From 75708fec6d421a247ba2fc832997de24ed10a085 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 31 Aug 2026 00:19:18 -0500 Subject: [PATCH] feat(runner): add ACPX sidecar transport (#12412) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The runner needs a bounded process boundary for each qualified provider runtime > - The package now provides a reviewed Codex-only ACPX sidecar executable > - Runnerd does not yet have a strict client for that sidecar protocol > - A later provider adapter must not mix process transport defects with provider mapping > - This pull request adds the package-local Rust transport and keeps it unselected > - The benefit is a tested fail-closed boundary for the later Codex provider adapter without a production behavior change ## Linked Issues or Issue Description Refs #12410 Refs #12386 ## What Changed - Add a Rust client for the generated ACPX sidecar v2 contract. - Validate the executable path, launch arguments, request timeout, and shutdown grace before process start. - Require exact request identities and contiguous event sequence numbers. - Reject replayed events, sequence gaps, wrong response identities, malformed frames, unknown fields, and unsupported protocol versions. - Bound stdout frames to 1 MiB and buffered events to 512 entries. - Bound event poll timeouts to 120 seconds before any `Instant` arithmetic. - Buffer valid events that arrive while a command waits for its response. - Treat an empty event poll as a normal timeout without poisoning the transport. - Discard retained events and reject all polling after a terminal transport failure. - Terminate the process group after a timeout, transport failure, or protocol failure. - Keep a valid sidecar command rejection separate from a transport failure so the next command can run. - Redact and bound stderr diagnostics before they enter an error. - Add a deterministic fake sidecar and twelve integration tests for success, polling, timeout bounds, poisoned queues, replay, gaps, identity mismatch, rejection, event floods, oversized frames, and secret redaction. - Document that the transport remains package-local and does not change runnerd provider selection. - Do not change dependencies, lockfiles, workflows, runnerd selection, server behavior, UI, or migrations. ## Verification - Replay base: `9ad8dbffa0a4759dcda2769042d6e8f02adcdf8d` (`master` after #12410 merged). - Exact replay head: `5c41111c4f9564405a6e87a02b6cf253a4424e5f`. - Stable patch ID: `60aca2620fdbb73fbbc203e928d39dcd450e085b`; this preserves the reviewed `1334a7f5..c5654218` six-file delta and keeps trusted bounded-reader failures distinct from fully redacted child stderr. - The exact delta is 6 files and 896 additions, all in `packages/paperclip-runner`; it contains no lockfile, workflow, server, UI, or migration change. - Focused Rust transport, package, repository, security, and Greptile checks: **PASSED** on the replayed exact head. Full CI run `33359202433` is green; its failed-job retry passed the two unrelated flaky jobs without a patch change. Greptile is exact-head 5/5, all security checks pass, and no review threads remain unresolved. - No local test result is claimed. GitHub Actions is the authoritative verification environment for this replayed revision. ## Risks - The transport crosses an untrusted process boundary and is security-sensitive. - It fails closed on malformed frames, identity drift, event replay, sequence gaps, queue overflow, frame overflow, timeout, process exit, and channel failure. - A terminal failure clears retained events before it marks the transport unavailable. - It redacts and bounds retained diagnostics before it returns them to a caller. - A valid remote command rejection does not corrupt the transport state. - The package exports a new Rust module, but no production path constructs it in this pull request. - The later provider adapter must validate run, turn, session, model, and tool bindings before it selects this transport. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex with GPT-5.6, agentic reasoning, tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- packages/paperclip-runner/README.md | 5 + .../runner/crates/runner-core/Cargo.toml | 4 + .../runner-core/src/acpx_sidecar_transport.rs | 582 ++++++++++++++++++ .../runner-core/src/bin/fake-acpx-sidecar.rs | 123 ++++ .../runner/crates/runner-core/src/lib.rs | 2 + .../tests/acpx_sidecar_transport.rs | 180 ++++++ 6 files changed, 896 insertions(+) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs create mode 100644 packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index a6bd70d1d1..764541787d 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -55,6 +55,11 @@ the exact model, session identity, tool catalog, structured input, and terminal settlement at the process boundary. Runnerd and the server do not select this sidecar in this slice. Other ACPX agents remain unavailable. +The Rust core includes a bounded client for this sidecar protocol. It enforces +request identity, event order, frame and queue limits, timeouts, redacted +diagnostics, and process-group cleanup. This transport remains package-local. +It does not change runnerd provider selection in this slice. + Run the complete contract gate with: ```sh diff --git a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml index cb5c9678ca..df141ae78e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml +++ b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml @@ -30,3 +30,7 @@ path = "src/bin/fake-harness.rs" [[bin]] name = "fake-codex-app-server" path = "src/bin/fake-codex-app-server.rs" + +[[bin]] +name = "fake-acpx-sidecar" +path = "src/bin/fake-acpx-sidecar.rs" diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs new file mode 100644 index 0000000000..5a635df5ca --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -0,0 +1,582 @@ +use std::collections::VecDeque; +use std::path::PathBuf; +use std::sync::mpsc::RecvTimeoutError; +use std::time::{Duration, Instant}; + +use serde::Deserialize; +use serde_json::{json, Value}; + +use crate::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GeneratedAcpxSidecarEventType, + GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, +}; +use crate::local_runner::LocalRunnerError; +use crate::process_supervisor::{BoundedLogBuffer, ProcessOutput, SupervisedProcess}; + +pub const ACPX_SIDECAR_MAX_FRAME_BYTES: usize = 1024 * 1024; +const MAX_BUFFERED_EVENTS: usize = 512; +const MAX_EVENT_POLL_TIMEOUT: Duration = Duration::from_secs(120); +const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991; + +#[derive(Clone, Debug)] +pub struct AcpxSidecarTransportConfig { + pub command: PathBuf, + pub args: Vec, + pub request_timeout: Duration, + pub shutdown_grace: Duration, +} + +impl AcpxSidecarTransportConfig { + pub fn validate(&self) -> Result<(), LocalRunnerError> { + if !self.command.is_absolute() || !self.command.is_file() { + return Err(LocalRunnerError::invalid( + "ACPX sidecar command must be an existing absolute file", + )); + } + if self.args.len() > 64 + || self.args.iter().any(|argument| { + argument.len() > 4_096 || argument.chars().any(|character| character == '\0') + }) + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar arguments exceed the bounded launch contract", + )); + } + if self.request_timeout < Duration::from_millis(1) + || self.request_timeout > Duration::from_secs(120) + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar request timeout must be in the range 1 ms through 120 s", + )); + } + if self.shutdown_grace < Duration::from_millis(1) + || self.shutdown_grace > Duration::from_secs(30) + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar shutdown grace must be in the range 1 ms through 30 s", + )); + } + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq)] +pub struct AcpxSidecarEvent { + pub sequence: u64, + pub event_type: GeneratedAcpxSidecarEventType, + pub run_id: Option, + pub turn_id: Option, + pub payload: Value, +} + +pub struct AcpxSidecarTransport { + process: SupervisedProcess, + request_timeout: Duration, + next_request_id: u64, + last_event_sequence: u64, + buffered_events: VecDeque, + stderr_tail: BoundedLogBuffer, + poisoned: bool, +} + +impl AcpxSidecarTransport { + pub fn start(config: &AcpxSidecarTransportConfig) -> Result { + config.validate()?; + let process = SupervisedProcess::spawn( + &config.command, + &config.args, + config.shutdown_grace, + ACPX_SIDECAR_MAX_FRAME_BYTES, + )?; + Ok(Self { + process, + request_timeout: config.request_timeout, + next_request_id: 1, + last_event_sequence: 0, + buffered_events: VecDeque::new(), + stderr_tail: BoundedLogBuffer::new(32, 8 * 1024), + poisoned: false, + }) + } + + pub fn process_id(&self) -> u32 { + self.process.id() + } + + pub fn request( + &mut self, + command: GeneratedAcpxSidecarCommand, + params: Value, + ) -> Result { + if self.poisoned { + return Err(LocalRunnerError::invalid( + "ACPX sidecar transport is unavailable after a protocol failure", + )); + } + let result = self.request_inner(command, params); + match result { + Ok(CommandOutcome::Success(value)) => Ok(value), + Ok(CommandOutcome::Rejected(error)) => Err(error), + Err(error) => { + self.poison(); + Err(error) + } + } + } + + pub fn poll_event( + &mut self, + timeout: Duration, + ) -> Result, LocalRunnerError> { + if self.poisoned { + return Err(LocalRunnerError::invalid( + "ACPX sidecar transport is unavailable after a protocol failure", + )); + } + if timeout > MAX_EVENT_POLL_TIMEOUT { + return Err(LocalRunnerError::invalid( + "ACPX sidecar event poll timeout must not exceed 120 s", + )); + } + if let Some(event) = self.buffered_events.pop_front() { + return Ok(Some(event)); + } + if timeout.is_zero() { + return Ok(None); + } + let result = self.poll_event_inner(timeout); + if result.is_err() { + self.poison(); + } + result + } + + pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> { + self.poisoned = true; + self.process.terminate_group().map(|_| ()) + } + + fn request_inner( + &mut self, + command: GeneratedAcpxSidecarCommand, + params: Value, + ) -> Result { + if !params.is_object() { + return Err(LocalRunnerError::invalid( + "ACPX sidecar command params must be an object", + )); + } + let request_id = self.next_request_id; + if request_id > MAX_JSON_SAFE_INTEGER { + return Err(LocalRunnerError::invalid( + "ACPX sidecar request sequence is exhausted", + )); + } + let frame = json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": request_id, + "command": command.as_str(), + "params": params, + }); + let frame_bytes = serde_json::to_vec(&frame).map_err(|error| { + LocalRunnerError::invalid(format!("ACPX sidecar request is not serializable: {error}")) + })?; + if frame_bytes.len() > ACPX_SIDECAR_MAX_FRAME_BYTES { + return Err(LocalRunnerError::invalid( + "ACPX sidecar request exceeds the frame limit", + )); + } + self.process.send(&frame).map_err(|error| { + LocalRunnerError::invalid(format!( + "ACPX sidecar request transport failed at {}: {error}", + command.as_str() + )) + })?; + self.next_request_id = request_id + 1; + + let deadline = Instant::now() + self.request_timeout; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(self.request_timeout_error(command)); + } + let Some(line) = self.receive_stdout_line(remaining, command.as_str())? else { + return Err(self.request_timeout_error(command)); + }; + match parse_frame(&line)? { + ParsedFrame::Event(event) => self.buffer_event(event)?, + ParsedFrame::Response(response) => { + if response.id != request_id { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar response id mismatch: expected {request_id}, received {}", + response.id + ))); + } + if response.ok { + return Ok(CommandOutcome::Success( + response.result.unwrap_or_else(|| json!({})), + )); + } + let error = response.error.expect("failed response has validated error"); + return Ok(CommandOutcome::Rejected(LocalRunnerError::invalid( + format!( + "ACPX sidecar command {} was rejected (retryable={})", + command.as_str(), + error.retryable, + ), + ))); + } + } + } + } + + fn poll_event_inner( + &mut self, + timeout: Duration, + ) -> Result, LocalRunnerError> { + let Some(line) = self.receive_stdout_line(timeout, "event.poll")? else { + return Ok(None); + }; + match parse_frame(&line)? { + ParsedFrame::Event(event) => { + self.validate_event_sequence(event.sequence)?; + Ok(Some(event)) + } + ParsedFrame::Response(response) => Err(LocalRunnerError::invalid(format!( + "ACPX sidecar emitted response {} without a pending request", + response.id + ))), + } + } + + fn receive_stdout_line( + &mut self, + timeout: Duration, + stage: &str, + ) -> Result, LocalRunnerError> { + let deadline = Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Ok(None); + } + match self.process.recv_timeout(remaining) { + Ok(ProcessOutput::Stdout(line)) => return Ok(Some(line)), + Ok(ProcessOutput::Stderr(line)) => { + self.stderr_tail.push(redact_diagnostic(&line)); + } + Ok(ProcessOutput::StdoutError(message)) => { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar stdout failed at {stage}: {}{}", + message, + self.diagnostic_suffix() + ))); + } + Ok(ProcessOutput::StdoutClosed) => return Err(self.closed_error(stage)), + Ok(ProcessOutput::StderrClosed) => {} + Err(RecvTimeoutError::Timeout) => return Ok(None), + Err(RecvTimeoutError::Disconnected) => { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar output channel closed at {stage}{}", + self.diagnostic_suffix() + ))); + } + } + } + } + + fn buffer_event(&mut self, event: AcpxSidecarEvent) -> Result<(), LocalRunnerError> { + if self.buffered_events.len() >= MAX_BUFFERED_EVENTS { + return Err(LocalRunnerError::invalid( + "ACPX sidecar exceeded the buffered event limit", + )); + } + self.validate_event_sequence(event.sequence)?; + self.buffered_events.push_back(event); + Ok(()) + } + + fn validate_event_sequence(&mut self, sequence: u64) -> Result<(), LocalRunnerError> { + let expected = self.last_event_sequence + 1; + if sequence != expected { + let disposition = if sequence <= self.last_event_sequence { + "replayed" + } else { + "has a gap" + }; + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar event sequence {disposition}: expected {expected}, received {sequence}" + ))); + } + self.last_event_sequence = sequence; + Ok(()) + } + + fn request_timeout_error(&self, command: GeneratedAcpxSidecarCommand) -> LocalRunnerError { + LocalRunnerError::invalid(format!( + "ACPX sidecar request timed out at {}{}", + command.as_str(), + self.diagnostic_suffix() + )) + } + + fn closed_error(&mut self, stage: &str) -> LocalRunnerError { + self.drain_diagnostics(Duration::from_millis(20)); + let suffix = self.diagnostic_suffix(); + match self.process.try_wait() { + Ok(Some(exit)) => LocalRunnerError::invalid(format!( + "ACPX sidecar exited at {stage}: exitCode={:?} signal={:?}{suffix}", + exit.exit_code, exit.signal + )), + Ok(None) => { + LocalRunnerError::invalid(format!("ACPX sidecar closed stdout at {stage}{suffix}")) + } + Err(error) => LocalRunnerError::invalid(format!( + "ACPX sidecar status failed at {stage}: {error}{suffix}" + )), + } + } + + fn drain_diagnostics(&mut self, max_wait: Duration) { + let deadline = Instant::now() + max_wait; + loop { + let output = if max_wait.is_zero() { + self.process.try_recv().ok() + } else { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + None + } else { + self.process.recv_timeout(remaining).ok() + } + }; + match output { + Some(ProcessOutput::Stderr(line)) => { + self.stderr_tail.push(redact_diagnostic(&line)); + } + Some(ProcessOutput::StderrClosed) | None => break, + Some(ProcessOutput::Stdout(_)) + | Some(ProcessOutput::StdoutError(_)) + | Some(ProcessOutput::StdoutClosed) => {} + } + } + } + + fn diagnostic_suffix(&self) -> String { + let diagnostics = self.stderr_tail.snapshot().lines.join("\n"); + if diagnostics.is_empty() { + String::new() + } else { + format!(" stderrTail={diagnostics:?}") + } + } + + fn poison(&mut self) { + if self.poisoned { + return; + } + self.poisoned = true; + self.buffered_events.clear(); + let _ = self.process.terminate_group(); + } +} + +enum CommandOutcome { + Success(Value), + Rejected(LocalRunnerError), +} + +enum ParsedFrame { + Response(ResponseFrame), + Event(AcpxSidecarEvent), +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ResponseFrame { + protocol_version: u64, + id: u64, + ok: bool, + #[serde(default)] + result: Option, + #[serde(default)] + error: Option, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ResponseError { + code: String, + message: String, + retryable: bool, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct EventFrame { + protocol_version: u64, + sequence: u64, + event_type: GeneratedAcpxSidecarEventType, + run_id: Value, + turn_id: Value, + payload: Value, +} + +fn parse_frame(line: &str) -> Result { + let value: Value = serde_json::from_str(line) + .map_err(|_| LocalRunnerError::invalid("ACPX sidecar emitted invalid JSON"))?; + let object = value + .as_object() + .ok_or_else(|| LocalRunnerError::invalid("ACPX sidecar frame must be an object"))?; + if object.contains_key("eventType") { + let frame: EventFrame = serde_json::from_value(value) + .map_err(|_| LocalRunnerError::invalid("ACPX sidecar event frame is invalid"))?; + if frame.protocol_version != GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION { + return Err(LocalRunnerError::invalid( + "ACPX sidecar event protocol version mismatch", + )); + } + if frame.sequence == 0 || frame.sequence > MAX_JSON_SAFE_INTEGER { + return Err(LocalRunnerError::invalid( + "ACPX sidecar event sequence is invalid", + )); + } + let run_id = nullable_identifier(frame.run_id, "event runId")?; + let turn_id = nullable_identifier(frame.turn_id, "event turnId")?; + if !frame.payload.is_object() { + return Err(LocalRunnerError::invalid( + "ACPX sidecar event payload must be an object", + )); + } + return Ok(ParsedFrame::Event(AcpxSidecarEvent { + sequence: frame.sequence, + event_type: frame.event_type, + run_id, + turn_id, + payload: frame.payload, + })); + } + + let result_is_present = object.contains_key("result"); + let error_is_present = object.contains_key("error"); + if result_is_present && !object.get("result").is_some_and(Value::is_object) { + return Err(LocalRunnerError::invalid( + "ACPX sidecar response result must be an object", + )); + } + if error_is_present && !object.get("error").is_some_and(Value::is_object) { + return Err(LocalRunnerError::invalid( + "ACPX sidecar response error must be an object", + )); + } + let frame: ResponseFrame = serde_json::from_value(value) + .map_err(|_| LocalRunnerError::invalid("ACPX sidecar response frame is invalid"))?; + if frame.protocol_version != GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION { + return Err(LocalRunnerError::invalid( + "ACPX sidecar response protocol version mismatch", + )); + } + if frame.id == 0 || frame.id > MAX_JSON_SAFE_INTEGER { + return Err(LocalRunnerError::invalid( + "ACPX sidecar response id is invalid", + )); + } + if frame.ok { + if error_is_present { + return Err(LocalRunnerError::invalid( + "successful ACPX sidecar response contains an error", + )); + } + } else if result_is_present || !error_is_present || frame.error.is_none() { + return Err(LocalRunnerError::invalid( + "failed ACPX sidecar response has an invalid result/error shape", + )); + } + if let Some(error) = frame.error.as_ref() { + if error.code.is_empty() + || error.code.chars().count() > 160 + || !error + .code + .chars() + .all(|character| character.is_ascii_alphanumeric() || "._:-".contains(character)) + || error.message.chars().count() > 8_192 + { + return Err(LocalRunnerError::invalid( + "ACPX sidecar response error exceeds its contract bounds", + )); + } + } + Ok(ParsedFrame::Response(frame)) +} + +fn nullable_identifier(value: Value, field: &str) -> Result, LocalRunnerError> { + if value.is_null() { + return Ok(None); + } + let Some(value) = value.as_str() else { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar {field} must be a string or null" + ))); + }; + if value.chars().count() > 160 || value.chars().any(char::is_control) { + return Err(LocalRunnerError::invalid(format!( + "ACPX sidecar {field} is invalid" + ))); + } + Ok(Some(value.to_owned())) +} + +fn redact_diagnostic(value: &str) -> String { + if value.is_empty() { + String::new() + } else { + "[REDACTED]".to_owned() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_frame_does_not_echo_untrusted_deserialization_details() { + let cases = [ + ( + json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": 1, + "ok": true, + "result": {}, + "opaque_field_canary_Q7Z9": true, + }) + .to_string(), + "ACPX sidecar response frame is invalid", + ), + ( + json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence": 1, + "eventType": "opaque_variant_canary_Q7Z9", + "runId": null, + "turnId": null, + "payload": {}, + }) + .to_string(), + "ACPX sidecar event frame is invalid", + ), + ( + r#"{"opaque_json_canary_Q7Z9":"#.to_owned(), + "ACPX sidecar emitted invalid JSON", + ), + ]; + + for (input, expected) in cases { + let message = match parse_frame(&input) { + Ok(_) => panic!("untrusted frame must be rejected"), + Err(error) => error.to_string(), + }; + assert!(message.contains(expected), "unexpected error: {message}"); + assert!(!message.contains("Q7Z9"), "error leaked input: {message}"); + } + } +} 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 new file mode 100644 index 0000000000..3a9e5c72e6 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs @@ -0,0 +1,123 @@ +use std::io::{self, BufRead, Write}; + +use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION; +use serde_json::{json, Value}; + +fn main() { + if let Err(error) = run() { + eprintln!("fake-acpx-sidecar: {error}"); + std::process::exit(1); + } +} + +fn run() -> Result<(), Box> { + let args = std::env::args().collect::>(); + let mode = args + .windows(2) + .find(|pair| pair[0] == "--mode") + .map(|pair| pair[1].as_str()) + .unwrap_or("happy"); + let stdin = io::stdin(); + let mut stdout = io::stdout().lock(); + let mut next_sequence = 1_u64; + for line in stdin.lock().lines() { + let request: Value = serde_json::from_str(&line?)?; + let id = request + .get("id") + .and_then(Value::as_u64) + .ok_or("request id is missing")?; + let command = request + .get("command") + .and_then(Value::as_str) + .ok_or("request command is missing")?; + match mode { + "silent" => continue, + "wrong-id" => { + write_json(&mut stdout, &success(id + 1, command, &request))?; + } + "event-wrong-id" => { + write_event(&mut stdout, next_sequence)?; + next_sequence += 1; + write_json(&mut stdout, &success(id + 1, command, &request))?; + } + "remote-error" => { + write_json( + &mut stdout, + &json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": id, + "ok": false, + "error": { + "code": "Q7Z9", + "message": "violet-circuit-4821", + "retryable": false, + }, + }), + )?; + } + "gap" => { + write_event(&mut stdout, 2)?; + write_json(&mut stdout, &success(id, command, &request))?; + } + "replay" => { + write_event(&mut stdout, 1)?; + write_event(&mut stdout, 1)?; + write_json(&mut stdout, &success(id, command, &request))?; + } + "flood" => { + for sequence in 1..=513 { + write_event(&mut stdout, sequence)?; + } + write_json(&mut stdout, &success(id, command, &request))?; + } + "oversized" => { + stdout.write_all(&vec![b'x'; 1024 * 1024 + 1])?; + stdout.write_all(b"\n")?; + stdout.flush()?; + } + "exit-secret" => { + eprintln!("amber-signal-7305"); + std::process::exit(9); + } + "happy" => { + write_event(&mut stdout, next_sequence)?; + next_sequence += 1; + write_json(&mut stdout, &success(id, command, &request))?; + } + _ => return Err(format!("unknown fake mode {mode}").into()), + } + } + Ok(()) +} + +fn success(id: u64, command: &str, request: &Value) -> Value { + json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id": id, + "ok": true, + "result": { + "command": command, + "params": request.get("params").cloned().unwrap_or_else(|| json!({})), + }, + }) +} + +fn write_event(output: &mut impl Write, sequence: u64) -> io::Result<()> { + write_json( + output, + &json!({ + "protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence": sequence, + "eventType": "runtime.diagnostic", + "runId": null, + "turnId": null, + "payload": { "code": "fake_event", "message": "bounded" }, + }), + ) +} + +fn write_json(output: &mut impl Write, value: &Value) -> io::Result<()> { + serde_json::to_writer(&mut *output, value)?; + output.write_all(b"\n")?; + output.flush() +} 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 247563c0a2..213dd5f107 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -1,8 +1,10 @@ #![forbid(unsafe_code)] +pub mod acpx_sidecar_transport; pub mod codex_provider; pub mod durable; pub mod fake_harness; +pub mod generated_acpx_sidecar_contract; pub mod local_runner; pub mod process_supervisor; pub mod provider_backend; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs new file mode 100644 index 0000000000..41d402c1d8 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs @@ -0,0 +1,180 @@ +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use paperclip_runner_core::acpx_sidecar_transport::{ + AcpxSidecarTransport, AcpxSidecarTransportConfig, +}; +use paperclip_runner_core::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GeneratedAcpxSidecarEventType, +}; +use serde_json::json; + +fn transport(mode: &str, timeout: Duration) -> AcpxSidecarTransport { + AcpxSidecarTransport::start(&AcpxSidecarTransportConfig { + command: PathBuf::from(env!("CARGO_BIN_EXE_fake-acpx-sidecar")), + args: vec!["--mode".to_owned(), mode.to_owned()], + request_timeout: timeout, + shutdown_grace: Duration::from_millis(50), + }) + .expect("fake ACPX sidecar should start") +} + +#[test] +fn buffers_events_that_arrive_before_a_response() { + let mut transport = transport("happy", Duration::from_secs(1)); + let result = transport + .request( + GeneratedAcpxSidecarCommand::Initialize, + json!({ "agent": "codex", "model": "gpt-5.6-sol" }), + ) + .expect("fake initialize should respond"); + assert_eq!(result["command"], "initialize"); + let event = transport + .poll_event(Duration::ZERO) + .expect("buffered event should parse") + .expect("buffered event should exist"); + assert_eq!(event.sequence, 1); + assert_eq!( + event.event_type, + GeneratedAcpxSidecarEventType::RuntimeDiagnostic + ); + assert_eq!(event.run_id, None); + transport.shutdown().expect("fake sidecar should stop"); +} + +#[test] +fn rejects_event_gaps_and_poisoned_transport_reuse() { + let mut transport = transport("gap", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("event gap must fail"); + assert!(error.to_string().contains("has a gap")); + let reuse = transport + .request(GeneratedAcpxSidecarCommand::SessionRead, json!({})) + .expect_err("poisoned transport must fail closed"); + assert!(reuse.to_string().contains("unavailable")); +} + +#[test] +fn rejects_event_replay() { + let mut transport = transport("replay", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("event replay must fail"); + assert!(error.to_string().contains("replayed")); +} + +#[test] +fn rejects_response_identity_mismatch() { + let mut transport = transport("wrong-id", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("wrong response id must fail"); + assert!(error.to_string().contains("response id mismatch")); +} + +#[test] +fn a_poisoned_transport_discards_events_buffered_before_failure() { + let mut transport = transport("event-wrong-id", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("wrong response id must fail after the event is buffered"); + assert!(error.to_string().contains("response id mismatch")); + let poll_error = transport + .poll_event(Duration::ZERO) + .expect_err("a poisoned transport must not expose retained events"); + assert!(poll_error.to_string().contains("unavailable")); +} + +#[test] +fn bounds_events_while_waiting_for_a_response() { + let mut transport = transport("flood", Duration::from_secs(2)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("event flood must fail"); + assert!(error.to_string().contains("buffered event limit")); +} + +#[test] +fn rejects_an_oversized_stdout_frame() { + let mut transport = transport("oversized", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("oversized frame must fail"); + assert!(error.to_string().contains("exceeded 1048576 bytes")); +} + +#[test] +fn times_out_and_terminates_a_silent_sidecar() { + let mut transport = transport("silent", Duration::from_millis(30)); + let started = Instant::now(); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("silent sidecar must time out"); + assert!(error.to_string().contains("timed out")); + assert!(started.elapsed() < Duration::from_secs(1)); +} + +#[test] +fn an_empty_event_poll_does_not_poison_the_transport() { + let mut transport = transport("silent", Duration::from_secs(1)); + assert_eq!( + transport + .poll_event(Duration::from_millis(20)) + .expect("an empty poll is not a protocol failure"), + None + ); + assert_eq!( + transport + .poll_event(Duration::from_millis(20)) + .expect("the transport remains available after an empty poll"), + None + ); + transport.shutdown().expect("fake sidecar should stop"); +} + +#[test] +fn rejects_an_unbounded_event_poll_without_poisoning_the_transport() { + let mut transport = transport("silent", Duration::from_secs(1)); + let error = transport + .poll_event(Duration::MAX) + .expect_err("an unbounded event poll must be rejected"); + assert!(error.to_string().contains("must not exceed 120 s")); + assert_eq!( + transport + .poll_event(Duration::ZERO) + .expect("a caller timeout error must not poison the transport"), + None + ); + transport.shutdown().expect("fake sidecar should stop"); +} + +#[test] +fn keeps_valid_command_rejections_separate_from_protocol_failures() { + let mut transport = transport("remote-error", Duration::from_secs(1)); + for command in [ + GeneratedAcpxSidecarCommand::Initialize, + GeneratedAcpxSidecarCommand::SessionRead, + ] { + let error = transport + .request(command, json!({})) + .expect_err("fake command should be rejected"); + let message = error.to_string(); + assert!(message.contains("was rejected")); + assert!(!message.contains("Q7Z9")); + assert!(!message.contains("violet-circuit-4821")); + assert!(!message.contains("unavailable")); + } + transport.shutdown().expect("fake sidecar should stop"); +} + +#[test] +fn redacts_sidecar_stderr_when_the_process_exits() { + let mut transport = transport("exit-secret", Duration::from_secs(1)); + let error = transport + .request(GeneratedAcpxSidecarCommand::Initialize, json!({})) + .expect_err("exited sidecar must fail"); + let message = error.to_string(); + assert!(message.contains("[REDACTED]")); + assert!(!message.contains("amber-signal-7305")); +}