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 <noreply@paperclip.ing>

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

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:

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.

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 } },

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,