chore: update lockfiles for 1.2.0a2 / 4.1.0a2

- Add test_add_messages_to_delta_migration_preserves_message_history (sync
  + async) covering the primary real-world BinaryOperatorAggregate →
  DeltaChannel migration path with real Message objects and IDs
- Hoist all in-function imports to module level in test_channels.py and
  fix _delta_channel_with_type helper accordingly
- Add section headers in test_channels.py for better navigation

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
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -26,10 +26,9 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_BLOBS_SQL,
-    SELECT_DELTA_PARENTS_SQL,
-    SELECT_DELTA_WRITES_SQL,
+    SELECT_DELTA_COMBINED_SQL,
     BasePostgresSaver,
+    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -442,10 +441,9 @@ class PostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
 
-        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`.
+        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.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
@@ -459,18 +457,25 @@ 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_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()
+            cur.execute(
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
+            )
+            rows = cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
             target_id=checkpoint_id,
-            parents_rows=parents_rows,
-            writes_rows=writes_rows,
-            blobs_rows=blobs_rows,
+            rows=cast("list[_DeltaCombinedRow]", rows),
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
@@ -485,7 +490,6 @@ 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": {
@@ -498,7 +502,7 @@ class PostgresSaver(BasePostgresSaver):
                 **value["checkpoint"],
                 "channel_values": {
                     **(value["checkpoint"].get("channel_values") or {}),
-                    **channel_values,
+                    **self._load_blobs(value["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
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -26,10 +26,9 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_BLOBS_SQL,
-    SELECT_DELTA_PARENTS_SQL,
-    SELECT_DELTA_WRITES_SQL,
+    SELECT_DELTA_COMBINED_SQL,
     BasePostgresSaver,
+    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -403,10 +402,10 @@ class AsyncPostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
 
-        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`.
+        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`.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
@@ -418,23 +417,24 @@ class AsyncPostgresSaver(BasePostgresSaver):
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
         async with self._cursor() as cur:
             await cur.execute(
-                SELECT_DELTA_PARENTS_SQL, (channel, thread_id, checkpoint_ns)
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
             )
-            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()
+            rows = await cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
             target_id=checkpoint_id,
-            parents_rows=parents_rows,
-            writes_rows=writes_rows,
-            blobs_rows=blobs_rows,
+            rows=cast("list[_DeltaCombinedRow]", rows),
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
@@ -449,18 +449,11 @@ 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": thread_id,
-                    "checkpoint_ns": checkpoint_ns,
+                    "thread_id": value["thread_id"],
+                    "checkpoint_ns": value["checkpoint_ns"],
                     "checkpoint_id": value["checkpoint_id"],
                 }
             },
@@ -468,15 +461,15 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 **value["checkpoint"],
                 "channel_values": {
                     **(value["checkpoint"].get("channel_values") or {}),
-                    **channel_values,
+                    **self._load_blobs(value["channel_values"]),
                 },
             },
             value["metadata"],
             (
                 {
                     "configurable": {
-                        "thread_id": thread_id,
-                        "checkpoint_ns": checkpoint_ns,
+                        "thread_id": value["thread_id"],
+                        "checkpoint_ns": value["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, cast
+from typing import Any, TypedDict, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
@@ -155,26 +155,58 @@ INSERT_CHECKPOINT_WRITES_SQL = """
     ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
 """
 
-# 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,
+
+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,
            parent_checkpoint_id,
-           checkpoint -> 'channel_versions' ->> %s AS ver
+           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
     FROM checkpoints
     WHERE thread_id = %s AND checkpoint_ns = %s
-"""
-
-SELECT_DELTA_WRITES_SQL = """
-    SELECT checkpoint_id, type, blob, task_id, idx
+    UNION ALL
+    SELECT 'w',
+           checkpoint_id, NULL, NULL,
+           type, blob, task_id, idx, NULL
     FROM checkpoint_writes
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
-"""
-
-SELECT_DELTA_BLOBS_SQL = """
-    SELECT version, type, blob
+    UNION ALL
+    SELECT 'b',
+           NULL, NULL, NULL,
+           type, blob, NULL, NULL, version
     FROM checkpoint_blobs
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
 """
@@ -212,32 +244,28 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         )
 
     def _load_blobs(
-        self,
-        blob_values: Any,
+        self, blob_values: list[tuple[bytes, bytes, bytes]]
     ) -> dict[str, Any]:
         if not blob_values:
             return {}
-        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
+        return {
+            k.decode(): self.serde.loads_typed((t.decode(), v))
+            for k, t, v in blob_values
+            if t.decode() != "empty"
+        }
 
     def _build_delta_channel_writes_history(
         self,
         *,
         channel: str,
         target_id: str,
-        parents_rows: Sequence[Any],
-        writes_rows: Sequence[Any],
-        blobs_rows: Sequence[Any],
+        rows: Sequence[_DeltaCombinedRow],
     ) -> _ChannelWritesHistory:
-        """Reconstruct one delta channel's history from rows of the three SELECTs.
+        """Reconstruct one delta channel's history from the combined UNION ALL rows.
 
         Pure data transform shared by sync (`PostgresSaver`) and async
-        (`AsyncPostgresSaver`); both paths run the queries themselves and
-        feed the rows here.
+        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
+        and feed the tagged rows here.
 
         Walk is newest → oldest from the target's parent. A non-sentinel
         blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
@@ -248,40 +276,49 @@ 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]]] = {}
-        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"])
-            )
+        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 ws in writes_by_cid.values():
             ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
-        blob_by_ver: dict[str, tuple[str, bytes]] = {
-            r["version"]: (r["type"], r["blob"]) for r in blobs_rows
-        }
+        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=[])
 
         collected: list[PendingWrite] = []  # newest first; reversed at the end
         for cid in ancestors:
-            # Pre-delta blob terminator: subsumes any writes at this ancestor.
+            # 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.
             ver = ver_of.get(cid)
             if ver is not None:
                 seed_blob = blob_by_ver.get(ver)
@@ -290,9 +327,6 @@ 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>=2.1.2,<5.0.0",
+  "langgraph-checkpoint>=4.0.3,<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 add_messages
+    from langgraph.graph.message import _messages_delta_reducer
     from typing_extensions import TypedDict
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
 
     def respond(state: State) -> dict:
         n = len(state["messages"])

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

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import contextvars
 import copy
 import logging
 from collections.abc import AsyncIterator, Collection, Iterator, Mapping, Sequence
@@ -33,14 +32,6 @@ 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__)
 
@@ -495,6 +486,20 @@ 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:
@@ -523,75 +528,61 @@ class BaseCheckpointSaver(Generic[V]):
 
         Underscore-prefixed because the method surface is experimental.
         """
-        # 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)
+        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)
 
     async def _aget_channel_writes_history(
         self, config: RunnableConfig, channel: str
     ) -> _ChannelWritesHistory:
         """Async version of `_get_channel_writes_history`. See docstring there."""
-        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)
+        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)
 
     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, cast
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 
@@ -27,6 +27,7 @@ from langgraph.checkpoint.base import (
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 logger = logging.getLogger(__name__)
 
@@ -185,8 +186,31 @@ class InMemorySaver(
                     if blob_entry is not None and blob_entry[0] != "empty":
                         blob_value = self.serde.loads_typed(blob_entry)
                         if blob_value is not DELTA_SENTINEL:
-                            # Pre-delta snapshot terminator. Skip this
-                            # ancestor's writes — the blob subsumes them.
+                            if isinstance(blob_value, _DeltaSnapshot):
+                                # Step-based snapshot: the blob is state AT this
+                                # ancestor, but the ancestor's pending_writes
+                                # encode the NEXT step's transition and are NOT
+                                # subsumed by the snapshot — collect them first.
+                                step_writes = self.writes.get(
+                                    (thread_id, checkpoint_ns, cp_id), {}
+                                )
+                                for (_task_id, _idx), (
+                                    tid,
+                                    ch,
+                                    serialized,
+                                    _,
+                                ) in sorted(step_writes.items(), reverse=True):
+                                    if ch != channel:
+                                        continue
+                                    collected.append(
+                                        (tid, ch, self.serde.loads_typed(serialized))
+                                    )
+                                collected.reverse()
+                                return _ChannelWritesHistory(
+                                    seed=blob_value, writes=collected
+                                )
+                            # Pre-delta blob: state AT this ancestor already
+                            # subsumes its pending_writes — skip them.
                             collected.reverse()
                             return _ChannelWritesHistory(
                                 seed=blob_value, writes=collected
@@ -231,16 +255,13 @@ 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": channel_values,
+                        "channel_values": self._load_blobs(
+                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
+                        ),
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -264,26 +285,19 @@ 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)
-                resolved_config = cast(
-                    RunnableConfig,
-                    {
+                return CheckpointTuple(
+                    config={
                         "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": channel_values,
+                        "channel_values": self._load_blobs(
+                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
+                        ),
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -378,27 +392,21 @@ class InMemorySaver(
 
                     checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
 
-                    list_config = cast(
-                        RunnableConfig,
-                        {
+                    yield CheckpointTuple(
+                        config={
                             "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": channel_values,
+                            "channel_values": self._load_blobs(
+                                thread_id,
+                                checkpoint_ns,
+                                checkpoint_["channel_versions"],
+                            ),
                         },
                         metadata=metadata,
                         parent_config=(

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

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

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

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

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_jsonplus.py

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

libs/checkpoint/uv.lock

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

libs/langgraph/langgraph/channels/__init__.py

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

libs/langgraph/langgraph/channels/_delta.py

@@ -1,177 +0,0 @@
-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,10 +22,9 @@ __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
 
 
@@ -33,11 +32,22 @@ 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 set(value.keys()) == {OVERWRITE}:
+    if isinstance(value, dict) and len(value) == 1 and OVERWRITE in value:
         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.
 
@@ -68,11 +78,8 @@ class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
             self.value = MISSING
 
     def __eq__(self, value: object) -> bool:
-        return isinstance(value, BinaryOperatorAggregate) and (
-            value.operator is self.operator
-            if value.operator.__name__ != "<lambda>"
-            and self.operator.__name__ != "<lambda>"
-            else True
+        return isinstance(value, BinaryOperatorAggregate) and _operators_equal(
+            self.operator, value.operator
         )
 
     @property

libs/langgraph/langgraph/channels/delta.py

@@ -0,0 +1,197 @@
+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

@@ -244,6 +244,52 @@ def add_messages(
     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,6 +1,5 @@
 from __future__ import annotations
 
-import collections.abc
 import inspect
 import logging
 import typing
@@ -47,9 +46,9 @@ from langgraph._internal._fields import (
 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._delta import DeltaChannel
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.binop import BinaryOperatorAggregate, _strip_extras
+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 (
@@ -1679,27 +1678,11 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None:
                         NotRequired,
                     ):
                         origin = origin.__args__[0]
-                    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 = []
+                    item = item.__class__(
+                        item.reducer,
+                        origin,
+                        snapshot_frequency=item.snapshot_frequency,
+                    )
                 return item
             elif isclass(item) and issubclass(item, BaseChannel):
                 # ex, Annotated[int, EphemeralValue, SomeOtherAnnotation]

libs/langgraph/langgraph/pregel/_checkpoint.py

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

libs/langgraph/langgraph/pregel/_loop.py

@@ -68,6 +68,7 @@ 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 (
@@ -189,6 +190,8 @@ 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
@@ -407,7 +410,7 @@ class PregelLoop:
                     task = self.tasks.get(task_id)
                 else:
                     task = None
-                self.submit(
+                fut = self.submit(
                     self.checkpointer_put_writes,
                     config,
                     writes_to_save,
@@ -415,12 +418,16 @@ class PregelLoop:
                     task_path_str(task.path) if task else "",
                 )
             else:
-                self.submit(
+                fut = self.submit(
                     self.checkpointer_put_writes,
                     config,
                     writes_to_save,
                     task_id,
                 )
+            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)
@@ -891,6 +898,10 @@ 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(
@@ -1372,6 +1383,11 @@ 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
@@ -1477,6 +1493,7 @@ 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.1.9"
+version = "1.2.0a2"
 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.0,<2",
-    "langgraph-checkpoint>=2.1.0,<5.0.0",
+    "langchain-core>=1.3.2,<2",
+    "langgraph-checkpoint>=4.0.3,<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_channels.py

@@ -1,22 +1,34 @@
 import operator
 from collections.abc import Sequence
+from typing import Annotated
 
 import pytest
-from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
 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._delta import DeltaChannel
 from langgraph.channels.binop import BinaryOperatorAggregate
+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.message import add_messages
+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
 
 pytestmark = pytest.mark.anyio
 
 
+# ---------------------------------------------------------------------------
+# Core channel primitives
+# ---------------------------------------------------------------------------
+
+
 def test_last_value() -> None:
     channel = LastValue(int).from_checkpoint(MISSING)
     assert channel.ValueType is int
@@ -99,50 +111,41 @@ 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:
-    from langchain_core.messages import AIMessage, HumanMessage
-    from langgraph.checkpoint.base import DELTA_SENTINEL
+    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
 
-    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"
@@ -150,12 +153,7 @@ 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."""
-    from langchain_core.messages import AIMessage, HumanMessage
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.graph.message import add_messages
-
-    spec = DeltaChannel(add_messages)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
     ch.replay_writes(
         [
@@ -172,48 +170,28 @@ def test_delta_channel_from_checkpoint_writes_list() -> None:
 
 
 def test_delta_channel_from_checkpoint_backwards_compat() -> None:
-    from langchain_core.messages import HumanMessage
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.graph.message import add_messages
-
-    # Old BinaryOperatorAggregate checkpoint: plain list treated as backward compat
-    spec = DeltaChannel(add_messages)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     old_value = [HumanMessage(content="old", id="h1")]
     ch = spec.from_checkpoint(old_value)
     assert ch.get() == old_value
 
 
 def test_delta_channel_overwrite() -> None:
-    from langchain_core.messages import HumanMessage
-    from langgraph.checkpoint.base import DELTA_SENTINEL
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.graph.message import add_messages
-    from langgraph.types import Overwrite
-
-    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
+    ch = DeltaChannel(_messages_delta_reducer, list).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."""
-    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)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     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() == [
@@ -221,11 +199,9 @@ 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(
         [
@@ -239,22 +215,13 @@ 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."""
-    from langchain_core.messages import HumanMessage
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.graph.message import add_messages
-
-    spec = DeltaChannel(add_messages)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     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(
         [
@@ -268,34 +235,123 @@ 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."""
-    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)
+    ch = DeltaChannel(_messages_delta_reducer, list).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
 
 
-def test_delta_channel_inmemory_saver_assembles_writes() -> None:
-    """InMemorySaver assembles writes from checkpoint_writes inside get_tuple."""
-    from typing import Annotated
+# ---------------------------------------------------------------------------
+# DeltaChannel — snapshot frequency
+# ---------------------------------------------------------------------------
 
-    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
+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(add_messages)]
+        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."""
+
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer, list)]
 
     n = {"v": 0}
 
@@ -313,9 +369,6 @@ 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"]
@@ -325,34 +378,38 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
 
 
-def _delta_channel_with_type(operator, typ):
+# ---------------------------------------------------------------------------
+# DeltaChannel — dict reducer
+# ---------------------------------------------------------------------------
+
+
+def _delta_channel_with_type(op, typ):
     """Build a DeltaChannel with an explicit type via the Annotated injection path."""
-    from typing import Annotated
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.graph.state import _get_channel
-
-    return _get_channel("_test", Annotated[typ, DeltaChannel(operator)])
+    return _get_channel("_test", Annotated[typ, DeltaChannel(op)])
 
 
 def test_delta_channel_dict_reducer_fresh_channel() -> None:
     """DeltaChannel with a dict reducer starts as empty dict on MISSING checkpoint."""
 
-    def merge_dicts(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
-    # Should be available (not raise EmptyChannelError) and start empty
     assert ch.is_available()
     assert ch.get() == {}
 
 
 def test_delta_channel_dict_reducer_basic_updates() -> None:
     """DeltaChannel with a dict reducer accumulates key/value pairs across steps."""
-    from langgraph.checkpoint.base import DELTA_SENTINEL
 
-    def merge_dicts(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
 
@@ -370,8 +427,11 @@ 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(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -386,29 +446,23 @@ def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
 
 
 def test_delta_channel_dict_reducer_with_deletions() -> None:
-    """Dict reducer that treats None values as deletions works end-to-end (deepagents pattern)."""
+    """Dict reducer that treats None values as deletions works end-to-end."""
 
-    def merge_files(left: dict | None, right: dict) -> dict:
-        if left is None:
-            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
+    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
         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(
@@ -422,24 +476,27 @@ 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(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     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(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -454,28 +511,15 @@ def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
 
 
 def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
-    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`.
+    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`."""
 
-    This is the shape the deepagents filesystem middleware uses for its
-    `files` field; without unwrapping NotRequired we'd fall through to `list`
-    and blow up on the first dict operator call.
-    """
-    from typing import Annotated
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
-    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),
-    ]
+    annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
     ch = _get_channel("files", annotation).from_checkpoint(MISSING)
     assert ch.get() == {}
     ch.update([{"a": 1}])
@@ -484,28 +528,16 @@ def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
 
 
 def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
-    """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel.
+    """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel."""
 
-    Mirrors the deepagents filesystem pattern: `files: Annotated[dict, reducer]`
-    where the reducer merges dicts and treats None values as deletions.
-    """
-    from typing import Annotated
-
-    from langgraph.checkpoint.memory import InMemorySaver
-    from typing_extensions import TypedDict
-
-    from langgraph.channels._delta import DeltaChannel
-    from langgraph.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
+    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
         return result
 
     class State(TypedDict):
@@ -528,12 +560,9 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
     for _ in range(3):
         graph.invoke({"files": {}}, config)
 
-    # Checkpoint stores only the sentinel — per-step writes live in checkpoint_writes.
     saved = saver.get_tuple(config)
     assert saved is not None
-    cv = saved.checkpoint["channel_values"]["files"]
-    assert cv is DELTA_SENTINEL
-
+    assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
     state = graph.get_state(config)
     assert state.values["files"] == {
         "/doc_1.txt": "content for turn 1",
@@ -541,7 +570,6 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
         "/doc_3.txt": "content for turn 3",
     }
 
-    # Deletion path must round-trip through writes replay.
     def delete_file(state: State) -> dict:
         return {"files": {"/doc_1.txt": None}}
 
@@ -562,8 +590,11 @@ 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(left: dict, right: dict) -> dict:
-        return {**left, **right}
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     old_value = {"a": 1, "b": 2}
@@ -572,7 +603,7 @@ def test_delta_channel_dict_reducer_backwards_compat() -> None:
 
 
 # ---------------------------------------------------------------------------
-# seed / pre-delta migration
+# DeltaChannel — seed / pre-delta migration
 # ---------------------------------------------------------------------------
 
 
@@ -583,7 +614,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(add_messages)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     seed = [HumanMessage(content="pre-delta", id="p1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes(
@@ -599,7 +630,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(add_messages)
+    spec = DeltaChannel(_messages_delta_reducer, list)
     seed = [HumanMessage(content="only-snap", id="s1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes([])
@@ -613,11 +644,10 @@ 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(left, right):
-        return right
+    def replace(state, writes):
+        return writes[-1] if writes else state
 
-    spec = DeltaChannel(replace)
+    spec = DeltaChannel(replace, list)
     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,22 +1,23 @@
-"""Benchmark: DeltaChannel vs BinaryOperatorAggregate storage and time.
+"""Benchmark: DeltaChannel snapshot_frequency — storage vs. read-depth tradeoff.
 
 Run directly:  python tests/test_delta_channel_benchmark.py
 Run via pytest: pytest tests/test_delta_channel_benchmark.py -s
 
-Simulates realistic multi-turn conversations with paragraph-length messages
-(~100 tokens each) scaling up to 1M-token-equivalent histories.
+Part 1 — baseline (original): DeltaChannel(inf) vs add_messages (BinOp).
+Part 2 — snapshot_frequency sweep: shows the storage/read-latency tradeoff
+  across frequencies [1, 5, 10, 50, inf] at scale.
 
-Token estimates: 1 token ≈ 4 chars; each turn ≈ 200 tokens (human + AI).
-A 1M-token conversation ≈ 5,000 turns of realistic messages.
-
-DeltaChannel stores only a zero-byte sentinel in checkpoint_blobs; the actual
-write data lives in checkpoint_writes (already stored there). Reconstruction
-walks the parent chain and replays writes through the operator — O(N) total
-storage vs O(N²) for plain add_messages.
+Key insight:
+  snapshot_frequency=inf  → O(N) storage, O(N) read depth (pure delta)
+  snapshot_frequency=N    → O(N²/N) storage, O(N) read depth bounded by freq
+  snapshot_frequency=1    → O(N²) storage, O(1) read depth (full snapshot)
 """
 
 from __future__ import annotations
 
+import contextlib
+import math
+import os
 import sys
 import time
 from typing import Annotated, Any
@@ -26,23 +27,17 @@ from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.checkpoint.memory import MemorySaver
 from typing_extensions import TypedDict
 
-from langgraph.channels._delta import DeltaChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import END, StateGraph
-from langgraph.graph.message import add_messages
-
-try:
-    from langgraph.checkpoint.sqlite import SqliteSaver
-
-    _SQLITE_AVAILABLE = True
-except ImportError:
-    _SQLITE_AVAILABLE = False
+from langgraph.graph.message import _messages_delta_reducer, add_messages
 
 try:
     from langgraph.checkpoint.postgres import PostgresSaver
 
     _POSTGRES_AVAILABLE = True
-    _POSTGRES_URI = (
-        "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
+    _POSTGRES_URI = os.environ.get(
+        "LANGGRAPH_BENCH_POSTGRES_URI",
+        "postgres://postgres@localhost:5432/postgres?sslmode=disable",
     )
 except ImportError:
     _POSTGRES_AVAILABLE = False
@@ -123,7 +118,21 @@ class BinaryState(TypedDict):
 
 
 class DeltaState(TypedDict):
-    messages: Annotated[list, DeltaChannel(add_messages)]
+    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]},
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -169,9 +178,8 @@ def _run_turns(
     """Run n_turns conversation turns.
 
     Returns (write_elapsed_s, read_elapsed_s, total_blob_bytes).
-    blob_bytes is -1 for savers without in-memory blob stores (e.g. SQLite).
-    Read latency is measured as the time to invoke the graph with no new
-    messages after the full history is built — this forces state rehydration.
+    Read latency is the average of 5 get_state calls after the full history
+    is built — forces state rehydration including ancestor replay if needed.
     """
     graph = _make_graph(state_cls, checkpointer)
     config = {"configurable": {"thread_id": "bench"}}
@@ -184,16 +192,16 @@ def _run_turns(
         )
     write_elapsed = time.perf_counter() - t0
 
-    # Measure read/rehydration: get_state forces the channel to rebuild
     t1 = time.perf_counter()
     for _ in range(5):
         graph.get_state(config)
     read_elapsed = (time.perf_counter() - t1) / 5
 
-    if isinstance(graph.checkpointer, MemorySaver):
-        blob_bytes = _total_blob_bytes(graph.checkpointer)
-    else:
-        blob_bytes = -1
+    blob_bytes = (
+        _total_blob_bytes(graph.checkpointer)
+        if isinstance(graph.checkpointer, MemorySaver)
+        else -1
+    )
     return write_elapsed, read_elapsed, blob_bytes
 
 
@@ -206,7 +214,6 @@ def _fmt_bytes(n: int) -> str:
 
 
 def _approx_tokens(n_turns: int) -> str:
-    # ~100 tokens human + ~100 tokens AI per turn
     tokens = n_turns * 200
     if tokens >= 1_000_000:
         return f"~{tokens / 1_000_000:.1f}M tok"
@@ -216,74 +223,56 @@ def _approx_tokens(n_turns: int) -> str:
 
 
 # ---------------------------------------------------------------------------
-# Benchmark matrix
+# Checkpointer factories
 # ---------------------------------------------------------------------------
 
-# Turn counts chosen to demonstrate O(N²) vs O(N) storage growth without running too long.
-# Extrapolation: 5,000 turns × ~200 tokens/turn ≈ 1M tokens (Claude's full context window).
-TURN_COUNTS = [10, 25, 50, 100, 500]
 
-# Deep-thread counts where add_messages blob storage would exceed 1 GB;
-# only DeltaChannel runs here.
-DELTA_ONLY_TURN_COUNTS = [1000]
+@contextlib.contextmanager
+def _pg_saver(thread_id: str = "bench"):
+    """Context manager that yields a fresh PostgresSaver and cleans up after."""
+    with PostgresSaver.from_conn_string(_POSTGRES_URI) as saver:
+        saver.setup()
+        with saver._cursor() as cur:
+            for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
+                cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
+        yield saver
+        with saver._cursor() as cur:
+            for tbl in ("checkpoints", "checkpoint_blobs", "checkpoint_writes"):
+                cur.execute(f"DELETE FROM {tbl} WHERE thread_id = %s", (thread_id,))
 
 
-def _checkpointer_factories() -> list[tuple[str, Any]]:
-    """Return (label, context_manager_or_none) pairs for available checkpointers."""
-    return [("InMemory", None)]
-
-
-def run_benchmark() -> None:
-    print()
-    print(
-        "DeltaChannel vs add_messages (BinaryOperatorAggregate) — checkpoint storage & latency"
-    )
-    print("Simulating realistic multi-turn conversations up to ~1M-token histories")
-    print("(5,000 turns × ~200 tokens/turn ≈ 1M tokens — Claude's full context window)")
-    print()
-
-    checkpointers: list[tuple[str, Any]] = [("InMemory", None)]
+def _checkpointers() -> list[tuple[str, Any]]:
+    """Return (label, saver_or_None) pairs for available checkpointers."""
+    result: list[tuple[str, Any]] = [("InMemory", None)]
     if _POSTGRES_AVAILABLE:
         try:
             import psycopg
 
             psycopg.connect(_POSTGRES_URI).close()
-            checkpointers.append(("Postgres (plain SELECT)", "postgres"))
+            result.append(("Postgres", "postgres"))
         except Exception:
             pass
-
-    for cp_label, cp_hint in checkpointers:
-        print(f"--- Checkpointer: {cp_label} ---")
-        _run_benchmark_for_checkpointer(cp_hint)
+    return result
 
 
-def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
-    import contextlib
-    import tempfile
+# ---------------------------------------------------------------------------
+# Part 1: baseline DeltaChannel(inf) vs add_messages
+# ---------------------------------------------------------------------------
+
+BASELINE_TURN_COUNTS = [10, 25, 50, 100, 500]
+DELTA_ONLY_TURN_COUNTS = [1000]
+
+
+def _run_baseline_for_checkpointer(cp_label: str, cp_hint: Any) -> None:
+    W = 72
 
-    @contextlib.contextmanager
     def _make_saver():
         if cp_hint is None:
-            yield None
-        elif cp_hint == "postgres":
-            with PostgresSaver.from_conn_string(_POSTGRES_URI) as saver:
-                saver.setup()
-                with saver._cursor() as cur:
-                    cur.execute("DELETE FROM checkpoints WHERE thread_id = 'bench'")
-                    cur.execute(
-                        "DELETE FROM checkpoint_blobs WHERE thread_id = 'bench'"
-                    )
-                    cur.execute(
-                        "DELETE FROM checkpoint_writes WHERE thread_id = 'bench'"
-                    )
-                yield saver
-        else:
-            with tempfile.NamedTemporaryFile(suffix=".db") as f:
-                with SqliteSaver.from_conn_string(f.name) as saver:
-                    yield saver
+            return contextlib.nullcontext(None)
+        return _pg_saver()
 
     rows: list[tuple[int, Any, Any, Any, Any, Any, Any]] = []
-    for turns in TURN_COUNTS:
+    for turns in BASELINE_TURN_COUNTS:
         with _make_saver() as saver:
             b_wt, b_rt, b_bytes = _run_turns(turns, BinaryState, saver)
         with _make_saver() as saver:
@@ -294,26 +283,19 @@ def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
             d_wt, d_rt, d_bytes = _run_turns(turns, DeltaState, saver)
         rows.append((turns, None, d_bytes, None, d_rt, None, d_wt))
 
-    # ── Table 1: Storage ─────────────────────────────────────────────────────
-    W = 64
-    print("Storage (checkpoint blob bytes)")
-    print("=" * W)
-    print(
-        f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}  "
-        f"{'savings':>8}"
-    )
-    print("-" * W)
-
     def _bytes_or_na(v: Any) -> str:
-        if v is None:
-            return "n/a"
-        if v < 0:
+        if v is None or v < 0:
             return "n/a"
         return _fmt_bytes(v)
 
     def _ms_or_na(v: Any) -> str:
         return "n/a" if v is None else f"{v * 1000:.1f}ms"
 
+    print(f"\n  [{cp_label}] Storage (blob bytes)")
+    print(
+        f"  {'turns':>6}  {'ctx':>10}  {'add_msgs':>12}  {'delta(inf)':>12}  {'savings':>8}"
+    )
+    print("  " + "-" * (W - 2))
     for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
         if b_bytes is None or b_bytes < 0 or d_bytes is None or d_bytes < 0:
             ratio_str = "n/a"
@@ -321,65 +303,132 @@ def _run_benchmark_for_checkpointer(cp_hint: Any) -> None:
             ratio = b_bytes / d_bytes if d_bytes else float("inf")
             ratio_str = f"{ratio:.0f}x"
         print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
-            f"{_bytes_or_na(b_bytes):>12}  {_bytes_or_na(d_bytes):>12}  "
-            f"{ratio_str:>8}"
+            f"  {turns:>6}  {_approx_tokens(turns):>10}  "
+            f"{_bytes_or_na(b_bytes):>12}  {_bytes_or_na(d_bytes):>12}  {ratio_str:>8}"
         )
-    print("=" * W)
-    print()
 
-    # ── Table 2: Read latency ─────────────────────────────────────────────────
-    print("Read latency (avg of 5 get_state calls)")
-    print("=" * W)
-    print(f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}")
-    print("-" * W)
+    print(f"\n  [{cp_label}] Read latency (avg of 5 get_state calls)")
+    print(f"  {'turns':>6}  {'ctx':>10}  {'add_msgs':>12}  {'delta(inf)':>12}")
+    print("  " + "-" * (W - 2))
     for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
         print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
+            f"  {turns:>6}  {_approx_tokens(turns):>10}  "
             f"{_ms_or_na(b_rt):>12}  {_ms_or_na(d_rt):>12}"
         )
-    print("=" * W)
+
+
+def run_baseline_benchmark() -> None:
     print()
-
-    # ── Table 3: Per-invoke latency (total write_elapsed / turns) ─────────────
-    print("Per-invoke latency (total graph.invoke time / turns)")
-    print("=" * W)
-    print(f"{'turns':>6}  {'ctx size':>10}  {'add_msgs':>12}  {'delta':>12}")
-    print("-" * W)
-    for turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt in rows:
-
-        def _per(wt: Any) -> str:
-            if wt is None:
-                return "n/a"
-            return f"{(wt / turns) * 1000:.1f}ms"
-
-        print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
-            f"{_per(b_wt):>12}  {_per(d_wt):>12}"
-        )
-    print("=" * W)
-    print()
-
-    print("Legend:")
-    print("  add_msgs  = Annotated[list, add_messages]  — O(N²) storage")
-    print("  delta     = Annotated[list, DeltaChannel(add_messages)]  — O(N) storage")
+    print("Part 1 — DeltaChannel(inf) vs add_messages: storage & latency")
+    print("=" * 72)
+    for cp_label, cp_hint in _checkpointers():
+        _run_baseline_for_checkpointer(cp_label, cp_hint)
     print()
 
 
 # ---------------------------------------------------------------------------
-# Pytest entry point
+# Part 2: snapshot_frequency sweep
+# ---------------------------------------------------------------------------
+
+# Frequencies to test. 1 = always snapshot (like BinOp), inf = pure delta.
+SNAPSHOT_FREQUENCIES: list[int | float] = [1, 5, 10, 50, math.inf]
+
+# Turn counts for the sweep — high enough to show storage divergence.
+SWEEP_TURN_COUNTS = [50, 100, 500]
+
+
+def _freq_label(freq: int | float) -> str:
+    if freq == math.inf:
+        return "inf"
+    return str(int(freq))
+
+
+def _run_sweep_for_checkpointer(cp_label: str, cp_hint: Any) -> None:
+    def _make_saver():
+        if cp_hint is None:
+            return contextlib.nullcontext(None)
+        return _pg_saver()
+
+    # Collect results: {turns: {freq_label: (write_s, read_s, bytes)}}
+    results: dict[int, dict[str, tuple[float, float, int]]] = {}
+    for turns in SWEEP_TURN_COUNTS:
+        results[turns] = {}
+        for freq in SNAPSHOT_FREQUENCIES:
+            state_cls = _make_delta_state(freq)
+            with _make_saver() as saver:
+                wt, rt, bb = _run_turns(turns, state_cls, saver)
+            results[turns][_freq_label(freq)] = (wt, rt, bb)
+
+    freq_labels = [_freq_label(f) for f in SNAPSHOT_FREQUENCIES]
+    col_w = 12
+
+    header = f"  {'turns':>6}  {'ctx':>10}" + "".join(
+        f"  {f'freq={freq_label}':>{col_w}}" for freq_label in freq_labels
+    )
+
+    print(f"\n  [{cp_label}] Storage (blob bytes) — lower is better")
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            _, _, bb = results[turns][label]
+            row += f"  {_fmt_bytes(bb) if bb >= 0 else 'n/a':>{col_w}}"
+        print(row)
+
+    print(f"\n  [{cp_label}] Read latency (avg of 5 get_state) — lower is better")
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            _, rt, _ = results[turns][label]
+            row += f"  {f'{rt * 1000:.1f}ms':>{col_w}}"
+        print(row)
+
+    print(
+        f"\n  [{cp_label}] Per-invoke write latency (total / turns) — lower is better"
+    )
+    print(header)
+    print("  " + "-" * (len(header) - 2))
+    for turns in SWEEP_TURN_COUNTS:
+        row = f"  {turns:>6}  {_approx_tokens(turns):>10}"
+        for label in freq_labels:
+            wt, _, _ = results[turns][label]
+            row += f"  {f'{(wt / turns) * 1000:.1f}ms':>{col_w}}"
+        print(row)
+
+
+def run_snapshot_freq_benchmark() -> None:
+    print()
+    print("Part 2 — DeltaChannel snapshot_frequency sweep")
+    print("Lower freq → fewer snapshots → less storage but deeper read replay")
+    print("=" * 80)
+    for cp_label, cp_hint in _checkpointers():
+        _run_sweep_for_checkpointer(cp_label, cp_hint)
+    print()
+    print("Legend:")
+    print(
+        "  freq=1    snapshot every write (full blob always — same as add_messages / BinOp)"
+    )
+    print("  freq=N    snapshot every N writes; read walks at most N ancestor writes")
+    print("  freq=inf  pure delta; read walks entire ancestor chain")
+    print()
+
+
+# ---------------------------------------------------------------------------
+# Pytest entry points
 # ---------------------------------------------------------------------------
 
 
 @pytest.mark.skip(
     reason="slow benchmark — run manually with: python tests/test_delta_channel_benchmark.py"
 )
-def test_delta_channel_benchmark(capsys: Any) -> None:
-    """Storage grows O(N²) for add_messages, O(N) for DeltaChannel."""
+def test_delta_channel_baseline_benchmark(capsys: Any) -> None:
+    """DeltaChannel(inf) uses less storage than add_messages at scale."""
     with capsys.disabled():
-        run_benchmark()
+        run_baseline_benchmark()
 
-    # Correctness assertion: DeltaChannel must use less storage at scale.
     for turns in [25, 50]:
         _, _, b_bytes = _run_turns(turns, BinaryState)
         _, _, d_bytes = _run_turns(turns, DeltaState)
@@ -389,10 +438,41 @@ def test_delta_channel_benchmark(capsys: Any) -> None:
         )
 
 
+@pytest.mark.skip(
+    reason="slow benchmark — run manually with: python tests/test_delta_channel_benchmark.py"
+)
+def test_snapshot_freq_benchmark(capsys: Any) -> None:
+    """snapshot_frequency trades storage for bounded read depth."""
+    with capsys.disabled():
+        run_snapshot_freq_benchmark()
+
+    # Correctness: results at all frequencies should agree on final state.
+    n_turns = 20
+    states: dict[str, list] = {}
+    for freq in SNAPSHOT_FREQUENCIES:
+        state_cls = _make_delta_state(freq)
+        graph = _make_graph(state_cls)
+        config = {"configurable": {"thread_id": "correctness"}}
+        for i in range(n_turns):
+            graph.invoke(
+                {"messages": [HumanMessage(content=_human_content(i), id=f"h{i}")]},
+                config,
+            )
+        state = graph.get_state(config)
+        states[_freq_label(freq)] = [m.id for m in state.values["messages"]]
+
+    ref = states["inf"]
+    for label, msg_ids in states.items():
+        assert msg_ids == ref, (
+            f"freq={label} produced different message IDs than freq=inf"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Script entry point
 # ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    run_benchmark()
+    run_baseline_benchmark()
+    run_snapshot_freq_benchmark()
     sys.exit(0)

libs/langgraph/tests/test_delta_channel_migration.py

@@ -46,12 +46,14 @@ 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._delta import DeltaChannel
 from langgraph.channels.binop import BinaryOperatorAggregate
+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
 
@@ -70,6 +72,13 @@ 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)]
@@ -85,7 +94,7 @@ def _binop_graph(checkpointer: Any) -> Any:
 
 def _delta_graph(checkpointer: Any) -> Any:
     class DeltaState(TypedDict):
-        items: Annotated[list, DeltaChannel(operator.add)]
+        items: Annotated[list, DeltaChannel(_list_concat)]
 
     return (
         StateGraph(DeltaState)
@@ -502,3 +511,103 @@ 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_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
+from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage
 from langchain_core.runnables import (
     RunnableConfig,
     RunnableLambda,
@@ -25,6 +25,7 @@ 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,
@@ -41,6 +42,7 @@ from typing_extensions import NotRequired, TypedDict
 
 from langgraph._internal._constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL
 from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
@@ -49,7 +51,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, add_messages
+from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
 from langgraph.pregel import (
     NodeBuilder,
     Pregel,
@@ -9404,15 +9406,9 @@ 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(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
 
     def respond(state: State) -> dict:
         n = len(state["messages"])
@@ -9446,15 +9442,9 @@ 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(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
 
     counter = {"n": 0}
 
@@ -9504,15 +9494,9 @@ 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(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
 
     def respond(state: State) -> dict:
         return {"messages": [AIMessage(content="reply", id="ai-1")]}
@@ -9551,15 +9535,9 @@ 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(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
 
     def update_msg(state: State) -> dict:
         # re-send h1 with updated content
@@ -9587,3 +9565,91 @@ 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,6 +6101,36 @@ 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,6 +7,9 @@ resolution-markers = [
     "python_full_version < '3.11'",
 ]
 
+[options]
+prerelease-mode = "allow"
+
 [[package]]
 name = "aiosqlite"
 version = "0.22.1"
@@ -1348,10 +1351,11 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
+    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -1360,14 +1364,26 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.1.9"
+version = "1.2.0a2"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1439,7 +1455,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.0,<2" },
+    { name = "langchain-core", specifier = ">=1.3.2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1548,7 +1564,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.0.2"
+version = "4.1.0a2"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

notes/delta_channel_query_bench.md

@@ -1,137 +0,0 @@
-# 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.

tmp_verify_send_blob.py

@@ -1,217 +0,0 @@
-"""Verify that SEND values land in checkpoint_blobs (TASKS channel),
-not just in checkpoint_writes.
-
-Strategy:
-  1. Build a tiny graph: node "1" emits a Send to node "2" via
-     conditional edges. interrupt_before=["2"] freezes the loop after
-     super-step 1 completes (cp_1 persisted) but before "2" runs.
-  2. Run graph.invoke against PostgresSaver.
-  3. Read checkpoint_blobs raw via psycopg.
-  4. Find the row with channel = '__pregel_tasks' and decode the blob
-     using the saver's serde. Assert it contains a non-empty list of
-     Send objects.
-  5. As a control, also dump checkpoint_writes to confirm the *same*
-     value is independently present in the writes table.
-"""
-
-from __future__ import annotations
-
-import operator
-from typing import Annotated
-from typing_extensions import TypedDict
-from uuid import uuid4
-
-from psycopg import Connection
-from psycopg.rows import dict_row
-
-from langgraph.checkpoint.postgres import PostgresSaver
-from langgraph.checkpoint.serde.types import TASKS
-from langgraph.graph import START, StateGraph
-from langgraph.types import Send
-
-DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5441/"
-
-
-class State(TypedDict):
-    history: Annotated[list[str], operator.add]
-
-
-def make_graph(checkpointer: PostgresSaver, *, with_interrupt: bool = True):
-    def node_one(state: State) -> State:
-        return {"history": ["1"]}
-
-    def node_two(state: dict) -> State:
-        return {"history": [f"2:{state['payload']}"]}
-
-    def fanout(state: State):
-        return [Send("two", {"payload": x}) for x in ("a", "b", "c")]
-
-    builder = StateGraph(State)
-    builder.add_node("one", node_one)
-    builder.add_node("two", node_two)
-    builder.add_edge(START, "one")
-    builder.add_conditional_edges("one", fanout, ["two"])
-    return builder.compile(
-        checkpointer=checkpointer,
-        interrupt_before=["two"] if with_interrupt else [],
-    )
-
-
-def main() -> None:
-    database = f"verify_{uuid4().hex[:12]}"
-    with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
-        conn.execute(f"CREATE DATABASE {database}")
-    try:
-        uri = DEFAULT_POSTGRES_URI + database
-        with Connection.connect(
-            uri, autocommit=True, prepare_threshold=0, row_factory=dict_row
-        ) as conn:
-            saver = PostgresSaver(conn)
-            saver.setup()
-
-            graph = make_graph(saver)
-            thread_id = "t1"
-            cfg = {"configurable": {"thread_id": thread_id}}
-            result = graph.invoke({"history": []}, cfg)
-            print("=== invoke result (interrupted before 'two') ===")
-            print(result)
-            print()
-
-            # Inspect raw checkpoint_blobs.
-            rows = conn.execute(
-                """
-                SELECT thread_id, checkpoint_ns, channel, version, type,
-                       octet_length(blob) AS nbytes, blob
-                FROM checkpoint_blobs
-                WHERE thread_id = %s
-                ORDER BY channel, version
-                """,
-                (thread_id,),
-            ).fetchall()
-
-            print(f"=== checkpoint_blobs ({len(rows)} rows) ===")
-            for r in rows:
-                preview = (
-                    "(empty)"
-                    if r["blob"] is None
-                    else f"{r['blob'][:60]!r}{'...' if len(r['blob']) > 60 else ''}"
-                )
-                print(
-                    f"  channel={r['channel']!r:30s} v={r['version']!s:18s} "
-                    f"type={r['type']:8s} nbytes={r['nbytes']!s:>5} blob={preview}"
-                )
-            print()
-
-            tasks_rows = [r for r in rows if r["channel"] == TASKS]
-            assert tasks_rows, (
-                f"expected at least one row in checkpoint_blobs for {TASKS!r}, "
-                f"got channels={[r['channel'] for r in rows]}"
-            )
-
-            # Decode the latest TASKS blob using the saver's serde.
-            tasks_row = tasks_rows[-1]
-            decoded = saver.serde.loads_typed((tasks_row["type"], tasks_row["blob"]))
-            print(f"=== decoded {TASKS!r} blob ({tasks_row['type']}) ===")
-            print(f"  type(decoded) = {type(decoded).__name__}")
-            print(f"  len(decoded)  = {len(decoded)}")
-            for i, v in enumerate(decoded):
-                print(f"  [{i}] type={type(v).__name__} value={v!r}")
-            print()
-
-            assert isinstance(decoded, list), (
-                f"expected list of Sends, got {type(decoded).__name__}"
-            )
-            assert len(decoded) == 3, f"expected 3 Sends (a, b, c), got {len(decoded)}"
-            for v in decoded:
-                assert isinstance(v, Send), (
-                    f"expected Send instance in blob, got {type(v).__name__}"
-                )
-                assert v.node == "two"
-            print("✅ ASSERTION PASSED: checkpoint_blobs contains the Sends.")
-            print()
-
-            # Control: confirm the same Sends are also in checkpoint_writes.
-            wrows = conn.execute(
-                """
-                SELECT task_id, idx, channel, type, octet_length(blob) AS nbytes, blob
-                FROM checkpoint_writes
-                WHERE thread_id = %s AND channel = %s
-                ORDER BY task_id, idx
-                """,
-                (thread_id, TASKS),
-            ).fetchall()
-            print(
-                f"=== checkpoint_writes (channel={TASKS!r}, {len(wrows)} rows) ==="
-            )
-            for r in wrows:
-                v = saver.serde.loads_typed((r["type"], r["blob"]))
-                print(
-                    f"  task={r['task_id']} idx={r['idx']} type={r['type']} "
-                    f"nbytes={r['nbytes']} -> {v!r}"
-                )
-
-            assert len(wrows) == 3, (
-                f"expected 3 writes-table rows for {TASKS!r}, got {len(wrows)}"
-            )
-            print(
-                "✅ ASSERTION PASSED: checkpoint_writes also contains the Sends."
-            )
-            print()
-            print(
-                "Conclusion (cp_1): SEND is double-stored. blobs row carries "
-                "the authoritative Send list; writes rows carry the same "
-                "payload as a redundant per-task record."
-            )
-
-            # ----------------------------------------------------------------
-            # Now resume: let 'two' actually run, so the Sends get consumed.
-            # We expect the next checkpoint's TASKS blob to be cleared,
-            # proving the accumulate=False lifecycle.
-            # ----------------------------------------------------------------
-            graph2 = make_graph(saver, with_interrupt=False)
-            result2 = graph2.invoke(None, cfg)
-            print()
-            print("=== invoke result (resumed; 'two' fanout completed) ===")
-            print(result2)
-            print()
-
-            rows2 = conn.execute(
-                """
-                SELECT channel, version, type, octet_length(blob) AS nbytes, blob
-                FROM checkpoint_blobs
-                WHERE thread_id = %s AND channel = %s
-                ORDER BY version
-                """,
-                (thread_id, TASKS),
-            ).fetchall()
-            print(f"=== checkpoint_blobs rows for {TASKS!r} across all cps ===")
-            for r in rows2:
-                decoded_v = saver.serde.loads_typed((r["type"], r["blob"]))
-                print(
-                    f"  v={r['version']} type={r['type']} nbytes={r['nbytes']} "
-                    f"-> {decoded_v!r}"
-                )
-
-            assert len(rows2) >= 2, (
-                "expected at least 2 versions of TASKS blob across the run"
-            )
-            latest = saver.serde.loads_typed(
-                (rows2[-1]["type"], rows2[-1]["blob"])
-            )
-            assert latest == [], (
-                f"expected latest TASKS blob to be cleared (==[]) "
-                f"after Sends were consumed, got {latest!r}"
-            )
-            print()
-            print(
-                "✅ ASSERTION PASSED: after Sends were consumed, the latest "
-                "TASKS blob is [] (Topic with accumulate=False clears itself "
-                "via update(EMPTY_SEQ) in apply_writes line 304-311)."
-            )
-    finally:
-        with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
-            conn.execute(f"DROP DATABASE {database}")
-
-
-if __name__ == "__main__":
-    main()