lean_ctx_server/server/

execute.rs

// SPDX-License-Identifier: Apache-2.0
use fs2::FileExt;
use std::io::Read;
use std::process::Stdio;
use std::sync::{Arc, Mutex, atomic::AtomicBool, mpsc};
use std::time::{Duration, Instant};

const READER_RESULT_TIMEOUT: Duration = Duration::from_secs(2);

/// Prefix of the timeout notice this module appends to a killed command.
const TIMEOUT_MARKER: &str = "ERROR: command timed out after ";

fn is_build_or_test_command(command: &str) -> bool {
    let tokens: Vec<_> = command
        .split(|ch: char| ch.is_ascii_whitespace() || matches!(ch, ';' | '&' | '|'))
        .filter(|token| !token.is_empty())
        .collect();
    tokens.windows(2).any(|pair| {
        let tool = std::path::Path::new(pair[0])
            .file_name()
            .and_then(|name| name.to_str())
            .unwrap_or(pair[0]);
        (tool == "cargo"
            && matches!(
                pair[1],
                "build" | "check" | "clippy" | "test" | "run" | "bench" | "install" | "fast"
            ))
            || (matches!(tool, "npm" | "pnpm" | "yarn" | "bun")
                && matches!(pair[1], "test" | "build" | "check"))
            || (tool == "lean-ctx" && pair[1] == "dev-install")
            || (matches!(tool, "make" | "gmake")
                && matches!(
                    pair[1],
                    "build"
                        | "check"
                        | "dev"
                        | "install"
                        | "install-release"
                        | "test"
                        | "preflight"
                        | "preflight-fast"
                ))
    }) || tokens.windows(3).any(|triple| {
        matches!(triple[0], "npm" | "pnpm" | "yarn" | "bun")
            && triple[1] == "run"
            && matches!(triple[2], "test" | "build" | "check" | "lint" | "typecheck")
    }) || tokens.iter().any(|token| {
        std::path::Path::new(token)
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| matches!(name, "rustc" | "dev-install.sh" | "install.sh"))
    })
}

fn acquire_build_lease(
    command: &str,
    cancel: Option<&AtomicBool>,
) -> Result<Option<std::fs::File>, String> {
    let config = lean_ctx::core::config::Config::load();
    if !config.agents.serialize_build_commands || !is_build_or_test_command(command) {
        return Ok(None);
    }
    let dir = lean_ctx::core::data_dir::lean_ctx_data_dir()?.join("resources");
    std::fs::create_dir_all(&dir).map_err(|error| error.to_string())?;
    let path = dir.join("build.lock");
    let file = std::fs::OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(&path)
        .map_err(|error| format!("open build lease {}: {error}", path.display()))?;
    loop {
        match file.try_lock_exclusive() {
            Ok(()) => return Ok(Some(file)),
            Err(error) if lock_is_contended(&error) => {
                if cancel.is_some_and(|flag| flag.load(std::sync::atomic::Ordering::Acquire)) {
                    return Err(
                        "build cancelled while waiting for the machine-wide slot".to_string()
                    );
                }
                std::thread::sleep(Duration::from_millis(100));
            }
            Err(error) => {
                return Err(format!("acquire build lease {}: {error}", path.display()));
            }
        }
    }
}

fn lock_is_contended(error: &std::io::Error) -> bool {
    error.kind() == std::io::ErrorKind::WouldBlock
        || cfg!(windows) && matches!(error.raw_os_error(), Some(32 | 33))
}

/// The child's own output preceding the timeout marker, or `None` when the
/// output carries no marker. `Some("")` means the command timed out having
/// produced nothing recoverable.
///
/// Single source of truth for "was anything captured before the timeout?":
/// `call_tool` decides `isError` with it (#1086) and `ctx_shell` decides
/// whether the notice is worth archiving (#995). It lives beside the code that
/// writes the marker because both callers previously re-derived it by matching
/// the whole output, so enriching the notice broke them silently (#1173).
pub(crate) fn output_before_timeout_marker(output: &str) -> Option<&str> {
    output.find(TIMEOUT_MARKER).map(|idx| output[..idx].trim())
}

#[cfg(test)]
pub(crate) fn execute_command_in(command: &str, cwd: &str) -> (String, i32) {
    execute_command_with_env(command, cwd, &std::collections::HashMap::new(), None)
}

#[cfg(test)]
pub(crate) fn execute_command_with_env(
    command: &str,
    cwd: &str,
    extra_env: &std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
) -> (String, i32) {
    execute_command_with_env_cancellable(command, cwd, extra_env, timeout_ms, None, false, None)
}

/// Execute a command under the normal shell policy, with an optional cooperative
/// cancellation signal used by explicit background jobs. The signal is checked
/// by the same watchdog that enforces `timeout_ms`, so cancellation kills the
/// complete process tree — the Unix session or the Windows job — rather than
/// only the shell leader.
///
/// `idle_keyed` turns `timeout_ms` into an *idle* budget instead of a wall-clock
/// one: the clock resets whenever the child produces new bytes (#1113/#1173). A
/// poll loop emitting a few hundred bytes over ten minutes is not the runaway
/// this watchdog protects against.
///
/// Every job managed by `background_shell` sets it, foreground runs included —
/// those detach at the soft cap and become exactly such a job. Only the
/// degraded no-session path runs strict wall-clock. Two bounds keep it honest:
/// genuinely runaway output hits the byte cap, which freezes the captured
/// length and so restarts the clock, and `LEAN_CTX_SHELL_BG_MAX_MS` (default
/// 1h) caps lifetime absolutely.
pub(crate) struct ShellExecution {
    pub output: String,
    pub exit_code: i32,
    pub completeness: lean_ctx::core::context_admission::OutputCompleteness,
    /// Q13: what the caller has to be told about the call's processes — a
    /// fixed, content-free line (`shell::containment`). It is NOT part of
    /// `output`: whoever delivers the result puts it in front of the text
    /// with [`with_notice`], after everything that admits, compresses or
    /// shortens that text, so none of it can drop or alter the line.
    pub notice: Option<&'static str>,
}

/// Puts a call's notice, if it has one, in front of the text delivered for
/// it, as a line of its own. The last step of a delivery, applied once.
pub(crate) fn with_notice(notice: Option<&'static str>, text: String) -> String {
    match notice {
        Some(notice) => format!("{notice}\n{text}"),
        None => text,
    }
}

