feat(prebuilt): hydrate ToolNode state from channels via CONFIG_KEY_READ

When ToolNode receives a bare `[tool_call]` list via the Send API (the
new dispatch shape that create_agent uses after langchain-ai/langchain
drops the ToolCallWithContext wrapper), hydrate ToolRuntime.state from
the current channel values instead of requiring the dispatcher to
inline the full agent state dict in the Send payload.

Implementation stays entirely in tool_node:

- Pregel installs CONFIG_KEY_READ as
  `functools.partial(local_read, scratchpad, channels, managed, task)`.
  Introspect the partial's positional args to learn channel + managed
  names, then read them all via `ChannelRead.do_read` with an explicit
  list. No changes to the pregel read machinery.
- Gracefully falls back to {} when invoked outside a Pregel context
  (e.g. direct ToolNode(...).invoke(...) from test harnesses).
- Legacy ToolCallWithContext path is preserved for external dispatchers.

Co-Authored-By: Claude Opus 4.7 (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, _DeltaSnapshot
+from langgraph.checkpoint.serde.types import TASKS
 from psycopg.types.json import Jsonb
 
 MetadataInput = dict[str, Any] | None
@@ -281,29 +281,13 @@ 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,7 +28,6 @@ from langgraph.checkpoint.serde.types import (
     RESUME,
     SCHEDULED,
     ChannelProtocol,
-    _DeltaSnapshot,
 )
 
 V = TypeVar("V", int, float, str)
@@ -546,16 +545,6 @@ 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:
@@ -589,15 +578,8 @@ 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,7 +27,6 @@ from langgraph.checkpoint.base import (
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 logger = logging.getLogger(__name__)
 
@@ -186,31 +185,8 @@ 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:
-                            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.
+                            # Pre-delta snapshot terminator. Skip this
+                            # ancestor's writes — the blob subsumes them.
                             collected.reverse()
                             return _ChannelWritesHistory(
                                 seed=blob_value, writes=collected
@@ -525,101 +501,6 @@ 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,12 +33,7 @@ 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,
-    _DeltaSentinel,
-    _DeltaSnapshot,
-)
+from langgraph.checkpoint.serde.types import DELTA_SENTINEL, SendProtocol
 from langgraph.store.base import Item
 
 if TYPE_CHECKING:
@@ -256,6 +251,8 @@ 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):
@@ -282,6 +279,8 @@ 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:
@@ -297,16 +296,10 @@ 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 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
+    if hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,
             _msgpack_enc(
@@ -620,15 +613,7 @@ def _create_msgpack_ext_hook(
         return False
 
     def ext_hook(code: int, data: bytes) -> Any:
-        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:
+        if 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,7 +1,6 @@
 from collections.abc import Sequence
 from typing import (
     Any,
-    NamedTuple,
     Protocol,
     TypeVar,
     runtime_checkable,
@@ -34,20 +33,6 @@ 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,148 +663,3 @@ 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,7 +1,6 @@
 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 (
@@ -21,7 +20,6 @@ __all__ = (
     "UntrackedValue",
     "EphemeralValue",
     "BinaryOperatorAggregate",
-    "DeltaChannel",
     "NamedBarrierValue",
     "NamedBarrierValueAfterFinish",
     # topics

libs/langgraph/langgraph/channels/_delta.py

@@ -5,19 +5,12 @@ 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.errors import (
-    EmptyChannelError,
-    ErrorCode,
-    InvalidUpdateError,
-    create_error_message,
-)
-from langgraph.types import Overwrite
+from langgraph.channels.binop import _get_overwrite
+from langgraph.errors import EmptyChannelError
 
 __all__ = ("DeltaChannel",)
 
@@ -29,51 +22,48 @@ 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]):
-    """Fold-reducer channel with configurable snapshot cadence.
+    """Experimental — private API, subject to change or removal without notice.
 
-    `snapshot_frequency=None` (default): pure delta — stores only
-    `DELTA_SENTINEL` in checkpoint blobs; reads replay all ancestor writes.
+    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=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.
+    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.
 
-    Parameters:
-        operator: Binary reducer `(Value, Value) -> Value`.
-        snapshot_frequency: Every Nth pregel step writes a snapshot blob.
-            `None` (default) = pure delta, never snapshot.
+    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)]
     """
 
-    __slots__ = ("value", "operator", "snapshot_frequency")
-    value: Value | Any
+    __slots__ = (
+        "value",
+        "operator",
+    )
 
     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>"
@@ -89,27 +79,20 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
     def UpdateType(self) -> Any:
         return self.typ
 
-    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__)
+    def copy(self) -> Self:
+        new: DeltaChannel[Value] = DeltaChannel(self.operator)
         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 (
@@ -121,24 +104,38 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         return self.operator(base, write)
 
     def from_checkpoint(self, checkpoint: Any) -> Self:
-        """Initialize from a stored blob or sentinel.
+        """Initialize from a seed value.
 
-        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.
+        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.
         """
-        new = self._clone_empty()
+        new: DeltaChannel[Value] = DeltaChannel(self.operator)
+        new.typ = self.typ
+        new.key = self.key
         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 ancestor writes oldest→newest into current value."""
+        """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.
+        """
         for _, _, value in writes:
             self.value = self._apply_write(self.value, value)
 
@@ -150,6 +147,12 @@ 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,
@@ -157,6 +160,7 @@ 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
@@ -170,12 +174,4 @@ 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,63 +184,97 @@ def add_messages(
         ```
 
     """
-    remove_all_idx = None
-    # coerce to list
+    # 1. Coerce scalars to lists.
     if not isinstance(left, list):
         left = [left]  # type: ignore[assignment]
     if not isinstance(right, list):
         right = [right]  # type: ignore[assignment]
-    # coerce to message
-    left = [
-        message_chunk_to_message(cast(BaseMessageChunk, m))
-        for m in convert_to_messages(left)
-    ]
-    right = [
+    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] = [
         message_chunk_to_message(cast(BaseMessageChunk, m))
         for m in convert_to_messages(right)
     ]
-    # assign missing ids
-    for m in left:
+    remove_all_idx: int | None = None
+    has_remove = False
+    for idx, m in enumerate(right_msgs):
         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 isinstance(m, RemoveMessage):
+            has_remove = True
+            if 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[remove_all_idx + 1 :]
+        return right_msgs[remove_all_idx + 1 :]
 
-    # 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)
+    # 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
             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)
-    merged = [m for m in merged if m.id not in ids_to_remove]
+                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":
-        merged = _format_messages(merged)
-    elif format:
+        return _format_messages(merged)
+    if 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 = None) -> int:
+def increment(current: int | None, channel: 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,23 +1,19 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
+from collections.abc import 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(
@@ -37,67 +33,30 @@ 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.
-
-    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.
-    """
+    """Create a checkpoint for the given channels."""
     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 channel_versions:
+            if k not in checkpoint["channel_versions"]:
                 continue
-            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
+            v = channels[k].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=channel_versions,
+        channel_versions=checkpoint["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,
@@ -108,10 +67,14 @@ def channels_from_checkpoint(
     """Hydrate channels from a checkpoint.
 
     For most channels, `spec.from_checkpoint(checkpoint["channel_values"][k])`
-    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`.
+    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.
     """
     channel_specs: dict[str, BaseChannel] = {}
     managed_specs: dict[str, ManagedValueSpec] = {}
@@ -125,12 +88,22 @@ def channels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         stored = checkpoint["channel_values"].get(k, MISSING)
-        if _needs_replay(spec, stored) and saver is not None and config is not None:
-            assert isinstance(spec, DeltaChannel)
+        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.
             history = saver._get_channel_writes_history(config, k)
-            replay_ch = spec.from_checkpoint(history.seed)
-            replay_ch.replay_writes(history.writes)
-            ch = replay_ch
+            delta_ch = spec.from_checkpoint(history.seed)
+            delta_ch.replay_writes(history.writes)
+            ch = delta_ch
         else:
             ch = spec.from_checkpoint(stored)
         channels[k] = ch
@@ -157,12 +130,16 @@ async def achannels_from_checkpoint(
     for k, spec in channel_specs.items():
         ch: BaseChannel
         stored = checkpoint["channel_values"].get(k, MISSING)
-        if _needs_replay(spec, stored) and saver is not None and config is not None:
-            assert isinstance(spec, DeltaChannel)
+        if (
+            isinstance(spec, DeltaChannel)
+            and saver is not None
+            and config is not None
+            and (stored is MISSING or stored is DELTA_SENTINEL)
+        ):
             history = await saver._aget_channel_writes_history(config, k)
-            replay_ch = spec.from_checkpoint(history.seed)
-            replay_ch.replay_writes(history.writes)
-            ch = replay_ch
+            delta_ch = spec.from_checkpoint(history.seed)
+            delta_ch.replay_writes(history.writes)
+            ch = delta_ch
         else:
             ch = spec.from_checkpoint(stored)
         channels[k] = ch

libs/langgraph/langgraph/pregel/_loop.py

@@ -25,7 +25,6 @@ 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,
@@ -199,7 +198,6 @@ 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",
@@ -409,7 +407,7 @@ class PregelLoop:
                     task = self.tasks.get(task_id)
                 else:
                     task = None
-                fut = self.submit(
+                self.submit(
                     self.checkpointer_put_writes,
                     config,
                     writes_to_save,
@@ -417,13 +415,12 @@ class PregelLoop:
                     task_path_str(task.path) if task else "",
                 )
             else:
-                fut = self.submit(
+                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)
@@ -894,9 +891,6 @@ 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(
@@ -934,17 +928,6 @@ 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
@@ -1289,7 +1272,6 @@ 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,
@@ -1495,7 +1477,6 @@ 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

@@ -0,0 +1,290 @@
+"""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,6 +127,7 @@ 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)
@@ -151,6 +152,7 @@ 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)
@@ -172,6 +174,7 @@ 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
@@ -185,6 +188,7 @@ 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
 
@@ -203,6 +207,7 @@ 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)
@@ -236,6 +241,7 @@ 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)
@@ -264,6 +270,7 @@ 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)
@@ -275,124 +282,6 @@ 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
@@ -401,6 +290,7 @@ 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
 
@@ -435,16 +325,11 @@ 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)])
@@ -457,12 +342,14 @@ 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}
@@ -499,7 +386,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."""
+    """Dict reducer that treats None values as deletions works end-to-end (deepagents pattern)."""
 
     def merge_files(left: dict | None, right: dict) -> dict:
         if left is None:
@@ -513,10 +400,15 @@ 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(
@@ -538,6 +430,7 @@ 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}
 
 
@@ -561,12 +454,17 @@ 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]`."""
+    """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.
+    """
     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:
@@ -574,7 +472,10 @@ 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}])
@@ -583,13 +484,17 @@ 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."""
+    """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.
+    """
     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:
@@ -623,9 +528,12 @@ 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
-    assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
+    cv = saved.checkpoint["channel_values"]["files"]
+    assert cv is DELTA_SENTINEL
+
     state = graph.get_state(config)
     assert state.values["files"] == {
         "/doc_1.txt": "content for turn 1",
@@ -633,6 +541,7 @@ 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 snapshot_frequency — storage vs. read-depth tradeoff.
+"""Benchmark: DeltaChannel vs BinaryOperatorAggregate storage and time.
 
 Run directly:  python tests/test_delta_channel_benchmark.py
 Run via pytest: pytest tests/test_delta_channel_benchmark.py -s
 
-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.
+Simulates realistic multi-turn conversations with paragraph-length messages
+(~100 tokens each) scaling up to 1M-token-equivalent histories.
 
-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)
+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.
 """
 
 from __future__ import annotations
 
-import contextlib
-import math
 import sys
 import time
 from typing import Annotated, Any
@@ -26,15 +26,24 @@ 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://sydney_runkle@localhost:5441/postgres?sslmode=disable"
+    _POSTGRES_URI = (
+        "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
+    )
 except ImportError:
     _POSTGRES_AVAILABLE = False
 
@@ -117,18 +126,6 @@ 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
 # ---------------------------------------------------------------------------
@@ -172,8 +169,9 @@ def _run_turns(
     """Run n_turns conversation turns.
 
     Returns (write_elapsed_s, read_elapsed_s, total_blob_bytes).
-    Read latency is the average of 5 get_state calls after the full history
-    is built — forces state rehydration including ancestor replay if needed.
+    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.
     """
     graph = _make_graph(state_cls, checkpointer)
     config = {"configurable": {"thread_id": "bench"}}
@@ -186,16 +184,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
 
-    blob_bytes = (
-        _total_blob_bytes(graph.checkpointer)
-        if isinstance(graph.checkpointer, MemorySaver)
-        else -1
-    )
+    if isinstance(graph.checkpointer, MemorySaver):
+        blob_bytes = _total_blob_bytes(graph.checkpointer)
+    else:
+        blob_bytes = -1
     return write_elapsed, read_elapsed, blob_bytes
 
 
@@ -208,6 +206,7 @@ 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"
@@ -217,56 +216,74 @@ def _approx_tokens(n_turns: int) -> str:
 
 
 # ---------------------------------------------------------------------------
-# Checkpointer factories
+# Benchmark matrix
 # ---------------------------------------------------------------------------
 
+# 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]
 
-@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,))
+# Deep-thread counts where add_messages blob storage would exceed 1 GB;
+# only DeltaChannel runs here.
+DELTA_ONLY_TURN_COUNTS = [1000]
 
 
-def _checkpointers() -> list[tuple[str, Any]]:
-    """Return (label, saver_or_None) pairs for available checkpointers."""
-    result: list[tuple[str, Any]] = [("InMemory", None)]
+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)]
     if _POSTGRES_AVAILABLE:
         try:
             import psycopg
 
             psycopg.connect(_POSTGRES_URI).close()
