Share one supervised process runner between server and CLI Git

The CLI's native Git runner and the server's run_git_plan each hand-rolled
the same mechanics: kill-on-drop, a wall-clock timeout, output capture,
and (only in the CLI) process-group teardown, bounded capture, and
cancellation. Add fabro_proc::SupervisedCommand, which owns stdio, the
process group, the timeout, cooperative cancellation, and bounded
capture, and put both runners on it. The server gains group teardown on
timeout, so helpers a stuck clone or fetch spawned no longer outlive it;
the CLI keeps discarding output on failure and gains nothing but less
code. The hardened -c overrides become one named list in the CLI.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Cargo.lock

@@ -2916,6 +2916,9 @@ dependencies = [
  "cc",
  "libc",
  "tempfile",
+ "thiserror 2.0.18",
+ "tokio",
+ "tokio-util",
 ]
 
 [[package]]

lib/apps/fabro-cli/src/commands/run/remote_workflow.rs

@@ -1,15 +1,15 @@
 //! Native Git acquisition owned by the CLI, with no server credential lookup.
 use std::future::Future;
 use std::path::{Path, PathBuf};
-use std::process::{ExitStatus, Stdio};
+use std::process::ExitStatus;
 use std::time::Duration;
 
 use anyhow::{Context as _, bail};
 use fabro_manifest::CollectedWorkflowClosure;
+use fabro_proc::{SupervisedCommand, SupervisedError};
 use fabro_types::{GitHubRepositorySlug, GitRunTarget, repository};
-use tokio::io::{AsyncRead, AsyncReadExt as _};
 use tokio::process::Command;
-use tokio::{fs, signal as tokio_signal, task, time};
+use tokio::{fs, signal as tokio_signal, task};
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 
@@ -18,6 +18,33 @@ use crate::args::RunArgs;
 
 const OUTPUT_LIMIT: usize = 64 * 1024;
 
