car_server_core/coder/

shell_tool.rs

//! `WorktreeExecutor` — the coder's host-side tool executor.
//!
//! Wraps `car_engine::agent_basics` file tools plus a new host `shell` tool,
//! with three hard guarantees enforced in code (not just policy):
//!
//! 1. **Pinned cwd** — shell commands always run at the worktree root; there
//!    is no cwd parameter. Relative file-tool paths are rooted there too, and
//!    clamped against lexical escape.
//! 2. **Bounded output** — combined output is capped (tail-kept) so a noisy
//!    build can't flood the conversation or the event stream.
//! 3. **Bounded time** — wall-clock timeout per command; on expiry the whole
//!    process group is killed (Unix), not just the shell.
//!
//! Every call is checked by the coder [`InspectorChain`] first; first Deny
//! wins and the denial reason is the tool error the model sees.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use car_engine::{agent_basics, CommandOutput, LocalSubstrate, Substrate, ToolExecutor};
use car_policy::InspectorChain;
use serde_json::{json, Value};

use super::policy::{
    coder_inspector_chain, coder_inspector_chain_with_project_policies, stays_under,
};

/// Default and ceiling for per-command wall-clock timeouts.
pub(crate) const DEFAULT_SHELL_TIMEOUT_SECS: u64 = 120;
pub(crate) const MAX_SHELL_TIMEOUT_SECS: u64 = 600;

/// Description of the coder `run_shell` tool's `command` parameter. Same reason
/// as the assistant's: the local leg of [`run_shell_on`] is `cmd /C` on Windows,
/// and telling the model otherwise makes it author POSIX the shell cannot run.
#[cfg(windows)]
const SHELL_COMMAND_PARAM_DESC: &str = concat!(
    "Command executed via `cmd /C` at the repository root on this Windows host ",
    "— cmd.exe, not a POSIX shell (no ls/grep/cat/tail/rm, no $(...))."
);
#[cfg(not(windows))]
const SHELL_COMMAND_PARAM_DESC: &str = "Command executed via sh -c at the repository root";
/// Combined stdout+stderr cap (tail kept).
pub(crate) const MAX_OUTPUT_BYTES: usize = 64 * 1024;

/// Keep the last `cap` bytes of `s`, on a char boundary, with a marker when
/// truncated.
pub(crate) fn tail(s: &str, cap: usize) -> String {
    if s.len() <= cap {
        return s.to_string();
    }
    let mut start = s.len() - cap;
    while !s.is_char_boundary(start) {
        start += 1;
    }
    format!("…[truncated]…{}", &s[start..])
}

/// Fold a `{stdout, stderr, exit_code}` triple into the coder/assistant shell
/// result shape `{exit_code, output, timed_out}`, with stderr appended after
/// stdout and the combined text tail-capped.
fn shell_result(stdout: &str, stderr: &str, exit_code: i32) -> Value {
    let mut combined = stdout.to_string();
    if !stderr.is_empty() {
        if !combined.is_empty() && !combined.ends_with('\n') {
            combined.push('\n');
        }
        combined.push_str(stderr);
    }
    json!({
        "exit_code": exit_code,
        "output": tail(&combined, MAX_OUTPUT_BYTES),
        "timed_out": false,
    })
}

/// Root relative `path` params at `root` and reject write escapes outside it —
/// the one shared implementation behind the coder's `WorktreeExecutor` and the
/// assistant's `GeneralExecutor` (which layers its own `clamp` opt-out on top).
/// `root_noun` names the boundary in the escape error ("worktree" vs "working
/// directory"). Reads may roam for context gathering; only `write_file`/
/// `edit_file` are pinned inside `root`, and relative paths are always
/// resolved against `root` (never the daemon cwd).
/// `clamp_reads` additionally pins the READ tools (`read_file`, `list_dir`,
/// `find_files`, `grep_files`) inside `root`. Off by default because the
/// general assistant is legitimately allowed to read the wider filesystem; the
/// `coder.discuss` surface turns it on, since a conversation whose whole
/// premise is "grounded in THIS repo" has no business reading outside it, and
/// leaving reads open there is an exfiltration path — a prompt-injected repo
/// file can ask for `grep_files {"path":"/Users/<user>","pattern":"sk-ant-"}`
/// and the hits stream to every `coder.discuss.event` subscriber.
pub(crate) fn clamp_paths_to(
    root: &std::path::Path,
    tool: &str,
    params: &Value,
    root_noun: &str,
    clamp_reads: bool,
) -> Result<Value, String> {
    let mut params = params.clone();
    let Some(obj) = params.as_object_mut() else {
        return Ok(params);
    };
    if let Some(Value::String(p)) = obj.get("path") {
        let pinned = matches!(tool, "write_file" | "edit_file")
            || (clamp_reads
                && matches!(tool, "read_file" | "list_dir" | "find_files" | "grep_files"));
        if !stays_under(root, p) && pinned {
            return Err(format!("path '{p}' resolves outside the {root_noun}"));
        }
        if Path::new(p).is_relative() {
            let abs = root.join(p);
            obj.insert("path".into(), json!(abs.to_string_lossy()));
        }
    } else if matches!(tool, "list_dir" | "find_files" | "grep_files") {
        obj.entry("path")
            .or_insert_with(|| json!(root.to_string_lossy()));
    }
    Ok(params)
}

/// Single-quote a string for POSIX `sh`: wrap in `'…'` and escape embedded
/// quotes as `'\''`. A PATH can contain spaces and (rarely) quotes.
fn sh_single_quote(s: &str) -> String {
    format!("'{}'", s.replace('\'', r"'\''"))
}

/// Remove every way a child shell could authenticate to a forge.
///
/// Three separate mechanisms, because closing one leaves the others open:
///
/// 1. **`gh`'s environment tokens.** `GH_TOKEN` / `GITHUB_TOKEN` and their
///    enterprise spellings are read before any config file.
/// 2. **`gh`'s config file.** With the env tokens gone, `gh` falls back to
///    `~/.config/gh/hosts.yml`, so `GH_CONFIG_DIR` is pointed at an empty
///    directory. The child can write into that directory, which does not help:
///    a `hosts.yml` still needs a token it no longer has.
/// 3. **git's credential helper.** This one is easy to miss and would have left
///    the hole wide open — `git push` over HTTPS does not read `GH_TOKEN` at
///    all, it asks the configured helper (`osxkeychain` on a Mac), which is
///    perfectly happy to hand over a stored credential. `GIT_CONFIG_COUNT` and
///    friends override `credential.helper` to empty for this child only, which
///    resets the helper list without touching the operator's `~/.gitconfig`.
///
/// `GIT_TERMINAL_PROMPT=0` so a de-credentialed push fails immediately instead
/// of blocking on a prompt no one can answer.
///
/// [`super::merge`] is host code: it builds its own `gh`/`git` argv outside this
/// function and keeps the real credential. That asymmetry is the whole design —
/// the runtime publishes, the model cannot.
fn withhold_forge_credentials(cmd: &mut tokio::process::Command) {
    for var in [
        "GH_TOKEN",
        "GITHUB_TOKEN",
        "GH_ENTERPRISE_TOKEN",
        "GITHUB_ENTERPRISE_TOKEN",
    ] {
        cmd.env_remove(var);
    }
    cmd.env("GH_CONFIG_DIR", empty_config_dir());
    cmd.env("GIT_CONFIG_COUNT", "1")
        .env("GIT_CONFIG_KEY_0", "credential.helper")
        .env("GIT_CONFIG_VALUE_0", "")
        .env("GIT_TERMINAL_PROMPT", "0")
        .env("GIT_ASKPASS", "")
        .env("SSH_ASKPASS", "")
        .env("SSH_ASKPASS_REQUIRE", "never");
}

/// Keep the runtime's own verification from writing files into the diff it is
/// about to deliver.
///
/// Running a check is not supposed to change the change. Python writes
/// `__pycache__/*.pyc` beside every module it imports, so a contract check of
/// `python3 -m pytest` generated bytecode in the worktree — and `stage_and_diff`
/// stages with `git add -A`, so those files went into the delivered commit.
///
/// Found by a live review panel, on the first run where real models saw a real
/// diff: all three approved the code change and refused the pull request over
/// the committed bytecode. They were right, and the contract could not have
/// caught it — the tests passed either way. A repository with a `.gitignore`
/// hides this; one without it silently receives build artifacts from every
/// session.
///
/// `PYTHONDONTWRITEBYTECODE` is the whole fix for Python: bytecode caching is a
/// warm-start optimisation and a one-shot check has no warm start to lose.
fn suppress_incidental_artifacts(cmd: &mut tokio::process::Command) {
    cmd.env("PYTHONDONTWRITEBYTECODE", "1");
}

/// A directory that exists and holds no forge configuration.
///
/// Under the CAR state root rather than a fresh temp dir per call: this is read
/// on every shell invocation, and a per-call temp directory would be both waste
/// and litter. Falls back to the OS temp dir if the state root cannot be
/// created — an unwritable path would make `gh` fall back to the real config,
/// which is the one outcome to avoid.
fn empty_config_dir() -> std::path::PathBuf {
    let dir = car_home::root_or_relative()
        .join("run")
        .join("no-forge-config");
    if std::fs::create_dir_all(&dir).is_ok() {
        return dir;
    }
    let fallback = std::env::temp_dir().join("car-no-forge-config");
    let _ = std::fs::create_dir_all(&fallback);
    fallback
}

/// Prefix `command` with an `export PATH=<inherited>:$PATH` so the PATH this
/// process was started with survives a login shell's profile.
///
/// macOS `path_helper` reorders PATH (see the call site); re-exporting ours in
/// front restores the operator's precedence while leaving the profile's own
/// additions on the tail. Returns `command` unchanged when PATH is unset/empty,
/// and on Linux this is a harmless no-op reassertion of the same value.
fn prepend_inherited_path(command: &str) -> String {
    match std::env::var("PATH") {
        Ok(p) if !p.trim().is_empty() => {
            format!("export PATH={}:\"$PATH\"; {}", sh_single_quote(&p), command)
        }
        _ => command.to_string(),
    }
}

