fix(native): resume a run whose turn.start a restart cut off before Codex created the turn (#20507)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A native Codex run sends `turn.start` to its runner, and the runner
sends `turn/start` to Codex. The runner marks the start "ambiguous"
before it sends the request
> - A control-plane restart can kill the runner and Codex while that
request is in flight. After the restart, neither journal says whether
Codex created the turn, and the resumed thread has no active turn
> - The restored runner then failed closed on every same-run resume with
"cannot safely recover an ambiguous Codex turn start without an active
replacement turn". The answer never reached the agent
> - The next run on the task was refused at the runner-state fence
(`prior_owner_terminal_state_indeterminate;
retirement_refused=journal_commands_unsettled(turn.start.failed)`)
before it asked for a runner. Retry after that run then waited forever
on `process_identity_missing` for a process that never existed
> - This pull request lets the runner prove from Codex's own thread
history that the start never created a turn, and then resume and start
the turn once. It also lets Retry continue after a run that the
runner-state fence refused before launch
> - The benefit is that a turn start cut off by a restart continues by
itself when Codex never saw it, and a person's Retry is no longer a dead
end

## Linked Issues or Issue Description

**What happened?**
Local restart-chaos wave `w-20261110T033426-s108`. Command: `--env local
--adapters codex --workloads Q1,Q2 --faults K1 --triggers answered
--tasks 4 --concurrency 2 --seed 108`. Task t03 (Q2): the answer run
`92a57f50` resumed after the first restart. A second K1 restart landed
0.25 s after `session.resumed`, while its `turn.start` was in flight.
Each same-run resume then failed with `PRP command session.snapshot
failed: cannot safely recover an ambiguous Codex turn start without an
active replacement turn`. When the prerequisite finished, the next run
`0e6aa9d1` failed before launch with `runner_state_identity_mismatch:
prior_owner_terminal_state_indeterminate;
retirement_refused=journal_commands_unsettled(turn.start.failed)`. The
accepted Retry (202) never created a run: the task stayed on
`retryWait.reason=process_identity_missing` until the wave timed out
(`R3:retry_no_new_run`).

**Expected behavior**
If Codex never created the turn, the run resumes and starts it once by
itself. If that cannot be proven, a person's Retry starts a new run
instead of waiting forever.

**Steps to reproduce**
1. Run the wave command above on master.
2. Or run `pnpm --filter @paperclipai/paperclip-runner exec vitest run
src/live/runnerd-codex-transport-cut-turn-start.test.ts` on master (case
"before" fails with the wave error).

## What Changed

- Runner (`codex_provider.rs`, `provider_backend.rs`): when the restored
provider state has an ambiguous turn start, no active turn before the
loss, and the resumed thread has no active turn, the runner reads the
newest turn of the thread (`thread/turns/list`, limit 1, metadata only,
the same request `read_thread` already sends). If there is no turn, or
the newest turn is one Paperclip already settled, Codex never created
the turn. The runner clears the ambiguity, emits a `harness.diagnostic`
with code `ambiguous_turn_start_not_created`, and restores an idle
session. The old `turn.start` stays `execution_indeterminate` and is
never executed again. The same-run resume then starts the run's turn
once, through the existing "recovered session without an active turn"
path.
- Any other history (a turn Paperclip never settled, an in-progress
turn, an incomplete page, a non-Codex provider) keeps today's
fail-closed error. The probe runs only on this recovery path. The normal
turn path does no new work before `turn.start`.
- Server (`native-local-process-stop.ts`): `nativeRunNeverLaunched` also
accepts a failed native run whose error is the runner-state fence
refusal `runner_state_identity_mismatch: prior_owner_*` (or
`legacy_prior_owner_*`). The executor throws it only while it migrates
another run's saved state, before it requests a launch. The existing
evidence check still applies: any launch request, identity record,
`adapter.invoke` or runner event keeps the run held. Other runner-state
errors (for example `authority_indeterminate`) stay held.
- Fake Codex (`fake-codex-app-server.rs`): keeps the thread's turn
history in a `<state-file>.turns` sidecar, so `thread/turns/list`
returns history like real Codex (newest first, honours `limit`). New
`--hang-turn-start <file>` holds `turn/start` "before" or "after" the
thread records the turn, to model a kill in flight.
- Tests:
- New real-runnerd test
`runnerd-codex-transport-cut-turn-start.test.ts`. A real runner and fake
Codex start a turn; both processes are killed while `turn/start` is in
flight. "before": the resumed run completes, the journal holds the old
`turn.start` (`execution_indeterminate`) and one new completed
`turn.start` with the same task text, and Codex saw exactly one more
`turn/start`. "after": recovery still fails closed and no new
`turn/start` is sent.
- Rust:
`ambiguous_turn_start_that_never_created_a_turn_recovers_and_starts_once`
and
`ambiguous_turn_start_that_created_an_unsettled_turn_still_fails_closed`.
Both rewrite saved provider state the way a master runner wrote it, so
they also cover state saved before this change.
- `explicit-native-continuation.test.ts`: five new rows for the
runner-state fence. Admitted: refused before launch (with and without a
coordinator). Held on `process_identity_missing`: refused after a launch
request, refused after a runner event, and a run that failed its own
`authority_indeterminate` check.
- `docs/durable-recovery.md`: describes both behaviours.

## Verification

- Red/green, real runnerd: on master (the staged master runnerd),
`runnerd-codex-transport-cut-turn-start.test.ts` "before" fails with
`PRP command runner.drain failed: ... cannot safely recover an ambiguous
Codex turn start without an active replacement turn`, the wave error.
With this change, both cases pass.
- Red/green, Rust: without the `provider_backend.rs` change, `cargo test
-p paperclip-runner-core --test codex_provider
ambiguous_turn_start_that` fails the "never created" test. With it, both
pass. The full `cargo test -p paperclip-runner-core` passes.
- Red/green, server: without the `native-local-process-stop.ts` change,
the two "refused at the prior-owner runner-state fence" rows fail with
`process_identity_missing`. With it, all 14 rows pass.
- Related suites pass: 11 Runner transport files, 419 tests
(`runnerd-codex-transport`, `-attachment-recovery`, `-cold-restore`,
`-warm-attachment`, `-release-resume`, `-cut-attach`, `-cut-turn-start`,
`-settlement`, `-suspension`, `runnerd-restart-continuation`,
`runnerd-saved-continuation`); server `explicit-native-continuation`,
`quarantined-task-retry` and `issue-confirm-run-stopped-routes`, 3
files, 470 tests; full `cargo test -p paperclip-runner-core`.
- `pnpm --filter @paperclipai/paperclip-runner exec tsc --noEmit` and
server `tsc --noEmit` pass.
- Pack `contractDigest` is unchanged: `sha256:3164357d…` before and
after (`scripts/contracts/contract-fingerprints.mjs`).
- Live proof: the wave command above on this branch. Results follow in a
comment.

**No duplicate provider work.** The runner starts a new turn only when
Codex's own history proves the cut-off start created nothing. The old
command is never executed again (the runner refuses an `indeterminate`
command). A recorded turn that Paperclip never settled keeps the run
failed closed.

## Risks

- Low-medium risk. The runner change runs only while it restores a
session whose saved state has an ambiguous turn start and no active
turn. Before this change, that run could never continue.
- The proof relies on Codex recording a turn in the thread before it
runs anything. If Codex accepted the request but died before it wrote
the turn, the runner treats the start as not created and starts the turn
once. Codex writes the user turn to the thread when it starts the turn,
before it asks the model for output, so no model output or tool call is
expected without a recorded turn.
- If the thread's newest turn is not in the runner's settled set (for
example after a provider identity epoch rollover forgot older
identities), the runner keeps failing closed, as today.
- **Open decision (not changed here):** when Codex did record the
cut-off turn, it may have done work. This pull request keeps the run
failed closed and does not replay or adopt that turn. Continuing is a
person's decision through Retry or a new message.
- The server change trusts the executor's own error text
`runner_state_identity_mismatch: prior_owner_*`, together with the
absence of every launch and runner event. A run that requested a launch
keeps every stop proof.
- **Runner release:** the pack `contractDigest` does not change. Local
runners use the runnerd built with the app, so the fix applies with this
release.
- Not covered: the related `authority_indeterminate` variant in the same
wave (t04, Q1) is a separate cause; its Retry already worked in that
wave.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use
(shell, file edits, test execution).

## 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

Refs #20477, #20482

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude <noreply@anthropic.com>

packages/paperclip-runner/docs/durable-recovery.md

@@ -197,6 +197,26 @@ startup fence over a `prepared` state that holds a saved thread. The next runner
 retires that fence once the recorded provider process and its group have exited,
 as it does for an open or active session.
 
+A restart can also kill the runner and Codex while `turn/start` is in flight.
+The runner marks the start ambiguous before it sends the request, so after the
+restart neither journal says whether Codex created the turn, and the resumed
+thread reports no active turn either way. Codex records a turn in the thread
+before it runs anything, so the restored runner reads the newest turn of the
+thread history (`thread/turns/list`, one turn, metadata only). If there is none,
+or it is a turn Paperclip already settled, the start never created a turn: the
+runner clears the ambiguity, emits the `ambiguous_turn_start_not_created`
+diagnostic and restores an idle session. The old `turn.start` stays settled as
+`execution_indeterminate` and never runs again; the same-run resume then starts
+the run's turn once, as for any recovered session without an active turn. If
+the newest turn is one Paperclip never settled, Codex did create the turn and it
+may have done work. The runner still fails closed and never starts that work
+again by itself; continuing is a person's decision (Retry or a new message).
+
+A follow-up run that a prior owner's unproven runner state refuses
+(`runner_state_identity_mismatch: prior_owner_*`) fails before it asks for a
+runner, so it never had a process. Retry and new messages after it continue
+without waiting for a stop record of that run.
+
 Saved OpenCode and Cursor Cloud provider state binds the digest of its own
 qualified launch profile (command, proxy script and, for OpenCode, the
 executable, each by path and SHA-256). Cursor Cloud does not re-bind. A changed

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

@@ -21,6 +21,41 @@ struct FakeState {
     goal: Option<Value>,
 }
 
+/// Every turn this thread accepted, oldest first, one per line, like the turn
+/// history a real Codex thread keeps across process loss. It sits beside the
+/// state file so existing state files and their shape stay unchanged.
+fn turn_history_path(state_path: &Path) -> PathBuf {
+    let mut name = state_path.as_os_str().to_owned();
+    name.push(".turns");
+    PathBuf::from(name)
+}
+
+fn load_turn_history(state_path: &Path) -> Vec<String> {
+    fs::read_to_string(turn_history_path(state_path))
+        .map(|text| {
+            text.lines()
+                .filter(|line| !line.is_empty())
+                .map(str::to_owned)
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+fn record_turn(state_path: &Path, turn_id: &str) -> io::Result<()> {
+    if load_turn_history(state_path)
+        .iter()
+        .any(|known| known == turn_id)
+    {
+        return Ok(());
+    }
+    let mut file = OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open(turn_history_path(state_path))?;
+    writeln!(file, "{turn_id}")?;
+    file.sync_all()
+}
+
 fn argument(args: &[String], name: &str) -> Option<String> {
     args.iter()
         .position(|value| value == name)
@@ -1443,9 +1478,30 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     send(json!({"id": id, "result": {"data": [], "nextCursor": "next-page"}}))?;
                     continue;
                 }
-                let turns = state.active_turn_id.as_ref()
-                    .map(|turn_id| vec![json!({"id": turn_id, "status": "inProgress", "items": [], "itemsView": "notLoaded"})])
-                    .unwrap_or_default();
+                // Newest first. The thread keeps every accepted turn; only
+                // the active one is in progress.
+                let mut history = load_turn_history(&state_path);
+                if let Some(active) = state.active_turn_id.as_ref() {
+                    history.retain(|turn_id| turn_id != active);
+                    history.push(active.clone());
+                }
+                let limit = message
+                    .pointer("/params/limit")
+                    .and_then(Value::as_u64)
+                    .unwrap_or(100) as usize;
+                let turns = history
+                    .iter()
+                    .rev()
+                    .take(limit)
+                    .map(|turn_id| {
+                        let status = if state.active_turn_id.as_ref() == Some(turn_id) {
+                            "inProgress"
+                        } else {
+                            "completed"
+                        };
+                        json!({"id": turn_id, "status": status, "items": [], "itemsView": "notLoaded"})
+                    })
+                    .collect::<Vec<_>>();
                 send(json!({"id": id, "result": {"data": turns, "nextCursor": null}}))?;
             }
             "thread/read" => {
@@ -1654,6 +1710,17 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                 }))?;
             }
             "turn/start" => {
+                // Models a provider killed while turn/start was in flight. The
+                // file names where it stops: "before" the turn exists, or
+                // "after" the thread recorded it but before the response.
+                let hang_turn_start = argument(&args, "--hang-turn-start")
+                    .and_then(|path| fs::read_to_string(path).ok())
+                    .map(|stage| stage.trim().to_owned());
+                if hang_turn_start.as_deref() == Some("before") {
+                    loop {
+                        std::thread::sleep(std::time::Duration::from_secs(60));
+                    }
+                }
                 if require_external_sandbox
                     && (message.pointer("/params/sandboxPolicy")
                         != Some(&json!({
@@ -1730,7 +1797,13 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     )
                 };
                 state.active_turn_id = Some(provider_turn_id.clone());
+                record_turn(&state_path, &provider_turn_id)?;
                 save_state(&state_path, &state)?;
+                if hang_turn_start.as_deref() == Some("after") {
+                    loop {
+                        std::thread::sleep(std::time::Duration::from_secs(60));
+                    }
+                }
                 if ambiguous_older_reused_turn && turn_start_count == 3 {
                     send(json!({
                         "method": "turn/started",

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

@@ -2514,6 +2514,38 @@ impl CodexProvider {
         Ok(())
     }
 
+    /// Provider evidence for a `turn/start` whose outcome a process loss hid.
+    /// Returns `true` only when this resumed Codex thread has no active turn
+    /// and its newest recorded turn is absent or one Paperclip already settled:
+    /// Codex records a turn in the thread before it runs anything, so the
+    /// ambiguous start never created one. Any other history, an incomplete
+    /// page, or another provider returns `false` and keeps the start ambiguous.
+    pub(crate) fn ambiguous_turn_start_never_created_a_turn(
+        &mut self,
+        settled_provider_turn_ids: &BTreeSet<String>,
+    ) -> Result<bool, LocalRunnerError> {
+        if self.config.provider != "codex" || self.active_provider_turn_id.is_some() {
+            return Ok(false);
+        }
+        let page = self.request("thread/turns/list", self.newest_turn_page_params())?;
+        Ok(newest_turn_page_proves_no_new_turn(
+            &page,
+            settled_provider_turn_ids,
+        ))
+    }
+
+    /// The first `thread/turns/list` page `read_thread` requests, narrowed to
+    /// the newest turn's metadata.
+    fn newest_turn_page_params(&self) -> Value {
+        let mut params = serde_json::Map::new();
+        params.insert("threadId".to_owned(), Value::String(self.thread_id.clone()));
+        params.insert("cursor".to_owned(), Value::Null);
+        params.insert("limit".to_owned(), Value::from(1));
+        params.insert("sortDirection".to_owned(), Value::from("desc"));
+        params.insert("itemsView".to_owned(), Value::from("notLoaded"));
+        Value::Object(params)
+    }
+
     pub(crate) fn read_thread_after_recovery_goal(
         &mut self,
         goal_is_dormant: fn(&Value) -> bool,
@@ -5020,6 +5052,35 @@ fn codex_question_response(
     Ok(json!({"answers": native}))
 }
 
+/// Codex turn statuses that mean the turn is over. `read_thread` accepts the
+/// same set plus `inProgress`.
+const ENDED_TURN_STATUSES: [&str; 4] = ["completed", "failed", "interrupted", "cancelled"];
+
+/// Whether the newest page of a thread's turn history proves that no turn
+/// exists beyond the ones Paperclip settled: the page is empty and complete,
+/// or its newest turn is a settled one with an explicit ended status. Any
+/// missing, malformed or unknown field proves nothing.
+fn newest_turn_page_proves_no_new_turn(
+    page: &Value,
+    settled_provider_turn_ids: &BTreeSet<String>,
+) -> bool {
+    let Some(turns) = page.pointer("/data").and_then(Value::as_array) else {
+        return false;
+    };
+    let Some(newest) = turns.first() else {
+        // An empty first page is the whole history only without a cursor.
+        return page.pointer("/nextCursor").is_none_or(Value::is_null);
+    };
+    newest
+        .pointer("/id")
+        .and_then(Value::as_str)
+        .is_some_and(|turn_id| settled_provider_turn_ids.contains(turn_id))
+        && newest
+            .pointer("/status")
+            .and_then(Value::as_str)
+            .is_some_and(|status| ENDED_TURN_STATUSES.contains(&status))
+}
+
 #[cfg(test)]
 mod tests {
 
@@ -5071,6 +5132,45 @@ mod tests {
     }
     use super::*;
 
+    #[test]
+    fn only_a_complete_settled_ended_history_proves_no_new_turn() {
+        let settled = BTreeSet::from(["turn-1".to_owned()]);
+        let page = |turns: Value, cursor: Value| json!({"data": turns, "nextCursor": cursor});
+        let proves = |value: Value| newest_turn_page_proves_no_new_turn(&value, &settled);
+        assert!(proves(page(json!([]), Value::Null)));
+        assert!(proves(json!({"data": []})));
+        assert!(proves(page(
+            json!([{"id": "turn-1", "status": "completed"}]),
+            Value::Null
+        )));
+        assert!(proves(page(
+            json!([{"id": "turn-1", "status": "interrupted"}]),
+            Value::Null
+        )));
+        // An incomplete page, an unsettled or in-progress turn, or a missing,
+        // null or unknown field keeps the start ambiguous.
+        assert!(!proves(page(json!([]), json!("next-page"))));
+        assert!(!proves(json!({"nextCursor": null})));
+        assert!(!proves(page(
+            json!([{"id": "turn-2", "status": "completed"}]),
+            Value::Null
+        )));
+        assert!(!proves(page(
+            json!([{"id": "turn-1", "status": "inProgress"}]),
+            Value::Null
+        )));
+        assert!(!proves(page(json!([{"id": "turn-1"}]), Value::Null)));
+        assert!(!proves(page(
+            json!([{"id": "turn-1", "status": null}]),
+            Value::Null
+        )));
+        assert!(!proves(page(
+            json!([{"id": "turn-1", "status": "paused"}]),
+            Value::Null
+        )));
+        assert!(!proves(page(json!([{"status": "completed"}]), Value::Null)));
+    }
+
     #[test]
     #[cfg(unix)]
     fn runtime_notification_policy_grants_only_discovery_files_and_ipc() {

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

@@ -2768,6 +2768,23 @@ impl CodexCommandExecutor {
                 self.save_state()?;
                 return Ok(());
             }
+            // A process loss can hide whether an in-flight turn/start reached
+            // Codex. When the resumed thread has no active turn and its history
+            // ends at a turn Paperclip already settled, Codex never created the
+            // turn: the start is settled as not started, and the controller
+            // may start it once. Any other evidence keeps it ambiguous.
+            let ambiguous_start_never_created = ambiguous_turn_start_pending
+                && recovered_active_turn_id.is_none()
+                && previous_active_turn_id.is_none()
+                && provider
+                    .ambiguous_turn_start_never_created_a_turn(&settled_provider_turn_ids)
+                    .map_err(|error| {
+                        DurableRunnerError::invalid(format!(
+                            "ambiguous {provider_name} turn start history probe failed: {error}"
+                        ))
+                    })?;
+            let ambiguous_turn_start_pending =
+                ambiguous_turn_start_pending && !ambiguous_start_never_created;
             if ambiguous_turn_start_pending {
                 let recovered_turn_id = recovered_active_turn_id.as_deref().ok_or_else(|| {
                 DurableRunnerError::invalid(
@@ -2844,6 +2861,19 @@ impl CodexCommandExecutor {
                 state.provider_session_id = resumed_provider_session_id.clone();
                 state.settled_provider_turn_ids = settled_provider_turn_ids;
                 state.settled_provider_turn_filter = settled_provider_turn_filter;
+                if ambiguous_start_never_created {
+                    state.ambiguous_turn_start_pending = false;
+                    state.push_event(NormalizedProviderEvent {
+                        event_type: "harness.diagnostic".to_owned(),
+                        priority: EventPriority::P0,
+                        payload: json!({
+                            "provider": provider_label,
+                            "code": "ambiguous_turn_start_not_created",
+                            "message": format!("{provider_name} recovery proved from the resumed thread's history that the turn start a process loss interrupted never created a turn"),
+                            "paperclipAccepted": false,
+                        }),
+                    })?;
+                }
                 state.push_terminal_event(NormalizedProviderEvent {
                     event_type: "session.resumed".to_owned(),
                     priority: EventPriority::P0,

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

@@ -3296,6 +3296,103 @@ fn completed_ambiguous_replacement_fails_closed_after_process_loss() {
     fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
 }
 
+/// The durable state a process loss leaves when it lands after the runner
+/// journaled a second `turn.start` and marked it ambiguous, but before Codex
+/// answered. `created` says whether Codex had already recorded that turn.
+fn cut_off_second_turn_start(label: &str, created: bool) -> PathBuf {
+    let directory = temporary_directory(label);
+    let config = provider_config(&directory, &["--durable-turn-ids"]);
+    let mut executor = CodexCommandExecutor::new(&directory);
+    executor
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider": config}),
+        ))
+        .expect("prepare Codex provider");
+    executor
+        .execute(&command("open", 2, "session.open", json!({})))
+        .expect("open Codex session");
+    executor
+        .execute(&command(
+            "first-turn",
+            3,
+            "turn.start",
+            json!({"text": "Complete the first turn."}),
+        ))
+        .expect("start first provider turn");
+    wait_for_executor_event(&mut executor, "turn.completed");
+    executor.shutdown().expect("stop first provider process");
+    drop(executor);
+    let provider_path = directory.join("codex-provider-state.json");
+    let mut provider: Value =
+        serde_json::from_slice(&fs::read(&provider_path).expect("read provider state"))
+            .expect("parse provider state");
+    provider["ambiguousTurnStartPending"] = json!(true);
+    fs::write(&provider_path, serde_json::to_vec(&provider).unwrap())
+        .expect("write the cut-off provider state");
+    if created {
+        // Codex recorded the second turn in its thread before it died.
+        use std::io::Write as _;
+        let mut history = fs::OpenOptions::new()
+            .append(true)
+            .open(directory.join("fake-state.json.turns"))
+            .expect("open fake turn history");
+        writeln!(history, "provider-turn-2").expect("record the cut-off turn");
+    }
+    directory
+}
+
+#[test]
+fn ambiguous_turn_start_that_never_created_a_turn_recovers_and_starts_once() {
+    let directory = cut_off_second_turn_start("durable-ambiguous-not-created", false);
+    let mut recovered = CodexCommandExecutor::new(&directory);
+    let snapshot = recovered
+        .execute(&command("snapshot", 5, "session.snapshot", json!({})))
+        .expect("history proves the cut-off start never created a turn");
+    assert_eq!(snapshot.result["status"], "session_open");
+    assert!(snapshot.result["activeProviderTurnId"].is_null());
+    let persisted: Value = serde_json::from_slice(
+        &fs::read(directory.join("codex-provider-state.json")).expect("read recovered state"),
+    )
+    .expect("parse recovered state");
+    assert_eq!(persisted["ambiguousTurnStartPending"], false);
+    assert_eq!(call_count(&directory, "thread/turns/list"), 1);
+    let started = recovered
+        .execute(&command(
+            "second-turn",
+            6,
+            "turn.start",
+            json!({"text": "Start once."}),
+        ))
+        .expect("start the turn the cut-off start never created");
+    assert_eq!(started.result["providerTurnId"], "provider-turn-2");
+    wait_for_executor_event(&mut recovered, "turn.completed");
+    assert_eq!(call_count(&directory, "turn/start"), 2);
+    recovered
+        .shutdown()
+        .expect("stop recovered provider process");
+    fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
+}
+
+#[test]
+fn ambiguous_turn_start_that_created_an_unsettled_turn_still_fails_closed() {
+    let directory = cut_off_second_turn_start("durable-ambiguous-created", true);
+    let mut recovered = CodexCommandExecutor::new(&directory);
+    let error = recovered
+        .execute(&command("snapshot", 5, "session.snapshot", json!({})))
+        .expect_err("a recorded turn Paperclip never settled may have done work");
+    assert!(
+        error
+            .to_string()
+            .contains("cannot safely recover an ambiguous Codex turn start"),
+        "unexpected ambiguous recovery error: {error}"
+    );
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    fs::remove_dir_all(directory).expect("remove Codex integration-test directory");
+}
+
 #[test]
 fn ambiguous_replacement_completion_replaces_durable_turn_authority() {
     let directory = temporary_directory("durable-ambiguous-turn-completion");

packages/paperclip-runner/src/live/runnerd-codex-transport-cut-turn-start.test.ts

@@ -0,0 +1,137 @@
+import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { expect, it, vi } from "vitest";
+import { createRunnerdNativeSessionBackend } from "../backends/codex-native-backend.js";
+import type { NativeExecutionInputV1 } from "../contracts/native-execution.js";
+import type { PrpEvent } from "../protocol/replay-contract.js";
+import { executeNativeSession } from "../native-session-runtime.js";
+import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js";
+
+/**
+ * A restart can kill the runner and its Codex process while `turn/start` is in
+ * flight. The runner journaled the command and marked the start ambiguous, so
+ * the restored runner cannot know from its own state whether Codex created the
+ * turn, and the resumed thread has no active turn either way.
+ *
+ * - "before": Codex never created the turn. The resumed thread's history ends
+ *   at a turn Paperclip already settled (here: no turn at all), which proves the
+ *   start never happened. The same-run resume starts the turn exactly once.
+ * - "after": Codex recorded the turn, then died. Its history names a turn
+ *   Paperclip never settled. The turn may have done work, so the runner still
+ *   fails closed and never starts it again by itself.
+ */
+it.each(["before", "after"] as const)("resumes a run whose turn.start a restart cut off (provider stopped %s creating the turn)", async (stage) => {
+  const root = await mkdtemp(join(tmpdir(), `runnerd-cut-turn-start-${stage}-`));
+  const callsPath = join(root, "calls.log");
+  const hangPath = join(root, "hang-turn-start");
+  const identity = { runId: "run-cut-start", sessionId: "session-cut-start", companyId: "company-cut-start", issueId: "issue-cut-start", agentId: "agent-cut-start" };
+  const execution: NativeExecutionInputV1 = {
+    schema: "paperclip.native-execution-input.v1",
+    binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-cut-start" },
+    session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
+    task: { identifier: "TEST-7", title: "Write the brief", description: null, prompt: "Write the brief with the answer marker.", workMode: "standard" },
+    workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
+    provider: { kind: "codex", model: null },
+    completionContract: { id: "contract-cut-start", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: {
+      revision: "1", objective: "Write the brief", criteria: [{ id: "brief", requirement: "Write the brief" }],
+    } },
+    interactionResponses: [], credentialBindings: [],
+  };
+  const options = {
+    runnerBinary: defaultCapabilityRunnerdBinary(),
+    codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"),
+    codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", callsPath, "--durable-turn-ids", "--hang-turn-start", hangPath],
+    stateDirectory: root,
+    turnStartTimeoutMs: 8_000,
+    prpIdentity: { runnerInstanceId: "runner-cut-start", environmentLeaseId: "lease-cut-start", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-cut-start", itemId: "item-cut-start" },
+  };
+  const first = createRunnerdCodexTransport(options);
+  let restored: ReturnType<typeof createRunnerdCodexTransport> | undefined;
+  const alive = (pid: number | null) => {
+    if (!pid) return false;
+    try { process.kill(pid, 0); return true; } catch { return false; }
+  };
+  const kill = (pid: number | null) => {
+    if (!pid) return;
+    try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; }
+  };
+  const turnStarts = async () => ((await readFile(callsPath, "utf8")).match(/^turn\/start$/gm) ?? []).length;
+  try {
+    const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-cut-start", transportFactory: () => first.transport });
+    const session = await backend.openSession({ identity, workingDirectory: root });
+    // The last checkpoint the controller persisted: no active turn.
+    const checkpoint = await session.snapshot();
+    expect(checkpoint.activeTurnId ?? null).toBeNull();
+    await writeFile(hangPath, stage);
+    const starting = session.startTurn({ message: { role: "user", text: execution.task.prompt } }).catch((error: unknown) => error);
+    // The runner journaled turn.start, marked the start ambiguous and sent it.
+    await vi.waitFor(async () => {
+      const provider = JSON.parse(await readFile(join(root, "runner", "codex-provider-state.json"), "utf8"));
+      expect(provider.ambiguousTurnStartPending).toBe(true);
+      expect(await turnStarts()).toBe(1);
+      if (stage === "after") expect(JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")).activeTurnId).toBe("provider-turn-1");
+    }, { timeout: 8_000, interval: 25 });
+    const owner = first.evidence();
+    await first.detachControllerForRestart();
+    kill(owner.runnerPid);
+    kill(owner.providerPid);
+    await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); });
+    await starting;
+    await rm(hangPath, { force: true });
+    // Codex's process-loss history: the thread remains, its in-memory turn does not.
+    const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8"));
+    providerState.activeTurnId = null;
+    await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState));
+
+    restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null });
+    const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-cut-start", transportFactory: () => restored!.transport });
+    const events: PrpEvent[] = [];
+    const executing = executeNativeSession({
+      input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true,
+      runnerInstanceId: "runner-cut-start", controlPlaneInstanceId: "controller-cut-start", timeoutMs: 15_000,
+      controlPlane: {
+        async openRun() {},
+        async checkpointSession() {},
+        async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
+        async replayEvents(request) {
+          const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId);
+          return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 };
+        },
+        async completeRun() {},
+      },
+    });
+    const control = async () => JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")) as {
+      commands: Array<{ type: string; status: string; payload: { text?: string }; result?: { result?: { code?: string } } }>;
+    };
+    if (stage === "before") {
+      const completion = await executing;
+      expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, terminal: { runTerminalState: "succeeded" } });
+      expect(events.find(event => event.eventType === "session.failed")).toBeUndefined();
+      // The cut-off start stays settled as unknown and is never executed again;
+      // one new start carries the same task, and Codex saw exactly one more.
+      const starts = (await control()).commands.filter(command => command.type === "turn.start");
+      expect(starts).toHaveLength(2);
+      expect(starts[0]).toMatchObject({ status: "failed", result: { result: { code: "execution_indeterminate" } } });
+      expect(starts[1]).toMatchObject({ status: "completed" });
+      expect(starts[1]!.payload.text).toContain(execution.task.prompt);
+      expect(await turnStarts()).toBe(2);
+    } else {
+      await expect(executing).rejects.toThrow(/cannot safely recover an ambiguous Codex turn start/);
+      // Possibly started work is never started again without a person.
+      expect(await turnStarts()).toBe(1);
+      const starts = (await control()).commands.filter(command => command.type === "turn.start");
+      expect(starts).toHaveLength(1);
+    }
+  } finally {
+    await rm(hangPath, { force: true });
+    await restored?.transport.close().catch(() => undefined);
+    for (const bundle of [first, restored]) {
+      if (!bundle) continue;
+      const evidence = bundle.evidence();
+      kill(evidence.runnerPid);
+      kill(evidence.providerPid);
+    }
+    await rm(root, { recursive: true, force: true });
+  }
+}, 60_000);

server/src/services/explicit-native-continuation.test.ts

@@ -94,6 +94,8 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
       onBlocked: reason => blocked.push(reason) });
     return { result, blocked, reads };
   }
+  const PRIOR_OWNER_REFUSAL = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate; "
+    + "retirement_refused=journal_commands_unsettled(turn.start.failed)";
   it.each([
     { shape: "refused by the cleanup quarantine before launch", errorCode: "native_session_cleanup_quarantined", launched: false,
       comment: "local_cleanup", retry: "cleanup_quarantined" },
@@ -109,6 +111,22 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
       prepared: true, comment: "process_identity_missing", retry: "process_identity_missing" },
     { shape: "failed without a process record after native selection", errorCode: "adapter_failed", launched: false,
       coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" },
+    // w-20261110T033426-s108/t03: a restart cut off the previous run's turn
+    // start, so its saved runner state could not be proven finished. The next
+    // run was refused at that runner-state fence before it asked for a runner,
+    // and Retry then waited forever for a stop record of a process that never
+    // existed.
+    { shape: "was refused at the prior-owner runner-state fence before launch", errorCode: "adapter_failed", launched: false,
+      prepared: true, error: PRIOR_OWNER_REFUSAL, comment: null, retry: null },
+    { shape: "was refused at the prior-owner runner-state fence with a coordinator", errorCode: "adapter_failed", launched: false,
+      prepared: true, coordinator: true, error: PRIOR_OWNER_REFUSAL, comment: null, retry: null },
+    { shape: "was refused at the prior-owner runner-state fence after it requested a launch", errorCode: "adapter_failed", launched: true,
+      prepared: true, error: PRIOR_OWNER_REFUSAL, comment: "process_identity_missing", retry: "process_identity_missing" },
+    { shape: "was refused at the prior-owner runner-state fence after a runner event", errorCode: "adapter_failed", launched: false,
+      runnerEvent: true, prepared: true, error: PRIOR_OWNER_REFUSAL, comment: "process_identity_missing", retry: "process_identity_missing" },
+    { shape: "failed its own runner-state authority check", errorCode: "adapter_failed", launched: false,
+      prepared: true, error: "runner_state_identity_mismatch: authority_indeterminate",
+      comment: "process_identity_missing", retry: "process_identity_missing" },
     { shape: "timed out before it requested a launch", status: "timed_out", errorCode: "timeout", launched: false,
       comment: null, retry: null },
     { shape: "timed out after a runner event without a process record", status: "timed_out", errorCode: "timeout", launched: false,
@@ -116,14 +134,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     { shape: "was interrupted without a process record after native selection", status: "interrupted", errorCode: "process_lost", launched: false,
       coordinator: true, comment: "process_identity_missing", retry: "process_identity_missing" },
   ] as Array<{ shape: string; status?: "timed_out" | "interrupted"; errorCode: string; launched: boolean; runnerEvent?: boolean; prepared?: boolean; coordinator?: boolean;
-    comment: string | null; retry: string | null }>)("routes input after a run $shape to the gate that can clear it", async ({
-    status, errorCode, launched, runnerEvent, prepared, coordinator, comment, retry }) => {
+    error?: string; comment: string | null; retry: string | null }>)("routes input after a run $shape to the gate that can clear it", async ({
+    status, errorCode, launched, runnerEvent, prepared, coordinator, error, comment, retry }) => {
     // s7532/t04: a Retry's run failed at the quarantine fence before it asked
     // for a runner. Input then waited for a stop record of a process that
     // never existed ("process_identity_missing", Retry still offered) instead
     // of the quarantine's own gate, which names the action that clears it.
     const f = await seed();
-    await db.update(heartbeatRuns).set({ processPid: null, errorCode, ...(status ? { status } : {}) })
+    await db.update(heartbeatRuns).set({ processPid: null, errorCode, ...(status ? { status } : {}), ...(error ? { error } : {}) })
       .where(eq(heartbeatRuns.id, f.sourceRunId));
     // A run that failed before native selection has no native issue and no
     // coordinator. A coordinator row requires the run's native issue (foreign

server/src/services/native-local-process-stop.ts

@@ -240,7 +240,11 @@ function record(value: unknown): Record<string, unknown> {
 
 /** A settled native run that never had a runner has nothing to stop:
  * - an admission fence (a quarantined prior-session cleanup) refused it before
- *   it requested a launch; or
+ *   it requested a launch;
+ * - the runner-state fence refused it before it requested a launch because a
+ *   previous run still owns the conversation's saved runner state and that
+ *   owner could not be proven finished (`prior_owner_*`). The root belonged
+ *   to another run, so this run started nothing; or
  * - it failed before native selection (for example a refused budget
  *   reservation or a failed preparation query). Selection records the run's
  *   native issue, native session, runner instance and coordinator, and a
@@ -248,12 +252,17 @@ function record(value: unknown): Record<string, unknown> {
  *   in any build.
  * Any launch request, identity record or runner event keeps the normal stop
  * proofs. */
+/** Thrown only by the executor's runner-state migration, before any launch,
+ * when the conversation's saved state names a different (prior) run. */
+const PRIOR_OWNER_FENCE_REFUSAL = /^runner_state_identity_mismatch: (?:legacy_)?prior_owner_/;
+
 export async function nativeRunNeverLaunched(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
   "id" | "companyId" | "runtimeMode" | "status" | "errorCode" | "processPid" | "processGroupId" |
-  "nativeSessionId" | "runnerInstanceId" | "nativeIssueId">): Promise<boolean> {
+  "nativeSessionId" | "runnerInstanceId" | "nativeIssueId" | "error">): Promise<boolean> {
   if (run.runtimeMode !== "native" || !SETTLED_RUN_STATUSES.includes(run.status) || run.processPid || run.processGroupId) return false;
-  const quarantinedBeforeLaunch = run.status === "failed" && run.errorCode === "native_session_cleanup_quarantined";
-  if (!quarantinedBeforeLaunch) {
+  const refusedBeforeLaunch = run.status === "failed" && (run.errorCode === "native_session_cleanup_quarantined" ||
+    PRIOR_OWNER_FENCE_REFUSAL.test(run.error ?? ""));
+  if (!refusedBeforeLaunch) {
     // A coordinator row requires the run's native issue (foreign key), so the
     // native issue also rules out a coordinator.
     if (run.nativeSessionId || run.runnerInstanceId || run.nativeIssueId) return false;