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(