From e3953135c0fbc00f847fcf4f80988a2412434971 Mon Sep 17 00:00:00 2001 From: cookerpapa Date: Mon, 14 Sep 2026 15:32:15 +0800 Subject: [PATCH] fix(review): preserve numbering and diagnostics after failed attempts --- docs/explanation/review-loop.md | 23 + docs/reference/status-format.md | 2 +- .../taskplane/agent-bridge-extension.ts | 399 ++++++++---------- extensions/taskplane/review-analysis.ts | 12 +- extensions/tests/review-analysis.test.ts | 27 +- .../tests/review-step-guard-runtime.test.ts | 21 +- .../tests/reviewer-failure-runtime.test.ts | 287 +++++++++++++ 7 files changed, 522 insertions(+), 249 deletions(-) create mode 100644 extensions/tests/reviewer-failure-runtime.test.ts diff --git a/docs/explanation/review-loop.md b/docs/explanation/review-loop.md index dcb0e74b..7d3743e5 100644 --- a/docs/explanation/review-loop.md +++ b/docs/explanation/review-loop.md @@ -98,6 +98,29 @@ During a review, the dashboard shows a **reviewer sub-row** below the active task with live metrics: elapsed time, tool count, last tool, cost, and context%. The worker row shows `[awaiting review]` until the reviewer finishes. +### Failed reviewer attempts + +If an inline reviewer exits without a nonempty review file, `review_step` +returns `UNAVAILABLE` and appends a `Review spawn failed` execution-log row +with the exit code, signal, timeout, or launch error. The attempt does not +increment `Review Counter`; a retry uses the same `R00N` filename. An empty +file is removed so it cannot supersede a previous review at the same gate. +Nonempty reviews retain their number even when the verdict is unclear; they +must still contain an explicit approval before the worker can proceed. + +Each subprocess attempt has separate event (`.jsonl`), exit-summary +(`-exit.json`), and stderr-tail (`-stderr.log`, up to 2 KB) files under +`.pi/runtime//agents/` in the project state root. Failed attempts' +files survive a successful retry. Launch errors before the subprocess starts +still receive an exit summary. The tool response and STATUS log identify +these diagnostic paths. + +Review rounds count `APPROVE`, `REVISE`, and `RETHINK` outcomes, both live +and after resume. `REFUSED`, `UNAVAILABLE`, and `UNKNOWN` do not advance the +round. The explicit `treatUnavailableAsNonApprove` setting still controls +whether `UNAVAILABLE` increases the non-approval streak; it does not turn a +failed attempt into a completed review round. + ### Review availability The `review_step` tool is registered during orchestrated execution (`/orch`). diff --git a/docs/reference/status-format.md b/docs/reference/status-format.md index 7766a087..babdb586 100644 --- a/docs/reference/status-format.md +++ b/docs/reference/status-format.md @@ -30,7 +30,7 @@ Common fields at top of file: | `Status` | Overall task status (Ready, In Progress, Complete, etc.) | | `Last Updated` | Last update date | | `Review Level` | Task review level from prompt | -| `Review Counter` | Number of reviews run so far | +| `Review Counter` | Last allocated review number; attempts producing no output do not increment it | | `Iteration` | Worker iteration counter | | `Size` | Task size metadata | diff --git a/extensions/taskplane/agent-bridge-extension.ts b/extensions/taskplane/agent-bridge-extension.ts index cf3be030..c485d7b1 100644 --- a/extensions/taskplane/agent-bridge-extension.ts +++ b/extensions/taskplane/agent-bridge-extension.ts @@ -33,12 +33,12 @@ import { unlinkSync, } from "fs"; import { join, dirname } from "path"; -import { spawn as nodeSpawn } from "child_process"; -import { resolvePiCliPath, resolveTaskplaneAgentTemplate } from "./path-resolver.ts"; +import { resolveTaskplaneAgentTemplate } from "./path-resolver.ts"; import { loadPiSettingsPackages, filterExcludedExtensions } from "./settings-loader.ts"; import { randomBytes } from "crypto"; import { buildExpansionRequestId, type SegmentExpansionRequest } from "./types.ts"; import { latestReviewFilesPerGate, parseReviewVerdict } from "./review-analysis.ts"; +import { spawnAgent, type AgentHostResult } from "./agent-host.ts"; /** * Resolve the outbox directory from environment variables. @@ -612,212 +612,108 @@ export default function (pi: ExtensionAPI) { /** * Spawn a reviewer Pi subprocess and wait for it to complete. - * Returns the process exit code. + * Returns the exit result and timeout cause; persists attempt diagnostics. */ - function spawnReviewer( + async function spawnReviewer( prompt: string, systemPrompt: string, cwd: string, taskFolder: string, + diagnosticBase: string, reviewType?: string, reviewStep?: number, - ): Promise { - // Pre-clean stale reviewer state from prior interrupted review + ): Promise { removeReviewerState(taskFolder); - return new Promise((resolve) => { - // Read reviewer config from env vars set by lane-runner from runnerConfig.reviewer. - // Empty string means inherit from session default (no flag passed to pi CLI). - const reviewerModel = process.env.TASKPLANE_REVIEWER_MODEL || ""; - const reviewerThinking = process.env.TASKPLANE_REVIEWER_THINKING || ""; - // Fall back to the schema default reviewer tool list (read-only + bash/grep). - // Must match config-schema.ts reviewer.tools default to avoid capability expansion. - const reviewerTools = process.env.TASKPLANE_REVIEWER_TOOLS || "read,bash,grep,find,ls"; - - const cliPath = resolvePiCliPath(); - const args = [ - cliPath, - "--mode", - "rpc", - "--no-session", - "--no-extensions", - "--no-skills", - "--tools", - reviewerTools, - "--system-prompt", - systemPrompt, - ]; - if (reviewerModel) args.push("--model", reviewerModel); - if (reviewerThinking) args.push("--thinking", reviewerThinking); + const settingsRoot = process.env.TASKPLANE_STATE_ROOT || cwd; + const reviewerPackages = loadPiSettingsPackages(settingsRoot); + let reviewerExclusions: string[] = []; + try { + const rawExclude = process.env.TASKPLANE_REVIEWER_EXCLUDE_EXTENSIONS; + if (rawExclude) { + const parsed = JSON.parse(rawExclude); + if (Array.isArray(parsed)) { + reviewerExclusions = parsed.filter((v: unknown): v is string => typeof v === "string"); + } + } + } catch { + /* ignore malformed */ + } - // TP-180: Forward user-installed extensions to reviewer agent - // Use TASKPLANE_STATE_ROOT (canonical project root) for settings resolution, - // falling back to cwd (which may be a worktree without .pi/settings.json). - const settingsRoot = process.env.TASKPLANE_STATE_ROOT || cwd; - const reviewerPackages = loadPiSettingsPackages(settingsRoot); - // Apply reviewer-specific exclusions from config (JSON array via env) - let reviewerExclusions: string[] = []; + const startedAt = Date.now(); + const telemetry = { + toolCalls: 0, + contextPct: 0, + costUsd: 0, + lastTool: "", + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheWriteTokens: 0, + }; + const emitState = (status: "running" | "done" | "error") => { try { - const rawExclude = process.env.TASKPLANE_REVIEWER_EXCLUDE_EXTENSIONS; - if (rawExclude) { - const parsed = JSON.parse(rawExclude); - if (Array.isArray(parsed)) { - reviewerExclusions = parsed.filter((v: unknown): v is string => typeof v === "string"); - } - } + writeReviewerState(taskFolder, { + ...telemetry, + status, + elapsedMs: Date.now() - startedAt, + updatedAt: Date.now(), + reviewType, + reviewStep, + }); } catch { - /* ignore malformed */ - } - const filteredReviewerPackages = filterExcludedExtensions(reviewerPackages, reviewerExclusions); - for (const pkg of filteredReviewerPackages) { - args.push("-e", pkg); + /* best effort */ } - const proc = nodeSpawn(process.execPath, args, { - shell: false, + }; + emitState("running"); + let timedOut = false; + const host = spawnAgent( + { + agentId: `${process.env.TASKPLANE_AGENT_ID || "worker"}-reviewer`, + role: "reviewer", + batchId: process.env.ORCH_BATCH_ID || "standalone", + laneNumber: null, + taskId: process.env.TASKPLANE_TASK_ID || null, + repoId: "", cwd, - stdio: ["pipe", "pipe", "pipe"], - env: { ...process.env }, - }); - - const startedAt = Date.now(); - let inputTokens = 0; - let outputTokens = 0; - let cacheReadTokens = 0; - let cacheWriteTokens = 0; - let costUsd = 0; - let toolCalls = 0; - let lastTool = ""; - let contextPct = 0; - let stdoutBuf = ""; - let finalized = false; - - const emitState = (status: "running" | "done" | "error") => { - try { - writeReviewerState(taskFolder, { - status, - elapsedMs: Date.now() - startedAt, - toolCalls, - contextPct, - costUsd, - lastTool, - inputTokens, - outputTokens, - cacheReadTokens, - cacheWriteTokens, - updatedAt: Date.now(), - reviewType, - reviewStep, - }); - } catch { - /* best effort */ - } - }; - - // Write initial "running" state immediately so dashboard shows - // the reviewer sub-row before the first message_end arrives. - emitState("running"); - - const closeStdin = () => { - setTimeout(() => { - try { - proc.stdin?.end(); - } catch { - /* ignore */ - } - }, 100); - }; - - const finalize = (code: number) => { - if (finalized) return; - finalized = true; - emitState(code === 0 ? "done" : "error"); - resolve(code); - }; - - const handleEvent = (event: any) => { - if (!event || typeof event.type !== "string") return; - switch (event.type) { - case "message_end": { - const usage = event.message?.usage; - if (usage) { - inputTokens += usage.input || 0; - outputTokens += usage.output || 0; - cacheReadTokens += usage.cacheRead || 0; - cacheWriteTokens += usage.cacheWrite || 0; - if (usage.cost) { - costUsd += - typeof usage.cost === "object" - ? usage.cost.total || 0 - : typeof usage.cost === "number" - ? usage.cost - : 0; - } - } - emitState("running"); - break; - } - case "tool_execution_start": { - toolCalls++; - const toolName = event.toolName || "tool"; - const argPreview = - typeof event.args === "string" - ? event.args.slice(0, 80) - : event.args && typeof Object.values(event.args)[0] === "string" - ? String(Object.values(event.args)[0]).slice(0, 80) - : ""; - lastTool = argPreview ? `${toolName}: ${argPreview}` : toolName; - emitState("running"); - break; - } - case "response": { - const pct = event.success === true ? event.data?.contextUsage?.percent : undefined; - if (typeof pct === "number" && Number.isFinite(pct)) { - contextPct = pct; - } - break; - } - case "agent_end": { - closeStdin(); - break; - } - } - }; - - // Send prompt immediately - proc.stdin?.write(JSON.stringify({ type: "prompt", message: prompt }) + "\n"); - - proc.stdout?.on("data", (chunk: Buffer | string) => { - stdoutBuf += typeof chunk === "string" ? chunk : chunk.toString("utf-8"); - let idx = -1; - while ((idx = stdoutBuf.indexOf("\n")) >= 0) { - let line = stdoutBuf.slice(0, idx); - stdoutBuf = stdoutBuf.slice(idx + 1); - if (line.endsWith("\r")) line = line.slice(0, -1); - if (!line.trim()) continue; - let event: any; - try { - event = JSON.parse(line); - } catch { - continue; - } - handleEvent(event); + prompt, + systemPrompt, + model: process.env.TASKPLANE_REVIEWER_MODEL || "", + thinking: process.env.TASKPLANE_REVIEWER_THINKING || "", + // Match the schema's reviewer allowlist, including when no override is set. + tools: process.env.TASKPLANE_REVIEWER_TOOLS || "read,bash,grep,find,ls", + extensions: filterExcludedExtensions(reviewerPackages, reviewerExclusions), + eventsPath: `${diagnosticBase}.jsonl`, + exitSummaryPath: `${diagnosticBase}-exit.json`, + timeoutMs: 10 * 60 * 1000, + }, + (event) => { + if (event.type === "agent_timeout") timedOut = true; + if (event.type === "tool_call") { + telemetry.toolCalls++; + const tool = String(event.payload?.tool || "tool"); + const preview = String(event.payload?.argsPreview || "").slice(0, 80); + telemetry.lastTool = preview ? `${tool}: ${preview}` : tool; + emitState("running"); } - }); - - proc.on("close", (code) => finalize(code ?? 1)); - proc.on("error", () => finalize(1)); - - // Timeout: 10 minutes - setTimeout( - () => { - try { - proc.kill("SIGTERM"); - } catch { - /* ignore */ - } - }, - 10 * 60 * 1000, - ); - }); + }, + (update) => { + telemetry.inputTokens = update.inputTokens ?? telemetry.inputTokens; + telemetry.outputTokens = update.outputTokens ?? telemetry.outputTokens; + telemetry.cacheReadTokens = update.cacheReadTokens ?? telemetry.cacheReadTokens; + telemetry.cacheWriteTokens = update.cacheWriteTokens ?? telemetry.cacheWriteTokens; + telemetry.costUsd = update.costUsd ?? telemetry.costUsd; + telemetry.contextPct = update.contextUsage?.percent ?? telemetry.contextPct; + emitState("running"); + }, + ); + const result = await host.promise; + try { + writeFileSync(`${diagnosticBase}-stderr.log`, result.stderrTail, "utf-8"); + } catch { + /* best effort */ + } + emitState(result.exitCode === 0 && !timedOut ? "done" : "error"); + return { ...result, timedOut }; } pi.registerTool({ @@ -971,32 +867,87 @@ export default function (pi: ExtensionAPI) { ].join("\n"); } + // Attempts get unique diagnostic files even when a retry reuses R00N. + const diagnosticBase = join( + process.env.TASKPLANE_STATE_ROOT || cwd, + ".pi", + "runtime", + process.env.ORCH_BATCH_ID || "standalone", + "agents", + `${process.env.TASKPLANE_AGENT_ID || "worker"}-reviewer-${Date.now()}-${randomBytes(3).toString("hex")}`, + ); + const failedReview = (reason: string) => { + // Setup errors may occur before the host has created its diagnostic files. + try { + mkdirSync(dirname(diagnosticBase), { recursive: true }); + if (!existsSync(`${diagnosticBase}-exit.json`)) { + writeFileSync( + `${diagnosticBase}-exit.json`, + JSON.stringify( + { + exitCode: null, + exitSignal: null, + error: reason, + }, + null, + 2, + ) + "\n", + ); + } + if (!existsSync(`${diagnosticBase}-stderr.log`)) + writeFileSync(`${diagnosticBase}-stderr.log`, ""); + } catch { + /* best effort */ + } + try { + const status = readFileSync(statusPath, "utf-8"); + const detail = + `${reviewType} Step ${stepNum}: ${reason}; diagnostics: ${diagnosticBase}`.replace( + /[|\r\n]/g, + " ", + ); + const logEntry = `| ${new Date().toISOString().slice(0, 16).replace("T", " ")} | Review spawn failed | ${detail} |\n`; + writeFileSync(statusPath, status.trimEnd() + "\n" + logEntry); + } catch { + /* best effort */ + } + removeReviewerState(taskFolder); + return { + content: [ + { + type: "text" as const, + text: `UNAVAILABLE — reviewer failed: ${reason}. Exit summary: ${diagnosticBase}-exit.json. Stderr: ${diagnosticBase}-stderr.log.`, + }, + ], + details: undefined, + }; + }; + try { const systemPrompt = loadReviewerPrompt(); - const exitCode = await spawnReviewer( + const result = await spawnReviewer( reviewPrompt, systemPrompt, cwd, taskFolder, + diagnosticBase, reviewType, stepNum, ); - // Update review counter in STATUS.md - try { - const status = readFileSync(statusPath, "utf-8"); - const updated = status.replace( - /\*\*Review Counter:\*\*\s*\d+/, - `**Review Counter:** ${reviewCounter}`, - ); - writeFileSync(statusPath, updated); - } catch { - /* best effort */ - } - - // Read review output and extract verdict - if (existsSync(outputPath)) { - const reviewContent = readFileSync(outputPath, "utf-8"); + const reviewContent = existsSync(outputPath) ? readFileSync(outputPath, "utf-8") : ""; + if (reviewContent.trim()) { + // Only consume a number once the reviewer has produced an artifact. + try { + const status = readFileSync(statusPath, "utf-8"); + const updated = status.replace( + /\*\*Review Counter:\*\*\s*\d+/, + `**Review Counter:** ${reviewCounter}`, + ); + writeFileSync(statusPath, updated); + } catch { + /* best effort */ + } // #624 (severity upgrade): robust, FAIL-CLOSED verdict extraction. The // old regex ('###?\s*Verdict[:\s]*…') missed common reviewer format // variants, and its fallback checked the substring "approve" FIRST — so @@ -1061,28 +1012,16 @@ export default function (pi: ExtensionAPI) { }; } } else { - removeReviewerState(taskFolder); - return { - content: [ - { - type: "text" as const, - text: `UNAVAILABLE — reviewer exited (code ${exitCode}) but produced no output.`, - }, - ], - details: undefined, - }; + // Empty files must not become the latest artifact for a review gate. + if (existsSync(outputPath)) unlinkSync(outputPath); + const cause = result.timedOut + ? "timeout after 10 minutes" + : result.error || + (result.signal ? `signal ${result.signal}` : `exit code ${result.exitCode}`); + return failedReview(`${cause}; no review output`); } } catch (err) { - removeReviewerState(taskFolder); - return { - content: [ - { - type: "text" as const, - text: `UNAVAILABLE — reviewer failed: ${err instanceof Error ? err.message : String(err)}`, - }, - ], - details: undefined, - }; + return failedReview(err instanceof Error ? err.message : String(err)); } }, }); diff --git a/extensions/taskplane/review-analysis.ts b/extensions/taskplane/review-analysis.ts index a86afdeb..151ceb20 100644 --- a/extensions/taskplane/review-analysis.ts +++ b/extensions/taskplane/review-analysis.ts @@ -170,7 +170,7 @@ export function computeFindingTrend( export interface ReviewStreakState { /** Consecutive REVISE/RETHINK reviews on this step (reset on APPROVE). */ consecutiveNonApprove: number; - /** Count of verdict/attempt reviews seen for this step (the review round). */ + /** Count of APPROVE/REVISE/RETHINK verdicts seen for this step (the review round). */ round: number; /** Finding counts from the previous round (for trend), or null. */ lastCounts: Record | null; @@ -186,7 +186,7 @@ export function freshReviewStreakState(): ReviewStreakState { /** * Apply ONE review-boundary outcome to a step's streak state (mutates it). This * is the single source of truth for the counter transitions: - * - every END boundary increments `round`; + * - only APPROVE/REVISE/RETHINK boundaries increment `round`; * - APPROVE resets the consecutive streak to 0; * - REVISE/RETHINK (and UNAVAILABLE iff `treatUnavailableAsNonApprove`) * increment the streak; @@ -207,7 +207,13 @@ export function advanceReviewStreak( recentCap: number; }, ): void { - state.round += 1; + if ( + opts.disposition === "APPROVE" || + opts.disposition === "REVISE" || + opts.disposition === "RETHINK" + ) { + state.round += 1; + } if (opts.counts && Object.keys(opts.counts).length > 0) { state.lastCounts = opts.counts; } diff --git a/extensions/tests/review-analysis.test.ts b/extensions/tests/review-analysis.test.ts index 240b13d2..507fffe3 100644 --- a/extensions/tests/review-analysis.test.ts +++ b/extensions/tests/review-analysis.test.ts @@ -29,7 +29,7 @@ import { const SPIRAL = { enabled: true, threshold: 3, cooldownReviews: 2 }; const adv = ( state: ReturnType, - disposition: string, + disposition: string | undefined, counts: Record | null = null, ) => advanceReviewStreak(state, { @@ -169,7 +169,7 @@ describe("review-analysis — computeFindingTrend", () => { }); describe("review-analysis — advanceReviewStreak", () => { - it("increments round every boundary and the streak on REVISE/RETHINK", () => { + it("increments round on verdicts and the streak on REVISE/RETHINK", () => { const s = freshReviewStreakState(); adv(s, "REVISE"); adv(s, "RETHINK"); @@ -186,20 +186,21 @@ describe("review-analysis — advanceReviewStreak", () => { expect(s.round).toBe(3); }); - it("REFUSED advances round but does NOT touch the streak", () => { + it("REFUSED leaves both the round and streak unchanged", () => { const s = freshReviewStreakState(); adv(s, "REVISE"); adv(s, "REFUSED"); expect(s.consecutiveNonApprove).toBe(1); - expect(s.round).toBe(2); + expect(s.round).toBe(1); }); it("UNAVAILABLE does not count by default; UNKNOWN never counts", () => { const s = freshReviewStreakState(); adv(s, "UNAVAILABLE"); adv(s, "UNKNOWN"); + adv(s, undefined); expect(s.consecutiveNonApprove).toBe(0); - expect(s.round).toBe(2); + expect(s.round).toBe(0); }); it("UNAVAILABLE counts when treatUnavailableAsNonApprove is true", () => { @@ -211,6 +212,7 @@ describe("review-analysis — advanceReviewStreak", () => { recentCap: 6, }); expect(s.consecutiveNonApprove).toBe(1); + expect(s.round).toBe(0); }); it("advances lastCounts only when counts are present, and bounds recentDispositions", () => { @@ -224,6 +226,21 @@ describe("review-analysis — advanceReviewStreak", () => { }); describe("review-analysis — reconstructReviewStreaks (resume)", () => { + it("resumes at round one after failed attempts followed by the first verdict", () => { + const events = [ + { reviewStep: 5, disposition: "UNAVAILABLE" }, + { reviewStep: 5, disposition: "UNKNOWN" }, + { reviewStep: 5, disposition: "REFUSED" }, + { reviewStep: 5, disposition: "REVISE" }, + ]; + const states = reconstructReviewStreaks(events, { + treatUnavailableAsNonApprove: false, + recentCap: 6, + }); + expect(states.get("5")?.round).toBe(1); + expect(states.get("5")?.consecutiveNonApprove).toBe(1); + }); + it("replays per-step history so an in-progress spiral survives resume", () => { const events = [ { reviewStep: 4, disposition: "REVISE", findingCounts: { critical: 2 } }, diff --git a/extensions/tests/review-step-guard-runtime.test.ts b/extensions/tests/review-step-guard-runtime.test.ts index 5e32c9e4..1a369afc 100644 --- a/extensions/tests/review-step-guard-runtime.test.ts +++ b/extensions/tests/review-step-guard-runtime.test.ts @@ -32,12 +32,13 @@ import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; // ── child_process mock (installed before importing agent-bridge-extension) ── // // The review_step handler spawns a Pi reviewer subprocess via -// `nodeSpawn(process.execPath, args, ...)`. We intercept that with a fake -// EventEmitter-shaped child process that immediately emits exit(0). The +// the shared agent host. We intercept that with a fake +// EventEmitter-shaped child process that immediately emits close(0). The // REFUSED path returns BEFORE reaching spawn — the mock exists so the // plan-NOT-blocked sanity check doesn't fork a real Pi process. // @@ -51,17 +52,17 @@ let spawnCallCount = 0; const mockSpawn = mock.fn((_cmd: string, _args: readonly string[], _opts: object) => { spawnCallCount++; const fake = new EventEmitter() as EventEmitter & { - stdout: EventEmitter; - stderr: EventEmitter; - stdin: { end: () => void }; + stdout: PassThrough; + stderr: PassThrough; + stdin: PassThrough; kill: (sig?: string) => boolean; }; - fake.stdout = new EventEmitter(); - fake.stderr = new EventEmitter(); - fake.stdin = { end: () => {} }; + fake.stdout = new PassThrough(); + fake.stderr = new PassThrough(); + fake.stdin = new PassThrough(); fake.kill = () => true; - // Emit exit on the next tick so the handler's listeners attach first. - setImmediate(() => fake.emit("exit", 0, null)); + // Emit close on the next tick so the handler's listeners attach first. + setImmediate(() => fake.emit("close", 0, null)); return fake; }); diff --git a/extensions/tests/reviewer-failure-runtime.test.ts b/extensions/tests/reviewer-failure-runtime.test.ts new file mode 100644 index 00000000..c842a1ce --- /dev/null +++ b/extensions/tests/reviewer-failure-runtime.test.ts @@ -0,0 +1,287 @@ +import { afterEach, beforeEach, describe, it, mock } from "node:test"; +import { strict as assert } from "node:assert"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync, existsSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { advanceReviewStreak, freshReviewStreakState } from "../taskplane/review-analysis.ts"; + +interface Attempt { + code: number | null; + signal?: string; + stderr?: string; + output?: string; + error?: string; + hang?: boolean; + throwOnSpawn?: boolean; + rpc?: object[]; +} + +const realChildProcess = await import("node:child_process"); +const attempts: Attempt[] = []; +const outputPaths: string[] = []; +let killCalls = 0; +const launchArgs: string[][] = []; +const reviewerStates: object[] = []; +mock.module("child_process", { + namedExports: { + ...realChildProcess, + spawn: (_cmd: string, args: string[]) => { + launchArgs.push(args); + const attempt = attempts.shift(); + assert.ok(attempt, "unexpected reviewer spawn"); + if (attempt.throwOnSpawn) throw new Error("synchronous launch failure"); + const proc = Object.assign(new EventEmitter(), { + stdin: new PassThrough(), + stdout: new PassThrough(), + stderr: new PassThrough(), + kill: (signal: string) => { + killCalls++; + queueMicrotask(() => proc.emit("close", null, signal)); + return true; + }, + }); + proc.stdin.on("data", (data) => { + const message = JSON.parse(data.toString()); + if (message.type !== "prompt") return; + const outputPath = message.message.match(/Write your review to: `([^`]+)`/)?.[1]; + assert.ok(outputPath); + outputPaths.push(outputPath); + queueMicrotask(() => { + if (attempt.stderr) proc.stderr.write(attempt.stderr); + for (const event of attempt.rpc ?? []) proc.stdout.write(JSON.stringify(event) + "\n"); + if (attempt.rpc) + reviewerStates.push( + JSON.parse(readFileSync(process.env.TASKPLANE_REVIEWER_STATE_PATH!, "utf-8")), + ); + if (attempt.output !== undefined) writeFileSync(outputPath, attempt.output); + if (attempt.error) proc.emit("error", new Error(attempt.error)); + else if (!attempt.hang) proc.emit("close", attempt.code, attempt.signal ?? null); + }); + }); + return proc; + }, + }, +}); + +const bridgeExtension = (await import("../taskplane/agent-bridge-extension.ts")).default; +const ENV_KEYS = [ + "TASKPLANE_TASK_FOLDER", + "TASKPLANE_STATUS_PATH", + "TASKPLANE_PROMPT_PATH", + "TASKPLANE_REVIEWS_DIR", + "TASKPLANE_REVIEWER_STATE_PATH", + "TASKPLANE_STATE_ROOT", + "TASKPLANE_AGENT_ID", + "TASKPLANE_TASK_ID", + "ORCH_BATCH_ID", + "TASKPLANE_REVIEWER_MODEL", + "TASKPLANE_REVIEWER_THINKING", + "TASKPLANE_REVIEWER_TOOLS", +] as const; +let previousEnv: Record; +let root: string; +let statusPath: string; +let reviewsDir: string; +let agentsDir: string; +let previousArgv: string; + +beforeEach(() => { + mock.timers.enable({ apis: ["setTimeout"] }); + attempts.length = 0; + outputPaths.length = 0; + launchArgs.length = 0; + reviewerStates.length = 0; + killCalls = 0; + root = mkdtempSync(join(tmpdir(), "reviewer-failure-")); + previousArgv = process.argv[1]; + process.argv[1] = join(root, "cli.js"); + writeFileSync(process.argv[1], "// mocked Pi CLI\n"); + statusPath = join(root, "STATUS.md"); + reviewsDir = join(root, ".reviews"); + agentsDir = join(root, ".pi", "runtime", "batch", "agents"); + previousEnv = Object.fromEntries(ENV_KEYS.map((key) => [key, process.env[key]])); + Object.assign(process.env, { + TASKPLANE_TASK_FOLDER: root, + TASKPLANE_STATUS_PATH: statusPath, + TASKPLANE_PROMPT_PATH: join(root, "PROMPT.md"), + TASKPLANE_REVIEWS_DIR: reviewsDir, + TASKPLANE_REVIEWER_STATE_PATH: join(root, ".reviewer-state.json"), + TASKPLANE_STATE_ROOT: root, + TASKPLANE_AGENT_ID: "worker-1", + TASKPLANE_TASK_ID: "TP-1", + ORCH_BATCH_ID: "batch", + }); + writeFileSync( + statusPath, + "**Review Counter:** 2\n\n### Step 5: Implement\n**Status:** 🟨 In Progress\n", + ); + writeFileSync(join(root, "PROMPT.md"), "### Step 5: Implement\n"); +}); + +afterEach(() => { + mock.timers.reset(); + process.argv[1] = previousArgv; + for (const key of ENV_KEYS) { + if (previousEnv[key] === undefined) delete process.env[key]; + else process.env[key] = previousEnv[key]; + } + rmSync(root, { recursive: true, force: true }); +}); + +function reviewTool() { + let execute: (id: string, params: object) => Promise<{ content: { text: string }[] }>; + bridgeExtension({ + registerTool(tool: { name: string; execute: typeof execute }) { + if (tool.name === "review_step") execute = tool.execute; + }, + } as never); + return () => execute("review-call", { step: 5, type: "code" }); +} + +describe("reviewer subprocess failures", () => { + for (const [name, output] of [ + ["missing", undefined], + ["empty", " \n"], + ] as const) { + it(`reuses the review number after ${name} output and reports the retry as round one`, async () => { + attempts.push({ code: 1, output }, { code: 0, output: "## Verdict: REVISE\n" }); + const review = reviewTool(); + const failed = await review(); + assert.match(failed.content[0].text, /^UNAVAILABLE/); + assert.match(readFileSync(statusPath, "utf-8"), /\*\*Review Counter:\*\* 2\b/); + assert.equal(existsSync(outputPaths[0]), false); + const failedDiagnostics = readdirSync(agentsDir).map((file) => [ + file, + readFileSync(join(agentsDir, file), "utf-8"), + ]); + const success = await review(); + for (const [file, content] of failedDiagnostics) + assert.equal(readFileSync(join(agentsDir, file), "utf-8"), content); + assert.equal(readdirSync(agentsDir).filter((file) => file.endsWith("-exit.json")).length, 2); + assert.match(success.content[0].text, /^REVISE/); + assert.equal(outputPaths[0], outputPaths[1]); + assert.match(outputPaths[1], /R003-code-step5\.md$/); + assert.match(readFileSync(statusPath, "utf-8"), /\*\*Review Counter:\*\* 3\b/); + const state = freshReviewStreakState(); + for (const disposition of ["UNAVAILABLE", "REVISE"] as const) { + advanceReviewStreak(state, { + disposition, + counts: null, + treatUnavailableAsNonApprove: false, + recentCap: 6, + }); + } + assert.equal(state.round, 1); + assert.equal(state.consecutiveNonApprove, 1); + }); + } + + for (const attempt of [ + { code: 1, stderr: "provider unavailable" }, + { code: null, signal: "SIGKILL", stderr: "terminated" }, + { code: null, error: "spawn ENOENT" }, + ] satisfies Attempt[]) { + it(`persists diagnostics for ${attempt.error ?? attempt.signal ?? "exit 1"}`, async () => { + attempts.push(attempt); + const result = await reviewTool()(); + assert.match(result.content[0].text, /^UNAVAILABLE/); + const files = readdirSync(agentsDir); + const summaryFile = files.find((name) => name.endsWith("-exit.json")); + assert.ok(summaryFile); + const summary = JSON.parse(readFileSync(join(agentsDir, summaryFile), "utf-8")); + assert.equal(summary.exitCode, attempt.code); + assert.equal(summary.exitSignal, attempt.signal ?? null); + if (attempt.error) assert.match(summary.error, /spawn ENOENT/); + const stderrFile = files.find((name) => name.endsWith("-stderr.log")); + assert.ok(stderrFile); + assert.equal(readFileSync(join(agentsDir, stderrFile), "utf-8").trim(), attempt.stderr ?? ""); + assert.ok(files.some((name) => name.endsWith(".jsonl"))); + const status = readFileSync(statusPath, "utf-8"); + assert.match(status, /Review spawn failed/); + assert.ok(status.includes(attempt.error ?? attempt.signal ?? "code 1")); + assert.equal(existsSync(join(root, ".reviewer-state.json")), false); + }); + } + + it("records synchronous launch errors without consuming a number", async () => { + attempts.push({ code: null, throwOnSpawn: true }); + const result = await reviewTool()(); + assert.match(result.content[0].text, /^UNAVAILABLE.*synchronous launch failure/); + const summary = readdirSync(agentsDir).find((file) => file.endsWith("-exit.json"))!; + assert.match(readFileSync(join(agentsDir, summary), "utf-8"), /synchronous launch failure/); + assert.match( + readFileSync(statusPath, "utf-8"), + /Review spawn failed.*synchronous launch failure/, + ); + assert.match(readFileSync(statusPath, "utf-8"), /\*\*Review Counter:\*\* 2\b/); + assert.equal(existsSync(join(root, ".reviewer-state.json")), false); + }); + + it("keeps nonempty unclear reviews fail-closed and preserves their artifact number", async () => { + attempts.push({ code: 0, output: "Review interrupted before a verdict.\n" }); + const result = await reviewTool()(); + assert.match(result.content[0].text, /verdict unclear/); + assert.match(result.content[0].text, /do NOT treat this as an approval/); + assert.match(readFileSync(statusPath, "utf-8"), /\*\*Review Counter:\*\* 3\b/); + assert.equal(readFileSync(outputPaths[0], "utf-8"), "Review interrupted before a verdict.\n"); + }); + + it("preserves reviewer configuration and dashboard telemetry through the host", async () => { + process.env.TASKPLANE_REVIEWER_MODEL = "test/reviewer"; + process.env.TASKPLANE_REVIEWER_THINKING = "high"; + process.env.TASKPLANE_REVIEWER_TOOLS = "read,grep"; + attempts.push({ + code: 0, + output: "## Verdict: APPROVE\n", + rpc: [ + { type: "tool_execution_start", toolName: "read", args: { path: "src/main.ts" } }, + { + type: "message_end", + message: { role: "assistant", usage: { input: 12, output: 3, cost: { total: 0.01 } } }, + }, + ], + }); + const result = await reviewTool()(); + assert.equal(result.content[0].text, "APPROVE"); + const args = launchArgs[0]; + for (const [flag, value] of [ + ["--model", "test/reviewer"], + ["--thinking", "high"], + ["--tools", "read,grep"], + ]) { + assert.equal(args[args.indexOf(flag) + 1], value); + } + assert.ok(args.includes("--no-extensions")); + assert.ok(args.includes("--no-skills")); + assert.ok(args.includes("--system-prompt")); + assert.ok( + reviewerStates.some( + (state: any) => + state.toolCalls === 1 && + state.lastTool === "read: src/main.ts" && + state.inputTokens === 12 && + state.outputTokens === 3 && + state.costUsd === 0.01, + ), + ); + }); + + it("records timeout as the cause of a missing review", async () => { + attempts.push({ code: null, hang: true }); + const pending = reviewTool()(); + mock.timers.tick(10 * 60 * 1000); + const result = await pending; + assert.match(result.content[0].text, /timeout/i); + assert.match(readFileSync(statusPath, "utf-8"), /Review spawn failed.*timeout/i); + }); + + it("retains a completed verdict after a nonzero exit and clears the timeout", async () => { + attempts.push({ code: 1, output: "## Verdict: REVISE\n" }); + const result = await reviewTool()(); + assert.match(result.content[0].text, /^REVISE/); + mock.timers.tick(10 * 60 * 1000); + assert.equal(killCalls, 0); + }); +});