refactor(serde): DELTA_SENTINEL uses msgpack ext type EXT_DELTA_SENTINEL=8

Aligns DELTA_SENTINEL serialization with _DeltaSnapshot: both now go
through the msgpack ext hook (EXT_DELTA_SENTINEL=8, EXT_DELTA_SNAPSHOT=7)
rather than a separate "delta" type tag in dumps_typed/loads_typed.

Removes the special-case "delta" branch from dumps_typed and loads_typed.
Updates prune() sentinel detection to deserialize and compare rather than
checking the raw type tag string.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -16,7 +16,7 @@ from langgraph.checkpoint.base import (
     _ChannelWritesHistory,
     get_checkpoint_id,
 )
-from langgraph.checkpoint.serde.types import TASKS
+from langgraph.checkpoint.serde.types import TASKS, _DeltaSnapshot
 from psycopg.types.json import Jsonb
 
 MetadataInput = dict[str, Any] | None
@@ -281,13 +281,29 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
 
         collected: list[PendingWrite] = []  # newest first; reversed at the end
         for cid in ancestors:
-            # Pre-delta blob terminator: subsumes any writes at this ancestor.
             ver = ver_of.get(cid)
             if ver is not None:
                 seed_blob = blob_by_ver.get(ver)
                 if seed_blob is not None and seed_blob[0] != "empty":
                     blob_value = self.serde.loads_typed(seed_blob)
                     if blob_value is not DELTA_SENTINEL:
+                        if isinstance(blob_value, _DeltaSnapshot):
+                            # Step-based snapshot: collect this ancestor's
+                            # pending_writes first (they encode the NEXT step's
+                            # transition, not subsumed by the snapshot blob).
+                            for (
+                                type_tag,
+                                write_blob,
+                                task_id,
+                                _idx,
+                            ) in writes_by_cid.get(cid, []):
+                                val = self.serde.loads_typed((type_tag, write_blob))
+                                collected.append((task_id, channel, val))
+                            collected.reverse()
+                            return _ChannelWritesHistory(
+                                seed=blob_value, writes=collected
+                            )
+                        # Pre-delta blob: subsumes this ancestor's writes.
                         collected.reverse()
                         return _ChannelWritesHistory(seed=blob_value, writes=collected)
             for type_tag, write_blob, task_id, _idx in writes_by_cid.get(cid, []):

libs/checkpoint-postgres/tests/test_async.py

@@ -377,13 +377,13 @@ async def test_get_checkpoint_no_channel_values(
 async def test_delta_channel_chain_reconstruction(saver_name: str) -> None:
     """AsyncPostgresSaver reconstructs DeltaChannel chain via point-lookup traversal."""
     pytest.importorskip(
-        "langgraph.channels._delta", reason="langgraph core not installed"
+        "langgraph.channels.delta", reason="langgraph core not installed"
     )
 
     from typing import Annotated
 
     from langchain_core.messages import AIMessage, HumanMessage
-    from langgraph.channels._delta import DeltaChannel
+    from langgraph.channels.delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
     from typing_extensions import TypedDict

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)
@@ -545,6 +546,16 @@ class BaseCheckpointSaver(Generic[V]):
                 # 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:
@@ -578,8 +589,15 @@ class BaseCheckpointSaver(Generic[V]):
                 tup = await self.aget_tuple(cursor_config)
                 if tup is None:
                     break
