diff --git a/apps/website/content/blog/2026-08-31-we-measured-the-runtime-swap.mdx b/apps/website/content/blog/2026-08-31-we-measured-the-runtime-swap.mdx index 309efb2ee..aa7f53631 100644 --- a/apps/website/content/blog/2026-08-31-we-measured-the-runtime-swap.mdx +++ b/apps/website/content/blog/2026-08-31-we-measured-the-runtime-swap.mdx @@ -45,7 +45,7 @@ A committed transcript replayed through the real client proves what the runtime | LangGraph (baseline) | Yes | Yes | Yes | Yes | Yes | | AWS Strands | Yes | Yes | Partial | Yes | Yes | | Microsoft Agent Framework | Yes | Yes | Yes | Yes | Yes | -| Mastra | Yes | Yes | Yes | Yes | Partial | +| Mastra | Yes | Yes | Yes | Yes | Yes | Messages, tool calls, and shared state crossed three non-LangGraph runtimes with zero changes to the adapter. Not one line. @@ -123,7 +123,7 @@ They were only findable by pointing the adapter at software written by people wh ## What stayed partial -Two cells did not go fully green. +One cell did not go fully green. **Shared state on Strands is partial.** Its bridge sends whole-document snapshots and never sends a patch, and a tool only contributes state if it opts in through a per-tool hook. @@ -138,11 +138,12 @@ The cause is the upstream bridge, not the protocol and not us. The protocol standardized the events — `SUBAGENT_STARTED`, `SUBAGENT_FINISHED`, `SUBAGENT_ERROR`, plus a `subagentRunId` attribution field on ordinary content events — and the adapter consumes them directly. What no runtime does is emit them natively. -Each one reports delegation in its own dialect: Strands surfaces the specialist's tool use and forwards its token stream, Microsoft Agent Framework streams the specialist's updates in-process into the tool body where the in-tree emitter merges them across the bridge boundary, and Mastra reports the delegation through its tool frames and returns the child's final text. -So each demo backend carries a small emitter — roughly 125 to 310 lines each — that translates its runtime's dialect into the standard events at the bridge boundary. +Each one reports delegation in its own dialect: Strands surfaces the specialist's tool use and forwards its token stream, Microsoft Agent Framework streams the specialist's updates in-process into the tool body where the in-tree emitter merges them across the bridge boundary, and Mastra forwards the child's chunks on the parent stream, where its bridge drops them and withholds the delegation tool call until the child resolves. +So each demo backend carries a small emitter — roughly 190 to 360 lines each — that translates its runtime's dialect into the standard events at the bridge boundary. -Two of the three cards stream live. -Mastra's fills in at completion, because its bridge does not forward child tokens to the wire; that is the one Partial cell, and it belongs to the runtime's bridge rather than to the protocol or the adapter. +All three cards stream live. +Mastra's took one more seam than the others: the emitter observes the runtime's public stream through a tee ahead of the bridge, emits the delegation tool call eagerly, and forwards the child's deltas itself, because the bridge would otherwise have painted the card only at completion. +That extra seam belongs to the runtime's bridge rather than to the protocol or the adapter, and the bridge itself is left unmodified. If you are building on server-declared subagents today, the contract to target is the protocol's own events; the emitter is the per-runtime cost, and it is small. ## What the deploy check found @@ -171,7 +172,7 @@ That is now measured against three implementations in two languages. Interrupts are portable as of these fixes, and they were not before, in a way no amount of internal testing would have revealed. -Subagents are portable through a per-runtime emitter that speaks the protocol's own events, streaming on two of the three runtimes and lifecycle-plus-final-text on the third. +Subagents are portable through a per-runtime emitter that speaks the protocol's own events, streaming on all three runtimes. The matrix now lives in the [adapter guide](/docs/choosing-an-adapter), with a cause column on every gap. Split three ways: the protocol cannot express it, the upstream bridge does not emit it, or our adapter failed to consume it. diff --git a/apps/website/content/docs/choosing-an-adapter/index.mdx b/apps/website/content/docs/choosing-an-adapter/index.mdx index b711d1aa4..8999c7df9 100644 --- a/apps/website/content/docs/choosing-an-adapter/index.mdx +++ b/apps/website/content/docs/choosing-an-adapter/index.mdx @@ -94,7 +94,7 @@ We tested it against three runtimes that have nothing to do with LangGraph, in t | **LangGraph** (via the AG-UI bridge) | Yes | Yes | Yes | Yes | Yes | — | | **AWS Strands** (Python) | Yes | Yes | Partial | Yes | Yes | State: upstream. Subagents: a small in-tree emitter translates native delegation signals to the protocol's `SUBAGENT_*` events | | **Microsoft Agent Framework** (Python) | Yes | Yes | Yes | Yes | Yes | Subagents: a small in-tree emitter translates native delegation signals to the protocol's `SUBAGENT_*` events | -| **Mastra** (TypeScript) | Yes | Yes | Yes | Yes | Partial | Subagents: lifecycle and final text via the emitter; the runtime's bridge does not forward child token streams | +| **Mastra** (TypeScript) | Yes | Yes | Yes | Yes | Yes | Subagents: a small in-tree emitter observes the runtime's public stream through a tee and emits the protocol's `SUBAGENT_*` events plus the child's token deltas, because the runtime's bridge drops child output | Every gap in that table falls into one of three causes, and the distinction is the point of the column: @@ -133,8 +133,9 @@ The protocol standardized the events — `SUBAGENT_STARTED`, `SUBAGENT_FINISHED` What differs per runtime is how the emitter learns about the delegation. Strands surfaces the specialist's tool use and streamed tokens through a per-tool stream handler, so the card streams live. On Microsoft Agent Framework the specialist's updates stream in-process into the tool body, and the in-tree emitter merges them across the bridge boundary, so the card streams live there too. -Mastra reports delegation through its tool frames and returns the child's final text; its bridge does not forward child tokens, so the card fills in at completion — the one Partial cell, a property of the runtime's bridge rather than of the protocol or the adapter. -Each demo backend ships its emitter in tree, roughly 125 to 310 lines per runtime. +Mastra streams the child's chunks in-process on the parent stream, but its bridge drops them and withholds the delegation tool call until the child resolves; the in-tree emitter therefore observes the runtime's public stream through a tee ahead of the bridge, emits the delegation tool call eagerly, and forwards the child's deltas under the subagent identity, so the card mounts while the child runs and streams live there too. +All three cards stream live; the Mastra card streams through the tee rather than through the bridge, a property of the runtime's bridge rather than of the protocol or the adapter. +Each demo backend ships its emitter in tree, roughly 190 to 360 lines per runtime. ### How this was measured diff --git a/apps/website/content/docs/runtimes/getting-started/introduction.mdx b/apps/website/content/docs/runtimes/getting-started/introduction.mdx index 80a6a2b46..6427852cd 100644 --- a/apps/website/content/docs/runtimes/getting-started/introduction.mdx +++ b/apps/website/content/docs/runtimes/getting-started/introduction.mdx @@ -41,7 +41,7 @@ TypeScript. Messages, tool calls, state, and interrupts all work, against a hand | **LangGraph** (via the AG-UI bridge) | Yes | Yes | Yes | Yes | Yes | | **AWS Strands** (Python) | Yes | Yes | Partial | Yes | Yes | | **Microsoft Agent Framework** (Python) | Yes | Yes | Yes | Yes | Yes | -| **Mastra** (TypeScript) | Yes | Yes | Yes | Yes | Partial | +| **Mastra** (TypeScript) | Yes | Yes | Yes | Yes | Yes | Every gap in that table is caused by an upstream integration, not by the AG-UI protocol and not by a defect in `@threadplane/ag-ui`. The full cause analysis, including the two adapter defects that were found and fixed, lives in [Choosing an adapter](/docs/choosing-an-adapter). @@ -88,7 +88,7 @@ Three differences turned up repeatedly, and each runtime page returns to them. **Resume payloads are not portable.** The adapter derives the wire shape from how the interrupt arrived, so application code passes one neutral `submit({ resume })` regardless of runtime. -**Subagents now work on every runtime measured here** — streaming on AWS Strands and Microsoft Agent Framework, and as lifecycle-plus-final-text on Mastra. Each backend ships a small emitter that translates its native delegation signals into the protocol's `SUBAGENT_*` events, which `@threadplane/ag-ui` consumes directly. The per-runtime pages show the emitter and the wire capture behind each cell. +**Subagents now stream on every runtime measured here** — AWS Strands, Microsoft Agent Framework, and Mastra. Each backend ships a small emitter that translates its native delegation signals into the protocol's `SUBAGENT_*` events, which `@threadplane/ag-ui` consumes directly. The per-runtime pages show the emitter and the wire capture behind each cell. ## Further reading diff --git a/apps/website/content/docs/runtimes/mastra/how-it-connects.mdx b/apps/website/content/docs/runtimes/mastra/how-it-connects.mdx index e817a9a94..a33b86e77 100644 --- a/apps/website/content/docs/runtimes/mastra/how-it-connects.mdx +++ b/apps/website/content/docs/runtimes/mastra/how-it-connects.mdx @@ -54,10 +54,11 @@ Shared state is a Mastra working-memory object under a Zod schema. The bridge em That places Mastra alongside Microsoft Agent Framework and apart from AWS Strands, whose bridge emits snapshots only. -## Subagents surface as lifecycle plus final text +## Subagents stream through a tee ahead of the bridge -An in-tree injector in the hosting service watches the wire for delegation tool frames — Mastra names them `agent-` — and adds `SUBAGENT_STARTED` on the tool-call start, one attributed text message carrying the child's final text, and `SUBAGENT_FINISHED` on the result. -The child's incremental tokens never reach the wire because `@ag-ui/mastra` drops its `tool-output` chunks and buffers the delegation burst until it resolves, which is why this cell is Partial. +Mastra forwards every chunk of a delegated child on the parent stream as a public `tool-output` chunk, but `@ag-ui/mastra` drops those chunks and withholds the delegation tool call — Mastra names it `agent-` — until the child resolves. +The hosting service wraps the agent so that each chunk is observed before the bridge processes it, and a per-run injector emits the delegation tool call eagerly, `SUBAGENT_STARTED`, one `TEXT_MESSAGE_CONTENT` per child delta attributed to the subagent, and `SUBAGENT_FINISHED` on the result, while dropping the bridge's later copy of the same tool call. +The card therefore mounts as soon as the delegation begins and its text grows while the child runs, which is what makes this cell Supported. ## Next steps diff --git a/apps/website/content/docs/runtimes/mastra/overview.mdx b/apps/website/content/docs/runtimes/mastra/overview.mdx index 9c407dd3c..2f1a50613 100644 --- a/apps/website/content/docs/runtimes/mastra/overview.mdx +++ b/apps/website/content/docs/runtimes/mastra/overview.mdx @@ -21,7 +21,7 @@ The hosted example runs at [examples.threadplane.ai/runtimes/mastra](https://exa | Tool calls | Supported | `check_conditions` executes server-side with no pause. | | Shared state | Supported | A working-memory packing list, over snapshots and real JSON-Patch deltas. | | Interrupts | Supported | `reserve_campsite` suspends the run and resumes from a persisted snapshot. | -| Subagents | Partial | The bridge emits delegation tool frames; an in-tree injector adds `SUBAGENT_*` lifecycle and the child's final text — the runtime's bridge does not forward child tokens. | +| Subagents | Supported | An in-tree stream tee observes the parent stream ahead of the bridge and emits the delegation tool call eagerly, `SUBAGENT_*` lifecycle, and the child's token deltas under the subagent identity. | ## Upstream ships no HTTP endpoint @@ -37,9 +37,11 @@ Mastra persists memory and suspended-run snapshots to LibSQL file storage. Resum ## How subagents surface -Mastra registers a child agent as a delegation tool named `agent-`, so a delegation crosses the wire as ordinary `TOOL_CALL_*` frames whose result carries the child's final text. An in-tree injector in the hosting service (`deployments/ag-ui-mastra/subagent-emitter.mjs`) keys off those frames: it emits `SUBAGENT_STARTED` when the delegation tool call starts, an attributed text message carrying the child's final text, and `SUBAGENT_FINISHED` when the result lands. +Mastra registers a child agent as a delegation tool named `agent-`, and while the child runs its every chunk is forwarded on the parent stream as a public `tool-output` chunk. The runtime's bridge drops those chunks and withholds the delegation tool call until the child resolves, so on its own the wire would carry only the child's final text, after a silent gap. -The cell is Partial rather than green because the runtime's bridge does not forward child tokens: the child's deltas exist in-process, but `@ag-ui/mastra` drops its `tool-output` chunks and buffers the delegation burst until it resolves, so the card fills in at completion instead of streaming. That is a property of the bridge, not of the protocol or the adapter. The wire capture behind this cell is committed at [`cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md`](https://github.com/cacheplane/angular-agent-framework/tree/main/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md). +The hosting service therefore wraps the agent in a small stream tee (`deployments/ag-ui-mastra/streaming-tee.mjs`) that observes each chunk before the bridge processes it, and a per-run injector (`deployments/ag-ui-mastra/subagent-emitter.mjs`) maps them to the protocol: the delegation `tool-call` chunk becomes an eager `TOOL_CALL_START`, `TOOL_CALL_ARGS`, and `TOOL_CALL_END` plus `SUBAGENT_STARTED`; each child text delta becomes a `TEXT_MESSAGE_CONTENT` attributed to the subagent; and the delegation result becomes `SUBAGENT_FINISHED` or `SUBAGENT_ERROR`. The bridge's own copy of the delegation tool call, flushed at the result, is dropped so the wire carries it once. + +The bridge itself is unmodified; the tee touches only the public agent members the bridge reads. The child's tokens reach the card while it is still running, which is what flips this cell to Supported. The wire capture behind this cell is committed at [`cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md`](https://github.com/cacheplane/angular-agent-framework/tree/main/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md). ## How the Mastra row was measured diff --git a/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md b/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md index da27af630..6f074133d 100644 --- a/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md +++ b/cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md @@ -14,7 +14,7 @@ this document is the only artifact. | Child registration | `agents: { weather_forecaster: childAgent }` on the supervisor's `Agent` config. | | Delegation hooks | `delegation: { onDelegationStart, onDelegationComplete }` in stream options; works via the supervisor's `defaultOptions` (verified live — both hooks fired). | | Delegation on the AG-UI wire today | An ordinary backend tool call named `agent-`: `TOOL_CALL_START` → one `TOOL_CALL_ARGS` blob → `TOOL_CALL_END` → `TOOL_CALL_RESULT` whose `content` is JSON `{text, subAgentThreadId, subAgentResourceId, subAgentToolResults}`. | -| Does child text stream incrementally anywhere? | **In-process yes, on the wire no.** The parent `fullStream` carries every child chunk wrapped as `tool-output` (`payload: {output: , toolCallId, toolName}`) — 82 inner `text-delta` chunks in the raw tap — but `@ag-ui/mastra`'s chunk processor drops them (`case "tool-output": break`). Only the final text reaches AG-UI, inside `TOOL_CALL_RESULT`. | +| Does child text stream incrementally anywhere? | **In-process yes, on the wire no.** The parent `fullStream` carries every child chunk wrapped as `tool-output` (`payload: {output: , toolCallId, toolName}`) — 82 inner `text-delta` chunks in the raw tap — but `@ag-ui/mastra`'s chunk processor drops them (`case "tool-output": break`). Only the final text reaches AG-UI, inside `TOOL_CALL_RESULT`. — superseded, see [After streaming](#after-streaming) | | Delegation tool-call id (for `parentToolCallId`) | The LLM's tool-call id (e.g. `call_aUfV9K0RCDZdZK3NWt9dRDKx`). Identical across `TOOL_CALL_START/ARGS/END/RESULT` and both hooks' `toolCallId`. | ## Installed-API details (`node_modules/@mastra/core/dist`) @@ -221,12 +221,13 @@ Notes: (`case "tool-output": break` in @ag-ui/mastra), so the one TEXT_MESSAGE_CONTENT carries the child's entire final text. -## Browser verification +## Browser verification (pre-streaming, superseded below) Live check 2026-09-02: real-key `deployments/ag-ui-mastra` on the topic port + `npx nx serve cockpit-runtimes-mastra-angular`, driving "Plan a trip to Bear -Lake this weekend - what will the weather be?" in the real UI. Screenshot of -the completed, expanded card: `e2e/manual/subagent-card-live.png`. +Lake this weekend - what will the weather be?" in the real UI. The screenshot +referenced here was replaced by the post-streaming capture (see "Browser +verification (after streaming)"). - The card renders from the injected events with ZERO component code: `chat-tool-calls` groups on `parentToolCallId` and mounts @@ -244,3 +245,158 @@ the completed, expanded card: `e2e/manual/subagent-card-live.png`. `data-state="done"` on first sight). The emitter is not the limiter; @ag-ui/mastra's buffered tool-call flush is (same upstream drop/buffer behavior documented above). + +## Streaming spike + +Task 0 of the streaming PR (spec: +`docs/superpowers/specs/2026-09-02-mastra-subagent-streaming-design.md`). +Captured 2026-09-02 by calling the supervisor's `stream()` directly with the +delegation prompt and logging every `fullStream` chunk. Pins unchanged +(`@mastra/core@1.63.2`, `@ag-ui/mastra@1.1.2`). + +### Bridge members touched (from `node_modules/@ag-ui/mastra/dist/mastra-*.mjs`) + +- `'getMemory' in agent` (`isLocalMastraAgent`) — decides the local dispatch path. +- `agent.stream(messages, options)` → reads `.fullStream` (consumed by + `processFullStream`), then `.traceId` and `.usage` for RUN_FINISHED. +- `agent.resumeStream(resume, options)` → the same `.fullStream` read. +- `agent.getMemory({requestContext})`, `agent.listTools(...)`, `agent.model`. +- `parentMessageId` on `TOOL_CALL_START` is the bridge's current message id, + set by `onMessageId` from the last `start` / `step-start` chunk's + `payload.messageId` (and re-randomized after `step-finish` / `finish`). +- The delegation `tool-call` chunk is buffered (`u = {toolCallId, toolName, + args}`) and flushed as START+ARGS+END only by the next flushing chunk — + for a delegation that is the `tool-result`, because every `tool-output` + in between hits `case "tool-output": break`. + +### Raw fullStream order (one delegation, 248 chunks) + +``` + 1 start messageId=b45dfabd-… ← parentMessageId source + 2 step-start messageId=b45dfabd-… + 3-51 tool-call-input-streaming-start / tool-call-delta x48 / -end (agent-weather_forecaster) +52 tool-call toolCallId=call_s7dO… toolName=agent-weather_forecaster + args={prompt, threadId:null, resourceId:null, instructions:null, maxSteps:5, …} ← complete +53 tool-output/start output.payload={id:'weather_forecaster', messageId:467ded64-…} +54 tool-output/step-start +55-74 tool-output/tool-call-input-streaming-* + tool-call + tool-result (inner updateWorkingMemory) +75 tool-output/step-finish +76 tool-output/step-start +77 tool-output/text-start output.payload.id=msg_0dad… +78-185 tool-output/text-delta x108 output.payload.text="Here's", " the", " weather", … +186 tool-output/text-end +187 tool-output/step-finish +188 tool-output/finish +189 tool-result toolCallId=call_s7dO… result keys={text, subAgentThreadId, subAgentResourceId, subAgentToolResults} +190 step-finish, 191 step-start, 192-207 parent updateWorkingMemory tool call, +208 step-finish, 209 step-start, 210 text-start, 211-245 text-delta x35, 246 text-end, +247 step-finish, 248 finish +``` + +Findings: + +- Chunk order is `tool-call` (args complete) → `tool-output` x135 (the whole + child stream, incrementally) → `tool-result`, exactly as the earlier tap. +- The child emits an explicit `text-start` before its deltas and `text-end` + after them; the lazy-START path in the injector is defensive only. +- `parentMessageId` for the eager `TOOL_CALL_START` is the `messageId` of the + last top-level `start` / `step-start` chunk (`b45dfabd-…` here), which is + the same id the bridge would stamp when it flushes at `tool-result`. +- `tool-call-suspended` never appears in this capture (the delegation tool + does not suspend; the demo child has no suspending tools). The injector + still maps it defensively to `SUBAGENT_FINISHED {outcome:{type:'suspended'}}`. +- Inner sub-agent tool chunks (`updateWorkingMemory` — 16 `tool-call-delta` + plus `tool-call` / `tool-result`) are present under `tool-output` and are + ignored by this PR (out of scope). + +### Card-mount check (`libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts`) + +`groups()` iterates the parent message's `toolCalls()` list and mounts +`chat-subagent-card` only when `subs.has(tc.id)` for an existing tool-call +entry (`Subagent.toolCallId` is the anchor). A `SUBAGENT_STARTED` with no +prior `TOOL_CALL_START` for its `parentToolCallId` therefore renders NO card: +synthesizing the eager `TOOL_CALL_START/ARGS/END` on the `tool-call` chunk +is required, not optional, for the card to appear before the child finishes. + +## After streaming + +Live smoke 2026-09-02 against the committed stream tee +(`deployments/ag-ui-mastra/streaming-tee.mjs`) + the chunk-aware injector +(`subagent-emitter.mjs`) wired in `server.mjs`. Pins unchanged +(`@ag-ui/mastra@1.1.2`, `@mastra/core@1.63.2`); the bridge itself is not +modified. Request: `POST /agent/mastra` with the same delegation prompt, +timestamped from the request start (scrubbed): + +``` +t= 169ms {"type":"RUN_STARTED","threadId":"t-live-…","runId":"r-live-…"} +t= 1578ms {"type":"TOOL_CALL_START","parentMessageId":"5764473e-…","toolCallId":"call_nXsj…","toolCallName":"agent-weather_forecaster"} ← eager (synthesized on the tool-call chunk) +t= 1578ms {"type":"TOOL_CALL_ARGS","toolCallId":"call_nXsj…","delta":"{\"prompt\":\"What is the weather forecast for Bear Lake this w…"} +t= 1578ms {"type":"TOOL_CALL_END","toolCallId":"call_nXsj…"} +t= 1578ms {"type":"SUBAGENT_STARTED","subagentRunId":"call_nXsj…-sub","name":"weather_forecaster","parentToolCallId":"call_nXsj…"} +t= 6263ms {"type":"TEXT_MESSAGE_START","messageId":"call_nXsj…-sub-m1","role":"assistant","subagentRunId":"call_nXsj…-sub"} +t= 6266ms {"type":"TEXT_MESSAGE_CONTENT","messageId":"call_nXsj…-sub-m1","delta":"Here's","subagentRunId":"call_nXsj…-sub"} +t= 6306ms {"type":"TEXT_MESSAGE_CONTENT",…,"delta":" the",…} +t= 6314ms {"type":"TEXT_MESSAGE_CONTENT",…,"delta":" weather",…} + … x94 attributed deltas, 399 chars, t=6266→7183ms … +t= 7183ms {"type":"TEXT_MESSAGE_END","messageId":"call_nXsj…-sub-m1","subagentRunId":"call_nXsj…-sub"} +t= 7382ms {"type":"SUBAGENT_FINISHED","subagentRunId":"call_nXsj…-sub","outcome":{"type":"success"}} +t= 7382ms {"type":"TOOL_CALL_RESULT","toolCallId":"call_nXsj…","content":"{\"text\":\"Here's the weather forecast for Bear Lake this weekend:…","messageId":"731b3738-…","role":"tool"} +t= 8138ms {"type":"STATE_SNAPSHOT","snapshot":{}} + {"type":"TEXT_MESSAGE_CHUNK",…} x68 — the parent's own summary + {"type":"RUN_FINISHED",…} +``` + +Event-type totals: 1 each RUN_STARTED / TOOL_CALL_START / TOOL_CALL_ARGS / +TOOL_CALL_END / SUBAGENT_STARTED / TEXT_MESSAGE_START / TEXT_MESSAGE_END / +SUBAGENT_FINISHED / TOOL_CALL_RESULT / STATE_SNAPSHOT / RUN_FINISHED, +94 TEXT_MESSAGE_CONTENT (all carrying `subagentRunId`), 68 TEXT_MESSAGE_CHUNK. + +Verified on the wire: + +- Eager `TOOL_CALL_START` + `SUBAGENT_STARTED` land at t=1.6 s, ~4.7 s before + the first attributed delta and ~5.8 s before `TOOL_CALL_RESULT` — the gap + the pre-streaming capture spent silent. +- Exactly ONE `TOOL_CALL_START` for the delegation id: the bridge's buffered + copy (flushed at `tool-result`) is dropped by the injector's dedupe. +- `SUBAGENT_FINISHED` precedes `TOOL_CALL_RESULT` (both written when the + `tool-result` chunk is observed, before the bridge processes it). +- The ~4.7 s between STARTED and the first delta is the child's own inner + `updateWorkingMemory` tool call plus model latency, not buffering: the raw + tap shows the same inner call preceding the child's `text-start`. + +Caveats and the fallback design: + +- Suspended delegations: if a child ever emitted `tool-call-suspended`, the + bridge would retract its buffered tool call (never emit START), but the + eager START is already painted; the injector closes the card with + `SUBAGENT_FINISHED {outcome:{type:'suspended'}}` and lets the bridge's + CUSTOM on_interrupt + RUN_FINISHED interrupt outcome through. The demo's + delegation does not suspend (no suspending tools on the child); this path + is unit-tested only. +- Inner sub-agent tool calls stay out of scope (ignored under `tool-output`). +- Alternative considered: subclassing the bridge and overriding its chunk + processor. Rejected because it couples to three TS-private methods, the + bridge's `clone()` constructs the base class (dropping overrides), and the + method signatures drift on upstream `main`. The Proxy tee touches only the + public agent surface the bridge reads (`'getMemory' in agent`, `stream`, + `resumeStream`, `getMemory`, `listTools`, `model`) and remains the fallback + design should the public seam ever move. + +## Browser verification (after streaming) + +Live check 2026-09-02, same servers, driving the real UI with "Plan a trip to +Bear Lake this weekend - what will the weather be?". Screenshot of the +completed, expanded card: `e2e/manual/subagent-card-live.png`. + +- The card mounts in the `running` state with "0 message(s)" at ~2.5–3.5 s + after send — before any child text exists — because the eager + `TOOL_CALL_START` gives `chat-tool-calls` the tool-call entry to anchor on. +- Headless 150 ms poll of the card's `innerText` length while + `data-state="running"` (distinct samples): 67 → 110 → 188 → 261 → 279 → + 306 → 330 → 457 → 494 chars (t=3454 ms → 10827 ms), then `done` (the card + collapses to its header, 68 chars) at t=10982 ms. Text visibly grows inside + the running card; the earlier capture never left `running` visible at all. +- A second, interactive run through the dev browser pane showed the same + shape at coarser (1 s, throttled) sampling: 67 → 131 → 474 chars while + `running`, then `done`. +- The parent's own summary streams below the card afterwards, unchanged. diff --git a/cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png b/cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png index fac129361..ab4157760 100644 Binary files a/cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png and b/cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png differ diff --git a/deployments/ag-ui-mastra/server.mjs b/deployments/ag-ui-mastra/server.mjs index 8987db394..6641f90f5 100644 --- a/deployments/ag-ui-mastra/server.mjs +++ b/deployments/ag-ui-mastra/server.mjs @@ -23,6 +23,7 @@ import { dirname, resolve } from 'node:path'; import { MastraAgent } from '@ag-ui/mastra'; import { createMastra } from './agents.mjs'; import { createSubagentInjector } from './subagent-emitter.mjs'; +import { withDelegationTee } from './streaming-tee.mjs'; const AG_UI_INTERNAL_TOKEN = process.env.AG_UI_INTERNAL_TOKEN; if (!AG_UI_INTERNAL_TOKEN) { @@ -100,27 +101,40 @@ export function createAgUiServer() { connection: 'keep-alive', }); + // One injector per run with two inputs, both writing through the same + // SSE frame writer: + // - `chunk()`: raw Mastra fullStream chunks observed through the stream + // tee BEFORE the bridge processes them — the eager TOOL_CALL_* + + // SUBAGENT_STARTED on the delegation `tool-call`, the child's + // attributed TEXT_MESSAGE_* deltas from `tool-output`, and + // SUBAGENT_FINISHED/ERROR on `tool-result`. + // - `eventsFor()`: the bridge's own AG-UI events, with its later buffered + // TOOL_CALL_START/ARGS/END copies for a synthesized id dropped. + const injector = createSubagentInjector(); + const write = (event) => res.write(sseFrame(event)); + const observe = (chunk) => { + for (const e of injector.chunk(chunk)) write(e); + }; + // A fresh bridge per request: MastraAgent carries per-run state. // resourceId scopes Mastra memory (threads live under a resource); - // keying it by AG-UI threadId gives per-conversation memory. + // keying it by AG-UI threadId gives per-conversation memory. The bridge + // receives the teed agent; it is otherwise unmodified. const bridge = new MastraAgent({ agentId: topic, - agent, + agent: withDelegationTee(agent, observe), resourceId: input.threadId, }); - // One injector per run: turns delegation tool calls (`agent-`) - // into SUBAGENT_* frames around the events the bridge already emits. - const injector = createSubagentInjector(); const sub = bridge.run(input).subscribe({ next: (event) => { - for (const e of injector.eventsFor(event)) res.write(sseFrame(e)); + for (const e of injector.eventsFor(event)) write(e); }, error: (err) => { // Map failures into the protocol instead of killing the socket: // the client finalizes the run as an error rather than hanging. const runError = { type: 'RUN_ERROR', message: String(err?.message ?? err) }; - for (const e of injector.eventsFor(runError)) res.write(sseFrame(e)); + for (const e of injector.eventsFor(runError)) write(e); res.end(); }, complete: () => { diff --git a/deployments/ag-ui-mastra/streaming-tee.mjs b/deployments/ag-ui-mastra/streaming-tee.mjs new file mode 100644 index 000000000..191c17916 --- /dev/null +++ b/deployments/ag-ui-mastra/streaming-tee.mjs @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: MIT +// Delegation stream tee: observe a Mastra Agent's `fullStream` chunks before +// the @ag-ui/mastra bridge consumes them. +// +// Why a Proxy and not a bridge subclass: the bridge (1.1.2) reads the agent +// only through public members — `'getMemory' in agent`, `stream()`, +// `resumeStream()`, `getMemory()`, `listTools()`, `model` — and consumes +// `stream()`'s `.fullStream` in a private chunk processor that DROPS every +// `tool-output` chunk (the in-process sub-agent deltas). Wrapping the agent +// keeps the bridge unmodified and version-independent; wrapping only the +// `fullStream` iterator keeps `.traceId` / `.usage` / everything else on the +// stream result intact. +// +// Ordering guarantee: the bridge is the single reader of the wrapped +// generator, so `observe(chunk)` runs strictly before the bridge processes +// that same chunk. Anything the observer writes to the SSE socket therefore +// lands ahead of the bridge's own events for the chunk. + +/** + * Wrap an async iterable so `observe` sees each item before it is yielded. + * Observer failures are logged and never break the consumer. + */ +async function* tee(source, observe) { + for await (const chunk of source) { + try { + observe(chunk); + } catch (err) { + console.warn('[streaming-tee] observer threw; chunk still forwarded:', err); + } + yield chunk; + } +} + +const WRAPPED_METHODS = new Set(['stream', 'resumeStream']); + +/** + * @template {object} T + * @param {T} agent the real Mastra Agent (or any object with `stream()`). + * @param {(chunk: object) => void} observe called with every `fullStream` + * chunk of every `stream()` / `resumeStream()` result, before the bridge + * receives it. + * @returns {T} a Proxy that forwards every member to `agent` with `this` + * bound to the real instance (so `#private` fields keep working), except + * that `stream` / `resumeStream` return their result with `fullStream` + * replaced by the teed generator. + */ +export function withDelegationTee(agent, observe) { + return new Proxy(agent, { + get(target, prop, receiver) { + const value = Reflect.get(target, prop, target); + if (typeof value !== 'function') return value; + if (!WRAPPED_METHODS.has(prop)) { + // Bind to the REAL target, not the receiver: class methods that touch + // `#private` state throw when `this` is the Proxy. + return value.bind(target); + } + return async (...args) => { + const result = await value.apply(target, args); + if (!result || typeof result !== 'object' || !result.fullStream) return result; + const wrapped = tee(result.fullStream, observe); + // Keep the original result object (its getters for traceId/usage/ + // text/etc. must still resolve); only shadow `fullStream`. + return new Proxy(result, { + get(res, key) { + if (key === 'fullStream') return wrapped; + const v = Reflect.get(res, key, res); + return typeof v === 'function' ? v.bind(res) : v; + }, + }); + }; + }, + has(target, prop) { + // Restates the default `has` behavior explicitly, so the bridge's + // `'getMemory' in agent` check reads as a deliberate, documented + // contract of this Proxy rather than an accident of the default trap. + return prop in target; + }, + }); +} diff --git a/deployments/ag-ui-mastra/subagent-emitter.mjs b/deployments/ag-ui-mastra/subagent-emitter.mjs index 24ab9fd9c..17c13993a 100644 --- a/deployments/ag-ui-mastra/subagent-emitter.mjs +++ b/deployments/ag-ui-mastra/subagent-emitter.mjs @@ -2,58 +2,252 @@ // SUBAGENT_* injection for Mastra delegation tool calls. // // Mastra surfaces a registered sub-agent as an ordinary backend tool named -// `agent-`: TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END → -// TOOL_CALL_RESULT whose `content` is JSON `{text, subAgentThreadId, ...}` -// (measured: cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md). -// The upstream @ag-ui/mastra bridge drops the in-process child deltas -// (`case "tool-output": break`), so the honest wire contract here is a -// single final text chunk per delegation. +// `agent-`. On the wire the upstream @ag-ui/mastra bridge buffers +// that tool call and flushes TOOL_CALL_START → ARGS → END only when the +// TOOL_CALL_RESULT arrives (content = JSON `{text, subAgentThreadId, ...}`), +// and it DROPS the in-process child deltas (`case "tool-output": break`) — +// measured in cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md. // -// This module is a pure transform over the outbound AG-UI event stream — -// keyed off the events themselves, NOT the Mastra delegation hooks, which -// fire in a different async context with no ordering guarantee relative to -// the Observable frames. +// This module is a pure per-run transform with TWO inputs: // -// Injected sequence per delegation tool call (child key = tool name -// minus the `agent-` prefix, subagentRunId = `-sub`): -// - AFTER TOOL_CALL_START: SUBAGENT_STARTED {subagentRunId, name, -// parentToolCallId} -// - BEFORE TOOL_CALL_RESULT (success): TEXT_MESSAGE_START/CONTENT/END -// carrying the child's final text under the subagent identity, then -// SUBAGENT_FINISHED {outcome:{type:'success'}} -// - BEFORE TOOL_CALL_RESULT (failure — parsed content says success:false -// or finishReason:'error'): SUBAGENT_ERROR {subagentRunId, message} -// - Terminal cleanup: a RUN_ERROR or RUN_FINISHED arriving while -// delegations are still pending (no TOOL_CALL_RESULT seen — e.g. the -// Observable errored mid-delegation) closes each pending card with -// SUBAGENT_ERROR before the terminal frame, so no card is left spinning. -// In the measured captures the RESULT always precedes the terminal frame, -// so this path is defensive only. +// - `chunk(c)`: raw Mastra `fullStream` chunks, observed through the stream +// tee (streaming-tee.mjs) BEFORE the bridge processes each one. This is +// where the child text actually streams, so it is the primary emitter: +// - `tool-call` named `agent-*` → synthesized EAGER TOOL_CALL_START/ARGS/END +// (args are complete in the chunk; parentMessageId = the last +// `start`/`step-start` chunk's messageId, the same id the bridge would +// stamp) + SUBAGENT_STARTED {subagentRunId: `-sub`, name, +// parentToolCallId}. The card needs the parent tool call present to mount, +// so synthesis is required rather than optional. +// - `tool-output` for a tracked id: inner `text-start` → attributed +// TEXT_MESSAGE_START (opened lazily on the first delta if absent), +// `text-delta` → TEXT_MESSAGE_CONTENT, `text-end` → TEXT_MESSAGE_END. +// Inner tool chunks are ignored (out of scope). +// - `tool-result` / `tool-error` → close any open message, then +// SUBAGENT_FINISHED {success} or SUBAGENT_ERROR (result says +// success:false / finishReason:'error', or the tool errored). When NO +// delta was observed the old single-chunk TEXT_MESSAGE_* synthesis from +// `result.text` fires as the fallback. +// - `tool-call-suspended` → close any open message, SUBAGENT_FINISHED +// {outcome:{type:'suspended'}}. The eager START has already been painted +// (the bridge would have retracted it) — accepted caveat; the demo's +// delegation never suspends. +// +// - `eventsFor(event)`: the bridge's outbound AG-UI events. For an id that +// was synthesized above, the bridge's later buffered TOOL_CALL_START/ARGS/ +// END copies are DROPPED (its TOOL_CALL_RESULT passes through untouched). +// For an id the chunk path never saw (tee not wired), the original +// event-keyed behavior stays: SUBAGENT_STARTED after TOOL_CALL_START and the +// single-chunk text + SUBAGENT_FINISHED/ERROR before TOOL_CALL_RESULT. +// Terminal cleanup: RUN_ERROR / RUN_FINISHED with delegations still pending +// closes open child messages and emits SUBAGENT_ERROR per pending id, +// exactly once, before the terminal frame. const AGENT_TOOL_PREFIX = 'agent-'; +/** + * @typedef {object} Entry + * @property {string} subagentRunId + * @property {string} name + * @property {boolean} synthesized TOOL_CALL_* were emitted from the chunk path + * @property {boolean} deltasSeen at least one child text delta was forwarded + * @property {boolean} messageOpen + * @property {string|undefined} messageId + * @property {number} messageCount + */ + +/** Result-shape failure check shared by both inputs. */ +function parseResult(raw) { + if (raw !== null && typeof raw === 'object') return raw; + if (typeof raw !== 'string') return undefined; + try { + return JSON.parse(raw); + } catch { + return undefined; + } +} + +function isFailure(parsed) { + return ( + parsed !== undefined && + typeof parsed === 'object' && + parsed !== null && + (parsed.success === false || parsed.finishReason === 'error') + ); +} + +function failureMessage(parsed) { + return String(parsed.error ?? parsed.text ?? 'sub-agent delegation failed'); +} + /** * Create a per-run injector. * - * @returns {{ eventsFor(event: object): object[] }} — for each outbound - * AG-UI event, the ordered list of frames to write (injections plus the - * original event). Non-delegation events pass through as `[event]`. + * @returns {{ chunk(chunk: object): object[], eventsFor(event: object): object[] }} + * `chunk` returns the AG-UI events to write for a raw Mastra chunk (usually + * none); `eventsFor` returns, for each outbound bridge event, the ordered + * list of frames to write (injections plus, unless deduped, the event). */ export function createSubagentInjector() { - /** @type {Map} pending delegations by toolCallId */ + /** @type {Map} pending delegations by toolCallId */ const pending = new Map(); + /** Ids whose TOOL_CALL_START/ARGS/END were synthesized — bridge copies drop. */ + const synthesized = new Set(); + /** The bridge's current parent message id (last start/step-start chunk). */ + let parentMessageId; + + function newEntry(toolCallId, toolName, isSynthesized) { + const entry = { + subagentRunId: `${toolCallId}-sub`, + name: toolName.slice(AGENT_TOOL_PREFIX.length), + synthesized: isSynthesized, + deltasSeen: false, + messageOpen: false, + messageId: undefined, + messageCount: 0, + }; + pending.set(toolCallId, entry); + return entry; + } + + function openMessage(toolCallId, entry) { + entry.messageCount += 1; + entry.messageId = `${toolCallId}-sub-m${entry.messageCount}`; + entry.messageOpen = true; + return { + type: 'TEXT_MESSAGE_START', + messageId: entry.messageId, + role: 'assistant', + subagentRunId: entry.subagentRunId, + }; + } + + function closeMessage(entry) { + if (!entry.messageOpen) return []; + entry.messageOpen = false; + return [{ type: 'TEXT_MESSAGE_END', messageId: entry.messageId, subagentRunId: entry.subagentRunId }]; + } + + /** Close + FINISHED/ERROR for a result; falls back to single-chunk text when no deltas streamed. */ + function finalize(toolCallId, entry, rawResult) { + pending.delete(toolCallId); + const { subagentRunId } = entry; + const parsed = parseResult(rawResult); + if (isFailure(parsed)) { + return [...closeMessage(entry), { type: 'SUBAGENT_ERROR', subagentRunId, message: failureMessage(parsed) }]; + } + const out = closeMessage(entry); + if (!entry.deltasSeen) { + const raw = typeof rawResult === 'string' ? rawResult : JSON.stringify(rawResult); + const text = typeof parsed?.text === 'string' ? parsed.text : raw; + out.push(openMessage(toolCallId, entry)); + out.push({ type: 'TEXT_MESSAGE_CONTENT', messageId: entry.messageId, delta: text, subagentRunId }); + out.push(...closeMessage(entry)); + } + out.push({ type: 'SUBAGENT_FINISHED', subagentRunId, outcome: { type: 'success' } }); + return out; + } + + function fail(toolCallId, entry, message) { + pending.delete(toolCallId); + return [...closeMessage(entry), { type: 'SUBAGENT_ERROR', subagentRunId: entry.subagentRunId, message }]; + } return { + chunk(chunk) { + const payload = chunk?.payload ?? {}; + switch (chunk?.type) { + case 'start': + case 'step-start': { + if (payload.messageId) parentMessageId = payload.messageId; + return []; + } + + case 'tool-call': { + const { toolCallId, toolName = '' } = payload; + if (!toolCallId || !toolName.startsWith(AGENT_TOOL_PREFIX)) return []; + const entry = newEntry(toolCallId, toolName, true); + synthesized.add(toolCallId); + return [ + { + type: 'TOOL_CALL_START', + ...(parentMessageId !== undefined ? { parentMessageId } : {}), + toolCallId, + toolCallName: toolName, + }, + { type: 'TOOL_CALL_ARGS', toolCallId, delta: JSON.stringify(payload.args ?? {}) }, + { type: 'TOOL_CALL_END', toolCallId }, + { + type: 'SUBAGENT_STARTED', + subagentRunId: entry.subagentRunId, + name: entry.name, + parentToolCallId: toolCallId, + }, + ]; + } + + case 'tool-output': { + const entry = pending.get(payload.toolCallId); + const inner = payload.output; + if (!entry || !inner) return []; + switch (inner.type) { + case 'text-start': + return [...closeMessage(entry), openMessage(payload.toolCallId, entry)]; + case 'text-delta': { + const text = inner.payload?.text; + if (typeof text !== 'string' || text.length === 0) return []; + const out = entry.messageOpen ? [] : [openMessage(payload.toolCallId, entry)]; + entry.deltasSeen = true; + out.push({ + type: 'TEXT_MESSAGE_CONTENT', + messageId: entry.messageId, + delta: text, + subagentRunId: entry.subagentRunId, + }); + return out; + } + case 'text-end': + return closeMessage(entry); + default: + return []; // inner tool chunks etc. — out of scope + } + } + + case 'tool-result': { + const entry = pending.get(payload.toolCallId); + if (!entry) return []; + return finalize(payload.toolCallId, entry, payload.result); + } + + case 'tool-error': { + const entry = pending.get(payload.toolCallId); + if (!entry) return []; + const err = payload.error; + return fail(payload.toolCallId, entry, String(err?.message ?? err ?? 'sub-agent delegation failed')); + } + + case 'tool-call-suspended': { + const entry = pending.get(payload.toolCallId); + if (!entry) return []; + pending.delete(payload.toolCallId); + return [ + ...closeMessage(entry), + { type: 'SUBAGENT_FINISHED', subagentRunId: entry.subagentRunId, outcome: { type: 'suspended' } }, + ]; + } + + default: + return []; + } + }, + eventsFor(event) { switch (event.type) { case 'TOOL_CALL_START': { + if (synthesized.has(event.toolCallId)) return []; // eager copy already on the wire const name = event.toolCallName ?? ''; if (!name.startsWith(AGENT_TOOL_PREFIX)) return [event]; - const entry = { - subagentRunId: `${event.toolCallId}-sub`, - name: name.slice(AGENT_TOOL_PREFIX.length), - }; - pending.set(event.toolCallId, entry); + const entry = newEntry(event.toolCallId, name, false); return [ event, { @@ -65,54 +259,22 @@ export function createSubagentInjector() { ]; } + case 'TOOL_CALL_ARGS': + case 'TOOL_CALL_END': + return synthesized.has(event.toolCallId) ? [] : [event]; + case 'TOOL_CALL_RESULT': { const entry = pending.get(event.toolCallId); - if (!entry) return [event]; // not a delegation (or unmatched) — pass through - pending.delete(event.toolCallId); - const { subagentRunId } = entry; - - const raw = typeof event.content === 'string' ? event.content : JSON.stringify(event.content); - let parsed; - try { - parsed = JSON.parse(raw); - } catch { - parsed = undefined; - } - const failed = - parsed !== undefined && - typeof parsed === 'object' && - parsed !== null && - (parsed.success === false || parsed.finishReason === 'error'); - if (failed) { - return [ - { - type: 'SUBAGENT_ERROR', - subagentRunId, - message: String(parsed.error ?? parsed.text ?? 'sub-agent delegation failed'), - }, - event, - ]; - } - - const text = typeof parsed?.text === 'string' ? parsed.text : raw; - const messageId = `${event.toolCallId}-sub-m1`; - return [ - { type: 'TEXT_MESSAGE_START', messageId, role: 'assistant', subagentRunId }, - { type: 'TEXT_MESSAGE_CONTENT', messageId, delta: text, subagentRunId }, - { type: 'TEXT_MESSAGE_END', messageId, subagentRunId }, - { type: 'SUBAGENT_FINISHED', subagentRunId, outcome: { type: 'success' } }, - event, - ]; + if (!entry) return [event]; // not a delegation, already finalized by the chunk path, or unmatched + return [...finalize(event.toolCallId, entry, event.content), event]; } case 'RUN_ERROR': case 'RUN_FINISHED': { if (pending.size === 0) return [event]; - const cleanup = [...pending.values()].map(({ subagentRunId }) => ({ - type: 'SUBAGENT_ERROR', - subagentRunId, - message: 'delegation did not complete before the run terminated', - })); + const cleanup = [...pending.entries()].flatMap(([toolCallId, entry]) => + fail(toolCallId, entry, 'delegation did not complete before the run terminated'), + ); pending.clear(); return [...cleanup, event]; } diff --git a/deployments/ag-ui-mastra/test/streaming-tee.test.mjs b/deployments/ag-ui-mastra/test/streaming-tee.test.mjs new file mode 100644 index 000000000..86cc9fb04 --- /dev/null +++ b/deployments/ag-ui-mastra/test/streaming-tee.test.mjs @@ -0,0 +1,214 @@ +// SPDX-License-Identifier: MIT +// Unit tests for the delegation stream tee — a Proxy over a Mastra Agent +// that lets an observer see every `fullStream` chunk BEFORE the @ag-ui/mastra +// bridge consumes it, while every other member forwards to the real agent. +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { Agent } from '@mastra/core/agent'; +import { withDelegationTee } from '../streaming-tee.mjs'; + +/** A real Agent (no model call is ever made) so `#private` fields exist. */ +function realAgent() { + return new Agent({ + id: 'tee-probe', + name: 'tee-probe', + instructions: 'probe', + model: 'openai/gpt-4o-mini', + }); +} + +/** Fake agent whose stream() yields two chunks and carries a sibling prop. */ +function fakeAgent(trace = []) { + return { + getMemory() {}, + listTools() {}, + async stream() { + return { + fullStream: (async function* () { + yield { type: 'a' }; + yield { type: 'b' }; + })(), + text: 'x', + }; + }, + async resumeStream() { + return { + fullStream: (async function* () { + yield { type: 'r1' }; + })(), + usage: 'u', + }; + }, + _trace: trace, + }; +} + +// ── Task 0 feasibility: the bridge's agent contract survives the Proxy ──── +test('proxy over a real Agent still satisfies the bridge contract', async () => { + const agent = realAgent(); + // Stub stream() on the instance: the bridge awaits it and reads fullStream. + agent.stream = async () => ({ + fullStream: (async function* () { + yield { type: 'start', payload: { messageId: 'm1' } }; + })(), + text: Promise.resolve('x'), + }); + + const proxy = withDelegationTee(agent, () => {}); + + assert.ok('getMemory' in proxy, "bridge's isLocalMastraAgent check: 'getMemory' in agent"); + assert.equal(typeof proxy.listTools, 'function'); + assert.equal(typeof proxy.getMemory, 'function'); + assert.equal(typeof proxy.resumeStream, 'function'); + assert.equal(proxy.model, agent.model, 'model property forwards'); + + const result = await proxy.stream([], {}); + assert.ok(result && typeof result === 'object'); + assert.equal(typeof result.fullStream[Symbol.asyncIterator], 'function', 'fullStream is async-iterable'); + const seen = []; + for await (const c of result.fullStream) seen.push(c.type); + assert.deepEqual(seen, ['start']); +}); + +// ── Task 1: ordering, passthrough, resumeStream, `this`, observer errors ── +test('observer sees each chunk BEFORE the consumer receives it', async () => { + const trace = []; + const proxy = withDelegationTee(fakeAgent(), (c) => trace.push(`obs:${c.type}`)); + const result = await proxy.stream([], {}); + for await (const c of result.fullStream) trace.push(`out:${c.type}`); + assert.deepEqual(trace, ['obs:a', 'out:a', 'obs:b', 'out:b']); +}); + +test('non-fullStream properties of the stream result are preserved', async () => { + const proxy = withDelegationTee(fakeAgent(), () => {}); + const result = await proxy.stream([], {}); + assert.equal(result.text, 'x'); +}); + +test('resumeStream is wrapped the same way', async () => { + const trace = []; + const proxy = withDelegationTee(fakeAgent(), (c) => trace.push(`obs:${c.type}`)); + const result = await proxy.resumeStream({}, {}); + assert.equal(result.usage, 'u'); + for await (const c of result.fullStream) trace.push(`out:${c.type}`); + assert.deepEqual(trace, ['obs:r1', 'out:r1']); +}); + +test('stream/resumeStream receive the original arguments', async () => { + const calls = []; + const agent = { + getMemory() {}, + async stream(...args) { + calls.push(['stream', ...args]); + return { fullStream: (async function* () {})() }; + }, + async resumeStream(...args) { + calls.push(['resume', ...args]); + return { fullStream: (async function* () {})() }; + }, + }; + const proxy = withDelegationTee(agent, () => {}); + await proxy.stream('msgs', { runId: 'r' }); + await proxy.resumeStream({ approved: true }, { runId: 'r2' }); + assert.deepEqual(calls, [ + ['stream', 'msgs', { runId: 'r' }], + ['resume', { approved: true }, { runId: 'r2' }], + ]); +}); + +test('other members forward with `this` bound to the real agent (#private fields work)', () => { + class Probe extends Agent { + #secret = 'hidden'; + readSecret() { + return this.#secret; + } + } + const agent = new Probe({ id: 'p', name: 'p', instructions: 'p', model: 'openai/gpt-4o-mini' }); + const proxy = withDelegationTee(agent, () => {}); + // Calling through the proxy must not throw "Cannot read private member". + assert.equal(proxy.readSecret(), 'hidden'); + assert.equal(proxy.id, 'p'); + assert.equal(typeof proxy.getMemory, 'function'); +}); + +test('a throwing observer does not break the consumer', async () => { + const warnings = []; + const originalWarn = console.warn; + console.warn = (...args) => warnings.push(args); + try { + const proxy = withDelegationTee(fakeAgent(), (c) => { + if (c.type === 'a') throw new Error('observer boom'); + }); + const result = await proxy.stream([], {}); + const seen = []; + for await (const c of result.fullStream) seen.push(c.type); + assert.deepEqual(seen, ['a', 'b']); + assert.equal(warnings.length, 1); + assert.match(String(warnings[0][0]), /observer/); + } finally { + console.warn = originalWarn; + } +}); + +test('a stream result without fullStream is returned untouched', async () => { + const agent = { getMemory() {}, async stream() { return { processDataStream() {} }; } }; + const proxy = withDelegationTee(agent, () => {}); + const result = await proxy.stream([], {}); + assert.equal(typeof result.processDataStream, 'function'); + assert.equal(result.fullStream, undefined); +}); + +// ── Consumer early-exit propagates to the source generator ──────────────── +test('consumer breaking out of the wrapped fullStream closes the source generator', async () => { + let finallyRan = false; + async function* source() { + try { + yield { type: 'a' }; + yield { type: 'b' }; + yield { type: 'c' }; + } finally { + finallyRan = true; + } + } + const agent = { + getMemory() {}, + async stream() { + return { fullStream: source() }; + }, + }; + const proxy = withDelegationTee(agent, () => {}); + const result = await proxy.stream([], {}); + const seen = []; + for await (const c of result.fullStream) { + seen.push(c.type); + if (c.type === 'a') break; + } + assert.deepEqual(seen, ['a']); + assert.equal(finallyRan, true, "breaking the consumer's loop must call the source generator's return()"); +}); + +// ── Result proxy forwards prototype getters, not just own properties ────── +test('traceId/usage defined as prototype getters still resolve through the result proxy', async () => { + class StreamResult { + get traceId() { + return 'trace-123'; + } + get usage() { + return { total: 42 }; + } + } + const resultInstance = new StreamResult(); + resultInstance.fullStream = (async function* () { + yield { type: 'a' }; + })(); + const agent = { + getMemory() {}, + async stream() { + return resultInstance; + }, + }; + const proxy = withDelegationTee(agent, () => {}); + const result = await proxy.stream([], {}); + assert.equal(result.traceId, 'trace-123'); + assert.deepEqual(result.usage, { total: 42 }); +}); diff --git a/deployments/ag-ui-mastra/test/subagent-emitter.test.mjs b/deployments/ag-ui-mastra/test/subagent-emitter.test.mjs index 5cedb98fd..b883655aa 100644 --- a/deployments/ag-ui-mastra/test/subagent-emitter.test.mjs +++ b/deployments/ag-ui-mastra/test/subagent-emitter.test.mjs @@ -144,6 +144,308 @@ test('pending delegation at RUN_FINISHED → SUBAGENT_ERROR cleanup before the t ]); }); +// ── chunk() — the stream-tee input (shapes copied from the Task 0 capture) ── + +const PARENT_MID = 'b45dfabd-fec1-4072-92db-215f394625a4'; +const CHILD_TEXT_ID = 'msg_0dadf9a2c94517c2006a986e0c8ecc87d0b86efe35436a91a4'; +const ARGS = { + prompt: 'What is the weather forecast for Bear Lake this weekend?', + threadId: null, + resourceId: null, + instructions: null, + maxSteps: 5, + suspendedToolRunId: null, + resumeData: null, +}; + +const startChunk = { type: 'start', runId: 'r', payload: { messageId: PARENT_MID } }; +const stepStartChunk = { type: 'step-start', runId: 'r', payload: { messageId: PARENT_MID } }; +function toolCallChunk(tid = TID, toolName = 'agent-weather_forecaster', args = ARGS) { + return { type: 'tool-call', runId: 'r', payload: { toolCallId: tid, toolName, args } }; +} +function toolOutput(output, tid = TID) { + return { + type: 'tool-output', + runId: 'r', + payload: { output, toolCallId: tid, toolName: 'agent-weather_forecaster' }, + }; +} +const childTextStart = toolOutput({ type: 'text-start', payload: { id: CHILD_TEXT_ID } }); +const childDelta = (text) => toolOutput({ type: 'text-delta', payload: { id: CHILD_TEXT_ID, text } }); +const childTextEnd = toolOutput({ type: 'text-end', payload: { id: CHILD_TEXT_ID } }); +function toolResultChunk(result, tid = TID) { + return { + type: 'tool-result', + runId: 'r', + payload: { toolCallId: tid, toolName: 'agent-weather_forecaster', result }, + }; +} +const SUCCESS_RESULT = { + text: "Here's the weather", + subAgentThreadId: 't-1-abc', + subAgentResourceId: 't-1-weather_forecaster', + subAgentToolResults: [], +}; + +const SUB = `${TID}-sub`; +const M1 = `${TID}-sub-m1`; +const eagerToolCall = [ + { type: 'TOOL_CALL_START', parentMessageId: PARENT_MID, toolCallId: TID, toolCallName: 'agent-weather_forecaster' }, + { type: 'TOOL_CALL_ARGS', toolCallId: TID, delta: JSON.stringify(ARGS) }, + { type: 'TOOL_CALL_END', toolCallId: TID }, + { type: 'SUBAGENT_STARTED', subagentRunId: SUB, name: 'weather_forecaster', parentToolCallId: TID }, +]; + +test('chunk: agent-* tool-call → eager TOOL_CALL_START/ARGS/END + SUBAGENT_STARTED (parentMessageId from step-start)', () => { + const injector = createSubagentInjector(); + assert.deepEqual(injector.chunk(startChunk), []); + assert.deepEqual(injector.chunk(stepStartChunk), []); + assert.deepEqual(injector.chunk(toolCallChunk()), eagerToolCall); +}); + +test('chunk: parentMessageId tracks the LAST start/step-start chunk', () => { + const injector = createSubagentInjector(); + injector.chunk(startChunk); + injector.chunk({ type: 'step-start', payload: { messageId: 'later-mid' } }); + const [start] = injector.chunk(toolCallChunk()); + assert.equal(start.parentMessageId, 'later-mid'); +}); + +test('chunk: tool-output text-start/delta/end → attributed TEXT_MESSAGE_START/CONTENT/END', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + assert.deepEqual(injector.chunk(childTextStart), [ + { type: 'TEXT_MESSAGE_START', messageId: M1, role: 'assistant', subagentRunId: SUB }, + ]); + assert.deepEqual(injector.chunk(childDelta("Here's")), [ + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: "Here's", subagentRunId: SUB }, + ]); + assert.deepEqual(injector.chunk(childDelta(' the')), [ + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: ' the', subagentRunId: SUB }, + ]); + assert.deepEqual(injector.chunk(childTextEnd), [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + ]); +}); + +test('chunk: lazy TEXT_MESSAGE_START when the child omits text-start; empty deltas skipped', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + assert.deepEqual(injector.chunk(childDelta('')), []); + assert.deepEqual(injector.chunk(childDelta('Hi')), [ + { type: 'TEXT_MESSAGE_START', messageId: M1, role: 'assistant', subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: 'Hi', subagentRunId: SUB }, + ]); +}); + +test('chunk: inner non-text child chunks are ignored', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + for (const inner of [ + { type: 'start', payload: { id: 'weather_forecaster', messageId: 'c-mid' } }, + { type: 'step-start', payload: { messageId: 'c-mid' } }, + { type: 'tool-call-input-streaming-start', payload: { toolName: 'updateWorkingMemory' } }, + { type: 'tool-call', payload: { toolCallId: 'inner-1', toolName: 'updateWorkingMemory', args: {} } }, + { type: 'tool-result', payload: { toolCallId: 'inner-1', toolName: 'updateWorkingMemory', result: {} } }, + { type: 'step-finish', payload: { messageId: 'c-mid' } }, + { type: 'finish', payload: { messageId: 'c-mid' } }, + ]) { + assert.deepEqual(injector.chunk(toolOutput(inner)), [], inner.type); + } + // An inner step-start must NOT move the parent's parentMessageId. + const [start] = injector.chunk(toolCallChunk('call_second')); + assert.equal(start.parentMessageId, PARENT_MID); +}); + +test('chunk: tool-result after deltas → close message + SUBAGENT_FINISHED (no fallback text)', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childTextStart); + injector.chunk(childDelta('Hi')); + // No text-end arrived: the result must close the open message first. + assert.deepEqual(injector.chunk(toolResultChunk(SUCCESS_RESULT)), [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_FINISHED', subagentRunId: SUB, outcome: { type: 'success' } }, + ]); +}); + +test('chunk: tool-result with text-end already seen → SUBAGENT_FINISHED only', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childTextStart); + injector.chunk(childDelta('Hi')); + injector.chunk(childTextEnd); + assert.deepEqual(injector.chunk(toolResultChunk(SUCCESS_RESULT)), [ + { type: 'SUBAGENT_FINISHED', subagentRunId: SUB, outcome: { type: 'success' } }, + ]); +}); + +test('chunk: tool-result with NO deltas observed → single-chunk fallback from result.text', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + assert.deepEqual(injector.chunk(toolResultChunk(SUCCESS_RESULT)), [ + { type: 'TEXT_MESSAGE_START', messageId: M1, role: 'assistant', subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: "Here's the weather", subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_FINISHED', subagentRunId: SUB, outcome: { type: 'success' } }, + ]); +}); + +test('chunk: failed tool-result → close message + SUBAGENT_ERROR', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childDelta('partial')); + assert.deepEqual(injector.chunk(toolResultChunk({ text: 'partial', finishReason: 'error' })), [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_ERROR', subagentRunId: SUB, message: 'partial' }, + ]); + const injector2 = createSubagentInjector(); + injector2.chunk(toolCallChunk()); + assert.deepEqual(injector2.chunk(toolResultChunk({ success: false, error: 'model refused' })), [ + { type: 'SUBAGENT_ERROR', subagentRunId: SUB, message: 'model refused' }, + ]); +}); + +test('chunk: tool-error for a tracked id → close message + SUBAGENT_ERROR', () => { + const injector = createSubagentInjector(); + injector.chunk(toolCallChunk()); + injector.chunk(childDelta('x')); + assert.deepEqual( + injector.chunk({ type: 'tool-error', payload: { toolCallId: TID, toolName: 'agent-weather_forecaster', error: new Error('boom') } }), + [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_ERROR', subagentRunId: SUB, message: 'boom' }, + ], + ); +}); + +test('chunk: tool-call-suspended for a tracked id → close message + SUBAGENT_FINISHED{suspended}', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childDelta('thinking')); + assert.deepEqual( + injector.chunk({ + type: 'tool-call-suspended', + payload: { toolCallId: TID, toolName: 'agent-weather_forecaster', suspendPayload: {}, runId: 'r' }, + }), + [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_FINISHED', subagentRunId: SUB, outcome: { type: 'suspended' } }, + ], + ); + // Suspended is terminal for the card: RUN_FINISHED must not error it. + const finished = { type: 'RUN_FINISHED', threadId: 't', runId: 'r', outcome: { type: 'interrupt' } }; + assert.deepEqual(injector.eventsFor(finished), [finished]); +}); + +test('chunk: non-agent tool-calls and untracked tool-output/tool-result are untouched', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + assert.deepEqual(injector.chunk(toolCallChunk('c1', 'check_conditions', { location: 'x' })), []); + assert.deepEqual(injector.chunk(toolOutput({ type: 'text-delta', payload: { text: 'nope' } }, 'c1')), []); + assert.deepEqual(injector.chunk(toolResultChunk({ forecast: 'clear' }, 'c1')), []); + // Parent-level text chunks are the bridge's job. + assert.deepEqual(injector.chunk({ type: 'text-delta', payload: { text: 'parent' } }), []); + assert.deepEqual(injector.chunk({ type: 'finish', payload: {} }), []); +}); + +test('eventsFor: bridge copies of a synthesized TOOL_CALL_START/ARGS/END are dropped; RESULT passes through', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childTextStart); + injector.chunk(childDelta('Hi')); + injector.chunk(childTextEnd); + injector.chunk(toolResultChunk(SUCCESS_RESULT)); + // The bridge flushes its buffered copy AFTER the tool-result chunk. + assert.deepEqual(injector.eventsFor(delegationStart()), []); + assert.deepEqual(injector.eventsFor({ type: 'TOOL_CALL_ARGS', toolCallId: TID, delta: '{}' }), []); + assert.deepEqual(injector.eventsFor({ type: 'TOOL_CALL_END', toolCallId: TID }), []); + const result = delegationResult(JSON.stringify(SUCCESS_RESULT)); + assert.deepEqual(injector.eventsFor(result), [result], 'no second SUBAGENT_* / TEXT_MESSAGE_* synthesis'); + const finished = { type: 'RUN_FINISHED', threadId: 't', runId: 'r' }; + assert.deepEqual(injector.eventsFor(finished), [finished]); +}); + +test('end-to-end interleave: exact wire order for one streamed delegation', () => { + const injector = createSubagentInjector(); + const out = []; + const chunkThenBridge = (chunk, bridgeEvents = []) => { + out.push(...injector.chunk(chunk)); + for (const e of bridgeEvents) out.push(...injector.eventsFor(e)); + }; + const runStarted = { type: 'RUN_STARTED', threadId: 't', runId: 'r' }; + out.push(...injector.eventsFor(runStarted)); + chunkThenBridge(startChunk); + chunkThenBridge(stepStartChunk); + chunkThenBridge(toolCallChunk()); // bridge buffers; emits nothing + chunkThenBridge(childTextStart); + chunkThenBridge(childDelta('A')); + chunkThenBridge(childDelta('B')); + chunkThenBridge(childTextEnd); + const result = delegationResult(JSON.stringify(SUCCESS_RESULT)); + chunkThenBridge(toolResultChunk(SUCCESS_RESULT), [ + delegationStart(), + { type: 'TOOL_CALL_ARGS', toolCallId: TID, delta: JSON.stringify(ARGS) }, + { type: 'TOOL_CALL_END', toolCallId: TID }, + result, + ]); + const parentText = { type: 'TEXT_MESSAGE_CHUNK', messageId: 'x', delta: 'ok' }; + chunkThenBridge({ type: 'text-delta', payload: { text: 'ok' } }, [parentText]); + const finished = { type: 'RUN_FINISHED', threadId: 't', runId: 'r' }; + chunkThenBridge({ type: 'finish', payload: {} }, [finished]); + + assert.deepEqual(out, [ + runStarted, + ...eagerToolCall, + { type: 'TEXT_MESSAGE_START', messageId: M1, role: 'assistant', subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: 'A', subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_CONTENT', messageId: M1, delta: 'B', subagentRunId: SUB }, + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_FINISHED', subagentRunId: SUB, outcome: { type: 'success' } }, + result, + parentText, + finished, + ]); + assert.equal(out.filter((e) => e.type === 'TOOL_CALL_START').length, 1); +}); + +test('terminal cleanup closes an open child message before SUBAGENT_ERROR, exactly once', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk()); + injector.chunk(childDelta('half')); + const out = injector.eventsFor({ type: 'RUN_ERROR', message: 'boom' }); + assert.deepEqual(out, [ + { type: 'TEXT_MESSAGE_END', messageId: M1, subagentRunId: SUB }, + { type: 'SUBAGENT_ERROR', subagentRunId: SUB, message: 'delegation did not complete before the run terminated' }, + { type: 'RUN_ERROR', message: 'boom' }, + ]); + assert.deepEqual(injector.eventsFor({ type: 'RUN_FINISHED' }), [{ type: 'RUN_FINISHED' }]); +}); + +test('two sequential delegations get distinct message ids and independent state', () => { + const injector = createSubagentInjector(); + injector.chunk(stepStartChunk); + injector.chunk(toolCallChunk('call_a')); + injector.chunk(childDelta('a1')); + injector.chunk(toolResultChunk(SUCCESS_RESULT, 'call_a')); + injector.chunk(toolCallChunk('call_b')); + const out = injector.chunk(toolOutput({ type: 'text-delta', payload: { text: 'b1' } }, 'call_b')); + assert.deepEqual(out, [ + { type: 'TEXT_MESSAGE_START', messageId: 'call_b-sub-m1', role: 'assistant', subagentRunId: 'call_b-sub' }, + { type: 'TEXT_MESSAGE_CONTENT', messageId: 'call_b-sub-m1', delta: 'b1', subagentRunId: 'call_b-sub' }, + ]); +}); + test('pending delegation at RUN_ERROR → SUBAGENT_ERROR cleanup, then no double-cleanup', () => { const injector = createSubagentInjector(); const out = [ @@ -155,3 +457,10 @@ test('pending delegation at RUN_ERROR → SUBAGENT_ERROR cleanup, then no double // A second terminal frame injects nothing more. assert.deepEqual(injector.eventsFor({ type: 'RUN_FINISHED' }), [{ type: 'RUN_FINISHED' }]); }); + +test('chunk: agent-* tool-call with NO prior start/step-start omits parentMessageId entirely', () => { + const injector = createSubagentInjector(); + const [start] = injector.chunk(toolCallChunk()); + assert.equal(start.type, 'TOOL_CALL_START'); + assert.equal('parentMessageId' in start, false, 'key must be absent, not present with an undefined value'); +}); diff --git a/docs/superpowers/plans/2026-09-02-mastra-subagent-streaming.md b/docs/superpowers/plans/2026-09-02-mastra-subagent-streaming.md new file mode 100644 index 000000000..6e9d81bbe --- /dev/null +++ b/docs/superpowers/plans/2026-09-02-mastra-subagent-streaming.md @@ -0,0 +1,51 @@ +# Mastra Sub-agent Streaming Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** The Mastra demo streams the sub-agent's child deltas to the AG-UI wire (with an eager `TOOL_CALL_START`), flipping the matrix cell from Partial to Yes. + +**Architecture:** See `docs/superpowers/specs/2026-09-02-mastra-subagent-streaming-design.md`. A public-API stream tee (`Proxy` over the Mastra `Agent`, wrapping `stream()`/`resumeStream()`'s `fullStream`) observes each chunk before the unmodified bridge consumes it; the per-run injector maps `tool-call`/`tool-output`/`tool-result`/`tool-call-suspended` chunks to synthesized eager `TOOL_CALL_*` + `SUBAGENT_*` + attributed `TEXT_MESSAGE_*` events and dedupes the bridge's later buffered copies. + +**Tech Stack:** Node 20+, `@mastra/core@1.63.2`, `@ag-ui/mastra@1.1.2` (unchanged pins), `node --test`, Playwright + aimock replay. + +**Branch:** `blove/mastra-subagent-streaming` (off origin/main; spec + plan committed). + +**Files:** `deployments/ag-ui-mastra/{streaming-tee.mjs (new), subagent-emitter.mjs, server.mjs, test/streaming-tee.test.mjs (new), test/subagent-emitter.test.mjs}`; `cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md`; docs/blog pages listed in Task 5. + +--- + +### Task 0: Spike (no production code) + +- [ ] Read `deployments/ag-ui-mastra/server.mjs`, `agents.mjs`, `subagent-emitter.mjs`, `test/subagent-emitter.test.mjs`, and the wire-capture doc; read `node_modules/@ag-ui/mastra/dist/*.mjs` around `isLocalMastraAgent`, `streamMastraAgent` (`this.agent.stream(A,s)` → `processFullStream(m.fullStream,…)`), and `resumeStream` usage, to list every agent member the bridge touches. +- [ ] Card-mount check: read `libs/chat/src/lib/primitives/chat-tool-calls/chat-tool-calls.component.ts` — confirm cards are anchored to entries in the parent `toolCalls()` list (i.e. a `SUBAGENT_STARTED` with no prior `TOOL_CALL_START` renders no card). Record the finding. +- [ ] Raw `fullStream` capture: export the key silently (`export OPENAI_API_KEY=$(grep '^OPENAI_API_KEY=' /Users/blove/repos/angular-agent-framework/.env | cut -d= -f2-)`), write a scratch script in `deployments/ag-ui-mastra/` (uncommitted) that calls the camping supervisor's `stream()` with the delegation prompt and logs every chunk `{type, payload.toolCallId?, payload.toolName?, payload.output?.type}`; note the order of `tool-call` → `tool-output(start/text-start/text-delta×N/text-end/…)` → `tool-result`, the `start`/`step-start` messageId available for `parentMessageId`, and whether `tool-call-suspended` appears. +- [ ] Proxy feasibility test (this one IS committed): `test/streaming-tee.test.mjs` — construct a real `Agent` from `agents.mjs`'s exports (no model call), wrap with a minimal Proxy prototype, assert `'getMemory' in proxy`, `typeof proxy.listTools === 'function'`, and that calling a stubbed `stream()` via the proxy returns an object whose `fullStream` is the wrapped generator. Run `node --test test/streaming-tee.test.mjs` → this fails until Task 1 provides the module; keep it as the failing test. +- [ ] Append "## Streaming spike" to `cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md` (chunk order, parentMessageId source, card-mount finding, suspended presence). Commit `docs(runtimes): mastra streaming spike — raw fullStream order and card-mount finding` + `Co-Authored-By: Claude Fable 5.1 `. + +### Task 1: `streaming-tee.mjs` (TDD) + +- [ ] Tests (`test/streaming-tee.test.mjs`): with a fake agent `{ getMemory(){}, listTools(){}, async stream(){ return { fullStream: (async function*(){ yield {type:'a'}; yield {type:'b'}; })(), text: 'x' } } }`: (1) observer sees `a` then `b`, each BEFORE the consumer receives it (record an interleaved trace `obs:a, out:a, obs:b, out:b`); (2) non-`fullStream` properties of the stream result are preserved (`text === 'x'`); (3) `resumeStream` wrapped the same way; (4) other members forward with `this` bound to the real agent (a method reading a `#private` field on a real `Agent` subclass instance still works through the proxy); (5) an observer exception does not break the consumer (logged, chunk still yielded). +- [ ] Implement `withDelegationTee(agent, observe)` per the spec. Commit `feat(runtimes): mastra stream tee — observe fullStream chunks before the bridge consumes them`. + +### Task 2: Injector chunk mapping + dedupe (TDD) + +- [ ] Tests (`test/subagent-emitter.test.mjs`, extend): feed `chunk()` sequences copied from the spike capture and assert exact events: eager `TOOL_CALL_START{toolCallId,toolCallName,parentMessageId}`/`ARGS{delta: JSON}`/`END` + `SUBAGENT_STARTED` on the `agent-*` `tool-call`; attributed `TEXT_MESSAGE_START/CONTENT×N/END` from `tool-output` inner text chunks (lazy START when `text-start` is absent); `SUBAGENT_FINISHED` on `tool-result` (and `SUBAGENT_ERROR` on the failure shape); `eventsFor` DROPS the bridge's later `TOOL_CALL_START/ARGS/END` for a synthesized id but passes `TOOL_CALL_RESULT`; the old single-chunk synthesis fires ONLY when no deltas were observed; `tool-call-suspended` → close message + `SUBAGENT_FINISHED{outcome:{type:'suspended'}}`; non-`agent-` tool-calls untouched; terminal cleanup closes open messages then `SUBAGENT_ERROR` for pending ids exactly once. +- [ ] Implement in `subagent-emitter.mjs` (pure per-run state: `pending` map with `{ name, messageOpen, messageId, deltasSeen, synthesized }`). Commit `feat(runtimes): mastra injector maps delegation chunks to eager TOOL_CALL_* and attributed child deltas`. + +### Task 3: `server.mjs` wiring + e2e + +- [ ] Per request: `const injector = createSubagentInjector(); const observe = (c) => { for (const e of injector.chunk(c)) write(e); }; const bridge = new MastraAgent({ ..., agent: withDelegationTee(agent, observe) });` keeping the existing `for (const e of injector.eventsFor(event)) write(e)` for bridge events. Make sure `write` is the same SSE frame writer for both paths. +- [ ] `npm test` (lane) green; free ports (cockpit/ports.mjs → cockpit-runtimes-mastra-angular), `npx playwright test --config cockpit/runtimes/mastra/angular/e2e/playwright.config.ts` → 5/5 (assertions unchanged). Commit `feat(runtimes): mastra server wires the delegation tee into the SSE stream`. + +### Task 4: Live verification + +- [ ] Real-key server + `npx nx serve cockpit-runtimes-mastra-angular`; POST the delegation prompt and tee the SSE: assert eager `TOOL_CALL_START` + `SUBAGENT_STARTED` precede the first attributed delta; count `TEXT_MESSAGE_CONTENT` with `subagentRunId`; confirm no duplicate `TOOL_CALL_START` for the delegation id. Browser: poll the card's innerText ~150 ms; record growth while `running`; screenshot over `cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png`. Append "## After streaming" + updated "## Browser verification" to the wire-capture doc; note the suspended caveat and the subclass alternative as fallback design. Kill servers. Commit `docs(runtimes): mastra streaming live verification`. + +### Task 5: Docs — flip the cell + +- [ ] `apps/website/content/docs/choosing-an-adapter/index.mdx`: Mastra Subagents cell → `Yes`; cause cell → the emitter (public-API stream tee) note; the "one Partial cell" sentences in the subagents section rewritten (all three stream live; note Mastra's card streams via the tee because the bridge itself drops child output). `runtimes/getting-started/introduction.mdx` matrix + the "lifecycle-plus-final-text on Mastra" clause; `runtimes/mastra/overview.mdx` Surface row + "How subagents surface" section; `runtimes/mastra/how-it-connects.mdx` sentence pair. Blog: `2026-08-31-we-measured-the-runtime-swap.mdx` matrix cell + the "Two of the three cards stream live. Mastra's fills in at completion…" passage + conclusion; `2026-08-31-what-changes-when-the-runtime-changes.mdx` if it names the Partial. No contractions; one sentence per line. `npx nx test website` green. Commit `docs(website,blog): mastra subagents stream — cell flips to Yes`. + +### Task 6: PR + follow-ups + +- [ ] Push; open PR `feat(runtimes): mastra sub-agent streaming via a public-API stream tee` (body: research verdict summary, evidence links, tallies, the suspended caveat; end with `🤖 Generated with [Claude Code](https://claude.com/claude-code)`). Do NOT arm auto-merge (the coordinator reviews first). +- [ ] Follow-up chip (coordinator): upstream `tool-output` mapping patch against `ag-ui-protocol/ag-ui` `integrations/mastra/typescript/src/mastra.ts`, referencing #2402/#2403. diff --git a/docs/superpowers/specs/2026-09-02-mastra-subagent-streaming-design.md b/docs/superpowers/specs/2026-09-02-mastra-subagent-streaming-design.md new file mode 100644 index 000000000..3d7e9f273 --- /dev/null +++ b/docs/superpowers/specs/2026-09-02-mastra-subagent-streaming-design.md @@ -0,0 +1,86 @@ +# Mastra sub-agent streaming: forward child deltas via a public-API stream tee + +**Date:** 2026-09-02 +**Status:** Approved (Brian, after deep research validated the layer and revised the seam) + +## Problem + +The Mastra runtime's subagent cell is Partial: Mastra streams the sub-agent's chunks +in-process as parent `tool-output` chunks (`{type:'tool-output', payload:{output: +, toolCallId, toolName}}` — a public, typed shape), but `@ag-ui/mastra` +1.1.2 drops them (`case "tool-output": break`) and buffers the delegation's +`TOOL_CALL_START/ARGS/END` until the tool result. The card therefore mounts already +complete after a ~5 s silent gap. + +## Research verdicts (evidence in the session's research report) + +- Pin not stale: 1.1.2 is the latest `@ag-ui/mastra`; upstream `main` still drops + `tool-output`; the only related upstream work is open PR #2403 (opt-in eager + `TOOL_CALL_START` for server tools), which also documents that buffering is + load-bearing for suspend retraction. +- No alternative layer: `@mastra/server` has no AG-UI route; Mastra's documented server + integration builds the same bridge inside a runtime we cannot depend on. `server.mjs` + is the canonical pattern and adds things upstream lacks. +- Mastra's official `onChunk` hook excludes `tool-output` in its compiled allow-list. +- A bridge subclass works today but couples to three TS-private methods, a `clone()` + that constructs the base class (drops overrides), and signature drift on `main`. + +## Design + +1. **Stream tee on a proxied agent** — `deployments/ag-ui-mastra/streaming-tee.mjs` + exports `withDelegationTee(agent, observe)`: a `Proxy` over the real Mastra `Agent` + that forwards every member bound to the real instance (so `#private` fields and the + bridge's `'getMemory' in agent` check keep working) except `stream()` and + `resumeStream()`, whose results are returned with `fullStream` replaced by an async + generator that calls `observe(chunk)` **before** yielding each chunk to the bridge. + One reader, one iteration → the observer's SSE writes are strictly ordered ahead of + the bridge's processing of the same chunk. The unmodified `MastraAgent` receives the + proxy. +2. **Injector grows a chunk input** — `subagent-emitter.mjs`'s per-run injector gains + `chunk(c)` returning AG-UI events to write, alongside the existing `eventsFor(event)`: + - `tool-call` whose `toolName` starts with `agent-` → synthesize eager + `TOOL_CALL_START/ARGS/END` (args are complete in the chunk; `parentMessageId` = the + last `start`/`step-start` chunk's message id) + `SUBAGENT_STARTED {subagentRunId: + -sub, name, parentToolCallId}`. Record the id so the bridge's later + buffered `TOOL_CALL_START/ARGS/END` for the same id are dropped in `eventsFor` + (its `TOOL_CALL_RESULT` still passes through). + - `tool-output` for a tracked id: inner `text-start` → attributed `TEXT_MESSAGE_START` + (lazy: open on first delta if no explicit start), `text-delta` → + `TEXT_MESSAGE_CONTENT`, `text-end` → `TEXT_MESSAGE_END`. Inner tool chunks are + out of scope for this PR (ignored). + - `tool-result` for a tracked id → close any open message, `SUBAGENT_FINISHED` + (success) or `SUBAGENT_ERROR` per the existing result check; mark deltas-seen. + - `tool-call-suspended` for a tracked id → close any open message, + `SUBAGENT_FINISHED {outcome:{type:'suspended'}}` (the eager START has already been + painted — accepted, documented caveat; the demo's delegation does not suspend). + - The existing `TOOL_CALL_RESULT` synthesis of a single-chunk message stays ONLY as + the fallback when no deltas were observed for that id. + - Terminal cleanup (RUN_FINISHED/RUN_ERROR) unchanged, extended to close open + messages. +3. **server.mjs wiring** — build the proxy per request with an observer that runs + `injector.chunk(c)` and writes each returned event as an SSE frame through the same + `res.write` path; the bridge's events keep flowing through `injector.eventsFor`. +4. **Spike first** (the two unverified items): (a) a unit test that a Proxy-wrapped + real `Agent` still satisfies the bridge (`'getMemory' in proxy`, `listTools`, + `stream()` returning a wrapped `fullStream`); (b) confirmation from + `libs/chat` that a subagent card needs the parent tool call present (it does — + `chat-tool-calls` anchors cards to tool-call entries), which is why synthesis is + required rather than optional; (c) a raw `fullStream` capture of one delegation + (chunk types + order, presence of `text-start/end`, whether `tool-call-suspended` + ever appears here). + +## Verification gates + +Wire capture showing eager `TOOL_CALL_START` + `SUBAGENT_STARTED` before the child +deltas and N attributed `TEXT_MESSAGE_CONTENT` events; browser check with the card's +text growing while `running` (char-growth samples + screenshot); e2e 5/5 in replay +(assertions unchanged); lane unit tests for the tee, the chunk mapping, dedupe, the +no-delta fallback, error, suspended, and terminal cleanup. Then the docs cell flips: +`choosing-an-adapter` matrix + section, runtimes intro + Mastra overview/how-it-connects, +both runtime blog posts ("one Partial cell" sentences), and the wire-capture doc. + +## Out of scope + +Inner sub-agent tool calls as attributed `TOOL_CALL_*`; the upstream `tool-output` +mapping patch (filed as a follow-up referencing #2403 for the eager-START half); the +bridge-subclass alternative (documented in the wire-capture doc as the fallback design).