kaish-kernel 0.16.0

//! Command dispatch — the single execution path for all commands.
//!
//! The `CommandDispatcher` trait defines how a single command is resolved and
//! executed. The Kernel implements this trait with the full dispatch chain:
//! user tools → builtins → .kai scripts → external commands → backend tools.
//!
//! `PipelineRunner` calls `dispatcher.dispatch()` for each command in a
//! pipeline, handling I/O routing (stdin piping, redirects) around each call.
//!
//! ```text
//! Stmt::Command ──┐
//!                  ├──▶ execute_pipeline() ──▶ PipelineRunner::run(dispatcher, commands, ctx)
//! Stmt::Pipeline ──┘                                  │
//!                                               for each command:
//!                                                 dispatcher.dispatch(cmd, ctx)
//!                                                     │
//!                                               ┌─────┼──────────────┐
//!                                               │     │              │
//!                                          user_tools builtins  .kai scripts
//!                                                                external cmds
//!                                                                backend tools
//! ```

use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;

use crate::ast::{Command, Expr, Stmt, Value};
use crate::interpreter::ExecResult;
use crate::tools::ExecContext;
#[cfg(test)]
use crate::tools::{
    external_commands_unavailable_error, ExternalCommandOutcome, ExternalCommandsUnavailable,
};

// The following imports are only used by the test-only `BackendDispatcher`.
#[cfg(test)]
use crate::ast::Arg;
#[cfg(test)]
use crate::backend::BackendError;
#[cfg(test)]
use crate::interpreter::apply_output_format;
#[cfg(test)]
use crate::scheduler::build_tool_args;
#[cfg(test)]
use crate::tools::{GlobalFlags, ToolRegistry};
#[cfg(all(test, feature = "subprocess"))]
use crate::tools::{resolve_in_path, virtual_cwd_error};

/// Arm `PR_SET_PDEATHSIG(SIGKILL)` in a freshly forked child, so the OS kills
/// it the moment `parent_pid` dies — for any reason, including `kill -9`, a
/// segfault, or an OOM kill, none of which let the parent run a single
/// instruction of cleanup. This is the one orphan guard that does not depend
/// on `setpgid` + a pidfd kill, `kill_on_drop`, or any other code of ours
/// getting to run.
///
/// **Call only between fork and exec.** `prctl` and `getppid` are both
/// async-signal-safe per POSIX, which is what makes that legal.
///
/// The `getppid` check closes `PR_SET_PDEATHSIG`'s documented race: if the
/// parent dies in the window between `fork` and the `prctl` above, the signal
/// is armed against a parent that is already gone and will never be delivered
/// — the exact orphan the flag exists to prevent, in the exact window it is
/// hardest to notice. Comparing against the pid the parent captured *before*
/// forking detects it, and failing the `pre_exec` fails the spawn loudly
/// rather than exec'ing a process nothing will ever reap.
///
/// Linux only. macOS has no equivalent that works without a live watcher
/// process, so this is compiled out there rather than faked with something
/// weaker — see `KernelConfig::kill_children_on_parent_death`.
#[cfg(all(unix, feature = "subprocess"))]
pub(crate) fn arm_parent_death_signal(parent_pid: u32) -> std::io::Result<()> {
    #[cfg(target_os = "linux")]
    {
        nix::sys::prctl::set_pdeathsig(nix::sys::signal::Signal::SIGKILL)
            .map_err(|e| std::io::Error::from_raw_os_error(e as i32))?;

        if nix::unistd::getppid().as_raw() as u32 != parent_pid {
            return Err(std::io::Error::other(
                "parent died before the parent-death signal was armed",
            ));
        }
    }
    #[cfg(not(target_os = "linux"))]
    let _ = parent_pid;
    Ok(())
}

/// Position of a command within a pipeline.
///
/// Used by external command execution to decide stdio inheritance:
/// - `Only` or `Last` in interactive mode → inherit terminal
/// - `First` or `Middle` → always capture
///
/// Not `#[non_exhaustive]`, deliberately: kaish pipelines are a strictly
/// linear chain of stages, so these four variants exhaust every position a
/// stage can occupy. A fifth would mean pipelines stopped being linear — a
/// grammar change, not a variant this enum grows on its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum PipelinePosition {
    /// Single command, no pipe.
    #[default]
    Only,
    /// First command in a pipeline (no stdin from pipe).
    First,
    /// Middle of a pipeline (piped stdin, piped stdout).
    Middle,
    /// Last command in a pipeline (piped stdin, final output).
    Last,
}

/// Trait for dispatching a single command through the full resolution chain.
///
/// Implementations handle argument parsing, tool lookup, and execution.
/// The pipeline runner handles I/O routing (stdin, redirects, piping).
#[async_trait]
pub trait CommandDispatcher: Send + Sync {
    /// Dispatch a single command for execution.
    ///
    /// The `ctx` provides stdin (from pipe or redirect), scope, and backend.
    /// Implementations should handle schema-aware argument parsing and
    /// output format extraction internally.
    async fn dispatch(&self, cmd: &Command, ctx: &mut ExecContext) -> Result<ExecResult>;

    /// Dispatch a compound statement (`if`, `for`, `while`, `case`) that sits
    /// in a pipeline stage.
    ///
    /// The statement runs to completion and its whole output comes back in the
    /// `ExecResult`; `PipelineRunner` then writes those bytes to the pipe. So
    /// `ctx.pipe_stdout` must stay with the runner — hand it to the statement
    /// and the first nested command inside it would take the writer and the
    /// rest of the loop would write nowhere.
    ///
    /// The default rejects the form. Only a dispatcher that can execute a
    /// whole statement (the `Kernel`) overrides it; a dispatcher that resolves
    /// one command at a time has nothing to run a loop body with, and saying
    /// so beats returning empty output at exit 0.
    async fn dispatch_stmt(&self, _stmt: &Stmt, _ctx: &mut ExecContext) -> Result<ExecResult> {
        anyhow::bail!("this dispatcher cannot run a compound statement in a pipeline stage")
    }

