FN-7835: auto-clear durable agent error state and retry on next heartbeat
Heartbeat-managed durable agents that land in state:"error" now self-recover on the next heartbeat instead of staying stuck until an operator intervenes. - HeartbeatTriggerScheduler keeps timers armed for durable heartbeat-managed agents in error state when the last error is transient and not operator-actionable (credential/quota/model-access/permanent-config failures stay parked). - executeHeartbeat clears recoverable errors at run entry (error → active, clears lastError), bounded by MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS (settings-overridable); a successful run resets the counter. - On budget exhaustion, the agent is parked paused with pauseReason:"error-retry-exhausted". - Emits new run-audit events agent:auto-recover-error-state and agent:error-retry-exhausted (added to DatabaseMutationType). - Adds heartbeat-error-recovery.test.ts and extends heartbeat-scheduler.test.ts to cover the recovery/exhaustion paths. - Adds changeset and documents the new behavior in AGENTS.md and docs/architecture.md. Files changed: .changeset/fn-7835-agent-error-auto-recovery.md | 7 + AGENTS.md | 1 + docs/architecture.md | 2 + .../src/__tests__/heartbeat-error-recovery.test.ts | 323 +++++++++++++++++++++ .../src/__tests__/heartbeat-scheduler.test.ts | 89 +++++- packages/engine/src/agent-heartbeat.ts | 209 ++++++++++++- packages/engine/src/run-audit.ts | 2 + 7 files changed, 618 insertions(+), 15 deletions(-) Fusion-Task-Id: FN-7835 Fusion-Task-Lineage: 1bbb28a3-8eb9-40e3-8177-6658ec5dae40 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-7835-agent-error-auto-recovery.md
Normal file
7
.changeset/fn-7835-agent-error-auto-recovery.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Agents now auto-clear error state and retry on their next heartbeat instead of getting stuck.
|
||||
category: fix
|
||||
dev: Heartbeat scheduler keeps transient, non-operator-actionable error-state durable agents timer-eligible; executeHeartbeat clears error (error→active, clears lastError) at run entry, bounded by MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS (settings-overridable). Operator-actionable errors stay parked; exhaustion pauses the agent with pauseReason "error-retry-exhausted"; a successful run resets the counter. Emits agent:auto-recover-error-state / agent:error-retry-exhausted run-audit events.
|
||||
@@ -224,6 +224,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
|
||||
- FN-7158: agent performance reflections emit `reflection:generated`, `reflection:skipped`, and `reflection:failed` with ids/counts/outcomes-only metadata; never persist reflection prose or prompt text in run-audit.
|
||||
- FN-7528: a deterministic, non-LLM post-task performance capture (`AgentReflectionService.captureTaskPerformance`) runs once per completed task and emits `reflection:captured` with ids/counts/outcomes-only metadata (`retryReworkCount?`, `filesTouchedCount?`, `packagesTouchedCount?`, `verificationFileScoped?`, `durationMs?`); never persists `verificationScopeReason` free-text or summary prose in run-audit.
|
||||
- FN-7787: `createResolvedAgentSession` enriches `session:runtime-resolved` with `noModelResolved: true` and `runtimeBuiltInFallbackModel` when a non-mock/non-test session reaches runtime creation without a complete provider/model pair; this is a visibility signal for runtime built-in fallback usage, not a fabricated model-resolution verdict.
|
||||
- FN-7835: heartbeat error-state recovery emits `agent:auto-recover-error-state` when a durable heartbeat-managed agent with transient, non-operator-actionable `lastError` clears `error` and retries on the next heartbeat; metadata stays ids/counts/outcomes-only (`agentId`, attempt, limit, source). It emits `agent:error-retry-exhausted` when the bounded recovery budget is exhausted and the agent is parked `paused` with `pauseReason:"error-retry-exhausted"`. Operator-actionable errors remain parked for human repair.
|
||||
- FN-7802: self-healing emits `task:reconcile-missing-worktree-merge-active` when it proves an `in-review` merge-active task (`merging`/`merging-pr`/`merging-fix`) is stranded by an unusable-worktree session-start failure, clears stale `worktree`/`branch`/`sessionFile`, resets the worktree-session retry budget, increments `recoveryRetryCount` as the bounded stale-metadata clear counter, and requeues to `todo`; it emits `task:reconcile-missing-worktree-merge-active-no-action` when `autoMerge:false`, workspace-task ownership, or triple-proof blocks the backward move.
|
||||
- FN-7011: self-healing emits `task:reconcile-engine-downtime-active-timing` when startup recovery shifts active task segment anchors to exclude proven engine-process downtime, and `task:reconcile-engine-downtime-active-timing-no-action` when no active task qualifies.
|
||||
- FN-5419: git run-audit now includes `pull:fast-forward` and `stash:pop-conflict`; dashboard git surfaces now include the extended `POST /api/git/pull` integration-worktree path plus companion `POST /api/git/stash-resolve`, `POST /api/git/stash-drop`, and `POST /api/git/stash-apply` routes.
|
||||
|
||||
@@ -675,6 +675,7 @@ Runtime action-gate flow (v1):
|
||||
- `StuckTaskDetector` (`stuck-task-detector.ts`) — inactivity/loop stall detection
|
||||
- `GridlockDetector` (`gridlock-detector.ts`) — detects all-blocked todo pipelines and emits notification events (plus explicit clear signals when gridlock resolves)
|
||||
- `TransientErrorDetector` (`transient-error-detector.ts`) — retriable error classification
|
||||
- Durable agent heartbeat recovery (FN-7835): a heartbeat-managed, runtime-enabled non-ephemeral agent that lands in `state:"error"` remains timer-eligible and clears `lastError` by transitioning `error → active` at the next heartbeat run entry. Recovery is bounded by `MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS` (settings-overridable through the engine's optional cast-based knob); success resets the metadata counter, while budget exhaustion parks the agent `paused` with `pauseReason:"error-retry-exhausted"` and emits `agent:error-retry-exhausted`.
|
||||
- `SelfHealingManager` (`self-healing.ts`) — auto-unpause/maintenance recovery actions
|
||||
- Batch 1 maintenance now includes `reconcile-orphaned-task-dirs` (FN-6783), a paused-safe housekeeping step that calls `TaskStore.reconcileOrphanedTaskDirs()` so valid live `.fusion/tasks/{ID}/task.json` records missing from the SQLite index become visible without waiting for process restart. The store-level guard skips any ID already present in active, soft-deleted, archived, or tombstoned storage and emits `task:reconcile-orphaned-task-dir` only for recovered rows.
|
||||
- Batch 1 maintenance also includes `reconcile-phantom-committed-reservations` (FN-7069), which calls `TaskStore.reconcilePhantomCommittedReservations()` for committed task-ID reservations that have no live/soft-deleted/archived task row and no `.fusion/tasks/{ID}/task.json`. The sweep prunes orphaned `activityLog` rows and `agents`/cascaded `agentRuns`, preserves `runAuditEvents`, and keeps the reservation `committed` per FN-5105 so the ID is permanently reserved rather than resurrected or handed out again.
|
||||
@@ -2158,6 +2159,7 @@ This section preserves the detailed lifecycle/self-healing contracts that were f
|
||||
- **Completion fan-out is synchronous**: `SelfHealingManager.reconcileCompletedTask()` runs on `in-review → done`. Downstream stale `blockedBy` links and residual `fusion/<task-id>` branch/worktree artifacts are reconciled immediately, not on a periodic sweep.
|
||||
- **In-review stall deadlock**: identical stalls (same code + reason) repeated past `inReviewStallDeadlockThreshold` (default 3) auto-pause with `pausedReason: "in-review-stall-deadlock"` and `status: "failed"`. User-initiated retry paths (dashboard retry, `fn_task_retry`, and CLI `task retry`) clear that automatic deadlock pause so the retry can execute, but they never override explicit/manual pauses or unrelated automatic pause reasons.
|
||||
- **Restart recovery**: `RestartRecoveryCoordinator` classifies interrupted `in-progress` runs. Unusable-worktree session-start failures (`missing`, `incomplete`, `unregistered git worktree`) are recoverable; retries are capped at `MAX_WORKTREE_SESSION_RETRIES=3` before escalating. `recoverMissingWorktreeReviewFailures` also owns durable unusable-worktree session-start failures that reached `in-review` with a merge-active status (`merging`, `merging-pr`, or `merging-fix`): before interrupted/deadlocked merge sweeps can re-drive the same phantom path, it applies auto-merge eligibility, workspace-task exclusion, and triple-proof, clears stale `worktree`/`branch`/`sessionFile`, resets the exhausted worktree-session retry budget, increments `recoveryRetryCount` as the bounded merge-active stale-metadata clear counter, and requeues to `todo` preserving progress. Operator retry surfaces (`fn_task_retry`, CLI `task retry`, dashboard retry) have the same signature-only reset primitive so a missing-worktree failure does not require a valid `merging` transition.
|
||||
- **Durable agent heartbeat error recovery (FN-7835)**: `HeartbeatTriggerScheduler` keeps timers armed for durable heartbeat-managed agents in `state:"error"` only when `lastError` classifies as transient and is not operator-actionable (credential, quota, model-access, or permanent configuration failures remain parked). `HeartbeatMonitor.executeHeartbeat()` clears recoverable errors at run entry (`error → active`, clears `lastError`), increments a consecutive `metadata.heartbeatErrorRecovery` counter, and emits `agent:auto-recover-error-state`; success resets the counter. Once the bounded budget (`MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS`, settings-overridable) is exhausted, the agent is parked `paused` with `pauseReason:"error-retry-exhausted"` and `agent:error-retry-exhausted` is emitted.
|
||||
- **Executor pre-session liveness gate (FN-4935/FN-6861)**: the gate now skips for fresh acquisitions (`acquisition.source === "fresh"`), emits structured `not_usable_task_worktree:<classification>` diagnostics (including canonicalized registered-path snapshots) and a `worktree:incomplete-detected` audit event with `source: "executor-liveness-gate"`, while preserving the existing `taskDoneRetryCount` / `MAX_TASK_DONE_REQUEUE_RETRIES` requeue contract. The project repo root is never a usable task worktree even though it is a legitimately registered Git worktree; `classifyTaskWorktree` returns `repo-root` for canonical root-equal paths, and resume acquisition treats that as self-healable stale metadata by clearing `task.worktree` and creating a fresh checkout under the configured worktrees directory. FN-5772 adds a bounded nested-root self-heal: when `task.worktree` points at a strict descendant of a registered worktree root inside the configured worktrees dir, executor re-anchors `task.worktree` to the git top-level, emits `worktree:reanchored` (`fromPath`, `toPath`, `source`), and proceeds; repo-root/outside-dir/unregistered top-level mismatches still fail. FN-4651 `worktreeSessionRetryCount` remains scoped to the in-review/session-start recovery path.
|
||||
- **Stale self-owned active-session reconcile on conflict cleanup (FN-4973)**: when executor worktree-conflict cleanup finds only a same-task stale `activeSessionRegistry` entry and no live in-memory `activeWorktrees` binding for that task/path, it must unregister the stale entry before `removeWorktree` (plus one-shot backstop reconcile on same-task `ActiveSessionWorktreeRemovalError` races). Foreign-task entries remain protected by FN-4811 and must never be reconciled by the requesting task.
|
||||
- **Same-task stale removal canonical helper (FN-5346)**: executor same-task cleanup paths now route pre-removal reconciliation through `reconcileSelfOwnedActiveSessionForRemoval` (via executor helper wiring), so stale self-owned `activeSessionRegistry` residues are cleared only when no live in-memory binding exists, while FN-4811 foreign-owner refusals and live-owner protections remain intact.
|
||||
|
||||
323
packages/engine/src/__tests__/heartbeat-error-recovery.test.ts
Normal file
323
packages/engine/src/__tests__/heartbeat-error-recovery.test.ts
Normal file
@@ -0,0 +1,323 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
import type { Agent, AgentHeartbeatRun, AgentStore, TaskStore } from "@fusion/core";
|
||||
import { createBudgetStatus } from "./heartbeat-test-helpers.js";
|
||||
|
||||
vi.mock("../logger.js", async () => {
|
||||
const { createMockLogger, formatMockError } = await import("./heartbeat-test-helpers.js");
|
||||
return {
|
||||
createLogger: vi.fn(() => createMockLogger()),
|
||||
heartbeatLog: createMockLogger(),
|
||||
formatError: formatMockError,
|
||||
};
|
||||
});
|
||||
|
||||
vi.mock("../pi.js", () => ({
|
||||
createFnAgent: vi.fn(),
|
||||
describeModel: vi.fn().mockReturnValue("mock-provider/mock-model"),
|
||||
promptWithFallback: vi.fn(async (session: { prompt: (prompt: string) => Promise<void> }, prompt: string) => {
|
||||
await session.prompt(prompt);
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock("../agent-session-helpers.js", async () => {
|
||||
const actual = await vi.importActual<typeof import("../agent-session-helpers.js")>("../agent-session-helpers.js");
|
||||
const pi = await import("../pi.js");
|
||||
return {
|
||||
...actual,
|
||||
createResolvedAgentSession: vi.fn(async () => ({
|
||||
session: await pi.createFnAgent(),
|
||||
sessionFile: undefined,
|
||||
runtimeId: "mock",
|
||||
wasConfigured: true,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
import { createFnAgent } from "../pi.js";
|
||||
import {
|
||||
buildHeartbeatErrorRecoveryMetadata,
|
||||
HEARTBEAT_ERROR_RECOVERY_METADATA_KEY,
|
||||
HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON,
|
||||
HeartbeatMonitor,
|
||||
HeartbeatTriggerScheduler,
|
||||
incrementHeartbeatErrorRecoveryMetadata,
|
||||
isErrorRecoveryEligible,
|
||||
readHeartbeatErrorRetryCount,
|
||||
resetHeartbeatErrorRecoveryMetadata,
|
||||
resolveErrorRecoveryLimit,
|
||||
} from "../agent-heartbeat.js";
|
||||
|
||||
const mockedCreateFnAgent = vi.mocked(createFnAgent);
|
||||
|
||||
const baseAgent = (patch: Partial<Agent> = {}): Agent => ({
|
||||
id: "agent-recovery",
|
||||
name: "Recovery Agent",
|
||||
role: "executor",
|
||||
state: "error",
|
||||
soul: "Keeps durable heartbeat agents healthy",
|
||||
createdAt: "2026-07-11T00:00:00.000Z",
|
||||
updatedAt: "2026-07-11T00:00:00.000Z",
|
||||
metadata: {},
|
||||
runtimeConfig: { enabled: true },
|
||||
...patch,
|
||||
}) as Agent;
|
||||
|
||||
function createNoTaskStore(settings: Record<string, unknown> = {}): TaskStore {
|
||||
return {
|
||||
getSettings: vi.fn().mockResolvedValue(settings),
|
||||
selectNextTaskForAgent: vi.fn().mockResolvedValue(null),
|
||||
listTasks: vi.fn().mockResolvedValue([]),
|
||||
getTaskDocuments: vi.fn().mockResolvedValue([]),
|
||||
recordRunAuditEvent: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
function createAgentStore(agent: Agent): AgentStore & { agent: Agent; runs: Map<string, AgentHeartbeatRun> } {
|
||||
let runSeq = 0;
|
||||
const runs = new Map<string, AgentHeartbeatRun>();
|
||||
const store = {
|
||||
agent,
|
||||
runs,
|
||||
recordHeartbeat: vi.fn().mockResolvedValue(undefined),
|
||||
getAgent: vi.fn(async () => store.agent),
|
||||
getCachedAgent: vi.fn(() => store.agent),
|
||||
listAgents: vi.fn(async () => [store.agent]),
|
||||
on: vi.fn(),
|
||||
off: vi.fn(),
|
||||
updateAgentState: vi.fn(async (_agentId: string, state: Agent["state"]) => {
|
||||
store.agent = { ...store.agent, state };
|
||||
return store.agent;
|
||||
}),
|
||||
updateAgent: vi.fn(async (_agentId: string, updates: Partial<Agent>) => {
|
||||
store.agent = { ...store.agent, ...updates };
|
||||
return store.agent;
|
||||
}),
|
||||
getBudgetStatus: vi.fn().mockResolvedValue(createBudgetStatus()),
|
||||
startHeartbeatRun: vi.fn(async () => {
|
||||
runSeq += 1;
|
||||
const run = {
|
||||
id: `run-${runSeq}`,
|
||||
agentId: store.agent.id,
|
||||
source: "timer",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
} as AgentHeartbeatRun;
|
||||
runs.set(run.id, run);
|
||||
return run;
|
||||
}),
|
||||
saveRun: vi.fn(async (run: AgentHeartbeatRun) => {
|
||||
runs.set(run.id, run);
|
||||
}),
|
||||
getRunDetail: vi.fn(async (_agentId: string, runId: string) => runs.get(runId) ?? null),
|
||||
endHeartbeatRun: vi.fn(async (_runId: string) => undefined),
|
||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||
getActiveHeartbeatRun: vi.fn().mockResolvedValue(null),
|
||||
getRecentRuns: vi.fn().mockResolvedValue([]),
|
||||
getRatingSummary: vi.fn().mockResolvedValue(undefined),
|
||||
claimTaskForAgent: vi.fn().mockResolvedValue({ ok: false, reason: "task_not_found" }),
|
||||
assignTask: vi.fn().mockResolvedValue(agent),
|
||||
syncExecutionTaskLink: vi.fn().mockResolvedValue(undefined),
|
||||
getAgentsByReportsTo: vi.fn().mockResolvedValue([]),
|
||||
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as AgentStore & { agent: Agent; runs: Map<string, AgentHeartbeatRun> };
|
||||
return store;
|
||||
}
|
||||
|
||||
function createSession(promptImpl: () => Promise<void>) {
|
||||
return {
|
||||
prompt: vi.fn(promptImpl),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
model: { provider: "mock", id: "mock-model" },
|
||||
};
|
||||
}
|
||||
|
||||
describe("heartbeat error-recovery primitives", () => {
|
||||
it("resolves a minimum bounded retry limit from optional settings", () => {
|
||||
expect(resolveErrorRecoveryLimit(undefined)).toBe(5);
|
||||
expect(resolveErrorRecoveryLimit({ heartbeatErrorRecoveryAttempts: 3 } as never)).toBe(3);
|
||||
expect(resolveErrorRecoveryLimit({ heartbeatErrorRecoveryAttempts: 0 } as never)).toBe(1);
|
||||
expect(resolveErrorRecoveryLimit({ heartbeatErrorRecoveryAttempts: Number.NaN } as never)).toBe(5);
|
||||
});
|
||||
|
||||
it("reads, increments, and resets the counter without clobbering unrelated metadata", () => {
|
||||
const agent = baseAgent({ metadata: { heartbeatTimerRepair: { repairedAt: "now" } } });
|
||||
expect(readHeartbeatErrorRetryCount(agent)).toBe(0);
|
||||
|
||||
const incremented = incrementHeartbeatErrorRecoveryMetadata(agent);
|
||||
expect(incremented.heartbeatTimerRepair).toEqual({ repairedAt: "now" });
|
||||
expect(readHeartbeatErrorRetryCount({ metadata: incremented })).toBe(1);
|
||||
|
||||
const forced = buildHeartbeatErrorRecoveryMetadata({ metadata: incremented }, 4);
|
||||
expect(readHeartbeatErrorRetryCount({ metadata: forced })).toBe(4);
|
||||
|
||||
const reset = resetHeartbeatErrorRecoveryMetadata({ metadata: forced });
|
||||
expect(reset.heartbeatTimerRepair).toEqual({ repairedAt: "now" });
|
||||
expect(readHeartbeatErrorRetryCount({ metadata: reset })).toBe(0);
|
||||
expect(reset[HEARTBEAT_ERROR_RECOVERY_METADATA_KEY]).toMatchObject({ consecutiveAttempts: 0 });
|
||||
});
|
||||
|
||||
it("only marks durable runtime-enabled under-budget transient error agents eligible", () => {
|
||||
expect(isErrorRecoveryEligible(baseAgent({ lastError: "socket hang up" }), 5)).toBe(true);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ state: "active", lastError: "socket hang up" }), 5)).toBe(false);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ runtimeConfig: { enabled: false }, lastError: "socket hang up" }), 5)).toBe(false);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ metadata: { agentKind: "task-worker" }, lastError: "socket hang up" }), 5)).toBe(false);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ metadata: buildHeartbeatErrorRecoveryMetadata(baseAgent(), 5), lastError: "socket hang up" }), 5)).toBe(false);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ lastError: "invalid api key" }), 5)).toBe(false);
|
||||
expect(isErrorRecoveryEligible(baseAgent({ lastError: "SyntaxError: Unexpected token" }), 5)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("HeartbeatMonitor error-state recovery", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
mockedCreateFnAgent.mockReset();
|
||||
});
|
||||
|
||||
it("reproduces a failed run, then clears error state and retries on the next heartbeat", async () => {
|
||||
const firstSession = createSession(async () => { throw new Error("socket hang up"); });
|
||||
const secondSession = createSession(async () => undefined);
|
||||
mockedCreateFnAgent.mockResolvedValueOnce(firstSession as never).mockResolvedValueOnce(secondSession as never);
|
||||
const store = createAgentStore(baseAgent({ state: "active", lastError: undefined }));
|
||||
const taskStore = createNoTaskStore();
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: process.cwd() });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
expect(store.agent.state).toBe("error");
|
||||
expect(store.agent.lastError).toContain("socket hang up");
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
|
||||
expect(secondSession.prompt).toHaveBeenCalledTimes(1);
|
||||
expect(store.agent.state).toBe("active");
|
||||
expect(store.agent.lastError).toBeUndefined();
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(0);
|
||||
expect(taskStore.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
mutationType: "agent:auto-recover-error-state",
|
||||
target: store.agent.id,
|
||||
metadata: expect.objectContaining({ attempt: 1, limit: 5 }),
|
||||
}));
|
||||
});
|
||||
|
||||
it("does not auto-recover operator-actionable error state", async () => {
|
||||
const session = createSession(async () => undefined);
|
||||
mockedCreateFnAgent.mockResolvedValueOnce(session as never);
|
||||
const store = createAgentStore(baseAgent({ lastError: "invalid api key" }));
|
||||
const taskStore = createNoTaskStore();
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: process.cwd() });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
|
||||
expect(session.prompt).not.toHaveBeenCalled();
|
||||
expect(store.agent.state).toBe("error");
|
||||
expect(store.agent.lastError).toBe("invalid api key");
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(0);
|
||||
expect(taskStore.recordRunAuditEvent).not.toHaveBeenCalledWith(expect.objectContaining({
|
||||
mutationType: "agent:auto-recover-error-state",
|
||||
}));
|
||||
});
|
||||
|
||||
it("parks the agent paused after the bounded recovery budget is exhausted", async () => {
|
||||
mockedCreateFnAgent
|
||||
.mockResolvedValueOnce(createSession(async () => { throw new Error("socket hang up 1"); }) as never)
|
||||
.mockResolvedValueOnce(createSession(async () => { throw new Error("socket hang up 2"); }) as never);
|
||||
const store = createAgentStore(baseAgent({ metadata: {}, lastError: "socket hang up" }));
|
||||
const taskStore = createNoTaskStore({ heartbeatErrorRecoveryAttempts: 2 });
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: process.cwd() });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
expect(store.agent.state).toBe("error");
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(1);
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
|
||||
expect(store.agent.state).toBe("paused");
|
||||
expect(store.agent.pauseReason).toBe(HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON);
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(2);
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(2);
|
||||
expect(taskStore.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
mutationType: "agent:error-retry-exhausted",
|
||||
target: store.agent.id,
|
||||
metadata: expect.objectContaining({ attempts: 2, limit: 2 }),
|
||||
}));
|
||||
});
|
||||
|
||||
it("parks an exhausted agent through the real timer scheduler path", async () => {
|
||||
vi.useFakeTimers();
|
||||
let scheduler: HeartbeatTriggerScheduler | undefined;
|
||||
try {
|
||||
mockedCreateFnAgent
|
||||
.mockResolvedValueOnce(createSession(async () => { throw new Error("socket hang up 1"); }) as never)
|
||||
.mockResolvedValueOnce(createSession(async () => { throw new Error("socket hang up 2"); }) as never);
|
||||
const store = createAgentStore(baseAgent({
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 },
|
||||
lastError: "socket hang up",
|
||||
}));
|
||||
const taskStore = createNoTaskStore({ heartbeatErrorRecoveryAttempts: 2 });
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: process.cwd() });
|
||||
const triggerPromises: Array<Promise<unknown>> = [];
|
||||
scheduler = new HeartbeatTriggerScheduler(
|
||||
store,
|
||||
(agentId, source) => {
|
||||
const run = monitor.executeHeartbeat({ agentId, source });
|
||||
triggerPromises.push(run);
|
||||
return run;
|
||||
},
|
||||
taskStore,
|
||||
);
|
||||
|
||||
scheduler.start();
|
||||
scheduler.registerAgent(store.agent.id, store.agent.runtimeConfig!);
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
await Promise.allSettled(triggerPromises.splice(0));
|
||||
expect(store.agent.state).toBe("error");
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(1);
|
||||
expect(scheduler.getRegisteredAgents()).toContain(store.agent.id);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
await Promise.allSettled(triggerPromises.splice(0));
|
||||
expect(store.agent.state).toBe("paused");
|
||||
expect(store.agent.pauseReason).toBe(HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON);
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(2);
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(2);
|
||||
expect(taskStore.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
mutationType: "agent:error-retry-exhausted",
|
||||
target: store.agent.id,
|
||||
metadata: expect.objectContaining({ attempts: 2, limit: 2 }),
|
||||
}));
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
await Promise.allSettled(triggerPromises.splice(0));
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(2);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(store.agent.id);
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
await Promise.allSettled(triggerPromises.splice(0));
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(2);
|
||||
scheduler.stop();
|
||||
} finally {
|
||||
scheduler?.stop();
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("resets the recovery budget after a successful run", async () => {
|
||||
const session = createSession(async () => undefined);
|
||||
mockedCreateFnAgent.mockResolvedValueOnce(session as never);
|
||||
const store = createAgentStore(baseAgent({
|
||||
lastError: "socket hang up",
|
||||
metadata: buildHeartbeatErrorRecoveryMetadata(baseAgent(), 3),
|
||||
}));
|
||||
const taskStore = createNoTaskStore();
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: process.cwd() });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: store.agent.id, source: "timer" });
|
||||
|
||||
expect(session.prompt).toHaveBeenCalledTimes(1);
|
||||
expect(store.agent.state).toBe("active");
|
||||
expect(store.agent.lastError).toBeUndefined();
|
||||
expect(readHeartbeatErrorRetryCount(store.agent)).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -99,6 +99,7 @@ describe("HeartbeatTriggerScheduler", () => {
|
||||
runtimeConfig: patch.runtimeConfig,
|
||||
lastHeartbeatAt: patch.lastHeartbeatAt,
|
||||
taskId: patch.taskId,
|
||||
lastError: patch.lastError,
|
||||
}) as Agent;
|
||||
|
||||
function createLifecycleStore(initialAgents: Agent[] = []): LifecycleStore {
|
||||
@@ -163,6 +164,50 @@ describe("HeartbeatTriggerScheduler", () => {
|
||||
expect(callback).toHaveBeenCalledWith(explicitAgent.id, "timer", expect.objectContaining({ intervalMs: 15_000 }));
|
||||
});
|
||||
|
||||
it("keeps recoverable error agents timer-eligible and dispatches their next tick", async () => {
|
||||
const recoverable = baseAgent("agent-recoverable", {
|
||||
state: "error",
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 },
|
||||
lastError: "socket hang up",
|
||||
});
|
||||
const exhausted = baseAgent("agent-exhausted", {
|
||||
state: "error",
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 },
|
||||
lastError: "socket hang up",
|
||||
metadata: { heartbeatErrorRecovery: { consecutiveAttempts: 5 } },
|
||||
});
|
||||
const disabled = baseAgent("agent-disabled-error", {
|
||||
state: "error",
|
||||
runtimeConfig: { enabled: false, heartbeatIntervalMs: 1_000 },
|
||||
});
|
||||
const ephemeral = baseAgent("agent-ephemeral-error", {
|
||||
state: "error",
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 },
|
||||
metadata: { agentKind: "task-worker" },
|
||||
});
|
||||
const operatorActionable = baseAgent("agent-operator-error", {
|
||||
state: "error",
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 },
|
||||
lastError: "invalid api key",
|
||||
});
|
||||
const eventStore = createLifecycleStore([recoverable, exhausted, disabled, ephemeral, operatorActionable]);
|
||||
scheduler = new HeartbeatTriggerScheduler(eventStore as unknown as AgentStore, callback);
|
||||
scheduler.start();
|
||||
|
||||
for (const agent of [recoverable, exhausted, disabled, ephemeral, operatorActionable]) {
|
||||
eventStore.emit("agent:created", agent);
|
||||
}
|
||||
|
||||
expect(scheduler.getRegisteredAgents()).toContain(recoverable.id);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(exhausted.id);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(disabled.id);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(ephemeral.id);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(operatorActionable.id);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
expect(callback).toHaveBeenCalledWith(recoverable.id, "timer", expect.objectContaining({ intervalMs: 1_000 }));
|
||||
});
|
||||
|
||||
it("keeps unrelated updates stable, re-arms interval changes, and clears paused timers", async () => {
|
||||
const agent = baseAgent("agent-lifecycle", { runtimeConfig: { enabled: true, heartbeatIntervalMs: 1_000 } });
|
||||
const eventStore = createLifecycleStore([agent]);
|
||||
@@ -326,6 +371,36 @@ describe("HeartbeatTriggerScheduler", () => {
|
||||
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
|
||||
});
|
||||
|
||||
it("re-arms a recoverable error agent when timer entry is missing and no lifecycle event fires", async () => {
|
||||
vi.useFakeTimers();
|
||||
const agent = {
|
||||
id: "agent-error-audit",
|
||||
name: "Agent Error Audit",
|
||||
role: "executor",
|
||||
state: "error",
|
||||
lastHeartbeatAt: "2026-01-01T00:00:00.000Z",
|
||||
runtimeConfig: { enabled: true, heartbeatIntervalMs: 30_000 },
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
lastError: "socket hang up",
|
||||
metadata: { heartbeatErrorRecovery: { consecutiveAttempts: 1 } },
|
||||
} as Agent;
|
||||
vi.mocked(store.listAgents).mockResolvedValue([agent]);
|
||||
vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null);
|
||||
|
||||
scheduler = new HeartbeatTriggerScheduler(store, callback);
|
||||
scheduler.start();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(scheduler.getRegisteredAgents()).toContain(agent.id);
|
||||
scheduler.unregisterAgent(agent.id);
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain(agent.id);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(60_000);
|
||||
|
||||
expect(scheduler.getRegisteredAgents()).toContain(agent.id);
|
||||
});
|
||||
|
||||
it("marks repaired agent metadata as stale when last heartbeat exceeds the default 2x threshold", async () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-01-01T02:00:00.000Z"));
|
||||
@@ -1935,11 +2010,11 @@ describe("HeartbeatTriggerScheduler", () => {
|
||||
expect(callback).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("timer is unregistered when agent becomes error state (should clear timer)", async () => {
|
||||
it("timer remains registered when agent becomes recoverable error state", async () => {
|
||||
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 5000 });
|
||||
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
|
||||
|
||||
// Update to error state
|
||||
// Update to a recoverable durable error state.
|
||||
(eventStore.getAgent as ReturnType<typeof vi.fn>).mockImplementation((agentId: string) => ({
|
||||
id: agentId,
|
||||
name: `Agent ${agentId}`,
|
||||
@@ -1947,15 +2022,15 @@ describe("HeartbeatTriggerScheduler", () => {
|
||||
state: "error" as const,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
lastError: "socket hang up",
|
||||
metadata: {},
|
||||
}));
|
||||
eventStore.emit("agent:updated", { id: "agent-001", state: "error", metadata: {} } as import("@fusion/core").Agent);
|
||||
eventStore.emit("agent:updated", { id: "agent-001", state: "error", lastError: "socket hang up", metadata: {} } as import("@fusion/core").Agent);
|
||||
|
||||
// Timer should be cleared for error agents
|
||||
expect(scheduler.getRegisteredAgents()).not.toContain("agent-001");
|
||||
expect(scheduler.getRegisteredAgents()).toContain("agent-001");
|
||||
|
||||
await vi.advanceTimersByTimeAsync(10000);
|
||||
expect(callback).not.toHaveBeenCalled();
|
||||
await vi.advanceTimersByTimeAsync(5000);
|
||||
expect(callback).toHaveBeenCalledWith("agent-001", "timer", expect.objectContaining({ intervalMs: 5000 }));
|
||||
});
|
||||
|
||||
it("timer is unregistered when agent becomes paused state (should clear timer)", async () => {
|
||||
|
||||
@@ -34,6 +34,7 @@ import { resolveHeartbeatPromptTemplate, resolveHeartbeatScopeDisciplineMode, se
|
||||
import { buildPromptLayers, collapsePromptLayers } from "./prompt-layers.js";
|
||||
import { resolveAndEmitGoalContext } from "./goal-injection-diagnostics.js";
|
||||
import { createLogger, heartbeatLog, formatError } from "./logger.js";
|
||||
import { classifyError, isOperatorActionableAgentError } from "./transient-error-detector.js";
|
||||
|
||||
/**
|
||||
* FNXC:WorktreeAcquisition 2026-07-09-00:00:
|
||||
@@ -51,6 +52,17 @@ import { createLogger, heartbeatLog, formatError } from "./logger.js";
|
||||
* as the cross-heartbeat counter.
|
||||
*/
|
||||
const MAX_HEARTBEAT_WORKTREE_ACQUISITION_RETRIES = 3;
|
||||
|
||||
/*
|
||||
FNXC:HeartbeatRecovery 2026-07-11-00:00:
|
||||
FN-7835 requires durable heartbeat-managed agents in error state to retry on their next heartbeat instead of staying stranded. Recovery is intentionally bounded by a consecutive-attempt budget so persistent failures park the agent instead of forming an infinite retry loop.
|
||||
|
||||
FNXC:HeartbeatRecovery 2026-07-11-00:00:
|
||||
FN-7672 requires durable agent error recovery to stay classification-gated: only transient, non-operator-actionable lastError values may be retried automatically. Credential, quota, model-access, and permanent configuration failures must remain parked for operator action instead of burning heartbeat retries.
|
||||
*/
|
||||
export const MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS = 5;
|
||||
export const HEARTBEAT_ERROR_RECOVERY_METADATA_KEY = "heartbeatErrorRecovery";
|
||||
export const HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON = "error-retry-exhausted";
|
||||
import { acquireTaskWorktree } from "./worktree-acquisition.js";
|
||||
import { createRunAuditor, generateSyntheticRunId, type DatabaseMutationType, type EngineRunContext } from "./run-audit.js";
|
||||
import { promptWithFallback } from "./pi.js";
|
||||
@@ -1723,14 +1735,66 @@ export class HeartbeatMonitor {
|
||||
if (!completionResult.skipStateTransition) {
|
||||
try {
|
||||
if (completionResult.status === "failed") {
|
||||
await this.store.updateAgentState(agentId, "error");
|
||||
await this.store.updateAgent(agentId, { lastError: completionResult.stderrExcerpt ?? "Run failed" });
|
||||
const latestAgent = await this.store.getAgent(agentId);
|
||||
const errorRecoveryLimit = this.taskStore
|
||||
? resolveErrorRecoveryLimit(await this.taskStore.getSettings().catch((settingsErr) => {
|
||||
heartbeatLog.warn(`Agent ${agentId} error-recovery limit lookup failed: ${settingsErr instanceof Error ? settingsErr.message : String(settingsErr)} — using default limit`);
|
||||
return undefined;
|
||||
}))
|
||||
: MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS;
|
||||
const retryCount = latestAgent ? readHeartbeatErrorRetryCount(latestAgent) : 0;
|
||||
const failedWithRecoverableError = isHeartbeatErrorRecoverable({ lastError: completionResult.stderrExcerpt ?? "Run failed" });
|
||||
/*
|
||||
FNXC:HeartbeatRecovery 2026-07-11-19:57:
|
||||
FN-7835's primary timer path cannot rely on a future heartbeat to perform exhaustion bookkeeping: once retryCount reaches the limit, timer eligibility intentionally stops dispatching error-state agents. Park the agent paused on the failing boundary run so the bounded retry contract is reachable in production.
|
||||
*/
|
||||
if (
|
||||
latestAgent
|
||||
&& isHeartbeatManaged(latestAgent)
|
||||
&& latestAgent.runtimeConfig?.enabled !== false
|
||||
&& retryCount >= errorRecoveryLimit
|
||||
&& retryCount > 0
|
||||
&& failedWithRecoverableError
|
||||
) {
|
||||
await this.store.updateAgentState(agentId, "paused");
|
||||
await this.store.updateAgent(agentId, {
|
||||
lastError: completionResult.stderrExcerpt ?? "Run failed",
|
||||
pauseReason: HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON,
|
||||
});
|
||||
heartbeatLog.warn(`Agent ${agentId} error recovery exhausted after ${retryCount}/${errorRecoveryLimit} attempts — pausing`);
|
||||
if (this.taskStore) {
|
||||
try {
|
||||
const runWithSource = run as unknown as { source?: unknown };
|
||||
const runSource = typeof runWithSource.source === "string" ? runWithSource.source : undefined;
|
||||
const audit = createRunAuditor(this.taskStore, {
|
||||
runId,
|
||||
agentId,
|
||||
phase: "heartbeat",
|
||||
source: runSource,
|
||||
});
|
||||
await audit.database({
|
||||
type: "agent:error-retry-exhausted",
|
||||
target: agentId,
|
||||
metadata: { agentId, attempts: retryCount, limit: errorRecoveryLimit, source: runSource },
|
||||
});
|
||||
} catch (auditErr) {
|
||||
heartbeatLog.warn(`Agent ${agentId} error-retry exhaustion audit failed: ${auditErr instanceof Error ? auditErr.message : String(auditErr)} — continuing`);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
await this.store.updateAgentState(agentId, "error");
|
||||
await this.store.updateAgent(agentId, { lastError: completionResult.stderrExcerpt ?? "Run failed" });
|
||||
}
|
||||
} else if (completionResult.status === "terminated") {
|
||||
await this.store.updateAgentState(agentId, "paused");
|
||||
} else {
|
||||
// Completed successfully - back to active and clear any stale failure marker.
|
||||
await this.store.updateAgentState(agentId, "active");
|
||||
await this.store.updateAgent(agentId, { lastError: undefined });
|
||||
const latestAgent = await this.store.getAgent(agentId);
|
||||
await this.store.updateAgent(agentId, {
|
||||
lastError: undefined,
|
||||
...(latestAgent ? { metadata: resetHeartbeatErrorRecoveryMetadata(latestAgent) } : {}),
|
||||
});
|
||||
}
|
||||
} catch (stateTransErr) {
|
||||
heartbeatLog.warn(`Agent ${agentId} state transition failed: ${stateTransErr instanceof Error ? stateTransErr.message : String(stateTransErr)} — continuing`);
|
||||
@@ -2174,7 +2238,7 @@ export class HeartbeatMonitor {
|
||||
}
|
||||
|
||||
// Resolve agent
|
||||
const agent = preloadedAgent ?? await this.store.getAgent(agentId);
|
||||
let agent = preloadedAgent ?? await this.store.getAgent(agentId);
|
||||
if (!agent) {
|
||||
heartbeatLog.warn(`Agent ${agentId} not found — completing run as failed`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
@@ -2184,6 +2248,66 @@ export class HeartbeatMonitor {
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
|
||||
if (agent.state === "error") {
|
||||
const errorRecoveryLimit = resolveErrorRecoveryLimit(heartbeatModelSettings);
|
||||
const currentRetryCount = readHeartbeatErrorRetryCount(agent);
|
||||
const canAttemptErrorRecovery = isErrorRecoveryEligible(agent, errorRecoveryLimit);
|
||||
const recoveryBudgetExhausted = isHeartbeatManaged(agent)
|
||||
&& agent.runtimeConfig?.enabled !== false
|
||||
&& isHeartbeatErrorRecoverable(agent)
|
||||
&& currentRetryCount >= errorRecoveryLimit;
|
||||
|
||||
if (canAttemptErrorRecovery) {
|
||||
const attempt = currentRetryCount + 1;
|
||||
const metadata = incrementHeartbeatErrorRecoveryMetadata(agent);
|
||||
try {
|
||||
await this.store.updateAgentState(agentId, "active");
|
||||
await this.store.updateAgent(agentId, { lastError: undefined, metadata });
|
||||
heartbeatLog.log(`Agent ${agentId} auto-recovered from error state for heartbeat retry attempt ${attempt}/${errorRecoveryLimit}`);
|
||||
await audit.database({
|
||||
type: "agent:auto-recover-error-state",
|
||||
target: agentId,
|
||||
metadata: { agentId, attempt, limit: errorRecoveryLimit, source },
|
||||
});
|
||||
agent = (await this.store.getAgent(agentId)) ?? { ...agent, state: "active", lastError: undefined, metadata };
|
||||
} catch (recoveryErr) {
|
||||
heartbeatLog.warn(`Agent ${agentId} error-state recovery bookkeeping failed: ${recoveryErr instanceof Error ? recoveryErr.message : String(recoveryErr)} — continuing with existing state`);
|
||||
}
|
||||
} else if (recoveryBudgetExhausted) {
|
||||
try {
|
||||
await this.store.updateAgentState(agentId, "paused");
|
||||
await this.store.updateAgent(agentId, { pauseReason: HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON });
|
||||
heartbeatLog.warn(`Agent ${agentId} error recovery exhausted after ${currentRetryCount}/${errorRecoveryLimit} attempts — pausing`);
|
||||
await audit.database({
|
||||
type: "agent:error-retry-exhausted",
|
||||
target: agentId,
|
||||
metadata: { agentId, attempts: currentRetryCount, limit: errorRecoveryLimit, source },
|
||||
});
|
||||
} catch (exhaustionErr) {
|
||||
heartbeatLog.warn(`Agent ${agentId} error-retry exhaustion bookkeeping failed: ${exhaustionErr instanceof Error ? exhaustionErr.message : String(exhaustionErr)} — completing run without retry`);
|
||||
}
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
resultJson: { reason: HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON, attempts: currentRetryCount, limit: errorRecoveryLimit },
|
||||
skipStateTransition: true,
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
} else {
|
||||
heartbeatLog.log(`Agent ${agentId} state is "error" but lastError is not eligible for heartbeat recovery — graceful exit`);
|
||||
try {
|
||||
await this.store.updateAgentState(agentId, "error");
|
||||
} catch (restoreErr) {
|
||||
heartbeatLog.warn(`Agent ${agentId} non-recoverable error-state restore failed: ${restoreErr instanceof Error ? restoreErr.message : String(restoreErr)} — preserving run completion`);
|
||||
}
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
resultJson: { reason: "invalid_state", state: agent.state, recoveryEligible: false },
|
||||
skipStateTransition: true,
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
}
|
||||
|
||||
// Check if agent has identity (used later for no-task run decisions)
|
||||
const agentHasIdentity = hasAgentIdentity(agent);
|
||||
const isAgentEphemeral = isEphemeralAgent(agent);
|
||||
@@ -3950,12 +4074,67 @@ function isHeartbeatManaged(agent: Agent): boolean {
|
||||
return !isEphemeralAgent(agent);
|
||||
}
|
||||
|
||||
type HeartbeatErrorRecoveryMetadata = {
|
||||
consecutiveAttempts: number;
|
||||
updatedAt?: string;
|
||||
};
|
||||
|
||||
export function resolveErrorRecoveryLimit(settings: Settings | null | undefined): number {
|
||||
const raw = (settings as { heartbeatErrorRecoveryAttempts?: unknown } | null | undefined)?.heartbeatErrorRecoveryAttempts;
|
||||
if (typeof raw !== "number" || !Number.isFinite(raw)) {
|
||||
return MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS;
|
||||
}
|
||||
return Math.max(1, Math.floor(raw));
|
||||
}
|
||||
|
||||
export function readHeartbeatErrorRetryCount(agent: Pick<Agent, "metadata">): number {
|
||||
const metadata = (agent.metadata ?? {}) as Record<string, unknown>;
|
||||
const raw = metadata[HEARTBEAT_ERROR_RECOVERY_METADATA_KEY];
|
||||
if (!raw || typeof raw !== "object") {
|
||||
return 0;
|
||||
}
|
||||
const candidate = raw as Record<string, unknown>;
|
||||
const count = candidate.consecutiveAttempts;
|
||||
return typeof count === "number" && Number.isFinite(count) && count > 0 ? Math.floor(count) : 0;
|
||||
}
|
||||
|
||||
export function buildHeartbeatErrorRecoveryMetadata(agent: Pick<Agent, "metadata">, consecutiveAttempts: number): Record<string, unknown> {
|
||||
return {
|
||||
...(agent.metadata ?? {}),
|
||||
[HEARTBEAT_ERROR_RECOVERY_METADATA_KEY]: {
|
||||
consecutiveAttempts: Math.max(0, Math.floor(consecutiveAttempts)),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} satisfies HeartbeatErrorRecoveryMetadata,
|
||||
};
|
||||
}
|
||||
|
||||
export function incrementHeartbeatErrorRecoveryMetadata(agent: Pick<Agent, "metadata">): Record<string, unknown> {
|
||||
return buildHeartbeatErrorRecoveryMetadata(agent, readHeartbeatErrorRetryCount(agent) + 1);
|
||||
}
|
||||
|
||||
export function resetHeartbeatErrorRecoveryMetadata(agent: Pick<Agent, "metadata">): Record<string, unknown> {
|
||||
return buildHeartbeatErrorRecoveryMetadata(agent, 0);
|
||||
}
|
||||
|
||||
export function isHeartbeatErrorRecoverable(agent: Pick<Agent, "lastError">): boolean {
|
||||
const lastError = agent.lastError ?? "";
|
||||
return classifyError(lastError) === "transient" && !isOperatorActionableAgentError(lastError);
|
||||
}
|
||||
|
||||
export function isErrorRecoveryEligible(agent: Agent, limit: number): boolean {
|
||||
return agent.state === "error"
|
||||
&& isHeartbeatManaged(agent)
|
||||
&& agent.runtimeConfig?.enabled !== false
|
||||
&& isHeartbeatErrorRecoverable(agent)
|
||||
&& readHeartbeatErrorRetryCount(agent) < Math.max(1, Math.floor(limit));
|
||||
}
|
||||
|
||||
/**
|
||||
* HeartbeatTriggerScheduler manages timer-based heartbeat triggers for agents.
|
||||
*
|
||||
* Timers are armed only for durable agents where all of the following hold:
|
||||
* - `runtimeConfig.enabled !== false`
|
||||
* - `state ∈ {active, running, idle}`
|
||||
* - `state ∈ {active, running, idle}` or `state === "error"` with retry budget remaining
|
||||
*
|
||||
* Any other state, or any ephemeral/task-worker agent, clears the timer.
|
||||
* State changes and heartbeat config updates are observed via AgentStore
|
||||
@@ -3998,6 +4177,7 @@ export class HeartbeatTriggerScheduler {
|
||||
private callback: TriggerCallback;
|
||||
private taskStore?: TaskStore;
|
||||
private timers: Map<string, AgentTimer> = new Map();
|
||||
private errorRecoveryLimit = MAX_HEARTBEAT_ERROR_RECOVERY_ATTEMPTS;
|
||||
private pendingAssignments: Map<string, PendingAssignment> = new Map();
|
||||
private registrationEpochs: Map<string, number> = new Map();
|
||||
private running = false;
|
||||
@@ -4501,10 +4681,15 @@ export class HeartbeatTriggerScheduler {
|
||||
}
|
||||
}
|
||||
|
||||
private updateErrorRecoveryLimit(settings: Settings | null | undefined): number {
|
||||
this.errorRecoveryLimit = resolveErrorRecoveryLimit(settings);
|
||||
return this.errorRecoveryLimit;
|
||||
}
|
||||
|
||||
private isTimerEligibleAgent(agent: Agent): boolean {
|
||||
return isHeartbeatManaged(agent)
|
||||
&& agent.runtimeConfig?.enabled !== false
|
||||
&& isTickableState(agent.state);
|
||||
&& (isTickableState(agent.state) || isErrorRecoveryEligible(agent, this.errorRecoveryLimit));
|
||||
}
|
||||
|
||||
private getAgentTimerConfig(agent: Agent): AgentHeartbeatConfig {
|
||||
@@ -4727,6 +4912,7 @@ export class HeartbeatTriggerScheduler {
|
||||
? await this.taskStore.getSettings()
|
||||
: null;
|
||||
const staleMultiplier = this.resolveRepairStaleMultiplier(settings);
|
||||
this.updateErrorRecoveryLimit(settings);
|
||||
const agents = await this.store.listAgents();
|
||||
let rearmedCount = 0;
|
||||
let zombieRearmedCount = 0;
|
||||
@@ -4851,12 +5037,20 @@ export class HeartbeatTriggerScheduler {
|
||||
this.unregisterAgent(agentId);
|
||||
return;
|
||||
}
|
||||
if (!isHeartbeatManaged(agent) || !isTickableState(agent.state)) {
|
||||
if (!isHeartbeatManaged(agent) || (agent.state !== "error" && !isTickableState(agent.state))) {
|
||||
heartbeatLog.log(`Timer tick skipped for ${agentId} (state=${agent.state})`);
|
||||
this.unregisterAgent(agentId);
|
||||
return;
|
||||
}
|
||||
|
||||
const settings = this.taskStore ? await this.taskStore.getSettings() : null;
|
||||
const errorRecoveryLimit = this.updateErrorRecoveryLimit(settings);
|
||||
if (agent.state === "error" && !isErrorRecoveryEligible(agent, errorRecoveryLimit)) {
|
||||
heartbeatLog.log(`Timer tick skipped for ${agentId} (state=${agent.state}, error recovery ineligible)`);
|
||||
this.unregisterAgent(agentId);
|
||||
return;
|
||||
}
|
||||
|
||||
// Guard: skip timer ticks for idle agents when configured
|
||||
const timerRc = (agent.runtimeConfig ?? {}) as {
|
||||
allowParallelExecution?: boolean;
|
||||
@@ -4885,7 +5079,6 @@ export class HeartbeatTriggerScheduler {
|
||||
|
||||
// Global/engine pause guard: scheduler should not dispatch timer callbacks
|
||||
// while globally paused (hard stop) or engine paused (soft stop for timers).
|
||||
const settings = this.taskStore ? await this.taskStore.getSettings() : null;
|
||||
if (settings?.globalPause) {
|
||||
heartbeatLog.log(`Timer tick skipped for ${agentId} (global pause active)`);
|
||||
return;
|
||||
|
||||
@@ -434,6 +434,8 @@ export type DatabaseMutationType =
|
||||
| "task:steering-comment:add"
|
||||
| "task:assign"
|
||||
| "task:checkout"
|
||||
| "agent:auto-recover-error-state"
|
||||
| "agent:error-retry-exhausted"
|
||||
| "task:release"
|
||||
| "task:pause"
|
||||
| "task:unpause"
|
||||
|
||||
Reference in New Issue
Block a user