From 7dbaaf00222bfe3fcb93f69f205df241f1fb99af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Tue, 28 Apr 2026 23:37:24 -0400 Subject: [PATCH] fix(worker): recover tasks after worker crash --- src/main/database/runs.ts | 16 +- src/main/worker-service.ts | 32 ++-- src/main/worker/entry/crash-recovery.ts | 139 +++++++------- src/main/worker/entry/types.ts | 15 +- tests/unit/worker-crash-recovery.test.ts | 201 +++++++++++++++++++++ tests/unit/worker-service-shutdown.test.ts | 135 +++++++++++++- 6 files changed, 437 insertions(+), 101 deletions(-) create mode 100644 tests/unit/worker-crash-recovery.test.ts diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 45e444b8..d84f362c 100644 --- a/src/main/database/runs.ts +++ b/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`. diff --git a/src/main/worker-service.ts b/src/main/worker-service.ts index 700d355a..18f1ba5f 100644 --- a/src/main/worker-service.ts +++ b/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, diff --git a/src/main/worker/entry/crash-recovery.ts b/src/main/worker/entry/crash-recovery.ts index 7dcb1e80..5c0cae21 100644 --- a/src/main/worker/entry/crash-recovery.ts +++ b/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, - createWorker: () => void, - getWorker: () => ReturnType | null, -) { - if (isQuitting()) return; + }: Pick, +): QueuedRunRecoveryCandidate[] { + if (isQuitting()) return []; logger.warn(`[worker] Worker exited with code ${code}, recovering...`); const snapshot = db.projection.snapshot.listSnapshot(); const affectedAgentIds = new Set(); const affectedCompanyIds = new Map(); - 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, +): 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); } } } diff --git a/src/main/worker/entry/types.ts b/src/main/worker/entry/types.ts index cbc391ba..d7c87177 100644 --- a/src/main/worker/entry/types.ts +++ b/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; - task: ReturnType | null; - connector: ReturnType; - workspace: ReturnType; - workspaceMode: "exclusive" | "shared" | "none"; - companyName: string; - companyDescription: string; - snapshot: ReturnType; - trigger: string; - }) => void; + dispatchRunToWorker: (params: RunDispatchParams) => void; processRunCompletion: (run: RunRecord, event: z.infer & { 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; diff --git a/tests/unit/worker-crash-recovery.test.ts b/tests/unit/worker-crash-recovery.test.ts new file mode 100644 index 00000000..94771f86 --- /dev/null +++ b/tests/unit/worker-crash-recovery.test.ts @@ -0,0 +1,201 @@ +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { mkdtemp, rm } from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("electron", () => ({ + safeStorage: { + isEncryptionAvailable: () => false, + getSelectedStorageBackend: () => "basic_text", + encryptString: (value: string) => Buffer.from(value, "utf8"), + decryptString: (value: Buffer) => value.toString("utf8"), + }, +})); + +import { AppDatabase } from "../../src/main/database"; +import { handleWorkerExit, requeueQueuedRunsAfterWorkerRestart } from "../../src/main/worker/entry/crash-recovery"; + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function createCompany(db: AppDatabase): string { + return db.core.company.saveCompany({ + name: "Crash Recovery Co", + description: "Worker crash recovery contract test", + status: "active", + }); +} + +function createConnector(db: AppDatabase): void { + db.core.connectors.saveConnector({ + id: "openai-codex", + model: "gpt-5.4", + envBindingText: "", + notes: "Test connector", + }); +} + +function createWorkspace(db: AppDatabase, companyId: string, localPath: string): string { + return db.core.workspaces.saveWorkspace({ + companyId, + name: "Primary", + localPath, + repoUrl: "", + repoRef: "", + isPrimary: true, + }); +} + +function createAgent(db: AppDatabase, companyId: string, workspaceId: string): string { + return db.core.agents.saveAgent({ + companyId, + name: "Recovery Agent", + role: "Engineer", + title: "Engineer", + status: "active", + reportsTo: null, + connectorId: "openai-codex", + workspaceId, + model: "gpt-5.4", + capabilities: "Continues work after runtime crashes", + budgetMonthlyUsd: 100, + }); +} + +describe("worker crash recovery", () => { + let tempDir: string; + let db: AppDatabase; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "agentcompany-worker-crash-")); + db = new AppDatabase(tempDir); + db.init(); + }); + + afterEach(async () => { + db.close(); + await rm(tempDir, { recursive: true, force: true }); + }); + + it("returns interrupted running tasks to checkoutable work and wakes the original agent", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Continue Phase 2", + description: "This task must remain recoverable after the worker process exits.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + db.core.runs.markRunStatus(run!.id, "running", "Started"); + expect(db.core.entity.getTask(taskId).status).toBe("in_progress"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const wakeAgentIfPossible = vi.fn(() => null); + handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible, + }, + ); + + expect(db.core.runs.getRun(run!.id).status).toBe("interrupted"); + expect(db.core.entity.getTask(taskId).status).toBe("todo"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(false); + expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "recovery"); + + const retryRun = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(retryRun).not.toBeNull(); + }); + + it("requeues queued runs after the replacement worker is ready", () => { + const companyId = createCompany(db); + createConnector(db); + const workspaceId = createWorkspace(db, companyId, tempDir); + const agentId = createAgent(db, companyId, workspaceId); + const taskId = db.core.tasks.saveTask({ + companyId, + title: "Queued recovery task", + description: "This run was queued when the worker process exited.", + assigneeAgentId: agentId, + workspaceId, + priority: "high", + status: "todo", + }); + const run = db.core.runs.createTaskRun({ + companyId, + taskId, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + expect(run).not.toBeNull(); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + + const dispatchRunToWorker = vi.fn(); + const requeueCandidates = handleWorkerExit( + 1, + { + db, + logger: createLogger(), + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + }, + ); + + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + expect(requeueCandidates).toHaveLength(1); + expect(db.core.runs.getRun(run!.id).status).toBe("queued"); + + requeueQueuedRunsAfterWorkerRestart( + requeueCandidates, + { + db, + logger: createLogger(), + dispatchRunToWorker, + }, + ); + + expect(dispatchRunToWorker).toHaveBeenCalledOnce(); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: run!.id, + companyId, + task: expect.objectContaining({ id: taskId }), + agent: expect.objectContaining({ id: agentId }), + trigger: "recovery", + })); + expect(db.core.runs.hasActiveRun({ companyId, agentId })).toBe(true); + }); +}); diff --git a/tests/unit/worker-service-shutdown.test.ts b/tests/unit/worker-service-shutdown.test.ts index 72698174..909828a4 100644 --- a/tests/unit/worker-service-shutdown.test.ts +++ b/tests/unit/worker-service-shutdown.test.ts @@ -1,5 +1,5 @@ import { EventEmitter } from "node:events"; -import { describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ forkMock: vi.fn(), @@ -25,6 +25,15 @@ function createWorker() { } describe("worker service shutdown", () => { + beforeEach(() => { + mocks.forkMock.mockReset(); + }); + + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + }); + it("keeps completed run outcomes when cleaning shutdown runs", async () => { const worker = createWorker(); mocks.forkMock.mockReturnValueOnce(worker); @@ -83,4 +92,128 @@ describe("worker service shutdown", () => { status: "interrupted", })); }); + + it("clears the crashed worker reference before recovery runs", () => { + vi.useFakeTimers(); + const worker = createWorker(); + mocks.forkMock.mockReturnValueOnce(worker); + const publishDomainChanged = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ runs: [] })), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged, + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker: vi.fn(), + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + expect(service.getWorker()).toBe(worker); + worker.emit("exit", 1); + + expect(service.getWorker()).toBeNull(); + expect(publishDomainChanged).toHaveBeenCalledOnce(); + }); + + it("requeues queued crash runs after the replacement worker is created", () => { + vi.useFakeTimers(); + const crashedWorker = createWorker(); + const replacementWorker = createWorker(); + mocks.forkMock + .mockReturnValueOnce(crashedWorker) + .mockReturnValueOnce(replacementWorker); + const queuedRun = { + id: "run-queued", + status: "queued", + taskId: "task-1", + agentId: "agent-1", + companyId: "company-1", + workspaceId: null, + connectorId: "openai-codex", + }; + const dispatchRunToWorker = vi.fn(); + const service = createWorkerService({ + db: { + projection: { + snapshot: { + listSnapshot: vi.fn(() => ({ + runs: [queuedRun], + companies: [{ id: "company-1", name: "Company", description: "Recovered company" }], + })), + }, + }, + core: { + runs: { + getRun: vi.fn(() => queuedRun), + }, + entity: { + getTask: vi.fn(() => ({ + id: "task-1", + companyId: "company-1", + workspaceId: null, + projectId: null, + requiresFilesystem: false, + })), + getAgent: vi.fn(() => ({ + id: "agent-1", + companyId: "company-1", + connectorId: "openai-codex", + workspaceId: null, + })), + }, + connectors: { + getConnector: vi.fn(() => ({ id: "openai-codex" })), + }, + workspaces: { + releaseWorkspaceLock: vi.fn(), + getCompanyObservationWorkspace: vi.fn(() => null), + }, + }, + }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + workerPath: "/tmp/worker.js", + isQuitting: () => false, + publishDomainChanged: vi.fn(), + emitEvent: vi.fn(), + wakeAgentIfPossible: vi.fn(() => null), + dispatchRunToWorker, + processRunCompletion: vi.fn(), + onRunStarted: vi.fn(), + onRunCompleted: vi.fn(), + notify: vi.fn(), + budgetService: undefined, + sessionService: undefined, + getRuntimeBridge: vi.fn(() => null), + getRuntimeMessageInjector: vi.fn(() => null), + } as never); + + service.createWorker(); + crashedWorker.emit("exit", 1); + expect(dispatchRunToWorker).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1_000); + + expect(service.getWorker()).toBe(replacementWorker); + expect(dispatchRunToWorker).toHaveBeenCalledWith(expect.objectContaining({ + runId: "run-queued", + companyId: "company-1", + trigger: "recovery", + })); + }); });