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.

<details>
<summary>Potential follow-ups</summary>

### 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.

</details>

## 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
This commit is contained in:
Michael Renner 2026-11-04 11:24:12 +01:00 • committed by GitHub
parent dbc61d647b
commit aba42ae6b9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1103 additions and 182 deletions

View file

@ -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

View file

@ -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<string>();
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<string, any> };
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<SingleResult> {
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<number>((resolve) => {
args.push(`Task: ${task}`);
emitUpdate();
let wasAborted = false;
const outcome = await new Promise<ProcessOutcome>((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<typeof setTimeout> | 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<typeof spawn>;
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<string>();
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<string, AgentConfig>();
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));

View file

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

View file

@ -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<unknown>): TestSubagentDetails {
return result.details as TestSubagentDetails;
}
function resultText(result: AgentToolResult<unknown>): 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<unknown>, 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<boolean>,
trusted = false,
): ExtensionContext {
return { ...base, hasUI: true, ui: { confirm }, isProjectTrusted: () => trusted } as unknown as ExtensionContext;
}
function singleTask(task: string, overrides: Record<string, unknown> = {}): Record<string, unknown> {
return {
agent: "worker",
task,
agentScope: "project",
confirmProjectAgents: false,
...overrides,
};
}
async function withFixtureInvocation<T>(run: () => Promise<T>): Promise<T> {
// 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<T>(
promise: Promise<T>,
): 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<void> {
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<boolean> {
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<unknown>, 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<string, unknown>,
options: {
signal?: AbortSignal;
onUpdate?: (result: AgentToolResult<unknown>) => void;
context?: ExtensionContext;
} = {},
): Promise<AgentToolResult<unknown>> {
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<unknown>[] = [];
let releaseUpdate: (() => void) | undefined;
const firstUpdate = new Promise<void>((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<unknown>[] = [];
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<AgentToolResult<unknown>> {
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");
});
});
});