diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs index 70a8adb2fad..414715f168d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs @@ -38,10 +38,6 @@ impl AcpxEventScope { self.active_turn_id.as_deref() } - pub(crate) fn has_settled_turns(&self) -> bool { - !self.settled_turn_ids.is_empty() - } - pub fn bind_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { let turn_id = turn_id.into(); validate_scope_id(&turn_id, "turn", DURABLE_STABLE_ID_CHARS)?; 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 new file mode 100644 index 00000000000..1b6f91c5bfa --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs @@ -0,0 +1,114 @@ +use std::time::Instant; + +use super::{ + await_provider_lifetime_fence, 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() + && self.state.as_ref().is_some_and(|state| { + state.descriptor.agent == "claude" + && state.lifecycle == "session_open" + && state.descriptor.run_id == self.context.run_id + && state.identity.is_some() + && state.active_turn_id.is_none() + && !state.provider_exit_unconfirmed + && state.pending_events.is_empty() + && state.pending_runtime_requests.is_empty() + }) + && self + .session + .as_ref() + .is_some_and(AcpxProviderSession::can_prepare_next_generation); + if !eligible { + return Ok(false); + } + let prepared = self + .session + .as_mut() + .expect("checked ACPX session") + .prepare_next_generation(deadline); + match prepared { + Ok(()) => Ok(true), + Err(error) => { + self.recover_failed_park_preparation()?; + Err(DurableRunnerError::invalid(format!( + "ACPX park preparation failed: {error}" + ))) + } + } + } + + fn recover_failed_park_preparation(&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"); + state.lifecycle = "prepared".to_owned(); + state.provider_exit_unconfirmed = true; + state + .identity + .as_ref() + .expect("park preparation requires a provider identity") + .provider_lifetime_fence_candidates + }; + self.save_state()?; + let _provider_lifetime_fence = await_provider_lifetime_fence(candidates)?; + let state = self + .state + .as_mut() + .expect("ACPX state remains available after park preparation"); + state.provider_exit_unconfirmed = false; + state.lifecycle = "suspended".to_owned(); + if let Err(error) = self.save_state() { + let state = self + .state + .as_mut() + .expect("ACPX state remains available after save failure"); + state.provider_exit_unconfirmed = true; + state.lifecycle = "prepared".to_owned(); + return Err(error); + } + Ok(()) + } + + pub(super) fn rebind_prepared_session( + &mut self, + descriptor: &AcpxProviderDescriptor, + tool_set: &AuthorizedToolSet, + next_candidate: bool, + ) -> bool { + if next_candidate + || self.candidate_admission().is_some() + || !self + .session + .as_ref() + .is_some_and(AcpxProviderSession::awaits_run_rebind) + { + return false; + } + let Some(identity) = self.state.as_ref().and_then(|state| state.identity.clone()) else { + return false; + }; + let Ok(config) = descriptor.session_config( + tool_set.clone(), + Some(identity), + self.launch_profile.as_ref(), + ) else { + return false; + }; + self.session + .as_mut() + .is_some_and(|session| session.rebind_run(&config).unwrap_or(false)) + } +} 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 e16c4f3e7c2..9063b2da0e1 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 @@ -39,6 +39,9 @@ use crate::provider_events::{ }; use crate::qualified_launch::{verify_executable_launch_artifact, verify_launch_artifact}; +#[path = "acpx_park_preparation.rs"] +mod park_preparation; + fn bounded_turn_timing(response: &Value, elapsed: Duration) -> Value { let mut receipt = serde_json::Map::new(); for key in [ @@ -1703,16 +1706,18 @@ impl AcpxCommandExecutor { "run.attach requires the same settled ACPX provider profile and session", )); } - if let Some(session) = self.session.as_mut() { - session - .shutdown("run authority rotation") - .map_err(|error| { - DurableRunnerError::invalid(format!( - "failed to checkpoint ACPX before attaching a new run: {error}" - )) - })?; + if !self.rebind_prepared_session(&descriptor, &tool_set, next_candidate.is_some()) { + if let Some(session) = self.session.as_mut() { + session + .shutdown("run authority rotation") + .map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to checkpoint ACPX before attaching a new run: {error}" + )) + })?; + } + self.session = None; } - self.session = None; if let Some(admission) = next_candidate { self.launch_profile .as_mut() @@ -1982,7 +1987,7 @@ impl AcpxCommandExecutor { if self .session .as_ref() - .is_some_and(|session| session.state().has_settled_turns()) + .is_some_and(AcpxProviderSession::generation_has_run_turn) { self.retain_buffered_provider_events()?; } @@ -1992,7 +1997,10 @@ impl AcpxCommandExecutor { let (provider_process_will_be_replaced, previous_process_id) = self .session .as_ref() - .map(|session| (session.state().has_settled_turns(), session.process_id())) + .map(|session| match session.prepared_from_process_id() { + Some(process_id) => (true, process_id), + None => (session.generation_has_run_turn(), session.process_id()), + }) .expect("ACPX session exists after availability check"); { let state = self @@ -2600,7 +2608,10 @@ impl AcpxCommandExecutor { .state .as_ref() .is_some_and(|state| !state.pending_events.is_empty() || state.lifecycle == "closed") - || self.session.is_none() + || self + .session + .as_ref() + .is_none_or(AcpxProviderSession::awaits_run_rebind) { return Ok(()); } @@ -2608,6 +2619,13 @@ impl AcpxCommandExecutor { } fn retain_buffered_provider_events(&mut self) -> Result<(), DurableRunnerError> { + if self + .session + .as_ref() + .is_some_and(AcpxProviderSession::awaits_run_rebind) + { + return Ok(()); + } if let Some(session) = self.session.as_mut() { session.settle_park_observation().map_err(|error| { DurableRunnerError::invalid(format!("ACPX observation settlement failed: {error}")) @@ -2887,6 +2905,10 @@ impl CommandExecutor for AcpxCommandExecutor { })) } + fn prepare_for_park(&mut self, deadline: Instant) -> Result { + self.prepare_session_for_park(deadline) + } + 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 new file mode 100644 index 00000000000..cd826f6df07 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs @@ -0,0 +1,203 @@ +use std::time::Instant; + +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use super::{ + reserved_terminal_tool_bridge, sidecar_run_tool_operations, AcpxProviderSession, + AcpxProviderSessionConfig, +}; +use crate::acpx_provider_state::AcpxProviderState; +use crate::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GeneratedAcpxSidecarEventType, +}; +use crate::local_runner::LocalRunnerError; +use crate::provider_bridge::ProviderToolBridge; + +fn generation_digest(config: &AcpxProviderSessionConfig) -> String { + let mut runtime_context = config.runtime_context.clone(); + let mcp_bound = runtime_context + .pointer("/mcp/bindingId") + .is_some_and(|binding| !binding.is_null()); + if let Some(object) = runtime_context.as_object_mut() { + object.remove("mcp"); + object.remove("aggregateDigest"); + } + let shape = json!({ + "candidateAdmission": config.candidate_admission, + "command": config.transport.command, + "args": config.transport.args, + "agent": config.agent, + "model": config.model, + "catalogRevision": config.catalog_revision, + "runtimeDirectory": config.runtime_directory, + "normalizedSessionId": config.normalized_session_id, + "workingDirectory": config.working_directory, + "permissionMode": config.permission_mode, + "mode": config.mode, + "piThinkingLevel": config.pi_thinking_level, + "claudeEffort": config.claude_effort, + "grokEffort": config.grok_effort, + "grokDisableWebSearch": config.grok_disable_web_search, + "permissionModePinned": config.permission_mode_pinned, + "providerPolicy": config.provider_policy, + "systemInstructions": config.system_instructions, + "runtimeContext": runtime_context, + "mcpBound": mcp_bound, + "toolSet": config.tool_set, + }); + format!("sha256:{:x}", Sha256::digest(shape.to_string().as_bytes())) +} + +impl AcpxProviderSession { + pub fn generation_has_run_turn(&self) -> bool { + self.generation_has_run_turn + } + + 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(super) fn park_tail_is_quiet(&self) -> bool { + if self.prepared_generation.is_none() { + return self.transport.buffered_events().len() == 0; + } + self.transport.buffered_events().all(|event| { + event.turn_id.is_none() + && matches!( + event.event_type, + GeneratedAcpxSidecarEventType::RuntimeProcess + | GeneratedAcpxSidecarEventType::RuntimeDiagnostic + ) + }) + } + + pub fn can_prepare_next_generation(&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 + } + + pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result<(), LocalRunnerError> { + 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() + { + return Err(self.fail_closed(error)); + } + } + 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(()) + } + + pub fn rebind_run( + &mut self, + next: &AcpxProviderSessionConfig, + ) -> Result { + let Some((prepared, _)) = self.prepared_generation.as_ref() else { + return Ok(false); + }; + if self.closed + || self.failure.is_some() + || self.transport_terminated + || self.runtime_retired + || self.generation_has_run_turn + || self.state.active_turn_id().is_some() + || self.state.has_pending_requests() + || next.run_id == self.config.run_id + || next.expected_identity.as_ref() != Some(&self.identity) + || &generation_digest(next) != prepared + || next.validate().is_err() + { + return Ok(false); + } + let state = AcpxProviderState::new(&next.run_id)?; + let mut tool_bridge = ProviderToolBridge::default(); + tool_bridge + .prepare(next.tool_set.clone()) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}")) + })?; + let reserved_tool_bridge = reserved_terminal_tool_bridge()?; + let previous_run_id = self.config.run_id.clone(); + let attached = match self.transport.request( + GeneratedAcpxSidecarCommand::RunAttach, + json!({ + "runId": next.run_id, + "catalogRevision": next.catalog_revision, + "tools": sidecar_run_tool_operations(&next.tool_set), + }), + ) { + Ok(attached) => attached, + Err(error) => return Err(self.fail_closed(error)), + }; + if attached.get("runId").and_then(Value::as_str) != Some(next.run_id.as_str()) + || attached.get("catalogRevision").and_then(Value::as_u64) + != Some(next.catalog_revision) + { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX sidecar did not confirm the rebound run", + ))); + } + if self.transport.buffered_events().any(|event| { + event.turn_id.is_some() || event.run_id.as_deref() == Some(previous_run_id.as_str()) + }) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX prepared provider emitted work bound to its previous run", + ))); + } + self.state = state; + self.tool_bridge = tool_bridge; + self.reserved_tool_bridge = reserved_tool_bridge; + self.config = next.clone(); + self.catalog_revision = next.catalog_revision; + self.working_directory = next.working_directory.clone(); + self.prepared_generation = None; + if let Err(error) = self.verify_dormant_session() { + return Err(self.fail_closed(error)); + } + Ok(true) + } +} 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 bb2045e82ba..a81d3db3742 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 @@ -22,6 +22,9 @@ use crate::provider_bridge::{ use crate::question_response::validate_question_response; use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; +#[path = "acpx_provider_generation.rs"] +mod generation; + const MAX_ID_CHARS: usize = 240; const MAX_MODEL_CHARS: usize = 240; pub(crate) const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024; @@ -353,6 +356,8 @@ pub struct AcpxProviderSession { rejected_start_resumable: bool, runtime_retired: bool, completed_turn_observed: bool, + generation_has_run_turn: bool, + prepared_generation: Option<(String, u32)>, } pub(crate) fn goal_is_dormant(projection: &Value) -> bool { @@ -411,6 +416,8 @@ impl AcpxProviderSession { rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, + generation_has_run_turn: false, + prepared_generation: None, }) } @@ -506,10 +513,10 @@ impl AcpxProviderSession { || self.failure.is_some() || self.transport_terminated || self.runtime_retired - || !self.completed_turn_observed + || !(self.completed_turn_observed || self.prepared_generation.is_some()) || self.state.active_turn_id().is_some() || self.state.has_pending_requests() - || self.transport.buffered_events().len() != 0 + || !self.park_tail_is_quiet() { return Ok(false); } @@ -529,8 +536,7 @@ impl AcpxProviderSession { return Ok(false); }; // Polling still owns every frame observed during these reads. - Ok(self.validate_dormant_snapshot(&snapshot).is_ok() - && self.transport.buffered_events().len() == 0) + Ok(self.validate_dormant_snapshot(&snapshot).is_ok() && self.park_tail_is_quiet()) } /// Read the same live sidecar before exposing its in-memory pending ledger. @@ -654,9 +660,9 @@ 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.state.has_settled_turns(); + let provider_restarted = self.generation_has_run_turn; if provider_restarted { - if let Err(error) = self.restart_idle_provider() { + if let Err(error) = self.restart_idle_provider(None) { return Err(self.fail_closed(error)); } } @@ -693,6 +699,8 @@ impl AcpxProviderSession { "ACPX reserved tool receipt rotation failed: {error}" )))); } + self.generation_has_run_turn = true; + self.prepared_generation = None; let response = match self.transport.request_turn_start( json!({"turnId":turn_id,"message":message,"performance":performance}), ) { @@ -1307,7 +1315,8 @@ impl AcpxProviderSession { Ok(()) } - fn restart_idle_provider(&mut self) -> Result<(), LocalRunnerError> { + fn restart_idle_provider(&mut self, deadline: Option) -> Result<(), LocalRunnerError> { + self.transport.bound_requests_by(deadline); let suspended = self.transport.request( GeneratedAcpxSidecarCommand::SessionSuspend, json!({"reason":"ACPX turn receipt epoch rotation"}), @@ -1322,6 +1331,7 @@ impl AcpxProviderSession { &restart_config.transport, &restart_config.agent, )?; + replacement.bound_requests_by(deadline); let (replacement_identity, _, turn_controls) = match bootstrap(&mut replacement, &restart_config) { Ok(value) => value, @@ -1340,6 +1350,7 @@ impl AcpxProviderSession { self.transport = replacement; self.turn_controls = turn_controls; self.transport_terminated = false; + self.generation_has_run_turn = false; Ok(()) } @@ -2244,6 +2255,8 @@ mod mode_tests { rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, + generation_has_run_turn: false, + prepared_generation: None, } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs index c55718e3453..58f537c07a8 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -155,10 +155,6 @@ impl AcpxProviderState { self.scope.active_turn_id() } - pub(crate) fn has_settled_turns(&self) -> bool { - self.scope.has_settled_turns() - } - pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> { self.scope.validate_new_turn_identity(turn_id) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs index be77e3be7ce..631a61b3c9b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -85,6 +85,7 @@ pub struct AcpxSidecarTransport { process: SupervisedProcess, request_timeout: Duration, session_open_timeout: Duration, + request_deadline: Option, next_request_id: u64, last_event_sequence: u64, buffered_events: VecDeque, @@ -389,6 +390,7 @@ impl AcpxSidecarTransport { process, request_timeout: config.request_timeout, session_open_timeout: config.request_timeout, + request_deadline: None, next_request_id: 1, last_event_sequence: 0, buffered_events: VecDeque::new(), @@ -407,6 +409,10 @@ impl AcpxSidecarTransport { } } + pub(crate) fn bound_requests_by(&mut self, deadline: Option) { + self.request_deadline = deadline; + } + pub fn process_id(&self) -> u32 { self.process.id() } @@ -508,13 +514,13 @@ impl AcpxSidecarTransport { command: GeneratedAcpxSidecarCommand, params: Value, ) -> Result { + let timeout_at = Instant::now() + self.command_timeout(command); + let deadline = self + .request_deadline + .map_or(timeout_at, |bound| bound.min(timeout_at)); let request_id = self.send_request(command, params)?; - self.receive_response( - command, - request_id, - Instant::now() + self.command_timeout(command), - )? - .ok_or_else(|| self.request_timeout_error(command)) + self.receive_response(command, request_id, deadline)? + .ok_or_else(|| self.request_timeout_error(command)) } /// A park observation may expire without revoking a healthy provider. 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 49b2d151dc9..76a6a309f09 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 @@ -74,11 +74,27 @@ fn run() -> Result<(), Box> { .windows(2) .find(|pair| pair[0] == "--journal") .ok_or("source journal is missing")?; + let live_predecessors = std::fs::read_to_string(&path[1]) + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter_map(|line| line["processStart"].as_u64()) + .filter(|pid| { + std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .stderr(std::process::Stdio::null()) + .status() + .is_ok_and(|status| status.success()) + }) + .collect::>(); let mut file = std::fs::OpenOptions::new() .create(true) .append(true) .open(&path[1])?; - write_json(&mut file, &json!({"processStart": std::process::id()}))?; + write_json( + &mut file, + &json!({"processStart": std::process::id(), "livePredecessors": live_predecessors}), + )?; if args.iter().any(|arg| arg == "--spawn-owned-child") { let child = std::process::Command::new("/bin/sleep") .arg("120") @@ -224,7 +240,45 @@ fn run() -> Result<(), Box> { response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false, "error":{"code":"temporarily_unavailable","message":"Optional read is temporarily unavailable.","retryable":true}}); } + if command == "session.open" && fault == "slow-open" { + std::thread::sleep(std::time::Duration::from_secs(10)); + } + if command == "session.open" && fault == "reject-open" { + response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false, + "error":{"code":"session_open_failed","message":"The replacement provider could not open.","retryable":false}}); + } write_json(&mut stdout, &response)?; + if command == "session.open" && args.iter().any(|arg| arg == "--emit-process") { + write_json( + &mut stdout, + &json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":next_sequence,"eventType":"runtime.process","runId":null,"turnId":null, + "payload":{"role":"sidecar","pid":std::process::id(),"processGroupId":null,"startedAt":"2026-01-01T00:00:00.000Z"}}), + )?; + next_sequence += 1; + } + if command == "session.suspend" && fault == "late-tool-after-suspend" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + &source_run, + "late-turn", + json!({"callId":"late-call","operationId":"issues.delete","input":{"source":"late-old-generation"}}), + )?; + next_sequence += 1; + } + if command == "run.attach" && fault == "pre-turn-tool" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + &source_run, + "pre-turn", + json!({"callId":"pre-turn-call","operationId":"issues.read","input":{"id":"issue-1"}}), + )?; + next_sequence += 1; + } if command == "turn.start" { let turn = request["params"]["turnId"].as_str().unwrap(); if args.iter().any(|arg| arg == "--wait-completion-feedback") { 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 fb0040aec08..5a45df27369 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 @@ -47,6 +47,7 @@ enum CommandLifecycle { } const TERMINAL_RESULT_ACK_TIMEOUT: Duration = Duration::from_secs(2); +const HOST_PARK_PREPARATION_BUDGET: Duration = Duration::from_secs(20); const CUMULATIVE_ACK_PERSIST_INTERVAL: usize = 16; fn reconnect_delay(base: Duration, max_delay: Duration, attempt: &mut u32) -> Duration { @@ -318,6 +319,10 @@ pub trait CommandExecutor { Ok(false) } + fn prepare_for_park(&mut self, _deadline: Instant) -> Result { + Ok(false) + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -1451,6 +1456,20 @@ fn process_host_control( ) && executor.host_park_ready()? => { + let ready = + match executor.prepare_for_park(Instant::now() + HOST_PARK_PREPARATION_BUDGET) { + Ok(false) => true, + Ok(true) => executor.host_park_ready()?, + Err(error) => { + state.record_diagnostic(format!("host park preparation failed: {error}")); + store.save(state)?; + false + } + }; + if !ready { + control.channel.reply(&request, "conflict")?; + return Ok(HostAction::None); + } control.park_id = Some(id.clone()); control.channel.reply(&request, "parked")?; Ok(HostAction::Park) @@ -3216,6 +3235,136 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + #[cfg(unix)] + fn hosted_park_reproves_a_prepared_provider_before_replying() { + use std::io::{BufRead, BufReader, Write}; + use std::os::unix::net::UnixStream; + struct PreparingExecutor { + calls: Vec<&'static str>, + prepared: Option, + ready_after_prepare: bool, + prepared_once: bool, + } + impl CommandExecutor for PreparingExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("private control cannot execute PRP commands"); + } + fn host_park_ready(&mut self) -> Result { + self.calls.push("proof"); + Ok(!self.prepared_once || self.ready_after_prepare) + } + fn prepare_for_park(&mut self, _deadline: Instant) -> Result { + self.calls.push("prepare"); + self.prepared_once = true; + self.prepared + .ok_or_else(|| DurableRunnerError::invalid("replacement provider failed")) + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("shutdown"); + Ok(()) + } + } + for (prepared, ready_after_prepare, status, calls) in [ + ( + Some(true), + true, + "parked", + vec!["proof", "prepare", "proof"], + ), + ( + Some(true), + false, + "conflict", + vec!["proof", "prepare", "proof"], + ), + (None, true, "conflict", vec!["proof", "prepare"]), + (Some(false), false, "parked", vec!["proof", "prepare"]), + ] { + let directory = std::env::temp_dir() + .join(format!("paperclip-host-prepare-{}", 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 = PreparingExecutor { + calls: Vec::new(), + prepared, + ready_after_prepare, + prepared_once: false, + }; + 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, + }); + let mut exchange = |request: Value, + state: &mut DurableState, + control: &mut HostedRunnerControl, + executor: &mut PreparingExecutor, + 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()) + }; + let (action, reply) = exchange( + json!({"op":"park","id":"park-1","identity":identity, + "lastControllerCommandSeq":0,"ackedSourceSeq":0}), + &mut state, + &mut control, + &mut executor, + false, + ); + assert_eq!(reply["status"], status); + assert_eq!(executor.calls, calls); + assert_eq!(matches!(action, HostAction::Park), status == "parked"); + assert_eq!(control.park_id.is_some(), status == "parked"); + assert_eq!( + state + .diagnostics + .iter() + .any(|message| message.contains("host park preparation failed")), + prepared.is_none() + ); + if prepared == Some(true) && status == "parked" { + let (action, _) = exchange( + json!({"op":"claim","id":"claim-1","parkId":"park-1","bootstrapTicket":"fresh-private-ticket"}), + &mut state, + &mut control, + &mut executor, + true, + ); + assert!(matches!(action, HostAction::Claim(_))); + let (action, reply) = exchange( + json!({"op":"retire","id":"retire-claim","parkId":"park-1","claimId":"claim-1", + "identity":identity,"lastControllerCommandSeq":0,"ackedSourceSeq":0}), + &mut state, + &mut control, + &mut executor, + false, + ); + assert_eq!(reply["status"], "retired"); + assert!(matches!(action, HostAction::Retire)); + assert_eq!( + executor.calls, + ["proof", "prepare", "proof", "proof", "shutdown"] + ); + } + fs::remove_dir_all(directory).unwrap(); + } + } + #[test] #[cfg(unix)] fn hosted_park_blocks_without_provider_polls_and_retires_on_eof_or_lifetime() { 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 index 6f25a381c8c..5a55c968543 100644 --- 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 @@ -10,6 +10,7 @@ use std::os::unix::net::UnixStream; struct RetentionExecutor { polls: Arc, shutdowns: Arc, + prepares: Arc, } impl CommandExecutor for RetentionExecutor { fn execute(&mut self, _: &Command) -> Result { @@ -18,6 +19,10 @@ impl CommandExecutor for RetentionExecutor { fn host_park_ready(&mut self) -> Result { Ok(true) } + fn prepare_for_park(&mut self, _deadline: std::time::Instant) -> Result { + self.prepares.fetch_add(1, Ordering::SeqCst); + Ok(true) + } fn poll_events(&mut self) -> Result, DurableRunnerError> { self.polls.fetch_add(1, Ordering::SeqCst); Ok(vec![]) @@ -174,9 +179,11 @@ fn exercise_hosted_loop(blocked_work: Option) { }); let polls = Arc::new(AtomicUsize::new(0)); let shutdowns = Arc::new(AtomicUsize::new(0)); + let prepares = Arc::new(AtomicUsize::new(0)); let executor = RetentionExecutor { polls: polls.clone(), shutdowns: shutdowns.clone(), + prepares: prepares.clone(), }; let (socket, mut host) = UnixStream::pair().unwrap(); host.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); @@ -201,6 +208,7 @@ fn exercise_hosted_loop(blocked_work: Option) { "parked" ); let parked_polls = polls.load(Ordering::SeqCst); + assert_eq!(prepares.load(Ordering::SeqCst), 1); // Outlast the ordinary reconnect grace without any polling or shutdown. thread::sleep(Duration::from_millis(80)); assert_eq!(polls.load(Ordering::SeqCst), parked_polls); @@ -245,6 +253,7 @@ fn exercise_hosted_loop(blocked_work: Option) { finish_tx.send(()).unwrap(); server.join().unwrap(); assert_eq!(shutdowns.load(Ordering::SeqCst), 1); + assert_eq!(prepares.load(Ordering::SeqCst), 1); assert_eq!( DurableStateStore::new(&directory) .unwrap() 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 7db36c3ea39..f1f593634d9 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 @@ -337,6 +337,13 @@ impl CommandExecutor for NativeProviderCommandExecutor { } } + fn prepare_for_park(&mut self, deadline: Instant) -> Result { + match self.selected.as_mut() { + Some(SelectedExecutor::Acpx(executor)) => executor.prepare_for_park(deadline), + _ => Ok(false), + } + } + 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 new file mode 100644 index 00000000000..0f7fcf96fc7 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs @@ -0,0 +1,594 @@ +#![cfg(unix)] + +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use paperclip_runner_core::durable::{ + AcpxLaunchProfile, Command, CommandExecution, CommandExecutor, DurableRunnerConfig, + QualifiedLaunchArtifact, +}; +use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; +use paperclip_runner_core::provider_bridge::{authorized_tool_catalog_digest, AuthorizedTool}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +struct Fixture(PathBuf); +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +impl Fixture { + fn fault(&self, fault: &str) { + fs::write(self.0.join("fault"), fault).unwrap(); + } + + fn journal(&self) -> Vec { + fs::read_to_string(self.0.join("sidecar.jsonl")) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() + } + + fn process_starts(&self) -> Vec { + self.journal() + .into_iter() + .filter(|line| line.get("processStart").is_some()) + .collect() + } + + fn state(&self) -> Value { + serde_json::from_slice(&fs::read(self.0.join("acpx-provider-state.json")).unwrap()).unwrap() + } +} + +struct Harness { + fixture: Fixture, + config: DurableRunnerConfig, + executor: NativeProviderCommandExecutor, + descriptor: Value, + opened: Value, + sequence: u64, +} + +impl Harness { + fn new() -> Self { + let catalog: Value = + serde_json::from_str(include_str!("../../../../acpx-profiles.json")).unwrap(); + let profile = &catalog["profiles"]["claude"]; + let digest = profile["commandDigest"].as_str().unwrap(); + let fixture = + Fixture(std::env::temp_dir().join(format!("acpx-park-prep-{}", uuid::Uuid::new_v4()))); + fs::create_dir_all(&fixture.0).unwrap(); + fs::set_permissions(&fixture.0, fs::Permissions::from_mode(0o700)).unwrap(); + let native = fixture.0.join("native-sidecar"); + fs::copy(env!("CARGO_BIN_EXE_fake-acpx-sidecar"), &native).unwrap(); + fs::set_permissions(&native, fs::Permissions::from_mode(0o700)).unwrap(); + let sidecar = fixture.0.join("sidecar"); + fs::write( + &sidecar, + format!( + "#!/bin/sh\nexec '{}' --mode source-reconcile --emit-process --journal '{}' --source-fault '{}' \"$@\"\n", + native.display(), + fixture.0.join("sidecar.jsonl").display(), + fixture.0.join("fault").display() + ), + ) + .unwrap(); + fs::set_permissions(&sidecar, fs::Permissions::from_mode(0o700)).unwrap(); + let args = vec!["--profile-digest".to_owned(), digest.to_owned()]; + let config = DurableRunnerConfig { + connect_url: "ws://127.0.0.1/runner".to_owned(), + ca_bundle_path: None, + state_dir: fixture.0.clone(), + runner_instance_id: "runner-1".to_owned(), + environment_lease_id: "lease-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: format!("sha256:{}", "a".repeat(64)), + acpx_launch_profile: Some(AcpxLaunchProfile { + candidate_admission: None, + authority_digest: format!("sha256:{}", "d".repeat(64)), + command: sidecar.clone(), + args: args.clone(), + artifacts: vec![QualifiedLaunchArtifact { + path: sidecar.clone(), + sha256: format!("sha256:{:x}", Sha256::digest(fs::read(&sidecar).unwrap())), + }], + }), + opencode_launch_profile: None, + cursor_cloud_launch_profile: None, + max_outbox_bytes: 1024 * 1024, + p0_reserve_bytes: 64 * 1024, + max_frame_bytes: 1024 * 1024, + reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, + max_runtime: Duration::from_secs(30), + }; + let descriptor = json!({ + "kind": "acpx", "provider": "acpx", "driver": "acpx_runtime", + "providerVersion": "0.13.1", "agent": "claude", "model": "explicit-model", + "acpxVersion": "0.13.1", "agentServerPackage": profile["agentServerPackage"], + "agentServerVersion": profile["agentServerVersion"], "agentRuntimePackage": profile["agentRuntimePackage"], + "agentRuntimeVersion": profile["agentRuntimeVersion"], "commandDigest": digest, + "sidecarCommand": sidecar, "sidecarArgs": args, "runtimeDirectory": fixture.0, + "normalizedSessionId": "session-1", "runId": "run-1", "cwd": fixture.0, + "instructions": "Complete the supplied work.", + "permissionMode": "approve-all", "permissionModePinned": true, + "runtimeContext": { + "aggregateDigest": "a".repeat(64), + "skills": [{"key": "skill-1", "bundle": {"digest": "d".repeat(64), "rootPath": "/skills/one"}}], + "mcp": {"assignmentSetId": "assignment-1", "digest": "e".repeat(64), "bindingId": "binding-run-1"}, + }, + }); + let mut executor = NativeProviderCommandExecutor::with_runner_config(&fixture.0, &config); + executor + .execute(&command(1, "run.prepare", json!({"provider": descriptor}))) + .unwrap(); + let opened = executor + .execute(&command(2, "session.open", json!({}))) + .unwrap() + .result; + let mut harness = Self { + fixture, + config, + executor, + descriptor, + opened, + sequence: 2, + }; + harness.drain(); + harness + } + + fn execute(&mut self, kind: &str, payload: Value) -> CommandExecution { + self.sequence += 1; + self.executor + .execute(&command(self.sequence, kind, payload)) + .unwrap() + } + + fn drain(&mut self) -> Vec { + let mut seen = Vec::new(); + loop { + let events = self.executor.poll_events().unwrap(); + if events.is_empty() { + return seen; + } + seen.extend( + events + .iter() + .map(|event| json!({"type":event.event_type,"payload":event.payload})), + ); + self.executor.acknowledge_events(events.len()).unwrap(); + } + } + + fn complete_turn(&mut self, turn: &str) -> Vec { + let started = self.execute( + "turn.start", + json!({"turnId": turn, "text": "Complete the work."}), + ); + let deadline = Instant::now() + Duration::from_secs(5); + let mut seen = started + .events + .iter() + .map(|(kind, _, payload)| json!({"type":kind,"payload":payload})) + .collect::>(); + while !seen + .iter() + .any(|event: &Value| event["type"] == "turn.completed") + { + assert!(Instant::now() < deadline, "no completed provider turn"); + seen.extend(self.drain()); + } + seen.extend(self.drain()); + seen + } + + fn attach_payload(&self, run: &str, mutate: impl FnOnce(&mut Value)) -> Value { + let mut descriptor = self.descriptor.clone(); + descriptor["runId"] = json!(run); + descriptor["runtimeContext"]["mcp"]["bindingId"] = json!(format!("binding-{run}")); + descriptor["runtimeContext"]["aggregateDigest"] = json!(format!("{run}-aggregate")); + let mut payload = json!({ + "provider": descriptor, + "paperclipNextAuthority": {"identity": { + "runId": run, "normalizedSessionId": self.config.normalized_session_id, + }}, + }); + mutate(&mut payload); + payload + } + + fn attach( + &mut self, + run: &str, + mutate: impl FnOnce(&mut Value), + ) -> (CommandExecution, Duration) { + let payload = self.attach_payload(run, mutate); + let started = Instant::now(); + let attached = self.execute("run.attach", payload); + let elapsed = started.elapsed(); + self.config.run_id = run.to_owned(); + self.executor.rotate_authority(&self.config); + (attached, elapsed) + } + + fn quiesce(&mut self) { + let ready = self.execute("session.snapshot", json!({"quiesceForWarmAttach": true})); + assert_eq!(self.drain(), Vec::::new()); + assert_eq!(ready.result["warmAttachReady"], true); + } +} + +fn command(sequence: u64, kind: &str, payload: Value) -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: format!("command-{sequence}"), + controller_seq: sequence, + command_type: kind.to_owned(), + issued_at: "2026-09-21T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload, + } +} + +fn alive(pid: &Value) -> bool { + std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .stderr(std::process::Stdio::null()) + .status() + .unwrap() + .success() +} + +fn commands_after(journal: &[Value], start: usize) -> Vec { + journal[start..] + .iter() + .filter_map(|line| line["command"].as_str().map(str::to_owned)) + .collect() +} + +fn last_process_start(journal: &[Value]) -> usize { + journal + .iter() + .rposition(|line| line.get("processStart").is_some()) + .unwrap() +} + +fn prepared_park(harness: &mut Harness) -> Value { + harness.complete_turn("first"); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.process_starts().last().unwrap()["processStart"].clone() +} + +#[test] +fn park_reaps_the_old_generation_before_spawning_and_proves_the_new_one_dormant() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let before = harness.fixture.journal().len(); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert!(spawned > before); + assert_eq!( + commands_after(&journal[..spawned], before), + ["session.goal.get", "session.snapshot", "session.suspend"] + ); + assert_eq!(journal[spawned]["livePredecessors"], json!([])); + assert_ne!(journal[spawned]["processStart"], old_pid); + assert_eq!( + commands_after(&journal, spawned), + [ + "initialize", + "session.open", + "run.attach", + "session.goal.get", + "session.snapshot" + ] + ); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-1"); + assert!(!alive(&old_pid)); + assert!(alive(&journal[spawned]["processStart"])); + assert!( + harness.executor.host_park_ready().unwrap(), + "a prepared dormant generation is park-ready" + ); + assert_eq!(harness.drain(), Vec::::new()); + assert_eq!(harness.fixture.state()["lifecycle"], "session_open"); + assert!( + !harness.executor.prepare_for_park(park_deadline()).unwrap(), + "a prepared generation is not prepared twice" + ); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); + assert!(!alive(&journal[spawned]["processStart"])); +} + +#[test] +fn compatible_attach_keeps_the_prepared_provider_and_rebinds_its_run() { + let mut cold = Harness::new(); + cold.complete_turn("first"); + cold.quiesce(); + let (_, cold_attach) = cold.attach("run-2", |_| {}); + cold.executor.shutdown().unwrap(); + + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + let before = harness.fixture.journal().len(); + let (attached, warm_attach) = harness.attach("run-2", |_| {}); + eprintln!("run.attach cold={cold_attach:?} prepared={warm_attach:?}"); + assert_eq!(attached.result["status"], "resumed"); + assert_eq!(attached.result["processId"], prepared_pid); + assert_eq!( + attached.result["providerSessionId"], + harness.opened["providerSessionId"] + ); + let journal = harness.fixture.journal(); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!( + commands_after(&journal, before), + [ + "run.attach", + "session.goal.get", + "session.snapshot", + "session.goal.get" + ] + ); + assert_eq!(journal[before]["params"]["runId"], "run-2"); + let state = harness.fixture.state(); + assert_eq!(state["descriptor"]["runId"], "run-2"); + assert_eq!( + state["descriptor"]["runtimeContext"]["mcp"]["bindingId"], + "binding-run-2" + ); + let events = harness.complete_turn("second"); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == prepared_pid)); + assert_eq!( + harness.fixture.process_starts().len(), + 2, + "the first turn of the rebound run reuses the prepared provider" + ); + let snapshot = harness.execute("session.snapshot", json!({})); + assert_eq!(snapshot.result["warmAttachReady"], true); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn changed_tools_or_skills_take_the_cold_path() { + let operations = vec![AuthorizedTool { + operation_id: "issues.read".to_owned(), + version: 1, + description: "Read an issue.".to_owned(), + input_schema: json!({"type":"object"}), + response_schema: json!({"type":"object"}), + }]; + let tools = json!({ + "schema": "paperclip.runner.authorized-tools.v1", "schemaVersion": 1, + "catalogDigest": authorized_tool_catalog_digest(&operations).unwrap(), + "operations": operations, + }); + let changes: [(&str, Box); 2] = [ + ( + "tools", + Box::new(move |payload| payload["authorizedTools"] = tools.clone()), + ), + ( + "skills", + Box::new(|payload| { + payload["provider"]["runtimeContext"]["skills"][0]["bundle"]["rootPath"] = + json!("/skills/two") + }), + ), + ]; + for (name, change) in changes { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + let (attached, _) = harness.attach("run-2", |payload| change(payload)); + assert_eq!(attached.result["status"], "resumed", "{name}"); + assert_ne!(attached.result["processId"], prepared_pid, "{name}"); + assert!(!alive(&prepared_pid), "{name}"); + assert_eq!(harness.fixture.process_starts().len(), 3, "{name}"); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert_eq!(journal[spawned]["livePredecessors"], json!([]), "{name}"); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-2", "{name}"); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); + } +} + +#[test] +fn a_generation_that_ran_a_turn_is_never_rebound() { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + harness.attach("run-2", |_| {}); + harness.complete_turn("second"); + harness.quiesce(); + let (attached, _) = harness.attach("run-3", |_| {}); + assert_ne!(attached.result["processId"], prepared_pid); + assert!(!alive(&prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn late_and_pre_turn_callbacks_never_reach_the_next_run() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("late-tool-after-suspend"); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.fault(""); + let (attached, _) = harness.attach("run-2", |_| {}); + let events = harness.complete_turn("second"); + assert!(!format!("{events:?}").contains("late-call")); + assert!(!format!("{events:?}").contains("late-old-generation")); + let reused_pid = attached.result["processId"].clone(); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); + assert!(!alive(&reused_pid)); + + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("pre-turn-tool"); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.fault(""); + let prepared_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!( + !harness.executor.host_park_ready().unwrap(), + "a turn-bound callback before any turn blocks the park" + ); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_ne!(attached.result["processId"], prepared_pid); + assert!(!alive(&prepared_pid)); + let events = harness.complete_turn("second"); + assert!(!format!("{events:?}").contains("pre-turn-call")); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn failed_preparation_conflicts_and_leaves_a_cold_recoverable_session() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + let error = harness + .executor + .prepare_for_park(park_deadline()) + .unwrap_err(); + assert!( + error.to_string().contains("park preparation failed"), + "{error}" + ); + harness.fixture.fault(""); + assert!(!alive(&old_pid)); + let failed_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(!alive(&failed_pid)); + let state = harness.fixture.state(); + assert_eq!(state["lifecycle"], "suspended"); + assert_eq!(state["providerExitUnconfirmed"], false); + assert!(state["identity"].is_object()); + assert!(!harness.executor.host_park_ready().unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert_eq!(harness.drain(), Vec::::new()); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["status"], "resumed"); + assert_eq!( + attached.result["providerSessionId"], + harness.opened["providerSessionId"] + ); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_slow_replacement_open_fails_at_the_park_deadline_and_leaves_the_runner_responsive() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("slow-open"); + let started = Instant::now(); + let error = harness + .executor + .prepare_for_park(Instant::now() + Duration::from_secs(1)) + .unwrap_err(); + let elapsed = started.elapsed(); + harness.fixture.fault(""); + assert!( + error.to_string().contains("timed out at session.open"), + "{error}" + ); + assert!( + elapsed < Duration::from_secs(5), + "park preparation took {elapsed:?}" + ); + assert!(!alive(&old_pid)); + let slow_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(!alive(&slow_pid)); + let state = harness.fixture.state(); + assert_eq!(state["lifecycle"], "suspended"); + assert_eq!(state["providerExitUnconfirmed"], false); + assert!(!harness.executor.host_park_ready().unwrap()); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["status"], "resumed"); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn failed_preparation_recovers_the_same_run_on_its_next_turn() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + assert!(harness.executor.prepare_for_park(park_deadline()).is_err()); + harness.fixture.fault(""); + 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.shutdown().unwrap(); +} + +#[test] +fn prepared_park_retires_through_ordinary_shutdown() { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.shutdown().unwrap(); + assert!(!alive(&prepared_pid)); + let journal = harness.fixture.journal(); + assert_eq!( + journal.last().unwrap()["command"], + "session.close", + "retirement closes the prepared provider" + ); + assert_eq!(harness.fixture.state()["descriptor"]["runId"], "run-1"); +} + +#[test] +fn a_conflicted_park_keeps_the_prepared_provider_for_the_same_run() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + let prepared_pid = prepared_park(&mut harness); + let events = harness.complete_turn("second"); + let reconciled = events + .iter() + .find(|event| event["type"] == "session.reconciled") + .expect("the turn reports the replaced provider process"); + assert_eq!(reconciled["payload"]["previousProcessId"], old_pid); + assert_eq!(reconciled["payload"]["processId"], prepared_pid); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 2); + let events = harness.complete_turn("third"); + assert!(events + .iter() + .any(|event| event["type"] == "session.reconciled" + && event["payload"]["previousProcessId"] == prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +fn park_deadline() -> Instant { + Instant::now() + Duration::from_secs(20) +} 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 9303af23960..4b56a5abc4c 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 @@ -266,6 +266,13 @@ fn hosted_park_acpx_requires_completion_in_this_lifetime_and_exact_authority() { .unwrap(); settle(&mut executor); assert!(executor.host_park_ready().unwrap()); + assert!(executor + .prepare_for_park(Instant::now() + Duration::from_secs(20)) + .unwrap()); + assert!( + executor.host_park_ready().unwrap(), + "a prepared dormant generation inherits its park proof" + ); config.run_id = "other-run".to_owned(); executor.rotate_authority(&config); assert!(!executor.host_park_ready().unwrap()); @@ -467,5 +474,8 @@ fn hosted_park_acpx_does_not_enable_an_unqualified_park_provider() { .unwrap(); settle(&mut executor); assert!(!executor.host_park_ready().unwrap()); + assert!(!executor + .prepare_for_park(Instant::now() + Duration::from_secs(20)) + .unwrap()); 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 e4485a9f7bb..1294ea1adf5 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 @@ -1,5 +1,5 @@ use std::path::PathBuf; -use std::time::Duration; +use std::time::{Duration, Instant}; use paperclip_runner_core::acpx_provider_session::{ AcpxPermissionMode, AcpxProviderSession, AcpxProviderSessionConfig, @@ -791,3 +791,146 @@ fn forced_cancellation_retirement_preserves_terminal_polling_but_rejects_live_re session.shutdown("test complete").unwrap(); } } + +fn reusable_config(directory: &std::path::Path) -> AcpxProviderSessionConfig { + let mut config = config("source-reconcile"); + config.transport.args = vec![ + "--mode".to_owned(), + "source-reconcile".to_owned(), + "--journal".to_owned(), + directory.join("journal.jsonl").display().to_string(), + "--source-fault".to_owned(), + directory.join("fault").display().to_string(), + ]; + config.runtime_context = json!({ + "aggregateDigest": "run-1-aggregate", + "skills": [{"key": "skill-1", "bundle": {"rootPath": "/skills/one"}}], + "mcp": {"bindingId": "binding-run-1"}, + }); + config +} + +fn settle_turn(session: &mut AcpxProviderSession, turn_id: &str) -> Vec { + session + .start_turn(turn_id, "Please continue", &std::env::temp_dir()) + .unwrap(); + let mut seen = Vec::new(); + loop { + let events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + let settled = events.iter().any(|event| { + matches!(event, AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } if settled == turn_id) + }); + seen.extend(events); + if settled { + return seen; + } + } +} + +fn process_starts(directory: &std::path::Path) -> usize { + std::fs::read_to_string(directory.join("journal.jsonl")) + .unwrap() + .lines() + .filter(|line| line.contains("processStart")) + .count() +} + +#[test] +fn prepared_generation_rebinds_only_an_identical_unused_configuration() { + let directory = std::env::temp_dir().join(format!("acpx-generation-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&directory).unwrap(); + let config = reusable_config(&directory); + let mut session = AcpxProviderSession::start(&config).unwrap(); + assert!(!session.can_prepare_next_generation()); + assert!(session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .is_err()); + settle_turn(&mut session, "turn-1"); + let first_pid = session.process_id(); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + let prepared_pid = session.process_id(); + assert_ne!(prepared_pid, first_pid); + assert!(session.awaits_run_rebind() && !session.generation_has_run_turn()); + let mut next = config.clone(); + next.run_id = "run-2".to_owned(); + next.expected_identity = Some(session.identity().clone()); + next.runtime_context["mcp"]["bindingId"] = json!("binding-run-2"); + next.runtime_context["aggregateDigest"] = json!("run-2-aggregate"); + let changes: [fn(&mut AcpxProviderSessionConfig); 7] = [ + |config| config.system_instructions.push_str(" Changed."), + |config| config.tool_set.operations.clear(), + |config| config.runtime_context["skills"][0]["bundle"]["rootPath"] = json!("/skills/two"), + |config| config.runtime_context["mcp"]["bindingId"] = serde_json::Value::Null, + |config| config.permission_mode = AcpxPermissionMode::DenyAll, + |config| config.run_id = "run-1".to_owned(), + |config| config.expected_identity = None, + ]; + for change in changes { + let mut changed = next.clone(); + change(&mut changed); + assert!(!session.rebind_run(&changed).unwrap()); + assert_eq!(session.process_id(), prepared_pid); + assert!(session.awaits_run_rebind()); + } + assert!(session.rebind_run(&next).unwrap()); + assert_eq!(session.process_id(), prepared_pid); + assert_eq!(session.state().run_id(), "run-2"); + assert!(!session.awaits_run_rebind()); + let attached_runs = std::fs::read_to_string(directory.join("journal.jsonl")) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|line| line["command"] == "run.attach") + .map(|line| line["params"]["runId"].clone()) + .collect::>(); + assert_eq!( + attached_runs, + [json!("run-1"), json!("run-1"), json!("run-2")] + ); + settle_turn(&mut session, "turn-1"); + assert_eq!(session.process_id(), prepared_pid); + assert!(session.generation_has_run_turn()); + let mut later = next.clone(); + later.run_id = "run-3".to_owned(); + assert!( + !session.rebind_run(&later).unwrap(), + "a used generation is never rebound" + ); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + settle_turn(&mut session, "turn-2"); + assert!( + !session.rebind_run(&later).unwrap(), + "a prepared generation that ran a turn is never rebound" + ); + assert_eq!(process_starts(&directory), 3); + session.shutdown("test complete").unwrap(); + std::fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn prepared_generation_never_admits_the_reaped_generation_late_callback() { + let directory = std::env::temp_dir().join(format!("acpx-generation-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&directory).unwrap(); + let mut session = AcpxProviderSession::start(&reusable_config(&directory)).unwrap(); + settle_turn(&mut session, "turn-1"); + std::fs::write(directory.join("fault"), "late-tool-after-suspend").unwrap(); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + std::fs::write(directory.join("fault"), "").unwrap(); + let events = settle_turn(&mut session, "turn-2"); + assert!(!events + .iter() + .any(|event| matches!(event, AcpxProviderStateEvent::ToolCall { .. }))); + assert_eq!( + process_starts(&directory), + 2, + "the prepared turn does not restart again" + ); + session.shutdown("test complete").unwrap(); + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts index db33af5ffd6..784b98e8c26 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts @@ -1404,6 +1404,50 @@ describe("qualified ACPX runtime sidecar", () => { expect(attachedRunId).toBe("run-1"); }); + it("rebinds a prepared sidecar to another run only before its first turn", async () => { + const root = mkdtempSync(join(tmpdir(), "acpx-run-rebind-")); + const loader = join(root, "provider-loader.mjs"), host = join(root, "rebind-host.mjs"); + const baseHost = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href; + writeFileSync(host, `import { AcpxRuntimeHost as BaseHost } from ${JSON.stringify(baseHost)}; + export class AcpxRuntimeHost extends BaseHost { + static async open(options) { const host = await BaseHost.open(options); Object.setPrototypeOf(host, this.prototype); return host; } + startTurn({ requestId }) { return { requestId, promptStarted: Promise.resolve(), + result: Promise.resolve({ status: "completed" }), events: (async function* () {})() }; } + }`); + writeFileSync(loader, `import { registerHooks } from "node:module"; + import { pathToFileURL } from "node:url"; + registerHooks({ resolve(specifier, context, next) { + if (context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js")) + return { url: pathToFileURL(${JSON.stringify(host)}).href, shortCircuit: true }; + return next(specifier, context); + } });`); + const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}}); + children.add(sidecar); + const send = (id:number, command:string, params:Record) => sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params}); + try { + sidecar.write(initializeRequest(1,"codex")); + expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true}); + send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root, + normalizedSessionId:"rebind-session",permissionMode:"deny-all",systemInstructions:"Complete the local fixture turn.",tools:[]}); + expect(await sidecar.next(frame=>frame.id === 2)).toMatchObject({ok:true}); + send(3,"run.attach",{runId:"parked-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 3)).toMatchObject({ok:true,result:{runId:"parked-run"}}); + send(4,"run.attach",{runId:"next-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 4)).toMatchObject({ok:true,result:{runId:"next-run"}}); + send(5,"session.snapshot",{}); + expect(await sidecar.next(frame=>frame.id === 5)).toMatchObject({ok:true,result:{runId:"next-run",turnId:null}}); + send(6,"turn.start",{turnId:"next-turn",message:"Complete the fixture turn."}); + expect(await sidecar.next(frame=>frame.id === 6)).toMatchObject({ok:true}); + expect(await sidecar.next(frame=>frame.eventType === "runtime.turn_terminal")) + .toMatchObject({runId:"next-run",turnId:"next-turn",payload:{status:"completed"}}); + send(7,"run.attach",{runId:"later-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 7)).toMatchObject({ok:false, + error:{message:"cannot attach another run after this sidecar started a turn"}}); + send(8,"session.snapshot",{}); + expect(await sidecar.next(frame=>frame.id === 8)).toMatchObject({ok:true,result:{runId:"next-run",turnId:null}}); + } finally {await sidecar.close();children.delete(sidecar);rmSync(root,{recursive:true,force:true});} + }, 15_000); + it("recovers after malformed input and reports its qualified Codex profile", async () => { const sidecar = startSidecar(); sidecar.write({ diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index a95d6a4f7a6..941ae867f17 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -176,6 +176,7 @@ let openParams: AcpxSidecarOpenParams | null = null; let claudeCompactionCount = 0; let runId: string | null = null; let turnId: string | null = null; +let turnStartedInProcess = false; /** * A restored session runs in a new provider process whose MCP request ids * restart. The controller keys semantic calls by run and call id, so prefix @@ -402,6 +403,7 @@ async function dispatch( if (request.command === "run.attach") { requireHost(); if (turnId) throw new Error("cannot attach a run during an active turn"); + if (turnStartedInProcess) throw new Error("cannot attach another run after this sidecar started a turn"); const attachedTools = parseTools(request.params.tools); if ( canonicalJson(attachedTools) !== canonicalJson(openParams?.tools ?? []) @@ -424,6 +426,7 @@ async function dispatch( const currentTurnId = boundedIdentity(request.params.turnId, "turnId"); const activeAgent = openParams!.agent; turnId = currentTurnId; + turnStartedInProcess = true; turnControls.begin(currentTurnId); let runtimeTurn: AcpxRuntimeTurn; const extensions = bindAcpxExtensionTurn({ diff --git a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts index c10017c71e9..837787940cc 100644 --- a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts @@ -296,7 +296,9 @@ describeEmbeddedPostgres("plugin Runner eviction through the Core execution entr expect((await state()).lifecycle).toBe("suspended"); } else { const providerPids: number[] = harness === "codex" ? [Number(await readFile(providerPidFile, "utf8"))] - : (await readFile(calls, "utf8")).trim().split(/\r?\n/).flatMap(line => { const entry = JSON.parse(line); return [entry.processStart, entry.childPid].filter(Number.isInteger); }); + : (await readFile(calls, "utf8")).trim().split(/\r?\n/).flatMap(line => { const entry = JSON.parse(line); return [entry.processStart, entry.childPid].filter(Number.isInteger); }) + .filter(pid => { try { process.kill(pid, 0); return true; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return false; throw error; } }); + expect(providerPids.length).toBeGreaterThan(0); const nativePid = handles.get(taskA)!.child.pid!; for (const pid of providerPids) { const [parent, group] = execFileSync("ps", ["-p", String(pid), "-o", "ppid=", "-o", "pgid="], { encoding: "utf8" }).trim().split(/\s+/).map(Number); diff --git a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts index 4da3feb7710..fed25e693ab 100644 --- a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts @@ -298,7 +298,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio expect(methods.filter(method => method === "turn/start")).toHaveLength(adopted ? 3 : 2); } else { const journal = methods.map(line => JSON.parse(line)); - expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 3 : 2); + expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 4 : 3); expect(journal.filter(entry => entry.command === "turn.start")).toHaveLength(adopted ? 3 : 2); } } catch (error) {