use crate::error::Result;
use crate::state::AppState;
use axum::extract::{Path, Query, State, WebSocketUpgrade};
use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use sqlx::Row;

mod promotion;
use promotion::{clear_partial, promote_inner, set_partial};

pub fn router(state: AppState) -> Router {
    let token = state.api_token.clone();

    // Every route requires the bearer token (when one is configured): the
    // mutators carry prod-deploy authority (CF2), and the reads expose prod
    // state — deployed versions, build SHAs, full gate logs, the live event
    // stream — that a tailnet peer should not get unauthenticated. There are no
    // unauthenticated routes: /metrics used to be the one opt-out, but the
    // Prometheus exporter was removed rather than leave an auth posture on it
    // that no scraper actually used. The bearer-success path logs the caller
    // identity for mutating (POST) requests so a prod ship is attributable.
    let protected = Router::new()
        // mutators — prod-deploy authority
        .route("/promote/{tier}", post(promote))
        .route("/rollback/{tier}", post(rollback))
        .route("/rebuild", post(rebuild))
        .route("/self-update", post(self_update))
        .route("/confirm/{tier}", post(confirm))
        .route("/backup/fetch", post(backup_fetch))
        // reads — prod state, now also gated
        .route("/state", get(get_state))
        .route("/status.json", get(get_status_json))
        .route("/runs/{id}", get(get_run))
        .route("/runs/{id}/wait", get(get_run_wait))
        .route("/logs/{version}/{gate}", get(get_gate_log))
        .route("/events", get(events_ws))
        .route_layer(axum::middleware::from_fn(move |req, next| {
            require_bearer(token.clone(), req, next)
        }));

    Router::new().merge(protected).with_state(state)
}

/// Bearer-token gate for the deploy mutators. When no token is configured the
/// request passes (main() only allows that on a loopback bind). Comparison is
/// constant-time to avoid leaking the token via timing.
async fn require_bearer(
    token: Option<std::sync::Arc<str>>,
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    let Some(expected) = token.as_deref() else {
        return next.run(req).await;
    };
    let path = req.uri().path().to_string();
    let method = req.method().clone();
    let ok = req
        .headers()
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|h| h.to_str().ok())
        .and_then(|h| h.strip_prefix("Bearer "))
        .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
    if ok {
        // Attribute mutations (POST) to a caller. Reads are GET and poll every
        // few seconds, so logging them would drown the journal; the audit value
        // is in *who shipped/rolled back prod*, which is always a POST. The peer
        // address is the closest identity we have without per-operator tokens
        // (`tailscale whois` is a future refinement); a shared token at least
        // gets a source host into the trail.
        if method == axum::http::Method::POST {
            let peer = req
                .extensions()
                .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
                .map_or_else(|| "unknown".into(), |ci| ci.0.to_string());
            tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation");
        }
        next.run(req).await
    } else {
        tracing::warn!(path = %path, "rejected unauthenticated request");
        (
            axum::http::StatusCode::UNAUTHORIZED,
            "missing or invalid bearer token\n",
        )
            .into_response()
    }
}

#[derive(Serialize)]
pub(crate) struct StateView {
    /// The running sandod's own package version. Lets a self-update caller
    /// confirm the new binary is live after the restart (the tier versions are
    /// the *deployed product*, not the controller).
    pub(crate) sandod_version: &'static str,
    pub(crate) tiers: Vec<TierView>,
    /// The most recent build run (the resource `GET /runs/{id}` exposes in
    /// full). Surfaced here so a `/state` poller sees an in-flight or failed
    /// build — the tier versions only ever reflect the last *success*, so
    /// without this `/state` looks frozen for the whole build. `null` until
    /// the first `/rebuild`.
    pub(crate) build: Option<crate::runs::BuildSummary>,
}

#[derive(Serialize)]
pub(crate) struct TierView {
    pub(crate) name: String,
    pub(crate) ord: i64,
    pub(crate) provisioned: bool,
    pub(crate) canary: String,
    pub(crate) current_version: Option<String>,
    pub(crate) previous_version: Option<String>,
    pub(crate) burn_in_started_at: Option<String>,
    /// Non-null when the tier was left in a partial / mixed-version state by a
    /// failed promote or rollback whose compensation could not fully restore
    /// consistency. NULL when the tier is consistent. The TUI flags it red.
    pub(crate) partial_reason: Option<String>,
    pub(crate) nodes: Vec<String>,
    pub(crate) gates: Vec<GateView>,
}

#[derive(Serialize)]
pub(crate) struct GateView {
    pub(crate) kind: String,
    pub(crate) finished_at: Option<String>,
    /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight. The TUI
    /// uses this to choose green/red/yellow rendering.
    pub(crate) status: Option<String>,
    /// Full typed `GateOutcome` as a JSON object, when present.
    /// Deserialized lazily by the consumer; sandod doesn't re-parse it.
    pub(crate) outcome: Option<serde_json::Value>,
    /// Relative path under `cfg.logs_root` to the persisted stdout/stderr.
    pub(crate) log_ref: Option<String>,
}

async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
    Ok(Json(state_view(&s).await?))
}

/// `GET /status.json` — the same state, in the shared cross-service payload
/// every operator surface renders. See `crate::status`.
async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
    let view = state_view(&s).await?;
    Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
}

/// The shared read behind `/state` and `/status.json`.
pub(crate) async fn state_view(s: &AppState) -> Result<StateView> {
    let rows = sqlx::query(
        "SELECT t.name, t.ord, t.provisioned, t.canary,
                ts.current_version, ts.previous_version, ts.burn_in_started_at,
                ts.partial_reason
         FROM tiers t
         LEFT JOIN tier_state ts ON ts.tier = t.name
         ORDER BY t.ord",
    )
    .fetch_all(&s.pool)
    .await?;

    let mut tiers: Vec<TierView> = Vec::with_capacity(rows.len());
    for r in rows {
        let name: String = r.get("name");
        let current_version: Option<String> = r.get("current_version");

        let nodes: Vec<String> =
            sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name")
                .bind(&name)
                .fetch_all(&s.pool)
                .await?;

        // Surface gates for current_version when set, otherwise for the most
        // recently attempted version on this tier. Without the fallback, a
        // tier that has never gone green (MM after a build failure, B before
        // first deploy) exposes no gate detail via /state — debugging required
        // SSH and direct SQLite access. See sando todo: gate observability.
        let gate_version: Option<String> = if current_version.is_some() {
            current_version.clone()
        } else {
            sqlx::query_scalar(
                "SELECT version FROM gate_runs WHERE tier = ?
                 ORDER BY id DESC LIMIT 1",
            )
            .bind(&name)
            .fetch_optional(&s.pool)
            .await?
        };

        let gates: Vec<GateView> = if let Some(ver) = gate_version.as_ref() {
            // Most recent gate_runs row per gate_kind for (tier, ver).
            sqlx::query(
                "SELECT gate_kind, finished_at, status, outcome_json, log_ref
                 FROM gate_runs g
                 WHERE tier = ?1 AND version = ?2
                   AND id = (SELECT MAX(id) FROM gate_runs
                             WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)
                 ORDER BY gate_kind",
            )
            .bind(&name)
            .bind(ver)
            .fetch_all(&s.pool)
            .await?
            .into_iter()
            .map(|gr| GateView {
                kind: gr.get("gate_kind"),
                finished_at: gr.get("finished_at"),
                status: gr.get("status"),
                outcome: gr
                    .get::<Option<String>, _>("outcome_json")
                    .and_then(|s| serde_json::from_str(&s).ok()),
                log_ref: gr.get("log_ref"),
            })
            .collect()
        } else {
            Vec::new()
        };

        tiers.push(TierView {
            name,
            ord: r.get("ord"),
            provisioned: r.get::<i64, _>("provisioned") != 0,
            canary: r.get("canary"),
            current_version,
            previous_version: r.get("previous_version"),
            burn_in_started_at: r.get("burn_in_started_at"),
            partial_reason: r.get("partial_reason"),
            nodes,
            gates,
        });
    }

    let build = crate::runs::latest_summary(&s.pool).await?;
    Ok(StateView {
        sandod_version: env!("CARGO_PKG_VERSION"),
        tiers,
        build,
    })
}

#[derive(Deserialize, Default)]
struct PromoteBody {
    /// Optional. If absent, defaults to the predecessor tier's `current_version`
    /// (i.e. promote whatever just finished baking on the previous tier).
    #[serde(default)]
    version: Option<String>,
    #[serde(default)]
    hotfix: bool,
    #[serde(default)]
    reset_burn_in: bool,
    /// Set when the release carries a database migration. Rollback restores the
    /// binary and release_contents only, never the schema (MNW migrates forward
    /// on boot and has no down path), so a migration-bearing promote is one-way.
    /// This forces a fresh `manual_confirm` on the predecessor tier even if the
    /// tier does not configure one, so the operator consciously acknowledges the
    /// rollback contract. `hotfix` does not suppress it. See
    /// `deploy/README.md` "Rollback contract".
    #[serde(default)]
    bears_migration: bool,
}

async fn promote(
    State(s): State<AppState>,
    Path(tier): Path<String>,
    body: Option<Json<PromoteBody>>,
) -> Result<Json<serde_json::Value>> {
    let body = body.map(|Json(b)| b).unwrap_or_default();
    // Run the deploy in a detached task, not inline in the request future.
    // A promote's rsync + restart can outlast a client's socket timeout; if the
    // client disconnects, axum drops the handler future — and if the deploy ran
    // inline that would abandon the rollout mid-flight (symlink half-swapped,
    // tier_state not advanced, and — before the kill_on_drop fix — an orphaned
    // rsync). Spawning detaches the work from the connection: the task owns the
    // deploy lock and runs to completion regardless, while a still-connected
    // client still receives the result. Mirrors how /rebuild already runs its
    // build off-request.
    tokio::spawn(promote_inner(s, tier, body))
        .await
        .map_err(|e| {
            crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}"))
        })?
}

