agent_engine/tools/

bash.rs

use super::{strip_ansi, Tool, ToolContext};
use crate::{Result, RuntimeError};
use serde_json::{json, Value};
use zeroize::Zeroizing;

const BASH_INTERMEDIARY_CHANNEL_CAPACITY: usize = 64;
static BASH_INTERMEDIARY_PRODUCED_BYTES: std::sync::atomic::AtomicU64 =
    std::sync::atomic::AtomicU64::new(0);
static BASH_INTERMEDIARY_ACCEPTED_BYTES: std::sync::atomic::AtomicU64 =
    std::sync::atomic::AtomicU64::new(0);
static BASH_INTERMEDIARY_CONSUMED_BYTES: std::sync::atomic::AtomicU64 =
    std::sync::atomic::AtomicU64::new(0);
static BASH_INTERMEDIARY_DROPPED_BYTES: std::sync::atomic::AtomicU64 =
    std::sync::atomic::AtomicU64::new(0);

#[derive(Debug, Clone, Copy)]
pub struct BashIntermediarySnapshot {
    pub produced_bytes: u64,
    pub accepted_bytes: u64,
    pub consumed_bytes: u64,
    pub dropped_bytes: u64,
    pub retained_bytes: u64,
}

impl BashIntermediarySnapshot {
    /// Every byte ever produced has been consumed or dropped: nothing is in
    /// flight anywhere in the process. A leaked byte keeps this false forever.
    pub fn is_settled(&self) -> bool {
        self.produced_bytes == self.consumed_bytes + self.dropped_bytes
    }
}

/// Counters are read downstream-first. Each one only grows, and upstream
/// counts are bumped before the bytes move on, so a snapshot torn by a
/// concurrent relay can overstate in-flight bytes but never understate them:
/// `is_settled` is never true unless the process really was settled.
pub fn bash_intermediary_snapshot() -> BashIntermediarySnapshot {
    use std::sync::atomic::Ordering::SeqCst;
    let consumed = BASH_INTERMEDIARY_CONSUMED_BYTES.load(SeqCst);
    let dropped = BASH_INTERMEDIARY_DROPPED_BYTES.load(SeqCst);
    let accepted = BASH_INTERMEDIARY_ACCEPTED_BYTES.load(SeqCst);
    let produced = BASH_INTERMEDIARY_PRODUCED_BYTES.load(SeqCst);
    BashIntermediarySnapshot {
        produced_bytes: produced,
        accepted_bytes: accepted,
        consumed_bytes: consumed,
        dropped_bytes: dropped,
        retained_bytes: produced.saturating_sub(consumed).saturating_sub(dropped),
    }
}

/// Test oracle for the handoff counters: wait until the whole process is
/// settled. Concurrent relays only delay it; a leaked byte times it out.
#[cfg(test)]
async fn wait_handoff_settled() -> BashIntermediarySnapshot {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    loop {
        let now = bash_intermediary_snapshot();
        if now.is_settled() {
            return now;
        }
        assert!(std::time::Instant::now() < deadline, "handoff bytes never settled: {now:?}");
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
}

pub struct BashTool;

/// One sanitized read in flight between a stream reader and the executor.
/// Its bytes are counted out exactly once: `consume` when the executor takes
/// it, or `Drop` when it never arrives (the channel closed before the send,
/// or it was still queued when the receiver went away on timeout or cancel).
/// Tokio drops queued values with the channel, so no path can leak a count.
struct HandoffChunk(Option<String>);

impl HandoffChunk {
    fn new(text: String) -> Self {
        BASH_INTERMEDIARY_PRODUCED_BYTES
            .fetch_add(text.len() as u64, std::sync::atomic::Ordering::SeqCst);
        Self(Some(text))
    }

    fn len(&self) -> usize {
        self.0.as_ref().map_or(0, String::len)
    }

    fn consume(mut self) -> String {
        let text = self.0.take().unwrap_or_default();
        BASH_INTERMEDIARY_CONSUMED_BYTES
            .fetch_add(text.len() as u64, std::sync::atomic::Ordering::SeqCst);
        text
    }
}

impl Drop for HandoffChunk {
    fn drop(&mut self) {
        if let Some(text) = self.0.take() {
            BASH_INTERMEDIARY_DROPPED_BYTES
                .fetch_add(text.len() as u64, std::sync::atomic::Ordering::SeqCst);
        }
    }
}

const READ_CHUNK_SIZE: usize = 1024;
const MAX_STREAMED_DELTA_BYTES: usize = 16 * 1024;

/// sudo passwords handed out during one command, redacted from its output
/// and zeroized when dropped.
pub(crate) type Redactions = std::sync::Arc<std::sync::Mutex<Vec<Zeroizing<String>>>>;

/// The askpass broker one bash call uses (Linux), prepared before spawn.
#[cfg(target_os = "linux")]
pub(crate) type Askpass = super::sudo_askpass::broker::PreparedBroker;
#[cfg(not(target_os = "linux"))]
pub(crate) type Askpass = std::convert::Infallible;

fn sanitize_output(input: &str) -> String {
    let stripped = strip_ansi(input);
    stripped
        .chars()
        .filter(|ch| {
            *ch == '\n' || *ch == '\r' || *ch == '\t' || (!ch.is_control() && *ch != '\u{7f}')
        })
        .collect()
}

/// Read one child stream in `READ_CHUNK_SIZE` chunks and forward the
/// sanitized text to the bash executor, tagged stdout/stderr. A character
/// split between two reads is held back and decoded whole (#286); only an
/// incomplete sequence (at most 3 bytes) waits, so prompts without a
/// trailing newline are still forwarded as soon as they are read.
async fn forward_output<R>(
    mut reader: R,
    is_stderr: bool,
    tx: tokio::sync::mpsc::Sender<(bool, HandoffChunk)>,
) where
    R: tokio::io::AsyncRead + Unpin,
{
    use tokio::io::AsyncReadExt;
    let mut buf = vec![0u8; READ_CHUNK_SIZE];
    let mut decoder = super::utf8_stream::Utf8StreamDecoder::new();
    loop {
        let (text, eof) = match reader.read(&mut buf).await {
            Ok(0) | Err(_) => (decoder.finish(), true),
            Ok(n) => (decoder.decode(&buf[..n]), false),
        };
        let msg = sanitize_output(&text);
        if !msg.is_empty() {
            let chunk = HandoffChunk::new(msg);
            // Count acceptance before the send makes the chunk visible, so
            // accepted >= consumed holds at every instant. A closed channel
            // drops the chunk here, and its Drop counts it.
            let Ok(permit) = tx.reserve().await else {
                break;
            };
            BASH_INTERMEDIARY_ACCEPTED_BYTES
                .fetch_add(chunk.len() as u64, std::sync::atomic::Ordering::SeqCst);
            permit.send((is_stderr, chunk));
        }
        if eof {
            break;
        }
    }
}

fn append_bounded(output: &mut String, text: &str, max_output: usize) -> bool {
    if output.len() >= max_output {
        return false;
    }
    let remaining = max_output - output.len();
    if text.len() <= remaining {
        output.push_str(text);
        true
    } else {
        let mut end = remaining;
        while end > 0 && !text.is_char_boundary(end) {
            end -= 1;
        }
        output.push_str(&text[..end]);
        false
    }
}

/// Ends the result of a command whose output passed `max_output`. The
/// command is killed there, so this is not a display cut: it never finished
/// and its exit status is unknown. Keeps the old `[output truncated at N`
/// prefix, so whatever counts these stops still matches.
fn output_limit_notice(max_output: usize) -> String {
    format!(
        "\n\n[output truncated at {max_output} bytes: the command was stopped at this limit \
         before it finished, so its exit status is unknown. Redirect large output to a file \
         (e.g. cmd > out.log 2>&1) and read only the parts you need.]"
    )
}

/// The result of a command the timeout stopped. The first line still starts
/// `Command timed out after`; what the command printed before the timeout
/// follows, already sanitized and redacted like any result, and cut to its
/// last `max_output` bytes if it must be: a hang shows at the end.
/// `requested_secs` is the timeout the model passed, if it passed one: a
/// configured default above `bash_max_timeout` was never asked for.
fn timed_out_message(
    timeout_secs: u64,
    requested_secs: Option<u64>,
    output: &str,
    max_output: usize,
) -> String {
    let mut message = format!("Command timed out after {timeout_secs}s and was stopped");
    if let Some(requested_secs) = requested_secs.filter(|&secs| secs > timeout_secs) {
        message.push_str(&format!(
            " ({requested_secs}s was requested; the maximum is {timeout_secs}s)"
        ));
    }
    message.push('.');
    if output.is_empty() {
        message.push_str(
            " No output was captured before the timeout; \
             output the program had buffered but not flushed is lost.",
        );
        return message;
    }
    let tail = tail_on_char_boundary(output, max_output);
    if tail.len() == output.len() {
        message.push_str(" Output before the timeout:\n");
    } else {
        message.push_str(&format!(
            " Output before the timeout (last {} of {} bytes):\n",
            tail.len(),
            output.len()
        ));
    }
    message.push_str(tail);
    message
}

/// The longest suffix of `text` that fits in `max_bytes` and starts on a
/// character boundary.
fn tail_on_char_boundary(text: &str, max_bytes: usize) -> &str {
    if text.len() <= max_bytes {
        return text;
    }
    let mut start = text.len() - max_bytes;
    while !text.is_char_boundary(start) {
        start += 1;
    }
    &text[start..]
}

/// Resolve the bash executable for the bash tool.
///
/// Unix: `bash` from PATH, as always. Windows: bash is usually NOT on PATH —
/// prefer an explicit PATH hit, then Git Bash's known install locations, then
/// WSL's `bash.exe` (runs inside the default distro) as a last resort.
pub(crate) fn bash_program() -> std::ffi::OsString {
    #[cfg(unix)]
    {
        "bash".into()
    }
    #[cfg(windows)]
    {
        let on_path = std::env::var_os("PATH")
            .map(|p| std::env::split_paths(&p).any(|d| d.join("bash.exe").is_file()))
            .unwrap_or(false);
        if on_path {
            return "bash".into();
        }
        for var in ["ProgramFiles", "ProgramFiles(x86)", "ProgramW6432"] {
            if let Some(pf) = std::env::var_os(var) {
                let candidate = std::path::Path::new(&pf).join(r"Git\bin\bash.exe");
                if candidate.is_file() {
                    return candidate.into_os_string();
                }
            }
        }
        let wsl = std::path::Path::new(r"C:\Windows\System32\bash.exe");
        if wsl.is_file() {
            return wsl.as_os_str().to_os_string();
        }
        // Nothing found — let spawn fail with a readable NotFound error.
        "bash".into()
    }
}

/// The command, behind a `sudo` shell function. The child has no terminal
/// (setsid), so a bare sudo could not prompt anyway. With the askpass
/// broker (Linux, see `sudo_askpass`) sudo asks the user through the
/// verified helper; without it (other platforms, embedders, setup failure)
/// sudo must not ask at all: `-n` fails with sudo's own "a password is
/// required", and NOPASSWD rules work either way. The password never
/// travels through the command's stdin, which is empty.
pub(crate) fn bash_script_with_secure_sudo(command: &str, askpass: bool) -> String {
    #[cfg(unix)]
    let call = if askpass {
        r#"command sudo -A "$@""#
    } else {
        r#"command sudo -n "$@""#
    };
    #[cfg(not(unix))]
    let call = {
        let _ = askpass;
        r#"command sudo -S -p '[sudo] password required: ' "$@""#
    };
    format!(
        r#"sudo() {{
    {call}
}}
{command}"#
    )
}

/// Resolve the PowerShell executable. Prefers pwsh (PowerShell 7+, the
/// cross-platform build with sane defaults) and falls back to the
/// Windows-bundled powershell.exe (5.1).
#[cfg(windows)]
pub(crate) fn powershell_program() -> std::ffi::OsString {
    let on_path = |exe: &str| {
        std::env::var_os("PATH")
            .map(|p| std::env::split_paths(&p).any(|d| d.join(exe).is_file()))
            .unwrap_or(false)
    };
    if on_path("pwsh.exe") {
        return "pwsh".into();
    }
    for var in ["ProgramFiles", "ProgramFiles(x86)", "ProgramW6432"] {
        if let Some(pf) = std::env::var_os(var) {
            let candidate = std::path::Path::new(&pf).join(r"PowerShell\7\pwsh.exe");
            if candidate.is_file() {
                return candidate.into_os_string();
            }
        }
    }
    // powershell.exe is always present on Windows (System32).
    "powershell".into()
}

/// Unix stub — the powershell tool is only registered on Windows, so this is
/// never called there; it exists so cross-compilation `cargo check` works.
#[cfg(not(windows))]
pub(crate) fn powershell_program() -> std::ffi::OsString {
    "pwsh".into()
}

#[async_trait::async_trait]
impl Tool for BashTool {
    fn origin(&self) -> crate::tools::ToolOrigin {
        crate::tools::ToolOrigin::Builtin
    }

    /// Explicitly NonIdempotent (Task 24): arbitrary shell side effects —
    /// serialized execution, no concurrency key.
    fn effect(&self) -> crate::tools::catalog::ToolEffect {
        crate::tools::catalog::ToolEffect::NonIdempotent
    }

    fn name(&self) -> &str {
        "bash"
    }

    fn description(&self) -> &str {
        "Execute a bash command and return its output. Use for running programs, installing packages, git operations, and any shell commands. Commands time out after 30 seconds by default; pass a larger timeout when needed. Stdin is empty. On Linux, if sudo needs a password, the user is asked in the TUI, shown the command sudo will run, and the password goes only to sudo, never to the model. Other password prompts are not answered."
    }

    fn parameters(&self) -> Value {
        json!({
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The bash command to execute"
                },
                "timeout": {
                    "type": "integer",
                    "description": "Timeout in seconds (default: 30). Use a larger value for long-running commands."
                }
            },
            "required": ["command"]
        })
    }

    async fn execute(&self, params: Value, ctx: ToolContext) -> Result<String> {
        let command = params["command"]
            .as_str()
            .ok_or_else(|| RuntimeError::Tool("Missing command parameter".to_string()))?;

        run_bash(command, params["timeout"].as_u64(), ctx, prepare_askpass()).await
    }
}

/// The production askpass broker, if this platform and binary support it.
fn prepare_askpass() -> Option<Askpass> {
    #[cfg(target_os = "linux")]
    {
        use super::sudo_askpass::broker::{AskpassConfig, PreparedBroker};
        let config = AskpassConfig::production()?;
        match PreparedBroker::prepare(&config) {
            Ok(broker) => Some(broker),
            Err(e) => {
                tracing::warn!("sudo askpass unavailable for this command: {e}");
                None
            }
        }
    }
    #[cfg(not(target_os = "linux"))]
    {
        None
    }
}

pub(crate) async fn run_bash(
    command: &str,
    timeout: Option<u64>,
    ctx: ToolContext,
    askpass: Option<Askpass>,
) -> Result<String> {
    let script = bash_script_with_secure_sudo(command, askpass.is_some());
    run_shell_command(&script, ShellSpec::Bash, timeout, ctx, askpass).await
}

/// Kills the command's whole process group unless the command finished on
/// its own. `child.kill()` stops only the shell: everything the command
/// started (a test runner's workers, `cmd &` jobs) used to keep running, and
/// keep using memory, after a timeout or cancel (S348). The command leads its
/// own group: `setsid` in `pre_exec`, and `systemd-run --scope` execs in
/// place, so the pid is the group id. Dropped before the `Child`, so the
/// leader is not reaped by Child drop before the kill. Explicit kill paths
/// signal/disarm first; normal wait disarms synchronously on completion.
struct KillGroupUnlessFinished {
    pgid: Option<u32>,
}

impl KillGroupUnlessFinished {
    fn new(pid: Option<u32>) -> Self {
        Self { pgid: pid }
    }

    /// The command exited by itself: nothing to kill.
    fn finished(&mut self) {
        self.pgid = None;
    }

    /// Signal and disarm BEFORE any explicit kill/wait can reap the leader.
    fn kill_now(&mut self) {
        #[cfg(unix)]
        if let Some(pgid) = self.pgid.take() {
            // SAFETY: killpg(2) on our own child's process group.
            unsafe {
                libc::killpg(pgid as libc::pid_t, libc::SIGKILL);
            }
        }
        #[cfg(not(unix))]
        {
            self.pgid = None;
        }
    }
}

impl Drop for KillGroupUnlessFinished {
    fn drop(&mut self) {
        self.kill_now();
    }
}

/// Which shell backend the shared executor drives.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ShellSpec {
    Bash,
    PowerShell,
}

