feat(FN-3958): add heartbeat timer reconciliation self-healing

Adds scheduler heartbeat timer reconciliation with automatic self-healing when timers drift, including tests for tracked-only monitor recovery and documentation in the agents reference.

Fusion-Task-Id: FN-3958
This commit is contained in:
Fusion
2026-05-10 23:24:27 -07:00
committed by gsxdsm
parent baed1db0c7
commit be404e5967
7 changed files with 176 additions and 5 deletions

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Harden durable-agent heartbeat timer self-healing by adding scheduler-owned timer registration reconciliation and aligning dashboard dev-mode startup timer eligibility with runtime behavior.

View File

@@ -970,6 +970,21 @@ Effects:
- Resume triggers one on-demand heartbeat restart only when `runtimeConfig.enabled !== false`
- `onTerminated` is a run-level callback for terminated heartbeat runs and is not used by unresponsive recovery
### Timer Reconciliation Self-Healing (FN-3958)
`HeartbeatTriggerScheduler` owns a periodic registration audit that reconciles durable-agent truth in `AgentStore` against the in-memory timer map.
- Audit cadence: once immediately on scheduler start, then every 60 seconds while running
- Repair target: durable, heartbeat-enabled agents in tickable states (`active`, `running`, `idle`) that are missing a timer entry
- Safety guards: skip ephemeral/task-worker agents, skip disabled agents, skip non-tickable states, and skip agents with an active heartbeat run
- Existing timer entries are left untouched (no interval reset/jitter churn)
This covers the untracked timer-loss failure mode where no `agent:updated` event fires after a timer entry disappears. Manual stop/start is no longer required to re-arm the timer in that case.
Separation of responsibilities:
- **HeartbeatMonitor recovery** handles **tracked stale sessions** (stuck in-memory run/session cleanup + pause/resume restart)
- **HeartbeatTriggerScheduler audit** handles **untracked missing-timer registration drift** (re-arm scheduling)
## Dashboard Health Status
The dashboard displays agent health status in AgentsView, AgentListModal, and AgentDetailView using a centralized health evaluation utility (`packages/dashboard/app/utils/agentHealth.ts`).

View File

@@ -1779,17 +1779,19 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
const agents = await agentStore.listAgents();
const missedCatchupTargets: { agentId: string; lastHeartbeatAt: string }[] = [];
for (const agent of agents) {
// State is the source of truth: arm timers only for non-ephemeral
// agents that are currently active/running. Transitions into
// State is the source of truth: arm timers only for non-ephemeral,
// heartbeat-enabled agents in tickable states. Transitions into
// tickable states while the scheduler is already running are
// handled by the scheduler's own agent:updated listener.
// handled by the scheduler's own lifecycle listeners.
if (isEphemeralAgent(agent)) continue;
if (agent.state !== "active" && agent.state !== "running") continue;
if (agent.runtimeConfig?.enabled === false) continue;
if (agent.state !== "active" && agent.state !== "running" && agent.state !== "idle") continue;
const rc = agent.runtimeConfig;
const intervalMs = (rc?.heartbeatIntervalMs as number | undefined) ?? DEFAULT_AGENT_HEARTBEAT_INTERVAL_MS;
triggerScheduler.registerAgent(
agent.id,
{
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
},

View File

@@ -719,7 +719,7 @@ describe("missed heartbeat detection", () => {
});
describe("unresponsive agent recovery", () => {
it("disposes session and pauses/resumes agent after 2x timeout", async () => {
it("recovers only tracked stale sessions via pause/resume restart", async () => {
const onTerminated = vi.fn();
const session = createMockSession();
const localStore = createMockStore({
@@ -751,6 +751,34 @@ describe("unresponsive agent recovery", () => {
vi.useRealTimers();
});
it("does not attempt stale recovery for untracked agents", async () => {
const localStore = createMockStore({
getAgent: vi.fn().mockResolvedValue({
id: "agent-001",
state: "active",
lastHeartbeatAt: new Date(Date.now() - 60_000).toISOString(),
runtimeConfig: { enabled: true },
}),
updateAgentState: vi.fn().mockResolvedValue(undefined),
updateAgent: vi.fn().mockResolvedValue(undefined),
});
const customMonitor = new HeartbeatMonitor({
store: localStore,
heartbeatTimeoutMs: 5_000,
pollIntervalMs: 1_000,
});
vi.useFakeTimers({ shouldAdvanceTime: true });
customMonitor.start();
await vi.advanceTimersByTimeAsync(12_000);
expect(localStore.updateAgentState).not.toHaveBeenCalledWith("agent-001", "paused");
expect(localStore.updateAgentState).not.toHaveBeenCalledWith("agent-001", "active");
customMonitor.stop();
vi.useRealTimers();
});
it("logs recovery warnings when dispose and pause fail", async () => {
const warnSpy = vi.mocked(heartbeatLog.warn);
warnSpy.mockClear();

View File

@@ -35,6 +35,7 @@ describe("HeartbeatTriggerScheduler", () => {
}),
getActiveHeartbeatRun: vi.fn().mockResolvedValue(null),
getBudgetStatus: vi.fn().mockResolvedValue(createBudgetStatus()),
listAgents: vi.fn().mockResolvedValue([]),
on: vi.fn(),
off: vi.fn(),
} as unknown as AgentStore;
@@ -73,6 +74,61 @@ describe("HeartbeatTriggerScheduler", () => {
});
});
describe("scheduler timer audit", () => {
it("re-arms a tickable durable agent when timer entry is missing and no lifecycle event fires", async () => {
vi.useFakeTimers();
const agent = {
id: "agent-001",
name: "Agent 001",
role: "executor",
state: "active",
lastHeartbeatAt: "2026-01-01T00:00:00.000Z",
runtimeConfig: { enabled: true, heartbeatIntervalMs: 30_000 },
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
metadata: {},
} as Agent;
vi.mocked(store.listAgents).mockResolvedValue([agent]);
vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null);
scheduler = new HeartbeatTriggerScheduler(store, callback);
scheduler.start();
await vi.advanceTimersByTimeAsync(0);
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
scheduler.unregisterAgent("agent-001");
expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
await vi.advanceTimersByTimeAsync(60_000);
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
});
it("skips audit re-arm when the agent already has an active heartbeat run", async () => {
vi.useFakeTimers();
const agent = {
id: "agent-001",
name: "Agent 001",
role: "executor",
state: "active",
runtimeConfig: { enabled: true, heartbeatIntervalMs: 30_000 },
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
metadata: {},
} as Agent;
vi.mocked(store.listAgents).mockResolvedValue([agent]);
vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue({ id: "run-1" } as any);
scheduler = new HeartbeatTriggerScheduler(store, callback);
scheduler.start();
await vi.advanceTimersByTimeAsync(0);
scheduler.unregisterAgent("agent-001");
await vi.advanceTimersByTimeAsync(60_000);
expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
});
});
describe("registerAgent", () => {
beforeEach(() => {
scheduler = new HeartbeatTriggerScheduler(store, callback);

View File

@@ -2856,6 +2856,9 @@ export class HeartbeatTriggerScheduler {
private configRevisionListener: ((agentId: string, revision: AgentConfigRevision) => void) | null = null;
private deletedListener: ((agentId: string) => void) | null = null;
private isTaskExecuting?: (taskId: string) => boolean;
private timerAuditIntervalHandle: ReturnType<typeof setInterval> | null = null;
private static readonly TIMER_AUDIT_INTERVAL_MS = 60_000;
constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore, options?: { isTaskExecuting?: (taskId: string) => boolean }) {
this.store = store;
@@ -2873,6 +2876,10 @@ export class HeartbeatTriggerScheduler {
this.running = true;
this.watchAssignments();
this.watchAgentLifecycle();
void this.auditTimerRegistrations("start");
this.timerAuditIntervalHandle = setInterval(() => {
void this.auditTimerRegistrations("interval");
}, HeartbeatTriggerScheduler.TIMER_AUDIT_INTERVAL_MS);
heartbeatLog.log("HeartbeatTriggerScheduler started");
}
@@ -2898,6 +2905,11 @@ export class HeartbeatTriggerScheduler {
}
this.timers.clear();
if (this.timerAuditIntervalHandle) {
clearInterval(this.timerAuditIntervalHandle);
this.timerAuditIntervalHandle = null;
}
heartbeatLog.log("HeartbeatTriggerScheduler stopped");
}
@@ -3340,6 +3352,37 @@ export class HeartbeatTriggerScheduler {
}
}
async auditTimerRegistrations(reason: "start" | "interval" = "interval"): Promise<void> {
if (!this.running) return;
try {
const agents = await this.store.listAgents();
let rearmedCount = 0;
for (const agent of agents) {
if (!this.isTimerEligibleAgent(agent)) continue;
if (this.timers.has(agent.id)) continue;
const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
if (activeRun) {
heartbeatLog.log(`Timer audit skipped re-arm for ${agent.id} (active run)`);
continue;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent), {
lastHeartbeatAt: agent.lastHeartbeatAt,
});
rearmedCount++;
heartbeatLog.log(`Timer re-armed for ${agent.id} (audit:${reason})`);
}
if (rearmedCount > 0) {
heartbeatLog.log(`Timer audit repaired ${rearmedCount} missing registration(s) (${reason})`);
}
} catch (error) {
heartbeatLog.warn(`Timer audit failed (${reason}): ${error instanceof Error ? error.message : String(error)}`);
}
}
/**
* Handle a timer tick for an agent.
* Checks for active runs before invoking the callback.

View File

@@ -1181,6 +1181,28 @@ describe("InProcessRuntime", () => {
);
});
it("reconciles a missing timer for a tickable durable agent without state changes", async () => {
const store = getAgentStore(runtime);
const scheduler = runtime.getTriggerScheduler();
expect(scheduler).toBeDefined();
const agent = await store.createAgent({
name: "audit-rearm-agent",
role: "executor",
runtimeConfig: {
enabled: true,
heartbeatIntervalMs: 1_000,
},
});
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
scheduler!.unregisterAgent(agent.id);
expect(scheduler!.getRegisteredAgents()).not.toContain(agent.id);
await vi.advanceTimersByTimeAsync(60_000);
expect(scheduler!.getRegisteredAgents()).toContain(agent.id);
});
it("unregisters an agent when enabled is set to false in update", async () => {
// Create a new agent with heartbeat enabled
const store = getAgentStore(runtime);