async fn rollback(
    State(s): State<AppState>,
    Path(tier): Path<String>,
) -> Result<Json<serde_json::Value>> {
    // Serialize against any concurrent promote/rollback (CF3).
    let _deploy_guard = s.deploy_lock.lock().await;
    let tier = crate::domain::TierId::new(tier);
    let target = s
        .topo
        .tiers
        .iter()
        .find(|t| t.name == tier)
        .ok_or(crate::error::Error::NotFound)?;

    let row: Option<(Option<String>, Option<String>)> =
        sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
            .bind(&tier)
            .fetch_optional(&s.pool)
            .await
            .map_err(crate::error::Error::Db)?;
    let (Some(current_str), Some(previous_str)) = row.unwrap_or((None, None)) else {
        return Err(crate::error::Error::GateBlocked(
            "no previous_version to roll back to".into(),
        ));
    };
    let current = crate::domain::Version::parse(&current_str)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
    let previous = crate::domain::Version::parse(&previous_str)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;

    let bin: Option<(String,)> =
        sqlx::query_as("SELECT artifact_path FROM versions WHERE version = ?")
            .bind(&previous)
            .fetch_optional(&s.pool)
            .await
            .map_err(crate::error::Error::Db)?;
    let Some((bin,)) = bin else {
        return Err(crate::error::Error::GateBlocked(format!(
            "previous version {previous} has no artifact_path; rollback impossible"
        )));
    };
    let bin_path = std::path::PathBuf::from(bin);
    let staged_dir = bin_path
        .parent()
        .ok_or_else(|| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))?
        .to_path_buf();

    // Deploy `previous` to every node. A failure partway through leaves the fleet
    // split-brain (some on `previous`, some still on `current`); record exactly how
    // far we got so /state surfaces it rather than reporting a clean rollback.
    let total = target.nodes.len();
    for (i, node) in target.nodes.iter().enumerate() {
        let executor = s
            .executors
            .get(&node.name)
            .cloned()
            .unwrap_or_else(|| crate::state::build_executor(node));
        if let Err(e) = crate::deploy::deploy_node(
            executor.as_ref(),
            node,
            &previous_str,
            &staged_dir,
            s.cfg.primary_bin(),
        )
        .await
        {
            set_partial(
                &s,
                &tier,
                &format!(
                    "rollback to {previous} incomplete: {i}/{total} nodes rolled back, \
                 node {} failed; remaining nodes still on {current} — manual check needed",
                    node.name,
                ),
            )
            .await;
            tracing::error!(tier = %tier, node = %node.name, done = i, of = total,
                to = %previous, "rollback failed mid-fleet; tier left partial");
            return Err(crate::error::Error::Other(e));
        }
    }

    // `previous_version` becomes NULL, not `current`. Sando tracks exactly one
    // step of history, so writing the version we just rolled *off* into the
    // rollback slot makes a second /rollback roll FORWARD onto the build the
    // operator was escaping. NULL makes the second call fail loudly with "no
    // previous_version to roll back to" — honest about the depth we keep.
    // Stepping back further is the deploys-table walk we deliberately don't do.
    //
    // `advanced_at = now` even though burn_in_started_at is nulled: a rollback IS
    // a change to the deployed identity, so the startup reconcile must treat the
    // roll-off `deploys` row as predating this state — otherwise a cleanly
    // rolled-back tier would look like an unrecorded deploy (migration 009 /
    // [`crate::reconcile`]).
    let now = chrono::Utc::now().to_rfc3339();
    sqlx::query(
        "UPDATE tier_state
         SET current_version = ?, previous_version = NULL,
             burn_in_started_at = NULL, advanced_at = ?
         WHERE tier = ?",
    )
    .bind(&previous)
    .bind(&now)
    .bind(&tier)
    .execute(&s.pool)
    .await
    .map_err(crate::error::Error::Db)?;

    // Full rollback succeeded on every node: the tier is consistent again.
    clear_partial(&s, &tier).await;

    tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");
    crate::events::emit(
        &s.events,
        crate::events::Event::Rollback {
            tier: tier.clone(),
            from: current.clone(),
            to: previous.clone(),
        },
    );

    Ok(Json(serde_json::json!({
        "tier": tier,
        "rolled_back_from": current,
        "now_running": previous,
    })))
}

#[derive(Deserialize, Default)]
struct RebuildBody {
    /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
    #[serde(default)]
    sha: Option<String>,
}

async fn rebuild(
    State(s): State<AppState>,
    body: Option<Json<RebuildBody>>,
) -> Result<Json<serde_json::Value>> {
    let body = body.map(|Json(b)| b).unwrap_or_default();
    let sha = match body.sha {
        Some(sha) => sha,
        None => {
            // Omitted sha = "build the deploy branch's tip". Fetch upstream
            // first so we resolve the *upstream* HEAD, not a possibly-stale
            // local branch ref — the build task fetches too, but only after the
            // sha is already chosen, so without this `/rebuild {}` could build
            // an old commit. A fetch failure is non-fatal: fall back to the
            // current bare-repo tip (same policy as the build task).
            let bare = std::path::Path::new(&s.topo.repo.bare_path);
            if let Some(upstream) = s.topo.repo.upstream.as_deref()
                && let Err(e) =
                    crate::git::fetch_upstream(bare, upstream, &s.topo.repo.branch).await
            {
                tracing::warn!(error = %e, "pre-resolve upstream fetch failed; resolving current bare-repo branch tip");
            }
            crate::git::resolve_ref(bare, &s.topo.repo.branch)
                .await
                .map_err(crate::error::Error::Other)?
        }
    };

    // Boundary parse: a sha entering Sando must be hex of plausible length.
    // The build pipeline downstream only ever sees `GitSha`.
    let sha = crate::domain::GitSha::parse(&sha)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;

    tracing::info!(sha = %sha, "rebuild requested");
    crate::events::emit(
        &s.events,
        crate::events::Event::RebuildRequested { sha: sha.clone() },
    );

    // One pollable resource per triggered build. Created before the spawn so
    // the run id is in the response even if the task is aborted milliseconds
    // later by a still-newer /rebuild.
    let run_id = crate::runs::create(&s.pool, sha.as_str())
        .await
        .map_err(crate::error::Error::Other)?;

    // Latest /rebuild wins: abort any in-flight build before spawning a new
    // one. Aborting drops the spawned task's future, which drops any
    // tokio::process::Child it owns; with `kill_on_drop(true)` set on the
    // cargo Command, SIGKILL propagates to cargo + its rustc children.
    let mut slot = s.active_build.lock().await;
    if let Some(prev) = slot.take()
        && !prev.handle.is_finished()
    {
        tracing::warn!("aborting in-flight build for newer /rebuild request");
        crate::events::emit(
            &s.events,
            crate::events::Event::BuildAborted {
                sha_aborted: sha.clone(),
            },
        );
        prev.handle.abort();
        // Aborting drops the task before it can settle its own row, so
        // record the supersession here.
        crate::runs::mark_aborted(&s.pool, prev.run_id).await.ok();
    }

    let pool = s.pool.clone();
    let cfg = s.cfg.clone();
    let topo = s.topo.clone();
    let events_for_task = s.events.clone();
    let sha_for_task = sha.clone();
    let sha_response = sha.to_string();
    let pool_for_task = s.pool.clone();
    let deploy_lock = s.deploy_lock.clone();
    let handle = tokio::spawn(async move {
        if let Err(e) = crate::build::build_and_run_host(
            pool,
            cfg,
            topo,
            sha_for_task.clone(),
            events_for_task,
            run_id,
            deploy_lock,
        )
        .await
        {
            tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
            // Pre-gate bails (fetch/checkout/version/scratch) don't settle the
            // run themselves; the build-step compile error already did. First
            // terminal write wins, so this is a safety net for the rest.
            crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
                .await
                .ok();
        }
    });
    *slot = Some(crate::state::ActiveBuild {
        handle: handle.abort_handle(),
        run_id,
    });

    Ok(Json(
        serde_json::json!({ "accepted": true, "sha": sha_response, "run_id": run_id.0 }),
    ))
}

/// `GET /runs/{id}` — the build-status resource a non-TUI driver polls after
/// `/rebuild`. Open (read-only) like `/state` and `/logs`.
async fn get_run(
    State(s): State<AppState>,
    Path(id): Path<i64>,
) -> Result<Json<crate::runs::RunView>> {
    crate::runs::get(&s.pool, crate::domain::RunId(id))
        .await
        .map_err(crate::error::Error::Other)?
        .map(Json)
        .ok_or(crate::error::Error::NotFound)
}

#[derive(Deserialize)]
struct WaitParams {
    /// How long to hold the request open before returning a still-building
    /// run. Default 30s, capped at 120s.
    #[serde(default)]
    timeout_ms: Option<u64>,
}

/// `GET /runs/{id}/wait` — long-poll: hold the request open until the run
/// settles (`result != building`) or the timeout elapses, then return the
/// current `RunView`. Removes polling-cadence guessing for a headless driver
/// (fire `/rebuild`, block on `/wait`). On timeout the run is returned
/// still-building (200) and the caller re-issues `/wait`. 404 if unknown.
async fn get_run_wait(
    State(s): State<AppState>,
    Path(id): Path<i64>,
    Query(p): Query<WaitParams>,
) -> Result<Json<crate::runs::RunView>> {
    let run_id = crate::domain::RunId(id);
    let timeout = std::time::Duration::from_millis(p.timeout_ms.unwrap_or(30_000).min(120_000));
    let deadline = tokio::time::Instant::now() + timeout;
    // Poll the row rather than wiring a per-run notifier: a build settles on
    // the minute scale, so a sub-second tick is plenty responsive and the
    // query is a single indexed read. The request releases its pool handle
    // between ticks.
    let tick = std::time::Duration::from_millis(750);
    loop {
        let view = crate::runs::get(&s.pool, run_id)
            .await
            .map_err(crate::error::Error::Other)?
            .ok_or(crate::error::Error::NotFound)?;
        let now = tokio::time::Instant::now();
        if view.result != "building" || now >= deadline {
            return Ok(Json(view));
        }
        tokio::time::sleep((deadline - now).min(tick)).await;
    }
}

#[derive(Deserialize)]
struct SelfUpdateBody {
    /// The commit to rebuild sandod from. Must already be on the canonical
    /// remote (the updater `git fetch`es it).
    sha: String,
}

/// The privileged updater unit instance for `sha`. A git sha is hex-only, so it
/// is a safe systemd instance name with no escaping needed.
fn self_update_unit(sha: &crate::domain::GitSha) -> String {
    format!("sando-update@{sha}.service")
}

/// Trigger a rebuild + restart of sandod *itself* to `sha`. sandod runs
/// unprivileged (User=sando, NoNewPrivileges, ProtectSystem=strict) and cannot
/// write `/usr/local/bin/sandod` or restart its own service — so it only
/// *triggers* the root oneshot `sando-update@<sha>.service` (which the sando
/// user is authorized to start by a scoped polkit rule). That unit builds
/// `sando/daemon` as the sando user in a dedicated checkout, installs the new
/// binary, and restarts sandod. Bearer-gated like the other mutators; the new
/// version shows up in `/state`'s `sandod_version` once the restart lands.
async fn self_update(
    State(s): State<AppState>,
    crate::error::TypedBody(body): crate::error::TypedBody<SelfUpdateBody>,
) -> Result<Json<serde_json::Value>> {
    let sha = crate::domain::GitSha::parse(&body.sha)
        .map_err(|e| crate::error::Error::BadRequest(format!("invalid sha: {e}")))?;

    let unit = self_update_unit(&sha);

    // Don't restart the controller out from under an in-flight server build — the
    // restart would SIGKILL it mid-deploy. Hold the active_build slot across the
    // trigger, not just the check: releasing it before triggering let a /rebuild
    // claim the slot in the gap and start a build the updater then killed (the
    // Run 2 TOCTOU). With the guard held, no build can start until the updater is
    // enqueued.
    {
        let slot = s.active_build.lock().await;
        if slot.as_ref().is_some_and(|b| !b.handle.is_finished()) {
            return Err(crate::error::Error::GateBlocked(
                "a server build is in flight; retry /self-update once it settles".into(),
            ));
        }
        // Also refuse during an in-flight promote/rollback: the updater's restart
        // would SIGKILL the daemon mid-deploy. try_lock (not lock) so we reject
        // rather than queue behind a long deploy; the guard is dropped immediately
        // — it only needs to observe that no deploy holds it right now.
        if s.deploy_lock.try_lock().is_err() {
            return Err(crate::error::Error::GateBlocked(
                "a promote/rollback is in flight; retry /self-update once it settles".into(),
            ));
        }

        tracing::warn!(sha = %sha, unit = %unit, "self-update requested; triggering privileged updater");
        // `--no-block`: return as soon as the job is enqueued. The build+restart
        // outcome lands in `journalctl -u <unit>`; sandod is restarted out from
        // under this request, so there is nothing more to await here.
        let status = tokio::process::Command::new("systemctl")
            .args(["start", "--no-block", &unit])
            .status()
            .await
            .map_err(|e| crate::error::Error::Other(anyhow::anyhow!("spawning systemctl: {e}")))?;
        if !status.success() {
            return Err(crate::error::Error::Other(anyhow::anyhow!(
                "systemctl start {unit} exited {status}; is sando-update@.service installed and the sando-user polkit rule in place?"
            )));
        }
    }
    Ok(Json(
        serde_json::json!({ "accepted": true, "sha": sha.to_string(), "unit": unit }),
    ))
}

async fn confirm(
    State(s): State<AppState>,
    Path(tier): Path<String>,
) -> Result<Json<serde_json::Value>> {
    // Operator-driven satisfaction of a `manual_confirm` gate. Looks up the
    // pending version (current MM version, or the tier's own if non-mm) and
    // inserts a passing gate_runs row so /promote can advance.
    //
    // Hold deploy_lock so the version we read and the row we insert reference the
    // same tier state — a concurrent rollback can't change current_version between
    // the read and the insert (ultra-fuzz Run 2, F3).
    let _deploy_guard = s.deploy_lock.lock().await;
    let tier = crate::domain::TierId::new(tier);
    let target = s
        .topo
        .tiers
        .iter()
        .find(|t| t.name == tier)
        .ok_or(crate::error::Error::NotFound)?;

    let version_str: Option<String> =
        sqlx::query_scalar("SELECT current_version FROM tier_state WHERE tier = ?")
            .bind(&target.name)
            .fetch_optional(&s.pool)
            .await
            .map_err(crate::error::Error::Db)?
            .flatten();
    let version_str = version_str.ok_or_else(|| {
        crate::error::Error::GateBlocked(format!(
            "tier {tier} has no current_version; nothing to confirm"
        ))
    })?;
    let version = crate::domain::Version::parse(&version_str)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;

    let now = chrono::Utc::now().to_rfc3339();
    let outcome =
        crate::outcome::GateOutcome::passed(crate::outcome::PassNote::OperatorConfirmed {
            at: chrono::Utc::now(),
        });
    let outcome_json = serde_json::to_string(&outcome)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
    sqlx::query(
        "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, outcome_json)
         VALUES (?, ?, 'manual_confirm', ?, ?, 'passed', ?)",
    )
    .bind(&version).bind(&target.name).bind(&now).bind(&now).bind(&outcome_json)
    .execute(&s.pool).await.map_err(crate::error::Error::Db)?;

    tracing::info!(tier = %tier, version = %version, "manual_confirm recorded");
    crate::events::emit(
        &s.events,
        crate::events::Event::ManualConfirm {
            tier: tier.clone(),
            version: version.clone(),
        },
    );

    Ok(Json(
        serde_json::json!({ "tier": tier, "version": version }),
    ))
}

async fn backup_fetch(State(s): State<AppState>) -> Result<Json<serde_json::Value>> {
    let fb = crate::backup::fetch(&s.pool, &s.cfg, &s.topo)
        .await
        .map_err(crate::error::Error::Other)?;
    crate::events::emit(
        &s.events,
        crate::events::Event::BackupFetched {
            source: fb.source.clone(),
            byte_size: fb.byte_size.unwrap_or(0),
        },
    );
    Ok(Json(serde_json::json!({
        "source": fb.source,
        "local_path": fb.local_path,
        "byte_size": fb.byte_size,
    })))
}

async fn get_gate_log(
    State(s): State<AppState>,
    Path((version, gate)): Path<(String, String)>,
) -> Result<axum::response::Response> {
    // Guard against `..` / absolute paths — the version segment must be a single
    // safe component. Without this, `GET /logs/..%2Fetc/passwd` would escape
    // logs_root.
    fn safe(seg: &str) -> bool {
        !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
    }
    if !safe(&version) {
        return Err(crate::error::Error::NotFound);
    }
    // The gate segment is an allowlisted kind, not a free-form filename: parse it
    // to `GateKind` so only the five known `<kind>.log` files are reachable. This
    // closes `*.log` filename probing within a version dir (traversal was already
    // blocked; this bounds the basename to the known set).
    let kind: crate::domain::GateKind = gate.parse().map_err(|_| crate::error::Error::NotFound)?;
    let path = s
        .cfg
        .logs_root
        .join(&version)
        .join(format!("{}.log", kind.as_str()));
    // Bound how much a single request can pull into the daemon's memory: a
    // runaway gate log shouldn't be read whole. Past the cap we return the tail
    // (the recent, relevant output) behind a truncation marker.
    const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024;
    use tokio::io::{AsyncReadExt, AsyncSeekExt};
    let mut file = match tokio::fs::File::open(&path).await {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            return Err(crate::error::Error::NotFound);
        }
        Err(e) => return Err(crate::error::Error::Other(e.into())),
    };
    let len = file
        .metadata()
        .await
        .map_err(|e| crate::error::Error::Other(e.into()))?
        .len();
    let body: Vec<u8> = if len <= MAX_LOG_BYTES {
        let mut buf = Vec::with_capacity(len as usize);
        file.read_to_end(&mut buf)
            .await
            .map_err(|e| crate::error::Error::Other(e.into()))?;
        buf
    } else {
        file.seek(std::io::SeekFrom::End(-(MAX_LOG_BYTES as i64)))
            .await
            .map_err(|e| crate::error::Error::Other(e.into()))?;
        let mut buf = format!(
            "[log truncated: showing the last {} MiB of {len} bytes]\n",
            MAX_LOG_BYTES / 1024 / 1024,
        )
        .into_bytes();
        file.read_to_end(&mut buf)
            .await
            .map_err(|e| crate::error::Error::Other(e.into()))?;
        buf
    };
    Ok((
        [(
            axum::http::header::CONTENT_TYPE,
            "text/plain; charset=utf-8",
        )],
        body,
    )
        .into_response())
}

async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
    use axum::extract::ws::Message;
    use tokio::sync::broadcast::error::RecvError;

    ws.on_upgrade(move |mut socket| async move {
        // Subscribe to both channels and merge them: a lag on the high-rate log
        // stream emits its own `lagged` frame without dropping anything on the
        // status stream (and vice versa), so a busy gate's chunk firehose can't
        // evict a PromoteComplete/GateDone the operator needs to see.
        let mut status_rx = s.events.subscribe_status();
        let mut logs_rx = s.events.subscribe_logs();
        loop {
            let recv = tokio::select! {
                r = status_rx.recv() => r,
                r = logs_rx.recv() => r,
            };
            match recv {
                Ok(env) => {
                    let json = match serde_json::to_string(&env) {
                        Ok(s) => s,
                        Err(e) => {
                            tracing::warn!(error = %e, "events ws: serialize failed");
                            continue;
                        }
                    };
                    if socket.send(Message::Text(json.into())).await.is_err() {
                        break;
                    }
                }
                Err(RecvError::Lagged(n)) => {
                    let _ = socket
                        .send(Message::Text(
                            format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
                        ))
                        .await;
                }
                Err(RecvError::Closed) => break,
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::promotion::{rollback_deployed_nodes, unsatisfied_gates};
    use super::*;
    use crate::config::Config;
    use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
    use async_trait::async_trait;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use http_body_util::BodyExt;
    use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts};
    use sqlx::SqlitePool;
    use sqlx::sqlite::SqlitePoolOptions;
    use std::os::unix::process::ExitStatusExt;
    use std::path::PathBuf;
    use std::sync::{Arc, Mutex as StdMutex};
    use tower::ServiceExt;

    async fn fresh_pool() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        pool
    }

    /// Two-tier topology used by the route tests: mm (provisioned, no nodes)
    /// → a (provisioned, one local node). Mirrors the production shape
    /// without involving real ssh / postgres.
    fn test_topo() -> Topology {
        Topology {
            repo: RepoConfig {
                bare_path: "/tmp/test.git".into(),
                branch: "main".into(),
                upstream: None,
            },
            backup: BackupConfig {
                source: "file:///tmp/test-backup.sql".into(),
                local_path: "/tmp/local-backup.sql".into(),
            },
            tiers: vec![
                Tier {
                    name: "host".into(),
                    provisioned: true,
                    gates: vec![],
                    canary: CanaryPolicy::Sequential,
                    nodes: vec![],
                },
                Tier {
                    name: "a".into(),
                    provisioned: true,
                    gates: vec![Gate::BootSmoke],
                    canary: CanaryPolicy::Sequential,
                    nodes: vec![Node {
                        name: "a-local".into(),
                        ssh_target: "local".into(),
                        release_root: "/tmp/a-node".into(),
                        service_name: "makenotwork.service".into(),
                        health_url: None,
                        config_check_env_file: None,
                        actuate: crate::topology::default_actuate(),
                        observe: crate::topology::default_observe(),
                        companions: Vec::new(),
                    }],
                },
            ],
        }
    }

    fn test_cfg() -> Config {
        Config {
            listen: "127.0.0.1:0".into(),
            db_path: PathBuf::from(":memory:"),
            topology_path: PathBuf::from("/tmp/test-sando.toml"),
            build_host: "test-host".into(),
            workdir: PathBuf::from("/tmp/sando-work"),
            release_root: PathBuf::from("/tmp/sando-releases"),
            scratch_db_url: None,
            scratch_owner_role: "makenotwork".into(),
            boot_smoke_port: 18181,
            code_smoke_port: 18182,
            bin_names: vec!["makenotwork".into()],
            logs_root: PathBuf::from("/tmp/sando-logs"),
            release_contents: vec![],
            cargo_target_dir: None,
            gate_timeout_secs: 2400,
            companions: Vec::new(),
            test_targets: vec![crate::config::TestTarget {
                dir: PathBuf::from("server"),
                features: vec!["fast-tests".into()],
                all_features: false,
                scratch_db: true,
            }],
        }
    }

    async fn test_state() -> AppState {
        let pool = fresh_pool().await;
        // Seed tier rows so FKs on tier_state / gate_runs are satisfied.
        for (i, name) in ["host", "a"].iter().enumerate() {
            sqlx::query(
                "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
            )
            .bind(name)
            .bind(i as i64)
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
                .bind(name)
                .execute(&pool)
                .await
                .unwrap();
        }
        // Don't call install_recorder in tests — it touches a process-global
        // and conflicts when tests run in parallel.
        let topo = test_topo();
        let executors = Arc::new(crate::state::build_executors(&topo));
        AppState {
            pool,
            topo: Arc::new(topo),
            cfg: Arc::new(test_cfg()),
            active_build: Arc::new(tokio::sync::Mutex::new(None)),
            deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
            events: crate::events::channel(),
            executors,
            api_token: None,
        }
    }

    async fn body_string(resp: axum::response::Response) -> String {
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        String::from_utf8(bytes.to_vec()).unwrap()
    }

    /// Insert the FK prerequisites for inserting gate_runs/tier_state rows.
    async fn seed(pool: &SqlitePool, tier: &str, version: &str) {
        sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING")
            .bind(tier).execute(pool).await.unwrap();
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
            .bind(version).execute(pool).await.unwrap();
        sqlx::query("INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING")
            .bind(tier).execute(pool).await.unwrap();
    }

    async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) {
        let status = if passed == 1 { "passed" } else { "failed" };
        sqlx::query(
            "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
             VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)",
        )
        .bind(version)
        .bind(tier)
        .bind(kind)
        .bind(status)
        .execute(pool)
        .await
        .unwrap();
    }

    // ---- unsatisfied_gates ----

    fn tid(s: &str) -> crate::domain::TierId {
        crate::domain::TierId::new(s)
    }

    #[tokio::test]
    async fn unsatisfied_gates_empty_when_no_configured_gates() {
        // A tier that configures no gates has nothing to satisfy.
        let pool = fresh_pool().await;
        seed(&pool, "host", "0.8.12").await;
        let pending = unsatisfied_gates(&pool, &tid("host"), &[], "0.8.12", None, false)
            .await
            .unwrap();
        assert_eq!(pending, Vec::<String>::new());
    }

    #[tokio::test]
    async fn unsatisfied_gates_flags_configured_gate_that_never_ran() {
        // THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied
        // (fail closed), NOT silently treated as green. Before this, an A tier
        // whose boot_smoke never executed exposed zero rows and waved promotion
        // straight through to prod.
        let pool = fresh_pool().await;
        seed(&pool, "a", "0.8.12").await;
        let pending =
            unsatisfied_gates(&pool, &tid("a"), &[Gate::BootSmoke], "0.8.12", None, false)
                .await
                .unwrap();
        assert_eq!(pending, vec!["boot_smoke".to_string()]);
    }

    #[tokio::test]
    async fn unsatisfied_gates_flags_failed_kind() {
        let pool = fresh_pool().await;
        seed(&pool, "host", "0.8.12").await;
        insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
        insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await;
        let pending = unsatisfied_gates(
            &pool,
            &tid("host"),
            &[Gate::CargoTest, Gate::BootSmoke],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert_eq!(pending, vec!["cargo_test".to_string()]);
    }

    #[tokio::test]
    async fn unsatisfied_gates_latest_row_wins() {
        // Two runs of the same gate; only the latest counts. A flap from
        // red to green should clear the pending entry.
        let pool = fresh_pool().await;
        seed(&pool, "host", "0.8.12").await;
        insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
        insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await;
        let pending = unsatisfied_gates(
            &pool,
            &tid("host"),
            &[Gate::CargoTest],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert!(pending.is_empty());
    }

    async fn insert_confirm(
        pool: &SqlitePool,
        tier: &str,
        version: &str,
        at: chrono::DateTime<chrono::Utc>,
    ) {
        sqlx::query(
            "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
             VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')",
        )
        .bind(version)
        .bind(tier)
        .bind(at.to_rfc3339())
        .bind(at.to_rfc3339())
        .execute(pool)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() {
        // A confirmation only satisfies the gate if it post-dates the version's
        // current landing on the tier (burn_in_started_at). A stale confirm left
        // over from before a rollback + rollback-forward must NOT wave it through.
        let pool = fresh_pool().await;
        seed(&pool, "a", "0.8.12").await;
        let landed = chrono::Utc::now();
        sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
            .bind(landed.to_rfc3339())
            .execute(&pool)
            .await
            .unwrap();

        // Stale confirmation (recorded before this landing) -> unsatisfied.
        insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await;
        let pending = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::ManualConfirm],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert_eq!(
            pending,
            vec!["manual_confirm".to_string()],
            "stale confirm must not satisfy"
        );

        // Fresh confirmation (after this landing) -> satisfied.
        insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await;
        let pending = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::ManualConfirm],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert!(pending.is_empty(), "fresh confirm satisfies");
    }

    #[tokio::test]
    async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() {
        // No landing clock (burn_in_started_at NULL) -> a passed confirm row is
        // not provably fresh, so fail closed and require a new confirmation.
        let pool = fresh_pool().await;
        seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL
        insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await;
        let pending = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::ManualConfirm],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert_eq!(
            pending,
            vec!["manual_confirm".to_string()],
            "no baseline -> fail closed"
        );
    }

    #[tokio::test]
    async fn unsatisfied_gates_hotfix_skips_only_burn_in() {
        // burn_in is evaluated live (no clock started -> not elapsed); cargo_test
        // has a failing row. Normal: both unsatisfied, in configured order.
        // hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic
        // so a future change doesn't widen the hotfix bypass.
        let pool = fresh_pool().await;
        seed(&pool, "a", "0.8.12").await;
        insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
        let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest];

        let normal = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, false)
            .await
            .unwrap();
        assert_eq!(
            normal,
            vec!["burn_in".to_string(), "cargo_test".to_string()]
        );

        let with_hotfix = unsatisfied_gates(&pool, &tid("a"), &gates, "0.8.12", None, true)
            .await
            .unwrap();
        assert_eq!(with_hotfix, vec!["cargo_test".to_string()]);
    }

    #[tokio::test]
    async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() {
        // A burn-in clock started far enough in the past satisfies the gate
        // live — no gate_runs row needed.
        let pool = fresh_pool().await;
        seed(&pool, "a", "0.8.12").await;
        sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
            .bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339())
            .execute(&pool)
            .await
            .unwrap();
        let pending = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::BurnIn { hours: 48 }],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in");
    }

    #[tokio::test]
    async fn unsatisfied_gates_ignores_other_tiers_and_versions() {
        let pool = fresh_pool().await;
        seed(&pool, "host", "0.8.12").await;
        seed(&pool, "host", "0.8.11").await;
        seed(&pool, "a", "0.8.12").await;
        // Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions
        // shouldn't pollute the query.
        insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
        insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
        insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await;

        let pending = unsatisfied_gates(
            &pool,
            &tid("host"),
            &[Gate::CargoTest],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert_eq!(pending, vec!["cargo_test".to_string()]);
    }

    #[tokio::test]
    async fn unsatisfied_gates_null_status_is_treated_as_failing() {
        // An in-flight gate (started_at set, finished_at + status NULL)
        // should NOT be treated as green. Otherwise a race could promote
        // before the gate concludes.
        let pool = fresh_pool().await;
        seed(&pool, "host", "0.8.12").await;
        sqlx::query(
            "INSERT INTO gate_runs (version, tier, gate_kind, started_at) \
             VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))",
        )
        .execute(&pool)
        .await
        .unwrap();

        let pending = unsatisfied_gates(
            &pool,
            &tid("host"),
            &[Gate::CargoTest],
            "0.8.12",
            None,
            false,
        )
        .await
        .unwrap();
        assert_eq!(pending, vec!["cargo_test".to_string()]);
    }

    // ---- /confirm/{tier} ----

    #[tokio::test]
    async fn confirm_rejects_when_tier_has_no_current_version() {
        // tier_state.a.current_version is NULL by default. /confirm has
        // nothing to confirm against → GateBlocked (400).
        let state = test_state().await;
        let app = router(state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/confirm/a")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body = body_string(resp).await;
        assert!(body.contains("no current_version"), "got: {body}");
    }

    #[tokio::test]
    async fn confirm_accepts_when_current_version_set_and_inserts_row() {
        let state = test_state().await;
        // Seed a version + advance tier a's state to it.
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
            .execute(&state.pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'")
            .execute(&state.pool)
            .await
            .unwrap();

        let app = router(state.clone());
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/confirm/a")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = body_string(resp).await;
        assert!(body.contains("\"tier\":\"a\""));
        assert!(body.contains("\"version\":\"0.8.12\""));

        // A passing gate_runs row was inserted.
        let count: (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'",
        )
        .fetch_one(&state.pool)
        .await
        .unwrap();
        assert_eq!(count.0, 1);
    }

    #[tokio::test]
    async fn confirm_404s_for_unknown_tier() {
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/confirm/zzzz")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn get_run_404s_for_unknown_id() {
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/runs/999")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn get_run_returns_view_with_gates() {
        let state = test_state().await;
        // A run that reached version 0.10.2 and ran two host gates (one red).
        let run_id = crate::runs::create(&state.pool, "abc1234def")
            .await
            .unwrap();
        let ver: crate::domain::Version = "0.10.2".parse().unwrap();
        seed(&state.pool, "host", "0.10.2").await;
        crate::runs::set_version(&state.pool, run_id, &ver)
            .await
            .unwrap();
        insert_gate(&state.pool, "host", "0.10.2", "cargo_test", 0).await;
        insert_gate(&state.pool, "host", "0.10.2", "boot_smoke", 1).await;

        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/runs/{}", run_id.0))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
        assert_eq!(v["run_id"], run_id.0);
        assert_eq!(v["sha"], "abc1234def");
        assert_eq!(v["version"], "0.10.2");
        assert_eq!(v["result"], "building");
        // Both host gates surface, latest-per-kind, alphabetized by kind.
        assert_eq!(v["gates"].as_array().unwrap().len(), 2);
        assert_eq!(v["gates"][0]["kind"], "boot_smoke");
        assert_eq!(v["gates"][0]["status"], "passed");
        assert_eq!(v["gates"][1]["kind"], "cargo_test");
        assert_eq!(v["gates"][1]["status"], "failed");
    }

    #[tokio::test]
    async fn get_run_wait_returns_immediately_when_settled() {
        let state = test_state().await;
        let run_id = crate::runs::create(&state.pool, "abc1234def")
            .await
            .unwrap();
        crate::runs::mark_passed(&state.pool, run_id).await.unwrap();

        let app = router(state);
        // Generous timeout, but an already-settled run must not wait for it.
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
        assert_eq!(v["result"], "passed");
    }

    #[tokio::test]
    async fn get_run_wait_returns_building_at_timeout() {
        let state = test_state().await;
        let run_id = crate::runs::create(&state.pool, "abc1234def")
            .await
            .unwrap();

        let app = router(state);
        // timeout_ms=0 → deadline is now → the first poll returns the
        // still-building run rather than blocking.
        let resp = app
            .oneshot(
                Request::builder()
                    .uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0))
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
        assert_eq!(v["result"], "building");
    }

    #[tokio::test]
    async fn get_run_wait_404s_for_unknown_id() {
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/runs/999/wait?timeout_ms=0")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[test]
    fn self_update_unit_maps_sha_to_instance() {
        let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap();
        assert_eq!(
            self_update_unit(&sha),
            "sando-update@abc1234def5678.service"
        );
    }

    #[tokio::test]
    async fn self_update_rejects_bad_sha_with_400() {
        // A malformed sha is a client error and must be rejected *before* any
        // privileged unit is triggered (so this test never shells out).
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/self-update")
                    .header("Content-Type", "application/json")
                    .body(Body::from(r#"{"sha":"not-a-sha!"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    // ---- /promote/{tier} default-version resolution ----

    #[tokio::test]
    async fn promote_to_first_tier_is_rejected() {
        // tier 0 is host — you /rebuild, not /promote.
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/host")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body = body_string(resp).await;
        assert!(
            body.contains("cannot /promote to the first tier"),
            "got: {body}"
        );
    }

    #[tokio::test]
    async fn promote_without_body_and_no_predecessor_version_errors() {
        // tier a has no body version supplied AND its predecessor mm has
        // current_version=NULL. Should fail before any deploy.
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body = body_string(resp).await;
        assert!(
            body.contains("no version specified") || body.contains("no current_version"),
            "got: {body}"
        );
    }

    #[tokio::test]
    async fn promote_blocked_when_predecessor_gate_never_ran() {
        // End-to-end CF1: the host tier configures boot_smoke but it never ran
        // (no gate_runs row). Promoting host -> a must be GateBlocked, citing the
        // unsatisfied gate, instead of waving through on zero evidence. A real
        // `versions` row is present so the ONLY thing that can block is the gate.
        let pool = fresh_pool().await;
        for (i, name) in ["host", "a"].iter().enumerate() {
            sqlx::query(
                "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
            )
            .bind(name)
            .bind(i as i64)
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
                .bind(name)
                .execute(&pool)
                .await
                .unwrap();
        }
        let mut topo = test_topo();
        topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate...
        let executors = Arc::new(crate::state::build_executors(&topo));
        let state = AppState {
            pool,
            topo: Arc::new(topo),
            cfg: Arc::new(test_cfg()),
            active_build: Arc::new(tokio::sync::Mutex::new(None)),
            deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
            events: crate::events::channel(),
            executors,
            api_token: None,
        };
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
            .execute(&state.pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
            .execute(&state.pool)
            .await
            .unwrap();

        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body = body_string(resp).await;
        assert!(
            body.contains("boot_smoke"),
            "expected boot_smoke to block; got: {body}"
        );
    }

    #[tokio::test]
    async fn migration_bearing_promote_requires_fresh_confirm() {
        // The predecessor (host) configures NO gates, so nothing would normally
        // block host -> a. A `bears_migration` promote must still be blocked on a
        // fresh `manual_confirm` it does not have: rollback restores the binary
        // only, so the one-way advance needs a conscious operator sign-off.
        let pool = fresh_pool().await;
        for (i, name) in ["host", "a"].iter().enumerate() {
            sqlx::query(
                "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
            )
            .bind(name)
            .bind(i as i64)
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
                .bind(name)
                .execute(&pool)
                .await
                .unwrap();
        }
        let mut topo = test_topo();
        topo.tiers[0].gates = vec![]; // host configures no gates at all
        let executors = Arc::new(crate::state::build_executors(&topo));
        let state = AppState {
            pool,
            topo: Arc::new(topo),
            cfg: Arc::new(test_cfg()),
            active_build: Arc::new(tokio::sync::Mutex::new(None)),
            deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
            events: crate::events::channel(),
            executors,
            api_token: None,
        };
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
            .execute(&state.pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
            .execute(&state.pool)
            .await
            .unwrap();

        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .header("content-type", "application/json")
                    .body(Body::from(r#"{"bears_migration": true}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::CONFLICT);
        let body = body_string(resp).await;
        assert!(
            body.contains("manual_confirm"),
            "expected the migration promote to block on manual_confirm; got: {body}"
        );
    }

    #[tokio::test]
    async fn tier_state_advance_is_atomic_previous_from_old_current() {
        // CF3: the promote advance is a single UPDATE where previous_version is
        // set from the row's *old* current_version (SQLite evaluates RHS against
        // the original row). No read-modify-write to lose under concurrency.
        let pool = fresh_pool().await;
        seed(&pool, "a", "1.0.0").await;
        // current_version FKs into versions, so the target must exist too.
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')")
            .execute(&pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'")
            .execute(&pool)
            .await
            .unwrap();

        // Exercise the sealed forward-advance primitive itself — the same op
        // /promote and the host build path both call (S1), not a copy of its SQL.
        let v = crate::domain::Version::parse("2.0.0").unwrap();
        crate::runs::advance_tier(&pool, "a", &v, None)
            .await
            .unwrap();

        let (cur, prev): (Option<String>, Option<String>) = sqlx::query_as(
            "SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(cur.as_deref(), Some("2.0.0"));
        assert_eq!(
            prev.as_deref(),
            Some("1.0.0"),
            "previous = the pre-update current, atomically"
        );
    }

    #[tokio::test]
    async fn canary_rollback_restores_deployed_nodes_to_previous_version() {
        use crate::topology::{Node, default_actuate, default_observe};
        let tmp = tempfile::tempdir().unwrap();

        // Two local nodes, each pre-seeded as if a promote had flipped them to
        // 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk.
        let mut nodes = Vec::new();
        for name in ["n1", "n2"] {
            let rr = tmp.path().join(name);
            for v in ["1.0.0", "2.0.0"] {
                tokio::fs::create_dir_all(rr.join("releases").join(v))
                    .await
                    .unwrap();
            }
            tokio::fs::symlink("releases/2.0.0", rr.join("current"))
                .await
                .unwrap();
            nodes.push(Node {
                name: name.into(),
                ssh_target: "local".into(),
                release_root: rr.to_string_lossy().into_owned(),
                service_name: "x.service".into(),
                health_url: None,
                config_check_env_file: None,
                actuate: default_actuate(),
                observe: default_observe(),
                companions: Vec::new(),
            });
        }

        let mut state = test_state().await;
        // The rollback target needs a versions row. The release dir name comes
        // from the artifact_path's parent (legacy layout: releases/<version>).
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('1.0.0','sha',datetime('now'),'/tmp/staged/releases/1.0.0/makenotwork')")
            .execute(&state.pool).await.unwrap();
        let execs: crate::state::ExecutorMap = nodes
            .iter()
            .map(|n| (n.name.clone(), crate::state::build_executor(n)))
            .collect();
        state.executors = std::sync::Arc::new(execs);

        let refs: Vec<&Node> = nodes.iter().collect();
        let restored = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
        assert_eq!(restored, 2, "both deployed nodes should be restored");

        for n in &nodes {
            let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current"))
                .await
                .unwrap();
            assert_eq!(
                cur.to_string_lossy(),
                "releases/1.0.0",
                "node {} rolled back",
                n.name
            );
        }
    }

    #[tokio::test]
    async fn canary_rollback_is_noop_without_a_previous_artifact() {
        use crate::topology::{Node, default_actuate, default_observe};
        let tmp = tempfile::tempdir().unwrap();
        let rr = tmp.path().join("n1");
        tokio::fs::create_dir_all(&rr).await.unwrap();
        let node = Node {
            name: "n1".into(),
            ssh_target: "local".into(),
            release_root: rr.to_string_lossy().into_owned(),
            service_name: "x.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: default_actuate(),
            observe: default_observe(),
            companions: Vec::new(),
        };
        let state = test_state().await; // no versions row for "9.9.9"
        let restored = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await;
        assert_eq!(
            restored, 0,
            "no artifact to roll back to -> nothing restored, no panic"
        );
    }

    // ---- FleetFake: a multi-node promote across recorded fake executors ----
    //
    // The route-level promote tests until now ran a single local node against a
    // real LocalExec, so the sequential-canary fan-out, the cross-node deploy
    // ordering, and the mid-canary rollback of already-flipped nodes had no
    // coverage. FleetFake records every deploy op (tagged by node) into one
    // shared log so ordering is visible across the fleet, and can fail any op
    // whose shell script or rsync target contains a marker — used to fail a
    // node's forward deploy of the new version while its rollback to the prior
    // version (a different `releases/<v>` path) still succeeds.

    struct FleetFake {
        tag: String,
        caps: CapabilitySet,
        log: Arc<StdMutex<Vec<String>>>,
        fail_if_contains: Option<String>,
    }

    impl FleetFake {
        fn fails(&self, text: &str) -> bool {
            self.fail_if_contains
                .as_deref()
                .is_some_and(|m| text.contains(m))
        }
    }

    #[async_trait]
    impl Executor for FleetFake {
        async fn run_streaming(
            &self,
            step: &Step,
            _sink: &mut dyn LogSink,
        ) -> anyhow::Result<RunOutput> {
            let script = step.argv.last().cloned().unwrap_or_default();
            self.log
                .lock()
                .unwrap()
                .push(format!("{}:run:{script}", self.tag));
            Ok(RunOutput {
                status: std::process::ExitStatus::from_raw(if self.fails(&script) {
                    1 << 8
                } else {
                    0
                }),
                stdout: Vec::new(),
                stderr: Vec::new(),
            })
        }
        async fn pull_file(
            &self,
            _r: &std::path::Path,
            _l: &std::path::Path,
            _o: &SyncOpts,
        ) -> anyhow::Result<()> {
            Ok(())
        }
        async fn pull_dir(
            &self,
            _r: &std::path::Path,
            _l: &std::path::Path,
            _o: &SyncOpts,
        ) -> anyhow::Result<()> {
            Ok(())
        }
        async fn pull_glob(
            &self,
            _g: &str,
            _l: &std::path::Path,
            _o: &SyncOpts,
        ) -> anyhow::Result<()> {
            Ok(())
        }
        async fn push_dir(
            &self,
            _local: &std::path::Path,
            remote: &std::path::Path,
            _o: &SyncOpts,
        ) -> anyhow::Result<()> {
            let dst = remote.display().to_string();
            self.log
                .lock()
                .unwrap()
                .push(format!("{}:push:{dst}", self.tag));
            if self.fails(&dst) {
                anyhow::bail!("fake rsync failure on {}", self.tag);
            }
            Ok(())
        }
        fn capabilities(&self) -> &CapabilitySet {
            &self.caps
        }
    }

    /// Rebuild tier "a" with `names` as remote fake nodes sharing one op log,
    /// seed the version/tier_state prerequisites for a promote of 3.0.0 up from
    /// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally
    /// make `fail_node` fail any op containing `fail_marker`. Returns the state
    /// and the shared log.
    async fn fleet_fixture(
        names: &[&str],
        fail_node: Option<&str>,
        fail_marker: &str,
    ) -> (AppState, Arc<StdMutex<Vec<String>>>) {
        use crate::topology::{default_actuate, default_observe};
        let mut state = test_state().await;
        let log = Arc::new(StdMutex::new(Vec::<String>::new()));

        let nodes: Vec<Node> = names
            .iter()
            .map(|name| Node {
                name: (*name).into(),
                ssh_target: format!("deploy@{name}"),
                release_root: format!("/tmp/fleet/{name}"),
                service_name: "makenotwork.service".into(),
                health_url: None,
                config_check_env_file: None,
                actuate: default_actuate(),
                observe: default_observe(),
                companions: Vec::new(),
            })
            .collect();

        let mut topo = (*state.topo).clone();
        topo.tiers[1].nodes = nodes.clone();
        // Drop the tier's post-deploy gate: the deploy fan-out is the subject
        // here, and node_health would need its own probe wiring.
        topo.tiers[1].gates = vec![];
        state.topo = Arc::new(topo);

        let execs: crate::state::ExecutorMap = nodes
            .iter()
            .map(|n| {
                let nm = n.name.to_string();
                let fail = fail_node == Some(nm.as_str());
                let exec: Arc<dyn Executor> = Arc::new(FleetFake {
                    tag: nm,
                    caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
                    log: log.clone(),
                    fail_if_contains: fail.then(|| fail_marker.to_string()),
                });
                (n.name.clone(), exec)
            })
            .collect();
        state.executors = Arc::new(execs);

        for n in &nodes {
            // deploys.node FKs into `nodes`.
            sqlx::query(
                "INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)",
            )
            .bind(&n.name)
            .bind(&n.ssh_target)
            .bind(&n.release_root)
            .execute(&state.pool)
            .await
            .unwrap();
        }
        for v in ["1.0.0", "2.0.0", "3.0.0"] {
            // Legacy (pre-identity) artifact_path is `releases/<version>/<bin>`,
            // so the release dir the node mirrors is named for the version. The
            // fixture reflects that real layout (parent basename == version).
            sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
                .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
        }
        sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
            .execute(&state.pool)
            .await
            .unwrap();
        sqlx::query(
            "UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'",
        )
        .execute(&state.pool)
        .await
        .unwrap();
        (state, log)
    }

    /// Index of the first logged op belonging to `node` (panics if the node was
    /// never touched — the message names it).
    fn first_touch(log: &[String], node: &str) -> usize {
        let prefix = format!("{node}:");
        log.iter()
            .position(|e| e.starts_with(&prefix))
            .unwrap_or_else(|| panic!("node {node:?} was never deployed to; log: {log:#?}"))
    }

    #[tokio::test]
    async fn promote_deploys_every_node_in_tier_order_and_advances() {
        let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await;
        let pool = state.pool.clone();

        let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect("all nodes deploy, so the promote succeeds");
        assert_eq!(
            body["nodes_deployed"],
            serde_json::json!(["a1", "a2", "a3"]),
            "the response names every node the promote reached",
        );

        let (cur, prev) = tier_versions(&pool, "a").await;
        assert_eq!(cur.as_deref(), Some("3.0.0"));
        assert_eq!(prev.as_deref(), Some("2.0.0"));

        // Sequential canary: a1 is fully touched before a2, a2 before a3.
        let log = log.lock().unwrap().clone();
        assert!(
            first_touch(&log, "a1") < first_touch(&log, "a2")
                && first_touch(&log, "a2") < first_touch(&log, "a3"),
            "nodes must deploy in tier order: {log:#?}",
        );

        // Every node has a green deploy row for the promoted version.
        let ok: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(ok, 3, "one ok deploy row per node");
    }

    #[tokio::test]
    async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() {
        // a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is
        // never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their
        // rollback ops target `releases/2.0.0`, which the marker does not match —
        // and tier_state must NOT advance.
        let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await;
        let pool = state.pool.clone();

        let err = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect_err("a mid-canary deploy failure must fail the promote");
        assert!(
            matches!(err, crate::error::Error::Other(_)),
            "a deploy failure propagates as Other, got: {err:?}",
        );

        // tier_state untouched: the failure returns before advance_tier.
        let (cur, prev) = tier_versions(&pool, "a").await;
        assert_eq!(
            cur.as_deref(),
            Some("2.0.0"),
            "a failed rollout must not advance"
        );
        assert_eq!(prev.as_deref(), Some("1.0.0"));

        let log = log.lock().unwrap().clone();
        // a3 sits after the failed a2 in the sequence and is never touched.
        assert!(
            !log.iter().any(|e| e.starts_with("a3:")),
            "nodes after the failure must not be deployed to: {log:#?}",
        );
        // Both touched nodes were rolled back to the prior version.
        for n in ["a1", "a2"] {
            assert!(
                log.iter()
                    .any(|e| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")),
                "touched node {n} must be restored to 2.0.0: {log:#?}",
            );
        }

        // The forward attempt is on the record: a1 ok, a2 failed.
        let a1: String = sqlx::query_scalar(
            "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(a1, "ok");
        let a2: String = sqlx::query_scalar(
            "SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(a2, "failed");

        // Both touched nodes restored => the tier is consistent on 2.0.0, so the
        // partial flag is cleared, not set.
        let reason: Option<String> =
            sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(
            reason, None,
            "a fully-restored canary leaves the tier consistent, not partial",
        );
    }

    /// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a
    /// promote had flipped it to `current` with `prev` still staged on disk.
    /// Returns the state (topology rewired to the tempdir node) and the tempdir,
    /// which the caller must keep alive.
    async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) {
        use crate::topology::{Node, default_actuate, default_observe};
        let tmp = tempfile::tempdir().unwrap();
        let rr = tmp.path().join("a-local");
        for v in [prev, current] {
            tokio::fs::create_dir_all(rr.join("releases").join(v))
                .await
                .unwrap();
        }
        tokio::fs::symlink(format!("releases/{current}"), rr.join("current"))
            .await
            .unwrap();
        let node = Node {
            name: "a-local".into(),
            ssh_target: "local".into(),
            release_root: rr.to_string_lossy().into_owned(),
            service_name: "x.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: default_actuate(),
            observe: default_observe(),
            companions: Vec::new(),
        };

        let mut state = test_state().await;
        let mut topo = (*state.topo).clone();
        topo.tiers[1].nodes = vec![node];
        state.executors = Arc::new(crate::state::build_executors(&topo));
        state.topo = Arc::new(topo);

        // deploys.node FKs into `nodes`, so the promote path needs the row.
        sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
            .bind(rr.to_string_lossy().into_owned())
            .execute(&state.pool).await.unwrap();
        for v in [prev, current] {
            sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
                .bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
        }
        sqlx::query(
            "UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'",
        )
        .bind(current)
        .bind(prev)
        .execute(&state.pool)
        .await
        .unwrap();
        (state, tmp)
    }

    async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option<String>, Option<String>) {
        sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
            .bind(tier)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn rollback_clears_previous_version_rather_than_swapping_it() {
        // The swap bug: writing the version we just rolled OFF into
        // previous_version made a second /rollback roll FORWARD onto the broken
        // build the operator was escaping. previous_version must go NULL.
        let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
        let pool = state.pool.clone();

        let _ = rollback(State(state), Path("a".to_string())).await.unwrap();

        let (cur, prev) = tier_versions(&pool, "a").await;
        assert_eq!(
            cur.as_deref(),
            Some("1.0.0"),
            "rolled back to the previous version"
        );
        assert_eq!(
            prev, None,
            "the version we rolled off must NOT become the rollback target"
        );
    }

    #[tokio::test]
    async fn second_rollback_refuses_instead_of_rolling_forward() {
        let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
        let pool = state.pool.clone();
        let release_root = state.topo.tiers[1].nodes[0].release_root.clone();

        let _ = rollback(State(state.clone()), Path("a".to_string()))
            .await
            .unwrap();
        let err = rollback(State(state), Path("a".to_string()))
            .await
            .expect_err("only one step of history is tracked; a second rollback must refuse");
        assert!(
            matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")),
            "expected a loud refusal, got: {err:?}",
        );

        // The refusal is total: neither the DB nor the node moved back to 2.0.0.
        let (cur, _) = tier_versions(&pool, "a").await;
        assert_eq!(cur.as_deref(), Some("1.0.0"));
        let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
            .await
            .unwrap();
        assert_eq!(
            link.to_string_lossy(),
            "releases/1.0.0",
            "node stays on the rolled-back version"
        );
    }

    #[tokio::test]
    async fn promote_refuses_an_unprovisioned_tier() {
        // Every step of a promote to a node-less tier is a silent no-op that
        // still reports success: the deploy loop iterates nothing and
        // advance_tier records a current_version the tier never received.
        let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
        let mut topo = (*state.topo).clone();
        topo.tiers[1].provisioned = false;
        topo.tiers[1].nodes.clear();
        state.topo = Arc::new(topo);
        let pool = state.pool.clone();
        sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'")
            .execute(&pool)
            .await
            .unwrap();

        let err = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect_err("promoting to an unprovisioned tier must be refused");
        assert!(
            matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")),
            "got: {err:?}",
        );

        // And nothing was recorded: the tier keeps whatever it had before.
        let (cur, _) = tier_versions(&pool, "a").await;
        assert_eq!(
            cur.as_deref(),
            Some("2.0.0"),
            "a refused promote must not advance tier_state",
        );
    }

    #[tokio::test]
    async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() {
        // The happy path. Every other promote test asserts a refusal or a red
        // outcome, so nothing pinned what a *successful* promote actually does:
        // deploy reaches the node, the `current` symlink flips, tier_state
        // advances with previous_version = the version we came off, any stale
        // partial flag clears, and the handler reports the nodes it touched.
        use crate::topology::Gate;
        let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;

        // Source tier `host` gates the promote on cargo_test. Satisfying it is
        // the point of the test: the sibling case asserts an unsatisfied gate
        // blocks, this one asserts a satisfied gate lets the promote through.
        let mut topo = (*state.topo).clone();
        topo.tiers[0].gates = vec![Gate::CargoTest];
        state.topo = Arc::new(topo);
        let pool = state.pool.clone();
        let release_root = state.topo.tiers[1].nodes[0].release_root.clone();

        // 3.0.0 is staged on the build host and green on `host`.
        tokio::fs::create_dir_all(
            std::path::Path::new(&release_root)
                .join("releases")
                .join("3.0.0"),
        )
        .await
        .unwrap();
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')")
            .execute(&pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
            .execute(&pool)
            .await
            .unwrap();
        insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await;

        // A stale flag from an earlier incident, which a clean rollout clears.
        set_partial(&state, &tid("a"), "left over from a previous canary").await;

        let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect("gates are green and the node deploys, so the promote succeeds");

        assert_eq!(body["tier"], "a");
        assert_eq!(body["version"], "3.0.0");
        assert_eq!(
            body["nodes_deployed"],
            serde_json::json!(["a-local"]),
            "the response names every node the promote reached",
        );

        let (cur, prev) = tier_versions(&pool, "a").await;
        assert_eq!(cur.as_deref(), Some("3.0.0"));
        assert_eq!(
            prev.as_deref(),
            Some("2.0.0"),
            "previous_version is the version we came off, so a rollback aims at it",
        );

        // The node genuinely moved: the promote is not just bookkeeping.
        let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
            .await
            .unwrap();
        assert_eq!(link.to_string_lossy(), "releases/3.0.0");

        let reason: Option<String> =
            sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(
            reason, None,
            "a clean full rollout clears a stale partial flag",
        );

        // The deploy is on the record as having succeeded, which is what the
        // next promote's gate check and /state both read.
        let (node, outcome): (String, String) =
            sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(node, "a-local");
        assert_eq!(outcome, "ok");
    }

    #[tokio::test]
    async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() {
        // The deploy reached every node, so tier_state advances (a stale
        // current_version would aim a later rollback at the wrong artifact), but
        // the promote must NOT report success: the tier is flagged partial and
        // the handler returns the gate failure.
        use crate::topology::Gate;
        let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
        // node_health is the tier's only gate, and it fails closed with no
        // probes — an empty executor map gives it nothing to probe while
        // deploy_node still falls back to a built executor and succeeds.
        let mut topo = (*state.topo).clone();
        topo.tiers[1].gates = vec![Gate::NodeHealth];
        state.topo = Arc::new(topo);
        state.executors = Arc::new(crate::state::ExecutorMap::new());
        let pool = state.pool.clone();
        // Promote 3.0.0 up from host, which configures no gates of its own.
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')")
            .execute(&pool).await.unwrap();
        sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
            .execute(&pool)
            .await
            .unwrap();

        let err = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect_err("red post-deploy gates must fail the promote");
        assert!(
            matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")),
            "the failure must name the red gate, got: {err:?}",
        );

        let (cur, _) = tier_versions(&pool, "a").await;
        assert_eq!(
            cur.as_deref(),
            Some("3.0.0"),
            "tier_state tracks what the nodes actually run"
        );
        let reason: Option<String> =
            sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert!(
            reason.as_deref().is_some_and(|r| r.contains("node_health")),
            "the tier must be flagged for /state and the TUI, got: {reason:?}",
        );
    }

    #[tokio::test]
    async fn set_partial_then_clear_roundtrips() {
        let state = test_state().await;
        let read = || async {
            sqlx::query_scalar::<_, Option<String>>(
                "SELECT partial_reason FROM tier_state WHERE tier = 'a'",
            )
            .fetch_one(&state.pool)
            .await
            .unwrap()
        };
        assert_eq!(read().await, None, "consistent tier starts clean");
        set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
        assert_eq!(
            read().await.as_deref(),
            Some("canary rollback incomplete: 1/2")
        );
        clear_partial(&state, &tid("a")).await;
        assert_eq!(read().await, None, "clear nulls it back out");
    }

    #[tokio::test]
    async fn state_surfaces_partial_reason() {
        use axum::extract::State;
        let state = test_state().await;
        set_partial(
            &state,
            &tid("a"),
            "first-deploy canary failed: 1 node(s) on 2.0.0",
        )
        .await;
        let Json(view) = get_state(State(state)).await.unwrap();
        let a = view.tiers.iter().find(|t| t.name == "a").unwrap();
        assert_eq!(
            a.partial_reason.as_deref(),
            Some("first-deploy canary failed: 1 node(s) on 2.0.0"),
        );
        let host = view.tiers.iter().find(|t| t.name == "host").unwrap();
        assert_eq!(
            host.partial_reason, None,
            "untouched tier stays clean in /state"
        );
    }

    #[tokio::test]
    async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() {
        use axum::extract::State;
        let state = test_state().await;
        // No build runs yet → build is null, so /state doesn't pretend a build
        // is happening.
        let Json(view) = get_state(State(state.clone())).await.unwrap();
        assert!(view.build.is_none());

        // A failed run must surface its cause in /state, not just in /runs.
        let run_id = crate::runs::create(&state.pool, "deadbeef").await.unwrap();
        crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed")
            .await
            .unwrap();
        let Json(view) = get_state(State(state)).await.unwrap();
        let b = view.build.expect("build surfaced");
        assert_eq!(b.run_id, run_id.0);
        assert_eq!(b.result, "failed");
        assert_eq!(
            b.failure_summary.as_deref(),
            Some("cargo_test: 3 test(s) failed")
        );
    }

    #[tokio::test]
    async fn status_json_serves_the_shared_payload_over_the_real_router() {
        // The mapping itself is tested in `crate::status`. This asserts the
        // route is wired, serves valid JSON, and stays internally consistent
        // (no dangling child or action references) against a real topology
        // rather than a hand-built fixture.
        let state = test_state().await;
        set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;

        let resp = router(state)
            .oneshot(
                Request::builder()
                    .uri("/status.json")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let body = http_body_util::BodyExt::collect(resp.into_body())
            .await
            .unwrap()
            .to_bytes();
        let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap();

        assert_eq!(payload.source, crate::status::SOURCE);
        assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
        assert_eq!(payload.validate(), Ok(()));
        assert_eq!(
            payload.node("tier:a").unwrap().status,
            ops_status::Status::Failed,
            "a partial tier must surface as failed over the wire"
        );
        assert_eq!(payload.worst_status(), ops_status::Status::Failed);
    }

    #[tokio::test]
    async fn promote_with_explicit_version_but_missing_artifact_404s() {
        // Explicit version supplied, gates trivially pass (mm has none in
        // test_topo), but `versions` table has no row → 404.
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .header("content-type", "application/json")
                    .body(Body::from(r#"{"version":"9.9.9"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    // ---- GET /logs/{version}/{gate} ----

    async fn state_with_logs_root(logs_root: PathBuf) -> AppState {
        let mut s = test_state().await;
        let mut cfg = (*s.cfg).clone();
        cfg.logs_root = logs_root;
        s.cfg = Arc::new(cfg);
        s
    }

    #[tokio::test]
    async fn get_gate_log_returns_file_contents() {
        let tmp = tempfile::tempdir().unwrap();
        let dir = tmp.path().join("0.9.5");
        tokio::fs::create_dir_all(&dir).await.unwrap();
        tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n")
            .await
            .unwrap();

        let state = state_with_logs_root(tmp.path().to_path_buf()).await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/0.9.5/cargo_test")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(body_string(resp).await, "hello sandod\n");
    }

    #[tokio::test]
    async fn get_gate_log_404s_when_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let state = state_with_logs_root(tmp.path().to_path_buf()).await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/0.9.5/cargo_test")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    // ---- CF2: bearer-token auth on deploy mutators ----

    #[tokio::test]
    async fn mutating_route_requires_bearer_when_token_set() {
        let mut state = test_state().await;
        state.api_token = Some(std::sync::Arc::from("s3cr3t"));
        let app = router(state);

        // No Authorization header -> 401, before any deploy logic runs.
        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Wrong token -> 401.
        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .header("authorization", "Bearer nope")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Correct token -> passes auth (then blocked downstream by gates /
        // missing predecessor version, but specifically NOT 401).
        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/promote/a")
                    .header("authorization", "Bearer s3cr3t")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn read_routes_require_token_when_set() {
        // Reads expose prod state (versions, SHAs, gate logs, the event stream),
        // so they are bearer-gated too — not just the mutators. A tailnet peer
        // without the token gets 401; the TUI presents the token and gets 200.
        let mut state = test_state().await;
        state.api_token = Some(std::sync::Arc::from("s3cr3t"));
        let app = router(state);

        // No token -> 401 on a read.
        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/state")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Correct token -> 200.
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/state")
                    .header("authorization", "Bearer s3cr3t")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn read_routes_open_without_a_token() {
        // The loopback/dev posture: no token configured, reads pass through so a
        // local TUI needs no credential.
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/state")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn self_update_malformed_body_is_400_not_422() {
        // TypedBody funnels a JSON deserialize failure through the Error envelope
        // (400), not axum's raw 422 — keeping every mutator on one error contract.
        let state = test_state().await; // no token -> auth passes, body is the gate
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/self-update")
                    .header("content-type", "application/json")
                    .body(Body::from("{ not valid json"))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn gate_log_rejects_unknown_gate_kind() {
        // The gate segment is an allowlisted GateKind, not a free-form filename:
        // an unknown kind is a 404, so `*.log` basenames can't be probed.
        let state = test_state().await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/0.9.6/passwd")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    /// Path-traversal guard: a `..` segment must not escape logs_root.
    /// axum's `{name}` param already rejects a literal `/` in the value, but
    /// `..` as a whole segment is structurally valid and must be blocked at
    /// the handler.
    #[tokio::test]
    async fn get_gate_log_rejects_dotdot_segments() {
        let tmp = tempfile::tempdir().unwrap();
        let state = state_with_logs_root(tmp.path().to_path_buf()).await;
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/../etc/passwd")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    // ---- build-identity path (wiki release-artifact-identity) ----

    /// Insert a settled (passed) build_runs row with content identity and return
    /// its id. `staged_path` is the content-addressed release dir the bundle was
    /// published to (`releases/<digest16>`).
    async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 {
        sqlx::query_scalar(
            "INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path)
             VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id",
        )
        .bind(sha)
        .bind(version)
        .bind(format!("{sha}-digest"))
        .bind(staged_path)
        .fetch_one(pool)
        .await
        .unwrap()
    }

    async fn insert_gate_build(
        pool: &SqlitePool,
        tier: &str,
        version: &str,
        kind: &str,
        passed: i64,
        build_id: i64,
    ) {
        let status = if passed == 1 { "passed" } else { "failed" };
        sqlx::query(
            "INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \
             VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)",
        )
        .bind(version)
        .bind(tier)
        .bind(kind)
        .bind(status)
        .bind(build_id)
        .execute(pool)
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn unsatisfied_gates_keys_build_evidence_on_build_id() {
        let pool = fresh_pool().await;
        seed(&pool, "a", "3.0.0").await;
        // Two builds of the SAME version string. Only b1's gate ran.
        let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await;
        let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await;
        insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await;

        // b1's own evidence satisfies.
        let ok = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::CargoTest],
            "3.0.0",
            Some(b1),
            false,
        )
        .await
        .unwrap();
        assert!(ok.is_empty(), "the build that passed its gate is satisfied");

        // b2 shares the version but has no gate row of its own: fail closed. This
        // is the hole — a rebuild reusing the version must not ride b1's evidence.
        let bad = unsatisfied_gates(
            &pool,
            &tid("a"),
            &[Gate::CargoTest],
            "3.0.0",
            Some(b2),
            false,
        )
        .await
        .unwrap();
        assert_eq!(
            bad,
            vec!["cargo_test".to_string()],
            "a different build of the same version does not inherit the evidence"
        );

        // Legacy (pre-identity) callers still resolve by version string.
        let legacy = unsatisfied_gates(&pool, &tid("a"), &[Gate::CargoTest], "3.0.0", None, false)
            .await
            .unwrap();
        assert!(legacy.is_empty(), "version-keyed legacy path unchanged");
    }

    #[tokio::test]
    async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() {
        // The burn-in hole: `promote --version Y` used to check the SOURCE tier's
        // clock/evidence (which belong to whatever is current there), letting Y
        // inherit another build's 48h. Now promote resolves the source's current
        // build and refuses an explicit version that isn't it.
        let state = test_state().await;
        seed_version(&state.pool, "3.0.0").await;
        let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await;
        sqlx::query(
            "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
        )
        .bind(b)
        .execute(&state.pool)
        .await
        .unwrap();

        let err = promote_inner(
            state,
            "a".into(),
            PromoteBody {
                version: Some("2.0.0".into()),
                ..Default::default()
            },
        )
        .await
        .expect_err("promoting a version other than the source build must be refused");
        assert!(
            matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")),
            "the refusal must name the mismatch, got: {err:?}",
        );
    }

    #[tokio::test]
    async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() {
        // Full promote through the identity path: source tier points at a build,
        // promote resolves the artifact through it, deploys build_runs.staged_path
        // (content-addressed), and advances the target's current_build_id.
        let mut state = test_state().await;
        let node_root = tempfile::tempdir().unwrap();
        // Point the a-local node at a real tempdir so the symlink swap is checkable.
        let mut topo = (*state.topo).clone();
        topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned();
        topo.tiers[1].gates = vec![]; // isolate the deploy fan-out
        state.topo = Arc::new(topo);
        state.executors = Arc::new(crate::state::build_executors(&state.topo));

        // deploys.node FKs into `nodes`, so the promote path needs the row.
        sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
            .bind(node_root.path().to_string_lossy().into_owned())
            .execute(&state.pool)
            .await
            .unwrap();

        seed_version(&state.pool, "3.0.0").await;
        let staged = format!(
            "{}/releases/abc123abc123abc1",
            node_root.path().to_string_lossy()
        );
        let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await;
        sqlx::query(
            "UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
        )
        .bind(b)
        .execute(&state.pool)
        .await
        .unwrap();

        let pool = state.pool.clone();
        let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
            .await
            .expect("identity-path promote succeeds");
        assert_eq!(body["version"], "3.0.0");

        // Target tier advanced with the build identity, not just the version.
        let (cur_v, cur_b): (Option<String>, Option<i64>) = sqlx::query_as(
            "SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(cur_v.as_deref(), Some("3.0.0"));
        assert_eq!(cur_b, Some(b), "target records the promoted build id");

        // The deploy row is attributed to the build.
        let deploy_b: Option<i64> = sqlx::query_scalar(
            "SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(deploy_b, Some(b));

        // The node's `current` points at the content-addressed release dir, whose
        // name is the staged_path's basename — not the version.
        let link = tokio::fs::read_link(node_root.path().join("current"))
            .await
            .unwrap();
        assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1");
    }

    /// Insert a bare `versions` row (FK target for tier_state.current_version).
    async fn seed_version(pool: &SqlitePool, version: &str) {
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
            .bind(version).execute(pool).await.unwrap();
    }
}