3059523172
done_summary previously only pointed at .out/.err without any visual distinction, so a completed task with clean-looking stdout and a nonzero exit still read as routine bookkeeping in the todo queue. Rather than gating a flag on the exit code, key it on has_stderr - a failed command mid-chain (cd bad-path && rm ...) can exit 0 while the real evidence sits in stderr, so an exit-code trigger would filter out precisely the cases where nothing looks wrong. .err's presence is already the scarce, meaningful signal the Read() pointer is built on; keying the flag on the same condition costs nothing on the common quiet-success path (no stderr, no pointer, unchanged) and fires on every case where something was written to stderr, including the ones the exit code can't be trusted to reveal. When stderr is present: header reads as a flag instead of neutral bookkeeping, and the .err pointer is listed before .out so it's not the last thing skimmed past on a long completion.
907 lines
35 KiB
Rust
907 lines
35 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()))
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Wake suppression: skip the completion wake when a caller already
|
|
// synchronously observed the task's terminal state via an inline
|
|
// `wait_seconds` poll on `BashRun` or `BashStatus` — the tool response
|
|
// already delivered the result in that same turn, so a follow-up wake
|
|
// message would just be a redundant duplicate of information the agent has.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// In-memory only — a daemon restart wipes it, which is fine: a task still
|
|
/// `running` across a restart is marked `interrupted` on boot (see module
|
|
/// docs) and gets its own fresh wake, independent of this set.
|
|
fn wake_suppressed() -> &'static Mutex<HashSet<String>> {
|
|
static SUPPRESSED: OnceLock<Mutex<HashSet<String>>> = OnceLock::new();
|
|
SUPPRESSED.get_or_init(|| Mutex::new(HashSet::new()))
|
|
}
|
|
|
|
/// Mark `id`'s completion wake as already-delivered-inline. Called after an
|
|
/// inline `wait_seconds` poll (on `BashRun` or `BashStatus`) observes a
|
|
/// terminal task, before the response carrying the full status is written
|
|
/// back to the caller. Idempotent — safe to call more than once per id.
|
|
pub(crate) fn suppress_wake(id: &str) {
|
|
wake_suppressed().lock().unwrap().insert(id.to_owned());
|
|
}
|
|
|
|
/// Consume (remove + report) `id`'s suppression flag. Returns `true` if the
|
|
/// wake should be skipped. One-shot by intent: a completed task's entry is
|
|
/// meant to be drained exactly once. There's a known low-probability race
|
|
/// (see `run_task`'s completion handler) where `suppress_wake` for a task
|
|
/// fires *after* this has already run for it — if the same name gets reused
|
|
/// before that late insert lands, the dangling flag could suppress the new
|
|
/// task's wake instead. Not eliminated, just narrow.
|
|
fn take_wake_suppressed(id: &str) -> bool {
|
|
wake_suppressed().lock().unwrap().remove(id)
|
|
}
|
|
|
|
/// 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
|
|
// ---------------------------------------------------------------------------
|
|
|
|
use hive_sh4re::wire_time::now_unix;
|
|
|
|
/// 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,
|
|
},
|
|
)
|
|
.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: now_unix(),
|
|
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.
|
|
///
|
|
/// Whenever a terminal task is observed here, the caller is about to receive
|
|
/// that result directly in its tool response — so the completion wake for
|
|
/// `id` is marked [`suppress_wake`]d: a status query (waited or not) that
|
|
/// already told the agent the outcome shouldn't be followed by a redundant
|
|
/// "task finished" inbox message for the same information.
|
|
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 observe_terminal(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
|
|
| TaskStatus::Killed
|
|
) {
|
|
suppress_wake(id);
|
|
return Some(task);
|
|
}
|
|
}
|
|
}
|
|
if tokio::time::Instant::now() >= deadline {
|
|
break;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(POLL_MS)).await;
|
|
}
|
|
observe_terminal(read_task(id))
|
|
}
|
|
|
|
/// Marks the wake suppressed if `task` is in a terminal state; passes
|
|
/// `task` through unchanged either way. Shared tail helper for both
|
|
/// `wait_for_task` return points.
|
|
fn observe_terminal(task: Option<TaskFile>) -> Option<TaskFile> {
|
|
if let Some(t) = &task
|
|
&& matches!(
|
|
t.status,
|
|
TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted | TaskStatus::Killed
|
|
)
|
|
{
|
|
suppress_wake(&t.id);
|
|
}
|
|
task
|
|
}
|
|
|
|
/// 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(now_unix());
|
|
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(now_unix());
|
|
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(now_unix());
|
|
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(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");
|
|
}
|
|
// If an inline `status`/`run` wait already handed this exact terminal
|
|
// result to the caller in a tool response (see `wait_for_task` /
|
|
// `observe_terminal`), there's nothing left to surface — retire the
|
|
// keyed todo without a `done` upsert so the agent gets no redundant
|
|
// wake. Narrow race: an inline waiter polling in the few hundred ms
|
|
// around this point may lose the race and still get a `done` todo
|
|
// alongside its inline result; best-effort, same tolerance as the rest
|
|
// of this daemon's guarantees.
|
|
if take_wake_suppressed(&id) {
|
|
clear_bash_todo(socket, &id).await;
|
|
tracing::debug!(id = %id, "bash_runner: done todo suppressed (already observed via status)");
|
|
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 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(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 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());
|
|
}
|
|
|
|
// Wake suppression is a single process-wide registry (see
|
|
// `wake_suppressed()`), so these run serially against distinct ids to
|
|
// avoid cross-test interference under parallel test execution.
|
|
|
|
#[test]
|
|
fn wake_suppression_is_one_shot() {
|
|
use super::{suppress_wake, take_wake_suppressed};
|
|
let id = "test-2270-one-shot";
|
|
assert!(!take_wake_suppressed(id), "unset id starts unsuppressed");
|
|
suppress_wake(id);
|
|
assert!(take_wake_suppressed(id), "set id reports suppressed once");
|
|
assert!(
|
|
!take_wake_suppressed(id),
|
|
"consuming the flag clears it — second read is unsuppressed"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn wake_suppression_is_idempotent_to_set() {
|
|
use super::{suppress_wake, take_wake_suppressed};
|
|
let id = "test-2270-idempotent";
|
|
suppress_wake(id);
|
|
suppress_wake(id); // simulates two concurrent observers of the same terminal task
|
|
assert!(take_wake_suppressed(id));
|
|
assert!(!take_wake_suppressed(id));
|
|
}
|
|
|
|
// 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;
|
|
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;
|
|
// 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;
|
|
let body = done_summary("t1", "exit=1", Some((false, true)));
|
|
assert!(body.contains(".err"));
|
|
assert!(!body.contains(".out"), "no stdout ⇒ no stdout pointer");
|
|
}
|
|
}
|