diff --git a/packages/paperclip-runner/runner/Cargo.toml b/packages/paperclip-runner/runner/Cargo.toml index 6a843e95b34..40b0ac1b1c4 100644 --- a/packages/paperclip-runner/runner/Cargo.toml +++ b/packages/paperclip-runner/runner/Cargo.toml @@ -33,7 +33,7 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std" rustls-native-certs = "0.8" rustls-pemfile = "2.2" reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] } -rustix = { version = "1.1", features = ["fs", "net"] } +rustix = { version = "1.1", features = ["fs", "net", "process"] } sha1 = "0.10" tungstenite = { version = "0.28", default-features = false, features = ["handshake", "rustls-tls-native-roots"] } toml_edit = "0.22" diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 35c816f46c2..68b4d15db5b 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -41,6 +41,13 @@ source sequence the runner has produced; acknowledged prefixes are removed atomically from durable state. After disconnect, every remaining event is sent again with the same identity and source sequence. +If Codex hits its tool-receipt limit during reconnect, runnerd stops its local +process group at the original deadline. It reads the remaining stdout in short +intervals, then replays the frames in order across as many batches as needed. +Only confirmed EOF with no terminal permits a missing-terminal result. +If stdout stays open or the buffer lacks room for one maximum-size frame, +runnerd reports uncertainty and keeps the events already committed to disk. + Executors retain polled events until runnerd acknowledges each event after its outbox commit. Batches commit one event at a time, so a later oversized event or capacity failure cannot roll back the accepted prefix or discard the diff --git a/packages/paperclip-runner/runner/crates/runner-core/examples/recovery-fixture.rs b/packages/paperclip-runner/runner/crates/runner-core/examples/recovery-fixture.rs index abdd1785e60..48805b685b0 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/examples/recovery-fixture.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/examples/recovery-fixture.rs @@ -3,7 +3,7 @@ use std::fs; use std::path::PathBuf; use std::sync::{Arc, Condvar, Mutex}; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use paperclip_runner_core::durable::{ capture_bootstrap_ticket, run_durable_runner, Command, CommandExecution, CommandExecutor, @@ -23,6 +23,10 @@ struct Script { receipt_file: PathBuf, #[serde(default)] accounting: bool, + #[serde(default)] + provider_stop_ms: Option, + #[serde(default)] + reconnect_delay_ms: Option, } #[derive(Default)] @@ -129,6 +133,8 @@ fn workload(script: Script, shared: Shared) { struct ScriptedExecutor { shared: Shared, worker: Option>, + stop_after: Option, + stop_at: Option, } impl ScriptedExecutor { @@ -137,6 +143,8 @@ impl ScriptedExecutor { let worker_shared = shared.clone(); Self { shared, + stop_after: script.provider_stop_ms.map(Duration::from_millis), + stop_at: None, worker: Some(thread::spawn(move || workload(script, worker_shared))), } } @@ -151,6 +159,7 @@ impl CommandExecutor for ScriptedExecutor { return Err(DurableRunnerError::invalid("fixture turn started twice")); } work.started = true; + self.stop_at = self.stop_after.map(|delay| Instant::now() + delay); } "turn.steer" => work.resumed = true, "session.snapshot" => {} @@ -170,6 +179,14 @@ impl CommandExecutor for ScriptedExecutor { self.retained_events() } + fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { + if self.stop_at.is_some_and(|due| due <= Instant::now()) { + self.shutdown()?; + self.stop_at = None; + } + Ok(self.stop_at) + } + fn retained_events(&mut self) -> Result, DurableRunnerError> { Ok(self .shared @@ -247,7 +264,7 @@ fn main() -> Result<(), Box> { max_outbox_bytes: value("--max-outbox-bytes")?.parse()?, p0_reserve_bytes: value("--p0-reserve-bytes")?.parse()?, max_frame_bytes: 1024 * 1024, - reconnect_delay: Duration::from_millis(20), + reconnect_delay: Duration::from_millis(script.reconnect_delay_ms.unwrap_or(20)), reconnect_grace: Some(Duration::from_millis( value("--reconnect-grace-ms")?.parse()?, )), 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 deb686987b6..8cfffc68aa2 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 @@ -835,6 +835,13 @@ pub struct CodexProvider { exit_drain: Option, } +#[derive(Debug)] +pub(crate) struct FencedProviderOutput { + pub stdout_drained: bool, + pub output_empty: bool, + pub next_drain_deadline: Option, +} + struct ProviderExitDrain { process_generation: u64, deadline: std::time::Instant, @@ -2482,7 +2489,7 @@ impl CodexProvider { } else { json!({"id": rpc_id, "error": {"code": -32000, "message": message}}) }; - self.send_frame(&response)?; + self.send_courtesy_response(&response)?; if self.notification_identity_diagnostics >= 32 { return Ok(None); } @@ -2521,7 +2528,7 @@ impl CodexProvider { // The terminal notification is already authoritative and may be // waiting in the durable outbox. A courtesy rejection must not turn a // provider that has closed stdin into a fatal polling error. - let _ = self.send_frame(&response); + let _ = self.send_courtesy_response(&response); Ok(Some(CodexProviderEvent::Notification { method: "warning".to_owned(), params: json!({"message": message, "providerMethod": bounded_method(method)}), @@ -2829,7 +2836,7 @@ impl CodexProvider { "Codex tool name", )?; if !self.authorized_tool_ids.contains(&operation_id) { - self.send_frame(&json!({ + self.send_courtesy_response(&json!({ "id": rpc_id, "result": codex_tool_failure("Paperclip did not authorize this tool for the run"), }))?; @@ -3016,7 +3023,7 @@ impl CodexProvider { if self.pending_runtime_requests.len() >= MAX_PENDING_RUNTIME_REQUESTS || retained_request_bytes.is_none() { - self.send_frame(&json!({ + self.send_courtesy_response(&json!({ "id": rpc_id, "error": { "code": -32000, @@ -3059,7 +3066,7 @@ impl CodexProvider { question_set, })); } - self.send_frame(&json!({ + self.send_courtesy_response(&json!({ "id": rpc_id, "error": {"code": -32601, "message": "provider request is unavailable in this runner layer"}, }))?; @@ -3302,6 +3309,21 @@ impl CodexProvider { } pub fn deliver_tool_result(&mut self, result: &ToolResult) -> Result<(), LocalRunnerError> { + self.deliver_tool_result_with_courtesy(result, false) + } + + pub(crate) fn deliver_tool_rejection( + &mut self, + result: &ToolResult, + ) -> Result<(), LocalRunnerError> { + self.deliver_tool_result_with_courtesy(result, true) + } + + fn deliver_tool_result_with_courtesy( + &mut self, + result: &ToolResult, + courtesy_rejection: bool, + ) -> Result<(), LocalRunnerError> { let pending = self .pending_tool_requests .get(&result.call_id) @@ -3324,13 +3346,18 @@ impl CodexProvider { } let text = String::from_utf8(result_bytes) .expect("serde_json always serializes JSON values as valid UTF-8"); - self.send_frame(&json!({ + let response = json!({ "id": pending.rpc_id, "result": { "success": !result.is_error, "contentItems": [{"type": "inputText", "text": text}], }, - }))?; + }); + if courtesy_rejection && result.is_error { + self.send_courtesy_response(&response)?; + } else { + self.send_frame(&response)?; + } if let Some(completed) = self.pending_tool_requests.remove(&result.call_id) { self.pending_tool_request_bytes = self .pending_tool_request_bytes @@ -3354,6 +3381,92 @@ impl CodexProvider { } } + #[cfg(test)] + pub(crate) fn process_exited(&mut self) -> Result { + self.process.try_wait().map(|exit| exit.is_some()) + } + + /// Fence only the already-owned local process. Keep every buffered frame + /// for connected interpretation; no courtesy response or provider RPC runs. + /// Reserve one maximum frame before reading; a full FIFO cannot certify EOF. + pub(crate) fn fence_local_receipt_limit( + &mut self, + ) -> Result { + if self.config.provider != "codex" { + return Err(LocalRunnerError::invalid( + "local receipt-limit fence requires Codex", + )); + } + self.expected_shutdown = true; + self.process.terminate_group_immediately()?; + if self + .exit_drain + .as_ref() + .is_none_or(|drain| drain.process_generation != self.process_generation) + { + self.exit_drain = Some(ProviderExitDrain { + process_generation: self.process_generation, + deadline: std::time::Instant::now() + self.process.shutdown_grace(), + timed_out: false, + warning_emitted: false, + }); + } + let tick = std::time::Instant::now() + Duration::from_millis(5); + while !self.process.stdout_drained() && std::time::Instant::now() < tick { + if self + .pending_messages + .len() + .saturating_add(self.deferred_ambiguous_messages.len()) + >= MAX_BUFFERED_MESSAGES + || self.pending_message_bytes + > MAX_BUFFERED_MESSAGE_BYTES.saturating_sub(CODEX_APP_SERVER_MAX_FRAME_BYTES) + { + return Err(LocalRunnerError::invalid( + "Fenced Codex output reached its bounded raw buffer; EOF remains unverified", + )); + } + let wait = Duration::from_millis(1) + .min(tick.saturating_duration_since(std::time::Instant::now())); + let Some(line) = self.process.receive_stdout_line(wait)? else { + break; + }; + let trace_frame_id = self.trace_inbound(&line); + let message = parse_provider_message(&line)?; + let bytes = json_size(&message, "buffered Codex provider message")?; + self.pending_message_bytes = + retain_buffered_message_bytes(self.pending_message_bytes, bytes).ok_or_else( + || { + LocalRunnerError::invalid( + "Codex buffered messages exceed the 16 MiB aggregate limit", + ) + }, + )?; + self.pending_messages.push_back(BufferedProviderMessage { + value: message, + trace_frame_id, + }); + } + let drain = self + .exit_drain + .as_ref() + .expect("fenced process has a drain deadline"); + let stdout_drained = !self.process.stdout_failed() && self.process.stdout_drained(); + if drain.timed_out || !stdout_drained && std::time::Instant::now() >= drain.deadline { + return Err(LocalRunnerError::invalid("Fenced Codex stdout did not reach verified EOF within shutdown grace; output requires reconciliation")); + } + Ok(FencedProviderOutput { + stdout_drained, + output_empty: stdout_drained + && self.pending_messages.is_empty() + && self.deferred_ambiguous_messages.is_empty(), + next_drain_deadline: (!stdout_drained).then_some( + drain + .deadline + .min(std::time::Instant::now() + Duration::from_millis(25)), + ), + }) + } + pub fn shutdown(&mut self) -> Result<(), LocalRunnerError> { self.shutdown_with_cursor_cloud_delivery(|_| Ok(false)) } @@ -3439,7 +3552,7 @@ impl CodexProvider { } else { json!({"answers": {}}) }; - if let Err(error) = self.send_frame(&json!({ + if let Err(error) = self.send_courtesy_response(&json!({ "id": request.rpc_id, "result": result, })) { @@ -3447,7 +3560,7 @@ impl CodexProvider { } } for request in pending.into_values() { - if let Err(error) = self.send_frame(&json!({ + if let Err(error) = self.send_courtesy_response(&json!({ "id": request.rpc_id, "result": codex_tool_failure("Paperclip stopped the active provider turn"), })) { @@ -3465,6 +3578,18 @@ impl CodexProvider { .map_err(ProviderRequestError::into_inner) } + fn send_courtesy_response(&mut self, value: &Value) -> Result<(), LocalRunnerError> { + // A reply cannot affect a deliberately stopped local conversation. + // Commands and authoritative results must still use strict delivery. + if self.config.provider == "codex" + && self.expected_shutdown + && self.process.try_wait()?.is_some() + { + return Ok(()); + } + self.send_frame(value) + } + fn send_frame(&mut self, value: &Value) -> Result<(), LocalRunnerError> { if let Some(trace) = self.trace.as_mut() { let raw = serde_json::to_vec(value).unwrap_or_default(); @@ -4796,6 +4921,125 @@ done provider } + #[cfg(unix)] + #[test] + fn fenced_tool_rejection_keeps_identity_and_real_delivery_strict() { + let mut provider = completion_tail_provider(); + provider.expected_shutdown = false; + provider.pending_tool_requests.insert( + "denied-call".to_owned(), + PendingToolRequest { + rpc_id: json!(91), + operation_id: "denied-operation".to_owned(), + input: json!({}), + retained_bytes: 0, + }, + ); + let mut rejection = ToolResult { + call_id: "denied-call".to_owned(), + operation_id: "denied-operation".to_owned(), + result: json!({"error": "denied"}), + is_error: true, + }; + provider.process.terminate_group_immediately().unwrap(); + assert!(provider.deliver_tool_rejection(&rejection).is_err()); + provider.fence_local_receipt_limit().unwrap(); + assert!(provider.deliver_tool_result(&rejection).is_err()); + assert!(provider.request("thread/read", json!({})).is_err()); + provider.config.provider = "cursor_cloud".to_owned(); + assert!(provider.deliver_tool_rejection(&rejection).is_err()); + provider.config.provider = "codex".to_owned(); + rejection.operation_id = "foreign-operation".to_owned(); + assert!(provider.deliver_tool_rejection(&rejection).is_err()); + rejection.operation_id = "denied-operation".to_owned(); + rejection.is_error = false; + assert!(provider.deliver_tool_rejection(&rejection).is_err()); + rejection.is_error = true; + provider.deliver_tool_rejection(&rejection).unwrap(); + assert!(provider.pending_tool_requests.is_empty()); + assert!(provider.completed_tool_call_ids.contains("denied-call")); + } + + #[cfg(unix)] + #[test] + fn fenced_receipt_output_keeps_prefix_when_stdout_never_closes() { + let mut provider = completion_tail_provider(); + let (sender, receiver) = mpsc::channel(); + let _original = provider.process.replace_output_receiver_for_test(receiver); + let prefix = json!({"method":"configWarning","params":{"message":"retained prefix"}}); + sender + .send(ProcessOutput::Stdout(prefix.to_string())) + .unwrap(); + let status = provider.fence_local_receipt_limit().unwrap(); + assert!(!status.stdout_drained && !status.output_empty); + assert!( + status.next_drain_deadline.unwrap() + <= std::time::Instant::now() + Duration::from_millis(25) + ); + assert!( + status.next_drain_deadline.unwrap() <= provider.exit_drain.as_ref().unwrap().deadline + ); + let bytes = provider.pending_message_bytes; + assert_eq!(provider.pending_messages.front().unwrap().value, prefix); + provider.exit_drain.as_mut().unwrap().deadline = std::time::Instant::now(); + assert!(provider + .fence_local_receipt_limit() + .unwrap_err() + .to_string() + .contains("EOF")); + assert_eq!(provider.pending_messages.front().unwrap().value, prefix); + assert_eq!(provider.pending_message_bytes, bytes); + } + + #[cfg(unix)] + #[test] + fn fenced_receipt_output_reports_capacity_without_dropping_known_frames() { + for (count, text_bytes, eof_waiting) in [ + (MAX_BUFFERED_MESSAGES, 1, false), + (MAX_BUFFERED_MESSAGES, 1, true), + (4, 13 * 1024 * 1024 / 4, false), + ] { + let mut provider = completion_tail_provider(); + let (sender, receiver) = mpsc::channel(); + let _original = provider.process.replace_output_receiver_for_test(receiver); + // A full FIFO cannot consume a possibly unstoreable next frame, + // even if this controlled reader has only EOF waiting. + if eof_waiting { + sender.send(ProcessOutput::StdoutClosed).unwrap(); + } + for index in 0..count { + let value = json!({"method":"configWarning","params":{"index":index,"message":"x".repeat(text_bytes)}}); + provider.pending_message_bytes += json_size(&value, "fixture").unwrap(); + provider + .pending_messages + .push_back(BufferedProviderMessage { + value, + trace_frame_id: Some(index as u64), + }); + } + let bytes = provider.pending_message_bytes; + assert!(provider + .fence_local_receipt_limit() + .unwrap_err() + .to_string() + .contains("bounded raw buffer")); + assert_eq!(provider.pending_messages.len(), count); + assert_eq!(provider.pending_message_bytes, bytes); + assert_eq!( + provider.pending_messages.front().unwrap().trace_frame_id, + Some(0) + ); + assert_eq!( + provider.pending_messages.back().unwrap().trace_frame_id, + Some((count - 1) as u64) + ); + assert_eq!( + provider.pending_messages.back().unwrap().value["params"]["index"], + count - 1 + ); + } + } + #[cfg(unix)] #[test] fn warm_attachment_ignores_an_unknown_notification_that_names_no_other_turn() { 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 d399b88db34..e25a3e209c6 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 @@ -49,7 +49,7 @@ enum CommandLifecycle { const TERMINAL_RESULT_ACK_TIMEOUT: Duration = Duration::from_secs(2); const CUMULATIVE_ACK_PERSIST_INTERVAL: usize = 16; -fn sleep_for_reconnect(base: Duration, max_delay: Duration, attempt: &mut u32) { +fn reconnect_delay(base: Duration, max_delay: Duration, attempt: &mut u32) -> Duration { let multiplier = 1_u128 << (*attempt).min(5); let uncapped = base.as_millis().saturating_mul(multiplier); let capped = uncapped.clamp(1, 5_000) as u64; @@ -58,23 +58,14 @@ fn sleep_for_reconnect(base: Duration, max_delay: Duration, attempt: &mut u32) { .map_or(0, |duration| u64::from(duration.subsec_nanos())); let jitter_percent = 75 + nanos % 51; *attempt = attempt.saturating_add(1); - let delay = Duration::from_millis(capped.saturating_mul(jitter_percent) / 100).min(max_delay); - if !delay.is_zero() { - thread::sleep(delay); - } -} - -fn sleep_before_deadline(delay: Duration, deadline: Instant) { - let bounded_delay = delay.min(deadline.saturating_duration_since(Instant::now())); - if !bounded_delay.is_zero() { - thread::sleep(bounded_delay); - } + Duration::from_millis(capped.saturating_mul(jitter_percent) / 100).min(max_delay) } fn connection_attempt_deadline( config: &DurableRunnerConfig, started: Instant, disconnected_since: Option, + lease_expires_at_unix_ms: Option, ) -> Instant { let now = Instant::now(); // Bound each connection/auth attempt independently of a productive @@ -91,6 +82,11 @@ fn connection_attempt_deadline( remaining = remaining.min(grace.saturating_sub(now.saturating_duration_since(disconnected_at))); } + if let (Some(expires_at), Ok(now_unix_ms)) = (lease_expires_at_unix_ms, current_unix_ms()) { + remaining = remaining.min(Duration::from_millis( + expires_at.saturating_sub(now_unix_ms), + )); + } now + remaining } @@ -333,6 +329,12 @@ pub trait CommandExecutor { Ok(Vec::new()) } + /// Advance already-owned local cleanup without provider RPCs or ordinary intake. + /// A pending deadline bounds both controller backoff and authentication. + fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { + Ok(None) + } + fn retained_runner_events(&mut self) -> Result, DurableRunnerError> { Ok(Vec::new()) } @@ -498,6 +500,7 @@ pub fn run_durable_runner( let mut authenticated_once = false; let mut disconnected_since: Option = None; let mut reconnect_attempt = 0_u32; + let mut next_connection_attempt = Instant::now(); loop { if authenticated_once { @@ -546,8 +549,42 @@ pub fn run_durable_runner( )); } + let provider_deadline = if authenticated_once + && state.warm_transition.is_none() + && state.pending_terminal_delivery.is_none() + && state.pending_provider_cleanup.is_none() + { + executor.maintain_disconnected_provider()? + } else { + None + }; + if Instant::now() < next_connection_attempt { + let wake_at = next_connection_attempt.min(connection_attempt_deadline( + &config, + started, + disconnected_since, + lease + .as_ref() + .map(|credential| credential.expires_at_unix_ms), + )); + let wake_at = provider_deadline.map_or(wake_at, |due| due.min(wake_at)); + let remaining = wake_at.saturating_duration_since(Instant::now()); + if !remaining.is_zero() { + thread::sleep(remaining); + } + continue; + } let using_bootstrap = lease.is_none(); - let connect_deadline = connection_attempt_deadline(&config, started, disconnected_since); + let connect_deadline = connection_attempt_deadline( + &config, + started, + disconnected_since, + lease + .as_ref() + .map(|credential| credential.expires_at_unix_ms), + ); + let connect_deadline = + provider_deadline.map_or(connect_deadline, |due| due.min(connect_deadline)); let connection = AuthenticatedTransport::connect( &endpoint, &config, @@ -559,7 +596,7 @@ pub fn run_durable_runner( let (mut transport, welcome) = match connection { Ok(Some(connection)) => connection, Ok(None) => { - sleep_before_deadline(config.reconnect_delay, connect_deadline); + next_connection_attempt = Instant::now() + config.reconnect_delay; continue; } Err(error) => { @@ -576,11 +613,12 @@ pub fn run_durable_runner( "bootstrap connection failed closed; provide a fresh one-use ticket", )); } - sleep_for_reconnect( - config.reconnect_delay, - connect_deadline.saturating_duration_since(Instant::now()), - &mut reconnect_attempt, - ); + next_connection_attempt = Instant::now() + + reconnect_delay( + config.reconnect_delay, + connect_deadline.saturating_duration_since(Instant::now()), + &mut reconnect_attempt, + ); continue; } }; @@ -863,9 +901,18 @@ pub fn run_durable_runner( disconnected_since.get_or_insert_with(Instant::now); state.reconnect_count = state.reconnect_count.saturating_add(1); store.save(&state)?; - let reconnect_deadline = - connection_attempt_deadline(&config, started, disconnected_since); - sleep_before_deadline(config.reconnect_delay, reconnect_deadline); + let reconnect_deadline = connection_attempt_deadline( + &config, + started, + disconnected_since, + lease + .as_ref() + .map(|credential| credential.expires_at_unix_ms), + ); + next_connection_attempt = Instant::now() + + config + .reconnect_delay + .min(reconnect_deadline.saturating_duration_since(Instant::now())); continue; } loop { @@ -1167,8 +1214,18 @@ pub fn run_durable_runner( } } disconnected_since.get_or_insert_with(Instant::now); - let reconnect_deadline = connection_attempt_deadline(&config, started, disconnected_since); - sleep_before_deadline(config.reconnect_delay, reconnect_deadline); + let reconnect_deadline = connection_attempt_deadline( + &config, + started, + disconnected_since, + lease + .as_ref() + .map(|credential| credential.expires_at_unix_ms), + ); + next_connection_attempt = Instant::now() + + config + .reconnect_delay + .min(reconnect_deadline.saturating_duration_since(Instant::now())); } } @@ -2368,14 +2425,18 @@ mod tests { config.validate().unwrap(); let started = Instant::now() - Duration::from_secs(21 * 24 * 60 * 60); let before = Instant::now(); - let deadline = connection_attempt_deadline(&config, started, None); + let deadline = connection_attempt_deadline(&config, started, None, None); assert!(deadline >= before + Duration::from_secs(29)); assert!(deadline <= Instant::now() + Duration::from_secs(30)); config.reconnect_grace = Some(Duration::from_secs(5)); - let deadline = connection_attempt_deadline(&config, started, Some(Instant::now())); + let deadline = connection_attempt_deadline(&config, started, Some(Instant::now()), None); assert!(deadline <= Instant::now() + Duration::from_secs(5)); + let lease_expires_at = current_unix_ms().unwrap() + 2_000; + let deadline = connection_attempt_deadline(&config, started, None, Some(lease_expires_at)); + assert!(deadline <= Instant::now() + Duration::from_secs(2)); + assert!(connection_attempt_deadline(&config, started, None, Some(1)) <= Instant::now()); config.max_runtime = Duration::from_secs(7 * 24 * 60 * 60); - assert!(connection_attempt_deadline(&config, started, None) <= Instant::now()); + assert!(connection_attempt_deadline(&config, started, None, None) <= Instant::now()); config.max_runtime = Duration::from_secs(30 * 24 * 60 * 60); config.validate().unwrap(); } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs index 6949763f04a..0de760471cb 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs @@ -1,4 +1,5 @@ use std::path::PathBuf; +use std::time::Instant; use serde_json::Value; @@ -60,6 +61,13 @@ impl CommandExecutor for SelectedExecutor { } } + fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { + match self { + Self::LocalFacade(executor) => executor.maintain_disconnected_provider(), + _ => Ok(None), + } + } + fn rotate_authority(&mut self, config: &DurableRunnerConfig) { match self { Self::LocalFacade(executor) => executor.rotate_authority(config), @@ -392,6 +400,12 @@ impl CommandExecutor for NativeProviderCommandExecutor { .map_or(events, |event| vec![event])) } + fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { + self.selected + .as_mut() + .map_or(Ok(None), CommandExecutor::maintain_disconnected_provider) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { // A provider that has not yet been selected/restored cannot have // in-process cleanup to advance. Never launch one merely for ACK debt. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs index 93a45779118..81488ca64d4 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/process_supervisor.rs @@ -1268,6 +1268,77 @@ impl SupervisedProcess { let _ = self.child.kill(); self.wait() } + + /// Fence this owned process group when an existing deadline leaves no grace. + pub fn terminate_group_immediately(&mut self) -> Result { + if self.finished { + return self.wait(); + } + self.stdin.take(); + #[cfg(unix)] + let signal = { + use rustix::process::{kill_process_group, Pid, Signal}; + let group = i32::try_from(self.process_group_id) + .ok() + .and_then(Pid::from_raw) + .ok_or_else(|| { + LocalRunnerError::invalid("owned process group identity is invalid") + })?; + match kill_process_group(group, Signal::KILL) { + Ok(()) => Ok(()), + Err(rustix::io::Errno::SRCH) => { + if let Some(exit) = self.try_wait()? { + self.finished = true; + return Ok(exit); + } + Err(LocalRunnerError::invalid( + "owned process group disappeared before leader exit was observed", + )) + } + Err(error) => Err(LocalRunnerError::invalid(format!( + "failed to signal owned process group: {error}" + ))), + } + }; + #[cfg(not(unix))] + let signal = { + if let Some(exit) = self.try_wait()? { + self.finished = true; + return Ok(exit); + } + let signal = self.child.kill().map_err(|error| { + LocalRunnerError::invalid(format!("failed to stop owned process: {error}")) + }); + if signal.is_err() { + if let Some(exit) = self.try_wait()? { + self.finished = true; + return Ok(exit); + } + } + signal + }; + self.reap_after_immediate_signal(signal) + } + + fn reap_after_immediate_signal( + &mut self, + signal: Result<(), LocalRunnerError>, + ) -> Result { + signal?; + let deadline = Instant::now() + Duration::from_millis(100); + loop { + if let Some(exit) = self.try_wait()? { + self.finished = true; + return Ok(exit); + } + if Instant::now() >= deadline { + return Err(LocalRunnerError::invalid( + "owned process termination remains unconfirmed", + )); + } + thread::sleep(Duration::from_millis(1)); + } + } } impl Drop for SupervisedProcess { @@ -1317,6 +1388,83 @@ mod tests { use super::*; use std::time::{SystemTime, UNIX_EPOCH}; + #[test] + #[cfg(unix)] + fn immediate_group_fence_does_not_wait_for_shutdown_grace() { + let mut process = SupervisedProcess::spawn( + Path::new("/bin/sh"), + &[ + "-c".to_owned(), + "trap '' TERM; printf 'ready\\n'; while :; do read -r line; done".to_owned(), + ], + Duration::from_secs(2), + 1024, + ) + .unwrap(); + assert_eq!( + process + .receive_stdout_line(Duration::from_secs(1)) + .unwrap() + .as_deref(), + Some("ready") + ); + let started = Instant::now(); + let exit = process.terminate_group_immediately().unwrap(); + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(!exit.success); + assert_eq!( + process.terminate_group_immediately().unwrap().exit_code, + exit.exit_code + ); + } + + #[test] + #[cfg(unix)] + fn immediate_group_fence_accepts_observed_leader_exit_and_missing_group() { + let mut process = SupervisedProcess::spawn( + Path::new("/bin/sh"), + &["-c".to_owned(), "exit 0".to_owned()], + Duration::from_secs(2), + 1024, + ) + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while process.try_wait().unwrap().is_none() { + assert!(Instant::now() < deadline); + thread::sleep(Duration::from_millis(1)); + } + assert!(!process.finished); + assert!(process.terminate_group_immediately().unwrap().success); + assert!(process.finished); + } + + #[test] + #[cfg(unix)] + fn failed_immediate_signal_does_not_wait_or_claim_process_exit() { + let mut process = SupervisedProcess::spawn( + Path::new("/bin/sh"), + &[ + "-c".to_owned(), + "trap '' TERM; printf 'ready\\n'; while :; do read -r line; done".to_owned(), + ], + Duration::from_secs(2), + 1024, + ) + .unwrap(); + assert!(process + .receive_stdout_line(Duration::from_secs(1)) + .unwrap() + .is_some()); + let started = Instant::now(); + assert!(process + .reap_after_immediate_signal(Err(LocalRunnerError::invalid("injected signal failure"))) + .is_err()); + assert!(started.elapsed() < Duration::from_secs(1)); + assert!(!process.finished); + assert!(process.try_wait().unwrap().is_none()); + process.terminate_group_immediately().unwrap(); + } + #[test] #[cfg(unix)] fn stdout_eof_is_sticky_across_consumers_while_stderr_remains_open() { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index bbcdcadde9e..5dab3b081c6 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -4232,7 +4232,7 @@ impl CodexCommandExecutor { self.provider .as_mut() .expect("provider remains present while rejecting its tool call") - .deliver_tool_result(&rejection) + .deliver_tool_rejection(&rejection) .map_err(|delivery_error| { DurableRunnerError::invalid(format!( "failed to return the semantic tool rejection: {delivery_error}" @@ -4313,15 +4313,22 @@ impl CodexCommandExecutor { } fn settle_receipt_limit_interrupt_after_deadline(&mut self) -> Result<(), DurableRunnerError> { - let interrupt_accepted = self - .state - .as_ref() - .is_some_and(|state| state.receipt_limit_interrupt_accepted); let provider_shutdown_failed = self .provider .as_mut() .is_some_and(|provider| provider.shutdown().is_err()); + self.settle_receipt_limit_after_shutdown(provider_shutdown_failed) + } + + fn settle_receipt_limit_after_shutdown( + &mut self, + provider_shutdown_failed: bool, + ) -> Result<(), DurableRunnerError> { self.provider = None; + let interrupt_accepted = self + .state + .as_ref() + .is_some_and(|state| state.receipt_limit_interrupt_accepted); let identity = self.event_identity()?; let state = self .state @@ -4381,44 +4388,40 @@ impl CodexCommandExecutor { self.save_state() } - fn retry_receipt_limit_interrupt(&mut self) -> Result<(), DurableRunnerError> { - let (should_retry, attempts, persisted_deadline) = - self.state.as_ref().map_or((false, 0, None), |state| { - ( - state.receipt_limit_interrupt_pending - && state.active_provider_turn_id.is_some(), - state.receipt_limit_interrupt_attempts, - state.receipt_limit_interrupt_deadline_unix_ms, - ) - }); - if !should_retry { - return Ok(()); - } - let now_unix_ms = current_unix_ms()?; - let deadline_unix_ms = match persisted_deadline { - Some(deadline) => deadline, - None => { - // Older durable state did not record this additive field. Give - // an already-pending interruption one complete bounded window - // after recovery rather than falling back on poll count alone. - let timeout_ms = if self - .state - .as_ref() - .is_some_and(|state| state.receipt_limit_interrupt_accepted) - { - RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS - } else { - RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS - }; - let deadline = receipt_limit_deadline_after(timeout_ms)?; - self.state - .as_mut() - .expect("Codex state remains available while adding its receipt-limit deadline") - .receipt_limit_interrupt_deadline_unix_ms = Some(deadline); - self.save_state()?; - deadline - } + fn receipt_limit_interrupt_deadline(&mut self) -> Result, DurableRunnerError> { + let Some(state) = self.state.as_ref().filter(|state| { + state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some() + }) else { + return Ok(None); }; + if let Some(deadline) = state.receipt_limit_interrupt_deadline_unix_ms { + return Ok(Some(deadline)); + } + // Preserve the existing legacy-journal window, once, without retrying a provider RPC. + let timeout = if state.receipt_limit_interrupt_accepted { + RECEIPT_LIMIT_ACCEPTED_TERMINAL_DEADLINE_MS + } else { + RECEIPT_LIMIT_INTERRUPT_TERMINAL_DEADLINE_MS + }; + let deadline = receipt_limit_deadline_after(timeout)?; + self.state + .as_mut() + .expect("pending interruption has state") + .receipt_limit_interrupt_deadline_unix_ms = Some(deadline); + self.save_state()?; + Ok(Some(deadline)) + } + + fn retry_receipt_limit_interrupt(&mut self) -> Result<(), DurableRunnerError> { + let Some(deadline_unix_ms) = self.receipt_limit_interrupt_deadline()? else { + return Ok(()); + }; + let attempts = self + .state + .as_ref() + .expect("pending interruption has state") + .receipt_limit_interrupt_attempts; + let now_unix_ms = current_unix_ms()?; // Attempts bound network traffic, while the durable wall-clock deadline // gives an accepted asynchronous interruption time to deliver its // authoritative terminal. The provider is always polled once more below @@ -4465,6 +4468,24 @@ impl CodexCommandExecutor { return Ok(()); }; if current_unix_ms()? >= deadline { + if self + .state + .as_ref() + .is_some_and(|state| state.config.provider == "codex") + { + if let Some(provider) = self.provider.as_mut() { + // A bounded poll may leave a real terminal behind its raw + // prefix. Fence now, but drain it across poll/ACK batches + // before a missing-terminal fallback can replace evidence. + let drained = provider + .fence_local_receipt_limit() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if drained.output_empty { + self.settle_receipt_limit_after_shutdown(false)?; + } + return Ok(()); + } + } self.settle_receipt_limit_interrupt_after_deadline()?; } Ok(()) @@ -4785,6 +4806,25 @@ impl CodexCommandExecutor { let receipt_limit_terminal_poll = self.state.as_ref().is_some_and(|state| { state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some() }); + let local_deadline = self + .state + .as_ref() + .filter(|state| receipt_limit_terminal_poll && state.config.provider == "codex") + .and_then(|state| state.receipt_limit_interrupt_deadline_unix_ms); + if let Some(deadline) = local_deadline { + if current_unix_ms()? >= deadline { + if let Some(provider) = self.provider.as_mut() { + // Verify EOF into bounded raw storage before slow per-event + // persistence can exhaust the ordinary process-exit grace. + let fenced = provider + .fence_local_receipt_limit() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + if !fenced.stdout_drained { + return Ok(()); + } + } + } + } if !receipt_limit_terminal_poll && self .state @@ -5478,6 +5518,46 @@ impl CommandExecutor for CodexCommandExecutor { .collect()) } + fn maintain_disconnected_provider( + &mut self, + ) -> Result, DurableRunnerError> { + // Local Codex ownership cannot prove termination of a remote facade. + // Never restore a provider merely to service its cleanup deadline. + if self.provider.is_none() + || !self.state.as_ref().is_some_and(|state| { + state.config.provider == "codex" && state.startup_attempt.is_none() + }) + { + return Ok(None); + } + let Some(deadline) = self.receipt_limit_interrupt_deadline()? else { + return Ok(None); + }; + let remaining = deadline.saturating_sub(current_unix_ms()?); + if remaining > 0 { + return std::time::Instant::now() + .checked_add(std::time::Duration::from_millis(remaining)) + .map(Some) + .ok_or_else(|| { + DurableRunnerError::invalid( + "Codex receipt-limit deadline is outside the local clock", + ) + }); + } + let drained = self + .provider + .as_mut() + .expect("local provider remains owned") + .fence_local_receipt_limit() + .map_err(|error| DurableRunnerError::invalid(error.to_string()))?; + // FIFO evidence may contain a real terminal. Keep it for ordinary + // connected processing instead of manufacturing a conflicting fallback. + if drained.output_empty { + self.settle_receipt_limit_after_shutdown(false)?; + } + Ok(drained.next_drain_deadline) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { if self.state.as_ref().is_some_and(|state| { state.receipt_limit_interrupt_pending && state.active_provider_turn_id.is_some() @@ -7872,6 +7952,437 @@ mod tests { recovered.validate().unwrap(); } + #[cfg(unix)] + fn disconnected_receipt_limit_fixture( + mode: &str, + deadline: u64, + ) -> (CodexCommandExecutor, PathBuf, u64) { + let directory = std::env::temp_dir().join(format!( + "paperclip-offline-receipt-limit-{}", + uuid::Uuid::new_v4() + )); + fs::create_dir_all(&directory).unwrap(); + let ready = directory.join("ready"); + let script = r#" +trap '' TERM +while IFS= read -r line; do + case "$line" in + *'"method":"initialize"'*) printf '%s\n' '{"id":1,"result":{}}' ;; + *'"method":"thread/start"'*) printf '%s\n' '{"id":2,"result":{"thread":{"id":"thread-1"}}}' ;; + *'"method":"turn/start"'*) + printf '%s\n' '{"id":3,"result":{"turn":{"id":"turn-1"}}}' + case "$1" in prefix*) printf '%s\n' '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"predecessor output"}}' ;; esac + case "$1" in many*) + count=0 + while [ "$count" -lt "$4" ]; do + printf '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"prefix-%s"}}\n' "$count" + count=$((count + 1)) + done ;; + esac + case "$1" in requests_terminal) + printf '%s\n' \ + '{"method":"item/completed","params":{"threadId":"thread-1","turnId":"turn-1","item":{"type":"collabAgentToolCall","tool":"spawnAgent","status":"completed","senderThreadId":"thread-1","receiverThreadIds":["child-thread"]}}}' \ + '{"id":91,"method":"item/tool/call","params":{"threadId":"child-thread"}}' \ + '{"id":92,"method":"item/tool/requestUserInput","params":{"threadId":"child-thread","questions":"malformed"}}' \ + '{"id":93,"method":"mcpServer/elicitation/request","params":{"threadId":"child-thread","_meta":{"codex_approval_kind":"mcp_tool_call"}}}' \ + '{"id":94,"method":"unsupported/withoutParams"}' \ + '{"id":95,"method":"unsupported/malformedParams","params":"malformed"}' \ + '{"id":96}' \ + '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"after denied child requests"}}' + ;; + esac + case "$1" in *terminal) printf '%s\n' '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}' ;; esac + printf ready > "$2" + while :; do sleep 1; done ;; + *'"id":'*) printf '%s\n' "$line" >> "$3" ;; + esac +done +"#; + let mut state = opencode_result_state(); + state.config.provider = "codex".to_owned(); + state.config.driver = "codex_app_server".to_owned(); + state.config.command = PathBuf::from("/bin/sh"); + state.config.args = vec![ + "-c".to_owned(), + script.to_owned(), + "fixture".to_owned(), + mode.to_owned(), + ready.to_string_lossy().into_owned(), + directory + .join("unexpected-rpc") + .to_string_lossy() + .into_owned(), + (MAX_EVENTS_PER_POLL * 2 + 17).to_string(), + ]; + state.config.model = None; + state.goal_capability = Some(SessionGoalCapability::codex_available()); + let mut provider = CodexProvider::start(&state.config, None).unwrap(); + provider.start_turn("Test", &state.config.cwd).unwrap(); + let until = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !ready.exists() { + assert!(std::time::Instant::now() < until); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + state.provider_process_generation = provider.process_generation(); + state + .begin_receipt_limit_stop( + "limited-call".to_owned(), + "fixture-tool".to_owned(), + deadline, + ) + .unwrap(); + state.record_receipt_limit_interrupt_attempt().unwrap(); + let mut executor = CodexCommandExecutor::new(&directory); + executor.event_identity = Some(ProviderEventIdentity { + runner_instance_id: "runner".to_owned(), + run_id: "run".to_owned(), + normalized_session_id: "session".to_owned(), + turn_id: "turn".to_owned(), + item_id: "item".to_owned(), + }); + executor.state = Some(state); + executor.provider = Some(provider); + executor.save_state().unwrap(); + (executor, directory, deadline) + } + + #[cfg(unix)] + #[test] + fn disconnected_receipt_limit_preserves_a_future_deadline_without_rpc_or_facade_claims() { + let (mut executor, directory, deadline) = + disconnected_receipt_limit_fixture("silent", current_unix_ms().unwrap() + 60_000); + for _ in 0..2 { + assert!( + executor.maintain_disconnected_provider().unwrap().unwrap() + > std::time::Instant::now() + ); + assert_eq!( + executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_deadline_unix_ms, + Some(deadline) + ); + assert_eq!( + executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_attempts, + 1 + ); + assert!(!executor + .provider + .as_mut() + .unwrap() + .process_exited() + .unwrap()); + } + // A local proxy handle cannot attest that a remote provider was stopped. + executor.state.as_mut().unwrap().config.provider = "cursor_cloud".to_owned(); + executor + .state + .as_mut() + .unwrap() + .receipt_limit_interrupt_deadline_unix_ms = Some(1); + assert!(executor.maintain_disconnected_provider().unwrap().is_none()); + assert!( + executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_pending + ); + assert!(!executor + .provider + .as_mut() + .unwrap() + .process_exited() + .unwrap()); + assert!(!directory.join("unexpected-rpc").exists()); + assert!(!executor.restore_checked); + executor + .provider + .as_mut() + .unwrap() + .fence_local_receipt_limit() + .unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + + #[cfg(unix)] + #[test] + fn disconnected_receipt_limit_stops_a_hanging_provider_at_its_original_deadline() { + for accepted in [false, true] { + let (mut executor, directory, deadline) = + disconnected_receipt_limit_fixture("silent", 1); + executor + .state + .as_mut() + .unwrap() + .receipt_limit_interrupt_accepted = accepted; + assert_eq!( + executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_deadline_unix_ms, + Some(deadline) + ); + let started = std::time::Instant::now(); + executor.maintain_disconnected_provider().unwrap(); + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + if let Some(provider) = executor.provider.as_mut() { + // The owned process is already dead even if the reader has not yet observed EOF. + assert!(provider.process_exited().unwrap()); + } + while executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_pending + { + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + executor.maintain_disconnected_provider().unwrap(); + } + let state = executor.state.as_ref().unwrap(); + assert_eq!(state.lifecycle, "closed"); + assert!(state.pending_events.iter().any(|event| event.event_type + == if accepted { + "turn.interrupted" + } else { + "turn.failed" + })); + assert!(!directory.join("unexpected-rpc").exists()); + assert!(!executor.restore_checked); + fs::remove_dir_all(directory).unwrap(); + } + } + + #[cfg(unix)] + #[test] + fn disconnected_receipt_limit_preserves_terminal_and_its_fifo_prefix_for_reconnect() { + for mode in ["terminal", "prefix_terminal", "prefix_only"] { + let (mut executor, directory, deadline) = disconnected_receipt_limit_fixture(mode, 1); + let retained = executor.state.as_ref().unwrap().pending_events.clone(); + assert!(executor.maintain_disconnected_provider().unwrap().is_none()); + assert!(executor + .provider + .as_mut() + .unwrap() + .process_exited() + .unwrap()); + let state = executor.state.as_ref().unwrap(); + assert_eq!(state.pending_events, retained); + assert!(state.receipt_limit_interrupt_pending); + assert_eq!(state.receipt_limit_interrupt_attempts, 1); + assert_eq!( + state.receipt_limit_interrupt_deadline_unix_ms, + Some(deadline) + ); + assert!(!directory.join("unexpected-rpc").exists()); + // The ordinary connected path drains the exact prefix before it + // accepts a terminal or applies the honest missing-terminal fallback. + executor.restore_checked = true; + let events = executor.poll_events().unwrap(); + let terminal = events + .iter() + .position(|event| { + event.event_type + == if mode == "prefix_only" { + "turn.failed" + } else { + "turn.completed" + } + }) + .unwrap(); + if mode.starts_with("prefix") { + let prefix = events + .iter() + .position(|event| { + serde_json::to_string(&event.payload) + .unwrap() + .contains("predecessor output") + }) + .unwrap(); + assert!(prefix < terminal); + } + assert!( + !executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_pending + ); + assert!(!directory.join("unexpected-rpc").exists()); + fs::remove_dir_all(directory).unwrap(); + } + } + + #[cfg(unix)] + #[test] + fn fenced_receipt_limit_replays_child_and_unavailable_requests_before_root_terminal() { + let (mut executor, directory, _) = + disconnected_receipt_limit_fixture("requests_terminal", 1); + executor.restore_checked = true; + let mut observed = Vec::new(); + for _ in 0..10 { + let events = executor.poll_events().unwrap(); + let count = events.len(); + observed.extend(events); + executor.acknowledge_events(count).unwrap(); + if !executor + .state + .as_ref() + .unwrap() + .receipt_limit_interrupt_pending + { + break; + } + } + let terminal = observed + .iter() + .position(|event| event.event_type == "turn.completed") + .expect("the real root terminal survives every queued courtesy rejection"); + assert_eq!( + observed + .iter() + .filter(|event| event.event_type == "turn.completed") + .count(), + 1 + ); + assert!(!observed.iter().any(|event| matches!( + event.event_type.as_str(), + "turn.failed" | "turn.interrupted" | "semantic_tool.input" | "runtime_request.created" + ))); + let prefix = observed + .iter() + .position(|event| event.payload["text"] == "after denied child requests") + .unwrap(); + assert!(prefix < terminal); + assert!( + observed[..terminal] + .iter() + .filter(|event| { + serde_json::to_string(&event.payload) + .unwrap() + .contains("unsupported Codex request") + || serde_json::to_string(&event.payload) + .unwrap() + .contains("authorized only for the parent task") + }) + .count() + >= 5 + ); + assert!(!directory.join("unexpected-rpc").exists()); + fs::remove_dir_all(directory).unwrap(); + } + + #[cfg(unix)] + fn assert_disconnected_receipt_limit_drains_batches(mode: &str, has_terminal: bool) { + let (mut executor, directory, _) = disconnected_receipt_limit_fixture(mode, 1); + executor.maintain_disconnected_provider().unwrap(); + assert!(executor + .provider + .as_mut() + .unwrap() + .process_exited() + .unwrap()); + executor.restore_checked = true; + let mut observed = Vec::new(); + let mut batches = 0; + loop { + let events = executor.poll_events().unwrap(); + assert!(events.len() <= MAX_EVENTS_PER_POLL + 3); + let count = events.len(); + observed.extend(events); + executor.acknowledge_events(count).unwrap(); + batches += 1; + assert!(batches <= 10); + let state = executor.state.as_ref().unwrap(); + if !state.receipt_limit_interrupt_pending + && state.pending_events.is_empty() + && state.queued_events.is_empty() + { + break; + } + } + let texts = observed + .iter() + .filter(|event| event.event_type == "item.delta") + .map(|event| event.payload["text"].as_str().unwrap().to_owned()) + .collect::>(); + assert_eq!( + texts.len(), + MAX_EVENTS_PER_POLL * 2 + 17, + "{mode}: {batches} batches, last={:?}", + texts.last() + ); + assert!(batches >= 3); + assert_eq!( + texts, + (0..MAX_EVENTS_PER_POLL * 2 + 17) + .map(|index| format!("prefix-{index}")) + .collect::>() + ); + let ids = observed + .iter() + .map(|event| &event.executor_event_id) + .collect::>(); + assert_eq!(ids.len(), observed.len()); + assert_eq!( + observed + .iter() + .filter(|event| event.event_type + == if has_terminal { + "turn.completed" + } else { + "turn.failed" + }) + .count(), + 1 + ); + assert!(!observed.iter().any(|event| event.event_type + == if has_terminal { + "turn.failed" + } else { + "turn.completed" + })); + let terminal = observed + .iter() + .position(|event| { + event.event_type + == if has_terminal { + "turn.completed" + } else { + "turn.failed" + } + }) + .unwrap(); + assert!( + observed + .iter() + .rposition(|event| event.event_type == "item.delta") + .unwrap() + < terminal + ); + assert!(!directory.join("unexpected-rpc").exists()); + fs::remove_dir_all(directory).unwrap(); + } + + #[cfg(unix)] + #[test] + fn disconnected_receipt_limit_drains_large_prefix_before_real_terminal() { + assert_disconnected_receipt_limit_drains_batches("many_terminal", true); + } + + #[cfg(unix)] + #[test] + fn disconnected_receipt_limit_drains_large_prefix_before_missing_terminal_fallback() { + assert_disconnected_receipt_limit_drains_batches("many_only", false); + } + #[test] fn receipt_limit_deadline_settlement_preserves_unacknowledged_events() { let directory = std::env::temp_dir().join(format!( diff --git a/packages/paperclip-runner/src/control-plane/runner-recovery-fixture.test.ts b/packages/paperclip-runner/src/control-plane/runner-recovery-fixture.test.ts index e346eee823f..150ee1705e8 100644 --- a/packages/paperclip-runner/src/control-plane/runner-recovery-fixture.test.ts +++ b/packages/paperclip-runner/src/control-plane/runner-recovery-fixture.test.ts @@ -1,6 +1,7 @@ import { execFileSync, spawn, type ChildProcess } from "node:child_process"; import { once } from "node:events"; import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { createServer, type Socket } from "node:net"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; @@ -23,13 +24,13 @@ beforeAll(() => { "-p", "paperclip-runner-core", "--example", "recovery-fixture"], { timeout: 120_000, stdio: "pipe" }); }, 120_000); -async function fixture(externalPersistence: boolean, graceMs = 20_000) { +async function fixture(externalPersistence: boolean, graceMs = 20_000, providerStopMs?: number, reconnectDelayMs?: number) { const root = mkdtempSync(resolve(tmpdir(), "runner-in-process-recovery-")); const stateDirectory = resolve(root, "control-plane"); const receiptFile = resolve(root, "work.json"); const releaseFile = resolve(root, "release"); const scriptFile = resolve(root, "script.json"); - writeFileSync(scriptFile, JSON.stringify({ steps: 64, waitAfter: 32, retentionLimit: 8, receiptFile, releaseFile })); + writeFileSync(scriptFile, JSON.stringify({ steps: 64, waitAfter: 32, retentionLimit: 8, receiptFile, releaseFile, providerStopMs, reconnectDelayMs })); const delivered: Array<{ sourceEventId: string; text: string }> = []; let checkpoint: DurablePrpCheckpoint | undefined; const create = () => { @@ -66,7 +67,7 @@ async function fixture(externalPersistence: boolean, graceMs = 20_000) { await vi.waitFor(() => expect(core.getCommand(commandId)?.status, `${commandId}: exit=${child?.exitCode}; ${diagnostics}`).toBe("completed"), { timeout: 10_000 }); } return { - root, receipt, child, exit, delivered, command, + root, port, receipt, child, exit, delivered, command, checkpoint: () => checkpoint, release: () => writeFileSync(releaseFile, "compute"), stopController: async () => { @@ -127,12 +128,14 @@ it.each([false, true])("preserves useful in-process work through repeated replac }, 35_000); it("stops after the original reconnect deadline while retaining recovery state", async () => { - const run = await fixture(true, 1500); + const run = await fixture(true, 1500, undefined, 5000); try { await run.command("turn.start", "start-once"); + const disconnectedAt = performance.now(); await run.stopController(); run.release(); await run.exit; + expect(performance.now() - disconnectedAt).toBeLessThan(4000); const state = JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8")); expect(state.recoverableFailure).toBe("transport_reconnect_grace_exceeded"); expect(run.receipt()).toMatchObject({ phase: "stopped", pid: run.child.pid }); @@ -140,3 +143,22 @@ it("stops after the original reconnect deadline while retaining recovery state", expect(run.receipt().peakRetainedEvents).toBeLessThanOrEqual(8); } finally { await run.cleanup(); } }, 15_000); + + +it("enforces local provider cleanup while a replacement controller holds the handshake open", async () => { + const run = await fixture(false, 20_000, 1500); + const sockets = new Set(); + const silent = createServer(socket => { sockets.add(socket); socket.once("close", () => sockets.delete(socket)); }); + try { + await run.command("turn.start", "start-with-deadline"); + await run.stopController(); + await new Promise((ready, reject) => { silent.once("error", reject); silent.listen(run.port, "127.0.0.1", ready); }); + await vi.waitFor(() => expect(sockets.size).toBeGreaterThan(0), { timeout: 4_000 }); + await vi.waitFor(() => expect(run.receipt().phase).toBe("stopped"), { timeout: 4_000 }); + expect(run.child.exitCode).toBeNull(); + } finally { + for (const socket of sockets) socket.destroy(); + await new Promise(done => silent.close(() => done())); + await run.cleanup(); + } +}, 15_000);