/// Run `command` against `substrate`, inspector-gated, bounded in time and
/// output — the one shared shell implementation behind both the coder's
/// `WorktreeExecutor` and the assistant's `GeneralExecutor`.
///
/// Returns `{exit_code, output, timed_out}`; a non-zero exit is a value, not an
/// error, so the model can read it. `cwd` pins the working directory on the
/// **local** path (ignored by non-local substrates, which carry their own root
/// — e.g. the Docker sandbox mount or the VM bridge). On the local path a
/// timeout kills the whole process group (Unix); non-local substrates enforce
/// their own command timeout via [`Substrate::run_command`].
///
/// `max_timeout_secs` is the ceiling `timeout_secs` is clamped to. Every
/// MODEL-facing caller passes [`MAX_SHELL_TIMEOUT_SECS`], which is what the
/// advertised tool description promises; the outcome-contract path passes the
/// operator's own ceiling instead, because a slow test gate is the operator's
/// decision about their repository, not a licence for the model to run one
/// command for an hour (car#1065).
/// Whether a shell child inherits the daemon's forge credentials.
///
/// The deny-list in [`super::policy`] is hardening, not a sandbox: it reads the
/// verb of each segment and then hands that segment to `/bin/sh`, so `sh -c`,
/// `env`, `timeout`, `$(…)`, a `\gh` escape, or a delegated `car do "…push it"`
/// all move or bypass the verb. No finite set of patterns enforces an any-route
/// property against an unrestricted shell (car#1076 documents the residue).
///
/// Credential separation does not have that shape. A shell holding no forge
/// credential cannot publish however the command is spelled, because every
/// route fails on **authentication** rather than on being recognised.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ForgeCredentials {
    /// The child keeps the forge credential. For contract checks, whose command
    /// the CONTRACT declares rather than the model.
    ///
    /// The name is exact: the difference between the two variants is the FORGE
    /// credential, not the environment. [`Withhold`](Self::Withhold) removes
    /// four `GH_*`/`GITHUB_*` token variables and neutralizes the gh/git/ssh
    /// credential helpers; it does not clear anything else, so `$DATABASE_URL`
    /// and friends reach both paths alike.
    ///
    /// Keeping the credential is also not permission to use one: a check runs
    /// the SAME inspector chain as the model's shell, so `DenyCredentialAccess`
    /// refuses the command on its text. That is substring hardening rather than
    /// a boundary — an unmarked spelling still gets through to the inherited
    /// environment. Both halves are stated beside the contract input in
    /// `docs/car-code-task.md` (car#1066).
    Inherit,
    /// The child gets none. For the model's own shell.
    Withhold,
}

#[cfg(unix)]
struct ProcessGroupGuard {
    pgid: i32,
    armed: bool,
}

#[cfg(unix)]
impl ProcessGroupGuard {
    fn new(pgid: u32) -> Self {
        Self {
            pgid: pgid as i32,
            armed: true,
        }
    }

    /// Sweep every process the shell left in its private group.
    ///
    /// The direct shell is reaped by `wait_with_output`; detached grandchildren
    /// are not. They keep the shell's group after being reparented, so `killpg`
    /// remains the one handle that covers both the ordinary return path and a
    /// partially-unwound command.
    fn terminate(&mut self) {
        if self.armed {
            unsafe {
                libc::killpg(self.pgid, libc::SIGKILL);
            }
            self.armed = false;
        }
    }
}

#[cfg(unix)]
impl Drop for ProcessGroupGuard {
    fn drop(&mut self) {
        // A future can be dropped by Ctrl-C/SIGTERM cancellation or unwinding
        // before `run_shell_on` reaches its ordinary cleanup. `kill_on_drop`
        // only owns the direct shell, so the group guard is the descendant
        // backstop for those paths.
        self.terminate();
    }
}

pub(crate) async fn run_shell_on(
    substrate: &Arc<dyn Substrate>,
    cwd: Option<&Path>,
    inspectors: &InspectorChain,
    command: &str,
    timeout_secs: Option<u64>,
    max_timeout_secs: u64,
    forge_credentials: ForgeCredentials,
) -> Result<Value, String> {
    if let Some(reason) = inspectors.check("shell", &json!({ "command": command })) {
        return Err(format!("denied by policy: {reason}"));
    }
    let secs = timeout_secs
        .unwrap_or(DEFAULT_SHELL_TIMEOUT_SECS)
        .clamp(1, max_timeout_secs.max(1));

    // Non-local substrates (Docker sandbox, VM-over-MCP) own their own
    // isolation and cwd — route straight through their `run_command`, which
    // enforces the timeout itself.
    if !substrate.is_local() {
        let CommandOutput {
            stdout,
            stderr,
            exit_code,
        } = substrate.run_command(command, Some(secs as f64)).await?;
        return Ok(shell_result(&stdout, &stderr, exit_code));
    }

    // Local path: `sh -lc` (Unix) / `cmd /C` (Windows) in a fresh process group
    // so a timeout can sweep the whole tree, not just the shell.
    let timeout = Duration::from_secs(secs);
    let mut cmd = if cfg!(target_os = "windows") {
        let mut c = tokio::process::Command::new("cmd");
        c.arg("/C").arg(command);
        // cmd.exe silently DROPS any env var over ~8191 chars. When that var is
        // PATH the model's shell loses its entire toolchain — `cargo`/`git`/`npm`
        // come back "is not recognized" — and the coder misreads the red checks as
        // its own broken code. See car_engine::win_env; `None` = inherit unchanged.
        if let Some(path) = car_engine::win_env::cmd_path_override() {
            c.env("PATH", path);
        }
        c
    } else {
        let mut c = tokio::process::Command::new("/bin/sh");
        // `-l` loads the user's profile so the agent inherits their toolchain
        // (nvm, rbenv, pyenv…). But on macOS `/etc/profile` runs `path_helper`,
        // which REBUILDS PATH with the system dirs first and merely appends
        // whatever this process inherited — silently demoting a PATH the operator
        // set *for the daemon* below `/usr/local/bin`. A stale system binary then
        // shadows the intended one, and the coder misreads the resulting red
        // check as its own broken code.
        //
        // This is the macOS twin of the Windows bug `car_engine::win_env` fixes
        // (cmd drops an over-long PATH, emptying the agent's shell). Surfaced by
        // the coder A/B: a daemon started with a venv first on PATH still had the
        // venv at position 16 inside the agent's shell, so `pip` resolved to
        // `/usr/local/bin/pip`, whose `#!/usr/bin/python` shebang no longer
        // exists on modern macOS. Every `pip` step in a derived contract then
        // failed forever — sinking sessions whose real work had already passed.
        //
        // Re-assert the inherited PATH *after* the profile has loaded, so the
        // operator's entries win while the profile's additions remain reachable.
        c.arg("-lc").arg(prepend_inherited_path(command));
        c
    };
    // UNCONDITIONAL, and deliberately not folded into
    // `withhold_forge_credentials`: that one is gated on the caller's credential
    // policy, and whether the runtime's own checks litter the worktree has
    // nothing to do with whether the model may reach a forge token. Hanging it
    // off that gate left the litter in place on every path that keeps
    // credentials — which is how the first attempt at this fix silently did
    // nothing.
    suppress_incidental_artifacts(&mut cmd);
    if forge_credentials == ForgeCredentials::Withhold {
        withhold_forge_credentials(&mut cmd);
    }
    cmd.stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true);
    if let Some(dir) = cwd {
        cmd.current_dir(dir);
    }
    #[cfg(unix)]
    cmd.process_group(0);

    let child = cmd
        .spawn()
        .map_err(|e| format!("failed to spawn shell: {e}"))?;
    #[cfg(unix)]
    let mut process_group = child.id().map(ProcessGroupGuard::new);

    // Windows has no process groups; assign the shell to a Job Object so a
    // timeout can atomically kill the whole tree (cmd.exe + everything it
    // spawns), not just cmd.exe. `kill_on_drop` only reaps the direct child,
    // orphaning grandchildren of a timed-out build. Best-effort: if the job
    // can't be created/assigned we fall back to `kill_on_drop`. The
    // KILL_ON_JOB_CLOSE flag also sweeps any stragglers when the job drops at
    // the end of this call.
    #[cfg(windows)]
    let job = match car_registry::supervisor::JobObject::new() {
        Ok(j) => {
            if let Some(pid) = child.id() {
                let _ = j.assign(pid);
            }
            Some(j)
        }
        Err(_) => None,
    };

    let result = match tokio::time::timeout(timeout, child.wait_with_output()).await {
        Ok(Ok(out)) => Ok(shell_result(
            &String::from_utf8_lossy(&out.stdout),
            &String::from_utf8_lossy(&out.stderr),
            out.status.code().unwrap_or(-1),
        )),
        Ok(Err(e)) => Err(format!("shell wait failed: {e}")),
        Err(_elapsed) => {
            // Kill the whole process group / job: `sh -c "sleep 999 & wait"`
            // (Unix) or a `cmd /C` build that spawned children (Windows) must
            // not outlive the timeout. kill_on_drop has already reaped the
            // shell itself; this sweeps descendants.
            #[cfg(windows)]
            if let Some(job) = &job {
                let _ = job.terminate(1);
            }
            Ok(json!({
                "exit_code": Value::Null,
                "output": format!("command timed out after {}s and was killed", timeout.as_secs()),
                "timed_out": true,
            }))
        }
    };

    // A successful/non-zero shell can daemonize a child after closing the
    // captured pipes. `wait_with_output` then returns while that process keeps
    // running (the observed shape was `car-server --no-auth` plus its
    // supervised `car do --serve`). A shell tool invocation is scoped to this
    // call, so no outcome grants a background-process lifetime.
    #[cfg(unix)]
    if let Some(group) = &mut process_group {
        group.terminate();
    }

    result
}

/// `recall` from the graph memory, and deliberately **not** `remember`.
///
/// car#1071's ask is the read: the coder could not recall a fact anyone had
/// stored about the project, which is the product's headline capability being
/// unavailable to the flagship coding agent inside it.
///
/// The write is a different grant and is withheld on purpose. `remember` is an
/// information-flow **sink** carrying `persistent_memory`
/// (`car_engine::builtin_tool_labels`) — it writes durable state that outlives
/// the session and is recalled by every later one. Compose that with car#1081,
/// where a coder session may be triaging an issue from a **public** tracker
/// whose body is attacker-authored, and a write path becomes a persistence
/// attack: hostile text lands in durable memory once and is read back as
/// trusted context indefinitely. A prompt injection that ends with the session
/// is recoverable; one that writes to memory is not.
///
/// The coder is not left unable to learn. `super::skill_memory::RepairMemory`
/// is its own write path — failure signatures and repair skills, scoped to what
/// a coder round actually establishes, and written by the runtime rather than
/// by the model.
fn recall_only_memory_defs() -> Vec<Value> {
    crate::assistant::memory::MemoryTools::tool_defs()
        .into_iter()
        .filter(|d| d["name"] == "recall")
        .collect()
}

/// One attached delegate and the tool names it advertises.
///
/// The defs are kept beside the executor rather than in a flat list so
/// dispatch can answer "who owns this name" instead of "does anyone", which is
/// what a single shared `Vec` could tell you.
struct Delegate {
    executor: Arc<dyn ToolExecutor>,
    defs: Vec<Value>,
}

impl Delegate {
    fn tool_names(&self) -> impl Iterator<Item = String> + '_ {
        self.defs
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
    }

    fn advertises(&self, tool: &str) -> bool {
        self.defs.iter().any(|d| d["name"] == tool)
    }
}

pub struct WorktreeExecutor {
    worktree: PathBuf,
    inspectors: InspectorChain,
    /// Executors for tools this one does not own: the Parslee platform tools,
    /// graph-memory `recall` (car#1071), the network pair (car#1073), and the
    /// browser surface when the session explicitly opts in (car#1069). Names a
    /// delegate advertises route to it, bypassing the worktree path-clamp. They
    /// still pass the inspector chain, so operator-authored deny rules govern
    /// delegate calls too.
    ///
    /// A **list**, not a single slot. It was `Option<Arc<dyn ToolExecutor>>`
    /// plus one `Vec<Value>`, so a second `with_delegate` silently replaced the
    /// first rather than adding to it — which is why three separate issues each
    /// hit "attach a second delegate" as their blocker.
    delegates: Vec<Delegate>,
    /// When set, the per-agent approval policy (`agent_permissions`) is consulted
    /// before every tool. `Deny` hard-blocks. `RequireApproval` hard-blocks
    /// `full_access` calls because coder/declarative runs have no interactive
    /// approval channel; lower tiers keep running so ordinary sandbox edits stay
    /// usable under the Balanced default.
    agent_id: Option<String>,
    /// Full-access delegate tools the operator explicitly approved for this
    /// session. The browser flag populates this set: it satisfies an ordinary
    /// `RequireApproval`, but never overrides an Agent Permissions `Deny` and
    /// never skips the coder inspector chain.
    session_approved_tools: std::collections::BTreeSet<String>,
    /// Per-session read ledgers backing the read-before-edit / staleness guard
    /// on built-in file tools. A shared worktree executor must not let one run
    /// authorize another run's mutation.
    read_ledgers: agent_basics::SessionReadLedgers,
    /// Latches once this executor has changed the worktree. Read by the
    /// no-change gate, which refuses a nomination from a session that ever
    /// mutated — see [`super::no_change::MutationLedger`] for why reverting
    /// does not clear it.
    mutations: Arc<super::no_change::MutationLedger>,
    /// Tools the operator's policy files forbid outright, captured when the
    /// chain was loaded in [`Self::for_coder_session`]. Empty on [`Self::new`],
    /// which loads no project rules.
    ///
    /// The inspector chain already REFUSES these at dispatch. This set exists
    /// so the coding loop can also stop OFFERING them, which the chain cannot
    /// do — an inspector sees a call, never the menu.
    denied_tools: BTreeSet<String>,
    /// Whether delegate-owned names may actually be **dispatched**.
    ///
    /// Attachment is not reachability, and conflating the two is a live hole:
    /// dispatch keys on `delegate_defs` — what the delegate *offers* — and
    /// returns before both `clamp_paths` and the inspector chain. A run that
    /// never advertised a delegate tool could still call one by name, and
    /// `classify_tool_tier` defaults an unrecognised name to `ReadOnly`, so the
    /// per-agent gate waves it through. `parslee_generate_document` writes to
    /// the user's connected drive.
    ///
    /// Default **false**: only a run that advertised the delegate surface may
    /// reach it. Today that is the declarative agent-build path, which calls
    /// `all_tool_defs()`; the coding loop advertises the static built-ins and
    /// so reaches nothing here.
    delegates_reachable: Arc<std::sync::atomic::AtomicBool>,
    /// Ceiling for outcome-contract check commands only ([`Self::run_check_shell`]).
    /// Defaults to [`MAX_SHELL_TIMEOUT_SECS`]; an operator raises it with
    /// `max_check_timeout_secs` in `~/.car/coder.toml` or
    /// `car code-task --max-check-timeout-secs`. The model's own `shell` calls
    /// keep the advertised 600s ceiling regardless (car#1065).
    check_timeout_ceiling: u64,
}

fn enforce_agent_permission(
    agent_id: &str,
    tool: &str,
    tier: car_policy::PermissionTier,
    mode: car_policy::ApprovalMode,
) -> Result<(), String> {
    match mode {
        car_policy::ApprovalMode::AlwaysAllow => Ok(()),
        car_policy::ApprovalMode::Deny => Err(format!(
            "denied for agent '{agent_id}' by your Agent Permissions settings: \
             '{tool}' is a {}-tier action this agent may not perform",
            tier.as_str()
        )),
        car_policy::ApprovalMode::RequireApproval
            if tier == car_policy::PermissionTier::FullAccess =>
        {
            Err(format!(
                "approval required for agent '{agent_id}' by your Agent Permissions \
                 settings: '{tool}' is a {}-tier action, but this runner has no \
                 interactive approval channel",
                tier.as_str()
            ))
        }
        car_policy::ApprovalMode::RequireApproval => Ok(()),
    }
}

impl WorktreeExecutor {
    /// Executor for `worktree` with the standard coder inspector chain.
    pub fn new(worktree: impl Into<PathBuf>) -> Self {
        let worktree: PathBuf = worktree.into();
        // Canonicalize so lexical clamping isn't fooled by `/var` vs
        // `/private/var` style aliasing of the worktree root itself.
        let worktree = worktree.canonicalize().unwrap_or(worktree);
        let inspectors = coder_inspector_chain(&worktree);
        Self {
            worktree,
            inspectors,
            delegates: Vec::new(),
            agent_id: None,
            session_approved_tools: std::collections::BTreeSet::new(),
            read_ledgers: agent_basics::SessionReadLedgers::new(),
            denied_tools: BTreeSet::new(),
            mutations: Arc::new(super::no_change::MutationLedger::new()),
            delegates_reachable: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            check_timeout_ceiling: MAX_SHELL_TIMEOUT_SECS,
        }
    }

    /// Raise (or lower) the ceiling for outcome-contract check commands.
    ///
    /// Floored at 1s: a `0` reaching here from config would otherwise clamp
    /// every check to a one-second timeout and turn a whole contract red.
    pub fn with_check_timeout_ceiling(mut self, secs: u64) -> Self {
        self.check_timeout_ceiling = secs.max(1);
        self
    }

    /// The ceiling [`Self::run_check_shell`] applies. Contract evaluation reads
    /// it to derive `deadline_clamped` against the ceiling that actually ran.
    pub fn check_timeout_ceiling(&self) -> u64 {
        self.check_timeout_ceiling
    }

    /// Whether this executor has changed the worktree at any point.
    ///
    /// One-way: see [`super::no_change::MutationLedger`]. A caller adjudicating
    /// a no-change nomination reads this, and must not interpret a currently
    /// clean worktree as equivalent.
    pub fn has_mutated(&self) -> bool {
        self.mutations.has_mutated()
    }

