From 007ba81c02842f7d47a01fe8288b9c4fc4ff2ab5 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 14:44:07 -0400 Subject: [PATCH] feat(pi): finish subagent runtime surfaces --- .../agents/pi/extensions/subagents/config.ts | 26 +++- .../agents/pi/extensions/subagents/index.ts | 100 +++++++++++- .../agents/pi/extensions/subagents/runner.ts | 10 +- .../extensions/subagents/supervisor.test.ts | 30 ++++ .../pi/extensions/subagents/supervisor.ts | 147 +++++++++++++----- .../agents/pi/extensions/subagents/types.ts | 1 + modules/agents/pi/extensions/subagents/ui.ts | 28 ++++ 7 files changed, 296 insertions(+), 46 deletions(-) create mode 100644 modules/agents/pi/extensions/subagents/ui.ts diff --git a/modules/agents/pi/extensions/subagents/config.ts b/modules/agents/pi/extensions/subagents/config.ts index 836793a..3fc6cf1 100644 --- a/modules/agents/pi/extensions/subagents/config.ts +++ b/modules/agents/pi/extensions/subagents/config.ts @@ -11,6 +11,11 @@ export interface Diagnostics { export interface SubagentsConfig { defaultContext: ContextMode; defaultTools: string; + maxConcurrent: number; + ui: { + enabled: boolean; + defaultExpanded: boolean; + }; toolProfiles: Record; } @@ -32,6 +37,8 @@ export const BUILT_IN_TOOL_PROFILES: Record = { const DEFAULT_CONFIG: SubagentsConfig = { defaultContext: "independent", defaultTools: "read-only", + maxConcurrent: 3, + ui: { enabled: true, defaultExpanded: false }, toolProfiles: { ...BUILT_IN_TOOL_PROFILES }, }; @@ -99,10 +106,25 @@ function normalizeConfig(raw: unknown, diagnostics: Diagnostics, label: string): else if (input.defaultContext !== undefined) diagnostics.warnings.push(`Invalid ${label} defaultContext ignored`); if (typeof input.defaultTools === "string") config.defaultTools = input.defaultTools; else if (input.defaultTools !== undefined) diagnostics.warnings.push(`Invalid ${label} defaultTools ignored`); + if (typeof input.maxConcurrent === "number" && Number.isInteger(input.maxConcurrent) && input.maxConcurrent > 0) config.maxConcurrent = input.maxConcurrent; + else if (input.maxConcurrent !== undefined) diagnostics.warnings.push(`Invalid ${label} maxConcurrent ignored`); + if (input.ui !== undefined) config.ui = normalizeUi(input.ui, diagnostics, label); if (input.toolProfiles !== undefined) config.toolProfiles = normalizeProfiles(input.toolProfiles, diagnostics, label); return config; } +function normalizeUi(raw: unknown, diagnostics: Diagnostics, label: string): SubagentsConfig["ui"] | undefined { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + diagnostics.warnings.push(`Invalid ${label} ui ignored`); + return undefined; + } + const input = raw as Record; + return { + enabled: typeof input.enabled === "boolean" ? input.enabled : DEFAULT_CONFIG.ui.enabled, + defaultExpanded: typeof input.defaultExpanded === "boolean" ? input.defaultExpanded : DEFAULT_CONFIG.ui.defaultExpanded, + }; +} + function normalizeProfiles(raw: unknown, diagnostics: Diagnostics, label: string): Record { const profiles: Record = {}; if (!raw || typeof raw !== "object" || Array.isArray(raw)) { @@ -136,6 +158,8 @@ function mergeConfig(base: SubagentsConfig, override: Partial | const merged = cloneConfig(base); if (override.defaultContext) merged.defaultContext = override.defaultContext; if (override.defaultTools) merged.defaultTools = override.defaultTools; + if (override.maxConcurrent) merged.maxConcurrent = override.maxConcurrent; + if (override.ui) merged.ui = { ...merged.ui, ...override.ui }; if (override.toolProfiles) merged.toolProfiles = { ...merged.toolProfiles, ...override.toolProfiles }; for (const key of Object.keys(merged.toolProfiles)) { if (key in BUILT_IN_TOOL_PROFILES) merged.toolProfiles[key] = BUILT_IN_TOOL_PROFILES[key]; @@ -144,7 +168,7 @@ function mergeConfig(base: SubagentsConfig, override: Partial | } function cloneConfig(config: SubagentsConfig): SubagentsConfig { - return { ...config, toolProfiles: { ...config.toolProfiles } }; + return { ...config, ui: { ...config.ui }, toolProfiles: { ...config.toolProfiles } }; } function defaultAgentDir(): string { diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts index 3643bc8..7394769 100644 --- a/modules/agents/pi/extensions/subagents/index.ts +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -4,14 +4,31 @@ import { loadAgents } from "./agents.ts"; import { loadConfig, resolveSpawn, type Diagnostics } from "./config.ts"; import { SubprocessRpcRunner } from "./runner.ts"; import { Supervisor } from "./supervisor.ts"; -import type { SpawnRequest } from "./types.ts"; +import type { SpawnRequest, SubagentStatus } from "./types.ts"; +import { widget } from "./ui.ts"; let supervisor: Supervisor | undefined; let lastDiagnostics: Diagnostics = { warnings: [] }; +let lastStatuses: SubagentStatus[] = []; +let uiExpanded = false; export default function subagents(pi: ExtensionAPI) { const getSupervisor = (ctx: ExtensionContext): Supervisor => { - if (!supervisor) supervisor = new Supervisor(new SubprocessRpcRunner(), cwdOf(ctx)); + if (supervisor) return supervisor; + const diagnostics: Diagnostics = { warnings: [] }; + const cwd = cwdOf(ctx); + const config = loadConfig(cwd, isProjectTrusted(ctx), diagnostics); + lastDiagnostics = diagnostics; + uiExpanded = config.ui.defaultExpanded; + supervisor = new Supervisor(new SubprocessRpcRunner(), cwd, { + maxConcurrent: config.maxConcurrent, + onMilestone: (status, event) => pi.appendEntry("subagent_milestone", { event, status }), + onChange: (statuses) => { + lastStatuses = statuses; + updateUi(ctx, config.ui.enabled); + }, + }); + updateUi(ctx, config.ui.enabled); return supervisor; }; @@ -22,7 +39,9 @@ export default function subagents(pi: ExtensionAPI) { const config = loadConfig(cwd, trusted, diagnostics); const agents = loadAgents(cwd, trusted, diagnostics); lastDiagnostics = diagnostics; - return resolveSpawn(request, config, agents); + const resolved = resolveSpawn(request, config, agents); + if (resolved.context === "fork") resolved.parentSessionFile = ctx.sessionManager.getSessionFile(); + return resolved; }; pi.registerTool({ @@ -32,7 +51,7 @@ export default function subagents(pi: ExtensionAPI) { parameters: Type.Object({ prompt: Type.String({ description: "Prompt for the delegated subagent" }), agent: Type.Optional(Type.String({ description: "Named agent definition to use" })), - context: Type.Optional(Type.Literal("independent")), + context: Type.Optional(Type.Union([Type.Literal("independent"), Type.Literal("fork")])), model: Type.Optional(Type.String({ description: "Optional model selector for the child" })), thinking: Type.Optional(Type.String({ description: "Optional thinking level for the child" })), tools: Type.Optional(Type.String({ description: "Tool profile name" })), @@ -44,6 +63,38 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerTool({ + name: "subagent_batch", + label: "Spawn subagent batch", + description: "Start multiple subagents and return immediately with accepted child ids and per-entry failures", + parameters: Type.Object({ + subagents: Type.Array( + Type.Object({ + prompt: Type.String({ description: "Prompt for the delegated subagent" }), + agent: Type.Optional(Type.String({ description: "Named agent definition to use" })), + context: Type.Optional(Type.Union([Type.Literal("independent"), Type.Literal("fork")])), + model: Type.Optional(Type.String({ description: "Optional model selector for the child" })), + thinking: Type.Optional(Type.String({ description: "Optional thinking level for the child" })), + tools: Type.Optional(Type.String({ description: "Tool profile name" })), + }), + ), + }), + async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + const requests = Array.isArray((params as { subagents?: unknown }).subagents) ? ((params as { subagents: SpawnRequest[] }).subagents) : []; + const accepted: SpawnRequest[] = []; + const failed: Array<{ index: number; error: string }> = []; + requests.forEach((request, index) => { + try { + accepted.push(resolve(ctx, request)); + } catch (error) { + failed.push({ index, error: error instanceof Error ? error.message : String(error) }); + } + }); + const result = getSupervisor(ctx).spawnBatch(accepted); + return textResult({ accepted: result.accepted, failed: [...failed, ...result.failed] }); + }, + }); + pi.registerTool({ name: "subagent_list", label: "List subagents", @@ -98,6 +149,18 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerCommand("subagent-batch", { + description: "Start ad hoc independent subagents split by |", + handler: async (args, ctx) => { + const requests = args + .split("|") + .map((prompt) => prompt.trim()) + .filter(Boolean) + .map((prompt) => resolve(ctx, { prompt })); + ctx.ui.notify(JSON.stringify(getSupervisor(ctx).spawnBatch(requests), null, 2), "info"); + }, + }); + pi.registerCommand("subagent-list", { description: "Show subagent status records", handler: async (_args, ctx) => { @@ -119,6 +182,15 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerCommand("subagent-ui", { + description: "Toggle the bundled subagent status inspector", + handler: async (_args, ctx) => { + uiExpanded = !uiExpanded; + updateUi(ctx, true); + ctx.ui.notify(`Subagent inspector ${uiExpanded ? "expanded" : "collapsed"}`, "info"); + }, + }); + pi.registerCommand("subagent-diagnostics", { description: "Show subagent configuration diagnostics from the last load", handler: async (_args, ctx) => { @@ -139,10 +211,24 @@ export default function subagents(pi: ExtensionAPI) { }); } +function updateUi(ctx: ExtensionContext, enabled: boolean) { + if (!ctx.hasUI) return; + ctx.ui.setWidget("subagents", enabled ? widget(lastStatuses, uiExpanded) : undefined); +} + function parseSpawnArgs(args: string): SpawnRequest { - const match = /^--agent\s+(\S+)\s+([\s\S]+)$/u.exec(args.trim()); - if (!match) return { prompt: args }; - return { agent: match[1], prompt: match[2] }; + const parts = args.trim().split(/\s+/u); + const request: Partial = {}; + while (parts.length >= 2 && parts[0].startsWith("--")) { + const flag = parts.shift(); + const value = parts.shift(); + if (flag === "--agent") request.agent = value; + else if (flag === "--context" && (value === "independent" || value === "fork")) request.context = value; + else if (flag === "--tools") request.tools = value; + else if (flag === "--model") request.model = value; + else if (flag === "--thinking") request.thinking = value; + } + return { ...request, prompt: parts.join(" ") || args } as SpawnRequest; } function isProjectTrusted(ctx: ExtensionContext): boolean { diff --git a/modules/agents/pi/extensions/subagents/runner.ts b/modules/agents/pi/extensions/subagents/runner.ts index 8edeaaa..7231fa8 100644 --- a/modules/agents/pi/extensions/subagents/runner.ts +++ b/modules/agents/pi/extensions/subagents/runner.ts @@ -156,7 +156,7 @@ class RpcChildHandle implements ChildHandle { export class SubprocessRpcRunner implements ChildRunner { async start(id: string, request: SpawnRequest, cwd: string, events: RunnerEvents): Promise { - const args = [process.argv[1], "--mode", "rpc", "--no-extensions", "--name", `subagent ${id}`, ...toolArgs(request), ...modelArgs(request)]; + const args = [process.argv[1], "--mode", "rpc", "--no-extensions", "--name", `subagent ${id}`, ...contextArgs(request), ...toolArgs(request), ...modelArgs(request)]; const child = spawn(process.execPath, args, { cwd, env: childEnvironment(), @@ -174,6 +174,11 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +function contextArgs(request: SpawnRequest): string[] { + if (request.context !== "fork" || !request.parentSessionFile) return []; + return ["--fork", request.parentSessionFile]; +} + function toolArgs(request: SpawnRequest): string[] { const activeTools = request.toolProfile?.activeTools; if (activeTools === undefined || activeTools === null) return []; @@ -200,5 +205,8 @@ function childEnvironment(): NodeJS.ProcessEnv { function independentPrompt(request: SpawnRequest): string { const base = request.agentBody ? `${request.agentBody}\n\n` : ""; + if (request.context === "fork") { + return `${base}You are running as a delegated subagent in fork context.\nUse the inherited parent session context, then return a concise final answer for the parent agent.\n\nTask:\n${request.prompt}`; + } return `${base}You are running as a delegated subagent in independent context.\nDo not assume access to the parent conversation transcript.\nReturn a concise final answer for the parent agent.\n\nTask:\n${request.prompt}`; } diff --git a/modules/agents/pi/extensions/subagents/supervisor.test.ts b/modules/agents/pi/extensions/subagents/supervisor.test.ts index 69f1bc8..5eba4ea 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.test.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.test.ts @@ -109,3 +109,33 @@ test("completed children ignore later cancel", async () => { assert.equal(result.result, "done"); assert.equal(runner.starts[0].handle.cancelCalls, 0); }); + +test("batch spawn returns accepted ids and per-entry failures", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + + const result = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "" }, { prompt: "two" }]); + await sleep(0); + + assert.equal(result.accepted.length, 2); + assert.equal(result.failed.length, 1); + assert.equal(result.failed[0].index, 1); + assert.equal(runner.starts.length, 2); +}); + +test("maxConcurrent preserves queued records", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { maxConcurrent: 1 }); + + const result = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "two" }]); + await sleep(0); + + assert.equal(result.accepted.length, 2); + assert.equal(runner.starts.length, 1); + assert.equal(supervisor.status(result.accepted[1].id).state, "queued"); + + runner.starts[0].events.completed("done", "agent_settled"); + await sleep(0); + + assert.equal(runner.starts.length, 2); +}); diff --git a/modules/agents/pi/extensions/subagents/supervisor.ts b/modules/agents/pi/extensions/subagents/supervisor.ts index fe42838..86f3b32 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -3,16 +3,26 @@ import { cloneResult, cloneStatus, toAccepted } from "./status.ts"; interface RunningChild { record: ChildRecord; + request: SpawnRequest; handle?: ChildHandle; startTimer?: ReturnType; runTimer?: ReturnType; } interface SupervisorOptions { + maxConcurrent?: number; + recentTerminalLimit?: number; timeouts?: { startMs?: number; runMs?: number; }; + onMilestone?: (status: SubagentStatus, event: string) => void; + onChange?: (statuses: SubagentStatus[]) => void; +} + +export interface BatchSpawnResult { + accepted: SpawnAccepted[]; + failed: Array<{ index: number; error: string }>; } const DEFAULT_TIMEOUTS = { @@ -23,6 +33,7 @@ const DEFAULT_TIMEOUTS = { export class Supervisor { private nextChild = 0; private readonly children = new Map(); + private readonly queue: RunningChild[] = []; constructor( private readonly runner: ChildRunner, @@ -31,6 +42,61 @@ export class Supervisor { ) {} spawn(request: SpawnRequest): SpawnAccepted { + return this.createChild(request); + } + + spawnBatch(requests: SpawnRequest[]): BatchSpawnResult { + const accepted: SpawnAccepted[] = []; + const failed: Array<{ index: number; error: string }> = []; + requests.forEach((request, index) => { + try { + accepted.push(this.createChild(request)); + } catch (error) { + failed.push({ index, error: error instanceof Error ? error.message : String(error) }); + } + }); + return { accepted, failed }; + } + + list(): SubagentStatus[] { + const statuses = [...this.children.values()].map((child) => cloneStatus(child.record.status)); + const active = statuses.filter((status) => !isTerminal(status.state)); + const terminal = statuses + .filter((status) => isTerminal(status.state)) + .sort((a, b) => Date.parse(b.completedAt ?? b.startedAt) - Date.parse(a.completedAt ?? a.startedAt)) + .slice(0, this.options.recentTerminalLimit ?? 10); + return [...active, ...terminal]; + } + + status(id: string): SubagentStatus { + return cloneStatus(this.require(id).record.status); + } + + result(id: string): SubagentResult { + return cloneResult(this.require(id).record); + } + + async cancel(id: string): Promise { + const child = this.require(id); + if (isTerminal(child.record.status.state)) return cloneStatus(child.record.status); + await child.handle?.cancel(); + this.completeWithoutResult(child, "cancelled", "cancelled"); + this.pumpQueue(); + return cloneStatus(child.record.status); + } + + async shutdown(): Promise { + await Promise.allSettled( + [...this.children.values()].map(async (child) => { + if (!isTerminal(child.record.status.state)) { + await child.handle?.cancel(); + this.completeWithoutResult(child, "cancelled", "shutdown"); + } + }), + ); + } + + private createChild(request: SpawnRequest): SpawnAccepted { const prompt = typeof request.prompt === "string" ? request.prompt.trim() : ""; if (!prompt) throw new Error("prompt is required"); @@ -53,15 +119,31 @@ export class Supervisor { lastEventAt: now, resultAvailable: false, }; - const child: RunningChild = { record: { status } }; + const child: RunningChild = { record: { status }, request: { ...request, prompt, context: status.context, tools: status.tools } }; this.children.set(id, child); + this.emitMilestone(child, "accepted"); + this.queue.push(child); + this.pumpQueue(); + return toAccepted(cloneStatus(status)); + } + + private pumpQueue() { + while (this.runningCount() < this.maxConcurrent()) { + const child = this.queue.shift(); + if (!child) break; + if (isTerminal(child.record.status.state)) continue; + this.start(child); + } + this.emitChange(); + } + + private start(child: RunningChild) { this.setState(child.record.status, "starting", "starting"); this.armStartTimer(child); - setTimeout(() => { if (isTerminal(child.record.status.state)) return; void this.runner - .start(id, { ...request, prompt, context: status.context, tools: status.tools }, this.cwd, this.eventsFor(child.record)) + .start(child.record.status.id, child.request, this.cwd, this.eventsFor(child.record)) .then((handle) => { child.handle = handle; if (isTerminal(child.record.status.state)) void handle.cancel(); @@ -70,39 +152,6 @@ export class Supervisor { this.fail(child.record, error instanceof Error ? error.message : String(error)); }); }, 0); - - return toAccepted(cloneStatus(status)); - } - - list(): SubagentStatus[] { - return [...this.children.values()].map((child) => cloneStatus(child.record.status)); - } - - status(id: string): SubagentStatus { - return cloneStatus(this.require(id).record.status); - } - - result(id: string): SubagentResult { - return cloneResult(this.require(id).record); - } - - async cancel(id: string): Promise { - const child = this.require(id); - if (isTerminal(child.record.status.state)) return cloneStatus(child.record.status); - await child.handle?.cancel(); - this.completeWithoutResult(child, "cancelled", "cancelled"); - return cloneStatus(child.record.status); - } - - async shutdown(): Promise { - await Promise.allSettled( - [...this.children.values()].map(async (child) => { - if (!isTerminal(child.record.status.state)) { - await child.handle?.cancel(); - this.completeWithoutResult(child, "cancelled", "shutdown"); - } - }), - ); } private eventsFor(record: ChildRecord): RunnerEvents { @@ -133,6 +182,8 @@ export class Supervisor { record.status.lastEventAt = now; record.status.stopReason = stopReason; record.status.resultAvailable = true; + if (child) this.emitMilestone(child, "completed"); + this.pumpQueue(); }, failed: (error) => this.fail(record, error), }; @@ -149,6 +200,8 @@ export class Supervisor { record.status.lastEventAt = now; record.status.error = error; record.status.stopReason = "failed"; + if (child) this.emitMilestone(child, "failed"); + this.pumpQueue(); } private completeWithoutResult(child: RunningChild, state: "cancelled" | "timed_out", reason: string) { @@ -160,6 +213,7 @@ export class Supervisor { child.record.status.lastEvent = state; child.record.status.lastEventAt = now; child.record.status.stopReason = reason; + this.emitMilestone(child, state); } private armStartTimer(child: RunningChild) { @@ -182,6 +236,7 @@ export class Supervisor { if (isTerminal(child.record.status.state)) return; void child.handle?.cancel(); this.completeWithoutResult(child, "timed_out", reason); + this.pumpQueue(); } private clearTimers(child: RunningChild) { @@ -206,6 +261,7 @@ export class Supervisor { status.state = state; status.lastEvent = event; status.lastEventAt = now; + this.emitChange(); } private require(id: string): RunningChild { @@ -216,10 +272,27 @@ export class Supervisor { private resolveContext(context: ContextMode | undefined): ContextMode { if (context === undefined) return "independent"; - if (context !== "independent") throw new Error("only independent context is implemented before fork mode lands"); + if (context !== "independent" && context !== "fork") throw new Error(`unknown context: ${context}`); return context; } + private maxConcurrent(): number { + return Math.max(1, this.options.maxConcurrent ?? 3); + } + + private runningCount(): number { + return [...this.children.values()].filter((child) => ["starting", "running", "settling"].includes(child.record.status.state)).length; + } + + private emitMilestone(child: RunningChild, event: string) { + this.options.onMilestone?.(cloneStatus(child.record.status), event); + this.emitChange(); + } + + private emitChange() { + this.options.onChange?.(this.list()); + } + private allocateId(): string { this.nextChild += 1; return `sg-${Date.now().toString(36)}-${this.nextChild.toString(36)}`; diff --git a/modules/agents/pi/extensions/subagents/types.ts b/modules/agents/pi/extensions/subagents/types.ts index 209a098..ac6f407 100644 --- a/modules/agents/pi/extensions/subagents/types.ts +++ b/modules/agents/pi/extensions/subagents/types.ts @@ -24,6 +24,7 @@ export interface SpawnRequest { tools?: string; toolProfile?: ToolProfile; agentBody?: string; + parentSessionFile?: string; } export interface SpawnAccepted { diff --git a/modules/agents/pi/extensions/subagents/ui.ts b/modules/agents/pi/extensions/subagents/ui.ts new file mode 100644 index 0000000..b466096 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/ui.ts @@ -0,0 +1,28 @@ +import type { SubagentStatus } from "./types.ts"; + +export function renderSummary(statuses: SubagentStatus[]): string[] { + const running = statuses.filter((status) => ["starting", "running", "settling"].includes(status.state)).length; + const queued = statuses.filter((status) => status.state === "queued").length; + const terminal = statuses.filter((status) => ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(status.state)).length; + if (running === 0 && queued === 0 && terminal === 0) return []; + return [`subagents: ${running} running · ${queued} queued · ${terminal} recent`]; +} + +export function renderInspector(statuses: SubagentStatus[]): string[] { + const lines = renderSummary(statuses); + for (const status of statuses) { + lines.push( + `${status.id} ${status.label} ${status.context} ${status.state} ${Math.round(status.elapsedMs / 1000)}s ${status.model ?? "inherit"} ${status.tools} ${status.lastEvent ?? "none"} result:${status.resultAvailable ? "yes" : "no"}`, + ); + } + return lines; +} + +export function widget(statuses: SubagentStatus[], expanded: boolean) { + return () => ({ + invalidate() {}, + render(width: number) { + return (expanded ? renderInspector(statuses) : renderSummary(statuses)).map((line) => (line.length > width ? line.slice(0, Math.max(0, width - 1)) : line)); + }, + }); +}