590 lines
20 KiB
Rust
590 lines
20 KiB
Rust
//! 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, bail};
|
|
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;
|
|
|
|
/// 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()
|
|
.cast_signed()
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Loose-ends file (generic MCP loose-ends protocol)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Rewrite `mcp-loose-ends/bash.json` with a summary of all currently
|
|
/// active (Pending or Running) tasks. The harness scans this directory
|
|
/// generically in `get_loose_ends` — no bash-specific code needed there.
|
|
///
|
|
/// File format: a JSON array of plain-text summary strings, one per
|
|
/// loose-end item. The harness includes them verbatim in the output.
|
|
/// Atomic write (tmp + rename) so the harness never reads a partial file.
|
|
fn refresh_loose_ends() {
|
|
let active = active_tasks();
|
|
let dir = paths::mcp_loose_ends_dir();
|
|
if let Err(e) = std::fs::create_dir_all(&dir) {
|
|
tracing::warn!(error = ?e, "bash_runner: create mcp-loose-ends dir failed");
|
|
return;
|
|
}
|
|
let items: Vec<String> = active
|
|
.iter()
|
|
.map(|t| {
|
|
let age = now_unix() - t.created_at;
|
|
format!(
|
|
"bash task `{}` status={:?}, cmd: `{}`, age {}s",
|
|
t.id, t.status, t.cmd, age
|
|
)
|
|
})
|
|
.collect();
|
|
let dest = dir.join("bash.json");
|
|
let tmp = dest.with_extension("json.tmp");
|
|
let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned());
|
|
if let Err(e) = std::fs::write(&tmp, &json).and_then(|()| std::fs::rename(&tmp, &dest)) {
|
|
tracing::warn!(error = ?e, "bash_runner: write mcp-loose-ends/bash.json failed");
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Public API used by daemon dispatch
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Longest accepted caller-chosen task name.
|
|
const MAX_TASK_NAME_LEN: usize = 64;
|
|
|
|
/// Validate a caller-chosen task name. The name doubles as the task id
|
|
/// and therefore as the `<name>.json` filename, so it must be a single
|
|
/// safe path segment. Allows ASCII alphanumerics plus `.`, `_`, `-`;
|
|
/// rejects empties, over-long names, `.`/`..`, and anything that could
|
|
/// escape the tasks dir or collide with the `.json.tmp` scratch suffix.
|
|
fn validate_task_name(name: &str) -> Result<()> {
|
|
if name.is_empty() {
|
|
bail!("task name must not be empty");
|
|
}
|
|
if name.len() > MAX_TASK_NAME_LEN {
|
|
bail!("task name too long (max {MAX_TASK_NAME_LEN} chars)");
|
|
}
|
|
if name == "." || name == ".." {
|
|
bail!("task name {name:?} is reserved");
|
|
}
|
|
if !name
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
|
|
{
|
|
bail!("task name {name:?} may only contain ASCII letters, digits, '.', '_', '-'");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Submit a new pending task. Returns the task ID.
|
|
///
|
|
/// When `name` is `Some`, it is validated and used as the task id (so it
|
|
/// surfaces in the wake `from`, status lookups, and the loose-ends list).
|
|
/// A name may be reused once any prior task of that name has finished;
|
|
/// submitting a name whose task is still `Pending`/`Running` is rejected.
|
|
/// `None` falls back to the auto-generated timestamp id.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns an error if the name is invalid, a task of that name is still
|
|
/// running, the tasks directory cannot be created, or the task file
|
|
/// cannot be written.
|
|
pub fn submit_task(cmd: String, timeout_secs: Option<u64>, name: Option<String>) -> Result<String> {
|
|
std::fs::create_dir_all(paths::tasks_dir())?;
|
|
let id = match name {
|
|
Some(name) => {
|
|
validate_task_name(&name)?;
|
|
if let Some(existing) = read_task(&name)
|
|
&& matches!(existing.status, TaskStatus::Pending | TaskStatus::Running)
|
|
{
|
|
bail!(
|
|
"a bash task named `{name}` is already running — wait for it to finish or pick another name"
|
|
);
|
|
}
|
|
name
|
|
}
|
|
None => new_task_id(),
|
|
};
|
|
let task = TaskFile {
|
|
id: id.clone(),
|
|
cmd,
|
|
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)?;
|
|
refresh_loose_ends();
|
|
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);
|
|
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");
|
|
}
|
|
refresh_loose_ends();
|
|
send_wake(socket, &id, "interrupted (daemon restarted)", None).await;
|
|
}
|
|
}
|
|
|
|
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");
|
|
}
|
|
refresh_loose_ends();
|
|
// Best-effort: tally the normalised command head for the /stats
|
|
// "favorite tools" view. Counted once per execution, regardless of
|
|
// exit status. Never fails the task.
|
|
crate::stats::record_command(&task.cmd);
|
|
|
|
let out_path = paths::task_out(&id);
|
|
let err_path = paths::task_err(&id);
|
|
|
|
let (timed_out, exit_code) =
|
|
match exec_cmd(&task.cmd, &out_path, &err_path, task.timeout_secs).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");
|
|
}
|
|
refresh_loose_ends();
|
|
|
|
let summary = if timed_out {
|
|
let secs = task.timeout_secs.unwrap_or(0);
|
|
format!("timed out after {secs}s")
|
|
} 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)`.
|
|
/// `timeout_secs = None` means no timeout — run until natural exit.
|
|
async fn exec_cmd(
|
|
cmd: &str,
|
|
out_path: &Path,
|
|
err_path: &Path,
|
|
timeout_secs: Option<u64>,
|
|
) -> Result<(i32, bool)> {
|
|
use tokio::process::Command;
|
|
// SAFETY: `nice` is async-signal-safe and modifies only the calling
|
|
// process's scheduling priority before exec. No allocations, no locks.
|
|
let mut cmd_builder = Command::new("sh");
|
|
cmd_builder
|
|
.arg("-c")
|
|
.arg(cmd)
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped());
|
|
unsafe {
|
|
cmd_builder.pre_exec(|| {
|
|
libc::nice(10);
|
|
Ok(())
|
|
});
|
|
}
|
|
let mut child = cmd_builder.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,
|
|
));
|
|
|
|
if let Some(secs) = timeout_secs {
|
|
match tokio::time::timeout(Duration::from_secs(secs), 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))
|
|
}
|
|
}
|
|
} else {
|
|
// No timeout — wait for natural exit.
|
|
match child.wait().await {
|
|
Ok(status) => {
|
|
let _ = copy_out.await;
|
|
let _ = copy_err.await;
|
|
Ok((status.code().unwrap_or(-1), false))
|
|
}
|
|
Err(e) => Err(e.into()),
|
|
}
|
|
}
|
|
}
|
|
|
|
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)>,
|
|
) {
|
|
use tokio::io::{AsyncBufReadExt as _, BufReader};
|
|
use tokio::net::UnixStream;
|
|
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,
|
|
};
|
|
|
|
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");
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{MAX_TASK_NAME_LEN, validate_task_name};
|
|
|
|
#[test]
|
|
fn accepts_reasonable_names() {
|
|
for ok in ["build", "ci-check", "nix_flake.check", "t1", "A.B-C_9"] {
|
|
assert!(validate_task_name(ok).is_ok(), "{ok} should be valid");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_unsafe_names() {
|
|
// Empty, traversal, separators, control/space, and over-long.
|
|
for bad in [
|
|
"",
|
|
".",
|
|
"..",
|
|
"a/b",
|
|
"../escape",
|
|
"has space",
|
|
"tab\tname",
|
|
"slash\\back",
|
|
] {
|
|
assert!(
|
|
validate_task_name(bad).is_err(),
|
|
"{bad:?} should be rejected"
|
|
);
|
|
}
|
|
assert!(validate_task_name(&"x".repeat(MAX_TASK_NAME_LEN + 1)).is_err());
|
|
// Exactly at the cap is allowed.
|
|
assert!(validate_task_name(&"x".repeat(MAX_TASK_NAME_LEN)).is_ok());
|
|
}
|
|
}
|