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 <cynarAI@users.noreply.github.com>

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",
+      }),
+    );
+  });
+});

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<boolean> {
+  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<void> {
+  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,
+    });
+  }
+}

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);
+        const retryDatabaseConnections = isTooManyDatabaseConnections(setupError);
-        if (!computerBusy) {
+        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)),