+                # See sync variant for rationale.
                 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:

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -27,6 +27,7 @@ from langgraph.checkpoint.base import (
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 logger = logging.getLogger(__name__)
 
@@ -185,8 +186,31 @@ class InMemorySaver(
                     if blob_entry is not None and blob_entry[0] != "empty":
                         blob_value = self.serde.loads_typed(blob_entry)
                         if blob_value is not DELTA_SENTINEL:
-                            # Pre-delta snapshot terminator. Skip this
-                            # ancestor's writes — the blob subsumes them.
+                            if isinstance(blob_value, _DeltaSnapshot):
+                                # Step-based snapshot: the blob is state AT this
+                                # ancestor, but the ancestor's pending_writes
+                                # encode the NEXT step's transition and are NOT
+                                # subsumed by the snapshot — collect them first.
+                                step_writes = self.writes.get(
+                                    (thread_id, checkpoint_ns, cp_id), {}
+                                )
+                                for (_task_id, _idx), (
+                                    tid,
+                                    ch,
+                                    serialized,
+                                    _,
+                                ) in sorted(step_writes.items(), reverse=True):
+                                    if ch != channel:
+                                        continue
+                                    collected.append(
+                                        (tid, ch, self.serde.loads_typed(serialized))
+                                    )
+                                collected.reverse()
+                                return _ChannelWritesHistory(
+                                    seed=blob_value, writes=collected
+                                )
+                            # Pre-delta blob: state AT this ancestor already
+                            # subsumes its pending_writes — skip them.
                             collected.reverse()
                             return _ChannelWritesHistory(
                                 seed=blob_value, writes=collected
@@ -501,6 +525,101 @@ class InMemorySaver(
                 task_path,
             )
 
+    def prune(
+        self,
+        thread_ids: Sequence[str],
+        *,
+        strategy: str = "keep_latest",
+    ) -> None:
+        """Prune checkpoints for the given threads.
+
+        For DeltaChannel channels, a checkpoint is only deleted if the walk
+        from the latest checkpoint would not need to traverse it — i.e., a
+        `_DeltaSnapshot` blob exists in the kept ancestry that covers all
+        sentinel channels.  Checkpoints that are still in the active walk
+        chain (because no snapshot has been taken yet, e.g. with
+        `snapshot_frequency=None`) are retained.
+
+        Args:
+            thread_ids: Thread IDs to prune.
+            strategy: ``"keep_latest"`` keeps only the most recent checkpoint
+                per namespace; ``"delete"`` removes all checkpoints.
+        """
+        for thread_id in thread_ids:
+            if strategy == "delete":
+                self.delete_thread(thread_id)
+                continue
+
+            if strategy != "keep_latest":
+                raise ValueError(
+                    f"Unknown pruning strategy {strategy!r}. "
+                    "Expected 'keep_latest' or 'delete'."
+                )
+
+            for checkpoint_ns, ns_storage in list(
+                self.storage.get(thread_id, {}).items()
+            ):
+                if not ns_storage:
+                    continue
+
+                # Latest checkpoint (uuid6 IDs are lexicographically monotonic)
+                latest_id = max(ns_storage.keys())
+                latest_data, _, _ = ns_storage[latest_id]
+                latest_cp = self.serde.loads_typed(latest_data)
+
+                # Which channels in the latest checkpoint still have sentinels?
+                sentinel_channels: set[str] = set()
+                for ch, ver in latest_cp.get("channel_versions", {}).items():
+                    blob = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                    if blob is not None and blob[0] != "empty":
+                        if self.serde.loads_typed(blob) is DELTA_SENTINEL:
+                            sentinel_channels.add(ch)
+
+                # Walk the parent chain to find the oldest ancestor still needed.
+                # We stop (and mark "safe to prune before here") when all
+                # sentinel channels are covered by a non-sentinel blob.
+                required_ids: set[str] = {latest_id}
+                if sentinel_channels:
+                    _, _, parent_id = ns_storage[latest_id]
+                    remaining = set(sentinel_channels)
+                    while parent_id is not None and remaining:
+                        entry = ns_storage.get(parent_id)
+                        if entry is None:
+                            break
+                        required_ids.add(parent_id)
+                        cp_data, _, grandparent_id = entry
+                        cp = self.serde.loads_typed(cp_data)
+                        resolved: set[str] = set()
+                        for ch in remaining:
+                            ver = cp.get("channel_versions", {}).get(ch)
+                            if ver is None:
+                                continue
+                            blob = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                            if blob is not None and blob[0] != "empty":
+                                if self.serde.loads_typed(blob) is not DELTA_SENTINEL:
+                                    resolved.add(ch)
+                        remaining -= resolved
+                        parent_id = grandparent_id
+
+                # Delete everything outside the required set
+                for cp_id in list(ns_storage.keys()):
+                    if cp_id in required_ids:
+                        continue
+                    cp_data, _, _ = ns_storage.pop(cp_id)
+                    self.writes.pop((thread_id, checkpoint_ns, cp_id), None)
+
+                # Clean up blobs no longer referenced by any kept checkpoint
+                live: set[tuple[str, str, str, Any]] = set()
+                for cp_data, _, _ in ns_storage.values():
+                    cp = self.serde.loads_typed(cp_data)
+                    for ch, ver in cp.get("channel_versions", {}).items():
+                        live.add((thread_id, checkpoint_ns, ch, ver))
+                for key in [
+                    k for k in self.blobs if k[:2] == (thread_id, checkpoint_ns)
+                ]:
+                    if key not in live:
+                        del self.blobs[key]
+
     def delete_thread(self, thread_id: str) -> None:
         """Delete all checkpoints and writes associated with a thread ID.
 

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -33,7 +33,12 @@ from langchain_core.load.load import Reviver
 from langgraph.checkpoint.serde import _msgpack as _lg_msgpack
 from langgraph.checkpoint.serde.base import SerializerProtocol
 from langgraph.checkpoint.serde.event_hooks import emit_serde_event
-from langgraph.checkpoint.serde.types import DELTA_SENTINEL, SendProtocol
+from langgraph.checkpoint.serde.types import (
+    DELTA_SENTINEL,
+    SendProtocol,
+    _DeltaSentinel,
+    _DeltaSnapshot,
+)
 from langgraph.store.base import Item
 
 if TYPE_CHECKING:
@@ -251,8 +256,6 @@ class JsonPlusSerializer(SerializerProtocol):
     def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
         if obj is None:
             return "null", EMPTY_BYTES
-        elif obj is DELTA_SENTINEL:
-            return "delta", EMPTY_BYTES
         elif isinstance(obj, bytes):
             return "bytes", obj
         elif isinstance(obj, bytearray):
@@ -279,8 +282,6 @@ class JsonPlusSerializer(SerializerProtocol):
             return ormsgpack.unpackb(
                 data_, ext_hook=self._unpack_ext_hook, option=ormsgpack.OPT_NON_STR_KEYS
             )
-        elif type_ == "delta":
-            return DELTA_SENTINEL
         elif self.pickle_fallback and type_ == "pickle":
             return pickle.loads(data_)
         else:
@@ -296,10 +297,16 @@ EXT_METHOD_SINGLE_ARG = 3
 EXT_PYDANTIC_V1 = 4
 EXT_PYDANTIC_V2 = 5
 EXT_NUMPY_ARRAY = 6
+EXT_DELTA_SNAPSHOT = 7
+EXT_DELTA_SENTINEL = 8
 
 
 def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
-    if hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
+    if isinstance(obj, _DeltaSentinel):
+        return ormsgpack.Ext(EXT_DELTA_SENTINEL, b"")
+    elif isinstance(obj, _DeltaSnapshot):
+        return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
+    elif hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,
             _msgpack_enc(
@@ -613,7 +620,15 @@ def _create_msgpack_ext_hook(
         return False
 
     def ext_hook(code: int, data: bytes) -> Any:
-        if code == EXT_CONSTRUCTOR_SINGLE_ARG:
+        if code == EXT_DELTA_SENTINEL:
+            return DELTA_SENTINEL
+        elif code == EXT_DELTA_SNAPSHOT:
+            return _DeltaSnapshot(
+                ormsgpack.unpackb(
+                    data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS
+                )
+            )
+        elif code == EXT_CONSTRUCTOR_SINGLE_ARG:
             try:
                 tup = ormsgpack.unpackb(
                     data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS

libs/checkpoint/langgraph/checkpoint/serde/types.py

@@ -1,6 +1,7 @@
 from collections.abc import Sequence
 from typing import (
     Any,
+    NamedTuple,
     Protocol,
     TypeVar,
     runtime_checkable,
@@ -33,6 +34,20 @@ class _DeltaSentinel:
 DELTA_SENTINEL = _DeltaSentinel()
 
 
+class _DeltaSnapshot(NamedTuple):
+    """Snapshot blob for a DeltaChannel with finite snapshot_frequency.
+
+    Stored in checkpoint_blobs via the `EXT_DELTA_SNAPSHOT` msgpack ext code.
+    The ancestor walk in `_get_channel_writes_history` terminates when it
+    encounters this type (any non-sentinel blob stops the walk).
+
+    `from_checkpoint` reconstructs the channel value directly from `.value`
+    without replaying writes — the snapshot IS the accumulated state.
+    """
+
+    value: Any
+
+
 Value = TypeVar("Value", covariant=True)
 Update = TypeVar("Update", contravariant=True)
 C = TypeVar("C")

libs/checkpoint/tests/test_memory.py

@@ -663,3 +663,148 @@ class TestPreDeltaBlobTerminator:
         assert "PRE-DELTA-WRITE" not in values
         # And the pending write at the target is never folded in.
         assert "PENDING-AT-TARGET" not in values
+
+
+class TestInMemorySaverPrune:
+    """Tests for InMemorySaver.prune with DeltaChannel awareness."""
+
+    def _build_chain(
+        self,
+        saver: InMemorySaver,
+        thread_id: str,
+        ns: str,
+        channel: str,
+        n: int,
+        *,
+        snapshot_at: set[int] | None = None,
+    ) -> list[str]:
+        """Build a chain of n checkpoints with DELTA_SENTINEL blobs.
+
+        If snapshot_at is provided, writes a _DeltaSnapshot blob at those steps.
+        Returns list of checkpoint IDs in order (oldest first).
+        """
+        from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+        serde = saver.serde
+        cp_ids = []
+        parent_id = None
+        ver_base = "0000000000000000000000000000000{i}.0000000000000000"
+
+        for i in range(n):
+            cp_id = f"cp{i:04d}"
+            ver = ver_base.format(i=i)
+            cp = empty_checkpoint()
+            cp["id"] = cp_id
+            cp["channel_versions"][channel] = ver
+
+            if snapshot_at and i in snapshot_at:
+                blob = serde.dumps_typed(_DeltaSnapshot(value=[f"msg{i}"]))
+            else:
+                blob = serde.dumps_typed(DELTA_SENTINEL)
+
+            saver.blobs[(thread_id, ns, channel, ver)] = blob
+            saver.storage[thread_id][ns][cp_id] = (
+                serde.dumps_typed(cp),
+                serde.dumps_typed({}),
+                parent_id,
+            )
+            # Add a dummy write for this checkpoint
+            saver.writes[(thread_id, ns, cp_id)][("task", i)] = (
+                "task",
+                channel,
+                serde.dumps_typed(f"write{i}"),
+                "",
+            )
+            cp_ids.append(cp_id)
+            parent_id = cp_id
+
+        return cp_ids
+
+    def test_prune_pure_delta_keeps_all(self) -> None:
+        """With no snapshots, all checkpoints are required for reconstruction."""
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        cp_ids = self._build_chain(saver, thread_id, ns, channel, n=5)
+
+        saver.prune([thread_id], strategy="keep_latest")
+
+        # All checkpoints must be retained (walk needs the full chain)
+        remaining = set(saver.storage[thread_id][ns].keys())
+        assert remaining == set(cp_ids)
+
+    def test_prune_with_snapshot_removes_pre_snapshot_checkpoints(self) -> None:
+        """Checkpoints older than the nearest snapshot can be safely pruned."""
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        # Snapshot at step 2; steps 3 and 4 are sentinels
+        cp_ids = self._build_chain(saver, thread_id, ns, channel, n=5, snapshot_at={2})
+
+        saver.prune([thread_id], strategy="keep_latest")
+
+        remaining = set(saver.storage[thread_id][ns].keys())
+        # cp0, cp1 (before snapshot) must be gone; cp2, cp3, cp4 must remain
+        assert cp_ids[0] not in remaining  # pre-snapshot
+        assert cp_ids[1] not in remaining  # pre-snapshot
+        assert cp_ids[2] in remaining  # the snapshot itself
+        assert cp_ids[3] in remaining  # sentinel after snapshot
+        assert cp_ids[4] in remaining  # latest
+
+    def test_prune_removes_orphaned_blobs(self) -> None:
+        """Blob entries for pruned checkpoints are cleaned up."""
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        self._build_chain(saver, thread_id, ns, channel, n=4, snapshot_at={1})
+        saver.prune([thread_id], strategy="keep_latest")
+
+        # cp0 blob should be gone (pruned); cp1, cp2, cp3 blobs remain
+        assert (
+            thread_id,
+            ns,
+            channel,
+            f"0000000000000000000000000000000{0}.0000000000000000",
+        ) not in saver.blobs
+        for i in range(1, 4):
+            ver = f"0000000000000000000000000000000{i}.0000000000000000"
+            assert (thread_id, ns, channel, ver) in saver.blobs
+
+    def test_prune_removes_writes_for_pruned_checkpoints(self) -> None:
+        """checkpoint_writes for pruned checkpoints are deleted."""
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        cp_ids = self._build_chain(saver, thread_id, ns, channel, n=4, snapshot_at={1})
+
+        saver.prune([thread_id], strategy="keep_latest")
+
+        # writes for cp0 must be gone
+        assert (thread_id, ns, cp_ids[0]) not in saver.writes
+        # writes for cp1+ must remain (they're in the walk chain)
+        for cp_id in cp_ids[1:]:
+            assert (thread_id, ns, cp_id) in saver.writes
+
+    def test_prune_delete_strategy_removes_everything(self) -> None:
+        """strategy='delete' removes all checkpoints for the thread."""
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        self._build_chain(saver, thread_id, ns, channel, n=3)
+
+        saver.prune([thread_id], strategy="delete")
+
+        assert not saver.storage.get(thread_id, {}).get(ns)
+        assert not any(k[0] == thread_id for k in saver.writes)
+        assert not any(k[0] == thread_id for k in saver.blobs)
+
+    def test_prune_non_delta_channel_always_pruneable(self) -> None:
+        """A channel with full snapshot blobs (no sentinels) allows full prune."""
+
+        saver = InMemorySaver()
+        thread_id, ns, channel = "t1", "", "messages"
+        # All snapshots, no sentinels
+        cp_ids = self._build_chain(
+            saver, thread_id, ns, channel, n=4, snapshot_at={0, 1, 2, 3}
+        )
+
+        saver.prune([thread_id], strategy="keep_latest")
+
+        remaining = set(saver.storage[thread_id][ns].keys())
+        # Only the latest checkpoint is needed (all blobs are snapshots)
+        assert remaining == {cp_ids[-1]}

libs/langgraph/langgraph/channels/__init__.py

@@ -1,6 +1,7 @@
 from langgraph.channels.any_value import AnyValue
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue, LastValueAfterFinish
 from langgraph.channels.named_barrier_value import (
@@ -20,6 +21,7 @@ __all__ = (
     "UntrackedValue",
     "EphemeralValue",
     "BinaryOperatorAggregate",
+    "DeltaChannel",
     "NamedBarrierValue",
     "NamedBarrierValueAfterFinish",
     # topics

libs/langgraph/langgraph/channels/delta.py

@@ -5,12 +5,19 @@ from collections.abc import Callable, Sequence
 from typing import Any, Generic
 
 from langgraph.checkpoint.base import DELTA_SENTINEL, PendingWrite
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import Self
 
+from langgraph._internal._constants import OVERWRITE
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel, Value
-from langgraph.channels.binop import _get_overwrite
-from langgraph.errors import EmptyChannelError
+from langgraph.errors import (
+    EmptyChannelError,
+    ErrorCode,
+    InvalidUpdateError,
+    create_error_message,
+)
+from langgraph.types import Overwrite
 
 __all__ = ("DeltaChannel",)
 
@@ -22,48 +29,51 @@ def _empty(typ: Any) -> Any:
         return []
 
 
+def _get_overwrite(value: Any) -> tuple[bool, Any]:
+    if isinstance(value, Overwrite):
+        return True, value.value
+    if isinstance(value, dict) and set(value.keys()) == {OVERWRITE}:
+        return True, value[OVERWRITE]
+    return False, None
+
+
 class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
-    """Experimental — private API, subject to change or removal without notice.
+    """Fold-reducer channel with configurable snapshot cadence.
 
-    Imported from the underscored module `langgraph.channels._delta` on purpose;
-    not re-exported from `langgraph.channels`. Intended for internal use only
-    while we validate the design on real workloads.
+    `snapshot_frequency=None` (default): pure delta — stores only
+    `DELTA_SENTINEL` in checkpoint blobs; reads replay all ancestor writes.
 
-    A channel that stores only a sentinel in checkpoints; per-step writes are
-    stored in checkpoint_writes and replayed through the operator at load time.
+    `snapshot_frequency=N`: pregel's `create_checkpoint` writes a full
+    `_DeltaSnapshot` blob every N steps (eagerly, even if the channel had
+    no write that step). Reads walk at most N ancestor checkpoints before
+    hitting the snapshot, bounding replay depth to N regardless of thread
+    length.
 
-    Use with append-style reducers (e.g. `add_messages`) on long-running threads
-    to eliminate O(N²) blob growth — storage is O(N) using the writes table that
-    every checkpointer already maintains.
-
-    Reconstruction replays every ancestor write through the operator, so
-    per-get cost scales with thread depth. Compaction for deep threads is
-    a follow-up — today, use this on threads of a few hundred turns.
-
-    Usage::
-
-        from langgraph.channels._delta import DeltaChannel
-
-        class State(TypedDict):
-            messages: Annotated[list[AnyMessage], DeltaChannel(add_messages)]
+    Parameters:
+        operator: Binary reducer `(Value, Value) -> Value`.
+        snapshot_frequency: Every Nth pregel step writes a snapshot blob.
+            `None` (default) = pure delta, never snapshot.
     """
 
-    __slots__ = (
-        "value",
-        "operator",
-    )
+    __slots__ = ("value", "operator", "snapshot_frequency")
+    value: Value | Any
 
     def __init__(
         self,
         operator: Callable[[Any, Any], Any],
+        *,
+        snapshot_frequency: int | None = None,
     ) -> None:
         super().__init__(list)
         self.operator = operator
+        self.snapshot_frequency = snapshot_frequency
         self.value: Any = []
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, DeltaChannel):
             return False
+        if self.snapshot_frequency != other.snapshot_frequency:
+            return False
         if (
             self.operator.__name__ != "<lambda>"
             and other.operator.__name__ != "<lambda>"
@@ -79,20 +89,27 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
     def UpdateType(self) -> Any:
         return self.typ
 
-    def copy(self) -> Self:
-        new: DeltaChannel[Value] = DeltaChannel(self.operator)
+    def is_snapshot_step(self, step: int) -> bool:
+        """True if pregel should write a snapshot blob at this step."""
+        return (
+            self.snapshot_frequency is not None and step % self.snapshot_frequency == 0
+        )
+
+    def _clone_empty(self) -> Self:
+        new = self.__class__.__new__(self.__class__)
         new.typ = self.typ
         new.key = self.key
+        new.operator = self.operator
+        new.snapshot_frequency = self.snapshot_frequency
+        new.value = MISSING
+        return new
+
+    def copy(self) -> Self:
+        new = self._clone_empty()
         new.value = self.value if self.value is MISSING else _copy.copy(self.value)
         return new
 
     def _apply_write(self, value: Any, write: Any) -> Any:
-        """Apply one write to `value` and return the new value.
-
-        An `Overwrite` replaces the value; any other write is folded through
-        the operator. Centralizes the Overwrite/reducer branching used by both
-        `update` (live super-step) and `from_checkpoint` (ancestor replay).
-        """
         is_overwrite, overwrite_value = _get_overwrite(write)
         if is_overwrite:
             return (
@@ -104,38 +121,24 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         return self.operator(base, write)
 
     def from_checkpoint(self, checkpoint: Any) -> Self:
-        """Initialize from a seed value.
+        """Initialize from a stored blob or sentinel.
 
-        Pregel's hydration path calls this with the `seed` returned by
-        `saver.get_channel_history`:
-
-          * `MISSING` / `DELTA_SENTINEL` → channel starts empty. The walk
-            either reached the root (fresh delta thread) or found nothing
-            to seed from.
-          * any other value → use as the base value. Typically a pre-delta
-            blob preserved across a channel-type migration; `replay_writes`
-            folds subsequent deltas on top.
+        Blob types (dispatched via serde ext code, not dict key inspection):
+          * `DELTA_SENTINEL` / `MISSING`: start empty; caller replays writes.
+          * `_DeltaSnapshot(value)`: restore value directly from snapshot.
+          * plain value (migration from old BinOp blobs): use directly.
         """
-        new: DeltaChannel[Value] = DeltaChannel(self.operator)
-        new.typ = self.typ
-        new.key = self.key
+        new = self._clone_empty()
         if checkpoint is MISSING or checkpoint is DELTA_SENTINEL:
             new.value = _empty(new.typ)
+        elif isinstance(checkpoint, _DeltaSnapshot):
+            new.value = checkpoint.value
         else:
             new.value = checkpoint
         return new
 
     def replay_writes(self, writes: Sequence[PendingWrite]) -> None:
-        """Fold a sequence of `PendingWrite` tuples into the current value.
-
-        Called after `from_checkpoint` during pregel hydration to replay
-        per-step deltas from on-path ancestors through the reducer. Writes
-        are oldest→newest. `Overwrite` values inside the stream reset the
-        reducer state at that point, same as during a live super-step.
-        The `task_id` and `channel` fields of each `PendingWrite` are
-        ignored — `_get_channel_writes_history` has already filtered to
-        this channel.
-        """
+        """Fold ancestor writes oldest→newest into current value."""
         for _, _, value in writes:
             self.value = self._apply_write(self.value, value)
 
@@ -147,12 +150,6 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
             is_overwrite, _ = _get_overwrite(value)
             if is_overwrite:
                 if seen_overwrite:
-                    from langgraph.errors import (
-                        ErrorCode,
-                        InvalidUpdateError,
-                        create_error_message,
-                    )
-
                     msg = create_error_message(
                         message="Can receive only one Overwrite value per super-step.",
                         error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
@@ -160,7 +157,6 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
                     raise InvalidUpdateError(msg)
                 seen_overwrite = True
             elif seen_overwrite:
-                # Post-Overwrite writes within the same super-step are dropped.
                 continue
             self.value = self._apply_write(self.value, value)
         return True
@@ -174,4 +170,12 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         return self.value is not MISSING
 
     def checkpoint(self) -> Any:
+        """Return stored representation: always `DELTA_SENTINEL`.
+
+        Snapshot decisions are made by `create_checkpoint` in pregel (which
+        has the step number) via `is_snapshot_step`. `checkpoint()` is only
+        called for non-snapshot steps or when no checkpointer is available.
+        """
+        if self.value is MISSING:
+            return MISSING
         return DELTA_SENTINEL

libs/langgraph/langgraph/graph/message.py

@@ -184,97 +184,63 @@ def add_messages(
         ```
 
     """
-    # 1. Coerce scalars to lists.
+    remove_all_idx = None
+    # coerce to list
     if not isinstance(left, list):
         left = [left]  # type: ignore[assignment]
     if not isinstance(right, list):
         right = [right]  # type: ignore[assignment]
-    left = cast(list, left)
-
-    # 2. Normalize `left`. After the first call, `left` is the previous return
-    #    value of `add_messages` — a list of fully-resolved BaseMessages with
-    #    IDs — and needs no work. Fresh user input (dicts, tuples, message
-    #    chunks, BaseMessages without IDs) falls through to full conversion.
-    left_msgs: list[BaseMessage]
-    if (
-        left
-        and isinstance(left[0], BaseMessage)
-        and not isinstance(left[0], BaseMessageChunk)
-        and left[0].id is not None
-    ):
-        left_msgs = left
-    else:
-        left_msgs = [
-            message_chunk_to_message(cast(BaseMessageChunk, m))
-            for m in convert_to_messages(left)
-        ]
-        for m in left_msgs:
-            if m.id is None:
-                m.id = str(uuid.uuid4())
-
-    # 3. Normalize `right` — always fresh input. Assign missing IDs and detect
-    #    any RemoveMessage sentinels in a single pass.
-    right_msgs: list[BaseMessage] = [
+    # coerce to message
+    left = [
+        message_chunk_to_message(cast(BaseMessageChunk, m))
+        for m in convert_to_messages(left)
+    ]
+    right = [
         message_chunk_to_message(cast(BaseMessageChunk, m))
         for m in convert_to_messages(right)
     ]
-    remove_all_idx: int | None = None
-    has_remove = False
-    for idx, m in enumerate(right_msgs):
+    # assign missing ids
+    for m in left:
         if m.id is None:
             m.id = str(uuid.uuid4())
-        if isinstance(m, RemoveMessage):
-            has_remove = True
-            if m.id == REMOVE_ALL_MESSAGES:
-                remove_all_idx = idx
+    for idx, m in enumerate(right):
+        if m.id is None:
+            m.id = str(uuid.uuid4())
+        if isinstance(m, RemoveMessage) and m.id == REMOVE_ALL_MESSAGES:
+            remove_all_idx = idx
 
-    # 4. REMOVE_ALL_MESSAGES: discard everything up to and including the sentinel.
     if remove_all_idx is not None:
-        return right_msgs[remove_all_idx + 1 :]
+        return right[remove_all_idx + 1 :]
 
-    # 5. Decide fast vs. slow path. The fast path (pure append) is only valid
-    #    when `right` has no removals, no intra-right duplicate IDs, and no IDs
-    #    that overlap with `left` — any of those would force the indexed merge
-    #    below to update or dedup.
-    pure_append = False
-    if not has_remove:
-        left_ids = {m.id for m in left_msgs}
-        right_ids = {m.id for m in right_msgs}
-        pure_append = len(right_ids) == len(right_msgs) and right_ids.isdisjoint(
-            left_ids
-        )
-
-    if pure_append:
-        merged = left_msgs + right_msgs
-    else:
-        # 6. Slow path: build id→index map over `left`, then replay `right`.
-        #    In-place replacement for matching IDs, append for new IDs, and a
-        #    deferred removal pass so RemoveMessages can target either side.
-        merged = left_msgs.copy()
-        merged_by_id = {m.id: i for i, m in enumerate(merged)}
-        ids_to_remove = set()
-        for m in right_msgs:
-            if (existing_idx := merged_by_id.get(m.id)) is not None:
-                if isinstance(m, RemoveMessage):
-                    ids_to_remove.add(m.id)
-                else:
-                    ids_to_remove.discard(m.id)
-                    merged[existing_idx] = m
+    # merge
+    merged = left.copy()
+    merged_by_id = {m.id: i for i, m in enumerate(merged)}
+    ids_to_remove = set()
+    for m in right:
+        if (existing_idx := merged_by_id.get(m.id)) is not None:
+            if isinstance(m, RemoveMessage):
+                ids_to_remove.add(m.id)
             else:
-                if isinstance(m, RemoveMessage):
-                    raise ValueError(
-                        f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
-                    )
-                merged_by_id[m.id] = len(merged)
-                merged.append(m)
-        merged = [m for m in merged if m.id not in ids_to_remove]
+                ids_to_remove.discard(m.id)
+                merged[existing_idx] = m
+        else:
+            if isinstance(m, RemoveMessage):
+                raise ValueError(
+                    f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
+                )
+
+            merged_by_id[m.id] = len(merged)
+            merged.append(m)
+    merged = [m for m in merged if m.id not in ids_to_remove]
 
-    # 7. Apply optional output format.
     if format == "langchain-openai":
-        return _format_messages(merged)
-    if format:
+        merged = _format_messages(merged)
+    elif format:
         msg = f"Unrecognized {format=}. Expected one of 'langchain-openai', None."
         raise ValueError(msg)
+    else:
+        pass
+
     return merged
 
 

libs/langgraph/langgraph/graph/state.py

@@ -49,7 +49,7 @@ from langgraph._internal._runnable import coerce_to_runnable
 from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.binop import BinaryOperatorAggregate, _strip_extras
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue, LastValueAfterFinish
 from langgraph.channels.named_barrier_value import (

libs/langgraph/langgraph/pregel/_algo.py

@@ -210,7 +210,7 @@ def local_read(
     return values
 
 
-def increment(current: int | None, channel: None) -> int:
+def increment(current: int | None, channel: None = None) -> int:
     """Default channel versioning function, increments the current int version."""
     return current + 1 if current is not None else 1
 

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -1,19 +1,23 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import DELTA_SENTINEL, BaseCheckpointSaver, Checkpoint
 from langgraph.checkpoint.base.id import uuid6
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
 
 LATEST_VERSION = 4
 
+GetNextVersion = Callable[[Any, None], Any]
+
 
 def empty_checkpoint() -> Checkpoint:
     return Checkpoint(
@@ -33,30 +37,67 @@ def create_checkpoint(
     *,
     id: str | None = None,
     updated_channels: set[str] | None = None,
+    get_next_version: GetNextVersion | None = None,
 ) -> Checkpoint:
-    """Create a checkpoint for the given channels."""
+    """Create a checkpoint for the given channels.
+
+    For `DeltaChannel` with `snapshot_frequency=N`, snapshot steps write a
+    `_DeltaSnapshot` blob rather than `DELTA_SENTINEL`, bounding the ancestor
+    walk to at most N steps. Snapshots are eager: even if the channel had no
+    write this step, a version bump is forced (via `get_next_version`) so the
+    blob is stored by `put()`. Without `get_next_version` (e.g. static
+    contexts), snapshot steps gracefully fall back to sentinel.
+    """
     ts = datetime.now(timezone.utc).isoformat()
     if channels is None:
         values = checkpoint["channel_values"]
+        channel_versions = checkpoint["channel_versions"]
     else:
         values = {}
+        channel_versions = dict(checkpoint["channel_versions"])
         for k in channels:
-            if k not in checkpoint["channel_versions"]:
+            if k not in channel_versions:
                 continue
-            v = channels[k].checkpoint()
-            if v is not MISSING:
-                values[k] = v
+            ch = channels[k]
+            if (
+                isinstance(ch, DeltaChannel)
+                and ch.is_snapshot_step(step)
+                and ch.is_available()
+            ):
+                # Eager snapshot: bump version if not already written this step
+                # so put() includes this channel in new_versions and stores blob.
+                if get_next_version is not None and (
+                    updated_channels is None or k not in updated_channels
+                ):
+                    channel_versions[k] = get_next_version(channel_versions[k], None)
+                values[k] = _DeltaSnapshot(ch.get())
+            else:
+                v = ch.checkpoint()
+                if v is not MISSING:
+                    values[k] = v
     return Checkpoint(
         v=LATEST_VERSION,
         ts=ts,
         id=id or str(uuid6(clock_seq=step)),
         channel_values=values,
-        channel_versions=checkpoint["channel_versions"],
+        channel_versions=channel_versions,
         versions_seen=checkpoint["versions_seen"],
         updated_channels=None if updated_channels is None else sorted(updated_channels),
     )
 
 
+def _needs_replay(spec: BaseChannel, stored: object) -> bool:
+    """True if `spec` is a `DeltaChannel` and the stored blob is a sentinel,
+    requiring an ancestor walk to reconstruct.
+
+    `_DeltaSnapshot` blobs and plain values (migration) resolve directly via
+    `from_checkpoint` — only `DELTA_SENTINEL` / `MISSING` trigger replay.
+    """
+    if not isinstance(spec, DeltaChannel):
+        return False
+    return stored is MISSING or stored is DELTA_SENTINEL
+
+
 def channels_from_checkpoint(
     specs: Mapping[str, BaseChannel | ManagedValueSpec],
     checkpoint: Checkpoint,
@@ -67,14 +108,10 @@ def channels_from_checkpoint(
     """Hydrate channels from a checkpoint.
 
     For most channels, `spec.from_checkpoint(checkpoint["channel_values"][k])`
-    is sufficient — the stored value IS the reconstructed state.
-
-    `DeltaChannel` is the exception: its stored value is a sentinel; the
-    full state is spread across `checkpoint_writes` along the ancestor
-    chain. When `saver` and `config` are provided, this function fetches
-    that history via `saver._get_channel_writes_history` and folds it
-    through the channel's reducer. Without them (static contexts — graph
-    drawing, unit tests), delta channels fall back to empty.
+    is sufficient. `DeltaChannel` is the exception: sentinel blobs require an
+    ancestor walk via `saver._get_channel_writes_history`. The walk terminates
+    at the nearest `_DeltaSnapshot` blob (step-based) or a pre-migration plain
+    value, so read depth is bounded by `snapshot_frequency`.
     """
     channel_specs: dict[str, BaseChannel] = {}
     managed_specs: dict[str, ManagedValueSpec] = {}
@@ -88,22 +125,12 @@ def channels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         stored = checkpoint["channel_values"].get(k, MISSING)
-        if (
-            isinstance(spec, DeltaChannel)
-            and saver is not None
-            and config is not None
-            and (stored is MISSING or stored is DELTA_SENTINEL)
-        ):
-            # Target's own blob is empty/sentinel — walk ancestors for
-            # seed + writes. Skipping this when `stored` is a real value
-            # preserves state written via `update_state` or sitting at the
-            # tip of a pre-migration thread: the saver's ancestor walk
-            # intentionally excludes the target's own blob, so without
-            # this short-circuit we'd lose it.
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
+            assert isinstance(spec, DeltaChannel)
             history = saver._get_channel_writes_history(config, k)
-            delta_ch = spec.from_checkpoint(history.seed)
-            delta_ch.replay_writes(history.writes)
-            ch = delta_ch
+            replay_ch = spec.from_checkpoint(history.seed)
+            replay_ch.replay_writes(history.writes)
+            ch = replay_ch
         else:
             ch = spec.from_checkpoint(stored)
         channels[k] = ch
@@ -130,16 +157,12 @@ async def achannels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         stored = checkpoint["channel_values"].get(k, MISSING)
-        if (
-            isinstance(spec, DeltaChannel)
-            and saver is not None
-            and config is not None
-            and (stored is MISSING or stored is DELTA_SENTINEL)
-        ):
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
+            assert isinstance(spec, DeltaChannel)
             history = await saver._aget_channel_writes_history(config, k)
-            delta_ch = spec.from_checkpoint(history.seed)
-            delta_ch.replay_writes(history.writes)
-            ch = delta_ch
+            replay_ch = spec.from_checkpoint(history.seed)
+            replay_ch.replay_writes(history.writes)
+            ch = replay_ch
         else:
             ch = spec.from_checkpoint(stored)
         channels[k] = ch

libs/langgraph/langgraph/pregel/_loop.py

@@ -25,6 +25,7 @@ from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
@@ -198,6 +199,7 @@ class PregelLoop:
     checkpoint_pending_writes: list[PendingWrite]
     checkpoint_previous_versions: dict[str, str | float | int]
     prev_checkpoint_config: RunnableConfig | None
+    _pending_write_futs: list[concurrent.futures.Future]
 
     status: Literal[
         "input",
@@ -407,7 +409,7 @@ class PregelLoop:
                     task = self.tasks.get(task_id)
                 else:
                     task = None
-                self.submit(
+                fut = self.submit(
                     self.checkpointer_put_writes,
                     config,
                     writes_to_save,
@@ -415,12 +417,13 @@ class PregelLoop:
                     task_path_str(task.path) if task else "",
                 )
             else:
-                self.submit(
+                fut = self.submit(
                     self.checkpointer_put_writes,
                     config,
                     writes_to_save,
                     task_id,
                 )
+            self._pending_write_futs.append(fut)
         # output writes
         if hasattr(self, "tasks"):
             self.output_writes(task_id, writes)
@@ -891,6 +894,9 @@ class PregelLoop:
             self.step,
             id=self.checkpoint["id"] if exiting else None,
             updated_channels=self.updated_channels,
+            get_next_version=self.checkpointer_get_next_version
+            if do_checkpoint
+            else None,
         )
         # sanitize TASK channel in the checkpoint before saving (durability=="exit")
         if TASKS in self.checkpoint["channel_values"] and any(
@@ -928,6 +934,17 @@ class PregelLoop:
             )
             self.checkpoint_previous_versions = channel_versions
 
+            # If the checkpoint has any DELTA_SENTINEL blobs, the sentinel is
+            # only meaningful if checkpoint_writes are durable first.  Flush
+            # pending write futures synchronously before committing the blob so
+            # we never end up with a sentinel blob backed by missing writes.
+            if self._pending_write_futs and any(
+                v is DELTA_SENTINEL for v in self.checkpoint["channel_values"].values()
+            ):
+                for fut in self._pending_write_futs:
+                    fut.result()
+                self._pending_write_futs.clear()
+
             # save it, without blocking
             # if there's a previous checkpoint save in progress, wait for it
             # ensuring checkpointers receive checkpoints in order
@@ -1272,6 +1289,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
+        self._pending_write_futs = []
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs,
@@ -1477,6 +1495,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
+        self._pending_write_futs = []
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )

libs/langgraph/tests/test_add_messages_benchmark.py

@@ -1,290 +0,0 @@
-"""Benchmark: add_messages fast-path optimizations.
-
-Both implementations are inlined so the benchmark is self-contained and
-immune to import-cache or installed-vs-local confusion.
-
-Run directly:
-    python tests/test_add_messages_benchmark.py
-
-Or via pytest (correctness only, numbers printed to stdout):
-    pytest tests/test_add_messages_benchmark.py -s -v
-"""
-
-import statistics
-import time
-import tracemalloc
-import uuid
-from typing import cast
-
-from langchain_core.messages import (
-    AIMessage,
-    BaseMessage,
-    BaseMessageChunk,
-    HumanMessage,
-    RemoveMessage,
-    convert_to_messages,
-    message_chunk_to_message,
-)
-
-from langgraph.graph.message import REMOVE_ALL_MESSAGES
-
-# ── original implementation (pre-optimisation) ────────────────────────────────
-
-
-def _add_messages_original(left, right):
-    remove_all_idx = None
-    if not isinstance(left, list):
-        left = [left]
-    if not isinstance(right, list):
-        right = [right]
-    left = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(left)
-    ]
-    right = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(right)
-    ]
-    for m in left:
-        if m.id is None:
-            m.id = str(uuid.uuid4())
-    for idx, m in enumerate(right):
-        if m.id is None:
-            m.id = str(uuid.uuid4())
-        if isinstance(m, RemoveMessage) and m.id == REMOVE_ALL_MESSAGES:
-            remove_all_idx = idx
-    if remove_all_idx is not None:
-        return right[remove_all_idx + 1 :]
-    merged = left.copy()
-    merged_by_id = {m.id: i for i, m in enumerate(merged)}
-    ids_to_remove = set()
-    for m in right:
-        if (existing_idx := merged_by_id.get(m.id)) is not None:
-            if isinstance(m, RemoveMessage):
-                ids_to_remove.add(m.id)
-            else:
-                ids_to_remove.discard(m.id)
-                merged[existing_idx] = m
-        else:
-            if isinstance(m, RemoveMessage):
-                raise ValueError(
-                    f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
-                )
-            merged_by_id[m.id] = len(merged)
-            merged.append(m)
-    return [m for m in merged if m.id not in ids_to_remove]
-
-
-# ── optimised implementation ──────────────────────────────────────────────────
-
-
-def _add_messages_optimized(left, right):
-    if not isinstance(left, list):
-        left = [left]
-    if not isinstance(right, list):
-        right = [right]
-
-    # Optimisation 1: skip conversion + ID assignment on left when it already
-    # contains fully-resolved BaseMessage objects (the common case after the
-    # first call, since add_messages always returns list[BaseMessage] with IDs).
-    if (
-        left
-        and isinstance(left[0], BaseMessage)
-        and not isinstance(left[0], BaseMessageChunk)
-    ):
-        left = cast(list[BaseMessage], left)
-    else:
-        left = [
-            message_chunk_to_message(cast(BaseMessageChunk, m))
-            for m in convert_to_messages(left)
-        ]
-        for m in left:
-            if m.id is None:
-                m.id = str(uuid.uuid4())
-
-    # always normalise right — it's fresh external input
-    right = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(right)
-    ]
-    remove_all_idx = None
-    has_remove = False
-    for idx, m in enumerate(right):
-        if m.id is None:
-            m.id = str(uuid.uuid4())
-        if isinstance(m, RemoveMessage):
-            has_remove = True
-            if m.id == REMOVE_ALL_MESSAGES:
-                remove_all_idx = idx
-
-    if remove_all_idx is not None:
-        return right[remove_all_idx + 1 :]
-
-    # Optimisation 2: pure-append fast path — no removals and no ID overlaps.
-    # Builds one set over left instead of copying left + building a full dict.
-    if not has_remove:
-        left_ids = {m.id for m in left}
-        if not any(m.id in left_ids for m in right):
-            return left + right
-
-    # slow path: updates or removals present — full indexed merge
-    merged = left.copy()
-    merged_by_id = {m.id: i for i, m in enumerate(merged)}
-    ids_to_remove = set()
-    for m in right:
-        if (existing_idx := merged_by_id.get(m.id)) is not None:
-            if isinstance(m, RemoveMessage):
-                ids_to_remove.add(m.id)
-            else:
-                ids_to_remove.discard(m.id)
-                merged[existing_idx] = m
-        else:
-            if isinstance(m, RemoveMessage):
-                raise ValueError(
-                    f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
-                )
-            merged_by_id[m.id] = len(merged)
-            merged.append(m)
-    return [m for m in merged if m.id not in ids_to_remove]
-
-
-# ── helpers ───────────────────────────────────────────────────────────────────
-
-
-def _make_messages(n: int) -> list[BaseMessage]:
-    return [
-        (HumanMessage if i % 2 == 0 else AIMessage)(
-            content=f"message {i}", id=str(uuid.uuid4())
-        )
-        for i in range(n)
-    ]
-
-
-def _bench_time(fn, left, right, *, iters: int = 2_000) -> float:
-    """Return median latency in microseconds."""
-    for _ in range(100):
-        fn(list(left), list(right))
-    times = []
-    for _ in range(iters):
-        left_copy, right_copy = list(left), list(right)
-        t0 = time.perf_counter()
-        fn(left_copy, right_copy)
-        times.append(time.perf_counter() - t0)
-    return statistics.median(times) * 1e6
-
-
-def _bench_memory(fn, left, right) -> int:
-    """Return peak memory allocated during a single call (bytes)."""
-    # one warm-up so any lazy init is excluded
-    fn(list(left), list(right))
-    left_copy, right_copy = list(left), list(right)
-    tracemalloc.start()
-    tracemalloc.clear_traces()
-    fn(left_copy, right_copy)
-    _, peak = tracemalloc.get_traced_memory()
-    tracemalloc.stop()
-    return peak
-
-
-# ── scenarios ─────────────────────────────────────────────────────────────────
-
-SCENARIOS = [
-    ("pure append  1 → 1  msg", 1, 1, "append"),
-    ("pure append  10 → 1  msg", 10, 1, "append"),
-    ("pure append  100 → 1  msg", 100, 1, "append"),
-    ("pure append  1000 → 1  msg", 1000, 1, "append"),
-    ("pure append  1000 → 5  msgs", 1000, 5, "append"),
-    ("update existing  100 → 1  msg", 100, 1, "update"),
-    ("remove message  100 → 1  msg", 100, 1, "remove"),
-]
-
-
-def _make_inputs(n_left, n_right, mode):
-    left = _make_messages(n_left)
-    right = _make_messages(n_right)
-    if mode == "update":
-        right[0] = AIMessage(content="updated", id=left[0].id)
-    elif mode == "remove":
-        right = [RemoveMessage(id=left[0].id)]
-    return left, right
-
-
-# ── main output ───────────────────────────────────────────────────────────────
-
-COL = 36
-
-
-def run_benchmarks() -> None:
-    print()
-    print("=" * 88)
-    print("add_messages benchmark — time (µs, median of 2 000 iterations)")
-    print("=" * 88)
-    print(f"{'Scenario':<{COL}} {'Original':>10} {'Optimized':>11} {'Speedup':>8}")
-    print("-" * 88)
-
-    for label, n_left, n_right, mode in SCENARIOS:
-        left, right = _make_inputs(n_left, n_right, mode)
-        t_orig = _bench_time(_add_messages_original, left, right)
-        t_opt = _bench_time(_add_messages_optimized, left, right)
-        print(f"{label:<{COL}} {t_orig:>10.2f} {t_opt:>11.2f} {t_orig / t_opt:>7.2f}x")
-
-    print()
-    print("=" * 88)
-    print("add_messages benchmark — peak memory allocated per call (bytes)")
-    print("=" * 88)
-    print(f"{'Scenario':<{COL}} {'Original':>10} {'Optimized':>11} {'Reduction':>10}")
-    print("-" * 88)
-
-    for label, n_left, n_right, mode in SCENARIOS:
-        left, right = _make_inputs(n_left, n_right, mode)
-        m_orig = _bench_memory(_add_messages_original, left, right)
-        m_opt = _bench_memory(_add_messages_optimized, left, right)
-        reduction = (1 - m_opt / m_orig) * 100 if m_orig else 0.0
-        print(f"{label:<{COL}} {m_orig:>10,} {m_opt:>11,} {reduction:>9.1f}%")
-
-    print()
-    print("=" * 88)
-    print("Simulated long thread — 200 steps × 2 msgs appended per step")
-    print("=" * 88)
-    for name, fn in [
-        ("original", _add_messages_original),
-        ("optimized", _add_messages_optimized),
-    ]:
-        state: list = []
-        t0 = time.perf_counter()
-        for step in range(200):
-            new_msgs = [
-                HumanMessage(content=f"step {step} human", id=str(uuid.uuid4())),
-                AIMessage(content=f"step {step} ai", id=str(uuid.uuid4())),
-            ]
-            state = fn(state, new_msgs)
-        elapsed = (time.perf_counter() - t0) * 1_000
-        print(f"  {name:<12}  {elapsed:.2f} ms  ({len(state)} messages)")
-    print()
-
-
-# ── pytest entry-points ───────────────────────────────────────────────────────
-
-
-def test_add_messages_correctness():
-    """Optimised implementation must match original output for every scenario."""
-    for label, n_left, n_right, mode in SCENARIOS:
-        left, right = _make_inputs(n_left, n_right, mode)
-        expected = _add_messages_original(list(left), list(right))
-        actual = _add_messages_optimized(list(left), list(right))
-        assert len(actual) == len(expected), f"[{label}] length mismatch"
-        for a, b in zip(actual, expected):
-            assert type(a) is type(b), f"[{label}] type mismatch"
-            assert a.id == b.id, f"[{label}] id mismatch"
-            assert a.content == b.content, f"[{label}] content mismatch"
-
-
-def test_add_messages_benchmark(capsys):
-    run_benchmarks()
-    out = capsys.readouterr().out
-    assert "Speedup" in out
-    assert "Optimized" in out
-
-
-if __name__ == "__main__":
-    run_benchmarks()

libs/langgraph/tests/test_channels.py

@@ -7,7 +7,7 @@ from langgraph.checkpoint.base import DELTA_SENTINEL
 
 from langgraph._internal._typing import MISSING
 from langgraph.channels.binop import BinaryOperatorAggregate
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
 from langgraph.channels.untracked_value import UntrackedValue
@@ -127,7 +127,6 @@ def test_delta_channel_basic_two_steps() -> None:
     from langchain_core.messages import AIMessage, HumanMessage
     from langgraph.checkpoint.base import DELTA_SENTINEL
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
@@ -152,7 +151,6 @@ def test_delta_channel_from_checkpoint_writes_list() -> None:
     """replay_writes on a fresh channel replays through the operator."""
     from langchain_core.messages import AIMessage, HumanMessage
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     spec = DeltaChannel(add_messages)
@@ -174,7 +172,6 @@ def test_delta_channel_from_checkpoint_writes_list() -> None:
 def test_delta_channel_from_checkpoint_backwards_compat() -> None:
     from langchain_core.messages import HumanMessage
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     # Old BinaryOperatorAggregate checkpoint: plain list treated as backward compat
@@ -188,7 +185,6 @@ def test_delta_channel_overwrite() -> None:
     from langchain_core.messages import HumanMessage
     from langgraph.checkpoint.base import DELTA_SENTINEL
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
     from langgraph.types import Overwrite
 
@@ -207,7 +203,6 @@ def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
     from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     spec = DeltaChannel(add_messages)
@@ -241,7 +236,6 @@ def test_delta_channel_update_by_id_and_replay() -> None:
     """Updating a message by ID must round-trip correctly through writes replay."""
     from langchain_core.messages import HumanMessage
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     spec = DeltaChannel(add_messages)
@@ -270,7 +264,6 @@ def test_delta_channel_checkpoint_returns_sentinel() -> None:
     """checkpoint() always returns DELTA_SENTINEL regardless of state."""
     from langgraph.checkpoint.base import DELTA_SENTINEL
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph.message import add_messages
 
     ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
@@ -282,6 +275,124 @@ def test_delta_channel_checkpoint_returns_sentinel() -> None:
     assert ch.checkpoint() is DELTA_SENTINEL
 
 
+def test_delta_channel_snapshot_step_based() -> None:
+    """Snapshots fire on every Nth step regardless of whether the channel was written.
+
+    With snapshot_frequency=N, every Nth pregel step produces a _DeltaSnapshot
+    blob — even if the channel had no write that step (eager snapshot). This
+    bounds the ancestor walk to at most N steps on any read.
+    """
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+    from typing_extensions import TypedDict
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
+
+    # snapshot_frequency=5: snapshot every 5 pregel steps
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=5)]
+        other: str
+
+    def node_a(state: State) -> dict:
+        # writes to messages
+        i = len(state["messages"]) // 2
+        return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
+
+    def node_b(state: State) -> dict:
+        # writes ONLY to other, not messages — snapshot must still fire at step N
+        return {"other": "y"}
+
+    g = StateGraph(State)
+    g.add_node("a", node_a)
+    g.add_node("b", node_b)
+    g.add_edge(START, "a")
+    g.add_edge("a", "b")
+    saver = InMemorySaver()
+    graph = g.compile(checkpointer=saver)
+
+    config = {"configurable": {"thread_id": "t1"}}
+    for i in range(6):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")], "other": ""},
+            config,
+        )
+
+    # Confirm at least one snapshot blob exists for messages
+    msg_blob_values = [
+        saver.serde.loads_typed((type_tag, blob))
+        for k, (type_tag, blob) in saver.blobs.items()
+        if k[2] == "messages" and type_tag == "msgpack" and blob
+    ]
+    snapshots = [v for v in msg_blob_values if isinstance(v, _DeltaSnapshot)]
+    assert snapshots, "expected at least one _DeltaSnapshot blob for messages"
+
+    # Final state must be correct regardless of snapshot cadence
+    state = graph.get_state(config)
+    assert len(state.values["messages"]) == 12  # 6 human + 6 AI
+
+
+def test_delta_channel_snapshot_fires_even_when_not_written() -> None:
+    """Eager snapshot: _DeltaSnapshot stored at snapshot step even when the
+    channel had no write that step (node_b doesn't touch messages).
+    """
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+    from typing_extensions import TypedDict
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
+
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=3)]
+        tick: int
+
+    def writer(state: State) -> dict:
+        i = len(state["messages"]) // 2
+        return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
+
+    def ticker(state: State) -> dict:
+        # never writes messages
+        return {"tick": state["tick"] + 1}
+
+    g = StateGraph(State)
+    g.add_node("writer", writer)
+    g.add_node("ticker", ticker)
+    g.add_edge(START, "writer")
+    g.add_edge("writer", "ticker")
+    saver = InMemorySaver()
+    graph = g.compile(checkpointer=saver)
+
+    config = {"configurable": {"thread_id": "t1"}}
+    for i in range(5):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")], "tick": 0},
+            config,
+        )
+
+    # Count distinct message channel blob versions
+    msg_blobs = {
+        k: saver.serde.loads_typed((t, b))
+        for k, (t, b) in saver.blobs.items()
+        if k[2] == "messages" and t == "msgpack" and b
+    }
+    snapshots = {k: v for k, v in msg_blobs.items() if isinstance(v, _DeltaSnapshot)}
+    # There must be snapshots (ticker steps are snapshot steps too)
+    assert snapshots, (
+        "eager snapshots must fire even on steps where messages wasn't written"
+    )
+
+    # All get_state calls must return the correct accumulated value
+    state = graph.get_state(config)
+    assert len(state.values["messages"]) == 10  # 5 human + 5 AI
+
+
 def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     """InMemorySaver assembles writes from checkpoint_writes inside get_tuple."""
     from typing import Annotated
@@ -290,7 +401,6 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     from langgraph.checkpoint.memory import InMemorySaver
     from typing_extensions import TypedDict
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
 
@@ -325,11 +435,16 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
 
 
+# ---------------------------------------------------------------------------
+# Dict-reducer tests
+# ---------------------------------------------------------------------------
+
+
 def _delta_channel_with_type(operator, typ):
     """Build a DeltaChannel with an explicit type via the Annotated injection path."""
     from typing import Annotated
 
-    from langgraph.channels._delta import DeltaChannel
+    from langgraph.channels.delta import DeltaChannel
     from langgraph.graph.state import _get_channel
 
     return _get_channel("_test", Annotated[typ, DeltaChannel(operator)])
@@ -342,14 +457,12 @@ def test_delta_channel_dict_reducer_fresh_channel() -> None:
         return {**left, **right}
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
-    # Should be available (not raise EmptyChannelError) and start empty
     assert ch.is_available()
     assert ch.get() == {}
 
 
 def test_delta_channel_dict_reducer_basic_updates() -> None:
     """DeltaChannel with a dict reducer accumulates key/value pairs across steps."""
-    from langgraph.checkpoint.base import DELTA_SENTINEL
 
     def merge_dicts(left: dict, right: dict) -> dict:
         return {**left, **right}
@@ -386,7 +499,7 @@ def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
 
 
 def test_delta_channel_dict_reducer_with_deletions() -> None:
-    """Dict reducer that treats None values as deletions works end-to-end (deepagents pattern)."""
+    """Dict reducer that treats None values as deletions works end-to-end."""
 
     def merge_files(left: dict | None, right: dict) -> dict:
         if left is None:
@@ -400,15 +513,10 @@ def test_delta_channel_dict_reducer_with_deletions() -> None:
         return result
 
     ch = _delta_channel_with_type(merge_files, dict).from_checkpoint(MISSING)
-
     ch.update([{"file1.py": "content1", "file2.py": "content2"}])
-
-    # Delete file1, add file3
     ch.update([{"file1.py": None, "file3.py": "content3"}])
-
     assert ch.get() == {"file2.py": "content2", "file3.py": "content3"}
 
-    # Confirm writes reconstruction produces the same result
     spec = _delta_channel_with_type(merge_files, dict)
     ch2 = spec.from_checkpoint(DELTA_SENTINEL)
     ch2.replay_writes(
@@ -430,7 +538,6 @@ def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
     ch.update([{"a": 1}])
     ch.update([Overwrite({"b": 2, "c": 3})])
-
     assert ch.get() == {"b": 2, "c": 3}
 
 
@@ -454,17 +561,12 @@ def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
 
 
 def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
-    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`.
-
-    This is the shape the deepagents filesystem middleware uses for its
-    `files` field; without unwrapping NotRequired we'd fall through to `list`
-    and blow up on the first dict operator call.
-    """
+    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`."""
     from typing import Annotated
 
     from typing_extensions import NotRequired
 
-    from langgraph.channels._delta import DeltaChannel
+    from langgraph.channels.delta import DeltaChannel
     from langgraph.graph.state import _get_channel
 
     def merge_dicts(left: dict | None, right: dict) -> dict:
@@ -472,10 +574,7 @@ def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
             return dict(right)
         return {**left, **right}
 
-    annotation = Annotated[
-        NotRequired[dict[str, int]],
-        DeltaChannel(merge_dicts),
-    ]
+    annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
     ch = _get_channel("files", annotation).from_checkpoint(MISSING)
     assert ch.get() == {}
     ch.update([{"a": 1}])
@@ -484,17 +583,13 @@ def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
 
 
 def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
-    """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel.
-
-    Mirrors the deepagents filesystem pattern: `files: Annotated[dict, reducer]`
-    where the reducer merges dicts and treats None values as deletions.
-    """
+    """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel."""
     from typing import Annotated
 
     from langgraph.checkpoint.memory import InMemorySaver
     from typing_extensions import TypedDict
 
-    from langgraph.channels._delta import DeltaChannel
+    from langgraph.channels.delta import DeltaChannel
     from langgraph.graph import START, StateGraph
 
     def merge_files(left: dict | None, right: dict) -> dict:
@@ -528,12 +623,9 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
     for _ in range(3):
         graph.invoke({"files": {}}, config)
 
-    # Checkpoint stores only the sentinel — per-step writes live in checkpoint_writes.
     saved = saver.get_tuple(config)
     assert saved is not None
-    cv = saved.checkpoint["channel_values"]["files"]
-    assert cv is DELTA_SENTINEL
-
+    assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
     state = graph.get_state(config)
     assert state.values["files"] == {
         "/doc_1.txt": "content for turn 1",
@@ -541,7 +633,6 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
         "/doc_3.txt": "content for turn 3",
     }
 
-    # Deletion path must round-trip through writes replay.
     def delete_file(state: State) -> dict:
         return {"files": {"/doc_1.txt": None}}
 

libs/langgraph/tests/test_delta_channel_benchmark.py

@@ -1,22 +1,22 @@
-"""Benchmark: DeltaChannel vs BinaryOperatorAggregate storage and time.
+"""Benchmark: DeltaChannel snapshot_frequency — storage vs. read-depth tradeoff.
 
 Run directly:  python tests/test_delta_channel_benchmark.py
 Run via pytest: pytest tests/test_delta_channel_benchmark.py -s
 
-Simulates realistic multi-turn conversations with paragraph-length messages
-(~100 tokens each) scaling up to 1M-token-equivalent histories.
+Part 1 — baseline (original): DeltaChannel(inf) vs add_messages (BinOp).
+Part 2 — snapshot_frequency sweep: shows the storage/read-latency tradeoff
+  across frequencies [1, 5, 10, 50, inf] at scale.
 
-Token estimates: 1 token ≈ 4 chars; each turn ≈ 200 tokens (human + AI).
-A 1M-token conversation ≈ 5,000 turns of realistic messages.
-
-DeltaChannel stores only a zero-byte sentinel in checkpoint_blobs; the actual
-write data lives in checkpoint_writes (already stored there). Reconstruction
-walks the parent chain and replays writes through the operator — O(N) total
-storage vs O(N²) for plain add_messages.
+Key insight:
+  snapshot_frequency=inf  → O(N) storage, O(N) read depth (pure delta)
+  snapshot_frequency=N    → O(N²/N) storage, O(N) read depth bounded by freq
+  snapshot_frequency=1    → O(N²) storage, O(1) read depth (full snapshot)
 """
 
 from __future__ import annotations
 
+import contextlib
+import math
 import sys
 import time
 from typing import Annotated, Any
@@ -26,24 +26,15 @@ from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.checkpoint.memory import MemorySaver
 from typing_extensions import TypedDict
 
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import END, StateGraph
 from langgraph.graph.message import add_messages
 
-try:
-    from langgraph.checkpoint.sqlite import SqliteSaver
-
-    _SQLITE_AVAILABLE = True
-except ImportError:
-    _SQLITE_AVAILABLE = False
-
 try:
     from langgraph.checkpoint.postgres import PostgresSaver
 
     _POSTGRES_AVAILABLE = True
-    _POSTGRES_URI = (
-        "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
-    )
+    _POSTGRES_URI = "postgres://sydney_runkle@localhost:5441/postgres?sslmode=disable"
 except ImportError:
     _POSTGRES_AVAILABLE = False
 
@@ -126,6 +117,18 @@ class DeltaState(TypedDict):
     messages: Annotated[list, DeltaChannel(add_messages)]
 
 
+def _make_delta_state(snapshot_frequency: int | float) -> type:
+    """Create a TypedDict with DeltaChannel at the given snapshot_frequency."""
+    channel = DeltaChannel(add_messages, snapshot_frequency=snapshot_frequency)
+    # Use the functional TypedDict form so the Annotated type is stored as an
+    # already-evaluated object rather than a forward-reference string (which
+    # would fail when get_type_hints tries to resolve 'snapshot_frequency').
+    return TypedDict(  # type: ignore[return-value]
+        f"DeltaState_freq{snapshot_frequency}",
+        {"messages": Annotated[list, channel]},
+    )
+
+
 # ---------------------------------------------------------------------------
 # Graph factory
 # ---------------------------------------------------------------------------
@@ -169,9 +172,8 @@ def _run_turns(
     """Run n_turns conversation turns.
 
     Returns (write_elapsed_s, read_elapsed_s, total_blob_bytes).
-    blob_bytes is -1 for savers without in-memory blob stores (e.g. SQLite).
-    Read latency is measured as the time to invoke the graph with no new
-    messages after the full history is built — this forces state rehydration.
+    Read latency is the average of 5 get_state calls after the full history
+    is built — forces state rehydration including ancestor replay if needed.
     """
     graph = _make_graph(state_cls, checkpointer)
     config = {"configurable": {"thread_id": "bench"}}
@@ -184,16 +186,16 @@ def _run_turns(
         )
     write_elapsed = time.perf_counter() - t0
 
-    # Measure read/rehydration: get_state forces the channel to rebuild
     t1 = time.perf_counter()
     for _ in range(5):
         graph.get_state(config)
     read_elapsed = (time.perf_counter() - t1) / 5
 
-    if isinstance(graph.checkpointer, MemorySaver):
-        blob_bytes = _total_blob_bytes(graph.checkpointer)
-    else:
-        blob_bytes = -1
+    blob_bytes = (
+        _total_blob_bytes(graph.checkpointer)
+        if isinstance(graph.checkpointer, MemorySaver)
+        else -1
+    )
     return write_elapsed, read_elapsed, blob_bytes
 
 
@@ -206,7 +208,6 @@ def _fmt_bytes(n: int) -> str:
 
 
 def _approx_tokens(n_turns: int) -> str:
-    # ~100 tokens human + ~100 tokens AI per turn
     tokens = n_turns * 200
     if tokens >= 1_000_000:
         return f"~{tokens / 1_000_000:.1f}M tok"
@@ -216,74 +217,56 @@ def _approx_tokens(n_turns: int) -> str:
 
 
 # ---------------------------------------------------------------------------
-# Benchmark matrix
+# Checkpointer factories
 # ---------------------------------------------------------------------------
 
-# Turn counts chosen to demonstrate O(N²) vs O(N) storage growth without running too long.
-# Extrapolation: 5,000 turns × ~200 tokens/turn ≈ 1M tokens (Claude's full context window).
-TURN_COUNTS = [10, 25, 50, 100, 500]
 
-# Deep-thread counts where add_messages blob storage would exceed 1 GB;
-# only DeltaChannel runs here.
-DELTA_ONLY_TURN_COUNTS = [1000]
+@contextlib.contextmanager
+def _pg_saver(thread_id: str = "bench"):
+    """Context manager that yields a fresh PostgresSaver and cleans up after."""
+    with PostgresSaver.from_conn_string(_POSTGRES_URI) as saver:
+        saver.setup()
+        with saver._cursor() as cur:
+            for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
+                cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
+        yield saver
+        with saver._cursor() as cur:
+            for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
+                cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
 
 
-def _checkpointer_factories() -> list[tuple[str, Any]]:
-    """Return (label, context_manager_or_none) pairs for available checkpointers."""
-    return [("InMemory", None)]
-
-
-def run_benchmark() -> None:
-    print()
-    print(
-        "DeltaChannel vs add_messages (BinaryOperatorAggregate) — checkpoint storage & latency"
-    )
-    print("Simulating realistic multi-turn conversations up to ~1M-token histories")
-    print("(5,000 turns × ~200 tokens/turn ≈ 1M tokens — Claude's full context window)")
-    print()
-
-    checkpointers: list[tuple[str, Any]] = [("InMemory", None)]
+def _checkpointers() -> list[tuple[str, Any]]:
+    """Return (label, saver_or_None) pairs for available checkpointers."""
+    result: list[tuple[str, Any]] = [("InMemory", None)]
     if _POSTGRES_AVAILABLE:
         try:
             import psycopg
 
             psycopg.connect(_POSTGRES_URI).close()
-            checkpointers.append(("Postgres (plain SELECT)", "postgres"))
+            result.append(("Postgres", "postgres"))
         except Exception:
             pass
-
-    for cp_label, cp_hint in checkpointers:
-        print(f"--- Checkpointer: {cp_label} ---")
-        _run_benchmark_for_checkpointer(cp_hint)
+    return result
 
 
-def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
-    import contextlib
-    import tempfile
+# ---------------------------------------------------------------------------
+# Part 1: baseline DeltaChannel(inf) vs add_messages
+# ---------------------------------------------------------------------------
+
+BASELINE_TURN_COUNTS = [10, 25, 50, 100, 500]
+DELTA_ONLY_TURN_COUNTS = [1000]
+
+
+def _run_baseline_for_checkpointer(cp_label: str, cp_hint: Any) -> None:
+    W = 72
 
-    @contextlib.contextmanager
     def _make_saver():
         if cp_hint is None:
-            yield None
-        elif cp_hint == "postgres":
-            with PostgresSaver.from_conn_string(_POSTGRES_URI) as saver:
-                saver.setup()
-                with saver._cursor() as cur:
-                    cur.execute("DELETE FROM checkpoints WHERE thread_id = 'bench'")
-                    cur.execute(
-                        "DELETE FROM checkpoint_blobs WHERE thread_id = 'bench'"
-                    )
-                    cur.execute(
-                        "DELETE FROM checkpoint_writes WHERE thread_id = 'bench'"
-                    )
-                yield saver
-        else:
-            with tempfile.NamedTemporaryFile(suffix=".db") as f:
-                with SqliteSaver.from_conn_string(f.name) as saver:
-                    yield saver
+            return contextlib.nullcontext(None)
+        return _pg_saver()
 
     rows: list[tuple[int, Any, Any, Any, Any, Any, Any]] = []
-    for turns in TURN_COUNTS:
+    for turns in BASELINE_TURN_COUNTS:
         with _make_saver() as saver:
             b_wt, b_rt, b_bytes = _run_turns(turns, BinaryState, saver)
         with _make_saver() as saver:
@@ -294,26 +277,19 @@ def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
             d_wt, d_rt, d_bytes = _run_turns(turns, DeltaState, saver)
         rows.append((turns, None, d_bytes, None, d_rt, None, d_wt))
 
-    # ── Table 1: Storage ─────────────────────────────────────────────────────
-    W = 64
-    print("Storage (checkpoint blob bytes)")
-    print("=" * W)
-    print(
-        f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}  "
-        f"{'savings':>8}"
-    )
-    print("-" * W)
-
     def _bytes_or_na(v: Any) -> str:
-        if v is None:
-            return "n/a"
-        if v < 0:
+        if v is None or v < 0:
             return "n/a"
         return _fmt_bytes(v)
 
     def _ms_or_na(v: Any) -> str:
         return "n/a" if v is None else f"{v * 1000:.1f}ms"
 
+    print(f"\n  [{cp_label}] Storage (blob bytes)")
+    print(
+        f"  {'turns':>6}  {'ctx':>10}  {'add_msgs':>12}  {'delta(inf)':>12}  {'savings':>8}"
+    )
+    print("  " + "-" * (W - 2))
     for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
         if b_bytes is None or b_bytes < 0 or d_bytes is None or d_bytes < 0:
             ratio_str = "n/a"
@@ -321,65 +297,132 @@ def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
             ratio = b_bytes / d_bytes if d_bytes else float("inf")
             ratio_str = f"{ratio:.0f}x"
         print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
-            f"{_bytes_or_na(b_bytes):>12}  {_bytes_or_na(d_bytes):>12}  "
-            f"{ratio_str:>8}"
+            f"  {turns:>6}  {_approx_tokens(turns):>10}  "
+            f"{_bytes_or_na(b_bytes):>12}  {_bytes_or_na(d_bytes):>12}  {ratio_str:>8}"
         )
-    print("=" * W)
-    print()
 
-    # ── Table 2: Read latency ─────────────────────────────────────────────────
-    print("Read latency (avg of 5 get_state calls)")
-    print("=" * W)
-    print(f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}")
-    print("-" * W)
+    print(f"\n  [{cp_label}] Read latency (avg of 5 get_state calls)")
+    print(f"  {'turns':>6}  {'ctx':>10}  {'add_msgs':>12}  {'delta(inf)':>12}")
+    print("  " + "-" * (W - 2))
     for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
         print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
+            f"  {turns:>6}  {_approx_tokens(turns):>10}  "
             f"{_ms_or_na(b_rt):>12}  {_ms_or_na(d_rt):>12}"
         )
-    print("=" * W)
+
+
+def run_baseline_benchmark() -> None:
     print()
-
-    # ── Table 3: Per-invoke latency (total write_elapsed / turns) ─────────────
-    print("Per-invoke latency (total graph.invoke time / turns)")
-    print("=" * W)
-    print(f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}")
-    print("-" * W)
-    for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
-
-        def _per(wt: Any) -> str:
-            if wt is None:
-                return "n/a"
-            return f"{(wt / turns) * 1000:.1f}ms"
-
-        print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
-            f"{_per(b_wt):>12}  {_per(d_wt):>12}"
-        )
-    print("=" * W)
-    print()
-
-    print("Legend:")
-    print("  add_msgs  = Annotated[list, add_messages]  — O(N²) storage")
-    print("  delta     = Annotated[list, DeltaChannel(add_messages)]  — O(N) storage")
+    print("Part 1 — DeltaChannel(inf) vs add_messages: storage & latency")
+    print("=" * 72)
+    for cp_label, cp_hint in _checkpointers():
+        _run_baseline_for_checkpointer(cp_label, cp_hint)
     print()
 
 
 # ---------------------------------------------------------------------------
