feat(runner): drive ACPX provider turns (#12419)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The Rust runner now validates and opens an exact Codex ACPX session and attaches one run/catalog identity > - A live session must start one turn, preserve immutable workspace authority, accept interruption, and consume provider events > - Command acknowledgements and event scope are trust boundaries; disagreement must terminate the session rather than permit divergent state > - The existing reducer already owns bounded state, ordering, correlation, and terminal cleanup > - This pull request connects turn commands and polling only to that reviewed reducer without selecting ACPX in runnerd > - The benefit is a fail-closed execution loop before semantic request resolution and production selection are added ## Linked Issues or Issue Description Refs #12418 Refs #12417 ## What Changed - Start one ACPX turn only after validating a bounded turn identifier, bounded message, and the session's immutable working directory. - Require `turn.start` to acknowledge the exact requested turn before mutating provider state. - Request interruption only for the active turn and require an affirmative cancellation acknowledgement. - Keep the turn active until the authoritative terminal event is polled and reduced. - Poll bounded sidecar events and route every event through the existing scope-first payload/state boundary. - Terminate the sidecar fail closed after a rejected or mismatched turn command, transport failure, or invalid scoped event. - Preserve ordinary local validation failures without mutating or closing an otherwise valid session. - Admit semantic tool calls through the run-scoped authorized tool bridge before returning them to runnerd. - Reduce events and authorize tool calls on candidate clones, then commit both together only after every check passes. - Rotate and scope tool receipts by run and turn; reserve pending, completed, and settled call IDs across both tool bridges so cross-bridge duplicates cannot be re-admitted. - Preserve exact turn correlation through payload, scope, provider state, fake sidecar, and integration fixtures. - Extend the package-local fake sidecar with deterministic turn activity, terminal, acknowledgement-mismatch, and scope-violation modes. - Add integration coverage for happy-path activity/interruption/settlement, local validation, turn mismatch, cancellation mismatch, cross-run denial, receipt rotation, and cross-bridge call-ID reuse. - Document the turn lifecycle boundary. - Do not change dependencies, lockfiles, workflows, runnerd selection, server behavior, UI, or migrations. ## Verification - Replay base: `10cf68295d5457ccf1010db20a06e2951e47162a` (`master` after #12418 merged). - Exact replay head: `1ec18368bdacafafa58720b9e2b982e1b0d11163`. - Stable patch ID: `4b43a5ef2a428850a05b0741d06e7cf4f48ec334`, identical to the prepared four-commit delta plus the focused cross-bridge call-ID reservation fix. - The exact delta is 13 files, 1,855 additions, and 37 deletions, all in `packages/paperclip-runner`; it contains no lockfile, workflow, server, UI, or migration change. - GitHub Actions run `33364033038`, attempt 2: **PASSED** on the exact replay head (23/23 jobs passed; the failed-job-only retry cleared one unrelated flaky server heartbeat test). - Greptile: **5/5** on the exact replay head with zero unresolved review threads; Superagent, Socket, and Snyk checks also passed. - No local test result is claimed. GitHub Actions is the authoritative verification environment for this replayed revision. ## Risks - A sidecar may emit events while a command response is in flight. The transport buffers them, then the session binds the acknowledged turn before polling and reducing those events. - A turn-start rejection can leave provider-native state ambiguous, so command rejection or acknowledgement mismatch terminates the session instead of attempting unsafe reuse. - Interruption does not settle local state; only the validated terminal event clears the active turn. - Any transport or event validation failure terminates the child process so a compromised stream cannot continue. - Call identifiers are authorization identities and remain reserved across pending, completed, and settled states in both bridges. - The package exposes new Rust methods, but no production path invokes them in this pull request. - Tool authorization is atomic with reducer state, but tool results, structured input, and permission resolution remain later slices. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex with GPT-5.6, agentic reasoning, tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
10cf68295d
commit
9a9fdf06ee
|
|
@ -93,6 +93,15 @@ The package-local session bootstrap starts the bounded sidecar transport,
|
|||
verifies the Codex-only capability handshake and effective model, opens one
|
||||
identity-bound session, and confirms its run attachment. Any failed bootstrap
|
||||
terminates the process; session shutdown preserves persistent provider state.
|
||||
The session can then start one immutable-workspace turn, request interruption,
|
||||
and reduce polled events through the scope-first state boundary. A mismatched
|
||||
command acknowledgement or invalid event terminates the session fail closed.
|
||||
Polled semantic calls pass through the run-scoped authorized tool bridge before
|
||||
they can be returned to a caller. Before a follow-up turn releases settled tool
|
||||
receipts, runner-core suspends and reaps the idle sidecar/provider generation,
|
||||
then resumes the same verified persistent identity in a fresh generation. This
|
||||
prevents a late session-lifetime MCP callback from inheriting the next turn's
|
||||
event authority.
|
||||
|
||||
Run the complete contract gate with:
|
||||
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ use crate::generated_acpx_sidecar_contract::{
|
|||
classify_generated_acpx_tool_operation, GeneratedAcpxSidecarEventType,
|
||||
};
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::provider_bridge::semantic_value_digest;
|
||||
|
||||
const MAX_EVENT_PAYLOAD_BYTES: usize = 256 * 1024;
|
||||
const MAX_ID_CHARS: usize = 160;
|
||||
|
|
@ -41,6 +42,7 @@ pub enum AcpxEventPayload {
|
|||
kind: AcpxRuntimeEventKind,
|
||||
tool_operation: Option<&'static str>,
|
||||
payload: Value,
|
||||
semantic_result_digest: Option<String>,
|
||||
},
|
||||
PermissionRequested {
|
||||
request_id: String,
|
||||
|
|
@ -57,6 +59,7 @@ pub enum AcpxEventPayload {
|
|||
call_id: String,
|
||||
operation_id: String,
|
||||
input: Value,
|
||||
input_digest: String,
|
||||
},
|
||||
TurnTerminal {
|
||||
status: AcpxTurnStatus,
|
||||
|
|
@ -123,6 +126,7 @@ pub fn decode_acpx_event(
|
|||
Ok(AcpxEventPayload::ToolCalled {
|
||||
call_id: required_id(&event.payload, "callId", "tool call")?,
|
||||
operation_id: required_id(&event.payload, "operationId", "tool operation")?,
|
||||
input_digest: semantic_value_digest(&input),
|
||||
input: sanitize_value(&input),
|
||||
})
|
||||
}
|
||||
|
|
@ -217,6 +221,7 @@ fn decode_runtime_event(payload: &Value) -> Result<AcpxEventPayload, LocalRunner
|
|||
.and_then(Value::as_str)
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX runtime event omitted its type"))?;
|
||||
let mut tool_operation = None;
|
||||
let mut semantic_result_digest = None;
|
||||
let kind = match runtime_type {
|
||||
"text_delta" => {
|
||||
bounded_required_text(payload, "text", MAX_RUNTIME_TEXT_CHARS, "runtime text")?;
|
||||
|
|
@ -277,11 +282,17 @@ fn decode_runtime_event(payload: &Value) -> Result<AcpxEventPayload, LocalRunner
|
|||
"semantic_result" => {
|
||||
required_id(payload, "callId", "semantic result call")?;
|
||||
required_id(payload, "operationId", "semantic result operation")?;
|
||||
if !payload.get("ok").is_some_and(Value::is_boolean) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX semantic result must contain a boolean outcome",
|
||||
));
|
||||
}
|
||||
if !payload.get("result").is_some_and(Value::is_object) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX semantic result must contain an object result",
|
||||
));
|
||||
}
|
||||
semantic_result_digest = payload.get("result").map(semantic_value_digest);
|
||||
AcpxRuntimeEventKind::SemanticResult
|
||||
}
|
||||
"provider_notice" => {
|
||||
|
|
@ -308,6 +319,7 @@ fn decode_runtime_event(payload: &Value) -> Result<AcpxEventPayload, LocalRunner
|
|||
kind,
|
||||
tool_operation,
|
||||
payload: sanitize_value(payload),
|
||||
semantic_result_digest,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
use std::collections::BTreeSet;
|
||||
|
||||
use crate::acpx_sidecar_transport::AcpxSidecarEvent;
|
||||
use crate::generated_acpx_sidecar_contract::GeneratedAcpxSidecarEventType;
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
|
||||
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.
|
||||
///
|
||||
|
|
@ -13,6 +16,7 @@ const MAX_SCOPE_ID_CHARS: usize = 160;
|
|||
pub struct AcpxEventScope {
|
||||
run_id: String,
|
||||
active_turn_id: Option<String>,
|
||||
settled_turn_ids: BTreeSet<String>,
|
||||
}
|
||||
|
||||
impl AcpxEventScope {
|
||||
|
|
@ -22,6 +26,7 @@ impl AcpxEventScope {
|
|||
Ok(Self {
|
||||
run_id,
|
||||
active_turn_id: None,
|
||||
settled_turn_ids: BTreeSet::new(),
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -33,6 +38,10 @@ impl AcpxEventScope {
|
|||
self.active_turn_id.as_deref()
|
||||
}
|
||||
|
||||
pub(crate) fn has_settled_turns(&self) -> bool {
|
||||
!self.settled_turn_ids.is_empty()
|
||||
}
|
||||
|
||||
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")?;
|
||||
|
|
@ -42,12 +51,57 @@ impl AcpxEventScope {
|
|||
"ACPX event scope already has a different active turn",
|
||||
)),
|
||||
None => {
|
||||
self.validate_new_turn_identity(&turn_id)?;
|
||||
self.active_turn_id = Some(turn_id);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> {
|
||||
self.validate_new_turn_identity_for_provider_restart(turn_id)?;
|
||||
if self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope exhausted its settled turn identity capacity",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn validate_new_turn_identity_for_provider_restart(
|
||||
&self,
|
||||
turn_id: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
validate_scope_id(turn_id, "turn")?;
|
||||
if self.settled_turn_ids.contains(turn_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope reused a settled turn identity",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn settled_turn_identity_capacity_reached(&self) -> bool {
|
||||
self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS
|
||||
}
|
||||
|
||||
pub(crate) fn rotate_settled_turn_identities_after_provider_restart(
|
||||
&mut self,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if self.active_turn_id.is_some() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope cannot rotate settled turn identities while a turn is active",
|
||||
));
|
||||
}
|
||||
if !self.settled_turn_identity_capacity_reached() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope cannot rotate settled turn identities before capacity",
|
||||
));
|
||||
}
|
||||
self.settled_turn_ids.clear();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn clear_turn(&mut self, turn_id: &str) -> Result<(), LocalRunnerError> {
|
||||
validate_scope_id(turn_id, "turn")?;
|
||||
if self.active_turn_id.as_deref() != Some(turn_id) {
|
||||
|
|
@ -55,6 +109,16 @@ impl AcpxEventScope {
|
|||
"ACPX event scope cannot clear a stale turn",
|
||||
));
|
||||
}
|
||||
if self.settled_turn_ids.len() >= MAX_SETTLED_TURN_IDS {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope exhausted its settled turn identity capacity",
|
||||
));
|
||||
}
|
||||
if !self.settled_turn_ids.insert(turn_id.to_owned()) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX event scope reused a settled turn identity",
|
||||
));
|
||||
}
|
||||
self.active_turn_id = None;
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -111,6 +175,34 @@ impl AcpxEventScope {
|
|||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{AcpxEventScope, MAX_SETTLED_TURN_IDS};
|
||||
|
||||
#[test]
|
||||
fn rotates_a_full_ledger_only_after_revalidating_the_next_identity() {
|
||||
let mut scope = AcpxEventScope::new("run-1").unwrap();
|
||||
for index in 0..MAX_SETTLED_TURN_IDS {
|
||||
let turn_id = format!("turn-{index}");
|
||||
scope.bind_turn(&turn_id).unwrap();
|
||||
scope.clear_turn(&turn_id).unwrap();
|
||||
}
|
||||
|
||||
assert!(scope.settled_turn_identity_capacity_reached());
|
||||
scope
|
||||
.validate_new_turn_identity_for_provider_restart("turn-next")
|
||||
.unwrap();
|
||||
assert!(scope
|
||||
.validate_new_turn_identity_for_provider_restart("turn-0")
|
||||
.is_err());
|
||||
scope
|
||||
.rotate_settled_turn_identities_after_provider_restart()
|
||||
.unwrap();
|
||||
scope.bind_turn("turn-next").unwrap();
|
||||
scope.clear_turn("turn-next").unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_scope_id(value: &str, label: &str) -> Result<(), LocalRunnerError> {
|
||||
if value.is_empty()
|
||||
|| value.chars().count() > MAX_SCOPE_ID_CHARS
|
||||
|
|
|
|||
|
|
@ -1,20 +1,26 @@
|
|||
use std::path::PathBuf;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Value};
|
||||
|
||||
use crate::acpx_provider_state::AcpxProviderState;
|
||||
use crate::acpx_provider_state::{AcpxProviderState, AcpxProviderStateEvent};
|
||||
use crate::acpx_sidecar_transport::{AcpxSidecarTransport, AcpxSidecarTransportConfig};
|
||||
use crate::generated_acpx_sidecar_contract::{
|
||||
GeneratedAcpxSidecarCommand, GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
|
||||
};
|
||||
use crate::local_runner::LocalRunnerError;
|
||||
use crate::provider_bridge::{AuthorizedToolSet, ProviderToolBridge};
|
||||
use crate::provider_bridge::{
|
||||
authorized_tool_catalog_digest, AuthorizedTool, AuthorizedToolSet, ProviderToolBridge,
|
||||
TOOL_SET_SCHEMA,
|
||||
};
|
||||
|
||||
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")]
|
||||
|
|
@ -109,10 +115,21 @@ impl AcpxProviderSessionConfig {
|
|||
"ACPX system instructions exceed their bounded contract",
|
||||
));
|
||||
}
|
||||
if self
|
||||
.tool_set
|
||||
.operations
|
||||
.iter()
|
||||
.any(|tool| is_reserved_terminal_operation(&tool.operation_id))
|
||||
{
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX run catalog cannot replace reserved terminal tools",
|
||||
));
|
||||
}
|
||||
let mut bridge = ProviderToolBridge::default();
|
||||
bridge.prepare(self.tool_set.clone()).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}"))
|
||||
})?;
|
||||
reserved_terminal_tool_bridge()?;
|
||||
if let Some(expected_identity) = self.expected_identity.as_ref() {
|
||||
expected_identity.validate()?;
|
||||
if expected_identity.normalized_session_id != self.normalized_session_id
|
||||
|
|
@ -162,9 +179,13 @@ impl AcpxProviderSessionIdentity {
|
|||
|
||||
pub struct AcpxProviderSession {
|
||||
transport: AcpxSidecarTransport,
|
||||
config: AcpxProviderSessionConfig,
|
||||
state: AcpxProviderState,
|
||||
tool_bridge: ProviderToolBridge,
|
||||
reserved_tool_bridge: ProviderToolBridge,
|
||||
identity: AcpxProviderSessionIdentity,
|
||||
catalog_revision: u64,
|
||||
working_directory: PathBuf,
|
||||
closed: bool,
|
||||
transport_terminated: bool,
|
||||
}
|
||||
|
|
@ -172,6 +193,13 @@ pub struct AcpxProviderSession {
|
|||
impl AcpxProviderSession {
|
||||
pub fn start(config: &AcpxProviderSessionConfig) -> Result<Self, LocalRunnerError> {
|
||||
config.validate()?;
|
||||
let mut tool_bridge = ProviderToolBridge::default();
|
||||
tool_bridge
|
||||
.prepare(config.tool_set.clone())
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}"))
|
||||
})?;
|
||||
let reserved_tool_bridge = reserved_terminal_tool_bridge()?;
|
||||
let mut transport = AcpxSidecarTransport::start(&config.transport)?;
|
||||
let bootstrap = bootstrap(&mut transport, config);
|
||||
let (identity, state) = match bootstrap {
|
||||
|
|
@ -183,9 +211,13 @@ impl AcpxProviderSession {
|
|||
};
|
||||
Ok(Self {
|
||||
transport,
|
||||
config: config.clone(),
|
||||
state,
|
||||
tool_bridge,
|
||||
reserved_tool_bridge,
|
||||
identity,
|
||||
catalog_revision: config.catalog_revision,
|
||||
working_directory: config.working_directory.clone(),
|
||||
closed: false,
|
||||
transport_terminated: false,
|
||||
})
|
||||
|
|
@ -207,6 +239,287 @@ impl AcpxProviderSession {
|
|||
self.catalog_revision
|
||||
}
|
||||
|
||||
pub fn start_turn(
|
||||
&mut self,
|
||||
turn_id: &str,
|
||||
message: &str,
|
||||
working_directory: &Path,
|
||||
) -> Result<Value, LocalRunnerError> {
|
||||
self.ensure_open()?;
|
||||
validate_text(turn_id, 160, "ACPX turn id")?;
|
||||
validate_turn_message(message)?;
|
||||
if working_directory != self.working_directory {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX turn working directory differs from its immutable session workspace",
|
||||
));
|
||||
}
|
||||
if self.state.active_turn_id().is_some() {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX provider session already has an active turn",
|
||||
));
|
||||
}
|
||||
let rotate_turn_identity_ledger = self.state.settled_turn_identity_capacity_reached();
|
||||
let identity_validation = if rotate_turn_identity_ledger {
|
||||
self.state
|
||||
.validate_new_turn_identity_for_provider_restart(turn_id)
|
||||
} else {
|
||||
self.state.validate_new_turn_identity(turn_id)
|
||||
};
|
||||
if let Err(error) = identity_validation {
|
||||
// Reusing a settled identity would let a delayed event from the
|
||||
// old turn alias the new receipt epoch. A full sidecar restart is
|
||||
// required before an exhausted ledger can rotate, while an exact
|
||||
// identity reuse remains forbidden within the current ledger.
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
// The MCP endpoint is session-lifetime and cannot authenticate which
|
||||
// provider turn originated a late HTTP callback. Reap the old sidecar
|
||||
// and provider before releasing its call-ID tombstones, then resume
|
||||
// the same verified persistent session in a fresh process generation.
|
||||
let provider_restarted = self.state.has_settled_turns();
|
||||
if provider_restarted {
|
||||
if let Err(error) = self.restart_idle_provider() {
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
}
|
||||
if rotate_turn_identity_ledger {
|
||||
if let Err(error) = self
|
||||
.state
|
||||
.rotate_settled_turn_identities_after_provider_restart()
|
||||
{
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
}
|
||||
// Prepare cloned receipt epochs before asking the replacement sidecar
|
||||
// to start work, then publish them only after both the provider and
|
||||
// reducer accept the new turn.
|
||||
let mut next_tool_bridge = self.tool_bridge.clone();
|
||||
let dynamic_preparation = if provider_restarted {
|
||||
next_tool_bridge.prepare_turn_after_provider_restart()
|
||||
} else {
|
||||
next_tool_bridge.prepare_turn()
|
||||
};
|
||||
if let Err(error) = dynamic_preparation {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX dynamic tool receipt rotation failed: {error}"
|
||||
))));
|
||||
}
|
||||
let mut next_reserved_tool_bridge = self.reserved_tool_bridge.clone();
|
||||
let reserved_preparation = if provider_restarted {
|
||||
next_reserved_tool_bridge.prepare_turn_after_provider_restart()
|
||||
} else {
|
||||
next_reserved_tool_bridge.prepare_turn()
|
||||
};
|
||||
if let Err(error) = reserved_preparation {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX reserved tool receipt rotation failed: {error}"
|
||||
))));
|
||||
}
|
||||
let response = match self.transport.request(
|
||||
GeneratedAcpxSidecarCommand::TurnStart,
|
||||
json!({"turnId":turn_id,"message":message}),
|
||||
) {
|
||||
Ok(response) => response,
|
||||
Err(error) => return Err(self.fail_closed(error)),
|
||||
};
|
||||
if response.get("turnId").and_then(Value::as_str) != Some(turn_id) {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX sidecar did not confirm the requested turn",
|
||||
)));
|
||||
}
|
||||
if let Err(error) = self.state.begin_turn(turn_id) {
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
self.tool_bridge = next_tool_bridge;
|
||||
self.reserved_tool_bridge = next_reserved_tool_bridge;
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
pub fn interrupt_turn(
|
||||
&mut self,
|
||||
turn_id: &str,
|
||||
reason: &str,
|
||||
) -> Result<Value, LocalRunnerError> {
|
||||
self.ensure_open()?;
|
||||
validate_text(turn_id, 160, "ACPX turn id")?;
|
||||
if self.state.active_turn_id() != Some(turn_id) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX interruption named a stale or inactive turn",
|
||||
));
|
||||
}
|
||||
let response = match self.transport.request(
|
||||
GeneratedAcpxSidecarCommand::TurnCancel,
|
||||
json!({"turnId":turn_id,"reason":bounded_reason(reason)}),
|
||||
) {
|
||||
Ok(response) => response,
|
||||
Err(error) => return Err(self.fail_closed(error)),
|
||||
};
|
||||
if response.get("cancelled").and_then(Value::as_bool) != Some(true) {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX sidecar did not confirm turn cancellation",
|
||||
)));
|
||||
}
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
pub fn poll_event(
|
||||
&mut self,
|
||||
timeout: Duration,
|
||||
) -> Result<Option<Vec<AcpxProviderStateEvent>>, LocalRunnerError> {
|
||||
self.ensure_open()?;
|
||||
let event = match self.transport.poll_event(timeout) {
|
||||
Ok(event) => event,
|
||||
Err(error) => return Err(self.fail_closed(error)),
|
||||
};
|
||||
let Some(event) = event else {
|
||||
return Ok(None);
|
||||
};
|
||||
let mut next_state = self.state.clone();
|
||||
let events = match next_state.accept_event(&event) {
|
||||
Ok(events) => events,
|
||||
Err(error) => return Err(self.fail_closed(error)),
|
||||
};
|
||||
let mut next_bridge = self.tool_bridge.clone();
|
||||
let mut next_reserved_bridge = self.reserved_tool_bridge.clone();
|
||||
let mut reconciled_events = Vec::with_capacity(events.len());
|
||||
for event in events {
|
||||
let mut expose_event = true;
|
||||
match &event {
|
||||
AcpxProviderStateEvent::ToolCall {
|
||||
call_id,
|
||||
operation_id,
|
||||
input,
|
||||
} => {
|
||||
let bridge = if is_reserved_terminal_operation(operation_id) {
|
||||
if let Err(error) = validate_reserved_terminal_value(operation_id, input) {
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
// These built-ins are authorized by the same ledger as
|
||||
// dynamic tools, but the server dispatcher must never
|
||||
// execute them as ordinary semantic operations.
|
||||
expose_event = false;
|
||||
if next_bridge.has_call_receipt(call_id) {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX reused a dynamic call id for a reserved terminal invocation",
|
||||
)));
|
||||
}
|
||||
&mut next_reserved_bridge
|
||||
} else {
|
||||
if next_reserved_bridge.has_call_receipt(call_id) {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX reused a reserved call id for a dynamic tool invocation",
|
||||
)));
|
||||
}
|
||||
&mut next_bridge
|
||||
};
|
||||
if let Err(error) =
|
||||
bridge.begin_call(call_id.clone(), operation_id.clone(), input.clone())
|
||||
{
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX provider tool authorization failed: {error}"
|
||||
))));
|
||||
}
|
||||
}
|
||||
AcpxProviderStateEvent::SemanticResult(result) => {
|
||||
if is_reserved_terminal_result(result) {
|
||||
if let Err(error) = validate_reserved_terminal_result(&next_state, result) {
|
||||
return Err(self.fail_closed(error));
|
||||
}
|
||||
if let Err(error) =
|
||||
next_reserved_bridge.apply_result(crate::provider_bridge::ToolResult {
|
||||
call_id: result.call_id.clone(),
|
||||
operation_id: result.operation_id.clone(),
|
||||
result: result.result.clone(),
|
||||
is_error: !result.ok,
|
||||
})
|
||||
{
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX reserved terminal result reconciliation failed: {error}"
|
||||
))));
|
||||
}
|
||||
} else {
|
||||
let replayed = next_bridge.has_completed_call(&result.call_id);
|
||||
if let Err(error) =
|
||||
next_bridge.apply_result(crate::provider_bridge::ToolResult {
|
||||
call_id: result.call_id.clone(),
|
||||
operation_id: result.operation_id.clone(),
|
||||
result: result.result.clone(),
|
||||
is_error: !result.ok,
|
||||
})
|
||||
{
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX provider tool result reconciliation failed: {error}"
|
||||
))));
|
||||
}
|
||||
if replayed {
|
||||
expose_event = false;
|
||||
}
|
||||
}
|
||||
if next_state.pending_tool(&result.call_id).is_some() {
|
||||
if let Err(error) =
|
||||
next_state.complete_tool(&result.call_id, &result.operation_id)
|
||||
{
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX provider tool completion reconciliation failed: {error}"
|
||||
))));
|
||||
}
|
||||
}
|
||||
}
|
||||
AcpxProviderStateEvent::TurnTerminal { .. } => {
|
||||
let settlements = match next_bridge.settle_turn("acpx_turn_settled") {
|
||||
Ok(settlements) => settlements,
|
||||
Err(error) => {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX provider tool settlement failed: {error}"
|
||||
))));
|
||||
}
|
||||
};
|
||||
let reserved_settlements = match next_reserved_bridge
|
||||
.settle_turn("acpx_reserved_terminal_unsettled")
|
||||
{
|
||||
Ok(settlements) => settlements,
|
||||
Err(error) => {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(format!(
|
||||
"ACPX reserved terminal settlement failed: {error}"
|
||||
))));
|
||||
}
|
||||
};
|
||||
if !reserved_settlements.is_empty() {
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX turn terminated before its reserved terminal invocation produced a correlated result",
|
||||
)));
|
||||
}
|
||||
// `accept_event` clears the candidate reducer's pending
|
||||
// tools while the bridge clones settle the corresponding
|
||||
// calls above. Prove both halves reached the same terminal
|
||||
// state before committing any of them to the reusable
|
||||
// session.
|
||||
if next_state.has_pending_tools()
|
||||
|| next_bridge.pending_calls().next().is_some()
|
||||
|| next_reserved_bridge.pending_calls().next().is_some()
|
||||
{
|
||||
return Err(self.fail_closed(LocalRunnerError::invalid(
|
||||
"ACPX terminal settlement left provider tool state inconsistent",
|
||||
)));
|
||||
}
|
||||
reconciled_events.extend(
|
||||
settlements
|
||||
.into_iter()
|
||||
.map(AcpxProviderStateEvent::ToolResult),
|
||||
);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
if expose_event {
|
||||
reconciled_events.push(event);
|
||||
}
|
||||
}
|
||||
self.state = next_state;
|
||||
self.tool_bridge = next_bridge;
|
||||
self.reserved_tool_bridge = next_reserved_bridge;
|
||||
Ok(Some(reconciled_events))
|
||||
}
|
||||
|
||||
pub fn shutdown(&mut self, reason: &str) -> Result<(), LocalRunnerError> {
|
||||
if self.closed {
|
||||
return self.terminate_transport();
|
||||
|
|
@ -235,6 +548,170 @@ impl AcpxProviderSession {
|
|||
self.transport_terminated = true;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn ensure_open(&self) -> Result<(), LocalRunnerError> {
|
||||
if self.closed {
|
||||
return Err(LocalRunnerError::invalid("ACPX provider session is closed"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn restart_idle_provider(&mut self) -> Result<(), LocalRunnerError> {
|
||||
let suspended = self.transport.request(
|
||||
GeneratedAcpxSidecarCommand::SessionSuspend,
|
||||
json!({"reason":"ACPX turn receipt epoch rotation"}),
|
||||
)?;
|
||||
verify_suspend_response(&suspended, &self.identity)?;
|
||||
self.transport.shutdown()?;
|
||||
self.transport_terminated = true;
|
||||
|
||||
let mut restart_config = self.config.clone();
|
||||
restart_config.expected_identity = Some(self.identity.clone());
|
||||
let mut replacement = AcpxSidecarTransport::start(&restart_config.transport)?;
|
||||
let (replacement_identity, _) = match bootstrap(&mut replacement, &restart_config) {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
return Err(self.reject_replacement(replacement, error));
|
||||
}
|
||||
};
|
||||
if replacement_identity != self.identity {
|
||||
return Err(self.reject_replacement(
|
||||
replacement,
|
||||
LocalRunnerError::invalid(
|
||||
"ACPX replacement provider changed its persistent session identity",
|
||||
),
|
||||
));
|
||||
}
|
||||
self.transport = replacement;
|
||||
self.transport_terminated = false;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn reject_replacement(
|
||||
&mut self,
|
||||
mut replacement: AcpxSidecarTransport,
|
||||
error: LocalRunnerError,
|
||||
) -> LocalRunnerError {
|
||||
let cleanup = replacement.shutdown();
|
||||
if cleanup.is_err() {
|
||||
// Keep the exact failed generation reachable so fail_closed can
|
||||
// retry its process-group termination instead of dropping the
|
||||
// only remaining cleanup authority.
|
||||
self.transport = replacement;
|
||||
self.transport_terminated = false;
|
||||
}
|
||||
with_cleanup_error(error, cleanup)
|
||||
}
|
||||
|
||||
fn fail_closed(&mut self, error: LocalRunnerError) -> LocalRunnerError {
|
||||
self.closed = true;
|
||||
with_cleanup_error(error, self.terminate_transport())
|
||||
}
|
||||
}
|
||||
|
||||
fn is_reserved_terminal_result(result: &crate::acpx_provider_state::AcpxSemanticResult) -> bool {
|
||||
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,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if !result.ok {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX reserved semantic result reported a failed outcome",
|
||||
));
|
||||
}
|
||||
let pending = state.pending_tool(&result.call_id).ok_or_else(|| {
|
||||
LocalRunnerError::invalid(
|
||||
"ACPX reserved semantic result has no authorized pending invocation",
|
||||
)
|
||||
})?;
|
||||
if pending.operation_id != result.operation_id || pending.input_digest != result.result_digest {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX reserved semantic result does not match its authorized invocation",
|
||||
));
|
||||
}
|
||||
validate_reserved_terminal_value(&result.operation_id, &result.result)
|
||||
}
|
||||
|
||||
fn validate_reserved_terminal_value(
|
||||
operation_id: &str,
|
||||
value: &Value,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
validate_prp_run_result(value)?;
|
||||
let disposition = value.get("reportedWorkDisposition").and_then(Value::as_str);
|
||||
let disposition_matches = match operation_id {
|
||||
PRP_BLOCK_TOOL_NAME => disposition == Some("blocked"),
|
||||
PRP_COMPLETION_TOOL_NAME => {
|
||||
matches!(disposition, Some("done" | "needs_review" | "yielded"))
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
if !disposition_matches {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX reserved semantic result disposition does not match its operation",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn reserved_terminal_tool_bridge() -> Result<ProviderToolBridge, LocalRunnerError> {
|
||||
let result_schema: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/schemas/result.schema.json"
|
||||
))
|
||||
.map_err(|_| LocalRunnerError::invalid("embedded Paperclip result schema is invalid"))?;
|
||||
let operations = vec![
|
||||
AuthorizedTool {
|
||||
operation_id: PRP_COMPLETION_TOOL_NAME.to_owned(),
|
||||
version: 1,
|
||||
description: "Return the authoritative Paperclip completion result.".to_owned(),
|
||||
input_schema: result_schema.clone(),
|
||||
response_schema: result_schema.clone(),
|
||||
},
|
||||
AuthorizedTool {
|
||||
operation_id: PRP_BLOCK_TOOL_NAME.to_owned(),
|
||||
version: 1,
|
||||
description: "Return the authoritative Paperclip blocked result.".to_owned(),
|
||||
input_schema: result_schema.clone(),
|
||||
response_schema: result_schema,
|
||||
},
|
||||
];
|
||||
let catalog_digest = authorized_tool_catalog_digest(&operations).map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX reserved terminal tools are invalid: {error}"))
|
||||
})?;
|
||||
let mut bridge = ProviderToolBridge::default();
|
||||
bridge
|
||||
.prepare(AuthorizedToolSet {
|
||||
schema: TOOL_SET_SCHEMA.to_owned(),
|
||||
schema_version: 1,
|
||||
catalog_digest,
|
||||
operations,
|
||||
})
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX reserved terminal tools are invalid: {error}"))
|
||||
})?;
|
||||
Ok(bridge)
|
||||
}
|
||||
|
||||
fn validate_prp_run_result(value: &Value) -> Result<(), LocalRunnerError> {
|
||||
let schema: Value = serde_json::from_str(include_str!(
|
||||
"../../../../protocol/schemas/result.schema.json"
|
||||
))
|
||||
.map_err(|_| LocalRunnerError::invalid("embedded Paperclip result schema is invalid"))?;
|
||||
let validator = jsonschema::validator_for(&schema).map_err(|_| {
|
||||
LocalRunnerError::invalid("embedded Paperclip result schema cannot compile")
|
||||
})?;
|
||||
if !validator.is_valid(value) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX reserved semantic result failed the Paperclip result schema",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl Drop for AcpxProviderSession {
|
||||
|
|
@ -363,6 +840,33 @@ fn verify_open_response(
|
|||
Ok(identity)
|
||||
}
|
||||
|
||||
fn verify_suspend_response(
|
||||
value: &Value,
|
||||
expected_identity: &AcpxProviderSessionIdentity,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
if value.get("suspended").and_then(Value::as_bool) != Some(true) {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX sidecar did not confirm provider suspension",
|
||||
));
|
||||
}
|
||||
let identity: AcpxProviderSessionIdentity = serde_json::from_value(
|
||||
value
|
||||
.get("identity")
|
||||
.cloned()
|
||||
.ok_or_else(|| LocalRunnerError::invalid("ACPX suspension omitted its identity"))?,
|
||||
)
|
||||
.map_err(|error| {
|
||||
LocalRunnerError::invalid(format!("ACPX suspension identity is invalid: {error}"))
|
||||
})?;
|
||||
identity.validate()?;
|
||||
if &identity != expected_identity {
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX suspension changed its persistent session identity",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_text(value: &str, max_chars: usize, label: &str) -> Result<(), LocalRunnerError> {
|
||||
if value.trim().is_empty()
|
||||
|| value.chars().count() > max_chars
|
||||
|
|
@ -385,6 +889,18 @@ fn bounded_reason(value: &str) -> String {
|
|||
value.chars().take(4_000).collect()
|
||||
}
|
||||
|
||||
fn validate_turn_message(value: &str) -> Result<(), LocalRunnerError> {
|
||||
if value.trim().is_empty()
|
||||
|| value.len() > MAX_SYSTEM_INSTRUCTIONS_BYTES
|
||||
|| value.contains('\0')
|
||||
{
|
||||
return Err(LocalRunnerError::invalid(
|
||||
"ACPX turn message exceeds its bounded contract",
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn with_cleanup_error(
|
||||
error: LocalRunnerError,
|
||||
cleanup: Result<(), LocalRunnerError>,
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ use crate::acpx_event_payload::{
|
|||
use crate::acpx_event_scope::AcpxEventScope;
|
||||
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};
|
||||
|
||||
const MAX_ASSISTANT_TEXT_BYTES: usize = 1024 * 1024;
|
||||
|
|
@ -15,11 +16,14 @@ 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";
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct AcpxPendingTool {
|
||||
pub operation_id: String,
|
||||
pub input: Value,
|
||||
pub(crate) input_digest: String,
|
||||
input_bytes: usize,
|
||||
}
|
||||
|
||||
|
|
@ -27,7 +31,9 @@ pub struct AcpxPendingTool {
|
|||
pub struct AcpxSemanticResult {
|
||||
pub call_id: String,
|
||||
pub operation_id: String,
|
||||
pub ok: bool,
|
||||
pub result: Value,
|
||||
pub(crate) result_digest: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
|
|
@ -38,6 +44,7 @@ pub enum AcpxProviderStateEvent {
|
|||
operation_id: String,
|
||||
input: Value,
|
||||
},
|
||||
ToolResult(ToolResult),
|
||||
PermissionRequest {
|
||||
request_id: String,
|
||||
kind: String,
|
||||
|
|
@ -115,6 +122,33 @@ impl AcpxProviderState {
|
|||
self.scope.active_turn_id()
|
||||
}
|
||||
|
||||
pub(crate) fn has_settled_turns(&self) -> bool {
|
||||
self.scope.has_settled_turns()
|
||||
}
|
||||
|
||||
pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> {
|
||||
self.scope.validate_new_turn_identity(turn_id)
|
||||
}
|
||||
|
||||
pub(crate) fn validate_new_turn_identity_for_provider_restart(
|
||||
&self,
|
||||
turn_id: &str,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
self.scope
|
||||
.validate_new_turn_identity_for_provider_restart(turn_id)
|
||||
}
|
||||
|
||||
pub(crate) fn settled_turn_identity_capacity_reached(&self) -> bool {
|
||||
self.scope.settled_turn_identity_capacity_reached()
|
||||
}
|
||||
|
||||
pub(crate) fn rotate_settled_turn_identities_after_provider_restart(
|
||||
&mut self,
|
||||
) -> Result<(), LocalRunnerError> {
|
||||
self.scope
|
||||
.rotate_settled_turn_identities_after_provider_restart()
|
||||
}
|
||||
|
||||
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()
|
||||
|
|
@ -147,7 +181,14 @@ impl AcpxProviderState {
|
|||
kind,
|
||||
tool_operation,
|
||||
payload,
|
||||
} => self.accept_runtime_event(event, kind, tool_operation, payload),
|
||||
semantic_result_digest,
|
||||
} => self.accept_runtime_event(
|
||||
event,
|
||||
kind,
|
||||
tool_operation,
|
||||
payload,
|
||||
semantic_result_digest,
|
||||
),
|
||||
AcpxEventPayload::PermissionRequested {
|
||||
request_id,
|
||||
kind,
|
||||
|
|
@ -193,6 +234,7 @@ impl AcpxProviderState {
|
|||
call_id,
|
||||
operation_id,
|
||||
input,
|
||||
input_digest,
|
||||
} => {
|
||||
let input_bytes = value_bytes(&input)?;
|
||||
if self.pending_tools.len() >= MAX_PENDING_TOOLS
|
||||
|
|
@ -215,6 +257,7 @@ impl AcpxProviderState {
|
|||
AcpxPendingTool {
|
||||
operation_id: operation_id.clone(),
|
||||
input: input.clone(),
|
||||
input_digest,
|
||||
input_bytes,
|
||||
},
|
||||
);
|
||||
|
|
@ -261,6 +304,10 @@ impl AcpxProviderState {
|
|||
self.pending_tools.get(call_id)
|
||||
}
|
||||
|
||||
pub fn has_pending_tools(&self) -> bool {
|
||||
!self.pending_tools.is_empty()
|
||||
}
|
||||
|
||||
pub fn complete_tool(
|
||||
&mut self,
|
||||
call_id: &str,
|
||||
|
|
@ -315,6 +362,7 @@ impl AcpxProviderState {
|
|||
kind: AcpxRuntimeEventKind,
|
||||
tool_operation: Option<&'static str>,
|
||||
payload: Value,
|
||||
semantic_result_digest: Option<String>,
|
||||
) -> Result<Vec<AcpxProviderStateEvent>, LocalRunnerError> {
|
||||
if kind == AcpxRuntimeEventKind::Thinking {
|
||||
if self.thinking_active {
|
||||
|
|
@ -351,21 +399,36 @@ impl AcpxProviderState {
|
|||
.and_then(Value::as_str)
|
||||
.expect("decoded ACPX semantic result has an operation id")
|
||||
.to_owned(),
|
||||
ok: payload
|
||||
.get("ok")
|
||||
.and_then(Value::as_bool)
|
||||
.expect("decoded ACPX semantic result has an outcome"),
|
||||
result: payload
|
||||
.get("result")
|
||||
.expect("decoded ACPX semantic result has a result")
|
||||
.clone(),
|
||||
result_digest: semantic_result_digest
|
||||
.expect("decoded ACPX semantic result has a raw correlation digest"),
|
||||
};
|
||||
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",
|
||||
)),
|
||||
};
|
||||
if matches!(
|
||||
result.operation_id.as_str(),
|
||||
PRP_COMPLETION_TOOL_NAME | PRP_BLOCK_TOOL_NAME
|
||||
) {
|
||||
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 terminal semantic results for one turn",
|
||||
)),
|
||||
};
|
||||
}
|
||||
// Dynamic results are independently authorized and deduplicated
|
||||
// by call ID in ProviderToolBridge. They must not compete for the
|
||||
// turn-wide terminal-result slot.
|
||||
return Ok(vec![AcpxProviderStateEvent::SemanticResult(result)]);
|
||||
}
|
||||
let turn_id = event
|
||||
.turn_id
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
use std::io::{self, BufRead, Write};
|
||||
use std::time::Duration;
|
||||
|
||||
use paperclip_runner_core::generated_acpx_sidecar_contract::GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION;
|
||||
use serde_json::{json, Value};
|
||||
|
|
@ -79,8 +80,361 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
|||
eprintln!("amber-signal-7305");
|
||||
std::process::exit(9);
|
||||
}
|
||||
"bootstrap" | "bootstrap-wrong-model" | "bootstrap-wrong-run" => {
|
||||
"bootstrap"
|
||||
| "bootstrap-wrong-model"
|
||||
| "bootstrap-wrong-run"
|
||||
| "turns"
|
||||
| "turns-wrong-turn"
|
||||
| "turns-wrong-cancel"
|
||||
| "turns-wrong-scope"
|
||||
| "turns-tool"
|
||||
| "turns-tool-terminal"
|
||||
| "turns-reused-tool-id-terminal"
|
||||
| "turns-late-tool-after-suspend"
|
||||
| "turns-tool-result-terminal"
|
||||
| "turns-tool-error-result-terminal"
|
||||
| "turns-multiple-tool-results-terminal"
|
||||
| "turns-reserved-result-terminal"
|
||||
| "turns-reserved-yielded-terminal"
|
||||
| "turns-reserved-block-terminal"
|
||||
| "turns-sensitive-reserved-result-terminal"
|
||||
| "turns-mismatched-sensitive-reserved-result-terminal"
|
||||
| "turns-invalid-reserved-block-terminal"
|
||||
| "turns-uncorrelated-reserved-result-terminal"
|
||||
| "turns-mismatched-reserved-result-terminal"
|
||||
| "turns-unauthorized-tool" => {
|
||||
write_json(&mut stdout, &bootstrap_success(id, command, &request, mode))?;
|
||||
let params = request.get("params").unwrap_or(&Value::Null);
|
||||
let turn_id = params
|
||||
.get("turnId")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("missing");
|
||||
let tool_call_id = if matches!(
|
||||
mode,
|
||||
"turns-reused-tool-id-terminal" | "turns-late-tool-after-suspend"
|
||||
) {
|
||||
"call-reused".to_owned()
|
||||
} else {
|
||||
turn_id
|
||||
.strip_prefix("turn-")
|
||||
.map(|suffix| format!("call-{suffix}"))
|
||||
.unwrap_or_else(|| "call-1".to_owned())
|
||||
};
|
||||
if command == "turn.start" && matches!(mode, "turns" | "turns-wrong-scope") {
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.event",
|
||||
if mode == "turns-wrong-scope" {
|
||||
"wrong-run"
|
||||
} else {
|
||||
"run-1"
|
||||
},
|
||||
turn_id,
|
||||
json!({"type":"text_delta","text":"hello"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.start"
|
||||
&& matches!(
|
||||
mode,
|
||||
"turns-tool"
|
||||
| "turns-tool-terminal"
|
||||
| "turns-reused-tool-id-terminal"
|
||||
| "turns-late-tool-after-suspend"
|
||||
| "turns-tool-result-terminal"
|
||||
| "turns-tool-error-result-terminal"
|
||||
| "turns-unauthorized-tool"
|
||||
)
|
||||
{
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.tool_called",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"callId":tool_call_id.clone(),
|
||||
"operationId":if matches!(mode, "turns-tool" | "turns-tool-terminal" | "turns-reused-tool-id-terminal" | "turns-late-tool-after-suspend" | "turns-tool-result-terminal" | "turns-tool-error-result-terminal") { "issues.read" } else { "issues.delete" },
|
||||
"input":{"id":"issue-1"},
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.start"
|
||||
&& matches!(
|
||||
mode,
|
||||
"turns-tool-result-terminal" | "turns-tool-error-result-terminal"
|
||||
)
|
||||
{
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.event",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"type":"semantic_result",
|
||||
"callId":tool_call_id,
|
||||
"operationId":"issues.read",
|
||||
"ok":mode == "turns-tool-result-terminal",
|
||||
"result":if mode == "turns-tool-result-terminal" { json!({"id":"issue-1"}) } else { json!({"error":{"code":"tool_failed"}}) },
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.turn_terminal",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({"status":"completed"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.start"
|
||||
&& matches!(
|
||||
mode,
|
||||
"turns-tool-terminal"
|
||||
| "turns-reused-tool-id-terminal"
|
||||
| "turns-late-tool-after-suspend"
|
||||
)
|
||||
{
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.turn_terminal",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({"status":"completed"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.start" && mode == "turns-multiple-tool-results-terminal" {
|
||||
for index in 1..=2 {
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.tool_called",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"callId":format!("call-{index}"),
|
||||
"operationId":"issues.read",
|
||||
"input":{"id":format!("issue-{index}")},
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
for index in 1..=2 {
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.event",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"type":"semantic_result",
|
||||
"callId":format!("call-{index}"),
|
||||
"operationId":"issues.read",
|
||||
"ok":true,
|
||||
"result":{"id":format!("issue-{index}")},
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.turn_terminal",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({"status":"completed"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.start"
|
||||
&& matches!(
|
||||
mode,
|
||||
"turns-reserved-result-terminal"
|
||||
| "turns-reserved-yielded-terminal"
|
||||
| "turns-reserved-block-terminal"
|
||||
| "turns-sensitive-reserved-result-terminal"
|
||||
| "turns-mismatched-sensitive-reserved-result-terminal"
|
||||
| "turns-invalid-reserved-block-terminal"
|
||||
| "turns-uncorrelated-reserved-result-terminal"
|
||||
| "turns-mismatched-reserved-result-terminal"
|
||||
)
|
||||
{
|
||||
let (operation_id, result) = if matches!(
|
||||
mode,
|
||||
"turns-reserved-block-terminal" | "turns-invalid-reserved-block-terminal"
|
||||
) {
|
||||
(
|
||||
"paperclip_block",
|
||||
json!({
|
||||
"schema":"paperclip.run_result.v1",
|
||||
"reportedWorkDisposition":"blocked",
|
||||
"summary":"Reserved blocker accepted.",
|
||||
"completionClaim":{
|
||||
"contractRevision":"acpx-provider-turns-v1",
|
||||
"objectiveSatisfied":false,
|
||||
"criteria":[],
|
||||
"remainingWork":[{
|
||||
"description":"Wait for external input.",
|
||||
"blocksCompletion":true,
|
||||
}],
|
||||
},
|
||||
"evidence":[],
|
||||
"verification":[],
|
||||
"blocker":{
|
||||
"reasonCode":"external_input",
|
||||
"owner":if mode == "turns-invalid-reserved-block-terminal" {
|
||||
json!({})
|
||||
} else {
|
||||
json!({"kind":"external","name":"External input"})
|
||||
},
|
||||
"unblockAction":"Provide the required input.",
|
||||
"scope":"current_track",
|
||||
},
|
||||
"attentionRequests":[],
|
||||
"artifacts":[],
|
||||
}),
|
||||
)
|
||||
} else if mode == "turns-reserved-yielded-terminal" {
|
||||
(
|
||||
"paperclip_finish",
|
||||
json!({
|
||||
"schema":"paperclip.run_result.v1",
|
||||
"reportedWorkDisposition":"yielded",
|
||||
"summary":"Reserved continuation accepted.",
|
||||
"completionClaim":{
|
||||
"contractRevision":"acpx-provider-turns-v1",
|
||||
"objectiveSatisfied":false,
|
||||
"criteria":[],
|
||||
"remainingWork":[],
|
||||
},
|
||||
"evidence":[],
|
||||
"verification":[],
|
||||
"continuation":{
|
||||
"kind":"same_agent",
|
||||
"summary":"Continue the current run.",
|
||||
"idempotencyKey":"continuation-1",
|
||||
},
|
||||
"attentionRequests":[],
|
||||
"artifacts":[],
|
||||
}),
|
||||
)
|
||||
} else {
|
||||
(
|
||||
"paperclip_finish",
|
||||
json!({
|
||||
"schema":"paperclip.run_result.v1",
|
||||
"reportedWorkDisposition":"done",
|
||||
"summary":if matches!(
|
||||
mode,
|
||||
"turns-sensitive-reserved-result-terminal"
|
||||
| "turns-mismatched-sensitive-reserved-result-terminal"
|
||||
) {
|
||||
"token=matching-sensitive-value"
|
||||
} else {
|
||||
"Reserved completion accepted."
|
||||
},
|
||||
"completionClaim":{
|
||||
"contractRevision":"acpx-provider-turns-v1",
|
||||
"objectiveSatisfied":true,
|
||||
"criteria":[],
|
||||
"remainingWork":[],
|
||||
},
|
||||
"evidence":[],
|
||||
"verification":[],
|
||||
"attentionRequests":[],
|
||||
"artifacts":[],
|
||||
}),
|
||||
)
|
||||
};
|
||||
if mode != "turns-uncorrelated-reserved-result-terminal" {
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.tool_called",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"callId":"call-finish",
|
||||
"operationId":operation_id,
|
||||
"input":result.clone(),
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
let semantic_result = if mode == "turns-mismatched-reserved-result-terminal" {
|
||||
let mut changed = result.clone();
|
||||
changed["summary"] = json!("A different terminal result.");
|
||||
changed
|
||||
} else if mode == "turns-mismatched-sensitive-reserved-result-terminal" {
|
||||
let mut changed = result.clone();
|
||||
changed["summary"] = json!("token=different-sensitive-value");
|
||||
changed
|
||||
} else {
|
||||
result
|
||||
};
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.event",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({
|
||||
"type":"semantic_result",
|
||||
"callId":"call-finish",
|
||||
"operationId":operation_id,
|
||||
"ok":true,
|
||||
"result":semantic_result,
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.turn_terminal",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({"status":"completed"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "session.suspend" && mode == "turns-late-tool-after-suspend" {
|
||||
// Simulate a session-lifetime callback that wakes after
|
||||
// suspension and reads the next mutable turn binding.
|
||||
// runner-core must reap this process before starting that
|
||||
// turn, so the relabeled event can never cross authority.
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.tool_called",
|
||||
"run-1",
|
||||
"turn-2",
|
||||
json!({
|
||||
"callId":"call-reused",
|
||||
"operationId":"issues.delete",
|
||||
"input":{"source":"late-old-turn"},
|
||||
}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
if command == "turn.cancel" && mode == "turns" {
|
||||
write_turn_event(
|
||||
&mut stdout,
|
||||
next_sequence,
|
||||
"runtime.turn_terminal",
|
||||
"run-1",
|
||||
turn_id,
|
||||
json!({"status":"interrupted"}),
|
||||
)?;
|
||||
next_sequence += 1;
|
||||
}
|
||||
}
|
||||
"happy" => {
|
||||
write_event(&mut stdout, next_sequence)?;
|
||||
|
|
@ -134,6 +488,25 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val
|
|||
"runId": if mode == "bootstrap-wrong-run" { "wrong-run" } else { params.get("runId").and_then(Value::as_str).unwrap_or("missing") },
|
||||
"catalogRevision": params.get("catalogRevision"),
|
||||
}),
|
||||
"turn.start" => json!({
|
||||
"turnId": if mode == "turns-wrong-turn" { "wrong-turn" } else { params.get("turnId").and_then(Value::as_str).unwrap_or("missing") },
|
||||
}),
|
||||
"turn.cancel" => json!({"cancelled":mode != "turns-wrong-cancel"}),
|
||||
"session.suspend" => json!({
|
||||
"suspended":true,
|
||||
"identity": {
|
||||
"kind": "acpx",
|
||||
"normalizedSessionId": "session-1",
|
||||
"acpxRecordId": "record-1",
|
||||
"backendSessionId": "backend-1",
|
||||
"agentSessionId": "agent-1",
|
||||
"profileDigest": format!("sha256:{}", "1".repeat(64)),
|
||||
"workspaceDigest": format!("sha256:{}", "2".repeat(64)),
|
||||
"requestedModel": "gpt-5.6-sol",
|
||||
"effectiveModel": "gpt-5.6-sol",
|
||||
"permissionMode": "approve-reads",
|
||||
},
|
||||
}),
|
||||
"session.close" => json!({"closed":true}),
|
||||
_ => json!({"command":command,"params":params}),
|
||||
};
|
||||
|
|
@ -145,6 +518,27 @@ fn bootstrap_success(id: u64, command: &str, request: &Value, mode: &str) -> Val
|
|||
})
|
||||
}
|
||||
|
||||
fn write_turn_event(
|
||||
output: &mut impl Write,
|
||||
sequence: u64,
|
||||
event_type: &str,
|
||||
run_id: &str,
|
||||
turn_id: &str,
|
||||
payload: Value,
|
||||
) -> io::Result<()> {
|
||||
write_json(
|
||||
output,
|
||||
&json!({
|
||||
"protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
|
||||
"sequence": sequence,
|
||||
"eventType": event_type,
|
||||
"runId": run_id,
|
||||
"turnId": turn_id,
|
||||
"payload": payload,
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
fn success(id: u64, command: &str, request: &Value) -> Value {
|
||||
json!({
|
||||
"protocolVersion": GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
|
||||
|
|
|
|||
|
|
@ -520,6 +520,20 @@ impl ProviderToolBridge {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// Start a receipt epoch after the process that owned the previous epoch
|
||||
/// has been reaped and a replacement provider generation is established.
|
||||
/// Providers that continue in the same process must retain their durable
|
||||
/// call-ID tombstones through ordinary `prepare_turn` transitions.
|
||||
pub(crate) fn prepare_turn_after_provider_restart(
|
||||
&mut self,
|
||||
) -> Result<(), ProviderBridgeError> {
|
||||
self.prepare_turn()?;
|
||||
self.settled_call_ids.clear();
|
||||
self.settled_call_filter = DurableReplayFilter::default();
|
||||
self.durable_run_receipt_limit_reached = false;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn retained_result_bytes_for_test(&self) -> usize {
|
||||
self.retained_result_bytes
|
||||
|
|
@ -550,6 +564,10 @@ impl ProviderToolBridge {
|
|||
self.completed.contains_key(call_id) || self.has_settled_call_id(call_id)
|
||||
}
|
||||
|
||||
pub(crate) fn has_call_receipt(&self, call_id: &str) -> bool {
|
||||
self.pending.contains_key(call_id) || self.has_completed_call(call_id)
|
||||
}
|
||||
|
||||
pub fn begin_call(
|
||||
&mut self,
|
||||
call_id: String,
|
||||
|
|
@ -1464,6 +1482,37 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cross_bridge_receipt_lookup_reserves_pending_call_ids() {
|
||||
let operation = AuthorizedTool {
|
||||
operation_id: "get_task_context".to_owned(),
|
||||
version: 1,
|
||||
description: "Read the active task context.".to_owned(),
|
||||
input_schema: json!({"type": "object"}),
|
||||
response_schema: json!({"type": "object"}),
|
||||
};
|
||||
let mut bridge = ProviderToolBridge::default();
|
||||
bridge
|
||||
.prepare(AuthorizedToolSet {
|
||||
schema: TOOL_SET_SCHEMA.to_owned(),
|
||||
schema_version: 1,
|
||||
catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation))
|
||||
.unwrap(),
|
||||
operations: vec![operation],
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert!(!bridge.has_call_receipt("shared-call"));
|
||||
bridge
|
||||
.begin_call(
|
||||
"shared-call".to_owned(),
|
||||
"get_task_context".to_owned(),
|
||||
json!({}),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(bridge.has_call_receipt("shared-call"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exact_completed_results_are_bounded_without_changing_replay() {
|
||||
let operation = AuthorizedTool {
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ fn decodes_every_admitted_runtime_event_shape() {
|
|||
AcpxRuntimeEventKind::ToolCall,
|
||||
),
|
||||
(
|
||||
json!({"type": "semantic_result", "callId": "call-1", "operationId": "paperclip_finish", "result": {"ok": true}}),
|
||||
json!({"type": "semantic_result", "callId": "call-1", "operationId": "paperclip_finish", "ok": true, "result": {"ok": true}}),
|
||||
AcpxRuntimeEventKind::SemanticResult,
|
||||
),
|
||||
(
|
||||
|
|
@ -103,6 +103,7 @@ fn admits_sidecar_replacement_scalars_in_tool_fields() {
|
|||
kind: AcpxRuntimeEventKind::ToolCall,
|
||||
tool_operation: Some("edit"),
|
||||
payload,
|
||||
..
|
||||
} if payload["toolCallId"].as_str() == Some("tool-\u{fffd}")
|
||||
&& payload["kind"].as_str() == Some(kind.as_str())
|
||||
&& payload["status"].as_str() == Some("pend\u{fffd}ing")
|
||||
|
|
@ -135,6 +136,7 @@ fn retains_full_kind_classification_from_a_bounded_sidecar_frame() {
|
|||
kind: AcpxRuntimeEventKind::ToolCall,
|
||||
tool_operation: Some("edit"),
|
||||
payload,
|
||||
..
|
||||
} if payload["type"] == "tool_call"
|
||||
&& payload["kind"].as_str().is_some_and(|value| value.chars().count() == 4_000)
|
||||
));
|
||||
|
|
@ -186,7 +188,7 @@ fn decodes_tool_and_permission_requests_after_scope_validation() {
|
|||
.unwrap();
|
||||
assert!(matches!(
|
||||
tool,
|
||||
AcpxEventPayload::ToolCalled { call_id, operation_id, input }
|
||||
AcpxEventPayload::ToolCalled { call_id, operation_id, input, .. }
|
||||
if call_id == "call-1"
|
||||
&& operation_id == "get_issue"
|
||||
&& input["apiToken"] == "[REDACTED]"
|
||||
|
|
|
|||
|
|
@ -131,6 +131,43 @@ fn binds_and_clears_one_turn_idempotently() {
|
|||
scope.clear_turn("turn-1").unwrap();
|
||||
assert_eq!(scope.active_turn_id(), None);
|
||||
assert!(scope.clear_turn("turn-1").is_err());
|
||||
let reused = scope.bind_turn("turn-1").unwrap_err();
|
||||
assert!(reused.to_string().contains("reused a settled turn"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_late_events_after_rotating_to_a_distinct_turn_identity() {
|
||||
let mut scope = AcpxEventScope::new("run-1").unwrap();
|
||||
scope.bind_turn("turn-1").unwrap();
|
||||
scope.clear_turn("turn-1").unwrap();
|
||||
scope.bind_turn("turn-2").unwrap();
|
||||
|
||||
let late = scope
|
||||
.validate_event(&event(
|
||||
GeneratedAcpxSidecarEventType::RuntimeToolCalled,
|
||||
Some("run-1"),
|
||||
Some("turn-1"),
|
||||
))
|
||||
.unwrap_err();
|
||||
assert!(late.to_string().contains("stale turn"), "{late}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bounds_settled_turn_identity_retention() {
|
||||
let mut scope = AcpxEventScope::new("run-1").unwrap();
|
||||
for index in 0..4_096 {
|
||||
let turn_id = format!("turn-{index}");
|
||||
scope.bind_turn(&turn_id).unwrap();
|
||||
scope.clear_turn(&turn_id).unwrap();
|
||||
}
|
||||
|
||||
let exhausted = scope.bind_turn("turn-overflow").unwrap_err();
|
||||
assert!(
|
||||
exhausted
|
||||
.to_string()
|
||||
.contains("exhausted its settled turn identity capacity"),
|
||||
"{exhausted}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
|||
|
|
@ -95,6 +95,7 @@ fn preserves_tool_operation_authority_across_payload_sanitization() {
|
|||
kind,
|
||||
tool_operation,
|
||||
payload,
|
||||
..
|
||||
} = decoded
|
||||
else {
|
||||
panic!("runtime event must decode as runtime payload");
|
||||
|
|
|
|||
|
|
@ -167,6 +167,7 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() {
|
|||
"type":"semantic_result",
|
||||
"callId":"finish-1",
|
||||
"operationId":"paperclip_finish",
|
||||
"ok":true,
|
||||
"result":{"reportedWorkDisposition":"done"}
|
||||
}),
|
||||
);
|
||||
|
|
@ -189,6 +190,7 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() {
|
|||
"type":"semantic_result",
|
||||
"callId":"finish-2",
|
||||
"operationId":"paperclip_finish",
|
||||
"ok":true,
|
||||
"result":{"reportedWorkDisposition":"done"}
|
||||
}),
|
||||
);
|
||||
|
|
@ -196,6 +198,32 @@ fn accepts_an_identical_semantic_result_once_and_rejects_a_conflict() {
|
|||
assert_eq!(state.semantic_result().unwrap().call_id, "finish-1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keeps_dynamic_results_independent_from_terminal_result_authority() {
|
||||
let mut state = AcpxProviderState::new("run-1").unwrap();
|
||||
state.begin_turn("turn-1").unwrap();
|
||||
for index in 1..=2 {
|
||||
let result = event(
|
||||
index,
|
||||
GeneratedAcpxSidecarEventType::RuntimeEvent,
|
||||
Some("turn-1"),
|
||||
json!({
|
||||
"type":"semantic_result",
|
||||
"callId":format!("call-{index}"),
|
||||
"operationId":"issues.read",
|
||||
"ok":true,
|
||||
"result":{"id":format!("issue-{index}")}
|
||||
}),
|
||||
);
|
||||
assert!(matches!(
|
||||
&state.accept_event(&result).unwrap()[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == format!("call-{index}")
|
||||
));
|
||||
}
|
||||
assert!(state.semantic_result().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validates_scope_before_mutating_pending_state() {
|
||||
let mut state = AcpxProviderState::new("run-1").unwrap();
|
||||
|
|
|
|||
|
|
@ -0,0 +1,594 @@
|
|||
use std::path::PathBuf;
|
||||
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,
|
||||
};
|
||||
use serde_json::json;
|
||||
|
||||
fn tool_set() -> AuthorizedToolSet {
|
||||
let operations = vec![AuthorizedTool {
|
||||
operation_id: "issues.read".to_owned(),
|
||||
version: 1,
|
||||
description: "Read an issue.".to_owned(),
|
||||
input_schema: json!({"type":"object"}),
|
||||
response_schema: json!({"type":"object"}),
|
||||
}];
|
||||
AuthorizedToolSet {
|
||||
schema: "paperclip.runner.authorized-tools.v1".to_owned(),
|
||||
schema_version: 1,
|
||||
catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(),
|
||||
operations,
|
||||
}
|
||||
}
|
||||
|
||||
fn config(mode: &str) -> AcpxProviderSessionConfig {
|
||||
AcpxProviderSessionConfig {
|
||||
transport: AcpxSidecarTransportConfig {
|
||||
command: PathBuf::from(env!("CARGO_BIN_EXE_fake-acpx-sidecar")),
|
||||
args: vec!["--mode".to_owned(), mode.to_owned()],
|
||||
request_timeout: Duration::from_secs(1),
|
||||
shutdown_grace: Duration::from_millis(100),
|
||||
},
|
||||
agent: "codex".to_owned(),
|
||||
model: "gpt-5.6-sol".to_owned(),
|
||||
run_id: "run-1".to_owned(),
|
||||
catalog_revision: 1,
|
||||
runtime_directory: std::env::temp_dir(),
|
||||
normalized_session_id: "session-1".to_owned(),
|
||||
working_directory: std::env::temp_dir(),
|
||||
permission_mode: AcpxPermissionMode::ApproveReads,
|
||||
permission_mode_pinned: true,
|
||||
system_instructions: "Complete the supplied task.".to_owned(),
|
||||
tool_set: tool_set(),
|
||||
expected_identity: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn starts_interrupts_and_settles_one_scoped_turn() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns")).unwrap();
|
||||
assert!(session
|
||||
.poll_event(Duration::from_millis(1))
|
||||
.unwrap()
|
||||
.is_none());
|
||||
let response = session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
assert_eq!(response["turnId"], "turn-1");
|
||||
assert_eq!(session.state().active_turn_id(), Some("turn-1"));
|
||||
|
||||
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-1", "Paperclip interruption")
|
||||
.unwrap();
|
||||
assert_eq!(session.state().active_turn_id(), Some("turn-1"));
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
assert_eq!(session.state().active_turn_id(), None);
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_invalid_turn_inputs_without_mutating_the_session() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns")).unwrap();
|
||||
let other_directory = std::env::current_dir().unwrap();
|
||||
assert!(session
|
||||
.start_turn("turn-1", "Please help", &other_directory)
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("immutable session workspace"));
|
||||
assert!(session
|
||||
.start_turn("turn-1", "bad\0message", &std::env::temp_dir())
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("bounded contract"));
|
||||
assert_eq!(session.state().active_turn_id(), None);
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
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();
|
||||
let error = session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("confirm the requested turn"), "{error}");
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_when_cancellation_is_not_confirmed() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-wrong-cancel")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let error = session
|
||||
.interrupt_turn("turn-1", "stop")
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("confirm turn cancellation"), "{error}");
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_when_a_polled_event_violates_run_scope() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-wrong-scope")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("stale run"), "{error}");
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admits_only_catalog_authorized_tool_calls() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-tool")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&events[0],
|
||||
AcpxProviderStateEvent::ToolCall { operation_id, .. }
|
||||
if operation_id == "issues.read"
|
||||
));
|
||||
assert_eq!(
|
||||
session.state().pending_tool("call-1").unwrap().operation_id,
|
||||
"issues.read"
|
||||
);
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn returns_pending_tool_cancellations_before_the_terminal_event() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-tool-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. }));
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&terminal[0],
|
||||
AcpxProviderStateEvent::ToolResult(result)
|
||||
if result.call_id == "call-1"
|
||||
&& result.operation_id == "issues.read"
|
||||
&& result.is_error
|
||||
&& result.result["error"]["code"] == "acpx_turn_settled"
|
||||
));
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
|
||||
assert!(session.state().pending_tool("call-1").is_none());
|
||||
session
|
||||
.start_turn("turn-2", "Please continue", &std::env::temp_dir())
|
||||
.expect("terminal settlement must leave the session reusable");
|
||||
let next_tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&next_tool[0],
|
||||
AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. }
|
||||
if call_id == "call-2" && operation_id == "issues.read"
|
||||
));
|
||||
let next_terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
next_terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2"
|
||||
));
|
||||
assert!(session.state().pending_tool("call-2").is_none());
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rotates_settled_tool_receipts_between_reusable_turns() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-reused-tool-id-terminal")).unwrap();
|
||||
|
||||
for turn_id in ["turn-1", "turn-2"] {
|
||||
session
|
||||
.start_turn(turn_id, "Please continue", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&tool[0],
|
||||
AcpxProviderStateEvent::ToolCall { call_id, operation_id, .. }
|
||||
if call_id == "call-reused" && operation_id == "issues.read"
|
||||
));
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&terminal[0],
|
||||
AcpxProviderStateEvent::ToolResult(result)
|
||||
if result.call_id == "call-reused"
|
||||
&& result.result["error"]["code"] == "acpx_turn_settled"
|
||||
));
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. }
|
||||
if settled == turn_id
|
||||
));
|
||||
}
|
||||
session.shutdown("test complete").unwrap();
|
||||
|
||||
let mut reserved_session =
|
||||
AcpxProviderSession::start(&config("turns-reserved-result-terminal")).unwrap();
|
||||
for turn_id in ["turn-1", "turn-2"] {
|
||||
reserved_session
|
||||
.start_turn(turn_id, "Please continue", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
assert!(reserved_session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.is_empty());
|
||||
let result = reserved_session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
&result[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == "call-finish"
|
||||
&& result.operation_id == "paperclip_finish"
|
||||
));
|
||||
let terminal = reserved_session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. }
|
||||
if settled == turn_id
|
||||
));
|
||||
}
|
||||
reserved_session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reaps_the_old_provider_before_admitting_a_late_tool_callback() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-late-tool-after-suspend")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please continue", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
|
||||
session
|
||||
.start_turn("turn-2", "Use a fresh provider", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let fresh = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&fresh[0],
|
||||
AcpxProviderStateEvent::ToolCall { call_id, operation_id, input }
|
||||
if call_id == "call-reused"
|
||||
&& operation_id == "issues.read"
|
||||
&& input["id"] == "issue-1"
|
||||
));
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_before_reusing_a_settled_turn_identity() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-reused-tool-id-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please continue", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
|
||||
let reused = session
|
||||
.start_turn("turn-1", "Do not alias old events", &std::env::temp_dir())
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
reused.to_string().contains("reused a settled turn"),
|
||||
"{reused}"
|
||||
);
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_tool_results_are_not_cancelled_when_the_turn_terminates() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-tool-result-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. }));
|
||||
|
||||
let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&result[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == "call-1" && result.operation_id == "issues.read" && result.ok
|
||||
));
|
||||
assert!(session.state().pending_tool("call-1").is_none());
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert_eq!(terminal.len(), 1);
|
||||
assert!(matches!(
|
||||
&terminal[0],
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completes_multiple_distinct_dynamic_tool_results_in_one_turn() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-multiple-tool-results-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
|
||||
for index in 1..=2 {
|
||||
let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&tool[0],
|
||||
AcpxProviderStateEvent::ToolCall { call_id, .. }
|
||||
if call_id == &format!("call-{index}")
|
||||
));
|
||||
}
|
||||
for index in 1..=2 {
|
||||
let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&result[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == format!("call-{index}")
|
||||
&& result.result["id"] == format!("issue-{index}")
|
||||
));
|
||||
assert!(session
|
||||
.state()
|
||||
.pending_tool(&format!("call-{index}"))
|
||||
.is_none());
|
||||
}
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert_eq!(terminal.len(), 1);
|
||||
assert!(matches!(
|
||||
&terminal[0],
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_semantic_results_preserve_error_status_and_release_pending_capacity() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-tool-error-result-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let tool = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(tool[0], AcpxProviderStateEvent::ToolCall { .. }));
|
||||
|
||||
let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&result[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == "call-1"
|
||||
&& !result.ok
|
||||
&& result.result["error"]["code"] == "tool_failed"
|
||||
));
|
||||
assert!(session.state().pending_tool("call-1").is_none());
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reserved_terminal_results_require_an_authorized_correlated_invocation() {
|
||||
for (mode, operation_id, disposition) in [
|
||||
("turns-reserved-result-terminal", "paperclip_finish", "done"),
|
||||
(
|
||||
"turns-reserved-yielded-terminal",
|
||||
"paperclip_finish",
|
||||
"yielded",
|
||||
),
|
||||
(
|
||||
"turns-reserved-block-terminal",
|
||||
"paperclip_block",
|
||||
"blocked",
|
||||
),
|
||||
] {
|
||||
let mut session = AcpxProviderSession::start(&config(mode)).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
|
||||
let invocation = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(invocation.is_empty());
|
||||
assert_eq!(
|
||||
session
|
||||
.state()
|
||||
.pending_tool("call-finish")
|
||||
.unwrap()
|
||||
.operation_id,
|
||||
operation_id
|
||||
);
|
||||
|
||||
let result = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&result[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.call_id == "call-finish"
|
||||
&& result.operation_id == operation_id
|
||||
&& result.ok
|
||||
&& result.result["reportedWorkDisposition"] == disposition
|
||||
));
|
||||
assert!(session.state().pending_tool("call-finish").is_none());
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn correlates_reserved_results_by_raw_digest_without_exposing_sensitive_values() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-sensitive-reserved-result-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
|
||||
assert!(session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.is_empty());
|
||||
let result_events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
&result_events[0],
|
||||
AcpxProviderStateEvent::SemanticResult(result)
|
||||
if result.result["summary"]
|
||||
.as_str()
|
||||
.is_some_and(|summary| summary.contains("REDACTED")
|
||||
&& !summary.contains("matching-sensitive-value"))
|
||||
));
|
||||
|
||||
let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap();
|
||||
assert!(matches!(
|
||||
terminal.last().unwrap(),
|
||||
AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-1"
|
||||
));
|
||||
session.shutdown("test complete").unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_sensitive_reserved_results_that_only_match_after_redaction() {
|
||||
let mut session = AcpxProviderSession::start(&config(
|
||||
"turns-mismatched-sensitive-reserved-result-terminal",
|
||||
))
|
||||
.unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
assert!(session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.is_empty());
|
||||
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
error.contains("does not match its authorized invocation"),
|
||||
"{error}"
|
||||
);
|
||||
assert!(!error.contains("matching-sensitive-value"), "{error}");
|
||||
assert!(!error.contains("different-sensitive-value"), "{error}");
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_before_returning_an_uncorrelated_reserved_result() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-uncorrelated-reserved-result-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
error.contains("no authorized pending invocation"),
|
||||
"{error}"
|
||||
);
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_before_returning_a_mismatched_reserved_result() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-mismatched-reserved-result-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
assert!(session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.is_empty());
|
||||
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
error.contains("does not match its authorized invocation"),
|
||||
"{error}"
|
||||
);
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_before_returning_a_malformed_reserved_result() {
|
||||
let mut session =
|
||||
AcpxProviderSession::start(&config("turns-invalid-reserved-block-terminal")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
error.contains("failed the Paperclip result schema"),
|
||||
"{error}"
|
||||
);
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fails_closed_before_returning_an_unauthorized_tool_call() {
|
||||
let mut session = AcpxProviderSession::start(&config("turns-unauthorized-tool")).unwrap();
|
||||
session
|
||||
.start_turn("turn-1", "Please help", &std::env::temp_dir())
|
||||
.unwrap();
|
||||
let error = session
|
||||
.poll_event(Duration::from_secs(1))
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(error.contains("unauthorized tool issues.delete"), "{error}");
|
||||
assert!(session.state().pending_tool("call-1").is_none());
|
||||
assert!(session.shutdown("already closed").is_ok());
|
||||
}
|
||||
|
|
@ -451,25 +451,25 @@ async function pumpTurn(
|
|||
currentTurnId: string,
|
||||
runtimeTurn: AcpxRuntimeTurn,
|
||||
): Promise<void> {
|
||||
let terminal: Record<string, unknown>;
|
||||
try {
|
||||
for await (const event of runtimeTurn.events) {
|
||||
emit("runtime.event", sanitizeRuntimeEvent(event), currentTurnId);
|
||||
}
|
||||
const result = await runtimeTurn.result;
|
||||
emit("runtime.turn_terminal", boundedSidecarValue(result), currentTurnId);
|
||||
terminal = boundedSidecarValue(result);
|
||||
} catch (error) {
|
||||
emit(
|
||||
"runtime.turn_terminal",
|
||||
{
|
||||
status: "failed",
|
||||
error: { message: safeMessage(error), retryable: false },
|
||||
},
|
||||
currentTurnId,
|
||||
);
|
||||
terminal = {
|
||||
status: "failed",
|
||||
error: { message: safeMessage(error), retryable: false },
|
||||
};
|
||||
} finally {
|
||||
rejectTurnWaiters(currentTurnId, "ACPX turn became terminal");
|
||||
if (turnId === currentTurnId) turnId = null;
|
||||
}
|
||||
// A terminal frame is also runnerd's permission to recycle this provider.
|
||||
// Publish it only after no callback can inherit this turn's mutable binding.
|
||||
emit("runtime.turn_terminal", terminal, currentTurnId);
|
||||
}
|
||||
|
||||
async function waitForTool(call: RunnerToolCall): Promise<unknown> {
|
||||
|
|
@ -497,22 +497,43 @@ async function waitForTool(call: RunnerToolCall): Promise<unknown> {
|
|||
"ACPX semantic result disposition does not match its terminal operation",
|
||||
);
|
||||
}
|
||||
emit("runtime.event", {
|
||||
type: "semantic_result",
|
||||
callId,
|
||||
operationId,
|
||||
result: validation.result,
|
||||
});
|
||||
// The authenticated runner bridge admitted this built-in invocation. Send
|
||||
// that fact across the sidecar boundary before its locally produced result
|
||||
// so runnerd can authorize and correlate the terminal claim.
|
||||
emit(
|
||||
"runtime.tool_called",
|
||||
{
|
||||
callId,
|
||||
operationId,
|
||||
input: validation.result,
|
||||
},
|
||||
activeTurnId,
|
||||
);
|
||||
emit(
|
||||
"runtime.event",
|
||||
{
|
||||
type: "semantic_result",
|
||||
callId,
|
||||
operationId,
|
||||
ok: true,
|
||||
result: validation.result,
|
||||
},
|
||||
activeTurnId,
|
||||
);
|
||||
return { accepted: true };
|
||||
}
|
||||
if (tools.size >= MAX_PENDING_TOOLS) {
|
||||
throw new Error("ACPX pending tool limit reached");
|
||||
}
|
||||
emit("runtime.tool_called", {
|
||||
callId,
|
||||
operationId,
|
||||
input: boundedSidecarValue(record(call.arguments)),
|
||||
});
|
||||
emit(
|
||||
"runtime.tool_called",
|
||||
{
|
||||
callId,
|
||||
operationId,
|
||||
input: boundedSidecarValue(record(call.arguments)),
|
||||
},
|
||||
activeTurnId,
|
||||
);
|
||||
return await new Promise((settle, reject) => {
|
||||
const abort = () => {
|
||||
const pending = tools.get(callId);
|
||||
|
|
|
|||
Loading…
Reference in New Issue