    /// Declare that this run advertises the delegate surface, making those
    /// names dispatchable.
    ///
    /// Call it immediately beside the `all_tool_defs()` that advertises them,
    /// so the two cannot drift: what a run can call should be what it was
    /// offered. Deliberately NOT folded into `all_tool_defs()` itself — a getter
    /// that widens a security boundary as a side effect is worse than one
    /// explicit line.
    pub fn advertise_delegates(&self) {
        self.delegates_reachable
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    /// Whether delegate names are dispatchable on this executor.
    pub fn delegates_reachable(&self) -> bool {
        self.delegates_reachable
            .load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Whether a `full_access`-tier tool would actually dispatch here, per the
    /// per-agent approval policy.
    ///
    /// A loop asks this before OFFERING one. A tool the gate will refuse is a
    /// worse failure than a tool that was never offered: the model spends turns
    /// on it and reads the refusal as the task being impossible rather than as a
    /// permission nobody granted. With no `agent_id` there is no per-agent gate
    /// at all, so the answer is yes. Otherwise it mirrors exactly what
    /// [`enforce_agent_permission`] lets through at that tier — `AlwaysAllow`
    /// and nothing else, since `RequireApproval` hard-blocks a runner that has
    /// no interactive approval channel.
    pub fn permits_full_access(&self) -> bool {
        let Some(agent_id) = &self.agent_id else {
            return true;
        };
        matches!(
            crate::agent_permissions::resolve(agent_id, car_policy::PermissionTier::FullAccess),
            car_policy::ApprovalMode::AlwaysAllow
        )
    }

    /// Executor for a coder session's `worktree`, configured identically for
    /// every entry point that starts one — the daemon's `coder.*` work loop and
    /// the headless `car code-task`. Both call this so the two cannot drift
    /// apart again (Parslee-ai/car#1063).
    ///
    /// The Parslee delegate is attached for the **agent-build** project kind:
    /// [`super::declarative::DeclarativeAgentRunner`] is the only caller of
    /// [`Self::all_tool_defs`], so those names reach a model only when a
    /// generated agent's spec allowlists them. A plain coding session runs
    /// [`super::native_loop::run_native_loop`], which advertises the static
    /// built-ins ([`Self::tool_defs`]) plus the delegate tools it names one by
    /// one — so the model-visible tool list is the same whichever entry point
    /// launched the session.
    ///
    /// Browser tools are attached separately by [`Self::with_browser_tools`]
    /// only after a session's explicit `browser` option is read. Keeping the
    /// default constructor browser-free makes omission a real absence rather
    /// than a prompt-only convention.
    pub fn for_coder_session(worktree: impl Into<PathBuf>) -> Result<Self, String> {
        let base = Self::new(worktree);
        let policy = coder_inspector_chain_with_project_policies(&base.worktree).map_err(|e| {
            format!(
                "refusing to start coder session with unreadable operator policy rules: {e}. \
                 Fix or remove the file — a deny rule that fails to load is a security \
                 control that would silently not exist"
            )
        })?;
        let denied_tools = policy.denied_tools;
        Ok(base
            .with_denied_tools(denied_tools)
            .with_chain(policy.chain)
            .with_delegate(
                Arc::new(crate::parslee_tools::ParsleeToolExecutor),
                crate::parslee_tools::ParsleeToolExecutor::tool_defs(),
            )
            .with_delegate(
                Arc::new(crate::assistant::memory::MemoryTools::open(
                    crate::assistant::default_memory_path(),
                )),
                recall_only_memory_defs(),
            )
            // The assistant's network pair — `http_request` and `web_search`
            // (Parslee-ai/car#1073). Attached rather than withheld, because
            // withholding buys no containment: the coder's `shell` can already
            // run `curl` with nothing inspecting it, as `coder::policy`'s own
            // note on the forge matcher records. What the coder lacked was never
            // egress, it was GOVERNED egress — these route through the inspector
            // chain, so an operator's `deny_tool` rule can refuse one by name,
            // and through the event log, so the call leaves a record that
            // `sh -c curl` does not.
            //
            // Default-closed all the same. Both defs declare `tier:
            // full_access`, which the Balanced default resolves to
            // `RequireApproval`, and `enforce_agent_permission` hard-blocks that
            // tier for a runner with no approval channel. Giving a coding agent
            // the network stays a decision an operator makes on the Agent
            // Permissions screen; this attachment does not make it for them.
            .with_delegate(
                Arc::new(crate::assistant::net_tools::NetTools::new()),
                crate::assistant::net_tools::net_tool_defs(),
            )
            // The coder agent runs under the stable `car-coder` policy subject, so an
            // operator can Deny it at a risk tier from the Agent Permissions screen.
            .with_agent_permissions("car-coder"))
    }

    /// Attach the assistant's browser integration for an explicitly opted-in
    /// coder session. Its fresh profile is deleted with the browser; cookies
    /// and sign-ins do not persist across coder sessions. Chromium still
    /// launches lazily on the first call. Every
    /// browser name remains a delegate, so dispatch passes the same per-agent
    /// permission check and frozen coder inspector chain as file and shell
    /// tools. The session opt-in satisfies `RequireApproval` for these exact
    /// names; an explicit `Deny` still wins.
    pub fn with_browser_tools(mut self) -> Self {
        let browser = Arc::new(crate::assistant::browser_tools::BrowserTools::isolated(
            self.worktree.clone(),
        ));
        let defs = browser.tool_defs();
        self.session_approved_tools.extend(
            defs.iter()
                .filter_map(|def| def["name"].as_str().map(String::from)),
        );
        self.with_delegate(browser, defs)
    }

    /// Enforce the per-agent approval policy for `agent_id`. Used by declarative
    /// agents (their `spec.id`) and the coder session, extending per-agent
    /// guardrails beyond the assistant loop.
    pub fn with_agent_permissions(mut self, agent_id: impl Into<String>) -> Self {
        self.agent_id = Some(agent_id.into());
        self
    }

    /// Replace the inspector chain (tests; callers wanting extra rules).
    pub fn with_chain(mut self, chain: InspectorChain) -> Self {
        self.inspectors = chain;
        self
    }

    /// Record the tools operator policy forbids outright — see the field.
    pub fn with_denied_tools(mut self, denied: BTreeSet<String>) -> Self {
        self.denied_tools = denied;
        self
    }

    /// Tools this session must not offer the model. Narrowing only: a name here
    /// is refused at dispatch whether or not the caller consults this.
    pub fn denied_tools(&self) -> &BTreeSet<String> {
        &self.denied_tools
    }

    /// Attach a delegate executor that handles the given tool `defs` (by name).
    /// Used to expose the Parslee platform tools to declarative agents without
    /// threading them through the worktree's file-tool path logic.
    ///
    /// **What a delegate gives up.** Delegate-owned names bypass the worktree
    /// path-clamp because they execute on another substrate, but still pass the
    /// coder [`InspectorChain`] so declarative deny rules apply. A delegate
    /// must still be side-effect-free or independently gated where repository
    /// path scoping cannot apply. The Parslee surface qualifies
    /// because it carries its own auth + entitlement gating (`parslee_*` refuses
    /// unless the account is signed in, has an active org, and holds the
    /// required entitlement); the per-agent approval check still runs first,
    /// since it precedes the delegate dispatch in `execute_in_session`. Do not
    /// attach a delegate that writes to the host on the strength of the caller's
    /// word.
    /// Attach a delegate. **Additive** — call it once per delegate.
    ///
    /// Name collisions resolve to the delegate attached FIRST, and that is a
    /// deliberate choice rather than an accident of iteration order: attachment
    /// order is written at the call site where a reader can see it, whereas
    /// last-wins would let a delegate added later silently capture a name an
    /// earlier one owns. [`Self::delegate_name_collisions`] reports any overlap
    /// so a test can refuse it outright.
    pub fn with_delegate(mut self, delegate: Arc<dyn ToolExecutor>, defs: Vec<Value>) -> Self {
        self.delegates.push(Delegate {
            executor: delegate,
            defs,
        });
        self
    }

    /// Attached delegate defs whose tool name is `name`.
    ///
    /// Lets a loop advertise a specific delegate tool without advertising the
    /// whole delegate surface — the coding loop wants graph-memory `recall`
    /// while leaving the Parslee document tools unoffered, and
    /// `all_tool_defs()` cannot express that.
    pub fn delegate_defs_named(&self, name: &str) -> Vec<Value> {
        self.delegates
            .iter()
            .flat_map(|d| d.defs.iter())
            .filter(|d| d["name"] == name)
            .cloned()
            .collect()
    }

    /// Attached delegate definitions whose names begin with `prefix`.
    pub fn delegate_defs_with_prefix(&self, prefix: &str) -> Vec<Value> {
        self.delegates
            .iter()
            .flat_map(|d| d.defs.iter())
            .filter(|d| {
                d["name"]
                    .as_str()
                    .is_some_and(|name| name.starts_with(prefix))
            })
            .cloned()
            .collect()
    }

    /// Tool names advertised by more than one delegate.
    ///
    /// Empty is the only healthy answer. Exposed so a call site that composes
    /// several delegates can assert it rather than discovering a shadowed tool
    /// at runtime.
    pub fn delegate_name_collisions(&self) -> Vec<String> {
        let mut seen: std::collections::BTreeMap<String, usize> = std::collections::BTreeMap::new();
        for delegate in &self.delegates {
            for name in delegate.tool_names() {
                *seen.entry(name).or_default() += 1;
            }
        }
        seen.into_iter()
            .filter(|(_, n)| *n > 1)
            .map(|(name, _)| name)
            .collect()
    }

    /// The delegate that owns `tool`, if any is attached and advertises it.
    fn delegate_for(&self, tool: &str) -> Option<&Delegate> {
        self.delegates.iter().find(|d| d.advertises(tool))
    }

    /// Every attached delegate's defs, flattened.
    ///
    /// Used for tier classification, which needs each tool's declared tier and
    /// does not care which delegate declared it. Dispatch deliberately does NOT
    /// go through this — it needs the owner, not the union.
    fn all_delegate_defs(&self) -> Vec<Value> {
        self.delegates
            .iter()
            .flat_map(|d| d.defs.iter().cloned())
            .collect()
    }

    /// All tool defs this executor exposes: the static built-ins plus any
    /// delegate tools. Agent loops should advertise these (not the static
    /// [`Self::tool_defs`]) so delegate tools are allowlistable.
    pub fn all_tool_defs(&self) -> Vec<Value> {
        let mut defs = Self::tool_defs();
        for delegate in &self.delegates {
            defs.extend(delegate.defs.iter().cloned());
        }
        defs
    }

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

    /// Tool definitions to expose to the model: the built-in file tools plus
    /// the coder's `shell` tool, in the `{name, description, parameters}`
    /// shape `GenerateRequest.tools` expects.
    pub fn tool_defs() -> Vec<Value> {
        let mut defs: Vec<Value> = agent_basics::entries()
            .iter()
            .map(|e| {
                json!({
                    "name": e.schema.name,
                    "description": e.schema.description,
                    "parameters": e.schema.parameters,
                })
            })
            .filter(|d| d["name"] != "calculate") // not useful for coding
            .collect();
        defs.push(json!({
            "name": "shell",
            "description": "Run a shell command at the repository root (the worktree). \
                            Use for builds, tests, and anything the file tools can't do. \
                            Output is the combined stdout+stderr tail. Publishing and \
                            privilege-escalating commands are denied by policy: `git \
                            push`, `gh`/`glab` writes (`pr create`, `release create`, \
                            non-GET `api`), `npm`/`cargo publish`, `docker push`, \
                            `sudo`, and destructive operations outside the repo. \
                            Reading the forge is allowed (`gh pr view`, `gh run view`, \
                            `gh api` GET) so you can watch CI. Do not try to route \
                            around these — the runtime opens the pull request itself \
                            after it has verified your work.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": SHELL_COMMAND_PARAM_DESC
                    },
                    "timeout_secs": {
                        "type": "integer",
                        "description": "Wall-clock limit (default 120, max 600)"
                    }
                },
                "required": ["command"]
            }
        }));
        defs
    }

    /// Root relative path params at the worktree and reject lexical escapes.
    /// Mirrors the param-name surface of `agent_basics` (everything keys on
    /// `path`).
    fn clamp_paths(&self, tool: &str, params: &Value) -> Result<Value, String> {
        clamp_paths_to(&self.worktree, tool, params, "worktree", false)
    }

    /// Run `command` via `sh -lc` at the worktree root. Returns
    /// `{exit_code, output, timed_out}` — non-zero exits are values, not
    /// errors, so the model (and contract evaluation) can read them.
    ///
    /// Thin wrapper over the shared [`run_shell_on`]: the coder always runs on
    /// the host, pinned to the worktree root.
    pub async fn run_shell(
        &self,
        command: &str,
        timeout_secs: Option<u64>,
    ) -> Result<Value, String> {
        let substrate: Arc<dyn Substrate> = Arc::new(LocalSubstrate::new());
        run_shell_on(
            &substrate,
            Some(&self.worktree),
            &self.inspectors,
            command,
            timeout_secs,
            MAX_SHELL_TIMEOUT_SECS,
            // The model's own shell holds no forge credential, so publication
            // fails on authentication however the command is spelled — the one
            // version of that property that does not depend on out-lexing
            // `/bin/sh` (car#1084).
            ForgeCredentials::Withhold,
        )
        .await
    }

    /// [`Self::run_shell`] for an outcome-contract check: same shell, same
    /// inspectors, but clamped to [`Self::check_timeout_ceiling`] instead of the
    /// model-facing [`MAX_SHELL_TIMEOUT_SECS`].
    ///
    /// Separate entry point rather than a field read inside `run_shell`, so the
    /// `shell` tool the model calls cannot reach the raised ceiling: a repo whose
    /// test gate legitimately needs twenty minutes should not thereby let the
    /// model sit on a hung command for twenty (car#1065).
    pub(crate) async fn run_check_shell(
        &self,
        command: &str,
        timeout_secs: Option<u64>,
    ) -> Result<Value, String> {
        let substrate: Arc<dyn Substrate> = Arc::new(LocalSubstrate::new());
        run_shell_on(
            &substrate,
            Some(&self.worktree),
            &self.inspectors,
            command,
            timeout_secs,
            self.check_timeout_ceiling,
            // A check runs a command the CONTRACT declares, not one the model
            // just wrote, so the forge credential stays. It does not follow
            // that the check may name a credential: `self.inspectors` above is
            // the full coder chain, `DenyCredentialAccess` included, so the
            // command is refused on its text (car#1066). See the
            // `ForgeCredentials::Inherit` docs for what that does and does not
            // cover.
            ForgeCredentials::Inherit,
        )
        .await
    }

    async fn execute_in_session(
        &self,
        tool: &str,
        params: &Value,
        session_id: Option<&str>,
    ) -> Result<Value, String> {
        if let Some(agent_id) = &self.agent_id {
            // Classify against the delegate's OWN declared tiers, not the bare
            // name map. `assistant_tool_tier`'s fallback arm is `ReadOnly`, so
            // any name it does not know — every `parslee_*` tool among them —
            // was being classified as the most permissive tier and waved
            // through the per-agent gate. `classify_tool_tier_with_defs` exists
            // for exactly this: honour a tool that declares its own tier.
            let tier = crate::agent_permissions::classify_tool_tier_with_defs(
                tool,
                params,
                &self.all_delegate_defs(),
            );
            let mode = crate::agent_permissions::resolve(agent_id, tier);
            // An explicit per-session grant answers RequireApproval for only
            // the names it carries. It cannot turn a configured Deny into an
            // allow, and the inspector chain below still gets the call.
            if !(mode == car_policy::ApprovalMode::RequireApproval
                && self.session_approved_tools.contains(tool))
            {
                enforce_agent_permission(agent_id, tool, tier, mode)?;
            }
        }

        if tool == "shell" {
            let command = params
                .get("command")
                .and_then(Value::as_str)
                .ok_or("missing 'command' parameter")?;
            let timeout_secs = params.get("timeout_secs").and_then(Value::as_u64);
            // A shell call cannot be treated as mutating on its face — the model
            // has to grep, build and test to investigate anything, and marking
            // every one of those would make a no-change finding unreachable.
            // So it is judged by effect: fingerprint either side and record a
            // mutation only if the worktree actually moved. If git cannot answer
            // on either side, assume it did — an unknown is not a clean bill.
            let before = super::no_change::worktree_fingerprint(&self.worktree);
            let result = self.run_shell(command, timeout_secs).await;
            let after = super::no_change::worktree_fingerprint(&self.worktree);
            match (&before, &after) {
                (Some(a), Some(b)) if a == b => {}
                _ => self.mutations.record_mutation(),
            }
            return result;
        }

        // The file-writing built-ins are mutating by definition, and only a
        // SUCCESSFUL one counts — a write the policy chain refused changed
        // nothing and must not disqualify the session.
        let is_mutating_tool = matches!(tool, "write_file" | "edit_file");

        // Reachability, not mere attachment — see `delegates_reachable`. An
        // unadvertised delegate name falls through to the ordinary path below
        // and ends as `unknown tool`, which is what a run that was never
        // offered the surface should get.
        if self.delegates_reachable() {
            if let Some(delegate) = self.delegate_for(tool) {
                // Delegate-owned tools bypass the worktree path clamp because
                // they execute on a different substrate, but operator policy
                // still governs them. Otherwise a deny_tool rule would stop a
                // built-in and silently miss the same coder session's delegate.
                if let Some(reason) = self.inspectors.check(tool, params) {
                    return Err(format!("denied by policy: {reason}"));
                }
                return delegate.executor.execute(tool, params).await;
            }
        }

        let clamped = self.clamp_paths(tool, params)?;
        if let Some(reason) = self.inspectors.check(tool, &clamped) {
            return Err(format!("denied by policy: {reason}"));
        }
        let substrate: Arc<dyn Substrate> = Arc::new(LocalSubstrate::new());
        let ledger = self.read_ledgers.ledger_for(session_id);
        match agent_basics::execute_with_ledger(&substrate, &ledger, tool, &clamped).await {
            Some(result) => {
                if is_mutating_tool && result.is_ok() {
                    self.mutations.record_mutation();
                }
                result
            }
            None => Err(format!("unknown tool: {tool}")),
        }
    }
}

