From 5c0407f4021159e2de9219f350420b3e38b3fbe0 Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Tue, 28 Apr 2026 16:11:04 -0400 Subject: [PATCH] fix(checkpoint): base saver _get_channel_writes_history handles _DeltaSnapshot correctly The base saver's _get_channel_writes_history walk terminated at any non-DELTA_SENTINEL blob without collecting that ancestor's pending_writes. This was correct for pre-delta migration blobs (which subsume their own writes), but wrong for _DeltaSnapshot blobs: the snapshot captures state AT the ancestor, while pending_writes encode the NEXT step's transition and must be collected before terminating. Applies the same fix as was already applied to InMemorySaver and PostgresSaver: when the ancestor blob is a _DeltaSnapshot, collect its pending_writes first, then terminate with the snapshot as seed. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- .../langgraph/checkpoint/base/__init__.py | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/libs/checkpoint/langgraph/checkpoint/base/__init__.py b/libs/checkpoint/langgraph/checkpoint/base/__init__.py index af4ab3dc4..7a482a3f9 100644 --- a/libs/checkpoint/langgraph/checkpoint/base/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/base/__init__.py @@ -28,6 +28,7 @@ from langgraph.checkpoint.serde.types import ( RESUME, SCHEDULED, ChannelProtocol, + _DeltaSnapshot, ) V = TypeVar("V", int, float, str) @@ -540,11 +541,18 @@ class BaseCheckpointSaver(Generic[V]): tup = self.get_tuple(cursor_config) if tup is None: break - # Pre-delta seed terminator: if the ancestor has a stored - # (non-sentinel) value for this channel, that snapshot - # subsumes any earlier writes on the chain. Stop here. ancestor_value = tup.checkpoint["channel_values"].get(channel) if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL: + if isinstance(ancestor_value, _DeltaSnapshot): + # Step-based snapshot: the blob is state AT this ancestor, + # but pending_writes encode the NEXT step's transition and + # are NOT subsumed — collect them before terminating. + if tup.pending_writes: + for write in reversed(tup.pending_writes): + if write[1] != channel: + continue + collected.append(write) + # Pre-delta blob: subsumes its own writes — stop immediately. collected.reverse() return _ChannelWritesHistory(seed=ancestor_value, writes=collected) if tup.pending_writes: @@ -580,6 +588,12 @@ class BaseCheckpointSaver(Generic[V]): break ancestor_value = tup.checkpoint["channel_values"].get(channel) if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL: + if isinstance(ancestor_value, _DeltaSnapshot): + if tup.pending_writes: + for write in reversed(tup.pending_writes): + if write[1] != channel: + continue + collected.append(write) collected.reverse() return _ChannelWritesHistory(seed=ancestor_value, writes=collected) if tup.pending_writes: