Broaden compile(transformers=...) signature to accept scope-aware factories

# Conflicts:
#	libs/langgraph/langgraph/stream/__init__.py
#	libs/langgraph/langgraph/stream/graph_streamer.py
#	libs/langgraph/langgraph/stream/run_stream.py
#	libs/langgraph/tests/test_graph_streamer.py
#	libs/langgraph/tests/test_pregel_stream_v2.py
#	libs/langgraph/tests/test_stream_messages_transformer.py
#	libs/langgraph/tests/test_stream_subgraph_transformer.py
#	libs/langgraph/tests/test_streaming_handler.py

libs/langgraph/langgraph/_internal/_constants.py

@@ -68,7 +68,7 @@ CONFIG_KEY_RESUME_MAP = sys.intern("__pregel_resume_map")
 # holds a mapping of task ns -> resume value for resuming tasks
 CONFIG_KEY_STREAM_MESSAGES_V2 = sys.intern("__pregel_stream_messages_v2")
 # when True, attach StreamMessagesHandlerV2 so content-block (v2) events
-# flow through stream_mode="messages"; set by StreamingHandler only.
+# flow through stream_mode="messages"; set by GraphStreamer only.
 
 # --- Other constants ---
 PUSH = sys.intern("__pregel_push")

libs/langgraph/langgraph/config.py

@@ -1,5 +1,7 @@
 import asyncio
 import sys
+from collections.abc import Callable
+from contextvars import ContextVar
 from typing import Any
 
 from langchain_core.runnables import RunnableConfig
@@ -9,6 +11,18 @@ from langgraph.store.base import BaseStore
 from langgraph._internal._constants import CONF, CONFIG_KEY_RUNTIME
 from langgraph.types import StreamWriter
 
+_tool_call_writer: ContextVar[Callable[[Any], None] | None] = ContextVar(
+    "langgraph_tool_call_writer", default=None
+)
+"""ContextVar holding the writer for the currently-executing tool call.
+
+Set by `StreamToolCallHandler.on_tool_start` and reset on end/error.
+Defined here (rather than alongside the handler in `pregel/_tools.py`)
+so `emit_tool_output_delta` can import it without triggering the
+pregel import chain — user tool code does
+`from langgraph.config import emit_tool_output_delta` at import time.
+"""
+
 
 def _no_op_stream_writer(c: Any) -> None:
     pass
@@ -194,3 +208,30 @@ def get_stream_writer() -> StreamWriter:
     """
     runtime = get_config()[CONF][CONFIG_KEY_RUNTIME]
     return runtime.stream_writer
+
+
+def emit_tool_output_delta(delta: Any) -> None:
+    """Emit a `tool-output-delta` event onto the `tools` stream mode.
+
+    Must be called from inside a tool's execution scope (sync or async).
+    While a tool is running, `StreamToolCallHandler.on_tool_start` sets a
+    writer closure on a ContextVar keyed to that call's `tool_call_id`
+    and namespace; this helper reads the ContextVar and forwards `delta`
+    through it.
+
+    When called outside any tool call, or when the graph was not
+    streamed with `"tools"` in `stream_mode`, this is a silent no-op —
+    tool authors can leave `emit_tool_output_delta` calls in place
+    without gating them on stream mode.
+
+    Args:
+        delta: The partial output chunk to stream. Shape is up to the
+            caller — strings are the common case, but any JSON-
+            serializable value is accepted and surfaced as-is on the
+            `tools` channel's `tool-output-delta` payload under
+            `"delta"`.
+    """
+    writer = _tool_call_writer.get()
+    if writer is None:
+        return
+    writer(delta)

libs/langgraph/langgraph/graph/state.py

@@ -1045,7 +1045,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         interrupt_after: All | list[str] | None = None,
         debug: bool = False,
         name: str | None = None,
-        transformers: Sequence[Callable[[], Any]] | None = None,
+        transformers: Sequence[Callable[..., Any]] | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 

libs/langgraph/langgraph/pregel/_lifecycle.py

@@ -0,0 +1,291 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Iterator
+from typing import Any, TypeVar, cast
+from uuid import UUID
+
+from langchain_core.callbacks import BaseCallbackHandler
+
+from langgraph._internal._constants import NS_SEP
+from langgraph.errors import GraphInterrupt
+from langgraph.pregel.protocol import StreamChunk
+
+try:
+    from langchain_core.tracers._streaming import _StreamingCallbackHandler
+except ImportError:
+    _StreamingCallbackHandler = object  # type: ignore[assignment,misc]
+
+
+T = TypeVar("T")
+
+
+_LANGGRAPH_SENTINEL_NODES = frozenset({"__start__", "__end__"})
+
+
+def _is_nested_pregel_start(
+    name: str | None,
+    metadata: dict[str, Any] | None,
+    parent_run_id: UUID | None,
+    task_run_ids: set[UUID],
+) -> bool:
+    """Recognize a nested `Pregel` invocation from its `on_chain_start` metadata.
+
+    When a compiled graph is added as a node, pregel fires two
+    `on_chain_start` callbacks at that task: first for the node chain
+    (whose `name` matches `metadata["langgraph_node"]`) and second for
+    the inner `Pregel` chain (whose `name` is the graph's `name`, not
+    the node name). Both share the same `langgraph_checkpoint_ns`.
+
+    Primary signal: a `langgraph_checkpoint_ns` is set AND `name`
+    differs from the owning task's `langgraph_node`. This covers the
+    common case where the compiled subgraph's name differs from the
+    node name it was registered under.
+
+    Fallback for name collisions (subgraph compiled with
+    `name == node_name`): the inner `Pregel` start's `parent_run_id`
+    is the run_id of the node chain's start event, which the handler
+    records in `task_run_ids` on the first start. Matching
+    `parent_run_id` to that set identifies the second start as the
+    nested `Pregel` even when names coincide.
+
+    Regular node chains are skipped; the root `Pregel` (which has no
+    `langgraph_node` metadata) isn't observed by this handler because
+    the root's start fires before the handler is attached.
+
+    Metadata-based detection is used because `on_chain_start`'s
+    `serialized` argument is `None` for compiled graphs in this
+    version of langchain-core, so class-based detection via
+    `serialized["id"]` isn't available.
+
+    Sentinel nodes (`__start__` / `__end__`) are excluded: conditional
+    edges from `START` fire an `on_chain_start` with `lg_node=__start__`
+    and the router function's name as `name`, which would otherwise
+    match the discriminator without representing an actual nested
+    `Pregel`.
+
+    Args:
+        name: The `name` kwarg from `on_chain_start`.
+        metadata: The `metadata` kwarg from `on_chain_start`.
+        parent_run_id: The `parent_run_id` kwarg from `on_chain_start`.
+        task_run_ids: The set of run_ids the handler has already seen
+            as node-chain starts (i.e. `name == langgraph_node`).
+    """
+    if not metadata:
+        return False
+    if not metadata.get("langgraph_checkpoint_ns"):
+        return False
+    lg_node = metadata.get("langgraph_node")
+    if lg_node is None or lg_node in _LANGGRAPH_SENTINEL_NODES:
+        return False
+    if name != lg_node:
+        return True
+    # Name collision fallback: the inner Pregel's parent_run_id is
+    # the node chain's run_id, which we recorded when that node
+    # chain's start fired.
+    return parent_run_id is not None and parent_run_id in task_run_ids
+
+
+class StreamLifecycleHandler(BaseCallbackHandler, _StreamingCallbackHandler):
+    """Callback handler that emits subgraph lifecycle events on the stream.
+
+    Pushes `LifecycleData`-shaped payloads onto the pregel stream under
+    the `"lifecycle"` mode, keyed by the subgraph's namespace tuple.
+    Drives the `started` → `running` → `completed` / `failed` /
+    `interrupted` state machine.
+
+    The handler is attached to `run_manager.inheritable_handlers` inside
+    a `Pregel.stream` / `astream` call, so it sees callbacks for every
+    descendant chain (nodes, nested `Pregel` subgraphs) but *not* for
+    the root `Pregel` whose start event has already fired. The root's
+    `started` event is emitted eagerly at construction; its terminal
+    state is emitted by `SubgraphTransformer.finalize` / `fail`.
+
+    `run_inline = True` keeps event ordering deterministic.
+    """
+
+    run_inline = True
+
+    def __init__(
+        self,
+        stream: Callable[[StreamChunk], None],
+        *,
+        root_graph_name: str | None = None,
+    ) -> None:
+        """Initialize the handler and emit the root graph's `started` event.
+
+        Args:
+            stream: Callable that accepts a `StreamChunk` tuple
+                `(namespace, mode, payload)` and enqueues it.
+            root_graph_name: The root `Pregel` instance's `name`, emitted
+                with the root's `started` lifecycle payload.
+        """
+        self.stream = stream
+        # Namespaces awaiting the started→running transition.
+        self._pending_running: set[tuple[str, ...]] = set()
+        # run_id → subgraph namespace; populated only for Pregel chains.
+        self._run_to_ns: dict[UUID, tuple[str, ...]] = {}
+        # run_ids of node-chain starts (name == langgraph_node); used
+        # as the parent_run_id fallback when a subgraph's name equals
+        # its node name. Cleared as each chain ends.
+        self._task_run_ids: set[UUID] = set()
+
+        root_payload: dict[str, Any] = {"event": "started"}
+        if root_graph_name is not None:
+            root_payload["graph_name"] = root_graph_name
+        self.stream(((), "lifecycle", root_payload))
+        self._pending_running.add(())
+
+    @staticmethod
+    def _subgraph_ns_from_metadata(metadata: dict[str, Any] | None) -> tuple[str, ...]:
+        """Return the running subgraph's own namespace from task metadata.
+
+        For a nested `Pregel` invoked as a node, `langgraph_checkpoint_ns`
+        ends at the node segment (no inner task appended yet), so
+        splitting on `NS_SEP` gives the subgraph's own namespace.
+        """
+        if not metadata:
+            return ()
+        nskey = metadata.get("langgraph_checkpoint_ns")
+        if not nskey:
+            return ()
+        return tuple(cast(str, nskey).split(NS_SEP))
+
+    @staticmethod
+    def _containing_ns_from_metadata(
+        metadata: dict[str, Any] | None,
+    ) -> tuple[str, ...]:
+        """Return the namespace of the subgraph that contains this task.
+
+        For an inner task with `langgraph_checkpoint_ns`
+        `"seg_a|seg_b"`, the containing subgraph is `("seg_a",)`.
+        """
+        if not metadata:
+            return ()
+        nskey = metadata.get("langgraph_checkpoint_ns")
+        if not nskey:
+            return ()
+        return tuple(cast(str, nskey).split(NS_SEP))[:-1]
+
+    @staticmethod
+    def _trigger_call_id(metadata: dict[str, Any] | None) -> str | None:
+        """Extract `trigger_call_id` from task metadata if present.
+
+        The task that spawned a nested `Pregel` has its task id encoded
+        in `langgraph_checkpoint_ns`'s last segment as
+        `node_name:task_id`. Returns the `task_id` portion, which
+        parents can correlate with their `tools` / `tasks` events.
+        """
+        if not metadata:
+            return None
+        nskey = cast(str | None, metadata.get("langgraph_checkpoint_ns"))
+        if not nskey:
+            return None
+        last = nskey.split(NS_SEP)[-1]
+        _, sep, task_id = last.rpartition(":")
+        return task_id if sep else None
+
+    def _emit(self, ns: tuple[str, ...], payload: dict[str, Any]) -> None:
+        self.stream((ns, "lifecycle", payload))
+
+    def tap_output_aiter(
+        self, run_id: UUID, output: AsyncIterator[T]
+    ) -> AsyncIterator[T]:
+        """Pass-through — required by the `_StreamingCallbackHandler` protocol.
+
+        Returns the iterator unchanged. A missing implementation lets
+        langchain's default `Protocol` body return `None`, which breaks
+        the `_consume_aiter` code path in `_runnable.py:900`.
+        """
+        return output
+
+    def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
+        """Pass-through — sync counterpart to `tap_output_aiter`."""
+        return output
+
+    def _fire_running_if_pending(self, ns: tuple[str, ...]) -> None:
+        if ns in self._pending_running:
+            self._pending_running.discard(ns)
+            self._emit(ns, {"event": "running"})
+
+    def on_chain_start(
+        self,
+        serialized: dict[str, Any],
+        inputs: dict[str, Any],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        # Any descendant activity transitions the containing subgraph to running.
+        containing = self._containing_ns_from_metadata(metadata)
+        self._fire_running_if_pending(containing)
+
+        name = cast(str | None, kwargs.get("name"))
+        lg_node = (metadata or {}).get("langgraph_node")
+
+        # Record node-chain starts so the name-collision fallback in
+        # `_is_nested_pregel_start` can match the inner Pregel's
+        # parent_run_id to them.
+        if (
+            lg_node is not None
+            and lg_node not in _LANGGRAPH_SENTINEL_NODES
+            and name == lg_node
+        ):
+            self._task_run_ids.add(run_id)
+
+        if not _is_nested_pregel_start(
+            name, metadata, parent_run_id, self._task_run_ids
+        ):
+            return
+
+        ns = self._subgraph_ns_from_metadata(metadata)
+        if not ns:
+            return
+
+        self._run_to_ns[run_id] = ns
+        payload: dict[str, Any] = {"event": "started"}
+        if name:
+            payload["graph_name"] = name
+        trigger_call_id = self._trigger_call_id(metadata)
+        if trigger_call_id:
+            payload["trigger_call_id"] = trigger_call_id
+        self._emit(ns, payload)
+        self._pending_running.add(ns)
+
+    def on_chain_end(
+        self,
+        response: Any,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._task_run_ids.discard(run_id)
+        ns = self._run_to_ns.pop(run_id, None)
+        if ns is None:
+            return
+        # Ensure started→running fired even for empty subgraphs.
+        if ns in self._pending_running:
+            self._pending_running.discard(ns)
+            self._emit(ns, {"event": "running"})
+        self._emit(ns, {"event": "completed"})
+
+    def on_chain_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._task_run_ids.discard(run_id)
+        ns = self._run_to_ns.pop(run_id, None)
+        if ns is None:
+            return
+        self._pending_running.discard(ns)
+        if isinstance(error, GraphInterrupt):
+            self._emit(ns, {"event": "interrupted"})
+        else:
+            self._emit(ns, {"event": "failed", "error": str(error)})

libs/langgraph/langgraph/pregel/_messages.py

@@ -273,7 +273,7 @@ class StreamMessagesHandlerV2(StreamMessagesHandler, _V2StreamingCallbackHandler
     which forwards protocol events onto the messages stream channel.
 
     Pregel attaches this class instead of the v1 handler only when
-    `StreamingHandler` opts in via the internal
+    `GraphStreamer` opts in via the internal
     `CONFIG_KEY_STREAM_MESSAGES_V2` config key; direct
     `graph.stream(stream_mode="messages")` callers keep the v1
     AIMessageChunk shape.

libs/langgraph/langgraph/pregel/_tools.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Iterator
+from contextvars import Token
+from typing import Any, TypeVar, cast
+from uuid import UUID
+
+from langchain_core.callbacks import BaseCallbackHandler
+
+from langgraph._internal._constants import NS_SEP
+from langgraph.config import _tool_call_writer
+from langgraph.pregel.protocol import StreamChunk
+
+try:
+    from langchain_core.tracers._streaming import _StreamingCallbackHandler
+except ImportError:
+    _StreamingCallbackHandler = object  # type: ignore[assignment,misc]
+
+
+T = TypeVar("T")
+
+ToolCallWriter = Callable[[Any], None]
+"""A closure bound to a single tool call that emits `tool-output-delta` events."""
+
+
+class StreamToolCallHandler(BaseCallbackHandler, _StreamingCallbackHandler):
+    """Callback handler that emits tool-call lifecycle events on the stream.
+
+    Fires on LangChain's `on_tool_*` callbacks and pushes to the `tools`
+    stream mode. Emits `tool-started` / `tool-output-delta` /
+    `tool-finished` / `tool-error` payloads keyed by `tool_call_id`.
+
+    While a tool is executing, this handler sets `_tool_call_writer` to a
+    closure bound to that call's namespace and `tool_call_id`. The
+    `emit_tool_output_delta` helper in `langgraph.config` reads that
+    ContextVar so tool bodies can stream partial output without threading
+    the writer through their own signature.
+
+    Attached by `Pregel.stream` / `astream` when `"tools"` is in
+    `stream_modes`. `run_inline = True` keeps event ordering
+    deterministic.
+    """
+
+    run_inline = True
+
+    def __init__(self, stream: Callable[[StreamChunk], None]) -> None:
+        """Initialize the handler.
+
+        Args:
+            stream: Callable that accepts a `StreamChunk` tuple
+                `(namespace, mode, payload)` and enqueues it.
+        """
+        self.stream = stream
+        # run_id → (namespace, tool_call_id, ContextVar token)
+        # `on_tool_end` does not receive `tool_call_id` in kwargs, so
+        # we correlate by `run_id` which is present on every callback.
+        self._run_to_call: dict[
+            UUID, tuple[tuple[str, ...], str, Token[ToolCallWriter | None]]
+        ] = {}
+
+    @staticmethod
+    def _containing_ns_from_metadata(
+        metadata: dict[str, Any] | None,
+    ) -> tuple[str, ...]:
+        """Return the namespace of the subgraph that contains this tool call.
+
+        `langgraph_checkpoint_ns` on a tool's callback metadata ends with
+        the `node_name:task_id` segment of the node that invoked the
+        tool. Dropping that segment gives the subgraph's own namespace,
+        which matches what other `tools` / `lifecycle` / `messages`
+        emitters use.
+        """
+        if not metadata:
+            return ()
+        nskey = metadata.get("langgraph_checkpoint_ns")
+        if not nskey:
+            return ()
+        return tuple(cast(str, nskey).split(NS_SEP))[:-1]
+
+    def _start(
+        self,
+        serialized: dict[str, Any] | None,
+        input_str: str,
+        *,
+        run_id: UUID,
+        metadata: dict[str, Any] | None,
+        inputs: dict[str, Any] | None,
+        kwargs: dict[str, Any],
+    ) -> None:
+        tool_call_id = cast("str | None", kwargs.get("tool_call_id")) or str(run_id)
+        tool_name = (
+            (serialized or {}).get("name")
+            or cast("str | None", kwargs.get("name"))
+            or ""
+        )
+        ns = self._containing_ns_from_metadata(metadata)
+
+        def writer(delta: Any) -> None:
+            self.stream(
+                (
+                    ns,
+                    "tools",
+                    {
+                        "event": "tool-output-delta",
+                        "tool_call_id": tool_call_id,
+                        "delta": delta,
+                    },
+                )
+            )
+
+        token = _tool_call_writer.set(writer)
+        self._run_to_call[run_id] = (ns, tool_call_id, token)
+
+        payload: dict[str, Any] = {
+            "event": "tool-started",
+            "tool_call_id": tool_call_id,
+            "tool_name": tool_name,
+        }
+        if inputs is not None:
+            payload["input"] = inputs
+        self.stream((ns, "tools", payload))
+
+    def _end(self, output: Any, *, run_id: UUID) -> None:
+        info = self._run_to_call.pop(run_id, None)
+        if info is None:
+            return
+        ns, tool_call_id, token = info
+        self._reset_writer(token)
+        self.stream(
+            (
+                ns,
+                "tools",
+                {
+                    "event": "tool-finished",
+                    "tool_call_id": tool_call_id,
+                    "output": output,
+                },
+            )
+        )
+
+    def _error(self, error: BaseException, *, run_id: UUID) -> None:
+        info = self._run_to_call.pop(run_id, None)
+        if info is None:
+            return
+        ns, tool_call_id, token = info
+        self._reset_writer(token)
+        self.stream(
+            (
+                ns,
+                "tools",
+                {
+                    "event": "tool-error",
+                    "tool_call_id": tool_call_id,
+                    "message": str(error),
+                },
+            )
+        )
+
+    def tap_output_aiter(
+        self, run_id: UUID, output: AsyncIterator[T]
+    ) -> AsyncIterator[T]:
+        """Pass-through — required by the `_StreamingCallbackHandler` protocol."""
+        return output
+
+    def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
+        """Pass-through — sync counterpart to `tap_output_aiter`."""
+        return output
+
+    @staticmethod
+    def _reset_writer(token: Token[ToolCallWriter | None]) -> None:
+        # Token is invalid if `on_tool_end` runs in a different context
+        # than `on_tool_start` (e.g. langchain may hand off to a thread
+        # worker without copying the context). Swallow that case; the
+        # ContextVar lifetime is bounded by the enclosing task anyway.
+        try:
+            _tool_call_writer.reset(token)
+        except ValueError:
+            pass
+
+    # ------------------------------------------------------------------
+    # Sync callbacks
+    # ------------------------------------------------------------------
+
+    def on_tool_start(
+        self,
+        serialized: dict[str, Any],
+        input_str: str,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        inputs: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._start(
+            serialized,
+            input_str,
+            run_id=run_id,
+            metadata=metadata,
+            inputs=inputs,
+            kwargs=kwargs,
+        )
+
+    def on_tool_end(
+        self,
+        output: Any,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._end(output, run_id=run_id)
+
+    def on_tool_error(
+        self,
+        error: BaseException,
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._error(error, run_id=run_id)

libs/langgraph/langgraph/pregel/main.py

@@ -130,6 +130,7 @@ from langgraph.pregel._checkpoint import (
 )
 from langgraph.pregel._draw import draw_graph
 from langgraph.pregel._io import map_input, read_channels
+from langgraph.pregel._lifecycle import StreamLifecycleHandler
 from langgraph.pregel._loop import (
     AsyncPregelLoop,
     SyncPregelLoop,
@@ -141,6 +142,7 @@ from langgraph.pregel._messages import (
 from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
 from langgraph.pregel._retry import RetryPolicy
 from langgraph.pregel._runner import PregelRunner
+from langgraph.pregel._tools import StreamToolCallHandler
 from langgraph.pregel._utils import get_new_channel_versions
 from langgraph.pregel._validate import validate_graph, validate_keys
 from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
@@ -344,15 +346,18 @@ class NodeBuilder:
         )
 
 
-_STREAM_V2_MODES: list[StreamMode] = [
-    "values",
-    "updates",
-    "messages",
-    "custom",
-    "checkpoints",
-    "tasks",
-    "debug",
-]
+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
+    default set. If zero transformers are registered (or none declares
+    a given mode), the graph does not stream events for that mode.
+    """
+    modes: set[str] = set()
+    for transformer in mux._transformers:
+        modes.update(transformer.required_stream_modes)
+    return cast("list[StreamMode]", list(modes))
 
 
 def _build_stream_factories(
@@ -361,20 +366,11 @@ def _build_stream_factories(
 ) -> list[Callable[..., Any]]:
     """Assemble the factory list handed to `StreamMux(factories=...)`.
 
-    Prepends the auto-registered built-ins — `ValuesTransformer`
-    (state snapshots backing `run.output` / `run.interrupted`),
-    `MessagesTransformer` (LLM token streaming), and
-    `SubgraphTransformer` (in-process subgraph handle discovery) —
-    then appends the graph's compile-time `stream_transformers`
-    followed by any call-site additions. Factories flow down into
-    subgraph mini-muxes, so per-scope instances propagate
-    automatically.
-
-    `LifecycleTransformer` is opt-in: add it via compile-time
-    `stream_transformers=[...]` or the per-call `transformers=[...]`
-    kwarg on `stream_v2()` / `astream_v2()`. Without it, no
-    `lifecycle` wire events are emitted and `run.lifecycle` is
-    absent.
+    Prepends the built-in `ValuesTransformer`, `MessagesTransformer`,
+    and `SubgraphTransformer` factories, then appends the graph's
+    compile-time `stream_transformers` followed by any call-site
+    additions. Factories flow down into subgraph mini-muxes, so
+    per-scope instances propagate automatically.
     """
     from langgraph.stream.transformers import (
         MessagesTransformer,
@@ -2710,6 +2706,21 @@ class Pregel(
                     )
                 )
 
+            # set up lifecycle stream mode
+            if "lifecycle" in stream_modes:
+                run_manager.inheritable_handlers.append(
+                    StreamLifecycleHandler(
+                        stream.put,
+                        root_graph_name=self.name,
+                    )
+                )
+
+            # set up tools stream mode
+            if "tools" in stream_modes:
+                run_manager.inheritable_handlers.append(
+                    StreamToolCallHandler(stream.put)
+                )
+
             # set up custom stream mode
             if "custom" in stream_modes:
 
@@ -3092,6 +3103,21 @@ class Pregel(
                     )
                 )
 
+            # set up lifecycle stream mode
+            if "lifecycle" in stream_modes:
+                run_manager.inheritable_handlers.append(
+                    StreamLifecycleHandler(
+                        stream_put,
+                        root_graph_name=self.name,
+                    )
+                )
+
+            # set up tools stream mode
+            if "tools" in stream_modes:
+                run_manager.inheritable_handlers.append(
+                    StreamToolCallHandler(stream_put)
+                )
+
             # set up custom stream mode
             def stream_writer(c: Any) -> None:
                 aioloop.call_soon_threadsafe(
@@ -3329,19 +3355,13 @@ class Pregel(
     ) -> Any:
         """Start a sync v2 streaming run driven by transformer projections.
 
-        Builds a `StreamMux` from the auto-registered built-ins
-        (`ValuesTransformer`, `MessagesTransformer`,
-        `SubgraphTransformer`), this graph's compile-time
+        Builds a `StreamMux` from the built-in `ValuesTransformer` /
+        `MessagesTransformer`, 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
+        supplied at the call site. Returns a `GraphRunStream` that the
+        caller drives by iterating any projection — no background
         thread.
 
-        `LifecycleTransformer` (emits `lifecycle` wire events,
-        exposes `run.lifecycle`) is opt-in — add it via
-        `stream_transformers` at compile time or via the
-        `transformers=` kwarg here.
-
         Args:
             input: Graph input.
             config: Optional runnable config forwarded to the graph.
@@ -3358,11 +3378,12 @@ class Pregel(
 
         factories = _build_stream_factories(self._stream_transformers, transformers)
         mux = StreamMux(factories=factories, is_async=False)
+        stream_modes = _collect_stream_modes(mux)
         graph_iter = iter(
             self.stream(
                 input,
                 _merge_v2_messages_flag(config),
-                stream_mode=_STREAM_V2_MODES,
+                stream_mode=stream_modes,
                 subgraphs=True,
                 version="v2",
                 interrupt_before=interrupt_before,
@@ -3399,10 +3420,11 @@ class Pregel(
 
         factories = _build_stream_factories(self._stream_transformers, transformers)
         mux = StreamMux(factories=factories, is_async=True)
+        stream_modes = _collect_stream_modes(mux)
         graph_aiter = self.astream(
             input,
             _merge_v2_messages_flag(config),
-            stream_mode=_STREAM_V2_MODES,
+            stream_mode=stream_modes,
             subgraphs=True,
             version="v2",
             interrupt_before=interrupt_before,

libs/langgraph/langgraph/pregel/remote.py

@@ -650,16 +650,29 @@ class RemoteGraph(PregelProtocol):
         """
         updated_stream_modes: list[StreamModeSDK] = []
         req_single = True
+        # `"lifecycle"` is emitted locally by the `StreamLifecycleHandler`
+        # attached inside `Pregel.stream` / `astream`. The remote graph
+        # API has no corresponding mode, so requests for it against a
+        # `RemoteGraph` are silently stripped here and a warning is
+        # logged so the caller isn't left wondering why no lifecycle
+        # events arrive.
+        dropped_lifecycle = False
         # coerce to list, or add default stream mode
         if stream_mode:
             if isinstance(stream_mode, str):
-                updated_stream_modes.append(cast(StreamModeSDK, stream_mode))
+                if stream_mode != "lifecycle":
+                    updated_stream_modes.append(cast(StreamModeSDK, stream_mode))
+                else:
+                    dropped_lifecycle = True
             else:
                 req_single = False
                 for m in stream_mode:
-                    updated_stream_modes.append(cast(StreamModeSDK, m))
+                    if m == "lifecycle":
+                        dropped_lifecycle = True
+                    else:
+                        updated_stream_modes.append(cast(StreamModeSDK, m))
         else:
-            updated_stream_modes.append(default)
+            updated_stream_modes.append(default)  # type: ignore[arg-type]
         requested_stream_modes = updated_stream_modes.copy()
         # add any from parent graph
         stream: StreamProtocol | None = (
@@ -667,7 +680,16 @@ class RemoteGraph(PregelProtocol):
         )
         if stream:
             for m in stream.modes:
-                updated_stream_modes.append(cast(StreamModeSDK, m))
+                if m == "lifecycle":
+                    dropped_lifecycle = True
+                else:
+                    updated_stream_modes.append(cast(StreamModeSDK, m))
+        if dropped_lifecycle:
+            logger.warning(
+                "Stream mode 'lifecycle' is not supported by RemoteGraph "
+                "and was stripped from the request; no lifecycle events "
+                "will be emitted for this remote run."
+            )
         # map "messages" to "messages-tuple"
         if "messages" in updated_stream_modes:
             updated_stream_modes.remove("messages")

libs/langgraph/langgraph/stream/_event_log.py

@@ -39,20 +39,13 @@ class EventLog(Generic[T]):
     skipped.
     """
 
-    def __init__(self, maxlen: int | None = None, *, retain: bool = False) -> None:
+    def __init__(self, maxlen: int | None = None) -> None:
         """Initialize an empty, unbound log.
 
         Args:
             maxlen: Accepted for forward compatibility; currently unused.
                 The caller-driven pump bounds memory naturally for
                 single-consumer use.
-            retain: If True, `push()` retains items regardless of whether
-                a consumer has subscribed yet. Used for projections
-                whose consumer only becomes visible after events have
-                already flowed (e.g. mini-mux logs inside dynamically
-                discovered subgraph handles, or the `lifecycle` channel
-                iterated after draining `values`). Subscription
-                exclusivity on `__iter__` is unchanged.
 
         Raises:
             ValueError: If `maxlen` is not a positive integer or `None`.
@@ -68,9 +61,8 @@ class EventLog(Generic[T]):
         self._is_async: bool | None = None
 
         # Flipped on first __iter__ / __aiter__. Pre-subscription
-        # pushes are silent no-ops unless `_retain` is True.
+        # pushes are silent no-ops.
         self._subscribed = False
-        self._retain = retain
 
         # Pump wiring set by the run stream after bind.
         self._request_more: Callable[[], bool] | None = None
@@ -107,14 +99,10 @@ class EventLog(Generic[T]):
         `put_nowait` producer shape. Memory is bounded by caller pace
         via the caller-driven pump.
 
-        When `retain=True` was set at construction, items are appended
-        regardless of subscription — for projections whose consumer
-        only reaches them after events have already flowed.
-
         Raises:
             RuntimeError: If the log is closed (and subscribed).
         """
-        if not self._subscribed and not self._retain:
+        if not self._subscribed:
             return
         if self._closed:
             raise RuntimeError("Cannot push to a closed EventLog")

libs/langgraph/langgraph/stream/_mux.py

@@ -140,18 +140,6 @@ class StreamMux:
             is_async=self._is_async,
             scope=scope,
         )
-        # Mini-muxes are created during the pump, after the first
-        # event at the child's scope has already been dispatched.
-        # Consumers reach child projections via the parent's
-        # `subgraphs` handle — necessarily after that first event.
-        # Flip retain on every log and channel so pushes are buffered
-        # until the consumer subscribes.
-        child._events._retain = True
-        for value in child.extensions.values():
-            if isinstance(value, EventLog):
-                value._retain = True
-            elif isinstance(value, StreamChannel):
-                value._log._retain = True
         if self._pump_fn is not None:
             child.bind_pump(self._pump_fn)
         if self._apump_fn is not None:
@@ -223,14 +211,13 @@ class StreamMux:
                 f"keys: {attributions}"
             )
         self._transformers.append(transformer)
-        is_native = bool(getattr(transformer, "_native", False))
-        self._bind_and_wire(projection, is_native=is_native)
+        self._bind_and_wire(projection)
         self.extensions.update(projection)
         owner_name = type(transformer).__name__
         for key in projection:
             self._projection_owners[key] = owner_name
             self._transformer_by_key[key] = transformer
-        if is_native:
+        if getattr(transformer, "_native", False):
             self.native_keys.update(projection.keys())
         on_register = getattr(transformer, "_on_register", None)
         if on_register is not None:
@@ -458,34 +445,26 @@ class StreamMux:
     # Binding and StreamChannel auto-wiring
     # ------------------------------------------------------------------
 
-    def _bind_and_wire(
-        self, projection: dict[str, Any], *, is_native: bool = False
-    ) -> None:
-        """Bind and wire EventLog / StreamChannel instances in a projection.
-
-        `is_native` controls wire naming: native transformer channels
-        emit events with `method` equal to the channel name, while
-        non-native channels get a `custom:` prefix to keep user-defined
-        projections from colliding with built-in method names.
-        """
+    def _bind_and_wire(self, projection: dict[str, Any]) -> None:
+        """Bind and wire EventLog / StreamChannel instances in a projection."""
         for value in projection.values():
             if isinstance(value, StreamChannel):
                 value._bind(is_async=self._is_async)
                 self._channels.append(value)
                 channel_name = value.name
 
-                def _make_forward(name: str, native: bool) -> Callable[[Any], None]:
+                def _make_forward(name: str) -> Callable[[Any], None]:
                     def _forward(item: Any) -> None:
-                        self._forward(name, item, native=native)
+                        self._forward(name, item)
 
                     return _forward
 
-                value._wire(_make_forward(channel_name, is_native))
+                value._wire(_make_forward(channel_name))
             elif isinstance(value, EventLog):
                 value._bind(is_async=self._is_async)
                 self._logs.append(value)
 
-    def _forward(self, channel_name: str, item: Any, *, native: bool) -> None:
+    def _forward(self, channel_name: str, item: Any) -> None:
         """Inject a ProtocolEvent for a StreamChannel push.
 
         Forwarded events bypass the transformer pipeline to avoid
@@ -493,18 +472,12 @@ class StreamMux:
         during `process()` would re-trigger itself). These events are
         visible in the main event log but are not passed through
         transformers' `process()` methods.
-
-        Native transformers emit with `method` equal to the channel
-        name (e.g. `"lifecycle"`); non-native transformers get a
-        `custom:` prefix so user-defined projections can't collide
-        with built-in method names.
         """
         self._seq += 1
-        method = channel_name if native else f"custom:{channel_name}"
         event: ProtocolEvent = {
             "type": "event",
             "seq": self._seq,
-            "method": method,
+            "method": f"custom:{channel_name}",
             "params": {
                 "namespace": [],
                 "timestamp": int(time.time() * 1000),

libs/langgraph/langgraph/stream/_types.py

@@ -94,10 +94,17 @@ class StreamTransformer(ABC):
             example, transformers that call `schedule()` from a sync
             `process`). The mux also auto-detects the async lane when
             `aprocess`, `afinalize`, or `afail` is overridden.
+        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 `GraphStreamer` run requests from the
+            graph. Empty tuple means the transformer consumes only
+            synthetic events (or is purely passive).
     """
 
     requires_async: ClassVar[bool] = False
     scope_exact: ClassVar[bool] = True
+    required_stream_modes: ClassVar[tuple[str, ...]] = ()
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         """Initialize the transformer with its mux's scope.

libs/langgraph/langgraph/stream/run_stream.py

@@ -49,23 +49,25 @@ class BaseRunStream:
 
     @property
     def _values_transformer(self) -> ValuesTransformer:
-        """Look up the `ValuesTransformer` registered on this mux.
+        """Look up the `ValuesTransformer` backing `output` / `interrupted`.
 
-        `output` / `interrupted` / `interrupts` need scalar state from
-        the `ValuesTransformer` without threading it through the
-        constructor. Raises if none is registered — `stream_v2` /
-        `astream_v2` always register one, so hitting this path means
-        the caller assembled the mux themselves and forgot.
+        Resolved lazily off the mux so subclasses don't have to thread
+        it through their constructors. Raises if no `ValuesTransformer`
+        is registered — `output` / `interrupted` / `interrupts` have
+        nothing to return in that case, so failing loudly is better
+        than returning `None` silently.
         """
         from langgraph.stream.transformers import ValuesTransformer
 
-        for t in self._mux._transformers:
-            if isinstance(t, ValuesTransformer):
-                return t
-        raise RuntimeError(
-            "No ValuesTransformer is registered on this mux — "
-            "`output` / `interrupted` / `interrupts` are unavailable."
-        )
+        vt = self._mux.transformer_by_key("values")
+        if not isinstance(vt, ValuesTransformer):
+            raise RuntimeError(
+                "No ValuesTransformer is registered on this mux — "
+                "`output`, `interrupted`, and `interrupts` require one. "
+                "Add it to your GraphStreamer subclass's "
+                "`builtin_factories` or pass it via `transformers=`."
+            )
+        return vt
 
     def __iter__(self) -> Iterator[ProtocolEvent]:
         """Sync iteration of protocol events on this mux's main log.
@@ -154,8 +156,9 @@ class GraphRunStream(BaseRunStream):
         Args:
             graph_iter: Pull-based iterator over the graph's stream.
             mux: The StreamMux owning projections and the main log.
-                Must have a `ValuesTransformer` registered for
-                `output` / `interrupted` / `interrupts` to work.
+                Must have a `ValuesTransformer` registered under the
+                `"values"` key — `output` / `interrupted` / `interrupts`
+                read from it lazily.
         """
         super().__init__(mux)
         self._graph_iter = graph_iter
@@ -214,10 +217,10 @@ class GraphRunStream(BaseRunStream):
     def output(self) -> dict[str, Any] | None:
         """Drive the run to completion and return the final state."""
         _drive_until_done(self._pump_next)
-        err = self._values_transformer.error
-        if err is not None:
-            raise err
-        return self._values_transformer._latest
+        vt = self._values_transformer
+        if vt.error is not None:
+            raise vt.error
+        return vt._latest
 
     @property
     def interrupted(self) -> bool:
@@ -227,10 +230,10 @@ class GraphRunStream(BaseRunStream):
             BaseException: If the run ended with an error.
         """
         _drive_until_done(self._pump_next)
-        err = self._values_transformer.error
-        if err is not None:
-            raise err
-        return self._values_transformer._interrupted
+        vt = self._values_transformer
+        if vt.error is not None:
+            raise vt.error
+        return vt._interrupted
 
     @property
     def interrupts(self) -> list[Any]:
@@ -240,10 +243,10 @@ class GraphRunStream(BaseRunStream):
             BaseException: If the run ended with an error.
         """
         _drive_until_done(self._pump_next)
-        err = self._values_transformer.error
-        if err is not None:
-            raise err
-        return self._values_transformer._interrupts
+        vt = self._values_transformer
+        if vt.error is not None:
+            raise vt.error
+        return vt._interrupts
 
 
 class AsyncGraphRunStream(BaseRunStream):
@@ -280,8 +283,9 @@ class AsyncGraphRunStream(BaseRunStream):
         Args:
             graph_aiter: Async iterator over the graph's stream.
             mux: The StreamMux owning projections and the main log.
-                Must have a `ValuesTransformer` registered for
-                `output` / `interrupted` / `interrupts` to work.
+                Must have a `ValuesTransformer` registered under the
+                `"values"` key — `output` / `interrupted` / `interrupts`
+                read from it lazily.
         """
         super().__init__(mux)
         self._graph_aiter = graph_aiter

libs/langgraph/langgraph/stream/stream_channel.py

@@ -37,9 +37,7 @@ class StreamChannel(Generic[T]):
     using only StreamChannels don't need `finalize` or `fail` hooks.
     """
 
-    def __init__(
-        self, name: str, *, maxlen: int | None = None, retain: bool = False
-    ) -> None:
+    def __init__(self, name: str, *, maxlen: int | None = None) -> None:
         """Initialize the channel with an empty inner log.
 
         Args:
@@ -47,13 +45,9 @@ class StreamChannel(Generic[T]):
                 events (`custom:<name>` on the wire).
             maxlen: Optional retention cap on the inner EventLog. See
                 `EventLog.__init__` for semantics.
-            retain: If True, the inner log retains pushes before any
-                consumer subscribes — needed for channels whose
-                consumer iterates after events have already flowed
-                (e.g. `lifecycle` inspected after draining `values`).
         """
         self.name = name
-        self._log: EventLog[T] = EventLog(maxlen=maxlen, retain=retain)
+        self._log: EventLog[T] = EventLog(maxlen=maxlen)
         self._wire_fn: Callable[[T], None] | None = None
 
     def _bind(self, *, is_async: bool) -> None:

libs/langgraph/langgraph/stream/transformers.py

@@ -9,14 +9,12 @@ from langchain_core.language_models.chat_model_stream import (
     ChatModelStream,
 )
 from langchain_core.messages import AIMessageChunk, BaseMessage
-from langchain_protocol.protocol import CheckpointRef, MessagesData
-from typing_extensions import TypedDict
+from langchain_protocol.protocol import CheckpointRef, LifecycleData, MessagesData
 
 from langgraph.errors import GraphInterrupt
 from langgraph.stream._event_log import EventLog
 from langgraph.stream._types import ProtocolEvent, StreamTransformer
 from langgraph.stream.run_stream import BaseRunStream
-from langgraph.stream.stream_channel import StreamChannel
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -27,38 +25,12 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-SubgraphStatus = Literal["started", "completed", "failed", "interrupted"]
+SubgraphStatus = Literal["started", "running", "completed", "failed", "interrupted"]
 _TERMINAL_STATUSES: frozenset[SubgraphStatus] = frozenset(
     {"completed", "failed", "interrupted"}
 )
 
 
-def _is_new_direct_child(
-    ns: tuple[str, ...],
-    scope: tuple[str, ...],
-    seen: set[tuple[str, ...]] | dict[tuple[str, ...], Any],
-) -> bool:
-    """Return True iff `ns` is a direct child of `scope` not yet seen.
-
-    Shared by `SubgraphTransformer` (in-process handle discovery) and
-    `LifecycleTransformer` (wire event emission) so the two can't
-    disagree on what counts as a new subgraph.
-    """
-    return len(ns) == len(scope) + 1 and ns[:-1] == scope and ns not in seen
-
-
-def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
-    """Split `node_name:task_id` into (node_name, task_id).
-
-    Task ids are present when Pregel spawned the subgraph as a task;
-    absent on synthesized namespaces (tests, hand-crafted events).
-    """
-    node_name, sep, task_id = segment.partition(":")
-    if not sep:
-        return segment, None
-    return node_name, task_id or None
-
-
 class ValuesTransformer(StreamTransformer):
     """Capture values events as a drainable stream of state snapshots.
 
@@ -77,6 +49,7 @@ class ValuesTransformer(StreamTransformer):
     """
 
     _native = True
