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/__init__.py

@@ -4,7 +4,7 @@ import threading
 from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
-from typing import Any, cast
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -26,9 +26,10 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_BLOBS_SQL,
+    SELECT_DELTA_PARENTS_SQL,
+    SELECT_DELTA_WRITES_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -441,9 +442,10 @@ class PostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
 
-        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
-        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
-        single roundtrip; the ancestor walk runs in Python.
+        Three indexed roundtrips (`checkpoints`, `checkpoint_writes`,
+        `checkpoint_blobs`) each filtered by `(thread_id, checkpoint_ns)` and
+        the per-table key. Plain SELECTs let the planner pick straight index
+        scans; rationale + benchmark in `notes/delta_channel_query_bench.md`.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
@@ -457,25 +459,18 @@ class PostgresSaver(BasePostgresSaver):
                 return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
         with self._cursor() as cur:
-            cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
-                (
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
-                ),
-            )
-            rows = cur.fetchall()
+            cur.execute(SELECT_DELTA_PARENTS_SQL, (channel, thread_id, checkpoint_ns))
+            parents_rows = cur.fetchall()
+            cur.execute(SELECT_DELTA_WRITES_SQL, (thread_id, checkpoint_ns, channel))
+            writes_rows = cur.fetchall()
+            cur.execute(SELECT_DELTA_BLOBS_SQL, (thread_id, checkpoint_ns, channel))
+            blobs_rows = cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
             target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            parents_rows=parents_rows,
+            writes_rows=writes_rows,
+            blobs_rows=blobs_rows,
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
@@ -490,6 +485,7 @@ class PostgresSaver(BasePostgresSaver):
             including its configuration, metadata, parent checkpoint (if any),
             and pending writes.
         """
+        channel_values = self._load_blobs(value["channel_values"])
         return CheckpointTuple(
             {
                 "configurable": {
@@ -502,7 +498,7 @@ class PostgresSaver(BasePostgresSaver):
                 **value["checkpoint"],
                 "channel_values": {
                     **(value["checkpoint"].get("channel_values") or {}),
-                    **self._load_blobs(value["channel_values"]),
+                    **channel_values,
                 },
             },
             value["metadata"],

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

@@ -4,7 +4,7 @@ import asyncio
 from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
-from typing import Any, cast
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -26,9 +26,10 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_BLOBS_SQL,
+    SELECT_DELTA_PARENTS_SQL,
+    SELECT_DELTA_WRITES_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -402,10 +403,10 @@ class AsyncPostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
 
-        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
-        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
-        single roundtrip; rows are assembled by the shared pure helper on
-        `BasePostgresSaver`.
+        Three indexed roundtrips (`checkpoints`, `checkpoint_writes`,
+        `checkpoint_blobs`); rows assembled by the shared pure helper on
+        `BasePostgresSaver`. Rationale + benchmark in
+        `notes/delta_channel_query_bench.md`.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
@@ -417,24 +418,23 @@ class AsyncPostgresSaver(BasePostgresSaver):
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
         async with self._cursor() as cur:
             await cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
-                (
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
-                ),
+                SELECT_DELTA_PARENTS_SQL, (channel, thread_id, checkpoint_ns)
             )
-            rows = await cur.fetchall()
+            parents_rows = await cur.fetchall()
+            await cur.execute(
+                SELECT_DELTA_WRITES_SQL, (thread_id, checkpoint_ns, channel)
+            )
+            writes_rows = await cur.fetchall()
+            await cur.execute(
+                SELECT_DELTA_BLOBS_SQL, (thread_id, checkpoint_ns, channel)
+            )
+            blobs_rows = await cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
             target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            parents_rows=parents_rows,
+            writes_rows=writes_rows,
+            blobs_rows=blobs_rows,
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
@@ -449,11 +449,18 @@ class AsyncPostgresSaver(BasePostgresSaver):
             including its configuration, metadata, parent checkpoint (if any),
             and pending writes.
         """
+        thread_id = value["thread_id"]
+        checkpoint_ns = value["checkpoint_ns"]
+        blob_values = value["channel_values"]
+        channel_values: dict[str, Any] = {}
+        if blob_values:
+            channel_values = self._load_blobs(blob_values)
+
         return CheckpointTuple(
             {
                 "configurable": {
-                    "thread_id": value["thread_id"],
-                    "checkpoint_ns": value["checkpoint_ns"],
+                    "thread_id": thread_id,
+                    "checkpoint_ns": checkpoint_ns,
                     "checkpoint_id": value["checkpoint_id"],
                 }
             },
@@ -461,15 +468,15 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 **value["checkpoint"],
                 "channel_values": {
                     **(value["checkpoint"].get("channel_values") or {}),
-                    **self._load_blobs(value["channel_values"]),
+                    **channel_values,
                 },
             },
             value["metadata"],
             (
                 {
                     "configurable": {
-                        "thread_id": value["thread_id"],
-                        "checkpoint_ns": value["checkpoint_ns"],
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
                         "checkpoint_id": value["parent_checkpoint_id"],
                     }
                 }

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

@@ -4,7 +4,7 @@ import random
 import warnings
 from collections.abc import Sequence
 from importlib.metadata import version as get_version
-from typing import Any, TypedDict, cast
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -155,58 +155,26 @@ INSERT_CHECKPOINT_WRITES_SQL = """
     ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
 """
 
-
-class _DeltaCombinedRow(TypedDict, total=False):
-    """One row from `SELECT_DELTA_COMBINED_SQL` (a UNION ALL of three tables).
-
-    Every row carries `_kind` ("p" / "w" / "b") plus whichever columns are
-    relevant for that kind; irrelevant columns are NULL and typed as `None`.
-    """
-
-    _kind: str  # always present: "p", "w", or "b"
-    # checkpoint row ("p")
-    checkpoint_id: str | None
-    parent_checkpoint_id: str | None
-    ver: str | None
-    # write / blob rows ("w", "b")
-    type: str | None
-    blob: bytes | None
-    # write row only ("w")
-    task_id: str | None
-    idx: int | None
-    # blob row only ("b")
-    version: str | None
-
-
-# DeltaChannel reconstruction: one UNION ALL query fetches checkpoints,
-# writes, and blobs for `channel` in one roundtrip; the ancestor walk runs
-# in Python in `_build_delta_channel_writes_history`.
-#
-# Parameter order: (channel, thread_id, checkpoint_ns,
-#                   thread_id, checkpoint_ns, channel,
-#                   thread_id, checkpoint_ns, channel)
-SELECT_DELTA_COMBINED_SQL = """
-    SELECT 'p'::text       AS _kind,
-           checkpoint_id,
+# DeltaChannel reconstruction: three plain indexed SELECTs per channel.
+# Bench (notes/delta_channel_query_bench.md) showed the prior recursive CTE
+# carried a hidden O(ancestors x blobs_in_thread) join; plain SELECTs are
+# 3x-100x faster in the realistic depth range and the Python walk is O(n).
+SELECT_DELTA_PARENTS_SQL = """
+    SELECT checkpoint_id,
            parent_checkpoint_id,
-           checkpoint -> 'channel_versions' ->> %s AS ver,
-           NULL::text       AS type,
-           NULL::bytea      AS blob,
-           NULL::text       AS task_id,
-           NULL::int        AS idx,
-           NULL::text       AS version
+           checkpoint -> 'channel_versions' ->> %s AS ver
     FROM checkpoints
     WHERE thread_id = %s AND checkpoint_ns = %s
-    UNION ALL
-    SELECT 'w',
-           checkpoint_id, NULL, NULL,
-           type, blob, task_id, idx, NULL
+"""
+
+SELECT_DELTA_WRITES_SQL = """
+    SELECT checkpoint_id, type, blob, task_id, idx
     FROM checkpoint_writes
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
-    UNION ALL
-    SELECT 'b',
-           NULL, NULL, NULL,
-           type, blob, NULL, NULL, version
+"""
+
+SELECT_DELTA_BLOBS_SQL = """
+    SELECT version, type, blob
     FROM checkpoint_blobs
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
 """
@@ -244,28 +212,32 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         )
 
     def _load_blobs(
-        self, blob_values: list[tuple[bytes, bytes, bytes]]
+        self,
+        blob_values: Any,
     ) -> dict[str, Any]:
         if not blob_values:
             return {}
-        return {
-            k.decode(): self.serde.loads_typed((t.decode(), v))
-            for k, t, v in blob_values
-            if t.decode() != "empty"
-        }
+        result: dict[str, Any] = {}
+        for k, t, v in blob_values:
+            type_tag = t.decode()
+            if type_tag != "empty":
+                result[k.decode()] = self.serde.loads_typed((type_tag, v))
+        return result
 
     def _build_delta_channel_writes_history(
         self,
         *,
         channel: str,
         target_id: str,
-        rows: Sequence[_DeltaCombinedRow],
+        parents_rows: Sequence[Any],
+        writes_rows: Sequence[Any],
+        blobs_rows: Sequence[Any],
     ) -> _ChannelWritesHistory:
-        """Reconstruct one delta channel's history from the combined UNION ALL rows.
+        """Reconstruct one delta channel's history from rows of the three SELECTs.
 
         Pure data transform shared by sync (`PostgresSaver`) and async
-        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
-        and feed the tagged rows here.
+        (`AsyncPostgresSaver`); both paths run the queries themselves and
+        feed the rows here.
 
         Walk is newest → oldest from the target's parent. A non-sentinel
         blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
@@ -276,49 +248,40 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         """
         parent_of: dict[str, str | None] = {}
         ver_of: dict[str, str | None] = {}
+        for r in parents_rows:
+            cid = r["checkpoint_id"]
+            parent_of[cid] = r["parent_checkpoint_id"]
+            ver_of[cid] = r["ver"]
+
+        ancestors: list[str] = []
+        cid = parent_of.get(target_id)
+        while cid is not None:
+            ancestors.append(cid)
+            cid = parent_of.get(cid)
+        if not ancestors:
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+        ancestor_set = set(ancestors)
+
+        # Group writes by ancestor cid; sort within (task_id DESC, idx DESC)
+        # to match the prior CTE ordering — newest write first per ancestor.
         writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
-        blob_by_ver: dict[str, tuple[str, bytes]] = {}
-
-        for r in rows:
-            kind = r["_kind"]
-            if kind == "p":
-                cid = cast(str, r["checkpoint_id"])
-                parent_of[cid] = r["parent_checkpoint_id"]
-                ver_of[cid] = r["ver"]
-            elif kind == "w":
-                cid = cast(str, r["checkpoint_id"])
-                writes_by_cid.setdefault(cid, []).append(
-                    cast(
-                        "tuple[str, bytes, str, int]",
-                        (r["type"], r["blob"], r["task_id"], r["idx"]),
-                    )
-                )
-            else:  # kind == "b"
-                blob_by_ver[cast(str, r["version"])] = cast(
-                    "tuple[str, bytes]", (r["type"], r["blob"])
-                )
-
-        # newest write first per ancestor (task_id DESC, idx DESC)
+        for r in writes_rows:
+            cid = r["checkpoint_id"]
+            if cid not in ancestor_set:
+                continue
+            writes_by_cid.setdefault(cid, []).append(
+                (r["type"], r["blob"], r["task_id"], r["idx"])
+            )
         for ws in writes_by_cid.values():
             ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
-        ancestors: list[str] = []
-        cur_cid: str | None = parent_of.get(target_id)
-        while cur_cid is not None:
-            ancestors.append(cur_cid)
-            cur_cid = parent_of.get(cur_cid)
-        if not ancestors:
-            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+        blob_by_ver: dict[str, tuple[str, bytes]] = {
+            r["version"]: (r["type"], r["blob"]) for r in blobs_rows
+        }
 
         collected: list[PendingWrite] = []  # newest first; reversed at the end
         for cid in ancestors:
-            # Collect writes first — they encode the transition FROM this
-            # ancestor's state to its child's and must be included even if
-            # this ancestor is also the seed checkpoint.
-            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))
-            # Then check seed terminator.
+            # 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)
@@ -327,6 +290,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     if blob_value is not DELTA_SENTINEL:
                         collected.reverse()
                         return _ChannelWritesHistory(seed=blob_value, writes=collected)
