//! Build orchestration: fan a `(app, version)` out across its targets, each
//! running its recipe concurrently on the host that can build it.
//!
//! A build inserts one `builds` row, then spawns one task per target. Each
//! target task registers itself in the single-slot guard (a newer build for
//! the same `(app, target)` aborts the in-flight one — latest wins, but other
//! targets keep running, which is the fan-out), then runs the recipe.

use crate::domain::{AppId, Status, Step, Target, Version};
use crate::engine::{self, RecipeCtx};
use crate::events::{self, Event};
use crate::state::AppState;
use anyhow::{Context, Result};
use ops_exec::{Action, LogSink, Step as OpStep};
use std::path::PathBuf;
use std::sync::Arc;

/// A [`LogSink`] that drops what it's handed — the release preflight runs git on
/// each host for its exit code and (via a separate `rev-parse`) the sha in
/// `RunOutput`, not for a streamed log, so there is no step to stream into.
struct DiscardSink;

#[async_trait::async_trait]
impl LogSink for DiscardSink {
    async fn write_chunk(&mut self, _bytes: &[u8]) {}
}

/// Preflight barrier: put every host that will build a target for this release
/// into a worktree detached at the tag `v<version>`, and refuse the build unless
/// they all report the SAME commit. Recipes used to `git pull --ff-only` per
/// host, so `mbp`/`astra`/`fw13` each built whatever `main` was at pull time and
/// the artifacts were filed under the daemon host's version. This runs before
/// any target task spawns, so a mixed-source release is stopped before a single
/// artifact is built.
///
/// "All" means all: a host that did not report is a refusal, not an abstention.
/// Comparing only the hosts that answered proves agreement among those, which is
/// not the claim the barrier is making.
///
/// The agreed sha is the release's provenance, and this is the only place it is
/// known to be one value rather than a per-host answer. Resolving it here and
/// carrying it forward is what lets an artifact record name the source it was
/// built from without asking a build host to be honest about it afterwards.
///
/// ## The tree a release builds in is Bento's, not the operator's
///
/// This used to check the tag out in the ordinary checkout and put that tree
/// back on its branch afterwards. Three things came out of that, and all three
/// are gone with it (ruled 2026-08-24, task `6b808f26`):
///
/// - A release pinned a tree somebody works in. Editing in Helix while one ran
///   meant editing the tag's content for the length of the build.
/// - The build dirtied that tree (`cargo` rewrites `Cargo.lock` under the
///   `[patch]` block), so the restore's `git checkout <branch>` failed and left
///   the tree detached at the tag — and the next release was refused for a
///   detached HEAD, or for an uncommitted change nobody made.
/// - The preflight therefore had to refuse a dirty or already-detached checkout,
///   which is a release blocked by the state of a tree it no longer reads.
///
/// A worktree costs the working files only — it shares the repository's object
/// store — and is re-pinned with `--force`, which is safe precisely because
/// nothing but Bento writes there. Where it lives is [`Host::worktree_root`],
/// and that doc carries why the path has to sit under `~/Code`.
///
/// Returns where each host will build, for the recipes to be pointed at, and the
/// commit they all agreed on. Nothing to unwind on a refusal: a host prepared
/// before the refusal is left holding a worktree at a tag, which is what the
/// next release would put it at anyway.
async fn pin_release(
    state: &AppState,
    app: &AppId,
    version: &Version,
    targets: &[Target],
) -> Result<Pinned> {
    let cfg = state
        .topo
        .app(app)
        .ok_or_else(|| anyhow::anyhow!("unknown app `{app}`"))?;
    // Spelled per app: a repo holding one product tags `v0.4.1`, a repo holding
    // several tags `pom-v0.4.1`. See `AppConfig::tag_format`.
    let tag = cfg.tag_for(version);

    // The distinct hosts across the requested targets (order-stable).
    //
    // A target whose host the topology cannot resolve is refused rather than
    // skipped. Skipping it would leave that target out of the barrier while it
    // still builds, which is the shape of hole this whole preflight exists to
    // close: agreement proven among some hosts reads as agreement among all.
    // `resolve_targets` rejects an unbuildable target before `/build`, so this
    // is unreachable through the API and cheap to state anyway.
    let mut hosts: Vec<&crate::topology::Host> = Vec::new();
    for t in targets {
        let h = state
            .topo
            .host_for(*t)
            .ok_or_else(|| anyhow::anyhow!("no host can build {t}"))?;
        if !hosts.iter().any(|seen| seen.name == h.name) {
            hosts.push(h);
        }
    }
    anyhow::ensure!(
        !hosts.is_empty(),
        "no build hosts for v{version}; refusing to build a release nothing was pinned for"
    );

    // What each host IS, before what each host is ON. A build host that was
    // rebuilt into something else is refused here rather than producing an
    // artifact whose libc floor nobody expected. Ordered before the worktree
    // work because it is one cheap round-trip and because a machine that is not
    // what the topology says it is should not be asked to check anything out.
    for host in &hosts {
        check_host_identity(state, host).await?;
    }

    let mut prepared: Vec<PinnedHost> = Vec::new();
    for host in &hosts {
        prepared.push(prepare_worktree(state, app, cfg, host, &tag).await?);
    }

    // The barrier: every participating host must have reported, and every report
    // must be the same commit. Reporting is checked first and separately from
    // agreement, because a barrier that compares only what it received cannot
    // tell unanimity from silence — three hosts agreeing while a fourth was
    // never asked is exactly the mixed-source release this refuses.
    let missing: Vec<&str> = hosts
        .iter()
        .filter(|h| !prepared.iter().any(|p| p.host == h.name))
        .map(|h| h.name.as_str())
        .collect();
    anyhow::ensure!(
        missing.is_empty(),
        "{} did not report a commit for v{version}; refusing to build a release the \
         barrier cannot vouch for",
        missing.join(", "),
    );
    let empty_shas: Vec<&str> = prepared
        .iter()
        .filter(|p| p.sha.is_empty())
        .map(|p| p.host.as_str())
        .collect();
    anyhow::ensure!(
        empty_shas.is_empty(),
        "`git rev-parse HEAD` returned nothing on {} for v{version}",
        empty_shas.join(", "),
    );

    let first = &prepared[0];
    let mismatch: Vec<String> = prepared
        .iter()
        .filter(|p| p.sha != first.sha)
        .map(|p| format!("{}={}", p.host, short(&p.sha)))
        .collect();
    anyhow::ensure!(
        mismatch.is_empty(),
        "build hosts are on different commits for v{version} \
         ({}={}, {}); refusing to build a release from mixed sources",
        first.host,
        short(&first.sha),
        mismatch.join(", "),
    );

    let sha = first.sha.clone();
    Ok(Pinned {
        build_dirs: prepared
            .into_iter()
            .map(|p| (p.host, p.build_dir))
            .collect(),
        sha,
    })
}

