kanade-agent 0.59.0

use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use crate::kill::KillSwitch;
use crate::live_tail::LiveTail;
use crate::output_cap::{CappedOutput, MAX_CAPTURE_BYTES};

use anyhow::{Context, Result};
use kanade_shared::wire::{Command, RunAs, Shell};
use rand::RngExt;
use tokio::io::AsyncReadExt;
use tokio::process::Command as ProcessCommand;
use tracing::{debug, info, warn};
use uuid::Uuid;

/// #43: PowerShell console-encoding prelude. Lives in the
/// launcher script (see [`TempPowerShellLaunch`]) so the user
/// script's `[CmdletBinding()] / param(...)` headers stay at the
/// top of their own physical file (PowerShell rejects them
/// anywhere else). Both `run_as: System` and `run_as: user /
/// system_gui` paths route through the same launcher.
pub(crate) const POWERSHELL_UTF8_PRELUDE: &str = "[Console]::OutputEncoding = New-Object System.Text.UTF8Encoding $false; \
     $OutputEncoding = [Console]::OutputEncoding; ";

/// Process-wide staging directory for temp `.ps1` files.
///
/// Layout: `%ProgramData%/Kanade/agent-scripts/<uuid>/` on Windows,
/// `/Library/Application Support/Kanade/agent-scripts/<uuid>/` on a root
/// macOS agent (0755 dirs / 0644 files, so a `run_as: user` child can read
/// its launcher — see `process_as_user_macos::create_staging_dir`), and
/// `$TMPDIR/kanade-agent-<uuid>/` elsewhere (dev / tests only —
/// the temp fallback skips the static `agent-scripts/` segment because
/// `$TMPDIR` is world-writable and a predictable parent would
/// open up a symlink-redirect attack; see staging_dir for the
/// full rationale).
///
/// Hierarchy rationale: the static `agent-scripts/` segment is the
/// logical category (greppable / observable / single-command
/// cleanup with `rmdir agent-scripts\`); the per-process `<uuid>/`
/// subdir provides isolation. Bundling the two into one segment
/// (`agent-scripts-<uuid>/`) — as the first pass did — made
/// per-process dirs look like siblings of the category itself,
/// which is harder to scan visually and to clean up in bulk.
///
/// Why `%ProgramData%` and not `%TEMP%`: the agent runs as
/// LocalSystem; `%TEMP%` for SYSTEM is `C:\Windows\Temp` whose
/// default ACL does NOT grant Users read access to SYSTEM-created
/// files. That breaks `run_as: user / system_gui` (child runs as a
/// non-admin user and would get "access denied" reading the
/// staged script). `C:\ProgramData` propagates an inherited
/// "Users: Read & execute" ACE to files SYSTEM creates inside it,
/// which is exactly what the user-session child needs. Scripts
/// already travel over NATS where the operator can see them, so
/// local-read by other users on the box is an acceptable trade.
///
/// Security shape:
/// - The static parent (`Kanade/agent-scripts/`) is created with
///   `create_dir_all`; pre-existence is fine because we never
///   write directly to it — only ever to a UUID subdir.
/// - The per-process UUID subdir is created with `create_dir`
///   (non-clobber). An attacker can't pre-create it because the
///   UUID is unguessable.
/// - Per-file staging uses `OpenOptions::create_new` (Windows
///   `CREATE_NEW` / POSIX `O_EXCL`) — defence in depth against
///   the astronomically-unlikely UUID collision and against
///   anyone with write into the UUID dir (Users default ACL
///   doesn't grant write, so this is belt-and-braces).
///
/// Gotcha for script authors targeting `run_as: user`:
/// `$PSScriptRoot` (= this UUID dir) is **read-only** for the
/// child user — the default ProgramData ACL grants Users
/// "Read & Execute" but not Modify. A user script that does
/// `New-Item $PSScriptRoot\out.txt` or similar will get access
/// denied. Use `$env:TEMP` / `$env:LOCALAPPDATA` / an absolute
/// path under the user's profile instead. Even for `run_as:
/// System` (where SYSTEM can write here), the dir is cleaned up
/// on script exit, so writing siblings is fragile either way.
fn staging_dir() -> Result<PathBuf> {
    static SLOT: OnceLock<Mutex<Option<PathBuf>>> = OnceLock::new();
    let slot = SLOT.get_or_init(|| Mutex::new(None));
    let mut guard = slot.lock().expect("staging_dir mutex poisoned");
    if let Some(p) = guard.as_ref() {
        return Ok(p.clone());
    }
    // Platforms diverge here for security reasons (Gemini PR #231
    // HIGH): the windows path has an admin-controlled category
    // parent under `%ProgramData%`, so we can safely nest a
    // grep-friendly static `agent-scripts/` segment in; a root
    // macOS agent does the same under the root/admin-only
    // `/Library/Application Support`. Elsewhere the natural parent
    // is `$TMPDIR` (typically `/tmp/`), which is world-writable; a
    // predictable static child like `/tmp/kanade-agent-scripts/`
    // could be pre-created by another local user as a symlink to
    // (say) `/etc/`, and a subsequent `create_dir_all` would follow
    // it and let us write files outside the intended tree. We
    // mitigate by skipping the static segment entirely there —
    // the per-process UUID dir (created with `create_dir`,
    // non-clobber) sits directly under `$TMPDIR`, with an
    // unguessable name that can't be pre-empted.
    let uuid = Uuid::new_v4().simple().to_string();
    #[cfg(target_os = "macos")]
    let dir = crate::process_as_user_macos::create_staging_dir(&uuid)?;
    #[cfg(not(target_os = "macos"))]
    let dir = {
        let dir = if cfg!(target_os = "windows") {
            let category = std::env::var_os("ProgramData")
                .map(PathBuf::from)
                .unwrap_or_else(|| PathBuf::from(r"C:\ProgramData"))
                .join("Kanade")
                .join("agent-scripts");
            std::fs::create_dir_all(&category)
                .with_context(|| format!("create_dir_all {}", category.display()))?;
            category.join(&uuid)
        } else {
            std::env::temp_dir().join(format!("kanade-agent-{uuid}"))
        };
        std::fs::create_dir(&dir).with_context(|| format!("create_dir {}", dir.display()))?;
        dir
    };
    *guard = Some(dir.clone());
    Ok(dir)
}

/// A PowerShell script staged to a single temp `.ps1` file.
///
/// Used as a building block by [`TempPowerShellLaunch`]; not invoked
/// directly by the spawn path anymore (the spawn path stages a
/// launcher/user pair so user-script `[CmdletBinding()] / param(...)`
/// blocks stay at the top of their file).
///
/// Cleanup-on-drop: the file is removed when the struct goes out of
/// scope. PowerShell reads the script into memory at parse time, so
/// deleting the file mid-run doesn't affect the running process.
pub(crate) struct TempPowerShellScript {
    path: PathBuf,
}

