Merge pull request #853 from fabro-sh/codex/local-subprocess-ownership

Consolidate server Git subprocess ownership

.config/nextest.toml

@@ -15,6 +15,11 @@ leak-timeout = "500ms"
     filter = "package(fabro-workflow)"
     slow-timeout = { period = "2s", terminate-after = 3 }
 
+    # Real descendant regressions include bounded reaping and process probes.
+    # Leave room for their own watchdogs to run fail-safe fixture cleanup.
+    [[profile.default.overrides]]
+    filter = "package(fabro-proc) & binary(lifecycle)"
+    slow-timeout = { period = "10s", terminate-after = 3 }
 
 [profile.e2e]
 # E2E (ignored) tests: flag SLOW after 10s, hard-kill after 30s

Cargo.lock

@@ -2919,6 +2919,10 @@ dependencies = [
  "cc",
  "libc",
  "tempfile",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-util",
+ "tracing",
 ]
 
 [[package]]

lib/apps/fabro-server/src/git_checkout.rs

@@ -4,10 +4,12 @@ use std::time::Duration;
 
 use base64::Engine as _;
 use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use fabro_proc::ProcessError;
 use fabro_store::KeyedMutex;
 use fabro_types::{GitHubRepositorySlug, GitRunTarget};
+use tokio::fs;
 use tokio::process::Command;
-use tokio::{fs, time};
+use tokio_util::sync::CancellationToken;
 
 const GIT_CLONE_TIMEOUT: Duration = Duration::from_mins(2);
 const GIT_FETCH_TIMEOUT: Duration = Duration::from_mins(1);
@@ -483,18 +485,28 @@ async fn run_git_plan(plan: GitCommandPlan) -> Result<Vec<u8>, GitCommandError>
     if let Some(current_dir) = plan.current_dir.as_ref() {
         command.current_dir(current_dir);
     }
-    command.kill_on_drop(true);
-
-    let output = time::timeout(plan.timeout, command.output())
-        .await
-        .map_err(|_| GitCommandError::Timeout {
+    let output = fabro_proc::capture(
+        &mut command,
+        Some(plan.timeout),
+        &CancellationToken::new(),
+        None,
+    )
+    .await
+    .map_err(|err| match err {
+        ProcessError::TimedOut => GitCommandError::Timeout {
             command:      safe_command_label(&plan),
             timeout_secs: plan.timeout.as_secs(),
-        })?
-        .map_err(|err| GitCommandError::Spawn {
+        },
+        ProcessError::Io(source) => GitCommandError::Spawn {
             command: safe_command_label(&plan),
-            source:  err,
-        })?;
+            source,
+        },
+        ProcessError::Cancelled => GitCommandError::Spawn {
+            command: safe_command_label(&plan),
+            source:  std::io::Error::new(std::io::ErrorKind::Interrupted, err),
+        },
+    })?
+    .output;
 
     if output.status.success() {
         return Ok(output.stdout);
@@ -546,13 +558,107 @@ mod tests {
         reason = "Git checkout unit tests build local git repositories and inspect temp files synchronously."
     )]
 
+    use std::error::Error;
     use std::fs;
     use std::path::Path;
 
     use tempfile::TempDir;
+    use tokio::{task, time};
 
     use super::*;
 
+    fn shell_plan(script: &str, directory: &Path) -> GitCommandPlan {
+        GitCommandPlan {
+            program:          "sh".to_string(),
+            args:             vec!["-c".to_string(), script.to_string()],
+            env:              vec![
+                ("HOME".to_string(), directory.display().to_string()),
+                ("GIT_CONFIG_NOSYSTEM".to_string(), "1".to_string()),
+                ("GIT_CONFIG_GLOBAL".to_string(), "/dev/null".to_string()),
+            ],
+            current_dir:      Some(directory.to_path_buf()),
+            timeout:          Duration::from_secs(5),
+            sensitive_values: vec![],
+        }
+    }
+
+    #[tokio::test]
+    async fn git_plan_capture_preserves_bytes_environment_and_diagnostics() {
+        let temp = TempDir::new().unwrap();
+        let mut plan = shell_plan(
+            "printf '%s' \"$TEST_CAPTURE_VALUE\"; head -c 200000 /dev/zero; printf '\\377'",
+            temp.path(),
+        );
+        plan.env.push((
+            "TEST_CAPTURE_VALUE".to_string(),
+            "literal $(command)".to_string(),
+        ));
+        let output = run_git_plan(plan).await.unwrap();
+        assert!(output.starts_with(b"literal $(command)"));
+        assert_eq!(output.len(), 200_019);
+        assert_eq!(output.last(), Some(&255));
+
+        let mut plan = shell_plan(
+            "printf stdout; printf '%s' \"$TEST_CAPTURE_VALUE\" >&2; exit 7",
+            temp.path(),
+        );
+        let secret = "fixture-secret-value";
+        plan.env
+            .push(("TEST_CAPTURE_VALUE".to_string(), secret.to_string()));
+        plan.sensitive_values.push(secret.to_string());
+        let error = run_git_plan(plan).await.unwrap_err();
+        let message = error.to_string();
+        assert!(!message.contains(secret));
+        assert!(message.ends_with("REDACTED"));
+
+        let mut plan = shell_plan("exit 0", temp.path());
+        plan.program = "/nonexistent/fabro-test-git".to_string();
+        let error = run_git_plan(plan).await.unwrap_err();
+        assert!(Error::source(&error).unwrap().is::<std::io::Error>());
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn git_plan_deadline_includes_held_pipe() {
+        struct Cleanup(TempDir);
+        impl Drop for Cleanup {
+            fn drop(&mut self) {
+                if let Some(pid) = fs::read_to_string(self.0.path().join("leader.pid"))
+                    .ok()
+                    .and_then(|value| value.trim().parse().ok())
+                {
+                    fabro_proc::sigkill_process_group(pid);
+                }
+            }
+        }
+        let fixture = Cleanup(TempDir::new().unwrap());
+        let mut plan = shell_plan(
+            "echo $$ > leader.pid; sleep 60 & echo $! > helper.pid; exit 0",
+            fixture.0.path(),
+        );
+        plan.timeout = Duration::from_millis(200);
+        let error = time::timeout(Duration::from_secs(8), run_git_plan(plan))
+            .await
+            .unwrap()
+            .unwrap_err();
+        assert!(matches!(error, GitCommandError::Timeout { .. }));
+        let helper: u32 = fs::read_to_string(fixture.0.path().join("helper.pid"))
+            .unwrap()
+            .trim()
+            .parse()
+            .unwrap();
+        time::timeout(Duration::from_secs(5), async {
+            while task::spawn_blocking(move || fabro_proc::process_running_strict(helper))
+                .await
+                .unwrap()
+            {
+                time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .unwrap();
+    }
+
     fn repository_slug(value: &str) -> GitHubRepositorySlug {
         GitHubRepositorySlug::try_new(value).expect("slug should parse")
     }

lib/apps/fabro-server/src/run_manifest.rs

@@ -17,6 +17,7 @@ use fabro_graphviz::render::apply_direction;
 use fabro_llm::FabroClient;
 use fabro_llm::lithos_catalog::Catalog;
 use fabro_llm::probe::{self, ModelTestStatus};
+use fabro_proc::ProcessError;
 use fabro_sandbox::{
     CloneRequest, ProviderAccess, RunSandbox, SandboxSpec, sandbox_spec_for_environment,
 };
@@ -42,7 +43,9 @@ use fabro_workflow::workflow_bundle::{BundledWorkflow, ParsedWorkflowConfig, Wor
 use futures_util::stream::{self, StreamExt};
 use lithos_llm::catalog::ProviderId;
 use tokio::process::Command;
+#[cfg(test)]
 use tokio::time;
+use tokio_util::sync::CancellationToken;
 
 use crate::run_compiler;
 use crate::server::AppState;
@@ -882,13 +885,19 @@ async fn check_git_remote_ref(
 /// failure to its most useful message: stderr, then stdout, then the exit
 /// status.
 async fn run_ls_remote(mut command: Command) -> std::result::Result<(), String> {
-    // Dropping a timed-out `Command::output` future does not stop the child
-    // unless kill-on-drop is enabled.
-    command.kill_on_drop(true);
-    let output = time::timeout(Duration::from_secs(10), command.output())
-        .await
-        .map_err(|_| "git ls-remote timed out after 10s".to_string())?
-        .map_err(|err| format!("Failed to run git ls-remote: {err}"))?;
+    let output = fabro_proc::capture(
+        &mut command,
+        Some(Duration::from_secs(10)),
+        &CancellationToken::new(),
+        None,
+    )
+    .await
+    .map_err(|err| match err {
+        ProcessError::TimedOut => "git ls-remote timed out after 10s".to_string(),
+        ProcessError::Io(source) => format!("Failed to run git ls-remote: {source}"),
+        ProcessError::Cancelled => "git ls-remote cancelled".to_string(),
+    })?
+    .output;
 
     if output.status.success() {
         return Ok(());
@@ -1661,6 +1670,32 @@ mod tests {
 
     use super::*;
 
+    #[tokio::test]
+    async fn ls_remote_capture_keeps_diagnostic_precedence_and_unlimited_output() {
+        for (script, expected) in [
+            ("printf stdout; printf stderr >&2; exit 7", "stderr"),
+            ("printf stdout; exit 7", "stdout"),
+        ] {
+            let mut command = Command::new("sh");
+            command.args(["-c", script]);
+            assert_eq!(super::run_ls_remote(command).await.unwrap_err(), expected);
+        }
+        let mut command = Command::new("sh");
+        command.args(["-c", "head -c 200000 /dev/zero | tr '\\0' x >&2; exit 7"]);
+        assert_eq!(
+            super::run_ls_remote(command).await.unwrap_err().len(),
+            200_000
+        );
+        let mut command = Command::new("sh");
+        command.args(["-c", "exit 7"]);
+        assert!(
+            super::run_ls_remote(command)
+                .await
+                .unwrap_err()
+                .starts_with("git ls-remote exited with status")
+        );
+    }
+
     fn minimal_manifest() -> types::RunManifest {
         types::RunManifest {
             args:      None,

lib/foundation/fabro-proc/Cargo.toml

@@ -4,11 +4,17 @@ edition.workspace = true
 version.workspace = true
 publish = false
 license.workspace = true
-description = "Safe wrappers for process management primitives (signals, pre-exec hooks, title rewriting)"
+description = "Local subprocess ownership, capture, and OS process primitives"
 
 [lints]
 workspace = true
 
+[dependencies]
+tokio.workspace = true
+tokio-util.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
+
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
 

lib/foundation/fabro-proc/src/lib.rs

@@ -1,3 +1,7 @@
+//! Local subprocess ownership and capture, plus OS process primitives.
+//! Finite commands use [`capture`]; command and credential policy stay with
+//! callers. Process ownership remains private to capture.
+
 #![allow(
     unsafe_code,
     reason = "This crate wraps low-level OS or FFI APIs that require unsafe code."
@@ -24,3 +28,6 @@ pub use signal::{
     sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2,
 };
 pub use title::{init as title_init, set as title_set};
+
+mod process;
+pub use process::{CapturedOutput, ProcessError, capture};

lib/foundation/fabro-proc/src/process.rs

@@ -0,0 +1,280 @@
+//! Finite local commands: the leader and required output share one deadline.
+//!
+//! Capture and child waiting are scoped futures, never spawned. Cancellation
+//! drops the readers and kills the owned group before reaping the direct child.
+use std::future::{self, Future};
+use std::io;
+use std::process::{ExitStatus, Output, Stdio};
+use std::time::Duration;
+
+use tokio::io::{AsyncRead, AsyncReadExt};
+use tokio::process::{Child, Command};
+use tokio::time;
+use tokio_util::sync::CancellationToken;
+
+/// Infrastructure failure; command arguments, environment and output are
+/// absent.
+#[derive(thiserror::Error)]
+pub enum ProcessError {
+    #[error("process operation cancelled")]
+    Cancelled,
+    #[error("process operation timed out")]
+    TimedOut,
+    #[error("process I/O failed")]
+    Io(#[from] io::Error),
+}
+
+impl std::fmt::Debug for ProcessError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(self, f)
+    }
+}
+
+/// Owns a child and its original process group through output completion.
+///
+/// On Unix, drop synchronously requests group KILL. Tokio's child reaper owns
+/// the remaining direct-child wait (its orphan queue on Unix); no application
+/// cleanup task or pipe task is detached. Orderly paths await reaping. Drop
+/// reaping needs a live Tokio runtime and is best effort, not a bounded
+/// promise. Non-Unix cleanup covers only the direct child. Escaped groups,
+/// runtime/process destruction and uninterruptible OS failures are outside this
+/// contract.
+///
+/// Normal completion, including nonzero exit, preserves helpers that closed
+/// inherited pipes. Dropping an unfinished owner always requests cleanup.
+#[must_use]
+struct ProcessOwner {
+    child: Child,
+    group: Option<u32>,
+    armed: bool,
+}
+
+// Bounds direct-child reaping after immediate termination. Output readers
+// are dropped when capture fails, so there is no separate drain allowance.
+const CLEANUP_ALLOWANCE: Duration = Duration::from_secs(2);
+
+impl ProcessOwner {
+    /// Spawn a prepared command in a fresh group. All command policy stays with
+    /// the caller; only group setup and kill-on-drop are supplied here.
+    fn spawn(command: &mut Command) -> Result<Self, ProcessError> {
+        #[cfg(unix)]
+        command.process_group(0);
+        let child = command.kill_on_drop(true).spawn()?;
+        let group = child.id();
+        tracing::debug!(pid = group, "Spawned owned local process");
+        Ok(Self {
+            child,
+            group,
+            armed: true,
+        })
+    }
+
+    /// Wait for the leader and both readers under one deadline. Dropping the
+    /// completion future closes the pipes; the owner remains armed until the
+    /// command completes normally or termination and reaping succeed.
+    async fn complete(
+        mut self,
+        output: impl Future<Output = io::Result<()>>,
+        timeout: Option<Duration>,
+        cancel: &CancellationToken,
+    ) -> Result<ExitStatus, ProcessError> {
+        let deadline = async {
+            match timeout {
+                Some(duration) => time::sleep(duration).await,
+                None => future::pending().await,
+            }
+        };
+        let result = tokio::select! {
+            result = async {
+                let (status, ()) = tokio::try_join!(self.child.wait(), output)?;
+                Ok(status)
+            } => result,
+            () = cancel.cancelled() => Err(ProcessError::Cancelled),
+            () = deadline => Err(ProcessError::TimedOut),
+        };
+        match result {
+            Ok(status) => {
+                self.armed = false;
+                tracing::debug!("Local process and output completed");
+                Ok(status)
+            }
+            Err(failure) => {
+                tracing::debug!(reason = %failure, "Stopping local process");
+                if let Err(error) = self.terminate().await {
+                    // Preserve an original I/O failure and leave drop's
+                    // fallback armed. Cleanup failure is never called success.
+                    tracing::warn!(error_kind = ?error.kind(), "Local process cleanup failed");
+                    return Err(match failure {
+                        ProcessError::Io(_) => failure,
+                        _ => ProcessError::Io(error),
+                    });
+                }
+                self.armed = false;
+                Err(failure)
+            }
+        }
+    }
+
+    async fn terminate(&mut self) -> io::Result<()> {
+        #[cfg(unix)]
+        if let Some(group) = self.group {
+            kill_group(group)?;
+        }
+        self.child.start_kill()?;
+        time::timeout(CLEANUP_ALLOWANCE, self.child.wait())
+            .await
+            .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "process reaping timed out"))??;
+        Ok(())
+    }
+}
+
+impl Drop for ProcessOwner {
+    fn drop(&mut self) {
+        if self.armed {
+            #[cfg(unix)]
+            if let Some(group) = self.group {
+                if let Err(error) = kill_group(group) {
+                    tracing::warn!(error_kind = ?error.kind(), "Dropped local process group cleanup failed");
+                }
+            }
+            // Also covers non-Unix and a failed group signal. Tokio retains
+            // direct-child reaping responsibility when Child is dropped.
+            let _ = self.child.start_kill();
+        }
+    }
+}
+
+#[cfg(unix)]
+fn kill_group(group: u32) -> io::Result<()> {
+    let group = i32::try_from(group)
+        .ok()
+        .filter(|group| *group > 0)
+        .ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "invalid owned process group")
+        })?;
+    // SAFETY: positive saved child id names the fresh group established at spawn;
+    // negation targets that group, never the caller's group or all processes.
+    if unsafe { libc::kill(-group, libc::SIGKILL) } == 0 {
+        return Ok(());
+    }
+    let error = io::Error::last_os_error();
+    if error.raw_os_error() == Some(libc::ESRCH) {
+        Ok(())
+    } else {
+        Err(error)
+    }
+}
+
+/// Raw capture with per-stream completeness, retaining std's output shape.
+/// Deliberately does not implement Debug: output may contain credentials.
+pub struct CapturedOutput {
+    pub output:           Output,
+    pub stdout_truncated: bool,
+    pub stderr_truncated: bool,
+}
+
+/// Capture raw bytes with optional per-stream prefix retention. Reaching the
+/// cap continues draining and is not process failure. `None` retains unlimited
+/// bytes. Like Tokio Command::output, stdin configuration is preserved (a piped
+/// stdin is closed before waiting); stdout/stderr are captured concurrently.
+///
+/// Timeout/cancellation covers both leader exit and EOF. On failure, readers
+/// close immediately, the owned group is killed, and direct-child reaping gets
+/// up to two seconds. Dropping this future also requests group KILL; Tokio
+/// owns eventual direct-child reaping while its runtime remains alive. On
+/// non-Unix platforms only the direct child is killed. Normal completion,
+/// including nonzero exit, preserves helpers that closed inherited pipes.
+pub async fn capture(
+    command: &mut Command,
+    timeout: Option<Duration>,
+    cancel: &CancellationToken,
+    prefix_cap: Option<usize>,
+) -> Result<CapturedOutput, ProcessError> {
+    command.stdout(Stdio::piped()).stderr(Stdio::piped());
+    let mut owner = ProcessOwner::spawn(command)?;
+    let stdout = owner.child.stdout.take();
+    let stderr = owner.child.stderr.take();
+    let mut stdout_bytes = Vec::new();
+    let mut stderr_bytes = Vec::new();
+    let mut stdout_truncated = false;
+    let mut stderr_truncated = false;
+    let status = owner
+        .complete(
+            async {
+                tokio::try_join!(
+                    drain(stdout, &mut stdout_bytes, &mut stdout_truncated, prefix_cap),
+                    drain(stderr, &mut stderr_bytes, &mut stderr_truncated, prefix_cap),
+                )?;
+                Ok(())
+            },
+            timeout,
+            cancel,
+        )
+        .await?;
+    Ok(CapturedOutput {
+        output: Output {
+            status,
+            stdout: stdout_bytes,
+            stderr: stderr_bytes,
+        },
+        stdout_truncated,
+        stderr_truncated,
+    })
+}
+
+async fn drain<R: AsyncRead + Unpin>(
+    reader: Option<R>,
+    bytes: &mut Vec<u8>,
+    truncated: &mut bool,
+    cap: Option<usize>,
+) -> io::Result<()> {
+    if let Some(mut reader) = reader {
+        let mut chunk = vec![0; 8192];
+        loop {
+            let count = reader.read(&mut chunk).await?;
+            if count == 0 {
+                break;
+            }
+            let retained = count.min(cap.unwrap_or(usize::MAX).saturating_sub(bytes.len()));
+            bytes.extend_from_slice(&chunk[..retained]);
+            *truncated |= retained < count;
+        }
+    }
+    Ok(())
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use std::error::Error;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn read_failure_preserves_source_and_reaps_child() {
+        let mut command = Command::new("sleep");
+        command
+            .arg("60")
+            .stdout(Stdio::null())
+            .stderr(Stdio::null());
+        let owner = ProcessOwner::spawn(&mut command).unwrap();
+        let pid = owner.child.id().unwrap();
+        let output = async {
+            Err(io::Error::new(
+                io::ErrorKind::ConnectionReset,
+                "reader failed",
+            ))
+        };
+        let error = owner
+            .complete(
+                output,
+                Some(Duration::from_secs(5)),
+                &CancellationToken::new(),
+            )
+            .await
+            .unwrap_err();
+        assert!(error.source().unwrap().is::<io::Error>());
+        assert_eq!(error.to_string(), "process I/O failed");
+        assert_eq!(format!("{error:?}"), "process I/O failed");
+        assert!(!crate::process_exists(pid));
+    }
+}

lib/foundation/fabro-proc/tests/lifecycle.rs

@@ -0,0 +1,244 @@
+#![cfg(unix)]
+use std::process::Stdio;
+use std::time::Duration;
+
+use fabro_proc::{self as process, ProcessError};
+use tokio::process::Command;
+use tokio::{fs, task, time};
+use tokio_util::sync::CancellationToken;
+
+struct Fixture(tempfile::TempDir);
+impl Fixture {
+    fn new() -> Self {
+        Self(tempfile::tempdir().expect("fixture directory"))
+    }
+    fn command(&self, script: &str) -> Command {
+        let mut command = Command::new("sh");
+        command.args(["-c", script]).current_dir(self.0.path());
+        command
+    }
+    async fn ready(&self) {
+        time::timeout(Duration::from_secs(5), async {
+            loop {
+                if fs::read_to_string(self.0.path().join("helper.pid"))
+                    .await
+                    .ok()
+                    .and_then(|s| s.trim().parse::<u32>().ok())
+                    .is_some()
+                {
+                    break;
+                }
+                time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("helper readiness");
+    }
+    async fn pid(&self, file: &str) -> u32 {
+        fs::read_to_string(self.0.path().join(file))
+            .await
+            .expect("fixture process observation")
+            .trim()
+            .parse()
+            .expect("fixture process observation")
+    }
+    async fn stopped(&self) {
+        for name in ["leader.pid", "helper.pid"] {
+            let pid = self.pid(name).await;
+            time::timeout(Duration::from_secs(5), async {
+                while task::spawn_blocking(move || process::process_running_strict(pid))
+                    .await
+                    .expect("fixture process observation")
+                {
+                    time::sleep(Duration::from_millis(20)).await;
+                }
+            })
+            .await
+            .expect("owned process must stop");
+        }
+        // Direct child must also be reaped, not merely a zombie.
+        let leader = self.pid("leader.pid").await;
+        time::timeout(Duration::from_secs(5), async {
+            while process::process_exists(leader) {
+                time::sleep(Duration::from_millis(20)).await;
+            }
+        })
+        .await
+        .expect("direct child must be reaped while runtime lives");
+    }
+}
+impl Drop for Fixture {
+    #[expect(
+        clippy::disallowed_methods,
+        reason = "Fail-safe test cleanup must run synchronously on panic"
+    )]
+    fn drop(&mut self) {
+        for name in ["leader.pid", "helper.pid"] {
+            if let Some(pid) = std::fs::read_to_string(self.0.path().join(name))
+                .ok()
+                .and_then(|s| s.trim().parse().ok())
+            {
+                if name == "leader.pid" {
+                    process::sigkill_process_group(pid);
+                } else {
+                    process::sigkill(pid);
+                }
+            }
+        }
+    }
+}
+const HELD: &str = "echo $$ > leader.pid; sleep 60 & echo $! > helper.pid; printf partial; exit 0";
+const WAIT: &str = "echo $$ > leader.pid; sleep 60 >/dev/null 2>&1 & echo $! > helper.pid; wait";
+const IGNORE_TERM: &str = "echo $$ > leader.pid; trap 'exit 0' TERM; sh -c 'trap \"\" TERM; echo $$ > helper.pid; exec sleep 60' & wait";
+
+#[tokio::test]
+async fn abort_execution_and_held_pipe_drain_reaps_and_kills_group() {
+    for script in [WAIT, HELD] {
+        let fixture = Fixture::new();
+        let mut command = fixture.command(script);
+        let task = tokio::spawn(async move {
+            process::capture(&mut command, None, &CancellationToken::new(), None).await
+        });
+        fixture.ready().await;
+        if script == HELD {
+            let leader = fixture.pid("leader.pid").await;
+            time::timeout(Duration::from_secs(5), async {
+                while process::process_exists(leader) {
+                    time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .unwrap();
+        }
+        task.abort();
+        assert!(matches!(task.await, Err(error) if error.is_cancelled()));
+        fixture.stopped().await;
+    }
+}
+
+#[tokio::test]
+async fn timeout_and_cancellation_stop_real_descendants() {
+    for cancel in [false, true] {
+        let fixture = Fixture::new();
+        let mut command = fixture.command(IGNORE_TERM);
+        let token = CancellationToken::new();
+        let run = process::capture(&mut command, Some(Duration::from_millis(500)), &token, None);
+        let trigger = async {
+            fixture.ready().await;
+            if cancel {
+                token.cancel();
+            }
+        };
+        let (result, ()) =
+            time::timeout(Duration::from_secs(8), async { tokio::join!(run, trigger) })
+                .await
+                .unwrap();
+        assert!(matches!(
+            (cancel, result),
+            (true, Err(ProcessError::Cancelled)) | (false, Err(ProcessError::TimedOut))
+        ));
+        fixture.stopped().await;
+    }
+}
+
+#[tokio::test]
+async fn normal_exit_preserves_helpers_with_closed_pipes() {
+    for code in [0, 7] {
+        let fixture = Fixture::new();
+        let mut command = fixture.command(&format!("echo $$ > leader.pid; sleep 60 </dev/null >/dev/null 2>&1 & echo $! > helper.pid; exit {code}"));
+        let output = process::capture(
+            &mut command,
+            Some(Duration::from_secs(5)),
+            &CancellationToken::new(),
+            None,
+        )
+        .await
+        .unwrap();
+        assert_eq!(output.output.status.code(), Some(code));
+        assert!(process::process_exists(fixture.pid("helper.pid").await));
+    }
+}
+
+#[tokio::test]
+async fn prefix_cap_drains_both_streams_and_preserves_raw_bytes() {
+    for cap in [None, Some(0), Some(257)] {
+        let mut command = Command::new("sh");
+        command.args([
+            "-c",
+            "head -c 200000 /dev/zero & head -c 200000 /dev/zero >&2 & wait; printf '\\377'",
+        ]);
+        let output = time::timeout(
+            Duration::from_secs(8),
+            process::capture(&mut command, None, &CancellationToken::new(), cap),
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        assert!(output.output.status.success());
+        assert_eq!(output.output.stdout.len(), cap.unwrap_or(200_001));
+        assert_eq!(output.output.stderr.len(), cap.unwrap_or(200_000));
+        assert_eq!(output.stdout_truncated, cap.is_some());
+        assert_eq!(output.stderr_truncated, cap.is_some());
+        if cap.is_none() {
+            assert_eq!(output.output.stdout.last(), Some(&255));
+        }
+    }
+    let mut command = Command::new("sh");
+    command.args(["-c", "printf abcd; printf x >&2"]);
+    let output = process::capture(&mut command, None, &CancellationToken::new(), Some(2))
+        .await
+        .unwrap();
+    assert!(output.stdout_truncated);
+    assert!(!output.stderr_truncated);
+}
+
+#[tokio::test]
+async fn spawn_failure_preserves_source_without_rendering_command_details() {
+    use std::error::Error;
+    let mut command = Command::new("/nonexistent/fabro-test-executable");
+    let error = process::capture(&mut command, None, &CancellationToken::new(), None)
+        .await
+        .err()
+        .unwrap();
+    assert!(error.source().unwrap().is::<std::io::Error>());
+    assert_eq!(error.to_string(), "process I/O failed");
+    assert_eq!(format!("{error:?}"), "process I/O failed");
+}
+
+#[tokio::test]
+async fn capture_preserves_prepared_stdin_and_literal_arguments() {
+    let fixture = Fixture::new();
+    let input_path = fixture.0.path().join("input");
+    fs::write(&input_path, b"configured stdin\n").await.unwrap();
+    let input = fs::File::open(&input_path).await.unwrap().into_std().await;
+    let mut command = Command::new("sh");
+    command
+        .args([
+            "-c",
+            "cat; printf '%s' \"$1\"",
+            "fixture",
+            "$(must-not-run)",
+        ])
+        .stdin(Stdio::from(input));
+    let result = process::capture(
+        &mut command,
+        Some(Duration::from_secs(5)),
+        &CancellationToken::new(),
+        None,
+    )
+    .await
+    .unwrap();
+    assert_eq!(result.output.stdout, b"configured stdin\n$(must-not-run)");
+    let mut command = Command::new("cat");
+    command.stdin(Stdio::piped());
+    let result = process::capture(
+        &mut command,
+        Some(Duration::from_secs(5)),
+        &CancellationToken::new(),
+        None,
+    )
+    .await
+    .unwrap();
+    assert!(result.output.stdout.is_empty());
+    assert!(result.output.status.success());
+}