From e8e0b208283355a1d800a4f4d915727798c87958 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 22 Sep 2026 02:57:21 +0000 Subject: [PATCH] Bound computer-busy run retries with a clear exhaustion error Computer-busy requeues previously retried forever via retryForever, which could hang runs and burn worker cycles on a stuck computer. Add a small retry policy helper that tracks consecutive busy waits in an internal run.error marker, preserves that marker across lease claims, caps retries at 24 with existing exponential backoff, and fails the run with a clear message when the bound is reached. Database connection pressure retries remain unbounded. Co-authored-by: cynarAI --- .../src/executor-computer-busy-retry.test.ts | 127 ++++++++++++++++ .../src/executor-computer-busy-retry.ts | 143 ++++++++++++++++++ packages/adapters/src/executor.ts | 52 ++++++- 3 files changed, 315 insertions(+), 7 deletions(-) create mode 100644 packages/adapters/src/executor-computer-busy-retry.test.ts create mode 100644 packages/adapters/src/executor-computer-busy-retry.ts 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)),