//! Promotion service: the deploy state machine lifted out of the HTTP
//! layer — promote orchestration, canary rollback of deployed nodes,
//! partial-state flags, and the promote-time gate-satisfaction check.
//! routes/mod.rs stays HTTP glue that calls into these.

use super::{AppState, Json, PromoteBody, Result};

pub(super) async fn promote_inner(
    s: AppState,
    tier: String,
    body: PromoteBody,
) -> Result<Json<serde_json::Value>> {
    // Serialize the whole check -> deploy -> advance against any concurrent
    // promote/rollback (CF3). Held for the entire task.
    let _deploy_guard = s.deploy_lock.lock().await;
    let tier = crate::domain::TierId::new(tier);
    let idx = s
        .topo
        .tiers
        .iter()
        .position(|t| t.name == tier)
        .ok_or(crate::error::Error::NotFound)?;
    if idx == 0 {
        return Err(crate::error::Error::GateBlocked(
            "cannot /promote to the first tier; use /rebuild".into(),
        ));
    }
    let target = &s.topo.tiers[idx];
    let source = &s.topo.tiers[idx - 1];
    // An unprovisioned tier has no nodes, so every step below is a no-op that
    // still *looks* like success: the deploy loop iterates nothing, node_health
    // returns Blocked, and advance_tier writes a current_version for a tier that
    // has never received a byte. /state then reports tier c running a version it
    // does not have. Refuse instead.
    if !target.provisioned {
        return Err(crate::error::Error::GateBlocked(format!(
            "tier {} is not provisioned (no nodes); promoting to it would record a              version it never received",
            target.name,
        )));
    }

    // Resolve the artifact through the SOURCE tier's evidence, not a version
    // string (invariant 2, wiki [[release-artifact-identity]]). The build that is
    // *current on the source tier* is the one that burned in and passed its gates
    // there; promoting anything else lets a version string ride on another build's
    // evidence and clock — the "promote --version Y inherits X's 48h" hole. A
    // given `body.version` must name that same build.
    let source_build: Option<(Option<i64>, Option<String>, Option<String>, Option<String>)> =
        sqlx::query_as(
            "SELECT ts.current_build_id, br.version, br.staged_path, br.platform
         FROM tier_state ts
         LEFT JOIN build_runs br ON br.id = ts.current_build_id
         WHERE ts.app = ? AND ts.tier = ?",
        )
        .bind(&s.cfg.id)
        .bind(&source.name)
        .fetch_optional(&s.pool)
        .await
        .map_err(crate::error::Error::Db)?;

    let (build_id, version_str, staged_dir, source_platform) = match source_build {
        // Identity path: the source tier points at a recorded build.
        Some((Some(bid), ver, staged_path, platform)) => {
            let version_str = ver.ok_or_else(|| {
                crate::error::Error::Other(anyhow::anyhow!(
                    "source build {bid} on tier {} has no recorded version",
                    source.name
                ))
            })?;
            let staged_path = staged_path.ok_or_else(|| {
                crate::error::Error::Other(anyhow::anyhow!(
                    "source build {bid} ({version_str}) has no staged_path; cannot promote"
                ))
            })?;
            if let Some(req) = &body.version
                && req != &version_str
            {
                return Err(crate::error::Error::GateBlocked(format!(
                    "tier {} is running build {bid} ({version_str}); refusing to promote an \
                     explicit version {req} that is not the build the tier vouched for",
                    source.name
                )));
            }
            let platform = platform
                .as_deref()
                .map(crate::domain::Platform::parse)
                .transpose()
                .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;
            (
                Some(bid),
                version_str,
                std::path::PathBuf::from(staged_path),
                platform,
            )
        }
        // Legacy fallback: NULL current_build_id (a pre-identity tier). Resolve by
        // version string through the versions table, exactly as before. One clean
        // build-and-promote cycle re-anchors the tier onto the identity path.
        _ => {
            let version_str = match body.version.clone() {
                Some(v) => v,
                None => sqlx::query_scalar::<_, Option<String>>(
                    "SELECT current_version FROM tier_state WHERE app = ? AND tier = ?",
                )
                .bind(&s.cfg.id)
                .bind(&source.name)
                .fetch_optional(&s.pool)
                .await
                .map_err(crate::error::Error::Db)?
                .flatten()
                .ok_or_else(|| {
                    crate::error::Error::GateBlocked(format!(
                        "no version specified and tier {} has no current_version",
                        source.name
                    ))
                })?,
            };
            let bin: Option<(String,)> =
                sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
                    .bind(&s.cfg.id)
                    .bind(&version_str)
                    .fetch_optional(&s.pool)
                    .await
                    .map_err(crate::error::Error::Db)?;
            let Some((bin,)) = bin else {
                return Err(crate::error::Error::NotFound);
            };
            // `artifact_path` is the primary binary; the release dir is its parent.
            let staged_dir = std::path::PathBuf::from(&bin)
                .parent()
                .ok_or_else(|| {
                    crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent"))
                })?
                .to_path_buf();
            (None, version_str, staged_dir, None)
        }
    };
    let version = crate::domain::Version::parse(&version_str)
        .map_err(|e| crate::error::Error::Other(anyhow::anyhow!(e)))?;

    // 1. Predecessor must have all of its configured gates satisfied for this
    //    build (with optional hotfix override that skips burn_in). Evaluated
    //    against the topology gate list, so a gate that never ran blocks the
    //    promote instead of being treated as green.
    //
    //    A migration-bearing promote (`bears_migration`) forces a fresh
    //    `manual_confirm` on the predecessor even if the tier does not configure
    //    one: rollback restores the binary + release_contents only, never the
    //    schema, so the operator must consciously acknowledge the one-way advance
    //    (deploy/README.md "Rollback contract"). `hotfix` skips only `burn_in`, so
    //    it does not suppress this confirm.
    let mut effective_gates = source.gates.clone();
    if body.bears_migration
        && !effective_gates
            .iter()
            .any(|g| matches!(g, crate::topology::Gate::ManualConfirm))
    {
        effective_gates.push(crate::topology::Gate::ManualConfirm);
    }
    //    Which bytes each node gets is resolved FIRST, because it is what says
    //    how many builds the gate check has to cover. A two-architecture promote
    //    ships two bundles with two sets of evidence, and checking only the one
    //    the source tier points at would wave the sibling through on gate rows
    //    nobody read. It also means a version missing its other half fails here,
    //    before any gate work, rather than halfway down the rollout.
    let target_nodes: Vec<&crate::topology::Node> = target.nodes.iter().collect();
    let bundles = bundles_for_nodes(
        &s,
        &version_str,
        &target_nodes,
        &staged_dir,
        source_platform.as_ref(),
        build_id,
    )
    .await?;
    let mut promoted_builds = distinct_builds(&bundles);
    if promoted_builds.is_empty() {
        // A provisioned tier with no nodes ships nothing, so there is nothing to
        // resolve — but the gate check still has to be keyed on the source build.
        // Left empty it would fall through to the version-string lookup, which is
        // the pre-identity path and weaker than what this tier is owed. The
        // `provisioned` guard above makes this unreachable today; it is written
        // out because the alternative fails quietly in the direction of less
        // evidence.
        promoted_builds.push(PromotedBuild {
            platform: source_platform.clone(),
            build_id,
        });
    }

    let pending = unsatisfied_gates(
        &s.pool,
        &s.cfg.id,
        &source.name,
        &effective_gates,
        &version_str,
        &promoted_builds,
        body.hotfix,
    )
    .await?;
    if !pending.is_empty() {
        return Err(crate::error::Error::GateBlocked(format!(
            "{} gate(s) not satisfied on tier {}: {}",
            pending.len(),
            source.name,
            pending.join(", "),
        )));
    }

    // The version this tier was running before this promote — the rollback
    // target if a canary node fails partway through a multi-node rollout.
    let prev_version: Option<String> =
        sqlx::query_scalar("SELECT current_version FROM tier_state WHERE app = ? AND tier = ?")
            .bind(&s.cfg.id)
            .bind(&target.name)
            .fetch_optional(&s.pool)
            .await
            .map_err(crate::error::Error::Db)?
            .flatten();

    // 3. Deploy to each node. Sequential canary is the only policy
    //    implemented in v0; parallel is a one-line change once we trust the
    //    sequential path. Track the nodes already flipped to the new version so
    //    a mid-rollout failure can roll them back (canary rollback).
    let mut deployed: Vec<&crate::topology::Node> = Vec::new();
    // The build id is not read here: it did its work above, keying the gate
    // check to each bundle's own evidence. What ships is decided by `Placement`.
    for (node, node_bundle, node_bundle_platform, _) in &bundles {
        // The proof that these bytes belong on this box. Built before the
        // deploy row is written, so a mismatch never becomes an `in_progress`
        // deploy that has to be reconciled.
        let placement =
            crate::deploy::Placement::check(node, node_bundle, node_bundle_platform.as_ref())
                .map_err(|e| crate::error::Error::GateBlocked(e.to_string()))?;
        let started = chrono::Utc::now().to_rfc3339();
        crate::events::emit(
            &s.events,
            crate::events::Event::DeployStart {
                tier: target.name.clone(),
                node: node.name.clone(),
                version: version.clone(),
            },
        );
        let executor = s
            .executors
            .get(&node.name)
            .cloned()
            .unwrap_or_else(|| crate::state::build_executor(node));
        // Record the deploy as `in_progress` BEFORE touching the node, so a crash
        // between the node's symlink swap and this promote's `advance_tier` leaves
        // a durable trace. Without the pre-write, a SIGKILL after the swap but
        // before the row is inserted would land the new binary on the node with no
        // DB evidence at all, and the startup reconcile — which reads `deploys` —
        // could not see it. The row is finalized to ok/failed immediately below;
        // any row still `in_progress` at startup is an orphan the reconcile settles
        // and flags ([`crate::reconcile`]).
        let deploy_id: i64 = sqlx::query_scalar(
            "INSERT INTO deploys (app, version, tier, node, started_at, outcome, hotfix, reset_burn_in, build_id)
             VALUES (?, ?, ?, ?, ?, 'in_progress', ?, ?, ?) RETURNING id",
        )
        .bind(&s.cfg.id)
        .bind(&version).bind(&target.name).bind(&node.name)
        .bind(&started)
        .bind(body.hotfix as i64).bind(body.reset_burn_in as i64)
        .bind(build_id)
        .fetch_one(&s.pool).await.map_err(crate::error::Error::Db)?;
        let result = crate::deploy::deploy_node(
            executor.as_ref(),
            placement,
            &version_str,
            s.cfg.primary_bin(),
        )
        .await;
        let finished = chrono::Utc::now().to_rfc3339();
        let (outcome_obj, err_for_propagation) = match result {
            Ok(_) => (crate::outcome::DeployOutcome::ok(), None),
            Err(e) => {
                let msg = format!("{e:#}");
                let kind = crate::classify::classify_deploy_error(&msg);
                (crate::outcome::DeployOutcome::failed(kind), Some(e))
            }
        };
        let outcome_json = serde_json::to_string(&outcome_obj)
            .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}"));
        sqlx::query(
            "UPDATE deploys SET finished_at = ?, outcome = ?, outcome_json = ? WHERE id = ?",
        )
        .bind(&finished)
        .bind(outcome_obj.status_str())
        .bind(&outcome_json)
        .bind(deploy_id)
        .execute(&s.pool)
        .await
        .map_err(crate::error::Error::Db)?;
        if let Some(e) = err_for_propagation {
            let crate::outcome::DeployStatus::Failed { failure } = outcome_obj.status else {
                unreachable!("err_for_propagation is Some iff status is Failed");
            };
            tracing::error!(
                tier = %target.name, node = %node.name, version = %version,
                failure = failure.summary(),
                "deploy failed; current symlink left intact, tier_state not advanced"
            );
            crate::events::emit(
                &s.events,
                crate::events::Event::DeployFailed {
                    tier: target.name.clone(),
                    node: node.name.clone(),
                    version: version.clone(),
                    failure,
                },
            );

            // Canary rollback: restore every node this promote touched — the
            // ones already flipped to the new version AND this failed node
            // (whose state is indeterminate: the symlink swap may have landed
            // before the restart failed) — back to the tier's prior version, so
            // the fleet is left consistent on `prev` rather than split-brain.
            // Nodes after this one were never touched and stay on `prev`.
            deployed.push(node);
            let touched = deployed.len();
            match prev_version.as_deref() {
                Some(prev) => {
                    let report = rollback_deployed_nodes(&s, &target.name, &deployed, prev).await;
                    // Say what is true, and say it differently when nothing is
                    // wrong. `restored=0 of=1` read as total failure when the
                    // accurate reading was "0 needed restoring" — the same
                    // confusion as the per-node message, one level up.
                    if report.is_consistent() {
                        tracing::warn!(
                            tier = %target.name,
                            restored = report.restored,
                            already_on_previous = report.already_on_previous,
                            of = report.touched(),
                            from = %version, to = prev,
                            "canary failed mid-rollout; every touched node is on the previous \
                             version and the tier is consistent",
                        );
                    } else {
                        tracing::error!(
                            tier = %target.name,
                            restored = report.restored,
                            already_on_previous = report.already_on_previous,
                            indeterminate = report.indeterminate,
                            of = report.touched(),
                            from = %version, to = prev,
                            "canary failed mid-rollout and the tier is NOT consistent; some nodes \
                             have an indeterminate version",
                        );
                    }
                    if report.restored > 0
                        && let Ok(prev_v) = crate::domain::Version::parse(prev)
                    {
                        crate::events::emit(
                            &s.events,
                            crate::events::Event::Rollback {
                                tier: target.name.clone(),
                                from: version.clone(),
                                to: prev_v,
                            },
                        );
                    }
                    // The tier is consistent when no node's version is unknown —
                    // which includes the case where a rollback "failed" before
                    // the swap and so left the node on `prev` already. Flagging
                    // that as partial would put a permanent scare on /state for
                    // a fleet that is entirely on one version.
                    if report.is_consistent() {
                        clear_partial(&s, &target.name).await;
                    } else {
                        set_partial(&s, &target.name, &format!(
                            "canary rollback incomplete: {indeterminate} of {touched} node(s) have an \
                             indeterminate version and may be on {version}; {restored} restored to \
                             {prev}, {already} were never swapped — manual check needed",
                            touched = report.touched(),
                            indeterminate = report.indeterminate,
                            restored = report.restored,
                            already = report.already_on_previous,
                        )).await;
                    }
                }
                None => {
                    tracing::error!(
                        tier = %target.name, count = touched, version = %version,
                        "canary failed on a first deploy (no previous version to restore to); \
                         touched nodes remain on the new version — manual cleanup needed",
                    );
                    set_partial(
                        &s,
                        &target.name,
                        &format!(
                            "first-deploy canary failed: {touched} node(s) left on {version}, \
                         no prior version to restore — manual cleanup needed",
                        ),
                    )
                    .await;
                }
            }
            return Err(crate::error::Error::Other(e));
        }
        deployed.push(node);
        crate::events::emit(
            &s.events,
            crate::events::Event::DeployOk {
                tier: target.name.clone(),
                node: node.name.clone(),
                version: version.clone(),
            },
        );
    }

    // 3b. Run this tier's post-deploy gates (node_health) against the freshly
    //     deployed nodes and record their outcomes. These rows are the evidence
    //     the NEXT promote (this tier -> the following one) checks via
    //     `unsatisfied_gates`. Before CF1, only the host tier ran gates, so
    //     A/B/C had no evidence and promotion waved through; node_health now
    //     proves the deployed nodes are serving (Run-2 SERIOUS-3: boot_smoke
    //     used to re-run the staged binary locally and proved nothing about the
    //     node). burn_in / manual_confirm are not run here — they are evaluated
    //     live / by the operator at the next promote. A failed gate does not
    //     unwind this deploy (the artifact is already live on the tier); it
    //     blocks the next promote, which is the fail-closed behavior we want.
    //
    //     It does, however, fail *this* promote's response. tier_state still
    //     advances below — the nodes genuinely run this version and a stale
    //     `current_version` would send a later rollback to the wrong artifact —
    //     but the tier is flagged partial and the handler returns the failure,
    //     so the operator learns at the promote instead of discovering it as an
    //     unexplained block on the next one.
    let post_deploy: Vec<crate::topology::Gate> = target
        .gates
        .iter()
        .filter(|g| g.runs_post_deploy())
        .cloned()
        .collect();
    let mut post_deploy_failure: Option<String> = None;
    if !post_deploy.is_empty() {
        // node_health probes each node the deploy just shipped to, over the same
        // executor the deploy used. Build the probe set from the tier's nodes and
        // the startup executor map; a node missing an executor (shouldn't happen
        // — both come from the same topology) is skipped, and an empty set makes
        // node_health Blocked (fail closed).
        let nodes: Vec<crate::gates::NodeProbe> = target
            .nodes
            .iter()
            .filter_map(|n| {
                s.executors
                    .get(&n.name)
                    .map(|exec| crate::gates::NodeProbe {
                        node: n.name.clone(),
                        service: n.service_name.clone(),
                        health_url: n.health_url.clone(),
                        executor: exec.clone(),
                    })
            })
            .collect();
        let ctx = crate::gates::GateCtx {
            pool: s.pool.clone(),
            cfg: s.cfg.clone(),
            tier: target.name.clone(),
            version: version.clone(),
            // No worktree at promote time; node_health works over executors, not
            // a checkout. No bundle either: the gates here are about the nodes,
            // not about bytes on this host.
            worktree: None,
            bundle: None,
            events: s.events.clone(),
            nodes,
            // These post-deploy gate rows are the evidence the NEXT promote (this
            // tier -> the following one) resolves through, so they must carry the
            // build they vouch for.
            build_id,
            // The hostname the public uses, straight from the tier. page_smoke
            // has to request the site the way a visitor does; anything derived
            // from a node would reach the origin and miss the CDN, which is the
            // layer the gate exists to watch.
            public_url: target.public_url.clone(),
            // No checkout at promote time, so nothing to resolve against.
            aux_dirs: std::collections::HashMap::new(),
        };
        post_deploy_failure = match crate::gates::run_all(&ctx, &post_deploy).await {
            Ok(failed) if failed.is_empty() => None,
            Ok(failed) => {
                let names = failed
                    .iter()
                    .map(|k| k.as_str())
                    .collect::<Vec<_>>()
                    .join(", ");
                tracing::warn!(
                    tier = %target.name, version = %version, gates = %names,
                    "post-deploy gate(s) failed; tier advanced but promotion to the next tier is blocked",
                );
                Some(format!(
                    "post-deploy gate(s) failed on {version}: {names}; \
                     the tier is serving {version} but cannot promote onward until they pass"
                ))
            }
            Err(e) => {
                tracing::error!(
                    tier = %target.name, version = %version, error = %e,
                    "post-deploy gate execution errored; promotion to the next tier is blocked",
                );
                Some(format!(
                    "post-deploy gate execution errored on {version}: {e}; \
                     the tier is serving {version} but cannot promote onward until the gates pass"
                ))
            }
        };
    }

    // 4. Advance tier_state through the single sealed forward-advance op (atomic
    //    self-referential UPDATE; no read-modify-write to lose under concurrency,
    //    CF3). We hold deploy_lock for this whole handler, so the advance is
    //    serialized against rollback and the host build path's advance.
    //    reset_burn_in on the *source* tier nulls its clock only when the operator
    //    explicitly asked.
    crate::runs::advance_tier(&s.pool, &s.cfg.id, target.name.as_str(), &version, build_id)
        .await
        .map_err(crate::error::Error::Db)?;

    if body.reset_burn_in {
        sqlx::query("UPDATE tier_state SET burn_in_started_at = NULL WHERE app = ? AND tier = ?")
            .bind(&s.cfg.id)
            .bind(&source.name)
            .execute(&s.pool)
            .await
            .map_err(crate::error::Error::Db)?;
    }

    // Red post-deploy gates: the rollout itself reached every node, but the tier
    // is not fit to promote onward. Flag it so /state and the TUI say so, and
    // return the failure rather than a 200 the operator would read as "shipped,
    // all good". tier_state has already advanced above — it describes what the
    // nodes are running, not whether we're happy about it.
    if let Some(reason) = post_deploy_failure {
        set_partial(&s, &target.name, &reason).await;
        return Err(crate::error::Error::GateBlocked(reason));
    }

    // A clean full rollout to every node clears any prior partial flag on this tier.
    clear_partial(&s, &target.name).await;

    crate::events::emit(
        &s.events,
        crate::events::Event::PromoteComplete {
            tier: target.name.clone(),
            version: version.clone(),
        },
    );
    tracing::info!(
        version = %version, tier = %target.name,
        hotfix = body.hotfix, reset_burn_in = body.reset_burn_in,
        "promote complete",
    );

    Ok(Json(serde_json::json!({
        "tier": target.name,
        "version": version,
        "nodes_deployed": target.nodes.iter().map(|n| n.name.clone()).collect::<Vec<_>>(),
    })))
}

