diff --git a/apps/website/content/blog/2026-08-27-langgraph-subgraphs-when-to-split.mdx b/apps/website/content/blog/2026-08-27-langgraph-subgraphs-when-to-split.mdx index 3a818b27d..f062d52f7 100644 --- a/apps/website/content/blog/2026-08-27-langgraph-subgraphs-when-to-split.mdx +++ b/apps/website/content/blog/2026-08-27-langgraph-subgraphs-when-to-split.mdx @@ -182,11 +182,14 @@ Its module docstring says so outright: ```text Mirrors cockpit/chat/subagents' orchestrator + `task` tool + `_run_subagent` -structure, but each dispatch emits `subagent_activity` CUSTOM events +structure, but each dispatch emits `subagent_activity` CUSTOM events [...] +The backend's SubagentEmittingAgent expands those CUSTOM events into the +protocol's standard SUBAGENT_STARTED / TEXT_MESSAGE_* (attributed via +subagentRunId) / SUBAGENT_FINISHED / SUBAGENT_ERROR events ``` -The thing that differs is the transport: AG-UI's already carries a first-class delegation event. -So the specialists stayed a flat `async` helper and progress reaches the frontend as a custom event dispatched from the tool body. +The thing that differs is the transport: AG-UI already carries first-class delegation events — `SUBAGENT_STARTED`, `SUBAGENT_FINISHED`, and content events attributed to a child run. +So the specialists stayed a flat `async` helper, the tool body dispatches its progress as custom events, and a thin wrapper on the server expands those into the protocol's standard subagent events on the wire. The subgraph was never required by the feature. It was required by the transport. @@ -209,7 +212,7 @@ A node is already a unit. When the child really is a different graph — and the repo has exactly one of those, which is the case I owe you after arguing the other side this whole time. -Our `examples/ag-ui` demo runs on that same AG-UI transport, and it emits the same `subagent_activity` events from the tool body. +Our `examples/ag-ui` demo runs on that same AG-UI transport, and its research tool reaches the frontend the same way: the protocol's standard `SUBAGENT_*` events, with the child's messages and its own `lookup` tool call attributed to the child run. So it is not buying observability. It already had it. It compiles a child graph anyway. diff --git a/apps/website/content/blog/2026-08-31-what-changes-when-the-runtime-changes.mdx b/apps/website/content/blog/2026-08-31-what-changes-when-the-runtime-changes.mdx index f5b9ddd62..7adc34cd7 100644 --- a/apps/website/content/blog/2026-08-31-what-changes-when-the-runtime-changes.mdx +++ b/apps/website/content/blog/2026-08-31-what-changes-when-the-runtime-changes.mdx @@ -224,8 +224,8 @@ Our LangGraph subagent tracker is 543 lines. It infers subagent identity from stream namespaces, correlates namespaces back to tool-call ids, and requires you to configure `subagentToolNames: ['task']` in the provider so it knows which tool calls are delegations. That is client-side inference of a server-side fact, and inference is exactly as reliable as it sounds. -AG-UI ships `ACTIVITY_SNAPSHOT` and `ACTIVITY_DELTA` as first-class events. -The server declares "this is a subagent, here is its type, here is its status, here is its content." +AG-UI ships `SUBAGENT_STARTED`, `SUBAGENT_FINISHED`, and `SUBAGENT_ERROR` as first-class events, and every content event can carry a `subagentRunId`. +The server declares "this is a subagent, here is its name, here is the tool call that dispatched it, and these messages and tool calls belong to it." Our reducer projects those onto the neutral `Subagent` contract, and the AG-UI provider config for the subagents demo needs no subagent option at all. Declared beats inferred. diff --git a/examples/ag-ui/angular/e2e/manual/subagent-card-live.png b/examples/ag-ui/angular/e2e/manual/subagent-card-live.png new file mode 100644 index 000000000..4ac03b2dc Binary files /dev/null and b/examples/ag-ui/angular/e2e/manual/subagent-card-live.png differ diff --git a/examples/ag-ui/angular/e2e/subagent-card.spec.ts b/examples/ag-ui/angular/e2e/subagent-card.spec.ts index 33e60206c..1cddcd456 100644 --- a/examples/ag-ui/angular/e2e/subagent-card.spec.ts +++ b/examples/ag-ui/angular/e2e/subagent-card.spec.ts @@ -26,9 +26,9 @@ interface SubagentProbe { // each carrying `toolCallIds`/reasoning) and `toolCalls()` (the child's own // `lookup` calls, rendered as ). We read the projected map // directly rather than scraping the rendered card: it IS the data the card -// renders, and asserting on it proves the ACTIVITY snapshot/delta pipeline -// reconstructed the full reason→tool→answer transcript and settled to -// `complete`, independent of card layout. +// renders, and asserting on it proves the SUBAGENT_* + subagentRunId-attributed +// event stream reconstructed the full reason→tool→answer transcript and +// settled to `complete`, independent of card layout. async function readSubagents(page: Page): Promise { return page.evaluate(() => { const ng = (window as unknown as { ng?: { getComponent?: (el: Element) => unknown } }).ng; @@ -64,10 +64,12 @@ async function readSubagents(page: Page): Promise { // `research` tool, the langgraph child subgraph runs a genuine reason → tool → // answer loop (an LLM call that returns a `lookup` tool_call, the offline // `lookup` tool, then a second plain LLM call that writes the summary), and the -// AG-UI server converts the subagent_activity CUSTOM events into native -// ACTIVITY_SNAPSHOT/ACTIVITY_DELTA. The @threadplane/ag-ui reducer projects the -// activity to agent.subagents() (the ordered transcript chat-subagent-card -// renders) and the child's research text must stay OUT of the parent's bubble. +// AG-UI server's SubagentEmittingAgent expands the subagent_activity CUSTOM +// events into the protocol's SUBAGENT_STARTED/FINISHED plus subagentRunId- +// attributed TEXT_MESSAGE_* / TOOL_CALL_* events. The @threadplane/ag-ui +// reducer projects them to agent.subagents() (the ordered transcript +// chat-subagent-card renders) and the child's research text must stay OUT of +// the parent's bubble. test('research delegation reconstructs the multi-message subagent transcript', async ({ page, }) => { diff --git a/examples/ag-ui/python/docs/wire-capture-subagents.md b/examples/ag-ui/python/docs/wire-capture-subagents.md new file mode 100644 index 000000000..c456a7f07 --- /dev/null +++ b/examples/ag-ui/python/docs/wire-capture-subagents.md @@ -0,0 +1,392 @@ +# examples/ag-ui (LangGraph): subagent wire capture + emitter-seam decision + +Evidence for migrating this demo's research subagent from the private +ACTIVITY convention (`ACTIVITY_SNAPSHOT`/`ACTIVITY_DELTA` with `activityType: +"subagent"`) to the protocol's standard `SUBAGENT_*` events plus +`subagentRunId`-attributed `TEXT_MESSAGE_*` / `TOOL_CALL_*` events. Captured +2026-09-02 against the live backend (`src/server.py`, `uv run uvicorn +src.server:app --port 8000`, real `OPENAI_API_KEY`, `gpt-5-mini` for the +orchestrator and the research child) with `ag-ui-langgraph 0.0.40` and +`ag-ui-protocol 0.1.22` (bumped in the same commit as this doc; the previous +transitive pin was 0.1.19). + +This demo is the richer fork of `cockpit/ag-ui/subagents`: the child is a +compiled LangGraph subgraph running a reason → `lookup` tool → answer loop, so +the transcript has two assistant turns and one child tool call (the flat +cockpit variant has one turn and no tools). The cockpit capture lives at +`cockpit/ag-ui/subagents/python/docs/wire-capture-subagents.md`; this doc +records only what differs. + +Bridge citations are into the installed venv source: +`.venv/lib/python3.12/site-packages/ag_ui_langgraph/agent.py` and +`.venv/lib/python3.12/site-packages/ag_ui/core/events.py`. + +## 1. SDK check + +``` +$ uv run python -c "from ag_ui.core import SubagentStartedEvent, TextMessageContentEvent, ToolCallStartEvent, ToolCallResultEvent; print(TextMessageContentEvent.model_fields['subagent_run_id']); print(list(ToolCallStartEvent.model_fields)); print(list(ToolCallResultEvent.model_fields))" +annotation=Union[str, NoneType] required=False default=None alias='subagentRunId' alias_priority=1 +['metadata', 'type', 'timestamp', 'raw_event', 'tool_call_id', 'tool_call_name', 'parent_message_id', 'subagent_run_id'] +['metadata', 'type', 'timestamp', 'raw_event', 'message_id', 'tool_call_id', 'content', 'role', 'subagent_run_id'] +``` + +`ToolCallStartEvent` carries `parent_message_id` + `subagent_run_id`, and +`ToolCallResultEvent` carries `message_id`, `tool_call_id`, `content`, `role`, +`subagent_run_id` — the fields the contract's `tool_call` / `tool_result` +expansions need. `SubagentStartedEvent` exposes `parent_tool_call_id`. + +## 2. Baseline (before the emitter) + +`RunAgentInput` POSTed to `/agent` (`Accept: text/event-stream`): + +```json +{"threadId":"capture-thread-2","runId":"capture-thread-2-run", + "messages":[{"id":"u1","role":"user","content":"I want an in-depth research deep-dive on Angular signals: history, motivation, and how they compare to zone.js. Dispatch your research subagent (the research tool, subagent_type research) now; do not use search_documents."}], + "tools":[],"context":[],"state":{},"forwardedProps":{}} +``` + +(The e2e's bare prompt *"Research Angular signals and summarize"* delegates +under aimock replay, but the live orchestrator answered it with +`search_documents` — the system prompt routes "simple lookups" there and +reserves `research` for "in-depth research". The longer prompt above +delegated on the first attempt.) + +Scrubbed capture — line numbers are event indices (1-based) in the SSE +stream; `rawEvent` mirrors are dropped from every line and repetitive runs +are elided with `# [elided: ...]`. No keys or org ids appeared in the stream. + +``` +1 {"type":"RUN_STARTED","threadId":"capture-thread-2","runId":"capture-thread-2-run"} +3 {"type":"STEP_STARTED","stepName":"generate"} +9 {"type":"TOOL_CALL_START","toolCallId":"call_9n4N3xc350eeejpCUahSoH4v","toolCallName":"research","parentMessageId":"lc_run--01a063d9-03e5-7521-90c4-cc6e84ddf9fa"} + # [elided: 29 TOOL_CALL_ARGS deltas spelling {"topic":"Angular signals: history, motivation, ...","subagent_type":"research"}] +69 {"type":"TOOL_CALL_END","toolCallId":"call_9n4N3xc350eeejpCUahSoH4v"} +77 {"type":"STEP_FINISHED","stepName":"generate"} +78 {"type":"STEP_STARTED","stepName":"tools"} +80 {"type":"RAW","event":{"event":"on_tool_start","name":"research"}} +82 {"type":"ACTIVITY_SNAPSHOT","messageId":"call_9n4N3xc350eeejpCUahSoH4v","activityType":"subagent","content":{"toolCallId":"call_9n4N3xc350eeejpCUahSoH4v","name":"research","status":"running","messages":[],"toolCalls":[]},"replace":true} +86 {"type":"STEP_FINISHED","stepName":"tools"} +87 {"type":"STEP_STARTED","stepName":"agent"} # the CHILD subgraph's node +90 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"add","path":"/messages/-","value":{"id":"call_9n4N3xc350eeejpCUahSoH4v-0","role":"assistant","content":"","toolCallIds":[]}}]} +93 {"type":"TOOL_CALL_START","toolCallId":"call_fgoFFeLMn9eA1V2voGF8pa2Q","toolCallName":"lookup","parentMessageId":"lc_run--01a063d9-0b11-7b50-95c5-ddf0404f2be9"} # UNATTRIBUTED — the child's own call, streamed by the bridge as if it were the parent's + # [elided: 12 TOOL_CALL_ARGS deltas for lookup] +120 {"type":"TOOL_CALL_END","toolCallId":"call_fgoFFeLMn9eA1V2voGF8pa2Q"} +124 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"add","path":"/toolCalls/-","value":{"id":"call_fgoFFeLMn9eA1V2voGF8pa2Q","name":"lookup","args":{"query":"Angular signals history m..."},"status":"running"}},{"op":"add","path":"/messages/0/toolCallIds/-","value":"call_fgoFFeLMn9eA1V2voGF8pa2Q"}]} +130 {"type":"STEP_FINISHED","stepName":"agent"} +131 {"type":"STEP_STARTED","stepName":"tools"} +134 {"type":"RAW","event":{"event":"on_tool_start","name":"lookup"}} +135 {"type":"RAW","event":{"event":"on_tool_end","name":"lookup"}} +137 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"replace","path":"/toolCalls/0/status","value":"complete"},{"op":"replace","path":"/toolCalls/0/result","value":"Angular signals are ..."}]} +141 {"type":"STEP_FINISHED","stepName":"tools"} +142 {"type":"STEP_STARTED","stepName":"agent"} +146 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"add","path":"/messages/-","value":{"id":"call_9n4N3xc350eeejpCUahSoH4v-1","role":"assistant","content":"","toolCallIds":[]}}]} +150 {"type":"TEXT_MESSAGE_START","messageId":"lc_run--01a063d9-195b-7dc2-9a29-2e20f8b34638","role":"assistant"} # UNATTRIBUTED — the child's answer, streamed by the bridge into the PARENT transcript +153 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"replace","path":"/messages/1/content","value":"-"}]} + # [elided: 343 more (TEXT_MESSAGE_CONTENT + RAW + RAW on_custom_event + ACTIVITY_DELTA) quads — each ACTIVITY_DELTA carries the FULL accumulated text ("- What", "- What signals", ...): 306,482 bytes of `value` across 344 deltas for a 1,810-char answer] +1525 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"replace","path":"/messages/1/content","value":""}]} +1530 {"type":"TEXT_MESSAGE_END","messageId":"lc_run--01a063d9-195b-7dc2-9a29-2e20f8b34638"} +1537 {"type":"STEP_FINISHED","stepName":"agent"} +1538 {"type":"STEP_STARTED","stepName":"tools"} +1541 {"type":"ACTIVITY_DELTA", ... "patch":[{"op":"replace","path":"/status","value":"complete"}]} +1542 {"type":"RAW","event":{"event":"on_tool_end","name":"research"}} +1543 {"type":"TOOL_CALL_RESULT","messageId":"c5fafb3e-cf02-4ee5-b247-c841b97ec603","toolCallId":"call_9n4N3xc350eeejpCUahSoH4v","content":"- What signals are: a fine-grained reactivity primitive in Angular ..."} +1550 {"type":"STEP_FINISHED","stepName":"tools"} +1551 {"type":"STEP_STARTED","stepName":"generate"} +1557 {"type":"TEXT_MESSAGE_START","messageId":"lc_run--01a063d9-565b-7eb3-a7ac-837d933d2e97","role":"assistant"} + # [elided: 68 TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own answer] +1695 {"type":"TEXT_MESSAGE_END","messageId":"lc_run--01a063d9-565b-7eb3-a7ac-837d933d2e97"} +1704 {"type":"STEP_STARTED","stepName":"attach_citations"} +1711 {"type":"STEP_STARTED","stepName":"generate_title"} +1720 {"type":"MESSAGES_SNAPSHOT", ...} # user, assistant(research call), tool(result), assistant(answer) — the child's lc_run message is NOT in it +1721 {"type":"RUN_FINISHED","threadId":"capture-thread-2","runId":"capture-thread-2-run"} +``` + +Event tally (1,721 events): 1 RUN_STARTED, 9 STEP_STARTED, 9 STEP_FINISHED, +2 TOOL_CALL_START, 41 TOOL_CALL_ARGS, 2 TOOL_CALL_END, 1 ACTIVITY_SNAPSHOT, +349 ACTIVITY_DELTA, 1 TOOL_CALL_RESULT, 2 TEXT_MESSAGE_START, +412 TEXT_MESSAGE_CONTENT, 2 TEXT_MESSAGE_END, 10 STATE_SNAPSHOT, +2 MESSAGES_SNAPSHOT, 1 RUN_FINISHED, 877 RAW. No CUSTOM (the +`ActivityEmittingAgent` swallowed all 350 `subagent_activity` CUSTOM events +and emitted an ACTIVITY event in each one's place), no SUBAGENT_*, zero +events carrying `subagentRunId`. + +RAW breakdown: 469 `on_chat_model_stream`, 350 `on_custom_event`, 16 +`on_chain_stream`, 15 `on_chain_start`, 15 `on_chain_end`, 4 +`on_chat_model_start`, 4 `on_chat_model_end`, 2 `on_tool_start`, 2 +`on_tool_end`. + +### 2a. Ordering finding (design §6) + +**`TOOL_CALL_START` for `research` precedes the first ACTIVITY event:** START +at 9, ARGS through 68, END at 69, `STEP_FINISHED(generate)` / +`STEP_STARTED(tools)` at 77/78, `on_tool_start` at 80, ACTIVITY_SNAPSHOT at +82. The tool call is fully announced before the tool body runs, and the +delegation window nests between `TOOL_CALL_END` (69) and `TOOL_CALL_RESULT` +(1543) — the same nesting the cockpit lane measured. The reducer's +`parentToolCallId` lookup therefore always finds an already-announced tool +call; the card never renders nameless. + +### 2b. The bridge streams the child subgraph unattributed + +Unlike the cockpit lane (whose child is a bare `llm.astream` inside the tool +body), this child is a compiled subgraph and `ag-ui-langgraph` streams +subgraphs by default (`forwarded_props.stream_subgraphs`, `agent.py:257`, +`:590-595`). The bridge therefore emits the child's nodes as `STEP_*` +(`agent` / `tools`, 87-141) AND the child's own content as bridge-native, +unattributed events: the `lookup` `TOOL_CALL_START/ARGS/END` (93-120) and the +answer's `TEXT_MESSAGE_START/CONTENT/END` (150-1530, 344 deltas) land in the +PARENT transcript while the run streams. The trailing `MESSAGES_SNAPSHOT` +(1720) omits the child's `lc_run--…` message, so the parent bubble reconciles +after the run — which is why the e2e's "child text must not leak into the +parent bubble" assertion (checked post-finalization) passes today. On the +wire, however, the child's answer is shipped twice: once verbatim as +unattributed `TEXT_MESSAGE_CONTENT` and once accumulated inside +`ACTIVITY_DELTA`. + +### 2c. Wire volume + +The `SubagentStreamHandler` accumulated `text_so_far` and shipped it in every +`message` event; the transform turned each into a JSON-patch `replace` of the +whole message. 344 deltas carried 306,482 bytes of `value` for a 1,810-char +answer — quadratic in the answer length. The per-token contract (each event +carries only the raw token) makes this linear. + +### 2d. Why the 1:1 `_dispatch_event` seam cannot carry the migration + +Identical to the cockpit finding: `ActivityEmittingAgent` overrode +`LangGraphAgent._dispatch_event`, which is called inline as `yield +self._dispatch_event(...)` at every yield site — strictly one in / one out. +The standard sequence needs 1:N expansion (`tool_call` → three `TOOL_CALL_*` +events; `finished` → `TEXT_MESSAGE_END` + `SUBAGENT_FINISHED`; the CUSTOM +event itself consumed → zero out), so the seam is `LangGraphAgent.run`, the +async generator the FastAPI endpoint consumes. + +## After the emitter + +Captured 2026-09-02 against the shipped scenario (`SubagentEmittingAgent` +mounted in `src/server.py`, per-token `subagent_activity` deltas from +`SubagentStreamHandler`, `message_id`-carrying phases from the research +subgraph), same `RunAgentInput` as §2 (thread `capture-thread-3`). No keys or +org ids appeared in the stream; only repetitive delta runs, +`STATE_SNAPSHOT`s and the bridge's RAW mirrors are elided, marked with +`# [elided: ...]`. + +``` +1 {"type":"RUN_STARTED","threadId":"capture-thread-3","runId":"capture-thread-3-run"} +3 {"type":"STEP_STARTED","stepName":"generate"} +9 {"type":"TOOL_CALL_START","toolCallId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc","toolCallName":"research","parentMessageId":"lc_run--01a063e0-aa28-7850-97bc-8118cdc8749b"} + # [elided: 48 TOOL_CALL_ARGS deltas spelling {"topic":"...","subagent_type":"research"}] +107 {"type":"TOOL_CALL_END","toolCallId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc"} +115 {"type":"STEP_FINISHED","stepName":"generate"} +116 {"type":"STEP_STARTED","stepName":"tools"} +118 {"type":"RAW","event":{"event":"on_tool_start","name":"research"}} +120 {"type":"SUBAGENT_STARTED","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub","name":"research","parentToolCallId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc"} +124 {"type":"STEP_FINISHED","stepName":"tools"} +125 {"type":"STEP_STARTED","stepName":"agent"} # child subgraph node (passes through, as before) +128 {"type":"TEXT_MESSAGE_START","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m1","role":"assistant","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} + # [elided: the bridge-native, UNATTRIBUTED TOOL_CALL_START/ARGS/END for lookup that §2b showed here are GONE (at this capture, filtered by the wrapper — superseded, see "Declared opt-out" below); their RAW on_chat_model_stream mirrors remain] +148 {"type":"TEXT_MESSAGE_END","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m1","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +149 {"type":"TOOL_CALL_START","toolCallId":"call_9FAovVPV9iy8ZR4l2XI6w5sE","toolCallName":"lookup","parentMessageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m1","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +150 {"type":"TOOL_CALL_ARGS","toolCallId":"call_9FAovVPV9iy8ZR4l2XI6w5sE","delta":"{\"query\": \"Angular Signals introduced in Angular 16 release\"}","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +151 {"type":"TOOL_CALL_END","toolCallId":"call_9FAovVPV9iy8ZR4l2XI6w5sE","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +157 {"type":"STEP_FINISHED","stepName":"agent"} +158 {"type":"STEP_STARTED","stepName":"tools"} +161 {"type":"RAW","event":{"event":"on_tool_start","name":"lookup"}} +162 {"type":"RAW","event":{"event":"on_tool_end","name":"lookup"}} +164 {"type":"TOOL_CALL_RESULT","messageId":"call_9FAovVPV9iy8ZR4l2XI6w5sE-result","toolCallId":"call_9FAovVPV9iy8ZR4l2XI6w5sE","content":"Angular signals are a fine-grained reactivity primitive: ...","role":"tool","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +168 {"type":"STEP_FINISHED","stepName":"tools"} +169 {"type":"STEP_STARTED","stepName":"agent"} +173 {"type":"TEXT_MESSAGE_START","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m2","role":"assistant","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +178 {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m2","delta":"-","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +181 {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m2","delta":" History","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} + # [elided: 532 more attributed TEXT_MESSAGE_CONTENT deltas — the CHILD's answer, one raw token each; the bridge-native unattributed TEXT_MESSAGE_START/CONTENT/END copy from §2b is GONE] +1788 {"type":"STEP_FINISHED","stepName":"agent"} +1789 {"type":"STEP_STARTED","stepName":"tools"} +1792 {"type":"TEXT_MESSAGE_END","messageId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub-m2","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub"} +1793 {"type":"SUBAGENT_FINISHED","subagentRunId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc-sub","outcome":{"type":"success"}} +1794 {"type":"RAW","event":{"event":"on_tool_end","name":"research"}} +1795 {"type":"TOOL_CALL_RESULT","messageId":"a55caffb-242c-4b28-8d50-767f457ee8da","toolCallId":"call_Ax1IOxHNk2UEIdCaKlDvCLtc","content":"- History & motivation: introduced as the opt-in fine-grained reactivity primitive in Angular 16 ..."} +1802 {"type":"STEP_FINISHED","stepName":"tools"} +1803 {"type":"STEP_STARTED","stepName":"generate"} +1809 {"type":"TEXT_MESSAGE_START","messageId":"lc_run--01a063e1-1c52-77b3-a6eb-f1a2541df8f0","role":"assistant"} + # [elided: 83 TEXT_MESSAGE_CONTENT deltas — the ORCHESTRATOR's own answer, no subagentRunId] +1977 {"type":"TEXT_MESSAGE_END","messageId":"lc_run--01a063e1-1c52-77b3-a6eb-f1a2541df8f0"} +1986 {"type":"STEP_STARTED","stepName":"attach_citations"} +1993 {"type":"STEP_STARTED","stepName":"generate_title"} +2002 {"type":"MESSAGES_SNAPSHOT", ...} +2003 {"type":"RUN_FINISHED","threadId":"capture-thread-3","runId":"capture-thread-3-run"} +``` + +Event tally (2,003 events): 1 RUN_STARTED, 9 STEP_STARTED, 9 STEP_FINISHED, +1 TOOL_CALL_START, 48 TOOL_CALL_ARGS, 1 TOOL_CALL_END, 1 SUBAGENT_STARTED, +2 TEXT_MESSAGE_START(sub), 534 TEXT_MESSAGE_CONTENT(sub), 2 +TEXT_MESSAGE_END(sub), 1 TOOL_CALL_START(sub), 1 TOOL_CALL_ARGS(sub), +1 TOOL_CALL_END(sub), 1 TOOL_CALL_RESULT(sub), 1 SUBAGENT_FINISHED, +1 TOOL_CALL_RESULT, 1 TEXT_MESSAGE_START, 83 TEXT_MESSAGE_CONTENT, +1 TEXT_MESSAGE_END, 10 STATE_SNAPSHOT, 2 MESSAGES_SNAPSHOT, 1 RUN_FINISHED, +1,291 RAW. No CUSTOM, no ACTIVITY_*, no SUBAGENT_ERROR. + +**Child deltas: streaming, one raw token per event.** 534 attributed content +events carried 2,810 bytes of `delta` for a 2,810-char answer — linear, versus +§2c's 306,482 bytes for a 1,810-char answer. The joined child deltas equal the +parent's `TOOL_CALL_RESULT.content` byte-for-byte. Every child event carries +`subagentRunId` derived from the wire `toolCallId` (`-sub`, +messages `-sub-m1` / `-m2`), `SUBAGENT_STARTED.parentToolCallId` +matches the bridge-native `TOOL_CALL_START.toolCallId` verbatim, and the +child's `lookup` call is a fully attributed `TOOL_CALL_START` (with +`parentMessageId` = the tool-calling turn `-sub-m1`) → `ARGS` → `END` → +`TOOL_CALL_RESULT` (`role: tool`, `messageId: -result`). The +`subagent_activity` CUSTOM events were consumed (0 on the wire); their RAW +`on_custom_event` mirrors (540) still pass through because the bridge yields +them before `_handle_single_event`, and the client ignores RAW. + +**The §2b duplicates are gone.** Exactly one `lookup` `TOOL_CALL_START` and +exactly two child `TEXT_MESSAGE_START`s are on the wire, all attributed; the +only unattributed `TEXT_MESSAGE_START` is the orchestrator's own answer +(1809). At this capture the wrapper achieved that by *inferring* the +duplicates — dropping every unattributed content event between +`SUBAGENT_STARTED` and `SUBAGENT_FINISHED`. That filter was replaced in review +by the bridge's declared opt-out (next section); the wire shape is the same, +the mechanism is not. `STEP_*` for the child nodes still pass through. + +**Measured order, `TOOL_CALL_START` vs `SUBAGENT_STARTED`:** START 9 → ARGS → +END 107 → `on_tool_start` 118 → SUBAGENT_STARTED 120 → … → SUBAGENT_FINISHED +1793 → TOOL_CALL_RESULT 1795. The tool call is fully announced before the tool +body runs, so the reducer attaches the card to an already-known +`parentToolCallId`; the whole `SUBAGENT_*` block nests between +`TOOL_CALL_END` and `TOOL_CALL_RESULT`, as in the cockpit lane. + +## Declared opt-out (review follow-up) + +Review of the emitter asked why the wrapper *inferred* the child's duplicates +(any unattributed `TEXT_MESSAGE_*` / `TOOL_CALL_*` inside a delegation window) +when the bridge already honors a declared opt-out. It does: +`ag_ui_langgraph/agent.py:993-994` reads the LangChain run metadata keys +`emit-messages` and `emit-tool-calls` (default `True`) and skips emitting +`TEXT_MESSAGE_*` / `TOOL_CALL_*` for runs that carry them as `False` — while +callbacks (`SubagentStreamHandler`, `adispatch_custom_event`) still fire, +`STEP_*` still pass, and the parent's own `TOOL_CALL_RESULT` / answer are +untouched. Metadata set on the tool's `subgraph.ainvoke(..., config=...)` +inherits into every run of the child subgraph. + +So the `research` tool now invokes the subgraph with +`config={"callbacks": [...], "metadata": {"emit-messages": False, +"emit-tool-calls": False}}` (`src/graph.py`), and `SubagentEmittingAgent` is +back to a pure 1:N expander: no window flag, no remembered ids, every +non-`subagent_activity` event passes through untouched +(`tests/test_subagent_emitting_agent.py` +`test_bridge_native_events_pass_through_untouched_even_inside_a_delegation`; +`tests/test_subagent_emission.py` +`test_research_tool_declares_the_child_silent_via_bridge_metadata` asserts the +metadata is on the invocation). + +Re-captured 2026-09-02 with the opt-out and no wrapper-side filtering, same +`RunAgentInput` as §2 (thread `capture-optout-1788383126`, real +`OPENAI_API_KEY`, `gpt-5-mini`): + +``` +1 {"type":"RUN_STARTED", ...} +9 {"type":"TOOL_CALL_START","toolCallId":"call_zDfLSjalNBgmmB45oNQ7ERvm","toolCallName":"research", ...} +69 {"type":"TOOL_CALL_END","toolCallId":"call_zDfLSjalNBgmmB45oNQ7ERvm"} +78 {"type":"STEP_STARTED","stepName":"tools"} +80 {"type":"RAW","event":{"event":"on_tool_start","name":"research"}} +82 {"type":"SUBAGENT_STARTED","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub","name":"research","parentToolCallId":"call_zDfLSjalNBgmmB45oNQ7ERvm"} +87 {"type":"STEP_STARTED","stepName":"agent"} # child node — still on the wire +90 {"type":"TEXT_MESSAGE_START","messageId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub-m1","role":"assistant","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} + # [no bridge-native TOOL_CALL_START for lookup here — the bridge skipped it (emit-tool-calls=False); its RAW on_chat_model_stream mirrors remain] +122 {"type":"TEXT_MESSAGE_END","messageId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub-m1","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} +123 {"type":"TOOL_CALL_START","toolCallId":"call_1E9G6oUvZ43IDK58Hc0TSnVC","toolCallName":"lookup","parentMessageId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub-m1","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} +124 {"type":"TOOL_CALL_ARGS", ... "subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} +125 {"type":"TOOL_CALL_END","toolCallId":"call_1E9G6oUvZ43IDK58Hc0TSnVC","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} +132 {"type":"STEP_STARTED","stepName":"tools"} # child node +138 {"type":"TOOL_CALL_RESULT","messageId":"call_1E9G6oUvZ43IDK58Hc0TSnVC-result","toolCallId":"call_1E9G6oUvZ43IDK58Hc0TSnVC","role":"tool","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub", ...} +143 {"type":"STEP_STARTED","stepName":"agent"} + # [259 attributed TEXT_MESSAGE_CONTENT deltas on -sub-m2 — the child's answer; no bridge-native TEXT_MESSAGE_START/CONTENT/END copy (emit-messages=False)] +941 {"type":"TEXT_MESSAGE_END","messageId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub-m2","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub"} +942 {"type":"SUBAGENT_FINISHED","subagentRunId":"call_zDfLSjalNBgmmB45oNQ7ERvm-sub","outcome":{"type":"success"}} +944 {"type":"TOOL_CALL_RESULT","toolCallId":"call_zDfLSjalNBgmmB45oNQ7ERvm","content":"- What they are: Angular Signals are a fine-grained reactivity primitive ..."} # the PARENT's result, untouched +952 {"type":"STEP_STARTED","stepName":"generate"} +958 {"type":"TEXT_MESSAGE_START","messageId":"lc_run--01a063f1-275c-79f0-8b5d-f6175c48fc09","role":"assistant"} # the ORCHESTRATOR's own answer — the only unattributed TEXT_MESSAGE_START + # [elided: 49 TEXT_MESSAGE_CONTENT deltas, no subagentRunId] +1084 {"type":"RUN_FINISHED", ...} +``` + +Event tally (1,084 events): 1 RUN_STARTED, 9 STEP_STARTED, 9 STEP_FINISHED, +1 TOOL_CALL_START, 29 TOOL_CALL_ARGS, 1 TOOL_CALL_END, 1 SUBAGENT_STARTED, +2 TEXT_MESSAGE_START(sub), 259 TEXT_MESSAGE_CONTENT(sub), +2 TEXT_MESSAGE_END(sub), 1 TOOL_CALL_START(sub), 1 TOOL_CALL_ARGS(sub), +1 TOOL_CALL_END(sub), 1 TOOL_CALL_RESULT(sub), 1 SUBAGENT_FINISHED, +1 TOOL_CALL_RESULT, 1 TEXT_MESSAGE_START, 49 TEXT_MESSAGE_CONTENT, +1 TEXT_MESSAGE_END, 10 STATE_SNAPSHOT, 2 MESSAGES_SNAPSHOT, 1 RUN_FINISHED, +700 RAW (377 `on_chat_model_stream`, 265 `on_custom_event`, 16 +`on_chain_stream`, 15/15 `on_chain_start`/`_end`, 4/4 +`on_chat_model_start`/`_end`, 2/2 `on_tool_start`/`_end`). No CUSTOM, no +ACTIVITY_*, no SUBAGENT_ERROR. + +- (a) **Zero** unattributed `TEXT_MESSAGE_*` / `TOOL_CALL_*` events between + `SUBAGENT_STARTED` (82) and `SUBAGENT_FINISHED` (942), with nothing + filtering them: the `lookup` call appears exactly once (123, attributed) and + the child's answer only as the 259 attributed deltas on `-sub-m2`. The only + unattributed `TEXT_MESSAGE_START` in the whole run is the orchestrator's + (958). The joined child deltas (1,386 chars) equal the parent's + `TOOL_CALL_RESULT.content` byte-for-byte. +- (b) The parent's `TOOL_CALL_RESULT` for the delegation (944) and its own + answer (958 → 1 START / 49 CONTENT / 1 END) are present and unattributed. +- (c) `STEP_*` still passes: 9 `STEP_STARTED` / 9 `STEP_FINISHED` + (`generate` ×2, `tools` ×3, `agent` ×2, `attach_citations`, + `generate_title`) — the child's `agent` / `tools` nodes included. + +The e2e suite (`examples/ag-ui/angular/e2e`, aimock replay) passed unchanged +after the swap. + +## Browser verification + +2026-09-02, live backend (real `OPENAI_API_KEY`, uvicorn on :8000) + `npx nx +serve examples-ag-ui-angular --port 4201`, driven headlessly with Playwright +(the §2 prompt typed into the composer). Screenshot, taken while the research +card was still `running` with the answer turn mid-stream: +`examples/ag-ui/angular/e2e/manual/subagent-card-live.png`. + +What rendered: the `research` dispatch produced an inline +`` anchored to its tool call — header `research` + wire +`toolCallId` + `running` badge + "2 message(s)" — with the child's transcript +inside it: the tool-calling turn as the first `.sac__msg` (empty text), then +the answer turn streaming below it as a second `.sac__msg`. Once the child +finished, the card flipped to `complete` and collapsed, and the +orchestrator's own summary streamed in the parent bubble. The child text +never appeared in the parent bubble (`chat-streaming-md` of the final +assistant message does not contain the child's "History & motivation" +sentence), and no stray `lookup` tool-call card appeared in the parent +transcript. + +The `lookup` call reached the projection — `agent.subagents()` reports +`toolCalls: [{name: "lookup", result: ...}]` (what the e2e asserts) — but the +card does not yet draw a `` for it: the card looks tool +calls up through `message.toolCallIds`, and the reducer's attributed +`TOOL_CALL_START` route (`libs/ag-ui/src/lib/reducer.ts` +`routeSubagentContentEvent`) pushes onto the entry's `toolCalls` without +linking the id into the open message (the legacy ACTIVITY transform used to +patch `/messages//toolCallIds/-` explicitly). The wire carries +`parentMessageId` on the attributed `TOOL_CALL_START`, so this is a reducer +follow-up, not a demo defect. + +Did the card stream mid-run: **yes**. Polling `agent.subagents()`, the +card's `innerText` and the `.sac__msg` count every 150ms: + +- t≈2.5s — card mounts on `SUBAGENT_STARTED`: `running`, 1 message (the + tool-calling turn, empty content), 0 tool calls, 1 `.sac__msg`. +- t≈18.9s — after gpt-5-mini's reasoning latency: `lookup` tool call present + with its result (`hasResult: true`), 2 messages, 2 `.sac__msg`, answer + turn at 46 chars. +- t≈18.9s → 22.5s — the answer turn grows monotonically while `running`: + 46 → 72 → 174 → 235 → 340 → 422 → 503 → 603 → 682 → 759 → 847 → 929 → + 1004 → … → 1862 chars across consecutive 150ms samples (card `innerText` + 103 → 1907 chars in step). +- t≈22.6s — `complete` at 1,878 chars; the card collapses (`innerText` + 1907 → 58 chars, 0 expanded `.sac__msg`). + +This confirms the attributed `TEXT_MESSAGE_CONTENT` deltas render +progressively in the card's second message while the attributed `lookup` +`TOOL_CALL_*` / `TOOL_CALL_RESULT` events render as the first message's tool +call — not as one post-hoc paste, and not as a stray tool call or bubble in +the parent transcript. diff --git a/examples/ag-ui/python/pyproject.toml b/examples/ag-ui/python/pyproject.toml index 80df87b99..8b3ebd38a 100644 --- a/examples/ag-ui/python/pyproject.toml +++ b/examples/ag-ui/python/pyproject.toml @@ -6,6 +6,7 @@ dependencies = [ "langgraph>=0.3", "langchain-openai>=0.3", "ag-ui-langgraph>=0.0.37", + "ag-ui-protocol>=0.1.22", "fastapi>=0.115", "uvicorn>=0.30", "python-dotenv>=1.0", diff --git a/examples/ag-ui/python/src/graph.py b/examples/ag-ui/python/src/graph.py index 642a68660..44a68ad55 100644 --- a/examples/ag-ui/python/src/graph.py +++ b/examples/ag-ui/python/src/graph.py @@ -278,10 +278,11 @@ def request_approval(reason: str) -> str: # node loops back to `agent`. A per-run `iterations` counter caps the loop # so it always terminates (the agent is told to answer after one lookup). # Each node emits the structured transcript (`message_start` / `tool_call` / -# `tool_result`) as `subagent_activity` CUSTOM events the L2 transform turns -# into AG-UI ACTIVITY DELTAs; live token text is streamed separately by the -# SubagentStreamHandler (which tags each `message` with the same -# `message_index` the subgraph opened the turn with). +# `tool_result`) as `subagent_activity` CUSTOM events that the server's +# SubagentEmittingAgent expands into the protocol's `subagentRunId`-attributed +# TEXT_MESSAGE_* / TOOL_CALL_* events; live token text is streamed +# separately by the SubagentStreamHandler as per-token `message` deltas +# tagged with the message id the subgraph opened the turn with. class ResearchState(TypedDict): messages: Annotated[list, add_messages] topic: Optional[str] @@ -346,7 +347,8 @@ def _build_research_subgraph(emit, run_state, llm_factory=_make_research_llm): `emit(payload)` dispatches a `subagent_activity` CUSTOM event already keyed by the parent tool_call_id. `run_state` is the SubagentRunState - the SubagentStreamHandler reads `message_index` from for live tokens. + that derives the `-sub-m` message ids; the + SubagentStreamHandler reads the open id from it for live tokens. `llm_factory(force_answer)` returns the model for a turn — overridable in tests with a fake tool-calling chat model. """ @@ -354,10 +356,10 @@ def _build_research_subgraph(emit, run_state, llm_factory=_make_research_llm): async def agent_node(state: ResearchState) -> dict: topic = state.get("topic") or "" iterations = state.get("iterations") or 0 - # Open a new assistant turn. The handler reads this index for the - # live `message` tokens it streams during this LLM call. - run_state.message_index = iterations - await emit({"phase": "message_start", "message_index": iterations}) + # Open a new assistant turn. The handler reads this id for the + # live `message` deltas it streams during this LLM call. + message_id = run_state.open_message() + await emit({"phase": "message_start", "message_id": message_id}) force_answer = iterations >= _RESEARCH_MAX_ITERATIONS system = SystemMessage(content=( @@ -380,15 +382,16 @@ async def agent_node(state: ResearchState) -> dict: call_llm = llm_factory(force_answer) response = await call_llm.ainvoke(messages) - # Emit one tool_call event per call on the returned AIMessage. + # Emit one tool_call event per call on the returned AIMessage. The + # emitter anchors it to the turn that is open (this one). tool_calls = getattr(response, "tool_calls", None) or [] for tc in tool_calls: await emit({ "phase": "tool_call", - "message_index": iterations, + "message_id": message_id, "tool_call_id": tc.get("id"), "name": tc.get("name"), - "args": tc.get("args"), + "args": tc.get("args") or {}, }) return {"messages": [response], "iterations": iterations + 1} @@ -405,15 +408,11 @@ async def tools_node(state: ResearchState) -> dict: else: result = f"(unknown tool: {name})" out.append(ToolMessage(content=result, tool_call_id=tc.get("id"))) - # tool_index is the 0-based position of this call in the run's - # toolCalls[] (same order tool_call events were emitted). await emit({ "phase": "tool_result", - "tool_index": run_state.tool_index, - "result": result, - "status": "complete", + "tool_call_id": tc.get("id"), + "content": result, }) - run_state.tool_index += 1 return {"messages": out} def should_continue(state: ResearchState) -> Literal["tools", "__end__"]: @@ -450,9 +449,11 @@ async def research( to populate `agent.subagents()` for the chat-subagents primitive). Always pass a stable identifier like "research". - The subagent run is also surfaced to the UI as a native AG-UI ACTIVITY - (activityType "subagent"): started → reason/tool/answer transcript → - finished, keyed by this tool's own call id. + The subagent run is also surfaced to the UI as the protocol's standard + subagent events (SUBAGENT_STARTED → attributed reason/tool/answer + transcript → SUBAGENT_FINISHED / SUBAGENT_ERROR), keyed by this tool's + own call id: the graph dispatches `subagent_activity` CUSTOM payloads + and the server's SubagentEmittingAgent expands them on the wire. """ async def _emit(payload: dict) -> None: @@ -463,14 +464,32 @@ async def _emit(payload: dict) -> None: except Exception: pass - run_state = SubagentRunState() + run_state = SubagentRunState(tool_call_id) subgraph = _build_research_subgraph(_emit, run_state) await _emit({"phase": "started", "name": subagent_type}) - result = await subgraph.ainvoke( - {"topic": topic, "messages": [], "iterations": 0}, - config={"callbacks": [SubagentStreamHandler(tool_call_id, run_state)]}, - ) + try: + result = await subgraph.ainvoke( + {"topic": topic, "messages": [], "iterations": 0}, + config={ + "callbacks": [SubagentStreamHandler(tool_call_id, run_state)], + # ag-ui-langgraph streams this compiled subgraph by default and + # would put the child's own LLM text and `lookup` tool call on + # the wire as UNATTRIBUTED TEXT_MESSAGE_* / TOOL_CALL_* events + # in the parent transcript. `emit-messages` / `emit-tool-calls` + # is the bridge's declared opt-out (read from LangChain run + # metadata, which inherits into the subgraph run): with them + # False the child's raw events never reach the wire, while + # callbacks (SubagentStreamHandler, adispatch_custom_event) and + # the child's STEP_* still fire. The attributed copies come from + # the `subagent_activity` payloads the SubagentEmittingAgent + # expands. + "metadata": {"emit-messages": False, "emit-tool-calls": False}, + }, + ) + except Exception as exc: + await _emit({"phase": "error", "message": f"{type(exc).__name__}: {exc}"}) + raise await _emit({"phase": "finished", "status": "complete"}) msgs = result.get("messages") if isinstance(result, dict) else None diff --git a/examples/ag-ui/python/src/server.py b/examples/ag-ui/python/src/server.py index 9db4e41e7..2c47b35ab 100644 --- a/examples/ag-ui/python/src/server.py +++ b/examples/ag-ui/python/src/server.py @@ -15,7 +15,7 @@ from ag_ui_langgraph import add_langgraph_fastapi_endpoint from .graph import _builder -from .streaming.activity_emitting_agent import ActivityEmittingAgent +from .streaming.subagent_emitting_agent import SubagentEmittingAgent # The exported graph is checkpointer-free for LangGraph Platform (which manages # persistence). The standalone ag-ui-langgraph endpoint reads graph state via @@ -42,4 +42,8 @@ def ok() -> dict: return {"ok": True} -add_langgraph_fastapi_endpoint(app, ActivityEmittingAgent(name="chat", graph=graph), path="/agent") +# SubagentEmittingAgent expands the research subagent's `subagent_activity` +# CUSTOM events into the protocol's SUBAGENT_* + subagentRunId-attributed +# content events (see streaming/subagent_emitting_agent.py). +agent = SubagentEmittingAgent(name="chat", graph=graph) +add_langgraph_fastapi_endpoint(app, agent, path="/agent") diff --git a/examples/ag-ui/python/src/streaming/activity_emitting_agent.py b/examples/ag-ui/python/src/streaming/activity_emitting_agent.py deleted file mode 100644 index e1b8f6fac..000000000 --- a/examples/ag-ui/python/src/streaming/activity_emitting_agent.py +++ /dev/null @@ -1,11 +0,0 @@ -"""LangGraphAgent subclass that converts subagent_activity CUSTOM events to -native AG-UI ACTIVITY events at the bridge's 1:1 dispatch point. Owned transport -adapter — keeps the wire protocol-native without patching the bridge.""" -from ag_ui_langgraph import LangGraphAgent -from src.streaming.activity_transform import subagent_custom_to_activity - - -class ActivityEmittingAgent(LangGraphAgent): - def _dispatch_event(self, event): - activity = subagent_custom_to_activity(event) - return super()._dispatch_event(activity if activity is not None else event) diff --git a/examples/ag-ui/python/src/streaming/activity_transform.py b/examples/ag-ui/python/src/streaming/activity_transform.py deleted file mode 100644 index e992f8eb8..000000000 --- a/examples/ag-ui/python/src/streaming/activity_transform.py +++ /dev/null @@ -1,143 +0,0 @@ -"""Maps a `subagent_activity` CUSTOM event (emitted by the research tool / -SubagentStreamHandler via adispatch_custom_event) to a native AG-UI ACTIVITY event. - -Pure and stateless (1:1): build patches purely from the event fields — never -track state in the transform. Anything that is not a `subagent_activity` CUSTOM -event returns None. - -Supported phases: started, message_start, message, tool_call, tool_result, -finished. Unknown phases return None. -""" -import json -from typing import Optional - -from ag_ui.core import ActivityDeltaEvent, ActivitySnapshotEvent, BaseEvent, EventType - -ACTIVITY_TYPE = "subagent" -_CUSTOM_NAME = "subagent_activity" - - -def subagent_custom_to_activity(event: BaseEvent) -> Optional[BaseEvent]: - if getattr(event, "type", None) != EventType.CUSTOM: - return None - if getattr(event, "name", None) != _CUSTOM_NAME: - return None - value = getattr(event, "value", None) - if isinstance(value, str): # bridge may JSON-serialize custom values - try: - value = json.loads(value) - except json.JSONDecodeError: - return None - if not isinstance(value, dict): - return None - - sid = value.get("subagent_id") - phase = value.get("phase") - if not sid or not phase: - return None - - if phase == "started": - return ActivitySnapshotEvent( - type=EventType.ACTIVITY_SNAPSHOT, - message_id=sid, - activity_type=ACTIVITY_TYPE, - content={ - "toolCallId": sid, - "name": value.get("name"), - "status": "running", - "messages": [], - "toolCalls": [], - }, - replace=True, - ) - - if phase == "message_start": - message_index = value.get("message_index") - return ActivityDeltaEvent( - type=EventType.ACTIVITY_DELTA, - message_id=sid, - activity_type=ACTIVITY_TYPE, - patch=[ - { - "op": "add", - "path": "/messages/-", - "value": { - "id": f"{sid}-{message_index}", - "role": "assistant", - "content": "", - "toolCallIds": [], - }, - } - ], - ) - - if phase == "message": - message_index = value.get("message_index") - return ActivityDeltaEvent( - type=EventType.ACTIVITY_DELTA, - message_id=sid, - activity_type=ACTIVITY_TYPE, - patch=[ - { - "op": "replace", - "path": f"/messages/{message_index}/content", - "value": value.get("text", ""), - } - ], - ) - - if phase == "tool_call": - message_index = value.get("message_index") - tool_call_id = value.get("tool_call_id") - return ActivityDeltaEvent( - type=EventType.ACTIVITY_DELTA, - message_id=sid, - activity_type=ACTIVITY_TYPE, - patch=[ - { - "op": "add", - "path": "/toolCalls/-", - "value": { - "id": tool_call_id, - "name": value.get("name"), - "args": value.get("args"), - "status": "running", - }, - }, - { - "op": "add", - "path": f"/messages/{message_index}/toolCallIds/-", - "value": tool_call_id, - }, - ], - ) - - if phase == "tool_result": - tool_index = value.get("tool_index") - return ActivityDeltaEvent( - type=EventType.ACTIVITY_DELTA, - message_id=sid, - activity_type=ACTIVITY_TYPE, - patch=[ - { - "op": "replace", - "path": f"/toolCalls/{tool_index}/status", - "value": value.get("status", "complete"), - }, - { - "op": "replace", - "path": f"/toolCalls/{tool_index}/result", - "value": value.get("result"), - }, - ], - ) - - if phase == "finished": - return ActivityDeltaEvent( - type=EventType.ACTIVITY_DELTA, - message_id=sid, - activity_type=ACTIVITY_TYPE, - patch=[{"op": "replace", "path": "/status", "value": value.get("status", "complete")}], - ) - - return None diff --git a/examples/ag-ui/python/src/streaming/subagent_emitting_agent.py b/examples/ag-ui/python/src/streaming/subagent_emitting_agent.py new file mode 100644 index 000000000..7134c1d4e --- /dev/null +++ b/examples/ag-ui/python/src/streaming/subagent_emitting_agent.py @@ -0,0 +1,271 @@ +# SPDX-License-Identifier: MIT +"""SUBAGENT_* emitter for the `research` delegation tool. + +The graph cannot reach the AG-UI wire directly: the research subgraph's nodes, +the `research` tool body and `SubagentStreamHandler` dispatch +`subagent_activity` CUSTOM events through LangChain's +``adispatch_custom_event``, which the ag-ui-langgraph bridge forwards 1:1 as +``CustomEvent`` items in ``LangGraphAgent.run`` (the async generator the +FastAPI endpoint consumes). The standard sequence needs 1:N expansion — a +``tool_call`` phase becomes three ``TOOL_CALL_*`` events, a ``finished`` +phase must close the open child message AND finish the subagent, and the +CUSTOM event itself must be consumed — so the seam is ``run`` rather than the +bridge's strictly one-in/one-out ``_dispatch_event`` hook (measured in +docs/wire-capture-subagents.md). + +Expansion contract (``tid`` = the payload's ``subagent_id`` = the ``research`` +tool call id, identical to the bridge's ``TOOL_CALL_START.toolCallId``): + + started {subagent_id, name} → SUBAGENT_STARTED {subagentRunId: -sub, name, parentToolCallId: } + message_start {subagent_id, message_id} → TEXT_MESSAGE_START {messageId: -sub-m, role: assistant, subagentRunId} + message {subagent_id, message_id, delta} → TEXT_MESSAGE_CONTENT {messageId, delta, subagentRunId} + (next message_start / tool_call / finished / error) → TEXT_MESSAGE_END for any open message first + tool_call {subagent_id, tool_call_id, name, args} → TOOL_CALL_START {toolCallId, toolCallName, parentMessageId: , subagentRunId} + + TOOL_CALL_ARGS {delta: json(args)} + TOOL_CALL_END + tool_result {subagent_id, tool_call_id, content} → TOOL_CALL_RESULT {messageId: -result, toolCallId, content, role: tool, subagentRunId} + finished {subagent_id} → SUBAGENT_FINISHED {subagentRunId, outcome: success} + error {subagent_id, message} → SUBAGENT_ERROR {subagentRunId, message} + +Unknown phases are dropped with a warning; malformed payloads are dropped; +CUSTOM events with any other name pass through untouched. No queue merge is +needed: the CUSTOM events already flow through the bridge generator live, +interleaved with the bridge's own events, so a plain ``for out in +expand(ev): yield out`` preserves streaming. Delegation state is per +``run()`` call (the endpoint clones the agent per request anyway). + +The child here is a compiled SUBGRAPH, and the bridge streams subgraphs — +so left alone it would ALSO emit the child's own LLM text and ``lookup`` tool +call as unattributed, bridge-native ``TEXT_MESSAGE_*`` / ``TOOL_CALL_*`` +events in the parent transcript (wire capture §2b). The graph opts the child +out at the source: the ``research`` tool invokes the subgraph with LangChain +run metadata ``emit-messages`` / ``emit-tool-calls`` = ``False``, the bridge's +declared switch for skipping those emissions while callbacks, CUSTOM events +and ``STEP_*`` still flow (see ``src/graph.py``). This wrapper therefore never +filters bridge-native events — every non-``subagent_activity`` event passes +through untouched. + +The encoder requires pydantic ``BaseEvent`` instances — raw dicts crash the +stream — so only typed ``ag_ui.core`` events are yielded. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from typing import Any, AsyncGenerator, Iterator + +from ag_ui.core import ( + BaseEvent, + EventType, + SubagentErrorEvent, + SubagentFinishedEvent, + SubagentFinishedSuccessOutcome, + SubagentStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, + ToolCallArgsEvent, + ToolCallEndEvent, + ToolCallResultEvent, + ToolCallStartEvent, +) +from ag_ui_langgraph import LangGraphAgent + +CUSTOM_NAME = "subagent_activity" + +logger = logging.getLogger(__name__) + + +@dataclass +class _Delegation: + """Lifecycle of one delegation call, keyed by parent tool-call id.""" + + run_id: str + open_message_id: str | None = None + message_count: int = 0 + + +@dataclass +class _RunState: + """Per-``run()`` expansion state.""" + + delegations: dict[str, _Delegation] = field(default_factory=dict) + + +def _subagent_run_id(tid: str) -> str: + return f"{tid}-sub" + + +def _message_id(tid: str, n: int) -> str: + return f"{tid}-sub-m{n}" + + +def _as_text(content: Any) -> str: + if isinstance(content, str): + return content + if content is None: + return "" + try: + return json.dumps(content) + except (TypeError, ValueError): + return str(content) + + +def _payload(event: BaseEvent) -> dict[str, Any] | None: + """Return the `subagent_activity` payload dict, or None if `event` is not + one (or is malformed).""" + if getattr(event, "type", None) != EventType.CUSTOM: + return None + if getattr(event, "name", None) != CUSTOM_NAME: + return None + value = getattr(event, "value", None) + if isinstance(value, str): # the bridge may JSON-serialize custom values + try: + value = json.loads(value) + except json.JSONDecodeError: + logger.warning("subagent_activity payload is not JSON; dropped") + return {} + if not isinstance(value, dict): + logger.warning("subagent_activity payload is not an object; dropped") + return {} + return value + + +class SubagentEmittingAgent(LangGraphAgent): + """LangGraphAgent whose ``run`` expands the graph's `subagent_activity` + CUSTOM events into standard SUBAGENT_* + attributed TEXT_MESSAGE_* / + TOOL_CALL_* events. Everything else passes through untouched. + + Keeps the bridge's ``__init__`` signature so ``clone()`` (called by the + FastAPI endpoint per request) reconstructs this subclass. + """ + + async def run(self, *args: Any, **kwargs: Any) -> AsyncGenerator[BaseEvent, None]: + state = _RunState() + async for event in super().run(*args, **kwargs): + for out in self._expand(event, state): + yield out + + def _expand(self, event: BaseEvent, state: _RunState) -> Iterator[BaseEvent]: + payload = _payload(event) + if payload is None: + yield event + return + if not payload: + return # malformed — already logged + tid = payload.get("subagent_id") + phase = payload.get("phase") + if not isinstance(tid, str) or not tid or not isinstance(phase, str): + logger.warning("subagent_activity missing subagent_id/phase; dropped: %r", payload) + return + + delegation = state.delegations.get(tid) + if delegation is None: + delegation = _Delegation(run_id=_subagent_run_id(tid)) + state.delegations[tid] = delegation + run_id = delegation.run_id + + if phase == "started": + yield SubagentStartedEvent( + type=EventType.SUBAGENT_STARTED, + subagent_run_id=run_id, + name=str(payload.get("name") or tid), + parent_tool_call_id=tid, + ) + elif phase == "message_start": + yield from self._close_message(delegation) + yield from self._open_message(delegation, tid, payload.get("message_id")) + elif phase == "message": + delta = payload.get("delta") + if not isinstance(delta, str) or not delta: + return + if delegation.open_message_id is None: + yield from self._open_message(delegation, tid, payload.get("message_id")) + yield TextMessageContentEvent( + type=EventType.TEXT_MESSAGE_CONTENT, + message_id=delegation.open_message_id, + delta=delta, + subagent_run_id=run_id, + ) + elif phase == "tool_call": + tool_call_id = payload.get("tool_call_id") + if not isinstance(tool_call_id, str) or not tool_call_id: + logger.warning("subagent_activity tool_call missing tool_call_id; dropped: %r", payload) + return + parent_message_id = delegation.open_message_id + yield from self._close_message(delegation) + yield ToolCallStartEvent( + type=EventType.TOOL_CALL_START, + tool_call_id=tool_call_id, + tool_call_name=str(payload.get("name") or tool_call_id), + parent_message_id=parent_message_id, + subagent_run_id=run_id, + ) + args = payload.get("args") + yield ToolCallArgsEvent( + type=EventType.TOOL_CALL_ARGS, + tool_call_id=tool_call_id, + delta=_as_text(args if args is not None else {}), + subagent_run_id=run_id, + ) + yield ToolCallEndEvent( + type=EventType.TOOL_CALL_END, + tool_call_id=tool_call_id, + subagent_run_id=run_id, + ) + elif phase == "tool_result": + tool_call_id = payload.get("tool_call_id") + if not isinstance(tool_call_id, str) or not tool_call_id: + logger.warning("subagent_activity tool_result missing tool_call_id; dropped: %r", payload) + return + yield ToolCallResultEvent( + type=EventType.TOOL_CALL_RESULT, + message_id=f"{tool_call_id}-result", + tool_call_id=tool_call_id, + content=_as_text(payload.get("content")), + role="tool", + subagent_run_id=run_id, + ) + elif phase == "finished": + yield from self._close_message(delegation) + yield SubagentFinishedEvent( + type=EventType.SUBAGENT_FINISHED, + subagent_run_id=run_id, + outcome=SubagentFinishedSuccessOutcome(), + ) + elif phase == "error": + yield from self._close_message(delegation) + yield SubagentErrorEvent( + type=EventType.SUBAGENT_ERROR, + subagent_run_id=run_id, + message=str(payload.get("message") or "subagent failed"), + ) + else: + logger.warning("subagent_activity phase %r not supported; dropped", phase) + + @staticmethod + def _open_message( + delegation: _Delegation, tid: str, message_id: Any + ) -> Iterator[BaseEvent]: + delegation.message_count += 1 + if not isinstance(message_id, str) or not message_id: + message_id = _message_id(tid, delegation.message_count) + delegation.open_message_id = message_id + yield TextMessageStartEvent( + type=EventType.TEXT_MESSAGE_START, + message_id=message_id, + role="assistant", + subagent_run_id=delegation.run_id, + ) + + @staticmethod + def _close_message(delegation: _Delegation) -> Iterator[BaseEvent]: + if delegation.open_message_id is None: + return + message_id, delegation.open_message_id = delegation.open_message_id, None + yield TextMessageEndEvent( + type=EventType.TEXT_MESSAGE_END, + message_id=message_id, + subagent_run_id=delegation.run_id, + ) diff --git a/examples/ag-ui/python/src/streaming/subagent_stream_handler.py b/examples/ag-ui/python/src/streaming/subagent_stream_handler.py index 1a00a1cc1..b1efaf058 100644 --- a/examples/ag-ui/python/src/streaming/subagent_stream_handler.py +++ b/examples/ag-ui/python/src/streaming/subagent_stream_handler.py @@ -1,60 +1,74 @@ -"""Taps a child subagent LLM's text tokens and emits them as `subagent_activity` -`message` events, keyed by the parent tool_call_id. Accumulates `text_so_far` -so the L2 transform stays stateless. `started`/`finished` are emitted by the -research tool body. Uses adispatch_custom_event (the bridge reads on_custom_event -from astream_events; get_stream_writer would surface only as a RAW event). - -Each `message` event also carries the current `message_index` — the 0-based -ordinal of the assistant turn the tokens belong to. The subgraph owns the -counter (it opens each turn with a `message_start`); the handler reads it -through a shared mutable ref (`SubagentRunState`) so the index it tags stays -in lock-step with the transcript the subgraph emits. The handler resets its -text buffer whenever the subgraph advances to a new turn so each message's -`text_so_far` starts fresh.""" +"""Taps a child subagent LLM's text tokens and forwards each one as a +`subagent_activity` payload keyed by the parent tool_call_id: + + message_start {subagent_id, message_id} once per assistant turn + message {subagent_id, message_id, delta} one per token (raw delta) + +`SubagentEmittingAgent` turns those into `subagentRunId`-attributed +TEXT_MESSAGE_START / TEXT_MESSAGE_CONTENT events and closes the message +(TEXT_MESSAGE_END) itself at the next `message_start` / `tool_call` / +`finished` / `error`. `started` / `tool_call` / `tool_result` / `finished` / +`error` are emitted by the research subgraph nodes and the `research` tool +body. + +Message ids follow the `-sub-m` convention. The research +subgraph runs several assistant turns per delegation (reason → tool → answer), +so it owns the turn counter: its `agent` node calls +`SubagentRunState.open_message()` and dispatches `message_start` BEFORE +invoking the model, and the handler reads the open id for the tokens that +follow. When no turn is open (a bare LLM call outside the subgraph, or the +unit tests) the handler opens one itself so a token is never orphaned. + +Uses adispatch_custom_event (the bridge reads on_custom_event from +astream_events; get_stream_writer would surface only as a RAW event).""" from typing import Any, Optional from uuid import UUID from langchain_core.callbacks import AsyncCallbackHandler, adispatch_custom_event +CUSTOM_NAME = "subagent_activity" + class SubagentRunState: - """Per-research-run shared state. The subgraph nodes own `message_index` - (bumping it as each assistant turn opens) and `tool_index` (the running - position in the run's toolCalls[]); the SubagentStreamHandler reads - `message_index` so its streamed `message` events tag the right turn.""" + """Per-delegation shared state: the message counter that derives + `-sub-m` ids and the id of the currently open assistant + turn. The research subgraph's nodes advance it; the SubagentStreamHandler + reads it so its streamed `message` deltas tag the right turn.""" + + def __init__(self, subagent_id: str) -> None: + self.subagent_id = subagent_id + self.message_count: int = 0 + self.message_id: Optional[str] = None - def __init__(self) -> None: - self.message_index: int = 0 - self.tool_index: int = 0 + def open_message(self) -> str: + """Advance to the next assistant turn and return its message id.""" + self.message_count += 1 + self.message_id = f"{self.subagent_id}-sub-m{self.message_count}" + return self.message_id class SubagentStreamHandler(AsyncCallbackHandler): def __init__(self, subagent_id: str, run_state: Optional[SubagentRunState] = None) -> None: self._id = subagent_id - self._buffer = "" - self._run_state = run_state if run_state is not None else SubagentRunState() - # Track which turn the current buffer belongs to so we reset the - # accumulated text when the subgraph advances to a new assistant turn. - self._buffer_index = self._run_state.message_index + self._run_state = run_state if run_state is not None else SubagentRunState(subagent_id) async def on_llm_new_token(self, token: str, *, run_id: UUID | None = None, **kwargs: Any) -> None: if not token: return - index = self._run_state.message_index - if index != self._buffer_index: - # New assistant turn opened since the last token — start fresh so - # `text_so_far` is scoped to this message, not the whole run. - self._buffer = "" - self._buffer_index = index - self._buffer += token try: + if self._run_state.message_id is None: + message_id = self._run_state.open_message() + await adispatch_custom_event( + CUSTOM_NAME, + {"subagent_id": self._id, "phase": "message_start", "message_id": message_id}, + ) await adispatch_custom_event( - "subagent_activity", + CUSTOM_NAME, { "subagent_id": self._id, "phase": "message", - "message_index": index, - "text": self._buffer, + "message_id": self._run_state.message_id, + "delta": token, }, ) except Exception: diff --git a/examples/ag-ui/python/tests/test_activity_transform.py b/examples/ag-ui/python/tests/test_activity_transform.py deleted file mode 100644 index 24c4baf5b..000000000 --- a/examples/ag-ui/python/tests/test_activity_transform.py +++ /dev/null @@ -1,212 +0,0 @@ -"""Tests for subagent_custom_to_activity — maps a `subagent_activity` CUSTOM -event to a native ACTIVITY event (1:1, stateless). Non-subagent events → None.""" -from ag_ui.core import CustomEvent, EventType, TextMessageStartEvent -from src.streaming.activity_transform import subagent_custom_to_activity - - -def _custom(data: dict) -> CustomEvent: - return CustomEvent(type=EventType.CUSTOM, name="subagent_activity", value=data) - - -def test_started_maps_to_activity_snapshot(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "started", "name": "research"})) - assert ev.type == EventType.ACTIVITY_SNAPSHOT - assert ev.message_id == "tc-1" - assert ev.activity_type == "subagent" - assert ev.content == { - "toolCallId": "tc-1", - "name": "research", - "status": "running", - "messages": [], - "toolCalls": [], - } - assert ev.replace is True - - -def test_message_start_maps_to_activity_delta_add_message(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "message_start", "message_index": 0})) - assert ev.type == EventType.ACTIVITY_DELTA - assert ev.message_id == "tc-1" - assert ev.patch == [ - { - "op": "add", - "path": "/messages/-", - "value": {"id": "tc-1-0", "role": "assistant", "content": "", "toolCallIds": []}, - } - ] - - -def test_message_start_uses_message_index_in_id(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "message_start", "message_index": 2})) - assert ev.patch[0]["value"]["id"] == "tc-1-2" - - -def test_message_maps_to_activity_delta_replace_content(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "message", "message_index": 0, "text": "Paris is"})) - assert ev.type == EventType.ACTIVITY_DELTA - assert ev.message_id == "tc-1" - assert ev.patch == [{"op": "replace", "path": "/messages/0/content", "value": "Paris is"}] - - -def test_message_uses_correct_index_in_path(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "message", "message_index": 3, "text": "hello"})) - assert ev.patch == [{"op": "replace", "path": "/messages/3/content", "value": "hello"}] - - -def test_tool_call_maps_to_two_op_patch(): - ev = subagent_custom_to_activity(_custom({ - "subagent_id": "tc-1", - "phase": "tool_call", - "message_index": 0, - "tool_call_id": "call-abc", - "name": "search", - "args": {"query": "Paris"}, - })) - assert ev.type == EventType.ACTIVITY_DELTA - assert ev.message_id == "tc-1" - assert ev.patch == [ - { - "op": "add", - "path": "/toolCalls/-", - "value": {"id": "call-abc", "name": "search", "args": {"query": "Paris"}, "status": "running"}, - }, - { - "op": "add", - "path": "/messages/0/toolCallIds/-", - "value": "call-abc", - }, - ] - - -def test_tool_call_uses_message_index_in_path(): - ev = subagent_custom_to_activity(_custom({ - "subagent_id": "tc-1", - "phase": "tool_call", - "message_index": 2, - "tool_call_id": "call-xyz", - "name": "lookup", - "args": {}, - })) - assert ev.patch[1]["path"] == "/messages/2/toolCallIds/-" - - -def test_tool_result_maps_to_two_op_patch(): - ev = subagent_custom_to_activity(_custom({ - "subagent_id": "tc-1", - "phase": "tool_result", - "tool_index": 0, - "result": "Paris, France", - "status": "complete", - })) - assert ev.type == EventType.ACTIVITY_DELTA - assert ev.message_id == "tc-1" - assert ev.patch == [ - {"op": "replace", "path": "/toolCalls/0/status", "value": "complete"}, - {"op": "replace", "path": "/toolCalls/0/result", "value": "Paris, France"}, - ] - - -def test_tool_result_defaults_status_to_complete(): - ev = subagent_custom_to_activity(_custom({ - "subagent_id": "tc-1", - "phase": "tool_result", - "tool_index": 1, - "result": "some result", - })) - assert ev.patch[0] == {"op": "replace", "path": "/toolCalls/1/status", "value": "complete"} - - -def test_tool_result_uses_tool_index_in_path(): - ev = subagent_custom_to_activity(_custom({ - "subagent_id": "tc-1", - "phase": "tool_result", - "tool_index": 3, - "result": "done", - })) - assert ev.patch[0]["path"] == "/toolCalls/3/status" - assert ev.patch[1]["path"] == "/toolCalls/3/result" - - -def test_finished_maps_to_activity_delta_replace_status(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "finished", "status": "complete"})) - assert ev.type == EventType.ACTIVITY_DELTA - assert ev.patch == [{"op": "replace", "path": "/status", "value": "complete"}] - - -def test_finished_defaults_status_to_complete(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "finished"})) - assert ev.patch == [{"op": "replace", "path": "/status", "value": "complete"}] - - -def test_finished_custom_status(): - ev = subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "finished", "status": "error"})) - assert ev.patch == [{"op": "replace", "path": "/status", "value": "error"}] - - -def test_unknown_phase_returns_none(): - assert subagent_custom_to_activity(_custom( - {"subagent_id": "tc-1", "phase": "unknown_phase"})) is None - - -def test_non_subagent_event_returns_none(): - assert subagent_custom_to_activity( - CustomEvent(type=EventType.CUSTOM, name="state_update", value={})) is None - assert subagent_custom_to_activity( - TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, message_id="m", role="assistant")) is None - - -def test_malformed_json_string_value_returns_none(): - ev = CustomEvent(type=EventType.CUSTOM, name="subagent_activity", value="not json {{{") - assert subagent_custom_to_activity(ev) is None - - -import pytest -from langgraph.graph import StateGraph, END -from langchain_core.callbacks.manager import adispatch_custom_event -from langgraph.checkpoint.memory import MemorySaver -from typing_extensions import TypedDict -from ag_ui.core import RunAgentInput -from src.streaming.activity_emitting_agent import ActivityEmittingAgent - - -class _S(TypedDict): - messages: list - - -# Emit via adispatch_custom_event (the LangChain callback API) — the SPIKE found -# that a plain get_stream_writer() payload surfaces only as an on_chain_stream -# RAW event in this bridge/LangGraph version and never becomes a discrete CUSTOM -# event at _dispatch_event, whereas adispatch_custom_event does. Layer 3's -# SubagentStreamHandler must use this same mechanism. -async def _emit_node(state: _S) -> dict: - await adispatch_custom_event( - "subagent_activity", - {"subagent_id": "tc-1", "phase": "started", "name": "research"}) - return {"messages": []} - - -def _tiny_graph(): - g = StateGraph(_S) - g.add_node("emit", _emit_node) - g.set_entry_point("emit") - g.add_edge("emit", END) - return g.compile(checkpointer=MemorySaver()) - - -@pytest.mark.asyncio -async def test_dispatch_event_seam_converts_custom_to_activity(): - agent = ActivityEmittingAgent(name="t", graph=_tiny_graph()) - run_input = RunAgentInput(thread_id="th", run_id="r", messages=[], - tools=[], context=[], state={}, forwarded_props={}) - types = [getattr(ev, "type", None) async for ev in agent.run(run_input)] - assert EventType.ACTIVITY_SNAPSHOT in types - assert EventType.CUSTOM not in [t for t in types] - assert isinstance(agent.clone(), ActivityEmittingAgent) diff --git a/examples/ag-ui/python/tests/test_subagent_emission.py b/examples/ag-ui/python/tests/test_subagent_emission.py index 051953cff..b51c26546 100644 --- a/examples/ag-ui/python/tests/test_subagent_emission.py +++ b/examples/ag-ui/python/tests/test_subagent_emission.py @@ -1,28 +1,41 @@ """In-process verification of the research subagent's reason → tool → answer -loop and its structured `subagent_activity` transcript emission. +loop and the standard SUBAGENT_* sequence it produces on the wire. Runs the enriched subgraph with a FAKE tool-calling chat model (no network): turn 0 returns an AIMessage carrying a `lookup` tool_call; turn 1 returns the -final answer. We intercept every `adispatch_custom_event` the subgraph nodes -fire and assert the ORDER + payloads of the structured phases: - - message_start(0) → tool_call(0, …) → tool_result(0, …) - → message_start(1) → finished-by-tool? no → final answer - -(Live `message` token events come from SubagentStreamHandler, which the fake -model doesn't drive — so this test asserts the node-emitted phases only, which -is exactly the transcript skeleton the L2 transform consumes.) +final answer. We intercept every `subagent_activity` payload the subgraph +nodes dispatch, bracket them with the `research` tool body's `started` / +`finished`, and push them through `SubagentEmittingAgent` exactly as the +bridge would (as CUSTOM events in `LangGraphAgent.run`) — asserting the +ORDER + fields of the standard events: + + SUBAGENT_STARTED → TEXT_MESSAGE_START(m1) → TEXT_MESSAGE_END(m1) + → TOOL_CALL_START/ARGS/END(lookup, parent m1) → TOOL_CALL_RESULT(lookup) + → TEXT_MESSAGE_START(m2) → TEXT_MESSAGE_END(m2) → SUBAGENT_FINISHED + +(Live `message` deltas come from SubagentStreamHandler, which the fake model +does not drive — so the answer turn has no TEXT_MESSAGE_CONTENT here; the +handler's own tests cover the per-token deltas.) """ +import json from typing import Any import pytest +from ag_ui.core import CustomEvent, EventType, RunAgentInput +from ag_ui_langgraph import LangGraphAgent from langchain_core.messages import AIMessage from langchain_core.runnables import Runnable +from langgraph.graph import END, MessagesState, StateGraph import src.graph as graph_mod -from src.graph import _build_research_subgraph +from src.graph import _build_research_subgraph, research +from src.streaming.subagent_emitting_agent import SubagentEmittingAgent from src.streaming.subagent_stream_handler import SubagentRunState +TID = "tc-research" +RUN_ID = f"{TID}-sub" +M1, M2 = f"{TID}-sub-m1", f"{TID}-sub-m2" + class _FakeToolCallingModel(Runnable): """A tiny Runnable standing in for ChatOpenAI. First invocation returns an @@ -55,53 +68,64 @@ async def ainvoke(self, input: Any, config: Any = None, **kwargs: Any) -> AIMess return self.invoke(input, config, **kwargs) -@pytest.mark.asyncio -async def test_subgraph_emits_reason_tool_answer_transcript(): - events: list[dict] = [] +def _noop_graph(): + g = StateGraph(MessagesState) + g.add_node("noop", lambda state: {}) + g.set_entry_point("noop") + g.add_edge("noop", END) + return g.compile() + + +async def _run_subgraph_and_collect_payloads() -> tuple[list[dict], dict, _FakeToolCallingModel]: + payloads: list[dict] = [] async def fake_emit(payload: dict) -> None: - events.append({"subagent_id": "tc-research", **payload}) + payloads.append({"subagent_id": TID, **payload}) fake_model = _FakeToolCallingModel() - run_state = SubagentRunState() subgraph = _build_research_subgraph( - fake_emit, run_state, llm_factory=lambda force_answer: fake_model + fake_emit, SubagentRunState(TID), llm_factory=lambda force_answer: fake_model ) + result = await subgraph.ainvoke({"topic": "Angular signals", "messages": [], "iterations": 0}) + return payloads, result, fake_model - result = await subgraph.ainvoke( - {"topic": "Angular signals", "messages": [], "iterations": 0} - ) - phases = [(e["phase"], e) for e in events] - phase_names = [p for p, _ in phases] +async def _expand(monkeypatch, payloads: list[dict]) -> list: + async def fake_run(self, input): + for p in payloads: + yield CustomEvent(type=EventType.CUSTOM, name="subagent_activity", value=p) + + monkeypatch.setattr(LangGraphAgent, "run", fake_run) + agent = SubagentEmittingAgent(name="chat", graph=_noop_graph()) + run_input = RunAgentInput( + thread_id="t", run_id="r", messages=[], tools=[], context=[], state={}, forwarded_props={} + ) + return [ev async for ev in agent.run(run_input)] - # Full structured phase sequence the node loop emits. - assert phase_names == [ - "message_start", # turn 0 opens - "tool_call", # turn 0 calls lookup - "tool_result", # tool node runs lookup - "message_start", # turn 1 opens (forced-answer turn) - ], phase_names - by_phase = {p: e for p, e in phases} +@pytest.mark.asyncio +async def test_subgraph_emits_reason_tool_answer_phases(): + payloads, result, fake_model = await _run_subgraph_and_collect_payloads() - # message_start indices: 0 then 1. - starts = [e["message_index"] for p, e in phases if p == "message_start"] - assert starts == [0, 1], starts + # Full structured phase sequence the node loop dispatches. + assert [p["phase"] for p in payloads] == [ + "message_start", # turn 1 opens + "tool_call", # turn 1 calls lookup + "tool_result", # tool node runs lookup + "message_start", # turn 2 opens (forced-answer turn) + ] + by_phase = {p["phase"]: p for p in payloads} + assert [p["message_id"] for p in payloads if p["phase"] == "message_start"] == [M1, M2] - # tool_call carries id/name/args + the originating message_index (0). tc = by_phase["tool_call"] - assert tc["message_index"] == 0 + assert tc["message_id"] == M1 assert tc["tool_call_id"] == "call_lookup_1" assert tc["name"] == "lookup" assert tc["args"] == {"query": "angular signals"} - # tool_result carries the matching tool_index (0), the lookup result text, - # and a complete status. tr = by_phase["tool_result"] - assert tr["tool_index"] == 0 - assert tr["status"] == "complete" - assert isinstance(tr["result"], str) and "signal" in tr["result"].lower() + assert tr["tool_call_id"] == "call_lookup_1" + assert isinstance(tr["content"], str) and "signal" in tr["content"].lower() # Loop terminates: the forced-answer turn returns a plain answer (no tool # calls), so the run ends with a final AIMessage and exactly two turns. @@ -112,6 +136,43 @@ async def fake_emit(payload: dict) -> None: assert "Signals" in last.content +@pytest.mark.asyncio +async def test_subgraph_run_expands_to_the_standard_subagent_sequence(monkeypatch): + payloads, _, _ = await _run_subgraph_and_collect_payloads() + # Bracket with what the `research` tool body dispatches around ainvoke. + script = [ + {"subagent_id": TID, "phase": "started", "name": "research"}, + *payloads, + {"subagent_id": TID, "phase": "finished", "status": "complete"}, + ] + out = await _expand(monkeypatch, script) + + assert [(ev.type, getattr(ev, "message_id", None) or getattr(ev, "tool_call_id", None)) for ev in out] == [ + (EventType.SUBAGENT_STARTED, None), + (EventType.TEXT_MESSAGE_START, M1), + (EventType.TEXT_MESSAGE_END, M1), + (EventType.TOOL_CALL_START, "call_lookup_1"), + (EventType.TOOL_CALL_ARGS, "call_lookup_1"), + (EventType.TOOL_CALL_END, "call_lookup_1"), + (EventType.TOOL_CALL_RESULT, "call_lookup_1-result"), + (EventType.TEXT_MESSAGE_START, M2), + (EventType.TEXT_MESSAGE_END, M2), + (EventType.SUBAGENT_FINISHED, None), + ] + assert not any(ev.type == EventType.CUSTOM for ev in out) + assert all(ev.subagent_run_id == RUN_ID for ev in out) + + assert out[0].name == "research" + assert out[0].parent_tool_call_id == TID + assert out[3].tool_call_name == "lookup" + assert out[3].parent_message_id == M1 + assert json.loads(out[4].delta) == {"query": "angular signals"} + assert out[6].tool_call_id == "call_lookup_1" + assert out[6].role == "tool" + assert "signal" in out[6].content.lower() + assert out[9].outcome.type == "success" + + @pytest.mark.asyncio async def test_lookup_tool_is_deterministic_and_offline(): # The canned fact lookup must be reproducible for the aimock fixture. @@ -120,3 +181,37 @@ async def test_lookup_tool_is_deterministic_and_offline(): # Unknown topic falls back to the default fact, never raises / hits network. assert graph_mod.lookup.invoke({"query": "quantum widgets"}) == \ graph_mod._RESEARCH_DEFAULT_FACT + + +@pytest.mark.asyncio +async def test_research_tool_declares_the_child_silent_via_bridge_metadata(monkeypatch): + # ag-ui-langgraph streams compiled subgraphs and would emit the child's + # own text / `lookup` call as UNATTRIBUTED TEXT_MESSAGE_* / TOOL_CALL_* + # events in the parent transcript. The bridge honors a declared opt-out: + # runs whose LangChain metadata carries `emit-messages` / `emit-tool-calls` + # = False are skipped for those emissions (callbacks and STEP_* still + # fire). The tool must pass it on the subgraph invocation so it inherits + # into the child run — no wrapper-side duplicate filtering. + captured: dict[str, Any] = {} + + class _CapturingSubgraph: + async def ainvoke(self, state: Any, config: Any = None, **kwargs: Any) -> dict: + captured["state"] = state + captured["config"] = config + return {"messages": [AIMessage(content="- Signals are reactive.")]} + + monkeypatch.setattr(graph_mod, "_build_research_subgraph", lambda *a, **k: _CapturingSubgraph()) + + result = await research.ainvoke( + {"type": "tool_call", "id": TID, "name": "research", "args": {"topic": "Angular signals"}} + ) + + assert result.content == "- Signals are reactive." + assert captured["state"]["topic"] == "Angular signals" + metadata = captured["config"]["metadata"] + assert metadata["emit-messages"] is False + assert metadata["emit-tool-calls"] is False + # The streaming callback still rides along — the opt-out silences only the + # bridge's raw emission, not the per-token subagent_activity source. + handler_types = [type(cb).__name__ for cb in captured["config"]["callbacks"]] + assert "SubagentStreamHandler" in handler_types diff --git a/examples/ag-ui/python/tests/test_subagent_emitting_agent.py b/examples/ag-ui/python/tests/test_subagent_emitting_agent.py new file mode 100644 index 000000000..71486aaab --- /dev/null +++ b/examples/ag-ui/python/tests/test_subagent_emitting_agent.py @@ -0,0 +1,461 @@ +"""Tests for SubagentEmittingAgent — the run-wrapping emitter that expands the +graph's `subagent_activity` CUSTOM events (started / message_start / message / +tool_call / tool_result / finished / error) into the protocol's standard +SUBAGENT_* + attributed TEXT_MESSAGE_* / TOOL_CALL_* events. Drives the +wrapper with a scripted inner `LangGraphAgent.run` generator and asserts the +exact output sequence field-for-field.""" +import json +import logging +from typing import Any + +import pytest +from ag_ui.core import ( + CustomEvent, + EventType, + RunAgentInput, + RunFinishedEvent, + RunStartedEvent, + StepStartedEvent, + TextMessageContentEvent, + TextMessageEndEvent, + TextMessageStartEvent, + ToolCallArgsEvent, + ToolCallEndEvent, + ToolCallResultEvent, + ToolCallStartEvent, +) +from ag_ui_langgraph import LangGraphAgent +from langgraph.graph import END, MessagesState, StateGraph + +from src.streaming.subagent_emitting_agent import SubagentEmittingAgent + +TID = "call_1" +TID2 = "call_2" +RUN_ID = f"{TID}-sub" +M1 = f"{TID}-sub-m1" +M2 = f"{TID}-sub-m2" +LOOKUP = "call_lookup_1" +LOOKUP_ARGS = {"query": "angular signals"} +LOOKUP_RESULT = "Angular signals are a reactivity primitive." + + +def _graph(): + g = StateGraph(MessagesState) + g.add_node("noop", lambda state: {}) + g.set_entry_point("noop") + g.add_edge("noop", END) + return g.compile() + + +def _input() -> RunAgentInput: + return RunAgentInput( + thread_id="t", run_id="r", messages=[], tools=[], context=[], state={}, forwarded_props={} + ) + + +def _activity(payload: dict[str, Any]) -> CustomEvent: + return CustomEvent(type=EventType.CUSTOM, name="subagent_activity", value=payload) + + +def _run_started(): + return RunStartedEvent(type=EventType.RUN_STARTED, thread_id="t", run_id="r") + + +def _run_finished(): + return RunFinishedEvent(type=EventType.RUN_FINISHED, thread_id="t", run_id="r") + + +def _tool_call(tid: str, name: str = "research", args: str = '{"topic":"signals"}'): + return [ + ToolCallStartEvent(type=EventType.TOOL_CALL_START, tool_call_id=tid, tool_call_name=name), + ToolCallArgsEvent(type=EventType.TOOL_CALL_ARGS, tool_call_id=tid, delta=args), + ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=tid), + ] + + +def _tool_result(tid: str, content: str): + return ToolCallResultEvent( + type=EventType.TOOL_CALL_RESULT, message_id=f"{tid}-result", tool_call_id=tid, content=content + ) + + +def _text(message_id: str, delta: str): + return [ + TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, message_id=message_id, role="assistant"), + TextMessageContentEvent(type=EventType.TEXT_MESSAGE_CONTENT, message_id=message_id, delta=delta), + TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END, message_id=message_id), + ] + + +def _delegation(tid: str = TID, name: str = "research") -> list: + """The research subgraph's phase sequence for one reason → tool → answer + loop, as the graph dispatches it (see test_subagent_emission.py).""" + m1, m2 = f"{tid}-sub-m1", f"{tid}-sub-m2" + return [ + _activity({"subagent_id": tid, "phase": "started", "name": name}), + _activity({"subagent_id": tid, "phase": "message_start", "message_id": m1}), + _activity({"subagent_id": tid, "phase": "tool_call", "message_id": m1, + "tool_call_id": LOOKUP, "name": "lookup", "args": LOOKUP_ARGS}), + _activity({"subagent_id": tid, "phase": "tool_result", + "tool_call_id": LOOKUP, "content": LOOKUP_RESULT}), + _activity({"subagent_id": tid, "phase": "message_start", "message_id": m2}), + _activity({"subagent_id": tid, "phase": "message", "message_id": m2, "delta": "- Signals"}), + _activity({"subagent_id": tid, "phase": "message", "message_id": m2, "delta": " are reactive."}), + _activity({"subagent_id": tid, "phase": "finished", "status": "complete"}), + ] + + +async def _collect(monkeypatch, script: list) -> list: + async def fake_run(self, input): + for ev in script: + yield ev + + monkeypatch.setattr(LangGraphAgent, "run", fake_run) + agent = SubagentEmittingAgent(name="chat", graph=_graph()) + return [ev async for ev in agent.run(_input())] + + +async def test_expands_the_reason_tool_answer_loop_field_for_field(monkeypatch): + script = [_run_started(), *_tool_call(TID), *_delegation(), _tool_result(TID, "- Signals are reactive."), _run_finished()] + out = await _collect(monkeypatch, script) + + assert [ev.type for ev in out] == [ + EventType.RUN_STARTED, + EventType.TOOL_CALL_START, + EventType.TOOL_CALL_ARGS, + EventType.TOOL_CALL_END, + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, # m1 — the tool-calling turn + EventType.TEXT_MESSAGE_END, # closed before the child's tool call + EventType.TOOL_CALL_START, # lookup (attributed) + EventType.TOOL_CALL_ARGS, + EventType.TOOL_CALL_END, + EventType.TOOL_CALL_RESULT, # lookup result (attributed) + EventType.TEXT_MESSAGE_START, # m2 — the answer turn + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + EventType.TOOL_CALL_RESULT, # the parent's research result (bridge-native) + EventType.RUN_FINISHED, + ] + # The CUSTOM subagent_activity events are consumed, never forwarded. + assert not any(ev.type == EventType.CUSTOM for ev in out) + + started = out[4] + assert started.subagent_run_id == RUN_ID + assert started.name == "research" + assert started.parent_tool_call_id == TID + + m1_start, m1_end = out[5], out[6] + assert (m1_start.message_id, m1_start.role, m1_start.subagent_run_id) == (M1, "assistant", RUN_ID) + assert (m1_end.message_id, m1_end.subagent_run_id) == (M1, RUN_ID) + + tc_start, tc_args, tc_end, tc_result = out[7:11] + assert tc_start.tool_call_id == LOOKUP + assert tc_start.tool_call_name == "lookup" + assert tc_start.parent_message_id == M1 + assert tc_start.subagent_run_id == RUN_ID + assert tc_args.tool_call_id == LOOKUP + assert json.loads(tc_args.delta) == LOOKUP_ARGS + assert tc_args.subagent_run_id == RUN_ID + assert tc_end.tool_call_id == LOOKUP + assert tc_end.subagent_run_id == RUN_ID + assert tc_result.tool_call_id == LOOKUP + assert tc_result.message_id == f"{LOOKUP}-result" + assert tc_result.content == LOOKUP_RESULT + assert tc_result.role == "tool" + assert tc_result.subagent_run_id == RUN_ID + + m2_start = out[11] + assert (m2_start.message_id, m2_start.role, m2_start.subagent_run_id) == (M2, "assistant", RUN_ID) + deltas = out[12:14] + assert [ev.delta for ev in deltas] == ["- Signals", " are reactive."] + for ev in deltas: + assert ev.message_id == M2 + assert ev.subagent_run_id == RUN_ID + m2_end = out[14] + assert (m2_end.message_id, m2_end.subagent_run_id) == (M2, RUN_ID) + + finished = out[15] + assert finished.subagent_run_id == RUN_ID + assert finished.outcome.type == "success" + + # Bridge-native events pass through untouched (same objects, unattributed). + assert out[1] is script[1] + assert out[16] is script[12] + assert out[16].subagent_run_id is None + + +async def test_bridge_native_events_pass_through_untouched_even_inside_a_delegation(monkeypatch): + # The wrapper is a pure 1:N expander: it never filters bridge-native + # events. The child subgraph's own LLM text / tool call are kept off the + # wire at the SOURCE — the research tool invokes the subgraph with the + # bridge's `emit-messages` / `emit-tool-calls` = False run metadata (see + # test_subagent_emission.py) — so an unattributed event that does arrive + # between started and finished (the child's STEP_*, or any future + # bridge-native event) is forwarded as-is, same object, in order. + step = StepStartedEvent(type=EventType.STEP_STARTED, step_name="tools") + parent_text = _text("lc_run--parent", "Here is what the subagent found.") + script = [ + _run_started(), + *_tool_call(TID), + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "message_start", "message_id": M1}), + step, + _activity({"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "- Signals"}), + _activity({"subagent_id": TID, "phase": "finished"}), + _tool_result(TID, "- Signals"), + *parent_text, + _run_finished(), + ] + out = await _collect(monkeypatch, script) + assert [(ev.type, getattr(ev, "subagent_run_id", None)) for ev in out] == [ + (EventType.RUN_STARTED, None), + (EventType.TOOL_CALL_START, None), + (EventType.TOOL_CALL_ARGS, None), + (EventType.TOOL_CALL_END, None), + (EventType.SUBAGENT_STARTED, RUN_ID), + (EventType.TEXT_MESSAGE_START, RUN_ID), + (EventType.STEP_STARTED, None), + (EventType.TEXT_MESSAGE_CONTENT, RUN_ID), + (EventType.TEXT_MESSAGE_END, RUN_ID), + (EventType.SUBAGENT_FINISHED, RUN_ID), + (EventType.TOOL_CALL_RESULT, None), + (EventType.TEXT_MESSAGE_START, None), + (EventType.TEXT_MESSAGE_CONTENT, None), + (EventType.TEXT_MESSAGE_END, None), + (EventType.RUN_FINISHED, None), + ] + assert out[6] is step + assert out[11:14] == parent_text + + +async def test_without_any_delegation_the_stream_is_the_identity(monkeypatch): + script = [_run_started(), *_tool_call("call_search", name="search_documents"), + _tool_result("call_search", "[]"), *_text("lc_run--parent", "hi"), _run_finished()] + out = await _collect(monkeypatch, script) + assert out == script + + +async def test_serialized_custom_value_is_decoded(monkeypatch): + # The bridge may JSON-serialize custom values; the expansion must cope. + script = [ + CustomEvent(type=EventType.CUSTOM, name="subagent_activity", + value='{"subagent_id": "call_1", "phase": "started", "name": "research"}'), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [EventType.SUBAGENT_STARTED, EventType.SUBAGENT_FINISHED] + assert out[0].name == "research" + + +async def test_no_deltas_still_brackets_with_started_and_finished(monkeypatch): + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [EventType.SUBAGENT_STARTED, EventType.SUBAGENT_FINISHED] + + +async def test_unrelated_custom_event_passes_through_untouched(monkeypatch): + other = CustomEvent(type=EventType.CUSTOM, name="PredictState", value={"x": 1}) + out = await _collect(monkeypatch, [_run_started(), other, _run_finished()]) + assert [ev.type for ev in out] == [EventType.RUN_STARTED, EventType.CUSTOM, EventType.RUN_FINISHED] + assert out[1] is other + + +async def test_error_closes_open_message_then_reports(monkeypatch): + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "message_start", "message_id": M1}), + _activity({"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "Par"}), + _activity({"subagent_id": TID, "phase": "error", "message": "RuntimeError: child exploded"}), + ] + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, # open message is closed before the error + EventType.SUBAGENT_ERROR, + ] + assert out[3].message_id == M1 + assert out[3].subagent_run_id == RUN_ID + err = out[4] + assert err.subagent_run_id == RUN_ID + assert err.message == "RuntimeError: child exploded" + + +async def test_tool_call_without_an_open_message_has_no_parent(monkeypatch): + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "tool_call", "tool_call_id": LOOKUP, "name": "lookup", "args": {}}), + _activity({"subagent_id": TID, "phase": "tool_result", "tool_call_id": LOOKUP, "content": {"ok": True}}), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TOOL_CALL_START, + EventType.TOOL_CALL_ARGS, + EventType.TOOL_CALL_END, + EventType.TOOL_CALL_RESULT, + EventType.SUBAGENT_FINISHED, + ] + assert out[1].parent_message_id is None + assert out[2].delta == "{}" + # Non-string tool content is JSON-serialized so the encoder never sees a dict. + assert out[4].content == '{"ok": true}' + + +async def test_message_start_without_message_id_derives_it(monkeypatch): + # Defensive: a message_start missing message_id gets -sub-m. + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "message_start"}), + _activity({"subagent_id": TID, "phase": "message", "delta": "x"}), + _activity({"subagent_id": TID, "phase": "message_start"}), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + out = await _collect(monkeypatch, script) + starts = [ev for ev in out if ev.type == EventType.TEXT_MESSAGE_START] + assert [ev.message_id for ev in starts] == [M1, M2] + content = [ev for ev in out if ev.type == EventType.TEXT_MESSAGE_CONTENT] + assert content[0].message_id == M1 + + +async def test_message_before_message_start_opens_the_message_lazily(monkeypatch): + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "x"}), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + assert out[1].message_id == M1 + + +async def test_two_sequential_delegations_get_distinct_run_ids(monkeypatch): + script = [ + _run_started(), + *_tool_call(TID), + *_delegation(TID, "research"), + _tool_result(TID, "intel"), + *_tool_call(TID2), + *_delegation(TID2, "research"), + _tool_result(TID2, "more intel"), + _run_finished(), + ] + out = await _collect(monkeypatch, script) + + started = [ev for ev in out if ev.type == EventType.SUBAGENT_STARTED] + assert [ev.subagent_run_id for ev in started] == [f"{TID}-sub", f"{TID2}-sub"] + assert [ev.parent_tool_call_id for ev in started] == [TID, TID2] + + expected = [ + EventType.SUBAGENT_STARTED, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_END, + EventType.TOOL_CALL_START, + EventType.TOOL_CALL_ARGS, + EventType.TOOL_CALL_END, + EventType.TOOL_CALL_RESULT, + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + a = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{TID}-sub"] + b = [ev for ev in out if getattr(ev, "subagent_run_id", None) == f"{TID2}-sub"] + assert [ev.type for ev in a] == expected + assert [ev.type for ev in b] == expected + assert {ev.message_id for ev in a if ev.type == EventType.TEXT_MESSAGE_START} == {f"{TID}-sub-m1", f"{TID}-sub-m2"} + assert {ev.message_id for ev in b if ev.type == EventType.TEXT_MESSAGE_START} == {f"{TID2}-sub-m1", f"{TID2}-sub-m2"} + # Every child block sits between its own tool call's END and RESULT. + types = [ev.type for ev in out] + parent_results = [i for i, ev in enumerate(out) if ev.type == EventType.TOOL_CALL_RESULT and ev.subagent_run_id is None] + assert types.index(EventType.SUBAGENT_STARTED) > types.index(EventType.TOOL_CALL_END) + assert types.index(EventType.SUBAGENT_FINISHED) < parent_results[0] + second_started = [i for i, ev in enumerate(out) if ev.type == EventType.SUBAGENT_STARTED][1] + assert parent_results[0] < second_started < parent_results[1] + + +async def test_unknown_phase_is_dropped_with_a_warning(monkeypatch, caplog): + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "reasoning", "delta": "hmm"}), + _activity({"subagent_id": TID, "phase": "finished"}), + ] + with caplog.at_level(logging.WARNING): + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [EventType.SUBAGENT_STARTED, EventType.SUBAGENT_FINISHED] + assert any("reasoning" in rec.getMessage() for rec in caplog.records) + + +async def test_malformed_payload_is_dropped(monkeypatch, caplog): + script = [ + _run_started(), + CustomEvent(type=EventType.CUSTOM, name="subagent_activity", value="not json"), + _activity({"phase": "started", "name": "research"}), # no subagent_id + _activity({"subagent_id": TID}), # no phase + _activity({"subagent_id": TID, "phase": "tool_call", "name": "lookup"}), # no tool_call_id + _activity({"subagent_id": TID, "phase": "tool_result", "content": "x"}), # no tool_call_id + _run_finished(), + ] + with caplog.at_level(logging.WARNING): + out = await _collect(monkeypatch, script) + assert [ev.type for ev in out] == [EventType.RUN_STARTED, EventType.RUN_FINISHED] + + +async def test_delegation_state_is_per_run(monkeypatch): + # A second run on the same agent must not see the first run's open message. + script = [ + _activity({"subagent_id": TID, "phase": "started", "name": "research"}), + _activity({"subagent_id": TID, "phase": "message_start", "message_id": M1}), + _activity({"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "x"}), + # run ends with the message still open (client disconnect, say) + ] + + async def fake_run(self, input): + for ev in script: + yield ev + + monkeypatch.setattr(LangGraphAgent, "run", fake_run) + agent = SubagentEmittingAgent(name="chat", graph=_graph()) + first = [ev async for ev in agent.run(_input())] + assert first[-1].type == EventType.TEXT_MESSAGE_CONTENT + + script[:] = [*_text("lc_run--parent", "hello"), _activity({"subagent_id": TID, "phase": "finished"})] + second = [ev async for ev in agent.run(_input())] + # No stale TEXT_MESSAGE_END from run 1 leaks into run 2, and the unknown + # delegation's finished is still expanded (bracketing the card). + assert [ev.type for ev in second] == [ + EventType.TEXT_MESSAGE_START, + EventType.TEXT_MESSAGE_CONTENT, + EventType.TEXT_MESSAGE_END, + EventType.SUBAGENT_FINISHED, + ] + + +def test_clone_preserves_the_subclass(): + # The FastAPI endpoint runs agent.clone() per request; the emitter must + # survive cloning or SUBAGENT_* events would silently vanish from the wire. + agent = SubagentEmittingAgent(name="chat", graph=_graph()) + assert isinstance(agent.clone(), SubagentEmittingAgent) + + +def test_server_mounts_the_emitting_agent(monkeypatch): + # ChatOpenAI validates credentials at construction (graph import time). + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-not-a-real-key") + from src import server + + assert isinstance(server.agent, SubagentEmittingAgent) diff --git a/examples/ag-ui/python/tests/test_subagent_stream_handler.py b/examples/ag-ui/python/tests/test_subagent_stream_handler.py index 8d5100ef0..55f6452ac 100644 --- a/examples/ag-ui/python/tests/test_subagent_stream_handler.py +++ b/examples/ag-ui/python/tests/test_subagent_stream_handler.py @@ -1,61 +1,104 @@ -"""Tests for SubagentStreamHandler — accumulates child LLM text tokens and -emits `subagent_activity` `message` events carrying the full `text_so_far`.""" +"""Tests for SubagentStreamHandler — forwards each child LLM token as a +`subagent_activity` payload: one `message_start` (carrying the message id) +before the first token of a message, then a `message` per token whose `delta` +is the raw token (no accumulation — the emitter turns these into attributed +TEXT_MESSAGE_START / TEXT_MESSAGE_CONTENT events). + +The research subgraph's `agent` node opens each assistant turn itself +(`SubagentRunState.open_message()` + its own `message_start` dispatch) before +invoking the model, so the handler only opens a message when none is open — +the subgraph and the handler never double-announce a turn.""" from unittest.mock import AsyncMock, patch from uuid import uuid4 import pytest from src.streaming.subagent_stream_handler import ( - SubagentStreamHandler, SubagentRunState, + SubagentStreamHandler, ) +TID = "call_7sxPY1sC236nPyHRTWAZMJB9" +M1 = f"{TID}-sub-m1" +M2 = f"{TID}-sub-m2" + + +class TestSubagentRunState: + def test_open_message_derives_sequential_ids(self): + state = SubagentRunState(TID) + assert state.message_id is None + assert state.open_message() == M1 + assert state.message_id == M1 + assert state.open_message() == M2 + assert state.message_id == M2 + assert state.message_count == 2 + class TestSubagentStreamHandler: @pytest.mark.asyncio - async def test_emits_accumulated_text_so_far(self): - handler = SubagentStreamHandler(subagent_id="tc-1") + async def test_emits_message_start_then_per_token_deltas(self): + handler = SubagentStreamHandler(subagent_id=TID) with patch("src.streaming.subagent_stream_handler.adispatch_custom_event", new_callable=AsyncMock) as dispatch: await handler.on_llm_new_token("Paris ", run_id=uuid4()) await handler.on_llm_new_token("is", run_id=uuid4()) - assert dispatch.call_args_list[0].args == ( - "subagent_activity", - {"subagent_id": "tc-1", "phase": "message", "message_index": 0, "text": "Paris "}) - assert dispatch.call_args_list[1].args == ( - "subagent_activity", - {"subagent_id": "tc-1", "phase": "message", "message_index": 0, "text": "Paris is"}) + assert [c.args for c in dispatch.call_args_list] == [ + ("subagent_activity", + {"subagent_id": TID, "phase": "message_start", "message_id": M1}), + ("subagent_activity", + {"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "Paris "}), + ("subagent_activity", + {"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "is"}), + ] @pytest.mark.asyncio - async def test_buffers_isolated_across_instances(self): - h1, h2 = SubagentStreamHandler("a"), SubagentStreamHandler("b") + async def test_empty_token_emits_nothing(self): + handler = SubagentStreamHandler(subagent_id=TID) with patch("src.streaming.subagent_stream_handler.adispatch_custom_event", new_callable=AsyncMock) as dispatch: - await h1.on_llm_new_token("x", run_id=uuid4()) - await h2.on_llm_new_token("y", run_id=uuid4()) - assert dispatch.call_args_list[0].args[1]["text"] == "x" - assert dispatch.call_args_list[1].args[1]["text"] == "y" + await handler.on_llm_new_token("", run_id=uuid4()) + assert dispatch.call_args_list == [] @pytest.mark.asyncio - async def test_tags_message_index_from_run_state(self): - run_state = SubagentRunState() - handler = SubagentStreamHandler(subagent_id="tc-1", run_state=run_state) + async def test_tags_the_message_the_subgraph_opened(self): + # The subgraph opens the turn (and dispatches message_start itself); + # the handler must reuse that id and NOT re-announce the message. + run_state = SubagentRunState(TID) + handler = SubagentStreamHandler(subagent_id=TID, run_state=run_state) with patch("src.streaming.subagent_stream_handler.adispatch_custom_event", new_callable=AsyncMock) as dispatch: + run_state.open_message() await handler.on_llm_new_token("first", run_id=uuid4()) # Subgraph advances to the next assistant turn. - run_state.message_index = 1 - await handler.on_llm_new_token("second", run_id=uuid4()) - first, second = dispatch.call_args_list - assert first.args[1]["message_index"] == 0 - assert first.args[1]["text"] == "first" - # Buffer resets per turn so text_so_far is scoped to the new turn. - assert second.args[1]["message_index"] == 1 - assert second.args[1]["text"] == "second" + run_state.open_message() + await handler.on_llm_new_token("sec", run_id=uuid4()) + await handler.on_llm_new_token("ond", run_id=uuid4()) + assert [c.args[1] for c in dispatch.call_args_list] == [ + {"subagent_id": TID, "phase": "message", "message_id": M1, "delta": "first"}, + {"subagent_id": TID, "phase": "message", "message_id": M2, "delta": "sec"}, + {"subagent_id": TID, "phase": "message", "message_id": M2, "delta": "ond"}, + ] + + @pytest.mark.asyncio + async def test_message_ids_isolated_across_instances(self): + h1, h2 = SubagentStreamHandler("a"), SubagentStreamHandler("b") + with patch("src.streaming.subagent_stream_handler.adispatch_custom_event", + new_callable=AsyncMock) as dispatch: + await h1.on_llm_new_token("x", run_id=uuid4()) + await h2.on_llm_new_token("y", run_id=uuid4()) + payloads = [c.args[1] for c in dispatch.call_args_list] + assert [p["phase"] for p in payloads] == [ + "message_start", "message", "message_start", "message"] + assert payloads[0]["message_id"] == "a-sub-m1" + assert payloads[1] == {"subagent_id": "a", "phase": "message", + "message_id": "a-sub-m1", "delta": "x"} + assert payloads[2]["message_id"] == "b-sub-m1" + assert payloads[3] == {"subagent_id": "b", "phase": "message", + "message_id": "b-sub-m1", "delta": "y"} @pytest.mark.asyncio async def test_dispatch_failure_is_silent(self): - handler = SubagentStreamHandler(subagent_id="tc-1") + handler = SubagentStreamHandler(subagent_id=TID) with patch("src.streaming.subagent_stream_handler.adispatch_custom_event", new_callable=AsyncMock, side_effect=RuntimeError): await handler.on_llm_new_token("hi", run_id=uuid4()) # must not raise diff --git a/examples/ag-ui/python/uv.lock b/examples/ag-ui/python/uv.lock index 69ff4bb94..3c80a1987 100644 --- a/examples/ag-ui/python/uv.lock +++ b/examples/ag-ui/python/uv.lock @@ -30,14 +30,14 @@ wheels = [ [[package]] name = "ag-ui-protocol" -version = "0.1.19" +version = "0.1.22" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a7/10/4ad299267a7d04b89935aa99eef62979758fcf95aee9f8bb5d70c35b1be1/ag_ui_protocol-0.1.19.tar.gz", hash = "sha256:43c27f60d41712dcad0e9e0a203cbdf1c8e248b22417374c5c68321c448af4ea", size = 10720, upload-time = "2026-06-02T17:26:15.627Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e8/f7/9bf788e7d3608725d022a248a58427040f4f930ab87ddb54bd29ee4d9a51/ag_ui_protocol-0.1.22.tar.gz", hash = "sha256:d21f265284a50d9fc87ad7bcbd58f737b4b16eef7b5375f13a6e925117b52046", size = 18110, upload-time = "2026-08-31T18:20:04.334Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/4c/0a/bcad8116eb058e4b4a305e3fc37ebd7efc879deeb86b854f1c5b8b6e97dd/ag_ui_protocol-0.1.19-py3-none-any.whl", hash = "sha256:898843b1410d378824da0c6a776486288b9c5828689d0bf563118868e37f390f", size = 13490, upload-time = "2026-06-02T17:26:16.313Z" }, + { url = "https://files.pythonhosted.org/packages/be/6c/af3d577e68c9474c99600e65b2c6283772aae187edca6f0f2f5cbd9f565a/ag_ui_protocol-0.1.22-py3-none-any.whl", hash = "sha256:fca13ee7820f8f53e869c19e09ddd75826c1799b27c2adb6f2e567295433c704", size = 22068, upload-time = "2026-08-31T18:20:03.43Z" }, ] [[package]] @@ -189,6 +189,7 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "ag-ui-langgraph" }, + { name = "ag-ui-protocol" }, { name = "fastapi" }, { name = "langchain-openai" }, { name = "langgraph" }, @@ -206,6 +207,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "ag-ui-langgraph", specifier = ">=0.0.37" }, + { name = "ag-ui-protocol", specifier = ">=0.1.22" }, { name = "fastapi", specifier = ">=0.115" }, { name = "langchain-openai", specifier = ">=0.3" }, { name = "langgraph", specifier = ">=0.3" },