-            result.append(("Postgres", "postgres"))
+            checkpointers.append(("Postgres (plain SELECT)", "postgres"))
         except Exception:
             pass
-    return result
+
+    for cp_label, cp_hint in checkpointers:
+        print(f"--- Checkpointer: {cp_label} ---")
+        _run_benchmark_for_checkpointer(cp_hint)
 
 
-# ---------------------------------------------------------------------------
-# 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
+def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
+    import contextlib
+    import tempfile
 
+    @contextlib.contextmanager
     def _make_saver():
         if cp_hint is None:
-            return contextlib.nullcontext(None)
-        return _pg_saver()
+            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
 
     rows: list[tuple[int, Any, Any, Any, Any, Any, Any]] = []
-    for turns in BASELINE_TURN_COUNTS:
+    for turns in TURN_COUNTS:
         with _make_saver() as saver:
             b_wt, b_rt, b_bytes = _run_turns(turns, BinaryState, saver)
         with _make_saver() as saver:
@@ -277,19 +294,26 @@ def _run_baseline_for_checkpointer(cp_label: str, 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 or v < 0:
+        if v is None:
+            return "n/a"
+        if 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"
@@ -297,132 +321,65 @@ def _run_baseline_for_checkpointer(cp_label: str, 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}  {ratio_str:>8}"
+            f"{turns:>6}  {_approx_tokens(turns):>10}  "
+            f"{_bytes_or_na(b_bytes):>12}  {_bytes_or_na(d_bytes):>12}  "
+            f"{ratio_str:>8}"
         )
