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 -->

.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.

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();

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<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,