fix(interactive-subagents): harden parallel lifecycle delivery

This commit is contained in:
jb
2026-09-03 23:40:14 -04:00
parent d99dbce93d
commit 7a1043e5e0
11 changed files with 848 additions and 131 deletions
@@ -0,0 +1,9 @@
---
name: test-default-autonomous
description: Integration test agent with omitted auto-exit
model: anthropic/claude-haiku-4-5
tools: bash
disable-model-invocation: true
---
You are a test agent. Run the requested bash command immediately, then report completion concisely. Do not ask questions or do anything else.
@@ -18,6 +18,7 @@
import { describe, it, before, after } from "node:test";
import assert from "node:assert/strict";
import { existsSync, readFileSync, unlinkSync } from "node:fs";
import { join } from "node:path";
import {
getAvailableBackends,
createTestEnv,
@@ -31,6 +32,7 @@ import {
trackTempFile,
readScreen,
sendCommand,
shellEscape,
PI_TIMEOUT,
type TestEnv,
} from "./harness.ts";
@@ -202,42 +204,115 @@ for (const backend of backends) {
// ── Parallel subagent spawn ──
it("spawns two subagents in parallel and both complete", async () => {
it("delivers two staggered sibling launches exactly once after the parent becomes idle", async () => {
const id = uniqueId();
const nameA = `ParaA-${id}`;
const nameB = `ParaB-${id}`;
const fileA = `/tmp/pi-integ-para-${id}-a.txt`;
const fileB = `/tmp/pi-integ-para-${id}-b.txt`;
const parentWorkFile = `/tmp/pi-integ-para-${id}-parent.txt`;
const parentSession = join(env.dir, `parallel-parent-${id}.jsonl`);
trackTempFile(env, fileA);
trackTempFile(env, fileB);
trackTempFile(env, parentWorkFile);
const surface = createTrackedSurface(env, `parallel-${id}`);
await sleep(1000);
const task = [
`You must call the subagent tool TWICE. Make both calls before waiting for results.`,
`Call the subagent tool twice in ONE assistant response so Pi executes them as parallel sibling tool calls.`,
`Use these exact parameters:`,
``,
`First call:`,
` name: "ParaA-${id}"`,
` agent: "test-echo"`,
` task: "Run: echo 'DONE_A_${id}' > '${fileA}'"`,
` name: "${nameA}"`,
` agent: "test-default-autonomous"`,
` cwd: "${env.dir}"`,
` task: "Run: sleep 6; echo 'DONE_A_${id}' > '${fileA}'"`,
``,
`Second call:`,
` name: "ParaB-${id}"`,
` agent: "test-echo"`,
` task: "Run: echo 'DONE_B_${id}' > '${fileB}'"`,
` name: "${nameB}"`,
` agent: "test-default-autonomous"`,
` cwd: "${env.dir}"`,
` task: "Run: sleep 12; echo 'DONE_B_${id}' > '${fileB}'"`,
``,
`Call both subagent tools NOW, do not wait between them.`,
`After both launch acknowledgements, use bash yourself to write PARENT_WORK_${id} to '${parentWorkFile}', then say you are waiting and stop.`,
`Never call subagent_message. When the first result arrives, acknowledge it and stop while the other child runs.`,
`When both results have arrived, say PARALLEL_IDLE_COMPLETE_${id} exactly once.`,
].join("\n");
startPi(surface, env.dir, task);
startPi(surface, env.dir, task, {
extraArgs: `--session ${shellEscape(parentSession)}`,
});
// Both marker files should appear
const [contentA, contentB] = await Promise.all([
waitForFile(fileA, PI_TIMEOUT, /DONE_A/),
waitForFile(fileB, PI_TIMEOUT, /DONE_B/),
]);
await waitForFile(parentWorkFile, PI_TIMEOUT, /PARENT_WORK_/);
const screen = await waitForScreen(
surface,
new RegExp(`^ PARALLEL_IDLE_COMPLETE_${id}\\s*$`, "m"),
PI_TIMEOUT,
400,
);
assert.equal(
screen.match(new RegExp(`^ PARALLEL_IDLE_COMPLETE_${id}\\s*$`, "gm"))?.length,
1,
);
assert.match(await waitForFile(fileA, PI_TIMEOUT, /DONE_A_/), new RegExp(`DONE_A_${id}`));
assert.match(await waitForFile(fileB, PI_TIMEOUT, /DONE_B_/), new RegExp(`DONE_B_${id}`));
assert.ok(contentA.includes(`DONE_A_${id}`), `File A should contain marker`);
assert.ok(contentB.includes(`DONE_B_${id}`), `File B should contain marker`);
const entries = readFileSync(parentSession, "utf8")
.trim()
.split("\n")
.map((line) => JSON.parse(line));
const siblingCall = entries.find((entry) => {
const calls = entry.message?.role === "assistant"
? entry.message.content?.filter((part: any) => part.type === "toolCall" && part.name === "subagent")
: [];
return calls?.length === 2;
});
assert.ok(siblingCall, "both wrapped launches should reach Pi as sibling subagent calls");
const calls = siblingCall.message.content.filter(
(part: any) => part.type === "toolCall" && part.name === "subagent",
);
assert.deepEqual(calls.map((call: any) => call.arguments.name), [nameA, nameB]);
assert.deepEqual(calls.map((call: any) => call.arguments.cwd), [env.dir, env.dir]);
const acknowledgements = entries.filter(
(entry) => entry.message?.role === "toolResult" && entry.message.toolName === "subagent",
);
assert.deepEqual(
acknowledgements.map((entry) => entry.message.details?.name).sort(),
[nameA, nameB].sort(),
);
for (const acknowledgement of acknowledgements) {
assert.equal(acknowledgement.message.details?.status, "started");
const childHeader = JSON.parse(
readFileSync(acknowledgement.message.details.sessionFile, "utf8").split("\n")[0],
);
assert.equal(childHeader.cwd, env.dir);
}
const completions = entries.filter(
(entry) => entry.type === "custom_message" && entry.customType === "subagent_result",
);
assert.deepEqual(completions.map((entry) => entry.details?.name).sort(), [nameA, nameB].sort());
assert.equal(completions.length, 2, "each child result should be delivered exactly once");
assert.ok(completions.every((entry) => entry.details?.exitCode === 0));
const idleBeforeCompletion = entries.findIndex((entry, index) => {
if (index <= entries.indexOf(acknowledgements.at(-1))) return false;
if (entry.message?.role !== "assistant" || entry.message.stopReason !== "stop") return false;
return completions.every((completion) => Date.parse(completion.timestamp) > Date.parse(entry.timestamp));
});
assert.notEqual(idleBeforeCompletion, -1, "the parent should settle before either child completes");
for (const completion of completions) {
assert.ok(
entries.some(
(entry) => entry.message?.role === "assistant"
&& Date.parse(entry.timestamp) > Date.parse(completion.timestamp),
),
`${completion.details?.name} should trigger a parent turn while idle`,
);
}
});
// ── Fork mode ──
+312 -8
View File
@@ -158,13 +158,16 @@ async function emitMockEvent(
type SharedRuntimeTestState = {
runningSubagents: Map<string, unknown>;
resumingNames: Set<string>;
latestCtx: unknown;
latestPi: unknown;
widgetInterval: ReturnType<typeof setInterval> | null;
statusInterval: ReturnType<typeof setInterval> | null;
standaloneWidgetSyncScheduled: boolean;
pendingDeliveries: { length: number };
resultDeliveryStates: Map<string, "queued" | "delivered">;
deliveryFlushScheduled: boolean;
parentGeneration: number;
activeRuntimeToken: symbol | null;
registeredRuntimeToken: symbol | null;
statusWidgetClaim: unknown;
@@ -178,11 +181,14 @@ function resetSharedRuntimeForTest(shared: SharedRuntimeTestState): void {
shared.statusInterval = null;
shared.standaloneWidgetSyncScheduled = false;
shared.runningSubagents.clear();
shared.resumingNames.clear();
shared.latestCtx = null;
shared.latestPi = null;
shared.statusWidgetClaim = null;
shared.pendingDeliveries.length = 0;
shared.resultDeliveryStates.clear();
shared.deliveryFlushScheduled = false;
shared.parentGeneration = 0;
shared.activeRuntimeToken = null;
shared.registeredRuntimeToken = null;
}
@@ -1201,28 +1207,53 @@ describe("subagent discovery", () => {
});
});
it("resolveEffectiveInteractive defaults to the inverse of auto-exit", () => {
// Autonomous agents (auto-exit: true) are NOT interactive — parent gets stall pings.
it("defaults unspecified launch behavior to autonomous while preserving explicit interactive profiles", () => {
assert.equal(testApi.resolveEffectiveAutoExit({ autoExit: true }), true);
assert.equal(testApi.resolveEffectiveAutoExit({ autoExit: false }), false);
assert.equal(testApi.resolveEffectiveAutoExit({}), true);
assert.equal(testApi.resolveEffectiveAutoExit(null), true);
assert.equal(testApi.autoExitEnvAssignment(true), "PI_SUBAGENT_AUTO_EXIT=1");
assert.equal(testApi.autoExitEnvAssignment(false), "PI_SUBAGENT_AUTO_EXIT=0");
assert.match(testApi.resolveClaudeSentinelFile("child", true), /pi-claude-child-done$/);
assert.equal(testApi.resolveClaudeSentinelFile("child", false), undefined);
// Autonomous agents are not user-driven, so parent status transitions stay enabled.
assert.equal(
testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, { autoExit: true }),
false,
);
// Agents without auto-exit ARE interactive — parent does not receive status transition pings.
// Explicitly non-auto-exit profiles remain interactive.
assert.equal(
testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, { autoExit: false }),
true,
);
// Missing auto-exit follows the fire-and-forget tool contract.
assert.equal(
testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, {}),
true,
false,
);
// Bare spawn with no agent defs (e.g. /iterate fork) is interactive by default.
assert.equal(
testApi.resolveEffectiveInteractive({ name: "A", task: "T" }, null),
true,
false,
);
});
it("resolves relative launch cwd from the parent session rather than the process", () => {
const resolved = testApi.resolveSubagentPaths(
{ agent: "worker", task: "T", cwd: "roles/reviewer" },
{},
"/workspace/current-session",
);
assert.equal(resolved.effectiveCwd, "/workspace/current-session/roles/reviewer");
const inherited = testApi.resolveSubagentPaths(
{ agent: "worker", task: "T" },
{},
"/workspace/current-session",
);
assert.equal(inherited.effectiveCwd, null);
});
it("resolveEffectiveInteractive honors explicit frontmatter over the auto-exit default", () => {
// Autonomous agent that still wants to be treated as interactive.
assert.equal(
@@ -1851,6 +1882,42 @@ describe("surface exit signals", () => {
}
});
it("terminates after repeated liveness failures when a completion-file surface is closed", async () => {
await assert.rejects(
pollForExit(
"closed-surface",
new AbortController().signal,
{
interval: 1,
completionFile: "/tmp/nonexistent-subagent-completion",
livenessCheck: async () => false,
livenessIntervalMs: 0,
livenessFailureThreshold: 2,
},
),
/surface closed before completion/i,
);
});
it("aborts an in-flight completion-file liveness probe", async () => {
const controller = new AbortController();
const polling = pollForExit(
"blocked-surface",
controller.signal,
{
interval: 1,
completionFile: "/tmp/nonexistent-subagent-completion",
livenessIntervalMs: 0,
livenessCheck: async (_surface: string, signal?: AbortSignal) =>
await new Promise<boolean>((_resolve, reject) => {
signal?.addEventListener("abort", () => reject(new Error("probe aborted")), { once: true });
}),
},
);
setTimeout(() => controller.abort(), 10);
await assert.rejects(polling, /aborted/i);
});
it("bounds an asynchronous Orca screen read and kills its CLI process", async () => {
const dir = createTestDir();
const pidFile = join(dir, "reader.pid");
@@ -1966,6 +2033,182 @@ describe("reload lifecycle", () => {
beforeEach(() => resetSharedRuntimeForTest(shared));
after(() => resetSharedRuntimeForTest(shared));
it("delivers parallel terminal results exactly once across active and temporarily unbound runtimes", async () => {
const activeRuntime = createMockExtensionApi();
(subagentsModule as any).default(activeRuntime.api);
await emitMockEvent(activeRuntime, "session_start", { reason: "startup" }, ctx);
const deliver = (name: string) => (pi: any) => {
pi.sendMessage({ customType: "subagent_result", content: name, display: true });
};
assert.equal(testApi.sendResultWhenRuntimeReady("child-a", deliver("child-a")), true);
assert.equal(testApi.sendResultWhenRuntimeReady("child-a", deliver("child-a duplicate")), false);
let transientAttempts = 0;
assert.equal(testApi.sendResultWhenRuntimeReady("child-retry", (pi: any) => {
transientAttempts += 1;
if (transientAttempts === 1) throw new Error("stale runtime");
pi.sendMessage({ customType: "subagent_result", content: "child-retry", display: true });
}), true);
assert.equal(testApi.sendResultWhenRuntimeReady("child-independent", deliver("child-independent")), true);
assert.deepEqual(
activeRuntime.sentMessages.map(({ message }) => message.content),
["child-a", "child-independent"],
"one failing delivery must not block its sibling",
);
// Registration happens before session_start during reload. Results in this
// gap must queue rather than use the still-bound old ExtensionAPI.
const freshRuntime = createMockExtensionApi();
(subagentsModule as any).default(freshRuntime.api);
assert.equal(testApi.sendResultWhenRuntimeReady("child-b", deliver("child-b")), true);
assert.equal(activeRuntime.sentMessages.length, 2, "the stale runtime must not receive child-b");
await emitMockEvent(activeRuntime, "session_shutdown", { reason: "reload" }, ctx);
assert.equal(testApi.sendResultWhenRuntimeReady("child-c", deliver("child-c failed")), true);
assert.equal(testApi.sendResultWhenRuntimeReady("child-b", deliver("child-b duplicate")), false);
assert.equal(shared.pendingDeliveries.length, 3);
await emitMockEvent(freshRuntime, "session_start", { reason: "reload" }, ctx);
await new Promise<void>((resolve) => setImmediate(resolve));
assert.deepEqual(
freshRuntime.sentMessages.map(({ message }) => message.content),
["child-retry", "child-b", "child-c failed"],
);
testApi.flushPendingDeliveries();
assert.equal(freshRuntime.sentMessages.length, 3);
assert.equal(testApi.sendResultWhenRuntimeReady("child-c", deliver("post-flush duplicate")), false);
});
it("bounds delivered result tombstones without evicting queued results", () => {
shared.latestPi = { sendMessage() {} };
shared.activeRuntimeToken = Symbol.for("test-runtime");
shared.registeredRuntimeToken = shared.activeRuntimeToken;
for (let index = 0; index < 1_100; index += 1) {
assert.equal(testApi.sendResultWhenRuntimeReady(`finished-${index}`, () => {}), true);
}
assert.ok(shared.resultDeliveryStates.size <= 1_024);
});
it("rejects launches after cancellation or runtime replacement", () => {
const controller = new AbortController();
const runtimeToken = Symbol("runtime");
shared.activeRuntimeToken = runtimeToken;
shared.registeredRuntimeToken = runtimeToken;
assert.doesNotThrow(() => testApi.assertLaunchActive(controller.signal, runtimeToken));
controller.abort();
assert.throws(
() => testApi.assertLaunchActive(controller.signal, runtimeToken),
/parent session ended/i,
);
const currentController = new AbortController();
shared.registeredRuntimeToken = Symbol("replacement");
assert.throws(
() => testApi.assertLaunchActive(currentController.signal, runtimeToken),
/parent session ended/i,
);
});
it("rejects a completion from a replaced parent even after its watcher left the running map", () => {
const completed = {
id: "old-child",
name: "old-child",
task: "done",
surface: "old-pane",
startTime: 0,
sessionFile: "/tmp/old-child.jsonl",
interactive: false,
parentGeneration: shared.parentGeneration,
statusState: createStatusState({ source: "pi", startTimeMs: 0 }),
};
assert.equal(testApi.shouldDeliverRunningResult(completed), true);
shared.parentGeneration += 1;
assert.equal(testApi.shouldDeliverRunningResult(completed), false);
});
it("drops a queued subagent question after parent replacement", () => {
const dir = createTestDir();
const sessionFile = join(dir, "question-child.jsonl");
const running = {
id: "question-child",
name: "question-child",
task: "ask",
agent: "worker",
surface: "question-pane",
startTime: 0,
sessionFile,
interactive: false,
parentGeneration: shared.parentGeneration,
statusState: createStatusState({ source: "pi", startTimeMs: 0 }),
};
writeFileSync(`${sessionFile}.ask`, "{");
try {
testApi.deliverPendingQuestion(running);
assert.equal(existsSync(`${sessionFile}.ask`), true, "partial signals must be retried");
assert.equal(shared.pendingDeliveries.length, 0);
writeFileSync(`${sessionFile}.ask`, JSON.stringify({ question: "Still there?" }));
testApi.deliverPendingQuestion(running);
assert.equal(existsSync(`${sessionFile}.ask`), false);
assert.equal(shared.pendingDeliveries.length, 1);
shared.parentGeneration += 1;
const replacement = createMockExtensionApi();
shared.latestPi = replacement.api;
shared.activeRuntimeToken = Symbol.for("replacement-runtime");
shared.registeredRuntimeToken = shared.activeRuntimeToken;
testApi.flushPendingDeliveries();
assert.equal(replacement.sentMessages.length, 0);
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it("invalidates old children when replacement session_start wins the shutdown race", async () => {
const oldRuntime = createMockExtensionApi();
(subagentsModule as any).default(oldRuntime.api);
await emitMockEvent(oldRuntime, "session_start", { reason: "startup" }, ctx);
const abortController = new AbortController();
shared.runningSubagents.set("replaced-child", {
id: "replaced-child",
name: "replaced-child",
task: "old work",
surface: "old-pane",
startTime: 0,
sessionFile: "/tmp/replaced-child.jsonl",
interactive: false,
parentGeneration: shared.parentGeneration,
statusState: createStatusState({ source: "pi", startTimeMs: 0 }),
abortController,
});
testApi.sendWhenRuntimeReady(() => {
throw new Error("queue this old callback");
});
assert.equal(shared.pendingDeliveries.length, 1);
const freshRuntime = createMockExtensionApi();
(subagentsModule as any).default(freshRuntime.api);
await emitMockEvent(
freshRuntime,
"session_start",
{ reason: "new", previousSessionFile: "/tmp/old-parent.jsonl" },
ctx,
);
assert.equal(abortController.signal.aborted, true);
assert.equal(shared.runningSubagents.size, 0);
assert.equal(shared.pendingDeliveries.length, 0);
assert.equal(shared.parentGeneration, 1);
await emitMockEvent(oldRuntime, "session_shutdown", { reason: "new" }, ctx);
assert.equal(shared.parentGeneration, 1, "delayed old shutdown must not invalidate twice");
});
it("preserves a live subagent and delivers its queued result through the fresh runtime", async () => {
const oldRuntime = createMockExtensionApi();
(subagentsModule as any).default(oldRuntime.api);
@@ -1981,6 +2224,7 @@ describe("reload lifecycle", () => {
sessionFile: string;
statusState: ReturnType<typeof createStatusState>;
interactive: boolean;
parentGeneration: number;
abortController: AbortController;
suppressDelivery?: boolean;
} = {
@@ -1992,6 +2236,7 @@ describe("reload lifecycle", () => {
sessionFile: "/tmp/reload-child.jsonl",
statusState: createStatusState({ source: "pi", startTimeMs: Date.now() }),
interactive: false,
parentGeneration: shared.parentGeneration,
abortController,
};
shared.runningSubagents.set(running.id, running);
@@ -2032,9 +2277,13 @@ describe("reload lifecycle", () => {
});
describe("tool registration", () => {
it("always resumes subagents as autonomous (auto-exit, non-interactive tracking)", () => {
it("always resumes subagents as autonomous with reliable follow-up delivery", () => {
const testApi = (subagentsModule as any).__test__;
assert.deepEqual(testApi.AUTOMATIC_DELIVERY_OPTIONS, {
triggerTurn: true,
deliverAs: "followUp",
});
assert.deepEqual(testApi.resolveResumeLaunchBehavior(), {
autoExit: true,
interactive: false,
@@ -2367,7 +2616,7 @@ describe("subagent interruption", () => {
}
});
it("uniqueRunningName suffixes defaulted names that collide with running subagents", () => {
it("uniqueRunningName suffixes explicit or defaulted names that collide with running subagents", () => {
const testApi = (subagentsModule as any).__test__;
const runningMap = testApi.runningSubagents as Map<string, any>;
runningMap.clear();
@@ -2432,6 +2681,61 @@ describe("subagent interruption", () => {
}
});
it("serializes concurrent finished-session resumes by name", () => {
const testApi = (subagentsModule as any).__test__;
const resumingNames = testApi.resumingNames as Set<string>;
resumingNames.clear();
try {
assert.equal(testApi.reserveResumeName("Worker"), true);
assert.equal(testApi.reserveResumeName("Worker"), false);
assert.equal(testApi.reserveResumeName("Scout"), true);
} finally {
resumingNames.clear();
}
});
it("clears stale completion signals before resuming a finished session", () => {
const testApi = (subagentsModule as any).__test__;
const dir = createTestDir();
const sessionFile = join(dir, "finished.jsonl");
writeFileSync(`${sessionFile}.exit`, JSON.stringify({ type: "error" }));
writeFileSync(`${sessionFile}.ask`, JSON.stringify({ question: "stale" }));
try {
testApi.clearStaleSessionSignals(sessionFile);
assert.equal(existsSync(`${sessionFile}.exit`), false);
assert.equal(existsSync(`${sessionFile}.ask`), false);
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it("recognizes terminal completion and waits for the watcher before late steering", async () => {
const testApi = (subagentsModule as any).__test__;
const runningMap = testApi.runningSubagents as Map<string, any>;
const dir = createTestDir();
const completionFile = join(dir, "child.exit-code");
const running: any = makeRunning({ sessionFile: join(dir, "child.jsonl"), completionFile });
runningMap.set(running.id, running);
try {
assert.equal(testApi.hasTerminalCompletionSignal(running), false);
writeFileSync(completionFile, "0\n");
assert.equal(testApi.hasTerminalCompletionSignal(running), true);
rmSync(completionFile);
running.completionObserved = true;
assert.equal(
testApi.hasTerminalCompletionSignal(running),
true,
"watcher state must preserve completion after consuming the file",
);
setTimeout(() => runningMap.delete(running.id), 10);
assert.equal(await testApi.waitForWatcherCleanup(running, new AbortController().signal, 200), true);
} finally {
runningMap.clear();
rmSync(dir, { recursive: true, force: true });
}
});
it("steers a running subagent by typing into its pane (newlines flattened)", () => {
const testApi = (subagentsModule as any).__test__;
let sentSurface = "";