diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts index 9837fa2003d..f2b0ef4fda2 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.terminal-replay.test.ts @@ -157,4 +157,55 @@ describe("Codex driver terminal-turn history replay", () => { events.find((event) => event.eventType === "session.failed")?.payload, ).toMatchObject({ code: "turn_binding_mismatch" }); }); + + it("records a restore-lost turn this session never bound without failing its own turn", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-unbound-restore-loss", + normalizedSessionId: "normalized-unbound-restore-loss", + workingDirectory: WORKSPACE, + }); + await session.startTurn({ message: { role: "user", text: "Continue." } }); + const lost = { + threadId: "thread-1", + turn: { id: "turn-uncheckpointed", status: "failed", items: [], + error: { code: "provider_turn_lost_on_restore", message: "lost", recoverable: true } }, + }; + transport.push("turn/completed", lost); + // A replay of the same report stays ignored. + transport.push("turn/completed", lost); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-1", status: "completed", items: [] }, + }); + const events = await collectUntilTerminal(session.events()); + expect(events.find((event) => event.eventType === "session.failed")).toBeUndefined(); + expect(events.filter((event) => event.eventType === "turn.failed")).toHaveLength(0); + expect(events.find((event) => event.eventType === "turn.completed")?.turnId).toBe("turn-1"); + expect(events.find((event) => event.eventType === "harness.diagnostic" + && event.payload.classification === "unbound_restored_turn_lost")?.payload) + .toMatchObject({ receivedTurnId: "turn-uncheckpointed", expectedTurnId: "turn-1" }); + }); + + it.each([ + ["an ordinary failure", { status: "failed", error: { code: "provider_failed", recoverable: false } }], + ["a completion", { status: "completed" }], + ])("still rejects %s for a turn this session never bound", async (_label, terminal) => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-unbound-terminal", + normalizedSessionId: "normalized-unbound-terminal", + workingDirectory: WORKSPACE, + }); + await session.startTurn({ message: { role: "user", text: "Continue." } }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { id: "turn-unknown", items: [], ...terminal }, + }); + const events: PrpEvent[] = []; + for await (const event of session.events()) events.push(event); + expect( + events.find((event) => event.eventType === "session.failed")?.payload, + ).toMatchObject({ code: "turn_binding_mismatch" }); + }); }); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 190bb1ab5db..26c79d213cf 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -4,7 +4,7 @@ import { classifyCodexNotification } from "./codex-notification-identity.js"; import { paperclipWorkspaceFileReferencesFromText } from "../../live/workspace-file-reference.js"; import { canonicalProviderEventsFromCodex, isCanonicalProviderEventType } from "../../provider-events.js"; import { harnessRuntimeRequestOutcome } from "../../contracts/harness-driver.js"; -import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; +import { isNativeRestartInterruption, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js"; import { isPrpAuthCredentialsUpdated, validatePrpStructuredRunResult } from "../../protocol/replay-contract.js"; import type { CodexRpcNotification, CodexTraceInterpretation } from "./app-server-transport.js"; import { redactCodexDiagnostic } from "./app-server-transport.js"; @@ -543,6 +543,7 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex mapTerminalTurn(state, turn, turnId); return; } + if (ignoreUnboundRestoredTurnLoss(state, turn, turnId)) return; if (!state.notificationNamesActiveTurn(turnId, "turn terminal")) return; mapTerminalTurn(state, turn, turnId); return; @@ -691,6 +692,41 @@ function withCompactionSizes( return { ...payload, preTokens, postTokens }; } +/** + * A controller can die after the provider accepted a turn but before that + * turn reached its checkpoint. The restored runner then reports the turn as + * lost on restore, while this session never bound it. The runner emits that + * cause only after it proved the old provider process ended and restored the + * same conversation, so the turn can produce no further work or result. It is + * not this session's turn: record it as restored history and keep the session + * usable for its own (continuation) turn instead of failing the protocol. + */ +function ignoreUnboundRestoredTurnLoss( + state: CodexSessionState, + turn: Record, + turnId: string, +): boolean { + if ( + turnId.length === 0 || + state.protocolFailed || + turnId === state.activeTurnId || + state.terminalTurns.has(turnId) || + text(turn.status) !== "failed" || + !isNativeRestartInterruption(turn.error) + ) { + return false; + } + state.restoredTerminalTurns.add(turnId); + state.emit("harness.diagnostic", { + code: "provider_notification_identity", + method: "turn/completed", + classification: "unbound_restored_turn_lost", + expectedTurnId: state.activeTurnId, + receivedTurnId: turnId.slice(0, 256), + }); + return true; +} + function ignoreRestoredTerminalReplay( state: CodexSessionState, method: string, diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index e16292d7641..98b2200236b 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1485,6 +1485,26 @@ it("defers turn starts until their command result and rejects stale identities", ).toBe("accept"); }); +it("does not hold a pending turn start behind another turn's restore-lost terminal", () => { + const { runnerdTerminalProviderTurnId: terminalTurn, terminalMayBelongToPendingTurnStart: mayBelong } = + runnerdRecoveryInternals; + // The restored runner reports the previous, unbound turn as lost after the + // new turn.start result named the new provider turn. + const lostTurn = terminalTurn("turn.failed", { + providerTurnId: "provider-turn-previous", status: "failed", + error: { code: "provider_turn_lost_on_restore", recoverable: true }, + }); + expect(lostTurn).toBe("provider-turn-previous"); + expect(mayBelong({ expectedProviderTurnId: "provider-turn-new", terminalProviderTurnId: lostTurn })).toBe(false); + // Its own terminal, an unnamed terminal, or any terminal before the command + // result stays held until turn.accepted is exposed. + expect(mayBelong({ expectedProviderTurnId: "provider-turn-new", terminalProviderTurnId: "provider-turn-new" })).toBe(true); + expect(mayBelong({ expectedProviderTurnId: "provider-turn-new", terminalProviderTurnId: null })).toBe(true); + expect(mayBelong({ expectedProviderTurnId: null, terminalProviderTurnId: lostTurn })).toBe(true); + expect(terminalTurn("provider.event", { events: [{ method: "turn/completed", + params: { turn: { id: "provider-turn-raw", status: "failed" } } }] })).toBe("provider-turn-raw"); +}); + it("requires ACPX command results to preserve the requested turn identity", () => { expect( runnerdRecoveryInternals.turnStartCommandResultValid({ diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 503ef9cf878..bdfd28d8734 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -709,6 +709,36 @@ function turnStartResponseReady(input: { ); } +/** The provider turn a durable terminal names, when it names one explicitly. */ +function runnerdTerminalProviderTurnId(eventType: string, payload: unknown): string | null { + const named = (value: unknown) => typeof value === "string" && value.length > 0 ? value : null; + if (eventType !== "provider.event") return named(record(payload).providerTurnId); + const terminal = unwrapRunnerdProviderNotifications(payload).find( + (notification) => notification.method === "turn/completed", + ); + const params = record(terminal?.params); + return named(record(params.turn).id) ?? named(params.turnId); +} + +/** + * A terminal held while turn/start is pending keeps it behind the new turn's + * accepted response. Once the durable command result names the new provider + * turn, a terminal that explicitly names another turn (for example, a turn the + * restored runner reports as lost on restore) cannot be the new turn's + * terminal. Holding it would also hold the turn/started event behind it, so + * the turn start could never be bound. + */ +function terminalMayBelongToPendingTurnStart(input: { + expectedProviderTurnId: string | null; + terminalProviderTurnId: string | null; +}): boolean { + return ( + input.expectedProviderTurnId === null || + input.terminalProviderTurnId === null || + input.terminalProviderTurnId === input.expectedProviderTurnId + ); +} + function turnStartNotificationDisposition(input: { responsePending: boolean; expectedProviderTurnId: string | null; @@ -7053,7 +7083,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { (event.eventType === "provider.event" && unwrapRunnerdProviderNotifications(eventPayload).some( (notification) => notification.method === "turn/completed", - ))); + ))) && + terminalMayBelongToPendingTurnStart({ + expectedProviderTurnId: this.#expectedProviderTurnId, + terminalProviderTurnId: runnerdTerminalProviderTurnId(event.eventType, eventPayload), + }); if (terminalWhileTurnStartPending) { if (!fromDeferredQueue) { this.#eventSourceSeq = event.sourceSeq; @@ -8251,6 +8285,8 @@ export const runnerdRecoveryInternals = Object.freeze({ runnerCloseGraceMs, rotatedRunAttachPayload, rotateExternalAuthorityEpoch, + runnerdTerminalProviderTurnId, + terminalMayBelongToPendingTurnStart, turnStartCommandResultValid, turnStartNotificationDisposition, turnStartResponseReady, diff --git a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts index efc2e95833b..1b31aa945cd 100644 --- a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -123,3 +123,88 @@ it.each(["none", "before-start", "after-start"] as const)("completes after the r await rm(root, { recursive: true, force: true }); } }, 30_000); + +it("starts a new turn when the restored runner reports a lost turn the checkpoint never bound", async () => { + // The controller died right after the provider accepted its turn, before the + // accepted turn reached the checkpoint. The restored runner then reports that + // unbound turn as lost while the recovery's new turn start is pending. + const root = await mkdtemp(join(tmpdir(), "runner-restart-unbound-")); + const identity = { runId: "run-unbound", sessionId: "session-unbound", companyId: "company-unbound", issueId: "issue-unbound", agentId: "agent-unbound" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-unbound" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-2", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-unbound", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn"], + stateDirectory: root, + turnStartTimeoutMs: 8_000, + prpIdentity: { runnerInstanceId: "runner-unbound", environmentLeaseId: "lease-unbound", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-unbound", itemId: "item-unbound" }, + }; + const first = createRunnerdCodexTransport(options); + let restored: ReturnType | undefined; + const alive = (pid: number | null) => { + if (!pid) return false; + try { process.kill(pid, 0); return true; } catch { return false; } + }; + const kill = (pid: number | null) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + }; + try { + const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-unbound", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + // The last checkpoint the controller persisted: no active turn yet. + const checkpoint = await session.snapshot(); + expect(checkpoint.activeTurnId ?? null).toBeNull(); + await session.startTurn({ message: { role: "user", text: execution.task.prompt } }); + const owner = first.evidence(); + await first.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")); + providerState.activeTurnId = null; + await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState)); + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-unbound", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const completion = await executeNativeSession({ + input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-unbound", controlPlaneInstanceId: "controller-unbound", timeoutMs: 15_000, + controlPlane: { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } }); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + // The unbound lost turn is not this execution's turn: it is not reported as its failure. + expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(0); + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2); + } finally { + await restored?.transport.close().catch(() => undefined); + for (const bundle of [first, restored]) { + if (!bundle) continue; + const evidence = bundle.evidence(); + kill(evidence.runnerPid); + kill(evidence.providerPid); + } + await rm(root, { recursive: true, force: true }); + } +}, 40_000);