release: alpha bump (a3) for langgraph, checkpoint, checkpoint-postgres (#7678)

## Summary
- Bumps `langgraph` 1.2.0a1 → 1.2.0a3
- Bumps `langgraph-checkpoint` 4.1.0a1 → 4.1.0a3
- Bumps `langgraph-checkpoint-postgres` 3.1.0a1 → 3.1.0a3
- Bumps min `langgraph-checkpoint` constraint in `langgraph` to
`>=4.1.0a3`
- Refreshes uv locks across the workspace

(Note: `a2` was already cut from another branch.)

## Test plan
- [ ] CI passes

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

@@ -19,6 +19,7 @@ from langgraph.checkpoint.base import (
     get_serializable_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.base import SerializerProtocol
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from psycopg import Capabilities, Connection, Cursor, Pipeline
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -26,9 +27,11 @@ from psycopg_pool import ConnectionPool
 
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_STAGE1_SQL,
+    SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
+    _DeltaStage1Row,
+    _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -308,7 +311,12 @@ class PostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            if v is None or isinstance(v, (str, int, float, bool)):
+            if v is DELTA_SENTINEL:
+                copy["channel_values"].pop(k)
+            elif isinstance(v, _DeltaSnapshot):
+                blob_values[k] = copy["channel_values"].pop(k)
+                copy["channel_values"][k] = True
+            elif v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -441,41 +449,49 @@ class PostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
 
-        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
-        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
-        single roundtrip; the ancestor walk runs in Python.
+        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.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
         checkpoint_id = get_checkpoint_id(config)
         if checkpoint_id is None:
-            # Caller didn't specify a target — resolve to the latest
-            # checkpoint on the thread. `get_tuple` without `checkpoint_id`
-            # returns the newest; its config carries the resolved id.
             target = self.get_tuple(config)
             if target is None:
                 return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
+
         with self._cursor() as cur:
             cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
+                SELECT_DELTA_STAGE1_SQL,
+                (channel, channel, thread_id, checkpoint_ns),
+            )
+            stage1_rows = cur.fetchall()
+        chain_cids, seed_version = self._walk_stage1(
+            cast("list[_DeltaStage1Row]", stage1_rows), checkpoint_id
+        )
+        seed_versions = [seed_version] if seed_version else []
+        with self._cursor() as cur:
+            cur.execute(
+                SELECT_DELTA_STAGE2_SQL,
                 (
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
                     thread_id,
                     checkpoint_ns,
                     channel,
+                    chain_cids,
                     thread_id,
                     checkpoint_ns,
                     channel,
+                    seed_versions,
                 ),
             )
-            rows = cur.fetchall()
+            stage2_rows = cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
-            target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            chain_cids=chain_cids,
+            seed_version=seed_version,
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -19,6 +19,7 @@ from langgraph.checkpoint.base import (
     get_serializable_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.base import SerializerProtocol
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
@@ -26,9 +27,11 @@ from psycopg_pool import AsyncConnectionPool
 
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
-    SELECT_DELTA_COMBINED_SQL,
+    SELECT_DELTA_STAGE1_SQL,
+    SELECT_DELTA_STAGE2_SQL,
     BasePostgresSaver,
-    _DeltaCombinedRow,
+    _DeltaStage1Row,
+    _DeltaStage2Row,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -267,7 +270,12 @@ class AsyncPostgresSaver(BasePostgresSaver):
         # others are stored in blobs table
         blob_values = {}
         for k, v in checkpoint["channel_values"].items():
-            if v is None or isinstance(v, (str, int, float, bool)):
+            if v is DELTA_SENTINEL:
+                copy["channel_values"].pop(k)
+            elif isinstance(v, _DeltaSnapshot):
+                blob_values[k] = copy["channel_values"].pop(k)
+                copy["channel_values"][k] = True
+            elif v is None or isinstance(v, (str, int, float, bool)):
                 pass
             else:
                 blob_values[k] = copy["channel_values"].pop(k)
@@ -402,10 +410,9 @@ class AsyncPostgresSaver(BasePostgresSaver):
     ) -> _ChannelWritesHistory:
         """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
 
-        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
-        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
-        single roundtrip; rows are assembled by the shared pure helper on
-        `BasePostgresSaver`.
+        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.
         """
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
@@ -415,26 +422,37 @@ class AsyncPostgresSaver(BasePostgresSaver):
             if target is None:
                 return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
             checkpoint_id = target.config["configurable"]["checkpoint_id"]
+
         async with self._cursor() as cur:
             await cur.execute(
-                SELECT_DELTA_COMBINED_SQL,
+                SELECT_DELTA_STAGE1_SQL,
+                (channel, channel, thread_id, checkpoint_ns),
+            )
+            stage1_rows = await cur.fetchall()
+        chain_cids, seed_version = self._walk_stage1(
+            cast("list[_DeltaStage1Row]", stage1_rows), checkpoint_id
+        )
+        seed_versions = [seed_version] if seed_version else []
+        async with self._cursor() as cur:
+            await cur.execute(
+                SELECT_DELTA_STAGE2_SQL,
                 (
-                    channel,
-                    thread_id,
-                    checkpoint_ns,
                     thread_id,
                     checkpoint_ns,
                     channel,
+                    chain_cids,
                     thread_id,
                     checkpoint_ns,
                     channel,
+                    seed_versions,
                 ),
             )
-            rows = await cur.fetchall()
+            stage2_rows = await cur.fetchall()
         return self._build_delta_channel_writes_history(
             channel=channel,
-            target_id=checkpoint_id,
-            rows=cast("list[_DeltaCombinedRow]", rows),
+            chain_cids=chain_cids,
+            seed_version=seed_version,
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -156,62 +156,62 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """
 
 
-class _DeltaCombinedRow(TypedDict, total=False):
-    """One row from `SELECT_DELTA_COMBINED_SQL` (a UNION ALL of three tables).
+class _DeltaStage2Row(TypedDict, total=False):
+    """One row from `SELECT_DELTA_STAGE2_SQL` (a UNION ALL of writes and blobs)."""
 
-    Every row carries `_kind` ("p" / "w" / "b") plus whichever columns are
-    relevant for that kind; irrelevant columns are NULL and typed as `None`.
-    """
-
-    _kind: str  # always present: "p", "w", or "b"
-    # checkpoint row ("p")
-    checkpoint_id: str | None
-    parent_checkpoint_id: str | None
-    ver: str | None
-    # write / blob rows ("w", "b")
+    _kind: str  # "w" or "b"
+    checkpoint_id: str | None  # "w" rows only
     type: str | None
     blob: bytes | None
-    # write row only ("w")
-    task_id: str | None
-    idx: int | None
-    # blob row only ("b")
-    version: str | None
+    task_id: str | None  # "w" rows only
+    idx: int | None  # "w" rows only
+    version: str | None  # "b" rows only
 
 
-# DeltaChannel reconstruction: one UNION ALL query fetches checkpoints,
-# writes, and blobs for `channel` in one roundtrip; the ancestor walk runs
-# in Python in `_build_delta_channel_writes_history`.
+# 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.
 #
-# Parameter order: (channel, thread_id, checkpoint_ns,
-#                   thread_id, checkpoint_ns, channel,
-#                   thread_id, checkpoint_ns, channel)
-SELECT_DELTA_COMBINED_SQL = """
-    SELECT 'p'::text       AS _kind,
-           checkpoint_id,
+# Parameter order:
+#   stage1: (channel, channel, thread_id, checkpoint_ns)
+#   stage2: (thread_id, checkpoint_ns, channel, chain_cids[],
+#            thread_id, checkpoint_ns, channel, seed_versions[])
+
+SELECT_DELTA_STAGE1_SQL = """
+    SELECT checkpoint_id,
            parent_checkpoint_id,
            checkpoint -> 'channel_versions' ->> %s AS ver,
-           NULL::text       AS type,
-           NULL::bytea      AS blob,
-           NULL::text       AS task_id,
-           NULL::int        AS idx,
-           NULL::text       AS version
+           (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS has_snapshot
     FROM checkpoints
     WHERE thread_id = %s AND checkpoint_ns = %s
-    UNION ALL
-    SELECT 'w',
-           checkpoint_id, NULL, NULL,
-           type, blob, task_id, idx, NULL
+"""
+
+SELECT_DELTA_STAGE2_SQL = """
+    SELECT 'w'::text AS _kind,
+           checkpoint_id,
+           type, blob, task_id, idx, NULL::text AS version
     FROM checkpoint_writes
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+      AND checkpoint_id = ANY(%s)
     UNION ALL
-    SELECT 'b',
-           NULL, NULL, NULL,
+    SELECT 'b', NULL,
            type, blob, NULL, NULL, version
     FROM checkpoint_blobs
     WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %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
+
+
 class BasePostgresSaver(BaseCheckpointSaver[str]):
     SELECT_SQL = SELECT_SQL
     SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
@@ -254,38 +254,59 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             if t.decode() != "empty"
         }
 
+    @staticmethod
+    def _walk_stage1(
+        stage1_rows: Sequence[_DeltaStage1Row],
+        target_id: str,
+    ) -> tuple[list[str], str | None]:
+        """Walk the parent chain from stage 1 metadata rows.
+
+        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.
+        """
+        parent_of: dict[str, str | None] = {}
+        ver_of: dict[str, str | None] = {}
+        snapshot_of: dict[str, bool] = {}
+        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"]
+
+        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
+
     def _build_delta_channel_writes_history(
         self,
         *,
         channel: str,
-        target_id: str,
-        rows: Sequence[_DeltaCombinedRow],
+        chain_cids: list[str],
+        seed_version: str | None,
+        stage2_rows: Sequence[_DeltaStage2Row],
     ) -> _ChannelWritesHistory:
-        """Reconstruct one delta channel's history from the combined UNION ALL rows.
+        """Reconstruct delta channel history from two-stage query results.
 
-        Pure data transform shared by sync (`PostgresSaver`) and async
-        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
-        and feed the tagged rows here.
-
-        Walk is newest → oldest from the target's parent. A non-sentinel
-        blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
-        walk and is returned as the seed so replay starts from it.
-
-        Writes stored at `target_id` itself are pending writes for the next
-        step and are excluded — the walk begins at the target's parent.
+        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.
         """
-        parent_of: dict[str, str | None] = {}
-        ver_of: dict[str, str | None] = {}
         writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
-        blob_by_ver: dict[str, tuple[str, bytes]] = {}
+        seed_blob: tuple[str, bytes] | None = None
 
-        for r in rows:
+        for r in stage2_rows:
             kind = r["_kind"]
-            if kind == "p":
-                cid = cast(str, r["checkpoint_id"])
-                parent_of[cid] = r["parent_checkpoint_id"]
-                ver_of[cid] = r["ver"]
-            elif kind == "w":
+            if kind == "w":
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_cid.setdefault(cid, []).append(
                     cast(
@@ -294,42 +315,26 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     )
                 )
             else:  # kind == "b"
-                blob_by_ver[cast(str, r["version"])] = cast(
-                    "tuple[str, bytes]", (r["type"], r["blob"])
-                )
+                seed_blob = cast("tuple[str, bytes]", (r["type"], r["blob"]))
 
-        # newest write first per ancestor (task_id DESC, idx DESC)
         for ws in writes_by_cid.values():
             ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
-        ancestors: list[str] = []
-        cur_cid: str | None = parent_of.get(target_id)
-        while cur_cid is not None:
-            ancestors.append(cur_cid)
-            cur_cid = parent_of.get(cur_cid)
-        if not ancestors:
+        if not chain_cids:
             return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
 
-        collected: list[PendingWrite] = []  # newest first; reversed at the end
-        for cid in ancestors:
-            # Collect writes first — they encode the transition FROM this
-            # ancestor's state to its child's and must be included even if
-            # this ancestor is also the seed checkpoint.
+        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))
-            # Then check seed terminator.
-            ver = ver_of.get(cid)
-            if ver is not None:
-                seed_blob = blob_by_ver.get(ver)
-                if seed_blob is not None and seed_blob[0] != "empty":
-                    blob_value = self.serde.loads_typed(seed_blob)
-                    if blob_value is not DELTA_SENTINEL:
-                        collected.reverse()
-                        return _ChannelWritesHistory(seed=blob_value, writes=collected)
 
