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

feat(FN-5414): merge fusion/fn-5414

Browse Source
This commit is contained in:
gsxdsm
2026-05-20 20:15:43 -07:00
parent 20a77ec29f
commit b936ab9bb5
5 changed files with 240 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
packages/engine/src/__tests__/heartbeat-monitor-per-agent-config.test.ts
View File

@@ -209,10 +209,10 @@ describe("per-agent heartbeat config", () => {
expect(config.maxConcurrentRuns).toBe(1); // fallback
});
it("applies project heartbeatMultiplier to pollIntervalMs", async () => {
it("applies project heartbeatMultiplier to pollIntervalMs and heartbeatTimeoutMs", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 60_000 },
runtimeConfig: { heartbeatIntervalMs: 60_000, heartbeatTimeoutMs: 30_000 },
});
const monitor = new HeartbeatMonitor({
store,
@@ -224,12 +224,31 @@ describe("per-agent heartbeat config", () => {
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(30_000);
expect(config.heartbeatTimeoutMs).toBe(15_000);
});
it("clamps multiplied pollIntervalMs to minimum 1000ms", async () => {
it("doubles pollIntervalMs and heartbeatTimeoutMs with multiplier 2", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 2000 },
runtimeConfig: { heartbeatIntervalMs: 20_000, heartbeatTimeoutMs: 10_000 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 2 }),
} as unknown as TaskStore,
});
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(40_000);
expect(config.heartbeatTimeoutMs).toBe(20_000);
});
it("clamps multiplied intervals to minimum floors", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 2000, heartbeatTimeoutMs: 10_000 },
});
const monitor = new HeartbeatMonitor({
store,
@@ -241,6 +260,28 @@ describe("per-agent heartbeat config", () => {
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(1000);
expect(config.heartbeatTimeoutMs).toBe(5000);
});
it("falls back to unscaled per-agent values when settings lookup fails", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 60_000, heartbeatTimeoutMs: 30_000 },
});
const warnSpy = vi.spyOn(heartbeatLog, "warn").mockImplementation(() => {});
const monitor = new HeartbeatMonitor({
store,
agentStore,
taskStore: {
getSettings: vi.fn().mockRejectedValue(new Error("boom")),
} as unknown as TaskStore,
});
const config = await monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(60_000);
expect(config.heartbeatTimeoutMs).toBe(30_000);
expect(warnSpy).toHaveBeenCalledTimes(1);
warnSpy.mockRestore();
});
});
@@ -270,6 +311,34 @@ describe("per-agent heartbeat config", () => {
vi.useRealTimers();
});
it("uses cached multiplier for sync health checks after async config resolve", async () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 10_000 },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 2 }),
} as unknown as TaskStore,
});
monitor.trackAgent("agent-001", session, "run-001");
await monitor.getAgentHeartbeatConfig("agent-001");
vi.advanceTimersByTime(15_000);
expect(monitor.isAgentHealthy("agent-001")).toBe(true);
vi.advanceTimersByTime(6_000);
expect(monitor.isAgentHealthy("agent-001")).toBe(false);
vi.useRealTimers();
});
});
describe("checkMissedHeartbeats with per-agent config", () => {
@@ -346,7 +415,130 @@ describe("per-agent heartbeat config", () => {
});
});
describe("reconcileOrphanedRunningAgents with multiplier cache", () => {
it("uses scaled timeout x3 after cache is warmed", async () => {
const nowIso = new Date().toISOString();
const staleIso = new Date(Date.now() - 50_000).toISOString();
const runningAgent = {
id: "agent-001",
name: "Agent",
role: "engineer",
state: "running",
createdAt: nowIso,
updatedAt: nowIso,
lastHeartbeatAt: staleIso,
runtimeConfig: { heartbeatTimeoutMs: 10_000 },
};
const monitorStore = {
listAgents: vi.fn().mockResolvedValue([runningAgent]),
getActiveHeartbeatRun: vi.fn().mockResolvedValue({ id: "run-001", status: "running" }),
getRunDetail: vi.fn().mockResolvedValue({ id: "run-001", status: "running" }),
saveRun: vi.fn().mockResolvedValue(undefined),
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
updateAgentState: vi.fn().mockResolvedValue(undefined),
} as unknown as AgentStore;
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 10_000 },
});
const monitor = new HeartbeatMonitor({
store: monitorStore,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 2 }),
} as unknown as TaskStore,
});
await monitor.getAgentHeartbeatConfig("agent-001");
await (monitor as any).reconcileOrphanedRunningAgents();
expect((monitorStore.endHeartbeatRun as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled();
expect((monitorStore.updateAgentState as ReturnType<typeof vi.fn>)).not.toHaveBeenCalled();
});
});
describe("backward compatibility", () => {
it("keeps missed and unresponsive thresholds unchanged when multiplier is 1", async () => {
const onMissed = vi.fn();
const runtimeStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { enabled: false },
});
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 5000 },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store: runtimeStore,
agentStore,
pollIntervalMs: 1000,
taskStore: {
getSettings: vi.fn().mockResolvedValue({ heartbeatMultiplier: 1 }),
} as unknown as TaskStore,
onMissed,
});
monitor.start();
monitor.trackAgent("agent-001", session, "run-001");
vi.advanceTimersByTime(4000);
await vi.advanceTimersByTimeAsync(100);
expect(onMissed).not.toHaveBeenCalled();
vi.advanceTimersByTime(1500);
await vi.advanceTimersByTimeAsync(100);
expect(onMissed).toHaveBeenCalledTimes(1);
vi.advanceTimersByTime(5000);
await vi.advanceTimersByTimeAsync(100);
expect(session.dispose).toHaveBeenCalledTimes(1);
monitor.stop();
vi.useRealTimers();
});
it("keeps orphan reconcile 3x timeout window unchanged when multiplier is unset", async () => {
const nowIso = new Date().toISOString();
const staleIso = new Date(Date.now() - 31_000).toISOString();
const runningAgent = {
id: "agent-001",
name: "Agent",
role: "engineer",
state: "running",
createdAt: nowIso,
updatedAt: nowIso,
lastHeartbeatAt: staleIso,
runtimeConfig: { heartbeatTimeoutMs: 10_000 },
};
const monitorStore = {
listAgents: vi.fn().mockResolvedValue([runningAgent]),
getActiveHeartbeatRun: vi.fn().mockResolvedValue({ id: "run-001", status: "running" }),
getRunDetail: vi.fn().mockResolvedValue({ id: "run-001", status: "running" }),
saveRun: vi.fn().mockResolvedValue(undefined),
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
updateAgentState: vi.fn().mockResolvedValue(undefined),
} as unknown as AgentStore;
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 10_000 },
});
const monitor = new HeartbeatMonitor({
store: monitorStore,
agentStore,
taskStore: {
getSettings: vi.fn().mockResolvedValue({}),
} as unknown as TaskStore,
});
await monitor.getAgentHeartbeatConfig("agent-001");
await (monitor as any).reconcileOrphanedRunningAgents();
expect((monitorStore.endHeartbeatRun as ReturnType<typeof vi.fn>)).toHaveBeenCalledTimes(1);
expect((monitorStore.updateAgentState as ReturnType<typeof vi.fn>)).toHaveBeenCalledWith("agent-001", "active");
});
it("works without agentStore (no per-agent config)", async () => {
const monitor = new HeartbeatMonitor({
store,

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

@@ -227,6 +227,13 @@ function getHeartbeatAgeMs(agent: Agent, now: number = Date.now()): number {
return Number.isFinite(lastTs) ? Math.max(0, now - lastTs) : Number.NaN;
}
function resolveHeartbeatMultiplier(rawMultiplier: unknown): number {
if (typeof rawMultiplier !== "number" || !Number.isFinite(rawMultiplier) || rawMultiplier <= 0) {
return 1;
}
return rawMultiplier;
}
async function terminatePersistedHeartbeatRun(
store: AgentStore,
agentId: string,
@@ -812,6 +819,8 @@ export class HeartbeatMonitor {
private agentStartLocks: Map<string, Promise<unknown>> = new Map();
private pollInterval: NodeJS.Timeout | null = null;
private isRunning = false;
private cachedHeartbeatMultiplier = 1;
private cachedHeartbeatMultiplierAt = 0;
/** Tasks created per agent during heartbeat runs (keyed by agentId) */
private runCreatedTasks: Map<string, Array<{ id: string; description: string }>> = new Map();
@@ -1098,6 +1107,8 @@ export class HeartbeatMonitor {
if (this.messageStore) {
this.messageStore.setMessageToAgentHook(this.handleMessageToAgent.bind(this));
}
// Warm heartbeat multiplier cache for sync health paths before reconcile.
void this.warmHeartbeatMultiplierCache();
// Reconcile any agents stuck in `state="running"` with no active run.
// Past versions of governance-skip paths (budget/global-pause) called
// completeRun with skipStateTransition=true after startRun had already
@@ -3235,9 +3246,26 @@ export class HeartbeatMonitor {
heartbeatLog.warn(`getAgentConfig(${agentId}) agent lookup failed: ${agentLookupErr instanceof Error ? agentLookupErr.message : String(agentLookupErr)} — using monitor defaults`);
}
// Sync health checks (isAgentHealthy / orphan reconcile / reports-health)
// are best-effort and use the most recent async-loaded multiplier cache.
const multiplier = this.cachedHeartbeatMultiplierAt > 0 ? this.cachedHeartbeatMultiplier : 1;
result.pollIntervalMs = Math.max(1000, Math.round(result.pollIntervalMs * multiplier));
result.heartbeatTimeoutMs = Math.max(5000, Math.round(result.heartbeatTimeoutMs * multiplier));
return result;
}
private async warmHeartbeatMultiplierCache(): Promise<void> {
if (!this.taskStore) return;
try {
const settings = await getHeartbeatMemorySettings(this.taskStore);
this.cachedHeartbeatMultiplier = resolveHeartbeatMultiplier(settings?.heartbeatMultiplier);
this.cachedHeartbeatMultiplierAt = Date.now();
} catch {
// Keep existing cache value on warm failures.
}
}
private async getAgentConfig(agentId: string): Promise<ResolvedHeartbeatConfig> {
const result = this.resolveAgentConfig(agentId);
@@ -3247,13 +3275,12 @@ export class HeartbeatMonitor {
try {
const settings = await getHeartbeatMemorySettings(this.taskStore);
const rawMultiplier = settings?.heartbeatMultiplier;
const multiplier =
typeof rawMultiplier === "number" && Number.isFinite(rawMultiplier) && rawMultiplier > 0
? rawMultiplier
: 1;
const multiplier = resolveHeartbeatMultiplier(settings?.heartbeatMultiplier);
this.cachedHeartbeatMultiplier = multiplier;
this.cachedHeartbeatMultiplierAt = Date.now();
result.pollIntervalMs = Math.max(1000, Math.round(result.pollIntervalMs * multiplier));
result.heartbeatTimeoutMs = Math.max(5000, Math.round(result.heartbeatTimeoutMs * multiplier));
} catch (settingsErr) {
heartbeatLog.warn(`getAgentConfig(${agentId}) settings lookup failed: ${settingsErr instanceof Error ? settingsErr.message : String(settingsErr)} — using base interval`);
}
@@ -3739,10 +3766,7 @@ export class HeartbeatTriggerScheduler {
}
private static resolveHeartbeatMultiplier(rawMultiplier: unknown): number {
if (typeof rawMultiplier !== "number" || !Number.isFinite(rawMultiplier) || rawMultiplier <= 0) {
return 1;
}
return rawMultiplier;
return resolveHeartbeatMultiplier(rawMultiplier);
}
/**