From aa941fb5fa8365b90052cf6c53108acdbed99087 Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Mon, 10 Aug 2026 16:59:43 -0300 Subject: [PATCH 1/5] build(deps): enable process spawn and process-group kill dependencies This commit adds the crate features the process infrastructure needs to spawn long-running jobs, wait asynchronously, and kill a process group. Tokio gains process and time support, async-trait makes a running-process handle object-safe, and nix provides a safe killpg wrapper. This commit is part of the kw integration's step 1. Signed-off-by: lorenzoberts --- Cargo.lock | 124 +++++++++++++++++++++++++++++++++++++++-------------- Cargo.toml | 4 +- 2 files changed, 96 insertions(+), 32 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 38f9236a..76265be0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -110,6 +110,17 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "autocfg" version = "1.5.0" @@ -185,6 +196,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2fd1289c04a9ea8cb22300a459a72a385d7c73d3259e2ed7dcb2af674838cfa9" +[[package]] +name = "cfg_aliases" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" + [[package]] name = "chrono" version = "0.4.41" @@ -196,7 +213,7 @@ dependencies = [ "js-sys", "num-traits", "wasm-bindgen", - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -230,7 +247,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -348,7 +365,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a2785755761f3ddc1492979ce1e48d2c00d09311c39e4466429188f3dd6501" dependencies = [ "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -372,7 +389,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.106", ] [[package]] @@ -383,7 +400,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -403,7 +420,7 @@ checksum = "74ef43543e701c01ad77d3a5922755c6a1d71b22d942cb8042be4994b380caff" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -581,7 +598,7 @@ dependencies = [ "indoc", "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -623,9 +640,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "libc" -version = "0.2.175" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a82ae493e598baaea5209805c49bbf2ea7de956d50d7da0da1164f9c6d28543" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "linux-raw-sys" @@ -687,14 +704,14 @@ dependencies = [ [[package]] name = "mio" -version = "1.0.4" +version = "1.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78bed444cc8a2160f01cbcf811ef18cac863ad68ae8ca62092e8db51d51c761c" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "log", "wasi", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -720,7 +737,19 @@ dependencies = [ "cfg-if", "proc-macro2", "quote", - "syn", + "syn 2.0.106", +] + +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", ] [[package]] @@ -825,6 +854,7 @@ name = "patch-hub" version = "0.1.7" dependencies = [ "ansi-to-tui", + "async-trait", "chrono", "clap", "color-eyre", @@ -832,6 +862,7 @@ dependencies = [ "derive-getters", "lazy_static", "mockall", + "nix", "once_cell", "patch-hub-proc-macros 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "ratatui", @@ -858,7 +889,7 @@ dependencies = [ "quote", "serde", "serde_json", - "syn", + "syn 2.0.106", ] [[package]] @@ -873,7 +904,7 @@ dependencies = [ "quote", "serde", "serde_json", - "syn", + "syn 2.0.106", ] [[package]] @@ -1134,7 +1165,7 @@ checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1176,9 +1207,9 @@ dependencies = [ [[package]] name = "signal-hook-mio" -version = "0.2.4" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "34db1a06d485c9142248b7a054f034b349b212551f3dfd19c94d45a754a217cd" +checksum = "b75a19a7a740b25bc7944bdee6172368f988763b744e3d4dfe753f6b4ece40cc" dependencies = [ "libc", "mio", @@ -1237,7 +1268,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn", + "syn 2.0.106", ] [[package]] @@ -1257,6 +1288,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "termtree" version = "0.5.1" @@ -1289,7 +1331,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1300,7 +1342,7 @@ checksum = "6c5e1be1c48b9172ee610da68fd9cd2770e7a4056cb3fc98710ee6906f0c7960" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1349,8 +1391,13 @@ version = "1.52.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" dependencies = [ + "bytes", + "libc", + "mio", "pin-project-lite", + "signal-hook-registry", "tokio-macros", + "windows-sys 0.61.2", ] [[package]] @@ -1361,7 +1408,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1395,7 +1442,7 @@ checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1573,7 +1620,7 @@ dependencies = [ "log", "proc-macro2", "quote", - "syn", + "syn 2.0.106", "wasm-bindgen-shared", ] @@ -1595,7 +1642,7 @@ checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -1659,7 +1706,7 @@ checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" dependencies = [ "windows-implement", "windows-interface", - "windows-link", + "windows-link 0.1.3", "windows-result", "windows-strings", ] @@ -1672,7 +1719,7 @@ checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1683,7 +1730,7 @@ checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.106", ] [[package]] @@ -1692,13 +1739,19 @@ version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + [[package]] name = "windows-result" version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" dependencies = [ - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -1707,7 +1760,7 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" dependencies = [ - "windows-link", + "windows-link 0.1.3", ] [[package]] @@ -1737,6 +1790,15 @@ dependencies = [ "windows-targets 0.53.3", ] +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -1759,7 +1821,7 @@ version = "0.53.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d5fe6031c4041849d7c496a8ded650796e7b6ecc19df1a431c1a363342e5dc91" dependencies = [ - "windows-link", + "windows-link 0.1.3", "windows_aarch64_gnullvm 0.53.0", "windows_aarch64_msvc 0.53.0", "windows_i686_gnu 0.53.0", diff --git a/Cargo.toml b/Cargo.toml index b6c51b4d..b27eb8e8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,7 +27,9 @@ chrono = "0.4.41" ansi-to-tui = "7.0.0" which = "8.0.0" ureq = { version = "3.0.12", features = ["rustls"] } -tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync"] } +tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "process", "time"] } +async-trait = "0.1" +nix = { version = "0.31", features = ["signal"] } [dev-dependencies] ctor = "0.2" From b36549d1f83be05723fbe559c844e37b3997f67b Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Tue, 11 Aug 2026 09:25:34 -0300 Subject: [PATCH 2/5] feat(infrastructure): add async process spawn, wait, and process-group kill This commit introduces ProcessTrait as a separate infrastructure boundary from ShellTrait. Spawn returns immediately with a RunningProcess handle, streams stdout and stderr into a log file, and supports async wait plus process-group kill so long-running kw jobs cannot freeze the TUI or leak grandchild processes. This commit is part of the kw integration's step 1. Signed-off-by: lorenzoberts --- src/infrastructure/mod.rs | 1 + src/infrastructure/process/mod.rs | 99 +++++++++++++ src/infrastructure/process/tests.rs | 221 ++++++++++++++++++++++++++++ src/infrastructure/process/trait.rs | 49 ++++++ 4 files changed, 370 insertions(+) create mode 100644 src/infrastructure/process/mod.rs create mode 100644 src/infrastructure/process/tests.rs create mode 100644 src/infrastructure/process/trait.rs diff --git a/src/infrastructure/mod.rs b/src/infrastructure/mod.rs index dd6b5be0..d2e604c9 100644 --- a/src/infrastructure/mod.rs +++ b/src/infrastructure/mod.rs @@ -3,5 +3,6 @@ pub mod errors; pub mod file_system; pub mod monitoring; pub mod net; +pub mod process; pub mod shell; pub mod terminal; diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs new file mode 100644 index 00000000..31a51d30 --- /dev/null +++ b/src/infrastructure/process/mod.rs @@ -0,0 +1,99 @@ +//! Unix-only process spawning: kw itself is Linux-only, and process-group +//! semantics (`process_group`/`killpg`) have no portable equivalent. + +mod r#trait; + +pub use r#trait::{ProcessError, ProcessTrait, RunningProcess}; + +#[cfg(test)] +pub use r#trait::MockRunningProcess; + +#[cfg(test)] +mod tests; + +use std::{ + fs::File, + io, + path::Path, + process::{ExitStatus, Stdio}, +}; + +use async_trait::async_trait; +use nix::{ + errno::Errno, + sys::signal::{killpg, Signal}, + unistd::Pid, +}; +use tokio::process::{Child, Command}; + +use super::shell::ShellCommand; + +// No production caller exists until the kw integration wires KwActor; +// kept per the CachePolicy precedent (src/lore/application/cache.rs). +#[allow(dead_code)] +pub struct OsProcess; + +impl ProcessTrait for OsProcess { + fn spawn( + &self, + cmd: &ShellCommand, + cwd: &Path, + log_path: &Path, + ) -> Result, ProcessError> { + let log_out = File::create(log_path)?; + let log_err = log_out.try_clone()?; + + // stdin is null so a prompt from the child gets EOF and fails fast + // instead of hanging on a TTY owned by the TUI. + let child = Command::new(&cmd.program) + .args(&cmd.args) + .current_dir(cwd) + .stdin(Stdio::null()) + .stdout(Stdio::from(log_out)) + .stderr(Stdio::from(log_err)) + .process_group(0) + .spawn()?; + + let pid = child + .id() + .ok_or_else(|| ProcessError::IoError(io::Error::other("spawned child has no pid")))?; + + Ok(Box::new(OsRunningProcess { + child, + pgid: Pid::from_raw(pid as i32), + reaped: false, + })) + } +} + +struct OsRunningProcess { + child: Child, + pgid: Pid, + reaped: bool, +} + +#[async_trait] +impl RunningProcess for OsRunningProcess { + async fn wait(&mut self) -> Result { + let status = self.child.wait().await?; + self.reaped = true; + Ok(status) + } + + fn kill(&mut self) -> Result<(), ProcessError> { + match killpg(self.pgid, Signal::SIGTERM) { + Ok(()) | Err(Errno::ESRCH) => Ok(()), + Err(errno) => Err(ProcessError::IoError(io::Error::from(errno))), + } + } +} + +impl Drop for OsRunningProcess { + fn drop(&mut self) { + // A handle dropped without wait() (panic, aborted task) must not leave + // the job's process group running. + if !self.reaped { + let _ = killpg(self.pgid, Signal::SIGTERM); + } + } +} diff --git a/src/infrastructure/process/tests.rs b/src/infrastructure/process/tests.rs new file mode 100644 index 00000000..22ecfe6a --- /dev/null +++ b/src/infrastructure/process/tests.rs @@ -0,0 +1,221 @@ +use std::{ + os::unix::process::ExitStatusExt, + path::{Path, PathBuf}, + process::ExitStatus, + time::{Duration, Instant}, +}; + +use nix::{errno::Errno, sys::signal::kill, unistd::Pid}; + +use super::{MockRunningProcess, OsProcess, ProcessTrait, RunningProcess}; +use crate::infrastructure::shell::ShellCommand; + +struct TempDir(PathBuf); + +impl TempDir { + fn new(test_name: &str) -> Self { + let dir = std::env::temp_dir().join(format!( + "patch_hub_process_test_{}_{test_name}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Self(dir) + } + + fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +async fn wait_for(mut cond: impl FnMut() -> bool, timeout: Duration) -> bool { + let start = Instant::now(); + while !cond() { + if start.elapsed() > timeout { + return false; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + true +} + +fn pid_is_gone(pid: i32) -> bool { + matches!( + kill(Pid::from_raw(pid), None::), + Err(Errno::ESRCH) + ) +} + +#[tokio::test] +async fn spawn_returns_while_process_still_running() { + let dir = TempDir::new("spawn_returns"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("sh").args(["-c", "sleep 0.5; echo done"]); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + + // spawn returned while the child was still inside its sleep + let early_log = std::fs::read_to_string(&log).unwrap(); + assert!(!early_log.contains("done")); + + let status = process.wait().await.unwrap(); + assert!(status.success()); + let final_log = std::fs::read_to_string(&log).unwrap(); + assert!(final_log.contains("done")); +} + +#[tokio::test] +async fn wait_returns_exit_code() { + let dir = TempDir::new("exit_code"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("sh").args(["-c", "exit 42"]); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + let status = process.wait().await.unwrap(); + + assert!(!status.success()); + assert_eq!(status.code(), Some(42)); +} + +#[tokio::test] +async fn log_file_grows_incrementally_with_stdout_and_stderr() { + let dir = TempDir::new("incremental_log"); + let log = dir.path().join("job.log"); + let cmd = + ShellCommand::new("sh").args(["-c", "echo first; echo errline >&2; sleep 1; echo second"]); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + + // output reaches the file while the process is still running, not at exit + let saw_partial_log = wait_for( + || { + let contents = std::fs::read_to_string(&log).unwrap(); + contents.contains("first") + && contents.contains("errline") + && !contents.contains("second") + }, + Duration::from_millis(800), + ) + .await; + assert!(saw_partial_log); + + let status = process.wait().await.unwrap(); + assert!(status.success()); + let contents = std::fs::read_to_string(&log).unwrap(); + assert!(contents.contains("first")); + assert!(contents.contains("errline")); + assert!(contents.contains("second")); +} + +#[tokio::test] +async fn spawn_runs_in_given_cwd() { + let dir = TempDir::new("cwd"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("pwd"); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + let status = process.wait().await.unwrap(); + + assert!(status.success()); + let out = std::fs::read_to_string(&log).unwrap(); + let expected = dir.path().canonicalize().unwrap(); + assert_eq!(out.trim(), expected.to_string_lossy()); +} + +#[tokio::test] +async fn kill_terminates_process_group() { + let dir = TempDir::new("kill_group"); + let log = dir.path().join("job.log"); + let sidecar = dir.path().join("grandchild.pid"); + let script = format!("sleep 60 & echo $! > \"{}\"; wait", sidecar.display()); + let cmd = ShellCommand::new("sh").args(["-c", &script]); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + + let sidecar_ready = wait_for(|| sidecar.exists(), Duration::from_secs(2)).await; + assert!(sidecar_ready); + let grandchild_pid: i32 = std::fs::read_to_string(&sidecar) + .unwrap() + .trim() + .parse() + .unwrap(); + + process.kill().unwrap(); + let status = tokio::time::timeout(Duration::from_secs(2), process.wait()) + .await + .expect("wait must complete shortly after kill") + .unwrap(); + // Signal death vs. exit code 128+SIGTERM is shell-dependent; what matters + // is the job did not succeed. + assert!(!status.success()); + + let grandchild_gone = wait_for(|| pid_is_gone(grandchild_pid), Duration::from_secs(2)).await; + assert!(grandchild_gone); +} + +#[tokio::test] +async fn kill_after_successful_exit_is_ok() { + let dir = TempDir::new("kill_idempotent"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("sh").args(["-c", "exit 0"]); + + let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + process.wait().await.unwrap(); + + process.kill().unwrap(); + process.kill().unwrap(); +} + +#[tokio::test] +async fn spawn_missing_binary_returns_error() { + let dir = TempDir::new("missing_binary"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("__nonexistent_binary_patch_hub__"); + + let result = OsProcess.spawn(&cmd, dir.path(), &log); + + assert!(result.is_err()); +} + +#[tokio::test] +async fn dropped_unreaped_process_group_is_killed() { + let dir = TempDir::new("drop_kills"); + let log = dir.path().join("job.log"); + let sidecar = dir.path().join("grandchild.pid"); + let script = format!("sleep 60 & echo $! > \"{}\"; wait", sidecar.display()); + let cmd = ShellCommand::new("sh").args(["-c", &script]); + + let process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); + + let sidecar_ready = wait_for(|| sidecar.exists(), Duration::from_secs(2)).await; + assert!(sidecar_ready); + let grandchild_pid: i32 = std::fs::read_to_string(&sidecar) + .unwrap() + .trim() + .parse() + .unwrap(); + + // no wait(): dropping the handle must not orphan the process group + drop(process); + + let grandchild_gone = wait_for(|| pid_is_gone(grandchild_pid), Duration::from_secs(2)).await; + assert!(grandchild_gone); +} + +#[tokio::test] +async fn running_process_is_dyn_compatible_and_mockable() { + let mut mock = MockRunningProcess::new(); + mock.expect_wait().returning(|| Ok(ExitStatus::from_raw(0))); + mock.expect_kill().returning(|| Ok(())); + + let mut process: Box = Box::new(mock); + process.kill().unwrap(); + let status = process.wait().await.unwrap(); + assert!(status.success()); +} diff --git a/src/infrastructure/process/trait.rs b/src/infrastructure/process/trait.rs new file mode 100644 index 00000000..7a97cd59 --- /dev/null +++ b/src/infrastructure/process/trait.rs @@ -0,0 +1,49 @@ +use async_trait::async_trait; +use mockall::automock; +use thiserror::Error; + +use std::{io, path::Path, process::ExitStatus}; + +use crate::infrastructure::shell::ShellCommand; + +#[derive(Debug, Error)] +pub enum ProcessError { + #[error("{0}")] + IoError(#[from] io::Error), +} + +// The two traits in this module have no production caller until the kw +// integration wires KwActor; kept per the CachePolicy precedent +// (src/lore/application/cache.rs). +#[allow(dead_code)] +#[automock] +pub trait ProcessTrait: Send + Sync { + /// Spawn `cmd` with `cwd` as its working directory, redirecting stdout and + /// stderr to `log_path` (created/truncated; the parent directory must + /// already exist). Returns immediately with a handle to the still-running + /// process, which is spawned as a process-group leader. + fn spawn( + &self, + cmd: &ShellCommand, + cwd: &Path, + log_path: &Path, + ) -> Result, ProcessError>; +} + +// `automock` must stay the outermost attribute: with `async_trait` listed +// first, the generated mock's async methods return an unusable type. +#[allow(dead_code)] +#[automock] +#[async_trait] +pub trait RunningProcess: Send { + /// Resolve once the process exits, without blocking a worker thread. + async fn wait(&mut self) -> Result; + + /// Send SIGTERM to the whole process group, not just the direct child. + /// Idempotent: an already-gone group (ESRCH) is reported as success. + /// + /// Both methods take `&mut self`: a consumer that waits while staying able + /// to cancel should `tokio::select!` between `wait()` and its cancel + /// signal, then call `kill()`. + fn kill(&mut self) -> Result<(), ProcessError>; +} From 129e1cf16b0462a3adceed342144a7c7c005c4a6 Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Tue, 11 Aug 2026 16:54:08 -0300 Subject: [PATCH 3/5] test(infrastructure): add controllable fake process double for actor tests This commit adds a test-only ProcessTrait double whose lifecycle a caller can hold open, finish, or kill. Each spawn is recorded so tests can assert argv, cwd, and log path without talking to a real process. This commit is part of the kw integration's step 1. Signed-off-by: lorenzoberts --- src/infrastructure/process/fake.rs | 162 ++++++++++++++++++++++++++++ src/infrastructure/process/mod.rs | 5 + src/infrastructure/process/tests.rs | 64 ++++++++++- 3 files changed, 230 insertions(+), 1 deletion(-) create mode 100644 src/infrastructure/process/fake.rs diff --git a/src/infrastructure/process/fake.rs b/src/infrastructure/process/fake.rs new file mode 100644 index 00000000..a87b6cb2 --- /dev/null +++ b/src/infrastructure/process/fake.rs @@ -0,0 +1,162 @@ +use std::{ + fs::{File, OpenOptions}, + io::Write, + os::unix::process::ExitStatusExt, + path::{Path, PathBuf}, + process::ExitStatus, + sync::{Arc, Mutex}, +}; + +use async_trait::async_trait; +use nix::sys::signal::Signal; +use tokio::sync::Notify; + +use super::{ProcessError, ProcessTrait, RunningProcess}; +use crate::infrastructure::shell::ShellCommand; + +/// Everything a [`FakeProcess::spawn`] call was asked to do. +#[derive(Debug, Clone)] +pub struct SpawnRecord { + pub program: String, + pub args: Vec, + pub cwd: PathBuf, + pub log_path: PathBuf, +} + +/// Test-side control over the [`FakeRunningProcess`] produced by one spawn: +/// when and how it finishes, whether it was killed, and what it "wrote" to +/// its log file while running. +pub struct FakeControl { + state: Mutex, + notify: Notify, + log_path: PathBuf, +} + +// Raw wait-status encoding (see waitpid(2)): an exit code lives in bits 8–15, +// a killing signal in the low 7 bits. Storing raw values lets `wait()` hand +// back a real `ExitStatus` via `ExitStatusExt::from_raw`. +struct FakeState { + raw_status: Option, + killed: bool, +} + +impl FakeControl { + pub fn write_log(&self, contents: &[u8]) { + let mut file = OpenOptions::new() + .append(true) + .open(&self.log_path) + .unwrap(); + file.write_all(contents).unwrap(); + } + + /// Unblock `wait()`, reporting exit with `exit_code`. + pub fn finish(&self, exit_code: i32) { + self.state.lock().unwrap().raw_status = Some(exit_code << 8); + self.notify.notify_one(); + } + + pub fn was_killed(&self) -> bool { + self.state.lock().unwrap().killed + } +} + +struct FakeSpawn { + record: SpawnRecord, + control: Arc, +} + +/// A [`ProcessTrait`] double that never executes anything: spawns are +/// recorded for later argv/cwd assertions and each returns a +/// [`FakeRunningProcess`] the test drives through its [`FakeControl`]. +#[derive(Default)] +pub struct FakeProcess { + spawns: Mutex>, +} + +impl FakeProcess { + pub fn new() -> Self { + Self::default() + } + + pub fn spawned(&self) -> Vec { + self.spawns + .lock() + .unwrap() + .iter() + .map(|spawn| spawn.record.clone()) + .collect() + } + + /// # Panics + /// + /// If nothing has been spawned yet. + pub fn last_child(&self) -> Arc { + self.spawns + .lock() + .unwrap() + .last() + .map(|spawn| spawn.control.clone()) + .expect("FakeProcess::last_child called before any spawn") + } +} + +impl ProcessTrait for FakeProcess { + fn spawn( + &self, + cmd: &ShellCommand, + cwd: &Path, + log_path: &Path, + ) -> Result, ProcessError> { + File::create(log_path)?; + + let control = Arc::new(FakeControl { + state: Mutex::new(FakeState { + raw_status: None, + killed: false, + }), + notify: Notify::new(), + log_path: log_path.to_path_buf(), + }); + self.spawns.lock().unwrap().push(FakeSpawn { + record: SpawnRecord { + program: cmd.program.clone(), + args: cmd.args.clone(), + cwd: cwd.to_path_buf(), + log_path: log_path.to_path_buf(), + }, + control: control.clone(), + }); + + Ok(Box::new(FakeRunningProcess { control })) + } +} + +struct FakeRunningProcess { + control: Arc, +} + +#[async_trait] +impl RunningProcess for FakeRunningProcess { + async fn wait(&mut self) -> Result { + loop { + // The state lock is released before the await; a finish()/kill() + // racing the check is not lost because Notify stores one permit. + let raw_status = self.control.state.lock().unwrap().raw_status; + if let Some(raw) = raw_status { + return Ok(ExitStatus::from_raw(raw)); + } + self.control.notify.notified().await; + } + } + + fn kill(&mut self) -> Result<(), ProcessError> { + let mut state = self.control.state.lock().unwrap(); + state.killed = true; + if state.raw_status.is_none() { + state.raw_status = Some(Signal::SIGTERM as i32); + } + drop(state); + self.control.notify.notify_one(); + Ok(()) + } +} diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs index 31a51d30..8e075c02 100644 --- a/src/infrastructure/process/mod.rs +++ b/src/infrastructure/process/mod.rs @@ -8,6 +8,11 @@ pub use r#trait::{ProcessError, ProcessTrait, RunningProcess}; #[cfg(test)] pub use r#trait::MockRunningProcess; +#[cfg(test)] +mod fake; +#[cfg(test)] +pub use fake::FakeProcess; + #[cfg(test)] mod tests; diff --git a/src/infrastructure/process/tests.rs b/src/infrastructure/process/tests.rs index 22ecfe6a..40634261 100644 --- a/src/infrastructure/process/tests.rs +++ b/src/infrastructure/process/tests.rs @@ -7,7 +7,7 @@ use std::{ use nix::{errno::Errno, sys::signal::kill, unistd::Pid}; -use super::{MockRunningProcess, OsProcess, ProcessTrait, RunningProcess}; +use super::{FakeProcess, MockRunningProcess, OsProcess, ProcessTrait, RunningProcess}; use crate::infrastructure::shell::ShellCommand; struct TempDir(PathBuf); @@ -219,3 +219,65 @@ async fn running_process_is_dyn_compatible_and_mockable() { let status = process.wait().await.unwrap(); assert!(status.success()); } + +#[tokio::test] +async fn fake_process_records_spawn_and_simulates_run() { + let dir = TempDir::new("fake_run"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").args(["build", "--alert=n"]); + + let mut process = fake.spawn(&cmd, dir.path(), &log).unwrap(); + + assert_eq!(fake.spawned().len(), 1); + let record = &fake.spawned()[0]; + assert_eq!(record.program, "kw"); + assert_eq!(record.args, vec!["build", "--alert=n"]); + assert_eq!(record.cwd, dir.path()); + assert_eq!(record.log_path, log); + + let control = fake.last_child(); + control.write_log(b"partial output\n"); + + // still running: wait must not resolve before finish() is called + let early_wait = tokio::time::timeout(Duration::from_millis(50), process.wait()).await; + assert!(early_wait.is_err()); + + let log_so_far = std::fs::read_to_string(&log).unwrap(); + assert_eq!(log_so_far, "partial output\n"); + + control.finish(0); + let status = process.wait().await.unwrap(); + assert!(status.success()); +} + +#[tokio::test] +async fn fake_finish_with_nonzero_code_yields_that_code() { + let dir = TempDir::new("fake_nonzero"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").arg("build"); + + let mut process = fake.spawn(&cmd, dir.path(), &log).unwrap(); + fake.last_child().finish(42); + + let status = process.wait().await.unwrap(); + assert!(!status.success()); + assert_eq!(status.code(), Some(42)); +} + +#[tokio::test] +async fn fake_kill_makes_wait_return_signal_status() { + let dir = TempDir::new("fake_kill"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").arg("build"); + + let mut process = fake.spawn(&cmd, dir.path(), &log).unwrap(); + + process.kill().unwrap(); + + assert!(fake.last_child().was_killed()); + let status = process.wait().await.unwrap(); + assert!(status.code().is_none()); +} From 2f980cbfe5714a38b9357abe3994a5e46d4608f8 Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Tue, 11 Aug 2026 17:21:57 -0300 Subject: [PATCH 4/5] fix(infrastructure): close process module races and fake fidelity gaps This commit hardens the process module after review. Kill tests wait for a real grandchild pid instead of racing an empty sidecar file, a failed spawn no longer leaves a truncated log behind, and the fake double treats terminal states as terminal so tests can simulate spawn failure faithfully. The module is unix-gated at declaration, matching kw's Linux-only process-group kill. This commit completes the kw integration's step 1. Signed-off-by: lorenzoberts --- src/infrastructure/mod.rs | 1 + src/infrastructure/process/fake.rs | 40 ++++++++-- src/infrastructure/process/mod.rs | 10 ++- src/infrastructure/process/tests.rs | 112 ++++++++++++++++++++++++---- src/infrastructure/process/trait.rs | 4 + 5 files changed, 140 insertions(+), 27 deletions(-) diff --git a/src/infrastructure/mod.rs b/src/infrastructure/mod.rs index d2e604c9..48908adf 100644 --- a/src/infrastructure/mod.rs +++ b/src/infrastructure/mod.rs @@ -3,6 +3,7 @@ pub mod errors; pub mod file_system; pub mod monitoring; pub mod net; +#[cfg(unix)] pub mod process; pub mod shell; pub mod terminal; diff --git a/src/infrastructure/process/fake.rs b/src/infrastructure/process/fake.rs index a87b6cb2..4d6c1af2 100644 --- a/src/infrastructure/process/fake.rs +++ b/src/infrastructure/process/fake.rs @@ -1,10 +1,13 @@ use std::{ fs::{File, OpenOptions}, - io::Write, + io::{self, Write}, os::unix::process::ExitStatusExt, path::{Path, PathBuf}, process::ExitStatus, - sync::{Arc, Mutex}, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Mutex, + }, }; use async_trait::async_trait; @@ -49,10 +52,15 @@ impl FakeControl { file.write_all(contents).unwrap(); } - /// Unblock `wait()`, reporting exit with `exit_code`. + /// Unblock `wait()`, reporting exit with `exit_code`. A terminal state is + /// terminal: a process already finished or killed does not exit later. pub fn finish(&self, exit_code: i32) { - self.state.lock().unwrap().raw_status = Some(exit_code << 8); - self.notify.notify_one(); + let mut state = self.state.lock().unwrap(); + if state.raw_status.is_none() { + state.raw_status = Some(exit_code << 8); + drop(state); + self.notify.notify_one(); + } } pub fn was_killed(&self) -> bool { @@ -71,6 +79,7 @@ struct FakeSpawn { #[derive(Default)] pub struct FakeProcess { spawns: Mutex>, + refuse_spawns: AtomicBool, } impl FakeProcess { @@ -78,6 +87,13 @@ impl FakeProcess { Self::default() } + /// Make `spawn` fail with an IO error, like a missing binary or an + /// unwritable log path would; no log file is created and nothing is + /// recorded, mirroring `OsProcess`'s failure behavior. + pub fn refuse_spawns(&self, refuse: bool) { + self.refuse_spawns.store(refuse, Ordering::Relaxed); + } + pub fn spawned(&self) -> Vec { self.spawns .lock() @@ -107,6 +123,12 @@ impl ProcessTrait for FakeProcess { cwd: &Path, log_path: &Path, ) -> Result, ProcessError> { + if self.refuse_spawns.load(Ordering::Relaxed) { + return Err(ProcessError::IoError(io::Error::new( + io::ErrorKind::NotFound, + "fake spawn failure", + ))); + } File::create(log_path)?; let control = Arc::new(FakeControl { @@ -151,12 +173,14 @@ impl RunningProcess for FakeRunningProcess { fn kill(&mut self) -> Result<(), ProcessError> { let mut state = self.control.state.lock().unwrap(); - state.killed = true; + // Mirrors the real kill()'s ESRCH tolerance: killing an already-dead + // process is a successful no-op, not a kill. if state.raw_status.is_none() { + state.killed = true; state.raw_status = Some(Signal::SIGTERM as i32); + drop(state); + self.control.notify.notify_one(); } - drop(state); - self.control.notify.notify_one(); Ok(()) } } diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs index 8e075c02..ae8e33c5 100644 --- a/src/infrastructure/process/mod.rs +++ b/src/infrastructure/process/mod.rs @@ -6,7 +6,7 @@ mod r#trait; pub use r#trait::{ProcessError, ProcessTrait, RunningProcess}; #[cfg(test)] -pub use r#trait::MockRunningProcess; +pub use r#trait::{MockProcessTrait, MockRunningProcess}; #[cfg(test)] mod fake; @@ -31,7 +31,7 @@ use nix::{ }; use tokio::process::{Child, Command}; -use super::shell::ShellCommand; +use crate::infrastructure::shell::ShellCommand; // No production caller exists until the kw integration wires KwActor; // kept per the CachePolicy precedent (src/lore/application/cache.rs). @@ -57,7 +57,11 @@ impl ProcessTrait for OsProcess { .stdout(Stdio::from(log_out)) .stderr(Stdio::from(log_err)) .process_group(0) - .spawn()?; + .spawn() + .inspect_err(|_| { + // a job that never started must not leave a truncated log + let _ = std::fs::remove_file(log_path); + })?; let pid = child .id() diff --git a/src/infrastructure/process/tests.rs b/src/infrastructure/process/tests.rs index 40634261..76df501b 100644 --- a/src/infrastructure/process/tests.rs +++ b/src/infrastructure/process/tests.rs @@ -1,4 +1,5 @@ use std::{ + io, os::unix::process::ExitStatusExt, path::{Path, PathBuf}, process::ExitStatus, @@ -7,7 +8,10 @@ use std::{ use nix::{errno::Errno, sys::signal::kill, unistd::Pid}; -use super::{FakeProcess, MockRunningProcess, OsProcess, ProcessTrait, RunningProcess}; +use super::{ + FakeProcess, MockProcessTrait, MockRunningProcess, OsProcess, ProcessError, ProcessTrait, + RunningProcess, +}; use crate::infrastructure::shell::ShellCommand; struct TempDir(PathBuf); @@ -52,6 +56,24 @@ fn pid_is_gone(pid: i32) -> bool { ) } +// The shell redirection creates the sidecar file before `echo $!` writes into +// it, so polling for existence can observe an empty file; poll for parseable +// content instead. +async fn await_sidecar_pid(path: &Path) -> i32 { + let mut pid = None; + wait_for( + || { + pid = std::fs::read_to_string(path) + .ok() + .and_then(|contents| contents.trim().parse::().ok()); + pid.is_some() + }, + Duration::from_secs(2), + ) + .await; + pid.expect("sidecar never received a grandchild pid") +} + #[tokio::test] async fn spawn_returns_while_process_still_running() { let dir = TempDir::new("spawn_returns"); @@ -138,13 +160,7 @@ async fn kill_terminates_process_group() { let mut process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); - let sidecar_ready = wait_for(|| sidecar.exists(), Duration::from_secs(2)).await; - assert!(sidecar_ready); - let grandchild_pid: i32 = std::fs::read_to_string(&sidecar) - .unwrap() - .trim() - .parse() - .unwrap(); + let grandchild_pid = await_sidecar_pid(&sidecar).await; process.kill().unwrap(); let status = tokio::time::timeout(Duration::from_secs(2), process.wait()) @@ -173,7 +189,7 @@ async fn kill_after_successful_exit_is_ok() { } #[tokio::test] -async fn spawn_missing_binary_returns_error() { +async fn spawn_missing_binary_returns_error_without_creating_log_file() { let dir = TempDir::new("missing_binary"); let log = dir.path().join("job.log"); let cmd = ShellCommand::new("__nonexistent_binary_patch_hub__"); @@ -181,6 +197,7 @@ async fn spawn_missing_binary_returns_error() { let result = OsProcess.spawn(&cmd, dir.path(), &log); assert!(result.is_err()); + assert!(!log.exists()); } #[tokio::test] @@ -193,13 +210,7 @@ async fn dropped_unreaped_process_group_is_killed() { let process = OsProcess.spawn(&cmd, dir.path(), &log).unwrap(); - let sidecar_ready = wait_for(|| sidecar.exists(), Duration::from_secs(2)).await; - assert!(sidecar_ready); - let grandchild_pid: i32 = std::fs::read_to_string(&sidecar) - .unwrap() - .trim() - .parse() - .unwrap(); + let grandchild_pid = await_sidecar_pid(&sidecar).await; // no wait(): dropping the handle must not orphan the process group drop(process); @@ -281,3 +292,72 @@ async fn fake_kill_makes_wait_return_signal_status() { let status = process.wait().await.unwrap(); assert!(status.code().is_none()); } + +#[tokio::test] +async fn fake_finish_after_kill_keeps_signal_status() { + let dir = TempDir::new("fake_finish_after_kill"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").arg("build"); + + let mut process = fake.spawn(&cmd, dir.path(), &log).unwrap(); + + process.kill().unwrap(); + // a killed process cannot exit 0 later; finish() must not resurrect it + fake.last_child().finish(0); + + let status = process.wait().await.unwrap(); + assert!(status.code().is_none()); + assert!(fake.last_child().was_killed()); +} + +#[tokio::test] +async fn fake_kill_after_finish_is_a_no_op_success() { + let dir = TempDir::new("fake_kill_after_finish"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").arg("build"); + + let mut process = fake.spawn(&cmd, dir.path(), &log).unwrap(); + fake.last_child().finish(0); + + process.kill().unwrap(); + + assert!(!fake.last_child().was_killed()); + let status = process.wait().await.unwrap(); + assert!(status.success()); +} + +#[tokio::test] +async fn fake_spawn_can_be_made_to_fail() { + let dir = TempDir::new("fake_spawn_failure"); + let log = dir.path().join("job.log"); + let fake = FakeProcess::new(); + let cmd = ShellCommand::new("kw").arg("build"); + + fake.refuse_spawns(true); + let result = fake.spawn(&cmd, dir.path(), &log); + + assert!(result.is_err()); + assert!(!log.exists()); + assert!(fake.spawned().is_empty()); +} + +#[tokio::test] +async fn mock_process_trait_can_simulate_spawn_failure() { + let dir = TempDir::new("mock_spawn_failure"); + let log = dir.path().join("job.log"); + let cmd = ShellCommand::new("kw").arg("build"); + + let mut mock = MockProcessTrait::new(); + mock.expect_spawn().return_once(|_, _, _| { + Err(ProcessError::IoError(io::Error::new( + io::ErrorKind::NotFound, + "kw not found", + ))) + }); + + let result = mock.spawn(&cmd, dir.path(), &log); + + assert!(result.is_err()); +} diff --git a/src/infrastructure/process/trait.rs b/src/infrastructure/process/trait.rs index 7a97cd59..32c53851 100644 --- a/src/infrastructure/process/trait.rs +++ b/src/infrastructure/process/trait.rs @@ -41,6 +41,10 @@ pub trait RunningProcess: Send { /// Send SIGTERM to the whole process group, not just the direct child. /// Idempotent: an already-gone group (ESRCH) is reported as success. + /// Signaling a leaderless group is intentional — it is how grandchildren + /// that outlive the leader get cleaned up — but in the narrow window where + /// the kernel has recycled the pgid, the signal could land on an unrelated + /// process group. /// /// Both methods take `&mut self`: a consumer that waits while staying able /// to cancel should `tokio::select!` between `wait()` and its cancel From 9c68a81aaad5aec397bd3e583c3ae96b03c20938 Mon Sep 17 00:00:00 2001 From: lorenzoberts Date: Mon, 24 Aug 2026 14:19:07 -0300 Subject: [PATCH 5/5] docs(infrastructure): drop forward-looking comments from the process module This commit keeps the dead_code allowances the unused OsProcess surface still needs, but stops explaining them in terms of a later KwActor that is not in this change. This commit is part of the kw integration's step 1. Signed-off-by: lorenzoberts Co-authored-by: Cursor --- src/infrastructure/process/mod.rs | 2 -- src/infrastructure/process/trait.rs | 3 --- 2 files changed, 5 deletions(-) diff --git a/src/infrastructure/process/mod.rs b/src/infrastructure/process/mod.rs index ae8e33c5..b97a88e2 100644 --- a/src/infrastructure/process/mod.rs +++ b/src/infrastructure/process/mod.rs @@ -33,8 +33,6 @@ use tokio::process::{Child, Command}; use crate::infrastructure::shell::ShellCommand; -// No production caller exists until the kw integration wires KwActor; -// kept per the CachePolicy precedent (src/lore/application/cache.rs). #[allow(dead_code)] pub struct OsProcess; diff --git a/src/infrastructure/process/trait.rs b/src/infrastructure/process/trait.rs index 32c53851..c9204c63 100644 --- a/src/infrastructure/process/trait.rs +++ b/src/infrastructure/process/trait.rs @@ -12,9 +12,6 @@ pub enum ProcessError { IoError(#[from] io::Error), } -// The two traits in this module have no production caller until the kw -// integration wires KwActor; kept per the CachePolicy precedent -// (src/lore/application/cache.rs). #[allow(dead_code)] #[automock] pub trait ProcessTrait: Send + Sync {