//! Build orchestration: resolve a sha to a worktree, read the server version,
//! shell out to `cargo build --release`, record a `versions` row.
//!
//! Runs as a tokio task spawned from `POST /rebuild`; the HTTP request
//! returns the version id immediately and the task drives the rest.

use crate::config::AppConfig;
use crate::deploy;
use crate::domain::{GitSha, Platform, RunId, TierId, Version};
use crate::gates::{self, GateCtx};
use crate::git;
use crate::topology::Topology;
use anyhow::{Context, Result};
use chrono::Utc;
use sqlx::SqlitePool;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::process::Command;

#[derive(Debug, Clone)]
pub struct BuildArtifact {
    pub version: Version,
    pub git_sha: GitSha,
    pub worktree: PathBuf,
    /// One entry per `cfg.bin_names` in declared order. First is the primary
    /// (referenced by the systemd unit's ExecStart). Paths are inside the
    /// worktree's `target/release/`.
    pub binary_paths: Vec<PathBuf>,
    /// `(companion name, built binary path)` for each `cfg.companions`, built
    /// from the same worktree/sha as the server. Staged into the release bundle
    /// under `companions/<name>/<bin>` and installed by the nodes that opt in.
    pub companion_paths: Vec<(String, PathBuf)>,
}

/// The live kernel hostname (`/proc/sys/kernel/hostname`, trimmed). Linux-only,
/// which Sando is. The build-host guard reads this rather than `$HOSTNAME`
/// (not reliably exported) so the check reflects the actual machine.
fn runtime_hostname() -> Result<String> {
    let raw = std::fs::read_to_string("/proc/sys/kernel/hostname")
        .context("reading /proc/sys/kernel/hostname for the build-host guard")?;
    Ok(raw.trim().to_string())
}

/// Pure half of the build-host guard: fail unless the live host matches the
/// configured build host. Split out from [`enforce_build_host`] so it is unit-
/// testable without depending on the test machine's hostname.
fn check_build_host(actual: &str, expected: &str) -> Result<()> {
    anyhow::ensure!(
        actual == expected,
        "refusing to build on host {actual}: configured build host is {expected} \
         (never build on a prod/serving node)",
    );
    Ok(())
}

/// Refuse to build unless this daemon is running on the configured build host.
fn enforce_build_host(expected: &str) -> Result<()> {
    check_build_host(&runtime_hostname()?, expected)
}

pub async fn run(
    pool: SqlitePool,
    cfg: Arc<AppConfig>,
    topo: Arc<Topology>,
    sha: GitSha,
    events: crate::events::EventTx,
    run_id: RunId,
) -> Result<BuildArtifact> {
    // Build-host guard: refuse to compile anywhere but the configured build
    // host. The build runs locally (`cargo build` in cfg.workdir), so a sandod
    // misdeployed onto a prod/serving node would otherwise build there — exactly
    // the "never build on prod" rule. Enforced before any cargo invocation so a
    // wrong-host daemon fails fast with a clear message rather than compiling.
    let Some(build_host) = cfg.build_host.as_deref() else {
        anyhow::bail!(
            "{} declares no build_host, which makes it intake-only: Sando does not compile \
             it. Ship it with POST /intake, from a builder that does.",
            cfg.id
        );
    };
    enforce_build_host(build_host)?;

    let repo = topo.repo.as_ref().with_context(|| {
        format!(
            "{} declares no [repo]: it is intake-only and Sando has no source to check out",
            cfg.id
        )
    })?;
    let worktree = cfg.workdir.join(sha.as_str());
    let bare = PathBuf::from(&repo.bare_path);

    crate::runs::set_phase(&pool, run_id, crate::runs::Phase::Fetching)
        .await
        .ok();

    // Pull-based ingestion: if an upstream remote is configured, fetch the
    // deploy branch so a just-pushed sha is locally resolvable. A fetch
    // failure is non-fatal — the sha may already be present from a prior
    // fetch or a direct push; the presence check below is the real gate.
    if let Some(upstream) = repo.upstream.as_deref()
        && let Err(e) = git::fetch_upstream(&bare, upstream, &repo.branch).await
    {
        tracing::warn!(error = %e, upstream, "upstream fetch failed; proceeding with current bare-repo state");
    }
    anyhow::ensure!(
        git::sha_present(&bare, sha.as_str()).await?,
        "sha {} not present in bare repo {} after fetch — push the commit to the upstream remote first",
        sha.as_str(),
        bare.display(),
    );

    git::checkout_worktree(&bare, sha.as_str(), &worktree).await?;

    // Check out any auxiliary repos (e.g. synckit) beside the worktree so a
    // cross-repo path dependency in the server or a companion resolves. Fails the
    // build if an aux repo can't be assembled — a companion that silently fails to
    // find its source would fail the compile downstream with a worse message.
    checkout_aux_repos(&cfg, &topo).await?;

    let server_dir = worktree.join("server");
    let version = read_pkg_version(&server_dir.join("Cargo.toml"))
        .await
        .with_context(|| format!("reading version from {}/Cargo.toml", server_dir.display()))?;
    crate::runs::set_version(&pool, run_id, &version).await.ok();

    // sqlx compile-time query checking needs a live DB with the current schema.
    // We point cargo at the scratch DB and prep it (drop public, re-migrate)
    // before invoking cargo build. The same DB is reset again by
    // `migration_dry_run` later if it runs as a gate.
    let mut cargo_cmd = Command::new("cargo");
    cargo_cmd
        .arg("build")
        .arg("--release")
        .current_dir(&server_dir)
        .kill_on_drop(true);
    // Shared build cache across per-sha worktrees: reuse one target dir so an
    // incremental diff doesn't clean-compile from scratch. Serialized builds
    // make this contention-free. Unset → cargo's default per-worktree target/.
    if let Some(target) = cfg.cargo_target_dir.as_deref() {
        cargo_cmd.env("CARGO_TARGET_DIR", target);
    }
    if let Some(scratch_url) = cfg.scratch_db_url.as_deref() {
        tracing::info!(sha = %sha.as_str(), "preparing scratch DB schema for sqlx compile-time checks");
        crate::gates::reset_scratch(scratch_url, &cfg.scratch_owner_role)
            .await
            .context("scratch DB reset before build")?;
        crate::gates::run_migrator(scratch_url, &server_dir.join("migrations"))
            .await
            .context("applying MNW migrations to scratch DB before build")?;
        cargo_cmd.env("DATABASE_URL", scratch_url);
    } else {
        tracing::warn!("scratch_db_url unset; sqlx will fall back to offline mode and may fail");
    }

    crate::runs::set_phase(&pool, run_id, crate::runs::Phase::Compiling)
        .await
        .ok();
    tracing::info!(sha = %sha, version = %version, dir = %server_dir.display(), "cargo build --release start");
    crate::events::emit(
        &events,
        crate::events::Event::BuildStart {
            sha: sha.clone(),
            version: version.clone(),
        },
    );
    let started = std::time::Instant::now();
    let out = cargo_cmd.output().await.context("spawning cargo build")?;
    let elapsed_s = started.elapsed().as_secs();
    if !out.status.success() {
        tracing::error!(sha = %sha, version = %version, elapsed_s, "cargo build --release failed");
        crate::events::emit(
            &events,
            crate::events::Event::BuildFailed {
                sha: sha.clone(),
                version: version.clone(),
                elapsed_s,
            },
        );
        // Settle the run with the headline compiler diagnostic (not the raw
        // 4 KB tail) so `GET /runs/{id}` answers "why" without a journald dive.
        let summary = crate::classify::classify_compile_error(&out.stdout, &out.stderr).summary();
        if let Err(e) = crate::runs::mark_failed(&pool, run_id, &summary).await {
            tracing::error!(run_id = %run_id, error = %e, "persisting compile-fail verdict failed; run may show stale 'building' until restart-reconcile");
        }
        anyhow::bail!(
            "cargo build --release failed:\n{}",
            tail(&out.stderr, 4_000)
        );
    }
    tracing::info!(sha = %sha, version = %version, elapsed_s, "cargo build --release ok");
    crate::events::emit(
        &events,
        crate::events::Event::BuildOk {
            sha: sha.clone(),
            version: version.clone(),
            elapsed_s,
        },
    );

    // Binaries land under `<target>/release/`; with a shared target dir that's
    // not inside the worktree, so resolve it the same way cargo did above.
    let release_dir = cfg
        .cargo_target_dir
        .as_deref()
        .map_or_else(|| server_dir.join("target/release"), |t| t.join("release"));
    let mut binary_paths = Vec::with_capacity(cfg.bin_names.len());
    for name in &cfg.bin_names {
        let p = release_dir.join(name);
        anyhow::ensure!(p.exists(), "expected binary at {} after build", p.display());
        binary_paths.push(p);
    }
    // Primary binary path is the one we record in `versions.artifact_path`
    // (everything downstream — promote, rollback — looks it up by version).
    let primary = binary_paths[0].clone();

    // Companion crates (e.g. mnw-cli): built from the SAME worktree/sha so a
    // service that shares the server's internal-API contract cannot drift out of
    // lockstep (the 2026-07-09 git-hosting outage). A companion build failure
    // fails the whole pipeline — the server never ships without its companions.
    let mut companion_paths = Vec::with_capacity(cfg.companions.len());
    for c in &cfg.companions {
        let bin = build_companion(&worktree, &cfg, c).await?;
        companion_paths.push((c.name.clone(), bin));
    }

    sqlx::query(
        "INSERT OR IGNORE INTO versions (app, version, git_sha, built_at, artifact_path)
         VALUES (?, ?, ?, ?, ?)",
    )
    .bind(&cfg.id)
    .bind(&version)
    .bind(&sha)
    .bind(Utc::now().to_rfc3339())
    .bind(primary.to_string_lossy().as_ref())
    .execute(&pool)
    .await?;

    Ok(BuildArtifact {
        version,
        git_sha: sha,
        worktree,
        binary_paths,
        companion_paths,
    })
}

