From ede3c0583f0e74519990b4185001ad3c5303feec Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 22:56:50 -0400 Subject: [PATCH] feat(pi): capture subagent activity events --- .../pi/extensions/subagents/runner.test.ts | 49 ++++++++ .../agents/pi/extensions/subagents/runner.ts | 5 +- .../agents/pi/extensions/subagents/status.ts | 7 +- .../extensions/subagents/supervisor.test.ts | 67 +++++++++++ .../pi/extensions/subagents/supervisor.ts | 110 +++++++++++++++++- .../agents/pi/extensions/subagents/types.ts | 26 ++++- .../agents/pi/extensions/subagents/ui.test.ts | 21 ++++ modules/agents/pi/extensions/subagents/ui.ts | 1 + 8 files changed, 279 insertions(+), 7 deletions(-) diff --git a/modules/agents/pi/extensions/subagents/runner.test.ts b/modules/agents/pi/extensions/subagents/runner.test.ts index 4a13680..55c54e4 100644 --- a/modules/agents/pi/extensions/subagents/runner.test.ts +++ b/modules/agents/pi/extensions/subagents/runner.test.ts @@ -26,6 +26,55 @@ function events(): RunnerEvents { }; } +test("child RPC process forwards structured activity before collecting the final result", async (t) => { + const running: unknown[] = []; + const completed: Array<{ result: string; stopReason?: string }> = []; + const fakeChild = new EventEmitter() as EventEmitter & { + stdout: FakeStream; + stderr: FakeStream; + stdin: FakeStream; + killed: boolean; + pid?: number; + kill(signal?: NodeJS.Signals): boolean; + }; + fakeChild.stdout = new FakeStream(); + fakeChild.stderr = new FakeStream(); + fakeChild.stdin = new FakeStream(); + fakeChild.killed = false; + fakeChild.kill = () => { + fakeChild.killed = true; + return true; + }; + t.mock.method(fakeChild.stdin, "write", (chunk, callback?: (error?: Error | null) => void) => { + const request = JSON.parse(String(chunk)) as { id: string; type: string }; + callback?.(); + if (request.type === "get_last_assistant_text") { + queueMicrotask(() => { + fakeChild.stdout.emit("data", `${JSON.stringify({ id: request.id, type: "response", success: true, data: { text: "final answer" } })}\n`); + }); + } + return true; + }); + t.mock.method(childProcess, "spawn", () => fakeChild as unknown as childProcess.ChildProcessWithoutNullStreams); + + const { SubprocessRpcRunner } = await import("./runner.ts"); + const runner = new SubprocessRpcRunner(); + await runner.start("child-1", { prompt: "work", label: "Review migration" }, "/tmp", { + ...events(), + running: (event) => running.push(event), + completed: (result, stopReason) => completed.push({ result, stopReason }), + }); + + const firstActivity = { type: "message_start", role: "assistant", message: { id: "msg-1" } }; + const secondActivity = { type: "tool_execution_start", tool: "read", input: { path: "runner.ts" } }; + const settledActivity = { type: "agent_settled" }; + fakeChild.stdout.emit("data", `${JSON.stringify(firstActivity)}\n${JSON.stringify(secondActivity)}\n${JSON.stringify(settledActivity)}\n`); + await new Promise((resolve) => setImmediate(resolve)); + + assert.deepEqual(running, [firstActivity, secondActivity, settledActivity]); + assert.deepEqual(completed, [{ result: "final answer", stopReason: "agent_settled" }]); +}); + test("child RPC process disables discovery while explicitly loading subagents extension", async (t) => { const calls: Array<{ command: string; args: string[] }> = []; const fakeChild = new EventEmitter() as EventEmitter & { diff --git a/modules/agents/pi/extensions/subagents/runner.ts b/modules/agents/pi/extensions/subagents/runner.ts index 5291f0f..c0db926 100644 --- a/modules/agents/pi/extensions/subagents/runner.ts +++ b/modules/agents/pi/extensions/subagents/runner.ts @@ -89,16 +89,17 @@ class RpcChildHandle implements ChildHandle { } if (payload.type === "agent_started") { - this.events.running("agent_started"); + this.events.running(payload as Record); return; } if (payload.type === "agent_settled") { + this.events.running(payload as Record); this.finish().catch((error) => this.fail(error instanceof Error ? error.message : String(error))); return; } - if (payload.type) this.events.running(payload.type); + if (payload.type) this.events.running(payload as Record); } private async finish() { diff --git a/modules/agents/pi/extensions/subagents/status.ts b/modules/agents/pi/extensions/subagents/status.ts index 1831b4e..68f8481 100644 --- a/modules/agents/pi/extensions/subagents/status.ts +++ b/modules/agents/pi/extensions/subagents/status.ts @@ -13,7 +13,12 @@ export function toAccepted(status: SubagentStatus): SpawnAccepted { } export function cloneStatus(status: SubagentStatus): SubagentStatus { - return { ...status, elapsedMs: elapsedMs(status) }; + return { + ...status, + currentActivity: status.currentActivity ? { ...status.currentActivity } : undefined, + activityHistory: status.activityHistory.map((event) => ({ ...event })), + elapsedMs: elapsedMs(status), + }; } export function cloneResult(record: ChildRecord): SubagentResult { diff --git a/modules/agents/pi/extensions/subagents/supervisor.test.ts b/modules/agents/pi/extensions/subagents/supervisor.test.ts index 7968dfd..f1503ba 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.test.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.test.ts @@ -73,6 +73,73 @@ test("runtime timeout reaches timed_out", async () => { assert.equal(runner.starts[0].handle.cancelCalls, 1); }); +test("activity exposes ordered transcript events while status and list keep only summaries", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor); + + runner.starts[0].events.running({ type: "message_started", role: "assistant" }); + runner.starts[0].events.running({ + type: "message_delta", + role: "assistant", + assistantMessageEvent: { type: "content_delta", delta: "private transcript body" }, + }); + runner.starts[0].events.running({ type: "tool_started", tool: "read", input: { path: "secret-notes.md" } }); + runner.starts[0].events.running({ type: "tool_completed", tool: "read", output: "secret file contents" }); + + type ActivityStatus = ReturnType & { + activityHistory: Array<{ type: string; summary: string }>; + currentActivity: { summary: string }; + }; + const activity = supervisor.activity(accepted.id); + const status = supervisor.status(accepted.id) as ActivityStatus; + const listed = supervisor.list().find((item) => item.id === accepted.id) as ActivityStatus | undefined; + + assert.deepEqual( + activity.map((event) => event.type), + ["queued", "starting", "prompt accepted", "message_started", "message_delta", "tool_started", "tool_completed"], + ); + assert.deepEqual(activity[4], { + type: "message_delta", + summary: "assistant message content_delta", + at: activity[4].at, + role: "assistant", + tool: undefined, + phase: "content_delta", + text: "private transcript body", + input: undefined, + output: undefined, + error: undefined, + payload: { + type: "message_delta", + role: "assistant", + assistantMessageEvent: { type: "content_delta", delta: "private transcript body" }, + }, + }); + assert.deepEqual(activity[5], { + type: "tool_started", + summary: "read secret-notes.md", + at: activity[5].at, + role: undefined, + tool: "read", + phase: "started", + text: undefined, + input: { path: "secret-notes.md" }, + output: undefined, + error: undefined, + payload: { type: "tool_started", tool: "read", input: { path: "secret-notes.md" } }, + }); + assert.equal(activity[6].output, "secret file contents"); + + assert.ok(Array.isArray(status.activityHistory), "status should expose structured activityHistory"); + assert.deepEqual(status.activityHistory.map((event) => event.type), activity.map((event) => event.type)); + assert.deepEqual(status.activityHistory.map((event) => event.summary), activity.map((event) => event.summary)); + assert.equal(status.currentActivity.summary, "read"); + assert.equal(listed?.currentActivity.summary, "read"); + assert.doesNotMatch(JSON.stringify(status), /private transcript body|secret file contents/u); + assert.doesNotMatch(JSON.stringify(listed), /private transcript body|secret file contents/u); +}); + test("process failure reaches failed with diagnostics", async () => { const runner = new FakeRunner(); const supervisor = new Supervisor(runner, "/tmp"); diff --git a/modules/agents/pi/extensions/subagents/supervisor.ts b/modules/agents/pi/extensions/subagents/supervisor.ts index de45d58..9bcf5a9 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -3,6 +3,7 @@ import type { ChildRecord, ChildRunner, ContextMode, + RunnerActivity, RunnerEvents, SpawnAccepted, SpawnRequest, @@ -179,9 +180,11 @@ export class Supervisor { elapsedMs: 0, lastEvent: "queued", lastEventAt: now, + currentActivity: { type: "queued", summary: "queued", at: now }, + activityHistory: [{ type: "queued", summary: "queued", at: now }], resultAvailable: false, }; - const child: RunningChild = { record: { status }, request: { ...request, prompt, label: status.label, context: status.context, tools: status.tools } }; + const child: RunningChild = { record: { status, activityEvents: [{ type: "queued", summary: "queued", at: now }] }, request: { ...request, prompt, label: status.label, context: status.context, tools: status.tools } }; this.children.set(id, child); this.emitMilestone(child, "accepted"); this.queue.push(child); @@ -242,6 +245,7 @@ export class Supervisor { record.status.completedAt = now; record.status.lastEvent = "completed"; record.status.lastEventAt = now; + this.recordActivity(record, "completed", now); record.status.stopReason = stopReason; record.status.resultAvailable = true; if (child) this.emitMilestone(child, "completed"); @@ -260,6 +264,7 @@ export class Supervisor { record.status.completedAt = now; record.status.lastEvent = "failed"; record.status.lastEventAt = now; + this.recordActivity(record, "failed", now); record.status.error = error; record.status.stopReason = "failed"; if (child) this.emitMilestone(child, "failed"); @@ -274,6 +279,7 @@ export class Supervisor { child.record.status.completedAt = now; child.record.status.lastEvent = state; child.record.status.lastEventAt = now; + this.recordActivity(child.record, state, now); child.record.status.stopReason = reason; this.emitMilestone(child, state); } @@ -317,15 +323,30 @@ export class Supervisor { return [...this.children.values()].find((child) => child.record === record); } - private setState(status: SubagentStatus, state: SubagentStatus["state"], event: string) { + activity(id: string) { + return this.require(id).record.activityEvents.map((event) => ({ ...event })); + } + + private setState(status: SubagentStatus, state: SubagentStatus["state"], event: RunnerActivity) { if (isTerminal(status.state)) return; + const record = this.require(status.id).record; const now = new Date().toISOString(); + const activity = this.recordActivity(record, event, now); status.state = state; - status.lastEvent = event; + status.lastEvent = activity.type; status.lastEventAt = now; this.emitChange(); } + private recordActivity(record: ChildRecord, event: RunnerActivity, at: string) { + const activity = normalizeActivity(event, at); + record.activityEvents.push(activity); + const summary = summarizeActivity(activity); + record.status.currentActivity = summary; + record.status.activityHistory.push(summary); + return activity; + } + private require(id: string): RunningChild { const child = this.children.get(id); if (!child) throw new Error(`unknown subagent id: ${id}`); @@ -423,3 +444,86 @@ function truncateLabel(label: string): string { function isTerminal(state: SubagentStatus["state"]): boolean { return isTerminalState(state); } + +function normalizeActivity(event: RunnerActivity, at: string) { + if (typeof event === "string") return { type: event, summary: event, at }; + const type = typeof event.type === "string" ? event.type : "activity"; + const role = typeof event.role === "string" ? event.role : undefined; + const tool = toolFromActivity(event); + const phase = typeof event.phase === "string" ? event.phase : phaseFromType(type, event); + const text = textFromActivity(event); + const input = inputFromActivity(event); + const output = "output" in event ? event.output : "result" in event ? event.result : "partialResult" in event ? event.partialResult : undefined; + const error = typeof event.error === "string" ? event.error : undefined; + return { type, summary: summaryFor({ type, role, tool, phase, input, output, error }), at, role, tool, phase, text, input, output, error, payload: { ...event } }; +} + +function summarizeActivity(activity: ReturnType) { + const { type, summary, at, role, tool, phase } = activity; + return { type, summary, at, role, tool, phase }; +} + +function toolFromActivity(event: Record): string | undefined { + for (const key of ["tool", "toolName", "name"]) { + const value = event[key]; + if (typeof value === "string") return value; + } + return undefined; +} + +function phaseFromType(type: string, event: Record): string | undefined { + const assistantEvent = event.assistantMessageEvent; + if (assistantEvent && typeof assistantEvent === "object" && !Array.isArray(assistantEvent)) { + const assistantType = (assistantEvent as { type?: unknown }).type; + if (typeof assistantType === "string") return assistantType; + } + if (type.endsWith("_start")) return "started"; + if (type.endsWith("_started")) return "started"; + if (type.endsWith("_update")) return "update"; + if (type.endsWith("_delta")) return "delta"; + if (type.endsWith("_end")) return "completed"; + if (type.endsWith("_completed")) return "completed"; + if (type.endsWith("_failed")) return "failed"; + return undefined; +} + +function textFromActivity(event: Record): string | undefined { + for (const key of ["text", "body", "content", "delta"]) { + const value = event[key]; + if (typeof value === "string") return value; + } + const assistantEvent = event.assistantMessageEvent; + if (assistantEvent && typeof assistantEvent === "object" && !Array.isArray(assistantEvent)) { + for (const key of ["delta", "content"]) { + const value = (assistantEvent as Record)[key]; + if (typeof value === "string") return value; + } + } + return undefined; +} + +function inputFromActivity(event: Record): unknown { + if ("input" in event) return event.input; + if ("args" in event) return event.args; + return undefined; +} + +function summaryFor(activity: { type: string; role?: string; tool?: string; phase?: string; input?: unknown; output?: unknown; error?: string }): string { + if (activity.error) return `${activity.tool ?? activity.type} failed: ${activity.error}`; + if (activity.tool) return `${activity.tool}${inputHint(activity.input)}`; + if (activity.type.startsWith("message")) return `${activity.role ?? "assistant"} message${activity.phase ? ` ${activity.phase}` : ""}`; + return activity.type; +} + +function inputHint(input: unknown): string { + if (!input || typeof input !== "object" || Array.isArray(input)) return ""; + const path = (input as { path?: unknown }).path; + if (typeof path === "string" && path.trim()) return ` ${path.trim()}`; + const command = (input as { command?: unknown }).command; + if (typeof command === "string" && command.trim()) return ` ${truncateActivityHint(command.trim())}`; + return ""; +} + +function truncateActivityHint(value: string): string { + return value.length <= 80 ? value : `${value.slice(0, 79).trimEnd()}…`; +} diff --git a/modules/agents/pi/extensions/subagents/types.ts b/modules/agents/pi/extensions/subagents/types.ts index 2274ccb..db36bb3 100644 --- a/modules/agents/pi/extensions/subagents/types.ts +++ b/modules/agents/pi/extensions/subagents/types.ts @@ -31,6 +31,27 @@ export interface SpawnAccepted { hint: string; } +export interface SubagentActivitySummary { + type: string; + summary: string; + at: string; + role?: string; + tool?: string; + phase?: string; +} + +export interface SubagentActivityEvent extends SubagentActivitySummary { + text?: string; + input?: unknown; + output?: unknown; + error?: string; + payload?: Record; +} + +export interface SubagentCurrentActivity extends SubagentActivitySummary {} + +export type RunnerActivity = string | Record; + export interface SubagentStatus { id: string; label: string; @@ -47,6 +68,8 @@ export interface SubagentStatus { elapsedMs: number; lastEvent?: string; lastEventAt?: string; + currentActivity?: SubagentCurrentActivity; + activityHistory: SubagentActivitySummary[]; stopReason?: string; resultAvailable: boolean; childSession?: string; @@ -79,12 +102,13 @@ export interface SubagentWaitResult { export interface ChildRecord { status: SubagentStatus; + activityEvents: SubagentActivityEvent[]; result?: string; } export interface RunnerEvents { accepted(childSession?: string): void; - running(event: string): void; + running(event: RunnerActivity): void; settling(): void; completed(result: string, stopReason?: string): void; failed(error: string): void; diff --git a/modules/agents/pi/extensions/subagents/ui.test.ts b/modules/agents/pi/extensions/subagents/ui.test.ts index 523eb4c..b659e64 100644 --- a/modules/agents/pi/extensions/subagents/ui.test.ts +++ b/modules/agents/pi/extensions/subagents/ui.test.ts @@ -9,6 +9,7 @@ function status(overrides: Partial & { id: string; label: string context: "independent", cwd: "/tmp", elapsedMs: 0, + activityHistory: [], resultAvailable: false, startedAt: "2026-08-01T00:00:00.000Z", tools: "inherit", @@ -34,6 +35,26 @@ test("compact monitor aggregates visible children by actionable lifecycle group" ); }); +test("expanded monitor shows concise current activity summaries instead of raw event types", () => { + const rendered = widget([ + status({ + id: "sg-reading", + label: "Audit guest enablement plan", + state: "running", + elapsedMs: 12_000, + lastEvent: "message_update", + currentActivity: { + type: "message_update", + summary: "read secret-notes.md", + at: "2026-08-01T00:00:12.000Z", + }, + }), + ], true)().render(240); + + assert.deepEqual(rendered, ["▶ running 12s Audit guest enablement plan last: read secret-notes.md"]); + assert.doesNotMatch(rendered.join("\n"), /message_update|private transcript body/u); +}); + test("expanded monitor renders one truncated row per child with state, elapsed time, and activity marker", () => { const lines = renderInspector([ status({ diff --git a/modules/agents/pi/extensions/subagents/ui.ts b/modules/agents/pi/extensions/subagents/ui.ts index e710a47..25907c4 100644 --- a/modules/agents/pi/extensions/subagents/ui.ts +++ b/modules/agents/pi/extensions/subagents/ui.ts @@ -55,6 +55,7 @@ function renderStatusRow(status: SubagentStatus): string { function statusMarker(status: SubagentStatus): string | undefined { if (status.error) return `error: ${status.error}`; if (status.resultAvailable) return "result: available"; + if (status.currentActivity) return `last: ${status.currentActivity.summary}`; if (status.lastEvent) return `last: ${status.lastEvent}`; if (status.state === "queued") return "waiting"; if (status.state === "settling") return "settling";