+            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()  # oldest → newest
         return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)

libs/checkpoint-postgres/pyproject.toml

@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-  "langgraph-checkpoint>=4.0.3,<5.0.0",
+  "langgraph-checkpoint>=2.1.2,<5.0.0",
   "orjson>=3.11.5",
   "psycopg>=3.2.0",
   "psycopg-pool>=3.2.0",

libs/checkpoint-postgres/tests/test_async.py

@@ -377,19 +377,19 @@ 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 _messages_delta_reducer
+    from langgraph.graph.message import add_messages
     from typing_extensions import TypedDict
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         n = len(state["messages"])

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextvars
 import copy
 import logging
 from collections.abc import AsyncIterator, Collection, Iterator, Mapping, Sequence
@@ -32,6 +33,14 @@ from langgraph.checkpoint.serde.types import (
 V = TypeVar("V", int, float, str)
 PendingWrite = tuple[str, str, Any]
 
+# Task-local guard: ContextVar is copied per asyncio Task, so concurrent
+# requests on the same event-loop thread do not share this flag. A plain
+# `threading.local()` would leak across tasks and let one in-flight
+# reconstruction silently short-circuit another.
+_DELTA_RECONSTRUCTION: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "_DELTA_RECONSTRUCTION", default=False
+)
+
 
 logger = logging.getLogger(__name__)
 
@@ -486,20 +495,6 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    def _get_tuple_raw(self, config: RunnableConfig) -> CheckpointTuple | None:
-        """Pure storage read used by `_get_channel_writes_history`.
-
-        Must return the same value as `get_tuple` but must NOT trigger channel
-        reconstruction; otherwise the channel-hydration path would re-enter
-        `_get_channel_writes_history`. Override only if `get_tuple` itself
-        performs channel hydration.
-        """
-        return self.get_tuple(config)
-
-    async def _aget_tuple_raw(self, config: RunnableConfig) -> CheckpointTuple | None:
-        """Async version of `_get_tuple_raw`. See docstring there."""
-        return await self.aget_tuple(config)
-
     def _get_channel_writes_history(
         self, config: RunnableConfig, channel: str
     ) -> _ChannelWritesHistory:
@@ -528,61 +523,75 @@ class BaseCheckpointSaver(Generic[V]):
 
         Underscore-prefixed because the method surface is experimental.
         """
-        collected: list[PendingWrite] = []  # newest first; reversed at the end
-        target_tuple = self._get_tuple_raw(config)
-        cursor_config: RunnableConfig | None = (
-            target_tuple.parent_config if target_tuple else None
-        )
-        while cursor_config is not None:
-            tup = self._get_tuple_raw(cursor_config)
-            if tup is None:
-                break
-            # Collect this ancestor's writes FIRST — they encode the
-            # transition from this ancestor's state to its child's, so
-            # they must be included whether or not this ancestor is the
-            # seed terminator.
-            if tup.pending_writes:
-                # Within a superstep, pending_writes are oldest→newest;
-                # reverse to scan newest-first.
-                for write in reversed(tup.pending_writes):
-                    if write[1] != channel:
-                        continue
-                    collected.append(write)
-            # Seed terminator: any non-sentinel blob on an ancestor
-            # establishes the reconstruction base. Stop here.
-            ancestor_value = tup.checkpoint["channel_values"].get(channel)
-            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
-                collected.reverse()
-                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
-            cursor_config = tup.parent_config
-        collected.reverse()
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        # Guard against re-entrant calls: when get_tuple() triggers
+        # reconstruction which calls get_tuple() again, the inner call
+        # short-circuits here.
+        if _DELTA_RECONSTRUCTION.get():
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+
+        token = _DELTA_RECONSTRUCTION.set(True)
+        try:
+            collected: list[PendingWrite] = []  # newest first; reversed at the end
+            target_tuple = self.get_tuple(config)
+            cursor_config: RunnableConfig | None = (
+                target_tuple.parent_config if target_tuple else None
+            )
+            while cursor_config is not None:
+                tup = self.get_tuple(cursor_config)
+                if tup is None:
+                    break
+                # Pre-delta seed terminator: if the ancestor has a stored
+                # (non-sentinel) value for this channel, that snapshot
+                # subsumes any earlier writes on the chain. Stop here.
+                ancestor_value = tup.checkpoint["channel_values"].get(channel)
+                if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                    collected.reverse()
+                    return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
+                if tup.pending_writes:
+                    # Within a superstep, pending_writes are oldest→newest;
+                    # reverse to scan newest-first.
+                    for write in reversed(tup.pending_writes):
+                        if write[1] != channel:
+                            continue
+                        collected.append(write)
+                cursor_config = tup.parent_config
+            collected.reverse()
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        finally:
+            _DELTA_RECONSTRUCTION.reset(token)
 
     async def _aget_channel_writes_history(
         self, config: RunnableConfig, channel: str
     ) -> _ChannelWritesHistory:
         """Async version of `_get_channel_writes_history`. See docstring there."""
-        collected: list[PendingWrite] = []
-        target_tuple = await self._aget_tuple_raw(config)
-        cursor_config: RunnableConfig | None = (
-            target_tuple.parent_config if target_tuple else None
-        )
-        while cursor_config is not None:
-            tup = await self._aget_tuple_raw(cursor_config)
-            if tup is None:
-                break
-            if tup.pending_writes:
-                for write in reversed(tup.pending_writes):
-                    if write[1] != channel:
-                        continue
-                    collected.append(write)
-            ancestor_value = tup.checkpoint["channel_values"].get(channel)
-            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
-                collected.reverse()
-                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
-            cursor_config = tup.parent_config
-        collected.reverse()
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        if _DELTA_RECONSTRUCTION.get():
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+
+        token = _DELTA_RECONSTRUCTION.set(True)
+        try:
+            collected: list[PendingWrite] = []
+            target_tuple = await self.aget_tuple(config)
+            cursor_config: RunnableConfig | None = (
+                target_tuple.parent_config if target_tuple else None
+            )
+            while cursor_config is not None:
+                tup = await self.aget_tuple(cursor_config)
+                if tup is None:
+                    break
+                ancestor_value = tup.checkpoint["channel_values"].get(channel)
+                if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                    collected.reverse()
+                    return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
+                if tup.pending_writes:
+                    for write in reversed(tup.pending_writes):
+                        if write[1] != channel:
+                            continue
+                        collected.append(write)
+                cursor_config = tup.parent_config
+            collected.reverse()
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        finally:
+            _DELTA_RECONSTRUCTION.reset(token)
 
     def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.

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

@@ -9,7 +9,7 @@ from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from types import TracebackType
-from typing import Any
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 
@@ -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
@@ -255,13 +231,16 @@ class InMemorySaver(
                 checkpoint, metadata, parent_checkpoint_id = saved
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                 checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
+                channel_values = self._load_blobs(
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint_["channel_versions"],
+                )
                 return CheckpointTuple(
                     config=config,
                     checkpoint={
                         **checkpoint_,
-                        "channel_values": self._load_blobs(
-                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
-                        ),
+                        "channel_values": channel_values,
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -285,19 +264,26 @@ class InMemorySaver(
                 checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                 checkpoint_ = self.serde.loads_typed(checkpoint)
-                return CheckpointTuple(
-                    config={
+                resolved_config = cast(
+                    RunnableConfig,
+                    {
                         "configurable": {
                             "thread_id": thread_id,
                             "checkpoint_ns": checkpoint_ns,
                             "checkpoint_id": checkpoint_id,
                         }
                     },
+                )
+                channel_values = self._load_blobs(
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint_["channel_versions"],
+                )
+                return CheckpointTuple(
+                    config=resolved_config,
                     checkpoint={
                         **checkpoint_,
-                        "channel_values": self._load_blobs(
-                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
-                        ),
+                        "channel_values": channel_values,
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -392,21 +378,27 @@ class InMemorySaver(
 
                     checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
 
-                    yield CheckpointTuple(
-                        config={
+                    list_config = cast(
+                        RunnableConfig,
+                        {
                             "configurable": {
                                 "thread_id": thread_id,
                                 "checkpoint_ns": checkpoint_ns,
                                 "checkpoint_id": checkpoint_id,
                             }
                         },
+                    )
+                    channel_values = self._load_blobs(
+                        thread_id,
+                        checkpoint_ns,
+                        checkpoint_["channel_versions"],
+                    )
+
+                    yield CheckpointTuple(
+                        config=list_config,
                         checkpoint={
                             **checkpoint_,
-                            "channel_values": self._load_blobs(
-                                thread_id,
-                                checkpoint_ns,
-                                checkpoint_["channel_versions"],
-                            ),
+                            "channel_values": channel_values,
                         },
                         metadata=metadata,
                         parent_config=(

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, _DeltaSnapshot):
-        return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
-    elif isinstance(obj, _DeltaSentinel):
-        return ormsgpack.Ext(EXT_DELTA_SENTINEL, b"")
-    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/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint"
-version = "4.1.0a2"
+version = "4.0.2"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.10"

libs/checkpoint/tests/test_jsonplus.py

@@ -1005,7 +1005,8 @@ def test_delta_sentinel_serde_round_trip() -> None:
 
     serde = JsonPlusSerializer()
     type_tag, blob = serde.dumps_typed(DELTA_SENTINEL)
-    assert type_tag == "msgpack"
-    assert blob  # non-empty ext envelope
+    # Zero-byte "delta" tag — no allowlist change needed.
+    assert type_tag == "delta"
+    assert blob == b""
     loaded = serde.loads_typed((type_tag, blob))
     assert loaded is DELTA_SENTINEL

libs/checkpoint/uv.lock

@@ -7,9 +7,6 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 
-[options]
-prerelease-mode = "allow"
-
 [[package]]
 name = "annotated-types"
 version = "0.7.0"
@@ -289,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a2"
+version = "4.0.2"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

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

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import copy as _copy
+from collections.abc import Callable, Sequence
+from typing import Any, Generic
+
+from langgraph.checkpoint.base import DELTA_SENTINEL, PendingWrite
+from typing_extensions import Self
+
+from langgraph._internal._typing import MISSING
+from langgraph.channels.base import BaseChannel, Value
+from langgraph.channels.binop import _get_overwrite
+from langgraph.errors import EmptyChannelError
+
+__all__ = ("DeltaChannel",)
+
+
+def _empty(typ: Any) -> Any:
+    try:
+        return typ()
+    except Exception:
+        return []
+
+
+class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
+    """Experimental — private API, subject to change or removal without notice.
+
+    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.
+
+    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.
+
+    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",
+    )
+
+    def __init__(
+        self,
+        operator: Callable[[Any, Any], Any],
+    ) -> None:
+        super().__init__(list)
+        self.operator = operator
+        self.value: Any = []
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, DeltaChannel):
+            return False
+        if (
+            self.operator.__name__ != "<lambda>"
+            and other.operator.__name__ != "<lambda>"
+        ):
+            return self.operator is other.operator
+        return True
+
+    @property
+    def ValueType(self) -> Any:
+        return self.typ
+
+    @property
+    def UpdateType(self) -> Any:
+        return self.typ
+
+    def copy(self) -> Self:
+        new: DeltaChannel[Value] = DeltaChannel(self.operator)
+        new.typ = self.typ
+        new.key = self.key
+        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 (
+                _copy.copy(overwrite_value)
+                if overwrite_value is not None
+                else _empty(self.typ)
+            )
+        base = _empty(self.typ) if value is MISSING else value
+        return self.operator(base, write)
+
+    def from_checkpoint(self, checkpoint: Any) -> Self:
+        """Initialize from a seed value.
+
+        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: 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)
+        else:
+            new.value = checkpoint
+        return new
+
+    def replay_writes(self, writes: Sequence[PendingWrite]) -> None:
+        """Fold a sequence of `PendingWrite` tuples into the current value.
+
+        Called after `from_checkpoint` during pregel hydration to replay
+        per-step deltas from on-path ancestors through the reducer. Writes
+        are oldest→newest. `Overwrite` values inside the stream reset the
+        reducer state at that point, same as during a live super-step.
+        The `task_id` and `channel` fields of each `PendingWrite` are
+        ignored — `_get_channel_writes_history` has already filtered to
+        this channel.
+        """
+        for _, _, value in writes:
+            self.value = self._apply_write(self.value, value)
+
+    def update(self, values: Sequence[Any]) -> bool:
+        if not values:
+            return False
+        seen_overwrite = False
+        for value in values:
+            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,
+                    )
+                    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
+
+    def get(self) -> Any:
+        if self.value is MISSING:
+            raise EmptyChannelError()
+        return self.value
+
+    def is_available(self) -> bool:
+        return self.value is not MISSING
+
+    def checkpoint(self) -> Any:
+        return DELTA_SENTINEL

libs/langgraph/langgraph/channels/binop.py

@@ -22,9 +22,10 @@ __all__ = ("BinaryOperatorAggregate",)
 def _strip_extras(t):  # type: ignore[no-untyped-def]
     """Strips Annotated, Required and NotRequired from a given type."""
     if hasattr(t, "__origin__"):
-        if t.__origin__ in (Required, NotRequired):
-            return _strip_extras(t.__args__[0])
         return _strip_extras(t.__origin__)
+    if hasattr(t, "__origin__") and t.__origin__ in (Required, NotRequired):
+        return _strip_extras(t.__args__[0])
+
     return t
 
 
@@ -32,22 +33,11 @@ def _get_overwrite(value: Any) -> tuple[bool, Any]:
     """Inspects the given value and returns (is_overwrite, overwrite_value)."""
     if isinstance(value, Overwrite):
         return True, value.value
-    if isinstance(value, dict) and len(value) == 1 and OVERWRITE in value:
+    if isinstance(value, dict) and set(value.keys()) == {OVERWRITE}:
         return True, value[OVERWRITE]
     return False, None
 
 
-def _operators_equal(a: Callable, b: Callable) -> bool:
-    """Return True if two reducer operators should be considered equal.
-
-    Lambdas all share the name '<lambda>' so identity comparison is
-    unreliable; treat any pairing that includes a lambda as equal.
-    """
-    if a.__name__ == "<lambda>" or b.__name__ == "<lambda>":
-        return True
-    return a is b
-
-
 class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
     """Stores the result of applying a binary operator to the current value and each new value.
 
@@ -78,8 +68,11 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
             self.value = MISSING
 
     def __eq__(self, value: object) -> bool:
-        return isinstance(value, BinaryOperatorAggregate) and _operators_equal(
-            self.operator, value.operator
+        return isinstance(value, BinaryOperatorAggregate) and (
+            value.operator is self.operator
+            if value.operator.__name__ != "<lambda>"
+            and self.operator.__name__ != "<lambda>"
+            else True
         )
 
     @property

libs/langgraph/langgraph/channels/delta.py

@@ -1,197 +0,0 @@
-from __future__ import annotations
-
-import collections.abc
-import copy as _copy
-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._typing import MISSING
-from langgraph.channels.base import BaseChannel, Value
-from langgraph.channels.binop import _get_overwrite, _operators_equal, _strip_extras
-from langgraph.errors import (
-    EmptyChannelError,
-    ErrorCode,
-    InvalidUpdateError,
-    create_error_message,
-)
-
-__all__ = ("DeltaChannel",)
-
-
-class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
-    """Reducer channel that stores only a sentinel in checkpoint blobs and
-    reconstructs state by replaying ancestor writes through the reducer.
-
-    The reducer receives the current accumulated value and a batch of writes
-    in one call: `reducer(state, [write1, write2, ...]) -> new_state`.
-
-    Reducers must be deterministic and batching-invariant (associative across
-    folds): applying two consecutive write batches separately must produce the
-    same state as applying their concatenation once:
-
-        reducer(reducer(state, xs), ys) == reducer(state, xs + ys)
-
-    This lets LangGraph replay checkpointed writes in larger batches than they
-    were originally produced without changing reconstructed state.
-
-    `snapshot_frequency=None` (default): pure delta; stores only
-    `DELTA_SENTINEL` in checkpoint blobs; reads replay all ancestor writes.
-
-    `snapshot_frequency=N`: `create_checkpoint` writes a full `_DeltaSnapshot`
-    blob every N steps, bounding replay depth to N.
-
-    Parameters:
-        reducer: `(state, list[writes]) -> new_state`. Must be deterministic
-            and batching-invariant as described above.
-        typ: The value type (e.g. `list`, `dict`). Inferred automatically
-            from the outer type when used inside `Annotated[T, DeltaChannel(...)]`.
-        snapshot_frequency: Every Nth pregel step writes a snapshot blob.
-            `None` (default) = pure delta, never snapshot.
-    """
-
-    __slots__ = ("value", "reducer", "snapshot_frequency")
-    value: Value | Any
-
-    def __init__(
-        self,
-        reducer: Callable[[Any, Sequence[Any]], Any],
-        typ: type[Value] | None = None,
-        *,
-        snapshot_frequency: int | None = None,
-    ) -> None:
-        if typ is None:
-            typ = list  # type: ignore[assignment]  # placeholder; overridden by _is_field_channel
-        super().__init__(typ)
-        self.reducer = reducer
-        self.snapshot_frequency = snapshot_frequency
-        typ = _strip_extras(typ)
-        if typ in (collections.abc.Sequence, collections.abc.MutableSequence):
-            typ = list
-        if typ in (collections.abc.Set, collections.abc.MutableSet):
-            typ = set
-        if typ in (collections.abc.Mapping, collections.abc.MutableMapping):
-            typ = dict
-        self.typ = typ
-        self.value: Any = MISSING
-
-    def __eq__(self, other: object) -> bool:
-        if not isinstance(other, DeltaChannel):
-            return False
-        if self.snapshot_frequency != other.snapshot_frequency:
-            return False
-        return _operators_equal(self.reducer, other.reducer)
-
-    @property
-    def ValueType(self) -> Any:
-        return self.typ
-
-    @property
-    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 > 0
-            and step % self.snapshot_frequency == 0
-        )
-
-    def copy(self) -> Self:
-        new = self.__class__(
-            self.reducer, self.typ, snapshot_frequency=self.snapshot_frequency
-        )
-        new.key = self.key
-        new.value = self.value if self.value is MISSING else _copy.copy(self.value)
-        return new
-
-    def from_checkpoint(self, checkpoint: Any) -> Self:
-        """Initialize from a stored blob or sentinel.
-
-        Blob types (dispatched via serde ext code, not dict key inspection):
-          * `DELTA_SENTINEL` / `MISSING`: start empty; caller replays writes.
-          * `_DeltaSnapshot(value)`: restore value directly from snapshot.
-          * plain value (migration from old BinOp blobs): use directly.
-        """
-        new = self.__class__(
-            self.reducer, self.typ, snapshot_frequency=self.snapshot_frequency
-        )
-        new.key = self.key
-        if checkpoint is MISSING or checkpoint is DELTA_SENTINEL:
-            new.value = self.typ()
-        elif isinstance(checkpoint, _DeltaSnapshot):
-            new.value = checkpoint.value
-        else:
-            new.value = checkpoint
-        return new
-
-    def replay_writes(self, writes: Sequence[PendingWrite]) -> None:
-        """Apply ancestor writes oldest-to-newest via a single reducer call.
-
-        If any write is an Overwrite, the last one in the sequence acts as
-        the reset point: its value becomes the new base and only writes
-        after it are passed to the reducer.
-        """
-        values = [v for _, _, v in writes]
-        if not values:
-            return
-        base = self.value
-        start = 0
-        for i, v in enumerate(values):
-            is_ow, ow_value = _get_overwrite(v)
-            if is_ow:
-                base = _copy.copy(ow_value) if ow_value is not None else self.typ()
-                start = i + 1
-        remaining = values[start:]
-        self.value = self.reducer(base, remaining) if remaining else base
-
-    def update(self, values: Sequence[Any]) -> bool:
-        if not values:
-            return False
-        overwrite_idx: int | None = None
-        for i, v in enumerate(values):
-            is_ow, _ = _get_overwrite(v)
-            if is_ow:
-                if overwrite_idx is not None:
-                    msg = create_error_message(
-                        message="Can receive only one Overwrite value per super-step.",
-                        error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
-                    )
-                    raise InvalidUpdateError(msg)
-                overwrite_idx = i
-        if overwrite_idx is not None:
-            _, overwrite_value = _get_overwrite(values[overwrite_idx])
-            base = (
-                _copy.copy(overwrite_value)
-                if overwrite_value is not None
-                else self.typ()
-            )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
-            self.value = self.reducer(base, remaining) if remaining else base
-            return True
-        base = self.typ() if self.value is MISSING else self.value
-        self.value = self.reducer(base, list(values))
-        return True
-
-    def get(self) -> Any:
-        if self.value is MISSING:
-            raise EmptyChannelError()
-        return self.value
-
-    def is_available(self) -> bool:
-        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,112 +184,100 @@ 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
 
 
-def _messages_delta_reducer(
-    state: list[AnyMessage], writes: list[list[AnyMessage]]
-) -> list[AnyMessage]:
-    """**Experimental.** Batch reducer for use with `DeltaChannel`.
-
-    Processes all writes in one pass — dedup by ID, `RemoveMessage`
-    tombstoning — without calling `add_messages`. Assumes writes contain
-    already-typed `BaseMessage` objects (no raw-dict coercion).
-
-    This reducer is batching-invariant, as required by `DeltaChannel`:
-    `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.
-
-    Use `add_messages` as the reducer for `BinaryOperatorAggregate` or
-    anywhere raw message dicts / strings need to be coerced first.
-
-    Example::
-
-        from typing import Annotated
-        from langgraph.channels.delta import DeltaChannel
-        from langgraph.graph.message import _messages_delta_reducer
-
-        class State(TypedDict):
-            messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-    """
-    from itertools import chain
-
-    index: dict[str, int] = {m.id: i for i, m in enumerate(state) if m.id is not None}
-    result: list[AnyMessage | None] = list(state)
-    for msg in chain.from_iterable(
-        [w] if isinstance(w, BaseMessage) else w for w in writes
-    ):
-        mid = msg.id
-        if mid is None:
-            result.append(msg)
-        elif isinstance(msg, RemoveMessage):
-            if mid in index:
-                result[index[mid]] = None
-                del index[mid]
-        elif mid in index:
-            result[index[mid]] = msg
-        else:
-            index[mid] = len(result)
-            result.append(msg)
-    return [m for m in result if m is not None]
-
-
 @deprecated(
     "MessageGraph is deprecated in langgraph 1.0.0, to be removed in 2.0.0. Please use StateGraph with a `messages` key instead.",
     category=None,

libs/langgraph/langgraph/graph/state.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import collections.abc
 import inspect
 import logging
 import typing
@@ -47,8 +48,8 @@ from langgraph._internal._pydantic import create_model
 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
-from langgraph.channels.delta import DeltaChannel
+from langgraph.channels.binop import BinaryOperatorAggregate, _strip_extras
+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 (
@@ -1678,11 +1679,27 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None:
                         NotRequired,
                     ):
                         origin = origin.__args__[0]
-                    item = item.__class__(
-                        item.reducer,
-                        origin,
-                        snapshot_frequency=item.snapshot_frequency,
-                    )
+                    outer = _strip_extras(origin)
+                    if outer in (
+                        collections.abc.Sequence,
+                        collections.abc.MutableSequence,
+                    ):
+                        outer = list
+                    elif outer in (
+                        collections.abc.Mapping,
+                        collections.abc.MutableMapping,
+                    ):
+                        outer = dict
+                    elif outer in (
+                        collections.abc.Set,
+                        collections.abc.MutableSet,
+                    ):
+                        outer = set
+                    item.typ = outer
+                    try:
+                        item.value = outer()
+                    except Exception:
+                        item.value = []
                 return item
             elif isclass(item) and issubclass(item, BaseChannel):
                 # ex, Annotated[int, EphemeralValue, SomeOtherAnnotation]

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, cast
 
 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,72 +33,30 @@ def create_checkpoint(
     *,
     id: str | None = None,
     updated_channels: set[str] | None = None,
-    get_next_version: GetNextVersion | None = None,
-    force_delta_snapshot: bool = False,
 ) -> 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.
-
-    `force_delta_snapshot` writes available `DeltaChannel` values as snapshots
-    regardless of `snapshot_frequency`. This is used by `durability="exit"`,
-    where intermediate writes are not stored as ancestor `checkpoint_writes`.
-    """
+    """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 (force_delta_snapshot or 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,
@@ -113,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] = {}
@@ -130,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:
-            delta_spec = cast(DeltaChannel, spec)
+        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 = delta_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
@@ -162,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:
-            delta_spec = cast(DeltaChannel, spec)
+        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 = delta_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

@@ -68,7 +68,6 @@ from langgraph.callbacks import (
     GraphResumeEvent,
 )
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import (
@@ -190,8 +189,6 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
-    # Only set on AsyncPregelLoop; sync loops keep this as None.
-    _delta_write_futs: list[Any] | None = None
     managed: ManagedValueMapping
     checkpoint: Checkpoint
     checkpoint_id_saved: str
@@ -410,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,
@@ -418,16 +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,
                 )
-            if self._delta_write_futs is not None and any(
-                isinstance(self.specs.get(c), DeltaChannel) for c, _ in writes_to_save
-            ):
-                self._delta_write_futs.append(fut)
         # output writes
         if hasattr(self, "tasks"):
             self.output_writes(task_id, writes)
@@ -898,10 +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,
-            force_delta_snapshot=exiting and self.durability == "exit",
         )
         # sanitize TASK channel in the checkpoint before saving (durability=="exit")
         if TASKS in self.checkpoint["channel_values"] and any(
@@ -1383,11 +1372,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         metadata: CheckpointMetadata,
         new_versions: ChannelVersions,
     ) -> RunnableConfig:
-        # Drain DeltaChannel write futures before committing the checkpoint so
-        # DELTA_SENTINEL blobs are never saved ahead of their backing writes.
-        if self._delta_write_futs:
-            futs, self._delta_write_futs = self._delta_write_futs, []
-            await asyncio.gather(*futs)
         try:
             if prev is not None:
                 await prev
@@ -1493,7 +1477,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
-        self._delta_write_futs = []
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.0a2"
+version = "1.1.9"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -24,8 +24,8 @@ classifiers = [
     'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.3.2,<2",
-    "langgraph-checkpoint>=4.0.3,<5.0.0",
+    "langchain-core>=1.3.0,<2",
+    "langgraph-checkpoint>=2.1.0,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
     "langgraph-prebuilt>=1.0.9,<1.1.0",
     "xxhash>=3.5.0",

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

@@ -1,34 +1,22 @@
 import operator
 from collections.abc import Sequence
-from typing import Annotated
 
 import pytest
-from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.checkpoint.base import DELTA_SENTINEL
-from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-from typing_extensions import NotRequired, TypedDict
 
 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
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
-from langgraph.graph import START, StateGraph
-from langgraph.graph.message import _messages_delta_reducer
-from langgraph.graph.state import _get_channel
-from langgraph.types import Overwrite
+from langgraph.graph.message import add_messages
 
 pytestmark = pytest.mark.anyio
 
 
-# ---------------------------------------------------------------------------
-# Core channel primitives
-# ---------------------------------------------------------------------------
-
-
 def test_last_value() -> None:
     channel = LastValue(int).from_checkpoint(MISSING)
     assert channel.ValueType is int
@@ -111,41 +99,50 @@ def test_untracked_value() -> None:
     assert channel.ValueType is dict
     assert channel.UpdateType is dict
 
+    # UntrackedValue should start empty
     with pytest.raises(EmptyChannelError):
         channel.get()
 
+    # Should be able to update with a value
     test_data = {"session": "test", "temp": "dir"}
     channel.update([test_data])
     assert channel.get() == test_data
 
+    # Update with new value
     new_data = {"session": "updated", "temp": "newdir"}
     channel.update([new_data])
     assert channel.get() == new_data
 
+    # On checkpoint, UntrackedValue should return MISSING
     checkpoint = channel.checkpoint()
     assert checkpoint is MISSING
 
+    # Creating from checkpoint with MISSING should start empty
     new_channel = UntrackedValue(dict).from_checkpoint(checkpoint)
     with pytest.raises(EmptyChannelError):
         new_channel.get()
 
 
-# ---------------------------------------------------------------------------
-# DeltaChannel — message reducer
-# ---------------------------------------------------------------------------
-
-
 def test_delta_channel_basic_two_steps() -> None:
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    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)
+
+    # Step 1: one message added
     ch.update([HumanMessage(content="hi", id="h1")])
     d1 = ch.checkpoint()
     assert d1 is DELTA_SENTINEL
 
+    # Step 2: another message
     ch.update([AIMessage(content="hello", id="a1")])
     d2 = ch.checkpoint()
     assert d2 is DELTA_SENTINEL
 
+    # Full accumulated value is preserved in memory
     assert len(ch.get()) == 2
     assert ch.get()[0].content == "hi"
     assert ch.get()[1].content == "hello"
@@ -153,7 +150,12 @@ def test_delta_channel_basic_two_steps() -> None:
 
 def test_delta_channel_from_checkpoint_writes_list() -> None:
     """replay_writes on a fresh channel replays through the operator."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import AIMessage, HumanMessage
+
+    from langgraph.channels._delta import DeltaChannel
+    from langgraph.graph.message import add_messages
+
+    spec = DeltaChannel(add_messages)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
     ch.replay_writes(
         [
@@ -170,28 +172,48 @@ def test_delta_channel_from_checkpoint_writes_list() -> None:
 
 
 def test_delta_channel_from_checkpoint_backwards_compat() -> None:
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    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
+    spec = DeltaChannel(add_messages)
     old_value = [HumanMessage(content="old", id="h1")]
     ch = spec.from_checkpoint(old_value)
     assert ch.get() == old_value
 
 
 def test_delta_channel_overwrite() -> None:
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    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
+
+    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
     ch.update([HumanMessage(content="old", id="h1")])
 
     ch.update([Overwrite([HumanMessage(content="new", id="h2")])])
     d = ch.checkpoint()
     assert d is DELTA_SENTINEL
+    # After overwrite, value is reset to only the new message
     assert len(ch.get()) == 1
     assert ch.get()[0].content == "new"
 
 
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    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)
     ch = spec.from_checkpoint(MISSING)
 
+    # Step 1: add two messages
     ch.update([HumanMessage(content="hi", id="h1")])
     ch.update([AIMessage(content="hello", id="a1")])
     assert ch.get() == [
@@ -199,9 +221,11 @@ def test_delta_channel_remove_message_and_replay() -> None:
         AIMessage(content="hello", id="a1"),
     ]
 
+    # Step 2: remove the AI message
     ch.update([RemoveMessage(id="a1")])
     assert ch.get() == [HumanMessage(content="hi", id="h1")]
 
+    # Replay the writes list from scratch — must reproduce the post-remove state
     ch2 = spec.from_checkpoint(DELTA_SENTINEL)
     ch2.replay_writes(
         [
@@ -215,13 +239,22 @@ def test_delta_channel_remove_message_and_replay() -> None:
 
 def test_delta_channel_update_by_id_and_replay() -> None:
     """Updating a message by ID must round-trip correctly through writes replay."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import HumanMessage
+
+    from langgraph.channels._delta import DeltaChannel
+    from langgraph.graph.message import add_messages
+
+    spec = DeltaChannel(add_messages)
     ch = spec.from_checkpoint(MISSING)
 
+    # Step 1: add a message
     ch.update([HumanMessage(content="original", id="h1")])
+
+    # Step 2: update the same message by ID
     ch.update([HumanMessage(content="updated", id="h1")])
     assert ch.get() == [HumanMessage(content="updated", id="h1")]
 
+    # Replay writes — must produce the updated message, not the original
     ch2 = spec.from_checkpoint(DELTA_SENTINEL)
     ch2.replay_writes(
         [
@@ -235,123 +268,34 @@ def test_delta_channel_update_by_id_and_replay() -> None:
 
 def test_delta_channel_checkpoint_returns_sentinel() -> None:
     """checkpoint() always returns DELTA_SENTINEL regardless of state."""
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    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)
     assert ch.checkpoint() is DELTA_SENTINEL
 
+    from langchain_core.messages import HumanMessage
+
     ch.update([HumanMessage(content="hi", id="h1")])
     assert ch.checkpoint() is DELTA_SENTINEL
 
 
-# ---------------------------------------------------------------------------
-# DeltaChannel — snapshot frequency
-# ---------------------------------------------------------------------------
-
-
-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.
-    """
-
-    class State(TypedDict):
-        messages: Annotated[
-            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=5)
-        ]
-        other: str
-
-    def node_a(state: State) -> dict:
-        i = len(state["messages"]) // 2
-        return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
-
-    def node_b(state: State) -> dict:
-        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,
-        )
-
-    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"
-
-    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).
-    """
-
-    class State(TypedDict):
-        messages: Annotated[
-            list, DeltaChannel(_messages_delta_reducer, 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:
-        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,
-        )
-
-    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)}
-    assert snapshots, (
-        "eager snapshots must fire even on steps where messages wasn't written"
-    )
-
-    state = graph.get_state(config)
-    assert len(state.values["messages"]) == 10  # 5 human + 5 AI
-
-
-# ---------------------------------------------------------------------------
-# DeltaChannel — end-to-end (InMemorySaver)
-# ---------------------------------------------------------------------------
-
-
 def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     """InMemorySaver assembles writes from checkpoint_writes inside get_tuple."""
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    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
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer, list)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     n = {"v": 0}
 
@@ -369,6 +313,9 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     graph.invoke({"messages": [HumanMessage(content="hi", id="h1")]}, config)
     graph.invoke({"messages": [HumanMessage(content="bye", id="h2")]}, config)
 
