RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(runner): reduce ACPX provider state (#12417) 

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The Rust runner now has bounded ACPX transport, scope, payload, and
provider-neutral normalization layers
> - A live provider still needs state across events to correlate
requests and preserve turn ordering
> - That state must not mutate before scope and payload validation
succeeds
> - It must bound retained text and pending values, suppress repeated
reasoning boundaries, and make one semantic result authoritative
> - This pull request adds that reducer without issuing process commands
or selecting ACPX in runnerd
> - The benefit is a separately reviewable state machine before
transport commands and production selection are connected

## Linked Issues or Issue Description

Refs #12416

Refs #12415

## What Changed

- Add a package-local ACPX provider state reducer with one run binding
and one active turn.
- Decode every sidecar event through the existing scope-first payload
boundary before state mutation.
- Bound retained assistant text, pending semantic tool inputs, and
pending runtime request values.
- Correlate semantic tool calls, structured input requests, and
permission requests by stable IDs.
- Keep pending tool resolution two-phase so callers remove state only
after a later sidecar command succeeds.
- Carry authoritative tool classification from validated payloads into
retained state.
- Suppress repeated reasoning-start activity within one turn.
- Accept one semantic result idempotently and fail closed on a
conflicting result.
- Flush the final assistant message before the authoritative terminal
event.
- Clear unresolved turn-scoped requests at terminal state and reject
late events for the settled turn.
- Admit redacted global process and diagnostic events without requiring
an active turn.
- Add seven integration tests for turn ordering, correlation, conflicts,
scope-before-mutation, redaction, and terminal cleanup.
- Document the state boundary.
- Do not change dependencies, lockfiles, workflows, runnerd selection,
server behavior, UI, or migrations.

## Verification

- Replay base: `7bb6cebeae727a16c205bb80b5c2b9e92ea6b5fa` (`master`
after #12416 merged).
- Exact replay head: `da82e7f67ecd6f0f2184f303b1b703721099cd86`.
- Stable patch ID: `3027df409450d08b2c32383585597a39e06c6f53`, identical
to the reviewed `f9cb4e54..e6e550f9` delta.
- The exact delta is 4 files and 720 additions, all in
`packages/paperclip-runner`; it contains no lockfile, workflow, server,
UI, or migration change.
- Focused Rust state, package, repository, security, and Greptile
checks: **PASSED** on the replayed exact head. Full CI run `33362166929`
completed 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 reducer owns security-sensitive correlation and terminal
ordering, so its only raw-event entry point always invokes scope-first
decoding.
- Pending resolution methods must be called only after the corresponding
sidecar transport command succeeds; the later process adapter owns that
sequencing.
- Terminal events intentionally clear unresolved turn-scoped requests so
late tool or input results fail closed.
- One semantic result remains readable after terminal state for later
durable finalization and is cleared only when a new turn begins.
- The package exports a new Rust module, but no production path
constructs it in this pull request.

> 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 2026-08-31 01:03:40 -05:00 committed by GitHub
parent 7bb6cebeae
commit f038633bf5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 720 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/paperclip-runner/README.md
View File

@ -83,6 +83,12 @@ are canonicalized, and consumers must not reinterpret the display value as
file-access authority. Operational semantic-result and terminal events remain
reserved for the stateful adapter rather than being duplicated.
The package-local ACPX provider reducer preserves that order while it tracks one
active turn, bounded assistant text, semantic results, and pending tool or input
correlations. Terminal events flush the final assistant message first and clear
unresolved turn-scoped requests. This reducer still does not select ACPX in
runnerd.
Run the complete contract gate with:
```sh

429
packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs Normal file
View File

@ -0,0 +1,429 @@
use std::collections::BTreeMap;
use serde_json::Value;
use crate::acpx_event_payload::{
decode_acpx_event, AcpxEventPayload, AcpxRuntimeEventKind, AcpxTurnStatus,
};
use crate::acpx_event_scope::AcpxEventScope;
use crate::acpx_sidecar_transport::AcpxSidecarEvent;
use crate::local_runner::LocalRunnerError;
use crate::provider_events::{normalize_acpx_runtime_event, NormalizedProviderEvent};
const MAX_ASSISTANT_TEXT_BYTES: usize = 1024 * 1024;
const MAX_PENDING_TOOLS: usize = 4_096;
const MAX_PENDING_TOOL_INPUT_BYTES: usize = 16 * 1024 * 1024;
const MAX_PENDING_RUNTIME_REQUESTS: usize = 1_024;
const MAX_PENDING_RUNTIME_REQUEST_BYTES: usize = 16 * 1024 * 1024;
#[derive(Clone, Debug, PartialEq)]
pub struct AcpxPendingTool {
pub operation_id: String,
pub input: Value,
input_bytes: usize,
}
#[derive(Clone, Debug, PartialEq)]
pub struct AcpxSemanticResult {
pub call_id: String,
pub operation_id: String,
pub result: Value,
}
#[derive(Clone, Debug, PartialEq)]
pub enum AcpxProviderStateEvent {
Activity(NormalizedProviderEvent),
ToolCall {
call_id: String,
operation_id: String,
input: Value,
},
PermissionRequest {
request_id: String,
kind: String,
title: String,
details: Value,
},
InputRequest {
request_id: String,
question_set: Value,
origin: Option<Value>,
},
SemanticResult(AcpxSemanticResult),
AssistantMessage {
turn_id: String,
text: String,
},
TurnTerminal {
turn_id: String,
status: AcpxTurnStatus,
error: Option<Value>,
},
Process(Value),
Diagnostic {
code: String,
message: String,
},
}
#[derive(Clone, Debug, PartialEq)]
struct PendingInput {
value_bytes: usize,
}
/// Reduces validated sidecar events into bounded provider state.
///
/// Raw sidecar values enter only through `accept_event`, which applies run and
/// turn authority before payload decoding. Transport commands remain outside
/// this reducer so callers can commit a pending resolution only after the
/// corresponding sidecar request succeeds.
#[derive(Clone, Debug, PartialEq)]
pub struct AcpxProviderState {
scope: AcpxEventScope,
provider_requests: u64,
thinking_active: bool,
assistant_text: String,
pending_tools: BTreeMap<String, AcpxPendingTool>,
pending_tool_input_bytes: usize,
pending_permissions: BTreeMap<String, usize>,
pending_inputs: BTreeMap<String, PendingInput>,
pending_runtime_request_bytes: usize,
semantic_result: Option<AcpxSemanticResult>,
}
impl AcpxProviderState {
pub fn new(run_id: impl Into<String>) -> Result<Self, LocalRunnerError> {
Ok(Self {
scope: AcpxEventScope::new(run_id)?,
provider_requests: 0,
thinking_active: false,
assistant_text: String::new(),
pending_tools: BTreeMap::new(),
pending_tool_input_bytes: 0,
pending_permissions: BTreeMap::new(),
pending_inputs: BTreeMap::new(),
pending_runtime_request_bytes: 0,
semantic_result: None,
})
}
pub fn run_id(&self) -> &str {
self.scope.run_id()
}
pub fn active_turn_id(&self) -> Option<&str> {
self.scope.active_turn_id()
}
pub fn begin_turn(&mut self, turn_id: impl Into<String>) -> Result<(), LocalRunnerError> {
if self.scope.active_turn_id().is_some()
|| !self.pending_tools.is_empty()
|| !self.pending_permissions.is_empty()
|| !self.pending_inputs.is_empty()
{
return Err(LocalRunnerError::invalid(
"ACPX provider state cannot start a turn while work is active",
));
}
let next_provider_requests = self
.provider_requests
.checked_add(1)
.ok_or_else(|| LocalRunnerError::invalid("ACPX provider request count is exhausted"))?;
self.scope.bind_turn(turn_id)?;
self.provider_requests = next_provider_requests;
self.thinking_active = false;
self.assistant_text.clear();
self.semantic_result = None;
Ok(())
}
pub fn accept_event(
&mut self,
event: &AcpxSidecarEvent,
) -> Result<Vec<AcpxProviderStateEvent>, LocalRunnerError> {
let payload = decode_acpx_event(&self.scope, event)?;
match payload {
AcpxEventPayload::Runtime {
kind,
tool_operation,
payload,
} => self.accept_runtime_event(event, kind, tool_operation, payload),
AcpxEventPayload::PermissionRequested {
request_id,
kind,
title,
details,
} => {
let value_bytes = value_bytes(&details)?;
self.admit_runtime_request(&request_id, value_bytes)?;
self.pending_permissions
.insert(request_id.clone(), value_bytes);
self.pending_runtime_request_bytes += value_bytes;
Ok(vec![AcpxProviderStateEvent::PermissionRequest {
request_id,
kind,
title,
details,
}])
}
AcpxEventPayload::InputRequested {
request_id,
question_set,
origin,
} => {
let value_bytes = value_bytes(&question_set)?;
self.admit_runtime_request(&request_id, value_bytes)?;
if self
.pending_inputs
.insert(request_id.clone(), PendingInput { value_bytes })
.is_some()
{
return Err(LocalRunnerError::invalid(
"ACPX reused a pending input request id",
));
}
self.pending_runtime_request_bytes += value_bytes;
Ok(vec![AcpxProviderStateEvent::InputRequest {
request_id,
question_set,
origin,
}])
}
AcpxEventPayload::ToolCalled {
call_id,
operation_id,
input,
} => {
let input_bytes = value_bytes(&input)?;
if self.pending_tools.len() >= MAX_PENDING_TOOLS
|| self
.pending_tool_input_bytes
.checked_add(input_bytes)
.is_none_or(|bytes| bytes > MAX_PENDING_TOOL_INPUT_BYTES)
{
return Err(LocalRunnerError::invalid(
"ACPX pending tool calls exceed their bounded capacity",
));
}
if self.pending_tools.contains_key(&call_id) {
return Err(LocalRunnerError::invalid(
"ACPX reused a pending tool call id",
));
}
self.pending_tools.insert(
call_id.clone(),
AcpxPendingTool {
operation_id: operation_id.clone(),
input: input.clone(),
input_bytes,
},
);
self.pending_tool_input_bytes += input_bytes;
Ok(vec![AcpxProviderStateEvent::ToolCall {
call_id,
operation_id,
input,
}])
}
AcpxEventPayload::TurnTerminal { status, error } => {
let turn_id = event
.turn_id
.as_deref()
.expect("a decoded terminal event has a turn binding")
.to_owned();
self.scope.clear_turn(&turn_id)?;
self.clear_pending_requests();
self.thinking_active = false;
let mut events = Vec::new();
if !self.assistant_text.is_empty() {
events.push(AcpxProviderStateEvent::AssistantMessage {
turn_id: turn_id.clone(),
text: std::mem::take(&mut self.assistant_text),
});
}
events.push(AcpxProviderStateEvent::TurnTerminal {
turn_id,
status,
error,
});
Ok(events)
}
AcpxEventPayload::Process { details } => {
Ok(vec![AcpxProviderStateEvent::Process(details)])
}
AcpxEventPayload::Diagnostic { code, message } => {
Ok(vec![AcpxProviderStateEvent::Diagnostic { code, message }])
}
}
}
pub fn pending_tool(&self, call_id: &str) -> Option<&AcpxPendingTool> {
self.pending_tools.get(call_id)
}
pub fn complete_tool(
&mut self,
call_id: &str,
operation_id: &str,
) -> Result<(), LocalRunnerError> {
let pending = self.pending_tools.get(call_id).ok_or_else(|| {
LocalRunnerError::invalid("ACPX tool result has no pending sidecar call")
})?;
if pending.operation_id != operation_id {
return Err(LocalRunnerError::invalid(
"ACPX tool result operation mismatch",
));
}
let pending = self
.pending_tools
.remove(call_id)
.expect("validated ACPX pending tool remains present");
self.pending_tool_input_bytes = self
.pending_tool_input_bytes
.saturating_sub(pending.input_bytes);
Ok(())
}
pub fn complete_permission(&mut self, request_id: &str) -> Result<(), LocalRunnerError> {
let value_bytes = self.pending_permissions.remove(request_id).ok_or_else(|| {
LocalRunnerError::invalid("ACPX permission result has no pending request")
})?;
self.pending_runtime_request_bytes = self
.pending_runtime_request_bytes
.saturating_sub(value_bytes);
Ok(())
}
pub fn complete_input(&mut self, request_id: &str) -> Result<(), LocalRunnerError> {
let pending = self
.pending_inputs
.remove(request_id)
.ok_or_else(|| LocalRunnerError::invalid("ACPX input result has no pending request"))?;
self.pending_runtime_request_bytes = self
.pending_runtime_request_bytes
.saturating_sub(pending.value_bytes);
Ok(())
}
pub fn semantic_result(&self) -> Option<&AcpxSemanticResult> {
self.semantic_result.as_ref()
}
fn accept_runtime_event(
&mut self,
event: &AcpxSidecarEvent,
kind: AcpxRuntimeEventKind,
tool_operation: Option<&'static str>,
payload: Value,
) -> Result<Vec<AcpxProviderStateEvent>, LocalRunnerError> {
if kind == AcpxRuntimeEventKind::Thinking {
if self.thinking_active {
return Ok(Vec::new());
}
self.thinking_active = true;
}
if kind == AcpxRuntimeEventKind::TextDelta {
let text = payload
.get("text")
.and_then(Value::as_str)
.unwrap_or_default();
if self
.assistant_text
.len()
.checked_add(text.len())
.is_none_or(|bytes| bytes > MAX_ASSISTANT_TEXT_BYTES)
{
return Err(LocalRunnerError::invalid(
"ACPX assistant text exceeds its retained limit",
));
}
self.assistant_text.push_str(text);
}
if kind == AcpxRuntimeEventKind::SemanticResult {
let result = AcpxSemanticResult {
call_id: payload
.get("callId")
.and_then(Value::as_str)
.expect("decoded ACPX semantic result has a call id")
.to_owned(),
operation_id: payload
.get("operationId")
.and_then(Value::as_str)
.expect("decoded ACPX semantic result has an operation id")
.to_owned(),
result: payload
.get("result")
.expect("decoded ACPX semantic result has a result")
.clone(),
};
return match self.semantic_result.as_ref() {
None => {
self.semantic_result = Some(result.clone());
Ok(vec![AcpxProviderStateEvent::SemanticResult(result)])
}
Some(existing) if existing == &result => Ok(Vec::new()),
Some(_) => Err(LocalRunnerError::invalid(
"ACPX emitted conflicting semantic results for one turn",
)),
};
}
let turn_id = event
.turn_id
.as_deref()
.expect("a decoded runtime event has a turn binding");
let fallback_item_id = format!("acpx-event-{}", event.sequence);
Ok(normalize_acpx_runtime_event(
kind,
&payload,
tool_operation,
&fallback_item_id,
turn_id,
self.provider_requests,
)
.into_iter()
.map(AcpxProviderStateEvent::Activity)
.collect())
}
fn admit_runtime_request(
&self,
request_id: &str,
value_bytes: usize,
) -> Result<(), LocalRunnerError> {
if self.pending_permissions.contains_key(request_id)
|| self.pending_inputs.contains_key(request_id)
{
return Err(LocalRunnerError::invalid(
"ACPX reused a pending runtime request id",
));
}
if self.pending_permissions.len() + self.pending_inputs.len()
>= MAX_PENDING_RUNTIME_REQUESTS
|| self
.pending_runtime_request_bytes
.checked_add(value_bytes)
.is_none_or(|bytes| bytes > MAX_PENDING_RUNTIME_REQUEST_BYTES)
{
return Err(LocalRunnerError::invalid(
"ACPX pending runtime requests exceed their bounded capacity",
));
}
Ok(())
}
fn clear_pending_requests(&mut self) {
self.pending_tools.clear();
self.pending_tool_input_bytes = 0;
self.pending_permissions.clear();
self.pending_inputs.clear();
self.pending_runtime_request_bytes = 0;
}
}
fn value_bytes(value: &Value) -> Result<usize, LocalRunnerError> {
serde_json::to_vec(value)
.map(|value| value.len())
.map_err(|error| {
LocalRunnerError::invalid(format!("ACPX retained value is invalid: {error}"))
})
}

1
packages/paperclip-runner/runner/crates/runner-core/src/lib.rs
View File

@ -2,6 +2,7 @@
pub mod acpx_event_payload;
pub mod acpx_event_scope;
pub mod acpx_provider_state;
pub mod acpx_sidecar_transport;
pub mod codex_provider;
pub mod durable;

284
packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_state.rs Normal file
View File

@ -0,0 +1,284 @@
use paperclip_runner_core::acpx_event_payload::AcpxTurnStatus;
use paperclip_runner_core::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent};
use paperclip_runner_core::acpx_sidecar_transport::AcpxSidecarEvent;
use paperclip_runner_core::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
use serde_json::{json, Value};
fn event(
sequence: u64,
event_type: GeneratedAcpxSidecarEventType,
turn_id: Option<&str>,
payload: Value,
) -> AcpxSidecarEvent {
AcpxSidecarEvent {
sequence,
event_type,
run_id: Some("run-1".to_owned()),
turn_id: turn_id.map(str::to_owned),
payload,
}
}
fn question_set() -> Value {
json!({
"schema":"paperclip.question_set.v1",
"title":"Choose",
"questions":[{
"id":"question-1",
"prompt":"Which option?",
"required":true,
"answerMode":"single_select",
"options":[{"id":"option-1","label":"First"}]
}]
})
}
#[test]
fn accumulates_assistant_text_deduplicates_reasoning_and_flushes_before_terminal() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let thinking = event(
1,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({"type":"thinking","text":"private"}),
);
assert_eq!(state.accept_event(&thinking).unwrap().len(), 1);
let repeated = event(
2,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({"type":"thinking","text":"still private"}),
);
assert!(state.accept_event(&repeated).unwrap().is_empty());
for (sequence, text) in [(3, "Hello "), (4, "world")] {
state
.accept_event(&event(
sequence,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({"type":"text_delta","text":text}),
))
.unwrap();
}
let terminal = state
.accept_event(&event(
5,
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
Some("turn-1"),
json!({"status":"completed"}),
))
.unwrap();
assert!(matches!(
&terminal[0],
AcpxProviderStateEvent::AssistantMessage { turn_id, text }
if turn_id == "turn-1" && text == "Hello world"
));
assert!(matches!(
&terminal[1],
AcpxProviderStateEvent::TurnTerminal {
turn_id,
status: AcpxTurnStatus::Completed,
error: None,
} if turn_id == "turn-1"
));
assert_eq!(state.active_turn_id(), None);
}
#[test]
fn correlates_semantic_tool_calls_until_the_sidecar_resolution_commits() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let called = event(
1,
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
Some("turn-1"),
json!({"callId":"call-1","operationId":"issues.read","input":{"id":"issue-1"}}),
);
let emitted = state.accept_event(&called).unwrap();
assert!(matches!(
&emitted[0],
AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. }
if call_id == "call-1" && operation_id == "issues.read"
));
assert_eq!(
state.pending_tool("call-1").unwrap().operation_id,
"issues.read"
);
assert!(state.complete_tool("call-1", "issues.write").is_err());
assert!(state.pending_tool("call-1").is_some());
state.complete_tool("call-1", "issues.read").unwrap();
assert!(state.pending_tool("call-1").is_none());
assert!(state.accept_event(&called).is_ok());
assert!(state.accept_event(&called).is_err());
}
#[test]
fn tracks_structured_input_and_permission_requests_without_cross_kind_reuse() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let input = event(
1,
GeneratedAcpxSidecarEventType::RuntimeInputRequested,
Some("turn-1"),
json!({"requestId":"request-1","questionSet":question_set()}),
);
assert!(matches!(
&state.accept_event(&input).unwrap()[0],
AcpxProviderStateEvent::InputRequest { request_id, .. }
if request_id == "request-1"
));
let permission_with_reused_id = event(
2,
GeneratedAcpxSidecarEventType::RuntimePermissionRequested,
Some("turn-1"),
json!({"requestId":"request-1","kind":"execute","title":"Run?"}),
);
assert!(state.accept_event(&permission_with_reused_id).is_err());
state.complete_input("request-1").unwrap();
assert!(state.complete_input("request-1").is_err());
let permission = event(
3,
GeneratedAcpxSidecarEventType::RuntimePermissionRequested,
Some("turn-1"),
json!({"requestId":"permission-1","kind":"execute","title":"Run?"}),
);
assert!(matches!(
&state.accept_event(&permission).unwrap()[0],
AcpxProviderStateEvent::PermissionRequest { request_id, .. }
if request_id == "permission-1"
));
state.complete_permission("permission-1").unwrap();
}
#[test]
fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let result = event(
1,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({
"type":"semantic_result",
"callId":"finish-1",
"operationId":"paperclip_finish",
"result":{"reportedWorkDisposition":"done"}
}),
);
assert!(matches!(
&state.accept_event(&result).unwrap()[0],
AcpxProviderStateEvent::SemanticResult(result)
if result.call_id == "finish-1"
));
assert!(state.accept_event(&result).unwrap().is_empty());
assert_eq!(
state.semantic_result().unwrap().result["reportedWorkDisposition"],
"done"
);
let conflict = event(
2,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({
"type":"semantic_result",
"callId":"finish-2",
"operationId":"paperclip_finish",
"result":{"reportedWorkDisposition":"done"}
}),
);
assert!(state.accept_event(&conflict).is_err());
assert_eq!(state.semantic_result().unwrap().call_id, "finish-1");
}
#[test]
fn validates_scope_before_mutating_pending_state() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
let mut wrong_run = event(
1,
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
Some("turn-1"),
json!({"callId":"call-1","operationId":"issues.read","input":{}}),
);
wrong_run.run_id = Some("run-2".to_owned());
assert!(state.accept_event(&wrong_run).is_err());
assert!(state.pending_tool("call-1").is_none());
let wrong_turn = event(
2,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-2"),
json!({"type":"text_delta","text":"wrong"}),
);
assert!(state.accept_event(&wrong_turn).is_err());
}
#[test]
fn accepts_global_process_and_diagnostic_events_without_an_active_turn() {
let mut state = AcpxProviderState::new("run-1").unwrap();
let mut process = event(
1,
GeneratedAcpxSidecarEventType::RuntimeProcess,
None,
json!({"pid":17,"accessToken":"secret"}),
);
process.run_id = None;
let process = state.accept_event(&process).unwrap();
assert!(matches!(
&process[0],
AcpxProviderStateEvent::Process(details)
if details["accessToken"] == "[REDACTED]"
));
let mut diagnostic = event(
2,
GeneratedAcpxSidecarEventType::RuntimeDiagnostic,
None,
json!({"code":"provider_notice","message":"token=super-secret"}),
);
diagnostic.run_id = None;
let diagnostic = state.accept_event(&diagnostic).unwrap();
assert!(matches!(
&diagnostic[0],
AcpxProviderStateEvent::Diagnostic { message, .. }
if message.contains("REDACTED") && !message.contains("super-secret")
));
}
#[test]
fn terminal_events_clear_pending_requests_and_reject_late_turn_events() {
let mut state = AcpxProviderState::new("run-1").unwrap();
state.begin_turn("turn-1").unwrap();
state
.accept_event(&event(
1,
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
Some("turn-1"),
json!({"callId":"call-1","operationId":"issues.read","input":{}}),
))
.unwrap();
state
.accept_event(&event(
2,
GeneratedAcpxSidecarEventType::RuntimeTurnTerminal,
Some("turn-1"),
json!({"status":"cancelled","error":{"message":"token=secret"}}),
))
.unwrap();
assert!(state.pending_tool("call-1").is_none());
assert!(state.complete_tool("call-1", "issues.read").is_err());
assert!(state
.accept_event(&event(
3,
GeneratedAcpxSidecarEventType::RuntimeEvent,
Some("turn-1"),
json!({"type":"text_delta","text":"late"}),
))
.is_err());
}