//! Rhai recipe engine + host-function API.
//!
//! A `(app, target)` resolves to a `.rhai` recipe composed from a shared step
//! vocabulary. The daemon embeds Rhai and registers the host functions recipes
//! call; the recipe is the orchestration, the host functions are the
//! privileged primitives (run a command, read a secret, collect artifacts,
//! publish). Recipes are otherwise sandboxed — no arbitrary FS/network except
//! through these functions — matching the Balanced Breakfast plugin model.
//!
//! Rhai is synchronous; the engine runs each recipe on a blocking thread
//! (`spawn_blocking`, see [`crate::runner`]) and host functions bridge to async
//! work via `Handle::block_on`. That is sound only off a runtime worker thread,
//! which `spawn_blocking` guarantees.

use crate::config::Config;
use crate::domain::{AppId, Status, Step, StepRunId, Target, Version};
use crate::events::{self, Event, EventTx};
use crate::ota::{OtaRegistry, PublishAuthority, Release};
use crate::state::ExecutorMap;
use crate::topology::{DeployTarget, Kind};
use anyhow::{Context as _, Result};
use ops_core::live_log::LiveLog;
use ops_exec::{Action, Executor, ObserveKind, Step as OpStep, SyncOpts};
use rhai::{Engine, EvalAltResult, Map};
use sha2::{Digest, Sha256};
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tokio::runtime::Handle;
use tokio::sync::Mutex as AsyncMutex;

/// The capability label for a command, derived from the open recipe step. A
/// recipe's `sh("mbp", …)` under `step("sign")` becomes an `Action::Sign`, gated
/// by the mac host's `sign` grant — so recipes stay unchanged while every command
/// is capability-checked at its transport. `Verify` is read-only (an observe).
/// The capability a step's commands are gated on.
///
/// `Verify` depends on what is being released, which is the one place this is
/// not a property of the step alone. An app's verify is a Gatekeeper check on a
/// signed bundle, and the `gatekeeper` observe is granted implicitly to hosts
/// that can `sign` (`CapabilitySet::from_tokens`) precisely so that pairing
/// holds. A library's verify is a crate preflight: it runs `cargo` on the build
/// host and asks the registry a question. Gating that on Gatekeeper asks a Linux
/// host for a macOS code-signing capability it can never honestly hold, and the
/// only way to satisfy it would be to declare the capability falsely.
fn action_for(step: Step, kind: Kind) -> Action {
    match step {
        Step::Checkout | Step::Prebuild | Step::Build => Action::Build,
        Step::Sign => Action::Sign,
        Step::Notarize => Action::Notarize,
        Step::Staple => Action::Staple,
        Step::Package => Action::Package,
        Step::Verify => match kind {
            Kind::App => Action::Observe(ObserveKind::Custom("gatekeeper".into())),
            // Running the build toolchain to inspect a crate or a service
            // binary, which is what `build` means on a host. Neither has a
            // bundle for Gatekeeper to have an opinion about.
            Kind::Library | Kind::Service => Action::Build,
        },
        // Publish/Collect/Handoff run on the daemon, not through a host
        // executor; this label only applies if a recipe runs a bare `sh` while
        // one is open. `handoff` is the daemon's own post-recipe motion and no
        // recipe should open it at all — naming it here costs nothing and beats
        // a wildcard that would silently absorb the next step somebody adds.
        Step::Publish | Step::Collect | Step::Handoff => Action::Package,
        // The one step that dispatches to a host OUTSIDE the build topology.
        // Every command a recipe runs while `deploy` is open — the install, the
        // restart, the health assertion — carries this action, so it reaches the
        // service host only through the deploy grant and reaches a build host
        // not at all (no build host is granted `deploy`).
        Step::Deploy => Action::Deploy,
    }
}

/// The currently-open step within a recipe run: its DB row id, which step it
/// is, and the live-log sink that `sh`/`log` stream into.
struct StepState {
    run_id: StepRunId,
    step: Step,
    log: Arc<AsyncMutex<LiveLog>>,
    /// Set when something in the step recorded a hard failure the recipe did
    /// not abort on (e.g. `verify_gatekeeper` rejected the artifact but the
    /// recipe ignored the bool). Forces the step's recorded status to `Failed`
    /// and bars `publish` (the step-success ledger).
    failed: bool,
    /// Wall-clock deadline for this step. A command that runs past it fails the
    /// step (and unwinds the recipe) rather than wedging under the old
    /// whole-build guillotine, which a legitimate 5-target fan-out plus notary
    /// queueing could trip — mismarking every target failed while the blocking
    /// recipe bodies kept signing.
    deadline: std::time::Instant,
}

/// Per-step wall-clock budget: a generous ceiling that catches a wedged command
/// (a hung ssh, a stuck notary poll) without killing legitimately slow work.
/// Bounding each step, rather than the whole build, keeps one slow step from
/// being blamed on another and keeps a fan-out of slow-but-fine targets from
/// being guillotined. `Config::step_timeout_secs` overrides these per-kind
/// defaults for every step; see [`RecipeCtx::step_budget`].
fn default_step_budget(step: Step) -> std::time::Duration {
    use std::time::Duration;
    let mins = match step {
        Step::Checkout => 10,
        // clippy + full test suite, cold, on a workspace.
        Step::Prebuild => 45,
        // cargo tauri build, cold, universal bundles.
        Step::Build => 90,
        Step::Sign => 15,
        // Apple's notary queue + this step's bounded retries.
        Step::Notarize => 60,
        Step::Staple => 10,
        Step::Package => 30,
        Step::Verify => 10,
        // rsync of multi-GiB artifacts off the build host.
        Step::Collect => 30,
        Step::Publish => 20,
        // A binary push, an install, a unit restart, and a health poll. Minutes
        // of work; the ceiling is for a wedged transport, not slow work.
        Step::Deploy => 15,
        // Unused by any recipe — the daemon runs the handoff itself, outside a
        // step's clock — and matched to `collect`, since it moves the same
        // bytes the same way and would wedge for the same reasons.
        Step::Handoff => 30,
    };
    Duration::from_secs(mins * 60)
}

/// Where a service's binary is staged on the host that will run it, before the
/// privileged installer moves it into place.
///
/// A fixed, unguessable-by-accident path rather than a recipe-chosen one,
/// because the installer refuses any source outside it. That refusal is the
/// only thing standing between the NOPASSWD sudo grant and `install`-as-root to
/// an arbitrary path, so both ends have to name the same constant. `/var/tmp`
/// rather than `/tmp` so a staged binary survives a `PrivateTmp` unit and a
/// systemd tmpfiles sweep between staging and install.
pub const DEPLOY_STAGING_ROOT: &str = "/var/tmp/bento-deploy";

/// Everything a recipe's host functions need, shared (Arc) into each closure.
pub struct RecipeCtx {
    pub app: AppId,
    pub version: Version,
    pub target: Target,
    /// Name of the host this target builds on (resolved from the topology by the
    /// runner). Recipes read it via `build_host()` so one per-platform recipe can
    /// dispatch to the right host across arches (linux x86_64 -> fw13, aarch64 ->
    /// astra) without hard-coding a host name.
    pub build_host: String,
    /// The build host's SSH destination (topology `ssh`), as opposed to its
    /// name. Deploy compares it against the service host's to tell "build and
    /// run on the same box" from "two hosts that need a transfer" — a question
    /// the host NAMES cannot answer, since a build host and a deploy
    /// destination are declared in different files and need not agree on one.
    pub build_host_ssh: String,
    /// This release's git tag, rendered from the app's `tag_format`. Held here
    /// rather than derived from the version because a repo holding several
    /// products spells it per product (`pom-v0.4.1`), and the recipe, the
    /// preflight barrier and the failure message all have to agree on it.
    pub tag: String,
    /// The app's default checkout path (topology `repo`, `~`-prefixed). Read it
    /// through [`RecipeCtx::repo_for`] for anything that runs on a build host;
    /// this field alone is the daemon-local answer.
    pub repo: String,
    /// Per-host checkout overrides (topology `repo_by_host`). Empty for every app
    /// that has not declared one, which is all of them but the Windows-shipping
    /// ones.
    pub repo_by_host: HashMap<String, String>,
    /// Cargo features this app's release builds enable (topology `features`).
    /// Recipes read it via `feature_flags()`.
    pub features: Vec<String>,
    /// App or library. Decides which capability a `verify` step is gated on;
    /// see [`action_for`].
    pub kind: Kind,
    pub target_run_id: i64,
    /// Capability-scoped executor per build host. Recipe commands dispatch
    /// through these — the transport (local / ssh / in-session agent) and the
    /// capability gate are the executor's, not the engine's.
    pub execs: Arc<ExecutorMap>,
    /// Sync transport per build host, used only by `collect` to pull artifacts
    /// back. Never the agent, even for an agent host — see `state::build_sync`.
    /// Not an execution path.
    pub syncs: Arc<ExecutorMap>,
    /// Where this target installs, for a `kind = "service"` app. `None` for an
    /// app or a library, which makes every deploy host function fail with that
    /// as the reason rather than with a missing-host error.
    ///
    /// The runner resolves it from the app manifest's `[[deploy]]` table and
    /// registers its executor into `execs` under the destination's host string,
    /// so `sh_ok(deploy_host(), ...)` reaches the service host through the same
    /// capability gate as everything else.
    pub deploy: Option<DeployTarget>,
    pub pool: SqlitePool,
    pub events: EventTx,
    pub cfg: Arc<Config>,
    pub ota: Arc<OtaRegistry>,
    pub rt: Handle,
    current: Mutex<Option<StepState>>,
    /// Gatekeeper verdict recorded by `verify_gatekeeper`: `None` = never run,
    /// `Some(false)` = ran and rejected, `Some(true)` = accepted. `publish`
    /// requires `Some(true)` for a macOS/iOS artifact (the proof it is signed +
    /// notarized).
    gatekeeper_ok: Mutex<Option<bool>>,
    /// Steps finalized as `Failed` during this run. A non-empty ledger bars
    /// `publish` — an artifact is never shipped after a step failed, even if the
    /// recipe ignored the failure and ran on.
    failed_steps: Mutex<Vec<Step>>,
    /// Set when a newer build supersedes this run. Checked at step boundaries
    /// and before publish so a superseded recipe stops promptly rather than
    /// running to completion (the blocking Rhai body can't be `abort()`ed).
    cancel: Arc<AtomicBool>,
    /// The all-targets-green publish gate (topology `require_all_targets`).
    /// `Some(declared)` ⇒ `publish` refuses unless every one of `declared` has a
    /// successful latest run for this `(app, version)`. `None` ⇒ gate off, each
    /// target publishes independently.
    all_green_required: Option<Vec<Target>>,
    /// sha256 of each artifact hashed at `collect`, keyed by file name. `publish`
    /// reads it to record `releases.artifact_hash` for the bytes it ships, so the
    /// hash is the one computed when the artifact landed rather than a re-read
    /// that could see a different file. Absent ⇒ `publish` hashes on demand.
    artifact_hashes: Mutex<HashMap<String, String>>,
}

