From 30a83f21fcbbdfc00d9852e44359a7520090c8df Mon Sep 17 00:00:00 2001 From: Phil Larson Date: Tue, 14 Jul 2026 08:21:19 -0700 Subject: [PATCH] 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` ## 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. --- .changeset/calm-taxis-requeue.md | 7 ++ .../__tests__/executor-step-session.test.ts | 105 ++++++++++++++++++ packages/engine/src/executor.ts | 103 +++++++++++++++++ 3 files changed, 215 insertions(+) create mode 100644 .changeset/calm-taxis-requeue.md diff --git a/.changeset/calm-taxis-requeue.md b/.changeset/calm-taxis-requeue.md new file mode 100644 index 0000000000..42fd439ad6 --- /dev/null +++ b/.changeset/calm-taxis-requeue.md @@ -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. diff --git a/packages/engine/src/__tests__/executor-step-session.test.ts b/packages/engine/src/__tests__/executor-step-session.test.ts index 8e02e31de2..9de4e975b3 100644 --- a/packages/engine/src/__tests__/executor-step-session.test.ts +++ b/packages/engine/src/__tests__/executor-step-session.test.ts @@ -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(); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 2ee50cb347..d35a4ac32f 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -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): 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,