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

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

Browse Source 
Fusion-Task-Id: FN-3497
This commit is contained in:
gsxdsm
2026-05-05 04:49:45 -07:00
parent f9dfab8961
commit af0bc4bd07
7 changed files with 247 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-3475-auto-pause-resume.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Auto-pause unresponsive agents with `pauseReason: "heartbeat-unresponsive"` and immediately auto-resume them through the shared heartbeat monitor lifecycle, including consistent assigned-task pause/unpause behavior and single on-demand restart semantics.

15
docs/agents.md
View File

@@ -657,6 +657,21 @@ Heartbeat timers are armed for agents in valid working states and remain armed a
- Ephemeral/task-worker agents are never armed with timers (managed directly by TaskExecutor)
- The `runtimeConfig.enabled` flag is respected for disabling heartbeat monitoring entirely
### Unresponsive Recovery (FN-3475)
When a tracked agent misses heartbeat for `2 × heartbeatTimeoutMs`, the monitor now performs recovery (not termination):
1. Dispose the stuck session and untrack the stale run
2. `pauseAgent(agentId, { pauseReason: "heartbeat-unresponsive", stopActiveRun: false })`
3. `resumeAgent(agentId, { triggerDetail: "unresponsive-recovery", triggerSource: "heartbeat-unresponsive", clearPauseReason: true })`
Effects:
- Agent state transitions `running/active → paused → active`
- `pauseReason` is set to `heartbeat-unresponsive` during recovery and cleared on resume
- Assigned tasks are auto-paused with `pausedByAgentId` during pause, then only those same tasks are auto-unpaused on resume
- Resume triggers one on-demand heartbeat restart only when `runtimeConfig.enabled !== false`
- `onTerminated` is reserved for true termination flows and is not used by unresponsive recovery
## Dashboard Health Status
The dashboard displays agent health status in AgentsView, AgentListModal, and AgentDetailView using a centralized health evaluation utility (`packages/dashboard/app/utils/agentHealth.ts`).

93
packages/dashboard/src/__tests__/routes-agent-runs.test.ts
View File

