//! Tests for [`super`].

use super::*;
use crate::config::Config;
use crate::ota::OtaRegistry;
use crate::topology::Topology;
use async_trait::async_trait;
use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, SyncOpts};
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::os::unix::process::ExitStatusExt;
use std::sync::Arc;
use tokio::sync::Mutex;

/// A no-transport [`Executor`] for the paths that don't need a real command
/// to run: its `preflight` is programmable (the agent host hits `/health`
/// there, and a dead `ops-agent` must fail the target before the recipe
/// dispatches), and every actual op is a success no-op.
struct FakeExec {
    caps: CapabilitySet,
    preflight_err: Option<String>,
}

impl FakeExec {
    fn preflight_fails(msg: &str) -> Arc<dyn Executor> {
        Arc::new(Self {
            caps: CapabilitySet::default(),
            preflight_err: Some(msg.to_string()),
        })
    }
}

#[async_trait]
impl Executor for FakeExec {
    async fn run_streaming(
        &self,
        _step: &ops_exec::Step,
        _sink: &mut dyn LogSink,
    ) -> anyhow::Result<RunOutput> {
        Ok(RunOutput {
            status: std::process::ExitStatus::from_raw(0),
            stdout: Vec::new(),
            stderr: Vec::new(),
        })
    }
    async fn pull_file(
        &self,
        _r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn pull_dir(
        &self,
        _r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn pull_glob(&self, _g: &str, _l: &std::path::Path, _o: &SyncOpts) -> anyhow::Result<()> {
        Ok(())
    }
    async fn push_dir(
        &self,
        _l: &std::path::Path,
        _r: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn preflight(&self) -> anyhow::Result<()> {
        match &self.preflight_err {
            Some(m) => anyhow::bail!("{m}"),
            None => Ok(()),
        }
    }
    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

/// A recording, programmable [`Executor`] for the macOS sign chain. Every
/// dispatched shell command is captured for assertion, and its exit code +
/// stdout is chosen by the first rule whose needle the command contains — so
/// a test can make `notarytool` report `Accepted`, make `codesign` fail, or
/// make `spctl` emit the Gatekeeper sentinel without a real Mac or SSH. A
/// rule may carry a *sequence* of responses (one per successive match) to
/// drive the notarize retry loop; the last entry repeats once the sequence
/// is exhausted. Unmatched commands succeed as empty no-ops (so a plain
/// build `sh_ok` passes), and the sync/preflight ops are no-ops.
struct ScriptedExec {
    caps: CapabilitySet,
    rules: Vec<ScriptRule>,
    log: Arc<std::sync::Mutex<Vec<String>>>,
}

struct ScriptRule {
    needle: String,
    responses: Vec<(i32, String)>,
    calls: std::sync::atomic::AtomicUsize,
}

impl ScriptedExec {
    fn new() -> Self {
        Self {
            // A mac host's real grant. Nothing in the dispatch path gates on
            // it (this fake never calls `gate`), but keep it coherent so
            // `capabilities()` is not a lie.
            caps: CapabilitySet::from_tokens(
                ["build", "sign", "notarize", "staple"],
                ["build-log", "artifact"],
            ),
            rules: Vec::new(),
            log: Arc::new(std::sync::Mutex::new(Vec::new())),
        }
    }

    /// Respond to every command containing `needle` with `(code, stdout)`.
    fn on(mut self, needle: &str, code: i32, stdout: &str) -> Self {
        self.rules.push(ScriptRule {
            needle: needle.to_string(),
            responses: vec![(code, stdout.to_string())],
            calls: std::sync::atomic::AtomicUsize::new(0),
        });
        self
    }

    /// Respond to successive `needle` matches with successive responses; the
    /// last repeats once the list is exhausted. Drives the notarize retry.
    fn on_seq(mut self, needle: &str, responses: &[(i32, &str)]) -> Self {
        self.rules.push(ScriptRule {
            needle: needle.to_string(),
            responses: responses
                .iter()
                .map(|(c, s)| (*c, (*s).to_string()))
                .collect(),
            calls: std::sync::atomic::AtomicUsize::new(0),
        });
        self
    }

    /// Every shell command this executor was asked to run, in order.
    fn commands(&self) -> Vec<String> {
        self.log.lock().unwrap().clone()
    }
}

#[async_trait]
impl Executor for ScriptedExec {
    async fn run_streaming(
        &self,
        step: &ops_exec::Step,
        _sink: &mut dyn LogSink,
    ) -> anyhow::Result<RunOutput> {
        let cmd = step.argv.last().cloned().unwrap_or_default();
        self.log.lock().unwrap().push(cmd.clone());
        let (code, stdout) =
            self.rules
                .iter()
                .find(|r| cmd.contains(&r.needle))
                .map_or((0, String::new()), |r| {
                    let i = r.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    r.responses[i.min(r.responses.len() - 1)].clone()
                });
        Ok(RunOutput {
            // Shift into the wait-status word's exit-code byte so
            // `ExitStatus::code()` reports `code` exactly (a bare
            // `from_raw(1)` reads as a signal, yielding `None`).
            status: std::process::ExitStatus::from_raw(code << 8),
            stdout: stdout.into_bytes(),
            stderr: Vec::new(),
        })
    }
    async fn pull_file(
        &self,
        _r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn pull_dir(
        &self,
        _r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn pull_glob(&self, _g: &str, _l: &std::path::Path, _o: &SyncOpts) -> anyhow::Result<()> {
        Ok(())
    }
    async fn push_dir(
        &self,
        _l: &std::path::Path,
        _r: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn preflight(&self) -> anyhow::Result<()> {
        Ok(())
    }
    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

/// Assemble an [`AppState`] from the three per-test inputs, filling in the
/// executors/syncs (built from `topo`) and the fixed test scaffolding
/// (metrics handle, event bus, standard OTA registry, empty active map, no
/// token). Every runner test builds the same struct around a different repo
/// + recipe; this is that struct in one place.
fn test_state(pool: SqlitePool, topo: Topology, cfg: Config) -> AppState {
    let executors = Arc::new(crate::state::build_executors(&topo));
    let syncs = Arc::new(crate::state::build_syncs(&topo));
    let host_locks = crate::state::build_host_locks(&topo);
    AppState {
        pool,
        topo: Arc::new(topo),
        cfg: Arc::new(cfg),
        prom: crate::metrics::test_handle(),
        events: crate::events::channel(),
        ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
        executors,
        syncs,
        active: Arc::new(Mutex::new(HashMap::new())),
        api_token: None,
        host_locks,
        distribution: Arc::new(Mutex::new(HashMap::new())),
        http: crate::tls::builder().build().unwrap(),
        // Port 1 refuses instantly. A test must never probe production, and
        // a refusal is also faster than any timeout would be.
        mnw_base_url: "http://127.0.0.1:1".into(),
    }
}

/// A `kind = "service"` release end to end: build, `glibc_check`, `deploy`,
/// and a health assertion the recipe makes itself against the service host.
///
/// The whole point of the deploy step is that it dispatches to a host
/// OUTSIDE the build topology, so this exercises the resolution
/// (target -> `[[deploy]]` entry -> executor registered for the run), the
/// staging, and the call into the privileged installer — with a fake
/// installer standing in for the root script, which is the one part a test
/// cannot run for real.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn service_recipe_builds_then_deploys_and_verifies() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("svc");
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    std::fs::write(repo.join("Cargo.toml"), "[package]\nversion = \"0.4.0\"\n").unwrap();

    // Stand-in for the root installer: same three arguments, records what it
    // was asked to do instead of writing to /usr/local/bin and restarting a
    // unit. `install` + a marker file, so the test can assert the binary
    // that arrived is the binary that was built.
    let installer = root.join("install-service.sh");
    std::fs::write(
        &installer,
        "#!/bin/sh\nset -eu\ninstall -m 0755 \"$1\" \"$2\"\necho \"restarted $3\" >> \"$2.log\"\n",
    )
    .unwrap();
    std::fs::set_permissions(
        &installer,
        <std::fs::Permissions as std::os::unix::fs::PermissionsExt>::from_mode(0o755),
    )
    .unwrap();
    let install_path = root.join("bin/svc");
    std::fs::create_dir_all(root.join("bin")).unwrap();

    std::fs::write(
            repo.join("dist/recipes/linux.rhai"),
            r#"
                let v = version();
                step("build");
                sh_ok("fw13", "mkdir -p REPO/target/release && echo built-BIN > REPO/target/release/svc");
                step("verify");
                log(glibc_check("REPO/target/release/svc"));
                step("deploy");
                log(deploy("REPO/target/release/svc"));
                // The recipe owns what "healthy" means, and asserts it itself
                // against the host it just restarted.
                sh_ok(deploy_host(), "test -x " + install_path());
            "#
            .replace("REPO", repo.to_str().unwrap())
            .replace("BIN", "0.4.0"),
        )
        .unwrap();

    std::fs::write(
        repo.join("bento.toml"),
        format!(
            r#"kind = "service"
targets = ["linux/x86_64"]
version_path = "Cargo.toml"

[[deploy]]
target = "linux/x86_64"
host = "local"
install_path = "{}"
service = "svc.service"
health_url = "http://localhost:9100/api/health"
"#,
            install_path.display()
        ),
    )
    .unwrap();

    let mut cfg = Config::for_tests(root);
    cfg.deploy_installer = installer.display().to_string();
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             pull_root = \"{repo}\"\n\n[app.svc]\nrepo = \"{repo}\"\n",
        repo = repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);

    let build_id = start_build(
        state.clone(),
        AppId::new("svc"),
        Version::parse("0.4.0").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    for _ in 0..100 {
        status = sqlx::query_scalar("SELECT status FROM target_runs WHERE build_id = ?")
            .bind(build_id)
            .fetch_optional(&pool)
            .await
            .unwrap()
            .unwrap_or_else(|| "running".to_string());
        if status != "running" {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(status, "ok", "service run should succeed");

    let steps: Vec<(String, String)> = sqlx::query_as(
        "SELECT step, status FROM step_runs WHERE target_run_id IN \
             (SELECT id FROM target_runs WHERE build_id = ?) ORDER BY id",
    )
    .bind(build_id)
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(
        steps.iter().map(|(s, _)| s.as_str()).collect::<Vec<_>>(),
        vec!["build", "verify", "deploy"],
        "a service ends at deploy, not collect"
    );
    assert!(steps.iter().all(|(_, st)| st == "ok"), "{steps:?}");

    // The bytes that were built are the bytes that landed, and the unit was
    // restarted only after the install succeeded.
    assert_eq!(
        std::fs::read_to_string(&install_path).unwrap().trim(),
        "built-0.4.0"
    );
    assert!(
        std::fs::read_to_string(install_path.with_extension("").with_file_name("svc.log"))
            .unwrap()
            .contains("restarted svc.service")
    );
}

/// Stand up a tmp app repo + topology and run a real local recipe end to
/// end: step transitions, streamed `sh_ok`, `version_of`, `log`, and a
/// `collect` that pulls a built artifact into dist_root.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn local_linux_recipe_runs_end_to_end() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    // Fake app checkout: tauri.conf.json + a linux recipe.
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    // Build writes an artifact into the repo; collect pulls it to dist_root.
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                let v = version_of("demo");
                log("building demo " + v);
                sh_ok("fw13", "echo compiling; mkdir -p REPO/out && echo bin > REPO/out/demo.bin");
                step("collect");
                collect("fw13", "REPO/out/demo.bin", "demo", v);
            "#
        .replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();

    let mut cfg = Config::for_tests(root);
    // Archive to a local directory, so the deposit a real release makes to
    // astra runs on the same code path here.
    cfg.archive = Some(crate::config::Archive {
        host: "local".into(),
        root: root.join("archive"),
    });
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]
pull_root = "{repo}"

[app.demo]
repo = "{repo}"
"#,
        repo = repo.display()
    ))
    .unwrap();

    let state = test_state(pool.clone(), topo, cfg);

    let app = AppId::new("demo");
    let version = Version::parse("0.0.1").unwrap();
    let build_id = start_build(
        state.clone(),
        app,
        version,
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    // Wait for the target run to settle. The row may not be inserted on the
    // first poll (run_target is spawned, not awaited), so treat a missing row
    // as still-pending rather than an error.
    let mut status = String::new();
    for _ in 0..100 {
        status = sqlx::query_scalar("SELECT status FROM target_runs WHERE build_id = ?")
            .bind(build_id)
            .fetch_optional(&pool)
            .await
            .unwrap()
            .unwrap_or_else(|| "running".to_string());
        if status != "running" {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(status, "ok", "target run should succeed");

    // Both steps recorded and finished ok.
    let steps: Vec<(String, String)> =
            sqlx::query_as("SELECT step, status FROM step_runs WHERE target_run_id IN (SELECT id FROM target_runs WHERE build_id = ?) ORDER BY id")
                .bind(build_id)
                .fetch_all(&pool)
                .await
                .unwrap();
    let names: Vec<&str> = steps.iter().map(|(s, _)| s.as_str()).collect();
    assert_eq!(names, vec!["build", "collect"]);
    assert!(steps.iter().all(|(_, st)| st == "ok"));

    // Artifact landed in dist_root, under its own target, and a step log was
    // written. Both trees are keyed the same way, `<app>/<version>/<target>/`.
    let artifact = state.cfg.dist_root.join("demo/0.0.1/linux-x86_64/demo.bin");
    assert!(artifact.exists(), "collect should copy the artifact");
    // ...and the same bytes reached the archive, at the same path under its
    // own root. This is the answer to "where is demo 0.0.1 for linux".
    let archived = root.join("archive/demo/0.0.1/linux-x86_64/demo.bin");
    assert!(
        archived.exists(),
        "collect should deposit into the archive: {}",
        archived.display()
    );
    assert_eq!(
        std::fs::read(&archived).unwrap(),
        std::fs::read(&artifact).unwrap()
    );
    // Read the path off the ledger rather than rebuilding it: the log is
    // named for its step run id, and the point of that is that the row
    // resolves to exactly one file.
    let (run_id, log_ref): (i64, String) = sqlx::query_as(
        "SELECT id, log_ref FROM step_runs WHERE step = 'build' AND target_run_id IN \
             (SELECT id FROM target_runs WHERE build_id = ?)",
    )
    .bind(build_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    let log = std::path::PathBuf::from(&log_ref);
    assert_eq!(
        log,
        state
            .cfg
            .logs_root
            .join(format!("demo/0.0.1/linux-x86_64/build.{run_id}.log")),
        "log path should be keyed on the step run id"
    );
    assert!(log.exists(), "build step log should exist");
    let body = std::fs::read_to_string(&log).unwrap();
    assert!(body.contains("compiling"));
    assert!(
        body.starts_with(&format!(
            "=== bento demo 0.0.1 linux/x86_64 step=build run_id={run_id} "
        )),
        "log should open with a run header naming its run: {body}"
    );
}

/// Multi-target fan-out, which is what the daemon exists for: every other
/// runner test drives exactly one target, so nothing covered `start_build`'s
/// `JoinSet` fan-out or `finalize_build`'s rollup. One target fails and one
/// succeeds — the failure must not abort its sibling, both must land their
/// own terminal row, and the build must finalize `failed` because any failed
/// target fails the build.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn multi_target_fan_out_rolls_up_partial_failure() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();

    // Linux succeeds and collects a real artifact. The artifact is the proof
    // this target ran to completion rather than being torn down when its
    // sibling failed.
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                let v = version_of("demo");
                sh_ok("fw13", "mkdir -p REPO/out && echo bin > REPO/out/demo.bin");
                step("collect");
                collect("fw13", "REPO/out/demo.bin", "demo", v);
            "#
        .replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();
    // Windows fails in its first step.
    std::fs::write(
        repo.join("dist/recipes/windows.rhai"),
        r#"
                step("build");
                sh_ok("winbox", "echo nope 1>&2; exit 1");
            "#,
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(
        repo.join("bento.toml"),
        "targets = [\"linux/x86_64\", \"windows/x86_64\"]\n",
    )
    .unwrap();
    // Two hosts so each target resolves its own, mirroring the real
    // per-architecture topology.
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]
pull_root = "{repo}"

[[host]]
name = "winbox"
ssh = "local"
targets = ["windows/x86_64"]

[app.demo]
repo = "{repo}"
"#,
        repo = repo.display()
    ))
    .unwrap();

    let state = test_state(pool.clone(), topo, cfg);

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec![
            "linux/x86_64".parse().unwrap(),
            "windows/x86_64".parse().unwrap(),
        ],
    )
    .await
    .unwrap();

    // Poll the build row, not the target rows: the build is terminal only
    // once finalize_build has joined every task and stamped it.
    let mut build_status = String::new();
    for _ in 0..200 {
        build_status = sqlx::query_scalar("SELECT status FROM builds WHERE id = ?")
            .bind(build_id)
            .fetch_one(&pool)
            .await
            .unwrap();
        if build_status != "running" {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    let runs: Vec<(String, String)> =
        sqlx::query_as("SELECT target, status FROM target_runs WHERE build_id = ? ORDER BY target")
            .bind(build_id)
            .fetch_all(&pool)
            .await
            .unwrap();
    assert_eq!(
        runs,
        vec![
            ("linux/x86_64".to_string(), "ok".to_string()),
            ("windows/x86_64".to_string(), "failed".to_string()),
        ],
        "each target lands its own terminal row; one failing does not take the other down",
    );

    // The surviving target finished its work, not merely its row.
    assert!(
        state
            .cfg
            .dist_root
            .join("demo/0.0.1/linux-x86_64/demo.bin")
            .exists(),
        "the succeeding target ran to completion and collected its artifact",
    );

    // The failure is attributed to the target that failed, and only it.
    let err: Option<String> = sqlx::query_scalar(
        "SELECT error FROM target_runs WHERE build_id = ? AND target = 'windows/x86_64'",
    )
    .bind(build_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(
        err.is_some_and(|e| !e.is_empty()),
        "a failed target records why",
    );

    assert_eq!(
        build_status, "failed",
        "any failed target fails the build; a partial release must not read as ok",
    );
    let finished: Option<String> =
        sqlx::query_scalar("SELECT finished_at FROM builds WHERE id = ?")
            .bind(build_id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(finished.is_some(), "finalize_build stamps the finish time");

    // finalize_build reaps its own latest-wins slots, so nothing is left
    // in flight to block a later build of the same targets.
    assert!(
        state.active.lock().await.is_empty(),
        "finalize_build reaps the slots it owned",
    );
}

/// Latest-wins supersession, driven through `start_build` (not the leaf
/// `begin_step` bail that's already covered). Two builds of the SAME
/// (app, target) race: the second must cooperatively cancel + abort the
/// first and take the single slot, the first must terminate non-`ok`, and
/// the second must run to completion.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_newer_build_supersedes_the_in_flight_one_for_the_same_target() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    // A sleep long enough that the first build is still mid-recipe when the
    // second arrives; the engine checks the cancel flag at the step boundary
    // before "done", so the superseded run bails rather than finishing.
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("fw13", "sleep 2");
            "#,
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[app.demo]
repo = "{}"
"#,
        repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);

    let target = "linux/x86_64".parse().unwrap();
    let first = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec![target],
    )
    .await
    .unwrap();
    // Let the first build register its slot and enter the recipe before the
    // second supersedes it (start_build inserts the slot synchronously, but
    // the recipe runs on a spawned task).
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    let second = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec![target],
    )
    .await
    .unwrap();
    assert_ne!(first, second);

    // Latest wins: exactly one slot for the key, owned by the second build.
    {
        let active = state.active.lock().await;
        assert_eq!(active.len(), 1, "supersession must not leave two slots");
        assert_eq!(
            active.values().next().unwrap().build_id,
            second,
            "the surviving slot belongs to the newer build",
        );
    }

    // Wait for the second (surviving) build to finalize.
    let mut second_status = String::new();
    for _ in 0..200 {
        second_status = sqlx::query_scalar("SELECT status FROM builds WHERE id = ?")
            .bind(second)
            .fetch_one(&pool)
            .await
            .unwrap();
        if second_status != "running" {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(
        second_status, "ok",
        "the superseding build runs to completion"
    );

    // The first build's target run terminated without succeeding — it was
    // cancelled/aborted, never stamped `ok`.
    let first_status: String =
        sqlx::query_scalar("SELECT status FROM target_runs WHERE build_id = ?")
            .bind(first)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_ne!(
        first_status, "ok",
        "the superseded build must not complete successfully",
    );

    // Both builds reaped their slots; nothing left in flight.
    assert!(
        state.active.lock().await.is_empty(),
        "every finalized build reaps its own slot",
    );
}

/// A failing `preflight` (the agent host's `/health` probe when `ops-agent`
/// is down) must fail the target BEFORE the recipe dispatches — the whole
/// point of preflighting. Covered only via a fake here, since a real
/// LocalExec/SshExec preflight is a no-op that always passes.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_failing_preflight_fails_the_target_before_the_recipe_runs() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    // If the recipe ran it would drop this marker; preflight failing first
    // means it never does.
    let marker = root.join("recipe-ran");
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("fw13", "touch MARKER");
            "#
        .replace("MARKER", marker.to_str().unwrap()),
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[app.demo]
repo = "{}"
"#,
        repo.display()
    ))
    .unwrap();
    let mut state = test_state(pool.clone(), topo, cfg);
    // Swap fw13's executor for one whose preflight fails.
    let mut execs = HashMap::new();
    execs.insert(
        "fw13".to_string(),
        FakeExec::preflight_fails("ops-agent not reachable at /health"),
    );
    state.executors = Arc::new(execs);

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(status, "failed", "a failed preflight must fail the target");
    assert!(
        error.contains("ops-agent not reachable"),
        "the failure must carry the preflight error, got: {error}"
    );
    assert!(
        !marker.exists(),
        "the recipe must NOT run when preflight fails"
    );
}

/// A target whose recipe file is absent must fail at the `checkout` boundary
/// with a readable "reading recipe" error, before any step runs — exercising
/// `read_recipe`'s error branch and the `fail_target` path in `run_target`
/// that neither the happy-path nor the failing-preflight test reaches. The
/// app is configured to ship linux, but no `linux.rhai` is written.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_target_with_no_recipe_file_fails_at_checkout() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    // The recipe dir exists but is empty — no linux.rhai.
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[app.demo]
repo = "{}"
"#,
        repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    // Poll the build row: it is terminal only once finalize_build joins the
    // one (failing) target task.
    let mut build_status = String::new();
    for _ in 0..100 {
        build_status = sqlx::query_scalar("SELECT status FROM builds WHERE id = ?")
            .bind(build_id)
            .fetch_one(&pool)
            .await
            .unwrap();
        if build_status != "running" {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    // The target failed, and the error points at the unreadable recipe.
    let (status, error): (String, Option<String>) =
        sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
            .bind(build_id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(status, "failed", "a missing recipe must fail the target");
    assert!(
        error.is_some_and(|e| e.contains("reading recipe")),
        "the failure must name the recipe it could not read",
    );

    // The recipe never ran, so no step_runs row was ever created — the
    // failure is at the checkout boundary, ahead of any step.
    let step_count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM step_runs WHERE target_run_id IN \
             (SELECT id FROM target_runs WHERE build_id = ?)",
    )
    .bind(build_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        step_count, 0,
        "no step should run when the recipe is absent"
    );

    assert_eq!(
        build_status, "failed",
        "the build rolls up the target failure",
    );
    assert!(
        state.active.lock().await.is_empty(),
        "finalize_build reaps the slot even on the recipe-read failure path",
    );
}

/// Run 2 S5: a publish whose version is not strictly newer than the latest
/// already published for the same (app, target, channel) is refused — an
/// older build cannot republish over a live newer release.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn publish_rejects_a_non_monotonic_version() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.2.0"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    let artifact = repo.join("out/app.tar.gz");
    // Build an artifact, publish 0.2.0 (records a release), then try to
    // publish the older 0.1.0 — the second publish must fail.
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("fw13", "mkdir -p REPO/out && echo bin > ARTIFACT");
                step("publish");
                publish("tauri-mnw", "demo", "linux/x86_64", "0.2.0", "ARTIFACT", #{});
                publish("tauri-mnw", "demo", "linux/x86_64", "0.1.0", "ARTIFACT", #{});
            "#
        .replace("ARTIFACT", artifact.to_str().unwrap())
        .replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[app.demo]
repo = "{}"
"#,
        repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.2.0").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(status, "failed", "non-monotonic publish must fail the run");
    assert!(
        error.contains("not newer"),
        "expected monotonicity error, got: {error}"
    );
    // Exactly one release was recorded (0.2.0); 0.1.0 never landed.
    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM releases")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        count, 1,
        "only the first (newer) publish should record a release"
    );
    // The recorded release carries the artifact's sha256 (64 hex chars),
    // not a NULL — the ledger says which bytes shipped.
    let hash: Option<String> =
        sqlx::query_scalar("SELECT artifact_hash FROM releases WHERE version = '0.2.0'")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        hash.as_deref().is_some_and(|h| h.len() == 64),
        "publish must record the artifact sha256, got {hash:?}"
    );
}

/// The artifact-identity fix at collect: a stale, wrongly-versioned artifact
/// left in the output dir fails the collect instead of silently winning a
/// later glob. Here the build is 0.0.1 but the recipe produces a 9.9.9 file.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn collect_rejects_a_stale_versioned_artifact() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                let v = version_of("demo");
                sh_ok("fw13", "mkdir -p REPO/out && echo bin > REPO/out/demo-9.9.9.bin");
                step("collect");
                collect("fw13", "REPO/out/demo-9.9.9.bin", "demo", v);
            "#
        .replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             pull_root = \"{repo}\"\n\n[app.demo]\nrepo = \"{repo}\"\n",
        repo = repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(
        status, "failed",
        "a mismatched-version artifact must fail collect"
    );
    assert!(
        error.contains("stale artifact"),
        "expected a stale-artifact error, got: {error}"
    );
}

/// A command that runs past its step's deadline fails THAT step and unwinds
/// the recipe. The per-step budget is overridden to 1s here; the recipe then
/// sleeps 30s, so the step deadline, not the sleep, decides the outcome.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_step_that_exceeds_its_deadline_fails() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("fw13", "sleep 30");
            "#,
    )
    .unwrap();

    let mut cfg = Config::for_tests(root);
    cfg.step_timeout_secs = Some(1); // every step's budget -> 1s
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             [app.demo]\nrepo = \"{}\"\n",
        repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let started = std::time::Instant::now();
    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(status, "failed", "a step past its deadline must fail");
    assert!(
        error.contains("per-step deadline"),
        "expected a deadline error, got: {error}"
    );
    // The deadline (1s), not the 30s sleep, decided it — proof the command
    // was actually interrupted rather than run to completion.
    assert!(
        started.elapsed() < std::time::Duration::from_secs(20),
        "the step deadline must fire well before the sleep would finish"
    );
}

/// A demo app that publishes linux and has the all-targets-green gate on; it
/// declares linux + macos, so publishing linux is gated on macos being green.
async fn gate_state(root: &std::path::Path) -> (AppState, sqlx::SqlitePool) {
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.2.0"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    let artifact = repo.join("out/app.tar.gz");
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("fw13", "mkdir -p REPO/out && echo bin > ARTIFACT");
                step("publish");
                publish("tauri-mnw", "demo", "linux/x86_64", "0.2.0", "ARTIFACT", #{});
            "#
        .replace("ARTIFACT", artifact.to_str().unwrap())
        .replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();
    std::fs::write(
        repo.join("bento.toml"),
        "targets = [\"linux/x86_64\", \"macos/aarch64\"]\nrequire_all_targets = true\n",
    )
    .unwrap();
    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             [[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
             [app.demo]\nrepo = \"{}\"\n",
        repo.display()
    ))
    .unwrap();
    (test_state(pool.clone(), topo, cfg), pool)
}

async fn await_target(pool: &sqlx::SqlitePool, build_id: i64) -> (String, String) {
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(pool)
                .await
                .unwrap();
        if let Some((s, e)) = row
            && s != "running"
        {
            return (s, e.unwrap_or_default());
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    panic!("target never settled");
}

/// The all-targets-green gate blocks a partial release: linux tries to
/// publish while macos has no successful run, so publish is refused and the
/// target fails at its publish step.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn all_green_gate_blocks_publish_when_a_sibling_is_not_green() {
    let tmp = tempfile::tempdir().unwrap();
    let (state, pool) = gate_state(tmp.path()).await;
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.2.0").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "failed", "a partial release must be blocked");
    assert!(
        error.contains("all-targets-green gate"),
        "expected the gate to name itself, got: {error}"
    );
    // Nothing shipped.
    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM releases")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 0, "a gated-off publish records no release");
}

/// With every sibling green, the gate lets the publish through. macos is
/// pre-recorded `ok` for this version, so linux's publish proceeds.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn all_green_gate_allows_publish_when_every_sibling_is_green() {
    let tmp = tempfile::tempdir().unwrap();
    let (state, pool) = gate_state(tmp.path()).await;
    // Pre-record a green macos run at 0.2.0 (as if its target already built).
    let bid: i64 = sqlx::query_scalar(
            "INSERT INTO builds (app, version, status, created_at) VALUES ('demo','0.2.0','ok','2026-07-23T00:00:00Z') RETURNING id",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
    sqlx::query(
        "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
             VALUES (?, 'demo', '0.2.0', 'macos/aarch64', 'ok', '2026-07-23T00:00:00Z')",
    )
    .bind(bid)
    .execute(&pool)
    .await
    .unwrap();

    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.2.0").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(
        status, "ok",
        "all siblings green -> publish proceeds ({error})"
    );
    let count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM releases WHERE target = 'linux/x86_64'")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(count, 1, "linux published once the gate was satisfied");
}

/// A committed + tagged git repo whose linux recipe pins the host to the tag
/// via `checkout_sha`. `tag` is created only when `Some`.
fn init_git_app(repo: &std::path::Path, tauri_version: &str, tag: Option<&str>) {
    init_git_app_with_recipe(
        repo,
        tauri_version,
        tag,
        "step(\"checkout\");\nlet s = checkout_sha(build_host());\nlog(\"pinned \" + s);\n\
             step(\"build\");\nsh_ok(build_host(), \"true\");\n",
    );
}

/// As [`init_git_app`], with the linux recipe spelled by the caller.
fn init_git_app_with_recipe(
    repo: &std::path::Path,
    tauri_version: &str,
    tag: Option<&str>,
    recipe: &str,
) {
    init_git_app_shipping(repo, tauri_version, tag, recipe, "[\"linux/x86_64\"]");
}

/// As [`init_git_app_with_recipe`], with the manifest's target list spelled
/// by the caller — a second target is what puts a second HOST in the
/// preflight, which is the only way to reach its unwind path.
fn init_git_app_shipping(
    repo: &std::path::Path,
    tauri_version: &str,
    tag: Option<&str>,
    recipe: &str,
    targets: &str,
) {
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        format!("{{\"version\":\"{tauri_version}\"}}"),
    )
    .unwrap();
    std::fs::write(repo.join("bento.toml"), format!("targets = {targets}\n")).unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    std::fs::write(repo.join("dist/recipes/linux.rhai"), recipe).unwrap();
    // Isolate from the dev's global git config (which forces signed tags).
    let run = |args: &[&str]| {
        let out = std::process::Command::new("git")
            .args(args)
            .current_dir(repo)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .output()
            .expect("git runs");
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    };
    run(&["init", "-q"]);
    run(&["-c", "user.email=t@t", "-c", "user.name=t", "add", "-A"]);
    run(&[
        "-c",
        "user.email=t@t",
        "-c",
        "user.name=t",
        "commit",
        "-q",
        "-m",
        "init",
    ]);
    if let Some(t) = tag {
        run(&["tag", t]);
    }
}

