diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index b67247d57a4..dd961a60dc2 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -469,6 +469,9 @@ import { stageRemoteRunnerFile, steerNativeSession, syncRemoteRunnerDirectoryOut, + isTransientRemoteCheckpointReadFailure, + remoteProviderStateUnreadable, + retryTransientRemoteCheckpointRead, verifyNativeHarnessBackup, shouldRestoreNativeHarnessBackupIntoSandbox, } from "./native-session-executor.js"; @@ -2312,8 +2315,14 @@ describe("remote provider pack manifest", () => { return JSON.stringify(value); }; - it.each(["opencode", "opencode-new-durable", "opencode-continuation", "opencode-continuation-missing", "opencode-continuation-no-receipt", "opencode-continuation-replacement", "opencode-continuation-cold", "opencode-continuation-legacy-durable", "opencode-dirty", "opencode-diagnostics", "opencode-diagnostics-stale", "cursor", "claude-x64", "claude-arm64"] as const)("uses the bundled Linux image authority for remote %s without controller overrides", async provider => { - const firstDurable = provider === "opencode-new-durable"; + it.each(["opencode", "opencode-new-durable", "opencode-new-durable-reset", "opencode-new-durable-inspect-reset", "opencode-new-durable-provider-reset", "opencode-continuation", "opencode-continuation-missing", "opencode-continuation-no-receipt", "opencode-continuation-replacement", "opencode-continuation-cold", "opencode-continuation-legacy-durable", "opencode-dirty", "opencode-diagnostics", "opencode-diagnostics-stale", "cursor", "claude-x64", "claude-arm64"] as const)("uses the bundled Linux image authority for remote %s without controller overrides", async provider => { + const firstDurable = provider.startsWith("opencode-new-durable"); + // A sandbox connection drops once while the settled state is read out. + let checkpointReadResets = provider === "opencode-new-durable-reset" ? 1 : 0; + // Or while the checkpoint inspects the runner state, or reads the provider state, before any copy starts. + let inspectionResets = 0; + const inspectionReset = provider === "opencode-new-durable-inspect-reset" ? "runner-state.json" + : provider === "opencode-new-durable-provider-reset" ? "codex-provider-state.json" : null; const architecture = provider === "claude-arm64" ? "arm64" : "x64"; const claude = provider.startsWith("claude-"); const opencode = provider.startsWith("opencode"); @@ -2372,6 +2381,10 @@ describe("remote provider pack manifest", () => { const suspendedState = durableRunnerState({ runId: remoteExecution.binding.runId, runnerInstanceId: "bundled-image-runner", environmentLeaseId: "lease", normalizedSessionId: remoteExecution.session.normalizedSessionId }, "suspended"); const syncOut = vi.fn(async (operations: Array<{ files: Array<{ kind: string; targetPath: string }> }>) => { + if (checkpointReadResets > 0 && operations.some(operation => operation.files.some(file => file.targetPath.includes("/.pending-")))) { + checkpointReadResets--; + throw Object.assign(new Error("write ECONNRESET"), { code: "ECONNRESET" }); + } for (const operation of operations) for (const file of operation.files) { if (file.kind === "directory") { await mkdir(file.targetPath, { recursive: true }); @@ -2402,7 +2415,14 @@ describe("remote provider pack manifest", () => { }); onTestFinished(() => { storage.mockRestore(); owner.mockRestore(); validate.mockRestore(); retain.mockRestore(); capture.mockRestore(); backup.mockRestore(); }); } - const execute = vi.fn(async (command: { command: string; args?: string[] }) => ({ + const execute = vi.fn(async (command: { command: string; args?: string[] }) => { + const script = command.args?.[1] ?? ""; + if (inspectionResets > 0 && inspectionReset && script.includes("base64 <") && script.includes(inspectionReset) + && (inspectionReset === "runner-state.json" || !script.includes("runner-state.json"))) { + inspectionResets--; + throw Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }); + } + return { exitCode: 0, signal: null, timedOut: false, stderr: "", stdout: command.args?.[0] === "--build-metadata" ? JSON.stringify({ schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", @@ -2412,7 +2432,8 @@ describe("remote provider pack manifest", () => { }) : firstDurable && command.args?.[1]?.startsWith("const os=require") ? JSON.stringify({ home: "/home/runtime-account", uid: 1000 }) : firstDurable && command.args?.[1]?.includes("base64 <") ? Buffer.from(JSON.stringify(command.args[1].includes("codex-provider-state.json") && !command.args[1].includes("runner-state.json") ? providerState : suspendedState)).toString("base64") : "", - })); + }; + }); try { let coldAdmission: ReturnType = null; const prepare = createRunnerdBackend({ db: preparedDb, execution: remoteExecution, @@ -2492,8 +2513,11 @@ describe("remote provider pack manifest", () => { expect(exportSources.every(source => source.startsWith("/workspace/"))).toBe(true); expect(providerArtifactStorage.retainNativeProviderArtifacts).toHaveBeenCalledWith(expect.objectContaining({ owner: expect.objectContaining({ runId: remoteExecution.binding.runId }), source: expect.objectContaining({ manifest: installed, launchRoot, accountUid: 1000 }) })); + if (inspectionReset) inspectionResets = 1; await driver.mock.calls.at(-1)![0].checkpoint!("settled"); expect(harnessStorage.persistNativeHarnessBackup).toHaveBeenCalledTimes(1); + expect(checkpointReadResets).toBe(0); + expect(inspectionResets).toBe(0); return; } expect(execute.mock.calls.some(([call]) => call.args?.[0] === "-e" || call.args?.[0] === "--version")).toBe(false); @@ -21223,3 +21247,32 @@ describe("HTTP Stop during remote runner staging", () => { } finally { release(); await outcome; await stopping; await bundle.transport.detachControllerForRestart?.().catch(() => {}); await temporary.cleanup(); await rm(root, { recursive: true, force: true }); } }, 120000); }); + +describe("isTransientRemoteCheckpointReadFailure", () => { + it("repeats only a dropped sandbox connection, never a state or safety refusal", () => { + expect(isTransientRemoteCheckpointReadFailure(Object.assign(new Error("write ECONNRESET"), { code: "ECONNRESET" }))).toBe(true); + expect(isTransientRemoteCheckpointReadFailure(new Error("Daytona syncOut directory download failed: socket hang up"))).toBe(true); + expect(isTransientRemoteCheckpointReadFailure(new Error("runner_harness_state_mismatch: checkpoint_identity_incomplete"))).toBe(false); + expect(isTransientRemoteCheckpointReadFailure(new Error("Daytona syncOut refusing tarball member that escapes the extraction dir: ../x"))).toBe(false); + expect(isTransientRemoteCheckpointReadFailure(Object.assign(new Error("certificate expired"), { code: "CERT_HAS_EXPIRED" }))).toBe(false); + // A provider-state refusal is transient only when the read behind it dropped. + expect(isTransientRemoteCheckpointReadFailure(remoteProviderStateUnreadable(Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" })))).toBe(true); + expect(isTransientRemoteCheckpointReadFailure(remoteProviderStateUnreadable(new Error("runner_remote_provider_state_unavailable")))).toBe(false); + expect(isTransientRemoteCheckpointReadFailure(remoteProviderStateUnreadable(new SyntaxError("Unexpected token")))).toBe(false); + expect(remoteProviderStateUnreadable(new Error("read ECONNRESET")).message).toBe("runner_harness_state_mismatch: provider_state_unreadable"); + }); + + it("stops after three reads, and never repeats a refusal", async () => { + const reset = () => Object.assign(new Error("read ECONNRESET"), { code: "ECONNRESET" }); + const onRetry = vi.fn(); + const dropping = vi.fn(async () => { throw reset(); }); + await expect(retryTransientRemoteCheckpointRead(dropping, onRetry, 0)).rejects.toThrow("read ECONNRESET"); + expect(dropping).toHaveBeenCalledTimes(3); + expect(onRetry.mock.calls).toEqual([[1, 3], [2, 3]]); + const refusing = vi.fn(async () => { throw new Error("runner_harness_state_mismatch: runner_identity"); }); + await expect(retryTransientRemoteCheckpointRead(refusing, vi.fn(), 0)).rejects.toThrow("runner_identity"); + expect(refusing).toHaveBeenCalledTimes(1); + let drops = 2; + await expect(retryTransientRemoteCheckpointRead(async () => { if (drops-- > 0) throw reset(); return "settled"; }, vi.fn(), 0)).resolves.toBe("settled"); + }); +}); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index fb5c533b658..cc2d29953b0 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -295,6 +295,7 @@ import { } from "./status-arbiter.js"; import { conflict, HttpError } from "../../errors.js"; import { redactSensitiveText } from "../../redaction.js"; +import { classifyRemoteConnectionError } from "../remote-connection-failure.js"; import { resolvePaperclipRunnerBinary, resolveRemotePaperclipRunnerBinary } from "./native-codex-runner.js"; import { createNativeRunTrace, @@ -13735,6 +13736,55 @@ function assertSafeExtractedCheckpoint(root: string): void { } } +const REMOTE_CHECKPOINT_CAPTURE_ATTEMPTS = 3; +const REMOTE_CHECKPOINT_CAPTURE_RETRY_MS = 1_000; +const TRANSIENT_REMOTE_READ_MESSAGE = /\b(?:ECONNRESET|EPIPE|ETIMEDOUT|ECONNREFUSED|socket hang up|fetch failed|no response returned)\b/; +/** `provider_state_unreadable` refusals whose underlying read was a dropped + * connection. The refusal keeps its message for every other caller; only the + * checkpoint read retry looks through it. */ +const providerStateReadTransportFailures = new WeakMap(); + +/** A dropped or refused sandbox connection while reading settled state. A + * mismatch, a missing directory or an unsafe archive is never transient. */ +export function isTransientRemoteCheckpointReadFailure(error: unknown): boolean { + if (error instanceof Error && providerStateReadTransportFailures.has(error)) { + return isTransientRemoteCheckpointReadFailure(providerStateReadTransportFailures.get(error)); + } + const reason = classifyRemoteConnectionError(error); + if (reason !== null) return reason !== "tls_failure"; + return error instanceof Error && !error.message.startsWith("runner_") + && TRANSIENT_REMOTE_READ_MESSAGE.test(error.message); +} + +/** The refusal `inspectRemoteHarnessState` throws when the provider state + * cannot be read, remembering the read failure behind it. */ +export function remoteProviderStateUnreadable(readFailure: unknown): Error { + const refusal = new Error("runner_harness_state_mismatch: provider_state_unreadable"); + providerStateReadTransportFailures.set(refusal, readFailure); + return refusal; +} + +/** + * Repeats a read of settled state out of the sandbox after a dropped + * connection, at most {@link REMOTE_CHECKPOINT_CAPTURE_ATTEMPTS} times in all. + * Any other failure, and the last transient one, is thrown unchanged. + */ +export async function retryTransientRemoteCheckpointRead( + read: () => Promise, + onRetry: (attempt: number, attempts: number) => Promise | void, + retryDelayMs = REMOTE_CHECKPOINT_CAPTURE_RETRY_MS, +): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await read(); + } catch (error) { + if (attempt >= REMOTE_CHECKPOINT_CAPTURE_ATTEMPTS || !isTransientRemoteCheckpointReadFailure(error)) throw error; + await onRetry(attempt, REMOTE_CHECKPOINT_CAPTURE_ATTEMPTS); + await new Promise((resolveDelay) => setTimeout(resolveDelay, retryDelayMs * attempt)); + } + } +} + /** * Copy a remote runner directory into durable local state. * @@ -15754,8 +15804,8 @@ async function createRunnerdBackendWithinSessionClaim( stateDirectory: remoteStateDirectory, execution: input.execution, }); - } catch { - throw new Error("runner_harness_state_mismatch: provider_state_unreadable"); + } catch (error) { + throw remoteProviderStateUnreadable(error); } const providerSessionIdentity = providerSessionIdentityFromDurableProviderState({ @@ -16301,14 +16351,23 @@ async function createRunnerdBackendWithinSessionClaim( // `runner_harness_state_mismatch`. Only checkpoint a harness that runnerd // has proved complete. A malformed or identity-conflicting state still // throws from inspectRemoteHarnessState and therefore fails closed. - let checkpointable = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState()); + // A dropped sandbox connection repeats the read (bounded) instead of + // losing the only backup of the provider session. + const inspectForCheckpoint = () => retryTransientRemoteCheckpointRead( + () => measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState()), + (attempt, attempts) => input.onLog?.( + "stderr", + `[paperclip-runner] remote checkpoint inspection interrupted (attempt ${attempt} of ${attempts}); reading the settled state again\n`, + ), + ); + let checkpointable = await inspectForCheckpoint(); // The remote runner writes its suspended lifecycle and provider state // before the outer process-owner RPC necessarily observes completion. // Allow a very small bounded visibility window without ever accepting an // active, incomplete, malformed, or identity-conflicting checkpoint. for (let attempt = 1; !checkpointable.complete && attempt < 3; attempt++) { await new Promise((resolve) => setTimeout(resolve, 100)); - checkpointable = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState()); + checkpointable = await inspectForCheckpoint(); } if (!checkpointable.complete) { const incompleteFailure = remoteCheckpointIncompleteFailure( @@ -16338,24 +16397,27 @@ async function createRunnerdBackendWithinSessionClaim( input.trace, "harness_state.backup.persist", async () => { - const verified = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState()); - if ( - !verified.complete || - !verified.runnerState || - !verified.providerSessionIdentity - ) { - throw new Error("runner_harness_state_mismatch: checkpoint_identity_incomplete"); - } - const providerSessionIdentity = verified.providerSessionIdentity; - const backupRoot = harnessBackupRoot(root); mkdirSync(backupRoot, { recursive: true, mode: 0o700 }); const pendingRoot = resolve( backupRoot, `.pending-${randomUUID()}`, ); - mkdirSync(pendingRoot, { recursive: true, mode: 0o700 }); - try { + // Reading the settled state out of the sandbox writes nothing + // durable, so a dropped sandbox connection can repeat the whole + // read. Without this, one reset loses the only copy of the + // provider session when the per-turn sandbox is released. + const captureSettledState = async () => { + rmSync(pendingRoot, { recursive: true, force: true }); + const verified = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState()); + if ( + !verified.complete || + !verified.runnerState || + !verified.providerSessionIdentity + ) { + throw new Error("runner_harness_state_mismatch: checkpoint_identity_incomplete"); + } + mkdirSync(pendingRoot, { recursive: true, mode: 0o700 }); for (const directory of persistenceProfile.directories) { const sourcePath = remotePersistencePath(directory); if (!sourcePath) @@ -16372,6 +16434,16 @@ async function createRunnerdBackendWithinSessionClaim( throw new Error("runner_harness_state_mismatch: checkpoint_directory_missing"); } } + return verified.providerSessionIdentity; + }; + try { + const providerSessionIdentity = await retryTransientRemoteCheckpointRead( + captureSettledState, + (attempt, attempts) => input.onLog?.( + "stderr", + `[paperclip-runner] remote checkpoint read interrupted (attempt ${attempt} of ${attempts}); reading the settled state again\n`, + ), + ); const manifest = buildNativeHarnessBackupManifest({ backupRoot: pendingRoot, execution: input.execution,