impl TempPowerShellScript {
    /// Stage `body` to a fresh BOM-prefixed `.ps1` under the
    /// per-process [`staging_dir`]. Uses `create_new` semantics
    /// (Windows `CREATE_NEW` / POSIX `O_EXCL`) so an attacker can't
    /// substitute the file via a TOCTOU race against the UUID name.
    pub fn write(body: &str) -> Result<Self> {
        let dir = staging_dir()?;
        let path = dir.join(format!("kanade-{}.ps1", Uuid::new_v4().simple()));
        let mut options = std::fs::OpenOptions::new();
        options.write(true).create_new(true);
        // macOS: a `run_as: user` child reads this as another user. Cap the
        // creation mode at 0644 (never group/other-writable, even under a
        // lax umask — the staged file is what root may run) and then chmod
        // to exactly 0644 in case the umask was stricter.
        #[cfg(target_os = "macos")]
        std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o644);
        let mut f = options
            .open(&path)
            .with_context(|| format!("create_new {}", path.display()))?;
        #[cfg(target_os = "macos")]
        {
            use std::os::unix::fs::PermissionsExt;
            f.set_permissions(std::fs::Permissions::from_mode(0o644))
                .with_context(|| format!("chmod 644 {}", path.display()))?;
        }
        // UTF-8 BOM (0xEF 0xBB 0xBF) — PowerShell uses it to detect
        // UTF-8 encoding without a `chcp 65001` dance. Without it,
        // a ja-JP host running default CP932 would mis-parse any
        // multi-byte sequence in the script body.
        f.write_all(&[0xEF, 0xBB, 0xBF])
            .with_context(|| format!("write BOM {}", path.display()))?;
        f.write_all(body.as_bytes())
            .with_context(|| format!("write body {}", path.display()))?;
        Ok(Self { path })
    }

    pub fn path(&self) -> &Path {
        &self.path
    }
}

impl Drop for TempPowerShellScript {
    fn drop(&mut self) {
        // Best-effort. The staging dir itself is never removed
        // (process-lifetime, cleaned by Storage Sense / TEMP GC).
        let _ = std::fs::remove_file(&self.path);
    }
}

/// A staged user script + launcher pair invoked as
/// `powershell -File <launcher>`.
///
/// Why a pair instead of one file: PowerShell only honors
/// `[CmdletBinding()] / param(...)` when they're at the **top** of
/// the script's physical file — prepending an encoding prelude to
/// the same file (the simpler approach) silently breaks any
/// operator-shipped `.ps1` that opens with those headers (e.g.
/// `scripts/deploy/backend.ps1`, surfaced by the
/// install-kanade-backend live test on 2026-05-26).
///
/// Solution: the launcher sets `[Console]::OutputEncoding` etc.
/// then calls the user script via `&` (call operator), which spawns
/// a fresh script scope where the user's `param(...)` block applies
/// to the user file's own arguments.
pub(crate) struct TempPowerShellLaunch {
    launcher: TempPowerShellScript,
    // Kept alive for the launcher's `-File` invocation. The
    // `_user` underscore marks it dead-code-wise; presence in the
    // struct is the entire point.
    _user: TempPowerShellScript,
}

impl TempPowerShellLaunch {
    pub fn stage(user_body: &str) -> Result<Self> {
        let user = TempPowerShellScript::write(user_body)?;
        // PowerShell single-quoted string literal — escape embedded
        // `'` by doubling. `to_string_lossy` instead of fallible
        // `to_str` so a TEMP path with non-UTF-8 surrogates (rare
        // on Windows but technically representable) still produces
        // a path we can hand to PowerShell.
        let user_path = user.path().to_string_lossy().replace('\'', "''");
        // Exit-code propagation. `&` runs the user script in a child
        // scope, so a user `exit N` only ends THAT scope — without the
        // lines below the launcher then ran off its end and the job
        // recorded 0 for every `exit N`. After `&` returns:
        //   * `$?` is False only when the script ended via an explicit
        //     `exit N` (N != 0) or a terminating error; then
        //     `$LASTEXITCODE` holds N (or 1 for a throw) → `exit` it.
        //   * `$?` is True for a script that ran to its end — including
        //     one that HANDLED a failing native command, which leaves a
        //     stale nonzero `$LASTEXITCODE` behind. That is why a plain
        //     `exit $LASTEXITCODE` is wrong: it would fail those runs.
        //     Falling off the end reports 0, as `pwsh -File` does.
        // `$global:LASTEXITCODE = 0` first so nothing from the prelude
        // (or an inherited value) can leak into the propagated code.
        // Measured on pwsh 7.6.6 against running the user script directly
        // with `pwsh -File`, over cases including: exit 3; handled native
        // failure then success; throw; plain output; trailing failing
        // native; exit 0; non-terminating error; exit inside a function;
        // EAP=Stop + Write-Error; CmdletBinding+param+exit 7 — identical
        // exit codes in every case. Windows PowerShell 5.1 shares this
        // launcher, and `$?` after `&` has the same semantics there.
        let launcher_body = format!(
            "{POWERSHELL_UTF8_PRELUDE}$global:LASTEXITCODE = 0\n\
             & '{user_path}' @args\n\
             if (-not $?) {{ exit $LASTEXITCODE }}\n"
        );
        let launcher = TempPowerShellScript::write(&launcher_body)?;
        Ok(Self {
            launcher,
            _user: user,
        })
    }

    pub fn launcher_path(&self) -> &Path {
        self.launcher.path()
    }
}

/// Outcome of a child-process run after kill / timeout / completion races.
pub enum ExecOutcome {
    Completed {
        exit_code: i32,
        stdout: String,
        stderr: String,
    },
    Killed {
        stdout: String,
        stderr: String,
    },
    Timeout {
        stdout: String,
        stderr: String,
    },
}

/// Spec §2.5.1 jitter — sleep a random `[0, jitter_secs)` interval so a
/// wide fan-out doesn't hit the OS at the same instant on every PC.
///
/// Called from `handle_command` *before* `started_at` is stamped (it used
/// to live at the top of `run_command_with_kill`, i.e. after the
/// timestamp). Keeping the stagger-wait outside the timing window means
/// the recorded duration (`finished_at - started_at`) measures only the
/// script's real runtime — matching `timeout:`, which has always bounded
/// the post-jitter execution alone. Ad-hoc `kanade run` sets
/// `jitter_secs: None`, so this is a no-op there.
///
/// The `> 1` guard (not `> 0`): `jitter_secs == 1` would make the range
/// `0..1`, which `random_range` collapses to a constant `0` — a
/// zero-duration sleep plus a misleading "applying jitter" log line. Skip
/// it so a 1-second jitter config is a true no-op.
pub async fn apply_jitter(cmd: &Command) {
    if let Some(j) = cmd.jitter_secs.filter(|&s| s > 1) {
        let secs = rand::rng().random_range(0..j);
        info!(
            jitter_secs = j,
            sleep_secs = secs,
            "applying jitter before exec"
        );
        tokio::time::sleep(Duration::from_secs(secs)).await;
    }
}

/// Spawn the command's shell child, race wait / kill / timeout, collect
/// stdout+stderr.
///
/// Spec §2.6 Layer 3 — if `cmd.exec_id` is set, subscribe to `kill.{exec_id}`
/// in parallel; a kill message causes `child.kill().await` and the outcome
/// is reported as `Killed`. A command without an `exec_id` (e.g. ad-hoc CLI
/// runs) still respects `timeout_secs`.
/// `live` (when `Some`) is the in-flight job's live-tail ring buffer:
/// every stdout/stderr chunk we read is appended to it as we go, so the
/// `job.tail.<pc_id>` handler can serve a running job's output to the
/// SPA. `None` for paths that don't want live capture (e.g. ad-hoc
/// `kanade run` from the CLI). It never affects the final captured
/// strings — those are still the complete byte stream.
/// The command's start deadline had passed at the moment the process would
/// have been created, so nothing was launched. Returned (inside an
/// `anyhow::Error`) rather than folded into [`ExecOutcome`] because no process
/// ran and the caller must tell "never started" apart from any real outcome.
#[derive(Debug)]
pub struct StartDeadlineExpired(pub chrono::DateTime<chrono::Utc>);

impl std::fmt::Display for StartDeadlineExpired {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "start deadline {} passed before launch", self.0)
    }
}

impl std::error::Error for StartDeadlineExpired {}

/// Refuse to launch past `deadline`. Called at the last point before the
/// process is actually created, so waits between the caller's own check and
/// here (kill-listener setup, a busy blocking pool) cannot carry a command
/// over its deadline. Inclusive, like every other start-deadline check.
pub(crate) fn ensure_start_deadline(deadline: Option<chrono::DateTime<chrono::Utc>>) -> Result<()> {
    match deadline {
        Some(d) if chrono::Utc::now() > d => Err(StartDeadlineExpired(d).into()),
        _ => Ok(()),
    }
}

pub async fn run_command_with_kill(
    kill: &KillSwitch,
    cmd: &Command,
    live: Option<Arc<LiveTail>>,
) -> Result<ExecOutcome> {
    run_command_with_start_deadline(kill, cmd, live, None).await
}