/// A worktree root beside the checkout, the shape production uses: a hidden
/// directory in the tree the repos live in, not inside any of them.
fn worktree_root(repo: &std::path::Path) -> std::path::PathBuf {
    repo.parent().expect("repo has a parent").join(".bento")
}

/// Where a release of `demo` out of `repo` builds: the worktree, plus the
/// app's own prefix inside it for a repo holding several products.
fn build_dir(repo: &std::path::Path, prefix: &str) -> std::path::PathBuf {
    let repo_dir = repo.file_name().expect("repo has a name");
    worktree_root(repo).join(repo_dir).join("demo").join(prefix)
}

fn one_host_topo(repo: &std::path::Path) -> Topology {
    Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             worktree_root = \"{}\"\n\
             [app.demo]\nrepo = \"{}\"\n",
        worktree_root(repo).display(),
        repo.display()
    ))
    .unwrap()
}

/// The release preflight (pin on) pins the host to the tag and the build
/// proceeds. Exercises both the barrier and the `checkout_sha` host function.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn release_preflight_pins_and_builds_when_the_tag_is_present() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "ok", "a pinned build should succeed ({error})");
    assert!(
        build_dir(&repo, "")
            .join("src-tauri/tauri.conf.json")
            .exists(),
        "the release must have built in the worktree, not the checkout"
    );
    assert_eq!(
        head_sha(&build_dir(&repo, "")),
        tag_sha(&repo, "v0.0.1"),
        "and that worktree must be at the release tag"
    );
}

