diff --git a/libs/langgraph/langgraph/_internal/_config.py b/libs/langgraph/langgraph/_internal/_config.py index 45092659a..61d3dbec7 100644 --- a/libs/langgraph/langgraph/_internal/_config.py +++ b/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 diff --git a/libs/langgraph/langgraph/pregel/_messages.py b/libs/langgraph/langgraph/pregel/_messages.py index 80c5df6f0..2b61c8349 100644 --- a/libs/langgraph/langgraph/pregel/_messages.py +++ b/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( diff --git a/libs/langgraph/langgraph/pregel/debug.py b/libs/langgraph/langgraph/pregel/debug.py index 519a36488..b3ef5a874 100644 --- a/libs/langgraph/langgraph/pregel/debug.py +++ b/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: diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index 6159634a2..d08a47c1e 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/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:` 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:` 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 diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index fa0bdc685..02eed9dbc 100644 --- a/libs/langgraph/langgraph/types.py +++ b/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): diff --git a/libs/langgraph/tests/test_pregel_debug.py b/libs/langgraph/tests/test_pregel_debug.py new file mode 100644 index 000000000..38e1a4737 --- /dev/null +++ b/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"]} diff --git a/libs/langgraph/tests/test_stream_lifecycle_transformer.py b/libs/langgraph/tests/test_stream_lifecycle_transformer.py index 1972a245d..74bbc5560 100644 --- a/libs/langgraph/tests/test_stream_lifecycle_transformer.py +++ b/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:",) with +# metadata["lc_agent_name"] == "weather_agent", sharing the SAME +# 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"}