From 04ce7eb2a8fddbb2a46c8ec99d3a748de53d7ae8 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 10 Oct 2026 22:06:55 -0500 Subject: [PATCH] fix: defer Boat admission while provider stop is pending --- .../computers/application/service.test.ts | 31 ++++- .../modules/computers/application/service.ts | 20 +++- server/src/modules/computers/domain/ledger.ts | 9 ++ server/src/modules/computers/index.ts | 2 +- .../modules/run-dispatch/adapters/postgres.ts | 3 +- .../run-dispatch/domain/policy.test.ts | 6 +- .../src/modules/run-dispatch/domain/policy.ts | 5 +- .../run-dispatch/domain/wake-context.ts | 3 + server/src/modules/run-dispatch/index.ts | 1 + .../src/services/cancelled-native-startup.ts | 37 +++++- .../services/computer-environment-driver.ts | 2 +- .../services/environment-run-orchestrator.ts | 2 + .../execution-recovery-attempt.test.ts | 2 +- .../services/execution-recovery-attempt.ts | 28 ++++- .../explicit-native-continuation.test.ts | 54 ++++++++- .../services/explicit-native-continuation.ts | 3 +- server/src/services/heartbeat.ts | 31 +++++ server/src/services/heartbeat/retries.test.ts | 111 +++++++++++++++++- server/src/services/heartbeat/retries.ts | 98 +++++++++++++++- .../src/services/legacy-execution-recovery.ts | 4 +- 20 files changed, 422 insertions(+), 30 deletions(-) diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index a576befd68..c47c30e5e7 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -14,7 +14,7 @@ import { spawnSync } from "node:child_process"; import { remoteProgram } from "../adapters/remote-program.js"; import { createComputerService } from "./service.js"; import type { ComputerBackend, ComputerRepository } from "./ports.js"; -import { ComputerError, type ComputerRecord } from "../domain/ledger.js"; +import { ComputerError, ComputerStopPendingError, type ComputerRecord } from "../domain/ledger.js"; function fixture() { let record: ComputerRecord | undefined; @@ -497,19 +497,44 @@ describe("computer ownership", () => { expect((await f.repository.get(f.scope)).ledger.owners[0]?.generation).toBe(first.owner.generation); }); - it("bounds the wait for terminal predecessor teardown without forcing retirement", async () => { + it.each([false, true])("bounds terminal predecessor teardown with pending-stop deferral=%s", async (deferPendingStop) => { const f = fixture(); await f.attach(); await f.admit(); vi.mocked(f.repository.runState).mockResolvedValue("terminal"); const wait = vi.fn(async () => {}); const waiting = createComputerService(f.repository, f.backend, () => new Date("2026-10-10T12:00:00Z"), { admissionWaitMs: 2000, wait }); - await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000 })) + await expect(waiting.admit({ ...f.scope, agentId: "agent", runId: "follow-up", sessionKey: "session", idleTimeoutMs: 60_000, deferPendingStop })) .rejects.toMatchObject({ code: "conflict" }); expect(wait).toHaveBeenCalledTimes(2); expect(f.backend.retire).not.toHaveBeenCalled(); }); + it("defers an exact pending stop before owner creation and admits the same session after completion", async () => { + const f = fixture(); await f.attach(); await f.service.reconcile(); + vi.mocked(f.backend.ready).mockClear(); + const request = { ...f.scope, agentId: "agent", runId: "run", sessionKey: "saved-session", idleTimeoutMs: 60_000, deferPendingStop: true }; + await expect(f.service.admit(request)).rejects.toBeInstanceOf(ComputerStopPendingError); + expect((await f.repository.get(f.scope)).ledger.owners).toEqual([]); + expect(f.backend.ready).not.toHaveBeenCalled(); expect(f.backend.launch).not.toHaveBeenCalled(); + f.completeStop(); await f.service.reconcile(); + const binding = await f.service.admit({ ...request, runId: "retry" }); + await binding.launch({ command: "runnerd" }); + expect((await f.repository.get(f.scope)).ledger.owners[0]).toMatchObject({ sessionKey: "saved-session", runId: "retry" }); + expect(f.backend.launch).toHaveBeenCalledOnce(); + }); + it.each(["failing", "superseded", "different-action", "detaching"])("does not defer %s stop evidence", async kind => { + const f = fixture(); await f.attach(); await f.service.reconcile(); + vi.mocked(f.backend.stopStatus).mockImplementation(async () => { + if (kind === "different-action") await f.repository.update(f.scope, record => { record.ledger.action!.id = "replacement"; }); + if (kind === "detaching") await f.repository.update(f.scope, record => { record.ledger.status = "detaching"; }); + return { id: "stop_1", status: ["failing", "superseded"].includes(kind) ? kind : "pending" }; + }); + const failure = await f.service.admit({ ...f.scope, agentId: "agent", runId: "run", sessionKey: "session", idleTimeoutMs: 60_000, deferPendingStop: true }).catch(error => error); + expect(failure).toBeInstanceOf(ComputerError); expect(failure).not.toBeInstanceOf(ComputerStopPendingError); + expect((await f.repository.get(f.scope)).ledger.owners).toEqual([]); + }); + it("waits for a normal provider snapshot stop before admitting the next turn", async () => { const f = fixture(); await f.attach(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 094abab739..3f5e0471b6 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import type { ComputerRepository, ComputerBackend } from "./ports.js"; import { ComputerError, + ComputerStopPendingError, assertAdmission, exactOwner, expired, @@ -57,6 +58,7 @@ export function createComputerService( async function admitRecord( scope: Scope, change: (record: ComputerRecord) => T, + pendingStopRunId?: string, ): Promise { const attempts = Math.ceil((options.admissionWaitMs ?? 120_000) / 1000); const wait = @@ -71,13 +73,13 @@ export function createComputerService( } catch (error) { if ( !(error instanceof ComputerError) || - error.code !== "conflict" || - attempt >= attempts + error.code !== "conflict" ) throw error; const current = await repository.get(scope); if (current.ledger.status !== "attached") throw error; if (error instanceof ComputerSessionBusyError && error.runId && error.agentId) { + if (attempt >= attempts) throw error; // A terminal database status does not prove the process has stopped. // Wait for the existing owner transition; never retire it here or // advance its generation while final commands might still run. @@ -86,6 +88,17 @@ export function createComputerService( continue; } if (!current.ledger.action) throw error; + if (pendingStopRunId && current.ledger.action.providerStopId) { + const action = current.ledger.action; + const stop = await backend.stopStatus(current, action.providerStopId!); + const latest = await repository.get(scope); + if (stop.id === action.providerStopId && stop.status === "pending" && + latest.ledger.status === "attached" && latest.ledger.action?.id === action.id && + latest.ledger.action.providerStopId === stop.id) { + throw new ComputerStopPendingError({ ...scope, computerId: current.id, stopId: stop.id, runId: pendingStopRunId }); + } + } + if (attempt >= attempts) throw error; await reconcileRecord(current); if ((await repository.get(scope)).ledger.action) await wait(1000); } @@ -387,6 +400,7 @@ finally: probeId?: string; sessionKey: string; idleTimeoutMs: number; + deferPendingStop?: boolean; }, ) { if (input.agentId) segment(input.agentId); @@ -455,7 +469,7 @@ finally: : agentHome, }; return { record: structuredClone(record), owner: structuredClone(owner) }; - }); + }, input.deferPendingStop ? input.runId : undefined); await backend.ready(result.record); await backend.claim(result.record); if (result.owner.generation === 1 && !result.owner.process) { diff --git a/server/src/modules/computers/domain/ledger.ts b/server/src/modules/computers/domain/ledger.ts index 2ef59796be..eceef75e21 100644 --- a/server/src/modules/computers/domain/ledger.ts +++ b/server/src/modules/computers/domain/ledger.ts @@ -60,6 +60,15 @@ export class ComputerError extends Error { this.name = "ComputerError"; } } +/** A verified provider stop is pending before this run acquired an owner. */ +export class ComputerStopPendingError extends ComputerError { + constructor(readonly admission: { + companyId: string; environmentId: string; computerId: string; stopId: string; runId: string; + }) { + super("conflict", "Computer is still saving its previous session."); + this.name = "ComputerStopPendingError"; + } +} export function segment(value: string): string { if (!/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(value)) throw new ComputerError("invalid", "Invalid computer resource identifier"); diff --git a/server/src/modules/computers/index.ts b/server/src/modules/computers/index.ts index af8cd09b62..5d42d27571 100644 --- a/server/src/modules/computers/index.ts +++ b/server/src/modules/computers/index.ts @@ -3,7 +3,7 @@ import { secretService } from "../../services/secrets.js"; import { computerRepository } from "./adapters/repository.js"; import { boatBackend } from "./adapters/boat.js"; import { createComputerService } from "./application/service.js"; -export { ComputerError } from "./domain/ledger.js"; +export { ComputerError, ComputerStopPendingError } from "./domain/ledger.js"; export type { OwnerRef as ComputerOwnerRef, ProcessClaim as ComputerProcessClaim, diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 693a68a22b..f1137cc635 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -127,6 +127,7 @@ function readNonEmptyString(value: unknown): string | null { } function classifyRetryReasonKind(retryReason: string | null): RetryReasonKind { + if (retryReason === "computer_admission_wait") return "computer_admission_wait"; if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) return "max_turn_continuation"; if (retryReason === ISSUE_DISPOSITION_REPAIR_RETRY_REASON) return "disposition_repair"; if ((retryReason === "ai_connection_busy" || retryReason === "ai_connection_pool_wait")) return "ai_connection_wait"; @@ -293,7 +294,7 @@ export function createPostgresRunDispatchAdapter( runAgentId: input.agentId, issueId, retryReasonKind, - enforceIssueExecutionLock: retryReasonKind === "max_turn_continuation" || retryReasonKind === "ai_connection_wait", + enforceIssueExecutionLock: retryReasonKind === "max_turn_continuation" || retryReasonKind === "ai_connection_wait" || retryReasonKind === "computer_admission_wait", isNonAssigneeWorkspaceBusyRetry: isNonAssigneeWorkspaceBusyRetry(retryReason, input.contextSnapshot), budgetBlock: null, agentInvokable: true, diff --git a/server/src/modules/run-dispatch/domain/policy.test.ts b/server/src/modules/run-dispatch/domain/policy.test.ts index 051c676d2c..38e0dc3b3e 100644 --- a/server/src/modules/run-dispatch/domain/policy.test.ts +++ b/server/src/modules/run-dispatch/domain/policy.test.ts @@ -510,13 +510,13 @@ describe("native replacement execution authority", () => { }); -describe("AI subscription wait ownership", () => { +describe.each(["ai_connection_wait", "computer_admission_wait"] as const)("%s ownership", (retryReasonKind) => { it.each([null, "another-run"])("preserves assignee lock guards and admits only recorded non-assignee waits: %s", (issueExecutionRunId) => { const gate = { - ...baseGateFacts(), retryReasonKind: "ai_connection_wait" as const, + ...baseGateFacts(), retryReasonKind, enforceIssueExecutionLock: true, issueExecutionRunId, }; - const queued = { ...baseStalenessFacts(), retryReasonKind: "ai_connection_wait" as const, issueExecutionRunId }; + const queued = { ...baseStalenessFacts(), retryReasonKind, issueExecutionRunId }; expect(decideScheduledRetryGate(gate, NOW)).toMatchObject({ allowed: false, errorCode: "issue_execution_lock_changed" }); expect(decideQueuedRunStaleness(queued, NOW)).toMatchObject({ stale: true, errorCode: "issue_execution_lock_changed" }); const nonAssignee = { issueAssigneeAgentId: "another-agent", isNonAssigneeWorkspaceBusyRetry: true }; diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index ec4b729839..a6ac7a5a5f 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -10,6 +10,7 @@ export type RetryReasonKind = | "max_turn_continuation" | "disposition_repair" | "native_safe_replacement" + | "computer_admission_wait" | "ai_connection_wait" | "other"; @@ -408,7 +409,7 @@ export function decideScheduledRetryGate( const lockOutcome = decideExecutionLock({ requiresExecutionLock: (requiresInProgress || - (facts.retryReasonKind === "ai_connection_wait" && !facts.isNonAssigneeWorkspaceBusyRetry)) && + ((facts.retryReasonKind === "ai_connection_wait" || facts.retryReasonKind === "computer_admission_wait") && !facts.isNonAssigneeWorkspaceBusyRetry)) && facts.enforceIssueExecutionLock, runId: facts.runId, issueExecutionRunId: facts.issueExecutionRunId, @@ -640,7 +641,7 @@ export function decideQueuedRunStaleness( const lockOutcome = decideExecutionLock({ // A server-recorded non-assignee wake never held the task execution lock. requiresExecutionLock: requiresInProgress || - (facts.retryReasonKind === "ai_connection_wait" && !facts.isNonAssigneeWorkspaceBusyRetry), + ((facts.retryReasonKind === "ai_connection_wait" || facts.retryReasonKind === "computer_admission_wait") && !facts.isNonAssigneeWorkspaceBusyRetry), runId: facts.runId, issueExecutionRunId: facts.issueExecutionRunId, }); diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 2100b357d9..2516e7a2eb 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -16,6 +16,7 @@ function readNonEmptyString(value: unknown): string | null { } export const MAX_TURN_CONTINUATION_RETRY_REASON = "max_turns_continuation"; +export const COMPUTER_ADMISSION_WAIT_RETRY_REASON = "computer_admission_wait"; export const WORKSPACE_BUSY_RETRY_REASON = "workspace_busy"; export const AI_CONNECTION_POOL_WAIT_RETRY_REASON = "ai_connection_pool_wait"; export const AI_CONNECTION_BUSY_RETRY_REASON = "ai_connection_busy"; @@ -40,6 +41,8 @@ export function isNonAssigneeWorkspaceBusyRetry( contextSnapshot: Record, ): boolean { return ( + (retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON && + contextSnapshot.computerAdmissionDeferredWhileAssignee === false) || (retryReason === WORKSPACE_BUSY_RETRY_REASON && contextSnapshot.workspaceBusyDeferredWhileAssignee === false) || ((retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index d29b078323..2fbeaa7591 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -12,6 +12,7 @@ import type { RunDispatchWriter, ScheduledRetryReader } from "./application/port export { MAX_TURN_CONTINUATION_RETRY_REASON, WORKSPACE_BUSY_RETRY_REASON, + COMPUTER_ADMISSION_WAIT_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 53a17527f4..a215b0bd66 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,6 +1,8 @@ +import { isComputerAdmissionWait } from "./execution-recovery-attempt.js"; +import { adapterExecutionControls } from "./adapter-execution-control.js"; import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { computers, environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { claimedAdapterType } from "./conversation-continuation.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; @@ -70,3 +72,36 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co )).limit(1); return !execution; } + +/** Resource waiting is replayable only before any provider admission, independently of operator Stop. */ +export async function canRetryComputerAdmissionWait(db: Db, run: Run): Promise { + if (!isComputerAdmissionWait(run) || !run.finishedAt || run.runtimeModeResolvedAt || + run.nativeIssueId || run.nativeSessionId || run.processPid || run.processGroupId || + run.processStartedAt || run.sessionIdAfter || adapterExecutionControls.has(run.id)) return false; + const settledAt = run.resultJson?.computerAdmissionPreparationSettledAt; + const settled = typeof settledAt === "string" && Number.isFinite(Date.parse(settledAt)); + if (!settled && !(run.executionStage === "preparing" && run.controllerBootId && + run.controllerBootId !== legacyControllerBootId && run.controllerLeaseExpiresAt && + run.controllerLeaseExpiresAt <= new Date())) return false; + const admission = run.resultJson!.computerAdmission as Record; + const [computer] = await db.select({ id: computers.id }).from(computers).where(and( + eq(computers.id, admission.computerId), eq(computers.companyId, run.companyId), + eq(computers.environmentId, admission.environmentId), eq(computers.provider, "boat"), + )).limit(1); + if (!computer) return false; + const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )).limit(1); + const [coordinator] = await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), + )).limit(1); + if (lease || coordinator) return false; + const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + or(isNotNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, + "harness.ready", "session.started", "session.resumed", "session.updated", "turn.started", + "provider.event", "provider.rpc_result", "tool.execution.started"])), + )).limit(1); + return !execution; +} diff --git a/server/src/services/computer-environment-driver.ts b/server/src/services/computer-environment-driver.ts index 505c2885ca..845a9f0cbd 100644 --- a/server/src/services/computer-environment-driver.ts +++ b/server/src/services/computer-environment-driver.ts @@ -48,7 +48,7 @@ export function createComputerEnvironmentDriver(db: Db): EnvironmentRuntimeDrive if (input.heartbeatRunId && !input.agentId) throw new Error("computer_agent_required"); const idleTimeoutMs = resolvePaperclipRunnerIdleTimeoutMs(input.environment.config.runnerIdleTimeoutMs); const binding = input.heartbeatRunId ? await computers.admit({ companyId: input.companyId, environmentId: input.environment.id, - agentId: input.agentId ?? undefined, runId: input.heartbeatRunId, idleTimeoutMs, + agentId: input.agentId ?? undefined, runId: input.heartbeatRunId, idleTimeoutMs, deferPendingStop: true, sessionKey: createHash("sha256").update(JSON.stringify([input.agentId, input.executionWorkspaceId ?? input.issueId, input.adapterType, input.executionConfigurationKey])).digest("hex") }) : await computers.admitProbe({ companyId: input.companyId, environmentId: input.environment.id, agentId: input.agentId ?? undefined, probeId: randomUUID(), idleTimeoutMs }); diff --git a/server/src/services/environment-run-orchestrator.ts b/server/src/services/environment-run-orchestrator.ts index c5cbb0ab95..a348a35f82 100644 --- a/server/src/services/environment-run-orchestrator.ts +++ b/server/src/services/environment-run-orchestrator.ts @@ -1,3 +1,4 @@ +import { ComputerStopPendingError } from "../modules/computers/index.js"; /** * Centralized environment run orchestrator. * @@ -218,6 +219,7 @@ export function environmentRunOrchestrator( try { return await environmentRuntime.acquireRunLease(input); } catch (err) { + if (err instanceof ComputerStopPendingError) throw err; throw new EnvironmentRunError( "lease_acquire_failed", `Failed to acquire lease for environment "${input.environment.name}" (${input.environment.driver}): ${err instanceof Error ? err.message : String(err)}`, diff --git a/server/src/services/execution-recovery-attempt.test.ts b/server/src/services/execution-recovery-attempt.test.ts index c434e7b60a..d71efa9dd9 100644 --- a/server/src/services/execution-recovery-attempt.test.ts +++ b/server/src/services/execution-recovery-attempt.test.ts @@ -27,7 +27,7 @@ describe("failure attempts across resource waits", () => { }); describe("persisted independent accounting", () => { - it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { + it.each(["computer_admission_wait", "max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { expect(executionFailureRetryCount({ scheduledRetryReason, scheduledRetryAttempt: 20, contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 1 } }, })).toBe(2); diff --git a/server/src/services/execution-recovery-attempt.ts b/server/src/services/execution-recovery-attempt.ts index dec0d847cf..85d8ef1063 100644 --- a/server/src/services/execution-recovery-attempt.ts +++ b/server/src/services/execution-recovery-attempt.ts @@ -32,6 +32,10 @@ function historicalFailureCount(run: RetryRun): number { const saved = count(run.contextSnapshot?.failureRetriesBeforeAiConnectionWait); if (saved !== null) return saved; } + if (run.scheduledRetryReason === "computer_admission_wait") { + const saved = count(run.contextSnapshot?.failureRetriesBeforeComputerWait); + if (saved !== null) return saved; + } if (run.scheduledRetryReason === "workspace_busy") { const saved = count(run.contextSnapshot?.failureRetriesBeforeWorkspaceWait); if (saved !== null) return saved; @@ -42,7 +46,7 @@ function historicalFailureCount(run: RetryRun): number { export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccounting { const saved = savedAccounting(run); - const nonFailureLane = ["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"].includes(run.scheduledRetryReason ?? ""); + const nonFailureLane = ["computer_admission_wait", "max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"].includes(run.scheduledRetryReason ?? ""); return { version: 1, failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)), @@ -57,7 +61,7 @@ export function executionFailureRetryCount(run: RetryRun): number { } export function executionRetryAttemptCount(run: RetryRun, reason: string): number { - if (reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait") { + if (reason === "computer_admission_wait" || reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait") { return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0; } const accounting = executionRetryAccounting(run); @@ -67,6 +71,24 @@ export function executionRetryAttemptCount(run: RetryRun, reason: string): numbe export function accountingForScheduledRetry(run: RetryRun, reason: string, attempt: number): ExecutionRetryAccounting { const accounting = executionRetryAccounting(run); if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt; - else if (reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait") accounting.failureRetries = attempt; + else if (reason !== "computer_admission_wait" && reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait") accounting.failureRetries = attempt; return accounting; } + +/** Only the executor can attest that a pending-stop wait never dispatched and finished cleanup. */ +export function isComputerAdmissionWait(run: { + id: string; companyId: string; status: string; runtimeMode: string; errorCode?: string | null; + resultJson?: Record | null; runnerProfileJson?: Record | null; +}): boolean { + const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + const result = object(run.resultJson), recovery = object(result.executionRecovery), cancellation = object(result.cancellation), admission = object(result.computerAdmission); + return run.runtimeMode === "legacy" && run.status === "cancelled" && run.errorCode === "computer_admission_wait" && + recovery.kind === "computer_admission_wait" && recovery.providerWorkStarted === false && + result.computerAdmissionRetryOutcome == null && + cancellation.expected === true && cancellation.source === "control_plane" && + object(cancellation.initiator).type === "system" && + admission.runId === run.id && admission.companyId === run.companyId && + [admission.computerId, admission.environmentId].every(value => typeof value === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value)) && + typeof admission.stopId === "string" && admission.stopId.length > 0 && + !object(run.runnerProfileJson).nativeExecutionInput; +} diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index b26bac40ab..4664d8a23d 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -13,7 +13,7 @@ import { join } from "node:path"; import { and, eq, sql } from "drizzle-orm"; import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { - approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections, + computers, approvals, issueApprovals, issueThreadInteractions, chatConversations, chatEndpoints, toolApplications, toolConnections, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, } from "@paperclipai/db"; @@ -470,6 +470,58 @@ const support = await getEmbeddedPostgresTestSupport(); context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false, }); + async function seedComputerWaitExplicitTurn() { + const f = await seedTimedOutExplicitTurn(); + const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "computer" }).returning(); + const [computer] = await db.insert(computers).values({ companyId: f.companyId, environmentId: environment.id, + providerId: randomUUID(), ledger: {} }).returning(); + const [parent] = await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "computer_admission_wait", + executionStage: "settled", controllerLeaseExpiresAt: null, + contextSnapshot: { ...f.parent.contextSnapshot, computerAdmissionDeferredWhileAssignee: true }, + resultJson: { executionRecovery: { kind: "computer_admission_wait", providerWorkStarted: false }, + computerAdmission: { companyId: f.companyId, runId: f.parent.id, environmentId: environment.id, computerId: computer.id, stopId: "stop_1" }, + cancellation: { expected: true, source: "control_plane", initiator: { type: "system" } }, + computerAdmissionPreparationSettledAt: f.now.toISOString() }, + }).where(eq(heartbeatRuns.id, f.parent.id)).returning(); + return { ...f, parent }; + } + + it("transfers an unchanged explicit message after a settled computer wait with one successor", async () => { + const f = await seedComputerWaitExplicitTurn(); + const heartbeat = heartbeatService(db); + const options = { now: f.now, retryReason: "computer_admission_wait", delayMs: 1000 }; + const [scheduled, duplicate] = await Promise.all([ + heartbeat.scheduleBoundedRetry(f.parent.id, options), heartbeat.scheduleBoundedRetry(f.parent.id, options), + ]); + expect(scheduled).toMatchObject({ outcome: "scheduled" }); + if (scheduled.outcome !== "scheduled") throw new Error("Expected computer wait successor"); + expect(duplicate).toMatchObject({ outcome: "scheduled", run: { id: scheduled.run.id } }); + expect(scheduled.run.contextSnapshot?.explicitUserContinuation).toEqual({ previousRunId: f.parent.id, commentId: f.commentId }); + await expect(dispatchExplicitRetry(f, scheduled.run)).resolves.toMatchObject({ interruptedRunId: f.parent.id }); + const receipts = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(receipts).toHaveLength(2); + expect(receipts.find(receipt => receipt.id === f.receipt.id)?.evidence).toEqual(f.receipt.evidence); + }); + + it.each(["edited", "operator-stop", "active-control", "interrupt-receipt", "superseded"])("does not transfer computer wait authority after %s", async kind => { + const f = await seedComputerWaitExplicitTurn(); + if (kind === "edited") await db.update(issueComments).set({ body: "Changed instructions" }).where(eq(issueComments.id, f.commentId)); + if (kind === "operator-stop") await db.update(heartbeatRuns).set({ errorCode: "operator_interrupted" }).where(eq(heartbeatRuns.id, f.parent.id)); + if (kind === "interrupt-receipt") await db.update(issueRecoveryActions).set({ evidence: { ...f.receipt.evidence, + explicitUserContinuation: { ...(f.receipt.evidence.explicitUserContinuation as Record), queuedCommentInterruptId: randomUUID() }, + } }).where(eq(issueRecoveryActions.id, f.receipt.id)); + if (kind === "superseded") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "succeeded", contextSnapshot: { issueId: f.issueId }, createdAt: new Date(f.now.getTime() + 1000) }); + const control = createAdapterExecutionControl(); + if (kind === "active-control") adapterExecutionControls.set(f.parent.id, control); + try { + const result = await heartbeatService(db).scheduleBoundedRetry(f.parent.id, { now: f.now, retryReason: "computer_admission_wait" }); + expect(result.outcome).not.toBe("scheduled"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.parent.id))).toHaveLength(0); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId))).toHaveLength(1); + } finally { adapterExecutionControls.delete(f.parent.id); control.finish(); } + }); + it("re-admits an explicit user turn's timeout retry with its own durable authorization", async () => { const f = await seedTimedOutExplicitTurn(); await expect(dispatchExplicitRetry(f, f.parent)).resolves.toMatchObject({ interruptedRunId: f.sourceRunId }); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index aabb3e1b78..8240f50932 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { canRetryComputerAdmissionWait } from "./cancelled-native-startup.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; @@ -358,7 +359,7 @@ export async function admitExplicitContinuationRetry(input: { )).for("update"); if (!task || task.assigneeAgentId !== agentId || task.executionRunId !== input.parentRunId || ["done", "cancelled"].includes(task.status) || !parent || - !["failed", "timed_out"].includes(parent.status) || !parent.finishedAt || + (!["failed", "timed_out"].includes(parent.status) && !(await canRetryComputerAdmissionWait(db, parent))) || !parent.finishedAt || parent.runtimeMode !== "legacy" || parent.contextSnapshot?.issueId !== issueId || (parent.nativeIssueId !== null && parent.nativeIssueId !== issueId) || adapterExecutionControls.has(parent.id) || diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1d7ca4f550..23594b880d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { ComputerStopPendingError } from "../modules/computers/index.js"; import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js"; import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js"; @@ -1106,6 +1107,8 @@ export function heartbeatService( scheduleInteractionContinuationInfrastructureRetryIfEligible, findSharedWorkspaceHolder, finalizeAiConnectionBusyDeferral, + finalizeComputerAdmissionDeferral, + scheduleComputerAdmissionRetry, finalizeWorkspaceBusyDeferral, promoteDueScheduledRetries, retryScheduledRetryNow, @@ -5143,6 +5146,15 @@ export function heartbeatService( eventMessage: "stale execution continuation cancelled before dispatch", suppressImmediateRecovery: true, }); + } else if (outerErr instanceof ComputerStopPendingError && + outerErr.admission.runId === run.id && outerErr.admission.companyId === run.companyId && + !legacyAdapterEntered && !nativeDispatchStarted && !nativeOwnershipHeld && + run.runtimeMode === "legacy" && !runOptions.nativeRestartRecovery && !parseObject(run.runnerProfileJson).nativeExecutionInput && + !parseObject(run.contextSnapshot?.explicitUserContinuation).failedRunId && + !run.contextSnapshot?.queuedCommentInterrupt && !executionControl.controller.signal.aborted) { + const nonAssignee = parseObject(run.runnerProfileJson).aiConnectionNonAssigneeCommentWake === true || + isNonAssigneeWorkspaceBusyRetry(run.scheduledRetryReason, parseObject(run.contextSnapshot)); + await finalizeComputerAdmissionDeferral(run, outerErr, !nonAssignee); } else if (isWorkspaceBusyDeferral(outerErr)) { // Expected contention on a shared project workspace, not a // failure: park the run as a bounded scheduled retry and leave the @@ -5349,6 +5361,7 @@ export function heartbeatService( controllerLeaseExpiresAt: null, }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.controllerBootId, legacyControllerBootId), + sql`${heartbeatRuns.errorCode} is distinct from 'computer_admission_wait'`, inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]))); } finally { @@ -5363,6 +5376,24 @@ export function heartbeatService( adapterExecutionControls.delete(run.id); } } + if (latestRun?.status === "cancelled" && latestRun.errorCode === "computer_admission_wait" && + !nativeDispatchStarted && !legacyAdapterEntered && !nativeOwnershipHeld) { + // Keep the controller lease until this atomic cleanup receipt: a crash + // before it is recoverable only after the old controller lease expires. + // An intervening Stop permanently suppresses automatic admission retry. + const aborted = executionControl.controller.signal.aborted; + const [settledWait] = await db.update(heartbeatRuns).set({ + executionStage: "settled", + controllerLeaseExpiresAt: null, + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + ${JSON.stringify(aborted + ? { computerAdmissionRetryOutcome: "aborted" } + : { computerAdmissionPreparationSettledAt: new Date().toISOString() })}::jsonb`, + }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled"), + eq(heartbeatRuns.controllerBootId, legacyControllerBootId), + eq(heartbeatRuns.errorCode, "computer_admission_wait"))).returning(); + if (settledWait && !aborted) await scheduleComputerAdmissionRetry(settledWait); + } // Terminalization precedes lease and adapter cleanup. Only now is the // owner gone; retry pending input for ordinary completions as well as Stop. if (latestRun?.runtimeMode === "legacy" && ["failed", "timed_out"].includes(latestRun.status) && diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts index 635e8ca9a5..7176303017 100644 --- a/server/src/services/heartbeat/retries.test.ts +++ b/server/src/services/heartbeat/retries.test.ts @@ -1,6 +1,8 @@ +import { canRetryComputerAdmissionWait } from "../cancelled-native-startup.js"; +import { ComputerStopPendingError } from "../../modules/computers/index.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; -import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db"; +import { computers, environments, environmentLeases, agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db"; import { and, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { HttpError } from "../../errors.js"; @@ -45,6 +47,11 @@ function guardedDatabase() { return { db: new Proxy({}, { get: access }) as Db, access }; } +function emptyComputerWaitScanDatabase() { + const limit = vi.fn(async () => []); + return { db: { select: () => ({ from: () => ({ where: () => ({ orderBy: () => ({ limit }) }) }) }) } as unknown as Db, limit }; +} + function unitRun(): Run { const run: Partial = { id: "run", companyId: "company", agentId: "agent", status: "running", @@ -109,7 +116,7 @@ describe("heartbeat retry module callbacks", () => { }); it("passes the service's worktree cutoff and publishes only returned effects", async () => { - const database = guardedDatabase(); + const database = emptyComputerWaitScanDatabase(); const deps = callbacks(); const now = new Date("2026-10-08T12:00:00Z"); const cutoff = new Date("2026-10-08T11:00:00Z"); @@ -123,21 +130,22 @@ describe("heartbeat retry module callbacks", () => { expect(deps.runDispatch.promoteDueScheduledRetries).toHaveBeenCalledWith({ now, cutoff }); expect(deps.applyRunDispatchPostCommitEffects).toHaveBeenCalledWith([effect]); expect(deps.applyRunDispatchPostCommitEffects).toHaveBeenCalledAfter(deps.runDispatch.promoteDueScheduledRetries); - expect(database.access).not.toHaveBeenCalled(); + expect(database.limit).toHaveBeenCalledWith(50); }); it("does not publish effects when promotion rejects", async () => { const deps = callbacks(); deps.runDispatch.promoteDueScheduledRetries.mockRejectedValue(new Error("promotion failed")); - await expect(extracted.createHeartbeatRetries(guardedDatabase().db, deps).promoteDueScheduledRetries()).rejects.toThrow("promotion failed"); + await expect(extracted.createHeartbeatRetries(emptyComputerWaitScanDatabase().db, deps).promoteDueScheduledRetries()).rejects.toThrow("promotion failed"); expect(deps.applyRunDispatchPostCommitEffects).not.toHaveBeenCalled(); }); - it.each(["workspace", "connection"] as const)("leaves a %s cancellation race winner alone", async (kind) => { + it.each(["workspace", "connection", "computer"] as const)("leaves a %s cancellation race winner alone", async (kind) => { const database = guardedDatabase(); const deps = callbacks(); const retries = extracted.createHeartbeatRetries(database.db, deps); if (kind === "workspace") await retries.finalizeWorkspaceBusyDeferral(unitRun(), deferral()); + else if (kind === "computer") await retries.finalizeComputerAdmissionDeferral(unitRun(), new ComputerStopPendingError({ companyId: "company", environmentId: "environment", computerId: "computer", stopId: "stop_1", runId: "run" }), true); else await retries.finalizeAiConnectionBusyDeferral(unitRun(), new HttpError(409, "busy"), true); expect(deps.setRunStatusIfRunning).toHaveBeenCalledTimes(1); expect(deps.setWakeupStatus).not.toHaveBeenCalled(); @@ -181,6 +189,9 @@ describePostgres("heartbeat retry module database wiring", () => { afterAll(async () => { await database?.cleanup(); }); afterEach(async () => { + await db.delete(environmentLeases); + await db.delete(computers); + await db.delete(environments); await db.delete(issues); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); @@ -215,6 +226,96 @@ describePostgres("heartbeat retry module database wiring", () => { return { companyId, agent, issue, run }; } + async function computerWaitFixture(settled = true) { + const f = await fixture(); + const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "computer" }).returning(); + const [computer] = await db.insert(computers).values({ companyId: f.companyId, environmentId: environment.id, + providerId: randomUUID(), ledger: {} }).returning(); + const resultJson = { executionRecovery: { kind: "computer_admission_wait", providerWorkStarted: false }, + computerAdmission: { companyId: f.companyId, runId: f.run.id, environmentId: environment.id, computerId: computer.id, stopId: "stop_1" }, + cancellation: { expected: true, source: "control_plane", initiator: { type: "system" } }, + ...(settled ? { computerAdmissionPreparationSettledAt: now.toISOString() } : {}) }; + const [run] = await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "computer_admission_wait", resultJson, + executionStage: settled ? "settled" : "preparing", controllerBootId: randomUUID(), controllerLeaseExpiresAt: now, + contextSnapshot: { ...f.run.contextSnapshot, computerAdmissionDeferredWhileAssignee: true, preserved: "original context" }, + }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + return { ...f, run, environment, computer }; + } + + it.each([true, false])("recovers a computer wait across restart with cleanup receipt=%s and one successor", async settled => { + const f = await computerWaitFixture(settled); + const first = boundRetries(), second = boundRetries(); + const opts = { now, retryReason: "computer_admission_wait", delayMs: 30_000 }; + const results = await Promise.all([first.retries.scheduleBoundedRetryForRun(f.run, f.agent, opts), second.retries.scheduleBoundedRetryForRun(f.run, f.agent, opts)]); + expect(results).toEqual([expect.objectContaining({ outcome: "scheduled" }), expect.objectContaining({ outcome: "scheduled" })]); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", contextSnapshot: { preserved: "original context", executionRetryAccounting: { failureRetries: 0 } } }); + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + expect(issue.executionRunId).toBe(successors[0].id); + expect(await boundRetries().retries.promoteDueScheduledRetries(new Date(now.getTime() + 31_000))) + .toMatchObject({ promoted: 1, runIds: [successors[0].id] }); + }); + + it.each([true, false])("sweeps a durable wait after restart with cleanup receipt=%s", async settled => { + const f = await computerWaitFixture(settled); + await boundRetries().retries.promoteDueScheduledRetries(new Date()); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", contextSnapshot: { preserved: "original context" } }); + await boundRetries().retries.promoteDueScheduledRetries(new Date(Date.now() + 31_000)); + const [queued] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successors[0].id)); + expect(queued.status).toBe("queued"); + }); + + it("rotates proof-ineligible waits so later recoverable waits are not starved", async () => { + const f = await computerWaitFixture(); + await db.update(heartbeatRuns).set({ resultJson: { ...f.run.resultJson, computerAdmissionRetryCheckedAt: now.toISOString() } }).where(eq(heartbeatRuns.id, f.run.id)); + await db.insert(heartbeatRuns).values(Array.from({ length: 50 }, () => ({ companyId: f.companyId, agentId: f.agent.id, + status: "cancelled", errorCode: "computer_admission_wait", finishedAt: now, resultJson: {} }))); + await boundRetries().retries.promoteDueScheduledRetries(new Date()); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id))).toHaveLength(0); + await boundRetries().retries.promoteDueScheduledRetries(new Date(Date.now() + 1000)); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id))).toHaveLength(1); + }); + + it.each(["aborted", "suppressed"])("rechecks %s disposition under the source lock after preflight", async outcome => { + const f = await computerWaitFixture(); + const deps = callbacks(); + deps.getAgentInvokability.mockImplementation(async () => { + await db.update(heartbeatRuns).set({ resultJson: { ...f.run.resultJson, computerAdmissionRetryOutcome: outcome } }).where(eq(heartbeatRuns.id, f.run.id)); + return { invokable: true }; + }); + const retry = await extracted.createHeartbeatRetries(db, deps).scheduleBoundedRetryForRun(f.run, f.agent, { now, retryReason: "computer_admission_wait" }); + expect(retry.outcome).toBe("not_scheduled"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id))).toHaveLength(0); + }); + + it.each(["reassigned", "cancelled", "lock-moved"])("does not queue a computer wait after %s", async kind => { + const f = await computerWaitFixture(); + await db.update(issues).set(kind === "reassigned" ? { assigneeAgentId: null } : kind === "cancelled" ? { status: "cancelled" } : { executionRunId: null }).where(eq(issues.id, f.issue.id)); + expect(await boundRetries().retries.scheduleBoundedRetryForRun(f.run, f.agent, { now, retryReason: "computer_admission_wait" })).toMatchObject({ outcome: "not_scheduled" }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.run.id))).toHaveLength(0); + }); + + it.each(["live-controller", "operator-stop", "foreign-computer", "lease", "provider-event", "native", "process", "aborted", "generic-receipt"])("rejects contradictory computer wait evidence: %s", async kind => { + const f = await computerWaitFixture(kind !== "live-controller"); + if (kind === "live-controller") f.run.controllerLeaseExpiresAt = new Date(Date.now() + 60_000); + if (kind === "operator-stop") f.run.errorCode = "operator_interrupted"; + if (kind === "foreign-computer") (f.run.resultJson!.computerAdmission as Record).computerId = randomUUID(); + if (kind === "native") f.run.runtimeMode = "native"; + if (kind === "aborted") f.run.resultJson!.computerAdmissionRetryOutcome = "aborted"; + if (kind === "generic-receipt") { + delete f.run.resultJson!.computerAdmissionPreparationSettledAt; + f.run.resultJson!.startupPreparationSettledAt = now.toISOString(); + f.run.controllerLeaseExpiresAt = null; + } + if (kind === "process") f.run.processPid = 123; + if (kind === "lease") await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: f.environment.id, heartbeatRunId: f.run.id, provider: "boat", providerLeaseId: "old", status: "released", releasedAt: now }); + if (kind === "provider-event") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.run.id, agentId: f.agent.id, seq: 1, eventType: "adapter.invoke", stream: "system" }); + expect(await canRetryComputerAdmissionWait(db, f.run)).toBe(false); + }); + it("reuses one successor across concurrent factories and keeps writes committed before callbacks", async () => { const { run, agent, issue, companyId } = await fixture(); const first = boundRetries(), second = boundRetries(); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index d2e7cb0e5b..ee5d00245a 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,3 +1,5 @@ +import { canRetryComputerAdmissionWait } from "../cancelled-native-startup.js"; +import type { ComputerStopPendingError } from "../../modules/computers/index.js"; import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js"; import { WORKSPACE_VALIDATION_FAILURE_CODE, @@ -22,6 +24,7 @@ import { asc, desc, eq, + gte, inArray, ne, notInArray, @@ -48,6 +51,7 @@ import { withRecoveryContext } from "../recovery/status-only-context.js"; import { MAX_TURN_CONTINUATION_RETRY_REASON, WORKSPACE_BUSY_RETRY_REASON, + COMPUTER_ADMISSION_WAIT_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, @@ -420,6 +424,9 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen opts?.maxAttempts ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS, ), ); + if (retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON && !(await canRetryComputerAdmissionWait(db, run))) { + return { outcome: "not_scheduled" as const, reason: "Computer admission retry requires a settled pre-provider wait.", issueId: run.issueId }; + } const consumedAttempts = executionRetryAttemptCount(run, retryReason); const nextAttempt = consumedAttempts + 1; const computedBaseSchedule = @@ -551,7 +558,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen const requiresIssueGate = isTransientWorkspaceGitScanCode(run.errorCode) || hasConversationContinuationPolicy(run.resultJson) || - (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) || + (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON || retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON) || retryReason === MAX_TURN_CONTINUATION_RETRY_REASON || retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON; if (requiresIssueGate) { @@ -562,6 +569,13 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen now, }); if (!gate.allowed) { + if (retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON && gate.errorCode === "issue_execution_lock_changed") { + // A concurrent scheduler may already have transferred the task lock. + const [existing] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.retryOfRunId, run.id), + )).limit(1); + if (existing) return { outcome: "scheduled" as const, run: existing, reusedExisting: true }; + } await appendRunEvent(run, { eventType: "lifecycle", stream: "system", @@ -615,6 +629,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen retryOfRunId: run.id, wakeReason, retryReason, + ...(retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON + ? { failureRetriesBeforeComputerWait: executionFailureRetryCount(run) } : {}), ...(retryReason === WORKSPACE_BUSY_RETRY_REASON ? { failureRetriesBeforeWorkspaceWait: @@ -707,6 +723,15 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen await tx.execute( sql`select id from heartbeat_runs where company_id = ${run.companyId} and id = ${run.id} for update`, ); + if (retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON) { + const [lockedRun] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), + )); + if (!lockedRun || !(await canRetryComputerAdmissionWait(tx as unknown as Db, lockedRun))) { + return { outcome: "not_scheduled", reason: "Computer admission wait changed before retry admission.", + errorCode: "issue_execution_lock_changed", issueId, details: {} }; + } + } const [existing] = await tx .select() .from(heartbeatRuns) @@ -948,7 +973,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen } if ( - (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && issueId && + (retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON || retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON) && issueId && !isNonAssigneeWorkspaceBusyRetry(retryReason, contextSnapshot) ) { // The issue row is locked above. Recheck after the preflight gate so @@ -966,7 +991,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen const scheduledRunId = randomUUID(); if (contextSnapshot.explicitUserContinuation) { - const continuation = issueId && retryReason === "transient_failure" ? await admitExplicitContinuationRetry({ + const continuation = issueId && (retryReason === "transient_failure" || retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON) ? await admitExplicitContinuationRetry({ db: tx as unknown as Db, companyId: run.companyId, issueId, agentId: run.agentId, parentRunId: run.id, successorRunId: scheduledRunId, now, }) : null; @@ -1365,6 +1390,58 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen .then((rows) => rows[0] ?? null); } + async function finalizeComputerAdmissionDeferral( + run: typeof heartbeatRuns.$inferSelect, + error: ComputerStopPendingError, + wasIssueAssignee: boolean, + ) { + const now = new Date(); + const message = `${error.message} This task will retry automatically.`; + const cancelled = await setRunStatusIfRunning(run.id, "cancelled", { + error: message, errorCode: COMPUTER_ADMISSION_WAIT_RETRY_REASON, finishedAt: now, + resultJson: { + executionRecovery: { kind: "computer_admission_wait", providerWorkStarted: false }, + computerAdmission: error.admission, + cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, + reason: "Waiting for the computer to finish saving", recordedAt: now.toISOString() }, + }, + contextSnapshot: { ...parseObject(run.contextSnapshot), computerAdmissionDeferredWhileAssignee: wasIssueAssignee }, + }); + // An operator Stop or another owner already terminalized this run. + if (!cancelled.updated) return; + await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: now, error: message }).catch(() => undefined); + return cancelled.run; + } + + async function scheduleComputerAdmissionRetry(run: typeof heartbeatRuns.$inferSelect) { + if (!(await canRetryComputerAdmissionWait(db, run))) return; + const agent = await getAgent(run.agentId); + if (!agent) return; + const retry = await scheduleBoundedRetryForRun(run, agent, { + retryReason: COMPUTER_ADMISSION_WAIT_RETRY_REASON, wakeReason: "computer_admission_retry", + maxAttempts: executionRetryAttemptCount(run, COMPUTER_ADMISSION_WAIT_RETRY_REASON) + 1, + delayMs: 30_000, + }); + const authorizationLost = retry.outcome !== "scheduled" && "errorCode" in retry && + retry.errorCode === "continuation_user_authorization_missing"; + if (retry.outcome !== "scheduled") { + const reason = "reason" in retry ? retry.reason : "The retry limit was reached."; + await db.update(heartbeatRuns).set({ + ...(authorizationLost ? { status: "failed", errorCode: "computer_admission_retry_unavailable" } : {}), + error: `Computer admission retry was not scheduled: ${reason}`, + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + ${JSON.stringify({ computerAdmissionRetryOutcome: "suppressed" })}::jsonb`, + }).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled"), + eq(heartbeatRuns.errorCode, COMPUTER_ADMISSION_WAIT_RETRY_REASON))); + await appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: authorizationLost ? "warn" : "info", + message: `Computer admission retry was not scheduled: ${reason}`, payload: { retryScheduled: false } }); + await releaseIssueExecutionAndPromote(run); + } + await finalizeAgentStatus(run.agentId, authorizationLost ? "failed" : "cancelled", null, + { wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run) }); + } + + // Credential rotation can briefly contend with a fresh runtime read. Keep // the task on its automatic pre-provider retry path while the lock clears. async function finalizeAiConnectionBusyDeferral( @@ -1575,6 +1652,19 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen async function promoteDueScheduledRetries(now = new Date()) { const cutoff = await getWorktreeExecutionCutoff(); + // Recover a crash after preparation cleanup but before successor scheduling. + // The scheduler rechecks task ownership and deduplicates under its locks. + const pendingComputerWaits = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.status, "cancelled"), eq(heartbeatRuns.errorCode, COMPUTER_ADMISSION_WAIT_RETRY_REASON), + sql`${heartbeatRuns.resultJson}->>'computerAdmissionRetryOutcome' is null`, + sql`not exists (select 1 from heartbeat_runs successor where successor.company_id = ${heartbeatRuns.companyId} and successor.retry_of_run_id = ${heartbeatRuns.id})`, + ...(cutoff ? [gte(heartbeatRuns.createdAt, cutoff)] : []), + )).orderBy(sql`coalesce(${heartbeatRuns.resultJson}->>'computerAdmissionRetryCheckedAt', '')`, asc(heartbeatRuns.id)).limit(50); + for (const wait of pendingComputerWaits) { + await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || + ${JSON.stringify({ computerAdmissionRetryCheckedAt: now.toISOString() })}::jsonb` }).where(eq(heartbeatRuns.id, wait.id)); + await scheduleComputerAdmissionRetry(wait); + } const result = await runDispatch.promoteDueScheduledRetries({ now, cutoff, @@ -1828,6 +1918,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen scheduleInteractionContinuationInfrastructureRetryIfEligible, findSharedWorkspaceHolder, finalizeAiConnectionBusyDeferral, + finalizeComputerAdmissionDeferral, + scheduleComputerAdmissionRetry, finalizeWorkspaceBusyDeferral, promoteDueScheduledRetries, retryScheduledRetryNow, diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 828ad95155..7d11a8ec6f 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -1,3 +1,4 @@ +import { canRetryComputerAdmissionWait } from "./cancelled-native-startup.js"; import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; import { hasUnrestoredRemoteWorkspace, preserveLegacyWorkspaceRestoreSources, type LegacyWorkspaceRestoreSource } from "./legacy-workspace-restore-recovery.js"; import { isPreDispatchReviewWaitVerified } from "./pre-dispatch-review-wait.js"; @@ -60,7 +61,8 @@ export function legacyExecutionNeedsReconciliation( * The synchronous classifier stays conservative for callers without a DB proof. */ export async function legacyExecutionNeedsReconciliationWithEvidence(db: Db, run: Run): Promise { return await hasUnrestoredRemoteWorkspace(db, run) || - (legacyExecutionNeedsReconciliation(run) && !(await isPreDispatchReviewWaitVerified(db, run))); + (legacyExecutionNeedsReconciliation(run) && !(await isPreDispatchReviewWaitVerified(db, run)) && + !(await canRetryComputerAdmissionWait(db, run))); } /** Persist the failed legacy run, owned lock release and operator decision together. */