diff --git a/packages/adapters/src/executor-computer-busy-retry.test.ts b/packages/adapters/src/executor-computer-busy-retry.test.ts new file mode 100644 index 0000000..eee2feb --- /dev/null +++ b/packages/adapters/src/executor-computer-busy-retry.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it, vi } from "vitest"; +import { + COMPUTER_BUSY_EXHAUSTED_ERROR, + COMPUTER_BUSY_MAX_ATTEMPTS, + computerBusyRetryMarker, + failComputerBusyExhaustedRun, + isComputerBusyRetryMarker, + nextComputerBusyRetryCount, + parseComputerBusyRetryCount, + preserveRunErrorForComputerBusyRetry, + requeueComputerBusyRun, + shouldRetryComputerBusy, +} from "./executor-computer-busy-retry.js"; +import type { ExecutorDeps } from "./executor.js"; + +describe("computer busy retry policy", () => { + it("tracks consecutive busy retries in an internal run.error marker", () => { + expect(parseComputerBusyRetryCount(null)).toBe(0); + expect(parseComputerBusyRetryCount("Computer is busy")).toBe(0); + expect(parseComputerBusyRetryCount(computerBusyRetryMarker(3))).toBe(3); + expect(isComputerBusyRetryMarker(computerBusyRetryMarker(1))).toBe(true); + expect(isComputerBusyRetryMarker("Run setup failed; retrying")).toBe(false); + }); + + it("preserves only the internal busy-retry marker across lease claims", () => { + expect(preserveRunErrorForComputerBusyRetry(null)).toBeNull(); + expect(preserveRunErrorForComputerBusyRetry("Run setup failed; retrying")).toBeNull(); + expect(preserveRunErrorForComputerBusyRetry(computerBusyRetryMarker(2))).toBe( + computerBusyRetryMarker(2), + ); + }); + + it("allows short busy waits but stops after the configured bound", () => { + expect(shouldRetryComputerBusy(0)).toBe(true); + expect(shouldRetryComputerBusy(COMPUTER_BUSY_MAX_ATTEMPTS - 1)).toBe(true); + expect(shouldRetryComputerBusy(COMPUTER_BUSY_MAX_ATTEMPTS)).toBe(false); + expect(shouldRetryComputerBusy(COMPUTER_BUSY_MAX_ATTEMPTS + 5)).toBe(false); + }); + + it("increments retry markers monotonically until exhausted", () => { + let prior = 0; + for (let attempt = 1; attempt <= COMPUTER_BUSY_MAX_ATTEMPTS; attempt += 1) { + expect(shouldRetryComputerBusy(prior)).toBe(true); + prior = nextComputerBusyRetryCount(prior); + expect(computerBusyRetryMarker(prior)).toBe(`computer_busy_retry:${attempt}`); + } + expect(shouldRetryComputerBusy(prior)).toBe(false); + expect(COMPUTER_BUSY_EXHAUSTED_ERROR).toMatch(/stayed busy/i); + }); + + it("requeues while under the bound and stops enqueueing once exhausted", async () => { + const enqueue = vi.fn(async () => undefined); + const updateMany = vi.fn(async () => ({ count: 1 })); + const deps = { + prisma: { run: { updateMany } }, + jobs: { enqueue }, + } as unknown as ExecutorDeps; + + const requeued = await requeueComputerBusyRun( + deps, + "run-1", + "worker-1", + 3, + COMPUTER_BUSY_MAX_ATTEMPTS - 1, + null, + ); + expect(requeued).toBe(true); + expect(enqueue).toHaveBeenCalledOnce(); + expect(updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + error: computerBusyRetryMarker(COMPUTER_BUSY_MAX_ATTEMPTS), + }), + }), + ); + + enqueue.mockClear(); + updateMany.mockClear(); + const exhausted = await requeueComputerBusyRun( + deps, + "run-1", + "worker-1", + 4, + COMPUTER_BUSY_MAX_ATTEMPTS, + null, + ); + expect(exhausted).toBe(false); + expect(enqueue).not.toHaveBeenCalled(); + expect(updateMany).not.toHaveBeenCalled(); + }); + + it("fails the run with a clear error once busy retries are exhausted", async () => { + const finalizeRun = vi.fn(async () => ({ continuationRunId: null })); + const deps = { + prisma: { + attempt: { create: vi.fn(async () => ({ id: "attempt-1" })) }, + bot: { findUniqueOrThrow: vi.fn(async () => ({ id: "bot-1", name: "Assistant" })) }, + }, + events: { finalizeRun }, + jobs: { enqueue: vi.fn(async () => undefined) }, + } as unknown as ExecutorDeps; + + await failComputerBusyExhaustedRun( + deps, + { + id: "run-1", + spaceId: "space-1", + threadId: "thread-1", + botId: "bot-1", + taskId: "task-1", + trigger: "user", + sourceMessageId: null, + userId: "user-1", + }, + "worker-1", + 2, + ); + + expect(finalizeRun).toHaveBeenCalledWith( + expect.objectContaining({ + outcome: "failed", + error: COMPUTER_BUSY_EXHAUSTED_ERROR, + attemptId: "attempt-1", + }), + ); + }); +}); diff --git a/packages/adapters/src/executor-computer-busy-retry.ts b/packages/adapters/src/executor-computer-busy-retry.ts new file mode 100644 index 0000000..5f75aab --- /dev/null +++ b/packages/adapters/src/executor-computer-busy-retry.ts @@ -0,0 +1,143 @@ +import { runContinueJob } from "@rakazo/adapter-kit"; +import { getLogger } from "@rakazo/logging"; +import { returnBotMessageOutcome } from "./bot-messages.js"; +import type { ExecutorDeps } from "./executor.js"; +import { + computerRetryDelay, + notifyRun, + writeComputerRunRequeue, +} from "./executor-computer.js"; +import type { TakeoverResumeCheckpoint } from "./takeover-resume.js"; + +export const COMPUTER_BUSY_MAX_ATTEMPTS = 24; + +export const COMPUTER_BUSY_EXHAUSTED_ERROR = + "Computer stayed busy after repeated retries; try again later"; + +const COMPUTER_BUSY_RETRY_MARKER = /^computer_busy_retry:(\d+)$/; + +export function parseComputerBusyRetryCount(error: string | null | undefined): number { + if (!error) return 0; + const match = error.match(COMPUTER_BUSY_RETRY_MARKER); + if (!match) return 0; + const parsed = Number(match[1]); + return Number.isFinite(parsed) && parsed > 0 ? parsed : 0; +} + +export function isComputerBusyRetryMarker(error: string | null | undefined): boolean { + return Boolean(error && COMPUTER_BUSY_RETRY_MARKER.test(error)); +} + +export function preserveRunErrorForComputerBusyRetry(error: string | null | undefined): string | null { + return isComputerBusyRetryMarker(error) ? error : null; +} + +export function shouldRetryComputerBusy(priorRetries: number): boolean { + return priorRetries < COMPUTER_BUSY_MAX_ATTEMPTS; +} + +export function nextComputerBusyRetryCount(priorRetries: number): number { + return priorRetries + 1; +} + +export function computerBusyRetryMarker(retryCount: number): string { + return `computer_busy_retry:${retryCount}`; +} + +export async function requeueComputerBusyRun( + deps: ExecutorDeps, + runId: string, + workerId: string, + fence: number, + priorBusyRetries: number, + resumeCheckpoint: TakeoverResumeCheckpoint | null, + heldForTakeover = false, +): Promise { + if (!shouldRetryComputerBusy(priorBusyRetries)) return false; + const retryCount = nextComputerBusyRetryCount(priorBusyRetries); + const released = await writeComputerRunRequeue( + deps, + runId, + workerId, + fence, + resumeCheckpoint, + heldForTakeover, + computerBusyRetryMarker(retryCount), + ); + if (!released) return false; + await deps.jobs.enqueue({ + ...runContinueJob(runId), + availableAt: new Date(Date.now() + computerRetryDelay(retryCount)), + }); + return true; +} + +type ComputerBusyRunRef = { + id: string; + spaceId: string; + threadId: string; + botId: string; + taskId: string; + trigger: string; + sourceMessageId: string | null; + userId: string; +}; + +export async function failComputerBusyExhaustedRun( + deps: ExecutorDeps, + run: ComputerBusyRunRef, + workerId: string, + fence: number, + attemptId?: string, + bot?: { id: string; name: string }, +): Promise { + const resolvedAttemptId = + attemptId ?? + ( + await deps.prisma.attempt.create({ + data: { runId: run.id, fence, status: "running" }, + }) + ).id; + const failed = await deps.events.finalizeRun({ + spaceId: run.spaceId, + threadId: run.threadId, + botId: run.botId, + runId: run.id, + taskId: run.taskId, + attemptId: resolvedAttemptId, + leaseOwner: workerId, + leaseFence: fence, + outcome: "failed", + error: COMPUTER_BUSY_EXHAUSTED_ERROR, + }); + if (!failed) return; + if (failed.continuationRunId) { + await deps.jobs + .enqueue(runContinueJob(failed.continuationRunId)) + .catch((error) => getLogger().error("steering continuation enqueue", error)); + } + const resolvedBot = + bot ?? + (await deps.prisma.bot.findUniqueOrThrow({ + where: { id: run.botId }, + select: { id: true, name: true }, + })); + if (run.trigger === "bot_message") { + await returnBotMessageOutcome( + deps, + { ...run, sourceMessageId: run.sourceMessageId }, + resolvedBot, + `Could not complete the delegated request: ${COMPUTER_BUSY_EXHAUSTED_ERROR}`, + "status", + ).catch((error) => getLogger().error("bot message failure return", error)); + } + if (!failed.continuationRunId) { + await notifyRun(deps, run, { + kind: "failure", + title: `${resolvedBot.name} failed`, + body: COMPUTER_BUSY_EXHAUSTED_ERROR, + botId: resolvedBot.id, + threadId: run.threadId, + }); + } +} diff --git a/packages/adapters/src/executor.ts b/packages/adapters/src/executor.ts index 1a2a6b5..1ccf4c7 100644 --- a/packages/adapters/src/executor.ts +++ b/packages/adapters/src/executor.ts @@ -155,6 +155,12 @@ import { threadContextForRun, subagentMarksUnread, } from "./executor-completion.js"; +import { + failComputerBusyExhaustedRun, + parseComputerBusyRetryCount, + preserveRunErrorForComputerBusyRetry, + requeueComputerBusyRun, +} from "./executor-computer-busy-retry.js"; import { computerScreenToolResult, notifyRun, @@ -878,6 +884,7 @@ export function createRunExecutor(deps: ExecutorDeps) { takeoverContinuePlan(run); const fence = nextFence(run.leaseFence); + const priorComputerBusyRetries = parseComputerBusyRetryCount(run.error); const now = new Date(); const leased = await deps.prisma.run.updateMany({ where: { @@ -896,7 +903,7 @@ export function createRunExecutor(deps: ExecutorDeps) { leaseOwner: workerId, leaseFence: fence, leaseExpiresAt: new Date(Date.now() + 5 * 60_000), - error: null, + error: preserveRunErrorForComputerBusyRetry(run.error), checkpoint: null, }, }); @@ -935,7 +942,18 @@ export function createRunExecutor(deps: ExecutorDeps) { }); } catch (error) { if (!(error instanceof ComputerBusyError)) throw error; - await requeueComputerRun(deps, runId, workerId, fence, resumeCheckpoint, heldForTakeover); + const requeued = await requeueComputerBusyRun( + deps, + runId, + workerId, + fence, + priorComputerBusyRetries, + resumeCheckpoint, + heldForTakeover, + ); + if (!requeued) { + await failComputerBusyExhaustedRun(deps, run, workerId, fence); + } return; } const attempt = await deps.prisma.attempt @@ -4143,8 +4161,28 @@ export function createRunExecutor(deps: ExecutorDeps) { } } catch (setupError) { const computerBusy = setupError instanceof ComputerBusyError; - const retryForever = computerBusy || isTooManyDatabaseConnections(setupError); - if (!computerBusy) { + const retryDatabaseConnections = isTooManyDatabaseConnections(setupError); + if (computerBusy) { + const requeued = await requeueComputerBusyRun( + deps, + runId, + workerId, + fence, + priorComputerBusyRetries, + resumeCheckpoint, + heldForTakeover, + ); + if (requeued) { + await deps.prisma.attempt.update({ + where: { id: attempt.id }, + data: { status: "setup_failed", error: null, finishedAt: new Date() }, + }); + return; + } + await failComputerBusyExhaustedRun(deps, run, workerId, fence, attempt.id); + return; + } + if (!retryDatabaseConnections) { // undici collapses every network failure to "fetch failed"; the cause names the // host and errno, which is the only part worth paging over. const causeMessage = @@ -4168,18 +4206,18 @@ export function createRunExecutor(deps: ExecutorDeps) { fence, resumeCheckpoint, heldForTakeover, - retryForever ? null : "Run setup failed; retrying", + retryDatabaseConnections ? null : "Run setup failed; retrying", ); if (released) { await deps.prisma.attempt.update({ where: { id: attempt.id }, data: { status: "setup_failed", - error: retryForever ? null : "Run setup failed; retrying", + error: retryDatabaseConnections ? null : "Run setup failed; retrying", finishedAt: new Date(), }, }); - if (retryForever) { + if (retryDatabaseConnections) { await deps.jobs.enqueue({ ...runContinueJob(runId), availableAt: new Date(Date.now() + computerRetryDelay(fence)),