diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 047f25aa087..cd88ba60094 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -162,3 +162,32 @@ after that drain, so a late active-goal notification cannot be hidden by the older executor journal. This hook does not detach a controller, stop a process, or advertise hosted parking support. The host-control consumer must separately validate the exact Runner identity and acknowledged sequence cut. + +### Hosted park and reclaim loop + +A trusted Launcher may pass a connected Unix stream at fd 4 and start durable +listener mode with `--host-control-fd 4`. The descriptor is close-on-exec before +provider startup. The private channel accepts the `park`, `claim`, and `retire` +requests defined in the host-control codec; it cannot execute PRP commands. + +`park` requires the exact six-field authority, acknowledged command/event cut, +no unfinished transition or cleanup, and the provider eligibility proof above. +The Runner drops its controller connection and credentials, retains its provider +and journals, and blocks on host control without provider polling. Ordinary +reconnect grace does not expire a parked Runner. The configured process lifetime +still applies. Mount ownership stays with Launcher. + +`claim` names the accepted park request and supplies a fresh bootstrap ticket. +It resumes authenticated PRP under the retained authority; ordinary `run.attach` +must still establish the next authority before new work. Pending reconnect and +live commands both reject execution before that attachment; snapshots, drain, +and cleanup remain available. Replaying the same claim returns its acknowledgement without consuming another ticket. A parked +retirement names the park request. Cancelling a claim additionally requires its +claim ID, the unchanged authority and sequence cut, and provider quiescence. +Retirement, parked host-channel loss, or process-lifetime expiry uses the existing +suspended lifecycle and provider shutdown path. There is no new checkpoint state. + +This native loop does not advertise hosted parking support: Launcher retention, +Core orchestration, provider-specific qualification, and live VM sleep/wake +verification are required before rollout. Providers without an eligibility +implementation remain ineligible. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 61c4dc2f7c3..4d3c6da791c 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -9,8 +9,8 @@ use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt}; use paperclip_runner_core::acpx_candidate_admission::AcpxCandidateAdmission; use paperclip_runner_core::durable::{ - capture_bootstrap_ticket, redact_diagnostic_text, run_durable_runner, AcpxLaunchProfile, - CursorCloudLaunchProfile, DurableRunnerConfig, OpenCodeLaunchProfile, QualifiedLaunchArtifact, + capture_bootstrap_ticket, redact_diagnostic_text, AcpxLaunchProfile, CursorCloudLaunchProfile, + DurableRunnerConfig, OpenCodeLaunchProfile, QualifiedLaunchArtifact, }; use paperclip_runner_core::local_runner::{run_local_runner, LocalRunnerError, RunnerConfig}; use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; @@ -417,6 +417,38 @@ fn durable_connect_url(args: &[String], run_id: &str) -> Result Result<(), LocalRunnerError> { + #[cfg(unix)] + let host_control = if let Some(fd) = optional_u64(args, "--host-control-fd")? { + use std::os::fd::{FromRawFd, OwnedFd}; + use std::os::unix::fs::FileTypeExt; + if fd != 4 + || !fs::metadata("/dev/fd/4") + .map(|metadata| metadata.file_type().is_socket()) + .unwrap_or(false) + { + return Err(LocalRunnerError::invalid( + "host control requires an inherited socket at fd 4", + )); + } + // SAFETY: fd 4 was inherited from the trusted host and verified open. + // No other Runner owner takes or closes it; HostControl sets CLOEXEC + // before any provider can spawn and validates connected AF_UNIX STREAM. + let socket = std::os::unix::net::UnixStream::from(unsafe { OwnedFd::from_raw_fd(4) }); + Some( + paperclip_runner_core::host_control::HostControl::new(socket) + .map_err(|error| LocalRunnerError::invalid(error.to_string()))?, + ) + } else { + None + }; + #[cfg(not(unix))] + let host_control = if args.iter().any(|arg| arg == "--host-control-fd") { + return Err(LocalRunnerError::invalid( + "host control is unavailable on this platform", + )); + } else { + None + }; let ticket = capture_bootstrap_ticket() .map_err(|error| LocalRunnerError::invalid(error.to_string()))? .ok_or_else(|| { @@ -462,6 +494,11 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { reconnect_grace: optional_u64(args, "--reconnect-grace-ms")?.map(Duration::from_millis), max_runtime: duration("--max-runtime-ms", 0)?, }; + if host_control.is_some() && !config.connect_url.starts_with("listen://") { + return Err(LocalRunnerError::invalid( + "host control requires trusted Runner listener mode", + )); + } #[cfg(unix)] std::env::set_var( paperclip_runner_core::runtime_notifications::CONTEXT_ENV, @@ -479,8 +516,13 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { } } let executor = NativeProviderCommandExecutor::with_runner_config(state_dir, &config); - run_durable_runner(config, ticket, executor) - .map_err(|error| LocalRunnerError::invalid(error.to_string())) + paperclip_runner_core::durable::run_durable_runner_with_host_control( + config, + ticket, + executor, + host_control, + ) + .map_err(|error| LocalRunnerError::invalid(error.to_string())) } fn apply_oom_score_adj(args: &[String]) -> Result<(), LocalRunnerError> { @@ -545,6 +587,11 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> { "ACPX candidate admission requires trusted durable runner startup", )); } + if args.iter().any(|arg| arg == "--host-control-fd") { + return Err(LocalRunnerError::invalid( + "host control requires trusted durable runner startup", + )); + } run_local_runner(RunnerConfig { run_id: value(args, "--run-id")?, normalized_session_id: value(args, "--session-id")?, @@ -678,6 +725,7 @@ mod tests { assert!(metadata.get("codexSourceRecoveryVersion").is_none()); assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1); assert!(metadata.get("acpxSourceRecoveryVersion").is_none()); + assert!(metadata.get("hostedParkingVersion").is_none()); assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA); assert_eq!(metadata["binaryContractVersion"], 2); assert_eq!( @@ -694,6 +742,20 @@ mod tests { ); } + #[test] + #[cfg(unix)] + fn host_control_rejects_other_fds_before_bootstrap_or_provider_startup() { + let args = vec!["--host-control-fd".to_owned(), "3".to_owned()]; + assert_eq!( + run_durable(&args).unwrap_err().to_string(), + "host control requires an inherited socket at fd 4" + ); + assert_eq!( + run(&args).unwrap_err().to_string(), + "host control requires trusted durable runner startup" + ); + } + #[test] #[cfg(all(target_os = "macos", target_arch = "aarch64"))] fn candidate_startup_requires_a_complete_trusted_launch_profile() { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs index 4b026797f66..20e21bd2a4a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs @@ -12,8 +12,8 @@ use sha2::{Digest, Sha256}; use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; pub use runner::{ - run_durable_runner, CommandExecution, CommandExecutor, PolledEvent, - TerminalDeliveryReconciliation, + run_durable_runner, run_durable_runner_with_host_control, CommandExecution, CommandExecutor, + PolledEvent, TerminalDeliveryReconciliation, }; pub(crate) use state::{ create_private_temporary_file, open_private_regular_file, redact_sensitive_text_values, 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 b1e7bf391a5..abd852e105c 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 @@ -447,10 +447,44 @@ impl CumulativeAckPersistence { } pub fn run_durable_runner( + config: DurableRunnerConfig, + bootstrap_ticket: BootstrapTicket, + executor: E, +) -> Result<(), DurableRunnerError> { + run_durable_runner_with_host_control(config, bootstrap_ticket, executor, None) +} + +#[cfg(not(unix))] +pub struct HostControl; +#[cfg(unix)] +use crate::host_control::{HostControl, HostIdentity, HostRequest}; + +#[cfg(unix)] +struct HostedRunnerControl { + channel: HostControl, + park_id: Option, + claiming: Option<(String, String)>, +} + +#[cfg(unix)] +impl HostedRunnerControl { + fn new(channel: HostControl) -> Self { + Self { + channel, + park_id: None, + claiming: None, + } + } +} + +pub fn run_durable_runner_with_host_control( mut config: DurableRunnerConfig, bootstrap_ticket: BootstrapTicket, mut executor: E, + #[allow(unused_variables)] host_control: Option, ) -> Result<(), DurableRunnerError> { + #[cfg(unix)] + let mut host_control = host_control.map(HostedRunnerControl::new); let started = Instant::now(); config.validate()?; if let Some(admission) = config @@ -508,7 +542,50 @@ pub fn run_durable_runner( let mut reconnect_attempt = 0_u32; let mut next_connection_attempt = Instant::now(); - loop { + 'session: loop { + #[cfg(unix)] + if let Some(control) = host_control.as_mut() { + if control.park_id.is_some() { + let remaining = (!config.max_runtime.is_zero()) + .then(|| config.max_runtime.saturating_sub(started.elapsed())); + if remaining.is_some_and(|duration| duration.is_zero()) { + persist_lifecycle_before_shutdown( + &mut state, + &store, + &mut executor, + "suspended", + )?; + return Ok(()); + } + match process_host_control( + control, + &mut state, + &store, + &mut executor, + true, + remaining, + ) { + Ok(HostAction::Claim(ticket)) => { + bootstrap_ticket = Some(ticket); + lease = None; + authenticated_once = false; + disconnected_since = None; + reconnect_attempt = 0; + } + Ok(HostAction::Retire) => return Ok(()), + Ok(_) => continue, + Err(error) => { + let _ = persist_lifecycle_before_shutdown( + &mut state, + &store, + &mut executor, + "suspended", + ); + return Err(error); + } + } + } + } if authenticated_once { let disconnected_at = disconnected_since.get_or_insert_with(Instant::now); if config @@ -584,6 +661,22 @@ pub fn run_durable_runner( continue; } let using_bootstrap = lease.is_none(); + #[cfg(unix)] + if let Some(control) = host_control.as_mut() { + match process_host_control(control, &mut state, &store, &mut executor, false, None) { + Ok(HostAction::Park) => { + lease = None; + bootstrap_ticket = None; + continue 'session; + } + Ok(HostAction::Retire) => return Ok(()), + Err(error) => { + let _ = shutdown_preserving_cleanup(&state, &mut executor); + return Err(error); + } + _ => {} + } + } let connect_deadline = connection_attempt_deadline( &config, started, @@ -743,6 +836,10 @@ pub fn run_durable_runner( &mut endpoint, next, )?; + #[cfg(unix)] + if let Some(control) = host_control.as_mut() { + control.claiming = None; + } } Err(error) => { state.record_diagnostic(format!( @@ -786,6 +883,8 @@ pub fn run_durable_runner( welcome.warm_transition_version, connection.protocol_version, )?; + #[cfg(unix)] + admit_host_claim(host_control.as_ref(), &command, next_authority.as_ref())?; let (result, lifecycle) = process_command(&mut state, &store, &config, &mut executor, &command)?; let next_authority = next_authority.filter(|_| completed_attachment(&result)); @@ -861,6 +960,10 @@ pub fn run_durable_runner( } if let Some(next) = authority_rotation { apply_authority_rotation(&mut state, &store, &mut config, &mut endpoint, next)?; + #[cfg(unix)] + if let Some(control) = host_control.as_mut() { + control.claiming = None; + } disconnected_since = Some(Instant::now()); continue; } @@ -994,6 +1097,34 @@ pub fn run_durable_runner( // Read control before starting another fsynced provider batch. // Consuming the last cumulative ACK must not let a new output // suffix overtake the stop/suspend already queued behind it. + #[cfg(unix)] + { + if let Some(control) = host_control.as_mut() { + match process_host_control( + control, + &mut state, + &store, + &mut executor, + false, + None, + ) { + Ok(HostAction::Park) => { + // The private owner now holds the exact idle cut. No + // old PRP frame, lease renewal, or provider poll may follow. + lease = None; + bootstrap_ticket = None; + drop(transport); + continue 'session; + } + Ok(HostAction::Retire) => return Ok(()), + Err(error) => { + let _ = shutdown_preserving_cleanup(&state, &mut executor); + return Err(error); + } + _ => {} + } + } + } let control_message = transport.receive_json(); poll_executor_events_when_control_idle( &mut state, @@ -1058,6 +1189,8 @@ pub fn run_durable_runner( welcome.warm_transition_version, connection.protocol_version, )?; + #[cfg(unix)] + admit_host_claim(host_control.as_ref(), &command, next_authority.as_ref())?; let (result, lifecycle) = process_command(&mut state, &store, &config, &mut executor, &command)?; let next_authority = next_authority.filter(|_| completed_attachment(&result)); @@ -1135,6 +1268,10 @@ pub fn run_durable_runner( &mut endpoint, next, )?; + #[cfg(unix)] + if let Some(control) = host_control.as_mut() { + control.claiming = None; + } disconnected_since = Some(Instant::now()); break; } @@ -1238,6 +1375,148 @@ pub fn run_durable_runner( } } +#[cfg(unix)] +enum HostAction { + None, + Park, + Claim(BootstrapTicket), + Retire, +} + +#[cfg(unix)] +fn admit_host_claim( + control: Option<&HostedRunnerControl>, + command: &Command, + next_authority: Option<&DurableRunnerConfig>, +) -> Result<(), DurableRunnerError> { + if control.is_some_and(|control| control.claiming.is_some()) + && !(matches!( + command.command_type.as_str(), + "session.snapshot" + | "runner.drain" + | "runner.suspend" + | "runner.shutdown" + | "turn.stop" + ) || (command.command_type == "run.attach" && next_authority.is_some())) + { + return Err(DurableRunnerError::invalid( + "claimed Runner requires authority attachment before execution", + )); + } + Ok(()) +} + +#[cfg(unix)] +fn process_host_control( + control: &mut HostedRunnerControl, + state: &mut DurableState, + store: &DurableStateStore, + executor: &mut E, + parked: bool, + remaining: Option, +) -> Result { + let Some(request) = control.channel.receive(parked, remaining)? else { + return Ok(HostAction::None); + }; + if control.channel.replay(&request)? { + return Ok(HostAction::None); + } + match &request { + HostRequest::Park { + id, + identity, + last_controller_command_seq, + acked_source_seq, + } if !parked + && control.park_id.is_none() + && control.claiming.is_none() + && host_park_cut_matches( + state, + identity, + *last_controller_command_seq, + *acked_source_seq, + ) + && executor.host_park_ready()? => + { + control.park_id = Some(id.clone()); + control.channel.reply(&request, "parked")?; + Ok(HostAction::Park) + } + HostRequest::Claim { + id, + park_id, + bootstrap_ticket, + .. + } if parked && control.park_id.as_ref() == Some(park_id) => { + let ticket = BootstrapTicket::new(bootstrap_ticket.clone())?; + control.claiming = Some((park_id.clone(), id.clone())); + control.park_id = None; + control.channel.reply(&request, "claimed")?; + Ok(HostAction::Claim(ticket)) + } + HostRequest::Retire { + park_id, + claim_id, + identity, + last_controller_command_seq, + acked_source_seq, + .. + } if match ( + claim_id, + identity, + last_controller_command_seq, + acked_source_seq, + ) { + (None, None, None, None) => parked && control.park_id.as_ref() == Some(park_id), + (Some(claim_id), Some(identity), Some(command_seq), Some(acked_seq)) => { + !parked + && control + .claiming + .as_ref() + .is_some_and(|(park, claim)| park == park_id && claim == claim_id) + && host_park_cut_matches(state, identity, *command_seq, *acked_seq) + && executor.host_park_ready()? + } + _ => false, + } => + { + persist_lifecycle_before_shutdown(state, store, executor, "suspended")?; + control.channel.reply(&request, "retired")?; + Ok(HostAction::Retire) + } + _ => { + control.channel.reply(&request, "conflict")?; + Ok(HostAction::None) + } + } +} + +#[cfg(unix)] +fn host_park_cut_matches( + state: &DurableState, + identity: &HostIdentity, + command_seq: u64, + acked_seq: u64, +) -> bool { + identity.runner_instance_id == state.runner_instance_id + && identity.environment_lease_id == state.environment_lease_id + && identity.run_id == state.run_id + && identity.normalized_session_id == state.normalized_session_id + && identity.turn_id == state.turn_id + && identity.item_id == state.item_id + && command_seq == state.last_controller_command_seq + && acked_seq == state.acked_source_seq + && state.outbox.is_empty() + && state.acked_source_seq == state.highest_source_seq() + && state.warm_transition.is_none() + && state.pending_terminal_delivery.is_none() + && state.pending_provider_cleanup.is_none() + && state + .processed_commands + .values() + .all(|command| !matches!(command.status.as_str(), "pending" | "indeterminate")) +} + fn lease_renewal_deadline(now: u64, expires_at: u64) -> u64 { now.saturating_add(expires_at.saturating_sub(now) / 2) } @@ -2721,6 +3000,271 @@ mod tests { } } + #[test] + #[cfg(unix)] + fn hosted_parking_requires_exact_cut_and_replays_claim_without_reusing_ticket() { + use std::io::{BufRead, BufReader, Write}; + use std::os::unix::net::UnixStream; + struct IdleExecutor { + proofs: usize, + shutdowns: usize, + ready: bool, + } + impl CommandExecutor for IdleExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("private control cannot execute PRP commands"); + } + fn host_park_ready(&mut self) -> Result { + self.proofs += 1; + Ok(self.ready) + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.shutdowns += 1; + Ok(()) + } + } + let directory = std::env::temp_dir().join(format!( + "paperclip-host-retirement-{}", + uuid::Uuid::new_v4() + )); + let config = config(directory.clone()); + let store = DurableStateStore::new(&config.state_dir).unwrap(); + let mut state = DurableState::new(&config); + let (socket, mut host) = UnixStream::pair().unwrap(); + let mut reader = BufReader::new(host.try_clone().unwrap()); + let mut control = HostedRunnerControl::new(HostControl::new(socket).unwrap()); + let mut executor = IdleExecutor { + proofs: 0, + shutdowns: 0, + ready: true, + }; + let park = json!({"op":"park","id":"park-1","identity":{ + "runnerInstanceId":state.runner_instance_id,"environmentLeaseId":state.environment_lease_id, + "runId":state.run_id,"normalizedSessionId":state.normalized_session_id,"turnId":state.turn_id,"itemId":state.item_id, + },"lastControllerCommandSeq":0,"ackedSourceSeq":0}); + let mut exchange = |request: &Value, + state: &mut DurableState, + control: &mut HostedRunnerControl, + executor: &mut IdleExecutor, + parked: bool| { + writeln!(host, "{request}").unwrap(); + let action = process_host_control( + control, + state, + &store, + executor, + parked, + Some(Duration::from_secs(1)), + ) + .unwrap(); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + (action, serde_json::from_str::(&line).unwrap()) + }; + state.last_controller_command_seq = 1; + assert_eq!( + exchange(&park, &mut state, &mut control, &mut executor, false).1["status"], + "conflict" + ); + state.last_controller_command_seq = 0; + assert!(matches!( + exchange(&park, &mut state, &mut control, &mut executor, false).0, + HostAction::Park + )); + assert!(matches!( + exchange(&park, &mut state, &mut control, &mut executor, true).0, + HostAction::None + )); + assert_eq!(executor.proofs, 1); + let claim = json!({"op":"claim","id":"claim-1","parkId":"park-1","bootstrapTicket":"fresh-private-ticket"}); + assert!(matches!( + exchange(&claim, &mut state, &mut control, &mut executor, true).0, + HostAction::Claim(_) + )); + assert!(matches!( + exchange(&claim, &mut state, &mut control, &mut executor, false).0, + HostAction::None + )); + assert!(admit_host_claim(Some(&control), &command("turn.start"), None).is_err()); + assert!(admit_host_claim(Some(&control), &command("session.snapshot"), None).is_ok()); + assert!(admit_host_claim(Some(&control), &command("run.attach"), None).is_err()); + let mut next_authority = config.clone(); + next_authority.run_id = "next-run".to_owned(); + assert!(admit_host_claim( + Some(&control), + &command("run.attach"), + Some(&next_authority) + ) + .is_ok()); + let mut changed = claim; + changed["bootstrapTicket"] = json!("different-private-ticket"); + assert_eq!( + exchange(&changed, &mut state, &mut control, &mut executor, false).1["status"], + "conflict" + ); + assert_eq!(executor.shutdowns, 0); + let mut premature_park = park.clone(); + premature_park["id"] = json!("park-before-attach"); + assert_eq!( + exchange( + &premature_park, + &mut state, + &mut control, + &mut executor, + false + ) + .1["status"], + "conflict" + ); + let retire = json!({"op":"retire","id":"retire-park","parkId":"park-1"}); + assert_eq!( + exchange(&retire, &mut state, &mut control, &mut executor, false).1["status"], + "conflict" + ); + let conditional_retire = json!({"op":"retire","id":"retire-claim","parkId":"park-1", + "claimId":"claim-1","identity":park["identity"], + "lastControllerCommandSeq":0,"ackedSourceSeq":0}); + for (field, value) in [ + ("parkId", json!("wrong-park")), + ("claimId", json!("wrong-claim")), + ("lastControllerCommandSeq", json!(1)), + ("ackedSourceSeq", json!(1)), + ] { + let mut stale = conditional_retire.clone(); + stale["id"] = json!(format!("stale-{field}")); + stale[field] = value; + assert_eq!( + exchange(&stale, &mut state, &mut control, &mut executor, false).1["status"], + "conflict" + ); + } + let mut wrong_identity = conditional_retire.clone(); + wrong_identity["id"] = json!("stale-identity"); + wrong_identity["identity"]["runId"] = json!("another-run"); + assert_eq!( + exchange( + &wrong_identity, + &mut state, + &mut control, + &mut executor, + false + ) + .1["status"], + "conflict" + ); + executor.ready = false; + assert_eq!( + exchange( + &conditional_retire, + &mut state, + &mut control, + &mut executor, + false + ) + .1["status"], + "conflict" + ); + assert_eq!(executor.shutdowns, 0); + executor.ready = true; + // Simulate completed run.attach before a later exact parked cut. + control.claiming = None; + assert!(admit_host_claim(Some(&control), &command("turn.start"), None).is_ok()); + let mut next_park = park.clone(); + next_park["id"] = json!("park-2"); + assert!(matches!( + exchange(&next_park, &mut state, &mut control, &mut executor, false).0, + HostAction::Park + )); + let retire = json!({"op":"retire","id":"retire-park","parkId":"park-2"}); + assert!(matches!( + exchange(&retire, &mut state, &mut control, &mut executor, true).0, + HostAction::Retire + )); + assert_eq!(executor.shutdowns, 1); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + #[cfg(unix)] + fn hosted_park_blocks_without_provider_polls_and_retires_on_eof_or_lifetime() { + use std::io::{BufRead, BufReader, Write}; + use std::os::unix::net::UnixStream; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + struct ParkedExecutor(Arc); + impl CommandExecutor for ParkedExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("parked provider cannot execute"); + } + fn host_park_ready(&mut self) -> Result { + Ok(true) + } + fn poll_events(&mut self) -> Result, DurableRunnerError> { + panic!("parked provider cannot poll"); + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + } + for eof in [false, true] { + let directory = std::env::temp_dir().join(format!( + "paperclip-hosted-park-stop-{}-{eof}", + std::process::id() + )); + let _ = fs::remove_dir_all(&directory); + let mut config = config(directory.clone()); + config.max_runtime = if eof { + Duration::ZERO + } else { + // Include durable startup under parallel-suite filesystem load. + Duration::from_secs(2) + }; + let (socket, mut host) = UnixStream::pair().unwrap(); + let control = HostControl::new(socket).unwrap(); + writeln!(host,"{}",json!({"op":"park","id":"park-stop","identity":{ + "runnerInstanceId":config.runner_instance_id,"environmentLeaseId":config.environment_lease_id, + "runId":config.run_id,"normalizedSessionId":config.normalized_session_id,"turnId":config.turn_id,"itemId":config.item_id, + },"lastControllerCommandSeq":0,"ackedSourceSeq":0})).unwrap(); + let shutdowns = Arc::new(AtomicUsize::new(0)); + let executor = ParkedExecutor(shutdowns.clone()); + let saved_config = config.clone(); + let task = std::thread::spawn(move || { + run_durable_runner_with_host_control( + config, + BootstrapTicket::new("unused-bootstrap".to_owned()).unwrap(), + executor, + Some(control), + ) + }); + let mut reply = String::new(); + BufReader::new(&host).read_line(&mut reply).unwrap(); + assert_eq!( + serde_json::from_str::(&reply).unwrap()["status"], + "parked" + ); + if eof { + drop(host); + assert!(task.join().unwrap().is_err()); + } else { + assert!(task.join().unwrap().is_ok()); + } + assert_eq!(shutdowns.load(Ordering::SeqCst), 1); + assert_eq!( + DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&saved_config) + .unwrap() + .0 + .lifecycle, + "suspended" + ); + fs::remove_dir_all(directory).unwrap(); + } + } + #[test] fn indeterminate_recovery_result_is_a_failed_wire_result() { let state = DurableState::new(&config(PathBuf::from("unused"))); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs new file mode 100644 index 00000000000..6f25a381c8c --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs @@ -0,0 +1,258 @@ +// Shares this module's authenticated PRP server helpers. +use crate::durable::{ + run_durable_runner_with_host_control, CommandExecution, CommandExecutor, DurableStateStore, + PolledEvent, +}; +use crate::host_control::HostControl; +use std::io::BufRead; +use std::os::unix::net::UnixStream; + +struct RetentionExecutor { + polls: Arc, + shutdowns: Arc, +} +impl CommandExecutor for RetentionExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("host control must not execute business commands"); + } + fn host_park_ready(&mut self) -> Result { + Ok(true) + } + fn poll_events(&mut self) -> Result, DurableRunnerError> { + self.polls.fetch_add(1, Ordering::SeqCst); + Ok(vec![]) + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.shutdowns.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +fn host_exchange(host: &mut UnixStream, request: Value) -> Value { + writeln!(host, "{request}").unwrap(); + let mut line = String::new(); + BufReader::new(host).read_line(&mut line).unwrap(); + serde_json::from_str(&line).unwrap() +} + +#[test] +fn hosted_loop_parks_and_reclaims_with_fresh_bootstrap_during_busy_prp() { + exercise_hosted_loop(None); +} + +#[test] +fn hosted_claim_blocks_pending_and_live_work_before_attachment() { + for pending in [true, false] { + exercise_hosted_loop(Some(pending)); + } +} + +fn exercise_hosted_loop(blocked_work: Option) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut config = config(listener.local_addr().unwrap().port()); + let directory = + std::env::temp_dir().join(format!("paperclip-host-loop-{}", uuid::Uuid::new_v4())); + config.state_dir = directory.clone(); + config.max_runtime = Duration::from_secs(15); + config.reconnect_grace = Some(Duration::from_millis(20)); + let state = test_state(&config); + let server_config = config.clone(); + let server_state = state.clone(); + let (busy_tx, busy_rx) = mpsc::channel(); + let (claimed_tx, claimed_rx) = mpsc::channel(); + let (finish_tx, finish_rx) = mpsc::channel(); + let server = thread::spawn(move || { + let expires = current_unix_ms().unwrap() + 60_000; + let (stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + stream + .set_write_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut socket = accept(stream).unwrap(); + let mut secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "initial-private-bootstrap", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + "connection-1", + Some("old-private-lease"), + expires, + 0, + vec![], + ), + ); + let mut announced = false; + // Keep PRP readable. Host control must not depend on an idle socket read. + loop { + let ping = control(&server_state, "connection-1", "ping", json!({})); + let encrypted = secure + .encrypt(&serde_json::to_vec(&ping).unwrap(), true) + .unwrap(); + if send_plain(&mut socket, &encrypted, server_config.max_frame_bytes).is_err() { + break; + } + let Ok(frame) = receive_plain(&mut socket, server_config.max_frame_bytes) else { + break; + }; + assert_eq!(secure.decrypt(&frame, true).unwrap()["kind"], "pong"); + if !announced { + busy_tx.send(()).unwrap(); + announced = true; + } + } + drop(socket); + // A parked Runner must accept the new bootstrap, never the old lease. + let (stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut socket = accept(stream).unwrap(); + let mut secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "fresh-private-bootstrap", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + let forbidden = json!({ + "schema":"paperclip.prp.command.v1", "commandId":"premature-turn", + "controllerSeq":1, "type":"turn.start", "issuedAt":"2026-01-01T00:00:00Z", "payload":{} + }); + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + "connection-2", + Some("new-private-lease"), + expires, + 0, + if blocked_work == Some(true) { vec![forbidden.clone()] } else { vec![] }, + ), + ); + if let Some(pending) = blocked_work { + if !pending { + send_secure(&mut socket, &mut secure, &server_config, + &control(&server_state, "connection-2", "command", forbidden)); + } + finish_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + return; + } + send_secure( + &mut socket, + &mut secure, + &server_config, + &control(&server_state, "connection-2", "ping", json!({})), + ); + assert_eq!( + receive_secure(&mut socket, &mut secure, &server_config)["kind"], + "pong" + ); + claimed_tx.send(()).unwrap(); + finish_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + }); + let polls = Arc::new(AtomicUsize::new(0)); + let shutdowns = Arc::new(AtomicUsize::new(0)); + let executor = RetentionExecutor { + polls: polls.clone(), + shutdowns: shutdowns.clone(), + }; + let (socket, mut host) = UnixStream::pair().unwrap(); + host.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let runner_config = config.clone(); + let runner = thread::spawn(move || { + run_durable_runner_with_host_control( + runner_config, + BootstrapTicket::new("initial-private-bootstrap".to_owned()).unwrap(), + executor, + Some(HostControl::new(socket).unwrap()), + ) + }); + busy_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + let identity = json!({"runnerInstanceId":state.runner_instance_id,"environmentLeaseId":state.environment_lease_id, + "runId":state.run_id,"normalizedSessionId":state.normalized_session_id,"turnId":state.turn_id,"itemId":state.item_id}); + assert_eq!( + host_exchange( + &mut host, + json!({"op":"park","id":"park-1","identity":identity, + "lastControllerCommandSeq":0,"ackedSourceSeq":0}) + )["status"], + "parked" + ); + let parked_polls = polls.load(Ordering::SeqCst); + // Outlast the ordinary reconnect grace without any polling or shutdown. + thread::sleep(Duration::from_millis(80)); + assert_eq!(polls.load(Ordering::SeqCst), parked_polls); + assert_eq!(shutdowns.load(Ordering::SeqCst), 0); + assert_eq!( + host_exchange( + &mut host, + json!({"op":"claim","id":"claim-1","parkId":"park-1", + "bootstrapTicket":"fresh-private-bootstrap"}) + )["status"], + "claimed" + ); + if blocked_work.is_some() { + let error = runner.join().unwrap().unwrap_err(); + assert_eq!(error.to_string(), "claimed Runner requires authority attachment before execution"); + finish_tx.send(()).unwrap(); + server.join().unwrap(); + let retained = DurableStateStore::new(&directory).unwrap().load_or_create(&config).unwrap().0; + assert_eq!(retained.last_controller_command_seq, 0); + assert!(retained.processed_commands.is_empty()); + std::fs::remove_dir_all(directory).unwrap(); + return; + } + claimed_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + // A stale parked-only retirement cannot kill a newly claimed process. + assert_eq!( + host_exchange( + &mut host, + json!({"op":"retire","id":"stale-retire","parkId":"park-1"}) + )["status"], + "conflict" + ); + assert_eq!( + host_exchange( + &mut host, + json!({"op":"retire","id":"retire-claim","parkId":"park-1", + "claimId":"claim-1","identity":identity,"lastControllerCommandSeq":0,"ackedSourceSeq":0}) + )["status"], + "retired" + ); + assert!(runner.join().unwrap().is_ok()); + finish_tx.send(()).unwrap(); + server.join().unwrap(); + assert_eq!(shutdowns.load(Ordering::SeqCst), 1); + assert_eq!( + DurableStateStore::new(&directory) + .unwrap() + .load_or_create(&config) + .unwrap() + .0 + .lifecycle, + "suspended" + ); + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index cda7ac7ffc7..569297dcbfc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -1747,6 +1747,9 @@ fn hex_nibble(byte: u8) -> Result { #[cfg(test)] mod tests { + #[cfg(unix)] + include!("tests/hosted_retention.rs"); + use std::net::{IpAddr, Ipv4Addr, TcpListener}; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering};