feat(langgraph): dispatch stream_events(version='v3') on Pregel (#7677)

## 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/pregel/main.py

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

libs/langgraph/langgraph/stream/run_stream.py

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

libs/langgraph/pyproject.toml

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

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

libs/langgraph/tests/test_pregel_stream_events_v3.py

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

libs/langgraph/tests/test_retry.py

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

libs/langgraph/tests/test_stream_data_transformers.py

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

libs/langgraph/tests/test_stream_events_v3_e2e.py

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

libs/langgraph/tests/test_stream_messages_transformer.py

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

libs/langgraph/uv.lock

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

libs/prebuilt/tests/test_tool_call_transformer.py

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

libs/prebuilt/uv.lock

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

libs/sdk-py/uv.lock

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