Fixeslangchain-ai/langgraph#8534
`put` splits stored values in two: primitives stay inline in the
checkpoint's `channel_values`, everything else moves to
`checkpoint_blobs`, and only `_DeltaSnapshot` leaves an inline marker
behind when it moves. Stage-1 seed detection tested for that marker, so
a plain value — what a thread migrated from `BinaryOperatorAggregate`
leaves behind — was invisible to the walk.
### Effect
Migrated threads found no seed, walked to the root, and replayed every
write on every read. Values still came out correct, because replaying an
additive reducer from empty rebuilds the same list, which is why nothing
looked wrong. What was lost is early termination — the entire point of
`DeltaChannel`:
<!-- linear:table-colwidths:266,266,266 -->
| thread length | writes replayed, before | after |
| -- | -- | -- |
| 2 turns | 3 | 1 |
| 6 turns | 7 | 1 |
| 20 turns | 21 | 1 |
Read latency is flat at \~0.6ms across all three after the change.
### Approach
Stage 1 now checks both places a value can live rather than trusting the
marker. It probes `checkpoint_blobs`:
```sql
EXISTS (SELECT 1 FROM checkpoint_blobs b0
WHERE b0.thread_id = checkpoints.thread_id
AND b0.checkpoint_ns = checkpoints.checkpoint_ns
AND b0.channel = %s
AND b0.version = checkpoint -> 'channel_versions' ->> %s
AND b0.type <> 'empty') AS hb_0
```
and selects the inline value alongside it, since `None`, `str`, `int`,
`float` and `bool` stay in `channel_values` with no blob row:
```sql
checkpoint -> 'channel_values' -> %s AS inline_0
```
The blob predicate matches `checkpoint_blobs`' primary key `(thread_id,
checkpoint_ns, channel, version)` exactly, so it is one index lookup per
row per channel, bounded by the 1024-row page.
I picked reading storage over the cheaper alternative — also writing the
marker for plain values — because **that would not fix any thread
already on disk.** Existing checkpoints have no marker and there is
nowhere to add one retroactively.
The seed resolves to the blob when one exists and the inline value
otherwise. That ordering is also what keeps a genuine inline `true` — a
`bool` channel holding `True` — distinguishable from the literal `true`
marker `put` inlines for a `_DeltaSnapshot`: only the snapshot has a
blob.
`None` is deliberately not treated as a seed; a JSON null is
indistinguishable from "nothing stored" at this layer, so the walk
continues and replay from empty is correct.
Params go from two to four per channel; both callers updated.
The inline half came out of review on this PR — a blob-only probe would
have left scalar-aggregate migrations (an integer sum, say) still
replaying their full history.
### On the `type <> 'empty'` predicate
Being upfront since it isn't demonstrable with a test: `put` does not
currently produce `empty` rows on this path — `blob_versions` is
filtered to keys present in `channel_values`, so `_dump_blobs`' empty
branch is unreachable from it. I confirmed there are no `empty` rows in
a populated test database.
I kept it because stage 2 already applies the same check when resolving
the seed blob. Without it the two stages could disagree: stage 1
terminates the walk on a row stage 2 then discards, producing no seed
*and* a truncated write chain — the same failure shape this function
exists to avoid. Rationale is in the docstring so the next reader
doesn't have to ask. Happy to drop it if you'd rather not carry an
unexercised predicate.
### Tests
`libs/checkpoint-postgres/tests/test_delta_plain_value_seed.py` —
blob-stored plain-value seed, `_DeltaSnapshot` seed, a version bump with
nothing stored (which must not stop the walk short of an older real
value), inline primitives (`int`, `str`, `float`, `None`), and inline
`True` versus the snapshot marker. Each fails against the behaviour it
fixes.
Verified: postgres suite 269 passed on PG 15 and 16; delta-channel
conformance against `AsyncPostgresSaver` went from 6 of 8 to 8 of 8,
including the pre-existing `test_history_migration_plain_value_as_seed`
failure this was causing; `make lint` clean.
### Not included
I wanted a Postgres conformance runner alongside `checkpoint-sqlite`'s,
but it needs `langgraph-checkpoint-conformance` as a dev dependency and
the contributing guide asks for maintainer sign-off before adding one.
The direct tests above cover the same ground without it.
Worth flagging separately: **conformance effectively runs against**
`InMemorySaver` **only today.** `libs/checkpoint-conformance/tests/`
contains just `test_validate_memory.py`, and `checkpoint-sqlite`'s
`test_conformance_delta.py` silently skips because the package isn't
installed in its test environment (`importorskip`). Wiring it up for
sqlite and postgres is what would have caught this bug, and
langchain-ai/langgraph#8534 notes it.
Sqlite is unaffected by the bug itself — it stores `channel_values`
inline and inspects them directly. `langgraph-api` already resolves
seeds by version rather than by marker.
## Summary
- Ports the langgraph API checkpointer fix to the OSS Postgres
checkpoint.
- The `'b'` (seed-blob) branch of `_build_delta_stage2_sql`'s `UNION
ALL` was missing column aliases (`AS _kind`, `AS checkpoint_id`, `AS
task_id`, `AS idx`).
- In PostgreSQL, `UNION ALL` column names are taken from the **first**
`SELECT`. When `channels_with_chain` is empty but `channels_with_seed`
is not (e.g. very first run of a DeltaChannel graph, or
`snapshot_frequency=1`), the `'b'` branch becomes the first `SELECT`, so
columns got default names (`text`, `null`) instead of the expected
aliases — causing `KeyError: '_kind'` in
`_build_delta_channels_writes_history`.
## Test plan
- [ ] Existing `checkpoint-postgres` delta channel tests pass (`make
test` in `libs/checkpoint-postgres`)
- [ ] Manually verified fix matches the correction suggested in the
upstream API PR review comment
🤖 Generated with [Claude Code](https://claude.com/claude-code)
## Summary
- Promotes the private K-channel batched ancestor-walk to a stable
public `get_delta_channel_history` / `aget_delta_channel_history` API on
`BaseCheckpointSaver` (returns `Mapping[str, DeltaChannelHistory]`, a
TypedDict with `writes` always present and `seed` `NotRequired`)
- Removes `DELTA_SENTINEL` / `_DeltaSentinel` entirely — the saver layer
is now delta-agnostic on both write and read paths
- Reworks `DeltaChannel` snapshot cadence from "every Nth superstep" to
"every N updates to this channel," persisted in
`CheckpointMetadata.delta_updates_since_snapshot`
- Adds Postgres optimizations: paged stage-1 with cursor (1024-row
pages) and per-channel UNION ALL stage-2 (no over-fetch when channels
have different chain depths)
- Default `snapshot_frequency` becomes a positive int (default `1000`);
the previous `None` opt-out is removed
## Public API
```python
class DeltaChannelHistory(TypedDict):
writes: list[PendingWrite] # always present, possibly empty
seed: NotRequired[Any] # absent if walk reached root
def get_delta_channel_history(
self, *, config: RunnableConfig, channels: Sequence[str]
) -> Mapping[str, DeltaChannelHistory]: ...
async def aget_delta_channel_history(
self, *, config: RunnableConfig, channels: Sequence[str]
) -> Mapping[str, DeltaChannelHistory]: ...
```
`config` and `channels` are keyword-only so later additions (e.g.
`page_size`) don't shift the positional API.
The TypedDict-with-`NotRequired[seed]` shape matches the existing
checkpoint-package convention (`CheckpointMetadata` is
`TypedDict(total=False)`) — absence-via-key-omission rather than
introducing a new sentinel. Pregel translates `"seed" not in hist` to
`MISSING` on its side at consume time.
The default impl walks `get_tuple` + `parent_config` correctly but is
slow on long chains; savers that care override (`InMemorySaver`,
`PostgresSaver`).
## Sentinel removal
`DELTA_SENTINEL` and `_DeltaSentinel` are deleted entirely. The saver
layer becomes delta-agnostic:
- `DeltaChannel.checkpoint()` returns `MISSING` for non-snapshot steps;
pregel's `create_checkpoint` skips MISSING so delta channels without a
snapshot simply don't appear in `channel_values`
- `InMemorySaver.put` and Postgres `put` no longer filter sentinels
(they have nothing to filter)
- `_needs_replay` becomes `stored is MISSING`
- `DeltaChannel.from_checkpoint` accepts: `MISSING` → empty,
`_DeltaSnapshot(value)` → snapshot value, plain value → pre-migration
legacy
## Snapshot cadence
`DeltaChannel.snapshot_frequency: int` (default `1000`, positive). The
previous `None` opt-out is gone.
```python
def should_snapshot(ch_name, ch):
if force_delta_snapshot: # durability="exit"
return True
return updates_since_snapshot.get(ch_name, 0) >= ch.snapshot_frequency
```
Per-channel update counters are persisted in
`CheckpointMetadata.delta_updates_since_snapshot` (`NotRequired`,
`total=False`). The counter is incremented by `_put_checkpoint` for any
delta channel in `updated_channels` and reset to `0` by
`create_checkpoint` for channels that fire a snapshot this step.
Version-format-independent — works for `int`, `float`, and `str`
versioning schemes alike.
## Postgres optimization
Two improvements internal to the override:
**Stage-1 paged with cursor** (`LIMIT 1024` internal const, `AND
checkpoint_id < ?` for subsequent pages). The previous unpaged form
scanned every checkpoint in `(thread_id, ns)` and was pathological at
high thread depths.
**Stage-2 per-channel UNION ALL**: one `WHERE channel='X' AND
checkpoint_id = ANY(chain_X)` branch per channel plus one seed-blob
branch per channel with a seed. The previous form filtered by `channel =
ANY(channels) AND checkpoint_id = ANY(union_chain_cids)`, over-fetching
writes when channels had different chain depths (`K ×
max(chain_lengths)` vs the correct `sum(chain_lengths)`).
Both improvements stay internal to `PostgresSaver`/`AsyncPostgresSaver`;
the public contract returns a single `Mapping`.
## Benchmarks
`libs/langgraph/tests/test_delta_channel_benchmark.py`. Run via `python
libs/langgraph/tests/test_delta_channel_benchmark.py`. Postgres against
local pg:5441.
Results below trimmed to the high-signal cells. Sub-millisecond /
sub-100-turn rows omitted as warmup-bound; freq=1 omitted (chain depth =
1, nothing to optimize); peak read-time memory and Postgres storage are
flat between branches and omitted. Deep-thread reads and the
cadence-rework storage win are the load-bearing numbers.
### Postgres reads, 500 turns
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| Single-channel deep read | 17.7 ms | **6.1 ms** | **-66%** |
| Single-channel, 1000 turns | 35.0 ms | **14.3 ms** | **-59%** |
| K=3 channels, freq=50 uniform | 70.5 ms | **41.4 ms** | **-41%** |
| K=8 channels, freq=50 uniform | 214.2 ms | **139.4 ms** | **-35%** |
| K=8 channels, mixed freq (25/50/100/.../1000) | 295.6 ms | **214.4
ms** | **-27%** |
K-channel batching + paged stage-1 + per-channel UNION ALL stage-2 doing
exactly what they should at depth.
### InMemory reads, 500 turns
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| Single-channel deep read | 7.9 ms | **3.8 ms** | **-52%** |
| Single-channel, 1000 turns | 15.6 ms | **7.2 ms** | **-54%** |
| K=8 channels, freq=50 uniform | 112.3 ms | 94.6 ms | -16% |
| K=8 channels, mixed freq | 184.9 ms | **134.5 ms** | **-27%** |
### InMemory storage, 500 turns (cadence-rework win)
| Scenario | main | branch | Δ |
|---|---:|---:|---:|
| K=3, freq=50 uniform | 8.7 MB | **3.3 MB** | **-62%** |
| K=3 mixed freq | 3.8 MB | **1.3 MB** | **-66%** |
| K=8, freq=50 uniform | 23.1 MB | **8.7 MB** | **-62%** |
| K=8 mixed freq | 11.5 MB | **4.2 MB** | **-64%** |
Snapshot frequency now counts **channel updates** instead of
**supersteps**. On graphs where supersteps outpace per-channel updates
(e.g., input/end steps that don't write to channels), branch stores ~3×
fewer snapshot blobs.
### Tradeoff worth flagging
InMemory K=3 with mixed frequencies (50/200/1000) at 500 turns: **+64%
read latency** (46.6 → 76.5 ms). The mixed scenario has a channel with
`freq=1000` that goes the entire 500-turn run with no snapshot. On main,
the old superstep-counted cadence happened to fire at step=500 anyway.
New cadence gives users explicit control over walk depth via
`snapshot_frequency`. The K=8 mixed case still wins overall (-27%); this
regression is specific to the K=3 mixed shape.
Default `snapshot_frequency=1000` is the upper bound on walk depth —
it's a tunable knob.
## Tests
- New sqlite smoke test (`test_get_delta_channel_history.py`) exercises
the inherited default `BaseCheckpointSaver` impl via `SqliteSaver` /
`AsyncSqliteSaver` end-to-end with a real `DeltaChannel`-backed graph.
Sqlite uses the default unchanged — this validates the default path
actually works on a real second saver, not just on the optimized
override.
- Module-level `pytest.importorskip("langgraph.channels.delta")` guards
the test for sqlite's standalone CI environment (matches the postgres
pattern).
## Test plan
- [x] `libs/checkpoint`: 150 passed, 16 skipped
- [x] `libs/langgraph` (channels + delta migration): 41/41 (post-merge)
- [x] `libs/langgraph` (full pregel suite): 1784 passing — 6 "failures"
verified via `env -i` clean shell are local LangSmith env vars + `git
describe revision_id` polluting LangChain metadata fixtures; CI is
unaffected
- [x] `libs/checkpoint-postgres`: 40/40 saver tests + 3/3 delta channel
reconstruction tests against local Postgres
- [x] `libs/checkpoint-sqlite`: 105/105 (incl. retry-passed flake
`test_ttl_refresh`, unrelated to this PR)
- [x] Lint clean across all four libs (`ruff format`, `ruff check`,
`mypy`)
- [x] Branch-vs-main benchmarks — see results above
---------
Co-authored-by: Quanzheng Long <long@langchain.dev>
Co-authored-by: Cursor <cursoragent@cursor.com>
## Summary
Replaces the single-roundtrip `UNION ALL` DeltaChannel read with a
two-stage query that avoids fetching unused snapshot blobs, then removes
the old combined path entirely.
### Problem
`_get_channel_writes_history` used a single `UNION ALL` query that
fetched **all** checkpoint metadata, writes, and blobs for a
`(thread_id, channel)` in one shot. With `snapshot_frequency=N`, this
pulled back O(N/freq) full-size snapshot blobs even though only the
nearest one is needed to seed reconstruction. At 500 turns with
`snapshot_frequency=10`, this meant fetching ~100 complete
message-history snapshots per read.
### Solution
Two-stage read:
- **Stage 1** — lightweight scan of `checkpoints` only (no blob bytes):
walks the parent chain from the target checkpoint and stops at the first
ancestor with a snapshot, returning `chain_cids` and `seed_version`
- **Stage 2** — targeted fetch: only the writes for `chain_cids` and the
single seed blob at `seed_version`
The two-stage path is now unconditional — the old combined query and
`LG_DELTA_TWO_STAGE_QUERY` env-var gate have been removed.
### Sentinel cleanup
`DELTA_SENTINEL` is now a pure in-memory signal and is never written to
storage:
- Postgres `put()` already stripped it from `channel_values` before
writing blobs
- Memory saver `put()` now stores `"empty"` instead of serializing the
sentinel
- `EXT_DELTA_SENTINEL` (msgpack ext code 8) removed from
`JsonPlusSerializer`
- `DELTA_SENTINEL` is kept as an in-memory marker:
`DeltaChannel.checkpoint()` returns it so savers know to skip it, and
`_ChannelWritesHistory.seed` uses it to mean "no snapshot found, start
from empty"
## Performance
Benchmarked at `snapshot_frequency=10` on Postgres (`~100 tok/msg`):
| turns | old combined query | two-stage |
|------:|-------------------:|----------:|
| 50 | 6.0ms | 2.8ms (2.1x faster) |
| 100 | 10.1ms | 5.6ms (1.8x faster) |
| 500 | **216.1ms** | 15.3ms (**14x faster**) |
The old query's read time grew super-linearly with turn count because
each read fetched O(N/freq) full snapshot blobs. Two-stage keeps read
depth bounded by `snapshot_frequency` regardless of thread length.
## Test plan
- `make test` in `libs/checkpoint`, `libs/checkpoint-postgres`,
`libs/langgraph`
- Removed `test_delta_sentinel_serde_round_trip` (sentinel no longer
serializable)
- Updated `test_memory.py` — delta channel blobs stored as `"empty"`,
not serialized sentinel
- Updated `test_channels.py` — `channel_values` no longer contains
sentinel key for DeltaChannels
- Deleted `test_delta_channel_two_stage_benchmark.py` (one-stage vs
two-stage comparison; path no longer exists)
---------
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
- **Description:** The final migration for the postgres checkpointer is
not currently idempotent. That presents problems when migrating from one
checkpointer to another or if migrations otherwise get applied twice.
This makes the final migration idempotent to avoid this problem.
- **Issue:** N/A
- **Dependencies:** N/A
- **Twitter handle:** N/A
Issue
Support for `Checkpoint.metadata.writes` was dropped in `langgraph`
v0.5.x.
In `langgraph-checkpoint-postgres` v2.0.23, metadata was serialized with
`BasePostgresSaver._dump_metadata` -> `JsonPlusSerializer.dumps` which
handles `pydantic.BaseModel`.
In v2.0.23, metadata is serialized with `psycopg.types.json.Jsonb`,
which raises `TypeError: Object of type AIMessage is not JSON
serializable` when trying to serialize `writes`.
Solution
- Add `BaseCheckpointSaver.get_serializable_checkpoint_metadata` which
pops the `writes` key.
- Log deprecation warning when strange version combinations are used
Solves https://github.com/langchain-ai/langgraph/issues/5769
---------
Co-authored-by: Alex Kondratev <56111142+soapun@users.noreply.github.com>
This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
To make tests pass:
* linting fixes
* whitespace fixes in snapshots
---------
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
Prepare langgraph-checkpoint for 0.5
- Given we have no upper bound on langgraph-checkpoint dep need to undo all changes in langgraph-checkpoint that might break previous versions of langgraph
- This has been superseded by saving the individual writes of each task through put_writes()
- Removing this speeds up checkpoint operations as it was duplicating data saved elsewhere already
- Instead store sends in a Topic channel, removing the need to fetch sends as writes against the parent checkpoint
- Remove deprecated/unused functions in langgraph-checkpoint (will require bumping min range for langgraph-checkpoint in langgraph lib)
- Implement migration of old pending sends in langgraph-checkpoint-postgres
- Ensure parent config of `checkpoint_during=False` checkpoints always points to checkpoints that were also saved
- Initializing the store with an 'embedding config' -> this contains the
'dims' (used to create the table) and the encoder object (rn langchain
embeddings object, though that is ......)
- Call setup() -> creates the vector table.
Each document has 1 or more vectors associated with it for each json
path in the embedding config.
Would welcome critique and requests!
Leaving the params as the defaults for pgvector but open to feedback if
you think it's important to be able to more transparently configure that
in setup()
```python
from typing import TypedDict, List, Dict, Any, Optional
from langchain_openai import OpenAIEmbeddings
from langgraph.graph import StateGraph
from langgraph.store.postgres import PostgresStore
emb_config = {
"dims": 1536, # OpenAI embedding dimensions
"embed": OpenAIEmbeddings(model="text-embedding-3-small"),
"distance_type": "cosine",
}
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
store.setup()
# Define the state type for our graph
class State(TypedDict):
query: str
results: Optional[List[Dict[str, Any]]]
def put_stuff(state: State) -> State:
docs = [
("doc1", {"text": "red apple in kitchen"}),
("doc2", {"text": "blue car in garage"}),
("doc3", {"text": "green apple on table"}),
]
for key, value in docs:
store.put(("docs",), key, value)
def search_stuff(state: State) -> State:
"""Search for documents using vector similarity."""
results = store.search(("docs",), query=state["query"])
return {"results": results}
builder = StateGraph(State)
builder.add_node(put_stuff)
builder.add_node(search_stuff)
builder.add_edge("__start__", "put_stuff")
builder.add_edge("put_stuff", "search_stuff")
# Compile
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
chain = builder.compile(store=store)
result = chain.invoke({"query": "sour apple"})
# Print results
for doc in result["results"]:
print(doc.key)
print(doc.value)
print(doc.response_metadata)
```
- updates from inside Send tasks are applied in the order the Sends were created, if when you fan out, and have each task write results to a list with reducer, the final list is in the order you used when triggering
* Implement serialization with msgpack library
- encode custom python objects with a msgpack extension type, with constructor path string, and args encoded as nested msgpack doc
* Smaller msgpack extension types
* Update lock files
* lock
* Don't delegate to pydantic json
* Fix kafka serde
- should use our serializer to load, as inputs to subgraphs are serialized using it
* Performance improvements in checkpointer libs
- Use sha1 instead of md5 for hashing (faster in python 3.x)
- Use orjson instead of json for json dumping (sadly can't use for json loading)
* Update tests
* Update
* Use random number instead of hash for get_version_number
* Avoid saving writes for the last task to complete in each step
- only when possible, exceptions for ERROR, INTERRUPT, SEND
* Make Channel.from_checkpoint a regular function
- context manager no longer needed since Context became a managed value
* Use __slots__ for Channels
* Fix for kafka
- Sends are stored through put_writes, so we don't need to also store them inside checkpoint object
- On reading checkpoint, reconstruct pending_sends from the stored writes
* Fix semantics of put_writes/list
- put_writes(error) should not prevent saving future successful if task is retried successfully
- put_writes(writes) should be a no-op if non-error writes already exist for that task (this prevents tasks executed more than once from modifying writes previously saved / acted on)
- checkpoints should not include channel default values (ie those without a version)
- list() should fetch and return writes for each checkpoint
* Lint
* Rm print
* Fix import
* Lint
