//! 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: pin every host that will build a target for this release
/// to 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.
///
/// Returns the branch each host was on before it was pinned, for
/// [`restore_branches`] to put back once the build settles, and the commit they
/// all agreed on. A host that is ALREADY detached has no branch to return to,
/// and is refused: that state means some earlier release never cleaned up, and
/// any commits made in the meantime are sitting on no branch at all.
///
/// 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.
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<String> = Vec::new();
    for t in targets {
        let h = state
            .topo
            .host_for(*t)
            .ok_or_else(|| anyhow::anyhow!("release preflight: no host can build {t}"))?;
        if !hosts.contains(&h.name) {
            hosts.push(h.name.clone());
        }
    }
    anyhow::ensure!(
        !hosts.is_empty(),
        "release preflight: no build hosts for v{version}; refusing to build a release \
         nothing was pinned for"
    );

    let mut shas: Vec<(String, String)> = Vec::new();
    let mut branches: Vec<(String, String)> = Vec::new();
    for host in &hosts {
        let exec = state
            .executors
            .get(host)
            .ok_or_else(|| anyhow::anyhow!("no executor for host `{host}`"))?;
        let mut sink = DiscardSink;
        // 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(host);

        // Read the branch BEFORE pinning, while there is still one to read.
        let branch_cmd = OpStep::shell(Action::Build, engine::git_current_branch_cmd(repo));
        let out = exec
            .run_streaming(&branch_cmd, &mut sink)
            .await
            .with_context(|| format!("release preflight: reading branch on `{host}`"))?;
        let branch = String::from_utf8_lossy(&out.stdout).trim().to_string();
        // `symbolic-ref -q` exits non-zero BOTH on a detached HEAD and when it
        // could not read the repo at all (wrong path, no checkout, no git).
        // Those need opposite responses, so a stderr that says anything is
        // reported as itself rather than folded into the detached-HEAD advice.
        let why = String::from_utf8_lossy(&out.stderr).trim().to_string();
        anyhow::ensure!(
            out.status.success() || why.is_empty(),
            "release preflight: reading the branch of `{repo}` on `{host}` failed: {why}"
        );
        anyhow::ensure!(
            out.status.success() && !branch.is_empty(),
            "release preflight: `{repo}` on `{host}` is on a detached HEAD, so there is \
             no branch to restore it to after the release. An earlier release left it \
             that way; any commits made since are on no branch and may exist nowhere \
             else. Reattach it (`git checkout <branch>`, fast-forwarding if the commits \
             should be kept) before releasing."
        );
        branches.push((host.clone(), branch));

        // A dirty tree defeats the pin silently, which is worse than not pinning
        // at all. `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 host with edits in the working tree builds those edits
        // while reporting the tagged sha, and a second host with a clean tree
        // builds something else. The rev-parse barrier below cannot see it:
        // both hosts genuinely are on the same commit. Two architectures, two
        // different binaries, one tag containing neither.
        let dirty_cmd = OpStep::shell(Action::Build, engine::git_dirty_cmd(repo));
        let out = exec
            .run_streaming(&dirty_cmd, &mut sink)
            .await
            .with_context(|| format!("release preflight: reading tree state on `{host}`"))?;
        let dirty = String::from_utf8_lossy(&out.stdout);
        let dirty: Vec<&str> = dirty
            .lines()
            .map(str::trim)
            .filter(|l| !l.is_empty())
            .collect();
        anyhow::ensure!(
            dirty.is_empty(),
            "release preflight: `{repo}` on `{host}` has uncommitted changes to tracked \
             files, so the build would not be the tagged commit:\n  {}\nCommit or stash \
             them before releasing.",
            dirty.join("\n  "),
        );

        // Advisory: `fetch --all` is non-zero if any one of a repo's remotes is
        // unreachable, and blaming the tag for a dead mirror is what made this
        // preflight misreport. Only the checkout below is allowed to fail.
        let fetch = OpStep::shell(Action::Build, engine::git_fetch_cmd(repo));
        let _ = exec.run_streaming(&fetch, &mut sink).await;

        let checkout = OpStep::shell(Action::Build, engine::git_checkout_tag_cmd(repo, &tag));
        let out = exec
            .run_streaming(&checkout, &mut sink)
            .await
            .with_context(|| format!("release preflight: checkout on `{host}`"))?;
        if !out.status.success() {
            // Ask git why before telling the operator. The two causes need
            // opposite responses: tag-and-push, versus clean the working tree.
            let probe = OpStep::shell(Action::Build, engine::git_tag_exists_cmd(repo, &tag));
            let tag_exists = exec
                .run_streaming(&probe, &mut sink)
                .await
                .is_ok_and(|o| o.status.success());
            anyhow::bail!(
                "release preflight: `git checkout {tag}` failed on `{host}`: {}",
                engine::checkout_failure_reason(&tag, tag_exists)
            );
        }

        let rev = OpStep::shell(Action::Build, engine::git_rev_parse_cmd(repo));
        let out = exec
            .run_streaming(&rev, &mut sink)
            .await
            .with_context(|| format!("release preflight: rev-parse on `{host}`"))?;
        anyhow::ensure!(
            out.status.success(),
            "release preflight: `git rev-parse HEAD` failed on `{host}`"
        );
        let sha = String::from_utf8_lossy(&out.stdout).trim().to_string();
        shas.push((host.clone(), sha));
    }

    // 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| !shas.iter().any(|(sh, _)| sh == *h))
        .map(String::as_str)
        .collect();
    anyhow::ensure!(
        missing.is_empty(),
        "release preflight: {} 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> = shas
        .iter()
        .filter(|(_, sha)| sha.is_empty())
        .map(|(h, _)| h.as_str())
        .collect();
    anyhow::ensure!(
        empty_shas.is_empty(),
        "release preflight: `git rev-parse HEAD` returned nothing on {} for v{version}",
        empty_shas.join(", "),
    );

    let (first_host, first_sha) = &shas[0];
    let mismatch: Vec<String> = shas
        .iter()
        .filter(|(_, sha)| sha != first_sha)
        .map(|(h, sha)| format!("{h}={}", short(sha)))
        .collect();
    anyhow::ensure!(
        mismatch.is_empty(),
        "release preflight: build hosts are on different commits for v{version} \
         ({}={}, {}); refusing to build a release from mixed sources",
        first_host,
        short(first_sha),
        mismatch.join(", "),
    );
    Ok(Pinned {
        branches,
        sha: first_sha.clone(),
    })
}

