fix: auto-recover orphaned heartbeat runs from crashed processes

When the dashboard crashes mid-heartbeat, the agentRuns row is left in
status='active' forever. HeartbeatTriggerScheduler.onTimerTick treats
any active run as "still running" and skips every subsequent tick, so
agents go silent indefinitely (observed: 6+ hours). The existing
in-memory missed-heartbeat watchdog can't help — its trackedAgents map
is wiped on process restart.

SelfHealingManager.recoverStaleHeartbeatRuns now reconciles these on
startup and during periodic maintenance: terminates active runs whose
processPid does not match the current process, has no recorded pid, or
has been active for more than 6 hours.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.changeset/auto-recover-orphan-heartbeat-runs.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Self-heal orphaned `agentRuns` rows left in `status='active'` when the dashboard process crashes mid-heartbeat. The trigger scheduler treats any active run as "still running" and silently skips every subsequent tick, so a single crashed run could leave an agent without heartbeats for hours. SelfHealingManager now reconciles these on startup and during periodic maintenance, terminating runs whose `processPid` does not match the current process or whose age exceeds 6 hours.

packages/core/src/agent-store.ts

@@ -1743,6 +1743,25 @@ export class AgentStore extends EventEmitter {
       .sort((a, b) => new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime());
   }
 
+  /**
+   * List every heartbeat run currently in `status = 'active'` across all
+   * agents. Used by self-healing to detect orphaned runs from prior process
+   * incarnations that crashed before calling endHeartbeatRun(). Without this
+   * sweep an active row blocks all subsequent timer ticks for the agent
+   * because HeartbeatTriggerScheduler.onTimerTick treats any active run as
+   * "already running".
+   */
+  async listActiveHeartbeatRuns(): Promise<AgentHeartbeatRun[]> {
+    const rows = this.db.prepare(`
+      SELECT data FROM agentRuns
+      WHERE status = 'active'
+      ORDER BY startedAt ASC
+    `).all() as Array<{ data: string }>;
+    return rows
+      .map((row) => this.parseJson<AgentHeartbeatRun | null>(row.data, null))
+      .filter((run): run is AgentHeartbeatRun => run !== null);
+  }
+
   // ─────────────────────────────────────────────────────────────────────────
   // Task Session Management
   // ─────────────────────────────────────────────────────────────────────────

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

@@ -594,6 +594,108 @@ describe("SelfHealingManager", () => {
     });
   });
 
+  describe("recoverStaleHeartbeatRuns", () => {
+    function createMockAgentStore(activeRuns: Array<{ id: string; agentId: string; startedAt: string; processPid?: number; status?: string }>): {
+      store: AgentStore;
+      ended: Array<{ runId: string; status: string }>;
+      saved: Array<Partial<{ id: string; status: string; stderrExcerpt: string }>>;
+    } {
+      const ended: Array<{ runId: string; status: string }> = [];
+      const saved: Array<Partial<{ id: string; status: string; stderrExcerpt: string }>> = [];
+      const detailById = new Map<string, any>();
+      for (const r of activeRuns) {
+        detailById.set(r.id, { id: r.id, agentId: r.agentId, startedAt: r.startedAt, endedAt: null, status: r.status ?? "active", processPid: r.processPid });
+      }
+      const agentStore = {
+        listActiveHeartbeatRuns: vi.fn().mockResolvedValue(
+          activeRuns.map((r) => ({ id: r.id, agentId: r.agentId, startedAt: r.startedAt, endedAt: null, status: "active" as const, processPid: r.processPid })),
+        ),
+        getRunDetail: vi.fn().mockImplementation((_agentId: string, runId: string) => Promise.resolve(detailById.get(runId) ?? null)),
+        saveRun: vi.fn().mockImplementation((run: any) => {
+          saved.push({ id: run.id, status: run.status, stderrExcerpt: run.stderrExcerpt });
+          return Promise.resolve();
+        }),
+        endHeartbeatRun: vi.fn().mockImplementation((runId: string, status: string) => {
+          ended.push({ runId, status });
+          return Promise.resolve();
+        }),
+      } as unknown as AgentStore;
+      return { store: agentStore, ended, saved };
+    }
+
+    it("returns 0 when no agentStore is configured", async () => {
+      const result = await manager.recoverStaleHeartbeatRuns();
+      expect(result).toBe(0);
+    });
+
+    it("terminates active runs whose processPid does not match this process", async () => {
+      const { store: agentStore, ended, saved } = createMockAgentStore([
+        { id: "run-orphan", agentId: "agent-a", startedAt: new Date().toISOString(), processPid: 999_999 },
+      ]);
+      const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
+
+      const result = await m.recoverStaleHeartbeatRuns();
+
+      expect(result).toBe(1);
+      expect(ended).toEqual([{ runId: "run-orphan", status: "terminated" }]);
+      expect(saved[0]?.status).toBe("terminated");
+      expect(saved[0]?.stderrExcerpt).toMatch(/Auto-recovered orphaned heartbeat run/);
+      m.stop();
+    });
+
+    it("leaves young runs from the current process alone", async () => {
+      const { store: agentStore, ended } = createMockAgentStore([
+        { id: "run-mine", agentId: "agent-b", startedAt: new Date().toISOString(), processPid: process.pid },
+      ]);
+      const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
+
+      const result = await m.recoverStaleHeartbeatRuns();
+
+      expect(result).toBe(0);
+      expect(ended).toEqual([]);
+      m.stop();
+    });
+
+    it("terminates legacy active runs that have no recorded processPid", async () => {
+      const { store: agentStore, ended } = createMockAgentStore([
+        { id: "run-legacy", agentId: "agent-c", startedAt: new Date().toISOString() },
+      ]);
+      const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
+
+      const result = await m.recoverStaleHeartbeatRuns();
+
+      expect(result).toBe(1);
+      expect(ended[0]?.runId).toBe("run-legacy");
+      m.stop();
+    });
+
+    it("terminates current-process runs that exceed the max-age threshold", async () => {
+      const tooOld = new Date(Date.now() - 7 * 60 * 60 * 1000).toISOString(); // 7h ago
+      const { store: agentStore, ended } = createMockAgentStore([
+        { id: "run-stuck", agentId: "agent-d", startedAt: tooOld, processPid: process.pid },
+      ]);
+      const m = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore });
+
+      const result = await m.recoverStaleHeartbeatRuns();
+
+      expect(result).toBe(1);
+      expect(ended[0]?.runId).toBe("run-stuck");
+      m.stop();
+    });
+
+    it("runStartupRecovery includes the stale heartbeat runs step", async () => {
+      vi.mocked(store.getSettings).mockResolvedValue({
+        globalPause: false,
+        enginePaused: false,
+      } as unknown as Settings);
+      const spy = vi.spyOn(manager, "recoverStaleHeartbeatRuns").mockResolvedValue(0);
+
+      await manager.runStartupRecovery();
+
+      expect(spy).toHaveBeenCalledTimes(1);
+    });
+  });
+
   describe("recoverNoProgressNoTaskDoneFailures", () => {
     it("requeues clean in-progress no-task_done failures with no step progress", async () => {
       const managerWithRecovery = new SelfHealingManager(store, {
@@ -3254,13 +3356,14 @@ describe("maintenance cycle concurrency", () => {
     makeSlow("recoverOrphanedPlanningTasks");
     makeSlow("recoverGhostReviewTasks");
     makeSlow("recoverOrphanedAgents");
+    makeSlow("recoverStaleHeartbeatRuns");
 
     await (manager as any).runMaintenance();
 
     // Operations run sequentially (one at a time), not in parallel.
     expect(maxConcurrent).toBe(1);
     // All operations should have run (including last one)
-    expect(executionOrder[executionOrder.length - 1]).toBe("recoverOrphanedAgents");
+    expect(executionOrder[executionOrder.length - 1]).toBe("recoverStaleHeartbeatRuns");
   });
 
   it("one failing batch 2 operation does not abort the batch", async () => {
@@ -3278,6 +3381,7 @@ describe("maintenance cycle concurrency", () => {
       "recoverOrphanedPlanningTasks",
       "recoverGhostReviewTasks",
       "recoverOrphanedAgents",
+      "recoverStaleHeartbeatRuns",
     ] as const;
 
     // Make one operation fail

packages/engine/src/self-healing.ts

@@ -204,6 +204,7 @@ export class SelfHealingManager {
       { name: "approved-triage", fn: () => this.recoverApprovedTriageTasks().then(() => undefined) },
       { 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) },
     ];
 
     for (const step of steps) {
@@ -658,6 +659,7 @@ export class SelfHealingManager {
           { name: "recover-orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks() },
           { name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
           { name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
+          { name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
         ];
         for (const fn of batch2Fns) {
           try {
@@ -1519,6 +1521,104 @@ export class SelfHealingManager {
     }
   }
 
+  /**
+   * Default cap (in ms) on how long an active heartbeat run from the current
+   * process is allowed to remain open before self-healing will terminate it.
+   * Six hours is well past any legitimate heartbeat tick (default 1 h
+   * interval, configurable up to a few hours) so reaching this threshold
+   * means the run record was never closed — typically a process that died
+   * without our watchdog catching it.
+   */
+  private static readonly STALE_ACTIVE_RUN_MAX_AGE_MS = 6 * 60 * 60 * 1000;
+
+  /**
+   * Terminate orphaned `agentRuns` rows left in `status = 'active'` by a
+   * process that crashed before calling endHeartbeatRun(). These rows
+   * silently break heartbeat scheduling: HeartbeatTriggerScheduler.onTimerTick
+   * skips every tick that finds an active run, so the agent never gets called
+   * again until something cleans up.
+   *
+   * A run is considered stale when:
+   *  - `processPid` was recorded and does not match the current `process.pid`
+   *    (i.e., the writer process is gone — guaranteed orphan), or
+   *  - `processPid` is missing (legacy data), or
+   *  - the run has been active for longer than STALE_ACTIVE_RUN_MAX_AGE_MS,
+   *    even from the current process (defense in depth against a writer that
+   *    leaks the row without crashing the whole runtime).
+   *
+   * The matching `processPid` + young run case is left alone — that is a
+   * legitimately in-flight heartbeat.
+   */
+  async recoverStaleHeartbeatRuns(): Promise<number> {
+    const agentStore = this.options.agentStore;
+    if (!agentStore) {
+      return 0;
+    }
+
+    let activeRuns;
+    try {
+      activeRuns = await agentStore.listActiveHeartbeatRuns();
+    } catch (err: unknown) {
+      const errorMessage = err instanceof Error ? err.message : String(err);
+      log.error(`Stale heartbeat run recovery — listing failed: ${errorMessage}`);
+      return 0;
+    }
+
+    if (activeRuns.length === 0) {
+      return 0;
+    }
+
+    const now = Date.now();
+    const currentPid = process.pid;
+    const maxAgeMs = SelfHealingManager.STALE_ACTIVE_RUN_MAX_AGE_MS;
+    let recovered = 0;
+
+    for (const run of activeRuns) {
+      const startedMs = Date.parse(run.startedAt);
+      const ageMs = Number.isFinite(startedMs) ? Math.max(0, now - startedMs) : Infinity;
+      const recordedPid = run.processPid;
+
+      const pidMismatch = typeof recordedPid === "number" && recordedPid !== currentPid;
+      const pidMissing = typeof recordedPid !== "number";
+      const tooOld = ageMs >= maxAgeMs;
+
+      if (!pidMismatch && !pidMissing && !tooOld) {
+        continue;
+      }
+
+      const reason = pidMismatch
+        ? `writer pid ${recordedPid} is no longer this process (current pid ${currentPid})`
+        : pidMissing
+          ? `no processPid recorded`
+          : `active for ${Math.round(ageMs / 1000)}s (>= ${Math.round(maxAgeMs / 1000)}s threshold)`;
+
+      try {
+        const detail = await agentStore.getRunDetail(run.agentId, run.id);
+        if (detail) {
+          await agentStore.saveRun({
+            ...detail,
+            endedAt: new Date().toISOString(),
+            status: "terminated",
+            stderrExcerpt: `Auto-recovered orphaned heartbeat run: ${reason}`,
+          });
+        }
+        await agentStore.endHeartbeatRun(run.id, "terminated");
+        log.log(
+          `Auto-recovered: orphan heartbeat run ${run.id} for ${run.agentId} (${reason})`,
+        );
+        recovered++;
+      } catch (err: unknown) {
+        const errorMessage = err instanceof Error ? err.message : String(err);
+        log.error(`Failed to recover stale heartbeat run ${run.id} for ${run.agentId}: ${errorMessage}`);
+      }
+    }
+
+    if (recovered > 0) {
+      log.log(`Recovered ${recovered} stale heartbeat run(s)`);
+    }
+    return recovered;
+  }
+
   /**
    * Recover `in-progress` tasks that failed only because the agent exited
    * without calling task_done, and where there is no sign of work to preserve.