WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-09 19:27:54 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(checkpoint): base saver _get_channel_writes_history handles _DeltaSnapshot correctly

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Sydney Runkle
2026-04-28 16:11:04 -04:00
co-authored by Claude Sonnet 4.6
parent 72343bdfb9
commit 5c0407f402
1 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint/langgraph/checkpoint/base/__init__.py
+17 -3
View File
@@ -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: