feat(langgraph): name tool-dispatched subagents via lc_agent_name (#7928)

libs/langgraph/langgraph/_internal/_config.py

@@ -402,3 +402,17 @@ _PROPAGATE_TO_METADATA = frozenset(
         "graph_id",
     )
 )
+
+
+def filter_to_user_tags(tags: Sequence[str] | None) -> list[str] | None:
+    """Drop langgraph's internal `seq:step:*` bookkeeping tags.
+
+    `seq:step:N` tags are added internally to mark sequence steps; everything
+    else (user-supplied tags and any other framework tags) is kept. Returns the
+    surviving tags, or `None` if none remain. Shared by the `messages` and
+    `tasks` stream handlers so both surface the same tag set on their metadata.
+    """
+    if not tags:
+        return None
+    filtered = [t for t in tags if not t.startswith("seq:step")]
+    return filtered or None

libs/langgraph/langgraph/pregel/_messages.py

@@ -15,6 +15,7 @@ from langchain_core.messages.utils import convert_to_messages
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
 from pydantic import BaseModel
 
+from langgraph._internal._config import filter_to_user_tags
 from langgraph._internal._constants import NS_SEP
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
 from langgraph.pregel.protocol import StreamChunk
@@ -143,9 +144,8 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             ]
             if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
                 return
-            if tags:
-                if filtered_tags := [t for t in tags if not t.startswith("seq:step")]:
-                    metadata["tags"] = filtered_tags
+            if (filtered_tags := filter_to_user_tags(tags)) is not None:
+                metadata["tags"] = filtered_tags
             self.metadata[run_id] = (ns, metadata)
 
     def on_llm_new_token(

libs/langgraph/langgraph/pregel/debug.py

@@ -6,9 +6,13 @@ from typing import Any
 from uuid import UUID
 
 from langchain_core.runnables import RunnableConfig
-from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
+from langgraph.checkpoint.base import (
+    EXCLUDED_METADATA_KEYS,
+    CheckpointMetadata,
+    PendingWrite,
+)
 
-from langgraph._internal._config import patch_checkpoint_map
+from langgraph._internal._config import filter_to_user_tags, patch_checkpoint_map
 from langgraph._internal._constants import (
     CONF,
     CONFIG_KEY_CHECKPOINT_NS,
@@ -40,12 +44,31 @@ def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPaylo
         if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
             continue
 
-        yield {
+        payload: TaskPayload = {
             "id": task.id,
             "name": task.name,
             "input": task.input,
             "triggers": task.triggers,
         }
+        # Forward user-meaningful metadata only — drop langgraph's internal
+        # framework keys (langgraph_node/step/triggers/path/checkpoint_ns,
+        # thread_id, ...), which are redundant with the task's own fields and
+        # namespace. Keys like `lc_agent_name`, `ls_integration`, and any
+        # user-supplied metadata ride along. Filtered config tags are folded in
+        # under `tags`, mirroring the messages stream handler. (The comprehension
+        # also yields a fresh dict, so mutating `md` doesn't touch task.config.)
+        if task.config is not None:
+            md = {
+                k: v
+                for k, v in (task.config.get("metadata") or {}).items()
+                if k not in EXCLUDED_METADATA_KEYS
+            }
+            filtered_tags = filter_to_user_tags(task.config.get("tags"))
+            if filtered_tags is not None:
+                md["tags"] = filtered_tags
+            if md:
+                payload["metadata"] = md
+        yield payload
 
 
 def is_multiple_channel_write(value: Any) -> bool:

libs/langgraph/langgraph/stream/transformers.py

@@ -9,7 +9,7 @@ from langchain_core.language_models.chat_model_stream import (
     ChatModelStream,
 )
 from langchain_core.messages import AIMessageChunk, BaseMessage, ToolMessage
-from langchain_protocol.protocol import MessagesData
+from langchain_protocol.protocol import LifecycleCause, MessagesData
 from typing_extensions import NotRequired, TypedDict
 
 from langgraph.errors import GraphDrained, GraphInterrupt
@@ -366,6 +366,7 @@ class LifecyclePayload(TypedDict, total=False):
     namespace: list[str]
     graph_name: NotRequired[str]
     trigger_call_id: NotRequired[str]
+    cause: NotRequired[LifecycleCause]
     error: NotRequired[str]
 
 
@@ -406,6 +407,18 @@ class _TasksLifecycleBase(StreamTransformer):
         # Maps tracked namespace -> task_id of the parent task whose
         # `TaskResultPayload` will close it.
         self._open: dict[tuple[str, ...], str] = {}
+        # lc_agent_name observed at each namespace (first task event wins).
+        # Not read by the base discriminator (which only checks whether the
+        # current task carries an lc_agent_name); maintained as extension state
+        # for subclasses that project named subagents — e.g. a `run.subagents`
+        # transformer reads this to filter to nested runs that have a name.
+        self._lc_by_ns: dict[tuple[str, ...], str | None] = {}
+        # Pregel task_id -> triggering LLM tool_call_id, harvested from a task
+        # whose `input` is a `tool_call_with_context` dict (current shape) or a
+        # list of tool-call dicts (legacy shape). The child subgraph's segment
+        # `node:<task_id>` shares this task_id, so a subagent recovers the tool
+        # call that spawned it (cross-payload).
+        self._pending_tool_calls: dict[str, str] = {}
 
     # --- Template-method hooks (subclass overrides) ---
 
@@ -418,6 +431,8 @@ class _TasksLifecycleBase(StreamTransformer):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         """Fired once per discovered namespace (first observed task event)."""
         raise NotImplementedError
@@ -443,18 +458,81 @@ class _TasksLifecycleBase(StreamTransformer):
         if "result" in data:
             self._handle_task_result(ns, data)
         else:
-            self._handle_task_start(ns)
+            self._record_identity(ns, data)
+            self._record_pending_tool_calls(data)
+            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 _record_identity(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
+        """Record this namespace's `lc_agent_name` (first task event wins).
+
+        Runs for every task-start event, including `ns == self.scope` and
+        tracked children. Pregel emits parent-namespace tasks before
+        child-namespace tasks, so under that ordering the parent's identity is
+        recorded by the time a child event is evaluated in `_handle_task_start`.
+        """
+        if ns in self._lc_by_ns:
+            return
+        metadata = data.get("metadata") or {}
+        self._lc_by_ns[ns] = metadata.get("lc_agent_name")
+
+    def _record_pending_tool_calls(self, data: dict[str, Any]) -> None:
+        """Harvest a task's triggering tool_call_id keyed by its task id.
+
+        A tool-dispatch task seeds `task_id -> tool_call_id`; the spawned
+        subgraph's namespace segment `node:<task_id>` shares that id, letting
+        a subagent recover the tool call that caused it across payloads. Two
+        input shapes are handled: the current Pregel push model schedules each
+        tool call as its own task whose `input` is a `tool_call_with_context`
+        dict, while a legacy / batched model passes a list of tool-call dicts.
+        """
+        task_id = data.get("id")
+        if not isinstance(task_id, str):
+            return
+        payload = data.get("input")
+        tool_call_id: str | None = None
+        # Current langgraph schedules each tool call as its own push task
+        # whose input is a `tool_call_with_context` dict.
+        if isinstance(payload, dict) and isinstance(payload.get("tool_call"), dict):
+            candidate = payload["tool_call"].get("id")
+            if isinstance(candidate, str):
+                tool_call_id = candidate
+        # Legacy / batched shape: input is a list of tool-call dicts.
+        elif isinstance(payload, list):
+            for tc in payload:
+                if isinstance(tc, dict) and isinstance(tc.get("id"), str):
+                    tool_call_id = tc["id"]  # first wins
+                    break
+        if tool_call_id is not None:
+            self._pending_tool_calls[task_id] = tool_call_id
+
+    def _handle_task_start(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
         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)
+        parsed_name, trigger_call_id = _parse_ns_segment(ns[-1])
+        metadata = data.get("metadata") or {}
+        child_lc = metadata.get("lc_agent_name")
+        # A subagent boundary is any nested run carrying an lc_agent_name (set
+        # by create_agent). Unnamed runs (lc_agent_name None) are excluded.
+        #
+        # A same-named nested agent — e.g. a subagent that invokes itself — is
+        # surfaced because it re-asserts its own lc_agent_name. The trade-off:
+        # a non-agent subgraph invoked inside a tool inherits the parent's
+        # lc_agent_name and will also surface (named after the parent). A caller
+        # that needs to exclude such a graph can null lc_agent_name in the
+        # config it invokes that graph with.
+        is_subagent = child_lc is not None
+        graph_name = child_lc if is_subagent else (parsed_name or None)
+        cause: LifecycleCause | None = None
+        if is_subagent and trigger_call_id is not None:
+            tool_call_id = self._pending_tool_calls.get(trigger_call_id)
+            if tool_call_id:
+                cause = {"type": "toolCall", "tool_call_id": str(tool_call_id)}
+        self._on_started(ns, graph_name, trigger_call_id, cause=cause)
         if trigger_call_id is not None:
             self._open[ns] = trigger_call_id
 
@@ -553,6 +631,8 @@ class LifecycleTransformer(_TasksLifecycleBase):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         if trigger_call_id is None:
             # Without a task id we can't correlate a parent-result
@@ -563,6 +643,8 @@ class LifecycleTransformer(_TasksLifecycleBase):
         if graph_name:
             payload["graph_name"] = graph_name
         payload["trigger_call_id"] = trigger_call_id
+        if cause is not None:
+            payload["cause"] = cause
         self._channel.push(payload)
 
     def _on_terminal(
@@ -625,6 +707,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         if self._mux is None:
             return
@@ -633,6 +717,10 @@ class SubgraphTransformer(_TasksLifecycleBase):
         except RuntimeError:
             return
         handle_cls = AsyncSubgraphRunStream if child_mux.is_async else SubgraphRunStream
+        # `cause` is intentionally ignored here: it is a wire/lifecycle-channel
+        # concern (carried on `LifecyclePayload`), not something the in-process
+        # subgraph navigation handle exposes. The argument is accepted only to
+        # keep the `_on_started` template signature uniform across transformers.
         handle = handle_cls(
             mux=child_mux,
             path=ns,
@@ -737,7 +825,12 @@ class SubgraphTransformer(_TasksLifecycleBase):
                 for child_ns, status, error in self._pop_terminal_transitions(ns, data):
                     await self._aon_terminal(child_ns, status, error)
             else:
-                self._handle_task_start(ns)
+                # Mirror the sync `process` bookkeeping so the async lane
+                # observes parent identity / tool calls before discriminating
+                # a subagent boundary.
+                self._record_identity(ns, data)
+                self._record_pending_tool_calls(data)
+                self._handle_task_start(ns, data)
             keep = False
         else:
             keep = True

libs/langgraph/langgraph/types.py

@@ -150,6 +150,16 @@ class TaskPayload(TypedDict):
     """Input data passed to the task."""
     triggers: list[str]
     """List of triggers that caused this task to be executed (e.g. channel writes)."""
+    metadata: NotRequired[dict[str, Any]]
+    """Framework-resolved metadata associated with the task.
+
+    Generic dict carrier following the messages-stream pattern. Populated by
+    `map_debug_tasks` from `task.config["metadata"]` when non-empty, so the
+    same keys `stream_mode="messages"` consumers see (e.g. `lc_agent_name`,
+    `langgraph_node`, `langgraph_step`) are available to stream transformers.
+
+    Consumers should ignore unrecognized keys.
+    """
 
 
 class TaskResultPayload(TypedDict):

libs/langgraph/tests/test_pregel_debug.py

@@ -0,0 +1,185 @@
+"""Tests for langgraph.pregel.debug helpers."""
+
+from __future__ import annotations
+
+from langgraph.pregel.debug import map_debug_tasks
+
+
+class _FakeTask:
+    """Minimal stand-in for PregelExecutableTask covering only what map_debug_tasks reads."""
+
+    def __init__(
+        self,
+        *,
+        id: str,
+        name: str,
+        input: object,
+        triggers: list[str],
+        config: dict | None,
+    ) -> None:
+        self.id = id
+        self.name = name
+        self.input = input
+        self.triggers = triggers
+        self.config = config
+
+
+def test_map_debug_tasks_forwards_metadata_when_present() -> None:
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {"lc_agent_name": "weather_agent"}},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert len(payloads) == 1
+    payload = payloads[0]
+    assert payload["id"] == "t1"
+    assert payload["name"] == "tools"
+    assert payload["metadata"] == {"lc_agent_name": "weather_agent"}
+
+
+def test_map_debug_tasks_omits_metadata_when_empty() -> None:
+    # Empty metadata dict in config: don't include metadata in the payload.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {}},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_omits_metadata_when_absent() -> None:
+    # No metadata key in config: don't include metadata in the payload.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_handles_none_config() -> None:
+    # task.config can be None; should not crash.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config=None,
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert len(payloads) == 1
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_filters_framework_metadata_keys() -> None:
+    """Internal framework keys are dropped from the forwarded metadata; only
+    user-meaningful keys (lc_agent_name, ls_integration, user metadata) ride
+    along. The framework keys (langgraph_*, thread_id, checkpoint_*) are
+    redundant with the task's own fields/namespace.
+    """
+    md = {
+        "lc_agent_name": "weather_agent",
+        "ls_integration": "langchain_create_agent",
+        "my_user_key": "x",
+        "thread_id": "thread-1",
+        "langgraph_step": 1,
+        "langgraph_node": "tools",
+        "langgraph_path": ("__pregel_pull", "tools"),
+        "langgraph_checkpoint_ns": "tools:abc",
+        "checkpoint_ns": "",
+    }
+    task = _FakeTask(
+        id="t1", name="tools", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"] == {
+        "lc_agent_name": "weather_agent",
+        "ls_integration": "langchain_create_agent",
+        "my_user_key": "x",
+    }
+
+
+def test_map_debug_tasks_omits_metadata_when_only_framework_keys() -> None:
+    """A task whose metadata is entirely framework keys (e.g. a plain
+    StateGraph node) yields no `metadata` key after filtering.
+    """
+    md = {
+        "thread_id": "thread-1",
+        "langgraph_step": 1,
+        "langgraph_node": "worker",
+        "langgraph_checkpoint_ns": "worker:abc",
+        "checkpoint_ns": "",
+    }
+    task = _FakeTask(
+        id="t1", name="worker", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert "metadata" not in payload
+
+
+def test_map_debug_tasks_metadata_is_copied_not_referenced() -> None:
+    """Mutating the source config after emission must not affect the
+    payload — TaskPayload.metadata is a defensive copy.
+    """
+    md = {"lc_agent_name": "a"}
+    task = _FakeTask(
+        id="t1", name="tools", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    md["lc_agent_name"] = "MUTATED"
+    assert payload["metadata"]["lc_agent_name"] == "a"
+
+
+def test_map_debug_tasks_folds_filtered_tags_into_metadata() -> None:
+    """Config tags are folded into TaskPayload.metadata under `tags`, with
+    langchain's internal `seq:step:*` tags filtered out — mirroring the
+    messages stream handler so both channels surface the same tag set."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={
+            "metadata": {"lc_agent_name": "weather_agent"},
+            "tags": ["seq:step:1", "user-tag", "session-123"],
+        },
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"]["lc_agent_name"] == "weather_agent"
+    assert payload["metadata"]["tags"] == ["user-tag", "session-123"]
+
+
+def test_map_debug_tasks_omits_tags_when_only_seq_step() -> None:
+    """If the only tags are internal `seq:step:*` markers, no `tags` key is
+    added (matches the messages handler's `if filtered_tags:` guard)."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {"lc_agent_name": "a"}, "tags": ["seq:step:1"]},
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert "tags" not in payload["metadata"]
+
+
+def test_map_debug_tasks_adds_tags_even_without_other_metadata() -> None:
+    """Filtered tags surface even when config has no metadata dict."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"tags": ["user-tag"]},
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"] == {"tags": ["user-tag"]}

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -35,20 +35,25 @@ def _tasks_start(
     *,
     task_id: str,
     name: str,
+    metadata: dict[str, Any] | None = None,
+    input: Any = None,
 ) -> dict[str, Any]:
     """Build a `tasks` ProtocolEvent carrying a TaskPayload (start)."""
+    data: dict[str, Any] = {
+        "id": task_id,
+        "name": name,
+        "input": input,
+        "triggers": [],
+    }
+    if metadata is not None:
+        data["metadata"] = metadata
     return {
         "type": "event",
         "method": "tasks",
         "params": {
             "namespace": namespace,
             "timestamp": TS,
-            "data": {
-                "id": task_id,
-                "name": name,
-                "input": None,
-                "triggers": [],
-            },
+            "data": data,
         },
     }
 
@@ -404,3 +409,272 @@ def test_stream_events_v3_with_nested_parent_ns_scopes_lifecycle() -> None:
         assert ns[:1] == ("outer:abc",), (
             f"namespace {ns} not within scoped prefix ('outer:abc',)"
         )
+
+
+# ---------------------------------------------------------------------------
+# Parsed-segment fallback (no subagent boundary)
+# ---------------------------------------------------------------------------
+
+
+def test_no_metadata_falls_through_to_existing_behavior() -> None:
+    """Tasks events without metadata produce the same output as before T4."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["graph_name"] == "agent"
+    assert payload["trigger_call_id"] == "abc"
+    assert "cause" not in payload
+
+
+def test_empty_metadata_dict_falls_through() -> None:
+    """An explicit empty metadata dict is treated the same as no metadata."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool", metadata={}))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["graph_name"] == "agent"
+    assert "cause" not in payload
+
+
+# ---------------------------------------------------------------------------
+# Subagent discrimination via lc_agent_name transition
+# ---------------------------------------------------------------------------
+#
+# A nested task is a subagent iff its metadata["lc_agent_name"] is present and
+# differs from its PARENT namespace's lc_agent_name. These tests replicate the
+# empirically-verified `create_agent` stream shape synthetically:
+#
+#   - A supervisor created via `create_agent(name="supervisor")` emits its own
+#     node tasks (model, tools) at ns=(), each with
+#     metadata["lc_agent_name"] == "supervisor".
+#   - The parent `tools` task at ns=() carries a `tool_call_with_context`
+#     dict as its `input` (with the LLM tool_call_id at
+#     input["tool_call"]["id"]) and a task `id`. (A legacy / batched shape
+#     passes a list of tool-call dicts instead; both are exercised below.)
+#   - When a tool body invokes an inner `create_agent(name="weather_agent")`,
+#     the inner agent's node tasks stream at ns=("tools:<taskid>",) with
+#     metadata["lc_agent_name"] == "weather_agent", sharing the SAME <taskid>
+#     as the parent `tools` task.
+#   - A plain StateGraph (no name) inherits the parent's lc_agent_name, so its
+#     child lc == parent lc -> NOT a subagent.
+
+
+def test_lifecycle_uses_lc_agent_name_for_subagent() -> None:
+    """A nested run whose lc_agent_name differs from its parent's is a subagent.
+
+    graph_name becomes the child's lc_agent_name; cause is recovered by joining
+    the child segment's task-id to the parent push task's tool call. This uses
+    the production `tool_call_with_context` dict input shape current langgraph
+    emits (tool_call_id at input["tool_call"]["id"]).
+    """
+    mux = _build_lifecycle_mux()
+    # Supervisor's `tools` push task at scope ns: carries its own lc_agent_name
+    # and a `tool_call_with_context` dict as `input`. Each tool call is its own
+    # push task, and the task id seeds the child segment.
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "call_weather",
+                    "args": {"city": "Boston"},
+                    "id": "call_w",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    # Inner weather_agent's first node task streams under the parent `tools`
+    # task's namespace segment (shared task id) with its own lc_agent_name.
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [subagent] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert subagent["graph_name"] == "weather_agent", (
+        "graph_name should be the child's lc_agent_name, not the parsed segment"
+    )
+    assert subagent["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}, (
+        "cause should recover the triggering tool_call_id from the parent push "
+        "task's tool_call_with_context input via the shared task id"
+    )
+
+
+def test_lifecycle_subagent_cause_from_legacy_list_input() -> None:
+    """cause recovery also handles the legacy / batched list input shape.
+
+    A parent task whose `input` is a list of tool-call dicts (rather than a
+    `tool_call_with_context` dict) still seeds the tool_call_id join.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input=[{"name": "call_weather", "args": {"city": "SF"}, "id": "call_w"}],
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [subagent] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert subagent["graph_name"] == "weather_agent"
+    assert subagent["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}
+
+
+def test_lifecycle_same_name_nested_run_is_surfaced() -> None:
+    """A nested run whose lc_agent_name matches the parent's is still surfaced.
+
+    A subagent that invokes itself re-asserts its own lc_agent_name, so child
+    lc == parent lc. The discriminator surfaces any nested run carrying an
+    lc_agent_name, so the recursive call is reported (named after the agent,
+    with the triggering tool call as cause).
+
+    Trade-off: a non-agent subgraph that merely inherited the parent's
+    lc_agent_name would also surface here. That is accepted; a caller can null
+    lc_agent_name in the config it invokes such a graph with to exclude it.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "weather_agent"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "recurse",
+                    "args": {},
+                    "id": "call_x",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    # The agent invokes itself: the nested run re-asserts the SAME lc_agent_name.
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_node_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [nested] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert nested["graph_name"] == "weather_agent", (
+        "a same-named nested run (e.g. self-recursion) must still be surfaced"
+    )
+    assert nested["cause"] == {"type": "toolCall", "tool_call_id": "call_x"}
+
+
+def test_lifecycle_unnamed_nested_agent_is_not_subagent() -> None:
+    """A nested run with lc_agent_name None is excluded (not a subagent)."""
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "lookup",
+                    "args": {},
+                    "id": "call_x",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["plain:tools_task_1"],
+            task_id="inner_node_1",
+            name="inner_node",
+            metadata={"lc_agent_name": None},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [nested] = [p for p in started if p["namespace"] == ["plain:tools_task_1"]]
+    assert nested["graph_name"] == "plain"
+    assert "cause" not in nested
+
+
+def test_lifecycle_subagent_terminal_roundtrip() -> None:
+    """A detected subagent closes with `completed` when its parent task results.
+
+    Pushes the subagent's `started` (via the `tool_call_with_context` parent
+    plus the child task event) and then the parent push task's terminal
+    result, asserting the namespace is closed and the `started` payload's
+    projected graph_name / cause survive the roundtrip.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "call_weather",
+                    "args": {"city": "Boston"},
+                    "id": "call_w",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+    # The parent push task (id=tools_task_1, at scope ns) finishes, closing
+    # the subagent subgraph that streamed under `tools:tools_task_1`.
+    mux.push(_tasks_result([], task_id="tools_task_1", name="tools"))
+
+    payloads = _drain_lifecycle(mux)
+    ns = ["tools:tools_task_1"]
+    subagent = [p for p in payloads if p["namespace"] == ns]
+    assert [p["event"] for p in subagent] == ["started", "completed"]
+    started, _completed = subagent
+    assert started["graph_name"] == "weather_agent"
+    assert started["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}