fix(native): keep restart continuations alive through early results and repeat restarts (#19699)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs continue a turn lost to a control-plane restart once, in the same conversation (#15297, #19501) > - In restart storm 3, two Codex runs each crossed one restart and still needed a Retry > - The continuation was attempted, but the driver failed it with `turn_binding_mismatch`, and the run then reported "runnerd exited before provider turn startup" > - A second restart that loses the continuation turn itself also needed a Retry, because the one-shot marker was already used > - A restart should not cost a click when continuing is safe, and a failure should name its real cause > - This pull request fixes the start race, reports the driver's own failure, and gives each admitted restart its own continuation > - The benefit is that interrupted tasks finish without a Retry, within the existing recovery budget ## Linked Issues or Issue Description Refs #19501, #15297, #19658, #19539 **What happened?** Restart storm 3 (seed 7103, RC06 mask on, load 9–12). Codex B5 Retry runs crossed one K1 restart mid-turn. After the restart, recovery resumed the dead runner. The runner restored the conversation and recorded `turn.failed` with `provider_turn_lost_on_restore` (`recoverable: true`). The control plane started the same-conversation continuation (`resumeInterruptedTurn` → `startTurn`). About 1 s later the transport closed with `protocol_failure:turn_binding_mismatch`. The run failed as `native_runner_process_exited: runnerd exited before provider turn startup`. The task needed another Retry. I reproduced it with the focused cell (`--workloads B5 --faults K1 --triggers retry`) and a temporary diagnostic. The failure is "Provider semantic result did not name the active turn": a `paperclip/runResult` reached the Codex driver while the continuation's `turn/start` was still pending and no turn was bound. The driver waits for a pending start before it handles `turn/completed`, but not before it handles `paperclip/runResult`. **Expected behavior** After a restart that loses a turn, the run continues automatically when that is safe. Each admitted restart attempt can continue once, within the provider-attempt budget. A failure names its real cause. **Steps to reproduce** 1. Run `node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --adapters codex --workloads B5 --faults K1 --triggers retry --tasks 2 --concurrency 2 --seed 4040 --rc06-masked`. 2. When a lost turn's continuation receives a semantic result before its start settles, the run fails with `turn_binding_mismatch` and the task needs a Retry. The race is intermittent: 1 of about 14 lost turns in my runs. **Root cause** - `packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts`: the `paperclip/runResult` branch checked the active-turn binding without waiting for `state.turnStartSettled`. The `turn/completed` branch already waits. - `codex-harness-session.ts` `startTurn`: the driver's own protocol failure closed the transport. The pending start then reported the runner exit, which hid the cause. - `codex-session-terminal.ts` `mapTerminalTurn`: a consumed one-shot marker blocked every later continuation in the run, even when the newly lost turn was that continuation itself. The one-continuation allowance is per run, not per issue or session lineage: attaching a new run resets the marker. The storm runs did attempt their continuation; the start race made it fail. ## What Changed - Codex driver: a `paperclip/runResult` waits for a pending turn start to settle before it is bound, like a terminal. - Codex driver: when this session's own protocol failure ends a pending turn start, `startTurn` throws `codex_turn_start_protocol_failure: <code>: <message>` instead of the runner exit. CodexRpc errors and protocol-integrity errors keep their own types. The server now classifies this as `native_session_interrupted` instead of `native_runner_process_exited`; both are retryable. - A later restart that loses the continuation turn itself re-arms the one-shot marker: the driver resets it as it was before the first continuation. It does this when it maps the lost terminal, including during recovery from a checkpoint taken while the continuation ran. The checkpoint gate `nativeRestartInterruptedTurnId` stays strict: a consumed marker still means the allowance is used. Each continuation still needs an admitted dead-runner restart. The existing provider-attempt budget bounds the repeats, and an exhausted budget still blocks with Retry. - `doc/execution-semantics.md`: an additive paragraph. ## Verification Red before the fix, green after: - `codex-app-server-driver.events.test.ts` "binds a semantic result that arrives before its pending turn start settles". Red: `session.failed`. Green: the result binds to `turn-1` and the turn completes. - `codex-app-server-driver.events.test.ts` "names its own protocol failure when that failure ends a pending turn start". Red: "runnerd exited before provider turn startup". Green: `codex_turn_start_protocol_failure: turn_binding_mismatch`. - `codex-app-server-driver.recovery.test.ts` "offers another restart continuation when a later restart loses the continuation turn itself". Red: `null`. Green: `turn-2`. A checkpoint with a consumed marker still returns `null`. - `codex-app-server-driver.recovery.test.ts` "re-arms the continuation when recovery finds the checkpointed continuation turn lost". This recovers from a checkpoint taken during the continuation, then starts the next continuation. Red without the re-arm (the marker stays consumed); green with it. - `native-runner-restart-recovery.integration.test.ts`: the same checkpoint (continuation active, consumed marker naming it) is claimed as `resume_dead_runner`. This is regression coverage. Suites run locally on the final code: - Runner: Codex driver, `native-session-runtime`, `runnerd-restart-continuation` and contracts (745 tests). - Server: `native-runner-restart-recovery.integration.test.ts`. - `tsc` for the runner tests and the server. - A few real-process tests timed out at about 6 s while six live agents ran. Each passed when rerun alone. The settlement suite passed both with and without this change at the same load. Live (local, real Codex, macOS arm64), focused restart-chaos cell `--adapters codex --workloads B5 --faults K1 --triggers retry,time --tasks 6 --seed 4042 --rc06-masked`: 6/6 crossed a restart and passed. R1 passed for all 6, so each continued without a Retry. ## Risks - A run can now continue more than once in the same conversation, once per admitted restart. Each continuation asks the agent to reconcile unfinished actions before repeating them. The provider-attempt budget still caps the repeats. - A semantic result that arrives during a pending start now waits for that start. If the start fails, the result is still rejected by the existing binding check. - The start race was intermittent (1 of about 14 lost turns), so the live cell shows absence of the failure, not a forced repro. The driver regression reproduces it deterministically. - No schema, API or migration change. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use, code execution, and extended reasoning. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 01:50:13 +01:00 · 2026-11-06 19:55:18 -06:00
1 parent 11c07c622c
commit c37bc749ac
@@ -2572,6 +2572,15 @@ result that exact turn proposed, recovery also settles the turn with that result
 accepted result instead of failing and rerunning finished work. A missing turn
 without such a result still fails closed, because its outcome is unknown.

+Each admitted restart attempt that loses a turn this way gets its own continuation.
+When a later restart loses the continuation turn itself, its outcome is also
+unknown, so the marker re-arms and the next admitted attempt continues again. The
+provider-attempt budget bounds the repeats; an exhausted budget still blocks with
+Retry. A semantic result that reaches the driver before its turn start settles
+waits for that start, like a terminal, instead of failing the new turn. If this
+session's own protocol failure ends a pending turn start, the run reports
+`codex_turn_start_protocol_failure` with the cause instead of a runner exit.
+
 Provider execution and control-plane finalization have different clocks. A healthy provider can think or execute a long tool without output. Once execution settles, recovery and finalization control steps have a 60-second deadline, checked on startup and every 15 seconds. With a healthy database and scheduler, an abandoned transition must be repaired or surfaced within 90 seconds. Terminal persistence must not wait on provider cleanup or publication; a late finalizer cannot change a reassigned or closed task or release another run's locks. In particular, a done report from a run whose task now belongs to another agent, a user or nobody is kept as an accepted result but records `completion_owner_superseded` and preserves the task status for the current owner. Historical ambiguous runs are never automatically replayed after an upgrade.

 When that deadline fails a native run with a prepared workspace, the same
@@ -103,6 +103,8 @@ export function nativeRestartInterruptedTurnId(snapshot: Pick<PersistedNativeSes
 >): string | null {
  // Runnerd records this cause for Codex app-server and ACPX sessions after
  // proving the old provider lifetime ended and restoring the conversation.
+  // A lost continuation re-arms the driver's marker before it is checkpointed,
+  // so a consumed marker here still means the allowance is used.
  if (!RESTART_CONTINUATION_DRIVER_KINDS.has(snapshot.driverKind ?? "") || snapshot.activeTurnId != null ||
      snapshot.semanticResult != null || snapshot.dispositionOnlyRecoveryConsumed ||
      snapshot.goal != null || snapshot.pendingRuntimeRequests?.length) return null;
@@ -46,6 +46,51 @@ import {
 import { RUNNERD_RUNTIME_ACTIVITY } from "./codex-driver-values.js";

 describe("Codex app-server Codex driver", () => {
+  it("binds a semantic result that arrives before its pending turn start settles", async () => {
+    // Runnerd can deliver the new turn's semantic result before turn/start's
+    // response binds the turn (as after a restart continuation). Like a
+    // terminal, the result waits for the start to settle instead of failing
+    // the session as an unbound result.
+    const transport = new FakeCodexTransport();
+    let releaseStart!: (value: Record<string, unknown>) => void;
+    transport.turnStartResponse = new Promise(resolve => { releaseStart = resolve; });
+    const session = await makeDriver([transport]).openSession({
+      runId: "early-result", normalizedSessionId: "early-result-session", workingDirectory: WORKSPACE,
+    });
+    const starting = session.startTurn({ message: { role: "user", text: "Finish." } });
+    await new Promise(resolve => setTimeout(resolve, 0));
+    transport.push("paperclip/runResult", { threadId: "thread-1", turnId: "turn-1", result });
+    await new Promise(resolve => setTimeout(resolve, 0));
+    releaseStart({ turn: { id: "turn-1", status: "inProgress", items: [] } });
+    await expect(starting).resolves.toMatchObject({ turnId: "turn-1" });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } });
+    const events = await collectUntilTerminal(session.events());
+    expect(events.some(event => event.eventType === "session.failed")).toBe(false);
+    expect(events.at(-1)?.eventType).toBe("turn.completed");
+    expect(events.filter(event => event.eventType === "run.result.proposed"))
+      .toEqual([expect.objectContaining({ turnId: "turn-1" })]);
+    await session.close({ reason: "fixture complete" });
+  });
+
+  it("names its own protocol failure when that failure ends a pending turn start", async () => {
+    // The protocol failure closes the transport, so runnerd exits and the
+    // pending start reports a runner exit. The run must keep the real cause.
+    const transport = new FakeCodexTransport();
+    let rejectStart!: (error: Error) => void;
+    transport.turnStartResponse = new Promise((_resolve, reject) => { rejectStart = reject; });
+    const session = await makeDriver([transport]).openSession({
+      runId: "start-protocol-failure", normalizedSessionId: "start-protocol-failure-session", workingDirectory: WORKSPACE,
+    });
+    const starting = session.startTurn({ message: { role: "user", text: "Finish." } });
+    await new Promise(resolve => setTimeout(resolve, 0));
+    transport.push("item/started", { threadId: "thread-1", turnId: "unbound-turn", item: { id: "item-1", type: "agentMessage" } });
+    await new Promise(resolve => setTimeout(resolve, 0));
+    rejectStart(new Error("runnerd exited before provider turn startup"));
+    await expect(starting).rejects.toThrow(/codex_turn_start_protocol_failure: turn_binding_mismatch/);
+    await expect(starting).rejects.not.toThrow(/runnerd exited/);
+    await session.close({ reason: "fixture complete" }).catch(() => undefined);
+  });
+
  it("rejects provider-forged and previous-run activity notifications", async () => {
    const transport = new FakeCodexTransport();
    const session = await makeDriver([transport]).openSession({
@@ -42,7 +42,7 @@ import {
  type PrpEvent,
  type PrpStructuredRunResult,
 } from "./codex-app-server-driver.test-support.js";
-import { NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";
+import { NativeSessionProtocolIntegrityError, nativeRestartInterruptedTurnId } from "../../contracts/native-session-backend.js";

 function cursorProviderIdentity(mode: unknown) {
  return {
@@ -62,6 +62,66 @@ function cursorProviderIdentity(mode: unknown) {
 }

 describe("Codex app-server Codex driver", () => {
+  it("offers another restart continuation when a later restart loses the continuation turn itself", async () => {
+    const transport = new FakeCodexTransport();
+    const session = await makeDriver([transport]).openSession({
+      runId: "run-two-restarts", normalizedSessionId: "session-two-restarts", workingDirectory: WORKSPACE,
+    });
+    const lost = { code: "provider_turn_lost_on_restore", recoverable: true };
+    await session.startTurn({ message: { role: "user", text: "Finish the request." } });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", error: lost, items: [] } });
+    await collectUntilTerminal(session.events());
+    expect(nativeRestartInterruptedTurnId({ ...await session.snapshot(), semanticResult: null })).toBe("turn-1");
+    // The first restart's continuation is accepted, then a second restart
+    // loses it before any result.
+    transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+    await session.startTurn({ message: { role: "user", text: "Continue." }, continuation: true });
+    transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "failed", error: lost, items: [] } });
+    await collectUntilTerminal(session.events());
+    const snapshot = await session.snapshot();
+    // The lost continuation re-arms the marker exactly like the first loss.
+    expect(snapshot.dispositionOnlyRecoveryConsumed).toBe(false);
+    expect(nativeRestartInterruptedTurnId({ ...snapshot, semanticResult: null })).toBe("turn-2");
+    // A consumed marker still means the allowance is used.
+    expect(nativeRestartInterruptedTurnId({ ...snapshot, semanticResult: null, dispositionOnlyRecoveryConsumed: true })).toBeNull();
+    await session.close({ reason: "fixture complete" });
+  });
+
+  it("re-arms the continuation when recovery finds the checkpointed continuation turn lost", async () => {
+    // Controller lost during the first continuation: the checkpoint names the
+    // continuation as the active turn and as the consumed marker's turn.
+    const lost = { code: "provider_turn_lost_on_restore", recoverable: true };
+    const first = new FakeCodexTransport();
+    const second = new FakeCodexTransport();
+    const driver = makeDriver([first, second]);
+    const original = await driver.openSession({
+      runId: "run-lost-continuation", normalizedSessionId: "session-lost-continuation", workingDirectory: WORKSPACE,
+    });
+    await original.startTurn({ message: { role: "user", text: "Finish the request." } });
+    first.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", error: lost, items: [] } });
+    await collectUntilTerminal(original.events());
+    first.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
+    await original.startTurn({ message: { role: "user", text: "Continue." }, continuation: true });
+    const checkpoint = await original.snapshot();
+    expect(checkpoint).toMatchObject({ activeTurnId: "turn-2", dispositionOnlyRecoveryConsumed: true,
+      dispositionOnlyRecoveryTurnId: "turn-2" });
+    await original.close({ reason: "controller lost during the continuation" });
+    second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [
+      { id: "turn-1", status: "failed", error: lost, items: [] },
+      { id: "turn-2", status: "failed", error: lost, items: [] },
+    ] } };
+    const recovery = await driver.recoverSession!(checkpoint);
+    expect(recovery).toMatchObject({ recovered: true });
+    const recovered = recovery.session!;
+    const restored = await recovered.snapshot();
+    expect(restored.dispositionOnlyRecoveryConsumed).toBe(false);
+    expect(nativeRestartInterruptedTurnId({ ...restored, semanticResult: null })).toBe("turn-2");
+    second.turnStartResponse = Promise.resolve({ turn: { id: "turn-3", status: "inProgress", items: [] } });
+    await expect(recovered.startTurn({ message: { role: "user", text: "Continue." }, continuation: true }))
+      .resolves.toMatchObject({ turnId: "turn-3" });
+    await recovered.close({ reason: "fixture complete" });
+  });
+
  it.each(["agent", "plan", "ask"])("retains native Cursor %s mode through checkpoint and recovery", async (mode) => {
    const identity = cursorProviderIdentity(mode);
    const first = new FakeCodexTransport("thread-1", "provider-session-1", identity);
@@ -1,7 +1,7 @@
 import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from "./codex-history.js";
 import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js";
 import { randomUUID } from "node:crypto";
-import { isNativeRestartInterruption, NativeProviderTerminalFailure } from "../../contracts/native-session-backend.js";
+import { isNativeRestartInterruption, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";

 import type {
  HarnessGoalOperation,
@@ -325,6 +325,14 @@ export class CodexHarnessSession
          this.terminal = true;
        }
      }
+      if (this.protocolFailed && !(error instanceof CodexRpcError) && !(error instanceof NativeSessionProtocolIntegrityError)) {
+        // This session's own protocol failure closed the transport, so the
+        // start then reports a lost runner. Keep the cause that ended it.
+        throw new Error(
+          `codex_turn_start_protocol_failure: ${this.protocolFailureCode}: ${this.protocolFailureMessage}`,
+          { cause: error },
+        );
+      }
      throw error;
    } finally {
      this.turnStartPending = false;
@@ -449,6 +449,10 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
      return;
    }
    if (notification.method === "paperclip/runResult") {
+      // Like a terminal, a semantic result can reach the notification channel
+      // before turn/start's response binds its turn (for example, a restart
+      // continuation). Bind it only after that start settles.
+      await state.turnStartSettled;
      if (
        threadId !== state.opened.threadId
        || !state.notificationNamesActiveTurn(turnId, "semantic result")
@@ -175,9 +175,16 @@ export function mapTerminalTurn(
    state.emit(eventType, payload, { turnId });
    finalize(state, status);
    if (status === "failed" && isNativeRestartInterruption(turn.error) && state.result === null &&
-        !state.dispositionOnlyRecoveryConsumed) {
+        (!state.dispositionOnlyRecoveryConsumed || state.dispositionOnlyRecoveryTurnId === turnId)) {
      // Orchestration may use the paid restart attempt for one same-thread
      // continuation. Its accepted turn id uses the existing durable marker.
+      // A later restart that loses that continuation turn itself leaves its
+      // outcome unknown too, so the next admitted restart attempt may
+      // continue again; the provider attempt budget bounds the repeats.
+      // Reset the marker so that continuation is submitted, checkpointed and
+      // adopted after a crash exactly like the first one.
+      state.dispositionOnlyRecoveryConsumed = false;
+      state.dispositionOnlyRecoveryTurnId = null;
      state.terminal = false;
      state.dispositionOnlyRecoveryAvailable = true;
    }
@@ -2509,8 +2509,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
    { attempt: 3, driverKind: "codex_app_server" },
    { attempt: 2, driverKind: "acpx_runtime" },
    { attempt: 2, driverKind: "opencode_server" },
-  ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind }) => {
-    const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}`);
+    // A second restart that loses the first restart's continuation turn.
+    { attempt: 2, driverKind: "codex_app_server", lostContinuation: true },
+  ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation }) => {
+    const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}`);
    await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {
      driverKind,
      sessionId: fixture.native.normalizedSessionId,
@@ -2523,6 +2525,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
        terminalState: "failed", result: null,
        error: { code: "provider_turn_lost_on_restore", recoverable: true },
      }) }],
+      // A restart during the first continuation: its turn is still the
+      // checkpointed active turn and the consumed marker names it.
+      ...(lostContinuation ? { activeTurnId: "continuation-turn", terminal: null,
+        dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "continuation-turn" } : {}),
    } } }).where(eq(heartbeatRuns.id, fixture.runId));
    await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId));
    const dispositions = await claimNativeRestartRecoveries({