impl RecipeCtx {
    /// Versions of `name` already on crates.io. A network failure yields an
    /// empty list: preflight then cannot claim a version is a duplicate, and
    /// `cargo publish` still refuses one, so the check degrades to advisory
    /// rather than blocking a release on registry availability.
    fn published_versions(name: &str) -> Vec<String> {
        let url = format!("https://crates.io/api/v1/crates/{name}");
        let Ok(out) = std::process::Command::new("curl")
            .args([
                "-sS",
                "--max-time",
                "15",
                "-H",
                "User-Agent: bento-preflight",
                &url,
            ])
            .output()
        else {
            return Vec::new();
        };
        let Ok(v) = serde_json::from_slice::<serde_json::Value>(&out.stdout) else {
            return Vec::new();
        };
        v.get("versions")
            .and_then(|x| x.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|x| x.get("num").and_then(|n| n.as_str()).map(str::to_string))
                    .collect()
            })
            .unwrap_or_default()
    }

    #[allow(clippy::too_many_arguments)]
    pub fn new(
        app: AppId,
        version: Version,
        target: Target,
        build_host: String,
        build_host_ssh: String,
        tag: String,
        repo: String,
        features: Vec<String>,
        kind: Kind,
        target_run_id: i64,
        execs: Arc<ExecutorMap>,
        syncs: Arc<ExecutorMap>,
        deploy: Option<DeployTarget>,
        pool: SqlitePool,
        events: EventTx,
        cfg: Arc<Config>,
        ota: Arc<OtaRegistry>,
        rt: Handle,
        cancel: Arc<AtomicBool>,
        all_green_required: Option<Vec<Target>>,
    ) -> Self {
        Self {
            app,
            version,
            target,
            build_host,
            build_host_ssh,
            tag,
            repo,
            repo_by_host: HashMap::new(),
            features,
            kind,
            target_run_id,
            execs,
            syncs,
            deploy,
            pool,
            events,
            cfg,
            ota,
            rt,
            current: Mutex::new(None),
            gatekeeper_ok: Mutex::new(None),
            failed_steps: Mutex::new(Vec::new()),
            cancel,
            all_green_required,
            artifact_hashes: Mutex::new(HashMap::new()),
        }
    }

    /// Declare the app's per-host checkout overrides (topology `repo_by_host`).
    ///
    /// Separate from `new` because it is empty for every app that has not opted
    /// in, and `new` already carries twenty arguments no test wants a
    /// twenty-first of.
    #[must_use]
    pub fn with_repo_by_host(mut self, repo_by_host: HashMap<String, String>) -> Self {
        self.repo_by_host = repo_by_host;
        self
    }

    /// Where this app is checked out on `host` (topology `AppConfig::repo_for`).
    ///
    /// Every git command a recipe or the engine runs on a build host goes
    /// through this. The bare `repo` field is the daemon-local path.
    pub fn repo_for(&self, host: &str) -> &str {
        self.repo_by_host
            .get(host)
            .map_or(self.repo.as_str(), String::as_str)
    }

    /// Whether a newer build has superseded this run.
    fn is_cancelled(&self) -> bool {
        self.cancel.load(Ordering::SeqCst)
    }

    fn now() -> String {
        chrono::Utc::now().to_rfc3339()
    }

    /// `<logs_root>/<app>/<version>/<target-with-slash-as-dash>/<step>.<run_id>.log`.
    ///
    /// The run id is in the filename because the rest of the key is not unique:
    /// re-running an app at a version it already built (a retry, or a rebuild of
    /// an already-published release) reopens the same path, and appending would
    /// show two runs' output with nothing marking the boundary. The
    /// step ledger records a run id per step, so keying the file on it makes a
    /// ledger row resolve to exactly one file and keeps the earlier run readable.
    fn log_path(&self, step: Step, run_id: StepRunId) -> PathBuf {
        self.log_dir()
            .join(format!("{}.{}.log", step.as_str(), run_id.0))
    }

    /// `<logs_root>/<app>/<version>/<target-with-slash-as-dash>/`.
    fn log_dir(&self) -> PathBuf {
        let target_dir = self.target.to_string().replace('/', "-");
        self.cfg
            .logs_root
            .join(self.app.as_str())
            .join(self.version.to_string())
            .join(target_dir)
    }

    /// Close the previous step (as `Ok`), open a new one: insert its DB row,
    /// open a live log whose chunks broadcast `StepLogChunk`, emit `StepStart`.
    fn begin_step(self: &Arc<Self>, step: Step) -> Result<()> {
        anyhow::ensure!(
            !self.is_cancelled(),
            "build superseded by a newer request; aborting before `{}`",
            step.as_str()
        );
        self.finish_step(Status::Ok)?;
        let me = self.clone();
        let started = Self::now();
        let started_for_header = started.clone();
        let run_id = self.rt.block_on(async move {
            // The step row and the target's current_step pointer are one logical
            // state — write them atomically so a failure can't leave a `running`
            // step row while current_step still names the previous step.
            let mut tx = me.pool.begin().await.context("begin step tx")?;
            // log_ref names the run id, which only exists once the row does, so
            // the path is written back inside the same transaction rather than
            // guessed beforehand.
            let id: i64 = sqlx::query_scalar(
                "INSERT INTO step_runs (target_run_id, step, status, started_at)
                 VALUES (?, ?, 'running', ?) RETURNING id",
            )
            .bind(me.target_run_id)
            .bind(step.as_str())
            .bind(&started)
            .fetch_one(&mut *tx)
            .await
            .context("insert step_run")?;
            let log_ref = me
                .log_path(step, StepRunId(id))
                .to_string_lossy()
                .into_owned();
            sqlx::query("UPDATE step_runs SET log_ref = ? WHERE id = ?")
                .bind(&log_ref)
                .bind(id)
                .execute(&mut *tx)
                .await
                .context("set step_run log_ref")?;
            sqlx::query("UPDATE target_runs SET current_step = ? WHERE id = ?")
                .bind(step.as_str())
                .bind(me.target_run_id)
                .execute(&mut *tx)
                .await
                .context("update current_step")?;
            tx.commit().await.context("commit step tx")?;
            anyhow::Ok(StepRunId(id))
        })?;

        // Live log: each chunk fans out as a StepLogChunk event keyed by run_id.
        let events = self.events.clone();
        let cb_run_id = run_id;
        let mut log = self.rt.block_on(LiveLog::open(
            self.log_path(step, run_id),
            Box::new(move |seq, text| {
                events::emit(
                    &events,
                    Event::StepLogChunk {
                        run_id: cb_run_id,
                        seq,
                        text: text.to_string(),
                    },
                );
            }),
        ));

        events::emit(
            &self.events,
            Event::StepStart {
                run_id,
                app: self.app.clone(),
                version: self.version.clone(),
                target: self.target,
                step,
            },
        );

        // A log that names its own run: reading one tells you which ledger row
        // it belongs to without going back to the DB, and a file that somehow
        // does get appended to still shows where the second run began. Written
        // after `StepStart` so its chunk event cannot precede the step it
        // belongs to.
        let header = format!(
            "=== bento {app} {version} {target} step={step} run_id={run_id} started={started_for_header} ===\n",
            app = self.app.as_str(),
            version = self.version,
            target = self.target,
            step = step.as_str(),
        );
        self.rt.block_on(async {
            use ops_core::remote::LogSink as _;
            log.write_chunk(header.as_bytes()).await;
        });

        *self.current.lock().unwrap() = Some(StepState {
            run_id,
            step,
            log: Arc::new(AsyncMutex::new(log)),
            failed: false,
            deadline: std::time::Instant::now() + self.step_budget(step),
        });
        Ok(())
    }

    /// This build's budget for `step`: the `Config` override if set, else the
    /// per-kind default.
    fn step_budget(&self, step: Step) -> std::time::Duration {
        self.cfg
            .step_timeout_secs
            .map_or_else(|| default_step_budget(step), std::time::Duration::from_secs)
    }

    /// The current step's wall-clock deadline (or a default if no step is open,
    /// which only happens before the first `step()` — commands then run under an
    /// implicit `Build` step opened by `ensure_step`).
    fn step_deadline(&self) -> std::time::Instant {
        self.current.lock().unwrap().as_ref().map_or_else(
            || std::time::Instant::now() + self.step_budget(Step::Build),
            |s| s.deadline,
        )
    }

    /// Drive `fut` on the runtime, but stop early on two conditions the recipe
    /// bodies otherwise cannot observe (they run synchronously on a blocking
    /// thread): the current step's deadline, and supersession by a newer build.
    /// Either turns into an error that fails the step and unwinds the recipe, so
    /// a wedged command cannot run unbounded and a superseded build stops
    /// mid-step instead of only at the next step boundary.
    fn run_bounded<F, T>(&self, what: &str, fut: F) -> Result<T>
    where
        F: std::future::Future<Output = Result<T>>,
    {
        let deadline = self.step_deadline();
        let cancel = self.cancel.clone();
        self.rt.block_on(async move {
            tokio::pin!(fut);
            let watch = async {
                // Poll the cooperative cancel flag; the finalizer and a
                // superseding build both set it. Cheap next to a build step.
                while !cancel.load(Ordering::SeqCst) {
                    tokio::time::sleep(std::time::Duration::from_millis(250)).await;
                }
            };
            tokio::select! {
                r = &mut fut => r,
                () = tokio::time::sleep_until(deadline.into()) => {
                    Err(anyhow::anyhow!("`{what}` exceeded its per-step deadline"))
                }
                () = watch => {
                    Err(anyhow::anyhow!("build superseded by a newer request; aborting `{what}`"))
                }
            }
        })
    }

    /// Flag the currently-open step as failed (no-op if none is open). Forces
    /// its recorded status to `Failed` at `finish_step` and adds it to the
    /// publish-barring ledger, even though the recipe kept running.
    fn fail_current_step(&self) {
        if let Some(st) = self.current.lock().unwrap().as_mut() {
            st.failed = true;
        }
    }

    /// Finalize the open step (if any): close its log, stamp the DB row, emit
    /// `StepDone`. Idempotent when no step is open. A step flagged via
    /// [`fail_current_step`] is recorded `Failed` regardless of the requested
    /// status, and added to the ledger `publish` consults.
    pub fn finish_step(self: &Arc<Self>, status: Status) -> Result<()> {
        let st = self.current.lock().unwrap().take();
        let Some(st) = st else { return Ok(()) };
        let status = if st.failed { Status::Failed } else { status };
        if status == Status::Failed {
            self.failed_steps.lock().unwrap().push(st.step);
        }
        let me = self.clone();
        self.rt.block_on(async move {
            // Drop all log refs so the sink can be owned + flushed.
            if let Ok(m) = Arc::try_unwrap(st.log) {
                m.into_inner().close().await;
            }
            if let Err(e) = sqlx::query(
                "UPDATE step_runs SET status = ?, finished_at = ? WHERE id = ?",
            )
            .bind(status.as_str())
            .bind(Self::now())
            .bind(st.run_id.0)
            .execute(&me.pool)
            .await
            {
                tracing::error!(step = st.step.as_str(), error = %e, "could not stamp step_run status");
            }
        });
        events::emit(
            &self.events,
            Event::StepDone {
                run_id: st.run_id,
                app: self.app.clone(),
                target: self.target,
                step: st.step,
                status,
            },
        );
        Ok(())
    }

    /// Ensure a step is open; default to `Build` if a recipe runs a command
    /// before declaring one.
    fn ensure_step(self: &Arc<Self>) -> Result<Arc<AsyncMutex<LiveLog>>> {
        if self.current.lock().unwrap().is_none() {
            self.begin_step(Step::Build)?;
        }
        Ok(self.current.lock().unwrap().as_ref().unwrap().log.clone())
    }

    /// The step currently open, or `Build` as a default for failure
    /// attribution before any step was declared.
    pub fn current_step(&self) -> Step {
        self.current
            .lock()
            .unwrap()
            .as_ref()
            .map_or(Step::Build, |s| s.step)
    }

    /// The capability-scoped executor for `name`, or an error if the host isn't
    /// in the topology.
    fn exec(&self, name: &str) -> Result<Arc<dyn ops_exec::Executor>> {
        self.execs
            .get(name)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("unknown build host `{name}` (not in topology)"))
    }

    /// The ssh string for `name` (for `collect`'s remote scp source).
    /// The transport that moves artifacts off `name`. Distinct from
    /// [`RecipeCtx::exec_for`]'s executor: an agent host signs over `AgentRpc`
    /// but is collected from over ssh (`state::build_sync`).
    fn host_sync(&self, name: &str) -> Result<Arc<dyn Executor>> {
        self.syncs
            .get(name)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("unknown build host `{name}` (not in topology)"))
    }

    /// Run `cmd` on `host` through its capability-scoped executor, streaming into
    /// the current step's log. The command's [`Action`] is derived from the open
    /// step (see [`action_for`]) and gated at the transport before dispatch — so a
    /// `build` step on a host without the `build` grant is denied, and the macOS
    /// sign steps ride the in-session `AgentRpc` transport automatically. Returns
    /// exit code + a tail of stdout for the recipe to branch on.
    fn run(self: &Arc<Self>, host: &str, cmd: &str) -> Result<(i32, String)> {
        // The service host is addressed as a service host whatever step is open.
        // Deriving the action from the step is right for a build host, where the
        // step IS the work; on a service host it would ask for `build` during a
        // `verify` and be denied for a reason unrelated to what was attempted.
        let action = match &self.deploy {
            Some(d) if d.host == host => Action::Deploy,
            _ => action_for(self.current_step(), self.kind),
        };
        self.run_as(host, cmd, action)
    }

    /// `run`, with the [`Action`] stated rather than resolved. Used where the
    /// caller already knows which plane it is on.
    fn run_as(self: &Arc<Self>, host: &str, cmd: &str, action: Action) -> Result<(i32, String)> {
        let sink = self.ensure_step()?;
        let exec = self.exec(host)?;
        let cur = self.current_step();
        let step = OpStep::shell(action, cmd.to_string());
        // Echo the command before running it. Without this a log says what
        // happened but not what was asked, and a gate that prints nothing when
        // it passes (`cargo fmt --all --check`) is indistinguishable from a gate
        // that never ran.
        let echo = format!("$ [{host}] {cmd}\n");
        // Bounded by the step's deadline and interruptible on supersession, so a
        // hung command fails its step instead of running unbounded, and a
        // superseded build stops mid-step rather than only at the next boundary.
        let label = format!("{cur} command on `{host}`");
        let out = self.run_bounded(&label, async move {
            use ops_core::remote::LogSink as _;
            let mut guard = sink.lock().await;
            guard.write_chunk(echo.as_bytes()).await;
            exec.run_streaming(&step, &mut *guard).await
        })?;
        let code = out.status.code().unwrap_or(-1);
        let stdout = String::from_utf8_lossy(&out.stdout);
        let tail: String = stdout
            .chars()
            .rev()
            .take(2000)
            .collect::<Vec<_>>()
            .into_iter()
            .rev()
            .collect();
        Ok((code, tail))
    }

    /// Assert `host`'s build tree is at the release tag and return the commit,
    /// which is what a recipe's `checkout` step logs. Fetch + checkout stream
    /// into the current step's log; the sha comes from a separate `rev-parse` so
    /// its stdout is only the sha.
    ///
    /// The preflight has already put this tree at the tag before any recipe ran —
    /// this is the same operation again, on purpose, so that a recipe's own
    /// `checkout` step is a real step with a real log rather than a claim about
    /// something that happened elsewhere. Re-running it is cheap and, because the
    /// tree is Bento's own worktree, forcing: a build that has already dirtied it
    /// must not be able to fail its own retry.
    fn checkout_sha(self: &Arc<Self>, host: &str) -> Result<String> {
        // Every command below runs ON `host`, so the path is that host's, not the
        // daemon's. Windows is why: its worktree is under `C:/Users/me/Code/...`.
        let repo = self.repo_for(host).to_string();
        // A failing mirror is not a failing release: fetch is advisory, and only
        // the checkout decides. Its output still streams into the step log, so an
        // unreachable remote stays visible without being fatal.
        let _ = self.run(host, &git_fetch_cmd(&repo))?;
        let (code, err) = self.run(host, &git_worktree_pin_cmd(&repo, &self.tag))?;
        if code != 0 {
            let (probe, _) = self.run(host, &git_tag_exists_cmd(&repo, &self.tag))?;
            anyhow::bail!(
                "checkout of {} failed on `{host}`: {}",
                self.tag,
                worktree_failure_reason(&self.tag, probe == 0, &err)
            );
        }
        let (code, tail) = self.run(host, &git_rev_parse_cmd(&repo))?;
        anyhow::ensure!(code == 0, "rev-parse failed on `{host}`");
        Ok(tail.trim().to_string())
    }

    /// Every artifact this run collected, `file name -> sha256`.
    ///
    /// Already computed at `collect`, which is the only moment the bytes are
    /// known to be the ones that landed.
    pub fn artifact_hashes(&self) -> HashMap<String, String> {
        self.artifact_hashes.lock().unwrap().clone()
    }

    /// Resolve `glob` on `host` to the single artifact it names. The `for` loop
    /// lists each existing match on its own line (and prints nothing — rather
    /// than a literal unexpanded pattern — when the glob matches no file), so
    /// the count is unambiguous. `required` controls whether zero matches is an
    /// error; more than one always is. See `resolve_artifact_match`.
    fn resolve_artifact(
        self: &Arc<Self>,
        host: &str,
        glob: &str,
        required: bool,
    ) -> Result<String> {
        ensure_glob_safe(glob)?;
        // `[ -e ]` guards against a non-matching glob surviving as its literal
        // self, and lists one path per line for the count.
        let cmd = format!("for __f in {glob}; do [ -e \"$__f\" ] && printf '%s\\n' \"$__f\"; done");
        let (code, tail) = self.run(host, &cmd)?;
        anyhow::ensure!(
            code == 0,
            "resolving artifact glob `{glob}` on `{host}` exited {code}"
        );
        resolve_artifact_match(&tail, glob, required)
    }
}

// ----- error bridging: anyhow -> Rhai runtime error -----

// Rhai host functions return `Result<_, Box<EvalAltResult>>` by convention, so
// this bridge must yield the boxed form to be usable with `.map_err(rhai_err)`.
#[allow(
    clippy::unnecessary_box_returns,
    reason = "rhai's error type is used boxed throughout its host-function API"
)]
fn rhai_err(e: impl std::fmt::Display) -> Box<EvalAltResult> {
    Box::new(EvalAltResult::ErrorRuntime(
        e.to_string().into(),
        rhai::Position::NONE,
    ))
}

/// A crate's publish-relevant metadata, read from `cargo metadata`.
#[derive(Debug, Clone)]
pub struct CrateMeta {
    pub name: String,
    pub version: String,
    pub repository: Option<String>,
    pub description: Option<String>,
    pub licensed: bool,
}

/// Parse the fields that matter for publishing out of `cargo metadata` JSON.
pub fn crate_meta_from_json(raw: &str) -> Result<CrateMeta> {
    let v: serde_json::Value = serde_json::from_str(raw).context("parsing cargo metadata")?;
    let p = v
        .get("packages")
        .and_then(|p| p.as_array())
        .and_then(|a| a.first())
        .context("cargo metadata reported no package")?;
    let str_field = |k: &str| {
        p.get(k)
            .and_then(|x| x.as_str())
            .filter(|s| !s.is_empty())
            .map(str::to_string)
    };
    Ok(CrateMeta {
        name: str_field("name").context("package has no name")?,
        version: str_field("version").context("package has no version")?,
        repository: str_field("repository"),
        description: str_field("description"),
        licensed: str_field("license").is_some() || str_field("license_file").is_some(),
    })
}

/// Everything wrong with a crate's metadata, as messages. Empty means publishable.
///
/// Checks only what crates.io records permanently. A published version cannot
/// be edited, only yanked, and yanking does not correct a wrong URL — so these
/// are the last moment any of it can be fixed.
pub fn crate_publish_problems(
    meta: &CrateMeta,
    repo_clonable: bool,
    published: &[String],
    credentials_present: bool,
) -> Vec<String> {
    let mut out = Vec::new();
    if !credentials_present {
        out.push(
            "no crates.io credentials on the publishing host: `cargo login` there first. \
             Checked now rather than at the upload, so this fails in seconds instead of \
             after a full build and verify."
                .to_string(),
        );
    }
    match &meta.repository {
        None => out.push(
            "no `repository` field: the crates.io page will show no source link, permanently"
                .to_string(),
        ),
        Some(url) if !repo_clonable => out.push(format!(
            "`repository` is not publicly clonable: {url} \
             (wrong URL, or the repo is private)"
        )),
        Some(_) => {}
    }
    if meta.description.is_none() {
        out.push("no `description`: crates.io requires one".to_string());
    }
    if !meta.licensed {
        out.push("no `license` or `license-file`".to_string());
    }
    if published.iter().any(|v| v == &meta.version) {
        out.push(format!(
            "version {} is already published; bump it",
            meta.version
        ));
    }
    out
}

/// Read the app's version from its checkout on the daemon host. With
/// `version_path` set (topology `version_path`), read exactly that file — a
/// `.json` as a Tauri config, anything else as a `Cargo.toml`. Unset (the Tauri
/// default), try `src-tauri/tauri.conf.json` then the root `Cargo.toml`. Used by
/// the runner's default-version path.
pub fn version_from_repo(repo: &str, version_path: Option<&str>) -> Result<Version> {
    let root = expand_tilde(repo);
    if let Some(vp) = version_path {
        let path = root.join(vp);
        let raw = std::fs::read_to_string(&path)
            .with_context(|| format!("reading version file {}", path.display()))?;
        let ver = if std::path::Path::new(vp)
            .extension()
            .is_some_and(|e| e.eq_ignore_ascii_case("json"))
        {
            version_from_tauri_json(&raw)?
        } else {
            version_from_cargo_toml(&raw)?
        };
        return Version::parse(&ver).map_err(|e| anyhow::anyhow!(e));
    }
    let tauri_conf = root.join("src-tauri").join("tauri.conf.json");
    if tauri_conf.exists() {
        let raw = std::fs::read_to_string(&tauri_conf)
            .with_context(|| format!("reading {}", tauri_conf.display()))?;
        return Version::parse(&version_from_tauri_json(&raw)?).map_err(|e| anyhow::anyhow!(e));
    }
    let cargo_toml = root.join("Cargo.toml");
    let raw = std::fs::read_to_string(&cargo_toml).with_context(|| {
        format!(
            "reading {} (no tauri.conf.json either)",
            cargo_toml.display()
        )
    })?;
    Version::parse(&version_from_cargo_toml(&raw)?).map_err(|e| anyhow::anyhow!(e))
}

/// Extract `version` from raw `tauri.conf.json` text.
fn version_from_tauri_json(raw: &str) -> Result<String> {
    let v: serde_json::Value = serde_json::from_str(raw).context("parsing tauri.conf.json")?;
    v.get("version")
        .and_then(|x| x.as_str())
        .map(str::to_owned)
        .context("no `version` in tauri.conf.json")
}

/// Extract the version from raw `Cargo.toml` text — `[package].version` (a leaf
/// crate) or `[workspace.package].version` (a workspace that sets it).
fn version_from_cargo_toml(raw: &str) -> Result<String> {
    let doc: toml::Value = toml::from_str(raw).context("parsing Cargo.toml")?;
    doc.get("package")
        .and_then(|p| p.get("version"))
        .or_else(|| {
            doc.get("workspace")
                .and_then(|w| w.get("package"))
                .and_then(|p| p.get("version"))
        })
        .and_then(|v| v.as_str())
        .map(str::to_owned)
        .context("no `[package].version` or `[workspace.package].version` in Cargo.toml")
}

/// Cross-check every version source in a repo and confirm they all agree with
/// the version being built, before a single host pulls or compiles.
///
/// `version_from_repo` reads exactly one file, so a `tauri.conf.json` at 0.5.0
/// and a root `Cargo.toml` still at 0.4.0 build happily and file artifacts under
/// whichever the runner happened to read. This reads every source present —
/// `version_path` (when set), `src-tauri/tauri.conf.json`, and the root
/// `Cargo.toml` — and fails loudly when any disagree, naming each file and its
/// version. A source that is absent is skipped (a library crate with only a
/// `Cargo.toml` has nothing to disagree with); the check never invents drift.
///
/// Scope: the JSON/TOML sources bentod itself reads. The iOS `gen/apple/project.yml`
/// path (rewritten by a build-time `sed`) is out of scope here — it is asserted at
/// its own build step — but the same drift class motivated this guard.
pub fn check_version_consistency(
    repo: &str,
    version_path: Option<&str>,
    expected: &Version,
) -> Result<()> {
    let root = expand_tilde(repo);
    let mut sources: Vec<(String, String)> = Vec::new();
    for rel in version_sources(version_path) {
        let path = root.join(&rel);
        // A source that is absent is skipped — a library crate with only a
        // `Cargo.toml` has nothing to disagree with — but one the app NAMES
        // must be readable, or the check would pass by failing to look.
        match std::fs::read_to_string(&path) {
            Ok(raw) => sources.push((rel, raw)),
            Err(e) if version_path == Some(rel.as_str()) => {
                return Err(e).with_context(|| format!("reading version file {}", path.display()));
            }
            Err(_) => {}
        }
    }
    versions_agree(repo, &sources, version_path, expected)
}

/// The files a repo can state its version in, in the order they are read:
/// whatever the app names, then the two conventional ones.
///
/// The app's own `version_path` is never read twice, which is why this is a
/// function rather than a constant.
pub fn version_sources(version_path: Option<&str>) -> Vec<String> {
    let mut rels: Vec<String> = version_path.into_iter().map(str::to_string).collect();
    for conventional in ["src-tauri/tauri.conf.json", "Cargo.toml"] {
        if version_path != Some(conventional) {
            rels.push(conventional.to_string());
        }
    }
    rels
}

/// The judgement half of [`check_version_consistency`], over sources somebody
/// else read.
///
/// Split out so the same rule can be applied to files read out of the release
/// TAG on a build host, which is where the question actually belongs: the tree
/// a release compiles is the tag's, so a `Cargo.toml` that disagrees with the
/// tag it is tagged in is the drift worth refusing. Reading the working copy
/// instead answered a question about a tree the release does not build.
///
/// `where_` is only for the error message — a path, or a tag and a host.
pub fn versions_agree(
    where_: &str,
    sources: &[(String, String)],
    version_path: Option<&str>,
    expected: &Version,
) -> Result<()> {
    let mut found: Vec<(String, Version)> = Vec::new();
    for (rel, raw) in sources {
        // The app's own `version_path` can be either shape, so it is decided by
        // extension; the two conventional sources are what they are.
        let as_json = std::path::Path::new(rel)
            .extension()
            .is_some_and(|e| e.eq_ignore_ascii_case("json"));
        let ver = if as_json {
            version_from_tauri_json(raw)
        } else {
            // A Cargo.toml with neither `[package].version` nor
            // `[workspace.package].version` (a pure virtual workspace) carries
            // no version to check — skip it rather than fail. An app that NAMED
            // this file is held to it.
            match version_from_cargo_toml(raw) {
                Ok(v) => Ok(v),
                Err(e) if version_path == Some(rel.as_str()) => Err(e),
                Err(_) => continue,
            }
        }?;
        found.push((
            rel.clone(),
            Version::parse(&ver).map_err(|e| anyhow::anyhow!(e))?,
        ));
    }

    let disagree: Vec<&(String, Version)> = found.iter().filter(|(_, v)| v != expected).collect();
    anyhow::ensure!(
        disagree.is_empty(),
        "version drift in {where_}: building {expected} but {}",
        disagree
            .iter()
            .map(|(src, v)| format!("{src} says {v}"))
            .collect::<Vec<_>>()
            .join(", ")
    );
    Ok(())
}

/// Read one file as it exists in `tag`, without checking anything out.
///
/// `<rev>:./<path>` resolves the path relative to `-C`, so this is asked from
/// the app's own directory and needs no knowledge of where that sits inside the
/// repository. A non-zero exit means the file is not in the tag, which is the
/// same "absent, so nothing to disagree with" the local read treats it as.
pub fn git_show_file_cmd(dir: &str, tag: &str, rel: &str) -> String {
    format!("git -C \"{dir}\" show \"{tag}:./{rel}\"")
}

/// Every `X.Y.Z`-shaped version embedded in an artifact file name. Each maximal
/// run of digits-and-dots contributes its first three numeric fields:
/// `GoingsOn_0.5.0_aarch64.dmg` and `demo-9.9.9.bin` both yield one version (the
/// trailing `.bin`/`.dmg` dot is tolerated), while `latest.json` yields `[]` and
/// the `64` in `x86_64` is not three fields. Only the `major.minor.patch` core is
/// taken; a prerelease/build suffix is separated by `-`/`+` and not needed here.
fn versions_in_filename(name: &str) -> Vec<Version> {
    name.split(|c: char| !(c.is_ascii_digit() || c == '.'))
        .filter_map(|run| {
            let f: Vec<&str> = run.split('.').filter(|s| !s.is_empty()).collect();
            if f.len() >= 3 && f[..3].iter().all(|s| s.chars().all(|c| c.is_ascii_digit())) {
                Version::parse(&format!("{}.{}.{}", f[0], f[1], f[2])).ok()
            } else {
                None
            }
        })
        .collect()
}

/// Fail when a collected file's name embeds a version whose `major.minor.patch`
/// is not the one being built. This is the guard against a stale checked-in
/// artifact winning a glob: `ls -t <glob>` once let
/// `AudioFiles-0.4.0-x86_64.AppImage` ship against 0.5.0. A file whose name
/// carries no version (an updater `latest.json`, a `.sig`) is not asserted —
/// there is nothing to compare. Compared on the core so a prerelease build's
/// plain `X.Y.Z` in the filename still matches.
fn assert_artifact_version(name: &str, expected: &Version) -> Result<()> {
    let versions = versions_in_filename(name);
    anyhow::ensure!(
        versions.is_empty() || versions.iter().any(|v| v.core() == expected.core()),
        "collected artifact `{name}` carries version {} but the build is {expected}; \
         a stale artifact was left in the output dir — clean it so only {expected} remains",
        versions
            .iter()
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join("/"),
    );
    Ok(())
}

/// sha256 of a file, lowercase hex. Streams in 64 KiB chunks so a multi-GiB
/// bundle never lands in memory whole.
fn sha256_file(path: &Path) -> Result<String> {
    let mut file =
        std::fs::File::open(path).with_context(|| format!("hashing {}", path.display()))?;
    let mut hasher = Sha256::new();
    std::io::copy(&mut file, &mut hasher)
        .with_context(|| format!("reading {} to hash", path.display()))?;
    Ok(hex_lower(&hasher.finalize()))
}

/// Every regular file under `root`, as `(path relative to root, absolute path)`,
/// sorted by the relative path.
///
/// **This walk has to match `bundle::digest_dir` in sando, file for file.** That
/// function re-hashes an incoming bundle and refuses it when the bytes disagree
/// with the manifest they arrived with, so a producer that walks differently
/// produces a manifest the consumer will reject for an artifact nothing is wrong
/// with. Three properties carry that agreement, and none is incidental:
///
/// - **Recursive.** A bundle may carry a directory (migrations, resources), and
///   a top-level-only listing would omit its contents from the manifest while
///   the verifier hashed them.
/// - **Symlinks are not followed, and not recorded.** Following one would let
///   content from outside the bundle into its identity; recording the link
///   itself would name a file the verifier does not hash.
/// - **Relative paths, `/`-separated, sorted.** Readdir order is not guaranteed,
///   so an unsorted manifest would differ run to run on one machine, never mind
///   between two.
fn collected_files(root: &Path) -> std::io::Result<Vec<(String, PathBuf)>> {
    fn walk(dir: &Path, root: &Path, out: &mut Vec<(String, PathBuf)>) -> std::io::Result<()> {
        for entry in std::fs::read_dir(dir)? {
            let entry = entry?;
            let ft = entry.file_type()?;
            let path = entry.path();
            if ft.is_dir() {
                walk(&path, root, out)?;
            } else if ft.is_file() {
                let rel = path
                    .strip_prefix(root)
                    .unwrap_or(&path)
                    .components()
                    .map(|c| c.as_os_str().to_string_lossy())
                    .collect::<Vec<_>>()
                    .join("/");
                out.push((rel, path));
            }
            // Symlinks and other special files are intentionally ignored,
            // matching the verifier.
        }
        Ok(())
    }
    let mut out = Vec::new();
    walk(root, root, &mut out)?;
    out.sort_by(|a, b| a.0.cmp(&b.0));
    Ok(out)
}

/// Lowercase-hex encode without pulling in a hex crate.
fn hex_lower(bytes: &[u8]) -> String {
    use std::fmt::Write as _;
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        let _ = write!(s, "{b:02x}");
    }
    s
}

/// Refresh every remote's refs and tags, so the tag a release names is present
/// locally however it was pushed. No branch/upstream assumptions — a bare
/// `git pull --ff-only` needs a tracking branch the release path shouldn't
/// depend on.
///
/// Best-effort on purpose. `fetch --all` exits non-zero if ANY remote fails, and
/// the library repos carry three (`astra`, `mnw`, `srht`), so chaining this into
/// the checkout with `&&` meant one unreachable mirror aborted the release and
/// reported it as a missing tag. The checkout below is the step allowed to fail;
/// this one only has to try. See [`git_worktree_pin_cmd`].
///
/// `repo` is interpolated UNQUOTED so a leading `~` is expanded by the remote
/// host's shell (the checkout path is trusted topology config, not user input),
/// matching how the recipes `cd` into it.
pub fn git_fetch_cmd(repo: &str) -> String {
    format!("git -C {repo} fetch --all --tags --prune")
}

/// Probe for the one failure a tracked `Cargo.lock` hits inside Bento's
/// worktree. Exits 0 when the crate tracks a lock AND the checkout sits under a
/// `.cargo/config.toml` declaring `[patch]`; 1 otherwise.
///
/// Both halves are needed and cargo reports neither. Under a `[patch]` block
/// cargo re-resolves and rewrites the lock's `[[patch.unused]]` entries, so
/// `cargo publish --dry-run` refuses the tree with "1 files in the working
/// directory contain changes that were not yet committed into git: Cargo.lock"
/// -- naming the lock and nothing about why it moved. pter 0.2.1 lost half an
/// hour to that message on build 325.
///
/// `~/Code/.bento` is under `~/Code` deliberately, so that the patch block
/// reaches the build (see [`crate::topology::Host::worktree_root`]). The patch
/// block is therefore not the half to remove, which is why this is worth
/// saying rather than leaving cargo to be cryptic about it.
///
/// `repo` is interpolated unquoted for the `~`, matching [`git_fetch_cmd`];
/// `pwd -P` then hands the loop an absolute path to walk up from.
pub fn tracked_lock_under_patch_cmd(repo: &str) -> String {
    format!(
        "git -C {repo} ls-files --error-unmatch Cargo.lock >/dev/null 2>&1 || exit 1; \
         d=$(cd {repo} && pwd -P) || exit 1; \
         while [ -n \"$d\" ] && [ \"$d\" != / ]; do \
         for c in \"$d/.cargo/config.toml\" \"$d/.cargo/config\"; do \
         [ -f \"$c\" ] && grep -q '^\\[patch' \"$c\" && exit 0; \
         done; d=$(dirname \"$d\"); done; exit 1"
    )
}

/// What to say when [`tracked_lock_under_patch_cmd`] answers yes. Names both
/// facts, because the error cargo would otherwise print names neither, and
/// closes off the two wrong fixes that are both one flag away.
pub fn tracked_lock_under_patch_problem(repo: &str) -> String {
    format!(
        "`Cargo.lock` is tracked and {repo} sits under a `.cargo/config.toml` \
         declaring `[patch]`. Cargo re-resolves there and rewrites the lock, so \
         `cargo publish --dry-run` will refuse the tree as dirty and name only \
         the lock. Untrack it: `git rm --cached Cargo.lock`. Every other library \
         in this tree already does. Not `--allow-dirty`, which publishes a lock \
         nobody reviewed, and not committing the rewritten lock, which resolves \
         differently on the next machine and fails there instead. The build \
         worktree is under `~/Code` on purpose so the patch block applies to it; \
         that is not the half to change."
    )
}

/// Does `tag` resolve to a commit in this checkout? Run only when the checkout
/// has already failed, to say WHY: an absent tag is an untagged or unpushed
/// release, while a tag that resolves fine means the checkout was refused for a
/// local reason (a dirty tree, most often) and the operator needs to hear that
/// instead.
pub fn git_tag_exists_cmd(repo: &str, tag: &str) -> String {
    format!("git -C {repo} rev-parse -q --verify \"refs/tags/{tag}^{{commit}}\"")
}

/// Which repository `repo` belongs to, and where `repo` sits inside it, in one
/// call: `--show-toplevel` then `--show-prefix`, one per line.
///
/// Both halves are needed to build in a worktree of a repo holding several
/// products. The worktree is made of the repository (`~/Code/MNW`), and the
/// recipe has to be pointed at the app inside it (`<worktree>/pom`).
pub fn git_toplevel_and_prefix_cmd(repo: &str) -> String {
    format!("git -C {repo} rev-parse --show-toplevel --show-prefix")
}

/// Read [`git_toplevel_and_prefix_cmd`]'s two lines.
///
/// The prefix is empty for a repo holding one product, where `repo` IS the
/// repository root — and git prints an empty second line for it, so a missing
/// line is a malformed answer rather than that case.
pub fn parse_toplevel_and_prefix(out: &str) -> Option<(String, String)> {
    let mut lines = out.split('\n');
    let toplevel = lines.next()?.trim().to_string();
    let prefix = lines.next()?.trim().to_string();
    (!toplevel.is_empty()).then_some((toplevel, prefix))
}

/// The repository's own directory name, which is what names its worktrees:
/// `MNW` for `/home/max/Code/MNW`.
///
/// Splits on `/` only. Git reports `--show-toplevel` with forward slashes on
/// every platform, Windows included, so this is the separator to read.
pub fn repo_dir_name(toplevel: &str) -> &str {
    toplevel
        .trim_end_matches('/')
        .rsplit('/')
        .next()
        .unwrap_or(toplevel)
}

/// Where the app being released sits inside its worktree: the worktree root for
/// a repo holding one product, `<worktree>/pom` for one holding several.
pub fn app_dir_in_worktree(worktree: &str, prefix: &str) -> String {
    let prefix = prefix.trim_matches('/');
    if prefix.is_empty() {
        worktree.to_string()
    } else {
        format!("{}/{prefix}", worktree.trim_end_matches('/'))
    }
}

/// Does this worktree already exist? Run before deciding whether to create one.
///
/// `rev-parse --git-dir` rather than a shell test, because the one non-unix
/// build host has no `test`: every command Bento renders for a host is a git
/// command or something a recipe wrote.
pub fn git_worktree_probe_cmd(worktree: &str) -> String {
    format!("git -C \"{worktree}\" rev-parse --git-dir")
}

/// Forget worktrees whose directories are gone. Run before creating one: a
/// directory somebody deleted by hand is still registered in the repository, and
/// `worktree add` refuses the path as in use rather than rebuilding it.
pub fn git_worktree_prune_cmd(toplevel: &str) -> String {
    format!("git -C \"{toplevel}\" worktree prune")
}

/// Create this app's build worktree, detached at the release tag. Git creates
/// the leading directories, so the worktree root needs no preparation.
pub fn git_worktree_add_cmd(toplevel: &str, worktree: &str, tag: &str) -> String {
    format!("git -C \"{toplevel}\" worktree add --detach --force \"{worktree}\" \"{tag}\"")
}

/// Put an existing build worktree at the release tag.
///
/// `--force` discards whatever the last release left in it — a rewritten
/// `Cargo.lock`, most often — and that is safe here in a way it never was in the
/// ordinary checkout: nothing but Bento writes in this tree, so there is no edit
/// of anybody's to lose. Owning the tree is what buys the forcing.
pub fn git_worktree_pin_cmd(worktree: &str, tag: &str) -> String {
    format!("git -C \"{worktree}\" checkout --detach --force \"{tag}\"")
}

/// The operator-facing explanation for a worktree that could not be put at the
/// tag.
///
/// An absent tag is an untagged or unpushed release and is the common case, so
/// it is answered plainly. Anything else is git's own stderr, which says more
/// about a path that is not a worktree, or a worktree another release holds,
/// than a guess would.
pub fn worktree_failure_reason(tag: &str, tag_exists: bool, stderr: &str) -> String {
    if !tag_exists {
        return format!("tag {tag} does not exist there (is it created and pushed?)");
    }
    let stderr = stderr.trim();
    if stderr.is_empty() {
        format!("tag {tag} exists, and git said nothing about why")
    } else {
        stderr.to_string()
    }
}

/// The command a host runs to report the commit it has checked out, for the
/// release preflight barrier.
pub fn git_rev_parse_cmd(repo: &str) -> String {
    format!("git -C {repo} rev-parse HEAD")
}

/// Expand a leading `~/` to `$HOME`. Paths in the topology are written with `~`.
pub fn expand_tilde(p: &str) -> PathBuf {
    if let Some(rest) = p.strip_prefix("~/")
        && let Ok(home) = std::env::var("HOME")
    {
        return Path::new(&home).join(rest);
    }
    PathBuf::from(p)
}

/// Reject a glob that carries shell command metacharacters. Path and wildcard
/// characters (`/ . * ? [ ] ~` etc.) are fine — the pattern reaches a login
/// shell to be expanded — but a `;` or `$(...)` must not ride along and run.
/// Not a privilege boundary (a recipe already runs arbitrary shell via `sh_ok`)
/// but it keeps a malformed pattern from turning into a command. Shared by
/// `collect` and `resolve_artifact`.
fn ensure_glob_safe(glob: &str) -> Result<()> {
    anyhow::ensure!(
        !glob.chars().any(|c| matches!(
            c,
            ';' | '&' | '|' | '$' | '`' | '\'' | '"' | '\\' | ' ' | '\n' | '(' | ')' | '<' | '>'
        )),
        "glob `{glob}` contains shell metacharacters"
    );
    Ok(())
}

/// Decide the single artifact a glob resolves to from a newline-separated
/// listing of the paths that matched it.
///
/// Demands exactly one match. Zero matches fail when `required` (return `""`
/// when optional); more than one is always an error rather than an arbitrary
/// newest-wins pick, because an ambiguous match means the build left stale
/// artifacts behind and the wrong one could ship.
fn resolve_artifact_match(listing: &str, glob: &str, required: bool) -> Result<String> {
    let matches: Vec<&str> = listing
        .lines()
        .map(str::trim)
        .filter(|l| !l.is_empty())
        .collect();
    match matches.as_slice() {
        [] if required => anyhow::bail!("no artifact matched glob `{glob}`"),
        [] => Ok(String::new()),
        [one] => Ok((*one).to_string()),
        many => anyhow::bail!(
            "glob `{glob}` is ambiguous: {} artifacts matched ({}). \
             The build left more than one behind; clean stale artifacts so exactly one remains.",
            many.len(),
            many.join(", ")
        ),
    }
}

/// Build a Rhai engine with the host API bound to `ctx`. Sandboxed: recipes
/// touch the outside world only through these functions.
pub fn build_engine(ctx: &Arc<RecipeCtx>) -> Engine {
    let mut engine = Engine::new();
    // Defensive caps — recipes are first-party but bound the blast radius.
    engine.set_max_operations(5_000_000);
    engine.set_max_call_levels(64);
    engine.set_max_string_size(0);

    // --- step(name) ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "step",
            move |name: &str| -> Result<(), Box<EvalAltResult>> {
                let step: Step = name.parse().map_err(rhai_err)?;
                ctx.begin_step(step).map_err(rhai_err)
            },
        );
    }

    // --- sh(host, cmd) -> #{ code, stdout_tail } ---
    //
    // The branch-on-exit-code primitive: the recipe OWNS the outcome. A non-zero
    // exit is returned, not raised, and does NOT fail the step or bar publish —
    // use this only when the recipe inspects `code` and decides. For a command
    // that must succeed (build/sign/etc.), use `sh_ok`, which fails the step (and
    // therefore bars publish via the failed-step ledger) on a non-zero exit.
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "sh",
            move |host: &str, cmd: &str| -> Result<Map, Box<EvalAltResult>> {
                let (code, tail) = ctx.run(host, cmd).map_err(rhai_err)?;
                let mut m = Map::new();
                m.insert("code".into(), (code as i64).into());
                m.insert("stdout_tail".into(), tail.into());
                Ok(m)
            },
        );
    }

    // --- sh_ok(host, cmd): run + assert exit 0 (the must-succeed primitive) ---
    //
    // A non-zero exit fails the current step (added to the publish-barring
    // ledger) and aborts the recipe, so an artifact is never shipped after a
    // must-succeed command failed.
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "sh_ok",
            move |host: &str, cmd: &str| -> Result<(), Box<EvalAltResult>> {
                let (code, _) = ctx.run(host, cmd).map_err(rhai_err)?;
                if code != 0 {
                    // Attribute the failure to the current step explicitly so the
                    // ledger bars publish even if a future caller swallowed the error.
                    ctx.fail_current_step();
                    return Err(rhai_err(format!(
                        "command on `{host}` exited {code}: {cmd}"
                    )));
                }
                Ok(())
            },
        );
    }

    // --- resolve_artifact(host, glob) -> path: the ONE artifact matching glob ---
    //
    // The artifact-selection primitive. Replaces `sh(host, "ls -t <glob> | head
    // -1").stdout_tail.trim()` guarded on an empty string, which let a non-zero
    // `ls` pass quietly and a stale newest-by-mtime file win. This resolves the
    // glob on the host and demands exactly one match: zero matches or more than
    // one both throw (an ambiguous match means the build left stale artifacts,
    // and silently picking the newest is how the wrong bytes ship). Use
    // `resolve_artifact_opt` for an artifact that may legitimately be absent.
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "resolve_artifact",
            move |host: &str, glob: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.resolve_artifact(host, glob, true).map_err(rhai_err)
            },
        );
    }

    // --- resolve_artifact_opt(host, glob) -> path | "": zero-or-one match ---
    //
    // Same strict resolution as `resolve_artifact` but tolerates zero matches
    // (returns ""); more than one is still an error. For optional outputs like a
    // `.deb` or an updater bundle a recipe collects only when present.
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "resolve_artifact_opt",
            move |host: &str, glob: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.resolve_artifact(host, glob, false).map_err(rhai_err)
            },
        );
    }

    // --- log(msg): operator-visible line into the current step's tail ---
    {
        let ctx = ctx.clone();
        engine.register_fn("log", move |msg: &str| -> Result<(), Box<EvalAltResult>> {
            let sink = ctx.ensure_step().map_err(rhai_err)?;
            let line = format!("[recipe] {msg}\n");
            ctx.rt.block_on(async {
                use ops_core::remote::LogSink;
                sink.lock().await.write_chunk(line.as_bytes()).await;
            });
            Ok(())
        });
    }

    // --- version_of(app) -> string ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "version_of",
            move |app: &str| -> Result<String, Box<EvalAltResult>> {
                // Only the current app is in scope; cross-app reads aren't needed.
                if app != ctx.app.as_str() {
                    return Err(rhai_err(format!(
                        "version_of: `{app}` is not the app being built"
                    )));
                }
                Ok(ctx.version.to_string())
            },
        );
    }

    // --- version() -> string: the version being built (no-arg form) ---
    {
        let ctx = ctx.clone();
        engine.register_fn("version", move || -> String { ctx.version.to_string() });
    }

    // --- build_host() -> string: the host this target builds on ---
    {
        let ctx = ctx.clone();
        engine.register_fn("build_host", move || -> String { ctx.build_host.clone() });
    }

    // --- repo() -> string: the app's checkout path on this target's build host
    //     (`~`-prefixed on a unix host). Host-correct rather than one path per
    //     app, so a recipe for a host whose checkout is elsewhere still calls
    //     this instead of hard-coding the path — which is what kept the Windows
    //     recipes off `checkout_sha`. ---
    {
        let ctx = ctx.clone();
        engine.register_fn("repo", move || -> String {
            ctx.repo_for(&ctx.build_host).to_string()
        });
    }

    // --- checkout_sha(host) -> sha: pin this host to the release tag and report
    //     its commit. Replaces a recipe's `git pull --ff-only`, which builds
    //     whatever `main` is at pull time; the daemon also runs the same pin as
    //     a cross-host preflight barrier before any target builds. ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "checkout_sha",
            move |host: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.checkout_sha(host).map_err(rhai_err)
            },
        );
    }

    // --- crate_preflight() -> string: verify this crate is safe to publish,
    //     or abort the run. Everything it checks is immutable once published:
    //     crates.io versions can be yanked but never edited, so a wrong
    //     repository URL is permanent. pter 0.1.0 shipped with a dead one. ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "crate_preflight",
            move || -> Result<String, Box<EvalAltResult>> {
                // `repo`, not `repo_for(...)`: `cargo metadata` runs on the
                // daemon's own box, so this is the one checkout that is always
                // the local one. It is not a missed call site.
                let repo = expand_tilde(&ctx.repo);

                let out = std::process::Command::new("cargo")
                    .args(["metadata", "--no-deps", "--format-version", "1"])
                    .current_dir(&repo)
                    .output()
                    .map_err(|e| format!("running cargo metadata in {}: {e}", repo.display()))?;
                if !out.status.success() {
                    return Err(format!(
                        "cargo metadata failed in {}: {}",
                        repo.display(),
                        String::from_utf8_lossy(&out.stderr).trim()
                    )
                    .into());
                }
                let meta = crate_meta_from_json(&String::from_utf8_lossy(&out.stdout))
                    .map_err(|e| e.to_string())?;

                // The real question is not whether a page renders but whether a
                // stranger with no credentials can fetch the source, so ask git.
                let clonable = meta.repository.as_ref().is_some_and(|url| {
                    std::process::Command::new("git")
                        .args(["ls-remote", url])
                        .env("GIT_TERMINAL_PROMPT", "0")
                        .output()
                        .is_ok_and(|o| o.status.success())
                });

                // Ask the publishing host whether cargo has credentials, rather
                // than moving the token anywhere. It stays in cargo's own 0600
                // store; a shell line carrying it would be visible in `ps`.
                // An exit code answers "are there credentials"; an Err answers
                // "the question could not be asked". Collapsing the second into
                // the first reported a capability denial as "no crates.io
                // credentials", which sent a real diagnosis three rounds the
                // wrong way. A check that cannot run is not a failed check.
                let creds =
                    ctx.run(
                        &ctx.build_host.clone(),
                        "cargo login --help >/dev/null 2>&1 && \
                     test -s \"${CARGO_HOME:-$HOME/.cargo}/credentials.toml\" \
                     || test -s \"${CARGO_HOME:-$HOME/.cargo}/credentials\"",
                    )
                    .map_err(|e| {
                        format!(
                            "could not check crates.io credentials on `{}`: {e}",
                            ctx.build_host
                        )
                    })?
                    .0 == 0;

                // Asked of the build host rather than the daemon: the tree
                // that gets published is the worktree over there, and it is the
                // one whose `[patch]` ancestry decides this. An Err is "the
                // question could not be asked" and is not a finding -- same
                // rule as the credentials check above, for the same reason it
                // was written that way.
                let build_repo = ctx.repo_for(&ctx.build_host).to_string();
                let patched_lock =
                    ctx.run(
                        &ctx.build_host.clone(),
                        &tracked_lock_under_patch_cmd(&build_repo),
                    )
                    .map_err(|e| {
                        format!(
                            "could not check for a tracked Cargo.lock on `{}`: {e}",
                            ctx.build_host
                        )
                    })?
                    .0 == 0;

                let published = RecipeCtx::published_versions(&meta.name);
                let mut problems = crate_publish_problems(&meta, clonable, &published, creds);
                if patched_lock {
                    problems.push(tracked_lock_under_patch_problem(&build_repo));
                }
                if !problems.is_empty() {
                    return Err(format!(
                        "{} {} is not safe to publish:\n  - {}",
                        meta.name,
                        meta.version,
                        problems.join("\n  - ")
                    )
                    .into());
                }
                Ok(format!("{} {} passed preflight", meta.name, meta.version))
            },
        );
    }

    // --- feature_flags() -> string: `--features a,b`, or "" when the app
    //     declares none. Returns the whole flag rather than a bare list so an
    //     app with no features cannot produce a dangling `--features`. ---
    {
        let ctx = ctx.clone();
        engine.register_fn("feature_flags", move || -> String {
            if ctx.features.is_empty() {
                String::new()
            } else {
                format!("--features {}", ctx.features.join(","))
            }
        });
    }

    // --- target() / platform() / arch(): the target axis, for one per-platform
    //     recipe to branch on arch (bundle paths differ between x86_64/aarch64). ---
    {
        let ctx = ctx.clone();
        engine.register_fn("target", move || -> String { ctx.target.to_string() });
    }
    {
        let ctx = ctx.clone();
        engine.register_fn("platform", move || -> String {
            ctx.target.platform.as_str().to_string()
        });
    }
    {
        let ctx = ctx.clone();
        engine.register_fn("arch", move || -> String {
            ctx.target.arch.as_str().to_string()
        });
    }

    // --- secret(key) -> string (file under secrets_root; never logged) ---
    {
        let ctx = ctx.clone();
        engine.register_fn("secret", move |key: &str| -> Result<String, Box<EvalAltResult>> {
            // Guard against traversal out of secrets_root. Require every path
            // component to be `Normal` (rejects `..`, `.`, absolute roots and
            // drive prefixes) and forbid backslashes (a literal filename char on
            // Linux, but a separator elsewhere) — the per-component strength of
            // Sando's `safe()`. A multi-segment key like `app/token` is still
            // allowed; `foo..bar` (a legit filename) is no longer falsely blocked.
            let safe = !key.is_empty()
                && !key.contains('\\')
                && std::path::Path::new(key)
                    .components()
                    .all(|c| matches!(c, std::path::Component::Normal(_)));
            if !safe {
                return Err(rhai_err(
                    "secret key must be a relative path under secrets_root (no `..`, `.`, absolute paths, or backslashes)",
                ));
            }
            let path = ctx.cfg.secrets_root.join(key);
            std::fs::read_to_string(&path)
                .map(|s| s.trim_end().to_string())
                .map_err(|e| rhai_err(format!("secret `{key}`: {e}")))
        });
    }

    // --- env(host, key) -> string ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "env",
            move |host: &str, key: &str| -> Result<String, Box<EvalAltResult>> {
                // The key is interpolated into a `${...}` shell expansion, so it must
                // be a bare shell identifier — anything else (quotes, `}`, `$`, `;`)
                // could break out and run arbitrary commands on the host. Validate
                // before building the command; this is the one env read that can't
                // sh-quote its argument (a quoted var name doesn't expand).
                if key.is_empty()
                    || !key
                        .chars()
                        .next()
                        .is_some_and(|c| c == '_' || c.is_ascii_alphabetic())
                    || !key.chars().all(|c| c == '_' || c.is_ascii_alphanumeric())
                {
                    return Err(rhai_err(format!(
                        "env name `{key}` must be a shell identifier ([A-Za-z_][A-Za-z0-9_]*)"
                    )));
                }
                // Read via the shell so it works on remote hosts too.
                let (code, tail) = ctx
                    .run(host, &format!("printf '%s' \"${{{key}}}\""))
                    .map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err(format!("env `{key}` on `{host}` failed")));
                }
                Ok(tail.trim().to_string())
            },
        );
    }

    // --- collect(host, glob, app, version): pull artifacts to dist_root ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "collect",
            move |host: &str,
                  glob: &str,
                  app: &str,
                  version: &str|
                  -> Result<(), Box<EvalAltResult>> {
                ctx.collect(host, glob, app, version).map_err(rhai_err)
            },
        );
    }

    // --- publish(channel, app, target, version, artifact, meta) ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "publish",
            move |channel: &str,
                  app: &str,
                  target: &str,
                  version: &str,
                  artifact: &str,
                  meta: Map|
                  -> Result<String, Box<EvalAltResult>> {
                ctx.publish(channel, app, target, version, artifact, &meta)
                    .map_err(rhai_err)
            },
        );
    }

    // --- deploy(binary) -> summary: install a service binary and restart its
    //     unit. The terminal step for `kind = "service"`, the counterpart of
    //     `publish` for something that is run rather than distributed.
    //
    //     Takes only the binary's path on the build host: where it lands, on
    //     which machine, and which unit restarts all come from the `[[deploy]]`
    //     entry for the target already being built. A recipe cannot deploy the
    //     aarch64 binary to the x86_64 box by naming the wrong host, because it
    //     never names a host at all.
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "deploy",
            move |binary: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.deploy(binary).map_err(rhai_err)
            },
        );
    }

    // --- deploy_host() -> string: the service host's ssh destination, so a
    //     recipe can run its own assertions there (`sh_ok(deploy_host(), ...)`).
    //     Commands run through it while the `deploy` step is open, so they are
    //     gated on the deploy grant like the install itself. ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "deploy_host",
            move || -> Result<String, Box<EvalAltResult>> {
                ctx.deploy_target()
                    .map(|d| d.host.clone())
                    .map_err(rhai_err)
            },
        );
    }

    // --- service_name() / install_path() / health_url(): the rest of the
    //     `[[deploy]]` entry, so a recipe asserts against the configured values
    //     rather than repeating them as literals that can drift. `health_url`
    //     is "" when unset. ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "service_name",
            move || -> Result<String, Box<EvalAltResult>> {
                ctx.deploy_target()
                    .map(|d| d.service.clone())
                    .map_err(rhai_err)
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "install_path",
            move || -> Result<String, Box<EvalAltResult>> {
                ctx.deploy_target()
                    .map(|d| d.install_path.clone())
                    .map_err(rhai_err)
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "health_url",
            move || -> Result<String, Box<EvalAltResult>> {
                ctx.deploy_target()
                    .map(|d| d.health_url.clone().unwrap_or_default())
                    .map_err(rhai_err)
            },
        );
    }

    // --- glibc_check(binary) -> string: assert the build host did not produce
    //     a binary the service host's glibc is too old to exec. Aborts the run
    //     if it did; returns "needs X, host has Y" for the log if it did not.
    //
    //     WHICH RECIPES CALL THIS, AND WHY THE OTHERS MUST NOT. The rule is not
    //     a style preference and it is not optional: this reads the recipe's
    //     `[[deploy]]` entry to learn which machine runs the bytes, so a recipe
    //     with no `[[deploy]]` cannot call it at all.
    //
    //     - A service that installs ITSELF (`[[deploy]]` present: magicmirror,
    //       wam, mnw-cli) SHOULD call it. Bento is both builder and installer
    //       there, so it knows the service host, and nothing downstream will
    //       check on its behalf.
    //     - A service HANDED OFF to Sando (`[[deploy]]` absent: pom) MUST NOT,
    //       and the absence is the Sando/Bento boundary rather than an omission.
    //       Which machine runs the bytes is environment knowledge, which is
    //       Sando's half. Sando covers it on the far side, more strongly: it runs
    //       the node's own loader against the rsynced bytes before the symlink
    //       swap (`sando_daemon::deploy::ldd_guard_script`), and since 0.2.12
    //       also compares the bundle's glibc floor against the node's declared
    //       `libc` before the rsync (`check_bundle_fits_node`).
    //
    //     So a new service recipe takes its answer from whether it carries a
    //     `[[deploy]]` table, not from whichever sibling recipe it was copied
    //     from. Wiki `sando-bento-boundary`, `host-base-images`. ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "glibc_check",
            move |binary: &str| -> Result<String, Box<EvalAltResult>> {
                let (needs, has) = ctx.glibc_check(binary).map_err(rhai_err)?;
                Ok(format!(
                    "glibc: binary needs {needs}, service host has {has}"
                ))
            },
        );
    }

    // --- macOS signing helpers. They dispatch through the named host's
    //     executor like any other step; when that host is the mac (transport =
    //     "agent"), codesign/notarize/staple ride the in-session `AgentRpc`
    //     transport — the only security session where the Developer ID key is
    //     usable (design §7 "THE WALL"). Capability-gated by the host's `sign`
    //     grant. ---
    register_macos_fns(&mut engine, ctx);

    engine
}

