feat(pi): add subagent tracer bullet

2026-08-01 13:54:57 -04:00
parent 585d4919e7
commit fdf0f89b37
5 changed files with 560 additions and 0 deletions
--- a/modules/agents/pi/extensions/subagents/index.ts
+++ 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,
+  };
+}
--- a/modules/agents/pi/extensions/subagents/runner.ts
+++ 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<string, PendingResponse>();
+
+  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<void> {
+    await this.send("prompt", { message });
+  }
+
+  async cancel(): Promise<void> {
+    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<string, unknown>): Promise<unknown> {
+    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<ChildHandle> {
+    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}`;
+}
--- a/modules/agents/pi/extensions/subagents/status.ts
+++ 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<SubagentStatus, "startedAt" | "completedAt">): 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);
+}
--- a/modules/agents/pi/extensions/subagents/supervisor.ts
+++ 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<string, RunningChild>();
+
+  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<SubagentStatus> {
+    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<void> {
+    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);
+}
--- a/modules/agents/pi/extensions/subagents/types.ts
+++ 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<void>;
+}
+
+export interface ChildRunner {
+  start(id: string, request: SpawnRequest, cwd: string, events: RunnerEvents): Promise<ChildHandle>;
+}