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): 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): boolean { return Boolean(error && COMPUTER_BUSY_RETRY_MARKER.test(error)); } export function preserveRunErrorForComputerBusyRetry(error: string | null): 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, }); } }