/// Highest `GLIBC_x.y` version referenced by a built binary, and the glibc a
/// host actually has, both parsed from the text the commands print.
///
/// Native-per-architecture builds remove the cross-compile hazard `deploy.sh`
/// was written against, but not this one: fw13 tracks a newer glibc than the
/// Ubuntu 24.04 box in Hetzner, so a binary built here can reference a symbol
/// version that box does not have and fail at exec — after the unit has already
/// been restarted onto it. Comparing the two before the install is what makes
/// that a failed step instead of a downed service.
fn max_glibc_symbol(objdump_out: &str) -> Option<(u64, u64)> {
    objdump_out
        .split(|c: char| !(c.is_ascii_digit() || c == '.' || c == '_' || c.is_ascii_alphabetic()))
        .filter_map(|tok| tok.strip_prefix("GLIBC_"))
        .filter_map(parse_glibc_version)
        .max()
}

/// Parse `2.39` (or `2.39.1`, keeping major/minor) into a comparable pair.
fn parse_glibc_version(s: &str) -> Option<(u64, u64)> {
    let mut parts = s.split('.');
    let major = parts.next()?.parse().ok()?;
    let minor = parts.next()?.parse().ok()?;
    Some((major, minor))
}

/// The glibc version out of `ldd --version`'s first line, whose tail is the
/// version however the distro decorates the rest (`ldd (Ubuntu GLIBC
/// 2.39-0ubuntu8.8) 2.39`).
fn glibc_from_ldd(ldd_out: &str) -> Option<(u64, u64)> {
    let first = ldd_out.lines().find(|l| !l.trim().is_empty())?;
    parse_glibc_version(first.split_whitespace().last()?)
}

impl RecipeCtx {
    /// This target's install destination, or an error naming why there is none.
    fn deploy_target(&self) -> Result<&DeployTarget> {
        self.deploy.as_ref().ok_or_else(|| {
            anyhow::anyhow!(
                "no deploy destination for {} {}: the app is `kind = \"{}\"`, and only a \
                 service declares [[deploy]] entries",
                self.app,
                self.target,
                match self.kind {
                    Kind::App => "app",
                    Kind::Library => "library",
                    Kind::Service => "service",
                }
            )
        })
    }

    /// Compare the built binary's glibc requirement against the service host's.
    /// Returns the two versions for the recipe to log.
    fn glibc_check(self: &Arc<Self>, binary: &str) -> Result<(String, String)> {
        let d = self.deploy_target()?.clone();
        // `objdump -T` on the build host; no symbols at all (a static binary)
        // means nothing to check, which is a pass rather than a failure.
        let (code, out) = self.run(
            &self.build_host.clone(),
            &format!(
                "objdump -T {binary} 2>/dev/null | grep -o 'GLIBC_[0-9.]*' | sort -uV || true"
            ),
        )?;
        anyhow::ensure!(code == 0, "reading glibc symbols from {binary} failed");
        let Some(needs) = max_glibc_symbol(&out) else {
            return Ok(("none".into(), "n/a".into()));
        };
        let (code, ldd) = self.run(&d.host, "ldd --version")?;
        anyhow::ensure!(
            code == 0,
            "could not read glibc version on service host `{}`",
            d.host
        );
        let has = glibc_from_ldd(&ldd).ok_or_else(|| {
            anyhow::anyhow!(
                "could not parse glibc version from `ldd --version` on `{}`",
                d.host
            )
        })?;
        anyhow::ensure!(
            needs <= has,
            "binary needs glibc {}.{} but `{}` has {}.{} — it would fail to exec after the \
             unit restarted onto it. Build on a host no newer than the service host.",
            needs.0,
            needs.1,
            d.host,
            has.0,
            has.1,
        );
        Ok((
            format!("{}.{}", needs.0, needs.1),
            format!("{}.{}", has.0, has.1),
        ))
    }

    /// Install `binary` (a path on the BUILD host) onto the service host and
    /// restart its unit, via the privileged installer the host holds a scoped
    /// sudo grant for.
    ///
    /// Bento never runs the install itself. It stages the bytes and calls a
    /// root script whose arguments are re-checked on the far side — the same
    /// shape as Sando's `install-companion.sh`, and for the same reason: the
    /// sudoers grant is then ONE auditable script rather than a broad
    /// `install`+`systemctl` grant on a production box.
    ///
    /// Only the binary moves. Config is deliberately untouched: pom's
    /// `pom-astra.toml` / `pom-hetzner.toml` differ per instance, and prod's
    /// carried a `[targets.mnw.tests]` block the repo did not have. A deploy
    /// that copies config over is how that block gets silently deleted.
    fn deploy(self: &Arc<Self>, binary: &str) -> Result<String> {
        anyhow::ensure!(
            !self.is_cancelled(),
            "build superseded by a newer request; refusing to deploy"
        );
        // A failed earlier step bars a deploy exactly as it bars a publish. An
        // artifact that failed its gates must not reach a production host just
        // because the recipe kept running.
        let failed = self.failed_steps.lock().unwrap().clone();
        anyhow::ensure!(
            failed.is_empty(),
            "refusing to deploy {} {}: {} failed earlier in this run",
            self.app,
            self.version,
            failed
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join(", "),
        );
        let d = self.deploy_target()?.clone();
        ensure_glob_safe(binary)?;

        // Stage under a fixed root the installer also insists on, so "what was
        // checked" and "what is installed" cannot drift apart.
        let staged = format!("{DEPLOY_STAGING_ROOT}/{}", self.app);
        let staged_bin = format!("{staged}/{}", self.app);
        let deploy_exec = self.exec(&d.host)?;
        anyhow::ensure!(
            deploy_exec.capabilities().permits(&Action::Deploy),
            "service host `{}` is not granted the `deploy` capability",
            d.host
        );

        self.run_ok(&d.host, &format!("mkdir -p {staged}"))?;
        if self.build_host_ssh == d.host {
            // Same box: the binary is already there. Routing it through the
            // daemon would be two transfers to end up where it started. This is
            // pom's aarch64 leg — astra builds it and astra runs it.
            self.run_ok(&d.host, &format!("cp -f {binary} {staged_bin}"))?;
        } else {
            // Build host -> daemon -> service host. Two hops because an executor
            // reaches one host; a direct host-to-host transport would mean the
            // build host holding a credential for the production box.
            let tmp = tempfile::tempdir().context("staging dir for deploy")?;
            let local = tmp.path().join(self.app.as_str());
            self.pull_for_deploy(binary, &local)?;
            let (dest, opts) = (PathBuf::from(&staged), SyncOpts::default());
            let dir = tmp.path().to_path_buf();
            self.run_bounded(&format!("stage {} on `{}`", self.app, d.host), async move {
                deploy_exec.push_dir(&dir, &dest, &opts).await
            })
            .with_context(|| format!("staging {} onto `{}`", self.app, d.host))?;
        }

        // The privileged half. Every argument is re-validated by the script,
        // which is the thing actually holding the sudo grant.
        self.run_ok(
            &d.host,
            &format!(
                "{} {staged_bin} {} {}",
                self.cfg.deploy_installer, d.install_path, d.service
            ),
        )?;
        Ok(format!(
            "{} {} installed at {} on `{}`; {} restarted",
            self.app, self.version, d.install_path, d.host, d.service
        ))
    }

    /// Fetch one file off a host into a daemon-local path for re-pushing.
    ///
    /// A local build host is read directly: `fw13` is the daemon's own box, so
    /// the file is already on this filesystem. Routing it through the
    /// artifact-pull gate instead would demand a `pull_root` covering every repo
    /// a service could be built in — today that is `~/Code/Apps`, and pom lives
    /// in `~/Code/MNW`. Widening it to `~/Code` would put `_private`, the
    /// secrets root, inside the collectable tree. This is pom's x86_64 leg.
    fn pull_for_deploy(self: &Arc<Self>, remote: &str, local: &Path) -> Result<()> {
        let host = self.build_host.clone();
        let remote_path = expand_tilde(remote);
        if self.build_host_ssh == "local" || self.build_host_ssh.is_empty() {
            std::fs::copy(&remote_path, local).with_context(|| {
                format!("staging {} from the daemon host", remote_path.display())
            })?;
            return Ok(());
        }
        let sync = self.host_sync(&host)?;
        let (src, dst, opts) = (remote_path, local.to_path_buf(), SyncOpts::default());
        self.run_bounded(&format!("fetch {remote} from `{host}`"), async move {
            sync.pull_file(&src, &dst, &opts).await
        })
        .with_context(|| format!("fetching {remote} from `{host}` to deploy"))
    }

    /// `run`, failing the step on a non-zero exit. The Rust-side twin of the
    /// recipe's `sh_ok`, for commands the deploy machinery issues itself.
    fn run_ok(self: &Arc<Self>, host: &str, cmd: &str) -> Result<String> {
        let (code, tail) = self.run(host, cmd)?;
        if code != 0 {
            self.fail_current_step();
            anyhow::bail!("command on `{host}` exited {code}: {cmd}\n{tail}");
        }
        Ok(tail)
    }

    /// Where this run's collected files land locally.
    ///
    /// Per target, not per version. Sharing one `dist_root/<app>/<version>/`
    /// across targets would have the hash loop below (which lists the directory)
    /// attribute a sibling's AppImage to the mac build's artifact record. It is
    /// also the layout the archive uses, and the two have to agree or the local
    /// copy and the deposited one are different shapes.
    fn collect_dest(&self, app: &str, version: &str) -> PathBuf {
        self.cfg
            .dist_root
            .join(app)
            .join(version)
            .join(crate::archive::target_slug(self.target))
    }

    fn collect(self: &Arc<Self>, host: &str, glob: &str, app: &str, version: &str) -> Result<()> {
        let dest = self.collect_dest(app, version);
        let dest_s = dest.to_string_lossy().into_owned();
        // The glob reaches a remote login shell intact (that's what expands it),
        // so command metacharacters stay barred. Path/wildcard chars are fine.
        ensure_glob_safe(glob)?;
        std::fs::create_dir_all(&dest)
            .with_context(|| format!("creating collect dest {dest_s}"))?;
        // The SYNC transport, not the host's exec executor: artifacts move over
        // ssh/rsync even from an agent host, whose `/pull` is confined to a
        // narrow `pull_root` that deliberately excludes the repo checkout these
        // artifacts are built in (see `state::build_sync`). The daemon still
        // runs the transfer itself, as it always has.
        let sync = self.host_sync(host)?;
        let opts = SyncOpts::precompressed();
        // Bounded by the collect step's deadline (rsync of a multi-GiB artifact
        // can wedge on a stalled transport) and interruptible on supersession.
        let dest_pull = dest.clone();
        self.run_bounded(&format!("collect {glob} from `{host}`"), async move {
            sync.pull_glob(glob, &dest_pull, &opts).await
        })
        .with_context(|| format!("collect {glob} from `{host}`"))?;
        // Assert the version and hash every collected file. This is where a
        // stale artifact is caught: a file whose name embeds a different version
        // fails the collect (rather than silently winning a later glob), and the
        // sha256 recorded here is what `publish` writes into the release ledger
        // and what the artifact record's manifest is built from.
        //
        // Recursive, and keyed by path relative to the collect dir. That is not
        // a preference: Sando's intake re-hashes the bundle with its own walker,
        // which recurses and keys the same way, and refuses a bundle whose bytes
        // do not match the manifest it was handed. A top-level `read_dir` keyed
        // by file name agrees with that walker for a flat directory and diverges
        // the moment a bundle carries a subdirectory — the honest artifact would
        // be refused for a manifest that omitted everything nested. The two
        // walkers have to be the same walk. See `bundle::digest_dir` in sando.
        for (rel, path) in
            collected_files(&dest).with_context(|| format!("listing collect dest {dest_s}"))?
        {
            // The version check stays on the file NAME rather than the relative
            // path: it is looking for a stale `app_1.2.3.AppImage` beside the
            // one this release built, and a directory component is not that.
            let name = path
                .file_name()
                .map_or_else(|| rel.clone(), |n| n.to_string_lossy().into_owned());
            assert_artifact_version(&name, &self.version)?;
            let digest = sha256_file(&path)?;
            self.artifact_hashes.lock().unwrap().insert(rel, digest);
        }
        // Deposit at the archive path, so this target's bytes have one address
        // whichever host produced them. A no-op when no archive is configured.
        //
        // Inside `collect`, not after the recipe: a failure here fails the
        // collect step, before sign and publish, rather than putting a red mark
        // on a release that has already shipped. And it is a failure, not a
        // warning — a deposit that is quietly skipped leaves the archive path
        // wrong for exactly the release nobody was watching, which is the thing
        // having one address is for.
        let (cfg, app_id, version, target) = (
            self.cfg.clone(),
            self.app.clone(),
            self.version.clone(),
            self.target,
        );
        let dest_archive = dest.clone();
        self.run_bounded("deposit in the archive", async move {
            crate::archive::deposit(&cfg, &dest_archive, &app_id, &version, target).await
        })?;
        // Best-effort size accounting for the event.
        events::emit(
            &self.events,
            Event::ArtifactCollected {
                app: self.app.clone(),
                target: self.target,
                path: dest_s,
                bytes: dir_size(&dest).unwrap_or(0),
            },
        );
        Ok(())
    }

    /// The all-targets-green gate: err unless every declared target OTHER than
    /// the one publishing has a latest `target_runs` row of `ok` for this
    /// `(app, version)`. A sibling with no run, a running run, or a failed
    /// latest run all block the publish, naming what is not green.
    fn assert_siblings_green(self: &Arc<Self>, declared: &[Target]) -> Result<()> {
        let me = self.clone();
        let (app_s, ver_s) = (self.app.to_string(), self.version.to_string());
        let rows: Vec<(String, String)> = self.rt.block_on(async move {
            sqlx::query_as(
                "SELECT target, status FROM target_runs tr
                 WHERE app = ?1 AND version = ?2
                   AND id = (SELECT MAX(id) FROM target_runs
                             WHERE app = ?1 AND version = ?2 AND target = tr.target)",
            )
            .bind(app_s)
            .bind(ver_s)
            .fetch_all(&me.pool)
            .await
            .unwrap_or_default()
        });
        let status_of = |t: &Target| -> Option<String> {
            let key = t.to_string();
            rows.iter()
                .find(|(name, _)| name == &key)
                .map(|(_, s)| s.clone())
        };
        let not_green: Vec<String> = declared
            .iter()
            .filter(|t| **t != self.target) // the publishing target is the last mile
            .filter(|t| status_of(t).as_deref() != Some("ok"))
            .map(|t| format!("{t} ({})", status_of(t).unwrap_or_else(|| "no run".into())))
            .collect();
        anyhow::ensure!(
            not_green.is_empty(),
            "all-targets-green gate: refusing to publish {} {} — not green: {}",
            self.app,
            self.version,
            not_green.join(", "),
        );
        Ok(())
    }