/// Confirm a build host is what the topology declares it to be.
///
/// The sha barrier below proves every host is on the same source. This proves
/// every host is the machine the topology thinks it is, which the sha barrier
/// cannot see: three hosts can agree on a commit while one of them was rebuilt
/// last week onto a base whose libc is two versions ahead, and the only symptom
/// is a binary that will not start on the box it ships to.
///
/// A host that declares neither field is skipped and logged as skipped. The
/// macOS and Windows hosts are legitimately in that state, so a refusal would
/// take those pipelines down to buy nothing.
async fn check_host_identity(state: &AppState, host: &crate::topology::Host) -> Result<()> {
    if host.base_image.is_none() && host.libc.is_none() {
        tracing::info!(host = %host.name, "preflight: host declares no base image; not checked");
        return Ok(());
    }
    let exec = state
        .executors
        .get(&host.name)
        .ok_or_else(|| anyhow::anyhow!("no executor for host `{}`", host.name))?;
    let step = OpStep::shell(Action::Build, ops_core::base_image::probe_cmd());
    let mut sink = DiscardSink;
    let out = exec
        .run_streaming(&step, &mut sink)
        .await
        .with_context(|| format!("asking `{}` what it is", host.name))?;
    anyhow::ensure!(
        out.status.success(),
        "`{}` could not report its base image: {}",
        host.name,
        String::from_utf8_lossy(&out.stderr).trim(),
    );
    let reported = ops_core::base_image::parse_probe(&String::from_utf8_lossy(&out.stdout));
    let checked = ops_core::base_image::check(
        &host.name,
        host.base_image.as_ref(),
        host.libc.as_deref(),
        &reported,
    )
    .context("a build host is not what the topology declares it to be")?;
    if let Some(what) = checked {
        tracing::info!(host = %host.name, "preflight: identity checked, {what}");
    }
    Ok(())
}

