fix(runner): keep a failed turn-end restart closed across a same-run Runner restart

The fail-closed restriction lived only in memory, so a Runner process
restart inside the same run reopened the saved suspended session and ran
later turns. Save the blocked run id in the ACPX state (written only while
set), skip same-run restore and fail turn.start while it names this run,
and clear it only when a different run attaches.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

@@ -49,6 +49,11 @@ impl AcpxCommandExecutor {
         // same conversation instead of starting a new one.
         self.failed_turn_end_restart = Some(error.to_string());
         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()
     }
 

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,9 +976,8 @@ 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 already failed a turn of this run closed.
-    turn_end_restart_failed_turn: bool,
     /// The failed turn-end restart's cause, until the Runner records it.
     unreported_turn_end_restart_failure: Option<String>,
 }
@@ -1011,7 +1020,6 @@ impl AcpxCommandExecutor {
             restore_checked: false,
             recovery_pending_effects: None,
             failed_turn_end_restart: None,
-            turn_end_restart_failed_turn: false,
             unreported_turn_end_restart_failure: None,
             restore_error: None,
             persistence_error: None,
@@ -1222,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"
@@ -1744,17 +1756,24 @@ impl AcpxCommandExecutor {
         state.semantic_result = None;
         state.pending_events.clear();
         state.lifecycle = "suspended".to_owned();
+        // 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()
         };
-        // 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() {
+        if saved.is_ok() && clears_turn_end_restart_failure {
             self.failed_turn_end_restart = None;
-            self.turn_end_restart_failed_turn = false;
         }
         saved
     }
@@ -1944,17 +1963,19 @@ impl AcpxCommandExecutor {
         turn_start_request(payload)?;
         // 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"));
-            }
+        // 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| {

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

@@ -410,3 +410,58 @@ fn the_next_run_resumes_the_conversation_after_a_failed_turn_end_restart() {
         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);
+    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 = 90;
+    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();
+}