docs(langgraph): rewrite stream_v2 references to stream_events v3

Update all docstrings and comments in production code that referred to
stream_v2/astream_v2 to use stream_events(version="v3") /
astream_events(version="v3") instead.

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, Mapping, Sequence
+from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
 from typing import Any, cast
 
@@ -19,7 +19,6 @@ 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
@@ -27,10 +26,9 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_STAGE2_SQL,
+    SELECT_DELTA_COMBINED_SQL,
     BasePostgresSaver,
-    _build_delta_stage1_sql,
-    _DeltaStage2Row,
+    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -310,12 +308,7 @@ class PostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            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)):
+            if v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -443,80 +436,46 @@ class PostgresSaver(BasePostgresSaver):
                 with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    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`.
+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
 
-        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).
+        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.
         """
-        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 {
-                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
-                    for ch in channels
-                }
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
             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_STAGE2_SQL,
+                SELECT_DELTA_COMBINED_SQL,
                 (
+                    channel,
                     thread_id,
                     checkpoint_ns,
-                    channels,
-                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channels,
-                    union_seed_versions,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
                 ),
             )
-            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),
+            rows = cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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, Mapping, Sequence
+from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
 from typing import Any, cast
 
@@ -19,7 +19,6 @@ 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
@@ -27,10 +26,9 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_STAGE2_SQL,
+    SELECT_DELTA_COMBINED_SQL,
     BasePostgresSaver,
-    _build_delta_stage1_sql,
-    _DeltaStage2Row,
+    _DeltaCombinedRow,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -269,12 +267,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            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)):
+            if v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -404,69 +397,44 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    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`.
+    async def _aget_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
 
-        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.
+        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`.
         """
-        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 {
-                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
-                    for ch in channels
-                }
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
             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_STAGE2_SQL,
+                SELECT_DELTA_COMBINED_SQL,
                 (
+                    channel,
                     thread_id,
                     checkpoint_ns,
-                    channels,
-                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channels,
-                    union_seed_versions,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
                 ),
             )
-            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),
+            rows = await cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import random
 import warnings
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from importlib.metadata import version as get_version
 from typing import Any, TypedDict, cast
 
@@ -156,100 +156,62 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """
 
 
-class _DeltaStage2Row(TypedDict, total=False):
-    """One row from `SELECT_DELTA_STAGE2_SQL` (a UNION ALL of writes and blobs)."""
+class _DeltaCombinedRow(TypedDict, total=False):
+    """One row from `SELECT_DELTA_COMBINED_SQL` (a UNION ALL of three tables).
 
-    _kind: str  # "w" or "b"
-    checkpoint_id: str | None  # "w" rows only
-    channel: str | None  # set on both "w" and "b" rows
+    Every row carries `_kind` ("p" / "w" / "b") plus whichever columns are
+    relevant for that kind; irrelevant columns are NULL and typed as `None`.
+    """
+
+    _kind: str  # always present: "p", "w", or "b"
+    # checkpoint row ("p")
+    checkpoint_id: str | None
+    parent_checkpoint_id: str | None
+    ver: str | None
+    # write / blob rows ("w", "b")
     type: str | None
     blob: bytes | None
-    task_id: str | None  # "w" rows only
-    idx: int | None  # "w" rows only
-    version: str | None  # "b" rows only
+    # write row only ("w")
+    task_id: str | None
+    idx: int | None
+    # blob row only ("b")
+    version: str | None
 
 
-# Multi-channel two-stage DeltaChannel reconstruction.
+# 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`.
 #
-# 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 = ANY(%s)
-      AND checkpoint_id = ANY(%s)
+# 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 'b', NULL, channel,
+    SELECT 'w',
+           checkpoint_id, NULL, NULL,
+           type, blob, task_id, idx, NULL
+    FROM checkpoint_writes
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    UNION ALL
+    SELECT 'b',
+           NULL, NULL, NULL,
            type, blob, NULL, NULL, version
     FROM checkpoint_blobs
-    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = ANY(%s)
-      AND version = ANY(%s)
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %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
@@ -292,120 +254,82 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             if t.decode() != "empty"
         }
 
