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

@@ -8,11 +8,13 @@ from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     ChannelVersions,
     Checkpoint,
     CheckpointMetadata,
     CheckpointTuple,
+    _ChannelWritesHistory,
     get_checkpoint_id,
     get_serializable_checkpoint_metadata,
 )
@@ -23,7 +25,11 @@ from psycopg.types.json import Jsonb
 from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
-from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.base import (
+    SELECT_DELTA_COMBINED_SQL,
+    BasePostgresSaver,
+    _DeltaCombinedRow,
+)
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
 Conn = _internal.Conn  # For backward compatibility
@@ -32,7 +38,7 @@ Conn = _internal.Conn  # For backward compatibility
 class PostgresSaver(BasePostgresSaver):
     """Checkpointer that stores checkpoints in a Postgres database."""
 
-    lock: threading.RLock
+    lock: threading.Lock
 
     def __init__(
         self,
@@ -48,7 +54,7 @@ class PostgresSaver(BasePostgresSaver):
 
         self.conn = conn
         self.pipe = pipe
-        self.lock = threading.RLock()
+        self.lock = threading.Lock()
         self.supports_pipeline = Capabilities().has_pipeline()
 
     @classmethod
@@ -430,6 +436,48 @@ class PostgresSaver(BasePostgresSaver):
                 with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
+
+        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
+        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
+        single roundtrip; the ancestor walk runs in Python.
+        """
+        thread_id = config["configurable"]["thread_id"]
+        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        checkpoint_id = get_checkpoint_id(config)
+        if checkpoint_id is None:
+            # Caller didn't specify a target — resolve to the latest
+            # checkpoint on the thread. `get_tuple` without `checkpoint_id`
+            # returns the newest; its config carries the resolved id.
+            target = self.get_tuple(config)
+            if target is None:
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
+        with self._cursor() as cur:
+            cur.execute(
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
+            )
+            rows = cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
+        )
+
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
         """
         Convert a database row into a CheckpointTuple object.
@@ -442,22 +490,6 @@ class PostgresSaver(BasePostgresSaver):
             including its configuration, metadata, parent checkpoint (if any),
             and pending writes.
         """
-        from langgraph.checkpoint.base import DeltaChannelSentinel
-
-        channel_values = self._load_blobs(value["channel_values"])
-        if any(isinstance(v, DeltaChannelSentinel) for v in channel_values.values()):
-            cp_config = cast(
-                RunnableConfig,
-                {
-                    "configurable": {
-                        "thread_id": value["thread_id"],
-                        "checkpoint_ns": value["checkpoint_ns"],
-                        "checkpoint_id": value["checkpoint_id"],
-                    }
-                },
-            )
-            with self._cursor() as cur:
-                self._resolve_delta_channels(cp_config, channel_values, cur)
         return CheckpointTuple(
             {
                 "configurable": {
@@ -470,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,16 +4,17 @@ 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 (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     ChannelVersions,
     Checkpoint,
     CheckpointMetadata,
     CheckpointTuple,
-    DeltaChannelSentinel,
+    _ChannelWritesHistory,
     get_checkpoint_id,
     get_serializable_checkpoint_metadata,
 )
@@ -24,7 +25,11 @@ from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
-from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.base import (
+    SELECT_DELTA_COMBINED_SQL,
+    BasePostgresSaver,
+    _DeltaCombinedRow,
+)
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
 Conn = _ainternal.Conn  # For backward compatibility
@@ -392,57 +397,45 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    async def _aget_channel_writes_cur(
-        self,
-        thread_id: str,
-        checkpoint_ns: str,
-        checkpoint_id: str,
-        channel: str,
-        cur: Any,
-    ) -> list[Any]:
-        """Async version of _get_channel_writes_cur — see sync version for rationale."""
-        await cur.execute(
-            "SELECT checkpoint_id, parent_checkpoint_id FROM checkpoints "
-            "WHERE thread_id = %s AND checkpoint_ns = %s",
-            (thread_id, checkpoint_ns),
-        )
-        parent_map: dict[str, str | None] = {
-            row["checkpoint_id"]: row["parent_checkpoint_id"]
-            for row in await cur.fetchall()
-        }
-        ancestor_ids: list[str] = []
-        cid: str | None = parent_map.get(checkpoint_id)
-        while cid is not None:
-            ancestor_ids.append(cid)
-            cid = parent_map.get(cid)
-        if not ancestor_ids:
-            return []
-        await cur.execute(
-            "SELECT checkpoint_id, type, blob FROM checkpoint_writes "
-            "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
-            "  AND checkpoint_id = ANY(%s) "
-            "ORDER BY task_id, idx",
-            (thread_id, checkpoint_ns, channel, ancestor_ids),
-        )
-        writes_by_cp: dict[str, list[tuple[str, bytes]]] = defaultdict(list)
-        for row in await cur.fetchall():
-            writes_by_cp[row["checkpoint_id"]].append((row["type"], row["blob"]))
-        result = []
-        for cid in reversed(ancestor_ids):
-            for type_tag, blob in writes_by_cp.get(cid, []):
-                result.append(self.serde.loads_typed((type_tag, blob)))
-        return result
-
-    async def aget_channel_writes(
+    async def _aget_channel_writes_history(
         self, config: RunnableConfig, channel: str
-    ) -> list[Any]:
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
+
+        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
+        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
+        single roundtrip; rows are assembled by the shared pure helper on
+        `BasePostgresSaver`.
+        """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-        checkpoint_id = config["configurable"]["checkpoint_id"]
+        checkpoint_id = get_checkpoint_id(config)
+        if checkpoint_id is None:
+            target = await self.aget_tuple(config)
+            if target is None:
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
         async with self._cursor() as cur:
-            return await self._aget_channel_writes_cur(
-                thread_id, checkpoint_ns, checkpoint_id, channel, cur
+            await cur.execute(
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
             )
+            rows = await cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
+        )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
         """
@@ -456,31 +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"]
-        checkpoint_id = value["checkpoint_id"]
-        blob_values = value["channel_values"]
-
-        channel_values: dict[str, Any] = {}
-        if blob_values:
-            channel_values = self._load_blobs(blob_values)
-            delta_channels = [
-                ch
-                for ch, v in channel_values.items()
-                if isinstance(v, DeltaChannelSentinel)
-            ]
-            if delta_channels:
-                async with self._cursor() as cur:
-                    for channel in delta_channels:
-                        channel_values[channel] = await self._aget_channel_writes_cur(
-                            thread_id, checkpoint_ns, checkpoint_id, channel, cur
-                        )
-
         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"],
                 }
             },
@@ -488,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

@@ -2,17 +2,18 @@ from __future__ import annotations
 
 import random
 import warnings
-from collections import defaultdict
 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 (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
-    DeltaChannelSentinel,
+    PendingWrite,
+    _ChannelWritesHistory,
     get_checkpoint_id,
 )
 from langgraph.checkpoint.serde.types import TASKS
@@ -155,6 +156,62 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """
 
 
+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,
+           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
+    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
+    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
+"""
+
+
 class BasePostgresSaver(BaseCheckpointSaver[str]):
     SELECT_SQL = SELECT_SQL
     SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
@@ -187,78 +244,92 @@ 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 _resolve_delta_channels(
+    def _build_delta_channel_writes_history(
         self,
-        config: RunnableConfig,
-        channel_values: dict[str, Any],
-        cur: Any,
-    ) -> None:
-        for channel, value in list(channel_values.items()):
-            if isinstance(value, DeltaChannelSentinel):
-                channel_values[channel] = self._get_channel_writes_cur(
-                    config["configurable"]["thread_id"],
-                    config["configurable"].get("checkpoint_ns", ""),
-                    config["configurable"]["checkpoint_id"],
-                    channel,
-                    cur,
+        *,
+        channel: str,
+        target_id: str,
+        rows: Sequence[_DeltaCombinedRow],
+    ) -> _ChannelWritesHistory:
+        """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 `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
+        walk and is returned as the seed so replay starts from it.
+
+        Writes stored at `target_id` itself are pending writes for the next
+        step and are excluded — the walk begins at the target's parent.
+        """
+        parent_of: dict[str, str | None] = {}
+        ver_of: dict[str, str | None] = {}
+        writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
+        blob_by_ver: dict[str, tuple[str, bytes]] = {}
+
+        for r in rows:
+            kind = r["_kind"]
+            if kind == "p":
+                cid = cast(str, r["checkpoint_id"])
+                parent_of[cid] = r["parent_checkpoint_id"]
+                ver_of[cid] = r["ver"]
+            elif kind == "w":
+                cid = cast(str, r["checkpoint_id"])
+                writes_by_cid.setdefault(cid, []).append(
+                    cast(
+                        "tuple[str, bytes, str, int]",
+                        (r["type"], r["blob"], r["task_id"], r["idx"]),
+                    )
+                )
+            else:  # kind == "b"
+                blob_by_ver[cast(str, r["version"])] = cast(
+                    "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-    def _get_channel_writes_cur(
-        self,
-        thread_id: str,
-        checkpoint_ns: str,
-        checkpoint_id: str,
-        channel: str,
-        cur: Any,
-    ) -> list[Any]:
-        """Fetch writes for `channel` across the checkpoint ancestor chain, oldest→newest.
+        # 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)
 
-        Two queries:
-        1. Fetch all (checkpoint_id, parent_checkpoint_id) for the thread — cheap, IDs only.
-        2. Walk the ancestor chain in Python, then fetch writes with a plain ANY() filter.
-        """
-        cur.execute(
-            "SELECT checkpoint_id, parent_checkpoint_id FROM checkpoints "
-            "WHERE thread_id = %s AND checkpoint_ns = %s",
-            (thread_id, checkpoint_ns),
-        )
-        parent_map: dict[str, str | None] = {
-            row["checkpoint_id"]: row["parent_checkpoint_id"] for row in cur.fetchall()
-        }
-        ancestor_ids: list[str] = []
-        cid: str | None = parent_map.get(checkpoint_id)
-        while cid is not None:
-            ancestor_ids.append(cid)
-            cid = parent_map.get(cid)
-        if not ancestor_ids:
-            return []
-        cur.execute(
-            "SELECT checkpoint_id, type, blob FROM checkpoint_writes "
-            "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
-            "  AND checkpoint_id = ANY(%s) "
-            "ORDER BY task_id, idx",
-            (thread_id, checkpoint_ns, channel, ancestor_ids),
-        )
-        writes_by_cp: dict[str, list[tuple[str, bytes]]] = defaultdict(list)
-        for row in cur.fetchall():
-            writes_by_cp[row["checkpoint_id"]].append((row["type"], row["blob"]))
-        result = []
-        for cid in reversed(ancestor_ids):
-            for type_tag, blob in writes_by_cp.get(cid, []):
-                result.append(self.serde.loads_typed((type_tag, blob)))
-        return result
+        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:
+            # 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)
+                if seed_blob is not None and seed_blob[0] != "empty":
+                    blob_value = self.serde.loads_typed(seed_blob)
+                    if blob_value is not DELTA_SENTINEL:
+                        collected.reverse()
+                        return _ChannelWritesHistory(seed=blob_value, writes=collected)
+
+        collected.reverse()  # oldest → newest
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
     def _dump_blobs(
         self,

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

@@ -361,9 +361,9 @@ async def test_get_checkpoint_no_channel_values(
 
         load_checkpoint_tuple = saver._load_checkpoint_tuple
 
-        def patched_load_checkpoint_tuple(value):
+        async def patched_load_checkpoint_tuple(value):
             value["checkpoint"].pop("channel_values", None)
-            return load_checkpoint_tuple(value)
+            return await load_checkpoint_tuple(value)
 
         monkeypatch.setattr(
             saver, "_load_checkpoint_tuple", patched_load_checkpoint_tuple
@@ -385,11 +385,11 @@ async def test_delta_channel_chain_reconstruction(saver_name: str) -> None:
     from langchain_core.messages import AIMessage, HumanMessage
     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,14 +1,11 @@
 from __future__ import annotations
 
 import copy
-import dataclasses
 import logging
-import threading
 from collections.abc import AsyncIterator, Collection, Iterator, Mapping, Sequence
-from typing import (  # noqa: UP035
+from typing import (
     Any,
     Generic,
-    List,
     Literal,
     NamedTuple,
     TypedDict,
@@ -21,6 +18,9 @@ from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.base import SerializerProtocol, maybe_add_typed_methods
 from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
+from langgraph.checkpoint.serde.types import (
+    DELTA_SENTINEL as DELTA_SENTINEL,
+)
 from langgraph.checkpoint.serde.types import (
     ERROR,
     INTERRUPT,
@@ -33,19 +33,6 @@ V = TypeVar("V", int, float, str)
 PendingWrite = tuple[str, str, Any]
 
 
-@dataclasses.dataclass
-class DeltaChannelSentinel:
-    """Marker stored in checkpoint_blobs for a DeltaChannel field.
-
-    No data is stored here — the actual per-step writes live in checkpoint_writes
-    and are replayed through the reducer at load time.
-    """
-
-    pass
-
-
-_DELTA_RECONSTRUCTION: threading.local = threading.local()
-
 logger = logging.getLogger(__name__)
 
 
@@ -137,6 +124,30 @@ class CheckpointTuple(NamedTuple):
     pending_writes: list[PendingWrite] | None = None
 
 
+class _ChannelWritesHistory(NamedTuple):
+    """Result of `BaseCheckpointSaver._get_channel_writes_history`.
+
+    Storage-level view of what one channel wrote across the ancestor chain
+    of a target checkpoint:
+
+      * `seed` — the nearest ancestor's stored blob value for this channel,
+        or `DELTA_SENTINEL` if the walk reached the root without finding a
+        stored value. A non-sentinel seed typically indicates a pre-delta
+        snapshot preserved across a channel-type migration (e.g.
+        `BinaryOperatorAggregate` storage extended under `DeltaChannel`).
+      * `writes` — on-path deltas oldest→newest, one `PendingWrite` per
+        step that wrote to this channel. Writes stored at the target
+        checkpoint itself are pending for the next super-step and are
+        excluded.
+
+    Experimental: method surface may change; the NamedTuple shape is the
+    contract.
+    """
+
+    seed: Any
+    writes: list[PendingWrite]
+
+
 class BaseCheckpointSaver(Generic[V]):
     """Base class for creating a graph checkpointer.
 
@@ -475,55 +486,103 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    def get_channel_writes(self, config: RunnableConfig, channel: str) -> List[Any]:  # noqa: UP006
-        """Collect all writes for `channel` across this checkpoint's ancestry, oldest→newest.
+    def _get_tuple_raw(self, config: RunnableConfig) -> CheckpointTuple | None:
+        """Pure storage read used by `_get_channel_writes_history`.
 
-        Default implementation walks the full thread history via `list()`. Savers can
-        override with a more efficient query (InMemorySaver and PostgresSaver do this).
+        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.
         """
-        # Guard against re-entrant calls: when list() triggers reconstruction which
-        # calls list() again, the inner call returns tuples with DeltaChannelSentinel
-        # in channel_values (which get_channel_writes ignores — it only reads
-        # pending_writes). This breaks the recursion safely.
-        if getattr(_DELTA_RECONSTRUCTION, "active", False):
-            return []
-        _DELTA_RECONSTRUCTION.active = True
-        try:
-            result: list[Any] = []
-            target_id = config["configurable"].get("checkpoint_id")
-            for tup in self.list(config):
-                if tup.config["configurable"].get("checkpoint_id") == target_id:
-                    continue  # skip the checkpoint itself; we want its ancestors' writes
-                if tup.pending_writes:
-                    for _, ch, value in tup.pending_writes:
-                        if ch == channel:
-                            result.append(value)
-            result.reverse()  # list() yields newest→oldest; we want oldest→newest
-            return result
-        finally:
-            _DELTA_RECONSTRUCTION.active = False
+        return self.get_tuple(config)
 
-    async def aget_channel_writes(
+    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
-    ) -> List[Any]:  # noqa: UP006
-        """Async version of get_channel_writes."""
-        if getattr(_DELTA_RECONSTRUCTION, "active", False):
-            return []
-        _DELTA_RECONSTRUCTION.active = True
-        try:
-            result: list[Any] = []
-            target_id = config["configurable"].get("checkpoint_id")
-            async for tup in self.alist(config):
-                if tup.config["configurable"].get("checkpoint_id") == target_id:
-                    continue
-                if tup.pending_writes:
-                    for _, ch, value in tup.pending_writes:
-                        if ch == channel:
-                            result.append(value)
-            result.reverse()
-            return result
-        finally:
-            _DELTA_RECONSTRUCTION.active = False
+    ) -> _ChannelWritesHistory:
+        """**Experimental.** Query one channel's writes along the parent chain.
+
+        Storage-level query, not channel semantics: returns `(seed, writes)`
+        reflecting what storage knows about a single channel across the
+        ancestor chain of the target checkpoint identified by `config`.
+
+        * `writes` — on-path deltas oldest→newest as `PendingWrite` tuples.
+          Writes stored at the target `checkpoint_id` itself are pending
+          for the next super-step and are excluded.
+        * `seed` — the nearest ancestor's stored blob value for this
+          channel; `DELTA_SENTINEL` if the walk reached the root without
+          finding a stored value. A non-sentinel seed typically indicates
+          a pre-delta snapshot preserved across a channel-type migration.
+
+        Walks the **parent chain** (not `list(before=...)`): for forked
+        threads, only on-path ancestors contribute.
+
+        Reference implementation walks `get_tuple` + `parent_config`,
+        inspecting each ancestor's `channel_values[channel]` for the seed
+        terminator. Savers with direct storage access (`InMemorySaver`,
+        `PostgresSaver`) override for performance; the return contract is
+        fixed here.
+
+        Underscore-prefixed because the method surface is experimental.
+        """
+        collected: list[PendingWrite] = []  # newest first; reversed at the end
+        target_tuple = self._get_tuple_raw(config)
+        cursor_config: RunnableConfig | None = (
+            target_tuple.parent_config if target_tuple else None
+        )
+        while cursor_config is not None:
+            tup = self._get_tuple_raw(cursor_config)
+            if tup is None:
+                break
+            # Collect this ancestor's writes FIRST — they encode the
+            # transition from this ancestor's state to its child's, so
+            # they must be included whether or not this ancestor is the
+            # seed terminator.
+            if tup.pending_writes:
+                # Within a superstep, pending_writes are oldest→newest;
+                # reverse to scan newest-first.
+                for write in reversed(tup.pending_writes):
+                    if write[1] != channel:
+                        continue
+                    collected.append(write)
+            # Seed terminator: any non-sentinel blob on an ancestor
+            # establishes the reconstruction base. Stop here.
+            ancestor_value = tup.checkpoint["channel_values"].get(channel)
+            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                collected.reverse()
+                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
+            cursor_config = tup.parent_config
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+
+    async def _aget_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Async version of `_get_channel_writes_history`. See docstring there."""
+        collected: list[PendingWrite] = []
+        target_tuple = await self._aget_tuple_raw(config)
+        cursor_config: RunnableConfig | None = (
+            target_tuple.parent_config if target_tuple else None
+        )
+        while cursor_config is not None:
+            tup = await self._aget_tuple_raw(cursor_config)
+            if tup is None:
+                break
+            if tup.pending_writes:
+                for write in reversed(tup.pending_writes):
+                    if write[1] != channel:
+                        continue
+                    collected.append(write)
+            ancestor_value = tup.checkpoint["channel_values"].get(channel)
+            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                collected.reverse()
+                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
+            cursor_config = tup.parent_config
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
     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,22 +9,25 @@ 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
 
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
     Checkpoint,
     CheckpointMetadata,
     CheckpointTuple,
-    DeltaChannelSentinel,
+    PendingWrite,
     SerializerProtocol,
+    _ChannelWritesHistory,
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 logger = logging.getLogger(__name__)
 
@@ -138,24 +141,20 @@ class InMemorySaver(
             result[k] = self.serde.loads_typed(vv)
         return result
 
-    def _resolve_delta_channels(
-        self,
-        config: RunnableConfig,
-        channel_values: dict[str, Any],
-    ) -> None:
-        """Replace DeltaChannelSentinel entries with reconstructed write lists."""
-        for channel, value in list(channel_values.items()):
-            if isinstance(value, DeltaChannelSentinel):
-                channel_values[channel] = self.get_channel_writes(config, channel)
-
-    def get_channel_writes(self, config: RunnableConfig, channel: str) -> list[Any]:
+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = config["configurable"].get("checkpoint_id", "")
         ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
-        # Walk the parent chain newest→oldest collecting checkpoint IDs.
+        # Walk the parent chain newest→oldest. Skip the target itself —
+        # writes stored AT `checkpoint_id` are pending for the next step
+        # (pregel applies them via `apply_writes`; they aren't part of the
+        # snapshot value AT `checkpoint_id`).
         chain: list[str] = []
-        current: str | None = checkpoint_id
+        target_entry = ns_storage.get(checkpoint_id)
+        current: str | None = target_entry[2] if target_entry is not None else None
         while current is not None:
             entry = ns_storage.get(current)
             if entry is None:
@@ -163,19 +162,77 @@ class InMemorySaver(
             chain.append(current)
             _, _, parent = entry
             current = parent
-        # Collect writes oldest→newest.
-        result: list[Any] = []
-        for cp_id in reversed(chain):
-            step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-            for (_task_id, _idx), (_, ch, serialized, _) in sorted(step_writes.items()):
-                if ch == channel:
-                    result.append(self.serde.loads_typed(serialized))
-        return result
+        # Scan newest→oldest. A pre-delta blob on an ancestor terminates the
+        # walk and is bound as `seed`; without this, a thread migrated from
+        # pre-delta storage would replay ancestor writes all the way to the
+        # root AND miss any value that lived only in the old blob (e.g. from
+        # `update_state`).
+        #
+        # At each ancestor, check the blob BEFORE processing its pending
+        # writes: a pre-delta blob represents the state AT that ancestor,
+        # which already subsumes any writes stored under it. Processing
+        # those writes first would fold them into the reconstructed value
+        # twice (once via the blob, once via replay).
+        collected: list[PendingWrite] = []  # newest first
+        for cp_id in chain:  # newest → oldest
+            entry = ns_storage.get(cp_id)
+            if entry is not None:
+                ckpt = self.serde.loads_typed(entry[0])
+                ver = ckpt.get("channel_versions", {}).get(channel)
+                if ver is not None:
+                    blob_entry = self.blobs.get(
+                        (thread_id, checkpoint_ns, channel, ver)
+                    )
+                    if blob_entry is not None and blob_entry[0] != "empty":
+                        blob_value = self.serde.loads_typed(blob_entry)
+                        if blob_value is not DELTA_SENTINEL:
+                            if isinstance(blob_value, _DeltaSnapshot):
+                                # Step-based snapshot: the blob is state AT this
+                                # ancestor, but the ancestor's pending_writes
+                                # encode the NEXT step's transition and are NOT
+                                # subsumed by the snapshot — collect them first.
+                                step_writes = self.writes.get(
+                                    (thread_id, checkpoint_ns, cp_id), {}
+                                )
+                                for (_task_id, _idx), (
+                                    tid,
+                                    ch,
+                                    serialized,
+                                    _,
+                                ) in sorted(step_writes.items(), reverse=True):
+                                    if ch != channel:
+                                        continue
+                                    collected.append(
+                                        (tid, ch, self.serde.loads_typed(serialized))
+                                    )
+                                collected.reverse()
+                                return _ChannelWritesHistory(
+                                    seed=blob_value, writes=collected
+                                )
+                            # Pre-delta blob: state AT this ancestor already
+                            # subsumes its pending_writes — skip them.
+                            collected.reverse()
+                            return _ChannelWritesHistory(
+                                seed=blob_value, writes=collected
+                            )
 
-    async def aget_channel_writes(
+            step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
+            # Within a superstep, sorted by (task_id, idx) = oldest → newest;
+            # reverse for newest-first scan.
+            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
+                step_writes.items(), reverse=True
+            ):
+                if ch != channel:
+                    continue
+                val = self.serde.loads_typed(serialized)
+                collected.append((tid, ch, val))
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+
+    async def _aget_channel_writes_history(
         self, config: RunnableConfig, channel: str
-    ) -> list[Any]:
-        return self.get_channel_writes(config, channel)
+    ) -> _ChannelWritesHistory:
+        return self._get_channel_writes_history(config, channel)
 
     def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the in-memory storage.
@@ -198,17 +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"],
-                )
-                self._resolve_delta_channels(config, channel_values)
                 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=[
@@ -232,27 +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"],
-                )
-                self._resolve_delta_channels(resolved_config, channel_values)
-                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=[
@@ -347,28 +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"],
-                    )
-                    self._resolve_delta_channels(list_config, channel_values)
-
-                    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,14 +33,18 @@ 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 SendProtocol
+from langgraph.checkpoint.serde.types import (
+    DELTA_SENTINEL,
+    SendProtocol,
+    _DeltaSentinel,
+    _DeltaSnapshot,
+)
 from langgraph.store.base import Item
 
 if TYPE_CHECKING:
     from langgraph.checkpoint.serde._msgpack import (
         AllowedMsgpackModules,
     )
-    from langgraph.checkpoint.serde.types import SendProtocol
 
 LC_REVIVER = Reviver()
 EMPTY_BYTES = b""
@@ -64,14 +68,6 @@ def _warn_once(
     logger.warning(msg, *args)
 
 
-def _get_delta_sentinel_cls() -> type:
-    from langgraph.checkpoint.base import (
-        DeltaChannelSentinel,
-    )  # lazy import avoids circular dep
-
-    return DeltaChannelSentinel
-
-
 class JsonPlusSerializer(SerializerProtocol):
     """Serializer that uses ormsgpack, with optional fallbacks.
 
@@ -264,8 +260,6 @@ class JsonPlusSerializer(SerializerProtocol):
             return "bytes", obj
         elif isinstance(obj, bytearray):
             return "bytearray", obj
-        elif isinstance(obj, _get_delta_sentinel_cls()):
-            return "delta", b""
         else:
             try:
                 return "msgpack", _msgpack_enc(obj)
@@ -288,10 +282,6 @@ class JsonPlusSerializer(SerializerProtocol):
             return ormsgpack.unpackb(
                 data_, ext_hook=self._unpack_ext_hook, option=ormsgpack.OPT_NON_STR_KEYS
             )
-        elif type_ == "delta":
-            from langgraph.checkpoint.base import DeltaChannelSentinel
-
-            return DeltaChannelSentinel()
         elif self.pickle_fallback and type_ == "pickle":
             return pickle.loads(data_)
         else:
@@ -307,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(
@@ -624,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,
@@ -14,6 +15,39 @@ INTERRUPT = "__interrupt__"
 RESUME = "__resume__"
 TASKS = "__pregel_tasks"
 
+
+class _DeltaSentinel:
+    """Singleton marker stored (as zero bytes) in checkpoint_blobs for a
+    DeltaChannel field. The actual per-step writes live in checkpoint_writes
+    and are replayed through the reducer at load time.
+
+    Compare with `is DELTA_SENTINEL` — `loads_typed` always returns the same
+    module-level instance.
+    """
+
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return "DELTA_SENTINEL"
+
+
+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

@@ -999,13 +999,13 @@ def test_msgpack_nested_pydantic_serializes_as_dict(
     assert result == obj
 
 
-def test_delta_channel_sentinel_serde_round_trip() -> None:
-    from langgraph.checkpoint.base import DeltaChannelSentinel
+def test_delta_sentinel_serde_round_trip() -> None:
+    from langgraph.checkpoint.base import DELTA_SENTINEL
     from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 
     serde = JsonPlusSerializer()
-    original = DeltaChannelSentinel()
-    type_tag, blob = serde.dumps_typed(original)
-    assert type_tag == "delta"
+    type_tag, blob = serde.dumps_typed(DELTA_SENTINEL)
+    assert type_tag == "msgpack"
+    assert blob  # non-empty ext envelope
     loaded = serde.loads_typed((type_tag, blob))
-    assert isinstance(loaded, DeltaChannelSentinel)
+    assert loaded is DELTA_SENTINEL

libs/checkpoint/tests/test_memory.py

@@ -6,6 +6,7 @@ from langchain_core.runnables import RunnableConfig
 from pydantic import BaseModel
 
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     Checkpoint,
     CheckpointMetadata,
     create_checkpoint,
@@ -208,8 +209,6 @@ class TestMemorySaver:
 
 
 async def test_memory_saver() -> None:
-    from langgraph.checkpoint.memory import InMemorySaver
-
     memory_saver = InMemorySaver()
     assert isinstance(memory_saver, InMemorySaver)
 
@@ -324,20 +323,14 @@ def test_memory_saver_with_allowlist_proxy_isolated() -> None:
 
 class TestInMemorySaverDeltaChannel:
     def test_load_blobs_returns_sentinel_for_delta_channel(self) -> None:
-        """_load_blobs returns DeltaChannelSentinel for delta channels (reconstruction deferred)."""
-        from langgraph.checkpoint.base import (
-            DeltaChannelSentinel,
-            empty_checkpoint,
-        )
-
+        """_load_blobs returns DELTA_SENTINEL for delta channels (reconstruction deferred)."""
         saver = InMemorySaver()
         serde = JsonPlusSerializer()
 
         thread_id, ns, channel = "t1", "", "messages"
         v1 = "00000000000000000000000000000001.0000000000000000"
 
-        sentinel = DeltaChannelSentinel()
-        saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(sentinel)
+        saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(DELTA_SENTINEL)
 
         cp1 = empty_checkpoint()
         cp1["id"] = "cp1"
@@ -348,12 +341,12 @@ class TestInMemorySaverDeltaChannel:
 
         result = saver._load_blobs(thread_id, ns, {channel: v1})
         assert channel in result
-        assert isinstance(result[channel], DeltaChannelSentinel)
-
-    def test_get_channel_writes_collects_writes(self) -> None:
-        """get_channel_writes collects per-step writes oldest→newest."""
-        from langgraph.checkpoint.base import empty_checkpoint
+        assert result[channel] is DELTA_SENTINEL
 
+    def test_get_channel_writes_collects_ancestor_writes_only(self) -> None:
+        """_get_channel_writes_history collects ancestor writes oldest→newest,
+        and excludes writes stored at the target checkpoint itself (those are
+        pending writes for the next step, applied separately by pregel)."""
         saver = InMemorySaver()
         serde = JsonPlusSerializer()
 
@@ -367,18 +360,20 @@ class TestInMemorySaverDeltaChannel:
             "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
             "cp2": (serde.dumps_typed(cp2), serde.dumps_typed({}), "cp1"),
         }
-        # cp1 has a write for channel
+        # Writes stored at cp1 produced the cp1 snapshot; part of history.
         saver.writes[(thread_id, ns, "cp1")][("task1", 0)] = (
             "task1",
             channel,
             serde.dumps_typed({"content": "hi"}),
             "",
         )
-        # cp2 has a write for channel
+        # Writes stored at cp2 are pending — they will produce cp3 when the
+        # step that loaded cp2 completes. They MUST NOT appear in the
+        # reconstructed snapshot value at cp2.
         saver.writes[(thread_id, ns, "cp2")][("task2", 0)] = (
             "task2",
             channel,
-            serde.dumps_typed({"content": "bye"}),
+            serde.dumps_typed({"content": "pending"}),
             "",
         )
 
@@ -389,5 +384,282 @@ class TestInMemorySaverDeltaChannel:
                 "checkpoint_id": "cp2",
             }
         }
-        result = saver.get_channel_writes(config, channel)
-        assert result == [{"content": "hi"}, {"content": "bye"}]
+        result = saver._get_channel_writes_history(config, channel)
+        assert result.seed is DELTA_SENTINEL
+        values = [v for _, _, v in result.writes]
+        assert values == [{"content": "hi"}]
+
+    def test_get_channel_writes_at_root_returns_empty(self) -> None:
+        """Reconstructing the root checkpoint's state: no ancestors → []."""
+        saver = InMemorySaver()
+        serde = JsonPlusSerializer()
+        thread_id, ns, channel = "t1", "", "messages"
+
+        cp1 = empty_checkpoint()
+        cp1["id"] = "cp1"
+        saver.storage[thread_id][ns] = {
+            "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
+        }
+        saver.writes[(thread_id, ns, "cp1")][("task1", 0)] = (
+            "task1",
+            channel,
+            serde.dumps_typed({"content": "pending"}),
+            "",
+        )
+
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": "cp1",
+            }
+        }
+        result = saver._get_channel_writes_history(config, channel)
+        assert result.seed is DELTA_SENTINEL
+        assert result.writes == []
+
+
+class TestBaseFallbackGetChannelWrites:
+    """Exercises the `BaseCheckpointSaver._get_channel_writes_history` default
+    implementation — the path third-party savers inherit when they don't
+    override `_get_channel_writes_history` themselves.
+
+    Regression guard for a bug where the fallback passed the caller's config
+    (with `checkpoint_id`) straight to `self.list()`, which most savers
+    collapse to a single row — causing the fallback to return `[]`.
+    """
+
+    def _build_saver_with_chain(self) -> tuple[InMemorySaver, str, str]:
+        """Build an InMemorySaver with a 3-checkpoint chain and per-step writes
+        for a `messages` channel.
+
+        Returns `(saver, thread_id, namespace)`. The saver subclass deletes the
+        InMemorySaver override so the base class fallback is exercised.
+        """
+
+        class _ThirdPartyStyleSaver(InMemorySaver):
+            _get_channel_writes_history = (
+                InMemorySaver.__mro__[1]._get_channel_writes_history  # type: ignore[attr-defined]
+            )
+            _aget_channel_writes_history = (
+                InMemorySaver.__mro__[1]._aget_channel_writes_history  # type: ignore[attr-defined]
+            )
+
+        saver = _ThirdPartyStyleSaver()
+        serde = JsonPlusSerializer()
+        thread_id, ns, channel = "t1", "", "messages"
+
+        cp0 = empty_checkpoint()
+        cp0["id"] = "00000000000000000000000000000001.0000000000000000"
+        cp1 = empty_checkpoint()
+        cp1["id"] = "00000000000000000000000000000002.0000000000000000"
+        cp2 = empty_checkpoint()
+        cp2["id"] = "00000000000000000000000000000003.0000000000000000"
+        saver.storage[thread_id][ns] = {
+            cp0["id"]: (serde.dumps_typed(cp0), serde.dumps_typed({}), None),
+            cp1["id"]: (serde.dumps_typed(cp1), serde.dumps_typed({}), cp0["id"]),
+            cp2["id"]: (serde.dumps_typed(cp2), serde.dumps_typed({}), cp1["id"]),
+        }
+        # Writes under cp0 produced cp1's state; writes under cp1 produced cp2's.
+        saver.writes[(thread_id, ns, cp0["id"])][("task1", 0)] = (
+            "task1",
+            channel,
+            serde.dumps_typed({"content": "first"}),
+            "",
+        )
+        saver.writes[(thread_id, ns, cp1["id"])][("task2", 0)] = (
+            "task2",
+            channel,
+            serde.dumps_typed({"content": "second"}),
+            "",
+        )
+        return saver, thread_id, ns
+
+    def test_fallback_returns_ancestor_writes_oldest_first(self) -> None:
+        saver, thread_id, ns = self._build_saver_with_chain()
+        target_id = "00000000000000000000000000000003.0000000000000000"
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": target_id,
+            }
+        }
+
+        result = saver._get_channel_writes_history(config, "messages")
+
+        assert result.seed is DELTA_SENTINEL
+        values = [v for _, _, v in result.writes]
+        assert values == [{"content": "first"}, {"content": "second"}]
+
+    async def test_async_fallback_returns_ancestor_writes_oldest_first(self) -> None:
+        saver, thread_id, ns = self._build_saver_with_chain()
+        target_id = "00000000000000000000000000000003.0000000000000000"
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": target_id,
+            }
+        }
+
+        result = await saver._aget_channel_writes_history(config, "messages")
+
+        assert result.seed is DELTA_SENTINEL
+        values = [v for _, _, v in result.writes]
+        assert values == [{"content": "first"}, {"content": "second"}]
+
+    async def test_async_fallback_concurrent_tasks_do_not_interfere(self) -> None:
+        """Regression: the re-entrancy guard must be task-local, not thread-local.
+
+        Two concurrent `_aget_channel_writes_history` calls on the same
+        event-loop thread must each see their full reconstructed writes. A
+        `threading.local()` guard would let whichever task set it first
+        short-circuit the other to `writes=[]`.
+        """
+        import asyncio
+
+        saver, thread_id, ns = self._build_saver_with_chain()
+
+        # Force the two tasks to interleave across the `set(True)` boundary:
+        # each `aget_tuple` yields control, so if the guard were thread-local
+        # the second task would observe `active=True` set by the first.
+        orig_aget_tuple = saver.aget_tuple
+
+        async def slow_aget_tuple(config: RunnableConfig) -> Any:
+            await asyncio.sleep(0)
+            return await orig_aget_tuple(config)
+
+        saver.aget_tuple = slow_aget_tuple  # type: ignore[method-assign]
+
+        target_id = "00000000000000000000000000000003.0000000000000000"
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": target_id,
+            }
+        }
+
+        results = await asyncio.gather(
+            saver._aget_channel_writes_history(config, "messages"),
+            saver._aget_channel_writes_history(config, "messages"),
+        )
+
+        expected_values = [{"content": "first"}, {"content": "second"}]
+        for result in results:
+            assert result.seed is DELTA_SENTINEL
+            values = [v for _, _, v in result.writes]
+            assert values == expected_values
+
+
+class TestPreDeltaBlobTerminator:
+    """Verify the pre-delta blob terminator: when the ancestor walk hits a
+    checkpoint whose blob for the channel is a real value (not
+    DELTA_SENTINEL), reconstruction seeds from it and stops. This guards
+
+      * back-compat: a thread written by pre-delta code, then extended under
+        delta — reconstruction must return the correct value without walking
+        past the last pre-delta ancestor;
+      * perf: without the terminator, every reconstruct-after-migration would
+        walk all the way to the thread root.
+    """
+
+    def _build_mixed_thread(self) -> tuple[InMemorySaver, str, str, str, str]:
+        """Three-checkpoint chain: cp1 (pre-delta, blob=[A]), cp2 (delta,
+        write=B), cp3 (delta, write=C). Reconstructing at cp3 must yield
+        seed=[A] + writes=[B, C].
+
+        Returns `(saver, thread_id, ns, channel, cp3_id)`.
+        """
+        saver = InMemorySaver()
+        serde = JsonPlusSerializer()
+        thread_id, ns, channel = "t1", "", "messages"
+
+        v1 = "00000000000000000000000000000001.0"
+        v2 = "00000000000000000000000000000002.0"
+        v3 = "00000000000000000000000000000003.0"
+
+        # Pre-delta: cp1 stored a real blob for the channel.
+        saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(["A"])
+        # Delta-era: cp2 and cp3 store sentinels; real writes in checkpoint_writes.
+        saver.blobs[(thread_id, ns, channel, v2)] = serde.dumps_typed(DELTA_SENTINEL)
+        saver.blobs[(thread_id, ns, channel, v3)] = serde.dumps_typed(DELTA_SENTINEL)
+
+        cp1 = empty_checkpoint()
+        cp1["id"] = "cp1"
+        cp1["channel_versions"][channel] = v1
+        cp2 = empty_checkpoint()
+        cp2["id"] = "cp2"
+        cp2["channel_versions"][channel] = v2
+        cp3 = empty_checkpoint()
+        cp3["id"] = "cp3"
+        cp3["channel_versions"][channel] = v3
+
+        saver.storage[thread_id][ns] = {
+            "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
+            "cp2": (serde.dumps_typed(cp2), serde.dumps_typed({}), "cp1"),
+            "cp3": (serde.dumps_typed(cp3), serde.dumps_typed({}), "cp2"),
+        }
+        # Write under cp1 would be from the pre-delta era and MUST be ignored
+        # (the blob already captures it). We add one and assert it is not
+        # folded into the reconstructed result.
+        saver.writes[(thread_id, ns, "cp1")][("task0", 0)] = (
+            "task0",
+            channel,
+            serde.dumps_typed("PRE-DELTA-WRITE"),
+            "",
+        )
+        saver.writes[(thread_id, ns, "cp2")][("task2", 0)] = (
+            "task2",
+            channel,
+            serde.dumps_typed("B"),
+            "",
+        )
+        saver.writes[(thread_id, ns, "cp3")][("task3", 0)] = (
+            "task3",
+            channel,
+            serde.dumps_typed("PENDING-AT-TARGET"),
+            "",
+        )
+        return saver, thread_id, ns, channel, "cp3"
+
+    def test_seed_from_pre_delta_ancestor_blob(self) -> None:
+        saver, thread_id, ns, channel, target = self._build_mixed_thread()
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": target,
+            }
+        }
+
+        result = saver._get_channel_writes_history(config, channel)
+
+        # Seed came from the pre-delta blob at cp1.
+        assert result.seed == ["A"]
+        # Delta-era writes from cp2 replay through the reducer on top of seed.
+        # cp3 is the target — its own write is pending for the NEXT step and
+        # must be excluded.
+        values = [v for _, _, v in result.writes]
+        assert values == ["B"]
+
+    def test_pre_delta_blob_terminates_walk_before_older_writes(self) -> None:
+        """Writes stored at the pre-delta ancestor itself must not be replayed
+        (the blob subsumes them)."""
+        saver, thread_id, ns, channel, target = self._build_mixed_thread()
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": ns,
+                "checkpoint_id": target,
+            }
+        }
+
+        result = saver._get_channel_writes_history(config, channel)
+
+        values = [v for _, _, v in result.writes]
+        # The pre-delta write under cp1 must not appear (the blob subsumes it).
+        assert "PRE-DELTA-WRITE" not in values
+        # And the pending write at the target is never folded in.
+        assert "PENDING-AT-TARGET" not in values

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

@@ -119,12 +119,3 @@ class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
         Returns `True` if the channel was updated, `False` otherwise.
         """
         return False
-
-    def after_checkpoint(self, version: Any, checkpoint_id: str | None = None) -> None:
-        """Called after checkpoint() with the assigned version, and after
-        from_checkpoint() with the current channel version.
-
-        No-op by default. Override in channels that track their own version
-        for incremental checkpointing (e.g. DeltaChannel).
-        """
-        pass

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

@@ -1,131 +1,183 @@
 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 DeltaChannelSentinel
+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
-from langgraph.errors import EmptyChannelError
+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[list[Value], Value, DeltaChannelSentinel]
-):
-    """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.
+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.
 
-    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.
+    The reducer receives the current accumulated value and a batch of writes
+    in one call: `reducer(state, [write1, write2, ...]) -> new_state`.
 
-    Works with all checkpointers. Savers with dedicated implementations
-    (InMemorySaver, PostgresSaver) reconstruct in one pass; others fall back to
-    walking the checkpoint list.
+    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:
 
-    Usage::
+        reducer(reducer(state, xs), ys) == reducer(state, xs + ys)
 
-        class State(TypedDict):
-            messages: Annotated[list[AnyMessage], DeltaChannel(add_messages)]
-            # Dict-type reducer (type inferred from the Annotated outer type):
-            files: Annotated[dict, DeltaChannel(merge_files)]
+    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", "operator")
+    __slots__ = ("value", "reducer", "snapshot_frequency")
+    value: Value | Any
 
     def __init__(
         self,
-        operator: Callable[[list[Value], Any], list[Value]],
+        reducer: Callable[[Any, Sequence[Any]], Any],
+        typ: type[Value] | None = None,
+        *,
+        snapshot_frequency: int | None = None,
     ) -> None:
-        super().__init__(list)
-        self.operator = operator
-        self.value: list[Value] = []
+        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.operator.__name__ != "<lambda>"
-            and other.operator.__name__ != "<lambda>"
-        ):
-            return self.operator is other.operator
-        return True
+        if self.snapshot_frequency != other.snapshot_frequency:
+            return False
+        return _operators_equal(self.reducer, other.reducer)
 
     @property
     def ValueType(self) -> Any:
-        return list[self.typ]  # type: ignore[name-defined]
+        return self.typ
 
     @property
     def UpdateType(self) -> Any:
-        return self.typ | list[self.typ]  # type: ignore[name-defined]
+        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 = DeltaChannel(self.operator)
-        new.typ = self.typ
+        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 self.value.copy()
+        new.value = self.value if self.value is MISSING else _copy.copy(self.value)
         return new
 
     def from_checkpoint(self, checkpoint: Any) -> Self:
-        new = DeltaChannel(self.operator)
-        new.typ = self.typ
+        """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:
-            try:
-                new.value = new.typ()
-            except Exception:
-                new.value = []
-        elif isinstance(checkpoint, list):
-            # Flat list of write values (oldest→newest) from get_channel_writes.
-            try:
-                value: Any = new.typ()
-            except Exception:
-                value = []
-            for write in checkpoint:
-                value = new.operator(value, write)
-            new.value = value
+        if checkpoint is MISSING or checkpoint is DELTA_SENTINEL:
+            new.value = self.typ()
+        elif isinstance(checkpoint, _DeltaSnapshot):
+            new.value = checkpoint.value
         else:
-            # Backward compat: plain accumulated value (e.g. from a migrated thread).
-            try:
-                new.value = list(checkpoint)
-            except Exception:
-                new.value = []
+            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
-        seen_overwrite = False
-        for value in values:
-            is_overwrite, overwrite_value = _get_overwrite(value)
-            if is_overwrite:
-                if seen_overwrite:
-                    from langgraph.errors import (
-                        ErrorCode,
-                        InvalidUpdateError,
-                        create_error_message,
-                    )
-
+        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)
-                self.value = (
-                    list(overwrite_value) if overwrite_value is not None else self.typ()
-                )
-                seen_overwrite = True
-            elif not seen_overwrite:
-                base = self.typ() if self.value is MISSING else self.value
-                self.value = self.operator(base, value)
+                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) -> list[Value]:
+    def get(self) -> Any:
         if self.value is MISSING:
             raise EmptyChannelError()
         return self.value
@@ -133,5 +185,13 @@ class DeltaChannel(
     def is_available(self) -> bool:
         return self.value is not MISSING
 
-    def checkpoint(self) -> DeltaChannelSentinel:
-        return DeltaChannelSentinel()
+    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
@@ -48,7 +47,7 @@ from langgraph._internal._pydantic import create_model
 from langgraph._internal._runnable import coerce_to_runnable
 from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.binop import BinaryOperatorAggregate, _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
@@ -1671,17 +1670,19 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None:
         for item in meta:
             if isinstance(item, BaseChannel):
                 if isinstance(item, DeltaChannel) and hasattr(typ, "__origin__"):
-                    outer = _strip_extras(typ.__origin__)
-                    if outer in (
-                        collections.abc.Sequence,
-                        collections.abc.MutableSequence,
+                    origin = typ.__origin__
+                    # Unwrap parameterized Required[X]/NotRequired[X] to X
+                    # (e.g. Annotated[NotRequired[dict[...]], ...]).
+                    if hasattr(origin, "__origin__") and origin.__origin__ in (
+                        Required,
+                        NotRequired,
                     ):
-                        outer = list
-                    item.typ = outer
-                    try:
-                        item.value = outer()
-                    except Exception:
-                        item.value = []
+                        origin = origin.__args__[0]
+                    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,22 @@
 from __future__ import annotations
 
-import logging
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
+from typing import Any, cast
 
-from langgraph.checkpoint.base import Checkpoint
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import DELTA_SENTINEL, BaseCheckpointSaver, Checkpoint
 from langgraph.checkpoint.base.id import uuid6
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
 
 LATEST_VERSION = 4
 
-logger = logging.getLogger(__name__)
+GetNextVersion = Callable[[Any, None], Any]
 
 
 def empty_checkpoint() -> Checkpoint:
@@ -34,35 +37,87 @@ 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,
+    *,
+    saver: BaseCheckpointSaver | None = None,
+    config: RunnableConfig | None = None,
 ) -> tuple[Mapping[str, BaseChannel], ManagedValueMapping]:
-    """Get channels from a checkpoint."""
+    """Hydrate channels from a checkpoint.
+
+    For most channels, `spec.from_checkpoint(checkpoint["channel_values"][k])`
+    is sufficient. `DeltaChannel` is the exception: sentinel blobs require an
+    ancestor walk via `saver._get_channel_writes_history`. The walk terminates
+    at the nearest `_DeltaSnapshot` blob (step-based) or a pre-migration plain
+    value, so read depth is bounded by `snapshot_frequency`.
+    """
     channel_specs: dict[str, BaseChannel] = {}
     managed_specs: dict[str, ManagedValueSpec] = {}
     for k, v in specs.items():
@@ -70,10 +125,51 @@ def channels_from_checkpoint(
             channel_specs[k] = v
         else:
             managed_specs[k] = v
+
     channels: dict[str, BaseChannel] = {}
-    for k, v in channel_specs.items():
-        ch = v.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
-        ch.after_checkpoint(checkpoint["channel_versions"].get(k), checkpoint.get("id"))
+    for k, spec in channel_specs.items():
+        ch: BaseChannel
+        stored = checkpoint["channel_values"].get(k, MISSING)
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
+            delta_spec = cast(DeltaChannel, spec)
+            history = saver._get_channel_writes_history(config, k)
+            replay_ch = delta_spec.from_checkpoint(history.seed)
+            replay_ch.replay_writes(history.writes)
+            ch = replay_ch
+        else:
+            ch = spec.from_checkpoint(stored)
+        channels[k] = ch
+    return channels, managed_specs
+
+
+async def achannels_from_checkpoint(
+    specs: Mapping[str, BaseChannel | ManagedValueSpec],
+    checkpoint: Checkpoint,
+    *,
+    saver: BaseCheckpointSaver | None = None,
+    config: RunnableConfig | None = None,
+) -> tuple[Mapping[str, BaseChannel], ManagedValueMapping]:
+    """Async version of `channels_from_checkpoint`. See docstring there."""
+    channel_specs: dict[str, BaseChannel] = {}
+    managed_specs: dict[str, ManagedValueSpec] = {}
+    for k, v in specs.items():
+        if isinstance(v, BaseChannel):
+            channel_specs[k] = v
+        else:
+            managed_specs[k] = v
+
+    channels: dict[str, BaseChannel] = {}
+    for k, spec in channel_specs.items():
+        ch: BaseChannel
+        stored = checkpoint["channel_values"].get(k, MISSING)
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
+            delta_spec = cast(DeltaChannel, spec)
+            history = await saver._aget_channel_writes_history(config, k)
+            replay_ch = delta_spec.from_checkpoint(history.seed)
+            replay_ch.replay_writes(history.writes)
+            ch = replay_ch
+        else:
+            ch = spec.from_checkpoint(stored)
         channels[k] = ch
     return channels, managed_specs
 

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 (
@@ -92,6 +93,7 @@ from langgraph.pregel._algo import (
     task_path_str,
 )
 from langgraph.pregel._checkpoint import (
+    achannels_from_checkpoint,
     channels_from_checkpoint,
     copy_checkpoint,
     create_checkpoint,
@@ -188,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
@@ -406,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,
@@ -414,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)
@@ -890,13 +898,11 @@ 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",
         )
-        if do_checkpoint and self.channels:
-            for k, ch in self.channels.items():
-                ch.after_checkpoint(
-                    self.checkpoint["channel_versions"].get(k),
-                    self.checkpoint.get("id"),
-                )
         # sanitize TASK channel in the checkpoint before saving (durability=="exit")
         if TASKS in self.checkpoint["channel_values"] and any(
             isinstance(channel, UntrackedValue) for channel in self.channels.values()
@@ -1279,7 +1285,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         )
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
-            self.specs, self.checkpoint
+            self.specs,
+            self.checkpoint,
+            saver=self.checkpointer,
+            config=self.checkpoint_config,
         )
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
@@ -1374,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
@@ -1479,11 +1493,15 @@ 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)
         )
-        self.channels, self.managed = channels_from_checkpoint(
-            self.specs, self.checkpoint
+        self.channels, self.managed = await achannels_from_checkpoint(
+            self.specs,
+            self.checkpoint,
+            saver=self.checkpointer,
+            config=self.checkpoint_config,
         )
         self.stack.push(self._suppress_interrupt)
         self.status = "input"

libs/langgraph/langgraph/pregel/_utils.py

@@ -1,11 +1,9 @@
 from __future__ import annotations
 
 import ast
-import functools
 import inspect
 import re
 import textwrap
-import types
 from collections.abc import Callable
 from typing import Any
 
@@ -66,74 +64,46 @@ def find_subgraph_pregel(candidate: Runnable) -> PregelProtocol | None:
     return None
 
 
-@functools.lru_cache(maxsize=256)
-def _get_nonlocal_names(code: types.CodeType) -> frozenset[str]:
-    """Return the set of nonlocal variable names referenced by a function.
-
-    Cached by code object so the expensive source fetch + AST parse only
-    happens once per unique function definition across repeated graph compiles.
-
-    Args:
-        code: The code object of the function to analyse.
-
-    Returns:
-        Frozenset of variable names that the function reads from its enclosing
-        scope (free variables and globals referenced in function bodies).
-    """
-    try:
-        source = inspect.getsource(code)
-        tree = ast.parse(textwrap.dedent(source))
-        visitor = FunctionNonLocals()
-        visitor.visit(tree)
-        return frozenset(visitor.nonlocals)
-    except (SyntaxError, TypeError, OSError, SystemError):
-        return frozenset()
-
-
 def get_function_nonlocals(func: Callable) -> list[Any]:
     """Get the nonlocal variables accessed by a function.
 
-    The expensive source-parsing step is cached by code object; only the
-    cheap closure-variable lookup runs on every call.
-
     Args:
         func: The function to check.
 
     Returns:
         List[Any]: The nonlocal variables accessed by the function.
     """
-    actual_func = (
-        func.__wrapped__
-        if hasattr(func, "__wrapped__") and callable(func.__wrapped__)
-        else func
-    )
-    # Fast path: no free variables means nothing to scan.
-    if not actual_func.__code__.co_freevars:
-        return []
-
-    nonlocal_names = _get_nonlocal_names(actual_func.__code__)
-    if not nonlocal_names:
-        return []
-
-    closure = inspect.getclosurevars(actual_func)
-    candidates = {**closure.globals, **closure.nonlocals}
-    values: list[Any] = []
-    for k, v in candidates.items():
-        if k in nonlocal_names:
-            values.append(v)
-        for kk in nonlocal_names:
-            if "." in kk and kk.startswith(k):
-                vv = v
-                for part in kk.split(".")[1:]:
-                    if vv is None:
-                        break
-                    else:
-                        try:
-                            vv = getattr(vv, part)
-                        except AttributeError:
+    try:
+        code = inspect.getsource(func)
+        tree = ast.parse(textwrap.dedent(code))
+        visitor = FunctionNonLocals()
+        visitor.visit(tree)
+        values: list[Any] = []
+        closure = (
+            inspect.getclosurevars(func.__wrapped__)
+            if hasattr(func, "__wrapped__") and callable(func.__wrapped__)
+            else inspect.getclosurevars(func)
+        )
+        candidates = {**closure.globals, **closure.nonlocals}
+        for k, v in candidates.items():
+            if k in visitor.nonlocals:
+                values.append(v)
+            for kk in visitor.nonlocals:
+                if "." in kk and kk.startswith(k):
+                    vv = v
+                    for part in kk.split(".")[1:]:
+                        if vv is None:
                             break
-                else:
-                    values.append(vv)
+                        else:
+                            try:
+                                vv = getattr(vv, part)
+                            except AttributeError:
+                                break
+                    else:
+                        values.append(vv)
+    except (SyntaxError, TypeError, OSError, SystemError):
+        return []
+
     return values
 
 

libs/langgraph/langgraph/pregel/main.py

@@ -122,6 +122,7 @@ from langgraph.pregel._algo import (
 )
 from langgraph.pregel._call import identifier
 from langgraph.pregel._checkpoint import (
+    achannels_from_checkpoint,
     channels_from_checkpoint,
     copy_checkpoint,
     create_checkpoint,
@@ -1049,14 +1050,17 @@ class Pregel(
 
         step = saved.metadata.get("step", -1) + 1
         stop = step + 2
-        checkpoint = saved.checkpoint
         channels, managed = channels_from_checkpoint(
             self.channels,
-            checkpoint,
+            saved.checkpoint,
+            saver=self.checkpointer
+            if isinstance(self.checkpointer, BaseCheckpointSaver)
+            else None,
+            config=saved.config,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
-            checkpoint,
+            saved.checkpoint,
             saved.pending_writes or [],
             self.nodes,
             channels,
@@ -1169,14 +1173,17 @@ class Pregel(
 
         step = saved.metadata.get("step", -1) + 1
         stop = step + 2
-        checkpoint = saved.checkpoint
-        channels, managed = channels_from_checkpoint(
+        channels, managed = await achannels_from_checkpoint(
             self.channels,
-            checkpoint,
+            saved.checkpoint,
+            saver=self.checkpointer
+            if isinstance(self.checkpointer, BaseCheckpointSaver)
+            else None,
+            config=saved.config,
         )
         # tasks for this checkpoint
         next_tasks = prepare_next_tasks(
-            checkpoint,
+            saved.checkpoint,
             saved.pending_writes or [],
             self.nodes,
             channels,
@@ -1522,8 +1529,9 @@ class Pregel(
             saved = checkpointer.get_tuple(config)
             if saved is not None:
                 self._migrate_checkpoint(saved.checkpoint)
-            base_checkpoint = saved.checkpoint if saved else empty_checkpoint()
-            checkpoint = copy_checkpoint(base_checkpoint) if saved else base_checkpoint
+            checkpoint = (
+                copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
+            )
             checkpoint_previous_versions = (
                 saved.checkpoint["channel_versions"].copy() if saved else {}
             )
@@ -1542,6 +1550,11 @@ class Pregel(
             channels, managed = channels_from_checkpoint(
                 self.channels,
                 checkpoint,
+                saver=self.checkpointer
+                if saved is not None
+                and isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None,
+                config=saved.config if saved is not None else None,
             )
             values, as_node = updates[0][:2]
 
@@ -1967,8 +1980,9 @@ class Pregel(
             saved = await checkpointer.aget_tuple(config)
             if saved is not None:
                 self._migrate_checkpoint(saved.checkpoint)
-            base_checkpoint = saved.checkpoint if saved else empty_checkpoint()
-            checkpoint = copy_checkpoint(base_checkpoint) if saved else base_checkpoint
+            checkpoint = (
+                copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
+            )
             checkpoint_previous_versions = (
                 saved.checkpoint["channel_versions"].copy() if saved else {}
             )
@@ -1984,9 +1998,14 @@ class Pregel(
             )
             if saved:
                 checkpoint_config = patch_configurable(config, saved.config[CONF])
-            channels, managed = channels_from_checkpoint(
+            channels, managed = await achannels_from_checkpoint(
                 self.channels,
                 checkpoint,
+                saver=self.checkpointer
+                if saved is not None
+                and isinstance(self.checkpointer, BaseCheckpointSaver)
+                else None,
+                config=saved.config if saved is not None else None,
             )
             values, as_node = updates[0][:2]
             # no values, just clear all tasks

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,18 +1,34 @@
 import operator
 from collections.abc import Sequence
+from typing import Annotated
 
 import pytest
+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.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 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
@@ -95,70 +111,57 @@ 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 DeltaChannelSentinel
+    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 isinstance(d1, DeltaChannelSentinel)
+    assert d1 is DELTA_SENTINEL
 
-    # Step 2: another message
     ch.update([AIMessage(content="hello", id="a1")])
     d2 = ch.checkpoint()
-    assert isinstance(d2, DeltaChannelSentinel)
+    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"
 
 
 def test_delta_channel_from_checkpoint_writes_list() -> None:
-    """from_checkpoint with a flat list of individual writes replays them 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)
-    # Each element is one write value (as stored in checkpoint_writes)
-    writes = [
-        HumanMessage(content="hi", id="h1"),
-        AIMessage(content="hello", id="a1"),
-        HumanMessage(content="bye", id="h2"),
-    ]
-    ch = spec.from_checkpoint(writes)
+    """replay_writes on a fresh channel replays through the operator."""
+    spec = DeltaChannel(_messages_delta_reducer, list)
+    ch = spec.from_checkpoint(DELTA_SENTINEL)
+    ch.replay_writes(
+        [
+            ("t0", "messages", HumanMessage(content="hi", id="h1")),
+            ("t1", "messages", AIMessage(content="hello", id="a1")),
+            ("t2", "messages", HumanMessage(content="bye", id="h2")),
+        ]
+    )
     msgs = ch.get()
     assert len(msgs) == 3
     assert msgs[0].content == "hi"
@@ -167,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 DeltaChannelSentinel
-
-    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 isinstance(d, DeltaChannelSentinel)
-    # After overwrite, value is reset to only the new message
+    assert d is DELTA_SENTINEL
     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() == [
@@ -216,77 +199,159 @@ 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
-    writes = [
-        HumanMessage(content="hi", id="h1"),
-        AIMessage(content="hello", id="a1"),
-        RemoveMessage(id="a1"),
-    ]
-    ch2 = spec.from_checkpoint(writes)
+    ch2 = spec.from_checkpoint(DELTA_SENTINEL)
+    ch2.replay_writes(
+        [
+            ("t0", "messages", HumanMessage(content="hi", id="h1")),
+            ("t1", "messages", AIMessage(content="hello", id="a1")),
+            ("t2", "messages", RemoveMessage(id="a1")),
+        ]
+    )
     assert ch2.get() == [HumanMessage(content="hi", id="h1")]
 
 
 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
-    writes = [
-        HumanMessage(content="original", id="h1"),
-        HumanMessage(content="updated", id="h1"),
-    ]
-    ch2 = spec.from_checkpoint(writes)
+    ch2 = spec.from_checkpoint(DELTA_SENTINEL)
+    ch2.replay_writes(
+        [
+            ("t0", "messages", HumanMessage(content="original", id="h1")),
+            ("t1", "messages", HumanMessage(content="updated", id="h1")),
+        ]
+    )
     assert len(ch2.get()) == 1
     assert ch2.get()[0].content == "updated"
 
 
 def test_delta_channel_checkpoint_returns_sentinel() -> None:
-    """checkpoint() always returns DeltaChannelSentinel regardless of state."""
-    from langgraph.checkpoint.base import DeltaChannelSentinel
-
-    from langgraph.channels.delta import DeltaChannel
-    from langgraph.graph.message import add_messages
-
-    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
-    assert isinstance(ch.checkpoint(), DeltaChannelSentinel)
-
-    from langchain_core.messages import HumanMessage
+    """checkpoint() always returns DELTA_SENTINEL regardless of state."""
+    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    assert ch.checkpoint() is DELTA_SENTINEL
 
     ch.update([HumanMessage(content="hi", id="h1")])
-    assert isinstance(ch.checkpoint(), DeltaChannelSentinel)
+    assert ch.checkpoint() is DELTA_SENTINEL
+
+
+# ---------------------------------------------------------------------------
+# DeltaChannel — snapshot frequency
+# ---------------------------------------------------------------------------
+
+
+def test_delta_channel_snapshot_step_based() -> None:
+    """Snapshots fire on every Nth step regardless of whether the channel was written.
+
+    With snapshot_frequency=N, every Nth pregel step produces a _DeltaSnapshot
+    blob — even if the channel had no write that step (eager snapshot). This
+    bounds the ancestor walk to at most N steps on any read.
+    """
+
+    class State(TypedDict):
+        messages: Annotated[
+            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=5)
+        ]
+        other: str
+
+    def node_a(state: State) -> dict:
+        i = len(state["messages"]) // 2
+        return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
+
+    def node_b(state: State) -> dict:
+        return {"other": "y"}
+
+    g = StateGraph(State)
+    g.add_node("a", node_a)
+    g.add_node("b", node_b)
+    g.add_edge(START, "a")
+    g.add_edge("a", "b")
+    saver = InMemorySaver()
+    graph = g.compile(checkpointer=saver)
+
+    config = {"configurable": {"thread_id": "t1"}}
+    for i in range(6):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")], "other": ""},
+            config,
+        )
+
+    msg_blob_values = [
+        saver.serde.loads_typed((type_tag, blob))
+        for k, (type_tag, blob) in saver.blobs.items()
+        if k[2] == "messages" and type_tag == "msgpack" and blob
+    ]
+    snapshots = [v for v in msg_blob_values if isinstance(v, _DeltaSnapshot)]
+    assert snapshots, "expected at least one _DeltaSnapshot blob for messages"
+
+    state = graph.get_state(config)
+    assert len(state.values["messages"]) == 12  # 6 human + 6 AI
+
+
+def test_delta_channel_snapshot_fires_even_when_not_written() -> None:
+    """Eager snapshot: _DeltaSnapshot stored at snapshot step even when the
+    channel had no write that step (node_b doesn't touch messages).
+    """
+
+    class State(TypedDict):
+        messages: Annotated[
+            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3)
+        ]
+        tick: int
+
+    def writer(state: State) -> dict:
+        i = len(state["messages"]) // 2
+        return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
+
+    def ticker(state: State) -> dict:
+        return {"tick": state["tick"] + 1}
+
+    g = StateGraph(State)
+    g.add_node("writer", writer)
+    g.add_node("ticker", ticker)
+    g.add_edge(START, "writer")
+    g.add_edge("writer", "ticker")
+    saver = InMemorySaver()
+    graph = g.compile(checkpointer=saver)
+
+    config = {"configurable": {"thread_id": "t1"}}
+    for i in range(5):
+        graph.invoke(
+            {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")], "tick": 0},
+            config,
+        )
+
+    msg_blobs = {
+        k: saver.serde.loads_typed((t, b))
+        for k, (t, b) in saver.blobs.items()
+        if k[2] == "messages" and t == "msgpack" and b
+    }
+    snapshots = {k: v for k, v in msg_blobs.items() if isinstance(v, _DeltaSnapshot)}
+    assert snapshots, (
+        "eager snapshots must fire even on steps where messages wasn't written"
+    )
+
+    state = graph.get_state(config)
+    assert len(state.values["messages"]) == 10  # 5 human + 5 AI
+
+
+# ---------------------------------------------------------------------------
+# DeltaChannel — end-to-end (InMemorySaver)
+# ---------------------------------------------------------------------------
 
 
 def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     """InMemorySaver assembles writes from checkpoint_writes inside get_tuple."""
-    from typing import Annotated
-
-    from langchain_core.messages import AIMessage, HumanMessage
-    from langgraph.checkpoint.memory import InMemorySaver
-    from typing_extensions import TypedDict
-
-    from langgraph.channels.delta import DeltaChannel
-    from langgraph.graph import START, StateGraph
-    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(add_messages)]
+        messages: Annotated[list, DeltaChannel(_messages_delta_reducer, list)]
 
     n = {"v": 0}
 
@@ -304,104 +369,285 @@ 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 must return a resolved list (not DeltaChannelSentinel)
-    from langgraph.checkpoint.base import DeltaChannelSentinel
-
     saved = saver.get_tuple(config)
     assert saved is not None
     assert "messages" in saved.checkpoint["channel_values"]
-    assert not isinstance(
-        saved.checkpoint["channel_values"]["messages"], DeltaChannelSentinel
-    )
-    assert isinstance(saved.checkpoint["channel_values"]["messages"], list)
+    assert saved.checkpoint["channel_values"]["messages"] is DELTA_SENTINEL
 
     state = graph.get_state(config)
     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 DeltaChannelSentinel
 
-    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}])
     d1 = ch.checkpoint()
-    assert isinstance(d1, DeltaChannelSentinel)
+    assert d1 is DELTA_SENTINEL
 
     ch.update([{"b": 2}])
     d2 = ch.checkpoint()
-    assert isinstance(d2, DeltaChannelSentinel)
+    assert d2 is DELTA_SENTINEL
 
     assert ch.get() == {"a": 1, "b": 2}
 
 
 def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
-    """from_checkpoint with a writes list replays correctly through a dict merge reducer."""
+    """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)
-    # Each element is one write value (oldest→newest)
-    writes = [{"a": 1}, {"b": 2}, {"c": 3}]
-    ch = spec.from_checkpoint(writes)
+    ch = spec.from_checkpoint(DELTA_SENTINEL)
+    ch.replay_writes(
+        [
+            ("t0", "files", {"a": 1}),
+            ("t1", "files", {"b": 2}),
+            ("t2", "files", {"c": 3}),
+        ]
+    )
     assert ch.get() == {"a": 1, "b": 2, "c": 3}
 
 
 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
-    writes = [
-        {"file1.py": "content1", "file2.py": "content2"},
-        {"file1.py": None, "file3.py": "content3"},
-    ]
     spec = _delta_channel_with_type(merge_files, dict)
-    ch2 = spec.from_checkpoint(writes)
+    ch2 = spec.from_checkpoint(DELTA_SENTINEL)
+    ch2.replay_writes(
+        [
+            ("t0", "files", {"file1.py": "content1", "file2.py": "content2"}),
+            ("t1", "files", {"file1.py": None, "file3.py": "content3"}),
+        ]
+    )
     assert ch2.get() == {"file2.py": "content2", "file3.py": "content3"}
+
+
+def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
+    """Overwrite(dict) in update() must preserve dict shape, not coerce to list."""
+
+    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."""
+
+    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)
+    ch.replay_writes(
+        [
+            ("t0", "files", {"a": 1}),
+            ("t1", "files", Overwrite({"x": 10, "y": 20})),
+            ("t2", "files", {"z": 30}),
+        ]
+    )
+    assert ch.get() == {"x": 10, "y": 20, "z": 30}
+
+
+def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
+    """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`."""
+
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
+
+    annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
+    ch = _get_channel("files", annotation).from_checkpoint(MISSING)
+    assert ch.get() == {}
+    ch.update([{"a": 1}])
+    ch.update([{"b": 2}])
+    assert ch.get() == {"a": 1, "b": 2}
+
+
+def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
+    """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel."""
+
+    def merge_files(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            for k, v in w.items():
+                if v is None:
+                    result.pop(k, None)
+                else:
+                    result[k] = v
+        return result
+
+    class State(TypedDict):
+        files: Annotated[dict[str, str], DeltaChannel(merge_files)]
+
+    turn = {"v": 0}
+
+    def write_file(state: State) -> dict:
+        turn["v"] += 1
+        n = turn["v"]
+        return {"files": {f"/doc_{n}.txt": f"content for turn {n}"}}
+
+    builder = StateGraph(State)
+    builder.add_node("write_file", write_file)
+    builder.add_edge(START, "write_file")
+    saver = InMemorySaver()
+    graph = builder.compile(checkpointer=saver)
+    config = {"configurable": {"thread_id": "fs"}}
+
+    for _ in range(3):
+        graph.invoke({"files": {}}, config)
+
+    saved = saver.get_tuple(config)
+    assert saved is not None
+    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",
+        "/doc_2.txt": "content for turn 2",
+        "/doc_3.txt": "content for turn 3",
+    }
+
+    def delete_file(state: State) -> dict:
+        return {"files": {"/doc_1.txt": None}}
+
+    builder2 = StateGraph(State)
+    builder2.add_node("write_file", write_file)
+    builder2.add_node("delete_file", delete_file)
+    builder2.add_edge(START, "write_file")
+    builder2.add_edge("write_file", "delete_file")
+    turn["v"] = 0
+    saver2 = InMemorySaver()
+    graph2 = builder2.compile(checkpointer=saver2)
+    config2 = {"configurable": {"thread_id": "fs2"}}
+    graph2.invoke({"files": {}}, config2)
+    state2 = graph2.get_state(config2)
+    assert state2.values["files"] == {}
+
+
+def test_delta_channel_dict_reducer_backwards_compat() -> None:
+    """A pre-DeltaChannel dict checkpoint must load as a dict, not be listified."""
+
+    def merge_dicts(state: dict, writes: list) -> dict:
+        result = dict(state)
+        for w in writes:
+            result.update(w)
+        return result
+
+    spec = _delta_channel_with_type(merge_dicts, dict)
+    old_value = {"a": 1, "b": 2}
+    ch = spec.from_checkpoint(old_value)
+    assert ch.get() == {"a": 1, "b": 2}
+
+
+# ---------------------------------------------------------------------------
+# DeltaChannel — seed / pre-delta migration
+# ---------------------------------------------------------------------------
+
+
+def test_delta_channel_from_checkpoint_honors_seed() -> None:
+    """A non-sentinel value to from_checkpoint is used as the pre-delta seed.
+
+    Guards the pre-delta migration path: when the saver's ancestor walk hits
+    a pre-DeltaChannel blob it passes it as `seed` so replay reconstructs
+    the post-migration state correctly rather than replaying from empty.
+    """
+    spec = DeltaChannel(_messages_delta_reducer, list)
+    seed = [HumanMessage(content="pre-delta", id="p1")]
+    ch = spec.from_checkpoint(seed)
+    ch.replay_writes(
+        [
+            ("t0", "messages", AIMessage(content="delta-1", id="d1")),
+            ("t1", "messages", HumanMessage(content="delta-2", id="d2")),
+        ]
+    )
+    msgs = ch.get()
+    assert [m.content for m in msgs] == ["pre-delta", "delta-1", "delta-2"]
+
+
+def test_delta_channel_from_checkpoint_seed_without_writes() -> None:
+    """Reconstruction at a pre-delta ancestor with no newer deltas returns
+    just the seed — the saver's terminator fired immediately."""
+    spec = DeltaChannel(_messages_delta_reducer, list)
+    seed = [HumanMessage(content="only-snap", id="s1")]
+    ch = spec.from_checkpoint(seed)
+    ch.replay_writes([])
+    assert ch.get() == seed
+
+
+def test_delta_channel_from_checkpoint_seed_none_is_distinct_from_sentinel() -> None:
+    """`seed=None` must start replay from None, not from an empty channel.
+
+    The DELTA_SENTINEL / MISSING sentinels mean 'no seed'; passing `None`
+    explicitly should feed None to the reducer as the left operand.
+    """
+
+    def replace(state, writes):
+        return writes[-1] if writes else state
+
+    spec = DeltaChannel(replace, list)
+    ch = spec.from_checkpoint(None)
+    ch.replay_writes([("t0", "x", "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
@@ -28,21 +29,15 @@ from typing_extensions import TypedDict
 
 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
@@ -54,7 +49,9 @@ except ImportError:
 _HUMAN_TEMPLATE = (
     "I need help understanding the implications of {topic} on our system architecture. "
     "Specifically, I'm concerned about how this interacts with our existing {concern} "
-    "and whether we need to refactor the {component} layer before proceeding."
+    "and whether we need to refactor the {component} layer before proceeding. "
+    "We've had prior incidents in this area and want to be deliberate. "
+    "What should we prioritize first, and are there known failure modes we should design around from the start?"
 )
 
 _AI_TEMPLATE = (
@@ -121,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]},
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -167,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"}}
@@ -182,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
 
 
@@ -204,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"
@@ -214,146 +223,212 @@ 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]
+
+@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 (recursive CTE)", "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 = []
-    for turns in TURN_COUNTS:
+    rows: list[tuple[int, Any, Any, Any, Any, Any, Any]] = []
+    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:
             d_wt, d_rt, d_bytes = _run_turns(turns, DeltaState, saver)
-        rows.append((turns, b_bytes, d_bytes, b_rt, d_rt))
+        rows.append((turns, b_bytes, d_bytes, b_rt, d_rt, b_wt, d_wt))
+    for turns in DELTA_ONLY_TURN_COUNTS:
+        with _make_saver() as saver:
+            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 = 70
-    print("Storage (checkpoint blob bytes)")
-    print("=" * W)
+    def _bytes_or_na(v: Any) -> str:
+        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 size':>10}  {'add_msgs':>12}  {'delta':>12}  {'savings':>8}"
+        f"  {'turns':>6}  {'ctx':>10}  {'add_msgs':>12}  {'delta(inf)':>12}  {'savings':>8}"
     )
-    print("-" * W)
-    storage_results = []
-    for turns, b_bytes, d_bytes, b_rt, d_rt in rows:
-        if b_bytes < 0:
-            print(
-                f"{turns:>6}  {_approx_tokens(turns):>10}  {'n/a':>12}  {'n/a':>12}  {'n/a':>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"
         else:
             ratio = b_bytes / d_bytes if d_bytes else float("inf")
-            storage_results.append((turns, b_bytes, d_bytes, ratio))
-            print(
-                f"{turns:>6}  {_approx_tokens(turns):>10}  "
-                f"{_fmt_bytes(b_bytes):>12}  {_fmt_bytes(d_bytes):>12}  "
-                f"{ratio:>7.0f}x"
-            )
-    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)
-    for turns, b_bytes, d_bytes, b_rt, d_rt in rows:
+            ratio_str = f"{ratio:.0f}x"
         print(
-            f"{turns:>6}  {_approx_tokens(turns):>10}  "
-            f"{b_rt * 1000:>10.1f}ms  {d_rt * 1000:>10.1f}ms"
+            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()
 
-    if storage_results:
-        turns, b_bytes, d_bytes, ratio = storage_results[-1]
-        b_rt = rows[-1][-2]
-        d_rt = rows[-1][-1]
+    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"At {turns} turns: {_fmt_bytes(b_bytes)} → {_fmt_bytes(d_bytes)} ({ratio:.0f}x less storage); "
-            f"read {b_rt * 1000:.1f}ms → {d_rt * 1000:.1f}ms"
+            f"  {turns:>6}  {_approx_tokens(turns):>10}  "
+            f"{_ms_or_na(b_rt):>12}  {_ms_or_na(d_rt):>12}"
         )
-        print()
 
-    print("Legend:")
-    print("  add_msgs = Annotated[list, add_messages]  — O(N²) storage")
-    print(
-        "  delta    = DeltaChannel(add_messages)     — O(N) storage, full chain replay"
-    )
+
+def run_baseline_benchmark() -> None:
+    print()
+    print("Part 1 — DeltaChannel(inf) vs add_messages: storage & latency")
+    print("=" * 72)
+    for cp_label, cp_hint in _checkpointers():
+        _run_baseline_for_checkpointer(cp_label, cp_hint)
     print()
 
 
 # ---------------------------------------------------------------------------
-# 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)
@@ -363,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

@@ -0,0 +1,613 @@
+"""Tests for the BinaryOperatorAggregate -> DeltaChannel migration path.
+
+A thread written under `BinaryOperatorAggregate(...)` must keep working
+after its annotation is swapped to `DeltaChannel(...)` on the same
+checkpointer — pre-migration state visible at each *settled* ancestor
+checkpoint is preserved, and post-migration writes fold on top through
+the reducer.
+
+Mechanism under test: the saver's `_get_channel_writes_history(config,
+channel)` walks the parent chain; when it encounters an ancestor whose
+`channel_values[channel]` is a real value (not `DELTA_SENTINEL`), it
+returns that as the `seed`. `DeltaChannel.from_checkpoint(seed)` uses
+it as the base value, and `replay_writes(writes)` folds on-path deltas.
+
+Scenarios covered:
+
+1. **Basic migration (sync + async)**: build pre-migration state with
+   `BinaryOperatorAggregate`, swap the annotation to `DeltaChannel` on
+   the same checkpointer, and verify that every settled pre-migration
+   super-step boundary (`next=('__start__',)`) round-trips exactly
+   under the delta-channel view.
+2. **Time travel into a pre-migration checkpoint** after migration —
+   `graph.get_state(pre_migration_config)` at a settled ancestor
+   returns the same state as under the binop channel.
+3. **Continuing a migrated thread**: driving one more super-step after
+   migration produces a state that includes the pre-migration settled
+   prefix plus the new delta write — proving `from_checkpoint(seed)` +
+   `replay_writes` correctly fold post-migration deltas onto the
+   pre-migration seed.
+4. **Base-saver fallback path**: a third-party-style subclass that
+   removes the optimized `InMemorySaver` override and falls back to
+   `BaseCheckpointSaver._get_channel_writes_history` must produce the
+   same result as the optimized path.
+5. **Channel-type isolation across threads**: two threads on the same
+   checkpointer under the delta-channel graph — one freshly-started,
+   one migrated from pre-migration state — don't cross-contaminate.
+   The parent-chain walk is scoped to the thread.
+
+TODO: add postgres variants in the existing `libs/checkpoint-postgres`
+test files (different fixture setup; not this file).
+"""
+
+from __future__ import annotations
+
+import operator
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage
+from langgraph.checkpoint.memory import InMemorySaver
+from typing_extensions import TypedDict
+
+from langgraph.channels.binop import BinaryOperatorAggregate
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import _messages_delta_reducer, add_messages
+
+pytestmark = pytest.mark.anyio
+
+
+# ---------------------------------------------------------------------------
+# Graph factories
+#
+# A minimal reducer (`operator.add` on lists of str) with a noop node keeps
+# state change localized to the HumanMessage-like payload passed through
+# `invoke`. That isolates the pre/post-migration parity assertions to
+# channel-hydration semantics.
+# ---------------------------------------------------------------------------
+
+
+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)]
+
+    return (
+        StateGraph(BinopState)
+        .add_node("noop", _noop)
+        .add_edge(START, "noop")
+        .add_edge("noop", END)
+        .compile(checkpointer=checkpointer)
+    )
+
+
+def _delta_graph(checkpointer: Any) -> Any:
+    class DeltaState(TypedDict):
+        items: Annotated[list, DeltaChannel(_list_concat)]
+
+    return (
+        StateGraph(DeltaState)
+        .add_node("noop", _noop)
+        .add_edge(START, "noop")
+        .add_edge("noop", END)
+        .compile(checkpointer=checkpointer)
+    )
+
+
+def _drive(graph: Any, config: dict, tag: str, n: int) -> None:
+    for i in range(n):
+        graph.invoke({"items": [f"{tag}{i}"]}, config)
+
+
+async def _adrive(graph: Any, config: dict, tag: str, n: int) -> None:
+    for i in range(n):
+        await graph.ainvoke({"items": [f"{tag}{i}"]}, config)
+
+
+def _settled_boundaries(history: list) -> list[tuple[dict, list]]:
+    """Return `[(config, items), ...]` for every checkpoint in `history`
+    whose `next == ('__start__',)` — the stable boundaries between invokes.
+    """
+    return [
+        (s.config, list(s.values.get("items", [])))
+        for s in history
+        if s.next == ("__start__",)
+    ]
+
+
+# ---------------------------------------------------------------------------
+# 1. Basic migration (sync + async)
+# ---------------------------------------------------------------------------
+
+
+def test_basic_migration_preserves_pre_migration_state() -> None:
+    """Build state under `BinaryOperatorAggregate`, migrate to
+    `DeltaChannel` on the same checkpointer, and verify that every
+    settled pre-migration super-step boundary round-trips exactly.
+
+    Settled boundaries (`next=('__start__',)`) are the stable hydration
+    targets for the migration path: writes that produced the NEXT
+    super-step are kept as `pending_writes` on the ancestor, so walking
+    from a descendant finds the ancestor's blob as the seed and
+    reconstructs the correct state.
+    """
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "basic-sync"}}
+
+    # Pre-migration: accumulate items across 3 invokes.
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 3)
+
+    pre_boundaries = _settled_boundaries(list(binop.get_state_history(config)))
+    assert len(pre_boundaries) >= 2, "expected multiple settled boundaries"
+
+    # Migrate: swap the annotation on the same checkpointer.
+    delta = _delta_graph(checkpointer)
+
+    for cfg, items in pre_boundaries:
+        snap = delta.get_state(cfg)
+        assert list(snap.values.get("items", [])) == items, (
+            f"snapshot mismatch at {cfg['configurable']['checkpoint_id']}: "
+            f"expected {items}, got {snap.values.get('items', [])}"
+        )
+
+
+async def test_basic_migration_preserves_pre_migration_state_async() -> None:
+    """Async variant of the basic migration scenario."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "basic-async"}}
+
+    binop = _binop_graph(checkpointer)
+    await _adrive(binop, config, "u", 3)
+
+    pre_history = [s async for s in binop.aget_state_history(config)]
+    pre_boundaries = _settled_boundaries(pre_history)
+    assert len(pre_boundaries) >= 2
+
+    delta = _delta_graph(checkpointer)
+
+    for cfg, items in pre_boundaries:
+        snap = await delta.aget_state(cfg)
+        assert list(snap.values.get("items", [])) == items, (
+            f"async snapshot mismatch at {cfg['configurable']['checkpoint_id']}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# 2. Time travel into a pre-migration checkpoint after migration
+# ---------------------------------------------------------------------------
+
+
+def test_time_travel_into_pre_migration_checkpoint() -> None:
+    """After migration, `graph.get_state(pre_migration_config)` at a
+    settled ancestor returns the state as stored at that point."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "time-travel"}}
+
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 3)
+
+    pre_boundaries = _settled_boundaries(list(binop.get_state_history(config)))
+    assert pre_boundaries, "no settled ancestors to time-travel to"
+
+    delta = _delta_graph(checkpointer)
+
+    # Pick the oldest non-empty boundary — a long distance to walk back.
+    non_empty = [(cfg, items) for cfg, items in pre_boundaries if items]
+    assert non_empty, "expected at least one non-empty boundary"
+    target_cfg, expected_items = non_empty[-1]
+
+    snap = delta.get_state(target_cfg)
+    assert list(snap.values.get("items", [])) == expected_items
+
+
+# ---------------------------------------------------------------------------
+# 3. Continuing a migrated thread: deltas fold onto pre-migration seed
+# ---------------------------------------------------------------------------
+
+
+def test_continuing_migrated_thread_folds_deltas_on_seed() -> None:
+    """Resume a pre-migration settled ancestor via `invoke(None, cfg)`
+    under the delta-channel graph. Since the pre-migration checkpoint
+    has an existing `pending_writes` entry (the input for the NEXT
+    super-step), re-running from that ancestor reproduces the same
+    post-ancestor state as the original binop run.
+
+    This proves the seed-terminator + write-replay pipeline works
+    end-to-end across the migration boundary.
+    """
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "continue"}}
+
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 2)
+
+    # Pick the oldest settled boundary with non-empty state.
+    pre_boundaries = _settled_boundaries(list(binop.get_state_history(config)))
+    target_cfg, seed_items = next(
+        (cfg, items) for cfg, items in reversed(pre_boundaries) if items
+    )
+    assert seed_items, "need a non-empty seed boundary"
+
+    # Migrate and resume from the pre-migration ancestor. `invoke(None,
+    # cfg)` replays the pending writes staged at `cfg` under the new
+    # channel; the reducer folds those deltas onto the seed.
+    delta = _delta_graph(checkpointer)
+    result = delta.invoke(None, target_cfg)
+
+    # The resumed state must include the pre-migration seed items in order.
+    result_items = list(result.get("items", []))
+    for idx, prefix_item in enumerate(seed_items):
+        assert result_items[idx] == prefix_item, (
+            f"pre-migration seed item at {idx} not preserved: "
+            f"got {result_items[: idx + 1]}, expected {seed_items}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# 4. Base-saver fallback path
+# ---------------------------------------------------------------------------
+
+
+class _ThirdPartyStyleSaver(InMemorySaver):
+    """Simulates a third-party saver that inherits the reference
+    `_get_channel_writes_history` implementation from
+    `BaseCheckpointSaver` rather than overriding it.
+
+    We rebind the two methods to the base-class versions (via MRO) so
+    the fallback path is exercised even though the storage layer is
+    still the in-memory one.
+    """
+
+    # MRO: [_ThirdPartyStyleSaver, InMemorySaver, BaseCheckpointSaver, ...]
+    _get_channel_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._get_channel_writes_history  # type: ignore[attr-defined]
+    )
+    _aget_channel_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._aget_channel_writes_history  # type: ignore[attr-defined]
+    )
+
+
+def test_base_saver_fallback_matches_optimized_override() -> None:
+    """The reference `BaseCheckpointSaver` implementation must produce
+    the same migration behavior as the optimized `InMemorySaver`
+    override. We drive the same migration scenario through both savers
+    and assert per-snapshot parity in the delta-channel view."""
+
+    # Fast path: optimized InMemorySaver override.
+    fast_saver = InMemorySaver()
+    fast_config = {"configurable": {"thread_id": "fast"}}
+    fast_binop = _binop_graph(fast_saver)
+    _drive(fast_binop, fast_config, "u", 3)
+    fast_delta = _delta_graph(fast_saver)
+    fast_history = [
+        (s.next, list(s.values.get("items", [])))
+        for s in fast_delta.get_state_history(fast_config)
+    ]
+
+    # Slow path: base-class fallback.
+    slow_saver = _ThirdPartyStyleSaver()
+    slow_config = {"configurable": {"thread_id": "slow"}}
+    slow_binop = _binop_graph(slow_saver)
+    _drive(slow_binop, slow_config, "u", 3)
+    slow_delta = _delta_graph(slow_saver)
+    slow_history = [
+        (s.next, list(s.values.get("items", [])))
+        for s in slow_delta.get_state_history(slow_config)
+    ]
+
+    assert slow_history == fast_history, (
+        "base-saver fallback should match optimized-override behavior; "
+        f"fast={fast_history}, slow={slow_history}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# 5. Thread isolation under mixed-generation storage
+# ---------------------------------------------------------------------------
+
+
+def test_delta_and_migrated_threads_do_not_cross_contaminate() -> None:
+    """Two threads sharing a checkpointer — one migrated from
+    pre-migration state, one freshly-started under DeltaChannel — must
+    maintain independent state. The parent-chain walk in
+    `_get_channel_writes_history` must be scoped to the target thread.
+    """
+
+    checkpointer = InMemorySaver()
+    migrated_cfg = {"configurable": {"thread_id": "migrated"}}
+    fresh_cfg = {"configurable": {"thread_id": "fresh"}}
+
+    # Thread A: pre-migration build-up.
+    binop = _binop_graph(checkpointer)
+    _drive(binop, migrated_cfg, "m", 2)
+
+    # Thread B: fresh delta-channel run.
+    delta = _delta_graph(checkpointer)
+    _drive(delta, fresh_cfg, "f", 2)
+
+    # Thread A: migrate and confirm its state is anchored in its own
+    # thread's pre-migration history (tag 'm'), never mixing in tag 'f'.
+    migrated_boundaries = _settled_boundaries(
+        list(delta.get_state_history(migrated_cfg))
+    )
+    assert migrated_boundaries, "migrated thread has no settled boundaries"
+    for _, items in migrated_boundaries:
+        for it in items:
+            assert it.startswith("m"), (
+                f"migrated thread leaked item from other thread: {it}"
+            )
+
+    # Thread B: settled boundaries must only contain 'f' tags.
+    fresh_boundaries = _settled_boundaries(list(delta.get_state_history(fresh_cfg)))
+    assert fresh_boundaries, "fresh thread has no settled boundaries"
+    for _, items in fresh_boundaries:
+        for it in items:
+            assert it.startswith("f"), (
+                f"fresh thread leaked item from migrated thread: {it}"
+            )
+
+
+# ---------------------------------------------------------------------------
+# 6. Tip-of-pre-migration hydration: the latest checkpoint from a binop-run
+# thread has a real accumulated value in its own `channel_values["items"]`.
+# When hydrated under the delta-channel graph via `get_state(config)` with no
+# `checkpoint_id`, the short-circuit must use that value directly instead of
+# walking ancestors (which would skip the tip's own blob).
+# ---------------------------------------------------------------------------
+
+
+def test_tip_of_pre_migration_hydrates_directly() -> None:
+    """`graph.get_state(config)` at the latest (pre-migration) checkpoint
+    returns the full accumulated list stored in that checkpoint's own
+    `channel_values`. The hydration must not walk ancestors past it."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "tip-sync"}}
+
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 3)
+
+    binop_tip = binop.get_state(config)
+    expected_items = list(binop_tip.values.get("items", []))
+    assert expected_items == ["u0", "u1", "u2"], (
+        f"sanity: pre-migration tip should accumulate all 3 items, got {expected_items}"
+    )
+
+    delta = _delta_graph(checkpointer)
+
+    snap = delta.get_state(config)
+    assert list(snap.values.get("items", [])) == expected_items, (
+        f"tip hydration mismatch: expected {expected_items}, "
+        f"got {snap.values.get('items', [])}"
+    )
+
+
+async def test_tip_of_pre_migration_hydrates_directly_async() -> None:
+    """Async variant of the tip-of-pre-migration hydration scenario."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "tip-async"}}
+
+    binop = _binop_graph(checkpointer)
+    await _adrive(binop, config, "u", 3)
+
+    binop_tip = await binop.aget_state(config)
+    expected_items = list(binop_tip.values.get("items", []))
+    assert expected_items == ["u0", "u1", "u2"]
+
+    delta = _delta_graph(checkpointer)
+
+    snap = await delta.aget_state(config)
+    assert list(snap.values.get("items", [])) == expected_items, (
+        f"async tip hydration mismatch: expected {expected_items}, "
+        f"got {snap.values.get('items', [])}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# 7. `update_state` after migration writes a real value to the new
+# checkpoint's `channel_values` (not a sentinel). Hydration must use it
+# directly — the ancestor walk would skip this blob and return stale state.
+# ---------------------------------------------------------------------------
+
+
+def test_update_state_after_migration_uses_written_value() -> None:
+    """After migrating and running at least one post-migration super-step
+    (so the thread's tip has a `DELTA_SENTINEL`), `update_state` writes a
+    concrete value to a new checkpoint's `channel_values`. `get_state`
+    must reflect that concrete value."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "update-state"}}
+
+    # Pre-migration: accumulate a little state.
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 2)
+
+    # Migrate and run one more super-step so the tip is a post-migration
+    # checkpoint with `DELTA_SENTINEL` in its own `channel_values`.
+    delta = _delta_graph(checkpointer)
+    delta.invoke({"items": ["post"]}, config)
+
+    # `update_state` writes a concrete value into a new checkpoint's blob
+    # via the reducer against the hydrated prior state.
+    delta.update_state(config, {"items": ["x", "y"]})
+
+    snap = delta.get_state(config)
+    updated_items = list(snap.values.get("items", []))
+    # Must include the "x","y" update; without the hydration fix, the
+    # update_state-written blob would be skipped in favor of an ancestor
+    # walk, and the update values would disappear.
+    assert "x" in updated_items and "y" in updated_items, (
+        f"update_state values missing from snapshot: {updated_items}"
+    )
+    # The "x","y" items should be folded onto the prior accumulated state,
+    # not stand alone. This verifies the update-written blob is used
+    # directly by `get_state` (no ancestor walk past it).
+    assert len(updated_items) >= 4, (
+        f"update_state snapshot should preserve pre-update state, got {updated_items}"
+    )
+    assert updated_items[-2:] == ["x", "y"], (
+        f"update_state deltas should be at the tail, got {updated_items}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# 8. Fork from an `update_state` checkpoint: a new run branched off the
+# update_state-produced checkpoint must see that checkpoint's concrete
+# `channel_values` as its base, with new deltas folded on top.
+# ---------------------------------------------------------------------------
+
+
+def test_fork_from_update_state_checkpoint() -> None:
+    """Branching a new run from the checkpoint produced by `update_state`
+    must use that checkpoint's concrete blob as the base. Additional
+    deltas from the forked run fold onto it through the reducer."""
+
+    checkpointer = InMemorySaver()
+    config = {"configurable": {"thread_id": "fork"}}
+
+    # Pre-migration build-up, then migrate and add one post-migration step.
+    binop = _binop_graph(checkpointer)
+    _drive(binop, config, "u", 2)
+    delta = _delta_graph(checkpointer)
+    delta.invoke({"items": ["post"]}, config)
+
+    # Apply `update_state` and capture the returned config (references
+    # the new checkpoint produced by the update).
+    update_cfg = delta.update_state(config, {"items": ["x", "y"]})
+
+    update_snap = delta.get_state(update_cfg)
+    base_items = list(update_snap.values.get("items", []))
+    assert "x" in base_items and "y" in base_items, (
+        f"update_state values missing from snapshot: {base_items}"
+    )
+    assert base_items[-2:] == ["x", "y"], (
+        f"sanity: update_state deltas should be at the tail, got {base_items}"
+    )
+
+    # Fork: invoke from the update_state checkpoint with a new delta.
+    forked = delta.invoke({"items": ["fork0"]}, update_cfg)
+    forked_items = list(forked.get("items", []))
+    # The fork must see the update_state-written blob as its base (not
+    # walk past it), and the new delta must fold on top of it.
+    assert forked_items[: len(base_items)] == base_items, (
+        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/tests/test_time_travel.py

@@ -1161,9 +1161,7 @@ def test_subgraph_interrupt_resume_with_explicit_head_checkpoint_id(
     assert called == ["step_a", "ask_human"]
 
     # Resume with explicit head checkpoint_id in config
-    head_checkpoint_id = graph.get_state(config).config["configurable"][
-        "checkpoint_id"
-    ]
+    head_checkpoint_id = graph.get_state(config).config["configurable"]["checkpoint_id"]
     called.clear()
     resume_config = {
         "configurable": {

libs/langgraph/tests/test_utils.py

@@ -427,49 +427,3 @@ def test_callback_manager_copies_configurable_ids_to_tracing_metadata() -> None:
         "thread_id": "th-123",
         "user_id": "uid-1",
     }
-
-
-def test_get_nonlocal_names_cached_by_code_object() -> None:
-    """_get_nonlocal_names caches by code object so repeated calls are cheap."""
-    from langgraph.pregel._utils import _get_nonlocal_names
-
-    x = 1
-
-    def my_func() -> int:
-        return x
-
-    result1 = _get_nonlocal_names(my_func.__code__)
-    result2 = _get_nonlocal_names(my_func.__code__)
-
-    # Same frozenset instance returned (cache hit)
-    assert result1 is result2
-    assert "x" in result1
-
-
-def test_get_function_nonlocals_fast_path_no_freevars() -> None:
-    """Functions with no free variables return [] without AST parsing."""
-    from langgraph.pregel._utils import _get_nonlocal_names, get_function_nonlocals
-
-    cache_info_before = _get_nonlocal_names.cache_info()
-
-    def pure_func(a: int, b: int) -> int:
-        return a + b
-
-    result = get_function_nonlocals(pure_func)
-
-    # Should have returned early without touching the cache
-    assert result == []
-    assert _get_nonlocal_names.cache_info().misses == cache_info_before.misses
-
-
-def test_get_function_nonlocals_returns_closure_values() -> None:
-    """get_function_nonlocals correctly extracts values from closures."""
-    from langgraph.pregel._utils import get_function_nonlocals
-
-    sentinel = object()
-
-    def my_func() -> object:
-        return sentinel
-
-    result = get_function_nonlocals(my_func)
-    assert sentinel in result

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

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -1842,17 +1842,9 @@ def _get_injection_from_type(
     return None
 
 
-# Cache keyed by tool object identity. Stores (tool, result) to keep a strong
-# reference that prevents GC from reusing the id for a different object.
-_INJECTED_ARGS_CACHE: dict[int, tuple[BaseTool, _InjectedArgs]] = {}
-
-
 def _get_all_injected_args(tool: BaseTool) -> _InjectedArgs:
     """Extract all injected arguments from tool in a single pass.
 
-    Results are cached by tool identity so the expensive type-hint and schema
-    inspection only runs once per unique tool object across ToolNode instances.
-
     This function analyzes both the tool's input schema and function signature
     to identify all arguments that should be injected (state, store, runtime).
 
@@ -1862,11 +1854,6 @@ def _get_all_injected_args(tool: BaseTool) -> _InjectedArgs:
     Returns:
         _InjectedArgs structure containing all detected injections.
     """
-    tool_id = id(tool)
-    entry = _INJECTED_ARGS_CACHE.get(tool_id)
-    if entry is not None and entry[0] is tool:
-        return entry[1]
-
     # Get annotations from both schema and function signature
     full_schema = tool.get_input_schema()
     schema_annotations = get_all_basemodel_annotations(full_schema)
@@ -1912,12 +1899,10 @@ def _get_all_injected_args(tool: BaseTool) -> _InjectedArgs:
         if _get_injection_from_type(type_, ToolRuntime):
             runtime_arg = name
 
-    result = _InjectedArgs(
+    return _InjectedArgs(
         state=state_args,
         store=store_arg,
         runtime=runtime_arg,
         all_injected_keys=all_injected_keys,
         _optional_state_args=_optional_state_args,
     )
-    _INJECTED_ARGS_CACHE[tool_id] = (tool, result)
-    return result

pr_delta_channel.md

@@ -1,133 +0,0 @@
-# feat(channels): DeltaChannel — O(N) incremental checkpoint storage
-
-## The problem
-
-LangGraph checkpoints store the **full accumulated value** of every channel on every step. For a `messages` channel backed by `add_messages`, that means each checkpoint blob contains the entire conversation history up to that point.
-
-Storage cost grows **O(N²)** in the number of turns:
-
-| Step | Checkpoint blob |
-|------|----------------|
-| 1    | [msg_1] |
-| 2    | [msg_1, msg_2] |
-| N    | [msg_1, ..., msg_N] |
-
-At 100K tokens of conversation data, a single thread accumulates ~250 MB; with large messages or file attachments costs scale even faster.
-
-## The fix: `DeltaChannel`
-
-`DeltaChannel` is an opt-in wrapper around any binary reducer that stores only a **sentinel marker** in `checkpoint_blobs` rather than the full accumulated value. The actual per-step writes stay in `checkpoint_writes` (which every checkpointer already writes unconditionally). At read time the saver walks the ancestor chain, collects all writes for the channel, and replays them through the reducer.
-
-Storage scales **O(N)** — the sentinel blob is effectively zero bytes, and the writes table already exists.
-
-```python
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph.message import add_messages
-
-class State(TypedDict):
-    # Before: O(N²) storage
-    messages: Annotated[list[AnyMessage], add_messages]
-
-    # After: O(N) storage
-    messages: Annotated[list[AnyMessage], DeltaChannel(add_messages)]
-```
-
-## Benchmarks
-
-Simulated with realistic paragraph-length messages (~100 tokens each, ~400 chars). Each turn = one human + one AI message (~200 tokens total).
-
-### Storage (InMemorySaver)
-
-| turns | ctx | add_msgs | delta | savings |
-|------:|----:|---------:|------:|--------:|
-| 10 | ~2K tok | 108.6 KB | 4.0 KB | 27x |
-| 25 | ~5K tok | 649.0 KB | 10.1 KB | 64x |
-| 50 | ~10K tok | 2.6 MB | 20.2 KB | 126x |
-| 100 | ~20K tok | 10.2 MB | 40.5 KB | 251x |
-| 500 | ~100K tok | 252.6 MB | 202.8 KB | 1245x |
-
-Savings grow with N because `add_messages` is O(N²) while `DeltaChannel` is O(N). The sentinel blob itself is essentially zero bytes.
-
-### Read latency (avg of 5 `get_state` calls = cost per `invoke`)
-
-| turns | ctx | add_msgs | delta |
-|------:|----:|---------:|------:|
-| 10 | ~2K tok | 0.1ms | 0.2ms |
-| 25 | ~5K tok | 0.2ms | 0.5ms |
-| 50 | ~10K tok | 0.4ms | 1.5ms |
-| 100 | ~20K tok | 0.7ms | 4.8ms |
-| 500 | ~100K tok | 5.8ms | 114.9ms |
-
-**This cost is paid once per `invoke`/`stream` call, not per node.** Within a single invocation, all channels are loaded into memory once at the start and shared across every node — there is no per-node reconstruction. The 114.9ms at 500 turns is what you pay each time a user sends a new message, not on each step of the graph.
-
-## How it works
-
-**Write:** `DeltaChannel.checkpoint()` always emits `DeltaChannelSentinel()` — a tiny marker (zero payload bytes) stored in `checkpoint_blobs`. Per-step writes flow into `checkpoint_writes` as they normally do for every channel.
-
-**Read:** The saver detects `DeltaChannelSentinel` values in `channel_values` and replaces them by calling `get_channel_writes` / `aget_channel_writes`, which walks the ancestor checkpoint chain and collects all writes for that channel (oldest→newest). `DeltaChannel.from_checkpoint()` replays those writes through the operator to reconstruct the full value.
-
-**Saver implementations:**
-- `InMemorySaver` — direct dict traversal of `self.storage` and `self.writes`, no I/O
-- `PostgresSaver` (sync + async) — two queries: one cheap ID walk across the thread, one `ANY()` fetch of writes; no recursive CTE
-- All other savers — `BaseCheckpointSaver.get_channel_writes` fallback via `list()`, with a re-entrancy guard to prevent infinite recursion
-
-## Changes
-
-**`libs/checkpoint`**
-- `base/__init__.py` — add `DeltaChannelSentinel` marker dataclass; add `get_channel_writes` / `aget_channel_writes` to `BaseCheckpointSaver` with a `list()`-based fallback and re-entrancy guard
-
-**`libs/checkpoint/memory`**
-- `memory/__init__.py` — `get_channel_writes` via direct dict traversal; `_resolve_delta_channels` helper called in `get_tuple` / `aget_tuple` to replace sentinels with reconstructed write lists
-
-**`libs/langgraph`**
-- `channels/delta.py` — `DeltaChannel` implementation: `checkpoint()` always emits sentinel, `from_checkpoint()` replays writes list
-- `channels/__init__.py` — export `DeltaChannel`
-- `graph/state.py` — recognize `DeltaChannel` as a valid channel annotation
-- `pregel/_checkpoint.py` / `pregel/_loop.py` — wire `after_checkpoint` hook; call it after each checkpointing step so `DeltaChannel` can advance internal state
-
-**`libs/checkpoint-postgres`**
-- `postgres/base.py` — `_get_channel_writes_cur` two-query ancestor walk (sync); `_resolve_delta_channels` called after `_load_blobs`
-- `postgres/aio.py` — `_aget_channel_writes_cur` (async counterpart)
-
-## Open questions
-
-**Should we add a compile-time capability check?**
-
-Currently misconfiguring `DeltaChannel` with an unsupported saver only errors at runtime on first reload. A protocol-based check at `compile()` time would give an early warning without requiring a manual boolean flag.
-
-**`snapshot_every` for bounded reconstruction cost?**
-
-Both per-invoke read latency and total write wall time grow O(N) per invoke / O(N²) total as the conversation lengthens. A `snapshot_every` parameter — periodically store a full snapshot in `checkpoint_blobs` to cap chain depth — would bound reconstruction cost and is a natural follow-up once the core design is stable.
-
-## Backwards compatibility
-
-| Scenario | Behaviour |
-|----------|-----------|
-| Existing graph using `add_messages` | Unaffected — no code or schema changes |
-| `DeltaChannel` loading an old full-list checkpoint blob | Handled via backwards-compat path in `from_checkpoint` |
-| `DeltaChannel` with `InMemorySaver` or `PostgresSaver` | Fully supported |
-| Time-travel to a past checkpoint | Ancestor walk uses the version at that checkpoint — correct by construction |
-| `Overwrite` value | Resets the effective chain; reconstruction starts from that step |
-
-## Test plan
-
-- [x] `DeltaChannel` unit tests: `update` → `checkpoint` lifecycle, `from_checkpoint` chain replay, backwards-compat with plain list, `Overwrite` resets chain
-- [x] `InMemorySaver` `get_channel_writes`: assembles write list from dict storage
-- [x] Serde round-trip for `DeltaChannelSentinel`
-- [x] End-to-end graph tests: multi-turn conversations accumulate correctly, time-travel reconstructs correct partial history
-- [x] `PostgresSaver` two-query chain reconstruction (sync + async)
-- [x] `BaseCheckpointSaver` fallback path via `list()` with re-entrancy guard
-- [x] Storage benchmark: `DeltaChannel` uses strictly less storage than `add_messages` at all measured turn counts
-
----
-
-## Changes from previous base branch
-
-The previous version stored `DeltaValue` objects (containing the per-step writes) directly in `checkpoint_blobs` and used a `DeltaChainValue` to represent the assembled chain. Reconstruction required a dedicated `get_delta_chain` / `aget_delta_chain` protocol and a recursive CTE in Postgres.
-
-This version pivots to a simpler design:
-- **Sentinel in blobs, writes in `checkpoint_writes`** — `checkpoint_blobs` stores only a zero-byte `DeltaChannelSentinel` marker. The actual per-step data already lives in `checkpoint_writes` (written unconditionally by every checkpointer), so blob storage is essentially free. This is why storage savings jump to 1245x at 500 turns.
-- **No custom serde type for the delta payload** — `DeltaValue` / `DeltaChainValue` and the `"delta"` serde type tag are gone. Writes are deserialized with the same serde path they were originally written with.
-- **Postgres: two queries instead of a recursive CTE** — fetch all `(checkpoint_id, parent_checkpoint_id)` pairs for the thread, walk the ancestor chain in Python, then fetch writes with a plain `ANY()` filter.
-- **Universal fallback on `BaseCheckpointSaver`** — the base class now provides `get_channel_writes` via `list()`, so any third-party saver works without modification.
-- **`snapshot_every` removed** — deferred as a follow-up; the simpler design is easier to reason about and delivers larger storage savings.