From c33ba2c42c7a26e1ea550290aaf6697dcc6a28d9 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 9 Nov 2026 08:57:36 -0600 Subject: [PATCH] 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 --- .../runner-core/src/acpx_park_preparation.rs | 26 +++---- .../runner-core/src/acpx_provider_backend.rs | 28 +++++-- .../tests/support/acpx_turn_end_restart.rs | 74 ++++++++++++++++++- 3 files changed, 105 insertions(+), 23 deletions(-) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs index 26d4afb8638..f4a692695f9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs +++ b/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() { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index 94c33a1ee44..dfbeb6f1c83 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -967,6 +967,8 @@ pub struct AcpxCommandExecutor { launch_profile: Option, recovery_pending_effects: Option, failed_turn_end_restart: Option, + /// 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 { @@ -1929,17 +1939,19 @@ impl AcpxCommandExecutor { fn start_turn(&mut self, payload: &Value) -> Result { 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| { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs index 181dc19fd06..887358b289b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs +++ b/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(); + } +}