hyperhive/hive-bash-mcp/src/runner.rs

//! 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)
    {
        body.push_str("\n\noutput captured — read the full text with:");
        if has_stdout {
            let _ = write!(body, "\n  Read({})", crate::paths::task_out(id).display());
        }
        if has_stderr {
            let _ = write!(
                body,
                "\n  Read({})  # stderr",
                crate::paths::task_err(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));
    }
}