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
|
* Find agents in `state="running"` that are not actually running and flip
|
||||||
* flip them to `"active"`. Called on monitor start to clean up orphans
|
* them to `"active"`. An agent is considered orphaned when either:
|
||||||
* left behind by older governance-skip code paths. Best-effort — failures
|
* (a) it has no active heartbeat run record, or
|
||||||
* are logged but do not block startup.
|
* (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> {
|
private async reconcileOrphanedRunningAgents(): Promise<void> {
|
||||||
try {
|
try {
|
||||||
const runningAgents = await this.store.listAgents({ state: "running", includeEphemeral: true });
|
const runningAgents = await this.store.listAgents({ state: "running", includeEphemeral: true });
|
||||||
|
const now = Date.now();
|
||||||
for (const agent of runningAgents) {
|
for (const agent of runningAgents) {
|
||||||
|
let reason: string | null = null;
|
||||||
const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
|
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 {
|
try {
|
||||||
await this.store.updateAgentState(agent.id, "active");
|
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) {
|
} catch (err) {
|
||||||
heartbeatLog.warn(`Failed to reconcile orphaned running agent ${agent.id}: ${err instanceof Error ? err.message : String(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> {
|
private async handleMissedHeartbeat(tracked: TrackedAgent, reason: string): Promise<void> {
|
||||||
@@ -2349,6 +2391,8 @@ export class HeartbeatMonitor {
|
|||||||
|
|
||||||
heartbeatLog.warn(`Recovering unresponsive agent ${tracked.agentId}: ${reason}`);
|
heartbeatLog.warn(`Recovering unresponsive agent ${tracked.agentId}: ${reason}`);
|
||||||
|
|
||||||
|
const runIdToTerminate = tracked.runId;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
tracked.session.dispose();
|
tracked.session.dispose();
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
@@ -2357,6 +2401,20 @@ export class HeartbeatMonitor {
|
|||||||
|
|
||||||
this.untrackAgent(tracked.agentId);
|
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 {
|
try {
|
||||||
await this.pauseAgent(tracked.agentId, { pauseReason: "heartbeat-unresponsive", stopActiveRun: false });
|
await this.pauseAgent(tracked.agentId, { pauseReason: "heartbeat-unresponsive", stopActiveRun: false });
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
|||||||
Reference in New Issue
Block a user