From e35620c9aa1166ce50b7e525ca4c4a2c963caca1 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 13 Jul 2026 07:51:07 -0700 Subject: [PATCH] FN-7939: supervise heartbeat timer-audit interval and bound non-advancing zombie re-arms Fixes agents silently going stale for hours even though the heartbeat repair audit process was running. - HeartbeatTriggerScheduler now runs an independent watchdog (armTimerAuditWatchdog/checkTimerAuditLiveness) that tracks the audit loop's last-run timestamp and re-arms + immediately re-runs the 60s audit interval if it goes stale beyond a bounded multiple of the cadence, so a silently dropped audit driver self-heals instead of leaving active agents unrepaired for hours. - Tracks consecutive non-advancing zombie-timer re-arms per agent (nonAdvancingRearmState) and escalates once the count crosses a threshold, recording consecutiveNonAdvancingRearms/nonAdvancingEscalated in agent.metadata.heartbeatTimerRepair and logging reason=heartbeat-rearm-nonadvancing-escalated instead of silently churning the same zombie-timer-rearmed repair forever. - Clears non-advancing rearm state on unregister, non-eligible agents, paused settings, and stale-run-reap skip paths so tracking never leaks stale per-agent counters. - Watchdog and its interval handle are armed in start() and cleared in stop() alongside the existing audit interval. - Adds a changeset (patch) describing the fix, and updates docs/agents.md and docs/architecture.md to document the FN-7939 audit watchdog and non-advancing escalation behavior. - Adds heartbeat-scheduler.test.ts coverage for watchdog re-arm/liveness and non-advancing escalation. Files changed: .changeset/fn-7939-heartbeat-audit-supervision.md | 7 + docs/agents.md | 8 +- docs/architecture.md | 1 + .../src/__tests__/heartbeat-scheduler.test.ts | 209 +++++++++++++++++++++ packages/engine/src/agent-heartbeat.ts | 128 ++++++++++++- 5 files changed, 341 insertions(+), 12 deletions(-) Fusion-Task-Id: FN-7939 Fusion-Task-Lineage: 9fa90240-4333-4588-b595-aef3811b1524 Co-authored-by: Fusion (runfusion.ai) --- .../fn-7939-heartbeat-audit-supervision.md | 7 + docs/agents.md | 8 +- docs/architecture.md | 1 + .../src/__tests__/heartbeat-scheduler.test.ts | 209 ++++++++++++++++++ packages/engine/src/agent-heartbeat.ts | 128 ++++++++++- 5 files changed, 341 insertions(+), 12 deletions(-) create mode 100644 .changeset/fn-7939-heartbeat-audit-supervision.md diff --git a/.changeset/fn-7939-heartbeat-audit-supervision.md b/.changeset/fn-7939-heartbeat-audit-supervision.md new file mode 100644 index 0000000000..59009e9be4 --- /dev/null +++ b/.changeset/fn-7939-heartbeat-audit-supervision.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix agents silently going stale for hours even though the heartbeat repair audit was running. +category: fix +dev: HeartbeatTriggerScheduler now supervises its own audit setInterval (a stalled/dropped audit driver is re-armed within a bounded window) and bounds/escalates non-advancing zombie-timer re-arms instead of churning silently, closing the ~62,348s silent-heartbeat window that survived the FN-7645/FN-7718 fixes (FN-7939). diff --git a/docs/agents.md b/docs/agents.md index 90270382fa..b651a767fb 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -1275,17 +1275,19 @@ Heartbeat runs from the Agents panel run on a **separate control-plane lane** th Cadence: - **Immediate startup audit**: runs once in `start()` after lifecycle watchers are attached - **Periodic audit**: runs every 60s while the scheduler is active -- **Cleanup**: periodic sweep interval is cleared in `stop()` +- **FN-7939 audit watchdog**: an independent watchdog supervises the 60s audit driver and re-arms/runs it when the audit liveness timestamp is stale for a bounded multiple of the cadence +- **Cleanup**: periodic sweep and watchdog intervals are cleared in `stop()` Repair eligibility: - Durable (non-ephemeral/task-worker) agent - `runtimeConfig.enabled !== false` - Agent state is tickable: `active`, `running`, or `idle` -- Agent is missing from the scheduler's in-memory `timers` map +- Agent is missing from the scheduler's in-memory `timers` map, or has a present timer entry but `lastHeartbeatAt` is stale beyond the repair threshold (zombie timer) Repair outcomes: - **Missing timer, not stale**: timer is re-armed and INFO diagnostics are logged (`agentId`, resolved interval, elapsed time since `lastHeartbeatAt`) -- **Missing timer, stale**: timer is re-armed, WARN diagnostics are logged, and `agent.metadata.heartbeatTimerRepair` is updated (`repairedAt`, `staleAtRepair`, `elapsedMs`, `staleThresholdMs`) +- **Missing/present timer, stale**: timer is re-armed, WARN diagnostics are logged, and `agent.metadata.heartbeatTimerRepair` is updated (`repairedAt`, `staleAtRepair`, `staleRepairReason`) +- **Repeated non-advancing zombie repair**: after the bounded scheduler threshold, metadata includes the consecutive count/escalation flag and logs `reason=heartbeat-rearm-nonadvancing-escalated` instead of silently churning forever Stale threshold: - Repair staleness defaults to **`2 × heartbeatIntervalMs`** diff --git a/docs/architecture.md b/docs/architecture.md index c48739ef12..f33387245e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1359,6 +1359,7 @@ Limits are controlled by project settings (`maxSpawnedAgentsPerParent`, `maxSpaw - on-demand runs - Assignment triggers skipped because a heartbeat run is already active are deferred and re-fired from `HeartbeatMonitor.onRunCompleted`, preserving the existing completion recovery path while avoiding timer-dependent stalls. - FN-7645: `HeartbeatTriggerScheduler.auditTimerRegistrations` (60s cadence) repairs both MISSING timer registrations and "zombie" ones — a timer map entry that stays present after its underlying `setInterval` silently stops firing. When a tickable, heartbeat-managed agent's `lastHeartbeatAt` exceeds the repair-stale threshold (`heartbeatIntervalMs * heartbeatRepairStaleMultiplier`, default 2x) even though a timer entry already exists, the audit clears and re-registers it (phase-aligned via `computeInitialDelayMs`), logging `reason=zombie-timer-rearmed`. This closes the gap where long-interval (~1h) agents could silently drift stale for hours because their sparse cadence meant a single lost tick was never re-armed by the previous missing-registration-only repair; short-interval agents were unaffected because their frequent ticks self-heal within minutes. All existing guards are preserved: pause suppression (`globalPause`/`enginePaused`) still gates dispatch in `onTimerTick`, FN-4119 stale active-run reaping still runs first for agents with a live run, ephemeral/task-worker agents stay excluded via `isTimerEligibleAgent`, and `registrationEpochs` staleness protection is untouched. +- FN-7939: the FN-7645/FN-7718 repair audit is itself self-supervised. `HeartbeatTriggerScheduler` records audit-loop liveness and runs an independent watchdog that re-arms the 60s audit interval plus immediately executes one audit when the driver is stale for a bounded multiple of the audit cadence, so a dropped auditor cannot strand active agents for hours while timer entries remain present. Repeated `zombie-timer-rearmed` repairs that do not advance `lastHeartbeatAt` are also bounded: after the non-advancing count crosses the scheduler threshold, the repair metadata records the count/escalation and logs the greppable `reason=heartbeat-rearm-nonadvancing-escalated` signal instead of silently churning forever; intentional `globalPause`/`enginePaused` suppression and healthy active runs remain non-escalating. - FN-7718: CLI-driven `fn agent stop`/`start` mutate the agent row from a SEPARATE process, so the in-process `agent:updated` listener never fires for those transitions — the 60s audit is the ONLY cross-process reconciliation path. The audit now invalidates a stopped/non-eligible agent's lingering timer entry (state made non-tickable, `runtimeConfig.enabled === false`, or ephemeral/`!isHeartbeatManaged`) instead of bare-`continue`ing past it, so the entry never survives to become an orphaned/"zombie" registration. `syncTimerForAgent` mirrors this for the in-process start seam: an eligible agent whose present timer entry is already stale beyond the same repair threshold is force-cleared and re-armed rather than left in place by the "already ticking" no-op. Net effect: a `stop`/`start` cycle durably clears the zombie-timer condition in one audit cycle instead of deferring repair to the FN-7645 stale-repair path minutes later. - FN-7723 (follow-up from FN-7718): `AgentStore` (`packages/core/src/agent-store.ts`) now supports an opt-in cross-process change-detection fast-path over the FN-7645/FN-7718 audit backstop — `startWatching()`/`stopWatching()`/`checkForChanges()`, modeled directly on `TaskStore`'s `fs.watch`+poll pattern (`packages/core/src/store.ts`): an `fs.watch` on the project's `.fusion` dir as a fail-soft fast-path nudge, plus an always-on poll fallback (default 2s) gated by `db.getLastModified()` so an unchanged DB costs one cheap `__meta` read. On a detected change it diffs current agent rows against a last-seen per-instance snapshot (comparing `state` explicitly, not just `updatedAt`, since two rapid writes can land in the same ISO-millisecond and mask a genuine transition) and re-emits the EXISTING `agent:updated`/`agent:stateChanged` events — no new event names, so `HeartbeatTriggerScheduler.watchAgentLifecycle`'s current listener reacts unchanged, funneling through the same `syncTimerForAgent` seam (including FN-7718's stale-present-entry force-re-arm). Only the long-lived engine `AgentStore` instance opts in (started/stopped alongside `HeartbeatTriggerScheduler` in `packages/engine/src/runtimes/in-process-runtime.ts`); the CLI's short-lived `AgentStore` (`packages/cli/src/commands/agent.ts`) and per-request dashboard stores never call `startWatching()`. The 60s `auditTimerRegistrations` sweep is UNCHANGED and remains the durable backstop — this is a purely additive latency improvement, not a replacement: a `fn agent stop`/`start` is now typically observed within one poll interval (~2s) instead of up to 60s. - FN-7726 (follow-up from FN-7723): the mechanical fs.watch+poll lifecycle `TaskStore.watch()` and `AgentStore.startWatching()` each hand-rolled (fail-soft `fs.watch` setup with its two canonical warn strings, the `setInterval` poll fallback, and idempotent teardown) is now a single shared `FsWatchPollController` (`packages/core/src/fs-watch-poll-controller.ts`). The controller owns ONLY that mechanism — `start({dir, recursive?, pollIntervalMs, onPoll, log, errorContext?})`/`stop()`/`isWatching()`/`watcher` (a live-handle getter kept for test seams). It does NOT own diff/emit logic or gating: `getLastModified()`-vs-`lastKnownModified` gating and the `pollingInProgress` re-entrancy guard remain private fields on each store, evaluated inside each store's own `checkForChanges()` (the function passed to the controller as `onPoll`) exactly as before extraction — a deliberate scoping decision (see FN-7726's task `plan` document) to avoid coupling the shared controller to each store's very different diff bodies (TaskStore's delete/archive/artifact-cursor diff vs. AgentStore's snapshot state compare). `TaskStore` and `AgentStore` each hold a private `watchPoll: FsWatchPollController` instance and pass their own logger (`storeLog`/`agentStoreLog`) so the `[task-store]`-prefixed and agent-prefixed fail-soft warn strings are unchanged. Public method names/signatures (`watch()`/`stopWatching()` on `TaskStore`; `startWatching()`/`stopWatching()`/`isWatching()`/`checkForChanges()` on `AgentStore`) and all existing events/latencies/gating are unchanged — this is a behavior-preserving internal refactor, not a new mechanism. diff --git a/packages/engine/src/__tests__/heartbeat-scheduler.test.ts b/packages/engine/src/__tests__/heartbeat-scheduler.test.ts index 5024b84417..df280f9860 100644 --- a/packages/engine/src/__tests__/heartbeat-scheduler.test.ts +++ b/packages/engine/src/__tests__/heartbeat-scheduler.test.ts @@ -836,6 +836,215 @@ describe("HeartbeatTriggerScheduler", () => { }); }); + describe("FN-7939: heartbeat audit driver supervision and bounded re-arm churn", () => { + /** + * FNXC:AgentHeartbeat 2026-07-13-00:00: + * FN-7939 proves the FN-7645/FN-7718 repair layer is not allowed to depend on a single unsupervised audit setInterval. The reported CEO outage kept a timer entry present for ~62,348s, so tests kill the audit driver itself and require the scheduler to self-rearm within a bounded watchdog window instead of waiting for an external stop/start. + */ + function buildAgent(overrides: Partial & { id: string; heartbeatIntervalMs: number }): Agent { + const { heartbeatIntervalMs, ...rest } = overrides; + return { + name: rest.id, + role: "executor", + state: "active", + lastHeartbeatAt: "2026-01-01T00:00:00.000Z", + runtimeConfig: { enabled: true, heartbeatIntervalMs }, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + metadata: {}, + ...rest, + } as Agent; + } + + it("self-rearms a stalled audit interval and dispatches a stale present short-interval timer within the bounded watchdog window", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agents: Record = { + "agent-short-stalled-audit": buildAgent({ id: "agent-short-stalled-audit", heartbeatIntervalMs: 300_000 }), + }; + vi.mocked(store.listAgents).mockImplementation(async () => Object.values(agents)); + vi.mocked(store.getAgent).mockImplementation(async (agentId: string) => agents[agentId] ?? null); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + vi.mocked(store.updateAgent).mockImplementation(async (agentId: string, updates: Partial) => { + agents[agentId] = { ...agents[agentId], ...updates } as Agent; + return agents[agentId]; + }); + callback.mockImplementation(async (agentId: string) => { + agents[agentId] = { ...agents[agentId], lastHeartbeatAt: new Date().toISOString() }; + }); + + scheduler = new HeartbeatTriggerScheduler(store, callback); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + + const internals = scheduler as unknown as { + timerAuditIntervalHandle: ReturnType | null; + timers: Map; + }; + expect(internals.timerAuditIntervalHandle).not.toBeNull(); + clearInterval(internals.timerAuditIntervalHandle!); + clearInterval(internals.timers.get("agent-short-stalled-audit")!.handle as ReturnType); + callback.mockClear(); + vi.mocked(heartbeatLog.warn).mockClear(); + + await vi.advanceTimersByTimeAsync(62_348_000); + + expect(callback).toHaveBeenCalledWith("agent-short-stalled-audit", "timer", expect.anything()); + expect(agents["agent-short-stalled-audit"].lastHeartbeatAt).not.toBe("2026-01-01T00:00:00.000Z"); + expect(heartbeatLog.warn).toHaveBeenCalledWith(expect.stringContaining("reason=heartbeat-audit-watchdog-rearmed")); + expect(heartbeatLog.warn).toHaveBeenCalledWith(expect.stringContaining("zombie-timer-rearmed")); + expect(heartbeatLog.warn).not.toHaveBeenCalledWith(expect.stringContaining("reason=heartbeat-rearm-nonadvancing-escalated")); + }); + + it("self-rearms a stalled audit interval for the long 3_600_000ms interval bucket without double-auditing while healthy", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agents: Record = { + "agent-long-stalled-audit": buildAgent({ id: "agent-long-stalled-audit", heartbeatIntervalMs: 3_600_000 }), + }; + vi.mocked(store.listAgents).mockImplementation(async () => Object.values(agents)); + vi.mocked(store.getAgent).mockImplementation(async (agentId: string) => agents[agentId] ?? null); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + callback.mockImplementation(async (agentId: string) => { + agents[agentId] = { ...agents[agentId], lastHeartbeatAt: new Date().toISOString() }; + }); + + scheduler = new HeartbeatTriggerScheduler(store, callback); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + + const internals = scheduler as unknown as { + timerAuditIntervalHandle: ReturnType | null; + timers: Map; + }; + clearInterval(internals.timerAuditIntervalHandle!); + clearInterval(internals.timers.get("agent-long-stalled-audit")!.handle as ReturnType); + callback.mockClear(); + + await vi.advanceTimersByTimeAsync(62_348_000); + + expect(callback).toHaveBeenCalledWith("agent-long-stalled-audit", "timer", expect.anything()); + const longTicks = callback.mock.calls.filter((call) => call[0] === "agent-long-stalled-audit").length; + expect(longTicks).toBeGreaterThanOrEqual(1); + expect(longTicks).toBeLessThanOrEqual(18); + }); + + it("does not resurrect a stopped scheduler after the audit watchdog cadence passes", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agent = buildAgent({ id: "agent-stopped-scheduler", heartbeatIntervalMs: 300_000 }); + vi.mocked(store.listAgents).mockResolvedValue([agent]); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + + scheduler = new HeartbeatTriggerScheduler(store, callback); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + scheduler.stop(); + const internals = scheduler as unknown as { + timerAuditIntervalHandle: ReturnType | null; + timerAuditWatchdogHandle: ReturnType | null; + }; + expect(internals.timerAuditIntervalHandle).toBeNull(); + expect(internals.timerAuditWatchdogHandle).toBeNull(); + + await vi.advanceTimersByTimeAsync(62_348_000); + + expect(scheduler.isActive()).toBe(false); + expect(callback).not.toHaveBeenCalled(); + }); + + it("does not watchdog-rearm or double-audit while the normal audit interval keeps firing", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agent = buildAgent({ id: "agent-healthy-auditor", heartbeatIntervalMs: 300_000 }); + vi.mocked(store.listAgents).mockResolvedValue([agent]); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + + scheduler = new HeartbeatTriggerScheduler(store, callback); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + vi.mocked(heartbeatLog.warn).mockClear(); + + await vi.advanceTimersByTimeAsync(10 * 60_000); + + expect(heartbeatLog.warn).not.toHaveBeenCalledWith(expect.stringContaining("reason=heartbeat-audit-watchdog-rearmed")); + }); + + it("does not escalate repeated zombie re-arms while globalPause intentionally suppresses timer dispatch", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agents: Record = { + "agent-global-pause": buildAgent({ id: "agent-global-pause", heartbeatIntervalMs: 120_000 }), + }; + vi.mocked(store.listAgents).mockImplementation(async () => Object.values(agents)); + vi.mocked(store.getAgent).mockImplementation(async (agentId: string) => agents[agentId] ?? null); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + const taskStore = { + getSettings: vi.fn().mockResolvedValue({ globalPause: true, enginePaused: false }), + } as unknown as TaskStore; + + scheduler = new HeartbeatTriggerScheduler(store, callback, taskStore); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + + const timers = (scheduler as unknown as { timers: Map }).timers; + clearInterval(timers.get("agent-global-pause")!.handle as ReturnType); + callback.mockClear(); + vi.mocked(heartbeatLog.warn).mockClear(); + + await vi.advanceTimersByTimeAsync(12 * 60_000); + + expect(callback).not.toHaveBeenCalled(); + expect(heartbeatLog.warn).toHaveBeenCalledWith(expect.stringContaining("zombie-timer-rearmed")); + expect(heartbeatLog.warn).not.toHaveBeenCalledWith(expect.stringContaining("reason=heartbeat-rearm-nonadvancing-escalated")); + }); + + it("escalates repeated zombie re-arms when skipHeartbeatWhenIdle prevents lastHeartbeatAt from advancing", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z")); + + const agents: Record = { + "agent-churn": buildAgent({ + id: "agent-churn", + heartbeatIntervalMs: 120_000, + runtimeConfig: { enabled: true, heartbeatIntervalMs: 120_000, skipHeartbeatWhenIdle: true }, + }), + }; + vi.mocked(store.listAgents).mockImplementation(async () => Object.values(agents)); + vi.mocked(store.getAgent).mockImplementation(async (agentId: string) => agents[agentId] ?? null); + vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null); + vi.mocked(store.updateAgent).mockImplementation(async (agentId: string, updates: Partial) => { + agents[agentId] = { ...agents[agentId], ...updates } as Agent; + return agents[agentId]; + }); + + scheduler = new HeartbeatTriggerScheduler(store, callback); + scheduler.start(); + await vi.advanceTimersByTimeAsync(0); + + const timers = (scheduler as unknown as { timers: Map }).timers; + clearInterval(timers.get("agent-churn")!.handle as ReturnType); + callback.mockClear(); + vi.mocked(heartbeatLog.warn).mockClear(); + + await vi.advanceTimersByTimeAsync(12 * 60_000); + + expect(callback).not.toHaveBeenCalled(); + expect(heartbeatLog.warn).toHaveBeenCalledWith(expect.stringContaining("reason=heartbeat-rearm-nonadvancing-escalated agentId=agent-churn")); + expect((agents["agent-churn"].metadata as Record).heartbeatTimerRepair).toEqual( + expect.objectContaining({ + staleAtRepair: true, + staleRepairReason: expect.stringContaining("heartbeat-rearm-nonadvancing-escalated"), + }), + ); + }); + }); + describe("FN-7718: orphaned/zombie timer invalidation on stop/start", () => { /** * FNXC:AgentHeartbeat 2026-07-09-00:00: diff --git a/packages/engine/src/agent-heartbeat.ts b/packages/engine/src/agent-heartbeat.ts index b4c37ac945..3e34a5dff5 100644 --- a/packages/engine/src/agent-heartbeat.ts +++ b/packages/engine/src/agent-heartbeat.ts @@ -4260,6 +4260,8 @@ type HeartbeatTimerRepairMetadata = { repairedAt?: string; staleAtRepair?: boolean; staleRepairReason?: string; + consecutiveNonAdvancingRearms?: number; + nonAdvancingEscalated?: boolean; }; function readHeartbeatTimerRepairMetadata(agent: Agent): HeartbeatTimerRepairMetadata { @@ -4273,6 +4275,8 @@ function readHeartbeatTimerRepairMetadata(agent: Agent): HeartbeatTimerRepairMet repairedAt: typeof candidate.repairedAt === "string" ? candidate.repairedAt : undefined, staleAtRepair: typeof candidate.staleAtRepair === "boolean" ? candidate.staleAtRepair : undefined, staleRepairReason: typeof candidate.staleRepairReason === "string" ? candidate.staleRepairReason : undefined, + consecutiveNonAdvancingRearms: typeof candidate.consecutiveNonAdvancingRearms === "number" ? candidate.consecutiveNonAdvancingRearms : undefined, + nonAdvancingEscalated: typeof candidate.nonAdvancingEscalated === "boolean" ? candidate.nonAdvancingEscalated : undefined, }; } @@ -4307,8 +4311,14 @@ export class HeartbeatTriggerScheduler { * Absent (legacy/no executor wiring) → treated as never effectively executing. */ private isAgentEffectivelyExecuting?: (agentId: string) => boolean; private timerAuditIntervalHandle: ReturnType | null = null; + private timerAuditWatchdogHandle: ReturnType | null = null; + private lastAuditRanAtMs = 0; + private nonAdvancingRearmState: Map = new Map(); private static readonly TIMER_AUDIT_INTERVAL_MS = 60_000; + private static readonly TIMER_AUDIT_WATCHDOG_INTERVAL_MS = 60_000; + private static readonly TIMER_AUDIT_WATCHDOG_STALE_MS = HeartbeatTriggerScheduler.TIMER_AUDIT_INTERVAL_MS * 3; + private static readonly NON_ADVANCING_REARM_ESCALATION_THRESHOLD = 3; private static readonly DEFAULT_REPAIR_STALE_MULTIPLIER = 2; private static readonly DEFAULT_HEARTBEAT_TIMEOUT_MS = 60_000; @@ -4327,13 +4337,59 @@ export class HeartbeatTriggerScheduler { start(): void { if (this.running) return; this.running = true; + this.lastAuditRanAtMs = Date.now(); this.watchAssignments(); this.watchAgentLifecycle(); void this.auditTimerRegistrations("start"); + this.armTimerAuditInterval(); + this.armTimerAuditWatchdog(); + heartbeatLog.log("HeartbeatTriggerScheduler started"); + } + + private armTimerAuditInterval(): void { + if (this.timerAuditIntervalHandle) { + clearInterval(this.timerAuditIntervalHandle); + this.timerAuditIntervalHandle = null; + } + if (!this.running) { + return; + } this.timerAuditIntervalHandle = setInterval(() => { void this.auditTimerRegistrations("interval"); }, HeartbeatTriggerScheduler.TIMER_AUDIT_INTERVAL_MS); - heartbeatLog.log("HeartbeatTriggerScheduler started"); + } + + private armTimerAuditWatchdog(): void { + if (this.timerAuditWatchdogHandle) { + clearInterval(this.timerAuditWatchdogHandle); + this.timerAuditWatchdogHandle = null; + } + if (!this.running) { + return; + } + /* + * FNXC:AgentHeartbeat 2026-07-13-07:38: + * FN-7939 — FN-7645's per-agent zombie-timer repair and FN-7718's stop/start invalidation depend on the 60s audit setInterval, but that auditor is itself a live timer that can silently stop firing. Supervise the auditor with an independent liveness timer so a stalled audit driver is re-armed inside a bounded window instead of leaving active agents unrepaired for hours (observed: 62,348s with a timer entry still present). + */ + this.timerAuditWatchdogHandle = setInterval(() => { + void this.checkTimerAuditLiveness(); + }, HeartbeatTriggerScheduler.TIMER_AUDIT_WATCHDOG_INTERVAL_MS); + } + + private async checkTimerAuditLiveness(): Promise { + if (!this.running) { + return; + } + const now = Date.now(); + const elapsedMs = this.lastAuditRanAtMs > 0 ? now - this.lastAuditRanAtMs : Number.POSITIVE_INFINITY; + if (elapsedMs <= HeartbeatTriggerScheduler.TIMER_AUDIT_WATCHDOG_STALE_MS) { + return; + } + heartbeatLog.warn( + `Heartbeat timer audit watchdog re-armed stalled auditor reason=heartbeat-audit-watchdog-rearmed elapsedMs=${elapsedMs} thresholdMs=${HeartbeatTriggerScheduler.TIMER_AUDIT_WATCHDOG_STALE_MS}`, + ); + this.armTimerAuditInterval(); + await this.auditTimerRegistrations("interval"); } /** @@ -4362,6 +4418,12 @@ export class HeartbeatTriggerScheduler { clearInterval(this.timerAuditIntervalHandle); this.timerAuditIntervalHandle = null; } + if (this.timerAuditWatchdogHandle) { + clearInterval(this.timerAuditWatchdogHandle); + this.timerAuditWatchdogHandle = null; + } + this.lastAuditRanAtMs = 0; + this.nonAdvancingRearmState.clear(); heartbeatLog.log("HeartbeatTriggerScheduler stopped"); } @@ -4570,6 +4632,7 @@ export class HeartbeatTriggerScheduler { unregisterAgent(agentId: string): void { this.registrationEpochs.set(agentId, (this.registrationEpochs.get(agentId) ?? 0) + 1); this.pendingAssignments.delete(agentId); + this.nonAdvancingRearmState.delete(agentId); if (this.timers.has(agentId)) { this.clearAgentTimer(agentId); heartbeatLog.log(`Unregistered timer for ${agentId}`); @@ -4988,7 +5051,12 @@ export class HeartbeatTriggerScheduler { return { reaped: true, elapsedMs, thresholdMs }; } - private async markRepairMetadata(agent: Agent, staleAtRepair: boolean, staleRepairReason?: string): Promise { + private async markRepairMetadata( + agent: Agent, + staleAtRepair: boolean, + staleRepairReason?: string, + options?: { consecutiveNonAdvancingRearms?: number; nonAdvancingEscalated?: boolean }, + ): Promise { const updater = (this.store as { updateAgent?: (agentId: string, updates: { metadata: Record }) => Promise }).updateAgent; if (typeof updater !== "function") { return; @@ -5000,12 +5068,16 @@ export class HeartbeatTriggerScheduler { repairedAt, staleAtRepair, ...(staleAtRepair && staleRepairReason ? { staleRepairReason } : {}), + ...(typeof options?.consecutiveNonAdvancingRearms === "number" ? { consecutiveNonAdvancingRearms: options.consecutiveNonAdvancingRearms } : {}), + ...(options?.nonAdvancingEscalated ? { nonAdvancingEscalated: true } : {}), }; const didChange = existing.repairedAt !== nextRepair.repairedAt || existing.staleAtRepair !== nextRepair.staleAtRepair || - existing.staleRepairReason !== nextRepair.staleRepairReason; + existing.staleRepairReason !== nextRepair.staleRepairReason || + existing.consecutiveNonAdvancingRearms !== nextRepair.consecutiveNonAdvancingRearms || + existing.nonAdvancingEscalated !== nextRepair.nonAdvancingEscalated; if (!didChange) { return; } @@ -5017,6 +5089,7 @@ export class HeartbeatTriggerScheduler { async auditTimerRegistrations(reason: "start" | "interval" = "interval"): Promise { if (!this.running) return; + this.lastAuditRanAtMs = Date.now(); try { const settings = this.taskStore && typeof this.taskStore.getSettings === "function" @@ -5046,6 +5119,7 @@ export class HeartbeatTriggerScheduler { * instead of inheriting a stale/orphaned timer. */ if (!this.isTimerEligibleAgent(agent)) { + this.nonAdvancingRearmState.delete(agent.id); if (this.timers.has(agent.id)) { this.unregisterAgent(agent.id); heartbeatLog.log(`Timer audit cleared orphaned timer for non-eligible agent ${agent.id} (audit:${reason})`); @@ -5072,7 +5146,10 @@ export class HeartbeatTriggerScheduler { * fresh (non-stale) present timer is left alone so healthy short-interval agents are never * force-re-armed or double-ticked. */ - if (hasTimerEntry && !staleAtRepair) continue; + if (hasTimerEntry && !staleAtRepair) { + this.nonAdvancingRearmState.delete(agent.id); + continue; + } const isZombieRearm = hasTimerEntry && staleAtRepair; @@ -5083,6 +5160,7 @@ export class HeartbeatTriggerScheduler { let activeRunThresholdMs = Number.NaN; if (activeRun) { if (settings?.globalPause || settings?.enginePaused) { + this.nonAdvancingRearmState.delete(agent.id); heartbeatLog.log(`Timer audit skipped re-arm for ${agent.id} (active run)`); continue; } @@ -5091,11 +5169,30 @@ export class HeartbeatTriggerScheduler { activeRunElapsedMs = reapResult.elapsedMs; activeRunThresholdMs = reapResult.thresholdMs; if (!reapedActiveRun) { + this.nonAdvancingRearmState.delete(agent.id); heartbeatLog.log(`Timer audit skipped re-arm for ${agent.id} (active run)`); continue; } } + let consecutiveNonAdvancingRearms: number | undefined; + let nonAdvancingEscalated = false; + let escalationReason: string | undefined; + if (isZombieRearm && !settings?.globalPause && !settings?.enginePaused) { + const heartbeatMarker = typeof agent.lastHeartbeatAt === "string" ? agent.lastHeartbeatAt : null; + const previous = this.nonAdvancingRearmState.get(agent.id); + consecutiveNonAdvancingRearms = previous && previous.lastHeartbeatAt === heartbeatMarker + ? previous.count + 1 + : 1; + this.nonAdvancingRearmState.set(agent.id, { lastHeartbeatAt: heartbeatMarker, count: consecutiveNonAdvancingRearms }); + nonAdvancingEscalated = consecutiveNonAdvancingRearms >= HeartbeatTriggerScheduler.NON_ADVANCING_REARM_ESCALATION_THRESHOLD; + if (nonAdvancingEscalated) { + escalationReason = `heartbeat-rearm-nonadvancing-escalated: ${consecutiveNonAdvancingRearms} consecutive zombie re-arms without lastHeartbeatAt advancing`; + } + } else { + this.nonAdvancingRearmState.delete(agent.id); + } + // registerAgent() clears any existing (including zombie) timer entry via // clearAgentTimer() before re-arming, so a present-but-dead interval handle // never leaks and the new registration phase-aligns via computeInitialDelayMs. @@ -5104,11 +5201,16 @@ export class HeartbeatTriggerScheduler { }); const staleRepairReason = staleAtRepair - ? isZombieRearm - ? `zombie-timer-rearmed: no heartbeat for ${Math.round(elapsedMs / 1000)}s while a timer entry remained present (threshold ${Math.round(staleThresholdMs / 1000)}s)` - : `No heartbeat for ${Math.round(elapsedMs / 1000)}s before timer audit repair (threshold ${Math.round(staleThresholdMs / 1000)}s)` + ? escalationReason + ? `${escalationReason}; zombie-timer-rearmed: no heartbeat for ${Math.round(elapsedMs / 1000)}s while a timer entry remained present (threshold ${Math.round(staleThresholdMs / 1000)}s)` + : isZombieRearm + ? `zombie-timer-rearmed: no heartbeat for ${Math.round(elapsedMs / 1000)}s while a timer entry remained present (threshold ${Math.round(staleThresholdMs / 1000)}s)` + : `No heartbeat for ${Math.round(elapsedMs / 1000)}s before timer audit repair (threshold ${Math.round(staleThresholdMs / 1000)}s)` : undefined; - await this.markRepairMetadata(agent, staleAtRepair, staleRepairReason); + await this.markRepairMetadata(agent, staleAtRepair, staleRepairReason, { + consecutiveNonAdvancingRearms, + nonAdvancingEscalated, + }); rearmedCount++; if (isZombieRearm) zombieRearmedCount++; @@ -5117,7 +5219,15 @@ export class HeartbeatTriggerScheduler { `Timer audit re-armed after stale-run reap reason=timer-audit-rearmed agentId=${agent.id} runId=${activeRunId} elapsedMs=${activeRunElapsedMs} thresholdMs=${activeRunThresholdMs}`, ); } - if (isZombieRearm) { + if (nonAdvancingEscalated) { + /* + * FNXC:AgentHeartbeat 2026-07-13-07:39: + * FN-7939 — a zombie-timer re-arm that never restores delivery must become visible after a bounded count. Persistent skip/parallel guards can otherwise leave lastHeartbeatAt frozen while the audit rewrites `zombie-timer-rearmed` metadata forever, recreating multi-hour silent drift under a nominally active timer entry. + */ + heartbeatLog.warn( + `Timer audit escalated non-advancing zombie re-arm reason=heartbeat-rearm-nonadvancing-escalated agentId=${agent.id} count=${consecutiveNonAdvancingRearms} threshold=${HeartbeatTriggerScheduler.NON_ADVANCING_REARM_ESCALATION_THRESHOLD} (audit:${reason}): ${staleRepairReason}`, + ); + } else if (isZombieRearm) { heartbeatLog.warn(`Timer audit force re-armed non-advancing agent ${agent.id} reason=zombie-timer-rearmed (audit:${reason}): ${staleRepairReason}`); } else if (staleAtRepair) { heartbeatLog.warn(`Timer re-armed stale agent ${agent.id} (audit:${reason}): ${staleRepairReason ?? "heartbeat exceeded stale threshold before repair"}`);