478 lines
17 KiB
Rust
478 lines
17 KiB
Rust
//! 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.
|
|
//!
|
|
//! 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. Exposed so
|
|
/// `format_bash_status` can detect truncation and surface the full-output
|
|
/// file path when the on-disk file is larger.
|
|
pub 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"))
|
|
}
|
|
|
|
/// Path to the full stdout capture file for `id`.
|
|
/// Exposed so `format_bash_status` can surface it when output is truncated.
|
|
#[must_use]
|
|
pub fn task_out_path(id: &str) -> PathBuf {
|
|
task_out(id)
|
|
}
|
|
|
|
/// Path to the full stderr capture file for `id`.
|
|
/// Exposed so `format_bash_status` can surface it when output is truncated.
|
|
#[must_use]
|
|
pub fn task_err_path(id: &str) -> PathBuf {
|
|
task_err(id)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Wire types (shared between MCP tool writers and runner readers)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum TaskStatus {
|
|
Pending,
|
|
Running,
|
|
Done,
|
|
TimedOut,
|
|
/// Harness was restarted while the task was running; process is gone.
|
|
Interrupted,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct TaskFile {
|
|
pub id: String,
|
|
pub cmd: String,
|
|
pub timeout_secs: u64,
|
|
pub status: TaskStatus,
|
|
pub created_at: i64,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
pub started_at: Option<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
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Return all tasks in `Pending` or `Running` state. Used by the
|
|
/// `get_loose_ends` tool to surface active background work alongside
|
|
/// broker-side items (questions, reminders). Silently skips unreadable
|
|
/// or unparseable files.
|
|
#[must_use]
|
|
pub fn active_tasks() -> Vec<TaskFile> {
|
|
let Ok(rd) = std::fs::read_dir(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
|
|
}
|
|
|
|
/// 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 (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<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,
|
|
// Transient: do not persist to the message broker. Bash task
|
|
// completions are fire-and-forget — the output is already on disk
|
|
// in harness/bash-tasks/; persisting would cause duplicate delivery
|
|
// after a harness restart.
|
|
transient: true,
|
|
};
|
|
match crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req).await {
|
|
Ok(_) => tracing::info!(id = %id, "bash_runner: wake delivered"),
|
|
Err(e) => tracing::warn!(id = %id, error = ?e, "bash_runner: wake delivery failed"),
|
|
}
|
|
}
|