diff --git a/packages/paperclip-runner/runner/Cargo.lock b/packages/paperclip-runner/runner/Cargo.lock index 5959542eede..5806ca57bd7 100644 --- a/packages/paperclip-runner/runner/Cargo.lock +++ b/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" diff --git a/packages/paperclip-runner/runner/Cargo.toml b/packages/paperclip-runner/runner/Cargo.toml index 71506a0efef..41ea065df0f 100644 --- a/packages/paperclip-runner/runner/Cargo.toml +++ b/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" diff --git a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml b/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml index e51e4bd9e60..cbc652c8271 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/Cargo.toml +++ b/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" diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider.rs index a30bf98bd91..60adb959c2d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider.rs +++ b/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 = 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::(), + "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"), + } + } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 077f648e502..6119c454d8c 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/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, } fn send(value: Value) -> io::Result<()> { @@ -28,6 +31,12 @@ fn run() -> Result<(), Box> { 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 = None; @@ -54,6 +63,17 @@ fn run() -> Result<(), Box> { "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> { .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> { "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> { .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": { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index fe860ba89e8..0998187abdf 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/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) -> 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"]) + ); } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index 2aa382e6eb6..98becf5f1f4 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -259,6 +259,8 @@ pub struct LocalProviderConfig { pub args: Vec, pub cwd: String, pub model: Option, + #[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, } +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 { .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, resume_thread_id: Option<&str>, ) -> Result { + 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( diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index 760707a7600..e01d287443d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/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 = None; let mut connection_lease_revocation_epoch: Option = None; let started = Instant::now(); + let mut reconnect_attempt = 0_u32; + let mut authenticated_once = false; + let mut disconnected_since: Option = 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"); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 6f474a8024e..69cbc62062f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/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, DurableRunn #[derive(Clone, Debug)] pub struct DurableRunnerConfig { pub connect_url: String, + pub ca_bundle_path: Option, 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, pub max_runtime: Duration, pub lifecycle_mode: String, pub idle_timeout: Option, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index ccb3292eacd..04bd4ba2e39 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/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 { - 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 { .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::() .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 { 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 { } struct ResolvedWsTarget { + secure: bool, + host: String, authority: String, path: String, addresses: Vec, @@ -80,6 +98,61 @@ impl ResolvedWsTarget { } } +enum RunnerTransportEndpoint { + Dial(String), + Listen { listener: TcpListener, path: String }, +} + +impl RunnerTransportEndpoint { + fn new(input: &str) -> Result { + 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, 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(input: &str, resolver: F) -> Result where F: FnOnce(&str, u16) -> io::Result>, @@ -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 WsReadWrite for T {} + struct WsClient { - stream: TcpStream, + stream: Box, mask_counter: u32, + mask_outbound: bool, + require_inbound_mask: bool, max_frame_bytes: usize, secure_channel: Option, } @@ -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, 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 Result { + 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, 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 { - let mut stream = TcpStream::connect(target.addresses.as_slice()) + fn connect( + target: &ResolvedWsTarget, + max_frame_bytes: usize, + ca_bundle_path: Option<&Path>, + ) -> Result { + 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 = 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 { + 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 diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs index 2b273383b4f..e2e68fedaa3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_events.rs +++ b/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, + old_path: Option, + new_path: Option, + rename_from: Option, + rename_to: Option, + additions: u64, + deletions: u64, + binary: bool, + mode_change: bool, + in_hunk: bool, +} + +fn git_diff_path(value: &str) -> Option { + let trimmed = value.trim(); + if trimmed == "/dev/null" { + return None; + } + let decoded = if trimmed.starts_with('"') && trimmed.ends_with('"') { + serde_json::from_str::(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, Option) { + 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) { + 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::()) }, + })); +} + +fn parse_turn_diff(value: &str) -> Vec { + let mut files = Vec::new(); + let mut current: Option = 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 = 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::(), "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::() + }); + let deletions = known_stats.then(|| { + files + .iter() + .filter_map(|file| file.get("deletions").and_then(Value::as_u64)) + .sum::() + }); + 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 = 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::(); + 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"); } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 691b47cffa4..749ca17b3ea 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/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,