    /// Evaluate an expression through the full async chain.
    ///
    /// Unlike the runner's sync `eval_simple_expr`, this can run command
    /// substitution (`$(...)`) because it has access to pipeline execution.
    /// Used for redirect targets and heredoc bodies so `cat < $(cmd)`,
    /// `echo x > $(cmd)`, and `$(...)` inside heredoc bodies work. The `ctx`
    /// carries scope/cwd/backend for dispatchers that evaluate against it;
    /// stateful dispatchers (Kernel) snapshot their own session state and
    /// only let command output escape (side effects like `cd` do not).
    async fn eval_expr(&self, expr: &Expr, ctx: &ExecContext) -> Result<Value>;

    /// Fork the dispatcher for concurrent execution (detached).
    ///
    /// Returns a subsidiary dispatcher with independent mutable state, safe
    /// to run concurrently with the parent and other forks without data
    /// races on shared scope/cwd/aliases. Used by background `&` jobs,
    /// where the fork must survive parent cancellation.
    ///
    /// For stateful dispatchers (e.g. Kernel) this snapshots per-session
    /// state into a fresh instance. Stateless dispatchers may clone.
    async fn fork(&self) -> Arc<dyn CommandDispatcher>;

    /// Fork the dispatcher for concurrent execution (attached to parent cancel).
    ///
    /// Like [`Self::fork`] but the fork's cancellation token is a *child* of
    /// the parent's. Cancelling the parent (timeout, Ctrl-C, embedder
    /// `Kernel::cancel`) cascades into the fork, which then kills its own
    /// external children via the usual SIGTERM/SIGKILL discipline.
    ///
    /// Used for foreground concurrency: scatter workers, concurrent pipeline
    /// stages, command substitution. Default implementation delegates to
    /// [`Self::fork`] for stateless dispatchers that don't track cancellation.
    async fn fork_attached(&self) -> Arc<dyn CommandDispatcher> {
        self.fork().await
    }
}

/// Minimal stateless dispatcher used by pipeline/runner unit tests.
///
/// Production code uses `Kernel` (via `Kernel::fork` for concurrent contexts).
/// This test-only dispatcher routes directly through `backend.call_tool()` so
/// the pipeline runner can be exercised without spinning up a full Kernel.
///
/// Limitations (intentional — these are test-only constraints):
/// - No user-defined tools
/// - No .kai script resolution
/// - No async argument evaluation (command substitution in args won't work)
#[cfg(test)]
pub(crate) struct BackendDispatcher {
    tools: Arc<ToolRegistry>,
}

#[cfg(test)]
impl BackendDispatcher {
    /// Create a new backend dispatcher with the given tool registry.
    pub(crate) fn new(tools: Arc<ToolRegistry>) -> Self {
        Self { tools }
    }

    /// Try to execute an external command (PATH lookup + process spawn).
    ///
    /// Used as fallback when no builtin/backend tool matches. `Unavailable`
    /// if kaish will not attempt this at all (kept in sync with
    /// kernel.rs::try_execute_external — see `ExternalCommandOutcome`).
    /// Always captures stdout/stderr (never inherits terminal — pipeline
    /// stages don't need interactive I/O).
    #[cfg(not(feature = "subprocess"))]
    async fn try_external(
        &self,
        _name: &str,
        _args: &[Arg],
        _ctx: &mut ExecContext,
    ) -> ExternalCommandOutcome {
        ExternalCommandOutcome::Unavailable(ExternalCommandsUnavailable::NotCompiled)
    }

    /// Try to execute an external command (PATH lookup + process spawn).
    #[cfg(feature = "subprocess")]
    async fn try_external(
        &self,
        name: &str,
        args: &[Arg],
        ctx: &mut ExecContext,
    ) -> ExternalCommandOutcome {
        if !ctx.allow_external_commands {
            return ExternalCommandOutcome::Unavailable(ExternalCommandsUnavailable::ConfiguredOff);
        }
        match self.try_external_on_path(name, args, ctx).await {
            Some(result) => ExternalCommandOutcome::Ran(Box::new(result)),
            None => ExternalCommandOutcome::NotFound,
        }
    }

    /// The actual PATH lookup + spawn, once the caller has confirmed
    /// external commands are allowed at all — kept in sync with
    /// kernel.rs::try_execute_external_on_path.
    #[cfg(feature = "subprocess")]
    async fn try_external_on_path(
        &self,
        name: &str,
        args: &[Arg],
        ctx: &mut ExecContext,
    ) -> Option<ExecResult> {
        // Real filesystem location of the shell's cwd, if any. A `None` real
        // path means the cwd is virtual (a CoW overlay, an in-memory VFS
        // mount, …) — there's nowhere for a child OS process to run. Don't
        // bail out here: a bare command name that isn't in PATH at all is a
        // genuine "not found" regardless of cwd. Once the command actually
        // resolves, `real_cwd` is checked again below and the honest reason
        // is given then — kept in sync with kernel.rs::try_execute_external
        // (issue #181).
        let real_cwd = ctx.backend.resolve_real_path(&ctx.cwd);

        // Resolve command: absolute/relative path or PATH lookup
        let executable = if name.contains('/') {
            // Resolve relative paths (./script, ../bin/tool) against the shell's cwd
            let resolved = if std::path::Path::new(name).is_absolute() {
                std::path::PathBuf::from(name)
            } else {
                match &real_cwd {
                    Some(real_cwd) => real_cwd.join(name),
                    // Can't resolve a relative path without a real cwd to
                    // join against, so we can't even tell whether it would
                    // exist — name the actual blocker.
                    None => return Some(virtual_cwd_error(name, &ctx.cwd)),
                }
            };
            // Kept in sync with kernel.rs::try_execute_external (issue
            // #229): `exists()` alone isn't enough — a directory or a
            // non-executable file both "exist" but must fail with the
            // clean, documented exit-126 class instead of falling through
            // to `Command::spawn()` and leaking whatever raw OS error comes
            // back (e.g. "Permission denied (os error 13)" under exit 127).
            if !resolved.exists() {
                return Some(ExecResult::failure(127, format!("{}: No such file or directory", name)));
            }
            if !resolved.is_file() {
                return Some(ExecResult::failure(126, format!("{}: Is a directory", name)));
            }
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let mode = std::fs::metadata(&resolved)
                    .map(|m| m.permissions().mode())
                    .unwrap_or(0);
                if mode & 0o111 == 0 {
                    return Some(ExecResult::failure(126, format!("{}: Permission denied", name)));
                }
            }
            resolved.to_string_lossy().into_owned()
        } else {
            // PATH from scope only — never OS env (keeps this test-only spawn
            // site in sync with kernel.rs::try_execute_external).
            let path_var = ctx.scope.get("PATH")
                .map(crate::interpreter::value_to_string)
                .unwrap_or_default();
            resolve_in_path(name, &path_var)?
        };

