diff --git a/extensions/pi-subagents/.gitignore b/extensions/pi-subagents/.gitignore index f2a57e9..f2537b4 100644 --- a/extensions/pi-subagents/.gitignore +++ b/extensions/pi-subagents/.gitignore @@ -7,3 +7,4 @@ node_modules/ package-lock.json .spec/ +bun.lock diff --git a/extensions/pi-subagents/skills/pi-subagents/SKILL.md b/extensions/pi-subagents/skills/pi-subagents/SKILL.md index bd08d18..4b32d52 100644 --- a/extensions/pi-subagents/skills/pi-subagents/SKILL.md +++ b/extensions/pi-subagents/skills/pi-subagents/SKILL.md @@ -16,6 +16,7 @@ Use this tool to launch unrestricted child Pi sessions. The caller must include - Timeout is only a notification threshold: the parent is informed that the child is still running; the child is not killed. - The returned subagent id is also the child Pi session id. - Child output is written to `result.log` under the child subagent directory. + - When `model` is omitted, the child inherits the parent session's active model (e.g., `openai-codex/gpt-5.6-sol`). Pass an explicit `model` to override (e.g., `"anthropic/claude-sonnet-4-5"`). ## Usage diff --git a/extensions/pi-subagents/src/extension/index.ts b/extensions/pi-subagents/src/extension/index.ts index e79df86..0cf472c 100644 --- a/extensions/pi-subagents/src/extension/index.ts +++ b/extensions/pi-subagents/src/extension/index.ts @@ -1188,7 +1188,38 @@ function startChild( throw new Error("Subagent process lifeline pipe was not created."); } lifelines.set(record.id, lifeline); - child.stdout.pipe(stdoutStream); + + // Monitor stdout for terminal provider errors (e.g. HTTP 402). + // When stopReason==="error" is emitted, the child Pi process may + // stay alive indefinitely. Kill it so the close handler finalizes + // the record promptly. + let terminalErrorText: string | undefined; + let errorTimer: NodeJS.Timeout | undefined; + child.stdout.on("data", (chunk: Buffer) => { + // Write to file (replaces pipe). + stdoutStream.write(chunk); + if (terminalErrorText) return; + const text = chunk.toString("utf-8"); + if (/"stopReason"\s*:\s*"error"/.test(text)) { + const errMatch = text.match(/"errorMessage"\s*:\s*"([^"]+)"/); + terminalErrorText = errMatch?.[1] ?? "Provider error"; + if (!record.error) { + record.error = terminalErrorText; + } + errorTimer = setTimeout(() => { + errorTimer = undefined; + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGTERM"); + setTimeout(() => { + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGKILL"); + } + }, 1000).unref(); + } + }, 1000); + errorTimer.unref(); + } + }); child.stderr.pipe(stderrStream); const done = new Promise((resolve) => { @@ -1201,6 +1232,12 @@ function startChild( if (finalized) return; finalized = true; + // Clear error-detection timer if still pending. + if (errorTimer) { + clearTimeout(errorTimer); + errorTimer = undefined; + } + await Promise.all([ new Promise((r) => stdoutStream.end(r)), new Promise((r) => stderrStream.end(r)), @@ -1227,7 +1264,10 @@ function startChild( record.completedAt = Date.now(); record.updatedAt = Date.now(); - if (code !== 0 && !record.error) + // Only assign a code-based error for actual non-zero exit + // codes (not null signals) and when we don't already have + // a provider error from stdout monitoring. + if (code !== null && code !== 0 && !record.error) record.error = stderr.trim() || `Subagent exited with code ${code}${signal ? ` (${signal})` : ""}`; @@ -1241,7 +1281,11 @@ function startChild( if (!hasExistingResult) { // Subagent did not write to result file — auto-save final output const finalOutput = extractFinalOutput(stdout); - if (finalOutput && record.outputFile) { + // When we have a provider error captured from stdout + // monitoring, use the (error) fallback. + if (record.error && record.outputFile) { + fs.writeFileSync(record.outputFile, "(error)\n", { mode: 0o600 }); + } else if (finalOutput && record.outputFile) { fs.writeFileSync(record.outputFile, `${finalOutput}\n`, { mode: 0o600, }); diff --git a/extensions/pi-subagents/src/runs/shared/pi-args.ts b/extensions/pi-subagents/src/runs/shared/pi-args.ts index 21f964c..98c7cb2 100644 --- a/extensions/pi-subagents/src/runs/shared/pi-args.ts +++ b/extensions/pi-subagents/src/runs/shared/pi-args.ts @@ -65,6 +65,10 @@ export function buildPiArgs(input: BuildPiArgsInput): BuildPiArgsResult { if (input.model) args.push("--model", input.model); + // Use JSON mode so child output is machine-parseable for error + // detection and result extraction. + args.push("--mode", "json"); + // Add only the minimal runtime extension. Do not disable normal extensions, // skills, tools, MCP direct tools, or project context. args.push("--extension", PROMPT_RUNTIME_EXTENSION_PATH); diff --git a/extensions/pi-subagents/test/support/mock-pi-script.mjs b/extensions/pi-subagents/test/support/mock-pi-script.mjs index 67a832e..82aa775 100644 --- a/extensions/pi-subagents/test/support/mock-pi-script.mjs +++ b/extensions/pi-subagents/test/support/mock-pi-script.mjs @@ -39,7 +39,7 @@ function claimNextResponse(dir) { return JSON.parse(fs.readFileSync(defaultPath, "utf-8")); } -function defaultAssistantMessage(output) { +function defaultAssistantMessage(output, overrides = {}) { return { type: "message_end", message: { @@ -54,6 +54,7 @@ function defaultAssistantMessage(output) { cacheWrite: 0, cost: { total: 0.001 }, }, + ...overrides, }, }; } @@ -191,6 +192,14 @@ async function main() { await new Promise((resolve) => setTimeout(resolve, response.delay)); } + const assistantOverrides = {}; + if (typeof response.stopReason === "string") { + assistantOverrides.stopReason = response.stopReason; + } + if (typeof response.errorMessage === "string") { + assistantOverrides.errorMessage = response.errorMessage; + } + if (Array.isArray(response.steps) && response.steps.length > 0) { for (const step of response.steps) { if (typeof step?.delay === "number" && step.delay > 0) { @@ -210,10 +219,10 @@ async function main() { response.echoEnv.map((key) => [key, process.env[key] ?? null]), ); if (jsonMode) - writeJsonlLine(defaultAssistantMessage(JSON.stringify(envSnapshot))); + writeJsonlLine(defaultAssistantMessage(JSON.stringify(envSnapshot), assistantOverrides)); else process.stdout.write(`${JSON.stringify(envSnapshot)}\n`); } else if (typeof response.output === "string") { - if (jsonMode) writeJsonlLine(defaultAssistantMessage(response.output)); + if (jsonMode) writeJsonlLine(defaultAssistantMessage(response.output, assistantOverrides)); else process.stdout.write(`${response.output}\n`); } diff --git a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts index 20704e4..310fa6a 100644 --- a/extensions/pi-subagents/test/unit/minimal-subagents.test.ts +++ b/extensions/pi-subagents/test/unit/minimal-subagents.test.ts @@ -2085,3 +2085,300 @@ test("lifeline: recursive cascade — grandparent death kills parent subagent", try { fs.rmSync(testDir, { recursive: true, force: true }); } catch {} } }); +// ── Bug: model inheritance ── + +test("model inheritance: omitted model inherits parent active model", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ output: "inherited model done", exitCode: 0 }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-model-inherit"); + // Simulate a parent context with an active model + const parentModel = { provider: "openai-codex", id: "gpt-5.6-sol" }; + const ctxWithModel = { + ...ctx, + model: parentModel, + }; + const { spawnTool } = registerTestTools(() => {}); + + try { + const result = await spawnTool.execute( + "inherit-model-child", + { task: "echo model" }, + new AbortController().signal, + undefined, + ctxWithModel, + ); + await waitForPersistedRecord(sessionId, result.details.id); + + // Verify --model was passed to child with the parent's active model + const captured = readLatestMockPiArgs(mockPi).args; + const modelIdx = captured.indexOf("--model"); + assert.ok(modelIdx !== -1, "--model must be passed to child"); + const modelValue = captured[modelIdx + 1]; + assert.equal(modelValue, "openai-codex/gpt-5.6-sol"); + + // Verify record persisted the model + const record = readPersistedRecord(sessionId, result.details.id); + assert.equal(record.model, "openai-codex/gpt-5.6-sol"); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("model inheritance: explicit model overrides inherited model", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ output: "explicit model done", exitCode: 0 }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-model-override"); + const parentModel = { provider: "openai-codex", id: "gpt-5.6-sol" }; + const ctxWithModel = { + ...ctx, + model: parentModel, + }; + const { spawnTool } = registerTestTools(() => {}); + + try { + const result = await spawnTool.execute( + "explicit-model-child", + { task: "echo model", model: "anthropic/claude-sonnet-4-5" }, + new AbortController().signal, + undefined, + ctxWithModel, + ); + await waitForPersistedRecord(sessionId, result.details.id); + + const captured = readLatestMockPiArgs(mockPi).args; + const modelIdx = captured.indexOf("--model"); + assert.ok(modelIdx !== -1); + const modelValue = captured[modelIdx + 1]; + assert.equal(modelValue, "anthropic/claude-sonnet-4-5"); + + const record = readPersistedRecord(sessionId, result.details.id); + assert.equal(record.model, "anthropic/claude-sonnet-4-5"); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +// ── Bug: fatal-error lifecycle ── + +test("error lifecycle: provider error marks child failed promptly", async () => { + const mockPi = createMockPi(); + mockPi.install(); + // Simulate an HTTP 402 payment-required error where child stays alive + mockPi.onCall({ + output: "payment required", + stopReason: "error", + errorMessage: "HTTP 402 Payment Required", + exitCode: 0, + keepAliveAfterFinalMessageMs: 20_000, // would hang for 20s + }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-error-lifecycle"); + const { spawnTool } = registerTestTools(() => {}); + + try { + const result = await spawnTool.execute( + "error-child", + { task: "do work" }, + new AbortController().signal, + undefined, + ctx, + ); + + // Wait up to 10s: 1s error-detection grace + SIGTERM + process exit. + let record: Record | undefined; + for (let i = 0; i < 200; i++) { + record = readPersistedRecord(sessionId, result.details.id); + if (record && !record.running) break; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + assert.ok(record, "record must exist"); + + // Child must be marked as not running + assert.equal(record.running, false, "record must show running=false"); + // completedAt must be set + assert.ok(record.completedAt, "completedAt must be set"); + // Error text must be captured + assert.ok(record.error, "error must be set"); + assert.match(record.error, /HTTP 402 Payment Required/); + // result must be set + assert.ok(record.result, "result must be set (file path)"); + + // Verify result.log has (error) fallback + const resultContent = fs.readFileSync(record.result, "utf-8"); + assert.equal(resultContent.trim(), "(error)"); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("error lifecycle: no duplicate completion notification on provider error", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ + output: "payment required", + stopReason: "error", + errorMessage: "HTTP 402 Payment Required", + exitCode: 0, + }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-error-dedup"); + let notifyCount = 0; + const { spawnTool } = registerTestTools(() => { + notifyCount += 1; + }); + + try { + const result = await spawnTool.execute( + "error-dedup-child", + { task: "do work" }, + new AbortController().signal, + undefined, + ctx, + ); + + await waitForPersistedRecord(sessionId, result.details.id); + assert.equal( + notifyCount, + 1, + "completion notification must fire exactly once", + ); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("error lifecycle: normal successful completion unchanged", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ output: "success output", exitCode: 0 }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-error-normal"); + const { spawnTool } = registerTestTools(() => {}); + + try { + const result = await spawnTool.execute( + "normal-child", + { task: "do work" }, + new AbortController().signal, + undefined, + ctx, + ); + + const record = await waitForPersistedRecord(sessionId, result.details.id); + assert.equal(record.running, false); + assert.equal(record.error, undefined); + assert.ok(record.completedAt); + const resultContent = fs.readFileSync(record.result, "utf-8"); + assert.match(resultContent, /success output/); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("error lifecycle: timeout notification-only unchanged", async () => { + const mockPi = createMockPi(); + mockPi.install(); + // Keep child alive for a long time; timeout must fire but not kill + mockPi.onCall({ + output: "running", + exitCode: 0, + keepAliveAfterFinalMessageMs: 2000, + }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-error-timeout"); + const timeoutMessages: string[] = []; + const { spawnTool } = registerTestTools((message: any) => { + const content = String(message.content ?? ""); + if (content.includes("timed out")) timeoutMessages.push(content); + }); + + try { + const result = await spawnTool.execute( + "timeout-child", + { task: "do work", timeout: 0.1 }, + new AbortController().signal, + undefined, + ctx, + ); + + // Wait for timeout notice + for (let i = 0; i < 100; i++) { + if (timeoutMessages.length > 0) break; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + assert.ok(timeoutMessages.length >= 1, "timeout notice must fire"); + assert.match(timeoutMessages[0], /still running; not killed/); + + // Verify record has timeoutAt but child is still running (not killed) + const runningRecord = readPersistedRecord(sessionId, result.details.id); + assert.equal(runningRecord.running, true, "child must still be running after timeout"); + assert.ok(runningRecord.timeoutAt, "timeoutAt must be set"); + + // Wait for child to actually finish before cleanup + await waitForPersistedRecord(sessionId, result.details.id); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +}); + +test("error lifecycle: lifeline cleanup unchanged after provider error", async () => { + const mockPi = createMockPi(); + mockPi.install(); + mockPi.onCall({ + output: "payment required", + stopReason: "error", + errorMessage: "HTTP 402 Payment Required", + exitCode: 0, + keepAliveAfterFinalMessageMs: 30_000, + }); + + const { sessionId, ctx } = makeTestCtx("pi-subagents-error-lifeline"); + const messages: string[] = []; + const { spawnTool } = registerTestTools((message: any) => { + messages.push(String(message.content ?? "")); + }); + + try { + const result = await spawnTool.execute( + "lifeline-error-child", + { task: "do work" }, + new AbortController().signal, + undefined, + ctx, + ); + + // Wait up to 10s for error detection + kill. + let record: Record | undefined; + for (let i = 0; i < 200; i++) { + record = readPersistedRecord(sessionId, result.details.id); + if (record && !record.running) break; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + assert.ok(record, "record must exist"); + + // Completion notification must fire + const completionMsg = messages.find((m) => + m.includes(record!.id) && m.includes("completed"), + ); + assert.ok(completionMsg, "completion notification must fire"); + + // running must be false + assert.equal(record.running, false); + // pendingCompletionNotice must be false (notification was sent) + assert.equal(record.pendingCompletionNotice, false); + assert.equal(record.notifiedCompletion, true); + } finally { + mockPi.uninstall(); + cleanupTestCtx(ctx, sessionId); + } +});