/// A pinned build writes an artifact record beside what it collected: the
/// manifest of those bytes, the commit the preflight pinned, and the steps
/// as artifact-scoped gates.
///
/// The three facts already existed and met nowhere — the per-file sha256 at
/// `collect`, the sha at the preflight, the step outcomes in `step_runs` —
/// which is how gates came to vouch for one thing while the deploy shipped
/// another.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_pinned_build_writes_an_artifact_record_for_what_it_collected() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    std::fs::create_dir_all(&repo).unwrap();
    // The recipe builds and collects out of `repo()`, which is the worktree.
    // That also proves the collect: an artifact built in a tree the sync gate
    // does not trust is refused, and the worktree root is an artifact root
    // for exactly this reason (`Host::artifact_roots`).
    let recipe = r#"
            step("checkout");
            let sha = checkout_sha(build_host());
            log("pinned " + sha);
            step("build");
            sh_ok(build_host(), "mkdir -p " + repo() + "/out && echo bin > " + repo() + "/out/demo.bin");
            step("collect");
            collect(build_host(), repo() + "/out/demo.bin", "demo", version());
        "#;
    init_git_app_with_recipe(&repo, "0.0.1", Some("v0.0.1"), recipe);

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let dist_root = cfg.dist_root.clone();
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    // No `pull_root`: the worktree root is the artifact root here, which is
    // the arrangement production uses.
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             worktree_root = \"{wt}\"\n[app.demo]\nrepo = \"{repo}\"\n",
        wt = worktree_root(&repo).display(),
        repo = repo.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "ok", "the build should succeed ({error})");

    let path = crate::artifact_record::record_path(
        &dist_root,
        &AppId::new("demo"),
        &Version::parse("0.0.1").unwrap(),
        "linux/x86_64".parse().unwrap(),
    );
    let json = std::fs::read_to_string(&path)
        .unwrap_or_else(|e| panic!("record at {}: {e}", path.display()));
    // Parsing revalidates, so this also asserts the digest matches the
    // manifest and no environment-scoped gate slipped in.
    let record = ops_artifact::ArtifactRecord::parse(&json).unwrap();

    assert_eq!(record.producer, "bento");
    assert_eq!(record.manifest.entries().len(), 1);
    assert_eq!(record.manifest.entries()[0].path, "demo.bin");
    assert_eq!(record.digest, record.manifest.digest());

    // The provenance names the commit the preflight pinned, not a rebuild of
    // whatever the branch is now.
    let head = std::process::Command::new("git")
        .args(["rev-parse", "v0.0.1^{commit}"])
        .current_dir(&repo)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .output()
        .unwrap();
    let head = String::from_utf8_lossy(&head.stdout).trim().to_string();
    assert_eq!(record.provenance.git_sha, head);
    assert_eq!(record.provenance.target, "linux/x86_64");
    assert_eq!(record.provenance.build_host, "fw13");
    assert!(!record.provenance.toolchain.is_empty());

    let gates: Vec<&str> = record.gates.iter().map(|g| g.gate.as_str()).collect();
    assert_eq!(gates, ["checkout", "build", "collect"]);
    assert!(record.all_gates_passed());
    assert!(
        record
            .gates
            .iter()
            .all(|g| g.scope == ops_artifact::Scope::Artifact),
        "a build host cannot vouch for an environment"
    );
}

/// The barrier refuses a target no host can build, instead of leaving it out
/// of the pin and letting the remaining hosts vouch for the release.
///
/// The hole this closes is silence reading as agreement: the comparison used
/// to run over the hosts that reported, so a target dropped for want of a
/// host still built while the other hosts' unanimity looked like proof the
/// whole release came from one commit.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn release_preflight_refuses_a_target_no_host_can_build() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    // The topology declares fw13 (linux/x86_64) only, so macOS has no host.
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let err = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["macos/aarch64".parse().unwrap()],
    )
    .await
    .unwrap_err();
    let msg = format!("{err:#}");
    assert!(
        msg.contains("no host can build macos/aarch64"),
        "the error must name the unbuildable target, got: {msg}"
    );
    let builds: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(builds, 0, "a refused preflight writes no build row");
}

/// A dirty checkout neither refuses a release nor reaches it.
///
/// `git checkout <tag>` does NOT fail on local modifications to files whose
/// content is unchanged in the tag: it succeeds and keeps them, so a dirty
/// host would build its edits while honestly reporting the tagged sha.
///
/// The worktree answers that instead of gating on it. The release is built
/// from a tree the edit is not in, so the edit cannot reach the binary and a
/// release is never refused because somebody has unsaved work.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_dirty_checkout_neither_refuses_a_release_nor_reaches_it() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));

    // Modify a TRACKED file, exactly as an editor session would.
    let tracked = repo.join("src-tauri/tauri.conf.json");
    // Still valid JSON and still version 0.0.1: the version preflight reads
    // this file on the daemon's box before anything is pinned, so an edit
    // that broke it would fail the release for a reason this test is not
    // about.
    let edited = "{\"version\":\"0.0.1\",\"unsaved\":true}".to_string();
    std::fs::write(&tracked, &edited).unwrap();

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(
        status, "ok",
        "an edit elsewhere must not stop a release ({error})"
    );
    assert_eq!(
        std::fs::read_to_string(&tracked).unwrap(),
        edited,
        "and the edit must still be there afterwards"
    );
    assert_ne!(
        std::fs::read_to_string(build_dir(&repo, "").join("src-tauri/tauri.conf.json")).unwrap(),
        edited,
        "what was built is the tag's content, not the edit"
    );
}

