chore: ruff format fixes

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

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import threading
 from collections import defaultdict
-from collections.abc import Iterator, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from contextlib import contextmanager
 from typing import Any, cast
 
@@ -27,10 +27,9 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_STAGE1_SQL,
     SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaStage1Row,
+    _build_delta_stage1_sql,
     _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
@@ -444,53 +443,79 @@ class PostgresSaver(BasePostgresSaver):
                 with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    def _get_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
-        """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
+    def _get_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Fast-path override of `BaseCheckpointSaver._get_all_delta_channels_writes_history`.
 
-        Two-stage query: stage 1 scans checkpoint metadata to walk the parent
-        chain and locate the nearest snapshot; stage 2 fetches only the
-        chain-limited writes and single seed blob.
+        Two-stage query, both stages cover ALL requested channels in a single
+        Postgres roundtrip each:
+
+        * Stage 1: dynamic SELECT over `checkpoints` with K parallel JSONB
+          key lookups (one column pair per channel) — no subquery, no
+          aggregation. Returns one row per checkpoint with versions and
+          snapshot flags for every requested channel.
+
+        * Stage 2: one UNION ALL over `checkpoint_writes` and
+          `checkpoint_blobs` filtered by `channel = ANY(?)` and per-channel
+          chain_cids / seed_versions (collapsed across channels).
         """
+        if not channels:
+            return {}
+        channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
         if checkpoint_id is None:
             target = self.get_tuple(config)
             if target is None:
-                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                return {
+                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                    for ch in channels
+                }
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
+        # Stage 1: K parallel JSONB lookups per row, one query for all channels.
+        stage1_sql = _build_delta_stage1_sql(channels)
+        stage1_params: list[Any] = []
+        for ch in channels:
+            stage1_params.extend([ch, ch])
+        stage1_params.extend([thread_id, checkpoint_ns])
         with self._cursor() as cur:
-            cur.execute(
-                SELECT_DELTA_STAGE1_SQL,
-                (channel, channel, thread_id, checkpoint_ns),
-            )
+            cur.execute(stage1_sql, stage1_params)
             stage1_rows = cur.fetchall()
-        chain_cids, seed_version = self._walk_stage1(
-            cast("list[_DeltaStage1Row]", stage1_rows), checkpoint_id
+
+        chain_by_ch, seed_ver_by_ch = self._walk_stage1_multi(
+            cast("list[Mapping[str, Any]]", stage1_rows), checkpoint_id, channels
         )
-        seed_versions = [seed_version] if seed_version else []
+        # 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,
                 (
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    chain_cids,
+                    channels,
+                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    seed_versions,
+                    channels,
+                    union_seed_versions,
                 ),
             )
             stage2_rows = cur.fetchall()
-        return self._build_delta_channel_writes_history(
-            channel=channel,
-            chain_cids=chain_cids,
-            seed_version=seed_version,
+        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),
         )
 

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 from collections import defaultdict
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from contextlib import asynccontextmanager
 from typing import Any, cast
 
@@ -27,10 +27,9 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_STAGE1_SQL,
     SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaStage1Row,
+    _build_delta_stage1_sql,
     _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
@@ -405,53 +404,68 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    async def _aget_channel_writes_history(
-        self, config: RunnableConfig, channel: str
-    ) -> _ChannelWritesHistory:
-        """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
+    async def _aget_all_delta_channels_writes_history(
+        self, config: RunnableConfig, channels: Sequence[str]
+    ) -> Mapping[str, _ChannelWritesHistory]:
+        """Fast-path override of `BaseCheckpointSaver._aget_all_delta_channels_writes_history`.
 
-        Two-stage query: stage 1 scans checkpoint metadata to walk the parent
-        chain and locate the nearest snapshot; stage 2 fetches only the
-        chain-limited writes and single seed blob.
+        Two-stage query, both stages cover ALL requested channels in a single
+        Postgres roundtrip each. See `PostgresSaver._get_all_delta_channels_writes_history`
+        for design notes.
         """
+        if not channels:
+            return {}
+        channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
         if checkpoint_id is None:
             target = await self.aget_tuple(config)
             if target is None:
-                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                return {
+                    ch: _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+                    for ch in channels
+                }
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
+        stage1_sql = _build_delta_stage1_sql(channels)
+        stage1_params: list[Any] = []
+        for ch in channels:
+            stage1_params.extend([ch, ch])
+        stage1_params.extend([thread_id, checkpoint_ns])
         async with self._cursor() as cur:
-            await cur.execute(
-                SELECT_DELTA_STAGE1_SQL,
-                (channel, channel, thread_id, checkpoint_ns),
-            )
+            await cur.execute(stage1_sql, stage1_params)
             stage1_rows = await cur.fetchall()
-        chain_cids, seed_version = self._walk_stage1(
-            cast("list[_DeltaStage1Row]", stage1_rows), checkpoint_id
+
+        chain_by_ch, seed_ver_by_ch = self._walk_stage1_multi(
+            cast("list[Mapping[str, Any]]", stage1_rows), checkpoint_id, channels
         )
-        seed_versions = [seed_version] if seed_version else []
+        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,
                 (
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    chain_cids,
+                    channels,
+                    union_chain_cids,
                     thread_id,
                     checkpoint_ns,
-                    channel,
-                    seed_versions,
+                    channels,
+                    union_seed_versions,
                 ),
             )
             stage2_rows = await cur.fetchall()
-        return self._build_delta_channel_writes_history(
-            channel=channel,
-            chain_cids=chain_cids,
-            seed_version=seed_version,
+        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),
         )
 

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

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import random
 import warnings
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from importlib.metadata import version as get_version
 from typing import Any, TypedDict, cast
 
@@ -161,6 +161,7 @@ class _DeltaStage2Row(TypedDict, total=False):
 
     _kind: str  # "w" or "b"
     checkpoint_id: str | None  # "w" rows only
+    channel: str | None  # set on both "w" and "b" rows
     type: str | None
     blob: bytes | None
     task_id: str | None  # "w" rows only
@@ -168,48 +169,85 @@ class _DeltaStage2Row(TypedDict, total=False):
     version: str | None  # "b" rows only
 
 
-# Two-stage DeltaChannel reconstruction.  Stage 1 scans checkpoint
-# metadata (no blob bytes) to walk the parent chain and locate the
-# nearest snapshot marker.  Stage 2 fetches only the chain-limited
-# writes and the single seed snapshot blob.
+# Multi-channel two-stage DeltaChannel reconstruction.
 #
-# Parameter order:
-#   stage1: (channel, channel, thread_id, checkpoint_ns)
-#   stage2: (thread_id, checkpoint_ns, channel, chain_cids[],
-#            thread_id, checkpoint_ns, channel, seed_versions[])
+# 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_STAGE1_SQL = """
-    SELECT checkpoint_id,
-           parent_checkpoint_id,
-           checkpoint -> 'channel_versions' ->> %s AS ver,
-           (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS has_snapshot
-    FROM checkpoints
-    WHERE thread_id = %s AND checkpoint_ns = %s
-"""
 
 SELECT_DELTA_STAGE2_SQL = """
     SELECT 'w'::text AS _kind,
-           checkpoint_id,
+           checkpoint_id, channel,
            type, blob, task_id, idx, NULL::text AS version
     FROM checkpoint_writes
-    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = ANY(%s)
       AND checkpoint_id = ANY(%s)
     UNION ALL
-    SELECT 'b', NULL,
+    SELECT 'b', NULL, channel,
            type, blob, NULL, NULL, version
     FROM checkpoint_blobs
-    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = ANY(%s)
       AND version = ANY(%s)
 """
 
 
-class _DeltaStage1Row(TypedDict):
-    """One row from `SELECT_DELTA_STAGE1_SQL`."""
-
-    checkpoint_id: str
-    parent_checkpoint_id: str | None
-    ver: str | None
-    has_snapshot: bool
+# 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]):
@@ -255,86 +293,119 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         }
 
     @staticmethod
-    def _walk_stage1(
-        stage1_rows: Sequence[_DeltaStage1Row],
+    def _walk_stage1_multi(
+        stage1_rows: Sequence[Mapping[str, Any]],
         target_id: str,
-    ) -> tuple[list[str], str | None]:
-        """Walk the parent chain from stage 1 metadata rows.
+        channels: Sequence[str],
+    ) -> tuple[dict[str, list[str]], dict[str, str | None]]:
+        """Walk the parent chain once for all requested channels.
 
-        Returns (chain_cids, seed_version):
-          chain_cids: ancestor checkpoint IDs from target's parent down to
-                      the seed (or root), in newest-first order.
-          seed_version: the channel blob version at the nearest ancestor
-                        with has_snapshot=True, or None if pure delta.
+        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] = {}
-        ver_of: dict[str, str | None] = {}
-        snapshot_of: dict[str, bool] = {}
+        # 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 = r["checkpoint_id"]
-            parent_of[cid] = r["parent_checkpoint_id"]
-            ver_of[cid] = r["ver"]
-            snapshot_of[cid] = r["has_snapshot"]
+            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_cids: list[str] = []
-        seed_version: str | None = None
-        cur_cid: str | None = parent_of.get(target_id)
-        while cur_cid is not None:
-            chain_cids.append(cur_cid)
-            if snapshot_of.get(cur_cid, False):
-                seed_version = ver_of.get(cur_cid)
-                break
-            cur_cid = parent_of.get(cur_cid)
-        return chain_cids, seed_version
+        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_channel_writes_history(
+    def _build_delta_channels_writes_history(
         self,
         *,
-        channel: str,
-        chain_cids: list[str],
-        seed_version: str | None,
+        channels: Sequence[str],
+        chain_by_ch: dict[str, list[str]],
+        seed_ver_by_ch: dict[str, str | None],
         stage2_rows: Sequence[_DeltaStage2Row],
-    ) -> _ChannelWritesHistory:
-        """Reconstruct delta channel history from two-stage query results.
+    ) -> dict[str, _ChannelWritesHistory]:
+        """Demux stage 2 rows per channel; produce per-channel histories.
 
-        chain_cids are in newest-first order (target's parent first).
-        stage2_rows contain only writes for chain_cids and the single
-        seed blob at seed_version.
+        stage2_rows carry `channel` on every row. We build per-channel
+        `writes_by_cid` and per-channel `seed_blob` dicts, then assemble
+        a `_ChannelWritesHistory` per requested channel.
         """