/// What a canary rollback actually left behind, per node.
///
/// Three outcomes, not two, and conflating the middle one with the last is the
/// bug this type exists to prevent: a rollback that fails *before* the symlink
/// swap leaves the node exactly where it already was, on the previous version.
/// Reporting that as "stranded on the new version, manual intervention needed"
/// sends an operator to do surgery on a healthy production box.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(super) struct RollbackReport {
    /// Put back on the previous version by a successful redeploy.
    pub(super) restored: usize,
    /// Rollback failed before the swap, so the node never left the previous
    /// version. Nothing is stranded and nothing needs doing.
    pub(super) already_on_previous: usize,
    /// Rollback failed at or after the swap, or could not be attempted at all.
    /// The node's version is not knowable from here. This is the only outcome
    /// that warrants a human.
    pub(super) indeterminate: usize,
}

impl RollbackReport {
    /// Every touched node accounted for.
    pub(super) fn touched(self) -> usize {
        self.restored + self.already_on_previous + self.indeterminate
    }

    /// True when no node is in an unknown state, whether or not every rollback
    /// "succeeded". A tier whose rollbacks all failed before the swap is
    /// consistent on the previous version and is not an incident.
    pub(super) fn is_consistent(self) -> bool {
        self.indeterminate == 0
    }
}