impl ShellSpec {
    /// The program and argv that run `script`.
    ///
    /// PowerShell gets `-ExecutionPolicy Bypass`: a stock Windows client has
    /// the `Restricted` policy, so `.\script.ps1` failed with "running
    /// scripts is disabled on this system". The flag is process-scoped, and
    /// PowerShells the command starts inherit it (through
    /// `PSExecutionPolicyPreference`). A policy set by Group Policy still
    /// wins. The policy is not a security boundary: the tool already runs any
    /// command as the user.
    fn invocation(self, script: &str) -> (std::ffi::OsString, Vec<std::ffi::OsString>) {
        match self {
            ShellSpec::Bash => (bash_program(), vec!["-c".into(), script.into()]),
            ShellSpec::PowerShell => (
                powershell_program(),
                ["-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script]
                    .into_iter()
                    .map(Into::into)
                    .collect(),
            ),
        }
    }

    /// Names the command's systemd scope (`systemctl --user list-units --type=scope`).
    fn scope_description(self) -> &'static str {
        match self {
            ShellSpec::Bash => "synaps tool: bash",
            ShellSpec::PowerShell => "synaps tool: powershell",
        }
    }
}

/// Shared execution core for bash + powershell: empty stdin, piped
/// stdout/stderr, chunked streaming with byte-conservation accounting, the
/// sudo askpass broker (bash on Linux), truncation, timeout. The only
/// per-shell differences are which program is spawned and which args it
/// takes. Nothing the command prints can raise a password prompt: only a
/// verified sudo can, through `askpass`.
pub(crate) async fn run_shell_command(
    script: &str,
    spec: ShellSpec,
    requested_timeout: Option<u64>,
    ctx: ToolContext,
    askpass: Option<Askpass>,
) -> Result<String> {
    {
        let timeout_secs = requested_timeout.unwrap_or(ctx.limits.bash_timeout);
        // H5: enforce bash_max_timeout cap — prevent DoS via prompt injection
        // requesting unbounded timeouts (e.g. timeout:2592000 + infinite loop).
        let timeout_secs = if ctx.limits.bash_max_timeout > 0 {
            timeout_secs.min(ctx.limits.bash_max_timeout)
        } else {
            timeout_secs
        };
        let max_output = ctx.limits.max_tool_buffer;

        let (program, args) = spec.invocation(script);
        let effective_env = ctx.capabilities.env.as_ref()
            .map(|env| env.spawn_vars().map(|(k, v)| (k.into(), v.into())).collect())
            .unwrap_or_else(crate::process_scope::inherited_env);
        let probe_env = effective_env.clone();
        let probe_cwd = ctx.capabilities.cwd.clone();
        // This worker can only launch a bounded throwaway probe. Abandoning
        // this future can never launch the real command from the worker.
        if ctx.capabilities.launch_cancel.as_ref().is_some_and(|c| c.is_cancelled()) {
            return Err(RuntimeError::Tool("Command canceled before launch".into()));
        }
        let probe_task = tokio::task::spawn_blocking(move || {
            crate::process_scope::launcher_for_effective_env(
                &probe_env, probe_cwd.as_deref(), crate::process_scope::ScopeKind::Tool)
        });
        let launcher = if let Some(cancel) = &ctx.capabilities.launch_cancel {
            tokio::select! {
                biased;
                _ = cancel.cancelled() => return Err(RuntimeError::Tool("Command canceled before launch".into())),
                result = probe_task => result,
            }
        } else {
            probe_task.await
        }.map_err(|e| RuntimeError::Tool(e.to_string()))?;
        let (program, args) = match launcher {
            Some(launcher) => launcher.wrap(spec.scope_description(),
                crate::process_scope::resolve_program(program, &effective_env,
                    ctx.capabilities.cwd.as_deref()).map_err(|e| RuntimeError::Tool(e.to_string()))?, args),
            None => (program, args),
        };
        let mut cmd = tokio::process::Command::new(program);
        if let Some(cwd) = ctx.capabilities.cwd.as_deref() {
            cmd.current_dir(cwd);
        }
        cmd.env_clear().envs(effective_env);
        #[cfg(target_os = "linux")]
        if let Some(broker) = &askpass {
            cmd.env("SUDO_ASKPASS", broker.helper_path());
        }
        cmd.args(&args)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);

        // Start the child in a new session (setsid) so it has no controlling
        // terminal. Programs that open /dev/tty directly (SSH fingerprint
        // prompts, gpg pinentry, git credential helpers, pagers) will get
        // ENXIO and fail with a readable error on stderr instead of hanging
        // invisibly until timeout. Sudo asks through its askpass helper
        // instead (see bash_script_with_secure_sudo()).
        #[cfg(unix)]
        unsafe {
            cmd.pre_exec(|| {
                libc::setsid();
                Ok(())
            });
        }

        if ctx.capabilities.launch_cancel.as_ref().is_some_and(|c| c.is_cancelled()) {
            return Err(RuntimeError::Tool("Command canceled before launch".into()));
        }
        let mut child = cmd.spawn().map_err(|e| RuntimeError::Tool(e.to_string()))?;
        // Armed until the command finishes on its own: a timeout, a cancel,
        // the output cap or an abandoned password prompt kills the whole
        // process group, not just the shell.
        let mut group = KillGroupUnlessFinished::new(child.id());

        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| RuntimeError::Tool("Failed to capture stdout".to_string()))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| RuntimeError::Tool("Failed to capture stderr".to_string()))?;
        // Serve sudo's password requests for as long as the command runs.
        let redactions = Redactions::default();
        let password_cancel = crate::CancellationToken::new();
        #[cfg(target_os = "linux")]
        let _askpass = askpass.and_then(|broker| {
            broker
                .start(
                    child.id(),
                    ctx.capabilities.secret_prompt.clone(),
                    redactions.clone(),
                    password_cancel.clone(),
                )
                .map_err(|e| tracing::warn!("sudo askpass unavailable for this command: {e}"))
                .ok()
        });
        #[cfg(not(target_os = "linux"))]
        let _ = askpass;

        let (tx_inter, mut rx_inter) =
            tokio::sync::mpsc::channel::<(bool, HandoffChunk)>(BASH_INTERMEDIARY_CHANNEL_CAPACITY);

        tokio::spawn(forward_output(stdout, false, tx_inter.clone()));
        tokio::spawn(forward_output(stderr, true, tx_inter.clone()));

        drop(tx_inter);

        // Outside the timed future: a timeout drops that future, and what
        // the command printed before it must outlive it.
        let mut full_output = String::new();
        let result = tokio::time::timeout(tokio::time::Duration::from_secs(timeout_secs), async {
            let mut truncated = false;
            let mut streamed_bytes = 0usize;

            loop {
                let item = tokio::select! {
                    biased;
                    // The user cancelled sudo's password prompt: stop the
                    // command rather than let it run on without it.
                    _ = password_cancel.cancelled() => {
                        group.kill_now();
                        let _ = child.kill().await;
                        return Err(RuntimeError::Tool(
                            "Command canceled while waiting for password".to_string(),
                        ));
                    }
                    item = rx_inter.recv() => item,
                };
                let Some((_is_stderr, chunk)) = item else {
                    break;
                };
                let mut msg = chunk.consume();
                for secret in redactions.lock().unwrap_or_else(|p| p.into_inner()).iter() {
                    if !secret.is_empty() {
                        msg = msg.replace(secret.as_str(), "[redacted]");
                    }
                }

                if truncated {
                    continue;
                }

                let kept_from = full_output.len();
                let added_all = append_bounded(&mut full_output, &msg, max_output);
                if let Some(ref txd) = ctx.channels.tx_delta {
                    if streamed_bytes < MAX_STREAMED_DELTA_BYTES {
                        // Stream only what the result keeps, never the
                        // bytes past the cap.
                        let kept = &full_output[kept_from..];
                        let remaining = MAX_STREAMED_DELTA_BYTES - streamed_bytes;
                        let delta = crate::truncate_str(kept, remaining).to_string();
                        streamed_bytes += delta.len();
                        if !delta.is_empty() {
                            txd.send(delta);
                        }
                    }
                }

                if !added_all {
                    // The command is stopped, not its output shortened for
                    // display: say so, in the result and the stream alike.
                    let notice = output_limit_notice(max_output);
                    full_output.push_str(&notice);
                    if let Some(ref txd) = ctx.channels.tx_delta {
                        txd.send(notice);
                    }
                    truncated = true;
                    group.kill_now();
                    let _ = child.kill().await;
                }
            }
            let status = child.wait().await;
            // wait errors may mean the leader was already reaped externally;
            // never signal a potentially recycled process-group ID afterwards.
            group.finished();
            let status = status.map_err(|e| RuntimeError::Tool(e.to_string()))?;
            // Finished on its own: leave anything it deliberately started in
            // the background (`server &`) running, as before.
            // Drop (and so zeroize) the passwords now the command is done.
            redactions.lock().unwrap_or_else(|p| p.into_inner()).clear();
            Ok::<_, RuntimeError>((status, std::mem::take(&mut full_output), truncated))
        })
        .await;

        match result {
            Ok(Ok((status, output, was_truncated))) => {
                if status.success() || was_truncated {
                    Ok(output)
                } else {
                    // T5: when a command fails and references a stripped secret,
                    // append a notice (once per name per session). Word-boundary
                    // match: covers $NAME, ${NAME}, ${NAME:?}, ${NAME:-x},
                    // $env:NAME and NAME=…
                    let notice = ctx.capabilities.stripped_env_notices(|name| references_env_name(script, name));
                    Err(RuntimeError::Tool(format!(
                        "Command failed (exit {}):\n{}{notice}",
                        status.code().unwrap_or(-1),
                        output
                    )))
                }
            }
            Ok(Err(e)) => Err(RuntimeError::Tool(format!(
                "Failed to execute command: {}",
                e
            ))),
            // The group is still armed: `group` drops on this return and
            // kills it (Unix), exactly as before. Elsewhere only the shell
            // is killed, by `kill_on_drop`.
            Err(_) => Err(RuntimeError::Tool(timed_out_message(
                timeout_secs,
                requested_timeout,
                &full_output,
                max_output,
            ))),
        }
    }
}