/// Fetch and check out every configured auxiliary repo at `cfg.workdir/<checkout_dir>`,
/// so a cross-repo path dependency built from the main worktree resolves (wiki
/// [[sando-overview]]; the synckit split, task sando-18cdb32f).
///
/// Each aux repo is a fixed, shared checkout refreshed to `branch` HEAD — not
/// per-sha — because the dependent's relative path resolves to that fixed spot
/// regardless of the main sha, and builds serialize. A fetch failure is a warning
/// (the branch may already be present from a prior build); an unresolvable branch
/// after that is fatal, as is a failed worktree — a half-assembled source tree
/// must fail the build here, loudly, not as a downstream compile error.
/// Where an aux repo's checkout lands. The single derivation: `checkout_aux_repos`
/// creates it here and `GateCtx::aux_dirs` resolves `test_target`s against it, so
/// the two cannot drift into looking in different places.
pub fn aux_checkout_dir(cfg: &AppConfig, aux: &crate::topology::AuxRepo) -> PathBuf {
    cfg.workdir.join(&aux.checkout_dir)
}

/// Every aux repo's checkout dir, keyed by name — what `GateCtx::aux_dirs` holds.
pub fn aux_checkout_dirs(
    cfg: &AppConfig,
    topo: &Topology,
) -> std::collections::HashMap<String, PathBuf> {
    topo.aux_repos
        .iter()
        .map(|a| (a.name.clone(), aux_checkout_dir(cfg, a)))
        .collect()
}

pub async fn checkout_aux_repos(cfg: &AppConfig, topo: &Topology) -> Result<()> {
    for aux in &topo.aux_repos {
        let bare = PathBuf::from(&aux.bare_path);
        git::ensure_bare_repo_no_hook(&bare)
            .await
            .with_context(|| format!("aux repo {}: init bare {}", aux.name, aux.bare_path))?;
        if let Err(e) = git::fetch_upstream(&bare, &aux.upstream, &aux.branch).await {
            tracing::warn!(
                aux = %aux.name, error = %e,
                "aux repo fetch failed; proceeding with current bare-repo state",
            );
        }
        let sha = git::resolve_ref(&bare, &aux.branch).await.with_context(|| {
            format!(
                "aux repo {}: branch {} not resolvable after fetch — is {} reachable with that branch?",
                aux.name, aux.branch, aux.upstream,
            )
        })?;
        let dest = aux_checkout_dir(cfg, aux);
        git::checkout_worktree(&bare, &sha, &dest)
            .await
            .with_context(|| {
                format!(
                    "aux repo {}: checking out {} ({}) at {}",
                    aux.name,
                    aux.branch,
                    sha,
                    dest.display()
                )
            })?;
        tracing::info!(
            aux = %aux.name, branch = %aux.branch, sha = %sha, dest = %dest.display(),
            "aux repo checked out beside worktree",
        );
    }
    Ok(())
}

/// Build one companion crate from the worktree, returning its release binary
/// path. Mirrors the server build's target-dir handling (shared
/// `cargo_target_dir` when set, for incremental reuse; else the crate's own
/// `target/`). A companion is an API client, not a sqlx crate, so it needs no
/// scratch DB. A non-zero exit propagates and fails the pipeline.
async fn build_companion(
    worktree: &Path,
    cfg: &AppConfig,
    c: &crate::config::Companion,
) -> Result<PathBuf> {
    let dir = worktree.join(&c.manifest_dir);
    anyhow::ensure!(
        dir.join("Cargo.toml").exists(),
        "companion {}: no Cargo.toml at {}",
        c.name,
        dir.display(),
    );
    // Match the server build: no `--locked` (the pipeline builds whatever the
    // sha pins; a stale lock shouldn't block a deploy the server build allows).
    let mut cmd = Command::new("cargo");
    cmd.arg("build")
        .arg("--release")
        .current_dir(&dir)
        .kill_on_drop(true);
    let release_dir = if let Some(target) = cfg.cargo_target_dir.as_deref() {
        cmd.env("CARGO_TARGET_DIR", target);
        target.join("release")
    } else {
        dir.join("target/release")
    };
    tracing::info!(companion = %c.name, dir = %dir.display(), "cargo build --release (companion) start");
    let started = std::time::Instant::now();
    let out = cmd
        .output()
        .await
        .context("spawning cargo build for companion")?;
    if !out.status.success() {
        anyhow::bail!(
            "companion {} build failed:\n{}",
            c.name,
            tail(&out.stderr, 4_000),
        );
    }
    let bin = release_dir.join(&c.bin);
    anyhow::ensure!(
        bin.exists(),
        "companion {} produced no binary at {} after build",
        c.name,
        bin.display(),
    );
    tracing::info!(companion = %c.name, elapsed_s = started.elapsed().as_secs(), "companion build ok");
    Ok(bin)
}

/// Full host-tier pipeline: build, stage the bundle into the host's
/// release_root, run the host tier's configured gates, advance tier_state
/// for "host" if all pass. Errors propagate back to the spawned task and
/// get logged. (Tier was called "mm" pre-Session-1; renamed to "host"
/// since sandod runs on whatever machine ends up being the Sando host.)
pub async fn build_and_run_host(
    pool: SqlitePool,
    cfg: Arc<AppConfig>,
    topo: Arc<Topology>,
    sha: GitSha,
    events: crate::events::EventTx,
    run_id: RunId,
    deploy_lock: Arc<tokio::sync::Mutex<()>>,
) -> Result<()> {
    let art = run(
        pool.clone(),
        cfg.clone(),
        topo.clone(),
        sha,
        events.clone(),
        run_id,
    )
    .await?;

    stage_and_gate(pool, cfg, topo, art, events, run_id, deploy_lock).await
}

/// A bundle assembled on disk and not yet published: the point both the build
/// path and the intake path have to reach before anything else can happen to it.
///
/// The two paths reach it differently. A build assembles it out of a worktree
/// (binaries, `release_contents`, companions); an intake is handed it already
/// assembled, with no worktree anywhere. Everything after this point (hashing,
/// publishing, recording identity, gating, advancing) is the same work for
/// both.
struct StagedBundle {
    version: Version,
    /// The staging dir under `release_root/staging/`, pre-publish.
    staging: PathBuf,
    /// What the bundle is built to run on, when it is known. A Sando build
    /// inherits the app's declared platform; an intake takes it from the
    /// record's provenance.
    platform: Option<Platform>,
}

/// Post-build half of the host pipeline: stage the artifact into the host's
/// release_root, run the host tier's gates, and advance `tier_state` for
/// "host" iff all pass. Split from [`build_and_run_host`] at the `run()`
/// boundary so the staging/gating/advance logic is reachable in tests from a
/// synthetic [`BuildArtifact`] — no real `cargo build --release` required.
pub async fn stage_and_gate(
    pool: SqlitePool,
    cfg: Arc<AppConfig>,
    topo: Arc<Topology>,
    art: BuildArtifact,
    events: crate::events::EventTx,
    run_id: RunId,
    deploy_lock: Arc<tokio::sync::Mutex<()>>,
) -> Result<()> {
    crate::runs::set_phase(&pool, run_id, crate::runs::Phase::Staging)
        .await
        .ok();
    let staged = assemble_from_source(&cfg, &art, run_id).await?;
    let published = publish(&pool, &cfg, staged, run_id).await?;
    record_and_gate(
        pool,
        cfg,
        topo,
        published,
        events,
        run_id,
        deploy_lock,
        Some(art.worktree),
    )
    .await
}

/// Accept an artifact built elsewhere and take it through the same host-tier
/// gating and advance a Sando-built one gets.
///
/// This is the whole point of the seam. `intake::accept` publishes the bundle
/// content-addressed once it has proved the bytes are the ones the record
/// vouches for, which lands it at exactly the state [`publish`] leaves a
/// Sando-built bundle in — so the two paths join at [`record_and_gate`] and
/// nothing downstream knows or cares which one it came from.
///
/// `staged` must already sit under `release_root/staging/` (publishing is an
/// atomic same-filesystem rename); getting the bytes there is the transport's
/// job, not this function's.
#[allow(clippy::too_many_arguments)]
pub async fn accept_intake(
    pool: &SqlitePool,
    cfg: &AppConfig,
    staged: &Path,
    record_json: &str,
    run_id: RunId,
) -> Result<Published> {
    crate::runs::set_phase(pool, run_id, crate::runs::Phase::Staging)
        .await
        .ok();

    let pinned = crate::retention::pinned_dirs(pool, &cfg.id).await?;
    let accepted = crate::intake::accept(&cfg.release_root, staged, record_json, &pinned)
        .await
        .map_err(|e| anyhow::anyhow!("{e}"))?;

    let version = Version::parse(&accepted.record.provenance.version).with_context(|| {
        format!(
            "artifact record carries version `{}`, which is not semver",
            accepted.record.provenance.version
        )
    })?;
    let platform = Platform::parse(&accepted.record.provenance.target).with_context(|| {
        format!(
            "artifact record carries target `{}`, which is not `os/arch`",
            accepted.record.provenance.target
        )
    })?;
    let git_sha = GitSha::parse(&accepted.record.provenance.git_sha).with_context(|| {
        format!(
            "artifact record carries git_sha `{}`",
            accepted.record.provenance.git_sha
        )
    })?;

    crate::runs::set_version(pool, run_id, &version).await.ok();
    upsert_version_row(
        pool,
        &cfg.id,
        &version,
        &git_sha,
        &accepted.released.join(cfg.primary_bin()),
    )
    .await?;

    let published = Published {
        version,
        released: accepted.released,
        digest_full: accepted.record.digest.to_string(),
        platform: Some(platform),
    };
    record_identity(pool, cfg, &published, run_id).await?;
    Ok(published)
}

/// Gate an artifact that has already been accepted.
///
/// Split from [`accept_intake`] so the two can be answered on different clocks.
/// Acceptance is fast and is the producer's business — it either believes the
/// bytes or names the file that drifted — so the caller waits for it and gets
/// the verdict. Gating is Sando's business and can take an hour, so the caller
/// does not.
///
/// An intake carries no worktree, and the gates that need one refuse rather than
/// resolve against nothing. That is the boundary showing up in the type:
/// artifact-scoped gates belong to the builder (wiki [[sando-bento-boundary]]),
/// so a tier that asks Sando to re-run them against an accepted artifact is
/// misconfigured and should be told so.
pub async fn gate_intake(
    pool: SqlitePool,
    cfg: Arc<AppConfig>,
    topo: Arc<Topology>,
    published: Published,
    events: crate::events::EventTx,
    run_id: RunId,
    deploy_lock: Arc<tokio::sync::Mutex<()>>,
) -> Result<()> {
    record_and_gate(
        pool,
        cfg,
        topo,
        published,
        events,
        run_id,
        deploy_lock,
        None,
    )
    .await
}

/// Record the `versions` label row for an artifact that arrived rather than was
/// built here. The build path writes its own inside [`run`]; this is the same
/// row for the path that never ran a compiler.
async fn upsert_version_row(
    pool: &SqlitePool,
    app: &crate::domain::AppId,
    version: &Version,
    git_sha: &GitSha,
    artifact_path: &Path,
) -> Result<()> {
    sqlx::query(
        "INSERT OR IGNORE INTO versions (app, version, git_sha, built_at, artifact_path)
         VALUES (?, ?, ?, ?, ?)",
    )
    .bind(app)
    .bind(version)
    .bind(git_sha)
    .bind(Utc::now().to_rfc3339())
    .bind(artifact_path.to_string_lossy().as_ref())
    .execute(pool)
    .await?;
    Ok(())
}

/// Assemble a bundle out of a worktree: binaries, `release_contents`, companions.
async fn assemble_from_source(
    cfg: &AppConfig,
    art: &BuildArtifact,
    run_id: RunId,
) -> Result<StagedBundle> {
    // Stage the bundle into `staging/<build_id>/` — a private scratch dir, not
    // yet a release. It is published content-addressed below, once its digest is
    // known. This is what makes overwrite unexpressible (wiki
    // [[release-artifact-identity]]): a build never touches another build's dir.
    let staging =
        deploy::stage_local_bundle(&cfg.release_root, run_id.0, &art.binary_paths).await?;

    // Stage every entry from cfg.release_contents into the staged bundle. This is
    // how non-binary version-coupled content (static assets, docs, error-pages,
    // ...) makes it into the atomic deploy bundle. Projects opt in via daemon
    // config — the sando code carries no MNW-specific knowledge.
    for entry in &cfg.release_contents {
        stage_entry(&art.worktree, &staging, entry).await?;
    }

    // Stage companion binaries as `companions/<name>` (the file itself) so they
    // ride the same atomic bundle rsync to the nodes, and a node can locate its
    // companion source from the logical name alone — no bin-filename coupling in
    // the topology. The nodes that opt in install them post-swap (see
    // deploy::deploy_remote).
    if !art.companion_paths.is_empty() {
        let dst_dir = staging.join("companions");
        tokio::fs::create_dir_all(&dst_dir)
            .await
            .with_context(|| format!("create staged companions dir {}", dst_dir.display()))?;
        for (name, built) in &art.companion_paths {
            let dst = dst_dir.join(name);
            tokio::fs::copy(built, &dst).await.with_context(|| {
                format!(
                    "stage companion {name}: {} -> {}",
                    built.display(),
                    dst.display()
                )
            })?;
        }
    }

    Ok(StagedBundle {
        version: art.version.clone(),
        staging,
        platform: cfg.platform.clone(),
    })
}