-    @staticmethod
-    def _walk_stage1_multi(
-        stage1_rows: Sequence[Mapping[str, Any]],
-        target_id: str,
-        channels: Sequence[str],
-    ) -> tuple[dict[str, list[str]], dict[str, str | None]]:
-        """Walk the parent chain once for all requested channels.
-
-        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`.
-
-        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] = {}
-        # 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 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(
+    def _build_delta_channel_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.
+        channel: str,
+        target_id: str,
+        rows: Sequence[_DeltaCombinedRow],
+    ) -> _ChannelWritesHistory:
+        """Reconstruct one delta channel's history from the combined UNION ALL rows.
 
-        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.
+        Pure data transform shared by sync (`PostgresSaver`) and async
+        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
+        and feed the tagged rows here.
+
+        Walk is newest → oldest from the target's parent. A non-sentinel
+        blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
+        walk and is returned as the seed so replay starts from it.
+
+        Writes stored at `target_id` itself are pending writes for the next
+        step and are excluded — the walk begins at the target's parent.
         """
-        # 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]] = {}
+        parent_of: dict[str, str | None] = {}
+        ver_of: dict[str, str | None] = {}
+        writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
+        blob_by_ver: dict[str, tuple[str, bytes]] = {}
 
-        for r in stage2_rows:
-            ch = cast(str, r["channel"])
+        for r in rows:
             kind = r["_kind"]
-            if kind == "w":
+            if kind == "p":
                 cid = cast(str, r["checkpoint_id"])
-                writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
+                parent_of[cid] = r["parent_checkpoint_id"]
+                ver_of[cid] = r["ver"]
+            elif kind == "w":
+                cid = cast(str, r["checkpoint_id"])
+                writes_by_cid.setdefault(cid, []).append(
                     cast(
                         "tuple[str, bytes, str, int]",
                         (r["type"], r["blob"], r["task_id"], r["idx"]),
                     )
                 )
             else:  # kind == "b"
-                ver = cast(str, r["version"])
-                seed_blob_by_ver[(ch, ver)] = cast(
+                blob_by_ver[cast(str, r["version"])] = cast(
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # 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)
+        # 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)
 
-        result: dict[str, _ChannelWritesHistory] = {}
-        for ch in channels:
-            chain_cids = chain_by_ch.get(ch, [])
-            seed_version = seed_ver_by_ch.get(ch)
+        ancestors: list[str] = []
+        cur_cid: str | None = parent_of.get(target_id)
+        while cur_cid is not None:
+            ancestors.append(cur_cid)
+            cur_cid = parent_of.get(cur_cid)
+        if not ancestors:
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
 
-            collected: list[PendingWrite] = []
-            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: 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)
 
-            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
+        collected.reverse()  # oldest → newest
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
     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.0a3"
+version = "3.1.0a1"
 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.0a3,<5.0.0",
+  "langgraph-checkpoint>=4.1.0a1,<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.0a3"
+version = "4.1.0a1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -307,7 +307,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a3"
+version = "3.1.0a1"
 source = { editable = "." }
 dependencies = [
     { name = "langgraph-checkpoint" },

libs/checkpoint-sqlite/uv.lock

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

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

@@ -125,8 +125,7 @@ class CheckpointTuple(NamedTuple):
 
 
 class _ChannelWritesHistory(NamedTuple):
-    """Result of `BaseCheckpointSaver._get_all_delta_channels_writes_history`
-    (a per-channel entry from the returned mapping).
+    """Result of `BaseCheckpointSaver._get_channel_writes_history`.
 
     Storage-level view of what one channel wrote across the ancestor chain
     of a target checkpoint:
@@ -488,12 +487,12 @@ class BaseCheckpointSaver(Generic[V]):
         raise NotImplementedError
 
     def _get_tuple_raw(self, config: RunnableConfig) -> CheckpointTuple | None:
-        """Pure storage read used by `_get_all_delta_channels_writes_history`.
+        """Pure storage read used by `_get_channel_writes_history`.
 
         Must return the same value as `get_tuple` but must NOT trigger channel
         reconstruction; otherwise the channel-hydration path would re-enter
-        `_get_all_delta_channels_writes_history`. Override only if `get_tuple`
-        itself performs channel hydration.
+        `_get_channel_writes_history`. Override only if `get_tuple` itself
+        performs channel hydration.
         """
         return self.get_tuple(config)
 
