diff --git a/hive-ag3nt/src/bash_runner.rs b/hive-ag3nt/src/bash_runner.rs deleted file mode 100644 index 8a907070..00000000 --- a/hive-ag3nt/src/bash_runner.rs +++ /dev/null @@ -1,419 +0,0 @@ -//! 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. -//! -//! 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. - -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 (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(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; -} - -/// 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, 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(); - - // 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, - )); - - match tokio::time::timeout(timeout, child.wait()).await { - Ok(Ok(status)) => { - let _ = copy_stdout.await; - let _ = copy_stderr.await; - Ok((status.code().unwrap_or(-1), false)) - } - Ok(Err(e)) => Err(e.into()), - Err(_elapsed) => { - // Kill the child so it doesn't linger after timeout. - let _ = child.kill().await; - let _ = child.wait().await; // reap zombie - let _ = copy_stdout.await; - let _ = copy_stderr.await; - Ok((-1, true)) - } - } -} - -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 f73266a0..e078fd03 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -551,7 +551,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 ee156366..e3f4a134 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -410,60 +410,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, -} - -/// 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 && 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); - } - } - 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 @@ -786,38 +732,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." - )] - 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 \ @@ -1826,8 +1740,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`, `execution`). Unrecognised tokens are logged and skipped. -/// Falls back to the flavor default when the env var is absent or empty. +/// `diagnostics`). 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 89c89140..f25557b4 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -736,8 +736,6 @@ pub enum ToolGroup { Scheduling, /// `get_logs` - *(privileged)* Diagnostics, - /// `bash_run`, `bash_status` - Execution, } impl ToolGroup { @@ -767,14 +765,13 @@ 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::Execution]; + &[Self::Messaging, Self::Meta, Self::Inbox]; /// Default tool groups for the manager harness — equivalent to the /// old `Flavor::Manager` allow-list. Used when `HIVE_TOOL_GROUPS` is unset. @@ -786,7 +783,6 @@ impl ToolGroup { Self::Approvals, Self::Scheduling, Self::Diagnostics, - Self::Execution, ]; }