/// Put one host's worktree at the release tag and report the commit it landed
/// on, creating the worktree the first time this app is released there.
///
/// The worktree's own path is derived on the host rather than assumed: a repo
/// holding several products is one `.git` over all of them, so `repo` for pom is
/// `~/Code/MNW/pom` and the tree to make a worktree of is `~/Code/MNW`. One
/// `rev-parse` answers both halves — which repository, and where the app sits
/// inside it.
async fn prepare_worktree(
    state: &AppState,
    app: &AppId,
    cfg: &crate::topology::AppConfig,
    host: &crate::topology::Host,
    tag: &str,
) -> Result<PinnedHost> {
    let exec = state
        .executors
        .get(&host.name)
        .ok_or_else(|| anyhow::anyhow!("no executor for host `{}`", host.name))?;
    let name = &host.name;
    // Resolved per host, not once for the set: the checkouts need not be at the
    // same path on every machine, and pinning windows-x86 with the unix path
    // failed on the first git command of the release.
    let repo = cfg.repo_for(name);

    let read = |cmd: String| async {
        let step = OpStep::shell(Action::Build, cmd);
        let mut sink = DiscardSink;
        exec.run_streaming(&step, &mut sink).await
    };

    let out = read(engine::git_toplevel_and_prefix_cmd(repo))
        .await
        .with_context(|| format!("reading `{repo}` on `{name}`"))?;
    let stderr = String::from_utf8_lossy(&out.stderr).trim().to_string();
    anyhow::ensure!(
        out.status.success(),
        "`{repo}` on `{name}` is not a git checkout: {stderr}"
    );
    let (toplevel, prefix) = engine::parse_toplevel_and_prefix(&String::from_utf8_lossy(
        &out.stdout,
    ))
    .ok_or_else(|| anyhow::anyhow!("could not read where `{repo}` is checked out on `{name}`"))?;

    let worktree = host
        .worktree_for(engine::repo_dir_name(&toplevel), app)
        .ok_or_else(|| {
            anyhow::anyhow!(
                "host `{name}` sets no worktree_root, so there is nowhere to build \
                 tagged source"
            )
        })?;

    // Advisory, and deliberately run against the ordinary checkout: a worktree
    // shares its repository's object store, so a tag fetched here is a tag the
    // worktree can be put at. `fetch --all` exits non-zero if ANY remote fails
    // and the library repos carry three, so a dead mirror must not decide a
    // release. Only the checkout below is allowed to fail.
    let _ = read(engine::git_fetch_cmd(repo)).await;

    let exists = read(engine::git_worktree_probe_cmd(&worktree))
        .await
        .is_ok_and(|o| o.status.success());
    let pin = if exists {
        engine::git_worktree_pin_cmd(&worktree, tag)
    } else {
        // Prune first: a worktree directory somebody deleted by hand is still
        // registered in the repository, and `worktree add` then refuses the path
        // as already in use rather than rebuilding it.
        let _ = read(engine::git_worktree_prune_cmd(&toplevel)).await;
        engine::git_worktree_add_cmd(&toplevel, &worktree, tag)
    };
    let out = read(pin)
        .await
        .with_context(|| format!("pinning the worktree on `{name}`"))?;
    if !out.status.success() {
        // Ask git why before telling the operator. The usual answer is a tag
        // that was never created or never pushed, which is a different job from
        // anything about the worktree itself.
        let tag_exists = read(engine::git_tag_exists_cmd(repo, tag))
            .await
            .is_ok_and(|o| o.status.success());
        anyhow::bail!(
            "could not put `{worktree}` on `{name}` at {tag}: {}",
            engine::worktree_failure_reason(tag, tag_exists, &String::from_utf8_lossy(&out.stderr))
        );
    }

    let out = read(engine::git_rev_parse_cmd(&worktree))
        .await
        .with_context(|| format!("rev-parse on `{name}`"))?;
    anyhow::ensure!(
        out.status.success(),
        "`git rev-parse HEAD` failed in `{worktree}` on `{name}`"
    );

    Ok(PinnedHost {
        host: name.clone(),
        build_dir: engine::app_dir_in_worktree(&worktree, &prefix),
        sha: String::from_utf8_lossy(&out.stdout).trim().to_string(),
    })
}

/// What the preflight established: where each host builds this release, and the
/// one commit every one of those trees is on.
struct Pinned {
    /// `(host, directory the recipe builds in)`. Becomes the run's
    /// `repo_by_host`, so `repo()` in a recipe is the worktree and no recipe
    /// knows this happened.
    build_dirs: Vec<(String, String)>,
    /// The commit all hosts agreed on. Empty when pinning is off (tests).
    sha: String,
}

/// One host's prepared worktree, before the barrier has compared them.
struct PinnedHost {
    host: String,
    /// The app's own directory inside the worktree — the worktree root for a
    /// repo holding one product, `<worktree>/pom` for one holding several.
    build_dir: String,
    sha: String,
}

/// Confirm the tag states the version being built, in every file that states a
/// version.
///
/// Read out of the tag with `git show` on a build host, so it needs no worktree
/// path of its own and no assumption that the daemon has a copy of the repo:
/// one host is enough, since the barrier has already proven they are all on the
/// same commit.
///
/// Skipped when nothing was pinned, which is `pin_release_sha = false` — tests,
/// whose repos are plain directories with no tag to read.
async fn check_version_at_tag(
    state: &AppState,
    app: &AppId,
    version: &Version,
    pinned: &Pinned,
) -> Result<()> {
    let Some((host, dir)) = pinned.build_dirs.first() else {
        return Ok(());
    };
    let Some(cfg) = state.topo.app(app) else {
        return Ok(());
    };
    let Some(exec) = state.executors.get(host) else {
        return Ok(());
    };
    let tag = cfg.tag_for(version);
    let version_path = cfg.version_path.as_deref();

    let mut sources: Vec<(String, String)> = Vec::new();
    for rel in engine::version_sources(version_path) {
        let step = OpStep::shell(Action::Build, engine::git_show_file_cmd(dir, &tag, &rel));
        let mut sink = DiscardSink;
        let out = exec
            .run_streaming(&step, &mut sink)
            .await
            .with_context(|| format!("reading {rel} from {tag} on `{host}`"))?;
        if out.status.success() {
            sources.push((rel, String::from_utf8_lossy(&out.stdout).into_owned()));
        } else if version_path == Some(rel.as_str()) {
            // A file the app NAMES as its version source has to be there, or
            // the check would pass by failing to look.
            anyhow::bail!("`{rel}` is not in {tag} on `{host}`, but the app declares it");
        }
    }
    engine::versions_agree(
        &format!("{tag} (read on `{host}`)"),
        &sources,
        version_path,
        version,
    )
}

/// First 12 chars of a sha for a readable error.
fn short(sha: &str) -> &str {
    sha.get(..12).unwrap_or(sha)
}

