fix(runner): keep a Claude session resumable when its turn-end provider restart fails (#20285)

## Thinking Path

- #20145 prepares the next Claude provider generation at park, and
#20190 restarts the ACPX provider when a turn settles. They move the
sidecar respawn off the follow-up's critical path.
- When that turn-end restart fails, the runner "fails closed":
`restart_session_after_turn` saves the session `closed` and drops it.
Within the run, the next turn fails exactly like a failed restart at
turn start. That's intended.
- Across runs it breaks the conversation:
- the next run's `run.attach` refuses a `closed` session ("run.attach
requires the same settled ACPX provider profile and session");
  - it reconnects for about 10 s;
- the controller then records a continuity break and starts a fresh
Claude conversation (`run.startup.reuse provider_session:
fresh/created`).
- A local A/B (`ttft-baseline --adapters claude --samples 3 --phases
coldwarm`, on a host with load average about 17–27) confirmed it:

  | | d40a7494 (before #20145/#20190) | a3a25ce6 (after) |
  |---|---|---|
| Follow-ups that resumed the conversation | 3 of 3 (`verified_reuse`,
resume 4.4–8.2 s) | 1 of 3 |
| Follow-ups that started a fresh conversation | 0 of 3 | 2 of 3
(bootstrap 16–17 s) |

The kept instance shows each lost follow-up's runner received provider
state `lifecycle: "closed"`, with `run.attach` refused.
- A failed park preparation already recovers safely: it waits for the
reaped provider's lifetime fence, then saves `suspended`. The turn-end
restart should do the same.

## Linked Issues or Issue Description

Refs #20190, #20145

**What happened?**
When the post-turn ACPX provider restart failed, the next Claude run
could not attach. It spent about 10 s and then started a new
conversation, losing the task's conversation context.

**Steps to reproduce**
1. Run a Claude task on a busy host so the turn-end provider restart
fails (the fixture `reject-open` fault reproduces it).
2. Post a follow-up comment (a new run).
3. The follow-up's `run.attach` is refused, then `run.startup.reuse`
reports `provider_session: fresh/created`.

**Expected behavior**
The follow-up resumes the same Claude conversation. Within the failing
run, later turns still fail closed.

## What Changed

- `acpx_park_preparation.rs`:
- A failed `restart_session_after_turn` now calls the shared
`recover_failed_generation_restart`, renamed from
`recover_failed_park_preparation`.
- That recovery drops the session, marks it `prepared` with the provider
exit unconfirmed, waits for the provider-lifetime fence, then saves
`suspended`.
- If the fence or the save fails, the state stays exit-unconfirmed,
which is the safe direction.
- `acpx_provider_backend.rs`:
- Later turns of the failing run still fail closed: the first with "ACPX
turn start failed closed: <cause>", then "ACPX session is not open".
That's the same as a failed restart at turn start.
- The restriction is saved in the ACPX state as
`turnEndRestartFailedRun` (serialized only while set), so it outlasts a
Runner process restart inside the same run. Same-run restore doesn't
reopen the session while it's set.
- Only a `run.attach` from a different run clears it; a same-run attach
keeps it. State validation requires it to match the descriptor's run.
- The cause is handed to the durable Runner once
(`take_turn_end_restart_failure`), which records it as a redacted
`turn-end provider restart failed: <cause>` diagnostic in
`runner-state.json`. That gives the under-load cause a first recorded
sample.
- Tests (`tests/support/acpx_turn_end_restart.rs`):
- New:
`the_next_run_resumes_the_conversation_after_a_failed_turn_end_restart`,
red on master (the state is `closed`). It covers the same Runner process
with a warm attach, and a new Runner process with a cold attach. The
conversation id is preserved, and the next run completes a turn.
- New:
`a_runner_restart_in_the_same_run_keeps_the_failed_turn_end_restart_closed`,
red without the saved field. A new Runner process with the same run id
fails `turn.start` without spawning a provider; then a new run
cold-attaches and resumes the same conversation.
- The existing
`a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restart`
now also asserts that the turn-end failure saves `suspended` (the
turn-start failure stays `closed`). Everything else about the run's
failing turns is unchanged.

## Verification

- `cargo test -p paperclip-runner-core`:
  - `--test acpx_park_preparation`: 21 passed.
  - `--lib acpx`: 93 passed.
  - `acpx_park_readiness`: 8 passed.
  - `acpx_warm_attachment`: 4 passed.
  - `acpx_provider_suspend`: 5 passed.
  - `acpx_provider_session`: 12 passed.
  - `acpx_provider_turns`: 29 passed.
  - `native_provider_backend`: 47 passed.
  - `durable_recovery`: 3 passed.
- Live A/B on this branch (`ttft-baseline --adapters claude --samples 3
--phases coldwarm`): follow-ups resumed 3/3 (1/3 before), follow-up
first activity p50 6.2 s (about 20–23 s before).
- `cargo clippy --all-targets`: no findings on changed lines.

## Risks

- **Late-callback guarantee:** unchanged. The failed restart already
reaped the old generation, and the next generation starts only via
`run.attach` or a later reopen. That's the same resume path a failed
park preparation uses.
- **Fence wait:** the failure path now waits for the reaped provider's
lifetime fence before saving, matching park recovery. If the fence never
clears, the state stays exit-unconfirmed and fails closed.
- **Rollback:** `AcpxDurableState` denies unknown fields. The new field
is written only between a failed turn-end restart and the next run's
attach. A rollback inside that window makes the older Runner refuse the
saved state, which still fails closed, as master's `closed` state does
today.
- **Scope:** the cause of restart failures under load isn't changed
here. This makes their failure mode safe.

## Model Used

Claude Opus 5.5 (`claude-opus-5-5`, Anthropic), 1M context window, tool
use and code execution, via Claude Code.

## 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/acpx_park_preparation.rs

@@ -43,15 +43,18 @@ impl AcpxCommandExecutor {
         let Err(error) = restarted else {
             return Ok(());
         };
-        let state = self
-            .state
-            .as_mut()
-            .expect("ACPX state remains available after a turn-end restart");
-        state.lifecycle = "closed".to_owned();
-        self.context.provider_turn_id = None;
-        self.session = None;
+        // Later turns of this run fail closed, exactly like a failed restart at
+        // turn start. The saved session stays resumable: the reaped provider is
+        // proven gone, then it is saved suspended, so the next run resumes the
+        // same conversation instead of starting a new one.
         self.failed_turn_end_restart = Some(error.to_string());
-        self.save_state()
+        self.unreported_turn_end_restart_failure = Some(error.to_string());
+        let run_id = self.context.run_id.clone();
+        self.state
+            .as_mut()
+            .expect("ACPX state remains available after a failed turn-end restart")
+            .turn_end_restart_failed_run = Some(run_id);
+        self.recover_failed_generation_restart()
     }
 
     pub(super) fn prepare_session_for_park(
@@ -70,7 +73,7 @@ impl AcpxCommandExecutor {
         match prepared {
             Ok(restarted) => Ok(restarted),
             Err(error) => {
-                self.recover_failed_park_preparation()?;
+                self.recover_failed_generation_restart()?;
                 Err(DurableRunnerError::invalid(format!(
                     "ACPX park preparation failed: {error}"
                 )))
@@ -78,20 +81,23 @@ impl AcpxCommandExecutor {
         }
     }
 
-    fn recover_failed_park_preparation(&mut self) -> Result<(), DurableRunnerError> {
+    /// A provider generation restart (at park or after a turn) failed: wait until
+    /// the reaped provider loses its lifetime fence, then save the session
+    /// suspended so a later run can resume it.
+    fn recover_failed_generation_restart(&mut self) -> Result<(), DurableRunnerError> {
         self.session = None;
         self.context.provider_turn_id = None;
         let candidates = {
             let state = self
                 .state
                 .as_mut()
-                .expect("ACPX state remains available after park preparation");
+                .expect("ACPX state remains available after a failed generation restart");
             state.lifecycle = "prepared".to_owned();
             state.provider_exit_unconfirmed = true;
             state
                 .identity
                 .as_ref()
-                .expect("park preparation requires a provider identity")
+                .expect("a generation restart requires a provider identity")
                 .provider_lifetime_fence_candidates
         };
         self.save_state()?;
@@ -99,7 +105,7 @@ impl AcpxCommandExecutor {
         let state = self
             .state
             .as_mut()
-            .expect("ACPX state remains available after park preparation");
+            .expect("ACPX state remains available after a failed generation restart");
         state.provider_exit_unconfirmed = false;
         state.lifecycle = "suspended".to_owned();
         if let Err(error) = self.save_state() {

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

@@ -627,6 +627,11 @@ struct AcpxDurableState {
     /// never sees their results; the controller may still deliver them.
     #[serde(default)]
     retired_semantic_calls: BTreeSet<String>,
+    /// The run whose turn-end provider restart failed. Its later turns fail
+    /// closed, even after a Runner process restart; the next run's attach
+    /// clears it and resumes the saved session. Written only while set.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    turn_end_restart_failed_run: Option<String>,
 }
 
 /// The saved launch names the same executable and sidecar files as this
@@ -677,6 +682,7 @@ impl AcpxDurableState {
             next_event_sequence: initial_event_sequence(),
             open_semantic_calls: BTreeSet::new(),
             retired_semantic_calls: BTreeSet::new(),
+            turn_end_restart_failed_run: None,
         }
     }
 
@@ -746,6 +752,10 @@ impl AcpxDurableState {
                 .is_some_and(|result| !result.is_object())
             || (self.identity.is_none()
                 && !matches!(self.lifecycle.as_str(), "prepared" | "closed"))
+            || self
+                .turn_end_restart_failed_run
+                .as_ref()
+                .is_some_and(|run_id| *run_id != self.descriptor.run_id)
         {
             return Err(DurableRunnerError::invalid(
                 "ACPX durable provider state is malformed or inconsistent",
@@ -966,7 +976,10 @@ pub struct AcpxCommandExecutor {
     fail_after_state_rename: bool,
     launch_profile: Option<AcpxLaunchProfile>,
     recovery_pending_effects: Option<usize>,
+    /// The failed turn-end restart's cause, reported by the next turn start.
     failed_turn_end_restart: Option<String>,
+    /// The failed turn-end restart's cause, until the Runner records it.
+    unreported_turn_end_restart_failure: Option<String>,
 }
 
 // The durable command's turn and the live provider callback have separate IDs.
@@ -1007,6 +1020,7 @@ impl AcpxCommandExecutor {
             restore_checked: false,
             recovery_pending_effects: None,
             failed_turn_end_restart: None,
+            unreported_turn_end_restart_failure: None,
             restore_error: None,
             persistence_error: None,
             #[cfg(test)]
@@ -1216,6 +1230,10 @@ impl AcpxCommandExecutor {
         if state.descriptor.run_id != self.context.run_id {
             return Ok(());
         }
+        // This run's turn-end provider restart failed; its turns fail closed.
+        if state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str()) {
+            return Ok(());
+        }
         if !matches!(
             state.lifecycle.as_str(),
             "session_open" | "turn_starting" | "turn_active" | "suspended"
@@ -1738,12 +1756,26 @@ impl AcpxCommandExecutor {
         state.semantic_result = None;
         state.pending_events.clear();
         state.lifecycle = "suspended".to_owned();
-        if cold_candidate_attachment {
+        // A new run starts a fresh provider generation; the previous run's
+        // failed turn-end restart no longer applies to it. A same-run attach
+        // keeps the restriction.
+        let clears_turn_end_restart_failure = state
+            .turn_end_restart_failed_run
+            .as_ref()
+            .is_some_and(|run_id| *run_id != attachment_context.run_id);
+        if clears_turn_end_restart_failure {
+            state.turn_end_restart_failed_run = None;
+        }
+        let saved = if cold_candidate_attachment {
             self.commit_staged_state(state)
         } else {
             self.state = Some(state);
             self.save_state()
+        };
+        if saved.is_ok() && clears_turn_end_restart_failure {
+            self.failed_turn_end_restart = None;
         }
+        saved
     }
 
     fn open_session(&mut self) -> Result<CommandExecution, DurableRunnerError> {
@@ -1756,6 +1788,17 @@ impl AcpxCommandExecutor {
                 "ACPX provider lifetime cleanup is not yet proven",
             ));
         }
+        // No path (session.open, a same-run run.attach) reopens a session whose
+        // run failed closed at its turn-end provider restart.
+        if self.session.is_none()
+            && self.state.as_ref().is_some_and(|state| {
+                state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str())
+            })
+        {
+            return Err(DurableRunnerError::invalid(
+                "ACPX session failed closed after this run's turn-end provider restart failed",
+            ));
+        }
         if self.session.is_none() {
             let recovering = self
                 .state
@@ -1929,17 +1972,21 @@ impl AcpxCommandExecutor {
 
     fn start_turn(&mut self, payload: &Value) -> Result<CommandExecution, DurableRunnerError> {
         turn_start_request(payload)?;
-        if let Some(error) = self.failed_turn_end_restart.take() {
-            if self.session.is_none()
-                && self
-                    .state
-                    .as_ref()
-                    .is_some_and(|state| state.lifecycle == "closed")
-            {
-                return Err(DurableRunnerError::invalid(format!(
-                    "ACPX turn start failed closed: {error}"
-                )));
-            }
+        // A failed turn-end restart fails every later turn of this run closed,
+        // like a failed restart at turn start: the first names the cause, later
+        // ones find no open session. The restriction is saved, so it outlasts a
+        // Runner process restart; only the next run's attach clears it.
+        if self.session.is_none()
+            && self.state.as_ref().is_some_and(|state| {
+                state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str())
+            })
+        {
+            return Err(match self.failed_turn_end_restart.take() {
+                Some(error) => {
+                    DurableRunnerError::invalid(format!("ACPX turn start failed closed: {error}"))
+                }
+                None => DurableRunnerError::invalid("ACPX session is not open"),
+            });
         }
         let reopenable = self.session.is_none()
             && self.state.as_ref().is_some_and(|state| {
@@ -2927,6 +2974,10 @@ impl CommandExecutor for AcpxCommandExecutor {
         self.restart_session_after_turn()
     }
 
+    fn take_turn_end_restart_failure(&mut self) -> Option<String> {
+        self.unreported_turn_end_restart_failure.take()
+    }
+
     fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError> {
         self.restore()?;
         if self

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

@@ -327,6 +327,12 @@ pub trait CommandExecutor {
         Ok(())
     }
 
+    /// The cause of a provider restart after a turn that failed and was
+    /// recovered, once; recorded as a runner diagnostic.
+    fn take_turn_end_restart_failure(&mut self) -> Option<String> {
+        None
+    }
+
     fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError>;
 
     fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> {
@@ -2175,6 +2181,10 @@ fn poll_executor_events_after_controller_ack<E: CommandExecutor>(
     poll_executor_events(state, store, config, executor)?;
     if state.acked_source_seq >= state.highest_source_seq() {
         executor.prepare_next_turn()?;
+        if let Some(failure) = executor.take_turn_end_restart_failure() {
+            state.record_diagnostic(format!("turn-end provider restart failed: {failure}"));
+            store.save(state)?;
+        }
     }
     Ok(())
 }

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

@@ -351,6 +351,13 @@ impl CommandExecutor for NativeProviderCommandExecutor {
         }
     }
 
+    fn take_turn_end_restart_failure(&mut self) -> Option<String> {
+        match self.selected.as_mut() {
+            Some(SelectedExecutor::Acpx(executor)) => executor.take_turn_end_restart_failure(),
+            _ => None,
+        }
+    }
+
     fn observe_recovery_state(
         &mut self,
         state: &crate::durable::DurableState,

packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs

@@ -237,14 +237,22 @@ fn a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restar
             starts,
         )
     };
-    let at_turn_end = outcome(true);
+    let mut at_turn_end = outcome(true);
     assert!(
         at_turn_end.0.starts_with("ACPX turn start failed closed:")
             && at_turn_end.0.contains("session.open"),
         "{}",
         at_turn_end.0
     );
-    assert_eq!(at_turn_end, outcome(false));
+    // The turns of this run fail the same way. Unlike a failed restart at turn
+    // start, the failed turn-end restart saves a resumable session, so the next
+    // run resumes the conversation (see the test below).
+    assert_eq!(at_turn_end.2, "suspended");
+    let mut at_turn_start = outcome(false);
+    assert_eq!(at_turn_start.2, "closed");
+    at_turn_end.2 = Value::Null;
+    at_turn_start.2 = Value::Null;
+    assert_eq!(at_turn_end, at_turn_start);
 }
 
 #[test]
@@ -331,3 +339,148 @@ fn an_active_goal_or_a_request_at_the_terminal_keeps_today_s_restart_timing() {
     assert_eq!(at_turn_end.3, at_turn_start.3);
     eprintln!("request at terminal: {at_turn_end:?}");
 }
+
+/// A turn-end restart that fails (for example a slow or loaded host) used to
+/// save the session `closed`. The next run's `run.attach` then refused it and
+/// the controller started a new conversation. The failed restart now saves a
+/// resumable session once the reaped provider is proven gone: later turns of
+/// this run still fail closed, and the next run resumes the same conversation,
+/// in this Runner process or in a new one.
+#[test]
+fn the_next_run_resumes_the_conversation_after_a_failed_turn_end_restart() {
+    for new_runner_process in [false, true] {
+        let mut harness = Harness::new();
+        let old_pid = harness.opened["processId"].clone();
+        let conversation = harness.opened["providerSessionId"].clone();
+        harness.complete_turn("first");
+        harness.fixture.fault("reject-open");
+        harness.executor.prepare_next_turn().unwrap();
+        assert!(!alive(&old_pid));
+        // The Runner records the cause once as a diagnostic.
+        let reported = harness.executor.take_turn_end_restart_failure();
+        assert!(
+            reported
+                .as_deref()
+                .is_some_and(|cause| cause.contains("session.open")),
+            "{reported:?}"
+        );
+        assert_eq!(harness.executor.take_turn_end_restart_failure(), None);
+        let saved = harness.fixture.state();
+        assert_eq!(saved["lifecycle"], "suspended", "{saved}");
+        assert_eq!(saved["providerExitUnconfirmed"], false);
+        let failed = harness
+            .executor
+            .execute(&command(
+                90,
+                "turn.start",
+                json!({"turnId": "second", "text": "Complete the work."}),
+            ))
+            .unwrap_err()
+            .to_string();
+        assert!(
+            failed.starts_with("ACPX turn start failed closed:"),
+            "{failed}"
+        );
+        harness.sequence = 90;
+        harness.fixture.fault("");
+        let attached = if new_runner_process {
+            // A new Runner process for the next run cold-attaches the saved session.
+            harness.executor.shutdown().unwrap();
+            harness.config.run_id = "run-2".to_owned();
+            harness.executor = NativeProviderCommandExecutor::with_runner_config(
+                &harness.fixture.0,
+                &harness.config,
+            );
+            let payload = harness.attach_payload("run-2", |payload| {
+                payload
+                    .as_object_mut()
+                    .unwrap()
+                    .remove("paperclipNextAuthority");
+            });
+            harness.execute("run.attach", payload)
+        } else {
+            harness.attach("run-2", |_| {}).0
+        };
+        assert_eq!(
+            attached.result["providerSessionId"], conversation,
+            "new_runner_process={new_runner_process}"
+        );
+        harness.complete_turn("next-run");
+        assert_eq!(harness.fixture.state()["descriptor"]["runId"], "run-2");
+        harness.executor.shutdown().unwrap();
+    }
+}
+
+#[test]
+fn a_runner_restart_in_the_same_run_keeps_the_failed_turn_end_restart_closed() {
+    let mut harness = Harness::new();
+    let conversation = harness.opened["providerSessionId"].clone();
+    harness.complete_turn("first");
+    harness.fixture.fault("reject-open");
+    harness.executor.prepare_next_turn().unwrap();
+    harness.fixture.fault("");
+    let saved = harness.fixture.state();
+    assert_eq!(
+        saved["turnEndRestartFailedRun"], harness.config.run_id,
+        "{saved}"
+    );
+    // The Runner process restarts inside the same run: its turns still fail
+    // closed instead of reopening the saved session.
+    harness.executor.shutdown().unwrap();
+    harness.executor =
+        NativeProviderCommandExecutor::with_runner_config(&harness.fixture.0, &harness.config);
+    let starts = harness.fixture.process_starts().len();
+    let failed = harness
+        .executor
+        .execute(&command(
+            90,
+            "turn.start",
+            json!({"turnId": "second", "text": "Complete the work."}),
+        ))
+        .unwrap_err()
+        .to_string();
+    assert_eq!(failed, "ACPX session is not open");
+    assert_eq!(harness.fixture.process_starts().len(), starts);
+    // Neither session.open nor a same-run attach reopens it.
+    let reopened = harness
+        .executor
+        .execute(&command(91, "session.open", json!({})))
+        .unwrap_err()
+        .to_string();
+    assert!(reopened.contains("failed closed"), "{reopened}");
+    let same_run_attach = json!({"provider": harness.descriptor.clone()});
+    let attach_error = harness
+        .executor
+        .execute(&command(92, "run.attach", same_run_attach))
+        .unwrap_err()
+        .to_string();
+    assert!(attach_error.contains("failed closed"), "{attach_error}");
+    assert_eq!(harness.fixture.process_starts().len(), starts);
+    assert_eq!(
+        harness.fixture.state()["turnEndRestartFailedRun"],
+        harness.config.run_id
+    );
+    assert_eq!(live_providers(&harness), Vec::<Value>::new());
+    assert_eq!(harness.fixture.state()["lifecycle"], "suspended");
+    // The next run still resumes the conversation and clears the restriction.
+    harness.executor.shutdown().unwrap();
+    harness.sequence = 92;
+    harness.config.run_id = "run-2".to_owned();
+    harness.executor =
+        NativeProviderCommandExecutor::with_runner_config(&harness.fixture.0, &harness.config);
+    let payload = harness.attach_payload("run-2", |payload| {
+        payload
+            .as_object_mut()
+            .unwrap()
+            .remove("paperclipNextAuthority");
+    });
+    let attached = harness.execute("run.attach", payload);
+    assert_eq!(attached.result["providerSessionId"], conversation);
+    assert!(harness
+        .fixture
+        .state()
+        .get("turnEndRestartFailedRun")
+        .is_none());
+    harness.complete_turn("next-run");
+    harness.executor.shutdown().unwrap();
+}