-# Pytest entry point
+# Part 2: snapshot_frequency sweep
+# ---------------------------------------------------------------------------
+
+# Frequencies to test. 1 = always snapshot (like BinOp), inf = pure delta.
+SNAPSHOT_FREQUENCIES: list[int | float] = [1, 5, 10, 50, math.inf]
+
+# Turn counts for the sweep — high enough to show storage divergence.
+SWEEP_TURN_COUNTS = [50, 100, 500]
+
+
+def _freq_label(freq: int | float) -> str:
+    if freq == math.inf:
+        return "inf"
+    return str(int(freq))
+
+
+def _run_sweep_for_checkpointer(cp_label: str, cp_hint: Any) -> None:
+    def _make_saver():
+        if cp_hint is None:
+            return contextlib.nullcontext(None)
+        return _pg_saver()
+
+    # Collect results: {turns: {freq_label: (write_s, read_s, bytes)}}
+    results: dict[int, dict[str, tuple[float, float, int]]] = {}
+    for turns in SWEEP_TURN_COUNTS:
+        results[turns] = {}
+        for freq in SNAPSHOT_FREQUENCIES:
+            state_cls = _make_delta_state(freq)
+            with _make_saver() as saver:
+                wt, rt, bb = _run_turns(turns, state_cls, saver)
+            results[turns][_freq_label(freq)] = (wt, rt, bb)
+
+    freq_labels = [_freq_label(f) for f in SNAPSHOT_FREQUENCIES]
+    col_w = 12
+
+    header = f"  {'turns':>6}  {'ctx':>10}" + "".join(
+        f"  {f'freq={freq_label}':>{col_w}}" for freq_label in freq_labels
+    )
+
+    print(f"\n  [{cp_label}] Storage (blob bytes) — lower is better")
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            _, _, bb = results[turns][label]
+            row += f"  {_fmt_bytes(bb) if bb >= 0 else 'n/a':>{col_w}}"
+        print(row)
+
+    print(f"\n  [{cp_label}] Read latency (avg of 5 get_state) — lower is better")
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            _, rt, _ = results[turns][label]
+            row += f"  {f'{rt * 1000:.1f}ms':>{col_w}}"
+        print(row)
+
+    print(
+        f"\n  [{cp_label}] Per-invoke write latency (total / turns) — lower is better"
+    )
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            wt, _, _ = results[turns][label]
+            row += f"  {f'{(wt / turns) * 1000:.1f}ms':>{col_w}}"
+        print(row)
+
+
+def run_snapshot_freq_benchmark() -> None:
+    print()
+    print("Part 2 — DeltaChannel snapshot_frequency sweep")
+    print("Lower freq → fewer snapshots → less storage but deeper read replay")
+    print("=" * 80)
+    for cp_label, cp_hint in _checkpointers():
+        _run_sweep_for_checkpointer(cp_label, cp_hint)
+    print()
+    print("Legend:")
+    print(
+        "  freq=1    snapshot every write (full blob always — same as add_messages / BinOp)"
+    )
+    print("  freq=N    snapshot every N writes; read walks at most N ancestor writes")
+    print("  freq=inf  pure delta; read walks entire ancestor chain")
+    print()
+
+
+# ---------------------------------------------------------------------------
+# Pytest entry points
 # ---------------------------------------------------------------------------
 
 
 @pytest.mark.skip(
     reason="slow benchmark — run manually with: python tests/test_delta_channel_benchmark.py"
 )
