fix(engine): reconcile agents stuck in running state after missed heartbeat
Recovery path now calls completeRun(terminated) so the canonical agent-state transition runs, and reconcileOrphanedRunningAgents both catches stale-heartbeat cases and runs every poll so pre-existing stuck rows self-heal post-upgrade. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
12
.changeset/heartbeat-reconcile-stuck-running.md
Normal file
12
.changeset/heartbeat-reconcile-stuck-running.md
Normal file
@@ -0,0 +1,12 @@
|
||||
---
|
||||
"@fusion/engine": patch
|
||||
---
|
||||
|
||||
Fix agents stuck in `state="running"` after a missed-heartbeat termination.
|
||||
|
||||
The unresponsive-agent recovery path disposed the session and called `pauseAgent`, but never explicitly ended the run via `completeRun` — relying on the in-flight execution to self-complete via its catch handler, which doesn't happen when the run is genuinely hung. The run record could still be terminated through other paths (safety-net or supersede-on-startRun), but those bypass the agent-state transition, leaving the agent permanently displayed as "running" with no active run.
|
||||
|
||||
Two fixes:
|
||||
|
||||
- `recoverUnresponsiveAgent` now calls `completeRun(..., status: "terminated")` so the canonical state transition runs alongside the existing `pauseAgent`/`resumeAgent` sequence.
|
||||
- `reconcileOrphanedRunningAgents` is broadened to also catch agents with stale `lastHeartbeatAt` (> 3× timeout) that aren't in the in-memory tracked set, terminating their stale run record. It now runs every poll instead of only at monitor start, so any pre-existing stuck rows from older versions self-heal within one poll interval after upgrade.
|
||||
@@ -580,20 +580,57 @@ export class HeartbeatMonitor {
|
||||
}
|
||||
|
||||
/**
|
||||
* Find agents in `state="running"` that have no active heartbeat run and
|
||||
* flip them to `"active"`. Called on monitor start to clean up orphans
|
||||
* left behind by older governance-skip code paths. Best-effort — failures
|
||||
* are logged but do not block startup.
|
||||
* Find agents in `state="running"` that are not actually running and flip
|
||||
* them to `"active"`. An agent is considered orphaned when either:
|
||||
* (a) it has no active heartbeat run record, or
|
||||
* (b) it is not in this monitor's in-memory tracked set AND its
|
||||
* lastHeartbeatAt is older than 3× the configured timeout.
|
||||
*
|
||||
* Case (a) covers historical bypass paths (governance-skip, supersede-on-
|
||||
* startRun, safety-net run termination) that ended the run record but
|
||||
* never propagated the agent-state transition. Case (b) covers a process
|
||||
* that crashed mid-run, leaving both the run row and the agent row stuck.
|
||||
*
|
||||
* Called on monitor start AND periodically from the polling loop to keep
|
||||
* the system self-healing across versions. Best-effort — failures are
|
||||
* logged but do not block the caller.
|
||||
*/
|
||||
private async reconcileOrphanedRunningAgents(): Promise<void> {
|
||||
try {
|
||||
const runningAgents = await this.store.listAgents({ state: "running", includeEphemeral: true });
|
||||
const now = Date.now();
|
||||
for (const agent of runningAgents) {
|
||||
let reason: string | null = null;
|
||||
const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
|
||||
if (activeRun) continue;
|
||||
if (!activeRun) {
|
||||
reason = "no active run";
|
||||
} else if (!this.trackedAgents.has(agent.id)) {
|
||||
const timeoutMs = this.resolveAgentConfig(agent.id).heartbeatTimeoutMs;
|
||||
const lastTs = agent.lastHeartbeatAt ? Date.parse(agent.lastHeartbeatAt) : NaN;
|
||||
const heartbeatAgeMs = Number.isFinite(lastTs) ? Math.max(0, now - lastTs) : Infinity;
|
||||
if (heartbeatAgeMs > timeoutMs * 3) {
|
||||
try {
|
||||
const detail = await this.store.getRunDetail(agent.id, activeRun.id);
|
||||
if (detail && detail.status !== "completed" && detail.status !== "failed" && detail.status !== "terminated") {
|
||||
await this.store.saveRun({
|
||||
...detail,
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "terminated",
|
||||
stderrExcerpt: `Reconciled stale run (no heartbeat for ${formatDuration(heartbeatAgeMs)}; threshold ${formatDuration(timeoutMs * 3)})`,
|
||||
});
|
||||
}
|
||||
await this.store.endHeartbeatRun(activeRun.id, "terminated");
|
||||
} catch (runEndErr) {
|
||||
heartbeatLog.warn(`Failed to terminate stale run ${activeRun.id} for ${agent.id}: ${runEndErr instanceof Error ? runEndErr.message : String(runEndErr)}`);
|
||||
}
|
||||
reason = `stale heartbeat (${formatDuration(heartbeatAgeMs)} since lastHeartbeatAt)`;
|
||||
}
|
||||
}
|
||||
if (!reason) continue;
|
||||
try {
|
||||
await this.store.updateAgentState(agent.id, "active");
|
||||
heartbeatLog.log(`Reconciled orphaned running agent ${agent.id} → active (no active run)`);
|
||||
this.clearRunState(agent.id);
|
||||
heartbeatLog.log(`Reconciled orphaned running agent ${agent.id} → active (${reason})`);
|
||||
} catch (err) {
|
||||
heartbeatLog.warn(`Failed to reconcile orphaned running agent ${agent.id}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
@@ -2332,6 +2369,11 @@ export class HeartbeatMonitor {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Periodically scan for orphaned `state="running"` rows so that a single
|
||||
// missed termination can't leave an agent permanently stuck. Cheap query
|
||||
// (indexed by state) so running it every poll is fine.
|
||||
await this.reconcileOrphanedRunningAgents();
|
||||
}
|
||||
|
||||
private async handleMissedHeartbeat(tracked: TrackedAgent, reason: string): Promise<void> {
|
||||
@@ -2349,6 +2391,8 @@ export class HeartbeatMonitor {
|
||||
|
||||
heartbeatLog.warn(`Recovering unresponsive agent ${tracked.agentId}: ${reason}`);
|
||||
|
||||
const runIdToTerminate = tracked.runId;
|
||||
|
||||
try {
|
||||
tracked.session.dispose();
|
||||
} catch (err) {
|
||||
@@ -2357,6 +2401,20 @@ export class HeartbeatMonitor {
|
||||
|
||||
this.untrackAgent(tracked.agentId);
|
||||
|
||||
// Canonically end the run record. Without this, dispose() relies on the
|
||||
// in-flight execution self-completing — which never happens when the run
|
||||
// is actually hung. completeRun also updates agent state, but we still
|
||||
// call pauseAgent below to set `pauseReason="heartbeat-unresponsive"`
|
||||
// and pause assigned tasks. The double state transition is harmless.
|
||||
try {
|
||||
await this.completeRun(tracked.agentId, runIdToTerminate, {
|
||||
status: "terminated",
|
||||
stderrExcerpt: reason,
|
||||
});
|
||||
} catch (err) {
|
||||
heartbeatLog.warn(`completeRun(terminated) failed for ${tracked.agentId}/${runIdToTerminate}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
|
||||
try {
|
||||
await this.pauseAgent(tracked.agentId, { pauseReason: "heartbeat-unresponsive", stopActiveRun: false });
|
||||
} catch (err) {
|
||||
|
||||
Reference in New Issue
Block a user