+    # get_tuple returns raw storage shape — channel_values stores DELTA_SENTINEL
+    # for delta channels; the reconstructed writes flow separately via
+    # saver._get_channel_writes_history.
     saved = saver.get_tuple(config)
     assert saved is not None
     assert "messages" in saved.checkpoint["channel_values"]
@@ -378,38 +325,34 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
 
 
-# ---------------------------------------------------------------------------
-# DeltaChannel — dict reducer
-# ---------------------------------------------------------------------------
-
-
-def _delta_channel_with_type(op, typ):
+def _delta_channel_with_type(operator, typ):
     """Build a DeltaChannel with an explicit type via the Annotated injection path."""
-    return _get_channel("_test", Annotated[typ, DeltaChannel(op)])
+    from typing import Annotated
+
+    from langgraph.channels._delta import DeltaChannel
+    from langgraph.graph.state import _get_channel
+
+    return _get_channel("_test", Annotated[typ, DeltaChannel(operator)])
 
 
 def test_delta_channel_dict_reducer_fresh_channel() -> None:
     """DeltaChannel with a dict reducer starts as empty dict on MISSING checkpoint."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        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(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
 
@@ -427,11 +370,8 @@ def test_delta_channel_dict_reducer_basic_updates() -> None:
 def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
     """replay_writes on a fresh channel replays through a dict merge reducer."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -446,23 +386,29 @@ 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(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            for k, v in w.items():
