Skip to content

Commit c60f525

Browse files
bloveclaude
blove
and
claude
authored
feat(runtimes): mastra subagent delegation demo with SUBAGENT_* injection in the bridge (#958)
* docs(runtimes): mastra delegation wire capture and hook findings Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(runtimes): mastra camping demo delegates forecasting; SUBAGENT_* injection in the bridge Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(runtimes): mastra post-emitter wire capture Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(runtimes): mastra delegation e2e — subagent card via fixture replay Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(runtimes): mastra live browser verification Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(runtimes): retake mastra live subagent-card screenshot with completed expanded body Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 5e581fa commit c60f525

9 files changed

Lines changed: 603 additions & 7 deletions

File tree

‎cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md‎

Lines changed: 246 additions & 0 deletions
Large diffs are not rendered by default.

‎cockpit/runtimes/mastra/angular/e2e/fixtures/mastra.json‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,18 @@
1818
"content": "North Pines is reserved for 2 nights — confirmation TP-0288."
1919
}
2020
},
21+
{
22+
"match": { "userMessage": "Bear Lake", "hasToolResult": true },
23+
"response": {
24+
"content": "Expect a sunny, mild weekend at Bear Lake — great weather for camping."
25+
}
26+
},
27+
{
28+
"match": { "systemMessage": "You are a weather forecaster" },
29+
"response": {
30+
"content": "Here's the Bear Lake weekend forecast:\n\n- Saturday: sunny, high 24°C\n- Sunday: partly cloudy, high 22°C\n- Overall: mild with light winds"
31+
}
32+
},
2133
{
2234
"match": { "userMessage": "packing list" },
2335
"response": {
@@ -60,6 +72,25 @@
6072
}
6173
]
6274
}
75+
},
76+
{
77+
"match": { "userMessage": "Bear Lake" },
78+
"response": {
79+
"toolCalls": [
80+
{
81+
"name": "agent-weather_forecaster",
82+
"arguments": {
83+
"prompt": "What will the weather be like at Bear Lake this weekend?",
84+
"threadId": null,
85+
"resourceId": null,
86+
"instructions": null,
87+
"maxSteps": 5,
88+
"suspendedToolRunId": null,
89+
"resumeData": null
90+
}
91+
}
92+
]
93+
}
6394
}
6495
]
6596
}

‎cockpit/runtimes/mastra/angular/e2e/manual/subagent-card-live.png‎

76 KB
Loading

‎cockpit/runtimes/mastra/angular/e2e/mastra.spec.ts‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
// SPDX-License-Identifier: MIT
22
import { test, expect } from '@playwright/test';
3+
import { submitAndWaitForResponse } from '@threadplane-internal/e2e-harness';
34

45
// First cockpit e2e whose backend is neither LangGraph nor Python: the
56
// rt-mastra topic runs against the deployments/ag-ui-mastra Node service
@@ -43,4 +44,16 @@ test.describe('cockpit runtimes/mastra: camping trip planner', () => {
4344
await dialog.getByRole('button', { name: 'Approve' }).click();
4445
await expect(page.getByText(/reserved for 2 nights/i)).toBeVisible({ timeout: 30_000 });
4546
});
47+
48+
// Delegation: the supervisor calls the registered weather_forecaster
49+
// sub-agent (wire tool `agent-weather_forecaster`); the server-side
50+
// emitter (deployments/ag-ui-mastra/subagent-emitter.mjs) injects
51+
// SUBAGENT_STARTED + attributed TEXT_MESSAGE_* + SUBAGENT_FINISHED, which
52+
// the adapter reduces into a subagent card on the tool-call group.
53+
test('rt-mastra: delegated forecast renders a subagent card with the final text', async ({ page }) => {
54+
const bubble = await submitAndWaitForResponse(page, 'Plan a trip to Bear Lake this weekend — what will the weather be?');
55+
await expect(page.locator('chat-subagent-card')).toHaveCount(1);
56+
await expect(page.locator('chat-subagent-card')).toContainText('weather_forecaster');
57+
await expect(bubble).toContainText(/forecast|weather/i);
58+
});
4659
});

‎cockpit/runtimes/mastra/angular/src/app/mastra.component.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,8 +56,11 @@ interface PackingList {
5656
* `submit({ resume })` goes out as
5757
* `forwardedProps.command = { resume, interruptEvent: { toolCallId, runId } }`
5858
* — exactly what the Mastra bridge requires to resume the suspended run.
59-
* - NO subagents surface here: Mastra reserves ACTIVITY_* for background
60-
* tasks, a measured red cell in the matrix.
59+
* - Sub-agent delegation (weather_forecaster) reaches the wire as a
60+
* `agent-<key>` tool call; the Node service's subagent emitter turns it
61+
* into SUBAGENT_* + attributed TEXT_MESSAGE_* frames, so the standard
62+
* chat-subagent-card renders with zero code in this component
63+
* (docs/wire-capture-subagents.md).
6164
*/
6265
@Component({
6366
selector: 'app-mastra',

‎deployments/ag-ui-mastra/agents.mjs‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,19 +96,37 @@ const reserveCampsiteTool = createTool({
9696
export function createMastra(dbUrl) {
9797
const store = (id) => new LibSQLStore({ id, url: dbUrl });
9898

99+
/**
100+
* Sub-agent (spike: wire-capture-subagents.md). Registered on the
101+
* supervisor via `agents:`; Mastra surfaces it as a backend tool named
102+
* `agent-weather_forecaster` whose TOOL_CALL_RESULT carries the child's
103+
* final text — server.mjs's subagent emitter turns that into SUBAGENT_*
104+
* frames. The `description` becomes the delegation tool's description.
105+
*/
106+
const weatherForecaster = new Agent({
107+
id: 'weather_forecaster',
108+
name: 'weather_forecaster',
109+
description: 'Forecasts weather for a campsite and date range. Use for any weather question.',
110+
instructions:
111+
'You are a weather forecaster. Given a campsite and dates, give a 3-bullet forecast summary. Be concise.',
112+
model: MODEL,
113+
});
114+
99115
const tripAgent = new Agent({
100116
id: 'mastra',
101117
name: 'mastra',
102118
instructions: `You are a terse camping trip planner.
103119
The packing list in working memory is the user's shared state: whenever the user adds, removes, or changes items (or starts a list), update working memory to match. 'items' is an array of {name, qty}. Never mention memory or the list mechanics.
104-
For questions about weather or trail conditions you MUST call check_conditions.
120+
For questions about trail conditions you MUST call check_conditions.
121+
For questions about weather forecasts you MUST delegate to the weather_forecaster agent.
105122
When the user asks to reserve or book a campsite you MUST call reserve_campsite; after it resumes, confirm the outcome.
106123
Always answer in one short sentence.`,
107124
model: MODEL,
108125
tools: {
109126
check_conditions: checkConditionsTool,
110127
reserve_campsite: reserveCampsiteTool,
111128
},
129+
agents: { weather_forecaster: weatherForecaster },
112130
memory: new Memory({
113131
storage: store('mastra-topic-memory'),
114132
options: {

‎deployments/ag-ui-mastra/server.mjs‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import { pathToFileURL } from 'node:url';
2222
import { dirname, resolve } from 'node:path';
2323
import { MastraAgent } from '@ag-ui/mastra';
2424
import { createMastra } from './agents.mjs';
25+
import { createSubagentInjector } from './subagent-emitter.mjs';
2526

2627
const AG_UI_INTERNAL_TOKEN = process.env.AG_UI_INTERNAL_TOKEN;
2728
if (!AG_UI_INTERNAL_TOKEN) {
@@ -108,16 +109,18 @@ export function createAgUiServer() {
108109
resourceId: input.threadId,
109110
});
110111

112+
// One injector per run: turns delegation tool calls (`agent-<childKey>`)
113+
// into SUBAGENT_* frames around the events the bridge already emits.
114+
const injector = createSubagentInjector();
111115
const sub = bridge.run(input).subscribe({
112116
next: (event) => {
113-
res.write(sseFrame(event));
117+
for (const e of injector.eventsFor(event)) res.write(sseFrame(e));
114118
},
115119
error: (err) => {
116120
// Map failures into the protocol instead of killing the socket:
117121
// the client finalizes the run as an error rather than hanging.
118-
res.write(
119-
sseFrame({ type: 'RUN_ERROR', message: String(err?.message ?? err) }),
120-
);
122+
const runError = { type: 'RUN_ERROR', message: String(err?.message ?? err) };
123+
for (const e of injector.eventsFor(runError)) res.write(sseFrame(e));
121124
res.end();
122125
},
123126
complete: () => {

‎deployments/ag-ui-mastra/subagent-emitter.mjs‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
// SPDX-License-Identifier: MIT
2+
// SUBAGENT_* injection for Mastra delegation tool calls.
3+
//
4+
// Mastra surfaces a registered sub-agent as an ordinary backend tool named
5+
// `agent-<childKey>`: TOOL_CALL_START → TOOL_CALL_ARGS → TOOL_CALL_END →
6+
// TOOL_CALL_RESULT whose `content` is JSON `{text, subAgentThreadId, ...}`
7+
// (measured: cockpit/runtimes/mastra/angular/docs/wire-capture-subagents.md).
8+
// The upstream @ag-ui/mastra bridge drops the in-process child deltas
9+
// (`case "tool-output": break`), so the honest wire contract here is a
10+
// single final text chunk per delegation.
11+
//
12+
// This module is a pure transform over the outbound AG-UI event stream —
13+
// keyed off the events themselves, NOT the Mastra delegation hooks, which
14+
// fire in a different async context with no ordering guarantee relative to
15+
// the Observable frames.
16+
//
17+
// Injected sequence per delegation tool call <tid> (child key = tool name
18+
// minus the `agent-` prefix, subagentRunId = `<tid>-sub`):
19+
// - AFTER TOOL_CALL_START: SUBAGENT_STARTED {subagentRunId, name,
20+
// parentToolCallId}
21+
// - BEFORE TOOL_CALL_RESULT (success): TEXT_MESSAGE_START/CONTENT/END
22+
// carrying the child's final text under the subagent identity, then
23+
// SUBAGENT_FINISHED {outcome:{type:'success'}}
24+
// - BEFORE TOOL_CALL_RESULT (failure — parsed content says success:false
25+
// or finishReason:'error'): SUBAGENT_ERROR {subagentRunId, message}
26+
// - Terminal cleanup: a RUN_ERROR or RUN_FINISHED arriving while
27+
// delegations are still pending (no TOOL_CALL_RESULT seen — e.g. the
28+
// Observable errored mid-delegation) closes each pending card with
29+
// SUBAGENT_ERROR before the terminal frame, so no card is left spinning.
30+
// In the measured captures the RESULT always precedes the terminal frame,
31+
// so this path is defensive only.
32+
33+
const AGENT_TOOL_PREFIX = 'agent-';
34+
35+
/**
36+
* Create a per-run injector.
37+
*
38+
* @returns {{ eventsFor(event: object): object[] }} — for each outbound
39+
* AG-UI event, the ordered list of frames to write (injections plus the
40+
* original event). Non-delegation events pass through as `[event]`.
41+
*/
42+
export function createSubagentInjector() {
43+
/** @type {Map<string, {subagentRunId: string, name: string}>} pending delegations by toolCallId */
44+
const pending = new Map();
45+
46+
return {
47+
eventsFor(event) {
48+
switch (event.type) {
49+
case 'TOOL_CALL_START': {
50+
const name = event.toolCallName ?? '';
51+
if (!name.startsWith(AGENT_TOOL_PREFIX)) return [event];
52+
const entry = {
53+
subagentRunId: `${event.toolCallId}-sub`,
54+
name: name.slice(AGENT_TOOL_PREFIX.length),
55+
};
56+
pending.set(event.toolCallId, entry);
57+
return [
58+
event,
59+
{
60+
type: 'SUBAGENT_STARTED',
61+
subagentRunId: entry.subagentRunId,
62+
name: entry.name,
63+
parentToolCallId: event.toolCallId,
64+
},
65+
];
66+
}
67+
68+
case 'TOOL_CALL_RESULT': {
69+
const entry = pending.get(event.toolCallId);
70+
if (!entry) return [event]; // not a delegation (or unmatched) — pass through
71+
pending.delete(event.toolCallId);
72+
const { subagentRunId } = entry;
73+
74+
const raw = typeof event.content === 'string' ? event.content : JSON.stringify(event.content);
75+
let parsed;
76+
try {
77+
parsed = JSON.parse(raw);
78+
} catch {
79+
parsed = undefined;
80+
}
81+
const failed =
82+
parsed !== undefined &&
83+
typeof parsed === 'object' &&
84+
parsed !== null &&
85+
(parsed.success === false || parsed.finishReason === 'error');
86+
if (failed) {
87+
return [
88+
{
89+
type: 'SUBAGENT_ERROR',
90+
subagentRunId,
91+
message: String(parsed.error ?? parsed.text ?? 'sub-agent delegation failed'),
92+
},
93+
event,
94+
];
95+
}
96+
97+
const text = typeof parsed?.text === 'string' ? parsed.text : raw;
98+
const messageId = `${event.toolCallId}-sub-m1`;
99+
return [
100+
{ type: 'TEXT_MESSAGE_START', messageId, role: 'assistant', subagentRunId },
101+
{ type: 'TEXT_MESSAGE_CONTENT', messageId, delta: text, subagentRunId },
102+
{ type: 'TEXT_MESSAGE_END', messageId, subagentRunId },
103+
{ type: 'SUBAGENT_FINISHED', subagentRunId, outcome: { type: 'success' } },
104+
event,
105+
];
106+
}
107+
108+
case 'RUN_ERROR':
109+
case 'RUN_FINISHED': {
110+
if (pending.size === 0) return [event];
111+
const cleanup = [...pending.values()].map(({ subagentRunId }) => ({
112+
type: 'SUBAGENT_ERROR',
113+
subagentRunId,
114+
message: 'delegation did not complete before the run terminated',
115+
}));
116+
pending.clear();
117+
return [...cleanup, event];
118+
}
119+
120+
default:
121+
return [event];
122+
}
123+
},
124+
};
125+
}

0 commit comments

Comments
 (0)