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 93e7e4005c..9e8134c365 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 @@ -3,7 +3,7 @@ use std::fs::{self, DirBuilder, File}; use std::io::{Read, Write}; use std::net::TcpListener; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, Instant}; #[cfg(unix)] use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; @@ -38,6 +38,8 @@ const ACPX_PROVIDER_STATE_SCHEMA: &str = "paperclip.runner.acpx-provider-state.v const MAX_PROVIDER_STATE_BYTES: u64 = 16 * 1024 * 1024; const MAX_PENDING_EVENTS: usize = 8_320; const MAX_EVENTS_PER_POLL: usize = 128; +const PROVIDER_LIFETIME_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5); +const PROVIDER_LIFETIME_CONFIRMATION_RETRY: Duration = Duration::from_millis(10); fn initial_event_sequence() -> u64 { 1 @@ -52,16 +54,16 @@ fn event_sequence(value: &str) -> Option { (event_id(sequence) == value).then_some(sequence) } -fn acquire_provider_lifetime_fence( +fn try_acquire_provider_lifetime_fence( candidates: [u16; 3], -) -> Result, DurableRunnerError> { +) -> Result>, DurableRunnerError> { let mut listeners = Vec::with_capacity(2); for port in candidates { match TcpListener::bind(("127.0.0.1", port)) { Ok(listener) => { listeners.push(listener); if listeners.len() == 2 { - return Ok(listeners); + return Ok(Some(listeners)); } } Err(error) if error.kind() == std::io::ErrorKind::AddrInUse => {} @@ -72,9 +74,34 @@ fn acquire_provider_lifetime_fence( } } } - Err(DurableRunnerError::invalid( - "ACPX original provider lifetime remains active; cleanup is not yet proven", - )) + Ok(None) +} + +fn acquire_provider_lifetime_fence( + candidates: [u16; 3], +) -> Result, DurableRunnerError> { + try_acquire_provider_lifetime_fence(candidates)?.ok_or_else(|| { + DurableRunnerError::invalid( + "ACPX original provider lifetime remains active; cleanup is not yet proven", + ) + }) +} + +fn await_provider_lifetime_fence( + candidates: [u16; 3], +) -> Result, DurableRunnerError> { + let deadline = Instant::now() + PROVIDER_LIFETIME_CONFIRMATION_TIMEOUT; + loop { + if let Some(listeners) = try_acquire_provider_lifetime_fence(candidates)? { + return Ok(listeners); + } + if Instant::now() >= deadline { + return Err(DurableRunnerError::invalid( + "ACPX original provider lifetime remains active after suspension; provider exit is not confirmed", + )); + } + std::thread::sleep(PROVIDER_LIFETIME_CONFIRMATION_RETRY); + } } #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] @@ -956,15 +983,27 @@ impl AcpxCommandExecutor { "reason": reason, }))); }; - self.session - .as_mut() - .ok_or_else(|| DurableRunnerError::invalid("ACPX session is unavailable"))? - .terminate_active_turn_for_suspension(&turn_id) - .map_err(|error| { - DurableRunnerError::invalid(format!( - "failed to terminate ACPX turn at the suspension boundary: {error}" - )) - })?; + let provider_lifetime_fence_candidates = { + let session = self + .session + .as_mut() + .ok_or_else(|| DurableRunnerError::invalid("ACPX session is unavailable"))?; + let candidates = session.identity().provider_lifetime_fence_candidates; + session + .terminate_active_turn_for_suspension(&turn_id) + .map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to terminate ACPX turn at the suspension boundary: {error}" + )) + })?; + candidates + }; + // Process-group termination reaps the sidecar leader and its ordinary + // descendants, but an escaped provider or guardian can outlive that + // group. Require the inherited listener quorum before making this + // durable session attachable, and retain it through the state write. + let _provider_lifetime_fence = + await_provider_lifetime_fence(provider_lifetime_fence_candidates)?; self.session = None; let state = self .state @@ -1879,6 +1918,20 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn suspension_waits_for_the_original_provider_lifetime_quorum() { + let (candidates, original_lifetime_fence) = reserve_provider_lifetime_fence(); + let releaser = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(25)); + drop(original_lifetime_fence); + }); + + let confirmed = await_provider_lifetime_fence(candidates) + .expect("suspension must wait until the original lifetime loses quorum"); + assert_eq!(confirmed.len(), 2); + releaser.join().unwrap(); + } + fn reserve_provider_lifetime_fence() -> ([u16; 3], Vec) { let mut listeners = Vec::new(); for port in 49_152..=u16::MAX { 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 ce5ced7913..a6dc391d47 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 @@ -701,9 +701,10 @@ impl AcpxProviderSession { /// A governed Paperclip result can settle the run while the model is still /// waiting for its semantic-tool callback to unwind. In that state the /// ordinary sidecar close path may wait for the callback longer than the - /// server process that owns this runner. Process-group termination is the - /// fail-closed boundary: only after it proves the old generation is gone - /// may runnerd persist the durable session as attachable by the next run. + /// server process that owns this runner. Process-group termination closes + /// this session's transport authority. The caller must additionally + /// acquire the identity's inherited lifetime-fence quorum before + /// persisting the durable session as attachable. pub fn terminate_active_turn_for_suspension( &mut self, turn_id: &str, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs index 4ea6d727ae..2d22ca9ff7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs @@ -820,6 +820,8 @@ impl SupervisedProcess { })?; // The group leader can exit while descendants remain alive. Reap the // leader first, then clear any remaining members of its private group. + // A caller that must exclude escaped or re-parented descendants still + // needs a separately inherited lifetime fence. #[cfg(unix)] signal_process_group(self.process_group_id, "KILL"); self.finished = true;