+/// Configuration overrides that keep an untrusted checkout from running code
+/// or rewriting bytes: no hooks or fsmonitor, no LFS smudge, no submodule
+/// recursion, no `ext::` transport, no background maintenance, no line-ending
+/// conversion.
+const HARDENED_GIT_CONFIG: &[&str] = &[
+    "-c",
+    "core.hooksPath=/dev/null",
+    "-c",
+    "core.fsmonitor=false",
+    "-c",
+    "filter.lfs.smudge=",
+    "-c",
+    "filter.lfs.process=",
+    "-c",
+    "filter.lfs.required=false",
+    "-c",
+    "submodule.recurse=false",
+    "-c",
+    "protocol.ext.allow=never",
+    "-c",
+    "maintenance.auto=0",
+    "-c",
+    "gc.auto=0",
+    "-c",
+    "core.autocrlf=false",
+];
+
 #[derive(Debug, thiserror::Error)]
 pub(super) enum RemoteWorkflowError {
     #[error(
@@ -64,34 +91,10 @@ impl NativeGit {
         args: &[&str],
         cancel: &CancellationToken,
     ) -> Result<Vec<u8>, RemoteWorkflowError> {
-        if cancel.is_cancelled() {
-            return Err(RemoteWorkflowError::Cancelled);
-        }
         let mut command = Command::new("git");
         command
             .current_dir(cwd)
-            .args([
-                "-c",
-                "core.hooksPath=/dev/null",
-                "-c",
-                "core.fsmonitor=false",
-                "-c",
-                "filter.lfs.smudge=",
-                "-c",
-                "filter.lfs.process=",
-                "-c",
-                "filter.lfs.required=false",
-                "-c",
-                "submodule.recurse=false",
-                "-c",
-                "protocol.ext.allow=never",
-                "-c",
-                "maintenance.auto=0",
-                "-c",
-                "gc.auto=0",
-                "-c",
-                "core.autocrlf=false",
-            ])
+            .args(HARDENED_GIT_CONFIG)
             .args(args)
             .env("GIT_TERMINAL_PROMPT", "0")
             .env("GIT_LFS_SKIP_SMUDGE", "1")
@@ -99,48 +102,30 @@ impl NativeGit {
             .env_remove("GIT_WORK_TREE")
             .env_remove("GIT_INDEX_FILE")
             .env_remove("GIT_OBJECT_DIRECTORY")
-            .env_remove("GIT_ALTERNATE_OBJECT_DIRECTORIES")
-            .stdin(Stdio::null())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .kill_on_drop(true);
-        #[cfg(unix)]
-        command.process_group(0);
+            .env_remove("GIT_ALTERNATE_OBJECT_DIRECTORIES");
         #[cfg(test)]
         command.envs(self.environment.iter().cloned());
-        let mut child = command.spawn()?;
-        let process_id = child.id();
-        let mut stdout = child.stdout.take().expect("piped Git stdout exists");
-        let mut stderr = child.stderr.take().expect("piped Git stderr exists");
-        let result = tokio::select! {
-            biased;
-            () = cancel.cancelled() => Err(RemoteWorkflowError::Cancelled),
-            () = time::sleep(self.timeout) => Err(RemoteWorkflowError::Timeout),
-            result = async {
-                let (status, (stdout, overflow), _) = tokio::try_join!(
-                    child.wait(), capture(&mut stdout), capture(&mut stderr)
-                )?;
-                if !status.success() {
-                    // Output may contain arbitrary helper/config secrets, even after pattern
-                    // redaction. Never retain it in an error/cause chain or tracing event.
-                    return Err(RemoteWorkflowError::Process { operation, status });
-                }
-                if overflow { return Err(RemoteWorkflowError::OutputLimit); }
-                Ok(stdout)
-            } => result,
-        };
-        // A timed-out, cancelled, or failed Git may leave helpers running in
-        // its process group; terminate the group, then reap Git. A successful
-        // Git has closed its pipes, and helpers it deliberately left behind
-        // (such as `credential-cache--daemon`) keep serving later commands.
-        if result.is_err() {
-            #[cfg(unix)]
-            if let Some(id) = process_id {
-                fabro_proc::sigkill_process_group(id);
-            }
-            child.kill().await?;
+        let output = SupervisedCommand::new(command, self.timeout)
+            .output_limit(OUTPUT_LIMIT)
+            .run(cancel)
+            .await
+            .map_err(|error| match error {
+                SupervisedError::Timeout { .. } => RemoteWorkflowError::Timeout,
+                SupervisedError::Cancelled => RemoteWorkflowError::Cancelled,
+                SupervisedError::Io(source) => RemoteWorkflowError::Io(source),
+            })?;
+        if !output.status.success() {
+            // Output may contain arbitrary helper/config secrets, even after pattern
+            // redaction. Never retain it in an error/cause chain or tracing event.
+            return Err(RemoteWorkflowError::Process {
+                operation,
+                status: output.status,
+            });
         }
-        result
+        if output.stdout_truncated {
+            return Err(RemoteWorkflowError::OutputLimit);
+        }
+        Ok(output.stdout)
     }
 
     /// Create and initialize an empty scratch repository. Temporary files are
@@ -360,21 +345,6 @@ impl NativeGit {
     }
 }
 
-async fn capture(reader: &mut (impl AsyncRead + Unpin)) -> std::io::Result<(Vec<u8>, bool)> {
-    let mut captured = Vec::new();
-    let mut overflow = false;
-    let mut buffer = vec![0; 8192];
-    loop {
-        let count = reader.read(&mut buffer).await?;
-        if count == 0 {
-            return Ok((captured, overflow));
-        }
-        let keep = count.min(OUTPUT_LIMIT - captured.len());
-        captured.extend_from_slice(&buffer[..keep]);
-        overflow |= keep != count;
-    }
-}
-
 fn exact_record(records: &str, reference: &str) -> anyhow::Result<Option<String>> {
     let mut found = None;
     for line in records.lines() {
@@ -544,6 +514,7 @@ mod tests {
     use nix::sys::signal::{self, Signal};
     use nix::sys::stat::Mode;
     use nix::unistd;
+    use tokio::time;
 
     use super::super::test_support::{commit_all, write_workflow};
     use super::*;

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::{SupervisedCommand, SupervisedError};
 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,17 +485,25 @@ 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())
+    // Server-side checkouts have no caller to cancel them; the timeout is the
+    // only deadline.
+    let output = SupervisedCommand::new(command, plan.timeout)
+        .run(&CancellationToken::new())
         .await
-        .map_err(|_| GitCommandError::Timeout {
-            command:      safe_command_label(&plan),
-            timeout_secs: plan.timeout.as_secs(),
-        })?
-        .map_err(|err| GitCommandError::Spawn {
-            command: safe_command_label(&plan),
-            source:  err,
+        .map_err(|error| match error {
+            SupervisedError::Timeout { .. } => GitCommandError::Timeout {
+                command:      safe_command_label(&plan),
+                timeout_secs: plan.timeout.as_secs(),
+            },
+            SupervisedError::Cancelled => GitCommandError::Spawn {
+                command: safe_command_label(&plan),
+                source:  std::io::Error::new(std::io::ErrorKind::Interrupted, "cancelled"),
+            },
+            SupervisedError::Io(source) => GitCommandError::Spawn {
+                command: safe_command_label(&plan),
+                source,
+            },
         })?;
 
     if output.status.success() {

lib/foundation/fabro-proc/Cargo.toml

@@ -9,6 +9,11 @@ description = "Safe wrappers for process management primitives (signals, pre-exe
 [lints]
 workspace = true
 
+[dependencies]
+thiserror.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true
+
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
 

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

@@ -8,6 +8,7 @@ mod flock;
 #[cfg(unix)]
 mod pre_exec;
 mod signal;
+mod supervised;
 mod title;
 
 #[cfg(unix)]
@@ -23,4 +24,5 @@ pub use signal::{process_exists, process_group_alive, process_running, process_r
 pub use signal::{
     sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2,
 };
+pub use supervised::{SupervisedCommand, SupervisedError, SupervisedOutput};
 pub use title::{init as title_init, set as title_set};

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

@@ -0,0 +1,235 @@
+//! Supervised child processes: an owned process group, a wall-clock timeout,
+//! cooperative cancellation, and bounded output capture.
+//!
+//! Callers configure a [`tokio::process::Command`] and hand it over; the
+//! supervisor owns stdio, the process group, and teardown. Helpers a child
+//! spawns (Git credential helpers, for example) share its group, so an
+//! interrupted or timed-out command takes them down with it.
+use std::process::{ExitStatus, Stdio};
+use std::time::Duration;
+
+use tokio::io::{AsyncRead, AsyncReadExt as _};
+use tokio::process::Command;
+use tokio::time;
+use tokio_util::sync::CancellationToken;
+
+/// Output of a child that ran to completion, successfully or not.
+#[derive(Debug)]
+pub struct SupervisedOutput {
+    pub status:           ExitStatus,
+    pub stdout:           Vec<u8>,
+    pub stderr:           Vec<u8>,
+    /// Stdout exceeded the configured limit and was cut off.
+    pub stdout_truncated: bool,
+    /// Stderr exceeded the configured limit and was cut off.
+    pub stderr_truncated: bool,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SupervisedError {
+    #[error("command timed out after {}s", timeout.as_secs())]
+    Timeout { timeout: Duration },
+    #[error("command cancelled")]
+    Cancelled,
+    #[error("failed to run command")]
+    Io(#[from] std::io::Error),
+}
+
+/// A command run under supervision. Stdin is closed; stdout and stderr are
+/// captured; on Unix the child leads its own process group.
+pub struct SupervisedCommand {
+    command:      Command,
+    timeout:      Duration,
+    output_limit: Option<usize>,
+}
+
+impl SupervisedCommand {
+    pub fn new(command: Command, timeout: Duration) -> Self {
+        Self {
+            command,
+            timeout,
+            output_limit: None,
+        }
+    }
+
+    /// Cap each captured stream at `limit` bytes; the remainder is drained
+    /// and reported as truncated rather than buffered.
+    #[must_use]
+    pub fn output_limit(mut self, limit: usize) -> Self {
+        self.output_limit = Some(limit);
+        self
+    }
+
+    /// Run the child to completion. A timeout, cancellation, or I/O failure
+    /// kills the child's whole process group before returning; a child that
+    /// exits on its own, with any status, leaves the helpers it deliberately
+    /// left behind alone.
+    pub async fn run(
+        self,
+        cancel: &CancellationToken,
+    ) -> Result<SupervisedOutput, SupervisedError> {
+        let Self {
+            mut command,
+            timeout,
+            output_limit,
+        } = self;
+        if cancel.is_cancelled() {
+            return Err(SupervisedError::Cancelled);
+        }
+        command
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true);
+        #[cfg(unix)]
+        command.process_group(0);
+        let mut child = command.spawn()?;
+        let process_id = child.id();
+        let mut stdout = child.stdout.take().expect("piped stdout exists");
+        let mut stderr = child.stderr.take().expect("piped stderr exists");
+        let result = tokio::select! {
+            biased;
+            () = cancel.cancelled() => Err(SupervisedError::Cancelled),
+            () = time::sleep(timeout) => Err(SupervisedError::Timeout { timeout }),
+            result = async {
+                let (status, (stdout, stdout_truncated), (stderr, stderr_truncated)) = tokio::try_join!(
+                    child.wait(),
+                    capture(&mut stdout, output_limit),
+                    capture(&mut stderr, output_limit),
+                )?;
+                Ok(SupervisedOutput { status, stdout, stderr, stdout_truncated, stderr_truncated })
+            } => result,
+        };
+        if result.is_err() {
+            #[cfg(unix)]
+            if let Some(id) = process_id {
+                crate::sigkill_process_group(id);
+            }
+            // The original error is the useful one; `kill_on_drop` backs this up.
+            let _reaped = child.kill().await;
+        }
+        result
+    }
+}
+
+async fn capture(
+    reader: &mut (impl AsyncRead + Unpin),
+    limit: Option<usize>,
+) -> std::io::Result<(Vec<u8>, bool)> {
+    let mut captured = Vec::new();
+    let mut truncated = false;
+    let mut buffer = vec![0; 8192];
+    loop {
+        let count = reader.read(&mut buffer).await?;
+        if count == 0 {
+            return Ok((captured, truncated));
+        }
+        let keep = limit.map_or(count, |limit| count.min(limit - captured.len()));
+        captured.extend_from_slice(&buffer[..keep]);
+        truncated |= keep != count;
+    }
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    #![expect(
+        clippy::disallowed_methods,
+        reason = "supervisor tests read small pid files written by fixture scripts"
+    )]
+
+    use super::*;
+
+    fn shell(script: &str) -> Command {
+        let mut command = Command::new("/bin/sh");
+        command.args(["-c", script]);
+        command
+    }
+
+    #[tokio::test]
+    async fn supervised_command_captures_both_streams_and_any_status() {
+        let output = SupervisedCommand::new(
+            shell("printf out; printf err >&2; exit 3"),
+            Duration::from_secs(5),
+        )
+        .run(&CancellationToken::new())
+        .await
+        .unwrap();
+        assert_eq!(output.status.code(), Some(3));
+        assert_eq!(output.stdout, b"out");
+        assert_eq!(output.stderr, b"err");
+        assert!(!output.stdout_truncated && !output.stderr_truncated);
+    }
+
+    #[tokio::test]
+    async fn supervised_command_drains_past_the_output_limit() {
+        let output = SupervisedCommand::new(
+            shell("i=0; while [ $i -lt 9000 ]; do printf 'sixteen-bytes---\\n'; i=$((i+1)); done"),
+            Duration::from_secs(10),
+        )
+        .output_limit(1024)
+        .run(&CancellationToken::new())
+        .await
+        .unwrap();
+        assert!(output.status.success());
+        assert_eq!(output.stdout.len(), 1024);
+        assert!(output.stdout_truncated);
+        assert!(!output.stderr_truncated);
+    }
+
+    #[tokio::test]
+    async fn supervised_command_kills_the_group_on_timeout_and_cancel() {
+        for timeout in [true, false] {
+            let dir = tempfile::tempdir().unwrap();
+            let pid_file = dir.path().join("pid");
+            let mut command = shell("printf '%s' $$ > pid; exec /bin/sleep 60");
+            command.current_dir(dir.path());
+            let cancel = CancellationToken::new();
+            let trigger = async {
+                if !timeout {
+                    time::timeout(Duration::from_secs(5), async {
+                        while !pid_file.exists() {
+                            time::sleep(Duration::from_millis(10)).await;
+                        }
+                    })
+                    .await
+                    .unwrap();
+                    cancel.cancel();
+                }
+            };
+            let supervised = SupervisedCommand::new(
+                command,
+                if timeout {
+                    Duration::from_millis(500)
+                } else {
+                    Duration::from_mins(1)
+                },
+            );
+            let (result, ()) = tokio::join!(supervised.run(&cancel), trigger);
+            let error = result.unwrap_err();
+            assert!(
+                matches!(error, SupervisedError::Timeout { .. }) == timeout,
+                "{error}"
+            );
+            let pid: u32 = std::fs::read_to_string(&pid_file).unwrap().parse().unwrap();
+            assert!(!crate::process_exists(pid));
+        }
+        assert!(matches!(
+            SupervisedCommand::new(shell("true"), Duration::from_secs(1))
+                .run(&CancellationToken::new().tap_cancel())
+                .await
+                .unwrap_err(),
+            SupervisedError::Cancelled
+        ));
+    }
+
+    trait TapCancel {
+        fn tap_cancel(self) -> Self;
+    }
+
+    impl TapCancel for CancellationToken {
+        fn tap_cancel(self) -> Self {
+            self.cancel();
+            self
+        }
+    }
+}