//! Concrete [`Executor`] transports.
//!
//! - [`LocalExec`] — spawn on the local machine (Sando's `ssh_target = "local"`
//!   fast-path; the agent's own in-process execution).
//! - [`SshExec`] — spawn `ssh` / `rsync` over the tailnet. Auth is the existing
//!   SSH keys; the rsync push/pull is extracted near-verbatim from Sando's
//!   `deploy.rs`.
//!
//! Both render a [`Step`] to one `/bin/sh` line and run it through the shared
//! [`RemoteHost`] command builder, so local and remote shell semantics are
//! identical. The capability gate is enforced here, caller-side, before any
//! command is dispatched.

use crate::capability::{CapabilityDenied, CapabilitySet};
use crate::executor::{Executor, SyncOpts, run_command_into_sink};
use crate::remote::{LogSink, RemoteHost, RunOutput, sh_quote};
use crate::step::{Action, ObserveKind, Step};
use anyhow::{Context, Result};
use async_trait::async_trait;
use std::fmt::Write as _;
use std::path::{Component, Path, PathBuf};
use tokio::process::Command;

/// Render a step to a single `/bin/sh` command line: optional `cd`, env
/// exports, then the body (a shell script verbatim, or sh-quoted argv).
fn render_shell_line(step: &Step) -> String {
    let mut line = String::new();
    if let Some(cwd) = &step.cwd {
        let _ = write!(line, "cd {} && ", sh_quote(&cwd.to_string_lossy()));
    }
    for (k, v) in &step.env {
        let _ = write!(line, "{}={} ", k, sh_quote(v));
    }
    match step.shell_script() {
        Some(script) => line.push_str(script),
        None => {
            let parts: Vec<String> = step.argv.iter().map(|a| sh_quote(a)).collect();
            line.push_str(&parts.join(" "));
        }
    }
    line
}

/// Enforce the caller-side capability gate, returning [`CapabilityDenied`] as
/// an `anyhow::Error` the caller can downcast for audit logging.
fn gate(caps: &CapabilitySet, host: &str, step: &Step) -> Result<()> {
    if caps.permits(&step.action) {
        Ok(())
    } else {
        Err(CapabilityDenied::new(host, &step.action).into())
    }
}

/// Gate a sync-plane artifact pull: the caller-side equivalent of the agent's
/// `/pull` handler (`agent::pull`). Two conditions, both **fail-closed**:
///
/// 1. The grant must include `observe:artifact` — retrieving a produced release
///    artifact is an observe-plane capability, distinct from reading the build
///    log. Denied → [`CapabilityDenied`], downcastable for audit.
/// 2. `requested` must lie under the executor's declared `pull_root`. A
///    transport with **no** root pulls nothing — that is the "declared per-host
///    artifact root" the audit finding calls for.
///
/// Without this, `pull_file`/`pull_dir`/`pull_glob` would rsync ANY path off the
/// host (`collect('mbp', '/Users/max/.tauri/passwords.env', …)` deposits the
/// notary credential into `dist_root`): "THE WALL" held on the agent plane and
/// not on the sync plane the agent hosts are actually collected over.
///
/// Confinement is **lexical** (reject `..`, require a component-wise prefix),
/// not canonicalizing like the agent's `confine_to_root`: the ssh transport's
/// path is on a remote host we cannot `canonicalize()` without an extra
/// round-trip, so both transports share the one check. The root is
/// operator-declared on trusted infra and the threat closed here is a recipe- or
/// daemon-supplied wild path; a symlink *inside* the root pointing out is the
/// agent plane's stronger (canonicalizing) guarantee, out of scope here.
fn gate_pull(
    caps: &CapabilitySet,
    host: &str,
    pull_root: Option<&Path>,
    requested: &Path,
) -> Result<()> {
    if !caps.permits_observe(&ObserveKind::Artifact) {
        return Err(CapabilityDenied::new(host, &Action::Observe(ObserveKind::Artifact)).into());
    }
    let Some(root) = pull_root else {
        anyhow::bail!(
            "pull from `{host}` denied: no artifact root declared for this host \
             (set `pull_root` in the host's topology entry)"
        );
    };
    anyhow::ensure!(
        !requested
            .components()
            .any(|c| matches!(c, Component::ParentDir)),
        "pull path `{}` from `{host}` contains `..`",
        requested.display()
    );
    anyhow::ensure!(
        requested.starts_with(root),
        "pull path `{}` escapes the declared artifact root `{}` on `{host}`",
        requested.display(),
        root.display()
    );
    Ok(())
}

/// An env *value* is sh-quoted at render time, but the *name* is interpolated
/// verbatim (`NAME=value`), so a name with shell metacharacters could inject a
/// command. Require each name to be a bare shell identifier before dispatch.
fn validate_env_names(step: &Step) -> Result<()> {
    for (k, _) in &step.env {
        let valid = !k.is_empty()
            && k.chars()
                .next()
                .is_some_and(|c| c == '_' || c.is_ascii_alphabetic())
            && k.chars().all(|c| c == '_' || c.is_ascii_alphanumeric());
        anyhow::ensure!(
            valid,
            "env name `{k}` must be a shell identifier ([A-Za-z_][A-Za-z0-9_]*)"
        );
    }
    Ok(())
}

/// Build the rsync `Command` shared by local and ssh push/pull. `src`/`dst` are
/// either plain paths (local) or `target:path` (ssh). `ssh_args`, when `Some`,
/// is the `ssh` argv rsync should transport over — [`RemoteHost::ssh_args`],
/// so the port and flags match the exec path exactly; `None` = a local rsync.
fn rsync_command(src: &str, dst: &str, ssh_args: Option<Vec<String>>, opts: &SyncOpts) -> Command {
    rsync_command_multi(std::slice::from_ref(&src.to_string()), dst, ssh_args, opts)
}

/// As [`rsync_command`], but with several sources into one destination — what a
/// glob expands to. `srcs` are passed as separate argv entries, never joined,
/// so a path is never re-split on whitespace.
fn rsync_command_multi(
    srcs: &[String],
    dst: &str,
    ssh_args: Option<Vec<String>>,
    opts: &SyncOpts,
) -> Command {
    let mut rsync = Command::new("rsync");
    // Kill the transfer if the caller's future is dropped (e.g. a promote whose
    // HTTP handler was cancelled by a client disconnect). Without this the
    // rsync orphans and keeps running; a retry then spawns a second rsync that
    // fights the first over the same `--delete` destination dir, wedging the
    // deploy. Matches the ssh-exec (remote.rs) and local-exec (executor.rs)
    // paths, which already set it.
    rsync.kill_on_drop(true);
    rsync.arg("-a");
    if opts.partial {
        rsync.arg("--partial");
    }
    if opts.compress {
        rsync.arg("-z");
    }
    if opts.delete {
        rsync.arg("--delete");
    }
    if let Some(chmod) = &opts.chmod {
        rsync.arg(format!("--chmod={chmod}"));
    }
    if opts.mkpath {
        rsync.arg("--mkpath");
    }
    for pattern in &opts.exclude {
        rsync.arg(format!("--exclude={pattern}"));
    }
    if let Some(args) = ssh_args {
        rsync.arg("-e").arg(format!("ssh {}", args.join(" ")));
    }
    rsync.args(srcs).arg(dst);
    rsync
}

/// Expand a glob against the local filesystem, for the transports that have no
/// remote shell to do it. Returns the matches sorted (deterministic argv), and
/// errors when nothing matches — a collect that quietly gathers zero files is
/// how an empty release ships.
fn expand_glob_locally(pattern: &str) -> Result<Vec<String>> {
    let paths = glob::glob(pattern).with_context(|| format!("bad glob pattern `{pattern}`"))?;
    let mut out: Vec<String> = Vec::new();
    for entry in paths {
        let path = entry.with_context(|| format!("reading glob match for `{pattern}`"))?;
        out.push(path.to_string_lossy().into_owned());
    }
    anyhow::ensure!(!out.is_empty(), "glob `{pattern}` matched no files");
    out.sort();
    Ok(out)
}

async fn run_rsync(mut cmd: Command, what: &str) -> Result<()> {
    let out = cmd
        .output()
        .await
        .with_context(|| format!("spawning rsync ({what})"))?;
    anyhow::ensure!(
        out.status.success(),
        "rsync {what} failed: {}",
        String::from_utf8_lossy(&out.stderr),
    );
    Ok(())
}

/// The local-machine transport.
pub struct LocalExec {
    host: RemoteHost,
    caps: CapabilitySet,
    pull_root: Option<PathBuf>,
}

impl LocalExec {
    pub fn new(caps: CapabilitySet) -> Self {
        Self {
            host: RemoteHost::new("local"),
            caps,
            pull_root: None,
        }
    }

    /// Confine this transport's artifact pulls to `root` (see [`gate_pull`]).
    /// Required before any `pull_*` succeeds — pulls are fail-closed, so an
    /// executor built without a root refuses every pull.
    #[must_use]
    pub fn with_pull_root(mut self, root: impl Into<PathBuf>) -> Self {
        self.pull_root = Some(root.into());
        self
    }
}

#[async_trait]
impl Executor for LocalExec {
    async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput> {
        gate(&self.caps, "local", step)?;
        validate_env_names(step)?;
        let (cmd, sentinel) = self.host.command_for(&render_shell_line(step));
        run_command_into_sink(cmd, sink, sentinel, "local").await
    }

    async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(&self.caps, "local", self.pull_root.as_deref(), remote)?;
        // No trailing slash: rsync copies the file itself.
        let src = remote.to_string_lossy().to_string();
        run_rsync(
            rsync_command(&src, &local.to_string_lossy(), None, opts),
            "pull_file(local)",
        )
        .await
    }

    async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(&self.caps, "local", self.pull_root.as_deref(), remote)?;
        // Local "pull" is just a local rsync; trailing slash = contents.
        let src = format!("{}/", remote.display());
        run_rsync(
            rsync_command(&src, &local.to_string_lossy(), None, opts),
            "pull_dir(local)",
        )
        .await
    }

    async fn pull_glob(&self, remote_glob: &str, local_dir: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(
            &self.caps,
            "local",
            self.pull_root.as_deref(),
            Path::new(remote_glob),
        )?;
        // No remote shell to expand for us, and `rsync` is spawned directly (no
        // shell), so expand in-process. This is the half of `pull_glob` that
        // differs from the ssh transport, and the reason the trait's docs warn
        // against assuming a shell.
        let matches = expand_glob_locally(remote_glob)?;
        let dst = format!("{}/", local_dir.display());
        run_rsync(
            rsync_command_multi(&matches, &dst, None, opts),
            "pull_glob(local)",
        )
        .await
    }

    async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
        let src = format!("{}/", local.display());
        run_rsync(
            rsync_command(&src, &remote.to_string_lossy(), None, opts),
            "push_dir(local)",
        )
        .await
    }

    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

/// The SSH transport: a tailnet host reached with the existing SSH keys.
pub struct SshExec {
    host: RemoteHost,
    caps: CapabilitySet,
    pull_root: Option<PathBuf>,
}

impl SshExec {
    pub fn new(ssh_target: impl Into<String>, caps: CapabilitySet) -> Self {
        Self {
            host: RemoteHost::new(ssh_target),
            caps,
            pull_root: None,
        }
    }

    /// Reach this host on a non-default SSH port. `None` keeps ssh's default,
    /// so a caller holding an `Option<u16>` can pass it straight through. The
    /// port applies to both the exec and sync paths (see [`RemoteHost`]).
    #[must_use]
    pub fn with_port(mut self, port: Option<u16>) -> Self {
        self.host = self.host.with_port(port);
        self
    }

    /// Confine this transport's artifact pulls to `root` — an absolute path as
    /// seen ON THIS HOST (a remote root, so it is never tilde-expanded by the
    /// caller). See [`gate_pull`]; required before any `pull_*` succeeds.
    #[must_use]
    pub fn with_pull_root(mut self, root: impl Into<PathBuf>) -> Self {
        self.pull_root = Some(root.into());
        self
    }

    pub fn ssh_target(&self) -> &str {
        self.host.ssh_target()
    }
}

#[async_trait]
impl Executor for SshExec {
    async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput> {
        gate(&self.caps, self.host.ssh_target(), step)?;
        validate_env_names(step)?;
        let (cmd, sentinel) = self.host.command_for(&render_shell_line(step));
        run_command_into_sink(cmd, sink, sentinel, self.host.ssh_target()).await
    }

    async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(
            &self.caps,
            self.host.ssh_target(),
            self.pull_root.as_deref(),
            remote,
        )?;
        // No trailing slash: rsync copies the file itself, not "contents of".
        let src = format!("{}:{}", self.host.ssh_target(), remote.display());
        let ssh = Some(self.host.ssh_args());
        run_rsync(
            rsync_command(&src, &local.to_string_lossy(), ssh, opts),
            "pull_file(ssh)",
        )
        .await
    }

    async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(
            &self.caps,
            self.host.ssh_target(),
            self.pull_root.as_deref(),
            remote,
        )?;
        let src = format!("{}:{}/", self.host.ssh_target(), remote.display());
        let ssh = Some(self.host.ssh_args());
        run_rsync(
            rsync_command(&src, &local.to_string_lossy(), ssh, opts),
            "pull_dir(ssh)",
        )
        .await
    }

    async fn pull_glob(&self, remote_glob: &str, local_dir: &Path, opts: &SyncOpts) -> Result<()> {
        gate_pull(
            &self.caps,
            self.host.ssh_target(),
            self.pull_root.as_deref(),
            Path::new(remote_glob),
        )?;
        // The REMOTE shell expands this one: rsync hands an un-`--protect-args`
        // remote path to the login shell on the far side, so the wildcard must
        // reach it intact — hence no quoting here. rsync exits non-zero when the
        // pattern matches nothing, which is the error we want.
        let src = format!("{}:{}", self.host.ssh_target(), remote_glob);
        let dst = format!("{}/", local_dir.display());
        let ssh = Some(self.host.ssh_args());
        run_rsync(rsync_command(&src, &dst, ssh, opts), "pull_glob(ssh)").await
    }

    async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()> {
        let src = format!("{}/", local.display());
        let dst = format!("{}:{}/", self.host.ssh_target(), remote.display());
        let ssh = Some(self.host.ssh_args());
        run_rsync(rsync_command(&src, &dst, ssh, opts), "push_dir(ssh)").await
    }

    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::step::Action;
    use std::sync::Arc;

    #[derive(Default)]
    struct VecSink(Vec<u8>);
    #[async_trait]
    impl LogSink for VecSink {
        async fn write_chunk(&mut self, bytes: &[u8]) {
            self.0.extend_from_slice(bytes);
        }
    }

    fn vec_sink() -> VecSink {
        VecSink::default()
    }

    /// The grant a sync transport needs to pull artifacts: `observe:artifact`
    /// and nothing else. Pair with `.with_pull_root(...)` in the pull tests.
    fn artifact_caps() -> CapabilitySet {
        CapabilitySet::from_tokens(Vec::<&str>::new(), ["artifact"])
    }

    #[test]
    fn render_plain_argv_quotes_each_token() {
        let step = Step::new(Action::Build, ["echo", "a b", "c"]);
        assert_eq!(render_shell_line(&step), "'echo' 'a b' 'c'");
    }

    #[test]
    fn render_shell_script_is_verbatim_with_env_and_cwd() {
        let step = Step::shell(Action::Deploy, "set -e; echo hi")
            .with_env("K", "v v")
            .with_cwd("/tmp/x");
        assert_eq!(
            render_shell_line(&step),
            "cd '/tmp/x' && K='v v' set -e; echo hi"
        );
    }

    #[tokio::test]
    async fn rejects_malicious_env_name_before_dispatch() {
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("pwned");
        let exec = LocalExec::new(CapabilitySet::actuate_only([Action::Build]));
        let mut sink = vec_sink();
        // A shell-metacharacter env name would otherwise inject a command.
        let step = Step::new(Action::Build, ["true"])
            .with_env(format!("X; touch {}", marker.display()), "v");
        let err = exec.run_streaming(&step, &mut sink).await.unwrap_err();
        assert!(err.to_string().contains("shell identifier"));
        assert!(!marker.exists(), "rejected env name must not execute");
    }

    #[tokio::test]
    async fn local_exec_runs_granted_step() {
        let exec = LocalExec::new(CapabilitySet::actuate_only([Action::Deploy]));
        let mut sink = vec_sink();
        let step = Step::shell(Action::Deploy, "printf ok");
        let out = exec.run_streaming(&step, &mut sink).await.unwrap();
        assert!(out.success());
        assert_eq!(sink.0, b"ok");
    }

    #[tokio::test]
    async fn local_exec_denies_ungranted_step_before_dispatch() {
        // Grant deploy only; ask it to sign. Must deny, and must NOT run the
        // command (the file the command would create must not appear).
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("ran");
        let exec = LocalExec::new(CapabilitySet::actuate_only([Action::Deploy]));
        let mut sink = vec_sink();
        let step = Step::shell(Action::Sign, format!("touch {}", marker.display()));
        let err = exec.run_streaming(&step, &mut sink).await.unwrap_err();
        let denied = err
            .downcast_ref::<CapabilityDenied>()
            .expect("CapabilityDenied");
        assert_eq!(denied.action, "sign");
        assert!(!marker.exists(), "denied step must not execute");
    }

    #[tokio::test]
    async fn dyn_executor_object_is_usable() {
        // Prove the trait is object-safe and Arc<dyn Executor> works.
        let exec: Arc<dyn Executor> = Arc::new(LocalExec::new(CapabilitySet::actuate_only([
            Action::Restart,
        ])));
        assert!(exec.capabilities().permits(&Action::Restart));
        let mut sink = vec_sink();
        let out = exec
            .run_streaming(&Step::shell(Action::Restart, "true"), &mut sink)
            .await
            .unwrap();
        assert!(out.success());
    }

    #[tokio::test]
    async fn local_push_and_pull_move_a_dir() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src");
        let mid = dir.path().join("mid");
        let dst = dir.path().join("dst");
        tokio::fs::create_dir_all(&src).await.unwrap();
        tokio::fs::write(src.join("f.txt"), b"hi").await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        exec.push_dir(&src, &mid, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(tokio::fs::read(mid.join("f.txt")).await.unwrap(), b"hi");
        exec.pull_dir(&mid, &dst, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(tokio::fs::read(dst.join("f.txt")).await.unwrap(), b"hi");
    }

    /// The distinction the split exists for: a single file lands AS a file.
    /// Under the old dir-shaped `pull` this path was `{file}/` — rsync would
    /// refuse it.
    #[tokio::test]
    async fn local_pull_file_copies_one_file() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("dump.sql.gz");
        let dst = dir.path().join("fetched.sql.gz");
        tokio::fs::write(&src, b"DUMPBYTES").await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        exec.pull_file(&src, &dst, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(tokio::fs::read(&dst).await.unwrap(), b"DUMPBYTES");
    }

    /// Pins what the file/dir split actually buys, which is NOT a type-level
    /// refusal: `-a` implies `-r`, so `pull_file` pointed at a directory does
    /// rsync's no-trailing-slash thing — copies the dir INTO the destination
    /// (`out/adir/f.txt`) rather than erroring. `pull_dir` would have put the
    /// contents at `out/f.txt`. Documented here because the difference is
    /// silent, and the docs on `pull_file` promise only file→file.
    #[tokio::test]
    async fn local_pull_file_on_a_dir_nests_rather_than_flattening() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("adir");
        let out = dir.path().join("out");
        tokio::fs::create_dir_all(&src).await.unwrap();
        tokio::fs::create_dir_all(&out).await.unwrap();
        tokio::fs::write(src.join("f.txt"), b"hi").await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        exec.pull_file(&src, &out, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(
            tokio::fs::read(out.join("adir").join("f.txt"))
                .await
                .unwrap(),
            b"hi"
        );
        assert!(
            !out.join("f.txt").exists(),
            "pull_file does not flatten; that's pull_dir"
        );
    }

    #[tokio::test]
    async fn local_pull_glob_gathers_matches_and_ignores_the_rest() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("bundle");
        let out = dir.path().join("dist");
        tokio::fs::create_dir_all(&src).await.unwrap();
        tokio::fs::create_dir_all(&out).await.unwrap();
        tokio::fs::write(src.join("a.msi"), b"A").await.unwrap();
        tokio::fs::write(src.join("b.msi"), b"B").await.unwrap();
        tokio::fs::write(src.join("notes.txt"), b"N").await.unwrap();

        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        let pattern = format!("{}/*.msi", src.display());
        exec.pull_glob(&pattern, &out, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(tokio::fs::read(out.join("a.msi")).await.unwrap(), b"A");
        assert_eq!(tokio::fs::read(out.join("b.msi")).await.unwrap(), b"B");
        assert!(
            !out.join("notes.txt").exists(),
            "glob must not gather non-matches"
        );
    }

    /// A collect that silently gathers nothing is how an empty release ships.
    #[tokio::test]
    async fn local_pull_glob_errors_when_nothing_matches() {
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("dist");
        tokio::fs::create_dir_all(&out).await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        let pattern = format!("{}/nope/*.msi", dir.path().display());
        let err = exec
            .pull_glob(&pattern, &out, &SyncOpts::default())
            .await
            .unwrap_err();
        assert!(err.to_string().contains("matched no files"), "{err}");
    }

    /// An exact path is a glob with no wildcard — recipes resolve globs
    /// host-side (`ls -t | head -1`) and pass concrete paths, so this is the
    /// common case, not an edge case.
    #[tokio::test]
    async fn local_pull_glob_accepts_a_concrete_path() {
        let dir = tempfile::tempdir().unwrap();
        let out = dir.path().join("dist");
        tokio::fs::create_dir_all(&out).await.unwrap();
        let f = dir.path().join("GoingsOn.dmg");
        tokio::fs::write(&f, b"DMG").await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(dir.path());
        exec.pull_glob(&f.to_string_lossy(), &out, &SyncOpts::default())
            .await
            .unwrap();
        assert_eq!(
            tokio::fs::read(out.join("GoingsOn.dmg")).await.unwrap(),
            b"DMG"
        );
    }

    /// The exfiltration the gate closes: a pull with no `observe:artifact`
    /// grant is denied before any rsync spawns, even with a root set.
    #[tokio::test]
    async fn pull_denied_without_artifact_grant() {
        let dir = tempfile::tempdir().unwrap();
        let f = dir.path().join("secret");
        tokio::fs::write(&f, b"S").await.unwrap();
        // A build/sign grant is not an artifact-read grant.
        let caps = CapabilitySet::from_tokens(["build", "sign"], ["build-log"]);
        let exec = LocalExec::new(caps).with_pull_root(dir.path());
        let err = exec
            .pull_file(&f, &dir.path().join("out"), &SyncOpts::default())
            .await
            .unwrap_err();
        let denied = err
            .downcast_ref::<CapabilityDenied>()
            .expect("CapabilityDenied so a caller can audit-log it");
        assert_eq!(denied.action, "observe:artifact");
        assert!(!dir.path().join("out").exists(), "denied pull must not run");
    }

    /// Fail-closed: an executor with the grant but NO declared root pulls
    /// nothing. This is what stops an un-configured host from being an open
    /// read primitive.
    #[tokio::test]
    async fn pull_denied_without_pull_root() {
        let dir = tempfile::tempdir().unwrap();
        let f = dir.path().join("a.dmg");
        tokio::fs::write(&f, b"D").await.unwrap();
        let exec = LocalExec::new(artifact_caps()); // no with_pull_root
        let err = exec
            .pull_file(&f, &dir.path().join("out"), &SyncOpts::default())
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("no artifact root declared"),
            "{err}"
        );
    }

    /// The `passwords.env` case: an authorized caller with a legitimate root
    /// still cannot reach a sibling path outside it.
    #[tokio::test]
    async fn pull_denied_outside_declared_root() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("artifacts");
        let secret = dir.path().join("passwords.env"); // sibling of root, not under it
        tokio::fs::create_dir_all(&root).await.unwrap();
        tokio::fs::write(&secret, b"NOTARY_PW=hunter2")
            .await
            .unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(&root);
        let err = exec
            .pull_file(&secret, &dir.path().join("out"), &SyncOpts::default())
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("escapes the declared artifact root")
        );
        assert!(!dir.path().join("out").exists());
    }

    /// A `..` component is refused even when the resolved target would land back
    /// inside the root — the check is lexical, so it never has to resolve it.
    #[tokio::test]
    async fn pull_denied_on_parent_dir_component() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("artifacts");
        tokio::fs::create_dir_all(&root).await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(&root);
        let sneaky = root.join("..").join("passwords.env");
        let err = exec
            .pull_file(&sneaky, &dir.path().join("out"), &SyncOpts::default())
            .await
            .unwrap_err();
        assert!(err.to_string().contains("contains `..`"), "{err}");
    }

    /// `starts_with` is component-wise, so a root prefix that is a string prefix
    /// but NOT a path prefix (`/x/artifacts` vs `/x/artifacts-evil`) does not
    /// leak. Pins that the confinement isn't a naive string compare.
    #[tokio::test]
    async fn pull_root_is_a_path_prefix_not_a_string_prefix() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("artifacts");
        let evil = dir.path().join("artifacts-evil");
        tokio::fs::create_dir_all(&root).await.unwrap();
        tokio::fs::create_dir_all(&evil).await.unwrap();
        let f = evil.join("x.dmg");
        tokio::fs::write(&f, b"D").await.unwrap();
        let exec = LocalExec::new(artifact_caps()).with_pull_root(&root);
        let err = exec
            .pull_file(&f, &dir.path().join("out"), &SyncOpts::default())
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("escapes the declared artifact root")
        );
    }

    #[test]
    fn ssh_pull_glob_leaves_the_wildcard_for_the_remote_shell() {
        // rsync hands an un---protect-args remote path to the far-side login
        // shell; quoting it here would defeat the expansion this depends on.
        let host = RemoteHost::new("windows-x86");
        let cmd = rsync_command(
            &format!("{}:{}", host.ssh_target(), "/c/build/bundle/msi/*.msi"),
            "/dist/",
            Some(host.ssh_args()),
            &SyncOpts::default(),
        );
        let args = render_args(&cmd);
        assert!(
            args.iter()
                .any(|a| a == "windows-x86:/c/build/bundle/msi/*.msi"),
            "wildcard reaches the remote intact: {args:?}"
        );
    }

    #[test]
    fn sync_opts_flags_are_opt_out() {
        // Default = the historical `-az --partial`.
        let cmd = rsync_command("s", "d", None, &SyncOpts::default());
        let args = render_args(&cmd);
        assert!(
            args.contains(&"-z".to_string()),
            "compress on by default: {args:?}"
        );
        assert!(
            args.contains(&"--partial".to_string()),
            "partial on by default: {args:?}"
        );

        // A precompressed payload drops -z but keeps --partial.
        let args = render_args(&rsync_command("s", "d", None, &SyncOpts::precompressed()));
        assert!(
            !args.contains(&"-z".to_string()),
            "precompressed drops -z: {args:?}"
        );
        assert!(args.contains(&"--partial".to_string()));

        // Sando's backup fetch drops both: a resumed dump could splice two
        // different backups (CF4).
        let opts = SyncOpts {
            compress: false,
            partial: false,
            ..SyncOpts::default()
        };
        let args = render_args(&rsync_command("s", "d", None, &opts));
        assert!(!args.contains(&"-z".to_string()));
        assert!(!args.contains(&"--partial".to_string()));

        // release_mirror keeps prune + chmod, and still compresses.
        let args = render_args(&rsync_command("s", "d", None, &SyncOpts::release_mirror()));
        assert!(args.contains(&"--delete".to_string()));
        assert!(args.iter().any(|a| a.starts_with("--chmod=")));
        assert!(args.contains(&"-z".to_string()));

        // An archive deposit creates its destination tree (the first build of a
        // version is what makes that directory exist) and never prunes it: a
        // second target of the same release deposits beside the first.
        let args = render_args(&rsync_command("s", "d", None, &SyncOpts::archive_deposit()));
        assert!(args.contains(&"--mkpath".to_string()), "{args:?}");
        assert!(!args.contains(&"--delete".to_string()), "{args:?}");
        assert!(!args.contains(&"-z".to_string()), "{args:?}");

        // ...and nothing else creates directories implicitly: a typo'd
        // destination for every other caller stays an error.
        let args = render_args(&rsync_command("s", "d", None, &SyncOpts::default()));
        assert!(!args.contains(&"--mkpath".to_string()), "{args:?}");
    }

    /// Every exclude reaches rsync as its own `--exclude=`, and none is emitted
    /// when the caller asked for none. A pattern lost here would ship a file the
    /// caller meant to leave behind, which for the handoff case is the document
    /// that changes the digest of what it describes.
    #[test]
    fn excludes_are_passed_through_one_flag_each() {
        let opts = SyncOpts {
            exclude: vec!["record.json".into(), "*.log".into()],
            ..SyncOpts::default()
        };
        let args = render_args(&rsync_command("s", "d", None, &opts));
        assert!(
            args.contains(&"--exclude=record.json".to_string()),
            "{args:?}"
        );
        assert!(args.contains(&"--exclude=*.log".to_string()), "{args:?}");

        let args = render_args(&rsync_command("s", "d", None, &SyncOpts::default()));
        assert!(!args.iter().any(|a| a.starts_with("--exclude")), "{args:?}");
    }

    /// End to end: an excluded file stays out of the destination while its
    /// siblings arrive. The unit test above proves the flag is built; this
    /// proves rsync honors it for the shape the handoff actually uses.
    #[tokio::test]
    async fn an_excluded_file_does_not_reach_the_destination() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        std::fs::write(src.join("demo.bin"), b"bytes").unwrap();
        std::fs::write(src.join("record.json"), b"{}").unwrap();

        let dst = dir.path().join("staged");
        let exec = LocalExec::new(CapabilitySet::from_tokens::<[&str; 0], [&str; 0]>([], []));
        exec.push_dir(
            &src,
            &dst,
            &SyncOpts {
                exclude: vec!["record.json".into()],
                ..SyncOpts::archive_deposit()
            },
        )
        .await
        .unwrap();
        assert!(dst.join("demo.bin").exists());
        assert!(
            !dst.join("record.json").exists(),
            "the excluded document must not land in the bundle"
        );
    }

    /// `--mkpath` end to end: a push into a destination whose parents do not
    /// exist creates them, which is the whole reason Bento's archive can name a
    /// per-`(app, version, target)` path before that version has ever built.
    #[tokio::test]
    async fn a_deposit_creates_its_missing_destination_tree() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        std::fs::write(src.join("demo.bin"), b"bytes").unwrap();

        let dst = dir.path().join("archive/demo/0.0.1/linux-x86_64");
        let exec = LocalExec::new(CapabilitySet::from_tokens::<[&str; 0], [&str; 0]>([], []));
        exec.push_dir(&src, &dst, &SyncOpts::archive_deposit())
            .await
            .unwrap();
        assert_eq!(std::fs::read(dst.join("demo.bin")).unwrap(), b"bytes");
    }

    #[test]
    fn rsync_over_ssh_carries_the_port_into_dash_e() {
        let host = RemoteHost::new("backup@db.example").with_port(Some(2222));
        let cmd = rsync_command("src", "dst", Some(host.ssh_args()), &SyncOpts::default());
        let args = render_args(&cmd);
        let e = args.iter().position(|a| a == "-e").expect("-e present");
        let ssh_spec = &args[e + 1];
        assert!(
            ssh_spec.contains("-p 2222"),
            "port reaches rsync's ssh: {ssh_spec}"
        );
        assert!(
            ssh_spec.contains("BatchMode=yes"),
            "shared flags kept: {ssh_spec}"
        );

        // No port set ⇒ no -p at all (ssh/ssh_config decides).
        let plain = RemoteHost::new("mbp");
        let args = render_args(&rsync_command(
            "s",
            "d",
            Some(plain.ssh_args()),
            &SyncOpts::default(),
        ));
        let e = args.iter().position(|a| a == "-e").unwrap();
        assert!(
            !args[e + 1].contains("-p"),
            "no port ⇒ no -p: {}",
            args[e + 1]
        );
    }

    fn render_args(cmd: &Command) -> Vec<String> {
        cmd.as_std()
            .get_args()
            .map(|a| a.to_string_lossy().to_string())
            .collect()
    }
}