/// [`run_command_with_kill`] that also enforces a start deadline immediately
/// before the process is created. The deadline bounds the launch only; a
/// process that has started is never killed for it.
pub async fn run_command_with_start_deadline(
    kill: &KillSwitch,
    cmd: &Command,
    live: Option<Arc<LiveTail>>,
    start_deadline: Option<chrono::DateTime<chrono::Utc>>,
) -> Result<ExecOutcome> {
    // v0.21: on Windows, run_as: user / system_gui take a separate Win32
    // path (CreateProcessAsUserW). System (default) stays on tokio::process
    // — backward-compatible for every pre-v0.21 manifest in the wild. macOS
    // needs no separate path: `host_command` wraps the same tokio::process
    // host in `launchctl asuser`, so it shares the capture / kill / timeout
    // machinery below.
    #[cfg(not(target_os = "macos"))]
    if !matches!(cmd.run_as, RunAs::System) {
        return run_in_user_session_dispatch(kill, cmd, live, start_deadline).await;
    }

    // #43: belt-and-braces. The tolerant decoder (below, around the
    // stdout_task / stderr_task spawn) keeps the capture useful even
    // when PowerShell emits CP932; this prelude makes the child
    // write UTF-8 to begin with, matching what the operator sees
    // when they test the script locally with the same `powershell`
    // binary. Combined, the agent's capture pipeline is both correct
    // AND consistent with manifest authors' local-test results.
    // cmd.exe doesn't have an equivalent one-liner that survives
    // across legacy / unicode commands; the Cmd branch relies on the
    // tolerant decoder alone.
    // Keep `_launch` alive through spawn + wait so both staged
    // files (launcher + user script) outlive PowerShell's parse.
    // PowerShell loads scripts into memory at parse time, so
    // removal-on-drop after wait is safe.
    // Both `_launch` and `launcher_path_owned` are declared in the
    // outer scope so their backing storage outlives the `args` Vec
    // (which borrows `&str` into `launcher_path_owned`). `Option`
    // sidesteps the "value assigned but never read" lint that
    // dummy-init in the Cmd branch would trip.
    //
    // `-File <launcher>` (vs the old `-Command "<body>"`): a body
    // with `[CmdletBinding()] / param(...)` headers is valid as a
    // script file but a parse error as a command-line expression.
    // The launcher sets UTF-8 console encoding then
    // `& '<user.ps1>' @args` — that call-operator boundary means
    // headers at the top of the user file stay at the top of THEIR
    // physical script, which is what PowerShell requires.
    //
    // `-ExecutionPolicy Bypass` is needed because -File honors the
    // host's ExecutionPolicy (Restricted blocks .ps1 entirely);
    // -Command mode silently bypassed it. Adding Bypass keeps
    // behavioural parity with the pre-fix path.
    let _launch: Option<TempPowerShellLaunch>;
    let launcher_path_owned: Option<String>;
    let (program, args): (&str, Vec<&str>) = match cmd.shell {
        Shell::Powershell => {
            let launch = TempPowerShellLaunch::stage(&cmd.script)?;
            launcher_path_owned = Some(launch.launcher_path().to_string_lossy().into_owned());
            _launch = Some(launch);
            (
                "powershell",
                vec![
                    "-NoProfile",
                    "-NonInteractive",
                    "-ExecutionPolicy",
                    "Bypass",
                    "-File",
                    launcher_path_owned.as_deref().unwrap(),
                ],
            )
        }
        Shell::Cmd => {
            _launch = None;
            // launcher_path_owned stays uninitialized — the Cmd
            // args Vec doesn't borrow from it, so the compiler
            // doesn't require an init on this branch.
            ("cmd", vec!["/C", &cmd.script])
        }
        Shell::Sh => {
            _launch = None;
            // Inline like Cmd — `sh -c <script>`. No launcher/prelude:
            // sh is UTF-8 native so the CP932 console dance the
            // PowerShell branch needs doesn't apply.
            ("sh", vec!["-c", &cmd.script])
        }
        Shell::Pwsh => {
            // PowerShell 7, cross-platform. Reuse the exact same temp
            // `.ps1` launcher as Windows PowerShell (script staging +
            // `[CmdletBinding()]/param()` header handling are identical
            // across both PowerShell hosts) — only the program name and
            // the execution-policy switch differ.
            let launch = TempPowerShellLaunch::stage(&cmd.script)?;
            launcher_path_owned = Some(launch.launcher_path().to_string_lossy().into_owned());
            _launch = Some(launch);
            let mut args = vec!["-NoProfile", "-NonInteractive"];
            // `-ExecutionPolicy` is a Windows-only concept; pwsh on
            // Linux/macOS rejects it. Only pass it where it means
            // something.
            #[cfg(target_os = "windows")]
            args.extend_from_slice(&["-ExecutionPolicy", "Bypass"]);
            args.extend_from_slice(&["-File", launcher_path_owned.as_deref().unwrap()]);
            ("pwsh", args)
        }
    };
    let mut builder = host_command(cmd, program, &args)?;
    builder
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);
    #[cfg(unix)]
    spawn_in_own_session(&mut builder);
    ensure_start_deadline(start_deadline)?;
    // A kill that arrived while the run was waiting (jitter, slot, backoff)
    // must not cost a process launch.
    if kill.is_killed() {
        return Ok(ExecOutcome::Killed {
            stdout: String::new(),
            stderr: String::new(),
        });
    }
    let mut child = builder
        .spawn()
        .with_context(|| format!("spawn {program}"))?;

    // Put the host (`powershell` / `pwsh` / `sh` / `cmd`) — and every
    // descendant it spawns — somewhere a kill/timeout can terminate the
    // WHOLE tree at once. Without this, `child.kill()` only reaps the
    // host; a grandchild (e.g. a job that runs `claude`) would be
    // orphaned AND keep the inherited stdout/stderr pipe handles open.
    // See [`KillTree`].
    let tree = KillTree::attach(&child);

    let stdout_handle = child.stdout.take();
    let stderr_handle = child.stderr.take();

    // #43: `read_to_string` is strict UTF-8 — a single invalid byte
    // sequence makes it return `Err(InvalidData)` AND discard
    // everything read so far. On ja-JP Windows that fires whenever
    // a PowerShell child emits CP932-encoded Japanese on stdout
    // (default `[Console]::OutputEncoding` is the system OEM
    // codepage, not UTF-8). The whole inventory probe output was
    // silently lost. Read as bytes + `String::from_utf8_lossy` so
    // we keep every byte of useful output; invalid runs become
    // U+FFFD and don't poison the rest of the capture. Same fix
    // for any future locale / cmd-shell / 3rd-party tool that
    // emits non-UTF-8 — not specific to PowerShell.
    //
    // Gemini #83 fix: return `(String, Option<Error>)` instead of
    // `Result<String, Error>` so a mid-stream I/O failure (broken
    // pipe, child crash partway through writing, etc.) preserves
    // every byte we DID manage to read instead of throwing the
    // partial buffer away with `?`. The caller logs the error +
    // annotates stderr with a marker but keeps the partial capture.
    // v0.4x: chunked drain instead of `read_to_end`. We still keep
    // every byte and decode the COMPLETE buffer with `from_utf8_lossy`
    // at the end (identical final output to the old path — the #43 /
    // Gemini #83 partial-capture + error-preserving semantics are
    // unchanged). The only difference: each chunk is *also* appended
    // to the live-tail ring as it arrives, so the `job.tail` handler
    // can serve a running job's output. The ring stores raw bytes and
    // lossy-decodes the whole buffer on snapshot, so a multi-byte char
    // split across two `read` calls doesn't produce a spurious U+FFFD.
    let (capture_stop, stop) = tokio::sync::watch::channel(false);
    let live_out = live.clone();
    let out_stop = stop.clone();
    let stdout_task = tokio::spawn(async move {
        drain_to_string(stdout_handle, live_out, Stream::Stdout, out_stop).await
    });
    let live_err = live.clone();
    let stderr_task = tokio::spawn(async move {
        drain_to_string(stderr_handle, live_err, Stream::Stderr, stop).await
    });

    let timeout_dur = Duration::from_secs(cmd.timeout_secs.max(1));

    let inner = match &cmd.exec_id {
        Some(eid) => {
            tokio::select! {
                // A clean host exit never touches the tree: a daemon the
                // script detached on purpose keeps running.
                status = child.wait() => {
                    debug!(exec_id = %eid, "child exited (wait arm fired)");
                    let s = status?;
                    OutcomeInner::Completed(s.code().unwrap_or(-1))
                }
                _ = kill.killed() => {
                    info!(exec_id = %eid, "kill arm fired");
                    // Terminate the whole tree (host + descendants) so
                    // orphaned grandchildren can't keep the pipes open
                    // and hang the drain.
                    tree.terminate(&mut child).await;
                    OutcomeInner::Killed
                }
                _ = tokio::time::sleep(timeout_dur) => {
                    info!(exec_id = %eid, "timeout arm fired");
                    tree.terminate(&mut child).await;
                    OutcomeInner::Timeout
                }
            }
        }
        None => {
            tokio::select! {
                status = child.wait() => {
                    let s = status?;
                    OutcomeInner::Completed(s.code().unwrap_or(-1))
                }
                _ = tokio::time::sleep(timeout_dur) => {
                    tree.terminate(&mut child).await;
                    OutcomeInner::Timeout
                }
            }
        }
    };

    let (stdout, stderr) = finish_capture(stdout_task, stderr_task, capture_stop).await?;

    Ok(match inner {
        OutcomeInner::Completed(code) => ExecOutcome::Completed {
            exit_code: code,
            stdout,
            stderr,
        },
        OutcomeInner::Killed => ExecOutcome::Killed { stdout, stderr },
        OutcomeInner::Timeout => ExecOutcome::Timeout { stdout, stderr },
    })
}

/// The host process for a job: `program args...` under the agent's own
/// identity — or, on macOS for `run_as: user` / `system_gui`, wrapped by
/// [`crate::process_as_user_macos::session_command`] to run in the console
/// user's GUI session.
fn host_command(cmd: &Command, program: &str, args: &[&str]) -> Result<ProcessCommand> {
    #[cfg(target_os = "macos")]
    if !matches!(cmd.run_as, RunAs::System) {
        return crate::process_as_user_macos::session_command(
            cmd.run_as,
            program,
            args,
            cmd.cwd.as_deref(),
        );
    }
    let mut builder = ProcessCommand::new(program);
    builder.args(args);
    // macOS: a LaunchDaemon's PATH is only /usr/bin:/bin:/usr/sbin:/sbin
    // and an installed plist is never rewritten, so the agent itself gives
    // system jobs the same PATH as session jobs; the rest of the env is
    // still inherited. Setting PATH on the Command also moves the lookup of
    // a bare `program` (`pwsh`) onto it — std resolves against the child's
    // PATH once `env("PATH", ..)` is set.
    #[cfg(target_os = "macos")]
    builder.env("PATH", crate::process_as_user_macos::job_path());
    if let Some(dir) = cmd.cwd.as_deref().filter(|s| !s.is_empty()) {
        // v0.21.2: expand `~` (and `%FOO%` on Windows) against the
        // agent's own identity before handing to current_dir (which
        // itself does no expansion).
        #[cfg(target_os = "windows")]
        {
            match crate::cwd_expand::open_self_token()
                .and_then(|tok| crate::cwd_expand::expand(dir, tok.handle()))
            {
                Ok(expanded) => {
                    builder.current_dir(expanded);
                }
                Err(e) => {
                    warn!(error = %e, raw_cwd = %dir, "cwd expansion failed; using raw value");
                    builder.current_dir(dir);
                }
            }
        }
        #[cfg(target_os = "macos")]
        {
            builder.current_dir(crate::process_as_user_macos::expand_agent_cwd(dir));
        }
        #[cfg(not(any(target_os = "windows", target_os = "macos")))]
        {
            builder.current_dir(dir);
        }
    }
    Ok(builder)
}

