wehub-resource-sync
a7d6d88f6f
CI / changes (push) Has been cancelled
CI / cd libs/checkpoint (push) Has been cancelled
CI / cd libs/checkpoint-conformance (push) Has been cancelled
CI / cd libs/checkpoint-postgres (push) Has been cancelled
CI / cd libs/checkpoint-sqlite (push) Has been cancelled
CI / cd libs/cli (push) Has been cancelled
CI / cd libs/prebuilt (push) Has been cancelled
CI / cd libs/sdk-py (push) Has been cancelled
CI / cd libs/langgraph (push) Has been cancelled
CI / Check SDK methods matching (push) Has been cancelled
CI / Check CLI schema hasn't changed #3.13 (push) Has been cancelled
CI / CLI integration test (push) Has been cancelled
CI / sdk-py integration test (push) Has been cancelled
CI / CI Success (push) Has been cancelled
baseline / benchmark (push) Has been cancelled
Deploy Redirects to GitHub Pages / deploy (push) Has been cancelled
713 lines
26 KiB
Python
713 lines
26 KiB
Python
"""Tests for LifecycleTransformer.
|
|
|
|
Consumes the `tasks` stream mode and emits subgraph lifecycle payloads
|
|
on the `lifecycle` channel for both in-process iteration via
|
|
`run.lifecycle` and wire delivery via `custom:lifecycle` protocol
|
|
events. Most tests dispatch synthetic protocol events through a
|
|
`StreamMux` to keep the inference logic isolated; the end-of-file
|
|
group exercises the path through real graphs (multi-depth
|
|
discovery, nested `stream_events(version="v3")` calls with non-empty `parent_ns`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import operator
|
|
import time
|
|
from typing import Annotated, Any
|
|
|
|
from typing_extensions import TypedDict
|
|
|
|
from langgraph._internal._constants import CONF, CONFIG_KEY_CHECKPOINT_NS
|
|
from langgraph.constants import END, START
|
|
from langgraph.errors import GraphInterrupt
|
|
from langgraph.graph import StateGraph
|
|
from langgraph.stream._mux import StreamMux
|
|
from langgraph.stream.transformers import (
|
|
LifecyclePayload,
|
|
LifecycleTransformer,
|
|
_TasksLifecycleBase,
|
|
)
|
|
|
|
TS = int(time.time() * 1000)
|
|
|
|
|
|
def _tasks_start(
|
|
namespace: list[str],
|
|
*,
|
|
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": data,
|
|
},
|
|
}
|
|
|
|
|
|
def _tasks_result(
|
|
namespace: list[str],
|
|
*,
|
|
task_id: str,
|
|
name: str,
|
|
error: str | None = None,
|
|
interrupts: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Build a `tasks` ProtocolEvent carrying a TaskResultPayload (finish)."""
|
|
return {
|
|
"type": "event",
|
|
"method": "tasks",
|
|
"params": {
|
|
"namespace": namespace,
|
|
"timestamp": TS,
|
|
"data": {
|
|
"id": task_id,
|
|
"name": name,
|
|
"error": error,
|
|
"interrupts": interrupts or [],
|
|
"result": {},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _arm(mux: StreamMux) -> None:
|
|
"""Force projection channels to accept pushes (skip lazy-subscribe gate).
|
|
|
|
`StreamChannel.push` only appends to the local buffer when a
|
|
subscriber is attached. Tests that inspect `_items` directly need
|
|
the gate flipped before any event is dispatched.
|
|
"""
|
|
mux._events._subscribed = True
|
|
for transformer in mux._transformers:
|
|
if isinstance(transformer, LifecycleTransformer):
|
|
transformer._channel._subscribed = True
|
|
|
|
|
|
def _unstamped(items):
|
|
"""Strip push stamps from a StreamChannel's internal buffer."""
|
|
return [item for _stamp, item in items]
|
|
|
|
|
|
def _drain_lifecycle(mux: StreamMux) -> list[LifecyclePayload]:
|
|
"""Snapshot the lifecycle channel's buffer."""
|
|
transformer = mux.transformer_by_key("lifecycle")
|
|
assert isinstance(transformer, LifecycleTransformer)
|
|
return _unstamped(transformer._channel._items)
|
|
|
|
|
|
def _build_lifecycle_mux(*, scope: tuple[str, ...] = ()) -> StreamMux:
|
|
mux = StreamMux([LifecycleTransformer(scope=scope)], is_async=False)
|
|
_arm(mux)
|
|
return mux
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LifecycleTransformer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_started_emitted_on_first_direct_child_task() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="tool"))
|
|
|
|
[payload] = _drain_lifecycle(mux)
|
|
assert payload["event"] == "started"
|
|
assert payload["namespace"] == ["agent:abc123"]
|
|
assert payload["graph_name"] == "agent"
|
|
assert payload["trigger_call_id"] == "abc123"
|
|
|
|
|
|
def test_started_dedup_on_repeat_namespace() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="a"))
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t2", name="b"))
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started"]
|
|
|
|
|
|
def test_grandchild_namespace_discovered() -> None:
|
|
"""Subgraphs at any depth below scope are tracked, not just direct children."""
|
|
mux = _build_lifecycle_mux()
|
|
# First-seen task at length-2 ns means a 2nd-level subgraph started.
|
|
mux.push(_tasks_start(["agent:abc", "tool:def"], task_id="t1", name="x"))
|
|
|
|
[payload] = _drain_lifecycle(mux)
|
|
assert payload["event"] == "started"
|
|
assert payload["namespace"] == ["agent:abc", "tool:def"]
|
|
|
|
|
|
def test_nested_chain_emits_started_at_each_depth() -> None:
|
|
"""A graph → subgraph → subgraph chain produces a started event per level."""
|
|
mux = _build_lifecycle_mux()
|
|
# Subgraph1 starts emitting tasks (events tagged with its own ns).
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
# Subgraph1 invokes subgraph2; subgraph2's first task event arrives.
|
|
mux.push(_tasks_start(["agent:abc", "tool:def"], task_id="t2", name="deep"))
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["namespace"] for p in payloads] == [
|
|
["agent:abc"],
|
|
["agent:abc", "tool:def"],
|
|
]
|
|
assert all(p["event"] == "started" for p in payloads)
|
|
|
|
|
|
def test_nested_chain_emits_completed_at_each_depth() -> None:
|
|
"""Each subgraph in a nested chain closes when its parent task result arrives."""
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
mux.push(_tasks_start(["agent:abc", "tool:def"], task_id="t2", name="deep"))
|
|
|
|
# Subgraph2's owning task (id=def, inside subgraph1) finishes.
|
|
mux.push(_tasks_result(["agent:abc"], task_id="def", name="tool"))
|
|
# Subgraph1's owning task (id=abc, at root) finishes.
|
|
mux.push(_tasks_result([], task_id="abc", name="agent"))
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
events = [(p["event"], p["namespace"]) for p in payloads]
|
|
assert events == [
|
|
("started", ["agent:abc"]),
|
|
("started", ["agent:abc", "tool:def"]),
|
|
("completed", ["agent:abc", "tool:def"]),
|
|
("completed", ["agent:abc"]),
|
|
]
|
|
|
|
|
|
def test_completed_on_parent_task_result() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
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"]
|
|
|
|
|
|
def test_failed_on_parent_task_result_with_error() -> None:
|
|
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"))
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started", "failed"]
|
|
assert payloads[1]["error"] == "boom"
|
|
|
|
|
|
def test_interrupted_on_parent_task_result_with_interrupts() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
mux.push(
|
|
_tasks_result(
|
|
[],
|
|
task_id="abc",
|
|
name="agent",
|
|
interrupts=[{"value": "pause"}],
|
|
)
|
|
)
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started", "interrupted"]
|
|
|
|
|
|
def test_interrupt_takes_precedence_over_error() -> None:
|
|
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="should-be-suppressed",
|
|
interrupts=[{"value": "pause"}],
|
|
)
|
|
)
|
|
|
|
last = _drain_lifecycle(mux)[-1]
|
|
assert last["event"] == "interrupted"
|
|
assert "error" not in last
|
|
|
|
|
|
def test_finalize_completes_open_subgraphs() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
|
|
mux.close()
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started", "completed"]
|
|
|
|
|
|
def test_fail_emits_interrupted_for_graph_interrupt() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
|
|
mux.fail(GraphInterrupt())
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started", "interrupted"]
|
|
assert "error" not in payloads[1]
|
|
|
|
|
|
def test_fail_emits_failed_for_other_exceptions() -> None:
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
|
|
mux.fail(RuntimeError("boom"))
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["event"] for p in payloads] == ["started", "failed"]
|
|
assert payloads[1]["error"] == "boom"
|
|
|
|
|
|
def test_unrelated_methods_pass_through() -> None:
|
|
"""Non-`tasks` events are not consumed and don't emit lifecycle."""
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(
|
|
{
|
|
"type": "event",
|
|
"method": "values",
|
|
"params": {"namespace": ["agent:abc"], "timestamp": TS, "data": {}},
|
|
}
|
|
)
|
|
assert _drain_lifecycle(mux) == []
|
|
|
|
|
|
def test_scoped_transformer_filters_outside_scope_but_tracks_all_depths() -> None:
|
|
"""Scope filters the prefix; subgraphs at any depth below scope are tracked."""
|
|
mux = _build_lifecycle_mux(scope=("agent:abc",))
|
|
# Root-level task — out of scope (no shared prefix).
|
|
mux.push(_tasks_start(["other:1"], task_id="t1", name="other"))
|
|
# Direct child of agent:abc — in scope.
|
|
mux.push(_tasks_start(["agent:abc", "tool:def"], task_id="t2", name="tool"))
|
|
# Grandchild of agent:abc — also in scope, tracked at its own depth.
|
|
mux.push(
|
|
_tasks_start(["agent:abc", "tool:def", "deep:ghi"], task_id="t3", name="deep")
|
|
)
|
|
|
|
payloads = _drain_lifecycle(mux)
|
|
assert [p["namespace"] for p in payloads] == [
|
|
["agent:abc", "tool:def"],
|
|
["agent:abc", "tool:def", "deep:ghi"],
|
|
]
|
|
|
|
|
|
def test_required_stream_modes_declared() -> None:
|
|
assert LifecycleTransformer.required_stream_modes == ("tasks",)
|
|
|
|
|
|
def test_protocol_event_method_is_native() -> None:
|
|
"""Native transformer — auto-forwarded events use `lifecycle`, not `custom:lifecycle`."""
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
|
|
methods = {evt["method"] for evt in _unstamped(mux._events._items)}
|
|
assert "lifecycle" in methods
|
|
assert "custom:lifecycle" not in methods
|
|
|
|
|
|
def test_tasks_events_suppressed_from_main_log() -> None:
|
|
"""Tasks events are folded into lifecycle and don't appear on the main log."""
|
|
mux = _build_lifecycle_mux()
|
|
mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
|
|
mux.push(_tasks_result([], task_id="abc", name="agent"))
|
|
|
|
methods = [evt["method"] for evt in _unstamped(mux._events._items)]
|
|
assert "tasks" not in methods
|
|
# Lifecycle events did make it through, though.
|
|
assert "lifecycle" in methods
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# End-to-end: real graphs through stream_events(version="v3")
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _State(TypedDict):
|
|
value: str
|
|
items: Annotated[list[str], operator.add]
|
|
|
|
|
|
def _passthrough(state: _State) -> dict[str, Any]:
|
|
return {"value": state["value"] + "!", "items": ["x"]}
|
|
|
|
|
|
def _make_two_level_nested() -> Any:
|
|
"""Build outer → middle → inner. Three Pregel instances, two nesting levels."""
|
|
inner_b: StateGraph = StateGraph(_State, input_schema=_State)
|
|
inner_b.add_node("inner_node", _passthrough)
|
|
inner_b.add_edge(START, "inner_node")
|
|
inner_b.add_edge("inner_node", END)
|
|
inner = inner_b.compile()
|
|
|
|
middle_b: StateGraph = StateGraph(_State, input_schema=_State)
|
|
middle_b.add_node("inner", inner)
|
|
middle_b.add_edge(START, "inner")
|
|
middle_b.add_edge("inner", END)
|
|
middle = middle_b.compile()
|
|
|
|
outer_b: StateGraph = StateGraph(_State, input_schema=_State)
|
|
outer_b.add_node("middle", middle)
|
|
outer_b.add_edge(START, "middle")
|
|
outer_b.add_edge("middle", END)
|
|
return outer_b.compile()
|
|
|
|
|
|
def test_stream_events_v3_real_graph_emits_lifecycle_at_each_depth() -> None:
|
|
"""Outer graph with two nested subgraphs surfaces lifecycle for both."""
|
|
graph = _make_two_level_nested()
|
|
run = graph.stream_events({"value": "x", "items": []}, version="v3")
|
|
|
|
# Iterating the projection drives the pump and drains synthesized
|
|
# lifecycle events at the same time.
|
|
payloads = list(run.lifecycle)
|
|
# Each subgraph instance produces a started + a terminal event. Two
|
|
# nested instances, so four payloads total in some interleaving.
|
|
by_event = {p["event"] for p in payloads}
|
|
assert "started" in by_event
|
|
assert "completed" in by_event
|
|
# Two distinct namespaces — direct child of root, and grandchild.
|
|
namespaces = {tuple(p["namespace"]) for p in payloads}
|
|
direct_children = {ns for ns in namespaces if len(ns) == 1}
|
|
grandchildren = {ns for ns in namespaces if len(ns) == 2}
|
|
assert direct_children, f"expected a level-1 lifecycle namespace, got {namespaces}"
|
|
assert grandchildren, f"expected a level-2 lifecycle namespace, got {namespaces}"
|
|
# Every direct-child namespace has a matching grandchild whose path extends it.
|
|
for parent in direct_children:
|
|
assert any(gc[: len(parent)] == parent for gc in grandchildren), (
|
|
f"grandchild does not extend parent {parent}: {grandchildren}"
|
|
)
|
|
|
|
|
|
def test_stream_events_v3_with_nested_parent_ns_scopes_lifecycle() -> None:
|
|
"""When `stream_events(version="v3")` is called with a non-empty checkpoint_ns in config,
|
|
`_resolve_parent_ns` returns that namespace and the registered
|
|
`LifecycleTransformer` is constructed with `scope=parent_ns`. This
|
|
exercises the path that exists today purely for nested-stream_events(version="v3")
|
|
callers; the test simulates such a caller by injecting a
|
|
checkpoint_ns into the config.
|
|
"""
|
|
graph = _make_two_level_nested()
|
|
config = {CONF: {CONFIG_KEY_CHECKPOINT_NS: "outer:abc"}}
|
|
run = graph.stream_events({"value": "x", "items": []}, config=config, version="v3")
|
|
|
|
payloads = list(run.lifecycle)
|
|
# Every emitted lifecycle namespace must extend the caller's scope —
|
|
# nothing at root-level, nothing under a sibling prefix.
|
|
for p in payloads:
|
|
ns = tuple(p["namespace"])
|
|
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"}
|
|
|
|
|
|
def test_on_started_override_without_cause_is_backward_compatible() -> None:
|
|
"""An `_on_started` override with the original 3-arg signature must work.
|
|
|
|
`cause` is delivered via `self._pending_cause`, not the call signature, so
|
|
older/third-party subclasses (e.g. deepagents' `SubagentTransformer`) that
|
|
override `_on_started(self, ns, graph_name, trigger_call_id)` keep working —
|
|
no `TypeError: _on_started() got an unexpected keyword argument 'cause'`.
|
|
"""
|
|
seen: list[tuple] = []
|
|
|
|
class _LegacyTransformer(_TasksLifecycleBase):
|
|
def init(self) -> dict[str, Any]:
|
|
return {}
|
|
|
|
def _should_track(self, ns: tuple[str, ...]) -> bool:
|
|
return len(ns) == 1 # direct child of the root scope
|
|
|
|
# Deliberately omits `cause` — mirrors a pre-`cause` override.
|
|
def _on_started(self, ns, graph_name, trigger_call_id) -> None: # type: ignore[override]
|
|
seen.append((ns, graph_name, trigger_call_id))
|
|
|
|
def _on_terminal(self, ns, status, error) -> None:
|
|
pass
|
|
|
|
transformer = _LegacyTransformer(scope=())
|
|
# Must not raise: the base calls `_on_started` without a `cause` kwarg.
|
|
transformer.process(_tasks_start(["agent:abc123"], task_id="abc123", name="agent"))
|
|
|
|
assert seen == [(("agent:abc123",), "agent", "abc123")]
|