diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 71604a5e3f..a4b60c0e32 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -93,6 +93,15 @@ 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. +The session can then start one immutable-workspace turn, request interruption, +and reduce polled events through the scope-first state boundary. A mismatched +command acknowledgement or invalid event terminates the session fail closed. +Polled semantic calls pass through the run-scoped authorized tool bridge before +they can be returned to a caller. Before a follow-up turn releases settled tool +receipts, runner-core suspends and reaps the idle sidecar/provider generation, +then resumes the same verified persistent identity in a fresh generation. This +prevents a late session-lifetime MCP callback from inheriting the next turn's +event authority. Run the complete contract gate with: 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 d420acf8b5..d78d2f168a 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 @@ -9,6 +9,7 @@ use crate::generated_acpx_sidecar_contract::{ classify_generated_acpx_tool_operation, GeneratedAcpxSidecarEventType, }; use crate::local_runner::LocalRunnerError; +use crate::provider_bridge::semantic_value_digest; const MAX_EVENT_PAYLOAD_BYTES: usize = 256 * 1024; const MAX_ID_CHARS: usize = 160; @@ -41,6 +42,7 @@ pub enum AcpxEventPayload { kind: AcpxRuntimeEventKind, tool_operation: Option<&'static str>, payload: Value, + semantic_result_digest: Option, }, PermissionRequested { request_id: String, @@ -57,6 +59,7 @@ pub enum AcpxEventPayload { call_id: String, operation_id: String, input: Value, + input_digest: String, }, TurnTerminal { status: AcpxTurnStatus, @@ -123,6 +126,7 @@ pub fn decode_acpx_event( Ok(AcpxEventPayload::ToolCalled { call_id: required_id(&event.payload, "callId", "tool call")?, operation_id: required_id(&event.payload, "operationId", "tool operation")?, + input_digest: semantic_value_digest(&input), input: sanitize_value(&input), }) } @@ -217,6 +221,7 @@ fn decode_runtime_event(payload: &Value) -> Result { bounded_required_text(payload, "text", MAX_RUNTIME_TEXT_CHARS, "runtime text")?; @@ -277,11 +282,17 @@ fn decode_runtime_event(payload: &Value) -> Result { required_id(payload, "callId", "semantic result call")?; required_id(payload, "operationId", "semantic result operation")?; + if !payload.get("ok").is_some_and(Value::is_boolean) { + return Err(LocalRunnerError::invalid( + "ACPX semantic result must contain a boolean outcome", + )); + } if !payload.get("result").is_some_and(Value::is_object) { return Err(LocalRunnerError::invalid( "ACPX semantic result must contain an object result", )); } + semantic_result_digest = payload.get("result").map(semantic_value_digest); AcpxRuntimeEventKind::SemanticResult } "provider_notice" => { @@ -308,6 +319,7 @@ fn decode_runtime_event(payload: &Value) -> Result, + settled_turn_ids: BTreeSet, } impl AcpxEventScope { @@ -22,6 +26,7 @@ impl AcpxEventScope { Ok(Self { run_id, active_turn_id: None, + settled_turn_ids: BTreeSet::new(), }) } @@ -33,6 +38,10 @@ impl AcpxEventScope { self.active_turn_id.as_deref() } + pub(crate) fn has_settled_turns(&self) -> bool { + !self.settled_turn_ids.is_empty() + } + pub fn bind_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { let turn_id = turn_id.into(); validate_scope_id(&turn_id, "turn")?; @@ -42,12 +51,57 @@ impl AcpxEventScope { "ACPX event scope already has a different active turn", )), None => { + self.validate_new_turn_identity(&turn_id)?; self.active_turn_id = Some(turn_id); Ok(()) } } } + pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> { + self.validate_new_turn_identity_for_provider_restart(turn_id)?; + if self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS { + return Err(LocalRunnerError::invalid( + "ACPX event scope exhausted its settled turn identity capacity", + )); + } + Ok(()) + } + + pub(crate) fn validate_new_turn_identity_for_provider_restart( + &self, + turn_id: &str, + ) -> Result<(), LocalRunnerError> { + validate_scope_id(turn_id, "turn")?; + if self.settled_turn_ids.contains(turn_id) { + return Err(LocalRunnerError::invalid( + "ACPX event scope reused a settled turn identity", + )); + } + Ok(()) + } + + pub(crate) fn settled_turn_identity_capacity_reached(&self) -> bool { + self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS + } + + pub(crate) fn rotate_settled_turn_identities_after_provider_restart( + &mut self, + ) -> Result<(), LocalRunnerError> { + if self.active_turn_id.is_some() { + return Err(LocalRunnerError::invalid( + "ACPX event scope cannot rotate settled turn identities while a turn is active", + )); + } + if !self.settled_turn_identity_capacity_reached() { + return Err(LocalRunnerError::invalid( + "ACPX event scope cannot rotate settled turn identities before capacity", + )); + } + self.settled_turn_ids.clear(); + Ok(()) + } + pub fn clear_turn(&mut self, turn_id: &str) -> Result<(), LocalRunnerError> { validate_scope_id(turn_id, "turn")?; if self.active_turn_id.as_deref() != Some(turn_id) { @@ -55,6 +109,16 @@ impl AcpxEventScope { "ACPX event scope cannot clear a stale turn", )); } + if self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS { + return Err(LocalRunnerError::invalid( + "ACPX event scope exhausted its settled turn identity capacity", + )); + } + if !self.settled_turn_ids.insert(turn_id.to_owned()) { + return Err(LocalRunnerError::invalid( + "ACPX event scope reused a settled turn identity", + )); + } self.active_turn_id = None; Ok(()) } @@ -111,6 +175,34 @@ impl AcpxEventScope { } } +#[cfg(test)] +mod tests { + use super::{AcpxEventScope, MAX_SETTLED_TURN_IDS}; + + #[test] + fn rotates_a_full_ledger_only_after_revalidating_the_next_identity() { + let mut scope = AcpxEventScope::new("run-1").unwrap(); + for index in 0..MAX_SETTLED_TURN_IDS { + let turn_id = format!("turn-{index}"); + scope.bind_turn(&turn_id).unwrap(); + scope.clear_turn(&turn_id).unwrap(); + } + + assert!(scope.settled_turn_identity_capacity_reached()); + scope + .validate_new_turn_identity_for_provider_restart("turn-next") + .unwrap(); + assert!(scope + .validate_new_turn_identity_for_provider_restart("turn-0") + .is_err()); + scope + .rotate_settled_turn_identities_after_provider_restart() + .unwrap(); + scope.bind_turn("turn-next").unwrap(); + scope.clear_turn("turn-next").unwrap(); + } +} + fn validate_scope_id(value: &str, label: &str) -> Result<(), LocalRunnerError> { if value.is_empty() || value.chars().count() > MAX_SCOPE_ID_CHARS 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 e544626feb..af272e60ea 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -1,20 +1,26 @@ -use std::path::PathBuf; +use std::path::{Path, PathBuf}; +use std::time::Duration; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -use crate::acpx_provider_state::AcpxProviderState; +use crate::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent}; 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}; +use crate::provider_bridge::{ + authorized_tool_catalog_digest, AuthorizedTool, AuthorizedToolSet, ProviderToolBridge, + TOOL_SET_SCHEMA, +}; 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")] @@ -109,10 +115,21 @@ impl AcpxProviderSessionConfig { "ACPX system instructions exceed their bounded contract", )); } + if self + .tool_set + .operations + .iter() + .any(|tool| is_reserved_terminal_operation(&tool.operation_id)) + { + return Err(LocalRunnerError::invalid( + "ACPX run catalog cannot replace reserved terminal tools", + )); + } let mut bridge = ProviderToolBridge::default(); bridge.prepare(self.tool_set.clone()).map_err(|error| { LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}")) })?; + reserved_terminal_tool_bridge()?; if let Some(expected_identity) = self.expected_identity.as_ref() { expected_identity.validate()?; if expected_identity.normalized_session_id != self.normalized_session_id @@ -162,9 +179,13 @@ impl AcpxProviderSessionIdentity { pub struct AcpxProviderSession { transport: AcpxSidecarTransport, + config: AcpxProviderSessionConfig, state: AcpxProviderState, + tool_bridge: ProviderToolBridge, + reserved_tool_bridge: ProviderToolBridge, identity: AcpxProviderSessionIdentity, catalog_revision: u64, + working_directory: PathBuf, closed: bool, transport_terminated: bool, } @@ -172,6 +193,13 @@ pub struct AcpxProviderSession { impl AcpxProviderSession { pub fn start(config: &AcpxProviderSessionConfig) -> Result { config.validate()?; + let mut tool_bridge = ProviderToolBridge::default(); + tool_bridge + .prepare(config.tool_set.clone()) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}")) + })?; + let reserved_tool_bridge = reserved_terminal_tool_bridge()?; let mut transport = AcpxSidecarTransport::start(&config.transport)?; let bootstrap = bootstrap(&mut transport, config); let (identity, state) = match bootstrap { @@ -183,9 +211,13 @@ impl AcpxProviderSession { }; Ok(Self { transport, + config: config.clone(), state, + tool_bridge, + reserved_tool_bridge, identity, catalog_revision: config.catalog_revision, + working_directory: config.working_directory.clone(), closed: false, transport_terminated: false, }) @@ -207,6 +239,287 @@ impl AcpxProviderSession { self.catalog_revision } + pub fn start_turn( + &mut self, + turn_id: &str, + message: &str, + working_directory: &Path, + ) -> Result { + self.ensure_open()?; + validate_text(turn_id, 160, "ACPX turn id")?; + validate_turn_message(message)?; + if working_directory != self.working_directory { + return Err(LocalRunnerError::invalid( + "ACPX turn working directory differs from its immutable session workspace", + )); + } + if self.state.active_turn_id().is_some() { + return Err(LocalRunnerError::invalid( + "ACPX provider session already has an active turn", + )); + } + let rotate_turn_identity_ledger = self.state.settled_turn_identity_capacity_reached(); + let identity_validation = if rotate_turn_identity_ledger { + self.state + .validate_new_turn_identity_for_provider_restart(turn_id) + } else { + self.state.validate_new_turn_identity(turn_id) + }; + if let Err(error) = identity_validation { + // Reusing a settled identity would let a delayed event from the + // old turn alias the new receipt epoch. A full sidecar restart is + // required before an exhausted ledger can rotate, while an exact + // identity reuse remains forbidden within the current ledger. + return Err(self.fail_closed(error)); + } + // The MCP endpoint is session-lifetime and cannot authenticate which + // provider turn originated a late HTTP callback. Reap the old sidecar + // and provider before releasing its call-ID tombstones, then resume + // the same verified persistent session in a fresh process generation. + let provider_restarted = self.state.has_settled_turns(); + if provider_restarted { + if let Err(error) = self.restart_idle_provider() { + return Err(self.fail_closed(error)); + } + } + if rotate_turn_identity_ledger { + if let Err(error) = self + .state + .rotate_settled_turn_identities_after_provider_restart() + { + return Err(self.fail_closed(error)); + } + } + // Prepare cloned receipt epochs before asking the replacement sidecar + // to start work, then publish them only after both the provider and + // reducer accept the new turn. + let mut next_tool_bridge = self.tool_bridge.clone(); + let dynamic_preparation = if provider_restarted { + next_tool_bridge.prepare_turn_after_provider_restart() + } else { + next_tool_bridge.prepare_turn() + }; + if let Err(error) = dynamic_preparation { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX dynamic tool receipt rotation failed: {error}" + )))); + } + let mut next_reserved_tool_bridge = self.reserved_tool_bridge.clone(); + let reserved_preparation = if provider_restarted { + next_reserved_tool_bridge.prepare_turn_after_provider_restart() + } else { + next_reserved_tool_bridge.prepare_turn() + }; + if let Err(error) = reserved_preparation { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX reserved tool receipt rotation failed: {error}" + )))); + } + let response = match self.transport.request( + GeneratedAcpxSidecarCommand::TurnStart, + json!({"turnId":turn_id,"message":message}), + ) { + Ok(response) => response, + Err(error) => return Err(self.fail_closed(error)), + }; + if response.get("turnId").and_then(Value::as_str) != Some(turn_id) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX sidecar did not confirm the requested turn", + ))); + } + if let Err(error) = self.state.begin_turn(turn_id) { + return Err(self.fail_closed(error)); + } + self.tool_bridge = next_tool_bridge; + self.reserved_tool_bridge = next_reserved_tool_bridge; + Ok(response) + } + + pub fn interrupt_turn( + &mut self, + turn_id: &str, + reason: &str, + ) -> Result { + self.ensure_open()?; + validate_text(turn_id, 160, "ACPX turn id")?; + if self.state.active_turn_id() != Some(turn_id) { + return Err(LocalRunnerError::invalid( + "ACPX interruption named a stale or inactive turn", + )); + } + let response = match self.transport.request( + GeneratedAcpxSidecarCommand::TurnCancel, + json!({"turnId":turn_id,"reason":bounded_reason(reason)}), + ) { + Ok(response) => response, + Err(error) => return Err(self.fail_closed(error)), + }; + if response.get("cancelled").and_then(Value::as_bool) != Some(true) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX sidecar did not confirm turn cancellation", + ))); + } + Ok(response) + } + + pub fn poll_event( + &mut self, + timeout: Duration, + ) -> Result>, LocalRunnerError> { + self.ensure_open()?; + let event = match self.transport.poll_event(timeout) { + Ok(event) => event, + Err(error) => return Err(self.fail_closed(error)), + }; + let Some(event) = event else { + return Ok(None); + }; + let mut next_state = self.state.clone(); + let events = match next_state.accept_event(&event) { + Ok(events) => events, + Err(error) => return Err(self.fail_closed(error)), + }; + let mut next_bridge = self.tool_bridge.clone(); + let mut next_reserved_bridge = self.reserved_tool_bridge.clone(); + let mut reconciled_events = Vec::with_capacity(events.len()); + for event in events { + let mut expose_event = true; + match &event { + AcpxProviderStateEvent::ToolCall { + call_id, + operation_id, + input, + } => { + let bridge = if is_reserved_terminal_operation(operation_id) { + if let Err(error) = validate_reserved_terminal_value(operation_id, input) { + return Err(self.fail_closed(error)); + } + // These built-ins are authorized by the same ledger as + // dynamic tools, but the server dispatcher must never + // execute them as ordinary semantic operations. + expose_event = false; + if next_bridge.has_call_receipt(call_id) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX reused a dynamic call id for a reserved terminal invocation", + ))); + } + &mut next_reserved_bridge + } else { + if next_reserved_bridge.has_call_receipt(call_id) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX reused a reserved call id for a dynamic tool invocation", + ))); + } + &mut next_bridge + }; + if let Err(error) = + bridge.begin_call(call_id.clone(), operation_id.clone(), input.clone()) + { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX provider tool authorization failed: {error}" + )))); + } + } + AcpxProviderStateEvent::SemanticResult(result) => { + if is_reserved_terminal_result(result) { + if let Err(error) = validate_reserved_terminal_result(&next_state, result) { + return Err(self.fail_closed(error)); + } + if let Err(error) = + next_reserved_bridge.apply_result(crate::provider_bridge::ToolResult { + call_id: result.call_id.clone(), + operation_id: result.operation_id.clone(), + result: result.result.clone(), + is_error: !result.ok, + }) + { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX reserved terminal result reconciliation failed: {error}" + )))); + } + } else { + let replayed = next_bridge.has_completed_call(&result.call_id); + if let Err(error) = + next_bridge.apply_result(crate::provider_bridge::ToolResult { + call_id: result.call_id.clone(), + operation_id: result.operation_id.clone(), + result: result.result.clone(), + is_error: !result.ok, + }) + { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX provider tool result reconciliation failed: {error}" + )))); + } + if replayed { + expose_event = false; + } + } + if next_state.pending_tool(&result.call_id).is_some() { + if let Err(error) = + next_state.complete_tool(&result.call_id, &result.operation_id) + { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX provider tool completion reconciliation failed: {error}" + )))); + } + } + } + AcpxProviderStateEvent::TurnTerminal { .. } => { + let settlements = match next_bridge.settle_turn("acpx_turn_settled") { + Ok(settlements) => settlements, + Err(error) => { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX provider tool settlement failed: {error}" + )))); + } + }; + let reserved_settlements = match next_reserved_bridge + .settle_turn("acpx_reserved_terminal_unsettled") + { + Ok(settlements) => settlements, + Err(error) => { + return Err(self.fail_closed(LocalRunnerError::invalid(format!( + "ACPX reserved terminal settlement failed: {error}" + )))); + } + }; + if !reserved_settlements.is_empty() { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX turn terminated before its reserved terminal invocation produced a correlated result", + ))); + } + // `accept_event` clears the candidate reducer's pending + // tools while the bridge clones settle the corresponding + // calls above. Prove both halves reached the same terminal + // state before committing any of them to the reusable + // session. + if next_state.has_pending_tools() + || next_bridge.pending_calls().next().is_some() + || next_reserved_bridge.pending_calls().next().is_some() + { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX terminal settlement left provider tool state inconsistent", + ))); + } + reconciled_events.extend( + settlements + .into_iter() + .map(AcpxProviderStateEvent::ToolResult), + ); + } + _ => {} + } + if expose_event { + reconciled_events.push(event); + } + } + self.state = next_state; + self.tool_bridge = next_bridge; + self.reserved_tool_bridge = next_reserved_bridge; + Ok(Some(reconciled_events)) + } + pub fn shutdown(&mut self, reason: &str) -> Result<(), LocalRunnerError> { if self.closed { return self.terminate_transport(); @@ -235,6 +548,170 @@ impl AcpxProviderSession { self.transport_terminated = true; Ok(()) } + + fn ensure_open(&self) -> Result<(), LocalRunnerError> { + if self.closed { + return Err(LocalRunnerError::invalid("ACPX provider session is closed")); + } + Ok(()) + } + + fn restart_idle_provider(&mut self) -> Result<(), LocalRunnerError> { + let suspended = self.transport.request( + GeneratedAcpxSidecarCommand::SessionSuspend, + json!({"reason":"ACPX turn receipt epoch rotation"}), + )?; + verify_suspend_response(&suspended, &self.identity)?; + self.transport.shutdown()?; + self.transport_terminated = true; + + let mut restart_config = self.config.clone(); + restart_config.expected_identity = Some(self.identity.clone()); + let mut replacement = AcpxSidecarTransport::start(&restart_config.transport)?; + let (replacement_identity, _) = match bootstrap(&mut replacement, &restart_config) { + Ok(value) => value, + Err(error) => { + return Err(self.reject_replacement(replacement, error)); + } + }; + if replacement_identity != self.identity { + return Err(self.reject_replacement( + replacement, + LocalRunnerError::invalid( + "ACPX replacement provider changed its persistent session identity", + ), + )); + } + self.transport = replacement; + self.transport_terminated = false; + Ok(()) + } + + fn reject_replacement( + &mut self, + mut replacement: AcpxSidecarTransport, + error: LocalRunnerError, + ) -> LocalRunnerError { + let cleanup = replacement.shutdown(); + if cleanup.is_err() { + // Keep the exact failed generation reachable so fail_closed can + // retry its process-group termination instead of dropping the + // only remaining cleanup authority. + self.transport = replacement; + self.transport_terminated = false; + } + with_cleanup_error(error, cleanup) + } + + fn fail_closed(&mut self, error: LocalRunnerError) -> LocalRunnerError { + self.closed = true; + with_cleanup_error(error, self.terminate_transport()) + } +} + +fn is_reserved_terminal_result(result: &crate::acpx_provider_state::AcpxSemanticResult) -> bool { + 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, +) -> Result<(), LocalRunnerError> { + if !result.ok { + return Err(LocalRunnerError::invalid( + "ACPX reserved semantic result reported a failed outcome", + )); + } + let pending = state.pending_tool(&result.call_id).ok_or_else(|| { + LocalRunnerError::invalid( + "ACPX reserved semantic result has no authorized pending invocation", + ) + })?; + if pending.operation_id != result.operation_id || pending.input_digest != result.result_digest { + return Err(LocalRunnerError::invalid( + "ACPX reserved semantic result does not match its authorized invocation", + )); + } + validate_reserved_terminal_value(&result.operation_id, &result.result) +} + +fn validate_reserved_terminal_value( + operation_id: &str, + value: &Value, +) -> Result<(), LocalRunnerError> { + validate_prp_run_result(value)?; + let disposition = value.get("reportedWorkDisposition").and_then(Value::as_str); + let disposition_matches = match operation_id { + PRP_BLOCK_TOOL_NAME => disposition == Some("blocked"), + PRP_COMPLETION_TOOL_NAME => { + matches!(disposition, Some("done" | "needs_review" | "yielded")) + } + _ => false, + }; + if !disposition_matches { + return Err(LocalRunnerError::invalid( + "ACPX reserved semantic result disposition does not match its operation", + )); + } + Ok(()) +} + +fn reserved_terminal_tool_bridge() -> Result { + let result_schema: Value = serde_json::from_str(include_str!( + "../../../../protocol/schemas/result.schema.json" + )) + .map_err(|_| LocalRunnerError::invalid("embedded Paperclip result schema is invalid"))?; + let operations = vec![ + AuthorizedTool { + operation_id: PRP_COMPLETION_TOOL_NAME.to_owned(), + version: 1, + description: "Return the authoritative Paperclip completion result.".to_owned(), + input_schema: result_schema.clone(), + response_schema: result_schema.clone(), + }, + AuthorizedTool { + operation_id: PRP_BLOCK_TOOL_NAME.to_owned(), + version: 1, + description: "Return the authoritative Paperclip blocked result.".to_owned(), + input_schema: result_schema.clone(), + response_schema: result_schema, + }, + ]; + let catalog_digest = authorized_tool_catalog_digest(&operations).map_err(|error| { + LocalRunnerError::invalid(format!("ACPX reserved terminal tools are invalid: {error}")) + })?; + let mut bridge = ProviderToolBridge::default(); + bridge + .prepare(AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest, + operations, + }) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX reserved terminal tools are invalid: {error}")) + })?; + Ok(bridge) +} + +fn validate_prp_run_result(value: &Value) -> Result<(), LocalRunnerError> { + let schema: Value = serde_json::from_str(include_str!( + "../../../../protocol/schemas/result.schema.json" + )) + .map_err(|_| LocalRunnerError::invalid("embedded Paperclip result schema is invalid"))?; + let validator = jsonschema::validator_for(&schema).map_err(|_| { + LocalRunnerError::invalid("embedded Paperclip result schema cannot compile") + })?; + if !validator.is_valid(value) { + return Err(LocalRunnerError::invalid( + "ACPX reserved semantic result failed the Paperclip result schema", + )); + } + Ok(()) } impl Drop for AcpxProviderSession { @@ -363,6 +840,33 @@ fn verify_open_response( Ok(identity) } +fn verify_suspend_response( + value: &Value, + expected_identity: &AcpxProviderSessionIdentity, +) -> Result<(), LocalRunnerError> { + if value.get("suspended").and_then(Value::as_bool) != Some(true) { + return Err(LocalRunnerError::invalid( + "ACPX sidecar did not confirm provider suspension", + )); + } + let identity: AcpxProviderSessionIdentity = serde_json::from_value( + value + .get("identity") + .cloned() + .ok_or_else(|| LocalRunnerError::invalid("ACPX suspension omitted its identity"))?, + ) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX suspension identity is invalid: {error}")) + })?; + identity.validate()?; + if &identity != expected_identity { + return Err(LocalRunnerError::invalid( + "ACPX suspension changed its persistent session identity", + )); + } + Ok(()) +} + fn validate_text(value: &str, max_chars: usize, label: &str) -> Result<(), LocalRunnerError> { if value.trim().is_empty() || value.chars().count() > max_chars @@ -385,6 +889,18 @@ fn bounded_reason(value: &str) -> String { value.chars().take(4_000).collect() } +fn validate_turn_message(value: &str) -> Result<(), LocalRunnerError> { + if value.trim().is_empty() + || value.len() > MAX_SYSTEM_INSTRUCTIONS_BYTES + || value.contains('\0') + { + return Err(LocalRunnerError::invalid( + "ACPX turn message exceeds its bounded contract", + )); + } + Ok(()) +} + fn with_cleanup_error( error: LocalRunnerError, cleanup: Result<(), LocalRunnerError>, 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 efc2938daa..fd223888e0 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 @@ -8,6 +8,7 @@ use crate::acpx_event_payload::{ use crate::acpx_event_scope::AcpxEventScope; 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}; const MAX_ASSISTANT_TEXT_BYTES: usize = 1024 * 1024; @@ -15,11 +16,14 @@ 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"; #[derive(Clone, Debug, PartialEq)] pub struct AcpxPendingTool { pub operation_id: String, pub input: Value, + pub(crate) input_digest: String, input_bytes: usize, } @@ -27,7 +31,9 @@ pub struct AcpxPendingTool { pub struct AcpxSemanticResult { pub call_id: String, pub operation_id: String, + pub ok: bool, pub result: Value, + pub(crate) result_digest: String, } #[derive(Clone, Debug, PartialEq)] @@ -38,6 +44,7 @@ pub enum AcpxProviderStateEvent { operation_id: String, input: Value, }, + ToolResult(ToolResult), PermissionRequest { request_id: String, kind: String, @@ -115,6 +122,33 @@ impl AcpxProviderState { self.scope.active_turn_id() } + pub(crate) fn has_settled_turns(&self) -> bool { + self.scope.has_settled_turns() + } + + pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> { + self.scope.validate_new_turn_identity(turn_id) + } + + pub(crate) fn validate_new_turn_identity_for_provider_restart( + &self, + turn_id: &str, + ) -> Result<(), LocalRunnerError> { + self.scope + .validate_new_turn_identity_for_provider_restart(turn_id) + } + + pub(crate) fn settled_turn_identity_capacity_reached(&self) -> bool { + self.scope.settled_turn_identity_capacity_reached() + } + + pub(crate) fn rotate_settled_turn_identities_after_provider_restart( + &mut self, + ) -> Result<(), LocalRunnerError> { + self.scope + .rotate_settled_turn_identities_after_provider_restart() + } + pub fn begin_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { if self.scope.active_turn_id().is_some() || !self.pending_tools.is_empty() @@ -147,7 +181,14 @@ impl AcpxProviderState { kind, tool_operation, payload, - } => self.accept_runtime_event(event, kind, tool_operation, payload), + semantic_result_digest, + } => self.accept_runtime_event( + event, + kind, + tool_operation, + payload, + semantic_result_digest, + ), AcpxEventPayload::PermissionRequested { request_id, kind, @@ -193,6 +234,7 @@ impl AcpxProviderState { call_id, operation_id, input, + input_digest, } => { let input_bytes = value_bytes(&input)?; if self.pending_tools.len() >= MAX_PENDING_TOOLS @@ -215,6 +257,7 @@ impl AcpxProviderState { AcpxPendingTool { operation_id: operation_id.clone(), input: input.clone(), + input_digest, input_bytes, }, ); @@ -261,6 +304,10 @@ impl AcpxProviderState { self.pending_tools.get(call_id) } + pub fn has_pending_tools(&self) -> bool { + !self.pending_tools.is_empty() + } + pub fn complete_tool( &mut self, call_id: &str, @@ -315,6 +362,7 @@ impl AcpxProviderState { kind: AcpxRuntimeEventKind, tool_operation: Option<&'static str>, payload: Value, + semantic_result_digest: Option, ) -> Result, LocalRunnerError> { if kind == AcpxRuntimeEventKind::Thinking { if self.thinking_active { @@ -351,21 +399,36 @@ impl AcpxProviderState { .and_then(Value::as_str) .expect("decoded ACPX semantic result has an operation id") .to_owned(), + ok: payload + .get("ok") + .and_then(Value::as_bool) + .expect("decoded ACPX semantic result has an outcome"), result: payload .get("result") .expect("decoded ACPX semantic result has a result") .clone(), + result_digest: semantic_result_digest + .expect("decoded ACPX semantic result has a raw correlation digest"), }; - return match self.semantic_result.as_ref() { - None => { - self.semantic_result = Some(result.clone()); - Ok(vec![AcpxProviderStateEvent::SemanticResult(result)]) - } - Some(existing) if existing == &result => Ok(Vec::new()), - Some(_) => Err(LocalRunnerError::invalid( - "ACPX emitted conflicting semantic results for one turn", - )), - }; + if matches!( + result.operation_id.as_str(), + PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME + ) { + return match self.semantic_result.as_ref() { + None => { + self.semantic_result = Some(result.clone()); + Ok(vec![AcpxProviderStateEvent::SemanticResult(result)]) + } + Some(existing) if existing == &result => Ok(Vec::new()), + Some(_) => Err(LocalRunnerError::invalid( + "ACPX emitted conflicting terminal semantic results for one turn", + )), + }; + } + // Dynamic results are independently authorized and deduplicated + // by call ID in ProviderToolBridge. They must not compete for the + // turn-wide terminal-result slot. + return Ok(vec![AcpxProviderStateEvent::SemanticResult(result)]); } let turn_id = event .turn_id 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 8aa20a1922..6b03af391d 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 @@ -1,4 +1,5 @@ use std::io::{self, BufRead, Write}; +use std::time::Duration; use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION; use serde_json::{json, Value}; @@ -79,8 +80,361 @@ fn run() -> Result<(), Box> { eprintln!("amber-signal-7305"); std::process::exit(9); } - "bootstrap" | "bootstrap-wrong-model" | "bootstrap-wrong-run" => { + "bootstrap" + | "bootstrap-wrong-model" + | "bootstrap-wrong-run" + | "turns" + | "turns-wrong-turn" + | "turns-wrong-cancel" + | "turns-wrong-scope" + | "turns-tool" + | "turns-tool-terminal" + | "turns-reused-tool-id-terminal" + | "turns-late-tool-after-suspend" + | "turns-tool-result-terminal" + | "turns-tool-error-result-terminal" + | "turns-multiple-tool-results-terminal" + | "turns-reserved-result-terminal" + | "turns-reserved-yielded-terminal" + | "turns-reserved-block-terminal" + | "turns-sensitive-reserved-result-terminal" + | "turns-mismatched-sensitive-reserved-result-terminal" + | "turns-invalid-reserved-block-terminal" + | "turns-uncorrelated-reserved-result-terminal" + | "turns-mismatched-reserved-result-terminal" + | "turns-unauthorized-tool" => { write_json(&mut stdout, &bootstrap_success(id, command, &request, mode))?; + let params = request.get("params").unwrap_or(&Value::Null); + let turn_id = params + .get("turnId") + .and_then(Value::as_str) + .unwrap_or("missing"); + let tool_call_id = if matches!( + mode, + "turns-reused-tool-id-terminal" | "turns-late-tool-after-suspend" + ) { + "call-reused".to_owned() + } else { + turn_id + .strip_prefix("turn-") + .map(|suffix| format!("call-{suffix}")) + .unwrap_or_else(|| "call-1".to_owned()) + }; + if command == "turn.start" && matches!(mode, "turns" | "turns-wrong-scope") { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + if mode == "turns-wrong-scope" { + "wrong-run" + } else { + "run-1" + }, + turn_id, + json!({"type":"text_delta","text":"hello"}), + )?; + next_sequence += 1; + } + if command == "turn.start" + && matches!( + mode, + "turns-tool" + | "turns-tool-terminal" + | "turns-reused-tool-id-terminal" + | "turns-late-tool-after-suspend" + | "turns-tool-result-terminal" + | "turns-tool-error-result-terminal" + | "turns-unauthorized-tool" + ) + { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + "run-1", + turn_id, + json!({ + "callId":tool_call_id.clone(), + "operationId":if matches!(mode, "turns-tool" | "turns-tool-terminal" | "turns-reused-tool-id-terminal" | "turns-late-tool-after-suspend" | "turns-tool-result-terminal" | "turns-tool-error-result-terminal") { "issues.read" } else { "issues.delete" }, + "input":{"id":"issue-1"}, + }), + )?; + next_sequence += 1; + } + if command == "turn.start" + && matches!( + mode, + "turns-tool-result-terminal" | "turns-tool-error-result-terminal" + ) + { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + "run-1", + turn_id, + json!({ + "type":"semantic_result", + "callId":tool_call_id, + "operationId":"issues.read", + "ok":mode == "turns-tool-result-terminal", + "result":if mode == "turns-tool-result-terminal" { json!({"id":"issue-1"}) } else { json!({"error":{"code":"tool_failed"}}) }, + }), + )?; + next_sequence += 1; + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + "run-1", + turn_id, + json!({"status":"completed"}), + )?; + next_sequence += 1; + } + if command == "turn.start" + && matches!( + mode, + "turns-tool-terminal" + | "turns-reused-tool-id-terminal" + | "turns-late-tool-after-suspend" + ) + { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + "run-1", + turn_id, + json!({"status":"completed"}), + )?; + next_sequence += 1; + } + if command == "turn.start" && mode == "turns-multiple-tool-results-terminal" { + for index in 1..=2 { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + "run-1", + turn_id, + json!({ + "callId":format!("call-{index}"), + "operationId":"issues.read", + "input":{"id":format!("issue-{index}")}, + }), + )?; + next_sequence += 1; + } + for index in 1..=2 { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + "run-1", + turn_id, + json!({ + "type":"semantic_result", + "callId":format!("call-{index}"), + "operationId":"issues.read", + "ok":true, + "result":{"id":format!("issue-{index}")}, + }), + )?; + next_sequence += 1; + } + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + "run-1", + turn_id, + json!({"status":"completed"}), + )?; + next_sequence += 1; + } + if command == "turn.start" + && matches!( + mode, + "turns-reserved-result-terminal" + | "turns-reserved-yielded-terminal" + | "turns-reserved-block-terminal" + | "turns-sensitive-reserved-result-terminal" + | "turns-mismatched-sensitive-reserved-result-terminal" + | "turns-invalid-reserved-block-terminal" + | "turns-uncorrelated-reserved-result-terminal" + | "turns-mismatched-reserved-result-terminal" + ) + { + let (operation_id, result) = if matches!( + mode, + "turns-reserved-block-terminal" | "turns-invalid-reserved-block-terminal" + ) { + ( + "paperclip_block", + json!({ + "schema":"paperclip.run_result.v1", + "reportedWorkDisposition":"blocked", + "summary":"Reserved blocker accepted.", + "completionClaim":{ + "contractRevision":"acpx-provider-turns-v1", + "objectiveSatisfied":false, + "criteria":[], + "remainingWork":[{ + "description":"Wait for external input.", + "blocksCompletion":true, + }], + }, + "evidence":[], + "verification":[], + "blocker":{ + "reasonCode":"external_input", + "owner":if mode == "turns-invalid-reserved-block-terminal" { + json!({}) + } else { + json!({"kind":"external","name":"External input"}) + }, + "unblockAction":"Provide the required input.", + "scope":"current_track", + }, + "attentionRequests":[], + "artifacts":[], + }), + ) + } else if mode == "turns-reserved-yielded-terminal" { + ( + "paperclip_finish", + json!({ + "schema":"paperclip.run_result.v1", + "reportedWorkDisposition":"yielded", + "summary":"Reserved continuation accepted.", + "completionClaim":{ + "contractRevision":"acpx-provider-turns-v1", + "objectiveSatisfied":false, + "criteria":[], + "remainingWork":[], + }, + "evidence":[], + "verification":[], + "continuation":{ + "kind":"same_agent", + "summary":"Continue the current run.", + "idempotencyKey":"continuation-1", + }, + "attentionRequests":[], + "artifacts":[], + }), + ) + } else { + ( + "paperclip_finish", + json!({ + "schema":"paperclip.run_result.v1", + "reportedWorkDisposition":"done", + "summary":if matches!( + mode, + "turns-sensitive-reserved-result-terminal" + | "turns-mismatched-sensitive-reserved-result-terminal" + ) { + "token=matching-sensitive-value" + } else { + "Reserved completion accepted." + }, + "completionClaim":{ + "contractRevision":"acpx-provider-turns-v1", + "objectiveSatisfied":true, + "criteria":[], + "remainingWork":[], + }, + "evidence":[], + "verification":[], + "attentionRequests":[], + "artifacts":[], + }), + ) + }; + if mode != "turns-uncorrelated-reserved-result-terminal" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + "run-1", + turn_id, + json!({ + "callId":"call-finish", + "operationId":operation_id, + "input":result.clone(), + }), + )?; + next_sequence += 1; + } + let semantic_result = if mode == "turns-mismatched-reserved-result-terminal" { + let mut changed = result.clone(); + changed["summary"] = json!("A different terminal result."); + changed + } else if mode == "turns-mismatched-sensitive-reserved-result-terminal" { + let mut changed = result.clone(); + changed["summary"] = json!("token=different-sensitive-value"); + changed + } else { + result + }; + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + "run-1", + turn_id, + json!({ + "type":"semantic_result", + "callId":"call-finish", + "operationId":operation_id, + "ok":true, + "result":semantic_result, + }), + )?; + next_sequence += 1; + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + "run-1", + turn_id, + json!({"status":"completed"}), + )?; + next_sequence += 1; + } + if command == "session.suspend" && mode == "turns-late-tool-after-suspend" { + // Simulate a session-lifetime callback that wakes after + // suspension and reads the next mutable turn binding. + // runner-core must reap this process before starting that + // turn, so the relabeled event can never cross authority. + std::thread::sleep(Duration::from_millis(50)); + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + "run-1", + "turn-2", + json!({ + "callId":"call-reused", + "operationId":"issues.delete", + "input":{"source":"late-old-turn"}, + }), + )?; + next_sequence += 1; + } + if command == "turn.cancel" && mode == "turns" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.turn_terminal", + "run-1", + turn_id, + json!({"status":"interrupted"}), + )?; + next_sequence += 1; + } } "happy" => { write_event(&mut stdout, next_sequence)?; @@ -134,6 +488,25 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val "runId": if mode == "bootstrap-wrong-run" { "wrong-run" } else { params.get("runId").and_then(Value::as_str).unwrap_or("missing") }, "catalogRevision": params.get("catalogRevision"), }), + "turn.start" => json!({ + "turnId": if mode == "turns-wrong-turn" { "wrong-turn" } else { params.get("turnId").and_then(Value::as_str).unwrap_or("missing") }, + }), + "turn.cancel" => json!({"cancelled":mode != "turns-wrong-cancel"}), + "session.suspend" => json!({ + "suspended":true, + "identity": { + "kind": "acpx", + "normalizedSessionId": "session-1", + "acpxRecordId": "record-1", + "backendSessionId": "backend-1", + "agentSessionId": "agent-1", + "profileDigest": format!("sha256:{}", "1".repeat(64)), + "workspaceDigest": format!("sha256:{}", "2".repeat(64)), + "requestedModel": "gpt-5.6-sol", + "effectiveModel": "gpt-5.6-sol", + "permissionMode": "approve-reads", + }, + }), "session.close" => json!({"closed":true}), _ => json!({"command":command,"params":params}), }; @@ -145,6 +518,27 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val }) } +fn write_turn_event( + output: &mut impl Write, + sequence: u64, + event_type: &str, + run_id: &str, + turn_id: &str, + payload: Value, +) -> io::Result<()> { + write_json( + output, + &json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence": sequence, + "eventType": event_type, + "runId": run_id, + "turnId": turn_id, + "payload": payload, + }), + ) +} + 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/provider_bridge.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_bridge.rs index a01d35172b..24f87e2729 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_bridge.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_bridge.rs @@ -520,6 +520,20 @@ impl ProviderToolBridge { Ok(()) } + /// Start a receipt epoch after the process that owned the previous epoch + /// has been reaped and a replacement provider generation is established. + /// Providers that continue in the same process must retain their durable + /// call-ID tombstones through ordinary `prepare_turn` transitions. + pub(crate) fn prepare_turn_after_provider_restart( + &mut self, + ) -> Result<(), ProviderBridgeError> { + self.prepare_turn()?; + self.settled_call_ids.clear(); + self.settled_call_filter = DurableReplayFilter::default(); + self.durable_run_receipt_limit_reached = false; + Ok(()) + } + #[cfg(test)] pub(crate) fn retained_result_bytes_for_test(&self) -> usize { self.retained_result_bytes @@ -550,6 +564,10 @@ impl ProviderToolBridge { self.completed.contains_key(call_id) || self.has_settled_call_id(call_id) } + pub(crate) fn has_call_receipt(&self, call_id: &str) -> bool { + self.pending.contains_key(call_id) || self.has_completed_call(call_id) + } + pub fn begin_call( &mut self, call_id: String, @@ -1464,6 +1482,37 @@ mod tests { ); } + #[test] + fn cross_bridge_receipt_lookup_reserves_pending_call_ids() { + let operation = AuthorizedTool { + operation_id: "get_task_context".to_owned(), + version: 1, + description: "Read the active task context.".to_owned(), + input_schema: json!({"type": "object"}), + response_schema: json!({"type": "object"}), + }; + let mut bridge = ProviderToolBridge::default(); + bridge + .prepare(AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)) + .unwrap(), + operations: vec![operation], + }) + .unwrap(); + + assert!(!bridge.has_call_receipt("shared-call")); + bridge + .begin_call( + "shared-call".to_owned(), + "get_task_context".to_owned(), + json!({}), + ) + .unwrap(); + assert!(bridge.has_call_receipt("shared-call")); + } + #[test] fn exact_completed_results_are_bounded_without_changing_replay() { let operation = AuthorizedTool { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_payload.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_payload.rs index 8112fc2e32..0b1f64a583 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_payload.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_event_payload.rs @@ -47,7 +47,7 @@ fn decodes_every_admitted_runtime_event_shape() { AcpxRuntimeEventKind::ToolCall, ), ( - json!({"type": "semantic_result", "callId": "call-1", "operationId": "paperclip_finish", "result": {"ok": true}}), + json!({"type": "semantic_result", "callId": "call-1", "operationId": "paperclip_finish", "ok": true, "result": {"ok": true}}), AcpxRuntimeEventKind::SemanticResult, ), ( @@ -103,6 +103,7 @@ fn admits_sidecar_replacement_scalars_in_tool_fields() { kind: AcpxRuntimeEventKind::ToolCall, tool_operation: Some("edit"), payload, + .. } if payload["toolCallId"].as_str() == Some("tool-\u{fffd}") && payload["kind"].as_str() == Some(kind.as_str()) && payload["status"].as_str() == Some("pend\u{fffd}ing") @@ -135,6 +136,7 @@ fn retains_full_kind_classification_from_a_bounded_sidecar_frame() { kind: AcpxRuntimeEventKind::ToolCall, tool_operation: Some("edit"), payload, + .. } if payload["type"] == "tool_call" && payload["kind"].as_str().is_some_and(|value| value.chars().count() == 4_000) )); @@ -186,7 +188,7 @@ fn decodes_tool_and_permission_requests_after_scope_validation() { .unwrap(); assert!(matches!( tool, - AcpxEventPayload::ToolCalled { call_id, operation_id, input } + AcpxEventPayload::ToolCalled { call_id, operation_id, input, .. } if call_id == "call-1" && operation_id == "get_issue" && input["apiToken"] == "[REDACTED]" 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 bd97b8a4b4..d916c58912 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 @@ -131,6 +131,43 @@ fn binds_and_clears_one_turn_idempotently() { scope.clear_turn("turn-1").unwrap(); assert_eq!(scope.active_turn_id(), None); assert!(scope.clear_turn("turn-1").is_err()); + let reused = scope.bind_turn("turn-1").unwrap_err(); + assert!(reused.to_string().contains("reused a settled turn")); +} + +#[test] +fn rejects_late_events_after_rotating_to_a_distinct_turn_identity() { + let mut scope = AcpxEventScope::new("run-1").unwrap(); + scope.bind_turn("turn-1").unwrap(); + scope.clear_turn("turn-1").unwrap(); + scope.bind_turn("turn-2").unwrap(); + + let late = scope + .validate_event(&event( + GeneratedAcpxSidecarEventType::RuntimeToolCalled, + Some("run-1"), + Some("turn-1"), + )) + .unwrap_err(); + assert!(late.to_string().contains("stale turn"), "{late}"); +} + +#[test] +fn bounds_settled_turn_identity_retention() { + let mut scope = AcpxEventScope::new("run-1").unwrap(); + for index in 0..4_096 { + let turn_id = format!("turn-{index}"); + scope.bind_turn(&turn_id).unwrap(); + scope.clear_turn(&turn_id).unwrap(); + } + + let exhausted = scope.bind_turn("turn-overflow").unwrap_err(); + assert!( + exhausted + .to_string() + .contains("exhausted its settled turn identity capacity"), + "{exhausted}" + ); } #[test] diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_events.rs index e0a27fbee0..5f736090e7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_events.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_events.rs @@ -95,6 +95,7 @@ fn preserves_tool_operation_authority_across_payload_sanitization() { kind, tool_operation, payload, + .. } = decoded else { panic!("runtime event must decode as runtime payload"); 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 b2b952013f..59dca6f390 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 @@ -167,6 +167,7 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() { "type":"semantic_result", "callId":"finish-1", "operationId":"paperclip_finish", + "ok":true, "result":{"reportedWorkDisposition":"done"} }), ); @@ -189,6 +190,7 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() { "type":"semantic_result", "callId":"finish-2", "operationId":"paperclip_finish", + "ok":true, "result":{"reportedWorkDisposition":"done"} }), ); @@ -196,6 +198,32 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() { assert_eq!(state.semantic_result().unwrap().call_id, "finish-1"); } +#[test] +fn keeps_dynamic_results_independent_from_terminal_result_authority() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + for index in 1..=2 { + let result = event( + index, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({ + "type":"semantic_result", + "callId":format!("call-{index}"), + "operationId":"issues.read", + "ok":true, + "result":{"id":format!("issue-{index}")} + }), + ); + assert!(matches!( + &state.accept_event(&result).unwrap()[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == format!("call-{index}") + )); + } + assert!(state.semantic_result().is_none()); +} + #[test] fn validates_scope_before_mutating_pending_state() { 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 new file mode 100644 index 0000000000..d2eba6cf2a --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -0,0 +1,594 @@ +use std::path::PathBuf; +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, +}; +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, + } +} + +#[test] +fn starts_interrupts_and_settles_one_scoped_turn() { + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + assert!(session + .poll_event(Duration::from_millis(1)) + .unwrap() + .is_none()); + let response = session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + assert_eq!(response["turnId"], "turn-1"); + assert_eq!(session.state().active_turn_id(), Some("turn-1")); + + 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-1", "Paperclip interruption") + .unwrap(); + assert_eq!(session.state().active_turn_id(), Some("turn-1")); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + assert_eq!(session.state().active_turn_id(), None); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn rejects_invalid_turn_inputs_without_mutating_the_session() { + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + let other_directory = std::env::current_dir().unwrap(); + assert!(session + .start_turn("turn-1", "Please help", &other_directory) + .unwrap_err() + .to_string() + .contains("immutable session workspace")); + assert!(session + .start_turn("turn-1", "bad\0message", &std::env::temp_dir()) + .unwrap_err() + .to_string() + .contains("bounded contract")); + assert_eq!(session.state().active_turn_id(), None); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + 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(); + let error = session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap_err() + .to_string(); + assert!(error.contains("confirm the requested turn"), "{error}"); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_when_cancellation_is_not_confirmed() { + let mut session = AcpxProviderSession::start(&config("turns-wrong-cancel")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let error = session + .interrupt_turn("turn-1", "stop") + .unwrap_err() + .to_string(); + assert!(error.contains("confirm turn cancellation"), "{error}"); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_when_a_polled_event_violates_run_scope() { + let mut session = AcpxProviderSession::start(&config("turns-wrong-scope")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!(error.contains("stale run"), "{error}"); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn admits_only_catalog_authorized_tool_calls() { + let mut session = AcpxProviderSession::start(&config("turns-tool")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &events[0], + AcpxProviderStateEvent::ToolCall { operation_id, .. } + if operation_id == "issues.read" + )); + assert_eq!( + session.state().pending_tool("call-1").unwrap().operation_id, + "issues.read" + ); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn returns_pending_tool_cancellations_before_the_terminal_event() { + let mut session = AcpxProviderSession::start(&config("turns-tool-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. })); + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &terminal[0], + AcpxProviderStateEvent::ToolResult(result) + if result.call_id == "call-1" + && result.operation_id == "issues.read" + && result.is_error + && result.result["error"]["code"] == "acpx_turn_settled" + )); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + + assert!(session.state().pending_tool("call-1").is_none()); + session + .start_turn("turn-2", "Please continue", &std::env::temp_dir()) + .expect("terminal settlement must leave the session reusable"); + let next_tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &next_tool[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. } + if call_id == "call-2" && operation_id == "issues.read" + )); + let next_terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + next_terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2" + )); + assert!(session.state().pending_tool("call-2").is_none()); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn rotates_settled_tool_receipts_between_reusable_turns() { + let mut session = AcpxProviderSession::start(&config("turns-reused-tool-id-terminal")).unwrap(); + + for turn_id in ["turn-1", "turn-2"] { + session + .start_turn(turn_id, "Please continue", &std::env::temp_dir()) + .unwrap(); + let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &tool[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. } + if call_id == "call-reused" && operation_id == "issues.read" + )); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &terminal[0], + AcpxProviderStateEvent::ToolResult(result) + if result.call_id == "call-reused" + && result.result["error"]["code"] == "acpx_turn_settled" + )); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } + if settled == turn_id + )); + } + session.shutdown("test complete").unwrap(); + + let mut reserved_session = + AcpxProviderSession::start(&config("turns-reserved-result-terminal")).unwrap(); + for turn_id in ["turn-1", "turn-2"] { + reserved_session + .start_turn(turn_id, "Please continue", &std::env::temp_dir()) + .unwrap(); + assert!(reserved_session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap() + .is_empty()); + let result = reserved_session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap(); + assert!(matches!( + &result[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == "call-finish" + && result.operation_id == "paperclip_finish" + )); + let terminal = reserved_session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } + if settled == turn_id + )); + } + reserved_session.shutdown("test complete").unwrap(); +} + +#[test] +fn reaps_the_old_provider_before_admitting_a_late_tool_callback() { + let mut session = AcpxProviderSession::start(&config("turns-late-tool-after-suspend")).unwrap(); + session + .start_turn("turn-1", "Please continue", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + + session + .start_turn("turn-2", "Use a fresh provider", &std::env::temp_dir()) + .unwrap(); + let fresh = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &fresh[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, input } + if call_id == "call-reused" + && operation_id == "issues.read" + && input["id"] == "issue-1" + )); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn fails_closed_before_reusing_a_settled_turn_identity() { + let mut session = AcpxProviderSession::start(&config("turns-reused-tool-id-terminal")).unwrap(); + session + .start_turn("turn-1", "Please continue", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + + let reused = session + .start_turn("turn-1", "Do not alias old events", &std::env::temp_dir()) + .unwrap_err(); + assert!( + reused.to_string().contains("reused a settled turn"), + "{reused}" + ); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn completed_tool_results_are_not_cancelled_when_the_turn_terminates() { + let mut session = AcpxProviderSession::start(&config("turns-tool-result-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. })); + + let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &result[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == "call-1" && result.operation_id == "issues.read" && result.ok + )); + assert!(session.state().pending_tool("call-1").is_none()); + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert_eq!(terminal.len(), 1); + assert!(matches!( + &terminal[0], + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn completes_multiple_distinct_dynamic_tool_results_in_one_turn() { + let mut session = + AcpxProviderSession::start(&config("turns-multiple-tool-results-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + + for index in 1..=2 { + let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &tool[0], + AcpxProviderStateEvent::ToolCall { call_id, .. } + if call_id == &format!("call-{index}") + )); + } + for index in 1..=2 { + let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &result[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == format!("call-{index}") + && result.result["id"] == format!("issue-{index}") + )); + assert!(session + .state() + .pending_tool(&format!("call-{index}")) + .is_none()); + } + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert_eq!(terminal.len(), 1); + assert!(matches!( + &terminal[0], + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn failed_semantic_results_preserve_error_status_and_release_pending_capacity() { + let mut session = + AcpxProviderSession::start(&config("turns-tool-error-result-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. })); + + let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &result[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == "call-1" + && !result.ok + && result.result["error"]["code"] == "tool_failed" + )); + assert!(session.state().pending_tool("call-1").is_none()); + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn reserved_terminal_results_require_an_authorized_correlated_invocation() { + for (mode, operation_id, disposition) in [ + ("turns-reserved-result-terminal", "paperclip_finish", "done"), + ( + "turns-reserved-yielded-terminal", + "paperclip_finish", + "yielded", + ), + ( + "turns-reserved-block-terminal", + "paperclip_block", + "blocked", + ), + ] { + let mut session = AcpxProviderSession::start(&config(mode)).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + + let invocation = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(invocation.is_empty()); + assert_eq!( + session + .state() + .pending_tool("call-finish") + .unwrap() + .operation_id, + operation_id + ); + + let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &result[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == "call-finish" + && result.operation_id == operation_id + && result.ok + && result.result["reportedWorkDisposition"] == disposition + )); + assert!(session.state().pending_tool("call-finish").is_none()); + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + session.shutdown("test complete").unwrap(); + } +} + +#[test] +fn correlates_reserved_results_by_raw_digest_without_exposing_sensitive_values() { + let mut session = + AcpxProviderSession::start(&config("turns-sensitive-reserved-result-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + + assert!(session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap() + .is_empty()); + let result_events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &result_events[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.result["summary"] + .as_str() + .is_some_and(|summary| summary.contains("REDACTED") + && !summary.contains("matching-sensitive-value")) + )); + + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn rejects_sensitive_reserved_results_that_only_match_after_redaction() { + let mut session = AcpxProviderSession::start(&config( + "turns-mismatched-sensitive-reserved-result-terminal", + )) + .unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + assert!(session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap() + .is_empty()); + + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!( + error.contains("does not match its authorized invocation"), + "{error}" + ); + assert!(!error.contains("matching-sensitive-value"), "{error}"); + assert!(!error.contains("different-sensitive-value"), "{error}"); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_before_returning_an_uncorrelated_reserved_result() { + let mut session = + AcpxProviderSession::start(&config("turns-uncorrelated-reserved-result-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!( + error.contains("no authorized pending invocation"), + "{error}" + ); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_before_returning_a_mismatched_reserved_result() { + let mut session = + AcpxProviderSession::start(&config("turns-mismatched-reserved-result-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + assert!(session + .poll_event(Duration::from_secs(1)) + .unwrap() + .unwrap() + .is_empty()); + + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!( + error.contains("does not match its authorized invocation"), + "{error}" + ); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_before_returning_a_malformed_reserved_result() { + let mut session = + AcpxProviderSession::start(&config("turns-invalid-reserved-block-terminal")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!( + error.contains("failed the Paperclip result schema"), + "{error}" + ); + assert!(session.shutdown("already closed").is_ok()); +} + +#[test] +fn fails_closed_before_returning_an_unauthorized_tool_call() { + let mut session = AcpxProviderSession::start(&config("turns-unauthorized-tool")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + let error = session + .poll_event(Duration::from_secs(1)) + .unwrap_err() + .to_string(); + assert!(error.contains("unauthorized tool issues.delete"), "{error}"); + assert!(session.state().pending_tool("call-1").is_none()); + assert!(session.shutdown("already closed").is_ok()); +} diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index 12371e19ea..ddc5fba004 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -451,25 +451,25 @@ async function pumpTurn( currentTurnId: string, runtimeTurn: AcpxRuntimeTurn, ): Promise { + let terminal: Record; try { for await (const event of runtimeTurn.events) { emit("runtime.event", sanitizeRuntimeEvent(event), currentTurnId); } const result = await runtimeTurn.result; - emit("runtime.turn_terminal", boundedSidecarValue(result), currentTurnId); + terminal = boundedSidecarValue(result); } catch (error) { - emit( - "runtime.turn_terminal", - { - status: "failed", - error: { message: safeMessage(error), retryable: false }, - }, - currentTurnId, - ); + terminal = { + status: "failed", + error: { message: safeMessage(error), retryable: false }, + }; } finally { rejectTurnWaiters(currentTurnId, "ACPX turn became terminal"); if (turnId === currentTurnId) turnId = null; } + // A terminal frame is also runnerd's permission to recycle this provider. + // Publish it only after no callback can inherit this turn's mutable binding. + emit("runtime.turn_terminal", terminal, currentTurnId); } async function waitForTool(call: RunnerToolCall): Promise { @@ -497,22 +497,43 @@ async function waitForTool(call: RunnerToolCall): Promise { "ACPX semantic result disposition does not match its terminal operation", ); } - emit("runtime.event", { - type: "semantic_result", - callId, - operationId, - result: validation.result, - }); + // The authenticated runner bridge admitted this built-in invocation. Send + // that fact across the sidecar boundary before its locally produced result + // so runnerd can authorize and correlate the terminal claim. + emit( + "runtime.tool_called", + { + callId, + operationId, + input: validation.result, + }, + activeTurnId, + ); + emit( + "runtime.event", + { + type: "semantic_result", + callId, + operationId, + ok: true, + result: validation.result, + }, + activeTurnId, + ); return { accepted: true }; } if (tools.size >= MAX_PENDING_TOOLS) { throw new Error("ACPX pending tool limit reached"); } - emit("runtime.tool_called", { - callId, - operationId, - input: boundedSidecarValue(record(call.arguments)), - }); + emit( + "runtime.tool_called", + { + callId, + operationId, + input: boundedSidecarValue(record(call.arguments)), + }, + activeTurnId, + ); return await new Promise((settle, reject) => { const abort = () => { const pending = tools.get(callId);