@@ -501,17 +500,14 @@ class BaseCheckpointSaver(Generic[V]):
         """Async version of `_get_tuple_raw`. See docstring there."""
         return await self.aget_tuple(config)
 
-    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.
+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """**Experimental.** Query one channel's writes along the parent chain.
 
-        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`:
+        Storage-level query, not channel semantics: returns `(seed, writes)`
+        reflecting what storage knows about a single channel across the
+        ancestor chain of the target checkpoint identified by `config`.
 
         * `writes` — on-path deltas oldest→newest as `PendingWrite` tuples.
           Writes stored at the target `checkpoint_id` itself are pending
@@ -524,86 +520,69 @@ 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` 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.
+        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.
 
-        Empty `channels` returns `{}`. Underscore-prefixed because the
-        method surface is experimental.
+        Underscore-prefixed because the method surface is experimental.
         """
-        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)
+        collected: list[PendingWrite] = []  # newest first; reversed at the end
         target_tuple = self._get_tuple_raw(config)
         cursor_config: RunnableConfig | None = (
             target_tuple.parent_config if target_tuple else None
         )
-        while cursor_config is not None and remaining:
+        while cursor_config is not None:
             tup = self._get_tuple_raw(cursor_config)
             if tup is None:
                 break
-            # Collect each ancestor's writes for any channel still searching.
+            # Collect this ancestor's writes FIRST — they encode the
+            # transition from this ancestor's state to its child's, so
+            # they must be included whether or not this ancestor is the
+            # seed terminator.
             if tup.pending_writes:
+                # Within a superstep, pending_writes are oldest→newest;
+                # reverse to scan newest-first.
                 for write in reversed(tup.pending_writes):
-                    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)
+                    if write[1] != channel:
+                        continue
+                    collected.append(write)
+            # Seed terminator: any non-sentinel blob on an ancestor
+            # establishes the reconstruction base. Stop here.
+            ancestor_value = tup.checkpoint["channel_values"].get(channel)
+            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                collected.reverse()
+                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
             cursor_config = tup.parent_config
