//! Atomic symlink-swap deploys.
//!
//! Layout on every target (local host, A nodes, B nodes, ...). Release dirs are
//! named for their content digest, not the version (wiki note
//! `release-artifact-identity`), so a rebuild can never overwrite an earlier
//! build's dir in place:
//!
//!   <release_root>/
//!     releases/
//!       a1b2c3d4e5f60718/       <- <digest16>
//!         <bin_name>
//!         MANIFEST              <- <sha256>  <relpath> per file, node-verified
//!       f0e1d2c3b4a59687/
//!         <bin_name>
//!     current -> releases/f0e1d2c3b4a59687
//!
//! `ln -sfn` swaps the symlink. systemd units point at
//! `<release_root>/current/<bin_name>` so reload-or-restart picks up the new
//! binary without ever pointing at a missing path.
//!
//! The host-side transport — `ssh` for shell steps, `rsync` for the release
//! dir — comes from the shared [`ops_exec::Executor`] (a `LocalExec` for
//! `ssh_target = "local"`, an `SshExec` otherwise), built once per node in
//! [`crate::state`]. This module owns the *deploy choreography* (mkdir, push,
//! atomic swap, restart, gc); the transport is the crate's. SSH push behavior
//! is identical to the pre-extraction code — this is a transport extraction,
//! not a model change.

use crate::topology::Node;
use anyhow::{Context, Result};
use async_trait::async_trait;
use ops_exec::{Action, Executor, LogSink, RunOutput, Step, SyncOpts, sh_quote};
use std::path::{Path, PathBuf};
use tokio::process::Command;

/// Keep this many release dirs per node; older ones get gc'd after a
/// successful deploy. Fixed for now; promote to config if the constant ever
/// needs to vary by tier.
const RELEASES_TO_KEEP: usize = 5;

/// A sink that drops streamed bytes. Deploy steps don't have a live-log handle
/// (gates do), so output is discarded as it streams; [`RunOutput`] still
/// captures the full stdout/stderr for error reporting, preserving the
/// pre-extraction behavior of surfacing `stderr` in failure messages.
struct DiscardSink;

#[async_trait]
impl LogSink for DiscardSink {
    async fn write_chunk(&mut self, _bytes: &[u8]) {}
}

/// Run a shell step through `executor`, treating a non-zero exit as an error
/// whose message carries the captured stderr — exactly as the old bespoke
/// `ssh()` helper did (`ssh <target> failed: <stderr>`).
async fn run_checked(executor: &dyn Executor, script: &str, what: &str) -> Result<RunOutput> {
    let step = Step::shell(Action::Deploy, script);
    let mut sink = DiscardSink;
    let out = executor
        .run_streaming(&step, &mut sink)
        .await
        .with_context(|| format!("{what}: spawning command"))?;
    anyhow::ensure!(
        out.status.success(),
        "{what} failed (exit {}): {}",
        out.status
            .code()
            .map_or_else(|| "signal".into(), |c| c.to_string()),
        String::from_utf8_lossy(&out.stderr),
    );
    Ok(out)
}

/// Stage built binaries into `staging/<build_id>/` on the Sando host — a
/// private, mutable scratch dir that is not yet a release (no symlink, no gc).
/// The caller adds `release_contents` + companions, hashes the result, writes
/// the `MANIFEST`, then publishes it content-addressed via
/// [`finalize_local_release`]. Splitting staging from publish is what lets the
/// bundle be hashed before it is named (wiki [[release-artifact-identity]]).
///
/// A stale `staging/<build_id>` from a killed prior run at the same id is
/// removed first, so a retry stages clean.
pub async fn stage_local_bundle(
    release_root: &Path,
    build_id: i64,
    binaries: &[PathBuf],
) -> Result<PathBuf> {
    let staging = release_root.join("staging").join(build_id.to_string());
    if tokio::fs::try_exists(&staging).await.unwrap_or(false) {
        tokio::fs::remove_dir_all(&staging)
            .await
            .with_context(|| format!("clearing stale staging dir {}", staging.display()))?;
    }
    tokio::fs::create_dir_all(&staging).await?;
    for binary in binaries {
        let name = binary.file_name().context("binary path has no file name")?;
        let dest = staging.join(name);
        tokio::fs::copy(binary, &dest)
            .await
            .with_context(|| format!("copy {} -> {}", binary.display(), dest.display()))?;
    }
    Ok(staging)
}

/// Publish a fully-staged bundle content-addressed: rename
/// `staging/<build_id>` to `releases/<digest16>` (atomic same-filesystem
/// rename), flip `current` to it, gc old releases. Returns the released dir.
///
/// The rename is the load-bearing step: **a directory whose name derives from
/// its contents cannot be rewritten, because rewriting it changes its name.**
/// The overwrite class that let a dev rebuild inherit an earlier build's gate
/// rows and burn-in clock stops being something to guard against and becomes
/// something that cannot be expressed. If a release with this digest already
/// exists (identical bytes rebuilt), the staging copy is redundant and dropped.
pub async fn finalize_local_release(
    release_root: &Path,
    staging: &Path,
    digest16: &str,
) -> Result<PathBuf> {
    let releases = release_root.join("releases");
    tokio::fs::create_dir_all(&releases).await?;
    let released = releases.join(digest16);

    if tokio::fs::try_exists(&released).await.unwrap_or(false) {
        // Same digest already published — reuse it, discard the redundant stage.
        tokio::fs::remove_dir_all(staging).await.ok();
    } else {
        tokio::fs::rename(staging, &released)
            .await
            .with_context(|| format!("publish {} -> {}", staging.display(), released.display()))?;
    }

    let current = release_root.join("current");
    let target = format!("releases/{digest16}");
    let out = Command::new("ln")
        .args(["-sfn", &target])
        .arg(&current)
        .output()
        .await?;
    anyhow::ensure!(
        out.status.success(),
        "symlink swap failed: {}",
        String::from_utf8_lossy(&out.stderr),
    );

    if let Err(e) = gc_local_releases(release_root).await {
        tracing::warn!(error = %e, "local release GC failed (non-fatal)");
    }
    Ok(released)
}

/// Deploy `staged_release_dir` (a directory built on the Sando host by
/// `deploy_local`) to `node` using `executor` (its transport from the topology
/// executor map). For `ssh_target=local`, this is just a symlink swap; for
/// remote nodes, we rsync the whole dir over the executor.
///
/// `primary_bin` is only used for logging — every file present in the staged
/// dir gets shipped.
pub async fn deploy_node(
    executor: &dyn Executor,
    node: &Node,
    version: &str,
    staged_release_dir: &Path,
    primary_bin: &str,
) -> Result<PathBuf> {
    // The release dir is named for its content digest (`releases/<digest16>`),
    // not the version. The node mirrors that name so host and node agree on the
    // artifact's identity; the version is only a log label here. Legacy staged
    // dirs (pre-identity, still `releases/<version>`) work unchanged — the name
    // is whatever the host staged under.
    let release_id = staged_release_dir
        .file_name()
        .and_then(|n| n.to_str())
        .with_context(|| {
            format!(
                "staged release dir {} has no usable name",
                staged_release_dir.display()
            )
        })?;
    if node.ssh_target == "local" || node.ssh_target.is_empty() {
        // Local deploy already happened when we staged on the Sando host.
        // Just re-point `current` at the staged dir.
        return reset_local_current(executor, Path::new(&node.release_root), release_id).await;
    }
    deploy_remote(
        executor,
        node,
        version,
        release_id,
        staged_release_dir,
        primary_bin,
    )
    .await
}

async fn reset_local_current(
    executor: &dyn Executor,
    release_root: &Path,
    release_id: &str,
) -> Result<PathBuf> {
    let current = release_root.join("current");
    let target = format!("releases/{release_id}");
    run_checked(
        executor,
        &format!(
            "ln -sfn {} {}",
            sh_quote(&target),
            sh_quote(&current.to_string_lossy())
        ),
        "local symlink swap",
    )
    .await?;
    Ok(release_root.join("releases").join(release_id))
}

async fn deploy_remote(
    executor: &dyn Executor,
    node: &Node,
    version: &str,
    release_id: &str,
    staged_release_dir: &Path,
    primary_bin: &str,
) -> Result<PathBuf> {
    let release_root = &node.release_root;
    let service = &node.service_name;
    let release_dir = format!("{release_root}/releases/{release_id}");

    tracing::info!(node = %node.name, version, release_id, "deploy: mkdir release dir");
    run_checked(
        executor,
        &format!("set -e; mkdir -p {q}", q = sh_quote(&release_dir)),
        "creating remote release dir",
    )
    .await?;

    tracing::info!(node = %node.name, version, primary = %primary_bin, "deploy: rsync release dir");
    // Rsync the whole staged dir (binaries + every release_contents entry).
    // `SyncOpts::release_mirror()` is the exact pre-extraction rsync flag set:
    // -az --partial --delete --chmod=Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X.
    //   --delete: removed assets across versions don't accumulate on the
    //     target; the bundle stays self-contained per version.
    //   --chmod: F+X preserves the execute bit per-file (binaries land 0755,
    //     data files 0644) instead of a blanket 0755.
    executor
        .push_dir(
            staged_release_dir,
            Path::new(&release_dir),
            &SyncOpts::release_mirror(),
        )
        .await
        .context("rsync failed (current symlink left intact)")?;

    // Verify the bundle on the node against its own MANIFEST before the swap
    // (invariant 3, wiki [[release-artifact-identity]]). The MANIFEST shipped in
    // the bundle is exactly `sha256sum` check format (`<hash>  <relpath>`), so
    // this re-hashes every file on the node and names any that drifted in
    // transit — the hash is load-bearing, not merely recorded. Bundles staged
    // by a pre-identity build carry no MANIFEST; those skip verification (logged)
    // rather than fail, so a mid-migration deploy of a legacy artifact still
    // ships. A mismatch fails the promote with the running service intact.
    run_checked(
        executor,
        &manifest_verify_script(&release_dir),
        "verifying bundle digest on node",
    )
    .await
    .context("node-side bundle verification failed (current symlink left intact)")?;

    // Fail closed on a wrong-architecture binary before the symlink swap. The
    // "never cross-compile" rule is enforced at build time (build_host check),
    // but nothing verified the artifact's arch matched the *target* node — so
    // adding an aarch64 node to a tier built on x86_64 would silently symlink an
    // unrunnable binary live. Compare the deployed binary's ELF e_machine to the
    // node's `uname -m`; unknown arches log and proceed (can't verify != known-bad).
    let deployed_bin = format!("{release_dir}/{primary_bin}");
    run_checked(
        executor,
        &arch_guard_script(&deployed_bin),
        "verifying binary arch matches node",
    )
    .await
    .context(
        "deployed binary architecture does not match the target node (current symlink left intact)",
    )?;

    // Config-drift guard (opt-in per node). Runs the freshly-rsynced binary in
    // config-only mode with the node's env sourced, BEFORE the swap, so a
    // required var missing on this node fails here — service still intact —
    // rather than after the restart, which would crash-loop it (how testnot
    // went down on a missing CDN_BASE_URL). Skipped unless the node sets
    // `config_check_env_file`.
    if let Some(env_file) = node.config_check_env_file.as_deref() {
        tracing::info!(node = %node.name, version, "deploy: pre-swap config check");
        check_target_config(executor, &deployed_bin, env_file)
            .await
            .context("pre-swap config check failed (current symlink left intact)")?;
    }

    tracing::info!(node = %node.name, version, "deploy: symlink swap + service reload");
    let restart_cmd = format!(
        "sudo /bin/systemctl reload-or-restart {}",
        sh_quote(service)
    );
    let swap_and_restart = swap_and_restart_script(release_root, release_id, &restart_cmd);
    run_checked(
        executor,
        &swap_and_restart,
        "symlink swap + systemctl reload-or-restart",
    )
    .await?;

    // Companion services (opt-in per node): install each from the just-rsynced
    // bundle and restart its unit via the node-side wrapper, AFTER the server is
    // up (mnw-cli is `After=makenotwork.service`). They shipped from the SAME sha
    // in this SAME bundle — the lockstep guarantee. A failure here fails the
    // promote: a companion is part of the deploy, not a best-effort side effect.
    for c in &node.companions {
        let src = format!(
            "{release_root}/releases/{release_id}/companions/{name}",
            name = c.name,
        );
        tracing::info!(node = %node.name, companion = %c.name, "deploy: install companion + restart");
        let cmd = install_companion_cmd(&src, &c.install_path, &c.service_name);
        run_checked(executor, &cmd, "install companion + restart")
            .await
            .with_context(|| {
                format!(
                    "companion {} deploy failed (server already swapped)",
                    c.name
                )
            })?;
    }

    if let Err(e) = gc_remote_releases(executor, release_root).await {
        tracing::warn!(error = %e, "remote release GC failed (non-fatal)");
    }

    Ok(PathBuf::from(release_root)
        .join("releases")
        .join(release_id))
}

/// Absolute path of the node-side companion installer (shipped once per node;
/// granted to the deploy user by a single scoped sudoers line). It installs the
/// staged binary to its `ExecStart` path and restarts the unit — keeping the
/// sudo grant to one script rather than a broad `install`/`systemctl` grant.
const COMPANION_INSTALLER: &str = "/usr/local/lib/mnw/install-companion.sh";

