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

feat(FN-1049): add per-agent heartbeat configuration via runtimeConfig

Browse Source 
- Define AgentHeartbeatConfig interface in core types (heartbeatIntervalMs, heartbeatTimeoutMs, maxConcurrentRuns)
- Update HeartbeatMonitor to resolve per-agent config from AgentStore with validated min/max clamping
- Wire AgentStore into HeartbeatMonitor via InProcessRuntime initialization
- Add heartbeat settings section to dashboard AgentDetailView ConfigTab
- Add PATCH /api/agents/:id endpoint accepting runtimeConfig updates
- Add comprehensive tests for per-agent heartbeat config resolution and validation
- Document per-agent heartbeat configuration in AGENTS.md
This commit is contained in:
gsxdsm
2026-04-07 11:50:17 -07:00
parent 39f4f4929f
commit 825ae1061b
10 changed files with 579 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
AGENTS.md
View File

@@ -634,6 +634,33 @@ spawn_agent({
- **Cleanup:** `terminateAllChildren()` / `terminateChildAgent()` methods
- **Settings:** `packages/core/src/types.ts` (`ProjectSettings` interface)
## Per-Agent Heartbeat Configuration
Each agent can override the global heartbeat monitoring settings via `runtimeConfig`. The `AgentHeartbeatConfig` interface (exported from `@fusion/core`) defines the available keys:
| Key | Default | Min | Description |
|-----|---------|-----|-------------|
| `heartbeatIntervalMs` | 30000 | 1000 | How often heartbeats are checked |
| `heartbeatTimeoutMs` | 60000 | 5000 | Time without heartbeat before agent is considered unresponsive |
| `maxConcurrentRuns` | 1 | 1 | Max concurrent heartbeat runs per agent |
### How It Works
1. `HeartbeatMonitor` reads per-agent config from `AgentStore.getCachedAgent()` (synchronous file read)
2. Values from `agent.runtimeConfig` are validated and clamped to minimums
3. Missing or invalid values fall back to the monitor-level constructor defaults
4. Both `isAgentHealthy()` and `checkMissedHeartbeats()` use the per-agent timeout
5. The dashboard health status indicator reads `runtimeConfig.heartbeatTimeoutMs` for display
### Dashboard Configuration
The agent detail ConfigTab includes a "Heartbeat Settings" section where users can configure interval and timeout per agent. Values are stored in `agent.runtimeConfig` and persisted via `PATCH /api/agents/:id` with `runtimeConfig` in the request body.
### API
- `HeartbeatMonitor.getAgentHeartbeatConfig(agentId)` — Returns the resolved config for an agent
- `AgentStore.getCachedAgent(agentId)` — Synchronous agent read for hot paths
## Dashboard Task Creation
The dashboard provides two UI surfaces for creating tasks:

17
packages/core/src/agent-store.ts
View File

@@ -10,7 +10,7 @@
*/
import { mkdir, readFile, writeFile, readdir, unlink } from "node:fs/promises";
import { existsSync } from "node:fs";
import { existsSync, readFileSync } from "node:fs";
import { join } from "node:path";
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
@@ -732,6 +732,21 @@ export class AgentStore extends EventEmitter {
return JSON.parse(content) as AgentData;
}
/**
* Synchronously read an agent from disk (for use in synchronous hot paths).
* Returns null if the agent file does not exist or cannot be parsed.
* @param agentId - The agent ID
*/
getCachedAgent(agentId: string): Agent | null {
try {
const path = join(this.agentsDir, `${agentId}.json`);
const content = readFileSync(path, "utf-8");
return this.parseAgent(JSON.parse(content) as AgentData);
} catch {
return null;
}
}
private parseAgent(data: AgentData): Agent {
return {
id: data.id,

2
packages/core/src/index.ts
View File

@@ -1,5 +1,5 @@
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES } from "./types.js";
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskCreateInput, TaskDetail, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentCapability, AgentHeartbeatEvent, AgentHeartbeatRun, HeartbeatInvocationSource, AgentTaskSession, AgentStats, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox } from "./types.js";
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskCreateInput, TaskDetail, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentCapability, AgentHeartbeatConfig, AgentHeartbeatEvent, AgentHeartbeatRun, HeartbeatInvocationSource, AgentTaskSession, AgentStats, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox } from "./types.js";
export { AGENT_VALID_TRANSITIONS } from "./types.js";
export { AgentStore } from "./agent-store.js";
export type { AgentStoreEvents } from "./agent-store.js";

12
packages/core/src/types.ts
View File

@@ -1455,7 +1455,7 @@ export interface Agent {
icon?: string;
/** Agent ID this agent reports to (org hierarchy) */
reportsTo?: string;
/** Runtime configuration (maxTurns, thinkingLevel, etc.) */
/** Runtime configuration. Supports: AgentHeartbeatConfig keys (heartbeatIntervalMs, heartbeatTimeoutMs, maxConcurrentRuns) */
runtimeConfig?: Record<string, unknown>;
/** Why the agent was paused (error, manual, etc.) */
pauseReason?: string;
@@ -1469,6 +1469,16 @@ export interface Agent {
lastError?: string;
}
/** Per-agent heartbeat configuration, stored in agent.runtimeConfig */
export interface AgentHeartbeatConfig {
/** Polling interval in ms (default: 30000). Min: 1000 */
heartbeatIntervalMs?: number;
/** Heartbeat timeout in ms (default: 60000). Min: 5000 */
heartbeatTimeoutMs?: number;
/** Max concurrent heartbeat runs per agent (default: 1). Min: 1 */
maxConcurrentRuns?: number;
}
/** Extended agent information including heartbeat history */
export interface AgentDetail extends Agent {
/** Recent heartbeat events (last N events) */

121
packages/dashboard/app/components/AgentDetailView.tsx
View File

@@ -202,7 +202,7 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
}
const lastHeartbeat = new Date(agent.lastHeartbeatAt).getTime();
const elapsed = Date.now() - lastHeartbeat;
const timeoutMs = 60000;
const timeoutMs = (agent as any).runtimeConfig?.heartbeatTimeoutMs ?? 60000;
if (elapsed > timeoutMs) {
return { label: "Unresponsive", color: "var(--state-error-text, #f85149)" };
}
@@ -767,15 +767,6 @@ interface AdvancedSettingField {
/** Well-known advanced setting definitions backed by agent.metadata */
const ADVANCED_SETTINGS: AdvancedSettingField[] = [
{
key: "heartbeatIntervalMs",
label: "Heartbeat Interval (ms)",
type: "number",
placeholder: "30000",
hint: "How often the agent sends heartbeats (minimum 1000ms, default 30000ms)",
min: 1000,
max: 600000,
},
{
key: "maxRetries",
label: "Max Retries",
@@ -870,6 +861,19 @@ function ConfigTab({
return initial;
});
// Heartbeat config state initialised from agent.runtimeConfig
const [heartbeatValues, setHeartbeatValues] = useState<Record<string, string>>(() => {
const rc = agent.runtimeConfig ?? {};
const initial: Record<string, string> = {};
if (rc.heartbeatIntervalMs !== undefined && rc.heartbeatIntervalMs !== null) {
initial.heartbeatIntervalMs = String(rc.heartbeatIntervalMs);
}
if (rc.heartbeatTimeoutMs !== undefined && rc.heartbeatTimeoutMs !== null) {
initial.heartbeatTimeoutMs = String(rc.heartbeatTimeoutMs);
}
return initial;
});
const [isSaving, setIsSaving] = useState(false);
const [errors, setErrors] = useState<ValidationErrors>({});
const [justSaved, setJustSaved] = useState(false);
@@ -883,6 +887,13 @@ function ConfigTab({
: "";
if (current !== persisted) return true;
}
// Check heartbeat values
const rc = agent.runtimeConfig ?? {};
for (const key of ["heartbeatIntervalMs", "heartbeatTimeoutMs"] as const) {
const current = heartbeatValues[key]?.trim() ?? "";
const persisted = rc[key] !== undefined && rc[key] !== null ? String(rc[key]) : "";
if (current !== persisted) return true;
}
return false;
})();
@@ -899,9 +910,37 @@ function ConfigTab({
}
};
const handleHeartbeatFieldChange = (key: string, value: string) => {
setHeartbeatValues((prev) => ({ ...prev, [key]: value }));
setJustSaved(false);
if (errors[key]) {
setErrors((prev) => {
const next = { ...prev };
delete next[key];
return next;
});
}
};
const handleSave = async () => {
// Validate before save
// Validate advanced settings
const validationErrors = validateAdvancedSettings(formValues);
// Validate heartbeat settings
for (const [key, config] of Object.entries({
heartbeatIntervalMs: { label: "Heartbeat Interval", min: 1000 },
heartbeatTimeoutMs: { label: "Heartbeat Timeout", min: 5000 },
})) {
const raw = heartbeatValues[key]?.trim();
if (!raw) continue;
const num = Number(raw);
if (Number.isNaN(num) || !Number.isFinite(num)) {
validationErrors[key] = `"${config.label}" must be a valid number`;
} else if (num < config.min) {
validationErrors[key] = `"${config.label}" must be at least ${config.min.toLocaleString()}`;
}
}
if (Object.keys(validationErrors).length > 0) {
setErrors(validationErrors);
addToast("Please fix validation errors before saving", "error");
@@ -922,10 +961,21 @@ function ConfigTab({
}
}
// Build the runtimeConfig payload — only include non-empty values
const newRuntimeConfig: Record<string, unknown> = { ...agent.runtimeConfig };
for (const key of ["heartbeatIntervalMs", "heartbeatTimeoutMs"] as const) {
const raw = heartbeatValues[key]?.trim();
if (!raw) {
delete newRuntimeConfig[key];
} else {
newRuntimeConfig[key] = Number(raw);
}
}
setIsSaving(true);
try {
await updateAgent(agent.id, { metadata: newMetadata }, projectId);
addToast("Advanced settings saved", "success");
await updateAgent(agent.id, { metadata: newMetadata, runtimeConfig: newRuntimeConfig }, projectId);
addToast("Settings saved", "success");
setJustSaved(true);
// Auto-hide the saved indicator after 3 seconds
setTimeout(() => setJustSaved(false), 3000);
@@ -972,6 +1022,51 @@ function ConfigTab({
</div>
</div>
<div className="config-section">
<h3>Heartbeat Settings</h3>
<p className="config-description">
Configure how this agent's heartbeat is monitored. Leave a field empty to use system defaults.
</p>
<div className="config-fields">
<div className="config-field">
<label htmlFor="hb-heartbeatIntervalMs">Heartbeat Interval (ms)</label>
<input
id="hb-heartbeatIntervalMs"
type="text"
inputMode="numeric"
className={cn("input", !!errors.heartbeatIntervalMs && "input--error")}
placeholder="30000"
value={heartbeatValues.heartbeatIntervalMs ?? ""}
onChange={(e) => handleHeartbeatFieldChange("heartbeatIntervalMs", e.target.value)}
/>
{errors.heartbeatIntervalMs ? (
<span className="config-error">{errors.heartbeatIntervalMs}</span>
) : (
<span className="config-hint">How often heartbeats are checked. Leave empty for system default (30000ms)</span>
)}
</div>
<div className="config-field">
<label htmlFor="hb-heartbeatTimeoutMs">Heartbeat Timeout (ms)</label>
<input
id="hb-heartbeatTimeoutMs"
type="text"
inputMode="numeric"
className={cn("input", !!errors.heartbeatTimeoutMs && "input--error")}
placeholder="60000"
value={heartbeatValues.heartbeatTimeoutMs ?? ""}
onChange={(e) => handleHeartbeatFieldChange("heartbeatTimeoutMs", e.target.value)}
/>
{errors.heartbeatTimeoutMs ? (
<span className="config-error">{errors.heartbeatTimeoutMs}</span>
) : (
<span className="config-hint">Time without heartbeat before agent is considered unresponsive. Leave empty for system default (60000ms)</span>
)}
</div>
</div>
</div>
<div className="config-section">
<h3>Advanced Settings</h3>
<p className="config-description">

52
packages/dashboard/app/components/__tests__/AgentDetailView.test.tsx
View File

@@ -27,6 +27,7 @@ describe("AgentDetailView", () => {
role: AgentCapability;
state: "idle" | "active" | "paused" | "terminated";
taskId?: string;
runtimeConfig?: Record<string, unknown>;
}> = {}): AgentDetail => ({
id: "agent-001",
name: "Test Agent",
@@ -37,6 +38,7 @@ describe("AgentDetailView", () => {
updatedAt: "2024-01-01T00:00:00.000Z",
lastHeartbeatAt: "2024-01-01T00:05:00.000Z",
metadata: {},
runtimeConfig: overrides.runtimeConfig,
heartbeatHistory: [],
activeRun: {
id: "run-001",
@@ -455,14 +457,17 @@ describe("AgentDetailView", () => {
await navigateToSettings(user);
await waitFor(() => {
// Heartbeat Settings section
expect(screen.getByLabelText("Heartbeat Interval (ms)")).toBeInTheDocument();
expect(screen.getByLabelText("Heartbeat Timeout (ms)")).toBeInTheDocument();
// Advanced Settings section
expect(screen.getByLabelText("Max Retries")).toBeInTheDocument();
expect(screen.getByLabelText("Task Timeout (ms)")).toBeInTheDocument();
expect(screen.getByLabelText("Log Level")).toBeInTheDocument();
});
});
it("shows empty fields when metadata has no advanced settings", async () => {
it("shows empty fields when metadata and runtimeConfig are empty", async () => {
mockFetchAgent.mockResolvedValue(createMockAgent({ metadata: {} }));
const user = userEvent.setup();
@@ -482,10 +487,13 @@ describe("AgentDetailView", () => {
});
});
it("pre-fills fields from agent metadata", async () => {
it("pre-fills heartbeat fields from agent runtimeConfig", async () => {
mockFetchAgent.mockResolvedValue(createMockAgent({
metadata: {
runtimeConfig: {
heartbeatIntervalMs: 15000,
heartbeatTimeoutMs: 120000,
},
metadata: {
maxRetries: 5,
logLevel: "debug",
},
@@ -506,6 +514,9 @@ describe("AgentDetailView", () => {
const heartbeatInput = screen.getByLabelText("Heartbeat Interval (ms)") as HTMLInputElement;
expect(heartbeatInput.value).toBe("15000");
const heartbeatTimeoutInput = screen.getByLabelText("Heartbeat Timeout (ms)") as HTMLInputElement;
expect(heartbeatTimeoutInput.value).toBe("120000");
const retriesInput = screen.getByLabelText("Max Retries") as HTMLInputElement;
expect(retriesInput.value).toBe("5");
@@ -595,15 +606,15 @@ describe("AgentDetailView", () => {
await navigateToSettings(user);
const heartbeatInput = await screen.findByLabelText("Heartbeat Interval (ms)");
const heartbeatTimeoutInput = await screen.findByLabelText("Heartbeat Timeout (ms)");
await user.clear(heartbeatInput);
await user.type(heartbeatInput, "500");
await user.clear(heartbeatTimeoutInput);
await user.type(heartbeatTimeoutInput, "500");
await user.click(screen.getByText("Save Settings"));
await waitFor(() => {
expect(screen.getByText(/must be at least 1,000/)).toBeInTheDocument();
expect(screen.getByText(/must be at least 5,000/)).toBeInTheDocument();
});
});
@@ -631,7 +642,7 @@ describe("AgentDetailView", () => {
});
});
it("calls updateAgent with correct metadata on save", async () => {
it("calls updateAgent with correct metadata and runtimeConfig on save", async () => {
const addToast = vi.fn();
mockUpdateAgent.mockResolvedValue(createMockAgent() as any);
@@ -656,12 +667,15 @@ describe("AgentDetailView", () => {
await waitFor(() => {
expect(mockUpdateAgent).toHaveBeenCalledWith(
"agent-001",
{ metadata: expect.objectContaining({ heartbeatIntervalMs: 15000 }) },
expect.objectContaining({
metadata: expect.any(Object),
runtimeConfig: expect.objectContaining({ heartbeatIntervalMs: 15000 }),
}),
undefined,
);
});
expect(addToast).toHaveBeenCalledWith("Advanced settings saved", "success");
expect(addToast).toHaveBeenCalledWith("Settings saved", "success");
});
it("forwards projectId to updateAgent", async () => {
@@ -799,9 +813,9 @@ describe("AgentDetailView", () => {
expect((logLevelSelect as HTMLSelectElement).value).toBe("debug");
});
it("clears metadata key when field is cleared to empty", async () => {
it("clears runtimeConfig key when heartbeat field is cleared to empty", async () => {
mockFetchAgent.mockResolvedValue(createMockAgent({
metadata: { heartbeatIntervalMs: 30000 },
runtimeConfig: { heartbeatIntervalMs: 30000 },
}));
mockUpdateAgent.mockResolvedValue(createMockAgent() as any);
@@ -827,16 +841,17 @@ describe("AgentDetailView", () => {
expect(mockUpdateAgent).toHaveBeenCalledWith(
"agent-001",
expect.objectContaining({
metadata: expect.not.objectContaining({ heartbeatIntervalMs: expect.anything() }),
runtimeConfig: expect.not.objectContaining({ heartbeatIntervalMs: expect.anything() }),
}),
undefined,
);
});
});
it("persists existing non-advanced metadata keys during save", async () => {
it("persists existing non-advanced metadata keys and runtimeConfig during save", async () => {
mockFetchAgent.mockResolvedValue(createMockAgent({
metadata: { customKey: "preserved", heartbeatIntervalMs: 30000 },
metadata: { customKey: "preserved" },
runtimeConfig: { heartbeatIntervalMs: 30000, otherConfig: "also-preserved" },
}));
mockUpdateAgent.mockResolvedValue(createMockAgent() as any);
@@ -860,9 +875,10 @@ describe("AgentDetailView", () => {
await waitFor(() => {
const call = mockUpdateAgent.mock.calls[0];
const metadata = (call as any)[1].metadata;
expect(metadata.customKey).toBe("preserved");
expect(metadata.heartbeatIntervalMs).toBe(45000);
const payload = (call as any)[1];
expect(payload.metadata.customKey).toBe("preserved");
expect(payload.runtimeConfig.heartbeatIntervalMs).toBe(45000);
expect(payload.runtimeConfig.otherConfig).toBe("also-preserved");
});
});
});

4
packages/dashboard/src/routes.ts
View File

@@ -7004,14 +7004,14 @@ Output ONLY the prompt text (no markdown, no explanations).`;
*/
router.patch("/agents/:id", async (req, res) => {
try {
const { name, role, metadata } = req.body;
const { name, role, metadata, runtimeConfig } = req.body;
const scopedStore = await getScopedStore(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
const agent = await agentStore.updateAgent(req.params.id, { name, role, metadata });
const agent = await agentStore.updateAgent(req.params.id, { name, role, metadata, runtimeConfig });
res.json(agent);
} catch (err: any) {
if (err.message?.includes("not found")) {

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

@@ -408,4 +408,315 @@ describe("HeartbeatMonitor", () => {
expect(() => monitor.untrackAgent("agent-001")).not.toThrow();
});
});
// ── Per-Agent Config Tests ──────────────────────────────────────────────
describe("per-agent heartbeat config", () => {
/** Create a mock store that returns a specific agent from getCachedAgent */
function createStoreWithAgent(agent: { id: string; runtimeConfig?: Record<string, unknown> }): AgentStore {
return {
recordHeartbeat: vi.fn().mockResolvedValue(undefined),
updateAgentState: vi.fn().mockResolvedValue(undefined),
getCachedAgent: vi.fn().mockReturnValue(agent),
} as unknown as AgentStore;
}
describe("getAgentHeartbeatConfig", () => {
it("returns monitor defaults when agentStore is not provided", () => {
const monitor = new HeartbeatMonitor({
store,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
maxConcurrentRuns: 2,
});
const config = 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", () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("returns per-agent values when runtimeConfig is set", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: {
heartbeatIntervalMs: 2000,
heartbeatTimeoutMs: 30000,
maxConcurrentRuns: 3,
},
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
maxConcurrentRuns: 1,
});
const config = 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", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatIntervalMs: 100 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(1000);
});
it("clamps heartbeatTimeoutMs to minimum of 5000", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 1000 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
heartbeatTimeoutMs: 60000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.heartbeatTimeoutMs).toBe(5000);
});
it("clamps maxConcurrentRuns to minimum of 1", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { maxConcurrentRuns: 0 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
maxConcurrentRuns: 1,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.maxConcurrentRuns).toBe(1);
});
it("falls back to monitor defaults when runtimeConfig values are NaN", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: {
heartbeatIntervalMs: NaN,
heartbeatTimeoutMs: "not a number" as any,
},
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
});
const config = 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", () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
(agentStore.getCachedAgent as ReturnType<typeof vi.fn>).mockReturnValue(null);
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
});
const config = monitor.getAgentHeartbeatConfig("agent-999");
expect(config.pollIntervalMs).toBe(5000);
expect(config.heartbeatTimeoutMs).toBe(10000);
});
it("returns monitor defaults when getCachedAgent throws", () => {
const agentStore = createStoreWithAgent({ id: "agent-001" });
(agentStore.getCachedAgent as ReturnType<typeof vi.fn>).mockImplementation(() => {
throw new Error("Read error");
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 10000,
});
const config = 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", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 120000 },
});
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 5000,
heartbeatTimeoutMs: 60000,
maxConcurrentRuns: 1,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.pollIntervalMs).toBe(5000); // fallback
expect(config.heartbeatTimeoutMs).toBe(120000); // overridden
expect(config.maxConcurrentRuns).toBe(1); // fallback
});
});
describe("isAgentHealthy with per-agent config", () => {
it("uses per-agent timeout for health check", () => {
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 30000 },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
agentStore,
heartbeatTimeoutMs: 5000, // Global default is 5000
});
monitor.trackAgent("agent-001", session, "run-001");
// Advance 10s — past the global 5s default, but within the per-agent 30s
vi.advanceTimersByTime(10000);
expect(monitor.isAgentHealthy("agent-001")).toBe(true);
// Advance past per-agent 30s timeout
vi.advanceTimersByTime(25000);
expect(monitor.isAgentHealthy("agent-001")).toBe(false);
vi.useRealTimers();
});
});
describe("checkMissedHeartbeats with per-agent config", () => {
it("detects missed heartbeat using per-agent timeout", async () => {
const onMissed = vi.fn();
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 10000 },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 1000,
heartbeatTimeoutMs: 5000, // Global default 5s — agent overrides to 10s
onMissed,
});
monitor.start();
monitor.trackAgent("agent-001", session, "run-001");
// Advance 6s — past global 5s but within per-agent 10s
vi.advanceTimersByTime(6000);
await vi.advanceTimersByTimeAsync(100);
// Should NOT have triggered onMissed because per-agent timeout is 10s
expect(onMissed).not.toHaveBeenCalled();
// Advance past the 10s per-agent timeout
vi.advanceTimersByTime(5000);
await vi.advanceTimersByTimeAsync(100);
expect(onMissed).toHaveBeenCalledWith("agent-001");
monitor.stop();
vi.useRealTimers();
});
it("terminates unresponsive agent using per-agent timeout", async () => {
const onTerminated = vi.fn();
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 5000 },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
agentStore,
pollIntervalMs: 1000,
heartbeatTimeoutMs: 60000, // Global default 60s — agent overrides to 5s
onTerminated,
});
monitor.start();
monitor.trackAgent("agent-001", session, "run-001");
// Wait for missed (5s) + termination at 2x timeout (10s)
vi.advanceTimersByTime(12000);
await vi.advanceTimersByTimeAsync(100);
expect(session.dispose).toHaveBeenCalled();
expect(onTerminated).toHaveBeenCalledWith("agent-001");
monitor.stop();
vi.useRealTimers();
});
});
describe("backward compatibility", () => {
it("works without agentStore (no per-agent config)", () => {
const monitor = new HeartbeatMonitor({
store,
heartbeatTimeoutMs: 5000,
});
const config = monitor.getAgentHeartbeatConfig("agent-001");
expect(config.heartbeatTimeoutMs).toBe(5000);
expect(config.pollIntervalMs).toBe(30000); // default
expect(config.maxConcurrentRuns).toBe(1); // default
});
it("existing isAgentHealthy works without per-agent config", () => {
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
heartbeatTimeoutMs: 5000,
});
monitor.trackAgent("agent-001", session, "run-001");
expect(monitor.isAgentHealthy("agent-001")).toBe(true);
vi.advanceTimersByTime(6000);
expect(monitor.isAgentHealthy("agent-001")).toBe(false);
vi.useRealTimers();
});
});
});
});

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

@@ -10,12 +10,22 @@
* - onTerminated: Called when an unresponsive agent is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig } from "@fusion/core";
/** Resolved per-agent heartbeat config after validation and fallback */
interface ResolvedHeartbeatConfig {
pollIntervalMs: number;
heartbeatTimeoutMs: number;
maxConcurrentRuns: number;
}
/** Options for HeartbeatMonitor constructor */
export interface HeartbeatMonitorOptions {
/** AgentStore instance for persistence */
store: AgentStore;
/** Optional separate AgentStore reference for reading per-agent runtimeConfig.
* If not provided, falls back to `store`. */
agentStore?: AgentStore;
/** Polling interval in milliseconds (default: 30000) */
pollIntervalMs?: number;
/** Heartbeat timeout in milliseconds (default: 60000) */
@@ -67,6 +77,7 @@ interface TrackedAgent {
*/
export class HeartbeatMonitor {
private store: AgentStore;
private configStore: AgentStore;
private pollIntervalMs: number;
private heartbeatTimeoutMs: number;
private maxConcurrentRuns: number;
@@ -83,6 +94,7 @@ export class HeartbeatMonitor {
constructor(options: HeartbeatMonitorOptions) {
this.store = options.store;
this.configStore = options.agentStore ?? options.store;
this.pollIntervalMs = options.pollIntervalMs ?? 30000;
this.heartbeatTimeoutMs = options.heartbeatTimeoutMs ?? 60000;
this.maxConcurrentRuns = options.maxConcurrentRuns ?? 1;
@@ -301,6 +313,8 @@ export class HeartbeatMonitor {
/**
* Check if an agent is healthy (heartbeat within timeout window).
* Uses per-agent heartbeatTimeoutMs from runtimeConfig if available,
* otherwise falls back to the monitor-level default.
* @param agentId - The agent ID
* @returns true if healthy, false if missed heartbeat or not tracked
*/
@@ -308,8 +322,9 @@ export class HeartbeatMonitor {
const tracked = this.trackedAgents.get(agentId);
if (!tracked) return false;
const config = this.getAgentConfig(agentId);
const elapsed = Date.now() - tracked.lastSeen;
return elapsed < this.heartbeatTimeoutMs;
return elapsed < config.heartbeatTimeoutMs;
}
/**
@@ -333,13 +348,62 @@ export class HeartbeatMonitor {
// Private methods
// ─────────────────────────────────────────────────────────────────────────
/**
* Get the resolved heartbeat configuration for an agent.
* Reads per-agent config from runtimeConfig with fallback to monitor defaults.
* @param agentId - The agent ID
* @returns Resolved config with validated values
*/
getAgentHeartbeatConfig(agentId: string): ResolvedHeartbeatConfig {
return this.getAgentConfig(agentId);
}
/**
* Resolve per-agent heartbeat config from runtimeConfig with validation and fallbacks.
*/
private getAgentConfig(agentId: string): ResolvedHeartbeatConfig {
// Defaults from monitor-level construction
const result: ResolvedHeartbeatConfig = {
pollIntervalMs: this.pollIntervalMs,
heartbeatTimeoutMs: this.heartbeatTimeoutMs,
maxConcurrentRuns: this.maxConcurrentRuns,
};
try {
// Synchronous read — AgentStore.getAgent is async, but we can't make this
// method async without changing the call chain. Instead, we'll resolve
// per-agent config on the checkMissedHeartbeats path (which is async).
// For synchronous callers (isAgentHealthy), we use a cached approach.
// For simplicity, we read from the store's underlying agent data.
const agent = this.configStore.getCachedAgent?.(agentId);
if (agent?.runtimeConfig) {
const rc = agent.runtimeConfig;
if (typeof rc.heartbeatIntervalMs === "number" && Number.isFinite(rc.heartbeatIntervalMs)) {
result.pollIntervalMs = Math.max(1000, rc.heartbeatIntervalMs);
}
if (typeof rc.heartbeatTimeoutMs === "number" && Number.isFinite(rc.heartbeatTimeoutMs)) {
result.heartbeatTimeoutMs = Math.max(5000, rc.heartbeatTimeoutMs);
}
if (typeof rc.maxConcurrentRuns === "number" && Number.isFinite(rc.maxConcurrentRuns)) {
result.maxConcurrentRuns = Math.max(1, Math.round(rc.maxConcurrentRuns));
}
}
} catch {
// If agent lookup fails, use monitor defaults
}
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 elapsed = now - tracked.lastSeen;
if (elapsed >= this.heartbeatTimeoutMs) {
if (elapsed >= config.heartbeatTimeoutMs) {
// Missed heartbeat detected
if (!tracked.missedHeartbeatReported) {
tracked.missedHeartbeatReported = true;
@@ -347,7 +411,7 @@ export class HeartbeatMonitor {
} else {
// Already reported - check if we should terminate
// Give 2x timeout for recovery before auto-terminate
if (elapsed >= this.heartbeatTimeoutMs * 2) {
if (elapsed >= config.heartbeatTimeoutMs * 2) {
await this.terminateUnresponsive(tracked);
}
}

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

@@ -213,6 +213,7 @@ export class InProcessRuntime
this.heartbeatMonitor = new HeartbeatMonitor({
store: this.agentStore,
agentStore: this.agentStore, // enables per-agent config resolution
onMissed: (agentId) => {
runtimeLog.warn(`Agent ${agentId} missed heartbeat`);
},