From fff70fe49ecd79a14bc6831d19571376db72f305 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:12:13 -0600 Subject: [PATCH] fix(runner): relaunch a provider whose runner died before admitting it (#19784) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs execute through `paperclip-runnerd`. When the control plane restarts, recovery starts a new runnerd process, and that process restores the provider (for example the Codex app server) from durable state. > - Each provider launch writes a durable startup fence before it spawns the process. Admission clears the fence. The fence stops a second runner from launching a provider while an unadmitted one may still run. > - A restart chaos wave sent SIGTERM and then SIGKILL to the control plane in quick succession. The second kill landed after the restored Codex provider was spawned and before it was admitted. > - The fence stayed in state with phase `spawned`. Every later runner refused to restore. The recovery probe's `runner.drain` failed with "provider startup ownership remains unadmitted", and the task needed a manual Retry. > - This pull request lets a new runner retire that fence when the recorded provider process and its process group have provably exited. It keeps every other fence closed. > - The benefit is that a double restart during provider relaunch continues automatically. ## Linked Issues or Issue Description **What happened?** A Codex run was active. A graceful restart (K2) suspended it, and recovery started a new runnerd that began restoring the provider. That runnerd spawned the Codex app server (pid recorded, phase `spawned`). A hard restart (K1) 9 s later killed it before admission. At the next boot, recovery chose `resume_dead_runner` and queued its `runner.drain` probe. The new runnerd failed the probe with `command_execution_failed: provider startup ownership remains unadmitted`. The run ended `native_session_recovery_failed`, and the task needed a Retry. The kept provider state had `lifecycle: "turn_active"` and a `startupAttempt` with `phase: "spawned"`, `trigger: "restore"`, `processId` = `processGroupId`, and no recorded exit. That process no longer existed. **Steps to reproduce** 1. Start a Codex task on a local runner. 2. Restart the control plane gracefully during the turn. 3. Kill the control plane again while recovery relaunches the provider, before the provider is admitted. **Expected behavior** Recovery relaunches the provider and continues the run. **Actual behavior** Every later runner refuses to restore, and the run fails. ## What Changed - `provider_backend.rs`: after a runner loads durable provider state, `retire_orphaned_startup_attempt` clears the startup fence only when all of these hold: - the attempt is in phase `spawned` (neither admitted nor failed); - the session is still restorable (`session_open`, `turn_active` or `provider_exited`); - signal 0 returns `ESRCH` for both the recorded process and its process group. It records a `provider_startup_retired` diagnostic event with the attempt, persists the state, and lets the normal restore path relaunch the provider. - Intent-phase fences (OpenCode). Three OpenCode runs after a K6 crash loop, and other runs on master, stopped at phase `intent` with no recorded process. Under load, the runner spends seconds verifying launch artifacts after it records intent. An intent alone could not show whether a process was started. - A new intent now carries `beforeSpawn: true`. - A new `SpawnStarting` observation clears the mark immediately before every provider spawn site in `start_with_tools_observed`. That clear is fsynced, but it emits no extra diagnostic event. - An intent still marked `beforeSpawn` (in a restorable session) proves that no process was started, so the fence is retired. - An intent that reached the spawn, or one written before this mark existed, stays fenced. - A runner that predates the field rejects the state as unknown (`deny_unknown_fields`), which fails closed. - `validate()` only allows the mark on an intent or on a spawn-stage failure. - The fence stays closed for: - an intent that reached the spawn, or one without the before-spawn mark; - a failed startup (initialization or spawn); - a session that recovery closed; - a process or group that may still run; - a permission error or an invalid id. - The existing test `failed_authenticated_identity_commit_keeps_the_durable_startup_fence` now uses a live process group instead of the fixed pid 123. That pid might not exist on a CI host, so the test would no longer prove a running provider keeps the fence. ## Verification - Red before the fix: the new integration test `restart_relaunches_a_provider_whose_runner_died_before_admitting_it` (fake Codex app server; turn active; a `spawned` restore fence with an exited process group; then the recovery `runner.drain` probe) failed with `provider startup ownership remains unadmitted`, the incident error. After the fix it passes. It shows one more `thread/resume`, the retired diagnostic, and a cleared fence. - Unit tests (unix-only module): retirement for each restorable lifecycle, and for an intent still before the spawn. A `SpawnStarting` outside its intent is rejected. The fence stays closed for: - an intent that reached the spawn; - an intent without the before-spawn mark; - a real child that survives its killed leader (process-group probe, no stubs); - a failed spawn; - a failed admission; - a closed session; - a live provider; - a surviving process group. - `cargo test -p paperclip-runner-core --test codex_provider`: 119 passed. The startup event counts are unchanged, because the spawn-start mark emits no event. The existing fence tests (`failed_provider_startup_is_persistently_fenced_before_another_process_can_resume`, `durable_recovery_closes_a_provider_that_reopens_a_settled_turn`) still pass. - `cargo test -p paperclip-runner-core --lib`: 570 passed. `cargo check --workspace --tests` passes. - `cargo fmt --check` passes. ## Risks - Process ids can be reused. A reused id only makes a gone process look alive, so the fence stays closed. Retirement needs `ESRCH` for both the pid and the group. - A provider process that moved to a new process group is outside this check. Before admission the runner has sent it no turn, so it cannot repeat provider work. - Intent fences written by older runners, including the kept incident states, have no before-spawn mark. They stay fenced and still need the Retry. New launches carry the mark. - A runner-side change takes effect on local runners after the runner build. Daytona needs a new image. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), used through Claude Code with tool use (shell, file edits, test execution). Default context window. Extended reasoning was available. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../crates/runner-core/src/codex_provider.rs | 7 + .../runner-core/src/provider_backend.rs | 414 +++++++++++++++++- .../runner-core/tests/codex_provider.rs | 95 ++++ ...runnerd-codex-transport-settlement.test.ts | 4 +- 4 files changed, 511 insertions(+), 9 deletions(-) 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 e4eb31abf5e..72997fd5ac6 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 @@ -192,6 +192,9 @@ pub(crate) enum ProviderStartupStage { } pub(crate) enum ProviderStartupObservation { + /// Launch checks passed and the process spawn is about to begin. Until + /// this is durable, a startup intent provably owns no process. + SpawnStarting, Spawned { process_id: u32, process_group_id: u32, @@ -1155,6 +1158,7 @@ impl CodexProvider { )); } let launch = verified_cursor_cloud_launch(profile)?; + observe(ProviderStartupObservation::SpawnStarting)?; SupervisedProcess::spawn_verified_with_environment_keys( &launch, Duration::from_secs(2), @@ -1176,6 +1180,7 @@ impl CodexProvider { )); } let launch = verified_opencode_launch(profile)?; + observe(ProviderStartupObservation::SpawnStarting)?; SupervisedProcess::spawn_verified_with_environment_keys( &launch, Duration::from_secs(2), @@ -1183,6 +1188,7 @@ impl CodexProvider { &environment_keys, ) } else { + observe(ProviderStartupObservation::SpawnStarting)?; SupervisedProcess::spawn_in_directory_with_environment_keys( &config.command, &launch_args, @@ -5773,6 +5779,7 @@ done None, None, &mut |observation| match observation { + ProviderStartupObservation::SpawnStarting => Ok(()), ProviderStartupObservation::Spawned { process_id, process_group_id, 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 c414061d6fb..af7e43def90 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 @@ -146,6 +146,11 @@ struct ProviderStartupAttempt { exit_code: Option, signal: Option, process_tree_retired: bool, + /// True from intent until the launch reaches the process spawn. Facts + /// written before this field existed omit it and stay fenced; a runner + /// that predates it rejects it as unknown, which also fails closed. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + before_spawn: bool, } impl ProviderStartupAttempt { @@ -196,7 +201,14 @@ impl ProviderStartupAttempt { None => false, }, }; + // 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 + || self.phase == ProviderStartupPhase::Intent + || (self.phase == ProviderStartupPhase::InitializationFailed + && self.failed_stage == Some(ProviderStartupStage::Spawn)); if self.schema != "paperclip.provider_startup.v1" + || !before_spawn_valid || uuid::Uuid::parse_str(&self.launch_id).is_err() || self.attempted_process_generation == 0 || self.authenticated_thread_id.is_some() @@ -429,6 +441,27 @@ fn restart_lost_turn_is_continuable(provider: &str) -> bool { matches!(provider, "codex" | "opencode") } +/// True only when signal 0 reports that no such process (or process group) +/// exists. A live process, a permission error, or an invalid id is not proof. +#[cfg(unix)] +fn local_process_definitely_gone(id: u32, group: bool) -> bool { + use rustix::process::{test_kill_process, test_kill_process_group, Pid}; + let Some(pid) = i32::try_from(id).ok().and_then(Pid::from_raw) else { + return false; + }; + let probe = if group { + test_kill_process_group(pid) + } else { + test_kill_process(pid) + }; + matches!(probe, Err(rustix::io::Errno::SRCH)) +} + +#[cfg(not(unix))] +fn local_process_definitely_gone(_id: u32, _group: bool) -> bool { + false +} + fn facade_provider_name(provider: &str) -> &'static str { match provider { "opencode" => "OpenCode", @@ -2111,6 +2144,12 @@ impl CodexCommandExecutor { } fn save_startup_fact(&mut self) -> Result<(), DurableRunnerError> { + self.save_startup_state(true) + } + + /// Persists the startup fence. `report` also queues the ownership + /// diagnostic; the spawn-start mark is internal and only needs fsync. + fn save_startup_state(&mut self, report: bool) -> Result<(), DurableRunnerError> { let outcome = (|| { let state = self .state @@ -2120,11 +2159,13 @@ impl CodexCommandExecutor { .startup_attempt .clone() .ok_or_else(|| DurableRunnerError::invalid("provider startup intent missing"))?; - state.push_terminal_event(NormalizedProviderEvent { - event_type: "harness.diagnostic".to_owned(), - priority: EventPriority::P0, - payload: json!({"code":"provider_startup_ownership", "startup": attempt}), - })?; + if report { + state.push_terminal_event(NormalizedProviderEvent { + event_type: "harness.diagnostic".to_owned(), + priority: EventPriority::P0, + payload: json!({"code":"provider_startup_ownership", "startup": attempt}), + })?; + } self.save_state() })(); if let Err(error) = &outcome { @@ -2164,6 +2205,7 @@ impl CodexCommandExecutor { exit_code: None, signal: None, process_tree_retired: false, + before_spawn: true, }); self.save_startup_fact() } @@ -2177,11 +2219,21 @@ impl CodexCommandExecutor { .as_mut() .and_then(|state| state.startup_attempt.as_mut()) .ok_or_else(|| DurableRunnerError::invalid("provider startup intent missing"))?; + let report = !matches!(observation, ProviderStartupObservation::SpawnStarting); match observation { + ProviderStartupObservation::SpawnStarting => { + if attempt.phase != ProviderStartupPhase::Intent || !attempt.before_spawn { + return Err(DurableRunnerError::invalid( + "provider spawn began outside its startup intent", + )); + } + attempt.before_spawn = false; + } ProviderStartupObservation::Spawned { process_id, process_group_id, } => { + attempt.before_spawn = false; attempt.phase = ProviderStartupPhase::Spawned; attempt.process_id = Some(process_id); attempt.process_group_id = Some(process_group_id); @@ -2194,7 +2246,7 @@ impl CodexCommandExecutor { attempt.signal = child_exit.as_ref().and_then(|fact| fact.signal); } } - self.save_startup_fact() + self.save_startup_state(report) } fn start_observed_provider( @@ -2302,9 +2354,74 @@ impl CodexCommandExecutor { fn restore_once(&mut self) -> Result<(), DurableRunnerError> { self.load_state_without_provider()?; + self.retire_orphaned_startup_attempt( + &|pid| local_process_definitely_gone(pid, false), + &|group| local_process_definitely_gone(group, true), + )?; self.restore_provider_if_needed() } + /// A runner killed during a provider launch leaves an unadmitted startup + /// fence in durable state while the session is still restorable. The + /// executor that loads that state owns no provider process. The fence is + /// retired, with a diagnostic, when nothing can still act for it: + /// + /// - 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. + /// + /// 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 + /// recovery deliberately closed. + fn retire_orphaned_startup_attempt( + &mut self, + process_gone: &dyn Fn(u32) -> bool, + group_gone: &dyn Fn(u32) -> bool, + ) -> Result<(), DurableRunnerError> { + if self.provider.is_some() || self.startup_evidence_error.is_some() { + return Ok(()); + } + let Some(state) = self.state.as_ref() else { + return Ok(()); + }; + let Some(attempt) = state.startup_attempt.clone() else { + return Ok(()); + }; + let restorable = matches!( + state.lifecycle.as_str(), + "session_open" | "turn_active" | "provider_exited" + ); + let orphaned = match attempt.phase { + // The runner died while checking the launch, before any spawn. + ProviderStartupPhase::Intent => attempt.before_spawn, + ProviderStartupPhase::Spawned => 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 { + return Ok(()); + } + let mut next = self + .state + .clone() + .ok_or_else(|| DurableRunnerError::invalid("provider startup state missing"))?; + next.startup_attempt = None; + next.push_terminal_event(NormalizedProviderEvent { + event_type: "harness.diagnostic".to_owned(), + priority: EventPriority::P0, + payload: json!({"code":"provider_startup_retired", "startup": attempt}), + })?; + self.persist_state(&next)?; + self.state = Some(next); + Ok(()) + } + fn load_state_without_provider(&mut self) -> Result<(), DurableRunnerError> { let path = self.state_path(); let mut file = match open_private_regular_file(&path) { @@ -5924,13 +6041,23 @@ mod tests { exit_code: None, signal: None, process_tree_retired: false, + before_spawn: false, }; attempt.validate().unwrap(); + attempt.before_spawn = true; + attempt.validate().unwrap(); + attempt.before_spawn = false; attempt.process_id = Some(123); attempt.process_group_id = Some(123); assert!(attempt.validate().is_err()); attempt.phase = ProviderStartupPhase::Spawned; attempt.validate().unwrap(); + attempt.before_spawn = true; + assert!( + attempt.validate().is_err(), + "a spawned launch is past the spawn" + ); + attempt.before_spawn = false; attempt.direct_child_exit_observed = true; assert!(attempt.validate().is_err()); attempt.phase = ProviderStartupPhase::InitializationFailed; @@ -5979,10 +6106,18 @@ mod tests { executor .begin_startup(ProviderStartupTrigger::Ensure, 1) .unwrap(); + // The unadmitted provider is still running, so a restarted executor + // must keep the fence closed. + #[cfg(unix)] + let mut live_provider = orphaned_startup::process_group_leader(); + #[cfg(unix)] + let live_pid = live_provider.id(); + #[cfg(not(unix))] + let live_pid = 123; executor .observe_startup(ProviderStartupObservation::Spawned { - process_id: 123, - process_group_id: 123, + process_id: live_pid, + process_group_id: live_pid, }) .unwrap(); let path = executor.state_path(); @@ -6023,9 +6158,272 @@ mod tests { .phase, ProviderStartupPhase::Spawned ); + #[cfg(unix)] + { + let _ = live_provider.kill(); + let _ = live_provider.wait(); + } fs::remove_dir_all(directory).unwrap(); } + #[cfg(unix)] + mod orphaned_startup { + use super::*; + + /// A child that leads its own process group, as provider launches do. + pub(super) fn process_group_leader() -> std::process::Child { + use std::os::unix::process::CommandExt; + std::process::Command::new("sleep") + .arg("30") + .process_group(0) + .spawn() + .expect("spawn a process group leader") + } + + /// A runner killed after spawning a provider and before admitting it. + fn orphaned_startup_fixture( + prefix: &str, + lifecycle: &str, + observe: impl FnOnce(&mut CodexCommandExecutor), + ) -> PathBuf { + let directory = + std::env::temp_dir().join(format!("paperclip-{prefix}-{}", uuid::Uuid::new_v4())); + let mut executor = CodexCommandExecutor::new(&directory); + 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("codex"); + state.lifecycle = lifecycle.to_owned(); + if lifecycle != "turn_active" { + state.active_provider_turn_id = None; + } + executor.state = Some(state); + executor + .begin_startup(ProviderStartupTrigger::Restore, 2) + .unwrap(); + observe(&mut executor); + drop(executor); + directory + } + + fn spawned(pid: u32) -> impl FnOnce(&mut CodexCommandExecutor) { + move |executor| { + executor + .observe_startup(ProviderStartupObservation::Spawned { + process_id: pid, + process_group_id: pid, + }) + .unwrap(); + } + } + + /// Loads the fixture as a new runner process would and applies the real + /// process probes. + fn reload_and_retire(directory: &std::path::Path) -> CodexCommandExecutor { + let mut restarted = CodexCommandExecutor::new(directory); + restarted.load_state_without_provider().unwrap(); + restarted + .retire_orphaned_startup_attempt( + &|pid| local_process_definitely_gone(pid, false), + &|group| local_process_definitely_gone(group, true), + ) + .unwrap(); + restarted + } + + fn exited_process_group() -> u32 { + let mut provider = process_group_leader(); + let pid = provider.id(); + provider.kill().unwrap(); + provider.wait().unwrap(); + pid + } + + #[test] + fn restart_retires_a_spawned_startup_whose_provider_process_exited() { + for lifecycle in ["turn_active", "session_open", "provider_exited"] { + let pid = exited_process_group(); + let directory = + orphaned_startup_fixture("orphaned-startup-exited", lifecycle, spawned(pid)); + let mut restarted = reload_and_retire(&directory); + assert!(restarted.assert_startup_admitted().is_ok(), "{lifecycle}"); + let events = restarted.retained_events().unwrap(); + assert!(events.iter().any(|event| { + event.event_type == "harness.diagnostic" + && event.payload["code"] == "provider_startup_retired" + && event.payload["startup"]["processId"] == pid + && event.payload["startup"]["phase"] == "spawned" + })); + let saved: CodexProviderState = + serde_json::from_slice(&fs::read(restarted.state_path()).unwrap()).unwrap(); + assert!(saved.startup_attempt.is_none()); + assert_eq!(saved.lifecycle, lifecycle); + fs::remove_dir_all(directory).unwrap(); + } + } + + #[test] + fn restart_keeps_the_fence_for_a_startup_that_failed() { + for (stage, child) in [ + (ProviderStartupStage::Spawn, None), + ( + ProviderStartupStage::Admission, + Some(exited_process_group()), + ), + ] { + let directory = + orphaned_startup_fixture("failed-startup", "turn_active", |executor| { + if let Some(pid) = child { + spawned(pid)(executor); + } + executor + .observe_startup(ProviderStartupObservation::Failed { + stage, + child_exit: None, + }) + .unwrap(); + }); + let restarted = reload_and_retire(&directory); + assert!(restarted.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. + let directory = + orphaned_startup_fixture("orphaned-startup-intent", "turn_active", |_| {}); + let saved: Value = serde_json::from_slice( + &fs::read(directory.join(CODEX_PROVIDER_STATE_FILE)).unwrap(), + ) + .unwrap(); + assert_eq!(saved["startupAttempt"]["phase"], "intent"); + assert_eq!(saved["startupAttempt"]["beforeSpawn"], 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"]["phase"] == "intent" + })); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_the_fence_for_an_intent_that_reached_the_spawn() { + // The spawn began: a process may exist that no record names. + let directory = + orphaned_startup_fixture("orphaned-startup-spawning", "turn_active", |executor| { + executor + .observe_startup(ProviderStartupObservation::SpawnStarting) + .unwrap(); + }); + let saved: Value = serde_json::from_slice( + &fs::read(directory.join(CODEX_PROVIDER_STATE_FILE)).unwrap(), + ) + .unwrap(); + assert_eq!(saved["startupAttempt"]["phase"], "intent"); + assert!(saved["startupAttempt"].get("beforeSpawn").is_none()); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_the_fence_for_an_intent_without_the_before_spawn_mark() { + // An intent written before the mark existed cannot prove anything. + let directory = + orphaned_startup_fixture("orphaned-startup-legacy", "turn_active", |_| {}); + let path = directory.join(CODEX_PROVIDER_STATE_FILE); + let mut saved: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + saved["startupAttempt"] + .as_object_mut() + .unwrap() + .remove("beforeSpawn"); + fs::write(&path, serde_json::to_vec(&saved).unwrap()).unwrap(); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn a_spawn_cannot_begin_twice_or_outside_its_intent() { + let directory = + orphaned_startup_fixture("spawn-starting-twice", "turn_active", |executor| { + executor + .observe_startup(ProviderStartupObservation::SpawnStarting) + .unwrap(); + assert!(executor + .observe_startup(ProviderStartupObservation::SpawnStarting) + .is_err()); + }); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_the_fence_for_a_session_recovery_closed() { + let directory = orphaned_startup_fixture( + "orphaned-startup-closed", + "closed", + spawned(exited_process_group()), + ); + let restarted = reload_and_retire(&directory); + assert!(restarted.assert_startup_admitted().is_err()); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_the_fence_while_the_provider_may_still_run() { + let mut provider = process_group_leader(); + let pid = provider.id(); + let directory = + orphaned_startup_fixture("orphaned-startup-live", "turn_active", spawned(pid)); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + provider.kill().unwrap(); + provider.wait().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } + + #[test] + fn restart_keeps_the_fence_while_a_child_survives_its_exited_leader() { + use std::io::{BufRead, BufReader}; + use std::os::unix::process::CommandExt; + // The leader starts a child in its own process group, reports the + // child pid, and is then killed and reaped. The child keeps the group. + let mut leader = std::process::Command::new("sh") + .args(["-c", "sleep 30 & echo $!; wait"]) + .process_group(0) + .stdout(std::process::Stdio::piped()) + .spawn() + .expect("spawn a provider stand-in with a child"); + let group = leader.id(); + let mut line = String::new(); + BufReader::new(leader.stdout.take().unwrap()) + .read_line(&mut line) + .unwrap(); + let child: u32 = line.trim().parse().unwrap(); + leader.kill().unwrap(); + leader.wait().unwrap(); + assert!(local_process_definitely_gone(group, false)); + assert!(!local_process_definitely_gone(group, true)); + + let directory = + orphaned_startup_fixture("orphaned-startup-group", "turn_active", spawned(group)); + assert!(reload_and_retire(&directory) + .assert_startup_admitted() + .is_err()); + + let _ = std::process::Command::new("kill") + .args(["-KILL", &child.to_string()]) + .status(); + fs::remove_dir_all(directory).unwrap(); + } + } + #[test] fn runtime_launch_rebinding_preserves_protected_arguments() { let before: Vec = vec![ 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 91a9e256d0b..d3bcf147e86 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 @@ -39,6 +39,101 @@ fn lock_receipt_limit_test() -> MutexGuard<'static, ()> { .unwrap_or_else(|poisoned| poisoned.into_inner()) } +/// A process group leader that has already exited, as a provider launch +/// leaves behind when its runner is killed before admitting it. +#[cfg(unix)] +fn exited_provider_process_group() -> u32 { + use std::os::unix::process::CommandExt; + let mut child = std::process::Command::new("sleep") + .arg("30") + .process_group(0) + .spawn() + .expect("spawn a provider stand-in"); + let pid = child.id(); + child.kill().unwrap(); + child.wait().unwrap(); + pid +} + +#[cfg(unix)] +#[test] +fn restart_relaunches_a_provider_whose_runner_died_before_admitting_it() { + let directory = temporary_directory("orphaned-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"])}), + )) + .unwrap(); + first + .execute(&command("open", 2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command( + "start", + 3, + "turn.start", + json!({"text": "Survive a restart during provider relaunch."}), + )) + .unwrap(); + first.shutdown().unwrap(); + drop(first); + + // The next runner began restoring the provider, spawned it, and was + // killed before admission. The spawned process group has since exited. + let state_path = directory.join("codex-provider-state.json"); + let mut state: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + assert_eq!(state["lifecycle"], "turn_active"); + let pid = exited_provider_process_group(); + let generation = state["providerProcessGeneration"].as_u64().unwrap() + 1; + state["startupAttempt"] = json!({ + "schema": "paperclip.provider_startup.v1", + "launchId": "afd87635-67c2-4b18-8ce1-9a97da4be2f4", + "phase": "spawned", + "trigger": "restore", + "attemptedProcessGeneration": generation, + "origin": null, + "command": {"commandId": "command_resume_probe_1", "controllerSeq": 9, "commandType": "runner.drain"}, + "configurationFingerprint": format!("sha256:{}", "a".repeat(64)), + "requestedThreadId": state["threadId"], + "authenticatedThreadId": null, + "processId": pid, + "processGroupId": pid, + "failedStage": null, + "directChildExitObserved": false, + "exitCode": null, + "signal": null, + "processTreeRetired": false, + }); + fs::write(&state_path, serde_json::to_vec(&state).unwrap()).unwrap(); + + 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("an exited, unadmitted provider launch 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"]["processId"] == pid)); + 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(); +} + fn temporary_directory(label: &str) -> PathBuf { let directory = std::env::temp_dir().join(format!( "paperclip-runner-codex-{label}-{}-{}", diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-settlement.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-settlement.test.ts index f06b0a3e8bf..f9b049f0c44 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-settlement.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-settlement.test.ts @@ -765,11 +765,13 @@ it.each([ const failedStop = failed.commands.find((command: { type: string; status: string }) => command.type === "turn.stop" && command.status === "failed"); for (const fact of facts) { + // Only an intent that has not reached the spawn carries beforeSpawn. expect(Object.keys(fact).sort()).toEqual([ "schema", "launchId", "phase", "trigger", "attemptedProcessGeneration", "origin", "command", "configurationFingerprint", "requestedThreadId", "authenticatedThreadId", "processId", "processGroupId", "failedStage", "directChildExitObserved", "exitCode", "signal", "processTreeRetired", + ...(fact === intent ? ["beforeSpawn"] : []), ].sort()); expect(fact).toMatchObject({ schema: "paperclip.provider_startup.v1", @@ -795,7 +797,7 @@ it.each([ }); } expect(intent).toMatchObject({ - phase: "intent", processId: null, processGroupId: null, + phase: "intent", beforeSpawn: true, processId: null, processGroupId: null, directChildExitObserved: false, failedStage: null, exitCode: null, signal: null, }); expect(spawned!.processId).toBeGreaterThan(0);