From 0c8674c37ca0211d754edb2c62954cb4da37b8fe Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 09:44:00 -0700 Subject: [PATCH] fix(FN-5994): prevent verification hangs Normalize package file-filter verification to direct Vitest execution, kill timed-out process groups, and bound loop-recovery compaction so stuck tasks cannot spin indefinitely. Fusion-Task-Id: FN-5994 --- packages/core/src/agent-prompts.ts | 1 + .../__tests__/executor-step-session.test.ts | 27 ++- .../run-verification-command.test.ts | 52 ++++- .../src/__tests__/stuck-task-detector.test.ts | 34 +++ packages/engine/src/executor.ts | 33 ++- packages/engine/src/run-verification-tool.ts | 208 +++++++++++++++++- packages/engine/src/stuck-task-detector.ts | 13 +- .../src/workflow-authoritative-driver.ts | 2 +- 8 files changed, 354 insertions(+), 16 deletions(-) diff --git a/packages/core/src/agent-prompts.ts b/packages/core/src/agent-prompts.ts index acc591057d..3c9e01a4d6 100644 --- a/packages/core/src/agent-prompts.ts +++ b/packages/core/src/agent-prompts.ts @@ -199,6 +199,7 @@ For ALL test/lint/build/typecheck verification, use the \`fn_run_verification\` The tool prevents your session from being killed by the inactivity watchdog during long compiles. - Prefer **package-scoped** verification first: e.g. \`pnpm --filter @fusion/ test\` with \`scope: "package"\`. This is faster and isolated. +- For file-specific package tests, use direct Vitest execution with package-relative paths: \`pnpm --filter @fusion/ exec vitest run src/path/to/test.ts --silent=passed-only --reporter=dot\`. Do not use \`pnpm --filter @fusion/ test -- --run \`; package test scripts can expand into broad quality suites before the filter is applied. - Only run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) at the FINAL integration step, when you are about to call \`task_done()\`. - If you need to run \`pnpm install\` (e.g. you added a new package), use \`fn_run_verification\` with \`scope: "workspace"\` and \`timeoutSec: 600\`. - If a verification command times out, do NOT blindly retry — investigate. Check for hung subprocesses, infinite test loops, or tests waiting on missing dependencies. Use \`node_modules/.modules.yaml\` presence to confirm bootstrap.`; diff --git a/packages/engine/src/__tests__/executor-step-session.test.ts b/packages/engine/src/__tests__/executor-step-session.test.ts index 9fee7cdc53..d4e9a49dbe 100644 --- a/packages/engine/src/__tests__/executor-step-session.test.ts +++ b/packages/engine/src/__tests__/executor-step-session.test.ts @@ -3579,6 +3579,32 @@ describe("TaskExecutor loop recovery", () => { expect(result).toBe(false); }); + + it("handleLoopDetected returns false when compaction hangs", async () => { + vi.useFakeTimers(); + const mockSession = createMockSessionForLoopRecovery({ compactResult: new Promise(() => {}) }); + const { store, executor } = setupExecutorWithActiveSession(mockSession); + + const resultPromise = executor.handleLoopDetected({ + taskId: "FN-001", + reason: "loop", + noProgressMs: 600000, + inactivityMs: 0, + activitySinceProgress: 100, + ignoredStepUpdateCount: 0, + shouldRequeue: true, + }); + + await vi.advanceTimersByTimeAsync(60000); + + await expect(resultPromise).resolves.toBe(false); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-001", + expect.stringContaining("Context compaction timed out"), + ); + + vi.useRealTimers(); + }); }); // ── Context limit error recovery tests ──────────────────────────────── @@ -3712,4 +3738,3 @@ describe("U2: fn_review_step RETHINK delegates to resetStepToBaseline (character ); }); }); - diff --git a/packages/engine/src/__tests__/run-verification-command.test.ts b/packages/engine/src/__tests__/run-verification-command.test.ts index bfe4c44f2d..8a7c94fe3b 100644 --- a/packages/engine/src/__tests__/run-verification-command.test.ts +++ b/packages/engine/src/__tests__/run-verification-command.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, vi } from "vitest"; import { tmpdir } from "node:os"; -import { runVerificationCommand, type RunVerificationOptions } from "../run-verification-tool.js"; +import { join } from "node:path"; +import { runVerificationCommand, normalizeVerificationCommand, type RunVerificationOptions } from "../run-verification-tool.js"; // Some tests use platform-appropriate shell syntax. On Windows, sh-style // quoting and pipes through `printf` are different — these tests are skipped @@ -23,6 +24,37 @@ const itPosix = onPosix ? it : it.skip; // so we fall back to os.tmpdir() which is always C:\Users\…\Temp there. describe("runVerificationCommand", { timeout: 30000 }, () => { const tempDir = onPosix ? "/tmp" : tmpdir(); + const workspaceRoot = join(process.cwd(), "../.."); + + describe("command normalization", () => { + it("rewrites package test -- --run filters to direct vitest with package-relative files", () => { + const result = normalizeVerificationCommand( + [ + "pnpm --filter @fusion/dashboard test -- --run", + "packages/dashboard/src/__tests__/routes-tasks.test.ts", + "packages/dashboard/src/__tests__/routes-settings.test.ts", + ].join(" "), + workspaceRoot, + ); + + expect(result.command).toBe( + [ + "pnpm --filter @fusion/dashboard exec vitest run", + "src/__tests__/routes-tasks.test.ts", + "src/__tests__/routes-settings.test.ts", + "--silent=passed-only --reporter=dot", + ].join(" "), + ); + expect(result.warnings).toEqual([ + expect.stringContaining("rewrote package test file filter"), + ]); + }); + + it("leaves ordinary package tests unchanged when no file filter is forwarded", () => { + const command = "pnpm --filter @fusion/dashboard test"; + expect(normalizeVerificationCommand(command, workspaceRoot)).toEqual({ command, warnings: [] }); + }); + }); describe("basic command execution", () => { it("executes a simple echo command and captures output", async () => { @@ -76,6 +108,24 @@ describe("runVerificationCommand", { timeout: 30000 }, () => { }); }); + describe("timeouts", () => { + itPosix("times out and kills a quiet long-running process group", async () => { + const onHeartbeat = vi.fn(); + const opts: RunVerificationOptions = { + command: "sh -c 'sleep 10 & wait'", + cwd: tempDir, + timeoutMs: 100, + onHeartbeat, + }; + + const result = await runVerificationCommand(opts); + + expect(result.success).toBe(false); + expect(result.timedOut).toBe(true); + expect(result.durationMs).toBeLessThan(5_000); + }); + }); + describe("output capture", () => { itPosix("captures multi-line stdout (POSIX shell)", async () => { // POSIX uses `;` as a command separator; cmd.exe uses `&`. Skip on Windows. diff --git a/packages/engine/src/__tests__/stuck-task-detector.test.ts b/packages/engine/src/__tests__/stuck-task-detector.test.ts index 52b497b5b1..1dd278287a 100644 --- a/packages/engine/src/__tests__/stuck-task-detector.test.ts +++ b/packages/engine/src/__tests__/stuck-task-detector.test.ts @@ -1082,6 +1082,40 @@ describe("StuckTaskDetector", () => { }); describe("onLoopDetected pre-kill callback", () => { + it("suppresses repeat loop detection while compact recovery callback is still pending", async () => { + let resolveLoop: (value: boolean) => void = () => {}; + const onLoopDetected = vi.fn(() => new Promise((resolve) => { + resolveLoop = resolve; + })); + const onStuck = vi.fn(); + const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck }); + const session = createMockSession(); + + vi.useFakeTimers({ shouldAdvanceTime: true }); + customDetector.trackTask("FN-001", session); + vi.advanceTimersByTime(61000); + for (let i = 0; i < 80; i++) { + customDetector.recordActivity("FN-001"); + } + + const firstDetection = customDetector.killAndRetry("FN-001", 60000); + await Promise.resolve(); + + expect(onLoopDetected).toHaveBeenCalledTimes(1); + expect(customDetector.classifyStuckReason("FN-001", 60000)).toBeNull(); + + await customDetector.killAndRetry("FN-001", 60000); + expect(onLoopDetected).toHaveBeenCalledTimes(1); + expect(session.dispose).not.toHaveBeenCalled(); + expect(onStuck).not.toHaveBeenCalled(); + + resolveLoop(true); + await firstDetection; + expect(customDetector.trackedCount).toBe(1); + + vi.useRealTimers(); + }); + it("calls onLoopDetected before onStuck when reason is loop", async () => { const callOrder: string[] = []; const onLoopDetected = vi.fn(async () => { callOrder.push("onLoopDetected"); return false; }); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 6439dbf5bb..d636a4a464 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -271,6 +271,8 @@ const MAX_TASK_DONE_REQUEUE_RETRIES = 3; const COMPLETED_TASK_WATCHDOG_MS = 60_000; /** How long to wait before retrying a workflow rerun handoff that never reached in-progress. */ const WORKFLOW_RERUN_WATCHDOG_MS = 15_000; +/** Upper bound for in-process loop recovery before falling through to kill/requeue. */ +const LOOP_COMPACTION_TIMEOUT_MS = 60_000; const TASK_DONE_REFUSAL_SUFFIX = "Either finish the work and resubmit, or do not call fn_task_done — exit the session and the engine will requeue."; @@ -1085,6 +1087,7 @@ For ALL test/lint/build/typecheck verification, use the \`fn_run_verification\` The tool prevents your session from being killed by the inactivity watchdog during long compiles. - Prefer **package-scoped** verification first: e.g. \`pnpm --filter @fusion/ test\` with \`scope: "package"\`. This is faster and isolated. +- For file-specific package tests, use direct Vitest execution with package-relative paths: \`pnpm --filter @fusion/ exec vitest run src/path/to/test.ts --silent=passed-only --reporter=dot\`. Do not use \`pnpm --filter @fusion/ test -- --run \`; package test scripts can expand into broad quality suites before the filter is applied. - Only run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) at the FINAL integration step, when you are about to call \`fn_task_done\`. - If you need to run \`pnpm install\` (e.g. you added a new package), use \`fn_run_verification\` with \`scope: "workspace"\` and \`timeoutSec: 600\`. - If a verification command times out, do NOT blindly retry — investigate. Check for hung subprocesses, infinite test loops, or tests waiting on missing dependencies. Use \`node_modules/.modules.yaml\` presence to confirm bootstrap. @@ -13416,10 +13419,34 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit executorLog.log(`${taskId} loop detected (attempt ${attempt}) — attempting compact-and-resume`); await this.store.logEntry(taskId, `Loop detected (${event.activitySinceProgress} events since last progress) — attempting compact-and-resume (attempt ${attempt})`); - const compactResult = await compactSessionContext(activeEntry.session); + let compactionTimedOut = false; + let compactionTimer: ReturnType | undefined; + let compactResult: Awaited> | null; + try { + compactResult = await Promise.race([ + compactSessionContext(activeEntry.session), + new Promise((resolve) => { + compactionTimer = setTimeout(() => { + compactionTimedOut = true; + resolve(null); + }, LOOP_COMPACTION_TIMEOUT_MS); + }), + ]); + } finally { + if (compactionTimer) clearTimeout(compactionTimer); + } if (!compactResult) { - executorLog.log(`${taskId} compaction failed or unavailable — falling back to kill/requeue`); - await this.store.logEntry(taskId, "Context compaction failed or unavailable — falling back to kill/requeue"); + const reason = compactionTimedOut + ? `Context compaction timed out after ${LOOP_COMPACTION_TIMEOUT_MS / 1000}s` + : "Context compaction failed or unavailable"; + executorLog.log(`${taskId} ${reason.toLowerCase()} — falling back to kill/requeue`); + await this.store.logEntry(taskId, `${reason} — falling back to kill/requeue`); + return false; + } + + if (this.activeSessions.get(taskId)?.session !== activeEntry.session) { + executorLog.log(`${taskId} compaction completed after session changed — falling back to kill/requeue`); + await this.store.logEntry(taskId, "Context compaction completed after session changed — falling back to kill/requeue"); return false; } diff --git a/packages/engine/src/run-verification-tool.ts b/packages/engine/src/run-verification-tool.ts index cb74841f52..fb0fce9194 100644 --- a/packages/engine/src/run-verification-tool.ts +++ b/packages/engine/src/run-verification-tool.ts @@ -19,8 +19,8 @@ */ import { spawn } from "node:child_process"; -import { existsSync } from "node:fs"; -import { isAbsolute, join } from "node:path"; +import { existsSync, readdirSync, readFileSync } from "node:fs"; +import { isAbsolute, join, relative } from "node:path"; import { Type, type Static } from "@earendil-works/pi-ai"; import type { ToolDefinition } from "@earendil-works/pi-coding-agent"; import { executorLog } from "./logger.js"; @@ -36,6 +36,190 @@ const DEFAULT_TIMEOUT_PACKAGE_SEC = 300; const DEFAULT_TIMEOUT_WORKSPACE_SEC = 900; const MAX_TIMEOUT_SEC = 1800; +function shellSplit(input: string): string[] { + const tokens: string[] = []; + let current = ""; + let quote: "'" | "\"" | null = null; + let escaped = false; + + for (const char of input) { + if (escaped) { + current += char; + escaped = false; + continue; + } + if (char === "\\" && quote !== "'") { + escaped = true; + continue; + } + if (quote) { + if (char === quote) { + quote = null; + } else { + current += char; + } + continue; + } + if (char === "'" || char === "\"") { + quote = char; + continue; + } + if (/\s/.test(char)) { + if (current.length > 0) { + tokens.push(current); + current = ""; + } + continue; + } + current += char; + } + + if (escaped) current += "\\"; + if (current.length > 0) tokens.push(current); + return tokens; +} + +function shellQuote(value: string): string { + if (/^[A-Za-z0-9_@%+=:,./-]+$/.test(value)) return value; + return `'${value.replace(/'/g, "'\\''")}'`; +} + +function findWorkspacePackageDir(rootDir: string, packageName: string): string | null { + if (packageName === "@runfusion/fusion") return "packages/cli"; + + const lastSegment = packageName.split("/").pop(); + const candidates = [ + lastSegment ? `packages/${lastSegment}` : "", + lastSegment === "fusion" ? "packages/cli" : "", + ].filter(Boolean); + + for (const candidate of candidates) { + if (!candidate) continue; + try { + const pkg = JSON.parse(readFileSync(join(rootDir, candidate, "package.json"), "utf8")) as { name?: string }; + if (pkg.name === packageName) return candidate; + } catch { + // Keep looking. + } + } + + const queue: Array<{ dir: string; depth: number }> = [ + { dir: "packages", depth: 0 }, + { dir: "plugins", depth: 0 }, + ]; + while (queue.length > 0) { + const current = queue.shift(); + if (!current) break; + const abs = join(rootDir, current.dir); + let entries: import("node:fs").Dirent[]; + try { + entries = readdirSync(abs, { withFileTypes: true }); + } catch { + continue; + } + for (const entry of entries) { + if (!entry.isDirectory()) continue; + if (entry.name === "node_modules" || entry.name === "dist") continue; + const child = join(current.dir, entry.name); + try { + const pkg = JSON.parse(readFileSync(join(rootDir, child, "package.json"), "utf8")) as { name?: string }; + if (pkg.name === packageName) return child; + } catch { + if (current.depth < 3) queue.push({ dir: child, depth: current.depth + 1 }); + } + } + } + + return null; +} + +function toPackageRelativeFilter(token: string, rootDir: string, packageDir: string): string { + const normalizedPackageDir = packageDir.replace(/\\/g, "/"); + const normalized = token.replace(/\\/g, "/").replace(/^\.\//, ""); + + if (normalized.startsWith(`${normalizedPackageDir}/`)) { + return normalized.slice(normalizedPackageDir.length + 1); + } + + if (isAbsolute(token)) { + const rel = relative(join(rootDir, packageDir), token).replace(/\\/g, "/"); + if (!rel.startsWith("../") && rel !== "..") return rel; + } + + return token; +} + +export function normalizeVerificationCommand(command: string, rootDir: string): { command: string; warnings: string[] } { + const tokens = shellSplit(command); + const warnings: string[] = []; + if (tokens[0] !== "pnpm") return { command, warnings }; + + const filterIndex = tokens.findIndex((token) => token === "--filter" || token === "-F"); + if (filterIndex < 0 || !tokens[filterIndex + 1]) return { command, warnings }; + const packageName = tokens[filterIndex + 1]!; + const separatorIndex = tokens.indexOf("--"); + if (separatorIndex < 0) return { command, warnings }; + + const scriptTokens = tokens.slice(filterIndex + 2, separatorIndex); + const runsTestScript = + scriptTokens.includes("test") + || (scriptTokens[0] === "run" && scriptTokens[1] === "test"); + if (!runsTestScript) return { command, warnings }; + + const forwarded = tokens.slice(separatorIndex + 1); + if (!forwarded.includes("--run")) return { command, warnings }; + + const packageDir = findWorkspacePackageDir(rootDir, packageName); + if (!packageDir) return { command, warnings }; + + const vitestArgs = forwarded + .filter((token) => token !== "--run") + .map((token) => ( + token.startsWith("-") + ? token + : toPackageRelativeFilter(token, rootDir, packageDir) + )); + + const hasReporter = vitestArgs.some((token) => token === "--reporter" || token.startsWith("--reporter=")); + const hasSilent = vitestArgs.some((token) => token === "--silent" || token.startsWith("--silent=")); + const normalizedTokens = [ + "pnpm", + "--filter", + packageName, + "exec", + "vitest", + "run", + ...vitestArgs, + ...(hasSilent ? [] : ["--silent=passed-only"]), + ...(hasReporter ? [] : ["--reporter=dot"]), + ]; + const normalizedCommand = normalizedTokens.map(shellQuote).join(" "); + + if (normalizedCommand !== command) { + warnings.push( + "rewrote package test file filter to direct vitest execution so package test scripts do not expand into broad quality suites", + ); + } + + return { command: normalizedCommand, warnings }; +} + +function killVerificationProcess(child: ReturnType, signal: NodeJS.Signals): void { + if (process.platform !== "win32" && child.pid) { + try { + process.kill(-child.pid, signal); + return; + } catch { + // Fall back to killing the immediate child below. + } + } + try { + child.kill(signal); + } catch { + // The process may already have exited. + } +} + // --------------------------------------------------------------------------- // Tool parameter schema // --------------------------------------------------------------------------- @@ -192,6 +376,7 @@ export async function runVerificationCommand( COREPACK_ENABLE_DOWNLOAD_PROMPT: "0", }, shell: true, + detached: process.platform !== "win32", }); let timedOut = false; @@ -211,20 +396,21 @@ export async function runVerificationCommand( }, QUIET_HEARTBEAT_INTERVAL_MS); // ── Hard timeout ──────────────────────────────────────────────────────── + let killTimer: ReturnType | null = null; const hardTimer = setTimeout(() => { if (settled) return; timedOut = true; executorLog.warn( `[fn_run_verification] hard timeout (${timeoutMs / 1000}s) — sending SIGTERM to: ${command}`, ); - child.kill("SIGTERM"); + killVerificationProcess(child, "SIGTERM"); - setTimeout(() => { + killTimer = setTimeout(() => { if (!settled) { executorLog.warn( `[fn_run_verification] SIGTERM ignored — sending SIGKILL to: ${command}`, ); - child.kill("SIGKILL"); + killVerificationProcess(child, "SIGKILL"); killed = true; } }, SIGKILL_GRACE_MS); @@ -266,6 +452,7 @@ export async function runVerificationCommand( settled = true; clearInterval(quietTimer); clearTimeout(hardTimer); + if (killTimer) clearTimeout(killTimer); // Flush remainders if (stdoutRemainder) appendToBuffer(stdoutBuf, stdoutRemainder); @@ -301,6 +488,7 @@ export async function runVerificationCommand( settled = true; clearInterval(quietTimer); clearTimeout(hardTimer); + if (killTimer) clearTimeout(killTimer); const durationMs = Date.now() - startMs; warnings.push(`Spawn error: ${err.message}`); resolve({ @@ -402,7 +590,13 @@ export function createRunVerificationTool( // If the command is package-scoped and the workspace has no .modules.yaml, // prepend a pnpm install so the agent doesn't stall on missing node_modules. let effectiveCommand = command; - if (command.trimStart().startsWith("pnpm --filter")) { + const normalized = normalizeVerificationCommand(effectiveCommand, rootDir); + if (normalized.command !== effectiveCommand) { + effectiveCommand = normalized.command; + warnings.push(...normalized.warnings); + } + + if (effectiveCommand.trimStart().startsWith("pnpm --filter")) { const modulesYaml = join(rootDir, "node_modules", ".modules.yaml"); if (!existsSync(modulesYaml)) { const installCmd = "pnpm install --prefer-offline"; @@ -411,7 +605,7 @@ export function createRunVerificationTool( `auto-prepending \`${installCmd}\` before running the command.`; warnings.push(msg); log.warn(`[fn_run_verification] ${taskId}: ${msg}`); - effectiveCommand = `${installCmd} && ${command}`; + effectiveCommand = `${installCmd} && ${effectiveCommand}`; } } diff --git a/packages/engine/src/stuck-task-detector.ts b/packages/engine/src/stuck-task-detector.ts index 5d0a5a7390..34c0705662 100644 --- a/packages/engine/src/stuck-task-detector.ts +++ b/packages/engine/src/stuck-task-detector.ts @@ -459,8 +459,10 @@ export class StuckTaskDetector { const activitySinceProgress = entry.activitySinceProgress; const ignoredStepUpdateCount = entry.ignoredStepUpdateCount; - // Classify the reason - const reason = this.classifyStuckReason(taskId, timeoutMs) ?? "inactivity"; + // Classify the reason. If recovery is already pending or fresh progress + // made the task no longer stuck, do not kill it. + const reason = this.classifyStuckReason(taskId, timeoutMs); + if (reason === null) return; const elapsedMin = Math.round(inactivityMs / 60_000); const noProgressMin = Math.round(noProgressMs / 60_000); @@ -529,6 +531,10 @@ export class StuckTaskDetector { // dispose/requeue but keep tracking alive for FN-5168 churn accounting. // Errors fall through to normal kill. if (reason === "loop" && this.onLoopDetected) { + // Mark recovery as pending before awaiting the callback. Compaction or + // steering can be slow; without this, the next poll can re-enter the same + // loop recovery path and spam compact-and-resume attempts for one session. + entry.recoveryInProgress = true; try { const handled = await this.onLoopDetected(event); if (handled) { @@ -539,10 +545,11 @@ export class StuckTaskDetector { // Keep the task tracked so FN-5168 can keep counting ignored // step-update churn after recovery resumes, but suppress detector // polling until the executor emits fresh progress on resume. - entry.recoveryInProgress = true; return; } + entry.recoveryInProgress = false; } catch (err) { + entry.recoveryInProgress = false; stuckLog.error(`onLoopDetected callback failed for ${canonicalId}:`, err); // Fall through to normal kill path } diff --git a/packages/engine/src/workflow-authoritative-driver.ts b/packages/engine/src/workflow-authoritative-driver.ts index 77c115981f..2604e8bb3d 100644 --- a/packages/engine/src/workflow-authoritative-driver.ts +++ b/packages/engine/src/workflow-authoritative-driver.ts @@ -15,7 +15,7 @@ import { executorLog } from "./logger.js"; import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; -const AUTHORITATIVE_WORKFLOW_ID = "builtin:workflow-interpreter-authoritative"; +const AUTHORITATIVE_WORKFLOW_ID = "internal:workflow-interpreter-authoritative"; export interface WorkflowAuthoritativeDriverStore { getSettings(): Promise;