    fn publish(
        self: &Arc<Self>,
        channel: &str,
        app: &str,
        target: &str,
        version: &str,
        artifact: &str,
        meta: &Map,
    ) -> Result<String> {
        // Never let a superseded build ship. This is the last and most important
        // cooperative-cancel checkpoint: even if a long-running step finished
        // after supersession, the artifact must not reach the backend.
        anyhow::ensure!(
            !self.is_cancelled(),
            "build superseded by a newer request; refusing to publish"
        );
        // Opt-in all-targets-green gate: refuse a partial release. Every OTHER
        // declared target of this (app, version) must have a successful latest
        // run before this one ships, so macOS can't publish while windows is red
        // or still building. The publishing target itself is the last mile (it
        // reached publish, so its steps passed) and is not required to be green
        // in the ledger yet.
        if let Some(declared) = self.all_green_required.clone() {
            self.assert_siblings_green(&declared)?;
        }
        let backend = self
            .ota
            .get(channel)
            .ok_or_else(|| anyhow::anyhow!("unknown publish channel `{channel}`"))?;
        let target: Target = target.parse().map_err(|e: String| anyhow::anyhow!(e))?;
        let version = Version::parse(version).map_err(|e| anyhow::anyhow!(e))?;
        let app = AppId::new(app);

        // The backend must actually handle this target (e.g. the desktop updater
        // disclaims iOS) — otherwise publish would push an artifact through a
        // backend that does not support it.
        anyhow::ensure!(
            backend.supports(target),
            "publish channel `{channel}` does not support target {target}",
        );

        // Monotonicity: never publish a version that is not strictly newer than
        // the latest already published for this (app, target, channel). Without
        // this an older build could republish over a live newer release. The
        // `releases` column is TEXT, so compare by parsed semver precedence
        // (Version: Ord), not lexically.
        {
            let (app_s, target_s, chan_s) =
                (app.to_string(), target.to_string(), channel.to_string());
            let me = self.clone();
            let latest: Option<Version> = self.rt.block_on(async move {
                let rows: Vec<(String,)> = sqlx::query_as(
                    "SELECT version FROM releases WHERE app = ? AND target = ? AND channel = ?",
                )
                .bind(app_s)
                .bind(target_s)
                .bind(chan_s)
                .fetch_all(&me.pool)
                .await
                .unwrap_or_default();
                rows.into_iter()
                    .filter_map(|(v,)| Version::parse(&v).ok())
                    .max()
            });
            if let Some(latest) = latest {
                anyhow::ensure!(
                    version > latest,
                    "refusing to publish {app} {version} to `{channel}` ({target}): \
                     not newer than the last published {latest}",
                );
            }
        }

        // Step-success ledger (the Bento analogue of Sando's gate fail-closed),
        // minted as an unforgeable PublishAuthority. `backend.publish` cannot be
        // called without one, so the unverified/post-failure ship path is sealed
        // at the type level rather than guarded by a separate runtime check.
        let authority = {
            let failed = self.failed_steps.lock().unwrap();
            let gatekeeper = *self.gatekeeper_ok.lock().unwrap();
            PublishAuthority::prove(target, failed.as_slice(), gatekeeper)?
        };
        let notes = meta
            .get("notes")
            .and_then(|v| v.clone().into_string().ok())
            .unwrap_or_default();
        // Resolve the artifact relative to the collected dist dir if not absolute.
        let artifact_path = {
            let p = PathBuf::from(artifact);
            if p.is_absolute() {
                p
            } else {
                self.collect_dest(app.as_str(), &version.to_string())
                    .join(artifact)
            }
        };
        let rel = Release {
            app: &app,
            target,
            version: &version,
            notes,
        };
        let receipt = backend
            .publish(&rel, &artifact_path, &authority)
            .with_context(|| format!("publish to `{channel}`"))?;
        // Record for idempotency / monotonicity. This write is CHECKED, not
        // fire-and-forget: a swallowed failure here would silently re-arm the
        // monotonicity guard (which reads this same table), letting an older
        // version republish over a live release. Concurrent same-(app,target)
        // publishers can't race the read-then-insert because the latest-wins slot
        // (state::ActiveSlot) serializes them and a superseded run is cancelled
        // before it reaches publish.
        // The artifact's hash, recorded so the release ledger says exactly which
        // bytes shipped. Prefer the digest computed at `collect`; fall back to
        // hashing the file now (an absolute-path artifact never routed through
        // `collect`). A hash failure must not fail an already-published release,
        // so degrade to NULL rather than erroring.
        let artifact_hash: Option<String> = artifact_path
            .file_name()
            .and_then(|n| n.to_str())
            .and_then(|n| self.artifact_hashes.lock().unwrap().get(n).cloned())
            .or_else(|| sha256_file(&artifact_path).ok());
        let me = self.clone();
        let (app_s, target_s, ver_s, chan_s) = (
            app.to_string(),
            target.to_string(),
            version.to_string(),
            channel.to_string(),
        );
        self.rt
            .block_on(async move {
                sqlx::query(
                    "INSERT OR IGNORE INTO releases (app, target, version, channel, artifact_hash, published_at)
                     VALUES (?, ?, ?, ?, ?, ?)",
                )
                .bind(app_s)
                .bind(target_s)
                .bind(ver_s)
                .bind(chan_s)
                .bind(artifact_hash)
                .bind(Self::now())
                .execute(&me.pool)
                .await
            })
            .context("recording release in the idempotency ledger (artifact published but ledger write failed)")?;
        events::emit(
            &self.events,
            Event::PublishOk {
                app: self.app.clone(),
                target: self.target,
                channel: channel.to_string(),
            },
        );
        Ok(receipt)
    }
}

fn dir_size(p: &Path) -> Option<i64> {
    let mut total = 0i64;
    for entry in std::fs::read_dir(p).ok()? {
        let entry = entry.ok()?;
        let md = entry.metadata().ok()?;
        if md.is_file() {
            total += md.len() as i64;
        } else if md.is_dir() {
            // Recurse so a bundle dir (a `.app`) reports its real size, not ~0.
            total += dir_size(&entry.path()).unwrap_or(0);
        }
    }
    Some(total)
}