/// A node, the bundle it is to receive, what that bundle runs on, and the build
/// row that bundle came from.
///
/// The build id is what makes the gate check per-platform: evidence is keyed on
/// the build that produced the bytes, so a promote that ships two architectures
/// has to look up two sets of gate rows, and this is where it learns which two.
/// `None` is the legacy path (a bundle resolved by version string, from a tier
/// with no `current_build_id`), where there is no build to key on.
pub(super) type NodeBundle<'a> = (
    &'a crate::topology::Node,
    std::path::PathBuf,
    Option<crate::domain::Platform>,
    Option<i64>,
);

/// One bundle a promote is about to ship, and the evidence key it is judged by.
///
/// A single-platform product has exactly one of these and it is the source
/// tier's own build, which is what every promote before two-architecture support
/// checked. A product like pom has one per architecture, each standing on its
/// own intake and its own gate run.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct PromotedBuild {
    /// What the bundle runs on, when it says. Used only to qualify the name of a
    /// failing gate, so an operator reading "cargo_test" knows which half.
    pub platform: Option<crate::domain::Platform>,
    /// The `build_runs` row the evidence is keyed on; `None` falls back to the
    /// version-keyed lookup for a pre-identity tier.
    pub build_id: Option<i64>,
}

/// Which bytes each node gets, resolved before any node is touched.
///
/// A single-platform product answers with `fallback` for every node, which is
/// the bundle the caller already resolved (the source tier's build on a promote,
/// the previous version's on a rollback). A product whose one version is several
/// bundles answers per node, because the node is the thing that says which
/// architecture it can run.
///
/// Resolving the whole tier up front is deliberate: a version missing its
/// x86_64 half fails the promote before its aarch64 half has been pushed
/// anywhere, instead of halfway down a rollout with nodes already flipped.
pub(super) async fn bundles_for_nodes<'a>(
    s: &AppState,
    version: &str,
    nodes: &[&'a crate::topology::Node],
    fallback: &std::path::Path,
    fallback_platform: Option<&crate::domain::Platform>,
    fallback_build_id: Option<i64>,
) -> Result<Vec<NodeBundle<'a>>> {
    let mut out = Vec::with_capacity(nodes.len());
    for node in nodes.iter().copied() {
        let resolved = match &node.platform {
            // The node states nothing, so there is nothing to resolve against;
            // it gets the caller's bundle, and `Placement::check` decides
            // whether that pairing is admissible at all.
            None => (
                fallback.to_path_buf(),
                fallback_platform.cloned(),
                fallback_build_id,
            ),
            // The node states a platform. Give it the bundle recorded for that
            // platform at this version — the caller's own when they agree, its
            // sibling when they do not.
            Some(want) if fallback_platform == Some(want) => (
                fallback.to_path_buf(),
                fallback_platform.cloned(),
                fallback_build_id,
            ),
            Some(want) => {
                let (build_id, path) = bundle_for_platform(s, version, want).await?;
                (path, Some(want.clone()), build_id)
            }
        };
        out.push((node, resolved.0, resolved.1, resolved.2));
    }
    Ok(out)
}

