chore: ruff format fixes

Co-authored-by: Cursor <cursoragent@cursor.com>

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import threading
 from collections import defaultdict
-from collections.abc import Iterator, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from contextlib import contextmanager
 from typing import Any, cast
 
@@ -19,6 +19,7 @@ from langgraph.checkpoint.base import (
     get_serializable_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.base import SerializerProtocol
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from psycopg import Capabilities, Connection, Cursor, Pipeline
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -26,9 +27,10 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
+    _build_delta_stage1_sql,
+    _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -308,7 +310,12 @@ class PostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            if v is None or isinstance(v, (str, int, float, bool)):
+            if v is DELTA_SENTINEL:
+                copy["channel_values"].pop(k)
+            elif isinstance(v, _DeltaSnapshot):
+                blob_values[k] = copy["channel_values"].pop(k)
+                copy["channel_values"][k] = True
+            elif v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -436,46 +443,80 @@ class PostgresSaver(BasePostgresSaver):
                 with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    def _get_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
-        """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
+    def _get_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Fast-path override of `BaseCheckpointSaver._get_all_delta_channels_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.
+        Two-stage query, both stages cover ALL requested channels in a single
+        Postgres roundtrip each:
+
+        * Stage 1: dynamic SELECT over `checkpoints` with K parallel JSONB
+          key lookups (one column pair per channel) — no subquery, no
+          aggregation. Returns one row per checkpoint with versions and
+          snapshot flags for every requested channel.
+
+        * Stage 2: one UNION ALL over `checkpoint_writes` and
+          `checkpoint_blobs` filtered by `channel = ANY(?)` and per-channel
+          chain_cids / seed_versions (collapsed across channels).
         """
+        if not channels:
+            return {}
+        channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
         if checkpoint_id is None:
-            # Caller didn't specify a target — resolve to the latest
-            # checkpoint on the thread. `get_tuple` without `checkpoint_id`
-            # returns the newest; its config carries the resolved id.
             target = self.get_tuple(config)
             if target is None:
-                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                return {
+                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                    for ch in channels
+                }
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
+
+        # Stage 1: K parallel JSONB lookups per row, one query for all channels.
+        stage1_sql = _build_delta_stage1_sql(channels)
+        stage1_params: list[Any] = []
+        for ch in channels:
+            stage1_params.extend([ch, ch])
+        stage1_params.extend([thread_id, checkpoint_ns])
+        with self._cursor() as cur:
+            cur.execute(stage1_sql, stage1_params)
+            stage1_rows = cur.fetchall()
+
+        chain_by_ch, seed_ver_by_ch = self._walk_stage1_multi(
+            cast("list[Mapping[str, Any]]", stage1_rows), checkpoint_id, channels
+        )
+        # Union of chain cids and seed versions across all channels.
+        union_chain_cids: list[str] = sorted(
+            {cid for chain in chain_by_ch.values() for cid in chain}
+        )
+        union_seed_versions: list[str] = sorted(
+            {ver for ver in seed_ver_by_ch.values() if ver is not None}
+        )
+
+        # Stage 2: chain-limited writes + chain-limited seed blobs for all channels.
         with self._cursor() as cur:
             cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
+                SELECT_DELTA_STAGE2_SQL,
                 (
-                    channel,
                     thread_id,
                     checkpoint_ns,
+                    channels,
+                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
+                    channels,
+                    union_seed_versions,
                 ),
             )