+    required_stream_modes = ("values",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -149,6 +122,7 @@ class MessagesTransformer(StreamTransformer):
     """
 
     _native = True
+    required_stream_modes = ("messages",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -292,17 +266,14 @@ class SubgraphRunStream(BaseRunStream):
     `make_child`'s pump inheritance, so any cursor on a subagent
     projection drives the whole run forward.
 
-    Handle fields:
+    Lifecycle fields update in place as events arrive:
 
     - `path`: the namespace tuple — stable for the life of the handle.
-    - `graph_name` / `trigger_call_id`: parsed from the namespace
-      segment at discovery (`node_name:task_id`).
-    - `status`: `started` on discovery; advances to `completed` when
-      the parent mux closes, or `failed` / `interrupted` when it
-      errors.
-    - `error`: set on terminal error.
-    - `checkpoint`: unused by the current discovery path — kept for
-      compatibility with consumers that inspect it.
+    - `graph_name` / `trigger_call_id`: set once from the `started`
+      payload.
+    - `status`: advances `started` → `running` → `completed` /
+      `failed` / `interrupted`.
+    - `error` / `checkpoint`: set on the terminal event when present.
 
     `.output` is a snapshot of the latest values seen at this
     namespace — it doesn't drive the pump (unlike root's
@@ -342,10 +313,10 @@ class SubgraphRunStream(BaseRunStream):
 class SubgraphTransformer(StreamTransformer):
     """Discover subgraphs and route events into per-subgraph mini-muxes.
 
-    Thin dispatcher. At its own `scope` (inherited from
-    `StreamTransformer`, determined by the enclosing mux), it watches
-    for the first event at exactly one namespace level deeper to
-    discover a direct child. Each discovered child gets its own
+    Thin state-machine + dispatcher. At its own `scope` (inherited
+    from `StreamTransformer`, determined by the enclosing mux), it
+    watches for `lifecycle` events at exactly one level deeper to
+    discover direct children. Each discovered child gets its own
     `SubgraphRunStream` backed by a mini-`StreamMux` — built via
     `parent_mux.make_child(path)`, so the same factory list produces
     fresh transformer instances at the child's scope.
@@ -357,15 +328,10 @@ class SubgraphTransformer(StreamTransformer):
     `SubgraphTransformer` for grandchildren) handle the rest. No
     duplicated routing or assembly logic.
 
-    Discovery is method-agnostic: the first event of any mode whose
-    namespace places it directly below `scope` spawns the handle.
-    `graph_name` and `trigger_call_id` are parsed from the namespace
-    segment, which encodes `node_name:task_id`.
-
-    Terminal status for each handle is set by the parent mux's
-    `close` / `fail` path. `finalize` transitions still-open handles
-    to `completed`; `fail` transitions them to `failed` or
-    `interrupted` depending on the error.
+    Lifecycle state for each handle (running / completed / failed /
+    interrupted) is updated in place as events fire. On terminal
+    events, the handle's mini-mux is closed so any subscribed cursors
+    unblock. `finalize` / `fail` handle dangling handles left mid-run.
 
     Native transformer — `subgraphs` exposes the direct-children log.
 
@@ -376,6 +342,7 @@ class SubgraphTransformer(StreamTransformer):
 
     _native = True
     scope_exact = False
+    required_stream_modes = ("lifecycle",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -393,23 +360,45 @@ class SubgraphTransformer(StreamTransformer):
 
     def process(self, event: ProtocolEvent) -> bool:
         ns = tuple(event["params"]["namespace"])
+        method = event["method"]
         depth = len(self.scope)
 
-        # 1. Discover: first-seen direct-child namespace registers a
-        #    handle. Any event method triggers discovery — no dedicated
-        #    channel.
-        if _is_new_direct_child(ns, self.scope, self._by_ns):
-            self._on_started(ns)
+        # 1. On `started` for a direct child (ns depth = mine + 1 and
+        #    ns prefix matches mine), register the handle.
+        if method == "lifecycle" and len(ns) == depth + 1 and ns[:-1] == self.scope:
+            data = cast(LifecycleData, event["params"]["data"])
+            if data.get("event") == "started":
+                self._on_started(ns, data)
 
-        # 2. Forward the event to the matching direct-child mini-mux.
-        #    Prefix-match: ns must start with some child's path.
+        # 2. Forward the event to the matching direct-child mini-mux
+        #    before the status-change step below so that terminal events
+        #    reach the child's log and grandchild transformers *before*
+        #    the child's mini-mux is closed. Prefix-match: ns must start
+        #    with some child's path.
         direct_child_ns = ns[: depth + 1] if len(ns) > depth else None
         if direct_child_ns is not None and direct_child_ns in self._by_ns:
             self._by_ns[direct_child_ns]._mux.push(event)
 
+        # 3. Status change for a direct child (ns = child's path, method
+        #    = lifecycle). Update handle fields, close mini-mux on
+        #    terminal.
+        if (
+            method == "lifecycle"
+            and ns in self._by_ns
+            and len(ns) == depth + 1
+            and ns[:-1] == self.scope
+        ):
+            data = cast(LifecycleData, event["params"]["data"])
+            event_type = data.get("event")
+            if event_type in ("running", "completed", "failed", "interrupted"):
+                self._on_status_change(ns, event_type, data)
+
         return True
 
-    def _on_started(self, ns: tuple[str, ...]) -> None:
+    def _on_started(self, ns: tuple[str, ...], data: LifecycleData) -> None:
+        if ns in self._by_ns:
+            # Duplicate started — ignore.
+            return
         # `_on_register` is called by the mux during registration, which
         # happens before any event can be dispatched — so this should
         # always be set by the time we process an event.
@@ -417,17 +406,33 @@ class SubgraphTransformer(StreamTransformer):
             "SubgraphTransformer processed an event before _on_register; "
             "transformer registration ordering is broken."
         )
-        graph_name, trigger_call_id = _parse_ns_segment(ns[-1])
         child_mux = self._mux.make_child(ns)
         handle = SubgraphRunStream(
             path=ns,
             mux=child_mux,
-            graph_name=graph_name,
-            trigger_call_id=trigger_call_id,
+            graph_name=data.get("graph_name"),
+            trigger_call_id=data.get("trigger_call_id"),
         )
         self._by_ns[ns] = handle
         self._root_log.push(handle)
 
+    def _on_status_change(
+        self,
+        ns: tuple[str, ...],
+        event_type: SubgraphStatus,
+        data: LifecycleData,
+    ) -> None:
+        handle = self._by_ns[ns]
+        handle.status = event_type
+        err = data.get("error")
+        if err is not None:
+            handle.error = err
+        checkpoint = data.get("checkpoint")
+        if checkpoint is not None:
+            handle.checkpoint = checkpoint
+        if event_type in _TERMINAL_STATUSES:
+            self._close_handle_mux(handle)
+
     @staticmethod
     def _close_handle_mux(handle: SubgraphRunStream) -> None:
         # Idempotent close — mux.close() runs finalize on its transformers
@@ -444,21 +449,10 @@ class SubgraphTransformer(StreamTransformer):
                 )
 
     def finalize(self) -> None:
-        """Transition any still-open direct children to `completed`.
-
-        Subgraph interrupts surface as a values event with a populated
-        `interrupts` field rather than as an exception at the graph
-        boundary — the parent pump exhausts normally and `finalize`
-        runs the close path. Inspect each child's `ValuesTransformer`
-        to distinguish "completed cleanly" from "interrupted".
-        """
+        """Transition any still-open direct children to `completed`."""
         for handle in self._by_ns.values():
             if handle.status not in _TERMINAL_STATUSES:
-                values_t = handle._mux.transformer_by_key("values")
-                if isinstance(values_t, ValuesTransformer) and values_t._interrupted:
-                    handle.status = "interrupted"
-                else:
-                    handle.status = "completed"
+                handle.status = "completed"
             self._close_handle_mux(handle)
 
     def fail(self, err: BaseException) -> None:
@@ -481,105 +475,3 @@ class SubgraphTransformer(StreamTransformer):
                         handle.path,
                         exc_info=True,
                     )
-
-
-class LifecyclePayload(TypedDict, total=False):
-    """Payload of a lifecycle event emitted by `LifecycleTransformer`."""
-
-    event: SubgraphStatus
-    namespace: list[str]
-    graph_name: str | None
-    trigger_call_id: str | None
-    error: str | None
-
-
-class LifecycleTransformer(StreamTransformer):
-    """Synthesize subgraph lifecycle events from observed namespaces.
-
-    Observes the same namespace signal `SubgraphTransformer` uses for
-    in-process discovery and emits `started` / `completed` / `failed`
-    / `interrupted` payloads onto its `lifecycle` channel. Consumers
-    subscribed to that channel see the events in-process; wire
-    consumers receive them as protocol events with `method:
-    "lifecycle"` (unprefixed because this transformer is `_native`).
-
-    No `running` event: the ns-discovery signal only fires once a
-    subgraph has emitted output, so `started` already implies
-    execution. Consumers needing finer-grained task-start visibility
-    should read the `tasks` stream mode alongside.
-
-    No root `started`: the run object itself signals run start.
-
-    Terminal events are synthesized — `finalize` emits `completed` for
-    still-open handles; `fail` emits `failed` or `interrupted`
-    depending on whether the error is a `GraphInterrupt`.
-
-    `scope_exact = False` so the transformer sees events at any
-    namespace (needed for discovery of direct children).
-    """
-
-    _native = True
-    scope_exact = False
-
-    def __init__(self, scope: tuple[str, ...] = ()) -> None:
-        super().__init__(scope)
-        # retain=True: lifecycle events are low-volume and consumers
-        # commonly inspect them after draining `values`; without
-        # retention those pushes would be dropped.
-        self._channel: StreamChannel[LifecyclePayload] = StreamChannel(
-            "lifecycle", retain=True
-        )
-        self._seen: set[tuple[str, ...]] = set()
-        self._open: set[tuple[str, ...]] = set()
-
-    def init(self) -> dict[str, Any]:
-        return {"lifecycle": self._channel}
-
-    def process(self, event: ProtocolEvent) -> bool:
-        ns = tuple(event["params"]["namespace"])
-        if _is_new_direct_child(ns, self.scope, self._seen):
-            self._emit_started(ns)
-        return True
-
-    def _emit_started(self, ns: tuple[str, ...]) -> None:
-        graph_name, trigger_call_id = _parse_ns_segment(ns[-1])
-        self._seen.add(ns)
-        self._open.add(ns)
-        payload: LifecyclePayload = {
-            "event": "started",
-            "namespace": list(ns),
-        }
-        if graph_name:
-            payload["graph_name"] = graph_name
-        if trigger_call_id is not None:
-            payload["trigger_call_id"] = trigger_call_id
-        self._channel.push(payload)
-
-    def finalize(self) -> None:
-        """Emit `completed` for every still-open direct child."""
-        for ns in list(self._open):
-            self._channel.push({"event": "completed", "namespace": list(ns)})
-        self._open.clear()
-
-    def fail(self, err: BaseException) -> None:
-        """Emit `failed` / `interrupted` for every still-open direct child.
-
-        Closes the channel after emitting rather than letting the mux
-        auto-fail it — the "failed" payload is the signal to
-        consumers, so they should be able to iterate it. A failed
-        channel would raise on iteration and hide the events that just
-        got pushed.
-        """
-        is_interrupt = isinstance(err, GraphInterrupt)
-        event_type: SubgraphStatus = "interrupted" if is_interrupt else "failed"
-        error_str = None if is_interrupt else str(err)
-        for ns in list(self._open):
-            payload: LifecyclePayload = {
-                "event": event_type,
-                "namespace": list(ns),
-            }
-            if error_str is not None:
-                payload["error"] = error_str
-            self._channel.push(payload)
-        self._open.clear()
-        self._channel._close()

libs/langgraph/langgraph/types.py

@@ -123,6 +123,8 @@ StreamMode = Literal[
     "debug",
     "messages",
     "custom",
+    "lifecycle",
+    "tools",
 ]
 """How the stream method should emit outputs.
 
@@ -135,6 +137,8 @@ StreamMode = Literal[
 - `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by `get_state()`.
 - `"tasks"`: Emit events when tasks start and finish, including their results and errors.
 - `"debug"`: Emit `"checkpoints"` and `"tasks"` events for debugging purposes.
+- `"lifecycle"`: Emit subgraph lifecycle events (`started`, `running`, `completed`, `failed`, `interrupted`) with payloads matching `LifecycleData`.
+- `"tools"`: Emit tool-call lifecycle events (`tool-started`, `tool-output-delta`, `tool-finished`, `tool-error`) keyed by `tool_call_id`.
 """
 
 StreamWriter = Callable[[Any], None]

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_pregel_stream_v2.py

@@ -131,6 +131,24 @@ def _build_custom_stream_graph():
     return builder.compile()
 
 
+class _CustomPassthroughTransformer(StreamTransformer):
+    """Opts a run into the `custom` stream mode without building a projection.
+
+    `stream_v2` requests only the modes that registered
+    transformers declare via `required_stream_modes`. Custom events are
+    raw user emissions from `StreamWriter`, so tests that want them
+    visible on the main event log register this pass-through transformer.
+    """
+
+    required_stream_modes = ("custom",)
+
+    def init(self) -> dict[str, Any]:
+        return {}
+
+    def process(self, event: ProtocolEvent) -> bool:
+        return True
+
+
 # ---------------------------------------------------------------------------
 # EventLog unit tests
 # ---------------------------------------------------------------------------
@@ -465,13 +483,29 @@ class TestStreamV2Sync:
 
     def test_custom_stream_events(self) -> None:
         graph = _build_custom_stream_graph()
-        handler = graph
-        run = handler.stream_v2({"value": "x", "items": []})
+        run = graph.stream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CustomPassthroughTransformer],
+        )
         custom_events = [e for e in run if e["method"] == "custom"]
         assert len(custom_events) == 2
         assert custom_events[0]["params"]["data"] == {"step": "start"}
         assert custom_events[1]["params"]["data"] == {"step": "end"}
 