/// Command run on the node to install a staged companion binary and restart its
/// unit, via the wrapper. Pure builder so it can be unit-tested; all three args
/// are shell-quoted (paths/unit names, operator config — but quoted regardless).
fn install_companion_cmd(src: &str, install_path: &str, service: &str) -> String {
    format!(
        "sudo {installer} {src} {dst} {svc}",
        installer = sh_quote(COMPANION_INSTALLER),
        src = sh_quote(src),
        dst = sh_quote(install_path),
        svc = sh_quote(service),
    )
}

/// Pre-swap config-drift check: load the node's env file the way systemd loads
/// it, then run the freshly-deployed binary in `MNW_CHECK_CONFIG=1` mode (loads
/// config, exits 0/1, no DB/migrations/bind). A non-zero exit — a required var
/// missing — is surfaced by `run_checked` as an error, failing the promote
/// before the swap.
///
/// Bounded by a timeout as a backstop: a binary predating `MNW_CHECK_CONFIG`
/// would ignore the var and try to start normally, which must not hang the
/// deploy. A timeout is reported as a failure (fail closed) — the operator only
/// opts a node in once a check-capable version is deployed, so a timeout means
/// something is wrong, not a routine older binary.
async fn check_target_config(
    executor: &dyn Executor,
    deployed_bin: &str,
    env_file: &str,
) -> Result<()> {
    let script = config_check_script(env_file, deployed_bin);
    let fut = run_checked(executor, &script, "pre-swap config check");
    match tokio::time::timeout(std::time::Duration::from_secs(20), fut).await {
        Ok(result) => result.map(|_| ()),
        Err(_) => anyhow::bail!(
            "pre-swap config check timed out after 20s — the binary may predate \
             MNW_CHECK_CONFIG or the check hung; refusing to swap"
        ),
    }
}

/// Shell that loads `env_file` with systemd `EnvironmentFile=` semantics, then
/// runs `bin` under `MNW_CHECK_CONFIG=1`.
///
/// Load the file line by line and `export` each `KEY=VALUE` verbatim rather than
/// `. env_file`. Dot-sourcing runs the file as a script, so any shell
/// metacharacter in a value (`$`, backticks, `;`, `&`, a glob, whitespace) is
/// expanded or word-split — a DB URL carrying a password silently dropped
/// `DATABASE_URL` to empty on our nodes, which would fail the check (and thus
/// every deploy) even though systemd starts the service fine. `export "$line"`
/// assigns the already-expanded word literally, matching systemd's "no variable
/// expansion" rule. Comments and blank lines are skipped; the `|| [ -n "$line" ]`
/// guard processes a final line with no trailing newline. (Quoted values —
/// `KEY="v"` — aren't unquoted here the way systemd would, but our env files use
/// bare `KEY=VALUE`, and a stray quote can only make the check stricter, never
/// wave a bad config through.)
fn config_check_script(env_file: &str, bin: &str) -> String {
    format!(
        "set -eu\n\
         while IFS= read -r __sando_l || [ -n \"$__sando_l\" ]; do\n\
         \tcase \"$__sando_l\" in ''|'#'*) continue ;; esac\n\
         \texport \"$__sando_l\"\n\
         done < {env}\n\
         MNW_CHECK_CONFIG=1 {bin}\n",
        env = sh_quote(env_file),
        bin = sh_quote(bin),
    )
}

/// Build the swap-and-restart shell script for a remote node.
///
/// The symlink swap is atomic via `mv -T` of a freshly-created symlink over the
/// old one (the rename(2) is the atomic step; `ln -sfn` alone does
/// unlink+symlink, which has a window). The load-bearing part: if `restart_cmd`
/// fails *after* the flip, `current` is rolled back to its prior target before
/// the script exits non-zero. Otherwise a failed restart would leave `current`
/// pointing at the new, un-activated release while the service still runs the
/// old one — and a later reboot/cron restart would then silently bring up the
/// release the deploy reported as failed. Best-effort re-restart of the prior
/// version keeps the running service consistent with the restored symlink.
///
/// `restart_cmd` is injected (rather than hardcoded) so tests can drive the
/// failure and success paths with a `false`/`true` stand-in.
fn swap_and_restart_script(release_root: &str, release_id: &str, restart_cmd: &str) -> String {
    format!(
        "set -e\n\
         cd {root}\n\
         prev=$(readlink current 2>/dev/null || true)\n\
         ln -sfn releases/{rel} current.new\n\
         mv -Tf current.new current\n\
         if ! {restart}; then\n\
         if [ -n \"$prev\" ]; then\n\
         ln -sfn \"$prev\" current.rollback\n\
         mv -Tf current.rollback current\n\
         {restart} || true\n\
         fi\n\
         echo \"deploy: restart failed; rolled symlink back to ${{prev:-<none>}}\" >&2\n\
         exit 1\n\
         fi\n",
        root = sh_quote(release_root),
        rel = sh_quote(release_id),
        restart = restart_cmd,
    )
}

/// Shell that re-hashes the rsynced bundle on the node against its shipped
/// `MANIFEST` and aborts (exit 1) if any file drifted (invariant 3, wiki note
/// `release-artifact-identity`). The `MANIFEST` is `sha256sum` check format,
/// so `sha256sum -c` verifies every listed file with node-native tooling and
/// names the one that failed. `--strict` fails on a malformed manifest line;
/// `--quiet` drops the per-file OK spam and keeps only failures.
///
/// A bundle staged by a pre-identity build carries no `MANIFEST`; that is not an
/// error — it logs a skip and exits 0, so a mid-migration deploy of a legacy
/// artifact still ships. Once every tier has cycled once, every bundle has one.
fn manifest_verify_script(release_dir: &str) -> String {
    format!(
        "set -e\n\
         cd {dir}\n\
         if [ ! -f MANIFEST ]; then\n\
         echo \"deploy: no MANIFEST in bundle; skipping digest verification (legacy artifact)\" >&2\n\
         exit 0\n\
         fi\n\
         sha256sum --quiet --strict -c MANIFEST\n",
        dir = sh_quote(release_dir),
    )
}

/// Shell that aborts (exit 1) if `bin`'s ELF architecture doesn't match the
/// node it's running on. Reads the ELF `e_machine` field (2 bytes LE at offset
/// 18) and compares it to the value implied by `uname -m`. An arch we don't have
/// a mapping for logs and proceeds — the guard exists to catch the concrete
/// x86_64-vs-aarch64 confusion, not to gate genuinely-new targets.
fn arch_guard_script(bin: &str) -> String {
    format!(
        "set -e\n\
         bin={bin}\n\
         arch=$(uname -m)\n\
         machine=$(od -An -tx1 -j18 -N2 \"$bin\" 2>/dev/null | tr -d ' \\n')\n\
         case \"$arch\" in\n\
         x86_64|amd64) want=3e00 ;;\n\
         aarch64|arm64) want=b700 ;;\n\
         *) echo \"deploy: arch check skipped (unmapped node arch $arch)\" >&2; want= ;;\n\
         esac\n\
         if [ -n \"$want\" ] && [ \"$machine\" != \"$want\" ]; then\n\
         echo \"deploy: arch mismatch — node $arch expects e_machine $want but binary has ${{machine:-<unreadable>}}\" >&2\n\
         exit 1\n\
         fi\n",
        bin = sh_quote(bin),
    )
}