-                if v is None:
-                    result.pop(k, None)
-                else:
-                    result[k] = v
+    def merge_files(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return {k: v for k, v in right.items() if v is not None}
+        result = {**left}
+        for k, v in right.items():
+            if v is None:
+                result.pop(k, None)
+            else:
+                result[k] = v
         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(
@@ -476,27 +422,24 @@ def test_delta_channel_dict_reducer_with_deletions() -> None:
 
 def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
     """Overwrite(dict) in update() must preserve dict shape, not coerce to list."""
+    from langgraph.types import Overwrite
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     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}
 
 
 def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
     """Overwrite(dict) embedded in replayed writes must reconstruct as dict."""
+    from langgraph.types import Overwrite
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -511,15 +454,28 @@ 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]`.
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    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
 
-    annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
+    from typing_extensions import NotRequired
+
+    from langgraph.channels._delta import DeltaChannel
+    from langgraph.graph.state import _get_channel
+
+    def merge_dicts(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return dict(right)
+        return {**left, **right}
+
+    annotation = Annotated[
+        NotRequired[dict[str, int]],
+        DeltaChannel(merge_dicts),
+    ]
     ch = _get_channel("files", annotation).from_checkpoint(MISSING)
     assert ch.get() == {}
     ch.update([{"a": 1}])
@@ -528,16 +484,28 @@ 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.
 
-    def merge_files(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            for k, v in w.items():
-                if v is None:
-                    result.pop(k, None)
-                else:
-                    result[k] = v
+    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.graph import START, StateGraph
+
+    def merge_files(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return {k: v for k, v in right.items() if v is not None}
+        result = {**left}
+        for k, v in right.items():
+            if v is None:
+                result.pop(k, None)
+            else:
+                result[k] = v
         return result
 
     class State(TypedDict):
@@ -560,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",
@@ -570,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}}
 
@@ -590,11 +562,8 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
 def test_delta_channel_dict_reducer_backwards_compat() -> None:
     """A pre-DeltaChannel dict checkpoint must load as a dict, not be listified."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     old_value = {"a": 1, "b": 2}
@@ -603,7 +572,7 @@ def test_delta_channel_dict_reducer_backwards_compat() -> None:
 
 
 # ---------------------------------------------------------------------------
-# DeltaChannel — seed / pre-delta migration
+# seed / pre-delta migration
 # ---------------------------------------------------------------------------
 
 
@@ -614,7 +583,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None:
     a pre-DeltaChannel blob it passes it as `seed` so replay reconstructs
     the post-migration state correctly rather than replaying from empty.
     """
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    spec = DeltaChannel(add_messages)
     seed = [HumanMessage(content="pre-delta", id="p1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes(
@@ -630,7 +599,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None:
 def test_delta_channel_from_checkpoint_seed_without_writes() -> None:
     """Reconstruction at a pre-delta ancestor with no newer deltas returns
     just the seed — the saver's terminator fired immediately."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    spec = DeltaChannel(add_messages)
     seed = [HumanMessage(content="only-snap", id="s1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes([])
@@ -644,10 +613,11 @@ def test_delta_channel_from_checkpoint_seed_none_is_distinct_from_sentinel() ->
     explicitly should feed None to the reducer as the left operand.
     """
 
-    def replace(state, writes):
-        return writes[-1] if writes else state
+    def replace(left, right):
+        return right
 
-    spec = DeltaChannel(replace, list)
+    spec = DeltaChannel(replace)
     ch = spec.from_checkpoint(None)
     ch.replay_writes([("t0", "x", "after")])
+    # Reducer replaces; seed=None → first write produces "after".
     assert ch.get() == "after"

libs/langgraph/tests/test_delta_channel_benchmark.py

@@ -1,23 +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 os
 import sys
 import time
 from typing import Annotated, Any
@@ -27,17 +26,23 @@ 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 _messages_delta_reducer, add_messages
+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 = os.environ.get(
-        "LANGGRAPH_BENCH_POSTGRES_URI",
-        "postgres://postgres@localhost:5432/postgres?sslmode=disable",
+    _POSTGRES_URI = (
+        "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
     )
 except ImportError:
     _POSTGRES_AVAILABLE = False
@@ -118,21 +123,7 @@ class BinaryState(TypedDict):
 
 
 class DeltaState(TypedDict):
-    messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-
-
-def _make_delta_state(snapshot_frequency: int | float) -> type:
-    """Create a TypedDict with DeltaChannel at the given snapshot_frequency."""
-    channel = DeltaChannel(
-        _messages_delta_reducer, 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]},
-    )
+    messages: Annotated[list, DeltaChannel(add_messages)]
 
 
 # ---------------------------------------------------------------------------
@@ -178,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"}}
@@ -192,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
 
 
@@ -214,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"
@@ -223,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:
@@ -283,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"
@@ -303,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)
@@ -438,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

@@ -46,14 +46,12 @@ import operator
 from typing import Annotated, Any
 
 import pytest
-from langchain_core.messages import AIMessage, HumanMessage
 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
-from langgraph.graph.message import _messages_delta_reducer, add_messages
 
 pytestmark = pytest.mark.anyio
 
@@ -72,13 +70,6 @@ def _noop(_state: Any) -> dict:
     return {}
 
 
-def _list_concat(state: list, writes: list) -> list:
-    result = list(state)
-    for w in writes:
-        result.extend(w if isinstance(w, list) else [w])
-    return result
-
-
 def _binop_graph(checkpointer: Any) -> Any:
     class BinopState(TypedDict):
         items: Annotated[list, BinaryOperatorAggregate(list, operator.add)]
@@ -94,7 +85,7 @@ def _binop_graph(checkpointer: Any) -> Any:
 
 def _delta_graph(checkpointer: Any) -> Any:
     class DeltaState(TypedDict):
-        items: Annotated[list, DeltaChannel(_list_concat)]
+        items: Annotated[list, DeltaChannel(operator.add)]
 
     return (
         StateGraph(DeltaState)
@@ -511,103 +502,3 @@ def test_fork_from_update_state_checkpoint() -> None:
         f"fork lost update_state base: base={base_items}, forked={forked_items}"
     )
     assert forked_items[-1] == "fork0", f"fork delta not appended: {forked_items}"
-
-
-# ---------------------------------------------------------------------------
-# 9. Migration from `add_messages` → `DeltaChannel(_messages_delta_reducer)`
-#
-# `add_messages` is the primary real-world use case: it creates a
-# BinaryOperatorAggregate with dedup-by-ID and RemoveMessage semantics.
-# After swapping the annotation to DeltaChannel, pre-migration blobs
-# (plain lists of Message objects) must be used directly as the seed.
-# ---------------------------------------------------------------------------
-
-
-def _add_messages_graph(checkpointer: Any) -> Any:
-    class MessagesState(TypedDict):
-        messages: Annotated[list, add_messages]
-
-    return (
-        StateGraph(MessagesState)
-        .add_node("noop", _noop)
-        .add_edge(START, "noop")
-        .add_edge("noop", END)
-        .compile(checkpointer=checkpointer)
-    )
-
-
-def _delta_messages_graph(checkpointer: Any) -> Any:
-    class DeltaMessagesState(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-
-    return (
-        StateGraph(DeltaMessagesState)
-        .add_node("noop", _noop)
-        .add_edge(START, "noop")
-        .add_edge("noop", END)
-        .compile(checkpointer=checkpointer)
-    )
-
-
-def test_add_messages_to_delta_migration_preserves_message_history() -> None:
-    """Migration from `add_messages` to `DeltaChannel(_messages_delta_reducer)`
-    preserves message ordering and IDs at both the tip and settled ancestor
-    boundaries.
-
-    The pre-migration blob is a plain list of Message objects; DeltaChannel
-    must use it directly as the seed without walking ancestors past it.
-    """
-    checkpointer = InMemorySaver()
-    config = {"configurable": {"thread_id": "add-messages-migration"}}
-
-    pre_graph = _add_messages_graph(checkpointer)
-    pre_graph.invoke({"messages": [HumanMessage(content="hello", id="h1")]}, config)
-    pre_graph.invoke({"messages": [AIMessage(content="hi", id="a1")]}, config)
-    pre_graph.invoke({"messages": [HumanMessage(content="thanks", id="h2")]}, config)
-
-    pre_tip = pre_graph.get_state(config)
-    assert [m.id for m in pre_tip.values["messages"]] == ["h1", "a1", "h2"]
-
-    delta_graph = _delta_messages_graph(checkpointer)
-
-    # Tip: latest checkpoint has a full list blob — must use it directly.
-    snap = delta_graph.get_state(config)
-    assert [m.id for m in snap.values["messages"]] == ["h1", "a1", "h2"], (
-        f"tip hydration mismatch: got {[m.id for m in snap.values['messages']]}"
-    )
-
-    # Settled ancestor boundaries must also match.
-    pre_settled = [
-        [m.id for m in s.values.get("messages", [])]
-        for s in pre_graph.get_state_history(config)
-        if s.next == ("__start__",)
-    ]
-    delta_settled = [
-        [m.id for m in s.values.get("messages", [])]
-        for s in delta_graph.get_state_history(config)
-        if s.next == ("__start__",)
-    ]
-    assert delta_settled == pre_settled, (
-        f"settled boundary mismatch after migration: "
-        f"pre={pre_settled}, delta={delta_settled}"
-    )
-
-
-async def test_add_messages_to_delta_migration_preserves_message_history_async() -> (
-    None
-):
-    """Async variant of the add_messages migration test."""
-    checkpointer = InMemorySaver()
-    config = {"configurable": {"thread_id": "add-messages-migration-async"}}
-
-    pre_graph = _add_messages_graph(checkpointer)
-    await pre_graph.ainvoke(
-        {"messages": [HumanMessage(content="hello", id="h1")]}, config
-    )
-    await pre_graph.ainvoke({"messages": [AIMessage(content="hi", id="a1")]}, config)
-
-    delta_graph = _delta_messages_graph(checkpointer)
-    snap = await delta_graph.aget_state(config)
-    assert [m.id for m in snap.values["messages"]] == ["h1", "a1"], (
-        f"async tip hydration mismatch: got {[m.id for m in snap.values['messages']]}"
-    )

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

@@ -16,7 +16,7 @@ from typing import Annotated, Any, Literal, get_type_hints
 
 import pytest
 from langchain_core.language_models import GenericFakeChatModel
-from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
 from langchain_core.runnables import (
     RunnableConfig,
     RunnableLambda,
@@ -25,7 +25,6 @@ from langchain_core.runnables import (
 from langchain_core.runnables.graph import Edge
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
-    DELTA_SENTINEL,
     BaseCheckpointSaver,
     Checkpoint,
     CheckpointMetadata,
@@ -42,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
@@ -51,7 +49,7 @@ from langgraph.config import get_stream_writer
 from langgraph.errors import GraphRecursionError, InvalidUpdateError, ParentCommand
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph
-from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
+from langgraph.graph.message import MessagesState, add_messages
 from langgraph.pregel import (
     NodeBuilder,
     Pregel,
@@ -9406,9 +9404,15 @@ def test_fork_does_not_apply_pending_writes(
 
 async def test_delta_channel_end_to_end_inmemory() -> None:
     """Full graph run: DeltaChannel accumulates correctly across multiple turns."""
+    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
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         n = len(state["messages"])
@@ -9442,9 +9446,15 @@ async def test_delta_channel_end_to_end_inmemory() -> None:
 
 async def test_delta_channel_time_travel() -> None:
     """Time-travel back to turn-1 checkpoint and resume; continuation must not include turn-2 deltas."""
+    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
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     counter = {"n": 0}
 
@@ -9494,9 +9504,15 @@ async def test_delta_channel_time_travel() -> None:
 
 async def test_delta_channel_remove_message_end_to_end() -> None:
     """RemoveMessage inside a DeltaChannel graph must persist and reload correctly."""
+    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
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         return {"messages": [AIMessage(content="reply", id="ai-1")]}
@@ -9535,9 +9551,15 @@ async def test_delta_channel_remove_message_end_to_end() -> None:
 
 async def test_delta_channel_update_by_id_end_to_end() -> None:
     """Updating a message by ID via DeltaChannel must persist and reload correctly."""
+    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
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def update_msg(state: State) -> dict:
         # re-send h1 with updated content
@@ -9565,91 +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_durability_exit_stores_snapshot() -> None:
-    """DeltaChannel must reload from a durability='exit' checkpoint."""
-
-    class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-
-    def respond(state: State) -> dict:
-        return {"messages": [AIMessage(content="reply", id="ai1")]}
-
-    builder = StateGraph(State)
-    builder.add_node("respond", respond)
-    builder.add_edge(START, "respond")
-    graph = builder.compile(checkpointer=InMemorySaver())
-    config = {"configurable": {"thread_id": "delta-exit-test"}}
-
-    result = graph.invoke(
-        {"messages": [HumanMessage(content="hello", id="h1")]},
-        config,
-        durability="exit",
-    )
-    assert [m.content for m in result["messages"]] == ["hello", "reply"]
-
-    state = graph.get_state(config)
-    assert [m.content for m in state.values["messages"]] == ["hello", "reply"]
-
-
-async def test_delta_channel_async_write_ordering() -> None:
-    """In async mode, DeltaChannel write futures are awaited before the checkpoint
-    is committed, so aput_writes always precedes aput for sentinel checkpoints."""
-
-    class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-
-    def respond(state: State) -> dict:
-        i = len(state["messages"])
-        return {"messages": [AIMessage(content=f"r{i}", id=f"ai{i}")]}
-
-    order: list[str] = []
-    original_aput_writes = InMemorySaver.aput_writes
-    original_aput = InMemorySaver.aput
-
-    async def tracked_aput_writes(self, config, writes, task_id, task_path=""):
-        result = await original_aput_writes(self, config, writes, task_id, task_path)
-        order.append("aput_writes")
-        return result
-
-    async def tracked_aput(self, config, checkpoint, metadata, new_versions):
-        has_sentinel = any(
-            v is DELTA_SENTINEL for v in checkpoint.get("channel_values", {}).values()
-        )
-        order.append("aput_sentinel" if has_sentinel else "aput_other")
-        return await original_aput(self, config, checkpoint, metadata, new_versions)
-
-    InMemorySaver.aput_writes = tracked_aput_writes
-    InMemorySaver.aput = tracked_aput
-    try:
-        builder = StateGraph(State)
-        builder.add_node("respond", respond)
-        builder.add_edge(START, "respond")
-        graph = builder.compile(checkpointer=InMemorySaver())
-        config = {"configurable": {"thread_id": "async-ordering-test"}}
-
-        for i in range(3):
-            await graph.ainvoke(
-                {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")]}, config
-            )
-
-        # Every aput_sentinel must be preceded by at least one aput_writes
-        for i, event in enumerate(order):
-            if event == "aput_sentinel":
-                preceding = order[:i]
-                assert "aput_writes" in preceding, (
-                    f"aput_sentinel at {i} had no preceding aput_writes: {order}"
-                )
-                last_write_idx = max(
-                    j for j, e in enumerate(order[:i]) if e == "aput_writes"
-                )
-                assert last_write_idx < i, (
-                    f"aput_writes at {last_write_idx} should precede aput_sentinel at {i}: {order}"
-                )
-    finally:
-        InMemorySaver.aput_writes = original_aput_writes
-        InMemorySaver.aput = original_aput
-
-    state = await graph.aget_state(config)
-    assert len(state.values["messages"]) == 6  # 3 human + 3 AI

libs/langgraph/tests/test_pregel_async.py

@@ -6101,36 +6101,6 @@ async def test_parent_command(
     )
 
 
-async def test_delta_channel_durability_exit_stores_snapshot_async() -> None:
-    """DeltaChannel must reload from an async durability='exit' checkpoint."""
-    from langchain_core.messages import AIMessage
-
-    from langgraph.channels.delta import DeltaChannel
-    from langgraph.graph.message import _messages_delta_reducer
-
-    class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
-
-    async def respond(state: State) -> dict:
-        return {"messages": [AIMessage(content="reply", id="ai1")]}
-
-    builder = StateGraph(State)
-    builder.add_node("respond", respond)
-    builder.add_edge(START, "respond")
-    graph = builder.compile(checkpointer=InMemorySaver())
-    config = {"configurable": {"thread_id": "delta-exit-async-test"}}
-
-    result = await graph.ainvoke(
-        {"messages": [HumanMessage(content="hello", id="h1")]},
-        config,
-        durability="exit",
-    )
-    assert [m.content for m in result["messages"]] == ["hello", "reply"]
-
-    state = await graph.aget_state(config)
-    assert [m.content for m in state.values["messages"]] == ["hello", "reply"]
-
-
 @NEEDS_CONTEXTVARS
 async def test_interrupt_subgraph(async_checkpointer: BaseCheckpointSaver) -> None:
     class State(TypedDict):

libs/langgraph/uv.lock

@@ -7,9 +7,6 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 
-[options]
-prerelease-mode = "allow"
-
 [[package]]
 name = "aiosqlite"
 version = "0.22.1"
@@ -1351,11 +1348,10 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.3.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
-    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -1364,26 +1360,14 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/92/fe/20190232d9b513242899dbb0c2bb77e31b4d61e343743adbe90ebc2603d2/langchain_core-1.3.0.tar.gz", hash = "sha256:14a39f528bf459aa3aa40d0a7f7f1bae7520d435ef991ae14a4ceb74d8c49046", size = 860755, upload-time = "2026-04-17T14:51:38.298Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.14"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/05/bf/efb5e2ed832e4d6d45590e25a9e5191986b291b543bc6a807b48bee070b0/langchain_protocol-0.0.14.tar.gz", hash = "sha256:bc1e8553122e6ede310280462d5813023a172ff2785ccbbdec54d43f3a15e5f2", size = 5862, upload-time = "2026-04-29T16:40:18.657Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
+    { url = "https://files.pythonhosted.org/packages/f8/e2/dbfa347aa072a6dc4cd38d6f9ebfc730b4c14c258c47f480f4c5c546f177/langchain_core-1.3.0-py3-none-any.whl", hash = "sha256:baf16ee028475df177b9ab8869a751c79406d64a6f12125b93802991b566cced", size = 515140, upload-time = "2026-04-17T14:51:36.274Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a2"
+version = "1.1.9"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1455,7 +1439,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.2,<2" },
+    { name = "langchain-core", specifier = ">=1.3.0,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1564,7 +1548,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a2"
+version = "4.0.2"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

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"}

notes/delta_channel_query_bench.md

@@ -0,0 +1,137 @@
+# Delta-channel reconstruction: query strategy benchmark
+
+**Branch:** `delta-channel-writes-based`
+**Question (Nuno):** Is the recursive CTE the right query shape for reconstructing a delta channel inside `get_tuple`, or would a plain `SELECT WHERE` be cheaper even though it returns more rows?
+**Answer:** Plain `SELECT WHERE` wins at every realistic depth. The recursion isn't the problem — the JSON-expression join inside the CTE is.
+
+## Setup
+
+- Postgres 16 on `localhost:5441` (the `compose-postgres.yml` instance, run directly without docker for this round)
+- Single delta channel `messages`, one write per checkpoint, `DELTA_SENTINEL` blob per checkpoint
+- Linear chain (`branch=1`) and 5-way branching at every step (`branch=5`) — branching is the case where plain over-fetches sibling rows
+- Median of 20 timed runs after 3 warmups, fresh psycopg cursor per strategy
+- Bench script: `bench_get_tuple_strategies.py` at repo root
+
+Three strategies compared:
+
+| name | roundtrips | shape |
+|------|-----------|-------|
+| `cte` | 1 | Current prod: recursive CTE walks ancestors, LEFT JOINs writes + blobs |
+| `plain` | 3 | Nuno's suggestion: thread-wide `SELECT WHERE` per table, Python walks parent chain and filters |
+| `cte+narrow` | 2 | CTE returns ancestor IDs only, then one `UNION ALL` of writes + blobs filtered by `ANY(ids)` |
+
+## Results (ms per get_tuple, median of 20)
+
+```
+ depth  branch       cte       plain    cte+narrow   rows_cte   rows_plain   plain/cte
+    10       1     0.14ms      0.26ms       0.24ms          9           30      1.89x
+    10       5     0.21ms      0.23ms       0.17ms          9          110      1.11x
+    50       1     0.89ms      0.27ms       0.33ms         49          150      0.31x
+    50       5     2.35ms      0.66ms       0.51ms         49          550      0.28x
+   200       1    11.61ms      0.78ms       1.30ms        199          600      0.07x
+   200       5    34.79ms      2.33ms       3.07ms        199         2200      0.07x
+  1000       1   274.60ms      2.59ms      13.29ms        999         3000      0.01x
+  1000       5   856.01ms     10.14ms      15.31ms        999        11000      0.01x
+```
+
+Lower is better. `plain/cte < 1` means plain is faster.
+
+### Headline numbers
+
+- depth 50: plain is **3x** faster
+- depth 200: plain is **15x** faster
+- depth 1000: plain is **~100x** faster
+- Branching makes plain over-fetch (3000 rows → 11000 rows at d=1000), but it remains ~85x faster than the CTE
+
+## Why the CTE collapses
+
+`EXPLAIN (ANALYZE, BUFFERS)` of the CTE at depth 1000 (linear). Excerpt with the load-bearing nodes:
+
+```
+Sort  ... actual time=137.798..137.827 rows=999
+  CTE ancestors
+    ->  Recursive Union  ... actual time=0.005..2.443 rows=999
+                                                 ^^^^^^
+                                                 recursion is 2.4 ms — fine
+  ->  Nested Loop Left Join  ... actual time=2.676..137.529 rows=999
+        Join Filter: (cw.checkpoint_id = a.cid)
+        Rows Removed by Join Filter: 998001
+                                     ^^^^^^^
+                                     999 ancestors x ~1000 writes
+        ->  Nested Loop Left Join  ... actual time=2.669..85.061 rows=999
+              Join Filter: (bl.version = ((c.checkpoint -> 'channel_versions'::text) ->> bl.channel))
+              Rows Removed by Join Filter: 998001
+                                           ^^^^^^^
+                                           same quadratic blow-up on the blob join
+```
+
+Two pathological things are happening:
+
+1. **The blob join filter is on a JSON expression**: `bl.version = (c.checkpoint -> 'channel_versions' ->> bl.channel)`. The planner cannot push this into an index lookup, so it materializes `checkpoint_blobs` for the thread and does a nested-loop comparison against every ancestor — a Cartesian product that grows as `O(ancestors × blobs_in_thread)`.
+2. **The writes join is similar**: writes for the thread are materialized once, then nested-loop joined against ancestors with a `Join Filter` rather than a hash/merge join over the indexed `checkpoint_id`.
+
+At depth 1000 that's **~2 million rows evaluated, 99.9% of them discarded**. The recursion itself is a rounding error.
+
+For comparison, the plain Q1 (`SELECT … FROM checkpoints WHERE thread_id=? AND checkpoint_ns=?`) at depth 1000:
+
+```
+Seq Scan on checkpoints  ... actual time=0.012..0.121 rows=1000
+Execution Time: 0.140 ms
+```
+
+A simple seq scan over 57 buffers. Q2 and Q3 follow the same shape and complete in well under 1 ms each.
+
+## Crossover and remote-DB reasoning
+
+- Pure local Postgres: plain wins from depth ~30 onward; CTE wins by fractions of a ms below that
+- Remote Postgres at ~5 ms RTT adds ~10 ms to plain (3 roundtrips vs 1). Crossover shifts to ~depth 30. Above that, the CTE's quadratic SQL cost still dominates the RTT savings.
+
+There is no realistic conversation depth where the CTE wins on a remote DB. At depth 200+ (anything resembling a real multi-turn agent run) plain is faster regardless of network.
+
+## Recommendation
+
+**Switch to plain SELECT WHERE, one delta channel at a time.**
+
+Three indexed queries per delta channel:
+
+```sql
+-- Q1: parent chain + per-checkpoint version of this channel
+SELECT checkpoint_id,
+       parent_checkpoint_id,
+       checkpoint -> 'channel_versions' ->> 'channel_name' AS ver
+FROM checkpoints
+WHERE thread_id = ? AND checkpoint_ns = ?;
+
+-- Q2: writes for this channel, anywhere in the thread
+SELECT checkpoint_id, type, blob, task_id, idx
+FROM checkpoint_writes
+WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ?;
+
+-- Q3: blobs for this channel, anywhere in the thread
+SELECT version, type, blob
+FROM checkpoint_blobs
+WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ?;
+```
+
+Python then:
+- Builds `parent_of: dict[cid, parent_cid]` from Q1
+- Walks from target's parent newest → oldest
+- Filters Q2 rows by `ancestor_set`, processes oldest → newest, applies overwrite-terminator
+- Picks seed blob via the per-ancestor `ver` map, terminates at first non-sentinel blob
+
+All O(n) on n = thread checkpoints, with tight constants (dict lookups). No recursion, no JSON-expression joins, no quadratic plans.
+
+If the 3-roundtrip cost ever shows up on remote-DB benchmarks, fold Q2 + Q3 into one `UNION ALL` to get back to 2 roundtrips. Bench says it isn't worth the SQL complexity right now.
+
+## Bonus: code simplification from single-channel scope
+
+Multi-channel reconstruction in the current `_reconstruct_delta_channels_cur` carries:
+
+- `rows_by_cid` nested dicts, keyed by cid then channel
+- `seen_blob: set[(cid, channel)]` and `seen_write: set[(cid, channel, task_id, idx)]` dedup
+- `collected: dict[channel, list]`, `done: set[channel]`, `seeds: dict[channel, value]`
+- Inner `for ch in channels_list` loops and an early-exit `if len(done) == len(channels_list)`
+
+Single-channel collapses these to a single list, a single bool, and one `Optional[Any]`. Roughly half the Python in that function, plus an obvious shape for splitting pure post-processing into `base.py` so sync and async stop duplicating it.
+
+If multi-channel coalescing turns out to matter later, it can come back as a SQL-level optimization without re-introducing the bookkeeping in Python.