-        writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
-        seed_blob: tuple[str, bytes] | None = None
+        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
+        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
+            ch: {} for ch in channels
+        }
+        # seed_blob_by_ver[(channel, version)] = (type, blob)
+        seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
         for r in stage2_rows:
+            ch = cast(str, r["channel"])
             kind = r["_kind"]
             if kind == "w":
                 cid = cast(str, r["checkpoint_id"])
-                writes_by_cid.setdefault(cid, []).append(
+                writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
                         "tuple[str, bytes, str, int]",
                         (r["type"], r["blob"], r["task_id"], r["idx"]),
                     )
                 )
             else:  # kind == "b"
-                seed_blob = cast("tuple[str, bytes]", (r["type"], r["blob"]))
+                ver = cast(str, r["version"])
+                seed_blob_by_ver[(ch, ver)] = cast(
+                    "tuple[str, bytes]", (r["type"], r["blob"])
+                )
 
-        for ws in writes_by_cid.values():
-            ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
+        # Sort writes per (channel, cid) newest-first by (task_id, idx)
+        for cid_map in writes_by_ch_by_cid.values():
+            for ws in cid_map.values():
+                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
-        if not chain_cids:
-            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+        result: dict[str, _ChannelWritesHistory] = {}
+        for ch in channels:
+            chain_cids = chain_by_ch.get(ch, [])
+            seed_version = seed_ver_by_ch.get(ch)
 
-        collected: list[PendingWrite] = []
-        for cid in chain_cids:
-            for type_tag, write_blob, task_id, _idx in writes_by_cid.get(cid, []):
-                val = self.serde.loads_typed((type_tag, write_blob))
-                collected.append((task_id, channel, val))
+            collected: 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))
 
-        seed: Any = DELTA_SENTINEL
-        if seed_blob is not None and seed_blob[0] != "empty":
-            seed = self.serde.loads_typed(seed_blob)
+            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()
-        return _ChannelWritesHistory(seed=seed, writes=collected)
+            collected.reverse()
+            result[ch] = _ChannelWritesHistory(seed=seed, writes=collected)
+        return result
 
     def _dump_blobs(
         self,

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

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

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

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

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

@@ -36,7 +36,7 @@ class _DeltaSnapshot(NamedTuple):
     """Snapshot blob for a DeltaChannel with finite snapshot_frequency.
 
     Stored in checkpoint_blobs via the `EXT_DELTA_SNAPSHOT` msgpack ext code.
-    The ancestor walk in `_get_channel_writes_history` terminates when it
+    The ancestor walk in `_get_all_delta_channels_writes_history` terminates when it
     encounters this type (any non-sentinel blob stops the walk).
 
     `from_checkpoint` reconstructs the channel value directly from `.value`

libs/checkpoint/tests/test_memory.py

@@ -335,9 +335,10 @@ class TestInMemorySaverDeltaChannel:
         assert channel not in result
 
     def test_get_channel_writes_collects_ancestor_writes_only(self) -> None:
-        """_get_channel_writes_history collects ancestor writes oldest→newest,
-        and excludes writes stored at the target checkpoint itself (those are
-        pending writes for the next step, applied separately by pregel)."""
+        """_get_all_delta_channels_writes_history collects ancestor writes
+        oldest→newest, and excludes writes stored at the target checkpoint
+        itself (those are pending writes for the next step, applied separately
+        by pregel)."""
         saver = InMemorySaver()
         serde = JsonPlusSerializer()
 
@@ -375,7 +376,9 @@ class TestInMemorySaverDeltaChannel:
                 "checkpoint_id": "cp2",
             }
         }
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
         assert values == [{"content": "hi"}]
@@ -405,15 +408,17 @@ class TestInMemorySaverDeltaChannel:
                 "checkpoint_id": "cp1",
             }
         }
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
         assert result.seed is DELTA_SENTINEL
         assert result.writes == []
 
 
 class TestBaseFallbackGetChannelWrites:
-    """Exercises the `BaseCheckpointSaver._get_channel_writes_history` default
-    implementation — the path third-party savers inherit when they don't
-    override `_get_channel_writes_history` themselves.
+    """Exercises the `BaseCheckpointSaver._get_all_delta_channels_writes_history`
+    default implementation — the path third-party savers inherit when they
+    don't override `_get_all_delta_channels_writes_history` themselves.
 
     Regression guard for a bug where the fallback passed the caller's config
     (with `checkpoint_id`) straight to `self.list()`, which most savers
@@ -429,11 +434,11 @@ class TestBaseFallbackGetChannelWrites:
         """
 
         class _ThirdPartyStyleSaver(InMemorySaver):
-            _get_channel_writes_history = (
-                InMemorySaver.__mro__[1]._get_channel_writes_history  # type: ignore[attr-defined]
+            _get_all_delta_channels_writes_history = (
+                InMemorySaver.__mro__[1]._get_all_delta_channels_writes_history  # type: ignore[attr-defined]
             )
-            _aget_channel_writes_history = (
-                InMemorySaver.__mro__[1]._aget_channel_writes_history  # type: ignore[attr-defined]
+            _aget_all_delta_channels_writes_history = (
+                InMemorySaver.__mro__[1]._aget_all_delta_channels_writes_history  # type: ignore[attr-defined]
             )
 
         saver = _ThirdPartyStyleSaver()
@@ -477,7 +482,9 @@ class TestBaseFallbackGetChannelWrites:
             }
         }
 
-        result = saver._get_channel_writes_history(config, "messages")
+        result = saver._get_all_delta_channels_writes_history(config, ["messages"])[
+            "messages"
+        ]
 
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
@@ -494,7 +501,9 @@ class TestBaseFallbackGetChannelWrites:
             }
         }
 
-        result = await saver._aget_channel_writes_history(config, "messages")
+        result = (
+            await saver._aget_all_delta_channels_writes_history(config, ["messages"])
+        )["messages"]
 
         assert result.seed is DELTA_SENTINEL
         values = [v for _, _, v in result.writes]
@@ -503,9 +512,9 @@ class TestBaseFallbackGetChannelWrites:
     async def test_async_fallback_concurrent_tasks_do_not_interfere(self) -> None:
         """Regression: the re-entrancy guard must be task-local, not thread-local.
 
-        Two concurrent `_aget_channel_writes_history` calls on the same
-        event-loop thread must each see their full reconstructed writes. A
-        `threading.local()` guard would let whichever task set it first
+        Two concurrent `_aget_all_delta_channels_writes_history` calls on the
+        same event-loop thread must each see their full reconstructed writes.
+        A `threading.local()` guard would let whichever task set it first
         short-circuit the other to `writes=[]`.
         """
         import asyncio
@@ -533,12 +542,13 @@ class TestBaseFallbackGetChannelWrites:
         }
 
         results = await asyncio.gather(
-            saver._aget_channel_writes_history(config, "messages"),
-            saver._aget_channel_writes_history(config, "messages"),
+            saver._aget_all_delta_channels_writes_history(config, ["messages"]),
+            saver._aget_all_delta_channels_writes_history(config, ["messages"]),
         )
 
         expected_values = [{"content": "first"}, {"content": "second"}]
-        for result in results:
+        for result_map in results:
+            result = result_map["messages"]
             assert result.seed is DELTA_SENTINEL
             values = [v for _, _, v in result.writes]
             assert values == expected_values
@@ -625,7 +635,9 @@ class TestPreDeltaBlobTerminator:
             }
         }
 
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
 
         # Seed came from the pre-delta blob at cp1.
         assert result.seed == ["A"]
