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

feat(FN-3319): enrich agent heartbeat callbacks with reason in UI

Browse Source 
Enriched agent heartbeat callbacks with a `reason` parameter propagated through the engine, in-process runtime, and dashboard components (AgentDetailView, AgentListModal, AgentsView). Added a new `agentHealth.tsx` utility and corresponding test file to surface the reason in UI health status, with te

Fusion-Task-Id: FN-3319
This commit is contained in:
Fusion
2026-05-03 12:47:08 -07:00
committed by gsxdsm
parent e153ae5443
commit a1b74de12b
9 changed files with 125 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -657,7 +657,7 @@ describe("HeartbeatMonitor", () => {
// Wait for async checkMissedHeartbeats
await vi.advanceTimersByTimeAsync(100);
expect(onMissed).toHaveBeenCalledWith("agent-001");
expect(onMissed).toHaveBeenCalledWith("agent-001", expect.any(String));
customMonitor.stop();
vi.useRealTimers();
@@ -711,7 +711,7 @@ describe("HeartbeatMonitor", () => {
expect(session.dispose).toHaveBeenCalled();
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "terminated");
expect(onTerminated).toHaveBeenCalledWith("agent-001");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
customMonitor.stop();
vi.useRealTimers();
@@ -769,7 +769,7 @@ describe("HeartbeatMonitor", () => {
const warnMessages = warnSpy.mock.calls.map(([message]) => String(message));
expect(warnMessages.some((message) => message.includes("Error disposing session for agent-001") && message.includes("dispose exploded"))).toBe(true);
expect(updateAgentState).toHaveBeenCalledWith("agent-001", "terminated");
expect(onTerminated).toHaveBeenCalledWith("agent-001");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
customMonitor.stop();
@@ -800,7 +800,7 @@ describe("HeartbeatMonitor", () => {
const warnMessages = warnSpy.mock.calls.map(([message]) => String(message));
expect(warnMessages.some((message) => message.includes("Error terminating agent agent-001") && message.includes("db connection lost"))).toBe(true);
expect(onTerminated).toHaveBeenCalledWith("agent-001");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
customMonitor.stop();
@@ -834,10 +834,11 @@ describe("HeartbeatMonitor", () => {
await vi.advanceTimersByTimeAsync(100);
const warnMessages = warnSpy.mock.calls.map(([message]) => String(message));
expect(warnMessages).toHaveLength(2);
expect(warnMessages).toHaveLength(3);
expect(warnMessages.some((message) => message.includes("Terminating unresponsive agent agent-001"))).toBe(true);
expect(warnMessages.some((message) => message.includes("Error disposing session for agent-001") && message.includes("dispose exploded"))).toBe(true);
expect(warnMessages.some((message) => message.includes("Error terminating agent agent-001") && message.includes("db connection lost"))).toBe(true);
expect(onTerminated).toHaveBeenCalledWith("agent-001");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
customMonitor.stop();
@@ -1137,7 +1138,7 @@ describe("HeartbeatMonitor", () => {
vi.advanceTimersByTime(5000);
await vi.advanceTimersByTimeAsync(100);
expect(onMissed).toHaveBeenCalledWith("agent-001");
expect(onMissed).toHaveBeenCalledWith("agent-001", expect.any(String));
monitor.stop();
vi.useRealTimers();
@@ -1167,7 +1168,7 @@ describe("HeartbeatMonitor", () => {
await vi.advanceTimersByTimeAsync(100);
expect(session.dispose).toHaveBeenCalled();
expect(onTerminated).toHaveBeenCalledWith("agent-001");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
monitor.stop();
vi.useRealTimers();

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

@@ -52,11 +52,11 @@ export interface HeartbeatMonitorOptions {
/** Max concurrent runs per agent (default: 1) */
maxConcurrentRuns?: number;
/** Callback when an agent misses its heartbeat */
onMissed?: (agentId: string) => void;
onMissed?: (agentId: string, reason: string) => void;
/** Callback when an agent recovers after a missed heartbeat */
onRecovered?: (agentId: string) => void;
/** Callback when an unresponsive agent is terminated */
onTerminated?: (agentId: string) => void;
onTerminated?: (agentId: string, reason: string) => void;
/** Callback when a run starts */
onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
/** Callback when a run completes */
@@ -116,6 +116,17 @@ interface TrackedAgent {
sessionIdBefore?: string;
}
/** Format milliseconds into a human-readable duration string (e.g. "5m", "1h 20m", "2h"). */
export function formatDuration(ms: number): string {
const totalMinutes = Math.floor(ms / 60_000);
if (totalMinutes < 1) return "<1m";
const hours = Math.floor(totalMinutes / 60);
const minutes = totalMinutes % 60;
if (hours > 0 && minutes > 0) return `${hours}h ${minutes}m`;
if (hours > 0) return `${hours}h`;
return `${minutes}m`;
}
/** Compare blocked-state snapshots to decide whether blocked messaging is duplicate noise. */
export function isBlockedStateDuplicate(current: BlockedStateSnapshot, previous: BlockedStateSnapshot): boolean {
return current.blockedBy === previous.blockedBy && current.contextHash === previous.contextHash;
@@ -429,9 +440,9 @@ export class HeartbeatMonitor {
private pollIntervalMs: number;
private heartbeatTimeoutMs: number;
private maxConcurrentRuns: number;
private onMissed?: (agentId: string) => void;
private onMissed?: (agentId: string, reason: string) => void;
private onRecovered?: (agentId: string) => void;
private onTerminated?: (agentId: string) => void;
private onTerminated?: (agentId: string, reason: string) => void;
private onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
private onRunCompleted?: (agentId: string, run: AgentHeartbeatRun) => void;
private taskStore?: TaskStore;
@@ -1975,30 +1986,37 @@ export class HeartbeatMonitor {
const elapsed = now - tracked.lastSeen;
if (elapsed >= config.heartbeatTimeoutMs) {
const reason = `No heartbeat for ${formatDuration(elapsed)} (threshold: ${formatDuration(config.heartbeatTimeoutMs)})`;
// Missed heartbeat detected
if (!tracked.missedHeartbeatReported) {
tracked.missedHeartbeatReported = true;
await this.handleMissedHeartbeat(tracked);
await this.handleMissedHeartbeat(tracked, reason);
} else {
// Already reported - check if we should terminate
// Give 2x timeout for recovery before auto-terminate
if (elapsed >= config.heartbeatTimeoutMs * 2) {
await this.terminateUnresponsive(tracked);
await this.terminateUnresponsive(tracked, config.heartbeatTimeoutMs);
}
}
}
}
}
private async handleMissedHeartbeat(tracked: TrackedAgent): Promise<void> {
private async handleMissedHeartbeat(tracked: TrackedAgent, reason: string): Promise<void> {
// Record missed heartbeat
await this.store.recordHeartbeat(tracked.agentId, "missed", tracked.runId);
// Notify callback
this.onMissed?.(tracked.agentId);
this.onMissed?.(tracked.agentId, reason);
}
private async terminateUnresponsive(tracked: TrackedAgent): Promise<void> {
private async terminateUnresponsive(tracked: TrackedAgent, heartbeatTimeoutMs: number): Promise<void> {
const now = Date.now();
const elapsed = now - tracked.lastSeen;
const reason = `No heartbeat for ${formatDuration(elapsed)} (2× timeout threshold: ${formatDuration(heartbeatTimeoutMs * 2)})`;
heartbeatLog.warn(`Terminating unresponsive agent ${tracked.agentId}: ${reason}`);
// Dispose the session
try {
tracked.session.dispose();
@@ -2018,7 +2036,7 @@ export class HeartbeatMonitor {
this.trackedAgents.delete(tracked.agentId);
// Notify callback
this.onTerminated?.(tracked.agentId);
this.onTerminated?.(tracked.agentId, reason);
}
}

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

@@ -478,11 +478,11 @@ export class InProcessRuntime
rootDir: this.config.workingDirectory,
messageStore: this.messageStore,
pluginRunner: this.pluginRunner,
onMissed: (agentId) => {
runtimeLog.warn(`Agent ${agentId} missed heartbeat`);
onMissed: (agentId, reason) => {
runtimeLog.warn(`Agent ${agentId} missed heartbeat: ${reason}`);
},
onTerminated: (agentId) => {
runtimeLog.warn(`Agent ${agentId} terminated (unresponsive)`);
onTerminated: (agentId, reason) => {
runtimeLog.warn(`Agent ${agentId} terminated (unresponsive): ${reason}`);
},
});
this.heartbeatMonitor.start();