-            rows = cur.fetchall()
-        return self._build_delta_channel_writes_history(
-            channel=channel,
-            target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            stage2_rows = cur.fetchall()
+        return self._build_delta_channels_writes_history(
+            channels=channels,
+            chain_by_ch=chain_by_ch,
+            seed_ver_by_ch=seed_ver_by_ch,
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 from collections import defaultdict
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from contextlib import asynccontextmanager
 from typing import Any, cast
 
@@ -19,6 +19,7 @@ from langgraph.checkpoint.base import (
     get_serializable_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.base import SerializerProtocol
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -26,9 +27,10 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
+    _build_delta_stage1_sql,
+    _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -267,7 +269,12 @@ class AsyncPostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            if v is None or isinstance(v, (str, int, float, bool)):
+            if v is DELTA_SENTINEL:
+                copy["channel_values"].pop(k)
+            elif isinstance(v, _DeltaSnapshot):
+                blob_values[k] = copy["channel_values"].pop(k)
+                copy["channel_values"][k] = True
+            elif v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -397,44 +404,69 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    async def _aget_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
-        """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
+    async def _aget_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Fast-path override of `BaseCheckpointSaver._aget_all_delta_channels_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`.
+        Two-stage query, both stages cover ALL requested channels in a single
+        Postgres roundtrip each. See `PostgresSaver._get_all_delta_channels_writes_history`
+        for design notes.
         """
+        if not channels:
+            return {}
+        channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
         if checkpoint_id is None:
             target = await self.aget_tuple(config)
             if target is None:
-                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                return {
+                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                    for ch in channels
+                }
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
+
+        stage1_sql = _build_delta_stage1_sql(channels)
+        stage1_params: list[Any] = []
+        for ch in channels:
+            stage1_params.extend([ch, ch])
+        stage1_params.extend([thread_id, checkpoint_ns])
+        async with self._cursor() as cur:
+            await cur.execute(stage1_sql, stage1_params)
+            stage1_rows = await cur.fetchall()
+
+        chain_by_ch, seed_ver_by_ch = self._walk_stage1_multi(
+            cast("list[Mapping[str, Any]]", stage1_rows), checkpoint_id, channels
+        )
+        union_chain_cids: list[str] = sorted(
+            {cid for chain in chain_by_ch.values() for cid in chain}
+        )
+        union_seed_versions: list[str] = sorted(
+            {ver for ver in seed_ver_by_ch.values() if ver is not None}
+        )
+
         async with self._cursor() as cur:
             await cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
+                SELECT_DELTA_STAGE2_SQL,
                 (
-                    channel,
                     thread_id,
                     checkpoint_ns,
+                    channels,
+                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
-                    channel,
+                    channels,
+                    union_seed_versions,
                 ),
             )
-            rows = await cur.fetchall()
-        return self._build_delta_channel_writes_history(
-            channel=channel,
-            target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            stage2_rows = await cur.fetchall()
+        return self._build_delta_channels_writes_history(
+            channels=channels,
+            chain_by_ch=chain_by_ch,
+            seed_ver_by_ch=seed_ver_by_ch,
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import random
 import warnings
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from importlib.metadata import version as get_version
 from typing import Any, TypedDict, cast
 
@@ -156,62 +156,100 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """
 
 
-class _DeltaCombinedRow(TypedDict, total=False):
-    """One row from `SELECT_DELTA_COMBINED_SQL` (a UNION ALL of three tables).
+class _DeltaStage2Row(TypedDict, total=False):
+    """One row from `SELECT_DELTA_STAGE2_SQL` (a UNION ALL of writes and blobs)."""
 
-    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")
+    _kind: str  # "w" or "b"
+    checkpoint_id: str | None  # "w" rows only
+    channel: str | None  # set on both "w" and "b" rows
     type: str | None
     blob: bytes | None
-    # write row only ("w")
-    task_id: str | None
-    idx: int | None
-    # blob row only ("b")
-    version: str | None
+    task_id: str | None  # "w" rows only
+    idx: int | None  # "w" rows only
+    version: str | None  # "b" rows only
 
 
-# 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`.
+# Multi-channel two-stage DeltaChannel reconstruction.
 #
-# Parameter order: (channel, thread_id, checkpoint_ns,
-#                   thread_id, checkpoint_ns, channel,
-#                   thread_id, checkpoint_ns, channel)
-SELECT_DELTA_COMBINED_SQL = """
-    SELECT 'p'::text       AS _kind,
-           checkpoint_id,
-           parent_checkpoint_id,
-           checkpoint -> 'channel_versions' ->> %s AS ver,
-           NULL::text       AS type,
-           NULL::bytea      AS blob,
-           NULL::text       AS task_id,
-           NULL::int        AS idx,
-           NULL::text       AS version
-    FROM checkpoints
-    WHERE thread_id = %s AND checkpoint_ns = %s
-    UNION ALL
-    SELECT 'w',
-           checkpoint_id, NULL, NULL,
-           type, blob, task_id, idx, NULL
+# Stage 1 scans checkpoint metadata (no blob bytes) and emits one row per
+# checkpoint with K parallel JSONB key lookups (one column pair per
+# requested delta channel: ver_i / hs_i).  No subqueries, no aggregation.
+# Python walks the parent chain once across all channels.
+#
+# Stage 2 fetches all writes and the seed blobs for ALL channels in a
+# single roundtrip via `channel = ANY(%s)` and chain/seed-version
+# filtering.
+#
+# Empirical comparison vs an alternative "ship full channel_versions /
+# channel_values JSONB and let Python pick" form (1000 checkpoints,
+# 8 total channels in graph, 3 delta channels requested):
+#
+#   Postgres execution:    A=0.24ms vs B=0.38ms   (both negligible)
+#   End-to-end latency:    A=6.83ms vs B=2.28ms   (B is 3.0x faster)
+#   Wire payload:          A=836KB  vs B=330KB    (61% smaller)
+#   Buffer hits:           identical (167 blocks)
+#
+# B (this dynamic-columns design) wins because it avoids JSONB
+# serialization on the wire and JSONB-to-dict deserialization in
+# psycopg.  Even at K=8 (8 delta channels = 16 dynamic columns), B
+# still beats A end-to-end (4.2ms vs 6.8ms).
+
+
+def _build_delta_stage1_sql(channels: Sequence[str]) -> str:
+    """Build stage 1 SQL with 2K parallel JSONB key lookups.
+
+    For channels=["messages", "files"] the result is::
+
+        SELECT checkpoint_id, parent_checkpoint_id,
+               checkpoint -> 'channel_versions' ->> %s AS ver_0,
+               (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_0,
+               checkpoint -> 'channel_versions' ->> %s AS ver_1,
+               (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_1
+        FROM checkpoints
+        WHERE thread_id = %s AND checkpoint_ns = %s
+
+    Channel names are passed as `%s` parameters (safe from SQL injection).
+    Only the column aliases `ver_i` / `hs_i` are interpolated into the
+    SQL string (i is bounded by len(channels) and uses safe identifiers).
+
+    Caller must extend params with `[ch_0, ch_0, ch_1, ch_1, ...,
+    thread_id, ns]`.
+    """
+    cols = []
+    for i in range(len(channels)):
+        cols.append(
+            f"checkpoint -> 'channel_versions' ->> %s AS ver_{i}, "
+            f"(checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_{i}"
+        )
+    return (
+        "SELECT checkpoint_id, parent_checkpoint_id, "
+        + ", ".join(cols)
+        + " FROM checkpoints WHERE thread_id = %s AND checkpoint_ns = %s"
+    )
+
+
+SELECT_DELTA_STAGE2_SQL = """
+    SELECT 'w'::text AS _kind,
+           checkpoint_id, channel,
+           type, blob, task_id, idx, NULL::text AS version
     FROM checkpoint_writes
-    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = ANY(%s)
+      AND checkpoint_id = ANY(%s)
     UNION ALL
-    SELECT 'b',
-           NULL, NULL, NULL,
+    SELECT 'b', NULL, channel,
            type, blob, NULL, NULL, version
     FROM checkpoint_blobs
-    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = ANY(%s)
+      AND version = ANY(%s)
 """
 
 
+# Stage 1 rows are dynamic-shape dicts: {checkpoint_id, parent_checkpoint_id,
+# ver_0, hs_0, ver_1, hs_1, ...}.  Walking is parameterized by the channel
+# list to map indices back to channel names — no static TypedDict here.
+# `dict[str, Any]` is the practical signature.
+
+
 class BasePostgresSaver(BaseCheckpointSaver[str]):
     SELECT_SQL = SELECT_SQL
     SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
@@ -254,82 +292,120 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             if t.decode() != "empty"
         }
 
-    def _build_delta_channel_writes_history(
-        self,
-        *,
-        channel: str,
+    @staticmethod
+    def _walk_stage1_multi(
+        stage1_rows: Sequence[Mapping[str, Any]],
         target_id: str,
-        rows: Sequence[_DeltaCombinedRow],
-    ) -> _ChannelWritesHistory:
-        """Reconstruct one delta channel's history from the combined UNION ALL rows.
+        channels: Sequence[str],
+    ) -> tuple[dict[str, list[str]], dict[str, str | None]]:
+        """Walk the parent chain once for all requested channels.
 
-        Pure data transform shared by sync (`PostgresSaver`) and async
-        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
-        and feed the tagged rows here.
+        Each row carries `ver_i` / `hs_i` per channel index. We walk the
+        parent chain from target's parent toward the root; for each
+        channel we stop at the nearest ancestor where `hs_i` is true and
+        record that ancestor's `ver_i` as the seed version. All
+        ancestors visited up to (and including) a channel's seed are in
+        that channel's `chain_cids`.
 
-        Walk is newest → oldest from the target's parent. A non-sentinel
-        blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
-        walk and is returned as the seed so replay starts from it.
-
-        Writes stored at `target_id` itself are pending writes for the next
-        step and are excluded — the walk begins at the target's parent.
+        Returns:
+          chain_cids_by_channel: per-channel list of ancestor cids in
+                                 newest-first order.
+          seed_version_by_channel: per-channel seed version (None if
+                                   walk reached root with no snapshot).
         """
         parent_of: dict[str, str | None] = {}
-        ver_of: dict[str, str | None] = {}
-        writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
-        blob_by_ver: dict[str, tuple[str, bytes]] = {}
+        # For each channel index, store ver and has_snapshot per cid.
+        ver_by_i_by_cid: list[dict[str, str | None]] = [
+            {} for _ in range(len(channels))
+        ]
+        hs_by_i_by_cid: list[dict[str, bool]] = [{} for _ in range(len(channels))]
 
-        for r in rows:
+        for r in stage1_rows:
+            cid = cast(str, r["checkpoint_id"])
+            parent_of[cid] = cast("str | None", r["parent_checkpoint_id"])
+            for i in range(len(channels)):
+                ver_by_i_by_cid[i][cid] = cast("str | None", r.get(f"ver_{i}"))
+                hs_by_i_by_cid[i][cid] = bool(r.get(f"hs_{i}"))
+
+        chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
+        seed_ver_by_ch: dict[str, str | None] = {ch: None for ch in channels}
+        # For each channel, walk from target's parent until we hit a
+        # snapshot or the root. Walks share the parent_of mapping but
+        # are otherwise independent.
+        for i, ch in enumerate(channels):
+            cur_cid: str | None = parent_of.get(target_id)
+            while cur_cid is not None:
+                chain_by_ch[ch].append(cur_cid)
+                if hs_by_i_by_cid[i].get(cur_cid, False):
+                    seed_ver_by_ch[ch] = ver_by_i_by_cid[i].get(cur_cid)
+                    break
+                cur_cid = parent_of.get(cur_cid)
+        return chain_by_ch, seed_ver_by_ch
+
+    def _build_delta_channels_writes_history(
+        self,
+        *,
+        channels: Sequence[str],
+        chain_by_ch: dict[str, list[str]],
+        seed_ver_by_ch: dict[str, str | None],
+        stage2_rows: Sequence[_DeltaStage2Row],
+    ) -> dict[str, _ChannelWritesHistory]:
+        """Demux stage 2 rows per channel; produce per-channel histories.
+
+        stage2_rows carry `channel` on every row. We build per-channel
+        `writes_by_cid` and per-channel `seed_blob` dicts, then assemble
+        a `_ChannelWritesHistory` per requested channel.
+        """
+        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
+        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
+            ch: {} for ch in channels
+        }
+        # seed_blob_by_ver[(channel, version)] = (type, blob)
+        seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
+
+        for r in stage2_rows:
+            ch = cast(str, r["channel"])
             kind = r["_kind"]
-            if kind == "p":
+            if kind == "w":
                 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(
+                writes_by_ch_by_cid.setdefault(ch, {}).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(
+                ver = cast(str, r["version"])
+                seed_blob_by_ver[(ch, ver)] = cast(
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # newest write first per ancestor (task_id DESC, idx DESC)
-        for ws in writes_by_cid.values():
-            ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
+        # Sort writes per (channel, cid) newest-first by (task_id, idx)
+        for cid_map in writes_by_ch_by_cid.values():
+            for ws in cid_map.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=[])
+        result: dict[str, _ChannelWritesHistory] = {}
+        for ch in channels:
+            chain_cids = chain_by_ch.get(ch, [])
+            seed_version = seed_ver_by_ch.get(ch)
 
-        collected: list[PendingWrite] = []  # newest first; reversed at the end
-        for cid in ancestors:
-            # Collect writes first — they encode the transition FROM this
-            # ancestor's state to its child's and must be included even if
-            # this ancestor is also the seed checkpoint.
-            for type_tag, write_blob, task_id, _idx in writes_by_cid.get(cid, []):
-                val = self.serde.loads_typed((type_tag, write_blob))
-                collected.append((task_id, channel, val))
-            # Then check seed terminator.
-            ver = ver_of.get(cid)
-            if ver is not None:
-                seed_blob = blob_by_ver.get(ver)
-                if seed_blob is not None and seed_blob[0] != "empty":
-                    blob_value = self.serde.loads_typed(seed_blob)
-                    if blob_value is not DELTA_SENTINEL:
-                        collected.reverse()
-                        return _ChannelWritesHistory(seed=blob_value, writes=collected)
+            collected: list[PendingWrite] = []
+            cid_writes = writes_by_ch_by_cid.get(ch, {})
+            for cid in chain_cids:
+                for type_tag, write_blob, task_id, _idx in cid_writes.get(cid, []):
+                    val = self.serde.loads_typed((type_tag, write_blob))
+                    collected.append((task_id, ch, val))
 
-        collected.reverse()  # oldest → newest
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+            seed: Any = DELTA_SENTINEL
+            if seed_version is not None:
+                blob = seed_blob_by_ver.get((ch, seed_version))
+                if blob is not None and blob[0] != "empty":
+                    seed = self.serde.loads_typed(blob)
+
+            collected.reverse()
+            result[ch] = _ChannelWritesHistory(seed=seed, writes=collected)
+        return result
 
     def _dump_blobs(
         self,

libs/checkpoint-postgres/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a1"
+version = "3.1.0a3"
 description = "Library with a Postgres implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.10"
@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-  "langgraph-checkpoint>=4.1.0a1,<5.0.0",
+  "langgraph-checkpoint>=4.1.0a3,<5.0.0",
   "orjson>=3.11.5",
   "psycopg>=3.2.0",
   "psycopg-pool>=3.2.0",

libs/checkpoint-postgres/uv.lock

@@ -259,7 +259,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -307,7 +307,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a1"
+version = "3.1.0a3"
 source = { editable = "." }
 dependencies = [
     { name = "langgraph-checkpoint" },

libs/checkpoint-sqlite/uv.lock

@@ -268,7 +268,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },

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

@@ -125,7 +125,8 @@ class CheckpointTuple(NamedTuple):
 
 
 class _ChannelWritesHistory(NamedTuple):
-    """Result of `BaseCheckpointSaver._get_channel_writes_history`.
+    """Result of `BaseCheckpointSaver._get_all_delta_channels_writes_history`
+    (a per-channel entry from the returned mapping).
 
     Storage-level view of what one channel wrote across the ancestor chain
     of a target checkpoint:
@@ -487,12 +488,12 @@ class BaseCheckpointSaver(Generic[V]):
         raise NotImplementedError
 
     def _get_tuple_raw(self, config: RunnableConfig) -> CheckpointTuple | None:
-        """Pure storage read used by `_get_channel_writes_history`.
+        """Pure storage read used by `_get_all_delta_channels_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.
+        `_get_all_delta_channels_writes_history`. Override only if `get_tuple`
+        itself performs channel hydration.
         """
         return self.get_tuple(config)
 
@@ -500,14 +501,17 @@ class BaseCheckpointSaver(Generic[V]):
         """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:
-        """**Experimental.** Query one channel's writes along the parent chain.
+    def _get_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """**Experimental.** Query multiple delta channels' writes along the parent chain.
 
-        Storage-level query, not channel semantics: returns `(seed, writes)`
-        reflecting what storage knows about a single channel across the
-        ancestor chain of the target checkpoint identified by `config`.
+        Storage-level query, not channel semantics: returns a per-channel
+        `(seed, writes)` reflecting what storage knows about each channel
+        across the ancestor chain of the target checkpoint identified by
+        `config`.
+
+        For every channel in `channels`:
 
         * `writes` — on-path deltas oldest→newest as `PendingWrite` tuples.
           Writes stored at the target `checkpoint_id` itself are pending
@@ -520,69 +524,86 @@ class BaseCheckpointSaver(Generic[V]):
         Walks the **parent chain** (not `list(before=...)`): for forked
         threads, only on-path ancestors contribute.
 
-        Reference implementation walks `get_tuple` + `parent_config`,
-        inspecting each ancestor's `channel_values[channel]` for the seed
-        terminator. Savers with direct storage access (`InMemorySaver`,
-        `PostgresSaver`) override for performance; the return contract is
-        fixed here.
+        Reference implementation walks `get_tuple` + `parent_config` ONCE
+        for all channels (each ancestor visited once, not once per channel),
+        inspecting each ancestor's `channel_values[channel]` for that
+        channel's seed terminator. Savers with direct storage access
+        (`InMemorySaver`, `PostgresSaver`) override for performance; the
+        return contract is fixed here.
 
-        Underscore-prefixed because the method surface is experimental.
+        Empty `channels` returns `{}`. Underscore-prefixed because the
+        method surface is experimental.
         """
-        collected: list[PendingWrite] = []  # newest first; reversed at the end
+        if not channels:
+            return {}
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {c: DELTA_SENTINEL for c in channels}
+        remaining: set[str] = set(channels)
         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:
+        while cursor_config is not None and remaining:
             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.
+            # Collect each ancestor's writes for any channel still searching.
             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)
+                    ch = write[1]
+                    if ch in remaining:
+                        collected_by_ch[ch].append(write)
+            # Per-channel seed terminator: a non-sentinel blob value at this
+            # ancestor establishes that channel's reconstruction base.
+            for ch in list(remaining):
+                ancestor_value = tup.checkpoint["channel_values"].get(ch)
+                if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                    seed_by_ch[ch] = ancestor_value
+                    remaining.discard(ch)
             cursor_config = tup.parent_config
-        collected.reverse()
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        return {
+            ch: _ChannelWritesHistory(
+                seed=seed_by_ch[ch],
+                writes=list(reversed(collected_by_ch[ch])),
+            )
+            for ch in channels
+        }
 
-    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] = []
+    async def _aget_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Async version of `_get_all_delta_channels_writes_history`."""
+        if not channels:
+            return {}
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {c: DELTA_SENTINEL for c in channels}
+        remaining: set[str] = set(channels)
         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:
+        while cursor_config is not None and remaining:
             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)
+                    ch = write[1]
+                    if ch in remaining:
+                        collected_by_ch[ch].append(write)
+            for ch in list(remaining):
+                ancestor_value = tup.checkpoint["channel_values"].get(ch)
+                if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                    seed_by_ch[ch] = ancestor_value
+                    remaining.discard(ch)
             cursor_config = tup.parent_config
-        collected.reverse()
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        return {
+            ch: _ChannelWritesHistory(
+                seed=seed_by_ch[ch],
+                writes=list(reversed(collected_by_ch[ch])),
+            )
+            for ch in channels
+        }
 
     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

@@ -6,7 +6,7 @@ import pickle
 import random
 import shutil
 from collections import defaultdict
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from types import TracebackType
 from typing import Any
@@ -141,17 +141,24 @@ class InMemorySaver(
             result[k] = self.serde.loads_typed(vv)
         return result
 
-    def _get_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
+    def _get_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Override: walk the parent chain ONCE for all requested channels.
+
+        For each channel we track its own seed terminator independently.
+        On a snapshot or pre-delta ancestor for a given channel, that
+        channel stops collecting further writes; other channels keep
+        walking until they find their own terminator or hit the root.
+        """
+        if not channels:
+            return {}
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = config["configurable"].get("checkpoint_id", "")
         ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
-        # Walk the parent chain newest→oldest. Skip the target itself —
-        # writes stored AT `checkpoint_id` are pending for the next step
-        # (pregel applies them via `apply_writes`; they aren't part of the
-        # snapshot value AT `checkpoint_id`).
+
+        # Build the parent chain (newest→oldest), skipping the target.
         chain: list[str] = []
         target_entry = ns_storage.get(checkpoint_id)
         current: str | None = target_entry[2] if target_entry is not None else None
@@ -162,77 +169,73 @@ class InMemorySaver(
             chain.append(current)
             _, _, parent = entry
             current = parent
-        # Scan newest→oldest. A pre-delta blob on an ancestor terminates the
-        # walk and is bound as `seed`; without this, a thread migrated from
-        # pre-delta storage would replay ancestor writes all the way to the
-        # root AND miss any value that lived only in the old blob (e.g. from
-        # `update_state`).
-        #
-        # At each ancestor, check the blob BEFORE processing its pending
-        # writes: a pre-delta blob represents the state AT that ancestor,
-        # which already subsumes any writes stored under it. Processing
-        # those writes first would fold them into the reconstructed value
-        # twice (once via the blob, once via replay).
-        collected: list[PendingWrite] = []  # newest first
-        for cp_id in chain:  # newest → oldest
-            entry = ns_storage.get(cp_id)
-            if entry is not None:
-                ckpt = self.serde.loads_typed(entry[0])
-                ver = ckpt.get("channel_versions", {}).get(channel)
-                if ver is not None:
-                    blob_entry = self.blobs.get(
-                        (thread_id, checkpoint_ns, channel, ver)
-                    )
-                    if blob_entry is not None and blob_entry[0] != "empty":
-                        blob_value = self.serde.loads_typed(blob_entry)
-                        if blob_value is not DELTA_SENTINEL:
-                            if isinstance(blob_value, _DeltaSnapshot):
-                                # Step-based snapshot: the blob is state AT this
-                                # ancestor, but the ancestor's pending_writes
-                                # encode the NEXT step's transition and are NOT
-                                # subsumed by the snapshot — collect them first.
-                                step_writes = self.writes.get(
-                                    (thread_id, checkpoint_ns, cp_id), {}
-                                )
-                                for (_task_id, _idx), (
-                                    tid,
-                                    ch,
-                                    serialized,
-                                    _,
-                                ) in sorted(step_writes.items(), reverse=True):
-                                    if ch != channel:
-                                        continue
-                                    collected.append(
-                                        (tid, ch, self.serde.loads_typed(serialized))
-                                    )
-                                collected.reverse()
-                                return _ChannelWritesHistory(
-                                    seed=blob_value, writes=collected
-                                )
-                            # Pre-delta blob: state AT this ancestor already
-                            # subsumes its pending_writes — skip them.
-                            collected.reverse()
-                            return _ChannelWritesHistory(
-                                seed=blob_value, writes=collected
-                            )
 
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {c: DELTA_SENTINEL for c in channels}
+        remaining: set[str] = set(channels)
+
+        for cp_id in chain:  # newest → oldest
+            if not remaining:
+                break
+            entry = ns_storage.get(cp_id)
+            ckpt = self.serde.loads_typed(entry[0]) if entry is not None else None
+
+            # Per-channel: check seed terminator at this ancestor first.
+            terminated_here: set[str] = set()
+            blob_value_by_ch: dict[str, Any] = {}
+            if ckpt is not None:
+                versions = ckpt.get("channel_versions", {})
+                for ch in remaining:
+                    ver = versions.get(ch)
+                    if ver is None:
+                        continue
+                    blob_entry = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                    if blob_entry is None or blob_entry[0] == "empty":
+                        continue
+                    blob_value = self.serde.loads_typed(blob_entry)
+                    if blob_value is DELTA_SENTINEL:
+                        continue
+                    blob_value_by_ch[ch] = blob_value
+                    terminated_here.add(ch)
+
+            # Process step writes: filter by channel, collect newest-first.
             step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-            # Within a superstep, sorted by (task_id, idx) = oldest → newest;
-            # reverse for newest-first scan.
             for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
                 step_writes.items(), reverse=True
             ):
-                if ch != channel:
+                if ch not in remaining:
                     continue
-                val = self.serde.loads_typed(serialized)
-                collected.append((tid, ch, val))
-        collected.reverse()
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+                blob_value = blob_value_by_ch.get(ch)
+                if blob_value is not None and not isinstance(
+                    blob_value, _DeltaSnapshot
+                ):
+                    # Pre-delta blob terminator: state at this ancestor
+                    # already subsumes these writes — skip them.
+                    continue
+                # Either no terminator at this ancestor for this channel,
+                # OR a `_DeltaSnapshot` terminator (writes here encode the
+                # transition to the child and are NOT subsumed).
+                collected_by_ch[ch].append(
+                    (tid, ch, self.serde.loads_typed(serialized))
+                )
 
-    async def _aget_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
-        return self._get_channel_writes_history(config, channel)
+            # Now apply terminators: channels that found a seed are done.
+            for ch in terminated_here:
+                seed_by_ch[ch] = blob_value_by_ch[ch]
+                remaining.discard(ch)
+
+        return {
+            ch: _ChannelWritesHistory(
+                seed=seed_by_ch[ch],
+                writes=list(reversed(collected_by_ch[ch])),
+            )
+            for ch in channels
+        }
+
+    async def _aget_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        return self._get_all_delta_channels_writes_history(config, channels)
 
     def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the in-memory storage.
@@ -452,7 +455,9 @@ class InMemorySaver(
         values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
         for k, v in new_versions.items():
             self.blobs[(thread_id, checkpoint_ns, k, v)] = (
-                self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
+                self.serde.dumps_typed(values[k])
+                if k in values and values[k] is not DELTA_SENTINEL
+                else ("empty", b"")
             )
         self.storage[thread_id][checkpoint_ns].update(
             {

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

@@ -34,9 +34,7 @@ from langgraph.checkpoint.serde import _msgpack as _lg_msgpack
 from langgraph.checkpoint.serde.base import SerializerProtocol
 from langgraph.checkpoint.serde.event_hooks import emit_serde_event
 from langgraph.checkpoint.serde.types import (
-    DELTA_SENTINEL,
     SendProtocol,
-    _DeltaSentinel,
     _DeltaSnapshot,
 )
 from langgraph.store.base import Item
@@ -322,14 +320,11 @@ EXT_PYDANTIC_V1 = 4
 EXT_PYDANTIC_V2 = 5
 EXT_NUMPY_ARRAY = 6
 EXT_DELTA_SNAPSHOT = 7
-EXT_DELTA_SENTINEL = 8
 
 
 def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
     if isinstance(obj, _DeltaSnapshot):
         return ormsgpack.Ext(EXT_DELTA_SNAPSHOT, _msgpack_enc(obj.value))
-    elif isinstance(obj, _DeltaSentinel):
-        return ormsgpack.Ext(EXT_DELTA_SENTINEL, b"")
     elif hasattr(obj, "model_dump") and callable(obj.model_dump):  # pydantic v2
         return ormsgpack.Ext(
             EXT_PYDANTIC_V2,
@@ -656,9 +651,7 @@ def _create_msgpack_ext_hook(
         return False
 
     def ext_hook(code: int, data: bytes) -> Any:
-        if code == EXT_DELTA_SENTINEL:
-            return DELTA_SENTINEL
-        elif code == EXT_DELTA_SNAPSHOT:
+        if code == EXT_DELTA_SNAPSHOT:
             return _DeltaSnapshot(
                 ormsgpack.unpackb(
                     data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS

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

@@ -17,12 +17,10 @@ TASKS = "__pregel_tasks"
 
 
 class _DeltaSentinel:
-    """Singleton marker stored (as zero bytes) in checkpoint_blobs for a
-    DeltaChannel field. The actual per-step writes live in checkpoint_writes
-    and are replayed through the reducer at load time.
+    """In-memory marker for a DeltaChannel field with no snapshot.
 
-    Compare with `is DELTA_SENTINEL` — `loads_typed` always returns the same
-    module-level instance.
+    Never serialized to storage — checkpointers strip it before writing.
+    Compare with `is DELTA_SENTINEL`; always the same module-level instance.
     """
 
     __slots__ = ()
@@ -38,7 +36,7 @@ class _DeltaSnapshot(NamedTuple):
     """Snapshot blob for a DeltaChannel with finite snapshot_frequency.
 
     Stored in checkpoint_blobs via the `EXT_DELTA_SNAPSHOT` msgpack ext code.
-    The ancestor walk in `_get_channel_writes_history` terminates when it
+    The ancestor walk in `_get_all_delta_channels_writes_history` terminates when it
     encounters this type (any non-sentinel blob stops the walk).
 
     `from_checkpoint` reconstructs the channel value directly from `.value`

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_jsonplus.py

@@ -1048,15 +1048,3 @@ def test_msgpack_nested_pydantic_serializes_as_dict(
     # No blocking should occur - inner is serialized as dict, not ext
     assert "blocked" not in caplog.text.lower()
     assert result == obj
-
-
-def test_delta_sentinel_serde_round_trip() -> None:
-    from langgraph.checkpoint.base import DELTA_SENTINEL
-    from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-
-    serde = JsonPlusSerializer()
-    type_tag, blob = serde.dumps_typed(DELTA_SENTINEL)
-    assert type_tag == "msgpack"
-    assert blob  # non-empty ext envelope
-    loaded = serde.loads_typed((type_tag, blob))
-    assert loaded is DELTA_SENTINEL

libs/checkpoint/tests/test_memory.py

@@ -322,31 +322,23 @@ def test_memory_saver_with_allowlist_proxy_isolated() -> None:
 
 
 class TestInMemorySaverDeltaChannel:
-    def test_load_blobs_returns_sentinel_for_delta_channel(self) -> None:
-        """_load_blobs returns DELTA_SENTINEL for delta channels (reconstruction deferred)."""
+    def test_load_blobs_omits_delta_channel(self) -> None:
+        """_load_blobs omits delta channels (stored as 'empty'); reconstruction deferred."""
         saver = InMemorySaver()
-        serde = JsonPlusSerializer()
 
         thread_id, ns, channel = "t1", "", "messages"
         v1 = "00000000000000000000000000000001.0000000000000000"
 
-        saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(DELTA_SENTINEL)
-
-        cp1 = empty_checkpoint()
-        cp1["id"] = "cp1"
-        cp1["channel_versions"][channel] = v1
-        saver.storage[thread_id][ns] = {
-            "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
-        }
+        saver.blobs[(thread_id, ns, channel, v1)] = ("empty", b"")
 
         result = saver._load_blobs(thread_id, ns, {channel: v1})
-        assert channel in result
-        assert result[channel] is DELTA_SENTINEL
+        assert channel not in result
 
     def test_get_channel_writes_collects_ancestor_writes_only(self) -> None:
-        """_get_channel_writes_history collects ancestor writes oldest→newest,
-        and excludes writes stored at the target checkpoint itself (those are
-        pending writes for the next step, applied separately by pregel)."""
+        """_get_all_delta_channels_writes_history collects ancestor writes
+        oldest→newest, and excludes writes stored at the target checkpoint
+        itself (those are pending writes for the next step, applied separately
+        by pregel)."""
         saver = InMemorySaver()
         serde = JsonPlusSerializer()
 
@@ -384,7 +376,9 @@ class TestInMemorySaverDeltaChannel:
                 "checkpoint_id": "cp2",
             }
         }
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
         assert values == [{"content": "hi"}]
@@ -414,15 +408,17 @@ class TestInMemorySaverDeltaChannel:
                 "checkpoint_id": "cp1",
             }
         }
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
         assert result.seed is DELTA_SENTINEL
         assert result.writes == []
 
 
 class TestBaseFallbackGetChannelWrites:
-    """Exercises the `BaseCheckpointSaver._get_channel_writes_history` default
-    implementation — the path third-party savers inherit when they don't
-    override `_get_channel_writes_history` themselves.
+    """Exercises the `BaseCheckpointSaver._get_all_delta_channels_writes_history`
+    default implementation — the path third-party savers inherit when they
+    don't override `_get_all_delta_channels_writes_history` themselves.
 
     Regression guard for a bug where the fallback passed the caller's config
     (with `checkpoint_id`) straight to `self.list()`, which most savers
@@ -438,11 +434,11 @@ class TestBaseFallbackGetChannelWrites:
         """
 
         class _ThirdPartyStyleSaver(InMemorySaver):
-            _get_channel_writes_history = (
-                InMemorySaver.__mro__[1]._get_channel_writes_history  # type: ignore[attr-defined]
+            _get_all_delta_channels_writes_history = (
+                InMemorySaver.__mro__[1]._get_all_delta_channels_writes_history  # type: ignore[attr-defined]
             )
-            _aget_channel_writes_history = (
-                InMemorySaver.__mro__[1]._aget_channel_writes_history  # type: ignore[attr-defined]
+            _aget_all_delta_channels_writes_history = (
+                InMemorySaver.__mro__[1]._aget_all_delta_channels_writes_history  # type: ignore[attr-defined]
             )
 
         saver = _ThirdPartyStyleSaver()
@@ -486,7 +482,9 @@ class TestBaseFallbackGetChannelWrites:
             }
         }
 
-        result = saver._get_channel_writes_history(config, "messages")
+        result = saver._get_all_delta_channels_writes_history(config, ["messages"])[
+            "messages"
+        ]
 
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
@@ -503,7 +501,9 @@ class TestBaseFallbackGetChannelWrites:
             }
         }
 
-        result = await saver._aget_channel_writes_history(config, "messages")
+        result = (
+            await saver._aget_all_delta_channels_writes_history(config, ["messages"])
+        )["messages"]
 
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
@@ -512,9 +512,9 @@ class TestBaseFallbackGetChannelWrites:
     async def test_async_fallback_concurrent_tasks_do_not_interfere(self) -> None:
         """Regression: the re-entrancy guard must be task-local, not thread-local.
 
-        Two concurrent `_aget_channel_writes_history` calls on the same
-        event-loop thread must each see their full reconstructed writes. A
-        `threading.local()` guard would let whichever task set it first
+        Two concurrent `_aget_all_delta_channels_writes_history` calls on the
+        same event-loop thread must each see their full reconstructed writes.
+        A `threading.local()` guard would let whichever task set it first
         short-circuit the other to `writes=[]`.
         """
         import asyncio
@@ -542,12 +542,13 @@ class TestBaseFallbackGetChannelWrites:
         }
 
         results = await asyncio.gather(
-            saver._aget_channel_writes_history(config, "messages"),
-            saver._aget_channel_writes_history(config, "messages"),
+            saver._aget_all_delta_channels_writes_history(config, ["messages"]),
+            saver._aget_all_delta_channels_writes_history(config, ["messages"]),
         )
 
         expected_values = [{"content": "first"}, {"content": "second"}]
-        for result in results:
+        for result_map in results:
+            result = result_map["messages"]
             assert result.seed is DELTA_SENTINEL
             values = [v for _, _, v in result.writes]
             assert values == expected_values
@@ -582,9 +583,9 @@ class TestPreDeltaBlobTerminator:
 
         # Pre-delta: cp1 stored a real blob for the channel.
         saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(["A"])
-        # Delta-era: cp2 and cp3 store sentinels; real writes in checkpoint_writes.
-        saver.blobs[(thread_id, ns, channel, v2)] = serde.dumps_typed(DELTA_SENTINEL)
-        saver.blobs[(thread_id, ns, channel, v3)] = serde.dumps_typed(DELTA_SENTINEL)
+        # Delta-era: cp2 and cp3 store "empty"; real writes in checkpoint_writes.
+        saver.blobs[(thread_id, ns, channel, v2)] = ("empty", b"")
+        saver.blobs[(thread_id, ns, channel, v3)] = ("empty", b"")
 
         cp1 = empty_checkpoint()
         cp1["id"] = "cp1"
@@ -634,7 +635,9 @@ class TestPreDeltaBlobTerminator:
             }
         }
 
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
 
         # Seed came from the pre-delta blob at cp1.
         assert result.seed == ["A"]
@@ -656,7 +659,9 @@ class TestPreDeltaBlobTerminator:
             }
         }
 
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
 
         values = [v for _, _, v in result.writes]
         # The pre-delta write under cp1 must not appear (the blob subsumes it).

libs/checkpoint/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/langgraph/langgraph/errors.py

@@ -164,12 +164,11 @@ class NodeError:
     """Exception raised by the failed node."""
 
 
-class NodeTimeoutError(TimeoutError):
+class NodeTimeoutError(Exception):
     """Raised when a node invocation exceeds one of its configured timeouts.
 
-    Subclasses the built-in `TimeoutError`, so existing `except TimeoutError`
-    handlers keep working. If the node has a `retry_policy` whose `retry_on`
-    permits `TimeoutError`, the attempt will be retried.
+    Does **not** inherit from the built-in `TimeoutError` (a subclass of
+    `OSError`) so that the default `RetryPolicy` treats it as retryable.
 
     Both `idle_timeout` and `run_timeout` reflect the configured policy at the
     time of the failure (each is `None` if not configured). `kind` and

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -114,9 +114,12 @@ def channels_from_checkpoint(
 
     For most channels, `spec.from_checkpoint(checkpoint["channel_values"][k])`
     is sufficient. `DeltaChannel` is the exception: sentinel blobs require an
-    ancestor walk via `saver._get_channel_writes_history`. The walk terminates
-    at the nearest `_DeltaSnapshot` blob (step-based) or a pre-migration plain
-    value, so read depth is bounded by `snapshot_frequency`.
+    ancestor walk via `saver._get_all_delta_channels_writes_history`. All
+    delta channels needing replay are batched into a single saver call to
+    save K-1 redundant scans of `checkpoint_writes` (which has no channel
+    index). The walk terminates per-channel at the nearest `_DeltaSnapshot`
+    blob or pre-migration plain value, so read depth is bounded by
+    `snapshot_frequency`.
     """
     channel_specs: dict[str, BaseChannel] = {}
     managed_specs: dict[str, ManagedValueSpec] = {}
@@ -126,18 +129,26 @@ def channels_from_checkpoint(
         else:
             managed_specs[k] = v
 
+    delta_channels: list[str] = [
+        k
+        for k, spec in channel_specs.items()
+        if _needs_replay(spec, checkpoint["channel_values"].get(k, MISSING))
+    ]
+    histories: Mapping[str, Any] = {}
+    if delta_channels and saver is not None and config is not None:
+        histories = saver._get_all_delta_channels_writes_history(config, delta_channels)
+
     channels: dict[str, BaseChannel] = {}
     for k, spec in channel_specs.items():
         ch: BaseChannel
-        stored = checkpoint["channel_values"].get(k, MISSING)
-        if _needs_replay(spec, stored) and saver is not None and config is not None:
+        if k in histories:
             delta_spec = cast(DeltaChannel, spec)
-            history = saver._get_channel_writes_history(config, k)
+            history = histories[k]
             replay_ch = delta_spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:
-            ch = spec.from_checkpoint(stored)
+            ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch
     return channels, managed_specs
 
@@ -158,18 +169,28 @@ async def achannels_from_checkpoint(
         else:
             managed_specs[k] = v
 
+    delta_channels: list[str] = [
+        k
+        for k, spec in channel_specs.items()
+        if _needs_replay(spec, checkpoint["channel_values"].get(k, MISSING))
+    ]
+    histories: Mapping[str, Any] = {}
+    if delta_channels and saver is not None and config is not None:
+        histories = await saver._aget_all_delta_channels_writes_history(
+            config, delta_channels
+        )
+
     channels: dict[str, BaseChannel] = {}
     for k, spec in channel_specs.items():
         ch: BaseChannel
-        stored = checkpoint["channel_values"].get(k, MISSING)
-        if _needs_replay(spec, stored) and saver is not None and config is not None:
+        if k in histories:
             delta_spec = cast(DeltaChannel, spec)
-            history = await saver._aget_channel_writes_history(config, k)
+            history = histories[k]
             replay_ch = delta_spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:
-            ch = spec.from_checkpoint(stored)
+            ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
         channels[k] = ch
     return channels, managed_specs
 

libs/langgraph/langgraph/pregel/main.py

@@ -30,6 +30,7 @@ from typing import (
 )
 from uuid import UUID, uuid5
 
+from langchain_core._api import beta
 from langchain_core.globals import get_debug
 from langchain_core.runnables import (
     RunnableSequence,
@@ -40,7 +41,6 @@ from langchain_core.runnables.config import (
     get_async_callback_manager_for_config,
     get_callback_manager_for_config,
 )
-from langchain_core._event_streaming import _AsyncEventsResult
 from langchain_core.runnables.graph import Graph
 from langchain_core.runnables.schema import StreamEvent
 from langgraph.cache.base import BaseCache
@@ -3449,6 +3449,7 @@ class Pregel(
                 await asyncio.shield(run_manager.on_chain_error(e))
             raise
 
+    @beta(message="The v3 streaming protocol on Pregel is experimental.")
     def _pregel_stream_v3(
         self,
         input: InputT | Command | None,
@@ -3459,7 +3460,12 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3')."""
+        """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3').
+
+        !!! warning
+
+            The v3 streaming protocol is experimental and may change.
+        """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3491,6 +3497,7 @@ class Pregel(
         )
         return GraphRunStream(graph_iter, mux)
 
+    @beta(message="The v3 streaming protocol on Pregel is experimental.")
     async def _apregel_stream_v3(
         self,
         input: InputT | Command | None,
@@ -3501,7 +3508,12 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Internal v3 async streaming implementation. Public entry: astream_events(version='v3')."""
+        """Internal v3 async streaming implementation. Public entry: astream_events(version='v3').
+
+        !!! warning
+
+            The v3 streaming protocol is experimental and may change.
+        """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3554,7 +3566,7 @@ class Pregel(
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any: ...
 
-    def stream_events(  # type: ignore[override]
+    def stream_events(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3570,25 +3582,51 @@ class Pregel(
 
         For `version="v1"` / `"v2"`, yields `StreamEvent` dicts (see
         `Runnable.stream_events`). For `version="v3"`, returns a
-        `GraphRunStream` exposing typed projections.
+        `GraphRunStream` whose typed projections the caller drives by
+        iterating — no background thread.
+
+        !!! warning
+
+            The `version="v3"` API is experimental and may change.
+
+        Builds a `StreamMux` from the built-in transformers, this
+        graph's compile-time `stream_transformers`, and any additional
+        `transformers=` supplied at the call site. `run.output`,
+        `run.interrupted`, and `run.interrupts` work regardless of
+        which transformers are registered.
+
+        Note:
+            Nesting v1 `stream(stream_mode="messages")` inside a node
+            of a `stream_events(version="v3")` run is not fully
+            supported. The outer v3 messages handler reroutes
+            `BaseChatModel.invoke` through the v2 event protocol, so
+            the inner v1 handler does not see `on_llm_new_token`
+            chunks. The inner stream still yields a finalized message
+            via `on_llm_end`. Use `stream_events(version="v3")` for the
+            inner graph as well, or call `chat_model.stream(...)`
+            explicitly, to get token-level streaming.
 
         Args:
             input: Graph input.
             config: Optional runnable config.
             version: Streaming-event schema version. `"v3"` selects the
                 content-block-centric streaming protocol.
-            interrupt_before: Nodes to interrupt before, if any. Only used
-                for `version="v3"`.
-            interrupt_after: Nodes to interrupt after, if any. Only used
-                for `version="v3"`.
-            control: Optional run control. Only used for `version="v3"`.
-            transformers: Extra transformer factories. Only used for
-                `version="v3"`.
+            interrupt_before: Nodes to interrupt before, if any. Only
+                used for `version="v3"`.
+            interrupt_after: Nodes to interrupt after, if any. Only
+                used for `version="v3"`.
+            control: Optional run control used to request cooperative
+                drain. Only used for `version="v3"`.
+            transformers: Extra transformer classes or configured
+                factories appended after compile-time
+                `stream_transformers`. Factories are called as
+                `factory(scope)` so they can propagate to subgraph
+                scopes. Only used for `version="v3"`.
             **kwargs: Forwarded to the v1/v2 path.
 
         Returns:
-            For `version="v3"`, a `GraphRunStream`. Otherwise an
-            `Iterator[StreamEvent]`.
+            For `version="v3"`, a `GraphRunStream` the caller iterates
+            to drive the run. Otherwise an `Iterator[StreamEvent]`.
         """
         if version == "v3":
             return self._pregel_stream_v3(
@@ -3609,7 +3647,7 @@ class Pregel(
         *,
         version: Literal["v1", "v2"] = "v2",
         **kwargs: Any,
-    ) -> _AsyncEventsResult: ...
+    ) -> AsyncIterator[StreamEvent]: ...
 
     @overload
     def astream_events(
@@ -3622,9 +3660,9 @@ class Pregel(
         interrupt_after: All | Sequence[str] | None = None,
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
-    ) -> _AsyncEventsResult: ...
+    ) -> Awaitable[Any]: ...
 
-    def astream_events(  # type: ignore[override]
+    def astream_events(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3635,21 +3673,31 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> _AsyncEventsResult:
-        """Async variant of `stream_events`. See `stream_events` for full docs."""
+    ) -> AsyncIterator[StreamEvent] | Awaitable[Any]:
+        """Async variant of `stream_events`.
+
+        For `version="v3"`, returns an `AsyncGraphRunStream` whose
+        projections can be awaited concurrently; each subscribed cursor
+        drives the pump when its buffer is empty. The same nesting
+        limitation as the sync path applies — see `stream_events` for
+        details.
+
+        !!! warning
+
+            The `version="v3"` API is experimental and may change.
+
+        See `stream_events` for full argument and return documentation.
+        """
         if version == "v3":
-            return _AsyncEventsResult(
-                awaitable=self._apregel_stream_v3(
-                    input,
-                    config,
-                    interrupt_before=interrupt_before,
-                    interrupt_after=interrupt_after,
-                    control=control,
-                    transformers=transformers,
-                )
+            return self._apregel_stream_v3(
+                input,
+                config,
+                interrupt_before=interrupt_before,
+                interrupt_after=interrupt_after,
+                control=control,
+                transformers=transformers,
             )
-        iterator = super().astream_events(input, config, version=version, **kwargs)
-        return _AsyncEventsResult(iterator=iterator)
+        return super().astream_events(input, config, version=version, **kwargs)
 
     @overload
     def invoke(

libs/langgraph/langgraph/stream/run_stream.py

@@ -5,6 +5,8 @@ from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mappin
 from types import MappingProxyType, TracebackType
 from typing import TYPE_CHECKING, Any
 
+from langchain_core._api import beta
+
 from langgraph.stream._convert import convert_to_protocol_event
 from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import ProtocolEvent
@@ -25,6 +27,7 @@ async def _adrive_until_done(pump: Callable[[], Awaitable[bool]]) -> None:
         pass
 
 
+@beta(message="The v3 streaming protocol on Pregel is experimental.")
 class GraphRunStream:
     """Sync run stream with caller-driven pumping.
 
@@ -38,6 +41,11 @@ class GraphRunStream:
     All transformer projections live in `extensions`. Native transformer
     projections (those with `_native = True`) are also set as direct
     attributes on this instance (e.g. `run.values`, `run.messages`).
+
+    !!! warning
+
+        Returned by `Pregel.stream_events(version="v3")`, which is
+        experimental and may change.
     """
 
     def __init__(
@@ -282,6 +290,7 @@ class GraphRunStream:
                 ch._subscribed = False
 
 
+@beta(message="The v3 streaming protocol on Pregel is experimental.")
 class AsyncGraphRunStream:
     """Async run stream with caller-driven pumping.
 
@@ -303,6 +312,11 @@ class AsyncGraphRunStream:
         async for msg in run.messages:
             ...
     ```
+
+    !!! warning
+
+        Awaited from `Pregel.astream_events(version="v3")`, which is
+        experimental and may change.
     """
 
     def __init__(

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.0a1"
+version = "1.2.0a4"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -24,10 +24,10 @@ classifiers = [
     'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.3.2,<2",
-    "langgraph-checkpoint>=4.1.0a1,<5.0.0",
+    "langchain-core>=1.4.0a2,<2",
+    "langgraph-checkpoint>=4.1.0a3,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
-    "langgraph-prebuilt>=1.0.12,<1.1.0",
+    "langgraph-prebuilt>=1.1.0a1,<1.2.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",
 ]
@@ -81,7 +81,6 @@ dev = [
 
 
 [tool.uv.sources]
-langchain-core = { git = "https://github.com/langchain-ai/langchain.git", branch = "nh/streaming-for-alpha-release", subdirectory = "libs/core" }
 langgraph-prebuilt = { path = "../prebuilt", editable = true }
 langgraph-checkpoint = { path = "../checkpoint", editable = true }
 langgraph-checkpoint-sqlite = { path = "../checkpoint-sqlite", editable = true }

libs/langgraph/tests/test_channels.py

@@ -371,8 +371,7 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
 
     saved = saver.get_tuple(config)
     assert saved is not None
-    assert "messages" in saved.checkpoint["channel_values"]
-    assert saved.checkpoint["channel_values"]["messages"] is DELTA_SENTINEL
+    assert "messages" not in saved.checkpoint["channel_values"]
 
     state = graph.get_state(config)
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
@@ -562,7 +561,7 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
 
     saved = saver.get_tuple(config)
     assert saved is not None
-    assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
+    assert "files" not in saved.checkpoint["channel_values"]
     state = graph.get_state(config)
     assert state.values["files"] == {
         "/doc_1.txt": "content for turn 1",

libs/langgraph/tests/test_delta_channel_migration.py

@@ -6,7 +6,7 @@ checkpointer — pre-migration state visible at each *settled* ancestor
 checkpoint is preserved, and post-migration writes fold on top through
 the reducer.
 
-Mechanism under test: the saver's `_get_channel_writes_history(config,
+Mechanism under test: the saver's `_get_all_delta_channels_writes_history(config,
 channel)` walks the parent chain; when it encounters an ancestor whose
 `channel_values[channel]` is a real value (not `DELTA_SENTINEL`), it
 returns that as the `seed`. `DeltaChannel.from_checkpoint(seed)` uses
@@ -29,7 +29,7 @@ Scenarios covered:
    pre-migration seed.
 4. **Base-saver fallback path**: a third-party-style subclass that
    removes the optimized `InMemorySaver` override and falls back to
-   `BaseCheckpointSaver._get_channel_writes_history` must produce the
+   `BaseCheckpointSaver._get_all_delta_channels_writes_history` must produce the
    same result as the optimized path.
 5. **Channel-type isolation across threads**: two threads on the same
    checkpointer under the delta-channel graph — one freshly-started,
@@ -266,7 +266,7 @@ def test_continuing_migrated_thread_folds_deltas_on_seed() -> None:
 
 class _ThirdPartyStyleSaver(InMemorySaver):
     """Simulates a third-party saver that inherits the reference
-    `_get_channel_writes_history` implementation from
+    `_get_all_delta_channels_writes_history` implementation from
     `BaseCheckpointSaver` rather than overriding it.
 
     We rebind the two methods to the base-class versions (via MRO) so
@@ -275,11 +275,11 @@ class _ThirdPartyStyleSaver(InMemorySaver):
     """
 
     # MRO: [_ThirdPartyStyleSaver, InMemorySaver, BaseCheckpointSaver, ...]
-    _get_channel_writes_history = (  # type: ignore[assignment]
-        InMemorySaver.__mro__[1]._get_channel_writes_history  # type: ignore[attr-defined]
+    _get_all_delta_channels_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._get_all_delta_channels_writes_history  # type: ignore[attr-defined]
     )
-    _aget_channel_writes_history = (  # type: ignore[assignment]
-        InMemorySaver.__mro__[1]._aget_channel_writes_history  # type: ignore[attr-defined]
+    _aget_all_delta_channels_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._aget_all_delta_channels_writes_history  # type: ignore[attr-defined]
     )
 
 
@@ -326,7 +326,7 @@ def test_delta_and_migrated_threads_do_not_cross_contaminate() -> None:
     """Two threads sharing a checkpointer — one migrated from
     pre-migration state, one freshly-started under DeltaChannel — must
     maintain independent state. The parent-chain walk in
-    `_get_channel_writes_history` must be scoped to the target thread.
+    `_get_all_delta_channels_writes_history` must be scoped to the target thread.
     """
 
     checkpointer = InMemorySaver()

libs/langgraph/tests/test_interleave_arrival_order.py

@@ -297,7 +297,9 @@ class TestInterleaveArrivalOrder:
 
 class TestInterleaveIntegration:
     def test_interleave_values_and_messages(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         tagged = list(run.interleave("values", "messages"))
         names = [name for name, _ in tagged]
         assert set(names).issubset({"values", "messages"})
@@ -319,7 +321,9 @@ class TestInterleaveIntegration:
             list(run.interleave("alpha"))
 
     def test_interleave_releases_projections_on_completion(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         list(run.interleave("values", "messages"))
         # Subscriptions should be released after the generator completes,
         # so the channels can be re-iterated (they'll be empty / closed).
@@ -327,7 +331,9 @@ class TestInterleaveIntegration:
         assert run.extensions["messages"]._subscribed is False
 
     def test_interleave_releases_projections_on_early_break(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         gen = run.interleave("values", "messages")
         next(gen)
         gen.close()

libs/langgraph/tests/test_pregel_stream_events_v3.py

@@ -396,19 +396,25 @@ class TestStreamChannelNamed:
 
 class TestStreamV2Sync:
     def test_values_projection(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         snapshots = list(run.values)
         assert len(snapshots) >= 1
         last = snapshots[-1]
         assert "A" in last["value"] and "B" in last["value"]
 
     def test_output(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         output = run.output
         assert output == {"value": "xAB", "items": ["a", "b"]}
 
     def test_raw_event_iteration(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         events = list(run)
         assert len(events) > 0
         for event in events:
@@ -418,21 +424,28 @@ class TestStreamV2Sync:
             assert isinstance(event["params"]["timestamp"], int)
 
     def test_extensions_has_native_keys(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         _ = run.output
         assert "values" in run.extensions and "messages" in run.extensions
         assert run.values is run.extensions["values"]
         assert run.messages is run.extensions["messages"]
 
     def test_extensions_is_read_only(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(TypeError):
             run.extensions["new_key"] = object()  # type: ignore[index]
         with pytest.raises(TypeError):
             del run.extensions["values"]  # type: ignore[attr-defined]
 
     def test_custom_stream_events(self) -> None:
-        run = _build_custom_stream_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = _build_custom_stream_graph().stream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CustomPassthroughTransformer],
         )
         custom_events = [e for e in run if e["method"] == "custom"]
         assert len(custom_events) == 2
@@ -448,12 +461,16 @@ class TestStreamV2Sync:
         registering a transformer whose `required_stream_modes`
         includes `"custom"`.
         """
-        run = _build_custom_stream_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_custom_stream_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         custom_events = [e for e in run if e["method"] == "custom"]
         assert custom_events == []
 
     def test_interleave_values_and_messages(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         tagged = list(run.interleave("values", "messages"))
         names = [name for name, _ in tagged]
         assert set(names).issubset({"values", "messages"})
@@ -463,7 +480,9 @@ class TestStreamV2Sync:
         assert run.extensions["messages"]._subscribed is False
 
     def test_abort_marks_exhausted_and_closes_mux(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         values_iter = iter(run.values)
         _ = next(values_iter)
         run.abort()
@@ -472,46 +491,64 @@ class TestStreamV2Sync:
         run.abort()  # idempotent
 
     def test_context_manager_calls_abort_on_exit(self) -> None:
-        with _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3") as run:
+        with _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        ) as run:
             _ = next(iter(run.values))
         assert run._exhausted is True
 
     def test_interleave_unknown_projection(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(KeyError):
             list(run.interleave("values", "does_not_exist"))
 
 
 class TestStreamV2SyncErrors:
     def test_error_propagation_output(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             _ = run.output
 
     def test_error_propagation_values(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             list(run.values)
 
     def test_error_propagation_raw_events(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             list(run)
 
     def test_error_propagation_interrupted(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             _ = run.interrupted
 
     def test_error_propagation_interrupts(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             _ = run.interrupts
 
 
 class TestStreamV2SyncInterrupt:
     def test_interrupted(self) -> None:
-        run = _build_interrupt_graph().stream_events({"value": "x", "items": []}, {"configurable": {"thread_id": "t1"}}, version="v3")
+        run = _build_interrupt_graph().stream_events(
+            {"value": "x", "items": []},
+            {"configurable": {"thread_id": "t1"}},
+            version="v3",
+        )
         _ = run.output
         assert run.interrupted is True
         assert len(run.interrupts) > 0
@@ -526,26 +563,34 @@ class TestStreamV2SyncInterrupt:
 @NEEDS_CONTEXTVARS
 class TestStreamV2Async:
     async def test_values_projection(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         snapshots = [s async for s in run.values]
         assert len(snapshots) >= 1
         last = snapshots[-1]
         assert "A" in last["value"] and "B" in last["value"]
 
     async def test_output(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         output = await run.output()
         assert output == {"value": "xAB", "items": ["a", "b"]}
 
     async def test_raw_event_iteration(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         events = [e async for e in run]
         assert len(events) > 0
         for event in events:
             assert event["type"] == "event"
 
     async def test_abort_marks_exhausted_and_closes_mux(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         values_iter = aiter(run.values)
         _ = await anext(values_iter)
         await run.abort()
@@ -555,20 +600,27 @@ class TestStreamV2Async:
         await run.abort()  # idempotent
 
     async def test_context_manager_calls_abort_on_exit(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         async with run:
             _ = await anext(aiter(run.values))
         assert run._exhausted is True
 
     async def test_extensions_has_native_keys(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         _ = await run.output()
         assert "values" in run.extensions and "messages" in run.extensions
         assert run.values is run.extensions["values"]
         assert run.messages is run.extensions["messages"]
 
     async def test_custom_stream_events(self) -> None:
-        run = await _build_custom_stream_graph().astream_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = await _build_custom_stream_graph().astream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CustomPassthroughTransformer],
         )
         events = [e async for e in run]
         custom_events = [e for e in events if e["method"] == "custom"]
@@ -581,29 +633,39 @@ class TestStreamV2Async:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncErrors:
     async def test_error_propagation_output(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             await run.output()
 
     async def test_error_propagation_values(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             async for _ in run.values:
                 pass
 
     async def test_error_propagation_raw_events(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             async for _ in run:
                 pass
 
     async def test_error_propagation_interrupted(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             await run.interrupted()
 
     async def test_error_propagation_interrupts(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             await run.interrupts()
 
@@ -612,7 +674,11 @@ class TestStreamV2AsyncErrors:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncInterrupt:
     async def test_interrupted(self) -> None:
-        run = await _build_interrupt_graph().astream_events({"value": "x", "items": []}, {"configurable": {"thread_id": "t2"}}, version="v3")
+        run = await _build_interrupt_graph().astream_events(
+            {"value": "x", "items": []},
+            {"configurable": {"thread_id": "t2"}},
+            version="v3",
+        )
         _ = await run.output()
         assert await run.interrupted() is True
         assert len(await run.interrupts()) > 0
@@ -975,7 +1041,8 @@ class TestCustomTransformer:
                     self._channel.push(self._count)
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[CounterTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[CounterTransformer]
         )
         assert "counter" in run.extensions
         counter_iter = iter(run.extensions["counter"])
@@ -1000,7 +1067,8 @@ class TestCustomTransformer:
                     self._log.push("saw_values")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[FooTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[FooTransformer]
         )
         foo_iter = iter(run.foo)
         _ = run.output
@@ -1016,7 +1084,10 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(TypeError, match="pre-built instance"):
-            _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[InstanceTransformer()]
+            _build_simple_graph().stream_events(
+                {"value": "x", "items": []},
+                version="v3",
+                transformers=[InstanceTransformer()],
             )
 
     def test_stream_channel_auto_forward(self) -> None:
@@ -1035,7 +1106,8 @@ class TestCustomTransformer:
                     self._channel.push("emitted")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[EmitterTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[EmitterTransformer]
         )
         custom_events = [e for e in run if e["method"] == "custom:emitter"]
         assert len(custom_events) > 0
@@ -1079,7 +1151,10 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(ValueError, match=r"conflict.*'values'.*ValuesTransformer"):
-            _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[ConflictTransformer]
+            _build_simple_graph().stream_events(
+                {"value": "x", "items": []},
+                version="v3",
+                transformers=[ConflictTransformer],
             )
 
 
@@ -1161,7 +1236,8 @@ class TestStreamChannelAutoLifecycle:
                     self._log.push("got_it")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[MinimalTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[MinimalTransformer]
         )
         minimal_iter = iter(run.extensions["minimal"])
         _ = run.output
@@ -1421,7 +1497,8 @@ class TestAsyncTransformerLane:
             async def afinalize(self) -> None:
                 self._log.close()
 
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3", transformers=[Scorer]
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[Scorer]
         )
         scores_cursor = aiter(run.extensions["scores"])
         _ = await run.output()
@@ -1437,7 +1514,9 @@ class TestAsyncTransformerLane:
 @NEEDS_CONTEXTVARS
 class TestMemoryBounds:
     def test_sync_subscribed_buffer_stays_at_most_one_between_yields(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         events_iter = iter(run)
         max_buffered = 0
         count = 0
@@ -1450,7 +1529,9 @@ class TestMemoryBounds:
         )
 
     def test_unsubscribed_projections_never_accumulate(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         list(run)
         values_log = run.extensions["values"]
         messages_log = run.extensions["messages"]
@@ -1458,19 +1539,25 @@ class TestMemoryBounds:
         assert len(messages_log._items) == 0 and not messages_log._subscribed
 
     def test_output_path_does_not_retain_values(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         _ = run.output
         values_log = run.extensions["values"]
         assert len(values_log._items) == 0 and not values_log._subscribed
 
     def test_drained_subscriber_buffer_returns_to_empty(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         list(run.values)
         assert len(run.extensions["values"]._items) == 0
 
     @pytest.mark.anyio
     async def test_async_single_consumer_buffer_stays_at_most_one(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         max_buffered = 0
         count = 0
         async for _ in run:
@@ -1481,7 +1568,9 @@ class TestMemoryBounds:
 
     @pytest.mark.anyio
     async def test_async_unsubscribed_projections_never_accumulate(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         _ = await run.output()
         values_log = run.extensions["values"]
         messages_log = run.extensions["messages"]

libs/langgraph/tests/test_retry.py

@@ -210,6 +210,15 @@ def test_should_retry_default_retry_on():
     req_error_no_resp.response = None
     assert _should_retry_on(policy, req_error_no_resp) is True
 
+    # NodeTimeoutError should be retryable by default
+    assert (
+        _should_retry_on(
+            policy,
+            NodeTimeoutError("node", 1.0, kind="run", run_timeout=0.5),
+        )
+        is True
+    )
+
     # Should retry on other exceptions by default
     class CustomException(Exception):
         pass
@@ -1456,14 +1465,14 @@ async def test_state_graph_add_node_timeout_composes_with_retry():
     async def flaky(state: _TimeoutState) -> _TimeoutState:
         attempts.append(len(attempts))
         if len(attempts) < 2:
-            await asyncio.sleep(0.5)
+            await asyncio.sleep(1.0)
         return {"x": state["x"] + 1}
 
     builder = StateGraph(_TimeoutState)
     builder.add_node(
         "flaky",
         flaky,
-        timeout=TimeoutPolicy(idle_timeout=0.1),
+        timeout=TimeoutPolicy(idle_timeout=0.3),
         retry_policy=RetryPolicy(
             max_attempts=3,
             initial_interval=0.0,
@@ -1758,6 +1767,212 @@ async def test_arun_with_retry_timeout_observer_treats_bubble_up_as_non_error():
     assert finish.error_message is None
 
 
+# ---------------------------------------------------------------------------
+# Watcher invariant: any timeout that retry/error_handler can recover from
+# MUST emit `finish=error` BEFORE the in-process recovery work happens. The
+# external watchdog (langgraph-api) relies on this so it only kills a worker
+# when no `finish` arrives within the deadline. The tests below pin down the
+# three recovery paths so a refactor that moves `_finish_timed_attempt` past
+# an `await` (or past the final `raise`) trips CI.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_arun_with_retry_observer_emits_finish_before_retry_backoff():
+    """`finish=error` of attempt N must arrive before the retry backoff sleep."""
+    timeline: list[tuple[float, Any]] = []
+
+    class TimingOutOnceProc:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def ainvoke(self, input, config):
+            self.calls += 1
+            if self.calls == 1:
+                await asyncio.sleep(1.0)
+            return "ok"
+
+    backoff = 0.25
+    policy = RetryPolicy(
+        max_attempts=2,
+        initial_interval=backoff,
+        backoff_factor=1.0,
+        max_interval=backoff,
+        jitter=False,
+        retry_on=NodeTimeoutError,
+    )
+    task = _make_task(
+        TimingOutOnceProc(),
+        timeout=_idle_timeout(0.05),
+        retry_policy=(policy,),
+        name="backoff_watcher",
+    )
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = lambda ev: timeline.append(
+        (time.monotonic(), ev)
+    )
+
+    assert await arun_with_retry(task, retry_policy=None) == "ok"
+
+    starts = [(t, ev) for t, ev in timeline if ev.event == "start"]
+    finishes = [(t, ev) for t, ev in timeline if ev.event == "finish"]
+    assert [ev.context.attempt for _, ev in starts] == [1, 2]
+    assert [ev.status for _, ev in finishes] == ["error", "success"]
+
+    first_finish_t = finishes[0][0]
+    second_start_t = starts[1][0]
+
+    # The watcher relies on this gap: `finish=error` for attempt 1 must arrive
+    # before `arun_with_retry` enters `await asyncio.sleep(backoff)`. We give a
+    # generous slack to keep this stable on slow CI; the structural invariant
+    # is "finish lands first", not "the gap equals exactly backoff".
+    assert second_start_t - first_finish_t >= backoff * 0.5, (
+        f"finish=error appears to be emitted after retry backoff sleep; "
+        f"gap was {second_start_t - first_finish_t:.3f}s, expected >= {backoff * 0.5:.3f}s"
+    )
+
+
+@pytest.mark.anyio
+async def test_state_graph_observer_emits_finish_before_error_handler_start():
+    """Original task's `finish=error` must arrive before the error_handler task's `start`."""
+
+    class State(TypedDict):
+        foo: str
+
+    async def slow_node(state: State) -> State:
+        await asyncio.sleep(1.0)
+        return {"foo": "should-not-happen"}
+
+    async def handler_node(state: State, error: NodeError) -> State:
+        return {"foo": "handled"}
+
+    events: list = []
+    graph = (
+        StateGraph(State)
+        .add_node(
+            "slow",
+            slow_node,
+            timeout=TimeoutPolicy(idle_timeout=0.05),
+            error_handler=handler_node,
+        )
+        .add_edge(START, "slow")
+        .compile()
+    )
+
+    result = await graph.ainvoke(
+        {"foo": ""},
+        config={
+            "configurable": {CONFIG_KEY_TIMED_ATTEMPT_OBSERVER: events.append},
+        },
+    )
+    assert result["foo"] == "handled"
+
+    # Filter to events from the failing node only — the handler node has no
+    # timeout configured here, so it doesn't appear in the observer stream.
+    slow_events = [ev for ev in events if ev.context.task_name == "slow"]
+    starts = [ev for ev in slow_events if ev.event == "start"]
+    finishes = [ev for ev in slow_events if ev.event == "finish"]
+    assert len(starts) == 1
+    assert len(finishes) == 1
+    assert finishes[0].status == "error"
+    assert finishes[0].error_type == "NodeTimeoutError"
+    # The slow task's finish-error event must precede every event for any
+    # follow-up task in the same observer stream.
+    slow_finish_index = events.index(finishes[0])
+    for ev in events[slow_finish_index + 1 :]:
+        assert ev.context.task_name == "slow" or ev.event == "start", (
+            f"unexpected event {ev.event} for {ev.context.task_name} "
+            f"after slow's finish=error"
+        )
+
+
+@pytest.mark.anyio
+async def test_arun_with_retry_observer_emits_finish_before_final_raise_on_exhaustion():
+    """When retry exhausts and the timeout propagates, the final `finish=error` must
+    be emitted before `arun_with_retry` re-raises."""
+    events: list = []
+
+    class AlwaysTimingOutProc:
+        async def ainvoke(self, input, config):
+            await asyncio.sleep(1.0)
+            return "never"
+
+    policy = RetryPolicy(
+        max_attempts=2,
+        initial_interval=0.0,
+        jitter=False,
+        retry_on=NodeTimeoutError,
+    )
+    task = _make_task(
+        AlwaysTimingOutProc(),
+        timeout=_idle_timeout(0.05),
+        retry_policy=(policy,),
+        name="never_succeeds",
+    )
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    with pytest.raises(NodeTimeoutError):
+        await arun_with_retry(task, retry_policy=None)
+
+    starts = [ev for ev in events if ev.event == "start"]
+    finishes = [ev for ev in events if ev.event == "finish"]
+    assert [ev.context.attempt for ev in starts] == [1, 2]
+    assert [ev.context.attempt for ev in finishes] == [1, 2]
+    assert [ev.status for ev in finishes] == ["error", "error"]
+    assert all(ev.error_type == "NodeTimeoutError" for ev in finishes)
+    # Both finish events were observed BEFORE arun_with_retry raised, otherwise
+    # the `with pytest.raises` block would have exited before `events` got
+    # populated with the second finish.
+
+
+@pytest.mark.anyio
+async def test_sync_sleep_in_async_node_bypasses_timeout_and_emits_finish_success():
+    """Sync `time.sleep` inside an async node blocks the event loop so the
+    in-process watchdog cannot fire. We document the resulting behavior here:
+
+    1. `NodeTimeoutError` is NOT raised, even though the sync sleep exceeds
+       `idle_timeout`.
+    2. The node's normal return value flows through.
+    3. `finish=success` is emitted to the observer.
+
+    This is the canonical case where the in-process timeout is defeated and
+    the only safety net is the external watcher (langgraph-api), which
+    SIGKILLs the worker when no `finish` arrives within its deadline. The
+    catch is that with a *short* sync sleep the event loop unblocks before
+    the watcher's deadline expires, so the watcher legitimately does not
+    kill — meaning the configured `idle_timeout` is silently honored at the
+    process level only when the block is long enough to outlast the
+    watcher's grace.
+
+    This is the documented "Cooperative cancellation" caveat on
+    `TimeoutPolicy`. The test pins the behavior so any future change that
+    starts raising `NodeTimeoutError` for sync-blocked async nodes (or stops
+    emitting `finish=success`) is caught.
+    """
+    events: list = []
+
+    class SyncSleepingProc:
+        async def ainvoke(self, input, config):
+            time.sleep(0.1)
+            return "completed_despite_timeout"
+
+    task = _make_task(
+        SyncSleepingProc(),
+        timeout=_idle_timeout(0.05),
+        name="sync_sleeper",
+    )
+    task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
+
+    result = await arun_with_retry(task, retry_policy=None)
+
+    assert result == "completed_despite_timeout"
+    starts = [ev for ev in events if ev.event == "start"]
+    finishes = [ev for ev in events if ev.event == "finish"]
+    assert len(starts) == 1
+    assert len(finishes) == 1
+    assert finishes[0].status == "success"
+    assert finishes[0].error_type is None
+
+
 def test_graph_error_handler_runs_after_retry_exhaustion():
     class State(TypedDict):
         foo: str
@@ -1995,7 +2210,6 @@ def test_graph_error_handler_does_not_swallow_interrupt_concurrent():
     )
 
 
-
 def test_node_error_handlers_route_to_matching_handler():
     class State(TypedDict):
         route: str
@@ -2053,4 +2267,3 @@ def test_node_without_error_handler_still_fails_run():
 
     with pytest.raises(ValueError, match="no handler"):
         graph.invoke({"foo": ""})
-

libs/langgraph/tests/test_stream_data_transformers.py

@@ -506,7 +506,8 @@ def _make_simple_graph() -> Any:
 def test_stream_events_v3_custom_projection_opt_in() -> None:
     """run.custom surfaces get_stream_writer() payloads when opted in."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
+    run = graph.stream_events(
+        {"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -517,7 +518,8 @@ def test_stream_events_v3_custom_projection_opt_in() -> None:
 def test_stream_events_v3_custom_and_values_coexist() -> None:
     """Both run.custom and run.values work in the same run."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
+    run = graph.stream_events(
+        {"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -529,7 +531,9 @@ def test_stream_events_v3_custom_and_values_coexist() -> None:
 def test_stream_events_v3_tasks_projection_opt_in() -> None:
     """run.tasks surfaces raw task events when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, transformers=[TasksTransformer], version="v3")
+    run = graph.stream_events(
+        {"value": "x", "items": []}, transformers=[TasksTransformer], version="v3"
+    )
 
     tasks_events = list(run.tasks)
     assert len(tasks_events) >= 1
@@ -540,7 +544,9 @@ def test_stream_events_v3_tasks_projection_opt_in() -> None:
 def test_stream_events_v3_debug_projection_opt_in() -> None:
     """run.debug surfaces debug events when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, transformers=[DebugTransformer], version="v3")
+    run = graph.stream_events(
+        {"value": "x", "items": []}, transformers=[DebugTransformer], version="v3"
+    )
 
     debug_events = list(run.debug)
     assert len(debug_events) >= 1
@@ -551,7 +557,8 @@ def test_stream_events_v3_debug_projection_opt_in() -> None:
 def test_stream_events_v3_updates_projection_opt_in() -> None:
     """run.updates surfaces node output dicts when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[UpdatesTransformer]
+    run = graph.stream_events(
+        {"value": "x", "items": []}, version="v3", transformers=[UpdatesTransformer]
     )
 
     updates = list(run.updates)
@@ -563,7 +570,10 @@ def test_stream_events_v3_updates_projection_opt_in() -> None:
 def test_stream_events_v3_all_transformers_interleaved() -> None:
     """All five transformers registered together, consumed via interleave."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[
+    run = graph.stream_events(
+        {"value": "x", "items": []},
+        version="v3",
+        transformers=[
             CustomTransformer,
             UpdatesTransformer,
             CheckpointsTransformer,
@@ -604,7 +614,10 @@ def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", config={"configurable": {"thread_id": "test-all"}},
+    run = graph.stream_events(
+        {"value": "x", "items": []},
+        version="v3",
+        config={"configurable": {"thread_id": "test-all"}},
         transformers=[
             CustomTransformer,
             UpdatesTransformer,
@@ -640,7 +653,10 @@ def test_stream_events_v3_checkpoints_projection_opt_in() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", config={"configurable": {"thread_id": "test-ckpt-standalone"}},
+    run = graph.stream_events(
+        {"value": "x", "items": []},
+        version="v3",
+        config={"configurable": {"thread_id": "test-ckpt-standalone"}},
         transformers=[CheckpointsTransformer],
     )
 
@@ -679,7 +695,10 @@ def test_tasks_and_lifecycle_coregistration_e2e() -> None:
     is present and suppressing them from the main log.
     """
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[TasksTransformer],
+    run = graph.stream_events(
+        {"value": "x", "items": []},
+        version="v3",
+        transformers=[TasksTransformer],
     )
 
     tasks_events = list(run.tasks)

libs/langgraph/tests/test_stream_events_v3_e2e.py

@@ -285,11 +285,15 @@ class TestStreamV2E2ESync:
 
     def test_output_matches_final_values_snapshot(self) -> None:
         """output property returns the same state as the last values snapshot."""
-        run1 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run1 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         snapshots = list(run1.values)
         final_via_values = snapshots[-1]
 
-        run2 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run2 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         final_via_output = run2.output
 
         assert final_via_values == final_via_output
@@ -407,7 +411,10 @@ class TestStreamV2E2ECustom:
         """Custom StreamWriter events appear on the main log when a
         transformer declares the custom mode."""
         graph = _make_custom_writer_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = graph.stream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CustomPassthroughTransformer],
         )
         events = list(run)
         custom = [e for e in events if e["method"] == "custom"]
@@ -426,7 +433,10 @@ class TestStreamV2E2ECustom:
     def test_custom_transformer_with_stream_channel(self) -> None:
         """A custom transformer with a StreamChannel produces extension data."""
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CounterTransformer],
         )
 
         assert "counter" in run.extensions
@@ -440,7 +450,10 @@ class TestStreamV2E2ECustom:
     def test_custom_channel_events_on_main_log(self) -> None:
         """StreamChannel auto-forward injects custom:<name> events into the main log."""
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CounterTransformer],
         )
         events = list(run)
         counter_events = [e for e in events if e["method"] == "custom:counter"]
@@ -577,7 +590,9 @@ class TestStreamV2E2EAsync:
         """Async interrupted run has correct flags."""
         graph = _make_interrupt_graph()
         config: dict[str, Any] = {"configurable": {"thread_id": "async-int-1"}}
-        run = await graph.astream_events({"value": "x", "items": []}, config, version="v3")
+        run = await graph.astream_events(
+            {"value": "x", "items": []}, config, version="v3"
+        )
 
         output = await run.output()
         assert output is not None
@@ -612,7 +627,10 @@ class TestStreamV2E2EAsync:
     async def test_async_custom_transformer(self) -> None:
         """Async custom transformer with StreamChannel works."""
         graph = _make_nested_graph()
-        run = await graph.astream_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = await graph.astream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CounterTransformer],
         )
         assert "counter" in run.extensions
         counter_cursor = aiter(run.extensions["counter"])
@@ -659,7 +677,10 @@ class TestStreamV2E2ECombined:
                 return True
 
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer, TagTransformer],
+        run = graph.stream_events(
+            {"value": "x", "items": []},
+            version="v3",
+            transformers=[_CounterTransformer, TagTransformer],
         )
 
         assert "counter" in run.extensions
@@ -730,13 +751,17 @@ class TestStreamV2E2ECombined:
 
     def test_lifecycle_matches_subgraph_handles(self) -> None:
         """Lifecycle events and subgraph handles agree on discovered subgraphs."""
-        run1 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run1 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         handle_paths: list[tuple[str, ...]] = []
         for handle in run1.subgraphs:
             list(handle.values)
             handle_paths.append(handle.path)
 
-        run2 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run2 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         lifecycle = list(run2.lifecycle)
 
         started_ns = [
@@ -763,7 +788,10 @@ class TestStreamV2E2ECombined:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "hi"}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_events(
+            {"messages": "hi"},
+            version="v3",
+            transformers=[_CounterTransformer],
         )
 
         counter_iter = iter(run.extensions["counter"])

libs/langgraph/tests/test_stream_messages_transformer.py

@@ -760,7 +760,9 @@ class TestDirectMessagesModeStaysV1:
             == "legacy path"
         )
 
-    def test_nested_graph_stream_messages_stays_v1_under_outer_stream_events_v3(self) -> None:
+    def test_nested_graph_stream_messages_stays_v1_under_outer_stream_events_v3(
+        self,
+    ) -> None:
         """An outer `stream_events(version="v3")` run must not flip an inner direct
         `stream_mode="messages"` call onto the v2 event protocol."""
         model = GenericFakeChatModel(messages=iter(["nested legacy path"]))

libs/langgraph/uv.lock

@@ -1348,8 +1348,8 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
-source = { git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release#ad4d43f38827e2aadce559e7e676b5eb4184a976" }
+version = "1.4.0a2"
+source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
     { name = "langchain-protocol" },
@@ -1361,6 +1361,10 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
+sdist = { url = "https://files.pythonhosted.org/packages/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
+]
 
 [[package]]
 name = "langchain-protocol"
@@ -1376,7 +1380,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a1"
+version = "1.2.0a4"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1448,7 +1452,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1460,7 +1464,7 @@ requires-dist = [
 dev = [
     { name = "httpx" },
     { name = "jupyter" },
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.0.0" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1493,7 +1497,7 @@ lint = [
 ]
 test = [
     { name = "httpx" },
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.0.0" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1557,7 +1561,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1605,7 +1609,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a1"
+version = "3.1.0a3"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -1751,7 +1755,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.12"
+version = "1.1.0a1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "1.0.12"
+version = "1.1.0a1"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.10"

libs/prebuilt/tests/test_tool_call_transformer.py

@@ -147,7 +147,10 @@ class TestToolCallTransformerUnit:
         mux.push(_tool_event("tool-output-delta", "a", delta="A1"))
         mux.push(_tool_event("tool-output-delta", "b", delta="B1"))
         mux.push(_tool_event("tool-output-delta", "a", delta="A2"))
-        assert _unstamped(transformer._active["a"]._output_deltas._items) == ["A1", "A2"]
+        assert _unstamped(transformer._active["a"]._output_deltas._items) == [
+            "A1",
+            "A2",
+        ]
         assert _unstamped(transformer._active["b"]._output_deltas._items) == ["B1"]
 
     def test_tools_event_passes_through_main_log(self) -> None:
@@ -199,7 +202,9 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [streamer])
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_events(
+            {"messages": []}, transformers=[ToolCallTransformer], version="v3"
+        )
 
         tool_calls: list[ToolCallStream] = []
         for tc in run.tool_calls:
@@ -239,7 +244,9 @@ class TestToolCallTransformerEndToEnd:
         assert "tool_calls" not in run_no_tc._mux.extensions  # type: ignore[attr-defined]
 
         # With ToolCallTransformer, the projection is present.
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_events(
+            {"messages": []}, transformers=[ToolCallTransformer], version="v3"
+        )
         assert "tool_calls" in run._mux.extensions  # type: ignore[attr-defined]
         # Drain so the run closes cleanly.
         list(run.tool_calls)
@@ -266,7 +273,8 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [astreamer])
-        run = await graph.astream_events({"messages": []}, version="v3", transformers=[ToolCallTransformer]
+        run = await graph.astream_events(
+            {"messages": []}, version="v3", transformers=[ToolCallTransformer]
         )
 
         collected: list[ToolCallStream] = []
@@ -295,7 +303,9 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [boom])
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_events(
+            {"messages": []}, transformers=[ToolCallTransformer], version="v3"
+        )
 
         collected: list[ToolCallStream] = []
         with pytest.raises(ValueError, match="nope"):

libs/prebuilt/uv.lock

@@ -249,7 +249,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -262,26 +262,26 @@ 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 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" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.12"
+version = "0.0.14"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" }
+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/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" },
+    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a1"
+version = "1.2.0a4"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -294,7 +294,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "." },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -365,7 +365,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -413,7 +413,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a1"
+version = "3.1.0a3"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -503,7 +503,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.12"
+version = "1.1.0a1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/uv.lock

@@ -266,7 +266,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -279,26 +279,26 @@ 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 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" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.12"
+version = "0.0.14"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" }
+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/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" },
+    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a1"
+version = "1.2.0a4"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -311,7 +311,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "." },
@@ -382,7 +382,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -430,7 +430,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.12"
+version = "1.1.0a1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },