feat(runner): project durable ACPX events (#12424)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The Rust runner now owns validated, scoped ACPX reducer events and safe session suspension > - Durable PRP transport must receive provider-neutral events rather than sidecar-native envelopes > - Semantic calls and questions must retain the exact run, session, turn, item, and provider-request authority used by the durable command stream > - Terminal, result, assistant, process, and diagnostic events also need one reviewed projection boundary > - Permission requests remain impossible under the pinned Codex policy and must fail closed if they reach projection > - This pull request adds only that package-local projection without selecting ACPX in runnerd ## Linked Issues or Issue Description Refs #12422 ## What Changed - Add a validated durable ACPX event projection context bound to one run, normalized session, turn, and item. - Pass already normalized activity events through without reintroducing provider-native envelopes. - Project authorized tool calls into canonical semantic input receipts with exact correlation and content digests. - Project structured questions into provider-neutral `paperclip.runtime_request.v2` events. - Preserve both the public projected request identity and the original provider request identity so responses resolve the exact sidecar request. - Project dynamic semantic operation results as `semantic_tool.result`; only reserved finish/block operations may propose the run result. - Project semantic completion results into `run.result.proposed`. - Project terminal-flushed assistant messages on the final channel and turn terminal states into existing provider-neutral event families. - Project sanitized process metadata and diagnostics into bounded harness diagnostics. - Validate runtime-request origins against their strict durable shape and fall back from empty optional titles to a valid question prompt. - Reject invalid identities, projected-identity collisions, unstable semantic receipt identities, permission requests, and cross-turn projection fail closed. - Add integration coverage across reducer event families, correlation, identity validation, projected question resolution, and pinned-policy denial. - Document the durable projection boundary. - Do not change dependencies, lockfiles, workflows, runnerd selection, server behavior, UI, or migrations. ## Verification - Replay base: `80639f4f69c8938eb74bdc0833df93e0ed91dab3` (`master` after #12422 merged). - Exact replay head: `3cb29581d2bcbc4b47f8069baffd721c6ce4e444`. - Stable patch ID: `92910b56575e67ae83960177d467a565019ba282`. - The exact delta is 22 files, 1,206 additions, and 59 deletions, all in `packages/paperclip-runner`; it contains no lockfile, workflow, server, UI, dependency, or migration change. - `git diff --check` and the Cargo formatting check pass on the replayed delta. - Exact-head GitHub Actions run `33372209037` (attempt 2): **PASSED** with 23/23 jobs passed. - Greptile reviewed exact head `3cb29581d2bcbc4b47f8069baffd721c6ce4e444`: **5/5**, with zero unresolved review threads. - Superagent, contributor trust, Socket, and Snyk security checks: **PASSED**. - No local test result is claimed. GitHub Actions is the authoritative verification environment for this replayed revision. ## Risks - This function accepts reducer output, not raw sidecar frames. Callers must preserve the existing scope-first decode and reduction order. - Semantic input includes the already sanitized provider input while its content receipt uses the same canonical digest. - Structured input preserves the validated provider-neutral question set and sanitized origin. - Noncanonical provider request identities are deterministically projected for PRP while the original identity remains authoritative for the sidecar resolution command. - Existing PRP v1 identifiers remain schema-compatible; the only public ID-schema change widens turn/item limits from 160 to 240 characters. The internal ACPX sidecar wire schema now mirrors the stable IDs its Rust transport already enforced. - The projector verifies event-carried terminal and assistant turn identifiers against the durable context. - No production path invokes this projector in this pull request. Durable command execution remains the next slice. > 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 linked the preceding public PR or described the issue in-PR - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id - [ ] 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
80639f4f69
commit
1ee738cf48
|
|
@ -114,6 +114,9 @@ closed.
|
|||
Safe suspension is available only with no active turn or pending request. The
|
||||
sidecar must return the exact persistent session identity before runnerd
|
||||
terminates the local process.
|
||||
Already validated ACPX reducer events project into provider-neutral durable
|
||||
events only with an exact run, session, turn, and item binding. Raw sidecar
|
||||
envelopes and permission requests are not admitted at this boundary.
|
||||
|
||||
Run the complete contract gate with:
|
||||
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@
|
|||
{
|
||||
"path": "schemas/event.schema.json",
|
||||
"id": "https://paperclip.dev/schemas/prp/v1/event.schema.json",
|
||||
"sha256": "e9827a7c31f004ae91388964a96b4ddf6ab40b85f29bf3922683d5dcf64f2559"
|
||||
"sha256": "d48916630976ff7e850b6e2d7d94131761fda322f5d8a05212651ec46b0db19d"
|
||||
},
|
||||
{
|
||||
"path": "schemas/fixture.schema.json",
|
||||
|
|
@ -70,7 +70,7 @@
|
|||
{
|
||||
"path": "schemas/request.schema.json",
|
||||
"id": "https://paperclip.dev/schemas/prp/v1/request.schema.json",
|
||||
"sha256": "bd41f01c00eddc1b635d3131a7183ddb616ff8fe68d5e348ddd9593a674c8d8e"
|
||||
"sha256": "c8d00e10150c90963f19641c075f50da7f67617dd817f41b356287ad85c53f2b"
|
||||
},
|
||||
{
|
||||
"path": "schemas/result.schema.json",
|
||||
|
|
|
|||
|
|
@ -100,8 +100,8 @@
|
|||
"protocolVersion": { "$ref": "#/$defs/protocolVersion" },
|
||||
"sequence": { "type": "integer", "minimum": 1 },
|
||||
"eventType": { "$ref": "#/$defs/eventType" },
|
||||
"runId": { "type": ["string", "null"], "maxLength": 160 },
|
||||
"turnId": { "type": ["string", "null"], "maxLength": 160 },
|
||||
"runId": { "type": ["string", "null"], "maxLength": 160, "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" },
|
||||
"turnId": { "type": ["string", "null"], "maxLength": 240, "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" },
|
||||
"payload": { "type": "object" }
|
||||
},
|
||||
"additionalProperties": false
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@
|
|||
"sourceKind": { "enum": ["runner", "control_plane"] },
|
||||
"runId": { "type": "string", "minLength": 1, "maxLength": 160 },
|
||||
"normalizedSessionId": { "type": "string", "minLength": 1, "maxLength": 160 },
|
||||
"turnId": { "type": "string", "minLength": 1, "maxLength": 160 },
|
||||
"itemId": { "type": "string", "minLength": 1, "maxLength": 160 },
|
||||
"turnId": { "type": "string", "minLength": 1, "maxLength": 240 },
|
||||
"itemId": { "type": "string", "minLength": 1, "maxLength": 240 },
|
||||
"eventType": {
|
||||
"enum": [
|
||||
"runner.connected", "runner.reconnected", "runner.reconciled", "runner.disconnected", "runner.draining", "runner.backpressure", "runner.suspending", "runner.suspended", "runner.stopped", "runner.diagnostic",
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@
|
|||
},
|
||||
"additionalProperties": false
|
||||
},
|
||||
"turnId": { "type": "string", "minLength": 1, "maxLength": 160 },
|
||||
"itemId": { "type": "string", "minLength": 1, "maxLength": 160 }
|
||||
"turnId": { "type": "string", "minLength": 1, "maxLength": 240 },
|
||||
"itemId": { "type": "string", "minLength": 1, "maxLength": 240 }
|
||||
},
|
||||
"additionalProperties": false
|
||||
},
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ use crate::provider_bridge::semantic_value_digest;
|
|||
|
||||
const MAX_EVENT_PAYLOAD_BYTES: usize = 256 * 1024;
|
||||
const MAX_ID_CHARS: usize = 160;
|
||||
const MAX_INPUT_REQUEST_ID_CHARS: usize = 240;
|
||||
const MAX_RUNTIME_TEXT_CHARS: usize = 64 * 1024;
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
|
|
@ -111,7 +112,12 @@ pub fn decode_acpx_event(
|
|||
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")?,
|
||||
request_id: required_id_with_limit(
|
||||
&event.payload,
|
||||
"requestId",
|
||||
"input request",
|
||||
MAX_INPUT_REQUEST_ID_CHARS,
|
||||
)?,
|
||||
question_set,
|
||||
origin: origin.map(|value| sanitize_value(&value)),
|
||||
})
|
||||
|
|
@ -405,11 +411,20 @@ fn validate_question_set(value: &Value) -> Result<(), LocalRunnerError> {
|
|||
}
|
||||
|
||||
fn required_id(value: &Value, key: &str, label: &str) -> Result<String, LocalRunnerError> {
|
||||
required_id_with_limit(value, key, label, MAX_ID_CHARS)
|
||||
}
|
||||
|
||||
fn required_id_with_limit(
|
||||
value: &Value,
|
||||
key: &str,
|
||||
label: &str,
|
||||
max_chars: usize,
|
||||
) -> 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) {
|
||||
if id.is_empty() || id.chars().count() > max_chars || id.chars().any(char::is_control) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX {label} identity is invalid"
|
||||
)));
|
||||
|
|
|
|||
|
|
@ -3,8 +3,8 @@ use std::collections::BTreeSet;
|
|||
use crate::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use crate::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS};
|
||||
|
||||
const MAX_SCOPE_ID_CHARS: usize = 160;
|
||||
const MAX_SETTLED_TURN_IDS: usize = 4_096;
|
||||
|
||||
/// Holds the run and turn authority used to admit ACPX sidecar events.
|
||||
|
|
@ -22,7 +22,7 @@ pub struct AcpxEventScope {
|
|||
impl AcpxEventScope {
|
||||
pub fn new(run_id: impl Into<String>) -> Result<Self, LocalRunnerError> {
|
||||
let run_id = run_id.into();
|
||||
validate_scope_id(&run_id, "run")?;
|
||||
validate_scope_id(&run_id, "run", SHORT_STABLE_ID_CHARS)?;
|
||||
Ok(Self {
|
||||
run_id,
|
||||
active_turn_id: None,
|
||||
|
|
@ -44,7 +44,7 @@ impl AcpxEventScope {
|
|||
|
||||
pub fn bind_turn(&mut self, turn_id: impl Into<String>) -> Result<(), LocalRunnerError> {
|
||||
let turn_id = turn_id.into();
|
||||
validate_scope_id(&turn_id, "turn")?;
|
||||
validate_scope_id(&turn_id, "turn", DURABLE_STABLE_ID_CHARS)?;
|
||||
match self.active_turn_id.as_deref() {
|
||||
Some(active_turn_id) if active_turn_id == turn_id.as_str() => Ok(()),
|
||||
Some(_) => Err(LocalRunnerError::invalid(
|
||||
|
|
@ -72,7 +72,7 @@ impl AcpxEventScope {
|
|||
&self,
|
||||
turn_id: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
validate_scope_id(turn_id, "turn")?;
|
||||
validate_scope_id(turn_id, "turn", DURABLE_STABLE_ID_CHARS)?;
|
||||
if self.settled_turn_ids.contains(turn_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope reused a settled turn identity",
|
||||
|
|
@ -103,7 +103,7 @@ impl AcpxEventScope {
|
|||
}
|
||||
|
||||
pub fn clear_turn(&mut self, turn_id: &str) -> Result<(), LocalRunnerError> {
|
||||
validate_scope_id(turn_id, "turn")?;
|
||||
validate_scope_id(turn_id, "turn", DURABLE_STABLE_ID_CHARS)?;
|
||||
if self.active_turn_id.as_deref() != Some(turn_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope cannot clear a stale turn",
|
||||
|
|
@ -125,10 +125,10 @@ impl AcpxEventScope {
|
|||
|
||||
pub fn validate_event(&self, event: &AcpxSidecarEvent) -> Result<(), LocalRunnerError> {
|
||||
if let Some(run_id) = event.run_id.as_deref() {
|
||||
validate_scope_id(run_id, "event run")?;
|
||||
validate_scope_id(run_id, "event run", SHORT_STABLE_ID_CHARS)?;
|
||||
}
|
||||
if let Some(turn_id) = event.turn_id.as_deref() {
|
||||
validate_scope_id(turn_id, "event turn")?;
|
||||
validate_scope_id(turn_id, "event turn", DURABLE_STABLE_ID_CHARS)?;
|
||||
}
|
||||
let global_event = matches!(
|
||||
event.event_type,
|
||||
|
|
@ -203,11 +203,8 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn validate_scope_id(value: &str, label: &str) -> Result<(), LocalRunnerError> {
|
||||
if value.is_empty()
|
||||
|| value.chars().count() > MAX_SCOPE_ID_CHARS
|
||||
|| value.chars().any(char::is_control)
|
||||
{
|
||||
fn validate_scope_id(value: &str, label: &str, max_chars: usize) -> Result<(), LocalRunnerError> {
|
||||
if !is_stable_id(value, max_chars) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX event scope {label} id is invalid"
|
||||
)));
|
||||
|
|
|
|||
|
|
@ -4,7 +4,10 @@ use std::time::Duration;
|
|||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent};
|
||||
use crate::acpx_provider_state::{
|
||||
is_reserved_terminal_operation, AcpxProviderState, AcpxProviderStateEvent, PRP_BLOCK_TOOL_NAME,
|
||||
PRP_COMPLETION_TOOL_NAME,
|
||||
};
|
||||
use crate::acpx_sidecar_transport::{AcpxSidecarTransport, AcpxSidecarTransportConfig};
|
||||
use crate::generated_acpx_sidecar_contract::{
|
||||
GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
|
||||
|
|
@ -15,13 +18,12 @@ use crate::provider_bridge::{
|
|||
ToolResult, TOOL_SET_SCHEMA,
|
||||
};
|
||||
use crate::question_response::validate_question_response;
|
||||
use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS};
|
||||
|
||||
const MAX_ID_CHARS: usize = 240;
|
||||
const MAX_MODEL_CHARS: usize = 240;
|
||||
const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024;
|
||||
const MAX_JSON_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
|
||||
const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish";
|
||||
const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block";
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
|
||||
#[serde(rename_all = "kebab-case")]
|
||||
|
|
@ -73,10 +75,10 @@ impl AcpxProviderSessionConfig {
|
|||
));
|
||||
}
|
||||
validate_text(&self.model, MAX_MODEL_CHARS, "ACPX model")?;
|
||||
validate_text(&self.run_id, 160, "ACPX run id")?;
|
||||
validate_text(
|
||||
validate_stable_id(&self.run_id, SHORT_STABLE_ID_CHARS, "ACPX run id")?;
|
||||
validate_stable_id(
|
||||
&self.normalized_session_id,
|
||||
160,
|
||||
SHORT_STABLE_ID_CHARS,
|
||||
"ACPX normalized session id",
|
||||
)?;
|
||||
if self.catalog_revision == 0 || self.catalog_revision > MAX_JSON_SAFE_INTEGER {
|
||||
|
|
@ -247,7 +249,7 @@ impl AcpxProviderSession {
|
|||
working_directory: &Path,
|
||||
) -> Result<Value, LocalRunnerError> {
|
||||
self.ensure_open()?;
|
||||
validate_text(turn_id, 160, "ACPX turn id")?;
|
||||
validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?;
|
||||
validate_turn_message(message)?;
|
||||
if working_directory != self.working_directory {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
|
|
@ -342,7 +344,7 @@ impl AcpxProviderSession {
|
|||
reason: &str,
|
||||
) -> Result<Value, LocalRunnerError> {
|
||||
self.ensure_open()?;
|
||||
validate_text(turn_id, 160, "ACPX turn id")?;
|
||||
validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?;
|
||||
if self.state.active_turn_id() != Some(turn_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX interruption named a stale or inactive turn",
|
||||
|
|
@ -572,7 +574,17 @@ impl AcpxProviderSession {
|
|||
resolution: &Value,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
self.ensure_bound_turn(turn_id)?;
|
||||
validate_text(request_id, 240, "ACPX input request id")?;
|
||||
validate_text(request_id, SHORT_STABLE_ID_CHARS, "ACPX input request id")?;
|
||||
if !is_stable_id(request_id, SHORT_STABLE_ID_CHARS) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX input request id is not a stable runtime request identity",
|
||||
));
|
||||
}
|
||||
let provider_request_id = self
|
||||
.state
|
||||
.pending_provider_input_request_id(request_id)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX input request is stale or unknown"))?
|
||||
.to_owned();
|
||||
let question_set = self
|
||||
.state
|
||||
.pending_question_set(request_id)
|
||||
|
|
@ -582,7 +594,7 @@ impl AcpxProviderSession {
|
|||
next_state.complete_input(request_id)?;
|
||||
let response = match self.transport.request(
|
||||
GeneratedAcpxSidecarCommand::InputResolve,
|
||||
json!({"requestId":request_id,"turnId":turn_id,"resolution":resolution}),
|
||||
json!({"requestId":provider_request_id,"turnId":turn_id,"resolution":resolution}),
|
||||
) {
|
||||
Ok(response) => response,
|
||||
Err(error) => return Err(self.fail_closed(error)),
|
||||
|
|
@ -727,7 +739,7 @@ impl AcpxProviderSession {
|
|||
}
|
||||
|
||||
fn ensure_bound_turn(&self, turn_id: &str) -> Result<(), LocalRunnerError> {
|
||||
validate_text(turn_id, 160, "ACPX turn id")?;
|
||||
validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?;
|
||||
if self.ensure_active_turn()? != turn_id {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX resolution named a stale or inactive turn",
|
||||
|
|
@ -755,10 +767,6 @@ fn is_reserved_terminal_result(result: &crate::acpx_provider_state::AcpxSemantic
|
|||
is_reserved_terminal_operation(&result.operation_id)
|
||||
}
|
||||
|
||||
fn is_reserved_terminal_operation(operation_id: &str) -> bool {
|
||||
matches!(operation_id, PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME)
|
||||
}
|
||||
|
||||
fn validate_reserved_terminal_result(
|
||||
state: &AcpxProviderState,
|
||||
result: &crate::acpx_provider_state::AcpxSemanticResult,
|
||||
|
|
@ -1019,6 +1027,13 @@ fn validate_text(value: &str, max_chars: usize, label: &str) -> Result<(), Local
|
|||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_stable_id(value: &str, max_chars: usize, label: &str) -> Result<(), LocalRunnerError> {
|
||||
if !is_stable_id(value, max_chars) {
|
||||
return Err(LocalRunnerError::invalid(format!("{label} is invalid")));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn is_sha256_digest(value: &str) -> bool {
|
||||
value.len() == 71
|
||||
&& value.starts_with("sha256:")
|
||||
|
|
|
|||
|
|
@ -10,14 +10,19 @@ use crate::acpx_sidecar_transport::AcpxSidecarEvent;
|
|||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::provider_bridge::ToolResult;
|
||||
use crate::provider_events::{normalize_acpx_runtime_event, NormalizedProviderEvent};
|
||||
use crate::stable_identity::project_acpx_runtime_request_id;
|
||||
|
||||
const MAX_ASSISTANT_TEXT_BYTES: usize = 1024 * 1024;
|
||||
const MAX_PENDING_TOOLS: usize = 4_096;
|
||||
const MAX_PENDING_TOOL_INPUT_BYTES: usize = 16 * 1024 * 1024;
|
||||
const MAX_PENDING_RUNTIME_REQUESTS: usize = 1_024;
|
||||
const MAX_PENDING_RUNTIME_REQUEST_BYTES: usize = 16 * 1024 * 1024;
|
||||
const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish";
|
||||
const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block";
|
||||
pub(crate) const PRP_COMPLETION_TOOL_NAME: &str = "paperclip_finish";
|
||||
pub(crate) const PRP_BLOCK_TOOL_NAME: &str = "paperclip_block";
|
||||
|
||||
pub(crate) fn is_reserved_terminal_operation(operation_id: &str) -> bool {
|
||||
matches!(operation_id, PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME)
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct AcpxPendingTool {
|
||||
|
|
@ -75,6 +80,7 @@ pub enum AcpxProviderStateEvent {
|
|||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct PendingInput {
|
||||
runtime_request_id: String,
|
||||
value_bytes: usize,
|
||||
question_set: Value,
|
||||
}
|
||||
|
|
@ -204,6 +210,15 @@ impl AcpxProviderState {
|
|||
} => {
|
||||
let value_bytes = value_bytes(&details)?;
|
||||
self.admit_runtime_request(&request_id, value_bytes)?;
|
||||
if self
|
||||
.pending_inputs
|
||||
.values()
|
||||
.any(|pending| pending.runtime_request_id == request_id)
|
||||
{
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX permission request identity collides with a pending input projection",
|
||||
));
|
||||
}
|
||||
self.pending_permissions
|
||||
.insert(request_id.clone(), value_bytes);
|
||||
self.pending_runtime_request_bytes += value_bytes;
|
||||
|
|
@ -221,11 +236,24 @@ impl AcpxProviderState {
|
|||
} => {
|
||||
let value_bytes = value_bytes(&question_set)?;
|
||||
self.admit_runtime_request(&request_id, value_bytes)?;
|
||||
let runtime_request_id = project_acpx_runtime_request_id(&request_id)
|
||||
.expect("a decoded ACPX input request has a bounded identity");
|
||||
if self
|
||||
.pending_inputs
|
||||
.values()
|
||||
.any(|pending| pending.runtime_request_id == runtime_request_id)
|
||||
|| self.pending_permissions.contains_key(&runtime_request_id)
|
||||
{
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX input request identity collides after durable projection",
|
||||
));
|
||||
}
|
||||
if self
|
||||
.pending_inputs
|
||||
.insert(
|
||||
request_id.clone(),
|
||||
PendingInput {
|
||||
runtime_request_id,
|
||||
value_bytes,
|
||||
question_set: question_set.clone(),
|
||||
},
|
||||
|
|
@ -354,9 +382,16 @@ impl AcpxProviderState {
|
|||
}
|
||||
|
||||
pub fn complete_input(&mut self, request_id: &str) -> Result<(), LocalRunnerError> {
|
||||
let provider_request_id = self
|
||||
.pending_inputs
|
||||
.iter()
|
||||
.find_map(|(provider_request_id, pending)| {
|
||||
(pending.runtime_request_id == request_id).then(|| provider_request_id.clone())
|
||||
})
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?;
|
||||
let pending = self
|
||||
.pending_inputs
|
||||
.remove(request_id)
|
||||
.remove(&provider_request_id)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?;
|
||||
self.pending_runtime_request_bytes = self
|
||||
.pending_runtime_request_bytes
|
||||
|
|
@ -366,10 +401,18 @@ impl AcpxProviderState {
|
|||
|
||||
pub fn pending_question_set(&self, request_id: &str) -> Option<&Value> {
|
||||
self.pending_inputs
|
||||
.get(request_id)
|
||||
.values()
|
||||
.find(|pending| pending.runtime_request_id == request_id)
|
||||
.map(|pending| &pending.question_set)
|
||||
}
|
||||
|
||||
pub(crate) fn pending_provider_input_request_id(&self, request_id: &str) -> Option<&str> {
|
||||
self.pending_inputs
|
||||
.iter()
|
||||
.find(|(_, pending)| pending.runtime_request_id == request_id)
|
||||
.map(|(provider_request_id, _)| provider_request_id.as_str())
|
||||
}
|
||||
|
||||
pub fn semantic_result(&self) -> Option<&AcpxSemanticResult> {
|
||||
self.semantic_result.as_ref()
|
||||
}
|
||||
|
|
@ -428,10 +471,7 @@ impl AcpxProviderState {
|
|||
result_digest: semantic_result_digest
|
||||
.expect("decoded ACPX semantic result has a raw correlation digest"),
|
||||
};
|
||||
if matches!(
|
||||
result.operation_id.as_str(),
|
||||
PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME
|
||||
) {
|
||||
if is_reserved_terminal_operation(&result.operation_id) {
|
||||
return match self.semantic_result.as_ref() {
|
||||
None => {
|
||||
self.semantic_result = Some(result.clone());
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ use crate::generated_acpx_sidecar_contract::{
|
|||
};
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::process_supervisor::{BoundedLogBuffer, ProcessOutput, SupervisedProcess};
|
||||
use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS};
|
||||
|
||||
pub const ACPX_SIDECAR_MAX_FRAME_BYTES: usize = 1024 * 1024;
|
||||
const MAX_BUFFERED_EVENTS: usize = 512;
|
||||
|
|
@ -441,8 +442,8 @@ fn parse_frame(line: &str) -> Result<ParsedFrame, LocalRunnerError> {
|
|||
"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")?;
|
||||
let run_id = nullable_identifier(frame.run_id, "event runId", SHORT_STABLE_ID_CHARS)?;
|
||||
let turn_id = nullable_identifier(frame.turn_id, "event turnId", DURABLE_STABLE_ID_CHARS)?;
|
||||
if !frame.payload.is_object() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX sidecar event payload must be an object",
|
||||
|
|
@ -509,7 +510,11 @@ fn parse_frame(line: &str) -> Result<ParsedFrame, LocalRunnerError> {
|
|||
Ok(ParsedFrame::Response(frame))
|
||||
}
|
||||
|
||||
fn nullable_identifier(value: Value, field: &str) -> Result<Option<String>, LocalRunnerError> {
|
||||
fn nullable_identifier(
|
||||
value: Value,
|
||||
field: &str,
|
||||
max_chars: usize,
|
||||
) -> Result<Option<String>, LocalRunnerError> {
|
||||
if value.is_null() {
|
||||
return Ok(None);
|
||||
}
|
||||
|
|
@ -518,7 +523,7 @@ fn nullable_identifier(value: Value, field: &str) -> Result<Option<String>, Loca
|
|||
"ACPX sidecar {field} must be a string or null"
|
||||
)));
|
||||
};
|
||||
if value.chars().count() > 160 || value.chars().any(char::is_control) {
|
||||
if !is_stable_id(value, max_chars) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX sidecar {field} is invalid"
|
||||
)));
|
||||
|
|
|
|||
|
|
@ -4,6 +4,8 @@ use std::time::Duration;
|
|||
use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
const PROJECTED_INPUT_PROVIDER_ID: &str = "input / réquest";
|
||||
|
||||
fn main() {
|
||||
if let Err(error) = run() {
|
||||
eprintln!("fake-acpx-sidecar: {error}");
|
||||
|
|
@ -110,6 +112,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|||
| "turns-permission"
|
||||
| "resolutions"
|
||||
| "resolutions-error-redaction"
|
||||
| "resolutions-projected-id"
|
||||
| "resolutions-wrong-ack"
|
||||
| "suspend"
|
||||
| "suspend-wrong-ack"
|
||||
|
|
@ -454,7 +457,10 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|||
if command == "turn.start"
|
||||
&& matches!(
|
||||
mode,
|
||||
"resolutions" | "resolutions-error-redaction" | "resolutions-wrong-ack"
|
||||
"resolutions"
|
||||
| "resolutions-error-redaction"
|
||||
| "resolutions-projected-id"
|
||||
| "resolutions-wrong-ack"
|
||||
)
|
||||
{
|
||||
for (event_type, payload) in [
|
||||
|
|
@ -469,7 +475,11 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|||
(
|
||||
"runtime.input_requested",
|
||||
json!({
|
||||
"requestId":"input-1",
|
||||
"requestId":if mode == "resolutions-projected-id" {
|
||||
PROJECTED_INPUT_PROVIDER_ID
|
||||
} else {
|
||||
"input-1"
|
||||
},
|
||||
"questionSet":{
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"questions":[{
|
||||
|
|
@ -601,7 +611,11 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val
|
|||
mode != "resolutions-wrong-ack"
|
||||
}
|
||||
}),
|
||||
"input.resolve" => json!({"resolved":true}),
|
||||
"input.resolve" => json!({
|
||||
"resolved": mode != "resolutions-projected-id"
|
||||
|| params.get("requestId").and_then(Value::as_str)
|
||||
== Some(PROJECTED_INPUT_PROVIDER_ID),
|
||||
}),
|
||||
"session.close" => json!({"closed":true}),
|
||||
_ => json!({"command":command,"params":params}),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ use std::fmt::{self, Display, Formatter};
|
|||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS};
|
||||
|
||||
pub use runner::{run_durable_runner, CommandExecution, CommandExecutor, PolledEvent};
|
||||
pub(crate) use state::{
|
||||
create_private_temporary_file, open_private_regular_file, redact_text, sanitize_value,
|
||||
|
|
@ -131,6 +133,22 @@ impl DurableRunnerConfig {
|
|||
)));
|
||||
}
|
||||
}
|
||||
for (name, value, max_chars) in [
|
||||
("run_id", self.run_id.as_str(), SHORT_STABLE_ID_CHARS),
|
||||
(
|
||||
"normalized_session_id",
|
||||
self.normalized_session_id.as_str(),
|
||||
SHORT_STABLE_ID_CHARS,
|
||||
),
|
||||
("turn_id", self.turn_id.as_str(), DURABLE_STABLE_ID_CHARS),
|
||||
("item_id", self.item_id.as_str(), DURABLE_STABLE_ID_CHARS),
|
||||
] {
|
||||
if !is_stable_id(value, max_chars) {
|
||||
return Err(DurableRunnerError::invalid(format!(
|
||||
"{name} must be a stable identity no longer than {max_chars} characters"
|
||||
)));
|
||||
}
|
||||
}
|
||||
if self.max_outbox_bytes == 0
|
||||
|| self.max_outbox_bytes > MAX_OUTBOX_BYTES
|
||||
|| self.p0_reserve_bytes >= self.max_outbox_bytes
|
||||
|
|
@ -162,3 +180,59 @@ impl DurableRunnerConfig {
|
|||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn config() -> DurableRunnerConfig {
|
||||
DurableRunnerConfig {
|
||||
connect_url: "ws://127.0.0.1/runner".to_owned(),
|
||||
state_dir: PathBuf::from("state"),
|
||||
runner_instance_id: "runner-1".to_owned(),
|
||||
environment_lease_id: "lease-1".to_owned(),
|
||||
run_id: "run-1".to_owned(),
|
||||
normalized_session_id: "session-1".to_owned(),
|
||||
turn_id: "turn-1".to_owned(),
|
||||
item_id: "item-1".to_owned(),
|
||||
runner_version: "1.0.0".to_owned(),
|
||||
runner_digest: "sha256:digest".to_owned(),
|
||||
max_outbox_bytes: 1024 * 1024,
|
||||
p0_reserve_bytes: 64 * 1024,
|
||||
max_frame_bytes: 64 * 1024,
|
||||
reconnect_delay: Duration::from_millis(1),
|
||||
max_runtime: Duration::from_secs(60),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validates_durable_event_identity_boundaries() {
|
||||
let mut boundary = config();
|
||||
boundary.run_id = "r".repeat(SHORT_STABLE_ID_CHARS);
|
||||
boundary.normalized_session_id = "s".repeat(SHORT_STABLE_ID_CHARS);
|
||||
boundary.turn_id = "t".repeat(DURABLE_STABLE_ID_CHARS);
|
||||
boundary.item_id = "i".repeat(DURABLE_STABLE_ID_CHARS);
|
||||
boundary.validate().unwrap();
|
||||
|
||||
for (field, invalid) in [
|
||||
("run_id", "run 1".to_owned()),
|
||||
("normalized_session_id", "session/1".to_owned()),
|
||||
("turn_id", "_turn-1".to_owned()),
|
||||
("item_id", "itém-1".to_owned()),
|
||||
("run_id", "r".repeat(SHORT_STABLE_ID_CHARS + 1)),
|
||||
("turn_id", "t".repeat(DURABLE_STABLE_ID_CHARS + 1)),
|
||||
] {
|
||||
let mut invalid_config = config();
|
||||
match field {
|
||||
"run_id" => invalid_config.run_id = invalid,
|
||||
"normalized_session_id" => invalid_config.normalized_session_id = invalid,
|
||||
"turn_id" => invalid_config.turn_id = invalid,
|
||||
"item_id" => invalid_config.item_id = invalid,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
let error = invalid_config.validate().unwrap_err().to_string();
|
||||
assert!(error.contains(field), "{error}");
|
||||
assert!(error.contains("stable identity"), "{error}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ pub mod provider_bridge;
|
|||
pub mod provider_events;
|
||||
pub mod question_response;
|
||||
pub mod replay;
|
||||
mod stable_identity;
|
||||
|
||||
use std::error::Error;
|
||||
use std::fmt::{self, Display, Formatter};
|
||||
|
|
|
|||
|
|
@ -1,8 +1,14 @@
|
|||
use serde_json::{json, Value};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::acpx_event_payload::AcpxRuntimeEventKind;
|
||||
use crate::durable::{redact_text, EventPriority};
|
||||
use crate::acpx_event_payload::{AcpxRuntimeEventKind, AcpxTurnStatus};
|
||||
use crate::acpx_provider_state::{is_reserved_terminal_operation, AcpxProviderStateEvent};
|
||||
use crate::durable::{redact_text, sanitize_value, EventPriority};
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::provider_bridge::{semantic_value_digest, ToolResult};
|
||||
use crate::stable_identity::{
|
||||
is_stable_id, project_acpx_runtime_request_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS,
|
||||
};
|
||||
|
||||
const MAX_TEXT_CHARS: usize = 4_000;
|
||||
|
||||
|
|
@ -36,6 +42,331 @@ pub(crate) fn normalized_codex_terminal_event_type(
|
|||
})
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct AcpxEventProjectionContext {
|
||||
pub run_id: String,
|
||||
pub normalized_session_id: String,
|
||||
pub turn_id: String,
|
||||
pub item_id: String,
|
||||
}
|
||||
|
||||
impl AcpxEventProjectionContext {
|
||||
pub fn validate(&self) -> Result<(), LocalRunnerError> {
|
||||
for (value, label, max_chars) in [
|
||||
(&self.run_id, "run", SHORT_STABLE_ID_CHARS),
|
||||
(
|
||||
&self.normalized_session_id,
|
||||
"normalized session",
|
||||
SHORT_STABLE_ID_CHARS,
|
||||
),
|
||||
// Turn and item correlation use the durable 240-character
|
||||
// identity contract shared by PRP events, runtime requests, and
|
||||
// semantic-tool receipts.
|
||||
(&self.turn_id, "turn", DURABLE_STABLE_ID_CHARS),
|
||||
(&self.item_id, "item", DURABLE_STABLE_ID_CHARS),
|
||||
] {
|
||||
validate_projection_identity(value, label, max_chars)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn correlation(&self) -> Value {
|
||||
json!({
|
||||
"runId": self.run_id,
|
||||
"normalizedSessionId": self.normalized_session_id,
|
||||
"turnId": self.turn_id,
|
||||
"itemId": self.item_id,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Projects already scope-checked ACPX reducer output into provider-neutral
|
||||
/// durable events. The reducer remains authoritative for bounds and request
|
||||
/// state; this boundary must not accept raw sidecar envelopes.
|
||||
pub fn project_acpx_state_event(
|
||||
context: &AcpxEventProjectionContext,
|
||||
event: &AcpxProviderStateEvent,
|
||||
) -> Result<Vec<NormalizedProviderEvent>, LocalRunnerError> {
|
||||
context.validate()?;
|
||||
let one = |event_type: &str, priority: EventPriority, payload: Value| {
|
||||
Ok(vec![NormalizedProviderEvent {
|
||||
event_type: event_type.to_owned(),
|
||||
priority,
|
||||
payload,
|
||||
}])
|
||||
};
|
||||
match event {
|
||||
AcpxProviderStateEvent::Activity(event) => Ok(vec![event.clone()]),
|
||||
AcpxProviderStateEvent::ToolCall {
|
||||
call_id,
|
||||
operation_id,
|
||||
input,
|
||||
} => {
|
||||
validate_semantic_projection_identity(call_id, operation_id)?;
|
||||
one(
|
||||
"semantic_tool.input",
|
||||
EventPriority::P0,
|
||||
json!({
|
||||
"semantic_tool": {
|
||||
"schema": "paperclip.prp.semantic_tool.v1",
|
||||
"schemaVersion": 1,
|
||||
"phase": "input",
|
||||
"operationId": operation_id,
|
||||
"callId": call_id,
|
||||
"correlation": context.correlation(),
|
||||
"idempotencyKey": Value::Null,
|
||||
"content": {
|
||||
"digest": semantic_value_digest(input),
|
||||
"redactionDisposition": "digest_only",
|
||||
"references": [],
|
||||
},
|
||||
"input": input,
|
||||
},
|
||||
}),
|
||||
)
|
||||
}
|
||||
AcpxProviderStateEvent::ToolResult(result) => {
|
||||
Ok(vec![project_acpx_tool_result(context, result)?])
|
||||
}
|
||||
AcpxProviderStateEvent::PermissionRequest { .. } => Err(LocalRunnerError::invalid(
|
||||
"ACPX permission request reached projection outside the pinned runner policy",
|
||||
)),
|
||||
AcpxProviderStateEvent::InputRequest {
|
||||
request_id,
|
||||
question_set,
|
||||
origin,
|
||||
} => {
|
||||
let request_id = project_acpx_runtime_request_id(request_id).ok_or_else(|| {
|
||||
LocalRunnerError::invalid("ACPX event projection request identity is invalid")
|
||||
})?;
|
||||
let prompt = question_set
|
||||
.get("title")
|
||||
.and_then(Value::as_str)
|
||||
.filter(|value| !value.is_empty())
|
||||
.or_else(|| {
|
||||
question_set
|
||||
.pointer("/questions/0/prompt")
|
||||
.and_then(Value::as_str)
|
||||
})
|
||||
.map(|value| bounded_text(value, MAX_TEXT_CHARS))
|
||||
.unwrap_or_else(|| "Codex needs your input".to_owned());
|
||||
let origin = project_runtime_request_origin(origin.as_ref())?;
|
||||
one(
|
||||
"runtime_request.created",
|
||||
EventPriority::P0,
|
||||
json!({
|
||||
"request": {
|
||||
"schema": "paperclip.runtime_request.v2",
|
||||
"requestKind": "runtime",
|
||||
"requestId": request_id,
|
||||
"turnId": context.turn_id,
|
||||
"itemId": context.item_id,
|
||||
"type": "input",
|
||||
"status": "pending",
|
||||
"prompt": prompt,
|
||||
"input": question_set,
|
||||
"origin": origin,
|
||||
},
|
||||
}),
|
||||
)
|
||||
}
|
||||
AcpxProviderStateEvent::SemanticResult(result) => {
|
||||
validate_semantic_projection_identity(&result.call_id, &result.operation_id)?;
|
||||
if is_reserved_terminal_operation(&result.operation_id) {
|
||||
one(
|
||||
"run.result.proposed",
|
||||
EventPriority::P0,
|
||||
result.result.clone(),
|
||||
)
|
||||
} else {
|
||||
Ok(vec![project_acpx_tool_result(
|
||||
context,
|
||||
&ToolResult {
|
||||
call_id: result.call_id.clone(),
|
||||
operation_id: result.operation_id.clone(),
|
||||
result: result.result.clone(),
|
||||
is_error: !result.ok,
|
||||
},
|
||||
)?])
|
||||
}
|
||||
}
|
||||
AcpxProviderStateEvent::AssistantMessage { turn_id, text } => {
|
||||
require_projected_turn(context, turn_id)?;
|
||||
one(
|
||||
"item.completed",
|
||||
EventPriority::P1,
|
||||
json!({
|
||||
"provider": "acpx",
|
||||
"itemId": context.item_id,
|
||||
"kind": "agentMessage",
|
||||
"status": "completed",
|
||||
"channel": "final",
|
||||
"text": bounded_text(text, MAX_TEXT_CHARS),
|
||||
}),
|
||||
)
|
||||
}
|
||||
AcpxProviderStateEvent::TurnTerminal {
|
||||
turn_id,
|
||||
status,
|
||||
error,
|
||||
} => {
|
||||
require_projected_turn(context, turn_id)?;
|
||||
let (event_type, status) = match status {
|
||||
AcpxTurnStatus::Completed => ("turn.completed", "completed"),
|
||||
AcpxTurnStatus::Failed => ("turn.failed", "failed"),
|
||||
AcpxTurnStatus::Cancelled => ("turn.cancelled", "cancelled"),
|
||||
AcpxTurnStatus::Interrupted => ("turn.interrupted", "interrupted"),
|
||||
};
|
||||
one(
|
||||
event_type,
|
||||
EventPriority::P0,
|
||||
json!({
|
||||
"provider": "acpx",
|
||||
"providerTurnId": turn_id,
|
||||
"status": status,
|
||||
"error": error,
|
||||
}),
|
||||
)
|
||||
}
|
||||
AcpxProviderStateEvent::Process(details) => one(
|
||||
"harness.diagnostic",
|
||||
EventPriority::P1,
|
||||
json!({
|
||||
"code": "acpx_process",
|
||||
"message": "The ACPX sidecar reported provider process metadata.",
|
||||
"details": details,
|
||||
}),
|
||||
),
|
||||
AcpxProviderStateEvent::Diagnostic { code, message } => one(
|
||||
"harness.diagnostic",
|
||||
EventPriority::P1,
|
||||
json!({"code": code, "message": bounded_text(message, MAX_TEXT_CHARS)}),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
fn project_acpx_tool_result(
|
||||
context: &AcpxEventProjectionContext,
|
||||
result: &ToolResult,
|
||||
) -> Result<NormalizedProviderEvent, LocalRunnerError> {
|
||||
validate_semantic_projection_identity(&result.call_id, &result.operation_id)?;
|
||||
let safe_result = sanitize_value(&result.result);
|
||||
Ok(NormalizedProviderEvent {
|
||||
event_type: "semantic_tool.result".to_owned(),
|
||||
priority: EventPriority::P0,
|
||||
payload: json!({
|
||||
"semantic_tool": {
|
||||
"schema": "paperclip.prp.semantic_tool.v1",
|
||||
"schemaVersion": 1,
|
||||
"phase": "result",
|
||||
"operationId": result.operation_id,
|
||||
"callId": result.call_id,
|
||||
"correlation": context.correlation(),
|
||||
"idempotencyKey": Value::Null,
|
||||
"content": {
|
||||
"digest": semantic_value_digest(&safe_result),
|
||||
"redactionDisposition": "digest_only",
|
||||
"references": [],
|
||||
},
|
||||
"outcome": if result.is_error { "failed" } else { "succeeded" },
|
||||
"code": if result.is_error { "semantic_tool_failed" } else { "semantic_tool_succeeded" },
|
||||
"retryable": false,
|
||||
"authorizationBoundary": "active_task",
|
||||
"operationReceiptId": format!("operation_{}", result.call_id),
|
||||
},
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_semantic_projection_identity(
|
||||
call_id: &str,
|
||||
operation_id: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
for (value, label) in [(call_id, "call"), (operation_id, "operation")] {
|
||||
if !is_stable_id(value, SHORT_STABLE_ID_CHARS) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX semantic {label} identity is invalid"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn project_runtime_request_origin(origin: Option<&Value>) -> Result<Value, LocalRunnerError> {
|
||||
let Some(origin) = origin else {
|
||||
return Ok(json!({
|
||||
"adapter": "codex-acpx",
|
||||
"provider": "codex",
|
||||
"method": "runtime.input_requested",
|
||||
}));
|
||||
};
|
||||
let object = origin.as_object().ok_or_else(|| {
|
||||
LocalRunnerError::invalid("ACPX runtime request origin must be an object")
|
||||
})?;
|
||||
if object
|
||||
.keys()
|
||||
.any(|key| !matches!(key.as_str(), "adapter" | "provider" | "method"))
|
||||
{
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX runtime request origin contains unsupported fields",
|
||||
));
|
||||
}
|
||||
validate_origin_field(object.get("adapter"), "adapter", 160, true)?;
|
||||
validate_origin_field(object.get("provider"), "provider", 160, false)?;
|
||||
validate_origin_field(object.get("method"), "method", 500, false)?;
|
||||
Ok(origin.clone())
|
||||
}
|
||||
|
||||
fn validate_origin_field(
|
||||
value: Option<&Value>,
|
||||
field: &str,
|
||||
max_chars: usize,
|
||||
required: bool,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
let Some(value) = value else {
|
||||
if required {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX runtime request origin omitted {field}"
|
||||
)));
|
||||
}
|
||||
return Ok(());
|
||||
};
|
||||
let text = value.as_str().ok_or_else(|| {
|
||||
LocalRunnerError::invalid(format!("ACPX runtime request origin {field} must be text"))
|
||||
})?;
|
||||
if text.is_empty() || text.chars().count() > max_chars {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX runtime request origin {field} is invalid"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_projection_identity(
|
||||
value: &str,
|
||||
label: &str,
|
||||
max_chars: usize,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if !is_stable_id(value, max_chars) {
|
||||
return Err(LocalRunnerError::invalid(format!(
|
||||
"ACPX event projection {label} identity is invalid"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn require_projected_turn(
|
||||
context: &AcpxEventProjectionContext,
|
||||
turn_id: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if turn_id != context.turn_id {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX state event does not match its durable turn projection",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn bounded_text(value: &str, max_chars: usize) -> String {
|
||||
redact_text(value).chars().take(max_chars).collect()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,39 @@
|
|||
use sha2::{Digest, Sha256};
|
||||
|
||||
pub(crate) const SHORT_STABLE_ID_CHARS: usize = 160;
|
||||
pub(crate) const DURABLE_STABLE_ID_CHARS: usize = 240;
|
||||
|
||||
const ACPX_RUNTIME_REQUEST_DOMAIN: &[u8] = b"paperclip.acpx.runtime-request.v1\0";
|
||||
const ACPX_RUNTIME_REQUEST_PREFIX: &str = "acpx-request-";
|
||||
|
||||
pub(crate) fn is_stable_id(value: &str, max_chars: usize) -> bool {
|
||||
let mut characters = value.chars();
|
||||
characters
|
||||
.next()
|
||||
.is_some_and(|character| character.is_ascii_alphanumeric())
|
||||
&& characters
|
||||
.all(|character| character.is_ascii_alphanumeric() || "._:-".contains(character))
|
||||
&& value.len() <= max_chars
|
||||
}
|
||||
|
||||
/// Preserves already-canonical runtime request IDs and deterministically
|
||||
/// projects other bounded ACPX IDs into the stricter PRP request namespace.
|
||||
/// The upstream ID remains authoritative for the sidecar resolution command.
|
||||
pub(crate) fn project_acpx_runtime_request_id(value: &str) -> Option<String> {
|
||||
if value.is_empty()
|
||||
|| value.chars().count() > DURABLE_STABLE_ID_CHARS
|
||||
|| value.chars().any(char::is_control)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
if is_stable_id(value, SHORT_STABLE_ID_CHARS) {
|
||||
return Some(value.to_owned());
|
||||
}
|
||||
let mut digest = Sha256::new();
|
||||
digest.update(ACPX_RUNTIME_REQUEST_DOMAIN);
|
||||
digest.update(value.as_bytes());
|
||||
Some(format!(
|
||||
"{ACPX_RUNTIME_REQUEST_PREFIX}{:x}",
|
||||
digest.finalize()
|
||||
))
|
||||
}
|
||||
|
|
@ -0,0 +1,471 @@
|
|||
use paperclip_runner_core::acpx_event_payload::AcpxTurnStatus;
|
||||
use paperclip_runner_core::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent};
|
||||
use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use paperclip_runner_core::durable::EventPriority;
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use paperclip_runner_core::provider_bridge::ToolResult;
|
||||
use paperclip_runner_core::provider_events::{
|
||||
project_acpx_state_event, AcpxEventProjectionContext, NormalizedProviderEvent,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn context() -> AcpxEventProjectionContext {
|
||||
AcpxEventProjectionContext {
|
||||
run_id: "run-1".to_owned(),
|
||||
normalized_session_id: "session-1".to_owned(),
|
||||
turn_id: "turn-1".to_owned(),
|
||||
item_id: "item-1".to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn project(event: AcpxProviderStateEvent) -> Vec<NormalizedProviderEvent> {
|
||||
project_acpx_state_event(&context(), &event).unwrap()
|
||||
}
|
||||
|
||||
fn reduced_semantic_result(
|
||||
call_id: &str,
|
||||
operation_id: &str,
|
||||
ok: bool,
|
||||
result: Value,
|
||||
) -> AcpxProviderStateEvent {
|
||||
let mut state = AcpxProviderState::new("run-1").unwrap();
|
||||
state.begin_turn("turn-1").unwrap();
|
||||
state
|
||||
.accept_event(&AcpxSidecarEvent {
|
||||
sequence: 1,
|
||||
event_type: GeneratedAcpxSidecarEventType::RuntimeEvent,
|
||||
run_id: Some("run-1".to_owned()),
|
||||
turn_id: Some("turn-1".to_owned()),
|
||||
payload: json!({
|
||||
"type":"semantic_result",
|
||||
"callId":call_id,
|
||||
"operationId":operation_id,
|
||||
"ok":ok,
|
||||
"result":result,
|
||||
}),
|
||||
})
|
||||
.unwrap()
|
||||
.remove(0)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_authorized_tools_with_exact_durable_correlation() {
|
||||
let events = project(AcpxProviderStateEvent::ToolCall {
|
||||
call_id: "call-1".to_owned(),
|
||||
operation_id: "get_task_context".to_owned(),
|
||||
input: json!({"taskId":"task-1"}),
|
||||
});
|
||||
|
||||
assert_eq!(events.len(), 1);
|
||||
assert_eq!(events[0].event_type, "semantic_tool.input");
|
||||
assert_eq!(events[0].priority, EventPriority::P0);
|
||||
assert_eq!(
|
||||
events[0].payload["semantic_tool"]["correlation"],
|
||||
json!({
|
||||
"runId":"run-1",
|
||||
"normalizedSessionId":"session-1",
|
||||
"turnId":"turn-1",
|
||||
"itemId":"item-1",
|
||||
})
|
||||
);
|
||||
assert_eq!(
|
||||
events[0].payload["semantic_tool"]["input"],
|
||||
json!({"taskId":"task-1"})
|
||||
);
|
||||
assert!(events[0].payload["semantic_tool"]["content"]["digest"]
|
||||
.as_str()
|
||||
.is_some_and(|value| value.starts_with("sha256:")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_terminal_tool_cancellations_as_correlated_results() {
|
||||
let events = project(AcpxProviderStateEvent::ToolResult(ToolResult {
|
||||
call_id: "call-1".to_owned(),
|
||||
operation_id: "issues.read".to_owned(),
|
||||
result: json!({
|
||||
"error": {
|
||||
"code": "acpx_turn_settled",
|
||||
"message": "The provider turn stopped before this semantic tool completed",
|
||||
"retryable": false,
|
||||
},
|
||||
}),
|
||||
is_error: true,
|
||||
}));
|
||||
|
||||
assert_eq!(events.len(), 1);
|
||||
assert_eq!(events[0].event_type, "semantic_tool.result");
|
||||
assert_eq!(events[0].priority, EventPriority::P0);
|
||||
assert_eq!(events[0].payload["semantic_tool"]["phase"], "result");
|
||||
assert_eq!(events[0].payload["semantic_tool"]["callId"], "call-1");
|
||||
assert_eq!(events[0].payload["semantic_tool"]["outcome"], "failed");
|
||||
assert_eq!(
|
||||
events[0].payload["semantic_tool"]["correlation"],
|
||||
json!({
|
||||
"runId":"run-1",
|
||||
"normalizedSessionId":"session-1",
|
||||
"turnId":"turn-1",
|
||||
"itemId":"item-1",
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_structured_input_and_semantic_results_without_provider_envelopes() {
|
||||
let question_set = json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"title":"Choose a target",
|
||||
"questions":[],
|
||||
});
|
||||
let input = project(AcpxProviderStateEvent::InputRequest {
|
||||
request_id: "request-1".to_owned(),
|
||||
question_set: question_set.clone(),
|
||||
origin: None,
|
||||
});
|
||||
assert_eq!(input[0].event_type, "runtime_request.created");
|
||||
assert_eq!(input[0].payload["request"]["requestId"], "request-1");
|
||||
assert_eq!(input[0].payload["request"]["turnId"], "turn-1");
|
||||
assert_eq!(input[0].payload["request"]["itemId"], "item-1");
|
||||
assert_eq!(input[0].payload["request"]["input"], question_set);
|
||||
assert_eq!(
|
||||
input[0].payload["request"]["origin"]["adapter"],
|
||||
"codex-acpx"
|
||||
);
|
||||
|
||||
let result = json!({
|
||||
"schema":"paperclip.run_result.v1",
|
||||
"reportedWorkDisposition":"done",
|
||||
"summary":"Finished",
|
||||
});
|
||||
let projected = project(reduced_semantic_result(
|
||||
"finish-1",
|
||||
"paperclip_finish",
|
||||
true,
|
||||
result.clone(),
|
||||
));
|
||||
assert_eq!(projected[0].event_type, "run.result.proposed");
|
||||
assert_eq!(projected[0].payload, result);
|
||||
|
||||
let dynamic = project(reduced_semantic_result(
|
||||
"call-1",
|
||||
"issues.read",
|
||||
true,
|
||||
json!({"id":"issue-1"}),
|
||||
));
|
||||
assert_eq!(dynamic[0].event_type, "semantic_tool.result");
|
||||
assert_eq!(dynamic[0].payload["semantic_tool"]["phase"], "result");
|
||||
assert_eq!(
|
||||
dynamic[0].payload["semantic_tool"]["operationId"],
|
||||
"issues.read"
|
||||
);
|
||||
assert_eq!(dynamic[0].payload["semantic_tool"]["callId"], "call-1");
|
||||
|
||||
let activity = NormalizedProviderEvent {
|
||||
event_type: "usage.reported".to_owned(),
|
||||
priority: EventPriority::P0,
|
||||
payload: json!({"cumulative":{"inputTokens":1}}),
|
||||
};
|
||||
assert_eq!(
|
||||
project(AcpxProviderStateEvent::Activity(activity.clone())),
|
||||
vec![activity]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_runtime_request_prompt_and_origin_into_the_strict_schema() {
|
||||
let empty_title = AcpxProviderStateEvent::InputRequest {
|
||||
request_id: "request-1".to_owned(),
|
||||
question_set: json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"title":"",
|
||||
"questions":[{
|
||||
"id":"target",
|
||||
"prompt":"Which target?",
|
||||
"required":true,
|
||||
"answerMode":"single_select",
|
||||
"options":[{"id":"first","label":"First"}],
|
||||
}],
|
||||
}),
|
||||
origin: Some(json!({
|
||||
"adapter":"codex-acpx",
|
||||
"provider":"codex",
|
||||
"method":"runtime.input_requested",
|
||||
})),
|
||||
};
|
||||
let projected = project(empty_title);
|
||||
assert_eq!(projected[0].payload["request"]["prompt"], "Which target?");
|
||||
|
||||
for origin in [
|
||||
json!({"provider":"codex"}),
|
||||
json!({"adapter":"codex-acpx","extra":true}),
|
||||
json!({"adapter":""}),
|
||||
json!({"adapter":"codex-acpx","method":null}),
|
||||
] {
|
||||
let invalid = AcpxProviderStateEvent::InputRequest {
|
||||
request_id: "request-1".to_owned(),
|
||||
question_set: json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"questions":[],
|
||||
}),
|
||||
origin: Some(origin),
|
||||
};
|
||||
assert!(project_acpx_state_event(&context(), &invalid).is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projects_assistant_terminal_and_diagnostic_events_fail_closed() {
|
||||
let assistant = project(AcpxProviderStateEvent::AssistantMessage {
|
||||
turn_id: "turn-1".to_owned(),
|
||||
text: "Done".to_owned(),
|
||||
});
|
||||
assert_eq!(assistant[0].event_type, "item.completed");
|
||||
assert_eq!(assistant[0].payload["itemId"], "item-1");
|
||||
assert_eq!(assistant[0].payload["channel"], "final");
|
||||
|
||||
for (status, expected) in [
|
||||
(AcpxTurnStatus::Completed, "turn.completed"),
|
||||
(AcpxTurnStatus::Failed, "turn.failed"),
|
||||
(AcpxTurnStatus::Cancelled, "turn.cancelled"),
|
||||
(AcpxTurnStatus::Interrupted, "turn.interrupted"),
|
||||
] {
|
||||
let terminal = project(AcpxProviderStateEvent::TurnTerminal {
|
||||
turn_id: "turn-1".to_owned(),
|
||||
status,
|
||||
error: None,
|
||||
});
|
||||
assert_eq!(terminal[0].event_type, expected);
|
||||
}
|
||||
|
||||
let process = project(AcpxProviderStateEvent::Process(json!({"pid":7})));
|
||||
assert_eq!(process[0].event_type, "harness.diagnostic");
|
||||
assert_eq!(process[0].payload["details"]["pid"], 7);
|
||||
let diagnostic = project(AcpxProviderStateEvent::Diagnostic {
|
||||
code: "provider_notice".to_owned(),
|
||||
message: "Retrying".to_owned(),
|
||||
});
|
||||
assert_eq!(diagnostic[0].payload["message"], "Retrying");
|
||||
|
||||
let wrong_turn = AcpxProviderStateEvent::TurnTerminal {
|
||||
turn_id: "turn-other".to_owned(),
|
||||
status: AcpxTurnStatus::Completed,
|
||||
error: None,
|
||||
};
|
||||
assert!(project_acpx_state_event(&context(), &wrong_turn)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("durable turn projection"));
|
||||
let permission = AcpxProviderStateEvent::PermissionRequest {
|
||||
request_id: "permission-1".to_owned(),
|
||||
kind: "write".to_owned(),
|
||||
title: "Allow write".to_owned(),
|
||||
details: Value::Null,
|
||||
};
|
||||
assert!(project_acpx_state_event(&context(), &permission)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("pinned runner policy"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_invalid_durable_projection_identity() {
|
||||
let mut invalid = context();
|
||||
invalid.run_id = "".to_owned();
|
||||
let event = AcpxProviderStateEvent::Diagnostic {
|
||||
code: "notice".to_owned(),
|
||||
message: "message".to_owned(),
|
||||
};
|
||||
assert!(project_acpx_state_event(&invalid, &event)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("run identity"));
|
||||
|
||||
for (field, max_chars) in [
|
||||
("run", 160),
|
||||
("normalized session", 160),
|
||||
("turn", 240),
|
||||
("item", 240),
|
||||
] {
|
||||
let mut invalid = context();
|
||||
let oversized = "x".repeat(max_chars + 1);
|
||||
match field {
|
||||
"run" => invalid.run_id = oversized,
|
||||
"normalized session" => invalid.normalized_session_id = oversized,
|
||||
"turn" => invalid.turn_id = oversized,
|
||||
"item" => invalid.item_id = oversized,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
let error = project_acpx_state_event(&invalid, &event)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains(&format!("{field} identity")), "{error}");
|
||||
}
|
||||
|
||||
for request_id in [String::new(), "x".repeat(241), "request\n1".to_owned()] {
|
||||
let request = AcpxProviderStateEvent::InputRequest {
|
||||
request_id,
|
||||
question_set: json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"questions":[],
|
||||
}),
|
||||
origin: None,
|
||||
};
|
||||
let error = project_acpx_state_event(&context(), &request)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("request identity"), "{error}");
|
||||
}
|
||||
|
||||
for field in ["run", "normalized session", "turn", "item"] {
|
||||
let mut invalid = context();
|
||||
match field {
|
||||
"run" => invalid.run_id = "run 1".to_owned(),
|
||||
"normalized session" => invalid.normalized_session_id = "session/1".to_owned(),
|
||||
"turn" => invalid.turn_id = "turn 1".to_owned(),
|
||||
"item" => invalid.item_id = "item/1".to_owned(),
|
||||
_ => unreachable!(),
|
||||
}
|
||||
let error = project_acpx_state_event(&invalid, &event)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains(&format!("{field} identity")), "{error}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_semantic_events_that_cannot_form_stable_receipts() {
|
||||
for event in [
|
||||
AcpxProviderStateEvent::ToolCall {
|
||||
call_id: "call 1".to_owned(),
|
||||
operation_id: "issues.read".to_owned(),
|
||||
input: json!({}),
|
||||
},
|
||||
AcpxProviderStateEvent::ToolResult(ToolResult {
|
||||
call_id: "call-1".to_owned(),
|
||||
operation_id: "issues/read".to_owned(),
|
||||
result: json!({}),
|
||||
is_error: false,
|
||||
}),
|
||||
reduced_semantic_result("réturn-1", "issues.read", true, json!({})),
|
||||
reduced_semantic_result("call-1", "issues read", true, json!({})),
|
||||
] {
|
||||
let error = project_acpx_state_event(&context(), &event)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("semantic"), "{error}");
|
||||
assert!(error.contains("identity"), "{error}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn deterministically_projects_bounded_upstream_request_ids() {
|
||||
let question_set = json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"questions":[],
|
||||
});
|
||||
for upstream_id in [
|
||||
"request 1".to_owned(),
|
||||
"réquest-1".to_owned(),
|
||||
"request/1".to_owned(),
|
||||
"_request-1".to_owned(),
|
||||
"x".repeat(240),
|
||||
] {
|
||||
let event = AcpxProviderStateEvent::InputRequest {
|
||||
request_id: upstream_id.clone(),
|
||||
question_set: question_set.clone(),
|
||||
origin: None,
|
||||
};
|
||||
let first = project_acpx_state_event(&context(), &event).unwrap();
|
||||
let second = project_acpx_state_event(&context(), &event).unwrap();
|
||||
let projected_id = first[0].payload["request"]["requestId"].as_str().unwrap();
|
||||
assert_ne!(projected_id, upstream_id);
|
||||
assert_eq!(second[0].payload["request"]["requestId"], projected_id);
|
||||
assert!(projected_id.starts_with("acpx-request-"));
|
||||
assert!(projected_id.len() <= 160);
|
||||
assert!(projected_id
|
||||
.chars()
|
||||
.all(|character| character.is_ascii_alphanumeric() || "._:-".contains(character)));
|
||||
}
|
||||
|
||||
let canonical = AcpxProviderStateEvent::InputRequest {
|
||||
request_id: "request-1".to_owned(),
|
||||
question_set,
|
||||
origin: None,
|
||||
};
|
||||
assert_eq!(
|
||||
project_acpx_state_event(&context(), &canonical).unwrap()[0].payload["request"]
|
||||
["requestId"],
|
||||
"request-1"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn runtime_request_projection_preserves_durable_identity_boundaries() {
|
||||
let canonical_request_schema: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/schemas/request.schema.json"
|
||||
))
|
||||
.unwrap();
|
||||
let mut runtime_request_schema = canonical_request_schema["oneOf"][0].clone();
|
||||
// This test owns identity projection. The question-set validator has its
|
||||
// own coverage, so replace its remote reference with an unconstrained
|
||||
// local schema before compiling the canonical runtime-request branch.
|
||||
runtime_request_schema["properties"]["input"] = json!({});
|
||||
let request_validator = jsonschema::validator_for(&runtime_request_schema).unwrap();
|
||||
let request_event = AcpxProviderStateEvent::InputRequest {
|
||||
request_id: "request-1".to_owned(),
|
||||
question_set: json!({
|
||||
"schema":"paperclip.question_set.v1",
|
||||
"questions":[],
|
||||
}),
|
||||
origin: None,
|
||||
};
|
||||
let valid = project_acpx_state_event(&context(), &request_event).unwrap();
|
||||
assert!(request_validator.is_valid(&valid[0].payload["request"]));
|
||||
|
||||
let mut durable_context = context();
|
||||
durable_context.turn_id = "t".repeat(240);
|
||||
durable_context.item_id = "i".repeat(240);
|
||||
let durable_request = project_acpx_state_event(&durable_context, &request_event).unwrap();
|
||||
assert_eq!(
|
||||
durable_request[0].payload["request"]["turnId"],
|
||||
durable_context.turn_id
|
||||
);
|
||||
assert_eq!(
|
||||
durable_request[0].payload["request"]["itemId"],
|
||||
durable_context.item_id
|
||||
);
|
||||
assert!(request_validator.is_valid(&durable_request[0].payload["request"]));
|
||||
|
||||
for request_id in ["request 1", "réquest-1", "request/1", "_request-1"] {
|
||||
let mut legacy_request = valid[0].payload["request"].clone();
|
||||
legacy_request["requestId"] = Value::String(request_id.to_owned());
|
||||
assert!(request_validator.is_valid(&legacy_request), "{request_id}");
|
||||
}
|
||||
|
||||
for field in ["turnId", "itemId"] {
|
||||
let mut invalid_request = valid[0].payload["request"].clone();
|
||||
invalid_request[field] = Value::String("x".repeat(241));
|
||||
assert!(!request_validator.is_valid(&invalid_request));
|
||||
}
|
||||
|
||||
let diagnostic = AcpxProviderStateEvent::Diagnostic {
|
||||
code: "notice".to_owned(),
|
||||
message: "message".to_owned(),
|
||||
};
|
||||
assert!(project_acpx_state_event(&durable_context, &diagnostic).is_ok());
|
||||
|
||||
let semantic = project_acpx_state_event(
|
||||
&durable_context,
|
||||
&AcpxProviderStateEvent::ToolCall {
|
||||
call_id: "call-1".to_owned(),
|
||||
operation_id: "issues.read".to_owned(),
|
||||
input: json!({"taskId":"task-1"}),
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
let semantic_schema: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/schemas/semantic-tool.schema.json"
|
||||
))
|
||||
.unwrap();
|
||||
let semantic_validator = jsonschema::validator_for(&semantic_schema).unwrap();
|
||||
assert!(semantic_validator.is_valid(&semantic[0].payload["semantic_tool"]));
|
||||
}
|
||||
|
|
@ -174,8 +174,20 @@ fn bounds_settled_turn_identity_retention() {
|
|||
fn rejects_invalid_scope_identifiers() {
|
||||
assert!(AcpxEventScope::new("").is_err());
|
||||
assert!(AcpxEventScope::new("run\n1").is_err());
|
||||
assert!(AcpxEventScope::new("run 1").is_err());
|
||||
let mut scope = AcpxEventScope::new("run-1").unwrap();
|
||||
assert!(scope.bind_turn("t".repeat(161)).is_err());
|
||||
let longest_turn_id = "t".repeat(240);
|
||||
scope.bind_turn(&longest_turn_id).unwrap();
|
||||
scope
|
||||
.validate_event(&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeDiagnostic,
|
||||
Some("run-1"),
|
||||
Some(&longest_turn_id),
|
||||
))
|
||||
.unwrap();
|
||||
scope.clear_turn(&longest_turn_id).unwrap();
|
||||
assert!(scope.bind_turn("t".repeat(241)).is_err());
|
||||
assert!(scope.bind_turn("turn 1").is_err());
|
||||
let oversized_run_id = "r".repeat(161);
|
||||
assert!(scope
|
||||
.validate_event(&event(
|
||||
|
|
|
|||
|
|
@ -6,10 +6,14 @@ use std::time::Duration;
|
|||
use paperclip_runner_core::acpx_provider_session::{
|
||||
AcpxPermissionMode, AcpxProviderSession, AcpxProviderSessionConfig,
|
||||
};
|
||||
use paperclip_runner_core::acpx_provider_state::AcpxProviderStateEvent;
|
||||
use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarTransportConfig;
|
||||
use paperclip_runner_core::provider_bridge::{
|
||||
authorized_tool_catalog_digest, AuthorizedTool, AuthorizedToolSet, ToolResult,
|
||||
};
|
||||
use paperclip_runner_core::provider_events::{
|
||||
project_acpx_state_event, AcpxEventProjectionContext,
|
||||
};
|
||||
use serde_json::json;
|
||||
|
||||
fn tool_set() -> AuthorizedToolSet {
|
||||
|
|
@ -98,6 +102,47 @@ fn commits_each_resolution_only_after_sidecar_acknowledgement() {
|
|||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn projected_request_id_resolves_the_exact_upstream_sidecar_request() {
|
||||
let mut session = AcpxProviderSession::start(&config("resolutions-projected-id")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let mut input = None;
|
||||
for _ in 0..2 {
|
||||
for event in session.poll_event(Duration::from_secs(1)).unwrap().unwrap() {
|
||||
if matches!(&event, AcpxProviderStateEvent::InputRequest { .. }) {
|
||||
input = Some(event);
|
||||
}
|
||||
}
|
||||
}
|
||||
let input = input.expect("observe the upstream input request");
|
||||
let projected = project_acpx_state_event(
|
||||
&AcpxEventProjectionContext {
|
||||
run_id: "run-1".to_owned(),
|
||||
normalized_session_id: "session-1".to_owned(),
|
||||
turn_id: "turn-1".to_owned(),
|
||||
item_id: "item-1".to_owned(),
|
||||
},
|
||||
&input,
|
||||
)
|
||||
.unwrap();
|
||||
let request_id = projected[0].payload["request"]["requestId"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_owned();
|
||||
assert!(request_id.starts_with("acpx-request-"));
|
||||
|
||||
session
|
||||
.deliver_tool_result(&tool_result("issues.read"))
|
||||
.unwrap();
|
||||
session
|
||||
.resolve_input(&request_id, "turn-1", &input_resolution("first"))
|
||||
.unwrap();
|
||||
assert!(session.state().pending_question_set(&request_id).is_none());
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_validation_preserves_pending_requests_for_a_correct_retry() {
|
||||
let mut session = started("resolutions");
|
||||
|
|
|
|||
|
|
@ -2,6 +2,9 @@ use paperclip_runner_core::acpx_event_payload::AcpxTurnStatus;
|
|||
use paperclip_runner_core::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent};
|
||||
use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use paperclip_runner_core::provider_events::{
|
||||
project_acpx_state_event, AcpxEventProjectionContext,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
fn event(
|
||||
|
|
@ -155,6 +158,38 @@ fn tracks_structured_input_and_permission_requests_without_cross_kind_reuse() {
|
|||
state.complete_permission("permission-1").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn correlates_projected_runtime_requests_to_the_upstream_input_id() {
|
||||
let mut state = AcpxProviderState::new("run-1").unwrap();
|
||||
state.begin_turn("turn-1").unwrap();
|
||||
let upstream_request_id = format!("input / {}", "é".repeat(200));
|
||||
let emitted = state
|
||||
.accept_event(&event(
|
||||
1,
|
||||
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
|
||||
Some("turn-1"),
|
||||
json!({"requestId":upstream_request_id,"questionSet":question_set()}),
|
||||
))
|
||||
.unwrap();
|
||||
let projected = project_acpx_state_event(
|
||||
&AcpxEventProjectionContext {
|
||||
run_id: "run-1".to_owned(),
|
||||
normalized_session_id: "session-1".to_owned(),
|
||||
turn_id: "turn-1".to_owned(),
|
||||
item_id: "item-1".to_owned(),
|
||||
},
|
||||
&emitted[0],
|
||||
)
|
||||
.unwrap();
|
||||
let runtime_request_id = projected[0].payload["request"]["requestId"]
|
||||
.as_str()
|
||||
.unwrap();
|
||||
assert!(runtime_request_id.starts_with("acpx-request-"));
|
||||
assert!(state.pending_question_set(runtime_request_id).is_some());
|
||||
state.complete_input(runtime_request_id).unwrap();
|
||||
assert!(state.pending_question_set(runtime_request_id).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() {
|
||||
let mut state = AcpxProviderState::new("run-1").unwrap();
|
||||
|
|
|
|||
|
|
@ -103,6 +103,43 @@ fn rejects_invalid_turn_inputs_without_mutating_the_session() {
|
|||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preserves_the_durable_turn_identity_boundary_through_the_sidecar() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns")).unwrap();
|
||||
let turn_id = "t".repeat(240);
|
||||
let response = session
|
||||
.start_turn(&turn_id, "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
assert_eq!(response["turnId"], turn_id);
|
||||
|
||||
let activity = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&activity[0],
|
||||
AcpxProviderStateEvent::Activity(event)
|
||||
if event.event_type == "item.delta" && event.payload["text"] == "hello"
|
||||
));
|
||||
session
|
||||
.interrupt_turn(&turn_id, "Paperclip interruption")
|
||||
.unwrap();
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } if settled == &turn_id
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
|
||||
let mut session = AcpxProviderSession::start(&config("turns")).unwrap();
|
||||
for invalid_turn_id in ["t".repeat(241), "turn 1".to_owned()] {
|
||||
assert!(session
|
||||
.start_turn(&invalid_turn_id, "Please help", &std::env::temp_dir(),)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("turn id is invalid"));
|
||||
}
|
||||
assert_eq!(session.state().active_turn_id(), None);
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_when_turn_start_acknowledges_another_turn() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-wrong-turn")).unwrap();
|
||||
|
|
|
|||
|
|
@ -2764,12 +2764,12 @@ export const requestSchema = {
|
|||
"turnId": {
|
||||
"type": "string",
|
||||
"minLength": 1,
|
||||
"maxLength": 160
|
||||
"maxLength": 240
|
||||
},
|
||||
"itemId": {
|
||||
"type": "string",
|
||||
"minLength": 1,
|
||||
"maxLength": 160
|
||||
"maxLength": 240
|
||||
}
|
||||
},
|
||||
"additionalProperties": false
|
||||
|
|
@ -3258,12 +3258,12 @@ export const eventSchema = {
|
|||
"turnId": {
|
||||
"type": "string",
|
||||
"minLength": 1,
|
||||
"maxLength": 160
|
||||
"maxLength": 240
|
||||
},
|
||||
"itemId": {
|
||||
"type": "string",
|
||||
"minLength": 1,
|
||||
"maxLength": 160
|
||||
"maxLength": 240
|
||||
},
|
||||
"eventType": {
|
||||
"enum": [
|
||||
|
|
|
|||
|
|
@ -47,6 +47,19 @@ test("the ACPX sidecar schema accepts each versioned message family", () => {
|
|||
}
|
||||
});
|
||||
|
||||
test("the ACPX sidecar schema shares the durable stable-identity boundary", () => {
|
||||
const longestTurnId = "t".repeat(240);
|
||||
assert.equal(validate({ ...messages[2], turnId: longestTurnId }), true);
|
||||
assert.equal(validate({ ...messages[2], turnId: "t".repeat(241) }), false);
|
||||
for (const turnId of ["turn 1", "réturn-1", "turn/1", "_turn-1"]) {
|
||||
assert.equal(validate({ ...messages[2], turnId }), false, turnId);
|
||||
}
|
||||
assert.equal(validate({ ...messages[2], runId: "r".repeat(161) }), false);
|
||||
for (const runId of ["run 1", "rún-1", "run/1", "_run-1"]) {
|
||||
assert.equal(validate({ ...messages[2], runId }), false, runId);
|
||||
}
|
||||
});
|
||||
|
||||
test("the ACPX sidecar schema fails closed on drift", () => {
|
||||
for (const message of [
|
||||
{ ...messages[0], protocolVersion: protocolVersion + 1 },
|
||||
|
|
|
|||
Loading…
Reference in New Issue