move more over

Commit 133082af preserved the full checkpoint ns on v2 message
events so subgraph chat-model tokens emit at their full path (matching
JS's handleChatModelStart). That promoted root-level chat tokens from
the empty tuple to depth 1 (e.g. ("call_model:<task>",)), but the root
MessagesTransformer was scope_exact=True at scope=(), so every
streamed message was dropped — run.messages yielded zero streams for
any graph calling model.invoke() / stream_v2().

Mirror JS's namespaces=[[]], depth=1 filter: accept events at the
transformer's scope or exactly one segment deeper. SubgraphTransformer
still forwards deeper events to the matching child mini-mux, whose
own MessagesTransformer applies the same scope+1 rule.

Also update stale test assertions: ChatModelStream.output.content is
now always a list of v1 content blocks by construction, so tests
compare via str(stream.text) (sync) or an explicit block-shape
assertion on the assembled AIMessage (async, since AsyncProjection
isn't str-able).

libs/langgraph/langgraph/pregel/_lifecycle.py

@@ -166,24 +166,6 @@ class StreamLifecycleHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             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))
 
@@ -248,9 +230,10 @@ class StreamLifecycleHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         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
+        # `cause` is intentionally not populated here: pregel does not know
+        # what on the parent namespace triggered this subgraph. Product-
+        # specific stream transformers populate `cause` before events
+        # reach the wire. See LifecycleCause in the protocol definition.
         self._emit(ns, payload)
         self._pending_running.add(ns)
 

libs/langgraph/langgraph/pregel/_messages.py

