From e86160820ad6a140c93aee899de5ee4469778da9 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 3 Jun 2026 17:03:16 +0200 Subject: [PATCH] feat(#1106): split bash mcp into hive-bash-daemon + hive-bash-mcp bridge - new hive-bash-mcp crate: daemon (subprocess runner, wake signals) + stdio bridge (mcp tools). mirrors hive-matrix-mcp architecture - hive-ag3nt: remove bash_runner.rs and bash_run/bash_status mcp tools; get_loose_ends uses hive_bash_mcp::runner::active_tasks() via crate dep - harness-base.nix: add hive-bash-daemon systemd service + auto-inject bash extraMcpServer into every agent (socket: /run/hive-bash/socket) --- Cargo.lock | 16 + Cargo.toml | 1 + hive-ag3nt/Cargo.toml | 1 + hive-ag3nt/src/bin/hive.rs | 1 - hive-ag3nt/src/lib.rs | 1 - hive-ag3nt/src/mcp.rs | 184 +--------- hive-bash-mcp/Cargo.toml | 35 ++ hive-bash-mcp/src/bin/mcp.rs | 239 ++++++++++++ hive-bash-mcp/src/lib.rs | 10 + hive-bash-mcp/src/main.rs | 32 ++ hive-bash-mcp/src/paths.rs | 67 ++++ hive-bash-mcp/src/protocol.rs | 110 ++++++ .../src/runner.rs | 346 ++++++++---------- hive-bash-mcp/src/socket.rs | 107 ++++++ nix/templates/harness-base.nix | 34 ++ 15 files changed, 811 insertions(+), 373 deletions(-) create mode 100644 hive-bash-mcp/Cargo.toml create mode 100644 hive-bash-mcp/src/bin/mcp.rs create mode 100644 hive-bash-mcp/src/lib.rs create mode 100644 hive-bash-mcp/src/main.rs create mode 100644 hive-bash-mcp/src/paths.rs create mode 100644 hive-bash-mcp/src/protocol.rs rename hive-ag3nt/src/bash_runner.rs => hive-bash-mcp/src/runner.rs (55%) create mode 100644 hive-bash-mcp/src/socket.rs diff --git a/Cargo.lock b/Cargo.lock index 9bf9e23f..4ac9ebe3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1226,6 +1226,7 @@ dependencies = [ "axum", "clap", "futures-util", + "hive-bash-mcp", "hive-sh4re", "reqwest", "rmcp", @@ -1241,6 +1242,21 @@ 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 33f9bb67..cae6b9d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,6 +2,7 @@ resolver = "3" members = [ "hive-ag3nt", + "hive-bash-mcp", "hive-c0re", "hive-forge", "hive-matrix-mcp", diff --git a/hive-ag3nt/Cargo.toml b/hive-ag3nt/Cargo.toml index ed5de53b..76a01aea 100644 --- a/hive-ag3nt/Cargo.toml +++ b/hive-ag3nt/Cargo.toml @@ -12,6 +12,7 @@ axum.workspace = true reqwest.workspace = true futures-util = "0.3" clap.workspace = true +hive-bash-mcp = { path = "../hive-bash-mcp" } hive-sh4re.workspace = true rmcp.workspace = true rusqlite.workspace = true diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index b31e5791..b880bb11 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -570,7 +570,6 @@ 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 72c94e7f..34c04282 100644 --- a/hive-ag3nt/src/lib.rs +++ b/hive-ag3nt/src/lib.rs @@ -1,7 +1,6 @@ //! 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; diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 0432b252..469930bd 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -494,96 +494,6 @@ 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 @@ -829,7 +739,7 @@ impl AgentServer { let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); // 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(); + let active = hive_bash_mcp::runner::active_tasks(); if !active.is_empty() { use std::fmt::Write as _; let _ = write!(out, "\n\n{} active bash task(s):", active.len()); @@ -964,94 +874,6 @@ 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 \ @@ -2110,8 +1932,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__hyperhive__bash_run` (background tasks -/// with structured output) instead of a raw interactive shell. `TodoWrite` +/// execution goes through `mcp__hive_bash__bash_run` (background tasks +/// with structured output via `hive-bash-mcp`) 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-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml new file mode 100644 index 00000000..81699f08 --- /dev/null +++ b/hive-bash-mcp/Cargo.toml @@ -0,0 +1,35 @@ +[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 new file mode 100644 index 00000000..90c5a047 --- /dev/null +++ b/hive-bash-mcp/src/bin/mcp.rs @@ -0,0 +1,239 @@ +//! `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 new file mode 100644 index 00000000..4ca4eb93 --- /dev/null +++ b/hive-bash-mcp/src/lib.rs @@ -0,0 +1,10 @@ +//! 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 new file mode 100644 index 00000000..8cfbe8ff --- /dev/null +++ b/hive-bash-mcp/src/main.rs @@ -0,0 +1,32 @@ +//! `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 new file mode 100644 index 00000000..ce960f66 --- /dev/null +++ b/hive-bash-mcp/src/paths.rs @@ -0,0 +1,67 @@ +//! 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) +} + +/// 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 new file mode 100644 index 00000000..91e74f57 --- /dev/null +++ b/hive-bash-mcp/src/protocol.rs @@ -0,0 +1,110 @@ +//! 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-ag3nt/src/bash_runner.rs b/hive-bash-mcp/src/runner.rs similarity index 55% rename from hive-ag3nt/src/bash_runner.rs rename to hive-bash-mcp/src/runner.rs index 7d8d3526..19f2dea4 100644 --- a/hive-ag3nt/src/bash_runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -1,50 +1,63 @@ -//! 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()`. +//! 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`. //! -//! Files under `harness_dir()/bash-tasks/`: +//! Files under `tasks_dir()`: //! - `.json` — task metadata + status (pending → running → done) -//! - `.out` — captured stdout (appended while running) -//! - `.err` — captured stderr (appended while running) +//! - `.out` — captured stdout (streamed while running) +//! - `.err` — captured stderr (streamed while running) //! -//! 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. +//! 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. //! //! The runner kills the child process on timeout — `tokio::process::Child::drop()` -//! does not kill children, so we explicitly call `child.kill().await` before -//! collecting the copy tasks. +//! does not kill children, so we explicitly call `child.kill().await`. use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::Result; -use serde::{Deserialize, Serialize}; -use tokio::io::AsyncWriteExt; +use tokio::io::AsyncWriteExt as _; +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. Exposed so -/// `format_bash_status` can detect truncation and surface the full-output -/// file path when the on-disk file is larger. +/// Full output always lives in the `.out`/`.err` files. pub const SUMMARY_BYTES: usize = 4096; -/// Default timeout for tasks that don't specify one. + +/// Default task timeout. 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); -/// Generate a task ID: `` — unique within a -/// harness session; collision chance across sessions negligible for our -/// volume. +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} + +/// Generate a task ID: ``. #[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() @@ -53,107 +66,13 @@ 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 = task_json(&task.id); + let dest = paths::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) } @@ -161,21 +80,43 @@ 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(task_json(id)).ok()?; + let s = std::fs::read_to_string(paths::task_json(id)).ok()?; serde_json::from_str(&s).ok() } // --------------------------------------------------------------------------- -// Public API used by MCP tools +// Public API used by daemon dispatch // --------------------------------------------------------------------------- -/// 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. +/// 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)?; + Ok(id) +} + +/// Return all tasks currently in `Pending` or `Running` state. #[must_use] pub fn active_tasks() -> Vec { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { return Vec::new(); }; let mut out = Vec::new(); @@ -195,49 +136,53 @@ pub fn active_tasks() -> Vec { out } -/// 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) +/// 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) } // --------------------------------------------------------------------------- // 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 the completion -/// `Wake`. Call once at harness startup from `serve_main`. -pub fn spawn(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 completion +/// wake signals. Call once at daemon startup. +pub fn spawn_loop(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(tasks_dir()) { + if let Err(e) = std::fs::create_dir_all(paths::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 { @@ -246,10 +191,9 @@ async fn run_loop(socket: PathBuf) { } } -/// On boot, find any task files in `running` state and flip them to -/// `interrupted`, then fire a wake so the agent unblocks. +/// On boot, flip any `running` tasks to `interrupted` and fire a wake. async fn mark_interrupted(socket: &Path) { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { return; }; for entry in rd.flatten() { @@ -268,16 +212,16 @@ async fn mark_interrupted(socket: &Path) { } tracing::warn!(id = %id, "bash_runner: marking interrupted task"); task.status = TaskStatus::Interrupted; - task.completed_at = Some(crate::serve_common::now_unix()); + task.completed_at = Some(now_unix()); if let Err(e) = write_task(&task) { - tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted task failed"); + tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed"); } - send_wake(socket, &id, "interrupted (harness restarted)", None).await; + send_wake(socket, &id, "interrupted (daemon restarted)", None).await; } } async fn poll_once(socket: &Path, claimed: &Arc>>) { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { + let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { return; }; for entry in rd.flatten() { @@ -304,8 +248,6 @@ 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); }); } @@ -320,13 +262,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(crate::serve_common::now_unix()); + task.started_at = Some(now_unix()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed"); } - let out_path = task_out(&id); - let err_path = task_err(&id); + let out_path = paths::task_out(&id); + let err_path = paths::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 { @@ -349,7 +291,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { } else { TaskStatus::Done }; - task.completed_at = Some(crate::serve_common::now_unix()); + task.completed_at = Some(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()); @@ -363,13 +305,12 @@ async fn run_task(mut task: TaskFile, socket: &Path) { } else { format!("exit={}", exit_code.unwrap_or(-1)) }; - let output_snippet = stdout_tail.as_deref().unwrap_or("").trim(); + 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((output_snippet, err_snippet))).await; + send_wake(socket, &id, &summary, Some((out_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, @@ -390,29 +331,27 @@ async fn exec_cmd( let out_path = out_path.to_path_buf(); let err_path = err_path.to_path_buf(); - // Stream stdout and stderr to files concurrently. - let copy_stdout = tokio::spawn(copy_stream_to_file( + let copy_out = tokio::spawn(copy_stream_to_file( tokio::io::BufReader::new(stdout), out_path, )); - let copy_stderr = tokio::spawn(copy_stream_to_file( + let copy_err = 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_stdout.await; - let _ = copy_stderr.await; + let _ = copy_out.await; + let _ = copy_err.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; // reap zombie - let _ = copy_stdout.await; - let _ = copy_stderr.await; + let _ = child.wait().await; + let _ = copy_out.await; + let _ = copy_err.await; Ok((-1, true)) } } @@ -428,12 +367,11 @@ 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 { @@ -448,7 +386,12 @@ fn tail_file(path: &Path, max_bytes: usize) -> Option { // Wake delivery // --------------------------------------------------------------------------- -async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str, &str)>) { +pub(crate) 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() { @@ -465,14 +408,37 @@ async fn send_wake(socket: &Path, id: &str, summary: &str, output: Option<(&str, 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, }; - 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"), + + 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"); + } } } diff --git a/hive-bash-mcp/src/socket.rs b/hive-bash-mcp/src/socket.rs new file mode 100644 index 00000000..d9e515be --- /dev/null +++ b/hive-bash-mcp/src/socket.rs @@ -0,0 +1,107 @@ +//! 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 491fb360..20abfb36 100644 --- a/nix/templates/harness-base.nix +++ b/nix/templates/harness-base.nix @@ -1118,6 +1118,40 @@ 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"; + }; + 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