Merge branch 'main' into nh/namespace-keyed-subagents

`metadata` reads more neutrally than `cause` — the field carries
descriptive info about what the lifecycle event is for, not a
strict causal relationship. Wire field, docstrings, and tests updated.

libs/langgraph/langgraph/stream/__init__.py

@@ -17,9 +17,9 @@ from langgraph.stream.transformers import (
     CheckpointsTransformer,
     CustomTransformer,
     DebugTransformer,
+    LifecycleEvent,
     LifecyclePayload,
     LifecycleTransformer,
-    SubgraphStatus,
     SubgraphTransformer,
     TasksTransformer,
     UpdatesTransformer,
@@ -32,13 +32,13 @@ __all__ = [
     "CustomTransformer",
     "DebugTransformer",
     "GraphRunStream",
+    "LifecycleEvent",
     "LifecyclePayload",
     "LifecycleTransformer",
     "ProtocolEvent",
     "StreamChannel",
     "StreamTransformer",
     "SubgraphRunStream",
-    "SubgraphStatus",
     "SubgraphTransformer",
     "TasksTransformer",
     "UpdatesTransformer",

libs/langgraph/langgraph/stream/run_stream.py

@@ -12,7 +12,7 @@ from langgraph.stream._mux import StreamMux
 from langgraph.stream._types import ProtocolEvent
 
 if TYPE_CHECKING:
-    from langgraph.stream.transformers import SubgraphStatus
+    from langgraph.stream.transformers import LifecycleEvent
 
 
 def _drive_until_done(pump: Callable[[], bool]) -> None:
@@ -523,8 +523,8 @@ class _SubgraphRunStreamMixin:
 
     path: tuple[str, ...]
     graph_name: str | None
-    trigger_call_id: str | None
-    status: SubgraphStatus
+    parent_task_id: str | None
+    status: LifecycleEvent
     error: str | None
     _seen_terminal: bool
 
@@ -538,7 +538,7 @@ class SubgraphRunStream(GraphRunStream, _SubgraphRunStreamMixin):
         *,
         path: tuple[str, ...],
         graph_name: str | None = None,
-        trigger_call_id: str | None = None,
+        parent_task_id: str | None = None,
     ) -> None:
         # Capture the parent-inherited pump before super().__init__
         # touches anything; we delegate to it from `_pump_next`.
@@ -550,7 +550,7 @@ class SubgraphRunStream(GraphRunStream, _SubgraphRunStreamMixin):
         )
         self.path = path
         self.graph_name = graph_name
-        self.trigger_call_id = trigger_call_id
+        self.parent_task_id = parent_task_id
         self.status = "started"
         self.error = None
         self._seen_terminal = False
@@ -581,7 +581,7 @@ class AsyncSubgraphRunStream(AsyncGraphRunStream, _SubgraphRunStreamMixin):
         *,
         path: tuple[str, ...],
         graph_name: str | None = None,
-        trigger_call_id: str | None = None,
+        parent_task_id: str | None = None,
     ) -> None:
         self._parent_apump_fn: Callable[[], Awaitable[bool]] | None = mux._apump_fn
         super().__init__(
@@ -591,7 +591,7 @@ class AsyncSubgraphRunStream(AsyncGraphRunStream, _SubgraphRunStreamMixin):
         )
         self.path = path
         self.graph_name = graph_name
-        self.trigger_call_id = trigger_call_id
+        self.parent_task_id = parent_task_id
         self.status = "started"
         self.error = None
         self._seen_terminal = False

libs/langgraph/langgraph/stream/transformers.py

@@ -327,11 +327,31 @@ class MessagesTransformer(StreamTransformer):
         self._by_run.clear()
 
 
-SubgraphStatus = Literal["started", "completed", "failed", "interrupted", "drained"]
+LifecycleEvent = Literal["started", "completed", "failed", "interrupted", "drained"]
+"""State transition surfaced on the `lifecycle` channel for a tracked subgraph.
+
+Each value:
+
+- `started` — first `tasks` event observed at the tracked namespace. Carries
+  `graph_name` / `parent_task_id` / optional `metadata` describing what spawned
+  the subgraph.
+- `completed` — the dispatching task's `TaskResultPayload` arrived with neither
+  error nor interrupts. Also emitted by `finalize` for any tracked namespace
+  still open at run end.
+- `failed` — the dispatching task's result carried an `error`, OR the run
+  failed at top level with a non-interrupt / non-drain exception. Carries
+  `error` (string).
+- `interrupted` — the dispatching task's result carried `interrupts` (takes
+  precedence over `error` if both present), OR the run failed with
+  `GraphInterrupt`.
+- `drained` — the run was cooperatively stopped at a superstep boundary via
+  `RunControl.request_drain()` (e.g. SIGTERM). The checkpoint is saved and
+  the run is resumable.
+"""
 
 
 def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
-    """Split a namespace segment into `(graph_name, trigger_call_id)`.
+    """Split a namespace segment into `(graph_name, parent_task_id)`.
 
     Segments are formatted `node_name:task_id` by `prepare_next_tasks`.
     Returns `(segment, None)` if no `:` is present.
@@ -340,6 +360,36 @@ def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
     return name, task_id if sep else None
 
 
+def _extract_dispatching_tool_call_id(payload: Any) -> str | None:
+    """Return the model-side `tool_call_id` from a per-call dispatched task's
+    `input`, or `None` if the payload doesn't match a recognised shape.
+
+    Two shapes are recognised; both are duck-typed so any tool runner
+    that mimics the layout participates without naming any specific
+    dispatcher's types:
+
+    1. Single-element list of tool-call dicts:
+       `[{"id": ..., "name": ..., "args": {...}}]`. The current public
+       shape — `langchain.agents.create_agent` Send-fans this out per
+       pending tool call.
+    2. Dict envelope wrapping a tool call:
+       `{"tool_call": {"id": ..., "args": {...}, ...}, ...}`. Older
+       prebuilt agent paths Send-fan-out this shape.
+    """
+    if isinstance(payload, dict):
+        tool_call = payload.get("tool_call")
+        if not isinstance(tool_call, dict):
+            return None
+    elif (
+        isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict)
+    ):
+        tool_call = payload[0]
+    else:
+        return None
+    raw_id = tool_call.get("id")
+    return raw_id if isinstance(raw_id, str) else None
+
+
 class LifecyclePayload(TypedDict, total=False):
     """Payload of a lifecycle event surfaced on the `lifecycle` channel.
 
@@ -349,11 +399,54 @@ class LifecyclePayload(TypedDict, total=False):
     `run.lifecycle`.
     """
 
-    event: SubgraphStatus
+    event: LifecycleEvent
+    """State transition. See `LifecycleEvent` for per-value semantics."""
     namespace: list[str]
+    """Checkpoint namespace of the subgraph the event is about. Always present.
+
+    A list of `node_name:task_id` segments, one per nesting level (root has
+    `[]`, a direct child of root has `["agent:abc123"]`, a grandchild has
+    `["agent:abc123", "tool:def456"]`, etc.). Stable identity across the
+    `started → terminal` pair for the same subgraph instance.
+    """
     graph_name: NotRequired[str]
-    trigger_call_id: NotRequired[str]
+    """Name of the parent-scope node that dispatched this subgraph
+    (`add_node` name, surrounding tool's name for in-tool invokes,
+    `Send` target name, etc.) — parsed from the namespace tail
+    segment. Absent when the segment has no `:` separator.
+    """
+    parent_task_id: str
+    """Pregel task id of the dispatching task — the task whose execution
+    spawned this subgraph.
+
+    Always present on every event for the same subgraph instance. This is
+    the join key for correlating `started` ↔ terminal events and for
+    matching a `started` back to its `tasks` parent. Each Send produces
+    its own pregel task with its own id, so the join is 1:1 even when a
+    model dispatches multiple parallel tool calls in one turn.
+    """
+    metadata: NotRequired[dict[str, Any]]
+    """Optional generic descriptor of *what triggered* this subgraph.
+    Forwarded by protocol layers as the wire `lifecycle.started.metadata` field.
+
+    Shape:
+
+    - `{"type": "tool_call", "tool_call_id": "<id>"}` — set when the subgraph
+      was triggered by a per-call tool dispatch (a model tool call routed
+      through whatever tool node the agent uses). `tool_call_id` is the
+      model-side id of the originating tool call, exposed so UI consumers
+      can anchor the lifecycle event back to the AI message that dispatched
+      it. The langgraph layer deliberately doesn't mine `args` — those live
+      on the AIMessage's `tool_calls[i].args` already, and consumers that
+      want descriptive intent (subagent type, prompt text, etc.) look it
+      up there to keep one source of truth.
+
+    Absent for structurally-triggered subgraphs (parallel branches via
+    `Send` with non-tool-call payloads, nested `graph.invoke()`, etc.) and
+    for tool dispatches whose envelope carried no `id`.
+    """
     error: NotRequired[str]
+    """Error string. Set on `failed` events; absent otherwise."""
 
 
 class _TasksLifecycleBase(StreamTransformer):
@@ -371,13 +464,13 @@ class _TasksLifecycleBase(StreamTransformer):
 
     - `_should_track(ns)` — scope filter (e.g. multi-depth vs
       direct-children-only).
-    - `_on_started(ns, graph_name, trigger_call_id)` — first sighting
-      action (push payload / build handle / etc.). Called once per
-      discovered namespace.
-    - `_on_terminal(ns, status, error)` — terminal action (push
-      terminal payload / mark handle status). Called once per
-      tracked namespace at result time, or via `finalize` / `fail`
-      sweeps if no parent result arrived.
+    - `_on_started(ns, graph_name, parent_task_id, tool_call_id)` —
+      first sighting action (push payload / build handle / etc.).
+      Called once per discovered namespace.
+    - `_on_terminal(ns, status, error, parent_task_id)` — terminal
+      action (push terminal payload / mark handle status). Called
+      once per tracked namespace at result time, or via `finalize` /
+      `fail` sweeps if no parent result arrived.
 
     Tasks events are suppressed from the main event log (`process`
     returns False) — they're folded into whichever projection the
@@ -390,9 +483,15 @@ class _TasksLifecycleBase(StreamTransformer):
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
         super().__init__(scope)
         self._seen: set[tuple[str, ...]] = set()
-        # Maps tracked namespace -> task_id of the parent task whose
+        # Maps tracked namespace -> task_id of the dispatching task whose
         # `TaskResultPayload` will close it.
         self._open: dict[tuple[str, ...], str] = {}
+        # Maps task_id -> model-side `tool_call_id` for tasks whose `input`
+        # matched a recognized per-call tool-dispatch shape. The lifecycle
+        # hook joins on this when a child subgraph fires its first task
+        # event so it can anchor the lifecycle.started to the originating
+        # AI message tool call.
+        self._dispatching_tool_call_id: dict[str, str] = {}
 
     # --- Template-method hooks (subclass overrides) ---
 
@@ -404,19 +503,33 @@ class _TasksLifecycleBase(StreamTransformer):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,
     ) -> None:
-        """Fired once per discovered namespace (first observed task event)."""
+        """Fired once per discovered namespace (first observed task event).
+
+        `tool_call_id` is the model-side id of the originating tool call
+        (from the per-call dispatched task's `input`). `None` for
+        structurally-triggered subgraphs or per-call envelopes that omitted
+        an `id`. Consumers join on `parent_task_id` (the pregel task id)
+        for identity; `tool_call_id` is purely an anchor back to the AI
+        message that dispatched the subgraph.
+        """
         raise NotImplementedError
 
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,
     ) -> None:
-        """Fired once per tracked namespace when its parent's result arrives,
-        or via finalize/fail safety-net sweeps.
+        """Fired once per tracked namespace when its dispatching task's
+        result arrives, or via finalize/fail safety-net sweeps.
+
+        `parent_task_id` is the same id paired with the namespace at
+        `_on_started` time, so subscribers can correlate the terminal
+        event back to its `started`.
         """
         raise NotImplementedError
 
@@ -430,56 +543,96 @@ class _TasksLifecycleBase(StreamTransformer):
         if "result" in data:
             self._handle_task_result(ns, data)
         else:
-            self._handle_task_start(ns)
+            self._handle_task_start(ns, data)
         # Tasks events are folded into the synthesized projections;
         # suppress from the main event log so iterators don't double-see
         # the same information in two shapes.
         return False
 
-    def _handle_task_start(self, ns: tuple[str, ...]) -> None:
+    def _handle_task_start(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
+        # Mine input shape on every tasks event (not just tracked ones)
+        # so we capture dispatching tasks that themselves live outside the
+        # tracked region but whose `id` will appear as `parent_task_id`
+        # for a child subgraph.
+        self._record_dispatching_tool_call_id(data)
         if not self._should_track(ns) or ns in self._seen:
             return
         self._seen.add(ns)
-        graph_name, trigger_call_id = _parse_ns_segment(ns[-1])
-        self._on_started(ns, graph_name or None, trigger_call_id)
-        if trigger_call_id is not None:
-            self._open[ns] = trigger_call_id
+        graph_name, parent_task_id = _parse_ns_segment(ns[-1])
+        tool_call_id = (
+            self._dispatching_tool_call_id.pop(parent_task_id, None)
+            if parent_task_id is not None
+            else None
+        )
+        self._on_started(
+            ns,
+            graph_name or None,
+            parent_task_id,
+            tool_call_id,
+        )
+        if parent_task_id is not None:
+            self._open[ns] = parent_task_id
+
+    def _record_dispatching_tool_call_id(self, data: dict[str, Any]) -> None:
+        """Remember `task_id -> tool_call_id` if the task input matches
+        a recognized per-call tool-dispatch shape.
+
+        Shape detection and id extraction both live in
+        `_extract_dispatching_tool_call_id`; this method just records the
+        mapping under the dispatching task's own `id` so the lifecycle hook
+        can anchor a child subgraph back to the originating AI message
+        tool call when that subgraph's first task event arrives.
+        """
+        task_id = data.get("id")
+        if not isinstance(task_id, str):
+            return
+        tool_call_id = _extract_dispatching_tool_call_id(data.get("input"))
+        if tool_call_id is None:
+            return
+        self._dispatching_tool_call_id[task_id] = tool_call_id
 
     def _pop_terminal_transitions(
         self, ns: tuple[str, ...], data: dict[str, Any]
-    ) -> list[tuple[tuple[str, ...], SubgraphStatus, str | None]]:
-        """Return and remove tracked children closed by this task result."""
+    ) -> list[tuple[tuple[str, ...], LifecycleEvent, str | None, str]]:
+        """Return and remove tracked children closed by this task result.
+
+        Each tuple is `(child_ns, status, error, parent_task_id)`.
+        `parent_task_id` is the dispatching task's id — the same id
+        we'd already paired with the namespace at `_on_started`.
+        """
         result_id = data.get("id")
         if not result_id:
             return []
-        transitions: list[tuple[tuple[str, ...], SubgraphStatus, str | None]] = []
-        for child_ns, parent_task_id in list(self._open.items()):
-            if child_ns[:-1] != ns or parent_task_id != result_id:
+        transitions: list[tuple[tuple[str, ...], LifecycleEvent, str | None, str]] = []
+        for child_ns, dispatching_task_id in list(self._open.items()):
+            if child_ns[:-1] != ns or dispatching_task_id != result_id:
                 continue
             status, error = _terminal_from_result(data)
-            transitions.append((child_ns, status, error))
+            transitions.append((child_ns, status, error, dispatching_task_id))
             del self._open[child_ns]
         return transitions
 
     def _handle_task_result(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
-        for child_ns, status, error in self._pop_terminal_transitions(ns, data):
-            self._on_terminal(child_ns, status, error)
+        for child_ns, status, error, parent_task_id in self._pop_terminal_transitions(
+            ns, data
+        ):
+            self._on_terminal(child_ns, status, error, parent_task_id)
 
     def finalize(self) -> None:
         """Emit `completed` for any tracked namespace still open at run end."""
-        for ns in list(self._open):
-            self._on_terminal(ns, "completed", None)
+        for ns, parent_task_id in list(self._open.items()):
+            self._on_terminal(ns, "completed", None, parent_task_id)
         self._open.clear()
 
     def fail(self, err: BaseException) -> None:
         """Emit terminal status for any tracked namespace still open."""
         status, error_str = _status_from_exception(err)
-        for ns in list(self._open):
-            self._on_terminal(ns, status, error_str)
+        for ns, parent_task_id in list(self._open.items()):
+            self._on_terminal(ns, status, error_str, parent_task_id)
         self._open.clear()
 
 
-def _status_from_exception(err: BaseException) -> tuple[SubgraphStatus, str | None]:
+def _status_from_exception(err: BaseException) -> tuple[LifecycleEvent, str | None]:
     """Map a run exception to a subgraph terminal status and error string."""
     if isinstance(err, GraphDrained):
         return "drained", None
@@ -490,7 +643,7 @@ def _status_from_exception(err: BaseException) -> tuple[SubgraphStatus, str | No
 
 def _terminal_from_result(
     payload: dict[str, Any],
-) -> tuple[SubgraphStatus, str | None]:
+) -> tuple[LifecycleEvent, str | None]:
     """Map a `TaskResultPayload` to a `(status, error)` pair.
 
     Order matters: a result with both `error` and `interrupts` prefers
@@ -539,26 +692,37 @@ class LifecycleTransformer(_TasksLifecycleBase):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,
     ) -> None:
-        if trigger_call_id is None:
-            # Without a task id we can't correlate a parent-result
+        if parent_task_id is None:
+            # Without a task id we can't correlate a dispatching-task-result
             # event back to this namespace — skip the started payload
             # and rely on finalize/fail to close.
             return
-        payload: LifecyclePayload = {"event": "started", "namespace": list(ns)}
+        payload: LifecyclePayload = {
+            "event": "started",
+            "namespace": list(ns),
+            "parent_task_id": parent_task_id,
+        }
         if graph_name:
             payload["graph_name"] = graph_name
-        payload["trigger_call_id"] = trigger_call_id
+        if tool_call_id is not None:
+            payload["metadata"] = {"type": "tool_call", "tool_call_id": tool_call_id}
         self._channel.push(payload)
 
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,
     ) -> None:
-        payload: LifecyclePayload = {"event": status, "namespace": list(ns)}
+        payload: LifecyclePayload = {
+            "event": status,
+            "namespace": list(ns),
+            "parent_task_id": parent_task_id,
+        }
         if error is not None:
             payload["error"] = error
         self._channel.push(payload)
@@ -611,7 +775,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
         self,
         ns: tuple[str, ...],
         graph_name: str | None,
-        trigger_call_id: str | None,
+        parent_task_id: str | None,
+        tool_call_id: str | None = None,  # noqa: ARG002
     ) -> None:
         if self._mux is None:
             return
@@ -624,7 +789,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
             mux=child_mux,
             path=ns,
             graph_name=graph_name,
-            trigger_call_id=trigger_call_id,
+            parent_task_id=parent_task_id,
         )
         self._handles[ns] = handle
         self._log.push(handle)
@@ -632,8 +797,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _on_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,  # noqa: ARG002
     ) -> None:
         handle = self._handles.get(ns)
         if handle is None or not self._mark_terminal(handle, status, error):
@@ -643,8 +809,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
     async def _aon_terminal(
         self,
         ns: tuple[str, ...],
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
+        parent_task_id: str,  # noqa: ARG002
     ) -> None:
         handle = self._handles.get(ns)
         if handle is None or not self._mark_terminal(handle, status, error):
@@ -654,7 +821,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _mark_terminal(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> bool:
         """Mark a handle terminal once. Returns True on first transition."""
@@ -669,7 +836,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     def _close_or_fail_handle(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> None:
         if handle._mux is None or handle._mux._events._closed:
@@ -682,7 +849,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
     async def _aclose_or_fail_handle(
         self,
         handle: SubgraphRunStream | AsyncSubgraphRunStream,
-        status: SubgraphStatus,
+        status: LifecycleEvent,
         error: str | None,
     ) -> None:
         if handle._mux is None or handle._mux._events._closed:
@@ -721,10 +888,15 @@ class SubgraphTransformer(_TasksLifecycleBase):
             ns = tuple(event["params"]["namespace"])
             data = event["params"]["data"]
             if "result" in data:
-                for child_ns, status, error in self._pop_terminal_transitions(ns, data):
-                    await self._aon_terminal(child_ns, status, error)
+                for (
+                    child_ns,
+                    status,
+                    error,
+                    parent_task_id,
+                ) in self._pop_terminal_transitions(ns, data):
+                    await self._aon_terminal(child_ns, status, error, parent_task_id)
             else:
-                self._handle_task_start(ns)
+                self._handle_task_start(ns, data)
             keep = False
         else:
             keep = True
@@ -736,9 +908,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
 
     def _complete_open_handles(self) -> BaseException | None:
         first_error: BaseException | None = None
-        for ns in list(self._open):
+        for ns, parent_task_id in list(self._open.items()):
             try:
-                self._on_terminal(ns, "completed", None)
+                self._on_terminal(ns, "completed", None, parent_task_id)
             except BaseException as e:
                 if first_error is None:
                     first_error = e
@@ -754,9 +926,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
 
     async def _acomplete_open_handles(self) -> BaseException | None:
         first_error: BaseException | None = None
-        for ns in list(self._open):
+        for ns, parent_task_id in list(self._open.items()):
             try:
-                await self._aon_terminal(ns, "completed", None)
+                await self._aon_terminal(ns, "completed", None, parent_task_id)
             except BaseException as e:
                 if first_error is None:
                     first_error = e

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -35,8 +35,16 @@ def _tasks_start(
     *,
     task_id: str,
     name: str,
+    input: Any = None,
 ) -> dict[str, Any]:
-    """Build a `tasks` ProtocolEvent carrying a TaskPayload (start)."""
+    """Build a `tasks` ProtocolEvent carrying a TaskPayload (start).
+
+    Pass `input=[{"id": ..., "name": ..., "args": {...}}]` (the per-call
+    list shape `langchain.agents.create_agent` Send-fans out) or
+    `input={"tool_call": {"id": ..., ...}, ...}` (the dict envelope older
+    prebuilt agent paths emit) to exercise the lifecycle transformer's
+    `tool_call_id` mining for `lifecycle.started.metadata`.
+    """
     return {
         "type": "event",
         "method": "tasks",
@@ -46,7 +54,7 @@ def _tasks_start(
             "data": {
                 "id": task_id,
                 "name": name,
-                "input": None,
+                "input": input,
                 "triggers": [],
             },
         },
@@ -123,7 +131,220 @@ def test_started_emitted_on_first_direct_child_task() -> None:
     assert payload["event"] == "started"
     assert payload["namespace"] == ["agent:abc123"]
     assert payload["graph_name"] == "agent"
-    assert payload["trigger_call_id"] == "abc123"
+    assert payload["parent_task_id"] == "abc123"
+
+
+def test_started_carries_metadata_for_dict_envelope_input() -> None:
+    """When the dispatching task's `input` is a dict envelope with a
+    `tool_call` field (the shape older prebuilt agent paths Send-fan
+    out per call), the transformer mines `tool_call_id` from
+    `tool_call.id` and remembers it keyed by the dispatching task id.
+    When that task triggers a subgraph (the child's namespace ends in
+    `name:<dispatching_task_id>`), `lifecycle.started.metadata` carries
+    `{"type": "tool_call", "tool_call_id": ...}`. Identity correlation
+    still uses `parent_task_id`; `tool_call_id` is exposed so UI
+    consumers can anchor the lifecycle event back to the originating
+    AI message tool call. Args are deliberately NOT mined — they live
+    on the AIMessage and have a single source of truth there.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_xyz",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["event"] == "started"
+    assert payload["parent_task_id"] == "abc123"
+    assert payload["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_xyz",
+    }
+
+
+def test_started_carries_metadata_for_list_shape_per_call_input() -> None:
+    """`langchain.agents.create_agent` Send-fans out a per-call task
+    whose `input` is a single-element list of tool-call dicts:
+    `[{"id": ..., "name": ..., "args": {...}}]`. The transformer mines
+    `tool_call_id` exactly as for the dict envelope shape, so
+    `lifecycle.started.metadata` fires regardless of which agent factory
+    drove the dispatch.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[
+                {
+                    "id": "tc-1",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            ],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["event"] == "started"
+    assert payload["parent_task_id"] == "abc123"
+    assert payload["metadata"] == {"type": "tool_call", "tool_call_id": "tc-1"}
+
+
+def test_started_carries_metadata_when_args_absent() -> None:
+    """`tool_call_id` is the only field metadata needs; the dispatching
+    envelope can omit `args` entirely (or have non-dict args) and we
+    still produce a metadata as long as `id` is a string."""
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[{"id": "tc-1", "name": "some_tool"}],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["metadata"] == {"type": "tool_call", "tool_call_id": "tc-1"}
+
+
+def test_list_shape_ignored_when_not_single_element() -> None:
+    """Only single-element lists are recognized as the per-call shape;
+    a 0- or 2+-element list is some other batched/multi-call payload
+    and must not be mined."""
+    # Two-element list — not the per-call shape.
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="abc123",
+            name="tools",
+            input=[
+                {"id": "tc-1", "name": "task"},
+                {"id": "tc-2", "name": "task"},
+            ],
+        )
+    )
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
+
+    # Empty list.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start([], task_id="def456", name="tools", input=[]))
+    mux2.push(_tasks_start(["agent:def456"], task_id="t1", name="model"))
+    [payload2] = _drain_lifecycle(mux2)
+    assert "metadata" not in payload2
+
+
+def test_list_shape_robust_to_non_dict_or_missing_id() -> None:
+    """Duck-typing safety: a single-element list whose element isn't a
+    dict, or whose dict has no string `id`, must not raise — it just
+    leaves `metadata` absent."""
+    # Element is not a dict.
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start([], task_id="t-a", name="tools", input=["not-a-dict"]))
+    mux.push(_tasks_start(["agent:t-a"], task_id="t1", name="model"))
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
+
+    # Element has no `id`.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start([], task_id="t-b", name="tools", input=[{"name": "task"}]))
+    mux2.push(_tasks_start(["agent:t-b"], task_id="t1", name="model"))
+    [payload2] = _drain_lifecycle(mux2)
+    assert "metadata" not in payload2
+
+    # Element's `id` is not a string.
+    mux3 = _build_lifecycle_mux()
+    mux3.push(_tasks_start([], task_id="t-c", name="tools", input=[{"id": 123}]))
+    mux3.push(_tasks_start(["agent:t-c"], task_id="t1", name="model"))
+    [payload3] = _drain_lifecycle(mux3)
+    assert "metadata" not in payload3
+
+
+def test_parallel_dispatches_attributed_to_correct_parent() -> None:
+    """Two dispatching task envelopes in the same model turn each fan
+    out to their own child subgraph; each child's `metadata.tool_call_id`
+    must reflect its own dispatching envelope, not the other.
+
+    Defends the `parent_task_id` (pregel task id) join: that id is
+    parsed from the child namespace segment and is unique per Send,
+    so it disambiguates parallel dispatches 1:1. Both children share
+    the same `subagent_type` (in args, not on metadata) — only the
+    pregel task id can tell them apart, so the `tool_call_id` must
+    follow the pregel id, not anything from `args`.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="parent_A",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_1",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="parent_B",
+            name="tools",
+            input={
+                "tool_call": {
+                    "id": "call_2",
+                    "name": "task",
+                    "args": {"subagent_type": "researcher"},
+                }
+            },
+        )
+    )
+    mux.push(_tasks_start(["agent:parent_A"], task_id="t1", name="model"))
+    mux.push(_tasks_start(["agent:parent_B"], task_id="t2", name="model"))
+
+    payloads = _drain_lifecycle(mux)
+    by_ns = {tuple(p["namespace"]): p for p in payloads}
+    assert by_ns[("agent:parent_A",)]["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_1",
+    }
+    assert by_ns[("agent:parent_B",)]["metadata"] == {
+        "type": "tool_call",
+        "tool_call_id": "call_2",
+    }
+
+
+def test_started_omits_metadata_for_structurally_triggered_subgraph() -> None:
+    """Subgraphs triggered without a recognizable tool-call envelope on
+    the parent's input (Send with custom payloads, plain nested
+    `graph.invoke`, etc.) don't get a `metadata` field on
+    `lifecycle.started`."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="tool"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert "metadata" not in payload
 
 
 def test_started_dedup_on_repeat_namespace() -> None:
@@ -188,8 +409,12 @@ def test_completed_on_parent_task_result() -> None:
     mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
     mux.push(_tasks_result([], task_id="abc", name="agent"))
 
-    events = [p["event"] for p in _drain_lifecycle(mux)]
-    assert events == ["started", "completed"]
+    payloads = _drain_lifecycle(mux)
+    assert [p["event"] for p in payloads] == ["started", "completed"]
+    # `parent_task_id` is required on every event for the same subgraph
+    # so consumers can correlate `started` ↔ terminal without joining
+    # via `namespace`.
+    assert all(p["parent_task_id"] == "abc" for p in payloads)
 
 
 def test_failed_on_parent_task_result_with_error() -> None:
@@ -265,6 +490,54 @@ def test_fail_emits_failed_for_other_exceptions() -> None:
     assert payloads[1]["error"] == "boom"
 
 
+def test_parent_task_id_present_on_every_terminal_path() -> None:
+    """Every exit path that emits a terminal event (parent-result with
+    error / interrupts, finalize sweep, fail sweep) must carry
+    `parent_task_id` so consumers can correlate the terminal event
+    back to its `started` without falling back to namespace joins."""
+    # Path 1: parent-result with error.
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
+    mux.push(_tasks_result([], task_id="abc", name="agent", error="boom"))
+    [_, terminal] = _drain_lifecycle(mux)
+    assert terminal["event"] == "failed"
+    assert terminal["parent_task_id"] == "abc"
+
+    # Path 2: parent-result with interrupts.
+    mux2 = _build_lifecycle_mux()
+    mux2.push(_tasks_start(["agent:def"], task_id="t1", name="tool"))
+    mux2.push(
+        _tasks_result([], task_id="def", name="agent", interrupts=[{"value": "pause"}])
+    )
+    [_, terminal2] = _drain_lifecycle(mux2)
+    assert terminal2["event"] == "interrupted"
+    assert terminal2["parent_task_id"] == "def"
+
+    # Path 3: finalize sweep (no parent result arrived).
+    mux3 = _build_lifecycle_mux()
+    mux3.push(_tasks_start(["agent:ghi"], task_id="t1", name="tool"))
+    mux3.close()
+    [_, terminal3] = _drain_lifecycle(mux3)
+    assert terminal3["event"] == "completed"
+    assert terminal3["parent_task_id"] == "ghi"
+
+    # Path 4: fail sweep with GraphInterrupt.
+    mux4 = _build_lifecycle_mux()
+    mux4.push(_tasks_start(["agent:jkl"], task_id="t1", name="tool"))
+    mux4.fail(GraphInterrupt())
+    [_, terminal4] = _drain_lifecycle(mux4)
+    assert terminal4["event"] == "interrupted"
+    assert terminal4["parent_task_id"] == "jkl"
+
+    # Path 5: fail sweep with generic exception.
+    mux5 = _build_lifecycle_mux()
+    mux5.push(_tasks_start(["agent:mno"], task_id="t1", name="tool"))
+    mux5.fail(RuntimeError("kaboom"))
+    [_, terminal5] = _drain_lifecycle(mux5)
+    assert terminal5["event"] == "failed"
+    assert terminal5["parent_task_id"] == "mno"
+
+
 def test_unrelated_methods_pass_through() -> None:
     """Non-`tasks` events are not consumed and don't emit lifecycle."""
     mux = _build_lifecycle_mux()

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -195,7 +195,7 @@ def test_handle_created_on_first_direct_child_task() -> None:
     [handle] = _drain_subgraphs(mux)
     assert handle.path == ("agent:abc",)
     assert handle.graph_name == "agent"
-    assert handle.trigger_call_id == "abc"
+    assert handle.parent_task_id == "abc"
     assert handle.status == "started"
     _child_mux(handle)  # mini-mux backed