root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(heartbeat): per-agent HEARTBEAT.md and phase-aligned scheduler

Browse Source 
Each agent now gets its own .fusion/agents/<id>/HEARTBEAT.md procedure
file instead of sharing a single project-wide file. A one-shot
migration in AgentStore.init() re-points existing agents off the legacy
shared path and copies the legacy file's contents into each agent's
new per-agent location so operator edits are preserved.

The HeartbeatTriggerScheduler now phase-aligns the first tick to
lastHeartbeatAt + intervalMs so a process restart resumes each agent's
existing schedule rather than waiting up to a full interval before
firing again. Overdue ticks fire promptly within a small jitter window
to avoid a thundering herd at boot.

Also fixes three pre-existing QuickChatFAB test failures introduced by
739e899b5: auto-select default model now switches to model mode whether
or not agents are present, the model tag only renders in model mode,
and one test scopes its option lookup to role="option" to disambiguate
the in-header tag from the dropdown entry.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-29 23:08:21 -07:00
parent 739e899b59
commit a3617850f5
12 changed files with 358 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
packages/engine/src/__tests__/agent-heartbeat.test.ts
View File

@@ -4619,6 +4619,61 @@ describe("HeartbeatTriggerScheduler", () => {
expect(callback).not.toHaveBeenCalled();
});
it("phase-aligns the first tick to lastHeartbeatAt when supplied", async () => {
// Simulate: last tick was 4s ago, interval is 5s.
// The next tick is due in 1s, not in a fresh full 5s window.
vi.setSystemTime(new Date("2026-04-30T05:00:00.000Z"));
const lastHeartbeatAt = new Date("2026-04-30T04:59:56.000Z").toISOString();
scheduler.registerAgent(
"agent-001",
{ heartbeatIntervalMs: 5000 },
{ lastHeartbeatAt },
);
await vi.advanceTimersByTimeAsync(999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledOnce();
// Subsequent ticks resume the steady cadence.
await vi.advanceTimersByTimeAsync(5000);
expect(callback).toHaveBeenCalledTimes(2);
});
it("fires promptly with jitter when lastHeartbeatAt is already overdue", async () => {
// Interval is 60s but the last tick was 10 minutes ago — fire immediately
// (within the OVERDUE_FIRE_JITTER_MS window) instead of waiting another
// full 60s. This is the core fix for "agents look unresponsive after a
// dashboard restart" — the previous setInterval-only scheduler would
// have made the user wait a full interval before the catch-up tick.
vi.setSystemTime(new Date("2026-04-30T05:00:00.000Z"));
const lastHeartbeatAt = new Date("2026-04-30T04:50:00.000Z").toISOString();
scheduler.registerAgent(
"agent-001",
{ heartbeatIntervalMs: 60_000 },
{ lastHeartbeatAt },
);
// Jitter window is 5s; advance past it to guarantee the fire happens.
await vi.advanceTimersByTimeAsync(5_000);
expect(callback).toHaveBeenCalledOnce();
});
it("falls back to full-interval delay when lastHeartbeatAt is missing", async () => {
// No options provided — preserves the original "wait one full interval"
// behavior for agents that have never ticked.
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 5000 });
await vi.advanceTimersByTimeAsync(4999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledOnce();
});
it("skips tick when agent has active run", async () => {
(store.getActiveHeartbeatRun as ReturnType<typeof vi.fn>).mockResolvedValue({
id: "run-active",

148
packages/engine/src/agent-heartbeat.ts
View File

@@ -1927,12 +1927,34 @@ export type TriggerCallback = (
context: WakeContext,
) => Promise<void>;
/** Per-agent timer state */
/** Per-agent timer state. The active handle is either the initial
* phase-aligned `setTimeout` waiting for the first overdue tick, or the
* steady-state `setInterval` installed once that first tick fires.
*/
interface AgentTimer {
intervalMs: number;
kind: "timeout" | "interval";
handle: ReturnType<typeof setInterval>;
}
/** Optional context passed to registerAgent, used to phase-align the
* initial timer fire to the agent's persisted heartbeat history.
*/
export interface RegisterAgentOptions {
/** ISO timestamp of the agent's last heartbeat. When set, the initial
* fire is scheduled at `lastHeartbeatAt + intervalMs` rather than
* `now + intervalMs`, so a process restart does not cost agents up to
* one full interval of silence.
*/
lastHeartbeatAt?: string | null;
}
/** Maximum random jitter (ms) added to the initial fire when an agent's
* next tick is already overdue. Prevents a thundering herd when the
* scheduler boots and many agents want to fire immediately.
*/
const OVERDUE_FIRE_JITTER_MS = 5_000;
/**
* True when an agent's state indicates it should be ticking right now.
* Heartbeats track liveness while the agent is meant to be doing work.
@@ -2018,9 +2040,13 @@ export class HeartbeatTriggerScheduler {
this.unwatchAssignments();
this.unwatchAgentLifecycle();
// Clear all timers
// Clear all timers (mix of phase-alignment timeouts and steady intervals).
for (const [agentId, timer] of this.timers) {
clearInterval(timer.handle);
if (timer.kind === "timeout") {
clearTimeout(timer.handle as unknown as ReturnType<typeof setTimeout>);
} else {
clearInterval(timer.handle);
}
heartbeatLog.log(`Cleared timer for ${agentId}`);
}
this.timers.clear();
@@ -2040,10 +2066,19 @@ export class HeartbeatTriggerScheduler {
/**
* Register an agent for timer-based heartbeat triggers.
*
* The first fire is phase-aligned to `options.lastHeartbeatAt + intervalMs`
* when supplied. This means a process restart resumes each agent's
* existing schedule rather than waiting up to a full interval before the
* first tick — the previous behavior caused agents on long intervals
* (e.g. 1h) to appear "overdue" in the UI for nearly a full interval after
* every dashboard restart even though nothing was actually wrong with them.
*
* @param agentId - The agent ID
* @param config - Per-agent heartbeat config
* @param options - Optional registration context (e.g., lastHeartbeatAt)
*/
registerAgent(agentId: string, config: AgentHeartbeatConfig): void {
registerAgent(agentId: string, config: AgentHeartbeatConfig, options?: RegisterAgentOptions): void {
if (config.enabled === false) {
this.unregisterAgent(agentId);
return;
@@ -2063,12 +2098,14 @@ export class HeartbeatTriggerScheduler {
const registrationEpoch = (this.registrationEpochs.get(agentId) ?? 0) + 1;
this.registrationEpochs.set(agentId, registrationEpoch);
const lastHeartbeatAt = options?.lastHeartbeatAt ?? null;
// Register immediately with multiplier=1 so agents don't wait for async settings I/O.
this.applyTimerRegistration(agentId, intervalMs, 1, usingDefaultInterval);
this.applyTimerRegistration(agentId, intervalMs, 1, usingDefaultInterval, lastHeartbeatAt);
// If project settings are available, refresh registration with the current multiplier.
if (this.taskStore && typeof (this.taskStore as { getSettings?: () => Promise<Settings> }).getSettings === "function") {
void this.applyProjectMultiplierRegistration(agentId, intervalMs, usingDefaultInterval, registrationEpoch);
void this.applyProjectMultiplierRegistration(agentId, intervalMs, usingDefaultInterval, registrationEpoch, lastHeartbeatAt);
}
}
@@ -2077,6 +2114,7 @@ export class HeartbeatTriggerScheduler {
baseIntervalMs: number,
usingDefaultInterval: boolean,
expectedEpoch: number,
lastHeartbeatAt: string | null,
): Promise<void> {
let multiplier = 1;
@@ -2095,7 +2133,33 @@ export class HeartbeatTriggerScheduler {
return;
}
this.applyTimerRegistration(agentId, baseIntervalMs, multiplier, usingDefaultInterval);
this.applyTimerRegistration(agentId, baseIntervalMs, multiplier, usingDefaultInterval, lastHeartbeatAt);
}
/**
* Compute the delay until the agent's next scheduled fire, given when it
* last heartbeat. When `lastHeartbeatAt` is missing or unparseable, falls
* back to a full-interval delay (matching the original behavior for
* agents that have never ticked). When the next fire is already overdue,
* returns a small randomized jitter to spread thundering herds at boot.
*/
private static computeInitialDelayMs(
intervalMs: number,
lastHeartbeatAt: string | null,
now: number = Date.now(),
): number {
if (!lastHeartbeatAt) {
return intervalMs;
}
const lastMs = Date.parse(lastHeartbeatAt);
if (!Number.isFinite(lastMs)) {
return intervalMs;
}
const remaining = lastMs + intervalMs - now;
if (remaining <= 0) {
return Math.floor(Math.random() * OVERDUE_FIRE_JITTER_MS);
}
return Math.min(remaining, intervalMs);
}
private applyTimerRegistration(
@@ -2103,28 +2167,67 @@ export class HeartbeatTriggerScheduler {
baseIntervalMs: number,
multiplier: number,
usingDefaultInterval: boolean,
lastHeartbeatAt: string | null,
): void {
const effectiveIntervalMs = Math.max(1000, Math.round(baseIntervalMs * multiplier));
const initialDelayMs = HeartbeatTriggerScheduler.computeInitialDelayMs(
effectiveIntervalMs,
lastHeartbeatAt,
);
this.clearAgentTimer(agentId);
const handle = setInterval(() => {
void this.onTimerTick(agentId, effectiveIntervalMs);
}, effectiveIntervalMs);
const armSteadyInterval = () => {
// The setTimeout fired and was consumed; replace it with the long-lived
// setInterval that drives every subsequent tick. Use the same
// effectiveIntervalMs so the cadence remains correct.
const intervalHandle = setInterval(() => {
void this.onTimerTick(agentId, effectiveIntervalMs);
}, effectiveIntervalMs);
this.timers.set(agentId, {
intervalMs: effectiveIntervalMs,
kind: "interval",
handle: intervalHandle,
});
};
this.timers.set(agentId, { intervalMs: effectiveIntervalMs, handle });
if (initialDelayMs >= effectiveIntervalMs) {
// No phase-shift needed (agent has never ticked, or the saved
// lastHeartbeatAt is somehow in the future). Skip the timeout hop and
// arm the steady-state interval directly so the behavior matches the
// pre-phase-alignment scheduler.
armSteadyInterval();
} else {
const timeoutHandle = setTimeout(() => {
// Fire the overdue/phase-aligned tick first, then transition to the
// steady cadence. The tick fires regardless of whether the steady
// interval install succeeds, so a missed tick can never silently
// happen here.
void this.onTimerTick(agentId, effectiveIntervalMs);
armSteadyInterval();
}, initialDelayMs);
this.timers.set(agentId, {
intervalMs: effectiveIntervalMs,
kind: "timeout",
handle: timeoutHandle as unknown as ReturnType<typeof setInterval>,
});
}
const phaseSuffix = lastHeartbeatAt
? `, first fire in ${initialDelayMs}ms (phase-aligned to lastHeartbeatAt)`
: "";
if (multiplier !== 1) {
heartbeatLog.log(
`Registered timer for ${agentId} (every ${baseIntervalMs}ms, multiplier ${multiplier} → ${effectiveIntervalMs}ms effective)`,
`Registered timer for ${agentId} (every ${baseIntervalMs}ms, multiplier ${multiplier} → ${effectiveIntervalMs}ms effective${phaseSuffix})`,
);
return;
}
heartbeatLog.log(
usingDefaultInterval
? `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms, default interval)`
: `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms)`,
? `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms, default interval${phaseSuffix})`
: `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms${phaseSuffix})`,
);
}
@@ -2133,7 +2236,14 @@ export class HeartbeatTriggerScheduler {
if (!timer) {
return;
}
clearInterval(timer.handle);
// Both kinds share the same opaque handle type at runtime, but we route
// through the matching clear function for clarity and to satisfy strict
// type narrowing on platforms that distinguish the two.
if (timer.kind === "timeout") {
clearTimeout(timer.handle as unknown as ReturnType<typeof setTimeout>);
} else {
clearInterval(timer.handle);
}
this.timers.delete(agentId);
}
@@ -2291,7 +2401,9 @@ export class HeartbeatTriggerScheduler {
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
this.registerAgent(agent.id, this.getAgentTimerConfig(agent), {
lastHeartbeatAt: agent.lastHeartbeatAt,
});
heartbeatLog.log(`Timer armed for ${agent.id} (${reason})`);
}
@@ -2307,7 +2419,9 @@ export class HeartbeatTriggerScheduler {
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
this.registerAgent(agent.id, this.getAgentTimerConfig(agent), {
lastHeartbeatAt: agent.lastHeartbeatAt,
});
heartbeatLog.log(`Timer refreshed for ${agent.id} (${reason})`);
}

14
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -563,11 +563,15 @@ export class InProcessRuntime
for (const agent of agents) {
if (!isTimerManagedAgent(agent)) continue;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(agent.id, {
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
});
this.triggerScheduler.registerAgent(
agent.id,
{
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
},
{ lastHeartbeatAt: agent.lastHeartbeatAt },
);
registeredCount++;
}
if (agents.length > 0) {