From 7cf00e00725f518984c5b5876164ff76d24d3b3f Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 28 Sep 2026 11:42:58 -0500 Subject: [PATCH] Deliver negotiated ACP steering and queued follow-ups through durable runner controls Bind controls to the active turn and reserve bounded delivery identities before dispatch; preserve explicit provider policy and Pi receipt estimates across both runner paths. Co-Authored-By: Paperclip --- .../provider-schemas/acpx-sidecar.schema.json | 42 +++- .../runner-core/src/acpx_provider_backend.rs | 193 ++++++++++++++++-- .../runner-core/src/acpx_provider_session.rs | 57 +++++- .../runner-core/src/acpx_provider_state.rs | 27 ++- .../runner-core/src/bin/fake-acpx-sidecar.rs | 6 + .../src/generated_acpx_sidecar_contract.rs | 2 + .../runner-core/src/provider_backend.rs | 8 + .../tests/acpx_provider_checkpoint.rs | 1 + .../tests/acpx_provider_resolutions.rs | 1 + .../tests/acpx_provider_session.rs | 9 +- .../tests/acpx_provider_suspend.rs | 1 + .../runner-core/tests/acpx_provider_turns.rs | 51 +++++ .../src/cli/acpx-runtime-sidecar.ts | 39 +++- .../src/contracts/harness-driver.ts | 2 + .../drivers/acpx/codex-acpx-driver.test.ts | 44 ++++ .../src/drivers/acpx/codex-acpx-driver.ts | 68 +++++- .../acpx/generated-sidecar-contract.ts | 1 + .../src/drivers/acpx/sidecar-protocol.ts | 1 + .../src/drivers/acpx/turn-controls.test.ts | 26 +++ .../src/drivers/acpx/turn-controls.ts | 42 ++++ .../drivers/codex/codex-harness-session.ts | 2 + .../test/acpx-sidecar-contract.test.mjs | 11 + 22 files changed, 604 insertions(+), 30 deletions(-) create mode 100644 packages/paperclip-runner/src/drivers/acpx/turn-controls.test.ts create mode 100644 packages/paperclip-runner/src/drivers/acpx/turn-controls.ts diff --git a/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json b/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json index 5689b3642a4..1d285c52add 100644 --- a/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json +++ b/packages/paperclip-runner/protocol/provider-schemas/acpx-sidecar.schema.json @@ -16,6 +16,7 @@ "run.attach", "turn.start", "turn.cancel", + "turn.steer", "permission.resolve", "input.resolve", "tool.resolve", @@ -50,7 +51,11 @@ "command": { "$ref": "#/$defs/command" }, "params": { "type": "object" } }, - "additionalProperties": false + "additionalProperties": false, + "allOf": [{ + "if": { "properties": { "command": { "const": "turn.steer" } } }, + "then": { "properties": { "params": { "$ref": "#/$defs/turnControl" } } } + }] }, "response": { "type": "object", @@ -171,6 +176,41 @@ } }, "additionalProperties": false + }, + "turnControl": { + "type": "object", + "required": [ + "turnId", + "controlId", + "mode", + "message" + ], + "properties": { + "turnId": { + "type": "string", + "minLength": 1, + "maxLength": 240, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "controlId": { + "type": "string", + "minLength": 1, + "maxLength": 160, + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]*$" + }, + "mode": { + "enum": [ + "steer", + "follow_up" + ] + }, + "message": { + "type": "string", + "minLength": 1, + "maxLength": 65536 + } + }, + "additionalProperties": false } } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index b333de86f81..65d56263d7f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -10,11 +10,11 @@ use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; -#[cfg(test)] use sha2::{Digest, Sha256}; use crate::acpx_provider_session::{ - AcpxPermissionMode, AcpxProviderSession, AcpxProviderSessionConfig, AcpxProviderSessionIdentity, + AcpxPermissionMode, AcpxProviderRuntimePolicy, AcpxProviderSession, AcpxProviderSessionConfig, + AcpxProviderSessionIdentity, }; use crate::acpx_sidecar_transport::AcpxSidecarTransportConfig; #[cfg(test)] @@ -137,6 +137,8 @@ struct AcpxProviderDescriptor { permission_mode: AcpxPermissionMode, permission_mode_pinned: bool, #[serde(default)] + provider_policy: Option, + #[serde(default)] runtime_context: Value, } @@ -162,12 +164,33 @@ impl AcpxProviderDescriptor { Some("0.153.4"), "sha256:c4538599d1ab767db5dff50934f13bb5ba313a59d9c4a83e993fac4617ea63d3", ), - "pi" => return Err(DurableRunnerError::invalid( - "ACPX agent pi is not executable through the verified runnerd provider boundary", - )), + "pi" => ( + "openrouter/deepseek/deepseek-v4-flash-0731", + "pi-acp", + "0.0.33", + Some("@earendil-works/pi-coding-agent"), + Some("0.84.2"), + "sha256:8c696f38296d53d0061fa11534570c5ddd951b63532aed30e0f1fcc676dc169f", + ), + "cursor" => ( + self.model.as_str(), + "cursor-agent", + "2026.09.26-dd393fe", + None, + None, + "sha256:1157a5d071abbd57ab132f22bace75c65e84cc47a045b0023475488755e14899", + ), + "copilot" => ( + self.model.as_str(), + "@github/copilot", + "1.0.88", + None, + None, + "sha256:b18c01603dd0169d233140709cfaa8bf5304a03cf5de78ca4f625f30013e8457", + ), _ => { return Err(DurableRunnerError::invalid( - "ACPX agent must be a qualified claude or codex profile", + "ACPX agent must name a known immutable profile", )) } }; @@ -178,7 +201,10 @@ impl AcpxProviderDescriptor { || self.acpx_version != "0.13.1" || (self.agent != "claude" && self.model != expected.0) || self.model.trim().is_empty() - || self.model.len() > 1024 + || self.model.len() > 240 + || self.model.contains('\0') + || (matches!(self.agent.as_str(), "pi" | "cursor" | "copilot") + && self.provider_policy.is_none()) || self.agent_server_package != expected.1 || self.agent_server_version != expected.2 || self.agent_runtime_package.as_deref() != expected.3 @@ -273,6 +299,7 @@ impl AcpxProviderDescriptor { working_directory: PathBuf::from(&self.cwd), permission_mode: self.permission_mode, permission_mode_pinned: self.permission_mode_pinned, + provider_policy: self.provider_policy.clone(), system_instructions: self.instructions.clone(), runtime_context: self.runtime_context.clone(), tool_set: provider_tool_set, @@ -387,6 +414,8 @@ struct AcpxDurableState { #[serde(default)] active_turn_id: Option, #[serde(default)] + attempted_turn_controls: HashSet, + #[serde(default)] provider_exit_unconfirmed: bool, #[serde(default)] semantic_result: Option, @@ -416,6 +445,7 @@ impl AcpxDurableState { tool_set, identity: None, active_turn_id: None, + attempted_turn_controls: HashSet::new(), provider_exit_unconfirmed: false, semantic_result: None, goal_projection: Value::Null, @@ -456,6 +486,11 @@ impl AcpxDurableState { | "closed" ) || self.next_event_sequence == 0 + || self.attempted_turn_controls.len() > 1024 + || self + .attempted_turn_controls + .iter() + .any(|id| !is_stable_id(id, 160)) || self.pending_events.len() > MAX_PENDING_EVENTS || self.pending_events.iter().any(|event| { event_sequence(&event.executor_event_id) @@ -1032,6 +1067,7 @@ impl AcpxCommandExecutor { )); } state.active_turn_id = Some(provider_turn_id.clone()); + state.attempted_turn_controls.clear(); state.semantic_result = None; state.lifecycle = "turn_starting".to_owned(); } @@ -1106,6 +1142,86 @@ impl AcpxCommandExecutor { }) } + fn steer_turn( + &mut self, + command_id: &str, + payload: &Value, + ) -> Result { + let text = payload + .get("text") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("turn.steer payload.text is required"))?; + let turn_id = payload + .get("turnId") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("turn.steer payload.turnId is required"))?; + let mode = payload + .get("mode") + .and_then(Value::as_str) + .unwrap_or("steer"); + if !matches!(mode, "steer" | "follow_up") + || text.trim().is_empty() + || text.len() > 65_536 + || text.contains('\0') + { + return Err(DurableRunnerError::invalid( + "turn.steer violates its bounded contract", + )); + } + let control_id = match payload.get("correlationId") { + Some(Value::String(id)) if is_stable_id(id, 160) => id.clone(), + None => format!("control-{:x}", Sha256::digest(command_id.as_bytes())), + _ => { + return Err(DurableRunnerError::invalid( + "turn.steer correlationId is invalid", + )) + } + }; + let state = self + .state + .as_mut() + .ok_or_else(|| DurableRunnerError::invalid("ACPX provider is not prepared"))?; + if state.lifecycle != "turn_active" + || state.active_turn_id.as_deref() != Some(turn_id) + || self.session.is_none() + { + return Err(DurableRunnerError::invalid( + "turn.steer named a stale or inactive turn", + )); + } + if state.attempted_turn_controls.contains(&control_id) + || state.attempted_turn_controls.len() >= 1024 + { + return Err(DurableRunnerError::invalid( + "turn.steer is duplicate or exceeds the turn control limit", + )); + } + state.attempted_turn_controls.insert(control_id.clone()); + // Retain before the potentially mutating RPC. Controller reconnects may + // read its acknowledgement but must not replay an ambiguous attempt. + self.save_state()?; + self.session + .as_mut() + .expect("checked live session") + .steer_turn(turn_id, &control_id, mode, text) + .map_err(|error| { + DurableRunnerError::invalid(format!("ACPX turn control failed: {error}")) + })?; + Ok(CommandExecution { + result: json!({"status": "accepted", "mode": mode, "providerTurnId": turn_id, "correlationId": control_id}), + events: vec![( + "item.completed".to_owned(), + EventPriority::P0, + json!({ + "provider": "acpx", "providerTurnId": turn_id, + "itemId": format!("acpx-control-{:x}", Sha256::digest(format!("{turn_id}:{control_id}").as_bytes())), + "kind": "steering_acknowledgement", "mode": mode, "status": "acknowledged", + "text": if mode == "steer" { "Steering acknowledged for the active turn." } else { "Follow-up queued after the active work." }, + }), + )], + }) + } + fn interrupt_turn(&mut self, reason: &str) -> Result { let turn_id = self .state @@ -1535,11 +1651,7 @@ impl CommandExecutor for AcpxCommandExecutor { self.goal_control(&command.command_type, &command.payload) } "turn.start" => self.start_turn(&command.payload), - "turn.steer" => Ok(CommandExecution::result(json!({ - "status": "rejected", - "code": "provider_command_unavailable", - "message": "ACPX does not support steering an active turn", - }))), + "turn.steer" => self.steer_turn(&command.command_id, &command.payload), "turn.interrupt" | "run.cancel" => self.interrupt_turn(&command.command_type), "turn.stop" => self.stop_turn_for_suspension(&command.command_type), "request.resolve" => self.resolve_request(&command.payload), @@ -1990,7 +2102,62 @@ mod tests { } #[test] - fn rejects_pi_before_process_launch() { + fn admits_candidate_descriptors_only_with_explicit_policy_and_exact_distribution() { + for (agent, package, version, digest, runtime_package, runtime_version, model) in [ + ( + "cursor", + "cursor-agent", + "2026.09.26-dd393fe", + "sha256:1157a5d071abbd57ab132f22bace75c65e84cc47a045b0023475488755e14899", + None, + None, + "explicit-model", + ), + ( + "copilot", + "@github/copilot", + "1.0.88", + "sha256:b18c01603dd0169d233140709cfaa8bf5304a03cf5de78ca4f625f30013e8457", + None, + None, + "explicit-model", + ), + ( + "pi", + "pi-acp", + "0.0.33", + "sha256:8c696f38296d53d0061fa11534570c5ddd951b63532aed30e0f1fcc676dc169f", + Some("@earendil-works/pi-coding-agent"), + Some("0.84.2"), + "openrouter/deepseek/deepseek-v4-flash-0731", + ), + ] { + let mut value = descriptor("codex"); + value["agent"] = json!(agent); + value["model"] = json!(model); + value["agentServerPackage"] = json!(package); + value["agentServerVersion"] = json!(version); + value["agentRuntimePackage"] = json!(runtime_package); + value["agentRuntimeVersion"] = json!(runtime_version); + value["commandDigest"] = json!(digest); + let missing: AcpxProviderDescriptor = serde_json::from_value(value.clone()).unwrap(); + assert!(missing.validate(&context()).is_err()); + value["providerPolicy"] = json!({"readOnly":true}); + let valid: AcpxProviderDescriptor = serde_json::from_value(value.clone()).unwrap(); + valid.validate(&context()).unwrap(); + for field in ["model", "agentServerVersion", "commandDigest"] { + let mut wrong = value.clone(); + wrong[field] = json!(""); + let invalid: AcpxProviderDescriptor = serde_json::from_value(wrong).unwrap(); + assert!(invalid.validate(&context()).is_err()); + } + value["providerPolicy"] = json!({"readOnly":true, "protectedPaths":[]}); + assert!(serde_json::from_value::(value).is_err()); + } + } + + #[test] + fn rejects_pi_with_another_profile_identity_before_process_launch() { let mut pi = descriptor("codex"); pi["agent"] = json!("pi"); let pi: AcpxProviderDescriptor = serde_json::from_value(pi).unwrap(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index 16a8b3e63ec..44dde107761 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -51,6 +51,12 @@ pub struct AcpxProviderSessionIdentity { pub provider_lifetime_fence_candidates: [u16; 3], } +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct AcpxProviderRuntimePolicy { + pub read_only: bool, +} + #[derive(Clone, Debug)] pub struct AcpxProviderSessionConfig { pub transport: AcpxSidecarTransportConfig, @@ -63,6 +69,7 @@ pub struct AcpxProviderSessionConfig { pub working_directory: PathBuf, pub permission_mode: AcpxPermissionMode, pub permission_mode_pinned: bool, + pub provider_policy: Option, pub system_instructions: String, pub runtime_context: Value, pub tool_set: AuthorizedToolSet, @@ -75,9 +82,11 @@ impl AcpxProviderSessionConfig { let qualified_model = match self.agent.as_str() { "claude" => "claude-sonnet-5", "codex" => "gpt-5.6-sol", + "pi" => "openrouter/deepseek/deepseek-v4-flash-0731", + "cursor" | "copilot" => self.model.as_str(), _ => { return Err(LocalRunnerError::invalid( - "ACPX agent must be claude or codex", + "ACPX agent must name a known immutable profile", )) } }; @@ -88,6 +97,13 @@ impl AcpxProviderSessionConfig { ))); } validate_text(&self.model, MAX_MODEL_CHARS, "ACPX model")?; + if matches!(self.agent.as_str(), "pi" | "cursor" | "copilot") + && self.provider_policy.is_none() + { + return Err(LocalRunnerError::invalid( + "ACPX candidate requires explicit provider read-only policy", + )); + } validate_stable_id(&self.run_id, SHORT_STABLE_ID_CHARS, "ACPX run id")?; validate_stable_id( &self.normalized_session_id, @@ -385,6 +401,44 @@ impl AcpxProviderSession { Ok(response) } + pub fn steer_turn( + &mut self, + turn_id: &str, + control_id: &str, + mode: &str, + message: &str, + ) -> Result { + self.ensure_open()?; + validate_stable_id(turn_id, DURABLE_STABLE_ID_CHARS, "ACPX turn id")?; + validate_stable_id(control_id, SHORT_STABLE_ID_CHARS, "ACPX control id")?; + if !matches!(mode, "steer" | "follow_up") + || message.trim().is_empty() + || message.len() > 65_536 + || message.contains('\0') + { + return Err(LocalRunnerError::invalid( + "ACPX turn control violates its bounded contract", + )); + } + self.state.reserve_turn_control(turn_id, control_id)?; + // The sidecar checks the negotiated live capability. An error may follow + // delivery, so the reservation must never be released for automatic retry. + let response = self.transport.request( + GeneratedAcpxSidecarCommand::TurnSteer, + json!({"turnId": turn_id, "controlId": control_id, "mode": mode, "message": message}), + )?; + if response.get("accepted").and_then(Value::as_bool) != Some(true) + || response.get("turnId").and_then(Value::as_str) != Some(turn_id) + || response.get("controlId").and_then(Value::as_str) != Some(control_id) + || response.get("mode").and_then(Value::as_str) != Some(mode) + { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX sidecar did not acknowledge the exact turn control", + ))); + } + Ok(response) + } + pub fn interrupt_turn( &mut self, turn_id: &str, @@ -1107,6 +1161,7 @@ fn bootstrap( "model": config.model, "permissionMode": config.permission_mode, "permissionModePinned": config.permission_mode_pinned, + "providerPolicy": config.provider_policy, "systemInstructions": config.system_instructions, "runtimeContext": config.runtime_context, "tools": &sidecar_tools, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs index 8aa4d5b5b2b..01d4c2294c6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use serde_json::Value; @@ -105,6 +105,7 @@ pub struct AcpxProviderState { pending_inputs: BTreeMap, pending_runtime_request_bytes: usize, semantic_result: Option, + attempted_turn_controls: BTreeSet, } impl AcpxProviderState { @@ -121,6 +122,7 @@ impl AcpxProviderState { pending_inputs: BTreeMap::new(), pending_runtime_request_bytes: 0, semantic_result: None, + attempted_turn_controls: BTreeSet::new(), }) } @@ -165,6 +167,28 @@ impl AcpxProviderState { || !self.pending_inputs.is_empty() } + pub(crate) fn reserve_turn_control( + &mut self, + turn_id: &str, + control_id: &str, + ) -> Result<(), LocalRunnerError> { + if self.active_turn_id() != Some(turn_id) { + return Err(LocalRunnerError::invalid( + "ACPX turn control named a stale or inactive turn", + )); + } + if self.attempted_turn_controls.contains(control_id) { + return Err(LocalRunnerError::invalid( + "ACPX turn control was already attempted", + )); + } + if self.attempted_turn_controls.len() >= 1024 { + return Err(LocalRunnerError::invalid("ACPX turn control limit reached")); + } + self.attempted_turn_controls.insert(control_id.to_owned()); + Ok(()) + } + pub fn begin_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { if self.scope.active_turn_id().is_some() || !self.pending_tools.is_empty() @@ -185,6 +209,7 @@ impl AcpxProviderState { self.assistant_text.clear(); self.assistant_message_id = None; self.semantic_result = None; + self.attempted_turn_controls.clear(); Ok(()) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs index 94fac4d48ed..593089a6e01 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs @@ -204,6 +204,8 @@ fn run() -> Result<(), Box> { | "goals" | "bootstrap-wrong-model" | "bootstrap-wrong-run" + | "controls" + | "controls-wrong-ack" | "turns" | "turns-wrong-turn" | "turns-wrong-cancel" @@ -737,6 +739,10 @@ fn bootstrap_success( "turn.start" => json!({ "turnId": if mode == "turns-wrong-turn" { "wrong-turn" } else { params.get("turnId").and_then(Value::as_str).unwrap_or("missing") }, }), + "turn.steer" => json!({ + "accepted": true, "turnId": params.get("turnId"), "controlId": params.get("controlId"), + "mode": if mode == "controls-wrong-ack" { json!("wrong") } else { params["mode"].clone() }, + }), "turn.cancel" => json!({"cancelled":mode != "turns-wrong-cancel"}), "session.suspend" => json!({ "suspended":mode != "suspend-wrong-ack", diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/generated_acpx_sidecar_contract.rs b/packages/paperclip-runner/runner/crates/runner-core/src/generated_acpx_sidecar_contract.rs index 8fa7cb526de..367a3bb54db 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/generated_acpx_sidecar_contract.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/generated_acpx_sidecar_contract.rs @@ -32,6 +32,7 @@ pub enum GeneratedAcpxSidecarCommand { RunAttach, TurnStart, TurnCancel, + TurnSteer, PermissionResolve, InputResolve, ToolResolve, @@ -52,6 +53,7 @@ impl GeneratedAcpxSidecarCommand { Self::RunAttach => "run.attach", Self::TurnStart => "turn.start", Self::TurnCancel => "turn.cancel", + Self::TurnSteer => "turn.steer", Self::PermissionResolve => "permission.resolve", Self::InputResolve => "input.resolve", Self::ToolResolve => "tool.resolve", diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index cf7f5468dbf..5d5d5415da4 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -3185,6 +3185,14 @@ impl CodexCommandExecutor { } fn steer_turn(&mut self, payload: &Value) -> Result { + if payload + .get("mode") + .is_some_and(|mode| mode.as_str() != Some("steer")) + { + return Err(DurableRunnerError::invalid( + "Codex does not expose queued follow-up through turn.steer", + )); + } let text = payload .get("text") .and_then(Value::as_str) diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs index 2ee2e712dcb..f00049620dc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_checkpoint.rs @@ -51,6 +51,7 @@ fn config(directory: &std::path::Path) -> AcpxProviderSessionConfig { working_directory: directory.to_owned(), permission_mode: AcpxPermissionMode::ApproveReads, permission_mode_pinned: true, + provider_policy: None, system_instructions: "Complete the supplied task.".to_owned(), runtime_context: serde_json::Value::Null, tool_set: AuthorizedToolSet { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs index 2352a0000f2..89e4cdee123 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_resolutions.rs @@ -50,6 +50,7 @@ fn config(mode: &str) -> AcpxProviderSessionConfig { working_directory: std::env::temp_dir(), permission_mode: AcpxPermissionMode::ApproveReads, permission_mode_pinned: true, + provider_policy: None, system_instructions: "Complete the supplied task.".to_owned(), runtime_context: serde_json::Value::Null, tool_set: tool_set(), diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs index b91d49ba21b..951a5697091 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_session.rs @@ -45,6 +45,7 @@ fn config(mode: &str) -> AcpxProviderSessionConfig { working_directory: std::env::temp_dir(), permission_mode: AcpxPermissionMode::ApproveReads, permission_mode_pinned: true, + provider_policy: None, system_instructions: "Complete the supplied task.".to_owned(), runtime_context: serde_json::Value::Null, tool_set: tool_set(), @@ -130,8 +131,12 @@ fn bootstraps_a_codex_session_and_confirms_run_identity() { #[test] fn validates_qualified_policy_and_tool_catalog_before_spawning() { let mut invalid_agent = config("bootstrap"); - invalid_agent.agent = "pi".to_owned(); - assert!(start_error(&invalid_agent).contains("claude or codex")); + invalid_agent.agent = "unknown".to_owned(); + assert!(start_error(&invalid_agent).contains("known immutable profile")); + + let mut missing_policy = config("bootstrap"); + missing_policy.agent = "copilot".to_owned(); + assert!(start_error(&missing_policy).contains("explicit provider read-only policy")); let mut unpinned = config("bootstrap"); unpinned.permission_mode_pinned = false; diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_suspend.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_suspend.rs index 92eb28ab277..eebd2e237fa 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_suspend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_suspend.rs @@ -26,6 +26,7 @@ fn config(mode: &str) -> AcpxProviderSessionConfig { working_directory: std::env::temp_dir(), permission_mode: AcpxPermissionMode::ApproveReads, permission_mode_pinned: true, + provider_policy: None, system_instructions: "Complete the supplied task.".to_owned(), runtime_context: serde_json::Value::Null, tool_set: AuthorizedToolSet { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs index 1436a25cd82..25bd85dc197 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -45,6 +45,7 @@ fn config(mode: &str) -> AcpxProviderSessionConfig { working_directory: std::env::temp_dir(), permission_mode: AcpxPermissionMode::ApproveReads, permission_mode_pinned: true, + provider_policy: None, system_instructions: "Complete the supplied task.".to_owned(), runtime_context: serde_json::Value::Null, tool_set: tool_set(), @@ -693,3 +694,53 @@ fn fails_closed_before_returning_an_unauthorized_tool_call() { assert!(session.state().pending_tool("call-1").is_none()); assert!(session.shutdown("already closed").is_ok()); } + +#[test] +fn turn_controls_preserve_mode_and_reject_stale_duplicate_and_oversized_delivery() { + let mut session = AcpxProviderSession::start(&config("controls")).unwrap(); + assert!(session + .steer_turn("turn-1", "control-1", "steer", "message") + .is_err()); + session + .start_turn("turn-1", "Work", &std::env::temp_dir()) + .unwrap(); + let first = session + .steer_turn("turn-1", "control-1", "steer", "Change focus") + .unwrap(); + assert_eq!(first["mode"], "steer"); + assert!(session + .steer_turn("turn-1", "control-1", "follow_up", "Duplicate") + .unwrap_err() + .to_string() + .contains("already attempted")); + let queued = session + .steer_turn("turn-1", "control-2", "follow_up", "Then validate") + .unwrap(); + assert_eq!(queued["mode"], "follow_up"); + assert!(session + .steer_turn("turn-2", "control-3", "steer", "Stale") + .is_err()); + assert!(session + .steer_turn("turn-1", "control-3", "cancel", "Wrong") + .is_err()); + assert!(session + .steer_turn("turn-1", "control-3", "steer", &"a".repeat(65_537)) + .is_err()); + session.shutdown("verified controls").unwrap(); +} + +#[test] +fn turn_controls_fail_closed_on_mismatched_acknowledgement() { + let mut session = AcpxProviderSession::start(&config("controls-wrong-ack")).unwrap(); + session + .start_turn("turn-1", "Work", &std::env::temp_dir()) + .unwrap(); + assert!(session + .steer_turn("turn-1", "control-1", "steer", "Change focus") + .unwrap_err() + .to_string() + .contains("exact turn control")); + assert!(session + .steer_turn("turn-1", "control-2", "follow_up", "Do not replay") + .is_err()); +} diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index 7c38d139938..9375efcd822 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -11,7 +11,7 @@ import type { AcpPermissionDecision, } from "acpx/runtime"; -import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "../drivers/acpx/profile-extensions.js"; +import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "../drivers/acpx/profile-extensions.js"; import type { PaperclipQuestionSet } from "../contracts/question-set.js"; import { createAcpxToolEventNormalizer } from "../provider-events.js"; import { parseNativeRuntimeContext } from "../contracts/runtime-context.js"; @@ -61,6 +61,7 @@ import { import { safeAcpxLocations } from "./acpx-sidecar-locations.js"; import { persistedAcpxTurnUsage, + acpxUsageEstimateNotice, qualifiedAcpxUsageBreakdown, } from "../drivers/acpx/usage-accounting.js"; import { validatePrpStructuredRunResult } from "../protocol/replay-contract.js"; @@ -85,6 +86,9 @@ import { verifyOpenedAcpxSidecarHost, } from "./acpx-sidecar-lifecycle.js"; +import { AcpxTurnControlLedger, parseAcpxTurnControl } from "../drivers/acpx/turn-controls.js"; + +const turnControls = new AcpxTurnControlLedger(); const MAX_PENDING_TOOLS = 512; const MAX_PENDING_INPUTS = 16; let goalSourceRevision = 0; @@ -277,6 +281,7 @@ async function dispatch( clientCapabilities: acpxProfileClientCapabilities(params.agent), model: params.model, permissionMode: params.permissionMode, + providerPolicy: params.providerPolicy, systemInstructions: params.systemInstructions, runtimeContext: params.runtimeContext, environment: process.env, @@ -345,13 +350,14 @@ async function dispatch( if (turnId) throw new Error("ACPX sidecar already has an active turn"); const currentTurnId = boundedIdentity(request.params.turnId, "turnId"); turnId = currentTurnId; + turnControls.begin(currentTurnId); let runtimeTurn: AcpxRuntimeTurn; const extensions = bindAcpxExtensionTurn({ adapter: createAcpxProfileExtensionAdapter(openParams!.agent, { - workspacePath: openParams!.workingDirectory, sessionId: activeHost.identity().agentSessionId, turnId: currentTurnId, + workspacePath: openParams!.workingDirectory, sessionId: activeHost.identity().backendSessionId, turnId: currentTurnId, }), active: () => turnId === currentTurnId && host === activeHost, - sessionId: activeHost.identity().agentSessionId, + sessionId: activeHost.identity().backendSessionId, waitForInput: (input, context) => waitForExtensionInput(currentTurnId, input, context), emit: event => emit("runtime.rich_event", { ...event }, currentTurnId), }); @@ -375,6 +381,21 @@ async function dispatch( void pumpTurn(currentTurnId, runtimeTurn, activeHost, usageBefore, extensions.drain); return { turnId: currentTurnId }; } + if (request.command === "turn.steer") { + const control = parseAcpxTurnControl(request.params); + const activeHost = requireHost(); + if (!runId || control.turnId !== turnId) throw new Error("cannot control a stale ACPX turn"); + const capability = activeHost.steeringCapability(); + if (!(control.mode === "steer" ? capability?.steering : capability?.queuedFollowUp)) { + throw new Error("ACP provider did not negotiate this turn control"); + } + turnControls.reserve(control, turnId); + const runtimeRequestId = `${runId}:${control.turnId}`; + if (control.mode === "steer") await activeHost.steerActiveTurn(control.message, runtimeRequestId); + else await activeHost.queueFollowUp(control.message, runtimeRequestId); + if (host !== activeHost || turnId !== control.turnId) throw new Error("ACP turn settled before control acknowledgement"); + return { accepted: true, turnId: control.turnId, controlId: control.controlId, mode: control.mode }; + } if (request.command === "turn.cancel") { const expected = boundedIdentity(request.params.turnId, "turnId"); if (expected !== turnId) throw new Error("cannot cancel a stale ACPX turn"); @@ -585,8 +606,11 @@ async function pumpTurn( usageBefore, await readSidecarHostStatusWithin(activeHost), runtimeTurn.requestId, + openParams?.agent, ); if (usage) { + const estimate = acpxUsageEstimateNotice(usage, `${currentTurnId}:usage-estimate`); + if (estimate) { validateAcpxRichEvent(estimate); emit("runtime.rich_event", { ...estimate }, currentTurnId); } emit( "runtime.event", sanitizeRuntimeEvent(usage as unknown as AcpRuntimeEvent), @@ -1118,6 +1142,7 @@ function parseOpenParams( model, permissionMode: requiredPermissionMode(value.permissionMode), permissionModePinned: value.permissionModePinned === true, + ...(value.providerPolicy == null ? {} : { providerPolicy: parseProviderPolicy(value.providerPolicy) }), systemInstructions: boundedText( value.systemInstructions, "systemInstructions", @@ -1133,6 +1158,14 @@ function parseOpenParams( }; } +function parseProviderPolicy(value: unknown): { readOnly: boolean } { + const policy = record(value); + if (typeof policy.readOnly !== "boolean" || Object.keys(policy).some(key => key !== "readOnly")) { + throw new Error("providerPolicy requires only an explicit readOnly boolean"); + } + return { readOnly: policy.readOnly }; +} + function parseTools(value: unknown): Readonly>[] { if (!Array.isArray(value) || value.length > 512) { throw new Error("tools must be an array with at most 512 entries"); diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 3400b419c9c..0e591de7173 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -512,6 +512,8 @@ export interface HarnessSession { effectiveCollaborationMode?: "default" | "plan"; }>; steer?(input: { + /** Queued follow-ups remain distinct from active-turn steering. */ + mode?: "steer" | "follow_up"; turnId: string; message: NativeUserMessage; correlationId?: string; diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts index 5d81f604869..0fb624b3ae7 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts @@ -14,6 +14,7 @@ import { type CodexAcpxDriverOptions, } from "./codex-acpx-driver.js"; import type { + AcpxRuntimeSteeringCapability, AcpxRuntimeTurnInput, AcpxRuntimeTurn, OpenAcpxRuntimeHostOptions, @@ -21,6 +22,46 @@ import type { import type { AcpxRecoveryWorkspaceLease } from "./runtime-sandbox.js"; describe("Codex ACPX harness driver", () => { + it("delivers negotiated steering and follow-up distinctly, once, for the active turn", async () => { + const fixture = driverFixture({ providerPolicy: { readOnly: true } }); + fixture.host.steeringCapability.mockReturnValue({ steering: true, queuedFollowUp: true }); + const session = await fixture.driver.openSession({ runId: "run-controls", normalizedSessionId: "session-1", workingDirectory: "/workspace" }); + expect(fixture.hostOptions?.providerPolicy).toEqual({ readOnly: true }); + const { turnId } = await session.startTurn({ message: { text: "Work" } }); + const turnInput = fixture.host.startTurn.mock.calls[0]![0]; + const context = { requestId: 0, signal: new AbortController().signal }; + await expect(turnInput.onExtensionRequest!("provider/request", { sessionId: "agent-1" }, context)).rejects.toThrow(/session mismatch/); + await expect(turnInput.onExtensionRequest!("provider/request", { sessionId: "backend-1" }, context)).rejects.toThrow(/adapter is unavailable/); + await session.steer!({ turnId, correlationId: "control-1", message: { text: "Change focus" } }); + await session.steer!({ turnId, correlationId: "control-2", mode: "follow_up", message: { text: "Then validate" } }); + expect(fixture.host.steerActiveTurn).toHaveBeenCalledExactlyOnceWith("Change focus", `run-controls:${turnId}`); + expect(fixture.host.queueFollowUp).toHaveBeenCalledExactlyOnceWith("Then validate", `run-controls:${turnId}`); + expect(fixture.host.interruptActiveTurn).not.toHaveBeenCalled(); + await expect(session.steer!({ turnId, correlationId: "control-2", message: { text: "Duplicate" } })).rejects.toThrow(/already attempted/); + await expect(session.steer!({ turnId: "wrong-turn", message: { text: "Stale" } })).rejects.toThrow(); + fixture.finishTurn({ status: "completed" }); + const events = await collectUntil(session.events(), "turn.completed"); + expect(events.filter(event => event.payload.kind === "steering_acknowledgement").map(event => event.payload.mode)).toEqual(["steer", "follow_up"]); + for (const event of events) expect(validatePrpEvent(event).ok).toBe(true); + await expect(session.steer!({ turnId, message: { text: "Already settled" } })).rejects.toThrow(); + await session.close({ reason: "controls verified" }); + }); + + it("rejects unnegotiated controls and retains ambiguous delivery attempts", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ runId: "run-controls", normalizedSessionId: "session-1", workingDirectory: "/workspace" }); + const { turnId } = await session.startTurn({ message: { text: "Work" } }); + await expect(session.steer!({ turnId, message: { text: "No handshake" } })).rejects.toThrow(/did not negotiate/); + expect(fixture.host.steerActiveTurn).not.toHaveBeenCalled(); + fixture.host.steeringCapability.mockReturnValue({ steering: true, queuedFollowUp: false }); + fixture.host.steerActiveTurn.mockRejectedValueOnce(new Error("delivery acknowledgement lost")); + const input = { turnId, correlationId: "ambiguous", message: { text: "Maybe delivered" } }; + await expect(session.steer!(input)).rejects.toThrow(/acknowledgement lost/); + await expect(session.steer!(input)).rejects.toThrow(/already attempted/); + expect(fixture.host.steerActiveTurn).toHaveBeenCalledOnce(); + await session.close({ reason: "ambiguous control verified" }); + }); + it.each([ ["claude", "claude-sonnet-5"], ["codex", "gpt-5.6-sol"], ] as const)("launches %s in full auto when no mode is supplied", async (agent, model) => { @@ -2758,6 +2799,9 @@ function fakeHost(createTurn: () => AcpxRuntimeTurn, onClose: () => void) { }, })), startTurn: vi.fn((_input: AcpxRuntimeTurnInput) => createTurn()), + steeringCapability: vi.fn<() => AcpxRuntimeSteeringCapability | null>(() => null), + steerActiveTurn: vi.fn(async (_text: string, _requestId?: string) => undefined), + queueFollowUp: vi.fn(async (_text: string, _requestId?: string) => undefined), interruptActiveTurn: vi.fn(async () => undefined), close: vi.fn(async () => { onClose(); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts index 160e2352570..f09e7243114 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts @@ -1,4 +1,4 @@ -import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "./profile-extensions.js"; +import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "./profile-extensions.js"; import { createHash, randomBytes } from "node:crypto"; import type { @@ -79,6 +79,9 @@ import { type AcpxRecoveryWorkspaceLease, } from "./runtime-sandbox.js"; +import { AcpxTurnControlLedger, parseAcpxTurnControl, type AcpxTurnControlMode } from "./turn-controls.js"; +import { acpxUsageEstimateNotice, persistedAcpxTurnUsage } from "./usage-accounting.js"; + const MAX_BUFFERED_EVENTS = 512; const TERMINAL_EVENT_RESERVE = 3; const TURN_START_EVENT_COUNT = 3; @@ -133,6 +136,7 @@ export interface CodexAcpxDriverOptions { runtimeDirectory: string; model: string; permissionMode?: NativeAcpxPermissionMode; + providerPolicy?: { readOnly: boolean }; systemInstructions?: string; environment?: NodeJS.ProcessEnv; managedCodexCredentialSourcePath?: string; @@ -150,6 +154,9 @@ interface CodexAcpxHost { startTurn( input: Parameters[0], ): AcpxRuntimeTurn; + steeringCapability?: AcpxRuntimeHost["steeringCapability"]; + steerActiveTurn?: AcpxRuntimeHost["steerActiveTurn"]; + queueFollowUp?: AcpxRuntimeHost["queueFollowUp"]; interruptActiveTurn(reason: string): Promise; close(input: { reason: string }): Promise; } @@ -203,6 +210,7 @@ export async function probeQualifiedAcpxEnvironment( agent: options.agent, model: options.model, permissionMode: "deny-all", + providerPolicy: { readOnly: true }, systemInstructions: "Paperclip Runner environment qualification probe. Do not execute a provider turn.", ...(options.environment === undefined @@ -262,14 +270,13 @@ export class CodexAcpxDriver implements HarnessDriver { options: CodexAcpxDriverOptions, dependencies: CodexAcpxDriverDependencies = {}, ) { - if (options.agent === "pi") { - throw new Error( - "Pi ACPX driver is unavailable until descriptor-confined verified launch is implemented", - ); + if (["pi", "cursor", "copilot"].includes(options.agent ?? "codex") && typeof options.providerPolicy?.readOnly !== "boolean") { + throw new Error("ACP candidate requires an explicit provider read-only policy"); } this.#options = { ...options, agent: options.agent ?? "codex", + ...(options.providerPolicy ? { providerPolicy: { readOnly: options.providerPolicy.readOnly } } : {}), ...(options.environment ? { environment: { ...options.environment } } : {}), @@ -441,6 +448,7 @@ export class CodexAcpxDriver implements HarnessDriver { clientCapabilities: acpxProfileClientCapabilities(this.#options.agent ?? "codex"), model: this.#options.model, permissionMode: this.#options.permissionMode ?? "approve-all", + providerPolicy: this.#options.providerPolicy, systemInstructions: this.#options.systemInstructions, environment: this.#options.environment, managedCodexCredentialSourcePath: @@ -764,6 +772,7 @@ class CodexAcpxSession implements HarnessSession { payload: Record; } | null = null; #runtimeRequestSequence = 0; + readonly #turnControls = new AcpxTurnControlLedger(); constructor(input: { host: CodexAcpxHost; @@ -856,16 +865,17 @@ class CodexAcpxSession implements HarnessSession { } const turnId = `turn-${randomBytes(12).toString("hex")}`; this.#activeTurnId = turnId; + this.#turnControls.begin(turnId); this.#assistantText = ""; this.#emit("turn.submitted", { text: input.message.text }, { turnId }); this.#emit("turn.accepted", { turnId }, { turnId }); this.#emit("turn.started", { status: "inProgress" }, { turnId }); const extensions = bindAcpxExtensionTurn({ adapter: createAcpxProfileExtensionAdapter(this.#agent, { - workspacePath: this.#input.workingDirectory, sessionId: this.ids().providerSessionId, turnId, + workspacePath: this.#input.workingDirectory, sessionId: this.#host.identity().backendSessionId, turnId, }), active: () => this.#activeTurnId === turnId && !this.#closingStarted, - sessionId: this.ids().providerSessionId, + sessionId: this.#host.identity().backendSessionId, waitForInput: (input, context) => this.#handleExtensionInput(turnId, input, context), emit: event => { if (!this.#emit(event.eventType, event.payload, { turnId, itemId: event.itemId })) { @@ -874,7 +884,10 @@ class CodexAcpxSession implements HarnessSession { }, }); let turn: AcpxRuntimeTurn; + const usageBefore = await readUsageStatus(this.#host); try { + this.#assertOpen(); + if (this.#activeTurnId !== turnId) throw new HarnessStaleTurnError(turnId); turn = this.#host.startTurn({ text: input.message.text, requestId: `${safeId(this.#input.runId, "run")}:${turnId}`, @@ -894,7 +907,7 @@ class CodexAcpxSession implements HarnessSession { ); throw error; } - const pump = this.#pumpTurn(turnId, turn, extensions.drain); + const pump = this.#pumpTurn(turnId, turn, extensions.drain, usageBefore); this.#activePump = pump; void pump .finally(() => { @@ -904,6 +917,25 @@ class CodexAcpxSession implements HarnessSession { return { turnId }; } + async steer(input: { turnId: string; message: NativeUserMessage; correlationId?: string; mode?: AcpxTurnControlMode }): Promise { + this.#assertOpen(); + if (input.turnId !== this.#activeTurnId || this.#pendingTerminal) throw new HarnessStaleTurnError(input.turnId); + const mode = input.mode ?? "steer"; + const capability = this.#host.steeringCapability?.(); + const operation = mode === "steer" ? this.#host.steerActiveTurn : this.#host.queueFollowUp; + if (!(mode === "steer" ? capability?.steering : capability?.queuedFollowUp) || !operation) { + throw new HarnessCapabilityUnavailableError(mode, "the ACP provider did not negotiate this turn control"); + } + const control = parseAcpxTurnControl({ turnId: input.turnId, message: input.message.text, mode, + controlId: input.correlationId ?? `control-${randomBytes(12).toString("hex")}` }); + this.#turnControls.reserve(control, this.#activeTurnId); + await operation.call(this.#host, control.message, `${safeId(this.#input.runId, "run")}:${control.turnId}`); + if (this.#activeTurnId !== control.turnId || this.#closingStarted) throw new HarnessStaleTurnError(control.turnId); + this.#emit("item.completed", { kind: "steering_acknowledgement", mode, + text: mode === "steer" ? "Steering acknowledged for the active turn." : "Follow-up queued after the active work.", status: "acknowledged" }, + { turnId: control.turnId, itemId: stableId("acpx-control", `${control.turnId}:${control.controlId}`) }); + } + async interrupt(input: { turnId?: string; reason?: string }): Promise { this.#assertOpen(); if (input.turnId && input.turnId !== this.#activeTurnId) { @@ -1369,7 +1401,7 @@ class CodexAcpxSession implements HarnessSession { .catch(() => undefined); } - async #pumpTurn(turnId: string, turn: AcpxRuntimeTurn, drainExtensions: () => Promise): Promise { + async #pumpTurn(turnId: string, turn: AcpxRuntimeTurn, drainExtensions: () => Promise, usageBefore: unknown): Promise { try { let index = 0; const normalizeToolEvent = @@ -1379,6 +1411,12 @@ class CodexAcpxSession implements HarnessSession { } const result = await turn.result; await drainExtensions(); + const receipt = persistedAcpxTurnUsage(usageBefore, await readUsageStatus(this.#host), turn.requestId, this.#agent); + if (receipt) { + this.#mapRuntimeEvent(receipt as unknown as AcpRuntimeEvent, turnId, ++index); + const estimate = acpxUsageEstimateNotice(receipt, `${turnId}:usage-estimate`); + if (estimate) { validateAcpxRichEvent(estimate); this.#emit(estimate.eventType, estimate.payload, { turnId, itemId: estimate.itemId }); } + } this.#cancelPendingRuntimeRequests("provider turn settled", turnId); if (this.#terminalTurns.has(turnId)) return; if (result.status === "completed") { @@ -2412,3 +2450,15 @@ async function settlesWithin( if (timer) clearTimeout(timer); } } + +/** Accounting failure cannot replace the provider's terminal result. */ +async function readUsageStatus(host: CodexAcpxHost): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + Promise.resolve().then(() => host.status()), + new Promise(resolve => { timer = setTimeout(() => resolve(undefined), 1_000); timer.unref?.(); }), + ]); + } catch { return undefined; } + finally { if (timer) clearTimeout(timer); } +} diff --git a/packages/paperclip-runner/src/drivers/acpx/generated-sidecar-contract.ts b/packages/paperclip-runner/src/drivers/acpx/generated-sidecar-contract.ts index a691384ae67..ed4bfbae44b 100644 --- a/packages/paperclip-runner/src/drivers/acpx/generated-sidecar-contract.ts +++ b/packages/paperclip-runner/src/drivers/acpx/generated-sidecar-contract.ts @@ -7,6 +7,7 @@ export const GENERATED_ACPX_SIDECAR_COMMANDS = [ "run.attach", "turn.start", "turn.cancel", + "turn.steer", "permission.resolve", "input.resolve", "tool.resolve", diff --git a/packages/paperclip-runner/src/drivers/acpx/sidecar-protocol.ts b/packages/paperclip-runner/src/drivers/acpx/sidecar-protocol.ts index 72cc6296908..0d2abe1273b 100644 --- a/packages/paperclip-runner/src/drivers/acpx/sidecar-protocol.ts +++ b/packages/paperclip-runner/src/drivers/acpx/sidecar-protocol.ts @@ -46,6 +46,7 @@ export interface AcpxSidecarOpenParams { model: string; permissionMode: NativeAcpxPermissionMode; permissionModePinned: boolean; + providerPolicy?: { readOnly: boolean }; systemInstructions: string; runtimeContext: NativeRuntimeContextSnapshot | null; tools: readonly Readonly>[]; diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-controls.test.ts b/packages/paperclip-runner/src/drivers/acpx/turn-controls.test.ts new file mode 100644 index 00000000000..cc97869d7cf --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-controls.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { AcpxTurnControlLedger, parseAcpxTurnControl } from "./turn-controls.js"; + +const control = { turnId: "turn-1", controlId: "control-1", mode: "follow_up", message: "Then validate it" }; +describe("ACP active turn controls", () => { + it("preserves explicit queued follow-up and rejects unbounded or ambiguous mutations", () => { + expect(parseAcpxTurnControl(control)).toEqual(control); + for (const invalid of [ { mode: "cancel" }, { mode: undefined }, { message: " " }, { message: "x\0y" }, + { message: "🙂".repeat(16_385) }, { turnId: "stale id" }, { controlId: "" }, { executable: "ambient" } ]) { + expect(() => parseAcpxTurnControl({ ...control, ...invalid })).toThrow(); + } + }); + it("tombstones ambiguous attempts, fences stale turns and bounds per-turn retention", () => { + const ledger = new AcpxTurnControlLedger(); + const parsed = parseAcpxTurnControl(control); + ledger.begin("turn-1"); + expect(() => ledger.reserve(parsed, null)).toThrow(/stale/); + ledger.reserve(parsed, "turn-1"); + expect(() => ledger.reserve(parsed, "turn-1")).toThrow(/already attempted/); + for (let index = 1; index < 1024; index++) ledger.reserve({ ...parsed, controlId: `control-${index + 1}` }, "turn-1"); + expect(() => ledger.reserve({ ...parsed, controlId: "overflow" }, "turn-1")).toThrow(/limit/); + ledger.begin("turn-2"); + expect(() => ledger.reserve(parsed, "turn-2")).toThrow(/stale/); + expect(() => ledger.reserve({ ...parsed, turnId: "turn-2" }, "turn-2")).not.toThrow(); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/acpx/turn-controls.ts b/packages/paperclip-runner/src/drivers/acpx/turn-controls.ts new file mode 100644 index 00000000000..0b2830dceb9 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/turn-controls.ts @@ -0,0 +1,42 @@ +export type AcpxTurnControlMode = "steer" | "follow_up"; +export interface AcpxTurnControl { + turnId: string; + controlId: string; + mode: AcpxTurnControlMode; + message: string; +} + +/** Closed, bounded mutation envelope; a follow-up never becomes a cancellation. */ +export function parseAcpxTurnControl(input: Record): AcpxTurnControl { + if (Object.keys(input).some(key => !["turnId", "controlId", "mode", "message"].includes(key))) { + throw new Error("ACP turn control contains unknown fields"); + } + const identity = (value: unknown, limit: number, name: string): string => { + if (typeof value !== "string" || value.length > limit || !/^[A-Za-z0-9][A-Za-z0-9._:-]*$/.test(value)) { + throw new Error(`ACP ${name} is invalid`); + } + return value; + }; + if (input.mode !== "steer" && input.mode !== "follow_up") throw new Error("ACP turn control mode is invalid"); + if (typeof input.message !== "string" || !input.message.trim() || input.message.includes("\0") + || Buffer.byteLength(input.message, "utf8") > 65_536) throw new Error("ACP turn control message is invalid"); + return { + turnId: identity(input.turnId, 240, "turn id"), + controlId: identity(input.controlId, 160, "control id"), + mode: input.mode, + message: input.message, + }; +} + +/** Reserve before dispatch and retain ambiguous attempts until the turn ends. */ +export class AcpxTurnControlLedger { + #turnId: string | null = null; + #attempts = new Set(); + begin(turnId: string): void { this.#turnId = turnId; this.#attempts.clear(); } + reserve(control: AcpxTurnControl, activeTurnId: string | null): void { + if (control.turnId !== activeTurnId || this.#turnId !== activeTurnId) throw new Error("ACP turn control names a stale or inactive turn"); + if (this.#attempts.has(control.controlId)) throw new Error("ACP turn control was already attempted"); + if (this.#attempts.size >= 1024) throw new Error("ACP turn control limit reached"); + this.#attempts.add(control.controlId); + } +} diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 87abaa1f400..414a4ab9d7f 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -300,11 +300,13 @@ export class CodexHarnessSession } async steer(input: { + mode?: "steer" | "follow_up"; turnId: string; message: NativeUserMessage; correlationId?: string; }): Promise { this.assertProtocolIntegrity(); + if (input.mode === "follow_up") throw this.unsupported("steering", "queued follow-up is not exposed by this driver"); this.requireCapability("steering"); this.requireActiveTurn(input.turnId, "steering"); if (input.correlationId) { diff --git a/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs b/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs index 80c32a11c6d..770f9ffece4 100644 --- a/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs +++ b/packages/paperclip-runner/test/acpx-sidecar-contract.test.mjs @@ -124,3 +124,14 @@ function error() { retryable: false, }; } + +test("turn control schema preserves explicit modes and rejects open-ended dispatch", () => { + const message = { protocolVersion, id: 1, command: "turn.steer", params: { + turnId: "turn-1", controlId: "control-1", mode: "follow_up", message: "Then validate", + } }; + assert.equal(validate(message), true, JSON.stringify(validate.errors)); + for (const params of [ { ...message.params, mode: "cancel" }, { ...message.params, method: "arbitrary" }, + { ...message.params, controlId: "" }, { ...message.params, turnId: "wrong turn" }, { ...message.params, mode: undefined } ]) { + assert.equal(validate({ ...message, params }), false); + } +});