diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 447a4892ba..e535d248bd 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -114,6 +114,9 @@ closed. Safe suspension is available only with no active turn or pending request. The sidecar must return the exact persistent session identity before runnerd 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. Run the complete contract gate with: diff --git a/packages/paperclip-runner/protocol/manifest.json b/packages/paperclip-runner/protocol/manifest.json index 64caa6a3e1..aded16edae 100644 --- a/packages/paperclip-runner/protocol/manifest.json +++ b/packages/paperclip-runner/protocol/manifest.json @@ -30,7 +30,7 @@ { "path": "schemas/event.schema.json", "id": "https://paperclip.dev/schemas/prp/v1/event.schema.json", - "sha256": "e9827a7c31f004ae91388964a96b4ddf6ab40b85f29bf3922683d5dcf64f2559" + "sha256": "d48916630976ff7e850b6e2d7d94131761fda322f5d8a05212651ec46b0db19d" }, { "path": "schemas/fixture.schema.json", @@ -70,7 +70,7 @@ { "path": "schemas/request.schema.json", "id": "https://paperclip.dev/schemas/prp/v1/request.schema.json", - "sha256": "bd41f01c00eddc1b635d3131a7183ddb616ff8fe68d5e348ddd9593a674c8d8e" + "sha256": "c8d00e10150c90963f19641c075f50da7f67617dd817f41b356287ad85c53f2b" }, { "path": "schemas/result.schema.json", diff --git a/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json b/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json index dbbbb53a12..71058b5d7d 100644 --- a/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json +++ b/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json @@ -100,8 +100,8 @@ "protocolVersion": { "$ref": "#/$defs/protocolVersion" }, "sequence": { "type": "integer", "minimum": 1 }, "eventType": { "$ref": "#/$defs/eventType" }, - "runId": { "type": ["string", "null"], "maxLength": 160 }, - "turnId": { "type": ["string", "null"], "maxLength": 160 }, + "runId": { "type": ["string", "null"], "maxLength": 160, "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" }, + "turnId": { "type": ["string", "null"], "maxLength": 240, "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" }, "payload": { "type": "object" } }, "additionalProperties": false diff --git a/packages/paperclip-runner/protocol/schemas/event.schema.json b/packages/paperclip-runner/protocol/schemas/event.schema.json index 9d2b3ed873..5beda54447 100644 --- a/packages/paperclip-runner/protocol/schemas/event.schema.json +++ b/packages/paperclip-runner/protocol/schemas/event.schema.json @@ -24,8 +24,8 @@ "sourceKind": { "enum": ["runner", "control_plane"] }, "runId": { "type": "string", "minLength": 1, "maxLength": 160 }, "normalizedSessionId": { "type": "string", "minLength": 1, "maxLength": 160 }, - "turnId": { "type": "string", "minLength": 1, "maxLength": 160 }, - "itemId": { "type": "string", "minLength": 1, "maxLength": 160 }, + "turnId": { "type": "string", "minLength": 1, "maxLength": 240 }, + "itemId": { "type": "string", "minLength": 1, "maxLength": 240 }, "eventType": { "enum": [ "runner.connected", "runner.reconnected", "runner.reconciled", "runner.disconnected", "runner.draining", "runner.backpressure", "runner.suspending", "runner.suspended", "runner.stopped", "runner.diagnostic", diff --git a/packages/paperclip-runner/protocol/schemas/request.schema.json b/packages/paperclip-runner/protocol/schemas/request.schema.json index 78441bc7bf..83a741ef44 100644 --- a/packages/paperclip-runner/protocol/schemas/request.schema.json +++ b/packages/paperclip-runner/protocol/schemas/request.schema.json @@ -24,8 +24,8 @@ }, "additionalProperties": false }, - "turnId": { "type": "string", "minLength": 1, "maxLength": 160 }, - "itemId": { "type": "string", "minLength": 1, "maxLength": 160 } + "turnId": { "type": "string", "minLength": 1, "maxLength": 240 }, + "itemId": { "type": "string", "minLength": 1, "maxLength": 240 } }, "additionalProperties": false }, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_payload.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_payload.rs index d78d2f168a..5278fc4db7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_payload.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_payload.rs @@ -13,6 +13,7 @@ use crate::provider_bridge::semantic_value_digest; const MAX_EVENT_PAYLOAD_BYTES: usize = 256 * 1024; const MAX_ID_CHARS: usize = 160; +const MAX_INPUT_REQUEST_ID_CHARS: usize = 240; const MAX_RUNTIME_TEXT_CHARS: usize = 64 * 1024; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -111,7 +112,12 @@ pub fn decode_acpx_event( let question_set = sanitize_question_set(question_set); let origin = optional_object(&event.payload, "origin", "input request origin")?; Ok(AcpxEventPayload::InputRequested { - request_id: required_id(&event.payload, "requestId", "input request")?, + request_id: required_id_with_limit( + &event.payload, + "requestId", + "input request", + MAX_INPUT_REQUEST_ID_CHARS, + )?, question_set, origin: origin.map(|value| sanitize_value(&value)), }) @@ -405,11 +411,20 @@ fn validate_question_set(value: &Value) -> Result<(), LocalRunnerError> { } fn required_id(value: &Value, key: &str, label: &str) -> Result { + required_id_with_limit(value, key, label, MAX_ID_CHARS) +} + +fn required_id_with_limit( + value: &Value, + key: &str, + label: &str, + max_chars: usize, +) -> Result { let id = value .get(key) .and_then(Value::as_str) .ok_or_else(|| LocalRunnerError::invalid(format!("ACPX {label} omitted its identity")))?; - if id.is_empty() || id.chars().count() > MAX_ID_CHARS || id.chars().any(char::is_control) { + if id.is_empty() || id.chars().count() > max_chars || id.chars().any(char::is_control) { return Err(LocalRunnerError::invalid(format!( "ACPX {label} identity is invalid" ))); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs index 523af67e59..1213f0899b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs @@ -3,8 +3,8 @@ use std::collections::BTreeSet; use crate::acpx_sidecar_transport::AcpxSidecarEvent; use crate::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType; use crate::local_runner::LocalRunnerError; +use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; -const MAX_SCOPE_ID_CHARS: usize = 160; const MAX_SETTLED_TURN_IDS: usize = 4_096; /// Holds the run and turn authority used to admit ACPX sidecar events. @@ -22,7 +22,7 @@ pub struct AcpxEventScope { impl AcpxEventScope { pub fn new(run_id: impl Into) -> Result { let run_id = run_id.into(); - validate_scope_id(&run_id, "run")?; + validate_scope_id(&run_id, "run", SHORT_STABLE_ID_CHARS)?; Ok(Self { run_id, active_turn_id: None, @@ -44,7 +44,7 @@ impl AcpxEventScope { pub fn bind_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { let turn_id = turn_id.into(); - validate_scope_id(&turn_id, "turn")?; + validate_scope_id(&turn_id, "turn", DURABLE_STABLE_ID_CHARS)?; match self.active_turn_id.as_deref() { Some(active_turn_id) if active_turn_id == turn_id.as_str() => Ok(()), Some(_) => Err(LocalRunnerError::invalid( @@ -72,7 +72,7 @@ impl AcpxEventScope { &self, turn_id: &str, ) -> Result<(), LocalRunnerError> { - validate_scope_id(turn_id, "turn")?; + validate_scope_id(turn_id, "turn", DURABLE_STABLE_ID_CHARS)?; if self.settled_turn_ids.contains(turn_id) { return Err(LocalRunnerError::invalid( "ACPX event scope reused a settled turn identity", @@ -103,7 +103,7 @@ impl AcpxEventScope { } pub fn clear_turn(&mut self, turn_id: &str) -> Result<(), LocalRunnerError> { - validate_scope_id(turn_id, "turn")?; + validate_scope_id(turn_id, "turn", DURABLE_STABLE_ID_CHARS)?; if self.active_turn_id.as_deref() != Some(turn_id) { return Err(LocalRunnerError::invalid( "ACPX event scope cannot clear a stale turn", @@ -125,10 +125,10 @@ impl AcpxEventScope { pub fn validate_event(&self, event: &AcpxSidecarEvent) -> Result<(), LocalRunnerError> { if let Some(run_id) = event.run_id.as_deref() { - validate_scope_id(run_id, "event run")?; + validate_scope_id(run_id, "event run", SHORT_STABLE_ID_CHARS)?; } if let Some(turn_id) = event.turn_id.as_deref() { - validate_scope_id(turn_id, "event turn")?; + validate_scope_id(turn_id, "event turn", DURABLE_STABLE_ID_CHARS)?; } let global_event = matches!( event.event_type, @@ -203,11 +203,8 @@ mod tests { } } -fn validate_scope_id(value: &str, label: &str) -> Result<(), LocalRunnerError> { - if value.is_empty() - || value.chars().count() > MAX_SCOPE_ID_CHARS - || value.chars().any(char::is_control) - { +fn validate_scope_id(value: &str, label: &str, max_chars: usize) -> Result<(), LocalRunnerError> { + if !is_stable_id(value, max_chars) { return Err(LocalRunnerError::invalid(format!( "ACPX event scope {label} id is invalid" ))); 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 486a7121ec..359b921f6f 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 @@ -4,7 +4,10 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use crate::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent}; +use crate::acpx_provider_state::{ + is_reserved_terminal_operation, AcpxProviderState, AcpxProviderStateEvent, PRP_BLOCK_TOOL_NAME, + PRP_COMPLETION_TOOL_NAME, +}; use crate::acpx_sidecar_transport::{AcpxSidecarTransport, AcpxSidecarTransportConfig}; use crate::generated_acpx_sidecar_contract::{ GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, @@ -15,13 +18,12 @@ use crate::provider_bridge::{ ToolResult, TOOL_SET_SCHEMA, }; use crate::question_response::validate_question_response; +use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; 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; -const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish"; -const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block"; #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] #[serde(rename_all = "kebab-case")] @@ -73,10 +75,10 @@ impl AcpxProviderSessionConfig { )); } validate_text(&self.model, MAX_MODEL_CHARS, "ACPX model")?; - validate_text(&self.run_id, 160, "ACPX run id")?; - validate_text( + validate_stable_id(&self.run_id, SHORT_STABLE_ID_CHARS, "ACPX run id")?; + validate_stable_id( &self.normalized_session_id, - 160, + SHORT_STABLE_ID_CHARS, "ACPX normalized session id", )?; if self.catalog_revision == 0 || self.catalog_revision > MAX_JSON_SAFE_INTEGER { @@ -247,7 +249,7 @@ impl AcpxProviderSession { working_directory: &Path, ) -> Result { self.ensure_open()?; - validate_text(turn_id, 160, "ACPX turn id")?; + validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?; validate_turn_message(message)?; if working_directory != self.working_directory { return Err(LocalRunnerError::invalid( @@ -342,7 +344,7 @@ impl AcpxProviderSession { reason: &str, ) -> Result { self.ensure_open()?; - validate_text(turn_id, 160, "ACPX turn id")?; + validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?; if self.state.active_turn_id() != Some(turn_id) { return Err(LocalRunnerError::invalid( "ACPX interruption named a stale or inactive turn", @@ -572,7 +574,17 @@ impl AcpxProviderSession { resolution: &Value, ) -> Result<(), LocalRunnerError> { self.ensure_bound_turn(turn_id)?; - validate_text(request_id, 240, "ACPX input request id")?; + validate_text(request_id, SHORT_STABLE_ID_CHARS, "ACPX input request id")?; + if !is_stable_id(request_id, SHORT_STABLE_ID_CHARS) { + return Err(LocalRunnerError::invalid( + "ACPX input request id is not a stable runtime request identity", + )); + } + let provider_request_id = self + .state + .pending_provider_input_request_id(request_id) + .ok_or_else(|| LocalRunnerError::invalid("ACPX input request is stale or unknown"))? + .to_owned(); let question_set = self .state .pending_question_set(request_id) @@ -582,7 +594,7 @@ impl AcpxProviderSession { next_state.complete_input(request_id)?; let response = match self.transport.request( GeneratedAcpxSidecarCommand::InputResolve, - json!({"requestId":request_id,"turnId":turn_id,"resolution":resolution}), + json!({"requestId":provider_request_id,"turnId":turn_id,"resolution":resolution}), ) { Ok(response) => response, Err(error) => return Err(self.fail_closed(error)), @@ -727,7 +739,7 @@ impl AcpxProviderSession { } fn ensure_bound_turn(&self, turn_id: &str) -> Result<(), LocalRunnerError> { - validate_text(turn_id, 160, "ACPX turn id")?; + validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?; if self.ensure_active_turn()? != turn_id { return Err(LocalRunnerError::invalid( "ACPX resolution named a stale or inactive turn", @@ -755,10 +767,6 @@ fn is_reserved_terminal_result(result: &crate::acpx_provider_state::AcpxSemantic is_reserved_terminal_operation(&result.operation_id) } -fn is_reserved_terminal_operation(operation_id: &str) -> bool { - matches!(operation_id, PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME) -} - fn validate_reserved_terminal_result( state: &AcpxProviderState, result: &crate::acpx_provider_state::AcpxSemanticResult, @@ -1019,6 +1027,13 @@ fn validate_text(value: &str, max_chars: usize, label: &str) -> Result<(), Local Ok(()) } +fn validate_stable_id(value: &str, max_chars: usize, label: &str) -> Result<(), LocalRunnerError> { + if !is_stable_id(value, max_chars) { + return Err(LocalRunnerError::invalid(format!("{label} is invalid"))); + } + Ok(()) +} + fn is_sha256_digest(value: &str) -> bool { value.len() == 71 && value.starts_with("sha256:") diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs index 8ad2058d7f..948dcff4e0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -10,14 +10,19 @@ use crate::acpx_sidecar_transport::AcpxSidecarEvent; use crate::local_runner::LocalRunnerError; use crate::provider_bridge::ToolResult; use crate::provider_events::{normalize_acpx_runtime_event, NormalizedProviderEvent}; +use crate::stable_identity::project_acpx_runtime_request_id; const MAX_ASSISTANT_TEXT_BYTES: usize = 1024 * 1024; const MAX_PENDING_TOOLS: usize = 4_096; const MAX_PENDING_TOOL_INPUT_BYTES: usize = 16 * 1024 * 1024; const MAX_PENDING_RUNTIME_REQUESTS: usize = 1_024; const MAX_PENDING_RUNTIME_REQUEST_BYTES: usize = 16 * 1024 * 1024; -const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish"; -const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block"; +pub(crate) const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish"; +pub(crate) const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block"; + +pub(crate) fn is_reserved_terminal_operation(operation_id: &str) -> bool { + matches!(operation_id, PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME) +} #[derive(Clone, Debug, PartialEq)] pub struct AcpxPendingTool { @@ -75,6 +80,7 @@ pub enum AcpxProviderStateEvent { #[derive(Clone, Debug, PartialEq)] struct PendingInput { + runtime_request_id: String, value_bytes: usize, question_set: Value, } @@ -204,6 +210,15 @@ impl AcpxProviderState { } => { let value_bytes = value_bytes(&details)?; self.admit_runtime_request(&request_id, value_bytes)?; + if self + .pending_inputs + .values() + .any(|pending| pending.runtime_request_id == request_id) + { + return Err(LocalRunnerError::invalid( + "ACPX permission request identity collides with a pending input projection", + )); + } self.pending_permissions .insert(request_id.clone(), value_bytes); self.pending_runtime_request_bytes += value_bytes; @@ -221,11 +236,24 @@ impl AcpxProviderState { } => { let value_bytes = value_bytes(&question_set)?; self.admit_runtime_request(&request_id, value_bytes)?; + let runtime_request_id = project_acpx_runtime_request_id(&request_id) + .expect("a decoded ACPX input request has a bounded identity"); + if self + .pending_inputs + .values() + .any(|pending| pending.runtime_request_id == runtime_request_id) + || self.pending_permissions.contains_key(&runtime_request_id) + { + return Err(LocalRunnerError::invalid( + "ACPX input request identity collides after durable projection", + )); + } if self .pending_inputs .insert( request_id.clone(), PendingInput { + runtime_request_id, value_bytes, question_set: question_set.clone(), }, @@ -354,9 +382,16 @@ impl AcpxProviderState { } pub fn complete_input(&mut self, request_id: &str) -> Result<(), LocalRunnerError> { + let provider_request_id = self + .pending_inputs + .iter() + .find_map(|(provider_request_id, pending)| { + (pending.runtime_request_id == request_id).then(|| provider_request_id.clone()) + }) + .ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?; let pending = self .pending_inputs - .remove(request_id) + .remove(&provider_request_id) .ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?; self.pending_runtime_request_bytes = self .pending_runtime_request_bytes @@ -366,10 +401,18 @@ impl AcpxProviderState { pub fn pending_question_set(&self, request_id: &str) -> Option<&Value> { self.pending_inputs - .get(request_id) + .values() + .find(|pending| pending.runtime_request_id == request_id) .map(|pending| &pending.question_set) } + pub(crate) fn pending_provider_input_request_id(&self, request_id: &str) -> Option<&str> { + self.pending_inputs + .iter() + .find(|(_, pending)| pending.runtime_request_id == request_id) + .map(|(provider_request_id, _)| provider_request_id.as_str()) + } + pub fn semantic_result(&self) -> Option<&AcpxSemanticResult> { self.semantic_result.as_ref() } @@ -428,10 +471,7 @@ impl AcpxProviderState { result_digest: semantic_result_digest .expect("decoded ACPX semantic result has a raw correlation digest"), }; - if matches!( - result.operation_id.as_str(), - PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME - ) { + if is_reserved_terminal_operation(&result.operation_id) { return match self.semantic_result.as_ref() { None => { self.semantic_result = Some(result.clone()); 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 5a635df5ca..591417df52 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 @@ -12,6 +12,7 @@ use crate::generated_acpx_sidecar_contract::{ }; use crate::local_runner::LocalRunnerError; use crate::process_supervisor::{BoundedLogBuffer, ProcessOutput, SupervisedProcess}; +use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; pub const ACPX_SIDECAR_MAX_FRAME_BYTES: usize = 1024 * 1024; const MAX_BUFFERED_EVENTS: usize = 512; @@ -441,8 +442,8 @@ fn parse_frame(line: &str) -> Result { "ACPX sidecar event sequence is invalid", )); } - let run_id = nullable_identifier(frame.run_id, "event runId")?; - let turn_id = nullable_identifier(frame.turn_id, "event turnId")?; + let run_id = nullable_identifier(frame.run_id, "event runId", SHORT_STABLE_ID_CHARS)?; + let turn_id = nullable_identifier(frame.turn_id, "event turnId", DURABLE_STABLE_ID_CHARS)?; if !frame.payload.is_object() { return Err(LocalRunnerError::invalid( "ACPX sidecar event payload must be an object", @@ -509,7 +510,11 @@ fn parse_frame(line: &str) -> Result { Ok(ParsedFrame::Response(frame)) } -fn nullable_identifier(value: Value, field: &str) -> Result, LocalRunnerError> { +fn nullable_identifier( + value: Value, + field: &str, + max_chars: usize, +) -> Result, LocalRunnerError> { if value.is_null() { return Ok(None); } @@ -518,7 +523,7 @@ fn nullable_identifier(value: Value, field: &str) -> Result, Loca "ACPX sidecar {field} must be a string or null" ))); }; - if value.chars().count() > 160 || value.chars().any(char::is_control) { + if !is_stable_id(value, max_chars) { return Err(LocalRunnerError::invalid(format!( "ACPX sidecar {field} is invalid" ))); 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 f9e7632a32..230dd374bd 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 @@ -4,6 +4,8 @@ use std::time::Duration; use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION; use serde_json::{json, Value}; +const PROJECTED_INPUT_PROVIDER_ID: &str = "input / réquest"; + fn main() { if let Err(error) = run() { eprintln!("fake-acpx-sidecar: {error}"); @@ -110,6 +112,7 @@ fn run() -> Result<(), Box> { | "turns-permission" | "resolutions" | "resolutions-error-redaction" + | "resolutions-projected-id" | "resolutions-wrong-ack" | "suspend" | "suspend-wrong-ack" @@ -454,7 +457,10 @@ fn run() -> Result<(), Box> { if command == "turn.start" && matches!( mode, - "resolutions" | "resolutions-error-redaction" | "resolutions-wrong-ack" + "resolutions" + | "resolutions-error-redaction" + | "resolutions-projected-id" + | "resolutions-wrong-ack" ) { for (event_type, payload) in [ @@ -469,7 +475,11 @@ fn run() -> Result<(), Box> { ( "runtime.input_requested", json!({ - "requestId":"input-1", + "requestId":if mode == "resolutions-projected-id" { + PROJECTED_INPUT_PROVIDER_ID + } else { + "input-1" + }, "questionSet":{ "schema":"paperclip.question_set.v1", "questions":[{ @@ -601,7 +611,11 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val mode != "resolutions-wrong-ack" } }), - "input.resolve" => json!({"resolved":true}), + "input.resolve" => json!({ + "resolved": mode != "resolutions-projected-id" + || params.get("requestId").and_then(Value::as_str) + == Some(PROJECTED_INPUT_PROVIDER_ID), + }), "session.close" => json!({"closed":true}), _ => json!({"command":command,"params":params}), }; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs index 3bbf2f9311..920a75770d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs @@ -7,6 +7,8 @@ use std::fmt::{self, Display, Formatter}; use std::path::PathBuf; use std::time::Duration; +use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; + pub use runner::{run_durable_runner, CommandExecution, CommandExecutor, PolledEvent}; pub(crate) use state::{ create_private_temporary_file, open_private_regular_file, redact_text, sanitize_value, @@ -131,6 +133,22 @@ impl DurableRunnerConfig { ))); } } + for (name, value, max_chars) in [ + ("run_id", self.run_id.as_str(), SHORT_STABLE_ID_CHARS), + ( + "normalized_session_id", + self.normalized_session_id.as_str(), + SHORT_STABLE_ID_CHARS, + ), + ("turn_id", self.turn_id.as_str(), DURABLE_STABLE_ID_CHARS), + ("item_id", self.item_id.as_str(), DURABLE_STABLE_ID_CHARS), + ] { + if !is_stable_id(value, max_chars) { + return Err(DurableRunnerError::invalid(format!( + "{name} must be a stable identity no longer than {max_chars} characters" + ))); + } + } if self.max_outbox_bytes == 0 || self.max_outbox_bytes > MAX_OUTBOX_BYTES || self.p0_reserve_bytes >= self.max_outbox_bytes @@ -162,3 +180,59 @@ impl DurableRunnerConfig { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn config() -> DurableRunnerConfig { + DurableRunnerConfig { + connect_url: "ws://127.0.0.1/runner".to_owned(), + state_dir: PathBuf::from("state"), + runner_instance_id: "runner-1".to_owned(), + environment_lease_id: "lease-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + runner_version: "1.0.0".to_owned(), + runner_digest: "sha256:digest".to_owned(), + max_outbox_bytes: 1024 * 1024, + p0_reserve_bytes: 64 * 1024, + max_frame_bytes: 64 * 1024, + reconnect_delay: Duration::from_millis(1), + max_runtime: Duration::from_secs(60), + } + } + + #[test] + fn validates_durable_event_identity_boundaries() { + let mut boundary = config(); + boundary.run_id = "r".repeat(SHORT_STABLE_ID_CHARS); + boundary.normalized_session_id = "s".repeat(SHORT_STABLE_ID_CHARS); + boundary.turn_id = "t".repeat(DURABLE_STABLE_ID_CHARS); + boundary.item_id = "i".repeat(DURABLE_STABLE_ID_CHARS); + boundary.validate().unwrap(); + + for (field, invalid) in [ + ("run_id", "run 1".to_owned()), + ("normalized_session_id", "session/1".to_owned()), + ("turn_id", "_turn-1".to_owned()), + ("item_id", "itém-1".to_owned()), + ("run_id", "r".repeat(SHORT_STABLE_ID_CHARS + 1)), + ("turn_id", "t".repeat(DURABLE_STABLE_ID_CHARS + 1)), + ] { + let mut invalid_config = config(); + match field { + "run_id" => invalid_config.run_id = invalid, + "normalized_session_id" => invalid_config.normalized_session_id = invalid, + "turn_id" => invalid_config.turn_id = invalid, + "item_id" => invalid_config.item_id = invalid, + _ => unreachable!(), + } + let error = invalid_config.validate().unwrap_err().to_string(); + assert!(error.contains(field), "{error}"); + assert!(error.contains("stable identity"), "{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 05b3ea4045..4d2abff54e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -16,6 +16,7 @@ pub mod provider_bridge; pub mod provider_events; pub mod question_response; pub mod replay; +mod stable_identity; use std::error::Error; use std::fmt::{self, Display, Formatter}; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index 844dd22867..47c6b9042f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs @@ -1,8 +1,14 @@ use serde_json::{json, Value}; use sha2::{Digest, Sha256}; -use crate::acpx_event_payload::AcpxRuntimeEventKind; -use crate::durable::{redact_text, EventPriority}; +use crate::acpx_event_payload::{AcpxRuntimeEventKind, AcpxTurnStatus}; +use crate::acpx_provider_state::{is_reserved_terminal_operation, AcpxProviderStateEvent}; +use crate::durable::{redact_text, sanitize_value, EventPriority}; +use crate::local_runner::LocalRunnerError; +use crate::provider_bridge::{semantic_value_digest, ToolResult}; +use crate::stable_identity::{ + is_stable_id, project_acpx_runtime_request_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS, +}; const MAX_TEXT_CHARS: usize = 4_000; @@ -36,6 +42,331 @@ pub(crate) fn normalized_codex_terminal_event_type( }) } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AcpxEventProjectionContext { + pub run_id: String, + pub normalized_session_id: String, + pub turn_id: String, + pub item_id: String, +} + +impl AcpxEventProjectionContext { + pub fn validate(&self) -> Result<(), LocalRunnerError> { + for (value, label, max_chars) in [ + (&self.run_id, "run", SHORT_STABLE_ID_CHARS), + ( + &self.normalized_session_id, + "normalized session", + SHORT_STABLE_ID_CHARS, + ), + // Turn and item correlation use the durable 240-character + // identity contract shared by PRP events, runtime requests, and + // semantic-tool receipts. + (&self.turn_id, "turn", DURABLE_STABLE_ID_CHARS), + (&self.item_id, "item", DURABLE_STABLE_ID_CHARS), + ] { + validate_projection_identity(value, label, max_chars)?; + } + Ok(()) + } + + fn correlation(&self) -> Value { + json!({ + "runId": self.run_id, + "normalizedSessionId": self.normalized_session_id, + "turnId": self.turn_id, + "itemId": self.item_id, + }) + } +} + +/// Projects already scope-checked ACPX reducer output into provider-neutral +/// durable events. The reducer remains authoritative for bounds and request +/// state; this boundary must not accept raw sidecar envelopes. +pub fn project_acpx_state_event( + context: &AcpxEventProjectionContext, + event: &AcpxProviderStateEvent, +) -> Result, LocalRunnerError> { + context.validate()?; + let one = |event_type: &str, priority: EventPriority, payload: Value| { + Ok(vec![NormalizedProviderEvent { + event_type: event_type.to_owned(), + priority, + payload, + }]) + }; + match event { + AcpxProviderStateEvent::Activity(event) => Ok(vec![event.clone()]), + AcpxProviderStateEvent::ToolCall { + call_id, + operation_id, + input, + } => { + validate_semantic_projection_identity(call_id, operation_id)?; + one( + "semantic_tool.input", + EventPriority::P0, + json!({ + "semantic_tool": { + "schema": "paperclip.prp.semantic_tool.v1", + "schemaVersion": 1, + "phase": "input", + "operationId": operation_id, + "callId": call_id, + "correlation": context.correlation(), + "idempotencyKey": Value::Null, + "content": { + "digest": semantic_value_digest(input), + "redactionDisposition": "digest_only", + "references": [], + }, + "input": input, + }, + }), + ) + } + AcpxProviderStateEvent::ToolResult(result) => { + Ok(vec![project_acpx_tool_result(context, result)?]) + } + AcpxProviderStateEvent::PermissionRequest { .. } => Err(LocalRunnerError::invalid( + "ACPX permission request reached projection outside the pinned runner policy", + )), + AcpxProviderStateEvent::InputRequest { + request_id, + question_set, + origin, + } => { + let request_id = project_acpx_runtime_request_id(request_id).ok_or_else(|| { + LocalRunnerError::invalid("ACPX event projection request identity is invalid") + })?; + let prompt = question_set + .get("title") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .or_else(|| { + question_set + .pointer("/questions/0/prompt") + .and_then(Value::as_str) + }) + .map(|value| bounded_text(value, MAX_TEXT_CHARS)) + .unwrap_or_else(|| "Codex needs your input".to_owned()); + let origin = project_runtime_request_origin(origin.as_ref())?; + one( + "runtime_request.created", + EventPriority::P0, + json!({ + "request": { + "schema": "paperclip.runtime_request.v2", + "requestKind": "runtime", + "requestId": request_id, + "turnId": context.turn_id, + "itemId": context.item_id, + "type": "input", + "status": "pending", + "prompt": prompt, + "input": question_set, + "origin": origin, + }, + }), + ) + } + AcpxProviderStateEvent::SemanticResult(result) => { + validate_semantic_projection_identity(&result.call_id, &result.operation_id)?; + if is_reserved_terminal_operation(&result.operation_id) { + one( + "run.result.proposed", + EventPriority::P0, + result.result.clone(), + ) + } else { + Ok(vec![project_acpx_tool_result( + context, + &ToolResult { + call_id: result.call_id.clone(), + operation_id: result.operation_id.clone(), + result: result.result.clone(), + is_error: !result.ok, + }, + )?]) + } + } + AcpxProviderStateEvent::AssistantMessage { turn_id, text } => { + require_projected_turn(context, turn_id)?; + one( + "item.completed", + EventPriority::P1, + json!({ + "provider": "acpx", + "itemId": context.item_id, + "kind": "agentMessage", + "status": "completed", + "channel": "final", + "text": bounded_text(text, MAX_TEXT_CHARS), + }), + ) + } + AcpxProviderStateEvent::TurnTerminal { + turn_id, + status, + error, + } => { + require_projected_turn(context, turn_id)?; + let (event_type, status) = match status { + AcpxTurnStatus::Completed => ("turn.completed", "completed"), + AcpxTurnStatus::Failed => ("turn.failed", "failed"), + AcpxTurnStatus::Cancelled => ("turn.cancelled", "cancelled"), + AcpxTurnStatus::Interrupted => ("turn.interrupted", "interrupted"), + }; + one( + event_type, + EventPriority::P0, + json!({ + "provider": "acpx", + "providerTurnId": turn_id, + "status": status, + "error": error, + }), + ) + } + AcpxProviderStateEvent::Process(details) => one( + "harness.diagnostic", + EventPriority::P1, + json!({ + "code": "acpx_process", + "message": "The ACPX sidecar reported provider process metadata.", + "details": details, + }), + ), + AcpxProviderStateEvent::Diagnostic { code, message } => one( + "harness.diagnostic", + EventPriority::P1, + json!({"code": code, "message": bounded_text(message, MAX_TEXT_CHARS)}), + ), + } +} + +fn project_acpx_tool_result( + context: &AcpxEventProjectionContext, + result: &ToolResult, +) -> Result { + validate_semantic_projection_identity(&result.call_id, &result.operation_id)?; + let safe_result = sanitize_value(&result.result); + Ok(NormalizedProviderEvent { + event_type: "semantic_tool.result".to_owned(), + priority: EventPriority::P0, + payload: json!({ + "semantic_tool": { + "schema": "paperclip.prp.semantic_tool.v1", + "schemaVersion": 1, + "phase": "result", + "operationId": result.operation_id, + "callId": result.call_id, + "correlation": context.correlation(), + "idempotencyKey": Value::Null, + "content": { + "digest": semantic_value_digest(&safe_result), + "redactionDisposition": "digest_only", + "references": [], + }, + "outcome": if result.is_error { "failed" } else { "succeeded" }, + "code": if result.is_error { "semantic_tool_failed" } else { "semantic_tool_succeeded" }, + "retryable": false, + "authorizationBoundary": "active_task", + "operationReceiptId": format!("operation_{}", result.call_id), + }, + }), + }) +} + +fn validate_semantic_projection_identity( + call_id: &str, + operation_id: &str, +) -> Result<(), LocalRunnerError> { + for (value, label) in [(call_id, "call"), (operation_id, "operation")] { + if !is_stable_id(value, SHORT_STABLE_ID_CHARS) { + return Err(LocalRunnerError::invalid(format!( + "ACPX semantic {label} identity is invalid" + ))); + } + } + Ok(()) +} + +fn project_runtime_request_origin(origin: Option<&Value>) -> Result { + let Some(origin) = origin else { + return Ok(json!({ + "adapter": "codex-acpx", + "provider": "codex", + "method": "runtime.input_requested", + })); + }; + let object = origin.as_object().ok_or_else(|| { + LocalRunnerError::invalid("ACPX runtime request origin must be an object") + })?; + if object + .keys() + .any(|key| !matches!(key.as_str(), "adapter" | "provider" | "method")) + { + return Err(LocalRunnerError::invalid( + "ACPX runtime request origin contains unsupported fields", + )); + } + validate_origin_field(object.get("adapter"), "adapter", 160, true)?; + validate_origin_field(object.get("provider"), "provider", 160, false)?; + validate_origin_field(object.get("method"), "method", 500, false)?; + Ok(origin.clone()) +} + +fn validate_origin_field( + value: Option<&Value>, + field: &str, + max_chars: usize, + required: bool, +) -> Result<(), LocalRunnerError> { + let Some(value) = value else { + if required { + return Err(LocalRunnerError::invalid(format!( + "ACPX runtime request origin omitted {field}" + ))); + } + return Ok(()); + }; + let text = value.as_str().ok_or_else(|| { + LocalRunnerError::invalid(format!("ACPX runtime request origin {field} must be text")) + })?; + if text.is_empty() || text.chars().count() > max_chars { + return Err(LocalRunnerError::invalid(format!( + "ACPX runtime request origin {field} is invalid" + ))); + } + Ok(()) +} + +fn validate_projection_identity( + value: &str, + label: &str, + max_chars: usize, +) -> Result<(), LocalRunnerError> { + if !is_stable_id(value, max_chars) { + return Err(LocalRunnerError::invalid(format!( + "ACPX event projection {label} identity is invalid" + ))); + } + Ok(()) +} + +fn require_projected_turn( + context: &AcpxEventProjectionContext, + turn_id: &str, +) -> Result<(), LocalRunnerError> { + if turn_id != context.turn_id { + return Err(LocalRunnerError::invalid( + "ACPX state event does not match its durable turn projection", + )); + } + Ok(()) +} + fn bounded_text(value: &str, max_chars: usize) -> String { redact_text(value).chars().take(max_chars).collect() } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/stable_identity.rs b/packages/paperclip-runner/runner/crates/runner-core/src/stable_identity.rs new file mode 100644 index 0000000000..6916404188 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/stable_identity.rs @@ -0,0 +1,39 @@ +use sha2::{Digest, Sha256}; + +pub(crate) const SHORT_STABLE_ID_CHARS: usize = 160; +pub(crate) const DURABLE_STABLE_ID_CHARS: usize = 240; + +const ACPX_RUNTIME_REQUEST_DOMAIN: &[u8] = b"paperclip.acpx.runtime-request.v1\0"; +const ACPX_RUNTIME_REQUEST_PREFIX: &str = "acpx-request-"; + +pub(crate) fn is_stable_id(value: &str, max_chars: usize) -> bool { + let mut characters = value.chars(); + characters + .next() + .is_some_and(|character| character.is_ascii_alphanumeric()) + && characters + .all(|character| character.is_ascii_alphanumeric() || "._:-".contains(character)) + && value.len() <= max_chars +} + +/// Preserves already-canonical runtime request IDs and deterministically +/// projects other bounded ACPX IDs into the stricter PRP request namespace. +/// The upstream ID remains authoritative for the sidecar resolution command. +pub(crate) fn project_acpx_runtime_request_id(value: &str) -> Option { + if value.is_empty() + || value.chars().count() > DURABLE_STABLE_ID_CHARS + || value.chars().any(char::is_control) + { + return None; + } + if is_stable_id(value, SHORT_STABLE_ID_CHARS) { + return Some(value.to_owned()); + } + let mut digest = Sha256::new(); + digest.update(ACPX_RUNTIME_REQUEST_DOMAIN); + digest.update(value.as_bytes()); + Some(format!( + "{ACPX_RUNTIME_REQUEST_PREFIX}{:x}", + digest.finalize() + )) +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_projection.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_projection.rs new file mode 100644 index 0000000000..31caa6d2a7 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_projection.rs @@ -0,0 +1,471 @@ +use paperclip_runner_core::acpx_event_payload::AcpxTurnStatus; +use paperclip_runner_core::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent}; +use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent; +use paperclip_runner_core::durable::EventPriority; +use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType; +use paperclip_runner_core::provider_bridge::ToolResult; +use paperclip_runner_core::provider_events::{ + project_acpx_state_event, AcpxEventProjectionContext, NormalizedProviderEvent, +}; +use serde_json::{json, Value}; + +fn context() -> AcpxEventProjectionContext { + AcpxEventProjectionContext { + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + } +} + +fn project(event: AcpxProviderStateEvent) -> Vec { + project_acpx_state_event(&context(), &event).unwrap() +} + +fn reduced_semantic_result( + call_id: &str, + operation_id: &str, + ok: bool, + result: Value, +) -> AcpxProviderStateEvent { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + state + .accept_event(&AcpxSidecarEvent { + sequence: 1, + event_type: GeneratedAcpxSidecarEventType::RuntimeEvent, + run_id: Some("run-1".to_owned()), + turn_id: Some("turn-1".to_owned()), + payload: json!({ + "type":"semantic_result", + "callId":call_id, + "operationId":operation_id, + "ok":ok, + "result":result, + }), + }) + .unwrap() + .remove(0) +} + +#[test] +fn projects_authorized_tools_with_exact_durable_correlation() { + let events = project(AcpxProviderStateEvent::ToolCall { + call_id: "call-1".to_owned(), + operation_id: "get_task_context".to_owned(), + input: json!({"taskId":"task-1"}), + }); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_type, "semantic_tool.input"); + assert_eq!(events[0].priority, EventPriority::P0); + assert_eq!( + events[0].payload["semantic_tool"]["correlation"], + json!({ + "runId":"run-1", + "normalizedSessionId":"session-1", + "turnId":"turn-1", + "itemId":"item-1", + }) + ); + assert_eq!( + events[0].payload["semantic_tool"]["input"], + json!({"taskId":"task-1"}) + ); + assert!(events[0].payload["semantic_tool"]["content"]["digest"] + .as_str() + .is_some_and(|value| value.starts_with("sha256:"))); +} + +#[test] +fn projects_terminal_tool_cancellations_as_correlated_results() { + let events = project(AcpxProviderStateEvent::ToolResult(ToolResult { + call_id: "call-1".to_owned(), + operation_id: "issues.read".to_owned(), + result: json!({ + "error": { + "code": "acpx_turn_settled", + "message": "The provider turn stopped before this semantic tool completed", + "retryable": false, + }, + }), + is_error: true, + })); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].event_type, "semantic_tool.result"); + assert_eq!(events[0].priority, EventPriority::P0); + assert_eq!(events[0].payload["semantic_tool"]["phase"], "result"); + assert_eq!(events[0].payload["semantic_tool"]["callId"], "call-1"); + assert_eq!(events[0].payload["semantic_tool"]["outcome"], "failed"); + assert_eq!( + events[0].payload["semantic_tool"]["correlation"], + json!({ + "runId":"run-1", + "normalizedSessionId":"session-1", + "turnId":"turn-1", + "itemId":"item-1", + }), + ); +} + +#[test] +fn projects_structured_input_and_semantic_results_without_provider_envelopes() { + let question_set = json!({ + "schema":"paperclip.question_set.v1", + "title":"Choose a target", + "questions":[], + }); + let input = project(AcpxProviderStateEvent::InputRequest { + request_id: "request-1".to_owned(), + question_set: question_set.clone(), + origin: None, + }); + assert_eq!(input[0].event_type, "runtime_request.created"); + assert_eq!(input[0].payload["request"]["requestId"], "request-1"); + assert_eq!(input[0].payload["request"]["turnId"], "turn-1"); + assert_eq!(input[0].payload["request"]["itemId"], "item-1"); + assert_eq!(input[0].payload["request"]["input"], question_set); + assert_eq!( + input[0].payload["request"]["origin"]["adapter"], + "codex-acpx" + ); + + let result = json!({ + "schema":"paperclip.run_result.v1", + "reportedWorkDisposition":"done", + "summary":"Finished", + }); + let projected = project(reduced_semantic_result( + "finish-1", + "paperclip_finish", + true, + result.clone(), + )); + assert_eq!(projected[0].event_type, "run.result.proposed"); + assert_eq!(projected[0].payload, result); + + let dynamic = project(reduced_semantic_result( + "call-1", + "issues.read", + true, + json!({"id":"issue-1"}), + )); + assert_eq!(dynamic[0].event_type, "semantic_tool.result"); + assert_eq!(dynamic[0].payload["semantic_tool"]["phase"], "result"); + assert_eq!( + dynamic[0].payload["semantic_tool"]["operationId"], + "issues.read" + ); + assert_eq!(dynamic[0].payload["semantic_tool"]["callId"], "call-1"); + + let activity = NormalizedProviderEvent { + event_type: "usage.reported".to_owned(), + priority: EventPriority::P0, + payload: json!({"cumulative":{"inputTokens":1}}), + }; + assert_eq!( + project(AcpxProviderStateEvent::Activity(activity.clone())), + vec![activity] + ); +} + +#[test] +fn projects_runtime_request_prompt_and_origin_into_the_strict_schema() { + let empty_title = AcpxProviderStateEvent::InputRequest { + request_id: "request-1".to_owned(), + question_set: json!({ + "schema":"paperclip.question_set.v1", + "title":"", + "questions":[{ + "id":"target", + "prompt":"Which target?", + "required":true, + "answerMode":"single_select", + "options":[{"id":"first","label":"First"}], + }], + }), + origin: Some(json!({ + "adapter":"codex-acpx", + "provider":"codex", + "method":"runtime.input_requested", + })), + }; + let projected = project(empty_title); + assert_eq!(projected[0].payload["request"]["prompt"], "Which target?"); + + for origin in [ + json!({"provider":"codex"}), + json!({"adapter":"codex-acpx","extra":true}), + json!({"adapter":""}), + json!({"adapter":"codex-acpx","method":null}), + ] { + let invalid = AcpxProviderStateEvent::InputRequest { + request_id: "request-1".to_owned(), + question_set: json!({ + "schema":"paperclip.question_set.v1", + "questions":[], + }), + origin: Some(origin), + }; + assert!(project_acpx_state_event(&context(), &invalid).is_err()); + } +} + +#[test] +fn projects_assistant_terminal_and_diagnostic_events_fail_closed() { + let assistant = project(AcpxProviderStateEvent::AssistantMessage { + turn_id: "turn-1".to_owned(), + text: "Done".to_owned(), + }); + assert_eq!(assistant[0].event_type, "item.completed"); + assert_eq!(assistant[0].payload["itemId"], "item-1"); + assert_eq!(assistant[0].payload["channel"], "final"); + + for (status, expected) in [ + (AcpxTurnStatus::Completed, "turn.completed"), + (AcpxTurnStatus::Failed, "turn.failed"), + (AcpxTurnStatus::Cancelled, "turn.cancelled"), + (AcpxTurnStatus::Interrupted, "turn.interrupted"), + ] { + let terminal = project(AcpxProviderStateEvent::TurnTerminal { + turn_id: "turn-1".to_owned(), + status, + error: None, + }); + assert_eq!(terminal[0].event_type, expected); + } + + let process = project(AcpxProviderStateEvent::Process(json!({"pid":7}))); + assert_eq!(process[0].event_type, "harness.diagnostic"); + assert_eq!(process[0].payload["details"]["pid"], 7); + let diagnostic = project(AcpxProviderStateEvent::Diagnostic { + code: "provider_notice".to_owned(), + message: "Retrying".to_owned(), + }); + assert_eq!(diagnostic[0].payload["message"], "Retrying"); + + let wrong_turn = AcpxProviderStateEvent::TurnTerminal { + turn_id: "turn-other".to_owned(), + status: AcpxTurnStatus::Completed, + error: None, + }; + assert!(project_acpx_state_event(&context(), &wrong_turn) + .unwrap_err() + .to_string() + .contains("durable turn projection")); + let permission = AcpxProviderStateEvent::PermissionRequest { + request_id: "permission-1".to_owned(), + kind: "write".to_owned(), + title: "Allow write".to_owned(), + details: Value::Null, + }; + assert!(project_acpx_state_event(&context(), &permission) + .unwrap_err() + .to_string() + .contains("pinned runner policy")); +} + +#[test] +fn rejects_invalid_durable_projection_identity() { + let mut invalid = context(); + invalid.run_id = "".to_owned(); + let event = AcpxProviderStateEvent::Diagnostic { + code: "notice".to_owned(), + message: "message".to_owned(), + }; + assert!(project_acpx_state_event(&invalid, &event) + .unwrap_err() + .to_string() + .contains("run identity")); + + for (field, max_chars) in [ + ("run", 160), + ("normalized session", 160), + ("turn", 240), + ("item", 240), + ] { + let mut invalid = context(); + let oversized = "x".repeat(max_chars + 1); + match field { + "run" => invalid.run_id = oversized, + "normalized session" => invalid.normalized_session_id = oversized, + "turn" => invalid.turn_id = oversized, + "item" => invalid.item_id = oversized, + _ => unreachable!(), + } + let error = project_acpx_state_event(&invalid, &event) + .unwrap_err() + .to_string(); + assert!(error.contains(&format!("{field} identity")), "{error}"); + } + + for request_id in [String::new(), "x".repeat(241), "request\n1".to_owned()] { + let request = AcpxProviderStateEvent::InputRequest { + request_id, + question_set: json!({ + "schema":"paperclip.question_set.v1", + "questions":[], + }), + origin: None, + }; + let error = project_acpx_state_event(&context(), &request) + .unwrap_err() + .to_string(); + assert!(error.contains("request identity"), "{error}"); + } + + for field in ["run", "normalized session", "turn", "item"] { + let mut invalid = context(); + match field { + "run" => invalid.run_id = "run 1".to_owned(), + "normalized session" => invalid.normalized_session_id = "session/1".to_owned(), + "turn" => invalid.turn_id = "turn 1".to_owned(), + "item" => invalid.item_id = "item/1".to_owned(), + _ => unreachable!(), + } + let error = project_acpx_state_event(&invalid, &event) + .unwrap_err() + .to_string(); + assert!(error.contains(&format!("{field} identity")), "{error}"); + } +} + +#[test] +fn rejects_semantic_events_that_cannot_form_stable_receipts() { + for event in [ + AcpxProviderStateEvent::ToolCall { + call_id: "call 1".to_owned(), + operation_id: "issues.read".to_owned(), + input: json!({}), + }, + AcpxProviderStateEvent::ToolResult(ToolResult { + call_id: "call-1".to_owned(), + operation_id: "issues/read".to_owned(), + result: json!({}), + is_error: false, + }), + reduced_semantic_result("réturn-1", "issues.read", true, json!({})), + reduced_semantic_result("call-1", "issues read", true, json!({})), + ] { + let error = project_acpx_state_event(&context(), &event) + .unwrap_err() + .to_string(); + assert!(error.contains("semantic"), "{error}"); + assert!(error.contains("identity"), "{error}"); + } +} + +#[test] +fn deterministically_projects_bounded_upstream_request_ids() { + let question_set = json!({ + "schema":"paperclip.question_set.v1", + "questions":[], + }); + for upstream_id in [ + "request 1".to_owned(), + "réquest-1".to_owned(), + "request/1".to_owned(), + "_request-1".to_owned(), + "x".repeat(240), + ] { + let event = AcpxProviderStateEvent::InputRequest { + request_id: upstream_id.clone(), + question_set: question_set.clone(), + origin: None, + }; + let first = project_acpx_state_event(&context(), &event).unwrap(); + let second = project_acpx_state_event(&context(), &event).unwrap(); + let projected_id = first[0].payload["request"]["requestId"].as_str().unwrap(); + assert_ne!(projected_id, upstream_id); + assert_eq!(second[0].payload["request"]["requestId"], projected_id); + assert!(projected_id.starts_with("acpx-request-")); + assert!(projected_id.len() <= 160); + assert!(projected_id + .chars() + .all(|character| character.is_ascii_alphanumeric() || "._:-".contains(character))); + } + + let canonical = AcpxProviderStateEvent::InputRequest { + request_id: "request-1".to_owned(), + question_set, + origin: None, + }; + assert_eq!( + project_acpx_state_event(&context(), &canonical).unwrap()[0].payload["request"] + ["requestId"], + "request-1" + ); +} + +#[test] +fn runtime_request_projection_preserves_durable_identity_boundaries() { + let canonical_request_schema: Value = serde_json::from_str(include_str!( + "../../../../protocol/schemas/request.schema.json" + )) + .unwrap(); + let mut runtime_request_schema = canonical_request_schema["oneOf"][0].clone(); + // This test owns identity projection. The question-set validator has its + // own coverage, so replace its remote reference with an unconstrained + // local schema before compiling the canonical runtime-request branch. + runtime_request_schema["properties"]["input"] = json!({}); + let request_validator = jsonschema::validator_for(&runtime_request_schema).unwrap(); + let request_event = AcpxProviderStateEvent::InputRequest { + request_id: "request-1".to_owned(), + question_set: json!({ + "schema":"paperclip.question_set.v1", + "questions":[], + }), + origin: None, + }; + let valid = project_acpx_state_event(&context(), &request_event).unwrap(); + assert!(request_validator.is_valid(&valid[0].payload["request"])); + + let mut durable_context = context(); + durable_context.turn_id = "t".repeat(240); + durable_context.item_id = "i".repeat(240); + let durable_request = project_acpx_state_event(&durable_context, &request_event).unwrap(); + assert_eq!( + durable_request[0].payload["request"]["turnId"], + durable_context.turn_id + ); + assert_eq!( + durable_request[0].payload["request"]["itemId"], + durable_context.item_id + ); + assert!(request_validator.is_valid(&durable_request[0].payload["request"])); + + for request_id in ["request 1", "réquest-1", "request/1", "_request-1"] { + let mut legacy_request = valid[0].payload["request"].clone(); + legacy_request["requestId"] = Value::String(request_id.to_owned()); + assert!(request_validator.is_valid(&legacy_request), "{request_id}"); + } + + for field in ["turnId", "itemId"] { + let mut invalid_request = valid[0].payload["request"].clone(); + invalid_request[field] = Value::String("x".repeat(241)); + assert!(!request_validator.is_valid(&invalid_request)); + } + + let diagnostic = AcpxProviderStateEvent::Diagnostic { + code: "notice".to_owned(), + message: "message".to_owned(), + }; + assert!(project_acpx_state_event(&durable_context, &diagnostic).is_ok()); + + let semantic = project_acpx_state_event( + &durable_context, + &AcpxProviderStateEvent::ToolCall { + call_id: "call-1".to_owned(), + operation_id: "issues.read".to_owned(), + input: json!({"taskId":"task-1"}), + }, + ) + .unwrap(); + let semantic_schema: Value = serde_json::from_str(include_str!( + "../../../../protocol/schemas/semantic-tool.schema.json" + )) + .unwrap(); + let semantic_validator = jsonschema::validator_for(&semantic_schema).unwrap(); + assert!(semantic_validator.is_valid(&semantic[0].payload["semantic_tool"])); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_scope.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_scope.rs index d916c58912..604d06ed21 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_scope.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_scope.rs @@ -174,8 +174,20 @@ fn bounds_settled_turn_identity_retention() { fn rejects_invalid_scope_identifiers() { assert!(AcpxEventScope::new("").is_err()); assert!(AcpxEventScope::new("run\n1").is_err()); + assert!(AcpxEventScope::new("run 1").is_err()); let mut scope = AcpxEventScope::new("run-1").unwrap(); - assert!(scope.bind_turn("t".repeat(161)).is_err()); + let longest_turn_id = "t".repeat(240); + scope.bind_turn(&longest_turn_id).unwrap(); + scope + .validate_event(&event( + GeneratedAcpxSidecarEventType::RuntimeDiagnostic, + Some("run-1"), + Some(&longest_turn_id), + )) + .unwrap(); + scope.clear_turn(&longest_turn_id).unwrap(); + assert!(scope.bind_turn("t".repeat(241)).is_err()); + assert!(scope.bind_turn("turn 1").is_err()); let oversized_run_id = "r".repeat(161); assert!(scope .validate_event(&event( diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs index e9a8d4b2ae..a551825a7f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs @@ -6,10 +6,14 @@ use std::time::Duration; use paperclip_runner_core::acpx_provider_session::{ AcpxPermissionMode, AcpxProviderSession, AcpxProviderSessionConfig, }; +use paperclip_runner_core::acpx_provider_state::AcpxProviderStateEvent; use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarTransportConfig; use paperclip_runner_core::provider_bridge::{ authorized_tool_catalog_digest, AuthorizedTool, AuthorizedToolSet, ToolResult, }; +use paperclip_runner_core::provider_events::{ + project_acpx_state_event, AcpxEventProjectionContext, +}; use serde_json::json; fn tool_set() -> AuthorizedToolSet { @@ -98,6 +102,47 @@ fn commits_each_resolution_only_after_sidecar_acknowledgement() { session.shutdown("test complete").unwrap(); } +#[test] +fn projected_request_id_resolves_the_exact_upstream_sidecar_request() { + let mut session = AcpxProviderSession::start(&config("resolutions-projected-id")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let mut input = None; + for _ in 0..2 { + for event in session.poll_event(Duration::from_secs(1)).unwrap().unwrap() { + if matches!(&event, AcpxProviderStateEvent::InputRequest { .. }) { + input = Some(event); + } + } + } + let input = input.expect("observe the upstream input request"); + let projected = project_acpx_state_event( + &AcpxEventProjectionContext { + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + }, + &input, + ) + .unwrap(); + let request_id = projected[0].payload["request"]["requestId"] + .as_str() + .unwrap() + .to_owned(); + assert!(request_id.starts_with("acpx-request-")); + + session + .deliver_tool_result(&tool_result("issues.read")) + .unwrap(); + session + .resolve_input(&request_id, "turn-1", &input_resolution("first")) + .unwrap(); + assert!(session.state().pending_question_set(&request_id).is_none()); + session.shutdown("test complete").unwrap(); +} + #[test] fn local_validation_preserves_pending_requests_for_a_correct_retry() { let mut session = started("resolutions"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs index 59dca6f390..471ca2c67b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs @@ -2,6 +2,9 @@ use paperclip_runner_core::acpx_event_payload::AcpxTurnStatus; use paperclip_runner_core::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent}; use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent; use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType; +use paperclip_runner_core::provider_events::{ + project_acpx_state_event, AcpxEventProjectionContext, +}; use serde_json::{json, Value}; fn event( @@ -155,6 +158,38 @@ fn tracks_structured_input_and_permission_requests_without_cross_kind_reuse() { state.complete_permission("permission-1").unwrap(); } +#[test] +fn correlates_projected_runtime_requests_to_the_upstream_input_id() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let upstream_request_id = format!("input / {}", "é".repeat(200)); + let emitted = state + .accept_event(&event( + 1, + GeneratedAcpxSidecarEventType::RuntimeInputRequested, + Some("turn-1"), + json!({"requestId":upstream_request_id,"questionSet":question_set()}), + )) + .unwrap(); + let projected = project_acpx_state_event( + &AcpxEventProjectionContext { + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + }, + &emitted[0], + ) + .unwrap(); + let runtime_request_id = projected[0].payload["request"]["requestId"] + .as_str() + .unwrap(); + assert!(runtime_request_id.starts_with("acpx-request-")); + assert!(state.pending_question_set(runtime_request_id).is_some()); + state.complete_input(runtime_request_id).unwrap(); + assert!(state.pending_question_set(runtime_request_id).is_none()); +} + #[test] fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() { let mut state = AcpxProviderState::new("run-1").unwrap(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs index d2eba6cf2a..c751b43b7a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -103,6 +103,43 @@ fn rejects_invalid_turn_inputs_without_mutating_the_session() { session.shutdown("test complete").unwrap(); } +#[test] +fn preserves_the_durable_turn_identity_boundary_through_the_sidecar() { + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + let turn_id = "t".repeat(240); + let response = session + .start_turn(&turn_id, "Please help", &std::env::temp_dir()) + .unwrap(); + assert_eq!(response["turnId"], turn_id); + + let activity = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &activity[0], + AcpxProviderStateEvent::Activity(event) + if event.event_type == "item.delta" && event.payload["text"] == "hello" + )); + session + .interrupt_turn(&turn_id, "Paperclip interruption") + .unwrap(); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } if settled == &turn_id + )); + session.shutdown("test complete").unwrap(); + + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + for invalid_turn_id in ["t".repeat(241), "turn 1".to_owned()] { + assert!(session + .start_turn(&invalid_turn_id, "Please help", &std::env::temp_dir(),) + .unwrap_err() + .to_string() + .contains("turn id is invalid")); + } + assert_eq!(session.state().active_turn_id(), None); + session.shutdown("test complete").unwrap(); +} + #[test] fn fails_closed_when_turn_start_acknowledges_another_turn() { let mut session = AcpxProviderSession::start(&config("turns-wrong-turn")).unwrap(); diff --git a/packages/paperclip-runner/src/protocol/generated/schema-bundle.ts b/packages/paperclip-runner/src/protocol/generated/schema-bundle.ts index fcfabda1b3..81873ee78b 100644 --- a/packages/paperclip-runner/src/protocol/generated/schema-bundle.ts +++ b/packages/paperclip-runner/src/protocol/generated/schema-bundle.ts @@ -2764,12 +2764,12 @@ export const requestSchema = { "turnId": { "type": "string", "minLength": 1, - "maxLength": 160 + "maxLength": 240 }, "itemId": { "type": "string", "minLength": 1, - "maxLength": 160 + "maxLength": 240 } }, "additionalProperties": false @@ -3258,12 +3258,12 @@ export const eventSchema = { "turnId": { "type": "string", "minLength": 1, - "maxLength": 160 + "maxLength": 240 }, "itemId": { "type": "string", "minLength": 1, - "maxLength": 160 + "maxLength": 240 }, "eventType": { "enum": [ diff --git a/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs b/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs index 10695fc9a6..f3d2314b8e 100644 --- a/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs +++ b/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs @@ -47,6 +47,19 @@ test("the ACPX sidecar schema accepts each versioned message family", () => { } }); +test("the ACPX sidecar schema shares the durable stable-identity boundary", () => { + const longestTurnId = "t".repeat(240); + assert.equal(validate({ ...messages[2], turnId: longestTurnId }), true); + assert.equal(validate({ ...messages[2], turnId: "t".repeat(241) }), false); + for (const turnId of ["turn 1", "réturn-1", "turn/1", "_turn-1"]) { + assert.equal(validate({ ...messages[2], turnId }), false, turnId); + } + assert.equal(validate({ ...messages[2], runId: "r".repeat(161) }), false); + for (const runId of ["run 1", "rún-1", "run/1", "_run-1"]) { + assert.equal(validate({ ...messages[2], runId }), false, runId); + } +}); + test("the ACPX sidecar schema fails closed on drift", () => { for (const message of [ { ...messages[0], protocolVersion: protocolVersion + 1 },