/// Resolve the version to build: explicit, or read from `tauri.conf.json`.
///
/// Returns typed errors so the route maps user mistakes (unknown app, bad
/// version string) to 400, and only genuine daemon-side failures (reading the
/// app's `tauri.conf.json`) to 500.
pub fn resolve_version(
    state: &AppState,
    app: &AppId,
    explicit: Option<String>,
) -> crate::error::Result<Version> {
    use crate::error::Error;
    if let Some(v) = explicit {
        return Version::parse(&v).map_err(Error::BadRequest);
    }
    let cfg = state
        .topo
        .app(app)
        .ok_or_else(|| Error::BadRequest(format!("unknown app `{app}`")))?;
    engine::version_from_repo(&cfg.repo, cfg.version_path.as_deref()).map_err(Error::Other)
}

/// Validate + default the target list against what the app ships. Unknown app,
/// an unshipped target, or a target no host can build are all client errors
/// (400), not server errors (500).
pub fn resolve_targets(
    state: &AppState,
    app: &AppId,
    requested: Vec<Target>,
) -> crate::error::Result<Vec<Target>> {
    use crate::error::Error;
    let cfg = state
        .topo
        .app(app)
        .ok_or_else(|| Error::BadRequest(format!("unknown app `{app}`")))?;
    if requested.is_empty() {
        return Ok(cfg.targets.clone());
    }
    for t in &requested {
        if !cfg.targets.contains(t) {
            return Err(Error::BadRequest(format!(
                "app `{app}` does not ship target {t}"
            )));
        }
        if state.topo.host_for(*t).is_none() {
            return Err(Error::BadRequest(format!("no host can build {t}")));
        }
    }
    Ok(requested)
}

/// Insert the build row and spawn per-target tasks. Returns the build id.
pub async fn start_build(
    state: AppState,
    app: AppId,
    version: Version,
    targets: Vec<Target>,
) -> Result<i64> {
    // Pin every build host to the release tag and verify they agree, before any
    // target task spawns. Off in tests (their repos aren't git checkouts).
    let pinned = Arc::new(if state.cfg.pin_release_sha {
        pin_release(&state, &app, &version, &targets)
            .await
            .context("release preflight")?
    } else {
        Pinned {
            build_dirs: Vec::new(),
            sha: String::new(),
        }
    });

    // Then: every version source in the TAG must agree with the version being
    // built, before a single host compiles. `version_from_repo` reads one file,
    // so a `tauri.conf.json`/`Cargo.toml` (or explicit-version) mismatch would
    // otherwise sail through and file artifacts under the wrong version.
    //
    // After the pin rather than before it, and read from the tag rather than
    // from the checkout, because those are two different trees now. Asking the
    // checkout answered a question about a tree the release does not build: a
    // working copy one commit ahead failed a release of the tag behind it, and
    // a `Cargo.toml` that disagreed with the tag it was tagged in passed.
    check_version_at_tag(&state, &app, &version, &pinned)
        .await
        .context("version preflight")?;

    let build_id: i64 = sqlx::query_scalar(
        "INSERT INTO builds (app, version, status, created_at) VALUES (?, ?, 'running', ?) RETURNING id",
    )
    .bind(app.as_str())
    .bind(version.to_string())
    .bind(chrono::Utc::now().to_rfc3339())
    .fetch_one(&state.pool)
    .await
    .context("insert build")?;

    events::emit(
        &state.events,
        Event::BuildRequested {
            app: app.clone(),
            version: version.clone(),
            targets: targets.clone(),
        },
    );
    crate::metrics::build_started();

    // Spawn every target into a JoinSet so finalize_build can await completion
    // event-driven (no DB polling) and observe each task's outcome (panic vs
    // clean) for supervision.
    let mut set: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
    for target in targets {
        let app = app.clone();
        let version = version.clone();
        let key = (app.clone(), target);
        let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));

        // Latest-wins, done as ONE critical section: supersede the prior occupant
        // (cooperatively cancel its recipe + abort its task) and install ours
        // without ever releasing the lock, so two concurrent /build+/retry for the
        // same (app, target) can't both pass "nothing to abort" and run together.
        let mut active = state.active.lock().await;
        if let Some(prev) = active.remove(&key)
            && !prev.abort.is_finished()
        {
            // The recipe runs on a blocking thread; abort() alone can't stop it,
            // so set the cooperative flag the engine checks at step boundaries.
            prev.cancel.store(true, std::sync::atomic::Ordering::SeqCst);
            prev.abort.abort();
            events::emit(
                &state.events,
                Event::TargetAborted {
                    app: app.clone(),
                    target,
                },
            );
        }
        let abort = set.spawn(run_target(
            state.clone(),
            build_id,
            app,
            version,
            target,
            cancel.clone(),
            pinned.clone(),
        ));
        active.insert(
            key,
            crate::state::ActiveSlot {
                build_id,
                abort,
                cancel,
            },
        );
        crate::metrics::set_in_flight(active.len());
    }

    // Mark the build done once all target tasks settle. Spawned so /build
    // returns immediately. Nothing to put back: the release built in Bento's own
    // worktrees, and the checkouts were never moved.
    tokio::spawn(finalize_build(state, build_id, set));
    Ok(build_id)
}

