diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 3115f399a8..71604a5e3f 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -89,6 +89,11 @@ correlations. Terminal events flush the final assistant message first and clear unresolved turn-scoped requests. This reducer still does not select ACPX in runnerd. +The package-local session bootstrap starts the bounded sidecar transport, +verifies the Codex-only capability handshake and effective model, opens one +identity-bound session, and confirms its run attachment. Any failed bootstrap +terminates the process; session shutdown preserves persistent provider state. + Run the complete contract gate with: ```sh 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 new file mode 100644 index 0000000000..e544626feb --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -0,0 +1,398 @@ +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::acpx_provider_state::AcpxProviderState; +use crate::acpx_sidecar_transport::{AcpxSidecarTransport, AcpxSidecarTransportConfig}; +use crate::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, +}; +use crate::local_runner::LocalRunnerError; +use crate::provider_bridge::{AuthorizedToolSet, ProviderToolBridge}; + +const MAX_ID_CHARS: usize = 240; +const MAX_MODEL_CHARS: usize = 240; +const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024; +const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991; + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "kebab-case")] +pub enum AcpxPermissionMode { + ApproveAll, + ApproveReads, + DenyAll, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct AcpxProviderSessionIdentity { + pub kind: String, + pub normalized_session_id: String, + pub acpx_record_id: String, + pub backend_session_id: String, + pub agent_session_id: String, + pub profile_digest: String, + pub workspace_digest: String, + pub requested_model: String, + pub effective_model: String, + #[serde(default)] + pub permission_mode: Option, +} + +#[derive(Clone, Debug)] +pub struct AcpxProviderSessionConfig { + pub transport: AcpxSidecarTransportConfig, + pub agent: String, + pub model: String, + pub run_id: String, + pub catalog_revision: u64, + pub runtime_directory: PathBuf, + pub normalized_session_id: String, + pub working_directory: PathBuf, + pub permission_mode: AcpxPermissionMode, + pub permission_mode_pinned: bool, + pub system_instructions: String, + pub tool_set: AuthorizedToolSet, + pub expected_identity: Option, +} + +impl AcpxProviderSessionConfig { + pub fn validate(&self) -> Result<(), LocalRunnerError> { + self.transport.validate()?; + if self.agent != "codex" { + return Err(LocalRunnerError::invalid( + "the initial ACPX provider session supports Codex only", + )); + } + validate_text(&self.model, MAX_MODEL_CHARS, "ACPX model")?; + validate_text(&self.run_id, 160, "ACPX run id")?; + validate_text( + &self.normalized_session_id, + 160, + "ACPX normalized session id", + )?; + if self.catalog_revision == 0 || self.catalog_revision > MAX_JSON_SAFE_INTEGER { + return Err(LocalRunnerError::invalid( + "ACPX catalog revision must be a positive JSON-safe integer", + )); + } + for (path, label) in [ + (&self.runtime_directory, "runtime directory"), + (&self.working_directory, "working directory"), + ] { + if !path.is_absolute() { + return Err(LocalRunnerError::invalid(format!( + "ACPX {label} must be an existing absolute directory" + ))); + } + if path.to_str().is_none() { + return Err(LocalRunnerError::invalid(format!( + "ACPX {label} must be valid UTF-8" + ))); + } + if !path.is_dir() { + return Err(LocalRunnerError::invalid(format!( + "ACPX {label} must be an existing absolute directory" + ))); + } + } + if !self.permission_mode_pinned { + return Err(LocalRunnerError::invalid( + "ACPX permission mode must be pinned by the runner policy", + )); + } + if self.system_instructions.len() > MAX_SYSTEM_INSTRUCTIONS_BYTES + || self.system_instructions.contains('\0') + { + return Err(LocalRunnerError::invalid( + "ACPX system instructions exceed their bounded contract", + )); + } + let mut bridge = ProviderToolBridge::default(); + bridge.prepare(self.tool_set.clone()).map_err(|error| { + LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}")) + })?; + if let Some(expected_identity) = self.expected_identity.as_ref() { + expected_identity.validate()?; + if expected_identity.normalized_session_id != self.normalized_session_id + || expected_identity.requested_model != self.model + || expected_identity.effective_model != self.model + || expected_identity.permission_mode != Some(self.permission_mode) + { + return Err(LocalRunnerError::invalid( + "ACPX expected identity conflicts with the requested session", + )); + } + } + Ok(()) + } +} + +impl AcpxProviderSessionIdentity { + pub fn validate(&self) -> Result<(), LocalRunnerError> { + if self.kind != "acpx" { + return Err(LocalRunnerError::invalid( + "ACPX session identity kind is invalid", + )); + } + for (value, label) in [ + (&self.normalized_session_id, "normalized session"), + (&self.acpx_record_id, "record"), + (&self.backend_session_id, "backend session"), + (&self.agent_session_id, "agent session"), + (&self.requested_model, "requested model"), + (&self.effective_model, "effective model"), + ] { + validate_text(value, MAX_ID_CHARS, &format!("ACPX {label} identity"))?; + } + for (value, label) in [ + (&self.profile_digest, "profile"), + (&self.workspace_digest, "workspace"), + ] { + if !is_sha256_digest(value) { + return Err(LocalRunnerError::invalid(format!( + "ACPX {label} digest is invalid" + ))); + } + } + Ok(()) + } +} + +pub struct AcpxProviderSession { + transport: AcpxSidecarTransport, + state: AcpxProviderState, + identity: AcpxProviderSessionIdentity, + catalog_revision: u64, + closed: bool, + transport_terminated: bool, +} + +impl AcpxProviderSession { + pub fn start(config: &AcpxProviderSessionConfig) -> Result { + config.validate()?; + let mut transport = AcpxSidecarTransport::start(&config.transport)?; + let bootstrap = bootstrap(&mut transport, config); + let (identity, state) = match bootstrap { + Ok(value) => value, + Err(error) => { + let cleanup = transport.shutdown(); + return Err(with_cleanup_error(error, cleanup)); + } + }; + Ok(Self { + transport, + state, + identity, + catalog_revision: config.catalog_revision, + closed: false, + transport_terminated: false, + }) + } + + pub fn process_id(&self) -> u32 { + self.transport.process_id() + } + + pub fn identity(&self) -> &AcpxProviderSessionIdentity { + &self.identity + } + + pub fn state(&self) -> &AcpxProviderState { + &self.state + } + + pub fn catalog_revision(&self) -> u64 { + self.catalog_revision + } + + pub fn shutdown(&mut self, reason: &str) -> Result<(), LocalRunnerError> { + if self.closed { + return self.terminate_transport(); + } + self.closed = true; + let close = self.transport.request( + GeneratedAcpxSidecarCommand::SessionClose, + json!({ + "reason": bounded_reason(reason), + "discardPersistentState": false, + }), + ); + let terminate = self.terminate_transport(); + match (close, terminate) { + (Ok(_), Ok(())) => Ok(()), + (Err(error), cleanup) => Err(with_cleanup_error(error, cleanup)), + (Ok(_), Err(error)) => Err(error), + } + } + + fn terminate_transport(&mut self) -> Result<(), LocalRunnerError> { + if self.transport_terminated { + return Ok(()); + } + self.transport.shutdown()?; + self.transport_terminated = true; + Ok(()) + } +} + +impl Drop for AcpxProviderSession { + fn drop(&mut self) { + if !self.transport_terminated { + self.closed = true; + let _ = self.terminate_transport(); + } + } +} + +fn bootstrap( + transport: &mut AcpxSidecarTransport, + config: &AcpxProviderSessionConfig, +) -> Result<(AcpxProviderSessionIdentity, AcpxProviderState), LocalRunnerError> { + let initialized = transport.request( + GeneratedAcpxSidecarCommand::Initialize, + json!({"agent": config.agent, "model": config.model}), + )?; + verify_initialize_response(&initialized, transport.process_id())?; + + let opened = transport.request( + GeneratedAcpxSidecarCommand::SessionOpen, + json!({ + "runtimeDirectory": config.runtime_directory, + "normalizedSessionId": config.normalized_session_id, + "workingDirectory": config.working_directory, + "agent": config.agent, + "model": config.model, + "permissionMode": config.permission_mode, + "permissionModePinned": config.permission_mode_pinned, + "systemInstructions": config.system_instructions, + "runtimeContext": Value::Null, + "tools": config.tool_set.operations, + "expectedIdentity": config.expected_identity, + }), + )?; + let identity = verify_open_response(&opened, transport.process_id(), config)?; + + let attached = transport.request( + GeneratedAcpxSidecarCommand::RunAttach, + json!({ + "runId": config.run_id, + "catalogRevision": config.catalog_revision, + "tools": config.tool_set.operations, + }), + )?; + if attached.get("runId").and_then(Value::as_str) != Some(config.run_id.as_str()) + || attached.get("catalogRevision").and_then(Value::as_u64) != Some(config.catalog_revision) + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar did not confirm the requested run attachment", + )); + } + Ok((identity, AcpxProviderState::new(&config.run_id)?)) +} + +fn verify_initialize_response(value: &Value, process_id: u32) -> Result<(), LocalRunnerError> { + if value.get("protocolVersion").and_then(Value::as_u64) + != Some(GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION) + || value.get("sidecarPid").and_then(Value::as_u64) != Some(u64::from(process_id)) + || !value.get("profile").is_some_and(Value::is_object) + || value + .pointer("/capabilities/persistentSessions") + .and_then(Value::as_bool) + != Some(true) + || value + .pointer("/capabilities/exactModelVerification") + .and_then(Value::as_bool) + != Some(true) + || value + .pointer("/capabilities/permissions") + .and_then(Value::as_str) + != Some("runner_policy") + || value + .pointer("/capabilities/semanticTools") + .and_then(Value::as_str) + != Some("runner_bridge") + || value + .pointer("/capabilities/structuredInput") + .and_then(Value::as_str) + != Some("paperclip.question_set.v1") + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar initialization capabilities are invalid", + )); + } + Ok(()) +} + +fn verify_open_response( + value: &Value, + process_id: u32, + config: &AcpxProviderSessionConfig, +) -> Result { + if value.get("sidecarPid").and_then(Value::as_u64) != Some(u64::from(process_id)) + || !value.get("status").is_some_and(Value::is_object) + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar session-open response is invalid", + )); + } + let identity: AcpxProviderSessionIdentity = serde_json::from_value( + value + .get("identity") + .cloned() + .ok_or_else(|| LocalRunnerError::invalid("ACPX sidecar omitted its identity"))?, + ) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX sidecar identity is invalid: {error}")) + })?; + 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 sidecar identity does not match the requested session", + )); + } + Ok(identity) +} + +fn validate_text(value: &str, max_chars: usize, label: &str) -> Result<(), LocalRunnerError> { + if value.trim().is_empty() + || value.chars().count() > max_chars + || value.chars().any(char::is_control) + { + return Err(LocalRunnerError::invalid(format!("{label} 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()) +} + +fn bounded_reason(value: &str) -> String { + value.chars().take(4_000).collect() +} + +fn with_cleanup_error( + error: LocalRunnerError, + cleanup: Result<(), LocalRunnerError>, +) -> LocalRunnerError { + match cleanup { + Ok(()) => error, + Err(cleanup) => LocalRunnerError::invalid(format!( + "{error}; ACPX sidecar cleanup also failed: {cleanup}" + )), + } +} 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 3a9e5c72e6..8aa20a1922 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 @@ -79,6 +79,9 @@ fn run() -> Result<(), Box> { eprintln!("amber-signal-7305"); std::process::exit(9); } + "bootstrap" | "bootstrap-wrong-model" | "bootstrap-wrong-run" => { + write_json(&mut stdout, &bootstrap_success(id, command, &request, mode))?; + } "happy" => { write_event(&mut stdout, next_sequence)?; next_sequence += 1; @@ -90,6 +93,58 @@ fn run() -> Result<(), Box> { Ok(()) } +fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Value { + let params = request.get("params").cloned().unwrap_or_else(|| json!({})); + let result = match command { + "initialize" => json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sidecarPid": std::process::id(), + "profile": {"agent":"codex"}, + "capabilities": { + "persistentSessions": true, + "exactModelVerification": true, + "permissions": "runner_policy", + "semanticTools": "runner_bridge", + "structuredInput": "paperclip.question_set.v1", + }, + }), + "session.open" => { + let model = params + .get("model") + .and_then(Value::as_str) + .unwrap_or("missing"); + json!({ + "sidecarPid": std::process::id(), + "identity": { + "kind": "acpx", + "normalizedSessionId": params.get("normalizedSessionId"), + "acpxRecordId": "record-1", + "backendSessionId": "backend-1", + "agentSessionId": "agent-1", + "profileDigest": format!("sha256:{}", "1".repeat(64)), + "workspaceDigest": format!("sha256:{}", "2".repeat(64)), + "requestedModel": model, + "effectiveModel": if mode == "bootstrap-wrong-model" { "wrong-model" } else { model }, + "permissionMode": params.get("permissionMode"), + }, + "status": {}, + }) + } + "run.attach" => json!({ + "runId": if mode == "bootstrap-wrong-run" { "wrong-run" } else { params.get("runId").and_then(Value::as_str).unwrap_or("missing") }, + "catalogRevision": params.get("catalogRevision"), + }), + "session.close" => json!({"closed":true}), + _ => json!({"command":command,"params":params}), + }; + json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": id, + "ok": true, + "result": result, + }) +} + fn success(id: u64, command: &str, request: &Value) -> Value { json!({ "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, 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 e4f2893055..48dbd09ad8 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_session; pub mod acpx_provider_state; pub mod acpx_sidecar_transport; pub mod codex_provider; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs new file mode 100644 index 0000000000..e9067210d7 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs @@ -0,0 +1,150 @@ +use std::path::PathBuf; +use std::time::Duration; + +use paperclip_runner_core::acpx_provider_session::{ + AcpxPermissionMode, AcpxProviderSession, 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 tool_set() -> AuthorizedToolSet { + 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"}), + }]; + AuthorizedToolSet { + schema: "paperclip.runner.authorized-tools.v1".to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + } +} + +fn config(mode: &str) -> AcpxProviderSessionConfig { + AcpxProviderSessionConfig { + transport: AcpxSidecarTransportConfig { + command: PathBuf::from(env!("CARGO_BIN_EXE_fake-acpx-sidecar")), + args: vec!["--mode".to_owned(), mode.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: 1, + runtime_directory: std::env::temp_dir(), + normalized_session_id: "session-1".to_owned(), + working_directory: std::env::temp_dir(), + permission_mode: AcpxPermissionMode::ApproveReads, + permission_mode_pinned: true, + system_instructions: "Complete the supplied task.".to_owned(), + tool_set: tool_set(), + expected_identity: None, + } +} + +fn expected_identity() -> AcpxProviderSessionIdentity { + AcpxProviderSessionIdentity { + kind: "acpx".to_owned(), + normalized_session_id: "session-1".to_owned(), + acpx_record_id: "record-1".to_owned(), + backend_session_id: "backend-1".to_owned(), + agent_session_id: "agent-1".to_owned(), + profile_digest: format!("sha256:{}", "1".repeat(64)), + workspace_digest: format!("sha256:{}", "2".repeat(64)), + requested_model: "gpt-5.6-sol".to_owned(), + effective_model: "gpt-5.6-sol".to_owned(), + permission_mode: Some(AcpxPermissionMode::ApproveReads), + } +} + +fn start_error(config: &AcpxProviderSessionConfig) -> String { + match AcpxProviderSession::start(config) { + Ok(mut session) => { + let _ = session.shutdown("unexpected successful bootstrap"); + panic!("ACPX provider session unexpectedly started") + } + Err(error) => error.to_string(), + } +} + +#[test] +fn bootstraps_a_codex_session_and_confirms_run_identity() { + let mut session = AcpxProviderSession::start(&config("bootstrap")).unwrap(); + assert!(session.process_id() > 0); + assert_eq!(session.identity(), &expected_identity()); + assert_eq!(session.state().run_id(), "run-1"); + assert_eq!(session.state().active_turn_id(), None); + assert_eq!(session.catalog_revision(), 1); + session.shutdown("test complete").unwrap(); + session.shutdown("already closed").unwrap(); +} + +#[test] +fn validates_codex_policy_and_tool_catalog_before_spawning() { + let mut invalid_agent = config("bootstrap"); + invalid_agent.agent = "opencode".to_owned(); + assert!(start_error(&invalid_agent).contains("Codex only")); + + let mut unpinned = config("bootstrap"); + unpinned.permission_mode_pinned = false; + assert!(start_error(&unpinned).contains("must be pinned")); + + let mut invalid_tools = config("bootstrap"); + invalid_tools.tool_set.catalog_digest = "invalid".to_owned(); + assert!(start_error(&invalid_tools).contains("authorized tools")); +} + +#[test] +fn rejects_a_sidecar_that_reports_another_effective_model() { + let error = start_error(&config("bootstrap-wrong-model")); + assert!(error.contains("identity does not match"), "{error}"); +} + +#[test] +fn rejects_a_sidecar_that_does_not_confirm_the_run_attachment() { + let error = start_error(&config("bootstrap-wrong-run")); + assert!(error.contains("run attachment"), "{error}"); +} + +#[test] +fn validates_recovery_identity_against_the_requested_session() { + let mut recovered = config("bootstrap"); + recovered.expected_identity = Some(expected_identity()); + let mut session = AcpxProviderSession::start(&recovered).unwrap(); + assert_eq!( + session.identity(), + recovered.expected_identity.as_ref().unwrap() + ); + session.shutdown("test complete").unwrap(); + + let mut mismatch = config("bootstrap"); + let mut expected = expected_identity(); + expected.normalized_session_id = "another-session".to_owned(); + mismatch.expected_identity = Some(expected); + assert!(start_error(&mismatch).contains("conflicts with the requested session")); +} + +#[cfg(unix)] +#[test] +fn rejects_non_utf8_directories_before_spawning() { + use std::ffi::OsString; + use std::os::unix::ffi::OsStringExt; + + let mut directory_name = + format!("paperclip-acpx-non-utf8-{}-", std::process::id()).into_bytes(); + directory_name.push(0xff); + let directory = std::env::temp_dir().join(OsString::from_vec(directory_name)); + + let mut invalid = config("bootstrap"); + invalid.working_directory = directory; + let error = start_error(&invalid); + + assert!(error.contains("must be valid UTF-8"), "{error}"); +}