feat(runner): add remote websocket transports

packages/paperclip-runner/runner/Cargo.lock

@@ -1820,8 +1820,12 @@ dependencies = [
  "hmac 0.12.1",
  "jsonschema",
  "reqwest",
+ "rustls",
+ "rustls-native-certs",
+ "rustls-pemfile",
  "serde",
  "serde_json",
+ "sha1 0.10.7",
  "sha2 0.10.9",
  "tokio",
  "uuid",
@@ -2217,6 +2221,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
 dependencies = [
  "aws-lc-rs",
+ "log",
  "once_cell",
  "ring",
  "rustls-pki-types",
@@ -2237,6 +2242,15 @@ dependencies = [
  "security-framework",
 ]
 
+[[package]]
+name = "rustls-pemfile"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50"
+dependencies = [
+ "rustls-pki-types",
+]
+
 [[package]]
 name = "rustls-pki-types"
 version = "1.15.1"

packages/paperclip-runner/runner/Cargo.toml

@@ -25,3 +25,7 @@ aws-smithy-types = "1.6"
 aws-types = "1.3"
 tokio = { version = "1.50", features = ["rt-multi-thread", "sync", "time"] }
 uuid = { version = "1.22", features = ["v4"] }
+rustls = "0.23"
+rustls-native-certs = "0.8"
+rustls-pemfile = "2.2"
+sha1 = "0.10"

packages/paperclip-runner/runner/crates/runner-core/Cargo.toml

@@ -22,6 +22,10 @@ aws-smithy-types.workspace = true
 aws-types.workspace = true
 tokio.workspace = true
 uuid.workspace = true
+rustls.workspace = true
+rustls-native-certs.workspace = true
+rustls-pemfile.workspace = true
+sha1.workspace = true
 
 [[bin]]
 name = "conformance-tracer"

packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider.rs

@@ -88,6 +88,8 @@ impl AcpxProvider {
                 "workingDirectory": config.cwd,
                 "agent": config.agent,
                 "model": config.model,
+                "permissionMode": config.permission_mode,
+                "permissionModePinned": config.permission_mode_pinned,
                 "systemInstructions": config.instructions,
                 "tools": provider.tools,
                 "expectedIdentity": expected_identity,
@@ -364,7 +366,15 @@ impl AcpxProvider {
                             .unwrap_or_default(),
                     );
                 }
-                Ok(Some(map_runtime_event(payload, self.provider_requests)))
+                let turn_id = message
+                    .get("turnId")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default();
+                Ok(Some(map_runtime_event(
+                    payload,
+                    self.provider_requests,
+                    turn_id,
+                )))
             }
             GeneratedAcpxSidecarEventType::RuntimeProcess => {
                 Ok(Some(ProviderEvent::Notification {
@@ -643,7 +653,10 @@ fn validate_config(config: &AcpxProviderConfig) -> Result<(), LocalRunnerError>
         || config.agent_runtime_package.as_deref() != runtime_package
         || config.agent_runtime_version.as_deref() != runtime_version
         || config.model != model
-        || config.permission_policy != "interactive"
+        || !matches!(
+            config.permission_mode.as_str(),
+            "approve-all" | "approve-reads" | "deny-all"
+        )
         || config.command_digest != command_digest
         || config.normalized_session_id.is_empty()
         || config.run_id.is_empty()
@@ -757,7 +770,7 @@ fn permission_kind(value: Option<&str>) -> String {
     }
 }
 
-fn map_runtime_event(payload: Value, provider_requests: u64) -> ProviderEvent {
+fn map_runtime_event(payload: Value, provider_requests: u64, turn_id: &str) -> ProviderEvent {
     match payload
         .get("type")
         .and_then(Value::as_str)
@@ -775,6 +788,32 @@ fn map_runtime_event(payload: Value, provider_requests: u64) -> ProviderEvent {
             method: "item/started".to_owned(),
             params: json!({ "item": { "type": "reasoning", "status": "inProgress" } }),
         },
+        "plan" => {
+            let plan: Vec<Value> = payload
+                .get("entries")
+                .and_then(Value::as_array)
+                .map(|entries| {
+                    entries
+                        .iter()
+                        .take(256)
+                        .filter_map(|entry| {
+                            let step = entry.get("content")?.as_str()?.trim();
+                            if step.is_empty() {
+                                return None;
+                            }
+                            Some(json!({
+                                "step": step.chars().take(4000).collect::<String>(),
+                                "status": entry.get("status").and_then(Value::as_str).unwrap_or("pending"),
+                            }))
+                        })
+                        .collect()
+                })
+                .unwrap_or_default();
+            ProviderEvent::Notification {
+                method: "turn/plan/updated".to_owned(),
+                params: json!({ "turnId": turn_id, "plan": plan }),
+            }
+        }
         "semantic_result" => ProviderEvent::SemanticResult {
             result: payload.get("result").cloned().unwrap_or(Value::Null),
             item_id: payload
@@ -870,7 +909,8 @@ mod tests {
             run_id: "run".to_owned(),
             cwd: "/tmp/workspace".to_owned(),
             instructions: "safe".to_owned(),
-            permission_policy: "interactive".to_owned(),
+            permission_mode: "approve-all".to_owned(),
+            permission_mode_pinned: true,
             runtime_context: None,
         };
         assert!(validate_config(&config).is_err());
@@ -881,6 +921,7 @@ mod tests {
         let event = map_runtime_event(
             json!({ "type": "thinking", "text": "secret chain of thought" }),
             1,
+            "turn-1",
         );
         match event {
             ProviderEvent::Notification { params, .. } => {
@@ -899,6 +940,7 @@ mod tests {
         let event = map_runtime_event(
             json!({ "type": "semantic_result", "callId": "finish-1", "result": result }),
             1,
+            "turn-1",
         );
         match event {
             ProviderEvent::SemanticResult { result, item_id } => {
@@ -908,4 +950,31 @@ mod tests {
             _ => panic!("expected semantic result"),
         }
     }
+
+    #[test]
+    fn preserves_every_structured_plan_entry_for_the_active_turn() {
+        let event = map_runtime_event(
+            json!({
+                "type": "plan",
+                "entries": [
+                    {"content": "Inspect", "status": "completed", "priority": "high"},
+                    {"content": "Implement", "status": "in_progress", "priority": "medium"},
+                    {"content": "Verify", "status": "pending", "priority": "low"}
+                ]
+            }),
+            1,
+            "turn-acp",
+        );
+        match event {
+            ProviderEvent::Notification { method, params } => {
+                assert_eq!(method, "turn/plan/updated");
+                assert_eq!(params["turnId"], "turn-acp");
+                assert_eq!(params["plan"].as_array().unwrap().len(), 3);
+                assert_eq!(params["plan"][0]["step"], "Inspect");
+                assert_eq!(params["plan"][1]["status"], "in_progress");
+                assert_eq!(params["plan"][2]["step"], "Verify");
+            }
+            _ => panic!("expected plan notification"),
+        }
+    }
 }

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs

@@ -4,13 +4,16 @@ use std::process::ExitCode;
 
 use serde_json::{json, Value};
 
-#[derive(Clone, Copy, Default)]
+#[derive(Clone, Default)]
 struct Mode {
     large_event: bool,
     oversized_event: bool,
     linger_after_turn_start: bool,
     runtime_question: bool,
     runtime_elicitation: bool,
+    structured_activity: bool,
+    include_skill_instructions: bool,
+    expected_approval_policy: Option<String>,
 }
 
 fn send(value: Value) -> io::Result<()> {
@@ -28,6 +31,12 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
         linger_after_turn_start: args.iter().any(|arg| arg == "--linger-after-turn-start"),
         runtime_question: args.iter().any(|arg| arg == "--runtime-question"),
         runtime_elicitation: args.iter().any(|arg| arg == "--runtime-elicitation"),
+        structured_activity: args.iter().any(|arg| arg == "--structured-activity"),
+        include_skill_instructions: args.iter().any(|arg| arg == "--include-skill-instructions"),
+        expected_approval_policy: args
+            .windows(2)
+            .find(|pair| pair[0] == "--expected-approval-policy")
+            .map(|pair| pair[1].clone()),
     };
     let mut turn_count = 0_u64;
     let mut active_turn_id: Option<String> = None;
@@ -54,6 +63,17 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                 "result": {"data": [{"name": "Plan", "mode": "plan", "model": "gpt-test", "reasoning_effort": "high"}]}
             }))?,
             "thread/start" | "thread/resume" => {
+                if let Some(expected) = mode.expected_approval_policy.as_deref() {
+                    if request
+                        .pointer("/params/approvalPolicy")
+                        .and_then(Value::as_str)
+                        != Some(expected)
+                    {
+                        return Err(
+                            format!("{method} omitted expected approvalPolicy={expected}").into(),
+                        );
+                    }
+                }
                 let tools = request
                     .pointer("/params/dynamicTools")
                     .and_then(Value::as_array)
@@ -69,8 +89,18 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     .pointer("/params/permissions")
                     .and_then(Value::as_str)
                     == Some("paperclip-runner-workspace-read-only");
+                if config
+                    .and_then(|value| value.get("skills.include_instructions"))
+                    .and_then(Value::as_bool)
+                    != Some(mode.include_skill_instructions)
+                {
+                    return Err(format!(
+                        "thread config omitted skills.include_instructions={}",
+                        mode.include_skill_instructions
+                    )
+                    .into());
+                }
                 for key in [
-                    "skills.include_instructions",
                     "include_apps_instructions",
                     "features.apps",
                     "features.plugins",
@@ -137,6 +167,43 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     "method": "turn/started",
                     "params": {"threadId": "fake-thread", "turn": {"id": turn_id, "status": "inProgress"}}
                 }))?;
+                if mode.structured_activity {
+                    send(json!({
+                        "method": "turn/plan/updated",
+                        "params": {
+                            "threadId": "fake-thread",
+                            "turnId": turn_id,
+                            "revision": 1,
+                            "plan": [
+                                {"step": "Inspect the runner path", "status": "inProgress"},
+                                {"step": "Implement the status island", "status": "pending"},
+                                {"step": "Verify replay and UI behavior", "status": "pending"}
+                            ]
+                        }
+                    }))?;
+                    send(json!({
+                        "method": "turn/diff/updated",
+                        "params": {
+                            "threadId": "fake-thread",
+                            "turnId": turn_id,
+                            "revision": 1,
+                            "diff": "diff --git a/ui/src/status.tsx b/ui/src/status.tsx\nnew file mode 100644\n--- /dev/null\n+++ b/ui/src/status.tsx\n@@ -0,0 +1,2 @@\n+export const status = 'working';\n+export const steps = 3;\n"
+                        }
+                    }))?;
+                    send(json!({
+                        "method": "turn/plan/updated",
+                        "params": {
+                            "threadId": "fake-thread",
+                            "turnId": turn_id,
+                            "revision": 2,
+                            "plan": [
+                                {"step": "Inspect the runner path", "status": "completed"},
+                                {"step": "Implement the status island", "status": "inProgress"},
+                                {"step": "Verify replay and UI behavior", "status": "pending"}
+                            ]
+                        }
+                    }))?;
+                }
                 if mode.large_event || mode.oversized_event {
                     let bytes = if mode.oversized_event {
                         4 * 1024 * 1024 + 128
@@ -262,6 +329,30 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
             .and_then(Value::as_str)
             .and_then(|request_id| pending_tool_turns.remove(request_id));
         if let Some(turn_id) = completed_turn.filter(|_| request.get("result").is_some()) {
+            if mode.structured_activity {
+                send(json!({
+                    "method": "turn/plan/updated",
+                    "params": {
+                        "threadId": "fake-thread",
+                        "turnId": turn_id,
+                        "revision": 3,
+                        "plan": [
+                            {"step": "Inspect the runner path", "status": "completed"},
+                            {"step": "Implement the status island", "status": "completed"},
+                            {"step": "Verify replay and UI behavior", "status": "completed"}
+                        ]
+                    }
+                }))?;
+                send(json!({
+                    "method": "turn/diff/updated",
+                    "params": {
+                        "threadId": "fake-thread",
+                        "turnId": turn_id,
+                        "revision": 2,
+                        "diff": "diff --git a/ui/src/status.tsx b/ui/src/status.tsx\nnew file mode 100644\n--- /dev/null\n+++ b/ui/src/status.tsx\n@@ -0,0 +1,3 @@\n+export const status = 'complete';\n+export const steps = 3;\n+export const verified = true;\n"
+                    }
+                }))?;
+            }
             send(json!({
                 "method": "paperclip/runResult",
                 "params": {

packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs

@@ -25,7 +25,8 @@ fn build_metadata() -> serde_json::Value {
             "name": "paperclip.runner",
             "minimumVersion": 1,
             "maximumVersion": 1
-        }
+        },
+        "prpTransportModes": ["dial_ws_loopback", "dial_wss", "listen_ws"]
     })
 }
 
@@ -225,8 +226,40 @@ fn run_durable_mode(
     let max_runtime = optional_duration_value(args, "--max-lifetime-ms")?
         .or(optional_duration_value(args, "--max-runtime-ms")?)
         .unwrap_or(Duration::MAX);
+    let connect_url = match (
+        args.iter().any(|argument| argument == "--connect-url"),
+        args.iter().any(|argument| argument == "--listen-address")
+            || args.iter().any(|argument| argument == "--listen-port")
+            || args.iter().any(|argument| argument == "--listen-path"),
+    ) {
+        (true, false) => value(args, "--connect-url")?,
+        (false, true) => {
+            let address = value(args, "--listen-address")?;
+            let port = value(args, "--listen-port")?;
+            let path = value(args, "--listen-path")?;
+            if address != "0.0.0.0" || port != "43127" {
+                return Err(LocalRunnerError::invalid(
+                    "runner listener requires --listen-address 0.0.0.0 and --listen-port 43127",
+                ));
+            }
+            if !path.starts_with("/api/runner/v1/connect/") || path.contains(['?', '#', '\\']) {
+                return Err(LocalRunnerError::invalid("runner listener path is invalid"));
+            }
+            format!("listen://{address}:{port}{path}")
+        }
+        _ => {
+            return Err(LocalRunnerError::invalid(
+                "durable runner requires exactly one of --connect-url or --listen-address/--listen-port/--listen-path",
+            ))
+        }
+    };
     let config = DurableRunnerConfig {
-        connect_url: value(args, "--connect-url")?,
+        connect_url,
+        ca_bundle_path: args
+            .iter()
+            .any(|argument| argument == "--ca-bundle-path")
+            .then(|| value(args, "--ca-bundle-path").map(PathBuf::from))
+            .transpose()?,
         state_dir: PathBuf::from(value(args, "--state-dir")?),
         runner_instance_id: value(args, "--runner-id")?,
         environment_lease_id: value(args, "--environment-lease-id")?,
@@ -250,6 +283,7 @@ fn run_durable_mode(
         p0_reserve_bytes: usize_value(args, "--p0-reserve-bytes", 32 * 1024)?,
         max_frame_bytes: usize_value(args, "--max-frame-bytes", 1024 * 1024)?,
         reconnect_delay: duration_value(args, "--reconnect-delay-ms", 25)?,
+        reconnect_grace: optional_duration_value(args, "--reconnect-grace-ms")?,
         max_runtime,
         lifecycle_mode,
         idle_timeout,
@@ -272,7 +306,9 @@ fn run(bootstrap_ticket: Option<BootstrapTicket>) -> Result<(), LocalRunnerError
     {
         return run_codex_app_server_proxy(&args);
     }
-    if args.iter().any(|argument| argument == "--connect-url") {
+    if args.iter().any(|argument| argument == "--connect-url")
+        || args.iter().any(|argument| argument == "--listen-address")
+    {
         return run_durable_mode(&args, bootstrap_ticket);
     }
     run_local_runner(RunnerConfig {
@@ -305,6 +341,10 @@ mod tests {
         assert_eq!(metadata["harnessDriverVersion"], 1);
         assert_eq!(metadata["prp"]["minimumVersion"], 1);
         assert_eq!(metadata["prp"]["maximumVersion"], 1);
+        assert_eq!(
+            metadata["prpTransportModes"],
+            json!(["dial_ws_loopback", "dial_wss", "listen_ws"])
+        );
     }
 }
 

packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs

@@ -259,6 +259,8 @@ pub struct LocalProviderConfig {
     pub args: Vec<String>,
     pub cwd: String,
     pub model: Option<String>,
+    #[serde(default = "default_codex_approval_policy")]
+    pub approval_policy: String,
     pub instructions: String,
     #[serde(default = "default_collaboration_mode")]
     pub collaboration_mode: String,
@@ -333,11 +335,22 @@ pub struct AcpxProviderConfig {
     pub run_id: String,
     pub cwd: String,
     pub instructions: String,
-    pub permission_policy: String,
+    #[serde(default = "default_legacy_acpx_permission_mode")]
+    pub permission_mode: String,
+    #[serde(default)]
+    pub permission_mode_pinned: bool,
     #[serde(default)]
     pub runtime_context: Option<Value>,
 }
 
+fn default_codex_approval_policy() -> String {
+    "never".to_owned()
+}
+
+fn default_legacy_acpx_permission_mode() -> String {
+    "approve-reads".to_owned()
+}
+
 #[derive(Clone, Debug, PartialEq)]
 pub enum ProviderConfig {
     Local(LocalProviderConfig),
@@ -700,7 +713,7 @@ fn codex_question_set(method: &str, params: &Value) -> Option<Value> {
                             .unwrap_or("Question"),
                         4_000,
                     ),
-                    "required": question.get("required").and_then(Value::as_bool).unwrap_or(true),
+                    "required": question.get("required").and_then(Value::as_bool).unwrap_or(false),
                     "answerMode": if options.is_empty() {
                         "text"
                     } else if question.get("multiSelect").and_then(Value::as_bool).unwrap_or(false)
@@ -1142,6 +1155,14 @@ impl CodexProvider {
         tools: impl Iterator<Item = AuthorizedTool>,
         resume_thread_id: Option<&str>,
     ) -> Result<Self, LocalRunnerError> {
+        if !matches!(
+            config.approval_policy.as_str(),
+            "never" | "on-request" | "untrusted"
+        ) {
+            return Err(LocalRunnerError::invalid(
+                "unsupported Codex approval policy",
+            ));
+        }
         let mut provider = Self {
             kind: config.kind,
             process: SupervisedProcess::spawn(
@@ -1193,7 +1214,7 @@ impl CodexProvider {
                     "threadId": thread_id,
                     "cwd": config.cwd,
                     "model": config.model,
-                    "approvalPolicy": "never",
+                    "approvalPolicy": config.approval_policy,
                     "permissions": permission_profile,
                     "runtimeWorkspaceRoots": [config.cwd],
                     "config": isolated_thread_config(
@@ -1212,7 +1233,7 @@ impl CodexProvider {
                 json!({
                     "cwd": config.cwd,
                     "model": config.model,
-                    "approvalPolicy": "never",
+                    "approvalPolicy": config.approval_policy,
                     "permissions": permission_profile,
                     "runtimeWorkspaceRoots": [config.cwd],
                     "config": isolated_thread_config(
@@ -1798,6 +1819,7 @@ mod provider_trace_tests {
         )
         .unwrap();
         assert_eq!(question_set["questions"].as_array().unwrap().len(), 3);
+        assert_eq!(question_set["questions"][0]["required"], false);
         assert_eq!(
             question_set["questions"][0]["options"][0]["label"],
             "Focused"
@@ -1877,6 +1899,7 @@ mod provider_trace_tests {
             "collaborationMode": "default"
         }))
         .unwrap();
+        assert_eq!(default_config.approval_policy, "never");
         assert!(default_config.include_collaboration_mode_instructions);
         assert_eq!(
             isolated_thread_config(

packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs

@@ -1,3 +1,17 @@
+fn sleep_for_reconnect(base: Duration, attempt: &mut u32) {
+    let multiplier = 1_u128 << (*attempt).min(5);
+    let uncapped = base.as_millis().saturating_mul(multiplier);
+    let capped = uncapped.min(5_000).max(1) as u64;
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |duration| u64::from(duration.subsec_nanos()));
+    let jitter_percent = 75 + nanos % 51;
+    *attempt = attempt.saturating_add(1);
+    thread::sleep(Duration::from_millis(
+        capped.saturating_mul(jitter_percent) / 100,
+    ));
+}
+
 pub fn run_durable_runner(
     config: DurableRunnerConfig,
     bootstrap_ticket: BootstrapTicket,
@@ -23,6 +37,9 @@ pub fn run_durable_runner(
         state.lifecycle = "ready".to_owned();
         state.record_diagnostic("resuming an explicitly suspended durable runner");
     }
+    // Bind listener mode before starting the harness. A fixed-port collision is
+    // an unhealthy lease and must fail without launching provider work.
+    let endpoint = RunnerTransportEndpoint::new(&config.connect_url)?;
     let mut provider = match start_configured_provider(&mut state) {
         Ok(provider) => provider,
         Err(error) => {
@@ -50,9 +67,6 @@ pub fn run_durable_runner(
         }
     }
     store.save(&state)?;
-    // Resolve once before authentication. Reconnects use only this validated,
-    // concrete address set, so DNS cannot redirect a retry.
-    let target = ResolvedWsTarget::resolve(&config.connect_url)?;
     if recovered {
         state.reconnect_count += 1;
         state.record_diagnostic("runner process restored the same durable identity");
@@ -78,8 +92,11 @@ pub fn run_durable_runner(
     let mut connection_lease_expires_at_unix_ms: Option<u64> = None;
     let mut connection_lease_revocation_epoch: Option<u64> = None;
     let started = Instant::now();
+    let mut reconnect_attempt = 0_u32;
+    let mut authenticated_once = false;
+    let mut disconnected_since: Option<Instant> = None;
 
-    loop {
+    'transport: loop {
         if started.elapsed() >= config.max_runtime && !state.active_turn {
             state.recoverable_failure = Some("transport_reconnect_deadline_exceeded".to_owned());
             state.lifecycle = "recoverable_failure".to_owned();
@@ -91,6 +108,23 @@ pub fn run_durable_runner(
                 "transport reconnect deadline exceeded; durable state is preserved",
             ));
         }
+        if authenticated_once {
+            let disconnected_at = disconnected_since.get_or_insert_with(Instant::now);
+            if config
+                .reconnect_grace
+                .is_some_and(|grace| disconnected_at.elapsed() >= grace)
+            {
+                state.recoverable_failure = Some("transport_reconnect_grace_exceeded".to_owned());
+                state.lifecycle = "recoverable_failure".to_owned();
+                state.record_diagnostic(
+                    "transport reconnect grace exceeded; durable state is preserved",
+                );
+                store.save(&state)?;
+                return Err(DurableRunnerError::invalid(
+                    "transport reconnect grace exceeded; durable state is preserved",
+                ));
+            }
+        }
         if connection_lease_expires_at_unix_ms
             .is_some_and(|expires_at| current_unix_ms().is_ok_and(|now| now >= expires_at))
         {
@@ -119,13 +153,20 @@ pub fn run_durable_runner(
                 )
             })?;
         let credential = CredentialMaterial::from_token(credential_token);
-        let mut client = match WsClient::connect(&target, config.max_frame_bytes) {
-            Ok(client) => client,
+        let mut client = match endpoint.open(
+            config.max_frame_bytes,
+            config.ca_bundle_path.as_deref(),
+        ) {
+            Ok(Some(client)) => client,
+            Ok(None) => {
+                thread::sleep(config.reconnect_delay);
+                continue;
+            }
             Err(error) => {
                 let text = error.to_string();
                 state.record_diagnostic(format!("transport reconnect scheduled: {text}"));
                 store.save(&state)?;
-                thread::sleep(config.reconnect_delay);
+                sleep_for_reconnect(config.reconnect_delay, &mut reconnect_attempt);
                 continue;
             }
         };
@@ -143,7 +184,7 @@ pub fn run_durable_runner(
                 "transport peer authentication failed closed: {error}"
             ));
             store.save(&state)?;
-            thread::sleep(config.reconnect_delay);
+            sleep_for_reconnect(config.reconnect_delay, &mut reconnect_attempt);
             continue;
         }
         let mut welcome = loop {
@@ -156,12 +197,15 @@ pub fn run_durable_runner(
                 Err(error) => {
                     state.record_diagnostic(error.to_string());
                     store.save(&state)?;
-                    thread::sleep(config.reconnect_delay);
-                    continue;
+                    sleep_for_reconnect(config.reconnect_delay, &mut reconnect_attempt);
+                    continue 'transport;
                 }
             }
         };
         let (_, connection) = validate_welcome(&welcome, &state)?;
+        authenticated_once = true;
+        disconnected_since = None;
+        reconnect_attempt = 0;
         let next_lease_token = match welcome.pointer_mut("/payload/connectionLeaseToken") {
             Some(Value::String(value)) => {
                 let token = std::mem::take(value);
@@ -226,7 +270,7 @@ pub fn run_durable_runner(
             state.record_diagnostic(error.to_string());
             state.reconnect_count += 1;
             store.save(&state)?;
-            thread::sleep(config.reconnect_delay);
+            sleep_for_reconnect(config.reconnect_delay, &mut reconnect_attempt);
             continue;
         }
 
@@ -480,7 +524,7 @@ pub fn run_durable_runner(
             }
         };
         if disconnected {
-            thread::sleep(config.reconnect_delay);
+            sleep_for_reconnect(config.reconnect_delay, &mut reconnect_attempt);
         }
     }
 }
@@ -636,12 +680,14 @@ fn provider_descriptor(
                 _ => None,
             },
             "agentProcessId": runtime.and_then(|item| item.agent_process_id()),
+            "permissionMode": value.permission_mode,
         }),
         Some(crate::codex_provider::ProviderConfig::Local(value)) => json!({
             "provider": provider_kind_name(value.kind),
             "driver": if value.kind == crate::codex_provider::ProviderKind::Opencode { "opencode_server" } else { "codex_app_server" },
             "model": value.model,
             "collaborationMode": value.collaboration_mode,
+            "approvalPolicy": value.approval_policy,
             "collaborationInstructions": value.include_collaboration_mode_instructions,
             "executionKind": "local_process",
             "providerVersion": if value.kind == crate::codex_provider::ProviderKind::Opencode { "1.18.17" } else { "unqualified" },
@@ -2094,7 +2140,8 @@ mod tests {
                 run_id: "run".to_owned(),
                 cwd: "/tmp/workspace".to_owned(),
                 instructions: "safe".to_owned(),
-                permission_policy: "interactive".to_owned(),
+                permission_mode: "approve-all".to_owned(),
+                permission_mode_pinned: true,
                 runtime_context: None,
             },
         ));
@@ -2134,7 +2181,8 @@ mod tests {
                 run_id: "run".to_owned(),
                 cwd: "/tmp/workspace".to_owned(),
                 instructions: "safe".to_owned(),
-                permission_policy: "interactive".to_owned(),
+                permission_mode: "approve-all".to_owned(),
+                permission_mode_pinned: true,
                 runtime_context: None,
             },
         ));
@@ -2253,6 +2301,7 @@ mod tests {
     fn config(root: &Path) -> DurableRunnerConfig {
         DurableRunnerConfig {
             connect_url: "ws://127.0.0.1:1/durable_runner/connect".to_owned(),
+            ca_bundle_path: None,
             state_dir: root.to_path_buf(),
             runner_instance_id: "runner_durable_runner_stable".to_owned(),
             environment_lease_id: "lease_durable_runner_stable".to_owned(),
@@ -2268,6 +2317,7 @@ mod tests {
             p0_reserve_bytes: 8 * 1024,
             max_frame_bytes: 1024 * 1024,
             reconnect_delay: Duration::from_millis(1),
+            reconnect_grace: None,
             max_runtime: Duration::from_millis(10),
             lifecycle_mode: "per_turn".to_owned(),
             idle_timeout: None,
@@ -2722,7 +2772,7 @@ mod tests {
             |_host, _port| Ok(vec![address]),
         )
         .unwrap();
-        let mut client = WsClient::connect(&target, config.max_frame_bytes).unwrap();
+        let mut client = WsClient::connect(&target, config.max_frame_bytes, None).unwrap();
         let error = authenticate_transport(
             &mut client,
             &state,
@@ -2733,7 +2783,8 @@ mod tests {
             None,
             None,
         )
-        .unwrap_err();
+        .err()
+        .unwrap();
         assert!(error
             .to_string()
             .contains("transport authentication proof is invalid"));
@@ -3068,9 +3119,7 @@ mod tests {
     #[test]
     fn destination_validation_rejects_malformed_userinfo_query_and_fragment_before_resolution() {
         for url in [
-            "wss://127.0.0.1:3100/durable_runner/connect",
             "WS://127.0.0.1:3100/durable_runner/connect",
-            "ws://127.0.0.1/durable_runner/connect",
             "ws://127.0.0.1:0/durable_runner/connect",
             "ws://[::1:3100/durable_runner/connect",
             "ws://user@127.0.0.1:3100/durable_runner/connect",
@@ -3091,6 +3140,31 @@ mod tests {
         }
     }
 
+    #[test]
+    fn destination_validation_allows_verified_wss_remote_addresses_and_default_ports() {
+        let target = resolve_ws_target_with(
+            "wss://paperclip.example.test/api/runner/v1/connect/run",
+            |host, port| {
+                assert_eq!(host, "paperclip.example.test");
+                assert_eq!(port, 443);
+                Ok(vec!["192.0.2.10:443".parse().unwrap()])
+            },
+        )
+        .unwrap();
+        assert!(target.secure);
+        assert_eq!(target.host, "paperclip.example.test");
+
+        let loopback = resolve_ws_target_with(
+            "ws://127.0.0.1/api/runner/v1/connect/run",
+            |_host, port| {
+                assert_eq!(port, 80);
+                Ok(vec!["127.0.0.1:80".parse().unwrap()])
+            },
+        )
+        .unwrap();
+        assert!(!loopback.secure);
+    }
+
     #[test]
     fn destination_validation_accepts_only_concrete_loopback_answers() {
         let target = resolve_ws_target_with(
@@ -3112,7 +3186,7 @@ mod tests {
 
     #[test]
     fn oversized_outbound_websocket_frame_is_rejected_before_write() {
-        let error = encode_masked_frame(0x1, b"ninebytes", [0; 4], 8).unwrap_err();
+        let error = encode_frame(0x1, b"ninebytes", Some([0; 4]), 8).unwrap_err();
         assert!(error
             .to_string()
             .contains("outbound WebSocket frame exceeds"));
@@ -3126,6 +3200,78 @@ mod tests {
             .contains("inbound WebSocket frame exceeds"));
     }
 
+    fn listener_client(request: &str) -> (TcpStream, TcpStream) {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let mut client = TcpStream::connect(listener.local_addr().unwrap()).unwrap();
+        client.write_all(request.as_bytes()).unwrap();
+        client.flush().unwrap();
+        let (server, _) = listener.accept().unwrap();
+        (client, server)
+    }
+
+    #[test]
+    fn runner_listener_accepts_only_the_exact_websocket_path_without_compression() {
+        let (mut wrong_path_client, wrong_path_server) = listener_client(
+            "GET /unrelated HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGVzdA==\r\nSec-WebSocket-Version: 13\r\n\r\n",
+        );
+        let error = WsClient::accept(
+            wrong_path_server,
+            "/api/runner/v1/connect/run-1",
+            1024,
+        )
+        .err()
+        .unwrap();
+        assert!(error.to_string().contains("unrelated HTTP path"));
+        let response = read_http_headers(&mut wrong_path_client, "test response").unwrap();
+        assert!(response.starts_with("HTTP/1.1 404 "));
+
+        let (_compression_client, compression_server) = listener_client(
+            "GET /api/runner/v1/connect/run-1 HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGVzdA==\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Extensions: permessage-deflate\r\n\r\n",
+        );
+        let error = WsClient::accept(
+            compression_server,
+            "/api/runner/v1/connect/run-1",
+            1024,
+        )
+        .err()
+        .unwrap();
+        assert!(error.to_string().contains("does not support WebSocket compression"));
+    }
+
+    #[test]
+    fn runner_listener_requires_masked_client_frames_and_enforces_control_limits() {
+        let (mut client, server) = listener_client(
+            "GET /api/runner/v1/connect/run-1 HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGVzdA==\r\nSec-WebSocket-Version: 13\r\n\r\n",
+        );
+        let mut accepted = WsClient::accept(
+            server,
+            "/api/runner/v1/connect/run-1",
+            1024,
+        )
+        .unwrap();
+        let response = read_http_headers(&mut client, "test response").unwrap();
+        assert!(response.starts_with("HTTP/1.1 101 "));
+        client.write_all(&encode_frame(0x1, b"{}", None, 1024).unwrap()).unwrap();
+        let error = accepted.receive_plain_json().unwrap_err();
+        assert!(error.to_string().contains("masking direction is invalid"));
+
+        let (mut control_client, control_server) = listener_client(
+            "GET /api/runner/v1/connect/run-1 HTTP/1.1\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGVzdA==\r\nSec-WebSocket-Version: 13\r\n\r\n",
+        );
+        let mut accepted = WsClient::accept(
+            control_server,
+            "/api/runner/v1/connect/run-1",
+            1024,
+        )
+        .unwrap();
+        let _ = read_http_headers(&mut control_client, "test response").unwrap();
+        control_client
+            .write_all(&[0x89, 0xFE, 0, 126])
+            .unwrap();
+        let error = accepted.receive_plain_json().unwrap_err();
+        assert!(error.to_string().contains("control frame exceeds 125 bytes"));
+    }
+
     #[test]
     fn revoked_runner_rejects_turn_with_zero_effects_and_preserves_existing_outbox() {
         let root = temporary_root("revoked-command");

packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs

@@ -3,9 +3,10 @@ use std::error::Error;
 use std::fmt::{self, Display, Formatter};
 use std::fs::{self, File, OpenOptions};
 use std::io::{self, Read, Write};
-use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
+use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
 use std::path::{Path, PathBuf};
 use std::thread;
+use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 #[cfg(unix)]
@@ -17,6 +18,10 @@ use hmac::{Hmac, Mac};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 use sha2::{Digest, Sha256};
+use sha1::Sha1;
+use base64::Engine as _;
+use rustls::pki_types::ServerName;
+use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned};
 
 use crate::codex_provider::ProviderConfig;
 use crate::process_supervisor::SupervisedProcess;
@@ -105,6 +110,7 @@ pub fn capture_bootstrap_ticket() -> Result<Option<BootstrapTicket>, DurableRunn
 #[derive(Clone, Debug)]
 pub struct DurableRunnerConfig {
     pub connect_url: String,
+    pub ca_bundle_path: Option<PathBuf>,
     pub state_dir: PathBuf,
     pub runner_instance_id: String,
     pub environment_lease_id: String,
@@ -120,6 +126,7 @@ pub struct DurableRunnerConfig {
     pub p0_reserve_bytes: usize,
     pub max_frame_bytes: usize,
     pub reconnect_delay: Duration,
+    pub reconnect_grace: Option<Duration>,
     pub max_runtime: Duration,
     pub lifecycle_mode: String,
     pub idle_timeout: Option<Duration>,

packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs

@@ -1,13 +1,20 @@
 struct ParsedWsUrl {
+    secure: bool,
     host: String,
     authority: String,
     port: u16,
     path: String,
 }
 fn parse_ws_url(input: &str) -> Result<ParsedWsUrl, DurableRunnerError> {
-    let remainder = input
-        .strip_prefix("ws://")
-        .ok_or_else(|| DurableRunnerError::invalid("runner core accepts exactly the ws:// scheme"))?;
+    let (secure, remainder, default_port) = if let Some(value) = input.strip_prefix("ws://") {
+        (false, value, 80)
+    } else if let Some(value) = input.strip_prefix("wss://") {
+        (true, value, 443)
+    } else {
+        return Err(DurableRunnerError::invalid(
+            "runner connect URL must use ws:// or wss://",
+        ));
+    };
     if remainder.is_empty()
         || remainder
             .chars()
@@ -33,16 +40,24 @@ fn parse_ws_url(input: &str) -> Result<ParsedWsUrl, DurableRunnerError> {
             .find(']')
             .ok_or_else(|| DurableRunnerError::invalid("bracketed IPv6 authority is malformed"))?;
         let host = &authority[1..closing];
-        let port = authority[closing + 1..]
-            .strip_prefix(':')
-            .ok_or_else(|| DurableRunnerError::invalid("bracketed IPv6 authority requires a port"))?;
+        let suffix = &authority[closing + 1..];
+        let port = if suffix.is_empty() {
+            default_port.to_string()
+        } else {
+            suffix
+                .strip_prefix(':')
+                .ok_or_else(|| DurableRunnerError::invalid("bracketed IPv6 authority is malformed"))?
+                .to_owned()
+        };
         host.parse::<std::net::Ipv6Addr>()
             .map_err(|_| DurableRunnerError::invalid("bracketed WebSocket host must be IPv6"))?;
         (host, port)
     } else {
         let (host, port) = authority
             .rsplit_once(':')
-            .ok_or_else(|| DurableRunnerError::invalid("WebSocket URL must include an explicit port"))?;
+            .map_or((authority, default_port.to_string()), |(host, port)| {
+                (host, port.to_owned())
+            });
         if host.is_empty() || host.contains(':') {
             return Err(DurableRunnerError::invalid(
                 "WebSocket host is empty or an unbracketed IPv6 literal",
@@ -57,6 +72,7 @@ fn parse_ws_url(input: &str) -> Result<ParsedWsUrl, DurableRunnerError> {
         return Err(DurableRunnerError::invalid("WebSocket port must be non-zero"));
     }
     Ok(ParsedWsUrl {
+        secure,
         host: host.to_owned(),
         authority: authority.to_owned(),
         port,
@@ -65,6 +81,8 @@ fn parse_ws_url(input: &str) -> Result<ParsedWsUrl, DurableRunnerError> {
 }
 
 struct ResolvedWsTarget {
+    secure: bool,
+    host: String,
     authority: String,
     path: String,
     addresses: Vec<SocketAddr>,
@@ -80,6 +98,61 @@ impl ResolvedWsTarget {
     }
 }
 
+enum RunnerTransportEndpoint {
+    Dial(String),
+    Listen { listener: TcpListener, path: String },
+}
+
+impl RunnerTransportEndpoint {
+    fn new(input: &str) -> Result<Self, DurableRunnerError> {
+        if let Some(remainder) = input.strip_prefix("listen://") {
+            let (authority, path) = remainder.split_once('/').ok_or_else(|| {
+                DurableRunnerError::invalid("runner_ingress_bind_conflict: listener path is required")
+            })?;
+            if authority != "0.0.0.0:43127" {
+                return Err(DurableRunnerError::invalid(
+                    "runner_ingress_bind_conflict: listener must bind 0.0.0.0:43127",
+                ));
+            }
+            let path = format!("/{path}");
+            let listener = TcpListener::bind(authority).map_err(|error| {
+                DurableRunnerError::invalid(format!(
+                    "runner_ingress_bind_conflict: failed to bind fixed listener: {error}"
+                ))
+            })?;
+            listener.set_nonblocking(true).map_err(|error| {
+                DurableRunnerError::invalid(format!(
+                    "runner_ingress_bind_conflict: failed to configure listener: {error}"
+                ))
+            })?;
+            return Ok(Self::Listen { listener, path });
+        }
+        Ok(Self::Dial(input.to_owned()))
+    }
+
+    fn open(
+        &self,
+        max_frame_bytes: usize,
+        ca_bundle_path: Option<&Path>,
+    ) -> Result<Option<WsClient>, DurableRunnerError> {
+        match self {
+            Self::Dial(url) => {
+                // Resolve on every reconnect. TLS authenticates the configured
+                // hostname, so a recovered connection is not pinned to a stale IP.
+                let target = ResolvedWsTarget::resolve(url)?;
+                WsClient::connect(&target, max_frame_bytes, ca_bundle_path).map(Some)
+            }
+            Self::Listen { listener, path } => match listener.accept() {
+                Ok((stream, _peer)) => WsClient::accept(stream, path, max_frame_bytes).map(Some),
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(None),
+                Err(error) => Err(DurableRunnerError::invalid(format!(
+                    "runner ingress listener accept failed: {error}"
+                ))),
+            },
+        }
+    }
+}
+
 fn resolve_ws_target_with<F>(input: &str, resolver: F) -> Result<ResolvedWsTarget, DurableRunnerError>
 where
     F: FnOnce(&str, u16) -> io::Result<Vec<SocketAddr>>,
@@ -95,21 +168,28 @@ where
             "WebSocket destination resolved to no addresses",
         ));
     }
-    if addresses.iter().any(|address| !address.ip().is_loopback()) {
+    if !parsed.secure && addresses.iter().any(|address| !address.ip().is_loopback()) {
         return Err(DurableRunnerError::invalid(
-            "every WebSocket destination must be loopback (127.0.0.0/8 or ::1)",
+            "plaintext WebSocket destinations must all be loopback (127.0.0.0/8 or ::1)",
         ));
     }
     Ok(ResolvedWsTarget {
+        secure: parsed.secure,
+        host: parsed.host,
         authority: parsed.authority,
         path: parsed.path,
         addresses,
     })
 }
 
+trait WsReadWrite: Read + Write {}
+impl<T: Read + Write> WsReadWrite for T {}
+
 struct WsClient {
-    stream: TcpStream,
+    stream: Box<dyn WsReadWrite + Send>,
     mask_counter: u32,
+    mask_outbound: bool,
+    require_inbound_mask: bool,
     max_frame_bytes: usize,
     secure_channel: Option<SecureChannel>,
 }
@@ -236,10 +316,10 @@ impl SecureChannel {
     }
 }
 
-fn encode_masked_frame(
+fn encode_frame(
     opcode: u8,
     payload: &[u8],
-    mask: [u8; 4],
+    mask: Option<[u8; 4]>,
     max_frame_bytes: usize,
 ) -> Result<Vec<u8>, DurableRunnerError> {
     if payload.len() > max_frame_bytes {
@@ -249,23 +329,27 @@ fn encode_masked_frame(
     }
     let mut frame = vec![0x80 | opcode];
     match payload.len() {
-        length if length <= 125 => frame.push(0x80 | length as u8),
+        length if length <= 125 => frame.push(mask.map_or(0, |_| 0x80) | length as u8),
         length if length <= u16::MAX as usize => {
-            frame.push(0x80 | 126);
+            frame.push(mask.map_or(0, |_| 0x80) | 126);
             frame.extend_from_slice(&(length as u16).to_be_bytes());
         }
         length => {
-            frame.push(0x80 | 127);
+            frame.push(mask.map_or(0, |_| 0x80) | 127);
             frame.extend_from_slice(&(length as u64).to_be_bytes());
         }
     }
-    frame.extend_from_slice(&mask);
-    frame.extend(
-        payload
-            .iter()
-            .enumerate()
-            .map(|(index, byte)| byte ^ mask[index % 4]),
-    );
+    if let Some(mask) = mask {
+        frame.extend_from_slice(&mask);
+        frame.extend(
+            payload
+                .iter()
+                .enumerate()
+                .map(|(index, byte)| byte ^ mask[index % 4]),
+        );
+    } else {
+        frame.extend_from_slice(payload);
+    }
     Ok(frame)
 }
 
@@ -280,9 +364,80 @@ fn checked_inbound_frame_length(length: u64, max_frame_bytes: usize) -> Result<u
     Ok(length)
 }
 
+fn read_http_headers(stream: &mut dyn Read, context: &str) -> Result<String, DurableRunnerError> {
+    let mut headers = Vec::new();
+    let mut byte = [0_u8; 1];
+    while !headers.ends_with(b"\r\n\r\n") {
+        if headers.len() >= MAX_HTTP_HEADER_BYTES {
+            return Err(DurableRunnerError::invalid(format!(
+                "WebSocket {context} headers are too large"
+            )));
+        }
+        stream.read_exact(&mut byte).map_err(|error| {
+            DurableRunnerError::invalid(format!("WebSocket {context} failed: {error}"))
+        })?;
+        headers.push(byte[0]);
+    }
+    String::from_utf8(headers)
+        .map_err(|error| DurableRunnerError::invalid(format!("invalid HTTP headers: {error}")))
+}
+
+fn tls_client_stream(
+    stream: TcpStream,
+    target: &ResolvedWsTarget,
+    ca_bundle_path: Option<&Path>,
+) -> Result<Box<dyn WsReadWrite + Send>, DurableRunnerError> {
+    let mut roots = RootCertStore::empty();
+    let native = rustls_native_certs::load_native_certs();
+    for certificate in native.certs {
+        roots.add(certificate).map_err(|error| {
+            DurableRunnerError::invalid(format!("failed to load a platform trust root: {error}"))
+        })?;
+    }
+    if let Some(path) = ca_bundle_path {
+        let file = File::open(path).map_err(|error| {
+            DurableRunnerError::invalid(format!("failed to open runner CA bundle: {error}"))
+        })?;
+        let mut reader = io::BufReader::new(file);
+        let mut added = 0_usize;
+        for certificate in rustls_pemfile::certs(&mut reader) {
+            roots.add(certificate.map_err(|error| {
+                DurableRunnerError::invalid(format!("runner CA bundle is invalid: {error}"))
+            })?)
+            .map_err(|error| {
+                DurableRunnerError::invalid(format!("runner CA certificate is invalid: {error}"))
+            })?;
+            added += 1;
+        }
+        if added == 0 {
+            return Err(DurableRunnerError::invalid(
+                "runner CA bundle contains no certificates",
+            ));
+        }
+    }
+    if roots.is_empty() {
+        return Err(DurableRunnerError::invalid(
+            "no TLS trust roots are available",
+        ));
+    }
+    let config = ClientConfig::builder()
+        .with_root_certificates(roots)
+        .with_no_client_auth();
+    let server_name = ServerName::try_from(target.host.clone()).map_err(|_| {
+        DurableRunnerError::invalid("runner WSS hostname is invalid for TLS verification")
+    })?;
+    let connection = ClientConnection::new(Arc::new(config), server_name)
+        .map_err(|error| DurableRunnerError::invalid(format!("TLS setup failed: {error}")))?;
+    Ok(Box::new(StreamOwned::new(connection, stream)))
+}
+
 impl WsClient {
-    fn connect(target: &ResolvedWsTarget, max_frame_bytes: usize) -> Result<Self, DurableRunnerError> {
-        let mut stream = TcpStream::connect(target.addresses.as_slice())
+    fn connect(
+        target: &ResolvedWsTarget,
+        max_frame_bytes: usize,
+        ca_bundle_path: Option<&Path>,
+    ) -> Result<Self, DurableRunnerError> {
+        let stream = TcpStream::connect(target.addresses.as_slice())
             .map_err(|error| DurableRunnerError::invalid(format!("WebSocket connect failed: {error}")))?;
         stream
             .set_read_timeout(Some(Duration::from_secs(2)))
@@ -290,6 +445,11 @@ impl WsClient {
         stream
             .set_write_timeout(Some(Duration::from_secs(2)))
             .map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
+        let mut stream: Box<dyn WsReadWrite + Send> = if target.secure {
+            tls_client_stream(stream, target, ca_bundle_path)?
+        } else {
+            Box::new(stream)
+        };
         let mut request = format!(
             "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {}\r\nSec-WebSocket-Version: 13\r\n\r\n",
             target.path,
@@ -306,21 +466,7 @@ impl WsClient {
             DurableRunnerError::invalid(format!("WebSocket upgrade write failed: {error}"))
         })?;
 
-        let mut response = Vec::new();
-        let mut byte = [0_u8; 1];
-        while !response.ends_with(b"\r\n\r\n") {
-            if response.len() >= MAX_HTTP_HEADER_BYTES {
-                return Err(DurableRunnerError::invalid(
-                    "WebSocket response headers are too large",
-                ));
-            }
-            stream.read_exact(&mut byte).map_err(|error| {
-                DurableRunnerError::invalid(format!("WebSocket upgrade response failed: {error}"))
-            })?;
-            response.push(byte[0]);
-        }
-        let response = String::from_utf8(response)
-            .map_err(|error| DurableRunnerError::invalid(format!("invalid HTTP response: {error}")))?;
+        let response = read_http_headers(stream.as_mut(), "upgrade response")?;
         if !response.starts_with("HTTP/1.1 101 ") {
             let status = response
                 .lines()
@@ -341,12 +487,92 @@ impl WsClient {
                 "WebSocket server returned an invalid acceptance proof",
             ));
         }
+        Ok(Self {
+            stream,
+            mask_counter: 1,
+            mask_outbound: true,
+            require_inbound_mask: false,
+            max_frame_bytes,
+            secure_channel: None,
+        })
+    }
+
+    fn accept(
+        mut stream: TcpStream,
+        expected_path: &str,
+        max_frame_bytes: usize,
+    ) -> Result<Self, DurableRunnerError> {
+        stream
+            .set_read_timeout(Some(Duration::from_secs(2)))
+            .map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
+        stream
+            .set_write_timeout(Some(Duration::from_secs(2)))
+            .map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
+        let request = read_http_headers(&mut stream, "upgrade request")?;
+        let mut lines = request.split("\r\n");
+        let request_line = lines.next().unwrap_or_default();
+        if request_line != format!("GET {expected_path} HTTP/1.1") {
+            let _ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nConnection: close\r\nContent-Length: 0\r\n\r\n");
+            return Err(DurableRunnerError::invalid(
+                "runner listener rejected an unrelated HTTP path",
+            ));
+        }
+        let mut websocket_key: Option<&str> = None;
+        let mut upgrade = false;
+        let mut connection_upgrade = false;
+        let mut version = false;
+        for line in lines {
+            let Some((name, value)) = line.split_once(':') else { continue };
+            let value = value.trim();
+            if name.eq_ignore_ascii_case("sec-websocket-key") {
+                websocket_key = Some(value);
+            } else if name.eq_ignore_ascii_case("upgrade") {
+                upgrade = value.eq_ignore_ascii_case("websocket");
+            } else if name.eq_ignore_ascii_case("connection") {
+                connection_upgrade = value
+                    .split(',')
+                    .any(|part| part.trim().eq_ignore_ascii_case("upgrade"));
+            } else if name.eq_ignore_ascii_case("sec-websocket-version") {
+                version = value == "13";
+            } else if name.eq_ignore_ascii_case("sec-websocket-extensions") {
+                let _ = stream.write_all(b"HTTP/1.1 400 Bad Request\r\nConnection: close\r\nContent-Length: 0\r\n\r\n");
+                return Err(DurableRunnerError::invalid(
+                    "runner listener does not support WebSocket compression",
+                ));
+            }
+        }
+        let websocket_key = websocket_key.filter(|value| !value.is_empty()).ok_or_else(|| {
+            DurableRunnerError::invalid("runner listener requires a WebSocket key")
+        })?;
+        if !upgrade || !connection_upgrade || !version {
+            let _ = stream.write_all(b"HTTP/1.1 400 Bad Request\r\nConnection: close\r\nContent-Length: 0\r\n\r\n");
+            return Err(DurableRunnerError::invalid(
+                "runner listener rejected an ordinary HTTP request",
+            ));
+        }
+        let mut hasher = Sha1::new();
+        hasher.update(websocket_key.as_bytes());
+        hasher.update(b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
+        let accept = base64::engine::general_purpose::STANDARD.encode(hasher.finalize());
+        stream
+            .write_all(
+                format!(
+                    "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {accept}\r\n\r\n"
+                )
+                .as_bytes(),
+            )
+            .and_then(|_| stream.flush())
+            .map_err(|error| {
+                DurableRunnerError::invalid(format!("WebSocket upgrade response failed: {error}"))
+            })?;
         stream
             .set_read_timeout(Some(Duration::from_millis(250)))
             .map_err(|error| DurableRunnerError::invalid(error.to_string()))?;
         Ok(Self {
-            stream,
+            stream: Box::new(stream),
             mask_counter: 1,
+            mask_outbound: false,
+            require_inbound_mask: true,
             max_frame_bytes,
             secure_channel: None,
         })
@@ -372,9 +598,12 @@ impl WsClient {
     }
 
     fn send_frame(&mut self, opcode: u8, payload: &[u8]) -> Result<(), DurableRunnerError> {
-        let mask = self.mask_counter.to_be_bytes();
-        self.mask_counter = self.mask_counter.wrapping_add(1);
-        let frame = encode_masked_frame(opcode, payload, mask, self.max_frame_bytes)?;
+        let mask = self.mask_outbound.then(|| {
+            let mask = self.mask_counter.to_be_bytes();
+            self.mask_counter = self.mask_counter.wrapping_add(1);
+            mask
+        });
+        let frame = encode_frame(opcode, payload, mask, self.max_frame_bytes)?;
         self.stream
             .write_all(&frame)
             .and_then(|_| self.stream.flush())
@@ -412,8 +641,18 @@ impl WsClient {
                     )))
                 }
             }
+            if header[0] & 0x80 == 0 || header[0] & 0x70 != 0 {
+                return Err(DurableRunnerError::invalid(
+                    "fragmented or extension WebSocket frames are not supported",
+                ));
+            }
             let opcode = header[0] & 0x0f;
             let masked = header[1] & 0x80 != 0;
+            if masked != self.require_inbound_mask {
+                return Err(DurableRunnerError::invalid(
+                    "WebSocket frame masking direction is invalid",
+                ));
+            }
             let mut length = u64::from(header[1] & 0x7f);
             if length == 126 {
                 let mut extended = [0_u8; 2];
@@ -429,6 +668,11 @@ impl WsClient {
                 length = u64::from_be_bytes(extended);
             }
             let length = checked_inbound_frame_length(length, self.max_frame_bytes)?;
+            if matches!(opcode, 0x8..=0xA) && length > 125 {
+                return Err(DurableRunnerError::invalid(
+                    "WebSocket control frame exceeds 125 bytes",
+                ));
+            }
             let mut mask = [0_u8; 4];
             if masked {
                 self.stream

packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs

@@ -4,6 +4,8 @@ use sha2::{Digest, Sha256};
 use crate::durable::redact_text;
 
 const MAX_OUTPUT_BYTES: usize = 64 * 1024;
+const MAX_TURN_DIFF_FILES: usize = 2_000;
+const MAX_TURN_DIFF_CHARS_PER_FILE: usize = 256 * 1024;
 
 fn text<'a>(value: Option<&'a Value>) -> &'a str {
     value.and_then(Value::as_str).unwrap_or("")
@@ -57,6 +59,161 @@ fn bounded_output(value: &str) -> Value {
     })
 }
 
+#[derive(Default)]
+struct ParsedDiffFile {
+    lines: Vec<String>,
+    old_path: Option<String>,
+    new_path: Option<String>,
+    rename_from: Option<String>,
+    rename_to: Option<String>,
+    additions: u64,
+    deletions: u64,
+    binary: bool,
+    mode_change: bool,
+    in_hunk: bool,
+}
+
+fn git_diff_path(value: &str) -> Option<String> {
+    let trimmed = value.trim();
+    if trimmed == "/dev/null" {
+        return None;
+    }
+    let decoded = if trimmed.starts_with('"') && trimmed.ends_with('"') {
+        serde_json::from_str::<String>(trimmed).ok()?
+    } else {
+        trimmed.to_owned()
+    };
+    let normalized = decoded.replace('\\', "/");
+    let relative = normalized
+        .strip_prefix("a/")
+        .or_else(|| normalized.strip_prefix("b/"))
+        .unwrap_or(&normalized);
+    if relative.is_empty()
+        || relative.starts_with('/')
+        || relative.split('/').any(|part| part == "..")
+    {
+        return None;
+    }
+    Some(relative.chars().take(4096).collect())
+}
+
+fn git_diff_header_token(value: &str) -> Option<(&str, &str)> {
+    let value = value.trim_start();
+    if value.starts_with('"') {
+        let mut escaped = false;
+        for (index, character) in value.char_indices().skip(1) {
+            if escaped {
+                escaped = false;
+            } else if character == '\\' {
+                escaped = true;
+            } else if character == '"' {
+                return Some((&value[..=index], &value[index + 1..]));
+            }
+        }
+        return None;
+    }
+    let end = value.find(char::is_whitespace).unwrap_or(value.len());
+    (end > 0).then_some((&value[..end], &value[end..]))
+}
+
+fn git_diff_header_paths(value: &str) -> (Option<String>, Option<String>) {
+    let Some((old, remaining)) = git_diff_header_token(value) else {
+        return (None, None);
+    };
+    let Some((new, _)) = git_diff_header_token(remaining) else {
+        return (None, None);
+    };
+    (git_diff_path(old), git_diff_path(new))
+}
+
+fn finish_diff_file(current: ParsedDiffFile, files: &mut Vec<Value>) {
+    if files.len() >= MAX_TURN_DIFF_FILES {
+        return;
+    }
+    let path = current
+        .rename_to
+        .as_ref()
+        .or(current.new_path.as_ref())
+        .or(current.old_path.as_ref())
+        .cloned();
+    let Some(path) = path else { return };
+    let previous_path = current
+        .rename_from
+        .clone()
+        .or_else(|| current.rename_to.as_ref().and(current.old_path.clone()));
+    let operation = if current.rename_to.is_some() && previous_path.is_some() {
+        "rename"
+    } else if current.old_path.is_none() {
+        "create"
+    } else if current.new_path.is_none() {
+        "delete"
+    } else if current.mode_change && current.additions == 0 && current.deletions == 0 {
+        "mode_change"
+    } else {
+        "modify"
+    };
+    let patch = format!("{}\n", current.lines.join("\n"));
+    files.push(json!({
+        "path": path,
+        "operation": operation,
+        "previousPath": previous_path,
+        "additions": if current.binary { Value::Null } else { json!(current.additions) },
+        "deletions": if current.binary { Value::Null } else { json!(current.deletions) },
+        "binary": current.binary,
+        "diff": if current.binary { Value::Null } else { json!(patch.chars().take(MAX_TURN_DIFF_CHARS_PER_FILE).collect::<String>()) },
+    }));
+}
+
+fn parse_turn_diff(value: &str) -> Vec<Value> {
+    let mut files = Vec::new();
+    let mut current: Option<ParsedDiffFile> = None;
+    for line in value.lines() {
+        if line.starts_with("diff --git ") {
+            if let Some(previous) = current.take() {
+                finish_diff_file(previous, &mut files);
+            }
+            let (old_path, new_path) = line
+                .strip_prefix("diff --git ")
+                .map(git_diff_header_paths)
+                .unwrap_or((None, None));
+            current = Some(ParsedDiffFile {
+                lines: vec![line.to_owned()],
+                old_path,
+                new_path,
+                ..ParsedDiffFile::default()
+            });
+            continue;
+        }
+        let Some(file) = current.as_mut() else {
+            continue;
+        };
+        file.lines.push(line.to_owned());
+        if let Some(path) = line.strip_prefix("--- ") {
+            file.old_path = git_diff_path(path);
+        } else if let Some(path) = line.strip_prefix("+++ ") {
+            file.new_path = git_diff_path(path);
+        } else if let Some(path) = line.strip_prefix("rename from ") {
+            file.rename_from = git_diff_path(path);
+        } else if let Some(path) = line.strip_prefix("rename to ") {
+            file.rename_to = git_diff_path(path);
+        } else if line.starts_with("old mode ") || line.starts_with("new mode ") {
+            file.mode_change = true;
+        } else if line.starts_with("Binary files ") || line == "GIT binary patch" {
+            file.binary = true;
+        } else if line.starts_with("@@") {
+            file.in_hunk = true;
+        } else if file.in_hunk && line.starts_with('+') && !line.starts_with("+++") {
+            file.additions += 1;
+        } else if file.in_hunk && line.starts_with('-') && !line.starts_with("---") {
+            file.deletions += 1;
+        }
+    }
+    if let Some(previous) = current {
+        finish_diff_file(previous, &mut files);
+    }
+    files
+}
+
 /// Convert native provider notifications into bounded provider-neutral PRP records.
 /// Unknown variants remain on the legacy diagnostic stream and are never interpreted from prose.
 pub fn canonical_provider_events(method: &str, params: &Value) -> Vec<(String, Value, String)> {
@@ -69,18 +226,87 @@ pub fn canonical_provider_events(method: &str, params: &Value) -> Vec<(String, V
         |result: &mut Vec<(String, Value, String)>, event: &str, payload: Value, item_id: &str| {
             result.push((event.to_owned(), payload, item_id.to_owned()))
         };
-    if (method == "item/started" || complete) && kind == "plan" || method == "turn/plan/updated" {
-        let steps: Vec<Value> = params.get("plan").and_then(Value::as_array).map(|values| values.iter().take(256).enumerate().map(|(index, value)| json!({"stepId": format!("step-{}", index + 1), "body": text(value.get("step")).chars().take(4000).collect::<String>(), "status": match text(value.get("status")) { "inProgress" | "in_progress" => "in_progress", "completed" => "completed", "blocked" => "blocked", _ => "pending" }})).collect()).unwrap_or_else(Vec::new);
-        let plan_complete = complete
-            || (!steps.is_empty()
-                && steps
+    if method == "turn/diff/updated" {
+        let turn_id = id(text(params.get("turnId")).trim(), "turn");
+        let patch = text(params.get("diff"));
+        let files = parse_turn_diff(patch);
+        if patch.trim().is_empty() || !files.is_empty() {
+            let known_stats = files.iter().all(|file| {
+                file.get("additions").and_then(Value::as_u64).is_some()
+                    && file.get("deletions").and_then(Value::as_u64).is_some()
+            });
+            let additions = known_stats.then(|| {
+                files
                     .iter()
-                    .all(|step| text(step.get("status")) == "completed"));
+                    .filter_map(|file| file.get("additions").and_then(Value::as_u64))
+                    .sum::<u64>()
+            });
+            let deletions = known_stats.then(|| {
+                files
+                    .iter()
+                    .filter_map(|file| file.get("deletions").and_then(Value::as_u64))
+                    .sum::<u64>()
+            });
+            let file_count = files.len();
+            let change_set_id = format!("{turn_id}:workspace");
+            push(
+                &mut result,
+                "workspace.change.updated",
+                json!({
+                    "schema": "paperclip.workspace.diff.v1",
+                    "changeSetId": change_set_id,
+                    "revision": params.get("revision").and_then(Value::as_u64).unwrap_or(1),
+                    "source": "harness_reported",
+                    "complete": false,
+                    "files": files,
+                    "totals": { "files": file_count, "additions": additions, "deletions": deletions },
+                    "patchArtifactRef": Value::Null,
+                }),
+                &change_set_id,
+            );
+        }
+    } else if method == "turn/plan/updated" {
+        let turn_plan_id = id(text(params.get("turnId")).trim(), "turn-plan");
+        let steps: Vec<Value> = params
+            .get("plan")
+            .and_then(Value::as_array)
+            .map(|values| {
+                values
+                    .iter()
+                    .take(256)
+                    .enumerate()
+                    .filter_map(|(index, value)| {
+                        let body = text(value.get("step"))
+                            .trim()
+                            .chars()
+                            .take(4000)
+                            .collect::<String>();
+                        if body.is_empty() {
+                            return None;
+                        }
+                        Some(json!({
+                            "stepId": format!("step-{}", index + 1),
+                            "body": body,
+                            "status": match text(value.get("status")) {
+                                "inProgress" | "in_progress" => "in_progress",
+                                "completed" => "completed",
+                                "blocked" | "failed" | "error" => "blocked",
+                                _ => "pending",
+                            }
+                        }))
+                    })
+                    .collect()
+            })
+            .unwrap_or_default();
+        let plan_complete = !steps.is_empty()
+            && steps
+                .iter()
+                .all(|step| text(step.get("status")) == "completed");
         push(
             &mut result,
             "plan.updated",
-            json!({"schema":"paperclip.plan.updated.v1","planId":item_id,"revision":params.get("revision").and_then(Value::as_u64).unwrap_or(1),"explanation":params.get("explanation").and_then(Value::as_str),"steps":steps,"complete":plan_complete,"syncStatus":if plan_complete {"pending"} else {"streaming"},"documentRevision":Value::Null}),
-            &item_id,
+            json!({"schema":"paperclip.plan.updated.v1","planId":turn_plan_id,"revision":params.get("revision").and_then(Value::as_u64).unwrap_or(1),"explanation":params.get("explanation").and_then(Value::as_str),"steps":steps,"complete":plan_complete,"syncStatus":"not_applicable","documentRevision":Value::Null}),
+            &turn_plan_id,
         );
     } else if (method == "item/started" || complete)
         && ["commandExecution", "mcpToolCall", "dynamicToolCall"].contains(&kind)
@@ -404,7 +630,46 @@ mod tests {
             ]}),
         );
         assert_eq!(events[0].0, "plan.updated");
+        assert_eq!(events[0].2, "turn-1");
+        assert_eq!(events[0].1["planId"], "turn-1");
         assert_eq!(events[0].1["complete"], true);
-        assert_eq!(events[0].1["syncStatus"], "pending");
+        assert_eq!(events[0].1["syncStatus"], "not_applicable");
+        assert_eq!(events[0].1["documentRevision"], Value::Null);
+    }
+
+    #[test]
+    fn proposed_plan_text_is_not_a_turn_checklist() {
+        let events = canonical_provider_events(
+            "item/completed",
+            &json!({"item":{"id":"proposed-plan-1","type":"plan","text":"Ship it"}}),
+        );
+        assert!(events.is_empty());
+    }
+
+    #[test]
+    fn empty_turn_plan_is_a_non_terminal_clearing_snapshot() {
+        let events =
+            canonical_provider_events("turn/plan/updated", &json!({"turnId":"turn-1","plan":[]}));
+        assert_eq!(events[0].1["steps"], json!([]));
+        assert_eq!(events[0].1["complete"], false);
+    }
+
+    #[test]
+    fn maps_turn_diff_to_a_stable_workspace_snapshot() {
+        let events = canonical_provider_events(
+            "turn/diff/updated",
+            &json!({
+                "turnId":"turn-1",
+                "diff":"diff --git a/src/a.ts b/src/a.ts\n--- a/src/a.ts\n+++ b/src/a.ts\n@@ -1 +1,2 @@\n-old\n+new\n+added\n"
+            }),
+        );
+        assert_eq!(events[0].0, "workspace.change.updated");
+        assert_eq!(events[0].2, "turn-1:workspace");
+        assert_eq!(events[0].1["changeSetId"], "turn-1:workspace");
+        assert_eq!(
+            events[0].1["totals"],
+            json!({"files":1,"additions":2,"deletions":1})
+        );
+        assert_eq!(events[0].1["files"][0]["path"], "src/a.ts");
     }
 }

packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs

@@ -26,6 +26,7 @@ fn codex_plan_mode_is_qualified_and_selected_on_turn_start() {
             args: Vec::new(),
             cwd: "/tmp".to_owned(),
             model: None,
+            approval_policy: "never".to_owned(),
             instructions: "Author a plan without editing files.".to_owned(),
             collaboration_mode: "plan".to_owned(),
             include_collaboration_mode_instructions: true,
@@ -40,6 +41,33 @@ fn codex_plan_mode_is_qualified_and_selected_on_turn_start() {
     provider.shutdown().unwrap();
 }
 
+#[test]
+fn codex_permission_policy_is_pinned_on_thread_start_and_resume() {
+    for policy in ["never", "on-request", "untrusted"] {
+        for resume_thread_id in [None, Some("existing-thread")] {
+            let mut provider = CodexProvider::start(
+                &LocalProviderConfig {
+                    kind: ProviderKind::Codex,
+                    command: PathBuf::from(env!("CARGO_BIN_EXE_fake-codex-app-server")),
+                    args: vec!["--expected-approval-policy".to_owned(), policy.to_owned()],
+                    cwd: "/tmp".to_owned(),
+                    model: None,
+                    approval_policy: policy.to_owned(),
+                    instructions: "Use the configured approval policy.".to_owned(),
+                    collaboration_mode: "default".to_owned(),
+                    include_collaboration_mode_instructions: true,
+                    include_skill_instructions: false,
+                    runtime_context: None,
+                },
+                vec![test_tool()].into_iter(),
+                resume_thread_id,
+            )
+            .unwrap();
+            provider.shutdown().unwrap();
+        }
+    }
+}
+
 #[test]
 fn codex_dynamic_tool_round_trips_through_the_rust_provider_boundary() {
     let mut provider = CodexProvider::start(
@@ -49,6 +77,7 @@ fn codex_dynamic_tool_round_trips_through_the_rust_provider_boundary() {
             args: Vec::new(),
             cwd: "/tmp".to_owned(),
             model: None,
+            approval_policy: "never".to_owned(),
             instructions: "Use the authorized Paperclip tools.".to_owned(),
             collaboration_mode: "default".to_owned(),
             include_collaboration_mode_instructions: true,
@@ -119,6 +148,92 @@ fn codex_dynamic_tool_round_trips_through_the_rust_provider_boundary() {
     }
 }
 
+#[test]
+fn deterministic_fixture_emits_replacing_plan_and_diff_snapshots() {
+    let mut provider = CodexProvider::start(
+        &LocalProviderConfig {
+            kind: ProviderKind::Codex,
+            command: PathBuf::from(env!("CARGO_BIN_EXE_fake-codex-app-server")),
+            args: vec!["--structured-activity".to_owned()],
+            cwd: "/tmp".to_owned(),
+            model: None,
+            approval_policy: "never".to_owned(),
+            instructions: "Emit deterministic structured activity.".to_owned(),
+            collaboration_mode: "default".to_owned(),
+            include_collaboration_mode_instructions: true,
+            include_skill_instructions: false,
+            runtime_context: None,
+        },
+        vec![test_tool()].into_iter(),
+        None,
+    )
+    .unwrap();
+    provider
+        .start_turn("Exercise the fixture.", "/tmp")
+        .unwrap();
+
+    let deadline = Instant::now() + Duration::from_secs(2);
+    let mut plan_revisions = Vec::new();
+    let mut diff_revisions = Vec::new();
+    loop {
+        assert!(
+            Instant::now() < deadline,
+            "fixture did not issue its tool call"
+        );
+        match provider.poll().unwrap() {
+            Some(ProviderEvent::Notification { method, params })
+                if method == "turn/plan/updated" =>
+            {
+                plan_revisions.push(params["revision"].as_u64().unwrap());
+                assert_eq!(params["turnId"], "fake-turn");
+            }
+            Some(ProviderEvent::Notification { method, params })
+                if method == "turn/diff/updated" =>
+            {
+                diff_revisions.push(params["revision"].as_u64().unwrap());
+            }
+            Some(ProviderEvent::ToolCall {
+                call_id,
+                operation_id,
+                ..
+            }) => {
+                provider
+                    .deliver_tool_result(&ToolResult {
+                        call_id,
+                        operation_id,
+                        result: json!({"ok": true}),
+                        is_error: false,
+                    })
+                    .unwrap();
+                break;
+            }
+            _ => {}
+        }
+    }
+
+    let deadline = Instant::now() + Duration::from_secs(2);
+    loop {
+        assert!(Instant::now() < deadline, "fixture did not finish its turn");
+        match provider.poll().unwrap() {
+            Some(ProviderEvent::Notification { method, params })
+                if method == "turn/plan/updated" =>
+            {
+                plan_revisions.push(params["revision"].as_u64().unwrap());
+            }
+            Some(ProviderEvent::Notification { method, params })
+                if method == "turn/diff/updated" =>
+            {
+                diff_revisions.push(params["revision"].as_u64().unwrap());
+            }
+            Some(ProviderEvent::Notification { method, .. }) if method == "turn/completed" => break,
+            _ => {}
+        }
+    }
+
+    assert_eq!(plan_revisions, vec![1, 2, 3]);
+    assert_eq!(diff_revisions, vec![1, 2]);
+}
+
 #[test]
 fn dot_185_runtime_question_round_trips_through_the_rust_provider_boundary() {
     let mut provider = CodexProvider::start(
@@ -128,6 +243,7 @@ fn dot_185_runtime_question_round_trips_through_the_rust_provider_boundary() {
             args: vec!["--runtime-question".to_owned()],
             cwd: "/tmp".to_owned(),
             model: None,
+            approval_policy: "never".to_owned(),
             instructions: "Ask for deployment input.".to_owned(),
             collaboration_mode: "default".to_owned(),
             include_collaboration_mode_instructions: true,
@@ -196,6 +312,7 @@ fn provider_contract_preserves_the_opencode_tag() {
             args: Vec::new(),
             cwd: "/tmp".to_owned(),
             model: Some("openrouter/deepseek/deepseek-v4-flash-0731".to_owned()),
+            approval_policy: "never".to_owned(),
             instructions: "Use the authorized Paperclip tools.".to_owned(),
             collaboration_mode: "default".to_owned(),
             include_collaboration_mode_instructions: false,
@@ -226,6 +343,7 @@ fn image_like_prompt_stays_skillless_and_accepts_a_valid_one_point_two_megabyte_
             args: vec!["--large-event".to_owned()],
             cwd: "/tmp".to_owned(),
             model: None,
+            approval_policy: "never".to_owned(),
             instructions: "Use only the authorized Paperclip tools.".to_owned(),
             collaboration_mode: "default".to_owned(),
             include_collaboration_mode_instructions: true,
@@ -288,6 +406,7 @@ fn codex_provider_rejects_an_event_above_the_four_megabyte_hard_limit() {
             args: vec!["--oversized-event".to_owned()],
             cwd: "/tmp".to_owned(),
             model: None,
+            approval_policy: "never".to_owned(),
             instructions: "Use only the authorized Paperclip tools.".to_owned(),
             collaboration_mode: "default".to_owned(),
             include_collaboration_mode_instructions: true,