/// An UNTRACKED file does not fail a release. A build host accumulates
/// editor scratch and stray logs, none of which reach the binary, so failing
/// on them would be noise that trains an operator to bypass the gate.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn release_preflight_tolerates_untracked_files() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    std::fs::write(repo.join("scratch.log"), "noise").unwrap();

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(
        status, "ok",
        "untracked files must not fail a release ({error})"
    );
}

/// The preflight refuses the build (before any row is written) when the
/// release tag does not exist on the host — a missing/unpushed tag can't
/// silently fall back to whatever `main` is.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn release_preflight_refuses_when_the_release_tag_is_missing() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    // App is 0.0.2 (so the version check passes) but only v0.0.1 is tagged.
    init_git_app(&repo, "0.0.2", Some("v0.0.1"));
    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let err = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.2").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap_err();
    let msg = format!("{err:#}");
    assert!(
        msg.contains("release preflight") && msg.contains("v0.0.2"),
        "expected a preflight tag error, got: {msg}"
    );
    assert!(
        msg.contains("does not exist"),
        "the error should name the absent tag as the cause, got: {msg}"
    );
    // Refused before anything was recorded.
    let builds: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(builds, 0, "a refused preflight writes no build row");
}

/// A release does not touch the checkout at all: same branch, same commit,
/// same working tree.
///
/// Checking the tag out there would move a tree somebody works in under them
/// for the length of a build, and leave it detached whenever the restore
/// failed. The worktree makes the question moot, so this asserts the strong
/// form rather than a recovery.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_release_never_moves_the_checkout() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    let branch_before = current_branch(&repo);
    let head_before = head_sha(&repo);
    assert!(!branch_before.is_empty(), "test repo starts on a branch");

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "ok", "the build itself should pass ({error})");

    assert_eq!(
        current_branch(&repo),
        branch_before,
        "the checkout must still be on its branch"
    );
    assert_eq!(head_sha(&repo), head_before, "and at the same commit");
    assert_eq!(repo_status(&repo), "", "and with the same working tree");
}

/// A detached checkout does not refuse a release.
///
/// Releases do not move that tree, so its HEAD is not their business.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_detached_checkout_no_longer_refuses_a_release() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    // Detach, standing in for a checkout an earlier release left on its tag.
    let out = std::process::Command::new("git")
        .args(["checkout", "--detach", "-q", "HEAD"])
        .current_dir(&repo)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .env("GIT_CONFIG_SYSTEM", "/dev/null")
        .output()
        .expect("git runs");
    assert!(out.status.success());
    assert!(current_branch(&repo).is_empty(), "repo is detached");

    let head_before = head_sha(&repo);

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(
        status, "ok",
        "a detached checkout is not a reason to refuse ({error})"
    );
    assert!(
        current_branch(&repo).is_empty() && head_sha(&repo) == head_before,
        "and the release left it exactly as detached as it found it"
    );
}

/// A build that rewrites a tracked file does not stop the next release.
///
/// `cargo build` rewrites `Cargo.lock` under the `[patch]` block, so a build
/// leaves the tree it ran in dirty. The dirt lands in Bento's own worktree
/// and the next release forces past it, which is safe precisely because
/// nothing else writes there. So this releases the same app twice with a
/// build that dirties the tree, and the second one is the assertion.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_build_that_dirties_its_tree_does_not_refuse_the_next_release() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    std::fs::create_dir_all(&repo).unwrap();
    std::fs::write(repo.join("Cargo.lock"), "version = 4\n").unwrap();
    init_git_app_with_recipe(
        &repo,
        "0.0.1",
        Some("v0.0.1"),
        "step(\"build\");\nsh_ok(build_host(), \"echo churn >> \" + repo() + \"/Cargo.lock\");\n",
    );
    // The tag and the branch disagree about the lockfile.
    let git = git_in(&repo);
    std::fs::write(repo.join("Cargo.lock"), "version = 4\n# moved on\n").unwrap();
    git(&["add", "-A"]);
    git(&["commit", "-q", "-m", "lock moves on"]);

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let topo = one_host_topo(&repo);

    for attempt in 1..=2 {
        let state = test_state(pool.clone(), topo.clone(), cfg.clone());
        let build_id = start_build(
            state,
            AppId::new("demo"),
            Version::parse("0.0.1").unwrap(),
            vec!["linux/x86_64".parse().unwrap()],
        )
        .await
        .unwrap_or_else(|e| panic!("release {attempt} refused: {e:#}"));
        let (status, error) = await_target(&pool, build_id).await;
        assert_eq!(status, "ok", "release {attempt} should build ({error})");
    }

    assert_eq!(
        std::fs::read_to_string(build_dir(&repo, "").join("Cargo.lock")).unwrap(),
        "version = 4\nchurn\n",
        "the second release started from the tag's lockfile, not the first's leavings"
    );
    assert_eq!(repo_status(&repo), "", "and the checkout was never in it");
}

/// A working copy ahead of the tag does not fail the version check.
///
/// The check reads the tag, not the checkout, so releasing v0.0.1 while
/// `main` has moved to 0.0.2 is not version drift: the tag says 0.0.1.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_working_copy_ahead_of_the_tag_does_not_fail_the_version_check() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    // main moves on, untagged, exactly as it does the day after a release.
    let git = git_in(&repo);
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        "{\"version\":\"0.0.2\"}",
    )
    .unwrap();
    git(&["add", "-A"]);
    git(&["commit", "-q", "-m", "0.0.2"]);

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .expect("releasing the tag behind main must not be version drift");
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "ok", "({error})");
}

/// A tag that disagrees with itself is refused, naming the file.
///
/// The drift worth catching: `tauri.conf.json` at 0.0.1 and `Cargo.toml`
/// still at 0.0.2, committed and tagged that way. Reading the checkout would
/// compare against whatever the working copy happens to say instead.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_tag_whose_manifest_disagrees_with_it_is_refused() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    std::fs::create_dir_all(&repo).unwrap();
    std::fs::write(
        repo.join("Cargo.toml"),
        "[package]\nname = \"demo\"\nversion = \"0.0.2\"\n",
    )
    .unwrap();
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let err = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap_err();
    let msg = format!("{err:#}");
    assert!(
        msg.contains("version drift") && msg.contains("Cargo.toml says 0.0.2"),
        "the refusal must name the file and what it says, got: {msg}"
    );
    assert!(
        msg.contains("v0.0.1"),
        "and say which tag it read, got: {msg}"
    );
    let builds: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(builds, 0, "a refused version preflight writes no build row");
}

/// A preflight that refuses on a LATER host leaves every checkout alone.
///
/// The hosts are prepared one at a time, so when the second is refused the
/// first has already been through the whole preflight. There is no unwind to
/// get right: the checkouts were never moved, and the worktrees a refusal
/// leaves behind are where the next release would have put them.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_refused_preflight_leaves_every_checkout_untouched() {
    let tmp = tempfile::tempdir().unwrap();
    // fw13's checkout is tagged and clean, so it pins. mbp's has no v0.0.1,
    // so it is refused after fw13 has already been moved.
    let repo = tmp.path().join("demo");
    init_git_app_shipping(
        &repo,
        "0.0.1",
        Some("v0.0.1"),
        "step(\"build\");\nsh_ok(build_host(), \"true\");\n",
        "[\"linux/x86_64\", \"macos/aarch64\"]",
    );
    let other = tmp.path().join("demo-mbp");
    init_git_app_shipping(
        &other,
        "0.0.1",
        None,
        "step(\"build\");\nsh_ok(build_host(), \"true\");\n",
        "[\"linux/x86_64\", \"macos/aarch64\"]",
    );
    let branch_before = current_branch(&repo);
    assert!(
        !branch_before.is_empty(),
        "fw13's checkout starts on a branch"
    );

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             worktree_root = \"{wt}\"\n\
             [[host]]\nname = \"mbp\"\nssh = \"local\"\ntargets = [\"macos/aarch64\"]\n\
             worktree_root = \"{wt}\"\n\
             [app.demo]\nrepo = \"{}\"\n[app.demo.repo_by_host]\nmbp = \"{}\"\n",
        repo.display(),
        other.display(),
        wt = worktree_root(&repo).display(),
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let err = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec![
            "linux/x86_64".parse().unwrap(),
            "macos/aarch64".parse().unwrap(),
        ],
    )
    .await
    .unwrap_err();
    let msg = format!("{err:#}");
    assert!(
        msg.contains("does not exist"),
        "the refusal should still be mbp's missing tag, got: {msg}"
    );
    assert_eq!(
        current_branch(&repo),
        branch_before,
        "a refused preflight must not have moved the host it prepared first"
    );
    assert_eq!(repo_status(&repo), "", "nor left anything in its tree");
}

/// An edit elsewhere in a repo holding several products does not touch a
/// release of one of them.
///
/// An edit in `server/` is not something pom's build compiles, but `git
/// checkout <tag>` acts on the whole repository, and MNW is one `.git` over
/// the server, sando, multithreaded and pom.
///
/// A worktree is made of the repository too, so the release still gets the
/// whole tagged tree, in its own copy, and the edit stays where its author
/// left it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_edit_elsewhere_in_the_repo_survives_a_release_and_stays_out_of_it() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path().join("monorepo");
    let app = root.join("pom");
    init_git_app(&app, "0.0.1", None);
    let git = |args: &[&str]| {
        let out = std::process::Command::new("git")
            .args(args)
            .current_dir(&root)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .env("GIT_AUTHOR_NAME", "t")
            .env("GIT_AUTHOR_EMAIL", "t@t")
            .env("GIT_COMMITTER_NAME", "t")
            .env("GIT_COMMITTER_EMAIL", "t@t")
            .output()
            .expect("git runs");
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    };
    // `init_git_app` made `pom/` its own repo; the monorepo is the parent, so
    // drop that and re-init one `.git` over both products.
    std::fs::remove_dir_all(app.join(".git")).unwrap();
    std::fs::create_dir_all(root.join("server")).unwrap();
    std::fs::write(root.join("server/Cargo.lock"), "version = 4\n").unwrap();
    git(&["init", "-q"]);
    git(&["add", "-A"]);
    git(&["commit", "-q", "-m", "init"]);
    git(&["tag", "v0.0.1"]);
    // The tag and the branch differ in `server/`, and the working copy of
    // that file is modified — so the checkout cannot carry the edit across.
    std::fs::write(root.join("server/Cargo.lock"), "version = 4\n# moved on\n").unwrap();
    git(&["add", "-A"]);
    git(&["commit", "-q", "-m", "server moves"]);
    std::fs::write(
        root.join("server/Cargo.lock"),
        "version = 4\n# local edit\n",
    )
    .unwrap();

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    // The worktree root sits beside the monorepo, not inside it.
    let topo = Topology::from_str_for_tests(&format!(
        "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             worktree_root = \"{wt}\"\n[app.demo]\nrepo = \"{repo}\"\n",
        wt = worktree_root(&root).display(),
        repo = app.display()
    ))
    .unwrap();
    let state = test_state(pool.clone(), topo, cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(
        status, "ok",
        "an edit in server/ must not stop pom's release ({error})"
    );
    assert_eq!(
        std::fs::read_to_string(root.join("server/Cargo.lock")).unwrap(),
        "version = 4\n# local edit\n",
        "the edit must be exactly where its author left it"
    );
    // The worktree is of the whole repository, at the tag: `server/` is
    // there, and it is the tagged content rather than either the branch tip
    // or the edit.
    let worktree = worktree_root(&root).join("monorepo").join("demo");
    assert_eq!(
        std::fs::read_to_string(worktree.join("server/Cargo.lock")).unwrap(),
        "version = 4\n",
        "the release built the tag's server/, not the working copy's"
    );
}

/// The branch `repo` is on, empty on a detached HEAD.
/// Tracked files with local changes, as `git status --porcelain` writes
/// them, joined by newlines and empty for a clean tree.
fn repo_status(repo: &std::path::Path) -> String {
    let out = std::process::Command::new("git")
        .args(["status", "--porcelain", "--untracked-files=no"])
        .current_dir(repo)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .env("GIT_CONFIG_SYSTEM", "/dev/null")
        .output()
        .expect("git runs");
    String::from_utf8_lossy(&out.stdout).trim_end().to_string()
}

/// Run git in `dir`, isolated from the dev's global config and with an
/// identity, panicking with git's own stderr on failure.
fn git_in(dir: &std::path::Path) -> impl Fn(&[&str]) + '_ {
    move |args: &[&str]| {
        let out = std::process::Command::new("git")
            .args(args)
            .current_dir(dir)
            .env("GIT_CONFIG_GLOBAL", "/dev/null")
            .env("GIT_CONFIG_SYSTEM", "/dev/null")
            .env("GIT_AUTHOR_NAME", "t")
            .env("GIT_AUTHOR_EMAIL", "t@t")
            .env("GIT_COMMITTER_NAME", "t")
            .env("GIT_COMMITTER_EMAIL", "t@t")
            .output()
            .expect("git runs");
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    }
}

/// The commit a checkout (or worktree) is on.
fn head_sha(repo: &std::path::Path) -> String {
    git_read(repo, &["rev-parse", "HEAD"])
}

/// The commit a tag names.
fn tag_sha(repo: &std::path::Path, tag: &str) -> String {
    git_read(repo, &["rev-parse", &format!("{tag}^{{commit}}")])
}

fn git_read(dir: &std::path::Path, args: &[&str]) -> String {
    let out = std::process::Command::new("git")
        .args(args)
        .current_dir(dir)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .env("GIT_CONFIG_SYSTEM", "/dev/null")
        .output()
        .expect("git runs");
    String::from_utf8_lossy(&out.stdout).trim().to_string()
}

fn current_branch(repo: &std::path::Path) -> String {
    let out = std::process::Command::new("git")
        .args(["symbolic-ref", "-q", "--short", "HEAD"])
        .current_dir(repo)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .env("GIT_CONFIG_SYSTEM", "/dev/null")
        .output()
        .expect("git runs");
    String::from_utf8_lossy(&out.stdout).trim().to_string()
}

/// An unreachable remote does not fail a release whose tag is present.
///
/// `fetch --all` is non-zero if ANY remote fails, and every library repo
/// carries three (`astra`, `mnw`, `srht`), so a dead mirror would abort the
/// release and blame it on a missing tag. Fetch is advisory; only the
/// checkout decides.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_dead_remote_does_not_fail_a_release_whose_tag_exists() {
    let tmp = tempfile::tempdir().unwrap();
    let repo = tmp.path().join("demo");
    init_git_app(&repo, "0.0.1", Some("v0.0.1"));
    // A remote that cannot possibly be fetched, standing in for an offline
    // astra or an srht mirror the repo was never pushed to.
    let out = std::process::Command::new("git")
        .args([
            "remote",
            "add",
            "srht",
            &tmp.path().join("nowhere.git").display().to_string(),
        ])
        .current_dir(&repo)
        .env("GIT_CONFIG_GLOBAL", "/dev/null")
        .env("GIT_CONFIG_SYSTEM", "/dev/null")
        .output()
        .expect("git runs");
    assert!(out.status.success());

    let mut cfg = Config::for_tests(tmp.path());
    cfg.pin_release_sha = true;
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let state = test_state(pool.clone(), one_host_topo(&repo), cfg);
    let build_id = start_build(
        state,
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .expect("a dead mirror must not refuse the release");
    let (status, error) = await_target(&pool, build_id).await;
    assert_eq!(status, "ok", "the tag is present, so this builds ({error})");
}

/// The audit fix: a `build` step dispatched to a host whose executor lacks the
/// `build` capability is denied at the transport BEFORE the command runs. This
/// is the structural guarantee behind "never build on prod" — a recipe naming
/// the wrong host can't compile there.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn build_on_a_host_without_the_build_grant_is_denied() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    // The recipe (wrongly) tries to compile on `prod`, a host with no build
    // grant. A marker file would appear if the command actually ran.
    let marker = root.join("ran-on-prod");
    std::fs::write(
        repo.join("dist/recipes/linux.rhai"),
        r#"
                step("build");
                sh_ok("prod", "touch MARKER");
            "#
        .replace("MARKER", marker.to_str().unwrap()),
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    // fw13 builds linux; `prod` is local-but-restart-only (no build/package).
    std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(
        r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[[host]]
name = "prod"
ssh = "local"
actuate = ["restart"]
observe = []

[app.demo]
repo = "REPO"
"#
        .replace("REPO", repo.to_str().unwrap())
        .as_str(),
    )
    .unwrap();

    let state = test_state(pool.clone(), topo, cfg);

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["linux/x86_64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    assert_eq!(status, "failed", "build on an ungranted host must fail");
    assert!(
        error.contains("capability denied") && error.contains("build"),
        "failure must be a capability denial, got: {error}"
    );
    assert!(!marker.exists(), "denied build step must NOT have executed");
}

// ---- macOS sign / notarize / staple execution chain (via ScriptedExec) ----

/// Stand up a single-macOS-target app whose recipe is `recipe_body` (with the
/// literal `ARTIFACT` replaced by a real, non-empty file on disk), dispatch
/// every host command through `scripted`, run the build to a terminal state,
/// and return the pool plus the final `(status, error)`. The returned
/// [`tempfile::TempDir`] must be kept alive by the caller: it holds the
/// sqlite DB the pool reads.
async fn run_macos_recipe(
    scripted: Arc<ScriptedExec>,
    recipe_body: &str,
    backoff_secs: Option<u64>,
) -> (tempfile::TempDir, SqlitePool, String, String) {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();

    // A real, non-empty artifact so `publish`'s size floor is satisfied; the
    // build step is faked, so nothing else creates it.
    let artifact = repo.join("out/demo.dmg");
    std::fs::create_dir_all(artifact.parent().unwrap()).unwrap();
    std::fs::write(&artifact, b"dmg-bytes").unwrap();

    std::fs::write(
        repo.join("dist/recipes/macos.rhai"),
        recipe_body.replace("ARTIFACT", artifact.to_str().unwrap()),
    )
    .unwrap();

    let cfg = Config {
        notarize_backoff_secs: backoff_secs,
        ..Config::for_tests(root)
    };
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    std::fs::write(repo.join("bento.toml"), "targets = [\"macos/aarch64\"]\n").unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        r#"
[[host]]
name = "mbp"
ssh = "local"
targets = ["macos/aarch64"]

[app.demo]
repo = "{}"
"#,
        repo.display()
    ))
    .unwrap();

    let mut state = test_state(pool.clone(), topo, cfg);
    // Route every host command through the scripted executor.
    let mut execs = HashMap::new();
    execs.insert("mbp".to_string(), scripted as Arc<dyn Executor>);
    state.executors = Arc::new(execs);

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec!["macos/aarch64".parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    (tmp, pool, status, error)
}

async fn release_count(pool: &SqlitePool) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM releases")
        .fetch_one(pool)
        .await
        .unwrap()
}