+    print("=" * W)
+    print()
 
-    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))
+    # ── 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)
     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}"
         )
-
-
-def run_baseline_benchmark() -> None:
-    print()
-    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("=" * W)
     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:
 
-# ---------------------------------------------------------------------------
-# Part 2: snapshot_frequency sweep
-# ---------------------------------------------------------------------------
+        def _per(wt: Any) -> str:
+            if wt is None:
+                return "n/a"
+            return f"{(wt / turns) * 1000:.1f}ms"
 
-# 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(
+            f"{turns:>6}  {_approx_tokens(turns):>10}  "
+            f"{_per(b_wt):>12}  {_per(d_wt):>12}"
+        )
+    print("=" * W)
     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("  add_msgs  = Annotated[list, add_messages]  — O(N²) storage")
+    print("  delta     = Annotated[list, DeltaChannel(add_messages)]  — O(N) storage")
     print()
 
 
 # ---------------------------------------------------------------------------
-# Pytest entry points
+# Pytest entry point
 # ---------------------------------------------------------------------------
 
 
 @pytest.mark.skip(
     reason="slow benchmark — run manually with: python tests/test_delta_channel_benchmark.py"
 )
-def test_delta_channel_baseline_benchmark(capsys: Any) -> None:
-    """DeltaChannel(inf) uses less storage than add_messages at scale."""
+def test_delta_channel_benchmark(capsys: Any) -> None:
+    """Storage grows O(N²) for add_messages, O(N) for DeltaChannel."""
     with capsys.disabled():
