From 50fe9d3aa800de96152024f634bb04bd466f5a36 Mon Sep 17 00:00:00 2001 From: Colton Padden Date: Tue, 10 Nov 2026 10:54:41 -0500 Subject: [PATCH 1/4] pi: add foreground delegate tool --- pi/.pi/agent/extensions/delegate/README.md | 51 ++++ .../agent/extensions/delegate/index.test.mjs | 131 +++++++++ pi/.pi/agent/extensions/delegate/index.ts | 108 ++++++++ .../agent/extensions/delegate/runtime.test.ts | 182 ++++++++++++ pi/.pi/agent/extensions/delegate/runtime.ts | 261 ++++++++++++++++++ 5 files changed, 733 insertions(+) create mode 100644 pi/.pi/agent/extensions/delegate/README.md create mode 100644 pi/.pi/agent/extensions/delegate/index.test.mjs create mode 100644 pi/.pi/agent/extensions/delegate/index.ts create mode 100644 pi/.pi/agent/extensions/delegate/runtime.test.ts create mode 100644 pi/.pi/agent/extensions/delegate/runtime.ts diff --git a/pi/.pi/agent/extensions/delegate/README.md b/pi/.pi/agent/extensions/delegate/README.md new file mode 100644 index 0000000..d63f826 --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/README.md @@ -0,0 +1,51 @@ +# Delegate + +A small foreground-only alternative to Pi's [upstream subagent example](https://github.com/earendil-works/pi/tree/main/packages/coding-agent/examples/extensions/subagent). One tool, no named agents, workflow engine, background jobs, or model-routing configuration. + +## Use + +This directory is auto-discovered at `~/.pi/agent/extensions/delegate/`. Run `/reload` after adding or changing it. + +Ask Pi naturally: + +> Use delegate with MODEL_A to plan this change without editing files. Then show me the plan before implementation. + +The tool accepts: + +```ts +delegate({ + task: "Inspect the caching implementation. Return a plan, relevant files, risks, and validation steps. Do not modify files.", + model: "provider/model-id", + thinking: "high", + tools: ["read", "grep", "find", "ls"], +}) +``` + +Replace `provider/model-id` with an exact ID from `pi --list-models`. Model IDs may themselves contain slashes. `model` and `task` are required; `thinking` inherits the parent's level when omitted (Pi clamps it to the child's model capabilities). + +`tools` defaults to `read,bash,edit,write`. The optional allowlist accepts `read`, `bash`, `edit`, `write`, `grep`, `find`, and `ls`. Use `read,grep,find,ls` for planning/review without write or shell tools. Bash can modify files even without `edit` or `write`. + +For implementation, call delegate again with the implementation model and include the approved plan in the task. There is no implicit handoff or automatic planning/review policy. Keep routing preferences in `AGENTS.md` if desired; this extension does not change them. + +## Behavior and boundaries + +- Each call starts a fresh Pi process in the parent's working directory. Repository/global context files and normal skill discovery remain available; the parent's conversation is **not** copied. +- The tool stays pending until the child exits. Progress is streamed by code, not model-driven polling. Calls execute sequentially with other parent tools to avoid concurrent checkout edits. +- The child uses the running Pi executable/version, normal configured providers, and inherited environment. Extensions, MCP, and prompt templates are disabled. Providers/authentication available **only** through parent extensions or transient parent API-key overrides are not transferred. +- The parent's current project-trust decision is passed explicitly; delegation does not silently approve an untrusted checkout. +- Cancellation sends SIGTERM and escalates to SIGKILL after five seconds if needed. On macOS/Linux it targets the child's process group, including ordinary descendants. Reload/shutdown cancels and awaits active children. This is not durable background execution; forcibly killing the parent can bypass cleanup. Windows cancellation targets the direct child only. +- The delegate tool is unavailable in children, and child instructions prohibit further delegation or background work. This is a workflow guard, **not a sandbox**: Bash still has normal OS access and can launch programs. Deliberately detached descendants may escape process-group cleanup. +- The child shares the checkout and credentials. There is no worktree isolation or parent permission-hook enforcement inside the child. Do not use this where disabling parent security extensions would violate your expectations. +- Only a normally stopped final text response counts as success. Provider errors, token limits, nonzero exits, invalid JSONL, and missing final responses are errors; available partial text, usage, and bounded stderr are preserved. +- Output is limited to 50 KiB/2,000 lines. Longer results are saved in a private temporary directory, with the path included in the result. These files are not automatically deleted; remove them when no longer needed. Child sessions are otherwise ephemeral (`--no-session`). +- Recent tool activity and text stream into the tool row; expand it for more output. Child model usage is returned to Pi's session accounting. + +## Tests + +From the `pi/` Stow package directory, with Node.js 24+ and the managed Pi installation present: + +```sh +node --test .pi/agent/extensions/delegate/runtime.test.ts .pi/agent/extensions/delegate/index.test.mjs +``` + +Tests cover awaited completion, streaming, model/tool arguments, UTF-8/JSONL framing, usage, errors, truncation, recursion guards, and cancellation/shutdown. They use local fake child processes, not paid model requests. `index.test.mjs` uses the installed Pi extension loader; `runtime.test.ts` tests the process lifecycle independently. diff --git a/pi/.pi/agent/extensions/delegate/index.test.mjs b/pi/.pi/agent/extensions/delegate/index.test.mjs new file mode 100644 index 0000000..9dc932d --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/index.test.mjs @@ -0,0 +1,131 @@ +// Loads the real extension with the installed Pi loader, but uses a fake child +// CLI. No model requests, credentials, or paid API calls are needed. +import assert from "node:assert/strict"; +import { mkdtemp, readFile, realpath, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const agentDir = fileURLToPath(new URL("../../", import.meta.url)); +const version = (await readFile(join(agentDir, "install/current-version"), "utf8")).trim(); +const loaderPath = join(agentDir, "install/releases", version, "node_modules/@earendil-works/pi-coding-agent/dist/core/extensions/loader.js"); +const { loadExtensions } = await import(pathToFileURL(loaderPath).href); +const extensionPath = fileURLToPath(new URL("index.ts", import.meta.url)); + +async function fixture(fn) { + const directory = await mkdtemp(join(tmpdir(), "delegate-extension-test-")); + const oldScript = process.argv[1]; + try { + const loaded = await loadExtensions([extensionPath], directory); + assert.deepEqual(loaded.errors, []); + const extension = loaded.extensions[0]; + const tool = extension.tools.get("delegate").definition; + const script = join(directory, "fake-pi.mjs"); + process.argv[1] = script; + const ctx = { + cwd: directory, thinkingLevel: "medium", isProjectTrusted: () => false, + modelRegistry: { find: (provider, id) => provider === "test" && id === "vendor/model" ? { provider, id } : undefined }, + }; + await writeFile(script, ` + let task = ''; for await (const chunk of process.stdin) task += chunk; + const event = {type:'message_end', message:{role:'assistant', stopReason:'stop', + content:[{type:'text',text:JSON.stringify({task,args:process.argv.slice(2),cwd:process.cwd()})}], + usage:{input:10,output:2,cacheRead:0,cacheWrite:0,totalTokens:12, + cost:{input:0,output:0,cacheRead:0,cacheWrite:0,total:0}}}}; + if (task.includes('LONG_OUTPUT')) event.message.content[0].text = 'x'.repeat(60000); + if (task.includes('FAIL')) { event.message.stopReason='error'; event.message.errorMessage='Provider unavailable'; } + if (task.includes('WAIT')) { process.stdout.write(JSON.stringify(event)+'\\n'); setInterval(()=>{},1000); } + else process.stdout.write(JSON.stringify(event)+'\\n'); + `); + await fn({ tool, ctx, extension, directory }); + } finally { + process.argv[1] = oldScript; + await rm(directory, { recursive: true, force: true }); + } +} + +test("tool loads with a small schema and forwards model/thinking/tools to an awaited child", async () => { + await fixture(async ({ tool, ctx }) => { + assert.equal(tool.executionMode, "sequential"); + assert.deepEqual(Object.keys(tool.parameters.properties).sort(), ["model", "task", "thinking", "tools"]); + const result = await tool.execute("test", { + task: "Plan this", model: "test/vendor/model", thinking: "high", tools: ["read", "grep"], + }, undefined, undefined, ctx); + assert.equal(result.isError, false); + assert.equal(result.usage.input, 10); + const child = JSON.parse(result.details.output); + assert.equal(child.task, "Task: Plan this"); + assert.equal(child.cwd, await realpath(ctx.cwd)); + assert.equal(child.args[child.args.indexOf("--model") + 1], "vendor/model"); + assert.equal(child.args[child.args.indexOf("--thinking") + 1], "high"); + assert.equal(child.args[child.args.indexOf("--tools") + 1], "read,grep"); + }); +}); + +test("omitted thinking inherits from the parent; tools use the documented default", async () => { + await fixture(async ({ tool, ctx }) => { + const result = await tool.execute("test", { task: "Implement", model: "test/vendor/model" }, undefined, undefined, ctx); + const { args } = JSON.parse(result.details.output); + assert.equal(args[args.indexOf("--thinking") + 1], "medium"); + assert.equal(args[args.indexOf("--tools") + 1], "read,bash,edit,write"); + }); +}); + +test("rejects unknown, ambiguous, and blank requests before launching", async () => { + await fixture(async ({ tool, ctx }) => { + for (const params of [ + { task: "Do work", model: "unknown/model" }, + { task: "Do work", model: "model" }, + { task: " ", model: "test/vendor/model" }, + ]) await assert.rejects(tool.execute("test", params, undefined, undefined, ctx)); + }); +}); + +test("provider errors are flagged while preserving partial text and usage", async () => { + await fixture(async ({ tool, ctx }) => { + const result = await tool.execute("test", { task: "FAIL", model: "test/vendor/model" }, undefined, undefined, ctx); + assert.equal(result.isError, true); + assert.match(result.content[0].text, /Provider unavailable/); + assert.match(result.content[0].text, /Task: FAIL/); + assert.equal(result.usage.input, 10); + }); +}); + +test("large results are bounded with full output saved privately outside the repository", async () => { + await fixture(async ({ tool, ctx }) => { + const result = await tool.execute("test", { task: "LONG_OUTPUT", model: "test/vendor/model" }, undefined, undefined, ctx); + const path = result.details.outputFile; + try { + assert.ok(Buffer.byteLength(result.content[0].text) < 52000); + assert.match(result.content[0].text, /Output truncated/); + assert.equal((await readFile(path, "utf8")).length, 60000); + assert.equal((await stat(path)).mode & 0o777, 0o600); + } finally { await rm(join(path, ".."), { recursive: true, force: true }); } + }); +}); + +test("session shutdown cancels and awaits active children", { timeout: 5000 }, async () => { + await fixture(async ({ tool, ctx, extension }) => { + let ready; + const updated = new Promise((resolve) => { ready = resolve; }); + const run = tool.execute("test", { task: "WAIT", model: "test/vendor/model" }, undefined, + (result) => { if (result.details.turns) ready(); }, ctx); + await updated; + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + assert.equal((await run).details.status, "aborted"); + }); +}); + +test("child environment guard prevents registering delegate recursively", async () => { + const previous = process.env.PI_DELEGATE_CHILD; + process.env.PI_DELEGATE_CHILD = "1"; + try { + const loaded = await loadExtensions([extensionPath], process.cwd()); + assert.deepEqual(loaded.errors, []); + assert.equal(loaded.extensions[0].tools.size, 0); + } finally { + if (previous === undefined) delete process.env.PI_DELEGATE_CHILD; + else process.env.PI_DELEGATE_CHILD = previous; + } +}); diff --git a/pi/.pi/agent/extensions/delegate/index.ts b/pi/.pi/agent/extensions/delegate/index.ts new file mode 100644 index 0000000..7c5ff92 --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/index.ts @@ -0,0 +1,108 @@ +import { mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { truncateHead, type ExtensionAPI } from "@earendil-works/pi-coding-agent"; +import { Text } from "@earendil-works/pi-tui"; +import { Type } from "typebox"; +import { runDelegate, type DelegateDetails } from "./runtime.ts"; + +const builtinTools = ["read", "bash", "edit", "write", "grep", "find", "ls"] as const; + +export default function (pi: ExtensionAPI) { + // Defense in depth if this extension is explicitly loaded by a child. + if (process.env.PI_DELEGATE_CHILD === "1") return; + + const active = new Map>(); + pi.on("session_shutdown", async () => { + for (const controller of active.keys()) controller.abort(); + await Promise.allSettled(active.values()); + }); + + pi.registerTool({ + name: "delegate", + label: "Delegate", + executionMode: "sequential", + description: [ + "Run a self-contained task in a fresh Pi child using an exact provider/model ID.", + "Waits for completion and streams progress; no polling or background jobs.", + "The child shares this checkout and credentials, NOT the conversation; include all necessary context.", + "Child extensions and MCP are disabled. This is not a sandbox.", + "Default tools: read,bash,edit,write. For read-only planning use read,grep,find,ls.", + "Output is capped at 50 KiB/2,000 lines; longer output is saved to a temporary file.", + ].join(" "), + promptSnippet: "Delegate a bounded task to a chosen model and await its result", + promptGuidelines: [ + "Give delegate a self-contained task, explicit edit permissions, and expected output; children cannot see this conversation.", + "Do not delegate trivial work or use delegate concurrently with edits to the same files. Verify child findings before acting.", + ], + parameters: Type.Object({ + task: Type.String({ minLength: 1, description: "Self-contained task, context, constraints, and expected output" }), + model: Type.String({ minLength: 3, description: "Exact provider/model ID; model IDs may contain slashes" }), + thinking: Type.Optional(Type.String({ enum: ["off", "minimal", "low", "medium", "high", "xhigh", "max"] })), + tools: Type.Optional(Type.Array(Type.String({ enum: [...builtinTools] }), { + minItems: 1, uniqueItems: true, description: "Built-in tool allowlist; omit for read,bash,edit,write", + })), + }), + async execute(_id, params, signal, onUpdate, ctx) { + if (!params.task.trim()) throw new Error("Delegated task must not be blank."); + const slash = params.model.indexOf("/"); + if (slash < 1 || slash === params.model.length - 1) throw new Error("Use an exact provider/model ID."); + const provider = params.model.slice(0, slash); + const model = params.model.slice(slash + 1); + if (!ctx.modelRegistry.find(provider, model)) throw new Error(`Unknown model: ${params.model}. Use pi --list-models to find exact IDs.`); + + const controller = new AbortController(); + const childSignal = signal ? AbortSignal.any([signal, controller.signal]) : controller.signal; + const run = runDelegate({ + task: params.task, provider, model, tools: params.tools, + thinking: params.thinking ?? ctx.thinkingLevel, + cwd: ctx.cwd, trusted: ctx.isProjectTrusted(), + }, childSignal, (details) => { + onUpdate?.({ + content: [{ type: "text", text: truncateHead(details.output || "Working…").content }], + details, + }); + }); + active.set(controller, run); + try { + const details = await run; + const text = [details.error, details.output, details.status !== "completed" ? details.stderr : ""].filter(Boolean).join("\n\n"); + const truncated = truncateHead(text); + let output = truncated.content; + if (truncated.truncated) { + const directory = await mkdtemp(join(tmpdir(), "pi-delegate-")); + details.outputFile = join(directory, "output.txt"); + await writeFile(details.outputFile, text, { mode: 0o600 }); + output += `\n\n[Output truncated. Full output: ${details.outputFile}]`; + } + return { + content: [{ type: "text" as const, text: output || "(no output)" }], + details, + usage: details.usage, + isError: details.status !== "completed", + }; + } finally { + active.delete(controller); + } + }, + renderCall(args, theme) { + const model = args.model ?? "…"; + const task = args.task?.replace(/\s+/g, " ").slice(0, 140) ?? "…"; + return new Text(`${theme.fg("toolTitle", theme.bold("delegate "))}${theme.fg("accent", model)}\n${theme.fg("dim", task)}`, 0, 0); + }, + renderResult(result, { expanded, isPartial }, theme) { + const details = result.details as DelegateDetails | undefined; + const content = result.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); + if (!details) return new Text(content, 0, 0); + const status = isPartial ? "running" : details.status; + const color = status === "completed" ? "success" : status === "running" ? "muted" : "error"; + const usage = `${details.turns} turns · ↑${details.usage.input} ↓${details.usage.output} · $${details.usage.cost.total.toFixed(4)}`; + const activity = (expanded ? details.activity : details.activity.slice(-3)).join("\n"); + const preview = expanded ? content : content.split("\n").slice(-6).join("\n").slice(-1500); + return new Text([ + theme.fg(color, status), theme.fg("dim", usage), + activity && theme.fg("muted", activity), preview, + ].filter(Boolean).join("\n"), 0, 0); + }, + }); +} diff --git a/pi/.pi/agent/extensions/delegate/runtime.test.ts b/pi/.pi/agent/extensions/delegate/runtime.test.ts new file mode 100644 index 0000000..6ccb096 --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/runtime.test.ts @@ -0,0 +1,182 @@ +import assert from "node:assert/strict"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { test } from "node:test"; +import { childArguments, runDelegate, type DelegateOptions } from "./runtime.ts"; + +const options: DelegateOptions = { + task: "Inspect this task", provider: "test-provider", model: "vendor/model", + cwd: process.cwd(), trusted: false, +}; +const usage = { + input: 10, output: 5, cacheRead: 2, cacheWrite: 1, totalTokens: 18, + cost: { input: 0.01, output: 0.02, cacheRead: 0, cacheWrite: 0, total: 0.03 }, +}; +const message = (text = "Done", stopReason = "stop") => ({ + type: "message_end", message: { + role: "assistant", content: [{ type: "text", text }], usage, stopReason, + errorMessage: stopReason === "error" ? "Provider failed" : undefined, + }, +}); +const line = (event: unknown) => `${JSON.stringify(event)}\n`; +const invocation = (script: string) => ({ command: process.execPath, args: ["--input-type=module", "-e", script, "--"] }); +const emit = (event: unknown) => `process.stdout.write(${JSON.stringify(line(event))});`; +const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitFor(check: () => Promise) { + for (let i = 0; i < 200; i++) { + if (await check()) return; + await delay(10); + } + throw new Error("Timed out waiting for child readiness"); +} + +test("child arguments pin provider/model, disable extensions, and preserve trust/tool scope", () => { + const args = childArguments({ ...options, thinking: "high", tools: ["read", "grep"] }); + assert.ok(args.includes("--no-extensions")); + assert.ok(args.includes("--no-session")); + assert.ok(args.includes("--no-mcp")); + assert.ok(args.includes("--no-approve")); + assert.equal(args[args.indexOf("--provider") + 1], "test-provider"); + assert.equal(args[args.indexOf("--model") + 1], "vendor/model"); + assert.equal(args[args.indexOf("--tools") + 1], "read,grep"); + assert.equal(args[args.indexOf("--thinking") + 1], "high"); + assert.ok(childArguments({ ...options, trusted: true }).includes("--approve")); +}); + +test("awaits process exit, streams text/tools, and counts final usage once", async () => { + let finished = false; + const updates: string[] = []; + const script = [ + emit({ type: "message_start", message: { role: "assistant" } }), + emit({ type: "message_update", assistantMessageEvent: { type: "text_delta", delta: "Working" } }), + emit({ type: "tool_execution_start", toolName: "read", args: { path: "README.md" } }), + `setTimeout(() => { ${emit(message())} }, 250);`, + ].join("\n"); + const run = runDelegate(options, undefined, (d) => updates.push(d.output), invocation(script)).then((d) => { finished = true; return d; }); + await delay(170); + assert.equal(finished, false); + const result = await run; + assert.equal(result.status, "completed"); + assert.equal(result.output, "Done"); + assert.ok(updates.includes("Working")); + assert.ok(result.activity.includes("→ read README.md")); + assert.deepEqual(result.usage, usage); + assert.equal(result.turns, 1); +}); + +test("transmits task on stdin rather than argv and marks children", async () => { + const task = "@not-a-file\n--option 'quoted' $HOME"; + const script = `let input = ''; for await (const chunk of process.stdin) input += chunk; + const text = JSON.stringify({input, args: process.argv, child: process.env.PI_DELEGATE_CHILD, cwd: process.cwd()}); + const event = ${JSON.stringify(message())}; event.message.content[0].text = text; + process.stdout.write(JSON.stringify(event));`; + const result = await runDelegate({ ...options, task }, undefined, undefined, invocation(script)); + const parsed = JSON.parse(result.output); + assert.equal(parsed.input, `Task: ${task}`); + assert.ok(!parsed.args.includes(task)); + assert.equal(parsed.child, "1"); + assert.equal(parsed.cwd, process.cwd()); +}); + +test("handles split UTF-8, CRLF, multiple text blocks, and trailing records without LF", async () => { + const event = message("Hello 🌍\u2028world"); + event.message.content.push({ type: "text", text: "Second block" }); + const script = `const bytes = Buffer.from(${JSON.stringify(JSON.stringify(event) + "\r\n")}); + for (const byte of bytes) { process.stdout.write(Buffer.from([byte])); await new Promise(r => setTimeout(r, 1)); }`; + const result = await runDelegate(options, undefined, undefined, invocation(script)); + assert.equal(result.output, "Hello 🌍\u2028world\nSecond block"); + assert.equal(result.status, "completed"); +}); + +test("accumulates multiple turns but returns only the final response", async () => { + const result = await runDelegate(options, undefined, undefined, + invocation(emit(message("First", "toolUse")) + emit(message("Final")))); + assert.equal(result.output, "Final"); + assert.equal(result.turns, 2); + assert.equal(result.usage.input, 20); +}); + +for (const [name, script, expected] of [ + ["nonzero exit", emit(message("Partial")) + 'process.stderr.write("CLI failure"); process.exitCode = 2;', /code 2/], + ["provider error with exit zero", emit(message("Partial", "error")), /Provider failed/], + ["token limit", emit(message("Incomplete", "length")), /ended with length/], + ["missing assistant response", 'process.stdout.write("{\\"type\\":\\"session\\"}\\n");', /without a final/], + ["invalid protocol", 'process.stdout.write("not json\\n");', /invalid JSONL/], + ["oversize incomplete record", 'process.stdout.write("x".repeat(8 * 1024 * 1024 + 1));', /exceeded 8 MiB/], +] as const) { + test(`reports ${name} without losing diagnostics`, async () => { + const result = await runDelegate(options, undefined, undefined, invocation(script)); + assert.equal(result.status, "failed"); + assert.match(result.error ?? "", expected); + if (name === "nonzero exit") { + assert.equal(result.output, "Partial"); + assert.match(result.stderr, /CLI failure/); + } + }); +} + +test("malformed event shapes fail without escaping the stream handler", async () => { + const result = await runDelegate(options, undefined, undefined, invocation(emit({ type: "message_end" }))); + assert.equal(result.status, "failed"); + assert.match(result.error ?? "", /Invalid child event/); +}); + +test("progress callback failures terminate the child rather than orphaning it", async () => { + const result = await runDelegate(options, undefined, () => { throw new Error("UI unavailable"); }, + invocation("setInterval(() => {}, 1000)"), 50); + assert.equal(result.status, "failed"); + assert.match(result.error ?? "", /UI unavailable/); +}); + +test("spawn errors are failures, never success", async () => { + const result = await runDelegate(options, undefined, undefined, { command: "/nonexistent/pi-delegate-test", args: [] }); + assert.equal(result.status, "failed"); + assert.match(result.error ?? "", /Could not start Pi/); +}); + +test("pre-aborted calls do not spawn", async () => { + const result = await runDelegate(options, AbortSignal.abort(), undefined, { command: "/nonexistent/pi-delegate-test", args: [] }); + assert.equal(result.status, "aborted"); + assert.equal(result.exitCode, undefined); +}); + +test("cancellation escalates even after SIGTERM was successfully sent", { skip: process.platform === "win32", timeout: 5000 }, async () => { + const directory = await mkdtemp(join(tmpdir(), "delegate-test-")); + const ready = join(directory, "ready"); + try { + const script = `import {writeFileSync} from 'node:fs'; + process.on('SIGTERM', () => {}); + ${emit(message("Partial progress"))} + writeFileSync(${JSON.stringify(ready)}, 'ready'); setInterval(() => {}, 1000);`; + const controller = new AbortController(); + const run = runDelegate(options, controller.signal, undefined, invocation(script), 50); + await waitFor(async () => readFile(ready).then(() => true, () => false)); + controller.abort(); + const result = await run; + assert.equal(result.status, "aborted"); + assert.equal(result.output, "Partial progress"); + assert.equal(result.exitSignal, "SIGKILL"); + } finally { await rm(directory, { recursive: true, force: true }); } +}); + +test("cancellation reaches ordinary descendants, not just the Pi process", { skip: process.platform === "win32", timeout: 5000 }, async () => { + const directory = await mkdtemp(join(tmpdir(), "delegate-tree-test-")); + const ready = join(directory, "ready"); + let descendant: number | undefined; + try { + const script = `import {spawn} from 'node:child_process'; import {writeFileSync} from 'node:fs'; + const child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], {stdio:'inherit'}); + writeFileSync(${JSON.stringify(ready)}, String(child.pid)); setInterval(() => {}, 1000);`; + const controller = new AbortController(); + const run = runDelegate(options, controller.signal, undefined, invocation(script), 50); + await waitFor(async () => readFile(ready).then((p) => { descendant = Number(p); return true; }, () => false)); + controller.abort(); + assert.equal((await run).status, "aborted"); + await waitFor(async () => { try { process.kill(descendant!, 0); return false; } catch { return true; } }); + } finally { + if (descendant) { try { process.kill(descendant, "SIGKILL"); } catch {} } + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/pi/.pi/agent/extensions/delegate/runtime.ts b/pi/.pi/agent/extensions/delegate/runtime.ts new file mode 100644 index 0000000..14b91c9 --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/runtime.ts @@ -0,0 +1,261 @@ +// Foreground subprocess lifecycle, adapted from Pi's MIT-licensed subagent example: +// https://github.com/earendil-works/pi/tree/main/packages/coding-agent/examples/extensions/subagent +import { spawn } from "node:child_process"; +import { existsSync } from "node:fs"; +import { basename } from "node:path"; +import type { Usage } from "@earendil-works/pi-ai"; +import type { JsonAgentSessionEvent } from "@earendil-works/pi-coding-agent"; + +export const CHILD_INSTRUCTIONS = [ + "You are a delegated child working on one self-contained task.", + "You do not have the parent's conversation. Follow the task and local repository instructions.", + "Do not delegate, spawn another coding agent, or leave background work running.", + "Return a concise report of findings, changes, validation, and unresolved issues.", +].join("\n"); + +export interface DelegateOptions { + task: string; + provider: string; + model: string; + thinking?: string; + tools?: string[]; + cwd: string; + trusted: boolean; +} + +export interface DelegateDetails { + model: string; + status: "running" | "completed" | "failed" | "aborted"; + output: string; + stderr: string; + activity: string[]; + turns: number; + usage: Usage; + error?: string; + exitCode?: number | null; + exitSignal?: string | null; + outputFile?: string; +} + +export interface Invocation { + command: string; + args: string[]; +} + +// Reuse the running Pi version, including standalone builds, rather than a +// potentially different `pi` on PATH. Mirrors the upstream example. +export function piInvocation(): Invocation { + const script = process.argv[1]; + if (script && !script.startsWith("/$bunfs/root/") && existsSync(script)) { + return { command: process.execPath, args: [script] }; + } + if (!/^(node|bun)(\.exe)?$/i.test(basename(process.execPath))) { + return { command: process.execPath, args: [] }; + } + return { command: "pi", args: [] }; +} + +export function childArguments(options: DelegateOptions): string[] { + const args = [ + "--mode", "json", "--print", "--no-session", + "--no-extensions", "--no-mcp", "--no-prompt-templates", + options.trusted ? "--approve" : "--no-approve", + "--provider", options.provider, "--model", options.model, + "--tools", (options.tools ?? ["read", "bash", "edit", "write"]).join(","), + "--append-system-prompt", CHILD_INSTRUCTIONS, + ]; + if (options.thinking) args.push("--thinking", options.thinking); + return args; +} + +export async function runDelegate( + options: DelegateOptions, + signal?: AbortSignal, + onUpdate?: (details: DelegateDetails) => void, + invocation: Invocation = piInvocation(), + killDelayMs = 5000, +): Promise { + const details: DelegateDetails = { + model: `${options.provider}/${options.model}`, + status: "running", output: "", stderr: "", activity: [], turns: 0, + usage: { + input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }; + if (signal?.aborted) return { ...details, status: "aborted", error: "Delegation cancelled before launch." }; + + return new Promise((resolve) => { + // A dedicated POSIX process group lets cancellation reach child shell tools, + // not just Pi itself. It is still piped, referenced, and awaited: never unref(). + const grouped = process.platform !== "win32"; + const proc = spawn(invocation.command, [...invocation.args, ...childArguments(options)], { + cwd: options.cwd, + shell: false, + detached: grouped, + stdio: ["pipe", "pipe", "pipe"], + env: { ...process.env, PI_DELEGATE_CHILD: "1" }, + }); + let closed = false; + let killTimer: ReturnType | undefined; + let updateTimer: ReturnType | undefined; + let buffer = ""; + let protocolError: string | undefined; + let spawnError: string | undefined; + let assistantSeen = false; + let stopReason: string | undefined; + let assistantError: string | undefined; + + const emit = () => { + // Snapshots must not mutate already emitted tool results/session entries. + try { + onUpdate?.({ ...details, activity: [...details.activity], usage: { ...details.usage, cost: { ...details.usage.cost } } }); + } catch (error) { + spawnError = `Could not publish child progress: ${String(error)}`; + terminate(); + } + }; + const scheduleUpdate = () => { + if (!closed && !updateTimer) updateTimer = setTimeout(() => { updateTimer = undefined; emit(); }, 100); + }; + const activity = (text: string) => { + details.activity.push(text.slice(0, 500)); + details.activity = details.activity.slice(-20); + scheduleUpdate(); + }; + const kill = (sig: NodeJS.Signals) => { + if (!proc.pid) return; + try { + if (grouped) process.kill(-proc.pid, sig); + else proc.kill(sig); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") { + details.stderr = `${details.stderr}\nUnable to send ${sig}: ${String(error)}`.slice(-65536); + } + } + }; + const terminate = () => { + if (closed || killTimer) return; + kill("SIGTERM"); + // proc.killed only means a signal was sent, not that the process exited. + killTimer = setTimeout(() => kill("SIGKILL"), killDelayMs); + }; + const abort = () => { details.status = "aborted"; terminate(); }; + + const processLine = (line: string) => { + if (!line.trim() || protocolError) return; + if (Buffer.byteLength(line) > 8 * 1024 * 1024) { + protocolError = "Child JSONL record exceeded 8 MiB."; + terminate(); + return; + } + let event: JsonAgentSessionEvent; + try { + const parsed = JSON.parse(line); + if (!parsed || typeof parsed.type !== "string") throw new Error("Missing event type"); + event = parsed; + } catch { + protocolError = "Child emitted invalid JSONL output."; + terminate(); + return; + } + try { + switch (event.type) { + case "message_start": + if (event.message.role === "assistant") details.output = ""; + break; + case "message_update": { + const delta = event.assistantMessageEvent; + if (delta.type === "text_delta") { + details.output = (details.output + delta.delta).slice(-16384); + scheduleUpdate(); + } + break; + } + case "message_end": { + const message = event.message; + if (message.role !== "assistant") break; + assistantSeen = true; + details.turns++; + stopReason = message.stopReason; + assistantError = message.errorMessage; + details.output = message.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); + for (const key of ["input", "output", "cacheRead", "cacheWrite", "totalTokens"] as const) { + details.usage[key] += message.usage?.[key] ?? 0; + } + for (const key of ["input", "output", "cacheRead", "cacheWrite", "total"] as const) { + details.usage.cost[key] += message.usage?.cost?.[key] ?? 0; + } + scheduleUpdate(); + break; + } + case "tool_execution_start": { + const args = event.args as Record; + const target = args?.command ?? args?.path ?? args?.pattern ?? ""; + activity(`→ ${event.toolName} ${String(target)}`); + break; + } + case "tool_execution_end": + activity(`${event.isError ? "✗" : "✓"} ${event.toolName}`); + break; + case "auto_retry_start": + activity(`Retry ${event.attempt}: ${event.errorMessage}`); + break; + } + } catch (error) { + protocolError = `Invalid child event: ${String(error)}`; + terminate(); + } + }; + + proc.stdout.setEncoding("utf8"); + proc.stdout.on("data", (chunk: string) => { + buffer += chunk; + let newline: number; + while ((newline = buffer.indexOf("\n")) !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + processLine(line); + } + if (Buffer.byteLength(buffer) > 8 * 1024 * 1024) { + protocolError = "Child JSONL record exceeded 8 MiB."; + buffer = ""; + terminate(); + } + }); + proc.stderr.setEncoding("utf8"); + proc.stderr.on("data", (chunk: string) => { details.stderr = (details.stderr + chunk).slice(-65536); }); + proc.stdin.on("error", (error: NodeJS.ErrnoException) => { + // An early CLI failure may close stdin before accepting the task. + if (error.code !== "EPIPE") { spawnError = `Child stdin failed: ${error.message}`; terminate(); } + }); + proc.on("error", (error) => { spawnError = `Could not start Pi: ${error.message}`; }); + proc.on("close", (code, exitSignal) => { + closed = true; + signal?.removeEventListener("abort", abort); + if (killTimer) clearTimeout(killTimer); + if (updateTimer) clearTimeout(updateTimer); + if (buffer.trim()) processLine(buffer); + details.exitCode = code; + details.exitSignal = exitSignal; + if (details.status === "aborted") { + details.error = "Delegation cancelled."; + } else { + const failure = spawnError ?? protocolError + ?? (code !== 0 ? `Child exited ${exitSignal ? `with signal ${exitSignal}` : `with code ${code}`}.` : undefined) + ?? (assistantSeen && stopReason !== "stop" ? assistantError || `Child ended with ${stopReason ?? "no stop reason"}.` : undefined) + ?? (!assistantSeen || !details.output.trim() ? "Child exited without a final text response." : undefined); + details.status = failure ? "failed" : "completed"; + details.error = failure; + } + resolve(details); + }); + + signal?.addEventListener("abort", abort, { once: true }); + if (signal?.aborted) abort(); + emit(); + // stdin avoids shell quoting, argument-size limits, @file interpretation, + // and exposing the delegated task in the process argument list. + proc.stdin.end(`Task: ${options.task}`); + }); +} From 47e2b17ca95000eea56e5acab449a3f05fc44bcd Mon Sep 17 00:00:00 2001 From: Colton Padden Date: Tue, 10 Nov 2026 11:04:33 -0500 Subject: [PATCH 2/4] pi: support parallel delegates in explicit workspaces --- pi/.pi/agent/extensions/delegate/README.md | 45 ++++- .../agent/extensions/delegate/index.test.mjs | 176 +++++++++++++++++- pi/.pi/agent/extensions/delegate/index.ts | 83 +++++++-- pi/.pi/agent/extensions/delegate/queue.ts | 40 ++++ .../agent/extensions/delegate/runtime.test.ts | 23 +++ pi/.pi/agent/extensions/delegate/runtime.ts | 34 +++- 6 files changed, 363 insertions(+), 38 deletions(-) create mode 100644 pi/.pi/agent/extensions/delegate/queue.ts diff --git a/pi/.pi/agent/extensions/delegate/README.md b/pi/.pi/agent/extensions/delegate/README.md index d63f826..4948d95 100644 --- a/pi/.pi/agent/extensions/delegate/README.md +++ b/pi/.pi/agent/extensions/delegate/README.md @@ -1,6 +1,6 @@ # Delegate -A small foreground-only alternative to Pi's [upstream subagent example](https://github.com/earendil-works/pi/tree/main/packages/coding-agent/examples/extensions/subagent). One tool, no named agents, workflow engine, background jobs, or model-routing configuration. +A small foreground-only alternative to Pi's [upstream subagent example](https://github.com/earendil-works/pi/tree/main/packages/coding-agent/examples/extensions/subagent). One tool, no named agents, workflow engine, background jobs, or model-routing configuration. Delegates run concurrently (up to three per parent session), with optional working directories prepared by the parent. ## Use @@ -23,19 +23,50 @@ delegate({ Replace `provider/model-id` with an exact ID from `pi --list-models`. Model IDs may themselves contain slashes. `model` and `task` are required; `thinking` inherits the parent's level when omitted (Pi clamps it to the child's model capabilities). +`cwd` selects an **existing directory**, defaulting to the parent's working directory. Relative paths resolve against the parent cwd; absolute paths, `~/` paths, and a leading `@` are supported. Symlinks are resolved and non-directories rejected before starting a child. The canonical path is shown in progress and final results. + `tools` defaults to `read,bash,edit,write`. The optional allowlist accepts `read`, `bash`, `edit`, `write`, `grep`, `find`, and `ls`. Use `read,grep,find,ls` for planning/review without write or shell tools. Bash can modify files even without `edit` or `write`. For implementation, call delegate again with the implementation model and include the approved plan in the task. There is no implicit handoff or automatic planning/review policy. Keep routing preferences in `AGENTS.md` if desired; this extension does not change them. +## Choosing isolation with Worktrunk + +The **parent agent decides before launching** whether isolation is needed. The extension does not classify tasks, create worktrees, switch branches, merge, or clean up worktrees. + +| Task | Suggested environment | +| --- | --- | +| Read code or review without writes | Existing checkout, preferably `tools: ["read", "grep", "find", "ls"]` | +| Fix independent PRs or implement separate changes concurrently | One dedicated worktree per writer | +| Modify the parent's uncommitted work | Parent checkout, with no concurrent parent/sibling writers or branch changes | +| Run builds/tests/formatters | Usually a dedicated worktree; these commands may write files | + +If uncertain, investigate read-only first. For independent PR fixes: + +1. Inspect `wt list --format json` and identify the PR branches/bases. Never assume an existing worktree is free just because it exists; check with the user when ownership/activity is uncertain. +2. Prepare worktrees with `wt`, then pass the returned paths to delegate. For example, `wt switch pr:6182 --no-cd --no-hooks --format json` selects/creates the PR's worktree, while `wt switch --create fix/cache --base origin/main --no-cd --no-hooks --format json` creates a new branch from an explicit base. Inspect the JSON result rather than guessing the configured path. Do preparation in an earlier tool batch, not alongside the delegates that depend on it. +3. Choose hooks/setup deliberately: `--no-hooks` avoids interactive approval and background setup. If hooks are needed, inspect them, honor approval decisions, and await required setup; never blindly use `--yes` to approve new project commands. Dependencies and ignored files may need project-specific setup. Do not indiscriminately copy secrets/caches or share writable build outputs. +4. Launch independent delegate calls in the **same assistant tool batch** with distinct `cwd` values: + + ```ts + delegate({ task: "Fix PR 6182 only. Commit/push only if authorized. Report changes and tests.", model: "provider/model-id", cwd: "/path/from/wt/pr-6182" }) + delegate({ task: "Fix PR 6183 only. Commit/push only if authorized. Report changes and tests.", model: "provider/model-id", cwd: "/path/from/wt/pr-6183" }) + ``` + +5. Review each child's report and changes. Keep worktrees with useful work, including after failures/cancellation. Cleanup is an explicit parent/user decision; do not force-remove dirty worktrees or delete unmerged branches automatically. + +New worktrees contain the selected committed base, **not the parent's uncommitted changes**. Different subdirectories of one checkout are not independent worktrees. Worktrees separate working files, HEAD, and index, but still share Git refs, credentials, ports, databases, and external effects. They are not sandboxes. + ## Behavior and boundaries -- Each call starts a fresh Pi process in the parent's working directory. Repository/global context files and normal skill discovery remain available; the parent's conversation is **not** copied. -- The tool stays pending until the child exits. Progress is streamed by code, not model-driven polling. Calls execute sequentially with other parent tools to avoid concurrent checkout edits. +- Each call starts a fresh Pi process in its selected working directory. Repository/global context files and normal skill discovery remain available; the parent's conversation is **not** copied. +- The tool stays pending until the child exits. Up to **three children run concurrently per parent session**; further calls wait in a cancellable FIFO queue. This limit does not cover other Pi sessions. Pi must also be in parallel tool mode: a sequential tool anywhere in the same assistant batch makes that entire batch sequential. +- Progress is streamed by code, not model-driven polling. Each row shows queued/running/final status, canonical cwd, time spent queued, and child execution time. One-second heartbeats update timing even when the child is quiet. Queueing does not launch a child or consume model tokens. +- Parent and sibling tools can now execute alongside delegates. The parent must avoid writes, staging, branch changes, and conflicting external actions in a workspace used by a child. The extension does not enforce workspace ownership or coordinate other processes. - The child uses the running Pi executable/version, normal configured providers, and inherited environment. Extensions, MCP, and prompt templates are disabled. Providers/authentication available **only** through parent extensions or transient parent API-key overrides are not transferred. -- The parent's current project-trust decision is passed explicitly; delegation does not silently approve an untrusted checkout. -- Cancellation sends SIGTERM and escalates to SIGKILL after five seconds if needed. On macOS/Linux it targets the child's process group, including ordinary descendants. Reload/shutdown cancels and awaits active children. This is not durable background execution; forcibly killing the parent can bypass cleanup. Windows cancellation targets the direct child only. +- The parent's current project-trust decision is passed explicitly **only when the canonical selected cwd equals the parent's canonical cwd**. Any different directory, including another worktree or subdirectory, starts with `--no-approve`; a trusted parent does not silently approve another project's dynamic configuration. Global context remains available, but project-local trust-gated resources in a new worktree are not enabled by delegation. +- Cancellation sends SIGTERM and escalates to SIGKILL after five seconds if needed. On macOS/Linux it targets the child's process group, including ordinary descendants. Reload/shutdown cancels and awaits both active and queued calls, without starting queued children. This is not durable background execution; forcibly killing the parent can bypass cleanup. Windows cancellation targets the direct child only. - The delegate tool is unavailable in children, and child instructions prohibit further delegation or background work. This is a workflow guard, **not a sandbox**: Bash still has normal OS access and can launch programs. Deliberately detached descendants may escape process-group cleanup. -- The child shares the checkout and credentials. There is no worktree isolation or parent permission-hook enforcement inside the child. Do not use this where disabling parent security extensions would violate your expectations. +- Children share the selected checkout and inherited credentials. Isolation exists only when the parent supplies independent worktrees; there is no automatic isolation or parent permission-hook enforcement inside the child. Do not use this where disabling parent security extensions would violate your expectations. - Only a normally stopped final text response counts as success. Provider errors, token limits, nonzero exits, invalid JSONL, and missing final responses are errors; available partial text, usage, and bounded stderr are preserved. - Output is limited to 50 KiB/2,000 lines. Longer results are saved in a private temporary directory, with the path included in the result. These files are not automatically deleted; remove them when no longer needed. Child sessions are otherwise ephemeral (`--no-session`). - Recent tool activity and text stream into the tool row; expand it for more output. Child model usage is returned to Pi's session accounting. @@ -48,4 +79,4 @@ From the `pi/` Stow package directory, with Node.js 24+ and the managed Pi insta node --test .pi/agent/extensions/delegate/runtime.test.ts .pi/agent/extensions/delegate/index.test.mjs ``` -Tests cover awaited completion, streaming, model/tool arguments, UTF-8/JSONL framing, usage, errors, truncation, recursion guards, and cancellation/shutdown. They use local fake child processes, not paid model requests. `index.test.mjs` uses the installed Pi extension loader; `runtime.test.ts` tests the process lifecycle independently. +Tests cover overlapping child processes and the three-child cap, queue cancellation/shutdown, cwd resolution and trust boundaries, queued/timing rendering, awaited completion, streaming, model/tool arguments, UTF-8/JSONL framing, usage, errors, truncation, recursion guards, and cancellation/shutdown. They use local fake child processes, not paid model requests. `index.test.mjs` uses the installed Pi extension loader; `runtime.test.ts` tests the process lifecycle independently. diff --git a/pi/.pi/agent/extensions/delegate/index.test.mjs b/pi/.pi/agent/extensions/delegate/index.test.mjs index 9dc932d..0058d15 100644 --- a/pi/.pi/agent/extensions/delegate/index.test.mjs +++ b/pi/.pi/agent/extensions/delegate/index.test.mjs @@ -1,7 +1,7 @@ // Loads the real extension with the installed Pi loader, but uses a fake child // CLI. No model requests, credentials, or paid API calls are needed. import assert from "node:assert/strict"; -import { mkdtemp, readFile, realpath, rm, stat, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, realpath, rm, stat, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { test } from "node:test"; @@ -28,6 +28,7 @@ async function fixture(fn) { modelRegistry: { find: (provider, id) => provider === "test" && id === "vendor/model" ? { provider, id } : undefined }, }; await writeFile(script, ` + import {existsSync, writeFileSync} from 'node:fs'; let task = ''; for await (const chunk of process.stdin) task += chunk; const event = {type:'message_end', message:{role:'assistant', stopReason:'stop', content:[{type:'text',text:JSON.stringify({task,args:process.argv.slice(2),cwd:process.cwd()})}], @@ -35,7 +36,14 @@ async function fixture(fn) { cost:{input:0,output:0,cacheRead:0,cacheWrite:0,total:0}}}}; if (task.includes('LONG_OUTPUT')) event.message.content[0].text = 'x'.repeat(60000); if (task.includes('FAIL')) { event.message.stopReason='error'; event.message.errorMessage='Provider unavailable'; } - if (task.includes('WAIT')) { process.stdout.write(JSON.stringify(event)+'\\n'); setInterval(()=>{},1000); } + if (task.startsWith('Task: HOLD:')) { + const id = task.slice('Task: HOLD:'.length); + writeFileSync(id+'.started', String(process.pid)); + const timer = setInterval(() => { + if (existsSync(id+'.release')) { clearInterval(timer); process.stdout.write(JSON.stringify(event)+'\\n'); } + }, 10); + } + else if (task.includes('WAIT')) { process.stdout.write(JSON.stringify(event)+'\\n'); setInterval(()=>{},1000); } else process.stdout.write(JSON.stringify(event)+'\\n'); `); await fn({ tool, ctx, extension, directory }); @@ -47,8 +55,8 @@ async function fixture(fn) { test("tool loads with a small schema and forwards model/thinking/tools to an awaited child", async () => { await fixture(async ({ tool, ctx }) => { - assert.equal(tool.executionMode, "sequential"); - assert.deepEqual(Object.keys(tool.parameters.properties).sort(), ["model", "task", "thinking", "tools"]); + assert.equal(tool.executionMode, "parallel"); + assert.deepEqual(Object.keys(tool.parameters.properties).sort(), ["cwd", "model", "task", "thinking", "tools"]); const result = await tool.execute("test", { task: "Plan this", model: "test/vendor/model", thinking: "high", tools: ["read", "grep"], }, undefined, undefined, ctx); @@ -78,6 +86,9 @@ test("rejects unknown, ambiguous, and blank requests before launching", async () { task: "Do work", model: "unknown/model" }, { task: "Do work", model: "model" }, { task: " ", model: "test/vendor/model" }, + { task: "Do work", model: "test/vendor/model", cwd: " " }, + { task: "Do work", model: "test/vendor/model", cwd: "missing-directory" }, + { task: "Do work", model: "test/vendor/model", cwd: "fake-pi.mjs" }, ]) await assert.rejects(tool.execute("test", params, undefined, undefined, ctx)); }); }); @@ -117,6 +128,163 @@ test("session shutdown cancels and awaits active children", { timeout: 5000 }, a }); }); +async function waitFor(check) { + const deadline = Date.now() + 5000; + while (Date.now() < deadline) { + if (await check()) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error("Timed out waiting for child readiness"); +} +const exists = (path) => stat(path).then(() => true, () => false); + +test("explicit cwd resolves relative paths and symlinks without transferring parent trust", async () => { + await fixture(async ({ tool, ctx, directory }) => { + const workspace = join(directory, "separate worktree"); + await mkdir(workspace); + await symlink(workspace, join(directory, "alias")); + ctx.isProjectTrusted = () => true; + const invoke = (cwd) => tool.execute("test", { task: "Inspect", model: "test/vendor/model", cwd }, undefined, undefined, ctx); + const same = JSON.parse((await invoke(".")).details.output); + assert.ok(same.args.includes("--approve")); + for (const cwd of ["separate worktree", workspace, "@alias"]) { + const result = await invoke(cwd); + const child = JSON.parse(result.details.output); + assert.equal(child.cwd, await realpath(workspace)); + assert.equal(result.details.cwd, child.cwd); + assert.ok(child.args.includes("--no-approve")); + assert.ok(!child.args.includes("--approve")); + assert.ok(result.details.elapsedMs >= 0); + assert.ok(result.details.queuedMs >= 0); + } + }); +}); + +test("three children overlap, excess calls queue, and queued cancellation never spawns", { timeout: 10000 }, async () => { + await fixture(async ({ tool, ctx, directory, extension }) => { + const updates = new Map(); + const controllers = Array.from({ length: 5 }, () => new AbortController()); + const runs = []; + try { + // Serialize submission through preflight, not execution: filesystem path + // validation is asynchronous and need not finish in invocation order. + for (const [id, controller] of controllers.entries()) { + runs.push(tool.execute(String(id), { + task: `HOLD:${id}`, model: "test/vendor/model", tools: ["read"], + }, controller.signal, (update) => updates.set(id, update.details), ctx)); + await waitFor(async () => updates.has(id)); + } + await waitFor(async () => (await Promise.all([0, 1, 2].map((id) => exists(join(directory, `${id}.started`))))).every(Boolean)); + assert.equal(updates.get(3).status, "queued"); + assert.equal(updates.get(4).status, "queued"); + assert.equal(await exists(join(directory, "3.started")), false); + assert.equal(await exists(join(directory, "4.started")), false); + controllers[4].abort(); + const cancelled = await runs[4]; + assert.equal(cancelled.details.status, "aborted"); + assert.equal(cancelled.details.exitCode, undefined); + await writeFile(join(directory, "0.release"), ""); + assert.equal((await runs[0]).details.status, "completed"); + await waitFor(() => exists(join(directory, "3.started"))); + for (const id of [1, 2, 3]) await writeFile(join(directory, `${id}.release`), ""); + const results = await Promise.all(runs); + assert.deepEqual(results.map((result) => result.details.status), ["completed", "completed", "completed", "completed", "aborted"]); + assert.equal(await exists(join(directory, "4.started")), false); + } finally { + for (const controller of controllers) controller.abort(); + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + await Promise.allSettled(runs); + } + }); +}); + +test("shutdown cancels running and queued calls, and rejects later launches", { timeout: 10000 }, async () => { + await fixture(async ({ tool, ctx, directory, extension }) => { + const states = new Map(); + const runs = []; + try { + for (let id = 0; id < 5; id++) { + runs.push(tool.execute(String(id), { + task: `HOLD:${id}`, model: "test/vendor/model", + }, undefined, (update) => states.set(id, update.details.status), ctx)); + await waitFor(async () => states.has(id)); + } + await waitFor(async () => (await Promise.all([0, 1, 2].map((id) => exists(join(directory, `${id}.started`))))).every(Boolean)); + assert.equal(states.get(3), "queued"); + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + assert.ok((await Promise.all(runs)).every((result) => result.details.status === "aborted")); + for (const id of [3, 4]) assert.equal(await exists(join(directory, `${id}.started`)), false); + await assert.rejects(tool.execute("later", { task: "Inspect", model: "test/vendor/model" }, undefined, undefined, ctx), /shutting down/); + } finally { + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + await Promise.allSettled(runs); + } + }); +}); + +test("child failures and progress exceptions release queue capacity", { timeout: 10000 }, async () => { + await fixture(async ({ tool, ctx }) => { + for (let i = 0; i < 4; i++) { + const failed = await tool.execute(`fail-${i}`, { task: "FAIL", model: "test/vendor/model" }, undefined, undefined, ctx); + assert.equal(failed.isError, true); + } + await assert.rejects(tool.execute("ui-error", { task: "Inspect", model: "test/vendor/model" }, undefined, + () => { throw new Error("UI failed"); }, ctx), /UI failed/); + const result = await tool.execute("after", { task: "Inspect", model: "test/vendor/model" }, undefined, undefined, ctx); + assert.equal(result.details.status, "completed"); + }); +}); + +test("quiet running and queued children publish live timing and stop updates after shutdown", { timeout: 10000 }, async () => { + await fixture(async ({ tool, ctx, extension }) => { + const updates = new Map(); + let updateCount = 0; + const runs = []; + try { + for (let id = 0; id < 4; id++) { + runs.push(tool.execute(String(id), { task: `HOLD:${id}`, model: "test/vendor/model" }, undefined, + (update) => { updates.set(id, update.details); updateCount++; }, ctx)); + await waitFor(async () => updates.has(id)); + } + const firstSnapshot = updates.get(0); + const firstElapsed = firstSnapshot.elapsedMs; + await waitFor(async () => updates.get(0).elapsedMs >= 1000 && updates.get(3).queuedMs >= 1000); + assert.equal(updates.get(0).status, "running"); + assert.equal(updates.get(3).status, "queued"); + assert.equal(firstSnapshot.elapsedMs, firstElapsed); // Emitted snapshots remain immutable. + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + await Promise.all(runs); + const count = updateCount; + await new Promise((resolve) => setTimeout(resolve, 1100)); + assert.equal(updateCount, count); + } finally { + for (const handler of extension.handlers.get("session_shutdown")) await handler({ reason: "reload" }, ctx); + await Promise.allSettled(runs); + } + }); +}); + +test("pre-aborted tool calls never start children", async () => { + await fixture(async ({ tool, ctx, directory }) => { + const result = await tool.execute("test", { task: "HOLD:cancelled", model: "test/vendor/model" }, AbortSignal.abort(), undefined, ctx); + assert.equal(result.details.status, "aborted"); + assert.equal(await exists(join(directory, "cancelled.started")), false); + }); +}); + +test("renderers preserve queued state and show canonical cwd and timing", async () => { + await fixture(async ({ tool, ctx }) => { + const theme = { fg: (_color, text) => text, bold: (text) => text }; + const result = await tool.execute("test", { task: "Inspect", model: "test/vendor/model" }, undefined, undefined, ctx); + const details = { ...result.details, status: "queued", queuedMs: 2100, elapsedMs: 0 }; + const rendered = tool.renderResult({ ...result, details }, { expanded: false, isPartial: true }, theme).render(200).join("\n"); + assert.match(rendered, /queued · 0s elapsed · 2s queued/); + assert.ok(rendered.includes(result.details.cwd)); + const header = tool.renderCall({ model: "test/vendor/model", task: "Inspect" }, theme, { cwd: ctx.cwd }).render(200).join("\n"); + assert.ok(header.includes(ctx.cwd)); + }); +}); + test("child environment guard prevents registering delegate recursively", async () => { const previous = process.env.PI_DELEGATE_CHILD; process.env.PI_DELEGATE_CHILD = "1"; diff --git a/pi/.pi/agent/extensions/delegate/index.ts b/pi/.pi/agent/extensions/delegate/index.ts index 7c5ff92..c49bb65 100644 --- a/pi/.pi/agent/extensions/delegate/index.ts +++ b/pi/.pi/agent/extensions/delegate/index.ts @@ -1,10 +1,11 @@ -import { mkdtemp, writeFile } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { mkdtemp, realpath, stat, writeFile } from "node:fs/promises"; +import { homedir, tmpdir } from "node:os"; +import { join, resolve } from "node:path"; import { truncateHead, type ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Text } from "@earendil-works/pi-tui"; import { Type } from "typebox"; -import { runDelegate, type DelegateDetails } from "./runtime.ts"; +import { initialDetails, runDelegate, type DelegateDetails, type DelegateOptions } from "./runtime.ts"; +import { DelegateQueue } from "./queue.ts"; const builtinTools = ["read", "bash", "edit", "write", "grep", "find", "ls"] as const; @@ -13,7 +14,10 @@ export default function (pi: ExtensionAPI) { if (process.env.PI_DELEGATE_CHILD === "1") return; const active = new Map>(); + const queue = new DelegateQueue(3); + let closing = false; pi.on("session_shutdown", async () => { + closing = true; for (const controller of active.keys()) controller.abort(); await Promise.allSettled(active.values()); }); @@ -21,11 +25,13 @@ export default function (pi: ExtensionAPI) { pi.registerTool({ name: "delegate", label: "Delegate", - executionMode: "sequential", + executionMode: "parallel", description: [ "Run a self-contained task in a fresh Pi child using an exact provider/model ID.", "Waits for completion and streams progress; no polling or background jobs.", - "The child shares this checkout and credentials, NOT the conversation; include all necessary context.", + "Up to three children run concurrently per session; additional calls queue and remain awaited.", + "cwd selects an existing directory (default: parent cwd); no worktrees are created or removed automatically.", + "The child shares credentials, NOT the conversation; include all necessary context.", "Child extensions and MCP are disabled. This is not a sandbox.", "Default tools: read,bash,edit,write. For read-only planning use read,grep,find,ls.", "Output is capped at 50 KiB/2,000 lines; longer output is saved to a temporary file.", @@ -33,18 +39,24 @@ export default function (pi: ExtensionAPI) { promptSnippet: "Delegate a bounded task to a chosen model and await its result", promptGuidelines: [ "Give delegate a self-contained task, explicit edit permissions, and expected output; children cannot see this conversation.", - "Do not delegate trivial work or use delegate concurrently with edits to the same files. Verify child findings before acting.", + "Before using delegate, decide whether the task needs isolation. For independent write tasks, use wt to prepare separate worktrees first, then pass each path as cwd. Never reuse a worktree being used by another agent or the user.", + "For read-only delegate investigations, prefer tools read,grep,find,ls without a new worktree. If isolation is uncertain, investigate read-only first. Bash and builds/tests may write files.", + "Do not delegate trivial work or run delegate concurrently with parent or sibling writes/branch changes in the same checkout. cwd is not a sandbox; different subdirectories are not separate worktrees. Verify child findings before acting.", + "When preparing delegate worktrees with wt, choose an explicit base; uncommitted changes are not copied. Do not silently approve hooks, assume background setup is finished, or automatically remove worktrees containing useful work.", ], parameters: Type.Object({ task: Type.String({ minLength: 1, description: "Self-contained task, context, constraints, and expected output" }), model: Type.String({ minLength: 3, description: "Exact provider/model ID; model IDs may contain slashes" }), + cwd: Type.Optional(Type.String({ minLength: 1, description: "Existing working directory, absolute or relative to parent cwd; defaults to parent cwd. Prepare independent write worktrees with wt first." })), thinking: Type.Optional(Type.String({ enum: ["off", "minimal", "low", "medium", "high", "xhigh", "max"] })), tools: Type.Optional(Type.Array(Type.String({ enum: [...builtinTools] }), { minItems: 1, uniqueItems: true, description: "Built-in tool allowlist; omit for read,bash,edit,write", })), }), async execute(_id, params, signal, onUpdate, ctx) { + if (closing) throw new Error("Delegate is shutting down."); if (!params.task.trim()) throw new Error("Delegated task must not be blank."); + if (params.cwd !== undefined && !params.cwd.trim()) throw new Error("Delegate cwd must not be blank."); const slash = params.model.indexOf("/"); if (slash < 1 || slash === params.model.length - 1) throw new Error("Use an exact provider/model ID."); const provider = params.model.slice(0, slash); @@ -53,16 +65,48 @@ export default function (pi: ExtensionAPI) { const controller = new AbortController(); const childSignal = signal ? AbortSignal.any([signal, controller.signal]) : controller.signal; - const run = runDelegate({ + let path = params.cwd ?? ctx.cwd; + if (path.startsWith("@")) path = path.slice(1); + if (path === "~" || path.startsWith("~/")) path = join(homedir(), path.slice(1)); + const options: DelegateOptions = { task: params.task, provider, model, tools: params.tools, thinking: params.thinking ?? ctx.thinkingLevel, - cwd: ctx.cwd, trusted: ctx.isProjectTrusted(), - }, childSignal, (details) => { + cwd: resolve(ctx.cwd, path), trusted: false, queuedAt: Date.now(), + }; + const publish = (details: DelegateDetails) => { onUpdate?.({ - content: [{ type: "text", text: truncateHead(details.output || "Working…").content }], + content: [{ type: "text", text: truncateHead(details.output || (details.status === "queued" ? "Queued (three-child session limit)…" : "Working…")).content }], details, }); - }); + }; + const run = (async () => { + options.cwd = await realpath(options.cwd); + if (!(await stat(options.cwd)).isDirectory()) throw new Error("Delegate cwd must be an existing directory."); + // A trusted parent must not silently approve a different project/path. + options.trusted = options.cwd === await realpath(ctx.cwd) && ctx.isProjectTrusted(); + const queued = initialDetails(options); + const updateQueue = () => { + queued.queuedMs = Date.now() - options.queuedAt!; + publish({ ...queued }); + }; + let timer: ReturnType | undefined; + let release: (() => void) | undefined; + try { + updateQueue(); + timer = setInterval(() => { + try { updateQueue(); } catch { controller.abort(); } + }, 1000); + release = await queue.acquire(childSignal); + clearInterval(timer); + if (!release || childSignal.aborted) { + return { ...queued, status: "aborted" as const, queuedMs: Date.now() - options.queuedAt!, error: "Delegation cancelled before launch." }; + } + return await runDelegate(options, childSignal, publish); + } finally { + if (timer) clearInterval(timer); + release?.(); + } + })(); active.set(controller, run); try { const details = await run; @@ -85,22 +129,25 @@ export default function (pi: ExtensionAPI) { active.delete(controller); } }, - renderCall(args, theme) { + renderCall(args, theme, context) { const model = args.model ?? "…"; const task = args.task?.replace(/\s+/g, " ").slice(0, 140) ?? "…"; - return new Text(`${theme.fg("toolTitle", theme.bold("delegate "))}${theme.fg("accent", model)}\n${theme.fg("dim", task)}`, 0, 0); + const cwd = args.cwd ?? context.cwd; + return new Text(`${theme.fg("toolTitle", theme.bold("delegate "))}${theme.fg("accent", model)}\n${theme.fg("dim", task)}\n${theme.fg("muted", `cwd: ${cwd}`)}`, 0, 0); }, - renderResult(result, { expanded, isPartial }, theme) { + renderResult(result, { expanded }, theme) { const details = result.details as DelegateDetails | undefined; const content = result.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); if (!details) return new Text(content, 0, 0); - const status = isPartial ? "running" : details.status; - const color = status === "completed" ? "success" : status === "running" ? "muted" : "error"; + const status = details.status; + const color = status === "completed" ? "success" : status === "queued" ? "warning" : status === "running" ? "muted" : "error"; + const timing = `${Math.floor((details.elapsedMs ?? 0) / 1000)}s elapsed · ${Math.floor((details.queuedMs ?? 0) / 1000)}s queued`; const usage = `${details.turns} turns · ↑${details.usage.input} ↓${details.usage.output} · $${details.usage.cost.total.toFixed(4)}`; const activity = (expanded ? details.activity : details.activity.slice(-3)).join("\n"); const preview = expanded ? content : content.split("\n").slice(-6).join("\n").slice(-1500); return new Text([ - theme.fg(color, status), theme.fg("dim", usage), + theme.fg(color, `${status} · ${timing}`), + details.cwd && theme.fg("muted", `cwd: ${details.cwd}`), theme.fg("dim", usage), activity && theme.fg("muted", activity), preview, ].filter(Boolean).join("\n"), 0, 0); }, diff --git a/pi/.pi/agent/extensions/delegate/queue.ts b/pi/.pi/agent/extensions/delegate/queue.ts new file mode 100644 index 0000000..f329ae3 --- /dev/null +++ b/pi/.pi/agent/extensions/delegate/queue.ts @@ -0,0 +1,40 @@ +// Session-local FIFO limit. Waiting calls remain awaited and can be cancelled +// without ever starting a subprocess. This does not coordinate other sessions. +export class DelegateQueue { + private running = 0; + private waiting: Array<() => void> = []; + + private readonly limit: number; + + constructor(limit = 3) { + this.limit = limit; + } + + acquire(signal: AbortSignal): Promise<(() => void) | undefined> { + if (signal.aborted) return Promise.resolve(undefined); + return new Promise((resolve) => { + const cancel = () => { + this.waiting = this.waiting.filter((entry) => entry !== start); + resolve(undefined); + }; + const start = () => { + signal.removeEventListener("abort", cancel); + this.running++; + let released = false; + resolve(() => { + if (released) return; + released = true; + this.running--; + this.drain(); + }); + }; + signal.addEventListener("abort", cancel, { once: true }); + this.waiting.push(start); + this.drain(); + }); + } + + private drain() { + while (this.running < this.limit && this.waiting.length) this.waiting.shift()!(); + } +} diff --git a/pi/.pi/agent/extensions/delegate/runtime.test.ts b/pi/.pi/agent/extensions/delegate/runtime.test.ts index 6ccb096..3db42b0 100644 --- a/pi/.pi/agent/extensions/delegate/runtime.test.ts +++ b/pi/.pi/agent/extensions/delegate/runtime.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { test } from "node:test"; import { childArguments, runDelegate, type DelegateOptions } from "./runtime.ts"; +import { DelegateQueue } from "./queue.ts"; const options: DelegateOptions = { task: "Inspect this task", provider: "test-provider", model: "vendor/model", @@ -32,6 +33,28 @@ async function waitFor(check: () => Promise) { throw new Error("Timed out waiting for child readiness"); } +test("queue preserves FIFO, skips a cancelled head waiter, and releases slots only once", async () => { + const queue = new DelegateQueue(1); + const signal = new AbortController().signal; + const first = await queue.acquire(signal); + const cancelled = new AbortController(); + const head = queue.acquire(cancelled.signal); + const order: number[] = []; + const second = queue.acquire(signal).then((release) => { order.push(2); return release; }); + const third = queue.acquire(signal).then((release) => { order.push(3); return release; }); + cancelled.abort(); + assert.equal(await head, undefined); + first!(); + const releaseSecond = await second; + assert.deepEqual(order, [2]); + first!(); // A duplicate release must not start the third waiter. + await Promise.resolve(); + assert.deepEqual(order, [2]); + releaseSecond!(); + (await third)!(); + assert.deepEqual(order, [2, 3]); +}); + test("child arguments pin provider/model, disable extensions, and preserve trust/tool scope", () => { const args = childArguments({ ...options, thinking: "high", tools: ["read", "grep"] }); assert.ok(args.includes("--no-extensions")); diff --git a/pi/.pi/agent/extensions/delegate/runtime.ts b/pi/.pi/agent/extensions/delegate/runtime.ts index 14b91c9..960af26 100644 --- a/pi/.pi/agent/extensions/delegate/runtime.ts +++ b/pi/.pi/agent/extensions/delegate/runtime.ts @@ -21,11 +21,15 @@ export interface DelegateOptions { tools?: string[]; cwd: string; trusted: boolean; + queuedAt?: number; } export interface DelegateDetails { model: string; - status: "running" | "completed" | "failed" | "aborted"; + status: "queued" | "running" | "completed" | "failed" | "aborted"; + cwd: string; + queuedMs: number; + elapsedMs: number; output: string; stderr: string; activity: string[]; @@ -68,6 +72,18 @@ export function childArguments(options: DelegateOptions): string[] { return args; } +export function initialDetails(options: DelegateOptions): DelegateDetails { + return { + model: `${options.provider}/${options.model}`, + status: "queued", cwd: options.cwd, queuedMs: 0, elapsedMs: 0, + output: "", stderr: "", activity: [], turns: 0, + usage: { + input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }; +} + export async function runDelegate( options: DelegateOptions, signal?: AbortSignal, @@ -75,14 +91,10 @@ export async function runDelegate( invocation: Invocation = piInvocation(), killDelayMs = 5000, ): Promise { - const details: DelegateDetails = { - model: `${options.provider}/${options.model}`, - status: "running", output: "", stderr: "", activity: [], turns: 0, - usage: { - input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }, - }; + const startedAt = Date.now(); + const details = initialDetails(options); + details.status = "running"; + details.queuedMs = Math.max(0, startedAt - (options.queuedAt ?? startedAt)); if (signal?.aborted) return { ...details, status: "aborted", error: "Delegation cancelled before launch." }; return new Promise((resolve) => { @@ -107,6 +119,7 @@ export async function runDelegate( let assistantError: string | undefined; const emit = () => { + details.elapsedMs = Date.now() - startedAt; // Snapshots must not mutate already emitted tool results/session entries. try { onUpdate?.({ ...details, activity: [...details.activity], usage: { ...details.usage, cost: { ...details.usage.cost } } }); @@ -235,6 +248,8 @@ export async function runDelegate( signal?.removeEventListener("abort", abort); if (killTimer) clearTimeout(killTimer); if (updateTimer) clearTimeout(updateTimer); + clearInterval(heartbeat); + details.elapsedMs = Date.now() - startedAt; if (buffer.trim()) processLine(buffer); details.exitCode = code; details.exitSignal = exitSignal; @@ -251,6 +266,7 @@ export async function runDelegate( resolve(details); }); + const heartbeat = setInterval(emit, 1000); signal?.addEventListener("abort", abort, { once: true }); if (signal?.aborted) abort(); emit(); From f6d8b54120c26c79ff78192d1a70459555901b3b Mon Sep 17 00:00:00 2001 From: Colton Padden Date: Tue, 10 Nov 2026 15:41:40 -0500 Subject: [PATCH 3/4] pi: extract delegate prompt constants --- pi/.pi/agent/extensions/delegate/index.ts | 52 +++++++++++++++-------- 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/pi/.pi/agent/extensions/delegate/index.ts b/pi/.pi/agent/extensions/delegate/index.ts index c49bb65..f957772 100644 --- a/pi/.pi/agent/extensions/delegate/index.ts +++ b/pi/.pi/agent/extensions/delegate/index.ts @@ -9,6 +9,39 @@ import { DelegateQueue } from "./queue.ts"; const builtinTools = ["read", "bash", "edit", "write", "grep", "find", "ls"] as const; +const DELEGATE_DESCRIPTION = ` +Run a self-contained task in a fresh Pi child using an exact provider/model ID. +Waits for completion and streams progress; no polling or background jobs. +Up to three children run concurrently per session; additional calls queue and remain awaited. +cwd selects an existing directory (default: parent cwd); +no worktrees are created or removed automatically. +The child shares credentials, NOT the conversation; include all necessary context. +Child extensions and MCP are disabled. This is not a sandbox. +Default tools: read,bash,edit,write. For read-only planning use read,grep,find,ls. +Output is capped at 50 KiB/2,000 lines; longer output is saved to a temporary file. +`.trim().replace(/\n/g, " "); + +const DELEGATE_PROMPT_GUIDELINES = ` +Give delegate a self-contained task, explicit edit permissions, and expected output; +children cannot see this conversation. + +Before using delegate, decide whether the task needs isolation. +For independent write tasks, use wt to prepare separate worktrees first, +then pass each path as cwd. +Never reuse a worktree being used by another agent or the user. + +For read-only delegate investigations, prefer tools read,grep,find,ls without a new worktree. +If isolation is uncertain, investigate read-only first. Bash and builds/tests may write files. + +Do not delegate trivial work or run delegate concurrently with parent or sibling +writes/branch changes in the same checkout. cwd is not a sandbox; +different subdirectories are not separate worktrees. Verify child findings before acting. + +When preparing delegate worktrees with wt, choose an explicit base; +uncommitted changes are not copied. Do not silently approve hooks, +assume background setup is finished, or automatically remove worktrees containing useful work. +`.trim().split("\n\n").map((guideline) => guideline.replace(/\n/g, " ")); + export default function (pi: ExtensionAPI) { // Defense in depth if this extension is explicitly loaded by a child. if (process.env.PI_DELEGATE_CHILD === "1") return; @@ -26,24 +59,9 @@ export default function (pi: ExtensionAPI) { name: "delegate", label: "Delegate", executionMode: "parallel", - description: [ - "Run a self-contained task in a fresh Pi child using an exact provider/model ID.", - "Waits for completion and streams progress; no polling or background jobs.", - "Up to three children run concurrently per session; additional calls queue and remain awaited.", - "cwd selects an existing directory (default: parent cwd); no worktrees are created or removed automatically.", - "The child shares credentials, NOT the conversation; include all necessary context.", - "Child extensions and MCP are disabled. This is not a sandbox.", - "Default tools: read,bash,edit,write. For read-only planning use read,grep,find,ls.", - "Output is capped at 50 KiB/2,000 lines; longer output is saved to a temporary file.", - ].join(" "), + description: DELEGATE_DESCRIPTION, promptSnippet: "Delegate a bounded task to a chosen model and await its result", - promptGuidelines: [ - "Give delegate a self-contained task, explicit edit permissions, and expected output; children cannot see this conversation.", - "Before using delegate, decide whether the task needs isolation. For independent write tasks, use wt to prepare separate worktrees first, then pass each path as cwd. Never reuse a worktree being used by another agent or the user.", - "For read-only delegate investigations, prefer tools read,grep,find,ls without a new worktree. If isolation is uncertain, investigate read-only first. Bash and builds/tests may write files.", - "Do not delegate trivial work or run delegate concurrently with parent or sibling writes/branch changes in the same checkout. cwd is not a sandbox; different subdirectories are not separate worktrees. Verify child findings before acting.", - "When preparing delegate worktrees with wt, choose an explicit base; uncommitted changes are not copied. Do not silently approve hooks, assume background setup is finished, or automatically remove worktrees containing useful work.", - ], + promptGuidelines: DELEGATE_PROMPT_GUIDELINES, parameters: Type.Object({ task: Type.String({ minLength: 1, description: "Self-contained task, context, constraints, and expected output" }), model: Type.String({ minLength: 3, description: "Exact provider/model ID; model IDs may contain slashes" }), From c7d110222a6ebcf45fba7c4dfb84e663ed20b7fd Mon Sep 17 00:00:00 2001 From: Colton Padden Date: Tue, 10 Nov 2026 15:45:20 -0500 Subject: [PATCH 4/4] pi: simplify delegate extension structure --- pi/.pi/agent/extensions/delegate/index.ts | 232 ++++++++++------ pi/.pi/agent/extensions/delegate/queue.ts | 4 +- pi/.pi/agent/extensions/delegate/runtime.ts | 278 ++++++++++++-------- 3 files changed, 326 insertions(+), 188 deletions(-) diff --git a/pi/.pi/agent/extensions/delegate/index.ts b/pi/.pi/agent/extensions/delegate/index.ts index f957772..7685f58 100644 --- a/pi/.pi/agent/extensions/delegate/index.ts +++ b/pi/.pi/agent/extensions/delegate/index.ts @@ -4,12 +4,25 @@ import { join, resolve } from "node:path"; import { truncateHead, type ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Text } from "@earendil-works/pi-tui"; import { Type } from "typebox"; -import { initialDetails, runDelegate, type DelegateDetails, type DelegateOptions } from "./runtime.ts"; import { DelegateQueue } from "./queue.ts"; +import { initialDetails, runDelegate, type DelegateDetails, type DelegateOptions } from "./runtime.ts"; + +const MAX_CONCURRENT_CHILDREN = 3; +const QUEUE_HEARTBEAT_MS = 1000; +const BUILTIN_TOOLS = ["read", "bash", "edit", "write", "grep", "find", "ls"]; +const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high", "xhigh", "max"]; -const builtinTools = ["read", "bash", "edit", "write", "grep", "find", "ls"] as const; +const STATUS_COLORS = { + queued: "warning", + running: "muted", + completed: "success", + failed: "error", + aborted: "error", +} as const; -const DELEGATE_DESCRIPTION = ` +const unwrap = (text: string) => text.trim().replace(/\n/g, " "); + +const DELEGATE_DESCRIPTION = unwrap(` Run a self-contained task in a fresh Pi child using an exact provider/model ID. Waits for completion and streams progress; no polling or background jobs. Up to three children run concurrently per session; additional calls queue and remain awaited. @@ -19,7 +32,7 @@ The child shares credentials, NOT the conversation; include all necessary contex Child extensions and MCP are disabled. This is not a sandbox. Default tools: read,bash,edit,write. For read-only planning use read,grep,find,ls. Output is capped at 50 KiB/2,000 lines; longer output is saved to a temporary file. -`.trim().replace(/\n/g, " "); +`); const DELEGATE_PROMPT_GUIDELINES = ` Give delegate a self-contained task, explicit edit permissions, and expected output; @@ -40,15 +53,82 @@ different subdirectories are not separate worktrees. Verify child findings befor When preparing delegate worktrees with wt, choose an explicit base; uncommitted changes are not copied. Do not silently approve hooks, assume background setup is finished, or automatically remove worktrees containing useful work. -`.trim().split("\n\n").map((guideline) => guideline.replace(/\n/g, " ")); +`.split("\n\n").map(unwrap); + +const DELEGATE_PARAMETERS = Type.Object({ + task: Type.String({ + minLength: 1, + description: "Self-contained task, context, constraints, and expected output", + }), + model: Type.String({ + minLength: 3, + description: "Exact provider/model ID; model IDs may contain slashes", + }), + cwd: Type.Optional(Type.String({ + minLength: 1, + description: unwrap(` +Existing working directory, absolute or relative to parent cwd; defaults to parent cwd. +Prepare independent write worktrees with wt first. +`), + })), + thinking: Type.Optional(Type.String({ enum: THINKING_LEVELS })), + tools: Type.Optional(Type.Array(Type.String({ enum: BUILTIN_TOOLS }), { + minItems: 1, + uniqueItems: true, + description: "Built-in tool allowlist; omit for read,bash,edit,write", + })), +}); + +function parseModelId(id: string): { provider: string; model: string } { + const slash = id.indexOf("/"); + if (slash < 1 || slash === id.length - 1) throw new Error("Use an exact provider/model ID."); + return { provider: id.slice(0, slash), model: id.slice(slash + 1) }; +} + +function resolvePath(path: string, base: string): string { + if (path.startsWith("@")) path = path.slice(1); + if (path === "~" || path.startsWith("~/")) path = join(homedir(), path.slice(1)); + return resolve(base, path); +} + +async function resolveDirectory(path: string, base: string): Promise { + const directory = await realpath(resolvePath(path, base)); + if (!(await stat(directory)).isDirectory()) { + throw new Error("Delegate cwd must be an existing directory."); + } + return directory; +} + +async function toToolResult(details: DelegateDetails) { + const failed = details.status !== "completed"; + const full = [details.error, details.output, failed && details.stderr] + .filter(Boolean) + .join("\n\n"); + const { content, truncated } = truncateHead(full); + let text = content; + if (truncated) { + details.outputFile = join(await mkdtemp(join(tmpdir(), "pi-delegate-")), "output.txt"); + await writeFile(details.outputFile, full, { mode: 0o600 }); + text += `\n\n[Output truncated. Full output: ${details.outputFile}]`; + } + return { + content: [{ type: "text" as const, text: text || "(no output)" }], + details, + usage: details.usage, + isError: failed, + }; +} + +const seconds = (ms?: number) => `${Math.floor((ms ?? 0) / 1000)}s`; export default function (pi: ExtensionAPI) { // Defense in depth if this extension is explicitly loaded by a child. if (process.env.PI_DELEGATE_CHILD === "1") return; const active = new Map>(); - const queue = new DelegateQueue(3); + const queue = new DelegateQueue(MAX_CONCURRENT_CHILDREN); let closing = false; + pi.on("session_shutdown", async () => { closing = true; for (const controller of active.keys()) controller.abort(); @@ -62,111 +142,111 @@ export default function (pi: ExtensionAPI) { description: DELEGATE_DESCRIPTION, promptSnippet: "Delegate a bounded task to a chosen model and await its result", promptGuidelines: DELEGATE_PROMPT_GUIDELINES, - parameters: Type.Object({ - task: Type.String({ minLength: 1, description: "Self-contained task, context, constraints, and expected output" }), - model: Type.String({ minLength: 3, description: "Exact provider/model ID; model IDs may contain slashes" }), - cwd: Type.Optional(Type.String({ minLength: 1, description: "Existing working directory, absolute or relative to parent cwd; defaults to parent cwd. Prepare independent write worktrees with wt first." })), - thinking: Type.Optional(Type.String({ enum: ["off", "minimal", "low", "medium", "high", "xhigh", "max"] })), - tools: Type.Optional(Type.Array(Type.String({ enum: [...builtinTools] }), { - minItems: 1, uniqueItems: true, description: "Built-in tool allowlist; omit for read,bash,edit,write", - })), - }), + parameters: DELEGATE_PARAMETERS, + async execute(_id, params, signal, onUpdate, ctx) { if (closing) throw new Error("Delegate is shutting down."); if (!params.task.trim()) throw new Error("Delegated task must not be blank."); - if (params.cwd !== undefined && !params.cwd.trim()) throw new Error("Delegate cwd must not be blank."); - const slash = params.model.indexOf("/"); - if (slash < 1 || slash === params.model.length - 1) throw new Error("Use an exact provider/model ID."); - const provider = params.model.slice(0, slash); - const model = params.model.slice(slash + 1); - if (!ctx.modelRegistry.find(provider, model)) throw new Error(`Unknown model: ${params.model}. Use pi --list-models to find exact IDs.`); + if (params.cwd !== undefined && !params.cwd.trim()) { + throw new Error("Delegate cwd must not be blank."); + } + const { provider, model } = parseModelId(params.model); + if (!ctx.modelRegistry.find(provider, model)) { + throw new Error(`Unknown model: ${params.model}. Use pi --list-models to find exact IDs.`); + } + const queuedAt = Date.now(); const controller = new AbortController(); const childSignal = signal ? AbortSignal.any([signal, controller.signal]) : controller.signal; - let path = params.cwd ?? ctx.cwd; - if (path.startsWith("@")) path = path.slice(1); - if (path === "~" || path.startsWith("~/")) path = join(homedir(), path.slice(1)); - const options: DelegateOptions = { - task: params.task, provider, model, tools: params.tools, - thinking: params.thinking ?? ctx.thinkingLevel, - cwd: resolve(ctx.cwd, path), trusted: false, queuedAt: Date.now(), - }; + const publish = (details: DelegateDetails) => { + const placeholder = details.status === "queued" + ? "Queued (three-child session limit)…" + : "Working…"; onUpdate?.({ - content: [{ type: "text", text: truncateHead(details.output || (details.status === "queued" ? "Queued (three-child session limit)…" : "Working…")).content }], + content: [{ type: "text", text: truncateHead(details.output || placeholder).content }], details, }); }; + const run = (async () => { - options.cwd = await realpath(options.cwd); - if (!(await stat(options.cwd)).isDirectory()) throw new Error("Delegate cwd must be an existing directory."); - // A trusted parent must not silently approve a different project/path. - options.trusted = options.cwd === await realpath(ctx.cwd) && ctx.isProjectTrusted(); - const queued = initialDetails(options); - const updateQueue = () => { - queued.queuedMs = Date.now() - options.queuedAt!; - publish({ ...queued }); + const cwd = await resolveDirectory(params.cwd ?? ctx.cwd, ctx.cwd); + const options: DelegateOptions = { + task: params.task, + provider, + model, + tools: params.tools, + thinking: params.thinking ?? ctx.thinkingLevel, + cwd, + // A trusted parent must not silently approve a different project/path. + trusted: cwd === await realpath(ctx.cwd) && ctx.isProjectTrusted(), + queuedAt, }; - let timer: ReturnType | undefined; + + const queued = initialDetails(options); + const publishQueued = () => publish({ ...queued, queuedMs: Date.now() - queuedAt }); + publishQueued(); + const heartbeat = setInterval(() => { + try { publishQueued(); } catch { controller.abort(); } + }, QUEUE_HEARTBEAT_MS); + let release: (() => void) | undefined; try { - updateQueue(); - timer = setInterval(() => { - try { updateQueue(); } catch { controller.abort(); } - }, 1000); release = await queue.acquire(childSignal); - clearInterval(timer); - if (!release || childSignal.aborted) { - return { ...queued, status: "aborted" as const, queuedMs: Date.now() - options.queuedAt!, error: "Delegation cancelled before launch." }; - } + } finally { + clearInterval(heartbeat); + } + if (!release) { + return { + ...queued, + status: "aborted" as const, + queuedMs: Date.now() - queuedAt, + error: "Delegation cancelled before launch.", + }; + } + try { return await runDelegate(options, childSignal, publish); } finally { - if (timer) clearInterval(timer); - release?.(); + release(); } })(); + active.set(controller, run); try { - const details = await run; - const text = [details.error, details.output, details.status !== "completed" ? details.stderr : ""].filter(Boolean).join("\n\n"); - const truncated = truncateHead(text); - let output = truncated.content; - if (truncated.truncated) { - const directory = await mkdtemp(join(tmpdir(), "pi-delegate-")); - details.outputFile = join(directory, "output.txt"); - await writeFile(details.outputFile, text, { mode: 0o600 }); - output += `\n\n[Output truncated. Full output: ${details.outputFile}]`; - } - return { - content: [{ type: "text" as const, text: output || "(no output)" }], - details, - usage: details.usage, - isError: details.status !== "completed", - }; + return await toToolResult(await run); } finally { active.delete(controller); } }, + renderCall(args, theme, context) { - const model = args.model ?? "…"; const task = args.task?.replace(/\s+/g, " ").slice(0, 140) ?? "…"; - const cwd = args.cwd ?? context.cwd; - return new Text(`${theme.fg("toolTitle", theme.bold("delegate "))}${theme.fg("accent", model)}\n${theme.fg("dim", task)}\n${theme.fg("muted", `cwd: ${cwd}`)}`, 0, 0); + return new Text([ + theme.fg("toolTitle", theme.bold("delegate ")) + theme.fg("accent", args.model ?? "…"), + theme.fg("dim", task), + theme.fg("muted", `cwd: ${args.cwd ?? context.cwd}`), + ].join("\n"), 0, 0); }, + renderResult(result, { expanded }, theme) { const details = result.details as DelegateDetails | undefined; - const content = result.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); + const content = result.content + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("\n"); if (!details) return new Text(content, 0, 0); - const status = details.status; - const color = status === "completed" ? "success" : status === "queued" ? "warning" : status === "running" ? "muted" : "error"; - const timing = `${Math.floor((details.elapsedMs ?? 0) / 1000)}s elapsed · ${Math.floor((details.queuedMs ?? 0) / 1000)}s queued`; - const usage = `${details.turns} turns · ↑${details.usage.input} ↓${details.usage.output} · $${details.usage.cost.total.toFixed(4)}`; + + const { status, usage } = details; + const timing = `${seconds(details.elapsedMs)} elapsed · ${seconds(details.queuedMs)} queued`; + const cost = `$${usage.cost.total.toFixed(4)}`; const activity = (expanded ? details.activity : details.activity.slice(-3)).join("\n"); const preview = expanded ? content : content.split("\n").slice(-6).join("\n").slice(-1500); return new Text([ - theme.fg(color, `${status} · ${timing}`), - details.cwd && theme.fg("muted", `cwd: ${details.cwd}`), theme.fg("dim", usage), - activity && theme.fg("muted", activity), preview, + theme.fg(STATUS_COLORS[status] ?? "error", `${status} · ${timing}`), + details.cwd && theme.fg("muted", `cwd: ${details.cwd}`), + theme.fg("dim", `${details.turns} turns · ↑${usage.input} ↓${usage.output} · ${cost}`), + activity && theme.fg("muted", activity), + preview, ].filter(Boolean).join("\n"), 0, 0); }, }); diff --git a/pi/.pi/agent/extensions/delegate/queue.ts b/pi/.pi/agent/extensions/delegate/queue.ts index f329ae3..3386bc1 100644 --- a/pi/.pi/agent/extensions/delegate/queue.ts +++ b/pi/.pi/agent/extensions/delegate/queue.ts @@ -1,9 +1,7 @@ -// Session-local FIFO limit. Waiting calls remain awaited and can be cancelled -// without ever starting a subprocess. This does not coordinate other sessions. +// Session-local FIFO concurrency limit; acquire resolves undefined if cancelled while waiting. export class DelegateQueue { private running = 0; private waiting: Array<() => void> = []; - private readonly limit: number; constructor(limit = 3) { diff --git a/pi/.pi/agent/extensions/delegate/runtime.ts b/pi/.pi/agent/extensions/delegate/runtime.ts index 960af26..d29b3f4 100644 --- a/pi/.pi/agent/extensions/delegate/runtime.ts +++ b/pi/.pi/agent/extensions/delegate/runtime.ts @@ -6,6 +6,19 @@ import { basename } from "node:path"; import type { Usage } from "@earendil-works/pi-ai"; import type { JsonAgentSessionEvent } from "@earendil-works/pi-coding-agent"; +const DEFAULT_TOOLS = ["read", "bash", "edit", "write"]; +const MAX_RECORD_BYTES = 8 * 1024 * 1024; +const MAX_STDERR_CHARS = 65536; +const MAX_STREAMED_OUTPUT_CHARS = 16384; +const MAX_ACTIVITY_ENTRIES = 20; +const MAX_ACTIVITY_CHARS = 500; +const UPDATE_THROTTLE_MS = 100; +const HEARTBEAT_MS = 1000; +const RECORD_TOO_LARGE = "Child JSONL record exceeded 8 MiB."; + +const USAGE_KEYS = ["input", "output", "cacheRead", "cacheWrite", "totalTokens"] as const; +const COST_KEYS = ["input", "output", "cacheRead", "cacheWrite", "total"] as const; + export const CHILD_INSTRUCTIONS = [ "You are a delegated child working on one self-contained task.", "You do not have the parent's conversation. Follow the task and local repository instructions.", @@ -46,8 +59,13 @@ export interface Invocation { args: string[]; } -// Reuse the running Pi version, including standalone builds, rather than a -// potentially different `pi` on PATH. Mirrors the upstream example. +interface Outcome { + assistantSeen: boolean; + stopReason?: string; + assistantError?: string; +} + +// Reuse the running Pi (including standalone builds), not whichever `pi` is on PATH. export function piInvocation(): Invocation { const script = process.argv[1]; if (script && !script.startsWith("/$bunfs/root/") && existsSync(script)) { @@ -65,7 +83,7 @@ export function childArguments(options: DelegateOptions): string[] { "--no-extensions", "--no-mcp", "--no-prompt-templates", options.trusted ? "--approve" : "--no-approve", "--provider", options.provider, "--model", options.model, - "--tools", (options.tools ?? ["read", "bash", "edit", "write"]).join(","), + "--tools", (options.tools ?? DEFAULT_TOOLS).join(","), "--append-system-prompt", CHILD_INSTRUCTIONS, ]; if (options.thinking) args.push("--thinking", options.thinking); @@ -75,8 +93,14 @@ export function childArguments(options: DelegateOptions): string[] { export function initialDetails(options: DelegateOptions): DelegateDetails { return { model: `${options.provider}/${options.model}`, - status: "queued", cwd: options.cwd, queuedMs: 0, elapsedMs: 0, - output: "", stderr: "", activity: [], turns: 0, + status: "queued", + cwd: options.cwd, + queuedMs: 0, + elapsedMs: 0, + output: "", + stderr: "", + activity: [], + turns: 0, usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, @@ -84,6 +108,83 @@ export function initialDetails(options: DelegateOptions): DelegateDetails { }; } +function snapshot(details: DelegateDetails): DelegateDetails { + return { + ...details, + activity: [...details.activity], + usage: { ...details.usage, cost: { ...details.usage.cost } }, + }; +} + +function parseEvent(line: string): JsonAgentSessionEvent { + const parsed = JSON.parse(line); + if (!parsed || typeof parsed.type !== "string") throw new Error("Missing event type"); + return parsed; +} + +function recordActivity(details: DelegateDetails, text: string) { + details.activity.push(text.slice(0, MAX_ACTIVITY_CHARS)); + details.activity = details.activity.slice(-MAX_ACTIVITY_ENTRIES); +} + +// Returns whether there is progress to publish. +function applyEvent(event: JsonAgentSessionEvent, details: DelegateDetails, outcome: Outcome): boolean { + switch (event.type) { + case "message_start": + if (event.message.role === "assistant") details.output = ""; + return false; + case "message_update": { + const delta = event.assistantMessageEvent; + if (delta.type !== "text_delta") return false; + details.output = (details.output + delta.delta).slice(-MAX_STREAMED_OUTPUT_CHARS); + return true; + } + case "message_end": { + const message = event.message; + if (message.role !== "assistant") return false; + outcome.assistantSeen = true; + outcome.stopReason = message.stopReason; + outcome.assistantError = message.errorMessage; + details.turns++; + details.output = message.content + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("\n"); + for (const key of USAGE_KEYS) details.usage[key] += message.usage?.[key] ?? 0; + for (const key of COST_KEYS) details.usage.cost[key] += message.usage?.cost?.[key] ?? 0; + return true; + } + case "tool_execution_start": { + const args = event.args as Record; + const target = args?.command ?? args?.path ?? args?.pattern ?? ""; + recordActivity(details, `→ ${event.toolName} ${String(target)}`); + return true; + } + case "tool_execution_end": + recordActivity(details, `${event.isError ? "✗" : "✓"} ${event.toolName}`); + return true; + case "auto_retry_start": + recordActivity(details, `Retry ${event.attempt}: ${event.errorMessage}`); + return true; + default: + return false; + } +} + +function failureReason( + code: number | null, + exitSignal: string | null, + outcome: Outcome, + output: string, +): string | undefined { + if (code !== 0) return `Child exited with ${exitSignal ? `signal ${exitSignal}` : `code ${code}`}.`; + if (outcome.assistantSeen && outcome.stopReason !== "stop") { + return outcome.assistantError || `Child ended with ${outcome.stopReason ?? "no stop reason"}.`; + } + if (!outcome.assistantSeen || !output.trim()) return "Child exited without a final text response."; + return undefined; +} + export async function runDelegate( options: DelegateOptions, signal?: AbortSignal, @@ -95,11 +196,12 @@ export async function runDelegate( const details = initialDetails(options); details.status = "running"; details.queuedMs = Math.max(0, startedAt - (options.queuedAt ?? startedAt)); - if (signal?.aborted) return { ...details, status: "aborted", error: "Delegation cancelled before launch." }; + if (signal?.aborted) { + return { ...details, status: "aborted", error: "Delegation cancelled before launch." }; + } return new Promise((resolve) => { - // A dedicated POSIX process group lets cancellation reach child shell tools, - // not just Pi itself. It is still piped, referenced, and awaited: never unref(). + // A POSIX process group lets cancellation reach the child's own subprocesses. const grouped = process.platform !== "win32"; const proc = spawn(invocation.command, [...invocation.args, ...childArguments(options)], { cwd: options.cwd, @@ -108,34 +210,19 @@ export async function runDelegate( stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, PI_DELEGATE_CHILD: "1" }, }); + + const outcome: Outcome = { assistantSeen: false }; let closed = false; - let killTimer: ReturnType | undefined; - let updateTimer: ReturnType | undefined; let buffer = ""; - let protocolError: string | undefined; let spawnError: string | undefined; - let assistantSeen = false; - let stopReason: string | undefined; - let assistantError: string | undefined; + let protocolError: string | undefined; + let killTimer: ReturnType | undefined; + let updateTimer: ReturnType | undefined; - const emit = () => { - details.elapsedMs = Date.now() - startedAt; - // Snapshots must not mutate already emitted tool results/session entries. - try { - onUpdate?.({ ...details, activity: [...details.activity], usage: { ...details.usage, cost: { ...details.usage.cost } } }); - } catch (error) { - spawnError = `Could not publish child progress: ${String(error)}`; - terminate(); - } - }; - const scheduleUpdate = () => { - if (!closed && !updateTimer) updateTimer = setTimeout(() => { updateTimer = undefined; emit(); }, 100); - }; - const activity = (text: string) => { - details.activity.push(text.slice(0, 500)); - details.activity = details.activity.slice(-20); - scheduleUpdate(); + const appendStderr = (text: string) => { + details.stderr = (details.stderr + text).slice(-MAX_STDERR_CHARS); }; + const kill = (sig: NodeJS.Signals) => { if (!proc.pid) return; try { @@ -143,7 +230,7 @@ export async function runDelegate( else proc.kill(sig); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") { - details.stderr = `${details.stderr}\nUnable to send ${sig}: ${String(error)}`.slice(-65536); + appendStderr(`\nUnable to send ${sig}: ${String(error)}`); } } }; @@ -153,71 +240,45 @@ export async function runDelegate( // proc.killed only means a signal was sent, not that the process exited. killTimer = setTimeout(() => kill("SIGKILL"), killDelayMs); }; - const abort = () => { details.status = "aborted"; terminate(); }; + const abort = () => { + details.status = "aborted"; + terminate(); + }; + const failProtocol = (message: string) => { + protocolError ??= message; + terminate(); + }; - const processLine = (line: string) => { - if (!line.trim() || protocolError) return; - if (Buffer.byteLength(line) > 8 * 1024 * 1024) { - protocolError = "Child JSONL record exceeded 8 MiB."; + const publish = () => { + details.elapsedMs = Date.now() - startedAt; + try { + onUpdate?.(snapshot(details)); + } catch (error) { + spawnError = `Could not publish child progress: ${String(error)}`; terminate(); - return; } + }; + const schedulePublish = () => { + if (closed || updateTimer) return; + updateTimer = setTimeout(() => { + updateTimer = undefined; + publish(); + }, UPDATE_THROTTLE_MS); + }; + + const handleLine = (line: string) => { + if (!line.trim() || protocolError) return; + if (Buffer.byteLength(line) > MAX_RECORD_BYTES) return failProtocol(RECORD_TOO_LARGE); let event: JsonAgentSessionEvent; try { - const parsed = JSON.parse(line); - if (!parsed || typeof parsed.type !== "string") throw new Error("Missing event type"); - event = parsed; + event = parseEvent(line); } catch { - protocolError = "Child emitted invalid JSONL output."; - terminate(); - return; + return failProtocol("Child emitted invalid JSONL output."); } try { - switch (event.type) { - case "message_start": - if (event.message.role === "assistant") details.output = ""; - break; - case "message_update": { - const delta = event.assistantMessageEvent; - if (delta.type === "text_delta") { - details.output = (details.output + delta.delta).slice(-16384); - scheduleUpdate(); - } - break; - } - case "message_end": { - const message = event.message; - if (message.role !== "assistant") break; - assistantSeen = true; - details.turns++; - stopReason = message.stopReason; - assistantError = message.errorMessage; - details.output = message.content.filter((part) => part.type === "text").map((part) => part.text).join("\n"); - for (const key of ["input", "output", "cacheRead", "cacheWrite", "totalTokens"] as const) { - details.usage[key] += message.usage?.[key] ?? 0; - } - for (const key of ["input", "output", "cacheRead", "cacheWrite", "total"] as const) { - details.usage.cost[key] += message.usage?.cost?.[key] ?? 0; - } - scheduleUpdate(); - break; - } - case "tool_execution_start": { - const args = event.args as Record; - const target = args?.command ?? args?.path ?? args?.pattern ?? ""; - activity(`→ ${event.toolName} ${String(target)}`); - break; - } - case "tool_execution_end": - activity(`${event.isError ? "✗" : "✓"} ${event.toolName}`); - break; - case "auto_retry_start": - activity(`Retry ${event.attempt}: ${event.errorMessage}`); - break; - } + if (applyEvent(event, details, outcome)) schedulePublish(); } catch (error) { - protocolError = `Invalid child event: ${String(error)}`; - terminate(); + failProtocol(`Invalid child event: ${String(error)}`); } }; @@ -226,52 +287,51 @@ export async function runDelegate( buffer += chunk; let newline: number; while ((newline = buffer.indexOf("\n")) !== -1) { - const line = buffer.slice(0, newline); + handleLine(buffer.slice(0, newline)); buffer = buffer.slice(newline + 1); - processLine(line); } - if (Buffer.byteLength(buffer) > 8 * 1024 * 1024) { - protocolError = "Child JSONL record exceeded 8 MiB."; + if (Buffer.byteLength(buffer) > MAX_RECORD_BYTES) { buffer = ""; - terminate(); + failProtocol(RECORD_TOO_LARGE); } }); proc.stderr.setEncoding("utf8"); - proc.stderr.on("data", (chunk: string) => { details.stderr = (details.stderr + chunk).slice(-65536); }); + proc.stderr.on("data", appendStderr); proc.stdin.on("error", (error: NodeJS.ErrnoException) => { // An early CLI failure may close stdin before accepting the task. - if (error.code !== "EPIPE") { spawnError = `Child stdin failed: ${error.message}`; terminate(); } + if (error.code === "EPIPE") return; + spawnError = `Child stdin failed: ${error.message}`; + terminate(); + }); + proc.on("error", (error) => { + spawnError = `Could not start Pi: ${error.message}`; }); - proc.on("error", (error) => { spawnError = `Could not start Pi: ${error.message}`; }); proc.on("close", (code, exitSignal) => { closed = true; signal?.removeEventListener("abort", abort); - if (killTimer) clearTimeout(killTimer); - if (updateTimer) clearTimeout(updateTimer); + clearTimeout(killTimer); + clearTimeout(updateTimer); clearInterval(heartbeat); + if (buffer.trim()) handleLine(buffer); + details.elapsedMs = Date.now() - startedAt; - if (buffer.trim()) processLine(buffer); details.exitCode = code; details.exitSignal = exitSignal; if (details.status === "aborted") { details.error = "Delegation cancelled."; } else { - const failure = spawnError ?? protocolError - ?? (code !== 0 ? `Child exited ${exitSignal ? `with signal ${exitSignal}` : `with code ${code}`}.` : undefined) - ?? (assistantSeen && stopReason !== "stop" ? assistantError || `Child ended with ${stopReason ?? "no stop reason"}.` : undefined) - ?? (!assistantSeen || !details.output.trim() ? "Child exited without a final text response." : undefined); - details.status = failure ? "failed" : "completed"; - details.error = failure; + details.error = spawnError ?? protocolError + ?? failureReason(code, exitSignal, outcome, details.output); + details.status = details.error ? "failed" : "completed"; } resolve(details); }); - const heartbeat = setInterval(emit, 1000); + const heartbeat = setInterval(publish, HEARTBEAT_MS); signal?.addEventListener("abort", abort, { once: true }); if (signal?.aborted) abort(); - emit(); - // stdin avoids shell quoting, argument-size limits, @file interpretation, - // and exposing the delegated task in the process argument list. + publish(); + // stdin avoids argv size limits, @file interpretation, and exposing the task in `ps`. proc.stdin.end(`Task: ${options.task}`); }); }