feat(runner): validate ACPX event payloads (#12415)
## 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 ACPX transport now verifies frame shape, sequence, run scope, and turn scope > - The sidecar schema keeps event payloads open so each event family needs a second validation boundary > - A later provider adapter must not retain or act on malformed payload fields > - This pull request validates and redacts package-local payloads and keeps the provider unselected > - The benefit is a typed fail-closed boundary before provider state or semantic tools can consume an event ## Linked Issues or Issue Description Refs #12414 Refs #12412 ## What Changed - Decode sidecar payloads only after run and turn scope validation passes. - Limit each decoded payload to 256 KiB. - Add typed payload variants for runtime events, permission requests, input requests, semantic tool calls, terminal events, process events, and diagnostics. - Admit only the nine runtime event shapes emitted by the reviewed Codex ACPX sidecar. - Validate runtime text, plan entries, tool locations, semantic result identities, notices, errors, and terminal status values. - Validate input requests against `paperclip.question_set.v1`. - Reject duplicate question IDs and duplicate option IDs within one question. - Require bounded control identities and object-shaped operational values. - Redact diagnostic, error, process, permission, tool, and retained runtime values before they can enter provider state. - Add six integration tests for every admitted shape, malformed values, scope-before-decode ordering, size limits, question ambiguity, and secret redaction. - Document the payload boundary. - Do not change dependencies, lockfiles, workflows, runnerd selection, server behavior, UI, or migrations. ## Verification - Replay base: `3db24d9366831559b1219782475e760ec041b639` (`master` after #12414 merged). - Exact replay head: `de045b42b5b52cca6c3021380747c56693ccd179`. - Stable patch ID: `90a6f0ed68b2ca7fa5397a8bf93e5e95df5bb58c`, identical to the reviewed `972a3b38..9bb85e93` delta. - The exact delta is 4 files and 806 additions, all in `packages/paperclip-runner`; it contains no lockfile, workflow, server, UI, or migration change. - Focused Rust payload, package, repository, security, and Greptile checks: **PASSED** on the replayed exact head. Full CI run `33360832792` completed 23/23 jobs successfully, 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 - This decoder is a security boundary because later code can act on decoded semantic tool calls and input requests. - It validates event authority before it inspects a payload. - It fails closed on unknown runtime event types, malformed fields, ambiguous question identifiers, unsupported terminal states, and oversized payloads. - It applies the existing durable redaction policy to retained values. - The package exports a new Rust module, but no production path constructs it in this pull request. - A later provider adapter must preserve this validation order and must not consume raw sidecar payloads directly. > 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
3db24d9366
commit
fe2ddfad2b
|
|
@ -65,6 +65,11 @@ also requires its optional or mandatory run and turn scope to match the active
|
|||
execution. Process and diagnostic events can remain global. All operational,
|
||||
tool, input, permission, and terminal events require the exact active binding.
|
||||
|
||||
A package-local payload boundary decodes events only after that scope check. It
|
||||
validates control identities, terminal status, question sets, and the admitted
|
||||
runtime event types and bounded fields. It redacts diagnostic and retained
|
||||
event values again before they can enter provider state.
|
||||
|
||||
Run the complete contract gate with:
|
||||
|
||||
```sh
|
||||
|
|
|
|||
|
|
@ -0,0 +1,468 @@
|
|||
use std::collections::BTreeSet;
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::acpx_event_scope::AcpxEventScope;
|
||||
use crate::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use crate::durable::{redact_text, sanitize_value};
|
||||
use crate::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
|
||||
const MAX_EVENT_PAYLOAD_BYTES: usize = 256 * 1024;
|
||||
const MAX_ID_CHARS: usize = 160;
|
||||
const MAX_RUNTIME_TEXT_CHARS: usize = 64 * 1024;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum AcpxRuntimeEventKind {
|
||||
TextDelta,
|
||||
Thinking,
|
||||
Plan,
|
||||
Status,
|
||||
ToolCall,
|
||||
SemanticResult,
|
||||
ProviderNotice,
|
||||
Error,
|
||||
Done,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum AcpxTurnStatus {
|
||||
Completed,
|
||||
Failed,
|
||||
Cancelled,
|
||||
Interrupted,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum AcpxEventPayload {
|
||||
Runtime {
|
||||
kind: AcpxRuntimeEventKind,
|
||||
payload: Value,
|
||||
},
|
||||
PermissionRequested {
|
||||
request_id: String,
|
||||
kind: String,
|
||||
title: String,
|
||||
details: Value,
|
||||
},
|
||||
InputRequested {
|
||||
request_id: String,
|
||||
question_set: Value,
|
||||
origin: Option<Value>,
|
||||
},
|
||||
ToolCalled {
|
||||
call_id: String,
|
||||
operation_id: String,
|
||||
input: Value,
|
||||
},
|
||||
TurnTerminal {
|
||||
status: AcpxTurnStatus,
|
||||
error: Option<Value>,
|
||||
},
|
||||
Process {
|
||||
details: Value,
|
||||
},
|
||||
Diagnostic {
|
||||
code: String,
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
/// Validates event authority before it decodes the bounded sidecar payload.
|
||||
pub fn decode_acpx_event(
|
||||
scope: &AcpxEventScope,
|
||||
event: &AcpxSidecarEvent,
|
||||
) -> Result<AcpxEventPayload, LocalRunnerError> {
|
||||
scope.validate_event(event)?;
|
||||
let payload_bytes = serde_json::to_vec(&event.payload).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX event payload is invalid: {error}"))
|
||||
})?;
|
||||
if payload_bytes.len() > MAX_EVENT_PAYLOAD_BYTES {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event payload exceeds the 256 KiB admission limit",
|
||||
));
|
||||
}
|
||||
|
||||
match event.event_type {
|
||||
GeneratedAcpxSidecarEventType::RuntimeEvent => decode_runtime_event(&event.payload),
|
||||
GeneratedAcpxSidecarEventType::RuntimePermissionRequested => {
|
||||
Ok(AcpxEventPayload::PermissionRequested {
|
||||
request_id: required_id(&event.payload, "requestId", "permission request")?,
|
||||
kind: bounded_optional_text(&event.payload, "kind", 160, "permission kind")?
|
||||
.map(|value| redact_text(&value))
|
||||
.unwrap_or_else(|| "permission".to_owned()),
|
||||
title: bounded_optional_text(&event.payload, "title", 4_000, "permission title")?
|
||||
.map(|value| redact_text(&value))
|
||||
.unwrap_or_else(|| "ACP permission request".to_owned()),
|
||||
details: sanitize_value(&event.payload),
|
||||
})
|
||||
}
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested => {
|
||||
let question_set = event.payload.get("questionSet").cloned().ok_or_else(|| {
|
||||
LocalRunnerError::invalid("ACPX input request omitted its question set")
|
||||
})?;
|
||||
validate_question_set(&question_set)?;
|
||||
let question_set = sanitize_question_set(question_set);
|
||||
let origin = optional_object(&event.payload, "origin", "input request origin")?;
|
||||
Ok(AcpxEventPayload::InputRequested {
|
||||
request_id: required_id(&event.payload, "requestId", "input request")?,
|
||||
question_set,
|
||||
origin: origin.map(|value| sanitize_value(&value)),
|
||||
})
|
||||
}
|
||||
GeneratedAcpxSidecarEventType::RuntimeToolCalled => {
|
||||
let input = event.payload.get("input").cloned().unwrap_or(Value::Null);
|
||||
if !input.is_object() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX tool call input must be an object",
|
||||
));
|
||||
}
|
||||
Ok(AcpxEventPayload::ToolCalled {
|
||||
call_id: required_id(&event.payload, "callId", "tool call")?,
|
||||
operation_id: required_id(&event.payload, "operationId", "tool operation")?,
|
||||
input: sanitize_value(&input),
|
||||
})
|
||||
}
|
||||
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal => {
|
||||
let status = match event.payload.get("status").and_then(Value::as_str) {
|
||||
Some("completed") => AcpxTurnStatus::Completed,
|
||||
Some("failed") => AcpxTurnStatus::Failed,
|
||||
Some("cancelled" | "canceled") => AcpxTurnStatus::Cancelled,
|
||||
Some("interrupted") => AcpxTurnStatus::Interrupted,
|
||||
_ => {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX terminal event has an unsupported status",
|
||||
))
|
||||
}
|
||||
};
|
||||
let error = optional_object(&event.payload, "error", "terminal error")?;
|
||||
Ok(AcpxEventPayload::TurnTerminal {
|
||||
status,
|
||||
error: error.map(|value| sanitize_value(&value)),
|
||||
})
|
||||
}
|
||||
GeneratedAcpxSidecarEventType::RuntimeProcess => Ok(AcpxEventPayload::Process {
|
||||
details: sanitize_value(&event.payload),
|
||||
}),
|
||||
GeneratedAcpxSidecarEventType::RuntimeDiagnostic => {
|
||||
let code = required_id(&event.payload, "code", "diagnostic code")?;
|
||||
let message = event
|
||||
.payload
|
||||
.get("message")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX diagnostic omitted its message"))?;
|
||||
if message.chars().count() > 8_192 {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX diagnostic message exceeds its bound",
|
||||
));
|
||||
}
|
||||
Ok(AcpxEventPayload::Diagnostic {
|
||||
code,
|
||||
message: redact_text(message),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn sanitize_question_set(mut value: Value) -> Value {
|
||||
let Some(question_set) = value.as_object_mut() else {
|
||||
return value;
|
||||
};
|
||||
redact_object_text(question_set, &["title", "description", "submitLabel"]);
|
||||
if let Some(questions) = question_set
|
||||
.get_mut("questions")
|
||||
.and_then(Value::as_array_mut)
|
||||
{
|
||||
for question in questions {
|
||||
let Some(question) = question.as_object_mut() else {
|
||||
continue;
|
||||
};
|
||||
// IDs, answer modes, and validation patterns are protocol values:
|
||||
// changing them would break response correlation or semantics.
|
||||
redact_object_text(question, &["header", "prompt", "helpText"]);
|
||||
if let Some(options) = question.get_mut("options").and_then(Value::as_array_mut) {
|
||||
for option in options {
|
||||
if let Some(option) = option.as_object_mut() {
|
||||
redact_object_text(option, &["label", "description"]);
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(custom_answer) = question
|
||||
.get_mut("customAnswer")
|
||||
.and_then(Value::as_object_mut)
|
||||
{
|
||||
redact_object_text(custom_answer, &["label", "placeholder"]);
|
||||
}
|
||||
}
|
||||
}
|
||||
value
|
||||
}
|
||||
|
||||
fn redact_object_text(object: &mut serde_json::Map<String, Value>, keys: &[&str]) {
|
||||
for key in keys {
|
||||
let Some(text) = object.get(*key).and_then(Value::as_str) else {
|
||||
continue;
|
||||
};
|
||||
let redacted = redact_text(text);
|
||||
object.insert((*key).to_owned(), Value::String(redacted));
|
||||
}
|
||||
}
|
||||
|
||||
fn decode_runtime_event(payload: &Value) -> Result<AcpxEventPayload, LocalRunnerError> {
|
||||
let runtime_type = payload
|
||||
.get("type")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX runtime event omitted its type"))?;
|
||||
let kind = match runtime_type {
|
||||
"text_delta" => {
|
||||
bounded_required_text(payload, "text", MAX_RUNTIME_TEXT_CHARS, "runtime text")?;
|
||||
AcpxRuntimeEventKind::TextDelta
|
||||
}
|
||||
"thinking" => {
|
||||
bounded_required_text(payload, "text", MAX_RUNTIME_TEXT_CHARS, "runtime thought")?;
|
||||
AcpxRuntimeEventKind::Thinking
|
||||
}
|
||||
"plan" => {
|
||||
validate_plan(payload)?;
|
||||
AcpxRuntimeEventKind::Plan
|
||||
}
|
||||
"status" => {
|
||||
bounded_optional_text(payload, "tag", 160, "runtime status tag")?;
|
||||
bounded_optional_text(payload, "text", 4_000, "runtime status text")?;
|
||||
AcpxRuntimeEventKind::Status
|
||||
}
|
||||
"tool_call" => {
|
||||
bounded_optional_text(payload, "toolCallId", 240, "runtime tool call id")?;
|
||||
bounded_optional_text(payload, "title", 4_000, "runtime tool title")?;
|
||||
bounded_optional_text(payload, "status", 100, "runtime tool status")?;
|
||||
if let Some(locations) = optional_array(payload, "locations", "runtime tool locations")?
|
||||
{
|
||||
if locations.len() > 2_000 {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime tool locations exceed their bound",
|
||||
));
|
||||
}
|
||||
for location in locations {
|
||||
if !location.is_object() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime tool location must be an object",
|
||||
));
|
||||
}
|
||||
bounded_optional_text(location, "path", 4_000, "runtime tool path")?;
|
||||
}
|
||||
}
|
||||
AcpxRuntimeEventKind::ToolCall
|
||||
}
|
||||
"semantic_result" => {
|
||||
required_id(payload, "callId", "semantic result call")?;
|
||||
required_id(payload, "operationId", "semantic result operation")?;
|
||||
if !payload.get("result").is_some_and(Value::is_object) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX semantic result must contain an object result",
|
||||
));
|
||||
}
|
||||
AcpxRuntimeEventKind::SemanticResult
|
||||
}
|
||||
"provider_notice" => {
|
||||
bounded_nonempty_text(payload, "category", 160, "provider notice category")?;
|
||||
bounded_nonempty_text(payload, "summary", 4_000, "provider notice summary")?;
|
||||
AcpxRuntimeEventKind::ProviderNotice
|
||||
}
|
||||
"error" => {
|
||||
bounded_optional_text(payload, "code", 160, "runtime error code")?;
|
||||
bounded_nonempty_text(payload, "message", 8_192, "runtime error message")?;
|
||||
AcpxRuntimeEventKind::Error
|
||||
}
|
||||
"done" => {
|
||||
bounded_optional_text(payload, "stopReason", 160, "runtime stop reason")?;
|
||||
AcpxRuntimeEventKind::Done
|
||||
}
|
||||
_ => {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime event type is not admitted",
|
||||
))
|
||||
}
|
||||
};
|
||||
Ok(AcpxEventPayload::Runtime {
|
||||
kind,
|
||||
payload: sanitize_value(payload),
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_plan(payload: &Value) -> Result<(), LocalRunnerError> {
|
||||
let entries = payload
|
||||
.get("entries")
|
||||
.and_then(Value::as_array)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX runtime plan must contain entries"))?;
|
||||
if entries.len() > 256 {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime plan exceeds 256 entries",
|
||||
));
|
||||
}
|
||||
for entry in entries {
|
||||
bounded_nonempty_text(entry, "content", 4_000, "runtime plan content")?;
|
||||
if !matches!(
|
||||
entry.get("status").and_then(Value::as_str),
|
||||
Some("pending" | "in_progress" | "completed")
|
||||
) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime plan contains an unsupported status",
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_question_set(value: &Value) -> Result<(), LocalRunnerError> {
|
||||
let schema: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/schemas/question-set.schema.json"
|
||||
))
|
||||
.map_err(|_| LocalRunnerError::invalid("embedded question-set schema is invalid"))?;
|
||||
let validator = jsonschema::validator_for(&schema)
|
||||
.map_err(|_| LocalRunnerError::invalid("embedded question-set schema cannot compile"))?;
|
||||
if !validator.is_valid(value) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX input request failed the Paperclip question-set schema",
|
||||
));
|
||||
}
|
||||
let mut ids = BTreeSet::new();
|
||||
for question in value
|
||||
.get("questions")
|
||||
.and_then(Value::as_array)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
{
|
||||
let id = question
|
||||
.get("id")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX input question omitted its id"))?;
|
||||
if !ids.insert(id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX input question ids must be unique",
|
||||
));
|
||||
}
|
||||
let mut option_ids = BTreeSet::new();
|
||||
for option in question
|
||||
.get("options")
|
||||
.and_then(Value::as_array)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
{
|
||||
let option_id = option
|
||||
.get("id")
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX input option omitted its id"))?;
|
||||
if !option_ids.insert(option_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX input option ids must be unique within one question",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn required_id(value: &Value, key: &str, label: &str) -> Result<String, LocalRunnerError> {
|
||||
let id = value
|
||||
.get(key)
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("ACPX {label} omitted its identity")))?;
|
||||
if id.is_empty() || id.chars().count() > MAX_ID_CHARS || id.chars().any(char::is_control) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} identity is invalid"
|
||||
)));
|
||||
}
|
||||
Ok(id.to_owned())
|
||||
}
|
||||
|
||||
fn bounded_required_text<'a>(
|
||||
value: &'a Value,
|
||||
key: &str,
|
||||
max_chars: usize,
|
||||
label: &str,
|
||||
) -> Result<&'a str, LocalRunnerError> {
|
||||
let text = value
|
||||
.get(key)
|
||||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("ACPX {label} is required")))?;
|
||||
if text.chars().count() > max_chars {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} exceeds its bound"
|
||||
)));
|
||||
}
|
||||
Ok(text)
|
||||
}
|
||||
|
||||
fn bounded_nonempty_text<'a>(
|
||||
value: &'a Value,
|
||||
key: &str,
|
||||
max_chars: usize,
|
||||
label: &str,
|
||||
) -> Result<&'a str, LocalRunnerError> {
|
||||
let text = bounded_required_text(value, key, max_chars, label)?;
|
||||
if text.trim().is_empty() {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} must not be empty"
|
||||
)));
|
||||
}
|
||||
Ok(text)
|
||||
}
|
||||
|
||||
fn bounded_optional_text(
|
||||
value: &Value,
|
||||
key: &str,
|
||||
max_chars: usize,
|
||||
label: &str,
|
||||
) -> Result<Option<String>, LocalRunnerError> {
|
||||
let Some(field) = value.get(key) else {
|
||||
return Ok(None);
|
||||
};
|
||||
if field.is_null() {
|
||||
return Ok(None);
|
||||
}
|
||||
let text = field
|
||||
.as_str()
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("ACPX {label} must be text")))?;
|
||||
if text.chars().count() > max_chars {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} exceeds its bound"
|
||||
)));
|
||||
}
|
||||
Ok(Some(text.to_owned()))
|
||||
}
|
||||
|
||||
fn optional_object(
|
||||
value: &Value,
|
||||
key: &str,
|
||||
label: &str,
|
||||
) -> Result<Option<Value>, LocalRunnerError> {
|
||||
let Some(field) = value.get(key) else {
|
||||
return Ok(None);
|
||||
};
|
||||
if field.is_null() {
|
||||
return Ok(None);
|
||||
}
|
||||
if !field.is_object() {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} must be an object"
|
||||
)));
|
||||
}
|
||||
Ok(Some(field.clone()))
|
||||
}
|
||||
|
||||
fn optional_array<'a>(
|
||||
value: &'a Value,
|
||||
key: &str,
|
||||
label: &str,
|
||||
) -> Result<Option<&'a Vec<Value>>, LocalRunnerError> {
|
||||
let Some(field) = value.get(key) else {
|
||||
return Ok(None);
|
||||
};
|
||||
if field.is_null() {
|
||||
return Ok(None);
|
||||
}
|
||||
field
|
||||
.as_array()
|
||||
.map(Some)
|
||||
.ok_or_else(|| LocalRunnerError::invalid(format!("ACPX {label} must be an array")))
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
#![forbid(unsafe_code)]
|
||||
|
||||
pub mod acpx_event_payload;
|
||||
pub mod acpx_event_scope;
|
||||
pub mod acpx_sidecar_transport;
|
||||
pub mod codex_provider;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,332 @@
|
|||
use paperclip_runner_core::acpx_event_payload::{
|
||||
decode_acpx_event, AcpxEventPayload, AcpxRuntimeEventKind, AcpxTurnStatus,
|
||||
};
|
||||
use paperclip_runner_core::acpx_event_scope::AcpxEventScope;
|
||||
use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn event(event_type: GeneratedAcpxSidecarEventType, payload: Value) -> AcpxSidecarEvent {
|
||||
AcpxSidecarEvent {
|
||||
sequence: 1,
|
||||
event_type,
|
||||
run_id: Some("run-1".to_owned()),
|
||||
turn_id: Some("turn-1".to_owned()),
|
||||
payload,
|
||||
}
|
||||
}
|
||||
|
||||
fn active_scope() -> AcpxEventScope {
|
||||
let mut scope = AcpxEventScope::new("run-1").unwrap();
|
||||
scope.bind_turn("turn-1").unwrap();
|
||||
scope
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_every_admitted_runtime_event_shape() {
|
||||
let scope = active_scope();
|
||||
let cases = [
|
||||
(
|
||||
json!({"type": "text_delta", "text": "Hello"}),
|
||||
AcpxRuntimeEventKind::TextDelta,
|
||||
),
|
||||
(
|
||||
json!({"type": "thinking", "text": "Inspect"}),
|
||||
AcpxRuntimeEventKind::Thinking,
|
||||
),
|
||||
(
|
||||
json!({"type": "plan", "entries": [{"content": "Inspect", "status": "pending"}]}),
|
||||
AcpxRuntimeEventKind::Plan,
|
||||
),
|
||||
(
|
||||
json!({"type": "status", "tag": "usage_update", "text": "Working"}),
|
||||
AcpxRuntimeEventKind::Status,
|
||||
),
|
||||
(
|
||||
json!({"type": "tool_call", "toolCallId": "tool-1", "status": "completed", "locations": []}),
|
||||
AcpxRuntimeEventKind::ToolCall,
|
||||
),
|
||||
(
|
||||
json!({"type": "semantic_result", "callId": "call-1", "operationId": "paperclip_finish", "result": {"ok": true}}),
|
||||
AcpxRuntimeEventKind::SemanticResult,
|
||||
),
|
||||
(
|
||||
json!({"type": "provider_notice", "category": "provider_update", "summary": "Update"}),
|
||||
AcpxRuntimeEventKind::ProviderNotice,
|
||||
),
|
||||
(
|
||||
json!({"type": "error", "code": "provider_error", "message": "Failed"}),
|
||||
AcpxRuntimeEventKind::Error,
|
||||
),
|
||||
(
|
||||
json!({"type": "done", "stopReason": "end_turn"}),
|
||||
AcpxRuntimeEventKind::Done,
|
||||
),
|
||||
];
|
||||
|
||||
for (payload, expected_kind) in cases {
|
||||
let decoded = decode_acpx_event(
|
||||
&scope,
|
||||
&event(GeneratedAcpxSidecarEventType::RuntimeEvent, payload),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
decoded,
|
||||
AcpxEventPayload::Runtime { kind, .. } if kind == expected_kind
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unclassified_and_malformed_runtime_payloads() {
|
||||
let scope = active_scope();
|
||||
for payload in [
|
||||
json!({"type": "future_event"}),
|
||||
json!({"type": "text_delta", "text": "x".repeat(65_537)}),
|
||||
json!({"type": "plan", "entries": [{"content": "Inspect", "status": "blocked"}]}),
|
||||
json!({"type": "semantic_result", "callId": "call-1", "operationId": "finish", "result": "not-an-object"}),
|
||||
json!({"type": "tool_call", "locations": vec![json!({}); 2_001]}),
|
||||
json!({"type": "tool_call", "locations": {"path": "file.txt"}}),
|
||||
json!({"type": "provider_notice", "category": "", "summary": "Update"}),
|
||||
] {
|
||||
assert!(decode_acpx_event(
|
||||
&scope,
|
||||
&event(GeneratedAcpxSidecarEventType::RuntimeEvent, payload),
|
||||
)
|
||||
.is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_tool_and_permission_requests_after_scope_validation() {
|
||||
let scope = active_scope();
|
||||
let tool = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
|
||||
json!({
|
||||
"callId": "call-1",
|
||||
"operationId": "get_issue",
|
||||
"input": {"issueId": "issue-1", "apiToken": "secret-value"},
|
||||
}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
tool,
|
||||
AcpxEventPayload::ToolCalled { call_id, operation_id, input }
|
||||
if call_id == "call-1"
|
||||
&& operation_id == "get_issue"
|
||||
&& input["apiToken"] == "[REDACTED]"
|
||||
));
|
||||
|
||||
let permission = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimePermissionRequested,
|
||||
json!({
|
||||
"requestId": "permission-1",
|
||||
"kind": "write",
|
||||
"title": "Authorization: Bearer permission-secret"
|
||||
}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
permission,
|
||||
AcpxEventPayload::PermissionRequested { request_id, kind, title, details }
|
||||
if request_id == "permission-1"
|
||||
&& kind == "write"
|
||||
&& title.contains("REDACTED")
|
||||
&& !title.contains("permission-secret")
|
||||
&& details["title"].as_str().is_some_and(|value| value.contains("REDACTED"))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validates_question_sets_and_rejects_duplicate_question_ids() {
|
||||
let scope = active_scope();
|
||||
let question = json!({
|
||||
"id": "choice",
|
||||
"prompt": "Choose one",
|
||||
"required": true,
|
||||
"answerMode": "single_select",
|
||||
"options": [{"id": "one", "label": "One"}],
|
||||
});
|
||||
let valid = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
|
||||
json!({
|
||||
"requestId": "input-1",
|
||||
"questionSet": {"schema": "paperclip.question_set.v1", "questions": [question.clone()]},
|
||||
"origin": {"provider": "codex"},
|
||||
}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
valid,
|
||||
AcpxEventPayload::InputRequested { request_id, .. } if request_id == "input-1"
|
||||
));
|
||||
|
||||
let sensitive = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
|
||||
json!({
|
||||
"requestId": "input-sensitive",
|
||||
"questionSet": {
|
||||
"schema": "paperclip.question_set.v1",
|
||||
"title": "token=question-set-secret",
|
||||
"description": "authorization: question-description-secret",
|
||||
"submitLabel": "secret=submit-label-secret",
|
||||
"questions": [{
|
||||
"id": "token=stable-question-id",
|
||||
"header": "token=question-header-secret",
|
||||
"prompt": "password=question-prompt-secret",
|
||||
"helpText": "api_key=question-help-secret",
|
||||
"required": true,
|
||||
"answerMode": "single_select",
|
||||
"options": [{
|
||||
"id": "token=stable-option-id",
|
||||
"label": "bearer option-label-secret",
|
||||
"description": "ticket=option-description-secret"
|
||||
}],
|
||||
"customAnswer": {
|
||||
"enabled": true,
|
||||
"label": "token=custom-label-secret",
|
||||
"placeholder": "secret=custom-placeholder-secret"
|
||||
},
|
||||
"textValidation": {"pattern": "^token=protocol-value$"}
|
||||
}]
|
||||
}
|
||||
}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
let AcpxEventPayload::InputRequested { question_set, .. } = sensitive else {
|
||||
panic!("input request must retain its question set");
|
||||
};
|
||||
let retained = question_set.to_string();
|
||||
assert!(!retained.contains("question-set-secret"));
|
||||
assert!(!retained.contains("question-prompt-secret"));
|
||||
assert!(!retained.contains("option-label-secret"));
|
||||
assert!(!retained.contains("custom-placeholder-secret"));
|
||||
assert_eq!(
|
||||
question_set["questions"][0]["id"],
|
||||
"token=stable-question-id"
|
||||
);
|
||||
assert_eq!(
|
||||
question_set["questions"][0]["options"][0]["id"],
|
||||
"token=stable-option-id"
|
||||
);
|
||||
assert_eq!(
|
||||
question_set["questions"][0]["textValidation"]["pattern"],
|
||||
"^token=protocol-value$"
|
||||
);
|
||||
|
||||
let duplicate = event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
|
||||
json!({
|
||||
"requestId": "input-2",
|
||||
"questionSet": {"schema": "paperclip.question_set.v1", "questions": [question.clone(), question]},
|
||||
}),
|
||||
);
|
||||
assert!(decode_acpx_event(&scope, &duplicate)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("must be unique"));
|
||||
|
||||
let duplicate_options = event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
|
||||
json!({
|
||||
"requestId": "input-3",
|
||||
"questionSet": {
|
||||
"schema": "paperclip.question_set.v1",
|
||||
"questions": [{
|
||||
"id": "choice",
|
||||
"prompt": "Choose one",
|
||||
"required": true,
|
||||
"answerMode": "single_select",
|
||||
"options": [
|
||||
{"id": "same", "label": "One"},
|
||||
{"id": "same", "label": "Two"},
|
||||
],
|
||||
}],
|
||||
},
|
||||
}),
|
||||
);
|
||||
assert!(decode_acpx_event(&scope, &duplicate_options)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("option ids must be unique"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_terminal_and_redacts_diagnostic_payloads() {
|
||||
let scope = active_scope();
|
||||
let terminal = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
|
||||
json!({"status": "cancelled", "error": {"authorization": "Bearer secret"}}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
terminal,
|
||||
AcpxEventPayload::TurnTerminal { status: AcpxTurnStatus::Cancelled, error: Some(error) }
|
||||
if error["authorization"] == "[REDACTED]"
|
||||
));
|
||||
|
||||
let no_error = decode_acpx_event(
|
||||
&scope,
|
||||
&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
|
||||
json!({"status": "completed", "error": null}),
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
no_error,
|
||||
AcpxEventPayload::TurnTerminal {
|
||||
status: AcpxTurnStatus::Completed,
|
||||
error: None
|
||||
}
|
||||
));
|
||||
|
||||
let mut diagnostic = event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeDiagnostic,
|
||||
json!({"code": "provider_warning", "message": "Authorization: Bearer secret"}),
|
||||
);
|
||||
diagnostic.run_id = None;
|
||||
diagnostic.turn_id = None;
|
||||
let decoded = decode_acpx_event(&scope, &diagnostic).unwrap();
|
||||
assert!(matches!(
|
||||
decoded,
|
||||
AcpxEventPayload::Diagnostic { message, .. }
|
||||
if message == "[REDACTED diagnostic containing a sensitive marker]"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_payloads_before_decoding_when_scope_or_size_is_invalid() {
|
||||
let scope = active_scope();
|
||||
let mut wrong_run = event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
|
||||
json!({"malformed": true}),
|
||||
);
|
||||
wrong_run.run_id = Some("run-2".to_owned());
|
||||
let error = decode_acpx_event(&scope, &wrong_run).unwrap_err();
|
||||
assert!(error.to_string().contains("stale run"));
|
||||
|
||||
let mut oversized = event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeProcess,
|
||||
json!({"output": "x".repeat(256 * 1024)}),
|
||||
);
|
||||
oversized.run_id = None;
|
||||
oversized.turn_id = None;
|
||||
let error = decode_acpx_event(&scope, &oversized).unwrap_err();
|
||||
assert!(error.to_string().contains("256 KiB"));
|
||||
}
|
||||
Loading…
Reference in New Issue