diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index a055804283..3115f399a8 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -83,6 +83,12 @@ are canonicalized, and consumers must not reinterpret the display value as file-access authority. Operational semantic-result and terminal events remain reserved for the stateful adapter rather than being duplicated. +The package-local ACPX provider reducer preserves that order while it tracks one +active turn, bounded assistant text, semantic results, and pending tool or input +correlations. Terminal events flush the final assistant message first and clear +unresolved turn-scoped requests. This reducer still does not select ACPX in +runnerd. + Run the complete contract gate with: ```sh 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 new file mode 100644 index 0000000000..efc2938daa --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -0,0 +1,429 @@ +use std::collections::BTreeMap; + +use serde_json::Value; + +use crate::acpx_event_payload::{ + decode_acpx_event, AcpxEventPayload, AcpxRuntimeEventKind, AcpxTurnStatus, +}; +use crate::acpx_event_scope::AcpxEventScope; +use crate::acpx_sidecar_transport::AcpxSidecarEvent; +use crate::local_runner::LocalRunnerError; +use crate::provider_events::{normalize_acpx_runtime_event, NormalizedProviderEvent}; + +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; + +#[derive(Clone, Debug, PartialEq)] +pub struct AcpxPendingTool { + pub operation_id: String, + pub input: Value, + input_bytes: usize, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct AcpxSemanticResult { + pub call_id: String, + pub operation_id: String, + pub result: Value, +} + +#[derive(Clone, Debug, PartialEq)] +pub enum AcpxProviderStateEvent { + Activity(NormalizedProviderEvent), + ToolCall { + call_id: String, + operation_id: String, + input: Value, + }, + PermissionRequest { + request_id: String, + kind: String, + title: String, + details: Value, + }, + InputRequest { + request_id: String, + question_set: Value, + origin: Option, + }, + SemanticResult(AcpxSemanticResult), + AssistantMessage { + turn_id: String, + text: String, + }, + TurnTerminal { + turn_id: String, + status: AcpxTurnStatus, + error: Option, + }, + Process(Value), + Diagnostic { + code: String, + message: String, + }, +} + +#[derive(Clone, Debug, PartialEq)] +struct PendingInput { + value_bytes: usize, +} + +/// Reduces validated sidecar events into bounded provider state. +/// +/// Raw sidecar values enter only through `accept_event`, which applies run and +/// turn authority before payload decoding. Transport commands remain outside +/// this reducer so callers can commit a pending resolution only after the +/// corresponding sidecar request succeeds. +#[derive(Clone, Debug, PartialEq)] +pub struct AcpxProviderState { + scope: AcpxEventScope, + provider_requests: u64, + thinking_active: bool, + assistant_text: String, + pending_tools: BTreeMap, + pending_tool_input_bytes: usize, + pending_permissions: BTreeMap, + pending_inputs: BTreeMap, + pending_runtime_request_bytes: usize, + semantic_result: Option, +} + +impl AcpxProviderState { + pub fn new(run_id: impl Into) -> Result { + Ok(Self { + scope: AcpxEventScope::new(run_id)?, + provider_requests: 0, + thinking_active: false, + assistant_text: String::new(), + pending_tools: BTreeMap::new(), + pending_tool_input_bytes: 0, + pending_permissions: BTreeMap::new(), + pending_inputs: BTreeMap::new(), + pending_runtime_request_bytes: 0, + semantic_result: None, + }) + } + + pub fn run_id(&self) -> &str { + self.scope.run_id() + } + + pub fn active_turn_id(&self) -> Option<&str> { + self.scope.active_turn_id() + } + + 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() + || !self.pending_permissions.is_empty() + || !self.pending_inputs.is_empty() + { + return Err(LocalRunnerError::invalid( + "ACPX provider state cannot start a turn while work is active", + )); + } + let next_provider_requests = self + .provider_requests + .checked_add(1) + .ok_or_else(|| LocalRunnerError::invalid("ACPX provider request count is exhausted"))?; + self.scope.bind_turn(turn_id)?; + self.provider_requests = next_provider_requests; + self.thinking_active = false; + self.assistant_text.clear(); + self.semantic_result = None; + Ok(()) + } + + pub fn accept_event( + &mut self, + event: &AcpxSidecarEvent, + ) -> Result, LocalRunnerError> { + let payload = decode_acpx_event(&self.scope, event)?; + match payload { + AcpxEventPayload::Runtime { + kind, + tool_operation, + payload, + } => self.accept_runtime_event(event, kind, tool_operation, payload), + AcpxEventPayload::PermissionRequested { + request_id, + kind, + title, + details, + } => { + let value_bytes = value_bytes(&details)?; + self.admit_runtime_request(&request_id, value_bytes)?; + self.pending_permissions + .insert(request_id.clone(), value_bytes); + self.pending_runtime_request_bytes += value_bytes; + Ok(vec![AcpxProviderStateEvent::PermissionRequest { + request_id, + kind, + title, + details, + }]) + } + AcpxEventPayload::InputRequested { + request_id, + question_set, + origin, + } => { + let value_bytes = value_bytes(&question_set)?; + self.admit_runtime_request(&request_id, value_bytes)?; + if self + .pending_inputs + .insert(request_id.clone(), PendingInput { value_bytes }) + .is_some() + { + return Err(LocalRunnerError::invalid( + "ACPX reused a pending input request id", + )); + } + self.pending_runtime_request_bytes += value_bytes; + Ok(vec![AcpxProviderStateEvent::InputRequest { + request_id, + question_set, + origin, + }]) + } + AcpxEventPayload::ToolCalled { + call_id, + operation_id, + input, + } => { + let input_bytes = value_bytes(&input)?; + if self.pending_tools.len() >= MAX_PENDING_TOOLS + || self + .pending_tool_input_bytes + .checked_add(input_bytes) + .is_none_or(|bytes| bytes > MAX_PENDING_TOOL_INPUT_BYTES) + { + return Err(LocalRunnerError::invalid( + "ACPX pending tool calls exceed their bounded capacity", + )); + } + if self.pending_tools.contains_key(&call_id) { + return Err(LocalRunnerError::invalid( + "ACPX reused a pending tool call id", + )); + } + self.pending_tools.insert( + call_id.clone(), + AcpxPendingTool { + operation_id: operation_id.clone(), + input: input.clone(), + input_bytes, + }, + ); + self.pending_tool_input_bytes += input_bytes; + Ok(vec![AcpxProviderStateEvent::ToolCall { + call_id, + operation_id, + input, + }]) + } + AcpxEventPayload::TurnTerminal { status, error } => { + let turn_id = event + .turn_id + .as_deref() + .expect("a decoded terminal event has a turn binding") + .to_owned(); + self.scope.clear_turn(&turn_id)?; + self.clear_pending_requests(); + self.thinking_active = false; + let mut events = Vec::new(); + if !self.assistant_text.is_empty() { + events.push(AcpxProviderStateEvent::AssistantMessage { + turn_id: turn_id.clone(), + text: std::mem::take(&mut self.assistant_text), + }); + } + events.push(AcpxProviderStateEvent::TurnTerminal { + turn_id, + status, + error, + }); + Ok(events) + } + AcpxEventPayload::Process { details } => { + Ok(vec![AcpxProviderStateEvent::Process(details)]) + } + AcpxEventPayload::Diagnostic { code, message } => { + Ok(vec![AcpxProviderStateEvent::Diagnostic { code, message }]) + } + } + } + + pub fn pending_tool(&self, call_id: &str) -> Option<&AcpxPendingTool> { + self.pending_tools.get(call_id) + } + + pub fn complete_tool( + &mut self, + call_id: &str, + operation_id: &str, + ) -> Result<(), LocalRunnerError> { + let pending = self.pending_tools.get(call_id).ok_or_else(|| { + LocalRunnerError::invalid("ACPX tool result has no pending sidecar call") + })?; + if pending.operation_id != operation_id { + return Err(LocalRunnerError::invalid( + "ACPX tool result operation mismatch", + )); + } + let pending = self + .pending_tools + .remove(call_id) + .expect("validated ACPX pending tool remains present"); + self.pending_tool_input_bytes = self + .pending_tool_input_bytes + .saturating_sub(pending.input_bytes); + Ok(()) + } + + pub fn complete_permission(&mut self, request_id: &str) -> Result<(), LocalRunnerError> { + let value_bytes = self.pending_permissions.remove(request_id).ok_or_else(|| { + LocalRunnerError::invalid("ACPX permission result has no pending request") + })?; + self.pending_runtime_request_bytes = self + .pending_runtime_request_bytes + .saturating_sub(value_bytes); + Ok(()) + } + + pub fn complete_input(&mut self, request_id: &str) -> Result<(), LocalRunnerError> { + let pending = self + .pending_inputs + .remove(request_id) + .ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?; + self.pending_runtime_request_bytes = self + .pending_runtime_request_bytes + .saturating_sub(pending.value_bytes); + Ok(()) + } + + pub fn semantic_result(&self) -> Option<&AcpxSemanticResult> { + self.semantic_result.as_ref() + } + + fn accept_runtime_event( + &mut self, + event: &AcpxSidecarEvent, + kind: AcpxRuntimeEventKind, + tool_operation: Option<&'static str>, + payload: Value, + ) -> Result, LocalRunnerError> { + if kind == AcpxRuntimeEventKind::Thinking { + if self.thinking_active { + return Ok(Vec::new()); + } + self.thinking_active = true; + } + if kind == AcpxRuntimeEventKind::TextDelta { + let text = payload + .get("text") + .and_then(Value::as_str) + .unwrap_or_default(); + if self + .assistant_text + .len() + .checked_add(text.len()) + .is_none_or(|bytes| bytes > MAX_ASSISTANT_TEXT_BYTES) + { + return Err(LocalRunnerError::invalid( + "ACPX assistant text exceeds its retained limit", + )); + } + self.assistant_text.push_str(text); + } + if kind == AcpxRuntimeEventKind::SemanticResult { + let result = AcpxSemanticResult { + call_id: payload + .get("callId") + .and_then(Value::as_str) + .expect("decoded ACPX semantic result has a call id") + .to_owned(), + operation_id: payload + .get("operationId") + .and_then(Value::as_str) + .expect("decoded ACPX semantic result has an operation id") + .to_owned(), + result: payload + .get("result") + .expect("decoded ACPX semantic result has a result") + .clone(), + }; + 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", + )), + }; + } + let turn_id = event + .turn_id + .as_deref() + .expect("a decoded runtime event has a turn binding"); + let fallback_item_id = format!("acpx-event-{}", event.sequence); + Ok(normalize_acpx_runtime_event( + kind, + &payload, + tool_operation, + &fallback_item_id, + turn_id, + self.provider_requests, + ) + .into_iter() + .map(AcpxProviderStateEvent::Activity) + .collect()) + } + + fn admit_runtime_request( + &self, + request_id: &str, + value_bytes: usize, + ) -> Result<(), LocalRunnerError> { + if self.pending_permissions.contains_key(request_id) + || self.pending_inputs.contains_key(request_id) + { + return Err(LocalRunnerError::invalid( + "ACPX reused a pending runtime request id", + )); + } + if self.pending_permissions.len() + self.pending_inputs.len() + >= MAX_PENDING_RUNTIME_REQUESTS + || self + .pending_runtime_request_bytes + .checked_add(value_bytes) + .is_none_or(|bytes| bytes > MAX_PENDING_RUNTIME_REQUEST_BYTES) + { + return Err(LocalRunnerError::invalid( + "ACPX pending runtime requests exceed their bounded capacity", + )); + } + Ok(()) + } + + fn clear_pending_requests(&mut self) { + self.pending_tools.clear(); + self.pending_tool_input_bytes = 0; + self.pending_permissions.clear(); + self.pending_inputs.clear(); + self.pending_runtime_request_bytes = 0; + } +} + +fn value_bytes(value: &Value) -> Result { + serde_json::to_vec(value) + .map(|value| value.len()) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX retained value is invalid: {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 1c1f937281..e4f2893055 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -2,6 +2,7 @@ pub mod acpx_event_payload; pub mod acpx_event_scope; +pub mod acpx_provider_state; pub mod acpx_sidecar_transport; pub mod codex_provider; pub mod durable; 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 new file mode 100644 index 0000000000..b2b952013f --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs @@ -0,0 +1,284 @@ +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 serde_json::{json, Value}; + +fn event( + sequence: u64, + event_type: GeneratedAcpxSidecarEventType, + turn_id: Option<&str>, + payload: Value, +) -> AcpxSidecarEvent { + AcpxSidecarEvent { + sequence, + event_type, + run_id: Some("run-1".to_owned()), + turn_id: turn_id.map(str::to_owned), + payload, + } +} + +fn question_set() -> Value { + json!({ + "schema":"paperclip.question_set.v1", + "title":"Choose", + "questions":[{ + "id":"question-1", + "prompt":"Which option?", + "required":true, + "answerMode":"single_select", + "options":[{"id":"option-1","label":"First"}] + }] + }) +} + +#[test] +fn accumulates_assistant_text_deduplicates_reasoning_and_flushes_before_terminal() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + + let thinking = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({"type":"thinking","text":"private"}), + ); + assert_eq!(state.accept_event(&thinking).unwrap().len(), 1); + let repeated = event( + 2, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({"type":"thinking","text":"still private"}), + ); + assert!(state.accept_event(&repeated).unwrap().is_empty()); + + for (sequence, text) in [(3, "Hello "), (4, "world")] { + state + .accept_event(&event( + sequence, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({"type":"text_delta","text":text}), + )) + .unwrap(); + } + let terminal = state + .accept_event(&event( + 5, + GeneratedAcpxSidecarEventType::RuntimeTurnTerminal, + Some("turn-1"), + json!({"status":"completed"}), + )) + .unwrap(); + assert!(matches!( + &terminal[0], + AcpxProviderStateEvent::AssistantMessage { turn_id, text } + if turn_id == "turn-1" && text == "Hello world" + )); + assert!(matches!( + &terminal[1], + AcpxProviderStateEvent::TurnTerminal { + turn_id, + status: AcpxTurnStatus::Completed, + error: None, + } if turn_id == "turn-1" + )); + assert_eq!(state.active_turn_id(), None); +} + +#[test] +fn correlates_semantic_tool_calls_until_the_sidecar_resolution_commits() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let called = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeToolCalled, + Some("turn-1"), + json!({"callId":"call-1","operationId":"issues.read","input":{"id":"issue-1"}}), + ); + let emitted = state.accept_event(&called).unwrap(); + assert!(matches!( + &emitted[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. } + if call_id == "call-1" && operation_id == "issues.read" + )); + assert_eq!( + state.pending_tool("call-1").unwrap().operation_id, + "issues.read" + ); + assert!(state.complete_tool("call-1", "issues.write").is_err()); + assert!(state.pending_tool("call-1").is_some()); + state.complete_tool("call-1", "issues.read").unwrap(); + assert!(state.pending_tool("call-1").is_none()); + + assert!(state.accept_event(&called).is_ok()); + assert!(state.accept_event(&called).is_err()); +} + +#[test] +fn tracks_structured_input_and_permission_requests_without_cross_kind_reuse() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let input = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeInputRequested, + Some("turn-1"), + json!({"requestId":"request-1","questionSet":question_set()}), + ); + assert!(matches!( + &state.accept_event(&input).unwrap()[0], + AcpxProviderStateEvent::InputRequest { request_id, .. } + if request_id == "request-1" + )); + let permission_with_reused_id = event( + 2, + GeneratedAcpxSidecarEventType::RuntimePermissionRequested, + Some("turn-1"), + json!({"requestId":"request-1","kind":"execute","title":"Run?"}), + ); + assert!(state.accept_event(&permission_with_reused_id).is_err()); + state.complete_input("request-1").unwrap(); + assert!(state.complete_input("request-1").is_err()); + + let permission = event( + 3, + GeneratedAcpxSidecarEventType::RuntimePermissionRequested, + Some("turn-1"), + json!({"requestId":"permission-1","kind":"execute","title":"Run?"}), + ); + assert!(matches!( + &state.accept_event(&permission).unwrap()[0], + AcpxProviderStateEvent::PermissionRequest { request_id, .. } + if request_id == "permission-1" + )); + state.complete_permission("permission-1").unwrap(); +} + +#[test] +fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let result = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({ + "type":"semantic_result", + "callId":"finish-1", + "operationId":"paperclip_finish", + "result":{"reportedWorkDisposition":"done"} + }), + ); + assert!(matches!( + &state.accept_event(&result).unwrap()[0], + AcpxProviderStateEvent::SemanticResult(result) + if result.call_id == "finish-1" + )); + assert!(state.accept_event(&result).unwrap().is_empty()); + assert_eq!( + state.semantic_result().unwrap().result["reportedWorkDisposition"], + "done" + ); + + let conflict = event( + 2, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({ + "type":"semantic_result", + "callId":"finish-2", + "operationId":"paperclip_finish", + "result":{"reportedWorkDisposition":"done"} + }), + ); + assert!(state.accept_event(&conflict).is_err()); + assert_eq!(state.semantic_result().unwrap().call_id, "finish-1"); +} + +#[test] +fn validates_scope_before_mutating_pending_state() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + let mut wrong_run = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeToolCalled, + Some("turn-1"), + json!({"callId":"call-1","operationId":"issues.read","input":{}}), + ); + wrong_run.run_id = Some("run-2".to_owned()); + assert!(state.accept_event(&wrong_run).is_err()); + assert!(state.pending_tool("call-1").is_none()); + + let wrong_turn = event( + 2, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-2"), + json!({"type":"text_delta","text":"wrong"}), + ); + assert!(state.accept_event(&wrong_turn).is_err()); +} + +#[test] +fn accepts_global_process_and_diagnostic_events_without_an_active_turn() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + let mut process = event( + 1, + GeneratedAcpxSidecarEventType::RuntimeProcess, + None, + json!({"pid":17,"accessToken":"secret"}), + ); + process.run_id = None; + let process = state.accept_event(&process).unwrap(); + assert!(matches!( + &process[0], + AcpxProviderStateEvent::Process(details) + if details["accessToken"] == "[REDACTED]" + )); + + let mut diagnostic = event( + 2, + GeneratedAcpxSidecarEventType::RuntimeDiagnostic, + None, + json!({"code":"provider_notice","message":"token=super-secret"}), + ); + diagnostic.run_id = None; + let diagnostic = state.accept_event(&diagnostic).unwrap(); + assert!(matches!( + &diagnostic[0], + AcpxProviderStateEvent::Diagnostic { message, .. } + if message.contains("REDACTED") && !message.contains("super-secret") + )); +} + +#[test] +fn terminal_events_clear_pending_requests_and_reject_late_turn_events() { + let mut state = AcpxProviderState::new("run-1").unwrap(); + state.begin_turn("turn-1").unwrap(); + state + .accept_event(&event( + 1, + GeneratedAcpxSidecarEventType::RuntimeToolCalled, + Some("turn-1"), + json!({"callId":"call-1","operationId":"issues.read","input":{}}), + )) + .unwrap(); + state + .accept_event(&event( + 2, + GeneratedAcpxSidecarEventType::RuntimeTurnTerminal, + Some("turn-1"), + json!({"status":"cancelled","error":{"message":"token=secret"}}), + )) + .unwrap(); + assert!(state.pending_tool("call-1").is_none()); + assert!(state.complete_tool("call-1", "issues.read").is_err()); + assert!(state + .accept_event(&event( + 3, + GeneratedAcpxSidecarEventType::RuntimeEvent, + Some("turn-1"), + json!({"type":"text_delta","text":"late"}), + )) + .is_err()); +}