feat(#665): harness-internal async bash task runner (option 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/<id>.json`; this background loop picks it up,
+//! runs `sh -c <cmd>` 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(<id>)`.
+//!
+//! Files under `harness_dir()/bash-tasks/`:
+//! - `<id>.json`  — task metadata + status (pending → running → done)
+//! - `<id>.out`   — captured stdout (appended while running)
+//! - `<id>.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: `<timestamp_hex><seq_hex>` — 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<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 `SUMMARY_BYTES` of stdout (full output in `<id>.out`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stdout_tail: Option<String>,
+    /// Last `SUMMARY_BYTES` of stderr (full output in `<id>.err`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stderr_tail: Option<String>,
+}
+
+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<TaskFile> {
+    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<u64>) -> Result<String> {
+    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<Mutex<HashSet<String>>> = 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<Mutex<HashSet<String>>>) {
+    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<i32> {
+    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<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"),
+    }
+}
+
+/// Read the last `max_bytes` of a file as a UTF-8 string.
+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
+// ---------------------------------------------------------------------------
+
+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"),
+    }
+}

hive-ag3nt/src/bin/hive.rs

@@ -551,6 +551,7 @@ async fn serve_main<S: Surface>(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

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;

hive-ag3nt/src/mcp.rs

@@ -410,6 +410,60 @@ pub struct RecvArgs {
     pub max: Option<u32>,
 }
 
+/// 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<u64>,
+}
+
+/// 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-<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."
+    )]
+    async fn bash_run(&self, Parameters(args): Parameters<BashRunArgs>) -> 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/<id>.out` / `.err`."
+    )]
+    async fn bash_status(&self, Parameters(args): Parameters<BashStatusArgs>) -> 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<hive_sh4re::ToolGroup> {
     let raw = match std::env::var(TOOL_GROUPS_ENV) {
         Ok(v) if !v.trim().is_empty() => v,

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,
     ];
 }