feat(runner): add ACPX sidecar transport (#12412)
## 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
This commit is contained in:
Dotta
GitHub
parent
9ad8dbffa0
commit
75708fec6d
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
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<String>,
|
||||
pub turn_id: Option<String>,
|
||||
pub payload: Value,
|
||||
}
|
||||
|
||||
pub struct AcpxSidecarTransport {
|
||||
process: SupervisedProcess,
|
||||
request_timeout: Duration,
|
||||
next_request_id: u64,
|
||||
last_event_sequence: u64,
|
||||
buffered_events: VecDeque<AcpxSidecarEvent>,
|
||||
stderr_tail: BoundedLogBuffer,
|
||||
poisoned: bool,
|
||||
}
|
||||
|
||||
impl AcpxSidecarTransport {
|
||||
pub fn start(config: &AcpxSidecarTransportConfig) -> Result<Self, LocalRunnerError> {
|
||||
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<Value, LocalRunnerError> {
|
||||
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<Option<AcpxSidecarEvent>, 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<CommandOutcome, LocalRunnerError> {
|
||||
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<Option<AcpxSidecarEvent>, 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<Option<String>, 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<Value>,
|
||||
#[serde(default)]
|
||||
error: Option<ResponseError>,
|
||||
}
|
||||
|
||||
#[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<ParsedFrame, LocalRunnerError> {
|
||||
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<Option<String>, 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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<dyn std::error::Error>> {
|
||||
let args = std::env::args().collect::<Vec<_>>();
|
||||
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()
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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"));
|
||||
}
|
||||
Loading…
Reference in New Issue