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

feat(FN-2115): add heartbeat multiplier and agent interval controls

Browse Source 
- Add heartbeatIntervalMultiplier to shared settings schema/types with settings parity coverage
- Apply heartbeat multiplier in engine scheduling logic while preserving explicit per-agent interval behavior
- Add Settings modal and Agents view controls for heartbeat multiplier and per-agent interval overrides, including new styling
- Expand dashboard and engine test coverage for multiplier/heartbeat controls and document the new setting
This commit is contained in:
Fusion
2026-04-19 06:40:03 -07:00
committed by gsxdsm
parent a70ff39e31
commit a4ce10a1c5
12 changed files with 559 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -857,7 +857,7 @@ describe("HeartbeatMonitor", () => {
}
describe("getAgentHeartbeatConfig", () => {
it("returns monitor defaults when agentStore is not provided", () => {
it("returns monitor defaults when agentStore is not provided", async () => {
const monitor = new HeartbeatMonitor({
store,
pollIntervalMs: 5000,
@@ -865,13 +865,13 @@ describe("HeartbeatMonitor", () => {
maxConcurrentRuns: 2,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
expect(config.maxConcurrentRuns).toBe(2);
});
it("returns monitor defaults when agent has no runtimeConfig", () => {
it("returns monitor defaults when agent has no runtimeConfig", async () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
const monitor = new HeartbeatMonitor({
store,
@@ -880,12 +880,12 @@ describe("HeartbeatMonitor", () => {
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("returns per-agent values when runtimeConfig is set", () => {
it("returns per-agent values when runtimeConfig is set", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: {
@@ -902,13 +902,13 @@ describe("HeartbeatMonitor", () => {
maxConcurrentRuns: 1,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(2000);
expect(config.heartbeatTimeoutMs).toBe(30000);
expect(config.maxConcurrentRuns).toBe(3);
});
it("clamps heartbeatIntervalMs to minimum of 1000", () => {
it("clamps heartbeatIntervalMs to minimum of 1000", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 100 },
@@ -919,11 +919,11 @@ describe("HeartbeatMonitor", () => {
pollIntervalMs: 5000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(1000);
});
it("clamps heartbeatTimeoutMs to minimum of 5000", () => {
it("clamps heartbeatTimeoutMs to minimum of 5000", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 1000 },
@@ -934,11 +934,11 @@ describe("HeartbeatMonitor", () => {
heartbeatTimeoutMs: 60000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.heartbeatTimeoutMs).toBe(5000);
});
it("clamps maxConcurrentRuns to minimum of 1", () => {
it("clamps maxConcurrentRuns to minimum of 1", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { maxConcurrentRuns: 0 },
@@ -949,11 +949,11 @@ describe("HeartbeatMonitor", () => {
maxConcurrentRuns: 1,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.maxConcurrentRuns).toBe(1);
});
it("falls back to monitor defaults when runtimeConfig values are NaN", () => {
it("falls back to monitor defaults when runtimeConfig values are NaN", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: {
@@ -968,12 +968,12 @@ describe("HeartbeatMonitor", () => {
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("falls back to monitor defaults when agent is not found", () => {
it("falls back to monitor defaults when agent is not found", async () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
(agentStore.getCachedAgent as ReturnType<typeof vi.fn>).mockReturnValue(null);
@@ -984,12 +984,12 @@ describe("HeartbeatMonitor", () => {
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-999");
const config = await monitor.getAgentHeartbeatConfig("agent-999");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("returns monitor defaults when getCachedAgent throws", () => {
it("returns monitor defaults when getCachedAgent throws", async () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
(agentStore.getCachedAgent as ReturnType<typeof vi.fn>).mockImplementation(() => {
throw new Error("Read error");
@@ -1002,12 +1002,12 @@ describe("HeartbeatMonitor", () => {
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("returns partial overrides when only some runtimeConfig keys are set", () => {
it("returns partial overrides when only some runtimeConfig keys are set", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 120000 },
@@ -1020,11 +1020,45 @@ describe("HeartbeatMonitor", () => {
maxConcurrentRuns: 1,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000); // fallback
expect(config.heartbeatTimeoutMs).toBe(120000); // overridden
expect(config.maxConcurrentRuns).toBe(1); // fallback
});
it("applies project heartbeatMultiplier to pollIntervalMs", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 60_000 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 0.5 }),
} as unknown as TaskStore,
});
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(30_000);
});
it("clamps multiplied pollIntervalMs to minimum 1000ms", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 2000 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 0.1 }),
} as unknown as TaskStore,
});
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(1000);
});
});
describe("isAgentHealthy with per-agent config", () => {
@@ -1124,13 +1158,13 @@ describe("HeartbeatMonitor", () => {
});
describe("backward compatibility", () => {
it("works without agentStore (no per-agent config)", () => {
it("works without agentStore (no per-agent config)", async () => {
const monitor = new HeartbeatMonitor({
store,
heartbeatTimeoutMs: 5000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.heartbeatTimeoutMs).toBe(5000);
expect(config.pollIntervalMs).toBe(3_600_000); // default
expect(config.maxConcurrentRuns).toBe(1); // default
@@ -3915,6 +3949,72 @@ describe("HeartbeatTriggerScheduler", () => {
vi.useRealTimers();
});
it("applies heartbeatMultiplier to timer interval", async () => {
scheduler.stop();
const taskStore = {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 0.5 }),
} as unknown as TaskStore;
scheduler = new HeartbeatTriggerScheduler(store, callback, taskStore);
scheduler.start();
vi.useFakeTimers();
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 60_000, enabled: true });
await Promise.resolve();
await Promise.resolve();
expect(taskStore.getSettings).toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(29_999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledWith("agent-001", "timer", expect.objectContaining({ intervalMs: 30_000 }));
vi.useRealTimers();
});
it("defaults multiplier to 1 when setting is missing", async () => {
scheduler.stop();
const taskStore = {
getSettings: vi.fn().mockResolvedValue({}),
} as unknown as TaskStore;
scheduler = new HeartbeatTriggerScheduler(store, callback, taskStore);
scheduler.start();
vi.useFakeTimers();
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 20_000, enabled: true });
await Promise.resolve();
await Promise.resolve();
expect(taskStore.getSettings).toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(19_999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledWith("agent-001", "timer", expect.objectContaining({ intervalMs: 20_000 }));
vi.useRealTimers();
});
it("clamps multiplied interval to 1000ms minimum", async () => {
scheduler.stop();
const taskStore = {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 0.1 }),
} as unknown as TaskStore;
scheduler = new HeartbeatTriggerScheduler(store, callback, taskStore);
scheduler.start();
vi.useFakeTimers();
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 2_000, enabled: true });
await Promise.resolve();
await Promise.resolve();
expect(taskStore.getSettings).toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledWith("agent-001", "timer", expect.objectContaining({ intervalMs: 1_000 }));
vi.useRealTimers();
});
it("clears previous timer when re-registering", () => {
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 10000 });
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 20000 });

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

@@ -644,7 +644,7 @@ export class HeartbeatMonitor {
const tracked = this.trackedAgents.get(agentId);
if (!tracked) return false;
const config = this.getAgentConfig(agentId);
const config = this.resolveAgentConfig(agentId);
const elapsed = Date.now() - tracked.lastSeen;
return elapsed < config.heartbeatTimeoutMs;
}
@@ -1495,14 +1495,14 @@ export class HeartbeatMonitor {
* @param agentId - The agent ID
* @returns Resolved config with validated values
*/
getAgentHeartbeatConfig(agentId: string): ResolvedHeartbeatConfig {
async getAgentHeartbeatConfig(agentId: string): Promise<ResolvedHeartbeatConfig> {
return this.getAgentConfig(agentId);
}
/**
* Resolve per-agent heartbeat config from runtimeConfig with validation and fallbacks.
*/
private getAgentConfig(agentId: string): ResolvedHeartbeatConfig {
private resolveAgentConfig(agentId: string): ResolvedHeartbeatConfig {
// Defaults from monitor-level construction
const result: ResolvedHeartbeatConfig = {
pollIntervalMs: this.pollIntervalMs,
@@ -1532,11 +1532,34 @@ export class HeartbeatMonitor {
return result;
}
private async getAgentConfig(agentId: string): Promise<ResolvedHeartbeatConfig> {
const result = this.resolveAgentConfig(agentId);
if (!this.taskStore) {
return result;
}
try {
const settings = await getHeartbeatMemorySettings(this.taskStore);
const rawMultiplier = settings?.heartbeatMultiplier;
const multiplier =
typeof rawMultiplier === "number" && Number.isFinite(rawMultiplier) && rawMultiplier > 0
? rawMultiplier
: 1;
result.pollIntervalMs = Math.max(1000, Math.round(result.pollIntervalMs * multiplier));
} catch (settingsErr) {
heartbeatLog.warn(`getAgentConfig(${agentId}) settings lookup failed: ${settingsErr instanceof Error ? settingsErr.message : String(settingsErr)} — using base interval`);
}
return result;
}
private async checkMissedHeartbeats(): Promise<void> {
const now = Date.now();
for (const tracked of this.trackedAgents.values()) {
const config = this.getAgentConfig(tracked.agentId);
const config = await this.getAgentConfig(tracked.agentId);
const elapsed = now - tracked.lastSeen;
if (elapsed >= config.heartbeatTimeoutMs) {
@@ -1648,6 +1671,7 @@ export class HeartbeatTriggerScheduler {
private callback: TriggerCallback;
private taskStore?: TaskStore;
private timers: Map<string, AgentTimer> = new Map();
private registrationEpochs: Map<string, number> = new Map();
private running = false;
private assignedListener: ((agent: import("@fusion/core").Agent, taskId: string) => void) | null = null;
private updatedListener: ((agent: import("@fusion/core").Agent) => void) | null = null;
@@ -1725,31 +1749,98 @@ export class HeartbeatTriggerScheduler {
}
const intervalMs = Math.max(1000, Math.round(rawIntervalMs));
const registrationEpoch = (this.registrationEpochs.get(agentId) ?? 0) + 1;
this.registrationEpochs.set(agentId, registrationEpoch);
// Clear existing timer if re-registering
this.unregisterAgent(agentId);
// Register immediately with multiplier=1 so agents don't wait for async settings I/O.
this.applyTimerRegistration(agentId, intervalMs, 1, usingDefaultInterval);
// 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);
}
}
private async applyProjectMultiplierRegistration(
agentId: string,
baseIntervalMs: number,
usingDefaultInterval: boolean,
expectedEpoch: number,
): Promise<void> {
let multiplier = 1;
try {
const settings = await getHeartbeatMemorySettings(this.taskStore!);
multiplier = HeartbeatTriggerScheduler.resolveHeartbeatMultiplier(settings?.heartbeatMultiplier);
} catch (settingsErr) {
heartbeatLog.warn(
`Failed to read heartbeatMultiplier for ${agentId}: ${settingsErr instanceof Error ? settingsErr.message : String(settingsErr)} — using 1x`,
);
multiplier = 1;
}
// Guard against stale async completions after subsequent register/unregister calls.
if (this.registrationEpochs.get(agentId) !== expectedEpoch) {
return;
}
this.applyTimerRegistration(agentId, baseIntervalMs, multiplier, usingDefaultInterval);
}
private applyTimerRegistration(
agentId: string,
baseIntervalMs: number,
multiplier: number,
usingDefaultInterval: boolean,
): void {
const effectiveIntervalMs = Math.max(1000, Math.round(baseIntervalMs * multiplier));
this.clearAgentTimer(agentId);
const handle = setInterval(() => {
void this.onTimerTick(agentId, intervalMs);
}, intervalMs);
void this.onTimerTick(agentId, effectiveIntervalMs);
}, effectiveIntervalMs);
this.timers.set(agentId, { intervalMs: effectiveIntervalMs, handle });
if (multiplier !== 1) {
heartbeatLog.log(
`Registered timer for ${agentId} (every ${baseIntervalMs}ms, multiplier ${multiplier} → ${effectiveIntervalMs}ms effective)`,
);
return;
}
this.timers.set(agentId, { intervalMs, handle });
heartbeatLog.log(
usingDefaultInterval
? `Registered timer for ${agentId} (every ${intervalMs}ms, default interval)`
: `Registered timer for ${agentId} (every ${intervalMs}ms)`,
? `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms, default interval)`
: `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms)`,
);
}
private clearAgentTimer(agentId: string): void {
const timer = this.timers.get(agentId);
if (!timer) {
return;
}
clearInterval(timer.handle);
this.timers.delete(agentId);
}
private static resolveHeartbeatMultiplier(rawMultiplier: unknown): number {
if (typeof rawMultiplier !== "number" || !Number.isFinite(rawMultiplier) || rawMultiplier <= 0) {
return 1;
}
return rawMultiplier;
}
/**
* Unregister an agent, clearing its timer.
* @param agentId - The agent ID
*/
unregisterAgent(agentId: string): void {
const timer = this.timers.get(agentId);
if (timer) {
clearInterval(timer.handle);
this.timers.delete(agentId);
this.registrationEpochs.set(agentId, (this.registrationEpochs.get(agentId) ?? 0) + 1);
if (this.timers.has(agentId)) {
this.clearAgentTimer(agentId);
heartbeatLog.log(`Unregistered timer for ${agentId}`);
}
}