-        run_baseline_benchmark()
+        run_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)
@@ -432,41 +389,10 @@ def test_delta_channel_baseline_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_baseline_benchmark()
-    run_snapshot_freq_benchmark()
+    run_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,6 +5,7 @@ import langchain_core
 import pytest
 from langchain_core.messages import (
     AIMessage,
+    AIMessageChunk,
     AnyMessage,
     HumanMessage,
     RemoveMessage,
@@ -338,6 +339,123 @@ 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,7 +41,6 @@ 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
@@ -9408,6 +9407,7 @@ 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,6 +9449,7 @@ 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
 
@@ -9506,6 +9507,7 @@ 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
 
@@ -9552,6 +9554,7 @@ 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
 
@@ -9584,90 +9587,3 @@ 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,6 +82,7 @@ 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
@@ -800,7 +801,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)
+            state = self._extract_state(input, cfg)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -835,7 +836,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)
+            state = self._extract_state(input, cfg)
             tool_runtime = ToolRuntime(
                 state=state,
                 tool_call_id=call["id"],
@@ -1273,18 +1274,37 @@ class ToolNode(RunnableCallable):
         return None
 
     def _extract_state(
-        self, input: list[AnyMessage] | dict[str, Any] | BaseModel
+        self,
+        input: list[AnyMessage] | dict[str, Any] | BaseModel,
+        config: RunnableConfig,
     ) -> list[AnyMessage] | dict[str, Any] | BaseModel:
-        """Extract state from input, handling ToolCallWithContext if present.
+        """Extract state from input.
 
-        Args:
-            input: The input which may be raw state or ToolCallWithContext.
+        Three input shapes:
 
-        Returns:
-            The actual state to pass to wrap_tool_call wrappers.
+        - `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.
         """
         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,6 +1320,98 @@ 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"}