From 205ffcb6069798d7e4c4f26789e1edca5eb8c5c0 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 5 Nov 2026 21:37:34 -0800 Subject: [PATCH] Verify live Claude and Codex ACPX eligibility before parking (#19295) --- .../runner/DURABLE_TRANSPORT.md | 10 +- .../runner-core/src/acpx_provider_backend.rs | 58 ++++++ .../runner-core/src/acpx_provider_session.rs | 89 +++++++-- .../runner-core/src/acpx_sidecar_transport.rs | 149 +++++++++++++-- .../runner-core/src/bin/fake-acpx-sidecar.rs | 19 ++ .../src/native_provider_backend.rs | 1 + .../src/tests/acpx_park_observation.rs | 94 +++++++++ .../runner-core/tests/acpx_park_readiness.rs | 178 ++++++++++++++++++ 8 files changed, 563 insertions(+), 35 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/tests/acpx_park_observation.rs diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index abd99258eb3..ca4697d8c46 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -157,12 +157,20 @@ for the existing process that completed the current turn. An eligibility check never selects or restores a provider. A restored process must complete its own authoritative turn before it can be eligible. Reconstructing an accepted finish during recovery preserves result delivery but does not qualify that process. -ACPX remains ineligible in this slice, including after qualified recovery. Active turns, goals, startup attempts, tool calls, pending credentials, and unsettled events block parking. Known helper threads also block it: their diagnostic lineage does not prove that their work finished. The bounded completed-turn tail drain retains late accounting and goal changes for durable state and normal event delivery. + +Claude and Codex through ACPX also need a successful turn in the live sidecar. +Runner checks its qualified profile, session identity and source run A. It gives +that sidecar 100 ms to report its goal and session state. If the sidecar is slow +or declines, Runner stays attached. Runner keeps at most one unanswered read and +recognizes its late reply. Events received during that read still belong to A +and are delivered before B attaches. The existing policy still replaces the +sidecar at the next turn or run attachment; the saved provider identity continues. + The check also inspects the live provider 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, 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 3298cd5ceff..65647805628 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 @@ -2307,6 +2307,11 @@ impl AcpxCommandExecutor { } fn retain_buffered_provider_events(&mut self) -> Result<(), DurableRunnerError> { + if let Some(session) = self.session.as_mut() { + session.settle_park_observation().map_err(|error| { + DurableRunnerError::invalid(format!("ACPX observation settlement failed: {error}")) + })?; + } loop { let count = self .session @@ -2516,6 +2521,59 @@ fn goal_terminal_disposition<'a>( } impl CommandExecutor for AcpxCommandExecutor { + fn host_park_ready(&mut self) -> Result { + self.ensure_persistence_healthy()?; + let Some(state) = self.state.as_ref() else { + return Ok(false); + }; + if !matches!(state.descriptor.agent.as_str(), "codex" | "claude") + || self.session.is_none() + || state.lifecycle != "session_open" + || state.provider_exit_unconfirmed + || state.descriptor.run_id != self.context.run_id + || state.identity.is_none() + || state.active_turn_id.is_some() + || !state.pending_events.is_empty() + || !state.pending_runtime_requests.is_empty() + || (!state.goal_projection.is_null() && !goal_is_dormant(&state.goal_projection)) + || !state.descriptor.names_qualified_release() + || self.candidate_admission().is_some() + || !acpx_release_profile(&state.descriptor.agent) + .is_some_and(|profile| !profile.qualification_pending) + { + return Ok(false); + } + state.validate(&self.context, &self.launch_profile_digest()?)?; + let session = self + .session + .as_mut() + .expect("checked surviving ACPX session"); + if state.identity.as_ref() != Some(session.identity()) { + return Ok(false); + } + let event_sequence = session.observed_event_sequence(); + let ready = session + .host_park_ready(std::time::Instant::now() + Duration::from_millis(100)) + .map_err(|error| { + DurableRunnerError::invalid(format!("ACPX parking observation failed: {error}")) + })?; + // Commit new provider observations through the ordinary event path. + // A retained prefix must be delivered and ACKed before a later park. + self.poll_provider(false)?; + Ok(ready + && self + .session + .as_ref() + .is_some_and(|session| session.observed_event_sequence() == event_sequence) + && self.state.as_ref().is_some_and(|state| { + state.pending_events.is_empty() + && state.lifecycle == "session_open" + && state.active_turn_id.is_none() + && !state.provider_exit_unconfirmed + && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection)) + })) + } + fn execute(&mut self, command: &Command) -> Result { self.restore()?; if self 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 53c1bf8ea3f..7c3c07a0f02 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 @@ -1,6 +1,6 @@ use crate::generated_acpx_profiles::acpx_release_profile; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; @@ -351,6 +351,7 @@ pub struct AcpxProviderSession { failure: Option, transport_terminated: bool, runtime_retired: bool, + completed_turn_observed: bool, } pub(crate) fn goal_is_dormant(projection: &Value) -> bool { @@ -407,6 +408,7 @@ impl AcpxProviderSession { failure: None, transport_terminated: false, runtime_retired: false, + completed_turn_observed: false, }) } @@ -446,6 +448,25 @@ impl AcpxProviderSession { let snapshot = self .transport .request(GeneratedAcpxSidecarCommand::SessionSnapshot, json!({}))?; + self.validate_dormant_snapshot(&snapshot)?; + // The status barrier can queue a newer goal while the snapshot is in flight. + // Inspect without consuming facts that ordinary polling still owns. + let mut checked_state = self.state.clone(); + for event in self.transport.buffered_events().skip(prior_events) { + if event.event_type != GeneratedAcpxSidecarEventType::RuntimeGoal { + continue; + } + checked_state.accept_event(event)?; + if !goal_is_dormant(&event.payload) { + return Err(LocalRunnerError::invalid( + "saved dormant ACPX received an active or unknown goal during its snapshot", + )); + } + } + Ok(()) + } + + fn validate_dormant_snapshot(&self, snapshot: &Value) -> Result<(), LocalRunnerError> { let identity: AcpxProviderSessionIdentity = serde_json::from_value(snapshot["identity"].clone()).map_err(|_| { LocalRunnerError::invalid("saved dormant ACPX snapshot identity is invalid") @@ -463,23 +484,49 @@ impl AcpxProviderSession { "saved dormant ACPX snapshot has unexpected work or identity", )); } - // The status barrier can queue a newer goal while the snapshot is in flight. - // Inspect without consuming facts that ordinary polling still owns. - let mut checked_state = self.state.clone(); - for event in self.transport.buffered_events().skip(prior_events) { - if event.event_type != GeneratedAcpxSidecarEventType::RuntimeGoal { - continue; - } - checked_state.accept_event(event)?; - if !goal_is_dormant(&event.payload) { - return Err(LocalRunnerError::invalid( - "saved dormant ACPX received an active or unknown goal during its snapshot", - )); - } - } Ok(()) } + pub(crate) fn observed_event_sequence(&self) -> u64 { + self.transport.observed_event_sequence() + } + + pub(crate) fn settle_park_observation(&mut self) -> Result<(), LocalRunnerError> { + self.transport.settle_observation() + } + + pub(crate) fn host_park_ready(&mut self, deadline: Instant) -> Result { + if self.closed + || self.failure.is_some() + || self.transport_terminated + || self.runtime_retired + || !self.completed_turn_observed + || self.state.active_turn_id().is_some() + || self.state.has_pending_requests() + || self.transport.buffered_events().len() != 0 + { + return Ok(false); + } + let Some(goal) = self + .transport + .observe(GeneratedAcpxSidecarCommand::SessionGoalGet, deadline)? + else { + return Ok(false); + }; + if !goal_is_dormant(&goal) { + return Ok(false); + } + let Some(snapshot) = self + .transport + .observe(GeneratedAcpxSidecarCommand::SessionSnapshot, deadline)? + else { + 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) + } + /// Read the same live sidecar before exposing its in-memory pending ledger. /// A durable file alone cannot authorize a request after process loss. pub fn verify_live_request_snapshot( @@ -661,6 +708,7 @@ impl AcpxProviderSession { } self.tool_bridge = next_tool_bridge; self.reserved_tool_bridge = next_reserved_tool_bridge; + self.completed_turn_observed = false; Ok(response) } @@ -956,6 +1004,16 @@ impl AcpxProviderSession { reconciled_events.push(event); } } + for event in &reconciled_events { + if let AcpxProviderStateEvent::TurnTerminal { + turn_id, status, .. + } = event + { + self.completed_turn_observed = *status + == crate::acpx_event_payload::AcpxTurnStatus::Completed + && self.state.active_turn_id() == Some(turn_id.as_str()); + } + } self.state = next_state; self.tool_bridge = next_bridge; self.reserved_tool_bridge = next_reserved_bridge; @@ -2170,6 +2228,7 @@ mod mode_tests { failure: None, transport_terminated: false, runtime_retired: false, + completed_turn_observed: false, } } 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 f402a48248d..07282d24adb 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 @@ -86,6 +86,7 @@ pub struct AcpxSidecarTransport { stderr_tail: BoundedLogBuffer, stderr_categories: BTreeSet<&'static str>, poisoned: bool, + abandoned_observation: Option, } // A fresh Pi process verifies and copies its native closure before ACP admission. @@ -387,6 +388,7 @@ impl AcpxSidecarTransport { stderr_tail: BoundedLogBuffer::new(32, 8 * 1024), stderr_categories: BTreeSet::new(), poisoned: false, + abandoned_observation: None, }) } @@ -469,6 +471,86 @@ impl AcpxSidecarTransport { command: GeneratedAcpxSidecarCommand, params: Value, ) -> Result { + 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)) + } + + /// A park observation may expire without revoking a healthy provider. + /// Retain one ID so its late reply cannot masquerade as an ordinary reply. + pub(crate) fn observe( + &mut self, + command: GeneratedAcpxSidecarCommand, + deadline: Instant, + ) -> Result, LocalRunnerError> { + if self.poisoned { + return Err(LocalRunnerError::invalid( + "ACPX sidecar transport is unavailable", + )); + } + if !matches!( + command, + GeneratedAcpxSidecarCommand::SessionGoalGet + | GeneratedAcpxSidecarCommand::SessionSnapshot + ) { + return Err(LocalRunnerError::invalid( + "ACPX observation must be read-only", + )); + } + if self.abandoned_observation.is_some() || Instant::now() >= deadline { + return Ok(None); + } + let result = self + .send_request(command, json!({})) + .and_then(|request_id| { + let result = self.receive_response(command, request_id, deadline)?; + if result.is_none() { + self.abandoned_observation = Some(request_id); + } + Ok(result) + }); + match result { + Ok(Some(CommandOutcome::Success(value))) => Ok(Some(value)), + Ok(Some(CommandOutcome::Rejected(_))) => Ok(None), + Ok(None) => Ok(None), + Err(error) => { + self.poison(); + Err(error) + } + } + } + + pub(crate) fn observed_event_sequence(&self) -> u64 { + self.last_event_sequence + } + + /// Ordinary turn admission drains an expired read before replacing its sidecar. + pub(crate) fn settle_observation(&mut self) -> Result<(), LocalRunnerError> { + let Some(request_id) = self.abandoned_observation.take() else { + return Ok(()); + }; + let command = GeneratedAcpxSidecarCommand::SessionSnapshot; + match self.receive_response(command, request_id, Instant::now() + self.request_timeout) { + Ok(Some(_)) => Ok(()), + result => { + self.poison(); + Err(match result { + Err(error) => error, + _ => self.request_timeout_error(command), + }) + } + } + } + + fn send_request( + &mut self, + command: GeneratedAcpxSidecarCommand, + params: Value, + ) -> Result { if !params.is_object() { return Err(LocalRunnerError::invalid( "ACPX sidecar command params must be an object", @@ -502,19 +584,30 @@ impl AcpxSidecarTransport { })?; self.next_request_id = request_id + 1; - let timeout = self.command_timeout(command); - let deadline = Instant::now() + timeout; + Ok(request_id) + } + + fn receive_response( + &mut self, + command: GeneratedAcpxSidecarCommand, + request_id: u64, + deadline: Instant, + ) -> Result, LocalRunnerError> { loop { let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { - return Err(self.request_timeout_error(command)); + return Ok(None); } let Some(line) = self.receive_stdout_line(remaining, command.as_str())? else { - return Err(self.request_timeout_error(command)); + return Ok(None); }; match parse_frame(&line)? { ParsedFrame::Event(event) => self.buffer_event(event)?, ParsedFrame::Response(response) => { + if self.abandoned_observation == Some(response.id) { + self.abandoned_observation = None; + continue; + } if response.id != request_id { return Err(LocalRunnerError::invalid(format!( "ACPX sidecar response id mismatch: expected {request_id}, received {}", @@ -522,20 +615,20 @@ impl AcpxSidecarTransport { ))); } if response.ok { - return Ok(CommandOutcome::Success( + return Ok(Some(CommandOutcome::Success( response.result.unwrap_or_else(|| json!({})), - )); + ))); } let error = response.error.expect("failed response has validated error"); let provider_failure = error.provider_failure.clone(); - return Ok(CommandOutcome::Rejected(LocalRunnerError::invalid( + return Ok(Some(CommandOutcome::Rejected(LocalRunnerError::invalid( format!( "ACPX sidecar command {} was rejected (retryable={}, classification={})", command.as_str(), error.retryable, response_error_classification(&error), ), - ).with_provider_failure(provider_failure))); + ).with_provider_failure(provider_failure)))); } } } @@ -545,18 +638,32 @@ impl AcpxSidecarTransport { &mut self, timeout: Duration, ) -> Result, LocalRunnerError> { - let Some(line) = self.receive_stdout_line(timeout, "event.poll")? else { - return Ok(None); - }; - match parse_frame(&line)? { - ParsedFrame::Event(event) => { - self.validate_event_sequence(event.sequence)?; - Ok(Some(event)) + let deadline = Instant::now() + timeout; + loop { + let Some(line) = self.receive_stdout_line( + deadline.saturating_duration_since(Instant::now()), + "event.poll", + )? + else { + return Ok(None); + }; + match parse_frame(&line)? { + ParsedFrame::Event(event) => { + self.validate_event_sequence(event.sequence)?; + return Ok(Some(event)); + } + ParsedFrame::Response(response) + if self.abandoned_observation == Some(response.id) => + { + self.abandoned_observation = None; + } + ParsedFrame::Response(response) => { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar emitted response {} without a pending request", + response.id + ))) + } } - ParsedFrame::Response(response) => Err(LocalRunnerError::invalid(format!( - "ACPX sidecar emitted response {} without a pending request", - response.id - ))), } } @@ -1404,3 +1511,7 @@ mod tests { ); } } + +#[cfg(all(test, unix))] +#[path = "tests/acpx_park_observation.rs"] +mod park_observation_tests; 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 42dc16658e6..280adde3421 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 @@ -124,6 +124,9 @@ fn run() -> Result<(), Box> { "pendingToolCount":if fault == "tool" || source_pending_completion.is_some() {1} else {0}, "pendingInputCount":0, "pendingRuntimeRequests":[]}); } + if command == "session.snapshot" && fault == "foreign-snapshot" { + response["result"]["identity"]["agentSessionId"] = json!("foreign-session"); + } if command == "session.goal.get" { source_goal_revision += 1; response["result"]["providerRevision"] = json!(source_goal_revision); @@ -153,6 +156,16 @@ fn run() -> Result<(), Box> { )?; next_sequence += 1; } + if command == "session.snapshot" && fault == "slow-snapshot" { + std::thread::sleep(std::time::Duration::from_secs(1)); + write_json( + &mut stdout, + &json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":next_sequence,"eventType":"runtime.diagnostic","runId":source_run,"turnId":null, + "payload":{"code":"slow_park_observation","message":"Retained after a slow read."}}), + )?; + next_sequence += 1; + } if command == "session.snapshot" && fault == "goal-during-snapshot" { source_goal_revision += 1; let frame = json!({ @@ -172,6 +185,12 @@ fn run() -> Result<(), Box> { response["result"]["sessionGoals"] = json!({"availability":"available", "actions":["set","pause","resume","clear"], "autonomousUpdates":true,"persistentAcrossResume":true,"maxObjectiveChars":4000,"tokenBudgetControl":false,"usageReporting":false}); } + if (command == "session.goal.get" && fault == "reject-goal") + || (command == "session.snapshot" && fault == "reject-snapshot") + { + response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false, + "error":{"code":"temporarily_unavailable","message":"Optional read is temporarily unavailable.","retryable":true}}); + } write_json(&mut stdout, &response)?; if command == "turn.start" { let turn = request["params"]["turnId"].as_str().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 86a0ff184c0..7db36c3ea39 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 @@ -332,6 +332,7 @@ impl CommandExecutor for NativeProviderCommandExecutor { } match self.selected.as_mut() { Some(SelectedExecutor::LocalFacade(executor)) => executor.host_park_ready(), + Some(SelectedExecutor::Acpx(executor)) => executor.host_park_ready(), _ => Ok(false), } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/tests/acpx_park_observation.rs b/packages/paperclip-runner/runner/crates/runner-core/src/tests/acpx_park_observation.rs new file mode 100644 index 00000000000..d96a5bbdc50 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/tests/acpx_park_observation.rs @@ -0,0 +1,94 @@ +use super::*; + +fn scripted_transport(reply_id: u64) -> AcpxSidecarTransport { + let event = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":1,"eventType":"runtime.diagnostic","runId":null,"turnId":null, + "payload":{"code":"park_tail","message":"Preserve this exact event."}}); + let late = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":1,"ok":true,"result":{"late":true}}); + let ordinary = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":reply_id,"ok":true,"result":{"ordinary":true}}); + AcpxSidecarTransport::start(&AcpxSidecarTransportConfig { + command: "/bin/sh".into(), + args: vec!["-c".to_owned(), format!("read first; sleep 0.1; printf '%s\\n' '{event}' '{late}'; read second; printf '%s\\n' '{ordinary}'; cat >/dev/null")], + verified_launch: None, + request_timeout: Duration::from_secs(1), + shutdown_grace: Duration::from_millis(20), + }).unwrap() +} + +#[test] +fn late_park_observation_reply_preserves_ordinary_commands_and_event_polling() { + for poll_first in [false, true] { + let mut transport = scripted_transport(2); + let start = Instant::now(); + assert!(transport + .observe( + GeneratedAcpxSidecarCommand::SessionSnapshot, + start + Duration::from_millis(10) + ) + .unwrap() + .is_none()); + assert_eq!(transport.abandoned_observation, Some(1)); + assert!(transport + .observe( + GeneratedAcpxSidecarCommand::SessionGoalGet, + Instant::now() + Duration::from_millis(10) + ) + .unwrap() + .is_none()); + assert_eq!(transport.next_request_id, 2); + let event = if poll_first { + let event = transport + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap(); + assert!(transport + .poll_event(Duration::from_millis(10)) + .unwrap() + .is_none()); + event + } else { + assert_eq!( + transport + .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({})) + .unwrap(), + json!({"ordinary":true}) + ); + transport.poll_event(Duration::ZERO).unwrap().unwrap() + }; + assert_eq!(event.sequence, 1); + assert_eq!( + event.payload, + json!({"code":"park_tail","message":"Preserve this exact event."}) + ); + assert_eq!(transport.abandoned_observation, None); + if poll_first { + assert_eq!( + transport + .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({})) + .unwrap(), + json!({"ordinary":true}) + ); + } + assert!(!transport.poisoned); + transport.shutdown().unwrap(); + } +} + +#[test] +fn late_park_observation_does_not_relax_ordinary_reply_identity() { + for wrong_id in [1, 99] { + let mut transport = scripted_transport(wrong_id); + assert!(transport + .observe( + GeneratedAcpxSidecarCommand::SessionSnapshot, + Instant::now() + Duration::from_millis(10) + ) + .unwrap() + .is_none()); + let error = transport + .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({})) + .unwrap_err(); + assert!(error.to_string().contains("response id mismatch")); + assert!(transport.poisoned); + } +} 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 26712386365..9303af23960 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 @@ -125,6 +125,169 @@ fn settle(executor: &mut NativeProviderCommandExecutor) -> Vec