/// The whole macOS release chain end to end through a fake host: codesign,
/// notarize (Accepted first try), staple, verify_gatekeeper, then publish.
/// The recorded commands lock the exact incantations each host function
/// dispatches, and a `releases` row proves the publish gate opened for a
/// signed + notarized + Gatekeeper-accepted artifact.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_macos_recipe_signs_notarizes_staples_verifies_and_publishes() {
    let scripted = Arc::new(
        ScriptedExec::new()
            .on("codesign", 0, "")
            .on("notarytool", 0, r#"{"status":"Accepted"}"#)
            .on("stapler staple", 0, "")
            .on(
                "spctl",
                0,
                "source=Notarized Developer ID\nBENTO_GATEKEEPER_OK",
            ),
    );
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "ARTIFACT");
            notarize(h, "ARTIFACT");
            staple(h, "ARTIFACT");
            step("verify");
            if !verify_gatekeeper(h, "ARTIFACT") { throw "not Gatekeeper-accepted"; }
            step("publish");
            publish("tauri-mnw", "demo", "macos/aarch64", "0.0.1", "ARTIFACT", #{});
        "#;
    let (_tmp, pool, status, error) = run_macos_recipe(scripted.clone(), recipe, None).await;

    assert_eq!(
        status, "ok",
        "signed+notarized macOS build should publish: {error}"
    );
    assert_eq!(
        release_count(&pool).await,
        1,
        "publish must record a release"
    );

    // The exact shell incantations each host function dispatched.
    let cmds = scripted.commands();
    let has = |needle: &str| cmds.iter().any(|c| c.contains(needle));
    assert!(
        has("codesign --force --options runtime --timestamp --sign"),
        "codesign runtime+timestamp incantation, got: {cmds:?}"
    );
    assert!(
        has("xcrun notarytool submit"),
        "notarytool submit: {cmds:?}"
    );
    assert!(
        has("--wait --output-format json"),
        "notarytool --wait json: {cmds:?}"
    );
    assert!(has("xcrun stapler staple"), "stapler staple: {cmds:?}");
    assert!(has("spctl --assess"), "gatekeeper assess: {cmds:?}");
}

/// A failing `codesign` fails the sign step and aborts the recipe before
/// publish — an unsigned artifact never ships.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_codesign_failure_fails_the_sign_step_and_blocks_publish() {
    let scripted = Arc::new(ScriptedExec::new().on("codesign", 1, ""));
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "ARTIFACT");
            step("publish");
            publish("tauri-mnw", "demo", "macos/aarch64", "0.0.1", "ARTIFACT", #{});
        "#;
    let (_tmp, pool, status, error) = run_macos_recipe(scripted, recipe, None).await;

    assert_eq!(status, "failed", "a failed codesign must fail the target");
    assert!(
        error.contains("codesign failed"),
        "error names the codesign failure, got: {error}"
    );
    assert_eq!(
        release_count(&pool).await,
        0,
        "nothing may publish after a codesign failure"
    );
}

/// Even if the recipe ignores `verify_gatekeeper`'s returned `false`, the
/// publish gate refuses the artifact: `verify_gatekeeper` both records the
/// rejection and fails its step, and `publish` proves neither passed. This
/// is the defense-in-depth the pure `PublishAuthority::prove` tests assert in
/// isolation, here exercised through the real host-function path.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_gatekeeper_rejection_bars_publish_even_if_the_recipe_ignores_it() {
    let scripted = Arc::new(
        ScriptedExec::new()
            .on("codesign", 0, "")
            .on("notarytool", 0, r#"{"status":"Accepted"}"#)
            .on("stapler staple", 0, "")
            // Gatekeeper says no: the sentinel is FAIL, not OK.
            .on("spctl", 0, "source=Unnotarized\nBENTO_GATEKEEPER_FAIL"),
    );
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "ARTIFACT");
            notarize(h, "ARTIFACT");
            staple(h, "ARTIFACT");
            step("verify");
            verify_gatekeeper(h, "ARTIFACT");
            step("publish");
            publish("tauri-mnw", "demo", "macos/aarch64", "0.0.1", "ARTIFACT", #{});
        "#;
    let (_tmp, pool, status, error) = run_macos_recipe(scripted, recipe, None).await;

    assert_eq!(
        status, "failed",
        "a Gatekeeper-rejected artifact must not publish"
    );
    assert!(
        !error.is_empty(),
        "the barred publish must surface an error"
    );
    assert_eq!(
        release_count(&pool).await,
        0,
        "no release for a rejected artifact"
    );
}

/// The one flaky, network-bound step: `notarize` retries a non-`Accepted`
/// result and succeeds on a later attempt. Two notarytool calls (reject then
/// accept) then a recorded release prove the retry ran and the chain
/// completed. Backoff is 0 so the retry sleep doesn't stall the test.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notarize_retries_a_non_accepted_result_then_succeeds() {
    let scripted = Arc::new(
        ScriptedExec::new()
            .on("codesign", 0, "")
            .on_seq(
                "notarytool",
                &[
                    (0, r#"{"status":"In Progress"}"#),
                    (0, r#"{"status":"Accepted"}"#),
                ],
            )
            .on("stapler staple", 0, "")
            .on(
                "spctl",
                0,
                "source=Notarized Developer ID\nBENTO_GATEKEEPER_OK",
            ),
    );
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "ARTIFACT");
            notarize(h, "ARTIFACT");
            staple(h, "ARTIFACT");
            step("verify");
            if !verify_gatekeeper(h, "ARTIFACT") { throw "not Gatekeeper-accepted"; }
            step("publish");
            publish("tauri-mnw", "demo", "macos/aarch64", "0.0.1", "ARTIFACT", #{});
        "#;
    let (_tmp, pool, status, error) = run_macos_recipe(scripted.clone(), recipe, Some(0)).await;

    assert_eq!(
        status, "ok",
        "notarize should succeed on the retry: {error}"
    );
    assert_eq!(
        release_count(&pool).await,
        1,
        "the retried build still publishes"
    );
    let notary_calls = scripted
        .commands()
        .iter()
        .filter(|c| c.contains("notarytool"))
        .count();
    assert_eq!(
        notary_calls, 2,
        "notarytool ran once, was rejected, then ran again"
    );
}

/// `notarize` gives up after its bounded retries: three notarytool attempts,
/// all non-`Accepted`, fail the target and bar publish.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn notarize_fails_the_target_after_exhausting_its_retries() {
    let scripted = Arc::new(
        ScriptedExec::new()
            .on("codesign", 0, "")
            // Every attempt reports a still-pending status, never Accepted.
            .on("notarytool", 0, r#"{"status":"In Progress"}"#),
    );
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "ARTIFACT");
            notarize(h, "ARTIFACT");
            step("publish");
            publish("tauri-mnw", "demo", "macos/aarch64", "0.0.1", "ARTIFACT", #{});
        "#;
    let (_tmp, pool, status, error) = run_macos_recipe(scripted.clone(), recipe, Some(0)).await;

    assert_eq!(
        status, "failed",
        "exhausted notarization must fail the target"
    );
    assert!(
        error.contains("notarization failed after 3 attempts"),
        "error names the exhausted retry, got: {error}"
    );
    assert_eq!(
        release_count(&pool).await,
        0,
        "an unnotarized artifact never publishes"
    );
    let notary_calls = scripted
        .commands()
        .iter()
        .filter(|c| c.contains("notarytool"))
        .count();
    assert_eq!(notary_calls, 3, "the retry is bounded at three attempts");
}
// ---- item (5): SSH and Agent transports end to end, via a recording fake ----
//
// Every other test declares `ssh = "local"`, so the runner's two-plane
// routing is only exercised at construction
// (state::agent_host_syncs_over_ssh_never_the_agent): steps run over the
// EXEC transport (state.executors -- SshExec, or the in-session AgentRpc for
// a mac host) while artifacts move over the SYNC transport (state.syncs --
// always ssh, NEVER the agent, whose confined `/pull` would 404 or force
// `pull_root` wide enough to expose `~/.tauri/passwords.env`). These drive a
// real recipe through `start_build` on a NON-local topology, replace the
// real ssh/agent transports with a recording fake, and assert which plane
// handled which operation -- the runtime form of that construction-time
// invariant, on the ssh string every other test pins to "local".

/// Records shell commands and artifact pulls on separate logs, so a test can
/// prove the exec transport built/signed and the sync transport collected --
/// and that neither did the other's job -- without a live host or ssh.
struct RecordingExec {
    caps: CapabilitySet,
    commands: Arc<std::sync::Mutex<Vec<String>>>,
    pulls: Arc<std::sync::Mutex<Vec<String>>>,
}

impl RecordingExec {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            // A mac build host's real grant. Nothing in this fake gates on it
            // (the real transports do), but keep it coherent so
            // `capabilities()` is not a lie.
            caps: CapabilitySet::from_tokens(
                ["build", "sign", "notarize", "staple"],
                ["build-log", "artifact"],
            ),
            commands: Arc::new(std::sync::Mutex::new(Vec::new())),
            pulls: Arc::new(std::sync::Mutex::new(Vec::new())),
        })
    }
    fn commands(&self) -> Vec<String> {
        self.commands.lock().unwrap().clone()
    }
    fn pulls(&self) -> Vec<String> {
        self.pulls.lock().unwrap().clone()
    }
}