/// Does `script` mention `name` as a whole identifier (not as a substring of
/// a longer one)? Used by the T5 stripped-secret notice; a false positive
/// costs one advisory line on an already-failing command, a false negative
/// hides the reason the command failed — so err towards matching.
fn references_env_name(script: &str, name: &str) -> bool {
    let is_ident = |c: char| c.is_ascii_alphanumeric() || c == '_';
    let mut from = 0;
    while let Some(pos) = script[from..].find(name) {
        let start = from + pos;
        let end = start + name.len();
        let before_ok = !script[..start].chars().next_back().is_some_and(is_ident);
        let after_ok = !script[end..].chars().next().is_some_and(is_ident);
        if before_ok && after_ok {
            return true;
        }
        from = end;
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A stock Windows client has the `Restricted` execution policy: without
    /// Bypass the powershell tool could not run `.\script.ps1`. The script is
    /// the last argument, after `-Command`, exactly as given.
    #[test]
    fn powershell_runs_with_execution_policy_bypass() {
        let (_, args) = ShellSpec::PowerShell.invocation(r".\test.ps1");
        let args: Vec<_> = args.iter().map(|a| a.to_string_lossy().into_owned()).collect();
        assert_eq!(args, ["-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", r".\test.ps1"]);
        let (_, args) = ShellSpec::Bash.invocation("echo hi");
        assert_eq!(args, [std::ffi::OsString::from("-c"), "echo hi".into()]);
    }

    #[test]
    fn explicit_child_kills_signal_and_disarm_group_before_reaping() {
        let source = include_str!("bash.rs");
        let executor = source.split("pub(crate) async fn run_shell_command").nth(1).unwrap()
            .split("#[cfg(test)]").next().unwrap();
        let kills: Vec<_> = executor.match_indices("let _ = child.kill().await;").collect();
        assert_eq!(kills.len(), 2, "audit every new explicit kill path");
        for (offset, _) in kills {
            assert!(executor[..offset].trim_end().ends_with("group.kill_now();"));
        }
        assert!(executor.contains("let status = child.wait().await;"));
        let wait = executor.split("let status = child.wait().await;").nth(1).unwrap();
        assert!(wait.find("group.finished();").unwrap() < wait.find("status.map_err").unwrap());
        assert!(executor.find("let mut child = cmd.spawn()").unwrap()
            < executor.find("let mut group = KillGroupUnlessFinished::new").unwrap());
    }

    #[tokio::test]
    async fn canceled_launch_never_runs_payload() {
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("must-not-exist");
        let token = crate::CancellationToken::new();
        token.cancel();
        let mut ctx = create_tool_context();
        ctx.capabilities.launch_cancel = Some(token);
        let error = BashTool.execute(json!({"command": format!("touch '{}'", marker.display())}), ctx)
            .await.unwrap_err().to_string();
        assert!(error.contains("canceled before launch"), "{error}");
        assert!(!marker.exists());
    }

    #[cfg(target_os = "linux")]
    fn own_cgroup() -> String {
        std::fs::read_to_string("/proc/self/cgroup")
            .unwrap_or_default()
            .trim()
            .to_string()
    }

    /// The command runs in its own systemd scope when the machine supports
    /// it, so an out-of-memory kill aimed at it cannot take the daemon's
    /// cgroup down with it (S348). Without a reachable user manager (CI
    /// containers), the command must still run, in this process's cgroup.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn bash_command_gets_its_own_scope_when_available() {
        if std::env::var("SYNAPS_TEST_REAL_SCOPES").as_deref() != Ok("1") { return; }
        let out = BashTool
            .execute(
                json!({ "command": "cat /proc/self/cgroup; echo pid=$$", "timeout": 30 }),
                create_tool_context(),
            )
            .await
            .unwrap();
        let child_cgroup = out.lines().next().unwrap_or_default().trim().to_string();
        assert!(child_cgroup.starts_with("0::/"), "cgroup v2 line: {out}");
        if crate::process_scope::launcher_for_env(None).is_some() {
            assert_ne!(
                child_cgroup,
                own_cgroup(),
                "the command must leave our cgroup"
            );
            assert!(child_cgroup.ends_with(".scope"), "{child_cgroup}");
            assert!(child_cgroup.contains("/app.slice/"), "{child_cgroup}");
        } else {
            assert_eq!(child_cgroup, own_cgroup(), "unwrapped fallback");
        }
    }

    /// A bash command's scope goes in `tool_scope_slice`, never in the
    /// daemon's slice; the probe carries the same `--slice`.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn a_bash_command_is_scoped_in_the_tool_slice() {
        use crate::process_scope::test_support::{fake_env, fake_systemd_run, recorded};
        let base = crate::test_env::BaseDirGuard::new();
        let _scopes = crate::test_env::EnvVarGuard::remove("SYNAPS_PROCESS_SCOPES");
        std::fs::write(
            base.path().join("config"),
            "tool_scope_slice = t_tools.slice\ndaemon_scope_slice = d_daemon.slice\n",
        )
        .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let fake = fake_systemd_run(dir.path());
        let mut env = fake_env(&fake);
        env.push(("PATH".to_string(), std::env::var("PATH").unwrap_or_default()));
        let mut ctx = create_tool_context();
        ctx.capabilities.env = Some(env.into());
        let out = BashTool
            .execute(json!({ "command": "echo scoped-run", "timeout": 30 }), ctx)
            .await
            .unwrap();
        assert!(out.contains("scoped-run"), "{out}");
        let launch = recorded(dir.path(), "launch.args");
        let slices: Vec<&String> = launch.iter().filter(|a| a.starts_with("--slice")).collect();
        assert_eq!(slices, ["--slice=t_tools.slice"], "{launch:?}");
        assert!(recorded(dir.path(), "probe.args").contains(&"--slice=t_tools.slice".to_string()));
    }

    /// A session environment rebuilt without the user-bus variables cannot
    /// reach systemd: the command starts unwrapped instead of failing.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn bash_command_with_a_rebuilt_env_lacking_the_user_bus_runs_unwrapped() {
        let mut ctx = create_tool_context();
        ctx.capabilities.env = Some(
            vec![("PATH".to_string(), std::env::var("PATH").unwrap_or_default())].into(),
        );
        let out = BashTool
            .execute(
                json!({ "command": "cat /proc/self/cgroup", "timeout": 30 }),
                ctx,
            )
            .await
            .unwrap();
        assert_eq!(out.lines().next().unwrap_or_default().trim(), own_cgroup());
    }

    #[cfg(target_os = "linux")]
    fn processes_named(name: &str) -> Vec<String> {
        let out = std::process::Command::new("pgrep")
            .args(["-f", name])
            .output()
            .unwrap();
        String::from_utf8_lossy(&out.stdout)
            .split_whitespace()
            .map(str::to_string)
            .collect()
    }

    /// A timed-out command takes everything it started with it, not just the
    /// shell (S348: a timed-out test run would leave all its workers running).
    /// Each child is renamed to the marker (`exec -a`), so a survivor is seen.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn bash_timeout_kills_the_whole_process_group() {
        let marker = format!("synaps-timeout-child-{}", std::process::id());
        let out = BashTool
            .execute(
                json!({
                    "command": format!("(exec -a {marker} sleep 300) & (exec -a {marker} sleep 300); wait"),
                    "timeout": 1
                }),
                create_tool_context(),
            )
            .await;
        let text = match out {
            Ok(s) => s,
            Err(e) => e.to_string(),
        };
        assert!(text.contains("timed out"), "{text}");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
        while !processes_named(&marker).is_empty() && std::time::Instant::now() < deadline {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        let left = processes_named(&marker);
        if !left.is_empty() {
            let _ = std::process::Command::new("pkill")
                .args(["-f", &marker])
                .status();
        }
        assert!(left.is_empty(), "children survived the timeout: {left:?}");
    }

    /// A command that exits by itself keeps what it deliberately started in
    /// the background, as before: only abnormal endings kill the group.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn bash_normal_exit_leaves_a_background_job_running() {
        let marker = format!("synaps-background-job-{}", std::process::id());
        let out = BashTool
            .execute(
                json!({
                    "command": format!("(exec -a {marker} sleep 30) >/dev/null 2>&1 & echo started"),
                    "timeout": 10
                }),
                create_tool_context(),
            )
            .await
            .unwrap();
        assert!(out.contains("started"), "{out}");
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        let alive = processes_named(&marker);
        let _ = std::process::Command::new("pkill")
            .args(["-f", &marker])
            .status();
        assert!(
            !alive.is_empty(),
            "a deliberate background job must survive a normal exit"
        );
    }

    #[tokio::test]
    async fn bash_intermediary_handoff_conserves_bytes_under_large_output() {
        let before = bash_intermediary_snapshot();
        let tool = BashTool;
        let mut ctx = create_tool_context();
        ctx.limits.max_tool_buffer = 64 * 1024;
        let result = tool
            .execute(
                json!({
                    "command": "python3 -c \"import sys; sys.stdout.write('x' * 1048576)\"",
                    "timeout": 30
                }),
                ctx,
            )
            .await
            .unwrap();
        // Stopped at the cap: the result is the cap's worth of output and
        // the notice saying so, nothing more (the memory bound holds).
        let notice = output_limit_notice(64 * 1024);
        assert!(result.ends_with(&notice));
        assert_eq!(result.len(), 64 * 1024 + notice.len());

        // The counters are process-global and sibling tests run bash
        // concurrently, so a before/after delta is skewed by whatever a
        // sibling had mid-channel when `before` was read, and that skew
        // never rebalances. Assert the invariant that does hold: once
        // in-flight relays drain, every byte ever produced in this process
        // was consumed or dropped. Siblings only delay this; a leaked byte
        // keeps it false forever.
        let after = wait_handoff_settled().await;
        assert_eq!(after.retained_bytes, 0);
        assert!(after.accepted_bytes >= after.consumed_bytes);
        assert!(after.produced_bytes - before.produced_bytes >= 64 * 1024);
    }

    #[test]
    fn sanitizes_terminal_control_sequences_and_nuls() {
        let cleaned = sanitize_output("ok\x1b[2J\x00done");
        assert_eq!(cleaned, "okdone");
    }

    use super::super::test_helpers::create_tool_context;
    use crate::tools::Tool;
    use serde_json::json;

    #[test]
    fn test_bash_tool_schema() {
        let tool = BashTool;
        assert_eq!(tool.name(), "bash");
        assert!(!tool.description().is_empty());

        let params = tool.parameters();
        assert_eq!(params["type"], "object");
        assert!(params["properties"].is_object());
        assert!(params["required"].is_array());
    }

    #[tokio::test]
    async fn test_bash_tool_execution() {
        let tool = BashTool;

        // Test simple echo command
        let ctx = create_tool_context();
        let params = json!({
            "command": "echo hello"
        });

        let result = tool.execute(params, ctx).await.unwrap();
        assert!(result.contains("hello"));

        // Test timeout parameter with quick command
        let ctx = create_tool_context();
        let params = json!({
            "command": "sleep 1",
            "timeout": 2
        });

        let result = tool.execute(params, ctx).await;
        // Should succeed (1 second sleep with 2 second timeout)
        assert!(result.is_ok());

        // Test timeout with longer command
        let ctx = create_tool_context();
        let params = json!({
            "command": "sleep 3",
            "timeout": 1
        });

        let result = tool.execute(params, ctx).await;
        // Should timeout
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("timed out"));
    }

    #[tokio::test]
    async fn test_bash_tool_requested_timeout_is_clamped_by_max_timeout() {
        let tool = BashTool;
        let mut ctx = create_tool_context();
        ctx.limits.bash_max_timeout = 1;

        let params = json!({
            "command": "sleep 5; echo done",
            "timeout": 10
        });

        let error = tool_error(tool.execute(params, ctx).await);
        assert!(
            error.contains("timed out"),
            "requested timeout MUST be clamped by bash_max_timeout"
        );
        // The model is told it got less time than it asked for.
        assert!(
            error.starts_with(
                "Tool execution failed: Command timed out after 1s and was stopped \
                 (10s was requested; the maximum is 1s)."
            ),
            "{error}"
        );
    }

    /// The configured default above `bash_max_timeout` is cut too, but the
    /// model asked for nothing, so the error does not say it did.
    #[tokio::test]
    async fn a_default_timeout_cut_by_the_maximum_is_not_called_requested() {
        let mut ctx = create_tool_context();
        ctx.limits.bash_timeout = 10;
        ctx.limits.bash_max_timeout = 1;

        let error = tool_error(BashTool.execute(json!({"command": "sleep 5"}), ctx).await);
        let head = "Tool execution failed: Command timed out after 1s and was stopped. ";
        assert!(error.starts_with(head), "{error}");
        assert!(!error.contains("requested"), "{error}");
    }

    fn prompt_channel() -> (
        crate::tools::SecretPromptHandle,
        tokio::sync::mpsc::UnboundedReceiver<crate::tools::SecretPromptRequest>,
    ) {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        (crate::tools::SecretPromptHandle::new(tx), rx)
    }

    /// Output that merely looks like a password prompt. The first line is
    /// what a diagnostic printing the tool's own command lines showed on
    /// the day this misfired: the old wrapper, prompt string included.
    const PROMPT_LOOKALIKES: &str = r#"printf '%s\n' "command sudo -S -p '[sudo] password required: ' \"\$@\"" >&2