/// Start the host as the leader of a new session (and so of a new process
/// group whose id is the host's pid), for [`KillTree`] to signal. `setsid`
/// rather than a bare process group so the chain also sheds any
/// controlling terminal of a dev-run agent: with a terminal present,
/// `sudo` (macOS `run_as: user`) may run the job behind a pseudo-terminal
/// (its `use_pty` default).
#[cfg(unix)]
fn spawn_in_own_session(builder: &mut ProcessCommand) {
    // SAFETY: the hook runs in the forked child before exec, where only
    // async-signal-safe calls are allowed; setsid(2) is one and touches no
    // Rust state.
    unsafe {
        builder.pre_exec(|| {
            if libc::setsid() == -1 {
                Err(std::io::Error::last_os_error())
            } else {
                Ok(())
            }
        });
    }
}

/// How long a unix host gets to exit on SIGTERM before its whole process
/// group is SIGKILLed.
#[cfg(unix)]
const KILL_TERM_GRACE: Duration = Duration::from_secs(5);

/// The host plus every descendant, as one unit to tear down on kill /
/// timeout — never on a clean host exit, which must leave a deliberately
/// detached daemon running.
///
/// - Windows: a Job Object ([`crate::job_object`]).
/// - Unix: the host's own process group ([`spawn_in_own_session`]):
///   SIGTERM to the group, up to [`KILL_TERM_GRACE`] for the host to exit,
///   then SIGKILL to whatever is left. A descendant that moved itself to
///   another group/session (a double-forked daemon) is out of reach, but it
///   can't hang the job: capture is bounded by [`OUTPUT_DRAIN_GRACE`].
///
/// With no tree (Job assignment failed) it degrades to killing the host.
struct KillTree {
    #[cfg(target_os = "windows")]
    job: Option<crate::job_object::JobObject>,
    #[cfg(unix)]
    group: Option<libc::pid_t>,
}

impl KillTree {
    fn attach(child: &tokio::process::Child) -> Self {
        #[cfg(target_os = "windows")]
        {
            let job = child.raw_handle().and_then(|h| {
                crate::job_object::JobObject::assign_handle(windows::Win32::Foundation::HANDLE(h))
                    .map_err(|e| {
                        warn!(error = %e, "job object assign failed; kill falls back to single-process terminate");
                    })
                    .ok()
            });
            Self { job }
        }
        #[cfg(unix)]
        {
            Self {
                group: child.id().and_then(|pid| libc::pid_t::try_from(pid).ok()),
            }
        }
    }

    async fn terminate(&self, child: &mut tokio::process::Child) {
        #[cfg(target_os = "windows")]
        if let Some(job) = &self.job {
            job.terminate();
            return;
        }
        #[cfg(unix)]
        if let Some(pgid) = self.group {
            signal_group(pgid, libc::SIGTERM);
            if tokio::time::timeout(KILL_TERM_GRACE, child.wait())
                .await
                .is_err()
            {
                info!(
                    pgid,
                    "host still running after SIGTERM grace; sending SIGKILL"
                );
            }
            signal_group(pgid, libc::SIGKILL);
            if let Err(e) = child.wait().await {
                warn!(error = %e, "wait after killing the process group failed");
            }
            return;
        }
        if let Err(e) = child.kill().await {
            warn!(error = %e, "child.kill failed (process may already be dead)");
        }
    }
}

/// `killpg`, treating "no such group" (everyone already gone) as success.
#[cfg(unix)]
fn signal_group(pgid: libc::pid_t, signal: libc::c_int) {
    // SAFETY: killpg(2) only reads its two integer arguments.
    if unsafe { libc::killpg(pgid, signal) } == -1 {
        let err = std::io::Error::last_os_error();
        if err.raw_os_error() != Some(libc::ESRCH) {
            warn!(error = %err, pgid, signal, "killpg failed");
        }
    }
}

/// Allow buffered output to drain after the host exits, without waiting for
/// intentionally detached descendants to close their inherited pipe handles.
/// Dropping the sender also stops readers on an early return/cancellation.
pub(crate) const OUTPUT_DRAIN_GRACE: Duration = Duration::from_secs(2);

type Capture = (String, Option<anyhow::Error>);

pub(crate) async fn finish_capture(
    stdout: tokio::task::JoinHandle<Capture>,
    stderr: tokio::task::JoinHandle<Capture>,
    stop: tokio::sync::watch::Sender<bool>,
) -> Result<(String, String)> {
    let joined = async { tokio::join!(stdout, stderr) };
    tokio::pin!(joined);
    let (out, err) = match tokio::time::timeout(OUTPUT_DRAIN_GRACE, &mut joined).await {
        Ok(pair) => pair,
        Err(_) => {
            let _ = stop.send(true);
            joined.await
        }
    };
    let (stdout, stdout_err) = out.context("stdout capture join")?;
    let (mut stderr, stderr_err) = err.context("stderr capture join")?;
    for (stream, error) in [("stdout", stdout_err), ("stderr", stderr_err)] {
        if let Some(error) = error {
            warn!(%stream, %error, "capture ended early (kept partial output)");
            stderr.push_str(&format!(
                "\n[agent: {stream} capture ended early: {error}]\n"
            ));
        }
    }
    Ok((stdout, stderr))
}

enum OutcomeInner {
    Completed(i32),
    Killed,
    Timeout,
}

/// Which captured stream a chunk belongs to — selects the live-tail
/// ring it gets mirrored into.
#[derive(Clone, Copy, Debug)]
enum Stream {
    Stdout,
    Stderr,
}

