mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-20 08:37:59 +02:00
Quanzheng LongandGitHub
9100f2c682
Add a standalone recovery tool at `examples/delta-channel-dump/` for operators rolling back from langgraph >= 1.2 / deepagents 0.6.x to an older runtime that does not understand `EXT_DELTA_SNAPSHOT` msgpack blobs. Without this, channels like `messages` can appear empty after rollback because the old reducer does not decode delta snapshots. `dump.py` connects directly to Postgres, walks the checkpoint parent chain (mirroring `aget_delta_channel_history`), decodes msgpack blobs via `ormsgpack` with a hand-rolled EXT 0–7 hook (no langgraph imports), and emits JSON with per-channel `seed` + oldest-to-newest `writes`. The operator reduces and re-applies via `update_state` manually. ## Scope - Postgres checkpointer only - Deps: `psycopg[binary]`, `ormsgpack` (not added to repo `pyproject.toml` — operator installs at runtime) - No reducer application, no DB writes, no AES/custom encryption support (fails loudly on encrypted blob types) ## How verified - E2E against local Postgres: 15 runs of `delta_channel_messages_freq` (`snapshot_frequency=10`) - Asserted `delta_kind == "snapshot"`, `len(seed) == 10`, `len(writes) == 5`, message ids `ai-0`..`ai-14` - Confirmed portability: fresh venv with only `pip install "psycopg[binary]" ormsgpack` produced identical output
133 lines
3.9 KiB
Markdown
133 lines
3.9 KiB
Markdown
# delta-channel-dump
|
|
|
|
Recover messages (and other channels) from a Postgres-backed LangGraph thread
|
|
written by **langgraph >= 1.2** (DeltaChannel format) — including **LangGraph
|
|
Server / langgraph-api** deployments on the Postgres runtime, **deepagents
|
|
0.6.x**, or any OSS app using `PostgresSaver` — before rolling back to an older
|
|
runtime such as **deepagents 0.5.x / langgraph < 1.2**.
|
|
|
|
langgraph-api uses the same `checkpoints` / `checkpoint_blobs` / `checkpoint_writes`
|
|
schema as OSS `checkpoint-postgres`; this script reads those tables directly.
|
|
|
|
On the older runtime, `add_messages` does not understand the `EXT_DELTA_SNAPSHOT`
|
|
msgpack ext code and silently returns an empty list for affected channels. This
|
|
tool reads the raw checkpoint blobs from Postgres and emits JSON you can inspect
|
|
and re-apply via `update_state` (LangGraph Server SDK) or `graph.update_state`
|
|
(OSS).
|
|
|
|
## Install
|
|
|
|
```bash
|
|
pip install "psycopg[binary]" ormsgpack
|
|
```
|
|
|
|
## Run
|
|
|
|
```bash
|
|
export DATABASE_URI=postgres://user:pass@host:5432/dbname
|
|
python3 dump.py \
|
|
--thread-id <uuid> \
|
|
--channel messages \
|
|
[--channel files ...] \
|
|
[--checkpoint-id <uuid>] \
|
|
[--checkpoint-ns ""] \
|
|
--output recovery.json
|
|
```
|
|
|
|
- `--thread-id` (required): thread UUID
|
|
- `--channel` (required, repeatable): channel names to recover
|
|
- `--checkpoint-id` (optional): target checkpoint; defaults to latest
|
|
- `--checkpoint-ns` (optional): namespace; defaults to `""`
|
|
- `--database-uri` (optional): Postgres URI; defaults to `DATABASE_URI` env var
|
|
- `--output` (optional): output file; defaults to stdout
|
|
|
|
## Output
|
|
|
|
```json
|
|
{
|
|
"thread_id": "...",
|
|
"checkpoint_ns": "",
|
|
"target_checkpoint_id": "...",
|
|
"parent_checkpoint_id": "...",
|
|
"channels": {
|
|
"messages": {
|
|
"delta_kind": "snapshot",
|
|
"seed_checkpoint_id": "...",
|
|
"seed_version": "...",
|
|
"seed": [{ "type": "ai", "content": "...", "id": "ai-0" }],
|
|
"writes": [
|
|
{
|
|
"checkpoint_id": "...",
|
|
"task_id": "...",
|
|
"idx": 0,
|
|
"value": [{ "type": "ai", "content": "...", "id": "ai-10" }]
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
```
|
|
|
|
`delta_kind` is one of:
|
|
|
|
- `snapshot` — DeltaChannel snapshot blob (`channel_values[ch] == true`)
|
|
- `legacy_plain` — pre-DeltaChannel inline or blob value
|
|
- `no_seed` — walked to root without finding a populated ancestor
|
|
|
|
`writes` are ordered oldest-to-newest (the order a reducer would replay them).
|
|
|
|
## Reducing back to a single list
|
|
|
|
For deepagents-style messages, combine seed and writes, then deduplicate:
|
|
|
|
```python
|
|
import json
|
|
|
|
data = json.load(open("recovery.json"))
|
|
ch = data["channels"]["messages"]
|
|
messages = list(ch["seed"] or [])
|
|
for w in ch["writes"]:
|
|
messages.extend(w["value"] or [])
|
|
|
|
# Dedup by id, keep last; drop RemoveMessage tombstones
|
|
by_id = {}
|
|
for m in messages:
|
|
if isinstance(m, dict) and m.get("type") == "remove":
|
|
by_id.pop(m.get("id"), None)
|
|
elif isinstance(m, dict) and m.get("id"):
|
|
by_id[m["id"]] = m
|
|
else:
|
|
by_id[id(m)] = m
|
|
reduced = list(by_id.values())
|
|
```
|
|
|
|
This approximates `_messages_delta_reducer` semantics; adjust for your graph.
|
|
|
|
## Re-applying
|
|
|
|
```python
|
|
from langgraph_sdk import get_client
|
|
|
|
client = get_client(url="http://localhost:8123")
|
|
await client.threads.update_state(
|
|
thread_id,
|
|
values={"messages": reduced},
|
|
)
|
|
```
|
|
|
|
Review the recovered JSON before calling `update_state`. This tool is
|
|
read-only and intentionally does not mutate the database.
|
|
|
|
## Scope / non-goals (v1)
|
|
|
|
- **Postgres only** — OSS `PostgresSaver` or langgraph-api Postgres runtime; not
|
|
inmem, gRPC core, Mongo, or Redis checkpointer backends
|
|
- **No AES decryption** (`LANGGRAPH_AES_KEY`) or custom encryption
|
|
- **No reducer** — raw seed + writes only
|
|
- **No automatic `update_state`** — operator applies manually
|
|
|
|
## Copying
|
|
|
|
`dump.py` is self-contained. Copy it anywhere; only `psycopg[binary]` and
|
|
`ormsgpack` are required at runtime. No langgraph imports.
|