/// What the preflight established: where each host's checkout was, and the one
/// commit every host is now on.
struct Pinned {
    /// `(host, branch)` for [`restore_branches`].
    branches: Vec<(String, String)>,
    /// The commit all hosts agreed on. Empty when pinning is off (tests).
    sha: String,
}

/// Put every host's checkout back on the branch [`pin_release`] found it on.
///
/// Best-effort and non-fatal: the release itself is already decided by the time
/// this runs, and failing a green build because a `git checkout` did not take
/// would be worse than the detached tree it is cleaning up. A failure is logged
/// loudly instead, because the state it leaves behind is the silent one.
async fn restore_branches(state: &AppState, app: &AppId, branches: &[(String, String)]) {
    let Some(cfg) = state.topo.app(app) else {
        return;
    };
    for (host, branch) in branches {
        let Some(exec) = state.executors.get(host) else {
            continue;
        };
        // Serialize against a target still holding this host for its recipe run,
        // so the restore can't move the tree mid-build.
        let _host_guard = match state.host_locks.get(host).cloned() {
            Some(lock) => Some(lock.lock_owned().await),
            None => None,
        };
        let mut sink = DiscardSink;
        // The same per-host path `pin_release` detached; restoring the daemon's
        // path on a host that keeps its checkout elsewhere would leave the real
        // one detached and report success.
        let repo = cfg.repo_for(host);
        let step = OpStep::shell(Action::Build, engine::git_restore_branch_cmd(repo, branch));
        match exec.run_streaming(&step, &mut sink).await {
            Ok(out) if out.status.success() => {
                tracing::debug!(%host, %branch, "restored checkout to its branch");
            }
            Ok(_) => tracing::error!(
                %host, %branch, %repo,
                "could not restore the checkout to its branch; it is left DETACHED at the \
                 release tag, and commits made there will belong to no branch"
            ),
            Err(e) => tracing::error!(
                %host, %branch, %repo, error = %e,
                "could not restore the checkout to its branch; it is left DETACHED at the \
                 release tag, and commits made there will belong to no branch"
            ),
        }
    }
}