/// Drain a child pipe to a `String` in chunks, mirroring each chunk to
/// the live-tail ring as it arrives.
///
/// Returns `(decoded, partial_error)` with the SAME contract as the
/// old `read_to_end` path: on a mid-stream I/O error we keep every
/// byte read so far (`from_utf8_lossy` applied to the whole buffer)
/// and surface the error so the caller can log + annotate. The live
/// ring sees only the bytes that actually arrived before the error.
async fn drain_to_string<R>(
    reader: Option<R>,
    live: Option<Arc<LiveTail>>,
    stream: Stream,
    mut stop: tokio::sync::watch::Receiver<bool>,
) -> (String, Option<anyhow::Error>)
where
    R: tokio::io::AsyncRead + Unpin,
{
    // Bounded, not a growing `Vec`: the capture is held in memory for the
    // whole run, so an unbounded one made a script's output the ceiling on
    // the agent's RSS (#1320).
    let mut buf = CappedOutput::new(MAX_CAPTURE_BYTES);
    let mut err: Option<anyhow::Error> = None;
    if let Some(mut s) = reader {
        // 8 KiB chunks: large enough that a chatty job doesn't thrash
        // the read loop, small enough that the live tail updates
        // promptly (a 5 s SPA poll sees output within one chunk).
        let mut chunk = [0u8; 8 * 1024];
        loop {
            let read = tokio::select! {
                biased;
                _ = stop.changed() => {
                    err = Some(anyhow::anyhow!("output drain deadline reached; a descendant may still hold the pipe"));
                    break;
                }
                read = s.read(&mut chunk) => read,
            };
            match read {
                Ok(0) => break,
                Ok(n) => {
                    let slice = &chunk[..n];
                    buf.push(slice);
                    // The live tail still sees every byte: it is its own
                    // bounded ring, and an operator watching a running job
                    // wants the newest output regardless of the capture cap.
                    if let Some(lt) = &live {
                        match stream {
                            Stream::Stdout => lt.push_stdout(slice),
                            Stream::Stderr => lt.push_stderr(slice),
                        }
                    }
                }
                Err(e) => {
                    err = Some(anyhow::Error::new(e));
                    break;
                }
            }
        }
    }
    // Operator-actionable, so it is not left to the payload marker alone: a
    // job whose output is being cut is a job that should be changed (or moved
    // to `collect:`), and the agent log is where that gets noticed.
    if buf.truncated() {
        warn!(
            stream = ?stream,
            bytes_written = buf.total(),
            kept = MAX_CAPTURE_BYTES,
            "output exceeded the capture cap and was truncated"
        );
    }
    (buf.finish(), err)
}

/// Glue between the run's shared [`KillSwitch`] and `process_as_user`'s
/// `oneshot::Receiver<()>` kill channel. We forward "fired" into the
/// channel, so the Win32 path's inner `tokio::select!` can use a plain
/// oneshot. macOS never gets here
/// (see `host_command`); Linux has no user-session launch yet.
#[cfg(not(target_os = "macos"))]
async fn run_in_user_session_dispatch(
    kill: &KillSwitch,
    cmd: &Command,
    live: Option<Arc<LiveTail>>,
    start_deadline: Option<chrono::DateTime<chrono::Utc>>,
) -> Result<ExecOutcome> {
    #[cfg(not(target_os = "windows"))]
    {
        let _ = kill;
        let _ = live;
        let _ = start_deadline;
        warn!(
            run_as = ?cmd.run_as,
            "run_as: user / system_gui is not supported on Linux agents — skipping the script",
        );
        // Synthesise an immediate "stub" outcome rather than silently
        // running as the wrong identity: Linux has no console-session
        // launch yet (Windows and macOS do).
        Ok(ExecOutcome::Completed {
            exit_code: 0,
            stdout: String::new(),
            stderr: format!(
                "run_as: {:?} is not supported on Linux agents; the script was skipped.\n",
                cmd.run_as
            ),
        })
    }

    #[cfg(target_os = "windows")]
    {
        let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<()>();
        // Forward the run's shared kill switch into the oneshot the
        // Win32 waiter selects on. An inert switch (no exec_id) never
        // fires, so the bridge just parks until it is aborted.
        let rx = kill.receiver();
        let bridge = tokio::spawn(async move {
            crate::kill::wait_killed(rx).await;
            info!("kill received → forwarding to user-session waiter");
            let _ = kill_tx.send(());
        });

        let timeout = Duration::from_secs(cmd.timeout_secs.max(1));
        let outcome = crate::process_as_user::run_command_in_user_session(
            cmd,
            cmd.run_as,
            timeout,
            kill_rx,
            live,
            start_deadline,
        )
        .await;

        bridge.abort();
        outcome
    }
}

#[cfg(test)]
mod start_deadline_tests {
    use super::{StartDeadlineExpired, ensure_start_deadline};

    #[test]
    fn a_passed_deadline_blocks_the_launch_and_no_deadline_does_not() {
        assert!(ensure_start_deadline(None).is_ok());
        let future = chrono::Utc::now() + chrono::Duration::seconds(60);
        assert!(ensure_start_deadline(Some(future)).is_ok());
        let past = chrono::Utc::now() - chrono::Duration::seconds(1);
        let err = ensure_start_deadline(Some(past)).unwrap_err();
        assert!(err.is::<StartDeadlineExpired>());
    }
}

#[cfg(test)]
mod tests {
    use tokio::io::AsyncReadExt;

