root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: defer stuck-kill requeue to executor finally block to prevent race

Browse Source 
When the stuck task detector killed a task and immediately called
moveTask("todo"), the scheduler could re-dispatch the task before the
old execution's finally block cleared this.executing. The new execute()
call hit the guard and silently returned, stranding the task in
"in-progress" with no active session or worktree (seen on FN-810/FN-912).

Move the requeue responsibility from StuckTaskDetector.killAndRetry to
the executor's finally block, which runs after this.executing.delete().
The beforeRequeue budget check now runs before session.dispose() and its
result is passed via StuckTaskEvent.shouldRequeue → markStuckAborted().

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-04 12:26:13 -07:00
parent 1ab1981557
commit fbda14717e
5 changed files with 144 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
packages/engine/src/executor.test.ts
View File

@@ -4635,7 +4635,7 @@ describe("TaskExecutor bounded recovery retries", () => {
const executor = new TaskExecutor(store, "/tmp/test", {});
mockedCreateHaiAgent.mockRejectedValue(new Error("Aborted"));
(executor as any).stuckAborted.add("FN-001");
(executor as any).stuckAborted.set("FN-001", true);
await executor.execute({
id: "FN-001",
@@ -4657,14 +4657,14 @@ describe("TaskExecutor bounded recovery retries", () => {
}));
});
it("exits cleanly when a stuck-killed session resolves without throwing", async () => {
it("requeues to todo when a stuck-killed session resolves without throwing", async () => {
const store = createMockStore();
const executor = new TaskExecutor(store, "/tmp/test", {});
mockedCreateHaiAgent.mockImplementation(async () => ({
session: {
prompt: vi.fn(async () => {
executor.markStuckAborted("FN-001");
executor.markStuckAborted("FN-001", true);
}),
dispose: vi.fn(),
state: {},
@@ -4690,6 +4690,46 @@ describe("TaskExecutor bounded recovery retries", () => {
expect.objectContaining({ status: "failed" }),
);
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" });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
});
it("does not requeue when stuck-kill budget is exhausted", async () => {
const store = createMockStore();
const executor = new TaskExecutor(store, "/tmp/test", {});
mockedCreateHaiAgent.mockImplementation(async () => ({
session: {
prompt: vi.fn(async () => {
// Budget exhausted — shouldRequeue=false
executor.markStuckAborted("FN-001", false);
}),
dispose: vi.fn(),
state: {},
},
}) as any);
await executor.execute({
id: "FN-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// 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" });
expect(store.updateTask).not.toHaveBeenCalledWith(
"FN-001",
expect.objectContaining({ status: "failed" }),
);
});
it("clears recovery metadata after successful run completes", async () => {

43
packages/engine/src/executor.ts
View File

@@ -184,8 +184,8 @@ export class TaskExecutor {
private pausedAborted = new Set<string>();
/** Tasks that had a dependency added mid-execution (abort + discard worktree). */
private depAborted = new Set<string>();
/** Tasks that were killed by stuck task detector (to avoid marking them as "failed"). */
private stuckAborted = new Set<string>();
/** Tasks killed by stuck task detector. Value = shouldRequeue (budget not exhausted). */
private stuckAborted = new Map<string, boolean>();
/**
* @param store — Task store instance (also used to listen for events)
@@ -417,6 +417,11 @@ export class TaskExecutor {
worktreePath = join(this.rootDir, ".worktrees", worktreeName);
}
// Set by stuck-abort handlers; the actual moveTask("todo") is deferred to
// the finally block so this.executing is cleared first (prevents re-dispatch race).
// true = requeue to todo, false = budget exhausted (already marked failed).
let stuckRequeue: boolean | null = null;
try {
// Check dependencies
const allTasks = await this.store.listTasks();
@@ -700,8 +705,11 @@ export class TaskExecutor {
}
// If the stuck task detector disposed the session and the agent exited
// cleanly, stop here. The detector already handled recovery/re-queueing.
// cleanly, stop here. The requeue is deferred to the finally block
// (after this.executing is cleared) to prevent a race where the
// scheduler re-dispatches while the old execution guard is still set.
if (this.stuckAborted.has(task.id)) {
stuckRequeue = this.stuckAborted.get(task.id) ?? true;
this.stuckAborted.delete(task.id);
executorLog.log(`${task.id} terminated by stuck task detector (graceful session exit)`);
return;
@@ -860,10 +868,11 @@ export class TaskExecutor {
await this.store.logEntry(task.id, "Execution paused — agent terminated, moved to todo");
await this.store.moveTask(task.id, "todo");
} else if (this.stuckAborted.has(task.id)) {
// Task was killed by stuck task detector — already moved to todo by killAndRetry.
// Don't mark as failed; the scheduler will retry it naturally.
executorLog.log(`${task.id} terminated by stuck task detector — will retry`);
// Task was killed by stuck task detector — defer requeue to finally block
// (after this.executing is cleared) to prevent re-dispatch race.
stuckRequeue = this.stuckAborted.get(task.id) ?? true;
this.stuckAborted.delete(task.id);
executorLog.log(`${task.id} terminated by stuck task detector — will ${stuckRequeue ? "retry" : "not retry (budget exhausted)"}`);
} else {
// Check if the error is a usage-limit error and trigger global pause
if (this.options.usageLimitPauser && isUsageLimitError(err.message)) {
@@ -907,6 +916,21 @@ export class TaskExecutor {
}
} finally {
this.executing.delete(task.id);
// Requeue stuck-killed task AFTER this.executing is cleared.
// This prevents the race where the scheduler re-dispatches the task
// (via task:moved → execute()) while the old execution guard is still set,
// which caused the new execute() call to silently no-op, stranding the
// task in "in-progress" with no active session or worktree.
if (stuckRequeue === true) {
try {
await this.store.updateTask(task.id, { status: "stuck-killed" });
await this.store.moveTask(task.id, "todo");
executorLog.log(`${task.id} moved to todo for retry after stuck kill`);
} catch (err: any) {
executorLog.error(`Failed to requeue stuck task ${task.id}: ${err.message}`);
}
}
}
}
@@ -2108,9 +2132,12 @@ If issues are found that need attention, describe them clearly.`;
* Mark a task as stuck-aborted so the executor's error handling
* knows not to treat the disposed session as a genuine failure.
* Called by the stuck task detector's onStuck callback.
*
* @param shouldRequeue — true to move the task back to "todo" for retry,
* false if the stuck kill budget is exhausted (task already marked failed).
*/
markStuckAborted(taskId: string): void {
this.stuckAborted.add(taskId);
markStuckAborted(taskId: string, shouldRequeue: boolean = true): void {
this.stuckAborted.set(taskId, shouldRequeue);
}
getWorktreePath(taskId: string): string | undefined {

63
packages/engine/src/stuck-task-detector.test.ts
View File

@@ -391,7 +391,7 @@ describe("StuckTaskDetector", () => {
vi.useRealTimers();
});
it("updates task status and moves to todo", async () => {
it("does not move task to todo directly (deferred to executor)", async () => {
const session = createMockSession();
detector.trackTask("FN-001", session);
@@ -400,13 +400,15 @@ describe("StuckTaskDetector", () => {
await detector.killAndRetry("FN-001", 60000);
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: "stuck-killed" });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
// The detector no longer moves the task — the executor handles this
// in its finally block after clearing the execution guard.
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "stuck-killed" });
expect(store.moveTask).not.toHaveBeenCalled();
vi.useRealTimers();
});
it("calls onStuck callback with structured event payload", async () => {
it("calls onStuck callback with structured event payload including shouldRequeue", async () => {
const onStuck = vi.fn();
const customDetector = new StuckTaskDetector(store, { onStuck });
const session = createMockSession();
@@ -425,6 +427,7 @@ describe("StuckTaskDetector", () => {
noProgressMs: expect.any(Number),
inactivityMs: expect.any(Number),
activitySinceProgress: 0,
shouldRequeue: true,
}),
);
@@ -485,10 +488,10 @@ describe("StuckTaskDetector", () => {
it("does nothing for untracked task", async () => {
await detector.killAndRetry("FN-001", 60000);
// Should not throw
expect(store.moveTask).not.toHaveBeenCalled();
expect(store.logEntry).not.toHaveBeenCalled();
});
it("calls beforeRequeue and skips re-queue when it returns false", async () => {
it("calls beforeRequeue and passes shouldRequeue=false to onStuck when budget exhausted", async () => {
const beforeRequeue = vi.fn().mockResolvedValue(false);
const onStuck = vi.fn();
const customDetector = new StuckTaskDetector(store, { beforeRequeue, onStuck });
@@ -503,18 +506,20 @@ describe("StuckTaskDetector", () => {
expect(beforeRequeue).toHaveBeenCalledWith("FN-001");
expect(session.dispose).toHaveBeenCalled();
// onStuck should still be called (so executor can mark stuck-aborted)
expect(onStuck).toHaveBeenCalled();
// But task should NOT be moved to todo
// onStuck should still be called with shouldRequeue=false
expect(onStuck).toHaveBeenCalledWith(
expect.objectContaining({ taskId: "FN-001", shouldRequeue: false }),
);
// Detector no longer moves tasks — executor handles it
expect(store.moveTask).not.toHaveBeenCalled();
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "stuck-killed" });
vi.useRealTimers();
});
it("calls beforeRequeue and proceeds with re-queue when it returns true", async () => {
it("calls beforeRequeue and passes shouldRequeue=true to onStuck when budget allows", async () => {
const beforeRequeue = vi.fn().mockResolvedValue(true);
const customDetector = new StuckTaskDetector(store, { beforeRequeue });
const onStuck = vi.fn();
const customDetector = new StuckTaskDetector(store, { beforeRequeue, onStuck });
const session = createMockSession();
customDetector.trackTask("FN-001", session);
@@ -525,15 +530,19 @@ describe("StuckTaskDetector", () => {
await customDetector.killAndRetry("FN-001", 60000);
expect(beforeRequeue).toHaveBeenCalledWith("FN-001");
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: "stuck-killed" });
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(onStuck).toHaveBeenCalledWith(
expect.objectContaining({ taskId: "FN-001", shouldRequeue: true }),
);
// Detector no longer moves tasks — executor handles it
expect(store.moveTask).not.toHaveBeenCalled();
vi.useRealTimers();
});
it("falls through to re-queue when beforeRequeue throws", async () => {
it("falls through with shouldRequeue=true when beforeRequeue throws", async () => {
const beforeRequeue = vi.fn().mockRejectedValue(new Error("check failed"));
const customDetector = new StuckTaskDetector(store, { beforeRequeue });
const onStuck = vi.fn();
const customDetector = new StuckTaskDetector(store, { beforeRequeue, onStuck });
const session = createMockSession();
customDetector.trackTask("FN-001", session);
@@ -543,19 +552,23 @@ describe("StuckTaskDetector", () => {
await customDetector.killAndRetry("FN-001", 60000);
// Should still re-queue on error (safe fallback)
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
// Should pass shouldRequeue=true on error (safe fallback)
expect(onStuck).toHaveBeenCalledWith(
expect.objectContaining({ taskId: "FN-001", shouldRequeue: true }),
);
expect(store.moveTask).not.toHaveBeenCalled();
vi.useRealTimers();
});
});
describe("checkNow", () => {
it("checks stuck tasks immediately", async () => {
it("checks stuck tasks immediately and disposes session", async () => {
store = createMockStore({
getSettings: vi.fn().mockResolvedValue({ taskStuckTimeoutMs: 60000 }),
});
const customDetector = new StuckTaskDetector(store);
const onStuck = vi.fn();
const customDetector = new StuckTaskDetector(store, { onStuck });
const session = createMockSession();
customDetector.trackTask("FN-001", session);
@@ -565,7 +578,12 @@ describe("StuckTaskDetector", () => {
await customDetector.checkNow();
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(session.dispose).toHaveBeenCalled();
expect(onStuck).toHaveBeenCalledWith(
expect.objectContaining({ taskId: "FN-001", shouldRequeue: true }),
);
// Detector no longer moves tasks — executor handles it
expect(store.moveTask).not.toHaveBeenCalled();
vi.useRealTimers();
});
@@ -670,9 +688,10 @@ describe("StuckTaskDetector", () => {
taskId: "FN-001",
reason: "inactivity",
activitySinceProgress: 0,
shouldRequeue: true,
}),
);
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(session.dispose).toHaveBeenCalled();
vi.useRealTimers();
});

53
packages/engine/src/stuck-task-detector.ts
View File

@@ -47,6 +47,8 @@ export interface StuckTaskEvent {
inactivityMs: number;
/** Number of activity heartbeats since the last progress event. */
activitySinceProgress: number;
/** Whether the task should be re-queued (budget not exhausted). */
shouldRequeue: boolean;
}
/** Minimum activity-since-progress count to classify as a loop.
@@ -257,13 +259,29 @@ export class StuckTaskDetector {
stuckLog.error(`Failed to log stuck event for ${taskId}:`, err);
}
// Build the event payload
// Check stuck kill budget BEFORE disposing the session so the result
// is available to the executor's cleanup path via the event payload.
let shouldRequeue = true;
if (this.beforeRequeue) {
try {
shouldRequeue = await this.beforeRequeue(taskId);
if (!shouldRequeue) {
stuckLog.log(`${taskId} exceeded stuck kill budget — not re-queuing`);
}
} catch (err) {
stuckLog.error(`beforeRequeue check failed for ${taskId}:`, err);
// Fall through with shouldRequeue=true — safer than dropping the task
}
}
// Build the event payload (includes requeue decision for executor)
const event: StuckTaskEvent = {
taskId,
reason,
noProgressMs,
inactivityMs,
activitySinceProgress,
shouldRequeue,
};
// Notify listeners before disposing the session so executor cleanup can
@@ -277,36 +295,11 @@ export class StuckTaskDetector {
stuckLog.error(`Failed to dispose session for ${taskId}:`, err);
}
// Remove from tracking
// Remove from tracking.
// The actual moveTask("todo") is handled by the executor's catch/finally
// block after it cleans up this.executing. This prevents a race where the
// scheduler re-dispatches the task while the old execution is still active.
this.tracked.delete(taskId);
// Check stuck kill budget before re-queuing (SelfHealingManager integration).
// If beforeRequeue returns false, the task has been marked failed — skip re-queue.
if (this.beforeRequeue) {
try {
const shouldRequeue = await this.beforeRequeue(taskId);
if (!shouldRequeue) {
stuckLog.log(`${taskId} exceeded stuck kill budget — not re-queuing`);
return;
}
} catch (err) {
stuckLog.error(`beforeRequeue check failed for ${taskId}:`, err);
// Fall through to re-queue on error — safer than dropping the task
}
}
// Set transient "stuck-killed" status, then move to "todo" for retry.
// moveTask from "in-progress" to "todo" automatically clears status,
// so no explicit status clear is needed after the move.
// currentStep and step statuses are preserved so execution resumes where it left off.
try {
await this.store.updateTask(taskId, { status: "stuck-killed" });
await this.store.moveTask(taskId, "todo");
stuckLog.log(`${taskId} moved to todo for retry`);
} catch (err) {
stuckLog.error(`Failed to move ${taskId} to todo:`, err);
}
}
/**