From a9b11399d97c3a0e521bcb75279e78b8ac53184d Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 12 Jun 2026 23:17:12 -0700 Subject: [PATCH] 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 --- .../fn-6336-reattach-orphaned-executions.md | 5 + docs/agents.md | 2 + docs/architecture.md | 2 + ...aling-reattach-orphaned-executions.test.ts | 217 ++++++++++++++++++ packages/engine/src/run-audit.ts | 1 + .../engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/self-healing.ts | 122 ++++++++++ 7 files changed, 350 insertions(+) create mode 100644 .changeset/fn-6336-reattach-orphaned-executions.md create mode 100644 packages/engine/src/__tests__/self-healing-reattach-orphaned-executions.test.ts diff --git a/.changeset/fn-6336-reattach-orphaned-executions.md b/.changeset/fn-6336-reattach-orphaned-executions.md new file mode 100644 index 0000000000..a85839d93c --- /dev/null +++ b/.changeset/fn-6336-reattach-orphaned-executions.md @@ -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. diff --git a/docs/agents.md b/docs/agents.md index f920b0639d..bd61143c97 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -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". diff --git a/docs/architecture.md b/docs/architecture.md index a5225fd922..4af1b5dcf3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -676,6 +676,7 @@ Runtime action-gate flow (v1): - Worktrees-dir sweeps that list direct children of `` (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 `/.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 ` 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. diff --git a/packages/engine/src/__tests__/self-healing-reattach-orphaned-executions.test.ts b/packages/engine/src/__tests__/self-healing-reattach-orphaned-executions.test.ts new file mode 100644 index 0000000000..c725dabab4 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-reattach-orphaned-executions.test.ts @@ -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 { + 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(), + hasActiveAgentExecution = () => false, + globalPause = false, + enginePaused = false, + executingTaskIds = new Set(), +}: { + tasks: Task[]; + agents?: Agent[]; + activeRuns?: Map; + hasActiveAgentExecution?: (agentId: string) => boolean; + globalPause?: boolean; + enginePaused?: boolean; + executingTaskIds?: Set; +}) { + 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)], ...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); + } + }); +}); diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index 9cf7077e64..611d4dea4e 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -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" } */ diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index ec32551b4b..f778f0fce5 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -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 }; diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 23bb92ff5a..2355e05f40 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -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; restartDurableAgentHeartbeat?: (agentId: string, context: { reason: string; attempt: number }) => Promise; 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 { + 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(); + 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(); + 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 | null }): { attempts: number; nextRetryAt?: string;