fadcac8479
copy_stream_to_file dropped the copy and flush errors, so an ENOSPC or pipe error left a truncated .out/.err with no trace while the task still reported its exit status. It now returns the error and the runner logs it with the stream and file path.
1280 lines
51 KiB
Rust
1280 lines
51 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.
|
|
//!
|
|
//! At daemon startup, before that interrupted-marking pass, any file under
|
|
//! `tasks_dir()` older than 30 days is deleted outright. Nothing swept
|
|
//! this directory before, and it had grown without bound, repeatedly
|
|
//! holding stale credentials in captured output.
|
|
//!
|
|
//! 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 hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups};
|
|
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());
|
|
if let Err(e) = write_task(&task) {
|
|
tracing::warn!(id, error = ?e, "kill: could not mark pending task killed");
|
|
return (false, false);
|
|
}
|
|
// 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, cgroups: Option<RunCgroups>) {
|
|
let cgroups = Arc::new(cgroups);
|
|
tokio::spawn(async move {
|
|
run_loop(socket, cgroups).await;
|
|
});
|
|
}
|
|
|
|
async fn run_loop(socket: PathBuf, cgroups: Arc<Option<RunCgroups>>) {
|
|
if let Err(e) = std::fs::create_dir_all(paths::tasks_dir()) {
|
|
tracing::warn!(error = ?e, "bash_runner: create tasks dir failed");
|
|
}
|
|
let removed = sweep_old_tasks(&paths::tasks_dir(), SystemTime::now() - RETENTION);
|
|
if removed > 0 {
|
|
tracing::info!(removed, "bash_runner: swept old task files at startup");
|
|
}
|
|
mark_interrupted(&socket).await;
|
|
|
|
let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
|
|
|
|
loop {
|
|
poll_once(&socket, &claimed, &cgroups);
|
|
tokio::time::sleep(POLL_INTERVAL).await;
|
|
}
|
|
}
|
|
|
|
/// How long a task's files are kept before [`sweep_old_tasks`] removes
|
|
/// them. `harness/bash-tasks/` grew without bound (and had repeatedly
|
|
/// held stale credentials in captured output) because nothing ever swept
|
|
/// it. Mara: "build automatic cleanup at daemon start to clear older
|
|
/// than 30d" — a flat age cutoff, no count-based retention knob.
|
|
const RETENTION: Duration = Duration::from_hours(720);
|
|
|
|
/// Delete every regular file directly under `dir` whose mtime is older
|
|
/// than `cutoff`. Runs once, at daemon startup (`run_loop`), before
|
|
/// [`mark_interrupted`] — an ancient still-`running` task is swept away
|
|
/// outright rather than given a fresh "interrupted" todo nobody's going
|
|
/// to act on 30 days later.
|
|
///
|
|
/// Whole-directory, extension-agnostic rather than `.json`-id-driven like
|
|
/// [`mark_interrupted`]/[`poll_once`]: a task's `.json`/`.out`/`.err`
|
|
/// triple is written together at completion (the same instant, well
|
|
/// inside the 30-day granularity this cares about), so sweeping
|
|
/// file-by-file on its own mtime can't split a still-referenced group —
|
|
/// and this shape also reaps orphaned `.out`/`.err` (a crash between
|
|
/// writing output and the final `.json` write) and stray `.json.tmp`
|
|
/// scratch files (`write_task`'s tmp+rename can leave one behind on a
|
|
/// crash mid-rename) that an id-driven walk would miss entirely.
|
|
/// Directories are skipped (`tasks_dir()` holds none today, but a future
|
|
/// entry shouldn't be treated as sweepable file content). Best-effort: a
|
|
/// per-entry `remove_file` failure is logged and the pass continues
|
|
/// rather than aborting on the first one. Returns the count actually
|
|
/// removed.
|
|
fn sweep_old_tasks(dir: &Path, cutoff: SystemTime) -> usize {
|
|
let Ok(rd) = std::fs::read_dir(dir) else {
|
|
return 0;
|
|
};
|
|
let mut removed = 0;
|
|
for entry in rd.flatten() {
|
|
let Ok(meta) = entry.metadata() else { continue };
|
|
if !meta.is_file() {
|
|
continue;
|
|
}
|
|
let Ok(mtime) = meta.modified() else { continue };
|
|
if mtime >= cutoff {
|
|
continue;
|
|
}
|
|
let path = entry.path();
|
|
match std::fs::remove_file(&path) {
|
|
Ok(()) => removed += 1,
|
|
Err(e) => {
|
|
tracing::warn!(path = %path.display(), error = ?e, "bash_runner: sweep remove failed");
|
|
}
|
|
}
|
|
}
|
|
removed
|
|
}
|
|
|
|
/// 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>>>,
|
|
cgroups: &Arc<Option<RunCgroups>>,
|
|
) {
|
|
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();
|
|
let cgroups = Arc::clone(cgroups);
|
|
tokio::spawn(async move {
|
|
run_task(task, &socket, cgroups.as_ref().as_ref()).await;
|
|
claimed.lock().unwrap().remove(&id);
|
|
});
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Task execution
|
|
// ---------------------------------------------------------------------------
|
|
|
|
async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups>) {
|
|
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");
|
|
}
|
|
// No command text in the summary: `id` is already the human-chosen
|
|
// `name` when the caller passed one, and even the auto-generated
|
|
// hex id was already returned to the agent in the `run` tool's own
|
|
// result — the todo only needs to say which task, not repeat the
|
|
// command. An unbounded raw command (e.g. a multi-line heredoc
|
|
// script) would balloon every UI that renders todo summaries; the
|
|
// full command stays on disk in the task file for `status`/`view`
|
|
// instead.
|
|
upsert_bash_todo(socket, &id, format!("bash task `{id}` running")).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 cgroup = cgroups.and_then(|cgroups| task_cgroup(cgroups, &id));
|
|
let outcome = exec_cmd(
|
|
&task.cmd,
|
|
&out_path,
|
|
&err_path,
|
|
task.timeout_secs,
|
|
&cancel,
|
|
&force,
|
|
cgroup.as_ref().map(|(_, procs)| procs),
|
|
)
|
|
.await;
|
|
drop(cgroup);
|
|
|
|
// 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;
|
|
}
|
|
|
|
/// `id`'s cgroup and its open `cgroup.procs`, or `None` (logged) when either
|
|
/// fails, in which case the task runs in the daemon's cgroup.
|
|
fn task_cgroup(cgroups: &RunCgroups, id: &str) -> Option<(RunCgroup, std::fs::File)> {
|
|
let opened = cgroups.create(id).and_then(|cgroup| {
|
|
let procs = cgroup.procs()?;
|
|
Ok((cgroup, procs))
|
|
});
|
|
opened
|
|
.inspect_err(|e| {
|
|
tracing::warn!(id = %id, error = %e, "bash_runner: task cgroup unavailable");
|
|
})
|
|
.ok()
|
|
}
|
|
|
|
/// 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,
|
|
cgroup_procs: Option<&std::fs::File>,
|
|
) -> 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());
|
|
let cgroup_procs = cgroup_procs.map(std::os::fd::AsRawFd::as_raw_fd);
|
|
// SAFETY: `setpgid`/`nice`/`write` 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, …)`).
|
|
// Entering the task's cgroup before exec means nothing the task forks
|
|
// starts outside it.
|
|
unsafe {
|
|
cmd_builder.pre_exec(move || {
|
|
libc::setpgid(0, 0);
|
|
libc::nice(10);
|
|
if let Some(procs) = cgroup_procs {
|
|
run_cgroup::enter_in_pre_exec(procs)?;
|
|
}
|
|
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 copy_out = {
|
|
let path = out_path.to_path_buf();
|
|
tokio::spawn(
|
|
async move { copy_stream_to_file(tokio::io::BufReader::new(stdout), &path).await },
|
|
)
|
|
};
|
|
let copy_err = {
|
|
let path = err_path.to_path_buf();
|
|
tokio::spawn(
|
|
async move { copy_stream_to_file(tokio::io::BufReader::new(stderr), &path).await },
|
|
)
|
|
};
|
|
|
|
// 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);
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
warn_if_copy_failed("stdout", out_path, copy_out.await);
|
|
warn_if_copy_failed("stderr", err_path, 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)
|
|
}
|
|
|
|
/// Copies `reader` into a new file at `path`, flushing at the end. On error
|
|
/// the bytes copied before it stay in the file.
|
|
async fn copy_stream_to_file<R>(mut reader: R, path: &Path) -> std::io::Result<()>
|
|
where
|
|
R: tokio::io::AsyncRead + Unpin,
|
|
{
|
|
let mut f = tokio::fs::File::create(path).await?;
|
|
let copied = tokio::io::copy(&mut reader, &mut f).await;
|
|
f.flush().await?;
|
|
copied.map(|_| ())
|
|
}
|
|
|
|
/// Warns when a stream copy failed or its task panicked: the log file is then
|
|
/// incomplete while the task still reports its exit status.
|
|
fn warn_if_copy_failed(
|
|
stream: &str,
|
|
path: &Path,
|
|
joined: Result<std::io::Result<()>, tokio::task::JoinError>,
|
|
) {
|
|
let error = match joined {
|
|
Ok(Ok(())) => return,
|
|
Ok(Err(e)) => format!("{e:?}"),
|
|
Err(e) => format!("{e:?}"),
|
|
};
|
|
tracing::warn!(
|
|
stream,
|
|
path = %path.display(),
|
|
error,
|
|
"bash_runner: capturing task output failed, the log file is incomplete"
|
|
);
|
|
}
|
|
|
|
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");
|
|
});
|
|
}
|
|
|
|
// sweep_old_tasks: a flat, whole-directory age sweep.
|
|
// `cutoff` is passed in explicitly (mirroring `hive-agent`'s
|
|
// `has_fresh_credentials(dir, since, ..)` pattern) so these don't need
|
|
// to fake a real 30-day-old file — `set_modified` moves a fresh
|
|
// tempfile's mtime directly, no sleeping required.
|
|
|
|
#[test]
|
|
fn sweep_removes_only_files_older_than_cutoff() {
|
|
use super::sweep_old_tasks;
|
|
use std::time::{Duration, SystemTime};
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let old = dir.path().join("old.json");
|
|
let fresh = dir.path().join("fresh.json");
|
|
std::fs::write(&old, b"{}").unwrap();
|
|
std::fs::write(&fresh, b"{}").unwrap();
|
|
let now = SystemTime::now();
|
|
std::fs::File::open(&old)
|
|
.unwrap()
|
|
.set_modified(now - Duration::from_hours(744)) // 31 days
|
|
.unwrap();
|
|
std::fs::File::open(&fresh)
|
|
.unwrap()
|
|
.set_modified(now)
|
|
.unwrap();
|
|
|
|
let removed = sweep_old_tasks(dir.path(), now - Duration::from_hours(720)); // 30 days
|
|
|
|
assert_eq!(removed, 1);
|
|
assert!(!old.exists(), "the 31-day-old file must be gone");
|
|
assert!(fresh.exists(), "today's file must survive");
|
|
}
|
|
|
|
#[test]
|
|
fn sweep_keeps_a_file_exactly_at_the_cutoff() {
|
|
// Boundary: `mtime >= cutoff` is kept, matching `sweep_old_tasks`'s
|
|
// doc — a task's mtime is never used to justify deleting it on the
|
|
// very instant it crosses the line, only once it's strictly older.
|
|
use super::sweep_old_tasks;
|
|
use std::time::SystemTime;
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("boundary.json");
|
|
std::fs::write(&path, b"{}").unwrap();
|
|
let cutoff = SystemTime::now();
|
|
std::fs::File::open(&path)
|
|
.unwrap()
|
|
.set_modified(cutoff)
|
|
.unwrap();
|
|
|
|
let removed = sweep_old_tasks(dir.path(), cutoff);
|
|
|
|
assert_eq!(removed, 0);
|
|
assert!(path.exists());
|
|
}
|
|
|
|
#[test]
|
|
fn sweep_is_extension_agnostic_and_reaps_orphans_and_tmp_files() {
|
|
// Whole-directory by design (see the doc comment): an orphaned
|
|
// `.out` with no `.json` (crash mid-task) and a stray `.json.tmp`
|
|
// (crash mid-rename in `write_task`) are exactly the garbage an
|
|
// id-driven walk like `mark_interrupted`'s would miss.
|
|
use super::sweep_old_tasks;
|
|
use std::time::{Duration, SystemTime};
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let orphan_out = dir.path().join("orphan.out");
|
|
let stray_tmp = dir.path().join("abc123.json.tmp");
|
|
std::fs::write(&orphan_out, b"stdout").unwrap();
|
|
std::fs::write(&stray_tmp, b"{}").unwrap();
|
|
let past = SystemTime::now() - Duration::from_hours(1440); // 60 days
|
|
std::fs::File::open(&orphan_out)
|
|
.unwrap()
|
|
.set_modified(past)
|
|
.unwrap();
|
|
std::fs::File::open(&stray_tmp)
|
|
.unwrap()
|
|
.set_modified(past)
|
|
.unwrap();
|
|
|
|
let removed = sweep_old_tasks(dir.path(), SystemTime::now());
|
|
|
|
assert_eq!(removed, 2);
|
|
assert!(!orphan_out.exists());
|
|
assert!(!stray_tmp.exists());
|
|
}
|
|
|
|
#[test]
|
|
fn sweep_skips_subdirectories() {
|
|
// Defensive: `tasks_dir()` holds no directories today, but a
|
|
// future entry must not be treated as sweepable file content
|
|
// (`remove_file` on a directory just errors and is swallowed —
|
|
// this pins the *intent*, not merely today's accidental safety).
|
|
use super::sweep_old_tasks;
|
|
use std::time::{Duration, SystemTime};
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let sub = dir.path().join("a-subdir");
|
|
std::fs::create_dir(&sub).unwrap();
|
|
|
|
let removed = sweep_old_tasks(dir.path(), SystemTime::now() + Duration::from_secs(1));
|
|
|
|
assert_eq!(removed, 0);
|
|
assert!(sub.exists());
|
|
}
|
|
|
|
#[test]
|
|
fn sweep_missing_dir_is_a_noop() {
|
|
// Defensive, same shape as `has_fresh_credentials_missing_dir_is_false`
|
|
// in `hive-agent`: a not-yet-created tasks dir must not panic.
|
|
use super::sweep_old_tasks;
|
|
use std::time::SystemTime;
|
|
let missing = tempfile::tempdir()
|
|
.unwrap()
|
|
.path()
|
|
.join("never-created-subdir");
|
|
assert_eq!(sweep_old_tasks(&missing, SystemTime::now()), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn kill_pending_reports_failure_when_the_write_fails() {
|
|
use super::{kill_task, read_task, write_task};
|
|
use crate::protocol::{TaskFile, TaskStatus};
|
|
with_harness_dir(|| {
|
|
let harness = tempfile::tempdir().unwrap();
|
|
// SAFETY: serialised by the lock in `with_harness_dir`, which
|
|
// restores the previous value when the closure returns.
|
|
unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", harness.path()) };
|
|
std::fs::create_dir_all(crate::paths::tasks_dir()).unwrap();
|
|
let task = TaskFile {
|
|
id: "t1".to_owned(),
|
|
cmd: "echo hi".to_owned(),
|
|
timeout_secs: None,
|
|
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).unwrap();
|
|
// A directory at the temp path makes `write_task` fail even as root.
|
|
std::fs::create_dir(crate::paths::task_json("t1").with_extension("json.tmp")).unwrap();
|
|
|
|
assert_eq!(kill_task("t1", false), (false, false));
|
|
assert_eq!(read_task("t1").unwrap().status, TaskStatus::Pending);
|
|
});
|
|
}
|
|
|
|
/// Yields `data`, then fails with an I/O error.
|
|
struct FailAfter {
|
|
data: &'static [u8],
|
|
sent: bool,
|
|
}
|
|
|
|
impl tokio::io::AsyncRead for FailAfter {
|
|
fn poll_read(
|
|
mut self: std::pin::Pin<&mut Self>,
|
|
_cx: &mut std::task::Context<'_>,
|
|
buf: &mut tokio::io::ReadBuf<'_>,
|
|
) -> std::task::Poll<std::io::Result<()>> {
|
|
if self.sent {
|
|
return std::task::Poll::Ready(Err(std::io::Error::other("pipe broke")));
|
|
}
|
|
self.sent = true;
|
|
buf.put_slice(self.data);
|
|
std::task::Poll::Ready(Ok(()))
|
|
}
|
|
}
|
|
|
|
fn block_on<F: std::future::Future>(f: F) -> F::Output {
|
|
tokio::runtime::Builder::new_current_thread()
|
|
.enable_all()
|
|
.build()
|
|
.unwrap()
|
|
.block_on(f)
|
|
}
|
|
|
|
#[test]
|
|
fn copy_error_is_returned_and_earlier_bytes_are_kept() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("t1.out");
|
|
let reader = FailAfter {
|
|
data: b"before",
|
|
sent: false,
|
|
};
|
|
let res = block_on(super::copy_stream_to_file(reader, &path));
|
|
assert_eq!(res.unwrap_err().to_string(), "pipe broke");
|
|
assert_eq!(std::fs::read(&path).unwrap(), b"before");
|
|
}
|
|
|
|
#[test]
|
|
fn successful_copy_writes_everything() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("t1.out");
|
|
let res = block_on(super::copy_stream_to_file(&b"all of it"[..], &path));
|
|
assert!(res.is_ok());
|
|
assert_eq!(std::fs::read(&path).unwrap(), b"all of it");
|
|
}
|
|
}
|