diff --git a/hive-ag3nt/src/bash_runner.rs b/hive-ag3nt/src/bash_runner.rs new file mode 100644 index 00000000..4633e9ba --- /dev/null +++ b/hive-ag3nt/src/bash_runner.rs @@ -0,0 +1,417 @@ +//! 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 `harness_dir()/bash-tasks/`: +//! - `.json` — task metadata + status (pending → running → done) +//! - `.out` — captured stdout (appended while running) +//! - `.err` — captured stderr (appended 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. +//! +//! See `docs/bash-runner.md` for the full design rationale. + +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 anyhow::Result; +use serde::{Deserialize, Serialize}; +use tokio::io::AsyncWriteExt; + +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. +const SUMMARY_BYTES: usize = 4096; +/// Default timeout for tasks that don't specify one. +pub const DEFAULT_TIMEOUT_SECS: u64 = 180; + +static TASK_SEQ: AtomicU64 = AtomicU64::new(0); + +/// 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() + .as_millis(); + let seq = TASK_SEQ.fetch_add(1, Ordering::Relaxed); + 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")) +} + +// --------------------------------------------------------------------------- +// 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 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 { + let s = std::fs::read_to_string(task_json(id)).ok()?; + serde_json::from_str(&s).ok() +} + +// --------------------------------------------------------------------------- +// Public API used by MCP tools +// --------------------------------------------------------------------------- + +/// 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 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(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 { + poll_once(&socket, &claimed).await; + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +/// 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(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(crate::serve_common::now_unix()); + if let Err(e) = write_task(&task) { + tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted task failed"); + } + send_wake(socket, &id, "interrupted (harness restarted)", None).await; + } +} + +async fn poll_once(socket: &Path, claimed: &Arc>>) { + let Ok(rd) = std::fs::read_dir(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; + // Remove from claimed so a resubmitted ID (rare) could be + // picked up again. In practice each ID is unique. + 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(crate::serve_common::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 timeout = Duration::from_secs(task.timeout_secs); + + let exec_result = tokio::time::timeout( + timeout, + exec_cmd(&task.cmd, &out_path, &err_path), + ) + .await; + + let timed_out; + let exit_code; + match exec_result { + Ok(Ok(code)) => { + timed_out = false; + exit_code = Some(code); + } + Ok(Err(e)) => { + tracing::warn!(id = %id, error = ?e, "bash_runner: exec error"); + timed_out = false; + exit_code = None; + } + Err(_elapsed) => { + tracing::warn!(id = %id, "bash_runner: task timed out"); + timed_out = true; + exit_code = 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(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()); + + 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 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((output_snippet, err_snippet))).await; +} + +async fn exec_cmd(cmd: &str, out_path: &Path, err_path: &Path) -> Result { + 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(); + + // 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_stderr = tokio::spawn(copy_stream_to_file( + tokio::io::BufReader::new(stderr), + err_path, + )); + + let status = child.wait().await?; + let _ = copy_stdout.await; + let _ = copy_stderr.await; + + Ok(status.code().unwrap_or(-1)) +} + +async fn copy_stream_to_file(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"), + } +} + +/// 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 { + &data[data.len() - max_bytes..] + } else { + &data + }; + Some(String::from_utf8_lossy(slice).into_owned()) +} + +// --------------------------------------------------------------------------- +// Wake delivery +// --------------------------------------------------------------------------- + +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, + }; + 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 e078fd03..f73266a0 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -551,6 +551,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 34c04282..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; diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index e3f4a134..85fbca5e 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -410,6 +410,60 @@ 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, +} + +/// MCP tool args for `bash_status`. +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct BashStatusArgs { + /// Task ID returned by `bash_run`. + pub id: String, +} + +/// Format the result of `bash_status` from a task ID. +#[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 { + 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); + } + } + if let Some(ref stdout) = task.stdout_tail { + if !stdout.trim().is_empty() { + let _ = write!(out, "\n\nstdout:\n```\n{}\n```", stdout.trim()); + } + } + if let Some(ref stderr) = task.stderr_tail { + if !stderr.trim().is_empty() { + let _ = write!(out, "\n\nstderr:\n```\n{}\n```", stderr.trim()); + } + } + 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 @@ -732,6 +786,38 @@ 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." + )] + async fn bash_run(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("bash_run", log, async move { + match crate::bash_runner::submit_task(args.cmd, args.timeout_secs) { + Ok(id) => format!("task started: id={id}"), + Err(e) => format!("bash_run failed: {e:#}"), + } + }) + .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`." + )] + async fn bash_status(&self, Parameters(args): Parameters) -> String { + let log = format!("{args:?}"); + run_tool_envelope("bash_status", log, async move { + 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 \ @@ -1740,8 +1826,8 @@ const TOOL_GROUPS_ENV: &str = "HIVE_TOOL_GROUPS"; /// Reads `HIVE_TOOL_GROUPS` from the environment first. Each comma-separated /// token is matched (case-insensitive) against the `ToolGroup` serde names /// (`messaging`, `meta`, `inbox`, `lifecycle`, `approvals`, `scheduling`, -/// `diagnostics`). Unrecognised tokens are logged and skipped. Falls back to -/// the flavor default when the env var is absent or empty. +/// `diagnostics`, `execution`). Unrecognised tokens are logged and skipped. +/// Falls back to the flavor default when the env var is absent or empty. fn effective_tool_groups(flavor: Flavor) -> Vec { let raw = match std::env::var(TOOL_GROUPS_ENV) { Ok(v) if !v.trim().is_empty() => v, diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index f25557b4..89c89140 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -736,6 +736,8 @@ pub enum ToolGroup { Scheduling, /// `get_logs` - *(privileged)* Diagnostics, + /// `bash_run`, `bash_status` + Execution, } impl ToolGroup { @@ -765,13 +767,14 @@ impl ToolGroup { "list_schedules", ], Self::Diagnostics => &["get_logs"], + Self::Execution => &["bash_run", "bash_status"], } } /// Default tool groups for a plain agent harness — equivalent to the /// old `Flavor::Agent` allow-list. Used when `HIVE_TOOL_GROUPS` is unset. pub const AGENT_DEFAULT: &'static [Self] = - &[Self::Messaging, Self::Meta, Self::Inbox]; + &[Self::Messaging, Self::Meta, Self::Inbox, Self::Execution]; /// Default tool groups for the manager harness — equivalent to the /// old `Flavor::Manager` allow-list. Used when `HIVE_TOOL_GROUPS` is unset. @@ -783,6 +786,7 @@ impl ToolGroup { Self::Approvals, Self::Scheduling, Self::Diagnostics, + Self::Execution, ]; }