967 lines
38 KiB
Rust
967 lines
38 KiB
Rust
//! Bash subprocess runner: spawns `bash -c <cmd>` tasks, writes status
|
|
//! files under `harness/bash-tasks/`, and surfaces task state to the agent
|
|
//! as *todos* on the harness's in-agent socket (loose-ends v2).
|
|
//!
|
|
//! 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)
|
|
//!
|
|
//! Todos (loose-ends v2): one keyed todo (`key = task id`) tracks the task
|
|
//! across its lifetime — upserted with a "running" summary at start, then
|
|
//! upserted again with the completion summary when it finishes. That summary
|
|
//! change signals the harness turn loop the same way the old completion wake
|
|
//! did; the agent clears the todo with `mark_todo_done` once it has read the
|
|
//! output.
|
|
//!
|
|
//! Tasks with status `running` on daemon boot are marked `interrupted`
|
|
//! (the process died with the previous daemon). The todo is still updated to
|
|
//! its interrupted-done summary 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::{HashMap, HashSet};
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
|
use std::sync::{Arc, Mutex, OnceLock};
|
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
|
|
|
use anyhow::{Result, bail};
|
|
use hive_agent_sock::Request as TodoReq;
|
|
use tokio::io::AsyncWriteExt as _;
|
|
use tokio::sync::Notify;
|
|
|
|
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);
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Running-task registry (for on-demand kill)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Handle to a currently-executing task, used by [`kill_task`] to signal it.
|
|
/// `exec_cmd` selects on `cancel`; when notified it signals the task's
|
|
/// process group with `SIGKILL` (if `force`) or `SIGINT`.
|
|
struct RunningHandle {
|
|
cancel: Arc<Notify>,
|
|
force: Arc<AtomicBool>,
|
|
}
|
|
|
|
/// Global registry of running tasks (id → handle). Populated by `run_task`
|
|
/// for the duration of execution and read by [`kill_task`] from the MCP
|
|
/// tool-call handler — a separate async context from the runner loop, so
|
|
/// a shared global (rather than the loop-local `claimed` set) is needed.
|
|
fn running() -> &'static Mutex<HashMap<String, RunningHandle>> {
|
|
static RUNNING: OnceLock<Mutex<HashMap<String, RunningHandle>>> = OnceLock::new();
|
|
RUNNING.get_or_init(|| Mutex::new(HashMap::new()))
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Inline-waiter registry: skip the completion todo when a caller is
|
|
// currently (or was, until moments ago) synchronously polling the task's
|
|
// terminal state via `wait_seconds` on `BashRun` or `BashStatus` — the tool
|
|
// response already delivers the result in that same turn, so a follow-up
|
|
// todo would just be a redundant duplicate of information the agent has.
|
|
//
|
|
// Superseded a one-shot flag set by the waiter *after* observing terminal
|
|
// state and checked by the runner on its own independent poll schedule —
|
|
// two independently timed reads of *different* state (task file vs. flag)
|
|
// can't be made race-free by reordering, only by sharing a lock. This
|
|
// shape does: both sides check/mutate **one registry** under **one lock**,
|
|
// deciding on presence rather than a flag that might be set too late. A
|
|
// still-registered waiter's own poll loop is guaranteed to see the
|
|
// just-written terminal file on its next iteration — same file, not a
|
|
// message that could be missed — so skipping its todo can't strand the
|
|
// agent. See the forge issue tracker for the full before/after trace.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Refcounted, not a plain set: two concurrent inline waiters on the same
|
|
/// `id` are possible (e.g. `run` and a separate `status` call racing each
|
|
/// other), and a plain `HashSet` would let the *first* one to drop
|
|
/// deregister the id out from under the second, still-live one. Counting
|
|
/// means presence only goes to zero once every registered guard has
|
|
/// dropped. In-memory only — a daemon restart wipes it, which is fine: any
|
|
/// waiter registered here belonged to an in-flight MCP call the restart
|
|
/// also tore down, and a task still `running` across a restart is marked
|
|
/// `interrupted` on boot (see module docs) and gets its own fresh todo,
|
|
/// independent of this map.
|
|
fn waiting_ids() -> &'static Mutex<HashMap<String, u32>> {
|
|
static WAITING: OnceLock<Mutex<HashMap<String, u32>>> = OnceLock::new();
|
|
WAITING.get_or_init(|| Mutex::new(HashMap::new()))
|
|
}
|
|
|
|
/// True if an inline waiter is currently registered for `id`. Checked by
|
|
/// `run_task`'s completion handler under the same lock [`WaiterGuard`]
|
|
/// (de)registers under — "is a waiter here" and "a waiter leaving" can
|
|
/// never observe torn state relative to each other, unlike the flag this
|
|
/// replaced.
|
|
fn waiter_present(id: &str) -> bool {
|
|
waiting_ids().lock().unwrap().contains_key(id)
|
|
}
|
|
|
|
/// RAII registration for one `wait_for_task` call. Increments `id`'s count
|
|
/// on construction, decrements (removing the entry at zero) on drop —
|
|
/// covers every `wait_for_task` return path (terminal found, deadline hit,
|
|
/// task vanished) with one code path instead of duplicating the removal at
|
|
/// each `return`. Rust drops locals after the return expression is
|
|
/// evaluated, so the guard stays registered through `wait_for_task`'s very
|
|
/// last read of the task file and only deregisters right before the value
|
|
/// actually returns to the caller.
|
|
struct WaiterGuard<'a>(&'a str);
|
|
|
|
impl<'a> WaiterGuard<'a> {
|
|
fn new(id: &'a str) -> Self {
|
|
*waiting_ids()
|
|
.lock()
|
|
.unwrap()
|
|
.entry(id.to_owned())
|
|
.or_insert(0) += 1;
|
|
Self(id)
|
|
}
|
|
}
|
|
|
|
impl Drop for WaiterGuard<'_> {
|
|
fn drop(&mut self) {
|
|
let mut waiting = waiting_ids().lock().unwrap();
|
|
if let Some(count) = waiting.get_mut(self.0) {
|
|
*count -= 1;
|
|
if *count == 0 {
|
|
waiting.remove(self.0);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Outcome of one `exec_cmd` run.
|
|
enum ExecOutcome {
|
|
/// Process exited on its own with this status code.
|
|
Exited(i32),
|
|
/// Killed by the per-task timeout (`timeout_secs`).
|
|
TimedOut,
|
|
/// Killed on request via [`kill_task`]. `forced` = SIGKILL vs SIGINT.
|
|
Killed { forced: bool },
|
|
}
|
|
|
|
/// Send `sig` to the task's process group. A negative pgid targets the whole
|
|
/// group, so `bash -c` plus any children it spawned all receive the signal.
|
|
/// No-op if the child had no pid (already reaped).
|
|
fn signal_group(pgid: Option<i32>, sig: i32) {
|
|
if let Some(pgid) = pgid {
|
|
// SAFETY: plain libc `kill(2)`; `-pgid` targets the child's own
|
|
// process group (it called `setpgid(0, 0)` so it leads its group).
|
|
unsafe {
|
|
libc::kill(-pgid, sig);
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// 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()
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Todo delivery (loose-ends v2)
|
|
// ---------------------------------------------------------------------------
|
|
//
|
|
// Bash-task state is surfaced to the agent as todos on the harness's
|
|
// in-agent socket (`HIVE_AGENT_SOCKET`), not as direct c0re wakes: one keyed
|
|
// todo (`key = task id`) tracks the task across its whole lifetime. It's
|
|
// upserted with a stable "running" summary when the task starts (an
|
|
// idempotent no-op that never re-wakes), then upserted again with the
|
|
// completion summary when it finishes — that summary change signals the
|
|
// harness turn loop exactly like the old completion wake did. The agent
|
|
// reads the output and clears the todo with `mark_todo_done`.
|
|
|
|
/// Send one todo request to the harness in-agent socket. Best-effort: any
|
|
/// connect / write error is logged and swallowed — the harness self-heals
|
|
/// on the next task transition, and a standalone daemon without the socket
|
|
/// simply has nowhere to push.
|
|
async fn send_todo(socket: &Path, req: &TodoReq) {
|
|
// Fail-fast rather than back off: the next task transition pushes the
|
|
// todo again, and a runner blocked in a retry schedule would delay the
|
|
// task bookkeeping behind it.
|
|
if let Err(e) = hive_sock_client::notify(socket, req, hive_sock_client::Retry::None).await {
|
|
tracing::warn!(error = ?e, socket = %socket.display(), "bash_runner: todo send failed");
|
|
}
|
|
}
|
|
|
|
/// Upsert the task's keyed todo (`key = id`) with `summary`. Used for both
|
|
/// the "running" surface at start and the "done" summary at completion — a
|
|
/// changed summary on the same row signals the turn loop, an unchanged
|
|
/// re-push is an idempotent no-op.
|
|
async fn upsert_bash_todo(socket: &Path, id: &str, summary: String) {
|
|
send_todo(
|
|
socket,
|
|
&TodoReq::UpsertTodo {
|
|
subsystem: "bash".to_owned(),
|
|
key: Some(id.to_owned()),
|
|
summary,
|
|
source: None,
|
|
reopen_if_acked: false,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
|
|
/// Clear a task's keyed todo — used when an inline `status`/`run` wait
|
|
/// already delivered the terminal result, so no `done` todo is warranted.
|
|
async fn clear_bash_todo(socket: &Path, id: &str) {
|
|
send_todo(
|
|
socket,
|
|
&TodoReq::ClearTodo {
|
|
subsystem: "bash".to_owned(),
|
|
key: Some(id.to_owned()),
|
|
all: false,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Public API used by daemon dispatch
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// 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 valid
|
|
/// [`hive_types::Ident`] — a single safe path segment of `[a-z0-9-]`,
|
|
/// non-empty and ≤63 bytes. Delegating to `Ident` also rejects `.`/`..`
|
|
/// traversal and the `.json.tmp` scratch suffix (both contain `.`), and
|
|
/// makes the derived `bash-task-<name>` wake sender a valid identifier.
|
|
fn validate_task_name(name: &str) -> Result<()> {
|
|
hive_types::Ident::parse(name)
|
|
.map(|_| ())
|
|
.map_err(|e| anyhow::anyhow!("invalid task name {name:?}: {e}"))
|
|
}
|
|
|
|
/// 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: chrono::Utc::now(),
|
|
started_at: None,
|
|
completed_at: None,
|
|
exit_code: None,
|
|
stdout_tail: None,
|
|
stderr_tail: None,
|
|
};
|
|
write_task(&task)?;
|
|
// No todo yet: the keyed "active" todo is upserted when the runner
|
|
// loop flips the task to Running (`run_task`), which has the in-agent
|
|
// socket handle. Surfacing a Pending task would only race the ~200ms
|
|
// until it starts and double-wake (pending→running).
|
|
Ok(id)
|
|
}
|
|
|
|
/// Inline wait: poll `read_task(id)` until terminal state or deadline.
|
|
/// Returns the final task on success, or `None` if it never completed.
|
|
///
|
|
/// Registered as an inline waiter for `id` (via [`WaiterGuard`]) for the
|
|
/// polling loop only — deliberately **not** across the final fallback read
|
|
/// below. A guard still held during that last read would let
|
|
/// `run_task`'s completion handler see "waiter present" and skip the
|
|
/// todo for a result this call already fixed (non-terminal, from the read
|
|
/// that just lost the deadline race) — a genuinely dropped notification,
|
|
/// not a redundant one. Dropping the guard first means the worst case if
|
|
/// `run_task` completes in the gap between the guard dropping and this
|
|
/// read running is a harmless extra todo (the pre-existing, documented
|
|
/// tolerance), never a missed one.
|
|
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);
|
|
{
|
|
let _waiting = WaiterGuard::new(id);
|
|
loop {
|
|
match read_task(id) {
|
|
None => break,
|
|
Some(task) => {
|
|
if matches!(
|
|
task.status,
|
|
TaskStatus::Done
|
|
| TaskStatus::TimedOut
|
|
| TaskStatus::Interrupted
|
|
| TaskStatus::Killed
|
|
) {
|
|
return Some(task);
|
|
}
|
|
}
|
|
}
|
|
if tokio::time::Instant::now() >= deadline {
|
|
break;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(POLL_MS)).await;
|
|
}
|
|
} // guard dropped before the final read below
|
|
read_task(id)
|
|
}
|
|
|
|
/// Kill a running or still-pending task.
|
|
///
|
|
/// Fire-and-forget: this **sends a signal and returns** — it does not wait
|
|
/// for the process to actually exit.
|
|
///
|
|
/// - **Running** (in the registry): signals its process group via the
|
|
/// `exec_cmd` cancel path — `SIGKILL` if `force`, else `SIGINT`. If the
|
|
/// process ignores `SIGINT`, the caller re-invokes with `force` to send
|
|
/// `SIGKILL` (no daemon-side auto-escalation). The task transitions to
|
|
/// [`TaskStatus::Killed`] once the process exits and a completion wake fires.
|
|
/// - **Pending** (queued, not yet started): marked `Killed` directly so the
|
|
/// runner loop never starts it. No process exists yet, so `force` is moot.
|
|
/// - **Terminal or unknown id**: no-op.
|
|
///
|
|
/// Returns `(killed, was_running)`: `killed` = a kill was issued (signal sent
|
|
/// or pending task cancelled); `was_running` = the task was actively running.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the running-task registry mutex is poisoned (a prior holder
|
|
/// panicked) — unrecoverable, consistent with the rest of the daemon's
|
|
/// `Mutex` usage.
|
|
#[must_use]
|
|
pub fn kill_task(id: &str, force: bool) -> (bool, bool) {
|
|
// Running task: signal via the registry.
|
|
{
|
|
let guard = running().lock().unwrap();
|
|
if let Some(handle) = guard.get(id) {
|
|
handle.force.store(force, Ordering::SeqCst);
|
|
handle.cancel.notify_one();
|
|
return (true, true);
|
|
}
|
|
}
|
|
// Pending task: mark Killed so the poll loop never starts it.
|
|
if let Some(mut task) = read_task(id)
|
|
&& task.status == TaskStatus::Pending
|
|
{
|
|
task.status = TaskStatus::Killed;
|
|
task.completed_at = Some(chrono::Utc::now());
|
|
let _ = write_task(&task);
|
|
// A Pending task never reached Running, so it has no keyed "active"
|
|
// todo to clear and (like the old code) fires no completion signal —
|
|
// the caller that killed it already knows.
|
|
return (true, false);
|
|
}
|
|
(false, false)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Runner background loop
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Spawn the background runner loop as a detached tokio task. `socket` is
|
|
/// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task
|
|
/// transitions are pushed as todos. 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(chrono::Utc::now());
|
|
if let Err(e) = write_task(&task) {
|
|
tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
|
|
}
|
|
upsert_bash_todo(
|
|
socket,
|
|
&id,
|
|
done_summary(&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(chrono::Utc::now());
|
|
if let Err(e) = write_task(&task) {
|
|
tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed");
|
|
}
|
|
upsert_bash_todo(
|
|
socket,
|
|
&id,
|
|
format!("bash task `{id}` running: `{}`", task.cmd),
|
|
)
|
|
.await;
|
|
// 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);
|
|
|
|
// Register a kill handle for the duration of execution so `kill_task`
|
|
// (called from the MCP tool-call handler) can signal this task.
|
|
let cancel = Arc::new(Notify::new());
|
|
let force = Arc::new(AtomicBool::new(false));
|
|
running().lock().unwrap().insert(
|
|
id.clone(),
|
|
RunningHandle {
|
|
cancel: cancel.clone(),
|
|
force: force.clone(),
|
|
},
|
|
);
|
|
|
|
let outcome = exec_cmd(
|
|
&task.cmd,
|
|
&out_path,
|
|
&err_path,
|
|
task.timeout_secs,
|
|
&cancel,
|
|
&force,
|
|
)
|
|
.await;
|
|
|
|
// Deregister before writing terminal state — the task is no longer killable.
|
|
running().lock().unwrap().remove(&id);
|
|
|
|
let stdout_tail = tail_file(&out_path, SUMMARY_BYTES);
|
|
let stderr_tail = tail_file(&err_path, SUMMARY_BYTES);
|
|
|
|
let (status, exit_code, summary) = match outcome {
|
|
Ok(ExecOutcome::Exited(code)) => (TaskStatus::Done, Some(code), format!("exit={code}")),
|
|
Ok(ExecOutcome::TimedOut) => {
|
|
tracing::warn!(id = %id, "bash_runner: task timed out");
|
|
let secs = task.timeout_secs.unwrap_or(0);
|
|
(
|
|
TaskStatus::TimedOut,
|
|
None,
|
|
format!("timed out after {secs}s"),
|
|
)
|
|
}
|
|
Ok(ExecOutcome::Killed { forced }) => {
|
|
let sig = if forced { "SIGKILL" } else { "SIGINT" };
|
|
tracing::warn!(id = %id, sig, "bash_runner: task killed");
|
|
(TaskStatus::Killed, None, format!("killed ({sig})"))
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(id = %id, error = ?e, "bash_runner: exec error");
|
|
(TaskStatus::Done, None, "exec error".to_owned())
|
|
}
|
|
};
|
|
|
|
task.status = status;
|
|
task.completed_at = Some(chrono::Utc::now());
|
|
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");
|
|
}
|
|
// If an inline `status`/`run` wait is currently registered for this
|
|
// task (see `WaiterGuard`), that call's own poll loop is guaranteed to
|
|
// observe the terminal file just written above on its very next
|
|
// iteration — same file, not a message that could be missed — so
|
|
// there's nothing left for a todo to surface: retire the keyed todo
|
|
// without a `done` upsert instead. `waiter_present` and
|
|
// `WaiterGuard`'s (de)registration share one lock, so this can't race
|
|
// the way the old flag-based check could (see the registry's module
|
|
// doc for the full before/after).
|
|
if waiter_present(&id) {
|
|
clear_bash_todo(socket, &id).await;
|
|
tracing::debug!(id = %id, "bash_runner: done todo suppressed (inline waiter registered)");
|
|
return;
|
|
}
|
|
|
|
// Transition the SAME keyed todo from "running" to its done summary: the
|
|
// summary change signals the turn loop, and the agent clears the todo
|
|
// with `mark_todo_done` after reading the output. Pass only whether each
|
|
// stream produced (trimmed) output — the summary carries a `Read(<path>)`
|
|
// pointer to the captured `.out`/`.err` files rather than inlining the tail.
|
|
let has_stdout = !stdout_tail.as_deref().unwrap_or("").trim().is_empty();
|
|
let has_stderr = !stderr_tail.as_deref().unwrap_or("").trim().is_empty();
|
|
upsert_bash_todo(
|
|
socket,
|
|
&id,
|
|
done_summary(&id, &summary, Some((has_stdout, has_stderr))),
|
|
)
|
|
.await;
|
|
}
|
|
|
|
/// Run `bash -c cmd` in its own process group, streaming output to files.
|
|
/// `timeout_secs = None` means no timeout — run until natural exit.
|
|
///
|
|
/// Tasks run under `bash`, not `sh` (on NixOS `/bin/sh` is bash in POSIX
|
|
/// mode, which disables bashisms — arrays, `[[ … ]]`, `local`, process
|
|
/// substitution), invoked via `/usr/bin/env bash`.
|
|
///
|
|
/// The invocation is prefixed with `set -o pipefail` so a pipeline's exit
|
|
/// status reflects its last *failing* stage, not just its last stage — a
|
|
/// piped `cargo build 2>&1 | tail -40 && cargo test ...` chain silently
|
|
/// reports success off `tail`'s exit code otherwise, masking a real build
|
|
/// failure from every downstream `&&`/exit-code check. `-e`/`-u`/`-x` are
|
|
/// deliberately NOT forced on: unlike `pipefail` (a pure exit-status
|
|
/// reporting fix), each of those changes control flow or output volume in
|
|
/// ways plenty of ad-hoc one-liners don't expect — an opt-in a task's own
|
|
/// command string can request, not a safe default for every task.
|
|
///
|
|
/// The task is cancellable via `cancel` (set `force` first): on cancel the
|
|
/// whole process group is signalled — `SIGKILL` if `force`, else `SIGINT` —
|
|
/// so children of the shell die too, not just the shell. The timeout path
|
|
/// likewise `SIGKILL`s the group.
|
|
async fn exec_cmd(
|
|
cmd: &str,
|
|
out_path: &Path,
|
|
err_path: &Path,
|
|
timeout_secs: Option<u64>,
|
|
cancel: &Notify,
|
|
force: &AtomicBool,
|
|
) -> Result<ExecOutcome> {
|
|
use tokio::process::Command;
|
|
// `/usr/bin/env bash` rather than a bare `bash`: `/usr/bin/env` is at a
|
|
// fixed absolute path (coreutils, present on NixOS), and it resolves
|
|
// `bash` via PATH — the same controlled PATH the daemon's systemd unit
|
|
// sets. Avoids hardcoding a nix store / generation path in the binary.
|
|
let mut cmd_builder = Command::new("/usr/bin/env");
|
|
cmd_builder
|
|
.arg("bash")
|
|
.arg("-c")
|
|
.arg(format!("set -o pipefail; {cmd}"))
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped());
|
|
// SAFETY: `setpgid`/`nice` are async-signal-safe and only touch the
|
|
// calling (child) process before exec — no allocations, no locks.
|
|
// `setpgid(0, 0)` makes the child its own process-group leader so the
|
|
// whole subtree can be signalled as a group (`kill(-pgid, …)`).
|
|
unsafe {
|
|
cmd_builder.pre_exec(|| {
|
|
libc::setpgid(0, 0);
|
|
libc::nice(10);
|
|
Ok(())
|
|
});
|
|
}
|
|
let mut child = cmd_builder.spawn()?;
|
|
// With `setpgid(0, 0)` the child's pgid equals its pid.
|
|
let pgid = child.id().map(u32_to_i32);
|
|
|
|
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,
|
|
));
|
|
|
|
// A future that fires when the timeout elapses, or never if `None`.
|
|
let timeout_fut = async {
|
|
match timeout_secs {
|
|
Some(secs) => tokio::time::sleep(Duration::from_secs(secs)).await,
|
|
None => std::future::pending::<()>().await,
|
|
}
|
|
};
|
|
|
|
let mut timed_out = false;
|
|
let mut killed: Option<bool> = None;
|
|
let wait_res = {
|
|
let wait = child.wait();
|
|
tokio::pin!(wait);
|
|
tokio::pin!(timeout_fut);
|
|
loop {
|
|
tokio::select! {
|
|
res = &mut wait => break res,
|
|
() = &mut timeout_fut => {
|
|
// Per-task timeout: hard-kill the whole group and reap.
|
|
signal_group(pgid, libc::SIGKILL);
|
|
timed_out = true;
|
|
break (&mut wait).await;
|
|
}
|
|
() = cancel.notified() => {
|
|
// A kill was requested: send the signal and keep waiting.
|
|
// `kill` is fire-and-forget — we do NOT block-for-exit or
|
|
// auto-escalate. If the process ignores SIGINT the caller
|
|
// re-invokes kill with `force`, which re-fires this arm
|
|
// with SIGKILL. `killed` tracks the last signal sent.
|
|
let forced = force.load(Ordering::SeqCst);
|
|
signal_group(pgid, if forced { libc::SIGKILL } else { libc::SIGINT });
|
|
killed = Some(forced);
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
let _ = copy_out.await;
|
|
let _ = copy_err.await;
|
|
|
|
let status = wait_res?;
|
|
let outcome = if let Some(forced) = killed {
|
|
ExecOutcome::Killed { forced }
|
|
} else if timed_out {
|
|
ExecOutcome::TimedOut
|
|
} else {
|
|
ExecOutcome::Exited(status.code().unwrap_or(-1))
|
|
};
|
|
Ok(outcome)
|
|
}
|
|
|
|
/// `u32` pid → `i32` for `kill(2)`. Pids fit in `i32`; saturates defensively.
|
|
fn u32_to_i32(v: u32) -> i32 {
|
|
i32::try_from(v).unwrap_or(i32::MAX)
|
|
}
|
|
|
|
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())
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Completion summary
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Build the completion-summary line carried by a `done` todo: the status
|
|
/// `summary` plus `Read(<path>)` pointers to any captured output.
|
|
///
|
|
/// `output` is `(has_stdout, has_stderr)` — whether the task produced
|
|
/// non-empty stdout / stderr; `None` for the interrupted path, which has no
|
|
/// captured output. Deliberately NOT the output text: the todo gives the
|
|
/// agent the status + a `Read(<path>)` pointer to the captured files rather
|
|
/// than inlining the tail, which would flood the agent's context on every
|
|
/// task completion.
|
|
fn done_summary(id: &str, summary: &str, output: Option<(bool, bool)>) -> String {
|
|
use std::fmt::Write as _;
|
|
let mut body = format!("bash task `{id}` finished: {summary}");
|
|
if let Some((has_stdout, has_stderr)) = output
|
|
&& (has_stdout || has_stderr)
|
|
{
|
|
// Header + pointer order are keyed on `has_stderr`, not on the exit
|
|
// code: a failed command mid-chain (`cd bad-path && rm ...`) can
|
|
// exit 0 while the real evidence sits in `.err` — an exit-code
|
|
// trigger would filter out precisely the cases where nothing looks
|
|
// wrong. `.err`'s presence is already the scarce, meaningful signal
|
|
// the pointer is built on; keying the flag on the same condition
|
|
// costs nothing on the common quiet-success path (no stderr, no
|
|
// pointer, exactly like before) and fires on every case where
|
|
// something was written to stderr, including the ones the exit
|
|
// code can't be trusted to reveal.
|
|
if has_stderr {
|
|
body.push_str("\n\n⚠️ stderr present — read the full output:");
|
|
let _ = write!(
|
|
body,
|
|
"\n Read({}) # stderr",
|
|
crate::paths::task_err(id).display()
|
|
);
|
|
if has_stdout {
|
|
let _ = write!(body, "\n Read({})", crate::paths::task_out(id).display());
|
|
}
|
|
} else {
|
|
body.push_str("\n\noutput captured — read the full text with:");
|
|
let _ = write!(body, "\n Read({})", crate::paths::task_out(id).display());
|
|
}
|
|
}
|
|
body
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::validate_task_name;
|
|
use crate::test_util::with_harness_dir;
|
|
use hive_types::Ident;
|
|
|
|
#[test]
|
|
fn accepts_ident_names() {
|
|
for ok in ["build", "ci-check", "t1", "task-123", "abc"] {
|
|
assert!(validate_task_name(ok).is_ok(), "{ok} should be valid");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_non_ident_names() {
|
|
// Empty, traversal, separators, control/space, and — now that names
|
|
// must be a valid `Ident` ([a-z0-9-]) — uppercase, `.`, and `_` too.
|
|
for bad in [
|
|
"",
|
|
".",
|
|
"..",
|
|
"a/b",
|
|
"../escape",
|
|
"has space",
|
|
"tab\tname",
|
|
"slash\\back",
|
|
"Uppercase",
|
|
"under_score",
|
|
"dot.name",
|
|
] {
|
|
assert!(
|
|
validate_task_name(bad).is_err(),
|
|
"{bad:?} should be rejected"
|
|
);
|
|
}
|
|
assert!(validate_task_name(&"x".repeat(Ident::MAX_LEN + 1)).is_err());
|
|
// Exactly at the cap is allowed.
|
|
assert!(validate_task_name(&"x".repeat(Ident::MAX_LEN)).is_ok());
|
|
}
|
|
|
|
// The waiter registry is a single process-wide `Mutex<HashMap<..>>`
|
|
// refcount (see `waiting_ids()`), so these run serially against
|
|
// distinct ids to avoid cross-test interference under parallel test
|
|
// execution.
|
|
|
|
#[test]
|
|
fn waiter_guard_registers_and_deregisters_on_drop() {
|
|
use super::{WaiterGuard, waiter_present};
|
|
let id = "test-2905-guard-lifecycle";
|
|
assert!(!waiter_present(id), "unregistered id starts absent");
|
|
{
|
|
let _guard = WaiterGuard::new(id);
|
|
assert!(waiter_present(id), "present for the guard's lifetime");
|
|
}
|
|
assert!(!waiter_present(id), "guard drop deregisters");
|
|
}
|
|
|
|
#[test]
|
|
fn waiter_guard_stays_present_until_every_sibling_drops() {
|
|
use super::{WaiterGuard, waiter_present};
|
|
let id = "test-2905-guard-refcount";
|
|
let first = WaiterGuard::new(id);
|
|
let second = WaiterGuard::new(id); // simulates two concurrent inline waiters
|
|
assert!(waiter_present(id));
|
|
drop(first);
|
|
// Refcounted, not a plain set: a sibling guard's early drop must
|
|
// not un-register an id a still-live guard needs — `second` is
|
|
// still registered, so presence must survive `first` alone
|
|
// dropping.
|
|
assert!(
|
|
waiter_present(id),
|
|
"still present while a sibling guard is live"
|
|
);
|
|
drop(second);
|
|
assert!(!waiter_present(id), "absent once every guard has dropped");
|
|
}
|
|
|
|
// done_summary: the stderr-present branch must trigger on has_stderr
|
|
// alone, never on the exit code — see the comment on the function. A
|
|
// completed task with exit=0 and stderr present (a swallowed failure
|
|
// mid-chain) still needs to read as a flag, not as routine bookkeeping.
|
|
|
|
#[test]
|
|
fn done_summary_no_output_is_bare() {
|
|
use super::done_summary;
|
|
assert_eq!(
|
|
done_summary("t1", "exit=0", None),
|
|
"bash task `t1` finished: exit=0"
|
|
);
|
|
assert_eq!(
|
|
done_summary("t1", "exit=0", Some((false, false))),
|
|
"bash task `t1` finished: exit=0",
|
|
"no captured output on either stream ⇒ no pointer block at all"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn done_summary_stdout_only_is_unflagged() {
|
|
use super::done_summary;
|
|
with_harness_dir(|| {
|
|
let body = done_summary("t1", "exit=0", Some((true, false)));
|
|
assert!(body.contains("output captured — read the full text with:"));
|
|
assert!(body.contains("Read("));
|
|
assert!(body.contains(".out"));
|
|
assert!(!body.contains(".err"));
|
|
assert!(!body.contains('\u{26a0}'), "no flag without stderr");
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn done_summary_stderr_present_is_flagged_regardless_of_summary_text() {
|
|
use super::done_summary;
|
|
with_harness_dir(|| {
|
|
// exit=0 in the summary — a swallowed mid-chain failure that
|
|
// still exits clean. The flag must fire anyway, keyed on
|
|
// has_stderr, not on the summary text.
|
|
let body = done_summary("t1", "exit=0", Some((true, true)));
|
|
assert!(
|
|
body.contains("⚠️ stderr present"),
|
|
"stderr present must flag even on a clean exit code"
|
|
);
|
|
let err_pos = body.find(".err").expect("err pointer present");
|
|
let out_pos = body.find(".out").expect("out pointer present");
|
|
assert!(err_pos < out_pos, ".err must be listed before .out");
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn done_summary_stderr_only_omits_out_pointer() {
|
|
use super::done_summary;
|
|
with_harness_dir(|| {
|
|
let body = done_summary("t1", "exit=1", Some((false, true)));
|
|
assert!(body.contains(".err"));
|
|
assert!(!body.contains(".out"), "no stdout ⇒ no stdout pointer");
|
|
});
|
|
}
|
|
}
|