feat(FN-4249): complete Step 3 — add running-agent mismatch self-healing

Fusion-Task-Id: FN-4249
Fusion-Task-Lineage: e0052fae-38e1-4d66-8240-43d26f5790bb

packages/engine/src/__tests__/scheduler-overlap-requeue.test.ts

@@ -87,6 +87,7 @@ function createAgentStore(agents: MutableAgent[]): AgentStore {
     updateAgentState: vi.fn(async (id: string, state: Agent["state"]) => {
       const existing = byId.get(id);
       if (!existing) return;
+      if (existing.state === state) return;
       existing.state = state;
     }),
     syncExecutionTaskLink: vi.fn(async (id: string, taskId?: string | null) => {

packages/engine/src/__tests__/self-healing.test.ts

@@ -440,6 +440,7 @@ describe("SelfHealingManager", () => {
       const recoverOrphanedExecutions = vi.spyOn(manager, "recoverOrphanedExecutions").mockResolvedValue(1);
       const recoverApprovedTriageTasks = vi.spyOn(manager, "recoverApprovedTriageTasks").mockResolvedValue(1);
       const recoverOrphanedAgents = vi.spyOn(manager, "recoverOrphanedAgents").mockResolvedValue(1);
+      const recoverAgentsRunningOnInactiveTasks = vi.spyOn(manager, "recoverAgentsRunningOnInactiveTasks").mockResolvedValue(1);
       const clearStaleBlockedBy = vi.spyOn(manager, "clearStaleBlockedBy").mockResolvedValue(1);
 
       await manager.runStartupRecovery();
@@ -452,6 +453,7 @@ describe("SelfHealingManager", () => {
       expect(recoverOrphanedExecutions).toHaveBeenCalledTimes(1);
       expect(recoverApprovedTriageTasks).toHaveBeenCalledTimes(1);
       expect(recoverOrphanedAgents).toHaveBeenCalledTimes(1);
+      expect(recoverAgentsRunningOnInactiveTasks).toHaveBeenCalledTimes(1);
       expect(clearStaleBlockedBy).toHaveBeenCalledTimes(1);
     });
 
@@ -867,6 +869,58 @@ describe("SelfHealingManager", () => {
     });
   });
 
+  describe("recoverAgentsRunningOnInactiveTasks", () => {
+    it("recovers durable running agents linked to todo tasks", async () => {
+      const now = Date.now();
+      const agents: Agent[] = [
+        {
+          id: "agent-recover",
+          state: "running",
+          executionTaskId: "FN-TODO",
+          updatedAt: new Date(now - 120_000).toISOString(),
+        } as Agent,
+        {
+          id: "agent-keep",
+          state: "running",
+          executionTaskId: "FN-IP",
+          updatedAt: new Date(now - 120_000).toISOString(),
+        } as Agent,
+      ];
+
+      const getTask = vi.fn(async (taskId: string) => {
+        if (taskId === "FN-TODO") return { id: "FN-TODO", column: "todo" } as Task;
+        if (taskId === "FN-IP") return { id: "FN-IP", column: "in-progress" } as Task;
+        return null;
+      });
+
+      const agentStore = {
+        listAgents: vi.fn(async () => agents),
+        getActiveHeartbeatRun: vi.fn(async () => null),
+        updateAgentState: vi.fn(async (agentId: string, state: Agent["state"]) => {
+          const agent = agents.find((candidate) => candidate.id === agentId);
+          if (agent) agent.state = state;
+        }),
+        syncExecutionTaskLink: vi.fn(async (agentId: string, taskId?: string) => {
+          const agent = agents.find((candidate) => candidate.id === agentId);
+          if (agent) agent.executionTaskId = taskId;
+        }),
+      } as unknown as AgentStore;
+
+      const managerWithAgents = new SelfHealingManager(
+        createMockStore({ getTask }),
+        { rootDir: "/tmp/test-project", agentStore },
+      );
+
+      const recovered = await managerWithAgents.recoverAgentsRunningOnInactiveTasks();
+
+      expect(recovered).toBe(1);
+      expect(agentStore.updateAgentState).toHaveBeenCalledWith("agent-recover", "active");
+      expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-recover", undefined);
+      expect(agentStore.updateAgentState).not.toHaveBeenCalledWith("agent-keep", "active");
+      managerWithAgents.stop();
+    });
+  });
+
   describe("recoverStaleHeartbeatRuns", () => {
     function createMockAgentStore(activeRuns: Array<{ id: string; agentId: string; startedAt: string; processPid?: number; status?: string }>): {
       store: AgentStore;

packages/engine/src/agent-heartbeat.ts

@@ -863,6 +863,10 @@ export class HeartbeatMonitor {
    * Called on monitor start AND periodically from the polling loop to keep
    * the system self-healing across versions. Best-effort — failures are
    * logged but do not block the caller.
+   *
+   * Complements SelfHealingManager.recoverAgentsRunningOnInactiveTasks():
+   * heartbeat reconciliation handles stale/no-run conditions, while self-healing
+   * handles task-column mismatches (for example running agents linked to todo tasks).
    */
   private async reconcileOrphanedRunningAgents(): Promise<void> {
     try {

packages/engine/src/scheduler.ts

@@ -497,7 +497,7 @@ export class Scheduler {
 
     for (const agent of linkedAgents) {
       await agentStore.updateAgentState(agent.id, "active");
-      await agentStore.syncExecutionTaskLink(agent.id, null);
+      await agentStore.syncExecutionTaskLink(agent.id, undefined);
       schedulerLog.log(`Rolled back running agent ${agent.id} after overlap requeue of ${taskId}`);
     }
   }

packages/engine/src/self-healing.ts

@@ -132,6 +132,7 @@ const DEFAULT_STALE_MERGING_STATUS_MIN_AGE_MS = 5 * 60_000;
 const DURABLE_ERROR_RECOVERY_MAX_RETRIES = 5;
 const DURABLE_ERROR_RECOVERY_BASE_COOLDOWN_MS = 30_000;
 const DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS = 15 * 60_000;
+const RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS = 5 * 60_000;
 
 interface LandedTaskCommit {
   sha: string;
@@ -316,6 +317,7 @@ export class SelfHealingManager {
       { name: "orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks().then(() => undefined) },
       { name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents().then(() => undefined) },
       { name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns().then(() => undefined) },
+      { name: "recover-running-on-inactive-tasks", fn: () => this.recoverAgentsRunningOnInactiveTasks().then(() => undefined) },
       { name: "clear-stale-blocked-by", fn: () => this.clearStaleBlockedBy().then(() => undefined) },
     ];
 
@@ -958,6 +960,7 @@ export class SelfHealingManager {
           { name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
           { name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
           { name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
+          { name: "recover-running-on-inactive-tasks", fn: () => this.recoverAgentsRunningOnInactiveTasks() },
           { name: "clear-stale-blocked-by", fn: () => this.clearStaleBlockedBy() },
         ];
         for (const fn of batch2Fns) {
@@ -2354,6 +2357,43 @@ export class SelfHealingManager {
     return Math.min(exponential, DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS);
   }
 
+  async recoverAgentsRunningOnInactiveTasks(): Promise<number> {
+    const agentStore = this.options.agentStore;
+    if (!agentStore) {
+      return 0;
+    }
+
+    const now = Date.now();
+    const recoveredAgentIds = new Set<string>();
+    const runningAgents = await agentStore.listAgents({ state: "running", includeEphemeral: true });
+
+    for (const agent of runningAgents) {
+      if (isEphemeralAgent(agent) || !agent.executionTaskId) {
+        continue;
+      }
+
+      const linkedTask = await this.store.getTask(agent.executionTaskId);
+      if (linkedTask && (linkedTask.column === "in-progress" || linkedTask.column === "in-review" || linkedTask.column === "done" || linkedTask.column === "archived")) {
+        continue;
+      }
+
+      const activeRun = await agentStore.getActiveHeartbeatRun(agent.id);
+      const runStartedAt = activeRun?.startedAt;
+      const runAgeMs = runStartedAt ? now - Date.parse(runStartedAt) : Number.POSITIVE_INFINITY;
+      const hasFreshRun = Boolean(activeRun) && Number.isFinite(runAgeMs) && runAgeMs <= RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS;
+      if (hasFreshRun || this.options.hasActiveAgentExecution?.(agent.id) === true) {
+        continue;
+      }
+
+      await agentStore.updateAgentState(agent.id, "active");
+      await agentStore.syncExecutionTaskLink(agent.id, undefined);
+      recoveredAgentIds.add(agent.id);
+      log.log(`Recovered running durable agent ${agent.id} on inactive task ${agent.executionTaskId}`);
+    }
+
+    return recoveredAgentIds.size;
+  }
+
   async recoverOrphanedAgents(): Promise<number> {
     const agentStore = this.options.agentStore;
     if (!agentStore) {