fix(worker): recover tasks after worker crash

src/main/database/runs.ts

@@ -166,20 +166,20 @@ export function buildRunMethods(
       model: string | null;
       unattributedCost: boolean;
       usageStatus?: RunUsageStatus;
+      recoverTaskForRetry?: boolean;
     }) {
       const run = getDb().prepare("select task_id as taskId, agent_id as agentId, company_id as companyId, connector_id as connectorId, started_at as startedAt from runs where id = ?").get(input.runId) as
         | { taskId: string | null; agentId: string; companyId: string; connectorId: ConnectorId; startedAt: string | null }
         | undefined;
       if (!run) return { budgetHardStopped: false };
-      // A run that never transitioned to 'running' (startedAt === null) and
-      // is now terminal with status 'failed' / 'cancelled' must roll the
-      // task back to 'todo'. Otherwise the task is stuck in 'in_progress'
-      // with no owner — checkoutTask refuses (WHERE status IN todo/backlog),
-      // and the scheduler loops forever logging "lost checkout race". The
-      // common trigger is workspace-lock contention: `dispatchRunToWorker`
-      // bails before work starts and calls finishRun('failed').
+      // Recoverable terminal paths must roll the task back to 'todo'.
+      // Otherwise the task is stuck in 'in_progress' with no owner:
+      // checkoutTask refuses (WHERE status IN todo/backlog), and the
+      // scheduler loops forever logging "lost checkout race".
       const runNeverStarted =
         run.startedAt === null && (input.status === "failed" || input.status === "cancelled");
+      const shouldReturnTaskToTodo =
+        run.taskId !== null && (runNeverStarted || input.recoverTaskForRetry === true);
       const ts = nowIso();
       const usageStatus = input.usageStatus ?? deriveLegacyRunUsageStatus({
         runStatus: input.status,
@@ -220,7 +220,7 @@ export function buildRunMethods(
           getDb()
             .prepare("delete from task_leases where task_id = ? and run_id = ?")
             .run(run.taskId, input.runId);
-          if (runNeverStarted) {
+          if (shouldReturnTaskToTodo) {
             // Roll the task back to `todo` when no *other* live run claims
             // it. Invariant: `task.status = 'in_progress'` ⇒ a queued or
             // running run owns it through `task_leases`.

src/main/worker-service.ts

@@ -1,6 +1,6 @@
 import { app, utilityProcess } from "electron";
 import { workerMessageSchema, type WorkerServiceDependencies } from "./worker/entry/types";
-import { handleWorkerExit } from "./worker/entry/crash-recovery";
+import { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "./worker/entry/crash-recovery";
 import { handleWorkerMessage } from "./worker/entry/message-handler";
 import { DEFAULT_MAX_CONSECUTIVE_SELF_WAKES } from "./run-attempt-policy";
 
@@ -44,7 +44,7 @@ export function createWorkerService({
   const HEALTHY_UPTIME_MS = 60_000;
   const crashLoopState = { consecutive: 0, halted: false, lastCreatedAt: 0 };
 
-  function scheduleWorkerRespawn() {
+  function scheduleWorkerRespawn(afterRespawn?: () => void) {
     if (crashLoopState.halted || destroyed || isQuitting()) return;
     const uptimeMs = Date.now() - crashLoopState.lastCreatedAt;
     if (uptimeMs >= HEALTHY_UPTIME_MS) {
@@ -70,6 +70,7 @@ export function createWorkerService({
     setTimeout(() => {
       if (destroyed || isQuitting() || crashLoopState.halted) return;
       createWorker();
+      afterRespawn?.();
     }, delay);
   }
 
@@ -91,29 +92,36 @@ export function createWorkerService({
       },
     });
 
-    worker.stdout?.on("data", (chunk) => {
+    const currentWorker = worker;
+
+    currentWorker.stdout?.on("data", (chunk) => {
       const text = String(chunk).trim();
       if (text) logger.debug(`[worker:stdout] ${text}`);
     });
-    worker.stderr?.on("data", (chunk) => {
+    currentWorker.stderr?.on("data", (chunk) => {
       const text = String(chunk).trim();
       if (text) logger.warn(`[worker:stderr] ${text}`);
     });
 
-    worker.on("exit", (code) => {
+    currentWorker.on("exit", (code) => {
+      if (worker === currentWorker) worker = null;
       if (isQuitting() || destroyed) return;
-      handleWorkerExit(
+      const requeueCandidates = handleWorkerExit(
         code,
-        { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible, dispatchRunToWorker },
-        scheduleWorkerRespawn,
-        getWorker,
+        { db, logger, isQuitting, publishDomainChanged, emitEvent, wakeAgentIfPossible },
+      );
+      scheduleWorkerRespawn(
+        () => requeueQueuedRunsAfterWorkerRestart(
+          requeueCandidates,
+          { db, logger, dispatchRunToWorker },
+        ),
       );
     });
 
-    worker.on("message", async (message) => {
+    currentWorker.on("message", async (message) => {
       if (isShutdownReadyMessage(message)) {
         try {
-          worker?.postMessage({ type: "shutdown-exit" });
+          currentWorker.postMessage({ type: "shutdown-exit" });
         } catch (error) {
           logger.debug(
             `[worker-service] shutdown-exit postMessage failed: ${error instanceof Error ? error.message : String(error)}`,
@@ -123,7 +131,7 @@ export function createWorkerService({
       }
       await handleWorkerMessage(
         message,
-        worker,
+        currentWorker,
         {
           db,
           logger,

src/main/worker/entry/crash-recovery.ts

@@ -1,6 +1,15 @@
 import type { WorkerServiceDependencies } from "./types";
 import { resolveWorkspaceWithLockMode } from "../../orchestration/resolve-workspace";
 
+export interface QueuedRunRecoveryCandidate {
+  id: string;
+  taskId: string | null;
+  agentId: string;
+  companyId: string;
+  workspaceId: string | null;
+  connectorId: string;
+}
+
 export function handleWorkerExit(
   code: number | null,
   {
@@ -10,26 +19,16 @@ export function handleWorkerExit(
     publishDomainChanged,
     emitEvent: _emitEvent,
     wakeAgentIfPossible,
-    dispatchRunToWorker,
-  }: Pick<WorkerServiceDependencies, "db" | "logger" | "isQuitting" | "publishDomainChanged" | "emitEvent" | "wakeAgentIfPossible" | "dispatchRunToWorker">,
-  createWorker: () => void,
-  getWorker: () => ReturnType<typeof import("electron").utilityProcess.fork> | null,
-) {
-  if (isQuitting()) return;
+  }: Pick<WorkerServiceDependencies, "db" | "logger" | "isQuitting" | "publishDomainChanged" | "emitEvent" | "wakeAgentIfPossible">,
+): QueuedRunRecoveryCandidate[] {
+  if (isQuitting()) return [];
 
   logger.warn(`[worker] Worker exited with code ${code}, recovering...`);
 
   const snapshot = db.projection.snapshot.listSnapshot();
   const affectedAgentIds = new Set<string>();
   const affectedCompanyIds = new Map<string, string>();
-  const requeueCandidates: Array<{
-    id: string;
-    taskId: string | null;
-    agentId: string;
-    companyId: string;
-    workspaceId: string | null;
-    connectorId: string;
-  }> = [];
+  const requeueCandidates: QueuedRunRecoveryCandidate[] = [];
 
   for (const run of snapshot.runs) {
     if (run.status === "running") {
@@ -42,14 +41,11 @@ export function handleWorkerExit(
         signal: null,
         model: null,
         unattributedCost: true,
+        recoverTaskForRetry: run.taskId !== null,
       });
       if (run.workspaceId) {
         db.core.workspaces.releaseWorkspaceLock(run.workspaceId, run.id);
       }
-      // EXEC-02: Release execution lock on crash recovery
-      if (run.taskId) {
-        db.core.tasks.releaseTaskLock(run.taskId, run.id);
-      }
       affectedAgentIds.add(run.agentId);
       affectedCompanyIds.set(run.agentId, run.companyId);
 
@@ -77,61 +73,70 @@ export function handleWorkerExit(
       logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${agentId} after crash recovery`);
     }
     const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(agentId, cid);
-    if (pendingWakeTrigger) {
-      wakeAgentIfPossible(agentId, cid, pendingWakeTrigger);
-    }
+    wakeAgentIfPossible(agentId, cid, pendingWakeTrigger ?? "recovery");
   }
 
   publishDomainChanged();
 
-  createWorker();
+  return requeueCandidates;
+}
 
-  const worker = getWorker();
-  if (requeueCandidates.length > 0 && worker) {
-    logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`);
-    const recoverySnapshot = db.projection.snapshot.listSnapshot();
-    for (const candidate of requeueCandidates) {
-      try {
-        // The crashed run may still own a workspace lock acquired during the
-        // original dispatch; release it before re-dispatch so the replay
-        // does not self-collide with "Workspace is locked by another run".
-        if (candidate.workspaceId) {
-          db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
-        }
-        const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null;
-        const agent = db.core.entity.getAgent(candidate.agentId);
-        const connector = db.core.connectors.getConnector(agent.connectorId);
-        const { workspace, lockMode } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId);
-        const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId);
+export function requeueQueuedRunsAfterWorkerRestart(
+  requeueCandidates: QueuedRunRecoveryCandidate[],
+  {
+    db,
+    logger,
+    dispatchRunToWorker,
+  }: Pick<WorkerServiceDependencies, "db" | "logger" | "dispatchRunToWorker">,
+): void {
+  if (requeueCandidates.length === 0) return;
+  logger.info(`[worker] Re-enqueuing ${requeueCandidates.length} queued run(s) after worker restart.`);
+  const recoverySnapshot = db.projection.snapshot.listSnapshot();
+  for (const candidate of requeueCandidates) {
+    try {
+      const currentRun = db.core.runs.getRun(candidate.id);
+      if (currentRun.status !== "queued") continue;
+      // The crashed run may still own a workspace lock acquired during the
+      // original dispatch; release it before re-dispatch so the replay
+      // does not self-collide with "Workspace is locked by another run".
+      if (candidate.workspaceId) {
+        db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
+      }
+      const task = candidate.taskId ? db.core.entity.getTask(candidate.taskId) : null;
+      const agent = db.core.entity.getAgent(candidate.agentId);
+      const connector = db.core.connectors.getConnector(agent.connectorId);
+      const { workspace, lockMode, observationPath } = resolveWorkspaceWithLockMode(db, task, agent, candidate.companyId);
+      const recoveryCompany = recoverySnapshot.companies.find((company) => company.id === candidate.companyId);
 
-        dispatchRunToWorker({
-          runId: candidate.id,
-          companyId: candidate.companyId,
-          agent,
-          task,
-          connector,
-          workspace,
-          workspaceMode: lockMode,
-          companyName: recoveryCompany?.name ?? "Company",
-          companyDescription: recoveryCompany?.description ?? "",
-          snapshot: recoverySnapshot,
-          trigger: "recovery",
-        });
-      } catch (error) {
-        logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`);
-        db.core.runs.finishRun({
-          runId: candidate.id,
-          status: "interrupted",
-          summary: "Could not re-enqueue after worker restart.",
-          errorMessage: error instanceof Error ? error.message : String(error),
-          exitCode: null,
-          signal: null,
-          model: null,
-          unattributedCost: true,
-        });
-        if (candidate.workspaceId) {
-          db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
-        }
+      dispatchRunToWorker({
+        runId: candidate.id,
+        companyId: candidate.companyId,
+        agent,
+        task,
+        connector,
+        workspace,
+        workspaceMode: lockMode,
+        observationPath,
+        companyName: recoveryCompany?.name ?? "Company",
+        companyDescription: recoveryCompany?.description ?? "",
+        snapshot: recoverySnapshot,
+        trigger: "recovery",
+      });
+    } catch (error) {
+      logger.warn(`[worker] Failed to re-enqueue run ${candidate.id}: ${error instanceof Error ? error.message : String(error)}`);
+      db.core.runs.finishRun({
+        runId: candidate.id,
+        status: "interrupted",
+        summary: "Could not re-enqueue after worker restart.",
+        errorMessage: error instanceof Error ? error.message : String(error),
+        exitCode: null,
+        signal: null,
+        model: null,
+        unattributedCost: true,
+        recoverTaskForRetry: candidate.taskId !== null,
+      });
+      if (candidate.workspaceId) {
+        db.core.workspaces.releaseWorkspaceLock(candidate.workspaceId, candidate.id);
       }
     }
   }

src/main/worker/entry/types.ts

@@ -5,6 +5,7 @@ import type { NotificationOptions } from "../../notification-service";
 import type { createBudgetEnforcementService } from "../../budget-enforcement-service";
 import type { createSessionManagementService } from "../../session-management-service";
 import type { RuntimeBridge, RuntimeMessageInjector } from "../../agent-runtime";
+import type { RunDispatchParams } from "../../orchestration/turn-context";
 
 const runStatusSchema = z.enum(["queued", "running", "succeeded", "failed", "cancelled", "timed_out", "interrupted"]);
 const runUsageStatusSchema = z.enum(RUN_USAGE_STATUSES);
@@ -110,19 +111,7 @@ export interface WorkerServiceDependencies {
   publishDomainChanged: () => void;
   emitEvent: (event: DesktopEvent) => void;
   wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
-  dispatchRunToWorker: (params: {
-    runId: string;
-    companyId: string;
-    agent: ReturnType<AppDatabase["core"]["entity"]["getAgent"]>;
-    task: ReturnType<AppDatabase["core"]["entity"]["getTask"]> | null;
-    connector: ReturnType<AppDatabase["core"]["connectors"]["getConnector"]>;
-    workspace: ReturnType<AppDatabase["core"]["workspaces"]["getWorkspace"]>;
-    workspaceMode: "exclusive" | "shared" | "none";
-    companyName: string;
-    companyDescription: string;
-    snapshot: ReturnType<AppDatabase["projection"]["snapshot"]["listSnapshot"]>;
-    trigger: string;
-  }) => void;
+  dispatchRunToWorker: (params: RunDispatchParams) => void;
   processRunCompletion: (run: RunRecord, event: z.infer<typeof workerMessageSchema> & { type: "run-finished" }) => void;
   onRunStarted?: (runId: string, taskId: string | null, agentId: string, companyId: string) => void;
   onRunCompleted?: (runId: string, taskId: string | null, agentId: string, companyId: string, status: string, summary: string | null, errorMessage: string | null) => void;