diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index e535d248bd..01a443d8dd 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -117,6 +117,11 @@ terminates the local process. Already validated ACPX reducer events project into provider-neutral durable events only with an exact run, session, turn, and item binding. Raw sidecar envelopes and permission requests are not admitted at this boundary. +A safely suspended session can be recorded as a bounded private checkpoint. +The checkpoint binds the exact provider identity, run, catalog revision, and +catalog digest and is replaced atomically before a later recovery attempt. +Recovery releases the stored identity only after those bindings match the +prospective session configuration exactly. Run the complete contract gate with: diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_checkpoint.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_checkpoint.rs new file mode 100644 index 0000000000..68e6526388 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_checkpoint.rs @@ -0,0 +1,337 @@ +use std::fs::{self, DirBuilder}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; + +#[cfg(unix)] +use std::fs::File; +#[cfg(unix)] +use std::os::unix::fs::DirBuilderExt; + +use serde::{Deserialize, Serialize}; + +use crate::acpx_provider_session::{ + AcpxPermissionMode, AcpxProviderSessionConfig, AcpxProviderSessionIdentity, +}; +use crate::durable::{ + create_private_temporary_file, open_private_regular_file, verify_private_directory, +}; +use crate::local_runner::LocalRunnerError; +use crate::stable_identity::{is_stable_id, SHORT_STABLE_ID_CHARS}; + +const CHECKPOINT_SCHEMA: &str = "paperclip.runner.acpx-suspension-checkpoint.v1"; +const CHECKPOINT_DIRECTORY: &str = "acpx-provider"; +const CHECKPOINT_FILE: &str = "suspension-checkpoint.json"; +const MAX_CHECKPOINT_BYTES: u64 = 1024 * 1024; +const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991; + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct AcpxSuspensionCheckpoint { + schema: String, + run_id: String, + normalized_session_id: String, + catalog_revision: u64, + catalog_digest: String, + identity: PersistedAcpxProviderSessionIdentity, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct PersistedAcpxProviderSessionIdentity { + kind: String, + normalized_session_id: String, + acpx_record_id: String, + backend_session_id: String, + agent_session_id: String, + profile_digest: String, + workspace_digest: String, + requested_model: String, + effective_model: String, + permission_mode: AcpxPermissionMode, +} + +impl PersistedAcpxProviderSessionIdentity { + fn from_runtime(identity: AcpxProviderSessionIdentity) -> Result { + identity.validate()?; + let permission_mode = identity.permission_mode.ok_or_else(|| { + LocalRunnerError::invalid( + "ACPX suspension checkpoint identity omitted its pinned permission mode", + ) + })?; + Ok(Self { + kind: identity.kind, + normalized_session_id: identity.normalized_session_id, + acpx_record_id: identity.acpx_record_id, + backend_session_id: identity.backend_session_id, + agent_session_id: identity.agent_session_id, + profile_digest: identity.profile_digest, + workspace_digest: identity.workspace_digest, + requested_model: identity.requested_model, + effective_model: identity.effective_model, + permission_mode, + }) + } + + fn runtime_identity(&self) -> AcpxProviderSessionIdentity { + AcpxProviderSessionIdentity { + kind: self.kind.clone(), + normalized_session_id: self.normalized_session_id.clone(), + acpx_record_id: self.acpx_record_id.clone(), + backend_session_id: self.backend_session_id.clone(), + agent_session_id: self.agent_session_id.clone(), + profile_digest: self.profile_digest.clone(), + workspace_digest: self.workspace_digest.clone(), + requested_model: self.requested_model.clone(), + effective_model: self.effective_model.clone(), + permission_mode: Some(self.permission_mode), + } + } + + fn validate(&self) -> Result<(), LocalRunnerError> { + self.runtime_identity().validate() + } +} + +impl AcpxSuspensionCheckpoint { + pub fn from_suspension( + config: &AcpxProviderSessionConfig, + identity: AcpxProviderSessionIdentity, + ) -> Result { + config.validate()?; + identity.validate()?; + if identity.normalized_session_id != config.normalized_session_id + || identity.requested_model != config.model + || identity.effective_model != config.model + || identity.permission_mode != Some(config.permission_mode) + || config + .expected_identity + .as_ref() + .is_some_and(|expected| expected != &identity) + { + return Err(LocalRunnerError::invalid( + "ACPX suspended identity conflicts with the admitted session configuration", + )); + } + let checkpoint = Self { + schema: CHECKPOINT_SCHEMA.to_owned(), + run_id: config.run_id.clone(), + normalized_session_id: config.normalized_session_id.clone(), + catalog_revision: config.catalog_revision, + catalog_digest: config.tool_set.catalog_digest.clone(), + identity: PersistedAcpxProviderSessionIdentity::from_runtime(identity)?, + }; + checkpoint.validate()?; + Ok(checkpoint) + } + + pub fn validate(&self) -> Result<(), LocalRunnerError> { + if self.schema != CHECKPOINT_SCHEMA { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint schema is unsupported", + )); + } + validate_stable_id(&self.run_id, "run")?; + validate_stable_id(&self.normalized_session_id, "normalized session")?; + if self.catalog_revision == 0 || self.catalog_revision > MAX_JSON_SAFE_INTEGER { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint catalog revision is invalid", + )); + } + if !is_sha256_digest(&self.catalog_digest) { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint catalog digest is invalid", + )); + } + self.identity.validate()?; + if self.identity.normalized_session_id != self.normalized_session_id { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint session identity is inconsistent", + )); + } + Ok(()) + } + + pub fn admit_recovery( + &self, + config: &AcpxProviderSessionConfig, + ) -> Result { + self.validate()?; + config.validate()?; + let identity = self.identity.runtime_identity(); + let expected = Self::from_suspension(config, identity.clone())?; + if &expected != self { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint conflicts with the recovery configuration", + )); + } + Ok(identity) + } +} + +#[derive(Clone, Debug)] +pub struct AcpxSuspensionCheckpointStore { + directory: PathBuf, + path: PathBuf, +} + +impl AcpxSuspensionCheckpointStore { + pub fn new(runner_state_directory: &Path) -> Result { + secure_directory(runner_state_directory, "runner state")?; + let directory = runner_state_directory.join(CHECKPOINT_DIRECTORY); + secure_directory(&directory, "ACPX checkpoint")?; + Ok(Self { + path: directory.join(CHECKPOINT_FILE), + directory, + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn load(&self) -> Result, LocalRunnerError> { + let file = match open_private_regular_file(&self.path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => { + return Err(LocalRunnerError::invalid(format!( + "failed to open the private ACPX suspension checkpoint: {error}" + ))) + } + }; + let length = file + .metadata() + .map_err(|error| { + LocalRunnerError::invalid(format!( + "failed to inspect the ACPX suspension checkpoint: {error}" + )) + })? + .len(); + if length > MAX_CHECKPOINT_BYTES { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint exceeds its 1 MiB bound", + )); + } + let bytes = read_checkpoint_bytes(file, length)?; + let checkpoint: AcpxSuspensionCheckpoint = + serde_json::from_slice(&bytes).map_err(|error| { + LocalRunnerError::invalid(format!( + "ACPX suspension checkpoint is malformed: {error}" + )) + })?; + checkpoint.validate()?; + Ok(Some(checkpoint)) + } + + pub fn save(&self, checkpoint: &AcpxSuspensionCheckpoint) -> Result<(), LocalRunnerError> { + checkpoint.validate()?; + verify_private_directory(&self.directory).map_err(|error| { + LocalRunnerError::invalid(format!( + "ACPX checkpoint directory is no longer private: {error}" + )) + })?; + let bytes = serde_json::to_vec_pretty(checkpoint).map_err(|error| { + LocalRunnerError::invalid(format!( + "failed to serialize the ACPX suspension checkpoint: {error}" + )) + })?; + if bytes.len() as u64 > MAX_CHECKPOINT_BYTES { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint exceeds its 1 MiB bound", + )); + } + let (temporary, mut file) = create_private_temporary_file(&self.path).map_err(|error| { + LocalRunnerError::invalid(format!( + "failed to create a private ACPX checkpoint file: {error}" + )) + })?; + let result = (|| -> std::io::Result<()> { + file.write_all(&bytes)?; + file.sync_all()?; + drop(file); + fs::rename(&temporary, &self.path)?; + #[cfg(unix)] + File::open(&self.directory)?.sync_all()?; + Ok(()) + })(); + if let Err(error) = result { + let _ = fs::remove_file(&temporary); + return Err(LocalRunnerError::invalid(format!( + "failed to atomically replace the ACPX suspension checkpoint: {error}" + ))); + } + Ok(()) + } +} + +fn read_checkpoint_bytes( + reader: impl Read, + capacity_hint: u64, +) -> Result, LocalRunnerError> { + let mut reader = reader.take(MAX_CHECKPOINT_BYTES + 1); + let mut bytes = Vec::with_capacity(capacity_hint.min(MAX_CHECKPOINT_BYTES) as usize); + reader.read_to_end(&mut bytes).map_err(|error| { + LocalRunnerError::invalid(format!( + "failed to read the ACPX suspension checkpoint: {error}" + )) + })?; + if bytes.len() as u64 > MAX_CHECKPOINT_BYTES { + return Err(LocalRunnerError::invalid( + "ACPX suspension checkpoint exceeds its 1 MiB bound", + )); + } + Ok(bytes) +} + +fn secure_directory(path: &Path, label: &str) -> Result<(), LocalRunnerError> { + let mut builder = DirBuilder::new(); + #[cfg(unix)] + builder.mode(0o700); + match builder.create(path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => { + return Err(LocalRunnerError::invalid(format!( + "failed to create {label} directory: {error}" + ))) + } + } + // The leaf is created with its private mode in the atomic mkdir operation. + // Never chmod a path after a metadata check: an attacker could replace the + // leaf with a symlink between those calls and redirect the permission write. + // Existing paths must already satisfy the same fail-closed contract. + verify_private_directory(path).map_err(|error| { + LocalRunnerError::invalid(format!("{label} directory is not private: {error}")) + }) +} + +fn validate_stable_id(value: &str, label: &str) -> Result<(), LocalRunnerError> { + if !is_stable_id(value, SHORT_STABLE_ID_CHARS) { + return Err(LocalRunnerError::invalid(format!( + "ACPX suspension checkpoint {label} identity is invalid" + ))); + } + Ok(()) +} + +fn is_sha256_digest(value: &str) -> bool { + value.len() == 71 + && value.starts_with("sha256:") + && value[7..] + .chars() + .all(|character| character.is_ascii_hexdigit() && !character.is_ascii_uppercase()) +} + +#[cfg(test)] +mod tests { + use super::read_checkpoint_bytes; + + #[test] + fn bounded_checkpoint_reader_rejects_growth_beyond_the_metadata_hint() { + let error = read_checkpoint_bytes(std::io::repeat(b'x'), 0) + .unwrap_err() + .to_string(); + + assert!(error.contains("1 MiB"), "{error}"); + } +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs index 4d2abff54e..8af533bacc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -2,6 +2,7 @@ pub mod acpx_event_payload; pub mod acpx_event_scope; +pub mod acpx_provider_checkpoint; pub mod acpx_provider_session; pub mod acpx_provider_state; pub mod acpx_sidecar_transport; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs new file mode 100644 index 0000000000..3da5419b7f --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs @@ -0,0 +1,248 @@ +use std::fs; +use std::path::PathBuf; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; + +use paperclip_runner_core::acpx_provider_checkpoint::{ + AcpxSuspensionCheckpoint, AcpxSuspensionCheckpointStore, +}; +use paperclip_runner_core::acpx_provider_session::{ + AcpxPermissionMode, AcpxProviderSessionConfig, AcpxProviderSessionIdentity, +}; +use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarTransportConfig; +use paperclip_runner_core::provider_bridge::{ + authorized_tool_catalog_digest, AuthorizedTool, AuthorizedToolSet, +}; +use serde_json::json; + +fn temporary_directory(label: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let directory = std::env::temp_dir().join(format!( + "paperclip-acpx-checkpoint-{label}-{}-{nonce}", + std::process::id() + )); + fs::create_dir_all(&directory).unwrap(); + #[cfg(unix)] + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700)).unwrap(); + directory +} + +fn config(directory: &std::path::Path) -> AcpxProviderSessionConfig { + let operations = Vec::new(); + AcpxProviderSessionConfig { + transport: AcpxSidecarTransportConfig { + command: PathBuf::from(env!("CARGO_BIN_EXE_fake-acpx-sidecar")), + args: vec!["--mode".to_owned(), "suspend".to_owned()], + request_timeout: Duration::from_secs(1), + shutdown_grace: Duration::from_millis(100), + }, + agent: "codex".to_owned(), + model: "gpt-5.6-sol".to_owned(), + run_id: "run-1".to_owned(), + catalog_revision: 7, + runtime_directory: directory.to_owned(), + normalized_session_id: "session-1".to_owned(), + working_directory: directory.to_owned(), + permission_mode: AcpxPermissionMode::ApproveReads, + permission_mode_pinned: true, + system_instructions: "Complete the supplied task.".to_owned(), + tool_set: AuthorizedToolSet { + schema: "paperclip.runner.authorized-tools.v1".to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + }, + expected_identity: None, + } +} + +fn identity() -> AcpxProviderSessionIdentity { + AcpxProviderSessionIdentity { + kind: "acpx".to_owned(), + normalized_session_id: "session-1".to_owned(), + acpx_record_id: "acpx-record-1".to_owned(), + backend_session_id: "backend-session-1".to_owned(), + agent_session_id: "agent-session-1".to_owned(), + profile_digest: format!("sha256:{}", "a".repeat(64)), + workspace_digest: format!("sha256:{}", "b".repeat(64)), + requested_model: "gpt-5.6-sol".to_owned(), + effective_model: "gpt-5.6-sol".to_owned(), + permission_mode: Some(AcpxPermissionMode::ApproveReads), + } +} + +#[test] +fn round_trips_an_exact_private_suspension_checkpoint_idempotently() { + let directory = temporary_directory("round-trip"); + let config = config(&directory); + let checkpoint = AcpxSuspensionCheckpoint::from_suspension(&config, identity()).unwrap(); + let store = AcpxSuspensionCheckpointStore::new(&directory).unwrap(); + + assert_eq!(store.load().unwrap(), None); + store.save(&checkpoint).unwrap(); + store.save(&checkpoint).unwrap(); + let recovered = store.load().unwrap().unwrap(); + assert_eq!(recovered, checkpoint); + assert_eq!(recovered.admit_recovery(&config).unwrap(), identity()); + + #[cfg(unix)] + { + assert_eq!( + fs::metadata(store.path()).unwrap().permissions().mode() & 0o077, + 0 + ); + assert_eq!( + fs::metadata(store.path().parent().unwrap()) + .unwrap() + .permissions() + .mode() + & 0o077, + 0 + ); + } + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn admits_recovery_only_for_the_exact_run_catalog_and_provider_identity() { + let directory = temporary_directory("recovery-admission"); + let config = config(&directory); + let checkpoint = AcpxSuspensionCheckpoint::from_suspension(&config, identity()).unwrap(); + + let mut changed_run = config.clone(); + changed_run.run_id = "run-2".to_owned(); + let mut changed_session = config.clone(); + changed_session.normalized_session_id = "session-2".to_owned(); + let mut changed_revision = config.clone(); + changed_revision.catalog_revision += 1; + let mut changed_catalog = config.clone(); + let operations = vec![AuthorizedTool { + operation_id: "issues.read".to_owned(), + version: 1, + description: "Read one issue".to_owned(), + input_schema: json!({"type":"object"}), + response_schema: json!({"type":"object"}), + }]; + changed_catalog.tool_set = AuthorizedToolSet { + schema: "paperclip.runner.authorized-tools.v1".to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + }; + let mut changed_model = config.clone(); + changed_model.model = "gpt-5.6-sol-mini".to_owned(); + let mut changed_permission = config.clone(); + changed_permission.permission_mode = AcpxPermissionMode::ApproveAll; + let mut changed_expected_identity = config.clone(); + let mut expected = identity(); + expected.backend_session_id = "backend-session-2".to_owned(); + changed_expected_identity.expected_identity = Some(expected); + + for (label, changed) in [ + ("run", changed_run), + ("session", changed_session), + ("revision", changed_revision), + ("catalog", changed_catalog), + ("model", changed_model), + ("permission", changed_permission), + ("expected identity", changed_expected_identity), + ] { + let error = checkpoint.admit_recovery(&changed).unwrap_err().to_string(); + assert!(error.contains("conflict"), "{label}: {error}"); + } + + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn rejects_identity_drift_before_a_checkpoint_is_created() { + let directory = temporary_directory("identity-drift"); + let config = config(&directory); + let mut mismatched = identity(); + mismatched.effective_model = "other-model".to_owned(); + let error = AcpxSuspensionCheckpoint::from_suspension(&config, mismatched) + .unwrap_err() + .to_string(); + assert!(error.contains("conflicts"), "{error}"); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn fails_closed_on_unknown_or_oversized_checkpoint_files() { + let directory = temporary_directory("malformed"); + let config = config(&directory); + let checkpoint = AcpxSuspensionCheckpoint::from_suspension(&config, identity()).unwrap(); + let store = AcpxSuspensionCheckpointStore::new(&directory).unwrap(); + store.save(&checkpoint).unwrap(); + + let valid: serde_json::Value = + serde_json::from_slice(&fs::read(store.path()).unwrap()).unwrap(); + let mut top_level_unknown = valid.clone(); + top_level_unknown["unexpected"] = json!(true); + let mut nested_unknown = valid.clone(); + nested_unknown["identity"]["unexpected"] = json!(true); + let mut missing_permission = valid.clone(); + missing_permission["identity"] + .as_object_mut() + .unwrap() + .remove("permissionMode"); + let mut invalid_run = valid.clone(); + invalid_run["runId"] = json!("run 1"); + let mut invalid_session = valid; + invalid_session["normalizedSessionId"] = json!("séssion-1"); + + for malformed in [ + top_level_unknown, + nested_unknown, + missing_permission, + invalid_run, + invalid_session, + ] { + fs::write(store.path(), serde_json::to_vec(&malformed).unwrap()).unwrap(); + assert!(store.load().is_err()); + } + + fs::write(store.path(), vec![b'x'; 1024 * 1024 + 1]).unwrap(); + assert!(store.load().unwrap_err().to_string().contains("1 MiB")); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +fn rejects_a_symlinked_runner_state_directory() { + use std::os::unix::fs::symlink; + + let directory = temporary_directory("symlink"); + let actual = directory.join("actual"); + fs::create_dir(&actual).unwrap(); + let linked = directory.join("linked"); + symlink(&actual, &linked).unwrap(); + let error = AcpxSuspensionCheckpointStore::new(&linked) + .unwrap_err() + .to_string(); + assert!(error.contains("must not be a symlink"), "{error}"); + fs::remove_dir_all(directory).unwrap(); +} + +#[cfg(unix)] +#[test] +fn rejects_an_existing_runner_state_directory_that_is_not_private() { + let directory = temporary_directory("public-state"); + fs::set_permissions(&directory, fs::Permissions::from_mode(0o755)).unwrap(); + + let error = AcpxSuspensionCheckpointStore::new(&directory) + .unwrap_err() + .to_string(); + + assert!(error.contains("not private"), "{error}"); + assert_eq!( + fs::metadata(&directory).unwrap().permissions().mode() & 0o077, + 0o055 + ); + fs::remove_dir_all(directory).unwrap(); +}