From fd2e7482c0c73174941db8d4817d713594e0624e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:19:41 -0600 Subject: [PATCH] fix(runner): stop a lost runnerd's provider and tools from a durable lineage journal (#20270) 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 > - A local native run uses one runner daemon (`paperclip-runnerd`). runnerd starts the provider (for example OpenCode), and the provider starts tool processes > - When runnerd dies in the middle of a tool call, the provider and its tools are re-parented. The only record of that process tree was in runnerd's memory > - Close then cannot prove that those processes stopped. It quarantines the session, and the next attempt stops with `native_session_cleanup_quarantined`: "Automatic recovery stopped / operator recovery required" > - Restarts and runner loss are product failures. Interrupted work must recover by itself, and it must never get two writers in one workspace > - This pull request makes runnerd journal its process lineage to disk, and makes close stop and prove gone every journalled process after runnerd is lost > - The benefit is that a killed runnerd leads to an automatic, safe retry within the normal attempt budget, instead of an operator stop ## Linked Issues or Issue Description Refs #20259 (runner loss classification and the lineage sampler panic). Refs #20260 (automatic continuation of an inert cleanup quarantine). **What happened?** A chat eval killed only the owned local `runnerd` process (SIGKILL) during an OpenCode tool call (`chat-features.runner-opencode.local.recovery.retry-exhausted`, `recovery.explain-stall` and `recovery.source-action`). Attempt 1 recorded `native_runner_process_exited` with "Retry scheduled 1/3". Attempt 2 stopped with `native_session_cleanup_quarantined`. The task was blocked with no owner, and the next action was operator recovery. **Expected behavior** After runnerd is lost, the product stops the provider and tool processes that runnerd started, proves them gone, and continues the run within its retry budget. When it cannot prove that, it keeps the fail-closed quarantine. **Steps to reproduce** 1. Start a local native OpenCode (or Codex) run. 2. Let the agent run a tool that leaves the provider's session (for example `nohup job &`). 3. Send SIGKILL to the `runnerd` PID only. 4. Close rejects with `runner did not durably suspend before checkpoint`, and the next attempt fails with `native_session_cleanup_quarantined`. **Paperclip version or commit** master at 4c15eea770 ## What Changed - `process_lineage.rs`: runnerd journals each supervised process's lineage to `/process-lineage/lineage--.json`. The journal holds the root, every member, and every group and session a member leads, each with its start time. The first record is written when the provider is spawned, before any tool can run. Later records are written by the existing 50 ms sampler only when the record changes. A teardown that proves the lineage gone removes the journal. - `process_lineage/journal.rs` (new module): the on-disk format, the write rules and the reaper `reap_journals`. The reaper refuses when runnerd never journalled (no writer marker) or when runnerd is still alive. It restores each journal into the same `LineageState` rules, sends TERM and then KILL only to processes whose PID and start time still match, and reports `verified` only when none remain. A journal that it cannot read refuses the proof. - Unsampled escapes: each record stores `sampled_through`, the newest process start that its read listed. The value is refreshed at most once a second while the membership is unchanged. The reaper refuses (`unattributed_orphan`) when a same-user orphan (parent PID 1) that is in no journal started after that point. Such a process can be a tool that started after the last read, so the reaper neither signals it nor ignores it. - A failed journal write removes the writer marker. Removal needs no free space. After that, no reap can verify from a record that may be missing a member. - `paperclip-runnerd`: durable mode enables the journal before any provider starts. A new `reap-lineage --state-dir --runner-pid` mode prints the report and exits 0 only when verified. - `runnerd-codex-transport.ts`: close samples whether runnerd exited unsuspended before close sent any signal. In that case, for a runnerd that this transport spawned on this host, it runs the reaper from the same executable. A verified report ends the session as a crash stop with its durable record intact. Recovery then continues the run. Any other result keeps the existing `NativeSessionCloseUnrecoverableError` quarantine. Launcher-owned (remote) and adopted runners do not use this path. - `runnerd-codex-transport.ts`: close does not queue a durable `runner.suspend` for a runnerd that it already knows is lost. Before, that suspend stayed pending in the saved control-plane state. A later runnerd on the same root then failed at startup with `terminal command result acknowledgement changed its durable identity`. The shared suspension barrier is unchanged for remote and adopted runners. - `runnerd-codex-transport.ts`: a runnerd that exits during close (for example because its provider snapshot is unreadable) is not a loss. It keeps the quarantine. - `runnerd-codex-transport.ts`: `#runnerHasExited` treated a signal-killed runnerd as alive, because `exitCode` is `null` and only `signalCode` is set. Close then spent its whole deadline trying to suspend a dead runner. It now checks `signalCode` too. - `fake-codex-app-server`: `--spawn-escaped-tool-job ` starts a tool job in its own session at turn start. - `doc/execution-semantics.md`: describes the journal, the reaper and the fallback. ## Verification - Rust, `cargo test -p paperclip-runner-core --lib process_lineage` (14 pass) and `process_supervisor` (20 pass). New cases: - `reaping_refuses_an_orphan_born_after_the_last_sample`: a tool starts after the last read, starts a new session and exits, and leaves a non-leader orphan job. The reap refuses and does not signal the job. It fails without the guard. - `a_failed_journal_write_withdraws_the_writer_marker` and `an_unchanged_record_is_rewritten_only_to_advance_sampled_through`. - `reaping_stops_a_dead_runners_provider_and_its_escaped_tool_job`: a real provider and a real `setsid` tool job, with a dead writer. The reap stops both, proves them gone and retires the journal and marker. - `reaping_refuses_without_a_marker_or_while_the_writer_lives` and `reaping_refuses_an_unreadable_journal`. - `a_journalled_record_restores_the_same_lineage`: after a round trip, the restored record still finds a re-parented job and its later child, and ignores an unrelated process. - TypeScript, live with the real debug runnerd: `stops a lost runnerd's provider and escaped tool job from its lineage journal`. - `journal=true`: SIGKILL runnerd mid-turn. The escaped tool job outlives runnerd. Close resolves, the job is gone, and the report is `verified`. Before the fix, close rejected with `runner did not durably suspend before checkpoint` and the job stayed alive. - `journal=false` (an older runnerd with no journal): close still rejects, and the job is not signalled. - `runner-lineage-reap.test.ts`: a report from an older runnerd, a malformed report, or an inconsistent `verified` report never counts as proof. - Turn-start cost: the journal adds one single-process read and one small file write when runnerd spawns its provider (about 1.5 ms in a macOS debug build), once for each runnerd. Nothing is added for each turn. - Paid runner E2E, `chat-features.runner-opencode.local.recovery.retry-exhausted`, on the eval branch with this fix (runs 42637596571 and 42644400434). Before the fix (run 42615758956), attempt 2 stopped with `native_session_cleanup_quarantined`. Now each of the three runnerd kills records `native_runner_process_exited`. Attempts 1 and 2 are `retryable_failure`, and attempt 3 is `terminal_failure` with cause `native_session_retry_exhausted` (attempt 3 of 3). The reaper ran after the first kill and verified. The cell still fails one eval setup check. The check waits for an *active* exhausted recovery action, but `execution-recovery-resolution.ts` settles that action as resolved and blocks the task (`preserve_without_replay_v1`). That is a separate grader decision, outside this PR. ## Risks - Medium. The change only affects a close where runnerd already exited without suspending and without a signal from close. Every other close is unchanged. - The reaper signals processes by PID. Each signal re-checks the PID's start time first, and a group or session is used only while its leader PID still names the recorded process (the same rules as the existing in-process teardown). - A crash stop leaves the durable root unsuspended. The coordinator already scheduled a retry for `native_runner_process_exited`. That retry now runs instead of failing on the quarantine. The paid E2E shows that attempts 2 and 3 start on the same run. - The escape guard can refuse when an unrelated same-user process becomes an orphan in the window of about one second after the last read. The fallback is today's quarantine. An orphan of a subreaper other than init is not seen by the guard. - The journal adds at most one small write each second for each runnerd while processes start on the host. This happens on the background sampler, not before turn start. - Previous-release story: no database or saved-session schema change. The journal is a new file in the runner state directory. A runnerd from an older release writes no journal and has no `reap-lineage` mode. Its loss keeps today's quarantine, and the existing board Retry and stopped-session proofs still apply. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use and code execution. ## 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 - [ ] All Paperclip CI gates are green - [ ] 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 --- doc/execution-semantics.md | 18 + .../src/bin/fake-codex-app-server.rs | 17 + .../runner-core/src/bin/paperclip-runnerd.rs | 56 ++ .../runner/crates/runner-core/src/lib.rs | 2 +- .../crates/runner-core/src/process_lineage.rs | 97 ++- .../src/process_lineage/journal.rs | 790 ++++++++++++++++++ .../runner-core/src/process_supervisor.rs | 2 +- .../src/live/runner-lineage-reap.test.ts | 23 + .../src/live/runner-lineage-reap.ts | 76 ++ .../src/live/runnerd-codex-transport.test.ts | 88 +- .../src/live/runnerd-codex-transport.ts | 56 +- 11 files changed, 1190 insertions(+), 35 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/process_lineage/journal.rs create mode 100644 packages/paperclip-runner/src/live/runner-lineage-reap.test.ts create mode 100644 packages/paperclip-runner/src/live/runner-lineage-reap.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 4d7260d3cdc..12197298fd0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1580,6 +1580,24 @@ and any run saved by an older release without these facts, keeps verified reattachment. A reattachment that loses its runner still requests the user's Retry as described above. +A local runnerd that dies (for example SIGKILL during a tool call) cannot +suspend, and its provider and tool processes are re-parented, so they cannot be +found from the controller's process records. Durable runnerd therefore journals +each provider's process lineage to `/process-lineage/` from the +moment the provider is spawned: the provider, every descendant, and every +process group or session a descendant leads, each named by PID and start time. +When close finds that runnerd exited without suspending and without a signal +from close, it runs `paperclip-runnerd reap-lineage` from the same executable. +The reaper refuses while runnerd is alive or when runnerd never journalled. It +signals only processes whose PID and start time still match the journal, and it +reports `verified` only when none remain. The journal also records how far the +process table was read. A same-user orphan that started after that point cannot +be attributed, so the reaper refuses without signalling it. A failed journal write removes the writer marker, so a +record that may be missing a member never verifies. A verified report ends the session as +a crash stop with its durable record intact, and recovery continues within the +normal attempt budget. Any other report keeps the cleanup quarantine below. +Launcher-owned (remote) and adopted runners are not reaped this way. + When a native attempt fails before launch because its predecessor's session is quarantined, explicit task Retry verifies the retained journal's actual owner. That owner must belong to the same company, task, agent, workspace, native session, 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 69603bfcf8d..382f76c02a3 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 @@ -774,6 +774,9 @@ fn run() -> Result<(), Box> { let linger_after_turn_start = args .iter() .any(|value| value == "--linger-after-turn-start"); + // A tool job that leaves the provider's session (like `nohup job &` in a + // Bash tool) and outlives the provider's process group. + let escaped_tool_pid_file = argument(&args, "--spawn-escaped-tool-job"); let hold_turn = args.iter().any(|value| value == "--hold-turn") || linger_after_turn_start; let held_output_count = argument(&args, "--held-output-count") .map(|value| value.parse::()) @@ -1661,6 +1664,20 @@ fn run() -> Result<(), Box> { { return Err("turn/start omitted the external sandbox boundary".into()); } + if let Some(pid_file) = escaped_tool_pid_file.as_deref() { + if turn_start_count == 0 { + std::process::Command::new("perl") + .args([ + "-e", + "use POSIX (); POSIX::setsid() or die; open(my $f, '>', \"$ARGV[0].tmp\") or die; print $f $$; close $f; rename(\"$ARGV[0].tmp\", $ARGV[0]) or die; exec '/bin/sh', '-c', 'while :; do sleep 1; done'", + pid_file, + ]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn()?; + } + } turn_start_count += 1; if durable_turn_ids { state.next_turn = state diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index 895de31d866..df70d461c93 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -519,6 +519,14 @@ fn run_durable(args: &[String]) -> Result<(), LocalRunnerError> { std::env::set_var(key, executable); } } + // Journal every provider's process lineage before any provider starts, so + // a controller can stop and prove gone the provider and its tools if this + // process dies. Without the journal the controller keeps its fail-closed + // cleanup quarantine; the run itself is unaffected. + #[cfg(any(target_os = "linux", target_os = "macos"))] + if let Err(error) = paperclip_runner_core::process_lineage::enable_journal(&state_dir) { + eprintln!("paperclip-runnerd: process lineage journal unavailable: {error}"); + } let executor = NativeProviderCommandExecutor::with_runner_config(state_dir, &config); paperclip_runner_core::durable::run_durable_runner_with_host_control( config, @@ -550,6 +558,51 @@ fn apply_task_memory_cgroup(args: &[String]) -> Result<(), LocalRunnerError> { Ok(()) } +/// `reap-lineage --state-dir --runner-pid [--grace-ms N] [--deadline-ms N]` +/// +/// Run by the controller after it confirmed that runnerd exited. Prints one +/// JSON report and exits 0 only when every journalled process is proven gone. +fn reap_lineage(args: &[String]) -> Result<(), LocalRunnerError> { + #[cfg(any(target_os = "linux", target_os = "macos"))] + { + let state_dir = PathBuf::from(value(args, "--state-dir")?); + let runner_pid = value(args, "--runner-pid")? + .parse::() + .ok() + .filter(|pid| *pid > 1) + .ok_or_else(|| LocalRunnerError::invalid("--runner-pid must be a process ID"))?; + let grace = Duration::from_millis(optional_u64(args, "--grace-ms")?.unwrap_or(1_000)); + let deadline = + Duration::from_millis(optional_u64(args, "--deadline-ms")?.unwrap_or(10_000)); + let report = paperclip_runner_core::process_lineage::reap_journals( + &state_dir, + runner_pid, + grace, + deadline.max(grace), + ); + println!( + "{}", + serde_json::to_string(&report) + .map_err(|error| LocalRunnerError::invalid(error.to_string()))? + ); + if report.verified { + Ok(()) + } else { + Err(LocalRunnerError::invalid(format!( + "process lineage not proven stopped: {}", + report.reason.unwrap_or("unknown") + ))) + } + } + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + { + let _ = args; + Err(LocalRunnerError::invalid( + "process lineage reaping is unavailable on this platform", + )) + } +} + fn run(args: &[String]) -> Result<(), LocalRunnerError> { #[cfg(unix)] if matches!( @@ -565,6 +618,9 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> { return paperclip_runner_core::stdio_mcp::run_proxy(Path::new(&args[1]), args.len() == 3) .map_err(|e| LocalRunnerError::invalid(e.to_string())); } + if args.first().map(String::as_str) == Some("reap-lineage") { + return reap_lineage(args); + } if args.first().map(String::as_str) == Some("--build-metadata") { println!("{}", metadata_with_provider_packs(args)?); return Ok(()); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs index 3e7d16b9f32..4c744e414c3 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/lib.rs @@ -32,7 +32,7 @@ pub mod managed_provider_backend; pub mod native_provider_backend; pub mod oom_score; #[cfg(any(target_os = "linux", target_os = "macos"))] -mod process_lineage; +pub mod process_lineage; pub mod process_supervisor; pub mod provider_backend; pub mod provider_bridge; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage.rs b/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage.rs index db9bfa68aea..6f90f15b286 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage.rs @@ -19,6 +19,15 @@ //! process's own session or an ancestor's group is never claimed through that //! session or group, because only groups and sessions that a recorded //! descendant leads are recorded. +//! +//! The record also has to survive the runner. When runnerd itself dies (for +//! example SIGKILL in the middle of a tool call), the provider and its tools +//! are re-parented and the in-memory record goes with runnerd. Durable runnerd +//! therefore journals every record to `/process-lineage/` from the +//! moment the provider is spawned. After runnerd's exit is confirmed, a later +//! `paperclip-runnerd reap-lineage` reads those journals, stops every process +//! that still matches a recorded identity, and proves the lineage gone. The +//! same PID-and-start-time rules apply, so a reused PID is never signalled. use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -26,6 +35,11 @@ use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; +mod journal; + +use journal::Journal; +pub use journal::{enable_journal, reap_journals, ReapReport, REAP_REPORT_SCHEMA}; + /// How often the background sampler reads the process table. A tool wrapper /// must live at least this long for the tracker to record its session. Claude /// Code's Bash wrapper sources a shell snapshot first and lives far longer. @@ -101,6 +115,20 @@ impl LineageState { } } + /// Starts from a root whose identity is already known, so the record is + /// complete before the background sampler's first read. + fn with_known_root(root: ProcessEntry) -> Self { + let mut state = Self { + root: Some((root.pid, root.started)), + ..Self::default() + }; + state.members.insert(root.pid, root.started); + if root.pgid == root.pid { + state.groups.insert(root.pid, root.started); + } + state + } + /// Updates membership from one table read and returns the live members. /// Callers must apply reads in the order they were taken; an older read /// would forget members that a newer one recorded. @@ -218,43 +246,76 @@ fn read_process_table() -> Option { } /// Reads the table while holding the state, so reads are applied in the -/// order they were taken. `None` when the table could not be read. -fn observe_now(state: &Mutex) -> Option { +/// order they were taken. `None` when the table could not be read. A changed +/// record is journalled under the same lock, so journal writes keep that order. +fn observe_now(state: &Mutex, journal: Option<&Journal>) -> Option { let mut state = state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); let table = read_process_table()?; - Some(state.observe(&table)) + let observation = state.observe(&table); + if let Some(journal) = journal { + // Every process this read listed started no later than its newest + // start time. A process born after it was not seen by this read. + let sampled_through = table + .entries() + .iter() + .map(|entry| entry.started) + .max() + .unwrap_or(0); + // A failed write withdraws the writer marker (see Journal::write), so + // no reaper can claim proof from the stale record on disk. + let _ = journal.write(&state, sampled_through); + } + Some(observation) } /// Records a supervised process's lineage from spawn until it is reaped. pub(crate) struct LineageTracker { state: Arc>, + journal: Option>, stop: Arc, sampler: Option>, } impl LineageTracker { - /// Starts recording. Fails when the sampler cannot run, so a caller never + /// Starts recording. Fails when the sampler cannot run, or when a + /// journalling runnerd cannot journal the root, so a caller never /// supervises a process whose tools it could not stop. pub(crate) fn start(root_pid: u32) -> Result { let root_pid = i32::try_from(root_pid) .map_err(|_| format!("process ID {root_pid} is out of range"))?; - let state = Arc::new(Mutex::new(LineageState::new(root_pid))); + // One read of the new process names the root by start time before + // any tool can run. It reads one process, not the whole table. + let initial = match paperclip_process_table::read(root_pid).entry() { + Some(root) if !root.zombie => LineageState::with_known_root(root), + _ => LineageState::new(root_pid), + }; + let journal = Journal::create(&initial) + .map_err(|error| format!("the process lineage journal could not be written: {error}"))? + .map(Arc::new); + let state = Arc::new(Mutex::new(initial)); let stop = Arc::new(AtomicBool::new(false)); let sampler_state = Arc::clone(&state); + let sampler_journal = journal.clone(); let sampler_stop = Arc::clone(&stop); let sampler = thread::Builder::new() .name("process-lineage".to_owned()) .spawn(move || { while !sampler_stop.load(Ordering::Acquire) { - let _ = observe_now(&sampler_state); + let _ = observe_now(&sampler_state, sampler_journal.as_deref()); thread::park_timeout(SAMPLE_INTERVAL); } }) - .map_err(|error| format!("the process lineage sampler could not start: {error}"))?; + .map_err(|error| { + if let Some(journal) = journal.as_ref() { + journal.retire(); + } + format!("the process lineage sampler could not start: {error}") + })?; Ok(Self { state, + journal, stop, sampler: Some(sampler), }) @@ -262,7 +323,7 @@ impl LineageTracker { /// Reads the table once now, so lineage visible at this moment is kept. pub(crate) fn sample(&self) { - let _ = observe_now(&self.state); + let _ = observe_now(&self.state, self.journal.as_deref()); } /// Stops every live process in the lineage and proves it is gone. @@ -272,12 +333,12 @@ impl LineageTracker { /// after `grace`, they send KILL. Fails with the members still alive at /// the deadline, or when the table could not be read to prove that none /// remain. - pub(crate) fn terminate(&self, grace: Duration, deadline: Duration) -> Result<(), String> { + pub(crate) fn terminate(&mut self, grace: Duration, deadline: Duration) -> Result<(), String> { let started = Instant::now(); let mut terminated: HashSet<(i32, u64)> = HashSet::new(); loop { let elapsed = started.elapsed(); - let Some(observation) = observe_now(&self.state) else { + let Some(observation) = observe_now(&self.state, self.journal.as_deref()) else { if elapsed >= deadline { return Err("the process table could not be read".to_owned()); } @@ -285,6 +346,12 @@ impl LineageTracker { continue; }; if observation.is_empty() { + // Stop journalling before retiring, so the sampler cannot + // write the file again after it is removed. + self.stop_sampler(); + if let Some(journal) = self.journal.as_ref() { + journal.retire(); + } return Ok(()); } if elapsed >= deadline { @@ -350,7 +417,7 @@ fn signal_member(entry: &ProcessEntry, signal: rustix::process::Signal) { mod tests { use super::*; - struct FixedTable { + pub(super) struct FixedTable { entries: Vec, unreadable: Vec, sessions: HashMap, @@ -368,7 +435,7 @@ mod tests { } } - fn entry(pid: i32, ppid: i32, pgid: i32, started: u64) -> ProcessEntry { + pub(super) fn entry(pid: i32, ppid: i32, pgid: i32, started: u64) -> ProcessEntry { ProcessEntry { pid, ppid, @@ -378,7 +445,7 @@ mod tests { } } - fn table(entries: Vec<(ProcessEntry, i32)>) -> FixedTable { + pub(super) fn table(entries: Vec<(ProcessEntry, i32)>) -> FixedTable { FixedTable { unreadable: Vec::new(), sessions: entries.iter().map(|(e, sid)| (e.pid, *sid)).collect(), @@ -387,12 +454,12 @@ mod tests { } impl LineageState { - fn observe_live(&mut self, table: &FixedTable) -> Vec { + pub(super) fn observe_live(&mut self, table: &FixedTable) -> Vec { self.observe(table).live } } - fn pids(entries: &[ProcessEntry]) -> Vec { + pub(super) fn pids(entries: &[ProcessEntry]) -> Vec { let mut pids: Vec = entries.iter().map(|e| e.pid).collect(); pids.sort_unstable(); pids diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage/journal.rs b/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage/journal.rs new file mode 100644 index 00000000000..361ab3dd594 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/process_lineage/journal.rs @@ -0,0 +1,790 @@ +//! The durable lineage journal and the reaper that reads it. +//! +//! When runnerd itself dies (for example SIGKILL in the middle of a tool call), +//! the provider and its tools are re-parented and the in-memory record goes +//! with runnerd. Durable runnerd therefore journals every record to +//! `/process-lineage/` from the moment the provider is spawned. +//! After runnerd's exit is confirmed, `paperclip-runnerd reap-lineage` reads +//! those journals, stops every process that still matches a recorded +//! identity, and proves the lineage gone. It applies the same PID-and-start +//! rules as the in-process tracker, so a reused PID is never signalled. +//! +//! This is its own module because it is a separate durability boundary: the +//! on-disk format, the write and withdrawal rules, and the cross-process +//! reaper change independently of the in-process sampling rules in the +//! parent module, which it reuses through `LineageState`. + +use std::collections::{HashMap, HashSet}; +use std::fs; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Mutex, OnceLock}; +use std::thread; +use std::time::{Duration, Instant}; + +use serde::{Deserialize, Serialize}; + +use super::{ + read_process_table, signal_member, LineageState, LiveProcessTable, ProcessEntry, ProcessTable, +}; + +impl LineageState { + fn journal_record(&self, writer: ProcessIdentity, sampled_through: u64) -> JournalRecord { + let sorted = |map: &HashMap| { + let mut entries: Vec<(i32, u64)> = map.iter().map(|(k, v)| (*k, *v)).collect(); + entries.sort_unstable(); + entries + }; + JournalRecord { + schema: JOURNAL_SCHEMA.to_owned(), + writer, + // A root whose start time is unknown cannot be told apart from a + // reused PID, so it is never journalled as an identity. + root: self.root.filter(|(_, started)| *started != 0), + members: sorted(&self.members), + groups: sorted(&self.groups), + sessions: sorted(&self.sessions), + sampled_through, + } + } + + fn from_journal_record(record: &JournalRecord) -> Self { + Self { + root: record.root, + members: record.members.iter().copied().collect(), + groups: record.groups.iter().copied().collect(), + sessions: record.sessions.iter().copied().collect(), + } + } +} + +/// Schema of one journalled lineage record. +const JOURNAL_SCHEMA: &str = "paperclip.runner.process-lineage.v1"; +/// Schema of the marker a journalling runnerd writes before any provider. +const WRITER_SCHEMA: &str = "paperclip.runner.process-lineage-writer.v1"; +/// Schema of the reaper's report. +pub const REAP_REPORT_SCHEMA: &str = "paperclip.runner.process-lineage-reap.v1"; +const JOURNAL_DIRECTORY_NAME: &str = "process-lineage"; + +static JOURNAL_DIRECTORY: OnceLock<(PathBuf, ProcessIdentity)> = OnceLock::new(); +static JOURNAL_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +/// A process named by PID and start time, which together cannot be reused. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct ProcessIdentity { + pid: i32, + started: u64, +} + +impl ProcessIdentity { + fn current() -> Option { + let pid = i32::try_from(std::process::id()).ok()?; + let entry = paperclip_process_table::read(pid).entry()?; + Some(Self { + pid, + started: entry.started, + }) + } + + /// True while the exact process still exists. An unreadable process is + /// treated as alive: its exit is not established. + fn is_alive(self) -> bool { + match paperclip_process_table::read(self.pid) { + paperclip_process_table::Read::Present(entry) => { + entry.started == self.started && !entry.zombie + } + paperclip_process_table::Read::Gone => false, + paperclip_process_table::Read::Unreadable => true, + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct JournalRecord { + schema: String, + writer: ProcessIdentity, + root: Option<(i32, u64)>, + members: Vec<(i32, u64)>, + groups: Vec<(i32, u64)>, + sessions: Vec<(i32, u64)>, + /// The newest process start time the recording read had listed. A + /// process that started later was never seen, so the record cannot + /// account for it. + #[serde(default)] + sampled_through: u64, +} + +#[derive(Debug, Serialize, Deserialize)] +struct WriterMarker { + schema: String, + writer: ProcessIdentity, +} + +/// One supervised process's journal file. +pub(super) struct Journal { + path: PathBuf, + marker: PathBuf, + writer: ProcessIdentity, + last: Mutex, +} + +#[derive(Default)] +struct JournalWrite { + /// The membership last written, without `sampled_through`. + membership: String, + sampled_through: u64, + at: Option, +} + +/// How often an unchanged record is rewritten only to advance +/// `sampled_through`. A shorter interval narrows the window a reaper must +/// treat as unsampled, at one small write per interval. +const SAMPLED_THROUGH_REFRESH: Duration = Duration::from_secs(1); + +impl Journal { + pub(super) fn create(state: &LineageState) -> io::Result> { + let Some((directory, writer)) = JOURNAL_DIRECTORY.get() else { + return Ok(None); + }; + let sequence = JOURNAL_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let journal = Self { + path: directory.join(format!("lineage-{}-{sequence}.json", writer.pid)), + marker: directory.join(format!("writer-{}.json", writer.pid)), + writer: *writer, + last: Mutex::new(JournalWrite::default()), + }; + // Nothing the root starts can predate the root itself. + let sampled_through = state.root.map_or(0, |(_, started)| started); + journal.write(state, sampled_through)?; + Ok(Some(journal)) + } + + /// Writes the record when membership changed, or when only + /// `sampled_through` advanced and the last write is older than the + /// refresh interval. A failed write removes the writer marker: removal + /// needs no free space, and without the marker no reaper can verify this + /// runnerd's lineage from a record that may be missing a member. + pub(super) fn write(&self, state: &LineageState, sampled_through: u64) -> io::Result<()> { + let membership = serde_json::to_string(&state.journal_record(self.writer, 0)) + .map_err(io::Error::other)?; + let mut last = self + .last + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let advanced = sampled_through > last.sampled_through + && last + .at + .is_none_or(|at| at.elapsed() >= SAMPLED_THROUGH_REFRESH); + if last.membership == membership && !advanced { + return Ok(()); + } + let sampled_through = sampled_through.max(last.sampled_through); + let written = serde_json::to_string(&state.journal_record(self.writer, sampled_through)) + .map_err(io::Error::other) + .and_then(|contents| write_private_atomically(&self.path, contents.as_bytes())); + if let Err(error) = written { + let _ = fs::remove_file(&self.marker); + return Err(error); + } + *last = JournalWrite { + membership, + sampled_through, + at: Some(Instant::now()), + }; + Ok(()) + } + + /// The lineage was proven gone; nothing is left for a reaper to stop. + pub(super) fn retire(&self) { + let _ = fs::remove_file(&self.path); + } +} + +fn write_private_atomically(path: &Path, contents: &[u8]) -> io::Result<()> { + use std::os::unix::fs::OpenOptionsExt; + let temporary = path.with_extension("json.tmp"); + let mut file = fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .mode(0o600) + .open(&temporary)?; + file.write_all(contents)?; + // No fsync: the journal must survive runnerd, not the host. A host + // restart ends every process the journal could name. + drop(file); + fs::rename(&temporary, path) +} + +/// Journals every lineage this runnerd records under `state_dir`, so the +/// processes can still be found and stopped after runnerd itself is gone. +/// Call once, before any supervised process starts. The writer marker tells +/// a reaper that this runnerd journals; without it, absence of journals +/// proves nothing. +pub fn enable_journal(state_dir: &Path) -> io::Result<()> { + use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; + let directory = std::path::absolute(state_dir)?.join(JOURNAL_DIRECTORY_NAME); + fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(&directory)?; + fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?; + let writer = ProcessIdentity::current() + .ok_or_else(|| io::Error::other("runnerd could not read its own process identity"))?; + let marker = serde_json::to_vec(&WriterMarker { + schema: WRITER_SCHEMA.to_owned(), + writer, + }) + .map_err(io::Error::other)?; + write_private_atomically( + &directory.join(format!("writer-{}.json", writer.pid)), + &marker, + )?; + JOURNAL_DIRECTORY + .set((directory, writer)) + .map_err(|_| io::Error::other("the process lineage journal is already enabled")) +} + +/// What one reap of a dead runnerd's journals proved. +#[derive(Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ReapReport { + pub schema: &'static str, + /// Every journalled process of the dead runnerd is proven gone. + pub verified: bool, + /// Why `verified` is false, or `None`. + pub reason: Option<&'static str>, + pub journals: usize, + pub signalled: usize, + /// PIDs still alive (or unreadable) at the deadline. + pub remaining: Vec, +} + +impl ReapReport { + fn refused(reason: &'static str) -> Self { + Self { + schema: REAP_REPORT_SCHEMA, + verified: false, + reason: Some(reason), + journals: 0, + signalled: 0, + remaining: Vec::new(), + } + } +} + +/// Stops and proves gone every process journalled by the runnerd that ran +/// with `state_dir` as PID `runner_pid`, after that runnerd has exited. +/// +/// Refuses (never signals) unless that runnerd journalled (its writer marker +/// exists) and is itself gone. Journals of other dead writers in the same +/// directory are reaped too; a live writer's journals are never touched. Each +/// process is signalled only while its PID and start time still match a +/// recorded identity: TERM first, KILL after `grace`. +pub fn reap_journals( + state_dir: &Path, + runner_pid: i32, + grace: Duration, + deadline: Duration, +) -> ReapReport { + let directory = state_dir.join(JOURNAL_DIRECTORY_NAME); + let marker_path = directory.join(format!("writer-{runner_pid}.json")); + let Some(marker) = fs::read(&marker_path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .filter(|marker| marker.schema == WRITER_SCHEMA && marker.writer.pid == runner_pid) + else { + return ReapReport::refused("journal_unavailable"); + }; + if marker.writer.is_alive() { + return ReapReport::refused("runner_alive"); + } + let Ok(listing) = fs::read_dir(&directory) else { + return ReapReport::refused("journal_unreadable"); + }; + let mut journals: Vec<(PathBuf, LineageState)> = Vec::new(); + // The oldest point after which the dead runnerd's sampler saw nothing. + let mut unsampled_since: Option = None; + for entry in listing { + let Ok(entry) = entry else { + return ReapReport::refused("journal_unreadable"); + }; + let name = entry.file_name(); + let Some(name) = name.to_str() else { continue }; + if !(name.starts_with("lineage-") && name.ends_with(".json")) { + continue; + } + let path = entry.path(); + let Some(record) = fs::read(&path) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .filter(|record| record.schema == JOURNAL_SCHEMA) + else { + // A journal that cannot be read may name live processes. + return ReapReport::refused("journal_unreadable"); + }; + if record.writer != marker.writer && record.writer.is_alive() { + continue; + } + if record.writer == marker.writer { + unsampled_since = Some( + unsampled_since.map_or(record.sampled_through, |since: u64| { + since.min(record.sampled_through) + }), + ); + } + journals.push((path, LineageState::from_journal_record(&record))); + } + let count = journals.len(); + let started = Instant::now(); + let mut signalled: HashSet<(i32, u64)> = HashSet::new(); + loop { + let elapsed = started.elapsed(); + let Some(table) = read_process_table() else { + if elapsed >= deadline { + return ReapReport { + journals: count, + signalled: signalled.len(), + ..ReapReport::refused("process_table_unreadable") + }; + } + thread::sleep(Duration::from_millis(10)); + continue; + }; + let mut live: Vec = Vec::new(); + let mut unresolved: Vec = Vec::new(); + for (_, state) in journals.iter_mut() { + let observation = state.observe(&table); + live.extend(observation.live); + unresolved.extend(observation.unresolved); + } + if live.is_empty() && unresolved.is_empty() { + // A tool that started after the last read, and whose parents have + // exited, matches no recorded parent, group or session: it is an + // orphan the record cannot account for. Its children stay under + // it, so the orphan names the whole unrecorded tree. Ownership + // cannot be proven either way, so it is never signalled, and the + // proof is refused. + if let Some(since) = unsampled_since { + let escaped = unattributed_orphans(&table, since, &journals); + if !escaped.is_empty() { + return ReapReport { + journals: count, + signalled: signalled.len(), + remaining: escaped, + ..ReapReport::refused("unattributed_orphan") + }; + } + } + for (path, _) in &journals { + let _ = fs::remove_file(path); + } + let _ = fs::remove_file(&marker_path); + return ReapReport { + schema: REAP_REPORT_SCHEMA, + verified: true, + reason: None, + journals: count, + signalled: signalled.len(), + remaining: Vec::new(), + }; + } + if elapsed >= deadline { + let mut remaining: Vec = live + .iter() + .map(|entry| entry.pid) + .chain(unresolved) + .collect(); + remaining.sort_unstable(); + remaining.dedup(); + return ReapReport { + journals: count, + signalled: signalled.len(), + remaining, + ..ReapReport::refused("processes_remain") + }; + } + for entry in &live { + let first_signal = signalled.insert((entry.pid, entry.started)); + if elapsed >= grace { + signal_member(entry, rustix::process::Signal::KILL); + } else if first_signal { + signal_member(entry, rustix::process::Signal::TERM); + } + } + thread::sleep(Duration::from_millis(10)); + } +} + +/// Live orphans (parent PID 1) of this user that started no earlier than +/// `since` and belong to no journal. An orphan of a subreaper other than init +/// is not seen here; such a host keeps the in-process teardown only. +fn unattributed_orphans( + table: &LiveProcessTable, + since: u64, + journals: &[(PathBuf, LineageState)], +) -> Vec { + let mut pids: Vec = table + .entries() + .iter() + .filter(|entry| { + !entry.zombie + && entry.ppid == 1 + && entry.started >= since + && !journals + .iter() + .any(|(_, state)| state.members.get(&entry.pid) == Some(&entry.started)) + && owned_by_this_user(entry.pid) + }) + .map(|entry| entry.pid) + .collect(); + pids.sort_unstable(); + pids +} + +#[cfg(target_os = "linux")] +fn owned_by_this_user(pid: i32) -> bool { + use std::os::unix::fs::MetadataExt; + fs::metadata(format!("/proc/{pid}")) + .is_ok_and(|metadata| metadata.uid() == rustix::process::geteuid().as_raw()) +} + +/// macOS lists only processes whose details this user can read. +#[cfg(not(target_os = "linux"))] +fn owned_by_this_user(_pid: i32) -> bool { + true +} + +#[cfg(test)] +mod tests { + use super::super::tests::{entry, pids, table}; + use super::super::SAMPLE_INTERVAL; + use super::*; + + #[test] + fn a_journalled_record_restores_the_same_lineage() { + let mut state = LineageState::new(100); + state.observe(&table(vec![ + (entry(100, 10, 100, 10), 1), + (entry(200, 100, 200, 20), 200), + (entry(201, 200, 200, 21), 200), + ])); + let writer = ProcessIdentity { + pid: 10, + started: 5, + }; + let record = state.journal_record(writer, 0); + let json = serde_json::to_string(&record).unwrap(); + let restored: JournalRecord = serde_json::from_str(&json).unwrap(); + assert_eq!(restored.writer, writer); + let mut restored = LineageState::from_journal_record(&restored); + // runnerd (10) is gone and the wrapper exited: provider 100 and job + // 201 are re-parented to init, and a later job 202 was started. + let after = table(vec![ + (entry(1, 0, 1, 1), 1), + (entry(100, 1, 100, 10), 1), + (entry(201, 1, 200, 21), 200), + (entry(202, 201, 200, 30), 200), + (entry(300, 1, 300, 40), 300), + ]); + assert_eq!(pids(&restored.observe_live(&after)), vec![100, 201, 202]); + } + + #[test] + fn an_unknown_root_start_is_never_journalled_as_an_identity() { + let state = LineageState::new(100); + let record = state.journal_record( + ProcessIdentity { + pid: 10, + started: 5, + }, + 0, + ); + assert_eq!(record.root, None); + } + + fn temporary_directory(name: &str) -> PathBuf { + let directory = std::env::temp_dir().join(format!( + "paperclip-lineage-{name}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir_all(directory.join(JOURNAL_DIRECTORY_NAME)).unwrap(); + directory + } + + fn write_marker(state_dir: &Path, writer: ProcessIdentity) { + let marker = WriterMarker { + schema: WRITER_SCHEMA.to_owned(), + writer, + }; + fs::write( + state_dir + .join(JOURNAL_DIRECTORY_NAME) + .join(format!("writer-{}.json", writer.pid)), + serde_json::to_vec(&marker).unwrap(), + ) + .unwrap(); + } + + /// A process that existed and has exited, so its identity is dead. + fn exited_writer() -> ProcessIdentity { + let mut child = std::process::Command::new("/bin/sleep") + .arg("5") + .spawn() + .unwrap(); + let pid = i32::try_from(child.id()).unwrap(); + let started = paperclip_process_table::read(pid).entry().unwrap().started; + child.kill().unwrap(); + child.wait().unwrap(); + ProcessIdentity { pid, started } + } + + #[test] + fn reaping_refuses_without_a_marker_or_while_the_writer_lives() { + let directory = temporary_directory("refuse"); + let report = reap_journals(&directory, 4242, Duration::ZERO, Duration::ZERO); + assert_eq!(report.reason, Some("journal_unavailable")); + assert!(!report.verified); + + let alive = ProcessIdentity::current().unwrap(); + write_marker(&directory, alive); + let report = reap_journals(&directory, alive.pid, Duration::ZERO, Duration::ZERO); + assert_eq!(report.reason, Some("runner_alive")); + assert!(!report.verified); + let _ = fs::remove_dir_all(&directory); + } + + #[test] + fn reaping_refuses_an_unreadable_journal() { + let directory = temporary_directory("unreadable"); + let writer = exited_writer(); + write_marker(&directory, writer); + fs::write( + directory + .join(JOURNAL_DIRECTORY_NAME) + .join("lineage-1-0.json"), + b"{", + ) + .unwrap(); + let report = reap_journals(&directory, writer.pid, Duration::ZERO, Duration::ZERO); + assert_eq!(report.reason, Some("journal_unreadable")); + let _ = fs::remove_dir_all(&directory); + } + + /// runnerd is SIGKILLed mid-tool: its provider and a tool job that left + /// the provider's session outlive it. The journal alone finds and stops + /// both, and proves them gone. + #[test] + fn reaping_stops_a_dead_runners_provider_and_its_escaped_tool_job() { + use std::os::unix::process::CommandExt; + let directory = temporary_directory("reap"); + let pid_file = directory.join("job.pid"); + // The wrapper lives past several samples, as a real tool wrapper does. + let tool = r#"sleep 0.3; nohup /bin/sh -c 'while :; do sleep 1; done' >/dev/null 2>&1 & echo $! > "$0.tmp"; mv "$0.tmp" "$0"; exit 0"#; + let wrapper = format!( + "use POSIX (); POSIX::setsid() or die; exec '/bin/sh', '-c', q{{{tool}}}, $ARGV[0];" + ); + let mut provider = std::process::Command::new("/bin/sh") + .args([ + "-c", + r#"sleep 0.2; perl -e "$1" "$2"; while :; do sleep 1; done"#, + "provider", + &wrapper, + &pid_file.display().to_string(), + ]) + .process_group(0) + .spawn() + .unwrap(); + let provider_pid = i32::try_from(provider.id()).unwrap(); + let root = paperclip_process_table::read(provider_pid).entry().unwrap(); + let mut state = LineageState::with_known_root(root); + let deadline = Instant::now() + Duration::from_secs(10); + let job = loop { + let live = state.observe(&read_process_table().unwrap()).live; + if let Some(pid) = fs::read_to_string(&pid_file) + .ok() + .and_then(|text| text.trim().parse::().ok()) + { + if let Some(job) = live.iter().find(|entry| entry.pid == pid) { + break *job; + } + } + assert!(Instant::now() < deadline, "tool job was not recorded"); + thread::sleep(SAMPLE_INTERVAL); + }; + assert_ne!(job.pgid, provider_pid); + + let writer = exited_writer(); + write_marker(&directory, writer); + let journal = directory + .join(JOURNAL_DIRECTORY_NAME) + .join(format!("lineage-{}-0.json", writer.pid)); + fs::write( + &journal, + serde_json::to_vec(&state.journal_record(writer, newest_start())).unwrap(), + ) + .unwrap(); + + let report = reap_journals( + &directory, + writer.pid, + Duration::from_millis(500), + Duration::from_secs(10), + ); + let _ = provider.wait(); + let job_alive = paperclip_process_table::read(job.pid) + .entry() + .is_some_and(|current| current.started == job.started && !current.zombie); + if job_alive { + if let Some(group) = rustix::process::Pid::from_raw(job.pgid) { + let _ = rustix::process::kill_process_group(group, rustix::process::Signal::KILL); + } + } + assert!(!job_alive, "tool job {} outlived the reap", job.pid); + assert!(report.verified, "{report:?}"); + assert_eq!(report.journals, 1); + assert!(report.signalled >= 2, "{report:?}"); + assert!(!journal.exists(), "a proven journal is retired"); + assert!(!directory + .join(JOURNAL_DIRECTORY_NAME) + .join(format!("writer-{}.json", writer.pid)) + .exists()); + let _ = fs::remove_dir_all(&directory); + } + + /// The newest start time a table read lists now, as the sampler records. + fn newest_start() -> u64 { + read_process_table() + .unwrap() + .entries() + .iter() + .map(|entry| entry.started) + .max() + .unwrap() + } + + /// A tool forked and left its session after the dead runnerd's last + /// read, and its parents are gone: nothing recorded leads to it. The reap + /// must not claim proof, and must not signal a process it cannot + /// attribute. + #[test] + fn reaping_refuses_an_orphan_born_after_the_last_sample() { + let directory = temporary_directory("escape"); + let writer = exited_writer(); + write_marker(&directory, writer); + let provider = exited_writer(); + let mut state = LineageState::default(); + state.root = Some((provider.pid, provider.started)); + let journal = directory + .join(JOURNAL_DIRECTORY_NAME) + .join(format!("lineage-{}-0.json", writer.pid)); + fs::write( + &journal, + serde_json::to_vec(&state.journal_record(writer, newest_start())).unwrap(), + ) + .unwrap(); + // runnerd died right after the tool started, before any read saw it. + // The tool starts a new session, starts a job and exits, so the job + // is re-parented to init and leads neither its group nor its session. + let pid_file = directory.join("orphan.pid"); + let status = std::process::Command::new("perl") + .args([ + "-e", + "use POSIX (); exit 0 if fork(); POSIX::setsid() or die; exit 0 if fork(); sleep 1 while getppid() != 1; open(my $f, '>', \"$ARGV[0].tmp\") or die; print $f $$; close $f; rename(\"$ARGV[0].tmp\", $ARGV[0]) or die; open(STDIN, '<', '/dev/null'); open(STDOUT, '>', '/dev/null'); open(STDERR, '>', '/dev/null'); exec '/bin/sh', '-c', 'while :; do sleep 1; done'", + &pid_file.display().to_string(), + ]) + .status() + .unwrap(); + assert!(status.success()); + let deadline = Instant::now() + Duration::from_secs(10); + let orphan = loop { + if let Some(entry) = fs::read_to_string(&pid_file) + .ok() + .and_then(|text| text.trim().parse::().ok()) + .and_then(|pid| paperclip_process_table::read(pid).entry()) + .filter(|entry| entry.ppid == 1 && entry.pgid != entry.pid) + { + break entry; + } + assert!(Instant::now() < deadline, "orphan job did not start"); + thread::sleep(Duration::from_millis(20)); + }; + let report = reap_journals( + &directory, + writer.pid, + Duration::ZERO, + Duration::from_secs(5), + ); + let alive = ProcessIdentity { + pid: orphan.pid, + started: orphan.started, + } + .is_alive(); + if let Some(group) = rustix::process::Pid::from_raw(orphan.pgid) { + let _ = rustix::process::kill_process_group(group, rustix::process::Signal::KILL); + } + let _ = fs::remove_dir_all(&directory); + assert_eq!(report.reason, Some("unattributed_orphan"), "{report:?}"); + assert!(!report.verified); + assert!(report.remaining.contains(&orphan.pid), "{report:?}"); + assert!(alive, "an unattributed process is never signalled"); + } + + /// A record that could not be written may be missing a member. The + /// failure withdraws the writer marker, so no reap can verify. + #[test] + fn a_failed_journal_write_withdraws_the_writer_marker() { + let directory = temporary_directory("write-failure"); + let writer = exited_writer(); + write_marker(&directory, writer); + let marker = directory + .join(JOURNAL_DIRECTORY_NAME) + .join(format!("writer-{}.json", writer.pid)); + let journal = Journal { + path: directory.join("missing").join("lineage.json"), + marker: marker.clone(), + writer, + last: Mutex::new(JournalWrite::default()), + }; + assert!(journal.write(&LineageState::new(100), 1).is_err()); + assert!(!marker.exists()); + let report = reap_journals(&directory, writer.pid, Duration::ZERO, Duration::ZERO); + assert_eq!(report.reason, Some("journal_unavailable")); + let _ = fs::remove_dir_all(&directory); + } + + #[test] + fn an_unchanged_record_is_rewritten_only_to_advance_sampled_through() { + let directory = temporary_directory("refresh"); + let path = directory.join(JOURNAL_DIRECTORY_NAME).join("lineage.json"); + let journal = Journal { + path: path.clone(), + marker: directory.join("marker"), + writer: ProcessIdentity { + pid: 10, + started: 5, + }, + last: Mutex::new(JournalWrite::default()), + }; + let state = LineageState::new(100); + let read = || { + serde_json::from_slice::(&fs::read(&path).unwrap()) + .unwrap() + .sampled_through + }; + journal.write(&state, 7).unwrap(); + assert_eq!(read(), 7); + // Within the refresh interval an advance alone is not written. + journal.write(&state, 9).unwrap(); + assert_eq!(read(), 7); + journal.last.lock().unwrap().at = Some(Instant::now() - SAMPLED_THROUGH_REFRESH); + journal.write(&state, 9).unwrap(); + assert_eq!(read(), 9); + let _ = fs::remove_dir_all(&directory); + } +} 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 eff002ac44f..64d27043b29 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 @@ -1282,7 +1282,7 @@ impl SupervisedProcess { /// never treats the provider as stopped while its tools still run. fn terminate_lineage(&mut self, grace: Duration) -> Result<(), LocalRunnerError> { #[cfg(any(target_os = "linux", target_os = "macos"))] - if let Some(lineage) = self.lineage.as_ref() { + if let Some(lineage) = self.lineage.as_mut() { lineage .terminate(grace, grace + LINEAGE_KILL_PROOF_WINDOW) .map_err(|reason| { diff --git a/packages/paperclip-runner/src/live/runner-lineage-reap.test.ts b/packages/paperclip-runner/src/live/runner-lineage-reap.test.ts new file mode 100644 index 00000000000..2f1663a4681 --- /dev/null +++ b/packages/paperclip-runner/src/live/runner-lineage-reap.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from "vitest"; +import { parseRunnerLineageReapReport, RUNNER_LINEAGE_REAP_SCHEMA } from "./runner-lineage-reap.js"; + +describe("parseRunnerLineageReapReport", () => { + const verified = { schema: RUNNER_LINEAGE_REAP_SCHEMA, verified: true, reason: null, journals: 1, signalled: 2, remaining: [] }; + + it("accepts the report runnerd prints last", () => { + expect(parseRunnerLineageReapReport(`warning\n${JSON.stringify(verified)}\n`)).toEqual(verified); + const refused = { ...verified, verified: false, reason: "processes_remain", remaining: [42] }; + expect(parseRunnerLineageReapReport(JSON.stringify(refused))).toEqual(refused); + }); + + it.each([ + ["no output", null], + ["an old runnerd's usage error", "unknown argument reap-lineage"], + ["another schema", JSON.stringify({ ...verified, schema: "other" })], + ["a verified report that names a survivor", JSON.stringify({ ...verified, remaining: [42] })], + ["a verified report with a refusal reason", JSON.stringify({ ...verified, reason: "runner_alive" })], + ["a malformed survivor list", JSON.stringify({ ...verified, verified: false, remaining: ["42"] })], + ])("never treats %s as proof", (_label, stdout) => { + expect(parseRunnerLineageReapReport(stdout)?.verified ?? false).toBe(false); + }); +}); diff --git a/packages/paperclip-runner/src/live/runner-lineage-reap.ts b/packages/paperclip-runner/src/live/runner-lineage-reap.ts new file mode 100644 index 00000000000..ae12f1b75d4 --- /dev/null +++ b/packages/paperclip-runner/src/live/runner-lineage-reap.ts @@ -0,0 +1,76 @@ +import { execFile } from "node:child_process"; + +/** Report printed by `paperclip-runnerd reap-lineage`. */ +export const RUNNER_LINEAGE_REAP_SCHEMA = "paperclip.runner.process-lineage-reap.v1"; + +export interface RunnerLineageReapReport { + schema: typeof RUNNER_LINEAGE_REAP_SCHEMA; + verified: boolean; + reason: string | null; + journals: number; + signalled: number; + remaining: number[]; +} + +/** + * Stops and proves gone every provider and tool process that a local runnerd + * journalled, after that runnerd has exited. + * + * runnerd is the only owner of its provider's process tree. When runnerd is + * killed (for example mid-tool), the provider and its tools are re-parented + * and can no longer be found from the controller's process records. Durable + * runnerd journals each lineage (PID, group and session with their start + * times) under `/process-lineage/`; the same runnerd executable + * reads that journal, signals only processes whose identity still matches, + * and reports whether none remain. + * + * Returns the report, or `null` when no report could be read (for example a + * runnerd built before the journal existed). Any result other than + * `verified: true` keeps the caller's fail-closed handling. + */ +export async function reapExitedRunnerLineage(input: { + runnerBinaryPath: string; + stateDirectory: string; + runnerPid: number; + timeoutMs: number; +}): Promise { + if (!Number.isSafeInteger(input.runnerPid) || input.runnerPid <= 1) return null; + // Reaping is bounded by the controller's close deadline. runnerd's own + // deadline is set below it so a report always arrives before the kill. + const timeoutMs = Math.max(1_000, Math.floor(input.timeoutMs)); + const deadlineMs = Math.max(500, timeoutMs - 500); + const graceMs = Math.min(1_000, Math.floor(deadlineMs / 2)); + const stdout = await new Promise((resolveOutput) => { + execFile( + input.runnerBinaryPath, + [ + "reap-lineage", + "--state-dir", input.stateDirectory, + "--runner-pid", String(input.runnerPid), + "--grace-ms", String(graceMs), + "--deadline-ms", String(deadlineMs), + ], + { timeout: timeoutMs, killSignal: "SIGKILL", maxBuffer: 64 * 1024, env: { PATH: process.env.PATH ?? "" } }, + // A refusal exits non-zero but still prints its report. + (_error, output) => resolveOutput(typeof output === "string" ? output : null), + ); + }); + return parseRunnerLineageReapReport(stdout); +} + +export function parseRunnerLineageReapReport(stdout: string | null): RunnerLineageReapReport | null { + const line = stdout?.trim().split("\n").at(-1); + if (!line) return null; + try { + const value = JSON.parse(line) as Record; + if (value.schema !== RUNNER_LINEAGE_REAP_SCHEMA || typeof value.verified !== "boolean" + || !Number.isSafeInteger(value.journals) || !Number.isSafeInteger(value.signalled) + || !Array.isArray(value.remaining) || !value.remaining.every(pid => Number.isSafeInteger(pid)) + || (value.reason !== null && typeof value.reason !== "string")) return null; + // A verified report never names a remaining process. + if (value.verified && (value.remaining.length > 0 || value.reason !== null)) return null; + return value as unknown as RunnerLineageReapReport; + } catch { + return null; + } +} 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 d9627e81fa6..8a8254cd41e 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -6028,13 +6028,34 @@ it("surfaces a runner exit while provider-ingress readiness is still pending", a } }); -it("rejects the notification stream promptly when runnerd exits after accepting a turn", async () => { +function processAlive(pid: number | null | undefined): boolean { + if (!pid) return false; + try { + process.kill(pid, 0); + } catch { + return false; + } + // An exited child of this test process stays a zombie until reaped. + try { + return !execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim().startsWith("Z"); + } catch { + return false; + } +} + +it.each([ + { journal: true }, + // A runnerd that never journalled (an older build) proves nothing: the + // cleanup stays fail-closed, as before the journal existed. + { journal: false }, +])("stops a lost runnerd's provider and escaped tool job from its lineage journal (journal=$journal)", async ({ journal }) => { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-exit-stream-")); + const jobPidFile = join(stateDirectory, "escaped-tool-job.pid"); const diagnostics: string[] = []; const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, - codexArgs: fakeCodexArgs(stateDirectory, "--linger-after-turn-start"), + codexArgs: fakeCodexArgs(stateDirectory, "--linger-after-turn-start", "--spawn-escaped-tool-job", jobPidFile), stateDirectory, closeGraceMs: 400, onDiagnostic: (message) => diagnostics.push(message), @@ -6043,6 +6064,8 @@ it("rejects the notification stream promptly when runnerd exits after accepting success: true, contentItems: [], })); + let jobPid: number | null = null; + let providerPid: number | null = null; try { await bundle.transport.request("initialize", {}); await bundle.transport.request("thread/start", { @@ -6066,8 +6089,23 @@ it("rejects the notification stream promptly when runnerd exits after accepting .notifications() [Symbol.asyncIterator](); await expectTurnStarted(notifications); + await vi.waitFor(async () => { + jobPid = Number((await readFile(jobPidFile, "utf8")).trim()); + expect(processAlive(jobPid)).toBe(true); + }, { timeout: 10_000, interval: 20 }); + // The tool job left the provider's session; give the lineage sampler + // (50ms) several reads to record it, as a real tool runs far longer. + await new Promise((resolveWait) => setTimeout(resolveWait, 400)); const runnerPid = bundle.evidence().runnerPid; expect(runnerPid).not.toBeNull(); + const lineage = join(stateDirectory, "runner", "process-lineage"); + const journals = (await readdir(lineage)).filter((name) => name.startsWith("lineage-")); + expect(journals).toHaveLength(1); + const recorded = JSON.parse(await readFile(join(lineage, journals[0]!), "utf8")); + providerPid = recorded.root[0]; + expect(processAlive(providerPid)).toBe(true); + expect(recorded.members.map((member: [number, number]) => member[0])).toContain(jobPid); + if (!journal) await rm(lineage, { recursive: true, force: true }); process.kill(runnerPid!, "SIGKILL"); await expect( Promise.race([ @@ -6080,27 +6118,47 @@ it("rejects the notification stream promptly when runnerd exits after accepting ), ]), ).rejects.toThrow("native_runner_process_exited"); - } finally { - try { + // The fake provider exits when runnerd's stdin pipe closes; the tool job + // that left its session is re-parented and keeps running. + expect(processAlive(jobPid)).toBe(true); + const runnerState = () => readFile(join(stateDirectory, "runner", "runner-state.json"), "utf8") + .then((text) => JSON.parse(text)); + if (journal) { + await bundle.transport.close(); + expect(processAlive(providerPid)).toBe(false); + expect(processAlive(jobPid)).toBe(false); + const report = diagnostics.find((message) => message.startsWith("native_session_runner_loss_lineage ")); + expect(JSON.parse(report!.slice("native_session_runner_loss_lineage ".length))).toMatchObject({ + verified: true, journals: 1, remaining: [], + }); + // The crash stop keeps the durable record for recovery to read, and + // queues no suspend that the next runnerd on this root would receive. + expect((await runnerState()).lifecycle).not.toBe("suspended"); + const control = JSON.parse(await readFile(join(stateDirectory, "control-plane", "control-plane-state.json"), "utf8")); + expect(control.commands.filter((command: { type: string; status: string }) => + command.type === "runner.suspend" && command.status === "pending")).toEqual([]); + } else { await expect(bundle.transport.close()).rejects.toThrow( "runner did not durably suspend before checkpoint", ); - const runnerState = JSON.parse( - await readFile( - join(stateDirectory, "runner", "runner-state.json"), - "utf8", - ), - ); - expect(runnerState.lifecycle).not.toBe("suspended"); + expect((await runnerState()).lifecycle).not.toBe("suspended"); + const report = diagnostics.find((message) => message.startsWith("native_session_runner_loss_lineage ")); + expect(JSON.parse(report!.slice("native_session_runner_loss_lineage ".length))).toMatchObject({ + verified: false, reason: "journal_unavailable", + }); const settlement = diagnostics.find((message) => message.startsWith("native_session_settlement_incomplete ")); - expect(settlement).toBeDefined(); expect(JSON.parse(settlement!.slice("native_session_settlement_incomplete ".length))).toMatchObject({ runnerSuspended: false, - suspensionState: { commandStatus: "pending", runnerIdentityMatches: true }, + suspensionState: { commandStatus: null }, }); - } finally { - await rm(stateDirectory, { recursive: true, force: true }); + // Nothing was signalled without proof. + expect(processAlive(jobPid)).toBe(true); } + } finally { + for (const pid of [jobPid, providerPid]) { + if (pid && processAlive(pid)) try { process.kill(pid, "SIGKILL"); } catch { /* gone */ } + } + await rm(stateDirectory, { recursive: true, force: true }); } }, 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index bddf9effa94..d52e8cca619 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -26,6 +26,7 @@ import { resolveCodexCommand } from "../drivers/codex/codex-command.js"; import { isCanonicalProviderEventType } from "../provider-events.js"; import { execFileSync } from "node:child_process"; import { readLinuxProcessStartedAt } from "./linux-process-start.js"; +import { reapExitedRunnerLineage } from "./runner-lineage-reap.js"; import { createHash, randomUUID } from "node:crypto"; import { appendFileSync, @@ -3905,6 +3906,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #resolveHandlerReady!: () => void; #core: DurablePrpControlPlane | null = null; #handle: RunnerProcessHandle | null = null; + /** The runnerd handle that exited on its own while the session was open. */ + #lostRunnerHandle: RunnerProcessHandle | null = null; #adoptedRunnerMonitor: NodeJS.Timeout | null = null; #adoptedRunnerAuthenticated = false; #pump: NodeJS.Timeout | null = null; @@ -5405,6 +5408,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { let providerDrained = false; let cleanupProviderStopConfirmed = false; let suspensionRequired = false; + // runnerd exited on its own, unsuspended, while the session was open: it + // was lost (killed or crashed), not stopped or broken by this close. + let runnerLost = false; let lastSuspensionState: Record | null = null; let suspensionState: { commandStatus: string | null; @@ -5453,7 +5459,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // their authority is safe to rotate; require the same exact suspension // barrier even when there is no remote checkpoint callback. suspensionRequired = true; - runnerSuspended = await awaitRunnerSuspensionBarrier({ + // A runnerd lost while the session was open can never answer. A queued + // suspend would stay pending in the saved control-plane state and reach + // the next runnerd on this root as a stale terminal command. + const lostLocalRunner = this.#handle !== null && this.#lostRunnerHandle === this.#handle + && !this.options.runnerProcessLauncher && !adoptedRunner; + runnerSuspended = !lostLocalRunner && await awaitRunnerSuspensionBarrier({ commands: () => this.#core?.store.state.commands ?? [], queueSuspend: (commandId) => { this.#core?.queueCommand("runner.suspend", {}, commandId, true); @@ -5468,6 +5479,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { pump: () => this.#pumpEventsSafely(), deadline: closeDeadline, }); + runnerLost = !runnerSuspended && this.#handle !== null && this.#lostRunnerHandle === this.#handle; if (!runnerSuspended) { const command = [...this.#core.store.state.commands].reverse().find( (candidate) => candidate.type === "runner.suspend", @@ -5591,7 +5603,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#controlPlaneCheckpoint = null; this.#controlPlaneRelease = null; } - if (suspensionRequired && !runnerSettled) { + // A lost local runnerd cannot suspend, and its provider and tools were + // re-parented when it died. When runnerd's own lineage journal proves + // every one of them stopped, nothing owns the state root any more: the + // session ends as a crash stop, and recovery decides from the durable + // record how to continue. Anything less keeps the quarantine below. + const runnerLossReaped = suspensionRequired && !runnerSettled && preparationConfirmed + && !runnerSuspended && runnerLost + && await this.#reapLostLocalRunnerLineage(closeDeadline); + if (suspensionRequired && !runnerSettled && !runnerLossReaped) { const settlement = { runnerSuspended, providerDrained, @@ -8259,8 +8279,32 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } } + /** True only when a runnerd this transport spawned on this host has exited + * and its journalled provider and tool processes are proven stopped. */ + async #reapLostLocalRunnerLineage(closeDeadline: number): Promise { + const handle = this.#handle; + const runnerPid = handle?.child.pid; + // A launcher-owned runnerd may run elsewhere (a remote sandbox), and an + // adopted runnerd is not this process's child, so its exit is not ours to + // confirm. Only a direct local spawn is reaped here. + if (!handle || runnerPid === undefined || this.options.runnerProcessLauncher + || this.options.adoptExistingRunner || !(await this.#runnerHasExited())) return false; + const report = await reapExitedRunnerLineage({ + runnerBinaryPath: this.options.runnerBinary + ?? (this.options.prepareLaunch ? "paperclip-runnerd" : defaultCapabilityRunnerdBinary()), + stateDirectory: this.options.runnerStateDirectory ?? resolve(this.#root, "runner"), + runnerPid, + // A small floor keeps the proof possible when close already used its budget. + timeoutMs: Math.max(5_000, closeDeadline - Date.now()), + }); + this.#diagnostic(`native_session_runner_loss_lineage ${JSON.stringify(report ?? { verified: false, reason: "report_unavailable" })}`); + return report?.verified === true; + } + async #runnerHasExited(): Promise { - if (this.#handle) return this.#handle.isExitConfirmed?.() ?? this.#handle.child.exitCode !== null; + // A runnerd killed by a signal reports exitCode null and its signalCode. + if (this.#handle) return this.#handle.isExitConfirmed?.() + ?? (this.#handle.child.exitCode !== null || (this.#handle.child.signalCode ?? null) !== null); const adoptedRunner = this.options.adoptExistingRunner; if (!adoptedRunner) return true; return !(await adoptedRunner.isAlive()); @@ -8347,6 +8391,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const detail = result.stderr.trim() || result.stdout.trim(); if (detail) this.#diagnostic(detail.slice(-4_096)); this.#publish(); + // runnerd exited on its own while the session was still open, without + // a clean suspend: it was lost (killed or crashed). Close may reap its + // journalled lineage. An exit during close is never treated as loss. + if (!this.#closed && exitConfirmed && !(result.code === 0 && + this.#core?.store.state.committedEvents.some((event) => event.eventType === "runner.suspending"))) + this.#lostRunnerHandle = handle; if (this.#closed || this.#failure !== null || this.#handle !== handle || (this.#releaseEpisode && !this.#releaseEpisode.claimed)) return; // A per-turn runner exits after its terminal suffix is durably ACKed.