fix(runner): keep a restart-lost Codex turn's late usage from failing its continuation (#19718)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Native Codex runs continue a turn lost to a control-plane restart
once, in the same conversation (#19699)
> - In restart storm 5, a Codex run crossed one graceful restart. The
continuation started, but the task stopped and needed a Retry
> - The runner log shows the cause. The resumed Codex thread sent the
usage of the lost turn after it accepted the continuation turn. The
runner rejected that usage as `turn_binding_mismatch` and failed the run
> - Recovery marks the lost turn as settled in the durable ledger. It
does not tell the live Codex provider, so the provider treats the old
turn's usage as a foreign event
> - This pull request records the lost turn as settled in the live
provider too
> - The benefit is that a restart-lost Codex turn continues without a
Retry

## Linked Issues or Issue Description

Refs #19699, #19608, #19501, #15297

**What happened?**

Restart storm 5 (seed 7106). Task t01 (Codex, B5) crossed one K2 SIGTERM
restart. Recovery resumed the dead runner. Runnerd restored the
conversation and recorded `turn.failed` with
`provider_turn_lost_on_restore` (recoverable). The control plane started
the continuation (#19699). Runnerd accepted it as a new provider turn.
About 0.5 s later, Codex sent `thread/tokenUsage/updated` for the lost
turn id. The runner's durable state records these events:

- `turn.accepted` for the new turn `01a1b3e2-…`
- `harness.diagnostic` `turn_binding_mismatch`, method
`thread/tokenUsage/updated`, `expectedTurnId` = new turn,
`receivedTurnId` = lost turn `01a1b3e0-…`
- `turn.failed` (`recoverable: false`) and `run.terminal` failed

The control plane did not see `turn/started` and failed after 30 s with
"runnerd did not report the provider turn identity". The task needed a
manual Retry.

**Expected behavior**

Usage for the lost turn is history. The runner keeps it as a resume
usage snapshot. The continuation runs, and the task continues without a
Retry.

**Steps to reproduce**

1. Start a Codex turn through the durable runner.
2. Kill the runner while the turn is active. Restore so that the
provider reports no active turn (`provider_turn_lost_on_restore`).
3. Start the continuation turn. Let the provider send
`thread/tokenUsage/updated` for the lost turn after it accepts the new
turn.
4. Before this change, the runner fails the continuation with
`turn_binding_mismatch`.

## What Changed

- `provider_backend.rs`: when recovery finds that the previous active
turn ended while the runner was offline, it now also restores that turn
id into the live provider's settled-turn set. The durable ledger already
settled it. The existing settled-turn path then turns its late usage
into `paperclip/resumeUsageSnapshot` and does not bill its `last` usage
to the run.
- The code skips this step when the settled-turn epoch needs a rollover,
the same as the durable ledger.
- Fake Codex: new `--stale-usage-after-turn-start` flag. With durable
turn ids, it sends usage for the previous turn right after it accepts
the next turn, before `turn/started`. This matches the live Codex order.
- New test
`restart_continuation_ignores_late_usage_for_the_turn_lost_on_restore`
in `tests/codex_provider.rs`. It uses the real durable Codex executor
with fake Codex.

## Verification

- Red before the fix: `cargo test -p paperclip-runner-core --test
codex_provider restart_continuation_ignores_late_usage` failed with the
same diagnostic as the live run (`turn_binding_mismatch`, method
`thread/tokenUsage/updated`, `receivedTurnId: provider-turn-1`,
`expectedTurnId: provider-turn-2`).
- Green after the fix: the same test passes. The continuation emits
`turn.started` for `provider-turn-2`, there is no `turn.failed` or
`run.terminal`, and fake Codex saw exactly two `turn/start` calls (the
original and one continuation).
- `cargo test -p paperclip-runner-core`: all suites pass.
`acpx_park_readiness::hosted_park_acpx_preserves_late_goals_and_rejects_foreign_or_busy_observations`
failed once and passed on rerun. This change does not touch ACPX.
- `cargo fmt --check` is clean.
- Not run: the live restart-chaos replay of
`w-20261107T023921-s7106/t01`. The local harness needs provider
credentials that were not available to this session.

## Risks

- Low risk. The change adds one already-ended turn id to the live
provider's settled set. That set is only used to ignore stale events and
to reject reuse of a settled id. So a provider that later reports the
lost turn id as a new active turn is now rejected, as the durable ledger
already does.
- The change does not address a second, separate problem: after the
runner failed the continuation, the control-plane transport waited the
full 30 s turn-start bound and reported "runnerd did not report the
provider turn identity" instead of the runner's `turn_binding_mismatch`.
That message only applies when such a failure occurs. This PR removes
this specific trigger.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file editing). Default context window and
reasoning mode.

## 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
- [ ] All Paperclip CI gates are green
- [ ] 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>

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs

@@ -1781,6 +1781,20 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     "id": id,
                     "result": {"turn": {"id": provider_turn_id, "status": "inProgress"}}
                 }))?;