    #[tokio::test]
    async fn capture_finishes_with_partial_output_while_writer_stays_open() {
        use tokio::io::AsyncWriteExt;
        let (mut writer, reader) = tokio::io::duplex(64);
        writer.write_all(b"launcher done").await.unwrap();
        let (stop, rx) = tokio::sync::watch::channel(false);
        let out = tokio::spawn(super::drain_to_string(
            Some(reader),
            None,
            super::Stream::Stdout,
            rx,
        ));
        let err = tokio::spawn(async { (String::new(), None) });
        let (stdout, stderr) = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            super::finish_capture(out, err, stop),
        )
        .await
        .expect("must not wait for descendant EOF")
        .unwrap();
        assert_eq!(stdout, "launcher done");
        assert!(stderr.contains("stdout capture ended early"));
        drop(writer);
    }

    #[tokio::test]
    async fn capture_preserves_clean_eof_without_warning() {
        let (stop, rx) = tokio::sync::watch::channel(false);
        let out = tokio::spawn(super::drain_to_string(
            Some(&b"done"[..]),
            None,
            super::Stream::Stdout,
            rx,
        ));
        let err = tokio::spawn(async { (String::new(), None) });
        let (stdout, stderr) = super::finish_capture(out, err, stop).await.unwrap();
        assert_eq!(stdout, "done");
        assert_eq!(stderr, "");
    }

    /// Mirror the production stdout/stderr reader: read every byte
    /// then `from_utf8_lossy`. Used to assert that invalid UTF-8
    /// (e.g. CP932-encoded Japanese on a non-Unicode console)
    /// doesn't wipe the capture the way `read_to_string` did pre-#43.
    async fn capture_lossy<R: tokio::io::AsyncRead + Unpin>(mut r: R) -> String {
        let mut buf = Vec::new();
        r.read_to_end(&mut buf).await.unwrap();
        String::from_utf8_lossy(&buf).into_owned()
    }

    #[tokio::test]
    async fn cp932_japanese_bytes_are_kept_lossy_not_dropped() {
        // CP932 (Shift-JIS) for "ちつ" — the byte sequence that
        // triggers `read_to_string`'s strict-UTF-8 rejection.
        // Pre-fix, the entire stdout buffer was discarded; post-
        // fix, the bytes survive (as U+FFFD) and the rest of the
        // payload around them stays intact.
        let raw: Vec<u8> = vec![
            b'{', b'"', b'k', b'"', b':', b'"', 0x82, 0xbf, 0x82, 0xc2, b'"', b'}',
        ];
        let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
        // The structural ASCII (`{"k":"…"}`) survives — that's
        // what was being lost pre-fix.
        assert!(captured.starts_with("{\"k\":\""), "ASCII frame preserved");
        assert!(captured.ends_with("\"}"), "ASCII frame preserved");
        // The Japanese bytes become U+FFFD replacement chars (not
        // dropped silently).
        assert!(captured.contains('\u{FFFD}'), "invalid runs marked");
    }

    #[tokio::test]
    async fn pure_utf8_payload_round_trips() {
        let raw = "こんにちは {\"ok\": true}".as_bytes().to_vec();
        let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
        assert_eq!(captured, "こんにちは {\"ok\": true}");
    }

    #[tokio::test]
    async fn empty_stream_yields_empty_string() {
        let raw: Vec<u8> = Vec::new();
        let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
        assert_eq!(captured, "");
    }

    #[test]
    fn powershell_prelude_constant_shape() {
        // Defensive: ensures the prelude itself ends with `; ` (so
        // the user script slots in cleanly without an explicit
        // newline) and contains both the Console + $OutputEncoding
        // statements operators expect when they read agent.log
        // or the script that actually ran.
        assert!(super::POWERSHELL_UTF8_PRELUDE.ends_with("; "));
        assert!(super::POWERSHELL_UTF8_PRELUDE.contains("[Console]::OutputEncoding"));
        assert!(super::POWERSHELL_UTF8_PRELUDE.contains("$OutputEncoding"));
    }

    #[test]
    fn temp_powershell_script_writes_bom_then_body_verbatim() {
        // Regression for the CodeRabbit finding on PR #230 first
        // pass: the staged user file MUST contain the body
        // verbatim. Any prelude prepended here would push
        // `[CmdletBinding()] / param(...)` headers off the top of
        // the file and re-introduce the parse error the PR was
        // meant to fix.
        let script = "[CmdletBinding()] param([string]$X='a'); Write-Output $X";
        let staged = super::TempPowerShellScript::write(script).expect("write");
        let bytes = std::fs::read(staged.path()).expect("read back");
        assert_eq!(
            &bytes[..3],
            &[0xEF, 0xBB, 0xBF],
            "BOM not at start of staged file",
        );
        let body_bytes = &bytes[3..];
        assert_eq!(
            std::str::from_utf8(body_bytes).unwrap(),
            script,
            "user body must be verbatim — no prelude prefix",
        );
        assert_eq!(
            staged.path().extension().and_then(|s| s.to_str()),
            Some("ps1"),
        );
    }

    #[test]
    fn temp_powershell_script_drop_removes_file() {
        let staged = super::TempPowerShellScript::write("Write-Output 'x'").expect("write");
        let path = staged.path().to_path_buf();
        assert!(path.exists(), "file should exist before drop");
        drop(staged);
        assert!(!path.exists(), "file should be gone after drop");
    }

    #[test]
    fn temp_powershell_launch_user_file_has_no_prelude() {
        let user_script =
            "[CmdletBinding()] param([string]$Foo = 'bar'); Write-Output \"got:$Foo\"";
        let launch = super::TempPowerShellLaunch::stage(user_script).expect("stage");
        // Find the user file via the launcher body (it embeds the
        // single-quoted path). We don't expose the user path
        // directly because callers never need it — but the test
        // does, to assert "no prelude on the user side".
        let launcher_text = std::fs::read_to_string(launch.launcher_path()).expect("read launcher");
        let start = launcher_text
            .find("& '")
            .expect("launcher should invoke user script via call operator");
        let after = &launcher_text[start + 3..];
        let end = after.find('\'').expect("launcher path closes its quote");
        let user_path_in_launcher: String = after[..end].replace("''", "'");

        let user_bytes = std::fs::read(&user_path_in_launcher).expect("read user file");
        // BOM + body verbatim. NO prelude. The launcher carries the
        // prelude so the user file's headers stay at the top.
        assert_eq!(&user_bytes[..3], &[0xEF, 0xBB, 0xBF]);
        let body = std::str::from_utf8(&user_bytes[3..]).unwrap();
        assert_eq!(body, user_script);
        assert!(
            !body.contains("[Console]::OutputEncoding"),
            "user file must not carry the encoding prelude",
        );
    }

    #[test]
    fn temp_powershell_launch_launcher_carries_prelude_then_invokes_user() {
        let launch = super::TempPowerShellLaunch::stage("Write-Output 'hi'").expect("stage");
        let launcher_text = std::fs::read_to_string(launch.launcher_path()).expect("read launcher");
        assert!(
            launcher_text.contains("[Console]::OutputEncoding"),
            "launcher must set console encoding before invoking user",
        );
        assert!(
            launcher_text.contains("& '") && launcher_text.contains("' @args"),
            "launcher must invoke user via call operator with @args splat",
        );
        // Prelude precedes the call operator (encoding takes
        // effect before any user output).
        let prelude_pos = launcher_text.find("[Console]::OutputEncoding").unwrap();
        let call_pos = launcher_text.find("& '").unwrap();
        assert!(prelude_pos < call_pos);
    }

    #[test]
    fn temp_powershell_launch_drop_removes_both_files() {
        let launch = super::TempPowerShellLaunch::stage("Write-Output 'x'").expect("stage");
        let launcher_path = launch.launcher_path().to_path_buf();
        // Pull user path out of the launcher body before drop.
        let launcher_text = std::fs::read_to_string(&launcher_path).expect("read launcher");
        let start = launcher_text.find("& '").unwrap() + 3;
        let end = launcher_text[start..].find('\'').unwrap();
        let user_path =
            std::path::PathBuf::from(launcher_text[start..start + end].replace("''", "'"));

        assert!(launcher_path.exists());
        assert!(user_path.exists());
        drop(launch);
        assert!(!launcher_path.exists(), "launcher must be removed on drop");
        assert!(!user_path.exists(), "user file must be removed on drop");
    }

    /// The launcher must report what `pwsh -File <user.ps1>` would: a user
    /// `exit 3` is 3 (the old `& '<user>' @args` body recorded 0), and a
    /// script that handled a failing native command before succeeding is 0
    /// (a naive `exit $LASTEXITCODE` would report 1).
    #[cfg(unix)]
    #[test]
    fn staged_launcher_reports_the_user_scripts_exit_code() {
        let Ok(pwsh) = which::which("pwsh") else {
            println!("skipping: `pwsh` not on PATH");
            return;
        };
        for (script, want) in [("exit 3", 3), ("/usr/bin/false; Write-Output ok", 0)] {
            let launch = super::TempPowerShellLaunch::stage(script).expect("stage");
            let status = std::process::Command::new(&pwsh)
                .args(["-NoProfile", "-NonInteractive", "-File"])
                .arg(launch.launcher_path())
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null())
                .status()
                .expect("run pwsh");
            assert_eq!(status.code(), Some(want), "script: {script}");
        }
    }

    /// Same contract on Windows PowerShell 5.1, which shares the launcher
    /// (`shell: powershell`), spawned the way the agent spawns it.
    #[cfg(windows)]
    #[test]
    fn staged_launcher_reports_the_user_scripts_exit_code_on_windows_powershell() {
        for (script, want) in [("exit 3", 3), ("cmd /c exit 1; Write-Output ok", 0)] {
            let launch = super::TempPowerShellLaunch::stage(script).expect("stage");
            let status = std::process::Command::new("powershell")
                .args([
                    "-NoProfile",
                    "-NonInteractive",
                    "-ExecutionPolicy",
                    "Bypass",
                    "-File",
                ])
                .arg(launch.launcher_path())
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null())
                .status()
                .expect("run powershell");
            assert_eq!(status.code(), Some(want), "script: {script}");
        }
    }
}

