diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 54c2e54e616..596b3a21c51 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1753,7 +1753,13 @@ The runtime's own cleanup after a failed turn start or a failed event stream (for example `runnerd did not report the provider turn identity` or the execution timeout) revokes the turn's output but is not a Stop, so it keeps the Runner for the same-run reattachment too. A protocol integrity fault, which -needs an operator, still stops the Runner. +needs an operator, still stops the Runner. A reattachment after a restart whose recovery read +times out or loses its transport (for example +`PRP command session.snapshot timed out`) detaches from the Runner the same way a server restart does: it sends no turn +stop and no suspend and does not signal the process. The Runner and its +in-flight turn stay for the next same-run reattachment. An abort (a Stop or the +recovery deadline), any identity or authority refusal, any other recovery +failure, and a protocol integrity fault still stop it. A remote native run can lose its original runner or sandbox before reattachment. This happens after a server restart, and for a same-run retry after a turn ends diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index c8b15fffb34..66b3d7a96ba 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -75,18 +75,34 @@ const DRIVER_KIND = "codex_app_server"; const DRIVER_VERSION = "codex-v2"; interface BootstrapCancellation { - close(): Promise; + /** `keepRunner` detaches this controller from a durable Runner exactly as a + * server restart does: no turn stop, no suspend, no signal. The Runner and + * its in-flight turn stay for the same run's next reattachment. Only the + * first close decides. */ + close(keepRunner?: boolean): Promise; detach(): void; wait(operation: Promise): Promise; } +/** A recovery failure that says nothing about the Runner's identity or state: + * a command or transport timed out or was lost. Any identity or authority + * refusal, even one that also mentions a timeout, is not transient. */ +export function isTransientRunnerRecoveryFailure(error: unknown): boolean { + if (error instanceof NativeSessionProtocolIntegrityError) return false; + const message = error instanceof Error ? error.message : String(error); + if (/mismatch|different|identity|authority|integrity|incompatible/i.test(message)) return false; + return /provider_transport_failed|PRP command \S+ timed out|runnerd transport closed while waiting/.test(message); +} + function bootstrapCancellation( transport: CodexAppServerTransport, signal: AbortSignal | undefined, ): BootstrapCancellation { let closePromise: Promise | null = null; - const close = (): Promise => { - closePromise ??= transport.close(); + const close = (keepRunner = false): Promise => { + closePromise ??= keepRunner && transport.detachControllerForRestart + ? transport.detachControllerForRestart() + : transport.close(); return closePromise; }; let rejectAborted!: (reason: unknown) => void; @@ -453,6 +469,12 @@ export class CodexAppServerDriver implements HarnessDriver { activeTurnId: snapshot.activeTurnId, }, }); + // A recovery whose Runner command timed out or lost its transport detaches + // from the Runner, as a restart does, for the same run's next + // reattachment: the Runner may still own an in-flight turn, and stopping + // or killing it loses the turn. An abort (a Stop or the recovery + // deadline), identity refusals, other recovery failures and protocol + // integrity faults still close and stop it. const cancellation = bootstrapCancellation(transport, options.signal); try { await cancellation.wait(this.#persistProcessOwnership(transport)); @@ -772,7 +794,7 @@ export class CodexAppServerDriver implements HarnessDriver { session, }; } catch (error) { - await cancellation.close().catch(() => {}); + await cancellation.close(isTransientRunnerRecoveryFailure(error)).catch(() => {}); if (error instanceof NativeSessionProtocolIntegrityError) throw error; if (options.signal.aborted) options.signal.throwIfAborted(); // A Runner with no saved control-plane state for this session can never diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts index 845eec9d141..133d8005941 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts @@ -250,6 +250,73 @@ describe("Codex app-server Codex driver", () => { }); expect(second.calls.some((call) => call.method === "turn/start" || call.method === "thread/goal/set")).toBe(false); }); + // Staging: a server replacement left the remote Runner alive mid-turn. The + // new controller reattached, verified the same process, and its recovery + // read failed with "PRP command session.snapshot timed out". The failure + // cleanup then killed the live Runner, so the same-run reattachment found + // the process gone and the in-flight turn was lost. + describe("a recovery that fails on its way to a live remote Runner", () => { + async function recoverWith(fail: (method: string) => Promise | null, signal?: AbortSignal) { + const originalTransport = new FakeCodexTransport(); + const recoveryTransport = new FakeCodexTransport(); + const request = recoveryTransport.request.bind(recoveryTransport); + vi.spyOn(recoveryTransport, "request").mockImplementation(async (method, params) => + fail(method) ?? request(method, params)); + const close = vi.spyOn(recoveryTransport, "close"); + // The runnerd transport's restart detach: no turn stop, no suspend, no signal. + const detach = vi.fn(async () => {}); + Object.assign(recoveryTransport, { detachControllerForRestart: detach }); + const driver = makeDriver([originalTransport, recoveryTransport]); + const original = await driver.openSession({ + runId: "run-runner-kept", normalizedSessionId: "session-runner-kept", workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Work." } }); + const checkpoint = await original.snapshot(); + await original.close({ reason: "controller detached for restart" }); + const recovery = driver.recoverSession!(checkpoint, { signal: signal ?? new AbortController().signal }); + return { recovery, close, detach, recoveryTransport }; + } + const rejectRead = (error: Error) => (method: string) => method === "thread/read" ? Promise.reject(error) : null; + + it.each([ + ["the runner snapshot timed out", new Error("provider_transport_failed: PRP command session.snapshot timed out")], + ["the history read failed behind a snapshot timeout (the staging message)", new HarnessReconciliationError( + "codex_history_read_failed: thread/turns/list requires supported paginated history; Error: provider_transport_failed: PRP command session.snapshot timed out")], + ])("detaches and keeps the Runner and its turn when %s", async (_title, error) => { + const { recovery, close, detach, recoveryTransport } = await recoverWith(rejectRead(error)); + expect(await recovery).toMatchObject({ recovered: false, reason: expect.stringContaining("session.snapshot timed out") }); + expect(detach).toHaveBeenCalledTimes(1); + expect(close).not.toHaveBeenCalled(); + expect(recoveryTransport.calls.some(call => ["turn/start", "thread/start", "turn/interrupt"].includes(call.method))).toBe(false); + }); + + it("still stops the Runner when an abort (a Stop or the recovery deadline) ends a hung read", async () => { + const controller = new AbortController(); + const { recovery, close, detach } = await recoverWith(method => { + if (method !== "thread/read") return null; + setTimeout(() => controller.abort(new Error("Stop requested")), 0); + return new Promise(() => {}); + }, controller.signal); + await expect(recovery).rejects.toThrow("Stop requested"); + expect(detach).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledTimes(1); + expect(close.mock.calls[0]).toEqual([]); + }); + + it.each([ + ["a thrown different-session refusal", new HarnessReconciliationError("thread/read returned a different driver session")], + ["an adopted provider identity mismatch", new Error("native_adopted_provider_identity_mismatch")], + ["an identity refusal that also timed out", new Error("provider_transport_failed: PRP command session.snapshot timed out; native_adopted_provider_identity_mismatch")], + ["an unclassified failure", new Error("codex_history_incomplete: thread/turns/list omitted data")], + ])("still stops the Runner after %s", async (_title, error) => { + const { recovery, close, detach } = await recoverWith(rejectRead(error)); + expect(await recovery).toMatchObject({ recovered: false }); + expect(detach).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledTimes(1); + expect(close.mock.calls[0]).toEqual([]); + }); + }); + it.each([ "initial-read", "reconcile-read", @@ -302,6 +369,8 @@ describe("Codex app-server Codex driver", () => { }), ).rejects.toBe(fault); expect(close).toHaveBeenCalledTimes(1); + // An integrity fault still stops the Runner. + expect(close.mock.calls[0]).toEqual([]); expect( recoveryTransport.calls.some((call) => call.method === "thread/start"), ).toBe(false);