diff --git a/Cargo.lock b/Cargo.lock index 3468417e..9bf9e23f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1241,21 +1241,6 @@ dependencies = [ "tracing-subscriber", ] -[[package]] -name = "hive-bash-mcp" -version = "0.1.0" -dependencies = [ - "anyhow", - "hive-sh4re", - "rmcp", - "schemars", - "serde", - "serde_json", - "tokio", - "tracing", - "tracing-subscriber", -] - [[package]] name = "hive-c0re" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index cae6b9d7..33f9bb67 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,6 @@ resolver = "3" members = [ "hive-ag3nt", - "hive-bash-mcp", "hive-c0re", "hive-forge", "hive-matrix-mcp", diff --git a/hive-bash-mcp/src/runner.rs b/hive-ag3nt/src/bash_runner.rs similarity index 55% rename from hive-bash-mcp/src/runner.rs rename to hive-ag3nt/src/bash_runner.rs index f6a8ee14..7d8d3526 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-ag3nt/src/bash_runner.rs @@ -1,63 +1,50 @@ -//! Bash subprocess runner: spawns `sh -c ` tasks, writes status -//! files under `harness/bash-tasks/`, and fires hyperhive wake signals -//! on completion. Mirrors the logic previously embedded in `hive-ag3nt`. +//! Harness-internal async bash task runner for `get_loose_ends`-compatible +//! background execution. An MCP tool call writes a task request file to +//! `harness_dir()/bash-tasks/.json`; this background loop picks it up, +//! runs `sh -c ` with a timeout, writes stdout/stderr to sibling files, +//! and fires a `Wake` via the broker socket on completion. The agent's next +//! turn finds the result via `bash_status()`. //! -//! Files under `tasks_dir()`: +//! Files under `harness_dir()/bash-tasks/`: //! - `.json` — task metadata + status (pending → running → done) -//! - `.out` — captured stdout (streamed while running) -//! - `.err` — captured stderr (streamed while running) +//! - `.out` — captured stdout (appended while running) +//! - `.err` — captured stderr (appended while running) //! -//! Tasks with status `running` on daemon boot are marked `interrupted` -//! (the process died with the previous daemon). A best-effort wake is -//! still sent so the agent is not silently blocked. +//! Tasks with status `running` on harness boot are marked `interrupted` +//! (the process died with the previous harness). Best-effort wake is still +//! sent so the agent is not silently blocked waiting forever. //! //! The runner kills the child process on timeout — `tokio::process::Child::drop()` -//! does not kill children, so we explicitly call `child.kill().await`. +//! does not kill children, so we explicitly call `child.kill().await` before +//! collecting the copy tasks. use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use anyhow::Result; -use tokio::io::AsyncWriteExt as _; +use serde::{Deserialize, Serialize}; +use tokio::io::AsyncWriteExt; -use crate::paths; -use crate::protocol::{TaskFile, TaskStatus}; - -/// Poll interval for the runner background loop. const POLL_INTERVAL: Duration = Duration::from_millis(200); - /// Soft cap on stdout/stderr captured in the done JSON summary. -/// Full output always lives in the `.out`/`.err` files. +/// Full output always lives in the .out/.err files. Exposed so +/// `format_bash_status` can detect truncation and surface the full-output +/// file path when the on-disk file is larger. pub const SUMMARY_BYTES: usize = 4096; - -/// Default task timeout. +/// Default timeout for tasks that don't specify one. pub const DEFAULT_TIMEOUT_SECS: u64 = 180; -/// Maximum inline wait (cap on `wait_seconds`). -pub const MAX_WAIT_SECS: u64 = 30; - -/// Poll interval used by the inline-wait loops. -const POLL_MS: u64 = 100; - static TASK_SEQ: AtomicU64 = AtomicU64::new(0); -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -fn now_unix() -> i64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i64 -} - -/// Generate a task ID: ``. +/// Generate a task ID: `` — unique within a +/// harness session; collision chance across sessions negligible for our +/// volume. #[must_use] pub fn new_task_id() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; let t = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() @@ -66,13 +53,107 @@ pub fn new_task_id() -> String { format!("{t:013x}{seq:04x}") } +/// Base directory for task files. Uses `HYPERHIVE_HARNESS_DIR` if set +/// (injected by hive-c0re meta flake after the harness/state split); +/// falls back to a sibling of `state_dir()` for pre-split deployments. +fn tasks_dir() -> PathBuf { + let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { + PathBuf::from(p) + } else { + // Pre-split fallback: derive harness/ as a sibling of state/. + let state = crate::paths::state_dir(); + state.parent().map(|p| p.join("harness")).unwrap_or(state) + }; + base.join("bash-tasks") +} + +fn task_json(id: &str) -> PathBuf { + tasks_dir().join(format!("{id}.json")) +} + +fn task_out(id: &str) -> PathBuf { + tasks_dir().join(format!("{id}.out")) +} + +fn task_err(id: &str) -> PathBuf { + tasks_dir().join(format!("{id}.err")) +} + +/// Path to the full stdout capture file for `id`. +/// Exposed so `format_bash_status` can surface it when output is truncated. +#[must_use] +pub fn task_out_path(id: &str) -> PathBuf { + task_out(id) +} + +/// Path to the full stderr capture file for `id`. +/// Exposed so `format_bash_status` can surface it when output is truncated. +#[must_use] +pub fn task_err_path(id: &str) -> PathBuf { + task_err(id) +} + +// --------------------------------------------------------------------------- +// Wire types (shared between MCP tool writers and runner readers) +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskStatus { + Pending, + Running, + Done, + TimedOut, + /// Harness was restarted while the task was running; process is gone. + Interrupted, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskFile { + pub id: String, + pub cmd: String, + pub timeout_secs: u64, + pub status: TaskStatus, + pub created_at: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub completed_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + /// Last `SUMMARY_BYTES` of stdout (full output in `.out`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stdout_tail: Option, + /// Last `SUMMARY_BYTES` of stderr (full output in `.err`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stderr_tail: Option, +} + +impl TaskFile { + #[must_use] + pub fn new(id: String, cmd: String, timeout_secs: u64) -> Self { + Self { + id, + cmd, + timeout_secs, + status: TaskStatus::Pending, + created_at: crate::serve_common::now_unix(), + started_at: None, + completed_at: None, + exit_code: None, + stdout_tail: None, + stderr_tail: None, + } + } +} + /// Write a task file atomically (tmp + rename). fn write_task(task: &TaskFile) -> std::io::Result<()> { let json = serde_json::to_string_pretty(task) .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; - let dest = paths::task_json(&task.id); + let dest = task_json(&task.id); let tmp = dest.with_extension("json.tmp"); - std::fs::write(&tmp, &json)?; + std::fs::write(&tmp, json)?; std::fs::rename(&tmp, &dest) } @@ -80,80 +161,21 @@ fn write_task(task: &TaskFile) -> std::io::Result<()> { /// unparseable. #[must_use] pub fn read_task(id: &str) -> Option { - let s = std::fs::read_to_string(paths::task_json(id)).ok()?; + let s = std::fs::read_to_string(task_json(id)).ok()?; serde_json::from_str(&s).ok() } // --------------------------------------------------------------------------- -// Loose-ends file (generic MCP loose-ends protocol) +// Public API used by MCP tools // --------------------------------------------------------------------------- -/// Rewrite `mcp-loose-ends/bash.json` with a summary of all currently -/// active (Pending or Running) tasks. The harness scans this directory -/// generically in `get_loose_ends` — no bash-specific code needed there. -/// -/// File format: a JSON array of plain-text summary strings, one per -/// loose-end item. The harness includes them verbatim in the output. -/// Atomic write (tmp + rename) so the harness never reads a partial file. -fn refresh_loose_ends() { - let active = active_tasks(); - let dir = paths::mcp_loose_ends_dir(); - if let Err(e) = std::fs::create_dir_all(&dir) { - tracing::warn!(error = ?e, "bash_runner: create mcp-loose-ends dir failed"); - return; - } - let items: Vec = active - .iter() - .map(|t| { - let age = now_unix() - t.created_at; - format!( - "bash task `{}` status={:?}, cmd: `{}`, age {}s", - t.id, t.status, t.cmd, age - ) - }) - .collect(); - let dest = dir.join("bash.json"); - let tmp = dest.with_extension("json.tmp"); - let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned()); - if let Err(e) = std::fs::write(&tmp, &json).and_then(|_| std::fs::rename(&tmp, &dest)) { - tracing::warn!(error = ?e, "bash_runner: write mcp-loose-ends/bash.json failed"); - } -} - -// --------------------------------------------------------------------------- -// Public API used by daemon dispatch -// --------------------------------------------------------------------------- - -/// Submit a new pending task. Returns the task ID. -/// -/// # Errors -/// -/// Returns an error if the tasks directory cannot be created or the -/// task file cannot be written. -pub fn submit_task(cmd: String, timeout_secs: Option) -> Result { - std::fs::create_dir_all(paths::tasks_dir())?; - let id = new_task_id(); - let task = TaskFile { - id: id.clone(), - cmd, - timeout_secs: timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS), - status: TaskStatus::Pending, - created_at: now_unix(), - started_at: None, - completed_at: None, - exit_code: None, - stdout_tail: None, - stderr_tail: None, - }; - write_task(&task)?; - refresh_loose_ends(); - Ok(id) -} - -/// Return all tasks currently in `Pending` or `Running` state. +/// Return all tasks in `Pending` or `Running` state. Used by the +/// `get_loose_ends` tool to surface active background work alongside +/// broker-side items (questions, reminders). Silently skips unreadable +/// or unparseable files. #[must_use] pub fn active_tasks() -> Vec { - let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return Vec::new(); }; let mut out = Vec::new(); @@ -173,53 +195,49 @@ pub fn active_tasks() -> Vec { out } -/// Inline wait: poll `read_task(id)` until terminal state or deadline. -/// Returns the final task on success, or `None` if it never completed. -pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { - let cap = wait_secs.min(MAX_WAIT_SECS); - if cap == 0 { - return read_task(id); - } - let deadline = tokio::time::Instant::now() + Duration::from_secs(cap); - loop { - match read_task(id) { - None => break, - Some(task) => { - if matches!( - task.status, - TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted - ) { - return Some(task); - } - } - } - if tokio::time::Instant::now() >= deadline { - break; - } - tokio::time::sleep(Duration::from_millis(POLL_MS)).await; - } - read_task(id) +/// Create a new pending task and write it to disk. Returns the task ID +/// the MCP tool should return to claude. The runner will pick it up +/// within the next poll interval (~200ms). +/// +/// # Errors +/// +/// Returns an error if the tasks directory cannot be created or the task +/// file cannot be written. +pub fn submit_task(cmd: String, timeout_secs: Option) -> Result { + std::fs::create_dir_all(tasks_dir())?; + let id = new_task_id(); + let task = TaskFile::new( + id.clone(), + cmd, + timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS), + ); + write_task(&task)?; + Ok(id) } // --------------------------------------------------------------------------- // Runner background loop // --------------------------------------------------------------------------- -/// Spawn the background runner loop as a detached tokio task. `socket` -/// is the path to the per-agent broker socket used to deliver completion -/// wake signals. Call once at daemon startup. -pub fn spawn_loop(socket: PathBuf) { +/// Spawn the background runner loop as a detached tokio task. `socket` is +/// the path to the per-agent broker socket, used to deliver the completion +/// `Wake`. Call once at harness startup from `serve_main`. +pub fn spawn(socket: PathBuf) { tokio::spawn(async move { run_loop(socket).await; }); } async fn run_loop(socket: PathBuf) { - if let Err(e) = std::fs::create_dir_all(paths::tasks_dir()) { + if let Err(e) = std::fs::create_dir_all(tasks_dir()) { tracing::warn!(error = ?e, "bash_runner: create tasks dir failed"); } + // Mark any tasks left in "running" state from a previous harness + // session as interrupted so agents waiting on them get unblocked. mark_interrupted(&socket).await; + // In-memory set of task IDs we have already claimed this session + // so we don't re-spawn on each poll iteration. let claimed: Arc>> = Arc::new(Mutex::new(HashSet::new())); loop { @@ -228,9 +246,10 @@ async fn run_loop(socket: PathBuf) { } } -/// On boot, flip any `running` tasks to `interrupted` and fire a wake. +/// On boot, find any task files in `running` state and flip them to +/// `interrupted`, then fire a wake so the agent unblocks. async fn mark_interrupted(socket: &Path) { - let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return; }; for entry in rd.flatten() { @@ -249,17 +268,16 @@ async fn mark_interrupted(socket: &Path) { } tracing::warn!(id = %id, "bash_runner: marking interrupted task"); task.status = TaskStatus::Interrupted; - task.completed_at = Some(now_unix()); + task.completed_at = Some(crate::serve_common::now_unix()); if let Err(e) = write_task(&task) { - tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed"); + tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted task failed"); } - refresh_loose_ends(); - send_wake(socket, &id, "interrupted (daemon restarted)", None).await; + send_wake(socket, &id, "interrupted (harness restarted)", None).await; } } async fn poll_once(socket: &Path, claimed: &Arc>>) { - let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(tasks_dir()) else { return; }; for entry in rd.flatten() { @@ -286,6 +304,8 @@ async fn poll_once(socket: &Path, claimed: &Arc>>) { let claimed = claimed.clone(); tokio::spawn(async move { run_task(task, &socket).await; + // Remove from claimed so a resubmitted ID (rare) could be + // picked up again. In practice each ID is unique. claimed.lock().unwrap().remove(&id); }); } @@ -300,14 +320,13 @@ async fn run_task(mut task: TaskFile, socket: &Path) { tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task"); task.status = TaskStatus::Running; - task.started_at = Some(now_unix()); + task.started_at = Some(crate::serve_common::now_unix()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed"); } - refresh_loose_ends(); - let out_path = paths::task_out(&id); - let err_path = paths::task_err(&id); + let out_path = task_out(&id); + let err_path = task_err(&id); let timeout = Duration::from_secs(task.timeout_secs); let (timed_out, exit_code) = match exec_cmd(&task.cmd, &out_path, &err_path, timeout).await { @@ -330,7 +349,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { } else { TaskStatus::Done }; - task.completed_at = Some(now_unix()); + task.completed_at = Some(crate::serve_common::now_unix()); task.exit_code = exit_code; task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); task.stderr_tail = stderr_tail.clone().filter(|s| !s.is_empty()); @@ -338,19 +357,19 @@ async fn run_task(mut task: TaskFile, socket: &Path) { if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed"); } - refresh_loose_ends(); let summary = if timed_out { format!("timed out after {}s", task.timeout_secs) } else { format!("exit={}", exit_code.unwrap_or(-1)) }; - let out_snippet = stdout_tail.as_deref().unwrap_or("").trim(); + let output_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; + send_wake(socket, &id, &summary, Some((output_snippet, err_snippet))).await; } /// Run `sh -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`. +/// On timeout the child process is explicitly killed before returning. async fn exec_cmd( cmd: &str, out_path: &Path, @@ -371,27 +390,29 @@ async fn exec_cmd( let out_path = out_path.to_path_buf(); let err_path = err_path.to_path_buf(); - let copy_out = tokio::spawn(copy_stream_to_file( + // Stream stdout and stderr to files concurrently. + let copy_stdout = tokio::spawn(copy_stream_to_file( tokio::io::BufReader::new(stdout), out_path, )); - let copy_err = tokio::spawn(copy_stream_to_file( + let copy_stderr = tokio::spawn(copy_stream_to_file( tokio::io::BufReader::new(stderr), err_path, )); match tokio::time::timeout(timeout, child.wait()).await { Ok(Ok(status)) => { - let _ = copy_out.await; - let _ = copy_err.await; + let _ = copy_stdout.await; + let _ = copy_stderr.await; Ok((status.code().unwrap_or(-1), false)) } Ok(Err(e)) => Err(e.into()), Err(_elapsed) => { + // Kill the child so it doesn't linger after timeout. let _ = child.kill().await; - let _ = child.wait().await; - let _ = copy_out.await; - let _ = copy_err.await; + let _ = child.wait().await; // reap zombie + let _ = copy_stdout.await; + let _ = copy_stderr.await; Ok((-1, true)) } } @@ -407,11 +428,12 @@ where let _ = f.flush().await; } Err(e) => { - tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed"); + tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed") } } } +/// Read the last `max_bytes` of a file as a UTF-8 string. fn tail_file(path: &Path, max_bytes: usize) -> Option { let data = std::fs::read(path).ok()?; let slice = if data.len() > max_bytes { @@ -426,12 +448,7 @@ fn tail_file(path: &Path, max_bytes: usize) -> Option { // Wake delivery // --------------------------------------------------------------------------- -pub(crate) async fn send_wake( - socket: &Path, - id: &str, - summary: &str, - output: Option<(&str, &str)>, -) { +async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str, &str)>) { let mut body = format!("bash task `{id}` finished: {summary}"); if let Some((stdout, stderr)) = output { if !stdout.is_empty() { @@ -448,37 +465,14 @@ pub(crate) async fn send_wake( let req = hive_sh4re::AgentRequest::Wake { from: format!("bash-task-{id}"), body, + // Transient: do not persist to the message broker. Bash task + // completions are fire-and-forget — the output is already on disk + // in harness/bash-tasks/; persisting would cause duplicate delivery + // after a harness restart. transient: true, }; - - use tokio::io::{AsyncBufReadExt as _, BufReader}; - use tokio::net::UnixStream; - - match UnixStream::connect(socket).await { - Ok(stream) => { - let (read, mut write) = stream.into_split(); - let line = match serde_json::to_string(&req) { - Ok(mut s) => { - s.push('\n'); - s - } - Err(e) => { - tracing::warn!(id = %id, error = ?e, "bash_runner: serialise wake failed"); - return; - } - }; - if write.write_all(line.as_bytes()).await.is_err() { - tracing::warn!(id = %id, "bash_runner: write wake failed"); - return; - } - let _ = write.shutdown().await; - // Drain the response so the server doesn't get ECONNRESET. - let mut resp = String::new(); - let _ = BufReader::new(read).read_line(&mut resp).await; - tracing::info!(id = %id, "bash_runner: wake delivered"); - } - Err(e) => { - tracing::warn!(id = %id, error = ?e, "bash_runner: connect wake socket failed"); - } + match crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req).await { + Ok(_) => tracing::info!(id = %id, "bash_runner: wake delivered"), + Err(e) => tracing::warn!(id = %id, error = ?e, "bash_runner: wake delivery failed"), } } diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index b880bb11..b31e5791 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -570,6 +570,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { S::send_to_parent(socket, failure).await; } tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf())); + hive_ag3nt::bash_runner::spawn(socket.to_path_buf()); // Log web_ui::serve's error instead of dropping it. A bare // `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so // any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points diff --git a/hive-ag3nt/src/lib.rs b/hive-ag3nt/src/lib.rs index 801cc003..72c94e7f 100644 --- a/hive-ag3nt/src/lib.rs +++ b/hive-ag3nt/src/lib.rs @@ -1,6 +1,7 @@ //! Shared in-container harness code used by both `hive-ag3nt` (agent) and //! `hive-m1nd` (manager) binaries. +pub mod bash_runner; pub mod client; pub mod events; pub mod forge_notify; @@ -8,7 +9,6 @@ pub mod identity; pub mod login; pub mod login_session; pub mod mcp; -pub mod mcp_loose_ends; pub mod paths; pub mod plugins; pub mod prompt; diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 540c37b9..0432b252 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -494,6 +494,96 @@ pub struct RecvArgs { pub max: Option, } +/// MCP tool args for `bash_run`. +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct BashRunArgs { + /// Shell command to run (passed to `sh -c`). + pub cmd: String, + /// Timeout in seconds. Defaults to 180. Task is killed and marked + /// `timed_out` when the limit is exceeded. + #[serde(default)] + pub timeout_secs: Option, + /// Optional inline wait: `bash_run` polls for up to `wait_seconds` + /// (capped at 30) before returning. When the task finishes within the + /// window the full status is returned immediately and no wake is fired; + /// when the timeout expires the task keeps running and the normal + /// `task started: id=` response is returned. Defaults to 3s. Pass + /// `0` to disable and always get the immediate response. + #[serde(default = "default_bash_run_wait")] + pub wait_seconds: Option, +} + +fn default_bash_run_wait() -> Option { + Some(3) +} + +/// MCP tool args for `bash_status`. +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct BashStatusArgs { + /// Task ID returned by `bash_run`. + pub id: String, + /// Optional inline wait: if set, `bash_status` polls for up to + /// `wait_seconds` (capped at 30) before returning. When the task + /// finishes within the window the full status is returned immediately. + /// Useful to avoid a separate round-trip when a task is expected to + /// finish soon. + pub wait_seconds: Option, +} + +/// Format the result of `bash_status` from a task ID. +/// +/// Includes inline output tails (up to [`crate::bash_runner::SUMMARY_BYTES`]) +/// and, when the full output file is larger than the inline tail, appends +/// the file path so the caller can read the rest with the `Read` tool. +#[must_use] +fn format_bash_status(id: &str) -> String { + use std::fmt::Write as _; + let Some(task) = crate::bash_runner::read_task(id) else { + return format!("bash_status: unknown task id `{id}`"); + }; + let mut out = format!("task `{id}`: status={status:?}", status = task.status); + if let Some(code) = task.exit_code { + let _ = write!(out, ", exit={code}"); + } + if let Some(t) = task.started_at + && task.completed_at.is_none() + { + let age = crate::serve_common::now_unix() - t; + let _ = write!(out, ", running for {age}s"); + } + if let Some(t) = task.completed_at { + if let Some(s) = task.started_at { + let _ = write!(out, ", took {}s", t - s); + } + } + + // Inline tail for stdout. + let out_path = crate::bash_runner::task_out_path(id); + let out_file_len = std::fs::metadata(&out_path).map(|m| m.len()).unwrap_or(0); + if let Some(ref stdout) = task.stdout_tail { + if !stdout.trim().is_empty() { + let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); + } + } + if out_file_len > crate::bash_runner::SUMMARY_BYTES as u64 { + let _ = write!(out, "\n\nFull stdout lives in `{}`", out_path.display()); + } + + // Inline tail for stderr. + let err_path = crate::bash_runner::task_err_path(id); + let err_file_len = std::fs::metadata(&err_path).map(|m| m.len()).unwrap_or(0); + if let Some(ref stderr) = task.stderr_tail { + if !stderr.trim().is_empty() { + let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); + } + } + if err_file_len > crate::bash_runner::SUMMARY_BYTES as u64 { + let _ = write!(out, "\n\nFull stderr lives in `{}`", err_path.display()); + } + + out +} + /// MCP tool args for `remind`. Exactly one of `delay_seconds` or /// `at_unix_timestamp` must be set; both / neither is a tool-side error. /// Hides the tagged `ReminderTiming` enum behind a flatter schema so the @@ -684,8 +774,8 @@ impl AgentServer { description = "List loose ends pending against this agent: unanswered questions \ where you are the asker (waiting on someone) or the target (someone's waiting on \ you), pending reminders you scheduled, plus — for the manager only — pending \ - approvals you submitted that the operator hasn't acted on yet. Also lists active \ - local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \ + approvals you submitted that the operator hasn't acted on yet. Also lists any \ + local bash tasks still in pending or running state. Cheap sweep, no args. Useful \ at turn start to remember what you owe / what's owed to you without scrolling \ inbox history. Output is a short bulleted list with ids, ages in seconds, and \ the relevant context. Each `question` or `reminder` row can be cancelled by \ @@ -737,16 +827,19 @@ impl AgentServer { } } let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); - // Append loose-end items published by external MCP daemons - // (e.g. active bash tasks from hive-bash-mcp). Generic — no - // per-MCP knowledge needed here. - let mcp_items = crate::mcp_loose_ends::collect(); - if !mcp_items.is_empty() { + // Append any local bash tasks still in pending/running state so + // the agent sees all outstanding work in one call. + let active = crate::bash_runner::active_tasks(); + if !active.is_empty() { use std::fmt::Write as _; - let n = mcp_items.len(); - let _ = write!(out, "\n\n{n} local task(s):"); - for item in &mcp_items { - let _ = write!(out, "\n- {item}"); + let _ = write!(out, "\n\n{} active bash task(s):", active.len()); + for task in &active { + let age = crate::serve_common::now_unix() - task.created_at; + let _ = write!( + out, + "\n- `{}` status={:?}, cmd: `{}`, age {}s", + task.id, task.status, task.cmd, age + ); } } out @@ -871,6 +964,94 @@ impl AgentServer { .await } + #[tool( + description = "Run a shell command in the background. Returns a task ID immediately — \ + do NOT wait inline. When the command finishes, the harness fires a wake with \ + `from: \"bash-task-\"` and the exit code + last stdout lines in the body; \ + handle it on a future turn. Use `bash_status` to poll the task status within \ + the same turn if needed. `timeout_secs` defaults to 180. Pass `wait_seconds` \ + (capped at 30) to wait inline for fast commands: when the task finishes within \ + the window the full status is returned immediately and no wake is fired; when \ + the timeout expires the task keeps running and the normal `task started: id=` \ + response is returned. `wait_seconds` defaults to 3; pass `wait_seconds: 0` to \ + disable inline waiting and always get the immediate response." + )] + async fn bash_run(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("bash_run", log, async move { + let id = match crate::bash_runner::submit_task(args.cmd, args.timeout_secs) { + Ok(id) => id, + Err(e) => return format!("bash_run failed: {e:#}"), + }; + // Inline wait: poll until done or deadline, whichever comes first. + if let Some(wait) = args.wait_seconds { + const MAX_WAIT_SECS: u64 = 30; + const POLL_MS: u64 = 100; + let deadline = tokio::time::Instant::now() + + std::time::Duration::from_secs(wait.min(MAX_WAIT_SECS)); + loop { + tokio::time::sleep(std::time::Duration::from_millis(POLL_MS)).await; + if let Some(task) = crate::bash_runner::read_task(&id) { + use crate::bash_runner::TaskStatus; + if matches!( + task.status, + TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted + ) { + return format_bash_status(&id); + } + } + if tokio::time::Instant::now() >= deadline { + break; + } + } + } + format!("task started: id={id}") + }) + .await + } + + #[tool( + description = "Check the status of a background bash task by its ID (from `bash_run`). \ + 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 \ + avoid a separate round-trip when the task is expected to finish soon." + )] + async fn bash_status(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("bash_status", log, async move { + // Inline wait: if the task isn't terminal yet, poll until done or deadline. + if let Some(wait) = args.wait_seconds { + const MAX_WAIT_SECS: u64 = 30; + const POLL_MS: u64 = 100; + let deadline = tokio::time::Instant::now() + + std::time::Duration::from_secs(wait.min(MAX_WAIT_SECS)); + loop { + match crate::bash_runner::read_task(&args.id) { + None => break, // unknown ID — no point waiting + Some(task) => { + use crate::bash_runner::TaskStatus; + if matches!( + task.status, + TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted + ) { + break; + } + } + } + if tokio::time::Instant::now() >= deadline { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(POLL_MS)).await; + } + } + format_bash_status(&args.id) + }) + .await + } + #[tool( description = "Ask the harness to start another turn immediately after this one \ completes, even if the inbox is empty. Use this when you have ongoing work that \ @@ -1929,8 +2110,8 @@ pub const SERVER_NAME: &str = "hyperhive"; /// exist in the session. Web egress (`WebFetch`/`WebSearch`) are /// tool-group-gated (`web_tools`) — off by default. Nested agents /// (`Task`) are intentionally omitted. `Bash` is disallowed — shell -/// execution goes through `mcp__hive_bash__bash_run` (background tasks -/// with structured output via `hive-bash-mcp`) instead of a raw interactive shell. `TodoWrite` +/// execution goes through `mcp__hyperhive__bash_run` (background tasks +/// with structured output) instead of a raw interactive shell. `TodoWrite` /// is omitted because the todo list lives in claude's in-process session /// state and silently evaporates on /compact or session reset — agents /// should plan in /state notes instead. diff --git a/hive-ag3nt/src/mcp_loose_ends.rs b/hive-ag3nt/src/mcp_loose_ends.rs deleted file mode 100644 index 222a6392..00000000 --- a/hive-ag3nt/src/mcp_loose_ends.rs +++ /dev/null @@ -1,53 +0,0 @@ -//! Generic scanner for MCP loose-end summary files. -//! -//! External MCP daemons (hive-bash-mcp, hive-matrix-mcp, etc.) write -//! JSON files to `$HYPERHIVE_HARNESS_DIR/mcp-loose-ends/.json`. -//! Each file contains a JSON array of plain-text summary strings. -//! -//! The harness reads all files in this directory in `get_loose_ends` to -//! surface active background work from any MCP without hardcoding -//! per-MCP knowledge here. - -use std::path::PathBuf; - -/// NOTE: the base-dir resolution logic here is intentionally mirrored in -/// `hive-bash-mcp/src/paths.rs::mcp_loose_ends_dir()`. They can't share -/// code across crates — keep them in sync if the fallback logic changes. -fn loose_ends_dir() -> PathBuf { - let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { - PathBuf::from(p) - } else { - let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); - let state_path = PathBuf::from(&state); - state_path - .parent() - .map(|p| p.join("harness")) - .unwrap_or_else(|| PathBuf::from(state)) - }; - base.join("mcp-loose-ends") -} - -/// Collect all loose-end summary strings published by external MCP daemons. -/// Each string is a single line suitable for inclusion in `get_loose_ends` -/// output. Returns an empty vec if the directory doesn't exist or is empty. -#[must_use] -pub fn collect() -> Vec { - let Ok(rd) = std::fs::read_dir(loose_ends_dir()) else { - return Vec::new(); - }; - let mut out = Vec::new(); - for entry in rd.flatten() { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some("json") { - continue; - } - let Ok(content) = std::fs::read_to_string(&path) else { - continue; - }; - let Ok(items) = serde_json::from_str::>(&content) else { - continue; - }; - out.extend(items); - } - out -} diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml deleted file mode 100644 index 81699f08..00000000 --- a/hive-bash-mcp/Cargo.toml +++ /dev/null @@ -1,35 +0,0 @@ -[package] -name = "hive-bash-mcp" -edition.workspace = true -version.workspace = true - -[lints] -workspace = true - -[dependencies] -anyhow.workspace = true -hive-sh4re.workspace = true -rmcp.workspace = true -schemars.workspace = true -serde.workspace = true -serde_json.workspace = true -tokio.workspace = true -tracing.workspace = true -tracing-subscriber.workspace = true - -# `hive-bash-daemon` — long-running per-agent bash task runner. -# Spawns `sh -c` subprocesses, monitors completion, writes task state -# files under harness/bash-tasks/, and fires hyperhive wake signals on -# completion. Listens on a unix socket for tool-call requests from the -# stdio MCP bridge. -[[bin]] -name = "hive-bash-daemon" -path = "src/main.rs" - -# `hive-bash-mcp` — thin stdio MCP bridge spawned by claude per turn. -# Forwards every tool call (bash_run, bash_status) to the daemon over -# the unix socket, returns results to claude. No subprocess management -# at this entrypoint — the daemon owns that. -[[bin]] -name = "hive-bash-mcp" -path = "src/bin/mcp.rs" diff --git a/hive-bash-mcp/src/bin/mcp.rs b/hive-bash-mcp/src/bin/mcp.rs deleted file mode 100644 index 90c5a047..00000000 --- a/hive-bash-mcp/src/bin/mcp.rs +++ /dev/null @@ -1,239 +0,0 @@ -//! `hive-bash-mcp` binary — stdio MCP server claude spawns per turn. -//! Thin protocol bridge: every tool call → connect to the daemon's -//! unix socket → write a JSON request line → read the JSON response → -//! return the result to claude. -//! -//! No subprocess management at this entrypoint — the daemon owns that. -//! Cold-starts in milliseconds. - -use anyhow::{Context, Result}; -use rmcp::{ - ServerHandler, ServiceExt, - handler::server::wrapper::Parameters, - schemars::{self, JsonSchema}, - tool, tool_handler, tool_router, - transport::stdio, -}; -use serde::Deserialize; -use std::fmt::Write as _; -use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; -use tokio::net::UnixStream; - -use hive_bash_mcp::paths; -use hive_bash_mcp::protocol::{DaemonRequest, DaemonResponse}; -use hive_bash_mcp::runner::SUMMARY_BYTES; - -/// Send `req` to the daemon and read back the response. Each call is a -/// fresh unix-socket connection — short-lived (single round-trip) so -/// connection pooling is unnecessary. -async fn round_trip(req: DaemonRequest) -> Result { - let socket = paths::daemon_socket(); - let stream = UnixStream::connect(&socket) - .await - .with_context(|| format!("connect bash daemon socket {}", socket.display()))?; - let (reader, mut writer) = stream.into_split(); - let mut line = serde_json::to_string(&req)?; - line.push('\n'); - writer - .write_all(line.as_bytes()) - .await - .context("write request to bash daemon socket")?; - writer.shutdown().await.ok(); - let mut buf = String::new(); - BufReader::new(reader) - .read_line(&mut buf) - .await - .context("read response from bash daemon socket")?; - serde_json::from_str(&buf).context("parse bash daemon response") -} - -/// Format a `TaskFile` JSON value as a human-readable status string. -/// Mirrors `format_bash_status` in the old hive-ag3nt, adapted to work -/// from the daemon's JSON payload. -fn format_task(id: &str, task: &serde_json::Value) -> String { - let status = task["status"].as_str().unwrap_or("unknown"); - let mut out = format!("task `{id}`: status={status}"); - - if let Some(code) = task["exit_code"].as_i64() { - let _ = write!(out, ", exit={code}"); - } - if let (Some(started), None) = ( - task["started_at"].as_i64(), - task["completed_at"].as_i64().map(|_| ()), - ) { - // Running — show age. - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i64; - let _ = write!(out, ", running for {}s", now - started); - } - if let (Some(completed), Some(started)) = - (task["completed_at"].as_i64(), task["started_at"].as_i64()) - { - let _ = write!(out, ", took {}s", completed - started); - } - - let out_file = paths::task_out(id); - let err_file = paths::task_err(id); - let out_len = std::fs::metadata(&out_file).map(|m| m.len()).unwrap_or(0); - let err_len = std::fs::metadata(&err_file).map(|m| m.len()).unwrap_or(0); - - if let Some(stdout) = task["stdout_tail"].as_str() { - let s = stdout.trim(); - if !s.is_empty() { - let _ = write!(out, "\n\nstdout:\n```\n{s}\n```"); - } - } - if out_len > SUMMARY_BYTES as u64 { - let _ = write!(out, "\n\nFull stdout lives in `{}`", out_file.display()); - } - - if let Some(stderr) = task["stderr_tail"].as_str() { - let s = stderr.trim(); - if !s.is_empty() { - let _ = write!(out, "\n\nstderr:\n```\n{s}\n```"); - } - } - if err_len > SUMMARY_BYTES as u64 { - let _ = write!(out, "\n\nFull stderr lives in `{}`", err_file.display()); - } - - out -} - -/// Turn a `DaemonResponse` from a `BashRun` call into the string -/// claude sees as the tool result. -fn render_bash_run(id: &str, resp: Result) -> String { - match resp { - Ok(DaemonResponse::Ok { payload }) => { - let finished = payload["finished"].as_bool().unwrap_or(false); - if finished { - if let Some(task) = payload.get("task") { - return format_task(id, task); - } - } - format!("task started: id={id}") - } - Ok(DaemonResponse::Error { message }) => format!("bash_run error: {message}"), - Err(e) => format!("bash bridge error: {e:#}"), - } -} - -/// Turn a `DaemonResponse` from a `BashStatus` call into the string -/// claude sees as the tool result. -fn render_bash_status(id: &str, resp: Result) -> String { - match resp { - Ok(DaemonResponse::Ok { payload }) => format_task(id, &payload), - Ok(DaemonResponse::Error { message }) => message, - Err(e) => format!("bash bridge error: {e:#}"), - } -} - -// --------------------------------------------------------------------------- -// MCP server -// --------------------------------------------------------------------------- - -#[derive(Debug, Deserialize, JsonSchema)] -struct BashRunArgs { - /// Shell command to run (passed to `sh -c`). - cmd: String, - /// Timeout in seconds. Defaults to 180. Task is killed and marked - /// `timed_out` when the limit is exceeded. - #[serde(default)] - timeout_secs: Option, - /// Optional inline wait: `bash_run` polls for up to `wait_seconds` - /// (capped at 30) before returning. When the task finishes within the - /// window the full status is returned immediately and no wake is fired; - /// when the timeout expires the task keeps running and the normal - /// `task started: id=` response is returned. Defaults to 3s. Pass - /// `0` to disable inline waiting and always get the immediate response. - #[serde(default = "default_wait")] - wait_seconds: Option, -} - -fn default_wait() -> Option { - Some(3) -} - -#[derive(Debug, Deserialize, JsonSchema)] -struct BashStatusArgs { - /// Task ID returned by `bash_run`. - id: String, - /// Optional inline wait: `bash_status` polls for up to `wait_seconds` - /// (capped at 30) before returning. Useful to avoid a separate - /// round-trip when the task is expected to finish soon. - #[serde(default)] - wait_seconds: Option, -} - -#[derive(Clone)] -struct BashMcp; - -#[tool_router] -impl BashMcp { - #[tool( - description = "Run a shell command in the background. Returns a task ID immediately — \ - do NOT wait inline. When the command finishes, the harness fires a wake with \ - `from: \"bash-task-\"` and the exit code + last stdout lines in the body; \ - handle it on a future turn. Use `bash_status` to poll the task status within \ - the same turn if needed. `timeout_secs` defaults to 180. Pass `wait_seconds` \ - (capped at 30) to wait inline for fast commands: when the task finishes within \ - the window the full status is returned immediately and no wake is fired; when \ - the timeout expires the task keeps running and the normal `task started: id=` \ - response is returned. `wait_seconds` defaults to 3; pass `wait_seconds: 0` to \ - disable inline waiting and always get the immediate response." - )] - async fn bash_run(&self, Parameters(args): Parameters) -> String { - let req = DaemonRequest::BashRun { - cmd: args.cmd, - timeout_secs: args.timeout_secs, - wait_seconds: args.wait_seconds, - }; - let resp = round_trip(req).await; - // Extract the id from the response to format the result. - match &resp { - Ok(DaemonResponse::Ok { payload }) => { - let id = payload["id"].as_str().unwrap_or("unknown").to_owned(); - render_bash_run(&id, resp) - } - _ => render_bash_run("unknown", resp), - } - } - - #[tool( - description = "Check the status of a background bash task by its ID (from `bash_run`). \ - 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 \ - avoid a separate round-trip when the task is expected to finish soon." - )] - async fn bash_status(&self, Parameters(args): Parameters) -> String { - let id = args.id.clone(); - let req = DaemonRequest::BashStatus { - id: args.id, - wait_seconds: args.wait_seconds, - }; - render_bash_status(&id, round_trip(req).await) - } -} - -#[tool_handler] -impl ServerHandler for BashMcp {} - -#[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn")), - ) - .with_writer(std::io::stderr) - .init(); - - let service = BashMcp.serve(stdio()).await?; - service.waiting().await?; - Ok(()) -} diff --git a/hive-bash-mcp/src/lib.rs b/hive-bash-mcp/src/lib.rs deleted file mode 100644 index 4ca4eb93..00000000 --- a/hive-bash-mcp/src/lib.rs +++ /dev/null @@ -1,10 +0,0 @@ -//! Shared library for `hive-bash-daemon` and `hive-bash-mcp`. -//! -//! The daemon owns the subprocess runner loop and unix socket server. -//! The stdio MCP bridge is a thin client that forwards each tool call -//! to the daemon over the unix socket. - -pub mod paths; -pub mod protocol; -pub mod runner; -pub mod socket; diff --git a/hive-bash-mcp/src/main.rs b/hive-bash-mcp/src/main.rs deleted file mode 100644 index 8cfbe8ff..00000000 --- a/hive-bash-mcp/src/main.rs +++ /dev/null @@ -1,32 +0,0 @@ -//! `hive-bash-daemon` binary — long-running per-agent bash task runner. -//! Spawns `sh -c` subprocesses, monitors completion, writes task state -//! files, and fires hyperhive wake signals. Listens on a unix socket -//! for tool-call requests from the `hive-bash-mcp` stdio bridge. - -use anyhow::Result; - -#[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), - ) - .init(); - - let socket_path = hive_bash_mcp::paths::daemon_socket(); - let wake_socket = hive_bash_mcp::paths::hyperhive_socket(); - - tracing::info!( - socket = %socket_path.display(), - wake = %wake_socket.display(), - "hive-bash-daemon starting" - ); - - // Start the background runner loop — scans for pending tasks and - // spawns them, sends wake signals on completion. - hive_bash_mcp::runner::spawn_loop(wake_socket); - - // Serve the unix socket forever. - hive_bash_mcp::socket::serve(&socket_path).await -} diff --git a/hive-bash-mcp/src/paths.rs b/hive-bash-mcp/src/paths.rs deleted file mode 100644 index dd4c5a70..00000000 --- a/hive-bash-mcp/src/paths.rs +++ /dev/null @@ -1,89 +0,0 @@ -//! Per-agent filesystem paths used by both `hive-bash-daemon` and the -//! stdio MCP bridge. -//! -//! All paths are overridable via env vars so the operator can redirect -//! them in agent.nix when needed. - -use std::path::PathBuf; - -/// Default unix socket path the daemon listens on inside the agent -/// container. Lives under systemd's `RuntimeDirectory=hive-bash` -/// (a tmpfs path that disappears on container restart — the daemon -/// recreates the socket on its own boot) so the agent unix user can -/// bind without root in `/run`. -pub const DEFAULT_DAEMON_SOCKET: &str = "/run/hive-bash/socket"; - -/// Resolve the daemon's unix socket path. Override via `HIVE_BASH_SOCKET`. -#[must_use] -pub fn daemon_socket() -> PathBuf { - std::env::var_os("HIVE_BASH_SOCKET") - .map_or_else(|| PathBuf::from(DEFAULT_DAEMON_SOCKET), PathBuf::from) -} - -/// Base directory for task files. Uses `HYPERHIVE_HARNESS_DIR` if set -/// (injected by hive-c0re meta flake after the harness/state split); -/// falls back to a sibling of the state dir for pre-split deployments. -#[must_use] -pub fn tasks_dir() -> PathBuf { - let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { - PathBuf::from(p) - } else { - // Pre-split fallback: derive harness/ as a sibling of state/. - let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); - let state_path = PathBuf::from(&state); - state_path - .parent() - .map(|p| p.join("harness")) - .unwrap_or_else(|| PathBuf::from(state)) - }; - base.join("bash-tasks") -} - -/// Hyperhive control socket — the daemon writes wake signals here so -/// the harness drives a new claude turn on bash task completion. -/// Mirrors the path used by `forge_notify` and `hive-matrix-mcp`. -#[must_use] -pub fn hyperhive_socket() -> PathBuf { - std::env::var_os("HIVE_CONTROL_SOCKET") - .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) -} - -/// Directory where MCP daemons write loose-end summary files for the harness. -/// Each daemon writes `.json` here; the harness scans the dir in -/// `get_loose_ends` to surface active work from all MCPs generically. -/// -/// NOTE: the base-dir resolution logic here is intentionally mirrored in -/// `hive-ag3nt/src/mcp_loose_ends.rs::loose_ends_dir()`. They can't share -/// code across crates — keep them in sync if the fallback logic changes. -#[must_use] -pub fn mcp_loose_ends_dir() -> PathBuf { - let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { - PathBuf::from(p) - } else { - let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); - let state_path = PathBuf::from(&state); - state_path - .parent() - .map(|p| p.join("harness")) - .unwrap_or_else(|| PathBuf::from(state)) - }; - base.join("mcp-loose-ends") -} - -/// Full path for a task's JSON metadata file. -#[must_use] -pub fn task_json(id: &str) -> PathBuf { - tasks_dir().join(format!("{id}.json")) -} - -/// Full path for a task's captured stdout. -#[must_use] -pub fn task_out(id: &str) -> PathBuf { - tasks_dir().join(format!("{id}.out")) -} - -/// Full path for a task's captured stderr. -#[must_use] -pub fn task_err(id: &str) -> PathBuf { - tasks_dir().join(format!("{id}.err")) -} diff --git a/hive-bash-mcp/src/protocol.rs b/hive-bash-mcp/src/protocol.rs deleted file mode 100644 index 91e74f57..00000000 --- a/hive-bash-mcp/src/protocol.rs +++ /dev/null @@ -1,110 +0,0 @@ -//! Wire types for the unix socket protocol between `hive-bash-daemon` -//! and `hive-bash-mcp`. One JSON request line in, one JSON response line -//! out per connection. Connections are short-lived (per tool call). - -use serde::{Deserialize, Serialize}; - -// --------------------------------------------------------------------------- -// Task state (shared between runner and protocol) -// --------------------------------------------------------------------------- - -/// Lifecycle state of a bash task. -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum TaskStatus { - Pending, - Running, - Done, - TimedOut, - /// Daemon was restarted while the task was running; process is gone. - Interrupted, -} - -/// Task metadata + result written to `.json` under the tasks dir. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TaskFile { - pub id: String, - pub cmd: String, - pub timeout_secs: u64, - pub status: TaskStatus, - pub created_at: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub completed_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub exit_code: Option, - /// Last [`crate::runner::SUMMARY_BYTES`] of stdout. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub stdout_tail: Option, - /// Last [`crate::runner::SUMMARY_BYTES`] of stderr. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub stderr_tail: Option, -} - -// --------------------------------------------------------------------------- -// Request / response -// --------------------------------------------------------------------------- - -/// Requests the MCP bridge sends to the daemon. -#[derive(Debug, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum DaemonRequest { - /// Liveness probe — fast round-trip that doesn't touch any subprocess. - Ping, - - /// Submit a new bash task. Returns the task ID on success. - /// `wait_seconds`: optional inline poll (capped at 30s); when the - /// task finishes within the window the response carries the full - /// status payload. When it doesn't, the response carries just the - /// task ID so the caller can check back with `BashStatus`. - BashRun { - cmd: String, - #[serde(default)] - timeout_secs: Option, - /// Inline wait cap: 30s. Pass `None` or `0` to get the - /// task-started-id response immediately. - #[serde(default)] - wait_seconds: Option, - }, - - /// Query the current status of a task. Returns the full `TaskFile` - /// (formatted as text by the bridge). `wait_seconds`: optional - /// inline poll (capped at 30s) — daemon returns as soon as the - /// task reaches a terminal state or the window expires. - BashStatus { - id: String, - #[serde(default)] - wait_seconds: Option, - }, - - /// Return all tasks currently in `Pending` or `Running` state. - /// Used by the harness `get_loose_ends` to surface active - /// background work. - ActiveTasks, -} - -/// Response shape from the daemon. `Ok` carries a JSON payload; `Error` -/// carries a human-readable message. -#[derive(Debug, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] -pub enum DaemonResponse { - Ok { payload: serde_json::Value }, - Error { message: String }, -} - -impl DaemonResponse { - /// Build an Ok response from any serialisable value. - pub fn ok(payload: &T) -> Self { - let payload = serde_json::to_value(payload) - .unwrap_or_else(|e| serde_json::json!({ "serialise_error": e.to_string() })); - Self::Ok { payload } - } - - /// Build an Error response from any `Display` value. - pub fn error(msg: impl std::fmt::Display) -> Self { - Self::Error { - message: msg.to_string(), - } - } -} diff --git a/hive-bash-mcp/src/socket.rs b/hive-bash-mcp/src/socket.rs deleted file mode 100644 index d9e515be..00000000 --- a/hive-bash-mcp/src/socket.rs +++ /dev/null @@ -1,107 +0,0 @@ -//! Unix socket server: the daemon listens here, the stdio MCP bridge -//! `connect()`s on every tool call. One JSON request line in, one -//! JSON response line out. Connections are short-lived (per tool call) -//! so the loop is just accept → dispatch → reply → close. - -use std::path::Path; - -use anyhow::{Context, Result}; -use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; -use tokio::net::{UnixListener, UnixStream}; - -use crate::protocol::{DaemonRequest, DaemonResponse, TaskStatus}; -use crate::runner; - -/// Start listening on `socket_path` and serve forever. Removes any -/// stale socket file first so a daemon restart doesn't hit EADDRINUSE. -pub async fn serve(socket_path: &Path) -> Result<()> { - let _ = tokio::fs::remove_file(socket_path).await; - if let Some(parent) = socket_path.parent() { - tokio::fs::create_dir_all(parent) - .await - .with_context(|| format!("mkdir {}", parent.display()))?; - } - let listener = UnixListener::bind(socket_path) - .with_context(|| format!("bind unix socket {}", socket_path.display()))?; - tracing::info!(path = %socket_path.display(), "bash daemon socket up"); - loop { - let (stream, _) = listener - .accept() - .await - .context("accept on bash daemon socket")?; - tokio::spawn(async move { - if let Err(e) = handle_connection(stream).await { - tracing::warn!(error = %e, "bash socket connection error"); - } - }); - } -} - -async fn handle_connection(stream: UnixStream) -> Result<()> { - let (reader, mut writer) = stream.into_split(); - let mut lines = BufReader::new(reader).lines(); - while let Some(line) = lines.next_line().await? { - let response = match serde_json::from_str::(&line) { - Ok(req) => dispatch(req).await, - Err(e) => DaemonResponse::error(format!("parse request: {e}")), - }; - let mut json = serde_json::to_string(&response)?; - json.push('\n'); - writer.write_all(json.as_bytes()).await?; - writer.flush().await?; - } - Ok(()) -} - -async fn dispatch(req: DaemonRequest) -> DaemonResponse { - match req { - DaemonRequest::Ping => DaemonResponse::ok(&serde_json::json!({"ok": true})), - - DaemonRequest::BashRun { - cmd, - timeout_secs, - wait_seconds, - } => { - let id = match runner::submit_task(cmd, timeout_secs) { - Ok(id) => id, - Err(e) => return DaemonResponse::error(format!("submit_task: {e:#}")), - }; - // Inline wait: if requested and the task finishes quickly, - // return the full status instead of just the task ID. - let wait = wait_seconds.unwrap_or(0); - if wait > 0 { - if let Some(task) = runner::wait_for_task(&id, wait).await { - if matches!( - task.status, - TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted - ) { - return DaemonResponse::ok(&serde_json::json!({ - "id": id, - "finished": true, - "task": task, - })); - } - } - } - DaemonResponse::ok(&serde_json::json!({ "id": id, "finished": false })) - } - - DaemonRequest::BashStatus { id, wait_seconds } => { - let wait = wait_seconds.unwrap_or(0); - let task = if wait > 0 { - runner::wait_for_task(&id, wait).await - } else { - runner::read_task(&id) - }; - match task { - Some(t) => DaemonResponse::ok(&t), - None => DaemonResponse::error(format!("unknown task id `{id}`")), - } - } - - DaemonRequest::ActiveTasks => { - let tasks = runner::active_tasks(); - DaemonResponse::ok(&tasks) - } - } -} diff --git a/nix/templates/harness-base.nix b/nix/templates/harness-base.nix index 7315e5cc..491fb360 100644 --- a/nix/templates/harness-base.nix +++ b/nix/templates/harness-base.nix @@ -1118,48 +1118,6 @@ in }; }; - # Bash task runner daemon — long-running process that owns subprocess - # monitoring + completion wake signals. Always enabled (every agent - # needs bash tools). The stdio MCP bridge `hive-bash-mcp` connects - # to this daemon's socket per turn. - # Socket dir: /run/hive-bash/ — RuntimeDirectory keeps it on tmpfs. - systemd.services.hive-bash-daemon = { - description = "bash task runner daemon for hive-bash-mcp"; - wantedBy = [ "multi-user.target" ]; - environment = { - HIVE_BASH_SOCKET = "/run/hive-bash/socket"; - HIVE_CONTROL_SOCKET = "/run/hive/mcp.sock"; - RUST_LOG = "info"; - # HYPERHIVE_HARNESS_DIR and HYPERHIVE_STATE_DIR are already - # injected via systemd.globalEnvironment by the meta flake - # (set to /agents//harness and /agents//state - # respectively). Listed here for explicitness — the daemon - # uses these to derive its task + loose-ends dir paths. - # Without them the daemon falls back to deriving harness/ as a - # sibling of state/, which produces the same value but is - # less robust if the two vars ever diverge. - }; - serviceConfig = { - ExecStart = "${pkgs.hyperhive}/bin/hive-bash-daemon"; - Restart = "on-failure"; - RestartSec = 3; - User = userName; - Group = userName; - RuntimeDirectory = "hive-bash"; - }; - }; - - # Inject the bash MCP bridge into every agent's extraMcpServers. - # The bridge connects to hive-bash-daemon on the same socket. - hyperhive.extraMcpServers = { - bash = lib.mkDefault { - command = "${pkgs.hyperhive}/bin/hive-bash-mcp"; - args = [ ]; - env.HIVE_BASH_SOCKET = "/run/hive-bash/socket"; - allowedTools = [ "*" ]; - }; - }; - # Re-fire the daemon when the matrix token appears (hive-c0re # provisions it after agent containers come up). Without this # the daemon would exit 0 silently on first boot and the MCP