diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 05dfcebf525..6407370d1ba 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1745,6 +1745,11 @@ this, the close stopped the Runner mid-turn, so the reattachment always found it gone with an unsettled journal (`runner_remote_process_identity_unavailable`) and the task waited for Retry. A Stop, a cancelled turn and a `stop_cleanup` recovery still stop the Runner, and a Runner that settled or exited is handled as before. +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. 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/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index cd2d6975156..33f017f8e0f 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -125,17 +125,20 @@ describe("HarnessDriverBackend", () => { }); it.each([ - { cancelled: false, retainUnsettledRemoteRunner: true }, - { cancelled: true, retainUnsettledRemoteRunner: false }, - ])("asks to retain an unsettled remote Runner on close only when the run was not cancelled (cancelled=$cancelled)", - async ({ cancelled, retainUnsettledRemoteRunner }) => { + { cancelled: false, failureCleanup: false, retainUnsettledRemoteRunner: true }, + { cancelled: true, failureCleanup: false, retainUnsettledRemoteRunner: false }, + { cancelled: true, failureCleanup: true, retainUnsettledRemoteRunner: true }, + ])("asks to retain an unsettled remote Runner on close only when the run was not stopped (cancelled=$cancelled, failureCleanup=$failureCleanup)", + async ({ cancelled, failureCleanup, retainUnsettledRemoteRunner }) => { const close = vi.fn(async () => {}); const session = Object.assign(new FakeHarnessSession(), { close }); const backend = new HarnessDriverBackend({ ...driver, openSession: async () => session }); const opened = await backend.openSession({ identity: { runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1", } }); - if (cancelled) await cancelSession(opened, { reason: "Stop requested", signal: new AbortController().signal }).cleanup; + if (cancelled) await cancelSession(opened, { reason: failureCleanup ? "Native session event consumption failed." : "Stop requested", + signal: new AbortController().signal, ...(failureCleanup ? { failureCleanup } : {}) }).cleanup; + if (failureCleanup) expect(await opened.result()).toBeNull(); await opened.close({ reason: "native session execution complete" }); expect(close).toHaveBeenCalledExactlyOnceWith({ reason: "native session execution complete", retainUnsettledRemoteRunner }); }); diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index e10724b1eba..e3922fb36b9 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -404,6 +404,7 @@ class HarnessNativeSession implements NativeSession { readonly #session: HarnessSession; #terminal: PrpTerminalState | null = null; #explicitlyCancelled = false; + #stopRequested = false; #protocolIntegrityFailure: NativeSessionProtocolIntegrityError | null = null; #assertProtocolIntegrity(): void { @@ -525,6 +526,7 @@ class HarnessNativeSession implements NativeSession { this.#input = { ...this.#input, identity: structuredClone(input.identity) }; this.#terminal = null; this.#explicitlyCancelled = false; + this.#stopRequested = false; } async release(): Promise { @@ -780,6 +782,7 @@ class HarnessNativeSession implements NativeSession { revokeTurnPublication() { this.#explicitlyCancelled = true; + this.#stopRequested = true; } settleCancelledTurn(input: { turnId: string; reason: string }) { @@ -788,7 +791,7 @@ class HarnessNativeSession implements NativeSession { return this.#withProtocolIntegrity(() => this.#session.settleCancelledTurn?.(input) ?? null); } - cancel(input: { reason: string; signal: AbortSignal }) { + cancel(input: { reason: string; signal: AbortSignal; failureCleanup?: boolean }) { if (input.signal.aborted) { throw ( input.signal.reason ?? new Error("native session cancellation aborted") @@ -797,7 +800,8 @@ class HarnessNativeSession implements NativeSession { // This flag is the adapter's synchronous publication boundary. Provider // interruption happens afterward as passive cleanup, so a slow or broken // transport cannot synthesize or publish new accepted output for the turn. - this.revokeTurnPublication(); + if (input.failureCleanup === true) this.#explicitlyCancelled = true; + else this.revokeTurnPublication(); const interrupt = this.#session.interrupt; return { ...(this.terminalDrainTimeoutMs === undefined ? {} : { terminalDrainTimeoutMs: this.terminalDrainTimeoutMs }), @@ -922,7 +926,7 @@ class HarnessNativeSession implements NativeSession { } close(input: { reason: string }) { - return this.#session.close({ ...input, retainUnsettledRemoteRunner: !this.#explicitlyCancelled }); + return this.#session.close({ ...input, retainUnsettledRemoteRunner: !this.#stopRequested && this.#protocolIntegrityFailure === null }); } } diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 0167561d1e2..647186f2691 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -270,6 +270,7 @@ export interface NativeSession { cancel?(input: { reason: string; signal: AbortSignal; + failureCleanup?: boolean; }): NativeSessionCancellation; settleCancelledTurn?(input: { turnId: string; reason: string }): Promise; resolveRuntimeRequest?(input: { diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 8da0c171821..6811cdfd7bc 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -5961,6 +5961,7 @@ describe("executeNativeSession recovery", () => { expect(cancel).toHaveBeenCalledWith({ reason: "Native session event consumption failed.", signal: expect.any(AbortSignal), + failureCleanup: true, }); expect(providerResult).not.toHaveBeenCalled(); expect(events.map((event) => event.eventType)).toEqual([ @@ -10973,3 +10974,62 @@ describe("first-turn queued input", () => { expect((JSON.parse(sent[0]!) as { task: { prompt: string } }).task.prompt).toBe(input.task.prompt); }); }); + +describe("remote Runner retention after a failed run", () => { + const port = (): ControlPlanePort => ({ + async openRun() {}, async checkpointSession() {}, async completeRun() {}, + async appendEvent(event: PrpEvent) { + return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + }); + const startCodexRun = (transport: FakeCodexTransport, options: { timeoutMs?: number; + onSession?: (session: NativeSession | null) => void } = {}) => { + const close = vi.spyOn(transport, "close"); + const executionInput = { ...input, workspace: { ...input.workspace, cwd: WORKSPACE } }; + const backend = createCodexNativeSessionBackend(executionInput, { + runnerInstanceId: "runner-recovery", transportFactory: () => transport, + }); + const execution = executeNativeSession({ input: executionInput, backend, controlPlane: port(), + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + timeoutMs: options.timeoutMs ?? 5_000, requireSessionCloseBeforeReturn: true, onSession: options.onSession }); + void execution.catch(() => undefined); + return { execution, close }; + }; + const retainedOn = (close: { mock: { calls: unknown[][] } }): Array => + close.mock.calls.map((call: unknown[]) => + (call[1] as { retainUnsettledRemoteRunner?: boolean } | undefined)?.retainUnsettledRemoteRunner); + + it("keeps the Runner for same-run reattachment when the turn start fails", async () => { + const transport = new FakeCodexTransport(); + transport.rejectMethods.set("turn/start", new Error("runnerd did not report the provider turn identity")); + const { execution, close } = startCodexRun(transport); + await expect(execution).rejects.toThrow("runnerd did not report the provider turn identity"); + expect(retainedOn(close)).not.toHaveLength(0); + expect(retainedOn(close).every(retained => retained === true)).toBe(true); + }); + + it("keeps the Runner for same-run reattachment when event consumption fails", async () => { + const transport = new FakeCodexTransport(); + const { execution, close } = startCodexRun(transport, { timeoutMs: 100 }); + await vi.waitFor(() => expect(transport.calls.some(call => call.method === "turn/start")).toBe(true)); + transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress" } }); + await expect(execution).rejects.toThrow("native session timed out after 100ms"); + expect(retainedOn(close)).not.toHaveLength(0); + expect(retainedOn(close).every(retained => retained === true)).toBe(true); + }); + + it("still stops the Runner after a Stop, even when the stopped run then fails", async () => { + const transport = new FakeCodexTransport(); + let published: NativeSession | null = null; + const { execution, close } = startCodexRun(transport, { timeoutMs: 300, + onSession: session => { published ??= session; } }); + await vi.waitFor(() => expect(transport.calls.some(call => call.method === "turn/start")).toBe(true)); + transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress" } }); + await vi.waitFor(() => expect(published).not.toBeNull()); + void published!.cancel!({ reason: "Stop requested", signal: new AbortController().signal }).cleanup.catch(() => undefined); + await expect(execution).rejects.toThrow(); + expect(retainedOn(close)).not.toHaveLength(0); + expect(retainedOn(close).every(retained => retained === false)).toBe(true); + }); +}); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 55cb68b4a67..f3bded13bf3 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -482,6 +482,7 @@ function terminalFromEvent( async function attemptOptionalSessionCancellation( session: NativeSession, reason: string, + failure: unknown, ): Promise<{ settlement: Promise[]> } | null> { if (session.cancel === undefined) return null; const cancellationAbort = new AbortController(); @@ -493,6 +494,7 @@ async function attemptOptionalSessionCancellation( cleanup = session.cancel({ reason, signal: cancellationAbort.signal, + ...(failure instanceof NativeSessionProtocolIntegrityError ? {} : { failureCleanup: true }), }).cleanup; } catch { return null; @@ -1629,6 +1631,7 @@ async function consumeTurn( const deferredCancellation = await attemptOptionalSessionCancellation( session, "Native session event consumption failed.", + error, ); deferredSessionCancellationSettlement = deferredCancellation?.settlement ?? null;