+    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 transformers require. Built-ins cover
+        `values` / `messages` / `lifecycle`; consumers that want raw
+        custom events surface them by registering a transformer whose
+        `required_stream_modes` includes `"custom"`.
+        """
+        graph = _build_custom_stream_graph()
+        run = graph.stream_v2({"value": "x", "items": []})
+        custom_events = [e for e in run if e["method"] == "custom"]
+        assert custom_events == []
+
     def test_interleave_values_and_messages(self) -> None:
         graph = _build_simple_graph()
         handler = graph
@@ -718,8 +752,10 @@ class TestStreamV2AsyncCustom:
     @NEEDS_CONTEXTVARS
     async def test_custom_stream_events(self) -> None:
         graph = _build_custom_stream_graph()
-        handler = graph
-        run = await handler.astream_v2({"value": "x", "items": []})
+        run = await graph.astream_v2(
+            {"value": "x", "items": []},
+            transformers=[_CustomPassthroughTransformer],
+        )
         events = [e async for e in run]
         custom_events = [e for e in events if e["method"] == "custom"]
         assert len(custom_events) == 2
@@ -1786,3 +1822,92 @@ class TestDrainOnConsume:
         log.close()
         assert list(a) == [0, 1, 2]
         assert list(b) == [0, 1, 2]
+
+
+# ---------------------------------------------------------------------------
+# compile(transformers=...) registration
+# ---------------------------------------------------------------------------
+
+
+class _CountingTransformer(StreamTransformer):
+    """Record how many events the transformer saw — nothing more."""
+
+    required_stream_modes = ("values",)
+
+    def __init__(self, scope: tuple[str, ...] = ()) -> None:
+        super().__init__(scope)
+        self._log: EventLog[int] = EventLog()
+        self.count = 0
+
+    def init(self) -> dict[str, Any]:
+        return {"counts": self._log}
+
+    def process(self, event: ProtocolEvent) -> bool:
+        if event["method"] == "values":
+            self.count += 1
+        return True
+
+
+def _build_counting_graph():
+    """Two-node graph compiled with `_CountingTransformer` pre-registered."""
+
+    def node_a(state: SimpleState) -> dict:
+        return {"value": state["value"] + "A", "items": ["a"]}
+
+    def node_b(state: SimpleState) -> dict:
+        return {"value": state["value"] + "B", "items": ["b"]}
+
+    builder = StateGraph(SimpleState)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_b", END)
+    return builder.compile(transformers=[_CountingTransformer])
+
+
+class TestCompileTimeTransformerRegistration:
+    def test_compile_time_transformer_runs_without_caller_opt_in(self) -> None:
+        graph = _build_counting_graph()
+        run = graph.stream_v2({"value": "", "items": []})
+
+        # Registered at compile time — no call-site `transformers=` needed.
+        counting = run._mux.transformer_by_key("counts")
+        assert isinstance(counting, _CountingTransformer)
+        run.output  # drive to completion
+        assert counting.count >= 1
+
+    @pytest.mark.anyio
+    async def test_compile_time_transformer_runs_on_astream_v2(self) -> None:
+        graph = _build_counting_graph()
+        run = await graph.astream_v2({"value": "", "items": []})
+
+        counting = run._mux.transformer_by_key("counts")
+        assert isinstance(counting, _CountingTransformer)
+        await run.output()
+        assert counting.count >= 1
+
+    def test_call_site_transformers_appended_after_compile_time(self) -> None:
+        """Call-site `transformers=` run after compile-time factories."""
+
+        class _UserTransformer(StreamTransformer):
+            required_stream_modes = ()
+
+            def __init__(self, scope: tuple[str, ...] = ()) -> None:
+                super().__init__(scope)
+
+            def init(self) -> dict[str, Any]:
+                return {"user_flag": EventLog()}
+
+            def process(self, event: ProtocolEvent) -> bool:
+                return True
+
+        graph = _build_counting_graph()
+        run = graph.stream_v2(
+            {"value": "", "items": []}, transformers=[_UserTransformer]
+        )
+
+        # Both compile-time and call-site projections are available.
+        assert "counts" in run.extensions
+        assert "user_flag" in run.extensions
+        run.output  # drain

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -1,258 +0,0 @@
-"""Tests for LifecycleTransformer — derives subgraph lifecycle from ns discovery."""
-
-from __future__ import annotations
-
-import operator
-import time
-from typing import Annotated, Any
-
-import pytest
-from typing_extensions import TypedDict
-
-from langgraph.constants import END, START
-from langgraph.errors import GraphInterrupt
-from langgraph.graph import StateGraph
-from langgraph.stream._mux import StreamMux
-from langgraph.stream._types import ProtocolEvent
-from langgraph.stream.transformers import (
-    LifecycleTransformer,
-    ValuesTransformer,
-)
-
-TS = int(time.time() * 1000)
-
-
-def _event(method: str, data: Any, *, namespace: list[str]) -> ProtocolEvent:
-    return {
-        "type": "event",
-        "method": method,
-        "params": {
-            "namespace": namespace,
-            "timestamp": TS,
-            "data": data,
-        },
-    }
-
-
-def _drain_channel(mux: StreamMux) -> list[dict[str, Any]]:
-    ch = mux.extensions["lifecycle"]
-    return list(ch._log._items)  # type: ignore[attr-defined]
-
-
-def _drain_main_events(mux: StreamMux) -> list[ProtocolEvent]:
-    return list(mux._events._items)  # type: ignore[attr-defined]
-
-
-def _subscribe(mux: StreamMux) -> None:
-    ch = mux.extensions["lifecycle"]
-    ch._log._subscribed = True  # type: ignore[attr-defined]
-    mux._events._subscribed = True  # type: ignore[attr-defined]
-
-
-class TestLifecycleTransformerUnit:
-    def _mux(self) -> StreamMux:
-        mux = StreamMux(
-            factories=[ValuesTransformer, LifecycleTransformer], is_async=False
-        )
-        _subscribe(mux)
-        return mux
-
-    def test_first_event_at_child_ns_emits_started(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["child:task_a"]))
-
-        events = _drain_channel(mux)
-        assert events == [
-            {
-                "event": "started",
-                "namespace": ["child:task_a"],
-                "graph_name": "child",
-                "trigger_call_id": "task_a",
-            }
-        ]
-
-    def test_no_started_at_root_ns(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=[]))
-        assert _drain_channel(mux) == []
-
-    def test_method_agnostic_discovery(self) -> None:
-        mux = self._mux()
-        mux.push(_event("messages", "x", namespace=["c:t"]))
-
-        (started,) = _drain_channel(mux)
-        assert started["event"] == "started"
-        assert started["namespace"] == ["c:t"]
-
-    def test_repeated_events_single_started(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-        mux.push(_event("values", {"v": 2}, namespace=["c:t"]))
-        mux.push(_event("updates", {"n": "x"}, namespace=["c:t"]))
-
-        events = _drain_channel(mux)
-        assert len(events) == 1
-        assert events[0]["event"] == "started"
-
-    def test_ns_without_task_id(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["child"]))
-
-        (started,) = _drain_channel(mux)
-        assert started["graph_name"] == "child"
-        assert "trigger_call_id" not in started
-
-    def test_finalize_emits_completed(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-        mux.close()
-
-        events = _drain_channel(mux)
-        assert events == [
-            {
-                "event": "started",
-                "namespace": ["c:t"],
-                "graph_name": "c",
-                "trigger_call_id": "t",
-            },
-            {"event": "completed", "namespace": ["c:t"]},
-        ]
-
-    def test_fail_with_graph_interrupt(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-        mux.fail(GraphInterrupt())
-
-        events = _drain_channel(mux)
-        assert events[-1] == {"event": "interrupted", "namespace": ["c:t"]}
-
-    def test_fail_with_generic_error(self) -> None:
-        mux = self._mux()
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-        mux.fail(RuntimeError("boom"))
-
-        events = _drain_channel(mux)
-        assert events[-1] == {
-            "event": "failed",
-            "namespace": ["c:t"],
-            "error": "boom",
-        }
-
-
-class TestLifecycleWireFormat:
-    """Native transformer: method on the wire is `"lifecycle"`, no `custom:` prefix."""
-
-    def test_wire_method_is_lifecycle_unprefixed(self) -> None:
-        mux = StreamMux(factories=[LifecycleTransformer], is_async=False)
-        _subscribe(mux)
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-
-        # Find the lifecycle event in the main log.
-        lifecycle_events = [
-            ev for ev in _drain_main_events(mux) if ev["method"] == "lifecycle"
-        ]
-        assert len(lifecycle_events) == 1
-        assert lifecycle_events[0]["params"]["data"]["event"] == "started"
-        # Also verify no `custom:lifecycle` leaks through.
-        assert not any(
-            ev["method"].startswith("custom:") for ev in _drain_main_events(mux)
-        )
-
-    def test_started_precedes_originating_event_on_wire(self) -> None:
-        """Seq ordering: synthesized lifecycle event lands before the event that triggered it."""
-        mux = StreamMux(
-            factories=[ValuesTransformer, LifecycleTransformer], is_async=False
-        )
-        _subscribe(mux)
-        mux.push(_event("values", {"v": 1}, namespace=["c:t"]))
-
-        wire = _drain_main_events(mux)
-        methods = [ev["method"] for ev in wire]
-        # Lifecycle's synthetic event is forwarded during process() and
-        # gets an earlier seq than the originating values event.
-        assert methods.index("lifecycle") < methods.index("values")
-
-
-# ---------------------------------------------------------------------------
-# End-to-end via stream_v2
-# ---------------------------------------------------------------------------
-
-
-class SimpleState(TypedDict):
-    value: str
-    items: Annotated[list[str], operator.add]
-
-
-def _build_nested_graph():
-    def inner_node(state: SimpleState) -> dict:
-        return {"value": state["value"] + "X", "items": ["x"]}
-
-    inner_builder = StateGraph(SimpleState)
-    inner_builder.add_node("inner_node", inner_node)
-    inner_builder.add_edge(START, "inner_node")
-    inner_builder.add_edge("inner_node", END)
-    inner = inner_builder.compile()
-
-    outer_builder = StateGraph(SimpleState)
-    outer_builder.add_node("sub", inner)
-    outer_builder.add_edge(START, "sub")
-    outer_builder.add_edge("sub", END)
-    return outer_builder.compile()
-
-
-class TestLifecycleEndToEnd:
-    def test_real_run_emits_started_and_completed(self) -> None:
-        graph = _build_nested_graph()
-        run = graph.stream_v2(
-            {"value": "", "items": []}, transformers=[LifecycleTransformer]
-        )
-
-        # Drain the run so finalize fires.
-        list(run.values)
-
-        lifecycle = list(run.lifecycle)  # type: ignore[attr-defined]
-        events = [e["event"] for e in lifecycle]
-        assert "started" in events
-        assert "completed" in events
-
-    def test_real_run_error_emits_failed(self) -> None:
-        def boom(state: SimpleState) -> dict:
-            raise RuntimeError("kaboom")
-
-        inner_builder = StateGraph(SimpleState)
-        inner_builder.add_node("inner", boom)
-        inner_builder.add_edge(START, "inner")
-        inner_builder.add_edge("inner", END)
-        inner = inner_builder.compile()
-
-        outer_builder = StateGraph(SimpleState)
-        outer_builder.add_node("sub", inner)
-        outer_builder.add_edge(START, "sub")
-        outer_builder.add_edge("sub", END)
-        graph = outer_builder.compile()
-
-        run = graph.stream_v2(
-            {"value": "", "items": []}, transformers=[LifecycleTransformer]
-        )
-        with pytest.raises(RuntimeError):
-            list(run.values)
-
-        lifecycle = list(run.lifecycle)  # type: ignore[attr-defined]
-        events = [e["event"] for e in lifecycle]
-        assert "failed" in events
-
-    def test_lifecycle_and_subgraphs_agree(self) -> None:
-        """SubgraphTransformer and LifecycleTransformer share the discovery predicate."""
-        graph = _build_nested_graph()
-        run = graph.stream_v2(
-            {"value": "", "items": []}, transformers=[LifecycleTransformer]
-        )
-
-        subs = list(run.subgraphs)
-        lifecycle = list(run.lifecycle)  # type: ignore[attr-defined]
-
-        sub_paths = {tuple(s.path) for s in subs}
-        started_paths = {
-            tuple(e["namespace"]) for e in lifecycle if e["event"] == "started"
-        }
-        assert sub_paths == started_paths

libs/langgraph/tests/test_stream_messages_transformer.py

@@ -183,7 +183,7 @@ class TestProtocolEventRouting:
         log.close()
         (stream,) = list(log._items)
         assert stream.done
-        assert stream.output.text == "hello world"
+        assert stream.output.content == "hello world"
 
     def test_message_finish_cleans_up_routing(self) -> None:
         t, log = _make_sync_transformer()
@@ -220,8 +220,8 @@ class TestProtocolEventRouting:
         streams = list(log._items)
         assert len(streams) == 2
         by_id = {s.message_id: s for s in streams}
-        assert by_id["run-a"].output.text == "aaaa"
-        assert by_id["run-b"].output.text == "bbbb"
+        assert by_id["run-a"].output.content == "aaaa"
+        assert by_id["run-b"].output.content == "bbbb"
 
     def test_text_deltas_accumulated_on_stream(self) -> None:
         t, log = _make_sync_transformer()
@@ -270,7 +270,7 @@ class TestWholeMessageFallback:
         log.close()
         (stream,) = list(log._items)
         assert stream.done
-        assert stream.output.text == "the full answer"
+        assert stream.output.content == "the full answer"
 
     def test_whole_message_has_full_lifecycle(self) -> None:
         t, log = _make_sync_transformer()
@@ -410,7 +410,7 @@ class TestAsyncMode:
             t.process(_proto_event(evt))
         (stream,) = list(log._items)
         msg = await stream.output
-        assert msg.text == "async"
+        assert msg.content == "async"
 
 
 # ---------------------------------------------------------------------------
@@ -471,7 +471,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.text == "mux stream"
+        assert stream.output.content == "mux stream"
 
     def test_whole_message_via_mux(self) -> None:
         t = MessagesTransformer()
@@ -485,7 +485,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.text == "result"
+        assert stream.output.content == "result"
 
     @pytest.mark.anyio
     async def test_async_streaming_via_mux(self) -> None:
@@ -501,7 +501,7 @@ class TestViaMux:
         streams = list(log._items)
         assert len(streams) == 1
         msg = await streams[0].output
-        assert msg.text == "async mux"
+        assert msg.content == "async mux"
         await mux.aclose()
 
 
@@ -545,7 +545,7 @@ class TestEndToEnd:
 
         assert len(streams) == 1
         assert isinstance(streams[0], ChatModelStream)
-        assert streams[0].output.text == "hello world"
+        assert streams[0].output.content == "hello world"
 
     def test_node_stream_v2_text_deltas_iterate(self) -> None:
         """Consumer can iterate `.text` on the streamed message in real time."""
@@ -588,7 +588,7 @@ class TestEndToEnd:
         streams = list(run.messages)
 
         assert len(streams) == 1
-        assert streams[0].output.text == "hardcoded"
+        assert streams[0].output.content == "hardcoded"
 
     @pytest.mark.anyio
     async def test_async_node_calling_astream_v2(self) -> None:
@@ -616,7 +616,7 @@ class TestEndToEnd:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.text == "async answer"
+        assert msg.content == "async answer"
 
     @pytest.mark.anyio
     async def test_nested_async_iteration_yields_text_deltas(self) -> None:
@@ -693,7 +693,7 @@ class TestEndToEndV2Invoke:
         )
         stream = streams[0]
         assert isinstance(stream, ChatModelStream)
-        assert stream.output.text == "hello world"
+        assert stream.output.content == "hello world"
 
     def test_invoke_v2_emits_protocol_events(self) -> None:
         """Iterating the stream yields the full v2 lifecycle (not v1 chunks)."""
@@ -726,7 +726,7 @@ class TestEndToEndV2Invoke:
             assert isinstance(event, dict)
             assert "event" in event
         # Typed projection still assembles the final text.
-        assert stream.output.text == "streamed answer"
+        assert stream.output.content == "streamed answer"
 
     def test_invoke_text_deltas_iterate_live(self) -> None:
         """`.text` projection yields deltas in order."""
@@ -774,7 +774,7 @@ class TestEndToEndV2Invoke:
         streams = list(run.messages)
 
         assert len(streams) == 2
-        contents = {s.output.text for s in streams}
+        contents = {s.output.content for s in streams}
         assert contents == {"alpha", "beta"}
 
     def test_invoke_plus_constructed_message_two_streams(self) -> None:
@@ -805,9 +805,9 @@ class TestEndToEndV2Invoke:
 
         assert len(streams) == 2
         assert streams[0].node == "streaming_node"
-        assert streams[0].output.text == "live stream"
+        assert streams[0].output.content == "live stream"
         assert streams[1].node == "constructed_node"
-        assert streams[1].output.text == "hardcoded"
+        assert streams[1].output.content == "hardcoded"
         assert streams[1].message_id == "constructed-1"
 
     @pytest.mark.anyio
@@ -835,7 +835,7 @@ class TestEndToEndV2Invoke:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.text == "async invoke"
+        assert msg.content == "async invoke"
 
 
 class TestDirectMessagesModeStaysV1:

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -1,4 +1,4 @@
-"""Tests for SubgraphTransformer namespace-based discovery."""
+"""Tests for subgraph lifecycle events and the SubgraphTransformer."""
 
 from __future__ import annotations
 
@@ -27,6 +27,32 @@ from langgraph.types import interrupt
 TS = int(time.time() * 1000)
 
 
+def _lifecycle(
+    event: str,
+    *,
+    namespace: list[str] | None = None,
+    graph_name: str | None = None,
+    trigger_call_id: str | None = None,
+    error: str | None = None,
+) -> ProtocolEvent:
+    data: dict[str, Any] = {"event": event}
+    if graph_name is not None:
+        data["graph_name"] = graph_name
+    if trigger_call_id is not None:
+        data["trigger_call_id"] = trigger_call_id
+    if error is not None:
+        data["error"] = error
+    return {
+        "type": "event",
+        "method": "lifecycle",
+        "params": {
+            "namespace": namespace or [],
+            "timestamp": TS,
+            "data": data,
+        },
+    }
+
+
 def _values(payload: dict[str, Any], *, namespace: list[str]) -> ProtocolEvent:
     return {
         "type": "event",
@@ -82,90 +108,96 @@ class TestSubgraphTransformerUnit:
         return mux, transformer
 
     def _handle(self, transformer: SubgraphTransformer) -> SubgraphRunStream:
+        """Return the single root handle after pushing one lifecycle started."""
         (handle,) = list(transformer._root_log._items)
         return handle
 
-    def test_root_event_does_not_create_handle(self) -> None:
+    def test_root_started_is_ignored(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=[]))
+        mux.push(_lifecycle("started", graph_name="root"))
         assert list(transformer._root_log._items) == []
         assert transformer._by_ns == {}
 
-    def test_first_event_at_child_depth_yields_handle(self) -> None:
+    def test_child_started_yields_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["child:task_a"]))
+        mux.push(
+            _lifecycle(
+                "started",
+                namespace=["task_a:child"],
+                graph_name="child",
+                trigger_call_id="task_a",
+            )
+        )
 
         handle = self._handle(transformer)
-        assert handle.path == ("child:task_a",)
+        assert handle.path == ("task_a:child",)
         assert handle.graph_name == "child"
         assert handle.trigger_call_id == "task_a"
         assert handle.status == "started"
 
-    def test_handle_without_task_id_suffix(self) -> None:
+    def test_status_transitions(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["child"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_lifecycle("running", namespace=["t:c"]))
+        mux.push(_lifecycle("completed", namespace=["t:c"]))
 
         handle = self._handle(transformer)
-        assert handle.path == ("child",)
-        assert handle.graph_name == "child"
-        assert handle.trigger_call_id is None
-
-    def test_discovery_is_method_agnostic(self) -> None:
-        mux, transformer = self._mux()
-        # Method-agnostic means "any event method triggers discovery".
-        # Use `updates` — neither ValuesTransformer nor
-        # MessagesTransformer care about it, so the test only exercises
-        # SubgraphTransformer's discovery path.
-        mux.push(
-            {
-                "type": "event",
-                "method": "updates",
-                "params": {"namespace": ["c:t"], "timestamp": TS, "data": "x"},
-            }
-        )
-        handle = self._handle(transformer)
-        assert handle.path == ("c:t",)
+        assert handle.status == "completed"
 
     def test_grandchild_surfaces_under_child(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["child:t"]))
+        mux.push(_lifecycle("started", namespace=["t:child"], graph_name="child"))
 
         child = self._handle(transformer)
         _pre_subscribe_handle(child)
 
-        mux.push(_values({"value": 2}, namespace=["child:t", "grand:u"]))
+        mux.push(
+            _lifecycle(
+                "started",
+                namespace=["t:child", "u:grand"],
+                graph_name="grand",
+            )
+        )
 
         (grand,) = _handle_subgraphs_items(child)
-        assert grand.path == ("child:t", "grand:u")
+        assert grand.path == ("t:child", "u:grand")
         assert grand.graph_name == "grand"
-        assert grand.trigger_call_id == "u"
+
+    def test_failed_stores_error(self) -> None:
+        mux, transformer = self._mux()
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_lifecycle("failed", namespace=["t:c"], error="boom"))
+
+        handle = self._handle(transformer)
+        assert handle.status == "failed"
+        assert handle.error == "boom"
 
     def test_values_routed_into_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
 
         handle = self._handle(transformer)
         _pre_subscribe_handle(handle)
 
-        mux.push(_values({"value": 2}, namespace=["c:t"]))
+        mux.push(_values({"value": 1}, namespace=["t:c"]))
+        mux.push(_values({"value": 2}, namespace=["t:c"]))
 
-        # Both the discovery event and subsequent values land in the child.
         assert _handle_values_items(handle) == [{"value": 1}, {"value": 2}]
         assert handle.output == {"value": 2}
 
     def test_root_values_not_routed(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
         handle = self._handle(transformer)
         _pre_subscribe_handle(handle)
 
         # Values event at root namespace — must not leak into child handle.
         mux.push(_values({"value": "root"}, namespace=[]))
-        assert _handle_values_items(handle) == [{"value": 1}]
+        assert _handle_values_items(handle) == []
 
     def test_finalize_closes_dangling(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
         handle = self._handle(transformer)
 
         mux.close()
@@ -175,7 +207,7 @@ class TestSubgraphTransformerUnit:
 
     def test_fail_with_graph_interrupt_marks_interrupted(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
         handle = self._handle(transformer)
 
         mux.fail(GraphInterrupt())
@@ -183,22 +215,32 @@ class TestSubgraphTransformerUnit:
 
     def test_fail_with_generic_error_marks_failed(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
         handle = self._handle(transformer)
 
         mux.fail(RuntimeError("explode"))
         assert handle.status == "failed"
         assert handle.error == "explode"
 
-    def test_repeated_events_same_ns_single_handle(self) -> None:
+    def test_duplicate_started_ignored(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_values({"value": 1}, namespace=["c:t"]))
-        mux.push(_values({"value": 2}, namespace=["c:t"]))
-        mux.push(_values({"value": 3}, namespace=["c:t"]))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="other"))
 
         handles = list(transformer._root_log._items)
         assert len(handles) == 1
-        assert handles[0].path == ("c:t",)
+        assert handles[0].graph_name == "c"
+
+    def test_non_lifecycle_non_values_passthrough(self) -> None:
+        mux, transformer = self._mux()
+        mux.push(
+            {
+                "type": "event",
+                "method": "messages",
+                "params": {"namespace": ["t:c"], "timestamp": TS, "data": "x"},
+            }
+        )
+        assert list(transformer._root_log._items) == []
 
 
 # ---------------------------------------------------------------------------

libs/langgraph/tests/test_tool_stream_handler.py

@@ -0,0 +1,290 @@
+"""Tests for StreamToolCallHandler and emit_tool_output_delta.
+
+These tests exercise the langgraph-core piece in isolation — the prebuilt
+`ToolCallTransformer` has its own test file. Here we feed real graphs
+through `Pregel.stream(stream_mode=["tools", ...])` and inspect the raw
+`(ns, mode, payload)` tuples on the `tools` channel.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage
+from langchain_core.tools import tool
+from langgraph.prebuilt import ToolNode
+from typing_extensions import TypedDict
+
+from langgraph.config import emit_tool_output_delta
+from langgraph.constants import END, START
+from langgraph.graph import StateGraph
+from langgraph.graph.message import add_messages
+
+
+class _State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+def _caller_sync(tool_name: str, tool_args: dict[str, Any], tc_id: str = "tc1"):
+    def caller(state: _State) -> dict:
+        return {
+            "messages": [
+                AIMessage(
+                    content="",
+                    tool_calls=[{"name": tool_name, "args": tool_args, "id": tc_id}],
+                )
+            ]
+        }
+
+    return caller
+
+
+def _caller_async(tool_name: str, tool_args: dict[str, Any], tc_id: str = "tc1"):
+    async def caller(state: _State) -> dict:
+        return {
+            "messages": [
+                AIMessage(
+                    content="",
+                    tool_calls=[{"name": tool_name, "args": tool_args, "id": tc_id}],
+                )
+            ]
+        }
+
+    return caller
+
+
+def _build_graph(caller, tools) -> Any:
+    sg = StateGraph(_State)
+    sg.add_node("caller", caller)
+    sg.add_node("tools", ToolNode(tools))
+    sg.add_edge(START, "caller")
+    sg.add_edge("caller", "tools")
+    sg.add_edge("tools", END)
+    return sg.compile()
+
+
+def _tool_events(stream) -> list[tuple[tuple[str, ...], dict]]:
+    """Collect `(ns, payload)` for every `tools`-mode chunk."""
+    out: list[tuple[tuple[str, ...], dict]] = []
+    for ns, mode, payload in stream:
+        if mode == "tools":
+            out.append((tuple(ns), payload))
+    return out
+
+
+class TestSyncGraphSyncTool:
+    def test_started_finished_cycle(self) -> None:
+        @tool
+        def echo(text: str) -> str:
+            """echo."""
+            return f"echoed:{text}"
+
+        graph = _build_graph(_caller_sync("echo", {"text": "hi"}), [echo])
+        events = _tool_events(
+            graph.stream(
+                {"messages": []},
+                stream_mode=["tools"],
+                subgraphs=True,
+            )
+        )
+
+        assert [p["event"] for _, p in events] == [
+            "tool-started",
+            "tool-finished",
+        ]
+        assert events[0][1]["tool_call_id"] == "tc1"
+        assert events[0][1]["tool_name"] == "echo"
+        assert events[0][1]["input"] == {"text": "hi"}
+        # ToolNode wraps the return in a ToolMessage.
+        assert events[1][1]["tool_call_id"] == "tc1"
+
+    def test_emit_tool_output_delta_produces_delta_events(self) -> None:
+        @tool
+        def streaming_echo(text: str) -> str:
+            """stream chunks."""
+            for chunk in ("a", "b", "c"):
+                emit_tool_output_delta(chunk)
+            return text
+
+        graph = _build_graph(
+            _caller_sync("streaming_echo", {"text": "x"}), [streaming_echo]
+        )
+        events = _tool_events(
+            graph.stream(
+                {"messages": []},
+                stream_mode=["tools"],
+                subgraphs=True,
+            )
+        )
+
+        deltas = [p["delta"] for _, p in events if p["event"] == "tool-output-delta"]
+        assert deltas == ["a", "b", "c"]
+        # The deltas must be bracketed by started and finished.
+        ordered = [p["event"] for _, p in events]
+        assert ordered[0] == "tool-started"
+        assert ordered[-1] == "tool-finished"
+
+    def test_tool_error_event(self) -> None:
+        @tool
+        def boom() -> str:
+            """raises."""
+            raise ValueError("nope")
+
+        graph = _build_graph(_caller_sync("boom", {}), [boom])
+        events: list[tuple[tuple[str, ...], dict]] = []
+        with pytest.raises(ValueError, match="nope"):
+            for ns, mode, payload in graph.stream(
+                {"messages": []},
+                stream_mode=["tools"],
+                subgraphs=True,
+            ):
+                if mode == "tools":
+                    events.append((tuple(ns), payload))
+
+        kinds = [p["event"] for _, p in events]
+        assert kinds == ["tool-started", "tool-error"]
+        assert events[1][1]["message"] == "nope"
+
+    def test_emit_outside_tool_is_noop(self) -> None:
+        # Called at import time (outside any tool body) — must not raise.
+        emit_tool_output_delta("ignored")
+        emit_tool_output_delta({"any": "payload"})
+
+    def test_no_events_without_tools_mode(self) -> None:
+        @tool
+        def echo(text: str) -> str:
+            """echo."""
+            return text
+
+        graph = _build_graph(_caller_sync("echo", {"text": "hi"}), [echo])
+        # No "tools" in stream_mode — handler is not attached and zero
+        # `tools`-method events fire.
+        chunks = list(
+            graph.stream(
+                {"messages": []},
+                stream_mode=["values"],
+                subgraphs=True,
+            )
+        )
+        assert all(
+            not (isinstance(c, tuple) and len(c) == 3 and c[1] == "tools")
+            for c in chunks
+        )
+
+
+class TestAsyncGraphAsyncTool:
+    @pytest.mark.anyio
+    async def test_async_tool_produces_events(self) -> None:
+        @tool
+        async def aecho(text: str) -> str:
+            """async echo."""
+            emit_tool_output_delta(text)
+            return f"got:{text}"
+
+        graph = _build_graph(_caller_async("aecho", {"text": "hi"}), [aecho])
+        events: list[tuple[tuple[str, ...], dict]] = []
+        async for ns, mode, payload in graph.astream(
+            {"messages": []},
+            stream_mode=["tools"],
+            subgraphs=True,
+        ):
+            if mode == "tools":
+                events.append((tuple(ns), payload))
+
+        kinds = [p["event"] for _, p in events]
+        assert kinds == ["tool-started", "tool-output-delta", "tool-finished"]
+        assert events[1][1]["delta"] == "hi"
+
+
+class TestConcurrentToolCalls:
+    def test_parallel_tool_calls_do_not_bleed(self) -> None:
+        @tool
+        def streamer(marker: str) -> str:
+            """emits marker twice."""
+            emit_tool_output_delta(f"{marker}-1")
+            emit_tool_output_delta(f"{marker}-2")
+            return marker
+
+        def caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {"name": "streamer", "args": {"marker": "A"}, "id": "a"},
+                            {"name": "streamer", "args": {"marker": "B"}, "id": "b"},
+                        ],
+                    )
+                ]
+            }
+
+        graph = _build_graph(caller, [streamer])
+        events = _tool_events(
+            graph.stream(
+                {"messages": []},
+                stream_mode=["tools"],
+                subgraphs=True,
+            )
+        )
+
+        # Group deltas by tool_call_id.
+        by_id: dict[str, list[str]] = {}
+        for _, p in events:
+            if p["event"] == "tool-output-delta":
+                by_id.setdefault(p["tool_call_id"], []).append(p["delta"])
+        assert by_id["a"] == ["A-1", "A-2"]
+        assert by_id["b"] == ["B-1", "B-2"]
+
+
+class TestSubgraphNamespacePropagation:
+    def test_tool_inside_subgraph_emits_with_subgraph_ns(self) -> None:
+        @tool
+        def inner_tool(text: str) -> str:
+            """inner tool."""
+            return text
+
+        def sub_caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {
+                                "name": "inner_tool",
+                                "args": {"text": "x"},
+                                "id": "tc1",
+                            }
+                        ],
+                    )
+                ]
+            }
+
+        inner = StateGraph(_State)
+        inner.add_node("sub_caller", sub_caller)
+        inner.add_node("sub_tools", ToolNode([inner_tool]))
+        inner.add_edge(START, "sub_caller")
+        inner.add_edge("sub_caller", "sub_tools")
+        inner.add_edge("sub_tools", END)
+        inner_graph = inner.compile()
+
+        outer = StateGraph(_State)
+        outer.add_node("sub", inner_graph)
+        outer.add_edge(START, "sub")
+        outer.add_edge("sub", END)
+        graph = outer.compile()
+
+        events = _tool_events(
+            graph.stream(
+                {"messages": []},
+                stream_mode=["tools"],
+                subgraphs=True,
+            )
+        )
+
+        # All `tools` events should carry a non-empty namespace rooted
+        # at the `sub` node.
+        assert events, "expected at least one tools event"
+        for ns, _ in events:
+            assert ns  # non-empty
+            assert ns[0].startswith("sub:")

