From a9a1f4c20978adf6508f1802d7ef7533eb4716a8 Mon Sep 17 00:00:00 2001 From: Fusion Date: Mon, 4 May 2026 16:41:00 -0700 Subject: [PATCH] feat(FN-3323): recover orphaned agents and add reports health section to he The merge lands FN-3323's self-healing system: a new "reports health" prompt section and `agent-heartbeat.ts` module that detect orphaned agents and recover their state (clearing stale error states and re-enabling heartbeat execution). The engine's `self-healing.ts` was expanded with wired recovery Fusion-Task-Id: FN-3323 --- .../src/__tests__/heartbeat-executor.test.ts | 81 ++++++++++ .../src/__tests__/heartbeat-test-helpers.ts | 1 + .../engine/src/__tests__/self-healing.test.ts | 140 +++++++++++++++++- packages/engine/src/agent-heartbeat.ts | 82 ++++++++++ .../engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/self-healing.ts | 76 +++++++++- 6 files changed, 378 insertions(+), 3 deletions(-) diff --git a/packages/engine/src/__tests__/heartbeat-executor.test.ts b/packages/engine/src/__tests__/heartbeat-executor.test.ts index b014d968a..7c4636684 100644 --- a/packages/engine/src/__tests__/heartbeat-executor.test.ts +++ b/packages/engine/src/__tests__/heartbeat-executor.test.ts @@ -159,6 +159,7 @@ describe("executeHeartbeat", () => { setLastBlockedState: vi.fn().mockResolvedValue(undefined), clearLastBlockedState: vi.fn().mockResolvedValue(undefined), appendRunLog: vi.fn().mockResolvedValue(undefined), + getAgentsByReportsTo: vi.fn().mockResolvedValue([]), } as unknown as AgentStore; } @@ -171,6 +172,86 @@ describe("executeHeartbeat", () => { vi.restoreAllMocks(); }); + describe("reports health check", () => { + it("buildReportsHealthSection returns null when agent has no reports", async () => { + const store = createStoreWithAgentForExec(); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" }); + + const result = await (monitor as any).buildReportsHealthSection("agent-001", store); + expect(result).toBeNull(); + }); + + it("buildReportsHealthSection returns formatted table for healthy reports", async () => { + const now = new Date().toISOString(); + const store = createStoreWithAgentForExec(); + vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([ + { id: "agent-002", name: "agent-2", state: "active", taskId: "FN-100", lastHeartbeatAt: now, updatedAt: now } as Agent, + { id: "agent-003", name: "agent-3", state: "running", taskId: "FN-101", lastHeartbeatAt: now, updatedAt: now } as Agent, + ]); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" }); + + const section = await (monitor as any).buildReportsHealthSection("agent-001", store); + expect(section).toContain("## Reports Health Check"); + expect(section).toContain("agent-2"); + expect(section).toContain("agent-3"); + expect(section).toContain("| Name | State | Task | Last Heartbeat | Health |"); + expect(section).toContain("healthy"); + }); + + it("buildReportsHealthSection classifies stuck agents", async () => { + const now = Date.now(); + const store = createStoreWithAgentForExec(); + vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([ + { id: "agent-002", name: "agent-2", state: "error", taskId: "FN-100", lastHeartbeatAt: new Date(now - 1000).toISOString(), updatedAt: new Date(now - 1000).toISOString(), lastError: "boom" } as Agent, + ]); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp", heartbeatTimeoutMs: 60_000 }); + + const section = await (monitor as any).buildReportsHealthSection("agent-001", store); + expect(section).toContain("**stuck**"); + expect(section).toContain("Actions for Unresponsive Reports"); + }); + + it("buildReportsHealthSection classifies stale agents", async () => { + const now = Date.now(); + const store = createStoreWithAgentForExec(); + vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([ + { id: "agent-003", name: "agent-3", state: "active", taskId: "FN-101", lastHeartbeatAt: new Date(now - 3 * 60 * 60 * 1000).toISOString(), updatedAt: new Date(now - 3 * 60 * 60 * 1000).toISOString() } as Agent, + ]); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp", heartbeatTimeoutMs: 60_000 }); + + const section = await (monitor as any).buildReportsHealthSection("agent-001", store); + expect(section).toContain("**stale**"); + }); + + it("executeHeartbeat includes reports health section when agent has reports", async () => { + const store = createStoreWithAgentForExec({ taskId: "FN-001" }); + const now = new Date().toISOString(); + vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([ + { id: "agent-010", name: "reporter", state: "running", taskId: "FN-200", lastHeartbeatAt: now, updatedAt: now } as Agent, + ]); + const mockSession = createMockAgentSession(); + mockedCreateFnAgent.mockResolvedValue({ session: mockSession as any }); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" }); + + await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" }); + const executionPrompt = mockSession.prompt.mock.calls.at(-1)?.[0] as string; + expect(executionPrompt).toContain("## Reports Health Check"); + expect(executionPrompt).toContain("reporter"); + }); + + it("executeHeartbeat omits reports health section when agent has no reports", async () => { + const store = createStoreWithAgentForExec({ taskId: "FN-001" }); + vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([]); + const mockSession = createMockAgentSession(); + mockedCreateFnAgent.mockResolvedValue({ session: mockSession as any }); + const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" }); + + await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" }); + const executionPrompt = mockSession.prompt.mock.calls.at(-1)?.[0] as string; + expect(executionPrompt).not.toContain("## Reports Health Check"); + }); + }); + describe("dependency validation", () => { it("throws when taskStore is not configured", async () => { const store = createStoreWithAgentForExec(); diff --git a/packages/engine/src/__tests__/heartbeat-test-helpers.ts b/packages/engine/src/__tests__/heartbeat-test-helpers.ts index 94e725ac3..27f4040b9 100644 --- a/packages/engine/src/__tests__/heartbeat-test-helpers.ts +++ b/packages/engine/src/__tests__/heartbeat-test-helpers.ts @@ -6,6 +6,7 @@ export function createMockStore(overrides: Partial = {}): AgentStore return { recordHeartbeat: vi.fn().mockResolvedValue(undefined), updateAgentState: vi.fn().mockResolvedValue(undefined), + getAgentsByReportsTo: vi.fn().mockResolvedValue([]), ...overrides, } as unknown as AgentStore; } diff --git a/packages/engine/src/__tests__/self-healing.test.ts b/packages/engine/src/__tests__/self-healing.test.ts index 332599a21..73784f88d 100644 --- a/packages/engine/src/__tests__/self-healing.test.ts +++ b/packages/engine/src/__tests__/self-healing.test.ts @@ -65,7 +65,7 @@ vi.mock("../logger.js", () => ({ })); import { SelfHealingManager } from "../self-healing.js"; -import type { TaskStore, Settings, Task } from "@fusion/core"; +import type { TaskStore, Settings, Task, AgentStore, Agent } from "@fusion/core"; import { EventEmitter } from "node:events"; import { execSync } from "node:child_process"; import { existsSync } from "node:fs"; @@ -438,6 +438,7 @@ describe("SelfHealingManager", () => { const recoverPartialProgressNoTaskDoneFailures = vi.spyOn(manager, "recoverPartialProgressNoTaskDoneFailures").mockResolvedValue(1); const recoverOrphanedExecutions = vi.spyOn(manager, "recoverOrphanedExecutions").mockResolvedValue(1); const recoverApprovedTriageTasks = vi.spyOn(manager, "recoverApprovedTriageTasks").mockResolvedValue(1); + const recoverOrphanedAgents = vi.spyOn(manager, "recoverOrphanedAgents").mockResolvedValue(1); await manager.runStartupRecovery(); @@ -447,6 +448,7 @@ describe("SelfHealingManager", () => { expect(recoverPartialProgressNoTaskDoneFailures).toHaveBeenCalledTimes(1); expect(recoverOrphanedExecutions).toHaveBeenCalledTimes(1); expect(recoverApprovedTriageTasks).toHaveBeenCalledTimes(1); + expect(recoverOrphanedAgents).toHaveBeenCalledTimes(1); }); it("runStartupRecovery skips while enginePaused is active", async () => { @@ -462,6 +464,136 @@ describe("SelfHealingManager", () => { }); }); + describe("recoverOrphanedAgents", () => { + function createMockAgentStore(agents: Agent[]): AgentStore { + return { + listAgents: vi.fn().mockResolvedValue(agents), + updateAgentState: vi.fn().mockResolvedValue(undefined), + updateAgent: vi.fn().mockResolvedValue(undefined), + } as unknown as AgentStore; + } + + it("returns 0 when no agentStore", async () => { + const result = await manager.recoverOrphanedAgents(); + expect(result).toBe(0); + }); + + it("skips agents with valid manager", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { id: "manager-1", state: "active", updatedAt: new Date(now).toISOString() } as Agent, + { id: "report-1", state: "error", reportsTo: "manager-1", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(0); + expect(agentStore.updateAgent).not.toHaveBeenCalled(); + managerWithAgents.stop(); + }); + + it("recovers orphaned agent in error state", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { id: "orphan-1", state: "error", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(1); + expect(agentStore.updateAgentState).toHaveBeenCalledWith("orphan-1", "active"); + expect(agentStore.updateAgent).toHaveBeenCalledWith("orphan-1", { lastError: undefined }); + managerWithAgents.stop(); + }); + + it("skips agents within grace period", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { id: "orphan-1", state: "error", updatedAt: new Date(now - 10_000).toISOString() } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(0); + expect(agentStore.updateAgent).not.toHaveBeenCalled(); + managerWithAgents.stop(); + }); + + it("skips ephemeral agents", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { + id: "ephemeral-1", + name: "ephemeral-1", + role: "executor", + state: "error", + createdAt: new Date(now - 240_000).toISOString(), + updatedAt: new Date(now - 120_000).toISOString(), + metadata: { agentKind: "task-worker" }, + } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(0); + expect(agentStore.updateAgent).not.toHaveBeenCalled(); + managerWithAgents.stop(); + }); + + it("recovers agent whose manager was deleted", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { id: "orphan-2", state: "running", reportsTo: "missing-manager", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(1); + expect(agentStore.updateAgentState).toHaveBeenCalledWith("orphan-2", "active"); + expect(agentStore.updateAgent).toHaveBeenCalledWith("orphan-2", { lastError: undefined }); + managerWithAgents.stop(); + }); + + it("ignores agents in healthy states", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ taskStuckTimeoutMs: 60_000 } as unknown as Settings); + const now = Date.now(); + const agentStore = createMockAgentStore([ + { id: "agent-a", state: "active", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + { id: "agent-b", state: "idle", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + { id: "agent-c", state: "paused", updatedAt: new Date(now - 120_000).toISOString() } as Agent, + ]); + const managerWithAgents = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore }); + + const result = await managerWithAgents.recoverOrphanedAgents(); + + expect(result).toBe(0); + expect(agentStore.updateAgent).not.toHaveBeenCalled(); + managerWithAgents.stop(); + }); + + it("runStartupRecovery includes orphaned agents step", async () => { + vi.mocked(store.getSettings).mockResolvedValue({ + globalPause: false, + enginePaused: false, + } as unknown as Settings); + const recoverOrphanedAgents = vi.spyOn(manager, "recoverOrphanedAgents").mockResolvedValue(1); + + await manager.runStartupRecovery(); + + expect(recoverOrphanedAgents).toHaveBeenCalledTimes(1); + }); + }); + describe("recoverNoProgressNoTaskDoneFailures", () => { it("requeues clean in-progress no-task_done failures with no step progress", async () => { const managerWithRecovery = new SelfHealingManager(store, { @@ -2996,13 +3128,15 @@ describe("maintenance cycle concurrency", () => { makeSlow("recoverOrphanedExecutions"); makeSlow("recoverApprovedTriageTasks"); makeSlow("recoverOrphanedPlanningTasks"); + makeSlow("recoverGhostReviewTasks"); + makeSlow("recoverOrphanedAgents"); 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("recoverOrphanedPlanningTasks"); + expect(executionOrder[executionOrder.length - 1]).toBe("recoverOrphanedAgents"); }); it("one failing batch 2 operation does not abort the batch", async () => { @@ -3018,6 +3152,8 @@ describe("maintenance cycle concurrency", () => { "recoverOrphanedExecutions", "recoverApprovedTriageTasks", "recoverOrphanedPlanningTasks", + "recoverGhostReviewTasks", + "recoverOrphanedAgents", ] as const; // Make one operation fail diff --git a/packages/engine/src/agent-heartbeat.ts b/packages/engine/src/agent-heartbeat.ts index 2d7649914..495b0e1fb 100644 --- a/packages/engine/src/agent-heartbeat.ts +++ b/packages/engine/src/agent-heartbeat.ts @@ -127,6 +127,15 @@ export function formatDuration(ms: number): string { return `${minutes}m`; } +function formatRelativeTime(iso?: string | null): string { + if (!iso) return "never"; + const parsed = Date.parse(iso); + if (!Number.isFinite(parsed)) return "unknown"; + const elapsed = Date.now() - parsed; + if (elapsed < 0) return "just now"; + return `${formatDuration(elapsed)} ago`; +} + /** Compare blocked-state snapshots to decide whether blocked messaging is duplicate noise. */ export function isBlockedStateDuplicate(current: BlockedStateSnapshot, previous: BlockedStateSnapshot): boolean { return current.blockedBy === previous.blockedBy && current.contextHash === previous.contextHash; @@ -1510,6 +1519,7 @@ export class HeartbeatMonitor { const customProcedure = await resolveAgentHeartbeatProcedure(agent, rootDir); const heartbeatProcedureText = customProcedure ?? (isNoTaskRun ? HEARTBEAT_NO_TASK_PROCEDURE : HEARTBEAT_PROCEDURE); + const reportsHealthSection = await this.buildReportsHealthSection(agent.id, this.store); if (isNoTaskRun) { // No-task heartbeat: agent has identity but no assigned task @@ -1578,6 +1588,7 @@ export class HeartbeatMonitor { "", "Your soul, instructions, and memory are already loaded in the system prompt.", "Focus on work that benefits the project without requiring a specific task context.", + ...(reportsHealthSection ? ["", reportsHealthSection] : []), "", "Call fn_heartbeat_done when finished.", ].join("\n"); @@ -1665,6 +1676,7 @@ export class HeartbeatMonitor { taskDetail!.prompt ? `PROMPT.md:\n${taskDetail!.prompt}` : "No PROMPT.md available.", ...triggeringCommentLines, ...pendingMessagesLines, + ...(reportsHealthSection ? ["", reportsHealthSection] : []), "", "Run the Heartbeat Procedure above. Call fn_heartbeat_done when finished.", ].join("\n"); @@ -1824,6 +1836,76 @@ export class HeartbeatMonitor { }); } + private async buildReportsHealthSection(agentId: string, agentStore: AgentStore): Promise { + const getReports = (agentStore as AgentStore & { getAgentsByReportsTo?: (id: string) => Promise }).getAgentsByReportsTo; + if (typeof getReports !== "function") { + return null; + } + + let reports: Agent[]; + try { + reports = await getReports(agentId); + } catch (err) { + heartbeatLog.warn(`Failed to load reports for ${agentId}: ${err instanceof Error ? err.message : String(err)}`); + return null; + } + if (reports.length === 0) { + return null; + } + + const now = Date.now(); + const rows = reports.map((report) => { + const timeoutMs = this.resolveAgentConfig(report.id).heartbeatTimeoutMs; + const lastHeartbeatTs = report.lastHeartbeatAt ? Date.parse(report.lastHeartbeatAt) : NaN; + const heartbeatAgeMs = Number.isFinite(lastHeartbeatTs) ? Math.max(0, now - lastHeartbeatTs) : Infinity; + + let health = "healthy"; + if (report.state === "paused") { + health = report.pauseReason ? `paused (${report.pauseReason})` : "paused"; + } else if (report.state === "terminated") { + health = "terminated"; + } else if (report.state === "error") { + health = "**stuck**"; + } else if (report.state === "running") { + health = heartbeatAgeMs <= timeoutMs * 2 ? "healthy" : "**stuck**"; + } else if ((report.state === "active" || report.state === "idle") && heartbeatAgeMs > timeoutMs * 3) { + health = "**stale**"; + } + + const task = report.taskId ?? "—"; + const state = report.state; + const heartbeat = formatRelativeTime(report.lastHeartbeatAt); + return `| ${report.name} | ${state} | ${task} | ${heartbeat} | ${health} |`; + }); + + const hasStuck = rows.some((row) => row.includes("**stuck**")); + const hasStale = rows.some((row) => row.includes("**stale**")); + const hasTerminated = rows.some((row) => row.includes("terminated")); + + const actionLines = ["### Actions for Unresponsive Reports"]; + if (hasStuck) { + actionLines.push("- For **stuck** reports: consider sending a message via fn_send_message asking for status, or reassigning their task via fn_delegate_task to a healthy agent."); + } + if (hasStale) { + actionLines.push("- For **stale** reports: the agent may have lost its heartbeat trigger — create a follow-up task to investigate."); + } + if (hasTerminated) { + actionLines.push("- For **terminated** reports: if they had active work, reassign their tasks or spawn replacement agents."); + } + + return [ + "## Reports Health Check", + "", + `You have ${reports.length} agent(s) reporting to you. Review their status and intervene if any are unresponsive.`, + "", + "| Name | State | Task | Last Heartbeat | Health |", + "|------|-------|------|----------------|--------|", + ...rows, + "", + ...actionLines, + ].join("\n"); + } + // ───────────────────────────────────────────────────────────────────────── // Heartbeat tools: createHeartbeatTools / clearRunState // ───────────────────────────────────────────────────────────────────────── diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 87eade2bd..5404c0a86 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -668,6 +668,7 @@ export class InProcessRuntime // 7. Initialize SelfHealingManager this.selfHealingManager = new SelfHealingManager(this.taskStore, { rootDir: this.config.workingDirectory, + agentStore: this.agentStore, recoverCompletedTask: (task) => this.executor.recoverCompletedTask(task), recoverFailedPreMergeStep: (task) => this.executor.recoverFailedPreMergeWorkflowStep(task), getExecutingTaskIds: () => this.executor.getExecutingTaskIds(), diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 845080de9..d1a254a26 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -17,7 +17,7 @@ import { exec } from "node:child_process"; import { promisify } from "node:util"; import { existsSync, readdirSync, rmSync, statSync } from "node:fs"; import { isAbsolute, join, relative, resolve } from "node:path"; -import { getTaskMergeBlocker, type TaskStore, type Settings, type Task, type MergeDetails } from "@fusion/core"; +import { getTaskMergeBlocker, isEphemeralAgent, type AgentStore, type TaskStore, type Settings, type Task, type MergeDetails } from "@fusion/core"; import { createLogger } from "./logger.js"; import { getRegisteredWorktreePaths, scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js"; @@ -27,6 +27,8 @@ const execAsync = promisify(exec); export interface SelfHealingOptions { /** Project root directory (parent of .worktrees/) */ rootDir: string; + /** Optional AgentStore for agent-level self-healing checks. */ + agentStore?: AgentStore; /** * Callback to recover a completed task that is stuck in in-progress. * Called by the periodic maintenance cycle when it detects a task whose @@ -196,6 +198,7 @@ export class SelfHealingManager { { name: "orphaned-executions", fn: () => this.recoverOrphanedExecutions().then(() => undefined) }, { 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) }, ]; for (const step of steps) { @@ -650,6 +653,7 @@ export class SelfHealingManager { { name: "recover-approved-triage", fn: () => this.recoverApprovedTriageTasks() }, { name: "recover-orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks() }, { name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() }, + { name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() }, ]; for (const fn of batch2Fns) { try { @@ -1369,6 +1373,76 @@ export class SelfHealingManager { } } + async recoverOrphanedAgents(): Promise { + const agentStore = this.options.agentStore; + if (!agentStore) { + return 0; + } + + try { + const settings = await this.store.getSettings(); + const timeoutMs = settings.taskStuckTimeoutMs; + if (!Number.isFinite(timeoutMs) || timeoutMs === undefined || timeoutMs <= 0) { + return 0; + } + const recoveryTimeoutMs = timeoutMs; + + const allAgents = await agentStore.listAgents(); + const allAgentIds = new Set(allAgents.map((agent) => agent.id)); + const now = Date.now(); + + const orphaned = allAgents.filter((agent) => { + if (isEphemeralAgent(agent)) { + return false; + } + if (agent.state !== "running" && agent.state !== "error") { + return false; + } + const managerMissing = !agent.reportsTo || !allAgentIds.has(agent.reportsTo); + if (!managerMissing) { + return false; + } + const updatedAt = Date.parse(agent.updatedAt ?? ""); + if (!Number.isFinite(updatedAt)) { + return false; + } + return now - updatedAt >= recoveryTimeoutMs; + }); + + if (orphaned.length === 0) { + return 0; + } + + let recovered = 0; + for (const agent of orphaned) { + const updatedAt = Date.parse(agent.updatedAt ?? ""); + const stuckForMs = Math.max(0, now - updatedAt); + try { + await agentStore.updateAgentState(agent.id, "active"); + await agentStore.updateAgent(agent.id, { + lastError: undefined, + }); + log.log( + `Auto-recovered: orphaned agent ${agent.id} stuck in ${agent.state} for ${Math.round(stuckForMs / 1000)}s — reset to active`, + ); + recovered++; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Failed to recover orphaned agent ${agent.id}: ${errorMessage}`); + } + } + + if (recovered > 0) { + log.log(`Recovered ${recovered} orphaned agent(s) → active`); + } + return recovered; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Orphaned agent recovery failed: ${errorMessage}`); + return 0; + } + } + /** * 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.