/// Run one target's recipe end to end, updating its `target_runs` row. `cancel`
/// is set when a newer build supersedes this one; the engine checks it at step
/// boundaries and before publish so a superseded recipe can't advance or ship.
async fn run_target(
    state: AppState,
    build_id: i64,
    app: AppId,
    version: Version,
    target: Target,
    cancel: Arc<std::sync::atomic::AtomicBool>,
    pinned: Arc<Pinned>,
) {
    let started = std::time::Instant::now();
    let target_run_id: i64 = match sqlx::query_scalar(
        "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
         VALUES (?, ?, ?, ?, 'running', ?) RETURNING id",
    )
    .bind(build_id)
    .bind(app.as_str())
    .bind(version.to_string())
    .bind(target.to_string())
    .bind(chrono::Utc::now().to_rfc3339())
    .fetch_one(&state.pool)
    .await
    {
        Ok(id) => id,
        Err(e) => {
            tracing::error!(%app, %target, error = %e, "could not create target_run");
            return;
        }
    };

    events::emit(
        &state.events,
        Event::TargetStart {
            app: app.clone(),
            version: version.clone(),
            target,
        },
    );

    let recipe_src = match read_recipe(&state, &app, target) {
        Ok(s) => s,
        Err(e) => {
            fail_target(
                &state,
                target_run_id,
                &app,
                &version,
                target,
                Step::Checkout,
                &format!("{e:#}"),
            )
            .await;
            crate::metrics::target_finished(
                &target.to_string(),
                "failed",
                started.elapsed().as_secs_f64(),
            );
            return;
        }
    };

    // Pre-flight the build host before running the recipe. For local/ssh hosts
    // this is a no-op; for the agent host (macOS in-session signing) it hits
    // `/health` so a dead ops-agent fails here with a clear message — matching
    // what the driver does — instead of erroring opaquely on the recipe's first
    // dispatch to that host.
    if let Some(host) = state.topo.host_for(target)
        && let Some(exec) = state.executors.get(&host.name)
        && let Err(e) = exec.preflight().await
    {
        fail_target(
            &state,
            target_run_id,
            &app,
            &version,
            target,
            Step::Checkout,
            &format!("{e:#}"),
        )
        .await;
        crate::metrics::target_finished(
            &target.to_string(),
            "failed",
            started.elapsed().as_secs_f64(),
        );
        return;
    }

    // Host + checkout path for this target, resolved from the topology. Both are
    // exposed to the recipe (`build_host()` / `repo()`); a missing host here is
    // unreachable (resolve_targets rejected unbuildable targets before /build),
    // but fall back rather than panic.
    let build_host = state
        .topo
        .host_for(target)
        .map(|h| h.name.clone())
        .unwrap_or_default();
    let build_host_ssh = state
        .topo
        .host_for(target)
        .map(|h| h.ssh.clone())
        .unwrap_or_default();
    // Where the recipe builds. The preflight prepared a worktree per host and
    // these are those paths, so `repo()` in a recipe is Bento's own tree and no
    // recipe knows the difference — which is the whole reason the checkout could
    // stop being what a release builds in. The topology's paths are the fallback
    // for `pin_release_sha = false`, which is tests: their repos are plain
    // directories with no git in them to make a worktree of.
    let repo_by_host: std::collections::HashMap<String, String> = if pinned.build_dirs.is_empty() {
        state
            .topo
            .app(&app)
            .map(|a| a.repo_by_host.clone())
            .unwrap_or_default()
    } else {
        pinned.build_dirs.iter().cloned().collect()
    };
    // The daemon-local path, which is a different question: `crate_preflight`
    // runs `cargo metadata` on the daemon's own box rather than on a build host.
    // Point it at the local host's worktree when there is one, so it reads the
    // tagged manifest the release is publishing and not whatever the working
    // copy happens to say.
    let repo = state
        .topo
        .hosts
        .iter()
        .find(|h| h.ssh == "local")
        .and_then(|h| repo_by_host.get(&h.name).cloned())
        .or_else(|| state.topo.app(&app).map(|a| a.repo.clone()))
        .unwrap_or_default();
    let features = state
        .topo
        .app(&app)
        .map(|a| a.features.clone())
        .unwrap_or_default();
    let tag = state
        .topo
        .app(&app)
        .map_or_else(|| format!("v{version}"), |a| a.tag_for(&version));
    // The opt-in all-targets-green publish gate: pass the declared target set
    // (so `publish` can require every sibling green) only when the app turns it
    // on; otherwise None leaves independent per-target publishing unchanged.
    let all_green_required = state
        .topo
        .app(&app)
        .and_then(|a| a.require_all_targets.then(|| a.targets.clone()));

    // A service's install destination for THIS target, plus an executor for it.
    //
    // The executor is added to a per-run copy of the map rather than to the
    // daemon-wide one, so a deploy grant on a production host exists only for
    // the duration of the run that needs it and only for the app that declared
    // it. Nothing else can address that host: it is not in the topology, so no
    // other app's recipe can name it.
    let deploy = state
        .topo
        .app(&app)
        .and_then(|a| a.deploy_for(target))
        .cloned();
    let execs = match &deploy {
        Some(d) => {
            let mut map = (*state.executors).clone();
            map.insert(d.host.clone(), crate::state::build_deploy_executor(d));
            Arc::new(map)
        }
        None => state.executors.clone(),
    };

    let ctx = Arc::new(
        RecipeCtx::new(
            app.clone(),
            version.clone(),
            target,
            build_host,
            build_host_ssh,
            tag.clone(),
            repo,
            features,
            // Gates the `verify` step's capability: a library's crate preflight is
            // not an app's Gatekeeper check. See engine::action_for.
            state.topo.app(&app).map(|a| a.kind).unwrap_or_default(),
            target_run_id,
            execs,
            state.syncs.clone(),
            deploy,
            state.pool.clone(),
            state.events.clone(),
            state.cfg.clone(),
            state.ota.clone(),
            tokio::runtime::Handle::current(),
            cancel,
            all_green_required,
        )
        .with_repo_by_host(repo_by_host),
    );

    // Serialize per host: hold this host's lock for the whole recipe run so a
    // second target on the same box (goingson macos + ios both on mbp) can't
    // build concurrently in one checkout and corrupt the shared target/ +
    // keychain. Targets on different hosts hold different locks and still fan
    // out. Acquired at an await point, so a supersede-abort while queued drops
    // the task cleanly before it ever takes the lock. Not held during the
    // read-only preflight above.
    let _host_guard = match state.host_locks.get(&ctx.build_host).cloned() {
        Some(lock) => Some(lock.lock_owned().await),
        None => None,
    };

    // Rhai is synchronous; run the recipe (and its final step finalization) on
    // a blocking thread so host functions can `block_on` without sitting on a
    // runtime worker.
    let ctx_run = ctx.clone();
    let outcome = tokio::task::spawn_blocking(move || {
        let engine = engine::build_engine(&ctx_run);
        let res = engine.run(&recipe_src);
        let last_step = ctx_run.current_step();
        match &res {
            Ok(()) => {
                let _ = ctx_run.finish_step(Status::Ok);
            }
            Err(_) => {
                let _ = ctx_run.finish_step(Status::Failed);
            }
        }
        res.map_err(|e| (last_step, e.to_string()))
    })
    .await;

    // Write the artifact record before the run is stamped terminal. It describes
    // what was collected, so it runs whether the recipe succeeded or failed: a
    // build that signed and collected an artifact and then failed at `publish`
    // still produced bytes somebody may want the provenance of. Emit-only and
    // non-fatal; nothing reads it yet.
    let record_path = crate::artifact_record::emit(&state, &ctx, &pinned.sha).await;

    match outcome {
        Ok(Ok(())) => {
            // Hand the artifact to Sando before the run is stamped ok, so a
            // target reported green is one whose bytes reached the controller
            // that decides whether they ship. Only on success: an artifact from
            // a failed recipe is exactly what should not be offered for a
            // deploy, whatever it managed to collect on the way down.
            if let Err(e) =
                handoff_for(&state, record_path.as_deref(), &app, &version, target).await
            {
                let msg = format!("{e:#}");
                tracing::error!(%app, %target, error = %msg, "handing the artifact to sando failed");
                fail_target(
                    &state,
                    target_run_id,
                    &app,
                    &version,
                    target,
                    Step::Handoff,
                    &msg,
                )
                .await;
                crate::metrics::target_finished(
                    &target.to_string(),
                    "failed",
                    started.elapsed().as_secs_f64(),
                );
                return;
            }

            let artifacts = collected_artifacts(&state, &app, &version, target);
            if let Err(e) = sqlx::query(
                "UPDATE target_runs SET status = 'ok', current_step = NULL, finished_at = ? WHERE id = ?",
            )
            .bind(chrono::Utc::now().to_rfc3339())
            .bind(target_run_id)
            .execute(&state.pool)
            .await
            {
                // A swallowed terminal write would leave the row `running`;
                // finalize_build reconciles it, but log so the cause is visible.
                tracing::error!(%app, %target, error = %e, "could not stamp target_run ok");
            }
            crate::metrics::target_finished(
                &target.to_string(),
                "ok",
                started.elapsed().as_secs_f64(),
            );
            events::emit(
                &state.events,
                Event::TargetOk {
                    app,
                    version,
                    target,
                    artifacts,
                },
            );
        }
        Ok(Err((step, msg))) => {
            fail_target(&state, target_run_id, &app, &version, target, step, &msg).await;
            crate::metrics::target_finished(
                &target.to_string(),
                "failed",
                started.elapsed().as_secs_f64(),
            );
        }
        Err(join_err) => {
            // Task was aborted (superseded) or panicked.
            let status = if join_err.is_cancelled() {
                "aborted"
            } else {
                "failed"
            };
            crate::metrics::target_finished(
                &target.to_string(),
                status,
                started.elapsed().as_secs_f64(),
            );
            let msg = if join_err.is_cancelled() {
                "aborted (superseded)".to_string()
            } else {
                format!("recipe task panicked: {join_err}")
            };
            fail_target(
                &state,
                target_run_id,
                &app,
                &version,
                target,
                Step::Build,
                &msg,
            )
            .await;
        }
    }
}