libs/langgraph/uv.lock

@@ -1348,11 +1348,10 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.3.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
-    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -1361,21 +1360,9 @@ 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/af/bc/0bff31fcaff174d86031cc713471a3e85ed4ec8e5cd95ad0217f2aced20e/langchain_core-1.3.0a2.tar.gz", hash = "sha256:52d978c84552b74b9a3f16c1fced84f9e27cc96d7a67c601925ce6cbc4ea3cf9", size = 854580, upload-time = "2026-04-13T14:37:55.745Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.11"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/d0/bb/38b5eaefa41c67735eedd9f9a2568b11c9eb376fa129a5edd7cc3dcde071/langchain_protocol-0.0.11.tar.gz", hash = "sha256:c276e2373b5ac691fc7ac9a72019d55182444ce8e89385c3f7e9f0185d0aace7", size = 6622, upload-time = "2026-04-23T22:13:16.771Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/0b/fa/6a8ecad8472b182f2caf9d83fd89f40fc1590cb96546d90089b7869b7f5e/langchain_protocol-0.0.11-py3-none-any.whl", hash = "sha256:364da1faf6f5d3001413bede792c1a822c0f23ae55d1ce1266ca7d8e80e79011", size = 6778, upload-time = "2026-04-23T22:13:15.712Z" },
+    { url = "https://files.pythonhosted.org/packages/0e/14/03c09686602567059f26af29de0c44546a83af2f2aa29925e61040e43ea2/langchain_core-1.3.0a2-py3-none-any.whl", hash = "sha256:9e929a34f0b0c6c1255e395a1de34f8626893ceb4cdae550a22a0bd18c87be54", size = 510233, upload-time = "2026-04-13T14:37:54.277Z" },
 ]
 
 [[package]]