@@ -647,7 +659,9 @@ class TestPreDeltaBlobTerminator:
             }
         }
 
-        result = saver._get_channel_writes_history(config, channel)
+        result = saver._get_all_delta_channels_writes_history(config, [channel])[
+            channel
+        ]
 
         values = [v for _, _, v in result.writes]
         # The pre-delta write under cp1 must not appear (the blob subsumes it).

libs/langgraph/langgraph/graph/state.py

@@ -1128,7 +1128,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             name: The name to use for the compiled graph.
             transformers: Optional sequence of `StreamTransformer` classes or
                 configured factories. Classes and factories are instantiated
-                per run whenever `stream_v2` / `astream_v2` is called and are
+                per run whenever `stream_events(version="v3")` / `astream_events(version="v3")` is called and are
                 propagated to subgraph scopes. Custom factories should follow
                 the standard `StreamTransformer` constructor shape by
                 accepting `scope` as their first argument. Appended after the

libs/langgraph/langgraph/pregel/_checkpoint.py

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

libs/langgraph/langgraph/pregel/_messages.py

@@ -349,7 +349,7 @@ class StreamMessagesHandlerV2(StreamMessagesHandler, _V2StreamingCallbackHandler
         tags: list[str] | None = None,
         **kwargs: Any,
     ) -> Any:
-        """Forward a protocol event from `stream_v2` as a messages stream part.
+        """Forward a protocol event from `stream_events(version="v3")` as a messages stream part.
 
         Fires once per `MessagesData` event (`message-start`, per-block
         `content-block-*`, `message-finish`). The transformer layer

libs/langgraph/langgraph/pregel/main.py

@@ -30,6 +30,7 @@ from typing import (
 )
 from uuid import UUID, uuid5
 
