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

Since #20190 the ACPX provider restarts after every turn. When that restart
failed (on a loaded host), the session was saved closed, so the next run's
run.attach refused it, spent about 10 s reconnecting, and the controller
started a new Claude conversation: 2 of 3 follow-ups on a3a25ce6 in a local
A/B, 0 of 3 before #20145/#20190. A failed turn-end restart now recovers
like a failed park preparation: once the reaped provider loses its lifetime
fence the session is saved suspended. Later turns of the run still fail
closed exactly as before, and the next run's attach resumes the same
conversation, in the same Runner process or a new one.

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,12 @@ 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.recover_failed_generation_restart()
     }
 
     pub(super) fn prepare_session_for_park(
@@ -70,7 +67,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 +75,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 +99,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

@@ -967,6 +967,8 @@ pub struct AcpxCommandExecutor {
     launch_profile: Option<AcpxLaunchProfile>,
     recovery_pending_effects: Option<usize>,
     failed_turn_end_restart: Option<String>,
+    /// The failed turn-end restart already failed a turn of this run closed.
+    turn_end_restart_failed_turn: bool,
 }
 
 // The durable command's turn and the live provider callback have separate IDs.
@@ -1007,6 +1009,7 @@ impl AcpxCommandExecutor {
             restore_checked: false,
             recovery_pending_effects: None,
             failed_turn_end_restart: None,
+            turn_end_restart_failed_turn: false,
             restore_error: None,
             persistence_error: None,
             #[cfg(test)]
@@ -1738,12 +1741,19 @@ impl AcpxCommandExecutor {
         state.semantic_result = None;
         state.pending_events.clear();
         state.lifecycle = "suspended".to_owned();
-        if cold_candidate_attachment {
+        let saved = if cold_candidate_attachment {
             self.commit_staged_state(state)
         } else {
             self.state = Some(state);
             self.save_state()
+        };
+        // A new run starts a fresh provider generation; the previous run's
+        // failed turn-end restart no longer applies to it.
+        if saved.is_ok() {
+            self.failed_turn_end_restart = None;
+            self.turn_end_restart_failed_turn = false;
         }
+        saved
     }
 
     fn open_session(&mut self) -> Result<CommandExecution, DurableRunnerError> {
@@ -1929,17 +1939,19 @@ 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")
-            {
+        // 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. Only the next run's attach clears it.
+        if self.session.is_none() {
+            if let Some(error) = self.failed_turn_end_restart.take() {
+                self.turn_end_restart_failed_turn = true;
                 return Err(DurableRunnerError::invalid(format!(
                     "ACPX turn start failed closed: {error}"
                 )));
             }
+            if self.turn_end_restart_failed_turn {
+                return Err(DurableRunnerError::invalid("ACPX session is not open"));
+            }
         }
         let reopenable = self.session.is_none()
             && self.state.as_ref().is_some_and(|state| {

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,65 @@ 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));
+        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();
+    }
+}