async fn fail_target(
    state: &AppState,
    target_run_id: i64,
    app: &AppId,
    version: &Version,
    target: Target,
    step: Step,
    error: &str,
) {
    if let Err(e) = sqlx::query(
        "UPDATE target_runs SET status = 'failed', current_step = NULL, error = ?, finished_at = ? WHERE id = ?",
    )
    .bind(error)
    .bind(chrono::Utc::now().to_rfc3339())
    .bind(target_run_id)
    .execute(&state.pool)
    .await
    {
        tracing::error!(%app, %target, error = %e, "could not stamp target_run failed");
    }
    events::emit(
        &state.events,
        Event::TargetFailed {
            app: app.clone(),
            version: version.clone(),
            target,
            step,
            error: error.to_string(),
        },
    );
}

/// Await every target task of a build, then stamp the build's terminal status.
///
/// Event-driven: it joins the `JoinSet` rather than polling the DB. Each task's
/// outcome is supervised — a panicked task is logged (and its still-`running`
/// row is reconciled below), an aborted (superseded) task is expected.
///
/// Per-step deadlines (see `engine::step_budget`) are the real bound on a wedged
/// step now; this overall deadline is only a generous last-resort backstop for a
/// hang outside a bounded command. When it trips it sets each still-running
/// target's cooperative cancel FIRST — the blocking recipe bodies observe that
/// at their next bounded command or step boundary and stop signing — then aborts
/// the async wrappers. The old code aborted only the wrappers, which could not
/// reach the blocking bodies, so a build was marked failed while codesign kept
/// running on the mac.
async fn finalize_build(state: AppState, build_id: i64, mut set: tokio::task::JoinSet<()>) {
    const MAX_WAIT: std::time::Duration = std::time::Duration::from_hours(6);
    let deadline = tokio::time::Instant::now() + MAX_WAIT;
    loop {
        match tokio::time::timeout_at(deadline, set.join_next()).await {
            Ok(Some(Ok(()))) => {}
            Ok(Some(Err(e))) => {
                // A panic in run_target itself (outside its inner spawn_blocking,
                // which is already caught). Cancellation = a superseding build.
                if !e.is_cancelled() {
                    tracing::error!(build_id, error = %e, "target task panicked");
                }
            }
            Ok(None) => break, // all targets settled
            Err(_elapsed) => {
                tracing::error!(
                    build_id,
                    "finalize_build backstop deadline hit; cancelling then aborting remaining targets"
                );
                // Set the cooperative cancel on this build's still-running slots
                // BEFORE aborting, so the blocking recipe bodies actually stop
                // (abort() alone cannot reach a spawn_blocking body).
                {
                    let active = state.active.lock().await;
                    for slot in active.values().filter(|s| s.build_id == build_id) {
                        slot.cancel.store(true, std::sync::atomic::Ordering::SeqCst);
                    }
                }
                set.abort_all();
                while set.join_next().await.is_some() {}
                break;
            }
        }
    }

    // Reap this build's latest-wins slots (only our own — a superseding build
    // owns a different build_id and is left intact).
    {
        let mut active = state.active.lock().await;
        active.retain(|_, slot| slot.build_id != build_id);
        crate::metrics::set_in_flight(active.len());
    }

    let now = chrono::Utc::now().to_rfc3339();
    // Any row still `running` is a panicked/aborted task that never stamped its
    // terminal status — reconcile it so the build finalizes truthfully.
    if let Err(e) = sqlx::query(
        "UPDATE target_runs SET status = 'failed', \
         error = COALESCE(error, 'target task ended before stamping its status'), \
         finished_at = ? WHERE build_id = ? AND status = 'running'",
    )
    .bind(&now)
    .bind(build_id)
    .execute(&state.pool)
    .await
    {
        tracing::error!(build_id, error = %e, "could not reconcile straggling target_runs");
    }

    let failed: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM target_runs WHERE build_id = ? AND status = 'failed'",
    )
    .bind(build_id)
    .fetch_one(&state.pool)
    .await
    .unwrap_or(0);
    let status = if failed == 0 { "ok" } else { "failed" };
    if let Err(e) = sqlx::query("UPDATE builds SET status = ?, finished_at = ? WHERE id = ?")
        .bind(status)
        .bind(&now)
        .bind(build_id)
        .execute(&state.pool)
        .await
    {
        tracing::error!(build_id, error = %e, "could not stamp build terminal status");
    }
}

/// Read the recipe text for `(app, target)` from the app's checkout on the
/// daemon host.
///
/// Apps use one recipe per platform (`<recipe_dir>/<platform>.rhai`), since
/// what they produce differs by platform. A library produces one crate whatever
/// host uploads it, so it uses a single `publish.rhai` — a `linux.rhai` naming
/// a registry upload would imply a per-platform artifact that does not exist.
fn read_recipe(state: &AppState, app: &AppId, target: Target) -> Result<String> {
    let cfg = state
        .topo
        .app(app)
        .ok_or_else(|| anyhow::anyhow!("unknown app `{app}`"))?;
    let file = match cfg.kind {
        crate::topology::Kind::Library => "publish.rhai".to_string(),
        // A service is built per target like an app — the binary genuinely
        // differs per platform, and so does where it lands — so it takes the
        // same per-platform recipe naming rather than a single deploy.rhai.
        crate::topology::Kind::App | crate::topology::Kind::Service => {
            format!("{}.rhai", target.platform.as_str())
        }
    };
    let path: PathBuf = engine::expand_tilde(&cfg.repo)
        .join(&cfg.recipe_dir)
        .join(file);
    std::fs::read_to_string(&path).with_context(|| format!("reading recipe {}", path.display()))
}

/// Send this target's finished bundle to the Sando configured for the app, if
/// one is. Nothing configured is a no-op and the ordinary case.
///
/// A configured handoff with no record is an error rather than a skip. It means
/// the recipe succeeded and collected nothing, or that the paperwork could not
/// be written — and for an app whose whole point is being deployed by Sando, a
/// build that produced nothing to hand over is not a green build. Everywhere
/// else a missing record stays the non-event it was.
async fn handoff_for(
    state: &AppState,
    record_path: Option<&std::path::Path>,
    app: &AppId,
    version: &Version,
    target: Target,
) -> anyhow::Result<()> {
    if !state.cfg.handoff.contains_key(app.as_str()) {
        return Ok(());
    }
    let record_path = record_path.ok_or_else(|| {
        anyhow::anyhow!(
            "{app} hands off to sando, but this {target} run wrote no artifact record \
             (nothing was collected, or the record could not be written)"
        )
    })?;
    let dir = crate::archive::target_dir(&state.cfg.dist_root, app, version, target);
    crate::handoff::send(&state.cfg, &dir, record_path, app, version, target).await
}

