From 7d28227feefacb63c49148ac3ed90c4c30a97184 Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 6 Nov 2026 03:08:55 +0100 Subject: [PATCH] hive-bash-mcp: a task writes to its files directly, so a backgrounded child cannot hold it open --- hive-bash-mcp/src/runner.rs | 146 +++++++++++++----------------------- 1 file changed, 54 insertions(+), 92 deletions(-) diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 1138eccf..48b15012 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -35,7 +35,6 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Result, bail}; use hive_agent_sock::Request as TodoReq; use hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups}; -use tokio::io::AsyncWriteExt as _; use tokio::sync::Notify; use crate::paths; @@ -755,13 +754,21 @@ async fn exec_cmd( // fixed absolute path (coreutils, present on NixOS), and it resolves // `bash` via PATH — the same controlled PATH the daemon's systemd unit // sets. Avoids hardcoding a nix store / generation path in the binary. + // + // The child writes straight to the task's `.out`/`.err` files, not through + // pipes this daemon copies: a backgrounded grandchild (`server &`) keeps a + // pipe's write end open after bash exits, so a copy that waits for EOF + // would hold the task in `running` for as long as that process lives, and + // `kill` and the timeout would have nothing left to act on. + let stdout_file = std::fs::File::create(out_path)?; + let stderr_file = std::fs::File::create(err_path)?; let mut cmd_builder = Command::new("/usr/bin/env"); cmd_builder .arg("bash") .arg("-c") .arg(format!("set -o pipefail; {cmd}")) - .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); + .stdout(std::process::Stdio::from(stdout_file)) + .stderr(std::process::Stdio::from(stderr_file)); let cgroup_procs = cgroup_procs.map(std::os::fd::AsRawFd::as_raw_fd); // SAFETY: `setpgid`/`nice`/`write` are async-signal-safe and only touch // the calling (child) process before exec — no allocations, no locks. @@ -783,22 +790,6 @@ async fn exec_cmd( // With `setpgid(0, 0)` the child's pgid equals its pid. let pgid = child.id().map(u32_to_i32); - let stdout = child.stdout.take().expect("stdout piped"); - let stderr = child.stderr.take().expect("stderr piped"); - - let copy_out = { - let path = out_path.to_path_buf(); - tokio::spawn( - async move { copy_stream_to_file(tokio::io::BufReader::new(stdout), &path).await }, - ) - }; - let copy_err = { - let path = err_path.to_path_buf(); - tokio::spawn( - async move { copy_stream_to_file(tokio::io::BufReader::new(stderr), &path).await }, - ) - }; - // A future that fires when the timeout elapses, or never if `None`. let timeout_fut = async { match timeout_secs { @@ -836,9 +827,6 @@ async fn exec_cmd( } }; - warn_if_copy_failed("stdout", out_path, copy_out.await); - warn_if_copy_failed("stderr", err_path, copy_err.await); - let status = wait_res?; let outcome = if let Some(forced) = killed { ExecOutcome::Killed { forced } @@ -855,38 +843,6 @@ fn u32_to_i32(v: u32) -> i32 { i32::try_from(v).unwrap_or(i32::MAX) } -/// Copies `reader` into a new file at `path`, flushing at the end. On error -/// the bytes copied before it stay in the file. -async fn copy_stream_to_file(mut reader: R, path: &Path) -> std::io::Result<()> -where - R: tokio::io::AsyncRead + Unpin, -{ - let mut f = tokio::fs::File::create(path).await?; - let copied = tokio::io::copy(&mut reader, &mut f).await; - f.flush().await?; - copied.map(|_| ()) -} - -/// Warns when a stream copy failed or its task panicked: the log file is then -/// incomplete while the task still reports its exit status. -fn warn_if_copy_failed( - stream: &str, - path: &Path, - joined: Result, tokio::task::JoinError>, -) { - let error = match joined { - Ok(Ok(())) => return, - Ok(Err(e)) => format!("{e:?}"), - Err(e) => format!("{e:?}"), - }; - tracing::warn!( - stream, - path = %path.display(), - error, - "bash_runner: capturing task output failed, the log file is incomplete" - ); -} - fn tail_file(path: &Path, max_bytes: usize) -> Option { let data = std::fs::read(path).ok()?; let slice = if data.len() > max_bytes { @@ -946,7 +902,12 @@ fn done_summary(id: &str, summary: &str, output: Option<(bool, bool)>) -> String #[cfg(test)] mod tests { - use super::validate_task_name; + use std::sync::atomic::AtomicBool; + use std::time::Duration; + + use tokio::sync::Notify; + + use super::{ExecOutcome, exec_cmd, validate_task_name}; use crate::test_util::with_harness_dir; use hive_types::Ident; @@ -1227,27 +1188,6 @@ mod tests { }); } - /// Yields `data`, then fails with an I/O error. - struct FailAfter { - data: &'static [u8], - sent: bool, - } - - impl tokio::io::AsyncRead for FailAfter { - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - if self.sent { - return std::task::Poll::Ready(Err(std::io::Error::other("pipe broke"))); - } - self.sent = true; - buf.put_slice(self.data); - std::task::Poll::Ready(Ok(())) - } - } - fn block_on(f: F) -> F::Output { tokio::runtime::Builder::new_current_thread() .enable_all() @@ -1256,25 +1196,47 @@ mod tests { .block_on(f) } - #[test] - fn copy_error_is_returned_and_earlier_bytes_are_kept() { + /// Runs `cmd` through `exec_cmd` with its output in a temp dir, giving up + /// after `within`. + fn run_within( + cmd: &str, + timeout_secs: Option, + within: std::time::Duration, + ) -> Option<(ExecOutcome, String)> { let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("t1.out"); - let reader = FailAfter { - data: b"before", - sent: false, - }; - let res = block_on(super::copy_stream_to_file(reader, &path)); - assert_eq!(res.unwrap_err().to_string(), "pipe broke"); - assert_eq!(std::fs::read(&path).unwrap(), b"before"); + let (out, err) = (dir.path().join("t.out"), dir.path().join("t.err")); + let (cancel, force) = (Notify::new(), AtomicBool::new(false)); + let outcome = block_on(async { + tokio::time::timeout( + within, + exec_cmd(cmd, &out, &err, timeout_secs, &cancel, &force, None), + ) + .await + }) + .ok()? + .unwrap(); + Some((outcome, std::fs::read_to_string(&out).unwrap())) } #[test] - fn successful_copy_writes_everything() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("t1.out"); - let res = block_on(super::copy_stream_to_file(&b"all of it"[..], &path)); - assert!(res.is_ok()); - assert_eq!(std::fs::read(&path).unwrap(), b"all of it"); + fn output_lands_in_the_files() { + let (outcome, out) = run_within("echo hi", None, Duration::from_secs(10)).unwrap(); + assert!(matches!(outcome, ExecOutcome::Exited(0))); + assert_eq!(out, "hi\n"); + } + + #[test] + fn a_backgrounded_child_does_not_hold_the_task_open() { + let (outcome, out) = run_within("sleep 30 & echo started", None, Duration::from_secs(5)) + .expect("the task must finish when bash does, not when `sleep` does"); + assert!(matches!(outcome, ExecOutcome::Exited(0))); + assert_eq!(out, "started\n"); + } + + #[test] + fn the_timeout_still_applies_with_a_backgrounded_child() { + let (outcome, _) = run_within("sleep 30 &\nsleep 30", Some(1), Duration::from_secs(5)) + .expect("the timeout must end the task"); + assert!(matches!(outcome, ExecOutcome::TimedOut)); } }