+                // A resumed Codex thread can report the cumulative usage of the
+                // turn lost to a restart after it accepts the next turn.
+                if durable_turn_ids
+                    && state.next_turn > 1
+                    && args
+                        .iter()
+                        .any(|value| value == "--stale-usage-after-turn-start")
+                {
+                    send(json!({"method": "thread/tokenUsage/updated", "params": {
+                        "threadId": state.thread_id,
+                        "turnId": format!("provider-turn-{}", state.next_turn - 1),
+                        "tokenUsage": {"total": {"inputTokens": 100, "outputTokens": 20}, "last": {"inputTokens": 50, "outputTokens": 10}}
+                    }}))?;
+                }
                 // A loaded host can accept a turn long before it starts it.
                 if let Some(delay) = argument(&args, "--turn-started-delay-ms") {
                     thread::sleep(Duration::from_millis(delay.parse()?));

packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs

@@ -2631,6 +2631,26 @@ impl CodexCommandExecutor {
                 ));
                 }
             }
+            if let Some(ended_turn_id) = previous_active_turn_id
+                .as_ref()
+                .filter(|_| recovered_active_turn_id.is_none() && !provider_epoch_requires_rollover)
+            {
+                // Recovery settles the turn that ended while the runner was
+                // offline in the durable ledger below. The live provider must
+                // know it too: a resumed thread can report that turn's usage
+                // after it accepts the restart continuation, which is history,
+                // not a foreign turn that fails the continuation.
+                provider
+                    .restore_settled_turn_identities(
+                        std::iter::once(ended_turn_id.clone()),
+                        settled_provider_turn_filter.clone(),
+                    )
+                    .map_err(|error| {
+                        DurableRunnerError::invalid(format!(
+                            "failed to restore the ended provider turn identity: {error}"
+                        ))
+                    })?;
+            }
             provider
                 .restore_completed_turn_authority(
                     (completed_turn_authoritative || recovered_turn_ended_with_result)

packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs

@@ -4383,6 +4383,98 @@ fn restart_lost_turn_preserves_rejected_completion_without_changing_ordinary_rec
     }
 }
 
+#[test]
+fn restart_continuation_ignores_late_usage_for_the_turn_lost_on_restore() {
+    // RC-20c: a resumed Codex thread reported the cumulative usage of the turn
+    // lost to the restart after it accepted the continuation turn. The runner
+    // must treat that usage as history of a settled turn, not as a foreign
+    // event that fails the continuation with turn_binding_mismatch.
+    let directory = temporary_directory("restart-continuation-stale-usage");
+    let config = provider_config(
+        &directory,
+        &[
+            "--durable-turn-ids",
+            "--linger-after-turn-start",
+            "--stale-usage-after-turn-start",
+        ],
+    );
+    let mut first = CodexCommandExecutor::new(&directory);
+    first
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider": config}),
+        ))
+        .unwrap();
+    first
+        .execute(&command("open", 2, "session.open", json!({})))
+        .unwrap();
+    first
+        .execute(&command(
+            "start",
+            3,
+            "turn.start",
+            json!({"text": "Run the long benchmark."}),
+        ))
+        .unwrap();
+    drop(first);
+    fs::write(
+        directory.join("fake-state.json"),
+        serde_json::to_vec(&json!({
+            "threadId": "codex-thread-1", "activeTurnId": null, "nextTurn": 1,
+        }))
+        .unwrap(),
+    )
+    .unwrap();
+    let mut restored = CodexCommandExecutor::new(&directory);
+    restored
+        .execute(&command("snapshot", 4, "session.snapshot", json!({})))
+        .unwrap();
+    let lost = poll_and_ack(&mut restored).unwrap();
+    assert!(lost.iter().any(|event| event.event_type == "turn.failed"
+        && event.payload["error"]["code"] == "provider_turn_lost_on_restore"));
+    let accepted = restored
+        .execute(&command(
+            "continue",
+            5,
+            "turn.start",
+            json!({"text": "Continue after the restart."}),
+        ))
+        .unwrap();
+    assert_eq!(accepted.result["providerTurnId"], "provider-turn-2");
+    let mut events = Vec::new();
+    for _ in 0..50 {
+        events.extend(poll_and_ack(&mut restored).unwrap());
+        if events
+            .iter()
+            .any(|event| event.event_type == "turn.started")
+        {
+            break;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(20));
+    }
+    assert!(
+        !events
+            .iter()
+            .any(|event| event.payload["code"] == "turn_binding_mismatch"
+                || event.payload["error"]["code"] == "turn_binding_mismatch"),
+        "late usage for the lost turn must not fail the continuation: {:?}",
+        events
+            .iter()
+            .map(|event| (&event.event_type, &event.payload))
+            .collect::<Vec<_>>()
+    );
+    assert!(!events
+        .iter()
+        .any(|event| event.event_type == "run.terminal" || event.event_type == "turn.failed"));
+    assert!(events.iter().any(|event| event.event_type == "turn.started"
+        && event.payload["providerTurnId"] == "provider-turn-2"));
+    assert_eq!(call_count(&directory, "turn/start"), 2);
+    restored.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
 #[test]
 fn durable_backend_settles_pending_tools_when_recovery_finds_the_turn_ended() {
     let directory = temporary_directory("durable-tool-ended-offline");