@@ -464,6 +464,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
let app: ReturnType<typeof import("../server.js").createServer>;
let mockExecuteHeartbeat: ReturnType<typeof vi.fn>;
let mockStopRun: ReturnType<typeof vi.fn>;
let mockPauseAgent: ReturnType<typeof vi.fn>;
let mockResumeAgent: ReturnType<typeof vi.fn>;
beforeEach(async () => {
vi.clearAllMocks();
@@ -475,6 +477,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
mockExecuteHeartbeat = vi.fn();
mockStopRun = vi.fn();
mockPauseAgent = vi.fn();
mockResumeAgent = vi.fn();
store = new MockStore();
const { createServer } = await import("../server.js");
@@ -482,6 +486,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
heartbeatMonitor: {
executeHeartbeat: mockExecuteHeartbeat,
stopRun: mockStopRun,
pauseAgent: mockPauseAgent,
resumeAgent: mockResumeAgent,
},
});
});
@@ -491,10 +497,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
});
describe("POST /api/agents/:id/state", () => {
it("pauses and stops active run asynchronously after responding", async () => {
mockGetActiveHeartbeatRun.mockResolvedValue(createMockRun({ id: "run-pause-1" }));
mockStopRun.mockResolvedValue(undefined);
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "paused" });
it("delegates pause transitions to heartbeat monitor lifecycle helper", async () => {
mockPauseAgent.mockResolvedValue({ id: "agent-001", state: "paused", pauseReason: "manual" });
const response = await request(
app,
@@ -505,44 +509,42 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
);
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "paused" });
expect(mockUpdateAgentState).toHaveBeenCalledWith("agent-001", "paused");
await vi.waitFor(() => {
expect(mockStopRun).toHaveBeenCalledWith("agent-001");
});
expect(response.body).toEqual({ id: "agent-001", state: "paused", pauseReason: "manual" });
expect(mockPauseAgent).toHaveBeenCalledWith("agent-001", { pauseReason: undefined, stopActiveRun: true });
expect(mockUpdateAgentState).not.toHaveBeenCalled();
expect(store.pauseTask).not.toHaveBeenCalled();
});
it("pausing agent auto-pauses only non-paused assigned tasks", async () => {
(store.getTasksByAssignedAgent as ReturnType<typeof vi.fn>).mockResolvedValueOnce([
{ id: "FN-1", paused: false },
{ id: "FN-2", paused: true },
{ id: "FN-3" },
]);
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "paused" });
it("delegates resume transitions to heartbeat monitor lifecycle helper", async () => {
mockResumeAgent.mockResolvedValue({ id: "agent-001", state: "active" });
const response = await request(
app,
"POST",
"/api/agents/agent-001/state",
JSON.stringify({ state: "paused" }),
JSON.stringify({ state: "active" }),
{ "content-type": "application/json" },
);
expect(response.status).toBe(200);
await vi.waitFor(() => {
expect(store.pauseTask).toHaveBeenCalledTimes(2);
expect(response.body).toEqual({ id: "agent-001", state: "active" });
expect(mockResumeAgent).toHaveBeenCalledWith("agent-001", {
triggerDetail: "Triggered from state resume",
triggerSource: "state-resume",
clearPauseReason: true,
});
expect(store.pauseTask).toHaveBeenCalledWith("FN-1", true, undefined, { pausedByAgentId: "agent-001" });
expect(store.pauseTask).toHaveBeenCalledWith("FN-3", true, undefined, { pausedByAgentId: "agent-001" });
expect(mockExecuteHeartbeat).not.toHaveBeenCalled();
});
it("resuming agent only unpauses tasks paused by that same agent", async () => {
it("falls back to direct state update when monitor lacks lifecycle helpers", async () => {
const { createServer } = await import("../server.js");
app = createServer(store as any, {
heartbeatMonitor: {
executeHeartbeat: mockExecuteHeartbeat,
stopRun: mockStopRun,
},
});
(store.getTasksByAssignedAgent as ReturnType<typeof vi.fn>).mockResolvedValueOnce([
{ id: "FN-1", paused: true, pausedByAgentId: "agent-001" },
{ id: "FN-2", paused: true, pausedByAgentId: "agent-002" },
{ id: "FN-3", paused: true },
]);
mockGetAgent.mockResolvedValue({ id: "agent-001", state: "paused" });
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "active" });
mockExecuteHeartbeat.mockResolvedValue(createMockRun({ id: "run-resume-1", status: "completed" }));
@@ -558,39 +560,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
await vi.waitFor(() => {
expect(store.pauseTask).toHaveBeenCalledWith("FN-1", false);
});
expect(store.pauseTask).toHaveBeenCalledTimes(1);
expect(mockExecuteHeartbeat).toHaveBeenCalledTimes(1);
});
it("resuming to active triggers on-demand heartbeat exactly once", async () => {
mockGetAgent.mockResolvedValue({ id: "agent-001", state: "paused" });
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "active" });
mockExecuteHeartbeat.mockResolvedValue(createMockRun({ id: "run-resume-1", status: "completed" }));
const response = await request(
app,
"POST",
"/api/agents/agent-001/state",
JSON.stringify({ state: "active" }),
{ "content-type": "application/json" },
);
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "active" });
await vi.waitFor(() => {
expect(mockExecuteHeartbeat).toHaveBeenCalledTimes(1);
});
expect(mockExecuteHeartbeat).toHaveBeenCalledWith({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "Triggered from state resume",
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from state resume",
triggerSource: "state-resume",
},
});
});
it("terminated agent also unpauses tasks paused by that agent", async () => {
(store.getTasksByAssignedAgent as ReturnType<typeof vi.fn>).mockResolvedValueOnce([
{ id: "FN-9", paused: true, pausedByAgentId: "agent-001" },
@@ -617,7 +588,7 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
state: "paused",
runtimeConfig: { enabled: false },
});
mockUpdateAgentState.mockResolvedValue({ id: "agent-001", state: "active" });
mockResumeAgent.mockResolvedValue({ id: "agent-001", state: "active" });
const response = await request(
app,
@@ -629,7 +600,7 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
expect(response.status).toBe(200);
expect(response.body).toEqual({ id: "agent-001", state: "active" });
await Promise.resolve();
expect(mockResumeAgent).toHaveBeenCalled();
expect(mockExecuteHeartbeat).not.toHaveBeenCalled();
});
});

