diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 94a47a39638..aea7fb9bd80 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1625,6 +1625,14 @@ controller must recover it through the authenticated ownership checks. This preserves active work and queued messages without treating a server restart as user cancellation. +Recovery restores a structured result proposal from the durable run events when +its append survived but the active turn's session checkpoint did not. Advancing +the replay cursor must preserve that proposal for the exact run, session, source, +and turn. Conflicting or invalid proposals fail closed. A proposal from another +turn cannot complete the current turn, and a later started turn prevents adoption +of the older proposal. The provider must still supply terminal evidence before +normal finalization; recovery must not repeat completed work to obtain its result. + When a native turn creates a governed approval wait, its checkpoint retains the exact committed triggering event and yielded disposition before interrupting the provider. Shutdown gives these turns a shared, bounded 20-second window to finish diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index bd9283471a0..0801a2bb062 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -69,6 +69,14 @@ prove successful recovery. ## `bounded-continuation` +Controller recovery can restore the exact active turn's saved completion proposal +before reconnecting its provider. It requires matching run/session/source/turn +identity and subsequent provider termination. This adds no agent retry, completion, +or recovery-resolution authority. The +[runtime regressions](../../packages/paperclip-runner/src/native-session-runtime.test.ts) +cover the driver boundary with simulated transport; live hosted recovery remains +separate qualification. + Review: `baseline`. API: `present` — representative operation IDs: diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 4e12774c949..0e0a0bfd1ca 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -134,6 +134,14 @@ and rejected identity evidence. The Runner restart tests require positive exit evidence before replacement. This coverage does not prove recovery of a historical retained sandbox or behavior during a live provider outage. +For completion recovery, restart after a native completion proposal is appended +but before its session checkpoint. Verify that the original result is finalized +after provider termination, with no new provider turn or repeated external action. +The [runtime regressions](../../packages/paperclip-runner/src/native-session-runtime.test.ts) +exercise the real Codex driver with a simulated transport, including a still-active +provider and invalid, conflicting, duplicate, or stale proposals. This does not +qualify a hosted sandbox restart. + The same database suite pauses an original preparation while recovery claims the remote run. It then verifies that the old preparation yields without opening a provider or changing the recovery owner, for both live and expired leases. diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 68ab9d278d8..8bc9af6512f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -1,6 +1,8 @@ import { createHash } from "node:crypto"; import { resolveQualifiedAcpxProfile } from "./drivers/acpx/qualified-profiles.js"; import { describe, expect, it, vi } from "vitest"; +import { createCodexNativeSessionBackend } from "./backends/codex-native-backend.js"; +import { FakeCodexTransport, WORKSPACE } from "./drivers/codex/codex-app-server-driver.test-support.js"; import type { ControlPlanePort } from "./contracts/control-plane-port.js"; import type { NativeExecutionInputV1, NativeExecutionInputV5 } from "./contracts/native-execution.js"; @@ -226,6 +228,117 @@ function highestContiguous(events: PrpEvent[]): number { } describe("executeNativeSession recovery", () => { + it.each([ + "completed", "pending", "duplicate", "conflicting", "other turn", "newer turn", + "wrong session", "invalid", + ])("recovers a journaled completion proposal after restart: %s", async (scenario) => { + const originalTransport = new FakeCodexTransport(); + const recoveredTransport = new FakeCodexTransport(); + const transports = [originalTransport, recoveredTransport]; + const executionInput = { ...input, workspace: { ...input.workspace, cwd: WORKSPACE } }; + const backend = createCodexNativeSessionBackend(executionInput, { + runnerInstanceId: "runner-recovery", + transportFactory: () => transports.shift()!, + }); + const original = await backend.openSession({ identity, workingDirectory: WORKSPACE }); + const journal: PrpEvent[] = []; + const iterator = original.events()[Symbol.asyncIterator](); + const drainThrough = async (eventType: string) => { + while (true) { + const next = await iterator.next(); + if (next.done) throw new Error(`stream closed before ${eventType}`); + journal.push(structuredClone(next.value)); + if (next.value.eventType === eventType) return; + } + }; + await original.startTurn({ message: { role: "user", text: "Finish the assigned review." } }); + originalTransport.push("turn/started", { + threadId: "thread-1", turn: { id: "turn-1", status: "inProgress" }, + }); + await drainThrough("turn.started"); + const checkpoint = await original.snapshot(); + expect(checkpoint.semanticResult).toBeNull(); + originalTransport.push("paperclip/runResult", { + threadId: "thread-1", turnId: "turn-1", result, + }); + await drainThrough("run.result.proposed"); + expect((await original.snapshot()).semanticResult).toEqual(result); + const proposal = journal.at(-1)!; + if (scenario === "other turn") proposal.turnId = "prior-turn"; + if (scenario === "wrong session") proposal.normalizedSessionId = "other-session"; + if (scenario === "invalid") proposal.payload = { reportedWorkDisposition: "done" }; + if (["duplicate", "conflicting", "newer turn"].includes(scenario)) { + journal.push({ + ...proposal, sourceSeq: proposal.sourceSeq + 1, + sourceEventId: `${proposal.sourceEventId}:next`, + ...(scenario === "newer turn" + ? { eventType: "turn.started" as const, turnId: "newer-turn", payload: {} } + : { payload: scenario === "conflicting" ? { ...result, summary: "A different result." } : result }), + }); + } + const persistedProposalCount = journal.filter(event => event.eventType === "run.result.proposed").length; + await original.close({ reason: "controller disconnected after persisting the proposal" }); + recoveredTransport.readResponse = { thread: { + id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "turn-1", status: scenario === "pending" ? "inProgress" : "completed", items: [ + { id: "answer", type: "agentMessage", text: "The review was published." }, + ] }], + } }; + const completeRun = vi.fn(async () => undefined); + const checkpoints: PersistedNativeSession[] = []; + const port: ControlPlanePort = { + async openRun() {}, + async loadSessionCheckpoint() { return structuredClone(checkpoint); }, + async checkpointSession(value) { checkpoints.push(structuredClone(value)); }, + async appendEvent(event) { + const existing = journal.find(value => value.sourceEventId === event.sourceEventId); + if (existing) expect(event).toEqual(existing); + else journal.push(structuredClone(event as PrpEvent)); + return { cursor: journal.length, highestContiguousSourceSeq: event.sourceSeq, + disposition: existing ? "duplicate" : "committed" }; + }, + async replayEvents(query) { + const events = journal.filter(event => event.sourceInstanceId === query.sourceInstanceId); + return { + events: structuredClone(events.filter(event => event.sourceSeq > query.afterSourceSeq).slice(0, query.limit)), + highestContiguousSourceSeq: highestContiguous(events), + }; + }, + completeRun, + }; + const execution = executeNativeSession({ + input: executionInput, backend, controlPlane: port, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + }); + if (["conflicting", "wrong session", "invalid"].includes(scenario)) { + await expect(execution).rejects.toBeInstanceOf(NativeSessionProtocolIntegrityError); + expect(completeRun).not.toHaveBeenCalled(); + expect(recoveredTransport.calls).toEqual([]); + return; + } + if (["other turn", "newer turn"].includes(scenario)) { + await expect(execution).rejects.toThrow("native_finalization_missing"); + expect(completeRun).not.toHaveBeenCalled(); + expect(recoveredTransport.calls.filter(call => call.method === "turn/start")).toEqual([]); + return; + } + if (scenario === "pending") { + await vi.waitFor(() => expect(checkpoints[0]?.semanticResult).toEqual(result)); + expect(completeRun).not.toHaveBeenCalled(); + recoveredTransport.push("turn/completed", { + threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] }, + }); + } + await expect(execution).resolves.toMatchObject({ result, turnId: "turn-1", terminal }); + expect(completeRun).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ result, turnId: "turn-1" }), + expect.anything(), + ); + expect(recoveredTransport.calls.filter(call => call.method === "turn/start")).toEqual([]); + expect(journal.filter(event => event.eventType === "run.result.proposed")).toHaveLength(persistedProposalCount); + expect(checkpoints.at(-1)).toMatchObject({ semanticResult: result, activeTurnId: null }); + }); + it.each([ { label: "continues a restart-lost Codex turn", allowed: true, error: { code: "provider_turn_lost_on_restore", recoverable: true } }, { label: "requires the control plane restart claim", allowed: false, error: { code: "provider_turn_lost_on_restore", recoverable: true } }, diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index c11c446aa63..e7875ec63a9 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1566,7 +1566,7 @@ function checkpointCursor(cursor: string | null | undefined): number { return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0; } -async function reconcileRecoveryCursor(input: { +async function reconcileRecoveryCheckpoint(input: { controlPlane: ControlPlanePort; checkpoint: PersistedNativeSession; runId: string; @@ -1576,6 +1576,8 @@ async function reconcileRecoveryCursor(input: { const checkpointHighWater = checkpointCursor(input.checkpoint.cursor); let afterSourceSeq = checkpointHighWater; let persistedHighWater = checkpointHighWater; + let proposalTurnId = input.checkpoint.activeTurnId ?? null; + let recoveredResult: PrpStructuredRunResult | null = null; while (true) { const replay = await input.controlPlane.replayEvents( { @@ -1588,6 +1590,34 @@ async function reconcileRecoveryCursor(input: { ); input.signal.throwIfAborted(); if (replay.events.length === 0) break; + for (const event of replay.events) { + if (event.eventType === "turn.started" && event.turnId !== proposalTurnId) { + proposalTurnId = null; + recoveredResult = null; + } + if ( + proposalTurnId && + event.turnId === proposalTurnId && + event.eventType === "run.result.proposed" + ) { + const validation = validatePrpStructuredRunResult(event.payload); + if ( + !validation.ok || + event.runId !== input.runId || + event.normalizedSessionId !== input.checkpoint.identity.sessionId || + event.sourceInstanceId !== input.sourceInstanceId + ) { + throw new NativeSessionProtocolIntegrityError("source_event_replay_conflict"); + } + const previous = recoveredResult ?? input.checkpoint.semanticResult; + const normalize = (result: PrpStructuredRunResult) => + canonicalJson(input.controlPlane.normalizeReplayPayload?.(result) ?? result); + if (previous && normalize(previous) !== normalize(validation.result)) { + throw new NativeSessionProtocolIntegrityError("source_event_replay_conflict"); + } + recoveredResult = validation.result; + } + } const pageHighWater = replay.events.reduce( (highest, event) => Math.max(highest, event.sourceSeq), afterSourceSeq, @@ -1600,11 +1630,16 @@ async function reconcileRecoveryCursor(input: { } if ( persistedHighWater === checkpointHighWater && - input.checkpoint.cursor === String(checkpointHighWater) + input.checkpoint.cursor === String(checkpointHighWater) && + recoveredResult === null ) { return input.checkpoint; } - return { ...input.checkpoint, cursor: String(persistedHighWater) }; + return { + ...input.checkpoint, + cursor: String(persistedHighWater), + ...(recoveredResult === null ? {} : { semanticResult: recoveredResult }), + }; } async function replayCheckpointedTurnTerminal(input: { @@ -2036,7 +2071,7 @@ export async function executeNativeSession( timeoutMs: recoveryTimeoutMs, timeoutMessage: `native session recovery replay timed out after ${recoveryTimeoutMs}ms`, operation: (signal) => - reconcileRecoveryCursor({ + reconcileRecoveryCheckpoint({ controlPlane: options.controlPlane, checkpoint: recoveryCheckpoint, runId: input.binding.runId,