From aa8bd3dc92c45c8c5d016f1a9cae7e8ef1558ec9 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 18:08:42 -0700 Subject: [PATCH 1/2] fix(FN-6043): recover stuck task processing Fusion-Task-Id: FN-6043 --- .changeset/stuck-task-recovery.md | 5 + .../core/src/__tests__/agent-prompts.test.ts | 20 ++- packages/core/src/agent-prompts.ts | 17 +- .../app/__tests__/pull-request-view.test.tsx | 13 +- .../__tests__/TaskCard.cli-states.test.tsx | 4 +- .../WorkflowNodeEditor.cli-agent.test.tsx | 12 ++ .../__tests__/run-vitest-with-heap.test.ts | 54 ++++++ .../scripts/run-vitest-with-heap.mjs | 33 +++- packages/dashboard/vitest.config.ts | 4 +- .../engine/src/__tests__/concurrency.test.ts | 44 +++++ .../src/__tests__/executor-prompt.test.ts | 20 +-- .../src/__tests__/executor-recovery.test.ts | 59 ++++++- .../__tests__/executor-step-session.test.ts | 30 +--- .../engine/src/__tests__/scheduler.test.ts | 28 +++ packages/engine/src/concurrency.ts | 32 ++++ packages/engine/src/executor.ts | 81 ++++++--- packages/engine/src/run-verification-tool.ts | 35 ++-- .../src/sandbox/__tests__/native.test.ts | 77 ++++++++- packages/engine/src/sandbox/native.ts | 161 +++++++++++++----- packages/engine/src/scheduler.ts | 41 +++++ packages/engine/src/triage.ts | 37 +++- 21 files changed, 643 insertions(+), 164 deletions(-) create mode 100644 .changeset/stuck-task-recovery.md diff --git a/.changeset/stuck-task-recovery.md b/.changeset/stuck-task-recovery.md new file mode 100644 index 0000000000..5633225e59 --- /dev/null +++ b/.changeset/stuck-task-recovery.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Fix stuck task recovery by preserving retryable requeues, supervising verification subprocesses, and narrowing executor verification guidance to impacted work. diff --git a/packages/core/src/__tests__/agent-prompts.test.ts b/packages/core/src/__tests__/agent-prompts.test.ts index 9d4a7de5a9..47829f8fba 100644 --- a/packages/core/src/__tests__/agent-prompts.test.ts +++ b/packages/core/src/__tests__/agent-prompts.test.ts @@ -127,15 +127,16 @@ describe("resolveAgentPrompt", () => { expect(result).toContain("task execution agent"); }); - it("built-in executor prompt requires resolving ALL lint and test failures including unrelated", () => { + it("built-in executor prompt limits fixes to impacted failures and follow-ups unrelated broad-suite failures", () => { const result = resolveAgentPrompt("executor"); - // The stricter language must be present to prevent "unrelated failure" deferrals - expect(result).toContain("Resolve ALL lint failures and test failures"); - expect(result).toContain("even if they appear unrelated or pre-existing"); - expect(result).toContain("do not defer them to a separate task"); + expect(result).toContain("Keep fixing failures caused by your change"); + expect(result).toContain("impacted tests"); + expect(result).toContain("unrelated or pre-existing failures"); + expect(result).toContain("create/link a follow-up task"); + expect(result).not.toContain("Resolve ALL lint failures and test failures"); }); - it("senior-engineer prompt requires resolving ALL lint and test failures including unrelated", () => { + it("senior-engineer prompt limits fixes to impacted failures and follow-ups unrelated broad-suite failures", () => { const config: AgentPromptsConfig = { roleAssignments: { executor: "senior-engineer", @@ -143,9 +144,10 @@ describe("resolveAgentPrompt", () => { }; const result = resolveAgentPrompt("executor", config); - expect(result).toContain("Resolve ALL lint failures and test failures"); - expect(result).toContain("even if they appear unrelated or pre-existing"); - expect(result).toContain("do not defer them to a separate task"); + expect(result).toContain("Lint, tests, and typecheck are also hard quality gates for failures caused by this task"); + expect(result).toContain("unrelated or pre-existing broad-suite failures"); + expect(result).toContain("create/link follow-up work"); + expect(result).not.toContain("Resolve ALL lint failures and test failures"); }); it("built-in executor prompt includes worktree boundary guidance", () => { diff --git a/packages/core/src/agent-prompts.ts b/packages/core/src/agent-prompts.ts index 3c9e01a4d6..50c6054db8 100644 --- a/packages/core/src/agent-prompts.ts +++ b/packages/core/src/agent-prompts.ts @@ -188,10 +188,11 @@ If a project build command is listed in the prompt, it is a hard completion gate - If the build fails, do NOT call \`task_done()\`; keep working until it passes Lint, tests, and typecheck are also hard quality gates: -- Keep fixing failures until lint, the configured/full test suite, and typecheck all pass -- If the repository exposes a typecheck command, run it and keep fixing failures until it passes -- Do not stop at "out of scope" if additional fixes are required to restore green lint, tests, build, or typecheck -- **CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task. +- Keep fixing failures caused by your change until lint, impacted tests, build, and typecheck pass. +- If the repository exposes a typecheck command, run it and fix failures caused by your change. +- When tests fail, classify whether the failure is caused by your change, a pre-existing defect, an unrelated flaky test, or an outdated test expectation. +- If broad workspace verification fails on unrelated or pre-existing failures after impacted checks pass, do NOT expand this task by fixing unrelated areas. Log the evidence, quarantine flakes per project policy, or create/link a follow-up task. +- Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command. ## Verification commands — use fn_run_verification @@ -200,7 +201,7 @@ The tool prevents your session from being killed by the inactivity watchdog duri - 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()\`. +- Run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) only when it is explicitly required by the task/workflow or after impacted/package-scoped checks pass and you are doing final integration. - 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.`; @@ -283,7 +284,7 @@ For bug-fix tasks, paste and fill in this checklist in the \`## Surface Enumerat > If keeping lint/tests/build/typecheck green requires edits outside the initial File Scope, make those fixes as part of this task. - [ ] Run lint check (\`pnpm lint\`) -- [ ] Run full test suite +- [ ] Run impacted tests - [ ] Run project typecheck if available - [ ] Fix all failures - [ ] Build passes @@ -727,8 +728,8 @@ Call \`task_done()\` to signal completion. \`\`\` If a project build command is listed in the prompt, it is a hard completion gate. -Lint, tests, and typecheck are also hard quality gates — keep fixing until green. -**CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.`; +Lint, tests, and typecheck are also hard quality gates for failures caused by this task. +If unrelated or pre-existing broad-suite failures remain after impacted checks pass, log the evidence and create/link follow-up work instead of expanding the task.`; const STRICT_REVIEWER_PROMPT_TEXT = `You are a strict code and plan reviewer with rigorous standards. diff --git a/packages/dashboard/app/__tests__/pull-request-view.test.tsx b/packages/dashboard/app/__tests__/pull-request-view.test.tsx index 67b1d695de..06d55dcfc4 100644 --- a/packages/dashboard/app/__tests__/pull-request-view.test.tsx +++ b/packages/dashboard/app/__tests__/pull-request-view.test.tsx @@ -5,7 +5,18 @@ import { PullRequestView, type PrDetail } from "../components/PullRequestView"; // Icons → simple stubs so assertions key on text/testids, not SVG internals. vi.mock("lucide-react", () => { const Stub = () => ; - return new Proxy({}, { get: () => Stub }); + return { + AlertTriangle: Stub, + CheckCircle: Stub, + Clock: Stub, + ExternalLink: Stub, + GitMerge: Stub, + GitPullRequest: Stub, + MessageSquare: Stub, + RotateCcw: Stub, + ThumbsUp: Stub, + XCircle: Stub, + }; }); function makeSummary(over: Partial = {}): PrDetail["summary"] { diff --git a/packages/dashboard/app/components/__tests__/TaskCard.cli-states.test.tsx b/packages/dashboard/app/components/__tests__/TaskCard.cli-states.test.tsx index 343bbfa53b..0a36e15492 100644 --- a/packages/dashboard/app/components/__tests__/TaskCard.cli-states.test.tsx +++ b/packages/dashboard/app/components/__tests__/TaskCard.cli-states.test.tsx @@ -6,7 +6,9 @@ import type { Task } from "@fusion/core"; vi.mock("lucide-react", () => { const Stub = () => null; - return new Proxy({}, { get: () => Stub }); + return new Proxy({}, { + get: (_target, prop) => prop === "then" ? undefined : Stub, + }); }); vi.mock("../ProviderIcon", () => ({ diff --git a/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.cli-agent.test.tsx b/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.cli-agent.test.tsx index 5dd9ad152e..d02cb97fc1 100644 --- a/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.cli-agent.test.tsx +++ b/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.cli-agent.test.tsx @@ -13,6 +13,14 @@ vi.mock("../../api", () => ({ fetchModels: vi.fn(), fetchAgents: vi.fn(), fetchDiscoveredSkills: vi.fn(), + fetchWorkflowStepTemplates: vi.fn().mockResolvedValue({ templates: [] }), + fetchPluginWorkflowStepTemplates: vi.fn().mockResolvedValue({ templates: [] }), + fetchConfig: vi.fn(), + fetchSettings: vi.fn(), + updateSettings: vi.fn(), + updateGlobalSettings: vi.fn(), + fetchWorkflowSettingValues: vi.fn().mockResolvedValue({ stored: {}, effective: {}, orphaned: [] }), + updateWorkflowSettingValues: vi.fn().mockResolvedValue({ stored: {}, effective: {}, orphaned: [] }), })); import { @@ -21,6 +29,8 @@ import { fetchStepParsers, updateWorkflow, fetchModels, + fetchConfig, + fetchSettings, } from "../../api"; import type { TraitCatalogEntry } from "../../api"; import { WorkflowNodeEditor } from "../WorkflowNodeEditor"; @@ -68,6 +78,8 @@ describe("WorkflowNodeEditor — cli-agent executor (U15)", () => { vi.mocked(fetchTraits).mockResolvedValue(TRAIT_CATALOG); vi.mocked(fetchStepParsers).mockResolvedValue([]); vi.mocked(fetchModels).mockResolvedValue({ models: [] }); + vi.mocked(fetchConfig).mockResolvedValue({ maxConcurrent: 2, rootDir: "/tmp/project" }); + vi.mocked(fetchSettings).mockResolvedValue({ autoMerge: true }); vi.mocked(updateWorkflow).mockResolvedValue(promptDef()); // Stub the adapter-catalog fetch. vi.stubGlobal( diff --git a/packages/dashboard/scripts/__tests__/run-vitest-with-heap.test.ts b/packages/dashboard/scripts/__tests__/run-vitest-with-heap.test.ts index de2ef12642..7dbdabd295 100644 --- a/packages/dashboard/scripts/__tests__/run-vitest-with-heap.test.ts +++ b/packages/dashboard/scripts/__tests__/run-vitest-with-heap.test.ts @@ -127,6 +127,56 @@ async function spawnWrapperTree(signal: NodeJS.Signals) { await waitFor(() => !isProcessAlive(pids!.childPid) && !isProcessAlive(pids!.grandchildPid)); } +async function spawnWrapperTreeUntilTimeout() { + const { pidFile, childPath, grandchildPath } = createStubProcessTree(); + const wrapper = spawn( + process.execPath, + [wrapperPath, "--heap=6144", "run", "--project", "dashboard-api-quality"], + { + cwd: dashboardRoot, + stdio: "pipe", + env: { + ...process.env, + FUSION_RUN_VITEST_TIMEOUT_MS: "100", + FUSION_RUN_VITEST_KILL_GRACE_MS: "50", + FUSION_RUN_VITEST_SPAWN_OVERRIDE: JSON.stringify({ + command: process.execPath, + args: [childPath, pidFile, grandchildPath], + }), + }, + }, + ); + activeWrappers.add(wrapper); + + let pids: { childPid: number; grandchildPid: number } | null = null; + await waitFor(() => { + try { + pids = JSON.parse(readFileSync(pidFile, "utf8")) as { childPid: number; grandchildPid: number }; + return Boolean( + pids && + Number.isInteger(pids.childPid) && + Number.isInteger(pids.grandchildPid) && + isProcessAlive(pids.childPid) && + isProcessAlive(pids.grandchildPid), + ); + } catch { + return false; + } + }); + + registerGroupLeader(pids!.childPid); + registerPid(pids!.grandchildPid); + + const exitCode = await new Promise((resolve, reject) => { + wrapper.once("error", reject); + wrapper.once("close", (code) => resolve(code)); + }); + activeWrappers.delete(wrapper); + + expect(exitCode).toBe(124); + await waitFor(() => !isProcessAlive(pids!.childPid) && !isProcessAlive(pids!.grandchildPid)); +} + afterEach(async () => { for (const wrapper of activeWrappers) { wrapper.kill("SIGKILL"); @@ -169,4 +219,8 @@ describe("run-vitest-with-heap", () => { it("reaps the spawned process group on SIGINT", async () => { await spawnWrapperTree("SIGINT"); }); + + it("times out and reaps the spawned process group", async () => { + await spawnWrapperTreeUntilTimeout(); + }); }); diff --git a/packages/dashboard/scripts/run-vitest-with-heap.mjs b/packages/dashboard/scripts/run-vitest-with-heap.mjs index f177b4bc8c..3e39b8402a 100644 --- a/packages/dashboard/scripts/run-vitest-with-heap.mjs +++ b/packages/dashboard/scripts/run-vitest-with-heap.mjs @@ -1,5 +1,5 @@ #!/usr/bin/env node -/* global clearInterval, console, process, setInterval */ +/* global clearInterval, clearTimeout, console, process, setInterval, setTimeout */ import { spawn } from "node:child_process"; @@ -16,6 +16,8 @@ if (vitestArgs.length === 0) { const nodeOptions = [`--max-old-space-size=${heapMb}`, process.env.NODE_OPTIONS || ""] .join(" ") .trim(); +const timeoutMs = Number.parseInt(process.env.FUSION_RUN_VITEST_TIMEOUT_MS || "900000", 10); +const forceKillGraceMs = Number.parseInt(process.env.FUSION_RUN_VITEST_KILL_GRACE_MS || "5000", 10); function resolveSpawnCommand() { const override = process.env.FUSION_RUN_VITEST_SPAWN_OVERRIDE; @@ -51,11 +53,31 @@ const child = spawn(command, args, { const heartbeat = setInterval(() => { console.log(`[dashboard-vitest] still running: ${vitestArgs.join(" ")}`); }, 5_000); +let timeoutExitCode = null; +let forceKillTimer = null; +const timeout = Number.isFinite(timeoutMs) && timeoutMs > 0 + ? setTimeout(() => { + timeoutExitCode = 124; + console.error(`[dashboard-vitest] timeout after ${timeoutMs}ms: ${vitestArgs.join(" ")}`); + forwardSignal("SIGTERM"); + forceKillTimer = setTimeout(() => { + forwardSignal("SIGKILL"); + }, Math.max(1, forceKillGraceMs)); + forceKillTimer.unref(); + }, timeoutMs) + : null; +timeout?.unref(); function clearHeartbeat() { clearInterval(heartbeat); } +function clearTimers() { + clearHeartbeat(); + if (timeout) clearTimeout(timeout); + if (forceKillTimer) clearTimeout(forceKillTimer); +} + function forwardSignal(signal) { clearHeartbeat(); @@ -86,7 +108,7 @@ for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"]) { } process.on("exit", () => { - clearHeartbeat(); + clearTimers(); try { process.kill(-child.pid, "SIGTERM"); } catch (error) { @@ -101,13 +123,16 @@ process.on("exit", () => { }); child.on("error", (error) => { - clearHeartbeat(); + clearTimers(); console.error(error); process.exit(1); }); child.on("close", (code, signal) => { - clearHeartbeat(); + clearTimers(); + if (timeoutExitCode !== null) { + process.exit(timeoutExitCode); + } if (signal) { process.kill(process.pid, signal); return; diff --git a/packages/dashboard/vitest.config.ts b/packages/dashboard/vitest.config.ts index 9aac1aaa52..03c9abeacc 100644 --- a/packages/dashboard/vitest.config.ts +++ b/packages/dashboard/vitest.config.ts @@ -209,9 +209,7 @@ const batchedQualityAppComponentTestsA = batchedQualityAppComponentTests.slice(0 const batchedQualityAppComponentTestsB = batchedQualityAppComponentTests.slice(batchedQualityAppSplitIndex); function buildComponentQualityInclude(testNames: readonly string[]): string[] { - return testNames.length > 0 - ? [`app/components/__tests__/{${testNames.join(",")}}.test.{ts,tsx}`] - : []; + return testNames.map((testName) => `app/components/__tests__/${testName}.test.{ts,tsx}`); } const qualityAppTests = [ diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index 258145ce98..5276c46af3 100644 --- a/packages/engine/src/__tests__/concurrency.test.ts +++ b/packages/engine/src/__tests__/concurrency.test.ts @@ -140,6 +140,50 @@ describe("AgentSemaphore", () => { expect(sem.availableCount).toBe(3); }); + it("reports waitingCount and diagnostic snapshot", async () => { + const sem = new AgentSemaphore(1); + await sem.acquire(); + + const waiter = sem.acquire(); + await Promise.resolve(); + + expect(sem.waitingCount).toBe(1); + expect(sem.snapshot()).toEqual({ + activeCount: 1, + waitingCount: 1, + availableCount: 0, + limit: 1, + }); + + sem.release(); + await waiter; + expect(sem.waitingCount).toBe(0); + sem.release(); + }); + + it("reconciles stale active counts down to persisted active work", async () => { + const sem = new AgentSemaphore(2); + await sem.acquire(); + await sem.acquire(); + + const result = sem.reconcileActiveCount(0); + + expect(result).toEqual({ before: 2, after: 0, changed: true }); + expect(sem.activeCount).toBe(0); + expect(sem.availableCount).toBe(2); + }); + + it("does not increase active counts during reconciliation", async () => { + const sem = new AgentSemaphore(2); + await sem.acquire(); + + const result = sem.reconcileActiveCount(3); + + expect(result).toEqual({ before: 1, after: 1, changed: false }); + expect(sem.activeCount).toBe(1); + sem.release(); + }); + it("run() gates concurrent calls", async () => { const sem = new AgentSemaphore(2); let concurrent = 0; diff --git a/packages/engine/src/__tests__/executor-prompt.test.ts b/packages/engine/src/__tests__/executor-prompt.test.ts index e1e6727942..f4d1f4fe9c 100644 --- a/packages/engine/src/__tests__/executor-prompt.test.ts +++ b/packages/engine/src/__tests__/executor-prompt.test.ts @@ -207,31 +207,30 @@ describe("buildExecutionPrompt", () => { expect(result).toContain("- **Build:** `pnpm build`"); }); - it("tells executors to fix quality-gate failures even outside initial file scope", () => { + it("tells executors to split unrelated broad-suite failures into follow-up work", () => { const task = createMockTaskDetail(); const result = buildExecutionPrompt(task, "/home/user/project", { testCommand: "pnpm test", buildCommand: "pnpm build", } as any); - expect(result).toContain("fix failures even when that requires edits outside the original File Scope"); + expect(result).toContain("caused-by-this-task failures are blocking"); + expect(result).toContain("unrelated or pre-existing failures should be logged and split into a follow-up"); expect(result).toContain("If the repo has a typecheck command, run it before `fn_task_done()`"); - expect(result).toContain("not for fixes required to get tests, build, or typecheck back to green"); + expect(result).toContain("including unrelated/pre-existing broad-suite failures"); }); - it("requires resolving ALL test failures, including unrelated or pre-existing ones", () => { + it("warns against repeated broad workspace verification loops", () => { const task = createMockTaskDetail(); const result = buildExecutionPrompt(task, "/home/user/project", { testCommand: "pnpm test", buildCommand: "pnpm build", } as any); - // The stricter language must be present to prevent "unrelated failure" deferrals - expect(result).toContain("Resolve ALL test failures"); - expect(result).toContain("even if they appear unrelated or pre-existing"); - expect(result).toContain("accumulate technical debt"); - expect(result).toContain("Investigate and fix or suppress them"); - expect(result).toContain("do not defer them to a separate task"); + expect(result).toContain("Do not repeatedly rerun a broad failing or hanging workspace command"); + expect(result).toContain("without a new hypothesis and a narrower confirming command"); + expect(result).toContain("unrelated or pre-existing failures should be logged and split into a follow-up"); + expect(result).not.toContain("Resolve ALL test failures"); }); it("includes source issue reference in commit instruction when task has github sourceIssue", () => { @@ -2571,4 +2570,3 @@ describe("fn_task_update bare-call guard (P1 api-contract)", () => { expect(text).not.toContain("fn_task_update requires at least one of"); }); }); - diff --git a/packages/engine/src/__tests__/executor-recovery.test.ts b/packages/engine/src/__tests__/executor-recovery.test.ts index 1269cbe888..41a673e6dd 100644 --- a/packages/engine/src/__tests__/executor-recovery.test.ts +++ b/packages/engine/src/__tests__/executor-recovery.test.ts @@ -465,7 +465,12 @@ describe("TaskExecutor bounded recovery retries", () => { ); expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review"); // Executor now handles the requeue in its finally block - expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: "stuck-killed", worktree: null, branch: null }); + expect(store.updateTask).toHaveBeenCalledWith("FN-001", { + status: "queued", + error: null, + worktree: null, + branch: null, + }); expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveProgress: true }); }); @@ -499,7 +504,11 @@ describe("TaskExecutor bounded recovery retries", () => { // Should NOT requeue or mark as failed (budget handler already did that) expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "todo"); - expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "stuck-killed", worktree: null, branch: null }); + expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", expect.objectContaining({ + status: "queued", + worktree: null, + branch: null, + })); expect(store.updateTask).not.toHaveBeenCalledWith( "FN-001", expect.objectContaining({ status: "failed" }), @@ -556,6 +565,48 @@ describe("TaskExecutor bounded recovery retries", () => { ); }); + it("does not let a late graph failure clobber a retryable requeue", async () => { + const store = createMockStore(); + const task = { + id: "FN-001", + title: "Test", + description: "Test", + column: "in-progress", + status: undefined, + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + } as Task; + store.getTask.mockResolvedValue({ + ...task, + column: "todo", + status: "queued", + error: null, + }); + const executor = new TaskExecutor(store, "/tmp/test", {}); + + await (executor as any).handleGraphFailure(task, { + visitedNodeIds: ["execute"], + }); + + expect(store.updateTask).not.toHaveBeenCalledWith( + "FN-001", + expect.objectContaining({ status: "failed" }), + expect.anything(), + ); + expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review", expect.anything()); + expect(store.handoffToReview).not.toHaveBeenCalled(); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-001", + "Workflow graph terminated with failure at node 'execute' (task already todo - preserving recovered lifecycle state)", + undefined, + undefined, + ); + }); + it("preserves step progress when requeuing stuck task by default", async () => { const store = createMockStore(); const executor = new TaskExecutor(store, "/tmp/test", {}); @@ -1465,8 +1516,8 @@ describe("Invalid transition error handling", () => { // then throws the Invalid transition error, // which is caught by the outer handler. expect(store.updateTask).toHaveBeenCalledWith("FN-001", { - status: "failed", - error: "Agent finished without calling fn_task_done (after 3 retries)", + status: "queued", + error: null, taskDoneRetryCount: 1, }); diff --git a/packages/engine/src/__tests__/executor-step-session.test.ts b/packages/engine/src/__tests__/executor-step-session.test.ts index ba2a4987b7..92d25db99a 100644 --- a/packages/engine/src/__tests__/executor-step-session.test.ts +++ b/packages/engine/src/__tests__/executor-step-session.test.ts @@ -116,8 +116,8 @@ describe("Workflow Steps Execution", () => { // Retries still didn't call fn_task_done, so it fails and requeues immediately. expect(store.updateTask).toHaveBeenCalledWith("FN-001", { - status: "failed", - error: "Agent finished without calling fn_task_done (after 3 retries)", + status: "queued", + error: null, taskDoneRetryCount: 1, }); expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveProgress: true }); @@ -242,8 +242,9 @@ describe("Workflow Steps Execution", () => { expect(mockedCreateFnAgent).toHaveBeenCalledTimes(4); expect(store.updateTask).toHaveBeenCalledWith("FN-5436-A", expect.objectContaining({ - status: "failed", - error: "Agent finished without calling fn_task_done (after 3 retries)", + status: "queued", + error: null, + taskDoneRetryCount: 1, })); expect(store.moveTask).toHaveBeenCalledWith("FN-5436-A", "todo", { preserveProgress: true }); expect(store.moveTask).not.toHaveBeenCalledWith("FN-5436-A", "in-review"); @@ -335,8 +336,8 @@ describe("Workflow Steps Execution", () => { expect(mockedCreateFnAgent).toHaveBeenCalledTimes(4); expect(store.updateTask).toHaveBeenCalledWith("FN-5436-C", { - status: "failed", - error: "Agent finished without calling fn_task_done (after 3 retries)", + status: "queued", + error: null, taskDoneRetryCount: 1, }); }); @@ -1224,7 +1225,7 @@ describe("Workflow Steps Execution", () => { store.getSettings.mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, - scripts: { test: "echo 'all tests passed'" }, + scripts: { test: `node -e "if (process.env.FN3968_SCRIPT_ENV !== 'workflow-script-env') process.exit(42)"` }, }); store.getTask.mockResolvedValue({ @@ -1254,14 +1255,6 @@ describe("Workflow Steps Execution", () => { updatedAt: new Date().toISOString(), }); - // Mock execSync to succeed for the script command - mockedExecSync.mockImplementation((cmd: string | string[]) => { - if (typeof cmd === "string" && cmd.includes("echo")) { - return Buffer.from("all tests passed\n"); - } - return Buffer.from(""); - }); - // Main agent with fn_task_done createAgentWithTaskDone(); @@ -1308,13 +1301,6 @@ describe("Workflow Steps Execution", () => { ]), }), ); - const updatePayloads = store.updateTask.mock.calls.map((call: any[]) => call[1]); - expect(JSON.stringify(updatePayloads)).not.toContain("all tests passed"); - - const scriptExecCall = mockedExecSync.mock.calls.find( - (call: any[]) => typeof call[0] === "string" && call[0].includes("echo 'all tests passed'") - ); - expect(scriptExecCall?.[1]?.env?.FN3968_SCRIPT_ENV).toBe("workflow-script-env"); delete process.env.FN3968_SCRIPT_ENV; }); diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index b07223c40b..ce91634262 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/packages/engine/src/__tests__/scheduler.test.ts @@ -1527,6 +1527,34 @@ describe("Scheduler", () => { expect(String(call?.[1])).toContain("semaphore slots may include triage/merge agents outside in-progress"); }); + it("recovers an idle leaked semaphore slot before dispatching", async () => { + vi.mocked(existsSync).mockReturnValue(true); + vi.mocked(readFile).mockResolvedValue("# Task\nDo something"); + + const semaphore = new AgentSemaphore(1); + await semaphore.acquire(); + const task = createMockTask({ id: "FN-A", column: "todo", dependencies: [] }); + const store = createMockStore({ + listTasks: vi.fn().mockResolvedValue([task]), + getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 10, maxWorktrees: 10 }), + }); + + const scheduler = new Scheduler(store, { semaphore }); + (scheduler as any).running = true; + (scheduler as any).idleSemaphoreLeakCandidateSince = Date.now() - 6_000; + await scheduler.schedule(); + + expect(semaphore.activeCount).toBe(0); + expect(schedulerLog.warn).toHaveBeenCalledWith( + expect.stringContaining("scheduler: recovered stale semaphore active count 1 -> 0"), + ); + expect(store.moveTask).toHaveBeenCalledWith( + "FN-A", + "in-progress", + expect.objectContaining({ moveSource: "scheduler" }), + ); + }); + it("lists tied binding gates in stable order", async () => { vi.mocked(existsSync).mockReturnValue(true); vi.mocked(readFile).mockResolvedValue("# Task\nDo something"); diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 6bdef4a97b..19d9d56ed1 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -66,6 +66,38 @@ export class AgentSemaphore { return this._active; } + /** Number of callers currently queued for a semaphore slot. */ + get waitingCount(): number { + return this._waiters.length; + } + + /** Snapshot of current semaphore pressure for diagnostics. */ + snapshot(): { activeCount: number; waitingCount: number; availableCount: number; limit: number } { + return { + activeCount: this.activeCount, + waitingCount: this.waitingCount, + availableCount: this.availableCount, + limit: this.limit, + }; + } + + /** + * Clamp stale active-slot accounting to a persisted upper bound. + * + * This is a recovery valve for crash/abort paths where the task/session that + * acquired a slot is gone but the in-memory semaphore did not observe its + * normal `finally` release. The caller owns the persisted-state judgment. + */ + reconcileActiveCount(maxActive: number): { before: number; after: number; changed: boolean } { + const bounded = Math.max(0, Math.floor(maxActive)); + const before = this._active; + if (before > bounded) { + this._active = bounded; + this._drain(); + } + return { before, after: this._active, changed: before !== this._active }; + } + /** Number of slots available for immediate acquisition. May be 0 or negative * if the limit was reduced below the current active count. * Returns 0 when the limit is not a valid positive number (defensive guard). */ diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 982b694951..8d2ef36d85 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1074,12 +1074,12 @@ If a project build command is listed in the prompt, it is a hard completion gate - If the build fails, do NOT call \`fn_task_done()\`; keep working until it passes Lint, tests, and typecheck are also hard quality gates: -- Keep fixing failures until lint, the configured/full test suite, and typecheck all pass -- If the repository exposes a typecheck command, run it and keep fixing failures until it passes -- Do not stop at "out of scope" if additional fixes are required to restore green lint, tests, build, or typecheck -- When tests fail, first identify whether the failure is caused by your change, a pre-existing defect, or an outdated test expectation; then fix code or tests accordingly so behavior and assertions match -- Update tests when intended behavior changed; fix implementation when behavior regressed unintentionally -- **CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task. +- Keep fixing failures caused by your change until lint, targeted tests, build, and typecheck pass. +- If the repository exposes a typecheck command, run it and fix failures caused by your change. +- When tests fail, first identify whether the failure is caused by your change, a pre-existing defect, an unrelated flaky test, or an outdated test expectation. +- Update tests when intended behavior changed; fix implementation when behavior regressed unintentionally. +- If broad workspace verification fails on unrelated or pre-existing failures after targeted checks pass, do NOT expand this task by fixing unrelated areas. Log the evidence, quarantine flakes per project policy, or create/link a follow-up task. +- Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command. ## Verification commands — use fn_run_verification @@ -1088,7 +1088,7 @@ The tool prevents your session from being killed by the inactivity watchdog duri - 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\`. +- Run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) only when it is explicitly required by the task/workflow or after impacted/package-scoped checks pass and you are doing final integration. - 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. @@ -5549,15 +5549,25 @@ export class TaskExecutor { await this.store.logEntry(task.id, `${message} (task paused — not parked)`, undefined, this.getRunContextFor(task.id)); return; } + if (live.column !== "in-progress") { + executorLog.log( + `${task.id}: graph run ended after task moved to '${live.column}' - preserving recovered lifecycle state`, + ); + await this.store.logEntry( + task.id, + `${message} (task already ${live.column} - preserving recovered lifecycle state)`, + undefined, + this.getRunContextFor(task.id), + ); + return; + } await this.store.logEntry(task.id, message, undefined, this.getRunContextFor(task.id)); // status "failed" doubles as the self-healing exemption: review-task // revival sweeps skip tasks carrying a non-null status, preventing the // FN-5704-style loop of re-running the graph from scratch. await this.store.updateTask(task.id, { error: message, status: "failed" }, this.getRunContextFor(task.id)); - if (live.column === "in-progress") { - await this.persistTokenUsage(task.id); - await this.handoffTaskToReview(live, "workflow-graph-failed"); - } + await this.persistTokenUsage(task.id); + await this.handoffTaskToReview(live, "workflow-graph-failed"); } catch (err) { executorLog.error( `${task.id}: failed to park graph-failed task: ${err instanceof Error ? err.message : String(err)}`, @@ -6089,8 +6099,8 @@ export class TaskExecutor { if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) { await this.store.updateTask(task.id, { - status: "failed", - error: failureMessage, + status: "queued", + error: null, worktree: null, branch: null, sessionFile: null, @@ -6669,7 +6679,12 @@ export class TaskExecutor { executorLog.warn(`${task.id}: worktree removal failed during stuck-requeue cleanup (${worktreePath}): ${msg}`); } } - await this.store.updateTask(task.id, { status: "stuck-killed", worktree: null, branch: null }); + await this.store.updateTask(task.id, { + status: "queued", + error: null, + worktree: null, + branch: null, + }); if (latestTask.column !== "todo") { await this.store.moveTask(task.id, "todo", preserveProgress ? { preserveProgress: true } : undefined); executorLog.log(`${task.id} moved to todo for retry after stuck kill${preserveProgress ? " (progress preserved)" : ""}`); @@ -7576,8 +7591,8 @@ export class TaskExecutor { if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) { await this.store.updateTask(task.id, { - status: "failed", - error: errorMessage, + status: "queued", + error: null, taskDoneRetryCount: nextRequeueCount, }); await this.store.logEntry( @@ -8338,7 +8353,12 @@ export class TaskExecutor { executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErrMessage}`); } } - await this.store.updateTask(task.id, { status: "stuck-killed", worktree: null, branch: null }); + await this.store.updateTask(task.id, { + status: "queued", + error: null, + worktree: null, + branch: null, + }); // Only move to todo if not already there. Use the freshly-read // latestTask.column rather than the stale captured task.column — // the captured snapshot can be hours old and would race against @@ -9095,8 +9115,8 @@ export class TaskExecutor { const nextRequeueCount = priorRequeues + 1; if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) { await this.store.updateTask(task.id, { - status: "failed", - error: refusal.message, + status: "queued", + error: null, taskDoneRetryCount: nextRequeueCount, paused: false, pausedByAgentId: null, @@ -9190,8 +9210,8 @@ export class TaskExecutor { const nextRequeueCount = priorRequeues + 1; if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) { await store.updateTask(taskId, { - status: "failed", - error: refusalMessage, + status: "queued", + error: null, taskDoneRetryCount: nextRequeueCount, paused: false, pausedByAgentId: null, @@ -9248,8 +9268,8 @@ export class TaskExecutor { const nextRequeueCount = priorRequeues + 1; if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) { await store.updateTask(taskId, { - status: "failed", - error: refusalMessage, + status: "queued", + error: null, taskDoneRetryCount: nextRequeueCount, paused: false, pausedByAgentId: null, @@ -13364,7 +13384,12 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit taskId, `Force-requeued after stuck-kill: executor did not unwind within ${FORCE_REQUEUE_GRACE_MS / 1000}s (hung subprocess)${preserveProgress ? " — progress preserved" : ""}`, ); - await this.store.updateTask(taskId, { status: "stuck-killed", worktree: null, branch: null }); + await this.store.updateTask(taskId, { + status: "queued", + error: null, + worktree: null, + branch: null, + }); await this.store.moveTask(taskId, "todo", preserveProgress ? { preserveProgress: true } : undefined); // Remove from executing so the scheduler can re-dispatch normally. // The old Promise is still running but the executing guard is cleared so @@ -13974,19 +13999,19 @@ ${hasProgress : "Start with Step 0 (Preflight). Work through each step in order."} Use \`fn_task_update\` to report progress on every step transition. Use \`fn_task_log\` for important actions and decisions. -Use \`fn_task_create\` for truly separate follow-up work, not for fixes required to get tests, build, or typecheck back to green. +Use \`fn_task_create\` for truly separate follow-up work, including unrelated/pre-existing broad-suite failures. Commit at step boundaries: \`git commit -m "feat(${task.id}): complete Step N — "${sourceIssueRef ? ` -m "Ref: ${sourceIssueRef}"` : ""}${authorArg}\` The \`\` is required — replace it with a concrete 5–10 word description of what the step changed. When all steps are complete: call \`fn_task_done()\` If a build command is configured, run that exact command in this worktree before calling \`fn_task_done()\`. Treat a non-zero exit code as a blocking failure. Do not claim success without a real passing run. -Run the configured/full test suite and fix failures even when that requires edits outside the original File Scope. +Run impacted/package-scoped tests before completion. Run the configured workspace test command only when the task/workflow explicitly requires it or after impacted checks pass for final integration. If any broad command fails, classify the failure before editing: caused-by-this-task failures are blocking; unrelated or pre-existing failures should be logged and split into a follow-up instead of expanding this task. If the repo has a lint command (e.g. \`pnpm lint\`, \`npm run lint\`), run it before \`fn_task_done()\` and fix any failures it reports. If the repo has a typecheck command, run it before \`fn_task_done()\` and fix any failures it reports. -Use \`fn_task_create\` for truly separate follow-up work, not for fixes required to get tests, build, or typecheck back to green. +Use \`fn_task_create\` for truly separate follow-up work, including unrelated/pre-existing broad-suite failures. If lint is configured and failing, fix that too before completion. -**CRITICAL: Resolve ALL test failures (and any lint/typecheck failures) before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.`; +Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command.`; } /** diff --git a/packages/engine/src/run-verification-tool.ts b/packages/engine/src/run-verification-tool.ts index b12ab69e12..0605c6b4ae 100644 --- a/packages/engine/src/run-verification-tool.ts +++ b/packages/engine/src/run-verification-tool.ts @@ -18,7 +18,7 @@ * without a full agent session. */ -import { spawn } from "node:child_process"; +import { superviseSpawn, type SupervisedChild } from "@fusion/core"; import { existsSync, readdirSync, readFileSync } from "node:fs"; import { isAbsolute, join, relative } from "node:path"; import { Type, type Static } from "@earendil-works/pi-ai"; @@ -218,20 +218,8 @@ export function normalizeVerificationCommand(command: string, rootDir: string): 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. - } +function killVerificationProcess(supervised: SupervisedChild, signal: NodeJS.Signals): void { + supervised.kill(signal); } // --------------------------------------------------------------------------- @@ -376,10 +364,7 @@ export async function runVerificationCommand( const stderrBuf = createBuffer(); return new Promise((resolve) => { - // Use shell: true so Node picks the platform default — /bin/sh on POSIX, - // cmd.exe on Windows. SIGTERM/SIGKILL semantics still apply on POSIX; - // on Windows the kill signals map to TerminateProcess. - const child = spawn(command, { + const supervised = superviseSpawn(command, [], { cwd, stdio: ["ignore", "pipe", "pipe"], env: { @@ -390,8 +375,10 @@ export async function runVerificationCommand( COREPACK_ENABLE_DOWNLOAD_PROMPT: "0", }, shell: true, - detached: process.platform !== "win32", + killGraceMs: SIGKILL_GRACE_MS, + maxLifetimeMs: timeoutMs + SIGKILL_GRACE_MS + 1_000, }); + const child = supervised.child; let timedOut = false; let killed = false; @@ -417,14 +404,14 @@ export async function runVerificationCommand( executorLog.warn( `[fn_run_verification] hard timeout (${timeoutMs / 1000}s) — sending SIGTERM to: ${command}`, ); - killVerificationProcess(child, "SIGTERM"); + killVerificationProcess(supervised, "SIGTERM"); killTimer = setTimeout(() => { if (!settled) { executorLog.warn( `[fn_run_verification] SIGTERM ignored — sending SIGKILL to: ${command}`, ); - killVerificationProcess(child, "SIGKILL"); + killVerificationProcess(supervised, "SIGKILL"); killed = true; } }, SIGKILL_GRACE_MS); @@ -432,7 +419,7 @@ export async function runVerificationCommand( // ── stdout ─────────────────────────────────────────────────────────────── let stdoutRemainder = ""; - child.stdout.on("data", (chunk: Buffer) => { + child.stdout?.on("data", (chunk: Buffer) => { const text = stdoutRemainder + chunk.toString("utf8"); const lines = text.split("\n"); stdoutRemainder = lines.pop() ?? ""; @@ -447,7 +434,7 @@ export async function runVerificationCommand( // ── stderr ─────────────────────────────────────────────────────────────── let stderrRemainder = ""; - child.stderr.on("data", (chunk: Buffer) => { + child.stderr?.on("data", (chunk: Buffer) => { const text = stderrRemainder + chunk.toString("utf8"); const lines = text.split("\n"); stderrRemainder = lines.pop() ?? ""; diff --git a/packages/engine/src/sandbox/__tests__/native.test.ts b/packages/engine/src/sandbox/__tests__/native.test.ts index 441f8ed74f..2104c5754b 100644 --- a/packages/engine/src/sandbox/__tests__/native.test.ts +++ b/packages/engine/src/sandbox/__tests__/native.test.ts @@ -1,10 +1,24 @@ +import { access, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { cwd } from "node:process"; +import { setTimeout as delay } from "node:timers/promises"; -import { describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { NativeSandboxBackend } from "../native.js"; describe("NativeSandboxBackend", () => { + let tempDir: string; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "fusion-native-sandbox-")); + }); + + afterEach(async () => { + await rm(tempDir, { recursive: true, force: true }); + }); + it("returns stdout on success", async () => { const backend = new NativeSandboxBackend(); const result = await backend.run("node -e 'process.stdout.write(\"ok\")'", { @@ -34,6 +48,67 @@ describe("NativeSandboxBackend", () => { expect(result.signal).toBe("SIGTERM"); }); + it.skipIf(process.platform === "win32")("times out and terminates descendant processes in the command process group", async () => { + const backend = new NativeSandboxBackend(); + const markerPath = join(tempDir, "descendant-survived.txt"); + const parentScriptPath = join(tempDir, "spawn-descendant.cjs"); + await writeFile( + parentScriptPath, + ` +const { spawn } = require("node:child_process"); +spawn(process.execPath, [ + "-e", + "setTimeout(() => require('node:fs').writeFileSync(process.env.MARKER, 'survived'), 450)", +], { + env: { ...process.env, MARKER: process.argv[2] }, + stdio: "ignore", +}).unref(); +setInterval(() => {}, 1000); +`, + "utf-8", + ); + + const result = await backend.run( + `${JSON.stringify(process.execPath)} ${JSON.stringify(parentScriptPath)} ${JSON.stringify(markerPath)}`, + { + cwd: tempDir, + timeoutMs: 75, + maxBuffer: 1024 * 1024, + encoding: "utf-8", + }, + ); + + expect(result.timedOut).toBe(true); + await delay(700); + await expect(access(markerPath)).rejects.toThrow(); + }); + + it.skipIf(process.platform === "win32")("cleans up background children after successful commands", async () => { + const backend = new NativeSandboxBackend(); + const markerPath = join(tempDir, "success-descendant-survived.txt"); + const parentScript = [ + "const { spawn } = require('node:child_process');", + `spawn(process.execPath, ['-e', ${JSON.stringify("setTimeout(() => require('node:fs').writeFileSync(process.env.MARKER, 'survived'), 450)")}], { env: { ...process.env, MARKER: process.env.MARKER }, stdio: 'ignore' }).unref();`, + "process.stdout.write('parent-done');", + ].join(" "); + + const result = await backend.run( + `${JSON.stringify(process.execPath)} -e ${JSON.stringify(parentScript)}`, + { + cwd: tempDir, + timeoutMs: 5_000, + maxBuffer: 1024 * 1024, + encoding: "utf-8", + env: { ...process.env, MARKER: markerPath }, + }, + ); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toBe("parent-done"); + await delay(700); + await expect(access(markerPath)).rejects.toThrow(); + }); + it("maps non-zero exits", async () => { const backend = new NativeSandboxBackend(); const result = await backend.run("node -e 'process.stderr.write(\"fail\"); process.exit(7)'", { diff --git a/packages/engine/src/sandbox/native.ts b/packages/engine/src/sandbox/native.ts index 58dd50dda0..cf580557ed 100644 --- a/packages/engine/src/sandbox/native.ts +++ b/packages/engine/src/sandbox/native.ts @@ -1,5 +1,3 @@ -import { exec } from "node:child_process"; -import { promisify } from "node:util"; import { superviseSpawn } from "@fusion/core"; import type { @@ -12,7 +10,8 @@ import type { SandboxStreamingResult, } from "./types.js"; -const execAsync = promisify(exec); +const FORCE_KILL_DELAY_MS = 5_000; +const NORMAL_CLEANUP_FORCE_KILL_DELAY_MS = 500; export class NativeSandboxBackend implements SandboxBackend { capabilities(): SandboxCapabilities { @@ -30,48 +29,110 @@ export class NativeSandboxBackend implements SandboxBackend { } async run(command: string, options: SandboxRunOptions): Promise { - try { - const execOptions: Parameters[1] = { - cwd: options.cwd, - timeout: options.timeoutMs, - maxBuffer: options.maxBuffer, - ...(options.encoding !== undefined && { encoding: options.encoding }), - ...(typeof options.shell === "string" && { shell: options.shell }), - ...(options.env !== undefined && { env: options.env }), - ...(options.signal !== undefined && { signal: options.signal }), - }; - const { stdout, stderr } = await execAsync(command, execOptions); - + if (options.signal?.aborted) { return { - stdout: stdout?.toString?.() ?? "", - stderr: stderr?.toString?.() ?? "", - exitCode: 0, + stdout: "", + stderr: "", + exitCode: null, signal: null, timedOut: false, bufferExceeded: false, - }; - } catch (error) { - const errObj = error as Record; - const code = errObj.code; - const status = typeof errObj.status === "number" ? errObj.status : null; - const exitCode = typeof code === "number" ? code : status; - const message = String(errObj.message ?? ""); - - return { - stdout: typeof (errObj.stdout as { toString?: unknown })?.toString === "function" ? String(errObj.stdout) : "", - stderr: typeof (errObj.stderr as { toString?: unknown })?.toString === "function" ? String(errObj.stderr) : "", - exitCode, - signal: (errObj.signal as NodeJS.Signals | null | undefined) ?? null, - bufferExceeded: - code === "ENOBUFS" - || code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER" - || message.includes("maxBuffer"), - timedOut: - code === "ETIMEDOUT" - || (errObj.killed === true && (errObj.signal === "SIGTERM" || message.includes("timed out"))), - spawnError: code === "ENOENT" || code === "EACCES" ? (error as Error) : undefined, + spawnError: new Error("Command aborted before start"), }; } + + return await new Promise((resolve) => { + const supervised = superviseSpawn(command, [], { + cwd: options.cwd, + shell: options.shell ?? true, + stdio: ["ignore", "pipe", "pipe"], + ...(options.env !== undefined && { env: options.env }), + maxLifetimeMs: options.timeoutMs > 0 ? options.timeoutMs + FORCE_KILL_DELAY_MS + 1_000 : undefined, + }); + const child = supervised.child; + + const encoding = options.encoding ?? "utf-8"; + let stdout = ""; + let stderr = ""; + let bufferExceeded = false; + let timedOut = false; + let settled = false; + let forceKillTimer: ReturnType | null = null; + + const killTree = (signal: NodeJS.Signals): void => { + supervised.kill(signal); + }; + + const scheduleForceKill = (delayMs = FORCE_KILL_DELAY_MS): void => { + if (forceKillTimer) return; + forceKillTimer = setTimeout(() => { + killTree("SIGKILL"); + }, delayMs); + forceKillTimer.unref(); + }; + + const killTreeForCommandFailure = (): void => { + killTree("SIGTERM"); + scheduleForceKill(); + }; + + const append = (current: string, chunk: Buffer): string => { + if (bufferExceeded) return current; + const text = chunk.toString(encoding); + if (current.length + text.length <= options.maxBuffer) { + return current + text; + } + bufferExceeded = true; + const remaining = Math.max(0, options.maxBuffer - current.length); + killTreeForCommandFailure(); + return current + text.slice(0, remaining); + }; + + const timeout = options.timeoutMs > 0 + ? setTimeout(() => { + timedOut = true; + killTreeForCommandFailure(); + }, options.timeoutMs) + : null; + timeout?.unref(); + + const onAbort = (): void => { + killTreeForCommandFailure(); + }; + options.signal?.addEventListener("abort", onAbort, { once: true }); + + const finish = (spawnError: Error | null, exitCode: number | null, signal: NodeJS.Signals | null): void => { + if (settled) return; + settled = true; + if (timeout) clearTimeout(timeout); + if (forceKillTimer) clearTimeout(forceKillTimer); + options.signal?.removeEventListener("abort", onAbort); + + if (!spawnError) { + killTree("SIGTERM"); + scheduleForceKill(NORMAL_CLEANUP_FORCE_KILL_DELAY_MS); + } + + resolve({ + stdout, + stderr, + exitCode, + signal, + timedOut, + bufferExceeded, + ...(spawnError ? { spawnError } : {}), + }); + }; + + child.stdout?.on("data", (chunk: Buffer) => { + stdout = append(stdout, chunk); + }); + child.stderr?.on("data", (chunk: Buffer) => { + stderr = append(stderr, chunk); + }); + child.on("error", (error) => finish(error, null, null)); + child.on("close", (code, signal) => finish(null, code, signal)); + }); } async runStreaming(command: string, options: SandboxRunStreamingOptions): Promise { @@ -105,28 +166,31 @@ export class NativeSandboxBackend implements SandboxBackend { let timedOut = false; let aborted = false; let settled = false; + let forceKillTimer: ReturnType | null = null; const killTree = (sig: NodeJS.Signals) => { supervised.kill(sig); }; + const scheduleForceKill = (delayMs = FORCE_KILL_DELAY_MS): void => { + if (forceKillTimer) return; + forceKillTimer = setTimeout(() => { + killTree("SIGKILL"); + }, delayMs); + forceKillTimer.unref(); + }; + const timer = setTimeout(() => { timedOut = true; killTree("SIGTERM"); - setTimeout(() => { - if (settled) return; - killTree("SIGKILL"); - }, 5_000).unref(); + scheduleForceKill(); }, options.timeout); timer.unref(); const onAbort = () => { aborted = true; killTree("SIGTERM"); - setTimeout(() => { - if (settled) return; - killTree("SIGKILL"); - }, 5_000).unref(); + scheduleForceKill(); }; options.signal?.addEventListener("abort", onAbort, { once: true }); @@ -154,6 +218,7 @@ export class NativeSandboxBackend implements SandboxBackend { if (settled) return; settled = true; clearTimeout(timer); + if (forceKillTimer) clearTimeout(forceKillTimer); options.signal?.removeEventListener("abort", onAbort); if (aborted) { @@ -172,6 +237,8 @@ export class NativeSandboxBackend implements SandboxBackend { } if (code === 0) { + killTree("SIGTERM"); + scheduleForceKill(NORMAL_CLEANUP_FORCE_KILL_DELAY_MS); resolve({ outcome: "success", stdout, diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index c099ccfcdb..188242ffc1 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -315,6 +315,40 @@ interface ConcurrencyGateDiagnostic { perColumnGates?: PerColumnCapacityGate[]; } +const IDLE_SEMAPHORE_LEAK_REPAIR_MS = 5_000; + +function persistedTopLevelAgentSlots(tasks: Task[]): number { + return tasks.filter((task) => ( + task.column === "in-progress" + || (task.column === "triage" && task.status === "planning" && !task.paused) + || (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? ""))) + )).length; +} + +function recoverIdleSemaphoreLeak( + semaphore: AgentSemaphore | undefined, + tasks: Task[], + source: string, + candidateSinceMs: number | null, +): number | null { + if (!semaphore) return null; + const persistedActive = persistedTopLevelAgentSlots(tasks); + if (persistedActive !== 0 || semaphore.activeCount <= 0) return null; + + const now = Date.now(); + if (candidateSinceMs === null) return now; + if (now - candidateSinceMs < IDLE_SEMAPHORE_LEAK_REPAIR_MS) return candidateSinceMs; + + const result = semaphore.reconcileActiveCount(0); + if (result.changed) { + schedulerLog.warn( + `${source}: recovered stale semaphore active count ${result.before} -> ${result.after} ` + + "(no persisted in-progress/planning/review agent work)", + ); + } + return null; +} + function computeConcurrencyGateDiagnostic(params: { agentSlots: number; maxConcurrent: number; @@ -495,6 +529,7 @@ export class Scheduler { private lastStaleTaskReportAt = 0; private lastBacklogPressureReportAt = 0; private lastUnlinkedMissionsAdvisoryReportAt = 0; + private idleSemaphoreLeakCandidateSince: number | null = null; private readonly lastHighOverlapFanoutWarningKey = new Map(); /** @@ -1208,6 +1243,12 @@ export class Scheduler { const settings = await this.store.getSettings(); const maxConcurrent = settings.maxConcurrent ?? this.options.maxConcurrent ?? 2; const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4; + this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak( + this.options.semaphore, + tasks, + "scheduler", + this.idleSemaphoreLeakCandidateSince, + ); // Refresh the poll interval if the persisted setting has changed this.refreshPollInterval(settings.pollIntervalMs); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 57c7798f69..180c560d9b 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -634,6 +634,7 @@ export class TriageProcessor { private processingSince = new Map(); private wasGlobalPaused = false; private wasEnginePaused = false; + private idleSemaphoreLeakCandidateSince: number | null = null; /** Active agent sessions per task, used to terminate on pause. */ private activeSessions = new Map void }>(); /** @@ -997,6 +998,31 @@ export class TriageProcessor { // Fetch all tasks (not just triage) to count active agents across columns. const allTasks = await this.store.listTasks({ slim: true, includeArchived: false }); const now = Date.now(); + + if (this.options.semaphore) { + const persistedActive = allTasks.filter((task) => ( + task.column === "in-progress" + || (task.column === "triage" && task.status === "planning" && !task.paused) + || (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? ""))) + )).length; + if (persistedActive === 0 && this.options.semaphore.activeCount > 0 && this.processing.size === 0) { + if (this.idleSemaphoreLeakCandidateSince === null) { + this.idleSemaphoreLeakCandidateSince = now; + } else if (now - this.idleSemaphoreLeakCandidateSince >= 5_000) { + const result = this.options.semaphore.reconcileActiveCount(0); + if (result.changed) { + planLog.warn( + `triage: recovered stale semaphore active count ${result.before} -> ${result.after} ` + + "(no persisted in-progress/planning/review agent work)", + ); + } + this.idleSemaphoreLeakCandidateSince = null; + } + } else { + this.idleSemaphoreLeakCandidateSince = null; + } + } + const eligibleTriageTasks = allTasks.filter( (t) => t.column === "triage" && !this.processing.has(t.id) && !t.paused // Skip tasks awaiting manual plan approval — they should not be auto-discovered @@ -1043,8 +1069,17 @@ export class TriageProcessor { const maxToStart = Math.min(perProjectAvailable, semaphoreAvailable); if (maxToStart <= 0 && triageTasks.length > 0) { + const semaphoreSnapshot = this.options.semaphore?.snapshot(); + const semaphoreDetail = semaphoreSnapshot + ? `, semaphore active=${semaphoreSnapshot.activeCount}/${semaphoreSnapshot.limit}, available=${semaphoreSnapshot.availableCount}, waiting=${semaphoreSnapshot.waitingCount}` + : ", semaphore unavailable"; + const processingIds = [...this.processing].slice(0, 5); + const eligibleIds = triageTasks.slice(0, 5).map((t) => t.id); + const blockedBy = perProjectAvailable <= 0 ? "triage concurrency" : "global semaphore"; planLog.log( - `Plan throttled: ${activeAgents} planning agents, limit ${maxTriageConcurrent}`, + `Plan throttled by ${blockedBy}: eligible=${triageTasks.length} [${eligibleIds.join(", ")}], ` + + `planning=${activeAgents}/${maxTriageConcurrent}, processing=${this.processing.size}` + + `${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}${semaphoreDetail}`, ); } From f0fa75b270d440a360fedaf1554a99a2a8156a7b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 18:31:58 -0700 Subject: [PATCH 2/2] fix(FN-6043): address stuck task PR feedback - align triage prompt templates on impacted-first verification - guard verification max lifetime when timeout is disabled - share idle semaphore leak recovery across scheduler and triage Fusion-Task-Id: FN-6043 --- packages/core/src/agent-prompts.ts | 4 +- .../engine/src/__tests__/concurrency.test.ts | 62 ++++++++++++++++++- .../src/__tests__/executor-core.test.ts | 6 +- .../src/__tests__/merger-post-merge.test.ts | 10 +-- packages/engine/src/concurrency.ts | 55 ++++++++++++++++ packages/engine/src/run-verification-tool.ts | 2 +- packages/engine/src/scheduler.ts | 32 +++------- packages/engine/src/triage.ts | 45 ++++++-------- 8 files changed, 155 insertions(+), 61 deletions(-) diff --git a/packages/core/src/agent-prompts.ts b/packages/core/src/agent-prompts.ts index 50c6054db8..739895c07b 100644 --- a/packages/core/src/agent-prompts.ts +++ b/packages/core/src/agent-prompts.ts @@ -280,7 +280,7 @@ For bug-fix tasks, paste and fill in this checklist in the \`## Surface Enumerat ### Step {N-1}: Testing & Verification -> ZERO test failures allowed. Full test suite as quality gate. +> ZERO failures allowed for checks required by this task's quality gates. Run impacted/package-scoped verification first; run workspace-wide suites only when the task or workflow explicitly requires them, or during final integration after impacted checks pass. > If keeping lint/tests/build/typecheck green requires edits outside the initial File Scope, make those fixes as part of this task. - [ ] Run lint check (\`pnpm lint\`) @@ -345,7 +345,7 @@ tests. Manual verification is NOT a test. - For bug fixes, the spec MUST include a \`## Surface Enumeration\` section. During self-review via \`fn_review_spec()\`, treat a missing section on a bug-fix spec as a blocking REVISE. - For bug fixes, populate \`## Surface Enumeration\` with this checklist from \`docs/testing.md\`: providers/bridges/execution paths; desktop + mobile breakpoints/platforms; empty/undefined/duplicate/populated data states; shared hooks/components/modules/helpers. - For bug fixes, regression tests must assert the invariant across all known surfaces — enumerate every provider/bridge, desktop + mobile breakpoints, and empty/undefined/populated data states — not just the reported repro (see FN-5787/FN-5789/FN-5803 and FN-5751) -- The final Testing step runs lint, the FULL test suite, and project typecheck when the repo exposes one +- The final Testing step runs lint, impacted/package-scoped tests first, and project typecheck when the repo exposes one. Run workspace-wide suites only when explicitly required by the task/workflow or during final integration after impacted checks pass. - Specs must instruct executors to fix lint failures and quality-gate failures directly, even when the required edits extend beyond the original File Scope - If the project has no test framework, the Testing step must include setting one up as part of this task (not just skipping tests) diff --git a/packages/engine/src/__tests__/concurrency.test.ts b/packages/engine/src/__tests__/concurrency.test.ts index 5276c46af3..9d8e51ba20 100644 --- a/packages/engine/src/__tests__/concurrency.test.ts +++ b/packages/engine/src/__tests__/concurrency.test.ts @@ -1,5 +1,12 @@ import { describe, it, expect, vi } from "vitest"; -import { AgentSemaphore, PRIORITY_MERGE, PRIORITY_EXECUTE, PRIORITY_SPECIFY } from "../concurrency.js"; +import type { Task } from "@fusion/core"; +import { + AgentSemaphore, + PRIORITY_MERGE, + PRIORITY_EXECUTE, + PRIORITY_SPECIFY, + recoverIdleSemaphoreLeakCandidate, +} from "../concurrency.js"; describe("AgentSemaphore", () => { it("allows immediate acquire when under limit", async () => { @@ -184,6 +191,59 @@ describe("AgentSemaphore", () => { sem.release(); }); + it("recovers idle semaphore leaks only after a stable persisted-idle window", async () => { + const sem = new AgentSemaphore(2); + await sem.acquire(); + const tasks: Task[] = []; + + const first = recoverIdleSemaphoreLeakCandidate({ + semaphore: sem, + tasks, + candidateSinceMs: null, + nowMs: 1_000, + }); + expect(first).toEqual({ candidateSinceMs: 1_000 }); + expect(sem.activeCount).toBe(1); + + const early = recoverIdleSemaphoreLeakCandidate({ + semaphore: sem, + tasks, + candidateSinceMs: first.candidateSinceMs, + nowMs: 5_000, + }); + expect(early).toEqual({ candidateSinceMs: 1_000 }); + expect(sem.activeCount).toBe(1); + + const repaired = recoverIdleSemaphoreLeakCandidate({ + semaphore: sem, + tasks, + candidateSinceMs: early.candidateSinceMs, + nowMs: 6_001, + }); + expect(repaired).toEqual({ + candidateSinceMs: null, + reconciliation: { before: 1, after: 0, changed: true }, + }); + expect(sem.activeCount).toBe(0); + }); + + it("does not recover while callers report in-flight work not yet persisted", async () => { + const sem = new AgentSemaphore(2); + await sem.acquire(); + + const result = recoverIdleSemaphoreLeakCandidate({ + semaphore: sem, + tasks: [], + candidateSinceMs: Date.now() - 6_000, + inFlightCount: 1, + nowMs: Date.now(), + }); + + expect(result).toEqual({ candidateSinceMs: null }); + expect(sem.activeCount).toBe(1); + sem.release(); + }); + it("run() gates concurrent calls", async () => { const sem = new AgentSemaphore(2); let concurrent = 0; diff --git a/packages/engine/src/__tests__/executor-core.test.ts b/packages/engine/src/__tests__/executor-core.test.ts index 5729bf16c1..72f2289d3d 100644 --- a/packages/engine/src/__tests__/executor-core.test.ts +++ b/packages/engine/src/__tests__/executor-core.test.ts @@ -990,13 +990,13 @@ describe("TaskExecutor messaging tools", () => { }); // Fast mode should still enforce fn_task_done requirement. - // After 3 retries it should fail and requeue. + // While retry budget remains, failures requeue instead of becoming terminal. expect(onError).toHaveBeenCalled(); expect(store.updateTask).toHaveBeenCalledWith( "FN-001", expect.objectContaining({ - status: "failed", - error: "Agent finished without calling fn_task_done (after 3 retries)", + status: "queued", + error: null, taskDoneRetryCount: 1, }), ); diff --git a/packages/engine/src/__tests__/merger-post-merge.test.ts b/packages/engine/src/__tests__/merger-post-merge.test.ts index b55f4d598f..ab06b2704a 100644 --- a/packages/engine/src/__tests__/merger-post-merge.test.ts +++ b/packages/engine/src/__tests__/merger-post-merge.test.ts @@ -149,13 +149,14 @@ import { } from "../merger.js"; import { mergerLog } from "../logger.js"; import { createFnAgent } from "../pi.js"; -import { execSync, exec } from "node:child_process"; +import { execSync, exec, spawn } from "node:child_process"; import * as core from "@fusion/core"; import { type TaskStore, type Task, type MergeResult, DEFAULT_SETTINGS } from "@fusion/core"; const mockedCreateFnAgent = vi.mocked(createFnAgent); const mockedExecSync = vi.mocked(execSync); const mockedExec = vi.mocked(exec); +const mockedSpawn = vi.mocked(spawn); const { existsSync: mockedExistsSyncRaw, readFileSync: mockedReadFileSyncRaw } = await import("node:fs"); const mockedExistsSync = vi.mocked(mockedExistsSyncRaw); const mockedReadFileSync = vi.mocked(mockedReadFileSyncRaw); @@ -707,9 +708,9 @@ describe("aiMergeTask — post-merge workflow steps", () => { const result = await aiMergeTask(store, "/tmp/root", "FN-050"); - const scriptExecCall = mockedExec.mock.calls.find((call: any) => String(call[0]) === "pnpm build"); - expect(scriptExecCall).toBeDefined(); - expect(scriptExecCall?.[1]?.cwd).toMatch(/\.worktrees\/post-merge-FN-050-[a-z0-9]+/); + const scriptSpawnCall = mockedSpawn.mock.calls.find((call: any) => String(call[0]) === "pnpm build"); + expect(scriptSpawnCall).toBeDefined(); + expect(scriptSpawnCall?.[2]?.cwd).toMatch(/\.worktrees\/post-merge-FN-050-[a-z0-9]+/); expect(result.merged).toBe(true); expect(store.moveTask).toHaveBeenCalledWith("FN-050", "done"); @@ -906,4 +907,3 @@ describe("aiMergeTask — post-merge workflow steps", () => { // ── Merge Details Collection Tests ───────────────────────────────────── - diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 19d9d56ed1..651d3032b2 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -1,3 +1,5 @@ +import type { Task } from "@fusion/core"; + /** Priority level for merge agents — served first. */ export const PRIORITY_MERGE = 2; /** Priority level for execution agents — served after merge, before specify. */ @@ -11,6 +13,59 @@ interface PriorityWaiter { resolve: () => void; } +export const IDLE_SEMAPHORE_LEAK_REPAIR_MS = 5_000; + +export function persistedTopLevelAgentSlots(tasks: Task[]): number { + return tasks.filter((task) => ( + task.column === "in-progress" + || (task.column === "triage" && task.status === "planning" && !task.paused) + || (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? ""))) + )).length; +} + +export interface IdleSemaphoreLeakRecoveryResult { + candidateSinceMs: number | null; + reconciliation?: { before: number; after: number; changed: boolean }; +} + +export function recoverIdleSemaphoreLeakCandidate(params: { + semaphore: AgentSemaphore | undefined; + tasks: Task[]; + candidateSinceMs: number | null; + inFlightCount?: number; + nowMs?: number; + repairAfterMs?: number; +}): IdleSemaphoreLeakRecoveryResult { + const { + semaphore, + tasks, + candidateSinceMs, + inFlightCount = 0, + nowMs = Date.now(), + repairAfterMs = IDLE_SEMAPHORE_LEAK_REPAIR_MS, + } = params; + + if (!semaphore) return { candidateSinceMs: null }; + + const persistedActive = persistedTopLevelAgentSlots(tasks); + if (persistedActive !== 0 || semaphore.activeCount <= 0 || inFlightCount > 0) { + return { candidateSinceMs: null }; + } + + if (candidateSinceMs === null) { + return { candidateSinceMs: nowMs }; + } + + if (nowMs - candidateSinceMs < repairAfterMs) { + return { candidateSinceMs }; + } + + return { + candidateSinceMs: null, + reconciliation: semaphore.reconcileActiveCount(0), + }; +} + /** * A concurrency semaphore that gates all agentic activities (triage specification, * task execution, and merge operations) behind a shared slot limit. diff --git a/packages/engine/src/run-verification-tool.ts b/packages/engine/src/run-verification-tool.ts index 0605c6b4ae..3bacdc4de2 100644 --- a/packages/engine/src/run-verification-tool.ts +++ b/packages/engine/src/run-verification-tool.ts @@ -376,7 +376,7 @@ export async function runVerificationCommand( }, shell: true, killGraceMs: SIGKILL_GRACE_MS, - maxLifetimeMs: timeoutMs + SIGKILL_GRACE_MS + 1_000, + maxLifetimeMs: timeoutMs > 0 ? timeoutMs + SIGKILL_GRACE_MS + 1_000 : undefined, }); const child = supervised.child; diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 188242ffc1..f3cb27743a 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -17,7 +17,7 @@ import { import { existsSync } from "node:fs"; import { readFile } from "node:fs/promises"; import { join } from "node:path"; -import type { AgentSemaphore } from "./concurrency.js"; +import { recoverIdleSemaphoreLeakCandidate, type AgentSemaphore } from "./concurrency.js"; import { planTaskWorktreePath, resolveTaskWorkingBranch } from "./worktree-names.js"; import { schedulerLog } from "./logger.js"; import { type PrMonitor, type PrComment } from "./pr-monitor.js"; @@ -315,38 +315,24 @@ interface ConcurrencyGateDiagnostic { perColumnGates?: PerColumnCapacityGate[]; } -const IDLE_SEMAPHORE_LEAK_REPAIR_MS = 5_000; - -function persistedTopLevelAgentSlots(tasks: Task[]): number { - return tasks.filter((task) => ( - task.column === "in-progress" - || (task.column === "triage" && task.status === "planning" && !task.paused) - || (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? ""))) - )).length; -} - function recoverIdleSemaphoreLeak( semaphore: AgentSemaphore | undefined, tasks: Task[], source: string, candidateSinceMs: number | null, ): number | null { - if (!semaphore) return null; - const persistedActive = persistedTopLevelAgentSlots(tasks); - if (persistedActive !== 0 || semaphore.activeCount <= 0) return null; - - const now = Date.now(); - if (candidateSinceMs === null) return now; - if (now - candidateSinceMs < IDLE_SEMAPHORE_LEAK_REPAIR_MS) return candidateSinceMs; - - const result = semaphore.reconcileActiveCount(0); - if (result.changed) { + const result = recoverIdleSemaphoreLeakCandidate({ + semaphore, + tasks, + candidateSinceMs, + }); + if (result.reconciliation?.changed) { schedulerLog.warn( - `${source}: recovered stale semaphore active count ${result.before} -> ${result.after} ` + + `${source}: recovered stale semaphore active count ${result.reconciliation.before} -> ${result.reconciliation.after} ` + "(no persisted in-progress/planning/review agent work)", ); } - return null; + return result.candidateSinceMs; } function computeConcurrencyGateDiagnostic(params: { diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 180c560d9b..d8ac33a5ba 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -42,7 +42,7 @@ import { formatExternalIntegrationEvidenceDiagnostic, } from "./spec-validation/external-integration-evidence.js"; import { buildSessionSkillContext } from "./session-skill-context.js"; -import { PRIORITY_SPECIFY, type AgentSemaphore } from "./concurrency.js"; +import { PRIORITY_SPECIFY, recoverIdleSemaphoreLeakCandidate, type AgentSemaphore } from "./concurrency.js"; import { AgentLogger } from "./agent-logger.js"; import { resolveAgentInstructions, @@ -169,11 +169,11 @@ For bug-fix tasks, paste and fill in this checklist in the \`## Surface Enumerat ### Step {N-1}: Testing & Verification -> ZERO test failures allowed. Full test suite as quality gate. +> ZERO failures allowed for checks required by this task's quality gates. Run impacted/package-scoped verification first; run workspace-wide suites only when the task or workflow explicitly requires them, or during final integration after impacted checks pass. > If keeping lint/tests/build/typecheck green requires edits outside the initial File Scope, make those fixes as part of this task. - [ ] Run lint check (\`pnpm lint\`) -- [ ] Run full test suite +- [ ] Run impacted tests - [ ] Run project typecheck if available - [ ] Fix all failures - [ ] Build passes @@ -241,7 +241,7 @@ tests. Manual verification is NOT a test. - For bug fixes, the spec MUST include a \`## Surface Enumeration\` section. During self-review via \`fn_review_spec()\`, treat a missing section on a bug-fix spec as a blocking REVISE. - For bug fixes, populate \`## Surface Enumeration\` with this checklist from \`docs/testing.md\`: providers/bridges/execution paths; desktop + mobile breakpoints/platforms; empty/undefined/duplicate/populated data states; shared hooks/components/modules/helpers. - For bug fixes, regression tests must assert the invariant across all known surfaces — enumerate every provider/bridge, desktop + mobile breakpoints, and empty/undefined/populated data states — not just the reported repro (see FN-5787/FN-5789/FN-5803 and FN-5751) -- The final Testing step runs lint, the FULL test suite, and project typecheck when the repo exposes one +- The final Testing step runs lint, impacted/package-scoped tests first, and project typecheck when the repo exposes one. Run workspace-wide suites only when explicitly required by the task/workflow or during final integration after impacted checks pass. - Specs must instruct executors to fix lint failures and quality-gate failures directly, even when the required edits extend beyond the original File Scope - If the project has no test framework, the Testing step must include setting one up as part of this task (not just skipping tests) @@ -473,11 +473,11 @@ For bug-fix tasks, paste and fill in this checklist in the \`## Surface Enumerat ### Step {N-1}: Testing & Verification -> ZERO test failures allowed. Full test suite as quality gate. +> ZERO failures allowed for checks required by this task's quality gates. Run impacted/package-scoped verification first; run workspace-wide suites only when the task or workflow explicitly requires them, or during final integration after impacted checks pass. > If keeping lint/tests/build/typecheck green requires edits outside the initial File Scope, make those fixes as part of this task. - [ ] Run lint check (\`pnpm lint\`) -- [ ] Run full test suite +- [ ] Run impacted tests - [ ] Run project typecheck if available - [ ] Build passes @@ -1000,27 +1000,20 @@ export class TriageProcessor { const now = Date.now(); if (this.options.semaphore) { - const persistedActive = allTasks.filter((task) => ( - task.column === "in-progress" - || (task.column === "triage" && task.status === "planning" && !task.paused) - || (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? ""))) - )).length; - if (persistedActive === 0 && this.options.semaphore.activeCount > 0 && this.processing.size === 0) { - if (this.idleSemaphoreLeakCandidateSince === null) { - this.idleSemaphoreLeakCandidateSince = now; - } else if (now - this.idleSemaphoreLeakCandidateSince >= 5_000) { - const result = this.options.semaphore.reconcileActiveCount(0); - if (result.changed) { - planLog.warn( - `triage: recovered stale semaphore active count ${result.before} -> ${result.after} ` + - "(no persisted in-progress/planning/review agent work)", - ); - } - this.idleSemaphoreLeakCandidateSince = null; - } - } else { - this.idleSemaphoreLeakCandidateSince = null; + const result = recoverIdleSemaphoreLeakCandidate({ + semaphore: this.options.semaphore, + tasks: allTasks, + candidateSinceMs: this.idleSemaphoreLeakCandidateSince, + inFlightCount: this.processing.size, + nowMs: now, + }); + if (result.reconciliation?.changed) { + planLog.warn( + `triage: recovered stale semaphore active count ${result.reconciliation.before} -> ${result.reconciliation.after} ` + + "(no persisted in-progress/planning/review agent work)", + ); } + this.idleSemaphoreLeakCandidateSince = result.candidateSinceMs; } const eligibleTriageTasks = allTasks.filter(