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 2a92b5880ec..2fe4ddc0c13 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 @@ -1388,6 +1388,31 @@ fn run() -> Result<(), Box> { if let Some(delay) = argument(&args, "--thread-resume-delay-ms") { thread::sleep(Duration::from_millis(delay.parse()?)); } + // A provider that closes its stdout without replying once and + // stays alive: a provider failure, not a silent stall. + #[cfg(unix)] + if args + .iter() + .any(|arg| arg == "--close-stdout-on-first-thread-resume") + { + let marker = state_path.with_file_name("fake-thread-resume-closed"); + if !marker.exists() { + fs::write(&marker, b"closed")?; + use std::os::fd::FromRawFd; + // SAFETY: fd 1 is this process's stdout; nothing writes to it after this. + drop(unsafe { std::os::fd::OwnedFd::from_raw_fd(1) }); + thread::sleep(Duration::from_secs(120)); + } + } + // A provider that stalls without replying once (host stall or + // freeze), then answers normally after a relaunch. + if let Some(delay) = argument(&args, "--stall-first-thread-resume-ms") { + let marker = state_path.with_file_name("fake-thread-resume-stalled"); + if !marker.exists() { + fs::write(&marker, b"stalled")?; + thread::sleep(Duration::from_millis(delay.parse()?)); + } + } if reject_missing_resume_state { send(json!({"id": id, "error": { "code": -32600, 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 9c2aa55af90..a648ae0e21e 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 @@ -202,6 +202,9 @@ pub(crate) enum ProviderStartupObservation { Failed { stage: ProviderStartupStage, child_exit: Option, + /// The provider gave no reply within the request deadline: a stalled + /// or frozen host, not a provider that rejected the request or exited. + no_reply: bool, }, } @@ -880,6 +883,12 @@ pub struct CodexProvider { completion_contract: Option, permission_profile: &'static str, exit_drain: Option, + /// The latest synchronous request ended because its deadline passed with + /// the provider still running and silent (no reply, no exit). + last_request_got_no_reply: bool, + /// The latest stdout receive returned nothing because its own timeout + /// passed, not because stdout closed or the reader disconnected. + last_receive_timed_out: bool, } #[derive(Debug)] @@ -1249,6 +1258,7 @@ impl CodexProvider { let _ = observe(ProviderStartupObservation::Failed { stage: ProviderStartupStage::Spawn, child_exit: None, + no_reply: false, }); return Err(error); } @@ -1261,6 +1271,7 @@ impl CodexProvider { let _ = observe(ProviderStartupObservation::Failed { stage: ProviderStartupStage::SpawnReceipt, child_exit, + no_reply: false, }); return Err(error); } @@ -1321,6 +1332,8 @@ impl CodexProvider { }), permission_profile, exit_drain: None, + last_request_got_no_reply: false, + last_receive_timed_out: false, }; let mut stage = ProviderStartupStage::Initialize; let initialized_result = (|| -> Result<(), LocalRunnerError> { @@ -1438,8 +1451,13 @@ impl CodexProvider { Ok(()) })(); if let Err(error) = initialized_result { + let no_reply = provider.last_request_got_no_reply; let child_exit = provider.retire_failed_startup(); - let _ = observe(ProviderStartupObservation::Failed { stage, child_exit }); + let _ = observe(ProviderStartupObservation::Failed { + stage, + child_exit, + no_reply, + }); return Err(error); } Ok(provider) @@ -1912,6 +1930,7 @@ impl CodexProvider { let _ = observe(ProviderStartupObservation::Failed { stage: ProviderStartupStage::Admission, child_exit, + no_reply: false, }); replacement.expected_shutdown = false; *self = replacement; @@ -1929,6 +1948,7 @@ impl CodexProvider { let _ = observe(ProviderStartupObservation::Failed { stage: ProviderStartupStage::Admission, child_exit, + no_reply: false, }); return Err(error); } @@ -3960,6 +3980,7 @@ impl CodexProvider { self.next_request_id = self.next_request_id.checked_add(1).ok_or_else(|| { ProviderRequestError::Rejected(LocalRunnerError::invalid("Codex request id exhausted")) })?; + self.last_request_got_no_reply = false; self.send_frame(&json!({"id": request_id, "method": method, "params": params})) .map_err(ProviderRequestError::Ambiguous)?; loop { @@ -3967,6 +3988,7 @@ impl CodexProvider { deadline.saturating_duration_since(std::time::Instant::now()) }); if remaining.is_zero() { + self.last_request_got_no_reply = true; return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid( format!("Provider {method} exceeded its bounded response deadline"), ))); @@ -3989,6 +4011,8 @@ impl CodexProvider { exit.exit_code, exit.signal ) } else { + // Only the receive's own timeout, never a closed stdout. + self.last_request_got_no_reply = self.last_receive_timed_out; format!("Codex {method} response timed out{diagnostic_suffix}") }; return Err(ProviderRequestError::Ambiguous(LocalRunnerError::invalid( @@ -4079,10 +4103,12 @@ impl CodexProvider { &mut self, timeout: Duration, ) -> Result, LocalRunnerError> { + self.last_receive_timed_out = false; let deadline = std::time::Instant::now() + timeout; loop { let remaining = deadline.saturating_duration_since(std::time::Instant::now()); if remaining.is_zero() { + self.last_receive_timed_out = true; return Ok(None); } match self.process.recv_timeout(remaining) { @@ -4095,7 +4121,10 @@ impl CodexProvider { } Ok(ProcessOutput::StdoutClosed) => return Ok(None), Ok(ProcessOutput::StderrClosed) => {} - Err(mpsc::RecvTimeoutError::Timeout) => return Ok(None), + Err(mpsc::RecvTimeoutError::Timeout) => { + self.last_receive_timed_out = true; + return Ok(None); + } Err(mpsc::RecvTimeoutError::Disconnected) => return Ok(None), } } @@ -5977,7 +6006,9 @@ done "durable spawned receipt write failed", )) } - ProviderStartupObservation::Failed { stage, child_exit } => { + ProviderStartupObservation::Failed { + stage, child_exit, .. + } => { failure = Some((stage, child_exit)); Ok(()) } 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 d3cc6e05690..8b41026b732 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 @@ -163,6 +163,13 @@ struct ProviderStartupAttempt { /// that predates it rejects it as unknown, which also fails closed. #[serde(default, skip_serializing_if = "std::ops::Not::not")] before_spawn: bool, + /// Initialization failed because the provider never replied within the + /// request deadline (a stalled or frozen host), not because it rejected + /// the request or exited. Omitted when false, so state saved without a + /// timed-out fence stays readable by runners that predate the field; a + /// runner that predates it rejects a fact that carries it (fails closed). + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + no_reply: bool, } impl ProviderStartupAttempt { @@ -213,6 +220,18 @@ impl ProviderStartupAttempt { None => false, }, }; + // Only a provider that was running and silent through its request + // deadline, during initialization, can have given no reply. + let no_reply_valid = !self.no_reply + || (self.phase == ProviderStartupPhase::InitializationFailed + && matches!( + self.failed_stage, + Some( + ProviderStartupStage::Initialize + | ProviderStartupStage::ThreadOpen + | ProviderStartupStage::ThreadRead + ) + )); // Only an intent, or a launch whose checks failed before the spawn, // can still be before the spawn. let before_spawn_valid = !self.before_spawn @@ -221,6 +240,7 @@ impl ProviderStartupAttempt { && self.failed_stage == Some(ProviderStartupStage::Spawn)); if self.schema != "paperclip.provider_startup.v1" || !before_spawn_valid + || !no_reply_valid || uuid::Uuid::parse_str(&self.launch_id).is_err() || self.attempted_process_generation == 0 || self.authenticated_thread_id.is_some() @@ -2233,6 +2253,7 @@ impl CodexCommandExecutor { signal: None, process_tree_retired: false, before_spawn: true, + no_reply: false, }); self.save_startup_fact() } @@ -2265,9 +2286,14 @@ impl CodexCommandExecutor { attempt.process_id = Some(process_id); attempt.process_group_id = Some(process_group_id); } - ProviderStartupObservation::Failed { stage, child_exit } => { + ProviderStartupObservation::Failed { + stage, + child_exit, + no_reply, + } => { attempt.phase = ProviderStartupPhase::InitializationFailed; attempt.failed_stage = Some(stage); + attempt.no_reply = no_reply; attempt.direct_child_exit_observed = child_exit.is_some(); attempt.exit_code = child_exit.as_ref().and_then(|fact| fact.exit_code); attempt.signal = child_exit.as_ref().and_then(|fact| fact.signal); @@ -2355,6 +2381,7 @@ impl CodexCommandExecutor { let _ = self.observe_startup(ProviderStartupObservation::Failed { stage: ProviderStartupStage::Admission, child_exit, + no_reply: false, }); } @@ -2396,13 +2423,16 @@ impl CodexCommandExecutor { /// - an intent still marked before the spawn (the runner died while /// checking the launch, so no process was ever started); or /// - a `spawned` launch whose recorded process and process group have - /// both exited. + /// both exited; or + /// - an initialization that failed only because the provider gave no + /// reply within the request deadline (`noReply`), whose recorded process + /// and process group have both exited. /// /// Restore then relaunches the provider through the normal resume path. /// Every other fence stays closed: an intent that reached the spawn or /// predates the before-spawn mark, a provider whose process or group may - /// still run, a startup that failed initialization or spawn (a real - /// provider failure that needs reconciliation), and a session that + /// still run, a startup that failed spawn or was rejected by the provider + /// (a real provider failure that needs reconciliation), and a session that /// recovery deliberately closed. fn retire_orphaned_startup_attempt( &mut self, @@ -2434,6 +2464,17 @@ impl CodexCommandExecutor { (Some(pid), Some(group)) => process_gone(pid) && group_gone(group), _ => false, }, + // A provider that never replied within the deadline (a stalled or + // frozen host) is not a provider failure; once it is provably gone + // it is relaunched like a cut-off launch. A provider that rejected + // its startup or exited, and every fact saved before `noReply` + // existed, stays fenced. + ProviderStartupPhase::InitializationFailed if attempt.no_reply => { + match (attempt.process_id, attempt.process_group_id) { + (Some(pid), Some(group)) => process_gone(pid) && group_gone(group), + _ => false, + } + } ProviderStartupPhase::InitializationFailed => false, }; if !restorable || !orphaned { @@ -6188,8 +6229,13 @@ mod tests { signal: None, process_tree_retired: false, before_spawn: false, + no_reply: false, }; attempt.validate().unwrap(); + // `noReply` belongs only to an initialization that failed silently. + attempt.no_reply = true; + assert!(attempt.validate().is_err()); + attempt.no_reply = false; attempt.before_spawn = true; attempt.validate().unwrap(); attempt.before_spawn = false; @@ -6442,6 +6488,7 @@ mod tests { .observe_startup(ProviderStartupObservation::Failed { stage, child_exit: None, + no_reply: false, }) .unwrap(); }); @@ -6451,6 +6498,83 @@ mod tests { } } + fn failed_without_reply( + pid: u32, + stage: ProviderStartupStage, + no_reply: bool, + ) -> impl FnOnce(&mut CodexCommandExecutor) { + move |executor| { + spawned(pid)(executor); + executor + .observe_startup(ProviderStartupObservation::Failed { + stage, + child_exit: None, + no_reply, + }) + .unwrap(); + } + } + + #[test] + fn restart_retires_an_initialization_that_got_no_reply_once_its_provider_exited() { + for stage in [ + ProviderStartupStage::Initialize, + ProviderStartupStage::ThreadOpen, + ProviderStartupStage::ThreadRead, + ] { + let pid = exited_process_group(); + let directory = orphaned_startup_fixture( + "no-reply-startup-exited", + "turn_active", + failed_without_reply(pid, stage, true), + ); + let mut restarted = reload_and_retire(&directory); + assert!(restarted.assert_startup_admitted().is_ok()); + assert!(restarted.retained_events().unwrap().iter().any(|event| { + event.payload["code"] == "provider_startup_retired" + && event.payload["startup"]["noReply"] == true + })); + fs::remove_dir_all(directory).unwrap(); + } + } + + #[test] + fn restart_keeps_a_no_reply_fence_while_its_provider_may_run() { + let mut live = process_group_leader(); + let directory = orphaned_startup_fixture( + "no-reply-startup-live", + "turn_active", + failed_without_reply(live.id(), ProviderStartupStage::ThreadOpen, true), + ); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + let _ = live.kill(); + let _ = live.wait(); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_a_rejected_or_previously_saved_initialization_fence() { + // A provider that answered with a rejection (no `noReply`), and a + // fact saved by a runner that predates the field, stay fenced. + let pid = exited_process_group(); + let directory = orphaned_startup_fixture( + "rejected-startup", + "turn_active", + failed_without_reply(pid, ProviderStartupStage::ThreadOpen, false), + ); + let saved: Value = serde_json::from_slice( + &fs::read(directory.join(CODEX_PROVIDER_STATE_FILE)).unwrap(), + ) + .unwrap(); + assert!(saved["startupAttempt"].get("noReply").is_none()); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn restart_retires_an_intent_that_never_reached_the_spawn() { // The runner died while it was still checking the launch. 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 8b0d751e644..c0122536536 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 @@ -134,6 +134,150 @@ fn restart_relaunches_a_provider_whose_runner_died_before_admitting_it() { fs::remove_dir_all(directory).unwrap(); } +/// A provider relaunch whose reply never came within the runner's 30 s request +/// deadline (a stalled or frozen host: restart-chaos TM2 freezes the runner +/// and the provider while the monotonic clock keeps running). The runner +/// retires that provider and records the failed startup. A later runner must +/// relaunch it instead of refusing every restore until the user clicks Retry. +#[cfg(unix)] +#[test] +fn restart_relaunches_a_provider_whose_relaunch_timed_out_without_a_reply() { + let directory = temporary_directory("timed-out-provider-startup"); + let runner_config = durable_config(&directory); + let mut first = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + first + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": provider_config( + &directory, + &["--hold-turn", "--stall-first-thread-resume-ms", "33000"], + )}), + )) + .unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Survive a stalled provider relaunch."}), + )) + .unwrap(); + first.shutdown().unwrap(); + drop(first); + + // The next runner relaunches the provider; thread/resume gets no reply. + let mut stalled = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + let error = stalled + .execute(&command( + "command_resume_probe_1", + 9, + "runner.drain", + json!({}), + )) + .unwrap_err(); + assert!(error.to_string().contains("timed out"), "{error}"); + drop(stalled); + let state_path = directory.join("codex-provider-state.json"); + let saved: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + let fact = &saved["startupAttempt"]; + assert_eq!(fact["phase"], "initialization_failed"); + assert_eq!(fact["failedStage"], "thread_open"); + assert_eq!(fact["noReply"], true); + assert_eq!(fact["directChildExitObserved"], true); + + let resumes_before = call_count(&directory, "thread/resume"); + let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + recovered + .execute(&command( + "command_resume_probe_2", + 10, + "runner.drain", + json!({}), + )) + .expect("a provider launch that timed out without a reply no longer blocks recovery"); + assert_eq!(call_count(&directory, "thread/resume"), resumes_before + 1); + let events = recovered.retained_events().unwrap(); + assert!(events + .iter() + .any(|event| event.event_type == "harness.diagnostic" + && event.payload["code"] == "provider_startup_retired" + && event.payload["startup"]["phase"] == "initialization_failed")); + let saved: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + assert!(saved["startupAttempt"].is_null()); + recovered.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + +/// A provider that closed its stdout without replying failed; it was not a +/// silent stall. Its fence stays closed even after its process is gone. +#[cfg(unix)] +#[test] +fn restart_keeps_the_fence_for_a_provider_that_closed_stdout_without_a_reply() { + let directory = temporary_directory("closed-stdout-provider-startup"); + let runner_config = durable_config(&directory); + let mut first = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + first + .execute(&command( + "prepare", + 1, + "run.prepare", + json!({"provider": provider_config( + &directory, + &["--hold-turn", "--close-stdout-on-first-thread-resume"], + )}), + )) + .unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Fail a provider relaunch by closing stdout."}), + )) + .unwrap(); + first.shutdown().unwrap(); + drop(first); + + let mut failed = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + failed + .execute(&command( + "command_resume_probe_1", + 9, + "runner.drain", + json!({}), + )) + .unwrap_err(); + drop(failed); + let state_path = directory.join("codex-provider-state.json"); + let saved: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + let fact = &saved["startupAttempt"]; + assert_eq!(fact["phase"], "initialization_failed"); + assert_eq!(fact["failedStage"], "thread_open"); + assert!(fact.get("noReply").is_none(), "{fact}"); + + let mut restarted = CodexCommandExecutor::with_runner_config(&directory, &runner_config); + let error = restarted + .execute(&command( + "command_resume_probe_2", + 10, + "runner.drain", + json!({}), + )) + .unwrap_err(); + assert!(error + .to_string() + .contains("provider startup ownership remains unadmitted")); + fs::remove_dir_all(directory).unwrap(); +} + fn temporary_directory(label: &str) -> PathBuf { let directory = std::env::temp_dir().join(format!( "paperclip-runner-codex-{label}-{}-{}", diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 3619db989c4..e1a3e703eb5 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -760,7 +760,7 @@ expires. A time fault makes deadlines lapse during a run or across a restart. No | Fault | What the harness does | Deadlines it crosses | | --- | --- | --- | | TM1 freeze | `SIGSTOP` the server process alone for 95–150 s, then `SIGCONT`. The runner, its provider and Postgres keep running. | Controller-instance lease (90 s), execution-control deadline (60 s), legacy run lease (60 s). The runner keeps calling a server that does not answer. | -| TM2 sleep | `SIGSTOP` the server and every process under it for 20–45 s. While they are stopped, shift every database timestamp back by the age. Then `SIGCONT` the same processes. | Everything stored in the database: native and remote execution leases, scheduled retries, liveness and silence watchdogs, accounting closure, and pending waits. The processes' own monotonic timers barely moved. This stands in for a laptop that slept. | +| TM2 sleep | `SIGSTOP` the server and every process under it for 20–45 s. While they are stopped, shift every database timestamp back by the age. Then `SIGCONT` the same processes. | Everything stored in the database: native and remote execution leases, scheduled retries, liveness and silence watchdogs, accounting closure, and pending waits. The pause itself is 20–45 s, not the age. A stopped process's monotonic clock keeps running (it stops only in a real system suspend), so in-process deadlines shorter than the pause, such as the runner's 30 s provider request deadline, also lapse. TM2 therefore stands in for a laptop that slept and for a long process stall at the same time. | | TM3 aged outage | `SIGKILL` the server tree (as K1). While it is down, shift every database timestamp back by the age. Then start a replacement. | The same deadlines, reconciled by a fresh boot that finds runs, leases, wakes and waits that look hours old. | The age is 26 hours by default; `--time-age-hours` sets 1–72. The pause and the age are the cheapest diff --git a/tests/restart-chaos/time-faults.ts b/tests/restart-chaos/time-faults.ts index feb8cd632ef..6606a1a227e 100644 --- a/tests/restart-chaos/time-faults.ts +++ b/tests/restart-chaos/time-faults.ts @@ -11,8 +11,10 @@ * the runner keeps working and calling a server that does not answer. * - TM2 sleep: SIGSTOP the server and every process under it (runners, providers), * shift every timestamp in the wave's database back by the age (default 26 h), - * then SIGCONT. A laptop-sleep stand-in: the database says the age has passed, - * while the frozen processes' own monotonic timers barely moved. + * then SIGCONT. A laptop-sleep stand-in: the database says the age has passed. + * The processes' monotonic clocks keep running while they are stopped (only a + * real system suspend stops them), so in-process deadlines shorter than the + * 20–45 s pause also lapse, as in a long process stall. * - TM3 aged outage: SIGKILL the server tree (as K1), shift every timestamp back * by the age while it is down, then restart. The replacement must reconcile * runs, leases, wakes and waits that look hours old. With `--workloads G1