diff --git a/.changeset/fix-pause-abort-leak-storm.md b/.changeset/fix-pause-abort-leak-storm.md new file mode 100644 index 0000000000..3d66855ff3 --- /dev/null +++ b/.changeset/fix-pause-abort-leak-storm.md @@ -0,0 +1,8 @@ +--- +"@runfusion/fusion": patch +--- + +Fix the global pause/resume failure mode that stalled the board: a pause-abort that left a task back in `todo` was parked `status:"failed"` ("operator action required") and leaked its in-memory worktree slot, producing an instant re-fail retry storm and concurrency-starving the whole queue. + +- Root cause: `handleGraphFailure` now treats a pause-abort that has re-queued a task to `todo` as benign (FN-6782) — it no longer parks it failed, clears the `pausedAborted` marker so the next dispatch starts clean, and releases the leaked worktree slot. +- Auto-recovery: a new `recoverPausedAbortFailures` self-healing sweep clears any pause-abort park (`status:"failed"` with "operator action required") still on the board and requeues it for normal scheduling, so the board self-heals without operator intervention. diff --git a/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts b/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts new file mode 100644 index 0000000000..9c5d55ec1b --- /dev/null +++ b/packages/engine/src/__tests__/executor-paused-abort-todo-benign.test.ts @@ -0,0 +1,98 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import "./executor-test-helpers.js"; +import { TaskExecutor } from "../executor.js"; +import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js"; +import type { TaskDetail } from "@fusion/core"; + +const now = "2026-06-20T00:00:00.000Z"; + +function makeTask(overrides: Partial = {}): TaskDetail { + return { + id: "FN-6782-T", + title: "pause-abort benign todo repro", + description: "Reproduces FN-6782 benign requeue-to-todo classification", + column: "todo", + dependencies: [], + steps: [{ name: "Implement", status: "pending" }], + currentStep: 0, + log: [], + branch: null, + baseBranch: "main", + worktree: "/tmp/fusion-fn-6782-t", + status: null, + error: null, + paused: false, + userPaused: false, + autoMerge: true, + mergeRetries: 0, + createdAt: now, + updatedAt: now, + ...overrides, + } as TaskDetail; +} + +function makeHarness(taskOverrides: Partial = {}) { + const store = createMockStore(); + const task = makeTask(taskOverrides); + store.getTask.mockResolvedValue(task); + store.getSettings.mockResolvedValue({ + maxConcurrent: 2, + maxWorktrees: 4, + pollIntervalMs: 15000, + autoMerge: true, + maxAutoMergeRetries: 3, + }); + const executor = new TaskExecutor(store, "/tmp/test", {}); + (executor as any).markPausedAborted(task.id, "hard-cancel"); + return { store, task, executor }; +} + +async function invokeGraphFailure(executor: TaskExecutor, task: TaskDetail) { + await (executor as any).handleGraphFailure(task, { + disposition: "failed", + outcome: "failure", + visitedNodeIds: ["plan", "execute"], + context: {}, + }); +} + +function logText(store: ReturnType): string { + return store.logEntry.mock.calls.map((call: unknown[]) => call[1]).join("\n"); +} + +describe("pause-abort benign requeue-to-todo (FN-6782)", () => { + beforeEach(() => { + resetExecutorMocks(); + }); + + it("does NOT park a todo-column pause-abort as failed (no retry storm)", async () => { + const { store, task, executor } = makeHarness({ column: "todo" }); + (executor as any).activeWorktrees.set(task.id, task.worktree); + + await invokeGraphFailure(executor, task); + + // Must NOT write status:"failed" — that was the storm trigger. + const parkedFailed = store.updateTask.mock.calls.some( + (call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed", + ); + expect(parkedFailed).toBe(false); + // Benign log surfaced. + expect(logText(store)).toContain("benign, cleared for normal scheduling"); + // pausedAborted marker cleared so the next dispatch starts clean. + expect((executor as any).pausedAborted.has(task.id)).toBe(false); + // Leaked worktree slot released. + expect((executor as any).activeWorktrees.has(task.id)).toBe(false); + }); + + it("STILL parks a non-todo (in-review) pause-abort as operator-action failed", async () => { + const { store, task, executor } = makeHarness({ column: "in-review" }); + + await invokeGraphFailure(executor, task); + + const parkedFailed = store.updateTask.mock.calls.some( + (call: unknown[]) => (call[1] as { status?: string } | undefined)?.status === "failed", + ); + expect(parkedFailed).toBe(true); + expect(logText(store)).toContain("operator action required"); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts b/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts new file mode 100644 index 0000000000..2946ee6fe8 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts @@ -0,0 +1,132 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { EventEmitter } from "node:events"; + +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, existsSync: vi.fn(actual.existsSync) }; +}); + +const { logger } = vi.hoisted(() => ({ logger: { log: vi.fn(), warn: vi.fn(), error: vi.fn() } })); +vi.mock("../logger.js", () => ({ + createLogger: vi.fn(() => logger), + schedulerLog: { log: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +vi.mock("../worktree-pool.js", () => ({ + WorktreePool: vi.fn(), + RemovalReason: {}, + scanIdleWorktrees: vi.fn().mockResolvedValue([]), + cleanupOrphanedWorktrees: vi.fn().mockResolvedValue(0), + isUsableTaskWorktree: vi.fn().mockResolvedValue(true), + removeWorktree: vi.fn().mockResolvedValue(undefined), + resolveWorktreeBackend: vi.fn(), +})); + +vi.mock("../merger.js", () => ({ classifyOwnedLandedEvidence: vi.fn() })); + +import { SelfHealingManager } from "../self-healing.js"; +import type { Settings, Task, TaskStore } from "@fusion/core"; + +const PARK_ERROR = + "Workflow graph failure surfaced after paused engine abort during pause/resume in 'todo' at node 'execute' — operator action required; retry or explicitly unpause/resume after inspecting the task"; + +function createMockStore(tasks: Task[]): TaskStore & EventEmitter { + const emitter = new EventEmitter(); + const byId = new Map(tasks.map((t) => [t.id, t])); + return Object.assign(emitter, { + getSettings: vi.fn().mockResolvedValue({ + autoMerge: true, + globalPause: false, + enginePaused: false, + maintenanceIntervalMs: 0, + } as unknown as Settings), + listTasks: vi.fn().mockResolvedValue(tasks), + getTask: vi.fn(async (id: string) => byId.get(id) ?? null), + updateTask: vi.fn().mockResolvedValue({} as Task), + logEntry: vi.fn().mockResolvedValue(undefined), + moveTask: vi.fn().mockResolvedValue(undefined), + recordRunAuditEvent: vi.fn().mockResolvedValue(undefined), + getRootDir: vi.fn().mockReturnValue("/tmp/test-project"), + }) as unknown as TaskStore & EventEmitter; +} + +function parkTask(overrides: Partial = {}): Task { + return { + id: "FN-7000", + column: "todo", + paused: false, + userPaused: false, + status: "failed", + error: PARK_ERROR, + steps: [{ status: "pending" }], + title: "parked task", + ...overrides, + } as unknown as Task; +} + +describe("recoverPausedAbortFailures", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-06-20T02:30:00.000Z")); + }); + afterEach(() => { + vi.useRealTimers(); + vi.clearAllMocks(); + }); + + it("clears a todo-column pause-abort park to schedulable (status:null) without moving it", async () => { + const store = createMockStore([parkTask({ id: "FN-7000", column: "todo" })]); + const release = vi.fn(); + const manager = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + getExecutingTaskIds: () => new Set(), + releaseExecutorWorktreeOwnership: release, + }); + + const recovered = await manager.recoverPausedAbortFailures(); + + expect(recovered).toBe(1); + expect(store.updateTask).toHaveBeenCalledWith("FN-7000", { status: null, error: null }); + // Already in todo — must NOT be moved. + expect(store.moveTask).not.toHaveBeenCalled(); + expect(release).toHaveBeenCalledWith("FN-7000"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith( + expect.objectContaining({ mutationType: "task:auto-recover-paused-abort-park", target: "FN-7000" }), + ); + }); + + it("rehomes an in-progress pause-abort park back to todo", async () => { + const store = createMockStore([parkTask({ id: "FN-7001", column: "in-progress" })]); + const manager = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + getExecutingTaskIds: () => new Set(), + }); + + const recovered = await manager.recoverPausedAbortFailures(); + + expect(recovered).toBe(1); + expect(store.moveTask).toHaveBeenCalledWith( + "FN-7001", + "todo", + { preserveProgress: true, moveSource: "engine", recoveryRehome: true }, + ); + }); + + it("skips paused, executing, in-review, and non-pause-abort failures", async () => { + const store = createMockStore([ + parkTask({ id: "FN-A", paused: true }), + parkTask({ id: "FN-B", column: "in-progress" }), // executing (below) + parkTask({ id: "FN-C", column: "in-review" }), // in-review park left for operator + parkTask({ id: "FN-D", error: "some other failure", status: "failed" }), + ]); + const manager = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + getExecutingTaskIds: () => new Set(["FN-B"]), + }); + + const recovered = await manager.recoverPausedAbortFailures(); + + expect(recovered).toBe(0); + expect(store.updateTask).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3358233f9f..5be0f7a3b3 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -6640,6 +6640,34 @@ export class TaskExecutor { ? "engine abort during pause/resume" : "task pause"; if (live.column !== "in-progress") { + // FN-6782: a pause/resume abort that has left the task back in `todo` + // is benign — the work is simply re-queued for a fresh dispatch, not + // stranded. Parking it `status: "failed"` (operator action required) + // here is what caused the retry storm: the scheduler re-dispatches the + // todo task, this branch re-fires on the still-set pausedAborted + // marker, and it re-parks instantly with no backoff. Treat `todo` like + // the in-progress benign case: clear the abort marker so the next + // dispatch starts clean, log, and return WITHOUT parking failed. The + // operator-action failure is preserved only for genuinely stranded + // non-todo columns (e.g. in-review), per FN-6478. + if (live.column === "todo") { + this.clearPausedAborted(task.id); + // FN-6782 leak fix: a task parked back to `todo` must not keep + // pinning its in-memory worktree slot. The execute() finally does + // not delete activeWorktrees on this early-return path, so without + // this release the slot leaks — a `todo` task stays a maxWorktrees + // holder and concurrency-blocks the whole queue (the FN-6756 + // "in todo yet still a holder, maxWorktrees=3/3" symptom). Mirror + // clearPhantomExecutorBinding's release semantics. Safe here: + // handleGraphFailure is terminal for this run (no seam re-entry), + // and the next dispatch re-acquires a fresh worktree. + this.activeWorktrees.delete(task.id); + const todoBenign = `Workflow graph run ended during ${pauseProvenance} with task re-queued to todo — benign, cleared for normal scheduling`; + executorLog.log(`${task.id}: ${todoBenign}`); + await this.store.logEntry(task.id, todoBenign, undefined, this.getRunContextFor(task.id)); + await this.persistTokenUsage(task.id); + return; + } const failedNode = result.visitedNodeIds[result.visitedNodeIds.length - 1] ?? "unknown"; const message = `Workflow graph failure surfaced after paused ${pauseProvenance} in '${live.column}' at node '${failedNode}' — operator action required; retry or explicitly unpause/resume after inspecting the task`; executorLog.warn(`${task.id}: ${message}`); diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index f43c202984..b6f8c5d43c 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -459,6 +459,10 @@ export type DatabaseMutationType = | "task:auto-recover-worktree-metadata-rebound" | "task:auto-recover-worktree-metadata-cleared" | "task:auto-recover-worktree-metadata-skipped-active" + // pause-abort auto-recovery (FN-6782): a global pause/resume park cleared and requeued + | "task:auto-recover-paused-abort-park" + // leaked concurrency-slot reaper: a worktree/lease/semaphore slot whose holder left in-progress + | "task:reap-leaked-concurrency-slot" // task:auto-archived-ghost-bug metadata: { findings: Array<{ construct: { kind: string; raw: string; filePath?: string; line?: number }; matched: boolean; probeError?: string; output?: string }>; reason: string } // task:auto-archived-duplicate metadata: { siblingTaskIds: string[]; scores: Record } | "task:auto-archived-ghost-bug" diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 457f96e9eb..64eebbafea 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -2111,6 +2111,7 @@ export class SelfHealingManager { { name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() }, { name: "recover-missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures() }, { name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() }, + { name: "recover-paused-abort-failures", fn: () => this.recoverPausedAbortFailures() }, { name: "recover-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() }, { name: "recover-orphaned-executions", fn: () => this.recoverOrphanedExecutions() }, { name: "recover-approved-triage", fn: () => this.recoverApprovedTriageTasks() }, @@ -7874,6 +7875,98 @@ export class SelfHealingManager { } } + /** + * FN-6782 / pause-abort auto-recovery: a global pause/resume cycle can leave a + * task parked `status: "failed"` with the "operator action required" message + * produced by the executor's genuine-pause-abort branch (executor.ts + * handleGraphFailure). Historically that required a human to retry/unpause. + * This sweep auto-recovers those parks: it clears the failed status/error and + * rehomes the task to `todo` with `status: null`. The scheduler treats a + * non-paused `todo` task with `status: null` as runnable (scheduler.ts builds + * its dispatch set from `column === "todo" && !paused`; `status: "queued"` is + * the *blocked* marker, not the runnable one), so clearing to null is what + * makes the task schedulable again. + * + * Guards mirror the other failed-task recoverers: never touch a task that is + * paused, currently executing, or whose error is not the pause-abort park. + */ + async recoverPausedAbortFailures(): Promise { + try { + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const tasks = await this.store.listTasks({ slim: true }); + + const isPausedAbortPark = (t: Task): boolean => + t.status === "failed" && + typeof t.error === "string" && + t.error.includes("operator action required") && + t.error.includes("Workflow graph failure surfaced after paused"); + + const parked = tasks.filter((t) => + isPausedAbortPark(t) && + !t.paused && + !t.userPaused && + !executingIds.has(t.id) && + // Only recover columns that are safe to requeue. done/archived parks are + // terminal and in-review parks may carry merge state — leave those for + // the existing review recoverers / operator inspection. + (t.column === "todo" || t.column === "in-progress"), + ); + + if (parked.length === 0) return 0; + + log.warn(`Found ${parked.length} pause-abort park(s) requiring auto-recovery`); + + let recovered = 0; + for (const task of parked) { + try { + // Re-read to avoid acting on a stale snapshot after awaits. + const fresh = await this.store.getTask(task.id); + if (!fresh || !isPausedAbortPark(fresh) || fresh.paused || executingIds.has(fresh.id)) { + continue; + } + + await this.store.updateTask(task.id, { status: null, error: null }); + if (fresh.column !== "todo") { + await this.store.moveTask(task.id, "todo", { + preserveProgress: true, + moveSource: "engine", + recoveryRehome: true, + }); + } + // Release any in-memory worktree ownership the leaked park may still + // pin, so the requeued task does not re-block the concurrency gate. + this.options.releaseExecutorWorktreeOwnership?.(task.id); + + await this.store.logEntry( + task.id, + "Auto-recovered: pause-abort park cleared — requeued for normal scheduling", + ); + await this.store.recordRunAuditEvent?.({ + taskId: task.id, + agentId: "self-healing", + runId: generateSyntheticRunId("self-healing", task.id), + domain: "database", + mutationType: "task:auto-recover-paused-abort-park", + target: task.id, + metadata: { fromColumn: fresh.column }, + }); + log.log(`Recovered pause-abort park ${task.id}: ${task.title || task.description?.slice(0, 60) || "(untitled)"}`); + recovered++; + } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Failed to recover pause-abort park ${task.id}: ${errorMessage}`); + } + } + + if (recovered > 0) { + log.log(`Recovered ${recovered} pause-abort park(s) → requeued to todo`); + } + return recovered; + } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Pause-abort park recovery failed: ${errorMessage}`); + return 0; + } + } + async auditNoCommitsExpectedCandidates(): Promise { try { const inReviewTasks = await this.store.listTasks({ column: "in-review", slim: true });