vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(#1106): split bash mcp into hive-bash-daemon + hive-bash-mcp bridge

Browse source 
- 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:🏃: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)
This commit is contained in:
damocles 2026-06-03 17:03:16 +02:00 • committed by mara
commit e86160820a
15 changed files with 811 additions and 373 deletions
Download patch file Download diff file Expand all files Collapse all files

35
hive-bash-mcp/Cargo.toml Normal file
View file

@ -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"

239
hive-bash-mcp/src/bin/mcp.rs Normal file
View file

@ -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<DaemonResponse> {
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<DaemonResponse>) -> 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<DaemonResponse>) -> 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<u64>,
/// 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=<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<u64>,
}
fn default_wait() -> Option<u64> {
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<u64>,
}
#[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-<id>\"` 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=<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<BashRunArgs>) -> 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/<id>.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<BashStatusArgs>) -> 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(())
}

10
hive-bash-mcp/src/lib.rs Normal file
View file

@ -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;

32
hive-bash-mcp/src/main.rs Normal file
View file

@ -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
}

67
hive-bash-mcp/src/paths.rs Normal file
View file

@ -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"))
}

110
hive-bash-mcp/src/protocol.rs Normal file
View file

@ -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 `<id>.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<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub completed_at: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub exit_code: Option<i32>,
/// Last [`crate::runner::SUMMARY_BYTES`] of stdout.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stdout_tail: Option<String>,
/// Last [`crate::runner::SUMMARY_BYTES`] of stderr.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stderr_tail: Option<String>,
}
// ---------------------------------------------------------------------------
// 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<u64>,
/// Inline wait cap: 30s. Pass `None` or `0` to get the
/// task-started-id response immediately.
#[serde(default)]
wait_seconds: Option<u64>,
},
/// 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<u64>,
},
/// 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<T: Serialize>(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(),
}
}
}

444
hive-bash-mcp/src/runner.rs Normal file
View file

@ -0,0 +1,444 @@
//! Bash subprocess runner: spawns `sh -c <cmd>` 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 `tasks_dir()`:
//! - `<id>.json` — task metadata + status (pending → running → done)
//! - `<id>.out` — captured stdout (streamed while running)
//! - `<id>.err` — captured stderr (streamed 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.
//!
//! The runner kills the child process on timeout — `tokio::process::Child::drop()`
//! does not kill children, so we explicitly call `child.kill().await`.
use std::collections::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 anyhow::Result;
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.
pub const SUMMARY_BYTES: usize = 4096;
/// 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);
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
fn now_unix() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64
}
/// Generate a task ID: `<timestamp_hex><seq_hex>`.
#[must_use]
pub fn new_task_id() -> String {
let t = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis();
let seq = TASK_SEQ.fetch_add(1, Ordering::Relaxed);
format!("{t:013x}{seq:04x}")
}
/// 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 tmp = dest.with_extension("json.tmp");
std::fs::write(&tmp, &json)?;
std::fs::rename(&tmp, &dest)
}
/// Read a task file. Returns `None` if the file doesn't exist or is
/// unparseable.
#[must_use]
pub fn read_task(id: &str) -> Option<TaskFile> {
let s = std::fs::read_to_string(paths::task_json(id)).ok()?;
serde_json::from_str(&s).ok()
}
// ---------------------------------------------------------------------------
// 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<u64>) -> Result<String> {
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<TaskFile> {
let Ok(rd) = std::fs::read_dir(paths::tasks_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 Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
continue;
};
let Some(task) = read_task(&id) else { continue };
if matches!(task.status, TaskStatus::Pending | TaskStatus::Running) {
out.push(task);
}
}
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<TaskFile> {
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 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(paths::tasks_dir()) {
tracing::warn!(error = ?e, "bash_runner: create tasks dir failed");
}
mark_interrupted(&socket).await;
let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
loop {
poll_once(&socket, &claimed).await;
tokio::time::sleep(POLL_INTERVAL).await;
}
}
/// 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(paths::tasks_dir()) else {
return;
};
for entry in rd.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
continue;
};
let Some(mut task) = read_task(&id) else {
continue;
};
if task.status != TaskStatus::Running {
continue;
}
tracing::warn!(id = %id, "bash_runner: marking interrupted task");
task.status = TaskStatus::Interrupted;
task.completed_at = Some(now_unix());
if let Err(e) = write_task(&task) {
tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
}
send_wake(socket, &id, "interrupted (daemon restarted)", None).await;
}
}
async fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
return;
};
for entry in rd.flatten() {
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
continue;
};
{
let guard = claimed.lock().unwrap();
if guard.contains(&id) {
continue;
}
}
let Some(task) = read_task(&id) else { continue };
if task.status != TaskStatus::Pending {
continue;
}
// Claim before spawning to avoid double-spawn across poll iterations.
claimed.lock().unwrap().insert(id.clone());
let socket = socket.to_path_buf();
let claimed = claimed.clone();
tokio::spawn(async move {
run_task(task, &socket).await;
claimed.lock().unwrap().remove(&id);
});
}
}
// ---------------------------------------------------------------------------
// Task execution
// ---------------------------------------------------------------------------
async fn run_task(mut task: TaskFile, socket: &Path) {
let id = task.id.clone();
tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task");
task.status = TaskStatus::Running;
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 = 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 {
Ok((code, false)) => (false, Some(code)),
Ok((_, true)) => {
tracing::warn!(id = %id, "bash_runner: task timed out");
(true, None)
}
Err(e) => {
tracing::warn!(id = %id, error = ?e, "bash_runner: exec error");
(false, None)
}
};
let stdout_tail = tail_file(&out_path, SUMMARY_BYTES);
let stderr_tail = tail_file(&err_path, SUMMARY_BYTES);
task.status = if timed_out {
TaskStatus::TimedOut
} else {
TaskStatus::Done
};
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());
if let Err(e) = write_task(&task) {
tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed");
}
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 err_snippet = stderr_tail.as_deref().unwrap_or("").trim();
send_wake(socket, &id, &summary, Some((out_snippet, err_snippet))).await;
}
/// Run `sh -c cmd`, streaming output to files. Returns `(exit_code, timed_out)`.
async fn exec_cmd(
cmd: &str,
out_path: &Path,
err_path: &Path,
timeout: Duration,
) -> Result<(i32, bool)> {
use tokio::process::Command;
let mut child = Command::new("sh")
.arg("-c")
.arg(cmd)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()?;
let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped");
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(
tokio::io::BufReader::new(stdout),
out_path,
));
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_out.await;
let _ = copy_err.await;
Ok((status.code().unwrap_or(-1), false))
}
Ok(Err(e)) => Err(e.into()),
Err(_elapsed) => {
let _ = child.kill().await;
let _ = child.wait().await;
let _ = copy_out.await;
let _ = copy_err.await;
Ok((-1, true))
}
}
}
async fn copy_stream_to_file<R>(mut reader: R, path: PathBuf)
where
R: tokio::io::AsyncRead + Unpin,
{
match tokio::fs::File::create(&path).await {
Ok(mut f) => {
let _ = tokio::io::copy(&mut reader, &mut f).await;
let _ = f.flush().await;
}
Err(e) => {
tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed");
}
}
}
fn tail_file(path: &Path, max_bytes: usize) -> Option<String> {
let data = std::fs::read(path).ok()?;
let slice = if data.len() > max_bytes {
&data[data.len() - max_bytes..]
} else {
&data
};
Some(String::from_utf8_lossy(slice).into_owned())
}
// ---------------------------------------------------------------------------
// Wake delivery
// ---------------------------------------------------------------------------
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() {
body.push_str("\n\nstdout:\n```\n");
body.push_str(stdout);
body.push_str("\n```");
}
if !stderr.is_empty() {
body.push_str("\n\nstderr:\n```\n");
body.push_str(stderr);
body.push_str("\n```");
}
}
let req = hive_sh4re::AgentRequest::Wake {
from: format!("bash-task-{id}"),
body,
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");
}
}
}

107
hive-bash-mcp/src/socket.rs Normal file
View file

@ -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::<DaemonRequest>(&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)
}
}
}