fix(FN-1290): recover orphaned stuck tasks
This commit is contained in:
@@ -8,10 +8,12 @@ const {
|
||||
mockSelfHealingStart,
|
||||
mockSelfHealingStop,
|
||||
mockSelfHealingCtor,
|
||||
mockExecutorCtor,
|
||||
} = vi.hoisted(() => ({
|
||||
mockSelfHealingStart: vi.fn(),
|
||||
mockSelfHealingStop: vi.fn(),
|
||||
mockSelfHealingCtor: vi.fn(),
|
||||
mockExecutorCtor: vi.fn(),
|
||||
}));
|
||||
|
||||
// Mock the TaskStore class
|
||||
@@ -79,9 +81,14 @@ vi.mock("../self-healing.js", async () => {
|
||||
// Mock the executor
|
||||
vi.mock("../executor.js", async () => {
|
||||
return {
|
||||
TaskExecutor: vi.fn().mockImplementation(() => {
|
||||
TaskExecutor: vi.fn().mockImplementation((_store, _rootDir, options) => {
|
||||
mockExecutorCtor(options);
|
||||
const self = {} as Record<string, unknown>;
|
||||
self.resumeOrphaned = vi.fn().mockResolvedValue(undefined);
|
||||
self.recoverCompletedTask = vi.fn().mockResolvedValue(true);
|
||||
self.getExecutingTaskIds = vi.fn().mockReturnValue(new Set());
|
||||
self.handleLoopDetected = vi.fn().mockResolvedValue(false);
|
||||
self.markStuckAborted = vi.fn();
|
||||
self.activeWorktrees = new Map();
|
||||
return self;
|
||||
}),
|
||||
@@ -146,6 +153,17 @@ describe("InProcessRuntime", () => {
|
||||
expect(mockSelfHealingStart).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("creates a stuck task detector and passes it to the executor", async () => {
|
||||
await runtime.start();
|
||||
|
||||
expect(mockExecutorCtor).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
stuckTaskDetector: expect.any(Object),
|
||||
}),
|
||||
);
|
||||
expect((runtime as any).stuckTaskDetector).toBeDefined();
|
||||
});
|
||||
|
||||
it("should transition to 'stopped' after stop", async () => {
|
||||
await runtime.start();
|
||||
await runtime.stop();
|
||||
|
||||
@@ -20,7 +20,7 @@ import type {
|
||||
ProjectRuntimeEvents,
|
||||
} from "../project-runtime.js";
|
||||
import { runtimeLog } from "../logger.js";
|
||||
import type { StuckTaskDetector } from "../stuck-task-detector.js";
|
||||
import { StuckTaskDetector } from "../stuck-task-detector.js";
|
||||
import type { UsageLimitPauser } from "../usage-limit-detector.js";
|
||||
import { SelfHealingManager } from "../self-healing.js";
|
||||
import { MissionAutopilot } from "../mission-autopilot.js";
|
||||
@@ -163,6 +163,18 @@ export class InProcessRuntime
|
||||
});
|
||||
|
||||
// 5. Initialize TaskExecutor
|
||||
this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, {
|
||||
beforeRequeue: (taskId) => this.selfHealingManager?.checkStuckBudget(taskId) ?? Promise.resolve(true),
|
||||
onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false),
|
||||
onStuck: (event) => {
|
||||
this.executor?.markStuckAborted(event.taskId, event.shouldRequeue);
|
||||
runtimeLog.warn(
|
||||
`Task ${event.taskId} stuck (${event.reason}) — ` +
|
||||
`${event.shouldRequeue ? "will retry" : "budget exhausted"}`,
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
const executorOptions: TaskExecutorOptions = {
|
||||
semaphore: this.globalSemaphore,
|
||||
pool: this.worktreePool,
|
||||
@@ -315,6 +327,7 @@ export class InProcessRuntime
|
||||
getExecutingTaskIds: () => this.executor.getExecutingTaskIds(),
|
||||
});
|
||||
this.selfHealingManager.start();
|
||||
this.stuckTaskDetector.start();
|
||||
|
||||
// 8. Set up event forwarding from TaskStore
|
||||
this.setupEventForwarding();
|
||||
@@ -376,13 +389,19 @@ export class InProcessRuntime
|
||||
runtimeLog.log("TriggerScheduler stopped");
|
||||
}
|
||||
|
||||
// 3. Stop heartbeat monitor
|
||||
// 3. Stop stuck task detector
|
||||
if (this.stuckTaskDetector) {
|
||||
this.stuckTaskDetector.stop();
|
||||
runtimeLog.log("StuckTaskDetector stopped");
|
||||
}
|
||||
|
||||
// 4. Stop heartbeat monitor
|
||||
if (this.heartbeatMonitor) {
|
||||
this.heartbeatMonitor.stop();
|
||||
runtimeLog.log("HeartbeatMonitor stopped");
|
||||
}
|
||||
|
||||
// 4. Stop scheduler (prevents new task scheduling)
|
||||
// 5. Stop scheduler (prevents new task scheduling)
|
||||
if (this.scheduler) {
|
||||
this.scheduler.stop();
|
||||
runtimeLog.log("Scheduler stopped");
|
||||
|
||||
@@ -596,6 +596,129 @@ describe("SelfHealingManager", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("recoverOrphanedExecutions", () => {
|
||||
it("requeues in-progress tasks whose reserved worktree is missing", async () => {
|
||||
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||
const managerWithRecovery = new SelfHealingManager(store, {
|
||||
rootDir: "/tmp/test-project",
|
||||
getExecutingTaskIds: getExecuting,
|
||||
});
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-200",
|
||||
column: "in-progress",
|
||||
paused: false,
|
||||
worktree: undefined,
|
||||
steps: [{ status: "in-progress" }],
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
|
||||
|
||||
const result = await managerWithRecovery.recoverOrphanedExecutions();
|
||||
|
||||
expect(result).toBe(1);
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-200", {
|
||||
status: "stuck-killed",
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
expect(store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-200",
|
||||
"Auto-recovered orphaned executor task — missing worktree/session, moved back to todo",
|
||||
);
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-200", "todo");
|
||||
|
||||
managerWithRecovery.stop();
|
||||
});
|
||||
|
||||
it("skips orphan recovery for actively executing tasks", async () => {
|
||||
const getExecuting = vi.fn().mockReturnValue(new Set(["FN-201"]));
|
||||
const managerWithRecovery = new SelfHealingManager(store, {
|
||||
rootDir: "/tmp/test-project",
|
||||
getExecutingTaskIds: getExecuting,
|
||||
});
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-201",
|
||||
column: "in-progress",
|
||||
paused: false,
|
||||
worktree: "/tmp/test-project/.worktrees/missing-tree",
|
||||
steps: [{ status: "in-progress" }],
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
|
||||
|
||||
const result = await managerWithRecovery.recoverOrphanedExecutions();
|
||||
|
||||
expect(result).toBe(0);
|
||||
expect(store.moveTask).not.toHaveBeenCalled();
|
||||
|
||||
managerWithRecovery.stop();
|
||||
});
|
||||
|
||||
it("skips tasks that are already complete", async () => {
|
||||
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||
const managerWithRecovery = new SelfHealingManager(store, {
|
||||
rootDir: "/tmp/test-project",
|
||||
getExecutingTaskIds: getExecuting,
|
||||
});
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-202",
|
||||
column: "in-progress",
|
||||
paused: false,
|
||||
worktree: "/tmp/test-project/.worktrees/missing-tree",
|
||||
steps: [{ status: "done" }],
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
|
||||
|
||||
const result = await managerWithRecovery.recoverOrphanedExecutions();
|
||||
|
||||
expect(result).toBe(0);
|
||||
expect(store.moveTask).not.toHaveBeenCalled();
|
||||
|
||||
managerWithRecovery.stop();
|
||||
});
|
||||
|
||||
it("skips tasks still within the grace window", async () => {
|
||||
const getExecuting = vi.fn().mockReturnValue(new Set<string>());
|
||||
const managerWithRecovery = new SelfHealingManager(store, {
|
||||
rootDir: "/tmp/test-project",
|
||||
getExecutingTaskIds: getExecuting,
|
||||
});
|
||||
|
||||
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
|
||||
{
|
||||
id: "FN-203",
|
||||
column: "in-progress",
|
||||
paused: false,
|
||||
worktree: "/tmp/test-project/.worktrees/missing-tree",
|
||||
steps: [{ status: "in-progress" }],
|
||||
updatedAt: "2026-01-01T00:04:30.000Z",
|
||||
},
|
||||
]);
|
||||
|
||||
vi.setSystemTime(new Date("2026-01-01T00:05:00.000Z"));
|
||||
|
||||
const result = await managerWithRecovery.recoverOrphanedExecutions();
|
||||
|
||||
expect(result).toBe(0);
|
||||
expect(store.moveTask).not.toHaveBeenCalled();
|
||||
|
||||
managerWithRecovery.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("recoverApprovedTriageTasks", () => {
|
||||
it("recovers approved specifying triage tasks that are not actively processing", async () => {
|
||||
const recoverFn = vi.fn().mockResolvedValue(true);
|
||||
|
||||
@@ -53,6 +53,7 @@ export interface SelfHealingOptions {
|
||||
}
|
||||
|
||||
const APPROVED_TRIAGE_RECOVERY_GRACE_MS = 60_000;
|
||||
const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000;
|
||||
|
||||
export class SelfHealingManager {
|
||||
// ── Auto-unpause state ──────────────────────────────────────────────
|
||||
@@ -264,6 +265,7 @@ export class SelfHealingManager {
|
||||
this.checkpointWal();
|
||||
await this.enforceWorktreeCap();
|
||||
await this.recoverCompletedTasks();
|
||||
await this.recoverOrphanedExecutions();
|
||||
await this.recoverApprovedTriageTasks();
|
||||
|
||||
const elapsedMs = Date.now() - startMs;
|
||||
@@ -322,6 +324,59 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover executor tasks stranded in `in-progress` before a real session was
|
||||
* established, typically when the scheduler reserved a worktree path but the
|
||||
* executor never materialized it or crashed before tracking the run.
|
||||
*/
|
||||
async recoverOrphanedExecutions(): Promise<number> {
|
||||
try {
|
||||
const tasks = await this.store.listTasks();
|
||||
const executingIds = this.options.getExecutingTaskIds?.() ?? new Set<string>();
|
||||
const now = Date.now();
|
||||
|
||||
const orphaned = tasks.filter((t) =>
|
||||
t.column === "in-progress" &&
|
||||
!t.paused &&
|
||||
!executingIds.has(t.id) &&
|
||||
!isTaskWorkComplete(t) &&
|
||||
(!t.worktree || !existsSync(t.worktree)) &&
|
||||
now - new Date(t.updatedAt).getTime() >= ORPHANED_EXECUTION_RECOVERY_GRACE_MS,
|
||||
);
|
||||
|
||||
if (orphaned.length === 0) return 0;
|
||||
|
||||
log.warn(`Found ${orphaned.length} orphaned executor task(s) stuck in in-progress`);
|
||||
|
||||
let recovered = 0;
|
||||
for (const task of orphaned) {
|
||||
try {
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "stuck-killed",
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
await this.store.logEntry(
|
||||
task.id,
|
||||
"Auto-recovered orphaned executor task — missing worktree/session, moved back to todo",
|
||||
);
|
||||
await this.store.moveTask(task.id, "todo");
|
||||
recovered++;
|
||||
} catch (err: any) {
|
||||
log.error(`Failed to recover orphaned executor task ${task.id}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (recovered > 0) {
|
||||
log.log(`Recovered ${recovered} orphaned executor task(s) → todo`);
|
||||
}
|
||||
return recovered;
|
||||
} catch (err: any) {
|
||||
log.error(`Orphaned executor recovery failed: ${err.message}`);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover triage tasks that already have an approved specification but were
|
||||
* left stuck in `status: "specifying"` without an active triage session.
|
||||
@@ -549,3 +604,8 @@ function hasLatestSpecReviewApproval(task: Task): boolean {
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
function isTaskWorkComplete(task: Task): boolean {
|
||||
if (task.steps.length === 0) return false;
|
||||
return task.steps.every((step) => step.status === "done" || step.status === "skipped");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user