-def test_delta_channel_benchmark(capsys: Any) -> None:
-    """Storage grows O(N²) for add_messages, O(N) for DeltaChannel."""
+def test_delta_channel_baseline_benchmark(capsys: Any) -> None:
+    """DeltaChannel(inf) uses less storage than add_messages at scale."""
     with capsys.disabled():
-        run_benchmark()
+        run_baseline_benchmark()
 
-    # Correctness assertion: DeltaChannel must use less storage at scale.
     for turns in [25, 50]:
         _, _, b_bytes = _run_turns(turns, BinaryState)
         _, _, d_bytes = _run_turns(turns, DeltaState)
@@ -389,10 +432,41 @@ def test_delta_channel_benchmark(capsys: Any) -> None:
         )
 
 
+@pytest.mark.skip(
+    reason="slow benchmark — run manually with: python tests/test_delta_channel_benchmark.py"
+)
+def test_snapshot_freq_benchmark(capsys: Any) -> None:
+    """snapshot_frequency trades storage for bounded read depth."""
+    with capsys.disabled():
+        run_snapshot_freq_benchmark()
+
+    # Correctness: results at all frequencies should agree on final state.
+    n_turns = 20
+    states: dict[str, list] = {}
+    for freq in SNAPSHOT_FREQUENCIES:
+        state_cls = _make_delta_state(freq)
+        graph = _make_graph(state_cls)
+        config = {"configurable": {"thread_id": "correctness"}}
+        for i in range(n_turns):
+            graph.invoke(
+                {"messages": [HumanMessage(content=_human_content(i), id=f"h{i}")]},
+                config,
+            )
+        state = graph.get_state(config)
+        states[_freq_label(freq)] = [m.id for m in state.values["messages"]]
+
+    ref = states["inf"]
+    for label, msg_ids in states.items():
+        assert msg_ids == ref, (
+            f"freq={label} produced different message IDs than freq=inf"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Script entry point
 # ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    run_benchmark()
+    run_baseline_benchmark()
+    run_snapshot_freq_benchmark()
     sys.exit(0)

libs/langgraph/tests/test_delta_channel_migration.py

@@ -50,7 +50,7 @@ from langgraph.checkpoint.memory import InMemorySaver
 from typing_extensions import TypedDict
 
 from langgraph.channels.binop import BinaryOperatorAggregate
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import END, START, StateGraph
 
 pytestmark = pytest.mark.anyio

libs/langgraph/tests/test_messages_state.py

@@ -5,7 +5,6 @@ import langchain_core
 import pytest
 from langchain_core.messages import (
     AIMessage,
-    AIMessageChunk,
     AnyMessage,
     HumanMessage,
     RemoveMessage,
@@ -339,123 +338,6 @@ def test_remove_all_messages():
     ]
 
 
-def test_fast_path_preserves_format_openai():
-    """Pure-append fast path must still apply the `langchain-openai` formatter."""
-    left = [HumanMessage(content="prior", id="1")]
-    right = [
-        AIMessage(
-            content=[
-                {
-                    "type": "tool_use",
-                    "name": "foo",
-                    "input": {"bar": "baz"},
-                    "id": "t1",
-                }
-            ],
-            id="2",
-        )
-    ]
-    result = add_messages(left, right, format="langchain-openai")
-    assert isinstance(result[0], HumanMessage)
-    assert result[0].content == "prior"
-    assert isinstance(result[1], AIMessage)
-    # formatter collapses the tool_use content block into `tool_calls`
-    assert result[1].content == ""
-    assert len(result[1].tool_calls) == 1
-    assert result[1].tool_calls[0]["name"] == "foo"
-    assert result[1].tool_calls[0]["args"] == {"bar": "baz"}
-    assert result[1].tool_calls[0]["id"] == "t1"
-
-
-def test_fast_path_rejects_invalid_format():
-    """Pure-append fast path must validate the `format` arg like the slow path."""
-    left = [HumanMessage(content="prior", id="1")]
-    right = [AIMessage(content="new", id="2")]
-    with pytest.raises(ValueError, match="Unrecognized format="):
-        add_messages(left, right, format="bogus")  # type: ignore[arg-type]
-
-
-def test_left_starting_with_chunk_is_normalized():
-    """Opt-1 guard: a `BaseMessageChunk` at left[0] must trigger full conversion."""
-    chunk = AIMessageChunk(content="chunk", id="c1")
-    result = add_messages([chunk], [HumanMessage(content="h", id="h1")])
-    assert len(result) == 2
-    # chunk must be converted to a non-chunk message
-    assert type(result[0]).__name__ == "AIMessage"
-    assert result[0].id == "c1"
-    assert result[1].id == "h1"
-
-
-def test_left_as_dicts_is_normalized():
-    """Opt-1 guard: dicts at left[0] must trigger full conversion."""
-    left = [{"role": "user", "content": "hi", "id": "d1"}]
-    right = [AIMessage(content="reply", id="a1")]
-    result = add_messages(left, right)
-    assert len(result) == 2
-    assert isinstance(result[0], HumanMessage)
-    assert result[0].id == "d1"
-    assert result[0].content == "hi"
-
-
-def test_left_as_tuples_is_normalized():
-    """Opt-1 guard: tuple-form messages must trigger full conversion."""
-    left = [("user", "hi")]
-    right = [AIMessage(content="reply", id="a1")]
-    result = add_messages(left, right)
-    assert len(result) == 2
-    assert isinstance(result[0], HumanMessage)
-    # id is auto-assigned
-    assert isinstance(result[0].id, str) and UUID(result[0].id, version=4)
-
-
-def test_left_first_msg_missing_id_is_normalized():
-    """Opt-1 guard: a BaseMessage without an id at left[0] falls to the else branch."""
-    left = [HumanMessage(content="hi")]  # no id
-    right = [AIMessage(content="reply", id="a1")]
-    result = add_messages(left, right)
-    assert len(result) == 2
-    # left's id must have been auto-assigned
-    assert isinstance(result[0].id, str) and UUID(result[0].id, version=4)
-
-
-def test_duplicate_ids_in_right_with_nonempty_left():
-    """Opt-2 guard: intra-right duplicate ids must take slow path (dedup kept)."""
-    left = [HumanMessage(content="prior", id="1")]
-    right = [
-        AIMessage(content="first", id="2"),
-        AIMessage(content="second", id="2"),
-    ]
-    result = add_messages(left, right)
-    assert len(result) == 2
-    assert result[0].id == "1"
-    assert result[1].id == "2"
-    assert result[1].content == "second"
-
-
-def test_right_with_none_ids_pure_append():
-    """Fast path still correct when right entries start with id=None (fresh uuids assigned)."""
-    left = [HumanMessage(content="prior", id="1")]
-    right = [AIMessage(content="a"), AIMessage(content="b")]
-    result = add_messages(left, right)
-    assert len(result) == 3
-    assert result[0].id == "1"
-    for m in result[1:]:
-        assert isinstance(m.id, str) and UUID(m.id, version=4)
-    # fresh uuids must be distinct
-    assert result[1].id != result[2].id
-
-
-def test_fast_path_returns_fresh_list():
-    """Fast path must return a new list object (not mutate or alias left)."""
-    left = [HumanMessage(content="prior", id="1")]
-    right = [AIMessage(content="new", id="2")]
-    result = add_messages(left, right)
-    assert result is not left
-    # left must be untouched
-    assert len(left) == 1
-    assert left[0].id == "1"
-
-
 def test_push_messages_in_graph():
     class MessagesState(TypedDict):
         messages: Annotated[list[AnyMessage], add_messages]

libs/langgraph/tests/test_pregel.py

@@ -41,6 +41,7 @@ from typing_extensions import NotRequired, TypedDict
 
 from langgraph._internal._constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL
 from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
@@ -9407,7 +9408,6 @@ async def test_delta_channel_end_to_end_inmemory() -> None:
     from langchain_core.messages import AIMessage, HumanMessage
     from langgraph.checkpoint.memory import InMemorySaver
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
 
@@ -9449,7 +9449,6 @@ async def test_delta_channel_time_travel() -> None:
     from langchain_core.messages import AIMessage, HumanMessage
     from langgraph.checkpoint.memory import InMemorySaver
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
 
@@ -9507,7 +9506,6 @@ async def test_delta_channel_remove_message_end_to_end() -> None:
     from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
     from langgraph.checkpoint.memory import InMemorySaver
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
 
@@ -9554,7 +9552,6 @@ async def test_delta_channel_update_by_id_end_to_end() -> None:
     from langchain_core.messages import HumanMessage
     from langgraph.checkpoint.memory import InMemorySaver
 
-    from langgraph.channels._delta import DeltaChannel
     from langgraph.graph import START, StateGraph
     from langgraph.graph.message import add_messages
 
@@ -9587,3 +9584,90 @@ async def test_delta_channel_update_by_id_end_to_end() -> None:
     assert "h1" in ids  # h1 persists (updated, not duplicated)
     assert "h2" in ids
     assert ids.count("h1") == 1, "h1 must not be duplicated"
+
+
+async def test_delta_channel_write_flushed_before_put() -> None:
+    """checkpoint_writes are flushed synchronously before put when DELTA_SENTINEL
+    is present, ensuring writes are durable before the sentinel blob is committed.
+
+    We verify this by intercepting put_writes and put calls and confirming
+    put_writes always completes before put is called for sentinel checkpoints.
+    """
+    import threading
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.base import DELTA_SENTINEL
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
+
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(add_messages)]
+
+    def respond(state: State) -> dict:
+        i = len(state["messages"])
+        return {"messages": [AIMessage(content=f"r{i}", id=f"ai{i}")]}
+
+    order: list[str] = []
+    lock = threading.Lock()
+    original_put_writes = InMemorySaver.put_writes
+    original_put = InMemorySaver.put
+
+    def tracked_put_writes(self, config, writes, task_id, task_path=""):
+        result = original_put_writes(self, config, writes, task_id, task_path)
+        with lock:
+            order.append("put_writes")
+        return result
+
+    def tracked_put(self, config, checkpoint, metadata, new_versions):
+        # Check if this checkpoint has any DELTA_SENTINEL blobs
+        has_sentinel = any(
+            v is DELTA_SENTINEL for v in checkpoint.get("channel_values", {}).values()
+        )
+        if has_sentinel:
+            with lock:
+                order.append("put_sentinel")
+        else:
+            with lock:
+                order.append("put_snapshot")
+        return original_put(self, config, checkpoint, metadata, new_versions)
+
+    InMemorySaver.put_writes = tracked_put_writes
+    InMemorySaver.put = tracked_put
+    try:
+        builder = StateGraph(State)
+        builder.add_node("respond", respond)
+        builder.add_edge(START, "respond")
+        saver = InMemorySaver()
+        graph = builder.compile(checkpointer=saver)
+        config = {"configurable": {"thread_id": "flush-test"}}
+
+        for i in range(3):
+            graph.invoke(
+                {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")]}, config
+            )
+
+        # For every sentinel put, all preceding put_writes must already be in order
+        for i, event in enumerate(order):
+            if event == "put_sentinel":
+                # All put_writes before this index must appear before this sentinel
+                preceding = order[:i]
+                assert "put_writes" in preceding, (
+                    f"put_sentinel at index {i} had no preceding put_writes: {order}"
+                )
+                # And the most recent put_writes must come before this sentinel
+                last_write_idx = max(
+                    j for j, e in enumerate(order[:i]) if e == "put_writes"
+                )
+                assert last_write_idx < i, (
+                    f"put_writes at {last_write_idx} not before put_sentinel at {i}"
+                )
+    finally:
+        InMemorySaver.put_writes = original_put_writes
+        InMemorySaver.put = original_put
+
+    # Final state must still be correct
+    state = graph.get_state(config)
+    assert len(state.values["messages"]) == 6  # 3 human + 3 AI

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -82,7 +82,6 @@ from langchain_core.tools.base import (
     _is_injected_arg_type,
     get_all_basemodel_annotations,
 )
-from langgraph._internal._constants import CONF, CONFIG_KEY_READ
 from langgraph._internal._runnable import RunnableCallable
 from langgraph.errors import GraphBubbleUp
 from langgraph.graph.message import REMOVE_ALL_MESSAGES
@@ -801,7 +800,7 @@ class ToolNode(RunnableCallable):
         # Construct ToolRuntime instances at the top level for each tool call
         tool_runtimes = []
         for call, cfg in zip(tool_calls, config_list, strict=False):
-            state = self._extract_state(input, cfg)
+            state = self._extract_state(input)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -836,7 +835,7 @@ class ToolNode(RunnableCallable):
         # Construct ToolRuntime instances at the top level for each tool call
         tool_runtimes = []
         for call, cfg in zip(tool_calls, config_list, strict=False):
-            state = self._extract_state(input, cfg)
+            state = self._extract_state(input)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -1274,37 +1273,18 @@ class ToolNode(RunnableCallable):
         return None
 
     def _extract_state(
-        self,
-        input: list[AnyMessage] | dict[str, Any] | BaseModel,
-        config: RunnableConfig,
+        self, input: list[AnyMessage] | dict[str, Any] | BaseModel
     ) -> list[AnyMessage] | dict[str, Any] | BaseModel:
-        """Extract state from input.
+        """Extract state from input, handling ToolCallWithContext if present.
 
-        Three input shapes:
+        Args:
+            input: The input which may be raw state or ToolCallWithContext.
 
-        - `ToolCallWithContext` dict — legacy Send payload carrying an inlined
-          state snapshot; return `input["state"]`.
-        - list of `ToolCall` dicts — new Send payload with no inlined state;
-          hydrate state from channels via `CONFIG_KEY_READ`.
-        - regular graph state (dict/list/BaseModel) — return `input` as-is.
+        Returns:
+            The actual state to pass to wrap_tool_call wrappers.
         """
         if isinstance(input, dict) and input.get("__type") == "tool_call_with_context":
             return input["state"]
-        if (
-            isinstance(input, list)
-            and input
-            and isinstance(input[-1], dict)
-            and input[-1].get("type") == "tool_call"
-        ):
-            read = config.get(CONF, {}).get(CONFIG_KEY_READ)
-            if read is None:
-                return {}
-            # Pregel installs CONFIG_KEY_READ as
-            # `functools.partial(local_read, scratchpad, channels, managed, task)`.
-            # Match the previous inlined-state contract by reading channels only;
-            # managed values have their own injection path (`ToolRuntime.context`).
-            channels = read.args[1]
-            return cast("dict[str, Any]", read(list(channels), False))
         return input
 
     def _inject_tool_args(

libs/prebuilt/tests/test_on_tool_call.py

@@ -1320,98 +1320,6 @@ async def test_state_extraction_with_tool_call_with_context_async() -> None:
     assert "tool_call" not in state_seen[0]
 
 
-def _config_with_channel_read(
-    channel_values: dict[str, object],
-    store: BaseStore | None = None,
-) -> RunnableConfig:
-    """Build a config that mimics `CONFIG_KEY_READ` as Pregel installs it.
-
-    Pregel always installs a `functools.partial(local_read, scratchpad,
-    channels, managed, task)`, and `ToolNode` introspects that partial to
-    learn channel names. The stub matches the shape: partial whose second and
-    third positional args are `channels` and `managed` mappings.
-    """
-    import functools
-
-    channels_stub = {k: None for k in channel_values}
-    managed_stub: dict[str, object] = {}
-
-    # Shape matches pregel's real partial:
-    # functools.partial(local_read, scratchpad, channels, managed, task)
-    def _read(scratchpad, channels, managed, task, select, fresh):  # noqa: ARG001
-        if isinstance(select, str):
-            return channel_values[select]
-        return {k: channel_values[k] for k in select if k in channel_values}
-
-    read = functools.partial(_read, None, channels_stub, managed_stub, None)
-    cfg = _create_config_with_runtime(store)
-    cfg["configurable"]["__pregel_read"] = read
-    return cfg
-
-
-def test_list_form_send_hydrates_state_from_channel_read() -> None:
-    """Send('tools', [tool_call]) with no inlined state should hydrate
-    ToolRuntime.state from CONFIG_KEY_READ (full state read)."""
-    state_seen = []
-
-    def state_inspector_handler(
-        request: ToolCallRequest,
-        execute: Callable[[ToolCallRequest], ToolMessage | Command],
-    ) -> ToolMessage | Command:
-        state_seen.append(request.state)
-        return execute(request)
-
-    channel_values = {
-        "messages": [AIMessage("from channels")],
-        "files": {"/a.md": "body"},
-    }
-
-    tool_node = ToolNode([add], wrap_tool_call=state_inspector_handler)
-
-    tool_call: ToolCall = {
-        "name": "add",
-        "args": {"a": 1, "b": 2},
-        "id": "call_1",
-        "type": "tool_call",
-    }
-
-    tool_node.invoke([tool_call], config=_config_with_channel_read(channel_values))
-
-    assert len(state_seen) == 1
-    got = state_seen[0]
-    assert got == channel_values
-    assert "messages" in got and "files" in got
-
-
-async def test_list_form_send_hydrates_state_async() -> None:
-    state_seen = []
-
-    def state_inspector_handler(
-        request: ToolCallRequest,
-        execute: Callable[[ToolCallRequest], ToolMessage | Command],
-    ) -> ToolMessage | Command:
-        state_seen.append(request.state)
-        return execute(request)
-
-    channel_values = {"messages": [AIMessage("from channels")], "files": {}}
-
-    tool_node = ToolNode([add], wrap_tool_call=state_inspector_handler)
-
-    tool_call: ToolCall = {
-        "name": "add",
-        "args": {"a": 1, "b": 2},
-        "id": "call_1",
-        "type": "tool_call",
-    }
-
-    await tool_node.ainvoke(
-        [tool_call], config=_config_with_channel_read(channel_values)
-    )
-
-    assert len(state_seen) == 1
-    assert state_seen[0] == channel_values
-
-
 def test_tool_call_request_is_frozen() -> None:
     """Test that ToolCallRequest raises deprecation warnings on direct attribute reassignment."""
     tool_call: ToolCall = {"name": "add", "args": {"a": 1, "b": 2}, "id": "call_1"}