@@ -1452,7 +1439,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.2" },
+    { name = "langchain-core", specifier = "==1.3.0a2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },

libs/prebuilt/langgraph/prebuilt/__init__.py

@@ -1,5 +1,7 @@
 """langgraph.prebuilt exposes a higher-level API for creating and executing agents and tools."""
 
+from langgraph.prebuilt._tool_call_stream import ToolCallStream
+from langgraph.prebuilt._tool_call_transformer import ToolCallTransformer
 from langgraph.prebuilt.chat_agent_executor import create_react_agent
 from langgraph.prebuilt.tool_node import (
     InjectedState,
@@ -13,6 +15,8 @@ from langgraph.prebuilt.tool_validator import ValidationNode
 __all__ = [
     "create_react_agent",
     "ToolNode",
+    "ToolCallStream",
+    "ToolCallTransformer",
     "tools_condition",
     "ValidationNode",
     "InjectedState",

libs/prebuilt/langgraph/prebuilt/_tool_call_stream.py

@@ -0,0 +1,117 @@
+"""In-process handle for a single tool call's streaming execution.
+
+Mirrors the shape of `ChatModelStream` from langchain-core but simpler —
+a tool has one output channel, no content-block multiplexing. Populated
+by `ToolCallTransformer` as `tool-started` / `tool-output-delta` /
+`tool-finished` / `tool-error` events flow in on the `tools` channel.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Iterator
+from typing import Any
+
+from langgraph.stream._event_log import EventLog
+
+
+class ToolCallStream:
+    """Scoped view of a single tool call's lifecycle.
+
+    Yielded on `run.tool_calls` once per `tool-started` event. Fields
+    are populated as events arrive:
+
+    - `tool_call_id`, `tool_name`, `input`: stable from the start event.
+    - `output_deltas`: an `EventLog` of delta chunks. Iterate (sync or
+      async) to consume partial output in arrival order.
+    - `output`: terminal payload from `tool-finished`, or `None` if the
+      call failed or is still in flight.
+    - `error`: terminal error string from `tool-error`, or `None` if the
+      call succeeded or is still in flight.
+    - `completed`: True once a terminal event (`tool-finished` or
+      `tool-error`) has been observed.
+
+    `ToolCallStream` is not meant to be constructed by end users — it's
+    produced by `ToolCallTransformer` as events flow through the mux.
+    """
+
+    def __init__(
+        self,
+        tool_call_id: str,
+        tool_name: str,
+        input: dict[str, Any] | None = None,
+    ) -> None:
+        """Initialize a fresh handle for a tool call.
+
+        Args:
+            tool_call_id: The `tool_call_id` from the AIMessage.
+            tool_name: The tool's name.
+            input: The tool's input arguments (as reported by
+                `on_tool_start`), or `None` if none were captured.
+        """
+        self.tool_call_id = tool_call_id
+        self.tool_name = tool_name
+        self.input = input
+        self._output_deltas: EventLog[Any] = EventLog()
+        self.output: Any = None
+        self.error: str | None = None
+        self.completed = False
+
+    @property
+    def output_deltas(self) -> EventLog[Any]:
+        """The EventLog of streamed `tool-output-delta` payloads.
+
+        Iterate (sync or async depending on how the run was started)
+        to consume partial output in arrival order. The log closes when
+        the tool finishes or errors.
+        """
+        return self._output_deltas
+
+    def _bind(self, *, is_async: bool) -> None:
+        """Bind the deltas log to sync or async iteration.
+
+        Called by `ToolCallTransformer` when constructing this handle so
+        the log matches the enclosing mux's mode.
+        """
+        self._output_deltas._bind(is_async=is_async)
+
+    def _push_delta(self, delta: Any) -> None:
+        self._output_deltas.push(delta)
+
+    def _finish(self, output: Any) -> None:
+        self.output = output
+        self.completed = True
+        self._output_deltas.close()
+
+    def _fail(self, message: str) -> None:
+        self.error = message
+        self.completed = True
+        self._output_deltas.close()
+
+    def __iter__(self) -> Iterator[Any]:
+        """Iterate delta chunks synchronously.
+
+        Equivalent to `iter(self.output_deltas)`. Raises `TypeError` if
+        the underlying log is bound to async mode.
+        """
+        return iter(self._output_deltas)
+
+    def __aiter__(self) -> AsyncIterator[Any]:
+        """Iterate delta chunks asynchronously.
+
+        Equivalent to `aiter(self.output_deltas)`. Raises `TypeError`
+        if the underlying log is bound to sync mode.
+        """
+        return self._output_deltas.__aiter__()
+
+    def __repr__(self) -> str:
+        status = (
+            "completed"
+            if self.completed and self.error is None
+            else "failed"
+            if self.completed
+            else "running"
+        )
+        return (
+            f"ToolCallStream(tool_call_id={self.tool_call_id!r}, "
+            f"tool_name={self.tool_name!r}, status={status})"
+        )

libs/prebuilt/langgraph/prebuilt/_tool_call_transformer.py

@@ -0,0 +1,128 @@
+"""Transformer that projects `tools` channel events into `ToolCallStream`s."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._types import ProtocolEvent, StreamTransformer
+
+from langgraph.prebuilt._tool_call_stream import ToolCallStream
+
+
+class ToolCallTransformer(StreamTransformer):
+    """Project `tools` channel events into `ToolCallStream` handles.
+
+    Each `tool-started` event spawns a `ToolCallStream`, pushed onto
+    `run.tool_calls`. Subsequent `tool-output-delta` events append to
+    that stream's deltas log; `tool-finished` and `tool-error` close it.
+
+    Native transformer — the `tool_calls` projection is exposed as a
+    direct attribute on the run stream.
+
+    `EventLog[ToolCallStream]` is used (not `StreamChannel`) because the
+    live handles are not serializable and should not be auto-forwarded
+    onto the main event log. Wire consumers subscribe to the `tools`
+    channel instead, where the raw protocol events flow through
+    untouched by this transformer (`process` returns `True`).
+
+    Registered explicitly by users at compile time via
+    `builder.compile(transformers=[ToolCallTransformer])` — not a
+    default built-in, so the `tools` channel is user-opt-in.
+    """
+
+    _native = True
+    required_stream_modes = ("tools",)
+
+    def __init__(self, scope: tuple[str, ...] = ()) -> None:
+        super().__init__(scope)
+        self._log: EventLog[ToolCallStream] = EventLog()
+        self._active: dict[str, ToolCallStream] = {}
+        self._is_async = False
+        self._pump_fn: Callable[[], bool] | None = None
+        self._apump_fn: Callable[[], Awaitable[bool]] | None = None
+
+    def init(self) -> dict[str, Any]:
+        return {"tool_calls": self._log}
+
+    def _bind_pump(self, fn: Callable[[], bool]) -> None:
+        """Wire the sync pull callback onto this transformer.
+
+        Called by `StreamMux.bind_pump`. Stored so each new
+        `ToolCallStream` created by `process` can wire its deltas log
+        for pump-driven iteration.
+        """
+        self._pump_fn = fn
+        self._is_async = False
+
+    def _bind_apump(self, fn: Callable[[], Awaitable[bool]]) -> None:
+        """Async counterpart to `_bind_pump`."""
+        self._apump_fn = fn
+        self._is_async = True
+
+    def _new_stream(
+        self,
+        tool_call_id: str,
+        tool_name: str,
+        tool_input: dict[str, Any] | None,
+    ) -> ToolCallStream:
+        stream = ToolCallStream(tool_call_id, tool_name, tool_input)
+        stream._bind(is_async=self._is_async)
+        if self._apump_fn is not None:
+            stream._output_deltas._arequest_more = self._apump_fn
+        if self._pump_fn is not None:
+            stream._output_deltas._request_more = self._pump_fn
+        return stream
+
+    def process(self, event: ProtocolEvent) -> bool:
+        # Namespace filtering is handled by the mux via `scope_exact`.
+        if event["method"] != "tools":
+            return True
+
+        data = event["params"]["data"]
+        tool_call_id = data.get("tool_call_id")
+        if tool_call_id is None:
+            return True
+        event_type = data.get("event")
+
+        stream: ToolCallStream | None
+        if event_type == "tool-started":
+            stream = self._new_stream(
+                tool_call_id,
+                data.get("tool_name", ""),
+                data.get("input"),
+            )
+            self._active[tool_call_id] = stream
+            self._log.push(stream)
+        elif event_type == "tool-output-delta":
+            stream = self._active.get(tool_call_id)
+            if stream is not None:
+                stream._push_delta(data.get("delta"))
+        elif event_type == "tool-finished":
+            stream = self._active.pop(tool_call_id, None)
+            if stream is not None:
+                stream._finish(data.get("output"))
+        elif event_type == "tool-error":
+            stream = self._active.pop(tool_call_id, None)
+            if stream is not None:
+                stream._fail(data.get("message", ""))
+
+        # Pass-through — wire consumers subscribe to the `tools` channel
+        # directly and reconstruct handles client-side.
+        return True
+
+    def finalize(self) -> None:
+        """Close any still-active tool streams left open at run end."""
+        for stream in self._active.values():
+            if not stream.completed:
+                stream._finish(None)
+        self._active.clear()
+
+    def fail(self, err: BaseException) -> None:
+        """Fail any still-active tool streams when the run errors."""
+        message = str(err)
+        for stream in self._active.values():
+            if not stream.completed:
+                stream._fail(message)
+        self._active.clear()

libs/prebuilt/tests/test_tool_call_transformer.py

@@ -0,0 +1,306 @@
+"""Tests for ToolCallTransformer and the ToolCallStream projection."""
+
+from __future__ import annotations
+
+import time
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.messages import AIMessage
+from langchain_core.tools import tool
+from langgraph.config import emit_tool_output_delta
+from langgraph.constants import END, START
+from langgraph.graph import StateGraph
+from langgraph.graph.message import add_messages
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._mux import StreamMux
+from langgraph.stream._types import ProtocolEvent
+from langgraph.stream.transformers import (
+    MessagesTransformer,
+    SubgraphTransformer,
+    ValuesTransformer,
+)
+from typing_extensions import TypedDict
+
+from langgraph.prebuilt import ToolCallStream, ToolCallTransformer, ToolNode
+
+TS = int(time.time() * 1000)
+
+
+def _tool_event(
+    event: str,
+    tool_call_id: str,
+    *,
+    tool_name: str = "",
+    input: dict[str, Any] | None = None,
+    delta: Any = None,
+    output: Any = None,
+    message: str = "",
+    namespace: list[str] | None = None,
+) -> ProtocolEvent:
+    data: dict[str, Any] = {"event": event, "tool_call_id": tool_call_id}
+    if event == "tool-started":
+        data["tool_name"] = tool_name
+        if input is not None:
+            data["input"] = input
+    elif event == "tool-output-delta":
+        data["delta"] = delta
+    elif event == "tool-finished":
+        data["output"] = output
+    elif event == "tool-error":
+        data["message"] = message
+    return {
+        "type": "event",
+        "method": "tools",
+        "params": {
+            "namespace": namespace or [],
+            "timestamp": TS,
+            "data": data,
+        },
+    }
+
+
+def _subscribe(log: EventLog) -> None:
+    log._subscribed = True
+
+
+def _mux() -> tuple[StreamMux, ToolCallTransformer]:
+    mux = StreamMux(
+        factories=[
+            ValuesTransformer,
+            MessagesTransformer,
+            SubgraphTransformer,
+            ToolCallTransformer,
+        ],
+        is_async=False,
+    )
+    transformer = mux.transformer_by_key("tool_calls")
+    assert isinstance(transformer, ToolCallTransformer)
+    _subscribe(transformer._log)
+    return mux, transformer
+
+
+class TestToolCallTransformerUnit:
+    def test_required_stream_modes_declares_tools(self) -> None:
+        assert ToolCallTransformer.required_stream_modes == ("tools",)
+
+    def test_tool_started_yields_handle(self) -> None:
+        mux, transformer = _mux()
+        mux.push(
+            _tool_event(
+                "tool-started",
+                "tc1",
+                tool_name="echo",
+                input={"text": "hi"},
+            )
+        )
+        handles = list(transformer._log._items)
+        assert len(handles) == 1
+        h = handles[0]
+        assert isinstance(h, ToolCallStream)
+        assert h.tool_call_id == "tc1"
+        assert h.tool_name == "echo"
+        assert h.input == {"text": "hi"}
+        assert h.completed is False
+
+    def test_delta_accumulates_on_active_stream(self) -> None:
+        mux, transformer = _mux()
+        mux.push(_tool_event("tool-started", "tc1", tool_name="echo"))
+        _subscribe(transformer._active["tc1"]._output_deltas)
+        mux.push(_tool_event("tool-output-delta", "tc1", delta="a"))
+        mux.push(_tool_event("tool-output-delta", "tc1", delta="b"))
+        stream = transformer._active["tc1"]
+        assert list(stream._output_deltas._items) == ["a", "b"]
+
+    def test_finish_closes_stream(self) -> None:
+        mux, transformer = _mux()
+        mux.push(_tool_event("tool-started", "tc1", tool_name="echo"))
+        stream = transformer._active["tc1"]
+        mux.push(_tool_event("tool-finished", "tc1", output="done"))
+        assert stream.completed is True
+        assert stream.output == "done"
+        assert stream.error is None
+        assert "tc1" not in transformer._active
+
+    def test_error_closes_stream(self) -> None:
+        mux, transformer = _mux()
+        mux.push(_tool_event("tool-started", "tc1", tool_name="boom"))
+        stream = transformer._active["tc1"]
+        mux.push(_tool_event("tool-error", "tc1", message="nope"))
+        assert stream.completed is True
+        assert stream.output is None
+        assert stream.error == "nope"
+        assert "tc1" not in transformer._active
+
+    def test_concurrent_tool_calls_do_not_bleed(self) -> None:
+        mux, transformer = _mux()
+        mux.push(_tool_event("tool-started", "a", tool_name="t"))
+        mux.push(_tool_event("tool-started", "b", tool_name="t"))
+        for tc in ("a", "b"):
+            _subscribe(transformer._active[tc]._output_deltas)
+        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 list(transformer._active["a"]._output_deltas._items) == ["A1", "A2"]
+        assert list(transformer._active["b"]._output_deltas._items) == ["B1"]
+
+    def test_tools_event_passes_through_main_log(self) -> None:
+        mux, transformer = _mux()
+        _subscribe(mux._events)
+        mux.push(_tool_event("tool-started", "tc1", tool_name="echo"))
+        kept = [e for e in mux._events._items if e["method"] == "tools"]
+        assert len(kept) == 1
+
+
+# ---------------------------------------------------------------------------
+# End-to-end tests with a real graph
+# ---------------------------------------------------------------------------
+
+
+class _State(TypedDict):
+    messages: Annotated[list, add_messages]
+
+
+def _build_graph(caller, tools):
+    sg = StateGraph(_State)
+    sg.add_node("caller", caller)
+    sg.add_node("tools", ToolNode(tools))
+    sg.add_edge(START, "caller")
+    sg.add_edge("caller", "tools")
+    sg.add_edge("tools", END)
+    return sg.compile()
+
+
+class TestToolCallTransformerEndToEnd:
+    def test_sync_streaming_tool_populates_tool_calls(self) -> None:
+        @tool
+        def streamer(text: str) -> str:
+            """streams chunks."""
+            for chunk in ("one", "two"):
+                emit_tool_output_delta(chunk)
+            return text
+
+        def caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {"name": "streamer", "args": {"text": "x"}, "id": "tc1"}
+                        ],
+                    )
+                ]
+            }
+
+        graph = _build_graph(caller, [streamer])
+        run = graph.stream_v2({"messages": []}, transformers=[ToolCallTransformer])
+
+        tool_calls: list[ToolCallStream] = []
+        for tc in run.tool_calls:
+            tool_calls.append(tc)
+            deltas = list(tc.output_deltas)
+            assert deltas == ["one", "two"]
+        assert len(tool_calls) == 1
+        tc = tool_calls[0]
+        assert tc.tool_call_id == "tc1"
+        assert tc.tool_name == "streamer"
+        assert tc.completed is True
+        assert tc.error is None
+
+    def test_stream_modes_union_includes_tools(self) -> None:
+        @tool
+        def echo(text: str) -> str:
+            """echo."""
+            return text
+
+        def caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {"name": "echo", "args": {"text": "x"}, "id": "tc1"}
+                        ],
+                    )
+                ]
+            }
+
+        graph = _build_graph(caller, [echo])
+        # 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": []})
+        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])
+        assert "tool_calls" in run._mux.extensions  # type: ignore[attr-defined]
+        # Drain so the run closes cleanly.
+        list(run.tool_calls)
+
+    @pytest.mark.anyio
+    async def test_async_streaming_tool_populates_tool_calls(self) -> None:
+        @tool
+        async def astreamer(text: str) -> str:
+            """async streams."""
+            emit_tool_output_delta(text)
+            emit_tool_output_delta(text + "!")
+            return text
+
+        async def caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {"name": "astreamer", "args": {"text": "hi"}, "id": "tc1"}
+                        ],
+                    )
+                ]
+            }
+
+        graph = _build_graph(caller, [astreamer])
+        run = await graph.astream_v2(
+            {"messages": []}, transformers=[ToolCallTransformer]
+        )
+
+        collected: list[ToolCallStream] = []
+        async for tc in run.tool_calls:
+            collected.append(tc)
+            deltas = [d async for d in tc.output_deltas]
+            assert deltas == ["hi", "hi!"]
+        assert len(collected) == 1
+        assert collected[0].completed is True
+        assert collected[0].error is None
+
+    def test_tool_error_populates_error_field(self) -> None:
+        @tool
+        def boom() -> str:
+            """raises."""
+            raise ValueError("nope")
+
+        def caller(state: _State) -> dict:
+            return {
+                "messages": [
+                    AIMessage(
+                        content="",
+                        tool_calls=[{"name": "boom", "args": {}, "id": "tc1"}],
+                    )
+                ]
+            }
+
+        graph = _build_graph(caller, [boom])
+        run = graph.stream_v2({"messages": []}, transformers=[ToolCallTransformer])
+
+        collected: list[ToolCallStream] = []
+        with pytest.raises(ValueError, match="nope"):
+            for tc in run.tool_calls:
+                collected.append(tc)
+                # Drain deltas so the error field is populated before we
+                # inspect it below.
+                list(tc.output_deltas)
+
+        assert len(collected) == 1
+        assert collected[0].error == "nope"
+        assert collected[0].output is None
+        assert collected[0].completed is True

libs/prebuilt/uv.lock

@@ -249,11 +249,10 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.3.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
-    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -262,21 +261,9 @@ 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/af/bc/0bff31fcaff174d86031cc713471a3e85ed4ec8e5cd95ad0217f2aced20e/langchain_core-1.3.0a2.tar.gz", hash = "sha256:52d978c84552b74b9a3f16c1fced84f9e27cc96d7a67c601925ce6cbc4ea3cf9", size = 854580, upload-time = "2026-04-13T14:37:55.745Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.11"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/d0/bb/38b5eaefa41c67735eedd9f9a2568b11c9eb376fa129a5edd7cc3dcde071/langchain_protocol-0.0.11.tar.gz", hash = "sha256:c276e2373b5ac691fc7ac9a72019d55182444ce8e89385c3f7e9f0185d0aace7", size = 6622, upload-time = "2026-04-23T22:13:16.771Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/0b/fa/6a8ecad8472b182f2caf9d83fd89f40fc1590cb96546d90089b7869b7f5e/langchain_protocol-0.0.11-py3-none-any.whl", hash = "sha256:364da1faf6f5d3001413bede792c1a822c0f23ae55d1ce1266ca7d8e80e79011", size = 6778, upload-time = "2026-04-23T22:13:15.712Z" },
+    { url = "https://files.pythonhosted.org/packages/0e/14/03c09686602567059f26af29de0c44546a83af2f2aa29925e61040e43ea2/langchain_core-1.3.0a2-py3-none-any.whl", hash = "sha256:9e929a34f0b0c6c1255e395a1de34f8626893ceb4cdae550a22a0bd18c87be54", size = 510233, upload-time = "2026-04-13T14:37:54.277Z" },
 ]
 
 [[package]]
@@ -294,7 +281,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.2" },
+    { name = "langchain-core", specifier = "==1.3.0a2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "." },
     { name = "langgraph-sdk", editable = "../sdk-py" },

libs/sdk-py/uv.lock

@@ -262,11 +262,10 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.2"
+version = "1.3.0a2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
-    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -275,21 +274,9 @@ 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/af/bc/0bff31fcaff174d86031cc713471a3e85ed4ec8e5cd95ad0217f2aced20e/langchain_core-1.3.0a2.tar.gz", hash = "sha256:52d978c84552b74b9a3f16c1fced84f9e27cc96d7a67c601925ce6cbc4ea3cf9", size = 854580, upload-time = "2026-04-13T14:37:55.745Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" },
-]
-
-[[package]]
-name = "langchain-protocol"
-version = "0.0.11"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "typing-extensions" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/d0/bb/38b5eaefa41c67735eedd9f9a2568b11c9eb376fa129a5edd7cc3dcde071/langchain_protocol-0.0.11.tar.gz", hash = "sha256:c276e2373b5ac691fc7ac9a72019d55182444ce8e89385c3f7e9f0185d0aace7", size = 6622, upload-time = "2026-04-23T22:13:16.771Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/0b/fa/6a8ecad8472b182f2caf9d83fd89f40fc1590cb96546d90089b7869b7f5e/langchain_protocol-0.0.11-py3-none-any.whl", hash = "sha256:364da1faf6f5d3001413bede792c1a822c0f23ae55d1ce1266ca7d8e80e79011", size = 6778, upload-time = "2026-04-23T22:13:15.712Z" },
+    { url = "https://files.pythonhosted.org/packages/0e/14/03c09686602567059f26af29de0c44546a83af2f2aa29925e61040e43ea2/langchain_core-1.3.0a2-py3-none-any.whl", hash = "sha256:9e929a34f0b0c6c1255e395a1de34f8626893ceb4cdae550a22a0bd18c87be54", size = 510233, upload-time = "2026-04-13T14:37:54.277Z" },
 ]
 
 [[package]]
@@ -307,7 +294,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.2" },
+    { name = "langchain-core", specifier = "==1.3.0a2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "." },