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