@@ -279,6 +279,113 @@ class StreamMessagesHandlerV2(StreamMessagesHandler, _V2StreamingCallbackHandler
     AIMessageChunk shape.
     """
 
+    def on_chat_model_start(
+        self,
+        serialized: dict[str, Any],
+        messages: list[list[BaseMessage]],
+        *,
+        run_id: UUID,
+        parent_run_id: UUID | None = None,
+        tags: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        """Record metadata with the FULL checkpoint namespace for v2.
+
+        v1's ``on_chat_model_start`` (inherited) slices the ns tuple
+        with ``[:-1]`` to re-position chat model tokens onto the
+        *containing pregel's* namespace — historically convenient for
+        consumers of ``stream_mode="messages"`` who want "where did
+        this node produce its output" rather than the chat-model's
+        own task ns.
+
+        For the protocol-v2 wire shape that is wrong: the client
+        subscribes the root feed at ``namespaces=[[]]`` with
+        ``depth=1``, and any message emitted at depth ``>=1`` that
+        still carries the containing node's ns must appear at the
+        *full* path from root so that depth filtering cleanly isolates
+        subgraph chatter from the root conversation. JS's
+        ``StreamMessagesHandlerV2`` already does
+        ``metadata.langgraph_checkpoint_ns.split("|")`` (no slice); this
+        override brings the Python v2 handler to the same shape.
+
+        Without this, a chat model invoked inside a nested subgraph
+        (e.g. ``research -> researcher``, ``research`` being a root
+        node that ``.ainvoke()``s a ``researcher`` subgraph) emits at
+        ``["research:<task>"]`` — a single level deep — which slips
+        through the root-feed depth-1 filter and pollutes the main
+        conversation with subgraph tokens. With this override we emit
+        at ``["research:<task>", "researcher:<task>"]`` so the client
+        routes those tokens to the subgraph card instead.
+        """
+        if metadata and (not tags or (TAG_NOSTREAM not in tags)):
+            task_checkpoint_ns = cast(str, metadata["langgraph_checkpoint_ns"])
+            # Keep the trailing ``:<task_id>`` segment (unlike the v1
+            # handler which strips it via ``[:-1]``). The client's
+            # lifecycle events land on the same ns, so message deltas
+            # now correlate 1:1 with a ``lifecycle: started`` event —
+            # ``useMessages(stream, subgraph)`` picks them up without
+            # needing to collapse sibling namespaces.
+            ns = tuple(task_checkpoint_ns.split(NS_SEP))
+            if not self.subgraphs and len(ns) > 1 and ns != self.parent_ns:
+                return
+            stream_metadata = dict(metadata)
+            # Preserve the v1-shaped ``langgraph_checkpoint_ns`` (task
+            # id stripped, trailing ``NS_END`` retained) so downstream
+            # consumers reading checkpoint metadata off a streamed
+            # message see the same shape they did pre-v2. Only the ns
+            # tuple emitted on the wire changes.
+            checkpoint_ns = (
+                f"{task_checkpoint_ns.rsplit(NS_END, 1)[0]}{NS_END}"
+                if NS_END in task_checkpoint_ns
+                else task_checkpoint_ns
+            )
+            stream_metadata["langgraph_checkpoint_ns"] = checkpoint_ns
+            stream_metadata["checkpoint_ns"] = checkpoint_ns
+            if tags:
+                if filtered_tags := [t for t in tags if not t.startswith("seq:step")]:
+                    stream_metadata["tags"] = filtered_tags
+            self.metadata[run_id] = (ns, stream_metadata)
+
+    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:
+        """Record chain (node) metadata with the FULL checkpoint ns.
+
+        Mirror of :meth:`on_chat_model_start` for the node-start path,
+        so messages returned by ``on_chain_end`` (``Command`` updates
+        and plain state dict outputs) land at the same full-path ns as
+        any chat-model deltas from within that node. See the
+        :meth:`on_chat_model_start` docstring for why the v1 ``[:-1]``
+        slice is dropped here.
+        """
+        if (
+            metadata
+            and kwargs.get("name") == metadata.get("langgraph_node")
+            and (not tags or TAG_HIDDEN not in tags)
+        ):
+            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))
+            if not self.subgraphs and len(ns) > 1:
+                return
+            self.metadata[run_id] = (ns, metadata)
+            for value in _state_values(inputs):
+                if isinstance(value, BaseMessage):
+                    if value.id is not None:
+                        self.seen.add(value.id)
+                elif isinstance(value, Sequence) and not isinstance(value, str):
+                    for item in value:
+                        if isinstance(item, BaseMessage):
+                            if item.id is not None:
+                                self.seen.add(item.id)
+
     def on_llm_new_token(
         self,
         token: str,

libs/langgraph/langgraph/pregel/main.py

@@ -375,11 +375,13 @@ def _build_stream_factories(
     from langgraph.stream.transformers import (
         MessagesTransformer,
         SubgraphTransformer,
+        ToolLifecycleTransformer,
         ValuesTransformer,
     )
 
     builtins: list[Callable[..., Any]] = [
         ValuesTransformer,
+        ToolLifecycleTransformer,
         MessagesTransformer,
         SubgraphTransformer,
     ]
@@ -874,6 +876,15 @@ class Pregel(
 
     def copy(self, update: dict[str, Any] | None = None) -> Self:
         attrs = {k: v for k, v in self.__dict__.items() if k != "__orig_class__"}
+        # ``__init__`` accepts ``stream_transformers`` (public parameter) but
+        # the attribute is stored as ``_stream_transformers`` (private). Map
+        # the private key back onto the public kwarg so compile-time
+        # transformers survive ``copy()`` / ``with_config()``. Without this,
+        # ``_stream_transformers`` gets captured by ``**deprecated_kwargs``
+        # and the resulting instance silently has an empty transformer
+        # pipeline.
+        if "_stream_transformers" in attrs:
+            attrs["stream_transformers"] = attrs.pop("_stream_transformers")
         attrs.update(update or {})
         return self.__class__(**attrs)
 
@@ -3352,6 +3363,9 @@ class Pregel(
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         transformers: Sequence[Any] | None = None,
+        stream_modes: Sequence[StreamMode] | None = None,
+        output_keys: str | Sequence[str] | None = None,
+        **kwargs: Any,
     ) -> Any:
         """Start a sync v2 streaming run driven by transformer projections.
 
@@ -3378,16 +3392,19 @@ class Pregel(
 
         factories = _build_stream_factories(self._stream_transformers, transformers)
         mux = StreamMux(factories=factories, is_async=False)
-        stream_modes = _collect_stream_modes(mux)
+        requested_stream_modes = set(_collect_stream_modes(mux))
+        requested_stream_modes.update(stream_modes or ())
         graph_iter = iter(
             self.stream(
                 input,
                 _merge_v2_messages_flag(config),
-                stream_mode=stream_modes,
+                stream_mode=list(requested_stream_modes),
                 subgraphs=True,
                 version="v2",
+                output_keys=output_keys,
                 interrupt_before=interrupt_before,
                 interrupt_after=interrupt_after,
+                **kwargs,
             )
         )
         return GraphRunStream(graph_iter, mux)
@@ -3400,6 +3417,9 @@ class Pregel(
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         transformers: Sequence[Any] | None = None,
+        stream_modes: Sequence[StreamMode] | None = None,
+        output_keys: str | Sequence[str] | None = None,
+        **kwargs: Any,
     ) -> Any:
         """Async counterpart to `stream_v2`.
 
@@ -3420,15 +3440,18 @@ class Pregel(
 
         factories = _build_stream_factories(self._stream_transformers, transformers)
         mux = StreamMux(factories=factories, is_async=True)
-        stream_modes = _collect_stream_modes(mux)
+        requested_stream_modes = set(_collect_stream_modes(mux))
+        requested_stream_modes.update(stream_modes or ())
         graph_aiter = self.astream(
             input,
             _merge_v2_messages_flag(config),
-            stream_mode=stream_modes,
+            stream_mode=list(requested_stream_modes),
             subgraphs=True,
             version="v2",
+            output_keys=output_keys,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
+            **kwargs,
         ).__aiter__()
         return AsyncGraphRunStream(graph_aiter, mux)
 

libs/langgraph/langgraph/stream/_convert.py

@@ -7,6 +7,31 @@ from langgraph.stream._types import ProtocolEvent, _ProtocolEventParams
 from langgraph.types import StreamPart
 
 
+def _is_v2_messages_payload(data: Any) -> bool:
+    return isinstance(data, dict) and isinstance(data.get("event"), str)
+
+
+def _normalize_messages_data(data: dict[str, Any]) -> dict[str, Any]:
+    """Normalize Python Core message fields to the protocol wire shape."""
+    normalized = {**data}
+    if (
+        normalized["event"] == "message-start"
+        and "id" not in normalized
+        and isinstance(normalized.get("message_id"), str)
+    ):
+        normalized["id"] = normalized["message_id"]
+    if (
+        normalized["event"]
+        in ("content-block-start", "content-block-delta", "content-block-finish")
+        and "content" not in normalized
+        and isinstance(normalized.get("content_block"), dict)
+    ):
+        normalized["content"] = normalized["content_block"]
+    normalized.pop("message_id", None)
+    normalized.pop("content_block", None)
+    return normalized
+
+
 def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent:
     """Convert a v2 StreamPart to a ProtocolEvent.
 
@@ -18,11 +43,25 @@ def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent:
         The equivalent ProtocolEvent.
     """
     part_dict = cast(dict[str, Any], part)
+    data = part_dict["data"]
     params: _ProtocolEventParams = {
         "namespace": list(part_dict["ns"]),
         "timestamp": int(time.time() * 1000),
-        "data": part_dict["data"],
+        "data": data,
     }
+    if (
+        part_dict["type"] == "messages"
+        and isinstance(data, tuple)
+        and len(data) == 2
+        and _is_v2_messages_payload(data[0])
+        and isinstance(data[1], dict)
+    ):
+        payload, metadata = data
+        params["data"] = _normalize_messages_data(payload)
+        if isinstance(metadata.get("langgraph_node"), str):
+            params["node"] = metadata["langgraph_node"]
+        if isinstance(metadata.get("run_id"), str):
+            params["run_id"] = metadata["run_id"]
     if "interrupts" in part_dict:
         params["interrupts"] = part_dict["interrupts"]
     return {

libs/langgraph/langgraph/stream/_mux.py

@@ -227,6 +227,18 @@ class StreamMux:
         """Return the transformer that owns the projection at `key`, if any."""
         return self._transformer_by_key.get(key)
 
+    def emit(self, event: ProtocolEvent) -> None:
+        """Append a protocol event directly to the main log.
+
+        Built-in transformers use this for protocol repair events that
+        must appear before the source event they are processing. Direct
+        emission intentionally bypasses the transformer pipeline, but
+        still lets this mux remain the only local sequencing authority.
+        """
+        self._seq += 1
+        event["seq"] = self._seq
+        self._events.push(event)
+
     def push(self, event: ProtocolEvent) -> None:
         """Route an event through all transformers, then append to the main log.
 
@@ -473,10 +485,8 @@ class StreamMux:
         visible in the main event log but are not passed through
         transformers' `process()` methods.
         """
-        self._seq += 1
         event: ProtocolEvent = {
             "type": "event",
-            "seq": self._seq,
             "method": f"custom:{channel_name}",
             "params": {
                 "namespace": [],
@@ -484,4 +494,4 @@ class StreamMux:
                 "data": item,
             },
         }
-        self._events.push(event)
+        self.emit(event)

libs/langgraph/langgraph/stream/_types.py

@@ -22,6 +22,8 @@ class _ProtocolEventParams(TypedDict):
     namespace: list[str]
     timestamp: int
     data: Any
+    node: NotRequired[str]
+    run_id: NotRequired[str]
     interrupts: NotRequired[tuple[Any, ...]]
 
 
@@ -35,7 +37,7 @@ class ProtocolEvent(TypedDict):
     """
 
     type: Literal["event"]
-    eventId: NotRequired[str]
+    event_id: NotRequired[str]
     seq: NotRequired[int]
     method: str  # StreamMode value: "values", "messages", "custom", etc.
     params: _ProtocolEventParams

libs/langgraph/langgraph/stream/transformers.py

@@ -9,7 +9,12 @@ from langchain_core.language_models.chat_model_stream import (
     ChatModelStream,
 )
 from langchain_core.messages import AIMessageChunk, BaseMessage
-from langchain_protocol.protocol import CheckpointRef, LifecycleData, MessagesData
+from langchain_protocol.protocol import (
+    CheckpointRef,
+    LifecycleCause,
+    LifecycleData,
+    MessagesData,
+)
 
 from langgraph.errors import GraphInterrupt
 from langgraph.stream._event_log import EventLog
@@ -31,6 +36,106 @@ _TERMINAL_STATUSES: frozenset[SubgraphStatus] = frozenset(
 )
 
 
+def _is_record(value: Any) -> bool:
+    return isinstance(value, dict)
+
+
+def _to_chat_model_stream_event(event: MessagesData) -> MessagesData:
+    """Convert wire-shaped message fields to ChatModelStream's internal shape."""
+    event_type = event.get("event")
+    converted: dict[str, Any] = dict(event)
+    if (
+        event_type == "message-start"
+        and "message_id" not in converted
+        and isinstance(converted.get("id"), str)
+    ):
+        converted["message_id"] = converted["id"]
+    if (
+        event_type in ("content-block-start", "content-block-delta", "content-block-finish")
+        and "content_block" not in converted
+        and isinstance(converted.get("content"), dict)
+    ):
+        converted["content_block"] = converted["content"]
+    return cast("MessagesData", converted)
+
+
+def _message_event_id(event: MessagesData) -> str | None:
+    raw_id = event.get("id") or event.get("message_id")
+    return str(raw_id) if raw_id is not None else None
+
+
+def _content_block_start_skeleton(content: Any) -> dict[str, Any] | None:
+    """Return a minimal content-block-start payload for a delta/finish block."""
+    if not _is_record(content) or not isinstance(content.get("type"), str):
+        return None
+
+    block_type = content["type"]
+    skeleton: dict[str, Any] = {"type": block_type}
+    if block_type == "text":
+        skeleton["text"] = ""
+    elif block_type == "reasoning":
+        skeleton["reasoning"] = ""
+    elif block_type in ("tool_call", "tool_call_chunk"):
+        skeleton["type"] = "tool_call_chunk"
+        if isinstance(content.get("id"), str):
+            skeleton["id"] = content["id"]
+        if isinstance(content.get("name"), str):
+            skeleton["name"] = content["name"]
+        skeleton["args"] = ""
+    elif block_type in ("server_tool_call", "server_tool_call_chunk"):
+        skeleton["type"] = "server_tool_call_chunk"
+        if isinstance(content.get("id"), str):
+            skeleton["id"] = content["id"]
+        if isinstance(content.get("name"), str):
+            skeleton["name"] = content["name"]
+        skeleton["args"] = ""
+    return skeleton
+
+
+def _copy_event(
+    source: ProtocolEvent,
+    *,
+    method: str,
+    namespace: list[str],
+    data: Any,
+) -> ProtocolEvent:
+    params = {**source["params"], "namespace": namespace, "data": data}
+    return {"type": "event", "method": method, "params": params}
+
+
+def _message_repair_key(event: ProtocolEvent, run_id: str) -> str:
+    namespace_key = "\x1f".join(event["params"]["namespace"])
+    return f"{namespace_key}\x1e{run_id}"
+
+
+def _extract_tool_calls_from_values(data: Any) -> dict[str, dict[str, Any]]:
+    if not _is_record(data):
+        return {}
+    messages = data.get("messages")
+    if not isinstance(messages, list):
+        return {}
+    known: dict[str, dict[str, Any]] = {}
+    for message in messages:
+        if not _is_record(message):
+            continue
+        tool_calls = message.get("tool_calls")
+        if not isinstance(tool_calls, list):
+            continue
+        for tool_call in tool_calls:
+            if not _is_record(tool_call):
+                continue
+            tool_call_id = tool_call.get("id")
+            if not isinstance(tool_call_id, str):
+                continue
+            name = tool_call.get("name")
+            args = tool_call.get("args")
+            known[tool_call_id] = {
+                "tool_name": name if isinstance(name, str) else "",
+                "input": args if _is_record(args) else {},
+            }
+    return known
+
+
 class ValuesTransformer(StreamTransformer):
     """Capture values events as a drainable stream of state snapshots.
 
@@ -83,6 +188,86 @@ class ValuesTransformer(StreamTransformer):
         return True
 
 
+class ToolLifecycleTransformer(StreamTransformer):
+    """Repair tool-start events needed for deterministic subagent discovery.
+
+    Some subagent frameworks expose a tool-caused subgraph lifecycle before
+    a LangChain tool callback has emitted the matching `tool-started`
+    frame. Core can infer the missing start from the latest values snapshot
+    (`messages[*].tool_calls`) and emit it before the lifecycle event leaves
+    the mux, keeping remote clients from guessing from values snapshots.
+    """
+
+    scope_exact = False
+    required_stream_modes = ("values", "tools", "lifecycle")
+
+    def __init__(self, scope: tuple[str, ...] = ()) -> None:
+        super().__init__(scope)
+        self._known_tool_calls: dict[str, dict[str, Any]] = {}
+        self._emitted_tool_starts: set[str] = set()
+        self._mux: StreamMux | None = None
+
+    def init(self) -> dict[str, Any]:
+        return {}
+
+    def _on_register(self, mux: StreamMux) -> None:
+        self._mux = mux
+
+    def process(self, event: ProtocolEvent) -> bool:
+        method = event["method"]
+        data = event["params"]["data"]
+        if method == "values":
+            self._known_tool_calls.update(_extract_tool_calls_from_values(data))
+            return True
+        if method == "tools" and _is_record(data):
+            if (
+                data.get("event") == "tool-started"
+                and isinstance(data.get("tool_call_id"), str)
+            ):
+                tool_call_id = cast("str", data["tool_call_id"])
+                if tool_call_id in self._emitted_tool_starts:
+                    return False
+                self._emitted_tool_starts.add(tool_call_id)
+            return True
+        if method == "lifecycle":
+            self._emit_missing_tool_started(event)
+        return True
+
+    def _emit_missing_tool_started(self, event: ProtocolEvent) -> None:
+        if self._mux is None:
+            return
+        data = event["params"]["data"]
+        if not _is_record(data) or data.get("event") != "started":
+            return
+        cause = data.get("cause")
+        if not _is_record(cause) or cause.get("type") != "toolCall":
+            return
+        tool_call_id = cause.get("tool_call_id")
+        if not isinstance(tool_call_id, str):
+            return
+        if tool_call_id in self._emitted_tool_starts:
+            return
+        known = self._known_tool_calls.get(tool_call_id)
+        if known is None:
+            return
+
+        self._emitted_tool_starts.add(tool_call_id)
+        namespace = event["params"]["namespace"]
+        self._mux.emit(
+            _copy_event(
+                event,
+                method="tools",
+                namespace=namespace[:-1],
+                data={
+                    "event": "tool-started",
+                    "tool_call_id": tool_call_id,
+                    "tool_name": known["tool_name"],
+                    "input": known["input"],
+                },
+            )
+        )
+
+
 class MessagesTransformer(StreamTransformer):
     """Capture messages events as ChatModelStream objects.
 
@@ -119,9 +304,19 @@ class MessagesTransformer(StreamTransformer):
 
     Native transformer — the `messages` projection is exposed as a
     direct attribute on the run stream.
+
+    `scope_exact = False`: matches events at the transformer's own
+    namespace **or** exactly one segment deeper (the chat-model /
+    node's own task ns). Mirrors JS's root-feed filter
+    (`namespaces=[[]], depth=1`) — root accepts depth-0 events plus
+    its own nodes' depth-1 tokens; subgraph mini-muxes accept their
+    own scope plus their internal nodes' tokens. Events deeper than
+    scope + 1 are dropped (the enclosing `SubgraphTransformer` has
+    already forwarded them to the matching child mini-mux).
     """
 
     _native = True
+    scope_exact = False
     required_stream_modes = ("messages",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
@@ -130,12 +325,17 @@ class MessagesTransformer(StreamTransformer):
         # Correlate protocol events back to a ChatModelStream by run_id
         # (attached to the event's metadata by StreamMessagesHandler).
         self._by_run: dict[str, ChatModelStream] = {}
+        self._started_blocks: dict[str, set[int]] = {}
+        self._mux: StreamMux | None = None
         self._pump_fn: Callable[[], bool] | None = None
         self._apump_fn: Callable[[], Awaitable[bool]] | None = None
 
     def init(self) -> dict[str, Any]:
         return {"messages": self._log}
 
+    def _on_register(self, mux: StreamMux) -> None:
+        self._mux = mux
+
     def _bind_pump(self, fn: Callable[[], bool]) -> None:
         """Wire the sync pull callback. Called by GraphRunStream._wire_request_more."""
         self._pump_fn = fn
@@ -186,16 +386,38 @@ class MessagesTransformer(StreamTransformer):
         )
 
     def process(self, event: ProtocolEvent) -> bool:
-        # Namespace filtering is handled by the mux via `scope_exact`.
         if event["method"] != "messages":
             return True
         params = event["params"]
+        # Accept events at our scope or exactly one segment deeper
+        # (the chat-model / node's own task ns). Deeper events belong
+        # to a subgraph and are routed by `SubgraphTransformer`.
+        ns = tuple(params["namespace"])
+        depth = len(self.scope)
+        if ns[:depth] != self.scope:
+            return True
 
-        payload, metadata = params["data"]
-        node: str | None = metadata.get("langgraph_node")
-        run_id = str(metadata.get("run_id", "")) if metadata else ""
+        raw_data = params["data"]
+        metadata: dict[str, Any] = {}
+        if isinstance(raw_data, tuple) and len(raw_data) == 2:
+            payload, raw_metadata = raw_data
+            metadata = raw_metadata if isinstance(raw_metadata, dict) else {}
+        else:
+            payload = raw_data
+        node = params.get("node")
+        if not isinstance(node, str):
+            node = metadata.get("langgraph_node")
+            if not isinstance(node, str):
+                node = None
+        raw_run_id = params.get("run_id", metadata.get("run_id"))
+        run_id = str(raw_run_id) if raw_run_id is not None else ""
 
         if isinstance(payload, dict) and "event" in payload:
+            self._repair_content_block_lifecycle(
+                event, cast("MessagesData", payload), run_id=run_id
+            )
+            if len(ns) > depth + 1:
+                return True
             self._route_protocol_event(
                 cast("MessagesData", payload), run_id=run_id, node=node
             )
@@ -216,23 +438,69 @@ class MessagesTransformer(StreamTransformer):
         run_id: str,
         node: str | None,
     ) -> None:
+        stream_event = _to_chat_model_stream_event(event)
         event_type = event.get("event")
         if event_type == "message-start":
-            message_id = event.get("message_id")
+            message_id = _message_event_id(event)
             stream = self._make_stream(
                 namespace=list(self.scope),
                 node=node,
-                message_id=str(message_id) if message_id is not None else None,
+                message_id=message_id,
             )
-            self._by_run[run_id] = stream
+            self._by_run[run_id or message_id or ""] = stream
             self._log.push(stream)
-            stream.dispatch(event)
+            stream.dispatch(stream_event)
         elif run_id in self._by_run:
             stream = self._by_run[run_id]
-            stream.dispatch(event)
+            stream.dispatch(stream_event)
             if event_type == "message-finish":
                 del self._by_run[run_id]
 
+    def _repair_content_block_lifecycle(
+        self,
+        source: ProtocolEvent,
+        event: MessagesData,
+        *,
+        run_id: str,
+    ) -> None:
+        if self._mux is None:
+            return
+        event_type = event.get("event")
+        key = _message_repair_key(source, run_id)
+        if event_type == "message-start":
+            self._started_blocks[key] = set()
+            return
+        if event_type == "content-block-start":
+            index = event.get("index")
+            if isinstance(index, int):
+                self._started_blocks.setdefault(key, set()).add(index)
+            return
+        if event_type in ("content-block-delta", "content-block-finish"):
+            index = event.get("index")
+            if not isinstance(index, int):
+                return
+            started = self._started_blocks.setdefault(key, set())
+            if index in started:
+                return
+            skeleton = _content_block_start_skeleton(event.get("content"))
+            if skeleton is None:
+                return
+            started.add(index)
+            self._mux.emit(
+                _copy_event(
+                    source,
+                    method="messages",
+                    namespace=list(source["params"]["namespace"]),
+                    data={
+                        "event": "content-block-start",
+                        "index": index,
+                        "content": skeleton,
+                    },
+                )
+            )
+        elif event_type == "message-finish":
+            self._started_blocks.pop(key, None)
+
     def _route_whole_message(self, message: BaseMessage, *, node: str | None) -> None:
         stream = self._make_stream(
             namespace=list(self.scope),
@@ -246,12 +514,14 @@ class MessagesTransformer(StreamTransformer):
     def finalize(self) -> None:
         """Clear any routing state — streams close themselves via `message-finish`."""
         self._by_run.clear()
+        self._started_blocks.clear()
 
     def fail(self, err: BaseException) -> None:
         """Propagate run error to any streams still open when the graph fails."""
         for stream in list(self._by_run.values()):
             stream.fail(err)
         self._by_run.clear()
+        self._started_blocks.clear()
 
 
 class SubgraphRunStream(BaseRunStream):
@@ -269,8 +539,11 @@ class SubgraphRunStream(BaseRunStream):
     Lifecycle fields update in place as events arrive:
 
     - `path`: the namespace tuple — stable for the life of the handle.
-    - `graph_name` / `trigger_call_id`: set once from the `started`
-      payload.
+    - `graph_name` / `cause`: set once from the `started` payload.
+      `cause` is populated by product-specific stream transformers
+      (see `LifecycleCause` in the protocol definition); pregel itself
+      emits no `cause`, so it may be `None` for subgraphs not covered
+      by a product transformer.
     - `status`: advances `started` → `running` → `completed` /
       `failed` / `interrupted`.
     - `error` / `checkpoint`: set on the terminal event when present.
@@ -287,12 +560,12 @@ class SubgraphRunStream(BaseRunStream):
         mux: StreamMux,
         *,
         graph_name: str | None = None,
-        trigger_call_id: str | None = None,
+        cause: LifecycleCause | None = None,
     ) -> None:
         super().__init__(mux)
         self.path: tuple[str, ...] = path
         self.graph_name: str | None = graph_name
-        self.trigger_call_id: str | None = trigger_call_id
+        self.cause: LifecycleCause | None = cause
         self.status: SubgraphStatus = "started"
         self.error: str | None = None
         self.checkpoint: CheckpointRef | None = None
@@ -411,7 +684,7 @@ class SubgraphTransformer(StreamTransformer):
             path=ns,
             mux=child_mux,
             graph_name=data.get("graph_name"),
-            trigger_call_id=data.get("trigger_call_id"),
+            cause=data.get("cause"),
         )
         self._by_ns[ns] = handle
         self._root_log.push(handle)

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

@@ -804,6 +804,30 @@ class TestConvertToProtocolEvent:
         assert isinstance(event["params"]["namespace"], list)
         assert event["params"]["namespace"] == ["a", "b", "c"]
 
+    def test_messages_conversion_uses_wire_shape(self) -> None:
+        part = {
+            "type": "messages",
+            "ns": ("call_model:task-1",),
+            "data": (
+                {
+                    "event": "content-block-delta",
+                    "index": 0,
+                    "content_block": {"type": "text", "text": "hi"},
+                },
+                {"langgraph_node": "call_model", "run_id": "run-1"},
+            ),
+        }
+        event = convert_to_protocol_event(part)
+        assert event["method"] == "messages"
+        assert event["params"]["namespace"] == ["call_model:task-1"]
+        assert event["params"]["node"] == "call_model"
+        assert event["params"]["run_id"] == "run-1"
+        assert event["params"]["data"] == {
+            "event": "content-block-delta",
+            "index": 0,
+            "content": {"type": "text", "text": "hi"},
+        }
+
 
 # ---------------------------------------------------------------------------
 # StreamMux unit tests
@@ -973,20 +997,24 @@ class TestMessagesTransformer:
         assert hasattr(items[0], "dispatch")
         assert items[0].message_id == "run-1"
 
-    def test_ignores_non_root_namespace(self) -> None:
-        """Namespace filtering is enforced by the mux via `scope_exact`."""
+    def test_ignores_deeper_than_scope_plus_one(self) -> None:
+        """Root MessagesTransformer accepts scope + 1 (matching JS's
+        `depth=1` root feed), so a depth-1 chat-model ns is kept, but
+        events nested two or more segments deep belong to a subgraph
+        and are dropped.
+        """
         mux = StreamMux([MessagesTransformer()], is_async=False)
         t = mux.transformer_by_key("messages")
         assert isinstance(t, MessagesTransformer)
         t._bind_pump(lambda: False)
         it = iter(t._log)
 
-        meta = {"langgraph_node": "llm", "run_id": "run-1"}
+        meta = {"langgraph_node": "llm", "run_id": "run-deep"}
         mux.push(
             _event(
                 "messages",
-                ({"event": "message-start", "message_id": "run-1"}, meta),
-                namespace=["sub"],
+                ({"event": "message-start", "message_id": "run-deep"}, meta),
+                namespace=["outer:task", "inner:task"],
             )
         )
         t._log.close()

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 str(stream.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 str(by_id["run-a"].text) == "aaaa"
+        assert str(by_id["run-b"].text) == "bbbb"
 
     def test_text_deltas_accumulated_on_stream(self) -> None:
         t, log = _make_sync_transformer()
@@ -257,6 +257,82 @@ class TestProtocolEventRouting:
         (stream,) = [*log._items]
         assert stream.node == "my_llm"
 
+    def test_wire_shape_routes_to_chat_model_stream(self) -> None:
+        t, log = _make_sync_transformer()
+        for evt in (
+            {"event": "message-start", "role": "ai", "id": "msg-1"},
+            {
+                "event": "content-block-start",
+                "index": 0,
+                "content": {"type": "text", "text": ""},
+            },
+            {
+                "event": "content-block-delta",
+                "index": 0,
+                "content": {"type": "text", "text": "hello"},
+            },
+            {
+                "event": "content-block-finish",
+                "index": 0,
+                "content": {"type": "text", "text": "hello"},
+            },
+            {"event": "message-finish", "reason": "stop"},
+        ):
+            t.process(
+                {
+                    "type": "event",
+                    "method": "messages",
+                    "params": {
+                        "namespace": [],
+                        "timestamp": TS,
+                        "data": evt,
+                        "node": "llm",
+                        "run_id": "run-1",
+                    },
+                }
+            )
+        (stream,) = [*log._items]
+        assert stream.message_id == "msg-1"
+        assert str(stream.text) == "hello"
+
+    def test_missing_content_block_start_is_synthesized_in_main_log(self) -> None:
+        mux = StreamMux([MessagesTransformer()], is_async=False)
+        events = iter(mux._events)
+
+        for evt in (
+            {"event": "message-start", "role": "ai", "id": "msg-1"},
+            {
+                "event": "content-block-delta",
+                "index": 0,
+                "content": {"type": "text", "text": "hello"},
+            },
+        ):
+            mux.push(
+                {
+                    "type": "event",
+                    "method": "messages",
+                    "params": {
+                        "namespace": [],
+                        "timestamp": TS,
+                        "data": evt,
+                        "run_id": "run-1",
+                    },
+                }
+            )
+        mux.close()
+
+        emitted = list(events)
+        assert [event["params"]["data"]["event"] for event in emitted] == [
+            "message-start",
+            "content-block-start",
+            "content-block-delta",
+        ]
+        assert emitted[1]["params"]["data"]["content"] == {
+            "type": "text",
+            "text": "",
+        }
+        assert emitted[0]["seq"] < emitted[1]["seq"] < emitted[2]["seq"]
+
 
 # ---------------------------------------------------------------------------
 # Non-streaming (whole AIMessage) fallback
@@ -270,7 +346,7 @@ class TestWholeMessageFallback:
         log.close()
         (stream,) = list(log._items)
         assert stream.done
-        assert stream.output.content == "the full answer"
+        assert str(stream.text) == "the full answer"
 
     def test_whole_message_has_full_lifecycle(self) -> None:
         t, log = _make_sync_transformer()
@@ -317,7 +393,12 @@ class TestFiltering:
         assert t.process(values_event) is True
 
     def test_subgraph_namespace_dropped(self) -> None:
-        """Root MessagesTransformer (via the mux) ignores non-root events."""
+        """Root MessagesTransformer accepts its scope + one segment (JS
+        `depth=1`), so a root-node chat-model ns (depth 1) is picked
+        up, but deeper subgraph chatter (depth 2+) is dropped —
+        `SubgraphTransformer` has already forwarded those into the
+        matching child mini-mux.
+        """
         from langgraph.stream._mux import StreamMux
 
         mux = StreamMux([MessagesTransformer()], is_async=False)
@@ -326,22 +407,34 @@ class TestFiltering:
         t._log._subscribed = True
         t._bind_pump(lambda: False)
 
-        mux.push(
-            {
-                "type": "event",
-                "method": "messages",
-                "params": {
-                    "namespace": ["subgraph"],
-                    "timestamp": TS,
-                    "data": (
-                        {"event": "message-start", "message_id": "run-x"},
-                        {"run_id": "run-x"},
-                    ),
-                },
-            }
-        )
+        def _push(ns: list[str], run_id: str) -> None:
+            mux.push(
+                {
+                    "type": "event",
+                    "method": "messages",
+                    "params": {
+                        "namespace": ns,
+                        "timestamp": TS,
+                        "data": (
+                            {
+                                "event": "message-start",
+                                "message_id": run_id,
+                            },
+                            {"run_id": run_id},
+                        ),
+                    },
+                }
+            )
+
+        # Depth 1 (a root-node chat model): accepted.
+        _push(["call_model:task-1"], run_id="run-root")
+        # Depth 2 (subgraph internal): dropped by the root transformer.
+        _push(["outer:task-1", "inner:task-2"], run_id="run-sub")
+
         t._log.close()
-        assert list(t._log._items) == []
+        streams = list(t._log._items)
+        assert len(streams) == 1
+        assert streams[0].message_id == "run-root"
 
 
 # ---------------------------------------------------------------------------
@@ -410,7 +503,7 @@ class TestAsyncMode:
             t.process(_proto_event(evt))
         (stream,) = list(log._items)
         msg = await stream.output
-        assert msg.content == "async"
+        assert msg.content == [{"type": "text", "text": "async", "index": 0}]
 
 
 # ---------------------------------------------------------------------------
@@ -471,7 +564,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.content == "mux stream"
+        assert str(stream.text) == "mux stream"
 
     def test_whole_message_via_mux(self) -> None:
         t = MessagesTransformer()
@@ -485,7 +578,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.content == "result"
+        assert str(stream.text) == "result"
 
     @pytest.mark.anyio
     async def test_async_streaming_via_mux(self) -> None:
@@ -501,7 +594,7 @@ class TestViaMux:
         streams = list(log._items)
         assert len(streams) == 1
         msg = await streams[0].output
-        assert msg.content == "async mux"
+        assert msg.content == [{"type": "text", "text": "async mux", "index": 0}]
         await mux.aclose()
 
 
@@ -545,7 +638,7 @@ class TestEndToEnd:
 
         assert len(streams) == 1
         assert isinstance(streams[0], ChatModelStream)
-        assert streams[0].output.content == "hello world"
+        assert str(streams[0].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 +681,7 @@ class TestEndToEnd:
         streams = list(run.messages)
 
         assert len(streams) == 1
-        assert streams[0].output.content == "hardcoded"
+        assert str(streams[0].text) == "hardcoded"
 
     @pytest.mark.anyio
     async def test_async_node_calling_astream_v2(self) -> None:
@@ -616,7 +709,7 @@ class TestEndToEnd:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.content == "async answer"
+        assert msg.content == [{"type": "text", "text": "async answer", "index": 0}]
 
     @pytest.mark.anyio
     async def test_nested_async_iteration_yields_text_deltas(self) -> None:
@@ -693,7 +786,7 @@ class TestEndToEndV2Invoke:
         )
         stream = streams[0]
         assert isinstance(stream, ChatModelStream)
-        assert stream.output.content == "hello world"
+        assert str(stream.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 +819,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 str(stream.text) == "streamed answer"
 
     def test_invoke_text_deltas_iterate_live(self) -> None:
         """`.text` projection yields deltas in order."""
@@ -774,7 +867,7 @@ class TestEndToEndV2Invoke:
         streams = list(run.messages)
 
         assert len(streams) == 2
-        contents = {s.output.content for s in streams}
+        contents = {str(s.text) for s in streams}
         assert contents == {"alpha", "beta"}
 
     def test_invoke_plus_constructed_message_two_streams(self) -> None:
@@ -805,9 +898,9 @@ class TestEndToEndV2Invoke:
 
         assert len(streams) == 2
         assert streams[0].node == "streaming_node"
-        assert streams[0].output.content == "live stream"
+        assert str(streams[0].text) == "live stream"
         assert streams[1].node == "constructed_node"
-        assert streams[1].output.content == "hardcoded"
+        assert str(streams[1].text) == "hardcoded"
         assert streams[1].message_id == "constructed-1"
 
     @pytest.mark.anyio
@@ -835,7 +928,7 @@ class TestEndToEndV2Invoke:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.content == "async invoke"
+        assert msg.content == [{"type": "text", "text": "async invoke", "index": 0}]
 
 
 class TestDirectMessagesModeStaysV1:

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -20,6 +20,7 @@ from langgraph.stream.transformers import (
     MessagesTransformer,
     SubgraphRunStream,
     SubgraphTransformer,
+    ToolLifecycleTransformer,
     ValuesTransformer,
 )
 from langgraph.types import interrupt
@@ -32,14 +33,14 @@ def _lifecycle(
     *,
     namespace: list[str] | None = None,
     graph_name: str | None = None,
-    trigger_call_id: str | None = None,
+    cause: dict[str, Any] | 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 cause is not None:
+        data["cause"] = cause
     if error is not None:
         data["error"] = error
     return {
@@ -75,7 +76,12 @@ def _subscribe(log: EventLog) -> None:
 # ---------------------------------------------------------------------------
 
 
-_FACTORIES = [ValuesTransformer, MessagesTransformer, SubgraphTransformer]
+_FACTORIES = [
+    ValuesTransformer,
+    ToolLifecycleTransformer,
+    MessagesTransformer,
+    SubgraphTransformer,
+]
 
 
 def _handle_values_items(handle: SubgraphRunStream) -> list:
@@ -125,16 +131,177 @@ class TestSubgraphTransformerUnit:
                 "started",
                 namespace=["task_a:child"],
                 graph_name="child",
-                trigger_call_id="task_a",
+                cause={"type": "toolCall", "tool_call_id": "call_abc"},
             )
         )
 
         handle = self._handle(transformer)
         assert handle.path == ("task_a:child",)
         assert handle.graph_name == "child"
-        assert handle.trigger_call_id == "task_a"
+        assert handle.cause == {"type": "toolCall", "tool_call_id": "call_abc"}
         assert handle.status == "started"
 
+    def test_tool_started_is_synthesized_before_tool_caused_lifecycle(self) -> None:
+        mux, transformer = self._mux()
+        events = iter(mux._events)
+        mux.push(
+            _values(
+                {
+                    "messages": [
+                        {
+                            "tool_calls": [
+                                {
+                                    "id": "call_abc",
+                                    "name": "task",
+                                    "args": {"subagent_type": "researcher"},
+                                }
+                            ]
+                        }
+                    ]
+                },
+                namespace=[],
+            )
+        )
+        mux.push(
+            _lifecycle(
+                "started",
+                namespace=["task:child"],
+                graph_name="child",
+                cause={"type": "toolCall", "tool_call_id": "call_abc"},
+            )
+        )
+
+        mux.close()
+        tool_started, lifecycle_started = list(events)[1:3]
+        assert tool_started["method"] == "tools"
+        assert tool_started["params"]["namespace"] == []
+        assert tool_started["params"]["data"] == {
+            "event": "tool-started",
+            "tool_call_id": "call_abc",
+            "tool_name": "task",
+            "input": {"subagent_type": "researcher"},
+        }
+        assert lifecycle_started["method"] == "lifecycle"
+        assert tool_started["seq"] < lifecycle_started["seq"]
+        assert self._handle(transformer).path == ("task:child",)
+
+    def test_core_golden_trace_uses_js_wire_shape_and_ordering(self) -> None:
+        mux, _transformer = self._mux()
+        events = iter(mux._events)
+        mux.push(
+            _values(
+                {
+                    "messages": [
+                        {
+                            "tool_calls": [
+                                {
+                                    "id": "call_abc",
+                                    "name": "task",
+                                    "args": {"subagent_type": "researcher"},
+                                }
+                            ]
+                        }
+                    ]
+                },
+                namespace=[],
+            )
+        )
+        for data in (
+            {"event": "message-start", "id": "msg-1", "role": "ai"},
+            {
+                "event": "content-block-delta",
+                "index": 0,
+                "content": {"type": "text", "text": "hi"},
+            },
+        ):
+            mux.push(
+                {
+                    "type": "event",
+                    "method": "messages",
+                    "params": {
+                        "namespace": ["call_model:task-1"],
+                        "timestamp": TS,
+                        "data": data,
+                        "run_id": "run-1",
+                    },
+                }
+            )
+        mux.push(
+            _lifecycle(
+                "started",
+                namespace=["task:child"],
+                graph_name="child",
+                cause={"type": "toolCall", "tool_call_id": "call_abc"},
+            )
+        )
+        mux.close()
+
+        trace = [
+            (event["method"], event["params"]["namespace"], event["params"]["data"])
+            for event in events
+        ]
+        assert trace == [
+            (
+                "values",
+                [],
+                {
+                    "messages": [
+                        {
+                            "tool_calls": [
+                                {
+                                    "id": "call_abc",
+                                    "name": "task",
+                                    "args": {"subagent_type": "researcher"},
+                                }
+                            ]
+                        }
+                    ]
+                },
+            ),
+            (
+                "messages",
+                ["call_model:task-1"],
+                {"event": "message-start", "id": "msg-1", "role": "ai"},
+            ),
+            (
+                "messages",
+                ["call_model:task-1"],
+                {
+                    "event": "content-block-start",
+                    "index": 0,
+                    "content": {"type": "text", "text": ""},
+                },
+            ),
+            (
+                "messages",
+                ["call_model:task-1"],
+                {
+                    "event": "content-block-delta",
+                    "index": 0,
+                    "content": {"type": "text", "text": "hi"},
+                },
+            ),
+            (
+                "tools",
+                [],
+                {
+                    "event": "tool-started",
+                    "tool_call_id": "call_abc",
+                    "tool_name": "task",
+                    "input": {"subagent_type": "researcher"},
+                },
+            ),
+            (
+                "lifecycle",
+                ["task:child"],
+                {
+                    "event": "started",
+                    "graph_name": "child",
+                    "cause": {"type": "toolCall", "tool_call_id": "call_abc"},
+                },
+            ),
+        ]
+
     def test_status_transitions(self) -> None:
         mux, transformer = self._mux()
         mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c"))
@@ -237,7 +404,14 @@ class TestSubgraphTransformerUnit:
             {
                 "type": "event",
                 "method": "messages",
-                "params": {"namespace": ["t:c"], "timestamp": TS, "data": "x"},
+                "params": {
+                    "namespace": ["t:c"],
+                    "timestamp": TS,
+                    "data": (
+                        {"event": "message-start", "message_id": "m1"},
+                        {"run_id": "m1"},
+                    ),
+                },
             }
         )
         assert list(transformer._root_log._items) == []
@@ -364,10 +538,10 @@ class TestSubgraphTransformerAsyncEndToEnd:
         assert child.status == "completed"
 
 
-class TestSubgraphTriggerCallId:
-    """Confirm `trigger_call_id` flows from real pregel metadata."""
+class TestSubgraphCause:
+    """Pregel core emits no `cause`; product transformers populate it."""
 
-    def test_trigger_call_id_populated_end_to_end(self) -> None:
+    def test_cause_not_populated_by_pregel(self) -> None:
         graph = _build_nested_graph()
         run = graph.stream_v2({"value": "", "items": []})
 
@@ -375,14 +549,15 @@ class TestSubgraphTriggerCallId:
         assert len(collected) == 1
         child = collected[0]
 
-        # The child's single-segment path encodes `node_name:task_id`.
-        # Both the parsed task_id (`trigger_call_id`) and the segment
-        # should match the same task_id suffix.
+        # The child's single-segment path still encodes `node_name:task_id`
+        # (that's pregel's internal namespace format), but `cause` is now
+        # product-agnostic and must be populated by a stream transformer,
+        # not by pregel itself.
         assert ":" in child.path[0]
         node_name, _, task_id = child.path[0].partition(":")
         assert node_name == "sub"
         assert task_id  # non-empty
-        assert child.trigger_call_id == task_id
+        assert child.cause is None
 
 
 class TestSubgraphInterrupt:

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.12"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" },
 ]
 
 [[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" },