-        collected.reverse()  # oldest → newest
-        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+        seed: Any = DELTA_SENTINEL
+        if seed_blob is not None and seed_blob[0] != "empty":
+            seed = self.serde.loads_typed(seed_blob)
+
+        collected.reverse()
+        return _ChannelWritesHistory(seed=seed, writes=collected)
 
     def _dump_blobs(
         self,

libs/checkpoint-postgres/pyproject.toml

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

libs/checkpoint-postgres/uv.lock

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

libs/checkpoint-sqlite/uv.lock

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

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

@@ -452,7 +452,9 @@ class InMemorySaver(
         values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
         for k, v in new_versions.items():
             self.blobs[(thread_id, checkpoint_ns, k, v)] = (
-                self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
+                self.serde.dumps_typed(values[k])
+                if k in values and values[k] is not DELTA_SENTINEL
+                else ("empty", b"")
             )
         self.storage[thread_id][checkpoint_ns].update(
             {

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

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

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

@@ -17,12 +17,10 @@ TASKS = "__pregel_tasks"
 
 
 class _DeltaSentinel:
-    """Singleton marker stored (as zero bytes) in checkpoint_blobs for a
-    DeltaChannel field. The actual per-step writes live in checkpoint_writes
-    and are replayed through the reducer at load time.
+    """In-memory marker for a DeltaChannel field with no snapshot.
 
-    Compare with `is DELTA_SENTINEL` — `loads_typed` always returns the same
-    module-level instance.
+    Never serialized to storage — checkpointers strip it before writing.
+    Compare with `is DELTA_SENTINEL`; always the same module-level instance.
     """
 
     __slots__ = ()

libs/checkpoint/pyproject.toml

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

libs/checkpoint/tests/test_jsonplus.py

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

libs/checkpoint/tests/test_memory.py

@@ -322,26 +322,17 @@ def test_memory_saver_with_allowlist_proxy_isolated() -> None:
 
 
 class TestInMemorySaverDeltaChannel:
-    def test_load_blobs_returns_sentinel_for_delta_channel(self) -> None:
-        """_load_blobs returns DELTA_SENTINEL for delta channels (reconstruction deferred)."""
+    def test_load_blobs_omits_delta_channel(self) -> None:
+        """_load_blobs omits delta channels (stored as 'empty'); reconstruction deferred."""
         saver = InMemorySaver()
-        serde = JsonPlusSerializer()
 
         thread_id, ns, channel = "t1", "", "messages"
         v1 = "00000000000000000000000000000001.0000000000000000"
 
-        saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(DELTA_SENTINEL)
-
-        cp1 = empty_checkpoint()
-        cp1["id"] = "cp1"
-        cp1["channel_versions"][channel] = v1
-        saver.storage[thread_id][ns] = {
-            "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
-        }
+        saver.blobs[(thread_id, ns, channel, v1)] = ("empty", b"")
 
         result = saver._load_blobs(thread_id, ns, {channel: v1})
-        assert channel in result
-        assert result[channel] is DELTA_SENTINEL
+        assert channel not in result
 
     def test_get_channel_writes_collects_ancestor_writes_only(self) -> None:
         """_get_channel_writes_history collects ancestor writes oldest→newest,
@@ -582,9 +573,9 @@ class TestPreDeltaBlobTerminator:
 
         # Pre-delta: cp1 stored a real blob for the channel.
         saver.blobs[(thread_id, ns, channel, v1)] = serde.dumps_typed(["A"])
-        # Delta-era: cp2 and cp3 store sentinels; real writes in checkpoint_writes.
-        saver.blobs[(thread_id, ns, channel, v2)] = serde.dumps_typed(DELTA_SENTINEL)
-        saver.blobs[(thread_id, ns, channel, v3)] = serde.dumps_typed(DELTA_SENTINEL)
+        # Delta-era: cp2 and cp3 store "empty"; real writes in checkpoint_writes.
+        saver.blobs[(thread_id, ns, channel, v2)] = ("empty", b"")
+        saver.blobs[(thread_id, ns, channel, v3)] = ("empty", b"")
 
         cp1 = empty_checkpoint()
         cp1["id"] = "cp1"

libs/checkpoint/uv.lock

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

libs/langgraph/langgraph/errors.py

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

libs/langgraph/langgraph/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_events(version="v3")` / `astream_events(version="v3")` is called and are
+                per run whenever `stream_v2` / `astream_v2` 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/_messages.py

@@ -349,7 +349,7 @@ class StreamMessagesHandlerV2(StreamMessagesHandler, _V2StreamingCallbackHandler
         tags: list[str] | None = None,
         **kwargs: Any,
     ) -> Any:
-        """Forward a protocol event from `stream_events(version="v3")` as a messages stream part.
+        """Forward a protocol event from `stream_v2` 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

@@ -40,9 +40,7 @@ from langchain_core.runnables.config import (
     get_async_callback_manager_for_config,
     get_callback_manager_for_config,
 )
-from langchain_core._event_streaming import _AsyncEventsResult
 from langchain_core.runnables.graph import Graph
-from langchain_core.runnables.schema import StreamEvent
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
@@ -378,7 +376,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_events(version="v3")` asks the graph for exactly that union — no hardcoded
+    `stream_v2` 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.
     """
@@ -408,14 +406,14 @@ def _normalize_stream_transformer_factories(
     for spec in specs or ():
         if isinstance(spec, StreamTransformer):
             raise TypeError(
-                "stream_events(version='v3') transformers must be scope-aware callables, "
+                "stream_v2 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_events(version='v3') transformers must be scope-aware callables, "
+                "stream_v2 transformers must be scope-aware callables, "
                 f"got {type(spec).__name__}."
             )
 
@@ -3449,7 +3447,7 @@ class Pregel(
                 await asyncio.shield(run_manager.on_chain_error(e))
             raise
 
-    def _pregel_stream_v3(
+    def stream_v2(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3459,7 +3457,42 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3')."""
+        """Start a sync v2 streaming run driven by transformer projections.
+
+        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.
+
+        `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.
+        """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3491,7 +3524,7 @@ class Pregel(
         )
         return GraphRunStream(graph_iter, mux)
 
-    async def _apregel_stream_v3(
+    async def astream_v2(
         self,
         input: InputT | Command | None,
         config: RunnableConfig | None = None,
@@ -3501,7 +3534,32 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
     ) -> Any:
-        """Internal v3 async streaming implementation. Public entry: astream_events(version='v3')."""
+        """Async counterpart to `stream_v2`.
+
+        Returns an `AsyncGraphRunStream` whose projections can be awaited
+        concurrently; each subscribed cursor drives the pump when its
+        buffer is empty.
+
+        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.
+        """
         parent_ns = _resolve_parent_ns(self.config, config)
         compiled_factories = _normalize_stream_transformer_factories(
             self.stream_transformers
@@ -3531,126 +3589,6 @@ 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(  # type: ignore[override]
-        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` exposing typed projections.
-
-        Args:
-            input: Graph input.
-            config: Optional runnable config.
-            version: Streaming-event schema version. `"v3"` selects the
-                content-block-centric streaming protocol.
-            interrupt_before: Nodes to interrupt before, if any. Only used
-                for `version="v3"`.
-            interrupt_after: Nodes to interrupt after, if any. Only used
-                for `version="v3"`.
-            control: Optional run control. Only used for `version="v3"`.
-            transformers: Extra transformer factories. Only used for
-                `version="v3"`.
-            **kwargs: Forwarded to the v1/v2 path.
-
-        Returns:
-            For `version="v3"`, a `GraphRunStream`. 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,
-    ) -> _AsyncEventsResult: ...
-
-    @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,
-    ) -> _AsyncEventsResult: ...
-
-    def astream_events(  # type: ignore[override]
-        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,
-    ) -> _AsyncEventsResult:
-        """Async variant of `stream_events`. See `stream_events` for full docs."""
-        if version == "v3":
-            return _AsyncEventsResult(
-                awaitable=self._apregel_stream_v3(
-                    input,
-                    config,
-                    interrupt_before=interrupt_before,
-                    interrupt_after=interrupt_after,
-                    control=control,
-                    transformers=transformers,
-                )
-            )
-        iterator = super().astream_events(input, config, version=version, **kwargs)
-        return _AsyncEventsResult(iterator=iterator)
-
     @overload
     def invoke(
         self,
@@ -4145,7 +4083,7 @@ def _resolve_parent_ns(
 ) -> tuple[str, ...]:
     """Return the checkpoint namespace the caller is running under.
 
-    `stream_events(version="v3")` uses this to scope its native projections
+    `stream_v2` 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_events(version="v3")` /
-`graph.astream_events(version="v3")` to drive a transformer pipeline that projects the
+Compile a graph with `transformers=[...]` and call `graph.stream_v2()` /
+`graph.astream_v2()` 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_events(version="v3")` run requests from the graph.
+            which modes a `stream_v2` run requests from the graph.
             Empty tuple means the transformer consumes only synthetic
             events (or is purely passive).
     """

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_events(version="v3")` / `astream_events(version="v3")` populate from the caller's
-    checkpoint namespace so that a nested `stream_events(version="v3")` call still
+    `stream_v2` / `astream_v2` populate from the caller's
+    checkpoint namespace so that a nested `stream_v2` 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_events(version="v3")` / `astream_events(version="v3")` via the `on_stream_event`
+       `stream_v2()` / `astream_v2()` 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_events(version="v3")` / `astream_events(version="v3")`. Models called via the legacy
+    `stream_v2()` / `astream_v2()`. 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_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
+    `stream_v2` / `astream_v2` populate from the caller's checkpoint
+    namespace so that a `stream_v2` 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_events(version="v3") to populate this
+        # v1 streaming callers must switch to stream_v2() to populate this
         # projection.
 
         return True

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_channels.py

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

libs/langgraph/tests/test_interleave_arrival_order.py

@@ -291,13 +291,13 @@ class TestInterleaveArrivalOrder:
 
 
 # ---------------------------------------------------------------------------
-# Integration test: interleave with stream_events(version="v3")
+# Integration test: interleave with stream_v2
 # ---------------------------------------------------------------------------
 
 
 class TestInterleaveIntegration:
     def test_interleave_values_and_messages(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         tagged = list(run.interleave("values", "messages"))
         names = [name for name, _ in tagged]
         assert set(names).issubset({"values", "messages"})
@@ -319,7 +319,7 @@ class TestInterleaveIntegration:
             list(run.interleave("alpha"))
 
     def test_interleave_releases_projections_on_completion(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         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 +327,7 @@ class TestInterleaveIntegration:
         assert run.extensions["messages"]._subscribed is False
 
     def test_interleave_releases_projections_on_early_break(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         gen = run.interleave("values", "messages")
         next(gen)
         gen.close()

libs/langgraph/tests/test_pregel_stream_v2.py

@@ -1,4 +1,4 @@
-"""Tests for Pregel.stream_events(version="v3") / astream_events(version="v3") and the transformer pipeline."""
+"""Tests for Pregel.stream_v2 / astream_v2 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_events(version="v3")` requests only the modes that registered transformers
+    `stream_v2` 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,25 @@ class TestStreamChannelNamed:
 
 
 # ---------------------------------------------------------------------------
-# stream_events(version="v3") sync tests
+# stream_v2 sync tests
 # ---------------------------------------------------------------------------
 
 
 class TestStreamV2Sync:
     def test_values_projection(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         snapshots = list(run.values)
         assert len(snapshots) >= 1
         last = snapshots[-1]
         assert "A" in last["value"] and "B" in last["value"]
 
     def test_output(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         output = run.output
         assert output == {"value": "xAB", "items": ["a", "b"]}
 
     def test_raw_event_iteration(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         events = list(run)
         assert len(events) > 0
         for event in events:
@@ -418,21 +418,23 @@ class TestStreamV2Sync:
             assert isinstance(event["params"]["timestamp"], int)
 
     def test_extensions_has_native_keys(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         _ = run.output
         assert "values" in run.extensions and "messages" in run.extensions
         assert run.values is run.extensions["values"]
         assert run.messages is run.extensions["messages"]
 
     def test_extensions_is_read_only(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(TypeError):
             run.extensions["new_key"] = object()  # type: ignore[index]
         with pytest.raises(TypeError):
             del run.extensions["values"]  # type: ignore[attr-defined]
 
     def test_custom_stream_events(self) -> None:
-        run = _build_custom_stream_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = _build_custom_stream_graph().stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CustomPassthroughTransformer],
         )
         custom_events = [e for e in run if e["method"] == "custom"]
         assert len(custom_events) == 2
@@ -442,18 +444,18 @@ class TestStreamV2Sync:
     def test_custom_events_suppressed_without_transformer(self) -> None:
         """Without a transformer declaring `"custom"`, no custom events flow.
 
-        `stream_events(version="v3")` asks the graph only for the modes that registered
+        `stream_v2` 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_events({"value": "x", "items": []}, version="v3")
+        run = _build_custom_stream_graph().stream_v2({"value": "x", "items": []})
         custom_events = [e for e in run if e["method"] == "custom"]
         assert custom_events == []
 
     def test_interleave_values_and_messages(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         tagged = list(run.interleave("values", "messages"))
         names = [name for name, _ in tagged]
         assert set(names).issubset({"values", "messages"})
@@ -463,7 +465,7 @@ class TestStreamV2Sync:
         assert run.extensions["messages"]._subscribed is False
 
     def test_abort_marks_exhausted_and_closes_mux(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         values_iter = iter(run.values)
         _ = next(values_iter)
         run.abort()
@@ -472,53 +474,56 @@ class TestStreamV2Sync:
         run.abort()  # idempotent
 
     def test_context_manager_calls_abort_on_exit(self) -> None:
-        with _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3") as run:
+        with _build_simple_graph().stream_v2({"value": "x", "items": []}) as run:
             _ = next(iter(run.values))
         assert run._exhausted is True
 
     def test_interleave_unknown_projection(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(KeyError):
             list(run.interleave("values", "does_not_exist"))
 
 
 class TestStreamV2SyncErrors:
     def test_error_propagation_output(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             _ = run.output
 
     def test_error_propagation_values(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             list(run.values)
 
     def test_error_propagation_raw_events(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             list(run)
 
     def test_error_propagation_interrupted(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             _ = run.interrupted
 
     def test_error_propagation_interrupts(self) -> None:
-        run = _build_error_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_error_graph().stream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             _ = run.interrupts
 
 
 class TestStreamV2SyncInterrupt:
     def test_interrupted(self) -> None:
-        run = _build_interrupt_graph().stream_events({"value": "x", "items": []}, {"configurable": {"thread_id": "t1"}}, version="v3")
+        run = _build_interrupt_graph().stream_v2(
+            {"value": "x", "items": []},
+            {"configurable": {"thread_id": "t1"}},
+        )
         _ = run.output
         assert run.interrupted is True
         assert len(run.interrupts) > 0
 
 
 # ---------------------------------------------------------------------------
-# astream_events(version="v3") async tests
+# astream_v2 async tests
 # ---------------------------------------------------------------------------
 
 
@@ -526,26 +531,26 @@ class TestStreamV2SyncInterrupt:
 @NEEDS_CONTEXTVARS
 class TestStreamV2Async:
     async def test_values_projection(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         snapshots = [s async for s in run.values]
         assert len(snapshots) >= 1
         last = snapshots[-1]
         assert "A" in last["value"] and "B" in last["value"]
 
     async def test_output(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         output = await run.output()
         assert output == {"value": "xAB", "items": ["a", "b"]}
 
     async def test_raw_event_iteration(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         events = [e async for e in run]
         assert len(events) > 0
         for event in events:
             assert event["type"] == "event"
 
     async def test_abort_marks_exhausted_and_closes_mux(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         values_iter = aiter(run.values)
         _ = await anext(values_iter)
         await run.abort()
@@ -555,20 +560,22 @@ class TestStreamV2Async:
         await run.abort()  # idempotent
 
     async def test_context_manager_calls_abort_on_exit(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         async with run:
             _ = await anext(aiter(run.values))
         assert run._exhausted is True
 
     async def test_extensions_has_native_keys(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         _ = await run.output()
         assert "values" in run.extensions and "messages" in run.extensions
         assert run.values is run.extensions["values"]
         assert run.messages is run.extensions["messages"]
 
     async def test_custom_stream_events(self) -> None:
-        run = await _build_custom_stream_graph().astream_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = await _build_custom_stream_graph().astream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CustomPassthroughTransformer],
         )
         events = [e async for e in run]
         custom_events = [e for e in events if e["method"] == "custom"]
@@ -581,29 +588,29 @@ class TestStreamV2Async:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncErrors:
     async def test_error_propagation_output(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             await run.output()
 
     async def test_error_propagation_values(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             async for _ in run.values:
                 pass
 
     async def test_error_propagation_raw_events(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             async for _ in run:
                 pass
 
     async def test_error_propagation_interrupted(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             await run.interrupted()
 
     async def test_error_propagation_interrupts(self) -> None:
-        run = await _build_error_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_error_graph().astream_v2({"value": "x", "items": []})
         with pytest.raises(ValueError, match="boom"):
             await run.interrupts()
 
@@ -612,7 +619,10 @@ class TestStreamV2AsyncErrors:
 @NEEDS_CONTEXTVARS
 class TestStreamV2AsyncInterrupt:
     async def test_interrupted(self) -> None:
-        run = await _build_interrupt_graph().astream_events({"value": "x", "items": []}, {"configurable": {"thread_id": "t2"}}, version="v3")
+        run = await _build_interrupt_graph().astream_v2(
+            {"value": "x", "items": []},
+            {"configurable": {"thread_id": "t2"}},
+        )
         _ = await run.output()
         assert await run.interrupted() is True
         assert len(await run.interrupts()) > 0
@@ -975,7 +985,8 @@ class TestCustomTransformer:
                     self._channel.push(self._count)
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[CounterTransformer]
+        run = _build_simple_graph().stream_v2(
+            {"value": "x", "items": []}, transformers=[CounterTransformer]
         )
         assert "counter" in run.extensions
         counter_iter = iter(run.extensions["counter"])
@@ -1000,14 +1011,15 @@ class TestCustomTransformer:
                     self._log.push("saw_values")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[FooTransformer]
+        run = _build_simple_graph().stream_v2(
+            {"value": "x", "items": []}, 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_events_v3_rejects_transformer_instances(self) -> None:
+    def test_stream_v2_rejects_transformer_instances(self) -> None:
         class InstanceTransformer(StreamTransformer):
             def init(self) -> dict[str, Any]:
                 return {}
@@ -1016,7 +1028,8 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(TypeError, match="pre-built instance"):
-            _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[InstanceTransformer()]
+            _build_simple_graph().stream_v2(
+                {"value": "x", "items": []}, transformers=[InstanceTransformer()]
             )
 
     def test_stream_channel_auto_forward(self) -> None:
@@ -1035,7 +1048,8 @@ class TestCustomTransformer:
                     self._channel.push("emitted")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[EmitterTransformer]
+        run = _build_simple_graph().stream_v2(
+            {"value": "x", "items": []}, transformers=[EmitterTransformer]
         )
         custom_events = [e for e in run if e["method"] == "custom:emitter"]
         assert len(custom_events) > 0
@@ -1079,7 +1093,8 @@ class TestCustomTransformer:
                 return True
 
         with pytest.raises(ValueError, match=r"conflict.*'values'.*ValuesTransformer"):
-            _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[ConflictTransformer]
+            _build_simple_graph().stream_v2(
+                {"value": "x", "items": []}, transformers=[ConflictTransformer]
             )
 
 
@@ -1161,7 +1176,8 @@ class TestStreamChannelAutoLifecycle:
                     self._log.push("got_it")
                 return True
 
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3", transformers=[MinimalTransformer]
+        run = _build_simple_graph().stream_v2(
+            {"value": "x", "items": []}, transformers=[MinimalTransformer]
         )
         minimal_iter = iter(run.extensions["minimal"])
         _ = run.output
@@ -1421,7 +1437,8 @@ class TestAsyncTransformerLane:
             async def afinalize(self) -> None:
                 self._log.close()
 
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3", transformers=[Scorer]
+        run = await _build_simple_graph().astream_v2(
+            {"value": "x", "items": []}, transformers=[Scorer]
         )
         scores_cursor = aiter(run.extensions["scores"])
         _ = await run.output()
@@ -1437,7 +1454,7 @@ class TestAsyncTransformerLane:
 @NEEDS_CONTEXTVARS
 class TestMemoryBounds:
     def test_sync_subscribed_buffer_stays_at_most_one_between_yields(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         events_iter = iter(run)
         max_buffered = 0
         count = 0
@@ -1450,7 +1467,7 @@ class TestMemoryBounds:
         )
 
     def test_unsubscribed_projections_never_accumulate(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         list(run)
         values_log = run.extensions["values"]
         messages_log = run.extensions["messages"]
@@ -1458,19 +1475,19 @@ class TestMemoryBounds:
         assert len(messages_log._items) == 0 and not messages_log._subscribed
 
     def test_output_path_does_not_retain_values(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         _ = run.output
         values_log = run.extensions["values"]
         assert len(values_log._items) == 0 and not values_log._subscribed
 
     def test_drained_subscriber_buffer_returns_to_empty(self) -> None:
-        run = _build_simple_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run = _build_simple_graph().stream_v2({"value": "x", "items": []})
         list(run.values)
         assert len(run.extensions["values"]._items) == 0
 
     @pytest.mark.anyio
     async def test_async_single_consumer_buffer_stays_at_most_one(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         max_buffered = 0
         count = 0
         async for _ in run:
@@ -1481,7 +1498,7 @@ class TestMemoryBounds:
 
     @pytest.mark.anyio
     async def test_async_unsubscribed_projections_never_accumulate(self) -> None:
-        run = await _build_simple_graph().astream_events({"value": "x", "items": []}, version="v3")
+        run = await _build_simple_graph().astream_v2({"value": "x", "items": []})
         _ = await run.output()
         values_log = run.extensions["values"]
         messages_log = run.extensions["messages"]

libs/langgraph/tests/test_retry.py

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

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_events(version="v3").
+group exercises real graphs through stream_v2.
 """
 
 from __future__ import annotations
@@ -478,7 +478,7 @@ def test_unrelated_events_ignored_by_all() -> None:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: real graphs through stream_events(version="v3")
+# End-to-end: real graphs through stream_v2
 # ---------------------------------------------------------------------------
 
 
@@ -503,10 +503,11 @@ def _make_simple_graph() -> Any:
     return builder.compile()
 
 
-def test_stream_events_v3_custom_projection_opt_in() -> None:
+def test_stream_v2_custom_projection_opt_in() -> None:
     """run.custom surfaces get_stream_writer() payloads when opted in."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
+    run = graph.stream_v2(
+        {"value": "hello", "items": []}, transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -514,10 +515,11 @@ def test_stream_events_v3_custom_projection_opt_in() -> None:
     assert any(e.get("status") == "working" for e in custom_events)
 
 
-def test_stream_events_v3_custom_and_values_coexist() -> None:
+def test_stream_v2_custom_and_values_coexist() -> None:
     """Both run.custom and run.values work in the same run."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "hello", "items": []}, version="v3", transformers=[CustomTransformer]
+    run = graph.stream_v2(
+        {"value": "hello", "items": []}, transformers=[CustomTransformer]
     )
 
     custom_events = list(run.custom)
@@ -526,10 +528,10 @@ def test_stream_events_v3_custom_and_values_coexist() -> None:
     assert len(custom_events) >= 1
 
 
-def test_stream_events_v3_tasks_projection_opt_in() -> None:
+def test_stream_v2_tasks_projection_opt_in() -> None:
     """run.tasks surfaces raw task events when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, transformers=[TasksTransformer], version="v3")
+    run = graph.stream_v2({"value": "x", "items": []}, transformers=[TasksTransformer])
 
     tasks_events = list(run.tasks)
     assert len(tasks_events) >= 1
@@ -537,10 +539,10 @@ def test_stream_events_v3_tasks_projection_opt_in() -> None:
     assert "my_node" in names
 
 
-def test_stream_events_v3_debug_projection_opt_in() -> None:
+def test_stream_v2_debug_projection_opt_in() -> None:
     """run.debug surfaces debug events when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, transformers=[DebugTransformer], version="v3")
+    run = graph.stream_v2({"value": "x", "items": []}, transformers=[DebugTransformer])
 
     debug_events = list(run.debug)
     assert len(debug_events) >= 1
@@ -548,10 +550,11 @@ def test_stream_events_v3_debug_projection_opt_in() -> None:
     assert types & {"checkpoint", "task", "task_result"}
 
 
-def test_stream_events_v3_updates_projection_opt_in() -> None:
+def test_stream_v2_updates_projection_opt_in() -> None:
     """run.updates surfaces node output dicts when opted in via transformers=."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[UpdatesTransformer]
+    run = graph.stream_v2(
+        {"value": "x", "items": []}, transformers=[UpdatesTransformer]
     )
 
     updates = list(run.updates)
@@ -560,10 +563,12 @@ def test_stream_events_v3_updates_projection_opt_in() -> None:
     assert "my_node" in node_names
 
 
-def test_stream_events_v3_all_transformers_interleaved() -> None:
+def test_stream_v2_all_transformers_interleaved() -> None:
     """All five transformers registered together, consumed via interleave."""
     graph = _make_simple_graph()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[
+    run = graph.stream_v2(
+        {"value": "x", "items": []},
+        transformers=[
             CustomTransformer,
             UpdatesTransformer,
             CheckpointsTransformer,
@@ -594,7 +599,7 @@ def test_stream_events_v3_all_transformers_interleaved() -> None:
     assert run.output["value"] == "x!"
 
 
-def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
+def test_stream_v2_all_transformers_with_checkpointer() -> None:
     """All transformers with a checkpointer — run.checkpoints populated."""
     from langgraph.checkpoint.memory import InMemorySaver
 
@@ -604,7 +609,9 @@ def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", config={"configurable": {"thread_id": "test-all"}},
+    run = graph.stream_v2(
+        {"value": "x", "items": []},
+        config={"configurable": {"thread_id": "test-all"}},
         transformers=[
             CustomTransformer,
             UpdatesTransformer,
@@ -630,7 +637,7 @@ def test_stream_events_v3_all_transformers_with_checkpointer() -> None:
     assert len(collected["custom"]) >= 1
 
 
-def test_stream_events_v3_checkpoints_projection_opt_in() -> None:
+def test_stream_v2_checkpoints_projection_opt_in() -> None:
     """run.checkpoints surfaces checkpoint data when opted in with a checkpointer."""
     from langgraph.checkpoint.memory import InMemorySaver
 
@@ -640,7 +647,9 @@ def test_stream_events_v3_checkpoints_projection_opt_in() -> None:
     builder.add_edge("my_node", END)
     graph = builder.compile(checkpointer=InMemorySaver())
 
-    run = graph.stream_events({"value": "x", "items": []}, version="v3", config={"configurable": {"thread_id": "test-ckpt-standalone"}},
+    run = graph.stream_v2(
+        {"value": "x", "items": []},
+        config={"configurable": {"thread_id": "test-ckpt-standalone"}},
         transformers=[CheckpointsTransformer],
     )
 
@@ -679,7 +688,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_events({"value": "x", "items": []}, version="v3", transformers=[TasksTransformer],
+    run = graph.stream_v2(
+        {"value": "x", "items": []},
+        transformers=[TasksTransformer],
     )
 
     tasks_events = list(run.tasks)

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_events(version="v3")` calls with non-empty `parent_ns`).
+discovery, nested `stream_v2` 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_events(version="v3")
+# End-to-end: real graphs through stream_v2
 # ---------------------------------------------------------------------------
 
 
@@ -358,10 +358,10 @@ def _make_two_level_nested() -> Any:
     return outer_b.compile()
 
 
-def test_stream_events_v3_real_graph_emits_lifecycle_at_each_depth() -> None:
+def test_stream_v2_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_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     # Iterating the projection drives the pump and drains synthesized
     # lifecycle events at the same time.
@@ -384,17 +384,17 @@ def test_stream_events_v3_real_graph_emits_lifecycle_at_each_depth() -> None:
         )
 
 
-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,
+def test_stream_v2_with_nested_parent_ns_scopes_lifecycle() -> None:
+    """When `stream_v2` 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_events(version="v3")
+    exercises the path that exists today purely for nested-stream_v2
     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_events({"value": "x", "items": []}, config=config, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []}, config=config)
 
     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_events(version="v3") / astream_events(version="v3")."""
+legacy v1 chunk filtering, and end-to-end via stream_v2 / astream_v2."""
 
 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_events(state["messages"], version="v3")
+        stream = model.stream_v2(state["messages"])
         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_events(version="v3").
+        # into this projection; callers must migrate to stream_v2.
         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_events(version="v3") → run.messages (node calls stream_events)
+# End-to-end: graph → stream_v2 → run.messages (node calls stream_v2)
 # ---------------------------------------------------------------------------
 
 
 class TestEndToEnd:
-    """stream_events(version="v3") path: node calls model.stream_events() explicitly."""
+    """stream_v2 path: node calls model.stream_v2() 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_events(state["messages"], version="v3")
+            stream = model.stream_v2(state["messages"])
             return {"messages": stream.output}
 
         graph = (
@@ -499,7 +499,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "hi"}, version="v3")
+        run = graph.stream_v2({"messages": "hi"})
         (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_events(state["messages"], version="v3")
+            stream = model.stream_v2(state["messages"])
             return {"messages": stream.output}
 
         graph = (
@@ -520,7 +520,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "go"}, version="v3")
+        run = graph.stream_v2({"messages": "go"})
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "streamed answer"
 
@@ -538,7 +538,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "hi"}, version="v3")
+        run = graph.stream_v2({"messages": "hi"})
         (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_events(state["messages"], version="v3")
+            stream = await model.astream_v2(state["messages"])
             return {"messages": await stream}
 
         graph = (
@@ -558,7 +558,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = await graph.astream_events({"messages": "hi"}, version="v3")
+        run = await graph.astream_v2({"messages": "hi"})
         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_events(state["messages"], version="v3")
+            stream = await model.astream_v2(state["messages"])
             return {"messages": await stream}
 
         graph = (
@@ -583,7 +583,7 @@ class TestEndToEnd:
             .compile()
         )
 
-        run = await graph.astream_events({"messages": "hi"}, version="v3")
+        run = await graph.astream_v2({"messages": "hi"})
 
         async def consume() -> list[str]:
             collected: list[str] = []
@@ -596,12 +596,12 @@ class TestEndToEnd:
 
 
 # ---------------------------------------------------------------------------
-# End-to-end: graph → stream_events(version="v3") → run.messages (node calls invoke)
+# End-to-end: graph → stream_v2 → run.messages (node calls invoke)
 # ---------------------------------------------------------------------------
 
 
 class TestEndToEndV2Invoke:
-    """Auto-routing path: stream_events(version="v3") injects CONFIG_KEY_STREAM_MESSAGES_V2,
+    """Auto-routing path: stream_v2 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_events({"messages": "hi"}, version="v3")
+        ).stream_v2({"messages": "hi"})
         (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_events({"messages": "go"}, version="v3")
+        ).stream_v2({"messages": "go"})
         (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_events({"messages": "hi"}, version="v3")
+        ).stream_v2({"messages": "hi"})
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "delta streaming works"
 
@@ -674,7 +674,7 @@ class TestEndToEndV2Invoke:
             .compile()
         )
 
-        streams = list(graph.stream_events({"messages": "hi"}, version="v3").messages)
+        streams = list(graph.stream_v2({"messages": "hi"}).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_events({"messages": "hi"}, version="v3")
+        run = graph.stream_v2({"messages": "hi"})
         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_events({"messages": "hi"}, version="v3")
+        run = await graph.astream_v2({"messages": "hi"})
         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_events(version="v3") / astream_events(version="v3")."""
+        the v2 flag is only injected by stream_v2 / astream_v2."""
         model = GenericFakeChatModel(messages=iter(["legacy path"]))
 
         def call_model(state: MessagesState) -> dict[str, Any]:
@@ -760,8 +760,8 @@ class TestDirectMessagesModeStaysV1:
             == "legacy path"
         )
 
-    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
+    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
         `stream_mode="messages"` call onto the v2 event protocol."""
         model = GenericFakeChatModel(messages=iter(["nested legacy path"]))
 
@@ -812,7 +812,7 @@ class TestDirectMessagesModeStaysV1:
             .compile()
         )
 
-        result = outer.stream_events({}, version="v3").output
+        result = outer.stream_v2({}).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_events(version="v3")`.
+`stream_v2`.
 """
 
 from __future__ import annotations
@@ -93,7 +93,7 @@ def _tasks_result(
 
 
 def _native_factories() -> list[Any]:
-    """Mirror the factory list `Pregel.stream_events(version="v3")` registers."""
+    """Mirror the factory list `Pregel.stream_v2` registers."""
     return [
         ValuesTransformer,
         MessagesTransformer,
@@ -761,10 +761,10 @@ def _make_failing_nested() -> Any:
     return outer_b.compile()
 
 
-def test_stream_events_v3_real_graph_yields_subgraph_handles() -> None:
+def test_stream_v2_real_graph_yields_subgraph_handles() -> None:
     """Iterating `run.subgraphs` yields handles for direct-child subgraphs."""
     graph = _make_two_level_nested()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     handle_paths: list[tuple[str, ...]] = []
     final_status: dict[tuple[str, ...], str] = {}
@@ -780,10 +780,10 @@ def test_stream_events_v3_real_graph_yields_subgraph_handles() -> None:
     assert final_status[handle_paths[0]] == "completed"
 
 
-def test_stream_events_v3_grandchild_visible_on_child_handle() -> None:
+def test_stream_v2_grandchild_visible_on_child_handle() -> None:
     """Drilling into `handle.subgraphs` surfaces nested grandchildren."""
     graph = _make_two_level_nested()
-    run = graph.stream_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     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_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     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_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     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_events({"value": "x", "items": []}, version="v3")
+    run = graph.stream_v2({"value": "x", "items": []})
 
     handle = next(iter(run.subgraphs))
     with pytest.raises(RuntimeError, match="child boom"):

libs/langgraph/tests/test_stream_v2.py

@@ -231,7 +231,7 @@ class TestV2Stream:
         for c in chunks:
             _assert_stream_part_shape(c)
 
-    def test_stream_events_v3_accepts_control_for_drain(self) -> None:
+    def test_stream_v2_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_events({}, control=control, version="v3")
+        run = graph.stream_v2({}, control=control)
         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_events_v3_pydantic_validation_error(self) -> None:
+    def test_stream_v2_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_v2_e2e.py

@@ -1,9 +1,9 @@
-"""End-to-end tests exercising all stream_events(version="v3") projections together.
+"""End-to-end tests exercising all stream_v2 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_events(version="v3") / astream_events(version="v3") run.
+results through a single stream_v2 / astream_v2 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_events(version="v3") and verify values + lifecycle."""
+        """Run a nested graph through stream_v2 and verify values + lifecycle."""
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
         events = list(run)
         assert len(events) > 0
 
@@ -285,11 +285,11 @@ class TestStreamV2E2ESync:
 
     def test_output_matches_final_values_snapshot(self) -> None:
         """output property returns the same state as the last values snapshot."""
-        run1 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run1 = _make_nested_graph().stream_v2({"value": "x", "items": []})
         snapshots = list(run1.values)
         final_via_values = snapshots[-1]
 
-        run2 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run2 = _make_nested_graph().stream_v2({"value": "x", "items": []})
         final_via_output = run2.output
 
         assert final_via_values == final_via_output
@@ -297,7 +297,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_events({"value": "x", "items": []}, version="v3") as run:
+        with graph.stream_v2({"value": "x", "items": []}) as run:
             first_val = next(iter(run.values))
             assert isinstance(first_val, dict)
         assert run._exhausted is True
@@ -305,7 +305,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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
         _ = run.output
 
         assert "values" in run.extensions
@@ -327,7 +327,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_events({"messages": "hi"}, version="v3")
+        run = graph.stream_v2({"messages": "hi"})
         streams = list(run.messages)
 
         assert len(streams) >= 1
@@ -350,7 +350,7 @@ class TestStreamV2E2EMessages:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "go"}, version="v3")
+        run = graph.stream_v2({"messages": "go"})
         (stream,) = list(run.messages)
         assert "".join(stream.text) == "streamed answer"
 
@@ -368,7 +368,7 @@ class TestStreamV2E2EMessages:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "hi"}, version="v3")
+        run = graph.stream_v2({"messages": "hi"})
         (stream,) = list(run.messages)
         assert stream.output.text == "hardcoded"
         assert stream.message_id == "msg-1"
@@ -376,7 +376,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_events({"messages": ["hi"], "done": False}, version="v3")
+        run = graph.stream_v2({"messages": ["hi"], "done": False})
         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 +385,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_events({"messages": ["hi"], "done": False}, version="v3")
+        run = graph.stream_v2({"messages": ["hi"], "done": False})
 
         found_messages = False
         for handle in run.subgraphs:
@@ -407,7 +407,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_events({"value": "x", "items": []}, version="v3", transformers=[_CustomPassthroughTransformer],
+        run = graph.stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CustomPassthroughTransformer],
         )
         events = list(run)
         custom = [e for e in events if e["method"] == "custom"]
@@ -418,7 +420,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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
         events = list(run)
         custom = [e for e in events if e["method"] == "custom"]
         assert custom == []
@@ -426,7 +428,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_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CounterTransformer],
         )
 
         assert "counter" in run.extensions
@@ -440,7 +444,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_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CounterTransformer],
         )
         events = list(run)
         counter_events = [e for e in events if e["method"] == "custom:counter"]
@@ -458,7 +464,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_events({"value": "x", "items": []}, config, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []}, config)
 
         output = run.output
         assert output is not None
@@ -471,7 +477,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_events({"value": "x", "items": []}, config, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []}, config)
 
         snapshots = list(run.values)
         assert len(snapshots) >= 1
@@ -488,14 +494,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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         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_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         with pytest.raises(ValueError, match="subgraph explosion"):
             list(run)
@@ -503,7 +509,7 @@ class TestStreamV2E2EErrors:
     def test_error_subgraph_handle_status(self) -> None:
         """Subgraph handle surfaces the error status."""
         graph = _make_error_subgraph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         handle = next(iter(run.subgraphs))
         with pytest.raises(RuntimeError, match="subgraph explosion"):
@@ -523,7 +529,7 @@ class TestStreamV2E2EAsync:
     async def test_all_projections_async(self) -> None:
         """Async run exercises values projection."""
         graph = _make_nested_graph()
-        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
 
         values_snapshots = [s async for s in run.values]
         assert len(values_snapshots) >= 1
@@ -534,7 +540,7 @@ class TestStreamV2E2EAsync:
     async def test_async_output(self) -> None:
         """Async output returns the final state."""
         graph = _make_nested_graph()
-        run = await graph.astream_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
         output = await run.output()
         assert output is not None
         assert output["value"] == "x_routed_processed"
@@ -544,7 +550,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_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
         events = [e async for e in run]
         assert len(events) > 0
         seqs = [e["seq"] for e in events]
@@ -566,7 +572,7 @@ class TestStreamV2E2EAsync:
             .compile()
         )
 
-        run = await graph.astream_events({"messages": "hi"}, version="v3")
+        run = await graph.astream_v2({"messages": "hi"})
         streams = [s async for s in run.messages]
         assert len(streams) >= 1
         for s in streams:
@@ -577,7 +583,7 @@ class TestStreamV2E2EAsync:
         """Async interrupted run has correct flags."""
         graph = _make_interrupt_graph()
         config: dict[str, Any] = {"configurable": {"thread_id": "async-int-1"}}
-        run = await graph.astream_events({"value": "x", "items": []}, config, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []}, config)
 
         output = await run.output()
         assert output is not None
@@ -587,14 +593,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_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
         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_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
         async with run:
             _ = await anext(aiter(run.values))
         assert run._exhausted is True
@@ -602,7 +608,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_events({"value": "x", "items": []}, version="v3")
+        run = await graph.astream_v2({"value": "x", "items": []})
         _ = await run.output()
         assert "values" in run.extensions
         assert "messages" in run.extensions
@@ -612,7 +618,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_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer],
+        run = await graph.astream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CounterTransformer],
         )
         assert "counter" in run.extensions
         counter_cursor = aiter(run.extensions["counter"])
@@ -631,7 +639,7 @@ class TestStreamV2E2ECombined:
     def test_interleave_all_native_projections(self) -> None:
         """Interleave values + messages + lifecycle without deadlock."""
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3")
+        run = graph.stream_v2({"value": "x", "items": []})
 
         seen_names: set[str] = set()
         for name, _item in run.interleave("values", "messages", "lifecycle"):
@@ -659,7 +667,9 @@ class TestStreamV2E2ECombined:
                 return True
 
         graph = _make_nested_graph()
-        run = graph.stream_events({"value": "x", "items": []}, version="v3", transformers=[_CounterTransformer, TagTransformer],
+        run = graph.stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CounterTransformer, TagTransformer],
         )
 
         assert "counter" in run.extensions
@@ -712,7 +722,7 @@ class TestStreamV2E2ECombined:
             .compile()
         )
 
-        run = outer.stream_events({"items": []}, version="v3")
+        run = outer.stream_v2({"items": []})
         handles = []
         for handle in run.subgraphs:
             list(handle.values)
@@ -730,13 +740,13 @@ class TestStreamV2E2ECombined:
 
     def test_lifecycle_matches_subgraph_handles(self) -> None:
         """Lifecycle events and subgraph handles agree on discovered subgraphs."""
-        run1 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run1 = _make_nested_graph().stream_v2({"value": "x", "items": []})
         handle_paths: list[tuple[str, ...]] = []
         for handle in run1.subgraphs:
             list(handle.values)
             handle_paths.append(handle.path)
 
-        run2 = _make_nested_graph().stream_events({"value": "x", "items": []}, version="v3")
+        run2 = _make_nested_graph().stream_v2({"value": "x", "items": []})
         lifecycle = list(run2.lifecycle)
 
         started_ns = [
@@ -763,7 +773,9 @@ class TestStreamV2E2ECombined:
             .compile()
         )
 
-        run = graph.stream_events({"messages": "hi"}, version="v3", transformers=[_CounterTransformer],
+        run = graph.stream_v2(
+            {"messages": "hi"},
+            transformers=[_CounterTransformer],
         )
 
         counter_iter = iter(run.extensions["counter"])

libs/langgraph/uv.lock

@@ -1349,7 +1349,7 @@ wheels = [
 [[package]]
 name = "langchain-core"
 version = "1.3.2"
-source = { git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release#ad4d43f38827e2aadce559e7e676b5eb4184a976" }
+source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
     { name = "langchain-protocol" },
@@ -1361,22 +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" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
+]
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.14"
+version = "0.0.12"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/05/bf/efb5e2ed832e4d6d45590e25a9e5191986b291b543bc6a807b48bee070b0/langchain_protocol-0.0.14.tar.gz", hash = "sha256:bc1e8553122e6ede310280462d5813023a172ff2785ccbbdec54d43f3a15e5f2", size = 5862, upload-time = "2026-04-29T16:40:18.657Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c2/e9/06c47ecb2aff08f83dfa30058da3bf86be64862c19569043ed5331bbeecd/langchain_protocol-0.0.14-py3-none-any.whl", hash = "sha256:ffc35089779bd8ca217015180cef5e660fc3b074efdaa0f2e95df73583f1a047", size = 6984, upload-time = "2026-04-29T16:40:17.841Z" },
+    { url = "https://files.pythonhosted.org/packages/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" },
 ]
 
 [[package]]
 name = "langgraph"
-version = "1.2.0a1"
+version = "1.2.0a3"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1448,7 +1452,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.3.2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },
@@ -1460,7 +1464,7 @@ requires-dist = [
 dev = [
     { name = "httpx" },
     { name = "jupyter" },
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.0.0" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1493,7 +1497,7 @@ lint = [
 ]
 test = [
     { name = "httpx" },
-    { name = "langchain-core", git = "https://github.com/langchain-ai/langchain.git?subdirectory=libs%2Fcore&branch=nh%2Fstreaming-for-alpha-release" },
+    { name = "langchain-core", specifier = ">=1.0.0" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
     { name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
@@ -1557,7 +1561,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "4.1.0a1"
+version = "4.1.0a3"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1605,7 +1609,7 @@ test = [
 
 [[package]]
 name = "langgraph-checkpoint-postgres"
-version = "3.1.0a1"
+version = "3.1.0a3"
 source = { editable = "../checkpoint-postgres" }
 dependencies = [
     { name = "langgraph-checkpoint" },

libs/prebuilt/tests/test_tool_call_transformer.py

@@ -199,7 +199,7 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [streamer])
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_v2({"messages": []}, transformers=[ToolCallTransformer])
 
         tool_calls: list[ToolCallStream] = []
         for tc in run.tool_calls:
@@ -235,11 +235,11 @@ 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_events({"messages": []}, version="v3")
+        run_no_tc = graph.stream_v2({"messages": []})
         assert "tool_calls" not in run_no_tc._mux.extensions  # type: ignore[attr-defined]
 
         # With ToolCallTransformer, the projection is present.
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_v2({"messages": []}, transformers=[ToolCallTransformer])
         assert "tool_calls" in run._mux.extensions  # type: ignore[attr-defined]
         # Drain so the run closes cleanly.
         list(run.tool_calls)
@@ -266,7 +266,8 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [astreamer])
-        run = await graph.astream_events({"messages": []}, version="v3", transformers=[ToolCallTransformer]
+        run = await graph.astream_v2(
+            {"messages": []}, transformers=[ToolCallTransformer]
         )
 
         collected: list[ToolCallStream] = []
@@ -295,7 +296,7 @@ class TestToolCallTransformerEndToEnd:
             }
 
         graph = _build_graph(caller, [boom])
-        run = graph.stream_events({"messages": []}, transformers=[ToolCallTransformer], version="v3")
+        run = graph.stream_v2({"messages": []}, transformers=[ToolCallTransformer])
 
         collected: list[ToolCallStream] = []
         with pytest.raises(ValueError, match="nope"):

libs/prebuilt/uv.lock

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

libs/sdk-py/uv.lock

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