+from langchain_core._api import beta
 from langchain_core.globals import get_debug
 from langchain_core.runnables import (
     RunnableSequence,
@@ -41,6 +42,7 @@ from langchain_core.runnables.config import (
     get_callback_manager_for_config,
 )
 from langchain_core.runnables.graph import Graph
+from langchain_core.runnables.schema import StreamEvent
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
@@ -376,7 +378,7 @@ def _collect_stream_modes(mux: Any) -> list[StreamMode]:
     """Return the union of `required_stream_modes` across registered transformers.
 
     Transformers declare the stream modes they need to function, and
-    `stream_v2` asks the graph for exactly that union — no hardcoded
+    `stream_events(version="v3")` asks the graph for exactly that union — no hardcoded
     default set. If zero transformers declare a given mode, the graph
     does not stream events for it.
     """
@@ -406,14 +408,14 @@ def _normalize_stream_transformer_factories(
     for spec in specs or ():
         if isinstance(spec, StreamTransformer):
             raise TypeError(
-                "stream_v2 transformers must be scope-aware callables, "
+                "stream_events(version='v3') transformers must be scope-aware callables, "
                 f"got pre-built instance {type(spec).__name__}. Pass the "
                 "transformer class or a factory like "
                 "`lambda scope: MyTransformer(scope, ...)`."
             )
         if not callable(spec):
             raise TypeError(
-                "stream_v2 transformers must be scope-aware callables, "
+                "stream_events(version='v3') transformers must be scope-aware callables, "
                 f"got {type(spec).__name__}."
             )
 
@@ -3447,7 +3449,8 @@ class Pregel(
                 await asyncio.shield(run_manager.on_chain_error(e))
             raise
 
-    def stream_v2(
+    @beta(message="The v3 streaming protocol on Pregel is experimental.")
+    def _pregel_stream_v3(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3457,41 +3460,11 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Start a sync v2 streaming run driven by transformer projections.
+        """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3').
 
-        Builds a `StreamMux` from the built-in transformers, this
-        graph's compile-time `stream_transformers`, and any additional
-        `transformers=` supplied at the call site. Returns a
-        `GraphRunStream` that the caller drives by iterating any
-        projection — no background thread.
+        !!! warning
 
-        `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_v2` run is not fully supported. The outer v2
-            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_v2` for
-            the inner graph as well, or call
-            `chat_model.stream(...)` explicitly, to get token-level
-            streaming.
-
-        Args:
-            input: Graph input.
-            config: Optional runnable config forwarded to the graph.
-            interrupt_before: Nodes to interrupt before, if any.
-            interrupt_after: Nodes to interrupt after, if any.
-            control: Optional run control used to request cooperative drain.
-            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.
-
-        Returns:
-            A `GraphRunStream` the caller iterates to drive the run.
+            The v3 streaming protocol is experimental and may change.
         """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
@@ -3524,7 +3497,8 @@ class Pregel(
         )
         return GraphRunStream(graph_iter, mux)
 
-    async def astream_v2(
+    @beta(message="The v3 streaming protocol on Pregel is experimental.")
+    async def _apregel_stream_v3(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3534,31 +3508,11 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Async counterpart to `stream_v2`.
+        """Internal v3 async streaming implementation. Public entry: astream_events(version='v3').
 
-        Returns an `AsyncGraphRunStream` whose projections can be awaited
-        concurrently; each subscribed cursor drives the pump when its
-        buffer is empty.
+        !!! warning
 
-        Note:
-            Same nesting limitation as `stream_v2`: nesting v1
-            `astream(stream_mode="messages")` inside a node of an
-            `astream_v2` run drops `on_llm_new_token` chunks because
-            the outer v2 handler reroutes `BaseChatModel.invoke`
-            through the v2 event protocol. Use `astream_v2` for the
-            inner graph as well, or call `chat_model.astream(...)`
-            explicitly, to get token-level streaming.
-
-        Args:
-            input: Graph input.
-            config: Optional runnable config forwarded to the graph.
-            interrupt_before: Nodes to interrupt before, if any.
-            interrupt_after: Nodes to interrupt after, if any.
-            control: Optional run control used to request cooperative drain.
-            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.
+            The v3 streaming protocol is experimental and may change.
         """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
@@ -3589,6 +3543,162 @@ class Pregel(
         ).__aiter__()
         return AsyncGraphRunStream(graph_aiter, mux)
 
+    @overload
+    def stream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v1", "v2"] = "v2",
+        **kwargs: Any,
+    ) -> Iterator[StreamEvent]: ...
+
+    @overload
+    def stream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v3"],
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        control: RunControl | None = None,
+        transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+    ) -> Any: ...
+
+    def stream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v1", "v2", "v3"] = "v2",
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        control: RunControl | None = None,
+        transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        """Stream events from this graph.
+
+        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.
+
+        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"`.
+            **kwargs: Forwarded to the v1/v2 path.
+
+        Returns:
+            For `version="v3"`, a `GraphRunStream` the caller iterates
+            to drive the run. Otherwise an `Iterator[StreamEvent]`.
+        """
+        if version == "v3":
+            return self._pregel_stream_v3(
+                input,
+                config,
+                interrupt_before=interrupt_before,
+                interrupt_after=interrupt_after,
+                control=control,
+                transformers=transformers,
+            )
+        return super().stream_events(input, config, version=version, **kwargs)
+
+    @overload
+    def astream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v1", "v2"] = "v2",
+        **kwargs: Any,
+    ) -> AsyncIterator[StreamEvent]: ...
+
+    @overload
+    def astream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v3"],
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        control: RunControl | None = None,
+        transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
+    ) -> Awaitable[Any]: ...
+
+    def astream_events(
+        self,
+        input: InputT | Command | None,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v1", "v2", "v3"] = "v2",
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        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.
+        """
+        if version == "v3":
+            return 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)
+
     @overload
     def invoke(
         self,
@@ -4083,7 +4193,7 @@ def _resolve_parent_ns(
 ) -> tuple[str, ...]:
     """Return the checkpoint namespace the caller is running under.
 
-    `stream_v2` uses this to scope its native projections
+    `stream_events(version="v3")` uses this to scope its native projections
     (`ValuesTransformer`, `MessagesTransformer`) to events emitted at
     the run's own level. A root call resolves to `()`; a call made
     from inside a node carries the outer graph's task namespace so the

libs/langgraph/langgraph/stream/__init__.py

@@ -1,7 +1,7 @@
 """Streaming infrastructure for LangGraph.
 
-Compile a graph with `transformers=[...]` and call `graph.stream_v2()` /
-`graph.astream_v2()` to drive a transformer pipeline that projects the
+Compile a graph with `transformers=[...]` and call `graph.stream_events(version="v3")` /
+`graph.astream_events(version="v3")` to drive a transformer pipeline that projects the
 graph's raw events into ergonomic per-channel streams.
 """
 

libs/langgraph/langgraph/stream/_types.py

@@ -88,7 +88,7 @@ class StreamTransformer(ABC):
         required_stream_modes: Stream modes the graph must emit for
             this transformer to have anything to process. Computed as
             the union across all registered transformers to determine
-            which modes a `stream_v2` run requests from the graph.
+            which modes a `stream_events(version="v3")` run requests from the graph.
             Empty tuple means the transformer consumes only synthetic
             events (or is purely passive).
     """

libs/langgraph/langgraph/stream/run_stream.py

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

libs/langgraph/langgraph/stream/transformers.py

@@ -38,8 +38,8 @@ class ValuesTransformer(StreamTransformer):
     Only values events at the run's own level are captured; snapshots
     from deeper subgraphs are left in the main event log but excluded
     from the projection. "Own level" is defined by `scope`, which
-    `stream_v2` / `astream_v2` populate from the caller's
-    checkpoint namespace so that a nested `stream_v2` call still
+    `stream_events(version="v3")` / `astream_events(version="v3")` populate from the caller's
+    checkpoint namespace so that a nested `stream_events(version="v3")` call still
     sees its own root snapshots.
     """
 
@@ -165,7 +165,7 @@ class MessagesTransformer(StreamTransformer):
     metadata)` from `StreamMessagesHandler`):
 
     1. Protocol event (dict with `"event"` key) — emitted by
-       `stream_v2()` / `astream_v2()` via the `on_stream_event`
+       `stream_events(version="v3")` / `astream_events(version="v3")` via the `on_stream_event`
        callback. Routed to an existing `ChatModelStream` by
        `metadata["run_id"]`. A `message-start` event creates a new
        stream; `message-finish` closes it.
@@ -177,15 +177,15 @@ class MessagesTransformer(StreamTransformer):
     V1 `AIMessageChunk` tuples (from `on_llm_new_token`) are not
     streamed into this projection: chat models that want to populate
     `run.messages` with content-block streaming must use
-    `stream_v2()` / `astream_v2()`. Models called via the legacy
+    `stream_events(version="v3")` / `astream_events(version="v3")`. Models called via the legacy
     `stream()` method still surface their final `AIMessage` via
     `on_chain_end` when a node returns it as state.
 
     Only events at the run's own level are projected; tokens from
     deeper subgraphs are left in the main event log but excluded from
     `.messages`. "Own level" is defined by `scope`, which
-    `stream_v2` / `astream_v2` populate from the caller's checkpoint
-    namespace so that a `stream_v2` call inside a node still sees its
+    `stream_events(version="v3")` / `astream_events(version="v3")` populate from the caller's checkpoint
+    namespace so that a `stream_events(version="v3")` call inside a node still sees its
     own root chat model streams on `.messages`. Consumers that need
     subgraph tokens should iterate the raw event stream or register a
     custom transformer.
@@ -281,7 +281,7 @@ class MessagesTransformer(StreamTransformer):
         ):
             self._route_whole_message(payload, node=node)
         # Legacy AIMessageChunk tuples (from on_llm_new_token) are ignored;
-        # v1 streaming callers must switch to stream_v2() to populate this
+        # v1 streaming callers must switch to stream_events(version="v3") to populate this
         # projection.
 
         return True

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.0a3"
+version = "1.2.0a4"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -24,10 +24,10 @@ classifiers = [
     'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.3.2,<2",
+    "langchain-core>=1.4.0a2,<2",
     "langgraph-checkpoint>=4.1.0a3,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
-    "langgraph-prebuilt>=1.0.12,<1.1.0",
+    "langgraph-prebuilt>=1.1.0a1,<1.2.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",
 ]

libs/langgraph/tests/test_delta_channel_migration.py

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

libs/langgraph/tests/test_interleave_arrival_order.py

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

@@ -1,4 +1,4 @@
-"""Tests for Pregel.stream_v2 / astream_v2 and the transformer pipeline."""
+"""Tests for Pregel.stream_events(version="v3") / astream_events(version="v3") and the transformer pipeline."""
 
 from __future__ import annotations
 
@@ -125,7 +125,7 @@ def _build_custom_stream_graph():
 class _CustomPassthroughTransformer(StreamTransformer):
     """Opts a run into the `custom` stream mode without building a projection.
 
-    `stream_v2` requests only the modes that registered transformers
+    `stream_events(version="v3")` requests only the modes that registered transformers
     declare via `required_stream_modes`. Custom events are raw user
     emissions from `StreamWriter`, so tests that want them visible on
     the main event log register this pass-through transformer.
@@ -390,25 +390,31 @@ class TestStreamChannelNamed:
 
 
 # ---------------------------------------------------------------------------
-# stream_v2 sync tests
+# stream_events(version="v3") sync tests
 # ---------------------------------------------------------------------------
 
 
 class TestStreamV2Sync:
     def test_values_projection(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         events = list(run)
         assert len(events) > 0
         for event in events:
@@ -418,22 +424,27 @@ class TestStreamV2Sync:
             assert isinstance(event["params"]["timestamp"], int)
 
     def test_extensions_has_native_keys(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2(
+        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"]
@@ -444,18 +455,22 @@ class TestStreamV2Sync:
     def test_custom_events_suppressed_without_transformer(self) -> None:
         """Without a transformer declaring `"custom"`, no custom events flow.
 
-        `stream_v2` asks the graph only for the modes that registered
+        `stream_events(version="v3")` asks the graph only for the modes that registered
         transformers require. Built-ins cover `values` / `messages`;
         consumers that want raw custom events surface them by
         registering a transformer whose `required_stream_modes`
         includes `"custom"`.
         """
-        run = _build_custom_stream_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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"})
@@ -465,7 +480,9 @@ class TestStreamV2Sync:
         assert run.extensions["messages"]._subscribed is False
 
     def test_abort_marks_exhausted_and_closes_mux(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         values_iter = iter(run.values)
         _ = next(values_iter)
         run.abort()
@@ -474,48 +491,63 @@ class TestStreamV2Sync:
         run.abort()  # idempotent
 
     def test_context_manager_calls_abort_on_exit(self) -> None:
-        with _build_simple_graph().stream_v2({"value": "x", "items": []}) 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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2(
+        run = _build_interrupt_graph().stream_events(
             {"value": "x", "items": []},
             {"configurable": {"thread_id": "t1"}},
+            version="v3",
         )
         _ = run.output
         assert run.interrupted is True
@@ -523,7 +555,7 @@ class TestStreamV2SyncInterrupt:
 
 
 # ---------------------------------------------------------------------------
-# astream_v2 async tests
+# astream_events(version="v3") async tests
 # ---------------------------------------------------------------------------
 
 
@@ -531,26 +563,34 @@ class TestStreamV2SyncInterrupt:
 @NEEDS_CONTEXTVARS
 class TestStreamV2Async:
     async def test_values_projection(self) -> None:
-        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         values_iter = aiter(run.values)
         _ = await anext(values_iter)
         await run.abort()
@@ -560,21 +600,26 @@ class TestStreamV2Async:
         await run.abort()  # idempotent
 
     async def test_context_manager_calls_abort_on_exit(self) -> None:
-        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2(
+        run = await _build_custom_stream_graph().astream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CustomPassthroughTransformer],
         )
         events = [e async for e in run]
@@ -588,29 +633,39 @@ class TestStreamV2Async:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncErrors:
     async def test_error_propagation_output(self) -> None:
-        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run = await _build_error_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         with pytest.raises(ValueError, match="boom"):
             await run.interrupts()
 
@@ -619,9 +674,10 @@ class TestStreamV2AsyncErrors:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncInterrupt:
     async def test_interrupted(self) -> None:
-        run = await _build_interrupt_graph().astream_v2(
+        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
@@ -985,8 +1041,8 @@ class TestCustomTransformer:
                     self._channel.push(self._count)
                 return True
 
-        run = _build_simple_graph().stream_v2(
-            {"value": "x", "items": []}, 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"])
@@ -1011,15 +1067,15 @@ class TestCustomTransformer:
                     self._log.push("saw_values")
                 return True
 
-        run = _build_simple_graph().stream_v2(
-            {"value": "x", "items": []}, transformers=[FooTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[FooTransformer]
         )
         foo_iter = iter(run.foo)
         _ = run.output
         assert "foo" in run.extensions and run.foo is run.extensions["foo"]
         assert "saw_values" in list(foo_iter)
 
-    def test_stream_v2_rejects_transformer_instances(self) -> None:
+    def test_stream_events_v3_rejects_transformer_instances(self) -> None:
         class InstanceTransformer(StreamTransformer):
             def init(self) -> dict[str, Any]:
                 return {}
@@ -1028,8 +1084,10 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(TypeError, match="pre-built instance"):
-            _build_simple_graph().stream_v2(
-                {"value": "x", "items": []}, transformers=[InstanceTransformer()]
+            _build_simple_graph().stream_events(
+                {"value": "x", "items": []},
+                version="v3",
+                transformers=[InstanceTransformer()],
             )
 
     def test_stream_channel_auto_forward(self) -> None:
@@ -1048,8 +1106,8 @@ class TestCustomTransformer:
                     self._channel.push("emitted")
                 return True
 
-        run = _build_simple_graph().stream_v2(
-            {"value": "x", "items": []}, 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
@@ -1093,8 +1151,10 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(ValueError, match=r"conflict.*'values'.*ValuesTransformer"):
-            _build_simple_graph().stream_v2(
-                {"value": "x", "items": []}, transformers=[ConflictTransformer]
+            _build_simple_graph().stream_events(
+                {"value": "x", "items": []},
+                version="v3",
+                transformers=[ConflictTransformer],
             )
 
 
@@ -1176,8 +1236,8 @@ class TestStreamChannelAutoLifecycle:
                     self._log.push("got_it")
                 return True
 
-        run = _build_simple_graph().stream_v2(
-            {"value": "x", "items": []}, transformers=[MinimalTransformer]
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3", transformers=[MinimalTransformer]
         )
         minimal_iter = iter(run.extensions["minimal"])
         _ = run.output
@@ -1437,8 +1497,8 @@ class TestAsyncTransformerLane:
             async def afinalize(self) -> None:
                 self._log.close()
 
-        run = await _build_simple_graph().astream_v2(
-            {"value": "x", "items": []}, 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()
@@ -1454,7 +1514,9 @@ class TestAsyncTransformerLane:
 @NEEDS_CONTEXTVARS
 class TestMemoryBounds:
     def test_sync_subscribed_buffer_stays_at_most_one_between_yields(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         events_iter = iter(run)
         max_buffered = 0
         count = 0
@@ -1467,7 +1529,9 @@ class TestMemoryBounds:
         )
 
     def test_unsubscribed_projections_never_accumulate(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        run = _build_simple_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         list(run)
         values_log = run.extensions["values"]
         messages_log = run.extensions["messages"]
@@ -1475,19 +1539,25 @@ class TestMemoryBounds:
         assert len(messages_log._items) == 0 and not messages_log._subscribed
 
     def test_output_path_does_not_retain_values(self) -> None:
-        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run = await _build_simple_graph().astream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         max_buffered = 0
         count = 0
         async for _ in run:
@@ -1498,7 +1568,9 @@ class TestMemoryBounds:
 
     @pytest.mark.anyio
     async def test_async_unsubscribed_projections_never_accumulate(self) -> None:
-        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
+        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

@@ -2210,7 +2210,6 @@ def test_graph_error_handler_does_not_swallow_interrupt_concurrent():
     )
 
 
-
 def test_node_error_handlers_route_to_matching_handler():
     class State(TypedDict):
         route: str
@@ -2268,4 +2267,3 @@ def test_node_without_error_handler_still_fails_run():
 
     with pytest.raises(ValueError, match="no handler"):
         graph.invoke({"foo": ""})
-

libs/langgraph/tests/test_stream_data_transformers.py

@@ -4,7 +4,7 @@ These transformers capture raw protocol events for their respective stream
 modes and expose them as native projections on the run stream (run.custom,
 run.updates, run.checkpoints, run.debug, run.tasks). Tests dispatch synthetic
 protocol events through a StreamMux to isolate transformer logic; the final
-group exercises real graphs through stream_v2.
+group exercises real graphs through stream_events(version="v3").
 """
 
 from __future__ import annotations
@@ -478,7 +478,7 @@ def test_unrelated_events_ignored_by_all() -> None:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: real graphs through stream_v2
+# End-to-end: real graphs through stream_events(version="v3")
 # ---------------------------------------------------------------------------
 
 
@@ -503,11 +503,11 @@ def _make_simple_graph() -> Any:
     return builder.compile()
 
 
-def test_stream_v2_custom_projection_opt_in() -> None:
+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_v2(
-        {"value": "hello", "items": []}, transformers=[CustomTransformer]
+    run = graph.stream_events(
+        {"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -515,11 +515,11 @@ def test_stream_v2_custom_projection_opt_in() -> None:
     assert any(e.get("status") == "working" for e in custom_events)
 
 
-def test_stream_v2_custom_and_values_coexist() -> 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_v2(
-        {"value": "hello", "items": []}, transformers=[CustomTransformer]
+    run = graph.stream_events(
+        {"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -528,10 +528,12 @@ def test_stream_v2_custom_and_values_coexist() -> None:
     assert len(custom_events) >= 1
 
 
-def test_stream_v2_tasks_projection_opt_in() -> 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_v2({"value": "x", "items": []}, transformers=[TasksTransformer])
+    run = graph.stream_events(
+        {"value": "x", "items": []}, transformers=[TasksTransformer], version="v3"
+    )
 
     tasks_events = list(run.tasks)
     assert len(tasks_events) >= 1
@@ -539,10 +541,12 @@ def test_stream_v2_tasks_projection_opt_in() -> None:
     assert "my_node" in names
 
 
-def test_stream_v2_debug_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_v2({"value": "x", "items": []}, transformers=[DebugTransformer])
+    run = graph.stream_events(
+        {"value": "x", "items": []}, transformers=[DebugTransformer], version="v3"
+    )
 
     debug_events = list(run.debug)
     assert len(debug_events) >= 1
@@ -550,11 +554,11 @@ def test_stream_v2_debug_projection_opt_in() -> None:
     assert types & {"checkpoint", "task", "task_result"}
 
 
-def test_stream_v2_updates_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_v2(
-        {"value": "x", "items": []}, transformers=[UpdatesTransformer]
+    run = graph.stream_events(
+        {"value": "x", "items": []}, version="v3", transformers=[UpdatesTransformer]
     )
 
     updates = list(run.updates)
@@ -563,11 +567,12 @@ def test_stream_v2_updates_projection_opt_in() -> None:
     assert "my_node" in node_names
 
 
-def test_stream_v2_all_transformers_interleaved() -> 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_v2(
+    run = graph.stream_events(
         {"value": "x", "items": []},
+        version="v3",
         transformers=[
             CustomTransformer,
             UpdatesTransformer,
@@ -599,7 +604,7 @@ def test_stream_v2_all_transformers_interleaved() -> None:
     assert run.output["value"] == "x!"
 
 
-def test_stream_v2_all_transformers_with_checkpointer() -> None:
+def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
     """All transformers with a checkpointer — run.checkpoints populated."""
     from langgraph.checkpoint.memory import InMemorySaver
 
@@ -609,8 +614,9 @@ def test_stream_v2_all_transformers_with_checkpointer() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_v2(
+    run = graph.stream_events(
         {"value": "x", "items": []},
+        version="v3",
         config={"configurable": {"thread_id": "test-all"}},
         transformers=[
             CustomTransformer,
@@ -637,7 +643,7 @@ def test_stream_v2_all_transformers_with_checkpointer() -> None:
     assert len(collected["custom"]) >= 1
 
 
-def test_stream_v2_checkpoints_projection_opt_in() -> None:
+def test_stream_events_v3_checkpoints_projection_opt_in() -> None:
     """run.checkpoints surfaces checkpoint data when opted in with a checkpointer."""
     from langgraph.checkpoint.memory import InMemorySaver
 
@@ -647,8 +653,9 @@ def test_stream_v2_checkpoints_projection_opt_in() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_v2(
+    run = graph.stream_events(
         {"value": "x", "items": []},
+        version="v3",
         config={"configurable": {"thread_id": "test-ckpt-standalone"}},
         transformers=[CheckpointsTransformer],
     )
@@ -688,8 +695,9 @@ def test_tasks_and_lifecycle_coregistration_e2e() -> None:
     is present and suppressing them from the main log.
     """
     graph = _make_simple_graph()
-    run = graph.stream_v2(
+    run = graph.stream_events(
         {"value": "x", "items": []},
+        version="v3",
         transformers=[TasksTransformer],
     )
 

libs/langgraph/tests/test_stream_events_v3.py

@@ -231,7 +231,7 @@ class TestV2Stream:
         for c in chunks:
             _assert_stream_part_shape(c)
 
-    def test_stream_v2_accepts_control_for_drain(self) -> None:
+    def test_stream_events_v3_accepts_control_for_drain(self) -> None:
         class DrainState(TypedDict, total=False):
             value: str
             skipped: str
@@ -253,7 +253,7 @@ class TestV2Stream:
         builder.add_edge("second", END)
         graph = builder.compile()
 
-        run = graph.stream_v2({}, control=control)
+        run = graph.stream_events({}, control=control, version="v3")
         with pytest.raises(GraphDrained, match="sigterm"):
             list(run.values)
 
@@ -1124,7 +1124,7 @@ class TestV2ValidationErrors:
 
     _INVALID_INPUT: dict[str, Any] = {"value": [1, 2, 3], "items": []}
 
-    def test_stream_v2_pydantic_validation_error(self) -> None:
+    def test_stream_events_v3_pydantic_validation_error(self) -> None:
         """Invalid input to stream with v2 + pydantic state raises ValidationError."""
         graph = _make_pydantic_graph()
         with pytest.raises(ValidationError):

libs/langgraph/tests/test_stream_events_v3_e2e.py

@@ -1,9 +1,9 @@
-"""End-to-end tests exercising all stream_v2 projections together.
+"""End-to-end tests exercising all stream_events(version="v3") projections together.
 
 Each test builds a realistic graph (subgraphs, LLM calls, custom writers,
 interrupts) and verifies that every projection — values, messages, lifecycle,
 subgraphs, raw events, output, interleave — produces correct, consistent
-results through a single stream_v2 / astream_v2 run.
+results through a single stream_events(version="v3") / astream_events(version="v3") run.
 """
 
 from __future__ import annotations
@@ -218,9 +218,9 @@ class _CounterTransformer(StreamTransformer):
 
 class TestStreamV2E2ESync:
     def test_all_projections_nested_graph(self) -> None:
-        """Run a nested graph through stream_v2 and verify values + lifecycle."""
+        """Run a nested graph through stream_events(version="v3") and verify values + lifecycle."""
         graph = _make_nested_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         values_snapshots: list[dict[str, Any]] = []
         lifecycle_events: list[dict[str, Any]] = []
@@ -246,7 +246,7 @@ class TestStreamV2E2ESync:
     def test_subgraph_handles_with_drill_down(self) -> None:
         """Subgraph handles yield and support values drill-down."""
         graph = _make_nested_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         handles = []
         for handle in run.subgraphs:
@@ -270,7 +270,7 @@ class TestStreamV2E2ESync:
     def test_raw_events_have_monotonic_seq(self) -> None:
         """Raw protocol events have monotonically increasing seq numbers."""
         graph = _make_nested_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
         events = list(run)
         assert len(events) > 0
 
@@ -285,11 +285,15 @@ class TestStreamV2E2ESync:
 
     def test_output_matches_final_values_snapshot(self) -> None:
         """output property returns the same state as the last values snapshot."""
-        run1 = _make_nested_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run2 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         final_via_output = run2.output
 
         assert final_via_values == final_via_output
@@ -297,7 +301,7 @@ class TestStreamV2E2ESync:
     def test_context_manager_and_abort(self) -> None:
         """Context manager calls abort, marking the stream exhausted."""
         graph = _make_nested_graph()
-        with graph.stream_v2({"value": "x", "items": []}) as run:
+        with graph.stream_events({"value": "x", "items": []}, version="v3") as run:
             first_val = next(iter(run.values))
             assert isinstance(first_val, dict)
         assert run._exhausted is True
@@ -305,7 +309,7 @@ class TestStreamV2E2ESync:
     def test_extensions_has_all_native_keys(self) -> None:
         """Extensions dict exposes all native projection keys."""
         graph = _make_nested_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
         _ = run.output
 
         assert "values" in run.extensions
@@ -327,7 +331,7 @@ class TestStreamV2E2EMessages:
     def test_messages_projection_from_invoke(self) -> None:
         """Messages projection captures LLM calls via model.invoke() auto-routing."""
         graph = _make_messages_graph()
-        run = graph.stream_v2({"messages": "hi"})
+        run = graph.stream_events({"messages": "hi"}, version="v3")
         streams = list(run.messages)
 
         assert len(streams) >= 1
@@ -350,7 +354,7 @@ class TestStreamV2E2EMessages:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "go"})
+        run = graph.stream_events({"messages": "go"}, version="v3")
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "streamed answer"
 
@@ -368,7 +372,7 @@ class TestStreamV2E2EMessages:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "hi"})
+        run = graph.stream_events({"messages": "hi"}, version="v3")
         (stream,) = list(run.messages)
         assert stream.output.text == "hardcoded"
         assert stream.message_id == "msg-1"
@@ -376,7 +380,7 @@ class TestStreamV2E2EMessages:
     def test_root_messages_only_shows_root_scope(self) -> None:
         """Root messages projection doesn't surface subgraph-scoped messages."""
         graph = _make_messages_subgraph()
-        run = graph.stream_v2({"messages": ["hi"], "done": False})
+        run = graph.stream_events({"messages": ["hi"], "done": False}, version="v3")
         root_streams = list(run.messages)
         # The message is emitted inside the subgraph, so the root
         # messages projection (scoped to root namespace) doesn't see it.
@@ -385,7 +389,7 @@ class TestStreamV2E2EMessages:
     def test_subgraph_handle_messages_drill_down(self) -> None:
         """Drilling into subgraph handle's messages surfaces subgraph messages."""
         graph = _make_messages_subgraph()
-        run = graph.stream_v2({"messages": ["hi"], "done": False})
+        run = graph.stream_events({"messages": ["hi"], "done": False}, version="v3")
 
         found_messages = False
         for handle in run.subgraphs:
@@ -407,8 +411,9 @@ 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_v2(
+        run = graph.stream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CustomPassthroughTransformer],
         )
         events = list(run)
@@ -420,7 +425,7 @@ class TestStreamV2E2ECustom:
     def test_custom_events_suppressed_without_transformer(self) -> None:
         """Without a custom-mode transformer, custom events don't flow."""
         graph = _make_custom_writer_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
         events = list(run)
         custom = [e for e in events if e["method"] == "custom"]
         assert custom == []
@@ -428,8 +433,9 @@ 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_v2(
+        run = graph.stream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CounterTransformer],
         )
 
@@ -444,8 +450,9 @@ 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_v2(
+        run = graph.stream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CounterTransformer],
         )
         events = list(run)
@@ -464,7 +471,7 @@ class TestStreamV2E2EInterrupt:
         """Interrupted run has correct flags and interrupt payloads."""
         graph = _make_interrupt_graph()
         config: dict[str, Any] = {"configurable": {"thread_id": "int-1"}}
-        run = graph.stream_v2({"value": "x", "items": []}, config)
+        run = graph.stream_events({"value": "x", "items": []}, config, version="v3")
 
         output = run.output
         assert output is not None
@@ -477,7 +484,7 @@ class TestStreamV2E2EInterrupt:
         """Values snapshots captured before the interrupt reflect partial state."""
         graph = _make_interrupt_graph()
         config: dict[str, Any] = {"configurable": {"thread_id": "int-2"}}
-        run = graph.stream_v2({"value": "x", "items": []}, config)
+        run = graph.stream_events({"value": "x", "items": []}, config, version="v3")
 
         snapshots = list(run.values)
         assert len(snapshots) >= 1
@@ -494,14 +501,14 @@ class TestStreamV2E2EErrors:
     def test_subgraph_error_propagates_through_output(self) -> None:
         """Error in a subgraph propagates through output."""
         graph = _make_error_subgraph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         with pytest.raises(ValueError, match="subgraph explosion"):
             _ = run.output
 
     def test_subgraph_error_propagates_through_raw_events(self) -> None:
         graph = _make_error_subgraph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         with pytest.raises(ValueError, match="subgraph explosion"):
             list(run)
@@ -509,7 +516,7 @@ class TestStreamV2E2EErrors:
     def test_error_subgraph_handle_status(self) -> None:
         """Subgraph handle surfaces the error status."""
         graph = _make_error_subgraph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         handle = next(iter(run.subgraphs))
         with pytest.raises(RuntimeError, match="subgraph explosion"):
@@ -529,7 +536,7 @@ class TestStreamV2E2EAsync:
     async def test_all_projections_async(self) -> None:
         """Async run exercises values projection."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
 
         values_snapshots = [s async for s in run.values]
         assert len(values_snapshots) >= 1
@@ -540,7 +547,7 @@ class TestStreamV2E2EAsync:
     async def test_async_output(self) -> None:
         """Async output returns the final state."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
         output = await run.output()
         assert output is not None
         assert output["value"] == "x_routed_processed"
@@ -550,7 +557,7 @@ class TestStreamV2E2EAsync:
     async def test_async_raw_events(self) -> None:
         """Async raw event iteration yields well-formed ProtocolEvents."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
         events = [e async for e in run]
         assert len(events) > 0
         seqs = [e["seq"] for e in events]
@@ -572,7 +579,7 @@ class TestStreamV2E2EAsync:
             .compile()
         )
 
-        run = await graph.astream_v2({"messages": "hi"})
+        run = await graph.astream_events({"messages": "hi"}, version="v3")
         streams = [s async for s in run.messages]
         assert len(streams) >= 1
         for s in streams:
@@ -583,7 +590,9 @@ class TestStreamV2E2EAsync:
         """Async interrupted run has correct flags."""
         graph = _make_interrupt_graph()
         config: dict[str, Any] = {"configurable": {"thread_id": "async-int-1"}}
-        run = await graph.astream_v2({"value": "x", "items": []}, config)
+        run = await graph.astream_events(
+            {"value": "x", "items": []}, config, version="v3"
+        )
 
         output = await run.output()
         assert output is not None
@@ -593,14 +602,14 @@ class TestStreamV2E2EAsync:
     async def test_async_error_propagation(self) -> None:
         """Async error from subgraph propagates through output."""
         graph = _make_error_subgraph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
         with pytest.raises(ValueError, match="subgraph explosion"):
             await run.output()
 
     async def test_async_context_manager(self) -> None:
         """Async context manager calls abort on exit."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
         async with run:
             _ = await anext(aiter(run.values))
         assert run._exhausted is True
@@ -608,7 +617,7 @@ class TestStreamV2E2EAsync:
     async def test_async_extensions_present(self) -> None:
         """Async run has all native extensions."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
         _ = await run.output()
         assert "values" in run.extensions
         assert "messages" in run.extensions
@@ -618,8 +627,9 @@ class TestStreamV2E2EAsync:
     async def test_async_custom_transformer(self) -> None:
         """Async custom transformer with StreamChannel works."""
         graph = _make_nested_graph()
-        run = await graph.astream_v2(
+        run = await graph.astream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CounterTransformer],
         )
         assert "counter" in run.extensions
@@ -639,7 +649,7 @@ class TestStreamV2E2ECombined:
     def test_interleave_all_native_projections(self) -> None:
         """Interleave values + messages + lifecycle without deadlock."""
         graph = _make_nested_graph()
-        run = graph.stream_v2({"value": "x", "items": []})
+        run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
         seen_names: set[str] = set()
         for name, _item in run.interleave("values", "messages", "lifecycle"):
@@ -667,8 +677,9 @@ class TestStreamV2E2ECombined:
                 return True
 
         graph = _make_nested_graph()
-        run = graph.stream_v2(
+        run = graph.stream_events(
             {"value": "x", "items": []},
+            version="v3",
             transformers=[_CounterTransformer, TagTransformer],
         )
 
@@ -722,7 +733,7 @@ class TestStreamV2E2ECombined:
             .compile()
         )
 
-        run = outer.stream_v2({"items": []})
+        run = outer.stream_events({"items": []}, version="v3")
         handles = []
         for handle in run.subgraphs:
             list(handle.values)
@@ -740,13 +751,17 @@ class TestStreamV2E2ECombined:
 
     def test_lifecycle_matches_subgraph_handles(self) -> None:
         """Lifecycle events and subgraph handles agree on discovered subgraphs."""
-        run1 = _make_nested_graph().stream_v2({"value": "x", "items": []})
+        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_v2({"value": "x", "items": []})
+        run2 = _make_nested_graph().stream_events(
+            {"value": "x", "items": []}, version="v3"
+        )
         lifecycle = list(run2.lifecycle)
 
         started_ns = [
@@ -773,8 +788,9 @@ class TestStreamV2E2ECombined:
             .compile()
         )
 
-        run = graph.stream_v2(
+        run = graph.stream_events(
             {"messages": "hi"},
+            version="v3",
             transformers=[_CounterTransformer],
         )
 

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -6,7 +6,7 @@ on the `lifecycle` channel for both in-process iteration via
 events. Most tests dispatch synthetic protocol events through a
 `StreamMux` to keep the inference logic isolated; the end-of-file
 group exercises the path through real graphs (multi-depth
-discovery, nested `stream_v2` calls with non-empty `parent_ns`).
+discovery, nested `stream_events(version="v3")` calls with non-empty `parent_ns`).
 """
 
 from __future__ import annotations
@@ -324,7 +324,7 @@ def test_tasks_events_suppressed_from_main_log() -> None:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: real graphs through stream_v2
+# End-to-end: real graphs through stream_events(version="v3")
 # ---------------------------------------------------------------------------
 
 
@@ -358,10 +358,10 @@ def _make_two_level_nested() -> Any:
     return outer_b.compile()
 
 
-def test_stream_v2_real_graph_emits_lifecycle_at_each_depth() -> None:
+def test_stream_events_v3_real_graph_emits_lifecycle_at_each_depth() -> None:
     """Outer graph with two nested subgraphs surfaces lifecycle for both."""
     graph = _make_two_level_nested()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     # Iterating the projection drives the pump and drains synthesized
     # lifecycle events at the same time.
@@ -384,17 +384,17 @@ def test_stream_v2_real_graph_emits_lifecycle_at_each_depth() -> None:
         )
 
 
-def test_stream_v2_with_nested_parent_ns_scopes_lifecycle() -> None:
-    """When `stream_v2` is called with a non-empty checkpoint_ns in config,
+def test_stream_events_v3_with_nested_parent_ns_scopes_lifecycle() -> None:
+    """When `stream_events(version="v3")` is called with a non-empty checkpoint_ns in config,
     `_resolve_parent_ns` returns that namespace and the registered
     `LifecycleTransformer` is constructed with `scope=parent_ns`. This
-    exercises the path that exists today purely for nested-stream_v2
+    exercises the path that exists today purely for nested-stream_events(version="v3")
     callers; the test simulates such a caller by injecting a
     checkpoint_ns into the config.
     """
     graph = _make_two_level_nested()
     config = {CONF: {CONFIG_KEY_CHECKPOINT_NS: "outer:abc"}}
-    run = graph.stream_v2({"value": "x", "items": []}, config=config)
+    run = graph.stream_events({"value": "x", "items": []}, config=config, version="v3")
 
     payloads = list(run.lifecycle)
     # Every emitted lifecycle namespace must extend the caller's scope —

libs/langgraph/tests/test_stream_messages_transformer.py

@@ -1,5 +1,5 @@
 """Tests for MessagesTransformer: protocol event routing, whole-message fallback,
-legacy v1 chunk filtering, and end-to-end via stream_v2 / astream_v2."""
+legacy v1 chunk filtering, and end-to-end via stream_events(version="v3") / astream_events(version="v3")."""
 
 from __future__ import annotations
 
@@ -152,7 +152,7 @@ def _lifecycle(
 def _simple_graph():
     def call_model(state: MessagesState) -> dict[str, Any]:
         model = GenericFakeChatModel(messages=iter(["hello world"]))
-        stream = model.stream_v2(state["messages"])
+        stream = model.stream_events(state["messages"], version="v3")
         return {"messages": stream.output}
 
     return (
@@ -327,7 +327,7 @@ class TestFiltering:
 
     def test_legacy_v1_chunks_ignored(self) -> None:
         # v1 AIMessageChunk tuples (from on_llm_new_token) are not streamed
-        # into this projection; callers must migrate to stream_v2.
+        # into this projection; callers must migrate to stream_events(version="v3").
         t, log = _make_sync_transformer()
         t.process(_v1_chunk("hello"))
         t.process(_v1_chunk(" world", finish=True))
@@ -477,18 +477,18 @@ class TestViaMux:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: graph → stream_v2 → run.messages (node calls stream_v2)
+# End-to-end: graph → stream_events(version="v3") → run.messages (node calls stream_events)
 # ---------------------------------------------------------------------------
 
 
 class TestEndToEnd:
-    """stream_v2 path: node calls model.stream_v2() explicitly."""
+    """stream_events(version="v3") path: node calls model.stream_events() explicitly."""
 
     def test_node_calling_stream_v2_populates_messages(self) -> None:
         model = GenericFakeChatModel(messages=iter(["hello world"]))
 
         def call_model(state: MessagesState) -> dict[str, Any]:
-            stream = model.stream_v2(state["messages"])
+            stream = model.stream_events(state["messages"], version="v3")
             return {"messages": stream.output}
 
         graph = (
@@ -499,7 +499,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "hi"})
+        run = graph.stream_events({"messages": "hi"}, version="v3")
         (stream,) = list(run.messages)
         assert isinstance(stream, ChatModelStream)
         assert stream.output.text == "hello world"
@@ -509,7 +509,7 @@ class TestEndToEnd:
         model = GenericFakeChatModel(messages=iter(["streamed answer"]))
 
         def call_model(state: MessagesState) -> dict[str, Any]:
-            stream = model.stream_v2(state["messages"])
+            stream = model.stream_events(state["messages"], version="v3")
             return {"messages": stream.output}
 
         graph = (
@@ -520,7 +520,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "go"})
+        run = graph.stream_events({"messages": "go"}, version="v3")
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "streamed answer"
 
@@ -538,7 +538,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "hi"})
+        run = graph.stream_events({"messages": "hi"}, version="v3")
         (stream,) = list(run.messages)
         assert stream.output.text == "hardcoded"
 
@@ -547,7 +547,7 @@ class TestEndToEnd:
         model = GenericFakeChatModel(messages=iter(["async answer"]))
 
         async def call_model(state: MessagesState) -> dict[str, Any]:
-            stream = await model.astream_v2(state["messages"])
+            stream = await model.astream_events(state["messages"], version="v3")
             return {"messages": await stream}
 
         graph = (
@@ -558,7 +558,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = await graph.astream_v2({"messages": "hi"})
+        run = await graph.astream_events({"messages": "hi"}, version="v3")
         streams = [s async for s in run.messages]
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
@@ -572,7 +572,7 @@ class TestEndToEnd:
         model = GenericFakeChatModel(messages=iter(["hello world"]))
 
         async def call_model(state: MessagesState) -> dict[str, Any]:
-            stream = await model.astream_v2(state["messages"])
+            stream = await model.astream_events(state["messages"], version="v3")
             return {"messages": await stream}
 
         graph = (
@@ -583,7 +583,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = await graph.astream_v2({"messages": "hi"})
+        run = await graph.astream_events({"messages": "hi"}, version="v3")
 
         async def consume() -> list[str]:
             collected: list[str] = []
@@ -596,12 +596,12 @@ class TestEndToEnd:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: graph → stream_v2 → run.messages (node calls invoke)
+# End-to-end: graph → stream_events(version="v3") → run.messages (node calls invoke)
 # ---------------------------------------------------------------------------
 
 
 class TestEndToEndV2Invoke:
-    """Auto-routing path: stream_v2 injects CONFIG_KEY_STREAM_MESSAGES_V2,
+    """Auto-routing path: stream_events(version="v3") injects CONFIG_KEY_STREAM_MESSAGES_V2,
     causing BaseChatModel to drive the v2 protocol event generator even for
     model.invoke()."""
 
@@ -620,7 +620,7 @@ class TestEndToEndV2Invoke:
     def test_invoke_populates_messages(self) -> None:
         run = self._graph(
             GenericFakeChatModel(messages=iter(["hello world"]))
-        ).stream_v2({"messages": "hi"})
+        ).stream_events({"messages": "hi"}, version="v3")
         (stream,) = list(run.messages)
         assert isinstance(stream, ChatModelStream)
         assert stream.output.text == "hello world"
@@ -629,7 +629,7 @@ class TestEndToEndV2Invoke:
         """Iterating the stream yields the full v2 lifecycle, not v1 chunks."""
         run = self._graph(
             GenericFakeChatModel(messages=iter(["streamed answer"]))
-        ).stream_v2({"messages": "go"})
+        ).stream_events({"messages": "go"}, version="v3")
         (stream,) = list(run.messages)
 
         events = list(stream)
@@ -650,7 +650,7 @@ class TestEndToEndV2Invoke:
     def test_invoke_text_deltas_iterate(self) -> None:
         run = self._graph(
             GenericFakeChatModel(messages=iter(["delta streaming works"]))
-        ).stream_v2({"messages": "hi"})
+        ).stream_events({"messages": "hi"}, version="v3")
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "delta streaming works"
 
@@ -674,7 +674,7 @@ class TestEndToEndV2Invoke:
             .compile()
         )
 
-        streams = list(graph.stream_v2({"messages": "hi"}).messages)
+        streams = list(graph.stream_events({"messages": "hi"}, version="v3").messages)
         assert len(streams) == 2
         assert {s.output.text for s in streams} == {"alpha", "beta"}
 
@@ -698,7 +698,7 @@ class TestEndToEndV2Invoke:
             .compile()
         )
 
-        run = graph.stream_v2({"messages": "hi"})
+        run = graph.stream_events({"messages": "hi"}, version="v3")
         streams = list(run.messages)
         assert len(streams) == 2
         assert streams[0].node == "streaming_node"
@@ -722,7 +722,7 @@ class TestEndToEndV2Invoke:
             .compile()
         )
 
-        run = await graph.astream_v2({"messages": "hi"})
+        run = await graph.astream_events({"messages": "hi"}, version="v3")
         streams = [s async for s in run.messages]
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
@@ -737,7 +737,7 @@ class TestEndToEndV2Invoke:
 class TestDirectMessagesModeStaysV1:
     def test_direct_graph_stream_messages_yields_ai_message_chunks(self) -> None:
         """graph.stream(stream_mode="messages") must not leak v2 event dicts —
-        the v2 flag is only injected by stream_v2 / astream_v2."""
+        the v2 flag is only injected by stream_events(version="v3") / astream_events(version="v3")."""
         model = GenericFakeChatModel(messages=iter(["legacy path"]))
 
         def call_model(state: MessagesState) -> dict[str, Any]:
@@ -760,8 +760,10 @@ class TestDirectMessagesModeStaysV1:
             == "legacy path"
         )
 
-    def test_nested_graph_stream_messages_stays_v1_under_outer_stream_v2(self) -> None:
-        """An outer `stream_v2()` run must not flip an inner direct
+    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"]))
 
@@ -812,7 +814,7 @@ class TestDirectMessagesModeStaysV1:
             .compile()
         )
 
-        result = outer.stream_v2({}).output
+        result = outer.stream_events({}, version="v3").output
 
         assert result is not None
         assert result["saw_only_chunks"] is True

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -4,7 +4,7 @@ Subscribes to `tasks` events and produces in-process `SubgraphRunStream`
 handles backed by mini-muxes (built via `StreamMux._make_child`). The
 synthetic-event tests isolate the inference / mini-mux wiring; the
 real-graph tests exercise the end-to-end navigation path through
-`stream_v2`.
+`stream_events(version="v3")`.
 """
 
 from __future__ import annotations
@@ -93,7 +93,7 @@ def _tasks_result(
 
 
 def _native_factories() -> list[Any]:
-    """Mirror the factory list `Pregel.stream_v2` registers."""
+    """Mirror the factory list `Pregel.stream_events(version="v3")` registers."""
     return [
         ValuesTransformer,
         MessagesTransformer,
@@ -761,10 +761,10 @@ def _make_failing_nested() -> Any:
     return outer_b.compile()
 
 
-def test_stream_v2_real_graph_yields_subgraph_handles() -> None:
+def test_stream_events_v3_real_graph_yields_subgraph_handles() -> None:
     """Iterating `run.subgraphs` yields handles for direct-child subgraphs."""
     graph = _make_two_level_nested()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     handle_paths: list[tuple[str, ...]] = []
     final_status: dict[tuple[str, ...], str] = {}
@@ -780,10 +780,10 @@ def test_stream_v2_real_graph_yields_subgraph_handles() -> None:
     assert final_status[handle_paths[0]] == "completed"
 
 
-def test_stream_v2_grandchild_visible_on_child_handle() -> None:
+def test_stream_events_v3_grandchild_visible_on_child_handle() -> None:
     """Drilling into `handle.subgraphs` surfaces nested grandchildren."""
     graph = _make_two_level_nested()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     grandchild_paths: list[tuple[str, ...]] = []
     middle_path: tuple[str, ...] | None = None
@@ -810,7 +810,7 @@ def test_subgraph_output_stops_at_own_terminal_without_draining_siblings() -> No
     inside the loop body misses its events.
     """
     graph = _make_two_sibling_subgraphs()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     paths: list[tuple[str, ...]] = []
     second_values: list[dict[str, Any]] = []
@@ -829,7 +829,7 @@ def test_subgraph_output_stops_at_own_terminal_without_draining_siblings() -> No
 
 def test_aborted_subgraph_handle_does_not_fail_parent_forwarding() -> None:
     graph = _make_two_sibling_subgraphs()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     seen: list[str | None] = []
     for handle in run.subgraphs:
@@ -847,7 +847,7 @@ def test_aborted_subgraph_handle_does_not_fail_parent_forwarding() -> None:
 
 def test_failed_subgraph_output_raises_terminal_error() -> None:
     graph = _make_failing_nested()
-    run = graph.stream_v2({"value": "x", "items": []})
+    run = graph.stream_events({"value": "x", "items": []}, version="v3")
 
     handle = next(iter(run.subgraphs))
     with pytest.raises(RuntimeError, match="child boom"):

libs/langgraph/uv.lock

@@ -1348,7 +1348,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -1361,26 +1361,26 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3c/93/68bafa047f8e1770d0cf0f61d6c70889f1dec42ef6bd263540d916c421b9/langchain_core-1.4.0a2.tar.gz", hash = "sha256:b723c7961b615c7f2180ce2bcf352fdad8247bc51a60adecd3d97088235c120d", size = 916486, upload-time = "2026-05-01T15:02:19.029Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
+    { url = "https://files.pythonhosted.org/packages/4e/8e/933e0ba7ba0430ce264e36b178d581b255239ad45093872483142d93478c/langchain_core-1.4.0a2-py3-none-any.whl", hash = "sha256:a5c689f8404357df797120c012da7704144a953b2ae18f258df263301e7badd5", size = 546297, upload-time = "2026-05-01T15:02:17.731Z" },
 ]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.12"
+version = "0.0.14"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/05/bf/efb5e2ed832e4d6d45590e25a9e5191986b291b543bc6a807b48bee070b0/langchain_protocol-0.0.14.tar.gz", hash = "sha256:bc1e8553122e6ede310280462d5813023a172ff2785ccbbdec54d43f3a15e5f2", size = 5862, upload-time = "2026-04-29T16:40:18.657Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" },
+    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a3"
+version = "1.2.0a4"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1452,7 +1452,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1755,7 +1755,7 @@ test = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "1.0.12"
+version = "1.1.0a1"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/pyproject.toml

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

libs/prebuilt/tests/test_tool_call_transformer.py

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

libs/prebuilt/uv.lock

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

libs/sdk-py/uv.lock

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