diff --git a/src/main/worker/handlers/run-finished-handler.ts b/src/main/worker/handlers/run-finished-handler.ts index 9ccd2707..e562e2c5 100644 --- a/src/main/worker/handlers/run-finished-handler.ts +++ b/src/main/worker/handlers/run-finished-handler.ts @@ -15,6 +15,10 @@ export interface RunFinishedHooks { | { ok: false; errorMessage: string }; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + export function handleRunFinished( event: Extract, deps: HandlerDeps, @@ -37,7 +41,11 @@ export function handleRunFinished( } : event; if (boundaryResult && !boundaryResult.ok) { - deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + try { + deps.db.core.runs.markRunCompletionFailure(event.runId, boundaryResult.errorMessage); + } catch (error) { + deps.logger.error(`[worker] markRunCompletionFailure failed for artifact boundary ${event.runId}: ${errorMessage(error)}`); + } } emitAndNotifyCompletion(effectiveEvent, deps, rfCtx); const { hadDeferredWakes } = runPostCompletionAutomation(effectiveEvent, deps, context, rfCtx); diff --git a/src/main/worker/handlers/run-finished-helpers.ts b/src/main/worker/handlers/run-finished-helpers.ts index 4b1e9cda..9717f8f8 100644 --- a/src/main/worker/handlers/run-finished-helpers.ts +++ b/src/main/worker/handlers/run-finished-helpers.ts @@ -16,6 +16,22 @@ export interface RunFinishedContext { runOutcome: ReturnType; } +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function runCompletionSideEffect( + logger: HandlerDeps["logger"], + label: string, + operation: () => void, +): void { + try { + operation(); + } catch (error) { + logger.error(`[worker] ${label} failed after run completion: ${errorMessage(error)}`); + } +} + export function finalizeRun( event: RunFinishedEvent, deps: HandlerDeps, @@ -71,13 +87,15 @@ export function finalizeRun( leaseRenewalMap.delete(event.runId); const runOutcome = db.core.runs.finishRun({ ...event, unattributedCost }); - db.core.activity.addActivity({ - companyId: run.companyId, - actor: run.agentId, - action: `run.${event.status}`, - entityType: "run", - entityId: run.id, - detail: event.summary ?? event.errorMessage ?? "", + runCompletionSideEffect(logger, "run activity logging", () => { + db.core.activity.addActivity({ + companyId: run.companyId, + actor: run.agentId, + action: `run.${event.status}`, + entityType: "run", + entityId: run.id, + detail: event.summary ?? event.errorMessage ?? "", + }); }); return { run, task, runOutcome }; @@ -85,11 +103,13 @@ export function finalizeRun( export function releaseCompletedRunWorkspaceLock( rfCtx: RunFinishedContext | null, - deps: Pick, + deps: Pick, ): void { const workspaceId = rfCtx?.run.workspaceId; if (!workspaceId) return; - deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + runCompletionSideEffect(deps.logger, "workspace lock release", () => { + deps.db.core.workspaces.releaseWorkspaceLock(workspaceId, rfCtx.run.id); + }); } export function accumulateAndRotateSession( @@ -120,22 +140,26 @@ export function accumulateAndRotateSession( const session = db.core.tasks.getSessionById(agentRecord.sessionId); if (!session) return; - sessionService.accumulateTokenUsage( - session.id, - event.inputTokens ?? 0, - event.outputTokens ?? 0, - event.runId, - ); - // Re-read after accumulation: evaluateRotation must see the counts - // that include this run, otherwise a run that just crossed the - // threshold waits a full additional cycle to rotate — the exact - // failure mode that let transcripts balloon past the context window. - const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); - if (!updatedSession) return; - const rotationResult = sessionService.evaluateRotation(updatedSession); - if (rotationResult.shouldRotate) { - sessionService.rotateSession(updatedSession); - logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + try { + sessionService.accumulateTokenUsage( + session.id, + event.inputTokens ?? 0, + event.outputTokens ?? 0, + event.runId, + ); + // Re-read after accumulation: evaluateRotation must see the counts + // that include this run, otherwise a run that just crossed the + // threshold waits a full additional cycle to rotate — the exact + // failure mode that let transcripts balloon past the context window. + const updatedSession = db.core.tasks.getSessionById(agentRecord.sessionId); + if (!updatedSession) return; + const rotationResult = sessionService.evaluateRotation(updatedSession); + if (rotationResult.shouldRotate) { + sessionService.rotateSession(updatedSession); + logger.info(`[session] Rotated session for agent ${run.agentId}: ${rotationResult.reason}`); + } + } catch (error) { + logger.error(`[session] accumulation/rotation failed for run ${run.id}: ${errorMessage(error)}`); } } @@ -147,20 +171,26 @@ export function emitAndNotifyCompletion( const { db, emitEvent, notify, logger } = deps; const { run, task, runOutcome } = rfCtx; - emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); - emitEvent({ - type: "run-status", - runId: run.id, - status: event.status, - message: event.summary ?? event.errorMessage ?? event.status, + runCompletionSideEffect(logger, "run-updated event emission", () => { + emitEvent({ type: "run-updated", runId: run.id, action: "completed" }); + }); + runCompletionSideEffect(logger, "run-status event emission", () => { + emitEvent({ + type: "run-status", + runId: run.id, + status: event.status, + message: event.summary ?? event.errorMessage ?? event.status, + }); }); if ((event.status === "failed" || event.status === "timed_out" || event.status === "interrupted") && task) { - notify({ - title: "Task Failed", - body: `${task.title} -- manual intervention needed`, - urgency: "informational", - navigation: { section: "tasks", entityId: task.id }, - batchKey: "failed_runs", + runCompletionSideEffect(logger, "task failure notification", () => { + notify({ + title: "Task Failed", + body: `${task.title} -- manual intervention needed`, + urgency: "informational", + navigation: { section: "tasks", entityId: task.id }, + batchKey: "failed_runs", + }); }); } @@ -175,22 +205,28 @@ export function emitAndNotifyCompletion( } if (updatedAgent && runOutcome.budgetHardStopped) { - db.core.activity.addActivity({ - companyId: updatedAgent.companyId, - actor: "system", - action: "agent.auto_paused", - entityType: "agent", - entityId: updatedAgent.id, - detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + runCompletionSideEffect(logger, "budget auto-pause activity logging", () => { + db.core.activity.addActivity({ + companyId: updatedAgent.companyId, + actor: "system", + action: "agent.auto_paused", + entityType: "agent", + entityId: updatedAgent.id, + detail: `Budget exhausted at ${updatedAgent.spentMonthlyUsd.toFixed(2)} / ${updatedAgent.budgetMonthlyUsd.toFixed(2)} USD.`, + }); }); - notify({ - title: "Agent Paused", - body: `${updatedAgent.name} paused -- budget limit reached`, - urgency: "informational", - navigation: { section: "agents", entityId: updatedAgent.id }, - batchKey: "agent_paused", + runCompletionSideEffect(logger, "budget auto-pause notification", () => { + notify({ + title: "Agent Paused", + body: `${updatedAgent.name} paused -- budget limit reached`, + urgency: "informational", + navigation: { section: "agents", entityId: updatedAgent.id }, + batchKey: "agent_paused", + }); + }); + runCompletionSideEffect(logger, "agent-updated event emission", () => { + emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); }); - emitEvent({ type: "agent-updated", agentId: updatedAgent.id, action: "paused" }); } } @@ -203,58 +239,97 @@ export function runPostCompletionAutomation( const { db, logger, isQuitting, wakeAgentIfPossible, processRunCompletion, onRunCompleted, notify, budgetService } = deps; const { destroyed, MAX_RETRY_ATTEMPTS, RETRY_COOLDOWN_MS, pendingRetryTimers } = context; const { run, task } = rfCtx; + let housekeepingStatus = event.status; + let housekeepingSummary = event.summary ?? null; + let housekeepingErrorMessage = event.errorMessage ?? null; // Policy-based incident check after run completion. Always runs once the // run is finalized — cost is read from the cost_entries SUM inside // finishRun/checkAndCreateIncidents, not from the worker event. if (budgetService && event.status === "succeeded") { - budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + try { + budgetService.checkAndCreateIncidents(run.companyId, run.agentId, task?.projectId ?? null); + } catch (error) { + logger.error(`[worker] budget incident check failed for run ${run.id}: ${errorMessage(error)}`); + } } if ((event.status === "failed" || event.status === "timed_out") && !isQuitting() && !destroyed && run.taskId) { - handleFailureEscalation( - { - db, - wakeAgentIfPossible, - notify: (title, body) => - notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), - logger, - }, - { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, - task, - event.status, - MAX_RETRY_ATTEMPTS, - RETRY_COOLDOWN_MS, - (fn, delayMs) => { - const retryTimer = setTimeout(() => { - pendingRetryTimers.delete(retryTimer); - if (isQuitting() || destroyed) return; - fn(); - }, delayMs); - pendingRetryTimers.add(retryTimer); - }, - ); + try { + handleFailureEscalation( + { + db, + wakeAgentIfPossible, + notify: (title, body) => + notify({ title, body, urgency: "critical", navigation: { section: "tasks", entityId: run.taskId ?? "" } }), + logger, + }, + { taskId: run.taskId, agentId: run.agentId, companyId: run.companyId }, + task, + event.status, + MAX_RETRY_ATTEMPTS, + RETRY_COOLDOWN_MS, + (fn, delayMs) => { + const retryTimer = setTimeout(() => { + pendingRetryTimers.delete(retryTimer); + if (isQuitting() || destroyed) return; + fn(); + }, delayMs); + pendingRetryTimers.add(retryTimer); + }, + ); + } catch (error) { + logger.error(`[worker] failure escalation failed for run ${run.id}: ${errorMessage(error)}`); + } } // EXEC-05: Promote deferred wakes after run completion let hadDeferredWakes = false; if (!isQuitting() && !destroyed) { - const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); - const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); - if (promoted > 0) { - logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); - } - if (pendingWakeTrigger) { - hadDeferredWakes = true; - logger.info( - `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, - ); - wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + try { + const promoted = db.automation.wakeup.promoteWakeupRequests(run.agentId, run.companyId); + const pendingWakeTrigger = db.automation.wakeup.getPreferredWakeTrigger(run.agentId, run.companyId); + if (promoted > 0) { + logger.info(`[worker] Promoted ${promoted} deferred wake(s) for agent ${run.agentId}`); + } + if (pendingWakeTrigger) { + hadDeferredWakes = true; + logger.info( + `[worker] Processing pending wake(s) for agent ${run.agentId} with priority trigger "${pendingWakeTrigger}"`, + ); + wakeAgentIfPossible(run.agentId, run.companyId, pendingWakeTrigger); + } + } catch (error) { + logger.error(`[worker] deferred wake processing failed for run ${run.id}: ${errorMessage(error)}`); } } if (!isQuitting() && !destroyed) { - processRunCompletion(run, event as z.infer & { type: "run-finished" }); + try { + processRunCompletion(run, event as z.infer & { type: "run-finished" }); + } catch (error) { + const message = errorMessage(error); + logger.error(`[worker] processRunCompletion failed for run ${run.id}: ${message}`); + housekeepingStatus = "failed"; + housekeepingSummary = null; + housekeepingErrorMessage = message; + try { + db.core.runs.markRunCompletionFailure(run.id, message); + } catch (markError) { + logger.error(`[worker] markRunCompletionFailure failed for run ${run.id}: ${errorMessage(markError)}`); + } + if (task?.status === "in_progress") { + try { + db.core.tasks.advanceTaskStatus( + task.id, + "blocked", + `Run completion automation failed: ${message}`, + ); + } catch (statusError) { + logger.error(`[worker] failed to block task ${task.id} after completion automation failure: ${errorMessage(statusError)}`); + } + } + } } // Run housekeeping: thread events + knowledge candidates @@ -265,9 +340,9 @@ export function runPostCompletionAutomation( run.taskId, run.agentId, run.companyId, - event.status, - event.summary ?? null, - event.errorMessage ?? null, + housekeepingStatus, + housekeepingSummary, + housekeepingErrorMessage, ); } catch (err) { logger.error(`[worker] onRunCompleted housekeeping failed: ${err instanceof Error ? err.message : String(err)}`); @@ -275,7 +350,7 @@ export function runPostCompletionAutomation( } // Reset retry count on successful completion - if (event.status === "succeeded" && run.taskId) { + if (housekeepingStatus === "succeeded" && run.taskId) { handleSuccessReset(db, run.taskId, logger); } diff --git a/tests/unit/worker-completion-boundary.test.ts b/tests/unit/worker-completion-boundary.test.ts index c68da7ee..d72cda6d 100644 --- a/tests/unit/worker-completion-boundary.test.ts +++ b/tests/unit/worker-completion-boundary.test.ts @@ -24,19 +24,45 @@ function runFinishedMessage(runId: string) { }; } -function createHandlerDeps(calls: string[]) { - const releaseWorkspaceLock = vi.fn(() => calls.push("releaseWorkspaceLock")); +function createHandlerDeps(calls: string[], options: { + taskId?: string | null; + taskStatus?: string; + sessionAccumulateThrows?: boolean; + processRunCompletionThrows?: boolean; + activityThrows?: boolean; + emitEventThrows?: boolean; + wakeupThrows?: boolean; + releaseWorkspaceThrows?: boolean; + markRunCompletionFailureThrows?: boolean; +} = {}) { + const releaseWorkspaceLock = vi.fn(() => { + calls.push("releaseWorkspaceLock"); + if (options.releaseWorkspaceThrows) { + throw new Error("workspace release failed"); + } + }); const finishRun = vi.fn(() => { calls.push("finishRun"); return { budgetHardStopped: false }; }); const rotateSession = vi.fn(() => calls.push("rotateSession")); - const processRunCompletion = vi.fn(() => calls.push("processRunCompletion")); + const processRunCompletion = vi.fn(() => { + calls.push("processRunCompletion"); + if (options.processRunCompletionThrows) { + throw new Error("processRunCompletion failed"); + } + }); const releaseRunResources = vi.fn((): RunResourceReleaseResult => { calls.push("releaseRunResources"); return { ok: true as const }; }); - const markRunCompletionFailure = vi.fn(() => calls.push("markRunCompletionFailure")); + const markRunCompletionFailure = vi.fn(() => { + calls.push("markRunCompletionFailure"); + if (options.markRunCompletionFailureThrows) { + throw new Error("mark completion failed"); + } + }); + const advanceTaskStatus = vi.fn(() => calls.push("advanceTaskStatus")); const db = { core: { @@ -45,14 +71,21 @@ function createHandlerDeps(calls: string[]) { id: RUN_ID, companyId: "company-1", agentId: "agent-1", - taskId: null, + taskId: options.taskId ?? null, workspaceId: "workspace-1", })), finishRun, markRunCompletionFailure, }, workspaces: { releaseWorkspaceLock }, - activity: { addActivity: vi.fn(() => calls.push("addActivity")) }, + activity: { + addActivity: vi.fn(() => { + calls.push("addActivity"); + if (options.activityThrows) { + throw new Error("activity write failed"); + } + }), + }, entity: { getAgent: vi.fn(() => ({ id: "agent-1", @@ -60,11 +93,19 @@ function createHandlerDeps(calls: string[]) { reportsTo: "manager-1", sessionId: "session-1", })), + getTask: vi.fn(() => ({ + id: options.taskId ?? "task-1", + companyId: "company-1", + agentId: "agent-1", + title: "Completion task", + status: options.taskStatus ?? "in_progress", + })), countActiveGoalsForCompany: vi.fn(() => 0), }, tasks: { getSessionById: vi.fn(() => ({ id: "session-1" })), listAgentTasks: vi.fn(() => []), + advanceTaskStatus, }, agents: { getDirectReports: vi.fn(() => []), @@ -76,14 +117,24 @@ function createHandlerDeps(calls: string[]) { }, automation: { wakeup: { - promoteWakeupRequests: vi.fn(() => 0), + promoteWakeupRequests: vi.fn(() => { + if (options.wakeupThrows) { + throw new Error("wakeup promotion failed"); + } + return 0; + }), getPreferredWakeTrigger: vi.fn(() => null), }, }, }; const sessionService = { - accumulateTokenUsage: vi.fn(() => calls.push("accumulateTokenUsage")), + accumulateTokenUsage: vi.fn(() => { + calls.push("accumulateTokenUsage"); + if (options.sessionAccumulateThrows) { + throw new Error("session accumulation failed"); + } + }), evaluateRotation: vi.fn(() => ({ shouldRotate: true, reason: "unit-test" })), rotateSession, }; @@ -98,7 +149,12 @@ function createHandlerDeps(calls: string[]) { error: vi.fn(), }, isQuitting: vi.fn(() => false), - emitEvent: vi.fn(() => calls.push("emitEvent")), + emitEvent: vi.fn(() => { + calls.push("emitEvent"); + if (options.emitEventThrows) { + throw new Error("emit failed"); + } + }), wakeAgentIfPossible: vi.fn(() => null), processRunCompletion, notify: vi.fn(), @@ -111,6 +167,7 @@ function createHandlerDeps(calls: string[]) { processRunCompletion, releaseRunResources, markRunCompletionFailure, + advanceTaskStatus, }; } @@ -227,6 +284,265 @@ describe("worker completion boundary", () => { expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); }); + it("continues completion when artifact failure marking throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + releaseRunResources, + markRunCompletionFailure, + } = createHandlerDeps(calls, { markRunCompletionFailureThrows: true }); + releaseRunResources.mockImplementationOnce(() => { + calls.push("releaseRunResources"); + return { ok: false as const, errorMessage: "completion_reconcile failed" }; + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "completion_reconcile failed"); + expect(processRunCompletion).toHaveBeenCalledWith( + expect.objectContaining({ id: RUN_ID }), + expect.objectContaining({ + status: "failed", + errorMessage: "completion_reconcile failed", + usageStatus: "error_stop", + }), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("continues completion automation when session accumulation fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { sessionAccumulateThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("marks completion failed and blocks the task when completion automation throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when activity logging fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { activityThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when renderer event emission fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { emitEventThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("releases workspace and acknowledges completion when deferred wake promotion fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { wakeupThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("acknowledges completion when workspace release fails", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + processRunCompletion, + } = createHandlerDeps(calls, { releaseWorkspaceThrows: true }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(processRunCompletion).toHaveBeenCalled(); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + + it("blocks the task and acknowledges completion when marking completion failure throws", async () => { + const calls: string[] = []; + const { + deps, + releaseWorkspaceLock, + markRunCompletionFailure, + advanceTaskStatus, + } = createHandlerDeps(calls, { + taskId: "task-1", + taskStatus: "in_progress", + processRunCompletionThrows: true, + markRunCompletionFailureThrows: true, + }); + const worker = { postMessage: vi.fn(() => calls.push("completionAck")) }; + + await handleWorkerMessage( + runFinishedMessage(RUN_ID), + worker as never, + deps as never, + workerMessageSchema, + false, + 2, + 30_000, + new Map(), + 3, + new Set(), + new Map(), + ); + + expect(markRunCompletionFailure).toHaveBeenCalledWith(RUN_ID, "processRunCompletion failed"); + expect(advanceTaskStatus).toHaveBeenCalledWith( + "task-1", + "blocked", + expect.stringContaining("Run completion automation failed: processRunCompletion failed"), + ); + expect(releaseWorkspaceLock).toHaveBeenCalledWith("workspace-1", RUN_ID); + expect(worker.postMessage).toHaveBeenCalledWith({ type: "run-completion-ack", runId: RUN_ID }); + }); + it("keeps same-workspace runs queued until parent acknowledges completion", async () => { const parentPort = new EventEmitter() as EventEmitter & { postMessage: ReturnType;