From 63da676b5afa5605e2312c13481662931380a447 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 13:54:57 -0400 Subject: [PATCH 1/6] feat(pi): add subagent tracer bullet --- .../agents/pi/extensions/subagents/index.ts | 123 ++++++++++++++ .../agents/pi/extensions/subagents/runner.ts | 157 +++++++++++++++++ .../agents/pi/extensions/subagents/status.ts | 38 +++++ .../pi/extensions/subagents/supervisor.ts | 158 ++++++++++++++++++ .../agents/pi/extensions/subagents/types.ts | 84 ++++++++++ 5 files changed, 560 insertions(+) create mode 100644 modules/agents/pi/extensions/subagents/index.ts create mode 100644 modules/agents/pi/extensions/subagents/runner.ts create mode 100644 modules/agents/pi/extensions/subagents/status.ts create mode 100644 modules/agents/pi/extensions/subagents/supervisor.ts create mode 100644 modules/agents/pi/extensions/subagents/types.ts diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts new file mode 100644 index 0000000..ac63f49 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -0,0 +1,123 @@ +import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; +import { Type } from "typebox"; +import { SubprocessRpcRunner } from "./runner.ts"; +import { Supervisor } from "./supervisor.ts"; +import type { SpawnRequest } from "./types.ts"; + +let supervisor: Supervisor | undefined; + +export default function subagents(pi: ExtensionAPI) { + const getSupervisor = (ctx: ExtensionContext): Supervisor => { + if (!supervisor) supervisor = new Supervisor(new SubprocessRpcRunner(), cwdOf(ctx)); + return supervisor; + }; + + pi.registerTool({ + name: "subagent_spawn", + label: "Spawn subagent", + description: "Start one ad hoc independent subagent and return immediately with its child id", + parameters: Type.Object({ + prompt: Type.String({ description: "Prompt for the delegated subagent" }), + context: Type.Optional(Type.Literal("independent")), + model: Type.Optional(Type.String({ description: "Optional model selector for status metadata" })), + thinking: Type.Optional(Type.String({ description: "Optional thinking level for status metadata" })), + }), + async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + const accepted = getSupervisor(ctx).spawn(params as SpawnRequest); + ctx.ui?.notify?.(`Started subagent ${accepted.id}`, "info"); + return textResult(accepted); + }, + }); + + pi.registerTool({ + name: "subagent_list", + label: "List subagents", + description: "List active and recent subagents for this parent session", + parameters: Type.Object({}), + async execute(_toolCallId, _params, _signal, _onUpdate, ctx) { + return textResult(getSupervisor(ctx).list()); + }, + }); + + pi.registerTool({ + name: "subagent_status", + label: "Get subagent status", + description: "Get current lifecycle status for one subagent", + parameters: Type.Object({ + id: Type.String({ description: "Subagent id returned by subagent_spawn" }), + }), + async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + return textResult(getSupervisor(ctx).status(String((params as { id: unknown }).id))); + }, + }); + + pi.registerTool({ + name: "subagent_result", + label: "Get subagent result", + description: "Return still-running before completion and the final answer after completion", + parameters: Type.Object({ + id: Type.String({ description: "Subagent id returned by subagent_spawn" }), + }), + async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + return textResult(getSupervisor(ctx).result(String((params as { id: unknown }).id))); + }, + }); + + pi.registerTool({ + name: "subagent_cancel", + label: "Cancel subagent", + description: "Cancel a running subagent", + parameters: Type.Object({ + id: Type.String({ description: "Subagent id returned by subagent_spawn" }), + }), + async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + return textResult(await getSupervisor(ctx).cancel(String((params as { id: unknown }).id))); + }, + }); + + pi.registerCommand("subagent-spawn", { + description: "Start an ad hoc independent subagent", + handler: async (args, ctx) => { + const accepted = getSupervisor(ctx).spawn({ prompt: args }); + ctx.ui.notify(`Started subagent ${accepted.id}`, "info"); + }, + }); + + pi.registerCommand("subagent-list", { + description: "Show subagent status records", + handler: async (_args, ctx) => { + ctx.ui.notify(JSON.stringify(getSupervisor(ctx).list(), null, 2), "info"); + }, + }); + + pi.registerCommand("subagent-status", { + description: "Show a subagent status by id", + handler: async (args, ctx) => { + ctx.ui.notify(JSON.stringify(getSupervisor(ctx).status(args.trim()), null, 2), "info"); + }, + }); + + pi.registerCommand("subagent-result", { + description: "Show a subagent result by id", + handler: async (args, ctx) => { + ctx.ui.notify(JSON.stringify(getSupervisor(ctx).result(args.trim()), null, 2), "info"); + }, + }); + + pi.on("session_shutdown", async () => { + await supervisor?.shutdown(); + supervisor = undefined; + }); +} + +function cwdOf(ctx: ExtensionContext): string { + const sessionCwd = (ctx as unknown as { sessionManager?: { getCwd?: () => string }; cwd?: string }).sessionManager?.getCwd?.(); + return sessionCwd ?? (ctx as unknown as { cwd?: string }).cwd ?? process.cwd(); +} + +function textResult(value: unknown) { + return { + content: [{ type: "text" as const, text: JSON.stringify(value, null, 2) }], + details: value, + }; +} diff --git a/modules/agents/pi/extensions/subagents/runner.ts b/modules/agents/pi/extensions/subagents/runner.ts new file mode 100644 index 0000000..7683827 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/runner.ts @@ -0,0 +1,157 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import type { ChildHandle, ChildRunner, RunnerEvents, SpawnRequest } from "./types.ts"; + +interface PendingResponse { + resolve(value: unknown): void; + reject(error: Error): void; + command: string; +} + +interface RpcLine { + id?: string; + type?: string; + command?: string; + success?: boolean; + data?: unknown; + error?: string; + message?: string; +} + +class RpcChildHandle implements ChildHandle { + private buffer = ""; + private nextRequest = 0; + private settled = false; + private readonly pending = new Map(); + + constructor( + private readonly child: ChildProcessWithoutNullStreams, + private readonly events: RunnerEvents, + ) { + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => this.onStdout(chunk)); + child.stderr.on("data", (chunk) => this.events.running(`stderr: ${String(chunk).trim().slice(0, 200)}`)); + child.on("error", (error) => this.fail(error.message)); + child.on("close", (code, signal) => { + for (const pending of this.pending.values()) { + pending.reject(new Error(`RPC process closed before ${pending.command} response`)); + } + this.pending.clear(); + if (!this.settled) this.fail(`RPC process closed with code ${code ?? "null"} signal ${signal ?? "null"}`); + }); + } + + async prompt(message: string): Promise { + await this.send("prompt", { message }); + } + + async cancel(): Promise { + try { + await this.send("abort", {}); + } catch {} + this.child.stdin.end(); + if (!this.child.killed) this.child.kill("SIGTERM"); + } + + private onStdout(chunk: string) { + this.buffer += chunk; + while (true) { + const newline = this.buffer.indexOf("\n"); + if (newline === -1) return; + const line = this.buffer.slice(0, newline).replace(/\r$/, ""); + this.buffer = this.buffer.slice(newline + 1); + if (line.trim() === "") continue; + this.onLine(line); + } + } + + private onLine(line: string) { + let payload: RpcLine; + try { + payload = JSON.parse(line); + } catch { + this.events.running(`non-json rpc output: ${line.slice(0, 200)}`); + return; + } + + if (payload.type === "response" && payload.id) { + const pending = this.pending.get(payload.id); + if (!pending) return; + this.pending.delete(payload.id); + if (payload.success) pending.resolve(payload.data); + else pending.reject(new Error(payload.error ?? payload.message ?? `${pending.command} failed`)); + return; + } + + if (payload.type === "agent_started") { + this.events.running("agent_started"); + return; + } + + if (payload.type === "agent_settled") { + this.finish().catch((error) => this.fail(error instanceof Error ? error.message : String(error))); + return; + } + + if (payload.type) this.events.running(payload.type); + } + + private async finish() { + if (this.settled) return; + this.settled = true; + this.events.settling(); + const result = await this.send("get_last_assistant_text", {}); + const text = typeof result === "string" ? result : result && typeof result === "object" && "text" in result ? String((result as { text: unknown }).text) : ""; + this.events.completed(text, "agent_settled"); + this.child.stdin.end(); + if (!this.child.killed) this.child.kill("SIGTERM"); + } + + private fail(error: string) { + if (this.settled) return; + this.settled = true; + this.events.failed(error); + } + + private send(command: string, body: Record): Promise { + const id = `subagent-${++this.nextRequest}`; + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject, command }); + this.child.stdin.write(`${JSON.stringify({ id, type: command, ...body })}\n`, (error) => { + if (!error) return; + this.pending.delete(id); + reject(error); + }); + }); + } +} + +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}`]; + const child = spawn(process.execPath, args, { + cwd, + env: childEnvironment(), + stdio: ["pipe", "pipe", "pipe"], + detached: process.platform !== "win32", + }); + const handle = new RpcChildHandle(child, events); + events.accepted(); + void handle.prompt(independentPrompt(request)).catch((error) => events.failed(error instanceof Error ? error.message : String(error))); + return handle; + } +} + +function childEnvironment(): NodeJS.ProcessEnv { + const env = { ...process.env }; + delete env.PI_SESSION_ID; + delete env.PI_SESSION_FILE; + delete env.PI_PROVIDER; + delete env.PI_MODEL; + delete env.PI_REASONING_LEVEL; + return env; +} + +function independentPrompt(request: SpawnRequest): string { + return `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/status.ts b/modules/agents/pi/extensions/subagents/status.ts new file mode 100644 index 0000000..76df5e7 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/status.ts @@ -0,0 +1,38 @@ +import type { ChildRecord, SpawnAccepted, SubagentResult, SubagentStatus } from "./types.ts"; + +export function toAccepted(status: SubagentStatus): SpawnAccepted { + return { + id: status.id, + label: status.label, + context: status.context, + tools: status.tools, + state: status.state, + hint: `Use subagent_status or subagent_result with id ${status.id}`, + }; +} + +export function cloneStatus(status: SubagentStatus): SubagentStatus { + return { ...status, elapsedMs: elapsedMs(status) }; +} + +export function cloneResult(record: ChildRecord): SubagentResult { + const status = cloneStatus(record.status); + const terminal = ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(status.state); + return { + id: status.id, + state: status.state, + running: !terminal, + resultAvailable: status.resultAvailable, + result: record.result, + error: status.error, + completedAt: status.completedAt, + elapsedMs: status.elapsedMs, + }; +} + +export function elapsedMs(status: Pick): number { + const start = Date.parse(status.startedAt); + const end = status.completedAt ? Date.parse(status.completedAt) : Date.now(); + if (!Number.isFinite(start) || !Number.isFinite(end)) return 0; + return Math.max(0, end - start); +} diff --git a/modules/agents/pi/extensions/subagents/supervisor.ts b/modules/agents/pi/extensions/subagents/supervisor.ts new file mode 100644 index 0000000..6f45ab7 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -0,0 +1,158 @@ +import type { ChildHandle, ChildRecord, ChildRunner, ContextMode, RunnerEvents, SpawnAccepted, SpawnRequest, SubagentResult, SubagentStatus } from "./types.ts"; +import { cloneResult, cloneStatus, toAccepted } from "./status.ts"; + +interface RunningChild { + record: ChildRecord; + handle?: ChildHandle; +} + +export class Supervisor { + private nextChild = 0; + private readonly children = new Map(); + + constructor( + private readonly runner: ChildRunner, + private readonly cwd: string, + ) {} + + spawn(request: SpawnRequest): SpawnAccepted { + const prompt = typeof request.prompt === "string" ? request.prompt.trim() : ""; + if (!prompt) throw new Error("prompt is required"); + + const id = this.allocateId(); + const now = new Date().toISOString(); + const status: SubagentStatus = { + id, + label: request.agent ?? `ad-hoc ${id}`, + agent: request.agent, + adHoc: !request.agent, + context: this.resolveContext(request.context), + state: "queued", + cwd: this.cwd, + model: request.model, + thinking: request.thinking, + tools: "pi-default", + startedAt: now, + elapsedMs: 0, + lastEvent: "queued", + lastEventAt: now, + resultAvailable: false, + }; + const child: RunningChild = { record: { status } }; + this.children.set(id, child); + this.setState(child.record.status, "starting", "starting"); + + 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)) + .then((handle) => { + child.handle = handle; + }) + .catch((error) => { + 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(); + const now = new Date().toISOString(); + child.record.status.state = "cancelled"; + child.record.status.completedAt = now; + child.record.status.lastEvent = "cancelled"; + child.record.status.lastEventAt = now; + child.record.status.stopReason = "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(); + }), + ); + } + + private eventsFor(record: ChildRecord): RunnerEvents { + return { + accepted: (childSession) => { + if (childSession) record.status.childSession = childSession; + this.setState(record.status, "running", "prompt accepted"); + }, + running: (event) => { + if (!isTerminal(record.status.state)) this.setState(record.status, "running", event); + }, + settling: () => { + if (!isTerminal(record.status.state)) this.setState(record.status, "settling", "agent_settled"); + }, + completed: (result, stopReason) => { + const now = new Date().toISOString(); + record.result = result; + record.status.state = "completed"; + record.status.completedAt = now; + record.status.lastEvent = "completed"; + record.status.lastEventAt = now; + record.status.stopReason = stopReason; + record.status.resultAvailable = true; + }, + failed: (error) => this.fail(record, error), + }; + } + + private fail(record: ChildRecord, error: string) { + if (isTerminal(record.status.state)) return; + const now = new Date().toISOString(); + record.status.state = "failed"; + record.status.completedAt = now; + record.status.lastEvent = "failed"; + record.status.lastEventAt = now; + record.status.error = error; + record.status.stopReason = "failed"; + } + + private setState(status: SubagentStatus, state: SubagentStatus["state"], event: string) { + if (isTerminal(status.state)) return; + const now = new Date().toISOString(); + status.state = state; + status.lastEvent = event; + status.lastEventAt = now; + } + + private require(id: string): RunningChild { + const child = this.children.get(id); + if (!child) throw new Error(`unknown subagent id: ${id}`); + return child; + } + + private resolveContext(context: ContextMode | undefined): ContextMode { + if (context === undefined) return "independent"; + if (context !== "independent") throw new Error("only independent context is implemented in this tracer bullet"); + return context; + } + + private allocateId(): string { + this.nextChild += 1; + return `sg-${Date.now().toString(36)}-${this.nextChild.toString(36)}`; + } +} + +function isTerminal(state: SubagentStatus["state"]): boolean { + return ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(state); +} diff --git a/modules/agents/pi/extensions/subagents/types.ts b/modules/agents/pi/extensions/subagents/types.ts new file mode 100644 index 0000000..4380099 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/types.ts @@ -0,0 +1,84 @@ +export type ContextMode = "independent" | "fork"; + +export type SubagentState = + | "queued" + | "starting" + | "running" + | "settling" + | "completed" + | "failed" + | "cancelled" + | "timed_out" + | "orphaned"; + +export interface SpawnRequest { + prompt: string; + context?: ContextMode; + agent?: string; + model?: string; + thinking?: string; + tools?: string; +} + +export interface SpawnAccepted { + id: string; + label: string; + context: ContextMode; + tools: string; + state: SubagentState; + hint: string; +} + +export interface SubagentStatus { + id: string; + label: string; + agent?: string; + adHoc: boolean; + context: ContextMode; + state: SubagentState; + cwd: string; + model?: string; + thinking?: string; + tools: string; + startedAt: string; + completedAt?: string; + elapsedMs: number; + lastEvent?: string; + lastEventAt?: string; + stopReason?: string; + resultAvailable: boolean; + childSession?: string; + error?: string; +} + +export interface SubagentResult { + id: string; + state: SubagentState; + running: boolean; + resultAvailable: boolean; + result?: string; + error?: string; + completedAt?: string; + elapsedMs: number; +} + +export interface ChildRecord { + status: SubagentStatus; + result?: string; +} + +export interface RunnerEvents { + accepted(childSession?: string): void; + running(event: string): void; + settling(): void; + completed(result: string, stopReason?: string): void; + failed(error: string): void; +} + +export interface ChildHandle { + cancel(): Promise; +} + +export interface ChildRunner { + start(id: string, request: SpawnRequest, cwd: string, events: RunnerEvents): Promise; +} -- 2.47.3 From 289ea1344c3b3f87c18720878c2ef18cca8af14c Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 14:13:44 -0400 Subject: [PATCH 2/6] feat(pi): harden subagent lifecycle --- .../agents/pi/extensions/subagents/index.ts | 7 ++ .../agents/pi/extensions/subagents/runner.ts | 44 ++++++- .../extensions/subagents/supervisor.test.ts | 111 ++++++++++++++++++ .../pi/extensions/subagents/supervisor.ts | 87 ++++++++++++-- 4 files changed, 236 insertions(+), 13 deletions(-) create mode 100644 modules/agents/pi/extensions/subagents/supervisor.test.ts diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts index ac63f49..2be89dc 100644 --- a/modules/agents/pi/extensions/subagents/index.ts +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -104,6 +104,13 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerCommand("subagent-cancel", { + description: "Cancel a running subagent by id", + handler: async (args, ctx) => { + ctx.ui.notify(JSON.stringify(await getSupervisor(ctx).cancel(args.trim()), null, 2), "info"); + }, + }); + pi.on("session_shutdown", async () => { await supervisor?.shutdown(); supervisor = undefined; diff --git a/modules/agents/pi/extensions/subagents/runner.ts b/modules/agents/pi/extensions/subagents/runner.ts index 7683827..8c88112 100644 --- a/modules/agents/pi/extensions/subagents/runner.ts +++ b/modules/agents/pi/extensions/subagents/runner.ts @@ -21,6 +21,9 @@ class RpcChildHandle implements ChildHandle { private buffer = ""; private nextRequest = 0; private settled = false; + private finishing = false; + private cancelling = false; + private killed = false; private readonly pending = new Map(); constructor( @@ -46,11 +49,12 @@ class RpcChildHandle implements ChildHandle { } async cancel(): Promise { + if (this.cancelling) return; + this.cancelling = true; try { - await this.send("abort", {}); + await Promise.race([this.send("abort", {}), delay(200)]); } catch {} - this.child.stdin.end(); - if (!this.child.killed) this.child.kill("SIGTERM"); + this.terminate(); } private onStdout(chunk: string) { @@ -97,14 +101,38 @@ class RpcChildHandle implements ChildHandle { } private async finish() { - if (this.settled) return; - this.settled = true; + if (this.settled || this.finishing) return; + this.finishing = true; this.events.settling(); const result = await this.send("get_last_assistant_text", {}); const text = typeof result === "string" ? result : result && typeof result === "object" && "text" in result ? String((result as { text: unknown }).text) : ""; + this.settled = true; this.events.completed(text, "agent_settled"); + this.terminate(); + } + + private terminate() { + if (this.killed) return; + this.killed = true; this.child.stdin.end(); - if (!this.child.killed) this.child.kill("SIGTERM"); + if (this.child.killed) return; + if (process.platform !== "win32" && this.child.pid) { + try { + process.kill(-this.child.pid, "SIGTERM"); + } catch { + this.child.kill("SIGTERM"); + } + setTimeout(() => { + if (this.child.killed || !this.child.pid) return; + try { + process.kill(-this.child.pid, "SIGKILL"); + } catch { + this.child.kill("SIGKILL"); + } + }, 2_000).unref(); + return; + } + this.child.kill("SIGTERM"); } private fail(error: string) { @@ -142,6 +170,10 @@ export class SubprocessRpcRunner implements ChildRunner { } } +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function childEnvironment(): NodeJS.ProcessEnv { const env = { ...process.env }; delete env.PI_SESSION_ID; diff --git a/modules/agents/pi/extensions/subagents/supervisor.test.ts b/modules/agents/pi/extensions/subagents/supervisor.test.ts new file mode 100644 index 0000000..69f1bc8 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/supervisor.test.ts @@ -0,0 +1,111 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { Supervisor } from "./supervisor.ts"; +import type { ChildHandle, ChildRunner, RunnerEvents, SpawnRequest } from "./types.ts"; + +class FakeHandle implements ChildHandle { + cancelCalls = 0; + + async cancel(): Promise { + this.cancelCalls += 1; + } +} + +class FakeRunner implements ChildRunner { + starts: Array<{ id: string; request: SpawnRequest; events: RunnerEvents; handle: FakeHandle }> = []; + autoAccept = true; + + async start(id: string, request: SpawnRequest, _cwd: string, events: RunnerEvents): Promise { + const handle = new FakeHandle(); + this.starts.push({ id, request, events, handle }); + if (this.autoAccept) events.accepted(`session-${id}`); + return handle; + } +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function spawnStarted(supervisor: Supervisor, prompt = "work") { + const accepted = supervisor.spawn({ prompt }); + await sleep(0); + return accepted; +} + +test("cancel is idempotent and reaches cancelled", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor); + + const first = await supervisor.cancel(accepted.id); + const second = await supervisor.cancel(accepted.id); + + assert.equal(first.state, "cancelled"); + assert.equal(second.state, "cancelled"); + assert.equal(runner.starts[0].handle.cancelCalls, 1); +}); + +test("startup timeout reaches timed_out", async () => { + const runner = new FakeRunner(); + runner.autoAccept = false; + const supervisor = new Supervisor(runner, "/tmp", { timeouts: { startMs: 5 } }); + const accepted = await spawnStarted(supervisor); + + await sleep(20); + + const status = supervisor.status(accepted.id); + assert.equal(status.state, "timed_out"); + assert.equal(status.stopReason, "start_timeout"); + assert.equal(runner.starts[0].handle.cancelCalls, 1); +}); + +test("runtime timeout reaches timed_out", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { timeouts: { runMs: 5 } }); + const accepted = await spawnStarted(supervisor); + + await sleep(20); + + const status = supervisor.status(accepted.id); + assert.equal(status.state, "timed_out"); + assert.equal(status.stopReason, "run_timeout"); + assert.equal(runner.starts[0].handle.cancelCalls, 1); +}); + +test("process failure reaches failed with diagnostics", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor); + + runner.starts[0].events.failed("process closed with code 1"); + + const status = supervisor.status(accepted.id); + assert.equal(status.state, "failed"); + assert.equal(status.error, "process closed with code 1"); +}); + +test("shutdown cancels running children", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor); + + await supervisor.shutdown(); + + const status = supervisor.status(accepted.id); + assert.equal(status.state, "cancelled"); + assert.equal(status.stopReason, "shutdown"); + assert.equal(runner.starts[0].handle.cancelCalls, 1); +}); + +test("completed children ignore later cancel", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor); + + runner.starts[0].events.completed("done", "agent_settled"); + await supervisor.cancel(accepted.id); + + const result = supervisor.result(accepted.id); + assert.equal(result.state, "completed"); + assert.equal(result.result, "done"); + assert.equal(runner.starts[0].handle.cancelCalls, 0); +}); diff --git a/modules/agents/pi/extensions/subagents/supervisor.ts b/modules/agents/pi/extensions/subagents/supervisor.ts index 6f45ab7..2129536 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -4,8 +4,22 @@ import { cloneResult, cloneStatus, toAccepted } from "./status.ts"; interface RunningChild { record: ChildRecord; handle?: ChildHandle; + startTimer?: ReturnType; + runTimer?: ReturnType; } +interface SupervisorOptions { + timeouts?: { + startMs?: number; + runMs?: number; + }; +} + +const DEFAULT_TIMEOUTS = { + startMs: 30_000, + runMs: 0, +}; + export class Supervisor { private nextChild = 0; private readonly children = new Map(); @@ -13,6 +27,7 @@ export class Supervisor { constructor( private readonly runner: ChildRunner, private readonly cwd: string, + private readonly options: SupervisorOptions = {}, ) {} spawn(request: SpawnRequest): SpawnAccepted { @@ -41,6 +56,7 @@ export class Supervisor { const child: RunningChild = { record: { status } }; this.children.set(id, child); this.setState(child.record.status, "starting", "starting"); + this.armStartTimer(child); setTimeout(() => { if (isTerminal(child.record.status.state)) return; @@ -48,6 +64,7 @@ export class Supervisor { .start(id, { ...request, prompt, context: status.context, tools: status.tools }, this.cwd, this.eventsFor(child.record)) .then((handle) => { child.handle = handle; + if (isTerminal(child.record.status.state)) void handle.cancel(); }) .catch((error) => { this.fail(child.record, error instanceof Error ? error.message : String(error)); @@ -73,19 +90,17 @@ export class Supervisor { const child = this.require(id); if (isTerminal(child.record.status.state)) return cloneStatus(child.record.status); await child.handle?.cancel(); - const now = new Date().toISOString(); - child.record.status.state = "cancelled"; - child.record.status.completedAt = now; - child.record.status.lastEvent = "cancelled"; - child.record.status.lastEventAt = now; - child.record.status.stopReason = "cancelled"; + 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(); + if (!isTerminal(child.record.status.state)) { + await child.handle?.cancel(); + this.completeWithoutResult(child, "cancelled", "shutdown"); + } }), ); } @@ -93,6 +108,11 @@ export class Supervisor { private eventsFor(record: ChildRecord): RunnerEvents { return { accepted: (childSession) => { + const child = this.findChild(record); + if (child) { + this.clearTimer(child, "startTimer"); + this.armRunTimer(child); + } if (childSession) record.status.childSession = childSession; this.setState(record.status, "running", "prompt accepted"); }, @@ -104,6 +124,8 @@ export class Supervisor { }, completed: (result, stopReason) => { const now = new Date().toISOString(); + const child = this.findChild(record); + if (child) this.clearTimers(child); record.result = result; record.status.state = "completed"; record.status.completedAt = now; @@ -118,6 +140,8 @@ export class Supervisor { private fail(record: ChildRecord, error: string) { if (isTerminal(record.status.state)) return; + const child = this.findChild(record); + if (child) this.clearTimers(child); const now = new Date().toISOString(); record.status.state = "failed"; record.status.completedAt = now; @@ -127,6 +151,55 @@ export class Supervisor { record.status.stopReason = "failed"; } + private completeWithoutResult(child: RunningChild, state: "cancelled" | "timed_out", reason: string) { + if (isTerminal(child.record.status.state)) return; + this.clearTimers(child); + const now = new Date().toISOString(); + child.record.status.state = state; + child.record.status.completedAt = now; + child.record.status.lastEvent = state; + child.record.status.lastEventAt = now; + child.record.status.stopReason = reason; + } + + private armStartTimer(child: RunningChild) { + const timeout = this.options.timeouts?.startMs ?? DEFAULT_TIMEOUTS.startMs; + if (timeout <= 0) return; + child.startTimer = setTimeout(() => { + this.timeout(child, "start_timeout"); + }, timeout); + } + + private armRunTimer(child: RunningChild) { + const timeout = this.options.timeouts?.runMs ?? DEFAULT_TIMEOUTS.runMs; + if (timeout <= 0) return; + child.runTimer = setTimeout(() => { + this.timeout(child, "run_timeout"); + }, timeout); + } + + private timeout(child: RunningChild, reason: string) { + if (isTerminal(child.record.status.state)) return; + void child.handle?.cancel(); + this.completeWithoutResult(child, "timed_out", reason); + } + + private clearTimers(child: RunningChild) { + this.clearTimer(child, "startTimer"); + this.clearTimer(child, "runTimer"); + } + + private clearTimer(child: RunningChild, key: "startTimer" | "runTimer") { + const timer = child[key]; + if (!timer) return; + clearTimeout(timer); + child[key] = undefined; + } + + private findChild(record: ChildRecord): RunningChild | undefined { + return [...this.children.values()].find((child) => child.record === record); + } + private setState(status: SubagentStatus, state: SubagentStatus["state"], event: string) { if (isTerminal(status.state)) return; const now = new Date().toISOString(); -- 2.47.3 From 3977ed682232e8e22e657722fafcc5ab3c5c67b7 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 14:27:36 -0400 Subject: [PATCH 3/6] feat(pi): add subagent config and agents --- .../agents/pi/extensions/subagents/agents.ts | 141 ++++++++++++++++ .../pi/extensions/subagents/config.test.ts | 117 ++++++++++++++ .../agents/pi/extensions/subagents/config.ts | 152 ++++++++++++++++++ .../agents/pi/extensions/subagents/index.ts | 41 ++++- .../agents/pi/extensions/subagents/runner.ts | 19 ++- .../pi/extensions/subagents/supervisor.ts | 4 +- .../agents/pi/extensions/subagents/types.ts | 6 + 7 files changed, 472 insertions(+), 8 deletions(-) create mode 100644 modules/agents/pi/extensions/subagents/agents.ts create mode 100644 modules/agents/pi/extensions/subagents/config.test.ts create mode 100644 modules/agents/pi/extensions/subagents/config.ts diff --git a/modules/agents/pi/extensions/subagents/agents.ts b/modules/agents/pi/extensions/subagents/agents.ts new file mode 100644 index 0000000..f56c1c8 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/agents.ts @@ -0,0 +1,141 @@ +import { existsSync, readdirSync, readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { basename, join } from "node:path"; +import type { ContextMode } from "./types.ts"; +import type { Diagnostics } from "./config.ts"; + +export interface AgentDefinition { + name: string; + description: string; + body: string; + context?: ContextMode; + model?: string; + thinking?: string; + tools?: string; + allowedContexts?: ContextMode[]; + hidden?: boolean; + source: string; +} + +export function loadAgents(cwd: string, projectTrusted: boolean, diagnostics: Diagnostics, agentDir = defaultAgentDir()): Map { + const user = loadTier(join(agentDir, "agents"), "user", diagnostics); + const project = projectTrusted ? loadTier(join(cwd, ".pi", "agents"), "project", diagnostics) : new Map(); + return new Map([...user, ...project]); +} + +function loadTier(dir: string, tier: string, diagnostics: Diagnostics): Map { + const agents = new Map(); + if (!existsSync(dir)) return agents; + for (const entry of readdirSync(dir, { withFileTypes: true })) { + if (!entry.isFile() || !entry.name.endsWith(".md")) continue; + const path = join(dir, entry.name); + const parsed = parseAgent(path, diagnostics); + if (!parsed) continue; + if (agents.has(parsed.name)) { + diagnostics.warnings.push(`Duplicate ${tier} agent '${parsed.name}' ignored at ${path}`); + continue; + } + const stem = basename(entry.name, ".md"); + if (stem !== parsed.name) diagnostics.warnings.push(`${tier} agent file '${entry.name}' name '${parsed.name}' does not match filename`); + agents.set(parsed.name, parsed); + } + return agents; +} + +export function parseAgent(path: string, diagnostics: Diagnostics): AgentDefinition | undefined { + try { + const text = readFileSync(path, "utf8"); + const match = /^---\n([\s\S]*?)\n---\n?([\s\S]*)$/u.exec(text); + if (!match) { + diagnostics.warnings.push(`Agent ${path} missing YAML frontmatter`); + return undefined; + } + const frontmatter = parseFrontmatter(match[1]); + const name = stringField(frontmatter, "name"); + const description = stringField(frontmatter, "description"); + if (!name || !/^[a-z0-9-]+$/.test(name)) { + diagnostics.warnings.push(`Agent ${path} has invalid name`); + return undefined; + } + if (!description) { + diagnostics.warnings.push(`Agent ${path} has invalid description`); + return undefined; + } + const context = contextField(frontmatter.context); + const allowedContexts = contextsField(frontmatter.allowedContexts); + if (frontmatter.context !== undefined && !context) diagnostics.warnings.push(`Agent ${path} has invalid context`); + if (frontmatter.allowedContexts !== undefined && !allowedContexts) diagnostics.warnings.push(`Agent ${path} has invalid allowedContexts`); + if (context && allowedContexts && !allowedContexts.includes(context)) diagnostics.warnings.push(`Agent ${path} context is outside allowedContexts`); + return { + name, + description, + body: match[2].trim(), + context, + model: stringField(frontmatter, "model"), + thinking: stringField(frontmatter, "thinking"), + tools: stringField(frontmatter, "tools"), + allowedContexts, + hidden: booleanField(frontmatter, "hidden"), + source: path, + }; + } catch (error) { + diagnostics.warnings.push(`Failed to load agent ${path}: ${error instanceof Error ? error.message : String(error)}`); + return undefined; + } +} + +function parseFrontmatter(text: string): Record { + const result: Record = {}; + const lines = text.split(/\r?\n/u); + for (let i = 0; i < lines.length; i += 1) { + const line = lines[i]; + if (!line.trim() || line.trimStart().startsWith("#")) continue; + const scalar = /^(\w+):\s*(.*?)\s*$/u.exec(line); + if (!scalar) continue; + const [, key, raw] = scalar; + if (raw !== "") { + result[key] = parseScalar(raw); + continue; + } + const values: string[] = []; + while (i + 1 < lines.length) { + const item = /^\s+-\s*(.*?)\s*$/u.exec(lines[i + 1]); + if (!item) break; + values.push(String(parseScalar(item[1]))); + i += 1; + } + result[key] = values; + } + return result; +} + +function parseScalar(raw: string): string | boolean { + const unquoted = raw.replace(/^['"]|['"]$/gu, ""); + if (unquoted === "true") return true; + if (unquoted === "false") return false; + return unquoted; +} + +function stringField(record: Record, key: string): string | undefined { + const value = record[key]; + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function booleanField(record: Record, key: string): boolean | undefined { + const value = record[key]; + return typeof value === "boolean" ? value : undefined; +} + +function contextField(value: unknown): ContextMode | undefined { + return value === "independent" || value === "fork" ? value : undefined; +} + +function contextsField(value: unknown): ContextMode[] | undefined { + if (!Array.isArray(value)) return undefined; + const contexts = value.map(contextField); + return contexts.every(Boolean) ? (contexts as ContextMode[]) : undefined; +} + +function defaultAgentDir(): string { + return process.env.PI_CODING_AGENT_DIR ?? join(homedir(), ".pi", "agent"); +} diff --git a/modules/agents/pi/extensions/subagents/config.test.ts b/modules/agents/pi/extensions/subagents/config.test.ts new file mode 100644 index 0000000..e292df3 --- /dev/null +++ b/modules/agents/pi/extensions/subagents/config.test.ts @@ -0,0 +1,117 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { loadAgents } from "./agents.ts"; +import { BUILT_IN_TOOL_PROFILES, loadConfig, resolveSpawn, type Diagnostics } from "./config.ts"; + +function fixture() { + const root = mkdtempSync(join(tmpdir(), "subagents-config-")); + const agentDir = join(root, "agent"); + const cwd = join(root, "project"); + mkdirSync(agentDir, { recursive: true }); + mkdirSync(cwd, { recursive: true }); + return { root, agentDir, cwd }; +} + +function diagnostics(): Diagnostics { + return { warnings: [] }; +} + +test("missing config files and agent directories are normal", () => { + const { cwd, agentDir } = fixture(); + const diag = diagnostics(); + + const config = loadConfig(cwd, true, diag, agentDir); + const agents = loadAgents(cwd, true, diag, agentDir); + + assert.equal(config.defaultContext, "independent"); + assert.equal(config.defaultTools, "read-only"); + assert.equal(agents.size, 0); + assert.deepEqual(diag.warnings, []); +}); + +test("global and trusted project config merge in order", () => { + const { cwd, agentDir } = fixture(); + mkdirSync(join(cwd, ".pi"), { recursive: true }); + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "global-profile", toolProfiles: { "global-profile": { activeTools: ["read"] } } })); + writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", toolProfiles: { "project-profile": { activeTools: ["ls"] } } })); + + const config = loadConfig(cwd, true, diagnostics(), agentDir); + + assert.equal(config.defaultTools, "project-profile"); + assert.deepEqual(config.toolProfiles["global-profile"].activeTools, ["read"]); + assert.deepEqual(config.toolProfiles["project-profile"].activeTools, ["ls"]); +}); + +test("project config is ignored when project is untrusted", () => { + const { cwd, agentDir } = fixture(); + mkdirSync(join(cwd, ".pi"), { recursive: true }); + writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", toolProfiles: { "project-profile": { activeTools: ["ls"] } } })); + + const config = loadConfig(cwd, false, diagnostics(), agentDir); + + assert.equal(config.defaultTools, "read-only"); + assert.equal(config.toolProfiles["project-profile"], undefined); +}); + +test("agents load with project precedence over user", () => { + const { cwd, agentDir } = fixture(); + mkdirSync(join(agentDir, "agents"), { recursive: true }); + mkdirSync(join(cwd, ".pi", "agents"), { recursive: true }); + writeFileSync(join(agentDir, "agents", "review.md"), "---\nname: review\ndescription: User review\ntools: read-only\n---\nuser body\n"); + writeFileSync(join(cwd, ".pi", "agents", "review.md"), "---\nname: review\ndescription: Project review\ntools: full-tools\n---\nproject body\n"); + + const agents = loadAgents(cwd, true, diagnostics(), agentDir); + + assert.equal(agents.get("review")?.description, "Project review"); + assert.equal(agents.get("review")?.body, "project body"); +}); + +test("duplicate same-tier definitions and invalid frontmatter produce diagnostics", () => { + const { cwd, agentDir } = fixture(); + const dir = join(agentDir, "agents"); + mkdirSync(dir, { recursive: true }); + writeFileSync(join(dir, "one.md"), "---\nname: same\ndescription: One\n---\none\n"); + writeFileSync(join(dir, "two.md"), "---\nname: same\ndescription: Two\n---\ntwo\n"); + writeFileSync(join(dir, "bad.md"), "---\nname: Bad Name\n---\nbad\n"); + const diag = diagnostics(); + + const agents = loadAgents(cwd, true, diag, agentDir); + + assert.equal(agents.size, 1); + assert.ok(diag.warnings.some((warning) => warning.includes("Duplicate user agent 'same'"))); + assert.ok(diag.warnings.some((warning) => warning.includes("invalid name"))); +}); + +test("named spawn resolves overrides, frontmatter, config, and defaults", () => { + const { cwd, agentDir } = fixture(); + mkdirSync(join(agentDir, "agents"), { recursive: true }); + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "local-review", toolProfiles: { "local-review": { activeTools: ["read"] } } })); + writeFileSync(join(agentDir, "agents", "review.md"), "---\nname: review\ndescription: Review\ncontext: independent\nmodel: inherit\nthinking: high\ntools: local-review\n---\nagent body\n"); + const diag = diagnostics(); + const config = loadConfig(cwd, true, diag, agentDir); + const agents = loadAgents(cwd, true, diag, agentDir); + + const resolved = resolveSpawn({ agent: "review", prompt: "check this", thinking: "low" }, config, agents); + + assert.equal(resolved.prompt, "check this"); + assert.equal(resolved.context, "independent"); + assert.equal(resolved.model, "inherit"); + assert.equal(resolved.thinking, "low"); + assert.equal(resolved.tools, "local-review"); + assert.deepEqual(resolved.toolProfile.activeTools, ["read"]); + assert.equal(resolved.agentBody, "agent body"); +}); + +test("built-in tool profile names cannot be overridden", () => { + const { cwd, agentDir } = fixture(); + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ toolProfiles: { "read-only": { activeTools: ["bash"] } } })); + const diag = diagnostics(); + + const config = loadConfig(cwd, true, diag, agentDir); + + assert.deepEqual(config.toolProfiles["read-only"], BUILT_IN_TOOL_PROFILES["read-only"]); + assert.ok(diag.warnings.some((warning) => warning.includes("Ignoring global override for built-in tool profile 'read-only'"))); +}); diff --git a/modules/agents/pi/extensions/subagents/config.ts b/modules/agents/pi/extensions/subagents/config.ts new file mode 100644 index 0000000..836793a --- /dev/null +++ b/modules/agents/pi/extensions/subagents/config.ts @@ -0,0 +1,152 @@ +import { existsSync, readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import type { ContextMode, SpawnRequest, ToolProfile } from "./types.ts"; +import type { AgentDefinition } from "./agents.ts"; + +export interface Diagnostics { + warnings: string[]; +} + +export interface SubagentsConfig { + defaultContext: ContextMode; + defaultTools: string; + toolProfiles: Record; +} + +export interface ResolvedSpawnRequest extends SpawnRequest { + prompt: string; + context: ContextMode; + tools: string; + toolProfile: ToolProfile; + agentBody?: string; +} + +export const BUILT_IN_TOOL_PROFILES: Record = { + none: { activeTools: [] }, + "read-only": { activeTools: ["read", "grep", "find", "ls"] }, + "read-only-with-safe-bash": { activeTools: ["read", "grep", "find", "ls", "bash"] }, + "full-tools": { activeTools: null }, +}; + +const DEFAULT_CONFIG: SubagentsConfig = { + defaultContext: "independent", + defaultTools: "read-only", + toolProfiles: { ...BUILT_IN_TOOL_PROFILES }, +}; + +export function loadConfig(cwd: string, projectTrusted: boolean, diagnostics: Diagnostics, agentDir = defaultAgentDir()): SubagentsConfig { + let config = cloneConfig(DEFAULT_CONFIG); + config = mergeConfig(config, readConfig(join(agentDir, "subagents.json"), diagnostics, "global"), diagnostics, "global"); + if (projectTrusted) { + config = mergeConfig(config, readConfig(join(cwd, ".pi", "subagents.json"), diagnostics, "project"), diagnostics, "project"); + } + if (!config.toolProfiles[config.defaultTools]) { + diagnostics.warnings.push(`Unknown defaultTools profile '${config.defaultTools}', using read-only`); + config.defaultTools = "read-only"; + } + return config; +} + +export function resolveSpawn(request: SpawnRequest, config: SubagentsConfig, agents: Map): ResolvedSpawnRequest { + const prompt = typeof request.prompt === "string" ? request.prompt.trim() : ""; + if (!prompt) throw new Error("prompt is required"); + const agent = request.agent ? agents.get(request.agent) : undefined; + if (request.agent && !agent) throw new Error(`unknown subagent agent: ${request.agent}`); + + const context = request.context ?? agent?.context ?? config.defaultContext; + if (context !== "independent" && context !== "fork") throw new Error(`unsupported context: ${context}`); + if (agent?.allowedContexts && !agent.allowedContexts.includes(context)) { + throw new Error(`agent '${agent.name}' does not allow ${context} context`); + } + + const tools = request.tools ?? agent?.tools ?? config.defaultTools; + const toolProfile = config.toolProfiles[tools]; + if (!toolProfile) throw new Error(`unknown tool profile: ${tools}`); + + return { + ...request, + prompt, + agent: agent?.name ?? request.agent, + context, + model: request.model ?? agent?.model, + thinking: request.thinking ?? agent?.thinking, + tools, + toolProfile, + agentBody: agent?.body, + }; +} + +function readConfig(path: string, diagnostics: Diagnostics, label: string): Partial | undefined { + if (!existsSync(path)) return undefined; + try { + const parsed = JSON.parse(readFileSync(path, "utf8")); + return normalizeConfig(parsed, diagnostics, label); + } catch (error) { + diagnostics.warnings.push(`Invalid ${label} subagents.json: ${error instanceof Error ? error.message : String(error)}`); + return undefined; + } +} + +function normalizeConfig(raw: unknown, diagnostics: Diagnostics, label: string): Partial | undefined { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + diagnostics.warnings.push(`Invalid ${label} subagents.json: root must be an object`); + return undefined; + } + const input = raw as Record; + const config: Partial = {}; + if (input.defaultContext === "independent" || input.defaultContext === "fork") config.defaultContext = input.defaultContext; + 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 (input.toolProfiles !== undefined) config.toolProfiles = normalizeProfiles(input.toolProfiles, diagnostics, label); + return config; +} + +function normalizeProfiles(raw: unknown, diagnostics: Diagnostics, label: string): Record { + const profiles: Record = {}; + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + diagnostics.warnings.push(`Invalid ${label} toolProfiles ignored`); + return profiles; + } + for (const [name, value] of Object.entries(raw as Record)) { + if (name in BUILT_IN_TOOL_PROFILES) { + diagnostics.warnings.push(`Ignoring ${label} override for built-in tool profile '${name}'`); + continue; + } + const profile = normalizeProfile(value); + if (!profile) { + diagnostics.warnings.push(`Invalid ${label} tool profile '${name}' ignored`); + continue; + } + profiles[name] = profile; + } + return profiles; +} + +function normalizeProfile(raw: unknown): ToolProfile | undefined { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined; + const activeTools = (raw as { activeTools?: unknown }).activeTools; + if (!Array.isArray(activeTools) || !activeTools.every((tool) => typeof tool === "string")) return undefined; + return { activeTools }; +} + +function mergeConfig(base: SubagentsConfig, override: Partial | undefined, diagnostics: Diagnostics, label: string): SubagentsConfig { + if (!override) return base; + const merged = cloneConfig(base); + if (override.defaultContext) merged.defaultContext = override.defaultContext; + if (override.defaultTools) merged.defaultTools = override.defaultTools; + 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]; + } + return merged; +} + +function cloneConfig(config: SubagentsConfig): SubagentsConfig { + return { ...config, toolProfiles: { ...config.toolProfiles } }; +} + +function defaultAgentDir(): string { + return process.env.PI_CODING_AGENT_DIR ?? join(homedir(), ".pi", "agent"); +} diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts index 2be89dc..3643bc8 100644 --- a/modules/agents/pi/extensions/subagents/index.ts +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -1,10 +1,13 @@ import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; +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"; let supervisor: Supervisor | undefined; +let lastDiagnostics: Diagnostics = { warnings: [] }; export default function subagents(pi: ExtensionAPI) { const getSupervisor = (ctx: ExtensionContext): Supervisor => { @@ -12,18 +15,30 @@ export default function subagents(pi: ExtensionAPI) { return supervisor; }; + const resolve = (ctx: ExtensionContext, request: SpawnRequest): SpawnRequest => { + const diagnostics: Diagnostics = { warnings: [] }; + const cwd = cwdOf(ctx); + const trusted = isProjectTrusted(ctx); + const config = loadConfig(cwd, trusted, diagnostics); + const agents = loadAgents(cwd, trusted, diagnostics); + lastDiagnostics = diagnostics; + return resolveSpawn(request, config, agents); + }; + pi.registerTool({ name: "subagent_spawn", label: "Spawn subagent", description: "Start one ad hoc independent subagent and return immediately with its child id", 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")), - model: Type.Optional(Type.String({ description: "Optional model selector for status metadata" })), - thinking: Type.Optional(Type.String({ description: "Optional thinking level for status metadata" })), + 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 accepted = getSupervisor(ctx).spawn(params as SpawnRequest); + const accepted = getSupervisor(ctx).spawn(resolve(ctx, params as SpawnRequest)); ctx.ui?.notify?.(`Started subagent ${accepted.id}`, "info"); return textResult(accepted); }, @@ -78,7 +93,7 @@ export default function subagents(pi: ExtensionAPI) { pi.registerCommand("subagent-spawn", { description: "Start an ad hoc independent subagent", handler: async (args, ctx) => { - const accepted = getSupervisor(ctx).spawn({ prompt: args }); + const accepted = getSupervisor(ctx).spawn(resolve(ctx, parseSpawnArgs(args))); ctx.ui.notify(`Started subagent ${accepted.id}`, "info"); }, }); @@ -104,6 +119,13 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerCommand("subagent-diagnostics", { + description: "Show subagent configuration diagnostics from the last load", + handler: async (_args, ctx) => { + ctx.ui.notify(JSON.stringify(lastDiagnostics, null, 2), "info"); + }, + }); + pi.registerCommand("subagent-cancel", { description: "Cancel a running subagent by id", handler: async (args, ctx) => { @@ -117,6 +139,17 @@ export default function subagents(pi: ExtensionAPI) { }); } +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] }; +} + +function isProjectTrusted(ctx: ExtensionContext): boolean { + const value = (ctx as unknown as { isProjectTrusted?: () => boolean }).isProjectTrusted?.(); + return value === true; +} + function cwdOf(ctx: ExtensionContext): string { const sessionCwd = (ctx as unknown as { sessionManager?: { getCwd?: () => string }; cwd?: string }).sessionManager?.getCwd?.(); return sessionCwd ?? (ctx as unknown as { cwd?: string }).cwd ?? process.cwd(); diff --git a/modules/agents/pi/extensions/subagents/runner.ts b/modules/agents/pi/extensions/subagents/runner.ts index 8c88112..8edeaaa 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}`]; + const args = [process.argv[1], "--mode", "rpc", "--no-extensions", "--name", `subagent ${id}`, ...toolArgs(request), ...modelArgs(request)]; const child = spawn(process.execPath, args, { cwd, env: childEnvironment(), @@ -174,6 +174,20 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +function toolArgs(request: SpawnRequest): string[] { + const activeTools = request.toolProfile?.activeTools; + if (activeTools === undefined || activeTools === null) return []; + if (activeTools.length === 0) return ["--no-tools"]; + return ["--tools", activeTools.join(",")]; +} + +function modelArgs(request: SpawnRequest): string[] { + const args: string[] = []; + if (request.model && request.model !== "inherit") args.push("--model", request.model); + if (request.thinking) args.push("--thinking", request.thinking); + return args; +} + function childEnvironment(): NodeJS.ProcessEnv { const env = { ...process.env }; delete env.PI_SESSION_ID; @@ -185,5 +199,6 @@ function childEnvironment(): NodeJS.ProcessEnv { } function independentPrompt(request: SpawnRequest): string { - return `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}`; + const base = request.agentBody ? `${request.agentBody}\n\n` : ""; + 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.ts b/modules/agents/pi/extensions/subagents/supervisor.ts index 2129536..fe42838 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -46,7 +46,7 @@ export class Supervisor { cwd: this.cwd, model: request.model, thinking: request.thinking, - tools: "pi-default", + tools: request.tools ?? "read-only", startedAt: now, elapsedMs: 0, lastEvent: "queued", @@ -216,7 +216,7 @@ 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 in this tracer bullet"); + if (context !== "independent") throw new Error("only independent context is implemented before fork mode lands"); return context; } diff --git a/modules/agents/pi/extensions/subagents/types.ts b/modules/agents/pi/extensions/subagents/types.ts index 4380099..209a098 100644 --- a/modules/agents/pi/extensions/subagents/types.ts +++ b/modules/agents/pi/extensions/subagents/types.ts @@ -11,6 +11,10 @@ export type SubagentState = | "timed_out" | "orphaned"; +export interface ToolProfile { + activeTools: string[] | null; +} + export interface SpawnRequest { prompt: string; context?: ContextMode; @@ -18,6 +22,8 @@ export interface SpawnRequest { model?: string; thinking?: string; tools?: string; + toolProfile?: ToolProfile; + agentBody?: string; } export interface SpawnAccepted { -- 2.47.3 From 007ba81c02842f7d47a01fe8288b9c4fc4ff2ab5 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 14:44:07 -0400 Subject: [PATCH 4/6] 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)); + }, + }); +} -- 2.47.3 From 3c4eaec76bfc45d2881e005f98eb0d175ba8ff58 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 16:30:06 -0400 Subject: [PATCH 5/6] feat(pi): add subagent wait tool --- .../agents/pi/extensions/subagents/index.ts | 46 +++++++++ .../extensions/subagents/supervisor.test.ts | 95 +++++++++++++++++++ .../pi/extensions/subagents/supervisor.ts | 83 +++++++++++++++- .../agents/pi/extensions/subagents/types.ts | 12 +++ 4 files changed, 235 insertions(+), 1 deletion(-) diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts index 7394769..5ea2958 100644 --- a/modules/agents/pi/extensions/subagents/index.ts +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -129,6 +129,24 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerTool({ + name: "subagent_wait", + label: "Wait for subagents", + description: "Block until multiple subagents are terminal or a timeout expires. Prefer setting timeoutMs so the parent turn cannot hang forever", + parameters: Type.Object({ + ids: Type.Array(Type.String({ description: "Subagent id returned by subagent_spawn or subagent_batch" })), + timeoutMs: Type.Optional(Type.Number({ description: "Maximum milliseconds to wait. Omit or use 0 to wait indefinitely" })), + mode: Type.Optional(Type.Union([Type.Literal("all"), Type.Literal("any")], { description: "Wait for all ids by default, or return after any id is terminal" })), + }), + async execute(_toolCallId, params, signal, _onUpdate, ctx) { + const input = params as { ids?: unknown; timeoutMs?: unknown; mode?: unknown }; + const ids = Array.isArray(input.ids) ? input.ids.map(String) : []; + const timeoutMs = typeof input.timeoutMs === "number" && Number.isFinite(input.timeoutMs) ? input.timeoutMs : undefined; + const mode = input.mode === "any" ? "any" : "all"; + return textResult(await getSupervisor(ctx).wait(ids, { timeoutMs, mode, signal })); + }, + }); + pi.registerTool({ name: "subagent_cancel", label: "Cancel subagent", @@ -182,6 +200,14 @@ export default function subagents(pi: ExtensionAPI) { }, }); + pi.registerCommand("subagent-wait", { + description: "Wait for subagent ids separated by spaces", + handler: async (args, ctx) => { + const { ids, timeoutMs, mode } = parseWaitArgs(args); + ctx.ui.notify(JSON.stringify(await getSupervisor(ctx).wait(ids, { timeoutMs, mode }), null, 2), "info"); + }, + }); + pi.registerCommand("subagent-ui", { description: "Toggle the bundled subagent status inspector", handler: async (_args, ctx) => { @@ -231,6 +257,26 @@ function parseSpawnArgs(args: string): SpawnRequest { return { ...request, prompt: parts.join(" ") || args } as SpawnRequest; } +function parseWaitArgs(args: string): { ids: string[]; timeoutMs?: number; mode?: "all" | "any" } { + const parts = args.trim().split(/\s+/u).filter(Boolean); + let timeoutMs: number | undefined; + let mode: "all" | "any" | undefined; + const ids: string[] = []; + while (parts.length > 0) { + const part = parts.shift(); + if (!part) continue; + if (part === "--timeout-ms" && parts[0]) { + const parsed = Number(parts.shift()); + if (Number.isFinite(parsed)) timeoutMs = parsed; + } else if (part === "--mode" && (parts[0] === "all" || parts[0] === "any")) { + mode = parts.shift() as "all" | "any"; + } else { + ids.push(part); + } + } + return { ids, timeoutMs, mode }; +} + function isProjectTrusted(ctx: ExtensionContext): boolean { const value = (ctx as unknown as { isProjectTrusted?: () => boolean }).isProjectTrusted?.(); return value === true; diff --git a/modules/agents/pi/extensions/subagents/supervisor.test.ts b/modules/agents/pi/extensions/subagents/supervisor.test.ts index 5eba4ea..7cff49e 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.test.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.test.ts @@ -139,3 +139,98 @@ test("maxConcurrent preserves queued records", async () => { assert.equal(runner.starts.length, 2); }); + +test("wait blocks until multiple subagents are terminal", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const first = await spawnStarted(supervisor, "one"); + const second = await spawnStarted(supervisor, "two"); + + const waiting = supervisor.wait([first.id, second.id], { timeoutMs: 100 }); + runner.starts[0].events.completed("one done", "agent_settled"); + await sleep(0); + + assert.equal(await Promise.race([waiting.then(() => "done"), sleep(10).then(() => "pending")]), "pending"); + + runner.starts[1].events.failed("two failed"); + const result = await waiting; + + assert.equal(result.timedOut, false); + assert.equal(result.ready, true); + assert.deepEqual(result.ids, [first.id, second.id]); + assert.equal(result.pending.length, 0); + assert.deepEqual(result.results.map((item) => item.state), ["completed", "failed"]); + assert.equal(result.results[0].result, "one done"); + assert.equal(result.results[1].error, "two failed"); +}); + +test("wait returns pending statuses on timeout", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const first = await spawnStarted(supervisor, "one"); + const second = await spawnStarted(supervisor, "two"); + + runner.starts[0].events.completed("one done", "agent_settled"); + const result = await supervisor.wait([first.id, second.id], { timeoutMs: 5 }); + + assert.equal(result.timedOut, true); + assert.equal(result.ready, false); + assert.deepEqual(result.results.map((item) => item.state), ["completed", "running"]); + assert.deepEqual(result.pending.map((item) => item.id), [second.id]); +}); + +test("wait any returns after the first terminal subagent", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const first = await spawnStarted(supervisor, "one"); + const second = await spawnStarted(supervisor, "two"); + + const waiting = supervisor.wait([first.id, second.id], { mode: "any", timeoutMs: 100 }); + runner.starts[1].events.completed("two done", "agent_settled"); + const result = await waiting; + + assert.equal(result.timedOut, false); + assert.equal(result.ready, true); + assert.deepEqual(result.results.map((item) => item.state), ["running", "completed"]); + assert.deepEqual(result.pending.map((item) => item.id), [first.id]); +}); + +test("wait rejects unknown and empty id sets", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + + await assert.rejects(() => supervisor.wait([]), /at least one subagent id is required/); + await assert.rejects(() => supervisor.wait(["missing"]), /unknown subagent id: missing/); +}); + +test("wait abort rejects without cancelling child", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp"); + const accepted = await spawnStarted(supervisor, "one"); + const controller = new AbortController(); + + const waiting = supervisor.wait([accepted.id], { signal: controller.signal }); + controller.abort(); + + await assert.rejects(waiting, /subagent wait aborted/); + assert.equal(runner.starts[0].handle.cancelCalls, 0); +}); + +test("wait follows queued subagents through queue start and completion", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { maxConcurrent: 1 }); + const batch = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "two" }]); + await sleep(0); + + const waiting = supervisor.wait([batch.accepted[1].id], { timeoutMs: 100 }); + assert.equal(await Promise.race([waiting.then(() => "done"), sleep(10).then(() => "pending")]), "pending"); + + runner.starts[0].events.completed("one done", "agent_settled"); + await sleep(0); + runner.starts[1].events.completed("two done", "agent_settled"); + const result = await waiting; + + assert.equal(result.timedOut, false); + assert.equal(result.ready, true); + assert.deepEqual(result.results.map((item) => item.result), ["two done"]); +}); diff --git a/modules/agents/pi/extensions/subagents/supervisor.ts b/modules/agents/pi/extensions/subagents/supervisor.ts index 86f3b32..743bfde 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -1,4 +1,16 @@ -import type { ChildHandle, ChildRecord, ChildRunner, ContextMode, RunnerEvents, SpawnAccepted, SpawnRequest, SubagentResult, SubagentStatus } from "./types.ts"; +import type { + ChildHandle, + ChildRecord, + ChildRunner, + ContextMode, + RunnerEvents, + SpawnAccepted, + SpawnRequest, + SubagentResult, + SubagentStatus, + SubagentWaitMode, + SubagentWaitResult, +} from "./types.ts"; import { cloneResult, cloneStatus, toAccepted } from "./status.ts"; interface RunningChild { @@ -34,6 +46,7 @@ export class Supervisor { private nextChild = 0; private readonly children = new Map(); private readonly queue: RunningChild[] = []; + private readonly waiters = new Set<() => void>(); constructor( private readonly runner: ChildRunner, @@ -76,6 +89,41 @@ export class Supervisor { return cloneResult(this.require(id).record); } + async wait( + ids: string[], + options: { timeoutMs?: number; signal?: AbortSignal; mode?: SubagentWaitMode } = {}, + ): Promise { + const uniqueIds = [...new Set(ids.map((id) => id.trim()).filter(Boolean))]; + if (uniqueIds.length === 0) throw new Error("at least one subagent id is required"); + for (const id of uniqueIds) this.require(id); + + const startedAt = Date.now(); + const mode = options.mode ?? "all"; + if (mode !== "all" && mode !== "any") throw new Error(`unknown wait mode: ${mode}`); + const deadline = options.timeoutMs && options.timeoutMs > 0 ? startedAt + options.timeoutMs : undefined; + let timedOut = false; + + while (!this.waitReady(uniqueIds, mode)) { + if (options.signal?.aborted) throw new Error("subagent wait aborted"); + const remainingMs = deadline === undefined ? undefined : deadline - Date.now(); + if (remainingMs !== undefined && remainingMs <= 0) { + timedOut = true; + break; + } + await this.nextChange(remainingMs, options.signal).catch((error) => { + if (error instanceof Error && error.message === "subagent wait timed out") timedOut = true; + else throw error; + }); + if (timedOut) break; + } + + const results = uniqueIds.map((id) => this.result(id)); + const pending = uniqueIds + .map((id) => this.status(id)) + .filter((status) => !isTerminal(status.state)); + return { ids: uniqueIds, mode, ready: this.waitReady(uniqueIds, mode), results, pending, timedOut, elapsedMs: Date.now() - startedAt }; + } + async cancel(id: string): Promise { const child = this.require(id); if (isTerminal(child.record.status.state)) return cloneStatus(child.record.status); @@ -291,6 +339,39 @@ export class Supervisor { private emitChange() { this.options.onChange?.(this.list()); + for (const waiter of this.waiters) waiter(); + } + + private waitReady(ids: string[], mode: SubagentWaitMode): boolean { + const terminal = (id: string) => isTerminal(this.require(id).record.status.state); + return mode === "all" ? ids.every(terminal) : ids.some(terminal); + } + + private nextChange(timeoutMs: number | undefined, signal: AbortSignal | undefined): Promise { + return new Promise((resolve, reject) => { + let timer: ReturnType | undefined; + const cleanup = () => { + this.waiters.delete(resolveOnce); + if (timer) clearTimeout(timer); + signal?.removeEventListener("abort", abort); + }; + const resolveOnce = () => { + cleanup(); + resolve(); + }; + const abort = () => { + cleanup(); + reject(new Error("subagent wait aborted")); + }; + this.waiters.add(resolveOnce); + signal?.addEventListener("abort", abort, { once: true }); + if (timeoutMs !== undefined) { + timer = setTimeout(() => { + cleanup(); + reject(new Error("subagent wait timed out")); + }, timeoutMs); + } + }); } private allocateId(): string { diff --git a/modules/agents/pi/extensions/subagents/types.ts b/modules/agents/pi/extensions/subagents/types.ts index ac6f407..14a5143 100644 --- a/modules/agents/pi/extensions/subagents/types.ts +++ b/modules/agents/pi/extensions/subagents/types.ts @@ -69,6 +69,18 @@ export interface SubagentResult { elapsedMs: number; } +export type SubagentWaitMode = "all" | "any"; + +export interface SubagentWaitResult { + ids: string[]; + mode: SubagentWaitMode; + ready: boolean; + results: SubagentResult[]; + pending: SubagentStatus[]; + timedOut: boolean; + elapsedMs: number; +} + export interface ChildRecord { status: SubagentStatus; result?: string; -- 2.47.3 From e143495d6c0dce8f03576ed276aba766196ea1d6 Mon Sep 17 00:00:00 2001 From: alexion Date: Sat, 1 Aug 2026 16:47:34 -0400 Subject: [PATCH 6/6] feat(pi): expire recent subagent statuses --- .../pi/extensions/subagents/config.test.ts | 25 ++++++- .../agents/pi/extensions/subagents/config.ts | 6 ++ .../agents/pi/extensions/subagents/index.ts | 1 + .../extensions/subagents/supervisor.test.ts | 70 +++++++++++++++++++ .../pi/extensions/subagents/supervisor.ts | 38 ++++++++++ 5 files changed, 138 insertions(+), 2 deletions(-) diff --git a/modules/agents/pi/extensions/subagents/config.test.ts b/modules/agents/pi/extensions/subagents/config.test.ts index e292df3..679cc91 100644 --- a/modules/agents/pi/extensions/subagents/config.test.ts +++ b/modules/agents/pi/extensions/subagents/config.test.ts @@ -28,6 +28,7 @@ test("missing config files and agent directories are normal", () => { assert.equal(config.defaultContext, "independent"); assert.equal(config.defaultTools, "read-only"); + assert.equal(config.recentTerminalTtlMs, 300000); assert.equal(agents.size, 0); assert.deepEqual(diag.warnings, []); }); @@ -35,16 +36,36 @@ test("missing config files and agent directories are normal", () => { test("global and trusted project config merge in order", () => { const { cwd, agentDir } = fixture(); mkdirSync(join(cwd, ".pi"), { recursive: true }); - writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "global-profile", toolProfiles: { "global-profile": { activeTools: ["read"] } } })); - writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", toolProfiles: { "project-profile": { activeTools: ["ls"] } } })); + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "global-profile", recentTerminalTtlMs: 1000, toolProfiles: { "global-profile": { activeTools: ["read"] } } })); + writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", recentTerminalTtlMs: 2000, toolProfiles: { "project-profile": { activeTools: ["ls"] } } })); const config = loadConfig(cwd, true, diagnostics(), agentDir); assert.equal(config.defaultTools, "project-profile"); + assert.equal(config.recentTerminalTtlMs, 2000); assert.deepEqual(config.toolProfiles["global-profile"].activeTools, ["read"]); assert.deepEqual(config.toolProfiles["project-profile"].activeTools, ["ls"]); }); +test("recent terminal ttl preserves zero and rejects invalid values", () => { + const { cwd, agentDir } = fixture(); + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ recentTerminalTtlMs: 0 })); + const zeroDiag = diagnostics(); + + const zeroConfig = loadConfig(cwd, true, zeroDiag, agentDir); + + assert.equal(zeroConfig.recentTerminalTtlMs, 0); + assert.deepEqual(zeroDiag.warnings, []); + + writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ recentTerminalTtlMs: -1 })); + const invalidDiag = diagnostics(); + + const invalidConfig = loadConfig(cwd, true, invalidDiag, agentDir); + + assert.equal(invalidConfig.recentTerminalTtlMs, 300000); + assert.ok(invalidDiag.warnings.some((warning) => warning.includes("Invalid global recentTerminalTtlMs ignored"))); +}); + test("project config is ignored when project is untrusted", () => { const { cwd, agentDir } = fixture(); mkdirSync(join(cwd, ".pi"), { recursive: true }); diff --git a/modules/agents/pi/extensions/subagents/config.ts b/modules/agents/pi/extensions/subagents/config.ts index 3fc6cf1..36d2785 100644 --- a/modules/agents/pi/extensions/subagents/config.ts +++ b/modules/agents/pi/extensions/subagents/config.ts @@ -12,6 +12,7 @@ export interface SubagentsConfig { defaultContext: ContextMode; defaultTools: string; maxConcurrent: number; + recentTerminalTtlMs: number; ui: { enabled: boolean; defaultExpanded: boolean; @@ -38,6 +39,7 @@ const DEFAULT_CONFIG: SubagentsConfig = { defaultContext: "independent", defaultTools: "read-only", maxConcurrent: 3, + recentTerminalTtlMs: 5 * 60 * 1000, ui: { enabled: true, defaultExpanded: false }, toolProfiles: { ...BUILT_IN_TOOL_PROFILES }, }; @@ -108,6 +110,9 @@ function normalizeConfig(raw: unknown, diagnostics: Diagnostics, label: string): 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 (typeof input.recentTerminalTtlMs === "number" && Number.isInteger(input.recentTerminalTtlMs) && input.recentTerminalTtlMs >= 0) { + config.recentTerminalTtlMs = input.recentTerminalTtlMs; + } else if (input.recentTerminalTtlMs !== undefined) diagnostics.warnings.push(`Invalid ${label} recentTerminalTtlMs 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; @@ -159,6 +164,7 @@ function mergeConfig(base: SubagentsConfig, override: Partial | if (override.defaultContext) merged.defaultContext = override.defaultContext; if (override.defaultTools) merged.defaultTools = override.defaultTools; if (override.maxConcurrent) merged.maxConcurrent = override.maxConcurrent; + if (override.recentTerminalTtlMs !== undefined) merged.recentTerminalTtlMs = override.recentTerminalTtlMs; 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)) { diff --git a/modules/agents/pi/extensions/subagents/index.ts b/modules/agents/pi/extensions/subagents/index.ts index 5ea2958..bb194e1 100644 --- a/modules/agents/pi/extensions/subagents/index.ts +++ b/modules/agents/pi/extensions/subagents/index.ts @@ -22,6 +22,7 @@ export default function subagents(pi: ExtensionAPI) { uiExpanded = config.ui.defaultExpanded; supervisor = new Supervisor(new SubprocessRpcRunner(), cwd, { maxConcurrent: config.maxConcurrent, + recentTerminalTtlMs: config.recentTerminalTtlMs, onMilestone: (status, event) => pi.appendEntry("subagent_milestone", { event, status }), onChange: (statuses) => { lastStatuses = statuses; diff --git a/modules/agents/pi/extensions/subagents/supervisor.test.ts b/modules/agents/pi/extensions/subagents/supervisor.test.ts index 7cff49e..edaca0c 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.test.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.test.ts @@ -110,6 +110,24 @@ test("completed children ignore later cancel", async () => { assert.equal(runner.starts[0].handle.cancelCalls, 0); }); +test("shutdown clears recent terminal expiry timer", async () => { + const runner = new FakeRunner(); + let changes = 0; + const supervisor = new Supervisor(runner, "/tmp", { + recentTerminalTtlMs: 5, + onChange: () => { + changes += 1; + }, + }); + await spawnStarted(supervisor); + + await supervisor.shutdown(); + const afterShutdown = changes; + await sleep(15); + + assert.equal(changes, afterShutdown); +}); + test("batch spawn returns accepted ids and per-entry failures", async () => { const runner = new FakeRunner(); const supervisor = new Supervisor(runner, "/tmp"); @@ -140,6 +158,58 @@ test("maxConcurrent preserves queued records", async () => { assert.equal(runner.starts.length, 2); }); +test("recent terminal statuses expire from list by ttl", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 5 }); + const accepted = await spawnStarted(supervisor); + + runner.starts[0].events.completed("done", "agent_settled"); + assert.equal(supervisor.list().some((status) => status.id === accepted.id), true); + + await sleep(10); + + assert.equal(supervisor.list().some((status) => status.id === accepted.id), false); + assert.equal(supervisor.result(accepted.id).result, "done"); +}); + +test("recent terminal ttl does not hide active statuses", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 0 }); + const accepted = await spawnStarted(supervisor); + + assert.equal(supervisor.list().some((status) => status.id === accepted.id), true); +}); + +test("zero recent terminal ttl hides terminal statuses immediately", async () => { + const runner = new FakeRunner(); + const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 0 }); + const accepted = await spawnStarted(supervisor); + + runner.starts[0].events.completed("done", "agent_settled"); + + assert.equal(supervisor.list().some((status) => status.id === accepted.id), false); + assert.equal(supervisor.result(accepted.id).result, "done"); +}); + +test("recent terminal ttl emits a change when an entry expires", async () => { + const runner = new FakeRunner(); + let changes = 0; + const supervisor = new Supervisor(runner, "/tmp", { + recentTerminalTtlMs: 5, + onChange: () => { + changes += 1; + }, + }); + await spawnStarted(supervisor); + const beforeComplete = changes; + + runner.starts[0].events.completed("done", "agent_settled"); + await sleep(15); + + assert.ok(changes > beforeComplete + 1); + assert.equal(supervisor.list().length, 0); +}); + test("wait blocks until multiple subagents are terminal", 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 743bfde..702aa46 100644 --- a/modules/agents/pi/extensions/subagents/supervisor.ts +++ b/modules/agents/pi/extensions/subagents/supervisor.ts @@ -24,6 +24,7 @@ interface RunningChild { interface SupervisorOptions { maxConcurrent?: number; recentTerminalLimit?: number; + recentTerminalTtlMs?: number; timeouts?: { startMs?: number; runMs?: number; @@ -47,6 +48,7 @@ export class Supervisor { private readonly children = new Map(); private readonly queue: RunningChild[] = []; private readonly waiters = new Set<() => void>(); + private recentTerminalTimer?: ReturnType; constructor( private readonly runner: ChildRunner, @@ -76,6 +78,7 @@ export class Supervisor { const active = statuses.filter((status) => !isTerminal(status.state)); const terminal = statuses .filter((status) => isTerminal(status.state)) + .filter((status) => this.isRecentTerminal(status)) .sort((a, b) => Date.parse(b.completedAt ?? b.startedAt) - Date.parse(a.completedAt ?? a.startedAt)) .slice(0, this.options.recentTerminalLimit ?? 10); return [...active, ...terminal]; @@ -134,6 +137,7 @@ export class Supervisor { } async shutdown(): Promise { + this.clearRecentTerminalTimer(); await Promise.allSettled( [...this.children.values()].map(async (child) => { if (!isTerminal(child.record.status.state)) { @@ -142,6 +146,7 @@ export class Supervisor { } }), ); + this.clearRecentTerminalTimer(); } private createChild(request: SpawnRequest): SpawnAccepted { @@ -340,6 +345,39 @@ export class Supervisor { private emitChange() { this.options.onChange?.(this.list()); for (const waiter of this.waiters) waiter(); + this.scheduleRecentTerminalExpiry(); + } + + private scheduleRecentTerminalExpiry() { + this.clearRecentTerminalTimer(); + const ttl = this.options.recentTerminalTtlMs; + if (ttl === undefined || ttl <= 0) return; + const now = Date.now(); + const nextExpiryMs = [...this.children.values()] + .map((child) => child.record.status) + .filter((status) => isTerminal(status.state)) + .map((status) => Date.parse(status.completedAt ?? status.startedAt)) + .filter((completed) => Number.isFinite(completed)) + .map((completed) => completed + ttl - now) + .filter((remaining) => remaining > 0) + .sort((a, b) => a - b)[0]; + if (nextExpiryMs === undefined) return; + this.recentTerminalTimer = setTimeout(() => this.emitChange(), nextExpiryMs + 1); + } + + private clearRecentTerminalTimer() { + if (!this.recentTerminalTimer) return; + clearTimeout(this.recentTerminalTimer); + this.recentTerminalTimer = undefined; + } + + private isRecentTerminal(status: SubagentStatus): boolean { + const ttl = this.options.recentTerminalTtlMs; + if (ttl === undefined) return true; + if (ttl <= 0) return false; + const completed = Date.parse(status.completedAt ?? status.startedAt); + if (!Number.isFinite(completed)) return true; + return Date.now() - completed <= ttl; } private waitReady(ids: string[], mode: SubagentWaitMode): boolean { -- 2.47.3