/// Takes a notice off the front of a result that arrived as plain text (the
/// protected launcher's supervisor runs the shell and answers with text
/// only), so that it travels on as the typed field.
#[cfg(unix)]
fn split_notice(text: String) -> (Option<&'static str>, String) {
    match leading_notice(&text) {
        Some(notice) => {
            let rest = text.get(notice.len() + 1..).unwrap_or_default().to_string();
            (Some(notice), rest)
        }
        None => (None, text),
    }
}

/// The notice a delivered text starts with, as a line of its own.
#[cfg(unix)]
pub(crate) fn leading_notice(text: &str) -> Option<&'static str> {
    use lean_ctx::shell::containment::{END_NOTICE, HELPER_NOTICE, RELEASE_NOTICE, UNCLEAR_NOTICE};
    [HELPER_NOTICE, END_NOTICE, UNCLEAR_NOTICE, RELEASE_NOTICE]
        .into_iter()
        .find(|notice| {
            text.strip_prefix(notice)
                .is_some_and(|rest| rest.is_empty() || rest.starts_with('\n'))
        })
}

/// No notice exists without process containment.
#[cfg(not(unix))]
pub(crate) fn leading_notice(_text: &str) -> Option<&'static str> {
    None
}

/// Keeps a call's notice in front of a text that replaced the one it stood
/// on: the digest of a result too large to deliver inline says nothing about
/// the call's processes by itself.
pub(crate) fn keep_notice(notice: Option<&'static str>, text: String) -> String {
    if leading_notice(&text).is_some() {
        text
    } else {
        with_notice(notice, text)
    }
}

impl ShellExecution {
    fn complete(output: String, exit_code: i32) -> Self {
        Self {
            output,
            exit_code,
            completeness: lean_ctx::core::context_admission::OutputCompleteness::complete(),
            notice: None,
        }
    }

    /// A brokered run: the supervisor captured it, so this process cannot
    /// vouch for the readers. Brokered execution exists on Unix only.
    #[cfg(unix)]
    fn unknown_capture(output: String, exit_code: i32) -> Self {
        Self {
            output,
            exit_code,
            completeness: lean_ctx::core::context_admission::OutputCompleteness {
                reader_incomplete: true,
                ..lean_ctx::core::context_admission::OutputCompleteness::complete()
            },
            notice: None,
        }
    }
}

/// Hand this executor to the protected supervisor
/// (`core::protected_execution`), which runs brokered `ctx_shell` commands
/// with it. The CLI calls this at start-up; the first registration wins.
#[cfg(unix)]
pub fn register_protected_shell_runner() {
    lean_ctx::core::protected_execution::register_shell_runner(
        execute_command_with_env_cancellable,
    );
}

// Production caller: core::protected_execution (via
// `register_protected_shell_runner`), which is unix-only.
#[cfg(any(unix, test))]
pub(crate) fn execute_command_with_env_cancellable(
    command: &str,
    cwd: &str,
    extra_env: &std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    cancel: Option<&AtomicBool>,
    idle_keyed: bool,
    live: Option<&std::sync::Mutex<String>>,
) -> (String, i32) {
    // Kept as a compatibility wrapper. Raw progressive output is no longer
    // written to caller-owned buffers.
    let _ = live;
    let execution = execute_command_with_env_cancellable_captured(
        command, cwd, extra_env, timeout_ms, cancel, idle_keyed,
    );
    // Nothing shortens the text on this path: the notice goes in front here.
    (
        with_notice(execution.notice, execution.output),
        execution.exit_code,
    )
}

pub(crate) fn execute_command_with_env_captured(
    command: &str,
    cwd: &str,
    extra_env: &std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
) -> ShellExecution {
    execute_command_with_env_cancellable_captured(command, cwd, extra_env, timeout_ms, None, false)
}