-        return {
-            ch: _ChannelWritesHistory(
-                seed=seed_by_ch[ch],
-                writes=list(reversed(collected_by_ch[ch])),
-            )
-            for ch in channels
-        }
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
-    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)
+    async def _aget_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Async version of `_get_channel_writes_history`. See docstring there."""
+        collected: list[PendingWrite] = []
         target_tuple = await self._aget_tuple_raw(config)
         cursor_config: RunnableConfig | None = (
             target_tuple.parent_config if target_tuple else None
         )
-        while cursor_config is not None and remaining:
+        while cursor_config is not None:
             tup = await self._aget_tuple_raw(cursor_config)
             if tup is None:
                 break
             if tup.pending_writes:
                 for write in reversed(tup.pending_writes):
-                    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)
+                    if write[1] != channel:
+                        continue
+                    collected.append(write)
+            ancestor_value = tup.checkpoint["channel_values"].get(channel)
+            if ancestor_value is not None and ancestor_value is not DELTA_SENTINEL:
+                collected.reverse()
+                return _ChannelWritesHistory(seed=ancestor_value, writes=collected)
             cursor_config = tup.parent_config
-        return {
-            ch: _ChannelWritesHistory(
-                seed=seed_by_ch[ch],
-                writes=list(reversed(collected_by_ch[ch])),
-            )
-            for ch in channels
-        }
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
     def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.

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

@@ -6,7 +6,7 @@ import pickle
 import random
 import shutil
 from collections import defaultdict
-from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
+from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack
 from types import TracebackType
 from typing import Any
@@ -141,24 +141,17 @@ class InMemorySaver(
             result[k] = self.serde.loads_typed(vv)
         return result
 
-    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 {}
+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = config["configurable"].get("checkpoint_id", "")
         ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
-
-        # Build the parent chain (newest→oldest), skipping the target.
+        # Walk the parent chain newest→oldest. Skip the target itself —
+        # writes stored AT `checkpoint_id` are pending for the next step
+        # (pregel applies them via `apply_writes`; they aren't part of the
+        # snapshot value AT `checkpoint_id`).
         chain: list[str] = []
         target_entry = ns_storage.get(checkpoint_id)
         current: str | None = target_entry[2] if target_entry is not None else None
@@ -169,73 +162,77 @@ class InMemorySaver(
             chain.append(current)
             _, _, parent = entry
             current = parent
-
-        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)
-
+        # 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
-            if not remaining:
-                break
             entry = ns_storage.get(cp_id)
-            ckpt = self.serde.loads_typed(entry[0]) if entry is not None else None
+            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
+                            )
 
-            # 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 not in remaining:
+                if ch != channel:
                     continue
-                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))
-                )
+                val = self.serde.loads_typed(serialized)
+                collected.append((tid, ch, val))
+        collected.reverse()
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
 
-            # 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)
+    async def _aget_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        return self._get_channel_writes_history(config, channel)
 
     def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the in-memory storage.
@@ -455,9 +452,7 @@ 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 and values[k] is not DELTA_SENTINEL
-                else ("empty", b"")
+                self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
             )
         self.storage[thread_id][checkpoint_ns].update(
             {

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

@@ -34,7 +34,9 @@ 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
@@ -320,11 +322,14 @@ 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,
@@ -651,7 +656,9 @@ def _create_msgpack_ext_hook(
         return False
 
     def ext_hook(code: int, data: bytes) -> Any:
-        if code == EXT_DELTA_SNAPSHOT:
+        if code == EXT_DELTA_SENTINEL:
+            return DELTA_SENTINEL
+        elif code == EXT_DELTA_SNAPSHOT:
             return _DeltaSnapshot(
                 ormsgpack.unpackb(
                     data, ext_hook=ext_hook, option=ormsgpack.OPT_NON_STR_KEYS

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

@@ -17,10 +17,12 @@ TASKS = "__pregel_tasks"
 
 
 class _DeltaSentinel:
-    """In-memory marker for a DeltaChannel field with no snapshot.
+    """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.
 
-    Never serialized to storage — checkpointers strip it before writing.
-    Compare with `is DELTA_SENTINEL`; always the same module-level instance.
+    Compare with `is DELTA_SENTINEL` — `loads_typed` always returns the same
+    module-level instance.
     """
 
     __slots__ = ()
@@ -36,7 +38,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_all_delta_channels_writes_history` terminates when it
+    The ancestor walk in `_get_channel_writes_history` terminates when it
     encounters this type (any non-sentinel blob stops the walk).
 
     `from_checkpoint` reconstructs the channel value directly from `.value`

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_jsonplus.py

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

libs/checkpoint/uv.lock

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

libs/langgraph/langgraph/errors.py

@@ -164,11 +164,12 @@ class NodeError:
     """Exception raised by the failed node."""
 
 
-class NodeTimeoutError(Exception):
+class NodeTimeoutError(TimeoutError):
     """Raised when a node invocation exceeds one of its configured timeouts.
 
-    Does **not** inherit from the built-in `TimeoutError` (a subclass of
-    `OSError`) so that the default `RetryPolicy` treats it as retryable.
+    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.
 
     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,12 +114,9 @@ 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_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`.