/// What this target run left in its collect directory.
///
/// Per target: the event reports what THIS run produced, and every target of a
/// version used to share one directory, so a mac build's `TargetOk` listed the
/// Linux AppImage a sibling had collected minutes earlier.
fn collected_artifacts(
    state: &AppState,
    app: &AppId,
    version: &Version,
    target: Target,
) -> Vec<String> {
    let dir = crate::archive::target_dir(&state.cfg.dist_root, app, version, target);
    let Ok(rd) = std::fs::read_dir(&dir) else {
        return Vec::new();
    };
    rd.filter_map(std::result::Result::ok)
        .map(|e| e.file_name().to_string_lossy().into_owned())
        .collect()
}

#[cfg(test)]
mod tests {
    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) rather than wedging under the old whole-build guillotine. The
    /// per-step budget is overridden to 1s here; the recipe then sleeps 30s, so
    /// the step deadline — not the sleep — decides the outcome, quickly.
    #[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.
    ///
    /// The preflight used to refuse one, and had to: `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 built its edits
    /// while honestly reporting the tagged sha. Caught on pom's first release,
    /// where fw13 had uncommitted changes in the serve path and astra did not.
    ///
    /// The worktree answers it 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
    /// refusing a release because somebody has unsaved work is no longer a thing
    /// this does.
    #[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.
    ///
    /// It used to check the tag out there and put the branch back afterwards,
    /// which is a tree somebody works in being moved under them for the length of
    /// a build — and when the restore failed, left detached. makeover shipped
    /// 2.3.0 from exactly that state on 2026-07-28, the published commit on no
    /// branch and no remote. The worktree makes the whole question moot, so this
    /// asserts the strong form rather than the 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 no longer refuses a release.
    ///
    /// It used to, and had to: a detached tree meant an earlier release never
    /// cleaned up, and the release about to start had no branch to put it back
    /// on. Releases do not move that tree any more, so its HEAD is not their
    /// business — and the state that produced the refusal is one releases can no
    /// longer create.
    #[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. That used to be the operator's problem:
    /// the tree was the ordinary checkout, the branch could not be checked back
    /// out over the edit, and the next release was refused for a dirty tree or a
    /// detached HEAD. Three releases on astra on 2026-08-23, three manual
    /// cleanups, none of the diffs meaning anything.
    ///
    /// Now 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, which is what made
        // the old restore fail rather than silently carry the edit across.
        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 used to read the checkout, so releasing v0.0.1 while `main` had
    /// already moved to 0.0.2 was refused for "version drift" — about a tree the
    /// release does not build. It reads the tag now, and 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.
    ///
    /// This is the drift worth catching and the one the old check could not see:
    /// `tauri.conf.json` at 0.0.1 and `Cargo.toml` still at 0.0.2, committed and
    /// tagged that way. Reading the checkout would have compared against
    /// whatever the working copy happened 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. That used to leave it
    /// detached at the tag with nothing to restore it, and the next attempt
    /// refused it for a detached HEAD — true, describing its own last run, and
    /// reading to the operator as their own mistake. Hit three times releasing
    /// pom on 2026-08-09. There is no unwind to get right now: 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.
    ///
    /// This is the case the old design could not get right, in either direction.
    /// The dirty gate was scoped to the app's own directory on purpose — 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. So the gate passed and the checkout
    /// then failed, on a file the operator had been told was none of this
    /// release's business.
    ///
    /// A worktree is made of the repository too, so the release still gets the
    /// whole tagged tree — it just gets its own copy of it, 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.
    ///
    /// The preflight used to run `fetch --all --tags --prune && checkout`, and
    /// `fetch --all` is non-zero if ANY remote fails. Every library repo carries
    /// three (`astra`, `mnw`, `srht`), so one dead mirror aborted the release and
    /// blamed it on a missing tag — which is how makeover v2.1.1 failed on
    /// 2026-07-26 four minutes after its tag was created. Fetch is advisory now;
    /// 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");
    }
}