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; +}