From e4f67dde58bef6cc5350fabfc4cd5da64698f290 Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 16:45:01 +0000 Subject: [PATCH] feat: add guarded Rust experiment process runner --- .github/workflows/rust-test.yml | 2 + .github/workflows/style.yml | 2 +- Cargo.lock | 1 + crates/lance-context-wal/Cargo.toml | 7 + crates/lance-context-wal/src/bin/runner.rs | 403 +++++++++++++++++++++ crates/lance-context-wal/tests/runner.rs | 265 ++++++++++++++ docs/design/wal-experiment-runner.md | 45 +++ 7 files changed, 724 insertions(+), 1 deletion(-) create mode 100644 crates/lance-context-wal/src/bin/runner.rs create mode 100644 crates/lance-context-wal/tests/runner.rs create mode 100644 docs/design/wal-experiment-runner.md diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index f28c7cf..bd22d21 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -67,6 +67,8 @@ jobs: run: | cargo test -p lance-context-wal --features placement,admission,ipc,record,arrow-contract --lib cargo test -p lance-context-wal --features arrow-contract,serde_json/preserve_order --lib ingest_contract::arrow + - name: WAL experiment process lifecycle tests + run: cargo test -p lance-context-wal --features runner --test runner - name: Start etcd for HA scheduler test run: | ETCD_VERSION=3.7.0 diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index de408eb..e5c0a2c 100644 --- a/.github/workflows/style.yml +++ b/.github/workflows/style.yml @@ -56,7 +56,7 @@ jobs: - name: Clippy (lance-context, default features) run: cargo clippy --manifest-path crates/lance-context/Cargo.toml --all-targets -- -D warnings - name: Clippy (WAL experiments) - run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record,arrow-contract --all-targets -- -D warnings + run: cargo clippy -p lance-context-wal --features lab,placement,admission,ipc,record,arrow-contract,runner --all-targets -- -D warnings typos: name: Spell Check diff --git a/Cargo.lock b/Cargo.lock index 6013d87..c5219a2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4194,6 +4194,7 @@ dependencies = [ "bytes", "clap", "hdrhistogram", + "libc", "raft-engine", "raft-proto", "serde", diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 824b66e..0d332f7 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -7,6 +7,7 @@ publish = false description = "Independent Rust WAL experiments owned by lance-context" [features] +runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"] contract = ["dep:serde", "dep:sha2"] arrow-contract = ["contract", "dep:arrow-schema", "arrow-schema/serde", "dep:serde_json"] record = ["dep:bytes", "bytes/serde", "dep:serde", "dep:sha2", "dep:uuid"] @@ -24,6 +25,7 @@ arrow-schema = { version = "58", optional = true } bytes = { version = "1", optional = true } clap = { version = "4", features = ["derive"], optional = true } hdrhistogram = { version = "7", optional = true } +libc = { version = "0.2", optional = true } raft-engine = { version = "=0.4.2", default-features = false, optional = true } raft-proto = { version = "=0.7.0", optional = true } serde_json = { version = "1", optional = true } @@ -36,6 +38,11 @@ bincode = "1.3" arrow-ipc = { version = "58", features = ["zstd"] } tempfile = "3" +[[bin]] +name = "context-wal-runner" +path = "src/bin/runner.rs" +required-features = ["runner"] + [[bin]] name = "context-wal-lab" path = "src/bin/lab.rs" diff --git a/crates/lance-context-wal/src/bin/runner.rs b/crates/lance-context-wal/src/bin/runner.rs new file mode 100644 index 0000000..341b51c --- /dev/null +++ b/crates/lance-context-wal/src/bin/runner.rs @@ -0,0 +1,403 @@ +//! Linux experiment supervisor. Each operation directory authorizes at most one +//! launch. Observation failures never authorize another launch or cleanup. +//! This is not a service supervisor: unknown outcomes require reconciliation. +use anyhow::{bail, ensure, Context, Result}; +use clap::{Parser, Subcommand}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, File, OpenOptions}, + io::{Read, Write}, + os::{ + fd::{AsRawFd, FromRawFd, OwnedFd}, + unix::process::ExitStatusExt, + }, + path::{Path, PathBuf}, + process::{Command, Stdio}, + thread, + time::Duration, +}; + +#[derive(Parser)] +struct Args { + #[command(subcommand)] + command: Action, +} + +#[derive(Subcommand)] +enum Action { + /// Use a new directory for each intentional launch. An existing directory + /// is observed only; it is never reset, even if its owner has disappeared. + Start { + directory: PathBuf, + #[arg(long)] + executable: PathBuf, + #[arg(long)] + sha256: String, + #[arg(last = true)] + arguments: Vec, + }, + Observe { + directory: PathBuf, + }, + /// Explicit fault injection/cleanup of this recorded process incarnation. + Stop { + directory: PathBuf, + }, + #[command(hide = true)] + Supervise { + directory: PathBuf, + }, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +struct Request { + executable: PathBuf, + sha256: String, + arguments: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +struct Identity { + pid: u32, + start_ticks: u64, + boot: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct Artifact { + bytes: u64, + sha256: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct Completed { + code: Option, + signal: Option, + spawn_error: Option, + stdout: Artifact, + stderr: Artifact, +} + +#[derive(Serialize)] +struct Observation { + phase: &'static str, + supervisor: Option, + child: Option, + completed: Option, +} + +fn read Deserialize<'a>>(path: &Path) -> Result> { + match fs::read(path) { + Ok(body) => Ok(Some(serde_json::from_slice(&body)?)), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error.into()), + } +} + +fn persist(directory: &Path, name: &str, value: &T) -> Result<()> { + let temp = directory.join(format!(".{name}.{}", std::process::id())); + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp)?; + serde_json::to_writer(&mut file, value)?; + file.write_all(b"\n")?; + file.sync_all()?; + fs::rename(temp, directory.join(name))?; + File::open(directory)?.sync_all()?; + Ok(()) +} + +fn artifact(path: &Path) -> Result { + let mut file = File::open(path)?; + let mut digest = Sha256::new(); + let mut bytes = 0; + let mut buffer = [0u8; 64 * 1024]; + loop { + let count = file.read(&mut buffer)?; + if count == 0 { + break; + } + digest.update(&buffer[..count]); + bytes += count as u64; + } + Ok(Artifact { + bytes, + sha256: format!("{:x}", digest.finalize()), + }) +} + +fn process(pid: u32) -> Result> { + let stat = match fs::read_to_string(format!("/proc/{pid}/stat")) { + Ok(stat) => stat, + Err(error) + if error.kind() == std::io::ErrorKind::NotFound + || error.raw_os_error() == Some(libc::ESRCH) => + { + return Ok(None); + } + Err(error) => return Err(error.into()), + }; + // comm can contain spaces and ')'; field 3 starts after the final ') '. + let (_, rest) = stat.rsplit_once(") ").context("invalid process stat")?; + let fields: Vec<_> = rest.split_whitespace().collect(); + let start_ticks = fields + .get(19) + .context("missing process start time")? + .parse()?; + let state = fields + .first() + .and_then(|v| v.chars().next()) + .context("missing process state")?; + Ok(Some(( + Identity { + pid, + start_ticks, + boot: fs::read_to_string("/proc/sys/kernel/random/boot_id")? + .trim() + .into(), + }, + state, + ))) +} + +fn live(identity: &Identity) -> Result { + Ok(process(identity.pid)?.is_some_and(|(current, state)| { + current.start_ticks == identity.start_ticks + && current.boot == identity.boot + && state != 'Z' + && state != 'X' + })) +} + +fn observe(directory: &Path) -> Result { + ensure!(directory.is_dir(), "operation directory does not exist"); + let supervisor: Option = read(&directory.join("supervisor.json"))?; + let child: Option = read(&directory.join("child.json"))?; + let completed: Option = read(&directory.join("completed.json"))?; + let phase = if completed.is_some() { + // The synced journal is authoritative after completion. Re-probing + // /proc can race reaping (ESRCH), or inspect a reused PID needlessly. + "completed" + } else { + let supervisor_live = supervisor.as_ref().map(live).transpose()?.unwrap_or(false); + let child_live = child.as_ref().map(live).transpose()?.unwrap_or(false); + if child_live && !supervisor_live { + "orphaned_running" + } else if supervisor_live { + "running" + } else { + "unknown" + } + }; + Ok(Observation { + phase, + supervisor, + child, + completed, + }) +} + +fn start(directory: &Path, request: Request) -> Result<()> { + match fs::create_dir(directory) { + Ok(()) => { + File::open(directory.parent().context("operation needs a parent")?)?.sync_all()?; + persist(directory, "request.json", &request)?; + ensure!( + artifact(&request.executable)?.sha256 == request.sha256, + "executable checksum mismatch; operation will not be retried" + ); + // Explicitly detached stdio: losing the launcher/exec connection + // cannot close the workload's output streams. + let _supervisor = Command::new(std::env::current_exe()?) + .arg("supervise") + .arg(directory) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr( + OpenOptions::new() + .write(true) + .create_new(true) + .open(directory.join("supervisor.stderr"))?, + ) + .spawn()?; + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + if let Some(saved) = read::(&directory.join("request.json"))? { + ensure!( + saved == request, + "operation ID already belongs to a different request" + ); + } + // Missing metadata can mean the first launcher died before writing. + // That ambiguity is never permission to launch another command. + } + Err(error) => return Err(error.into()), + } + print_observation(directory) +} + +fn supervise(directory: &Path) -> Result<()> { + // SAFETY: process-local session/signal setup, before creating any threads. + unsafe { + ensure!(libc::setsid() != -1, "could not detach supervisor session"); + ensure!( + libc::signal(libc::SIGHUP, libc::SIG_IGN) != libc::SIG_ERR, + "could not ignore hangup" + ); + } + let lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(directory.join("lock"))?; + lock.try_lock() + .context("operation already has a supervisor")?; + ensure!( + !directory.join("supervisor.json").exists(), + "operation was already started; refusing replay" + ); + let identity = process(std::process::id())? + .context("supervisor identity unavailable")? + .0; + persist(directory, "supervisor.json", &identity)?; + let request: Request = read(&directory.join("request.json"))?.context("missing request")?; + ensure!( + artifact(&request.executable)?.sha256 == request.sha256, + "executable changed before launch" + ); + let stdout = OpenOptions::new() + .write(true) + .create_new(true) + .open(directory.join("stdout"))?; + let stderr = OpenOptions::new() + .write(true) + .create_new(true) + .open(directory.join("stderr"))?; + let result = if directory.join("stop.json").exists() { + Err(std::io::Error::other( + "operation stopped before child launch", + )) + } else { + Command::new(&request.executable) + .args(&request.arguments) + .stdin(Stdio::null()) + .stdout(stdout.try_clone()?) + .stderr(stderr.try_clone()?) + .spawn() + }; + let (code, signal, spawn_error) = match result { + Err(error) => (None, None, Some(error.to_string())), + Ok(mut child) => { + persist(directory, "pid", &child.id())?; + if let Some((identity, _)) = process(child.id())? { + persist(directory, "child.json", &identity)?; + } + let status = loop { + if let Some(status) = child.try_wait()? { + break status; + } + if directory.join("stop.json").exists() { + child.kill()?; + break child.wait()?; + } + thread::sleep(Duration::from_millis(25)); + }; + (status.code(), status.signal(), None) + } + }; + stdout.sync_all()?; + stderr.sync_all()?; + persist( + directory, + "completed.json", + &Completed { + code, + signal, + spawn_error, + stdout: artifact(&directory.join("stdout"))?, + stderr: artifact(&directory.join("stderr"))?, + }, + ) +} + +fn stop(directory: &Path) -> Result<()> { + // The marker fences a supervisor that has not spawned its child yet. + persist(directory, "stop.json", &true)?; + if let Some(identity) = read::(&directory.join("child.json"))? { + if live(&identity)? { + // Bind the signal to a kernel process handle, then check the saved + // incarnation again. PID reuse cannot signal an unrelated process. + // SAFETY: pidfd_open has scalar arguments and returns an owned fd. + let raw = unsafe { libc::syscall(libc::SYS_pidfd_open, identity.pid, 0) }; + if raw < 0 { + let error = std::io::Error::last_os_error(); + if error.raw_os_error() != Some(libc::ESRCH) { + return Err(error.into()); + } + } else { + // SAFETY: successful pidfd_open returned a new owned descriptor. + let fd = unsafe { OwnedFd::from_raw_fd(raw as i32) }; + if live(&identity)? { + // SAFETY: pidfd references only this process; no siginfo pointer. + let result = unsafe { + libc::syscall( + libc::SYS_pidfd_send_signal, + fd.as_raw_fd(), + libc::SIGKILL, + std::ptr::null::(), + 0, + ) + }; + if result < 0 { + let error = std::io::Error::last_os_error(); + if error.raw_os_error() != Some(libc::ESRCH) { + return Err(error.into()); + } + } + } + } + } + } + print_observation(directory) +} + +fn print_observation(directory: &Path) -> Result<()> { + println!("{}", serde_json::to_string(&observe(directory)?)?); + Ok(()) +} + +fn main() -> Result<()> { + match Args::parse().command { + Action::Start { + directory, + executable, + sha256, + arguments, + } => { + ensure!( + sha256.len() == 64 && sha256.bytes().all(|v| v.is_ascii_hexdigit()), + "invalid SHA-256" + ); + let parent = directory + .parent() + .context("operation needs parent")? + .canonicalize()?; + let name = directory.file_name().context("operation needs name")?; + start( + &parent.join(name), + Request { + executable: executable.canonicalize()?, + sha256: sha256.to_ascii_lowercase(), + arguments, + }, + ) + } + Action::Observe { directory } => print_observation(&directory), + Action::Stop { directory } => stop(&directory), + Action::Supervise { directory } => supervise(&directory), + } + .or_else(|error| bail!("{error:#}")) +} diff --git a/crates/lance-context-wal/tests/runner.rs b/crates/lance-context-wal/tests/runner.rs new file mode 100644 index 0000000..cf80131 --- /dev/null +++ b/crates/lance-context-wal/tests/runner.rs @@ -0,0 +1,265 @@ +#![cfg(feature = "runner")] + +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::{ + fs::{self, OpenOptions}, + io::Write, + path::{Path, PathBuf}, + process::{Command, Output}, + thread, + time::{Duration, Instant}, +}; + +const RUNNER: &str = env!("CARGO_BIN_EXE_context-wal-runner"); + +struct Run { + root: tempfile::TempDir, + operation: PathBuf, + executable: PathBuf, + sha: String, +} + +impl Run { + fn new() -> Self { + let root = tempfile::tempdir().unwrap(); + let executable = std::env::current_exe().unwrap(); + let sha = format!("{:x}", Sha256::digest(fs::read(&executable).unwrap())); + Self { + operation: root.path().join("operation"), + root, + executable, + sha, + } + } + + fn launch(&self, mode: &str) -> Command { + let mut command = Command::new(RUNNER); + command + .arg("start") + .arg(&self.operation) + .arg("--executable") + .arg(&self.executable) + .arg("--sha256") + .arg(&self.sha) + .arg("--") + .args(["--exact", "runner_fixture", "--ignored", "--nocapture"]) + .env("CONTEXT_RUNNER_FIXTURE", self.root.path()) + .env("CONTEXT_RUNNER_FIXTURE_MODE", mode); + command + } + + fn observe(&self) -> Value { + json( + Command::new(RUNNER) + .arg("observe") + .arg(&self.operation) + .output() + .unwrap(), + ) + } + + fn wait(&self, predicate: impl Fn(&Value) -> bool) -> Value { + let deadline = Instant::now() + Duration::from_secs(15); + loop { + let status = self.observe(); + if predicate(&status) { + return status; + } + assert!( + Instant::now() < deadline, + "operation did not reach expected state: {status}" + ); + thread::sleep(Duration::from_millis(10)); + } + } +} + +impl Drop for Run { + fn drop(&mut self) { + let _ = Command::new(RUNNER) + .arg("stop") + .arg(&self.operation) + .output(); + // Give the detached supervisor time to reap its child before TempDir. + for _ in 0..100 { + if !self.operation.exists() { + break; + } + let status = self.observe(); + if status["phase"] != "running" && status["phase"] != "orphaned_running" { + break; + } + thread::sleep(Duration::from_millis(10)); + } + } +} + +fn json(output: Output) -> Value { + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).unwrap() +} + +fn assert_artifact(path: &Path, saved: &Value) { + let bytes = fs::read(path).unwrap(); + assert_eq!(saved["bytes"].as_u64(), Some(bytes.len() as u64)); + assert_eq!(saved["sha256"], format!("{:x}", Sha256::digest(&bytes))); +} + +#[test] +fn concurrent_launch_and_lost_observation_never_repeat_the_rust_command() { + let run = Run::new(); + thread::scope(|scope| { + let workers: Vec<_> = (0..8) + .map(|_| scope.spawn(|| json(run.launch("finish").output().unwrap()))) + .collect(); + for worker in workers { + worker.join().unwrap(); + } + }); + let completed = run.wait(|value| value["phase"] == "completed"); + assert_eq!(completed["completed"]["code"], 0); + assert_artifact( + &run.operation.join("stdout"), + &completed["completed"]["stdout"], + ); + assert_artifact( + &run.operation.join("stderr"), + &completed["completed"]["stderr"], + ); + for _ in 0..3 { + assert_eq!(json(run.launch("finish").output().unwrap()), completed); + } + assert_eq!(fs::read(run.root.path().join("launches")).unwrap(), b"x"); + let changed = run.launch("finish").arg("--show-output").output().unwrap(); + assert!(!changed.status.success()); + assert!(String::from_utf8_lossy(&changed.stderr).contains("different request")); + let replay = Command::new(RUNNER) + .arg("supervise") + .arg(&run.operation) + .output() + .unwrap(); + assert!(!replay.status.success()); + assert_eq!(run.observe(), completed); +} + +#[test] +fn supervisor_loss_preserves_unknown_outcome_and_stop_targets_the_orphan() { + let run = Run::new(); + json(run.launch("wait").output().unwrap()); + let active = + run.wait(|value| value["child"].is_object() && run.root.path().join("launches").exists()); + let supervisor = active["supervisor"]["pid"].as_i64().unwrap(); + // SAFETY: the target is this test's live supervisor, not a borrowed PID. + assert_eq!(unsafe { libc::kill(supervisor as i32, libc::SIGKILL) }, 0); + run.wait(|value| value["phase"] == "orphaned_running"); + assert_eq!( + json(run.launch("wait").output().unwrap())["phase"], + "orphaned_running" + ); + assert_eq!(fs::read(run.root.path().join("launches")).unwrap(), b"x"); + + // A mismatched process incarnation must not be signaled. Keep the original + // child live, then restore its record and stop that exact orphan via pidfd. + let original = fs::read(run.operation.join("child.json")).unwrap(); + let mut foreign: Value = serde_json::from_slice(&original).unwrap(); + foreign["start_ticks"] = Value::from(foreign["start_ticks"].as_u64().unwrap() + 1); + fs::write( + run.operation.join("child.json"), + serde_json::to_vec(&foreign).unwrap(), + ) + .unwrap(); + json( + Command::new(RUNNER) + .arg("stop") + .arg(&run.operation) + .output() + .unwrap(), + ); + fs::write(run.operation.join("child.json"), &original).unwrap(); + assert_eq!(run.observe()["phase"], "orphaned_running"); + json( + Command::new(RUNNER) + .arg("stop") + .arg(&run.operation) + .output() + .unwrap(), + ); + let unknown = run.wait(|value| value["phase"] == "unknown"); + assert!(unknown["completed"].is_null()); + assert_eq!( + json(run.launch("wait").output().unwrap())["phase"], + "unknown" + ); + assert_eq!(fs::read(run.root.path().join("launches")).unwrap(), b"x"); +} + +#[test] +fn claimed_directory_and_checksum_failure_are_never_relaunched() { + let run = Run::new(); + fs::create_dir(&run.operation).unwrap(); + assert_eq!( + json(run.launch("finish").output().unwrap())["phase"], + "unknown" + ); + assert!(!run.root.path().join("launches").exists()); + let mut corrupt = Run::new(); + corrupt.sha = "0".repeat(64); + let result = corrupt.launch("finish").output().unwrap(); + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("checksum mismatch")); + assert_eq!( + json(corrupt.launch("finish").output().unwrap())["phase"], + "unknown" + ); + assert!(!corrupt.root.path().join("launches").exists()); +} + +#[test] +fn hangup_does_not_stop_the_workload_and_explicit_stop_retains_its_exit() { + let run = Run::new(); + json(run.launch("wait").output().unwrap()); + let active = + run.wait(|value| value["child"].is_object() && run.root.path().join("launches").exists()); + let supervisor = active["supervisor"]["pid"].as_i64().unwrap(); + // SAFETY: signal only this test's live supervisor. + assert_eq!(unsafe { libc::kill(supervisor as i32, libc::SIGHUP) }, 0); + thread::sleep(Duration::from_millis(50)); + assert_eq!(run.observe()["phase"], "running"); + json( + Command::new(RUNNER) + .arg("stop") + .arg(&run.operation) + .output() + .unwrap(), + ); + let completed = run.wait(|value| value["phase"] == "completed"); + assert_eq!(completed["completed"]["signal"], libc::SIGKILL); + assert!(completed["completed"]["code"].is_null()); + assert_eq!(json(run.launch("wait").output().unwrap()), completed); + assert_eq!(fs::read(run.root.path().join("launches")).unwrap(), b"x"); +} + +#[test] +#[ignore = "separate Rust child process launched only by runner integration tests"] +fn runner_fixture() { + let root = PathBuf::from(std::env::var_os("CONTEXT_RUNNER_FIXTURE").unwrap()); + let mut marker = OpenOptions::new() + .create(true) + .append(true) + .open(root.join("launches")) + .unwrap(); + marker.write_all(b"x").unwrap(); + marker.sync_all().unwrap(); + println!("complete stdout payload"); + eprintln!("complete stderr payload"); + if std::env::var("CONTEXT_RUNNER_FIXTURE_MODE").unwrap() == "wait" { + loop { + thread::sleep(Duration::from_secs(1)); + } + } +} diff --git a/docs/design/wal-experiment-runner.md b/docs/design/wal-experiment-runner.md new file mode 100644 index 0000000..f0954ee --- /dev/null +++ b/docs/design/wal-experiment-runner.md @@ -0,0 +1,45 @@ +# Guarded experiment process runner + +Tracking: https://github.com/lance-format/lance-context/issues/378 + +A disconnected terminal or timed-out remote command can leave an ingestion +experiment running. Starting another attempt changes the workload and can hide +the original failure. The optional Linux `context-wal-runner` records one launch +per operation directory and keeps observation separate from execution. + +```sh +cargo build --release -p lance-context-wal --features runner --bin context-wal-runner +context-wal-runner start /tmp/example-operation \ + --executable /absolute/path/to/program --sha256 -- +context-wal-runner observe /tmp/example-operation +context-wal-runner stop /tmp/example-operation +``` + +Use an existing parent directory and a fresh absolute operation path. The +create-only directory, synced request and locked supervisor journal prevent +concurrent launch or replay. Existing operations are observed; a different +executable/hash/argument request is rejected. Environment variables and working +directory are inherited from the first launcher, not part of the stored request; +callers must freeze and separately record their nonsecret configuration. + +The supervisor detaches its streams and session, ignores hangup, checks the +executable hash again, records the child's boot/PID/start-time identity and waits +for completion. It syncs stdout/stderr and records their sizes/hashes plus exit +code, signal or spawn error. Callers still verify retrieved logs against those +hashes before treating a remote download as complete. + +An `unknown` or `orphaned_running` result is not permission to restart. A crash +between process creation and identity recording can leave an unidentifiable +child; the tool preserves that uncertainty. `stop` explicitly marks the operation +and uses a Linux pidfd plus incarnation checks to signal a recorded child without +signaling a reused PID. It does not stop an entire descendant process tree. +Completed journals remain authoritative after process reaping. + +This is a trusted, isolated experiment tool, not a service supervisor or an +exactly-once business operation mechanism. Do not remove/reuse operation +directories to recover an uncertain result. Keep them on durable storage if +their evidence must survive container/node loss. The executable and evidence +directory must not be modified by other actors; the hash is a reproducibility +check, not protection against executable replacement races or malicious input. +Arguments and logs can contain deployment information, so keep raw journals and +configuration private. The public tests use synthetic child processes only.