fix: defer Boat admission while provider stop is pending

This commit is contained in:
Dotta committed 2026-10-10 22:07:06 -05:00
1 parent 5308f15032
commit 04ce7eb2a8
20 files changed
+422 -30

No files matched your search

@@ -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();
@@ -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<T>(
scope: Scope,
change: (record: ComputerRecord) => T,
pendingStopRunId?: string,
): Promise<T> {
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) {
@@ -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");
+1 -1
View File
@@ -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,
@@ -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,
@@ -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 };
@@ -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,
});
@@ -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<string, unknown>,
): 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) &&
+1
View File
@@ -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,
@@ -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<boolean> {
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<string, string>;
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;
}
@@ -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 });
@@ -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)}`,
@@ -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);
@@ -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<string, unknown> | null; runnerProfileJson?: Record<string, unknown> | null;
}): boolean {
const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
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;
}
@@ -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<string, unknown>), 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 });
@@ -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) ||
+31
View File
@@ -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) &&
+106 -5
View File
@@ -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<Run> = {
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<string, unknown>).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();
+95 -3
View File
@@ -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,
@@ -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<boolean> {
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. */