/// The unix tree-kill contract of `run_command_with_kill`, through the real
/// spawn: a timeout takes the host's whole process group down, a clean exit
/// leaves a detached daemon alone. `run_as: system` + `sh`, so it runs as
/// any user.
#[cfg(all(test, unix))]
mod process_tree_tests {
    use std::path::Path;
    use std::time::{Duration, Instant};

    use super::{ExecOutcome, run_command_with_kill};
    use crate::kill::{KillSwitch, trigger_local};

    fn sh_job(script: &str, timeout_secs: u64) -> kanade_shared::wire::Command {
        serde_json::from_value(serde_json::json!({
            "id": "tree-test",
            "version": "1",
            "request_id": "req",
            "exec_id": null,
            "shell": "sh",
            "script": script,
            "timeout_secs": timeout_secs,
            "jitter_secs": null,
            "run_as": "system",
        }))
        .expect("command")
    }

    fn read_pid(file: &Path) -> libc::pid_t {
        std::fs::read_to_string(file)
            .expect("pid file")
            .trim()
            .parse()
            .expect("pid")
    }

    fn alive(pid: libc::pid_t) -> bool {
        // SAFETY: signal 0 only checks that `pid` exists.
        unsafe { libc::kill(pid, 0) == 0 }
    }

    /// SIGKILLs the test's own background process however the test ends.
    struct Reap(libc::pid_t);

    impl Drop for Reap {
        fn drop(&mut self) {
            // SAFETY: plain kill(2) on a pid this test started.
            unsafe { libc::kill(self.0, libc::SIGKILL) };
        }
    }

    #[tokio::test]
    async fn timeout_kills_the_whole_process_tree() {
        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("grandchild.pid");
        let script = format!("sleep 300 & echo $! > '{}'; sleep 300", pid_file.display());

        let outcome = run_command_with_kill(&KillSwitch::inert(), &sh_job(&script, 1), None)
            .await
            .expect("run");

        assert!(matches!(outcome, ExecOutcome::Timeout { .. }));
        let grandchild = read_pid(&pid_file);
        let _reap = Reap(grandchild);
        // Killed by the group signal; allow its new parent a moment to reap it.
        let deadline = Instant::now() + Duration::from_secs(5);
        while alive(grandchild) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert!(
            !alive(grandchild),
            "grandchild {grandchild} survived the timeout"
        );
    }

    #[tokio::test]
    async fn clean_exit_leaves_a_detached_daemon_running() {
        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("daemon.pid");
        let script = format!(
            "sleep 300 >/dev/null 2>&1 </dev/null & echo $! > '{}'; exit 0",
            pid_file.display()
        );

        let outcome = run_command_with_kill(&KillSwitch::inert(), &sh_job(&script, 60), None)
            .await
            .expect("run");

        let daemon = read_pid(&pid_file);
        let _reap = Reap(daemon);
        assert!(matches!(
            outcome,
            ExecOutcome::Completed { exit_code: 0, .. }
        ));
        assert!(alive(daemon), "daemon {daemon} was killed on a clean exit");
    }

    #[tokio::test]
    async fn local_kill_terminates_a_running_child_without_a_broker() {
        let mut cmd = sh_job("sleep 300", 600);
        cmd.exec_id = Some("proc-local-kill".into());
        let switch = KillSwitch::arm(None, cmd.exec_id.as_deref()).await;
        let run = tokio::spawn({
            let cmd = cmd.clone();
            async move { run_command_with_kill(&switch, &cmd, None).await }
        });
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(trigger_local("proc-local-kill"));
        let outcome = tokio::time::timeout(Duration::from_secs(10), run)
            .await
            .expect("run ends after kill")
            .unwrap()
            .expect("run");
        assert!(matches!(outcome, ExecOutcome::Killed { .. }));
        assert!(!crate::kill::registered("proc-local-kill"));
    }

    #[tokio::test]
    async fn a_kill_latched_before_launch_prevents_the_launch() {
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("ran");
        let mut cmd = sh_job(&format!("touch '{}'", marker.display()), 60);
        cmd.exec_id = Some("proc-pre-kill".into());
        let switch = KillSwitch::arm(None, cmd.exec_id.as_deref()).await;
        trigger_local("proc-pre-kill");
        let outcome = run_command_with_kill(&switch, &cmd, None).await.unwrap();
        assert!(matches!(outcome, ExecOutcome::Killed { .. }));
        assert!(!marker.exists());
    }

    #[tokio::test]
    #[ignore = "requires a live nats-server"]
    async fn remote_kill_terminates_a_running_child() {
        let client = crate::kill::broker_test::connect().await;
        let mut cmd = sh_job("sleep 300", 600);
        cmd.exec_id = Some("proc-remote-kill".into());
        let switch = KillSwitch::arm(Some(&client), cmd.exec_id.as_deref()).await;
        let run = tokio::spawn({
            let cmd = cmd.clone();
            async move { run_command_with_kill(&switch, &cmd, None).await }
        });
        tokio::time::sleep(Duration::from_millis(300)).await;
        crate::kill::broker_test::publish_kill(&client, "proc-remote-kill").await;
        let outcome = tokio::time::timeout(Duration::from_secs(10), run)
            .await
            .expect("run ends after remote kill")
            .unwrap()
            .expect("run");
        assert!(matches!(outcome, ExecOutcome::Killed { .. }));
    }
}