pub(crate) fn execute_command_with_env_cancellable_captured(
    command: &str,
    cwd: &str,
    extra_env: &std::collections::HashMap<String, String>,
    timeout_ms: Option<u64>,
    cancel: Option<&AtomicBool>,
    idle_keyed: bool,
) -> ShellExecution {
    // Store isolation for model-directed subprocesses of a protected session.
    //
    // macOS refuses a nested `sandbox_apply` under the whole-MCP profile, so
    // this process cannot wrap its own children in the launcher's rules. The
    // request goes to the launcher's supervisor instead, which starts the child
    // from outside that sandbox under the profile the launcher pinned. Decided
    // *before* the build lease and before any caller-supplied `extra_env`: the
    // supervisor takes the machine-wide lease itself (taking it on both sides
    // would deadlock), and a tool call can neither select nor drop the rules. A
    // protected session that cannot reach its supervisor runs nothing.
    //
    // This is the one exec seam `ctx_shell` has — foreground, background and the
    // interpreter heredoc reroute all arrive here — and the Engine's own
    // persistence never passes through it.
    #[cfg(unix)]
    let protected_prefix: Option<Vec<String>> = {
        use lean_ctx::core::protected_execution::{Mode, launch_mode, sandbox_prefix};
        match launch_mode() {
            Err(error) => return ShellExecution::complete(format!("ERROR: {error}"), 1),
            Ok(Mode::Brokered(client)) => {
                let (output, exit_code) = client.shell(
                    command, cwd, extra_env, timeout_ms, cancel, idle_keyed, None,
                );
                let (notice, output) = split_notice(output);
                return ShellExecution {
                    notice,
                    ..ShellExecution::unknown_capture(output, exit_code)
                };
            }
            Ok(Mode::Sandboxed(profile)) => Some(sandbox_prefix(&profile)),
            Ok(Mode::Direct) => None,
        }
    };
    #[cfg(not(unix))]
    let protected_prefix = match lean_ctx::core::protected_store::protected_child_prefix() {
        Ok(prefix) => prefix,
        Err(error) => return ShellExecution::complete(format!("ERROR: {error}"), 1),
    };

    let _build_lease = match acquire_build_lease(command, cancel) {
        Ok(lease) => lease,
        Err(error) => return ShellExecution::complete(format!("ERROR: {error}"), 130),
    };
    let resource_config = lean_ctx::core::config::Config::load().agents;
    let (shell, flag) = lean_ctx::shell::shell_and_flag();
    let normalized_cmd = crate::tools::ctx_shell::normalize_command_for_shell(command);
    let normalized_cmd = lean_ctx::shell::platform::zsh_safe_command(&normalized_cmd, &shell);
    let dir = std::path::Path::new(cwd);
    let mut cmd = std::process::Command::new(&shell);
    if let Some([program, arguments @ ..]) = protected_prefix.as_deref() {
        cmd = std::process::Command::new(program);
        cmd.args(arguments).arg(&shell);
    }
    if cfg!(windows) && lean_ctx::shell::platform::is_powershell(&shell) {
        cmd.args(["-NoProfile", "-ExecutionPolicy", "Bypass"]);
    }
    cmd.arg(&flag)
        .arg(&normalized_cmd)
        .env("GIT_TERMINAL_PROMPT", "0")
        .stdin(Stdio::null());
    lean_ctx::shell::reentry::mark_child(&mut cmd);

    if !extra_env.contains_key("GIT_PAGER") {
        cmd.env("GIT_PAGER", "cat");
    }
    if !extra_env.contains_key("PAGER") {
        cmd.env("PAGER", "cat");
    }

    ensure_utf8_locale(&mut cmd, extra_env);
    lean_ctx::shell::platform::apply_profile_free_env(&mut cmd);

    // Auto-forward agent runtime env vars (CODEX_THREAD_ID, CLAUDE_*, …) so
    // session-aware commands run through ctx_shell can see the active session.
    //   1. From this process's own env — covers agents that pass the vars to the
    //      MCP server process.
    //   2. From the captured agent-env store — covers agents like Codex where the
    //      vars live only in the native agent shell, not the MCP server process
    //      (#370). Hooks / `lean-ctx -c` capture them; the process env wins on
    //      conflict, and explicit `extra_env` (below) wins over both.
    let mut forwarded: std::collections::HashSet<String> = std::collections::HashSet::new();
    for (key, val) in std::env::vars() {
        if lean_ctx::core::agent_runtime_env::is_forwardable(&key) {
            cmd.env(&key, &val);
            forwarded.insert(key);
        }
    }
    for (key, val) in lean_ctx::core::agent_runtime_env::load() {
        if !forwarded.contains(&key) {
            cmd.env(&key, &val);
        }
    }

    // Explicit env vars from tool call (highest priority)
    for (key, val) in extra_env {
        cmd.env(key, val);
    }
    if is_build_or_test_command(command) {
        if resource_config.cargo_build_jobs > 0 && !extra_env.contains_key("CARGO_BUILD_JOBS") {
            cmd.env(
                "CARGO_BUILD_JOBS",
                resource_config.cargo_build_jobs.to_string(),
            );
        }
        if resource_config.shared_cargo_target
            && !extra_env.contains_key("CARGO_TARGET_DIR")
            && let Ok(data_dir) = lean_ctx::core::data_dir::lean_ctx_data_dir()
        {
            cmd.env(
                "CARGO_TARGET_DIR",
                data_dir.join("build-cache/cargo-target"),
            );
        }
    }
    if dir.is_dir() {
        cmd.current_dir(dir);
    } else {
        return ShellExecution::complete(
            format!("ERROR: working directory does not exist or is not a directory: {cwd}"),
            1,
        );
    }
    let cap = lean_ctx::core::limits::max_shell_bytes();

    // Isolate the shell in its own process group on Unix. A timeout must kill
    // descendants too; otherwise a child can retain a pipe write-end and make
    // the caller report an empty result despite bytes already captured (#995).
    //
    // A new *session* rather than just a group: it also drops the controlling
    // terminal. Otherwise an interactive descendant (`zsh -i`, `bash -i`, a
    // test suite that spawns one) takes the terminal's foreground with
    // tcsetpgrp, and the agent that launched the MCP server is stopped by
    // SIGTTIN — "zsh: suspended claude", with mouse reporting still on.
    // The session leader's pid is its pgid, so the timeout kill is unchanged.
    //
    // Q13: on Unix the shell's parent is this process's spawn helper
    // (`shell::containment`), which starts it with that same `setsid` and
    // ends its session should this process die during the call. Everything
    // decided above — command line, environment, directory, sandbox prefix —
    // is handed over as it is. Without a helper the shell is started here, as
    // before, and the result says so in its first line.
    //
    // Windows has no process group to kill: contain the tree in a job object
    // instead, so a timeout or cancel ends the `python -` a heredoc started
    // rather than only its shell (#1920).
    lean_ctx::shell::process_tree::ProcessTree::prepare(&mut cmd);

    let started = match lean_ctx::shell::process_tree::ShellChild::start(&mut cmd) {
        Ok(started) => started,
        Err(e) => return ShellExecution::complete(format!("ERROR: {e}"), 1),
    };
    let mut child = started.child;
    // Stream each pipe into a shared, cap-bounded buffer that the main thread can
    // read at any time. Crucially this lets a timed-out wait recover the bytes
    // captured so far instead of discarding all output (#945).
    let (out_buf, out_done) = spawn_capture(started.stdout, cap);
    let (err_buf, err_done) = spawn_capture(started.stderr, cap);

    let timeout = command_timeout(command, timeout_ms);
    let start = Instant::now();
    let hard_deadline = idle_keyed.then(|| start + streaming_max_lifetime());
    let mut last_len = 0usize;
    let mut last_output_at = start;
    // Named on timeout so a multi-segment pipeline says *which* part hung (#1086).
    let mut still_running: Vec<String> = Vec::new();
    let (code, timed_out, cancelled) = loop {
        match child.try_wait() {
            Ok(Some(status)) => {
                break (
                    lean_ctx::shell::exit_status::exit_code(status),
                    false,
                    false,
                );
            }
            Ok(None) => {
                if cancel.is_some_and(|signal| signal.load(std::sync::atomic::Ordering::Acquire)) {
                    child.kill();
                    child.wait();
                    break (130, false, true);
                }
                let idle_for = if idle_keyed {
                    let len = captured_len(&out_buf) + captured_len(&err_buf);
                    if len != last_len {
                        last_len = len;
                        last_output_at = Instant::now();
                    }
                    last_output_at.elapsed()
                } else {
                    start.elapsed()
                };
                if idle_for >= timeout || hard_deadline.is_some_and(|d| Instant::now() >= d) {
                    still_running = running_segments(child.id(), &normalized_cmd);
                    child.kill();
                    child.wait();
                    break (124, true, false);
                }
                std::thread::sleep(Duration::from_millis(25));
            }
            Err(_) => {
                child.kill();
                break (1, false, false);
            }
        }
    };
    // Q13: a shell whose helper went away before its exit was known has no
    // parent that answers for it; the caller is told with the fixed notice.
    // Such a call's outcome is not known in full — whether the command ran
    // at all, or how it ended — so its capture counts as incomplete.
    let lost_notice = child.lost_notice();
    // A command that exited on its own keeps what it deliberately left running,
    // as the Unix session does; a killed tree has nothing left to release. A
    // release that could not be handed over is said to the caller as well.
    let release_notice = child.release();
    let containment_notice = started.notice.or(lost_notice).or(release_notice);

    // Bounded grace period for the readers to reach EOF, but always read the
    // shared buffers afterwards so output is never silently lost (#945). A reader
    // that misses the deadline means something still holds the pipe open; we then
    // surface the partial capture plus an explicit note rather than nothing.
    let reader_deadline = Instant::now() + READER_RESULT_TIMEOUT;
    let out_complete = wait_for_reader(&out_done, reader_deadline);
    let err_complete = wait_for_reader(&err_done, reader_deadline);
    let (out_bytes, out_trunc, out_read_failed) = snapshot(&out_buf);
    let (err_bytes, err_trunc, err_read_failed) = snapshot(&err_buf);
    let reader_incomplete = !out_complete || !err_complete || out_read_failed || err_read_failed;
    // The legacy decoder may replace invalid UTF-8 bytes. On Windows this also
    // conservatively flags ACP decoding, whose fallback does not prove lossless.
    let decode_lossy =
        std::str::from_utf8(&out_bytes).is_err() || std::str::from_utf8(&err_bytes).is_err();

    let stdout =
        lean_ctx::shell::resolve_carriage_returns(&lean_ctx::shell::decode_output(&out_bytes));
    let stderr =
        lean_ctx::shell::resolve_carriage_returns(&lean_ctx::shell::decode_output(&err_bytes));
    // On failure both streams are labeled so the agent can attribute the error
    // (#812); success keeps the plain join.
    let mut text = lean_ctx::shell::combine_streams(&stdout, &stderr, code);

    if out_trunc || err_trunc {
        text.push_str(&format!(
            "\n[truncated: cap={}B stdout={}B stderr={}B]",
            cap,
            out_bytes.len(),
            err_bytes.len()
        ));
    }
    // The command finished but a reader never hit EOF: a leftover process is
    // holding the pipe open. We still return what was captured; flag it so the
    // agent knows the tail may be missing instead of seeing a bare exit code
    // (#945). Suppressed on timeout, which carries its own message below.
    if reader_incomplete && !timed_out {
        if !text.is_empty() && !text.ends_with('\n') {
            text.push('\n');
        }
        text.push_str(&format!(
            "[lean-ctx: output reader still draining after {}s — a background process is likely \
             holding the pipe open; output above may be partial]",
            READER_RESULT_TIMEOUT.as_secs()
        ));
    }
    if timed_out {
        if !text.ends_with('\n') && !text.is_empty() {
            text.push('\n');
        }
        text.push_str(&format!(
            "{TIMEOUT_MARKER}{}ms{}",
            timeout.as_millis(),
            if idle_keyed {
                " without new output"
            } else {
                ""
            }
        ));
        // #1086: for a compound command the captured output is often complete
        // for every segment but one. Name what is still alive in the child's
        // process group so the caller can fix that part instead of re-running
        // the whole pipeline.
        //
        // Joined with "; " rather than " | " (#1654): these are processes, not
        // stages, and a pipe character asserted a relationship the process
        // table never established — rendering a line that was not runnable and
        // named a stage the caller never wrote.
        if !still_running.is_empty() {
            text.push_str(&format!(
                "\n[still running at timeout: {}]",
                still_running.join("; ")
            ));
        }
    }
    // #1655/#1662: a recursive walk does not read `.gitignore`, so a nested
    // checkout the other ctx_* tools never see can dominate it. Say so after
    // the fact — the command's own traversal is left alone.
    //
    // Not only on timeout (#1662): `grep -r … | head` is the shape agents
    // actually type, and it is the shape where the cost hides best. `head`
    // closes early, the ten visible lines arrive at once, and the walk keeps
    // running for two more minutes behind a result that already looked
    // finished. Reported as "no redirect, no hint" — because the only place
    // this was emitted was a timeout that a fast-looking command never
    // reached.
    if let Some(hint) = super::walk_hint::walk_hint(command, cwd) {
        if !text.is_empty() && !text.ends_with('\n') {
            text.push('\n');
        }
        text.push_str(&hint);
    }
    if cancelled {
        if !text.ends_with('\n') && !text.is_empty() {
            text.push('\n');
        }
        // #1246: a cancel is always caller-requested, so it is not an error.
        text.push_str("[cancelled: command stopped on request]");
    }

    ShellExecution {
        output: text,
        exit_code: code,
        // Q13: fixed and content-free; carried beside the text, not in it.
        notice: containment_notice,
        completeness: lean_ctx::core::context_admission::OutputCompleteness {
            truncated: out_trunc || err_trunc,
            timed_out,
            reader_incomplete: reader_incomplete || lost_notice.is_some(),
            cancelled,
            decode_lossy,
        },
    }
}

/// Shared, cap-bounded capture buffer for one child pipe. The reader thread
/// appends here as bytes arrive — not only at EOF — so the caller can recover
/// whatever was read so far even if the reader never reaches EOF, e.g. when a
/// backgrounded or otherwise-inherited process keeps the pipe's write-end open
/// after the direct child exits (#945).
#[derive(Default)]
struct CaptureBuf {
    bytes: Vec<u8>,
    truncated: bool,
    read_failed: bool,
}

/// Spawn a reader that streams `pipe` into a shared [`CaptureBuf`] (bounded to
/// `cap` bytes) and signals completion (EOF or read error) over the returned
/// channel. The buffer is readable at any time via [`snapshot`], so a timed-out
/// wait yields partial output instead of nothing.
fn spawn_capture<R: Read + Send + 'static>(
    pipe: Option<R>,
    cap: usize,
) -> (Arc<Mutex<CaptureBuf>>, mpsc::Receiver<()>) {
    let shared = Arc::new(Mutex::new(CaptureBuf::default()));
    let (done_tx, done_rx) = mpsc::channel();
    let writer = Arc::clone(&shared);
    std::thread::spawn(move || {
        if let Some(mut r) = pipe {
            let mut buf = [0u8; 8192];
            loop {
                match r.read(&mut buf) {
                    Ok(0) => break,
                    Err(ref error) if error.kind() == std::io::ErrorKind::Interrupted => {}
                    Err(_) => {
                        writer
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .read_failed = true;
                        break;
                    }
                    Ok(n) => {
                        let mut s = writer
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner);
                        if s.bytes.len() < cap {
                            let remaining = cap - s.bytes.len();
                            let take = remaining.min(n);
                            s.bytes.extend_from_slice(&buf[..take]);
                            if take < n {
                                s.truncated = true;
                            }
                        } else {
                            s.truncated = true;
                        }
                    }
                }
            }
        }
        let _ = done_tx.send(());
    });
    (shared, done_rx)
}

/// Snapshot a capture buffer (clone bytes + truncation flag) under its lock.
/// Safe to call while the reader is still writing — yields a consistent prefix.
fn snapshot(buf: &Arc<Mutex<CaptureBuf>>) -> (Vec<u8>, bool, bool) {
    let s = buf
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    (s.bytes.clone(), s.truncated, s.read_failed)
}

/// Bytes captured so far, used as the liveness signal for an idle-keyed
/// timeout. Cheaper than [`snapshot`], which clones the whole buffer.
fn captured_len(buf: &Arc<Mutex<CaptureBuf>>) -> usize {
    buf.lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .bytes
        .len()
}

/// Absolute ceiling on an idle-keyed (background) job, so a command that keeps
/// emitting output forever still cannot outlive the daemon's patience (#1173).
fn streaming_max_lifetime() -> Duration {
    Duration::from_millis(
        std::env::var("LEAN_CTX_SHELL_BG_MAX_MS")
            .ok()
            .and_then(|v| v.parse().ok())
            .filter(|&ms: &u64| ms > 0)
            .unwrap_or(3_600_000),
    )
}

/// One live process in the timed-out child's group.
#[cfg(unix)]
struct LiveProc {
    pid: u32,
    ppid: u32,
    args: String,
}

/// Command lines still alive in the timed-out child's process group (#1086).
///
/// These are **processes**, not pipeline segments. The distinction matters
/// because a segment can spawn its own children into the same group: `go run x`
/// execs a compiled binary under `$TMPDIR`, which then shows up here alongside
/// its parent. Rendering the set as `a | b` claimed a pipeline the process
/// table never established, and pointed the caller at a stage they never wrote
/// (#1654).
///
/// Rows still carrying the *whole* command are the un-exec'd shell wrapper and
/// say nothing specific, so they are dropped — note the leader is not simply
/// skipped by pid, because `sh -c 'a; b'` execs into its final segment and so
/// *is* the leader.
///
/// Best-effort: an unavailable or unparsable `ps` yields no attribution.
#[cfg(unix)]
fn running_segments(pgid: u32, command: &str) -> Vec<String> {
    // POSIX-portable field selection; `ps -g` differs between BSD and Linux.
    // `ppid` is read so a descendant can be labelled as one rather than
    // presented as a sibling stage.
    let Ok(out) = std::process::Command::new("ps")
        .args(["-A", "-o", "pid=,ppid=,pgid=,args="])
        .output()
    else {
        return Vec::new();
    };
    let needle = command.split_whitespace().collect::<Vec<_>>().join(" ");
    let live: Vec<LiveProc> = String::from_utf8_lossy(&out.stdout)
        .lines()
        .filter_map(|line| parse_ps_row(line, pgid, &needle))
        .collect();

    describe_live(&live)
}

/// Render the live set, marking any process whose parent is also in the set.
///
/// A descendant is shown under its parent rather than as a peer, so the caller
/// can tell "the stage I wrote" from "something that stage started".
#[cfg(unix)]
fn describe_live(live: &[LiveProc]) -> Vec<String> {
    let pids: std::collections::HashSet<u32> = live.iter().map(|p| p.pid).collect();
    live.iter()
        .filter(|p| !pids.contains(&p.ppid))
        .map(|parent| {
            let kids: Vec<&str> = live
                .iter()
                .filter(|c| c.ppid == parent.pid)
                .map(|c| c.args.as_str())
                .collect();
            if kids.is_empty() {
                parent.args.clone()
            } else {
                format!("{} (spawned: {})", parent.args, kids.join(", "))
            }
        })
        .collect()
}

/// One `pid ppid pgid args` row: `Some(..)` iff it belongs to `pgid` and names
/// a narrower command than the one we launched. `needle` is the whole command
/// with runs of whitespace collapsed, matching how `ps` renders `args`.
#[cfg(unix)]
fn parse_ps_row(line: &str, pgid: u32, needle: &str) -> Option<LiveProc> {
    let rest = line.trim_start();
    let (pid, rest) = rest.split_once(char::is_whitespace)?;
    let rest = rest.trim_start();
    let (ppid, rest) = rest.split_once(char::is_whitespace)?;
    let args = rest
        .trim_start()
        .split_once(char::is_whitespace)
        .and_then(|(row_pgid, args)| {
            (row_pgid.parse::<u32>().ok()? == pgid).then_some(args.trim())
        })?;
    if args.is_empty() || args.contains(needle) {
        return None;
    }
    Some(LiveProc {
        pid: pid.parse().ok()?,
        ppid: ppid.parse().ok()?,
        args: args.chars().take(200).collect(),
    })
}

#[cfg(not(unix))]
fn running_segments(_pgid: u32, _command: &str) -> Vec<String> {
    // No process groups here, so a descendant cannot be attributed to this run.
    Vec::new()
}

/// Block until the reader signals completion or `deadline` passes. Returns true
/// iff the reader completed (reached EOF) within the deadline.
fn wait_for_reader(done: &mpsc::Receiver<()>, deadline: Instant) -> bool {
    let remaining = deadline.saturating_duration_since(Instant::now());
    done.recv_timeout(remaining).is_ok()
}

fn ensure_utf8_locale(
    cmd: &mut std::process::Command,
    extra_env: &std::collections::HashMap<String, String>,
) {
    if extra_env.contains_key("LC_ALL") || extra_env.contains_key("LC_CTYPE") {
        return;
    }
    lean_ctx::shell::platform::apply_utf8_locale(cmd);
}

fn command_timeout(command: &str, timeout_ms: Option<u64>) -> Duration {
    // Single source of truth: operator env pin > per-call `timeout_ms` >
    // per-tier env/config > built-in heavy/normal ceilings. Keeps this path
    // identical to the interactive hook (`shell::exec::shell_timeout`).
    lean_ctx::shell::shell_timeout_with_override(command, timeout_ms)
}

