fix(engine): requeue stale assistant continuations (#2095)
## Summary - detect persisted executor sessions that cannot continue from an assistant message - clear the stale session pointer after the executor lock is released - requeue the task with workflow progress preserved instead of marking it failed ## Test plan - `pnpm --filter @fusion/engine exec vitest run src/__tests__/executor-step-session.test.ts -t "clears a stale assistant-continuation resume session and requeues without marking the task failed" --project=engine-default --silent=passed-only --reporter=dot` - `pnpm --filter @fusion/engine typecheck` - `pnpm build` <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Bug Fixes** * Improved recovery when an assistant continuation session becomes stale by restarting a fresh session with bounded retries, preserving overall task progress. * Clears invalid persisted session/continuation state and defers requeue until coordination cleanup is safe. * When retries are exhausted, tasks are marked failed and the error callback runs (without routing to review). * **Tests** * Added coverage for stale-session recovery, repeated-stale behavior, correct (or skipped) requeue decisions, and progress/error handling paths. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
Phil Larson
7
.changeset/calm-taxis-requeue.md
Normal file
7
.changeset/calm-taxis-requeue.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Recover stale executor sessions with bounded fresh-session retries while preserving task progress.
|
||||
category: fix
|
||||
dev: Clears the persisted assistant-last transcript, defers requeue until lock release, and exhausts through the shared recovery budget.
|
||||
@@ -16,6 +16,8 @@ import { SessionManager } from "@earendil-works/pi-coding-agent";
|
||||
import { StepSessionExecutor } from "../step-session-executor.js";
|
||||
import { executorLog } from "../logger.js";
|
||||
import { withRateLimitRetry } from "../rate-limit-retry.js";
|
||||
import { MAX_RECOVERY_RETRIES } from "../recovery-policy.js";
|
||||
import { executingTaskLock } from "../active-session-registry.js";
|
||||
import { runVerificationCommand as mockedRunVerificationCommand } from "../verification-utils.js";
|
||||
import {
|
||||
createMockStore,
|
||||
@@ -246,6 +248,109 @@ describe("Workflow Steps Execution", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("clears a stale assistant-continuation resume session and requeues without marking the task failed", async () => {
|
||||
const store = createMockStore();
|
||||
const task = {
|
||||
id: "FN-ASSISTANT-STALE",
|
||||
title: "Stale assistant continuation",
|
||||
description: "Test stale assistant continuation recovery",
|
||||
column: "in-progress",
|
||||
dependencies: [],
|
||||
steps: [{ name: "Preflight", status: "in-progress" as const }],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
|
||||
sessionFile: "/tmp/stale-session.jsonl",
|
||||
worktree: "/tmp/test/.worktrees/fn-assistant-stale",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
store.getTask.mockResolvedValue(task as any);
|
||||
store.moveTask.mockImplementation(async () => {
|
||||
expect(executingTaskLock.has("FN-ASSISTANT-STALE")).toBe(false);
|
||||
expect((executor as any).activeWorktrees.has("FN-ASSISTANT-STALE")).toBe(false);
|
||||
return task as any;
|
||||
});
|
||||
|
||||
const staleSession = {
|
||||
prompt: vi.fn().mockRejectedValue(new Error("Cannot continue from message role: assistant")),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
on: vi.fn(),
|
||||
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
|
||||
state: {},
|
||||
};
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: staleSession } as any);
|
||||
|
||||
const onError = vi.fn();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { onError });
|
||||
const markGraphExecuteSelfRequeued = vi.spyOn(executor as any, "markGraphExecuteSelfRequeued");
|
||||
(executor as any).activeWorktrees.set("FN-ASSISTANT-STALE", new Set([task.worktree]));
|
||||
|
||||
await executor.execute(task as any);
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-ASSISTANT-STALE", {
|
||||
sessionFile: null,
|
||||
recoveryRetryCount: 1,
|
||||
nextRecoveryAt: expect.any(String),
|
||||
});
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-ASSISTANT-STALE", {
|
||||
sessionFile: null,
|
||||
status: null,
|
||||
error: null,
|
||||
});
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-ASSISTANT-STALE", "todo", { preserveResumeState: true });
|
||||
expect(markGraphExecuteSelfRequeued).toHaveBeenCalledWith("FN-ASSISTANT-STALE");
|
||||
expect(executingTaskLock.has("FN-ASSISTANT-STALE")).toBe(false);
|
||||
expect((executor as any).activeWorktrees.has("FN-ASSISTANT-STALE")).toBe(false);
|
||||
expect(store.handoffToReview).not.toHaveBeenCalled();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("fails a repeated stale assistant-continuation after the fresh-session retry budget is exhausted", async () => {
|
||||
const store = createMockStore();
|
||||
const task = {
|
||||
id: "FN-ASSISTANT-STALE-EXHAUSTED",
|
||||
title: "Repeated stale assistant continuation",
|
||||
description: "Test bounded stale assistant continuation recovery",
|
||||
column: "in-progress",
|
||||
dependencies: [],
|
||||
steps: [{ name: "Preflight", status: "in-progress" as const }],
|
||||
currentStep: 0,
|
||||
recoveryRetryCount: MAX_RECOVERY_RETRIES,
|
||||
log: [],
|
||||
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
|
||||
sessionFile: "/tmp/stale-session.jsonl",
|
||||
worktree: "/tmp/test/.worktrees/fn-assistant-stale-exhausted",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
store.getTask.mockResolvedValue(task as any);
|
||||
mockedCreateFnAgent.mockResolvedValue({
|
||||
session: {
|
||||
prompt: vi.fn().mockRejectedValue(new Error("Cannot continue from message role: assistant")),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
on: vi.fn(),
|
||||
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
|
||||
state: {},
|
||||
},
|
||||
} as any);
|
||||
const onError = vi.fn();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { onError });
|
||||
|
||||
await executor.execute(task as any);
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-ASSISTANT-STALE-EXHAUSTED", {
|
||||
status: "failed",
|
||||
error: "Cannot continue from message role: assistant",
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
});
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-ASSISTANT-STALE-EXHAUSTED", "todo", expect.anything());
|
||||
expect(onError).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
describe("FN-5436: pending-review skip on no-fn_task_done exit", () => {
|
||||
it("does not park in-review when code review REVISE requires more executor work", async () => {
|
||||
const store = createMockStore();
|
||||
|
||||
@@ -566,6 +566,12 @@ function buildExecuteRequeueLoopHighWaterSignature(live: TaskDetail, previousSig
|
||||
};
|
||||
}
|
||||
|
||||
const INVALID_ASSISTANT_CONTINUATION_PATTERN = /cannot continue from message role:\s*assistant/i;
|
||||
|
||||
function isInvalidAssistantContinuationErrorMessage(errorMessage: string): boolean {
|
||||
return INVALID_ASSISTANT_CONTINUATION_PATTERN.test(errorMessage);
|
||||
}
|
||||
|
||||
const TRANSIENT_WORKTREE_TASK_JSON_ENOENT_PATTERN = /ENOENT:\s+no such file or directory,\s+open\s+'([^']+\/\.fusion\/tasks\/([^/]+)\/task\.json)'/;
|
||||
|
||||
export function isTransientMissingTaskJsonError(error: unknown, task: Pick<Task, "id" | "worktree">): boolean {
|
||||
@@ -9824,6 +9830,7 @@ export class TaskExecutor {
|
||||
// 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;
|
||||
let staleAssistantContinuationRequeue = false;
|
||||
let taskDone = false;
|
||||
let reviewAddressingActivated = false;
|
||||
let taskEnv: NodeJS.ProcessEnv | undefined;
|
||||
@@ -11793,6 +11800,54 @@ export class TaskExecutor {
|
||||
// Dependency added mid-execution — discard worktree and move to triage
|
||||
this.depAborted.delete(task.id);
|
||||
await this.handleDepAbortCleanup(task.id, worktreePath);
|
||||
} else if (isInvalidAssistantContinuationErrorMessage(errorMessage)) {
|
||||
/*
|
||||
FNXC:ExecutorSessionRecovery 2026-07-14-06:03:
|
||||
A stale assistant-last transcript gets a bounded fresh-session retry with the shared recovery backoff. The retry counter must survive the deferred move so repeated fresh-session failures eventually become a visible execution failure instead of cycling through Todo forever.
|
||||
|
||||
FNXC:ExecutorSessionRecovery 2026-07-14-06:19:
|
||||
Deferred self-requeues must mark the workflow graph recovery and release the active worktree slot after the executor lock drops; otherwise graph failure cleanup can overwrite the recovery and the parked task can keep consuming maxWorktrees capacity.
|
||||
*/
|
||||
const liveTask = await this.store.getTask(task.id);
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: liveTask.recoveryRetryCount,
|
||||
nextRecoveryAt: liveTask.nextRecoveryAt,
|
||||
});
|
||||
if (!decision.shouldRetry) {
|
||||
executorLog.error(`✗ ${task.id} stale assistant-continuation retries exhausted (${MAX_RECOVERY_RETRIES} attempts): ${errorMessage}`);
|
||||
await this.store.logEntry(
|
||||
task.id,
|
||||
`Stale assistant-continuation fresh-session retries exhausted after ${MAX_RECOVERY_RETRIES} attempts: ${errorMessage}`,
|
||||
errorStack ?? errorDetail,
|
||||
this.getRunContextFor(task.id),
|
||||
);
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "failed",
|
||||
error: errorMessage,
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
});
|
||||
await this.persistTokenUsage(task.id);
|
||||
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
|
||||
return;
|
||||
}
|
||||
|
||||
staleAssistantContinuationRequeue = true;
|
||||
const attempt = decision.nextState.recoveryRetryCount;
|
||||
const delay = formatDelay(decision.delayMs);
|
||||
executorLog.warn(`${task.id} stale assistant-continuation session detected — fresh-session retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay} after executor lock release`);
|
||||
await this.store.logEntry(
|
||||
task.id,
|
||||
`Detected stale assistant-continuation session — fresh-session retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay} with progress preserved: ${errorMessage}`,
|
||||
undefined,
|
||||
this.getRunContextFor(task.id),
|
||||
);
|
||||
await this.store.updateTask(task.id, {
|
||||
sessionFile: null,
|
||||
recoveryRetryCount: decision.nextState.recoveryRetryCount,
|
||||
nextRecoveryAt: decision.nextState.nextRecoveryAt,
|
||||
});
|
||||
return;
|
||||
} else if (errorMessage.includes("Invalid transition")) {
|
||||
// Task was moved by user/process while executor was running — already in desired state
|
||||
// This check must come before pausedAborted since it's more specific
|
||||
@@ -12489,6 +12544,54 @@ export class TaskExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
// Requeue stale assistant-continuation sessions AFTER this.executing is cleared.
|
||||
// Moving the task while the execution guard is still held can cause the scheduler's
|
||||
// task:moved dispatch to no-op, stranding the task in todo with no fresh run.
|
||||
if (staleAssistantContinuationRequeue) {
|
||||
/*
|
||||
FNXC:ExecutorSessionRecovery 2026-07-14-06:26:
|
||||
Claim the process-wide executor lock for deferred cleanup, release it immediately before moveTask emits task:moved, and always drop the claim on errors. This closes the guard-release race without recreating the original no-op dispatch: a fresh retry cannot start while stale state is being cleared, but can claim the task when the committed move event fires.
|
||||
|
||||
FNXC:ExecutorSessionRecovery 2026-07-14-06:34:
|
||||
Release the stale run's activeWorktrees slot before releasing the executor lock. Once the lock is open, the fresh retry may install its own slot while moveTask dispatches; deleting afterward would erase the new run's capacity and liveness tracking.
|
||||
*/
|
||||
const cleanupClaimed = executingTaskLock.tryClaim(task.id);
|
||||
if (!cleanupClaimed) {
|
||||
executorLog.log(`${task.id} stale assistant-continuation requeue skipped — a fresh executor already claimed the task`);
|
||||
} else {
|
||||
let cleanupLockHeld = true;
|
||||
try {
|
||||
const latestTask = await this.store.getTask(task.id);
|
||||
if (latestTask.column === "in-progress" || latestTask.column === "todo") {
|
||||
await this.store.updateTask(task.id, {
|
||||
sessionFile: null,
|
||||
status: null,
|
||||
error: null,
|
||||
});
|
||||
if (latestTask.column !== "todo") {
|
||||
this.markGraphExecuteSelfRequeued(task.id);
|
||||
this.activeWorktrees.delete(task.id);
|
||||
executingTaskLock.release(task.id);
|
||||
cleanupLockHeld = false;
|
||||
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
|
||||
} else {
|
||||
this.activeWorktrees.delete(task.id);
|
||||
}
|
||||
executorLog.log(`${task.id} stale assistant-continuation session cleared — requeued to todo with progress preserved`);
|
||||
} else {
|
||||
executorLog.log(`${task.id} stale assistant-continuation requeue skipped — task is now in '${latestTask.column}'`);
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
executorLog.error(`Failed to requeue stale assistant-continuation task ${task.id}: ${errorMessage}`);
|
||||
} finally {
|
||||
if (cleanupLockHeld) {
|
||||
executingTaskLock.release(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,
|
||||
|
||||
Reference in New Issue
Block a user