/// A bundle that has been hashed and published content-addressed. Both paths
/// produce one; nothing downstream can tell them apart.
///
/// Public because the intake route now hands one from `accept_intake` to
/// `gate_intake`: proving the bytes answers the producer, gating them does not,
/// so the two run on different clocks and the value passes between them.
#[derive(Debug)]
pub struct Published {
    version: Version,
    released: PathBuf,
    digest_full: String,
    platform: Option<Platform>,
}

/// Hash the assembled bundle, write its MANIFEST, and publish it at
/// `releases/<digest16>`.
///
/// The intake path does not call this: `intake::accept` does the same three
/// steps itself, because it has to hash the bytes to verify them and hashing
/// them twice would be the one place the two implementations could disagree.
async fn publish(
    pool: &SqlitePool,
    cfg: &AppConfig,
    staged: StagedBundle,
    run_id: RunId,
) -> Result<Published> {
    // Content identity: hash the fully-staged bundle, write its MANIFEST into the
    // bundle (for node-side verification), then publish it at `releases/<digest16>`.
    // The digest is now load-bearing — a hashing failure fails the build rather
    // than shipping an unidentifiable artifact.
    let digest = crate::bundle::digest_dir(&staged.staging)
        .await
        .context("hashing the staged bundle for content addressing")?;
    tokio::fs::write(
        staged.staging.join(crate::bundle::MANIFEST_NAME),
        digest.manifest.as_bytes(),
    )
    .await
    .context("writing bundle MANIFEST")?;
    let pinned = crate::retention::pinned_dirs(pool, &cfg.id).await?;
    let released =
        deploy::finalize_local_release(&cfg.release_root, &staged.staging, digest.short(), &pinned)
            .await?;

    let staged_bin = released.join(cfg.primary_bin());
    sqlx::query("UPDATE versions SET artifact_path = ? WHERE app = ? AND version = ?")
        .bind(staged_bin.to_string_lossy().as_ref())
        .bind(&cfg.id)
        .bind(&staged.version)
        .execute(pool)
        .await?;

    let published = Published {
        version: staged.version,
        released,
        digest_full: digest.full,
        platform: staged.platform,
    };
    record_identity(pool, cfg, &published, run_id).await?;
    Ok(published)
}

/// Record the identity on the build row: the digest, the content-addressed dir
/// the bundle was published to, and what it runs on. This is what promote
/// resolves the artifact through, and burn-in/retention key on.
async fn record_identity(
    pool: &SqlitePool,
    cfg: &AppConfig,
    published: &Published,
    run_id: RunId,
) -> Result<()> {
    let released_path = published.released.to_string_lossy();
    crate::runs::set_identity(pool, run_id, &published.digest_full, &released_path)
        .await
        .ok();
    // Platform is what lets two bundles of one version be told apart, so a
    // dropped write here would leave a pom artifact that can be placed nowhere
    // (a node declaring a platform refuses an artifact that records none). Fail
    // rather than ship an unplaceable bundle.
    if let Some(p) = &published.platform {
        crate::runs::set_platform(pool, run_id, p)
            .await
            .with_context(|| format!("recording platform {p} for {}", cfg.id))?;
    }
    Ok(())
}

/// The shared tail of both paths: run the host tier's gates against a published
/// bundle and advance `tier_state` iff all pass.
#[allow(clippy::too_many_arguments)]
async fn record_and_gate(
    pool: SqlitePool,
    cfg: Arc<AppConfig>,
    topo: Arc<Topology>,
    published: Published,
    events: crate::events::EventTx,
    run_id: RunId,
    deploy_lock: Arc<tokio::sync::Mutex<()>>,
    worktree: Option<PathBuf>,
) -> Result<()> {
    let host = topo
        .tiers
        .iter()
        .find(|t| t.name.as_str() == "host")
        .context("topology has no `host` tier")?;

    crate::runs::set_phase(&pool, run_id, crate::runs::Phase::Gating)
        .await
        .ok();
    let ctx = GateCtx {
        pool: pool.clone(),
        cfg: cfg.clone(),
        tier: TierId::new("host"),
        version: published.version.clone(),
        worktree,
        // The published bundle. `migration_dry_run` prefers it over the
        // worktree, so the migrations it proves are the ones inside the digest
        // rather than ones sitting beside them in a checkout.
        bundle: Some(published.released.clone()),
        events: events.clone(),
        // Host runs build-time gates (cargo_test / migration_dry_run /
        // boot_smoke) only — `node_health` never appears here, so there are no
        // nodes to probe.
        nodes: Vec::new(),
        // These gates vouch for this build; record its id so promote can resolve
        // the artifact through the evidence rather than a version string.
        build_id: Some(run_id.0),
        // Where checkout_aux_repos put each aux repo, so a test_target naming
        // one resolves. Shared derivation, so the two cannot disagree.
        public_url: None,
        aux_dirs: aux_checkout_dirs(&cfg, &topo),
    };
    let failed = gates::run_all(&ctx, &host.gates).await?;

    if failed.is_empty() {
        // Advance the host tier through the single sealed forward-advance op, under
        // deploy_lock so this can't interleave with a concurrent `/rollback host`
        // (the old fetch-then-write here was the one CF3 site outside the lock —
        // ultra-fuzz Run 2, S1). Held only for the atomic UPDATE, never the gates.
        {
            let _deploy_guard = deploy_lock.lock().await;
            crate::runs::advance_tier(&pool, &cfg.id, "host", &published.version, Some(run_id.0))
                .await?;
        }
        // Terminal verdict: unlike the phase pings above (best-effort), a dropped
        // pass/fail write leaves the run wedged at `building`. Log it loudly if it
        // fails — the startup reconcile (main) is the backstop that settles such a
        // row on the next restart.
        if let Err(e) = crate::runs::mark_passed(&pool, run_id).await {
            tracing::error!(run_id = %run_id, error = %e, "persisting host-green verdict failed; run may show stale 'building' until restart-reconcile");
        }
        tracing::info!(version = %published.version, "host pipeline green; ready to promote to next tier");
    } else {
        // Pull the first red gate's typed summary into the run so the API
        // answers "which gate, and why" — not just "failed".
        let summary = crate::runs::first_failed_gate_summary(&pool, &cfg.id, run_id)
            .await
            .unwrap_or_else(|| "host pipeline red".to_string());
        if let Err(e) = crate::runs::mark_failed(&pool, run_id, &summary).await {
            tracing::error!(run_id = %run_id, error = %e, "persisting host-red verdict failed; run may show stale 'building' until restart-reconcile");
        }
        tracing::warn!(version = %published.version, "host pipeline red; not advancing tier_state");
    }
    Ok(())
}