/// 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> {
    // Preflight: every version source in the repo must agree with the version
    // being built, before any host pulls or 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.
    if let Some(cfg) = state.topo.app(&app) {
        engine::check_version_consistency(&cfg.repo, cfg.version_path.as_deref(), &version)
            .context("version preflight")?;
    }

    // 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 = if state.cfg.pin_release_sha {
        pin_release(&state, &app, &version, &targets)
            .await
            .context("release preflight")?
    } else {
        Pinned {
            branches: Vec::new(),
            sha: String::new(),
        }
    };

    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.sha.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, and put the pinned
    // checkouts back on their branches. Spawned so /build returns immediately.
    tokio::spawn(finalize_build(state, build_id, set, app, pinned.branches));
    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_sha: String,
) {
    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();
    let repo = state
        .topo
        .app(&app)
        .map(|a| a.repo.clone())
        .unwrap_or_default();
    let repo_by_host = state
        .topo
        .app(&app)
        .map(|a| a.repo_by_host.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<()>,
    app: AppId,
    pinned_branches: Vec<(String, String)>,
) {
    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 app_still_building = {
        let mut active = state.active.lock().await;
        active.retain(|_, slot| slot.build_id != build_id);
        crate::metrics::set_in_flight(active.len());
        active.keys().any(|(a, _)| a == &app)
    };

    // Undo the preflight's pin. Skipped if a superseding build for this same app
    // still holds the checkout — it pinned the tree to ITS tag, and restoring the
    // branch here would move that build off the commit it is releasing.
    if !pinned_branches.is_empty() {
        if app_still_building {
            tracing::debug!(
                build_id, %app,
                "leaving the checkout pinned; a superseding build for this app is still running"
            );
        } else {
            restore_branches(&state, &app, &pinned_branches).await;
        }
    }

    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,
    ) {
        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"), "targets = [\"linux/x86_64\"]\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]);
        }
    }

    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\
             [app.demo]\nrepo = \"{}\"\n",
            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})");
    }

    /// 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();
        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());
        "#
        .replace("REPO", repo.to_str().unwrap());
        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();
        let topo = Topology::from_str_for_tests(&format!(
            "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             pull_root = \"{repo}\"\n[app.demo]\nrepo = \"{repo}\"\n",
            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");
    }

    /// The preflight refuses a host whose tree has uncommitted changes to
    /// tracked files.
    ///
    /// This is the hole the rev-parse barrier cannot see. `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 builds its edits
    /// while honestly reporting the tagged sha, and a clean host builds
    /// something else: two architectures, two different binaries, one tag
    /// containing neither combination. Caught on pom's first release, where
    /// fw13 had uncommitted changes in the serve path and astra did not.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn release_preflight_refuses_a_dirty_working_tree() {
        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");
        let body = std::fs::read_to_string(&tracked).unwrap();
        std::fs::write(&tracked, format!("{body}\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();
        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("uncommitted changes") && msg.contains("tauri.conf.json"),
            "the error must name the host and the files, 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");
    }

    /// 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");
    }

    /// The branch a release was launched from is still checked out afterwards.
    ///
    /// The preflight pins every host to `v<version>`, which detaches HEAD. That
    /// is correct during the build and wrong to leave behind: git does not warn,
    /// and later commits succeed while belonging to no branch. makeover shipped
    /// 2.3.0 from exactly that state on 2026-07-28 — the published commit existed
    /// only as a detached HEAD on one machine, on no branch and no remote.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_release_leaves_the_checkout_on_its_branch() {
        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);
        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})");

        // finalize_build restores after the targets settle, so give it a moment.
        for _ in 0..50 {
            if current_branch(&repo) == branch_before {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert_eq!(
            current_branch(&repo),
            branch_before,
            "the release must put the checkout back on its branch, not leave it detached"
        );
    }

    /// A checkout that is ALREADY detached is refused, rather than released from
    /// and left detached again. It means an earlier release never cleaned up, and
    /// anything committed since is on no branch — which is precisely the state
    /// that has to be looked at by a human before more releases pile onto it.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_detached_checkout_is_refused_before_it_is_pinned_again() {
        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 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("detached HEAD"),
            "the refusal should name the detached checkout, 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");
    }

    /// The branch `repo` is on, empty on a detached HEAD.
    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");
    }
}