        // The executable resolved — found in PATH, or a path that exists —
        // but there's still nowhere to run it without a real cwd.
        let real_cwd = match real_cwd {
            Some(p) => p,
            None => return Some(virtual_cwd_error(name, &ctx.cwd)),
        };

        // Build flat argv from args. A for-loop (not filter_map) so the
        // Decision D collection-argv guard can short-circuit the whole spawn
        // — kept in sync with the production build in kernel.rs::build_args_flat.
        let mut argv: Vec<String> = Vec::new();
        for arg in args {
            match arg {
                Arg::Positional(expr) => match expr {
                    Expr::Literal(Value::String(s)) => argv.push(s.clone()),
                    Expr::Literal(Value::Int(i)) => argv.push(i.to_string()),
                    Expr::Literal(Value::Float(f)) => argv.push(f.to_string()),
                    Expr::VarRef(path) => {
                        if let Ok(v) = ctx.scope.resolve_path(path) {
                            if let Some(msg) = crate::interpreter::structured_boundary_error("a command argument", &v) {
                                return Some(ExecResult::failure(1, msg));
                            }
                            // Text sink: binary goes loud (kept in sync with
                            // kernel.rs::build_args_flat).
                            match crate::interpreter::value_to_text_sink(&v) {
                                Ok(s) => argv.push(s),
                                Err(e) => return Some(ExecResult::failure(1, e.to_string())),
                            }
                        }
                    }
                    // Remaining literal types (Bool/Json/Null/Bytes) — kept in
                    // sync with the production build_args_flat, which resolves
                    // every positional through value_to_text_sink (binary loud).
                    Expr::Literal(other) => match crate::interpreter::value_to_text_sink(other) {
                        Ok(s) => argv.push(s),
                        Err(e) => return Some(ExecResult::failure(1, e.to_string())),
                    },
                    _ => {}
                },
                Arg::ShortFlag(f) => argv.push(format!("-{f}")),
                Arg::LongFlag(f) => argv.push(format!("--{f}")),
                Arg::Named { key, value } => match value {
                    Expr::Literal(Value::String(s)) => argv.push(format!("--{key}={s}")),
                    _ => argv.push(format!("--{key}=")),
                },
                Arg::WordAssign { key, value } => match value {
                    Expr::Literal(Value::String(s)) => argv.push(format!("{key}={s}")),
                    _ => argv.push(format!("{key}=")),
                },
                Arg::DoubleDash => argv.push("--".to_string()),
            }
        }

        // Check for streaming pipes
        let has_pipe_stdin = ctx.pipe_stdin.is_some();
        let has_buffered_stdin = ctx.stdin.is_some();

        // Spawn process
        use tokio::process::Command;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let mut cmd = Command::new(&executable);
        cmd.args(&argv);
        cmd.current_dir(&real_cwd);
        cmd.kill_on_drop(true);

        // Hermetic env: child sees only kaish's exported vars, not the kaish
        // process's OS env. Frontends that want OS-env passthrough (REPL, MCP)
        // populate it via KernelConfig::initial_vars at construction.
        cmd.env_clear();
        let exported = ctx.scope.exported_vars();
        // A structured value can't cross the process boundary; refuse rather than
        // silently JSON-serialize it into the child's environment. Kept in sync
        // with the production spawn site in kernel.rs::try_execute_external.
        if let Some(msg) = crate::interpreter::structured_export_error(&exported) {
            return Some(ExecResult::failure(1, msg));
        }
        for (var_name, value) in exported {
            // Binary can't cross the process boundary as an env var value
            // either — loud, not the `[binary: N bytes]` placeholder (kept in
            // sync with the production spawn site).
            match crate::interpreter::value_to_text_sink_named(
                &value,
                "an exported environment variable value",
            ) {
                Ok(s) => {
                    cmd.env(var_name, s);
                }
                Err(e) => return Some(ExecResult::failure(1, e.to_string())),
            }
        }

        // Stdin: pipe_stdin or buffered bytes or inherit (interactive) or null
        cmd.stdin(if has_pipe_stdin || has_buffered_stdin {
            std::process::Stdio::piped()
        } else if ctx.interactive && matches!(ctx.pipeline_position, PipelinePosition::First | PipelinePosition::Only) {
            std::process::Stdio::inherit()
        } else {
            std::process::Stdio::null()
        });
        cmd.stdout(std::process::Stdio::piped());
        cmd.stderr(std::process::Stdio::piped());

        // On Unix, always put the child in its own process group so a
        // cancel can `killpg` the whole tree (the child plus any
        // grandchildren) — matching the production spawn site
        // (kernel.rs::try_execute_external) exactly. Without this, `killpg`
        // targets a group nobody is actually in (an ESRCH no-op), and a
        // grandchild spawned by the child survives cancellation — the exact
        // gap GH #133 item 4 closes. This dispatcher has no job-control
        // terminal integration (no `terminal_state`), so unlike production
        // there is no signal-handler restoration to gate here.
        #[cfg(unix)]
        {
            let kill_on_parent_death = ctx.kill_children_on_parent_death;
            let parent_pid = std::process::id();
            // SAFETY: setpgid, prctl, and getppid are async-signal-safe per
            // POSIX; safe to call between fork and exec.
            #[allow(unsafe_code)]
            unsafe {
                cmd.pre_exec(move || {
                    nix::unistd::setpgid(nix::unistd::Pid::from_raw(0), nix::unistd::Pid::from_raw(0))
                        .map_err(|e| std::io::Error::from_raw_os_error(e as i32))?;
                    if kill_on_parent_death {
                        arm_parent_death_signal(parent_pid)?;
                    }
                    Ok(())
                });
            }
        }

        let mut child = match cmd.spawn() {
            Ok(c) => c,
            Err(e) => return Some(ExecResult::failure(127, format!("{}: {}", name, e))),
        };
        // Open a pidfd (Linux) for race-free direct-child kill via wait_or_kill.
        let kill_target = crate::pidfd::KillTarget::from_child(&child);

        // Stream stdin: copy pipe_stdin → child stdin in chunks (bounded memory)
        let stdin_task: Option<tokio::task::JoinHandle<()>> = if let Some(mut pipe_in) = ctx.pipe_stdin.take() {
            let prefix = ctx.stdin.take();
            child.stdin.take().map(|mut child_stdin| {
                tokio::spawn(async move {
                    // A buffered prefix and a live pipe are one stream, not two
                    // candidates — see the same reasoning in
                    // `kernel.rs::try_execute_external`, which this twin mirrors.
                    if let Some(data) = prefix
                        && child_stdin.write_all(&data).await.is_err()
                    {
                        return; // child closed stdin; drop signals EOF
                    }
                    let mut buf = [0u8; 8192];
                    loop {
                        match pipe_in.read(&mut buf).await {
                            Ok(0) => break, // EOF
                            Ok(n) => {
                                if child_stdin.write_all(&buf[..n]).await.is_err() {
                                    break; // child closed stdin
                                }
                            }
                            Err(_) => break,
                        }
                    }
                    // Drop child_stdin signals EOF to child
                })
            })
        } else if let Some(data) = ctx.stdin.take() {
            // Buffered stdin bytes written from a DETACHED task, not inline:
            // an inline write deadlocks once the stdin pipe fills before the
            // output drain below has spawned (mirrors the kernel.rs fix; keeps
            // the two spawn sites in sync). Drop signals EOF; a broken pipe
            // (child closed stdin early) is fine.
            child.stdin.take().map(|mut child_stdin| {
                tokio::spawn(async move {
                    let _ = child_stdin.write_all(&data).await;
                })
            })
        } else {
            None
        };

        // Capture stdout via the spill-aware collector, regardless of whether
        // this is a pipeline stage (`ctx.pipe_stdout` set) or the last/only
        // stage. This intentionally does NOT special-case `ctx.pipe_stdout`
        // — production's `try_execute_external` never touches that field at
        // all; a middle/first pipeline stage's forwarding to the next stage
        // is entirely `PipelineRunner::run_pipeline`'s job (pipeline.rs),
        // which reads `stage_ctx.pipe_stdout` (still `Some`, untouched here)
        // after `dispatch()` returns and forwards `result.out` itself.
        //
        // Before this fix, this dispatcher special-cased `pipe_stdout` and
        // streamed the child's stdout straight through in 8KB chunks — full
        // fidelity, no cap. Production has no such fast path: every external
        // stage's stdout is captured here first, then forwarded by the
        // runner, so a >10MB intermediate stage silently loses its head in
        // production (the runner's forward goes through the SAME capture,
        // still true after this fix — see GH #133 item 2 for the capture
        // primitive itself). Losing the pipe_stdout special case is what lets
        // a test reproduce that production bug class at all (GH #133 item 3).
        let Some(child_stdout) = child.stdout.take() else {
            return Some(ExecResult::failure(1, "internal: stdout not available"));
        };
        let Some(mut child_stderr) = child.stderr.take() else {
            return Some(ExecResult::failure(1, "internal: stderr not available"));
        };

        // Capture stdout into a fixed 10MB tail-evicting ring (`BoundedStream`
        // + `drain_to_stream`) — the SAME capture primitive the production
        // spawn site uses (kernel.rs::try_execute_external), not the
        // limit-aware `spill_aware_collect` this used to call. Production
        // does not spill-check an external command's own capture inline
        // against `ctx.output_limit`; the pipeline-level post-hoc
        // `spill_if_needed` (`Kernel::execute_pipeline`) is what applies that
        // afterward, and `did_spill` is left `false` here for THAT reason — a
        // caller wanting the limit-aware post-hoc behavior applies it
        // separately, same as the real pipeline path (GH #133 item 2).
        // Independently, `did_spill` CAN still end up `true` below: if the
        // ring itself overflows (unconditionally, regardless of
        // `ctx.output_limit`), that's the GH #191 loud-overflow signal, not
        // the limit-aware spill this comment is about.
        let stdout_stream = Arc::new(crate::scheduler::BoundedStream::new(
            crate::scheduler::DEFAULT_STREAM_MAX_SIZE,
        ));
        let stdout_clone = stdout_stream.clone();
        let stdout_task = tokio::spawn(async move {
            crate::scheduler::drain_to_stream(child_stdout, stdout_clone).await;
        });

        // Stderr streaming is intentionally left as-is (live to
        // `ctx.stderr` when present, else buffered) — production instead
        // caps stderr into its own 10MB ring with no live streaming. That
        // divergence is out of scope for this PR; see GH #133 follow-ups.
        let stderr_stream_handle = ctx.stderr.clone();
        let stderr_task = tokio::spawn(async move {
            let mut buf = Vec::new();
            let mut chunk = [0u8; 8192];
            loop {
                match child_stderr.read(&mut chunk).await {
                    Ok(0) => break,
                    Ok(n) => {
                        if let Some(ref stream) = stderr_stream_handle {
                            stream.write(&chunk[..n]);
                        } else {
                            buf.extend_from_slice(&chunk[..n]);
                        }
                    }
                    Err(_) => break,
                }
            }
            if stderr_stream_handle.is_some() {
                String::new()
            } else {
                String::from_utf8_lossy(&buf).into_owned()
            }
        });

        let cancel = ctx.cancel.clone();
        // Mirror production's cancel-aware drain handling: spawn the
        // drains concurrently with the wait (not after collection
        // completes) so a cancel can actually interrupt a still-running,
        // still-silent child instead of blocking until it produces EOF.
        let cancelled_before_wait = cancel.is_cancelled();
        let status = crate::kernel::wait_or_kill(
            &mut child,
            kill_target.as_ref(),
            &cancel,
            std::time::Duration::from_secs(2),
        ).await;
        if let Some(task) = stdin_task { task.abort(); }
        let mut stderr = if cancelled_before_wait || cancel.is_cancelled() {
            // The child's pipes are gone; late output is lost but
            // predictable death beats partial capture (same tradeoff
            // production makes).
            stdout_task.abort();
            stderr_task.abort();
            String::new()
        } else {
            let _ = stdout_task.await;
            stderr_task.await.unwrap_or_default()
        };

        // Signal-death mapping (128+signal, e.g. SIGKILL→137) must match
        // the production spawn site exactly — kept in sync via the shared
        // `exit_code_from_status` helper (GH #133 item 1). A `wait_or_kill`
        // I/O error (not a signal death) falls back to 1, same as before.
        let code = match status {
            Ok(s) => crate::kernel::exit_code_from_status(&s),
            Err(_) => 1,
        };
        let stdout = stdout_stream.read().await;
        // stdout came back as raw bytes: text if valid UTF-8, else a Bytes
        // result (so `curl url`, `curl url > file.bin`, etc. keep binary intact).
        let mut result = ExecResult::success_text_or_bytes(stdout).with_code(code);

        // Mirror production's overflow signaling (GH #191) for the piece this
        // twin actually shares with `kernel.rs::try_execute_external`: the
        // stdout `BoundedStream` ring. Stderr here is captured differently
        // from production (live-streamed to `ctx.stderr` when set, else an
        // unbounded `Vec` — see the comment above `stderr_stream_handle`,
        // GH #133 follow-up), so there is no stderr `BoundedStream` overflow
        // to mirror; only the stdout side applies. `did_spill` stays `false`
        // otherwise, matching `output_limit_is_not_applied_inline_matching_production`
        // below — this is the fixed-ring overflow signal, not the
        // limit-aware post-hoc spill Kernel::execute_pipeline applies.
        if stdout_stream.has_overflowed().await {
            let stats = stdout_stream.stats().await;
            stderr = format!("{}{stderr}", stats.overflow_marker("stdout"));
            result.did_spill = true;
        }
        result.err = stderr;
        Some(result)
    }
}

#[cfg(test)]
#[async_trait]
impl CommandDispatcher for BackendDispatcher {
    async fn dispatch(&self, cmd: &Command, ctx: &mut ExecContext) -> Result<ExecResult> {
        // Handle built-in true/false/: (`:` is another spelling of `true`)
        match cmd.name.as_str() {
            "true" | ":" => return Ok(ExecResult::success("")),
            "false" => return Ok(ExecResult::failure(1, "")),
            _ => {}
        }

        // Build tool args through the reduced sync evaluator (no command
        // substitution) — see `SyncEvalSource` in `scheduler::pipeline`.
        // A bad/subscripted collection access is a genuine PathError here too —
        // propagate it via `?` rather than swallowing, same as the production
        // Kernel::dispatch_command's `execute_command(..).await?`.
        let schema = self.tools.get(&cmd.name).map(|t| t.schema());
        let tool_args = build_tool_args(&cmd.args, ctx, schema.as_ref())
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        // Honor --json before the tool runs so a parse failure inside the
        // builtin doesn't drop the format on the floor. See kernel.rs for the
        // matching call in the production path.
        let raw_argv = schema.as_ref().is_some_and(|s| s.raw_argv);
        GlobalFlags::apply_from_args(&tool_args, raw_argv, ctx);

        // Execute via backend
        let backend = ctx.backend.clone();
        let result = match backend.call_tool(&cmd.name, tool_args, ctx).await {
            // Route through the same `From<ToolResult> for ExecResult` the
            // production dispatch path uses (kernel.rs) rather than
            // hand-rolling the field-by-field copy: the old inline version
            // wrapped `data` unconditionally as `Value::Json`, which skipped
            // `json_to_value_no_envelope`'s scalar-unwrap (`Value::Int`/
            // `Value::String`/…) and silently dropped `did_spill`/
            // `original_code` — a divergence this test-only dispatcher must
            // not have from the real path (GH #93 item 4).
            Ok(tool_result) => ExecResult::from(tool_result),
            Err(BackendError::ToolNotFound(_)) => {
                // Fall back to external command execution. The backend
                // registry already had its chance above, so — unlike the
                // production path in kernel.rs, which tries external first —
                // `Unavailable` is a final answer here, not something that
                // needs to survive a later backend lookup.
                match self.try_external(&cmd.name, &cmd.args, ctx).await {
                    ExternalCommandOutcome::Ran(result) => *result,
                    ExternalCommandOutcome::NotFound => {
                        ExecResult::failure(127, format!("command not found: {}", cmd.name))
                    }
                    ExternalCommandOutcome::Unavailable(reason) => {
                        external_commands_unavailable_error(&cmd.name, reason)
                    }
                }
            }
            Err(e) => ExecResult::failure(127, e.to_string()),
        };

        // Migrated builtins parse --json via the GlobalFlags flatten and
        // write ctx.output_format. The kernel just applies it.
        let result = match ctx.output_format {
            Some(format) => apply_output_format(result, format),
            None => result,
        };

        Ok(result)
    }

    /// Sync-only evaluation (no command substitution) — matches this
    /// test dispatcher's documented "no async argument evaluation" limit.
    async fn eval_expr(&self, expr: &Expr, ctx: &ExecContext) -> Result<Value> {
        crate::scheduler::pipeline::eval_simple_expr(expr, ctx)
            .map_err(|e| anyhow::anyhow!(e))?
            .ok_or_else(|| anyhow::anyhow!("cannot evaluate expression in test dispatcher"))
    }

    /// BackendDispatcher is stateless, so a fork is just a clone.
    async fn fork(&self) -> Arc<dyn CommandDispatcher> {
        Arc::new(Self { tools: Arc::clone(&self.tools) })
    }
}

/// Tests that spawn real external processes through `try_external`, to catch
/// behavioral drift from the production spawn site (`kernel.rs::try_execute_external`)
/// — GH #133. Unlike the `BackendDispatcher` tests in `scheduler::pipeline`,
/// which exercise builtins over a `MemoryFs` (virtual cwd, so `try_external`
/// never spawns), these give the dispatcher a real tempdir cwd + PATH so the
/// external fallback actually runs a child process.
#[cfg(all(test, feature = "subprocess"))]
mod external_process_tests {
    // Test-fixture helpers (not `#[test]` bodies themselves), so the
    // workspace's usual allow-in-tests clippy.toml carve-out doesn't cover
    // them — see CLAUDE.md's "clap builtin gotchas" / test-code conventions.
    #![allow(clippy::unwrap_used, clippy::expect_used)]
    use super::*;
    use crate::ast::{Arg, Command, Expr, Value};
    use crate::tools::{ExecContext, ToolRegistry};
    use crate::vfs::{LocalFs, VfsRouter};

    /// A `BackendDispatcher` + `ExecContext` rooted at a real tempdir, with an
    /// empty tool registry (every command name falls through to
    /// `try_external`, exactly like a real external command with no matching
    /// builtin/user tool) and PATH seeded from the test process's own OS env.
    /// Reading OS env here is fixture code, not kaish's hermetic runtime — see
    /// CLAUDE.md and `external_command_tests.rs::repl_kernel`.
    fn real_cwd_dispatcher() -> (BackendDispatcher, ExecContext, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("tempdir");
        let mut vfs = VfsRouter::new();
        vfs.mount("/", LocalFs::new(dir.path().to_path_buf()));
        let tools = Arc::new(ToolRegistry::new());
        let mut ctx = ExecContext::with_vfs_and_tools(Arc::new(vfs), tools.clone());
        // Exported (not just set): try_external's own PATH lookup reads
        // ctx.scope directly, but the CHILD process only inherits exported
        // vars (cmd.env_clear() + exported_vars()) — a script that shells
        // out further (`sh -c "yes | head"`) needs PATH in ITS env too,
        // not just kaish's resolver.
        ctx.scope.set_exported(
            "PATH",
            Value::String(std::env::var("PATH").unwrap_or_default()),
        );
        let dispatcher = BackendDispatcher::new(tools);
        (dispatcher, ctx, dir)
    }

    /// `sh -c <script>` as a `Command`, matching how the parser would build it
    /// from `sh -c 'script'` (a short flag, then a positional literal).
    fn sh_cmd(script: &str) -> Command {
        Command {
            name: "sh".to_string(),
            args: vec![
                Arg::ShortFlag("c".to_string()),
                Arg::Positional(Expr::Literal(Value::String(script.to_string()))),
            ],
            redirects: vec![],
        }
    }

    /// GH #133 item 1: production maps a signal-killed child to `128 + signal`
    /// (SIGKILL -> 137); the twin used to hardcode `code().unwrap_or(1)` -> 1,
    /// so a cancel/timeout test run through this dispatcher observed an exit
    /// code production never actually produces. Fails at `code == 1` pre-fix.
    #[tokio::test]
    async fn signal_killed_child_maps_to_128_plus_signal() {
        let (dispatcher, mut ctx, _dir) = real_cwd_dispatcher();
        let cmd = sh_cmd("kill -KILL $$");
        let result = dispatcher.dispatch(&cmd, &mut ctx).await.expect("dispatch");
        assert_eq!(
            result.code, 137,
            "SIGKILL should map to 128+9=137 (production's mapping), got {}",
            result.code
        );
    }

    /// GH #133 item 2: the twin used to call the limit-aware
    /// `spill_aware_collect` in its non-pipe capture branch, applying
    /// `ctx.output_limit` inline and setting `did_spill` itself. Production's
    /// `try_execute_external` never spill-checks its own capture that way —
    /// spill is a pipeline-level, post-hoc step (`Kernel::execute_pipeline`
    /// calls `spill_if_needed` AFTER the dispatcher returns). So even with a
    /// tiny `output_limit` configured, `try_external` itself must return the
    /// full (up to the 10MB ring) captured output with `did_spill == false`.
    /// Pre-fix, the twin truncated inline and set `did_spill = true` here.
    #[tokio::test]
    async fn output_limit_is_not_applied_inline_matching_production() {
        let (dispatcher, mut ctx, _dir) = real_cwd_dispatcher();
        // A tiny in-memory limit (no disk spill file — CLAUDE.md: no real
        // system paths in tests) — if try_external still spill-checked
        // inline (the bug), this would trigger truncation right here.
        ctx.output_limit = crate::output_limit::OutputLimitConfig::agent().in_memory();
        ctx.output_limit.set_limit(Some(64));

        let cmd = sh_cmd("yes x | head -c 1000");
        let result = dispatcher.dispatch(&cmd, &mut ctx).await.expect("dispatch");

        assert_eq!(result.code, 0, "err: {}", result.err);
        assert_eq!(
            result.text_out().len(),
            1000,
            "try_external must return the full captured output — production \
             defers spill to the post-hoc pipeline step, not its own capture; \
             got {} bytes: {:?}",
            result.text_out().len(),
            result.text_out()
        );
        assert!(
            !result.did_spill,
            "try_external itself must not set did_spill — that's \
             Kernel::execute_pipeline's post-hoc spill_if_needed's job, \
             matching production"
        );
    }

    /// GH #133 item 3: before this fix, `try_external` special-cased
    /// `ctx.pipe_stdout` — taking it out of the context and hand-streaming
    /// the child's stdout straight into it in 8KB chunks, bypassing the
    /// capture logic a non-pipeline external goes through, and always
    /// returning an empty `result.out` ("output was streamed to pipe").
    /// Production's `try_execute_external` has no such special case: it never
    /// reads or writes `ctx.pipe_stdout` at all — `PipelineRunner::run_pipeline`
    /// (pipeline.rs) is solely responsible for reading a stage's captured
    /// `result.out` back out and forwarding it to the next stage.
    #[tokio::test]
    async fn pipeline_stage_leaves_pipe_stdout_for_the_runner_to_forward() {
        let (dispatcher, mut ctx, _dir) = real_cwd_dispatcher();

        // Simulate what PipelineRunner::run_pipeline wires onto a first/middle
        // stage's ctx before calling dispatch(): a pipe_stdout the runner
        // expects to read back out afterward.
        let (writer, reader) = crate::scheduler::pipe_stream_default();
        ctx.pipe_stdout = Some(writer);

        // Drain the reader concurrently — a full-fidelity writer (the old
        // special case) would otherwise still work here for a small payload,
        // but this also lets the pipe close out cleanly either way.
        let drain = tokio::spawn(async move {
            use tokio::io::AsyncReadExt;
            let mut reader = reader;
            let mut buf = Vec::new();
            let _ = reader.read_to_end(&mut buf).await;
            buf
        });

        let cmd = sh_cmd("echo hello");
        // A generous but bounded timeout: a real hang here (e.g. an
        // accidental deadlock reintroduced by a future edit) should fail
        // loud and fast in CI, not stall the suite indefinitely.
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(15),
            dispatcher.dispatch(&cmd, &mut ctx),
        )
        .await
        .expect("dispatch timed out")
        .expect("dispatch");