#[async_trait]
impl Executor for RecordingExec {
    async fn run_streaming(
        &self,
        step: &ops_exec::Step,
        _sink: &mut dyn LogSink,
    ) -> anyhow::Result<RunOutput> {
        self.commands
            .lock()
            .unwrap()
            .push(step.argv.last().cloned().unwrap_or_default());
        Ok(RunOutput {
            status: std::process::ExitStatus::from_raw(0),
            stdout: Vec::new(),
            stderr: Vec::new(),
        })
    }
    async fn pull_file(
        &self,
        r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        self.pulls
            .lock()
            .unwrap()
            .push(r.to_string_lossy().into_owned());
        Ok(())
    }
    async fn pull_dir(
        &self,
        r: &std::path::Path,
        _l: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        self.pulls
            .lock()
            .unwrap()
            .push(r.to_string_lossy().into_owned());
        Ok(())
    }
    async fn pull_glob(&self, g: &str, _l: &std::path::Path, _o: &SyncOpts) -> anyhow::Result<()> {
        self.pulls.lock().unwrap().push(g.to_string());
        Ok(())
    }
    async fn push_dir(
        &self,
        _l: &std::path::Path,
        _r: &std::path::Path,
        _o: &SyncOpts,
    ) -> anyhow::Result<()> {
        Ok(())
    }
    async fn preflight(&self) -> anyhow::Result<()> {
        Ok(())
    }
    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}

/// Stand up a single-target app (host named `h1`) whose recipe is
/// `recipe_body` (with `REPO` replaced by the checkout path), inject
/// `exec_fake` as the host's EXEC transport and `sync_fake` as its SYNC
/// transport, run the build to a terminal state, and return the tmpdir plus
/// `(status, error)`. Unlike `test_state`'s `build_executors`, this replaces
/// BOTH planes so no real ssh/agent transport is dialed. The returned
/// [`tempfile::TempDir`] holds the sqlite DB and must outlive the caller's
/// assertions.
async fn run_two_plane(
    host_toml: &str,
    target: &str,
    recipe_file: &str,
    recipe_body: &str,
    exec_fake: Arc<dyn Executor>,
    sync_fake: Arc<dyn Executor>,
) -> (tempfile::TempDir, String, String) {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();
    let repo = root.join("app");
    std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
    std::fs::write(
        repo.join("src-tauri/tauri.conf.json"),
        r#"{"version":"0.0.1"}"#,
    )
    .unwrap();
    std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
    std::fs::write(
        repo.join("dist/recipes").join(recipe_file),
        recipe_body.replace("REPO", repo.to_str().unwrap()),
    )
    .unwrap();
    std::fs::write(
        repo.join("bento.toml"),
        format!("targets = [\"{target}\"]\n"),
    )
    .unwrap();

    let cfg = Config::for_tests(root);
    let pool = crate::db::open(&cfg.db_path).await.unwrap();
    let topo = Topology::from_str_for_tests(&format!(
        "{}\n[app.demo]\nrepo = \"{}\"\n",
        host_toml.replace("REPO", repo.to_str().unwrap()),
        repo.display()
    ))
    .unwrap();

    let mut state = test_state(pool.clone(), topo, cfg);
    state.executors = Arc::new(HashMap::from([("h1".to_string(), exec_fake)]));
    state.syncs = Arc::new(HashMap::from([("h1".to_string(), sync_fake)]));

    let build_id = start_build(
        state.clone(),
        AppId::new("demo"),
        Version::parse("0.0.1").unwrap(),
        vec![target.parse().unwrap()],
    )
    .await
    .unwrap();

    let mut status = String::new();
    let mut error = String::new();
    for _ in 0..100 {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
                .bind(build_id)
                .fetch_optional(&pool)
                .await
                .unwrap();
        if let Some((s, e)) = row {
            status = s;
            error = e.unwrap_or_default();
            if status != "running" {
                break;
            }
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    (tmp, status, error)
}

/// An agent (macOS) host signs over the AGENT transport but is collected
/// from over SSH -- driven through the whole runner, not just `build_sync`.
/// The sign chain's commands land on the exec plane and the artifact pull on
/// the sync plane; crucially, the agent plane is asked to move NOTHING (a
/// regression to one transport would route collect at `AgentRpc::pull_glob`,
/// refused by design, or widen `pull_root` over the secret-bearing home dir).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_agent_host_signs_over_the_agent_and_collects_over_ssh_end_to_end() {
    let agent = RecordingExec::new(); // exec plane (AgentRpc in prod)
    let ssh = RecordingExec::new(); // sync plane (SshExec in prod)
    let host = r#"
[[host]]
name = "h1"
ssh = "mbp"
targets = ["macos/aarch64"]
transport = "agent"
agent_url = "http://mbp:8765"
actuate = ["build", "sign", "notarize", "staple"]
observe = ["build-log", "gatekeeper", "artifact"]
pull_root = "REPO"
"#;
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo built");
            step("sign");
            codesign(h, "Developer ID Application: Test", "REPO/out/demo.dmg");
            step("collect");
            collect(h, "REPO/out/*.dmg", "demo", "0.0.1");
        "#;
    let (_tmp, status, error) = run_two_plane(
        host,
        "macos/aarch64",
        "macos.rhai",
        recipe,
        agent.clone(),
        ssh.clone(),
    )
    .await;
    assert_eq!(status, "ok", "the recipe should complete: {error}");

    // The build + sign commands ran on the AGENT (exec) transport.
    let agent_cmds = agent.commands();
    assert!(
        agent_cmds.iter().any(|c| c.contains("codesign")),
        "codesign rides the agent exec transport: {agent_cmds:?}"
    );
    assert!(
        agent_cmds.iter().any(|c| c.contains("echo built")),
        "the build step rides the agent exec transport: {agent_cmds:?}"
    );
    // ...and the agent moved NO artifacts. This is the load-bearing half:
    // AgentRpc::pull_glob is refused by design, so collect must not touch it.
    assert!(
        agent.pulls().is_empty(),
        "the agent transport must never collect artifacts: {:?}",
        agent.pulls()
    );

    // The artifact was collected over the SSH (sync) transport...
    let ssh_pulls = ssh.pulls();
    assert!(
        ssh_pulls
            .iter()
            .any(|p| p.contains("demo.dmg") || p.contains("*.dmg")),
        "collect rides the ssh sync transport: {ssh_pulls:?}"
    );
    // ...and the sync transport was never asked to run a build/sign command.
    assert!(
        ssh.commands().is_empty(),
        "the sync transport must never run host commands: {:?}",
        ssh.commands()
    );
}

/// A plain (non-agent) host whose `ssh` is a remote alias, not "local" --
/// the case every other test avoids. The recipe runs end to end through the
/// fake, proving the runner drives a non-local host and still splits exec
/// (build) from sync (collect) across the two transport maps.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_non_local_ssh_host_runs_a_recipe_end_to_end() {
    let exec = RecordingExec::new();
    let sync = RecordingExec::new();
    let host = r#"
[[host]]
name = "h1"
ssh = "astra"
targets = ["linux/x86_64"]
pull_root = "REPO"
"#;
    let recipe = r#"
            let h = build_host();
            step("build");
            sh_ok(h, "echo compiling");
            step("collect");
            collect(h, "REPO/out/demo.bin", "demo", "0.0.1");
        "#;
    let (_tmp, status, error) = run_two_plane(
        host,
        "linux/x86_64",
        "linux.rhai",
        recipe,
        exec.clone(),
        sync.clone(),
    )
    .await;
    assert_eq!(status, "ok", "the recipe should complete: {error}");

    assert!(
        exec.commands().iter().any(|c| c.contains("echo compiling")),
        "the build command rides the exec transport: {:?}",
        exec.commands()
    );
    assert!(
        exec.pulls().is_empty(),
        "the exec transport must not collect: {:?}",
        exec.pulls()
    );
    assert!(
        sync.pulls().iter().any(|p| p.contains("demo.bin")),
        "collect rides the sync transport: {:?}",
        sync.pulls()
    );
    assert!(
        sync.commands().is_empty(),
        "the sync transport must not run commands: {:?}",
        sync.commands()
    );
}

/// Every recipe of every configured app must parse.
///
/// A recipe is read and compiled at release time, on the build host, after the
/// checkout has already run — so a typo in one is discovered at the worst
/// possible moment. Compiling them here costs nothing and moves that discovery
/// to `cargo test`. Skips when the live config is absent (CI, another host),
/// like [`crate::topology`]'s live-config smoke test.
#[cfg(test)]
mod live_recipe_smoke {
    use crate::topology::Topology;
    use std::path::{Path, PathBuf};

    #[test]
    fn live_recipes_compile_if_present() {
        let Some(home) = std::env::var_os("HOME") else {
            return;
        };
        let path = Path::new(&home).join(".config/bento/bento.toml");
        if !path.exists() {
            return;
        }
        let topo = Topology::load(&path).expect("live bento.toml must load");
        // Syntax only: the host functions are bound per run against a live
        // context, and Rhai resolves calls at eval time regardless.
        let engine = rhai::Engine::new();
        let mut checked = 0;
        for (name, cfg) in &topo.app {
            let dir = crate::engine::expand_tilde(&cfg.repo).join(&cfg.recipe_dir);
            // Every recipe in the directory, not just the ones the app's current
            // targets name. A recipe for a target that is temporarily not
            // shipped (windows, dropped from the manifests until its host is
            // real) still has to parse, and it is the one nothing else is
            // watching.
            let Ok(entries) = std::fs::read_dir(&dir) else {
                continue;
            };
            let mut files: Vec<PathBuf> = entries
                .filter_map(Result::ok)
                .map(|e| e.path())
                .filter(|p| p.extension().is_some_and(|x| x == "rhai"))
                .collect();
            files.sort();
            for p in files {
                let file = p.file_name().unwrap_or_default().to_string_lossy();
                let Ok(src) = std::fs::read_to_string(&p) else {
                    continue;
                };
                engine
                    .compile(&src)
                    .unwrap_or_else(|e| panic!("recipe {name}/{file} does not parse: {e}"));
                checked += 1;
            }
        }
        assert!(checked > 0, "live config resolved no readable recipes");
    }
}