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 1b6f91c5bfa..26d4afb8638 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 @@ -1,18 +1,15 @@ use std::time::Instant; use super::{ - await_provider_lifetime_fence, AcpxCommandExecutor, AcpxProviderDescriptor, AcpxProviderSession, + await_provider_lifetime_fence, goal_is_dormant, AcpxCommandExecutor, AcpxProviderDescriptor, + AcpxProviderSession, }; use crate::durable::DurableRunnerError; use crate::provider_bridge::AuthorizedToolSet; impl AcpxCommandExecutor { - pub(super) fn prepare_session_for_park( - &mut self, - deadline: Instant, - ) -> Result { - self.ensure_persistence_healthy()?; - let eligible = self.candidate_admission().is_none() + fn generation_restartable(&self, session_ready: fn(&AcpxProviderSession) -> bool) -> bool { + self.candidate_admission().is_none() && self.state.as_ref().is_some_and(|state| { state.descriptor.agent == "claude" && state.lifecycle == "session_open" @@ -22,12 +19,47 @@ impl AcpxCommandExecutor { && !state.provider_exit_unconfirmed && state.pending_events.is_empty() && state.pending_runtime_requests.is_empty() + && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection)) }) - && self - .session - .as_ref() - .is_some_and(AcpxProviderSession::can_prepare_next_generation); - if !eligible { + && self.session.as_ref().is_some_and(session_ready) + } + + pub(super) fn restart_session_after_turn(&mut self) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; + if !self.generation_restartable(AcpxProviderSession::can_restart_after_turn) { + return Ok(()); + } + let session = self.session.as_mut().expect("checked ACPX session"); + if session.settle_park_observation().is_err() + || !self.generation_restartable(AcpxProviderSession::can_restart_after_turn) + { + return Ok(()); + } + let restarted = self + .session + .as_mut() + .expect("checked ACPX session") + .restart_after_turn(None); + 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; + self.failed_turn_end_restart = Some(error.to_string()); + self.save_state() + } + + pub(super) fn prepare_session_for_park( + &mut self, + deadline: Instant, + ) -> Result { + self.ensure_persistence_healthy()?; + if !self.generation_restartable(AcpxProviderSession::can_prepare_next_generation) { return Ok(false); } let prepared = self @@ -36,7 +68,7 @@ impl AcpxCommandExecutor { .expect("checked ACPX session") .prepare_next_generation(deadline); match prepared { - Ok(()) => Ok(true), + Ok(restarted) => Ok(restarted), Err(error) => { self.recover_failed_park_preparation()?; Err(DurableRunnerError::invalid(format!( 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 9063b2da0e1..94c33a1ee44 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 @@ -966,6 +966,7 @@ pub struct AcpxCommandExecutor { fail_after_state_rename: bool, launch_profile: Option, recovery_pending_effects: Option, + failed_turn_end_restart: Option, } // The durable command's turn and the live provider callback have separate IDs. @@ -1005,6 +1006,7 @@ impl AcpxCommandExecutor { session: None, restore_checked: false, recovery_pending_effects: None, + failed_turn_end_restart: None, restore_error: None, persistence_error: None, #[cfg(test)] @@ -1927,6 +1929,18 @@ 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}" + ))); + } + } let reopenable = self.session.is_none() && self.state.as_ref().is_some_and(|state| { state.identity.is_some() @@ -1997,7 +2011,7 @@ impl AcpxCommandExecutor { let (provider_process_will_be_replaced, previous_process_id) = self .session .as_ref() - .map(|session| match session.prepared_from_process_id() { + .map(|session| match session.restarted_after_turn_from() { Some(process_id) => (true, process_id), None => (session.generation_has_run_turn(), session.process_id()), }) @@ -2611,7 +2625,7 @@ impl AcpxCommandExecutor { || self .session .as_ref() - .is_none_or(AcpxProviderSession::awaits_run_rebind) + .is_none_or(AcpxProviderSession::awaits_next_turn) { return Ok(()); } @@ -2622,7 +2636,7 @@ impl AcpxCommandExecutor { if self .session .as_ref() - .is_some_and(AcpxProviderSession::awaits_run_rebind) + .is_some_and(AcpxProviderSession::awaits_next_turn) { return Ok(()); } @@ -2909,6 +2923,10 @@ impl CommandExecutor for AcpxCommandExecutor { self.prepare_session_for_park(deadline) } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + self.restart_session_after_turn() + } + fn execute(&mut self, command: &Command) -> Result { self.restore()?; if self diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs index cd826f6df07..6d7cd78c6ba 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs @@ -54,18 +54,20 @@ impl AcpxProviderSession { self.generation_has_run_turn } + pub fn restarted_after_turn_from(&self) -> Option { + self.restarted_after_turn_from + } + pub fn awaits_run_rebind(&self) -> bool { self.prepared_generation.is_some() } - pub fn prepared_from_process_id(&self) -> Option { - self.prepared_generation - .as_ref() - .map(|(_, process_id)| *process_id) + pub fn awaits_next_turn(&self) -> bool { + self.prepared_generation.is_some() || self.restarted_after_turn_from.is_some() } pub(super) fn park_tail_is_quiet(&self) -> bool { - if self.prepared_generation.is_none() { + if !self.awaits_next_turn() { return self.transport.buffered_events().len() == 0; } self.transport.buffered_events().all(|event| { @@ -78,65 +80,70 @@ impl AcpxProviderSession { }) } - pub fn can_prepare_next_generation(&self) -> bool { + fn at_settled_turn_boundary(&self) -> bool { !self.closed && self.failure.is_none() && !self.transport_terminated && !self.runtime_retired - && self.generation_has_run_turn && self.completed_turn_observed && self.prepared_generation.is_none() && self.config.candidate_admission.is_none() && self.state.active_turn_id().is_none() && !self.state.has_pending_requests() && !self.state.has_pending_tools() - && self.transport.buffered_events().len() == 0 + && self.park_tail_is_quiet() } - pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result<(), LocalRunnerError> { + pub fn can_restart_after_turn(&self) -> bool { + self.generation_has_run_turn && self.at_settled_turn_boundary() + } + + pub fn can_prepare_next_generation(&self) -> bool { + (self.generation_has_run_turn || self.restarted_after_turn_from.is_some()) + && self.at_settled_turn_boundary() + } + + pub fn restart_after_turn( + &mut self, + deadline: Option, + ) -> Result<(), LocalRunnerError> { + if !self.can_restart_after_turn() { + return Err(LocalRunnerError::invalid( + "ACPX provider is not at a settled turn boundary", + )); + } + let previous_process_id = self.process_id(); + if let Err(error) = self.restart_idle_provider(deadline) { + return Err(self.fail_closed(error)); + } + self.restarted_after_turn_from = Some(previous_process_id); + Ok(()) + } + + pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result { if !self.can_prepare_next_generation() { return Err(LocalRunnerError::invalid( "ACPX provider is not at a reusable generation boundary", )); } - let previous_process_id = self.process_id(); - let mut tool_bridge = self.tool_bridge.clone(); - let mut reserved_tool_bridge = self.reserved_tool_bridge.clone(); - tool_bridge - .prepare_turn_after_provider_restart() - .and_then(|()| reserved_tool_bridge.prepare_turn_after_provider_restart()) - .map_err(|error| { - LocalRunnerError::invalid(format!( - "ACPX tool receipt rotation before park failed: {error}" - )) - })?; - if let Err(error) = self.restart_idle_provider(Some(deadline)) { - return Err(self.fail_closed(error)); - } - if self.state.settled_turn_identity_capacity_reached() { - if let Err(error) = self - .state - .rotate_settled_turn_identities_after_provider_restart() - { + let restarted = self.generation_has_run_turn; + if restarted { + self.restart_after_turn(Some(deadline))?; + if let Err(error) = self.verify_dormant_session() { return Err(self.fail_closed(error)); } + self.transport.bound_requests_by(None); } - self.tool_bridge = tool_bridge; - self.reserved_tool_bridge = reserved_tool_bridge; self.completed_turn_observed = false; - if let Err(error) = self.verify_dormant_session() { - return Err(self.fail_closed(error)); - } - self.transport.bound_requests_by(None); - self.prepared_generation = Some((generation_digest(&self.config), previous_process_id)); - Ok(()) + self.prepared_generation = Some(generation_digest(&self.config)); + Ok(restarted) } pub fn rebind_run( &mut self, next: &AcpxProviderSessionConfig, ) -> Result { - let Some((prepared, _)) = self.prepared_generation.as_ref() else { + let Some(prepared) = self.prepared_generation.as_ref() else { return Ok(false); }; if self.closed @@ -194,6 +201,7 @@ impl AcpxProviderSession { self.config = next.clone(); self.catalog_revision = next.catalog_revision; self.working_directory = next.working_directory.clone(); + self.restarted_after_turn_from = None; self.prepared_generation = None; if let Err(error) = self.verify_dormant_session() { return Err(self.fail_closed(error)); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index a81d3db3742..e588b0a4d68 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -357,7 +357,8 @@ pub struct AcpxProviderSession { runtime_retired: bool, completed_turn_observed: bool, generation_has_run_turn: bool, - prepared_generation: Option<(String, u32)>, + restarted_after_turn_from: Option, + prepared_generation: Option, } pub(crate) fn goal_is_dormant(projection: &Value) -> bool { @@ -417,6 +418,7 @@ impl AcpxProviderSession { runtime_retired: false, completed_turn_observed: false, generation_has_run_turn: false, + restarted_after_turn_from: None, prepared_generation: None, }) } @@ -660,12 +662,13 @@ impl AcpxProviderSession { // provider turn originated a late HTTP callback. Reap the old sidecar // and provider before releasing its call-ID tombstones, then resume // the same verified persistent session in a fresh process generation. - let provider_restarted = self.generation_has_run_turn; - if provider_restarted { + let restart_now = self.generation_has_run_turn; + if restart_now { if let Err(error) = self.restart_idle_provider(None) { return Err(self.fail_closed(error)); } } + let provider_restarted = restart_now || self.restarted_after_turn_from.is_some(); if rotate_turn_identity_ledger { if let Err(error) = self .state @@ -700,6 +703,7 @@ impl AcpxProviderSession { )))); } self.generation_has_run_turn = true; + self.restarted_after_turn_from = None; self.prepared_generation = None; let response = match self.transport.request_turn_start( json!({"turnId":turn_id,"message":message,"performance":performance}), @@ -2256,6 +2260,7 @@ mod mode_tests { runtime_retired: false, completed_turn_observed: false, generation_has_run_turn: false, + restarted_after_turn_from: None, prepared_generation: None, } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs index 76a6a309f09..cfba5cc3ba9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs @@ -299,6 +299,20 @@ fn run() -> Result<(), Box> { source_pending_completion = Some(turn.to_owned()); continue; } + if fault == "pending-input" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.input_requested", + &source_run, + turn, + json!({"requestId":"input-at-terminal","questionSet":{ + "schema":"paperclip.question_set.v1", + "questions":[{"id":"target","prompt":"Which target?","required":true, + "answerMode":"single_select","options":[{"id":"first","label":"First"}]}]}}), + )?; + next_sequence += 1; + } write_turn_event( &mut stdout, next_sequence, 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 5a45df27369..d786da4a324 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 @@ -323,6 +323,10 @@ pub trait CommandExecutor { Ok(false) } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + Ok(()) + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -2168,7 +2172,11 @@ fn poll_executor_events_after_controller_ack( if state.acked_source_seq < sent_source_seq { return executor.maintain_backpressured_provider(); } - poll_executor_events(state, store, config, executor) + poll_executor_events(state, store, config, executor)?; + if state.acked_source_seq >= state.highest_source_seq() { + executor.prepare_next_turn()?; + } + Ok(()) } fn poll_executor_events_when_control_idle( @@ -5075,6 +5083,98 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn turn_boundary_preparation_waits_until_the_controller_acks_the_terminal_event() { + struct BoundaryExecutor { + events: VecDeque, + calls: Vec<&'static str>, + } + impl CommandExecutor for BoundaryExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("the poll gate cannot execute a command") + } + fn poll_events(&mut self) -> Result, DurableRunnerError> { + self.calls.push("poll"); + Ok(self.events.iter().cloned().collect()) + } + fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> { + self.events.drain(..count); + Ok(()) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("maintain"); + Ok(()) + } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("prepare"); + Ok(()) + } + } + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-turn-boundary-{}", + uuid::Uuid::new_v4() + )); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = BoundaryExecutor { + events: VecDeque::from([PolledEvent { + executor_event_id: "provider-terminal".to_owned(), + event_type: "turn.completed".to_owned(), + priority: EventPriority::P0, + payload: json!({"status": "completed"}), + }]), + calls: Vec::new(), + }; + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 0) + .unwrap(); + assert_eq!(executor.calls, ["poll"]); + assert_eq!(state.highest_source_seq(), 1); + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 0) + .unwrap(); + assert_eq!( + executor.calls, + ["poll", "poll"], + "an unsent terminal is not out" + ); + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 1) + .unwrap(); + assert_eq!(executor.calls[2..], ["maintain"]); + state.apply_ack(1, 2).unwrap(); + store.save(&state).unwrap(); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Ok(Some(json!({"kind": "command"}))), + ) + .unwrap(); + assert_eq!(executor.calls[3..], ["maintain"], "control frames go first"); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Err(DurableRunnerError::invalid("disconnected")), + ) + .unwrap(); + assert_eq!(executor.calls.len(), 4); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Ok(None), + ) + .unwrap(); + assert_eq!(executor.calls[4..], ["poll", "prepare"]); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn controller_ack_debt_runs_cleanup_without_moving_event_or_ack_cursors() { struct MaintenanceExecutor { 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 f1f593634d9..13551eb72b9 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 @@ -344,6 +344,13 @@ impl CommandExecutor for NativeProviderCommandExecutor { } } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + match self.selected.as_mut() { + Some(SelectedExecutor::Acpx(executor)) => executor.prepare_next_turn(), + _ => Ok(()), + } + } + fn observe_recovery_state( &mut self, state: &crate::durable::DurableState, diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs index 0f7fcf96fc7..f9d9a3072ca 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs @@ -14,6 +14,9 @@ use paperclip_runner_core::provider_bridge::{authorized_tool_catalog_digest, Aut use serde_json::{json, Value}; use sha2::{Digest, Sha256}; +#[path = "support/acpx_turn_end_restart.rs"] +mod turn_end_restart; + struct Fixture(PathBuf); impl Drop for Fixture { fn drop(&mut self) { @@ -267,8 +270,9 @@ fn last_process_start(journal: &[Value]) -> usize { fn prepared_park(harness: &mut Harness) -> Value { harness.complete_turn("first"); + harness.executor.prepare_next_turn().unwrap(); assert!(harness.executor.host_park_ready().unwrap()); - assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); harness.fixture.process_starts().last().unwrap()["processStart"].clone() } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs index 4b56a5abc4c..f1994f3b21a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs @@ -265,8 +265,12 @@ fn hosted_park_acpx_requires_completion_in_this_lifetime_and_exact_authority() { )) .unwrap(); settle(&mut executor); - assert!(executor.host_park_ready().unwrap()); - assert!(executor + executor.prepare_next_turn().unwrap(); + assert!( + executor.host_park_ready().unwrap(), + "a generation restarted at the turn boundary is park-ready" + ); + assert!(!executor .prepare_for_park(Instant::now() + Duration::from_secs(20)) .unwrap()); assert!( @@ -479,3 +483,45 @@ fn hosted_park_acpx_does_not_enable_an_unqualified_park_provider() { .unwrap()); executor.shutdown().unwrap(); } + +#[test] +fn a_turn_end_restart_waits_until_an_abandoned_park_observation_tail_is_delivered() { + let (fixture, _config, mut executor, opened) = fixture("claude"); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"first", "text":"Complete first work."}), + )) + .unwrap(); + settle(&mut executor); + fs::write(fixture.0.join("fault"), "slow-snapshot").unwrap(); + assert!(!executor.host_park_ready().unwrap()); + fs::write(fixture.0.join("fault"), "").unwrap(); + let starts = || { + fs::read_to_string(fixture.0.join("sidecar.jsonl")) + .unwrap() + .lines() + .filter(|line| line.contains("processStart")) + .count() + }; + executor.prepare_next_turn().unwrap(); + assert_eq!( + starts(), + 1, + "the old generation's tail is not yet delivered" + ); + assert!(std::process::Command::new("kill") + .args(["-0", &opened["processId"].to_string()]) + .status() + .unwrap() + .success()); + let events = executor.poll_events().unwrap(); + assert!(events + .iter() + .any(|event| event.payload["code"] == "slow_park_observation")); + executor.acknowledge_events(events.len()).unwrap(); + executor.prepare_next_turn().unwrap(); + assert_eq!(starts(), 2); + executor.shutdown().unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs index 1294ea1adf5..f31bb5e1da5 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -934,3 +934,83 @@ fn prepared_generation_never_admits_the_reaped_generation_late_callback() { session.shutdown("test complete").unwrap(); std::fs::remove_dir_all(directory).unwrap(); } + +#[test] +fn a_turn_end_restart_reaps_the_late_callback_and_releases_receipts_at_the_next_turn() { + let mut session = AcpxProviderSession::start(&config("turns-late-tool-after-suspend")).unwrap(); + session + .start_turn("turn-1", "Please continue", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(!session.can_restart_after_turn()); + session + .deliver_tool_result(&late_result("call-reused")) + .unwrap(); + let first_pid = session.process_id(); + session.restart_after_turn(None).unwrap(); + let fresh_pid = session.process_id(); + assert_ne!(fresh_pid, first_pid); + assert_eq!(session.restarted_after_turn_from(), Some(first_pid)); + assert!(!session.generation_has_run_turn() && !session.can_restart_after_turn()); + assert!(session + .poll_event(Duration::from_millis(200)) + .unwrap() + .is_none()); + + session + .start_turn("turn-2", "Use the fresh provider", &std::env::temp_dir()) + .unwrap(); + assert_eq!(session.process_id(), fresh_pid); + assert_eq!(session.restarted_after_turn_from(), None); + let fresh = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &fresh[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, input } + if call_id == "call-reused" + && operation_id == "issues.read" + && input["id"] == "issue-1" + )); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn an_interrupted_or_retired_turn_restarts_at_the_next_turn_start() { + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(!session.can_restart_after_turn()); + session + .interrupt_turn("turn-1", "Paperclip interruption") + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + let interrupted_pid = session.process_id(); + assert!(!session.can_restart_after_turn()); + assert!(session.restart_after_turn(None).is_err()); + assert_eq!(session.process_id(), interrupted_pid); + session + .start_turn("turn-2", "Start again", &std::env::temp_dir()) + .unwrap(); + assert_ne!(session.process_id(), interrupted_pid); + session.shutdown("test complete").unwrap(); + + let mut retired = AcpxProviderSession::start(&config("turns-retired")).unwrap(); + retired + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + retired + .interrupt_turn("turn-1", "Paperclip cancellation") + .unwrap(); + retired.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(retired.state().active_turn_id().is_none()); + assert!(retired.runtime_retired()); + assert!(!retired.can_restart_after_turn()); + retired.shutdown("test complete").unwrap(); +} 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 new file mode 100644 index 00000000000..181dc19fd06 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs @@ -0,0 +1,333 @@ +use super::*; + +const WARM_TURN: [&str; 2] = ["turn.start", "session.goal.get"]; + +fn live_providers(harness: &Harness) -> Vec { + harness + .fixture + .process_starts() + .into_iter() + .map(|line| line["processStart"].clone()) + .filter(alive) + .collect() +} + +fn settle(harness: &mut Harness, turn: &str) -> Vec { + let events = harness.complete_turn(turn); + harness.executor.prepare_next_turn().unwrap(); + events +} + +fn start_follow_up(harness: &mut Harness, turn: &str) -> (Vec, Vec, Duration) { + let before = harness.fixture.journal().len(); + let started = Instant::now(); + let execution = harness.execute( + "turn.start", + json!({"turnId": turn, "text": "Complete the work."}), + ); + let elapsed = started.elapsed(); + let mut events = execution + .events + .iter() + .map(|(kind, _, payload)| json!({"type":kind,"payload":payload})) + .collect::>(); + let deadline = Instant::now() + Duration::from_secs(5); + while !events + .iter() + .any(|event: &Value| event["type"] == "turn.completed") + { + assert!(Instant::now() < deadline, "no completed provider turn"); + events.extend(harness.drain()); + } + events.extend(harness.drain()); + let commands = commands_after(&harness.fixture.journal(), before); + (events, commands, elapsed) +} + +fn reconciled(events: &[Value]) -> &Value { + &events + .iter() + .find(|event| event["type"] == "session.reconciled") + .expect("the turn reports the replaced provider process")["payload"] +} + +#[test] +fn a_settled_turn_restarts_its_provider_and_the_next_turn_starts_without_one() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let before = harness.fixture.journal().len(); + let started = Instant::now(); + harness.executor.prepare_next_turn().unwrap(); + let restart = started.elapsed(); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert_eq!( + commands_after(&journal[..spawned], before), + ["session.suspend"] + ); + assert_eq!(journal[spawned]["livePredecessors"], json!([])); + assert_eq!( + commands_after(&journal, spawned), + ["initialize", "session.open", "run.attach"] + ); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-1"); + let fresh_pid = journal[spawned]["processStart"].clone(); + assert!(!alive(&old_pid)); + assert_eq!(live_providers(&harness), [fresh_pid.clone()]); + assert_eq!( + harness.drain(), + Vec::::new(), + "the fresh generation's output waits for its first turn" + ); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 2); + + let (events, commands, turn_start) = start_follow_up(&mut harness, "second"); + eprintln!("turn-end restart={restart:?} follow-up turn.start={turn_start:?}"); + assert_eq!(commands, WARM_TURN); + assert_eq!(reconciled(&events)["previousProcessId"], old_pid); + assert_eq!(reconciled(&events)["processId"], fresh_pid); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == fresh_pid)); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 3); + assert_eq!(live_providers(&harness).len(), 1); + let (_, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + harness.executor.shutdown().unwrap(); + assert_eq!(live_providers(&harness), Vec::::new()); +} + +#[test] +fn park_reuses_the_generation_a_settled_turn_restarted() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + let fresh_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(harness.executor.host_park_ready().unwrap()); + let before = harness.fixture.journal().len(); + let started = Instant::now(); + assert!( + !harness.executor.prepare_for_park(park_deadline()).unwrap(), + "park neither restarts nor asks for a second proof" + ); + eprintln!( + "park preparation after a turn-end restart={:?}", + started.elapsed() + ); + assert_eq!( + commands_after(&harness.fixture.journal(), before), + Vec::::new() + ); + assert!(harness.executor.host_park_ready().unwrap()); + assert_eq!(harness.fixture.process_starts().len(), 2); + + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["processId"], fresh_pid); + let events = settle(&mut harness, "second"); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == fresh_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + let (events, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + assert_eq!(reconciled(&events)["previousProcessId"], fresh_pid); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn the_reaped_generation_late_callback_never_reaches_the_next_turn() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("late-tool-after-suspend"); + harness.executor.prepare_next_turn().unwrap(); + harness.fixture.fault(""); + assert_eq!(harness.drain(), Vec::::new()); + let (events, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands, WARM_TURN); + assert!(!format!("{events:?}").contains("late-call")); + assert!(!format!("{events:?}").contains("late-old-generation")); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_turn_that_arrives_before_the_turn_end_restart_restarts_at_its_start() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let (events, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands[..2], ["session.suspend", "initialize"]); + assert_eq!(reconciled(&events)["previousProcessId"], old_pid); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!(live_providers(&harness).len(), 1); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 3); + assert_eq!(live_providers(&harness).len(), 1); + let (_, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_park_before_the_turn_end_restart_restarts_once() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!(live_providers(&harness).len(), 1); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["processId"], live_providers(&harness)[0]); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restart() { + let outcome = |turn_end: bool| { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + if turn_end { + harness.executor.prepare_next_turn().unwrap(); + assert!(!alive(&old_pid)); + assert!(!harness.executor.host_park_ready().unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert_eq!(harness.drain(), Vec::::new()); + harness.executor.prepare_next_turn().unwrap(); + } + let first = harness + .executor + .execute(&command( + 90, + "turn.start", + json!({"turnId": "second", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + harness.fixture.fault(""); + let retry = harness + .executor + .execute(&command( + 91, + "turn.start", + json!({"turnId": "third", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + assert_eq!(live_providers(&harness), Vec::::new()); + let state = harness.fixture.state(); + let starts = harness.fixture.process_starts().len(); + harness.executor.shutdown().unwrap(); + ( + first, + retry, + state["lifecycle"].clone(), + state["providerExitUnconfirmed"].clone(), + state["activeTurnId"].clone(), + state["pendingEvents"].clone(), + starts, + ) + }; + let 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)); +} + +#[test] +fn cancellation_after_a_turn_end_restart_is_already_settled() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + for kind in ["run.cancel", "turn.interrupt", "turn.stop"] { + let result = harness.execute(kind, json!({})).result; + assert_eq!(result["status"], "already_settled", "{kind}"); + } + let (_, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands, WARM_TURN); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn suspension_and_retirement_after_a_turn_end_restart_reap_the_fresh_generation() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + let fresh_pid = live_providers(&harness)[0].clone(); + harness.execute("runner.suspend", json!({})); + assert!(!alive(&fresh_pid)); + assert_eq!(harness.fixture.state()["lifecycle"], "suspended"); + let events = harness.complete_turn("second"); + assert!(events + .iter() + .any(|event| event["type"] == "session.resumed")); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.prepare_next_turn().unwrap(); + let fresh_pid = live_providers(&harness)[0].clone(); + assert_eq!(harness.fixture.process_starts().len(), 4); + harness.executor.shutdown().unwrap(); + assert!(!alive(&fresh_pid)); +} + +#[test] +fn an_active_goal_or_a_request_at_the_terminal_keeps_today_s_restart_timing() { + let mut harness = Harness::new(); + harness.fixture.fault("goal"); + harness.complete_turn("first"); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!( + harness.fixture.process_starts().len(), + 1, + "an active goal can keep working after its prompt" + ); + harness.executor.shutdown().unwrap(); + + let resolve = |turn_end: bool| { + let mut harness = Harness::new(); + harness.fixture.fault("pending-input"); + let events = harness.complete_turn("first"); + harness.fixture.fault(""); + if turn_end { + harness.executor.prepare_next_turn().unwrap(); + } + let resolved = harness + .executor + .execute(&command( + 90, + "request.resolve", + json!({"requestId": "input-at-terminal", "answers": {"target": "first"}}), + )) + .map(|execution| execution.result) + .map_err(|error| error.to_string()); + let (_, commands, _) = start_follow_up(&mut harness, "second"); + let starts = harness.fixture.process_starts().len(); + harness.executor.shutdown().unwrap(); + ( + events + .iter() + .map(|event| event["type"].clone()) + .collect::>(), + resolved, + commands.contains(&"session.suspend".to_owned()), + starts, + ) + }; + let at_turn_end = resolve(true); + let at_turn_start = resolve(false); + assert_eq!(at_turn_end.0, at_turn_start.0); + assert_eq!(at_turn_end.1, at_turn_start.1); + assert_eq!(at_turn_end.3, at_turn_start.3); + eprintln!("request at terminal: {at_turn_end:?}"); +}