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..a68e5fce2fa 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,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() { 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..be646761028 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 @@ -627,6 +627,11 @@ struct AcpxDurableState { /// never sees their results; the controller may still deliver them. #[serde(default)] retired_semantic_calls: BTreeSet, + /// 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, } /// 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, recovery_pending_effects: Option, + /// The failed turn-end restart's cause, reported by the next turn start. failed_turn_end_restart: Option, + /// The failed turn-end restart's cause, until the Runner records it. + unreported_turn_end_restart_failure: Option, } // 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 { @@ -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 { 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 { + self.unreported_turn_end_restart_failure.take() + } + fn execute(&mut self, command: &Command) -> Result { self.restore()?; if self diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index d786da4a324..2386f4218d7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/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 { + None + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -2175,6 +2181,10 @@ fn poll_executor_events_after_controller_ack( 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(()) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs index 13551eb72b9..d906e5ffa03 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs +++ b/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 { + 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, 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..55a7a11f534 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,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::::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(); +}