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

feat(FN-5362): add heartbeat executor sparse-cache reports-health diagnosti

Browse Source 
The merge fixes a stale cache-miss path in the heartbeat executor and adds `reports-health` diagnostics for cache-state reporting, with a regression test covering a sparse-cache false-positive scenario. It also adds a two-line tweak to the merger and updates the diagnostics documentation.

Fusion-Task-Id: FN-5362
This commit is contained in:
Fusion (runfusion.ai)
2026-05-20 12:16:37 -07:00
committed by gsxdsm
parent 5548dc57de
commit 6fb4b9ee47
4 changed files with 130 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
packages/engine/src/__tests__/heartbeat-executor.test.ts
View File

@@ -261,6 +261,65 @@ describe("executeHeartbeat", () => {
expect(section).toContain("**stale**");
});
it("buildReportsHealthSection reproduces sparse-cache false positive without persisted interval (FN-5362)", async () => {
const now = Date.now();
const store = createStoreWithAgentForExec();
vi.mocked(store.getCachedAgent).mockReturnValue(null);
vi.mocked(store.getAgent).mockResolvedValue({
id: "agent-frontend",
} as unknown as Agent);
vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([
{
id: "agent-frontend",
name: "Frontend Engineer",
state: "idle",
taskId: null,
lastHeartbeatAt: new Date(now - 20 * 60_000).toISOString(),
updatedAt: new Date(now - 20 * 60_000).toISOString(),
} as unknown as Agent,
]);
const monitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
pollIntervalMs: 5 * 60_000,
});
const section = await (monitor as any).buildReportsHealthSection("agent-001", store);
expect(section).toContain("Frontend Engineer");
expect(section).toContain("**stale**");
});
it("buildReportsHealthSection uses persisted agent interval when cache is sparse (FN-5362)", async () => {
const now = Date.now();
const store = createStoreWithAgentForExec();
vi.mocked(store.getCachedAgent).mockReturnValue(null);
vi.mocked(store.getAgent).mockResolvedValue({
id: "agent-frontend",
runtimeConfig: { heartbeatIntervalMs: 60 * 60_000 },
} as unknown as Agent);
vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([
{
id: "agent-frontend",
name: "Frontend Engineer",
state: "idle",
taskId: null,
lastHeartbeatAt: new Date(now - 20 * 60_000).toISOString(),
updatedAt: new Date(now - 20 * 60_000).toISOString(),
} as unknown as Agent,
]);
const monitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
pollIntervalMs: 5 * 60_000,
});
const section = await (monitor as any).buildReportsHealthSection("agent-001", store);
expect(section).toContain("Frontend Engineer");
expect(section).not.toContain("**stale**");
});
it("buildReportsHealthSection keeps 60m-interval reports healthy within the grace window", async () => {
const now = Date.now();
const store = createStoreWithAgentForExec();
@@ -418,6 +477,35 @@ describe("executeHeartbeat", () => {
expect(section).not.toContain("**stale**");
});
it("buildReportsHealthSection logs stale decisions with interval source (FN-5362)", async () => {
const now = Date.now();
const store = createStoreWithAgentForExec();
vi.mocked(store.getCachedAgent).mockReturnValue(null);
vi.mocked(store.getAgent).mockResolvedValue({
id: "agent-overdue",
runtimeConfig: { heartbeatIntervalMs: 60 * 60_000 },
} as unknown as Agent);
vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([
{
id: "agent-overdue",
name: "Overdue",
state: "active",
taskId: null,
lastHeartbeatAt: new Date(now - 100 * 60_000).toISOString(),
updatedAt: new Date(now - 100 * 60_000).toISOString(),
} as unknown as Agent,
]);
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp", pollIntervalMs: 5 * 60_000 });
await (monitor as any).buildReportsHealthSection("agent-001", store);
expect(heartbeatLog.log).toHaveBeenCalledWith(
expect.stringContaining("[reports-health] stale report agent-overdue intervalSource=persisted-agent"),
);
expect(heartbeatLog.log).toHaveBeenCalledWith(expect.stringContaining("staleThresholdMs="));
expect(heartbeatLog.log).toHaveBeenCalledWith(expect.stringContaining("heartbeatAgeMs="));
});
it("buildReportsHealthSection preserves AgentStore method binding for direct-report lookups", async () => {
const now = new Date().toISOString();
const report = { id: "agent-004", name: "bound-report", state: "active", taskId: "FN-102", reportsTo: "agent-001", lastHeartbeatAt: now, updatedAt: now } as Agent;

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

@@ -2995,7 +2995,10 @@ export class HeartbeatMonitor {
}
private async buildReportsHealthSection(agentId: string, agentStore: AgentStore): Promise<string | null> {
const storeWithReports = agentStore as AgentStore & { getAgentsByReportsTo?: (id: string) => Promise<Agent[]> };
const storeWithReports = agentStore as AgentStore & {
getAgentsByReportsTo?: (id: string) => Promise<Agent[]>;
getAgent?: (id: string) => Promise<Agent | null>;
};
if (typeof storeWithReports.getAgentsByReportsTo !== "function") {
return null;
}
@@ -3012,8 +3015,28 @@ export class HeartbeatMonitor {
}
const now = Date.now();
const rows = reports.map((report) => {
const { pollIntervalMs, heartbeatTimeoutMs } = this.resolveAgentConfig(report.id);
const rows = await Promise.all(reports.map(async (report) => {
const resolvedConfig = this.resolveAgentConfig(report.id);
let pollIntervalMs = resolvedConfig.pollIntervalMs;
let intervalSource: "runtimeConfig" | "persisted-agent" | "monitor-default" = "monitor-default";
try {
const cachedAgent = this.configStore.getCachedAgent?.(report.id);
if (cachedAgent?.runtimeConfig && typeof cachedAgent.runtimeConfig.heartbeatIntervalMs === "number" && Number.isFinite(cachedAgent.runtimeConfig.heartbeatIntervalMs)) {
pollIntervalMs = Math.max(1000, cachedAgent.runtimeConfig.heartbeatIntervalMs);
intervalSource = "runtimeConfig";
} else if (typeof storeWithReports.getAgent === "function") {
const persisted = await storeWithReports.getAgent(report.id);
if (persisted?.runtimeConfig && typeof persisted.runtimeConfig.heartbeatIntervalMs === "number" && Number.isFinite(persisted.runtimeConfig.heartbeatIntervalMs)) {
pollIntervalMs = Math.max(1000, persisted.runtimeConfig.heartbeatIntervalMs);
intervalSource = "persisted-agent";
}
}
} catch (reportsHealthConfigErr) {
heartbeatLog.warn(`[reports-health] failed to resolve interval for ${report.id}: ${reportsHealthConfigErr instanceof Error ? reportsHealthConfigErr.message : String(reportsHealthConfigErr)} — using monitor-default`);
}
const { heartbeatTimeoutMs } = resolvedConfig;
const staleThresholdMs = Math.max(
pollIntervalMs * REPORTS_STALE_INTERVAL_MULTIPLIER,
MIN_HEARTBEAT_STALENESS_MS,
@@ -3030,13 +3053,14 @@ export class HeartbeatMonitor {
health = heartbeatAgeMs <= heartbeatTimeoutMs * 2 ? "healthy" : "**stuck**";
} else if ((report.state === "active" || report.state === "idle") && heartbeatAgeMs > staleThresholdMs) {
health = "**stale**";
heartbeatLog.log(`[reports-health] stale report ${report.id} intervalSource=${intervalSource} staleThresholdMs=${staleThresholdMs} heartbeatAgeMs=${heartbeatAgeMs}`);
}
const task = report.taskId ?? "—";
const state = report.state;
const heartbeat = formatRelativeTime(report.lastHeartbeatAt);
return `| ${report.name} | ${state} | ${task} | ${heartbeat} | ${health} |`;
});
}));
const hasStuck = rows.some((row) => row.includes("**stuck**"));
const hasStale = rows.some((row) => row.includes("**stale**"));

2
packages/engine/src/merger.ts
View File

@@ -6430,7 +6430,7 @@ export async function aiMergeTask(
audit,
runContext: engineRunContext,
runInitCommand: false,
createWorktree: async (branch, path, taskId, startPoint, allowSiblingBranchRename) => {
createWorktree: async (branch, path, _taskId, _startPoint, _allowSiblingBranchRename) => {
await execAsync(`git worktree add -f ${quoteArg(path)} ${quoteArg(branch)}`, {
cwd: projectRootDir,
encoding: "utf-8",