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

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

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

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

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

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

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

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

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

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

libs/checkpoint-postgres/pyproject.toml

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

libs/checkpoint-postgres/tests/test_async.py

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

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

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

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

@@ -9,7 +9,7 @@ from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from types import TracebackType
-from typing import Any
+from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 
@@ -255,13 +255,16 @@ class InMemorySaver(
                 checkpoint, metadata, parent_checkpoint_id = saved
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                 checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
+                channel_values = self._load_blobs(
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint_["channel_versions"],
+                )
                 return CheckpointTuple(
                     config=config,
                     checkpoint={
                         **checkpoint_,
-                        "channel_values": self._load_blobs(
-                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
-                        ),
+                        "channel_values": channel_values,
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -285,19 +288,26 @@ class InMemorySaver(
                 checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                 writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
                 checkpoint_ = self.serde.loads_typed(checkpoint)
-                return CheckpointTuple(
-                    config={
+                resolved_config = cast(
+                    RunnableConfig,
+                    {
                         "configurable": {
                             "thread_id": thread_id,
                             "checkpoint_ns": checkpoint_ns,
                             "checkpoint_id": checkpoint_id,
                         }
                     },
+                )
+                channel_values = self._load_blobs(
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint_["channel_versions"],
+                )
+                return CheckpointTuple(
+                    config=resolved_config,
                     checkpoint={
                         **checkpoint_,
-                        "channel_values": self._load_blobs(
-                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
-                        ),
+                        "channel_values": channel_values,
                     },
                     metadata=self.serde.loads_typed(metadata),
                     pending_writes=[
@@ -392,21 +402,27 @@ class InMemorySaver(
 
                     checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
 
-                    yield CheckpointTuple(
-                        config={
+                    list_config = cast(
+                        RunnableConfig,
+                        {
                             "configurable": {
                                 "thread_id": thread_id,
                                 "checkpoint_ns": checkpoint_ns,
                                 "checkpoint_id": checkpoint_id,
                             }
                         },
+                    )
+                    channel_values = self._load_blobs(
+                        thread_id,
+                        checkpoint_ns,
+                        checkpoint_["channel_versions"],
+                    )
+
+                    yield CheckpointTuple(
+                        config=list_config,
                         checkpoint={
                             **checkpoint_,
-                            "channel_values": self._load_blobs(
-                                thread_id,
-                                checkpoint_ns,
-                                checkpoint_["channel_versions"],
-                            ),
+                            "channel_values": channel_values,
                         },
                         metadata=metadata,
                         parent_config=(
@@ -509,6 +525,101 @@ class InMemorySaver(
                 task_path,
             )
 
+    def prune(
+        self,
+        thread_ids: Sequence[str],
+        *,
+        strategy: str = "keep_latest",
+    ) -> None:
+        """Prune checkpoints for the given threads.
+
+        For DeltaChannel channels, a checkpoint is only deleted if the walk
+        from the latest checkpoint would not need to traverse it — i.e., a
+        `_DeltaSnapshot` blob exists in the kept ancestry that covers all
+        sentinel channels.  Checkpoints that are still in the active walk
+        chain (because no snapshot has been taken yet, e.g. with
+        `snapshot_frequency=None`) are retained.
+
+        Args:
+            thread_ids: Thread IDs to prune.
+            strategy: ``"keep_latest"`` keeps only the most recent checkpoint
+                per namespace; ``"delete"`` removes all checkpoints.
+        """
+        for thread_id in thread_ids:
+            if strategy == "delete":
+                self.delete_thread(thread_id)
+                continue
+
+            if strategy != "keep_latest":
+                raise ValueError(
+                    f"Unknown pruning strategy {strategy!r}. "
+                    "Expected 'keep_latest' or 'delete'."
+                )
+
+            for checkpoint_ns, ns_storage in list(
+                self.storage.get(thread_id, {}).items()
+            ):
+                if not ns_storage:
+                    continue
+
+                # Latest checkpoint (uuid6 IDs are lexicographically monotonic)
+                latest_id = max(ns_storage.keys())
+                latest_data, _, _ = ns_storage[latest_id]
+                latest_cp = self.serde.loads_typed(latest_data)
+
+                # Which channels in the latest checkpoint still have sentinels?
+                sentinel_channels: set[str] = set()
+                for ch, ver in latest_cp.get("channel_versions", {}).items():
+                    blob = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                    if blob is not None and blob[0] != "empty":
+                        if self.serde.loads_typed(blob) is DELTA_SENTINEL:
+                            sentinel_channels.add(ch)
+
+                # Walk the parent chain to find the oldest ancestor still needed.
+                # We stop (and mark "safe to prune before here") when all
+                # sentinel channels are covered by a non-sentinel blob.
+                required_ids: set[str] = {latest_id}
+                if sentinel_channels:
+                    _, _, parent_id = ns_storage[latest_id]
+                    remaining = set(sentinel_channels)
+                    while parent_id is not None and remaining:
+                        entry = ns_storage.get(parent_id)
+                        if entry is None:
+                            break
+                        required_ids.add(parent_id)
+                        cp_data, _, grandparent_id = entry
+                        cp = self.serde.loads_typed(cp_data)
+                        resolved: set[str] = set()
+                        for ch in remaining:
+                            ver = cp.get("channel_versions", {}).get(ch)
+                            if ver is None:
+                                continue
+                            blob = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                            if blob is not None and blob[0] != "empty":
+                                if self.serde.loads_typed(blob) is not DELTA_SENTINEL:
+                                    resolved.add(ch)
+                        remaining -= resolved
+                        parent_id = grandparent_id
+
+                # Delete everything outside the required set
+                for cp_id in list(ns_storage.keys()):
+                    if cp_id in required_ids:
+                        continue
+                    cp_data, _, _ = ns_storage.pop(cp_id)
+                    self.writes.pop((thread_id, checkpoint_ns, cp_id), None)
+
+                # Clean up blobs no longer referenced by any kept checkpoint
+                live: set[tuple[str, str, str, Any]] = set()
+                for cp_data, _, _ in ns_storage.values():
+                    cp = self.serde.loads_typed(cp_data)
+                    for ch, ver in cp.get("channel_versions", {}).items():
+                        live.add((thread_id, checkpoint_ns, ch, ver))
+                for key in [
+                    k for k in self.blobs if k[:2] == (thread_id, checkpoint_ns)
+                ]:
+                    if key not in live:
+                        del self.blobs[key]
+
     def delete_thread(self, thread_id: str) -> None:
         """Delete all checkpoints and writes associated with a thread ID.
 

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

@@ -302,10 +302,10 @@ EXT_DELTA_SENTINEL = 8
 
 
 def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
-    if isinstance(obj, _DeltaSnapshot):
-        return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
-    elif isinstance(obj, _DeltaSentinel):
+    if isinstance(obj, _DeltaSentinel):
         return ormsgpack.Ext(EXT_DELTA_SENTINEL, b"")
+    elif isinstance(obj, _DeltaSnapshot):
+        return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
     elif hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_jsonplus.py

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

libs/checkpoint/tests/test_memory.py

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

libs/checkpoint/uv.lock

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

libs/langgraph/langgraph/channels/binop.py

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

libs/langgraph/langgraph/channels/delta.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import collections.abc
 import copy as _copy
 from collections.abc import Callable, Sequence
 from typing import Any, Generic
@@ -9,81 +8,78 @@ from langgraph.checkpoint.base import DELTA_SENTINEL, PendingWrite
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import Self
 
+from langgraph._internal._constants import OVERWRITE
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel, Value
-from langgraph.channels.binop import _get_overwrite, _operators_equal, _strip_extras
 from langgraph.errors import (
     EmptyChannelError,
     ErrorCode,
     InvalidUpdateError,
     create_error_message,
 )
+from langgraph.types import Overwrite
 
 __all__ = ("DeltaChannel",)
 
 
+def _empty(typ: Any) -> Any:
+    try:
+        return typ()
+    except Exception:
+        return []
+
+
+def _get_overwrite(value: Any) -> tuple[bool, Any]:
+    if isinstance(value, Overwrite):
+        return True, value.value
+    if isinstance(value, dict) and set(value.keys()) == {OVERWRITE}:
+        return True, value[OVERWRITE]
+    return False, None
+
+
 class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
-    """Reducer channel that stores only a sentinel in checkpoint blobs and
-    reconstructs state by replaying ancestor writes through the reducer.
+    """Fold-reducer channel with configurable snapshot cadence.
 
-    The reducer receives the current accumulated value and a batch of writes
-    in one call: `reducer(state, [write1, write2, ...]) -> new_state`.
-
-    Reducers must be deterministic and batching-invariant (associative across
-    folds): applying two consecutive write batches separately must produce the
-    same state as applying their concatenation once:
-
-        reducer(reducer(state, xs), ys) == reducer(state, xs + ys)
-
-    This lets LangGraph replay checkpointed writes in larger batches than they
-    were originally produced without changing reconstructed state.
-
-    `snapshot_frequency=None` (default): pure delta; stores only
+    `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.
+    `snapshot_frequency=N`: pregel's `create_checkpoint` writes a full
+    `_DeltaSnapshot` blob every N steps (eagerly, even if the channel had
+    no write that step). Reads walk at most N ancestor checkpoints before
+    hitting the snapshot, bounding replay depth to N regardless of thread
+    length.
 
     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(...)]`.
+        operator: Binary reducer `(Value, Value) -> Value`.
         snapshot_frequency: Every Nth pregel step writes a snapshot blob.
             `None` (default) = pure delta, never snapshot.
     """
 
-    __slots__ = ("value", "reducer", "snapshot_frequency")
+    __slots__ = ("value", "operator", "snapshot_frequency")
     value: Value | Any
 
     def __init__(
         self,
-        reducer: Callable[[Any, Sequence[Any]], Any],
-        typ: type[Value] | None = None,
+        operator: Callable[[Any, Any], Any],
         *,
         snapshot_frequency: int | None = None,
     ) -> None:
-        if typ is None:
-            typ = list  # type: ignore[assignment]  # placeholder; overridden by _is_field_channel
-        super().__init__(typ)
-        self.reducer = reducer
+        super().__init__(list)
+        self.operator = operator
         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
+        self.value: Any = []
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, DeltaChannel):
             return False
         if self.snapshot_frequency != other.snapshot_frequency:
             return False
-        return _operators_equal(self.reducer, other.reducer)
+        if (
+            self.operator.__name__ != "<lambda>"
+            and other.operator.__name__ != "<lambda>"
+        ):
+            return self.operator is other.operator
+        return True
 
     @property
     def ValueType(self) -> Any:
@@ -96,19 +92,34 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
     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
+            self.snapshot_frequency is not None and step % self.snapshot_frequency == 0
         )
 
-    def copy(self) -> Self:
-        new = self.__class__(
-            self.reducer, self.typ, snapshot_frequency=self.snapshot_frequency
-        )
+    def _clone_empty(self) -> Self:
+        new = self.__class__.__new__(self.__class__)
+        new.typ = self.typ
         new.key = self.key
+        new.operator = self.operator
+        new.snapshot_frequency = self.snapshot_frequency
+        new.value = MISSING
+        return new
+
+    def copy(self) -> Self:
+        new = self._clone_empty()
         new.value = self.value if self.value is MISSING else _copy.copy(self.value)
         return new
 
+    def _apply_write(self, value: Any, write: Any) -> Any:
+        is_overwrite, overwrite_value = _get_overwrite(write)
+        if is_overwrite:
+            return (
+                _copy.copy(overwrite_value)
+                if overwrite_value is not None
+                else _empty(self.typ)
+            )
+        base = _empty(self.typ) if value is MISSING else value
+        return self.operator(base, write)
+
     def from_checkpoint(self, checkpoint: Any) -> Self:
         """Initialize from a stored blob or sentinel.
 
@@ -117,12 +128,9 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
           * `_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
+        new = self._clone_empty()
         if checkpoint is MISSING or checkpoint is DELTA_SENTINEL:
-            new.value = self.typ()
+            new.value = _empty(new.typ)
         elif isinstance(checkpoint, _DeltaSnapshot):
             new.value = checkpoint.value
         else:
@@ -130,51 +138,27 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
         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
+        """Fold ancestor writes oldest→newest into current value."""
+        for _, _, value in writes:
+            self.value = self._apply_write(self.value, value)
 
     def update(self, values: Sequence[Any]) -> bool:
         if not values:
             return False
-        overwrite_idx: int | None = None
-        for i, v in enumerate(values):
-            is_ow, _ = _get_overwrite(v)
-            if is_ow:
-                if overwrite_idx is not None:
+        seen_overwrite = False
+        for value in values:
+            is_overwrite, _ = _get_overwrite(value)
+            if is_overwrite:
+                if seen_overwrite:
                     msg = create_error_message(
                         message="Can receive only one Overwrite value per super-step.",
                         error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
                     )
                     raise InvalidUpdateError(msg)
-                overwrite_idx = i
-        if overwrite_idx is not None:
-            _, overwrite_value = _get_overwrite(values[overwrite_idx])
-            base = (
-                _copy.copy(overwrite_value)
-                if overwrite_value is not None
-                else self.typ()
-            )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
-            self.value = self.reducer(base, remaining) if remaining else base
-            return True
-        base = self.typ() if self.value is MISSING else self.value
-        self.value = self.reducer(base, list(values))
+                seen_overwrite = True
+            elif seen_overwrite:
+                continue
+            self.value = self._apply_write(self.value, value)
         return True
 
     def get(self) -> Any:

libs/langgraph/langgraph/graph/message.py

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

libs/langgraph/langgraph/pregel/_algo.py

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

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
-from typing import Any, cast
+from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import DELTA_SENTINEL, BaseCheckpointSaver, Checkpoint
@@ -38,7 +38,6 @@ 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.
 
@@ -48,10 +47,6 @@ def create_checkpoint(
     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:
@@ -66,7 +61,7 @@ def create_checkpoint(
             ch = channels[k]
             if (
                 isinstance(ch, DeltaChannel)
-                and (force_delta_snapshot or ch.is_snapshot_step(step))
+                and ch.is_snapshot_step(step)
                 and ch.is_available()
             ):
                 # Eager snapshot: bump version if not already written this step
@@ -131,9 +126,9 @@ def channels_from_checkpoint(
         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)
+            assert isinstance(spec, DeltaChannel)
             history = saver._get_channel_writes_history(config, k)
-            replay_ch = delta_spec.from_checkpoint(history.seed)
+            replay_ch = spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:
@@ -163,9 +158,9 @@ async def achannels_from_checkpoint(
         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)
+            assert isinstance(spec, DeltaChannel)
             history = await saver._aget_channel_writes_history(config, k)
-            replay_ch = delta_spec.from_checkpoint(history.seed)
+            replay_ch = spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:

libs/langgraph/langgraph/pregel/_loop.py

@@ -25,6 +25,7 @@ from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
@@ -68,7 +69,6 @@ from langgraph.callbacks import (
     GraphResumeEvent,
 )
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.constants import TAG_HIDDEN
 from langgraph.errors import (
@@ -190,8 +190,6 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
-    # Only set on AsyncPregelLoop; sync loops keep this as None.
-    _delta_write_futs: list[Any] | None = None
     managed: ManagedValueMapping
     checkpoint: Checkpoint
     checkpoint_id_saved: str
@@ -201,6 +199,7 @@ class PregelLoop:
     checkpoint_pending_writes: list[PendingWrite]
     checkpoint_previous_versions: dict[str, str | float | int]
     prev_checkpoint_config: RunnableConfig | None
+    _pending_write_futs: list[concurrent.futures.Future]
 
     status: Literal[
         "input",
@@ -424,10 +423,7 @@ class PregelLoop:
                     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)
+            self._pending_write_futs.append(fut)
         # output writes
         if hasattr(self, "tasks"):
             self.output_writes(task_id, writes)
@@ -901,7 +897,6 @@ class PregelLoop:
             get_next_version=self.checkpointer_get_next_version
             if do_checkpoint
             else None,
-            force_delta_snapshot=exiting and self.durability == "exit",
         )
         # sanitize TASK channel in the checkpoint before saving (durability=="exit")
         if TASKS in self.checkpoint["channel_values"] and any(
@@ -939,6 +934,17 @@ class PregelLoop:
             )
             self.checkpoint_previous_versions = channel_versions
 
+            # If the checkpoint has any DELTA_SENTINEL blobs, the sentinel is
+            # only meaningful if checkpoint_writes are durable first.  Flush
+            # pending write futures synchronously before committing the blob so
+            # we never end up with a sentinel blob backed by missing writes.
+            if self._pending_write_futs and any(
+                v is DELTA_SENTINEL for v in self.checkpoint["channel_values"].values()
+            ):
+                for fut in self._pending_write_futs:
+                    fut.result()
+                self._pending_write_futs.clear()
+
             # save it, without blocking
             # if there's a previous checkpoint save in progress, wait for it
             # ensuring checkpointers receive checkpoints in order
@@ -1283,6 +1289,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
+        self._pending_write_futs = []
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs,
@@ -1383,11 +1390,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         metadata: CheckpointMetadata,
         new_versions: ChannelVersions,
     ) -> RunnableConfig:
-        # Drain DeltaChannel write futures before committing the checkpoint so
-        # DELTA_SENTINEL blobs are never saved ahead of their backing writes.
-        if self._delta_write_futs:
-            futs, self._delta_write_futs = self._delta_write_futs, []
-            await asyncio.gather(*futs)
         try:
             if prev is not None:
                 await prev
@@ -1493,7 +1495,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
-        self._delta_write_futs = []
+        self._pending_write_futs = []
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_channels.py

@@ -1,13 +1,9 @@
 import operator
 from collections.abc import Sequence
-from typing import Annotated
 
 import pytest
-from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import AIMessage, HumanMessage
 from langgraph.checkpoint.base import DELTA_SENTINEL
-from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-from typing_extensions import NotRequired, TypedDict
 
 from langgraph._internal._typing import MISSING
 from langgraph.channels.binop import BinaryOperatorAggregate
@@ -16,19 +12,11 @@ from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
-from langgraph.graph import START, StateGraph
-from langgraph.graph.message import _messages_delta_reducer
-from langgraph.graph.state import _get_channel
-from langgraph.types import Overwrite
+from langgraph.graph.message import add_messages
 
 pytestmark = pytest.mark.anyio
 
 
-# ---------------------------------------------------------------------------
-# Core channel primitives
-# ---------------------------------------------------------------------------
-
-
 def test_last_value() -> None:
     channel = LastValue(int).from_checkpoint(MISSING)
     assert channel.ValueType is int
@@ -111,41 +99,49 @@ def test_untracked_value() -> None:
     assert channel.ValueType is dict
     assert channel.UpdateType is dict
 
+    # UntrackedValue should start empty
     with pytest.raises(EmptyChannelError):
         channel.get()
 
+    # Should be able to update with a value
     test_data = {"session": "test", "temp": "dir"}
     channel.update([test_data])
     assert channel.get() == test_data
 
+    # Update with new value
     new_data = {"session": "updated", "temp": "newdir"}
     channel.update([new_data])
     assert channel.get() == new_data
 
+    # On checkpoint, UntrackedValue should return MISSING
     checkpoint = channel.checkpoint()
     assert checkpoint is MISSING
 
+    # Creating from checkpoint with MISSING should start empty
     new_channel = UntrackedValue(dict).from_checkpoint(checkpoint)
     with pytest.raises(EmptyChannelError):
         new_channel.get()
 
 
-# ---------------------------------------------------------------------------
-# DeltaChannel — message reducer
-# ---------------------------------------------------------------------------
-
-
 def test_delta_channel_basic_two_steps() -> None:
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.base import DELTA_SENTINEL
 
+    from langgraph.graph.message import add_messages
+
+    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
+
+    # Step 1: one message added
     ch.update([HumanMessage(content="hi", id="h1")])
     d1 = ch.checkpoint()
     assert d1 is DELTA_SENTINEL
 
+    # Step 2: another message
     ch.update([AIMessage(content="hello", id="a1")])
     d2 = ch.checkpoint()
     assert d2 is DELTA_SENTINEL
 
+    # Full accumulated value is preserved in memory
     assert len(ch.get()) == 2
     assert ch.get()[0].content == "hi"
     assert ch.get()[1].content == "hello"
@@ -153,7 +149,11 @@ def test_delta_channel_basic_two_steps() -> None:
 
 def test_delta_channel_from_checkpoint_writes_list() -> None:
     """replay_writes on a fresh channel replays through the operator."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import AIMessage, HumanMessage
+
+    from langgraph.graph.message import add_messages
+
+    spec = DeltaChannel(add_messages)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
     ch.replay_writes(
         [
@@ -170,28 +170,45 @@ def test_delta_channel_from_checkpoint_writes_list() -> None:
 
 
 def test_delta_channel_from_checkpoint_backwards_compat() -> None:
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import HumanMessage
+
+    from langgraph.graph.message import add_messages
+
+    # Old BinaryOperatorAggregate checkpoint: plain list treated as backward compat
+    spec = DeltaChannel(add_messages)
     old_value = [HumanMessage(content="old", id="h1")]
     ch = spec.from_checkpoint(old_value)
     assert ch.get() == old_value
 
 
 def test_delta_channel_overwrite() -> None:
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    from langchain_core.messages import HumanMessage
+    from langgraph.checkpoint.base import DELTA_SENTINEL
+
+    from langgraph.graph.message import add_messages
+    from langgraph.types import Overwrite
+
+    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
     ch.update([HumanMessage(content="old", id="h1")])
 
     ch.update([Overwrite([HumanMessage(content="new", id="h2")])])
     d = ch.checkpoint()
     assert d is DELTA_SENTINEL
+    # After overwrite, value is reset to only the new message
     assert len(ch.get()) == 1
     assert ch.get()[0].content == "new"
 
 
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+
+    from langgraph.graph.message import add_messages
+
+    spec = DeltaChannel(add_messages)
     ch = spec.from_checkpoint(MISSING)
 
+    # Step 1: add two messages
     ch.update([HumanMessage(content="hi", id="h1")])
     ch.update([AIMessage(content="hello", id="a1")])
     assert ch.get() == [
@@ -199,9 +216,11 @@ def test_delta_channel_remove_message_and_replay() -> None:
         AIMessage(content="hello", id="a1"),
     ]
 
+    # Step 2: remove the AI message
     ch.update([RemoveMessage(id="a1")])
     assert ch.get() == [HumanMessage(content="hi", id="h1")]
 
+    # Replay the writes list from scratch — must reproduce the post-remove state
     ch2 = spec.from_checkpoint(DELTA_SENTINEL)
     ch2.replay_writes(
         [
@@ -215,13 +234,21 @@ def test_delta_channel_remove_message_and_replay() -> None:
 
 def test_delta_channel_update_by_id_and_replay() -> None:
     """Updating a message by ID must round-trip correctly through writes replay."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    from langchain_core.messages import HumanMessage
+
+    from langgraph.graph.message import add_messages
+
+    spec = DeltaChannel(add_messages)
     ch = spec.from_checkpoint(MISSING)
 
+    # Step 1: add a message
     ch.update([HumanMessage(content="original", id="h1")])
+
+    # Step 2: update the same message by ID
     ch.update([HumanMessage(content="updated", id="h1")])
     assert ch.get() == [HumanMessage(content="updated", id="h1")]
 
+    # Replay writes — must produce the updated message, not the original
     ch2 = spec.from_checkpoint(DELTA_SENTINEL)
     ch2.replay_writes(
         [
@@ -235,18 +262,19 @@ def test_delta_channel_update_by_id_and_replay() -> None:
 
 def test_delta_channel_checkpoint_returns_sentinel() -> None:
     """checkpoint() always returns DELTA_SENTINEL regardless of state."""
-    ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)
+    from langgraph.checkpoint.base import DELTA_SENTINEL
+
+    from langgraph.graph.message import add_messages
+
+    ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
     assert ch.checkpoint() is DELTA_SENTINEL
 
+    from langchain_core.messages import HumanMessage
+
     ch.update([HumanMessage(content="hi", id="h1")])
     assert ch.checkpoint() is DELTA_SENTINEL
 
 
-# ---------------------------------------------------------------------------
-# DeltaChannel — snapshot frequency
-# ---------------------------------------------------------------------------
-
-
 def test_delta_channel_snapshot_step_based() -> None:
     """Snapshots fire on every Nth step regardless of whether the channel was written.
 
@@ -254,18 +282,28 @@ def test_delta_channel_snapshot_step_based() -> None:
     blob — even if the channel had no write that step (eager snapshot). This
     bounds the ancestor walk to at most N steps on any read.
     """
+    from typing import Annotated
 
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+    from typing_extensions import TypedDict
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
+
+    # snapshot_frequency=5: snapshot every 5 pregel steps
     class State(TypedDict):
-        messages: Annotated[
-            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=5)
-        ]
+        messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=5)]
         other: str
 
     def node_a(state: State) -> dict:
+        # writes to messages
         i = len(state["messages"]) // 2
         return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
 
     def node_b(state: State) -> dict:
+        # writes ONLY to other, not messages — snapshot must still fire at step N
         return {"other": "y"}
 
     g = StateGraph(State)
@@ -283,6 +321,7 @@ def test_delta_channel_snapshot_step_based() -> None:
             config,
         )
 
+    # Confirm at least one snapshot blob exists for messages
     msg_blob_values = [
         saver.serde.loads_typed((type_tag, blob))
         for k, (type_tag, blob) in saver.blobs.items()
@@ -291,6 +330,7 @@ def test_delta_channel_snapshot_step_based() -> None:
     snapshots = [v for v in msg_blob_values if isinstance(v, _DeltaSnapshot)]
     assert snapshots, "expected at least one _DeltaSnapshot blob for messages"
 
+    # Final state must be correct regardless of snapshot cadence
     state = graph.get_state(config)
     assert len(state.values["messages"]) == 12  # 6 human + 6 AI
 
@@ -299,11 +339,18 @@ def test_delta_channel_snapshot_fires_even_when_not_written() -> None:
     """Eager snapshot: _DeltaSnapshot stored at snapshot step even when the
     channel had no write that step (node_b doesn't touch messages).
     """
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+    from typing_extensions import TypedDict
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[
-            list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=3)
-        ]
+        messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=3)]
         tick: int
 
     def writer(state: State) -> dict:
@@ -311,6 +358,7 @@ def test_delta_channel_snapshot_fires_even_when_not_written() -> None:
         return {"messages": [AIMessage(content=f"a{i}", id=f"a{i}")]}
 
     def ticker(state: State) -> dict:
+        # never writes messages
         return {"tick": state["tick"] + 1}
 
     g = StateGraph(State)
@@ -328,30 +376,36 @@ def test_delta_channel_snapshot_fires_even_when_not_written() -> None:
             config,
         )
 
+    # Count distinct message channel blob versions
     msg_blobs = {
         k: saver.serde.loads_typed((t, b))
         for k, (t, b) in saver.blobs.items()
         if k[2] == "messages" and t == "msgpack" and b
     }
     snapshots = {k: v for k, v in msg_blobs.items() if isinstance(v, _DeltaSnapshot)}
+    # There must be snapshots (ticker steps are snapshot steps too)
     assert snapshots, (
         "eager snapshots must fire even on steps where messages wasn't written"
     )
 
+    # All get_state calls must return the correct accumulated value
     state = graph.get_state(config)
     assert len(state.values["messages"]) == 10  # 5 human + 5 AI
 
 
-# ---------------------------------------------------------------------------
-# 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.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer, list)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     n = {"v": 0}
 
@@ -369,6 +423,9 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     graph.invoke({"messages": [HumanMessage(content="hi", id="h1")]}, config)
     graph.invoke({"messages": [HumanMessage(content="bye", id="h2")]}, config)
 
+    # get_tuple returns raw storage shape — channel_values stores DELTA_SENTINEL
+    # for delta channels; the reconstructed writes flow separately via
+    # saver._get_channel_writes_history.
     saved = saver.get_tuple(config)
     assert saved is not None
     assert "messages" in saved.checkpoint["channel_values"]
@@ -379,23 +436,25 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
 
 
 # ---------------------------------------------------------------------------
-# DeltaChannel — dict reducer
+# Dict-reducer tests
 # ---------------------------------------------------------------------------
 
 
-def _delta_channel_with_type(op, typ):
+def _delta_channel_with_type(operator, typ):
     """Build a DeltaChannel with an explicit type via the Annotated injection path."""
-    return _get_channel("_test", Annotated[typ, DeltaChannel(op)])
+    from typing import Annotated
+
+    from langgraph.channels.delta import DeltaChannel
+    from langgraph.graph.state import _get_channel
+
+    return _get_channel("_test", Annotated[typ, DeltaChannel(operator)])
 
 
 def test_delta_channel_dict_reducer_fresh_channel() -> None:
     """DeltaChannel with a dict reducer starts as empty dict on MISSING checkpoint."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
     assert ch.is_available()
@@ -405,11 +464,8 @@ def test_delta_channel_dict_reducer_fresh_channel() -> None:
 def test_delta_channel_dict_reducer_basic_updates() -> None:
     """DeltaChannel with a dict reducer accumulates key/value pairs across steps."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
 
@@ -427,11 +483,8 @@ def test_delta_channel_dict_reducer_basic_updates() -> None:
 def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
     """replay_writes on a fresh channel replays through a dict merge reducer."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -448,14 +501,15 @@ def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
 def test_delta_channel_dict_reducer_with_deletions() -> None:
     """Dict reducer that treats None values as deletions works end-to-end."""
 
-    def merge_files(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            for k, v in w.items():
-                if v is None:
-                    result.pop(k, None)
-                else:
-                    result[k] = v
+    def merge_files(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return {k: v for k, v in right.items() if v is not None}
+        result = {**left}
+        for k, v in right.items():
+            if v is None:
+                result.pop(k, None)
+            else:
+                result[k] = v
         return result
 
     ch = _delta_channel_with_type(merge_files, dict).from_checkpoint(MISSING)
@@ -476,12 +530,10 @@ def test_delta_channel_dict_reducer_with_deletions() -> None:
 
 def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
     """Overwrite(dict) in update() must preserve dict shape, not coerce to list."""
+    from langgraph.types import Overwrite
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
     ch.update([{"a": 1}])
@@ -491,12 +543,10 @@ def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
 
 def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
     """Overwrite(dict) embedded in replayed writes must reconstruct as dict."""
+    from langgraph.types import Overwrite
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     ch = spec.from_checkpoint(DELTA_SENTINEL)
@@ -512,12 +562,17 @@ def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
 
 def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
     """DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`."""
+    from typing import Annotated
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    from typing_extensions import NotRequired
+
+    from langgraph.channels.delta import DeltaChannel
+    from langgraph.graph.state import _get_channel
+
+    def merge_dicts(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return dict(right)
+        return {**left, **right}
 
     annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
     ch = _get_channel("files", annotation).from_checkpoint(MISSING)
@@ -529,15 +584,23 @@ def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
 
 def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
     """End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel."""
+    from typing import Annotated
 
-    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
+    from langgraph.checkpoint.memory import InMemorySaver
+    from typing_extensions import TypedDict
+
+    from langgraph.channels.delta import DeltaChannel
+    from langgraph.graph import START, StateGraph
+
+    def merge_files(left: dict | None, right: dict) -> dict:
+        if left is None:
+            return {k: v for k, v in right.items() if v is not None}
+        result = {**left}
+        for k, v in right.items():
+            if v is None:
+                result.pop(k, None)
+            else:
+                result[k] = v
         return result
 
     class State(TypedDict):
@@ -590,11 +653,8 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
 def test_delta_channel_dict_reducer_backwards_compat() -> None:
     """A pre-DeltaChannel dict checkpoint must load as a dict, not be listified."""
 
-    def merge_dicts(state: dict, writes: list) -> dict:
-        result = dict(state)
-        for w in writes:
-            result.update(w)
-        return result
+    def merge_dicts(left: dict, right: dict) -> dict:
+        return {**left, **right}
 
     spec = _delta_channel_with_type(merge_dicts, dict)
     old_value = {"a": 1, "b": 2}
@@ -603,7 +663,7 @@ def test_delta_channel_dict_reducer_backwards_compat() -> None:
 
 
 # ---------------------------------------------------------------------------
-# DeltaChannel — seed / pre-delta migration
+# seed / pre-delta migration
 # ---------------------------------------------------------------------------
 
 
@@ -614,7 +674,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None:
     a pre-DeltaChannel blob it passes it as `seed` so replay reconstructs
     the post-migration state correctly rather than replaying from empty.
     """
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    spec = DeltaChannel(add_messages)
     seed = [HumanMessage(content="pre-delta", id="p1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes(
@@ -630,7 +690,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None:
 def test_delta_channel_from_checkpoint_seed_without_writes() -> None:
     """Reconstruction at a pre-delta ancestor with no newer deltas returns
     just the seed — the saver's terminator fired immediately."""
-    spec = DeltaChannel(_messages_delta_reducer, list)
+    spec = DeltaChannel(add_messages)
     seed = [HumanMessage(content="only-snap", id="s1")]
     ch = spec.from_checkpoint(seed)
     ch.replay_writes([])
@@ -644,10 +704,11 @@ def test_delta_channel_from_checkpoint_seed_none_is_distinct_from_sentinel() ->
     explicitly should feed None to the reducer as the left operand.
     """
 
-    def replace(state, writes):
-        return writes[-1] if writes else state
+    def replace(left, right):
+        return right
 
-    spec = DeltaChannel(replace, list)
+    spec = DeltaChannel(replace)
     ch = spec.from_checkpoint(None)
     ch.replay_writes([("t0", "x", "after")])
+    # Reducer replaces; seed=None → first write produces "after".
     assert ch.get() == "after"

libs/langgraph/tests/test_delta_channel_benchmark.py

@@ -17,7 +17,6 @@ from __future__ import annotations
 
 import contextlib
 import math
-import os
 import sys
 import time
 from typing import Annotated, Any
@@ -29,16 +28,13 @@ from typing_extensions import TypedDict
 
 from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import END, StateGraph
-from langgraph.graph.message import _messages_delta_reducer, add_messages
+from langgraph.graph.message import add_messages
 
 try:
     from langgraph.checkpoint.postgres import PostgresSaver
 
     _POSTGRES_AVAILABLE = True
-    _POSTGRES_URI = os.environ.get(
-        "LANGGRAPH_BENCH_POSTGRES_URI",
-        "postgres://postgres@localhost:5432/postgres?sslmode=disable",
-    )
+    _POSTGRES_URI = "postgres://sydney_runkle@localhost:5441/postgres?sslmode=disable"
 except ImportError:
     _POSTGRES_AVAILABLE = False
 
@@ -118,14 +114,12 @@ class BinaryState(TypedDict):
 
 
 class DeltaState(TypedDict):
-    messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+    messages: Annotated[list, DeltaChannel(add_messages)]
 
 
 def _make_delta_state(snapshot_frequency: int | float) -> type:
     """Create a TypedDict with DeltaChannel at the given snapshot_frequency."""
-    channel = DeltaChannel(
-        _messages_delta_reducer, snapshot_frequency=snapshot_frequency
-    )
+    channel = DeltaChannel(add_messages, snapshot_frequency=snapshot_frequency)
     # Use the functional TypedDict form so the Annotated type is stored as an
     # already-evaluated object rather than a forward-reference string (which
     # would fail when get_type_hints tries to resolve 'snapshot_frequency').

libs/langgraph/tests/test_delta_channel_migration.py

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

libs/langgraph/tests/test_pregel.py

@@ -16,7 +16,7 @@ from typing import Annotated, Any, Literal, get_type_hints
 
 import pytest
 from langchain_core.language_models import GenericFakeChatModel
-from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
 from langchain_core.runnables import (
     RunnableConfig,
     RunnableLambda,
@@ -25,7 +25,6 @@ from langchain_core.runnables import (
 from langchain_core.runnables.graph import Edge
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
-    DELTA_SENTINEL,
     BaseCheckpointSaver,
     Checkpoint,
     CheckpointMetadata,
@@ -51,7 +50,7 @@ from langgraph.config import get_stream_writer
 from langgraph.errors import GraphRecursionError, InvalidUpdateError, ParentCommand
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph
-from langgraph.graph.message import MessagesState, _messages_delta_reducer, add_messages
+from langgraph.graph.message import MessagesState, add_messages
 from langgraph.pregel import (
     NodeBuilder,
     Pregel,
@@ -9406,9 +9405,14 @@ 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.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         n = len(state["messages"])
@@ -9442,9 +9446,14 @@ 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.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     counter = {"n": 0}
 
@@ -9494,9 +9503,14 @@ 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.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         return {"messages": [AIMessage(content="reply", id="ai-1")]}
@@ -9535,9 +9549,14 @@ 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.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def update_msg(state: State) -> dict:
         # re-send h1 with updated content
@@ -9567,89 +9586,88 @@ async def test_delta_channel_update_by_id_end_to_end() -> None:
     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."""
+async def test_delta_channel_write_flushed_before_put() -> None:
+    """checkpoint_writes are flushed synchronously before put when DELTA_SENTINEL
+    is present, ensuring writes are durable before the sentinel blob is committed.
+
+    We verify this by intercepting put_writes and put calls and confirming
+    put_writes always completes before put is called for sentinel checkpoints.
+    """
+    import threading
+    from typing import Annotated
+
+    from langchain_core.messages import AIMessage, HumanMessage
+    from langgraph.checkpoint.base import DELTA_SENTINEL
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    from langgraph.graph import START, StateGraph
+    from langgraph.graph.message import add_messages
 
     class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(_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)]
+        messages: Annotated[list, DeltaChannel(add_messages)]
 
     def respond(state: State) -> dict:
         i = len(state["messages"])
         return {"messages": [AIMessage(content=f"r{i}", id=f"ai{i}")]}
 
     order: list[str] = []
-    original_aput_writes = InMemorySaver.aput_writes
-    original_aput = InMemorySaver.aput
+    lock = threading.Lock()
+    original_put_writes = InMemorySaver.put_writes
+    original_put = InMemorySaver.put
 
-    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")
+    def tracked_put_writes(self, config, writes, task_id, task_path=""):
+        result = original_put_writes(self, config, writes, task_id, task_path)
+        with lock:
+            order.append("put_writes")
         return result
 
-    async def tracked_aput(self, config, checkpoint, metadata, new_versions):
+    def tracked_put(self, config, checkpoint, metadata, new_versions):
+        # Check if this checkpoint has any DELTA_SENTINEL blobs
         has_sentinel = any(
             v is DELTA_SENTINEL for v in checkpoint.get("channel_values", {}).values()
         )
-        order.append("aput_sentinel" if has_sentinel else "aput_other")
-        return await original_aput(self, config, checkpoint, metadata, new_versions)
+        if has_sentinel:
+            with lock:
+                order.append("put_sentinel")
+        else:
+            with lock:
+                order.append("put_snapshot")
+        return original_put(self, config, checkpoint, metadata, new_versions)
 
-    InMemorySaver.aput_writes = tracked_aput_writes
-    InMemorySaver.aput = tracked_aput
+    InMemorySaver.put_writes = tracked_put_writes
+    InMemorySaver.put = tracked_put
     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"}}
+        saver = InMemorySaver()
+        graph = builder.compile(checkpointer=saver)
+        config = {"configurable": {"thread_id": "flush-test"}}
 
         for i in range(3):
-            await graph.ainvoke(
+            graph.invoke(
                 {"messages": [HumanMessage(content=f"h{i}", id=f"h{i}")]}, config
             )
 
-        # Every aput_sentinel must be preceded by at least one aput_writes
+        # For every sentinel put, all preceding put_writes must already be in order
         for i, event in enumerate(order):
-            if event == "aput_sentinel":
+            if event == "put_sentinel":
+                # All put_writes before this index must appear before this sentinel
                 preceding = order[:i]
-                assert "aput_writes" in preceding, (
-                    f"aput_sentinel at {i} had no preceding aput_writes: {order}"
+                assert "put_writes" in preceding, (
+                    f"put_sentinel at index {i} had no preceding put_writes: {order}"
                 )
+                # And the most recent put_writes must come before this sentinel
                 last_write_idx = max(
-                    j for j, e in enumerate(order[:i]) if e == "aput_writes"
+                    j for j, e in enumerate(order[:i]) if e == "put_writes"
                 )
                 assert last_write_idx < i, (
-                    f"aput_writes at {last_write_idx} should precede aput_sentinel at {i}: {order}"
+                    f"put_writes at {last_write_idx} not before put_sentinel at {i}"
                 )
     finally:
-        InMemorySaver.aput_writes = original_aput_writes
-        InMemorySaver.aput = original_aput
+        InMemorySaver.put_writes = original_put_writes
+        InMemorySaver.put = original_put
 
-    state = await graph.aget_state(config)
+    # Final state must still be correct
+    state = graph.get_state(config)
     assert len(state.values["messages"]) == 6  # 3 human + 3 AI

libs/langgraph/tests/test_pregel_async.py

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

libs/langgraph/uv.lock

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

notes/delta_channel_query_bench.md

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