feat(FN-4033): handle stale timers in heartbeat scheduler
Adds stale timer repair handling to the heartbeat scheduler with test coverage, and updates the agents documentation accordingly. Fusion-Task-Id: FN-4033
This commit is contained in:
@@ -1025,6 +1025,8 @@ Effects:
|
|||||||
- Safety guards: skip ephemeral/task-worker agents, skip disabled agents, skip non-tickable states, and skip agents with an active heartbeat run
|
- 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)
|
- Existing timer entries are left untouched (no interval reset/jitter churn)
|
||||||
- Repair metadata: each audit re-arm writes `metadata.heartbeatTimerRepair` with `repairedAt` and a stale-at-repair indicator when the agent had already missed its expected cadence
|
- Repair metadata: each audit re-arm writes `metadata.heartbeatTimerRepair` with `repairedAt` and a stale-at-repair indicator when the agent had already missed its expected cadence
|
||||||
|
- Stale-at-repair threshold: defaults to `2 × heartbeatIntervalMs`; override with project setting `heartbeatRepairStaleMultiplier` (> 0) when you need a different sensitivity
|
||||||
|
- Stale repairs emit a WARN log entry and still flow through the existing `agent:updated` refresh path for dashboard surfacing
|
||||||
|
|
||||||
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.
|
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.
|
||||||
|
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ describe("HeartbeatTriggerScheduler", () => {
|
|||||||
let scheduler: import("../agent-heartbeat.js").HeartbeatTriggerScheduler;
|
let scheduler: import("../agent-heartbeat.js").HeartbeatTriggerScheduler;
|
||||||
|
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
|
vi.clearAllMocks();
|
||||||
callback = vi.fn().mockResolvedValue(undefined);
|
callback = vi.fn().mockResolvedValue(undefined);
|
||||||
store = {
|
store = {
|
||||||
getAgent: vi.fn().mockResolvedValue({
|
getAgent: vi.fn().mockResolvedValue({
|
||||||
@@ -108,7 +109,7 @@ describe("HeartbeatTriggerScheduler", () => {
|
|||||||
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
|
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
|
||||||
});
|
});
|
||||||
|
|
||||||
it("marks repaired agent metadata as stale when last heartbeat is old", async () => {
|
it("marks repaired agent metadata as stale when last heartbeat exceeds the default 2x threshold", async () => {
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
vi.setSystemTime(new Date("2026-01-01T02:00:00.000Z"));
|
vi.setSystemTime(new Date("2026-01-01T02:00:00.000Z"));
|
||||||
const agent = {
|
const agent = {
|
||||||
@@ -137,6 +138,75 @@ describe("HeartbeatTriggerScheduler", () => {
|
|||||||
}),
|
}),
|
||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
|
expect(heartbeatLog.warn).toHaveBeenCalledWith(expect.stringContaining("Timer re-armed stale agent agent-001"));
|
||||||
|
});
|
||||||
|
|
||||||
|
it("marks repaired agent metadata as healthy when heartbeat is within stale threshold", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
vi.setSystemTime(new Date("2026-01-01T00:00:10.000Z"));
|
||||||
|
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(store.updateAgent).toHaveBeenCalledWith(
|
||||||
|
"agent-001",
|
||||||
|
expect.objectContaining({
|
||||||
|
metadata: expect.objectContaining({
|
||||||
|
heartbeatTimerRepair: expect.objectContaining({ staleAtRepair: false }),
|
||||||
|
}),
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
expect(heartbeatLog.warn).not.toHaveBeenCalledWith(expect.stringContaining("Timer re-armed stale agent"));
|
||||||
|
});
|
||||||
|
|
||||||
|
it("uses project heartbeatRepairStaleMultiplier when configured", async () => {
|
||||||
|
vi.useFakeTimers();
|
||||||
|
vi.setSystemTime(new Date("2026-01-01T00:00:50.000Z"));
|
||||||
|
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);
|
||||||
|
|
||||||
|
const taskStore = {
|
||||||
|
getSettings: vi.fn().mockResolvedValue({ heartbeatRepairStaleMultiplier: 1 }),
|
||||||
|
} as unknown as TaskStore;
|
||||||
|
|
||||||
|
scheduler = new HeartbeatTriggerScheduler(store, callback, taskStore);
|
||||||
|
scheduler.start();
|
||||||
|
await vi.advanceTimersByTimeAsync(0);
|
||||||
|
|
||||||
|
expect(taskStore.getSettings).toHaveBeenCalled();
|
||||||
|
expect(store.updateAgent).toHaveBeenCalledWith(
|
||||||
|
"agent-001",
|
||||||
|
expect.objectContaining({
|
||||||
|
metadata: expect.objectContaining({
|
||||||
|
heartbeatTimerRepair: expect.objectContaining({ staleAtRepair: true }),
|
||||||
|
}),
|
||||||
|
}),
|
||||||
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("skips audit re-arm when the agent already has an active heartbeat run", async () => {
|
it("skips audit re-arm when the agent already has an active heartbeat run", async () => {
|
||||||
|
|||||||
@@ -2891,7 +2891,7 @@ export class HeartbeatTriggerScheduler {
|
|||||||
private timerAuditIntervalHandle: ReturnType<typeof setInterval> | null = null;
|
private timerAuditIntervalHandle: ReturnType<typeof setInterval> | null = null;
|
||||||
|
|
||||||
private static readonly TIMER_AUDIT_INTERVAL_MS = 60_000;
|
private static readonly TIMER_AUDIT_INTERVAL_MS = 60_000;
|
||||||
private static readonly REPAIR_STALE_GRACE_MULTIPLIER = 1.5;
|
private static readonly DEFAULT_REPAIR_STALE_MULTIPLIER = 2;
|
||||||
|
|
||||||
constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore, options?: { isTaskExecuting?: (taskId: string) => boolean }) {
|
constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore, options?: { isTaskExecuting?: (taskId: string) => boolean }) {
|
||||||
this.store = store;
|
this.store = store;
|
||||||
@@ -3385,14 +3385,22 @@ export class HeartbeatTriggerScheduler {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private getRepairStaleThresholdMs(agent: Agent): number {
|
private resolveRepairStaleMultiplier(settings: Settings | null | undefined): number {
|
||||||
|
const value = (settings as Record<string, unknown> | undefined)?.heartbeatRepairStaleMultiplier;
|
||||||
|
if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
|
||||||
|
return HeartbeatTriggerScheduler.DEFAULT_REPAIR_STALE_MULTIPLIER;
|
||||||
|
}
|
||||||
|
return value;
|
||||||
|
}
|
||||||
|
|
||||||
|
private getRepairStaleThresholdMs(agent: Agent, staleMultiplier: number): number {
|
||||||
const config = this.getAgentTimerConfig(agent);
|
const config = this.getAgentTimerConfig(agent);
|
||||||
let rawIntervalMs = config.heartbeatIntervalMs;
|
let rawIntervalMs = config.heartbeatIntervalMs;
|
||||||
if (!rawIntervalMs || typeof rawIntervalMs !== "number" || !Number.isFinite(rawIntervalMs) || rawIntervalMs <= 0) {
|
if (!rawIntervalMs || typeof rawIntervalMs !== "number" || !Number.isFinite(rawIntervalMs) || rawIntervalMs <= 0) {
|
||||||
rawIntervalMs = HeartbeatTriggerScheduler.DEFAULT_HEARTBEAT_INTERVAL_MS;
|
rawIntervalMs = HeartbeatTriggerScheduler.DEFAULT_HEARTBEAT_INTERVAL_MS;
|
||||||
}
|
}
|
||||||
const intervalMs = Math.max(1000, Math.round(rawIntervalMs));
|
const intervalMs = Math.max(1000, Math.round(rawIntervalMs));
|
||||||
return Math.round(intervalMs * HeartbeatTriggerScheduler.REPAIR_STALE_GRACE_MULTIPLIER);
|
return Math.round(intervalMs * staleMultiplier);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async markRepairMetadata(agent: Agent, staleAtRepair: boolean, staleRepairReason?: string): Promise<void> {
|
private async markRepairMetadata(agent: Agent, staleAtRepair: boolean, staleRepairReason?: string): Promise<void> {
|
||||||
@@ -3426,6 +3434,10 @@ export class HeartbeatTriggerScheduler {
|
|||||||
if (!this.running) return;
|
if (!this.running) return;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
const settings = this.taskStore && typeof this.taskStore.getSettings === "function"
|
||||||
|
? await this.taskStore.getSettings()
|
||||||
|
: null;
|
||||||
|
const staleMultiplier = this.resolveRepairStaleMultiplier(settings);
|
||||||
const agents = await this.store.listAgents();
|
const agents = await this.store.listAgents();
|
||||||
let rearmedCount = 0;
|
let rearmedCount = 0;
|
||||||
for (const agent of agents) {
|
for (const agent of agents) {
|
||||||
@@ -3442,7 +3454,7 @@ export class HeartbeatTriggerScheduler {
|
|||||||
lastHeartbeatAt: agent.lastHeartbeatAt,
|
lastHeartbeatAt: agent.lastHeartbeatAt,
|
||||||
});
|
});
|
||||||
|
|
||||||
const staleThresholdMs = this.getRepairStaleThresholdMs(agent);
|
const staleThresholdMs = this.getRepairStaleThresholdMs(agent, staleMultiplier);
|
||||||
const lastHeartbeatMs = agent.lastHeartbeatAt ? Date.parse(agent.lastHeartbeatAt) : Number.NaN;
|
const lastHeartbeatMs = agent.lastHeartbeatAt ? Date.parse(agent.lastHeartbeatAt) : Number.NaN;
|
||||||
const elapsedMs = Number.isFinite(lastHeartbeatMs) ? Date.now() - lastHeartbeatMs : Number.NaN;
|
const elapsedMs = Number.isFinite(lastHeartbeatMs) ? Date.now() - lastHeartbeatMs : Number.NaN;
|
||||||
const staleAtRepair = Number.isFinite(elapsedMs) && elapsedMs > staleThresholdMs;
|
const staleAtRepair = Number.isFinite(elapsedMs) && elapsedMs > staleThresholdMs;
|
||||||
@@ -3452,7 +3464,11 @@ export class HeartbeatTriggerScheduler {
|
|||||||
await this.markRepairMetadata(agent, staleAtRepair, staleRepairReason);
|
await this.markRepairMetadata(agent, staleAtRepair, staleRepairReason);
|
||||||
|
|
||||||
rearmedCount++;
|
rearmedCount++;
|
||||||
heartbeatLog.log(`Timer re-armed for ${agent.id} (audit:${reason}${staleAtRepair ? ", stale" : ""})`);
|
if (staleAtRepair) {
|
||||||
|
heartbeatLog.warn(`Timer re-armed stale agent ${agent.id} (audit:${reason}): ${staleRepairReason ?? "heartbeat exceeded stale threshold before repair"}`);
|
||||||
|
} else {
|
||||||
|
heartbeatLog.log(`Timer re-armed for ${agent.id} (audit:${reason})`);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (rearmedCount > 0) {
|
if (rearmedCount > 0) {
|
||||||
|
|||||||
Reference in New Issue
Block a user