async fn gc_local_releases(release_root: &Path) -> Result<()> {
    let releases = release_root.join("releases");
    if !releases.exists() {
        return Ok(());
    }
    let mut entries = Vec::new();
    let mut rd = tokio::fs::read_dir(&releases).await?;
    while let Some(entry) = rd.next_entry().await? {
        if !entry.file_type().await?.is_dir() {
            continue;
        }
        let meta = entry.metadata().await?;
        entries.push((entry.path(), meta.modified()?));
    }
    entries.sort_by_key(|e| std::cmp::Reverse(e.1));
    for (path, _) in entries.into_iter().skip(RELEASES_TO_KEEP) {
        if let Err(e) = tokio::fs::remove_dir_all(&path).await {
            tracing::warn!(path = %path.display(), error = %e, "gc: rm failed");
        } else {
            tracing::debug!(path = %path.display(), "gc: removed old release");
        }
    }
    Ok(())
}

async fn gc_remote_releases(executor: &dyn Executor, release_root: &str) -> Result<()> {
    // `ls -t` orders by mtime desc. Skip the first N, rm the rest. `xargs -r`
    // is a no-op when stdin is empty (avoids `rm` complaining).
    let script = format!(
        "set -e; cd {root}/releases 2>/dev/null || exit 0; \
         ls -1t | tail -n +{keep_plus_one} | xargs -r -I{{}} rm -rf -- {{}}",
        root = sh_quote(release_root),
        keep_plus_one = RELEASES_TO_KEEP + 1,
    );
    run_checked(executor, &script, "remote release gc")
        .await
        .map(|_| ())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::topology::NodeCompanion;
    use ops_exec::{CapabilitySet, LocalExec, SshExec};
    use std::os::unix::process::ExitStatusExt;
    use std::sync::{Arc, Mutex as StdMutex};
    use std::time::SystemTime;

    /// A LocalExec granted the default node capabilities (deploy + restart).
    fn local_executor() -> LocalExec {
        LocalExec::new(CapabilitySet::from_tokens(
            ["deploy", "restart"],
            ["health"],
        ))
    }

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

        let src_dir = root.join("src");
        tokio::fs::create_dir_all(&src_dir).await.unwrap();
        let primary = src_dir.join("makenotwork");
        let admin = src_dir.join("mnw-admin");
        tokio::fs::write(&primary, b"PRIMARY").await.unwrap();
        tokio::fs::write(&admin, b"ADMIN").await.unwrap();

        let release_root = root.join("releases-root");
        tokio::fs::create_dir_all(&release_root).await.unwrap();

        // Stage into staging/<build_id> (no publish yet).
        let staging = stage_local_bundle(&release_root, 42, &[primary.clone(), admin.clone()])
            .await
            .expect("stage_local_bundle should succeed");
        assert_eq!(staging, release_root.join("staging").join("42"));
        assert!(
            !release_root.join("current").exists(),
            "staging must not publish or flip current"
        );

        // Publish content-addressed at releases/<digest16>.
        let released = finalize_local_release(&release_root, &staging, "deadbeefcafe0000")
            .await
            .expect("finalize_local_release should succeed");
        assert_eq!(
            released,
            release_root.join("releases").join("deadbeefcafe0000")
        );
        assert!(
            !staging.exists(),
            "staging dir is consumed by the publish rename"
        );
        assert_eq!(
            tokio::fs::read(released.join("makenotwork")).await.unwrap(),
            b"PRIMARY"
        );
        assert_eq!(
            tokio::fs::read(released.join("mnw-admin")).await.unwrap(),
            b"ADMIN"
        );

        let current = release_root.join("current");
        let target = tokio::fs::read_link(&current).await.unwrap();
        assert_eq!(target.to_string_lossy(), "releases/deadbeefcafe0000");
        let via_current = tokio::fs::read(current.join("makenotwork")).await.unwrap();
        assert_eq!(via_current, b"PRIMARY");
    }

    #[tokio::test]
    async fn finalize_second_release_swaps_symlink_and_keeps_old_dir() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let src_dir = root.join("src");
        tokio::fs::create_dir_all(&src_dir).await.unwrap();
        let bin = src_dir.join("server");
        tokio::fs::write(&bin, b"V1").await.unwrap();

        let release_root = root.join("rr");
        tokio::fs::create_dir_all(&release_root).await.unwrap();

        // Two builds, distinct digests (distinct content) -> two release dirs.
        let s1 = stage_local_bundle(&release_root, 1, std::slice::from_ref(&bin))
            .await
            .unwrap();
        finalize_local_release(&release_root, &s1, "1111111111111111")
            .await
            .unwrap();
        tokio::fs::write(&bin, b"V2").await.unwrap();
        let s2 = stage_local_bundle(&release_root, 2, std::slice::from_ref(&bin))
            .await
            .unwrap();
        finalize_local_release(&release_root, &s2, "2222222222222222")
            .await
            .unwrap();

        assert!(
            release_root
                .join("releases/1111111111111111/server")
                .exists()
        );
        assert!(
            release_root
                .join("releases/2222222222222222/server")
                .exists()
        );
        let target = tokio::fs::read_link(release_root.join("current"))
            .await
            .unwrap();
        assert_eq!(target.to_string_lossy(), "releases/2222222222222222");
        let via_current = tokio::fs::read(release_root.join("current/server"))
            .await
            .unwrap();
        assert_eq!(via_current, b"V2");
    }

    #[tokio::test]
    async fn finalize_reuses_an_existing_release_of_the_same_digest() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let bin = root.join("server");
        tokio::fs::write(&bin, b"BYTES").await.unwrap();
        let release_root = root.join("rr");
        tokio::fs::create_dir_all(&release_root).await.unwrap();

        let s1 = stage_local_bundle(&release_root, 1, std::slice::from_ref(&bin))
            .await
            .unwrap();
        finalize_local_release(&release_root, &s1, "abc123abc123abc1")
            .await
            .unwrap();
        // Same digest rebuilt (e.g. a re-run at the same content): finalize must
        // reuse the existing release and drop the redundant staging dir, not error.
        let s2 = stage_local_bundle(&release_root, 2, std::slice::from_ref(&bin))
            .await
            .unwrap();
        let released = finalize_local_release(&release_root, &s2, "abc123abc123abc1")
            .await
            .expect("finalize is idempotent on a repeated digest");
        assert_eq!(released, release_root.join("releases/abc123abc123abc1"));
        assert!(!s2.exists(), "redundant staging dropped");
    }

    #[tokio::test]
    async fn manifest_verify_script_passes_on_match_fails_on_drift_and_skips_when_absent() {
        // The node-side verification is a shell running `sha256sum -c MANIFEST`;
        // drive the real script through bash to prove it accepts a good bundle,
        // rejects a tampered one, and no-ops on a legacy (MANIFEST-less) bundle.
        let dir = tempfile::tempdir().unwrap();
        tokio::fs::write(dir.path().join("server"), b"BINARY")
            .await
            .unwrap();
        tokio::fs::create_dir(dir.path().join("static"))
            .await
            .unwrap();
        tokio::fs::write(dir.path().join("static/app.css"), b"body{}")
            .await
            .unwrap();
        let digest = crate::bundle::digest_dir(dir.path()).await.unwrap();
        tokio::fs::write(dir.path().join("MANIFEST"), digest.manifest.as_bytes())
            .await
            .unwrap();

        let run = |d: &std::path::Path| {
            let script = manifest_verify_script(d.to_str().unwrap());
            async move {
                Command::new("bash")
                    .arg("-c")
                    .arg(&script)
                    .output()
                    .await
                    .unwrap()
            }
        };

        let ok = run(dir.path()).await;
        assert!(
            ok.status.success(),
            "matching bundle verifies: {}",
            String::from_utf8_lossy(&ok.stderr)
        );

        // Drift one file: sha256sum -c must fail (current symlink left intact).
        tokio::fs::write(dir.path().join("static/app.css"), b"TAMPERED")
            .await
            .unwrap();
        let bad = run(dir.path()).await;
        assert!(!bad.status.success(), "a drifted file fails verification");

        // Legacy bundle with no MANIFEST: skip, not fail.
        let legacy = tempfile::tempdir().unwrap();
        tokio::fs::write(legacy.path().join("server"), b"x")
            .await
            .unwrap();
        let skip = run(legacy.path()).await;
        assert!(
            skip.status.success(),
            "a bundle without a MANIFEST skips verification rather than failing"
        );
    }

    #[tokio::test]
    async fn gc_local_releases_keeps_last_n_by_mtime() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let releases = root.join("releases");
        tokio::fs::create_dir_all(&releases).await.unwrap();

        let total = RELEASES_TO_KEEP + 3;
        let mut names = Vec::new();
        for i in 0..total {
            let name = format!("v{i:02}");
            let dir = releases.join(&name);
            tokio::fs::create_dir(&dir).await.unwrap();
            let f = std::fs::File::open(&dir).unwrap();
            let when =
                SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000 + i as u64);
            let times = std::fs::FileTimes::new().set_modified(when);
            f.set_times(times).unwrap();
            names.push(name);
        }

        gc_local_releases(root).await.unwrap();

        let surviving_expected: Vec<_> = names
            .iter()
            .skip(total - RELEASES_TO_KEEP)
            .cloned()
            .collect();
        for name in &surviving_expected {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
        for name in names.iter().take(total - RELEASES_TO_KEEP) {
            assert!(
                !releases.join(name).exists(),
                "expected to be pruned: {name}"
            );
        }
    }

    #[tokio::test]
    async fn gc_local_releases_noop_when_below_threshold() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let releases = root.join("releases");
        tokio::fs::create_dir_all(&releases).await.unwrap();
        for i in 0..3 {
            tokio::fs::create_dir(releases.join(format!("v{i}")))
                .await
                .unwrap();
        }
        gc_local_releases(root).await.unwrap();
        for i in 0..3 {
            assert!(releases.join(format!("v{i}")).exists());
        }
    }

    #[tokio::test]
    async fn gc_local_releases_noop_when_releases_dir_missing() {
        let tmp = tempfile::tempdir().unwrap();
        gc_local_releases(tmp.path()).await.unwrap();
    }

    #[tokio::test]
    async fn deploy_remote_fails_cleanly_when_host_unreachable() {
        // 192.0.2.0/24 is reserved for documentation and routes nowhere.
        // ConnectTimeout=10 limits the test wallclock to ~10s worst case.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.0.1");
        tokio::fs::create_dir_all(&staged).await.unwrap();
        tokio::fs::write(staged.join("server"), b"x").await.unwrap();

        let node = crate::topology::Node {
            name: "unreachable".into(),
            ssh_target: "deploy@192.0.2.1".into(),
            release_root: "/opt/never".into(),
            service_name: "makenotwork.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: crate::topology::default_actuate(),
            observe: crate::topology::default_observe(),
            companions: Vec::new(),
        };
        let executor = SshExec::new(
            node.ssh_target.clone(),
            CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
        );

        let result = deploy_node(&executor, &node, "0.0.1", &staged, "server").await;
        let err = result.expect_err("deploy to unreachable host should fail");
        let msg = format!("{err:#}");
        // Don't pin exact wording, just that the failure is attributed (ssh /
        // rsync / connection) and that no panic / hang happened.
        assert!(
            msg.contains("ssh")
                || msg.contains("rsync")
                || msg.contains("connection")
                || msg.contains("Connection"),
            "unexpected error: {msg}"
        );
    }

    #[tokio::test]
    async fn deploy_node_with_local_ssh_target_swaps_symlink() {
        // ssh_target="local" routes to the local fast-path: just a symlink
        // swap, no remote calls.
        let tmp = tempfile::tempdir().unwrap();
        let release_root = tmp.path().to_path_buf();
        let staged = release_root.join("releases").join("0.0.1");
        tokio::fs::create_dir_all(&staged).await.unwrap();
        tokio::fs::write(staged.join("server"), b"x").await.unwrap();

        let node = crate::topology::Node {
            name: "local-dev".into(),
            ssh_target: "local".into(),
            release_root: release_root.to_string_lossy().into_owned(),
            service_name: "makenotwork.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: crate::topology::default_actuate(),
            observe: crate::topology::default_observe(),
            companions: Vec::new(),
        };
        let executor = local_executor();

        let out = deploy_node(&executor, &node, "0.0.1", &staged, "server")
            .await
            .unwrap();
        assert_eq!(out, staged);
        let target = tokio::fs::read_link(release_root.join("current"))
            .await
            .unwrap();
        assert_eq!(target.to_string_lossy(), "releases/0.0.1");
    }

    // ---- swap_and_restart_script: symlink/restart consistency ----

    async fn run_script(script: &str) -> std::process::Output {
        Command::new("sh")
            .arg("-c")
            .arg(script)
            .output()
            .await
            .unwrap()
    }

    async fn setup_release_root(with_current: bool) -> tempfile::TempDir {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        tokio::fs::create_dir_all(root.join("releases/old"))
            .await
            .unwrap();
        tokio::fs::create_dir_all(root.join("releases/new"))
            .await
            .unwrap();
        if with_current {
            std::os::unix::fs::symlink("releases/old", root.join("current")).unwrap();
        }
        tmp
    }

    #[tokio::test]
    async fn swap_and_restart_keeps_new_symlink_when_restart_succeeds() {
        let tmp = setup_release_root(true).await;
        let root = tmp.path().to_string_lossy().into_owned();
        let out = run_script(&swap_and_restart_script(&root, "new", "true")).await;
        assert!(
            out.status.success(),
            "script should succeed when restart succeeds"
        );
        let target = tokio::fs::read_link(tmp.path().join("current"))
            .await
            .unwrap();
        assert_eq!(
            target.to_string_lossy(),
            "releases/new",
            "symlink advanced to new"
        );
    }

    #[tokio::test]
    async fn swap_and_restart_rolls_symlink_back_when_restart_fails() {
        // The bug: a restart failure after the flip must NOT leave `current`
        // pointing at the new (un-activated) release.
        let tmp = setup_release_root(true).await;
        let root = tmp.path().to_string_lossy().into_owned();
        let out = run_script(&swap_and_restart_script(&root, "new", "false")).await;
        assert!(!out.status.success(), "script must fail when restart fails");
        let target = tokio::fs::read_link(tmp.path().join("current"))
            .await
            .unwrap();
        assert_eq!(
            target.to_string_lossy(),
            "releases/old",
            "symlink rolled back to prev so a later restart can't silently activate new",
        );
    }

    // ---- arch_guard_script: wrong-arch artifacts fail closed ----

    /// A 20-byte stub whose ELF e_machine field (offset 18, 2 bytes LE) is set.
    fn elf_stub_with_machine(b18: u8, b19: u8) -> tempfile::NamedTempFile {
        let mut data = vec![0u8; 20];
        data[18] = b18;
        data[19] = b19;
        let f = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(f.path(), &data).unwrap();
        f
    }

    /// e_machine low byte for the host running the test, if mapped.
    fn host_machine_lo() -> Option<u8> {
        match std::env::consts::ARCH {
            "x86_64" => Some(0x3e),
            "aarch64" => Some(0xb7),
            _ => None,
        }
    }

    #[tokio::test]
    async fn arch_guard_passes_for_matching_binary() {
        let Some(lo) = host_machine_lo() else { return };
        let f = elf_stub_with_machine(lo, 0x00);
        let out = run_script(&arch_guard_script(&f.path().to_string_lossy())).await;
        assert!(
            out.status.success(),
            "matching arch must pass: {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }

    #[tokio::test]
    async fn arch_guard_fails_closed_for_wrong_binary() {
        // Use the other arch's e_machine so it can't match the host.
        let wrong = match std::env::consts::ARCH {
            "x86_64" => 0xb7,  // aarch64 binary on an x86_64 node
            "aarch64" => 0x3e, // x86_64 binary on an aarch64 node
            _ => return,
        };
        let f = elf_stub_with_machine(wrong, 0x00);
        let out = run_script(&arch_guard_script(&f.path().to_string_lossy())).await;
        assert!(
            !out.status.success(),
            "wrong-arch binary must fail closed before the symlink swap"
        );
    }

    #[tokio::test]
    async fn swap_and_restart_first_deploy_failure_has_no_prev_to_restore() {
        // No prior `current`. A restart failure leaves `current` at new (the only
        // version) and still reports failure — documented degenerate case.
        let tmp = setup_release_root(false).await;
        let root = tmp.path().to_string_lossy().into_owned();
        let out = run_script(&swap_and_restart_script(&root, "new", "false")).await;
        assert!(!out.status.success(), "script must fail when restart fails");
        let target = tokio::fs::read_link(tmp.path().join("current"))
            .await
            .unwrap();
        assert_eq!(
            target.to_string_lossy(),
            "releases/new",
            "no prev existed to roll back to"
        );
    }

    // ---- config_check_script: systemd-faithful env loading ----

    #[tokio::test]
    async fn config_check_script_loads_values_with_shell_metachars() {
        // The bug: `. env_file` expands/word-splits values, so a URL or a
        // password containing a shell metacharacter is mangled — it dropped
        // DATABASE_URL to empty on a real node, which would fail every deploy.
        // The export-loop must load such a value intact. The "binary" is a
        // checker script (a real path, like a deployed binary) that exits 0 only
        // if the var arrived byte-for-byte — it compares against the expected
        // value read from a file, so nothing re-interprets the metacharacters.
        let tricky = "postgres://u:p$ss;w&rd@h/db `x` $(y)";
        // Plain files in a tempdir: no lingering write fd, so the checker can be
        // exec'd (a NamedTempFile stays open and would ETXTBSY).
        let dir = tempfile::tempdir().unwrap();
        let expected_path = dir.path().join("expected");
        std::fs::write(&expected_path, tricky).unwrap(); // no trailing newline

        let env_path = dir.path().join("node.env");
        std::fs::write(
            &env_path,
            format!(
                "# a comment\n\nDATABASE_URL={tricky}\nOTHER=plain\nEXPECTED_FILE={ef}\n",
                ef = expected_path.display(),
            ),
        )
        .unwrap();

        let checker_path = dir.path().join("checker.sh");
        std::fs::write(
            &checker_path,
            "#!/bin/sh\nwant=$(cat \"$EXPECTED_FILE\")\n\
             [ \"$DATABASE_URL\" = \"$want\" ] || { echo \"DB [$DATABASE_URL] != [$want]\" >&2; exit 1; }\n\
             [ \"$OTHER\" = plain ] || { echo \"OTHER [$OTHER]\" >&2; exit 1; }\n",
        )
        .unwrap();
        std::fs::set_permissions(
            &checker_path,
            std::os::unix::fs::PermissionsExt::from_mode(0o755),
        )
        .unwrap();

        let script =
            config_check_script(&env_path.to_string_lossy(), &checker_path.to_string_lossy());
        let out = run_script(&script).await;
        assert!(
            out.status.success(),
            "value with shell metachars must load intact; stderr: {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }

    // ---- install-companion.sh: the node-side guard rails ----

    /// Run the shipped installer script with three args; returns its exit code.
    /// Exercises the real file rather than a copy of its logic, because the
    /// script is the ONLY control on a NOPASSWD sudo grant.
    fn run_installer(src: &str, dst: &str, service: &str) -> i32 {
        let script =
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../deploy/install-companion.sh");
        std::process::Command::new("bash")
            .arg(&script)
            .args([src, dst, service])
            .output()
            .expect("running install-companion.sh")
            .status
            .code()
            .expect("script exited via signal")
    }

    // Guards run before any filesystem write, so these never install anything.
    // Exit 3 = refused by a guard; exit 4 = guards passed, src simply absent.
    const REFUSED: i32 = 3;
    const PASSED_GUARDS: i32 = 4;

    #[test]
    fn installer_refuses_a_dst_that_escapes_opt_via_dotdot() {
        // `/opt/../etc/...` matches a bare `/opt/*` glob. With the sudoers
        // wildcard that meant `install -m 0755` as root to anywhere, plus a
        // restart of any unit — so the path must be normalised before the test.
        assert_eq!(
            run_installer(
                "/opt/mnw/releases/1.0.0/companions/mnw-cli",
                "/opt/../etc/systemd/system/evil.service",
                "mnw-cli.service",
            ),
            REFUSED,
        );
    }

    #[test]
    fn installer_refuses_a_src_that_escapes_the_bundle_via_dotdot() {
        assert_eq!(
            run_installer(
                "/opt/mnw/releases/1.0.0/companions/../../../../../etc/shadow",
                "/opt/mnw-cli/mnw-cli",
                "mnw-cli.service",
            ),
            REFUSED,
        );
    }

    #[test]
    fn installer_accepts_the_real_companion_paths() {
        // The guards must not have been tightened into uselessness: the shape
        // Sando actually sends has to get past them. It stops at the missing
        // src (exit 4), which is proof the guards accepted it.
        assert_eq!(
            run_installer(
                "/opt/mnw/releases/1.0.0/companions/mnw-cli",
                "/opt/mnw-cli/mnw-cli",
                "mnw-cli.service",
            ),
            PASSED_GUARDS,
        );
    }

    #[test]
    fn installer_refuses_a_service_name_with_a_path_separator() {
        assert_eq!(
            run_installer(
                "/opt/mnw/releases/1.0.0/companions/mnw-cli",
                "/opt/mnw-cli/mnw-cli",
                "../../etc/evil.service",
            ),
            REFUSED,
        );
    }

    // ---- install_companion_cmd: shape + quoting ----

    #[test]
    fn install_companion_cmd_shape_and_quoting() {
        let cmd = install_companion_cmd(
            "/opt/mnw/releases/0.10.14/companions/mnw-cli",
            "/opt/mnw-cli/mnw-cli",
            "mnw-cli.service",
        );
        // Routes through the wrapper (single sudoers grant), sudo-invoked, with
        // src, dst, service in that order.
        assert!(cmd.starts_with("sudo "), "must be sudo-invoked: {cmd}");
        assert!(
            cmd.contains("/usr/local/lib/mnw/install-companion.sh"),
            "{cmd}"
        );
        let installer_pos = cmd.find("install-companion.sh").unwrap();
        let src_pos = cmd.find("companions/mnw-cli").unwrap();
        let dst_pos = cmd.find("/opt/mnw-cli/mnw-cli").unwrap();
        let svc_pos = cmd.find("mnw-cli.service").unwrap();
        assert!(
            installer_pos < src_pos && src_pos < dst_pos && dst_pos < svc_pos,
            "arg order: {cmd}"
        );
    }

    #[test]
    fn install_companion_cmd_quotes_metachars() {
        // A path with a space/quote must be shell-safe (defense in depth even
        // though these come from operator config).
        let cmd = install_companion_cmd("/a b/src", "/dst'x", "u.service");
        let out = std::process::Command::new("sh")
            .arg("-c")
            .arg(format!(
                "set -- {}; echo \"$#\"",
                cmd.strip_prefix("sudo ").unwrap()
            ))
            .output()
            .unwrap();
        // installer + 3 args = 4 positional words after quoting.
        assert_eq!(
            String::from_utf8_lossy(&out.stdout).trim(),
            "4",
            "quoting split wrong: {cmd}"
        );
    }

    #[tokio::test]
    async fn config_check_script_propagates_binary_failure() {
        // A required var missing (the binary exits non-zero) must fail the check.
        let env = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(env.path(), "FOO=bar\n").unwrap();
        let script = config_check_script(&env.path().to_string_lossy(), "false");
        let out = run_script(&script).await;
        assert!(
            !out.status.success(),
            "a non-zero MNW_CHECK_CONFIG exit must fail the check"
        );
    }

    #[tokio::test]
    async fn deploy_node_denied_when_executor_lacks_deploy_grant() {
        // Defense in depth: an executor without the deploy grant refuses the
        // step before any filesystem / ssh action.
        let tmp = tempfile::tempdir().unwrap();
        let release_root = tmp.path().to_path_buf();
        let staged = release_root.join("releases").join("0.0.1");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = crate::topology::Node {
            name: "local-dev".into(),
            ssh_target: "local".into(),
            release_root: release_root.to_string_lossy().into_owned(),
            service_name: "makenotwork.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: vec!["restart".into()], // no deploy
            observe: vec![],
            companions: Vec::new(),
        };
        let executor = LocalExec::new(CapabilitySet::from_tokens(["restart"], Vec::<&str>::new()));
        let err = deploy_node(&executor, &node, "0.0.1", &staged, "server")
            .await
            .unwrap_err();
        assert!(
            format!("{err:#}").contains("capability denied"),
            "expected capability denial"
        );
    }

    // ---- FakeExec: the deploy_remote choreography without a real host ----
    //
    // deploy_node's local fast-path is covered above with a real LocalExec, but
    // the remote path (rsync + arch guard + config-drift + swap + companions +
    // gc) short-circuits on `ssh_target != "local"` and so never ran under test
    // without a reachable node. FakeExec records every executor call in order
    // and can be told to fail one shell step (matched by substring) or the rsync
    // push, so the ordering and the fail-closed-before-swap contract are
    // assertable in-process.

    struct FakeExec {
        caps: CapabilitySet,
        calls: Arc<StdMutex<Vec<String>>>,
        /// The first `run_streaming` whose script contains this substring exits
        /// non-zero (a failed shell step), e.g. the arch guard.
        fail_run_matching: Option<String>,
        /// `push_dir` (the rsync) returns an error.
        fail_push_dir: bool,
    }

    impl FakeExec {
        fn new() -> Self {
            Self {
                caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
                calls: Arc::new(StdMutex::new(Vec::new())),
                fail_run_matching: None,
                fail_push_dir: false,
            }
        }
        fn log(&self) -> Vec<String> {
            self.calls.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl Executor for FakeExec {
        async fn run_streaming(&self, step: &Step, _sink: &mut dyn LogSink) -> Result<RunOutput> {
            // Every deploy step is a `Step::shell`, so the script is argv's tail.
            let script = step.argv.last().cloned().unwrap_or_default();
            self.calls.lock().unwrap().push(format!("run:{script}"));
            let fail = self
                .fail_run_matching
                .as_deref()
                .is_some_and(|m| script.contains(m));
            Ok(RunOutput {
                status: std::process::ExitStatus::from_raw(if fail { 1 << 8 } else { 0 }),
                stdout: Vec::new(),
                stderr: if fail {
                    b"fake step failure".to_vec()
                } else {
                    Vec::new()
                },
            })
        }
        async fn pull_file(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
            self.calls.lock().unwrap().push("pull_file".into());
            Ok(())
        }
        async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
            self.calls.lock().unwrap().push("pull_dir".into());
            Ok(())
        }
        async fn pull_glob(&self, _glob: &str, _local: &Path, _opts: &SyncOpts) -> Result<()> {
            self.calls.lock().unwrap().push("pull_glob".into());
            Ok(())
        }
        async fn push_dir(&self, _local: &Path, remote: &Path, _opts: &SyncOpts) -> Result<()> {
            self.calls
                .lock()
                .unwrap()
                .push(format!("push_dir:{}", remote.display()));
            if self.fail_push_dir {
                anyhow::bail!("fake rsync failure");
            }
            Ok(())
        }
        fn capabilities(&self) -> &CapabilitySet {
            &self.caps
        }
    }

    fn remote_node(config_check: bool, companions: Vec<NodeCompanion>) -> Node {
        Node {
            name: "web-a".into(),
            ssh_target: "deploy@web-a".into(),
            release_root: "/opt/mnw".into(),
            service_name: "makenotwork.service".into(),
            health_url: None,
            config_check_env_file: config_check.then(|| "/etc/mnw/node.env".to_string()),
            actuate: crate::topology::default_actuate(),
            observe: crate::topology::default_observe(),
            companions,
        }
    }

    fn companion() -> NodeCompanion {
        NodeCompanion {
            name: "mnw-cli".into(),
            install_path: "/opt/mnw-cli/mnw-cli".into(),
            service_name: "mnw-cli.service".into(),
        }
    }

    /// Index of the first recorded call whose text contains `needle` (panics if
    /// absent — the assertion message names what was missing).
    fn pos(log: &[String], needle: &str) -> usize {
        log.iter()
            .position(|c| c.contains(needle))
            .unwrap_or_else(|| panic!("no call matched {needle:?} in {log:#?}"))
    }

    #[tokio::test]
    async fn deploy_remote_runs_the_full_choreography_in_order() {
        // A node opted into the config-drift check and carrying one companion:
        // mkdir -> rsync -> arch guard -> config check -> swap+restart ->
        // companion install -> gc, in that order.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.9.0");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = remote_node(true, vec![companion()]);
        let exec = FakeExec::new();
        let out = deploy_node(&exec, &node, "0.9.0", &staged, "makenotwork")
            .await
            .expect("deploy_remote should succeed against the fake");
        assert_eq!(out, PathBuf::from("/opt/mnw/releases/0.9.0"));

        let log = exec.log();
        let mkdir = pos(&log, "mkdir -p");
        let rsync = pos(&log, "push_dir:/opt/mnw/releases/0.9.0");
        let arch = pos(&log, "e_machine");
        let cfg = pos(&log, "MNW_CHECK_CONFIG=1");
        let swap = pos(&log, "reload-or-restart");
        let comp = pos(&log, "install-companion.sh");
        let gc = pos(&log, "ls -1t");
        assert!(
            mkdir < rsync && rsync < arch && arch < cfg && cfg < swap && swap < comp && comp < gc,
            "deploy steps out of order: {log:#?}"
        );
    }

    #[tokio::test]
    async fn deploy_remote_aborts_before_swap_when_rsync_fails() {
        // The rsync failing must fail the deploy BEFORE the symlink swap — the
        // "current symlink left intact" contract. Assert the swap never ran.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.9.0");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = remote_node(false, Vec::new());
        let mut exec = FakeExec::new();
        exec.fail_push_dir = true;
        let err = deploy_node(&exec, &node, "0.9.0", &staged, "makenotwork")
            .await
            .expect_err("rsync failure must fail the deploy");
        assert!(
            format!("{err:#}").contains("rsync"),
            "error should attribute the rsync: {err:#}"
        );
        let log = exec.log();
        assert!(
            !log.iter().any(|c| c.contains("reload-or-restart")),
            "swap must not run after a failed rsync: {log:#?}"
        );
    }

    #[tokio::test]
    async fn deploy_remote_aborts_before_swap_when_arch_guard_fails() {
        // A wrong-arch binary must fail closed before the swap. The fake fails
        // the arch-guard shell step; the swap must not follow.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.9.0");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = remote_node(false, Vec::new());
        let mut exec = FakeExec::new();
        exec.fail_run_matching = Some("e_machine".into());
        let err = deploy_node(&exec, &node, "0.9.0", &staged, "makenotwork")
            .await
            .expect_err("arch mismatch must fail the deploy");
        assert!(
            format!("{err:#}").contains("architecture"),
            "error should mention the arch check: {err:#}"
        );
        let log = exec.log();
        assert!(
            !log.iter().any(|c| c.contains("reload-or-restart")),
            "swap must not run after a failed arch guard: {log:#?}"
        );
    }

    #[tokio::test]
    async fn deploy_remote_skips_config_check_when_node_opts_out() {
        // No config_check_env_file => the pre-swap config check is skipped, but
        // the rest of the choreography (including the swap) still runs.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.9.0");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = remote_node(false, Vec::new());
        let exec = FakeExec::new();
        deploy_node(&exec, &node, "0.9.0", &staged, "makenotwork")
            .await
            .unwrap();
        let log = exec.log();
        assert!(
            !log.iter().any(|c| c.contains("MNW_CHECK_CONFIG=1")),
            "config check must be skipped when the node opts out: {log:#?}"
        );
        assert!(
            log.iter().any(|c| c.contains("reload-or-restart")),
            "the swap must still run: {log:#?}"
        );
    }

    #[tokio::test]
    async fn deploy_remote_installs_companion_after_the_swap() {
        // Companions are After= the server: their install must land after the
        // symlink swap + service restart, never before.
        let tmp = tempfile::tempdir().unwrap();
        let staged = tmp.path().join("releases").join("0.9.0");
        tokio::fs::create_dir_all(&staged).await.unwrap();

        let node = remote_node(false, vec![companion()]);
        let exec = FakeExec::new();
        deploy_node(&exec, &node, "0.9.0", &staged, "makenotwork")
            .await
            .unwrap();
        let log = exec.log();
        assert!(
            pos(&log, "reload-or-restart") < pos(&log, "install-companion.sh"),
            "companion install must follow the swap: {log:#?}"
        );
    }
}