        assert!(
            ctx.pipe_stdout.is_some(),
            "try_external must leave ctx.pipe_stdout untouched — forwarding \
             to the next stage is PipelineRunner's job, matching production, \
             which never reads or writes this field at all"
        );

        // Drop the writer now (the runner would take it back out and, after
        // forwarding, let it go) so the reader sees EOF and `drain` actually
        // completes — nothing else in this test closes the pipe, since
        // try_external no longer touches it at all post-fix.
        drop(ctx.pipe_stdout.take());
        let _ = drain.await;

        assert!(
            result.text_out().contains("hello"),
            "try_external must capture and return stdout the same way for a \
             pipeline stage as a non-pipeline call (not force it empty \
             because a pipe was attached) — got: {:?}",
            result.text_out()
        );
    }

    /// GH #133 item 3, large-payload consequence: before this fix, a pipeline
    /// stage's stdout went through the hand-rolled full-fidelity streamer,
    /// which ignored any size cap entirely and forwarded byte-for-byte no
    /// matter the size — an intermediate stage had NO cap at all, of any
    /// kind. Post-fix, every stage (pipe or not) goes through the same
    /// capture path a non-pipeline external uses.
    ///
    /// Updated for GH #133 item 2 (landed since this test was written): the
    /// shared capture path now caps via an *unconditional* ~10MB
    /// `BoundedStream` ring regardless of `ctx.output_limit` configuration —
    /// production never spill-checks its own capture inline against
    /// `ctx.output_limit`, deferring THAT to the pipeline-level, post-hoc
    /// `spill_if_needed`. So `ctx.output_limit` is configured below only to
    /// prove it's inert here (matching item 2's contract) — it plays no part
    /// in why this payload gets capped.
    ///
    /// Updated again for GH #191: the fixed ring overflowing IS now loud on
    /// its own terms, independent of `ctx.output_limit`. `did_spill` flips to
    /// `true` (this dispatcher calling `dispatch()` directly, not through
    /// `Kernel::execute_pipeline`, is exactly why `code` stays `0` here — the
    /// exit-3 remap lives in that caller, not in `try_external` itself), and
    /// stderr carries a truncation marker. Stdout still comes back as a
    /// clean, marker-free tail — the marker is never prepended into stdout
    /// (which may be binary), only into stderr. This test still pins the
    /// piece item 3 alone is responsible for: a pipeline stage is no longer
    /// special-cased into a no-cap-of-any-kind fast path.
    #[tokio::test]
    async fn oversized_pipeline_stage_output_is_no_longer_forwarded_losslessly() {
        let (dispatcher, mut ctx, _dir) = real_cwd_dispatcher();

        ctx.output_limit = crate::output_limit::OutputLimitConfig::agent().in_memory();
        ctx.output_limit.set_limit(Some(1024)); // tiny vs. the >10MB payload below

        let (writer, reader) = crate::scheduler::pipe_stream_default();
        ctx.pipe_stdout = Some(writer);

        // Drain the pipe concurrently — a full-fidelity writer would
        // otherwise block on the 64KB pipe capacity well before finishing an
        // 11MB write, deadlocking the test.
        let drain = tokio::spawn(async move {
            use tokio::io::AsyncReadExt;
            let mut reader = reader;
            let mut buf = Vec::new();
            let _ = reader.read_to_end(&mut buf).await;
            buf
        });

        let cmd = sh_cmd("yes x | head -c 11000000");
        // A generous but bounded timeout: a real hang here should fail loud
        // and fast in CI, not stall the suite indefinitely.
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(15),
            dispatcher.dispatch(&cmd, &mut ctx),
        )
        .await
        .expect("dispatch timed out")
        .expect("dispatch");

        // Drop the writer (try_external no longer touches it post-fix, so
        // nothing else will) so the reader sees EOF and `drain` completes.
        drop(ctx.pipe_stdout.take());
        let _ = drain.await;

        // The exit-3 remap lives in `Kernel::execute_pipeline` (`if
        // result.did_spill { code = 3 }`), which this test never calls —
        // it drives `dispatcher.dispatch()` directly. So `code` stays the
        // child's own exit status (0) even though `did_spill` is now `true`.
        assert_eq!(result.code, 0, "err: {}", result.err);
        assert!(
            result.text_out().len() < 11_000_000,
            "an oversized (~11MB) pipeline stage's output must now be capped, \
             not forwarded byte-for-byte losslessly — the pre-fix special \
             case ignored any cap entirely; post-fix it goes through the same \
             capped capture (the unconditional ~10MB ring) a non-pipeline \
             external uses. got {} bytes",
            result.text_out().len()
        );
        assert!(
            !result.text_out().contains("truncated"),
            "the loud-overflow marker (GH #191) must never contaminate stdout \
             — it belongs in stderr only, since stdout may be binary: got {:?}",
            &result.text_out()[..result.text_out().len().min(80)]
        );
        assert!(
            result.did_spill,
            "the fixed ~10MB ring overflowing must set did_spill (GH #191) so \
             a real `Kernel::execute_pipeline` caller remaps to exit 3 — this \
             is independent of ctx.output_limit's own spill_if_needed, which \
             stays out of scope for try_external as before"
        );
        assert!(
            result.err.contains("stdout truncated"),
            "stderr must carry the loud overflow marker (GH #191): {}",
            result.err
        );
    }

    /// GH #133 item 4: production always puts the spawned child in its own
    /// process group (`setpgid(0,0)` in `pre_exec`) so a cancel's `killpg`
    /// reaches the whole tree — the direct child AND any grandchildren it
    /// spawns. Pre-fix, this dispatcher never called `setpgid`, so `killpg`
    /// targeted a process group nobody was actually in (an ESRCH no-op): a
    /// grandchild survived cancellation even though the direct child died.
    /// Any existing test asserting "grandchild cleanup" against this
    /// dispatcher was passing trivially, verifying nothing real.
    ///
    /// # Why this test checks the structural fact, not an end-to-end kill
    ///
    /// The most faithful reproduction of the issue would background a
    /// grandchild (`sleep N &`), cancel mid-flight, and assert the
    /// grandchild dies too — pinning the exact "existing test passes
    /// trivially" symptom. That reproduction turned out to be **blocked by a
    /// separate, pre-existing ordering issue** in this dispatcher, not
    /// introduced by this PR: `try_external`'s output collection used to run
    /// to completion BEFORE `wait_or_kill` was even called, so cancellation
    /// had no observable effect until the child's stdout closed on its own —
    /// which, for a `sh -c '... & wait'` script producing no stdout, only
    /// happened once the whole script finished naturally. GH #133 item 2 (PR
    /// #152, already landed on main alongside this fix) restructured
    /// collection to run *concurrently* with `wait_or_kill`, matching
    /// production — an end-to-end grandchild-kill test is now meaningful and
    /// fast, and remains a natural follow-up. Until then, this test pins the
    /// concrete, fast, unconfounded consequence of *this* PR's diff: the
    /// spawned child's own pgid equals its own pid, i.e. `setpgid(0, 0)` in
    /// `pre_exec` actually took effect. `ps -p $$` runs and exits almost
    /// immediately, producing no stdout for kaish to block draining — so the
    /// ordering issue above never enters into it either way.
    #[cfg(unix)]
    #[tokio::test]
    async fn spawned_child_becomes_its_own_process_group_leader() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let out_file = tmp.path().join("pgid_info");

        let (dispatcher, mut ctx, _dir) = real_cwd_dispatcher();

        // `$$` is the running shell's own PID; `ps -o pid=,pgid= -p $$`
        // reports that shell's pid and process-group id. If setpgid(0,0)
        // took effect in pre_exec (before `ps` even execs), the two must be
        // equal. Redirected straight to a file — sh's own captured stdout
        // (what kaish pipes) stays empty, so collection returns immediately.
        let script = format!("ps -o pid=,pgid= -p $$ > {}", out_file.display());
        let cmd = sh_cmd(&script);

        let result = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            dispatcher.dispatch(&cmd, &mut ctx),
        )
        .await
        .expect("dispatch timed out")
        .expect("dispatch");
        assert_eq!(result.code, 0, "err: {}", result.err);

        let contents = std::fs::read_to_string(&out_file).expect("read pgid info");
        let mut fields = contents.split_whitespace();
        let pid: i32 = fields.next().expect("pid field").parse().expect("pid parse");
        let pgid: i32 = fields.next().expect("pgid field").parse().expect("pgid parse");

        assert_eq!(
            pid, pgid,
            "the spawned child's pgid must equal its own pid — setpgid(0,0) \
             in pre_exec should make it its own process-group leader (so a \
             later killpg reaches it and any of its own children), matching \
             production (kernel.rs::try_execute_external); got pid={pid} \
             pgid={pgid}"
        );
    }

    /// GH #229: `try_external`'s path-with-slash branch checked only
    /// `resolved.exists()` before spawning, diverging from production
    /// (`kernel.rs::try_execute_external`), which additionally checks
    /// `is_file()` (exit 126 "Is a directory") and the Unix executable bit
    /// (exit 126 "Permission denied"). Spawning a directory through this
    /// test-only dispatcher used to fall through to `Command::spawn()`,
    /// which fails with a raw OS error (mapped to exit 127 here, "{name}:
    /// {e}") instead of the clean, documented exit-126 class production
    /// gives every `kernel.execute()` test.
    #[tokio::test]
    async fn path_with_slash_to_a_directory_is_126_not_a_leaked_os_error() {
        let (dispatcher, mut ctx, dir) = real_cwd_dispatcher();
        std::fs::create_dir(dir.path().join("adir")).expect("mkdir");

        let cmd = Command { name: "./adir".to_string(), args: vec![], redirects: vec![] };
        let result = dispatcher.dispatch(&cmd, &mut ctx).await.expect("dispatch");

        assert_eq!(
            result.code, 126,
            "spawning a directory must report the clean 'Is a directory' class \
             (matching kernel.rs::try_execute_external), not leak whatever raw \
             OS spawn error Command::spawn() happens to produce: {:?}",
            result
        );
        assert!(
            result.err.contains("Is a directory"),
            "err should name the reason: {}",
            result.err
        );
    }

    /// GH #229 companion: a resolved-but-non-executable regular file must
    /// report exit 126 "Permission denied", matching production's Unix mode
    /// check. Pre-fix, this dispatcher had no mode check at all and fell
    /// through to `Command::spawn()`, leaking whatever raw OS error resulted
    /// instead of the clean exit-126 class. The mode check reads the file's
    /// own permission bits directly (not an effective-permission check via
    /// the OS), so this is deterministic even when the test runs as root.
    #[cfg(unix)]
    #[tokio::test]
    async fn path_with_slash_to_a_non_executable_file_is_126_not_a_leaked_os_error() {
        use std::os::unix::fs::PermissionsExt;

        let (dispatcher, mut ctx, dir) = real_cwd_dispatcher();
        let file_path = dir.path().join("not_executable");
        std::fs::write(&file_path, b"#!/bin/sh\necho hi\n").expect("write file");
        let mut perms = std::fs::metadata(&file_path).expect("metadata").permissions();
        perms.set_mode(0o644); // no exec bits, regardless of effective uid
        std::fs::set_permissions(&file_path, perms).expect("chmod");

        let cmd = Command { name: "./not_executable".to_string(), args: vec![], redirects: vec![] };
        let result = dispatcher.dispatch(&cmd, &mut ctx).await.expect("dispatch");

        assert_eq!(
            result.code, 126,
            "a non-executable file must report the clean 'Permission denied' \
             class (matching kernel.rs::try_execute_external), not leak a raw \
             OS spawn error: {:?}",
            result
        );
        assert!(
            result.err.contains("Permission denied"),
            "err should name the reason: {}",
            result.err
        );
    }
}