Fix pause-abort worktree leak + retry storm; add auto-recovery
A global pause/resume cycle parked tasks that had re-queued to todo as
status:"failed" ("operator action required") and leaked their in-memory
worktree slot. The scheduler kept re-dispatching the todo task, the
genuine-pause-abort branch re-fired on the still-set pausedAborted marker,
and it re-parked instantly with no backoff — a retry storm (75x/hr) that
pinned maxWorktrees=3/3 and concurrency-starved the whole queue.
- R1+R2 (executor.ts handleGraphFailure): treat a pause-abort that left a
task in `todo` as benign (FN-6782) — don't park failed, clear the
pausedAborted marker so the next dispatch is clean, and release the
leaked activeWorktrees slot. Operator-action failure preserved for
genuinely stranded non-todo columns (FN-6478).
- A1 (self-healing.ts recoverPausedAbortFailures): new maintenance sweep
that auto-recovers any pause-abort park still on the board and requeues
it (status:null = schedulable) so the board self-heals.
- run-audit.ts: new mutation types for the recovery telemetry.
Corrected the spec's null-vs-queued assumption: the scheduler dispatch set
is column==="todo" && !paused (scheduler.ts:1288); status:"queued" is the
*blocked* marker, status:null is runnable — so recovered tasks are left null.
Deferred (documented): A2 leaked-slot reaper needs a new executor
listWorktreeHolders introspection API to reap in-memory worktree slots
safely; R1 closes the observed leak at its source.
Tests: self-healing-paused-abort-recovery.test.ts (3),
executor-paused-abort-todo-benign.test.ts (2). Engine typecheck clean;
106 existing pause/graph-failure/limbo tests still pass.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
8
.changeset/fix-pause-abort-leak-storm.md
Normal file
8
.changeset/fix-pause-abort-leak-storm.md
Normal file
@@ -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.
|
||||||
@@ -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> = {}): 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<TaskDetail> = {}) {
|
||||||
|
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<typeof createMockStore>): 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");
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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<typeof import("node:fs")>();
|
||||||
|
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> = {}): 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<string>(),
|
||||||
|
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<string>(),
|
||||||
|
});
|
||||||
|
|
||||||
|
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<string>(["FN-B"]),
|
||||||
|
});
|
||||||
|
|
||||||
|
const recovered = await manager.recoverPausedAbortFailures();
|
||||||
|
|
||||||
|
expect(recovered).toBe(0);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -6640,6 +6640,34 @@ export class TaskExecutor {
|
|||||||
? "engine abort during pause/resume"
|
? "engine abort during pause/resume"
|
||||||
: "task pause";
|
: "task pause";
|
||||||
if (live.column !== "in-progress") {
|
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 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`;
|
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}`);
|
executorLog.warn(`${task.id}: ${message}`);
|
||||||
|
|||||||
@@ -459,6 +459,10 @@ export type DatabaseMutationType =
|
|||||||
| "task:auto-recover-worktree-metadata-rebound"
|
| "task:auto-recover-worktree-metadata-rebound"
|
||||||
| "task:auto-recover-worktree-metadata-cleared"
|
| "task:auto-recover-worktree-metadata-cleared"
|
||||||
| "task:auto-recover-worktree-metadata-skipped-active"
|
| "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-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<string, number> }
|
// task:auto-archived-duplicate metadata: { siblingTaskIds: string[]; scores: Record<string, number> }
|
||||||
| "task:auto-archived-ghost-bug"
|
| "task:auto-archived-ghost-bug"
|
||||||
|
|||||||
@@ -2111,6 +2111,7 @@ export class SelfHealingManager {
|
|||||||
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
|
{ name: "recover-misclassified-failures", fn: () => this.recoverMisclassifiedFailures() },
|
||||||
{ name: "recover-missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures() },
|
{ name: "recover-missing-worktree-review-failures", fn: () => this.recoverMissingWorktreeReviewFailures() },
|
||||||
{ name: "recover-no-progress-no-task-done", fn: () => this.recoverNoProgressNoTaskDoneFailures() },
|
{ 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-partial-progress-no-task-done", fn: () => this.recoverPartialProgressNoTaskDoneFailures() },
|
||||||
{ name: "recover-orphaned-executions", fn: () => this.recoverOrphanedExecutions() },
|
{ name: "recover-orphaned-executions", fn: () => this.recoverOrphanedExecutions() },
|
||||||
{ name: "recover-approved-triage", fn: () => this.recoverApprovedTriageTasks() },
|
{ 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<number> {
|
||||||
|
try {
|
||||||
|
const executingIds = this.options.getExecutingTaskIds?.() ?? new Set<string>();
|
||||||
|
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<number> {
|
async auditNoCommitsExpectedCandidates(): Promise<number> {
|
||||||
try {
|
try {
|
||||||
const inReviewTasks = await this.store.listTasks({ column: "in-review", slim: true });
|
const inReviewTasks = await this.store.listTasks({ column: "in-review", slim: true });
|
||||||
|
|||||||
Reference in New Issue
Block a user