diff --git a/hive-bash-mcp/src/bin/mcp.rs b/hive-bash-mcp/src/bin/mcp.rs index 10ba08b8..f31ee081 100644 --- a/hive-bash-mcp/src/bin/mcp.rs +++ b/hive-bash-mcp/src/bin/mcp.rs @@ -145,26 +145,6 @@ fn render_bash_status(id: &str, resp: Result, waited: bool) -> S } } -/// Turn a `DaemonResponse` from a `BashKill` call into the string claude sees. -fn render_bash_kill(resp: Result) -> String { - match resp { - Ok(DaemonResponse::Ok { payload }) => { - let id = payload["id"].as_str().unwrap_or("unknown"); - if payload["was_running"].as_bool().unwrap_or(false) { - let sig = payload["signal"].as_str().unwrap_or("SIGINT"); - format!( - "task `{id}`: {sig} sent to its process group; it transitions to `killed` \ - once the process exits and the usual completion wake fires." - ) - } else { - format!("task `{id}` was pending — cancelled before it started.") - } - } - Ok(DaemonResponse::Error { message }) => format!("kill error: {message}"), - Err(e) => format!("bash bridge error: {e:#}"), - } -} - // --------------------------------------------------------------------------- // MCP server // --------------------------------------------------------------------------- @@ -216,20 +196,6 @@ struct BashStatusArgs { wait_seconds: Option, } -#[derive(Debug, Deserialize, JsonSchema)] -struct BashKillArgs { - /// Task ID (from `run`) to kill. - id: String, - /// `false` (default): SIGINT to the task's process group — graceful, the - /// process can clean up. `true`: SIGKILL. Either way the whole process - /// group is signalled, so children of the shell (e.g. a `cargo`/`nix` - /// invocation) are stopped too, not just the shell. Fire-and-forget: the - /// call sends the signal and returns without waiting. If a SIGINT'd task - /// doesn't exit, call `kill` again with `force: true` to send SIGKILL. - #[serde(default)] - force: bool, -} - #[derive(Clone)] struct BashMcp; @@ -272,8 +238,8 @@ impl BashMcp { #[tool( description = "Check the status of a background bash task by its ID (from `run`). \ - Returns the current status (pending/running/done/timed_out/interrupted/killed), exit \ - code if finished, and a tail of stdout/stderr. Full output lives in \ + Returns the current status (pending/running/done/timed_out/interrupted), exit code \ + if finished, and a tail of stdout/stderr. Full output lives in \ `harness/bash-tasks/.out` / `.err`. \ Pass `wait_seconds` (capped at 30) to wait inline for the task to finish: when the \ task finishes within the window the full status is returned immediately. Useful to \ @@ -288,24 +254,6 @@ impl BashMcp { }; render_bash_status(&id, round_trip(req).await, waited) } - - #[tool( - description = "Kill a background bash task you started (by its ID from `run`). \ - `force: false` (default) sends SIGINT — graceful, lets the process clean up; \ - `force: true` sends SIGKILL. Either way the task's whole process group is \ - signalled, so a runaway child (cargo/nix/etc.) is stopped too, not just the shell. \ - Fire-and-forget: sends the signal and returns without waiting. If a SIGINT'd task \ - doesn't exit, call kill again with `force: true` to SIGKILL. A still-pending task \ - is cancelled before it starts. The task ends as `killed` and fires the usual \ - completion wake." - )] - async fn kill(&self, Parameters(args): Parameters) -> String { - let req = DaemonRequest::BashKill { - id: args.id, - force: args.force, - }; - render_bash_kill(round_trip(req).await) - } } #[tool_handler] diff --git a/hive-bash-mcp/src/protocol.rs b/hive-bash-mcp/src/protocol.rs index 4638a409..e1e97b9c 100644 --- a/hive-bash-mcp/src/protocol.rs +++ b/hive-bash-mcp/src/protocol.rs @@ -18,10 +18,6 @@ pub enum TaskStatus { TimedOut, /// Daemon was restarted while the task was running; process is gone. Interrupted, - /// Killed on request via `BashKill` (SIGINT or SIGKILL to the task's - /// process group). Distinct from `Interrupted` (daemon-restart) and - /// `TimedOut` (exceeded `timeout_secs`). - Killed, } /// Task metadata + result written to `.json` under the tasks dir. @@ -97,17 +93,6 @@ pub enum DaemonRequest { /// Used by the harness `get_loose_ends` to surface active /// background work. ActiveTasks, - - /// Kill a running (or still-pending) task. `force = false` sends - /// `SIGINT` to the task's process group (graceful — the process can - /// clean up); `force = true` sends `SIGKILL` (immediate). Signalling - /// the whole process group reaps `sh -c` plus any children it spawned, - /// so a runaway grandchild (e.g. `cargo`/`nix`) is actually stopped. - BashKill { - id: String, - #[serde(default)] - force: bool, - }, } /// Response shape from the daemon. `Ok` carries a JSON payload; `Error` diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 058b9932..a4fd2719 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -14,15 +14,14 @@ //! The runner kills the child process on timeout — `tokio::process::Child::drop()` //! does not kill children, so we explicitly call `child.kill().await`. -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex, OnceLock}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Result, bail}; use tokio::io::AsyncWriteExt as _; -use tokio::sync::Notify; use crate::paths; use crate::protocol::{TaskFile, TaskStatus}; @@ -42,50 +41,6 @@ const POLL_MS: u64 = 100; static TASK_SEQ: AtomicU64 = AtomicU64::new(0); -// --------------------------------------------------------------------------- -// Running-task registry (for on-demand kill) -// --------------------------------------------------------------------------- - -/// Handle to a currently-executing task, used by [`kill_task`] to signal it. -/// `exec_cmd` selects on `cancel`; when notified it signals the task's -/// process group with `SIGKILL` (if `force`) or `SIGINT`. -struct RunningHandle { - cancel: Arc, - force: Arc, -} - -/// Global registry of running tasks (id → handle). Populated by `run_task` -/// for the duration of execution and read by [`kill_task`] from the socket -/// dispatch path — a separate async context from the runner loop, so a -/// shared global (rather than the loop-local `claimed` set) is needed. -fn running() -> &'static Mutex> { - static RUNNING: OnceLock>> = OnceLock::new(); - RUNNING.get_or_init(|| Mutex::new(HashMap::new())) -} - -/// Outcome of one `exec_cmd` run. -enum ExecOutcome { - /// Process exited on its own with this status code. - Exited(i32), - /// Killed by the per-task timeout (`timeout_secs`). - TimedOut, - /// Killed on request via [`kill_task`]. `forced` = SIGKILL vs SIGINT. - Killed { forced: bool }, -} - -/// Send `sig` to the task's process group. A negative pgid targets the whole -/// group, so `bash -c` plus any children it spawned all receive the signal. -/// No-op if the child had no pid (already reaped). -fn signal_group(pgid: Option, sig: i32) { - if let Some(pgid) = pgid { - // SAFETY: plain libc `kill(2)`; `-pgid` targets the child's own - // process group (it called `setpgid(0, 0)` so it leads its group). - unsafe { - libc::kill(-pgid, sig); - } - } -} - // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -277,10 +232,7 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { Some(task) => { if matches!( task.status, - TaskStatus::Done - | TaskStatus::TimedOut - | TaskStatus::Interrupted - | TaskStatus::Killed + TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted ) { return Some(task); } @@ -294,52 +246,6 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { read_task(id) } -/// Kill a running or still-pending task. -/// -/// Fire-and-forget: this **sends a signal and returns** — it does not wait -/// for the process to actually exit. -/// -/// - **Running** (in the registry): signals its process group via the -/// `exec_cmd` cancel path — `SIGKILL` if `force`, else `SIGINT`. If the -/// process ignores `SIGINT`, the caller re-invokes with `force` to send -/// `SIGKILL` (no daemon-side auto-escalation). The task transitions to -/// [`TaskStatus::Killed`] once the process exits and a completion wake fires. -/// - **Pending** (queued, not yet started): marked `Killed` directly so the -/// runner loop never starts it. No process exists yet, so `force` is moot. -/// - **Terminal or unknown id**: no-op. -/// -/// Returns `(killed, was_running)`: `killed` = a kill was issued (signal sent -/// or pending task cancelled); `was_running` = the task was actively running. -/// -/// # Panics -/// -/// Panics if the running-task registry mutex is poisoned (a prior holder -/// panicked) — unrecoverable, consistent with the rest of the daemon's -/// `Mutex` usage. -#[must_use] -pub fn kill_task(id: &str, force: bool) -> (bool, bool) { - // Running task: signal via the registry. - { - let guard = running().lock().unwrap(); - if let Some(handle) = guard.get(id) { - handle.force.store(force, Ordering::SeqCst); - handle.cancel.notify_one(); - return (true, true); - } - } - // Pending task: mark Killed so the poll loop never starts it. - if let Some(mut task) = read_task(id) - && task.status == TaskStatus::Pending - { - task.status = TaskStatus::Killed; - task.completed_at = Some(now_unix()); - let _ = write_task(&task); - refresh_loose_ends(); - return (true, false); - } - (false, false) -} - // --------------------------------------------------------------------------- // Runner background loop // --------------------------------------------------------------------------- @@ -452,57 +358,27 @@ async fn run_task(mut task: TaskFile, socket: &Path) { let out_path = paths::task_out(&id); let err_path = paths::task_err(&id); - // Register a kill handle for the duration of execution so `kill_task` - // (called from the socket dispatch path) can signal this task. - let cancel = Arc::new(Notify::new()); - let force = Arc::new(AtomicBool::new(false)); - running().lock().unwrap().insert( - id.clone(), - RunningHandle { - cancel: cancel.clone(), - force: force.clone(), - }, - ); - - let outcome = exec_cmd( - &task.cmd, - &out_path, - &err_path, - task.timeout_secs, - &cancel, - &force, - ) - .await; - - // Deregister before writing terminal state — the task is no longer killable. - running().lock().unwrap().remove(&id); + let (timed_out, exit_code) = + match exec_cmd(&task.cmd, &out_path, &err_path, task.timeout_secs).await { + Ok((code, false)) => (false, Some(code)), + Ok((_, true)) => { + tracing::warn!(id = %id, "bash_runner: task timed out"); + (true, None) + } + Err(e) => { + tracing::warn!(id = %id, error = ?e, "bash_runner: exec error"); + (false, None) + } + }; let stdout_tail = tail_file(&out_path, SUMMARY_BYTES); let stderr_tail = tail_file(&err_path, SUMMARY_BYTES); - let (status, exit_code, summary) = match outcome { - Ok(ExecOutcome::Exited(code)) => (TaskStatus::Done, Some(code), format!("exit={code}")), - Ok(ExecOutcome::TimedOut) => { - tracing::warn!(id = %id, "bash_runner: task timed out"); - let secs = task.timeout_secs.unwrap_or(0); - ( - TaskStatus::TimedOut, - None, - format!("timed out after {secs}s"), - ) - } - Ok(ExecOutcome::Killed { forced }) => { - let sig = if forced { "SIGKILL" } else { "SIGINT" }; - tracing::warn!(id = %id, sig, "bash_runner: task killed"); - (TaskStatus::Killed, None, format!("killed ({sig})")) - } - Err(e) => { - tracing::warn!(id = %id, error = ?e, "bash_runner: exec error"); - (TaskStatus::Done, None, "exec error".to_owned()) - } + task.status = if timed_out { + TaskStatus::TimedOut + } else { + TaskStatus::Done }; - - task.status = status; task.completed_at = Some(now_unix()); task.exit_code = exit_code; task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); @@ -513,31 +389,33 @@ async fn run_task(mut task: TaskFile, socket: &Path) { } refresh_loose_ends(); + let summary = if timed_out { + let secs = task.timeout_secs.unwrap_or(0); + format!("timed out after {secs}s") + } else { + format!("exit={}", exit_code.unwrap_or(-1)) + }; let out_snippet = stdout_tail.as_deref().unwrap_or("").trim(); let err_snippet = stderr_tail.as_deref().unwrap_or("").trim(); send_wake(socket, &id, &summary, Some((out_snippet, err_snippet))).await; } -/// Run `bash -c cmd` in its own process group, streaming output to files. +/// Run `bash -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`. /// `timeout_secs = None` means no timeout — run until natural exit. /// -/// Tasks run under `bash`, not `sh` (on NixOS `/bin/sh` is bash in POSIX -/// mode, which disables bashisms — arrays, `[[ … ]]`, `local`, process -/// substitution), invoked via `/usr/bin/env bash`. -/// -/// The task is cancellable via `cancel` (set `force` first): on cancel the -/// whole process group is signalled — `SIGKILL` if `force`, else `SIGINT` — -/// so children of the shell die too, not just the shell. The timeout path -/// likewise `SIGKILL`s the group. +/// Tasks run under `bash`, not `sh`: on NixOS `/bin/sh` is bash in POSIX +/// mode, which disables bashisms (arrays, `[[ … ]]`, `local`, process +/// substitution, …). Agents write bash, so we invoke `bash` (via +/// `/usr/bin/env bash`). async fn exec_cmd( cmd: &str, out_path: &Path, err_path: &Path, timeout_secs: Option, - cancel: &Notify, - force: &AtomicBool, -) -> Result { +) -> Result<(i32, bool)> { use tokio::process::Command; + // SAFETY: `nice` is async-signal-safe and modifies only the calling + // process's scheduling priority before exec. No allocations, no locks. // `/usr/bin/env bash` rather than a bare `bash`: `/usr/bin/env` is at a // fixed absolute path (coreutils, present on NixOS), and it resolves // `bash` via PATH — the same controlled PATH the daemon's systemd unit @@ -549,20 +427,13 @@ async fn exec_cmd( .arg(cmd) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()); - // SAFETY: `setpgid`/`nice` are async-signal-safe and only touch the - // calling (child) process before exec — no allocations, no locks. - // `setpgid(0, 0)` makes the child its own process-group leader so the - // whole subtree can be signalled as a group (`kill(-pgid, …)`). unsafe { cmd_builder.pre_exec(|| { - libc::setpgid(0, 0); libc::nice(10); Ok(()) }); } let mut child = cmd_builder.spawn()?; - // 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"); @@ -579,60 +450,33 @@ async fn exec_cmd( err_path, )); - // A future that fires when the timeout elapses, or never if `None`. - let timeout_fut = async { - match timeout_secs { - Some(secs) => tokio::time::sleep(Duration::from_secs(secs)).await, - None => std::future::pending::<()>().await, - } - }; - - let mut timed_out = false; - let mut killed: Option = None; - let wait_res = { - let wait = child.wait(); - tokio::pin!(wait); - tokio::pin!(timeout_fut); - loop { - tokio::select! { - res = &mut wait => break res, - () = &mut timeout_fut => { - // Per-task timeout: hard-kill the whole group and reap. - signal_group(pgid, libc::SIGKILL); - timed_out = true; - break (&mut wait).await; - } - () = cancel.notified() => { - // A kill was requested: send the signal and keep waiting. - // `kill` is fire-and-forget — we do NOT block-for-exit or - // auto-escalate. If the process ignores SIGINT the caller - // re-invokes kill with `force`, which re-fires this arm - // with SIGKILL. `killed` tracks the last signal sent. - let forced = force.load(Ordering::SeqCst); - signal_group(pgid, if forced { libc::SIGKILL } else { libc::SIGINT }); - killed = Some(forced); - } + if let Some(secs) = timeout_secs { + match tokio::time::timeout(Duration::from_secs(secs), child.wait()).await { + Ok(Ok(status)) => { + let _ = copy_out.await; + let _ = copy_err.await; + Ok((status.code().unwrap_or(-1), false)) + } + Ok(Err(e)) => Err(e.into()), + Err(_elapsed) => { + let _ = child.kill().await; + let _ = child.wait().await; + let _ = copy_out.await; + let _ = copy_err.await; + Ok((-1, true)) } } - }; - - let _ = copy_out.await; - let _ = copy_err.await; - - let status = wait_res?; - let outcome = if let Some(forced) = killed { - ExecOutcome::Killed { forced } - } else if timed_out { - ExecOutcome::TimedOut } else { - ExecOutcome::Exited(status.code().unwrap_or(-1)) - }; - Ok(outcome) -} - -/// `u32` pid → `i32` for `kill(2)`. Pids fit in `i32`; saturates defensively. -fn u32_to_i32(v: u32) -> i32 { - i32::try_from(v).unwrap_or(i32::MAX) + // No timeout — wait for natural exit. + match child.wait().await { + Ok(status) => { + let _ = copy_out.await; + let _ = copy_err.await; + Ok((status.code().unwrap_or(-1), false)) + } + Err(e) => Err(e.into()), + } + } } async fn copy_stream_to_file(mut reader: R, path: PathBuf) diff --git a/hive-bash-mcp/src/socket.rs b/hive-bash-mcp/src/socket.rs index cad8f672..d6c19dd6 100644 --- a/hive-bash-mcp/src/socket.rs +++ b/hive-bash-mcp/src/socket.rs @@ -74,10 +74,7 @@ async fn dispatch(req: DaemonRequest) -> DaemonResponse { && let Some(task) = runner::wait_for_task(&id, wait).await && matches!( task.status, - TaskStatus::Done - | TaskStatus::TimedOut - | TaskStatus::Interrupted - | TaskStatus::Killed + TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted ) { return DaemonResponse::ok(&serde_json::json!({ @@ -106,30 +103,5 @@ async fn dispatch(req: DaemonRequest) -> DaemonResponse { let tasks = runner::active_tasks(); DaemonResponse::ok(&tasks) } - - DaemonRequest::BashKill { id, force } => { - let (killed, was_running) = runner::kill_task(&id, force); - if !killed { - return DaemonResponse::error(format!( - "no running or pending task with id `{id}` (already finished or unknown)" - )); - } - let payload = if was_running { - serde_json::json!({ - "id": id, - "killed": true, - "was_running": true, - "signal": if force { "SIGKILL" } else { "SIGINT" }, - }) - } else { - serde_json::json!({ - "id": id, - "killed": true, - "was_running": false, - "note": "task was pending — cancelled before it started", - }) - }; - DaemonResponse::ok(&payload) - } } }