/// The distinct builds `bundles` will ship, in a stable order.
///
/// Deduplicated because a tier is usually several nodes on one architecture, and
/// evaluating one build's gates once per node would say the same thing three
/// times in the error an operator reads. Keyed on the whole entry rather than on
/// the build id alone, so the legacy `None` case does not collapse two
/// version-resolved bundles into one.
pub(super) fn distinct_builds(bundles: &[NodeBundle<'_>]) -> Vec<PromotedBuild> {
    let mut out: Vec<PromotedBuild> = Vec::new();
    for (_, _, platform, build_id) in bundles {
        let entry = PromotedBuild {
            platform: platform.clone(),
            build_id: *build_id,
        };
        if !out.contains(&entry) {
            out.push(entry);
        }
    }
    out
}

/// The bundle recorded for `version` on `platform`.
///
/// This is what makes a two-architecture product promotable. One pom version is
/// two bundles with two digests, each built natively by Bento and accepted
/// through its own intake; the tier ladder carries the version forward and this
/// answers "which of that version's bundles does this box take".
///
/// Only a run that settled green qualifies. A sibling that failed its own host
/// gates is not a fallback for the one that passed — the source tier's evidence
/// says nothing about bytes it never saw, so each architecture's bundle stands
/// on its own intake and its own gate run.
async fn bundle_for_platform(
    s: &AppState,
    version: &str,
    platform: &crate::domain::Platform,
) -> Result<(Option<i64>, std::path::PathBuf)> {
    let row: Option<(i64, Option<String>)> = sqlx::query_as(
        "SELECT id, staged_path FROM build_runs
         WHERE app = ? AND version = ? AND platform = ? AND result = 'passed'
         ORDER BY id DESC LIMIT 1",
    )
    .bind(&s.cfg.id)
    .bind(version)
    .bind(platform.to_string())
    .fetch_optional(&s.pool)
    .await
    .map_err(crate::error::Error::Db)?;

    match row {
        // The id comes back alongside the path so the caller can check this
        // build's own gate evidence rather than the source tier's. A green build
        // row is not a green gate run: the build says the bytes compiled, the
        // gate run says the tier vouched for them.
        Some((id, Some(path))) => Ok((Some(id), std::path::PathBuf::from(path))),
        Some((_, None)) => Err(crate::error::Error::Other(anyhow::anyhow!(
            "the {platform} build of {version} has no staged_path; cannot promote it"
        ))),
        None => Err(crate::error::Error::GateBlocked(format!(
            "no green {platform} bundle recorded for {version}. Each architecture is \
             a separate artifact with its own evidence, so this one has to be built \
             and accepted before a {platform} node can take this version"
        ))),
    }
}

/// After a canary node fails mid-promote, restore the nodes already flipped to
/// the new version back to `prev_version`, leaving the tier consistent (all on
/// the old version) rather than split-brain. Best-effort: every node is
/// attempted; a per-node failure is logged but never propagated (the promote is
/// already failing).
///
/// Returns a [`RollbackReport`] rather than a bare count, because "the rollback
/// failed" is not the same claim as "the node is on the new version" and the
/// caller has to be able to tell them apart. When the previous version has no
/// recorded artifact, no rollback can be attempted and every touched node is
/// reported indeterminate.
pub(super) async fn rollback_deployed_nodes(
    s: &AppState,
    tier: &crate::domain::TierId,
    nodes: &[&crate::topology::Node],
    prev_version: &str,
) -> RollbackReport {
    let bin: Option<(String,)> =
        match sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
            .bind(&s.cfg.id)
            .bind(prev_version)
            .fetch_optional(&s.pool)
            .await
        {
            Ok(b) => b,
            Err(e) => {
                tracing::error!(tier = %tier, prev = prev_version, error = %e,
                "canary rollback: looking up the previous artifact failed; no rollback attempted");
                return RollbackReport {
                    indeterminate: nodes.len(),
                    ..Default::default()
                };
            }
        };
    let Some((bin,)) = bin else {
        tracing::error!(tier = %tier, prev = prev_version, nodes = nodes.len(),
            "canary rollback: previous version has no artifact_path; no rollback attempted");
        return RollbackReport {
            indeterminate: nodes.len(),
            ..Default::default()
        };
    };
    let Some(staged_dir) = std::path::PathBuf::from(&bin)
        .parent()
        .map(std::path::Path::to_path_buf)
    else {
        tracing::error!(tier = %tier, prev = prev_version,
            "canary rollback: previous artifact_path has no parent dir; no rollback attempted");
        return RollbackReport {
            indeterminate: nodes.len(),
            ..Default::default()
        };
    };

    // A rollback ships bytes too, so it needs the same per-node resolution a
    // promote does: on a two-architecture product the previous version is also
    // two bundles, and restoring the wrong one would leave the node no better
    // off than the failed canary did. A resolution failure here cannot be
    // propagated (the promote is already failing), so it reports the node as
    // indeterminate — which is the truth: nothing was attempted and the node's
    // version is whatever the failed deploy left.
    // No fallback build id: a rollback resolves the previous version's bundles
    // by version string and re-runs no gates, so there is no evidence to key.
    // The bytes are ones this tier already ran.
    let bundles = match bundles_for_nodes(s, prev_version, nodes, &staged_dir, None, None).await {
        Ok(b) => b,
        Err(e) => {
            tracing::error!(tier = %tier, prev = prev_version, error = %e,
                "canary rollback: could not resolve a previous bundle per node; no rollback attempted");
            return RollbackReport {
                indeterminate: nodes.len(),
                ..Default::default()
            };
        }
    };

    let mut report = RollbackReport::default();
    for (node, node_bundle, node_bundle_platform, _) in &bundles {
        let executor = s
            .executors
            .get(&node.name)
            .cloned()
            .unwrap_or_else(|| crate::state::build_executor(node));
        let placement =
            match crate::deploy::Placement::check(node, node_bundle, node_bundle_platform.as_ref())
            {
                Ok(p) => p,
                Err(e) => {
                    report.indeterminate += 1;
                    tracing::error!(tier = %tier, node = %node.name, error = %e,
                        "canary rollback: refused to place the previous bundle on this node");
                    continue;
                }
            };
        match crate::deploy::deploy_node(
            executor.as_ref(),
            placement,
            prev_version,
            s.cfg.primary_bin(),
        )
        .await
        {
            Ok(_) => {
                report.restored += 1;
                tracing::warn!(tier = %tier, node = %node.name, version = prev_version,
                    "canary rollback: node restored to the previous version");
            }
            // A rollback is itself a deploy, so it fails at a stage too. Failing
            // before the swap means it never touched `current` — the node is
            // still on the version it was already running, which is the one we
            // were rolling back TO. That is the intended end state reached by a
            // different route, not an incident.
            Err(e) => match crate::deploy::stage_of(&e) {
                Some(crate::deploy::FailureStage::BeforeSwap) => {
                    report.already_on_previous += 1;
                    tracing::warn!(
                        tier = %tier, node = %node.name, version = prev_version,
                        error = %format!("{e:#}"),
                        "canary rollback did not run, and did not need to: it failed before the \
                         symlink swap, so the node is already on the previous version",
                    );
                }
                // Unannotated errors land here deliberately. Guessing "safe"
                // would reintroduce the original bug in the worse direction.
                stage => {
                    report.indeterminate += 1;
                    tracing::error!(
                        tier = %tier, node = %node.name, version = prev_version,
                        error = %format!("{e:#}"),
                        stage = ?stage,
                        "canary rollback FAILED for node at or after the symlink swap; its version \
                         is indeterminate — manual intervention needed",
                    );
                }
            },
        }
    }
    report
}

/// Flag a tier as left in a partial / mixed-version state, with a human-readable
/// reason surfaced through `/state` and the TUI. Best-effort: a failure to record
/// the flag is logged, never propagated — the caller is already on an error path
/// and the worse outcome is to mask the original failure with a bookkeeping one.
pub(super) async fn set_partial(s: &AppState, tier: &crate::domain::TierId, reason: &str) {
    if let Err(e) =
        sqlx::query("UPDATE tier_state SET partial_reason = ? WHERE app = ? AND tier = ?")
            .bind(reason)
            .bind(&s.cfg.id)
            .bind(tier)
            .execute(&s.pool)
            .await
    {
        tracing::error!(tier = %tier, reason, error = %e,
            "failed to record tier partial state; the fleet may be inconsistent without a /state flag");
    }
}

/// Clear a tier's partial flag after a clean full promote or rollback. Errors are
/// logged but not propagated: the deploy itself succeeded, and a stale flag is a
/// visible nuisance, not a safety regression (the operator sees a partial marker
/// on a tier that is actually fine, and re-checks).
pub(super) async fn clear_partial(s: &AppState, tier: &crate::domain::TierId) {
    if let Err(e) =
        sqlx::query("UPDATE tier_state SET partial_reason = NULL WHERE app = ? AND tier = ?")
            .bind(&s.cfg.id)
            .bind(tier)
            .execute(&s.pool)
            .await
    {
        tracing::warn!(tier = %tier, error = %e, "failed to clear tier partial flag");
    }
}

/// Returns the kinds of `tier`'s *configured* gates that are not satisfied for
/// `version`. `hotfix` suppresses the `burn_in` requirement only.
///
/// Fail-closed against the topology gate list (the CF1 fix). The previous
/// version inspected only existing `gate_runs` rows, so a configured gate that
/// had *never run* produced no row and was invisibly treated as green — letting
/// a promote wave through with zero evidence (it shipped 0.9.5 to prod with
/// tier A's `boot_smoke` never recorded). Now every configured gate must show
/// positive evidence:
/// - `burn_in` is evaluated live against the tier's clock (a stored `blocked`
///   row would otherwise never flip to passed as time elapses);
/// - every other kind requires a `passed` row for (tier, version) — a missing
///   or non-passed latest row counts as unsatisfied.
///
/// `builds` are the artifacts this promote will actually ship (wiki note
/// `release-artifact-identity`). Each carries a `build_id`, and the deterministic
/// build-evidence gates are checked against the rows that vouched for *that
/// build* rather than against any row that happens to carry the version string —
/// this is what stops a `promote --version Y` from riding on gate rows a
/// different build left under the same version. A `None` build id is the
/// legacy/pre-identity path: fall back to the version-keyed lookup so a
/// mid-migration tier (NULL `current_build_id`) still promotes.
///
/// **One promote can ship several builds**, and every one of them is checked.
/// A pom version is two bundles with two digests, each built natively and
/// accepted through its own intake, so the source tier's evidence for one
/// architecture says nothing about the other. Checking only the build the tier
/// points at would let an x86_64 node take bytes whose `cargo_test` row nobody
/// looked at. When there is more than one, a failing gate is reported qualified
/// by platform, because "cargo_test not satisfied" is not actionable if the
/// operator cannot tell which half it is about.
///
/// `burn_in` and `manual_confirm` are evaluated once regardless: both are keyed
/// on the tier's own clock (`tier_state.burn_in_started_at`), not on a build, so
/// asking per build would ask the same question N times and answer it N times in
/// the error.
pub(super) async fn unsatisfied_gates(
    pool: &sqlx::SqlitePool,
    app: &crate::domain::AppId,
    tier: &crate::domain::TierId,
    gates: &[crate::topology::Gate],
    version: &str,
    builds: &[PromotedBuild],
    hotfix: bool,
) -> std::result::Result<Vec<String>, crate::error::Error> {
    use crate::topology::Gate;
    let mut bad = Vec::new();
    for gate in gates {
        let kind = gate.kind();
        match gate {
            Gate::BurnIn { hours } => {
                if hotfix {
                    continue;
                }
                let ok = crate::gates::burn_in_satisfied(pool, app, tier, *hours)
                    .await
                    .map_err(crate::error::Error::Other)?;
                if !ok {
                    bad.push(kind.as_str().to_string());
                }
            }
            Gate::ManualConfirm => {
                // A confirmation must be *fresh*: recorded at or after the
                // version's current landing on this tier (tier_state
                // .burn_in_started_at, the per-deploy clock). Without this a
                // confirmation row survives a rollback + rollback-forward and
                // waves a re-deploy of the same version through with no fresh
                // operator sign-off — weaker than burn_in, which is clock-based.
                // No baseline (NULL) => fail closed: require a fresh confirm.
                let confirmed_at: Option<String> = sqlx::query_scalar(
                    "SELECT finished_at FROM gate_runs
                     WHERE app = ?1 AND tier = ?2 AND version = ?3
                       AND gate_kind = 'manual_confirm' AND status = 'passed'
                     ORDER BY id DESC LIMIT 1",
                )
                .bind(app)
                .bind(tier.as_str())
                .bind(version)
                .fetch_optional(pool)
                .await
                .map_err(crate::error::Error::Db)?
                .flatten();
                let landed_at: Option<String> = sqlx::query_scalar(
                    "SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?",
                )
                .bind(app)
                .bind(tier.as_str())
                .fetch_optional(pool)
                .await
                .map_err(crate::error::Error::Db)?
                .flatten();
                let fresh = match (confirmed_at, landed_at) {
                    (Some(c), Some(l)) => {
                        match (
                            chrono::DateTime::parse_from_rfc3339(&c),
                            chrono::DateTime::parse_from_rfc3339(&l),
                        ) {
                            (Ok(cd), Ok(ld)) => cd >= ld,
                            _ => false, // unparseable timestamp -> fail closed
                        }
                    }
                    _ => false,
                };
                if !fresh {
                    bad.push(kind.as_str().to_string());
                }
            }
            // Build/post-deploy gates that leave a `gate_runs` row: the latest
            // row for this (tier, version, kind) must be `passed`. Listed
            // explicitly (no `_` catch-all) so adding a new `Gate` variant is a
            // compile error here until its promotion semantics are decided —
            // a transient-`blocked` kind silently falling into "needs a passed
            // row" would be permanently unsatisfiable.
            Gate::CargoTest
            | Gate::HardeningTest
            | Gate::Clippy
            | Gate::Fmt
            | Gate::CargoAudit
            | Gate::CargoDeny
            | Gate::MigrationDryRun
            | Gate::CodeSmoke
            | Gate::BootSmoke
            | Gate::NodeHealth
            // Same evidence rule as node_health: a passed row for this tier's
            // current build, or the promote out of here is refused.
            | Gate::PageSmoke => {
                // Latest row for this configured gate kind; NULL/missing/any
                // non-'passed' status all count as unsatisfied (fail closed).
                // Every build this promote ships has to show its own passed row.
                // An empty `builds` is the caller saying "no identities to key
                // on"; one version-keyed lookup is the pre-identity behaviour.
                let lookups: &[PromotedBuild] = if builds.is_empty() {
                    &[PromotedBuild {
                        platform: None,
                        build_id: None,
                    }]
                } else {
                    builds
                };
                let qualify = lookups.len() > 1;
                for b in lookups {
                    // Keyed on build_id when the artifact has an identity (the
                    // evidence must be for *this* build), else on the version.
                    let status: Option<String> = match b.build_id {
                        Some(bid) => sqlx::query_scalar(
                            "SELECT status FROM gate_runs
                             WHERE app = ?1 AND tier = ?2 AND build_id = ?3 AND gate_kind = ?4
                             ORDER BY id DESC LIMIT 1",
                        )
                        .bind(app)
                        .bind(tier.as_str())
                        .bind(bid)
                        .bind(kind.as_str()),
                        None => sqlx::query_scalar(
                            "SELECT status FROM gate_runs
                             WHERE app = ?1 AND tier = ?2 AND version = ?3 AND gate_kind = ?4
                             ORDER BY id DESC LIMIT 1",
                        )
                        .bind(app)
                        .bind(tier.as_str())
                        .bind(version)
                        .bind(kind.as_str()),
                    }
                    .fetch_optional(pool)
                    .await
                    .map_err(crate::error::Error::Db)?
                    .flatten();
                    if status.as_deref() != Some("passed") {
                        // Qualified only when there is more than one build to
                        // tell apart, so a single-platform product's message is
                        // exactly what it always was.
                        let name = match (&b.platform, qualify) {
                            (Some(p), true) => format!("{} ({p})", kind.as_str()),
                            _ => kind.as_str().to_string(),
                        };
                        if !bad.contains(&name) {
                            bad.push(name);
                        }
                    }
                }
            }
        }
    }
    Ok(bad)
}