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 <noreply@paperclip.ing>

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
     }
   }
 }

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<AcpxProviderRuntimePolicy>,
+    #[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<String>,
     #[serde(default)]
+    attempted_turn_controls: HashSet<String>,
+    #[serde(default)]
     provider_exit_unconfirmed: bool,
     #[serde(default)]
     semantic_result: Option<Value>,
@@ -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<CommandExecution, DurableRunnerError> {
+        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<CommandExecution, DurableRunnerError> {
         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::<AcpxProviderDescriptor>(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();

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<AcpxProviderRuntimePolicy>,
     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<Value, LocalRunnerError> {
+        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,

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<String, PendingInput>,
     pending_runtime_request_bytes: usize,
     semantic_result: Option<AcpxSemanticResult>,
+    attempted_turn_controls: BTreeSet<String>,
 }
 
 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<String>) -> 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(())
     }
 

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs

@@ -204,6 +204,8 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
             | "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",

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",

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<CommandExecution, DurableRunnerError> {
+        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)

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 {

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(),

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;

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 {

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());
+}

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<Record<string, unknown>>[] {
   if (!Array.isArray(value) || value.length > 512) {
     throw new Error("tools must be an array with at most 512 entries");

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;

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();

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<AcpxRuntimeHost["startTurn"]>[0],
   ): AcpxRuntimeTurn;
+  steeringCapability?: AcpxRuntimeHost["steeringCapability"];
+  steerActiveTurn?: AcpxRuntimeHost["steerActiveTurn"];
+  queueFollowUp?: AcpxRuntimeHost["queueFollowUp"];
   interruptActiveTurn(reason: string): Promise<void>;
   close(input: { reason: string }): Promise<void>;
 }
@@ -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<string, unknown>;
   } | 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<void> {
+    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<void> {
     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<void>): Promise<void> {
+  async #pumpTurn(turnId: string, turn: AcpxRuntimeTurn, drainExtensions: () => Promise<void>, usageBefore: unknown): Promise<void> {
     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<unknown> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  try {
+    return await Promise.race([
+      Promise.resolve().then(() => host.status()),
+      new Promise<undefined>(resolve => { timer = setTimeout(() => resolve(undefined), 1_000); timer.unref?.(); }),
+    ]);
+  } catch { return undefined; }
+  finally { if (timer) clearTimeout(timer); }
+}

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",

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<Record<string, unknown>>[];

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();
+  });
+});

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<string, unknown>): 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<string>();
+  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);
+  }
+}

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<void> {
     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) {

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);
+  }
+});