diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index ea9c726a09d..832375b1170 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -107,3 +107,16 @@ Cancellation cleanup only settles retained work. Live qualification still needs to verify that restoring a provider starts no work, deployed storage honors fsync, and the old process has exited. + +## Recovering ACPX between turns + +`paperclip-runnerd --build-metadata` prints `acpxRecoveryIntegrityVersion: 1` +when the Runner implements these ACPX checks. The hosted ROM build copies +that field into its manifest. The host driver advertises a separate +`acpxSourceRecoveryVersion: 1` and checks whether this run qualifies. + +The Runner compares the restored session identity, pending requests, turn +and goal with A’s saved idle state. After reading the goal, it checks updates +that arrive during the final status read and keeps them queued for delivery. +As with Codex, a prepared session can continue or be cleaned up. This handoff +requires a completed turn. Suspended sessions use their usual recovery path. 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 d1bef5eaec3..dc10623db86 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 @@ -17,8 +17,9 @@ use sha2::{Digest, Sha256}; use crate::acpx_provider_capabilities::{run_attachment_policy, RunAttachmentPolicy}; use crate::acpx_provider_session::{ - AcpxPermissionMode, AcpxProviderRuntimePolicy, AcpxProviderSession, AcpxProviderSessionConfig, - AcpxProviderSessionIdentity, AcpxTurnControlCapabilities, PiThinkingLevel, + goal_is_dormant, AcpxPermissionMode, AcpxProviderRuntimePolicy, AcpxProviderSession, + AcpxProviderSessionConfig, AcpxProviderSessionIdentity, AcpxTurnControlCapabilities, + PiThinkingLevel, }; use crate::acpx_sidecar_transport::AcpxSidecarTransportConfig; #[cfg(test)] @@ -885,6 +886,7 @@ pub struct AcpxCommandExecutor { #[cfg(test)] fail_after_state_rename: bool, launch_profile: Option, + recovery_pending_effects: Option, } // The durable command's turn and the live provider callback have separate IDs. @@ -923,6 +925,7 @@ impl AcpxCommandExecutor { state: None, session: None, restore_checked: false, + recovery_pending_effects: None, restore_error: None, persistence_error: None, #[cfg(test)] @@ -935,6 +938,10 @@ impl AcpxCommandExecutor { self.state_dir.join(ACPX_PROVIDER_STATE_FILE) } + pub(crate) fn observe_recovery_state(&mut self, state: &crate::durable::DurableState) { + self.recovery_pending_effects = Some(state.recovery_pending_effects()); + } + fn candidate_admission( &self, ) -> Option<&crate::acpx_candidate_admission::AcpxCandidateAdmission> { @@ -1032,6 +1039,11 @@ impl AcpxCommandExecutor { } fn restore_once(&mut self) -> Result<(), DurableRunnerError> { + self.load_state_without_session()?; + self.restore_session_if_needed() + } + + fn load_state_without_session(&mut self) -> Result<(), DurableRunnerError> { let path = self.state_path(); let mut file = match open_private_regular_file(&path) { Ok(file) => file, @@ -1078,7 +1090,7 @@ impl AcpxCommandExecutor { } state.validate(&self.context, &launch_profile_digest)?; self.state = Some(state); - self.restore_session_if_needed() + Ok(()) } fn restore_session_if_needed(&mut self) -> Result<(), DurableRunnerError> { @@ -1154,6 +1166,7 @@ impl AcpxCommandExecutor { priority: EventPriority::P0, payload: session_event_payload(&state.descriptor, &identity, process_id, turn_controls), })?; + self.recovery_pending_effects = None; self.session = Some(session); self.save_state() } @@ -1179,6 +1192,21 @@ impl AcpxCommandExecutor { "ACPX provider identity did not attest the qualified command digest", )); } + let expected_dormant = recovering + && self.recovery_pending_effects == Some(0) + && self.candidate_admission().is_none() + && acpx_release_profile(&state.descriptor.agent) + .is_some_and(|profile| !profile.qualification_pending) + && matches!(state.lifecycle.as_str(), "session_open" | "suspended") + && state.active_turn_id.is_none() + && !state.provider_exit_unconfirmed + && state.pending_runtime_requests.is_empty() + && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection)); + if expected_dormant { + session + .verify_dormant_session() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + } Ok(session) } @@ -2513,6 +2541,7 @@ impl CommandExecutor for AcpxCommandExecutor { } fn rotate_authority(&mut self, config: &DurableRunnerConfig) { + self.recovery_pending_effects = None; self.context.run_id = config.run_id.clone(); self.context.normalized_session_id = config.normalized_session_id.clone(); self.context.turn_id = config.turn_id.clone(); @@ -4959,6 +4988,9 @@ mod tests { let config = test_config(&directory, Some(launch_profile)); let mut executor = AcpxCommandExecutor::with_runner_config(&directory, &config); executor.state = Some(state); + executor.recovery_pending_effects = Some(0); + executor.restore_session_if_needed().unwrap(); + assert!(executor.session.is_none()); let open_error = executor.open_session().unwrap_err(); assert!(open_error .to_string() 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 ca505adab0f..550469d08a9 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 @@ -353,6 +353,26 @@ pub struct AcpxProviderSession { runtime_retired: bool, } +pub(crate) fn goal_is_dormant(projection: &Value) -> bool { + projection["schema"] == "paperclip.session_goal.snapshot.v1" + && projection["workingNow"].as_bool() == Some(false) + && projection + .pointer("/goal/workingNow") + .is_none_or(|working| working.as_bool() == Some(false)) + && (projection.get("goal") == Some(&Value::Null) + || matches!( + projection.pointer("/goal/status").and_then(Value::as_str), + Some( + "paused" + | "blocked" + | "limited" + | "usage_limited" + | "budget_limited" + | "complete" + ) + )) +} + impl AcpxProviderSession { pub fn start(config: &AcpxProviderSessionConfig) -> Result { config.validate()?; @@ -410,6 +430,56 @@ impl AcpxProviderSession { &self.state } + pub(crate) fn verify_dormant_session(&mut self) -> Result<(), LocalRunnerError> { + if self.state.active_turn_id().is_some() || self.state.has_pending_requests() { + return Err(LocalRunnerError::invalid( + "saved dormant ACPX resumed pending work", + )); + } + let goal = self.goal_control(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({}))?; + let prior_events = self.transport.buffered_events().len(); + if !goal_is_dormant(&goal) { + return Err(LocalRunnerError::invalid( + "saved dormant ACPX resumed an active or unknown goal", + )); + } + let snapshot = self + .transport + .request(GeneratedAcpxSidecarCommand::SessionSnapshot, json!({}))?; + let identity: AcpxProviderSessionIdentity = + serde_json::from_value(snapshot["identity"].clone()).map_err(|_| { + LocalRunnerError::invalid("saved dormant ACPX snapshot identity is invalid") + })?; + if identity != self.identity + || snapshot["runId"].as_str() != Some(self.config.run_id.as_str()) + || snapshot.get("turnId") != Some(&Value::Null) + || snapshot["pendingToolCount"].as_u64() != Some(0) + || snapshot["pendingInputCount"].as_u64() != Some(0) + || !snapshot["pendingRuntimeRequests"] + .as_array() + .is_some_and(Vec::is_empty) + { + return Err(LocalRunnerError::invalid( + "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(()) + } + /// 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( @@ -1718,6 +1788,54 @@ fn with_cleanup_error( #[cfg(test)] mod tests { + #[test] + fn dormant_goals_accept_known_inactive_statuses_and_reject_active_or_unknown() { + use super::*; + for (status, expected) in [ + ("paused", true), + ("blocked", true), + ("limited", true), + ("usage_limited", true), + ("budget_limited", true), + ("complete", true), + ("active", false), + ("unknown", false), + ] { + let mut projection = json!({ + "schema":"paperclip.session_goal.snapshot.v1", "workingNow":false, + "goal":{"status":status,"workingNow":false} + }); + assert_eq!(goal_is_dormant(&projection), expected, "{status}"); + projection["goal"] + .as_object_mut() + .unwrap() + .remove("workingNow"); + assert_eq!( + goal_is_dormant(&projection), + expected, + "missing workingNow: {status}" + ); + for working in [ + json!(true), + Value::Null, + json!("false"), + json!("true"), + json!(0), + json!({}), + json!([]), + ] { + projection["goal"]["workingNow"] = working; + assert!( + !goal_is_dormant(&projection), + "{status} has invalid workingNow" + ); + } + projection["goal"]["workingNow"] = json!(false); + projection["workingNow"] = json!(true); + assert!(!goal_is_dormant(&projection), "{status} session is working"); + } + } + #[test] fn startup_cleanup_failure_keeps_the_validated_provider_cause() { use super::*; @@ -1939,6 +2057,180 @@ mod mode_tests { provider_lifetime_fence_candidates: [60_001, 60_002, 60_003], } } + + #[cfg(unix)] + fn snapshot_wire_session( + prior_events: &[crate::acpx_sidecar_transport::AcpxSidecarEvent], + new_events: &[crate::acpx_sidecar_transport::AcpxSidecarEvent], + foreign_session: bool, + ) -> AcpxProviderSession { + let mut cfg = config(); + let identity = identity(); + let mut snapshot_identity = identity.clone(); + if foreign_session { + snapshot_identity.acpx_record_id = "foreign-record".to_owned(); + } + let reply = |id, result| { + json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id":id,"ok":true,"result":result}) + }; + let print = |value: Value| { + format!( + "printf '%s\\n' '{}'\n", + value.to_string().replace('\'', "'\"'\"'") + ) + }; + let mut script = String::new(); + for (id, events, result) in [ + ( + 1, + prior_events, + json!({"schema":"paperclip.session_goal.snapshot.v1","providerRevision":2,"goal":null,"workingNow":false}), + ), + ( + 2, + new_events, + json!({"identity":snapshot_identity,"runId":"run-1","turnId":null, + "pendingToolCount":0,"pendingInputCount":0,"pendingRuntimeRequests":[]}), + ), + (3, &[][..], json!({"closed":true})), + ] { + script.push_str("IFS= read -r request_line\n"); + for event in events { + let event_type = match event.event_type { + GeneratedAcpxSidecarEventType::RuntimeGoal => "runtime.goal", + GeneratedAcpxSidecarEventType::RuntimeDiagnostic => "runtime.diagnostic", + _ => panic!("unsupported fixture event"), + }; + script.push_str(&print( + json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":event.sequence,"eventType":event_type,"runId":event.run_id, + "turnId":event.turn_id,"payload":event.payload}), + )); + } + script.push_str(&print(reply(id, result))); + } + cfg.transport.command = PathBuf::from("/bin/sh"); + cfg.transport.args = vec!["-c".to_owned(), script]; + let transport = AcpxSidecarTransport::start_for_agent(&cfg.transport, &cfg.agent).unwrap(); + AcpxProviderSession { + transport, + state: AcpxProviderState::new(&cfg.run_id).unwrap(), + identity, + tool_bridge: ProviderToolBridge::default(), + reserved_tool_bridge: ProviderToolBridge::default(), + turn_controls: AcpxTurnControlCapabilities::default(), + catalog_revision: cfg.catalog_revision, + working_directory: cfg.working_directory.clone(), + config: cfg, + closed: false, + failure: None, + transport_terminated: false, + runtime_retired: false, + } + } + + #[cfg(unix)] + fn wire_goal( + sequence: u64, + revision: Option, + status: &str, + ) -> crate::acpx_sidecar_transport::AcpxSidecarEvent { + crate::acpx_sidecar_transport::AcpxSidecarEvent { + sequence, + event_type: GeneratedAcpxSidecarEventType::RuntimeGoal, + run_id: Some("run-1".to_owned()), + turn_id: None, + payload: json!({"schema":"paperclip.session_goal.snapshot.v1","providerRevision":revision, + "goal":{"status":status,"workingNow":false},"workingNow":false}), + } + } + + #[test] + #[cfg(unix)] + fn dormant_snapshot_rejects_new_active_goals_regardless_of_revision() { + for (revision, status) in [ + (Some(3), "active"), + (Some(2), "active"), + (Some(1), "active"), + (None, "active"), + (Some(3), "unknown"), + ] { + let events = vec![wire_goal(1, revision, status)]; + let mut session = snapshot_wire_session(&[], &events, false); + let error = session.verify_dormant_session().unwrap_err(); + assert!(error.to_string().contains("during its snapshot"), "{error}"); + assert_eq!( + session + .transport + .buffered_events() + .cloned() + .collect::>(), + events + ); + session.shutdown("test complete").unwrap(); + } + } + + #[test] + #[cfg(unix)] + fn dormant_snapshot_preserves_prior_and_benign_new_events_for_polling() { + let events = vec![ + crate::acpx_sidecar_transport::AcpxSidecarEvent { + sequence: 1, + event_type: GeneratedAcpxSidecarEventType::RuntimeDiagnostic, + run_id: Some("run-1".to_owned()), + turn_id: None, + payload: json!({"code":"probe-observation","message":"Retained observation"}), + }, + wire_goal(2, Some(1), "active"), + wire_goal(3, Some(3), "paused"), + ]; + let mut session = snapshot_wire_session(&events[..2], &events[2..], false); + session.verify_dormant_session().unwrap(); + assert_eq!( + session + .transport + .buffered_events() + .cloned() + .collect::>(), + events + ); + for expected in &events { + let delivered = session.poll_event(Duration::ZERO).unwrap().unwrap(); + if expected.event_type == GeneratedAcpxSidecarEventType::RuntimeGoal { + assert!( + matches!(&delivered[..], [AcpxProviderStateEvent::Goal(goal)] if goal == &expected.payload) + ); + } else { + assert_eq!(delivered.len(), 1); + } + } + assert!(session.poll_event(Duration::ZERO).unwrap().is_none()); + session.shutdown("test complete").unwrap(); + } + + #[test] + #[cfg(unix)] + fn dormant_snapshot_requires_the_same_run_and_session() { + for foreign_session in [false, true] { + let mut event = wire_goal(1, Some(3), "paused"); + if !foreign_session { + event.run_id = Some("foreign-run".to_owned()); + } + let mut session = snapshot_wire_session(&[], &[event], foreign_session); + let error = session.verify_dormant_session().unwrap_err(); + assert!( + error.to_string().contains(if foreign_session { + "identity" + } else { + "stale run" + }), + "{error}" + ); + session.shutdown("test complete").unwrap(); + } + } #[test] fn pending_profile_cannot_launch_even_with_explicit_policy() { let mut config = config(); 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 c2e0bbee6a6..a712f59f430 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 @@ -402,6 +402,10 @@ impl AcpxSidecarTransport { self.process.id() } + pub(crate) fn buffered_events(&self) -> impl ExactSizeIterator { + self.buffered_events.iter() + } + pub fn request( &mut self, command: GeneratedAcpxSidecarCommand, 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 35b3f8b9a1f..b12f224226c 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 @@ -66,6 +66,29 @@ fn run() -> Result<(), Box> { let mut goal = Value::Null; let mut provider_lost = false; let mut session_identity = Value::Null; + let mut source_run = String::new(); + let mut source_pending_completion: Option = None; + let mut source_goal_revision = 0_u64; + let mut source_journal = if mode == "source-reconcile" { + let path = args + .windows(2) + .find(|pair| pair[0] == "--journal") + .ok_or("source journal is missing")?; + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&path[1])?; + write_json(&mut file, &json!({"processStart": std::process::id()}))?; + if args.iter().any(|arg| arg == "--spawn-owned-child") { + let child = std::process::Command::new("/bin/sleep") + .arg("120") + .spawn()?; + write_json(&mut file, &json!({"childPid": child.id()}))?; + } + Some(file) + } else { + None + }; for line in stdin.lock().lines() { let request: Value = serde_json::from_str(&line?)?; let id = request @@ -76,6 +99,115 @@ fn run() -> Result<(), Box> { .get("command") .and_then(Value::as_str) .ok_or("request command is missing")?; + if let Some(journal) = source_journal.as_mut() { + write_json(journal, &request)?; + if command == "run.attach" { + source_run = request["params"]["runId"].as_str().unwrap().to_owned(); + } + let fault = args + .windows(2) + .find(|pair| pair[0] == "--source-fault") + .and_then(|pair| std::fs::read_to_string(&pair[1]).ok()) + .unwrap_or_default(); + let mut response = bootstrap_success( + id, + command, + &request, + mode, + profile_digest, + lifetime_fence_candidates, + &mut session_identity, + ); + if command == "session.snapshot" { + response["result"] = json!({"identity":session_identity, "runId":source_run, + "turnId":if fault == "active" {json!("unexpected-turn")} else {json!(source_pending_completion)}, + "pendingToolCount":if fault == "tool" || source_pending_completion.is_some() {1} else {0}, "pendingInputCount":0, + "pendingRuntimeRequests":[]}); + } + if command == "session.goal.get" { + source_goal_revision += 1; + response["result"]["providerRevision"] = json!(source_goal_revision); + } + if command == "session.snapshot" && fault == "goal-during-snapshot" { + source_goal_revision += 1; + let frame = json!({ + "protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":next_sequence,"eventType":"runtime.goal","runId":source_run,"turnId":null, + "payload":{"schema":"paperclip.session_goal.snapshot.v1","providerRevision":source_goal_revision, + "goal":{"objective":"Changed during final status","status":"active","workingNow":false},"workingNow":false} + }); + write_json(journal, &json!({"emitted":frame}))?; + write_json(&mut stdout, &frame)?; + next_sequence += 1; + } + if command == "session.goal.get" && matches!(fault.as_str(), "goal" | "paused") { + response["result"]["goal"] = json!({"objective":"Continue independently", "status":if fault == "goal" {"active"} else {"paused"}, + "tokenBudget":null,"tokensUsed":null,"elapsedSeconds":null,"iterations":null, + "lastReason":null,"createdAt":null,"updatedAt":null,"completedAt":null,"workingNow":false}); + response["result"]["sessionGoals"] = json!({"availability":"available", "actions":["set","pause","resume","clear"], + "autonomousUpdates":true,"persistentAcrossResume":true,"maxObjectiveChars":4000,"tokenBudgetControl":false,"usageReporting":false}); + } + write_json(&mut stdout, &response)?; + if command == "turn.start" { + let turn = request["params"]["turnId"].as_str().unwrap(); + if args.iter().any(|arg| arg == "--wait-completion-feedback") { + let report = json!({"schema":"paperclip.run_result.v1", "reportedWorkDisposition":"done", "summary":"Completed fake ACPX work", + "completionClaim":{"contractRevision":"1","objectiveSatisfied":true, + "criteria":[{"criterionId":"objective","status":"satisfied","evidenceRefs":[]}],"remainingWork":[]}, + "evidence":[],"verification":[],"attentionRequests":[],"artifacts":[]}); + let call_id = format!("finish-{turn}"); + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + &source_run, + turn, + json!({"callId":call_id,"operationId":"paperclip_finish","input":report}), + )?; + next_sequence += 1; + source_pending_completion = Some(turn.to_owned()); + continue; + } + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + &source_run, + turn, + json!({"type":"text_delta", "text":"Completed fake ACPX work."}), + )?; + next_sequence += 1; + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + &source_run, + turn, + json!({"status":"completed"}), + )?; + next_sequence += 1; + } + if command == "tool.resolve" { + let turn = source_pending_completion + .take() + .ok_or("missing source completion")?; + if request["params"]["turnId"] != turn + || request["params"]["callId"] != format!("finish-{turn}") + { + return Err("source completion identity mismatch".into()); + } + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + &source_run, + &turn, + json!({"status":if request["params"]["error"].is_null() {"completed"} else {"failed"}}), + )?; + next_sequence += 1; + } + continue; + } if mode == "turns-provider-death" && provider_lost && command == "session.goal.get" { write_json( &mut stdout, 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 9b0d056156d..523df43429d 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 @@ -122,6 +122,7 @@ fn build_metadata() -> serde_json::Value { "packageVersion": env!("CARGO_PKG_VERSION"), "binaryContractVersion": 2, "recoveryIntegrityVersion": 1, + "acpxRecoveryIntegrityVersion": 1, "durableSessionCapabilities": ["unlimited_runtime", "connection_lease_renewal"], "externalProviderCapabilities": ["openai_dot_mcp", "grok_bot_mcp", "muse_external_v1"], "nativeExecutionVersion": 1, @@ -674,6 +675,8 @@ mod tests { let metadata = build_metadata(); assert_eq!(metadata["recoveryIntegrityVersion"], 1); assert!(metadata.get("sourceRecoveryVersion").is_none()); + assert_eq!(metadata["acpxRecoveryIntegrityVersion"], 1); + assert!(metadata.get("acpxSourceRecoveryVersion").is_none()); assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA); assert_eq!(metadata["binaryContractVersion"], 2); assert_eq!( 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 2c856c2b68f..5bcc344f329 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 @@ -286,8 +286,10 @@ impl CommandExecutor for NativeProviderCommandExecutor { return Ok(()); } self.select_recovery()?; - if let Some(SelectedExecutor::LocalFacade(executor)) = self.selected.as_mut() { - executor.observe_recovery_state(state); + match self.selected.as_mut() { + Some(SelectedExecutor::LocalFacade(executor)) => executor.observe_recovery_state(state), + Some(SelectedExecutor::Acpx(executor)) => executor.observe_recovery_state(state), + _ => {} } Ok(()) } diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts index 681bf7234fb..482cd81261d 100644 --- a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts +++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts @@ -11,6 +11,8 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; export interface RunnerExecutionDriver { /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */ sourceRecoveryVersion?: 1; + /** Same trusted prelaunch attestation for a qualified ACPX provider. */ + acpxSourceRecoveryVersion?: 1; /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */ authorityRunId?: string; /** Own provider configuration and file preparation in the execution environment. */ diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts new file mode 100644 index 00000000000..10dfe108514 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts @@ -0,0 +1,272 @@ +import { createHash } from "node:crypto"; +import { execFileSync, spawn } from "node:child_process"; +import { mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; +import { appendFileSync, closeSync, openSync, readFileSync } from "node:fs"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import type { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js"; +import { QUALIFIED_ACPX_PROFILES, resolveQualifiedAcpxProfile } from "../drivers/acpx/qualified-profiles.js"; +import { createCapabilityRunnerdCodexTransport, readRunnerdArtifactBinding, type CapabilityRunnerdCodexTransportOptions } from "./runnerd-codex-transport.js"; +import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js"; +import { defaultCapabilityRunnerdBinary } from "../../test/support/runnerd-transport.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; +import { createNativeSessionBackend } from "../backends/native-backend-factory.js"; +import type { NativeExecutionInput } from "../contracts/native-execution.js"; +import type { PersistedNativeSession } from "../contracts/native-session-backend.js"; + +type Bundle = ReturnType; +const read = async (path: string) => JSON.parse(await readFile(path, "utf8")); +const sha = (value: string | Buffer) => `sha256:${createHash("sha256").update(value).digest("hex")}`; +const quote = (value: string) => `'${value.replaceAll("'", "'\"'\"'")}'`; +const kill = (pid: number) => { try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } }; +const absent = (pid: number) => { try { process.kill(pid, 0); return false; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; } }; +const stopped = (pid: number) => { + try { process.kill(pid, 0); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; } + try { return execFileSync("ps", ["-p", String(pid), "-o", "stat="], { encoding: "utf8" }).trim().startsWith("Z"); } + catch (error) { if ((error as { status?: number }).status === 1) return true; throw error; } +}; + +async function fixture(agent: "codex" | "claude" = "codex", nativeSource = false) { + const root = await realpath(await mkdtemp(join(tmpdir(), "runner-acpx-source-"))); + const journal = join(root, "calls"), fault = join(root, "fault"), wrapper = join(root, "sidecar"), script = join(root, "sidecar-entry"); + const binary = defaultCapabilityRunnerdBinary(), sidecar = resolve(import.meta.dirname, "../../runner/target/debug/fake-acpx-sidecar"); + await mkdir(join(root, "acpx"), { mode: 0o700 }); await mkdir(join(root, "diagnostics"), { mode: 0o700 }); + await writeFile(script, "qualified fixture entry"); + await writeFile(wrapper, `#!/bin/sh\nexec ${quote(sidecar)} --mode source-reconcile --profile-digest ${quote(QUALIFIED_ACPX_PROFILES[agent].commandDigest)} --journal ${quote(journal)} --source-fault ${quote(fault)} --spawn-owned-child${nativeSource ? " --wait-completion-feedback" : ""}\n`, { mode: 0o700 }); + let authority: DurablePrpControlPlane | null = null; + let registrationGeneration = 0; + const server = createServer(); + server.on("upgrade", (request, socket, head) => authority ? authority.handleUpgrade(request, socket, "/runner", head) : socket.destroy()); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); if (!address || typeof address === "string") throw new Error("missing fixture listener"); + const identity = { runnerInstanceId: "runner-source", environmentLeaseId: "lease-source", runId: "run-source", normalizedSessionId: "session-source", turnId: "turn-source", itemId: "item-source" }; + const nativeIdentity = { companyId: "company", agentId: "agent", issueId: "issue", runId: identity.runId, sessionId: identity.normalizedSessionId }; + const execution: NativeExecutionInput = { schema: "paperclip.native-execution-input.v1", + binding: { companyId: nativeIdentity.companyId, agentId: nativeIdentity.agentId, issueId: nativeIdentity.issueId, runId: identity.runId, executionWorkspaceId: "workspace" }, + task: { identifier: "TASK-1", title: "Complete fake work", description: null, prompt: "Complete fake work", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: identity.normalizedSessionId, driverKind: "acpx_runtime", protocolVersion: 1 }, + provider: { kind: "acpx", agent, model: "explicit-source-model", permissionPolicy: "interactive", profile: resolveQualifiedAcpxProfile(agent, "explicit-source-model") }, + completionContract: { id: "contract", sha256: sha("contract"), schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Complete fake work", criteria: [{ id: "objective", requirement: "Complete fake work" }] } }, interactionResponses: [], credentialBindings: [] }; + const bundles: Bundle[] = [], runnerPids = new Set(); + let requirements: RunnerProviderLaunchRequirements | undefined; + let recovering = false; + const options: CapabilityRunnerdCodexTransportOptions = { + provider: "acpx", acpxAgent: agent, runnerBinary: binary, stateDirectory: root, prpIdentity: identity, + runnerArtifact: readRunnerdArtifactBinding(binary), sourceRecoveryVersion: 1, acpxSourceRecoveryVersion: 1, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000, + controlPlaneRegistration: async next => { + const generation = ++registrationGeneration; authority = next; + return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: () => { if (generation === registrationGeneration) authority = null; } }; + }, + prepareLaunch: async input => { + requirements ??= structuredClone(input); + return { configurationId: `${recovering ? "source" : "initial"}-${input.identity.runId}`, requirementsDigest: runnerLaunchRequirementsDigest(input), + provider: createRunnerProviderLaunch(input, { options: {}, root, cwd: root, codexHome: root, codexCommand: "", providerNodeCommand: wrapper, opencodeProxyPath: "", acpxSidecarPath: script }) }; + }, + runnerProcessLauncher: spec => { + const stderr = openSync(join(root, "native.stderr"), "a", 0o600); + const child = spawn(binary, [...spec.args, "--acpx-launch-authority-digest", sha("fixture authority"), + "--acpx-sidecar-command", wrapper, "--acpx-sidecar-command-sha256", sha(execFileSync("cat", [wrapper])), + "--acpx-sidecar-script", script, "--acpx-sidecar-script-sha256", sha("qualified fixture entry")], + { env: { PATH: process.env.PATH, HOME: root, ...spec.environment }, stdio: ["ignore", "ignore", stderr], detached: true }); + closeSync(stderr); + if (child.pid) runnerPids.add(child.pid); + return { child, completion: new Promise((resolve, reject) => { child.once("error", reject); child.once("exit", (code, signal) => { + try { appendFileSync(join(root, "native.stderr"), readFileSync(join(root, "diagnostics", "runnerd.stderr.log"))); } catch {} + resolve({ code, signal, stdout: "", stderr: "" }); + }); }) }; + }, + }; + const runnerFile = join(root, "runner", "runner-state.json"), providerFile = join(root, "runner", "acpx-provider-state.json"), coreFile = join(root, "control-plane", "control-plane-state.json"); + const calls = async () => (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + const count = async (command: string) => (await calls()).filter(entry => command === "processStart" ? entry.processStart : entry.command === command).length; + const settle = async () => vi.waitFor(async () => { + const runner = await read(runnerFile), provider = await read(providerFile), core = await read(coreFile); + expect(runner.outbox).toEqual([]); expect(runner.ackedSourceSeq).toBe(runner.nextSourceSeq - 1); + expect(provider.activeTurnId).toBeNull(); expect(provider.pendingEvents).toEqual([]); expect(provider.pendingRuntimeRequests).toEqual({}); + expect(core.commands.some((command: { status: string }) => command.status === "pending")).toBe(false); + expect(authority?.semanticToolResultsSettled()).toBe(true); + }, { timeout: 5_000 }); + const complete = async (bundle: Bundle) => { + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete the fake ACPX turn." }] }); + for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return; + throw new Error("missing completed turn"); + }; + const cleanup = async () => { + for (const bundle of bundles.reverse()) await bundle.detachControllerForRestart().catch(() => {}); + for (const pid of runnerPids) kill(pid); + for (const entry of await calls().catch(() => [])) if (entry.processStart) kill(entry.processStart); + server.closeAllConnections(); await new Promise(resolve => server.close(() => resolve())); + await rm(root, { recursive: true, force: true }); + }; + try { + const first = createCapabilityRunnerdCodexTransport(options); bundles.push(first); + let nativeSnapshot: PersistedNativeSession | undefined; + let thread: { id: string; sessionId: string }; + if (nativeSource) { + const session = await createNativeSessionBackend(execution, { codexTransportFactory: () => first.transport, + environment: { PAPERCLIP_WORKSPACE_CWD: root }, dynamicTools: [] }).openSession({ identity: nativeIdentity, workingDirectory: root }); + await session.startTurn({ message: { role: "user", text: "Complete before abrupt loss." } }); + for await (const event of session.events()) if (event.eventType === "turn.completed") break; + nativeSnapshot = await session.snapshot(); + expect(nativeSnapshot.terminal?.runTerminalState).toBe("succeeded"); + expect(nativeSnapshot.semanticResult?.reportedWorkDisposition).toBe("done"); + thread = { id: nativeSnapshot.sessionId, sessionId: nativeSnapshot.providerSessionId! }; + } else { + const opened = await first.transport.request("thread/start", { cwd: root, model: "explicit-source-model", dynamicTools: [] }); + thread = { id: (opened.thread as { id: string }).id, sessionId: (opened.thread as { sessionId: string }).sessionId }; + await complete(first); + } + await settle(); + expect((await read(coreFile)).committedEvents.some((event: { eventType: string; envelope: { payload: { payload: { runTerminalState?: string } } } }) => event.eventType === "run.terminal" && event.envelope.payload.payload.runTerminalState === "succeeded")).toBe(true); + const providerIdentity = (await read(providerFile)).identity; + const pid = first.evidence().runnerPid!; + await first.detachControllerForRestart(); kill(pid); + for (const entry of await calls()) if (entry.processStart) kill(entry.processStart); + await vi.waitFor(async () => { expect(stopped(pid)).toBe(true); for (const entry of await calls()) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true); }); + const recover = (overrides: Partial = {}) => { + recovering = true; + const bundle = createCapabilityRunnerdCodexTransport({ ...options, recoveryLaunch: { requirements: requirements! }, + resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId, providerIdentity }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return overrides.authorityRunId ?? "run-target"; } }); + bundles.push(bundle); return bundle; + }; + return { root, fault, runnerFile, providerFile, coreFile, identity, providerIdentity, thread, calls, count, settle, complete, recover, cleanup, nativeIdentity, nativeSnapshot, execution }; + } catch (error) { await cleanup(); throw error; } +} + +it.skipIf(process.platform === "win32")("recovers and adopts ACPX through the real native backend before attachment", async () => { + const f = await fixture("claude", true); + try { + let owner = "run-target"; + let source = f.recover({ get authorityRunId() { return owner; } }); + const backend = () => createNativeSessionBackend(f.execution, { codexTransportFactory: () => source.transport, + environment: { PAPERCLIP_WORKSPACE_CWD: f.root }, dynamicTools: [] }); + let recovered = await backend().recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal }); + expect(recovered.recovered).toBe(true); + const pid = source.evidence().runnerPid, startedAt = source.transport.processInfo().startedAt; + expect(await f.count("turn.start")).toBe(1); expect(await f.count("processStart")).toBe(2); + await recovered.session!.detachControllerForRestart!(); + const launch = vi.fn(() => { throw new Error("duplicate source launch"); }); + source = f.recover({ get authorityRunId() { return owner; }, runnerProcessLauncher: launch, authorizeRunnerAdoption: async () => { throw new Error("old source configuration"); }, + adoptExistingRunner: { pid, processGroupId: pid, startedAt, isAlive: () => !stopped(pid!), signal: () => false } }); + recovered = await backend().recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal }); + expect(recovered.recovered).toBe(true); expect(launch).not.toHaveBeenCalled(); expect(source.evidence().runnerPid).toBe(pid); + await expect(source.transport.request("turn/start", {})).rejects.toThrow("source_attachment_pending"); + for (const runId of ["run-target", "run-warm-third"]) { + owner = runId; + await recovered.session!.attachRun!({ identity: { ...f.nativeIdentity, runId } }); + expect(source.evidence().runnerPid).toBe(pid); + await recovered.session!.startTurn({ message: { role: "user", text: "Continue after authenticated attachment." } }); + let completed = false; + for await (const event of recovered.session!.events()) { + if (event.eventType === "turn.completed" && event.runId === runId) { completed = true; break; } + } + expect(completed).toBe(true); await f.settle(); + const snapshot = await recovered.session!.snapshot(); + expect(snapshot.sessionId).toBe(f.thread.id); expect(snapshot.providerIdentity).toEqual(f.providerIdentity); + expect(snapshot.semanticResult?.reportedWorkDisposition).toBe("done"); expect(snapshot.terminal).toMatchObject({ runTerminalState: "succeeded" }); + } + expect(await f.count("turn.start")).toBe(3); expect(await f.count("processStart")).toBe(4); + expect(await f.count("tool.resolve")).toBe(3); + await recovered.session!.close({ reason: "test complete" }); + } catch (error) { + const diagnostics = await readFile(join(f.root, "native.stderr"), "utf8").catch(() => ""); + throw new Error(`${String(error)}\n${diagnostics}`, { cause: error }); + } finally { await f.cleanup(); } +}, 30_000); + +it.skipIf(process.platform === "win32").each(["codex", "claude", "adopt", "paused", "retained"] as const)("recovers qualified ACPX %s under source authority before the second turn", async mode => { + const f = await fixture(mode === "claude" ? "claude" : "codex"); + try { + await expect(f.recover({ acpxSourceRecoveryVersion: undefined }).transport.request("thread/read", {})).rejects.toThrow("recovery_integrity_unsupported"); + if (mode === "paused") await writeFile(f.fault, "paused"); + if (mode === "retained") { + const state = await read(f.providerFile); + for (const [eventType, payload] of [["usage.reported", { cumulative: { inputTokens: 3 } }], ["harness.diagnostic", { code: "retained-source-maintenance", message: "Retained maintenance fact" }]]) { + state.pendingEvents.push({ executorEventId: `acpx_provider_${String(state.nextEventSequence++).padStart(16, "0")}`, eventType, priority: "p0", payload }); + } + await writeFile(f.providerFile, JSON.stringify(state)); + } + const accounting = vi.fn(async (_event: PrpEvent) => {}); + let source = f.recover({ onRecoveryAccountingEvent: accounting }); + expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread); await f.settle(); + await expect(source.transport.request("turn/start", {})).rejects.toThrow("source_attachment_pending"); + const pid = source.evidence().runnerPid; + if (mode === "adopt") { + const startedAt = source.transport.processInfo().startedAt; + await source.detachControllerForRestart(); + const launch = vi.fn(() => { throw new Error("duplicate source launch"); }); + source = f.recover({ runnerProcessLauncher: launch, authorizeRunnerAdoption: async () => { throw new Error("old source configuration"); }, + adoptExistingRunner: { pid, processGroupId: pid, startedAt, isAlive: () => !stopped(pid!), signal: () => false } }); + expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread); + expect(source.evidence().runnerPid).toBe(pid); expect(launch).not.toHaveBeenCalled(); + } + await source.transport.attachRun!({ runId: "run-target", turnId: "turn-target", itemId: "item-target" }); + expect(source.evidence().runnerPid).toBe(pid); + await f.complete(source); await f.settle(); + expect((await read(f.providerFile)).identity).toEqual(f.providerIdentity); + expect(await f.count("turn.start")).toBe(2); expect(await f.count("processStart")).toBe(3); + if (mode === "retained") { + for (const type of ["usage.reported", "harness.diagnostic"]) { + const delivered = accounting.mock.calls.filter(call => call[0].eventType === type); + expect(delivered).toHaveLength(1); expect(delivered[0]![0]).toHaveProperty("runId", f.identity.runId); + } + } + await source.transport.close(); expect((await read(f.runnerFile)).lifecycle).toBe("suspended"); + expect(await f.count("processStart")).toBe(3); + } catch (error) { + const diagnostics = (await readFile(join(f.root, "native.stderr"), "utf8").catch(() => "")) + + (await readFile(join(f.root, "diagnostics", "runnerd.stderr.log"), "utf8").catch(() => "")); + throw new Error(`${String(error)}\n${diagnostics}`, { cause: error }); + } finally { await f.cleanup(); } +}, 30_000); + +it.skipIf(process.platform === "win32").each(["goal", "active", "tool", "goal-during-snapshot"] as const)("stops the owned ACPX process group after live %s admission fails", async fault => { + const f = await fixture(); + try { + await writeFile(f.fault, fault); + await expect(f.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant ACPX"); + await vi.waitFor(async () => { + const calls = await f.calls(); + expect(absent(calls.filter(entry => entry.processStart).at(-1)!.processStart)).toBe(true); + for (const entry of calls) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true); + }, { timeout: 5_000 }); + expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1); + if (fault === "goal-during-snapshot") { + expect((await f.calls()).find(entry => entry.emitted?.eventType === "runtime.goal")?.emitted) + .toMatchObject({ runId: f.identity.runId, turnId: null, payload: { providerRevision: 2, goal: { status: "active" } } }); + } + } finally { await f.cleanup(); } +}, 30_000); + +it.skipIf(process.platform === "win32").each(["limited", "usage_limited", "budget_limited"] as const)("rejects a saved %s ACPX goal that resumes active and reaps the sidecar", async status => { + const f = await fixture("claude"); + try { + const state = await read(f.providerFile); + state.goalProjection = { schema: "paperclip.session_goal.snapshot.v1", goal: { objective: "Settled limited goal", status, workingNow: false }, workingNow: false }; + await writeFile(f.providerFile, JSON.stringify(state)); await writeFile(f.fault, "goal"); + await expect(f.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant ACPX resumed an active or unknown goal"); + await vi.waitFor(async () => { + const calls = await f.calls(); + expect(absent(calls.filter(entry => entry.processStart).at(-1)!.processStart)).toBe(true); + for (const entry of calls) if (entry.processStart || entry.childPid) expect(stopped(entry.processStart ?? entry.childPid)).toBe(true); + }, { timeout: 5_000 }); + expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1); + } finally { await f.cleanup(); } +}, 30_000); + +it.skipIf(process.platform === "win32")("preserves ordinary same-run active goal restoration", async () => { + const f = await fixture(); + try { + const state = await read(f.providerFile); + state.goalProjection = { schema: "paperclip.session_goal.snapshot.v1", goal: { objective: "Authorized same-run goal", status: "active" }, workingNow: false }; + await writeFile(f.providerFile, JSON.stringify(state)); await writeFile(f.fault, "goal"); + const sameRun = f.recover({ authorityRunId: f.identity.runId }); + expect((await sameRun.transport.request("thread/read", {})).thread).toMatchObject(f.thread); + expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1); + } finally { await f.cleanup(); } +}, 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 7452a2e5944..d81e7a08123 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -5605,8 +5605,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { })))) { throw new Error("native_runner_source_reconciliation_requires_exact_authority"); } - if (lostPredecessor && (this.options.provider ?? "codex") !== "codex") throw new Error("native_runner_recovery_integrity_unsupported"); - if (lostPredecessor && this.options.sourceRecoveryVersion !== 1) { + if (lostPredecessor && (!["codex", "acpx"].includes(this.options.provider ?? "codex") || this.options.acpxCandidateAdmission)) throw new Error("native_runner_recovery_integrity_unsupported"); + const recoveryCapability = this.options.provider === "acpx" ? this.options.acpxSourceRecoveryVersion : this.options.sourceRecoveryVersion; + if (lostPredecessor && recoveryCapability !== 1) { throw new Error("native_runner_recovery_integrity_unsupported"); } const hasControlPlaneWarmBoundary =