Merge branch 'nh/streaming-transformer' into nh/subgraph-lifecycle

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 GraphStreamer only.
+# flow through stream_mode="messages"; set by StreamingHandler only.
 
 # --- Other constants ---
 PUSH = sys.intern("__pregel_push")

libs/langgraph/langgraph/config.py

@@ -1,7 +1,5 @@
 import asyncio
 import sys
-from collections.abc import Callable
-from contextvars import ContextVar
 from typing import Any
 
 from langchain_core.runnables import RunnableConfig
@@ -11,18 +9,6 @@ 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
@@ -208,30 +194,3 @@ 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

@@ -1,291 +0,0 @@
-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
-    `GraphStreamer` opts in via the internal
+    `StreamingHandler` 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

@@ -1,223 +0,0 @@
-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,7 +130,6 @@ 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,
@@ -142,7 +141,6 @@ 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
@@ -346,18 +344,15 @@ class NodeBuilder:
         )
 
 
-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))
+_STREAM_V2_MODES: list[StreamMode] = [
+    "values",
+    "updates",
+    "messages",
+    "custom",
+    "checkpoints",
+    "tasks",
+    "debug",
+]
 
 
 def _build_stream_factories(
@@ -366,11 +361,20 @@ def _build_stream_factories(
 ) -> list[Callable[..., Any]]:
     """Assemble the factory list handed to `StreamMux(factories=...)`.
 
-    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.
+    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.
     """
     from langgraph.stream.transformers import (
         MessagesTransformer,
@@ -2706,21 +2710,6 @@ 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:
 
@@ -3103,21 +3092,6 @@ 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(
@@ -3355,13 +3329,19 @@ class Pregel(
     ) -> Any:
         """Start a sync v2 streaming run driven by transformer projections.
 
-        Builds a `StreamMux` from the built-in `ValuesTransformer` /
-        `MessagesTransformer`, this graph's compile-time
+        Builds a `StreamMux` from the auto-registered built-ins
+        (`ValuesTransformer`, `MessagesTransformer`,
+        `SubgraphTransformer`), 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.
@@ -3378,12 +3358,11 @@ 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_modes,
+                stream_mode=_STREAM_V2_MODES,
                 subgraphs=True,
                 version="v2",
                 interrupt_before=interrupt_before,
@@ -3420,11 +3399,10 @@ 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_modes,
+            stream_mode=_STREAM_V2_MODES,
             subgraphs=True,
             version="v2",
             interrupt_before=interrupt_before,

libs/langgraph/langgraph/pregel/remote.py

@@ -650,29 +650,16 @@ 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):
-                if stream_mode != "lifecycle":
-                    updated_stream_modes.append(cast(StreamModeSDK, stream_mode))
-                else:
-                    dropped_lifecycle = True
+                updated_stream_modes.append(cast(StreamModeSDK, stream_mode))
             else:
                 req_single = False
                 for m in stream_mode:
-                    if m == "lifecycle":
-                        dropped_lifecycle = True
-                    else:
-                        updated_stream_modes.append(cast(StreamModeSDK, m))
+                    updated_stream_modes.append(cast(StreamModeSDK, m))
         else:
-            updated_stream_modes.append(default)  # type: ignore[arg-type]
+            updated_stream_modes.append(default)
         requested_stream_modes = updated_stream_modes.copy()
         # add any from parent graph
         stream: StreamProtocol | None = (
@@ -680,16 +667,7 @@ class RemoteGraph(PregelProtocol):
         )
         if stream:
             for m in stream.modes:
-                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."
-            )
+                updated_stream_modes.append(cast(StreamModeSDK, m))
         # map "messages" to "messages-tuple"
         if "messages" in updated_stream_modes:
             updated_stream_modes.remove("messages")

libs/langgraph/langgraph/stream/_event_log.py

@@ -39,13 +39,20 @@ class EventLog(Generic[T]):
     skipped.
     """
 
-    def __init__(self, maxlen: int | None = None) -> None:
+    def __init__(self, maxlen: int | None = None, *, retain: bool = False) -> 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`.
@@ -61,8 +68,9 @@ class EventLog(Generic[T]):
         self._is_async: bool | None = None
 
         # Flipped on first __iter__ / __aiter__. Pre-subscription
-        # pushes are silent no-ops.
+        # pushes are silent no-ops unless `_retain` is True.
         self._subscribed = False
+        self._retain = retain
 
         # Pump wiring set by the run stream after bind.
         self._request_more: Callable[[], bool] | None = None
@@ -99,10 +107,14 @@ 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:
+        if not self._subscribed and not self._retain:
             return
         if self._closed:
             raise RuntimeError("Cannot push to a closed EventLog")

libs/langgraph/langgraph/stream/_mux.py

@@ -140,6 +140,18 @@ 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:
@@ -211,13 +223,14 @@ class StreamMux:
                 f"keys: {attributions}"
             )
         self._transformers.append(transformer)
-        self._bind_and_wire(projection)
+        is_native = bool(getattr(transformer, "_native", False))
+        self._bind_and_wire(projection, is_native=is_native)
         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 getattr(transformer, "_native", False):
+        if is_native:
             self.native_keys.update(projection.keys())
         on_register = getattr(transformer, "_on_register", None)
         if on_register is not None:
@@ -445,26 +458,34 @@ class StreamMux:
     # Binding and StreamChannel auto-wiring
     # ------------------------------------------------------------------
 
-    def _bind_and_wire(self, projection: dict[str, Any]) -> None:
-        """Bind and wire EventLog / StreamChannel instances in a projection."""
+    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.
+        """
         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) -> Callable[[Any], None]:
+                def _make_forward(name: str, native: bool) -> Callable[[Any], None]:
                     def _forward(item: Any) -> None:
-                        self._forward(name, item)
+                        self._forward(name, item, native=native)
 
                     return _forward
 
-                value._wire(_make_forward(channel_name))
+                value._wire(_make_forward(channel_name, is_native))
             elif isinstance(value, EventLog):
                 value._bind(is_async=self._is_async)
                 self._logs.append(value)
 
-    def _forward(self, channel_name: str, item: Any) -> None:
+    def _forward(self, channel_name: str, item: Any, *, native: bool) -> None:
         """Inject a ProtocolEvent for a StreamChannel push.
 
         Forwarded events bypass the transformer pipeline to avoid
@@ -472,12 +493,18 @@ 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": f"custom:{channel_name}",
+            "method": method,
             "params": {
                 "namespace": [],
                 "timestamp": int(time.time() * 1000),

libs/langgraph/langgraph/stream/_types.py

@@ -94,17 +94,10 @@ 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,25 +49,23 @@ class BaseRunStream:
 
     @property
     def _values_transformer(self) -> ValuesTransformer:
-        """Look up the `ValuesTransformer` backing `output` / `interrupted`.
+        """Look up the `ValuesTransformer` registered on this mux.
 
-        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.
+        `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.
         """
         from langgraph.stream.transformers import ValuesTransformer
 
-        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
+        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."
+        )
 
     def __iter__(self) -> Iterator[ProtocolEvent]:
         """Sync iteration of protocol events on this mux's main log.
@@ -156,9 +154,8 @@ 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 under the
-                `"values"` key — `output` / `interrupted` / `interrupts`
-                read from it lazily.
+                Must have a `ValuesTransformer` registered for
+                `output` / `interrupted` / `interrupts` to work.
         """
         super().__init__(mux)
         self._graph_iter = graph_iter
@@ -217,10 +214,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)
-        vt = self._values_transformer
-        if vt.error is not None:
-            raise vt.error
-        return vt._latest
+        err = self._values_transformer.error
+        if err is not None:
+            raise err
+        return self._values_transformer._latest
 
     @property
     def interrupted(self) -> bool:
@@ -230,10 +227,10 @@ class GraphRunStream(BaseRunStream):
             BaseException: If the run ended with an error.
         """
         _drive_until_done(self._pump_next)
-        vt = self._values_transformer
-        if vt.error is not None:
-            raise vt.error
-        return vt._interrupted
+        err = self._values_transformer.error
+        if err is not None:
+            raise err
+        return self._values_transformer._interrupted
 
     @property
     def interrupts(self) -> list[Any]:
@@ -243,10 +240,10 @@ class GraphRunStream(BaseRunStream):
             BaseException: If the run ended with an error.
         """
         _drive_until_done(self._pump_next)
-        vt = self._values_transformer
-        if vt.error is not None:
-            raise vt.error
-        return vt._interrupts
+        err = self._values_transformer.error
+        if err is not None:
+            raise err
+        return self._values_transformer._interrupts
 
 
 class AsyncGraphRunStream(BaseRunStream):
@@ -283,9 +280,8 @@ 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 under the
-                `"values"` key — `output` / `interrupted` / `interrupts`
-                read from it lazily.
+                Must have a `ValuesTransformer` registered for
+                `output` / `interrupted` / `interrupts` to work.
         """
         super().__init__(mux)
         self._graph_aiter = graph_aiter

libs/langgraph/langgraph/stream/stream_channel.py

@@ -37,7 +37,9 @@ class StreamChannel(Generic[T]):
     using only StreamChannels don't need `finalize` or `fail` hooks.
     """
 
-    def __init__(self, name: str, *, maxlen: int | None = None) -> None:
+    def __init__(
+        self, name: str, *, maxlen: int | None = None, retain: bool = False
+    ) -> None:
         """Initialize the channel with an empty inner log.
 
         Args:
@@ -45,9 +47,13 @@ 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)
+        self._log: EventLog[T] = EventLog(maxlen=maxlen, retain=retain)
         self._wire_fn: Callable[[T], None] | None = None
 
     def _bind(self, *, is_async: bool) -> None:

libs/langgraph/langgraph/stream/transformers.py

@@ -9,12 +9,14 @@ from langchain_core.language_models.chat_model_stream import (
     ChatModelStream,
 )
 from langchain_core.messages import AIMessageChunk, BaseMessage
-from langchain_protocol.protocol import CheckpointRef, LifecycleData, MessagesData
+from langchain_protocol.protocol import CheckpointRef, MessagesData
+from typing_extensions import TypedDict
 
 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
@@ -25,12 +27,38 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-SubgraphStatus = Literal["started", "running", "completed", "failed", "interrupted"]
+SubgraphStatus = Literal["started", "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.
 
@@ -49,7 +77,6 @@ class ValuesTransformer(StreamTransformer):
     """
 
     _native = True
-    required_stream_modes = ("values",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -122,7 +149,6 @@ class MessagesTransformer(StreamTransformer):
     """
 
     _native = True
-    required_stream_modes = ("messages",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -266,14 +292,17 @@ class SubgraphRunStream(BaseRunStream):
     `make_child`'s pump inheritance, so any cursor on a subagent
     projection drives the whole run forward.
 
-    Lifecycle fields update in place as events arrive:
+    Handle fields:
 
     - `path`: the namespace tuple — stable for the life of the handle.
-    - `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.
+    - `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.
 
     `.output` is a snapshot of the latest values seen at this
     namespace — it doesn't drive the pump (unlike root's
@@ -313,10 +342,10 @@ class SubgraphRunStream(BaseRunStream):
 class SubgraphTransformer(StreamTransformer):
     """Discover subgraphs and route events into per-subgraph mini-muxes.
 
-    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
+    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
     `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.
@@ -328,10 +357,15 @@ class SubgraphTransformer(StreamTransformer):
     `SubgraphTransformer` for grandchildren) handle the rest. No
     duplicated routing or assembly logic.
 
-    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.
+    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.
 
     Native transformer — `subgraphs` exposes the direct-children log.
 
@@ -342,7 +376,6 @@ class SubgraphTransformer(StreamTransformer):
 
     _native = True
     scope_exact = False
-    required_stream_modes = ("lifecycle",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
@@ -360,45 +393,23 @@ class SubgraphTransformer(StreamTransformer):
 
     def process(self, event: ProtocolEvent) -> bool:
         ns = tuple(event["params"]["namespace"])
-        method = event["method"]
         depth = len(self.scope)
 
-        # 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)
+        # 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)
 
-        # 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.
+        # 2. Forward the event to the matching direct-child mini-mux.
+        #    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, ...], data: LifecycleData) -> None:
-        if ns in self._by_ns:
-            # Duplicate started — ignore.
-            return
+    def _on_started(self, ns: tuple[str, ...]) -> None:
         # `_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.
@@ -406,33 +417,17 @@ 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=data.get("graph_name"),
-            trigger_call_id=data.get("trigger_call_id"),
+            graph_name=graph_name,
+            trigger_call_id=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
@@ -449,10 +444,21 @@ class SubgraphTransformer(StreamTransformer):
                 )
 
     def finalize(self) -> None:
-        """Transition any still-open direct children to `completed`."""
+        """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".
+        """
         for handle in self._by_ns.values():
             if handle.status not in _TERMINAL_STATUSES:
-                handle.status = "completed"
+                values_t = handle._mux.transformer_by_key("values")
+                if isinstance(values_t, ValuesTransformer) and values_t._interrupted:
+                    handle.status = "interrupted"
+                else:
+                    handle.status = "completed"
             self._close_handle_mux(handle)
 
     def fail(self, err: BaseException) -> None:
@@ -475,3 +481,105 @@ 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,8 +123,6 @@ StreamMode = Literal[
     "debug",
     "messages",
     "custom",
-    "lifecycle",
-    "tools",
 ]
 """How the stream method should emit outputs.
 
@@ -137,8 +135,6 @@ 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.0a2",
+    "langchain-core==1.3.2",
     "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,24 +131,6 @@ 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
 # ---------------------------------------------------------------------------
@@ -483,29 +465,13 @@ class TestStreamV2Sync:
 
     def test_custom_stream_events(self) -> None:
         graph = _build_custom_stream_graph()
-        run = graph.stream_v2(
-            {"value": "x", "items": []},
-            transformers=[_CustomPassthroughTransformer],
-        )
+        handler = graph
+        run = handler.stream_v2({"value": "x", "items": []})
         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
@@ -752,10 +718,8 @@ class TestStreamV2AsyncCustom:
     @NEEDS_CONTEXTVARS
     async def test_custom_stream_events(self) -> None:
         graph = _build_custom_stream_graph()
-        run = await graph.astream_v2(
-            {"value": "x", "items": []},
-            transformers=[_CustomPassthroughTransformer],
-        )
+        handler = graph
+        run = await handler.astream_v2({"value": "x", "items": []})
         events = [e async for e in run]
         custom_events = [e for e in events if e["method"] == "custom"]
         assert len(custom_events) == 2
@@ -1822,92 +1786,3 @@ 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

@@ -0,0 +1,258 @@
+"""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.content == "hello world"
+        assert stream.output.text == "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.content == "aaaa"
-        assert by_id["run-b"].output.content == "bbbb"
+        assert by_id["run-a"].output.text == "aaaa"
+        assert by_id["run-b"].output.text == "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.content == "the full answer"
+        assert stream.output.text == "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.content == "async"
+        assert msg.text == "async"
 
 
 # ---------------------------------------------------------------------------
@@ -471,7 +471,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.content == "mux stream"
+        assert stream.output.text == "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.content == "result"
+        assert stream.output.text == "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.content == "async mux"
+        assert msg.text == "async mux"
         await mux.aclose()
 
 
@@ -545,7 +545,7 @@ class TestEndToEnd:
 
         assert len(streams) == 1
         assert isinstance(streams[0], ChatModelStream)
-        assert streams[0].output.content == "hello world"
+        assert streams[0].output.text == "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.content == "hardcoded"
+        assert streams[0].output.text == "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.content == "async answer"
+        assert msg.text == "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.content == "hello world"
+        assert stream.output.text == "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.content == "streamed answer"
+        assert stream.output.text == "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.content for s in streams}
+        contents = {s.output.text 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.content == "live stream"
+        assert streams[0].output.text == "live stream"
         assert streams[1].node == "constructed_node"
-        assert streams[1].output.content == "hardcoded"
+        assert streams[1].output.text == "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.content == "async invoke"
+        assert msg.text == "async invoke"
 
 
 class TestDirectMessagesModeStaysV1:

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -1,4 +1,4 @@
-"""Tests for subgraph lifecycle events and the SubgraphTransformer."""
+"""Tests for SubgraphTransformer namespace-based discovery."""
 
 from __future__ import annotations
 
@@ -27,32 +27,6 @@ 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",
@@ -108,96 +82,90 @@ 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_started_is_ignored(self) -> None:
+    def test_root_event_does_not_create_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", graph_name="root"))
+        mux.push(_values({"value": 1}, namespace=[]))
         assert list(transformer._root_log._items) == []
         assert transformer._by_ns == {}
 
-    def test_child_started_yields_handle(self) -> None:
+    def test_first_event_at_child_depth_yields_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(
-            _lifecycle(
-                "started",
-                namespace=["task_a:child"],
-                graph_name="child",
-                trigger_call_id="task_a",
-            )
-        )
+        mux.push(_values({"value": 1}, namespace=["child:task_a"]))
 
         handle = self._handle(transformer)
-        assert handle.path == ("task_a:child",)
+        assert handle.path == ("child:task_a",)
         assert handle.graph_name == "child"
         assert handle.trigger_call_id == "task_a"
         assert handle.status == "started"
 
-    def test_status_transitions(self) -> None:
+    def test_handle_without_task_id_suffix(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
-        mux.push(_lifecycle("running", namespace=["t:c"]))
-        mux.push(_lifecycle("completed", namespace=["t:c"]))
+        mux.push(_values({"value": 1}, namespace=["child"]))
 
         handle = self._handle(transformer)
-        assert handle.status == "completed"
+        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",)
 
     def test_grandchild_surfaces_under_child(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:child"], graph_name="child"))
+        mux.push(_values({"value": 1}, namespace=["child:t"]))
 
         child = self._handle(transformer)
         _pre_subscribe_handle(child)
 
-        mux.push(
-            _lifecycle(
-                "started",
-                namespace=["t:child", "u:grand"],
-                graph_name="grand",
-            )
-        )
+        mux.push(_values({"value": 2}, namespace=["child:t", "grand:u"]))
 
         (grand,) = _handle_subgraphs_items(child)
-        assert grand.path == ("t:child", "u:grand")
+        assert grand.path == ("child:t", "grand:u")
         assert grand.graph_name == "grand"
-
-    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"
+        assert grand.trigger_call_id == "u"
 
     def test_values_routed_into_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
 
         handle = self._handle(transformer)
         _pre_subscribe_handle(handle)
 
-        mux.push(_values({"value": 1}, namespace=["t:c"]))
-        mux.push(_values({"value": 2}, namespace=["t:c"]))
+        mux.push(_values({"value": 2}, namespace=["c:t"]))
 
+        # 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(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
         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) == []
+        assert _handle_values_items(handle) == [{"value": 1}]
 
     def test_finalize_closes_dangling(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
         handle = self._handle(transformer)
 
         mux.close()
@@ -207,7 +175,7 @@ class TestSubgraphTransformerUnit:
 
     def test_fail_with_graph_interrupt_marks_interrupted(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
         handle = self._handle(transformer)
 
         mux.fail(GraphInterrupt())
@@ -215,32 +183,22 @@ class TestSubgraphTransformerUnit:
 
     def test_fail_with_generic_error_marks_failed(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
         handle = self._handle(transformer)
 
         mux.fail(RuntimeError("explode"))
         assert handle.status == "failed"
         assert handle.error == "explode"
 
-    def test_duplicate_started_ignored(self) -> None:
+    def test_repeated_events_same_ns_single_handle(self) -> None:
         mux, transformer = self._mux()
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
-        mux.push(_lifecycle("started", namespace=["t:c"], graph_name="other"))
+        mux.push(_values({"value": 1}, namespace=["c:t"]))
+        mux.push(_values({"value": 2}, namespace=["c:t"]))
+        mux.push(_values({"value": 3}, namespace=["c:t"]))
 
         handles = list(transformer._root_log._items)
         assert len(handles) == 1
-        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) == []
+        assert handles[0].path == ("c:t",)
 
 
 # ---------------------------------------------------------------------------

libs/langgraph/tests/test_tool_stream_handler.py

@@ -1,290 +0,0 @@
-"""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,10 +1348,11 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0a2"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
+    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -1360,9 +1361,21 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-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" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/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" },
+    { 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" },
 ]
 
 [[package]]
@@ -1439,7 +1452,7 @@ test = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.0a2" },
+    { name = "langchain-core", specifier = "==1.3.2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "../sdk-py" },

libs/prebuilt/langgraph/prebuilt/__init__.py

@@ -1,7 +1,5 @@
 """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,
@@ -15,8 +13,6 @@ 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

@@ -1,117 +0,0 @@
-"""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

@@ -1,128 +0,0 @@
-"""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

@@ -1,306 +0,0 @@
-"""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,10 +249,11 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0a2"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
+    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -261,9 +262,21 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-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" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/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" },
+    { 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" },
 ]
 
 [[package]]
@@ -281,7 +294,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.0a2" },
+    { name = "langchain-core", specifier = "==1.3.2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "." },
     { name = "langgraph-sdk", editable = "../sdk-py" },

libs/sdk-py/uv.lock

@@ -262,10 +262,11 @@ wheels = [
 
 [[package]]
 name = "langchain-core"
-version = "1.3.0a2"
+version = "1.3.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "jsonpatch" },
+    { name = "langchain-protocol" },
     { name = "langsmith" },
     { name = "packaging" },
     { name = "pydantic" },
@@ -274,9 +275,21 @@ dependencies = [
     { name = "typing-extensions" },
     { name = "uuid-utils" },
 ]
-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" }
+sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/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" },
+    { 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" },
 ]
 
 [[package]]
@@ -294,7 +307,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
-    { name = "langchain-core", specifier = "==1.3.0a2" },
+    { name = "langchain-core", specifier = "==1.3.2" },
     { name = "langgraph-checkpoint", editable = "../checkpoint" },
     { name = "langgraph-prebuilt", editable = "../prebuilt" },
     { name = "langgraph-sdk", editable = "." },