async fn read_pkg_version(cargo_toml: &Path) -> Result<Version> {
    let raw = tokio::fs::read_to_string(cargo_toml).await?;
    let parsed: toml::Value = toml::from_str(&raw)?;
    let v = parsed
        .get("package")
        .and_then(|p| p.get("version"))
        .and_then(|v| v.as_str())
        .context("package.version not found")?;
    Version::parse(v).with_context(|| format!("parsing package.version `{v}`"))
}

fn tail(buf: &[u8], max: usize) -> String {
    let s = String::from_utf8_lossy(buf);
    if s.len() <= max {
        return s.into_owned();
    }
    // `s.len() - max` can land mid-codepoint; walk forward to the next char
    // boundary so the slice never panics (returns slightly fewer than `max`
    // bytes in that case). `floor_char_boundary` is still unstable, so do it by
    // hand.
    let mut start = s.len() - max;
    while start < s.len() && !s.is_char_boundary(start) {
        start += 1;
    }
    s[start..].to_string()
}

/// Copy `worktree/<entry.src>` into `staged/<entry.dst>`. Handles file or
/// directory sources transparently. Missing source policy depends on
/// `entry.required`:
/// - required=true  -> error (build fails)
/// - required=false -> log warn + skip (e.g. older shas missing a dir)
///
/// Uses `cp -a` to preserve modes/symlinks/etc; parent of dst is created if
/// needed so entries like `dst = "docs/assumptions.toml"` work without
/// extra config.
async fn stage_entry(
    worktree: &Path,
    staged: &Path,
    entry: &crate::config::ReleaseEntry,
) -> Result<()> {
    let src = worktree.join(&entry.src);
    let dst = staged.join(&entry.dst);
    if !src.exists() {
        if entry.required {
            anyhow::bail!(
                "required release_contents source missing: {}",
                src.display()
            );
        }
        tracing::warn!(src = %src.display(), "release_contents source missing (optional); skipping");
        return Ok(());
    }
    if let Some(parent) = dst.parent() {
        tokio::fs::create_dir_all(parent)
            .await
            .with_context(|| format!("create staged parent {}", parent.display()))?;
    }
    // Multiple entries with the same dst (e.g. site-docs/public/ +
    // site-docs/examples/ both landing under docs/) need additive merging.
    // `cp -a SRC/. DST/` copies SRC's contents into DST without overwriting
    // the dst dir itself; that's the merge-friendly form when dst is a dir
    // that may already exist from a prior entry. For non-dir sources or a
    // missing dst we fall back to the plain `cp -a SRC DST` form.
    let merge_into_existing_dir = src.is_dir() && dst.is_dir();
    let mut cmd = Command::new("cp");
    cmd.arg("-a");
    if merge_into_existing_dir {
        let mut src_arg = src.clone().into_os_string();
        src_arg.push("/.");
        cmd.arg(src_arg);
        let mut dst_arg = dst.clone().into_os_string();
        dst_arg.push("/");
        cmd.arg(dst_arg);
    } else {
        cmd.arg(&src).arg(&dst);
    }
    let out = cmd
        .output()
        .await
        .with_context(|| format!("spawning cp for {} -> {}", src.display(), dst.display()))?;
    anyhow::ensure!(
        out.status.success(),
        "stage {} -> {}: {}",
        src.display(),
        dst.display(),
        String::from_utf8_lossy(&out.stderr),
    );
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        BuildArtifact, accept_intake, check_build_host, checkout_aux_repos, gate_intake,
        runtime_hostname, stage_and_gate, tail,
    };
    use crate::config::{AppConfig, TestTarget};
    use crate::domain::{GitSha, RunId, Version};
    use crate::topology::{AuxRepo, BackupConfig, CanaryPolicy, Gate, RepoConfig, Tier, Topology};
    use sqlx::SqlitePool;
    use sqlx::sqlite::SqlitePoolOptions;
    use std::collections::BTreeMap;
    use std::path::PathBuf;
    use std::sync::Arc;

    /// Post-build pipeline fixture: an in-memory store with the `host` tier
    /// seeded, a synthetic worktree holding a fake primary binary, and a
    /// build_runs row in flight. Returns everything `stage_and_gate` needs plus
    /// the tempdir root (drop it to clean up) and the run/version it seeded.
    ///
    /// `gates` is the host tier's gate list: `[]` is the green path;
    /// `[Gate::ManualConfirm]` is a deterministic red — with no prior operator
    /// confirmation row that gate blocks, and it shells out to nothing.
    async fn stage_fixture(
        gates: Vec<Gate>,
    ) -> (
        SqlitePool,
        Arc<AppConfig>,
        Arc<Topology>,
        BuildArtifact,
        RunId,
        Version,
        tempfile::TempDir,
    ) {
        let tmp = tempfile::tempdir().unwrap();
        let release_root = tmp.path().join("release-root");
        let worktree = tmp.path().join("worktree");
        let bin_dir = worktree.join("target").join("release");
        tokio::fs::create_dir_all(&bin_dir).await.unwrap();
        let bin_path = bin_dir.join("makenotwork");
        tokio::fs::write(&bin_path, b"#!/bin/false\nfake sando artifact\n")
            .await
            .unwrap();

        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();

        // gate_runs and tier_state FK into `tiers`; the pipeline only touches host.
        sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 1, 'sequential')")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
            .execute(&pool)
            .await
            .unwrap();

        let version = Version::parse("1.2.3").unwrap();
        let git_sha = GitSha::parse("abc1234").unwrap();
        // gate_runs.version and the `SET artifact_path` UPDATE both need the row.
        sqlx::query(
            "INSERT INTO versions (version, git_sha, built_at, artifact_path)
             VALUES (?, ?, datetime('now'), '')",
        )
        .bind(version.to_string())
        .bind(git_sha.to_string())
        .execute(&pool)
        .await
        .unwrap();

        let run_id = crate::runs::create(
            &pool,
            &crate::domain::AppId::default(),
            &git_sha.to_string(),
        )
        .await
        .unwrap();

        let cfg = AppConfig {
            page_smoke_cmd: None,
            platform: None,
            code_smoke_env: BTreeMap::default(),
            id: crate::domain::AppId::default(),
            topology_path: PathBuf::from("/tmp/test-sando.toml"),
            build_host: Some("test-host".into()),
            workdir: tmp.path().to_path_buf(),
            release_root: release_root.clone(),
            scratch_db_url: None,
            scratch_owner_role: "makenotwork".into(),
            boot_smoke_port: 18181,
            code_smoke_port: 18182,
            bin_names: vec!["makenotwork".into()],
            logs_root: tmp.path().join("logs"),
            release_contents: vec![],
            cargo_target_dir: None,
            gate_timeout_secs: 2400,
            companions: Vec::new(),
            test_targets: vec![TestTarget {
                dir: PathBuf::from("server"),
                aux_repo: None,
                features: vec!["fast-tests".into()],
                all_features: false,
                scratch_db: true,
            }],
            migration_checks: vec![],
            frontend_builds: vec![],
            backup_max_age_hours: 48,
        };

        let topo = Topology {
            repo: Some(RepoConfig {
                bare_path: "/tmp/test.git".into(),
                branch: "main".into(),
                upstream: None,
            }),
            backup: vec![BackupConfig {
                name: "server".into(),
                source: "file:///tmp/test-backup.sql".into(),
                local_path: "/tmp/local-backup.sql".into(),
            }],
            tiers: vec![Tier {
                public_url: None,
                name: "host".into(),
                provisioned: true,
                gates,
                canary: CanaryPolicy::Sequential,
                nodes: Vec::new(),
            }],
            aux_repos: Vec::new(),
        };

        let art = BuildArtifact {
            version: version.clone(),
            git_sha,
            worktree,
            binary_paths: vec![bin_path],
            companion_paths: Vec::new(),
        };

        (
            pool,
            Arc::new(cfg),
            Arc::new(topo),
            art,
            run_id,
            version,
            tmp,
        )
    }

    // ---- intake through the seam ----

    /// The `ArtifactRecord` Bento would have written for a staged bundle.
    async fn record_for(staged: &std::path::Path, version: &str, target: &str) -> String {
        use ops_artifact::{ArtifactRecord, GateRecord, Manifest, Provenance, Scope, Verdict};
        let computed = crate::bundle::digest_dir(staged).await.unwrap();
        let manifest = Manifest::parse(&computed.manifest).unwrap();
        let at = chrono::DateTime::<chrono::Utc>::from_timestamp(1_754_000_000, 0).unwrap();
        ArtifactRecord::new(
            "bento",
            manifest,
            Provenance {
                app: "pom".into(),
                version: version.into(),
                tag: format!("pom-v{version}"),
                git_sha: "a".repeat(40),
                target: target.into(),
                build_host: "astra".into(),
                toolchain: "rustc 1.97.0".into(),
                built_at: at,
            },
            vec![GateRecord::new(
                "prebuild",
                Scope::Artifact,
                Verdict::Passed,
                "prebuild passed in 90s",
                at,
            )],
        )
        .unwrap()
        .to_json()
    }

    #[tokio::test]
    async fn an_accepted_artifact_is_published_gated_and_advances_the_tier() {
        // The seam: an artifact Sando did not build reaches the same published,
        // gated, tier-advanced end state a Sando-built one does. No worktree
        // exists anywhere in this test, which is the point — everything from
        // `finalize_local_release` onward stopped caring where the bytes came
        // from.
        let (pool, cfg, topo, _art, run_id, _version, tmp) = stage_fixture(vec![]).await;
        let deploy_lock = Arc::new(tokio::sync::Mutex::new(()));

        let staged = cfg.release_root.join("staging").join("intake-1");
        tokio::fs::create_dir_all(&staged).await.unwrap();
        tokio::fs::write(staged.join("makenotwork"), b"bytes built elsewhere")
            .await
            .unwrap();
        let record = record_for(&staged, "1.2.3", "linux/aarch64").await;

        // Two calls now, on purpose: the route answers its caller on the first
        // and spawns the second. Acceptance is what the producer waits for.
        let published = accept_intake(&pool, &cfg, &staged, &record, run_id)
            .await
            .expect("the bytes are believed");
        gate_intake(
            pool.clone(),
            cfg.clone(),
            topo,
            published,
            crate::events::channel(),
            run_id,
            deploy_lock,
        )
        .await
        .expect("a green intake settles the run");

        // Published content-addressed, and the staging dir is gone: renamed,
        // not copied.
        let (digest, staged_path, platform): (Option<String>, Option<String>, Option<String>) =
            sqlx::query_as(
                "SELECT bundle_digest, staged_path, platform FROM build_runs WHERE id = ?",
            )
            .bind(run_id.0)
            .fetch_one(&pool)
            .await
            .unwrap();
        let digest = digest.expect("bundle_digest recorded");
        let staged_path = staged_path.expect("staged_path recorded");
        assert_eq!(digest.len(), 64);
        assert!(
            !staged.exists(),
            "staging was renamed into the release root"
        );
        assert_eq!(
            std::path::Path::new(&staged_path),
            tmp.path()
                .join("release-root")
                .join("releases")
                .join(&digest[..16]),
        );

        // The platform came off the record's provenance and is on the row. This
        // is what makes two bundles of one version tellable apart later.
        assert_eq!(platform.as_deref(), Some("linux/aarch64"));

        // Green gates advanced the host tier, exactly as a build would have.
        let (result, _summary) = run_result(&pool, run_id).await;
        assert_eq!(result, "passed");
        let (current, _prev) = tier_versions(&pool, "host").await;
        assert_eq!(current.as_deref(), Some("1.2.3"));
    }

    #[tokio::test]
    async fn an_intake_whose_bytes_drifted_never_reaches_the_gates() {
        // Identity is decided before anything else happens to the bundle, so a
        // record vouching for one set of bytes arriving with another fails the
        // run rather than gating and shipping.
        let (pool, cfg, topo, _art, run_id, _version, _tmp) = stage_fixture(vec![]).await;
        let deploy_lock = Arc::new(tokio::sync::Mutex::new(()));

        let staged = cfg.release_root.join("staging").join("intake-1");
        tokio::fs::create_dir_all(&staged).await.unwrap();
        tokio::fs::write(staged.join("makenotwork"), b"bytes built elsewhere")
            .await
            .unwrap();
        let record = record_for(&staged, "1.2.3", "linux/aarch64").await;
        tokio::fs::write(staged.join("makenotwork"), b"other bytes entirely")
            .await
            .unwrap();

        // Refused by ACCEPTANCE, not by gating — which is what lets the route
        // answer the producer with the refusal instead of a `202`-shaped lie.
        let _ = (&topo, &deploy_lock);
        let err = accept_intake(&pool, &cfg, &staged, &record, run_id)
            .await
            .expect_err("a drifted bundle is refused");
        assert!(err.to_string().contains("makenotwork"), "{err}");

        // Nothing advanced, and the bytes were left where they were.
        let (current, _prev) = tier_versions(&pool, "host").await;
        assert_eq!(current, None);
        assert!(staged.exists(), "a refused intake leaves the bytes alone");
    }

    #[tokio::test]
    async fn a_gate_that_reads_source_refuses_against_an_accepted_artifact() {
        // The boundary showing up at runtime. `code_smoke` is artifact-scoped —
        // it compiles frontends and boots the binary against a scratch DB — and
        // an accepted artifact has no checkout for it to read. It has to say so
        // rather than pass on having run nothing, which is what an unwrapped
        // `worktree.join(..)` against an empty path would have done.
        let (pool, cfg, _topo, _art, run_id, version, _tmp) = stage_fixture(vec![]).await;
        let ctx = crate::gates::GateCtx {
            pool,
            cfg,
            tier: crate::domain::TierId::new("host"),
            version,
            worktree: None,
            bundle: Some(PathBuf::from("/r/abc")),
            events: crate::events::channel(),
            nodes: Vec::new(),
            build_id: Some(run_id.0),
            public_url: None,
            aux_dirs: std::collections::HashMap::default(),
        };
        let outcome = ctx
            .worktree_for(crate::domain::GateKind::CodeSmoke)
            .expect_err("no worktree means no source-reading gate");
        assert!(!outcome.is_passed());
        let crate::outcome::GateStatus::Failed { failure } = &outcome.status else {
            panic!("expected a failure, got {:?}", outcome.status)
        };
        assert!(
            matches!(failure, crate::outcome::GateFailure::NeedsSource { .. }),
            "{failure:?}"
        );
        assert!(
            failure.summary().contains("built elsewhere"),
            "{}",
            failure.summary()
        );
    }

    #[tokio::test]
    async fn migrations_come_from_the_bundle_before_the_worktree() {
        // What the gate proves has to be what ships. Migrations staged into the
        // bundle are inside its digest; the same files sitting in a checkout are
        // not, and a checkout can be edited between the dry run and the deploy.
        // So when both hold a copy, the bundle wins.
        let (pool, cfg, _topo, _art, run_id, version, tmp) = stage_fixture(vec![]).await;
        let bundle = tmp.path().join("bundle");
        let worktree = tmp.path().join("wt");
        for root in [&bundle, &worktree] {
            tokio::fs::create_dir_all(root.join("server/migrations"))
                .await
                .unwrap();
        }
        let ctx = crate::gates::GateCtx {
            pool,
            cfg,
            tier: crate::domain::TierId::new("host"),
            version,
            worktree: Some(worktree.clone()),
            bundle: Some(bundle.clone()),
            events: crate::events::channel(),
            nodes: Vec::new(),
            build_id: Some(run_id.0),
            public_url: None,
            aux_dirs: std::collections::HashMap::default(),
        };
        assert_eq!(
            ctx.migrations_dir(std::path::Path::new("server/migrations")),
            Some(bundle.join("server/migrations")),
        );

        // The worktree is the fallback for a build whose config has not opted
        // into bundling them yet, which is every MNW build before this lands.
        let ctx = crate::gates::GateCtx {
            bundle: Some(tmp.path().join("empty-bundle")),
            ..ctx
        };
        assert_eq!(
            ctx.migrations_dir(std::path::Path::new("server/migrations")),
            Some(worktree.join("server/migrations")),
        );

        // And neither is not silently green: an accepted artifact whose builder
        // did not bundle its migrations has nothing to dry-run, and the gate
        // has to be told so rather than restore a dump and report success.
        let ctx = crate::gates::GateCtx {
            worktree: None,
            ..ctx
        };
        assert_eq!(
            ctx.migrations_dir(std::path::Path::new("server/migrations")),
            None
        );
    }

    // ---- checkout_aux_repos ----

    async fn git_in(dir: &std::path::Path, args: &[&str]) {
        let out = tokio::process::Command::new("git")
            .args(["-c", "user.email=t@t", "-c", "user.name=t"])
            .current_dir(dir)
            .args(args)
            .output()
            .await
            .unwrap();
        assert!(
            out.status.success(),
            "git {args:?}: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    }

    /// A minimal `Config` whose only field this test path reads is `workdir`.
    fn cfg_with_workdir(workdir: PathBuf) -> AppConfig {
        AppConfig {
            page_smoke_cmd: None,
            platform: None,
            code_smoke_env: BTreeMap::default(),
            id: crate::domain::AppId::default(),
            topology_path: PathBuf::from("/tmp/test-sando.toml"),
            build_host: Some("test-host".into()),
            workdir,
            release_root: PathBuf::from("/tmp/rr"),
            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/logs"),
            release_contents: vec![],
            cargo_target_dir: None,
            gate_timeout_secs: 2400,
            companions: Vec::new(),
            test_targets: vec![],
            migration_checks: vec![],
            frontend_builds: vec![],
            backup_max_age_hours: 48,
        }
    }

    fn topo_with_aux(aux_repos: Vec<AuxRepo>) -> Topology {
        Topology {
            repo: Some(RepoConfig {
                bare_path: "/tmp/x.git".into(),
                branch: "main".into(),
                upstream: None,
            }),
            backup: vec![BackupConfig {
                name: "server".into(),
                source: "s".into(),
                local_path: "/tmp/d".into(),
            }],
            tiers: vec![],
            aux_repos,
        }
    }

    #[tokio::test]
    async fn a_gate_looks_where_the_aux_checkout_actually_landed() {
        // The two halves of the aux-repo test_target path: checkout_aux_repos
        // writes the tree, and GateCtx::target_dir reads it. Nothing but this
        // stops one from being changed without the other, and the failure would
        // be a warn-and-skip — a green gate that ran one crate fewer.
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("docengine-src");
        tokio::fs::create_dir_all(&src).await.unwrap();
        git_in(&src, &["init", "-q", "-b", "main"]).await;
        tokio::fs::write(src.join("Cargo.toml"), b"[package]\nname = \"docengine\"\n")
            .await
            .unwrap();
        git_in(&src, &["add", "."]).await;
        git_in(&src, &["commit", "-q", "-m", "one"]).await;

        let workdir = tmp.path().join("work");
        tokio::fs::create_dir_all(&workdir).await.unwrap();
        let cfg = cfg_with_workdir(workdir.clone());
        let topo = topo_with_aux(vec![AuxRepo {
            name: "docengine".into(),
            bare_path: tmp
                .path()
                .join("docengine.git")
                .to_string_lossy()
                .into_owned(),
            upstream: src.to_string_lossy().into_owned(),
            branch: "main".into(),
            // Nested, as the real one is: it must not be mistaken for a path
            // under the per-sha worktree.
            checkout_dir: "Libraries/docengine".into(),
        }]);
        checkout_aux_repos(&cfg, &topo).await.unwrap();

        let ctx = crate::gates::GateCtx {
            pool: sqlx::SqlitePool::connect_lazy("sqlite::memory:").unwrap(),
            cfg: Arc::new(cfg.clone()),
            tier: crate::domain::TierId::new("host"),
            version: "0.1.0".parse().unwrap(),
            worktree: Some(workdir.join("abc123")),
            bundle: None,
            events: crate::events::channel(),
            nodes: Vec::new(),
            build_id: None,
            public_url: None,
            aux_dirs: super::aux_checkout_dirs(&cfg, &topo),
        };
        let target = crate::config::TestTarget {
            dir: PathBuf::new(),
            aux_repo: Some("docengine".into()),
            features: Vec::new(),
            all_features: true,
            scratch_db: false,
        };
        let resolved = ctx.target_dir(&target).expect("aux repo is checked out");
        assert!(
            resolved.join("Cargo.toml").is_file(),
            "gate would skip the aux target as absent; resolved {}",
            resolved.display(),
        );
        assert!(
            !resolved.starts_with(ctx.worktree.as_ref().unwrap()),
            "an aux checkout is a sibling of the worktree, not under it",
        );
    }

    #[tokio::test]
    async fn checkout_aux_repos_places_repo_beside_worktree_and_refreshes_to_branch_head() {
        let tmp = tempfile::tempdir().unwrap();

        // An "upstream" source repo with a marker file on main.
        let src = tmp.path().join("synckit-src");
        tokio::fs::create_dir_all(&src).await.unwrap();
        git_in(&src, &["init", "-q", "-b", "main"]).await;
        tokio::fs::write(src.join("VERSION"), b"v1").await.unwrap();
        git_in(&src, &["add", "."]).await;
        git_in(&src, &["commit", "-q", "-m", "one"]).await;

        let workdir = tmp.path().join("work");
        tokio::fs::create_dir_all(&workdir).await.unwrap();
        let cfg = cfg_with_workdir(workdir.clone());
        let topo = topo_with_aux(vec![AuxRepo {
            name: "synckit".into(),
            bare_path: tmp
                .path()
                .join("synckit.git")
                .to_string_lossy()
                .into_owned(),
            upstream: src.to_string_lossy().into_owned(),
            branch: "main".into(),
            checkout_dir: "synckit".into(),
        }]);

        // First build: the aux repo lands at workdir/synckit at v1.
        checkout_aux_repos(&cfg, &topo).await.unwrap();
        let dest = workdir.join("synckit");
        assert_eq!(
            tokio::fs::read(dest.join("VERSION")).await.unwrap(),
            b"v1",
            "aux repo checked out beside the worktree",
        );

        // Upstream advances; a later build refreshes the shared checkout to HEAD.
        tokio::fs::write(src.join("VERSION"), b"v2").await.unwrap();
        git_in(&src, &["add", "."]).await;
        git_in(&src, &["commit", "-q", "-m", "two"]).await;
        checkout_aux_repos(&cfg, &topo).await.unwrap();
        assert_eq!(
            tokio::fs::read(dest.join("VERSION")).await.unwrap(),
            b"v2",
            "aux checkout refreshed to the new branch HEAD",
        );

        // The aux bare carries no build-trigger hook.
        assert!(
            !tmp.path().join("synckit.git/hooks/post-receive").exists(),
            "aux bare must be hookless",
        );
    }

    #[tokio::test]
    async fn checkout_aux_repos_is_a_noop_without_aux_repos() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = cfg_with_workdir(tmp.path().to_path_buf());
        checkout_aux_repos(&cfg, &topo_with_aux(vec![]))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn checkout_aux_repos_fails_on_an_unresolvable_branch() {
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src");
        tokio::fs::create_dir_all(&src).await.unwrap();
        git_in(&src, &["init", "-q", "-b", "main"]).await;
        tokio::fs::write(src.join("f"), b"x").await.unwrap();
        git_in(&src, &["add", "."]).await;
        git_in(&src, &["commit", "-q", "-m", "c"]).await;

        let cfg = cfg_with_workdir(tmp.path().join("work"));
        let topo = topo_with_aux(vec![AuxRepo {
            name: "synckit".into(),
            bare_path: tmp.path().join("s.git").to_string_lossy().into_owned(),
            upstream: src.to_string_lossy().into_owned(),
            branch: "nonexistent".into(),
            checkout_dir: "synckit".into(),
        }]);
        let err = checkout_aux_repos(&cfg, &topo).await.unwrap_err();
        assert!(
            format!("{err:#}").contains("synckit"),
            "error names the aux repo: {err:#}",
        );
    }

    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()
    }

    async fn run_result(pool: &SqlitePool, run_id: RunId) -> (String, Option<String>) {
        sqlx::query_as("SELECT result, failure_summary FROM build_runs WHERE id = ?")
            .bind(run_id.0)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn stage_and_gate_stages_advances_and_flips_the_symlink_when_gates_are_green() {
        let (pool, cfg, topo, art, run_id, version, tmp) = stage_fixture(vec![]).await;
        let deploy_lock = Arc::new(tokio::sync::Mutex::new(()));

        stage_and_gate(
            pool.clone(),
            cfg.clone(),
            topo,
            art,
            crate::events::channel(),
            run_id,
            deploy_lock,
        )
        .await
        .expect("green host pipeline returns Ok");

        // Tier advanced to the built version (previous was NULL -> stays NULL).
        let (current, previous) = tier_versions(&pool, "host").await;
        assert_eq!(current.as_deref(), Some(version.to_string().as_str()));
        assert_eq!(previous, None);

        // Run settled green.
        let (result, summary) = run_result(&pool, run_id).await;
        assert_eq!(result, "passed");
        assert_eq!(summary, None);

        // Identity: the build row carries the bundle digest (64 hex) and the
        // content-addressed dir it was published to (releases/<digest16>).
        let (digest, staged_path): (Option<String>, Option<String>) =
            sqlx::query_as("SELECT bundle_digest, staged_path FROM build_runs WHERE id = ?")
                .bind(run_id.0)
                .fetch_one(&pool)
                .await
                .unwrap();
        let digest = digest.expect("bundle_digest recorded");
        let staged_path = staged_path.expect("staged_path recorded");
        assert_eq!(digest.len(), 64);
        let releases = tmp.path().join("release-root").join("releases");
        assert_eq!(
            std::path::Path::new(&staged_path),
            releases.join(&digest[..16]),
            "bundle is published content-addressed at releases/<digest16>"
        );

        // versions.artifact_path points at the primary binary inside that dir,
        // and it exists on disk.
        let staged_bin: String =
            sqlx::query_scalar("SELECT artifact_path FROM versions WHERE version = ?")
                .bind(version.to_string())
                .fetch_one(&pool)
                .await
                .unwrap();
        let expected_bin = releases.join(&digest[..16]).join("makenotwork");
        assert_eq!(staged_bin, expected_bin.to_string_lossy());
        assert!(
            expected_bin.exists(),
            "staged binary missing at {expected_bin:?}"
        );

        // The bundle carries its MANIFEST (for node-side verification), and the
        // recorded digest recomputes over the published dir (MANIFEST excluded).
        assert!(
            releases.join(&digest[..16]).join("MANIFEST").exists(),
            "MANIFEST written into the bundle"
        );
        let recomputed = crate::bundle::digest_dir(std::path::Path::new(&staged_path))
            .await
            .unwrap();
        assert_eq!(
            digest, recomputed.full,
            "recorded digest matches the bundle"
        );

        // The `current` symlink flipped to the content-addressed release.
        let link = tmp.path().join("release-root").join("current");
        let target = std::fs::read_link(&link).expect("current is a symlink");
        assert_eq!(target, PathBuf::from(format!("releases/{}", &digest[..16])));
    }

    #[tokio::test]
    async fn stage_and_gate_marks_the_run_failed_and_does_not_advance_when_a_gate_is_red() {
        // ManualConfirm with no prior confirmation row blocks deterministically.
        let (pool, cfg, topo, art, run_id, _version, _tmp) =
            stage_fixture(vec![Gate::ManualConfirm]).await;
        let deploy_lock = Arc::new(tokio::sync::Mutex::new(()));

        // A red gate is a pipeline outcome, not an error: the fn records the
        // failure and returns Ok so the spawned task settles the run cleanly.
        stage_and_gate(
            pool.clone(),
            cfg,
            topo,
            art,
            crate::events::channel(),
            run_id,
            deploy_lock,
        )
        .await
        .expect("a red gate settles the run, it does not error out");

        // Tier did NOT advance — still the seeded NULL/NULL.
        let (current, previous) = tier_versions(&pool, "host").await;
        assert_eq!(current, None);
        assert_eq!(previous, None);

        // Run settled red with a non-empty summary.
        let (result, summary) = run_result(&pool, run_id).await;
        assert_eq!(result, "failed");
        assert!(
            summary.as_deref().is_some_and(|s| !s.is_empty()),
            "failed run must carry a summary, got {summary:?}"
        );
    }

    #[test]
    fn check_build_host_accepts_matching_host() {
        assert!(check_build_host("fw13", "fw13").is_ok());
    }

    #[test]
    fn check_build_host_refuses_mismatched_host() {
        // A daemon misdeployed onto prod (e.g. a Hetzner host) must refuse.
        let err = check_build_host("alpha-west-1", "fw13")
            .unwrap_err()
            .to_string();
        assert!(err.contains("refusing to build"), "{err}");
        assert!(
            err.contains("alpha-west-1") && err.contains("fw13"),
            "{err}"
        );
    }

    #[test]
    fn runtime_hostname_reads_a_nonempty_trimmed_name() {
        let h = runtime_hostname().expect("hostname readable on Linux");
        assert!(!h.is_empty());
        assert_eq!(h, h.trim(), "must be trimmed");
    }

    #[test]
    fn tail_does_not_panic_on_multibyte_boundary() {
        // Each '€' is 3 bytes; a byte cap landing mid-codepoint must not panic.
        let s = "€".repeat(10); // 30 bytes
        for max in 1..=30 {
            let out = tail(s.as_bytes(), max);
            assert!(out.len() <= max, "max={max} got {} bytes", out.len());
            // Result is always valid UTF-8 made only of whole '€'s.
            assert!(out.chars().all(|c| c == '€'), "max={max}: {out:?}");
        }
    }

    #[test]
    fn tail_returns_whole_input_when_under_cap() {
        assert_eq!(tail(b"hello", 100), "hello");
    }
}