diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 68b4d15db5b..a852fe0f07b 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -41,6 +41,12 @@ 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. +After disconnect, runnerd tries Core when reconnect is due. Between attempts, +live Codex processes can drain local output into the outbox. Work that +needs a provider request waits for Core, including managed-provider history +reads and unanswered tools or approvals. Runnerd suspends +intake during terminal delivery or an authority transition. + 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. @@ -76,6 +82,11 @@ the P0 reserve. When the soft limit is reached, the runner enters backpressure and rejects the new non-P0 event rather than silently losing it. Exhausting the P0 reserve is an explicit unrecoverable condition. +While disconnected, soft-limit backpressure retains the rejected event and its +suffix at the provider owner until replay ACKs free outbox space. Oversized +events or exhausted hard limits stop the owned provider and preserve the +accepted durable prefix. + ## Recovering Codex between turns Suppose run A finishes a turn, its Runner exits, and run B wants to continue 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 48805b685b0..7d32ed22622 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 @@ -187,6 +187,10 @@ impl CommandExecutor for ScriptedExecutor { Ok(self.stop_at) } + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + self.retained_events() + } + fn retained_events(&mut self) -> Result, DurableRunnerError> { Ok(self .shared diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/claude_managed_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/claude_managed_provider.rs index da44421ad4c..1871c885580 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/claude_managed_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/claude_managed_provider.rs @@ -2491,7 +2491,7 @@ fn extract_event_list(value: Value) -> Vec { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use std::io::{ErrorKind, Read, Write}; use std::net::{TcpListener, TcpStream}; @@ -3000,6 +3000,33 @@ mod tests { .map_err(|error| error.cause) } + pub(crate) fn disconnected_executor_fixture( + config: ClaudeManagedProviderConfig, + ) -> (Box, impl Fn() -> Vec) { + let service = FakeAnthropicService::start(Vec::new(), Vec::new()); + let origin = service.origin.clone(); + let provider = ClaudeManagedProvider::start_with_worker( + &config, + Vec::new(), + "test-scope", + None, + None, + 0, + Vec::new(), + SensitiveApiKey("anthropic-test-secret".to_owned()), + move |key, beta| NetworkWorker::start_at(key, beta, &origin, false), + ) + .map_err(|error| error.cause) + .unwrap(); + (Box::new(provider), move || { + service + .captured() + .iter() + .map(|request| format!("{} {}", request.method, request.path)) + .collect() + }) + } + #[test] fn tool_names_are_anthropic_safe_stable_and_collision_resistant() { let first = remote_tool_name("issues.comment:create"); 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 8cfffc68aa2..abfc3bdb241 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 @@ -2571,6 +2571,59 @@ impl CodexProvider { } pub fn poll(&mut self) -> Result, LocalRunnerError> { + self.poll_with_control(true) + } + + pub(crate) fn poll_disconnected( + &mut self, + ) -> Result, LocalRunnerError> { + self.poll_with_control(false) + } + + fn disconnected_message_is_passive(&self, message: &Value) -> bool { + let Some(method) = message.get("method").and_then(Value::as_str) else { + return false; + }; + let params = message.get("params").unwrap_or(&Value::Null); + // Preserve frames that need a request, authority reconciliation or a + // terminal tail drain for the ordinary connected path. + if method == "turn/started" + && !notification_turn_id(params) + .is_some_and(|id| Some(id) == self.active_provider_turn_id.as_deref()) + { + return false; + } + message.get("id").is_none() + && normalized_codex_terminal_event_type(method, params).is_none() + && params.pointer("/item/type").and_then(Value::as_str) != Some("collabAgentToolCall") + && matches!( + classify_notification_thread( + method, + &self.thread_id, + &self.descendant_thread_ids, + params + ), + Ok(NotificationThread::Root | NotificationThread::UnrelatedInformation) + ) + && validate_notification_binding( + &self.thread_id, + self.active_provider_turn_id.as_deref(), + params, + ) + .is_ok() + } + + fn poll_with_control( + &mut self, + allow_control: bool, + ) -> Result, LocalRunnerError> { + if !allow_control + && (self.quarantined + || self.ambiguous_turn_start_pending + || self.process.try_wait()?.is_some()) + { + return Ok(None); + } if self.process.stdout_failed() { return Err(LocalRunnerError::invalid( "Codex stdout closed without a valid reader EOF", @@ -2676,6 +2729,22 @@ impl CodexProvider { })?; (message, trace_frame_id) }; + if !allow_control && !self.disconnected_message_is_passive(&message) { + 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_front(BufferedProviderMessage { + value: message, + trace_frame_id, + }); + return Ok(None); + } self.last_trace_frame_id = trace_frame_id; if let (Some(trace), Some(frame_id)) = (self.trace.as_mut(), trace_frame_id) { let method = message.get("method").and_then(Value::as_str); @@ -4960,6 +5029,86 @@ done assert!(provider.completed_tool_call_ids.contains("denied-call")); } + #[cfg(unix)] + #[test] + fn disconnected_poll_retains_terminal_frame_and_exact_connected_replay() { + let mut provider = completion_tail_provider(); + provider.active_provider_turn_id = Some("offline-turn".to_owned()); + provider.completed_turn_authority = None; + let accepted_start = json!({"method":"turn/started","params":{ + "threadId":"reader-tail-thread","turn":{"id":"offline-turn"}}}); + let progress = json!({"method":"item/agentMessage/delta", "params":{ + "threadId":"reader-tail-thread", "turnId":"offline-turn", "delta":"offline output"}}); + let terminal = json!({"method":"turn/completed", "params":{ + "threadId":"reader-tail-thread", "turn":{"id":"offline-turn","status":"completed"}}}); + for (value, trace_frame_id) in [ + (accepted_start, Some(9)), + (progress, Some(10)), + (terminal.clone(), Some(11)), + ] { + provider.pending_message_bytes += json_size(&value, "fixture").unwrap(); + provider + .pending_messages + .push_back(BufferedProviderMessage { + value, + trace_frame_id, + }); + } + assert!(matches!(provider.poll_disconnected().unwrap(), + Some(CodexProviderEvent::Notification { method, .. }) if method == "turn/started")); + assert_eq!(provider.take_provider_trace_frame_id(), Some(9)); + assert!( + matches!(provider.poll_disconnected().unwrap(), Some(CodexProviderEvent::Notification { method, .. }) if method == "item/agentMessage/delta") + ); + assert_eq!(provider.take_provider_trace_frame_id(), Some(10)); + let bytes = provider.pending_message_bytes; + let started = std::time::Instant::now(); + for _ in 0..5 { + assert!(provider.poll_disconnected().unwrap().is_none()); + assert_eq!(provider.pending_messages.front().unwrap().value, terminal); + assert_eq!(provider.pending_message_bytes, bytes); + assert_eq!(provider.active_provider_turn_id(), Some("offline-turn")); + } + assert!(started.elapsed() < Duration::from_secs(1)); + assert!( + matches!(provider.poll().unwrap(), Some(CodexProviderEvent::Notification { method, params }) if method == "turn/completed" && params == terminal["params"]) + ); + assert_eq!(provider.take_provider_trace_frame_id(), Some(11)); + assert!(provider.pending_messages.is_empty()); + assert_eq!(provider.pending_message_bytes, 0); + assert_eq!(provider.active_provider_turn_id(), None); + provider.shutdown().unwrap(); + } + + #[cfg(unix)] + #[test] + fn disconnected_poll_defers_requests_lineage_and_turn_reconciliation() { + let mut provider = completion_tail_provider(); + provider.active_provider_turn_id = Some("offline-turn".to_owned()); + for message in [ + json!({"id":91,"method":"item/tool/call","params":{}}), + json!({"method":"item/agentMessage/delta","params":{"threadId":"unverified-child","delta":"child"}}), + json!({"method":"turn/started","params":{"threadId":"reader-tail-thread"}}), + json!({"method":"turn/started","params":{"threadId":"reader-tail-thread","turn":{"id":"next-turn"}}}), + ] { + provider.pending_message_bytes = json_size(&message, "fixture").unwrap(); + provider + .pending_messages + .push_back(BufferedProviderMessage { + value: message.clone(), + trace_frame_id: Some(12), + }); + let started = std::time::Instant::now(); + assert!(provider.poll_disconnected().unwrap().is_none()); + assert!(started.elapsed() < Duration::from_secs(1)); + let retained = provider.pending_messages.pop_front().unwrap(); + assert_eq!(retained.value, message); + assert_eq!(retained.trace_frame_id, Some(12)); + assert_eq!(provider.active_provider_turn_id(), Some("offline-turn")); + } + provider.shutdown().unwrap(); + } + #[cfg(unix)] #[test] fn fenced_receipt_output_keeps_prefix_when_stdout_never_closes() { 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 e25a3e209c6..b1e7bf391a5 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 @@ -329,6 +329,12 @@ pub trait CommandExecutor { Ok(Vec::new()) } + /// Drain live local output between controller reconnect attempts. Providers + /// whose ordinary poll can start recovery or wait for RPCs must defer that work. + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + 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> { @@ -559,6 +565,9 @@ pub fn run_durable_runner( None }; if Instant::now() < next_connection_attempt { + if authenticated_once { + poll_disconnected_executor_events(&mut state, &store, &config, &mut executor)?; + } let wake_at = next_connection_attempt.min(connection_attempt_deadline( &config, started, @@ -570,7 +579,7 @@ pub fn run_durable_runner( 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); + thread::sleep(remaining.min(Duration::from_millis(25))); } continue; } @@ -1778,6 +1787,40 @@ fn stop_after_terminal_result_delivery_failure( Err(error) } +fn poll_disconnected_executor_events( + state: &mut DurableState, + store: &DurableStateStore, + config: &DurableRunnerConfig, + executor: &mut E, +) -> Result<(), DurableRunnerError> { + // Maintenance has already run. Leave the rejected suffix at the provider until + // a reconnect ACK frees space. + if state.warm_transition.is_some() + || state.pending_terminal_delivery.is_some() + || state.pending_provider_cleanup.is_some() + || state.backpressure + { + return Ok(()); + } + let events = executor.poll_disconnected_events(); + let polled = persist_polled_events(state, store, config, executor, events); + match polled { + Err(error) + if error.classification() == Some("outbox_backpressure") + && !state.outbox.is_empty() => + { + store.save(state) + } + Err(error) => { + let _ = shutdown_preserving_cleanup(state, executor); + state.record_diagnostic(format!("disconnected provider ingestion failed: {error}")); + store.save(state)?; + Err(error) + } + Ok(()) => Ok(()), + } +} + fn poll_executor_events_after_controller_ack( state: &mut DurableState, store: &DurableStateStore, @@ -1838,7 +1881,18 @@ fn poll_executor_events( if state.pending_provider_cleanup.is_some() { return Ok(()); } - let events = match executor.poll_events() { + let events = executor.poll_events(); + persist_polled_events(state, store, config, executor, events) +} + +fn persist_polled_events( + state: &mut DurableState, + store: &DurableStateStore, + config: &DurableRunnerConfig, + executor: &mut E, + events: Result, DurableRunnerError>, +) -> Result<(), DurableRunnerError> { + let events = match events { Ok(events) => events, Err(error) => { drain_retained_events(state, store, config, executor).map_err(|secondary| { @@ -2608,6 +2662,10 @@ mod tests { Ok(self.events.iter().cloned().collect()) } + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + self.poll_events() + } + fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> { self.acknowledgements.push(count); if self.fail_acknowledgement { @@ -4139,6 +4197,91 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn disconnected_ingestion_keeps_durable_prefix_and_resumes_after_backpressure_ack() { + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-offline-backpressure-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&directory); + let mut config = config(directory.clone()); + config.max_outbox_bytes = 16 * 1024; + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = RetainingEventExecutor { + events: (1..=128) + .map(|index| PolledEvent { + executor_event_id: format!("offline-{index}"), + event_type: "item.delta".to_owned(), + priority: EventPriority::P1, + payload: json!({"index":index,"text":"x".repeat(512)}), + }) + .collect(), + fail_acknowledgement: false, + acknowledgements: Vec::new(), + }; + poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap(); + assert!(state.backpressure); + assert!(!state.outbox.is_empty()); + assert!(!executor.events.is_empty()); + assert_eq!(state.outbox.len() + executor.events.len(), 128); + assert_eq!(state.last_controller_command_seq, 0); + let (reloaded, _) = store.load_or_create(&config).unwrap(); + assert_eq!(reloaded.outbox, state.outbox); + assert!(reloaded.backpressure); + let prefix = state.highest_source_seq(); + poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap(); + assert_eq!(state.highest_source_seq(), prefix); + while !executor.events.is_empty() { + state.apply_ack(state.highest_source_seq(), 2).unwrap(); + poll_disconnected_executor_events(&mut state, &store, &config, &mut executor).unwrap(); + } + assert_eq!(state.highest_source_seq(), 128); + assert_eq!(executor.acknowledgements.iter().sum::(), 128); + assert!(state.peak_outbox_bytes <= config.max_outbox_bytes); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn disconnected_ingestion_retires_provider_on_unrecoverable_output() { + struct OversizedExecutor { + shutdowns: usize, + } + impl CommandExecutor for OversizedExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("offline ingestion cannot execute commands"); + } + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + Ok(vec![PolledEvent { + executor_event_id: "oversized".to_owned(), + event_type: "item.delta".to_owned(), + priority: EventPriority::P1, + payload: json!({"text":"x".repeat(2048)}), + }]) + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.shutdowns += 1; + Ok(()) + } + } + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-offline-output-failure-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&directory); + let mut config = config(directory.clone()); + config.max_frame_bytes = 1024; + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = OversizedExecutor { shutdowns: 0 }; + let error = poll_disconnected_executor_events(&mut state, &store, &config, &mut executor) + .unwrap_err(); + assert!(error.to_string().contains("transport frame limit")); + assert_eq!(executor.shutdowns, 1); + assert_eq!(state.highest_source_seq(), 0); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn provider_poll_yields_until_the_sent_controller_ack_prefix_is_consumed() { let directory = std::env::temp_dir().join(format!( 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 d7729216c32..832383bb752 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 @@ -662,7 +662,8 @@ impl DurableState { self.record_diagnostic("outbox soft limit reached; non-P0 event rejected"); return Err(DurableRunnerError::invalid( "outbox soft limit reached; reserved storage is available only to P0 events", - )); + ) + .with_classification("outbox_backpressure")); } if projected > config.max_outbox_bytes { self.lifecycle = "unrecoverable".to_owned(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs index a8afb907fb4..8614ef99ad1 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/managed_provider_backend.rs @@ -2698,6 +2698,70 @@ mod tests { }) } + #[test] + fn disconnected_managed_executor_does_not_restore_or_reconcile_remote_streams() { + let directory = + std::env::temp_dir().join(format!("paperclip-managed-disconnected-{}", Uuid::new_v4())); + let config = test_config(&directory); + let mut executor = ManagedProviderCommandExecutor::with_runner_config(&directory, &config); + let mut payload = claude_prepare_payload(); + payload["provider"]["runtimeContext"] = Value::Null; + executor + .execute(&test_command(1, "run.prepare", payload)) + .unwrap(); + let ManagedProviderDescriptor::ClaudeManaged(descriptor) = + executor.state.as_ref().unwrap().descriptor.clone() + else { + panic!("expected Claude managed descriptor"); + }; + let (provider, requests) = + crate::claude_managed_provider::tests::disconnected_executor_fixture(descriptor); + executor.provider = Some(provider); + executor.state.as_mut().unwrap().lifecycle = "session_open".to_owned(); + executor.refresh_provider_checkpoint(); + executor.save_state().unwrap(); + let retained = fs::read(executor.state_path()).unwrap(); + let before = requests(); + assert_eq!( + before + .iter() + .filter(|path| path.contains("/events/stream")) + .count(), + 1 + ); + + for _ in 0..64 { + assert!(executor.poll_disconnected_events().unwrap().is_empty()); + } + assert_eq!(requests(), before); + assert_eq!(fs::read(executor.state_path()).unwrap(), retained); + + let mut replacement = + ManagedProviderCommandExecutor::with_runner_config(&directory, &config); + assert!(replacement.poll_disconnected_events().unwrap().is_empty()); + assert!(!replacement.restore_checked); + assert_eq!(requests(), before); + + // The real worker's closed stream needs history and a reopen on ordinary + // polling. This control proves disconnected polling left it untouched. + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while requests().len() == before.len() && std::time::Instant::now() < deadline { + let _ = executor.poll_events(); + std::thread::sleep(Duration::from_millis(1)); + } + let after = requests(); + assert!(after.iter().any(|path| path.contains("/events?"))); + assert_eq!( + after + .iter() + .filter(|path| path.contains("/events/stream")) + .count(), + 2 + ); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn attach_refreshes_tools_in_the_same_session_and_preserves_profile_fences() { let directory = 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 0de760471cb..86a0ff184c0 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 @@ -61,6 +61,15 @@ impl CommandExecutor for SelectedExecutor { } } + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + match self { + Self::LocalFacade(executor) => executor.poll_disconnected_events(), + Self::Acpx(executor) => executor.poll_disconnected_events(), + Self::Managed(executor) => executor.poll_disconnected_events(), + Self::Dot(executor) => executor.poll_disconnected_events(), + } + } + fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { match self { Self::LocalFacade(executor) => executor.maintain_disconnected_provider(), @@ -215,6 +224,35 @@ impl NativeProviderCommandExecutor { Ok(()) } + fn poll_with_credentials( + &mut self, + poll: fn(&mut SelectedExecutor) -> Result, DurableRunnerError>, + ) -> Result, DurableRunnerError> { + self.credentials.poll(false)?; + if let Some(event) = self.credentials.pending() { + return Ok(vec![event]); + } + let events = self + .selected + .as_mut() + .map_or_else(|| Ok(Vec::new()), poll)?; + let terminal = events.iter().any(|event| { + matches!( + event.event_type.as_str(), + "turn.completed" + | "turn.failed" + | "turn.interrupted" + | "turn.cancelled" + | "harness.exited" + ) + }); + self.credentials.poll(terminal)?; + Ok(self + .credentials + .pending() + .map_or(events, |event| vec![event])) + } + fn select_from_payload(&mut self, payload: &Value) -> Result<(), DurableRunnerError> { let kind = payload .pointer("/provider/kind") @@ -375,29 +413,11 @@ impl CommandExecutor for NativeProviderCommandExecutor { fn poll_events(&mut self) -> Result, DurableRunnerError> { self.select_recovery()?; - self.credentials.poll(false)?; - if let Some(event) = self.credentials.pending() { - return Ok(vec![event]); - } - let events = self - .selected - .as_mut() - .map_or_else(|| Ok(Vec::new()), CommandExecutor::poll_events)?; - let terminal = events.iter().any(|event| { - matches!( - event.event_type.as_str(), - "turn.completed" - | "turn.failed" - | "turn.interrupted" - | "turn.cancelled" - | "harness.exited" - ) - }); - self.credentials.poll(terminal)?; - Ok(self - .credentials - .pending() - .map_or(events, |event| vec![event])) + self.poll_with_credentials(CommandExecutor::poll_events) + } + + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + self.poll_with_credentials(CommandExecutor::poll_disconnected_events) } fn maintain_disconnected_provider(&mut self) -> Result, DurableRunnerError> { 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 cd74a65a470..aa0101022f8 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 @@ -4806,6 +4806,13 @@ impl CodexCommandExecutor { } fn poll_current_provider(&mut self) -> Result<(), DurableRunnerError> { + self.poll_current_provider_with_control(true) + } + + fn poll_current_provider_with_control( + &mut self, + allow_control: bool, + ) -> Result<(), DurableRunnerError> { // Receipt-limit interruption is autonomous recovery. It must advance // even while older durable events await acknowledgement, otherwise a // slow or disconnected controller can keep an exhausted provider turn @@ -4845,14 +4852,18 @@ impl CodexCommandExecutor { return self.settle_receipt_limit_interrupt_if_deadline_elapsed(); } for _ in 0..MAX_EVENTS_PER_POLL { - let event = self + let provider = self .provider .as_mut() - .expect("provider remains present while polling") - .poll() - .map_err(|error| { - DurableRunnerError::invalid(format!("Codex provider failed: {error}")) - })?; + .expect("provider remains present while polling"); + let event = (if allow_control { + provider.poll() + } else { + provider.poll_disconnected() + }) + .map_err(|error| { + DurableRunnerError::invalid(format!("Codex provider failed: {error}")) + })?; let Some(event) = event else { break }; let trace_frame_id = self .provider @@ -5513,6 +5524,21 @@ impl CommandExecutor for CodexCommandExecutor { self.retained_events() } + fn poll_disconnected_events(&mut self) -> Result, DurableRunnerError> { + // Only an already-admitted local Codex process can supply passive + // output. Recovery, facades and cleanup retain their connected path. + if self.provider.is_some() + && self.state.as_ref().is_some_and(|state| { + state.config.provider == "codex" + && state.startup_attempt.is_none() + && !state.receipt_limit_interrupt_pending + }) + { + self.poll_current_provider_with_control(false)?; + } + self.retained_events() + } + fn retained_events(&mut self) -> Result, DurableRunnerError> { if let Some(error) = &self.startup_evidence_error { return Err(error.clone()); @@ -6008,6 +6034,123 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn disconnected_poll_never_restores_an_unopened_facade() { + let directory = std::env::temp_dir().join(format!( + "paperclip-offline-no-restore-{}", + uuid::Uuid::new_v4() + )); + fs::create_dir_all(&directory).unwrap(); + fs::write( + directory.join(CODEX_PROVIDER_STATE_FILE), + b"not a provider state", + ) + .unwrap(); + let mut executor = CodexCommandExecutor::new(&directory); + assert!(executor.poll_disconnected_events().unwrap().is_empty()); + assert!(executor.maintain_disconnected_provider().unwrap().is_none()); + assert!(!executor.restore_checked); + executor.state = Some(opencode_result_state()); + assert!(executor.poll_disconnected_events().unwrap().is_empty()); + assert!(executor.maintain_disconnected_provider().unwrap().is_none()); + assert!(!executor.restore_checked); + fs::remove_dir_all(directory).unwrap(); + } + + #[cfg(unix)] + #[test] + fn disconnected_codex_output_is_durable_and_terminal_control_waits_for_reconnect() { + let directory = + std::env::temp_dir().join(format!("paperclip-offline-codex-{}", uuid::Uuid::new_v4())); + fs::create_dir_all(&directory).unwrap(); + let goal_called = directory.join("goal-called"); + let script = format!( + r#" +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"}}}}}}' + printf '%s\n' '{{"method":"turn/started","params":{{"threadId":"thread-1","turn":{{"id":"turn-1","status":"inProgress"}}}}}}' + printf '%s\n' '{{"method":"item/agentMessage/delta","params":{{"threadId":"thread-1","turnId":"turn-1","delta":"retained offline output"}}}}' + printf '%s\n' '{{"method":"turn/completed","params":{{"threadId":"thread-1","turn":{{"id":"turn-1","status":"completed"}}}}}}' ;; + *'"method":"thread/goal/get"'*) + touch '{}' + printf '%s\n' '{{"id":4,"result":{{"goal":null}}}}' ;; + esac +done +"#, + goal_called.display() + ); + 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]; + 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(); + state.provider_process_generation = provider.process_generation(); + let mut executor = CodexCommandExecutor::new(&directory); + executor.state = Some(state); + executor.provider = Some(provider); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let mut observed = Vec::new(); + loop { + assert!(std::time::Instant::now() < deadline); + let events = executor.poll_disconnected_events().unwrap(); + if events.is_empty() { + continue; + } + let saved: CodexProviderState = + serde_json::from_slice(&fs::read(executor.state_path()).unwrap()).unwrap(); + assert_eq!(saved.pending_events, events); + let count = events.len(); + observed.extend(events); + executor.acknowledge_events(count).unwrap(); + if observed + .iter() + .any(|event| event.payload["text"] == "retained offline output") + { + break; + } + } + let start = observed + .iter() + .position(|event| event.event_type == "turn.started") + .unwrap(); + let text = observed + .iter() + .position(|event| event.payload["text"] == "retained offline output") + .unwrap(); + assert!(start < text); + let started = std::time::Instant::now(); + for _ in 0..5 { + assert!(executor.poll_disconnected_events().unwrap().is_empty()); + } + assert!(started.elapsed() < std::time::Duration::from_secs(1)); + assert!(!goal_called.exists()); + assert_eq!( + executor + .state + .as_ref() + .unwrap() + .active_provider_turn_id + .as_deref(), + Some("turn-1") + ); + let events = executor.poll_events().unwrap(); + assert!(goal_called.exists()); + assert!(events + .iter() + .any(|event| event.event_type == "turn.completed")); + executor.provider.as_mut().unwrap().shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + fn opencode_result_state() -> CodexProviderState { let mut state = CodexProviderState::new( CodexProviderConfig { 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 150ee1705e8..8f91b6d5aa0 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 @@ -99,12 +99,17 @@ it.each([false, true])("preserves useful in-process work through repeated replac expect(run.receipt().completedSteps).toBe(0); await run.stopController(); run.release(); - await vi.waitFor(() => expect(run.receipt()).toMatchObject({ phase: "backpressured", completedSteps: 8, retainedEvents: 8 }), { timeout: 5000 }); + await vi.waitFor(() => expect(run.receipt()).toMatchObject({ phase: "awaiting_controller", completedSteps: 32, retainedEvents: 0, acknowledgedEvents: 32 }), { timeout: 10_000 }); expect(run.receipt().pid).toBe(run.child.pid); expect(run.child.exitCode).toBeNull(); expect(run.delivered).toHaveLength(0); + const offline = JSON.parse(readFileSync(resolve(run.root, "runner/runner-state.json"), "utf8")); + expect(offline.ackedSourceSeq).toBe(0); + expect(offline.outbox).toHaveLength(32); + expect(offline.peakOutboxBytes).toBeLessThanOrEqual(offline.maxOutboxBytes); await run.replaceController(); await vi.waitFor(() => expect(run.delivered).toHaveLength(32), { timeout: 10_000 }); + expect(run.delivered.map(event => event.sourceEventId)).toEqual(offline.outbox.map((event: any) => event.envelope.payload.sourceEventId)); expect(run.receipt()).toMatchObject({ phase: "awaiting_controller", completedSteps: 32 }); await run.stopController(); expect(run.receipt().completedSteps).toBe(32); @@ -125,7 +130,7 @@ it.each([false, true])("preserves useful in-process work through repeated replac await run.command("runner.shutdown", "shutdown"); expect((await run.exit)[0]).toBe(0); } finally { await run.cleanup(); } -}, 35_000); +}, 60_000); it("stops after the original reconnect deadline while retaining recovery state", async () => { const run = await fixture(true, 1500, undefined, 5000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 650e4f3475a..fd6c8da5ac2 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -3900,7 +3900,7 @@ it.each(["admission-only", "admission-only-mismatched-model", "complete", "api-k } else { await expect(probe).rejects.toThrow(mode === "missing-runtime" ? /ENOENT|unavailable|cleanup/i : mode === "admission-only-mismatched-model" || mode === "mismatched-model" || mode === "remote-mismatched-model" || mode === "model-refresh-error" || mode === "model-cleanup-error" || mode === "missing-model" ? /observed model|mismatched provider model/i : /timed out|cleanup|suspension/i); if (mode === "late-remote-release-error") { - expect(registrationReleased).toBe(true); expect(cleanupConfirmed).toBe(false); + await vi.waitFor(() => expect(registrationReleased).toBe(true)); expect(cleanupConfirmed).toBe(false); expect((await stat(runtimeDirectory)).isDirectory()).toBe(true); } if (mode === "model-refresh-error") {