FN-6336: reattach orphaned assigned executions
Self-healing now resumes stranded assigned in-progress work when an agent has no active run or execution. - Add a self-healing pass that groups stale assigned in-progress tasks by durable agent and calls the executor resume seam without moving tasks backward. - Emit run-audit telemetry for reattached orphaned executions and wire the resume hook through the in-process runtime. - Cover grace windows, pause/active-run/active-execution guards, task filtering, and registration order with engine tests. - Document the recovery behavior and add a patch changeset for the published CLI package. Files changed: .changeset/fn-6336-reattach-orphaned-executions.md | 5 + docs/agents.md | 2 + docs/architecture.md | 2 + ...lf-healing-reattach-orphaned-executions.test.ts | 217 +++++++++++++++++++++ packages/engine/src/run-audit.ts | 1 + packages/engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/self-healing.ts | 122 ++++++++++++ 7 files changed, 350 insertions(+) Fusion-Task-Id: FN-6336 Fusion-Task-Lineage: 6d8519b9-d8b0-4726-9b45-bfa445b7883e
This commit is contained in:
gsxdsm
5
.changeset/fn-6336-reattach-orphaned-executions.md
Normal file
5
.changeset/fn-6336-reattach-orphaned-executions.md
Normal file
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
Self-healing now automatically re-dispatches an assigned in-progress task when its durable agent loses both the heartbeat run and active execution session, preventing the task from stranding until the next engine restart.
|
||||
@@ -530,6 +530,8 @@ The `runtimeConfig` field on agents supports the following options:
|
||||
|
||||
Assignment-triggered heartbeats are completion-resilient: if an `agent:assigned` wake is skipped only because the durable agent already has an active heartbeat run, Fusion records the latest assigned task as a pending assignment and re-fires that assignment wake once the active run completes. This prevents assigned work from being stranded by long heartbeat intervals or `skipHeartbeatWhenIdle`; disabled agents (`enabled === false`) and budget-exhausted agents still do not defer assignment wakes.
|
||||
|
||||
Self-healing also covers abnormal run/session loss for assigned `in-progress` work. If the task remains assigned but the durable agent has no active heartbeat run and no active executor session after the orphan grace window, `reattach-orphaned-assigned-executions` re-dispatches the task forward via `executor.resumeTaskForAgent(agentId)` without pausing, failing, or moving the task backward.
|
||||
|
||||
Heartbeat values are validated and minimum-clamped to 5 minutes (300,000 ms).
|
||||
Project setting `heartbeatMultiplier` (default `1`) scales resolved heartbeat timing globally: both the heartbeat interval (`pollIntervalMs`) and unresponsive timeout base (`heartbeatTimeoutMs`) are multiplied. Per-agent `heartbeatIntervalMs`/`heartbeatTimeoutMs` remain base values before multiplier scaling. This setting is configured from the **Agents** screen's **Controls** popup under "Heartbeat Speed".
|
||||
|
||||
|
||||
@@ -676,6 +676,7 @@ Runtime action-gate flow (v1):
|
||||
- Worktrees-dir sweeps that list direct children of `<worktreesDir>` (pool idle scan, orphan cleanup/reap, self-healing unregistered-orphan reap, and cap enforcement) must exclude the `.ai-merge` container by name; those one-level sweeps never inspect or recycle clean rooms beneath it. Batch 1 sweeps stale AI merge clean-room worktrees under the new `<worktreesDir>/.ai-merge/` root and still scans legacy `.fusion/ai-merge/` plus legacy `tmpdir()` locations for pre-relocation leftovers; candidates are bounded to names starting with `fusion-ai-merge-`. `runAiMerge` registers each live clean-room worktree in `activeSessionRegistry` with kind `ai-merge` as soon as the directory exists and keeps both raw and canonical paths registered for the duration of the merge, so the dedicated periodic sweep and pre-merge prune defer when either path is active (including concurrent same-task merge attempts). The default age gate is 2 hours; task-aware cleanup uses a 10-minute grace period for `done`/`archived` tasks and for genuinely missing/deleted task rows, and every removal path is clamped by the same 10-minute minimum-age floor so a freshly created worktree is never reaped. Transient `getTask` lookup failures (for example SQLite busy/parse errors) are not treated as deletion evidence; they log a warning, emit `lookup-error` only if eventually removed, and retain the conservative 2-hour gate. The sweep canonicalizes paths before checking `activeSessionRegistry`, attempts `git worktree remove --force <path>` before filesystem removal, runs `git worktree prune` after cleanup attempts, and emits `worktree:tempdir-sweep` run-audit telemetry for removal attempts and failures. Fresh directories, active-session paths, and individual removal failures are skipped/logged without aborting the maintenance cycle.
|
||||
|
||||
- `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`.
|
||||
- `reattach-orphaned-assigned-executions` is a forward-resume safety net for durable-agent assignments. During startup recovery and periodic maintenance, after orphaned-agent and stale-heartbeat-run repairs, self-healing finds `in-progress` tasks with an `assignedAgentId` whose agent has no active heartbeat run and no active executor session after the orphan grace window. It re-dispatches in place via `executor.resumeTaskForAgent(agentId)` (the same seam used by clean `HeartbeatMonitor.onRunCompleted` and guarded by executor double-execution checks), emits `task:reattach-orphaned-execution`, and never moves the task backward. This complements engine-start `executor.resumeOrphaned()` and leaves unassigned/role-based execution recovery to the existing startup/limbo/stuck-task paths.
|
||||
- Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`.
|
||||
|
||||
#### Stuck-loop exhaustion terminal contract
|
||||
@@ -1075,6 +1076,7 @@ The run-audit system records every mutation performed by the engine across four
|
||||
- **Git / `merge:no-op-attribution-mismatch-skipped`** — emitted when the FN-5304 source-tip guard cannot run because the source branch ref is unavailable (for example already pruned). `target` is the task ID; metadata includes `reason` (`"source-ref-unavailable"`).
|
||||
- **Database / `task:auto-recover-misrouted-foreign-commit`** — emitted per dropped misrouted commit during FN-4948 contamination recovery. `target` is the recovering task; metadata carries `{ droppedSha, foreignTaskId, paths }`.
|
||||
- **Database / `task:orphan-detected-no-action`** — emitted by `recoverOrphanedExecutions` (FN-5337) when row metadata looks orphaned after grace windows; annotation-only event with no lifecycle mutation (`in-progress` task stays put).
|
||||
- **Database / `task:reattach-orphaned-execution`** — emitted by `reattachOrphanedAssignedExecutions` (FN-6336) when self-healing re-dispatches an idle assigned `in-progress` task forward via `executor.resumeTaskForAgent(agentId)` after proving the assigned agent has no active heartbeat run or active execution.
|
||||
- **Database / `task:soft-delete-column-reconciled`** — emitted by `reconcileSoftDeletedColumnDrift` (FN-5566, re-land FN-5446) when a soft-deleted row (`deletedAt IS NOT NULL`) is found with legacy `column != 'archived'`; rewrites only `column` (no resurrection), with metadata `{ previousColumn }`.
|
||||
- **Database / `session:runtime-resolved`** — emitted once per `createResolvedAgentSession` call with metadata `{ sessionPurpose, runtimeId, wasConfigured, provider, modelId, mockProviderActive, testModeActive, runtimeHint? }` for per-lane runtime/provider attribution.
|
||||
- **Database / `task:reconcile-dependency-blocking-lease`** — emitted by `reconcileDependencyBlockingLeases()` (FN-6292) when self-healing rebounds an `in-progress` holder to `todo` because an unmet dependency is blocked by the holder's stale file-scope lease. Metadata includes the dependency ID, blocked-by marker, and unmet dependency list.
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
import { mkdtempSync, rmSync, readFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { Agent, AgentHeartbeatRun, AgentStore, Task } from "@fusion/core";
|
||||
|
||||
import { SelfHealingManager } from "../self-healing.js";
|
||||
|
||||
const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000;
|
||||
const ORPHANED_WITH_WORKTREE_GRACE_MS = 300_000;
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
afterEach(() => {
|
||||
while (tempDirs.length > 0) {
|
||||
const dir = tempDirs.pop();
|
||||
if (dir) rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
function makeWorktree(): string {
|
||||
const dir = mkdtempSync(join(tmpdir(), "fn-6336-reattach-"));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
function isoAge(ms: number): string {
|
||||
return new Date(Date.now() - ms).toISOString();
|
||||
}
|
||||
|
||||
function makeTask(overrides: Partial<Task> = {}): Task {
|
||||
return {
|
||||
id: "FN-1",
|
||||
title: "assigned execution",
|
||||
description: "assigned execution",
|
||||
column: "in-progress",
|
||||
status: "in-progress",
|
||||
lineageId: "lineage-1",
|
||||
branch: "fusion/fn-1",
|
||||
worktree: makeWorktree(),
|
||||
assignedAgentId: "agent-1",
|
||||
paused: false,
|
||||
steps: [{ title: "execute", status: "in-progress" }],
|
||||
createdAt: isoAge(ORPHANED_WITH_WORKTREE_GRACE_MS + 10_000),
|
||||
updatedAt: isoAge(ORPHANED_WITH_WORKTREE_GRACE_MS + 10_000),
|
||||
...overrides,
|
||||
} as Task;
|
||||
}
|
||||
|
||||
function makeAgent(id = "agent-1"): Agent {
|
||||
return {
|
||||
id,
|
||||
name: id,
|
||||
role: "executor",
|
||||
state: "active",
|
||||
createdAt: isoAge(120_000),
|
||||
updatedAt: isoAge(120_000),
|
||||
metadata: {},
|
||||
} as Agent;
|
||||
}
|
||||
|
||||
function makeActiveRun(agentId = "agent-1"): AgentHeartbeatRun {
|
||||
return {
|
||||
id: "run-1",
|
||||
agentId,
|
||||
status: "active",
|
||||
startedAt: new Date().toISOString(),
|
||||
} as AgentHeartbeatRun;
|
||||
}
|
||||
|
||||
function buildManager({
|
||||
tasks,
|
||||
agents = [makeAgent()],
|
||||
activeRuns = new Map<string, AgentHeartbeatRun | null>(),
|
||||
hasActiveAgentExecution = () => false,
|
||||
globalPause = false,
|
||||
enginePaused = false,
|
||||
executingTaskIds = new Set<string>(),
|
||||
}: {
|
||||
tasks: Task[];
|
||||
agents?: Agent[];
|
||||
activeRuns?: Map<string, AgentHeartbeatRun | null>;
|
||||
hasActiveAgentExecution?: (agentId: string) => boolean;
|
||||
globalPause?: boolean;
|
||||
enginePaused?: boolean;
|
||||
executingTaskIds?: Set<string>;
|
||||
}) {
|
||||
const resumeAssignedTaskForAgent = vi.fn(async () => undefined);
|
||||
const recordRunAuditEvent = vi.fn(async () => undefined);
|
||||
const store = {
|
||||
getSettings: vi.fn(async () => ({ globalPause, enginePaused })),
|
||||
listTasks: vi.fn(async () => tasks),
|
||||
recordRunAuditEvent,
|
||||
} as any;
|
||||
const agentStore = {
|
||||
getAgent: vi.fn(async (agentId: string) => agents.find((agent) => agent.id === agentId) ?? null),
|
||||
getActiveHeartbeatRun: vi.fn(async (agentId: string) => activeRuns.get(agentId) ?? null),
|
||||
} as unknown as AgentStore;
|
||||
|
||||
const manager = new SelfHealingManager(store, {
|
||||
rootDir: "/tmp/fn-6336-project",
|
||||
agentStore,
|
||||
getExecutingTaskIds: () => executingTaskIds,
|
||||
hasActiveAgentExecution,
|
||||
resumeAssignedTaskForAgent,
|
||||
});
|
||||
|
||||
return { manager, resumeAssignedTaskForAgent, agentStore, store, recordRunAuditEvent };
|
||||
}
|
||||
|
||||
describe("FN-6336: reattach orphaned assigned in-progress executions", () => {
|
||||
it("re-dispatches an orphaned assigned task past worktree grace via the assigned-agent seam", async () => {
|
||||
const task = makeTask();
|
||||
const { manager, resumeAssignedTaskForAgent, recordRunAuditEvent } = buildManager({ tasks: [task] });
|
||||
|
||||
const recovered = await manager.reattachOrphanedAssignedExecutions();
|
||||
|
||||
expect(recovered).toBe(1);
|
||||
expect(resumeAssignedTaskForAgent).toHaveBeenCalledTimes(1);
|
||||
expect(resumeAssignedTaskForAgent).toHaveBeenCalledWith("agent-1");
|
||||
expect(recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
domain: "database",
|
||||
mutationType: "task:reattach-orphaned-execution",
|
||||
target: "FN-1",
|
||||
}));
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("uses the shorter grace when no task worktree exists", async () => {
|
||||
const task = makeTask({ worktree: undefined, updatedAt: isoAge(ORPHANED_EXECUTION_RECOVERY_GRACE_MS + 1_000) });
|
||||
const { manager, resumeAssignedTaskForAgent } = buildManager({ tasks: [task] });
|
||||
|
||||
await expect(manager.reattachOrphanedAssignedExecutions()).resolves.toBe(1);
|
||||
|
||||
expect(resumeAssignedTaskForAgent).toHaveBeenCalledOnce();
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("does not reattach a task that is still within the longer worktree grace window", async () => {
|
||||
const task = makeTask({ updatedAt: isoAge(ORPHANED_WITH_WORKTREE_GRACE_MS - 1_000) });
|
||||
const { manager, resumeAssignedTaskForAgent } = buildManager({ tasks: [task] });
|
||||
|
||||
await expect(manager.reattachOrphanedAssignedExecutions()).resolves.toBe(0);
|
||||
|
||||
expect(resumeAssignedTaskForAgent).not.toHaveBeenCalled();
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["an active heartbeat run exists", { activeRuns: new Map([["agent-1", makeActiveRun()]]) }, {}],
|
||||
["an active agent execution exists", { hasActiveAgentExecution: (agentId: string) => agentId === "agent-1" }, {}],
|
||||
["the task is within the no-worktree grace window", {}, { worktree: undefined, updatedAt: isoAge(ORPHANED_EXECUTION_RECOVERY_GRACE_MS - 1_000) }],
|
||||
["the task is paused", {}, { paused: true }],
|
||||
["the project is globally paused", { globalPause: true }, {}],
|
||||
["the engine is paused", { enginePaused: true }, {}],
|
||||
["the task is soft-deleted", {}, { deletedAt: new Date().toISOString() }],
|
||||
["task work is already complete", {}, { steps: [{ title: "execute", status: "done" }] }],
|
||||
["the executor is already executing the task", { executingTaskIds: new Set(["FN-1"]) }, {}],
|
||||
["the task has no assigned agent", {}, { assignedAgentId: undefined }],
|
||||
["the assigned agent is missing", { agents: [] }, {}],
|
||||
] as const)("does not reattach when %s", async (_name, managerOverrides, taskOverrides) => {
|
||||
const { manager, resumeAssignedTaskForAgent } = buildManager({ tasks: [makeTask(taskOverrides as Partial<Task>)], ...managerOverrides });
|
||||
|
||||
await expect(manager.reattachOrphanedAssignedExecutions()).resolves.toBe(0);
|
||||
|
||||
expect(resumeAssignedTaskForAgent).not.toHaveBeenCalled();
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("deduplicates multiple orphaned tasks sharing the same assigned agent", async () => {
|
||||
const first = makeTask({ id: "FN-1", lineageId: "lineage-1" });
|
||||
const second = makeTask({ id: "FN-2", lineageId: "lineage-2", branch: "fusion/fn-2" });
|
||||
const { manager, resumeAssignedTaskForAgent, recordRunAuditEvent } = buildManager({ tasks: [first, second] });
|
||||
|
||||
await expect(manager.reattachOrphanedAssignedExecutions()).resolves.toBe(1);
|
||||
|
||||
expect(resumeAssignedTaskForAgent).toHaveBeenCalledTimes(1);
|
||||
expect(resumeAssignedTaskForAgent).toHaveBeenCalledWith("agent-1");
|
||||
expect(recordRunAuditEvent).toHaveBeenCalledTimes(2);
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("only considers in-progress tasks and ignores review, done, todo, triage, and archived tasks", async () => {
|
||||
const tasks = [
|
||||
makeTask({ id: "FN-review", column: "in-review" }),
|
||||
makeTask({ id: "FN-done", column: "done" }),
|
||||
makeTask({ id: "FN-todo", column: "todo" }),
|
||||
makeTask({ id: "FN-triage", column: "triage" }),
|
||||
makeTask({ id: "FN-archived", column: "archived" }),
|
||||
] as Task[];
|
||||
const { manager, resumeAssignedTaskForAgent } = buildManager({ tasks });
|
||||
|
||||
await expect(manager.reattachOrphanedAssignedExecutions()).resolves.toBe(0);
|
||||
|
||||
expect(resumeAssignedTaskForAgent).not.toHaveBeenCalled();
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("is registered after agent and stale-run recovery in startup and periodic self-healing loops", () => {
|
||||
const source = readFileSync("src/self-healing.ts", "utf8");
|
||||
const startup = source.slice(source.indexOf("async runStartupRecovery"), source.indexOf(" stop(): void"));
|
||||
const periodicStart = source.lastIndexOf("recover-ghost-review");
|
||||
const periodicEnd = source.indexOf("reconcile-task-worktree-metadata", periodicStart);
|
||||
const periodic = source.slice(periodicStart, periodicEnd);
|
||||
|
||||
for (const block of [startup, periodic]) {
|
||||
const orphanedAgents = block.indexOf("recover-orphaned-agents");
|
||||
const staleRuns = block.indexOf("recover-stale-heartbeat-runs");
|
||||
const reattach = block.indexOf("reattach-orphaned-assigned-executions");
|
||||
expect(orphanedAgents).toBeGreaterThanOrEqual(0);
|
||||
expect(staleRuns).toBeGreaterThan(orphanedAgents);
|
||||
expect(reattach).toBeGreaterThan(staleRuns);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -507,6 +507,7 @@ export type DatabaseMutationType =
|
||||
/** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */
|
||||
| "task:reclaim-self-owned-branch-conflict-no-action"
|
||||
| "task:orphan-detected-no-action"
|
||||
| "task:reattach-orphaned-execution"
|
||||
/** Metadata: { taskId, lastReason, stuckKillCount, attemptedStuckKillCount, maxStuckKills, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch } */
|
||||
| "task:stuck-loop-exhausted-no-action"
|
||||
/** Metadata: { taskId: string; ignoredStepUpdateCount: number; stuckKillStreak: number; lastReason: "no-progress-churn" } */
|
||||
|
||||
@@ -797,6 +797,7 @@ export class InProcessRuntime
|
||||
getActiveMergeTaskId: () => this.activeMergeTaskIdProvider?.() ?? null,
|
||||
leaseManager: this.leaseManager,
|
||||
hasActiveAgentExecution: (agentId: string) => this.heartbeatMonitor?.getTrackedAgents().includes(agentId) ?? false,
|
||||
resumeAssignedTaskForAgent: (agentId: string) => this.executor.resumeTaskForAgent(agentId),
|
||||
recoverActiveMissionValidations: async () => {
|
||||
if (!this.missionExecutionLoop) {
|
||||
return { recoveredCount: 0 };
|
||||
|
||||
@@ -316,6 +316,12 @@ export interface SelfHealingOptions {
|
||||
*/
|
||||
unbackedMergingFanoutGraceMs?: number;
|
||||
hasActiveAgentExecution?: (agentId: string) => boolean;
|
||||
/**
|
||||
* Re-dispatches an agent's orphaned assigned in-progress execution forward,
|
||||
* via Executor.resumeTaskForAgent. This must never move the task backward in
|
||||
* lifecycle; the executor seam owns all in-memory double-execution guards.
|
||||
*/
|
||||
resumeAssignedTaskForAgent?: (agentId: string) => Promise<void>;
|
||||
restartDurableAgentHeartbeat?: (agentId: string, context: { reason: string; attempt: number }) => Promise<boolean>;
|
||||
autoRecoveryDispatcher?: AutoRecoveryDispatcher;
|
||||
/** Optional ChatStore for maintenance chat-retention cleanup. */
|
||||
@@ -1029,6 +1035,7 @@ export class SelfHealingManager {
|
||||
{ name: "orphaned-planning", fn: () => this.recoverOrphanedPlanningTasks().then(() => undefined) },
|
||||
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents().then(() => undefined) },
|
||||
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns().then(() => undefined) },
|
||||
{ name: "reattach-orphaned-assigned-executions", fn: () => this.reattachOrphanedAssignedExecutions().then(() => undefined) },
|
||||
{
|
||||
name: "reap-stale-mission-validator-runs",
|
||||
fn: async () => {
|
||||
@@ -1988,6 +1995,7 @@ export class SelfHealingManager {
|
||||
{ name: "recover-ghost-review", fn: () => this.recoverGhostReviewTasks() },
|
||||
{ name: "recover-orphaned-agents", fn: () => this.recoverOrphanedAgents() },
|
||||
{ name: "recover-stale-heartbeat-runs", fn: () => this.recoverStaleHeartbeatRuns() },
|
||||
{ name: "reattach-orphaned-assigned-executions", fn: () => this.reattachOrphanedAssignedExecutions() },
|
||||
{ name: "recover-running-on-inactive-tasks", fn: () => this.recoverAgentsRunningOnInactiveTasks() },
|
||||
{ name: "recover-drifted-agent-task-links", fn: () => this.recoverDriftedAgentTaskLinks() },
|
||||
{ name: "reconcile-soft-delete-column-drift", fn: () => this.reconcileSoftDeletedColumnDrift() },
|
||||
@@ -7896,6 +7904,120 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Re-dispatch assigned in-progress tasks whose durable agent has no active
|
||||
* heartbeat run and no active executor session. This is a forward resume via
|
||||
* Executor.resumeTaskForAgent; it never moves lifecycle backward and
|
||||
* complements the observation-only recoverOrphanedExecutions pass.
|
||||
*/
|
||||
async reattachOrphanedAssignedExecutions(): Promise<number> {
|
||||
try {
|
||||
const settings = await this.store.getSettings();
|
||||
if (settings.globalPause || settings.enginePaused) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const agentStore = this.options.agentStore;
|
||||
const resumeAssignedTaskForAgent = this.options.resumeAssignedTaskForAgent;
|
||||
if (!agentStore || !resumeAssignedTaskForAgent) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const tasks = await this.store.listTasks({ column: "in-progress", slim: true });
|
||||
const executingIds = this.options.getExecutingTaskIds?.() ?? new Set<string>();
|
||||
const now = Date.now();
|
||||
const candidates: Task[] = [];
|
||||
|
||||
for (const task of tasks) {
|
||||
if (task.column !== "in-progress") continue;
|
||||
if (task.paused || task.deletedAt) continue;
|
||||
if (!task.assignedAgentId) continue;
|
||||
if (executingIds.has(task.id)) continue;
|
||||
if (isTaskWorkComplete(task)) continue;
|
||||
|
||||
const updatedAtMs = new Date(task.updatedAt).getTime();
|
||||
if (!Number.isFinite(updatedAtMs)) continue;
|
||||
const hadWorktree = Boolean(task.worktree && existsSync(task.worktree));
|
||||
const graceMs = hadWorktree ? ORPHANED_WITH_WORKTREE_GRACE_MS : ORPHANED_EXECUTION_RECOVERY_GRACE_MS;
|
||||
if (now - updatedAtMs < graceMs) continue;
|
||||
|
||||
candidates.push(task);
|
||||
}
|
||||
|
||||
if (candidates.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const tasksByAgent = new Map<string, Task[]>();
|
||||
for (const task of candidates) {
|
||||
const agentId = task.assignedAgentId;
|
||||
if (!agentId) continue;
|
||||
|
||||
const agent = await agentStore.getAgent(agentId);
|
||||
if (!agent) continue;
|
||||
|
||||
const activeRun = await agentStore.getActiveHeartbeatRun(agentId);
|
||||
if (activeRun) continue;
|
||||
if (this.options.hasActiveAgentExecution?.(agentId) === true) continue;
|
||||
|
||||
const agentTasks = tasksByAgent.get(agentId) ?? [];
|
||||
agentTasks.push(task);
|
||||
tasksByAgent.set(agentId, agentTasks);
|
||||
}
|
||||
|
||||
let reattachedAgents = 0;
|
||||
for (const [agentId, agentTasks] of tasksByAgent) {
|
||||
try {
|
||||
await resumeAssignedTaskForAgent(agentId);
|
||||
reattachedAgents += 1;
|
||||
|
||||
for (const task of agentTasks) {
|
||||
try {
|
||||
const hadWorktree = Boolean(task.worktree && existsSync(task.worktree));
|
||||
const stalenessMs = now - new Date(task.updatedAt).getTime();
|
||||
const reason = hadWorktree
|
||||
? "assigned-agent-no-active-run-or-execution-worktree-exists"
|
||||
: "assigned-agent-no-active-run-or-execution";
|
||||
|
||||
await createRunAuditor(this.store, {
|
||||
runId: generateSyntheticRunId("self-healing-reattach-orphaned-execution", task.id),
|
||||
agentId: "self-healing",
|
||||
taskId: task.id,
|
||||
taskLineageId: task.lineageId,
|
||||
phase: "reattach-orphaned-assigned-executions",
|
||||
}).database({
|
||||
type: "task:reattach-orphaned-execution",
|
||||
target: task.id,
|
||||
metadata: {
|
||||
assignedAgentId: agentId,
|
||||
priorWorktree: task.worktree ?? null,
|
||||
priorBranch: task.branch ?? null,
|
||||
hadWorktree,
|
||||
stalenessMs,
|
||||
reason,
|
||||
},
|
||||
});
|
||||
|
||||
log.log(`[reattach-orphaned-execution] ${task.id}: re-dispatched agent ${agentId} (${reason})`);
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Failed to annotate reattached orphaned execution ${task.id}: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Failed to reattach orphaned assigned executions for ${agentId}: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
return reattachedAgents;
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Orphaned assigned execution reattach failed: ${errorMessage}`);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
private getDurableAgentRecoveryState(agent: { metadata?: Record<string, unknown> | null }): {
|
||||
attempts: number;
|
||||
nextRetryAt?: string;
|
||||
|
||||
Reference in New Issue
Block a user