- Add test_add_messages_to_delta_migration_preserves_message_history (sync
+ async) covering the primary real-world BinaryOperatorAggregate →
DeltaChannel migration path with real Message objects and IDs
- Hoist all in-function imports to module level in test_channels.py and
fix _delta_channel_with_type helper accordingly
- Add section headers in test_channels.py for better navigation
prune was not previously implemented on InMemorySaver (raised
NotImplementedError); adding a DeltaChannel-aware implementation is a
follow-up concern, not required for the core feature.
channel: None = None -> channel: None; all call sites pass None explicitly
and the default was never needed. Restores the original signature.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Renames `operator` → `reducer` and flips arg order to `(reducer, typ=None)`,
matching the new batch contract: `reducer(state, list[writes]) -> state`. The
reducer receives all writes for a step in one call instead of being folded
pairwise, enabling single-pass implementations that avoid O(N²) reprocessing.
`typ` is now optional — `_is_field_channel` in `graph/state.py` always
overwrites it from the `Annotated[T, ...]` outer type, so users can write
`DeltaChannel(my_reducer)` rather than `DeltaChannel(list, my_reducer)`.
Adds `_messages_delta_reducer` to `langgraph.graph.message` (experimental):
a single-pass bulk reducer for message lists that deduplicates by ID and
handles `RemoveMessage` tombstoning without calling `add_messages`, avoiding
repeated dedup passes that `add_messages` would incur in a fold.
Also fixes the `_delta_write_futs` mypy error in `AsyncPregelLoop` by moving
the type annotation to the class body, and unignores `new_pr_desc.md` from
the repo via `.gitignore`.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
In durability="async" mode (the default), put_writes calls are
fire-and-forget coroutines — a process crash between write submission and
checkpoint commit leaves a DELTA_SENTINEL blob with no backing writes,
causing silent data loss on replay.
AsyncPregelLoop now maintains _delta_write_futs: any write to a
DeltaChannel channel appends its asyncio.Future to this list in
accept_writes. _checkpointer_put_after_previous drains the list with
await asyncio.gather() before calling aput(), guaranteeing
checkpoint_writes are durable before the sentinel blob is committed.
The sync loop is unchanged: BackgroundExecutor.__exit__ already ensures
all background tasks complete before invoke() returns.
Also fixes DeltaChannel(list, add_messages) constructor call in
checkpoint-postgres async test (missing typ arg).
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
DELTA_SENTINEL is now serialized as a msgpack ext code (EXT_DELTA_SENTINEL=8)
alongside _DeltaSnapshot (ext 7), keeping both sentinel types in the same
codec path. The dedicated "delta" string type tag and its special-case in
dumps_typed/loads_typed are removed — no migration needed since this is
introduced fresh.
InMemorySaver.prune() updated to deserialize blobs and check `is DELTA_SENTINEL`
rather than comparing the raw type tag string, making it codec-agnostic.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Previously DeltaChannel.__init__ hardcoded typ=list and _is_field_channel
patched item.typ/item.value after construction. This mirrors BinaryOperatorAggregate:
- DeltaChannel(typ, operator, *, snapshot_frequency=None) — typ is now
a required first argument; __init__ strips abstract/parameterized types
to their concrete counterparts (same logic as BinaryOperatorAggregate)
- _is_field_channel reconstructs the channel via its constructor instead
of patching typ and value externally
- copy() and from_checkpoint() use self.__class__(self.typ, self.operator, ...)
— no post-construction attribute hacking needed
- _empty() helper removed; self.typ() is always a concrete callable
- All call sites updated: DeltaChannel(list, op), DeltaChannel(dict, op), etc.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
The helper was three lines called from exactly two places — inlining
it removes indirection without adding duplication.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Replace Any with the concrete element type — each row is (key, type_tag,
blob) all as bytes — matching how _load_blobs unpacks and decodes them.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
_needs_replay already gates on isinstance(spec, DeltaChannel), so the
assert/isinstance checks inside the replay branch were unreachable.
Replace with cast(DeltaChannel, spec) for zero-cost type narrowing that
survives -O and avoids any runtime check.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
delta.py — value consistency:
- `__init__` now starts with `value=MISSING` (was `[]`); both fresh
construction and clones are consistently uninitialised until
`from_checkpoint()` or `copy()` sets the real value
- `_clone_empty` drops `__new__` in favour of the normal constructor;
`typ` and `key` are restored explicitly afterwards (`typ` may differ
from `list` when set via Annotated injection; `key` is injected by
the graph builder after construction)
delta.py — snapshot cadence:
- `is_snapshot_step` now guards `step > 0`; snapshots fire at steps N,
2N, 3N, … instead of also at step 0 where `0 % N == 0` always held
_checkpoint.py — runtime guards:
- replace both `assert isinstance(spec, DeltaChannel)` with proper
`if not isinstance: raise TypeError`; `assert` is stripped by `-O`
and is wrong for production invariant checks
binop.py — avoid unnecessary allocation:
- `_get_overwrite`: replace `set(value.keys()) == {OVERWRITE}` with
`len(value) == 1 and OVERWRITE in value` to avoid allocating a
throwaway set on every call
checkpoint-postgres — typed rows:
- add `_DeltaCombinedRow(TypedDict, total=False)` documenting the nine
columns emitted by `SELECT_DELTA_COMBINED_SQL`'s UNION ALL; change
`_build_delta_channel_writes_history` parameter from `Sequence[Any]`
to `Sequence[_DeltaCombinedRow]`; call sites cast the psycopg
`DictRow` result accordingly
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Both BinaryOperatorAggregate and DeltaChannel had identical inline logic for
comparing operators that may be lambdas. Extract _operators_equal into
binop.py (alongside _get_overwrite) and use it in both __eq__ methods.
Also removes the duplicate _get_overwrite definition from delta.py — it was
identical to binop.py's and is now imported from there instead, along with
the now-unused OVERWRITE constant and Overwrite imports.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Dropped after removing the _DeltaSnapshot special-case in the seed-terminator
logic — write-collection ordering fix handles both blob types uniformly.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Replaces the three sequential SELECT roundtrips in _get_channel_writes_history
(checkpoints, checkpoint_writes, checkpoint_blobs) with one combined
UNION ALL query tagged by a _kind discriminator column. Both sync and async
paths now do one execute + one fetchall regardless of pipeline mode.
_build_delta_channel_writes_history is updated to accept the single tagged
rows list and dispatch on _kind while building its lookup dicts; the three
old SQL constants are removed.
Also fixes write-collection ordering in _build_delta_channel_writes_history:
the seed-terminator blob check previously fired before collecting that
ancestor's writes, silently dropping the transition writes needed to
reconstruct the child's state. Writes are now collected first.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Adds _get_tuple_raw / _aget_tuple_raw as the pure-storage-read layer that
_get_channel_writes_history calls instead of get_tuple. Default
implementation delegates to get_tuple for full backward compatibility — no
changes needed for existing savers whose get_tuple is a plain storage query.
Savers that perform channel hydration inside get_tuple can override
_get_tuple_raw with the raw read to structurally break any possible cycle;
a Python RecursionError surfaces the problem if they don't, rather than the
previous silent data corruption (returning empty writes).
Also fixes write-collection ordering in the reference implementation: pending
writes from the seed-terminator ancestor were silently dropped because the
terminator check fired before the collection loop. Writes are now collected
first so the seed ancestor's transition writes are included in reconstruction.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
copy() and from_checkpoint() hardcoded DeltaChannel instead of
self.__class__, breaking subclasses. Now mirrors the BinaryOperatorAggregate
pattern: self.__class__(self.operator) with explicit typ/key assignment.
update() applied non-overwrite values that arrived before an Overwrite in
the sequence, then discarded them when the Overwrite fired — order-dependent
behaviour in a method whose contract says order is arbitrary. Now pre-scans
for an Overwrite and applies only it (or folds all values normally if none).
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
The second `if hasattr(t, "__origin__")` block was unreachable — the first
branch always returned, so Required[T] / NotRequired[T] resolved to the bare
class instead of the inner type. Check Required/NotRequired before the generic
__origin__ fallback.
Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Two fixes exposed by running the postgres test suite against a local
postgres instance:
1. `PostgresSaver._get_channel_writes_history` /
`AsyncPostgresSaver._aget_channel_writes_history` required
`checkpoint_id` in the passed config, raising `KeyError` when called
with just `thread_id` (e.g. `graph.aget_state({"thread_id": "..."})`).
Now resolves to the latest checkpoint via `get_tuple`/`aget_tuple`
when the id is missing.
2. `test_get_checkpoint_no_channel_values` (sync + async) monkeypatched
`_load_checkpoint_tuple` with the old `(value, cur)` signature. Method
now takes `(value)` only since delta reconstruction moved out of the
tuple-load path — updated both tests.
Local postgres (`brew install pgvector postgresql@16`, running on port
5441) now exercises all 40 non-vector postgres tests green.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Restructure DeltaChannel reconstruction so the hydration path matches
pregel's storage axes (blobs + writes) without leaking internal DTOs
into the public checkpoint contract.
Key changes:
* Deleted `DeltaChannelWrites` dataclass and `SEED_UNSET` sentinel.
Reconstruction data no longer flows through `Checkpoint.channel_values`
as a wrapped DTO — that field now carries a value or `DELTA_SENTINEL`,
never a reconstruction shape.
* Added private `_ChannelWritesHistory(seed: Any, writes: list[PendingWrite])`
NamedTuple as the return type for the new storage-level query.
* Added private, experimental `_get_channel_writes_history` /
`_aget_channel_writes_history` on `BaseCheckpointSaver` — reference
impl via `get_tuple` + `parent_config` walk, overridden on
`InMemorySaver` / `PostgresSaver` / `AsyncPostgresSaver` for perf.
Fixes a latent migration bug in the base fallback (now inspects
ancestor `channel_values` for pre-delta seed).
* `DeltaChannel.from_checkpoint(seed)` simplified to two cases
(sentinel/MISSING → empty, else → seed). New `replay_writes` method
folds `list[PendingWrite]` through the reducer.
* Delta hydration consolidated inside `channels_from_checkpoint` via
optional `saver` + `config` kwargs (+ async mirror
`achannels_from_checkpoint`). All six pregel call sites updated.
`get_tuple` no longer patches `channel_values` — removed
`_resolve_delta_channels` (memory) and per-tuple reconstruction from
`_load_checkpoint_tuple` (postgres sync + async).
* Hydration short-circuits on the target's own blob: if
`channel_values[k]` is a real value (pre-migration tip, `update_state`
result), use it directly. Only walks ancestors when the target holds
sentinel or is missing. Fixes a correctness bug where migration-tip
and `update_state` values would be lost.
* New test_delta_channel_migration.py: 10 scenarios covering
BinaryOperatorAggregate → DeltaChannel migration (basic + async,
time-travel, fork, `update_state`, tip-of-pre-migration, base-saver
fallback parity, cross-thread isolation).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
snapshot_every was a knob for bounding reconstruction cost on deep threads.
Benchmarks (notes/add_messages_replay_problem.md + scratch work on
sr/add-messages-replay-bench) showed the add_messages fast-path
(optimize/add-messages-fast-path) closes the quadratic replay cost for
threads under ~1000 turns, where the crossover to snapshots makes sense.
For deeper threads we'll ship a first-class compaction primitive instead.
Removals:
* DeltaChannel: snapshot_every ctor param, _writes_since_snapshot counter,
should_snapshot() / snapshot_write() methods, counter threading through
_apply_write / update / from_checkpoint / copy.
* Pregel loop: post-checkpoint snapshot-injection block and
SNAPSHOT_TASK_ID import + constant.
* Checkpoint base: _overwrite_types() helper and the ancestor-walk
short-circuit on user-emitted Overwrite in sync + async
get_channel_writes.
* InMemory + Postgres savers: same walk-terminator shortcut. The
pre-delta blob terminator (seed-from-ancestor-blob) stays — it's
required for migration correctness, not a snapshot optimization.
* Tests for all of the above.
Preserved:
* Channel-level Overwrite semantics in DeltaChannel / BinOpAggregate:
Overwrite still resets the value at reducer level; same-super-step
dedup and InvalidUpdateError on multiple Overwrites still enforced.
* Pre-delta migration seeding.
The recursive CTE was bottlenecked by a JSON-expression join
(`bl.version = checkpoint->'channel_versions'->>bl.channel`) that the
planner could not index, producing an O(ancestors x blobs) nested-loop.
At depth 1000 it ran ~275 ms and removed ~2M filter rows; the recursion
itself was 2.4 ms.
Switch to three plain indexed SELECTs per delta channel
(checkpoints, checkpoint_writes, checkpoint_blobs); a pure helper on
BasePostgresSaver walks the parent chain and assembles
DeltaChannelWrites. Sync (__init__.py) and async (aio.py) each own
their three-roundtrip I/O wrappers.
Bench numbers (notes/delta_channel_query_bench.md): 3x at depth 50,
15x at depth 200, ~100x at depth 1000. Plain over-fetches sibling rows
when the thread branches but still wins at every realistic depth on
both local and remote postgres.
Multi-channel coalescing dropped — reconstruction is per-channel now.
Same shape as InMemorySaver. Can come back as a SQL-level
optimization later if needed.
Four fixes from an independent review of the reconstruction pipeline, plus
a structural cleanup:
1. Ancestor walk excludes the target checkpoint itself (matches pregel:
writes stored under checkpoint_id=T are pending for the NEXT step and
applied separately via apply_writes). Memory saver previously included
them, diverging from Postgres and causing pending writes to be folded
into the reconstructed snapshot — visible via get_state during
interrupts and time-travel into a non-leaf checkpoint.
2. Pre-delta blob terminator. When the walk hits an ancestor whose blob
for the channel is a real value (not DELTA_SENTINEL), bind that blob
as DeltaChannelWrites.seed and stop. Without this, threads migrated
from pre-delta storage would replay ancestor writes to the root
forever AND lose any value that lived only in the old blob
(e.g. from update_state). Per-ancestor, the blob is checked BEFORE
its writes — a pre-delta blob subsumes writes at the same checkpoint,
so including them would double-count.
3. Base-fallback get_channel_writes follows parent_checkpoint_id instead
of list(before=...). The previous form returned every tuple with
id<target, including sibling branches on forked threads.
4. seed replaces the Overwrite-wrapping hack for pre-delta values.
DeltaChannelWrites(writes, seed=SEED_UNSET) makes the saver's
reconstruction terminator semantically explicit; drops the lazy
_make_overwrite import dance. User-emitted Overwrite still reset the
chain via _apply_write as before.
Postgres: recursive CTE enumerates on-path ancestors and joins once
against checkpoint_writes and once against checkpoint_blobs for every
delta channel in the get_tuple — one roundtrip instead of the previous
3 queries × N channels.
Tests added:
- Pre-delta blob seeding (seed binding, no double-counting of ancestor
writes at the terminator, pending-at-target excluded).
- Root checkpoint returns empty writes.
- Seed-based from_checkpoint replay (three scenarios: with writes,
seed-only, seed=None distinct from SEED_UNSET).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Annotated[NotRequired[dict[...]], DeltaChannel(reducer)] (the shape used
by deepagents' filesystem middleware) fell through type inference to
`list`, so the first operator call blew up with
"'list' object is not a mapping". `_is_field_channel` now unwraps a
parameterized Required[X]/NotRequired[X] before stripping extras, which
lets dict/set/mapping outer types reach the abc normalization block.
Also type-annotates the `new` locals in DeltaChannel.copy() and
from_checkpoint() so mypy can infer them through the abstract return
type.
Adds tests covering: dict Overwrite in update and in writes replay,
snapshot_write with a dict reducer, dict backwards-compat checkpoints,
NotRequired type inference, and a filesystem-shaped end-to-end graph.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Instead of a recursive SQL CTE, collect the ancestor checkpoint ID chain in
Python by fetching all (checkpoint_id, parent_checkpoint_id) for the thread
in one query, then fetch writes with a plain WHERE checkpoint_id = ANY(...).
Simpler, avoids recursive query planner overhead, and uses well-indexed lookups.
DeltaChannel.checkpoint() now returns a zero-byte DeltaChannelSentinel
instead of duplicating delta data in checkpoint_blobs. Reconstruction
walks the parent checkpoint chain via checkpoint_writes (which already
holds per-step writes) and replays them through the operator.
In-memory benchmark (100 turns, ~20K tokens):
storage: 10.2 MB → 40.5 KB (251x reduction)
read: 0.6ms → 7.9ms (reconstruction cost, amortized by storage savings)
InMemorySaver and PostgresSaver override get_channel_writes() with
efficient implementations (Python dict walk and recursive CTE respectively).
The base class fallback uses self.list() with a thread-local recursion guard.
Remove the positional `typ` parameter from `DeltaChannel.__init__`. The type is
now injected automatically from the `Annotated` outer type in `_is_field_channel`
(matching how `BinaryOperatorAggregate` receives its type). `copy()` and
`from_checkpoint()` propagate `self.typ` explicitly. Test helpers updated to
use `_get_channel` with the proper `Annotated` path.
Rely on the runtime raise in DeltaChannel.from_checkpoint() instead of
a compile-time boolean flag. Savers that assemble DeltaChainValue inside
_load_blobs work transparently; savers that don't will pass through a raw
DeltaValue and hit a clear ValueError on first reload.
Removes: BaseCheckpointSaver.supports_delta_channels, the attribute on
InMemorySaver / PostgresSaver / AsyncPostgresSaver, the compile-time
UserWarning in StateGraph.compile(), and the associated test.
