feat(langgraph): type v3 stream_events return and native projections (#8389)

The `version="v3"` overloads of `stream_events`/`astream_events`
returned `Any`, and `GraphRunStream`/`AsyncGraphRunStream` attached
native projections via a runtime `setattr` loop invisible to type
checkers.

- Return `GraphRunStream` / `Awaitable[AsyncGraphRunStream]` from the v3
overloads.
- Declare the always-registered native projections (`values`,
`messages`, `lifecycle`, `subgraphs`) as typed class attributes on both
run streams.
- Add `assert_type` checks in `test_stream_events_v3.py`.

Opt-in native projections (`updates`, `custom`, `checkpoints`, `debug`,
`tasks`) are only present when their transformer is registered, so they
remain reached via `extensions[...]` rather than annotated as
always-present.

libs/langgraph/langgraph/pregel/main.py

@@ -3512,7 +3512,7 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> Any:
+    ) -> GraphRunStream:
         """Internal v3 sync streaming implementation. Public entry: stream_events(version='v3').
 
         Extra keyword arguments are forwarded to the underlying ``stream(...)``
@@ -3568,7 +3568,7 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> Any:
+    ) -> AsyncGraphRunStream:
         """Internal v3 async streaming implementation. Public entry: astream_events(version='v3').
 
         Extra keyword arguments are forwarded to the underlying ``astream(...)``
@@ -3633,7 +3633,7 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> Any: ...
+    ) -> GraphRunStream: ...
 
     def stream_events(
         self,
@@ -3738,7 +3738,7 @@ class Pregel(
         control: RunControl | None = None,
         transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
         **kwargs: Any,
-    ) -> Awaitable[Any]: ...
+    ) -> Awaitable[AsyncGraphRunStream]: ...
 
     def astream_events(
         self,

libs/langgraph/langgraph/stream/run_stream.py

@@ -12,7 +12,13 @@ from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import ProtocolEvent
 
 if TYPE_CHECKING:
-    from langgraph.stream.transformers import SubgraphStatus
+    from langchain_core.language_models.chat_model_stream import (
+        AsyncChatModelStream,
+        ChatModelStream,
+    )
+
+    from langgraph.stream.stream_channel import StreamChannel
+    from langgraph.stream.transformers import LifecyclePayload, SubgraphStatus
 
 
 def _drive_until_done(pump: Callable[[], bool]) -> None:
@@ -48,6 +54,16 @@ class GraphRunStream:
         experimental and may change.
     """
 
+    # Native projections always registered by `stream_events(version="v3")`.
+    # Attached dynamically by the `setattr` loop in `__init__`; declared here
+    # so type checkers see them. Opt-in native projections (`updates`,
+    # `custom`, `checkpoints`, `debug`, `tasks`) are only present when their
+    # transformer is registered, so they are reached via `extensions[...]`.
+    values: StreamChannel[dict[str, Any]]
+    messages: StreamChannel[ChatModelStream]
+    lifecycle: StreamChannel[LifecyclePayload]
+    subgraphs: StreamChannel[SubgraphRunStream]
+
     def __init__(
         self,
         graph_iter: Iterator[Any] | None,
@@ -329,6 +345,16 @@ class AsyncGraphRunStream:
         experimental and may change.
     """
 
+    # Native projections always registered by `astream_events(version="v3")`.
+    # Attached dynamically by the `setattr` loop in `__init__`; declared here
+    # so type checkers see them. Opt-in native projections (`updates`,
+    # `custom`, `checkpoints`, `debug`, `tasks`) are only present when their
+    # transformer is registered, so they are reached via `extensions[...]`.
+    values: StreamChannel[dict[str, Any]]
+    messages: StreamChannel[AsyncChatModelStream]
+    lifecycle: StreamChannel[LifecyclePayload]
+    subgraphs: StreamChannel[AsyncSubgraphRunStream]
+
     def __init__(
         self,
         graph_aiter: AsyncIterator[Any] | None,

libs/langgraph/tests/test_stream_events_v3.py

@@ -12,6 +12,10 @@ from dataclasses import dataclass
 from typing import Annotated, Any, TypeVar
 
 import pytest
+from langchain_core.language_models.chat_model_stream import (
+    AsyncChatModelStream,
+    ChatModelStream,
+)
 from langchain_core.messages import AIMessage, BaseMessage
 from langgraph.checkpoint.memory import InMemorySaver
 from pydantic import BaseModel, ValidationError
@@ -24,6 +28,14 @@ from langgraph.func import entrypoint
 from langgraph.graph import StateGraph
 from langgraph.graph.message import MessagesState
 from langgraph.runtime import RunControl
+from langgraph.stream import (
+    AsyncGraphRunStream,
+    AsyncSubgraphRunStream,
+    GraphRunStream,
+    LifecyclePayload,
+    StreamChannel,
+    SubgraphRunStream,
+)
 from langgraph.types import (
     CheckpointPayload,
     CheckpointStreamPart,
@@ -1178,3 +1190,32 @@ def _check_type_narrowing(part: StreamPart[_StateT, _OutputT]) -> None:
         assert_type(part, DebugStreamPart[_StateT])
         assert_type(part["data"], DebugPayload[_StateT])
     assert_type(part["ns"], tuple[str, ...])
+
+
+# --- v3 stream_events return / projection typing checks ---
+# These functions are never called at runtime; `ty` validates the
+# assert_type calls. They pin the public typing surface of
+# stream_events(version="v3") / astream_events(version="v3"): the handle
+# type and the always-registered native projections.
+
+
+def _check_stream_events_v3_typing() -> None:
+    """Compile-time checks for sync v3 typing — never called at runtime."""
+    graph = _make_simple_graph().compile()
+    run = graph.stream_events(_SIMPLE_INPUT, version="v3")
+    assert_type(run, GraphRunStream)
+    assert_type(run.values, StreamChannel[dict[str, Any]])
+    assert_type(run.messages, StreamChannel[ChatModelStream])
+    assert_type(run.lifecycle, StreamChannel[LifecyclePayload])
+    assert_type(run.subgraphs, StreamChannel[SubgraphRunStream])
+
+
+async def _check_astream_events_v3_typing() -> None:
+    """Compile-time checks for async v3 typing — never called at runtime."""
+    graph = _make_simple_graph().compile()
+    run = await graph.astream_events(_SIMPLE_INPUT, version="v3")
+    assert_type(run, AsyncGraphRunStream)
+    assert_type(run.values, StreamChannel[dict[str, Any]])
+    assert_type(run.messages, StreamChannel[AsyncChatModelStream])
+    assert_type(run.lifecycle, StreamChannel[LifecyclePayload])
+    assert_type(run.subgraphs, StreamChannel[AsyncSubgraphRunStream])