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 · 中文

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).

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.

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.