langgraph-ledger

Full, tamper-evident traceability for LangGraph agents — a port of DeepSeek Harness (dsh)'s traceability design: content-addressed labels on every tool call, a hash-chained append-only event ledger, the execution DAG, replay, crash recovery, and first-class fork/rollback.

English · 中文

pip install langgraph-ledger

Why

LangGraph already checkpoints state and can time-travel. What it does not give you is an audit-grade record: were these events edited after the fact? Which tool call exactly — same name, or same content? What did the run look like as a graph, and can I fork from any point of it?

This plugin ports the traceability design of DeepSeek Harness (dsh) — append-only session log as the single source of truth, format versioning, fork lineage — onto LangGraph's checkpointer contract, and adds what neither has out of the box: content hashing.

The combination is the point:

Differentiation (as of 2026-08)

Honest snapshot of the neighborhood — see POSITIONING for details:

Nobody else ships hash-labels + DAG + rollback as one LangGraph-native unit. That is the slot this project occupies.

Install

pip install langgraph-ledger

Quickstart

from langgraph.checkpoint.memory import InMemorySaver
from langgraph_ledger import TracingCheckpointSaver, DshTraceCallbackHandler

saver = TracingCheckpointSaver(InMemorySaver(), trace_root="./traces")
graph = builder.compile(checkpointer=saver)          # your graph, unchanged

graph.invoke(input, config={
    "configurable": {"thread_id": "run-42"},
    "callbacks": [DshTraceCallbackHandler()],         # LLM/tool/node events
})

Every run now leaves ./traces/run-42.jsonl — one hash-chained JSON event per line:

{"v":0,"seq":7,"ts":"…","kind":"tool/call","payload":{"label":"tl_9f2e…","name":"search",…},"id":"…","prev":"…"}
{"v":0,"seq":8,"ts":"…","kind":"state/snapshot","payload":{"checkpoint_id":"…","label":"cp_2c01…","parent_checkpoint_id":"…","checkpoint_sha256":"…"},"id":"…","prev":"…"}

Rollback & fork

from langgraph_ledger import time_travel_config, fork_thread

# resume/fork from any recorded checkpoint (LangGraph-native time travel)
cfg = time_travel_config("run-42", checkpoint_id="<past-id>")
graph.update_state(cfg, {"count": 100})              # forks from that point

# dsh-style: a new thread seeded with the ancestry up to a checkpoint
new_tid = fork_thread(saver, "run-42", at_checkpoint_id="<past-id>")

Audit

python -m langgraph_ledger verify  traces/run-42.jsonl   # hash chain intact?
python -m langgraph_ledger analyze traces/run-42.jsonl   # errors, loops, timeline
python -m langgraph_ledger dag     traces/run-42.jsonl --mermaid
python -m langgraph_ledger repair  traces/               # close crash-orphaned runs
python -m langgraph_ledger replay  traces/run-42.jsonl   # rebuild the message timeline

from langgraph_ledger import verify_thread
report = verify_thread(saver, "traces/run-42.jsonl")  # stored state == logged claim?
assert report["ok"]

Crash recovery & replay

If the process dies mid-run, the log ends with an unclosed run/start. verify flags it as an open run; repair appends an honest run/end {status: "interrupted"} through the hash chain (idempotent, never rewrites history). replay_messages() rebuilds the conversation timeline from the log — digests by default, full text when the handler was created with record_full=True. For fail-closed operation (a run that cannot record must not proceed), use TraceRecorder(..., strict=True) — enforced at the recorder and the checkpointer. Caveat: LangChain's callback manager may swallow handler exceptions, so for hard guarantees prefer the checkpointer path.

Design mapping from DeepSeek Harness

A design study, not a port of code (dsh is TypeScript/Node; this is Python/LangGraph):

Deliberate deviations: prompt/response full text is not stored by default (digest only, opt-in via record_full=True); dsh's byte-level replay and compaction are out of scope.

Proven in production

Before extraction, this design was battle-tested inside a production LangGraph agent product (private, Aug 2026) — a multi-round, tool-calling, real-model workload:

• Tracing overhead measured <0.1% of a turn — microsecond-level append+flush per event vs. hundreds of ms per LLM round-trip. "ON by default" was justified by measurement, not hope.
• 10 hook points wired with zero signature changes to existing code (contextvars propagation); with tracing OFF, behavior is bit-identical — pinned by tests.
• Snapshot/rollback drill passed end-to-end: file-level preimages, fail-closed on missing preimage (refuses rather than destroys), idempotent re-application.
• 3,150-test suite green after wiring the full hook surface.
• The trace layer was how failures were audited: one reviewed failure class went from 17/17 of cases to 0 after the trace made its mechanism visible; routing dead-ends went 2 → 0. (The latency cost of that refactor came from extra model rounds, not from tracing — the ledger itself stayed under 0.1%.)

Threat model (read before you rely on "tamper-evident")

The hash chain is keyless by default — and since 0.3.0, optionally keyed:

• ✅ Proven (keyless): given a trusted chain head, ANY deletion, reordering, or edit of ANY logged line is detectable (verify recomputes the full chain).
• ❌ Not proven (keyless): an attacker with write access to the log file who rewrites the entire file and re-chains it from genesis. No secret is involved, so nothing stops a full rewrite.
• 🔑 Keyed mode (HMAC-SHA256): pass hmac_key= to TraceRecorder / recorder_for, and the chain cannot be re-chained without the key — the full-rewrite attack fails even against an attacker who controls the file. Keep the key out of the log's trust domain (env var, secrets manager); losing the key means losing the ability to verify.

TraceRecorder("traces", thread_id, hmac_key=os.environ["LEDGER_HMAC_KEY"])

export LANGGRAPH_LEDGER_HMAC_KEY=...          # never pass keys as CLI args
python -m langgraph_ledger verify traces/run-42.jsonl

Keyed mode is fully backward compatible: keyless logs verify exactly as before (just don't pass a key), and content labels (tl_*, cp_*, payload digests) stay keyless so cross-log dedup and loop detection keep working without any secret.

Whether keyed or not: anchor the head outside the log's trust domain. Export it after each run and store it somewhere the process cannot write:

python -m langgraph_ledger head traces/run-42.jsonl   # {"seq": N, "id": "<hex>"}

Later, verify + a head comparison gives you end-to-end integrity. A daily cron appending heads to an append-only store (or an email to yourself) is enough.

Two more honest limits:

• One writer process per log. The append lock is in-process (threading.Lock); two processes appending to the same <thread>.jsonl will fork the chain. Use one trace root per process, or serialize externally.
• verify_thread needs the live saver. Checkpoint drift detection re-reads the saver, so it works across restarts only with a persistent backend (SQLite/Postgres). The log itself (verify_log) verifies anywhere.

What it is NOT

• Not a hosted observability platform — the log is a local file you own.
• Not byte-level replay of model streams — it is for error localization, audit and state rollback.
• Rollback restores agent state; side effects your tools made in the world are yours to undo (pair tool calls with your own preimages if you need that).

Development

pip install -e ".[test]"
pytest tests/

License

MIT — see LICENSE.