From d84061b9236855049b360f2bfae07e1d40ae8a0d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 20 Oct 2026 09:09:22 -0500 Subject: [PATCH] fix(runner): recover journaled completion proposals (#16962) ## Thinking Path > - Paperclip manages AI agents and their work. > - Native runs preserve provider sessions across controller restarts. > - A completion event can reach durable storage before the next session checkpoint. > - Recovery advanced the event cursor but lost the saved completion proposal. > - The completed turn then appeared to need another provider turn to report its result. > - This change restores the exact active turn's proposal before advancing the cursor. > - Recovery can finalize completed work without repeating its external actions. ## Linked Issues or Issue Description **What happened?** After a controller restart, a native run could fail with `native_finalization_missing` even though `run.result.proposed` was already durable. The recovered driver skipped that event when its cursor advanced, so the result was absent from its state. Automatic disposition retries could then encounter outstanding tool receipts and exhaust recovery. **Expected behavior** Restore the saved result for the exact active turn. Wait for provider termination, then use normal finalization. Do not start another provider turn to repeat completed work. **Steps to reproduce** 1. Start a native Codex task and save its active-turn checkpoint. 2. Persist its completion proposal, then lose the controller before another checkpoint. 3. Recover the same run after the provider completes. Before this change, finalization reports a missing result. **Paperclip version or commit** Reproduced on `92b238137b`. The repair is based on `7c247243e5`. **Deployment mode** Native Runner with Codex. The incident used a remote sandbox; the regression covers controller event recovery. Related: #12845 and #15297. The older #12241 addresses provider process continuity; this change repairs the controller's saved event projection. ## What Changed - Restore valid completion proposals for the checkpoint's exact active turn during event replay. - Reject conflicting, invalid, or incorrectly bound proposals. Ignore proposals from another turn and prevent adoption after a newer turn starts. - Keep provider termination and normal finalization as separate requirements. - Add eight behavioral cases using the real Codex driver and a simulated transport, and document the recovery contract and verification scope. ## Verification - Red: the new restart regression failed with `native_finalization_missing` before the production change. - Green: 254 tests passed across the native runtime, backend adapter, and Codex recovery suites. The new cases check no repeated provider work, no early finalization, duplicate proposals, conflicts, invalid proposals, and stale turn/session bindings. - `node scripts/check-maps.mjs` and `git diff --check` passed. - `pnpm -r typecheck` and `pnpm build` passed locally. All CI gates passed on `2ac025a15c`; Greptile rated that commit 5/5 with no outstanding findings. The full local `pnpm test:run` was started after completing the embedded PostgreSQL installation, then stopped while still running once the complete CI test matrix passed; it is not reported as a completed local run. - Real local acceptance passed on `2ac025a15c` with the production Rust Runner transport, Codex CLI `0.160.0`, and `gpt-6.1-sol` on macOS arm64. Persisted the completion proposal while retaining the older active-turn checkpoint; let the provider finish without persisting its later controller events; stopped the exact owned runner/provider; recovered the same provider session and turn. Observed one accepted result, one durable `turn.start` command, and one line in the receipt file. Hosted sandbox restart and recovery of the original stopped task have not been performed. - Initial setup failures are retained: dependency drift caused diff-viewer type errors, and skipped PostgreSQL postinstall links prevented database startup. Dependency alignment and the package postinstall repaired those local prerequisites. Probe setup iterations failed on a direct-provider response schema, invalid probe input fields, and an original provider that still held its writer lock. Acceptance uses the production transport, validated input, and exact-owner shutdown. No production code changes were made for those probe setup failures. ## Risks - A stale proposal could finalize the wrong work. Recovery requires the checkpoint's active turn and matching run, session, and source. It rejects conflicting results and waits for provider termination. - The change uses the existing paginated recovery scan. It adds no database query per event or new retry allowance. - No schema migration or change to task authorization is included. ## Model Used OpenAI GPT-6, running as Codex with reasoning, tool use, and code execution. The exact served model snapshot and context-window size are not exposed in this session. The live acceptance probe uses `gpt-6.1-sol`. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/execution-semantics.md | 8 ++ maps/agent-access-map/recovery.md | 8 ++ maps/feature-map/recovery.md | 8 ++ .../src/native-session-runtime.test.ts | 113 ++++++++++++++++++ .../src/native-session-runtime.ts | 43 ++++++- 5 files changed, 176 insertions(+), 4 deletions(-) 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,