/// macOS signing/notarization host functions. Thin wrappers over the right
/// shell incantations, dispatched through the named host's executor. On the mac
/// host (`transport = "agent"`) they run via the in-session `ops-agent`, the only
/// context where codesign can use the Developer ID key (design §7 "THE WALL"); a
/// plain SSH session cannot. Each is gated by the host's `sign` capability.
fn register_macos_fns(engine: &mut Engine, ctx: &Arc<RecipeCtx>) {
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "verify_gatekeeper",
            move |host: &str, path: &str| -> Result<bool, Box<EvalAltResult>> {
                // spctl has no JSON mode, so assess on-host and decide there,
                // emitting an unambiguous sentinel as the final line. We match the
                // sentinel rather than substring-hunting `source=Notarized...` in a
                // 2000-char tail: truncation only drops the front, so the sentinel
                // is always present, and it can't be spoofed by spctl's own prose.
                // The full assess output is still streamed to the step log.
                let q = ops_core::remote::sh_quote(path);
                let cmd = format!(
                    "out=$(spctl --assess -vv --type install {q} 2>&1); printf '%s\\n' \"$out\"; \
                     printf '%s' \"$out\" | grep -q 'source=Notarized Developer ID' \
                     && echo BENTO_GATEKEEPER_OK || echo BENTO_GATEKEEPER_FAIL",
                );
                let (_, tail) = ctx.run(host, &cmd).map_err(rhai_err)?;
                let accepted = tail.contains("BENTO_GATEKEEPER_OK");
                // Record the verdict for the publish gate. A rejection also
                // fails the step, so the matrix shows red and `publish` is barred
                // even if the recipe ignores the returned bool.
                *ctx.gatekeeper_ok.lock().unwrap() = Some(accepted);
                if !accepted {
                    ctx.fail_current_step();
                }
                Ok(accepted)
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "codesign",
            move |host: &str, identity: &str, path: &str| -> Result<(), Box<EvalAltResult>> {
                let cmd = format!(
                    "codesign --force --options runtime --timestamp --sign {} {}",
                    ops_core::remote::sh_quote(identity),
                    ops_core::remote::sh_quote(path),
                );
                let (code, _) = ctx.run(host, &cmd).map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("codesign failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "staple",
            move |host: &str, path: &str| -> Result<(), Box<EvalAltResult>> {
                let (code, _) = ctx
                    .run(
                        host,
                        &format!("xcrun stapler staple {}", ops_core::remote::sh_quote(path)),
                    )
                    .map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("stapler failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "notarize",
            move |host: &str, path: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.notarize(host, path).map_err(rhai_err)
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "keychain_open",
            move |host: &str, name: &str| -> Result<(), Box<EvalAltResult>> {
                // The full build-keychain lifecycle lives in dist/build-keychain.sh
                // (design §7); this drives it by name so the recipe stays short.
                let (code, _) = ctx
                    .run(
                        host,
                        &format!(
                            ". ~/.tauri/passwords.env && ./dist/build-keychain.sh open {}",
                            ops_core::remote::sh_quote(name)
                        ),
                    )
                    .map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("keychain_open failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "keychain_close",
            move |host: &str, name: &str| -> Result<(), Box<EvalAltResult>> {
                let _ = ctx.run(
                    host,
                    &format!(
                        "./dist/build-keychain.sh close {}",
                        ops_core::remote::sh_quote(name)
                    ),
                );
                Ok(())
            },
        );
    }
}

impl RecipeCtx {
    /// `xcrun notarytool submit --wait` with bounded retry (the one flaky,
    /// network-bound step). Emits `NotarizeRetry` per attempt.
    fn notarize(self: &Arc<Self>, host: &str, path: &str) -> Result<String> {
        const MAX_ATTEMPTS: u32 = 3;
        let backoff = self
            .cfg
            .notarize_backoff_secs
            .map_or(std::time::Duration::from_secs(15), |s| {
                std::time::Duration::from_secs(s)
            });
        let cmd = format!(
            ". ~/.tauri/passwords.env && xcrun notarytool submit {} \
             --key \"$NOTARY_KEY\" --key-id \"$NOTARY_KEY_ID\" --issuer \"$NOTARY_ISSUER\" \
             --wait --output-format json",
            ops_core::remote::sh_quote(path),
        );
        let mut last = String::new();
        for attempt in 1..=MAX_ATTEMPTS {
            let (code, tail) = self.run(host, &cmd)?;
            if code == 0 && notary_accepted(&tail) {
                return Ok(tail);
            }
            last = tail;
            if attempt < MAX_ATTEMPTS {
                events::emit(
                    &self.events,
                    Event::NotarizeRetry {
                        app: self.app.clone(),
                        target: self.target,
                        attempt,
                        reason: format!("exit {code}"),
                    },
                );
                self.rt.block_on(tokio::time::sleep(backoff));
            }
        }
        anyhow::bail!("notarization failed after {MAX_ATTEMPTS} attempts: {last}")
    }
}

/// True iff `notarytool --output-format json` output reports `status: Accepted`.
/// Isolates the JSON object (`{`..`}`) from any shell-sourcing noise and reads
/// the typed `status` field, rather than substring-matching `"status":"Accepted"`
/// in a possibly-truncated tail — which could match the literal inside an error
/// message or miss it across a whitespace variant. Fails closed: any parse or
/// field miss returns false.
fn notary_accepted(output: &str) -> bool {
    let (Some(start), Some(end)) = (output.find('{'), output.rfind('}')) else {
        return false;
    };
    if start > end {
        return false;
    }
    serde_json::from_str::<serde_json::Value>(&output[start..=end])
        .ok()
        .and_then(|v| {
            v.get("status")
                .and_then(|s| s.as_str())
                .map(|s| s.eq_ignore_ascii_case("accepted"))
        })
        .unwrap_or(false)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Build the shared cross-crate bundle fixture under `root`.
    ///
    /// A binary at the top and two files in a subdirectory: the shape a service
    /// that ships its migrations has, and the case a flat walk gets wrong.
    pub(crate) fn write_bundle_fixture(root: &Path) {
        std::fs::create_dir_all(root.join("migrations")).unwrap();
        std::fs::write(root.join("pom"), b"binary-bytes").unwrap();
        std::fs::write(root.join("migrations/001_init.sql"), b"create table a;").unwrap();
        std::fs::write(root.join("migrations/002_next.sql"), b"alter table a;").unwrap();
    }

    /// The manifest text the fixture must produce, in BOTH crates.
    ///
    /// Sando's `bundle::digest_dir` has the identical constant and the identical
    /// fixture. That is the whole point: bento writes this text into the artifact
    /// record, sando recomputes it from the bytes that arrive, and an artifact is
    /// refused when they differ. Two walks, one answer, pinned from both ends —
    /// if either crate's walk drifts, its own test fails and names the drift
    /// rather than a release failing intake for a bundle nothing is wrong with.
    pub(crate) const BUNDLE_FIXTURE_MANIFEST: &str = concat!(
        "e4c908e219c533fa7ad7ea1634398f9bf51637ba20717769ada545bab26d7368  migrations/001_init.sql\n",
        "b026fd51bae096b34672cefdb781b6585b13efb53bc301d50c305f422552a380  migrations/002_next.sql\n",
        "71227a7f160afca3fb3c39f448735886dda7bd366252580c2222fb87d4bb4d85  pom\n",
    );

    /// The producer half of the contract above: what `collect` hashes, turned
    /// into a manifest, is exactly the text the verifier will recompute.
    ///
    /// Nested files are included and addressed by relative path. Before this,
    /// `collect` listed only the top level, so `migrations/` contributed nothing
    /// to the manifest while sando's walker hashed both files in it — and the
    /// honest bundle was refused for a manifest that had omitted them.
    #[test]
    fn a_collected_bundle_manifests_exactly_as_the_verifier_will_read_it() {
        let dir = tempfile::tempdir().unwrap();
        write_bundle_fixture(dir.path());

        let files = collected_files(dir.path()).unwrap();
        assert_eq!(
            files.iter().map(|(r, _)| r.as_str()).collect::<Vec<_>>(),
            vec!["migrations/001_init.sql", "migrations/002_next.sql", "pom"],
            "recursive, relative, sorted"
        );

        let hashes: Vec<(String, String)> = files
            .into_iter()
            .map(|(rel, path)| (rel, sha256_file(&path).unwrap()))
            .collect();
        let manifest = ops_artifact::Manifest::new(hashes).unwrap();
        assert_eq!(manifest.to_text(), BUNDLE_FIXTURE_MANIFEST);
    }

    /// A symlink is neither followed nor named. Following one would let bytes
    /// from outside the bundle into its identity; naming it would put a path in
    /// the manifest the verifier does not hash, which reads as a corrupt bundle.
    #[test]
    #[cfg(unix)]
    fn a_symlink_in_the_collect_dir_is_not_part_of_the_bundle() {
        let dir = tempfile::tempdir().unwrap();
        write_bundle_fixture(dir.path());
        let outside = dir.path().join("..").join("secret.env");
        std::fs::write(&outside, b"TOKEN=1").ok();
        std::os::unix::fs::symlink(&outside, dir.path().join("link.env")).unwrap();

        let files = collected_files(dir.path()).unwrap();
        assert!(
            !files.iter().any(|(rel, _)| rel.contains("link.env")),
            "{files:?}"
        );
    }

    /// Run 3 S1: once the cooperative cancel flag is set (a newer build
    /// superseded this run), a step boundary refuses to proceed — the blocking
    /// recipe stops at the next `step()` instead of running on and publishing.
    #[tokio::test]
    async fn begin_step_bails_when_cancelled() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let cancel = Arc::new(AtomicBool::new(true));
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::App,
            1,
            Arc::new(std::collections::HashMap::new()),
            Arc::new(std::collections::HashMap::new()),
            None,
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            cancel.clone(),
            None,
        ));
        // Cancelled: begin_step refuses before touching the DB (the ensure! is
        // ahead of any block_on, so this is safe to call from the async test).
        let err = ctx.begin_step(Step::Build).unwrap_err();
        assert!(err.to_string().contains("supersede"), "got: {err}");
        assert!(ctx.is_cancelled());
    }

    /// `feature_flags()` returns a whole flag or nothing at all. An app with
    /// no declared features must not yield a bare `--features`, which would
    /// swallow the next word of the build command as its argument.
    #[tokio::test]
    async fn feature_flags_renders_whole_flag_or_empty() {
        async fn flags_for(features: Vec<String>) -> String {
            let dir = tempfile::tempdir().unwrap();
            let cfg = Arc::new(Config::for_tests(dir.path()));
            let pool = crate::db::open(&cfg.db_path).await.unwrap();
            let ctx = Arc::new(RecipeCtx::new(
                AppId::new("demo"),
                Version::parse("0.1.0").unwrap(),
                "linux/x86_64".parse().unwrap(),
                "fw13".into(),
                "local".into(),
                "v0.1.0".into(),
                "/tmp".into(),
                features,
                Kind::App,
                1,
                Arc::new(std::collections::HashMap::new()),
                Arc::new(std::collections::HashMap::new()),
                None,
                pool,
                crate::events::channel(),
                cfg,
                Arc::new(OtaRegistry::standard("https://makenot.work")),
                tokio::runtime::Handle::current(),
                Arc::new(AtomicBool::new(false)),
                None,
            ));
            let engine = build_engine(&ctx);
            engine.eval::<String>("feature_flags()").unwrap()
        }

        assert_eq!(flags_for(vec![]).await, "");
        assert_eq!(
            flags_for(vec!["supernote".into()]).await,
            "--features supernote"
        );
        assert_eq!(
            flags_for(vec!["supernote".into(), "extra".into()]).await,
            "--features supernote,extra"
        );
    }

    /// `repo()` answers for the host this target builds on, not for the daemon.
    ///
    /// This is what lets a Windows recipe call `repo()` and `checkout_sha(h)`
    /// instead of hard-coding `C:/Users/me/...` — and hard-coding it is what
    /// kept those recipes off the release-tag pin, since `checkout_sha` builds
    /// its git commands from the app's path and takes no override.
    #[tokio::test]
    async fn repo_resolves_per_build_host() {
        async fn repo_on(build_host: &str) -> String {
            let dir = tempfile::tempdir().unwrap();
            let cfg = Arc::new(Config::for_tests(dir.path()));
            let pool = crate::db::open(&cfg.db_path).await.unwrap();
            let ctx = Arc::new(
                RecipeCtx::new(
                    AppId::new("demo"),
                    Version::parse("0.1.0").unwrap(),
                    "linux/x86_64".parse().unwrap(),
                    build_host.into(),
                    "local".into(),
                    "v0.1.0".into(),
                    "~/Code/Apps/demo".into(),
                    vec![],
                    Kind::App,
                    1,
                    Arc::new(std::collections::HashMap::new()),
                    Arc::new(std::collections::HashMap::new()),
                    None,
                    pool,
                    crate::events::channel(),
                    cfg,
                    Arc::new(OtaRegistry::standard("https://makenot.work")),
                    tokio::runtime::Handle::current(),
                    Arc::new(AtomicBool::new(false)),
                    None,
                )
                .with_repo_by_host(HashMap::from([(
                    "windows-x86".to_string(),
                    "C:/Users/me/Code/Apps/demo".to_string(),
                )])),
            );
            build_engine(&ctx).eval::<String>("repo()").unwrap()
        }

        assert_eq!(repo_on("windows-x86").await, "C:/Users/me/Code/Apps/demo");
        assert_eq!(repo_on("fw13").await, "~/Code/Apps/demo");
    }

    /// `secret(key)` reads a file under `secrets_root`, trims its trailing
    /// newline (the shape of a here-doc'd token file), and refuses any key that
    /// could escape the root. Covers the host-fn registered in `build_engine`.
    #[tokio::test]
    async fn secret_reads_under_root_and_blocks_traversal() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::for_tests(dir.path());
        // Seed a secret and one in a nested subdir; a trailing newline that the
        // read must strip.
        std::fs::create_dir_all(&cfg.secrets_root).unwrap();
        std::fs::write(cfg.secrets_root.join("token"), "s3cr3t\n").unwrap();
        std::fs::create_dir_all(cfg.secrets_root.join("app")).unwrap();
        std::fs::write(cfg.secrets_root.join("app").join("key"), "nested").unwrap();
        // Plant a file OUTSIDE the root that a traversal key would reach.
        std::fs::write(dir.path().join("outside"), "leak").unwrap();

        let cfg = Arc::new(cfg);
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::App,
            1,
            Arc::new(std::collections::HashMap::new()),
            Arc::new(std::collections::HashMap::new()),
            None,
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));
        let engine = build_engine(&ctx);

        // Happy path: read + trim.
        assert_eq!(
            engine.eval::<String>(r#"secret("token")"#).unwrap(),
            "s3cr3t"
        );
        // A multi-segment relative key is allowed.
        assert_eq!(
            engine.eval::<String>(r#"secret("app/key")"#).unwrap(),
            "nested"
        );

        // Traversal, absolute paths, and empty keys are refused BEFORE any read,
        // so the file one `..` above the root is never disclosed.
        for bad in [
            r#"secret("../outside")"#,
            r#"secret("/etc/passwd")"#,
            r#"secret("")"#,
        ] {
            let err = engine.eval::<String>(bad).unwrap_err().to_string();
            assert!(
                err.contains("relative path under secrets_root"),
                "`{bad}` should hit the traversal guard, got: {err}"
            );
        }
        // A missing key surfaces the filesystem error, not a panic, and does not
        // trip the traversal guard (it is a legitimate relative path).
        let err = engine
            .eval::<String>(r#"secret("nope")"#)
            .unwrap_err()
            .to_string();
        assert!(err.contains("secret `nope`"), "got: {err}");
    }

    /// The two failures that actually shipped, as regression cases.
    #[test]
    fn preflight_catches_a_dead_repository_url() {
        // pter 0.1.0: repository pointed at a URL that does not exist. It
        // published clean and the link is now permanent for that version.
        let meta = CrateMeta {
            name: "pter".into(),
            version: "0.1.0".into(),
            repository: Some("https://github.com/maxjacobson/pter".into()),
            description: Some("d".into()),
            licensed: true,
        };
        let problems = crate_publish_problems(&meta, false, &[], true);
        assert_eq!(problems.len(), 1, "{problems:?}");
        assert!(
            problems[0].contains("not publicly clonable"),
            "{problems:?}"
        );

        // Same metadata, reachable URL: nothing to report.
        assert!(crate_publish_problems(&meta, true, &[], true).is_empty());
    }

    #[test]
    fn preflight_requires_the_fields_crates_io_bakes_in() {
        let bare = CrateMeta {
            name: "x".into(),
            version: "0.1.0".into(),
            repository: None,
            description: None,
            licensed: false,
        };
        let problems = crate_publish_problems(&bare, false, &[], true);
        assert_eq!(problems.len(), 3, "{problems:?}");
        assert!(problems.iter().any(|p| p.contains("repository")));
        assert!(problems.iter().any(|p| p.contains("description")));
        assert!(problems.iter().any(|p| p.contains("license")));
    }

    // A library's verify is a crate preflight, not a Gatekeeper check on a
    // signed bundle. Gating it on `gatekeeper` asked a Linux host for a macOS
    // code-signing capability it can never hold, so the step was denied before
    // it ran a command; the denial then surfaced as "no crates.io credentials",
    // which is not what went wrong. The only way to satisfy the old gate was to
    // declare the capability falsely in the topology.
    #[test]
    fn a_library_verify_is_not_gated_on_gatekeeper() {
        assert_eq!(
            action_for(Step::Verify, Kind::Library),
            Action::Build,
            "a crate preflight runs the build toolchain; that is what it needs",
        );
        assert_eq!(
            action_for(Step::Verify, Kind::App),
            Action::Observe(ObserveKind::Custom("gatekeeper".into())),
            "an app's verify still proves the bundle is signed and notarized",
        );
    }

    // The capability the default host grant actually carries. Without this the
    // fix above is only true by inspection.
    #[test]
    fn a_default_host_can_run_a_library_verify_and_not_an_app_one() {
        let caps =
            ops_exec::CapabilitySet::from_tokens(["build", "package"], ["build-log", "artifact"]);
        assert!(caps.permits(&action_for(Step::Verify, Kind::Library)));
        assert!(!caps.permits(&action_for(Step::Verify, Kind::App)));
    }

    // Every other step is a property of the step alone; verify is the one that
    // depends on what is being released.
    #[test]
    fn no_other_step_changes_with_the_kind() {
        for step in [
            Step::Checkout,
            Step::Prebuild,
            Step::Build,
            Step::Sign,
            Step::Notarize,
            Step::Staple,
            Step::Package,
            Step::Publish,
            Step::Collect,
        ] {
            assert_eq!(
                action_for(step, Kind::App),
                action_for(step, Kind::Library),
                "{step:?} should not depend on the kind",
            );
        }
    }

    #[test]
    fn preflight_rejects_republishing_the_same_version() {
        let meta = CrateMeta {
            name: "makeover".into(),
            version: "0.10.0".into(),
            repository: Some("https://git.sr.ht/~maxmj/makeover".into()),
            description: Some("d".into()),
            licensed: true,
        };
        let problems =
            crate_publish_problems(&meta, true, &["0.9.0".into(), "0.10.0".into()], true);
        assert_eq!(problems.len(), 1, "{problems:?}");
        assert!(problems[0].contains("already published"), "{problems:?}");

        // An unreleased version against the same history is fine.
        let mut next = meta.clone();
        next.version = "0.11.0".into();
        assert!(crate_publish_problems(&next, true, &["0.10.0".into()], true).is_empty());
    }

    /// Missing credentials must surface at preflight, not at the upload. The
    /// publish step is the irreversible one and runs last, after a full build
    /// and verify; discovering there that cargo cannot authenticate wastes the
    /// whole run.
    #[test]
    fn preflight_reports_missing_credentials_up_front() {
        let meta = CrateMeta {
            name: "makeover".into(),
            version: "0.11.0".into(),
            repository: Some("https://git.sr.ht/~maxmj/makeover".into()),
            description: Some("d".into()),
            licensed: true,
        };
        // Metadata is perfect; only the token is absent.
        let problems = crate_publish_problems(&meta, true, &[], false);
        assert_eq!(problems.len(), 1, "{problems:?}");
        assert!(problems[0].contains("credentials"), "{problems:?}");
        assert!(
            problems[0].contains("cargo login"),
            "should say how to fix it"
        );

        // Present: nothing to report.
        assert!(crate_publish_problems(&meta, true, &[], true).is_empty());
    }

    #[test]
    fn crate_meta_reads_cargo_metadata_json() {
        let raw = r#"{"packages":[{"name":"makeover","version":"0.10.0",
            "repository":"https://git.sr.ht/~maxmj/makeover","description":"themes",
            "license":"MIT"}]}"#;
        let m = crate_meta_from_json(raw).unwrap();
        assert_eq!(m.name, "makeover");
        assert_eq!(m.version, "0.10.0");
        assert!(m.licensed);
        assert_eq!(
            m.repository.as_deref(),
            Some("https://git.sr.ht/~maxmj/makeover")
        );

        // license_file alone also counts as licensed; empty strings do not
        // count as present.
        let lf = r#"{"packages":[{"name":"x","version":"0.1.0","license":"",
            "license_file":"LICENSE","description":""}]}"#;
        let m = crate_meta_from_json(lf).unwrap();
        assert!(m.licensed);
        assert!(m.description.is_none());
    }

    /// Reads the ambient `HOME` rather than setting one. `set_var` is
    /// process-global and unsynchronized, so a test that overwrote HOME changed
    /// it for every other test in the binary — which is what silently disabled
    /// `topology::live_config_smoke` (it skips when `$HOME/.config/bento` is
    /// absent, and `/home/test` always is).
    #[test]
    fn expand_tilde_handles_home() {
        let home = PathBuf::from(std::env::var("HOME").expect("HOME is set"));
        assert_eq!(expand_tilde("~/Code/x"), home.join("Code/x"));
        assert_eq!(expand_tilde("/abs/path"), PathBuf::from("/abs/path"));
    }

    // ---- artifact resolution (the M3 silent-`sh` fix) ----

    #[test]
    fn resolve_artifact_match_wants_exactly_one() {
        // Exactly one match: the path, trimmed of the listing's line noise.
        assert_eq!(
            resolve_artifact_match("  /d/App.AppImage \n", "*.AppImage", true).unwrap(),
            "/d/App.AppImage"
        );
    }

    #[test]
    fn resolve_artifact_match_zero_depends_on_required() {
        // Required + zero matches is the case the old empty-string guard caught;
        // keep failing it.
        let err = resolve_artifact_match("", "*.dmg", true).unwrap_err();
        assert!(err.to_string().contains("no artifact matched"), "{err}");
        // Optional + zero matches resolves to empty (recipe skips the collect).
        assert_eq!(
            resolve_artifact_match("\n  \n", "*.deb", false).unwrap(),
            ""
        );
    }

    #[test]
    fn resolve_artifact_match_rejects_ambiguous() {
        // Two matches must throw rather than silently pick one — this is the
        // stale-newest-mtime hole the audit flagged. Applies even when optional.
        for required in [true, false] {
            let err =
                resolve_artifact_match("/d/old.deb\n/d/new.deb\n", "*.deb", required).unwrap_err();
            let msg = err.to_string();
            assert!(msg.contains("ambiguous"), "{msg}");
            assert!(msg.contains("old.deb") && msg.contains("new.deb"), "{msg}");
        }
    }

    #[test]
    fn ensure_glob_safe_allows_paths_bars_commands() {
        // Path and wildcard characters pass.
        assert!(ensure_glob_safe("~/Code/app/dist/*.AppImage").is_ok());
        assert!(ensure_glob_safe("/t/App_1.2.3-x86_64.dmg").is_ok());
        // A command substitution or separator does not.
        for bad in ["*.dmg; rm -rf /", "$(evil)", "a|b", "a b"] {
            assert!(ensure_glob_safe(bad).is_err(), "should reject {bad:?}");
        }
    }

    // ---- version resolution ----

    #[test]
    fn version_from_tauri_json_reads_version() {
        assert_eq!(
            version_from_tauri_json(r#"{"version":"0.4.2"}"#).unwrap(),
            "0.4.2"
        );
        assert!(version_from_tauri_json(r#"{"productName":"X"}"#).is_err());
    }

    #[test]
    fn version_from_cargo_toml_prefers_package_then_workspace() {
        // A leaf crate's [package].version.
        assert_eq!(
            version_from_cargo_toml("[package]\nname = \"x\"\nversion = \"0.5.0\"\n").unwrap(),
            "0.5.0"
        );
        // A workspace that sets [workspace.package].version.
        assert_eq!(
            version_from_cargo_toml("[workspace.package]\nversion = \"1.2.3\"\n").unwrap(),
            "1.2.3"
        );
        // No version anywhere -> error, not a panic.
        assert!(version_from_cargo_toml("[workspace]\nmembers = []\n").is_err());
    }

    #[test]
    fn version_from_repo_default_and_explicit_paths() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();

        // Tauri app: default path reads src-tauri/tauri.conf.json.
        let tauri = root.join("tauri");
        std::fs::create_dir_all(tauri.join("src-tauri")).unwrap();
        std::fs::write(
            tauri.join("src-tauri/tauri.conf.json"),
            r#"{"version":"0.4.2"}"#,
        )
        .unwrap();
        assert_eq!(
            version_from_repo(tauri.to_str().unwrap(), None)
                .unwrap()
                .to_string(),
            "0.4.2"
        );

        // Workspace egui app: no tauri.conf.json, explicit version_path at a member crate.
        let ws = root.join("ws");
        std::fs::create_dir_all(ws.join("crates/app")).unwrap();
        std::fs::write(
            ws.join("Cargo.toml"),
            "[workspace]\nmembers = [\"crates/app\"]\n",
        )
        .unwrap();
        std::fs::write(
            ws.join("crates/app/Cargo.toml"),
            "[package]\nname = \"app\"\nversion = \"0.5.0\"\n",
        )
        .unwrap();
        assert_eq!(
            version_from_repo(ws.to_str().unwrap(), Some("crates/app/Cargo.toml"))
                .unwrap()
                .to_string(),
            "0.5.0"
        );
    }

    // ---- version-source cross-check (drift preflight) ----

    fn ver(s: &str) -> Version {
        Version::parse(s).unwrap()
    }

    #[test]
    fn version_consistency_passes_when_all_sources_agree() {
        let tmp = tempfile::tempdir().unwrap();
        let repo = tmp.path();
        std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
        std::fs::write(
            repo.join("src-tauri/tauri.conf.json"),
            r#"{"version":"0.5.0"}"#,
        )
        .unwrap();
        std::fs::write(
            repo.join("Cargo.toml"),
            "[package]\nname = \"app\"\nversion = \"0.5.0\"\n",
        )
        .unwrap();
        check_version_consistency(repo.to_str().unwrap(), None, &ver("0.5.0")).unwrap();
    }

    #[test]
    fn version_consistency_flags_tauri_vs_cargo_drift() {
        // The concrete finding: tauri.conf.json bumped to 0.5.0 but the root
        // Cargo.toml left at 0.4.0. version_from_repo (one file) would miss it.
        let tmp = tempfile::tempdir().unwrap();
        let repo = tmp.path();
        std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
        std::fs::write(
            repo.join("src-tauri/tauri.conf.json"),
            r#"{"version":"0.5.0"}"#,
        )
        .unwrap();
        std::fs::write(
            repo.join("Cargo.toml"),
            "[package]\nname = \"app\"\nversion = \"0.4.0\"\n",
        )
        .unwrap();
        let err =
            check_version_consistency(repo.to_str().unwrap(), None, &ver("0.5.0")).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("Cargo.toml says 0.4.0"), "{msg}");
    }

    #[test]
    fn version_consistency_flags_explicit_version_the_repo_does_not_reflect() {
        let tmp = tempfile::tempdir().unwrap();
        let repo = tmp.path();
        std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
        std::fs::write(
            repo.join("src-tauri/tauri.conf.json"),
            r#"{"version":"0.5.0"}"#,
        )
        .unwrap();
        let err =
            check_version_consistency(repo.to_str().unwrap(), None, &ver("9.9.9")).unwrap_err();
        assert!(format!("{err:#}").contains("building 9.9.9"));
    }

    #[test]
    fn version_consistency_single_source_never_invents_drift() {
        // A virtual-workspace root Cargo.toml (no version) alongside the member
        // crate the version_path points at: only one real source, so no drift.
        let tmp = tempfile::tempdir().unwrap();
        let repo = tmp.path();
        std::fs::create_dir_all(repo.join("crates/app")).unwrap();
        std::fs::write(
            repo.join("Cargo.toml"),
            "[workspace]\nmembers = [\"crates/app\"]\n",
        )
        .unwrap();
        std::fs::write(
            repo.join("crates/app/Cargo.toml"),
            "[package]\nname = \"app\"\nversion = \"0.5.0\"\n",
        )
        .unwrap();
        check_version_consistency(
            repo.to_str().unwrap(),
            Some("crates/app/Cargo.toml"),
            &ver("0.5.0"),
        )
        .unwrap();
    }

    // ---- artifact filename version assertion + hashing ----

    #[test]
    fn versions_in_filename_extracts_only_real_semvers() {
        assert_eq!(
            versions_in_filename("GoingsOn_0.5.0_aarch64.dmg"),
            vec![ver("0.5.0")]
        );
        assert_eq!(
            versions_in_filename("AudioFiles-0.4.0-x86_64.AppImage"),
            vec![ver("0.4.0")]
        );
        // No three-part token ⇒ nothing (an updater manifest, a bare signature).
        assert!(versions_in_filename("latest.json").is_empty());
        assert!(versions_in_filename("app.sig").is_empty());
    }

    #[test]
    fn assert_artifact_version_rejects_a_stale_artifact() {
        // The 0.4.0 file sitting in the output dir against a 0.5.0 build.
        let err =
            assert_artifact_version("AudioFiles-0.4.0-x86_64.AppImage", &ver("0.5.0")).unwrap_err();
        assert!(format!("{err:#}").contains("stale artifact"), "{err:#}");
        // The matching version passes, and a versionless file is not asserted.
        assert_artifact_version("GoingsOn_0.5.0_aarch64.dmg", &ver("0.5.0")).unwrap();
        assert_artifact_version("latest.json", &ver("0.5.0")).unwrap();
    }

    /// The comparison that decides whether a binary can exec on the box that is
    /// about to be restarted onto it. Both sides are parsed out of text a tool
    /// printed, so both parsers are worth pinning: fw13 tracks a newer glibc
    /// than the Ubuntu 24.04 host in Hetzner, and getting this backwards means a
    /// dead unit rather than a failed step.
    #[test]
    fn glibc_versions_parse_from_what_the_tools_actually_print() {
        // `objdump -T | grep -o 'GLIBC_[0-9.]*'` output: highest wins, and the
        // comparison is numeric (2.9 must not beat 2.34 lexically).
        let objdump = "GLIBC_2.2.5\nGLIBC_2.34\nGLIBC_2.9\nGLIBC_2.17\n";
        assert_eq!(max_glibc_symbol(objdump), Some((2, 34)));
        // A static binary references none: nothing to check.
        assert_eq!(max_glibc_symbol(""), None);

        // `ldd --version` first line, however the distro decorates it.
        assert_eq!(
            glibc_from_ldd("ldd (Ubuntu GLIBC 2.39-0ubuntu8.8) 2.39\nCopyright...\n"),
            Some((2, 39))
        );
        assert_eq!(
            glibc_from_ldd("ldd (GNU libc) 2.41\nCopyright (C) 2025\n"),
            Some((2, 41))
        );
        assert_eq!(glibc_from_ldd(""), None);
    }

    /// A binary needing MORE than the host has is the failure this check exists
    /// for; equal and less are both fine (glibc symbol versioning is backward
    /// compatible, so an older requirement runs on a newer host).
    #[test]
    fn glibc_requirement_is_satisfied_by_equal_or_newer_only() {
        let needs = max_glibc_symbol("GLIBC_2.41").unwrap();
        assert!(needs > glibc_from_ldd("ldd (Ubuntu GLIBC 2.39) 2.39").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 2.41").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 2.42").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 3.0").unwrap());
    }

    /// Every deploy host function fails with the app's KIND as the reason when
    /// there is no destination, rather than with a missing-host error from
    /// somewhere deeper. A recipe calling `deploy()` on a library is a recipe
    /// written against the wrong kind, and the message should say so.
    #[tokio::test]
    async fn deploy_host_fns_explain_a_missing_destination_by_kind() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Library,
            1,
            Arc::new(std::collections::HashMap::new()),
            Arc::new(std::collections::HashMap::new()),
            None,
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));
        let engine = build_engine(&ctx);
        for call in [
            "deploy_host()",
            "service_name()",
            "install_path()",
            "health_url()",
            r#"deploy("/tmp/x")"#,
        ] {
            let err = engine.eval::<String>(call).unwrap_err().to_string();
            assert!(
                err.contains("library") && err.contains("no deploy destination"),
                "`{call}` must fail on the kind, got: {err}"
            );
        }
    }

    /// A service host is addressed on the DEPLOY plane whatever step is open.
    ///
    /// The subtle one. Actions are normally derived from the step, which is
    /// right for a build host — the step is what that host is being asked to do.
    /// A service host is granted `deploy`/`restart` and must never be granted
    /// `build`, so the same rule would have `glibc_check` ask it for `build`
    /// during a `verify` step and get denied for a reason unrelated to what was
    /// attempted. `verify` is the step that check belongs in, so without this
    /// routing the glibc gate cannot run at all.
    #[tokio::test]
    async fn a_service_host_is_addressed_on_the_deploy_plane_in_any_step() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        sqlx::query(
            "INSERT INTO builds (id, app, version, status, created_at) \
             VALUES (1, 'demo', '0.1.0', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO target_runs (id, build_id, app, version, target, status, started_at) \
             VALUES (1, 1, 'demo', '0.1.0', 'linux/x86_64', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();

        let deploy = crate::topology::DeployTarget {
            target: "linux/x86_64".parse().unwrap(),
            host: "local".into(),
            port: None,
            install_path: "/usr/local/bin/demo".into(),
            service: "demo.service".into(),
            health_url: None,
        };
        let mut execs: crate::state::ExecutorMap = std::collections::HashMap::new();
        execs.insert("local".into(), crate::state::build_deploy_executor(&deploy));
        // The service host's grant is exactly deploy + restart. If this ever
        // widens to include `build`, the test below stops proving anything.
        assert!(!execs["local"].capabilities().permits(&Action::Build));
        assert!(execs["local"].capabilities().permits(&Action::Deploy));

        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Service,
            1,
            Arc::new(execs),
            Arc::new(std::collections::HashMap::new()),
            Some(deploy),
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));

        let ctx_blocking = ctx.clone();
        tokio::task::spawn_blocking(move || {
            // `verify` on a service derives Action::Build — which the service
            // host does not grant. The command must still run.
            ctx_blocking.begin_step(Step::Verify).unwrap();
            assert_eq!(
                action_for(Step::Verify, Kind::Service),
                Action::Build,
                "the step's own action is the one that would be denied",
            );
            let (code, out) = ctx_blocking
                .run("local", "echo reached-the-service-host")
                .expect("a service host must be reachable during a verify step");
            assert_eq!(code, 0, "{out}");
            assert!(out.contains("reached-the-service-host"), "{out}");
        })
        .await
        .unwrap();
    }

    /// A step that finalized `Failed` bars the deploy, exactly as it bars a
    /// publish. Without this, a recipe that inspects `sh(...).code` and carries
    /// on regardless still lands a binary on a production host — the precise
    /// hazard a pipeline exists to remove. The check is the ledger, not the
    /// control flow, so it holds whether or not the recipe noticed.
    #[tokio::test]
    async fn a_failed_step_bars_the_deploy() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let deploy = crate::topology::DeployTarget {
            target: "linux/x86_64".parse().unwrap(),
            host: "local".into(),
            port: None,
            install_path: "/usr/local/bin/demo".into(),
            service: "demo.service".into(),
            health_url: None,
        };
        // A real build + target run, so the step rows this test finalizes have
        // the parents the schema requires.
        sqlx::query(
            "INSERT INTO builds (id, app, version, status, created_at) \
             VALUES (1, 'demo', '0.1.0', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO target_runs (id, build_id, app, version, target, status, started_at) \
             VALUES (1, 1, 'demo', '0.1.0', 'linux/x86_64', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();

        let mut execs: crate::state::ExecutorMap = std::collections::HashMap::new();
        execs.insert("local".into(), crate::state::build_deploy_executor(&deploy));
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Service,
            1,
            Arc::new(execs),
            Arc::new(std::collections::HashMap::new()),
            Some(deploy),
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));

        // A gate ran, failed, and the recipe did not abort — the swallowed
        // failure. Finalizing it is what puts it in the ledger.
        let ctx_blocking = ctx.clone();
        tokio::task::spawn_blocking(move || {
            ctx_blocking.begin_step(Step::Prebuild).unwrap();
            ctx_blocking.fail_current_step();
            ctx_blocking.finish_step(Status::Ok).unwrap();

            let err = ctx_blocking.deploy("/tmp/demo").unwrap_err().to_string();
            assert!(
                err.contains("refusing to deploy") && err.contains("prebuild"),
                "must refuse and name the failed step, got: {err}"
            );
        })
        .await
        .unwrap();
    }

    #[test]
    fn every_step_has_a_nonzero_default_budget() {
        // A zero/missing budget would deadline-fail a step instantly. Cover the
        // whole matrix so a new Step variant can't silently get a 0 budget.
        for step in Step::ALL {
            assert!(
                default_step_budget(step) >= std::time::Duration::from_mins(1),
                "{step} budget must be a sane ceiling",
            );
        }
    }

    #[test]
    fn sha256_file_is_lowercase_hex_of_contents() {
        let tmp = tempfile::tempdir().unwrap();
        let f = tmp.path().join("a.bin");
        std::fs::write(&f, b"abc").unwrap();
        // Known SHA-256 of "abc".
        assert_eq!(
            sha256_file(&f).unwrap(),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
        );
    }

    // ---- publish step-success gate ----

    fn target(s: &str) -> Target {
        s.parse().unwrap()
    }

    #[test]
    fn publish_gate_blocks_macos_without_verification() {
        // Never verified -> blocked, with a message pointing at verify_gatekeeper.
        let err = PublishAuthority::prove(target("macos/aarch64"), &[], None).unwrap_err();
        assert!(format!("{err:#}").contains("never verified"), "{err:#}");
    }

    #[test]
    fn publish_gate_blocks_macos_when_gatekeeper_rejected() {
        let err = PublishAuthority::prove(target("macos/aarch64"), &[], Some(false)).unwrap_err();
        assert!(
            format!("{err:#}").contains("Gatekeeper rejected"),
            "{err:#}"
        );
    }

    #[test]
    fn publish_gate_allows_macos_when_gatekeeper_accepted() {
        PublishAuthority::prove(target("macos/aarch64"), &[], Some(true)).unwrap();
        // iOS is gated the same way.
        PublishAuthority::prove(target("ios/universal"), &[], Some(true)).unwrap();
        assert!(PublishAuthority::prove(target("ios/universal"), &[], None).is_err());
    }

    #[test]
    fn publish_gate_does_not_require_gatekeeper_for_non_apple_targets() {
        // Linux/Windows aren't notarized; no gatekeeper proof needed.
        PublishAuthority::prove(target("linux/x86_64"), &[], None).unwrap();
        PublishAuthority::prove(target("windows/x86_64"), &[], None).unwrap();
    }

    #[test]
    fn publish_gate_blocks_when_any_prior_step_failed() {
        // A failed step bars publish on every target, even a verified macOS one.
        let err =
            PublishAuthority::prove(target("linux/x86_64"), &[Step::Build], None).unwrap_err();
        assert!(
            format!("{err:#}").contains("prior step(s) failed"),
            "{err:#}"
        );
        assert!(
            format!("{err:#}").contains("build"),
            "names the failed step: {err:#}"
        );

        let err = PublishAuthority::prove(target("macos/aarch64"), &[Step::Sign], Some(true))
            .unwrap_err();
        assert!(
            format!("{err:#}").contains("prior step(s) failed"),
            "{err:#}"
        );
    }

    #[test]
    fn notary_accepted_parses_status_field() {
        assert!(notary_accepted(
            r#"{"id":"abc","status":"Accepted","message":"ok"}"#
        ));
        // Embedded in shell-sourcing noise: the object is isolated and parsed.
        assert!(notary_accepted(
            "sourcing env...\n{\n  \"status\": \"Accepted\"\n}\nbye"
        ));
        // Whitespace variant that a tight substring `"status":"Accepted"` misses.
        assert!(notary_accepted(r#"{ "status" : "Accepted" }"#));
    }

    #[test]
    fn notary_accepted_rejects_non_accepted_and_garbage() {
        assert!(!notary_accepted(r#"{"status":"Invalid"}"#));
        assert!(!notary_accepted(r#"{"status":"In Progress"}"#));
        assert!(!notary_accepted("no json here"));
        assert!(!notary_accepted("")); // empty / truncated -> fail closed
        // A truncated tail whose opening brace was cut off cannot parse -> closed.
        assert!(!notary_accepted(r#""status":"Accepted"}"#));
        // The literal appearing inside an error string must NOT pass as success.
        assert!(!notary_accepted(
            r#"{"status":"Invalid","message":"expected status:Accepted"}"#
        ));
    }

    /// Both shapes of repo: one holding several products, and one holding a
    /// single crate, where git prints an empty prefix line.
    #[test]
    fn toplevel_and_prefix_read_both_shapes_of_repo() {
        let (top, prefix) =
            parse_toplevel_and_prefix("/home/max/Code/MNW\npom/\n").expect("two lines");
        assert_eq!(top, "/home/max/Code/MNW");
        assert_eq!(prefix, "pom/");
        // A repo holding one product: git prints an empty second line.
        let (top, prefix) =
            parse_toplevel_and_prefix("/home/max/Code/Libraries/pter\n\n").expect("two lines");
        assert_eq!(top, "/home/max/Code/Libraries/pter");
        assert_eq!(prefix, "");
        assert!(
            parse_toplevel_and_prefix("").is_none(),
            "no answer is not an answer"
        );
    }

    #[test]
    fn repo_dir_name_is_the_last_segment_on_every_platform() {
        assert_eq!(repo_dir_name("/home/max/Code/MNW"), "MNW");
        assert_eq!(repo_dir_name("/home/max/Code/MNW/"), "MNW");
        // Git reports forward slashes on Windows too.
        assert_eq!(repo_dir_name("C:/Users/me/Code/Apps/goingson"), "goingson");
    }

    /// The app's directory inside its worktree, for both repo shapes.
    #[test]
    fn app_dir_in_worktree_follows_the_prefix() {
        assert_eq!(
            app_dir_in_worktree("/home/max/Code/.bento/MNW/pom", "pom/"),
            "/home/max/Code/.bento/MNW/pom/pom"
        );
        assert_eq!(
            app_dir_in_worktree("/home/max/Code/.bento/pter/pter", ""),
            "/home/max/Code/.bento/pter/pter"
        );
    }

    /// A missing tag is the common failure and gets a plain answer; anything
    /// else is git's own stderr, which says more than a guess.
    #[test]
    fn worktree_failure_reason_names_the_tag_or_repeats_git() {
        let missing = worktree_failure_reason("pom-v0.4.5", false, "irrelevant");
        assert!(missing.contains("does not exist"), "{missing}");
        let held = worktree_failure_reason(
            "pom-v0.4.5",
            true,
            "fatal: '/home/max/Code/.bento/MNW/pom' already exists",
        );
        assert!(held.contains("already exists"), "{held}");
        let silent = worktree_failure_reason("pom-v0.4.5", true, "  ");
        assert!(silent.contains("said nothing"), "{silent}");
    }

    /// Run the probe for real rather than asserting on its text: it is shell,
    /// and the thing worth knowing is whether `sh` agrees, not whether the
    /// string looks right.
    fn probe(repo: &std::path::Path) -> bool {
        std::process::Command::new("sh")
            .arg("-c")
            .arg(tracked_lock_under_patch_cmd(&repo.display().to_string()))
            .status()
            .unwrap()
            .success()
    }

    /// A crate under a `[patch]` root, with and without its lock tracked.
    ///
    /// pter 0.2.1 is the case: the tracked half was true, the patch half was
    /// true, and cargo reported only "Cargo.lock" (build 325). Untracking the
    /// lock in `a9969a9` is what fixed it, and this asserts the probe agrees
    /// with that fix in both directions.
    #[test]
    fn a_tracked_lock_is_only_a_problem_under_a_patch_block() {
        let root = tempfile::tempdir().unwrap();
        let repo = root.path().join("crate");
        std::fs::create_dir_all(&repo).unwrap();
        let git = |args: &[&str]| {
            std::process::Command::new("git")
                .args(args)
                .current_dir(&repo)
                .env("GIT_AUTHOR_NAME", "t")
                .env("GIT_AUTHOR_EMAIL", "t@t")
                .env("GIT_COMMITTER_NAME", "t")
                .env("GIT_COMMITTER_EMAIL", "t@t")
                .output()
                .unwrap()
        };
        git(&["init", "-q", "."]);
        std::fs::write(repo.join("Cargo.lock"), "# lock\n").unwrap();

        // Lock present but untracked, no patch anywhere: nothing to say.
        assert!(!probe(&repo));

        // Tracked, still no patch root. Every library that commits a lock and
        // builds outside `~/Code` lives here, and it publishes fine.
        git(&["add", "Cargo.lock"]);
        git(&["commit", "-qm", "lock"]);
        assert!(!probe(&repo));

        // The ancestor declares `[patch]`. Both halves now hold.
        std::fs::create_dir_all(root.path().join(".cargo")).unwrap();
        std::fs::write(
            root.path().join(".cargo/config.toml"),
            "[patch.\"https://makenot.work/git/max/docengine.git\"]\ndocengine = { path = \"x\" }\n",
        )
        .unwrap();
        assert!(probe(&repo));

        // Untracking the lock is the fix, and the probe has to agree that it is.
        git(&["rm", "-q", "--cached", "Cargo.lock"]);
        assert!(!probe(&repo));
    }

    /// The message exists because cargo's names neither fact and both wrong
    /// fixes are one flag away. Assert it still says all four things.
    #[test]
    fn the_tracked_lock_message_names_both_facts_and_refuses_both_wrong_fixes() {
        let msg = tracked_lock_under_patch_problem("~/Code/.bento/pter/pter");
        assert!(msg.contains("Cargo.lock"), "{msg}");
        assert!(msg.contains("[patch]"), "{msg}");
        assert!(msg.contains("git rm --cached"), "{msg}");
        assert!(msg.contains("--allow-dirty"), "{msg}");
        assert!(msg.contains("~/Code/.bento/pter/pter"), "{msg}");
    }
}