#[cfg(all(test, target_os = "macos"))]
#[path = "execute_supervised_tests.rs"]
mod supervised_tests;

#[cfg(all(test, unix))]
#[path = "execute_containment_tests.rs"]
mod containment_tests;

#[cfg(test)]
mod tests {
    use super::{
        command_timeout, ensure_utf8_locale, execute_command_in, is_build_or_test_command,
    };

    #[test]
    fn resource_broker_build_slot_classifier_is_precise() {
        for command in [
            "cargo test --lib",
            "cargo fast",
            "cd rust && cargo fast --features dev-tools",
            "make dev",
            "make build",
            "make preflight-fast",
            "make install-release",
            "./install.sh --build-only",
            "bash rust/dev-install.sh --fast",
            "./rust/dev-install.sh --release",
            "lean-ctx dev-install",
            "/opt/bin/lean-ctx dev-install --fast",
            "cd rust && cargo clippy --all-features",
            "/opt/bin/cargo build --release",
            "npm test",
            "npm run typecheck",
            "pnpm build",
            "rustc main.rs",
        ] {
            assert!(is_build_or_test_command(command), "{command}");
        }
        for command in [
            "git status",
            "cargo metadata",
            "make help",
            "lean-ctx --version",
            "npm view lean-ctx",
            "rg cargo",
        ] {
            assert!(!is_build_or_test_command(command), "{command}");
        }
    }

    #[test]
    fn resource_broker_serializes_build_leases() {
        let isolation = lean_ctx::core::data_dir::isolated_data_dir();
        // Off by default: a user's builds are never queued behind other sessions.
        assert!(
            super::acquire_build_lease("cargo test", None)
                .expect("lease")
                .is_none()
        );
        std::fs::write(
            isolation.path().join("config.toml"),
            "[agents]\nserialize_build_commands = true\n",
        )
        .expect("opt in");
        let first = super::acquire_build_lease("cargo test first", None)
            .expect("first lease")
            .expect("build command gets lease");
        let acquired = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
        let worker_acquired = std::sync::Arc::clone(&acquired);
        let worker = std::thread::spawn(move || {
            let _second = super::acquire_build_lease("cargo test second", None)
                .expect("second lease")
                .expect("build command gets lease");
            worker_acquired.store(true, std::sync::atomic::Ordering::Release);
        });
        std::thread::sleep(std::time::Duration::from_millis(50));
        assert!(!acquired.load(std::sync::atomic::Ordering::Acquire));
        drop(first);
        worker.join().expect("lease waiter");
        assert!(acquired.load(std::sync::atomic::Ordering::Acquire));
    }

    #[test]
    fn command_timeout_delegates_to_shell_timeout() {
        // `command_timeout` is a thin alias for `shell::exec::shell_timeout`
        // (full precedence coverage lives there). Smoke-test the delegation:
        // heavy beats normal, and the universal MS override pins both.
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let saved = std::env::var("LEAN_CTX_SHELL_TIMEOUT_MS").ok();
        lean_ctx::test_env::remove_var("LEAN_CTX_SHELL_TIMEOUT_MS");

        assert!(
            command_timeout("cargo install --path .", None) > command_timeout("git status", None)
        );

        lean_ctx::test_env::set_var("LEAN_CTX_SHELL_TIMEOUT_MS", "5000");
        assert_eq!(
            command_timeout("cargo install --path .", None),
            std::time::Duration::from_secs(5)
        );
        assert_eq!(
            command_timeout("git status", None),
            std::time::Duration::from_secs(5)
        );

        lean_ctx::test_env::remove_var("LEAN_CTX_SHELL_TIMEOUT_MS");
        if let Some(v) = saved {
            lean_ctx::test_env::set_var("LEAN_CTX_SHELL_TIMEOUT_MS", v);
        }
    }

    /// The one exec seam: foreground runs, detached background jobs and the
    /// interpreter-heredoc reroute all build their command here, so covering it
    /// once covers all three. Synthetic store, isolated data dir — never the
    /// developer's installed knowledge.
    #[cfg(target_os = "macos")]
    #[test]
    fn the_supervisor_runs_the_shell_under_the_pinned_child_rules() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let isolation = lean_ctx::core::data_dir::isolated_data_dir();
        let fact = isolation.path().join("knowledge/knowledge.json");
        std::fs::create_dir_all(fact.parent().unwrap()).unwrap();
        std::fs::write(&fact, "synthetic-provider-fact").unwrap();
        let work = tempfile::tempdir().unwrap();
        let cwd = work.path().to_str().unwrap();
        let read_fact = format!("cat {}", fact.display());
        let write_fact = format!("printf downgraded > {}", fact.display());

        // Community: the store is an ordinary file, the shell is unwrapped.
        lean_ctx::core::protected_store::unpin_synthetic_session();
        let (text, code) = execute_command_in(&read_fact, cwd);
        assert_eq!(code, 0, "{text}");
        assert!(text.contains("synthetic-provider-fact"), "{text}");

        let _session = lean_ctx::core::protected_store::pin_synthetic_session(isolation.path())
            .expect("pinned rules");
        lean_ctx::core::protected_execution::with_test_supervisor(|| {
            let (text, code) = execute_command_in(&read_fact, cwd);
            assert_ne!(code, 0, "{text}");
            assert!(!text.contains("synthetic-provider-fact"), "{text}");
            let (text, code) = execute_command_in(&write_fact, cwd);
            assert_ne!(code, 0, "{text}");
            // Ordinary project work in the session's own directory is untouched.
            let (text, code) =
                execute_command_in("printf 'fn main() {}' > main.rs && cat main.rs", cwd);
            assert_eq!(code, 0, "{text}");
            assert!(text.contains("fn main()"), "{text}");
        });
        // The Engine keeps persisting from outside the child rules.
        assert_eq!(
            std::fs::read_to_string(&fact).unwrap(),
            "synthetic-provider-fact"
        );
        std::fs::write(&fact, "engine-write").unwrap();

        // A protected session whose rules went missing runs nothing at all —
        // there is no unconfined fallback shell.
        lean_ctx::test_env::remove_var(lean_ctx::core::protected_store::CHILD_PROFILE_ENV);
        let (text, code) = execute_command_in("printf ran-unconfined", cwd);
        assert_ne!(code, 0, "{text}");
        assert!(text.starts_with("ERROR:"), "{text}");
        assert!(!text.contains("ran-unconfined"), "{text}");
    }

    /// #1113/#1173: a poll loop that emits a short line every so often must
    /// outlive the wall-clock budget — the cap protects against runaway output,
    /// not against a long-lived monitor. Wall-clock mode still kills it.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn idle_keyed_timeout_survives_a_trickling_loop() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        lean_ctx::test_env::remove_var("LEAN_CTX_SHELL_TIMEOUT_MS");
        let env = std::collections::HashMap::new();
        // Emits every 100ms for ~900ms under a 400ms budget.
        let cmd = "for i in 1 2 3 4 5 6 7 8 9; do printf T; sleep 0.1; done; printf DONE";

        let (idle_out, idle_code) = super::execute_command_with_env_cancellable(
            cmd,
            ".",
            &env,
            Some(400),
            None,
            true,
            None,
        );
        assert_eq!(
            idle_code, 0,
            "output kept resetting the idle clock: {idle_out}"
        );
        assert!(idle_out.contains("DONE"));

        let (wall_out, wall_code) = super::execute_command_with_env_cancellable(
            cmd,
            ".",
            &env,
            Some(400),
            None,
            false,
            None,
        );
        assert_eq!(wall_code, 124, "wall-clock mode must still enforce the cap");
        assert!(wall_out.contains("timed out"));
    }

    /// #1086: a timed-out pipeline names the segment that was still running,
    /// so the caller fixes that part instead of re-running everything.
    #[test]
    #[cfg_attr(windows, ignore)]
    fn timeout_names_the_still_running_segment() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        lean_ctx::test_env::remove_var("LEAN_CTX_SHELL_TIMEOUT_MS");
        let env = std::collections::HashMap::new();
        let (out, code) = super::execute_command_with_env_cancellable(
            "printf FIRST_OK; sleep 47",
            ".",
            &env,
            Some(600),
            None,
            false,
            None,
        );
        assert_eq!(code, 124);
        assert!(
            out.contains("FIRST_OK"),
            "partial output must survive: {out}"
        );
        assert!(
            out.contains("still running at timeout") && out.contains("47"),
            "the hung segment must be named: {out}"
        );
    }

    #[test]
    #[cfg(unix)]
    fn ps_row_names_segments_and_drops_the_whole_command() {
        // Rows are `pid ppid pgid args`.
        let whole = "printf hi; grep -rn needle /repo";
        assert_eq!(
            super::parse_ps_row(" 4242 4200 4200 grep -rn needle /repo", 4200, whole)
                .map(|p| p.args),
            Some("grep -rn needle /repo".to_string()),
            "a segment must be named"
        );
        // The un-exec'd shell wrapper still carries the whole command.
        assert!(
            super::parse_ps_row(
                "4200 1 4200 /bin/sh -c printf hi; grep -rn needle /repo",
                4200,
                whole
            )
            .is_none()
        );
        // Another job's process group.
        assert!(super::parse_ps_row("4242 4200 9999 sleep 5", 4200, whole).is_none());
        assert!(super::parse_ps_row("garbage", 4200, whole).is_none());
    }

    // --- GH #1654: descendants are not pipeline stages ---

    /// The reporter's case. `go run ./x` execs the binary it just compiled,
    /// which joins the same process group. Rendering the set as `a | b` claimed
    /// a pipeline the process table never established, producing a line that
    /// is not runnable and naming a stage the caller never wrote.
    #[test]
    #[cfg(unix)]
    fn a_spawned_child_is_labelled_not_presented_as_a_stage() {
        let live = vec![
            super::LiveProc {
                pid: 100,
                ppid: 1,
                args: "go run ./ --template x".into(),
            },
            super::LiveProc {
                pid: 101,
                ppid: 100,
                args: "/var/folders/T/go-build/b001/exe/evcc --template x".into(),
            },
        ];
        let out = super::describe_live(&live);
        assert_eq!(out.len(), 1, "the child is not a peer entry: {out:?}");
        assert!(out[0].starts_with("go run ./"), "{out:?}");
        assert!(
            out[0].contains("(spawned: /var/folders/T/go-build/b001/exe/evcc"),
            "the descendant is named as one: {out:?}"
        );
    }

    /// Real pipeline siblings share a parent that is *not* in the set (the
    /// shell wrapper was dropped for carrying the whole command), so both stay
    /// top-level.
    #[test]
    #[cfg(unix)]
    fn genuine_pipeline_siblings_both_stay_top_level() {
        let live = vec![
            super::LiveProc {
                pid: 200,
                ppid: 50,
                args: "grep -rn needle /repo".into(),
            },
            super::LiveProc {
                pid: 201,
                ppid: 50,
                args: "head -30".into(),
            },
        ];
        let mut out = super::describe_live(&live);
        out.sort();
        assert_eq!(out, vec!["grep -rn needle /repo", "head -30"]);
    }

    /// The rendered separator must not be shell syntax — that was the whole
    /// defect. `;` reads as a list; `|` read as a pipeline that did not exist.
    #[test]
    #[cfg(unix)]
    fn the_timeout_line_does_not_imply_a_pipeline() {
        let joined = ["a".to_string(), "b".to_string()].join("; ");
        assert!(!joined.contains('|'), "no pipe may be invented: {joined}");
    }

    #[test]
    fn ensure_utf8_locale_sets_fallback_when_none_inherited() {
        let _env_lock = lean_ctx::core::data_dir::test_env_lock();
        let empty: std::collections::HashMap<String, String> = std::collections::HashMap::new();
        let mut cmd = std::process::Command::new("true");

        // Temporarily unset locale vars to test fallback
        let saved = (
            std::env::var("LC_ALL").ok(),
            std::env::var("LC_CTYPE").ok(),
            std::env::var("LANG").ok(),
        );
        lean_ctx::test_env::remove_var("LC_ALL");
        lean_ctx::test_env::remove_var("LC_CTYPE");
        lean_ctx::test_env::remove_var("LANG");

        ensure_utf8_locale(&mut cmd, &empty);

        // Restore
        if let Some(v) = saved.0 {
            lean_ctx::test_env::set_var("LC_ALL", v);
        }
        if let Some(v) = saved.1 {
            lean_ctx::test_env::set_var("LC_CTYPE", v);
        }
        if let Some(v) = saved.2 {
            lean_ctx::test_env::set_var("LANG", v);
        }

        // Command internal env isn't inspectable, but we verify the fn doesn't panic
        // and the real integration test below checks byte-level correctness.
    }

    #[test]
    fn ensure_utf8_locale_skips_when_extra_env_has_lc_all() {
        let mut extra = std::collections::HashMap::new();
        extra.insert("LC_ALL".to_string(), "C".to_string());
        let mut cmd = std::process::Command::new("true");
        ensure_utf8_locale(&mut cmd, &extra);
        // Should not panic or override
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn utf8_bytes_survive_shell_roundtrip() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let (output, code) = execute_command_in(
            "printf '\\xD0\\x9F\\xD1\\x80\\xD0\\xB8\\xD0\\xB2\\xD0\\xB5\\xD1\\x82'",
            ".",
        );
        assert_eq!(code, 0, "printf failed: {output}");
        assert_eq!(output, "Привет", "Cyrillic bytes must survive roundtrip");
    }

    #[test]
    #[cfg_attr(windows, ignore)] // ReadToEnd() blocks indefinitely on Windows CI
    fn execute_command_closes_stdin() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let command = "sh -c 'if read -t 1 line; then echo 67890; else echo 12345; fi'";
        let (output, code) = execute_command_in(command, ".");
        assert_eq!(code, 0, "command failed: {output}");
        assert!(
            output.contains("12345"),
            "child process should receive EOF on stdin, got: {output}"
        );
    }

    /// #1881: a command terminated by a signal reports `128 + signal`, not 1.
    /// Exit 1 plus output would even pass as a grep-style success (#1090).
    #[test]
    #[cfg(unix)]
    fn execute_command_reports_signal_termination_as_128_plus_signal() {
        let (output, code) = execute_command_in("echo before-kill; kill -TERM $$", ".");
        assert_eq!(code, 143, "SIGTERM must map to 143, got {code}: {output}");
        assert!(output.contains("before-kill"), "output kept: {output}");

        let (_, code) = execute_command_in("kill -KILL $$", ".");
        assert_eq!(code, 137, "SIGKILL must map to 137");
    }

    /// #945: a command that finishes but leaves a process holding the stdout
    /// pipe open (here a backgrounded `sleep`) must NOT lose the foreground
    /// output. The old `recv_timeout(...).unwrap_or_default()` discarded
    /// everything on the reader timeout; now the captured prefix survives and
    /// the still-draining reader is flagged instead of returning a bare exit.
    #[test]
    #[cfg_attr(windows, ignore)] // POSIX backgrounding (`&`) + sleep
    fn background_pipe_holder_keeps_foreground_output() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let (output, code) = execute_command_in("echo REPRO_CANARY_945; sleep 4 &", ".");
        assert_eq!(code, 0, "command should succeed: {output:?}");
        assert!(
            output.contains("REPRO_CANARY_945"),
            "foreground stdout must survive a lingering background pipe holder, got: {output:?}"
        );
        assert!(
            output.contains("output reader still draining"),
            "an incomplete reader must be flagged, got: {output:?}"
        );
    }

    /// A command runs without a controlling terminal, so an interactive
    /// descendant cannot take the terminal's foreground from the agent.
    /// (Vacuous where the test runner itself has no terminal, as in CI.)
    #[test]
    #[cfg(unix)]
    fn command_has_no_controlling_terminal() {
        let (output, code) = execute_command_in(
            "if (exec 3</dev/tty) 2>/dev/null; then echo HAS_TTY; else echo NO_TTY; fi",
            ".",
        );
        assert_eq!(code, 0, "{output:?}");
        assert!(output.contains("NO_TTY"), "got: {output:?}");
    }

    #[test]
    #[cfg_attr(windows, ignore)]
    fn forwards_captured_agent_runtime_env() {
        // #370: the MCP server process lacks CODEX_THREAD_ID; a hook captured it
        // from the agent shell. ctx_shell must still forward it to the child.
        // The capture lands in `state_dir()`, so all category dirs must be
        // isolated — pinning only the data dir let parallel tests race it.
        let _iso = lean_ctx::core::data_dir::isolated_data_dir();

        // Simulate a hook capturing the var from the native agent environment.
        lean_ctx::test_env::remove_var("CODEX_THREAD_ID");
        lean_ctx::test_env::set_var("CODEX_THREAD_ID", "thread-from-hook");
        lean_ctx::core::agent_runtime_env::capture();
        // The MCP server process itself does not carry the var.
        lean_ctx::test_env::remove_var("CODEX_THREAD_ID");

        let (output, code) = execute_command_in("printf 'TID=%s' \"$CODEX_THREAD_ID\"", ".");

        assert_eq!(code, 0, "command failed: {output}");
        assert!(
            output.contains("TID=thread-from-hook"),
            "captured agent runtime var must be forwarded, got: {output}"
        );
    }

    /// Per-call `timeout_ms` must reach the kill loop: a command that sleeps
    /// past its 200ms budget is killed with the timeout exit code and message.
    #[test]
    #[cfg_attr(windows, ignore)] // POSIX sleep
    fn per_call_timeout_kills_long_command() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let (output, code) = super::execute_command_with_env(
            "sleep 3",
            ".",
            &std::collections::HashMap::new(),
            Some(200),
        );
        assert_eq!(code, 124, "timed-out command must exit 124: {output}");
        assert!(
            output.contains("timed out after 200ms"),
            "timeout message must carry the per-call budget, got: {output}"
        );
    }

    /// #995: bytes emitted before a timeout remain visible; a timeout notice is
    /// additive rather than a replacement for useful subprocess output.
    #[test]
    #[cfg_attr(windows, ignore)] // POSIX sleep
    fn per_call_timeout_preserves_partial_output() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let (output, code) = super::execute_command_with_env(
            "printf TIMEOUT_PARTIAL_995; sleep 3",
            ".",
            &std::collections::HashMap::new(),
            Some(200),
        );
        assert_eq!(code, 124, "timed-out command must exit 124: {output}");
        assert!(
            output.contains("TIMEOUT_PARTIAL_995"),
            "stdout emitted before timeout must be preserved: {output:?}"
        );
        assert!(
            output.contains("timed out after 200ms"),
            "timeout notice must remain explicit: {output:?}"
        );
    }

    #[test]
    fn git_version_returns_when_git_is_available() {
        let _lock = lean_ctx::core::data_dir::test_env_lock();
        let git_available = std::process::Command::new("git")
            .arg("--version")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .is_ok();
        if !git_available {
            return;
        }

        let (output, code) = execute_command_in("git --version", ".");
        assert_eq!(code, 0, "git command failed: {output}");
        assert!(
            output.to_ascii_lowercase().contains("git version"),
            "unexpected git output: {output}"
        );
    }
}