feat(FN-3934): add restart recovery coordinator

Adds a new restart recovery coordinator (FN-3934) to manage task recovery on agent restart, including the core coordinator class, tests, and integration into the in-process runtime; also documents the coordinator in AGENTS.md.

Fusion-Task-Id: FN-3934

packages/engine/src/__tests__/restart-recovery-coordinator.test.ts

@@ -0,0 +1,68 @@
+import { describe, expect, it, vi } from "vitest";
+import type { TaskStore, Task } from "@fusion/core";
+import { RestartRecoveryCoordinator } from "../restart-recovery-coordinator.js";
+
+function createTask(overrides: Partial<Task>): Task {
+  return {
+    id: "FN-1",
+    description: "test",
+    column: "in-progress",
+    priority: "normal",
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
+    steps: [],
+    log: [],
+    dependencies: [],
+    attachments: [],
+    ...overrides,
+  } as Task;
+}
+
+describe("RestartRecoveryCoordinator", () => {
+  it("requeues interrupted failed tasks with no progress, then resumes remaining orphans", async () => {
+    const store = {
+      listTasks: vi.fn().mockResolvedValue([
+        createTask({ id: "FN-1", status: "failed", error: "Agent finished without calling fn_task_done", steps: [] }),
+        createTask({ id: "FN-2", steps: [{ id: "s1", title: "x", status: "done" }] as any }),
+      ]),
+      updateTask: vi.fn().mockResolvedValue({}),
+      logEntry: vi.fn().mockResolvedValue(undefined),
+      moveTask: vi.fn().mockResolvedValue(undefined),
+    } as unknown as TaskStore;
+
+    const executor = {
+      resumeOrphaned: vi.fn().mockResolvedValue(undefined),
+    } as any;
+
+    const coordinator = new RestartRecoveryCoordinator(store, executor);
+    await coordinator.recoverInterruptedRuns();
+
+    expect(store.updateTask).toHaveBeenCalledWith("FN-1", expect.objectContaining({ status: "stuck-killed" }));
+    expect(store.moveTask).toHaveBeenCalledWith("FN-1", "todo");
+    expect(executor.resumeOrphaned).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not requeue when step progress exists", async () => {
+    const store = {
+      listTasks: vi.fn().mockResolvedValue([
+        createTask({
+          id: "FN-9",
+          status: "failed",
+          error: "Agent finished without calling fn_task_done",
+          steps: [{ id: "s1", title: "x", status: "in-progress" }] as any,
+        }),
+      ]),
+      updateTask: vi.fn(),
+      logEntry: vi.fn(),
+      moveTask: vi.fn(),
+    } as unknown as TaskStore;
+
+    const executor = { resumeOrphaned: vi.fn().mockResolvedValue(undefined) } as any;
+    const coordinator = new RestartRecoveryCoordinator(store, executor);
+    await coordinator.recoverInterruptedRuns();
+
+    expect(store.updateTask).not.toHaveBeenCalled();
+    expect(store.moveTask).not.toHaveBeenCalled();
+    expect(executor.resumeOrphaned).toHaveBeenCalledTimes(1);
+  });
+});

packages/engine/src/restart-recovery-coordinator.ts

@@ -0,0 +1,62 @@
+import type { Task, TaskStore } from "@fusion/core";
+import type { TaskExecutor } from "./executor.js";
+import { createLogger } from "./logger.js";
+
+const log = createLogger("restart-recovery");
+
+function hasStepProgress(task: Task): boolean {
+  const steps = Array.isArray(task.steps) ? task.steps : [];
+  return steps.some((step) => step.status === "done" || step.status === "in-progress" || step.status === "skipped");
+}
+
+function isNoTaskDoneFailure(task: Task): boolean {
+  return task.status === "failed"
+    && typeof task.error === "string"
+    && task.error.toLowerCase().includes("without calling fn_task_done");
+}
+
+export class RestartRecoveryCoordinator {
+  constructor(
+    private readonly store: TaskStore,
+    private readonly executor: TaskExecutor,
+  ) {}
+
+  async recoverInterruptedRuns(): Promise<void> {
+    const allInProgress = await this.store.listTasks({ slim: true, column: "in-progress" });
+    const candidates = allInProgress.filter((task) => task.column === "in-progress" && !task.paused);
+
+    if (candidates.length === 0) return;
+
+    let requeued = 0;
+    for (const task of candidates) {
+      if (!this.mustSafeRetry(task)) continue;
+      await this.safeRequeue(task);
+      requeued++;
+    }
+
+    if (requeued > 0) {
+      log.log(`Restart recovery requeued ${requeued} interrupted task(s) for safe retry`);
+    }
+
+    await this.executor.resumeOrphaned();
+  }
+
+  private mustSafeRetry(task: Task): boolean {
+    return isNoTaskDoneFailure(task) && !hasStepProgress(task);
+  }
+
+  private async safeRequeue(task: Task): Promise<void> {
+    await this.store.updateTask(task.id, {
+      status: "stuck-killed",
+      worktree: null,
+      branch: null,
+      sessionFile: null,
+      error: null,
+    });
+    await this.store.logEntry(
+      task.id,
+      "Restart recovery: interrupted run had no step progress and no fn_task_done — requeued to todo for safe retry",
+    );
+    await this.store.moveTask(task.id, "todo");
+  }
+}

packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts

@@ -15,6 +15,7 @@ const {
   mockSelfHealingCtor,
   mockRecoverNoProgressNoTaskDoneFailures,
   mockRunStartupRecovery,
+  mockRecoverInterruptedRuns,
   mockExecutorCtor,
   mockResumeOrphaned,
   mockTaskStoreSettings,
@@ -31,6 +32,7 @@ const {
   mockSelfHealingCtor: vi.fn(),
   mockRecoverNoProgressNoTaskDoneFailures: vi.fn().mockResolvedValue(0),
   mockRunStartupRecovery: vi.fn().mockResolvedValue(undefined),
+  mockRecoverInterruptedRuns: vi.fn().mockResolvedValue(undefined),
   mockExecutorCtor: vi.fn(),
   mockResumeOrphaned: vi.fn().mockResolvedValue(undefined),
   mockTaskStoreSettings: {} as Record<string, unknown>,
@@ -157,6 +159,14 @@ vi.mock("../../self-healing.js", async () => {
   };
 });
 
+vi.mock("../../restart-recovery-coordinator.js", async () => {
+  return {
+    RestartRecoveryCoordinator: vi.fn().mockImplementation(() => ({
+      recoverInterruptedRuns: mockRecoverInterruptedRuns,
+    })),
+  };
+});
+
 // Mock the plugin runner
 vi.mock("../../plugin-runner.js", async () => {
   return {
@@ -320,11 +330,11 @@ describe("InProcessRuntime", () => {
       expect(mockSelfHealingStart).toHaveBeenCalled();
     }, 30000);
 
-    it("runs self-healing startup recovery immediately after orphan resume on startup", async () => {
+    it("runs startup recovery immediately after interrupted-run coordination on startup", async () => {
       await runtime.start();
 
-      expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
-      expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
+      expect(mockRecoverInterruptedRuns).toHaveBeenCalledTimes(1);
+      expect(mockResumeOrphaned).not.toHaveBeenCalled();
       expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
     }, 30000);
 
@@ -333,7 +343,7 @@ describe("InProcessRuntime", () => {
 
       await runtime.start();
 
-      expect(mockRecoverNoProgressNoTaskDoneFailures).not.toHaveBeenCalled();
+      expect(mockRecoverInterruptedRuns).not.toHaveBeenCalled();
       expect(mockResumeOrphaned).not.toHaveBeenCalled();
       expect(mockRunStartupRecovery).not.toHaveBeenCalled();
     }, 30000);
@@ -342,20 +352,17 @@ describe("InProcessRuntime", () => {
       mockTaskStoreSettings.enginePaused = true;
 
       await runtime.start();
-      mockRecoverNoProgressNoTaskDoneFailures.mockClear();
+      mockRecoverInterruptedRuns.mockClear();
       mockResumeOrphaned.mockClear();
       mockRunStartupRecovery.mockClear();
 
       mockTaskStoreSettings.enginePaused = false;
       await runtime.resumeAfterUnpause();
 
-      expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
-      expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
+      expect(mockRecoverInterruptedRuns).toHaveBeenCalledTimes(1);
+      expect(mockResumeOrphaned).not.toHaveBeenCalled();
       expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
-      expect(mockRecoverNoProgressNoTaskDoneFailures.mock.invocationCallOrder[0]).toBeLessThan(
-        mockResumeOrphaned.mock.invocationCallOrder[0],
-      );
-      expect(mockResumeOrphaned.mock.invocationCallOrder[0]).toBeLessThan(
+      expect(mockRecoverInterruptedRuns.mock.invocationCallOrder[0]).toBeLessThan(
         mockRunStartupRecovery.mock.invocationCallOrder[0],
       );
     }, 30000);
@@ -364,15 +371,15 @@ describe("InProcessRuntime", () => {
       mockTaskStoreSettings.enginePaused = true;
 
       await runtime.start();
-      mockRecoverNoProgressNoTaskDoneFailures.mockClear();
+      mockRecoverInterruptedRuns.mockClear();
       mockResumeOrphaned.mockClear();
       mockRunStartupRecovery.mockClear();
 
       mockTaskStoreSettings.enginePaused = false;
       await Promise.all([runtime.resumeAfterUnpause(), runtime.resumeAfterUnpause()]);
 
-      expect(mockRecoverNoProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1);
-      expect(mockResumeOrphaned).toHaveBeenCalledTimes(1);
+      expect(mockRecoverInterruptedRuns).toHaveBeenCalledTimes(1);
+      expect(mockResumeOrphaned).not.toHaveBeenCalled();
       expect(mockRunStartupRecovery).toHaveBeenCalledTimes(1);
     }, 30000);
 

packages/engine/src/runtimes/in-process-runtime.ts

@@ -33,6 +33,7 @@ import { runtimeLog } from "../logger.js";
 import { StuckTaskDetector } from "../stuck-task-detector.js";
 import type { UsageLimitPauser } from "../usage-limit-detector.js";
 import { SelfHealingManager } from "../self-healing.js";
+import { RestartRecoveryCoordinator } from "../restart-recovery-coordinator.js";
 import { MeshLeaseManager } from "../mesh-lease-manager.js";
 import { PluginRunner } from "../plugin-runner.js";
 import { MissionAutopilot } from "../mission-autopilot.js";
@@ -125,6 +126,7 @@ export class InProcessRuntime
   private startupRecoveryDeferred = false;
   /** Prevent duplicate unpause recovery dispatches from racing each other. */
   private resumeAfterUnpauseRunning = false;
+  private restartRecoveryCoordinator?: RestartRecoveryCoordinator;
 
   /**
    * @param config - Runtime configuration
@@ -636,6 +638,7 @@ export class InProcessRuntime
       });
       this.selfHealingManager.start();
       this.stuckTaskDetector.start();
+      this.restartRecoveryCoordinator = new RestartRecoveryCoordinator(this.taskStore, this.executor);
 
       // 8. Set up event forwarding from TaskStore
       this.setupEventForwarding();
@@ -912,13 +915,9 @@ export class InProcessRuntime
   }
 
   private async resumeStartupRecoverySequence(): Promise<void> {
-    // Requeue no-progress no-task_done failures before resumeOrphaned can
-    // restart other orphaned executions.
-    await this.selfHealingManager!.recoverNoProgressNoTaskDoneFailures();
-
-    // Resume orphaned in-progress tasks before the broader self-healing scan
-    // so the executor can claim or fast-path eligible tasks first.
-    await this.executor!.resumeOrphaned();
+    // Restart recovery decides when interrupted runs can safely resume versus
+    // when they must be reset to todo for a clean retry.
+    await this.restartRecoveryCoordinator!.recoverInterruptedRuns();
 
     // Some "stuck" tasks are already orphaned by the time the runtime boots:
     // they no longer have a tracked session/worktree, so the stuck detector