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

libs/langgraph/langgraph/graph/state.py

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

libs/langgraph/langgraph/pregel/_messages.py

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

libs/langgraph/langgraph/pregel/main.py

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

libs/langgraph/langgraph/stream/__init__.py

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

libs/langgraph/langgraph/stream/_types.py

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

libs/langgraph/langgraph/stream/run_stream.py

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

libs/langgraph/langgraph/stream/transformers.py

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

libs/langgraph/pyproject.toml

@@ -24,7 +24,7 @@ classifiers = [
     'Programming Language :: Python :: 3.13',
 ]
 dependencies = [
-    "langchain-core>=1.3.2,<2",
+    "langchain-core>=1.4.0a2,<2",
     "langgraph-checkpoint>=4.1.0a3,<5.0.0",
     "langgraph-sdk>=0.3.0,<0.4.0",
     "langgraph-prebuilt>=1.0.12,<1.1.0",

libs/langgraph/tests/test_interleave_arrival_order.py

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

libs/langgraph/tests/test_pregel_stream_events_v3.py

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

libs/langgraph/tests/test_retry.py

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

libs/langgraph/tests/test_stream_data_transformers.py

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

libs/langgraph/tests/test_stream_events_v3.py

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

libs/langgraph/tests/test_stream_events_v3_e2e.py

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

libs/langgraph/tests/test_stream_lifecycle_transformer.py

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

libs/langgraph/tests/test_stream_messages_transformer.py

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

libs/langgraph/tests/test_stream_subgraph_transformer.py

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

libs/langgraph/uv.lock

@@ -1348,7 +1348,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -1361,21 +1361,21 @@ 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]]
@@ -1452,7 +1452,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = ">=1.3.2,<2" },
+    { name = "langchain-core", specifier = ">=1.4.0a2,<2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },

libs/prebuilt/tests/test_tool_call_transformer.py

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

libs/prebuilt/uv.lock

@@ -249,7 +249,7 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.4.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
@@ -262,21 +262,21 @@ 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]]
@@ -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" },

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,21 +279,21 @@ 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]]
@@ -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 = "." },