printf '[sudo] password for x: ' >&2
printf 'Password: ' >&2
if read -r got; then echo "stdin gave: $got"; else echo stdin-eof; fi"#;

    /// Regression: text on stderr can no longer raise a password prompt,
    /// and nothing is ever written to the command's stdin (it is empty).
    #[tokio::test]
    async fn sudo_prompt_lookalike_output_raises_no_prompt_and_feeds_no_stdin() {
        let (handle, mut prompts) = prompt_channel();
        let mut ctx = create_tool_context();
        ctx.capabilities.secret_prompt = Some(handle);
        let started = std::time::Instant::now();
        let out = BashTool
            .execute(json!({"command": PROMPT_LOOKALIKES, "timeout": 30}), ctx)
            .await
            .unwrap();
        assert!(out.contains("stdin-eof"), "{out}");
        assert!(out.contains("[sudo] password for x: "), "it is only output: {out}");
        assert!(out.contains("password required: "), "{out}");
        assert!(prompts.try_recv().is_err(), "no prompt was raised");
        assert!(started.elapsed() < std::time::Duration::from_secs(10));
    }

    #[cfg(unix)]
    #[test]
    fn the_sudo_wrapper_asks_through_askpass_or_not_at_all() {
        let script = super::bash_script_with_secure_sudo("sudo id", true);
        assert_eq!(script, "sudo() {\n    command sudo -A \"$@\"\n}\nsudo id");
        let script = super::bash_script_with_secure_sudo("sudo id", false);
        assert_eq!(script, "sudo() {\n    command sudo -n \"$@\"\n}\nsudo id");
        for askpass in [true, false] {
            let script = super::bash_script_with_secure_sudo("x", askpass);
            assert!(!script.contains(" -S") && !script.contains("password"), "{script}");
        }
    }

    /// Without the broker (this test binary does not dispatch the helper),
    /// sudo is told never to ask: it fails with its own error, no prompt.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn without_askpass_sudo_never_asks() {
        let fake = tempfile::tempdir().unwrap();
        crate::tools::sudo_askpass::testing::write_fixtures(fake.path()).unwrap();
        let (handle, mut prompts) = prompt_channel();
        let mut ctx = create_tool_context();
        ctx.capabilities.secret_prompt = Some(handle);
        let command = format!(
            "export PATH=\"{}:$PATH\"; sudo echo ran; echo after:$?",
            fake.path().display()
        );
        let out = BashTool
            .execute(json!({"command": command, "timeout": 30}), ctx)
            .await
            .unwrap();
        assert!(out.contains("sudo: a password is required"), "{out}");
        assert!(out.contains("after:1") && !out.contains("ran\n"), "{out}");
        assert!(prompts.try_recv().is_err());
    }

    #[tokio::test]
    async fn test_bash_control_char_output_is_sanitized_and_bounded_in_deltas() {
        let tool = BashTool;
        let channel = crate::tools::output::delta_channel();
        let (delta_tx, mut delta_rx) = (channel.sender, channel.receiver);
        let mut ctx = create_tool_context();
        ctx.channels.tx_delta = Some(delta_tx);
        ctx.limits.max_tool_buffer = 256;

        let params = json!({
            "command": "python3 -c \"import sys; sys.stdout.buffer.write(b'clean\\x1b[2J\\x00' + b'A' * 2000); sys.stdout.flush()\"",
            "timeout": 30
        });

        let result = tool.execute(params, ctx).await.unwrap();
        let mut streamed = String::new();
        while let Some(delta) = delta_rx.try_drain() {
            streamed.push_str(&delta);
        }

        assert!(result.ends_with(&output_limit_notice(256)), "{result}");
        // The stream carries exactly what the result keeps, notice included.
        assert_eq!(streamed, result);
        assert!(!result.contains('\u{1b}'));
        assert!(!result.contains('\0'));
        assert!(!streamed.contains('\u{1b}'));
        assert!(!streamed.contains('\0'));
        assert!(
            streamed.len() <= 2048,
            "streamed deltas must be bounded, got {} bytes",
            streamed.len()
        );
    }

    #[tokio::test]
    async fn test_bash_binary_output_is_sanitized() {
        let tool = BashTool;
        let ctx = create_tool_context();
        let params = json!({
            "command": "python3 -c \"import sys; sys.stdout.buffer.write(bytes(range(32)) + b'visible')\"",
            "timeout": 30
        });

        let result = tool.execute(params, ctx).await.unwrap();

        assert!(result.contains("visible"));
        assert!(!result.contains('\0'));
        assert!(!result.contains('\u{1b}'));
    }

    #[tokio::test]
    async fn test_bash_tool_timeout() {
        let tool = BashTool;
        let ctx = create_tool_context();

        let params = json!({
            "command": "sleep 10",
            "timeout": 1
        });

        let error = tool_error(tool.execute(params, ctx).await);
        // Still an error starting `Command timed out after`; a command that
        // printed nothing says so instead of showing an empty section.
        assert_eq!(
            error,
            "Tool execution failed: Command timed out after 1s and was stopped. \
             No output was captured before the timeout; \
             output the program had buffered but not flushed is lost."
        );
    }

    #[tokio::test]
    async fn test_bash_tool_failure() {
        let tool = BashTool;
        let ctx = create_tool_context();

        let params = json!({
            "command": "exit 1"
        });

        let result = tool.execute(params, ctx).await;

        // Should fail and return error
        assert!(result.is_err());
        let error = result.unwrap_err().to_string();
        assert!(error.contains("failed") || error.contains("exit"));
    }

    /// The text the model reads for a failed call.
    fn tool_error(result: crate::Result<String>) -> String {
        match result {
            Ok(out) => panic!("expected an error, got: {out}"),
            Err(e) => e.to_string(),
        }
    }

    /// The exact words the model reads when the output cap stops a command.
    #[test]
    fn the_output_limit_notice_says_the_command_was_stopped() {
        assert_eq!(
            output_limit_notice(256),
            "\n\n[output truncated at 256 bytes: the command was stopped at this limit \
             before it finished, so its exit status is unknown. Redirect large output \
             to a file (e.g. cmd > out.log 2>&1) and read only the parts you need.]"
        );
    }

    /// Past the cap the command is killed and the result says so: the cap's
    /// worth of output, then the notice, in the result, the stream and the
    /// history the stream feeds alike. Still a success result (unchanged),
    /// and back at once, not at the end of the command or the timeout.
    #[cfg(unix)]
    #[tokio::test]
    async fn the_output_cap_stops_the_command_and_says_its_status_is_unknown() {
        let channel = crate::tools::output::delta_channel();
        let history = channel.output_handle();
        let (delta_tx, mut delta_rx) = (channel.sender, channel.receiver);
        let mut ctx = create_tool_context();
        ctx.channels.tx_delta = Some(delta_tx);
        ctx.limits.max_tool_buffer = 256;
        let started = std::time::Instant::now();
        let result = BashTool
            .execute(
                json!({"command": "printf 'x%.0s' {1..2000}; sleep 30; echo finished", "timeout": 30}),
                ctx,
            )
            .await
            .unwrap();
        assert!(
            started.elapsed() < std::time::Duration::from_secs(20),
            "stopped at the cap"
        );
        assert_eq!(result, "x".repeat(256) + &output_limit_notice(256));
        assert!(result.contains("the command was stopped"), "{result}");
        assert!(result.contains("its exit status is unknown"), "{result}");
        let mut streamed = String::new();
        while let Some(delta) = delta_rx.try_drain() {
            streamed.push_str(&delta);
        }
        assert_eq!(streamed, result);
        assert_eq!(history.model_history().text, result);
    }

    /// Output up to the cap, the cap exactly included, comes back whole and
    /// unmarked; one byte more stops the command.
    #[cfg(unix)]
    #[tokio::test]
    async fn output_up_to_the_cap_is_unchanged_and_one_byte_more_is_stopped() {
        for (bytes, expected) in [
            (255, "y".repeat(255)),
            (256, "y".repeat(256)),
            (257, "y".repeat(256) + &output_limit_notice(256)),
        ] {
            let mut ctx = create_tool_context();
            ctx.limits.max_tool_buffer = 256;
            let command = format!("printf 'y%.0s' {{1..{bytes}}}");
            let result = BashTool
                .execute(json!({"command": command, "timeout": 30}), ctx)
                .await
                .unwrap();
            assert_eq!(result, expected, "{bytes} bytes");
        }
    }

    /// A command that fails after a little output is still a failure with
    /// its exit status: the cap only speaks when it stopped something.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_nonzero_exit_under_the_cap_is_still_a_failure() {
        let mut ctx = create_tool_context();
        ctx.limits.max_tool_buffer = 256;
        let error = tool_error(
            BashTool
                .execute(
                    json!({"command": "printf partial; exit 3", "timeout": 30}),
                    ctx,
                )
                .await,
        );
        assert_eq!(
            error,
            "Tool execution failed: Command failed (exit 3):\npartial"
        );
    }

    #[test]
    fn the_timeout_message_names_the_stop_and_keeps_the_output() {
        assert_eq!(
            timed_out_message(30, None, "", 256),
            "Command timed out after 30s and was stopped. No output was captured before \
             the timeout; output the program had buffered but not flushed is lost."
        );
        assert_eq!(
            timed_out_message(30, Some(30), "step 1\nstep 2\n", 256),
            "Command timed out after 30s and was stopped. Output before the timeout:\n\
             step 1\nstep 2\n"
        );
        assert_eq!(
            timed_out_message(300, Some(600), "x", 256),
            "Command timed out after 300s and was stopped (600s was requested; \
             the maximum is 300s). Output before the timeout:\nx"
        );
        // No timeout from the model: a default the maximum cut was never
        // requested, so the message does not say it was.
        assert_eq!(
            timed_out_message(300, None, "x", 256),
            "Command timed out after 300s and was stopped. Output before the timeout:\nx"
        );
        let long = format!("{}END", "a".repeat(1000));
        assert_eq!(
            timed_out_message(1, Some(1), &long, 10),
            "Command timed out after 1s and was stopped. \
             Output before the timeout (last 10 of 1003 bytes):\naaaaaaaEND"
        );
    }

    /// The kept tail starts on a character boundary for every budget, and is
    /// the longest one that fits.
    #[test]
    fn the_timeout_tail_is_cut_on_a_character_boundary() {
        let text = "ab▀é😀c▀";
        for max in 0..=text.len() + 1 {
            let tail = tail_on_char_boundary(text, max);
            assert!(
                text.ends_with(tail) && tail.len() <= max,
                "max {max}: {tail:?}"
            );
            let before = &text[..text.len() - tail.len()];
            let next = before.chars().next_back().map_or(0, char::len_utf8);
            assert!(
                before.is_empty() || tail.len() + next > max,
                "max {max}: {tail:?}"
            );
        }
        assert_eq!(
            timed_out_message(1, None, "▀▀▀", 4),
            "Command timed out after 1s and was stopped. \
             Output before the timeout (last 3 of 9 bytes):\n▀"
        );
    }

    /// A timed-out command's error carries what it printed before the
    /// timeout, stdout and stderr, exactly as it was streamed.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_timeout_keeps_the_output_printed_before_it() {
        let channel = crate::tools::output::delta_channel();
        let (delta_tx, mut delta_rx) = (channel.sender, channel.receiver);
        let mut ctx = create_tool_context();
        ctx.channels.tx_delta = Some(delta_tx);
        let error = tool_error(
            BashTool
                .execute(
                    json!({"command": "echo out-before; echo err-before >&2; sleep 30", "timeout": 1}),
                    ctx,
                )
                .await,
        );
        let head = "Tool execution failed: Command timed out after 1s and was stopped. \
                    Output before the timeout:\n";
        assert!(error.starts_with(head), "{error}");
        let output = &error[head.len()..];
        assert!(
            output.contains("out-before\n") && output.contains("err-before\n"),
            "{error}"
        );
        assert_eq!(output.len(), "out-before\nerr-before\n".len(), "{error}");
        let mut streamed = String::new();
        while let Some(delta) = delta_rx.try_drain() {
            streamed.push_str(&delta);
        }
        assert_eq!(output, streamed);
    }

    /// The cap kills the command's group, but a process that left the group
    /// (setsid) still holds the pipes, so the call ends at the timeout. The
    /// error stays within `max_output`: the tail of what was captured, the
    /// cap's notice included, cut inside a 3-byte character and moved to
    /// the next whole one.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_timeout_after_a_flood_keeps_a_bounded_tail() {
        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("holder.pid");
        let command = format!(
            r#"python3 -c "import os, sys, time; os.setsid(); open(sys.argv[1], 'w').write(str(os.getpid())); time.sleep(60)" '{pid}' &
while [ ! -s '{pid}' ]; do sleep 0.02; done
python3 -c "import sys; sys.stdout.buffer.write('▀'.encode() * 100000)"
wait"#,
            pid = pid_file.display()
        );
        let mut ctx = create_tool_context();
        ctx.limits.max_tool_buffer = 256;
        let error = tool_error(
            BashTool
                .execute(json!({"command": command, "timeout": 5}), ctx)
                .await,
        );
        if let Some(pid) = std::fs::read_to_string(&pid_file)
            .ok()
            .and_then(|p| p.trim().parse::<libc::pid_t>().ok())
        {
            // SAFETY: kills the holder this test started.
            unsafe {
                libc::kill(pid, libc::SIGKILL);
            }
        }
        let notice = output_limit_notice(256);
        let (head, tail) = error.split_once('\n').unwrap_or_else(|| panic!("{error}"));
        // 85 whole characters (255 bytes) fit under the cap, then the notice.
        let captured = 255 + notice.len();
        assert_eq!(
            head,
            format!(
                "Tool execution failed: Command timed out after 5s and was stopped. \
                 Output before the timeout (last {} of {captured} bytes):",
                tail.len()
            ),
        );
        assert!(
            tail.len() <= 256 && tail.len() > 256 - '▀'.len_utf8(),
            "{}",
            tail.len()
        );
        assert!(tail.ends_with(&notice), "{error}");
        let chars = &tail[..tail.len() - notice.len()];
        assert!(
            !chars.is_empty() && chars.chars().all(|c| c == '▀'),
            "{chars:?}"
        );
    }

    /// What a timed-out command printed goes through the same redaction as
    /// any result: a sudo password it echoed is never shown.
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn a_timeout_redacts_a_password_in_the_output_before_it() {
        let bin = tempfile::tempdir().unwrap();
        let helper = crate::tools::sudo_askpass::testing::write_fixtures(bin.path()).unwrap();
        let (handle, mut prompts) = prompt_channel();
        let user = tokio::spawn(async move {
            let request = prompts.recv().await.expect("sudo asks for the password");
            let _ = request.response_tx.send(Some("swordfish".to_string()));
        });
        let mut ctx = create_tool_context();
        ctx.capabilities.secret_prompt = Some(handle);
        let command = format!(
            "export PATH=\"{}:$PATH\"\nFAKE_SUDO_ECHO=1 sudo true; echo done; sleep 30",
            bin.path().display()
        );
        let error = tool_error(
            crate::tools::sudo_askpass::testing::run_bash_with_test_askpass(
                &command,
                Some(5),
                ctx,
                &helper,
            )
            .await,
        );
        user.await.unwrap();
        assert!(
            error.starts_with(
                "Tool execution failed: Command timed out after 5s and was stopped. \
                 Output before the timeout:\n"
            ),
            "{error}"
        );
        assert!(
            error.contains("sudo got: [redacted]\n") && error.contains("done\n"),
            "{error}"
        );
        assert!(!error.contains("swordfish"), "{error}");
    }

    /// Is `pid` still running? One that was killed but not yet reaped by its
    /// new parent (a zombie) is not.
    #[cfg(unix)]
    fn process_running(pid: libc::pid_t) -> bool {
        // SAFETY: signal 0 only checks that the process exists.
        if unsafe { libc::kill(pid, 0) } != 0 {
            return false;
        }
        #[cfg(target_os = "linux")]
        let state = std::fs::read_to_string(format!("/proc/{pid}/stat"))
            .ok()
            .and_then(|stat| {
                stat.rsplit_once(')')
                    .map(|(_, rest)| rest.trim_start().to_string())
            });
        #[cfg(not(target_os = "linux"))]
        let state = std::process::Command::new("ps")
            .args(["-o", "stat=", "-p", &pid.to_string()])
            .output()
            .ok()
            .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_string());
        state.is_some_and(|s| !s.is_empty() && !s.starts_with('Z'))
    }

    /// After a timeout nothing the command started is left running: a child
    /// that records its pid and sleeps is gone, not just the shell.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_timeout_leaves_no_process_of_the_command_running() {
        let dir = tempfile::tempdir().unwrap();
        let pid_file = dir.path().join("child.pid");
        let command = format!(
            "sleep 300 & echo $! > '{}'; echo started; wait",
            pid_file.display()
        );
        let error = tool_error(
            BashTool
                .execute(
                    json!({"command": command, "timeout": 1}),
                    create_tool_context(),
                )
                .await,
        );
        assert_eq!(
            error,
            "Tool execution failed: Command timed out after 1s and was stopped. \
             Output before the timeout:\nstarted\n"
        );
        let pid: libc::pid_t = std::fs::read_to_string(&pid_file)
            .unwrap()
            .trim()
            .parse()
            .unwrap();
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while process_running(pid) && std::time::Instant::now() < deadline {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        let survived = process_running(pid);
        if survived {
            // SAFETY: kills the child this test's command started.
            unsafe {
                libc::kill(pid, libc::SIGKILL);
            }
        }
        assert!(!survived, "the command's child {pid} outlived the timeout");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn bash_honours_capability_cwd_when_set() {
        let tmp = tempfile::TempDir::new().unwrap();
        let expected = tmp.path().canonicalize().unwrap();
        let mut ctx = create_tool_context();
        ctx.capabilities.cwd = Some(tmp.path().to_path_buf());
        let out = BashTool
            .execute(json!({"command": "pwd -P"}), ctx)
            .await
            .unwrap();
        assert_eq!(out.trim(), expected.to_string_lossy());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn bash_inherits_process_cwd_when_capability_cwd_is_none() {
        let expected = std::env::current_dir().unwrap().canonicalize().unwrap();
        let ctx = create_tool_context();
        assert!(ctx.capabilities.cwd.is_none());
        let out = BashTool
            .execute(json!({"command": "pwd -P"}), ctx)
            .await
            .unwrap();
        assert_eq!(out.trim(), expected.to_string_lossy());
    }

    #[test]
    fn references_env_name_is_word_bounded() {
        assert!(references_env_name("echo ${GH_TOKEN:?}", "GH_TOKEN"));
        assert!(references_env_name("echo $GH_TOKEN", "GH_TOKEN"));
        assert!(references_env_name("GH_TOKEN=x gh auth", "GH_TOKEN"));
        assert!(references_env_name("$env:GH_TOKEN", "GH_TOKEN"));
        assert!(!references_env_name("echo $GH_TOKEN_FILE", "GH_TOKEN"));
        assert!(!references_env_name("echo $MY_GH_TOKEN", "GH_TOKEN"));
        assert!(!references_env_name("gh auth status", "GH_TOKEN"));
    }

    /// T5 notice: non-zero exit + script references stripped var → notice.
    #[cfg(unix)]
    #[tokio::test]
    async fn bash_notice_on_stripped_secret_reference() {
        let warned = std::sync::Arc::new(std::sync::Mutex::new(
            std::collections::HashSet::<String>::new(),
        ));
        // --- case 1: non-zero exit + reference → notice
        let mut ctx = create_tool_context();
        ctx.capabilities.env_stripped = vec!["GH_TOKEN".into()];
        ctx.capabilities.env_warned = warned.clone();
        let err = BashTool
            .execute(json!({"command": "echo ${GH_TOKEN:?} 2>&1; exit 1"}), ctx)
            .await
            .unwrap_err();
        let msg = format!("{err}");
        assert!(
            msg.contains("note: $GH_TOKEN was stripped"),
            "notice must name the var: {msg}"
        );

        // --- case 2: zero exit → no notice
        let mut ctx2 = create_tool_context();
        ctx2.capabilities.env_stripped = vec!["GH_TOKEN".into()];
        ctx2.capabilities.env_warned = warned.clone();
        let out = BashTool
            .execute(json!({"command": "echo $GH_TOKEN; exit 0"}), ctx2)
            .await
            .unwrap();
        assert!(
            !out.contains("note:"),
            "zero exit must not produce a notice: {out}"
        );

        // --- case 3: second failure → no duplicate (same Arc)
        let mut ctx3 = create_tool_context();
        ctx3.capabilities.env_stripped = vec!["GH_TOKEN".into()];
        ctx3.capabilities.env_warned = warned.clone();
        let err2 = BashTool
            .execute(json!({"command": "echo ${GH_TOKEN}; exit 1"}), ctx3)
            .await
            .unwrap_err();
        let msg2 = format!("{err2}");
        assert!(
            !msg2.contains("note: $GH_TOKEN was stripped"),
            "dedup must suppress the second notice: {msg2}"
        );
    }
}

/// Issue #286: a multi-byte character split across two reads must come back
/// whole, not as U+FFFD.
#[cfg(test)]
mod utf8_chunk_tests {
    use super::*;
    use crate::tools::Tool;
    use serde_json::json;
    use std::collections::VecDeque;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    /// Yields exactly the given chunks, one per `read`, then EOF.
    struct ChunkedReader(VecDeque<Vec<u8>>);

    impl tokio::io::AsyncRead for ChunkedReader {
        fn poll_read(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut tokio::io::ReadBuf<'_>,
        ) -> Poll<std::io::Result<()>> {
            if let Some(chunk) = self.0.pop_front() {
                assert!(chunk.len() <= buf.remaining(), "test chunk larger than read buffer");
                buf.put_slice(&chunk);
            }
            Poll::Ready(Ok(()))
        }
    }

    async fn forward_chunks(chunks: Vec<Vec<u8>>, is_stderr: bool) -> String {
        let (tx, mut rx) = tokio::sync::mpsc::channel(1024);
        forward_output(ChunkedReader(chunks.into()), is_stderr, tx).await;
        let mut out = String::new();
        while let Some((stderr, chunk)) = rx.recv().await {
            let msg = chunk.consume();
            assert_eq!(stderr, is_stderr);
            out.push_str(&msg);
        }
        out
    }

    #[tokio::test]
    async fn characters_split_at_every_offset_survive_both_streams() {
        for ch in ['é', '▀', '😀'] {
            let text = format!("a{ch}b{ch}{ch}c");
            let bytes = text.as_bytes();
            for split in 1..bytes.len() {
                for is_stderr in [false, true] {
                    let out = forward_chunks(
                        vec![bytes[..split].to_vec(), bytes[split..].to_vec()],
                        is_stderr,
                    )
                    .await;
                    assert!(!out.contains('\u{FFFD}'), "{ch:?} split at {split}: {out:?}");
                    assert_eq!(out.as_bytes(), bytes, "{ch:?} split at {split}");
                }
            }
        }
    }

    #[tokio::test]
    async fn one_byte_reads_reassemble_every_character() {
        let text = "é▀😀 mixed ascii ✓ 日本語";
        let chunks = text.bytes().map(|b| vec![b]).collect();
        assert_eq!(forward_chunks(chunks, false).await, text);
    }

    #[tokio::test]
    async fn genuinely_invalid_bytes_still_become_replacement_characters() {
        assert_eq!(forward_chunks(vec![b"ok\xffok".to_vec()], false).await, "ok\u{FFFD}ok");
        // A lead byte followed by a non-continuation byte in the next read.
        assert_eq!(
            forward_chunks(vec![b"a\xe2".to_vec(), b"Xb".to_vec()], false).await,
            "a\u{FFFD}Xb"
        );
        // A stray continuation byte at the start of a read.
        assert_eq!(
            forward_chunks(vec![b"a".to_vec(), b"\x80b".to_vec()], false).await,
            "a\u{FFFD}b"
        );
    }

    #[tokio::test]
    async fn incomplete_sequence_at_eof_is_flushed_not_dropped() {
        assert_eq!(forward_chunks(vec![b"ab\xe2\x96".to_vec()], false).await, "ab\u{FFFD}");
        assert_eq!(
            forward_chunks(vec![b"ab".to_vec(), b"\xf0\x9f".to_vec()], true).await,
            "ab\u{FFFD}"
        );
    }

    /// A consumer that gives up (timeout, cancelled password prompt) drops
    /// the receiver with chunks still queued. Those bytes must be counted as
    /// dropped, or the process-wide accounting never settles again.
    #[tokio::test]
    async fn chunks_still_queued_when_the_receiver_goes_away_count_as_dropped() {
        let (tx, rx) = tokio::sync::mpsc::channel(16);
        let chunks: Vec<Vec<u8>> = (0..8).map(|_| b"queued".to_vec()).collect();
        forward_output(ChunkedReader(chunks.into()), false, tx).await;
        let before = bash_intermediary_snapshot();
        drop(rx);
        let after = wait_handoff_settled().await;
        assert!(after.dropped_bytes - before.dropped_bytes >= 8 * 6);
    }

    /// A password prompt has no trailing newline; it must be forwarded as soon
    /// as it is read, not held back waiting for more output or EOF.
    #[tokio::test]
    async fn prompt_without_newline_is_forwarded_before_eof() {
        let (mut writer, reader) = tokio::io::duplex(4096);
        let (tx, mut rx) = tokio::sync::mpsc::channel(16);
        let task = tokio::spawn(forward_output(reader, true, tx));
        use tokio::io::AsyncWriteExt;
        writer.write_all("[sudo] password for x: ".as_bytes()).await.unwrap();
        let (is_stderr, chunk) =
            tokio::time::timeout(std::time::Duration::from_secs(5), rx.recv())
                .await
                .expect("prompt must arrive without EOF")
                .unwrap();
        let msg = chunk.consume();
        assert!(is_stderr);
        assert_eq!(msg, "[sudo] password for x: ");
        drop(writer);
        task.await.unwrap();
    }

    /// Through the real tool: more than 1 KiB of 3-byte characters on both
    /// streams. 1024 is not a multiple of 3, so the reads split characters.
    #[tokio::test]
    async fn bash_tool_output_over_1kib_keeps_multibyte_characters_whole() {
        let channel = crate::tools::output::delta_channel();
        let (delta_tx, mut delta_rx) = (channel.sender, channel.receiver);
        let mut ctx = super::super::test_helpers::create_tool_context();
        ctx.channels.tx_delta = Some(delta_tx);
        let result = BashTool
            .execute(json!({"command": "printf '▀%.0s' {1..2000}", "timeout": 30}), ctx)
            .await
            .unwrap();
        let mut streamed = String::new();
        while let Some(delta) = delta_rx.try_drain() {
            streamed.push_str(&delta);
        }
        let expected = "▀".repeat(2000);
        assert!(!result.contains('\u{FFFD}'), "result has U+FFFD");
        assert_eq!(result, expected);
        assert!(!streamed.contains('\u{FFFD}'), "deltas have U+FFFD");
        assert!(expected.starts_with(&streamed) && !streamed.is_empty());

        let ctx = super::super::test_helpers::create_tool_context();
        let err = BashTool
            .execute(
                json!({"command": "python3 -c \"import sys; sys.stderr.write(('▀'*119+'\\n')*45)\"; exit 3", "timeout": 30}),
                ctx,
            )
            .await
            .unwrap_err()
            .to_string();
        assert!(!err.contains('\u{FFFD}'), "stderr has U+FFFD: {err}");
        assert!(err.contains(&("▀".repeat(119) + "\n").repeat(45)), "{err}");
    }
}