+    ancestor walk via `saver._get_channel_writes_history`. The walk terminates
+    at the nearest `_DeltaSnapshot` blob (step-based) or a pre-migration plain
+    value, so read depth is bounded by `snapshot_frequency`.
     """
     channel_specs: dict[str, BaseChannel] = {}
     managed_specs: dict[str, ManagedValueSpec] = {}
@@ -129,26 +126,18 @@ 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
-        if k in histories:
+        stored = checkpoint["channel_values"].get(k, MISSING)
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
             delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
+            history = saver._get_channel_writes_history(config, k)
             replay_ch = delta_spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:
-            ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
+            ch = spec.from_checkpoint(stored)
         channels[k] = ch
     return channels, managed_specs
 
@@ -169,28 +158,18 @@ 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
-        if k in histories:
+        stored = checkpoint["channel_values"].get(k, MISSING)
+        if _needs_replay(spec, stored) and saver is not None and config is not None:
             delta_spec = cast(DeltaChannel, spec)
-            history = histories[k]
+            history = await saver._aget_channel_writes_history(config, k)
             replay_ch = delta_spec.from_checkpoint(history.seed)
             replay_ch.replay_writes(history.writes)
             ch = replay_ch
         else:
-            ch = spec.from_checkpoint(checkpoint["channel_values"].get(k, MISSING))
+            ch = spec.from_checkpoint(stored)
         channels[k] = ch
     return channels, managed_specs
 

libs/langgraph/langgraph/pregel/main.py

@@ -30,7 +30,6 @@ 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,
@@ -41,6 +40,7 @@ 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,7 +3449,6 @@ 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,
@@ -3460,12 +3459,7 @@ 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').
-
-        !!! warning
-
-            The v3 streaming protocol is experimental and may change.
-        """
+        """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3')."""
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3497,7 +3491,6 @@ 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,
@@ -3508,12 +3501,7 @@ 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').
-
-        !!! warning
-
-            The v3 streaming protocol is experimental and may change.
-        """
+        """Internal v3 async streaming implementation. Public entry: astream_events(version='v3')."""
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3566,7 +3554,7 @@ class Pregel(
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any: ...
 
-    def stream_events(
+    def stream_events(  # type: ignore[override]
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3582,51 +3570,25 @@ class Pregel(
 
         For `version="v1"` / `"v2"`, yields `StreamEvent` dicts (see
         `Runnable.stream_events`). For `version="v3"`, returns a
-        `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.
+        `GraphRunStream` exposing typed projections.
 
         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 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"`.
+            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"`.
             **kwargs: Forwarded to the v1/v2 path.
 
         Returns:
-            For `version="v3"`, a `GraphRunStream` the caller iterates
-            to drive the run. Otherwise an `Iterator[StreamEvent]`.
+            For `version="v3"`, a `GraphRunStream`. Otherwise an
+            `Iterator[StreamEvent]`.
         """
         if version == "v3":
             return self._pregel_stream_v3(
@@ -3647,7 +3609,7 @@ class Pregel(
         *,
         version: Literal["v1", "v2"] = "v2",
         **kwargs: Any,
-    ) -> AsyncIterator[StreamEvent]: ...
+    ) -> _AsyncEventsResult: ...
 
     @overload
     def astream_events(
@@ -3660,9 +3622,9 @@ class Pregel(
         interrupt_after: All | Sequence[str] | None = None,
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
-    ) -> Awaitable[Any]: ...
+    ) -> _AsyncEventsResult: ...
 