47
packages/dashboard/src/routes/register-agent-runtime-routes.ts
View File

@@ -39,7 +39,7 @@ interface AgentRuntimeRouteDeps {
}
export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRuntimeRouteDeps): void {
const { router, getProjectContext, rethrowAsApiError } = ctx;
const { router, getProjectContext, rethrowAsApiError, runtimeLogger } = ctx;
const {
validateAgentInstructionsPayload,
serializeAccessState,
@@ -421,6 +421,33 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
? heartbeatMonitor
: null;
const lifecycleMonitor = projectHeartbeatMonitor as ({
pauseAgent?: (agentId: string, options?: { pauseReason?: string; stopActiveRun?: boolean }) => Promise<unknown>;
resumeAgent?: (agentId: string, options?: { triggerDetail?: string; triggerSource?: string; clearPauseReason?: boolean }) => Promise<unknown>;
} | null);
const pauseAgentHelper = lifecycleMonitor?.pauseAgent;
const resumeAgentHelper = lifecycleMonitor?.resumeAgent;
const supportsLifecycleHelpers = pauseAgentHelper && resumeAgentHelper;
if (supportsLifecycleHelpers && nextState === "paused") {
const paused = await pauseAgentHelper(agentId, {
pauseReason: currentAgent.pauseReason,
stopActiveRun: true,
});
res.json(paused);
return;
}
if (supportsLifecycleHelpers && nextState === "active") {
const resumed = await resumeAgentHelper(agentId, {
triggerDetail: "Triggered from state resume",
triggerSource: "state-resume",
clearPauseReason: true,
});
res.json(resumed);
return;
}
const updatedAgent = await agentStore.updateAgentState(agentId, nextState);
res.json(updatedAgent);
@@ -445,7 +472,11 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
);
results.forEach((result, index) => {
if (result.status === "rejected") {
console.error(`[agent-state] failed to pause assigned task ${toPause[index]?.id} for ${agentId}:`, result.reason);
runtimeLogger.child("agent-state").warn("Failed to auto-pause assigned task", {
agentId,
taskId: toPause[index]?.id,
error: String(result.reason),
});
}
});
}
@@ -461,7 +492,11 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
);
results.forEach((result, index) => {
if (result.status === "rejected") {
console.error(`[agent-state] failed to unpause assigned task ${toUnpause[index]?.id} for ${agentId}:`, result.reason);
runtimeLogger.child("agent-state").warn("Failed to auto-unpause assigned task", {
agentId,
taskId: toUnpause[index]?.id,
error: String(result.reason),
});
}
});
}
@@ -480,7 +515,11 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
});
}
} catch (err) {
console.error(`[agent-state] async heartbeat work failed for ${agentId}:`, err);
runtimeLogger.child("agent-state").warn("Async state transition follow-up failed", {
agentId,
nextState,
error: err instanceof Error ? err.message : String(err),
});
}
})();
} catch (err: unknown) {

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

@@ -48,6 +48,8 @@ describe("per-agent heartbeat config", () => {
return {
recordHeartbeat: vi.fn().mockResolvedValue(undefined),
updateAgentState: vi.fn().mockResolvedValue(undefined),
updateAgent: vi.fn().mockResolvedValue(undefined),
getAgent: vi.fn().mockResolvedValue({ ...agent, state: "running" }),
getCachedAgent: vi.fn().mockReturnValue(agent),
} as unknown as AgentStore;
}
@@ -322,17 +324,21 @@ describe("per-agent heartbeat config", () => {
vi.useRealTimers();
});
it("terminates unresponsive agent using per-agent timeout", async () => {
it("recovers unresponsive agent using per-agent timeout", async () => {
const onTerminated = vi.fn();
const agentStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { heartbeatTimeoutMs: 5000 },
});
const runtimeStore = createStoreWithAgent({
id: "agent-001",
runtimeConfig: { enabled: false },
});
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
const monitor = new HeartbeatMonitor({
store,
store: runtimeStore,
agentStore,
pollIntervalMs: 1000,
heartbeatTimeoutMs: 60000, // Global default 60s — agent overrides to 5s
@@ -346,7 +352,9 @@ describe("per-agent heartbeat config", () => {
await vi.advanceTimersByTimeAsync(100);
expect(session.dispose).toHaveBeenCalled();
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
expect(runtimeStore.updateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(runtimeStore.updateAgentState).toHaveBeenCalledWith("agent-001", "active");
expect(onTerminated).not.toHaveBeenCalled();
monitor.stop();
vi.useRealTimers();

124
packages/engine/src/__tests__/heartbeat-monitor.test.ts
View File

@@ -603,12 +603,17 @@ describe("missed heartbeat detection", () => {
});
});
describe("unresponsive agent termination", () => {
it("disposes session and terminates agent after 2x timeout", async () => {
describe("unresponsive agent recovery", () => {
it("disposes session and pauses/resumes agent after 2x timeout", async () => {
const onTerminated = vi.fn();
const session = createMockSession();
const localStore = createMockStore({
getAgent: vi.fn().mockResolvedValue({ id: "agent-001", state: "running", runtimeConfig: { enabled: false } }),
updateAgentState: vi.fn().mockResolvedValue(undefined),
updateAgent: vi.fn().mockResolvedValue(undefined),
});
const customMonitor = new HeartbeatMonitor({
store,
store: localStore,
heartbeatTimeoutMs: 5000,
pollIntervalMs: 1000,
onTerminated,
@@ -618,113 +623,20 @@ describe("unresponsive agent termination", () => {
customMonitor.start();
customMonitor.trackAgent("agent-001", session, "run-001");
// Wait for missed heartbeat (1x timeout)
vi.advanceTimersByTime(6000);
await vi.advanceTimersByTimeAsync(100);
// Wait for termination (2x timeout = 10 seconds total from start)
vi.advanceTimersByTime(6000);
await vi.advanceTimersByTimeAsync(100);
expect(session.dispose).toHaveBeenCalled();
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "terminated");
expect(onTerminated).toHaveBeenCalledWith("agent-001", expect.any(String));
customMonitor.stop();
vi.useRealTimers();
});
it("removes agent from tracking after termination", async () => {
const session = createMockSession();
const customMonitor = new HeartbeatMonitor({
store,
heartbeatTimeoutMs: 5000,
pollIntervalMs: 1000,
});
vi.useFakeTimers({ shouldAdvanceTime: true });
customMonitor.start();
customMonitor.trackAgent("agent-001", session, "run-001");
expect(customMonitor.getTrackedAgents()).toContain("agent-001");
// Wait for termination
vi.advanceTimersByTime(12000);
await vi.advanceTimersByTimeAsync(100);
expect(customMonitor.getTrackedAgents()).not.toContain("agent-001");
customMonitor.stop();
vi.useRealTimers();
});
it("logs warning when session dispose throws during termination", async () => {
const warnSpy = vi.mocked(heartbeatLog.warn);
warnSpy.mockClear();
const session: AgentSession = {
dispose: vi.fn(() => {
throw new Error("dispose exploded");
}),
};
const updateAgentState = vi.fn().mockResolvedValue(undefined);
const localStore = createMockStore({ updateAgentState });
const onTerminated = vi.fn();
const customMonitor = new HeartbeatMonitor({
store: localStore,
heartbeatTimeoutMs: 5000,
pollIntervalMs: 1000,
onTerminated,
});
vi.useFakeTimers({ shouldAdvanceTime: true });
customMonitor.start();
customMonitor.trackAgent("agent-001", session, "run-001");
vi.advanceTimersByTime(10100);
await vi.advanceTimersByTimeAsync(100);
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.any(String));
expect(session.dispose).toHaveBeenCalled();
expect(localStore.updateAgentState).toHaveBeenCalledWith("agent-001", "paused");
expect(localStore.updateAgentState).toHaveBeenCalledWith("agent-001", "active");
expect(onTerminated).not.toHaveBeenCalled();
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
customMonitor.stop();
vi.useRealTimers();
});
it("logs warning when updateAgentState throws during termination", async () => {
const warnSpy = vi.mocked(heartbeatLog.warn);
warnSpy.mockClear();
const session = createMockSession();
const localStore = createMockStore({
updateAgentState: vi.fn().mockRejectedValue(new Error("db connection lost")),
});
const onTerminated = vi.fn();
const customMonitor = new HeartbeatMonitor({
store: localStore,
heartbeatTimeoutMs: 5000,
pollIntervalMs: 1000,
onTerminated,
});
vi.useFakeTimers({ shouldAdvanceTime: true });
customMonitor.start();
customMonitor.trackAgent("agent-001", session, "run-001");
vi.advanceTimersByTime(10100);
await vi.advanceTimersByTimeAsync(100);
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.any(String));
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
customMonitor.stop();
vi.useRealTimers();
});
it("logs warnings from both dispose and state update when both fail", async () => {
it("logs recovery warnings when dispose and pause fail", async () => {
const warnSpy = vi.mocked(heartbeatLog.warn);
warnSpy.mockClear();
const session: AgentSession = {
@@ -733,14 +645,13 @@ describe("unresponsive agent termination", () => {
}),
};
const localStore = createMockStore({
getAgent: vi.fn().mockResolvedValue({ id: "agent-001", state: "running", runtimeConfig: { enabled: false } }),
updateAgentState: vi.fn().mockRejectedValue(new Error("db connection lost")),
});
const onTerminated = vi.fn();
const customMonitor = new HeartbeatMonitor({
store: localStore,
heartbeatTimeoutMs: 5000,
pollIntervalMs: 1000,
onTerminated,
});
vi.useFakeTimers({ shouldAdvanceTime: true });
@@ -751,12 +662,9 @@ describe("unresponsive agent termination", () => {
await vi.advanceTimersByTimeAsync(100);
const warnMessages = warnSpy.mock.calls.map(([message]) => String(message));
expect(warnMessages).toHaveLength(3);
expect(warnMessages.some((message) => message.includes("Terminating unresponsive agent agent-001"))).toBe(true);
expect(warnMessages.some((message) => message.includes("Recovering 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.any(String));
expect(customMonitor.getTrackedAgents()).toHaveLength(0);
expect(warnMessages.some((message) => message.includes("Error pausing unresponsive agent agent-001") && message.includes("db connection lost"))).toBe(true);
customMonitor.stop();
vi.useRealTimers();

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

@@ -112,6 +112,17 @@ export interface HeartbeatExecutionOptions {
contextSnapshot?: Record<string, unknown>;
}
export interface PauseAgentOptions {
pauseReason?: string;
stopActiveRun?: boolean;
}
export interface ResumeAgentOptions {
triggerDetail?: string;
triggerSource?: string;
clearPauseReason?: boolean;
}
/** Session interface for disposing agent resources */
export interface AgentSession {
/** Dispose the agent session (stop execution, cleanup resources) */
@@ -880,6 +891,104 @@ export class HeartbeatMonitor {
this.clearRunState(agentId);
}
async pauseAgent(agentId: string, options: PauseAgentOptions = {}): Promise<Agent> {
const { pauseReason, stopActiveRun = false } = options;
if (stopActiveRun) {
try {
await this.stopRun(agentId);
} catch (error) {
heartbeatLog.warn(`pauseAgent(${agentId}) stopRun failed: ${error instanceof Error ? error.message : String(error)}`);
}
}
const current = await this.store.getAgent(agentId);
if (!current) {
throw new Error(`Agent ${agentId} not found`);
}
let updated = current;
if (current.state !== "paused") {
updated = await this.store.updateAgentState(agentId, "paused");
}
if (pauseReason !== undefined && updated.pauseReason !== pauseReason) {
updated = await this.store.updateAgent(agentId, { pauseReason });
}
if (this.taskStore) {
const assignedTasks = await this.taskStore.getTasksByAssignedAgent(agentId, { excludeArchived: true });
const toPause = assignedTasks.filter((task) => task.paused !== true);
const results = await Promise.allSettled(
toPause.map((task) => this.taskStore!.pauseTask(task.id, true, undefined, { pausedByAgentId: agentId })),
);
results.forEach((result, index) => {
if (result.status === "rejected") {
heartbeatLog.warn(`pauseAgent(${agentId}) failed to pause assigned task ${toPause[index]?.id}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`);
}
});
}
return updated;
}
async resumeAgent(agentId: string, options: ResumeAgentOptions = {}): Promise<Agent> {
const {
triggerDetail = "Triggered from state resume",
triggerSource = "state-resume",
clearPauseReason = true,
} = options;
const current = await this.store.getAgent(agentId);
if (!current) {
throw new Error(`Agent ${agentId} not found`);
}
let updated = current;
if (current.state !== "active") {
updated = await this.store.updateAgentState(agentId, "active");
}
if (clearPauseReason && updated.pauseReason !== undefined) {
updated = await this.store.updateAgent(agentId, { pauseReason: undefined });
}
if (this.taskStore) {
const pausedTasks = await this.taskStore.getTasksByAssignedAgent(agentId, {
pausedOnly: true,
excludeArchived: true,
});
const toUnpause = pausedTasks.filter((task) => task.pausedByAgentId === agentId);
const results = await Promise.allSettled(toUnpause.map((task) => this.taskStore!.pauseTask(task.id, false)));
results.forEach((result, index) => {
if (result.status === "rejected") {
heartbeatLog.warn(`resumeAgent(${agentId}) failed to unpause assigned task ${toUnpause[index]?.id}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`);
}
});
}
const latest = await this.store.getAgent(agentId);
const isHeartbeatEnabled = latest?.runtimeConfig?.enabled !== false;
if (isHeartbeatEnabled) {
try {
await this.executeHeartbeat({
agentId,
source: "on_demand",
triggerDetail,
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail,
triggerSource,
},
});
} catch (error) {
heartbeatLog.warn(`resumeAgent(${agentId}) executeHeartbeat failed: ${error instanceof Error ? error.message : String(error)}`);
}
}
return (await this.store.getAgent(agentId)) ?? updated;
}
/**
* Remove an agent from monitoring.
* Does NOT end the heartbeat run - caller's responsibility.
@@ -2228,7 +2337,7 @@ export class HeartbeatMonitor {
// Already reported - check if we should terminate
// Give 2x timeout for recovery before auto-terminate
if (elapsed >= config.heartbeatTimeoutMs * 2) {
await this.terminateUnresponsive(tracked, config.heartbeatTimeoutMs);
await this.recoverUnresponsiveAgent(tracked, config.heartbeatTimeoutMs);
}
}
}
@@ -2243,33 +2352,36 @@ export class HeartbeatMonitor {
this.onMissed?.(tracked.agentId, reason);
}
private async terminateUnresponsive(tracked: TrackedAgent, heartbeatTimeoutMs: number): Promise<void> {
private async recoverUnresponsiveAgent(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}`);
heartbeatLog.warn(`Recovering unresponsive agent ${tracked.agentId}: ${reason}`);
// Dispose the session
try {
tracked.session.dispose();
} catch (err) {
// Log but don't stop termination
heartbeatLog.warn(`Error disposing session for ${tracked.agentId}: ${err instanceof Error ? err.message : String(err)}`);
}
// Update agent state to terminated
this.untrackAgent(tracked.agentId);
try {
await this.store.updateAgentState(tracked.agentId, "terminated");
await this.pauseAgent(tracked.agentId, { pauseReason: "heartbeat-unresponsive", stopActiveRun: false });
} catch (err) {
heartbeatLog.warn(`Error terminating agent ${tracked.agentId}: ${err instanceof Error ? err.message : String(err)}`);
heartbeatLog.warn(`Error pausing unresponsive agent ${tracked.agentId}: ${err instanceof Error ? err.message : String(err)}`);
}
// Remove from tracking
this.trackedAgents.delete(tracked.agentId);
// Notify callback
this.onTerminated?.(tracked.agentId, reason);
try {
await this.resumeAgent(tracked.agentId, {
triggerDetail: "unresponsive-recovery",
triggerSource: "heartbeat-unresponsive",
clearPauseReason: true,
});
} catch (err) {
heartbeatLog.warn(`Error resuming unresponsive agent ${tracked.agentId}: ${err instanceof Error ? err.message : String(err)}`);
}
}
}