From aba42ae6b90fdb5e3f3e7aac0c4bd0577a761952 Mon Sep 17 00:00:00 2001 From: Michael Renner Date: Wed, 4 Nov 2026 11:24:12 +0100 Subject: [PATCH] fix(coding-agent): make subagent progress and failures reliable (#8250) The shipped subagent example could say a child was done while it was still working, lose the useful part of process failures, and return a normal-looking tool result when the delegated work had actually failed. Single and chain output could also exceed the normal tool limit. This also adds a dedicated regression suite covering progress state, cancellation, process failures, output limits, cwd-based agent selection, and persisted tool-result error state. After this change: - queued, running, completed, failed, and aborted work stay distinct throughout the run; - interrupting the parent run preserves partial child results, stops queued work from starting, and escalates when a child ignores SIGTERM; - signal exits, spawn failures, provider errors, stderr, and assistant output remain visible instead of collapsing to `(no output)`; - failed subagents become failed tool results, so the parent model can react instead of treating them as successful calls; - single and chain results stay within the normal 50 KB / 2,000-line limit; - project agents are selected and confirmed from each task's actual working directory rather than the dispatcher's directory. ## Further potential improvements These are useful follow-ups, but they are separate product and storage decisions rather than part of the failure-handling fixes in this PR.
Potential follow-ups ### Keep persisted transcripts bounded Subagent tool results persist the full child message history in `details`, which can make parent session files large and slow to reload. A follow-up could compact that history or spill it to disk. ### Define an aggregate parallel-output budget Parallel mode still gives every task its own 50 KB budget. A follow-up could keep every task represented while enforcing one overall tool-result limit. ### Report child usage to the calling session Per-child token and cost usage is visible in result cards but is not added to the calling session's totals. ### Isolate malformed agent definitions One malformed YAML agent can still break discovery for every valid agent in the same directory. ### Limit subagent recursion A nesting limit would prevent accidental process and paid-request multiplication when subagents can invoke the subagent tool again. * https://github.com/earendil-works/pi/issues/8195 ### Add a configurable timeout Unattended runs would benefit from a generous opt-in timeout for children that never complete.
## Related issues ### Spawn failures lose their diagnostics This directly addresses the reported case where a nonexistent child working directory produces `(no output)` instead of the actual spawn error. The regression suite includes that same reproduction. * https://github.com/earendil-works/pi/issues/8129 ### Persisted child history can bloat sessions This PR intentionally keeps full child history in tool-result details, so session compaction or spill-to-disk remains follow-up work rather than something fixed here. * https://github.com/earendil-works/pi/issues/7452 --- .../examples/extensions/subagent/README.md | 25 +- .../examples/extensions/subagent/index.ts | 559 +++++++++++----- .../test/fixtures/subagent-child.mjs | 80 +++ .../test/subagent-extension.test.ts | 621 ++++++++++++++++++ 4 files changed, 1103 insertions(+), 182 deletions(-) create mode 100644 packages/coding-agent/test/fixtures/subagent-child.mjs create mode 100644 packages/coding-agent/test/subagent-extension.test.ts diff --git a/packages/coding-agent/examples/extensions/subagent/README.md b/packages/coding-agent/examples/extensions/subagent/README.md index da74f3266..95d7ebe2a 100644 --- a/packages/coding-agent/examples/extensions/subagent/README.md +++ b/packages/coding-agent/examples/extensions/subagent/README.md @@ -8,7 +8,7 @@ Delegate tasks to specialized subagents with isolated context windows. - **Streaming output**: See tool calls and progress as they happen - **Parallel streaming**: All parallel tasks stream updates simultaneously - **Markdown rendering**: Final output rendered with proper formatting (expanded view) -- **Usage tracking**: Shows turns, tokens, cost, and context usage per agent +- **Usage display**: Shows turns, tokens, cost, and context usage inside each subagent result - **Abort support**: Ctrl+C propagates to kill subagent processes ## Structure @@ -62,7 +62,7 @@ This tool executes a separate `pi` subprocess with a delegated system prompt and To enable project-local agents, pass `agentScope: "both"` (or `"project"`). Only do this for repositories you trust. -When running interactively, the tool prompts for confirmation before running project-local agents in untrusted projects. Trusted projects skip the additional prompt. Set `confirmProjectAgents: false` to disable confirmation. +When running interactively, the tool prompts for confirmation before running project-local agents unless the session trust context reports trusted and the agents come from that session project's `.pi/agents` directory. Agents loaded from another task `cwd` still require confirmation. Set `confirmProjectAgents: false` to disable confirmation. ## Usage @@ -112,8 +112,8 @@ Use a chain: first have scout find the read tool, then have planner suggest impr **Parallel mode streaming**: - Shows all tasks with live status (⏳ running, ✓ done, ✗ failed) - Updates as each task makes progress -- Shows "2/3 done, 1 running" status -- Returns each completed task's final output to the parent model, capped at 50 KB per task +- Distinguishes pending, running, completed, failed, and aborted tasks +- Returns each completed task summary to the parent model with its own 50 KB limit - Returns failure diagnostics from stderr/error messages when a child exits before producing output **Tool call formatting** (mimics built-in tools): @@ -140,7 +140,7 @@ System prompt for the agent goes here. When `model` is omitted, the subagent inherits the dispatching session's active model and thinking level. **Locations:** -- `~/.pi/agent/agents/*.md` - User-level (always loaded) +- `~/.pi/agent/agents/*.md` - User-level (with `agentScope: "user"` or `"both"`) - `.pi/agents/*.md` - Project-level (only with `agentScope: "project"` or `"both"`) Project agents override user agents with the same name when `agentScope: "both"`. @@ -164,14 +164,17 @@ Project agents override user agents with the same name when `agentScope: "both"` ## Error Handling -- **Exit code != 0**: Tool returns error with stderr/output -- **stopReason "error"**: LLM error propagated with error message -- **stopReason "aborted"**: User abort (Ctrl+C) kills subprocess, throws error -- **Chain mode**: Stops at first failing step, reports which step failed +- **Exit code != 0 or signal termination**: Tool returns an error with preserved stderr/output +- **Spawn errors**: Tool returns the spawn diagnostic +- **stopReason "error"**: LLM error is propagated with its error message +- **Abort**: Ctrl+C sends SIGTERM, escalates to SIGKILL after five seconds if needed, and preserves partial details/usage +- **Chain mode**: Stops at the first failing step and reports which step failed ## Limitations - Output truncated to last 10 items in collapsed view (expand to see all) -- Parallel model-visible output is capped at 50 KB per task; full results remain in tool details -- Agents discovered fresh on each invocation (allows editing mid-session) +- Single and chain output is capped at 50 KB/2,000 lines; full results remain in tool details +- Parallel output keeps a separate 50 KB budget per task, allowing up to roughly 400 KB across eight tasks +- Child usage is displayed in tool details but is not added to the calling session's usage totals +- Agents are snapshotted from each task's effective cwd when the tool invocation starts; relative task cwd values resolve from the session cwd - Parallel mode limited to 8 tasks, 4 concurrent diff --git a/packages/coding-agent/examples/extensions/subagent/index.ts b/packages/coding-agent/examples/extensions/subagent/index.ts index 843297454..0b911a341 100644 --- a/packages/coding-agent/examples/extensions/subagent/index.ts +++ b/packages/coding-agent/examples/extensions/subagent/index.ts @@ -21,10 +21,14 @@ import type { Message } from "@earendil-works/pi-ai"; import { StringEnum } from "@earendil-works/pi-ai"; import { CONFIG_DIR_NAME, + DEFAULT_MAX_BYTES, + DEFAULT_MAX_LINES, type ExtensionAPI, formatReadLineRange, getAgentDir, getMarkdownTheme, + type JsonAgentSessionEvent, + truncateHead, withFileMutationQueue, } from "@earendil-works/pi-coding-agent"; import { Container, Markdown, Spacer, Text } from "@earendil-works/pi-tui"; @@ -35,6 +39,7 @@ const MAX_PARALLEL_TASKS = 8; const MAX_CONCURRENCY = 4; const COLLAPSED_ITEM_COUNT = 10; const PER_TASK_OUTPUT_CAP = 50 * 1024; +const TRUNCATION_NOTICE_RESERVE = 256; function formatTokens(count: number): string { if (count < 1000) return count.toString(); @@ -144,25 +149,28 @@ interface UsageStats { turns: number; } +type ChildStatus = "pending" | "running" | "completed" | "failed" | "aborted"; + interface SingleResult { agent: string; agentSource: "user" | "project" | "unknown"; task: string; - exitCode: number; + status: ChildStatus; + exitCode: number | null; messages: Message[]; stderr: string; usage: UsageStats; model?: string; stopReason?: string; errorMessage?: string; + terminationSignal?: NodeJS.Signals; step?: number; } interface SubagentDetails { mode: "single" | "parallel" | "chain"; - agentScope: AgentScope; - projectAgentsDir: string | null; results: SingleResult[]; + isError: boolean; } function getFinalOutput(messages: Message[]): string { @@ -178,14 +186,38 @@ function getFinalOutput(messages: Message[]): string { } function isFailedResult(result: SingleResult): boolean { - return result.exitCode !== 0 || result.stopReason === "error" || result.stopReason === "aborted"; + return result.status === "failed" || result.status === "aborted"; +} + +function getFailureEvidence(result: SingleResult): string { + const evidence: string[] = []; + const seen = new Set(); + const add = (label: string, value: string | undefined) => { + const text = value?.trim(); + if (!text || seen.has(text)) return; + seen.add(text); + evidence.push(`${label}: ${text}`); + }; + + add("Error", result.errorMessage); + add("Signal", result.terminationSignal); + add("stderr", result.stderr); + add("Assistant output", getFinalOutput(result.messages)); + return evidence.join("\n\n") || "(no output)"; +} + +function getFailureSummary(result: SingleResult): string { + return ( + result.errorMessage?.trim() || + (result.terminationSignal ? `Terminated by signal ${result.terminationSignal}` : "") || + result.stderr.trim() || + getFinalOutput(result.messages) || + "(no output)" + ); } function getResultOutput(result: SingleResult): string { - if (isFailedResult(result)) { - return result.errorMessage || result.stderr || getFinalOutput(result.messages) || "(no output)"; - } - return getFinalOutput(result.messages) || "(no output)"; + return isFailedResult(result) ? getFailureEvidence(result) : getFinalOutput(result.messages) || "(no output)"; } function truncateParallelOutput(output: string): string { @@ -199,6 +231,19 @@ function truncateParallelOutput(output: string): string { return `${truncated}\n\n[Output truncated: ${byteLength - Buffer.byteLength(truncated, "utf8")} bytes omitted. Full output preserved in tool details.]`; } +function truncateModelOutput(output: string): string { + const initial = truncateHead(output); + if (!initial.truncated) return output; + + const body = truncateHead(output, { + maxLines: DEFAULT_MAX_LINES - 2, + maxBytes: DEFAULT_MAX_BYTES - TRUNCATION_NOTICE_RESERVE, + }); + const omittedBytes = initial.totalBytes - body.outputBytes; + const notice = `[Output truncated: ${omittedBytes} bytes omitted. Full output preserved in tool details.]`; + return body.content ? `${body.content}\n\n${notice}` : notice; +} + type DisplayItem = { type: "text"; text: string } | { type: "toolCall"; name: string; args: Record }; function getDisplayItems(messages: Message[]): DisplayItem[] { @@ -267,32 +312,74 @@ interface DispatchDefaults { thinkingLevel?: ThinkingLevel; } +interface ProcessOutcome { + code: number | null; + signal: NodeJS.Signals | null; + spawnError?: Error; +} + +interface ResolvedAgent { + agent?: AgentConfig; + available: AgentConfig[]; + cwd: string; +} + +function resolveAgent(defaultCwd: string, scope: AgentScope, name: string, cwd?: string): ResolvedAgent { + const effectiveCwd = path.resolve(defaultCwd, cwd ?? "."); + const discovery = discoverAgents(effectiveCwd, scope); + return { + agent: discovery.agents.find((candidate) => candidate.name === name), + available: discovery.agents, + cwd: effectiveCwd, + }; +} + +function createUnstartedResult( + agent: string, + task: string, + step: number | undefined, + status: "aborted" | "failed", + errorMessage?: string, +): SingleResult { + return { + agent, + agentSource: "unknown", + task, + status, + exitCode: null, + messages: [], + stderr: "", + usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, + errorMessage, + step, + }; +} + async function runSingleAgent( - defaultCwd: string, dispatchDefaults: DispatchDefaults, - agents: AgentConfig[], + resolvedAgent: ResolvedAgent, agentName: string, task: string, - cwd: string | undefined, step: number | undefined, signal: AbortSignal | undefined, onUpdate: OnUpdateCallback | undefined, makeDetails: (results: SingleResult[]) => SubagentDetails, ): Promise { - const agent = agents.find((a) => a.name === agentName); + if (signal?.aborted) { + return createUnstartedResult(agentName, task, step, "aborted", "Subagent was aborted before starting"); + } + const effectiveCwd = resolvedAgent.cwd; + const agent = resolvedAgent.agent; if (!agent) { - const available = agents.map((a) => `"${a.name}"`).join(", ") || "none"; - return { - agent: agentName, - agentSource: "unknown", + const available = resolvedAgent.available.map((candidate) => `"${candidate.name}"`).join(", ") || "none"; + return createUnstartedResult( + agentName, task, - exitCode: 1, - messages: [], - stderr: `Unknown agent: "${agentName}". Available agents: ${available}.`, - usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, step, - }; + "failed", + `Unknown agent: "${agentName}". Available agents: ${available}.`, + ); } const args: string[] = ["--mode", "json", "-p", "--no-session"]; @@ -306,12 +393,12 @@ async function runSingleAgent( let tmpPromptDir: string | null = null; let tmpPromptPath: string | null = null; - const currentResult: SingleResult = { agent: agentName, agentSource: agent.source, task, - exitCode: 0, + status: "running", + exitCode: null, messages: [], stderr: "", usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, @@ -320,12 +407,10 @@ async function runSingleAgent( }; const emitUpdate = () => { - if (onUpdate) { - onUpdate({ - content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }], - details: makeDetails([currentResult]), - }); - } + onUpdate?.({ + content: [{ type: "text", text: getFinalOutput(currentResult.messages) || "(running...)" }], + details: makeDetails([currentResult]), + }); }; try { @@ -336,104 +421,154 @@ async function runSingleAgent( args.push("--append-system-prompt", tmpPromptPath); } - args.push(`Task: ${task}`); - let wasAborted = false; + if (signal?.aborted) { + currentResult.status = "aborted"; + currentResult.errorMessage = "Subagent was aborted before starting"; + return currentResult; + } - const exitCode = await new Promise((resolve) => { + args.push(`Task: ${task}`); + emitUpdate(); + let wasAborted = false; + const outcome = await new Promise((resolve) => { const invocation = getPiInvocation(args); - const proc = spawn(invocation.command, invocation.args, { - cwd: cwd ?? defaultCwd, - shell: false, - stdio: ["ignore", "pipe", "pipe"], - }); + let settled = false; + let exited = false; + let closed = false; + let killTimer: ReturnType | undefined; let buffer = ""; const processLine = (line: string) => { if (!line.trim()) return; - let event: any; + let event: JsonAgentSessionEvent; try { - event = JSON.parse(line); + event = JSON.parse(line) as JsonAgentSessionEvent; } catch { return; } - - if (event.type === "message_end" && event.message) { - const msg = event.message as Message; - currentResult.messages.push(msg); - - if (msg.role === "assistant") { - currentResult.usage.turns++; - const usage = msg.usage; - if (usage) { - currentResult.usage.input += usage.input || 0; - currentResult.usage.output += usage.output || 0; - currentResult.usage.cacheRead += usage.cacheRead || 0; - currentResult.usage.cacheWrite += usage.cacheWrite || 0; - currentResult.usage.cost += usage.cost?.total || 0; - currentResult.usage.contextTokens = usage.totalTokens || 0; - } - if (!currentResult.model && msg.model) currentResult.model = msg.model; - if (msg.stopReason) currentResult.stopReason = msg.stopReason; - if (msg.errorMessage) currentResult.errorMessage = msg.errorMessage; - } - emitUpdate(); + if (event.type !== "message_end") return; + const eventMessage = event.message; + if ( + eventMessage.role !== "user" && + eventMessage.role !== "assistant" && + eventMessage.role !== "toolResult" + ) { + return; } - - if (event.type === "tool_result_end" && event.message) { - currentResult.messages.push(event.message as Message); - emitUpdate(); + const message = eventMessage as Message; + currentResult.messages.push(message); + if (message.role === "assistant") { + currentResult.usage.turns++; + currentResult.usage.input += message.usage.input; + currentResult.usage.output += message.usage.output; + currentResult.usage.cacheRead += message.usage.cacheRead; + currentResult.usage.cacheWrite += message.usage.cacheWrite; + currentResult.usage.cost += message.usage.cost.total; + currentResult.usage.contextTokens = message.usage.totalTokens; + if (!currentResult.model) currentResult.model = message.model; + currentResult.stopReason = message.stopReason; + currentResult.errorMessage = message.errorMessage; } + emitUpdate(); }; - proc.stdout.on("data", (data) => { + let proc: ReturnType; + const finish = (result: ProcessOutcome) => { + if (settled) return; + settled = true; + closed = true; + if (killTimer) clearTimeout(killTimer); + if (signal) signal.removeEventListener("abort", abortProcess); + resolve(result); + }; + const abortProcess = () => { + if (closed || exited || proc.exitCode !== null || proc.signalCode !== null) return; + if (!proc.kill("SIGTERM")) return; + wasAborted = true; + killTimer = setTimeout(() => { + if (!closed && !exited) proc.kill("SIGKILL"); + }, 5000); + }; + + try { + proc = spawn(invocation.command, invocation.args, { + cwd: effectiveCwd, + shell: false, + stdio: ["ignore", "pipe", "pipe"], + }); + } catch (error) { + finish({ + code: null, + signal: null, + spawnError: error instanceof Error ? error : new Error(String(error)), + }); + return; + } + + proc.stdout?.on("data", (data) => { buffer += data.toString(); const lines = buffer.split("\n"); buffer = lines.pop() || ""; for (const line of lines) processLine(line); }); - - proc.stderr.on("data", (data) => { + proc.stderr?.on("data", (data) => { currentResult.stderr += data.toString(); }); - - proc.on("close", (code) => { - if (buffer.trim()) processLine(buffer); - resolve(code ?? 0); + proc.on("exit", () => { + exited = true; }); - - proc.on("error", () => { - resolve(1); + proc.on("close", (code, terminationSignal) => { + if (buffer.trim()) processLine(buffer); + finish({ code, signal: terminationSignal }); + }); + proc.on("error", (error) => { + finish({ code: null, signal: null, spawnError: error }); }); if (signal) { - const killProc = () => { - wasAborted = true; - proc.kill("SIGTERM"); - setTimeout(() => { - if (!proc.killed) proc.kill("SIGKILL"); - }, 5000); - }; - if (signal.aborted) killProc(); - else signal.addEventListener("abort", killProc, { once: true }); + if (signal.aborted) abortProcess(); + else signal.addEventListener("abort", abortProcess, { once: true }); } }); - currentResult.exitCode = exitCode; - if (wasAborted) throw new Error("Subagent was aborted"); + currentResult.exitCode = outcome.code; + currentResult.terminationSignal = outcome.signal ?? undefined; + + if (wasAborted) { + currentResult.status = "aborted"; + currentResult.errorMessage ||= "Subagent was aborted"; + } else if (outcome.spawnError) { + currentResult.status = "failed"; + currentResult.errorMessage = `Failed to spawn subagent: ${outcome.spawnError.message}`; + } else if (outcome.signal) { + currentResult.status = "failed"; + currentResult.errorMessage ||= `Subagent terminated by signal ${outcome.signal}`; + } else if (outcome.code !== 0) { + currentResult.status = "failed"; + currentResult.errorMessage ||= `Subagent exited with code ${outcome.code}`; + } else if (currentResult.stopReason === "aborted") { + currentResult.status = "aborted"; + } else if (currentResult.stopReason === "error") { + currentResult.status = "failed"; + } else { + currentResult.status = "completed"; + } return currentResult; } finally { - if (tmpPromptPath) + if (tmpPromptPath) { try { fs.unlinkSync(tmpPromptPath); } catch { /* ignore */ } - if (tmpPromptDir) + } + if (tmpPromptDir) { try { fs.rmdirSync(tmpPromptDir); } catch { /* ignore */ } + } } } @@ -467,6 +602,13 @@ const SubagentParams = Type.Object({ }); export default function (pi: ExtensionAPI) { + // Tool execute results cannot set runtime error state directly, so patch it through the result hook. + pi.on("tool_result", (event) => { + if (event.toolName !== "subagent") return; + const details = event.details as SubagentDetails | undefined; + if (details?.isError) return { isError: true }; + }); + pi.registerTool({ name: "subagent", label: "Subagent", @@ -494,12 +636,11 @@ export default function (pi: ExtensionAPI) { const modeCount = Number(hasChain) + Number(hasTasks) + Number(hasSingle); const makeDetails = - (mode: "single" | "parallel" | "chain") => + (mode: "single" | "parallel" | "chain", forceError = false) => (results: SingleResult[]): SubagentDetails => ({ mode, - agentScope, - projectAgentsDir: discovery.projectAgentsDir, results, + isError: forceError || results.some(isFailedResult), }); if (modeCount !== 1) { @@ -511,37 +652,50 @@ export default function (pi: ExtensionAPI) { text: `Invalid parameters. Provide exactly one mode.\nAvailable agents: ${available}`, }, ], - details: makeDetails("single")([]), + details: makeDetails("single", true)([]), }; } - if ( - (agentScope === "project" || agentScope === "both") && - confirmProjectAgents && - ctx.hasUI && - !ctx.isProjectTrusted() - ) { - const requestedAgentNames = new Set(); - if (params.chain) for (const step of params.chain) requestedAgentNames.add(step.agent); - if (params.tasks) for (const t of params.tasks) requestedAgentNames.add(t.agent); - if (params.agent) requestedAgentNames.add(params.agent); + const resolvedChain = (params.chain ?? []).map((step) => + resolveAgent(ctx.cwd, agentScope, step.agent, step.cwd), + ); + const resolvedTasks = (params.tasks ?? []).map((task) => + resolveAgent(ctx.cwd, agentScope, task.agent, task.cwd), + ); + const resolvedSingle = params.agent ? resolveAgent(ctx.cwd, agentScope, params.agent, params.cwd) : undefined; - const projectAgentsRequested = Array.from(requestedAgentNames) - .map((name) => agents.find((a) => a.name === name)) - .filter((a): a is AgentConfig => a?.source === "project"); + if ((agentScope === "project" || agentScope === "both") && confirmProjectAgents && ctx.hasUI) { + let selectedAgents: ResolvedAgent[] = []; + if (hasChain) selectedAgents = resolvedChain; + else if (hasTasks) selectedAgents = resolvedTasks; + else if (resolvedSingle) selectedAgents = [resolvedSingle]; + const projectAgentsByPath = new Map(); + for (const resolved of selectedAgents) { + if (resolved.agent?.source === "project") { + projectAgentsByPath.set(resolved.agent.filePath, resolved.agent); + } + } + const trustedProjectAgentsDir = + ctx.isProjectTrusted() && discovery.projectAgentsDir + ? fs.realpathSync(discovery.projectAgentsDir) + : undefined; + const projectAgentsRequested = Array.from(projectAgentsByPath.values()).filter( + (agent) => + !trustedProjectAgentsDir || fs.realpathSync(path.dirname(agent.filePath)) !== trustedProjectAgentsDir, + ); if (projectAgentsRequested.length > 0) { - const names = projectAgentsRequested.map((a) => a.name).join(", "); - const dir = discovery.projectAgentsDir ?? "(unknown)"; + const sources = projectAgentsRequested.map((agent) => `- ${agent.name}: ${agent.filePath}`).join("\n"); const ok = await ctx.ui.confirm( "Run project-local agents?", - `Agents: ${names}\nSource: ${dir}\n\nProject agents are repo-controlled. Only continue for trusted repositories.`, + `Sources:\n${sources}\n\nProject agents are repo-controlled. Only continue for trusted repositories.`, ); - if (!ok) + if (!ok) { return { content: [{ type: "text", text: "Canceled: project-local agents not approved." }], details: makeDetails(hasChain ? "chain" : hasTasks ? "parallel" : "single")([]), }; + } } } @@ -569,12 +723,10 @@ export default function (pi: ExtensionAPI) { : undefined; const result = await runSingleAgent( - ctx.cwd, dispatchDefaults, - agents, + resolvedChain[i], step.agent, taskWithContext, - step.cwd, i + 1, signal, chainUpdate, @@ -586,15 +738,24 @@ export default function (pi: ExtensionAPI) { if (isError) { const errorMsg = getResultOutput(result); return { - content: [{ type: "text", text: `Chain stopped at step ${i + 1} (${step.agent}): ${errorMsg}` }], + content: [ + { + type: "text", + text: truncateModelOutput(`Chain stopped at step ${i + 1} (${step.agent}): ${errorMsg}`), + }, + ], details: makeDetails("chain")(results), - isError: true, }; } previousOutput = getFinalOutput(result.messages); } return { - content: [{ type: "text", text: getFinalOutput(results[results.length - 1].messages) || "(no output)" }], + content: [ + { + type: "text", + text: truncateModelOutput(getFinalOutput(results[results.length - 1].messages) || "(no output)"), + }, + ], details: makeDetails("chain")(results), }; } @@ -608,7 +769,7 @@ export default function (pi: ExtensionAPI) { text: `Too many parallel tasks (${params.tasks.length}). Max is ${MAX_PARALLEL_TASKS}.`, }, ], - details: makeDetails("parallel")([]), + details: makeDetails("parallel", true)([]), }; // Track all results for streaming updates @@ -620,7 +781,8 @@ export default function (pi: ExtensionAPI) { agent: params.tasks[i].agent, agentSource: "unknown", task: params.tasks[i].task, - exitCode: -1, // -1 = still running + status: "pending", + exitCode: null, messages: [], stderr: "", usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, contextTokens: 0, turns: 0 }, @@ -629,11 +791,15 @@ export default function (pi: ExtensionAPI) { const emitParallelUpdate = () => { if (onUpdate) { - const running = allResults.filter((r) => r.exitCode === -1).length; - const done = allResults.filter((r) => r.exitCode !== -1).length; + const running = allResults.filter((result) => result.status === "running").length; + const pending = allResults.filter((result) => result.status === "pending").length; + const done = allResults.length - running - pending; onUpdate({ content: [ - { type: "text", text: `Parallel: ${done}/${allResults.length} done, ${running} running...` }, + { + type: "text", + text: `Parallel: ${done}/${allResults.length} done, ${running} running, ${pending} pending...`, + }, ], details: makeDetails("parallel")([...allResults]), }); @@ -642,12 +808,10 @@ export default function (pi: ExtensionAPI) { const results = await mapWithConcurrencyLimit(params.tasks, MAX_CONCURRENCY, async (t, index) => { const result = await runSingleAgent( - ctx.cwd, dispatchDefaults, - agents, + resolvedTasks[index], t.agent, t.task, - t.cwd, undefined, signal, // Per-task update callback @@ -667,9 +831,10 @@ export default function (pi: ExtensionAPI) { const successCount = results.filter((r) => !isFailedResult(r)).length; const summaries = results.map((r) => { const output = truncateParallelOutput(getResultOutput(r)); - const status = isFailedResult(r) - ? `failed${r.stopReason && r.stopReason !== "end" ? ` (${r.stopReason})` : ""}` - : "completed"; + const status = + isFailedResult(r) && r.stopReason && r.stopReason !== "end" + ? `${r.status} (${r.stopReason})` + : r.status; return `### [${r.agent}] ${status}\n\n${output}`; }); return { @@ -683,14 +848,12 @@ export default function (pi: ExtensionAPI) { }; } - if (params.agent && params.task) { + if (params.agent && params.task && resolvedSingle) { const result = await runSingleAgent( - ctx.cwd, dispatchDefaults, - agents, + resolvedSingle, params.agent, params.task, - params.cwd, undefined, signal, onUpdate, @@ -700,13 +863,22 @@ export default function (pi: ExtensionAPI) { if (isError) { const errorMsg = getResultOutput(result); return { - content: [{ type: "text", text: `Agent ${result.stopReason || "failed"}: ${errorMsg}` }], + content: [ + { + type: "text", + text: truncateModelOutput(`Agent ${result.stopReason || result.status}: ${errorMsg}`), + }, + ], details: makeDetails("single")([result]), - isError: true, }; } return { - content: [{ type: "text", text: getFinalOutput(result.messages) || "(no output)" }], + content: [ + { + type: "text", + text: truncateModelOutput(getFinalOutput(result.messages) || "(no output)"), + }, + ], details: makeDetails("single")([result]), }; } @@ -714,7 +886,7 @@ export default function (pi: ExtensionAPI) { const available = agents.map((a) => `${a.name} (${a.source})`).join(", ") || "none"; return { content: [{ type: "text", text: `Invalid parameters. Available agents: ${available}` }], - details: makeDetails("single")([]), + details: makeDetails("single", true)([]), }; }, @@ -787,20 +959,33 @@ export default function (pi: ExtensionAPI) { return text.trimEnd(); }; + const resultIcon = (child: SingleResult): string => { + if (child.status === "pending" || child.status === "running") { + return theme.fg("warning", "⏳"); + } + if (isFailedResult(child)) return theme.fg("error", "✗"); + return theme.fg("success", "✓"); + }; + if (details.mode === "single" && details.results.length === 1) { const r = details.results[0]; const isError = isFailedResult(r); - const icon = isError ? theme.fg("error", "✗") : theme.fg("success", "✓"); + const isRunning = r.status === "pending" || r.status === "running"; + const icon = resultIcon(r); const displayItems = getDisplayItems(r.messages); const finalOutput = getFinalOutput(r.messages); if (expanded) { const container = new Container(); let header = `${icon} ${theme.fg("toolTitle", theme.bold(r.agent))}${theme.fg("muted", ` (${r.agentSource})`)}`; - if (isError && r.stopReason) header += ` ${theme.fg("error", `[${r.stopReason}]`)}`; + if (isRunning) header += ` ${theme.fg("warning", `[${r.status}]`)}`; + else if (isError) header += ` ${theme.fg("error", `[${r.stopReason ?? r.status}]`)}`; container.addChild(new Text(header, 0, 0)); if (isError && r.errorMessage) container.addChild(new Text(theme.fg("error", `Error: ${r.errorMessage}`), 0, 0)); + if (r.terminationSignal) + container.addChild(new Text(theme.fg("error", `Signal: ${r.terminationSignal}`), 0, 0)); + if (r.stderr.trim()) container.addChild(new Text(theme.fg("error", `stderr: ${r.stderr.trim()}`), 0, 0)); container.addChild(new Spacer(1)); container.addChild(new Text(theme.fg("muted", "─── Task ───"), 0, 0)); container.addChild(new Text(theme.fg("dim", r.task), 0, 0)); @@ -833,7 +1018,8 @@ export default function (pi: ExtensionAPI) { } let text = `${icon} ${theme.fg("toolTitle", theme.bold(r.agent))}${theme.fg("muted", ` (${r.agentSource})`)}`; - if (isError && r.stopReason) text += ` ${theme.fg("error", `[${r.stopReason}]`)}`; + if (isRunning) text += ` ${theme.fg("warning", `[${r.status}]`)}`; + else if (isError) text += ` ${theme.fg("error", `[${r.stopReason ?? r.status}]`)}`; if (isError && r.errorMessage) text += `\n${theme.fg("error", `Error: ${r.errorMessage}`)}`; else if (displayItems.length === 0) text += `\n${theme.fg("muted", "(no output)")}`; else { @@ -845,22 +1031,35 @@ export default function (pi: ExtensionAPI) { return new Text(text, 0, 0); } - const aggregateUsage = (results: SingleResult[]) => { - const total = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, cost: 0, turns: 0 }; - for (const r of results) { - total.input += r.usage.input; - total.output += r.usage.output; - total.cacheRead += r.usage.cacheRead; - total.cacheWrite += r.usage.cacheWrite; - total.cost += r.usage.cost; - total.turns += r.usage.turns; + const aggregateUsage = (results: SingleResult[]): UsageStats => { + const total: UsageStats = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + cost: 0, + contextTokens: 0, + turns: 0, + }; + for (const result of results) { + total.input += result.usage.input; + total.output += result.usage.output; + total.cacheRead += result.usage.cacheRead; + total.cacheWrite += result.usage.cacheWrite; + total.cost += result.usage.cost; + total.turns += result.usage.turns; } return total; }; if (details.mode === "chain") { - const successCount = details.results.filter((r) => r.exitCode === 0).length; - const icon = successCount === details.results.length ? theme.fg("success", "✓") : theme.fg("error", "✗"); + const successCount = details.results.filter((result) => result.status === "completed").length; + const hasActive = details.results.some( + (result) => result.status === "pending" || result.status === "running", + ); + let icon = theme.fg("error", "✗"); + if (hasActive) icon = theme.fg("warning", "⏳"); + else if (successCount === details.results.length) icon = theme.fg("success", "✓"); if (expanded) { const container = new Container(); @@ -876,7 +1075,7 @@ export default function (pi: ExtensionAPI) { ); for (const r of details.results) { - const rIcon = r.exitCode === 0 ? theme.fg("success", "✓") : theme.fg("error", "✗"); + const rIcon = resultIcon(r); const displayItems = getDisplayItems(r.messages); const finalOutput = getFinalOutput(r.messages); @@ -889,6 +1088,12 @@ export default function (pi: ExtensionAPI) { ), ); container.addChild(new Text(theme.fg("muted", "Task: ") + theme.fg("dim", r.task), 0, 0)); + if (isFailedResult(r) && r.errorMessage) + container.addChild(new Text(theme.fg("error", `Error: ${r.errorMessage}`), 0, 0)); + if (r.terminationSignal) + container.addChild(new Text(theme.fg("error", `Signal: ${r.terminationSignal}`), 0, 0)); + if (r.stderr.trim()) + container.addChild(new Text(theme.fg("error", `stderr: ${r.stderr.trim()}`), 0, 0)); // Show tool calls for (const item of displayItems) { @@ -928,11 +1133,16 @@ export default function (pi: ExtensionAPI) { theme.fg("toolTitle", theme.bold("chain ")) + theme.fg("accent", `${successCount}/${details.results.length} steps`); for (const r of details.results) { - const rIcon = r.exitCode === 0 ? theme.fg("success", "✓") : theme.fg("error", "✗"); + const rIcon = resultIcon(r); const displayItems = getDisplayItems(r.messages); text += `\n\n${theme.fg("muted", `─── Step ${r.step}: `)}${theme.fg("accent", r.agent)} ${rIcon}`; - if (displayItems.length === 0) text += `\n${theme.fg("muted", "(no output)")}`; - else text += `\n${renderDisplayItems(displayItems, 5)}`; + if (displayItems.length === 0) { + const failed = isFailedResult(r); + const summary = failed ? getFailureSummary(r) : "(no output)"; + text += `\n${theme.fg(failed ? "error" : "muted", summary)}`; + } else { + text += `\n${renderDisplayItems(displayItems, 5)}`; + } } const usageStr = formatUsageStats(aggregateUsage(details.results)); if (usageStr) text += `\n\n${theme.fg("dim", `Total: ${usageStr}`)}`; @@ -941,17 +1151,16 @@ export default function (pi: ExtensionAPI) { } if (details.mode === "parallel") { - const running = details.results.filter((r) => r.exitCode === -1).length; - const successCount = details.results.filter((r) => r.exitCode !== -1 && !isFailedResult(r)).length; - const failCount = details.results.filter((r) => r.exitCode !== -1 && isFailedResult(r)).length; - const isRunning = running > 0; - const icon = isRunning - ? theme.fg("warning", "⏳") - : failCount > 0 - ? theme.fg("warning", "◐") - : theme.fg("success", "✓"); + const running = details.results.filter((result) => result.status === "running").length; + const pending = details.results.filter((result) => result.status === "pending").length; + const successCount = details.results.filter((result) => result.status === "completed").length; + const failCount = details.results.filter(isFailedResult).length; + const isRunning = running > 0 || pending > 0; + let icon = theme.fg("success", "✓"); + if (isRunning) icon = theme.fg("warning", "⏳"); + else if (failCount > 0) icon = theme.fg("warning", "◐"); const status = isRunning - ? `${successCount + failCount}/${details.results.length} done, ${running} running` + ? `${successCount + failCount}/${details.results.length} done, ${running} running, ${pending} pending` : `${successCount}/${details.results.length} tasks`; if (expanded && !isRunning) { @@ -965,7 +1174,7 @@ export default function (pi: ExtensionAPI) { ); for (const r of details.results) { - const rIcon = isFailedResult(r) ? theme.fg("error", "✗") : theme.fg("success", "✓"); + const rIcon = resultIcon(r); const displayItems = getDisplayItems(r.messages); const finalOutput = getFinalOutput(r.messages); @@ -974,6 +1183,12 @@ export default function (pi: ExtensionAPI) { new Text(`${theme.fg("muted", "─── ") + theme.fg("accent", r.agent)} ${rIcon}`, 0, 0), ); container.addChild(new Text(theme.fg("muted", "Task: ") + theme.fg("dim", r.task), 0, 0)); + if (isFailedResult(r) && r.errorMessage) + container.addChild(new Text(theme.fg("error", `Error: ${r.errorMessage}`), 0, 0)); + if (r.terminationSignal) + container.addChild(new Text(theme.fg("error", `Signal: ${r.terminationSignal}`), 0, 0)); + if (r.stderr.trim()) + container.addChild(new Text(theme.fg("error", `stderr: ${r.stderr.trim()}`), 0, 0)); // Show tool calls for (const item of displayItems) { @@ -1009,17 +1224,19 @@ export default function (pi: ExtensionAPI) { // Collapsed view (or still running) let text = `${icon} ${theme.fg("toolTitle", theme.bold("parallel "))}${theme.fg("accent", status)}`; for (const r of details.results) { - const rIcon = - r.exitCode === -1 - ? theme.fg("warning", "⏳") - : isFailedResult(r) - ? theme.fg("error", "✗") - : theme.fg("success", "✓"); + const rIcon = resultIcon(r); const displayItems = getDisplayItems(r.messages); text += `\n\n${theme.fg("muted", "─── ")}${theme.fg("accent", r.agent)} ${rIcon}`; - if (displayItems.length === 0) - text += `\n${theme.fg("muted", r.exitCode === -1 ? "(running...)" : "(no output)")}`; - else text += `\n${renderDisplayItems(displayItems, 5)}`; + if (displayItems.length === 0) { + const failed = isFailedResult(r); + let summary = "(no output)"; + if (r.status === "pending") summary = "(pending...)"; + else if (r.status === "running") summary = "(running...)"; + else if (failed) summary = getFailureSummary(r); + text += `\n${theme.fg(failed ? "error" : "muted", summary)}`; + } else { + text += `\n${renderDisplayItems(displayItems, 5)}`; + } } if (!isRunning) { const usageStr = formatUsageStats(aggregateUsage(details.results)); diff --git a/packages/coding-agent/test/fixtures/subagent-child.mjs b/packages/coding-agent/test/fixtures/subagent-child.mjs new file mode 100644 index 000000000..e68426d93 --- /dev/null +++ b/packages/coding-agent/test/fixtures/subagent-child.mjs @@ -0,0 +1,80 @@ +import fs from "node:fs"; + +const task = process.argv.find((arg) => arg.startsWith("Task: "))?.slice(6) ?? "success"; +const requestedModelIndex = process.argv.indexOf("--model"); +const requestedModel = requestedModelIndex >= 0 ? process.argv[requestedModelIndex + 1] : "fixture-model"; + +const zeroUsage = () => ({ + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, +}); + +const assistant = (text, options = {}) => ({ + role: "assistant", + content: options.toolCall + ? [ + { type: "text", text }, + { type: "toolCall", id: "call-1", name: "read", arguments: { path: "README.md" } }, + ] + : [{ type: "text", text }], + api: "anthropic-messages", + provider: "fixture", + model: options.model ?? requestedModel, + usage: options.usage ?? zeroUsage(), + stopReason: options.stopReason ?? "stop", + ...(options.errorMessage ? { errorMessage: options.errorMessage } : {}), + timestamp: Date.now(), +}); + +const emit = (event) => process.stdout.write(`${JSON.stringify(event)}\n`); +const emitMessage = (message) => emit({ type: "message_end", message }); + +if (task.startsWith("record-argv:")) { + fs.writeFileSync(task.slice("record-argv:".length), JSON.stringify(process.argv)); + emitMessage(assistant("done")); +} else if (task.startsWith("record-invocation:")) { + fs.writeFileSync( + task.slice("record-invocation:".length), + JSON.stringify({ argv: process.argv, cwd: process.cwd() }), + ); + emitMessage(assistant("done")); +} else if (task === "tool-then-final") { + emitMessage(assistant("working", { stopReason: "toolUse", toolCall: true })); + setTimeout(() => emitMessage(assistant("finished")), 150); +} else if (task.startsWith("tool-wait:")) { + const releaseFile = task.slice("tool-wait:".length); + emitMessage(assistant("working", { stopReason: "toolUse", toolCall: true })); + const timer = setInterval(() => { + if (!fs.existsSync(releaseFile)) return; + clearInterval(timer); + emitMessage(assistant("finished")); + }, 10); +} else if (task === "signal") { + process.kill(process.pid, "SIGTERM"); +} else if (task === "error" || task === "model-error") { + process.stderr.write("stderr evidence\n"); + emitMessage( + assistant("assistant evidence", { + stopReason: "error", + errorMessage: "provider evidence", + }), + ); + if (task === "error") process.exitCode = 1; +} else if (task.startsWith("large")) { + emitMessage(assistant(`${"x".repeat(60 * 1024)}\n${"line\n".repeat(2100)}`)); +} else if (task.startsWith("ignore-sigterm:")) { + fs.writeFileSync(task.slice("ignore-sigterm:".length), String(process.pid)); + process.on("SIGTERM", () => {}); + emitMessage(assistant("partial", { stopReason: "toolUse" })); + setInterval(() => {}, 1000); +} else if (task.startsWith("hold:")) { + fs.appendFileSync(task.slice("hold:".length), `${process.pid}\n`); + setInterval(() => {}, 1000); +} else { + process.stderr.write(`Unknown fixture task: ${task}\n`); + process.exitCode = 2; +} diff --git a/packages/coding-agent/test/subagent-extension.test.ts b/packages/coding-agent/test/subagent-extension.test.ts new file mode 100644 index 000000000..46c4184d3 --- /dev/null +++ b/packages/coding-agent/test/subagent-extension.test.ts @@ -0,0 +1,621 @@ +import { existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import type { AgentToolResult } from "@earendil-works/pi-agent-core"; +import { fauxAssistantMessage, fauxToolCall, type Message } from "@earendil-works/pi-ai"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import subagentExtension from "../examples/extensions/subagent/index.ts"; +import { CONFIG_DIR_NAME } from "../src/config.ts"; +import type { ExtensionAPI, ExtensionContext, ToolDefinition } from "../src/core/extensions/index.ts"; +import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES } from "../src/core/tools/truncate.ts"; +import { createHarness, type Harness } from "./suite/harness.ts"; + +const SUBAGENT_FIXTURE_PATH = fileURLToPath(new URL("./fixtures/subagent-child.mjs", import.meta.url)); + +const fixtureScenario = { + error: "error", + modelError: "model-error", + signal: "signal", + toolThenFinal: "tool-then-final", + large: (name: string) => `large-${name}`, + hold: (startsFile: string) => `hold:${startsFile}`, + ignoreSigterm: (pidFile: string) => `ignore-sigterm:${pidFile}`, + recordArgv: (outputFile: string) => `record-argv:${outputFile}`, + recordInvocation: (outputFile: string) => `record-invocation:${outputFile}`, + toolWait: (releaseFile: string) => `tool-wait:${releaseFile}`, +} as const; + +type ChildStatus = "pending" | "running" | "completed" | "failed" | "aborted"; + +interface TestSingleResult { + status: ChildStatus; + messages: Message[]; + stderr: string; + model?: string; + errorMessage?: string; + terminationSignal?: NodeJS.Signals; +} + +interface TestSubagentDetails { + results: TestSingleResult[]; +} + +function createProjectAgent( + cwd: string, + name = "worker", + options: { model?: string; body?: string; raw?: string } = {}, +): void { + const dir = join(cwd, CONFIG_DIR_NAME, "agents"); + mkdirSync(dir, { recursive: true }); + const content = + options.raw ?? + [ + "---", + `name: ${name}`, + `description: ${name} test agent`, + ...(options.model === undefined ? [] : [`model: ${options.model}`]), + "---", + options.body ?? "", + ].join("\n"); + writeFileSync(join(dir, `${name}.md`), content); +} + +function readStartedPids(path: string): string[] { + return readFileSync(path, "utf8").trim().split("\n"); +} + +function hasStartedPids(path: string, count: number): boolean { + return existsSync(path) && readStartedPids(path).length === count; +} + +function selectedModel(path: string): string | undefined { + const args = JSON.parse(readFileSync(path, "utf8")) as string[]; + return args[args.indexOf("--model") + 1]; +} + +function detailsOf(result: AgentToolResult): TestSubagentDetails { + return result.details as TestSubagentDetails; +} + +function resultText(result: AgentToolResult): string { + return result.content + .filter((part): part is { type: "text"; text: string } => part.type === "text") + .map((part) => part.text) + .join("\n"); +} + +function finalAssistantText(result: AgentToolResult, resultIndex = 0): string { + const messages = detailsOf(result).results[resultIndex]?.messages ?? []; + for (let index = messages.length - 1; index >= 0; index--) { + const message = messages[index]; + if (message.role !== "assistant") continue; + return message.content + .filter((part): part is { type: "text"; text: string } => part.type === "text") + .map((part) => part.text) + .join("\n"); + } + return ""; +} + +function createTool(): ToolDefinition { + let registered: ToolDefinition | undefined; + const api = { + on: vi.fn(), + registerTool(tool: ToolDefinition) { + registered = tool; + }, + } as unknown as ExtensionAPI; + subagentExtension(api); + if (!registered) throw new Error("subagent extension did not register its tool"); + return registered; +} + +function createContext(cwd: string): ExtensionContext { + return { + cwd, + hasUI: false, + model: { provider: "test", id: "dispatch-alias" }, + thinkingLevel: "high", + isProjectTrusted: () => false, + } as unknown as ExtensionContext; +} + +function createInteractiveContext( + base: ExtensionContext, + confirm: (title: string, message: string) => Promise, + trusted = false, +): ExtensionContext { + return { ...base, hasUI: true, ui: { confirm }, isProjectTrusted: () => trusted } as unknown as ExtensionContext; +} + +function singleTask(task: string, overrides: Record = {}): Record { + return { + agent: "worker", + task, + agentScope: "project", + confirmProjectAgents: false, + ...overrides, + }; +} + +async function withFixtureInvocation(run: () => Promise): Promise { + // The example re-executes process.argv[1] as Pi. Point that invocation at + // the deterministic fixture while retaining the real child_process.spawn. + const originalScript = process.argv[1]; + process.argv[1] = SUBAGENT_FIXTURE_PATH; + try { + return await run(); + } finally { + process.argv[1] = originalScript; + } +} + +function captureSettlement( + promise: Promise, +): Promise<{ status: "fulfilled"; result: T } | { status: "rejected"; error: unknown }> { + return promise.then( + (result) => ({ status: "fulfilled" as const, result }), + (error: unknown) => ({ status: "rejected" as const, error }), + ); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 3000): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error(`Timed out after ${timeoutMs}ms`); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function waitForProcessExit(pid: number, timeoutMs = 1000): Promise { + const deadline = Date.now() + timeoutMs; + while (isProcessAlive(pid) && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return !isProcessAlive(pid); +} + +function renderText(tool: ToolDefinition, result: AgentToolResult, expanded = false): string { + if (!tool.renderResult) throw new Error("subagent tool has no result renderer"); + const theme = { + fg: (_color: string, text: string) => text, + bold: (text: string) => text, + } as never; + const component = tool.renderResult(result, { expanded, isPartial: false }, theme, { + args: {}, + state: {}, + lastComponent: undefined, + invalidate: () => {}, + toolCallId: "test", + cwd: "/tmp", + executionStarted: true, + argsComplete: true, + isPartial: false, + expanded, + showImages: false, + isError: false, + }); + return component.render(160).join("\n"); +} + +describe("subagent example extension", () => { + let tempDir: string; + let tool: ToolDefinition; + let ctx: ExtensionContext; + const harnesses: Harness[] = []; + + function executeSubagent( + params: Record, + options: { + signal?: AbortSignal; + onUpdate?: (result: AgentToolResult) => void; + context?: ExtensionContext; + } = {}, + ): Promise> { + return withFixtureInvocation(() => + tool.execute("test", params, options.signal, options.onUpdate, options.context ?? ctx), + ); + } + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), "pi-subagent-test-")); + createProjectAgent(tempDir); + tool = createTool(); + ctx = createContext(tempDir); + }); + + afterEach(() => { + vi.useRealTimers(); + while (harnesses.length > 0) harnesses.pop()?.cleanup(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + describe("child lifecycle", () => { + it("keeps a child running across a tool-calling message until process close", async () => { + const updates: AgentToolResult[] = []; + let releaseUpdate: (() => void) | undefined; + const firstUpdate = new Promise((resolve) => { + releaseUpdate = resolve; + }); + + const execution = executeSubagent(singleTask(fixtureScenario.toolThenFinal), { + onUpdate: (update) => { + updates.push(update); + releaseUpdate?.(); + }, + }); + + await firstUpdate; + const streamingStatus = detailsOf(updates[0]).results[0].status; + const streamingRender = renderText(tool, updates[0]); + const result = await execution; + + expect(streamingStatus).toBe("running"); + expect(streamingRender).toContain("running"); + expect(detailsOf(result).results[0].status).toBe("completed"); + }); + + it("distinguishes pending tasks from four active workers", async () => { + const updates: AgentToolResult[] = []; + const releaseFile = join(tempDir, "release-workers"); + const execution = executeSubagent( + { + tasks: Array.from({ length: 5 }, () => ({ + agent: "worker", + task: fixtureScenario.toolWait(releaseFile), + })), + agentScope: "project", + confirmProjectAgents: false, + }, + { onUpdate: (update) => updates.push(update) }, + ); + + await waitFor(() => updates.length >= 4); + const statuses = detailsOf(updates.at(-1)!).results.map((result) => result.status); + writeFileSync(releaseFile, "release"); + await execution; + + expect(statuses.filter((status) => status === "running")).toHaveLength(4); + expect(statuses.filter((status) => status === "pending")).toHaveLength(1); + }); + + it("treats signal termination as a failed child with a diagnostic", async () => { + const result = await executeSubagent(singleTask(fixtureScenario.signal)); + + const child = detailsOf(result).results[0]; + expect(child.status).toBe("failed"); + expect(child.terminationSignal).toBe("SIGTERM"); + expect(resultText(result)).toContain("SIGTERM"); + }); + + it("preserves spawn errors for a nonexistent working directory", async () => { + const result = await executeSubagent( + singleTask("spawn-error", { cwd: join(tempDir, "missing-working-directory") }), + ); + + expect(detailsOf(result).results[0].status).toBe("failed"); + expect(resultText(result)).toMatch(/ENOENT|no such file or directory/i); + }); + + it("escalates SIGTERM to SIGKILL based on close state and preserves partial details", async () => { + const pidFile = join(tempDir, "ignores-sigterm.pid"); + const controller = new AbortController(); + const execution = captureSettlement( + executeSubagent(singleTask(fixtureScenario.ignoreSigterm(pidFile)), { signal: controller.signal }), + ); + + await waitFor(() => existsSync(pidFile)); + const pid = Number(readFileSync(pidFile, "utf8")); + vi.useFakeTimers(); + controller.abort(); + await vi.advanceTimersByTimeAsync(5000); + vi.useRealTimers(); + const exitedAfterEscalation = await waitForProcessExit(pid); + if (!exitedAfterEscalation) process.kill(pid, "SIGKILL"); + const settled = await execution; + + expect(exitedAfterEscalation).toBe(true); + expect(settled.status).toBe("fulfilled"); + if (settled.status === "fulfilled") { + const child = detailsOf(settled.result).results[0]; + expect(child.status).toBe("aborted"); + expect(child.messages).toHaveLength(1); + } + }); + + it("returns aborted task details when the signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(); + vi.useFakeTimers(); + const settled = await captureSettlement( + executeSubagent( + { + tasks: Array.from({ length: 5 }, () => ({ + agent: "worker", + task: fixtureScenario.toolThenFinal, + })), + agentScope: "project", + confirmProjectAgents: false, + }, + { signal: controller.signal }, + ), + ); + vi.useRealTimers(); + + expect(settled.status).toBe("fulfilled"); + if (settled.status === "fulfilled") { + expect(detailsOf(settled.result).results).toHaveLength(5); + expect(detailsOf(settled.result).results.every((result) => result.status === "aborted")).toBe(true); + expect(detailsOf(settled.result).results.every((result) => result.messages.length === 0)).toBe(true); + expect(resultText(settled.result)).toContain("aborted"); + } + }); + + it("does not start queued work after aborting active workers", async () => { + const controller = new AbortController(); + const startsFile = join(tempDir, "started-children.txt"); + const execution = captureSettlement( + executeSubagent( + { + tasks: Array.from({ length: 5 }, () => ({ + agent: "worker", + task: fixtureScenario.hold(startsFile), + })), + agentScope: "project", + confirmProjectAgents: false, + }, + { signal: controller.signal }, + ), + ); + + await waitFor(() => hasStartedPids(startsFile, 4)); + vi.useFakeTimers(); + controller.abort(); + const settled = await execution; + vi.useRealTimers(); + + expect(readStartedPids(startsFile)).toHaveLength(4); + expect(settled.status).toBe("fulfilled"); + if (settled.status === "fulfilled") { + expect(detailsOf(settled.result).results).toHaveLength(5); + expect(detailsOf(settled.result).results.every((result) => result.status === "aborted")).toBe(true); + } + }); + }); + + describe("runtime integration", () => { + it("marks a failed subagent invocation as an error tool result while preserving details", async () => { + const harness = await createHarness({ extensionFactories: [subagentExtension] }); + harnesses.push(harness); + createProjectAgent(harness.tempDir); + harness.setResponses([ + fauxAssistantMessage( + fauxToolCall("subagent", { + agent: "worker", + task: fixtureScenario.error, + agentScope: "project", + confirmProjectAgents: false, + }), + { stopReason: "toolUse" }, + ), + fauxAssistantMessage("done"), + ]); + + await withFixtureInvocation(() => harness.session.prompt("delegate")); + + const toolResult = harness.session.messages.find( + (message) => message.role === "toolResult" && message.toolName === "subagent", + ); + expect(toolResult?.role).toBe("toolResult"); + if (toolResult?.role === "toolResult") { + expect(toolResult.isError).toBe(true); + expect((toolResult.details as TestSubagentDetails).results[0].errorMessage).toBe("provider evidence"); + } + }); + + it("marks invalid mode combinations as error tool results", async () => { + const harness = await createHarness({ extensionFactories: [subagentExtension] }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage( + fauxToolCall("subagent", { + agent: "worker", + task: "single", + tasks: [{ agent: "worker", task: "parallel" }], + }), + { stopReason: "toolUse" }, + ), + fauxAssistantMessage("done"), + ]); + + await harness.session.prompt("delegate"); + + const toolResult = harness.session.messages.find( + (message) => message.role === "toolResult" && message.toolName === "subagent", + ); + expect(toolResult?.role === "toolResult" ? toolResult.isError : false).toBe(true); + }); + }); + + describe("task cwd agent discovery", () => { + it("resolves a relative task cwd against the session cwd for discovery and execution", async () => { + const startupRoot = join(tempDir, "startup-project"); + const sessionRoot = join(tempDir, "session-project"); + const startupChild = join(startupRoot, "child"); + const sessionChild = join(sessionRoot, "child"); + mkdirSync(startupChild, { recursive: true }); + mkdirSync(sessionChild, { recursive: true }); + createProjectAgent(startupChild, "worker", { model: "startup-model" }); + createProjectAgent(sessionChild, "worker", { model: "session-model" }); + const invocationFile = join(tempDir, "relative-cwd-invocation.json"); + const originalCwd = process.cwd(); + + try { + process.chdir(startupRoot); + await executeSubagent(singleTask(fixtureScenario.recordInvocation(invocationFile), { cwd: "child" }), { + context: createContext(sessionRoot), + }); + } finally { + process.chdir(originalCwd); + } + + const invocation = JSON.parse(readFileSync(invocationFile, "utf8")) as { argv: string[]; cwd: string }; + expect(invocation.cwd).toBe(realpathSync(sessionChild)); + expect(invocation.argv[invocation.argv.indexOf("--model") + 1]).toBe("session-model"); + }); + + it("uses and confirms the project agent from the child task cwd", async () => { + const childCwd = join(tempDir, "confirmed-project"); + mkdirSync(childCwd); + createProjectAgent(tempDir, "worker", { model: "root-model" }); + createProjectAgent(childCwd, "worker", { model: "child-model" }); + const argvFile = join(tempDir, "confirmed-child-argv.json"); + const confirm = vi.fn(async (_title: string, _message: string) => true); + await executeSubagent( + singleTask(fixtureScenario.recordArgv(argvFile), { + cwd: childCwd, + confirmProjectAgents: true, + }), + { context: createInteractiveContext(ctx, confirm) }, + ); + + expect.soft(selectedModel(argvFile)).toBe("child-model"); + expect(confirm).toHaveBeenCalledWith( + "Run project-local agents?", + expect.stringContaining(join(childCwd, CONFIG_DIR_NAME, "agents")), + ); + }); + + // Regression coverage for #8261 when a trusted session delegates into another project. + it("confirms an external task cwd agent even when the session trust context reports trusted", async () => { + const projectsRoot = join(tempDir, "projects"); + const sessionCwd = join(projectsRoot, "session-project"); + const externalCwd = join(projectsRoot, "external-project"); + mkdirSync(sessionCwd, { recursive: true }); + mkdirSync(externalCwd, { recursive: true }); + createProjectAgent(sessionCwd, "worker", { model: "session-model" }); + createProjectAgent(externalCwd, "worker", { model: "external-model" }); + const argvFile = join(tempDir, "external-child-argv.json"); + const confirm = vi.fn(async (_title: string, _message: string) => true); + await executeSubagent( + singleTask(fixtureScenario.recordArgv(argvFile), { + cwd: externalCwd, + confirmProjectAgents: true, + }), + { context: createInteractiveContext(createContext(sessionCwd), confirm, true) }, + ); + + expect.soft(selectedModel(argvFile)).toBe("external-model"); + expect(confirm).toHaveBeenCalledOnce(); + expect(confirm).toHaveBeenCalledWith( + "Run project-local agents?", + expect.stringContaining(join(externalCwd, CONFIG_DIR_NAME, "agents")), + ); + expect(String(confirm.mock.calls[0][1])).not.toContain(join(sessionCwd, CONFIG_DIR_NAME, "agents")); + }); + + it("resolves and confirms each project-agent source for parallel task cwd values", async () => { + const projectA = join(tempDir, "parallel-a"); + const projectB = join(tempDir, "parallel-b"); + mkdirSync(projectA); + mkdirSync(projectB); + createProjectAgent(projectA, "worker", { model: "model-a" }); + createProjectAgent(projectB, "worker", { model: "model-b" }); + const argvA1 = join(tempDir, "parallel-a1-argv.json"); + const argvA2 = join(tempDir, "parallel-a2-argv.json"); + const argvB = join(tempDir, "parallel-b-argv.json"); + const confirm = vi.fn(async (_title: string, _message: string) => true); + await executeSubagent( + { + tasks: [ + { agent: "worker", task: fixtureScenario.recordArgv(argvA1), cwd: projectA }, + { agent: "worker", task: fixtureScenario.recordArgv(argvA2), cwd: projectA }, + { agent: "worker", task: fixtureScenario.recordArgv(argvB), cwd: projectB }, + ], + agentScope: "project", + confirmProjectAgents: true, + }, + { context: createInteractiveContext(ctx, confirm) }, + ); + + expect.soft(selectedModel(argvA1)).toBe("model-a"); + expect.soft(selectedModel(argvA2)).toBe("model-a"); + expect.soft(selectedModel(argvB)).toBe("model-b"); + + const confirmationText = confirm.mock.calls.map((call) => String(call[1])).join("\n"); + const sourceA = join(projectA, CONFIG_DIR_NAME, "agents"); + const sourceB = join(projectB, CONFIG_DIR_NAME, "agents"); + expect.soft(confirmationText.split(sourceA).length - 1).toBe(1); + expect.soft(confirmationText.split(sourceB).length - 1).toBe(1); + }); + }); + + describe("model-facing output and rendering", () => { + it.each([ + { name: "single", params: singleTask(fixtureScenario.large("single")) }, + { + name: "chain", + params: { + chain: [{ agent: "worker", task: fixtureScenario.large("chain") }], + agentScope: "project", + confirmProjectAgents: false, + }, + }, + ])("limits $name output while retaining complete details", async ({ params }) => { + const result = await executeSubagent(params); + + const text = resultText(result); + expect(Buffer.byteLength(text, "utf8")).toBeLessThanOrEqual(DEFAULT_MAX_BYTES); + expect(text.split("\n").length).toBeLessThanOrEqual(DEFAULT_MAX_LINES); + expect(text).toMatch(/truncat/i); + const fullText = finalAssistantText(result); + expect(Buffer.byteLength(fullText, "utf8")).toBeGreaterThan(DEFAULT_MAX_BYTES); + expect(fullText.split("\n").length).toBeGreaterThan(DEFAULT_MAX_LINES); + }); + + it("shows failure diagnostics in collapsed results", async () => { + const result = await executeSubagent({ + chain: [{ agent: "worker", task: fixtureScenario.signal }], + agentScope: "project", + confirmProjectAgents: false, + }); + + expect(renderText(tool, result)).toContain("SIGTERM"); + }); + + function runFailedChain(task: string = fixtureScenario.error): Promise> { + return executeSubagent({ + chain: [{ agent: "worker", task }], + agentScope: "project", + confirmProjectAgents: false, + }); + } + + it("preserves provider errors, stderr, and assistant output as distinct evidence", async () => { + const result = await runFailedChain(); + const child = detailsOf(result).results[0]; + expect.soft(child.errorMessage).toBe("provider evidence"); + expect.soft(child.stderr).toContain("stderr evidence"); + expect.soft(finalAssistantText(result)).toContain("assistant evidence"); + + const expanded = renderText(tool, result, true); + expect.soft(expanded).toContain("provider evidence"); + expect.soft(expanded).toContain("stderr evidence"); + expect.soft(expanded).toContain("assistant evidence"); + }); + + it("renders zero-exit model errors as failed chain steps", async () => { + expect(renderText(tool, await runFailedChain(fixtureScenario.modelError))).toContain("0/1 steps"); + }); + }); +});