#[async_trait]
impl ToolExecutor for WorktreeExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        self.execute_in_session(tool, params, None).await
    }

    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        _action_id: &str,
        _timeout_ms: Option<u64>,
        session_id: Option<&str>,
        _attempt: u32,
    ) -> Result<Value, String> {
        self.execute_in_session(tool, params, session_id).await
    }
}

#[cfg(test)]
mod tests {
    /// A child that reports `TOK`, `GHT`, `CFG` and `HELPER` — one field per
    /// credential route — spelled for the platform's shell. Read a field back
    /// with [`probe_field`] rather than by eye: the fields are `|`-separated on
    /// Unix and newline-separated on Windows, because `cmd`'s `echo` cannot
    /// suppress its newline the way `printf` can.
    ///
    /// `/bin/sh` does not exist on Windows — the coder shells out through
    /// `cmd /C` there, see [`WorktreeExecutor::run_shell`] — and the two tests
    /// below were the only ones in this module without the Windows spelling
    /// their neighbours already have. They died with "the system cannot find
    /// the path specified" and took the whole Windows leg of CI with them,
    /// while asserting nothing at all about the strip (car#1096). `cmd` leaves
    /// `%VAR%` literal when VAR is unset, so `if defined` stands in for the
    /// POSIX default here. The two are not identical — `if defined` is the
    /// analogue of `${VAR-…}`, which treats a defined-but-empty variable as
    /// present, where `${VAR:-…}` substitutes for it — but every variable
    /// these tests set carries a value, so the mapping is exact for them.
    fn credential_probe() -> tokio::process::Command {
        #[cfg(unix)]
        {
            let mut cmd = tokio::process::Command::new("/bin/sh");
            cmd.arg("-c").arg(
                "printf 'TOK=%s|GHT=%s|CFG=%s|HELPER=%s' \
                 \"${GH_TOKEN:-EMPTY}\" \"${GITHUB_TOKEN:-EMPTY}\" \
                 \"${GH_CONFIG_DIR:-UNSET}\" \"${GIT_CONFIG_COUNT:-UNSET}\"",
            );
            cmd
        }
        #[cfg(windows)]
        {
            let mut cmd = tokio::process::Command::new("cmd");
            cmd.arg("/C").arg(
                "(if defined GH_TOKEN (echo TOK=%GH_TOKEN%) else (echo TOK=EMPTY)) & \
                 (if defined GITHUB_TOKEN (echo GHT=%GITHUB_TOKEN%) else (echo GHT=EMPTY)) & \
                 (if defined GH_CONFIG_DIR (echo CFG=%GH_CONFIG_DIR%) else (echo CFG=UNSET)) & \
                 (if defined GIT_CONFIG_COUNT (echo HELPER=%GIT_CONFIG_COUNT%) else (echo HELPER=UNSET))",
            );
            cmd
        }
    }

    /// One named field out of a [`credential_probe`] record.
    ///
    /// Asked field-wise, never as a substring of the whole record: a
    /// `contains("TOK=x")` is also satisfied by `TOK=xy`, so it cannot tell an
    /// exact credential value from one that merely starts with it. Splits on
    /// both separators the probe can emit (see its doc comment).
    fn probe_field(text: &str, name: &str) -> Option<String> {
        let prefix = format!("{name}=");
        text.split(['|', '\n', '\r'])
            .find_map(|field| field.trim().strip_prefix(&prefix))
            .map(str::to_string)
    }

    /// The mechanism, tested directly rather than through `run_shell`.
    ///
    /// The inspector chain already refuses a command that *names* a credential
    /// variable (`DenyCredentialAccess`), so a shell command cannot be used to
    /// observe this. That deny is a pattern defence of exactly the class
    /// car#1076 showed cannot be made complete against `/bin/sh`; this one is
    /// structural, and the two are complementary — the pattern stops the
    /// obvious read, and the strip means there is nothing to read when the
    /// pattern is evaded.
    ///
    /// The credentials are planted on the *builder*, never with
    /// `std::env::set_var`. Both spellings reach the child, but a process-wide
    /// set is shared with every other test in the binary: `check-windows` runs
    /// `cargo test --lib`, which is the threaded harness, so a sibling's
    /// `set_var`/`remove_var` landing between this one's set and its spawn
    /// would flip either test's answer. Planting on the builder also makes the
    /// assertion sharper — `env_remove` now has to beat an explicit value on
    /// the same `Command`, not merely an inherited one.
    #[tokio::test]
    async fn withholding_removes_every_route_to_a_forge_credential() {
        let mut cmd = credential_probe();
        cmd.env("GH_TOKEN", "ghp_secret_do_not_leak")
            .env("GITHUB_TOKEN", "gho_secret_do_not_leak");
        withhold_forge_credentials(&mut cmd);
        let out = cmd.output().await.expect("child ran");
        let text = String::from_utf8_lossy(&out.stdout).to_string();

        assert_eq!(
            probe_field(&text, "TOK").as_deref(),
            Some("EMPTY"),
            "GH_TOKEN survived: {text}"
        );
        assert_eq!(
            probe_field(&text, "GHT").as_deref(),
            Some("EMPTY"),
            "GITHUB_TOKEN survived: {text}"
        );
        assert!(
            !text.contains("ghp_secret_do_not_leak") && !text.contains("gho_secret_do_not_leak"),
            "a credential leaked: {text}"
        );
        // gh must not fall back to the real ~/.config/gh/hosts.yml.
        assert!(
            !text.contains("CFG=UNSET"),
            "GH_CONFIG_DIR not pinned: {text}"
        );
        assert_eq!(
            probe_field(&text, "HELPER").as_deref(),
            Some("1"),
            "git config override not applied: {text}"
        );

        // Behavioural, not env-shape: git in this child must resolve NO
        // credential helper. That is the route which ignores GH_TOKEN entirely
        // and would otherwise have left `git push` over HTTPS working.
        let mut git = tokio::process::Command::new("git");
        git.arg("config").arg("--get").arg("credential.helper");
        withhold_forge_credentials(&mut git);
        let helper = git.output().await.expect("git ran");
        let resolved = String::from_utf8_lossy(&helper.stdout).trim().to_string();
        assert!(
            resolved.is_empty(),
            "a credential helper survived into the child: {resolved}"
        );
    }

    /// A child that was NOT withheld from still sees the environment — proving
    /// the test above is observing the strip and not an already-empty env.
    ///
    /// Builder-scoped for the same reason as its sibling. `GITHUB_TOKEN` is
    /// pinned to a fixture even though nothing reads it back: the probe prints
    /// every field it knows, so leaving that one to the ambient environment
    /// would put a runner's real token into `text`.
    #[tokio::test]
    async fn an_untouched_child_still_sees_the_credential() {
        let mut cmd = credential_probe();
        cmd.env("GH_TOKEN", "ghp_inherit_me")
            .env("GITHUB_TOKEN", "gho_not_read_back");
        let out = cmd.output().await.expect("child ran");
        let text = String::from_utf8_lossy(&out.stdout).to_string();
        // The TOK field is compared exactly, so this proves the child saw the
        // value we set and not merely something beginning with it.
        assert_eq!(
            probe_field(&text, "TOK").as_deref(),
            Some("ghp_inherit_me"),
            "control case failed — the strip test would pass vacuously"
        );
    }

    /// `GH_CONFIG_DIR` must point somewhere that exists and holds no hosts.yml,
    /// or `gh` falls straight back to the operator's real config.
    #[test]
    fn the_empty_config_dir_exists_and_is_empty_of_forge_config() {
        let dir = empty_config_dir();
        assert!(
            dir.is_dir(),
            "gh falls back to the real config if this is absent"
        );
        assert!(!dir.join("hosts.yml").exists());
    }

    /// macOS `/etc/profile` runs `path_helper`, which rebuilds PATH with the
    /// system dirs FIRST and appends the inherited ones — so a toolchain the
    /// operator put first for the daemon lands at the tail inside the agent's
    /// shell, and a stale system binary shadows it. Surfaced by the coder A/B: a
    /// venv-first PATH still lost `pip` to `/usr/local/bin/pip`, whose
    /// `#!/usr/bin/python` shebang does not exist on modern macOS, so every
    /// `pip` step in a derived contract failed forever and sank sessions whose
    /// real work had already gone green. The macOS twin of the Windows
    /// over-long-PATH bug `car_engine::win_env` fixes.
    #[cfg(unix)]
    #[tokio::test]
    async fn login_shell_keeps_the_inherited_path_ahead_of_the_profiles() {
        let dir = tempfile::tempdir().unwrap();
        // A fake tool that must win over anything the profile puts earlier.
        let bin = dir.path().join("car-path-probe");
        std::fs::write(&bin, "#!/bin/sh\necho WINNER\n").unwrap();
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap();

        let orig = std::env::var("PATH").unwrap_or_default();
        std::env::set_var("PATH", format!("{}:{}", dir.path().display(), orig));
        let script =
            prepend_inherited_path("command -v car-path-probe >/dev/null && car-path-probe");
        std::env::set_var("PATH", &orig);

        let out = tokio::process::Command::new("/bin/sh")
            .arg("-lc")
            .arg(&script)
            .output()
            .await
            .unwrap();
        assert_eq!(
            String::from_utf8_lossy(&out.stdout).trim(),
            "WINNER",
            "the daemon's PATH must survive the login shell's profile"
        );
    }

    /// No PATH to re-assert → the command must pass through untouched.
    #[test]
    fn path_prepend_is_a_no_op_without_a_path() {
        let orig = std::env::var("PATH").ok();
        std::env::remove_var("PATH");
        assert_eq!(prepend_inherited_path("echo hi"), "echo hi");
        if let Some(p) = orig {
            std::env::set_var("PATH", p);
        }
    }

    /// A PATH with a space or a quote must not break out of the export. The
    /// POSIX idiom for a literal `'` inside single quotes is `'\''` — close,
    /// escaped quote, reopen. Unix-only: it round-trips through `/bin/sh`,
    /// which doesn't exist on Windows (where the coder shells out differently).
    #[cfg(unix)]
    #[test]
    fn path_prepend_quotes_hostile_paths() {
        assert_eq!(
            sh_single_quote("/a b/bin:/it's/bin"),
            r#"'/a b/bin:/it'\''s/bin'"#
        );
        // And it must actually round-trip through a real shell.
        let out = std::process::Command::new("/bin/sh")
            .arg("-c")
            .arg(format!("printf %s {}", sh_single_quote("/a b/x:/it's/y")))
            .output()
            .unwrap();
        assert_eq!(String::from_utf8_lossy(&out.stdout), "/a b/x:/it's/y");
    }
    use super::*;

    fn executor() -> (tempfile::TempDir, WorktreeExecutor) {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::new(dir.path());
        (dir, exec)
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn shell_runs_at_worktree_root() {
        let (dir, exec) = executor();
        let out = exec.run_shell("pwd", Some(10)).await.unwrap();
        let cwd = out["output"].as_str().unwrap().trim();
        assert_eq!(
            PathBuf::from(cwd).canonicalize().unwrap(),
            dir.path().canonicalize().unwrap()
        );
        assert_eq!(out["exit_code"], 0);
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn shell_runs_at_worktree_root() {
        let (dir, exec) = executor();
        // `cmd /C cd` prints the current directory on Windows.
        let out = exec.run_shell("cd", Some(10)).await.unwrap();
        let cwd = out["output"].as_str().unwrap().trim();
        assert_eq!(
            PathBuf::from(cwd).canonicalize().unwrap(),
            dir.path().canonicalize().unwrap()
        );
        assert_eq!(out["exit_code"], 0);
    }

    #[tokio::test]
    async fn shell_reports_nonzero_exit_as_value() {
        let (_dir, exec) = executor();
        let out = exec.run_shell("exit 3", Some(10)).await.unwrap();
        assert_eq!(out["exit_code"], 3);
        assert_eq!(out["timed_out"], false);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn shell_captures_stderr() {
        let (_dir, exec) = executor();
        let out = exec
            .run_shell("echo to-out; echo to-err 1>&2", Some(10))
            .await
            .unwrap();
        let text = out["output"].as_str().unwrap();
        assert!(text.contains("to-out") && text.contains("to-err"));
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn shell_captures_stderr() {
        let (_dir, exec) = executor();
        // `&` is cmd's command separator; `1>&2` redirects stderr.
        let out = exec
            .run_shell("echo to-out & echo to-err 1>&2", Some(10))
            .await
            .unwrap();
        let text = out["output"].as_str().unwrap();
        assert!(text.contains("to-out") && text.contains("to-err"), "{text}");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn shell_timeout_kills_and_reports() {
        let (_dir, exec) = executor();
        let started = std::time::Instant::now();
        let out = exec.run_shell("sleep 30", Some(1)).await.unwrap();
        assert!(
            started.elapsed() < Duration::from_secs(10),
            "did not wait out the sleep"
        );
        assert_eq!(out["timed_out"], true);
        assert!(out["exit_code"].is_null());
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn shell_timeout_kills_and_reports() {
        let (_dir, exec) = executor();
        let started = std::time::Instant::now();
        // An infinite `cmd` loop is a deterministic blocker that needs no
        // console or stdin (unlike `timeout`/`pause`); the 1s wall-clock limit
        // must fire and kill it.
        let out = exec
            .run_shell("for /L %i in () do @rem", Some(1))
            .await
            .unwrap();
        assert!(
            started.elapsed() < Duration::from_secs(10),
            "did not enforce the timeout"
        );
        assert_eq!(out["timed_out"], true);
        assert!(out["exit_code"].is_null());
    }

    #[tokio::test]
    async fn shell_denied_by_policy() {
        let (_dir, exec) = executor();
        let err = exec
            .run_shell("git push origin main", Some(5))
            .await
            .unwrap_err();
        assert!(err.contains("denied by policy"), "{err}");
    }

    #[test]
    fn noninteractive_agent_permissions_fail_closed_for_full_access_approval() {
        assert!(
            enforce_agent_permission(
                "writer",
                "shell",
                car_policy::PermissionTier::SandboxEdit,
                car_policy::ApprovalMode::RequireApproval,
            )
            .is_ok(),
            "sandbox edits remain usable under the Balanced default"
        );

        let err = enforce_agent_permission(
            "writer",
            "shell",
            car_policy::PermissionTier::FullAccess,
            car_policy::ApprovalMode::RequireApproval,
        )
        .unwrap_err();
        assert!(err.contains("approval required"), "{err}");
        assert!(err.contains("no interactive approval channel"), "{err}");

        let err = enforce_agent_permission(
            "writer",
            "shell",
            car_policy::PermissionTier::ReadOnly,
            car_policy::ApprovalMode::Deny,
        )
        .unwrap_err();
        assert!(err.contains("denied for agent"), "{err}");
    }

    #[tokio::test]
    async fn relative_file_writes_land_in_worktree() {
        let (dir, exec) = executor();
        exec.execute(
            "write_file",
            &json!({"path": "sub/out.txt", "content": "hi"}),
        )
        .await
        .unwrap();
        assert_eq!(
            std::fs::read_to_string(dir.path().join("sub/out.txt")).unwrap(),
            "hi"
        );
    }

    /// (#1a) The read-before-edit guard is LIVE through the coder's
    /// WorktreeExecutor: editing a worktree file the session never read is
    /// refused. Reverting this call site to the ungated `agent_basics::execute`
    /// makes this pass silently — that's the regression this test pins.
    #[tokio::test]
    async fn edit_requires_prior_read_through_worktree_executor() {
        let (dir, exec) = executor();
        std::fs::write(dir.path().join("f.txt"), "hello world").unwrap();
        let err = exec
            .execute(
                "edit_file",
                &json!({ "path": "f.txt", "old_text": "hello", "new_text": "hi" }),
            )
            .await
            .unwrap_err();
        assert!(err.contains("before editing it"), "{err}");
    }

    #[tokio::test]
    async fn escaping_writes_are_rejected_in_code() {
        let (_dir, exec) = executor();
        let err = exec
            .execute(
                "write_file",
                &json!({"path": "../escape.txt", "content": "x"}),
            )
            .await
            .unwrap_err();
        assert!(err.contains("outside the worktree"), "{err}");

        let err = exec
            .execute(
                "write_file",
                &json!({"path": "/tmp/abs-escape.txt", "content": "x"}),
            )
            .await
            .unwrap_err();
        assert!(err.contains("outside the worktree"), "{err}");
    }

    #[tokio::test]
    async fn list_dir_defaults_to_worktree_not_process_cwd() {
        let (dir, exec) = executor();
        std::fs::write(dir.path().join("marker.txt"), "x").unwrap();
        let out = exec.execute("list_dir", &json!({})).await.unwrap();
        assert!(
            out.to_string().contains("marker.txt"),
            "expected worktree listing, got: {out}"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn output_is_tail_capped() {
        let (_dir, exec) = executor();
        // ~200KB of output → capped to the 64KB tail.
        let out = exec
            .run_shell("i=0; while [ $i -lt 5000 ]; do echo 'line of output 40 bytes long....'; i=$((i+1)); done", Some(30))
            .await
            .unwrap();
        let text = out["output"].as_str().unwrap();
        assert!(text.len() <= MAX_OUTPUT_BYTES + 32, "len={}", text.len());
        assert!(text.starts_with("…[truncated]…"));
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn output_is_tail_capped() {
        let (_dir, exec) = executor();
        // ~200KB of output → capped to the 64KB tail.
        let out = exec
            .run_shell(
                "for /L %i in (1,1,5000) do @echo line of output 40 bytes long....",
                Some(60),
            )
            .await
            .unwrap();
        let text = out["output"].as_str().unwrap();
        assert!(text.len() <= MAX_OUTPUT_BYTES + 32, "len={}", text.len());
        assert!(text.starts_with("…[truncated]…"));
    }

    #[tokio::test]
    async fn unknown_tool_errors() {
        let (_dir, exec) = executor();
        assert!(exec.execute("teleport", &json!({})).await.is_err());
    }

    struct StubDelegate;
    #[async_trait]
    impl ToolExecutor for StubDelegate {
        async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
            Ok(json!({ "via": "delegate", "tool": tool, "echo": params.clone() }))
        }
    }

    /// car#1071: a coder session can recall from the graph memory.
    ///
    /// The product's headline capability was unavailable to the flagship coding
    /// agent inside it — the coder could not read a fact anyone had stored about
    /// the project.
    #[test]
    fn a_coder_session_carries_graph_memory_recall() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        let names: Vec<String> = exec
            .all_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(
            names.iter().any(|n| n == "recall"),
            "the coder must be able to recall stored project facts"
        );
    }

    /// And CANNOT write to it. This is the security half of car#1071 and the
    /// assertion most worth keeping.
    ///
    /// `remember` is an information-flow sink carrying `persistent_memory`: it
    /// writes durable state that every later session reads. car#1081 says a
    /// coder session may be triaging an issue from a public tracker whose body
    /// is attacker-authored, so a write path turns a single prompt injection
    /// into a persistence attack — hostile text stored once and recalled as
    /// trusted context indefinitely.
    #[test]
    fn a_coder_session_cannot_write_to_graph_memory() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        let names: Vec<String> = exec
            .all_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(
            !names.iter().any(|n| n == "remember"),
            "a coder must not write durable memory later sessions will trust"
        );
        // The filter is the mechanism, so assert it directly too: if
        // MemoryTools grows a second write tool, this catches it.
        let attached = recall_only_memory_defs();
        assert_eq!(attached.len(), 1);
        assert_eq!(attached[0]["name"], "recall");
    }

    /// Two delegates coexist. This is the whole point of the refactor.
    ///
    /// `with_delegate` used to assign a single `Option` slot and one shared
    /// `Vec<Value>`, so a second call replaced the first — silently, with the
    /// first delegate's tools vanishing from `all_tool_defs` and its dispatch
    /// falling through to `unknown tool`. Three separate issues (car#1073
    /// network, car#1069 browser, car#1071 memory) each hit that as their
    /// blocker.
    #[tokio::test]
    async fn a_second_delegate_does_not_evict_the_first() {
        let dir = tempfile::tempdir().unwrap();
        let first = vec![json!({
            "name": "alpha_tool",
            "description": "first",
            "parameters": { "type": "object", "properties": {} }
        })];
        let second = vec![json!({
            "name": "beta_tool",
            "description": "second",
            "parameters": { "type": "object", "properties": {} }
        })];
        let exec = WorktreeExecutor::new(dir.path())
            .with_delegate(Arc::new(StubDelegate), first)
            .with_delegate(Arc::new(StubDelegate), second);

        let names: Vec<String> = exec
            .all_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(
            names.iter().any(|n| n == "alpha_tool"),
            "first delegate evicted"
        );
        assert!(
            names.iter().any(|n| n == "beta_tool"),
            "second delegate missing"
        );
        assert!(names.iter().any(|n| n == "read_file"), "built-ins lost");

        exec.advertise_delegates();
        for tool in ["alpha_tool", "beta_tool"] {
            let out = exec.execute(tool, &json!({ "x": 1 })).await.unwrap();
            assert_eq!(out["via"], "delegate", "{tool} did not route to a delegate");
            assert_eq!(out["tool"], tool, "{tool} routed to the wrong delegate");
        }
    }

    /// A name advertised by two delegates resolves to the one attached FIRST,
    /// and the overlap is reportable rather than silent.
    #[tokio::test]
    async fn a_name_collision_resolves_to_the_first_delegate_and_is_reportable() {
        let dir = tempfile::tempdir().unwrap();
        let def = |name: &str| {
            vec![json!({
                "name": name,
                "description": "x",
                "parameters": { "type": "object", "properties": {} }
            })]
        };
        let exec = WorktreeExecutor::new(dir.path())
            .with_delegate(Arc::new(StubDelegate), def("shared_name"))
            .with_delegate(Arc::new(StubDelegate), def("shared_name"));

        assert_eq!(
            exec.delegate_name_collisions(),
            vec!["shared_name".to_string()],
            "an overlap a call site could assert on must be visible"
        );

        // Still dispatches — deterministically, to the first.
        exec.advertise_delegates();
        let out = exec.execute("shared_name", &json!({})).await.unwrap();
        assert_eq!(out["via"], "delegate");
    }

    /// The healthy case: nothing overlaps.
    #[test]
    fn a_coder_session_has_no_delegate_name_collisions() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        assert!(
            exec.delegate_name_collisions().is_empty(),
            "two attached delegates advertise the same tool name"
        );
    }

    #[tokio::test]
    async fn delegate_tool_routes_through_delegate_and_is_advertised() {
        let dir = tempfile::tempdir().unwrap();
        let defs = vec![json!({
            "name": "ext_tool",
            "description": "external",
            "parameters": { "type": "object", "properties": {} }
        })];
        let exec = WorktreeExecutor::new(dir.path()).with_delegate(Arc::new(StubDelegate), defs);

        // all_tool_defs surfaces the delegate tool alongside the built-ins…
        let names: Vec<String> = exec
            .all_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        assert!(names.iter().any(|n| n == "ext_tool"));
        assert!(names.iter().any(|n| n == "read_file")); // built-ins still present

        // This run advertised the delegate surface, so it may call it. Without
        // this the name is not reachable at all — see
        // `an_unadvertised_delegate_tool_is_not_reachable`.
        exec.advertise_delegates();

        // …and execute() routes it to the delegate (no worktree path clamp).
        let out = exec.execute("ext_tool", &json!({ "x": 1 })).await.unwrap();
        assert_eq!(out["via"], "delegate");
        assert_eq!(out["tool"], "ext_tool");
        assert_eq!(out["echo"]["x"], 1);

        // Tools the delegate doesn't own still fall through to "unknown".
        assert!(exec.execute("teleport", &json!({})).await.is_err());
    }

    #[tokio::test]
    async fn project_policy_denies_shell_file_and_advertised_delegate_calls() {
        let dir = tempfile::tempdir().unwrap();
        let policies = dir.path().join(".car").join("policies");
        std::fs::create_dir_all(&policies).unwrap();
        std::fs::write(
            policies.join("rules.toml"),
            "deny_tool = [\"write_file\", \"ext_tool\"]\ndeny_keyword = [\"BLOCKED CHECK\"]\n",
        )
        .unwrap();

        let defs = vec![json!({
            "name": "ext_tool",
            "description": "external",
            "parameters": { "type": "object", "properties": {} }
        })];
        let mut exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        exec = exec.with_delegate(Arc::new(StubDelegate), defs);

        let file_err = exec
            .execute(
                "write_file",
                &json!({"path": "blocked.txt", "content": "x"}),
            )
            .await
            .expect_err("project deny_tool must govern coder file tools");
        assert!(file_err.contains("operator policy"), "{file_err}");

        let check_err = exec
            .run_check_shell("echo BLOCKED CHECK", Some(5))
            .await
            .expect_err("contract checks use the same governed chain");
        assert!(check_err.contains("operator policy"), "{check_err}");

        exec.advertise_delegates();
        let delegate_err = exec
            .execute("ext_tool", &json!({}))
            .await
            .expect_err("advertised delegates remain governed by operator policy");
        assert!(delegate_err.contains("operator policy"), "{delegate_err}");
    }

    /// The browser flag is approval to offer the surface, not a policy bypass.
    /// This call must be denied before BrowserTools can launch Chromium.
    #[tokio::test]
    async fn opted_in_browser_calls_still_cross_the_coder_policy_chain() {
        let dir = tempfile::tempdir().unwrap();
        let policies = dir.path().join(".car").join("policies");
        std::fs::create_dir_all(&policies).unwrap();
        std::fs::write(
            policies.join("browser.toml"),
            "deny_tool = [\"browse_navigate\"]\n",
        )
        .unwrap();

        let exec = WorktreeExecutor::for_coder_session(dir.path())
            .unwrap()
            .with_browser_tools();
        exec.advertise_delegates();
        let err = exec
            .execute("browse_navigate", &json!({"url": "https://example.com"}))
            .await
            .expect_err("project policy must intercept browser delegate calls");
        assert!(err.contains("operator policy"), "{err}");
        assert!(err.contains("browse_navigate"), "{err}");
    }

    #[test]
    fn malformed_project_policy_refuses_coder_session() {
        let dir = tempfile::tempdir().unwrap();
        let policies = dir.path().join(".car").join("policies");
        std::fs::create_dir_all(&policies).unwrap();
        std::fs::write(
            policies.join("broken.toml"),
            "deny_tool = [this is not TOML\n",
        )
        .unwrap();

        let err = WorktreeExecutor::for_coder_session(dir.path())
            .err()
            .expect("a session must not start with silently missing denies");
        assert!(err.contains("refusing to start coder session"), "{err}");
        assert!(err.contains("operator policy"), "{err}");
        assert!(err.contains("broken.toml"), "{err}");
    }

    /// Both coder entry points go through `for_coder_session`, so the delegate
    /// and the policy subject cannot drift apart between them
    /// (Parslee-ai/car#1063). Dropping either from the shared constructor fails
    /// here.
    /// Attaching a delegate must not make it callable.
    ///
    /// Dispatch used to key on `delegate_defs` — what the delegate *offers* —
    /// and return before both the path clamp and the inspector chain, so a
    /// coding run that advertised only the built-ins could still invoke a
    /// `parslee_*` tool by name. `parslee_generate_document` saves to the
    /// user's connected drive, and the per-agent gate does not catch it because
    /// an unrecognised name classifies as `ReadOnly`.
    ///
    /// This test is about REACHABILITY, not advertisement — the sibling test
    /// covering the advertised list would still pass with the hole open.
    #[tokio::test]
    async fn an_unadvertised_delegate_tool_is_not_reachable() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();

        // The coding loop advertises the static built-ins and never calls
        // advertise_delegates().
        assert!(
            !exec.delegates_reachable(),
            "delegates must be closed until a run advertises them"
        );

        let delegate_name = crate::parslee_tools::ParsleeToolExecutor::tool_defs()
            .first()
            .and_then(|d| d["name"].as_str().map(String::from))
            .expect("the parslee delegate advertises at least one tool");

        let err = exec
            .execute(&delegate_name, &json!({}))
            .await
            .expect_err("an unadvertised delegate name must not dispatch");
        assert!(
            err.contains("unknown tool"),
            "expected it to fall through to the ordinary path, got: {err}"
        );

        // And a run that DOES advertise the surface still reaches it, so the
        // agent-build path is unaffected.
        exec.advertise_delegates();
        assert!(exec.delegates_reachable());
    }

    #[tokio::test]
    async fn for_coder_session_carries_the_parslee_delegate_and_policy_subject() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();

        let names: Vec<String> = exec
            .all_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(String::from))
            .collect();
        for parslee in crate::parslee_tools::ParsleeToolExecutor::tool_names() {
            assert!(
                names.contains(&parslee),
                "{parslee} missing from all_tool_defs: {names:?}"
            );
        }
        assert!(names.iter().any(|n| n == "read_file")); // built-ins still present

        // The stable `car-coder` policy subject is what the Agent Permissions
        // screen denies against.
        assert_eq!(exec.agent_id.as_deref(), Some("car-coder"));
    }

    /// The governed network pair reaches a coder session (car#1073), and stays
    /// default-closed while it does.
    ///
    /// The tier assertion is the load-bearing half. `full_access` is what makes
    /// the per-agent gate hard-block these until an operator grants the tier; a
    /// re-tier to `read_only` would hand every unattended coder run the network
    /// silently, and would still pass a test that only checked attachment.
    #[tokio::test]
    async fn for_coder_session_carries_the_governed_network_pair_at_full_access() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();

        for tool in ["http_request", "web_search"] {
            let defs = exec.delegate_defs_named(tool);
            assert_eq!(defs.len(), 1, "{tool} must be attached exactly once");
            assert_eq!(
                defs[0]["tier"], "full_access",
                "{tool} must stay full_access — that tier is what keeps the \
                 per-agent gate closed by default"
            );
        }
    }

    /// A third delegate must not shadow a name an earlier one owns. Collisions
    /// resolve first-wins, so an overlap would be silent at runtime.
    #[tokio::test]
    async fn coder_session_delegates_do_not_collide() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::for_coder_session(dir.path()).unwrap();
        assert_eq!(
            exec.delegate_name_collisions(),
            Vec::<String>::new(),
            "two delegates advertise the same tool name"
        );
    }

    /// With no per-agent subject there is no gate, so nothing is withheld — the
    /// `None` arm of the accessor a loop consults before offering a
    /// `full_access` tool.
    #[test]
    fn an_executor_with_no_agent_subject_permits_full_access() {
        let dir = tempfile::tempdir().unwrap();
        assert!(WorktreeExecutor::new(dir.path()).permits_full_access());
    }

    #[cfg(unix)]
    struct TestProcessGroup {
        pgid: i32,
        armed: bool,
    }

    #[cfg(unix)]
    impl TestProcessGroup {
        fn from_file(path: &Path) -> Self {
            let pgid = std::fs::read_to_string(path)
                .unwrap_or_else(|error| {
                    panic!("read fixture process group {}: {error}", path.display())
                })
                .trim()
                .parse()
                .unwrap_or_else(|error| {
                    panic!("parse fixture process group {}: {error}", path.display())
                });
            Self { pgid, armed: true }
        }

        fn exists(&self) -> bool {
            let result = unsafe { libc::killpg(self.pgid, 0) };
            result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
        }

        async fn assert_reaped(mut self, outcome: &str) {
            for _ in 0..100 {
                if !self.exists() {
                    self.armed = false;
                    return;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            panic!(
                "shell process group {} survived the {outcome} path",
                self.pgid
            );
        }
    }

    #[cfg(unix)]
    impl Drop for TestProcessGroup {
        fn drop(&mut self) {
            if self.armed {
                unsafe {
                    libc::killpg(self.pgid, libc::SIGKILL);
                }
            }
        }
    }

    #[cfg(unix)]
    async fn wait_for_fixture_file(path: &Path) {
        for _ in 0..200 {
            if path.is_file() {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("fixture did not write {}", path.display());
    }

    #[cfg(unix)]
    fn publish_fixture_process_group(group_file: &Path) -> String {
        // The path is the readiness signal. Publish only after printf closes
        // the temporary file, so cancellation cannot observe an empty PID.
        // Match the executor's canonical worktree on macOS (/var -> /private/var).
        let group_file = group_file
            .parent()
            .unwrap()
            .canonicalize()
            .unwrap()
            .join(group_file.file_name().unwrap());
        let temporary = group_file.with_extension("pgid.pending");
        format!(
            "printf '%s' \"$$\" > {} && mv -f {} {}",
            sh_single_quote(&temporary.display().to_string()),
            sh_single_quote(&temporary.display().to_string()),
            sh_single_quote(&group_file.display().to_string()),
        )
    }

    #[cfg(unix)]
    fn long_lived_shell_command(group_file: &Path, finish: &str) -> String {
        format!(
            "{}; trap '' HUP; sleep 120 >/dev/null 2>&1 & {finish}",
            publish_fixture_process_group(group_file)
        )
    }

    /// A shell returning zero or non-zero does not grant its detached children
    /// a lifetime beyond the tool call. This reproduces the reported
    /// `car-server --no-auth` plus `car do --serve` shape reparented to pid 1
    /// after a foreground review completed, without claiming which command
    /// originally launched those observed processes.
    #[cfg(unix)]
    #[tokio::test]
    async fn local_shell_lifecycle_reaps_background_descendants_after_completed_outcomes() {
        let (dir, exec) = executor();
        for (name, finish, expected) in [("success", "exit 0", 0), ("error", "exit 7", 7)] {
            let group_file = dir.path().join(format!("{name}.pgid"));
            let out = exec
                .run_shell(&long_lived_shell_command(&group_file, finish), Some(10))
                .await
                .unwrap();
            assert_eq!(out["exit_code"], expected);
            TestProcessGroup::from_file(&group_file)
                .assert_reaped(name)
                .await;
        }
    }

    /// Dropping an in-flight shell future is the cancellation primitive used by
    /// the foreground CLI's SIGINT/SIGTERM path. The guard must sweep the group
    /// even though `run_shell_on` never reaches its ordinary return cleanup.
    #[cfg(unix)]
    #[tokio::test]
    async fn local_shell_lifecycle_reaps_process_group_on_cancellation() {
        let (dir, exec) = executor();
        let group_file = dir.path().join("cancelled.pgid");
        let command = format!(
            "{}; exec sleep 120",
            publish_fixture_process_group(&group_file)
        );
        let task = tokio::spawn(async move { exec.run_shell(&command, Some(180)).await });
        wait_for_fixture_file(&group_file).await;
        let group = TestProcessGroup::from_file(&group_file);
        task.abort();
        assert!(task.await.unwrap_err().is_cancelled());
        group.assert_reaped("cancellation").await;
    }

    /// Rust unwinding drops local futures. Keep that path load-bearing because
    /// `kill_on_drop` owns only the direct shell; without the group guard a
    /// panic still leaves grandchildren behind.
    #[cfg(unix)]
    #[tokio::test]
    async fn local_shell_lifecycle_reaps_process_group_on_panic() {
        let (dir, exec) = executor();
        let group_file = dir.path().join("panic.pgid");
        let task_group_file = group_file.clone();
        let command = format!(
            "{}; exec sleep 120",
            publish_fixture_process_group(&group_file)
        );
        let task = tokio::spawn(async move {
            let shell = exec.run_shell(&command, Some(180));
            tokio::pin!(shell);
            tokio::select! {
                result = &mut shell => panic!("fixture shell returned before panic: {result:?}"),
                _ = wait_for_fixture_file(&task_group_file) => panic!("intentional lifecycle fixture panic"),
            }
        });
        let join = task.await.unwrap_err();
        assert!(join.is_panic());
        TestProcessGroup::from_file(&group_file)
            .assert_reaped("panic")
            .await;
    }

    #[test]
    fn tail_respects_char_boundaries() {
        let s = "ééééé"; // 2 bytes each
        let t = tail(s, 3);
        assert!(t.ends_with('é'));
    }

    /// The two shell entry points read DIFFERENT ceilings, and that separation
    /// is the whole shape of the car#1065 fix: `run_check_shell` honors the
    /// operator's contract-check ceiling, `run_shell` — the one behind the
    /// model's `shell` tool — stays pinned at [`MAX_SHELL_TIMEOUT_SECS`].
    ///
    /// Read at a ceiling of 1s rather than a raised one so the assertion costs
    /// three seconds instead of ten minutes; the direction under test is which
    /// ceiling each path reads, and that is the same either way.
    #[cfg(unix)]
    #[tokio::test]
    async fn the_check_ceiling_binds_run_check_shell_and_not_the_model_facing_shell() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::new(dir.path()).with_check_timeout_ceiling(1);

        let checked = exec.run_check_shell("sleep 3", Some(10)).await.unwrap();
        assert_eq!(
            checked["timed_out"], true,
            "a contract check is bound by the executor's check ceiling"
        );

        let modelled = exec.run_shell("sleep 3", Some(10)).await.unwrap();
        assert_eq!(
            modelled["timed_out"], false,
            "the model's own shell keeps the advertised 600s ceiling — a slow \
             test gate is not a licence to hang"
        );
    }

    /// A contract check runs the SAME inspector chain as the model's shell, so
    /// `DenyCredentialAccess` refuses it on the command text — and that check
    /// is a substring matcher, not a boundary.
    ///
    /// Both directions are asserted, because `docs/car-code-task.md` now states
    /// both beside the contract input and either one alone reads as a promise
    /// the code does not keep. Denying only would suggest a check is sealed off
    /// from credentials; it is not, since `ForgeCredentials::Inherit` leaves the
    /// environment intact and an unmarked spelling carries no marker to match
    /// (car#1066).
    #[cfg(unix)]
    #[tokio::test]
    async fn a_contract_check_may_not_name_a_credential_even_though_it_inherits_one() {
        let dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::new(dir.path());

        for command in [
            "curl -H \"Authorization: Bearer $STAGING_API_TOKEN\" https://example.invalid/health",
            "sqlcmd -Q \"select 1\" -C \"$DB_CONNECTION_STRING\"",
            "cat ~/.aws/credentials",
        ] {
            let err = exec
                .run_check_shell(command, Some(5))
                .await
                .expect_err("a contract check naming a credential is refused");
            assert!(
                err.starts_with("denied by policy:"),
                "expected a policy refusal for {command:?}, got {err}"
            );
        }

        // …and the matcher is hardening, not a sandbox. None of these carries a
        // built-in marker, so all of them reach the shell with the daemon's
        // environment still intact. A contract author must not read the
        // refusals above as "a check cannot touch a credential".
        for command in [
            "echo \"Authorization: Bearer $TOKEN\"",
            "echo \"$DBURL\"",
            "echo ok",
        ] {
            let out = exec
                .run_check_shell(command, Some(5))
                .await
                .unwrap_or_else(|e| panic!("expected {command:?} to reach the shell, got {e}"));
            assert_eq!(out["exit_code"], 0, "{out}");
        }
    }

    /// The default is the constant the tool description advertises, and a `0`
    /// from config is floored rather than honored (it would clamp every check
    /// to one second).
    #[test]
    fn the_check_ceiling_defaults_to_the_shell_max_and_floors_zero() {
        let dir = tempfile::tempdir().unwrap();
        assert_eq!(
            WorktreeExecutor::new(dir.path()).check_timeout_ceiling(),
            MAX_SHELL_TIMEOUT_SECS
        );
        assert_eq!(
            WorktreeExecutor::new(dir.path())
                .with_check_timeout_ceiling(0)
                .check_timeout_ceiling(),
            1
        );
        assert_eq!(
            WorktreeExecutor::new(dir.path())
                .with_check_timeout_ceiling(1800)
                .check_timeout_ceiling(),
            1800
        );
    }
}