WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-10-01 05:55:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

slim lifecycle.cause to {type, tool_call_id}, require trigger_call_id

Browse Source 
The args-mining of `subagent_type` and `description` coupled the
streaming layer to one specific tool's args schema (the deepagents
`task` tool). Drop it: cause now carries only `{type: "tool_call",
tool_call_id}`. Args still live on the AIMessage's `tool_calls[i].args`
where they originate — consumers that want descriptive intent look
them up there via the `tool_call_id` join, keeping one source of truth.

Also promote `trigger_call_id` from `NotRequired` to required and
emit it on every event (started + every terminal path). Consumers
no longer need a namespace fallback to correlate `started` ↔ terminal
events for the same subgraph instance.

Renames:
- `_extract_per_call_args` -> `_extract_dispatching_tool_call_id`
  (now returns just `str | None`)
- `_record_invocation_metadata` -> `_record_dispatching_tool_call_id`
- `_invocation_metadata` cache -> `_dispatching_tool_call_id`
- `_on_started` parameter `invocation_metadata` -> `tool_call_id`
- `_on_terminal` gains a `trigger_call_id` parameter; `_pop_terminal_transitions`
  yields it through the tuple

Tests refreshed: cause-shape assertions reflect the slimmed payload,
new `test_trigger_call_id_present_on_every_terminal_path` covers all
five terminal exit paths, list-shape robustness test re-aimed at
missing `id` instead of missing args.
This commit is contained in:
Nick Hollon
2026-05-08 10:57:29 -04:00
parent ee9e234da0
commit 8c4bfa6d07
2 changed files with 221 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/stream/transformers.py
+108 -108
View File
@@ -360,8 +360,9 @@ def _parse_ns_segment(segment: str) -> tuple[str, str | None]:
return name, task_id if sep else None
def _extract_per_call_args(payload: Any) -> tuple[dict[str, Any], str | None] | None:
"""Pull `(args, tool_call_id)` out of a per-call dispatched task's `input`.
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
@@ -374,10 +375,6 @@ def _extract_per_call_args(payload: Any) -> tuple[dict[str, Any], str | None] |
2. Dict envelope wrapping a tool call:
`{"tool_call": {"id": ..., "args": {...}, ...}, ...}`. Older
prebuilt agent paths Send-fan-out this shape.
Returns `None` if the payload doesn't match either shape or its
`args` isn't a dict. `tool_call_id` is `None` when the envelope
omits an `id` or its `id` isn't a string.
"""
if isinstance(payload, dict):
tool_call = payload.get("tool_call")
@@ -389,11 +386,8 @@ def _extract_per_call_args(payload: Any) -> tuple[dict[str, Any], str | None] |
tool_call = payload[0]
else:
return None
args = tool_call.get("args")
if not isinstance(args, dict):
return None
raw_id = tool_call.get("id")
return args, raw_id if isinstance(raw_id, str) else None
return raw_id if isinstance(raw_id, str) else None
class LifecyclePayload(TypedDict, total=False):
@@ -422,15 +416,15 @@ class LifecyclePayload(TypedDict, total=False):
split (it does for normally-dispatched subgraphs). Absent if the segment
has no `:` separator.
"""
trigger_call_id: NotRequired[str]
trigger_call_id: str
"""Pregel task id of the dispatching task — the task whose execution
spawned this subgraph.
Set on `started` when the namespace tail segment carries one. This is
the join key for correlating multiple lifecycle events about the same
subgraph 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.
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.
"""
cause: NotRequired[dict[str, Any]]
"""Optional generic descriptor of *what triggered* this subgraph.
@@ -438,24 +432,19 @@ class LifecyclePayload(TypedDict, total=False):
Shape:
- `{"type": "tool_call", "subagent_type": "<name>", "description": "<text>",
"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). Mined from the per-call task's `input` —
see `_extract_per_call_args` for the recognised shapes. `subagent_type`
and `description` describe the invoking intent; `tool_call_id` is the
- `{"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 per-call Send fan-out gives each tool_call its own pregel task,
so each `tool_call_id` here corresponds to exactly one `trigger_call_id`).
Identity-level correlation across lifecycle events for the same
invocation still uses `trigger_call_id`.
Any field inside the dict is optional — partial metadata still produces
a `cause` if at least one field was extractable.
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.).
`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."""
@@ -476,13 +465,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, invocation_metadata)` —
- `_on_started(ns, graph_name, trigger_call_id, tool_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_terminal(ns, status, error, trigger_call_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
@@ -495,15 +484,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 -> invocation metadata for tasks whose `input` matched
# a recognized per-call tool-dispatch shape (see `_extract_per_call_args`
# for the accepted layouts). The lifecycle hook joins on this when a
# child subgraph fires its first task event so it can attribute the
# invocation to the model tool call's args.
self._invocation_metadata: dict[str, dict[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) ---
@@ -516,16 +505,16 @@ class _TasksLifecycleBase(StreamTransformer):
ns: tuple[str, ...],
graph_name: str | None,
trigger_call_id: str | None,
invocation_metadata: dict[str, str] | None = None,
tool_call_id: str | None = None,
) -> None:
"""Fired once per discovered namespace (first observed task event).
`invocation_metadata` carries invocation-intent fields mined from the
per-call dispatched task's `input.tool_call.args` envelope:
`{"subagent_type": str, "description": str}` (either or both
may be absent). `None` for structurally-triggered subgraphs.
Consumers join on `trigger_call_id` (the pregel task id) for
identity; this dict is purely descriptive metadata.
`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 `trigger_call_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
@@ -534,9 +523,14 @@ class _TasksLifecycleBase(StreamTransformer):
ns: tuple[str, ...],
status: LifecycleEvent,
error: str | None,
trigger_call_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.
`trigger_call_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
@@ -558,16 +552,16 @@ class _TasksLifecycleBase(StreamTransformer):
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 parent tasks that themselves live outside the
# so we capture dispatching tasks that themselves live outside the
# tracked region but whose `id` will appear as `trigger_call_id`
# for a child subgraph.
self._record_invocation_metadata(data)
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])
invocation_metadata = (
self._invocation_metadata.pop(trigger_call_id, None)
tool_call_id = (
self._dispatching_tool_call_id.pop(trigger_call_id, None)
if trigger_call_id is not None
else None
)
@@ -575,74 +569,67 @@ class _TasksLifecycleBase(StreamTransformer):
ns,
graph_name or None,
trigger_call_id,
invocation_metadata,
tool_call_id,
)
if trigger_call_id is not None:
self._open[ns] = trigger_call_id
def _record_invocation_metadata(self, data: dict[str, Any]) -> None:
"""Remember `task_id -> invocation metadata` if the task input matches
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 lives in `_extract_per_call_args`; this method
mines `subagent_type` / `description` from the resulting `args`
and tags on the model-side `tool_call_id` for UI anchoring.
Identity-level correlation still uses `trigger_call_id` (the
pregel task id, == this `task.id`).
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
extracted = _extract_per_call_args(data.get("input"))
if extracted is None:
tool_call_id = _extract_dispatching_tool_call_id(data.get("input"))
if tool_call_id is None:
return
args, tool_call_id = extracted
metadata: dict[str, str] = {}
for key in ("subagent_type", "description"):
value = args.get(key)
if isinstance(value, str):
metadata[key] = value
# `tool_call_id` rides along as anchoring metadata only when we
# already have descriptive intent (`subagent_type`/`description`).
# A per-call envelope without descriptive args is a non-subagent
# tool dispatch — it stays causeless, matching the structurally-
# triggered subgraph case.
if metadata and tool_call_id is not None:
metadata["tool_call_id"] = tool_call_id
if metadata:
self._invocation_metadata[task_id] = metadata
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, ...], LifecycleEvent, 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, trigger_call_id)`.
`trigger_call_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, ...], LifecycleEvent, 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, trigger_call_id in self._pop_terminal_transitions(
ns, data
):
self._on_terminal(child_ns, status, error, trigger_call_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, trigger_call_id in list(self._open.items()):
self._on_terminal(ns, "completed", None, trigger_call_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, trigger_call_id in list(self._open.items()):
self._on_terminal(ns, status, error_str, trigger_call_id)
self._open.clear()
@@ -707,21 +694,22 @@ class LifecycleTransformer(_TasksLifecycleBase):
ns: tuple[str, ...],
graph_name: str | None,
trigger_call_id: str | None,
invocation_metadata: dict[str, str] | None = 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
# 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),
"trigger_call_id": trigger_call_id,
}
if graph_name:
payload["graph_name"] = graph_name
payload["trigger_call_id"] = trigger_call_id
if invocation_metadata:
cause: dict[str, Any] = {"type": "tool_call"}
cause.update(invocation_metadata)
payload["cause"] = cause
if tool_call_id is not None:
payload["cause"] = {"type": "tool_call", "tool_call_id": tool_call_id}
self._channel.push(payload)
def _on_terminal(
@@ -729,8 +717,13 @@ class LifecycleTransformer(_TasksLifecycleBase):
ns: tuple[str, ...],
status: LifecycleEvent,
error: str | None,
trigger_call_id: str,
) -> None:
payload: LifecyclePayload = {"event": status, "namespace": list(ns)}
payload: LifecyclePayload = {
"event": status,
"namespace": list(ns),
"trigger_call_id": trigger_call_id,
}
if error is not None:
payload["error"] = error
self._channel.push(payload)
@@ -784,7 +777,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
ns: tuple[str, ...],
graph_name: str | None,
trigger_call_id: str | None,
invocation_metadata: dict[str, str] | None = None, # noqa: ARG002
tool_call_id: str | None = None, # noqa: ARG002
) -> None:
if self._mux is None:
return
@@ -807,6 +800,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
ns: tuple[str, ...],
status: LifecycleEvent,
error: str | None,
trigger_call_id: str, # noqa: ARG002
) -> None:
handle = self._handles.get(ns)
if handle is None or not self._mark_terminal(handle, status, error):
@@ -818,6 +812,7 @@ class SubgraphTransformer(_TasksLifecycleBase):
ns: tuple[str, ...],
status: LifecycleEvent,
error: str | None,
trigger_call_id: str, # noqa: ARG002
) -> None:
handle = self._handles.get(ns)
if handle is None or not self._mark_terminal(handle, status, error):
@@ -894,8 +889,13 @@ 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,
trigger_call_id,
) in self._pop_terminal_transitions(ns, data):
await self._aon_terminal(child_ns, status, error, trigger_call_id)
else:
self._handle_task_start(ns, data)
keep = False
@@ -909,9 +909,9 @@ class SubgraphTransformer(_TasksLifecycleBase):
def _complete_open_handles(self) -> BaseException | None:
first_error: BaseException | None = None
for ns in list(self._open):
for ns, trigger_call_id in list(self._open.items()):
try:
self._on_terminal(ns, "completed", None)
self._on_terminal(ns, "completed", None, trigger_call_id)
except BaseException as e:
if first_error is None:
first_error = e
@@ -927,9 +927,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, trigger_call_id in list(self._open.items()):
try:
await self._aon_terminal(ns, "completed", None)
await self._aon_terminal(ns, "completed", None, trigger_call_id)
except BaseException as e:
if first_error is None:
first_error = e
libs/langgraph/tests/test_stream_lifecycle_transformer.py
+113 -122
View File
@@ -41,10 +41,9 @@ def _tasks_start(
Pass `input=[{"id": ..., "name": ..., "args": {...}}]` (the per-call
list shape `langchain.agents.create_agent` Send-fans out) or
`input={"tool_call": {"args": {...}}, ...}` (the dict envelope older
`input={"tool_call": {"id": ..., ...}, ...}` (the dict envelope older
prebuilt agent paths emit) to exercise the lifecycle transformer's
input mining of invocation-intent metadata (`subagent_type`,
`description`).
`tool_call_id` mining for `lifecycle.started.cause`.
"""
return {
"type": "event",
@@ -135,20 +134,20 @@ def test_started_emitted_on_first_direct_child_task() -> None:
assert payload["trigger_call_id"] == "abc123"
def test_started_carries_cause_when_parent_input_has_invocation_metadata() -> None:
"""When a parent task's `input` is a dict envelope with a `tool_call`
field (`{"tool_call": {"args": {...}}, ...}`, the layout older
prebuilt agent paths Send-fan out per call), the transformer mines
`subagent_type`, `description`, and `tool_call_id` from `tool_call`
and remembers them keyed by `parent_task_id`. When that parent task
triggers a subgraph (the child's namespace ends in
`name:<parent_task_id>`), the `lifecycle.started` payload carries
`cause = {"type": "tool_call", "subagent_type": ..., "description": ...,
"tool_call_id": ...}`. Identity-level correlation still uses
`trigger_call_id`; `tool_call_id` is exposed so UI consumers can
anchor the lifecycle event back to the originating AI message."""
def test_started_carries_cause_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.cause` carries
`{"type": "tool_call", "tool_call_id": ...}`. Identity correlation
still uses `trigger_call_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()
# Parent task at root ns whose input matches the Send envelope.
mux.push(
_tasks_start(
[],
@@ -158,15 +157,11 @@ def test_started_carries_cause_when_parent_input_has_invocation_metadata() -> No
"tool_call": {
"id": "call_xyz",
"name": "task",
"args": {
"subagent_type": "researcher",
"description": "look up weather",
},
"args": {"subagent_type": "researcher"},
}
},
)
)
# Child subgraph's first task — trigger_call_id parsed from segment.
mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
[payload] = _drain_lifecycle(mux)
@@ -174,38 +169,6 @@ def test_started_carries_cause_when_parent_input_has_invocation_metadata() -> No
assert payload["trigger_call_id"] == "abc123"
assert payload["cause"] == {
"type": "tool_call",
"subagent_type": "researcher",
"description": "look up weather",
"tool_call_id": "call_xyz",
}
def test_started_cause_with_description_but_no_subagent_type() -> None:
"""Partial invocation metadata (only `description`, or only `subagent_type`)
still produces a cause — every field other than `type` is optional.
`tool_call_id` rides outside `args` and is extracted independently of
args content, so it shows up here even when args is sparse."""
mux = _build_lifecycle_mux()
mux.push(
_tasks_start(
[],
task_id="abc123",
name="tools",
input={
"tool_call": {
"id": "call_xyz",
"name": "task",
"args": {"description": "do a thing"},
}
},
)
)
mux.push(_tasks_start(["agent:abc123"], task_id="t1", name="model"))
[payload] = _drain_lifecycle(mux)
assert payload["cause"] == {
"type": "tool_call",
"description": "do a thing",
"tool_call_id": "call_xyz",
}
@@ -214,9 +177,10 @@ def test_started_carries_cause_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
`subagent_type`, `description`, and `tool_call_id` exactly as for
the dict envelope shape, so `lifecycle.started.cause` fires
regardless of which agent factory drove the dispatch."""
`tool_call_id` exactly as for the dict envelope shape, so
`lifecycle.started.cause` fires regardless of which agent factory
drove the dispatch.
"""
mux = _build_lifecycle_mux()
mux.push(
_tasks_start(
@@ -227,10 +191,7 @@ def test_started_carries_cause_for_list_shape_per_call_input() -> None:
{
"id": "tc-1",
"name": "task",
"args": {
"subagent_type": "researcher",
"description": "Do X",
},
"args": {"subagent_type": "researcher"},
}
],
)
@@ -240,12 +201,26 @@ def test_started_carries_cause_for_list_shape_per_call_input() -> None:
[payload] = _drain_lifecycle(mux)
assert payload["event"] == "started"
assert payload["trigger_call_id"] == "abc123"
assert payload["cause"] == {
"type": "tool_call",
"subagent_type": "researcher",
"description": "Do X",
"tool_call_id": "tc-1",
}
assert payload["cause"] == {"type": "tool_call", "tool_call_id": "tc-1"}
def test_started_carries_cause_when_args_absent() -> None:
"""`tool_call_id` is the only field cause needs; the dispatching
envelope can omit `args` entirely (or have non-dict args) and we
still produce a cause 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["cause"] == {"type": "tool_call", "tool_call_id": "tc-1"}
def test_list_shape_ignored_when_not_single_element() -> None:
@@ -260,16 +235,8 @@ def test_list_shape_ignored_when_not_single_element() -> None:
task_id="abc123",
name="tools",
input=[
{
"id": "tc-1",
"name": "task",
"args": {"subagent_type": "researcher"},
},
{
"id": "tc-2",
"name": "task",
"args": {"subagent_type": "writer"},
},
{"id": "tc-1", "name": "task"},
{"id": "tc-2", "name": "task"},
],
)
)
@@ -286,10 +253,10 @@ def test_list_shape_ignored_when_not_single_element() -> None:
assert "cause" not in payload2
def test_list_shape_robust_to_non_dict_or_missing_args() -> None:
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/non-dict `args`, or whose `args` lacks
both fields, must not raise — it just no-ops."""
dict, or whose dict has no string `id`, must not raise — it just
leaves `cause` absent."""
# Element is not a dict.
mux = _build_lifecycle_mux()
mux.push(_tasks_start([], task_id="t-a", name="tools", input=["not-a-dict"]))
@@ -297,47 +264,32 @@ def test_list_shape_robust_to_non_dict_or_missing_args() -> None:
[payload] = _drain_lifecycle(mux)
assert "cause" not in payload
# Args is not a dict.
# Element has no `id`.
mux2 = _build_lifecycle_mux()
mux2.push(
_tasks_start(
[],
task_id="t-b",
name="tools",
input=[{"id": "tc", "name": "task", "args": "nope"}],
)
)
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 "cause" not in payload2
# Args dict lacks both subagent_type and description.
# Element's `id` is not a string.
mux3 = _build_lifecycle_mux()
mux3.push(
_tasks_start(
[],
task_id="t-c",
name="tools",
input=[{"id": "tc", "name": "task", "args": {"other": "field"}}],
)
)
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 "cause" not in payload3
def test_parallel_dispatches_attributed_to_correct_parent() -> None:
"""Two parent task envelopes dispatched in the same model turn each
fan out to their own child subgraph; each child's `cause` must
reflect its own parent's invocation metadata, not the other
parent's.
"""Two dispatching task envelopes in the same model turn each fan
out to their own child subgraph; each child's `cause.tool_call_id`
must reflect its own dispatching envelope, not the other.
Defends the `trigger_call_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. Any join key derived
from the originating tool call's args (e.g. `subagent_type`)
would collide here because both parents target the same subagent
type.
so it disambiguates parallel dispatches 1:1. Both children share
the same `subagent_type` (in args, not on cause) — 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(
@@ -349,10 +301,7 @@ def test_parallel_dispatches_attributed_to_correct_parent() -> None:
"tool_call": {
"id": "call_1",
"name": "task",
"args": {
"subagent_type": "researcher",
"description": "find X",
},
"args": {"subagent_type": "researcher"},
}
},
)
@@ -366,17 +315,11 @@ def test_parallel_dispatches_attributed_to_correct_parent() -> None:
"tool_call": {
"id": "call_2",
"name": "task",
"args": {
"subagent_type": "researcher",
"description": "find Y",
},
"args": {"subagent_type": "researcher"},
}
},
)
)
# Each parent triggers its own child subgraph; the child's namespace
# ends in `name:<parent_task_id>` so the trigger_call_id parsed out
# of the segment is what disambiguates the two.
mux.push(_tasks_start(["agent:parent_A"], task_id="t1", name="model"))
mux.push(_tasks_start(["agent:parent_B"], task_id="t2", name="model"))
@@ -384,14 +327,10 @@ def test_parallel_dispatches_attributed_to_correct_parent() -> None:
by_ns = {tuple(p["namespace"]): p for p in payloads}
assert by_ns[("agent:parent_A",)]["cause"] == {
"type": "tool_call",
"subagent_type": "researcher",
"description": "find X",
"tool_call_id": "call_1",
}
assert by_ns[("agent:parent_B",)]["cause"] == {
"type": "tool_call",
"subagent_type": "researcher",
"description": "find Y",
"tool_call_id": "call_2",
}
@@ -470,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"]
# `trigger_call_id` is required on every event for the same subgraph
# so consumers can correlate `started` ↔ terminal without joining
# via `namespace`.
assert all(p["trigger_call_id"] == "abc" for p in payloads)
def test_failed_on_parent_task_result_with_error() -> None:
@@ -547,6 +490,54 @@ def test_fail_emits_failed_for_other_exceptions() -> None:
assert payloads[1]["error"] == "boom"
def test_trigger_call_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
`trigger_call_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["trigger_call_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["trigger_call_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["trigger_call_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["trigger_call_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["trigger_call_id"] == "mno"
def test_unrelated_methods_pass_through() -> None:
"""Non-`tasks` events are not consumed and don't emit lifecycle."""
mux = _build_lifecycle_mux()