-    def astream_events(
+    def astream_events(  # type: ignore[override]
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3673,31 +3635,21 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> 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.
-        """
+    ) -> _AsyncEventsResult:
+        """Async variant of `stream_events`. See `stream_events` for full docs."""
         if version == "v3":
-            return self._apregel_stream_v3(
-                input,
-                config,
-                interrupt_before=interrupt_before,
-                interrupt_after=interrupt_after,
-                control=control,
-                transformers=transformers,
+            return _AsyncEventsResult(
+                awaitable=self._apregel_stream_v3(
+                    input,
+                    config,
+                    interrupt_before=interrupt_before,
+                    interrupt_after=interrupt_after,
+                    control=control,
+                    transformers=transformers,
+                )
             )
-        return super().astream_events(input, config, version=version, **kwargs)
+        iterator = super().astream_events(input, config, version=version, **kwargs)
+        return _AsyncEventsResult(iterator=iterator)
 
     @overload
     def invoke(

libs/langgraph/langgraph/stream/run_stream.py

@@ -5,8 +5,6 @@ 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
@@ -27,7 +25,6 @@ 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.
 
@@ -41,11 +38,6 @@ 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__(
@@ -290,7 +282,6 @@ class GraphRunStream:
                 ch._subscribed = False
 
 
-@beta(message="The v3 streaming protocol on Pregel is experimental.")
 class AsyncGraphRunStream:
     """Async run stream with caller-driven pumping.
 
@@ -312,11 +303,6 @@ 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.0a4"
+version = "1.2.0a1"
 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.4.0a2,<2",
-    "langgraph-checkpoint>=4.1.0a3,<5.0.0",
+    "langchain-core>=1.3.2,<2",
+    "langgraph-checkpoint>=4.1.0a1,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
-    "langgraph-prebuilt>=1.1.0a1,<1.2.0",
+    "langgraph-prebuilt>=1.0.12,<1.1.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",
 ]
@@ -81,6 +81,7 @@ 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,7 +371,8 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
 
     saved = saver.get_tuple(config)
     assert saved is not None
-    assert "messages" not in saved.checkpoint["channel_values"]
+    assert "messages" in saved.checkpoint["channel_values"]
+    assert saved.checkpoint["channel_values"]["messages"] is DELTA_SENTINEL
 
     state = graph.get_state(config)
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
@@ -561,7 +562,7 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
 
     saved = saver.get_tuple(config)
     assert saved is not None
-    assert "files" not in saved.checkpoint["channel_values"]
+    assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
     state = graph.get_state(config)
     assert state.values["files"] == {
         "/doc_1.txt": "content for turn 1",

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_all_delta_channels_writes_history(config,
+Mechanism under test: the saver's `_get_channel_writes_history(config,
 channel)` walks the parent chain; when it encounters an ancestor whose
 `channel_values[channel]` is a real value (not `DELTA_SENTINEL`), it
 returns that as the `seed`. `DeltaChannel.from_checkpoint(seed)` uses
@@ -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_all_delta_channels_writes_history` must produce the
+   `BaseCheckpointSaver._get_channel_writes_history` must produce the
    same result as the optimized path.
 5. **Channel-type isolation across threads**: two threads on the same
    checkpointer under the delta-channel graph — one freshly-started,
@@ -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_all_delta_channels_writes_history` implementation from
+    `_get_channel_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_all_delta_channels_writes_history = (  # type: ignore[assignment]
-        InMemorySaver.__mro__[1]._get_all_delta_channels_writes_history  # type: ignore[attr-defined]
+    _get_channel_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._get_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]
+    _aget_channel_writes_history = (  # type: ignore[assignment]
+        InMemorySaver.__mro__[1]._aget_channel_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_all_delta_channels_writes_history` must be scoped to the target thread.
+    `_get_channel_writes_history` must be scoped to the target thread.
     """
 
     checkpointer = InMemorySaver()

libs/langgraph/tests/test_interleave_arrival_order.py

@@ -297,9 +297,7 @@ 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"})
@@ -321,9 +319,7 @@ 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).
@@ -331,9 +327,7 @@ 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,25 +396,19 @@ 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:
@@ -424,28 +418,21 @@ 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
@@ -461,16 +448,12 @@ 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"})
@@ -480,9 +463,7 @@ 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()
@@ -491,64 +472,46 @@ 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
@@ -563,34 +526,26 @@ 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()
@@ -600,27 +555,20 @@ 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"]
@@ -633,39 +581,29 @@ 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()
 
@@ -674,11 +612,7 @@ 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
@@ -1041,8 +975,7 @@ 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"])
@@ -1067,8 +1000,7 @@ 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
@@ -1084,10 +1016,7 @@ 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:
@@ -1106,8 +1035,7 @@ 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
@@ -1151,10 +1079,7 @@ 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]
             )
 
 
@@ -1236,8 +1161,7 @@ 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
@@ -1497,8 +1421,7 @@ 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()
@@ -1514,9 +1437,7 @@ 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
@@ -1529,9 +1450,7 @@ 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"]
@@ -1539,25 +1458,19 @@ 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:
@@ -1568,9 +1481,7 @@ 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,15 +210,6 @@ 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
@@ -1465,14 +1456,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(1.0)
+            await asyncio.sleep(0.5)
         return {"x": state["x"] + 1}
 
     builder = StateGraph(_TimeoutState)
     builder.add_node(
         "flaky",
         flaky,
-        timeout=TimeoutPolicy(idle_timeout=0.3),
+        timeout=TimeoutPolicy(idle_timeout=0.1),
         retry_policy=RetryPolicy(
             max_attempts=3,
             initial_interval=0.0,
@@ -1767,212 +1758,6 @@ 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
@@ -2210,6 +1995,7 @@ def test_graph_error_handler_does_not_swallow_interrupt_concurrent():
     )
 
 
+
 def test_node_error_handlers_route_to_matching_handler():
     class State(TypedDict):
         route: str
@@ -2267,3 +2053,4 @@ 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,8 +506,7 @@ 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)
@@ -518,8 +517,7 @@ 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)
@@ -531,9 +529,7 @@ 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
@@ -544,9 +540,7 @@ 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
@@ -557,8 +551,7 @@ 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)
@@ -570,10 +563,7 @@ 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,
@@ -614,10 +604,7 @@ 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,
@@ -653,10 +640,7 @@ 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],
     )
 
@@ -695,10 +679,7 @@ 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,15 +285,11 @@ 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
@@ -411,10 +407,7 @@ 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"]
@@ -433,10 +426,7 @@ 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
@@ -450,10 +440,7 @@ 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"]
@@ -590,9 +577,7 @@ 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
@@ -627,10 +612,7 @@ 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"])
@@ -677,10 +659,7 @@ 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
@@ -751,17 +730,13 @@ 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 = [
@@ -788,10 +763,7 @@ 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,9 +760,7 @@ 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.4.0a2"
-source = { registry = "https://pypi.org/simple" }
+version = "1.3.2"
+source = { git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release#ad4d43f38827e2aadce559e7e676b5eb4184a976" }
 dependencies = [
     { name = "jsonpatch" },
     { name = "langchain-protocol" },
@@ -1361,10 +1361,6 @@ 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"
@@ -1380,7 +1376,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a4"
+version = "1.2.0a1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1452,7 +1448,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1464,7 +1460,7 @@ requires-dist = [
 dev = [
     { name = "httpx" },
     { name = "jupyter" },
-    { name = "langchain-core", specifier = ">=1.0.0" },
+    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1497,7 +1493,7 @@ lint = [
 ]
 test = [
     { name = "httpx" },
-    { name = "langchain-core", specifier = ">=1.0.0" },
+    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1561,7 +1557,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a3"
+version = "4.1.0a1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1609,7 +1605,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a3"
+version = "3.1.0a1"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -1755,7 +1751,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0a1"
+version = "1.0.12"
 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.1.0a1"
+version = "1.0.12"
 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,10 +147,7 @@ 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:
@@ -202,9 +199,7 @@ 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:
@@ -244,9 +239,7 @@ 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)
@@ -273,8 +266,7 @@ 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] = []
@@ -303,9 +295,7 @@ 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.4.0a2"
+version = "1.3.2"
 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/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" },
+    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
 ]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.14"
+version = "0.0.12"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/05/bf/efb5e2ed832e4d6d45590e25a9e5191986b291b543bc6a807b48bee070b0/langchain_protocol-0.0.14.tar.gz", hash = "sha256:bc1e8553122e6ede310280462d5813023a172ff2785ccbbdec54d43f3a15e5f2", size = 5862, upload-time = "2026-04-29T16:40:18.657Z" }
+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" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
+    { url = "https://files.pythonhosted.org/packages/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" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a4"
+version = "1.2.0a1"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -294,7 +294,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", specifier = ">=1.3.2,<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.0a3"
+version = "4.1.0a1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -413,7 +413,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a3"
+version = "3.1.0a1"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },
@@ -503,7 +503,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0a1"
+version = "1.0.12"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/uv.lock

@@ -266,7 +266,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.4.0a2"
+version = "1.3.2"
 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/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/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" },
+    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
 ]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.14"
+version = "0.0.12"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/05/bf/efb5e2ed832e4d6d45590e25a9e5191986b291b543bc6a807b48bee070b0/langchain_protocol-0.0.14.tar.gz", hash = "sha256:bc1e8553122e6ede310280462d5813023a172ff2785ccbbdec54d43f3a15e5f2", size = 5862, upload-time = "2026-04-29T16:40:18.657Z" }
+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" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
+    { url = "https://files.pythonhosted.org/packages/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" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a4"
+version = "1.2.0a1"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -311,7 +311,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
+    { name = "langchain-core", specifier = ">=1.3.2,<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.0a3"
+version = "4.1.0a1"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -430,7 +430,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.1.0a1"
+version = "1.0.12"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },