//! 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::domain::Platform;
use crate::retention::PinnedReleases;
use crate::topology::Node;
use anyhow::{Context, Result};
use async_trait::async_trait;
use ops_core::base_image;
use ops_exec::{Action, Executor, LogSink, RunOutput, Step, SyncOpts, sh_quote};
use std::path::{Path, PathBuf};
use tokio::process::Command;

/// A staged bundle proven to be for the node it is about to be pushed to.
///
/// This exists because "ship aarch64 bytes to an x86_64 box" was, until pom, a
/// mistake nobody could make: one product, one build host, one architecture, so
/// the pairing of a bundle and a node was correct by having no alternative. pom
/// has two architectures under one version, so the pairing becomes a real
/// choice, and a wrong choice deploys a binary the node cannot exec.
///
/// The answer is not a check before the call. A check is something a later
/// caller forgets, and the failure it guards is discovered by a production node
/// failing to start. [`Placement::check`] is the *only* way to obtain one of
/// these, and [`deploy_node`] takes one instead of a loose `(node, dir)` pair —
/// so a mismatched deploy is not a bug the code has to avoid, it is a value the
/// code cannot construct.
#[derive(Debug, Clone)]
pub struct Placement<'a> {
    node: &'a Node,
    bundle: &'a Path,
}

/// Why a bundle may not be placed on a node.
///
/// All four cases are refusals, including both "one side said nothing" cases.
/// Silence is not agreement: a node that does not state its platform cannot
/// vouch that it runs a bundle built for a stated one, and a bundle that does
/// not state its platform cannot satisfy a node that requires one. The only
/// admissible pairing besides a match is both sides silent, which is the
/// single-platform world Sando lived in and MNW still lives in.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum PlacementError {
    #[error(
        "node {node} runs {node_platform} and this bundle was built for {artifact_platform}; \
         refusing to deploy a binary the node cannot execute"
    )]
    Mismatch {
        node: String,
        node_platform: Platform,
        artifact_platform: Platform,
    },
    #[error(
        "node {node} does not declare a platform, and this bundle was built for \
         {artifact_platform}. Declare `platform` on the node so the two can be compared"
    )]
    NodeSilent {
        node: String,
        artifact_platform: Platform,
    },
    #[error(
        "node {node} requires {node_platform} and this bundle records no platform. \
         An artifact whose platform is unknown cannot be shown to satisfy one that is"
    )]
    ArtifactSilent {
        node: String,
        node_platform: Platform,
    },
}

impl<'a> Placement<'a> {
    /// The one constructor. `artifact` is the platform the bundle records, which
    /// for an accepted artifact comes from its `ArtifactRecord` provenance and
    /// for a Sando-built one comes from the app config.
    pub fn check(
        node: &'a Node,
        bundle: &'a Path,
        artifact: Option<&Platform>,
    ) -> Result<Self, PlacementError> {
        match (node.platform.as_ref(), artifact) {
            (Some(n), Some(a)) if n == a => Ok(Self { node, bundle }),
            (Some(n), Some(a)) => Err(PlacementError::Mismatch {
                node: node.name.to_string(),
                node_platform: n.clone(),
                artifact_platform: a.clone(),
            }),
            (None, Some(a)) => Err(PlacementError::NodeSilent {
                node: node.name.to_string(),
                artifact_platform: a.clone(),
            }),
            (Some(n), None) => Err(PlacementError::ArtifactSilent {
                node: node.name.to_string(),
                node_platform: n.clone(),
            }),
            // Both silent: the single-platform world. MNW is here, and stays
            // here until its nodes declare a platform — at which point its
            // builds have to as well, which is the forcing function rather than
            // a silently mixed state.
            (None, None) => Ok(Self { node, bundle }),
        }
    }

    pub fn node(&self) -> &'a Node {
        self.node
    }

    pub fn bundle(&self) -> &'a Path {
        self.bundle
    }
}

/// Keep at least 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.
///
/// A floor, not a ceiling. Whatever the deployed state still references is set
/// aside first and this count applies to the remainder — see
/// [`crate::retention`] for why a count alone could not express that.
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.
///
/// `pinned` names the release dirs the deployed state still points at
/// ([`crate::retention::pinned_dirs`]); they are never gc'd here, however old
/// they are. It arrives as data rather than as a pool handle so this stays a
/// filesystem operation and the intake seam keeps working without a database.
pub async fn finalize_local_release(
    release_root: &Path,
    staging: &Path,
    digest16: &str,
    pinned: &PinnedReleases,
) -> 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, pinned).await {
        tracing::warn!(error = %e, "local release GC failed (non-fatal)");
    }
    Ok(released)
}

/// Deploy a [`Placement`]'s bundle to its node using `executor` (the node's
/// 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.
///
/// The bundle and the node arrive together inside the placement, so there is no
/// signature here that accepts a bundle and a node that were never compared.
///
/// `primary_bin` is only used for logging — every file present in the staged
/// dir gets shipped.
///
/// `pinned` protects the node's own `releases/` from the gc that runs after a
/// successful remote deploy. The node mirrors the host's directory name, so the
/// set is the host's ([`crate::retention::pinned_dirs`]) with nothing recomputed
/// per node. `None` says the caller could not determine it and the gc is skipped
/// rather than run blind; `Some(PinnedReleases::none())` says there is genuinely
/// nothing deployed to protect. The distinction matters — the first is ignorance
/// and the second is knowledge — which is why this is an `Option` and not an
/// empty set standing in for both.
///
/// Unused on the `ssh_target=local` path: that deploy is a symlink swap over a
/// store the host gc already owns.
pub async fn deploy_node(
    executor: &dyn Executor,
    placement: Placement<'_>,
    version: &str,
    primary_bin: &str,
    pinned: Option<&PinnedReleases>,
) -> Result<PathBuf> {
    let node = placement.node();
    let staged_release_dir = placement.bundle();
    // 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,
        pinned,
    )
    .await
}

/// Where a node deploy failed, relative to the symlink swap.
///
/// The distinction is the whole difference between "nothing happened" and "go
/// look at production now", and it used to be carried only in the wording of a
/// `.context()` string, which meant the reporting layer could not act on it. It
/// reported every rollback failure as though the node were stranded on the new
/// version — including the case where the node had never left the old one,
/// which is the safe case and the common one.
///
/// Attached as `anyhow` context, so it both reads correctly in the error chain
/// and can be recovered with `stage_of`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FailureStage {
    /// Failed before the swap ran. `current` still points at the old release
    /// and the service was never restarted, so the node is on the OLD version.
    /// Nothing is stranded and nothing needs doing.
    BeforeSwap,
    /// Failed at or after the swap. The node's version is not knowable from
    /// here: the swap script rolls `current` back if the restart fails, but a
    /// failure between the two, or in a companion after the server is already
    /// live, can leave the node on either version.
    AtOrAfterSwap,
}

impl std::fmt::Display for FailureStage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BeforeSwap => {
                f.write_str("current symlink left intact; node is on the previous version")
            }
            Self::AtOrAfterSwap => {
                f.write_str("the symlink swap had already run; node version is indeterminate")
            }
        }
    }
}

/// Recover the [`FailureStage`] from a deploy error's context chain.
///
/// `None` means the error predates the stage annotation or came from somewhere
/// that does not set one. Callers must treat that as indeterminate rather than
/// as safe: guessing "before the swap" would reintroduce the bug in the
/// opposite, worse direction.
pub fn stage_of(err: &anyhow::Error) -> Option<FailureStage> {
    // anyhow's own downcast_ref searches attached context values, not just the
    // source chain, which is where a `.context(FailureStage::…)` lands.
    err.downcast_ref::<FailureStage>().copied()
}

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

/// Confirm the node is what the topology says it is, before anything is pushed.
///
/// The declared half of what the pre-swap `ldd` guard does at runtime. That
/// guard compares a specific binary against a specific box and is the last line;
/// this compares the box against its own declaration and is the first. Both
/// exist on purpose: the guard catches a binary nobody declared anything about,
/// and this catches a machine that stopped being what the config claims, which
/// the guard can only report as a symbol it cannot resolve.
///
/// A node that declares nothing is not checked. That is a skip and it is logged
/// as one, so an unchecked node is visible rather than looking checked. See
/// [`ops_core::base_image`] for why this differs from [`Placement::check`].
async fn check_node_identity(executor: &dyn Executor, node: &Node) -> Result<()> {
    if node.base_image.is_none() && node.libc.is_none() {
        tracing::info!(
            node = %node.name,
            "deploy: node declares no base image; identity not checked"
        );
        return Ok(());
    }
    let out = run_checked(
        executor,
        &base_image::probe_cmd(),
        "asking the node what it is",
    )
    .await
    .context(FailureStage::BeforeSwap)?;
    let reported = base_image::parse_probe(&String::from_utf8_lossy(&out.stdout));
    match base_image::check(
        node.name.as_str(),
        node.base_image.as_ref(),
        node.libc.as_deref(),
        &reported,
    ) {
        Ok(checked) => {
            if let Some(what) = checked {
                tracing::info!(node = %node.name, "deploy: identity checked, {what}");
            }
            Ok(())
        }
        Err(drift) => Err(anyhow::Error::new(drift))
            .context("the node is not what the topology declares it to be")
            .context(FailureStage::BeforeSwap),
    }
}

/// Refuse a bundle whose glibc floor is above what the node declares, before
/// the rsync.
///
/// The weak, early half of a pair. `ldd_guard_script` runs the node's own loader
/// against the actual bytes one step before the symlink swap, which covers every
/// shared library and every symbol version rather than glibc alone. Nothing here
/// replaces it, and a bundle that passes this can still fail that.
///
/// What this adds is *when*. The loader check happens after the bundle is built
/// and rsynced; this happens before either, so the subset of failures that two
/// declared numbers already prove is refused at the start of the promote instead
/// of most of the way through it. The zero-margin state on production makes that
/// subset a live one: three of its five binaries sit exactly on the box's glibc,
/// so a build host drifting one point release ahead puts every promote here.
///
/// Skipped, and logged as skipped, in all three cases where there is nothing to
/// compare: the node declares no `libc`, the bundle states no floor (static, or
/// no ELF this parser reads), or the declared `libc` is not a version string.
/// The last is a config typo rather than a bad bundle, and refusing a deploy
/// over it would be answering the wrong question loudly.
async fn check_bundle_fits_node(node: &Node, staged_release_dir: &Path) -> Result<()> {
    let Some(declared) = node.libc.as_deref() else {
        return Ok(());
    };
    let Some(node_libc) = crate::elf::GlibcVersion::parse(declared) else {
        tracing::warn!(
            node = %node.name,
            declared,
            "deploy: node's declared libc is not a version; glibc floor not compared"
        );
        return Ok(());
    };
    let digest = crate::bundle::digest_dir(staged_release_dir)
        .await
        .context("reading the staged bundle's glibc floor")
        .context(FailureStage::BeforeSwap)?;
    let Some(floor) = digest.glibc_floor else {
        tracing::info!(
            node = %node.name,
            "deploy: bundle states no glibc floor; nothing to compare"
        );
        return Ok(());
    };
    if floor > node_libc {
        return Err(anyhow::anyhow!(
            "this bundle needs glibc {floor} and `{node}` declares {node_libc}; \
             refusing to ship a binary the node cannot load. Either the build host \
             drifted ahead of the node, or the node's declared libc is stale",
            node = node.name,
        ))
        .context(FailureStage::BeforeSwap);
    }
    tracing::info!(
        node = %node.name,
        "deploy: glibc floor {floor} fits the node's {node_libc}"
    );
    Ok(())
}

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

    // Identity check first, before a single byte moves. A node that was rebuilt
    // into something else is refused here rather than after an rsync, and long
    // before the pre-swap `ldd` guard would have caught the consequence without
    // naming the cause. Cheap: one shell round-trip that reads /etc/os-release.
    check_node_identity(executor, node).await?;

    // And that the bundle could load there at all, from two numbers, before the
    // bytes move. The `ldd` guard below asks the stronger question on the node
    // itself; this one is only earlier.
    check_bundle_fits_node(node, staged_release_dir).await?;

    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
    .context(FailureStage::BeforeSwap)?;

    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")
        .context(FailureStage::BeforeSwap)?;

    // 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")
    .context(FailureStage::BeforeSwap)?;

    // 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")
    .context(FailureStage::BeforeSwap)?;

    // And that the node can actually resolve what the binary links, which the
    // arch check above cannot see: right architecture, right ELF, and still
    // unrunnable because it wants a glibc symbol version this box does not have.
    // Bento used to catch that at build time; under the Sando/Bento boundary the
    // builder no longer knows which machine runs the bytes, so it lands here.
    run_checked(
        executor,
        &ldd_guard_script(&deployed_bin),
        "verifying the node can resolve the binary's dynamic dependencies",
    )
    .await
    .context("the target node cannot satisfy the deployed binary's dynamic dependencies")
    .context(FailureStage::BeforeSwap)?;

    // The same two guards for every companion, and for the same reason. Both
    // checks above take the primary binary alone, so a companion of the wrong
    // architecture, or one linking a symbol version this node lacks, used to be
    // discovered by its unit failing to start — during the install loop below,
    // which runs AFTER the swap. That is the expensive side of the line these
    // guards exist to stay on: the promote fails either way, but with the server
    // already restarted onto the new release.
    //
    // The companion bytes are present and verified by now: they arrived in the
    // same rsync and `manifest_verify_script` above covers the whole release
    // directory, `companions/` included. So there is nothing to wait for.
    for c in &node.companions {
        let src = companion_src(&release_dir, &c.name);
        run_checked(
            executor,
            &arch_guard_script(&src),
            "verifying companion arch matches node",
        )
        .await
        .with_context(|| {
            format!(
                "companion {} architecture does not match the target node",
                c.name
            )
        })
        .context(FailureStage::BeforeSwap)?;
        run_checked(
            executor,
            &ldd_guard_script(&src),
            "verifying the node can resolve the companion's dynamic dependencies",
        )
        .await
        .with_context(|| {
            format!(
                "the target node cannot satisfy companion {}'s dynamic dependencies",
                c.name
            )
        })
        .context(FailureStage::BeforeSwap)?;
    }

    // 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")
            .context(FailureStage::BeforeSwap)?;
    }

    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
    .context(FailureStage::AtOrAfterSwap)?;

    // 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 = companion_src(&release_dir, &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
                )
            })
            .context(FailureStage::AtOrAfterSwap)?;
    }

    // No pinned set means the caller could not determine what is referenced,
    // and a gc that cannot tell is the exact failure this parameter exists to
    // stop. Skipping costs disk on the node; running blind costs the artifact a
    // rollback resolves to. `finalize_local_release` takes the same position by
    // refusing to publish at all when the pin query fails.
    match pinned {
        Some(pinned) => {
            if let Err(e) = gc_remote_releases(executor, release_root, pinned).await {
                tracing::warn!(error = %e, "remote release GC failed (non-fatal)");
            }
        }
        None => tracing::warn!(
            node = %node.name,
            "remote release GC skipped: the pinned set is unknown, and a gc that \
             cannot see what is referenced is what stranded the host store twice"
        ),
    }

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

/// Where a companion's binary sits inside the staged release directory.
///
/// One function because two places need it and they must not drift: the guards
/// that run before the swap check this path, and the installer after the swap
/// reads it. A guard that checked a path the installer did not use would be a
/// check of nothing, and would look exactly like a passing check.
fn companion_src(release_dir: &str, name: &str) -> String {
    format!("{release_dir}/companions/{name}")
}

/// 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<()> {
    // Readability first, as its own step with its own message.
    //
    // The env file is read by this check AS THE DEPLOY USER, and it is the only
    // thing that does. systemd loads `EnvironmentFile=` as root before dropping
    // to `User=`, so the running service does not care about the mode — which
    // means a file rewritten 0600 breaks the next deploy while the current one
    // keeps serving, and the breakage is invisible until someone ships. That is
    // exactly how prod deploy 0.11.3 failed on 2026-08-01.
    //
    // Without this step the operator gets `bash: line 9: <file>: Permission
    // denied` out of a generated script and has to reverse-engineer which user
    // and which file. Naming the user, the mode and the owner turns that into a
    // one-line read.
    let probe = readability_probe_script(env_file);
    if let Ok(Err(e)) = tokio::time::timeout(
        std::time::Duration::from_secs(20),
        run_checked(executor, &probe, "env file readability"),
    )
    .await
    {
        return Err(e).context(format!(
            "the deploy user cannot read {env_file}. systemd reads EnvironmentFile= as root, so \
             the running service is unaffected and this breaks only deploys. Expected mode 0640 \
             owned root:<service user> (see sando/deploy/bootstrap-node.sh); something that \
             rewrote the file likely did so with a 077 umask"
        ));
    }

    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"
        ),
    }
}

/// Assert the deploy user can read `env_file`, reporting who it is and what the
/// file actually looks like when it cannot.
///
/// `stat` output is best-effort: a node without it (or a file that does not
/// exist) still gets the identity line, which is the half an operator cannot
/// derive from the failure on their own.
fn readability_probe_script(env_file: &str) -> String {
    format!(
        "if [ ! -e {env} ]; then\n\
         \techo \"{env_disp}: does not exist on this node\" >&2; exit 1\n\
         fi\n\
         if [ ! -r {env} ]; then\n\
         \techo \"cannot read {env_disp} as $(id -un) (groups: $(id -Gn))\" >&2\n\
         \tstat -c 'actual: mode %a owner %U:%G' {env} >&2 2>/dev/null || true\n\
         \texit 1\n\
         fi\n",
        env = sh_quote(env_file),
        env_disp = env_file,
    )
}

/// 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),
    )
}

/// Refuse a binary whose dynamic dependencies the node cannot satisfy, before
/// the symlink swap.
///
/// The sibling of [`arch_guard_script`], and it exists because the boundary took
/// the check away from the builder. Bento's recipe used to compare the binary's
/// highest `GLIBC_` symbol against `ldd --version` on the service host, which it
/// could only do while it held a `[[deploy]]` entry naming that host. A
/// handed-off service has none: which machine runs the bytes is environment
/// knowledge, which is Sando's half. So the check moves here, where the node and
/// the artifact are already in the same value.
///
/// It asks the stronger question, because here it can. Bento compared two
/// version numbers from two machines; this runs the node's own loader against
/// the bytes that were just rsynced onto it. That covers every shared library
/// and every symbol version, not glibc alone, and it answers "will this exec
/// here" rather than "is this number smaller than that one".
///
/// Three outcomes, and only one of them fails:
///
/// - `not found` in `ldd` output — a missing library or an unsatisfiable symbol
///   version. This is the failure, and it is exactly what would otherwise be
///   discovered by the unit failing to start after the swap.
/// - not a dynamic executable — `ldd` exits non-zero and says so. A static
///   binary has nothing to resolve, so it passes.
/// - no `ldd` on the node — nothing to check with. Logged and passed: "cannot
///   verify" is not "known bad", the same call `arch_guard_script` makes for an
///   unmapped arch.
///
/// `ldd` runs the loader, which for an arbitrary binary is code execution. These
/// bytes are ours, already verified against their MANIFEST on this node, and
/// about to be exec'd by the service unit a second later.
fn ldd_guard_script(bin: &str) -> String {
    format!(
        "set -e\n\
         bin={bin}\n\
         command -v ldd >/dev/null 2>&1 || {{ echo \"deploy: ldd check skipped (no ldd on node)\" >&2; exit 0; }}\n\
         out=$(ldd \"$bin\" 2>&1) || {{ \n\
         case \"$out\" in\n\
         *\"not a dynamic executable\"*) echo \"deploy: ldd check passed (static binary)\" >&2; exit 0 ;;\n\
         *) echo \"deploy: ldd failed on $bin: $out\" >&2; exit 1 ;;\n\
         esac\n\
         }}\n\
         if printf '%s' \"$out\" | grep -q 'not found'; then\n\
         echo \"deploy: this node cannot satisfy the binary's dynamic dependencies:\" >&2\n\
         printf '%s\\n' \"$out\" | grep 'not found' >&2\n\
         exit 1\n\
         fi\n",
        bin = sh_quote(bin),
    )
}

/// Trim `releases/` to the pinned set plus the [`RELEASES_TO_KEEP`] newest of
/// what is left.
///
/// Pinning is applied before the count, so a referenced artifact cannot be aged
/// out by rebuilds of a newer version — the failure that stranded production
/// twice. See [`crate::retention`].
async fn gc_local_releases(release_root: &Path, pinned: &PinnedReleases) -> 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;
        }
        // Set aside before anything is ordered or counted: a pinned dir is not
        // a candidate, so it can never occupy one of the count's slots either.
        if entry
            .file_name()
            .to_str()
            .is_some_and(|n| pinned.contains(n))
        {
            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(())
}

/// Trim a node's `releases/` to the pinned set plus the [`RELEASES_TO_KEEP`]
/// newest of what is left.
///
/// The node mirrors the host's directory name (see [`deploy_node`]), so the same
/// [`PinnedReleases`] the host gc subtracts is the right set here — nothing has
/// to be recomputed per node.
///
/// Weaker instance of the host defect: every promote and rollback rsyncs from
/// the host store, so a node directory evicted here is re-pushed rather than
/// lost. What it costs is a full rsync of a bundle the node already had, at the
/// worst moment — mid-rollback, with a tier already failing.
///
/// Pinning is applied before the count, matching [`gc_local_releases`]: a pinned
/// directory is not a candidate, so it cannot occupy one of the count's slots.
async fn gc_remote_releases(
    executor: &dyn Executor,
    release_root: &str,
    pinned: &PinnedReleases,
) -> Result<()> {
    run_checked(
        executor,
        &remote_gc_script(release_root, pinned),
        "remote release gc",
    )
    .await
    .map(|_| ())
}

/// The remote gc as shell.
///
/// Split out so the script is testable without an executor: it is the half of
/// this that can be wrong in a way `rm -rf` makes expensive.
///
/// `ls -1t` orders by mtime desc. Pinned names are carried in the positional
/// parameters rather than a here-doc piped through `grep -v`, because `grep`
/// exits 1 when it selects no lines — which happens on both edges that matter
/// (nothing pinned, or everything pinned) and would abort the script under
/// `set -e` for the two cases that are perfectly normal. Comparing with `case`
/// has no exit status to trip over, and matches whole names rather than
/// substrings, which a `grep -F` without `-x` would not.
fn remote_gc_script(release_root: &str, pinned: &PinnedReleases) -> String {
    let pins = pinned
        .sorted_names()
        .into_iter()
        .map(sh_quote)
        .collect::<Vec<_>>()
        .join(" ");
    // `set --` with no operands unsets the positional parameters, which is
    // exactly what a nothing-pinned gc wants: the `for` below then iterates zero
    // times and every directory is a candidate. Written as one branch because
    // `set -- ` with an empty expansion is the same statement.
    let set_pins = format!("set -- {pins}");
    format!(
        "set -e; cd {root}/releases 2>/dev/null || exit 0; \
         {set_pins}; \
         n=0; \
         ls -1t | while IFS= read -r d; do \
           for p in \"$@\"; do \
             if [ \"$d\" = \"$p\" ]; then continue 2; fi; \
           done; \
           n=$((n+1)); \
           if [ \"$n\" -le {keep} ]; then continue; fi; \
           rm -rf -- \"$d\"; \
         done",
        root = sh_quote(release_root),
        keep = RELEASES_TO_KEEP,
    )
}

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

    /// Nothing deployed, so nothing pinned: the tests that exercise the count
    /// alone pass this, and the ones that exercise pinning build their own set.
    fn no_pins() -> PinnedReleases {
        PinnedReleases::none()
    }

    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;

    // ---- placement ----
    //
    // The whole table, because the interesting cases are the two where one side
    // said nothing. Treating silence as agreement is how a wrong-architecture
    // deploy would get through, and it is the shape a "check it before you call"
    // guard tends to end up with.

    fn node_on(platform: Option<&str>) -> Node {
        Node {
            name: crate::domain::NodeId::new("n1"),
            ssh_target: "deploy@n1".into(),
            release_root: "/opt/x".into(),
            platform: platform.map(|p| Platform::parse(p).unwrap()),
            base_image: None,
            libc: None,
            service_name: "x.service".into(),
            config_check_env_file: None,
            actuate: crate::topology::default_actuate(),
            observe: crate::topology::default_observe(),
            health_url: None,
            companions: Vec::new(),
        }
    }

    /// A node that declares a glibc older than the bundle needs is refused
    /// before the rsync, and the message names both numbers so the operator
    /// knows which side to fix.
    #[tokio::test]
    async fn a_bundle_above_the_node_s_declared_glibc_is_refused_before_the_rsync() {
        let dir = tempfile::tempdir().unwrap();
        let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
        std::fs::write(dir.path().join("bin"), &exe).unwrap();
        let Some(floor) = crate::elf::glibc_floor(&exe) else {
            return; // a static test binary states no floor; nothing to compare
        };

        let mut node = node_on(None);
        node.libc = Some("2.0".into()); // older than anything real
        let err = check_bundle_fits_node(&node, dir.path())
            .await
            .expect_err("a bundle above the node's glibc must be refused");
        // `{:#}` walks the context chain: the outermost context is the
        // `FailureStage`, whose Display is the operator-facing "nothing moved"
        // line, and the cause below it is the reason.
        let msg = format!("{err:#}");
        assert!(
            msg.contains(&floor.to_string()) && msg.contains("2.0"),
            "the refusal must name both numbers: {msg}"
        );
        assert_eq!(
            stage_of(&err),
            Some(FailureStage::BeforeSwap),
            "refusing here must be recoverable: nothing has moved yet"
        );
    }

    #[tokio::test]
    async fn a_bundle_within_the_node_s_declared_glibc_passes() {
        let dir = tempfile::tempdir().unwrap();
        let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
        std::fs::write(dir.path().join("bin"), &exe).unwrap();

        let mut node = node_on(None);
        node.libc = Some("99.0".into()); // newer than anything real
        check_bundle_fits_node(&node, dir.path())
            .await
            .expect("a bundle the node can load must pass");
    }

    /// The three ways there is nothing to compare. All three pass, because
    /// "cannot verify" is not "known bad" — the same call `arch_guard_script`
    /// makes for an unmapped architecture.
    #[tokio::test]
    async fn nothing_to_compare_is_a_pass_not_a_refusal() {
        let dir = tempfile::tempdir().unwrap();
        let exe = std::fs::read(std::env::current_exe().unwrap()).unwrap();
        std::fs::write(dir.path().join("bin"), &exe).unwrap();

        // 1. The node declares no libc.
        let node = node_on(None);
        check_bundle_fits_node(&node, dir.path()).await.unwrap();

        // 2. The node's declared libc is not a version (a config typo).
        let mut typo = node_on(None);
        typo.libc = Some("noble".into());
        check_bundle_fits_node(&typo, dir.path()).await.unwrap();

        // 3. The bundle holds no ELF, so it states no floor.
        let empty = tempfile::tempdir().unwrap();
        std::fs::write(empty.path().join("style.css"), b"body{}").unwrap();
        let mut strict = node_on(None);
        strict.libc = Some("2.0".into());
        check_bundle_fits_node(&strict, empty.path())
            .await
            .expect("a bundle with no binaries has no floor to exceed");
    }

    #[test]
    fn matching_platforms_are_placeable() {
        let node = node_on(Some("linux/aarch64"));
        let art = Platform::parse("linux/aarch64").unwrap();
        let p = Placement::check(&node, Path::new("/r/abc"), Some(&art)).expect("a match places");
        assert_eq!(p.bundle(), Path::new("/r/abc"));
        assert_eq!(p.node().name.as_str(), "n1");
    }

    #[test]
    fn a_different_architecture_is_refused() {
        // The failure this type exists for: pom's aarch64 bundle reaching the
        // x86_64 box, which execs nothing and takes the watcher down.
        let node = node_on(Some("linux/x86_64"));
        let art = Platform::parse("linux/aarch64").unwrap();
        let err = Placement::check(&node, Path::new("/r/abc"), Some(&art)).unwrap_err();
        assert!(
            matches!(err, PlacementError::Mismatch { .. }),
            "expected a mismatch, got {err}"
        );
        // The message has to name both, or an operator cannot tell which half
        // is wrong.
        let msg = err.to_string();
        assert!(
            msg.contains("linux/x86_64") && msg.contains("linux/aarch64"),
            "{msg}"
        );
    }

    #[test]
    fn a_silent_node_refuses_a_stated_artifact() {
        // Not "the node probably runs it". A node that never said what it is
        // cannot vouch for a bundle that did, and the pairing that looks
        // harmless here is exactly the one that ships the wrong half of a
        // two-architecture release.
        let node = node_on(None);
        let art = Platform::parse("linux/aarch64").unwrap();
        assert!(matches!(
            Placement::check(&node, Path::new("/r/abc"), Some(&art)),
            Err(PlacementError::NodeSilent { .. })
        ));
    }

    #[test]
    fn a_stated_node_refuses_a_silent_artifact() {
        let node = node_on(Some("linux/aarch64"));
        assert!(matches!(
            Placement::check(&node, Path::new("/r/abc"), None),
            Err(PlacementError::ArtifactSilent { .. })
        ));
    }

    #[test]
    fn both_silent_is_the_single_platform_world_and_still_places() {
        // MNW is here and stays here. Its nodes declare nothing and its builds
        // record nothing, which is the truth about a product with one build host
        // and one architecture. The moment either side starts stating, the other
        // has to as well — that is the forcing function, and it is why this cell
        // is the only admissible non-match.
        let node = node_on(None);
        Placement::check(&node, Path::new("/r/abc"), None).expect("the pre-pom world still ships");
    }

    #[test]
    fn platform_parsing_is_a_shape_not_a_spelling() {
        assert_eq!(
            Platform::parse("Linux/AArch64").unwrap(),
            Platform::parse("linux/aarch64").unwrap(),
            "case is not a distinction between two machines"
        );
        for bad in ["linux", "linux/", "/aarch64", "linux/aarch64/gnu", ""] {
            assert!(Platform::parse(bad).is_err(), "{bad:?} should not parse");
        }
    }

    // ---- failure stage ----
    //
    // The 2026-08-01 prod deploy failed its pre-swap config check, and the
    // rollback then failed the same way — which left the node safely on the old
    // version, and was reported as "it remains on the new version, manual
    // intervention needed". These pin the distinction the reporting layer now
    // depends on.

    #[test]
    fn a_pre_swap_failure_is_recoverable_as_such() {
        let e = anyhow::anyhow!("Permission denied")
            .context("pre-swap config check failed")
            .context(FailureStage::BeforeSwap);
        assert_eq!(stage_of(&e), Some(FailureStage::BeforeSwap));
        // The reason survives alongside the stage; the stage does not replace it.
        let rendered = format!("{e:#}");
        assert!(
            rendered.contains("pre-swap config check failed"),
            "{rendered}"
        );
        assert!(rendered.contains("Permission denied"), "{rendered}");
    }

    #[test]
    fn a_post_swap_failure_is_recoverable_as_such() {
        let e = anyhow::anyhow!("unit failed to start")
            .context("companion x deploy failed (server already swapped)")
            .context(FailureStage::AtOrAfterSwap);
        assert_eq!(stage_of(&e), Some(FailureStage::AtOrAfterSwap));
    }

    #[test]
    fn an_unannotated_failure_has_no_stage() {
        // Must be None, not a default. A caller seeing None has to treat the
        // node as indeterminate; inferring "before the swap" would reintroduce
        // the original bug pointing the other way, which is the dangerous way.
        let e = anyhow::anyhow!("something older, from before stages existed");
        assert_eq!(stage_of(&e), None);
    }

    // ---- env file readability probe ----

    #[tokio::test]
    async fn readability_probe_passes_on_a_readable_file() {
        let tmp = tempfile::tempdir().unwrap();
        let f = tmp.path().join("ok.env");
        tokio::fs::write(&f, "A=1\n").await.unwrap();
        let script = readability_probe_script(&f.to_string_lossy());
        let out = run_checked(&local_executor(), &script, "probe").await;
        assert!(out.is_ok(), "{:?}", out.err().map(|e| format!("{e:#}")));
    }

    #[tokio::test]
    async fn readability_probe_names_the_user_and_mode_when_unreadable() {
        // Root can read anything, so a mode-based test would pass spuriously
        // there. Skip rather than assert something false. No libc dependency
        // for one probe: a 0-mode temp file is readable iff we are root.
        let probe_dir = tempfile::tempdir().unwrap();
        let probe_file = probe_dir.path().join("root-check");
        tokio::fs::write(&probe_file, "x").await.unwrap();
        tokio::fs::set_permissions(
            &probe_file,
            std::os::unix::fs::PermissionsExt::from_mode(0o000),
        )
        .await
        .unwrap();
        if tokio::fs::read(&probe_file).await.is_ok() {
            return; // running as root
        }
        let tmp = tempfile::tempdir().unwrap();
        let f = tmp.path().join("locked.env");
        tokio::fs::write(&f, "A=1\n").await.unwrap();
        tokio::fs::set_permissions(&f, std::os::unix::fs::PermissionsExt::from_mode(0o000))
            .await
            .unwrap();

        let script = readability_probe_script(&f.to_string_lossy());
        let err = run_checked(&local_executor(), &script, "probe")
            .await
            .expect_err("an unreadable file must fail the probe");
        let msg = format!("{err:#}");
        // The two things the raw bash error does not tell you.
        assert!(msg.contains("cannot read"), "{msg}");
        assert!(msg.contains("mode 0") || msg.contains("mode "), "{msg}");
    }

    #[tokio::test]
    async fn readability_probe_distinguishes_missing_from_unreadable() {
        let tmp = tempfile::tempdir().unwrap();
        let missing = tmp.path().join("nope.env");
        let script = readability_probe_script(&missing.to_string_lossy());
        let err = run_checked(&local_executor(), &script, "probe")
            .await
            .expect_err("a missing file must fail the probe");
        let msg = format!("{err:#}");
        assert!(msg.contains("does not exist"), "{msg}");
    }

    #[test]
    fn the_two_stages_read_differently() {
        // These strings end up in an operator's terminal during an incident.
        let before = FailureStage::BeforeSwap.to_string();
        let after = FailureStage::AtOrAfterSwap.to_string();
        assert!(before.contains("previous version"), "{before}");
        assert!(after.contains("indeterminate"), "{after}");
        assert_ne!(before, after);
    }

    /// 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", &no_pins())
                .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", &no_pins())
            .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", &no_pins())
            .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", &no_pins())
            .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", &no_pins())
            .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, &no_pins()).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_never_evicts_a_pinned_dir() {
        // The 2026-08-25 shape exactly: the oldest dir is the one production is
        // running, and enough newer rebuilds exist to push it past the count.
        // Under the count alone it was the first thing deleted.
        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);
            f.set_times(std::fs::FileTimes::new().set_modified(when))
                .unwrap();
            names.push(name);
        }

        // The two oldest: what a tier is running and what it would roll back to.
        let pinned: PinnedReleases = [names[0].clone(), names[1].clone()].into_iter().collect();
        gc_local_releases(root, &pinned).await.unwrap();

        for name in [&names[0], &names[1]] {
            assert!(
                releases.join(name).exists(),
                "a referenced artifact was evicted: {name}"
            );
        }
        // And the count still applies to everything else, from a floor that the
        // pinned pair did not eat into: the newest RELEASES_TO_KEEP unpinned
        // dirs survive, so pinning two costs two extra slots rather than two of
        // the five.
        let unpinned: Vec<_> = names.iter().filter(|n| !pinned.contains(n)).collect();
        let cut = unpinned.len() - RELEASES_TO_KEEP;
        for name in unpinned.iter().take(cut) {
            assert!(
                !releases.join(name).exists(),
                "expected to be pruned: {name}"
            );
        }
        for name in unpinned.iter().skip(cut) {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
    }

    #[tokio::test]
    async fn gc_local_releases_keeps_a_pinned_dir_that_is_not_even_present() {
        // A pinned name with nothing on disk must not disturb the count. This is
        // the state the bug leaves behind, and gc runs again while it holds.
        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..=RELEASES_TO_KEEP {
            tokio::fs::create_dir(releases.join(format!("v{i}")))
                .await
                .unwrap();
        }
        let pinned: PinnedReleases = ["gone-already".to_string()].into_iter().collect();
        gc_local_releases(root, &pinned).await.unwrap();

        let left = std::fs::read_dir(&releases).unwrap().count();
        assert_eq!(left, RELEASES_TO_KEEP);
    }

    #[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, &no_pins()).await.unwrap();
        for i in 0..3 {
            assert!(releases.join(format!("v{i}")).exists());
        }
    }

    // ---- remote gc ----
    //
    // Driven through `LocalExec`, so these run the real shell the node runs
    // rather than asserting on the script's text. The script is the half of the
    // remote gc that can be wrong, and it is wrong with `rm -rf`.

    /// `releases/` with `total` dirs named `v00..`, oldest first by mtime.
    async fn releases_by_age(root: &Path, total: usize) -> Vec<String> {
        let releases = root.join("releases");
        tokio::fs::create_dir_all(&releases).await.unwrap();
        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();
            // A file inside, so a deletion is visible as more than an empty dir.
            tokio::fs::write(dir.join("makenotwork"), b"x")
                .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);
            f.set_times(std::fs::FileTimes::new().set_modified(when))
                .unwrap();
            names.push(name);
        }
        names
    }

    /// Parity with the count-only script this replaced: nothing pinned, newest
    /// `RELEASES_TO_KEEP` survive. If this drifts the change was not a
    /// refinement of the old behaviour but a replacement of it.
    #[tokio::test]
    async fn gc_remote_releases_keeps_last_n_by_mtime_when_nothing_is_pinned() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let total = RELEASES_TO_KEEP + 3;
        let names = releases_by_age(root, total).await;

        gc_remote_releases(&local_executor(), root.to_str().unwrap(), &no_pins())
            .await
            .unwrap();

        let releases = root.join("releases");
        for name in names.iter().take(total - RELEASES_TO_KEEP) {
            assert!(!releases.join(name).exists(), "expected pruned: {name}");
        }
        for name in names.iter().skip(total - RELEASES_TO_KEEP) {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
    }

    /// The done condition, on the node: the dirs a tier's current and previous
    /// artifacts name survive even when they are the oldest on disk and well
    /// past the count. Same shape as the host-store test above, which is the
    /// point — the two stores now answer the same question the same way.
    #[tokio::test]
    async fn gc_remote_releases_never_evicts_a_pinned_dir() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let total = RELEASES_TO_KEEP + 3;
        let names = releases_by_age(root, total).await;

        let pinned: PinnedReleases = [names[0].clone(), names[1].clone()].into_iter().collect();
        gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
            .await
            .unwrap();

        let releases = root.join("releases");
        for name in [&names[0], &names[1]] {
            assert!(
                releases.join(name).exists(),
                "a referenced artifact was evicted from the node: {name}"
            );
        }
        // And pinning does not spend the count's slots, again matching the host.
        let unpinned: Vec<_> = names.iter().filter(|n| !pinned.contains(n)).collect();
        let cut = unpinned.len() - RELEASES_TO_KEEP;
        for name in unpinned.iter().take(cut) {
            assert!(!releases.join(name).exists(), "expected pruned: {name}");
        }
        for name in unpinned.iter().skip(cut) {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
    }

    /// Every dir pinned means the loop deletes nothing and the script still
    /// exits 0. Worth its own test because the obvious implementation of this
    /// filter is `grep -v`, which exits 1 when it selects no lines and would
    /// have failed the deploy here under `set -e`.
    #[tokio::test]
    async fn gc_remote_releases_succeeds_when_everything_is_pinned() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let names = releases_by_age(root, RELEASES_TO_KEEP + 3).await;
        let pinned: PinnedReleases = names.iter().cloned().collect();

        gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
            .await
            .unwrap();

        let releases = root.join("releases");
        for name in &names {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
    }

    /// A `releases/` that does not exist is not an error: a node's first deploy
    /// creates the dir, and gc runs on the same path.
    #[tokio::test]
    async fn gc_remote_releases_is_a_noop_when_the_store_is_missing() {
        let tmp = tempfile::tempdir().unwrap();
        gc_remote_releases(&local_executor(), tmp.path().to_str().unwrap(), &no_pins())
            .await
            .unwrap();
    }

    /// Names reach the script as positional parameters, so a name that looks
    /// like shell must be compared whole rather than expanded or split. None of
    /// these can be a digest16, but the pre-identity names are version strings
    /// and the pinned set is data read out of a database.
    #[tokio::test]
    async fn gc_remote_releases_quotes_pinned_names() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let releases = root.join("releases");
        tokio::fs::create_dir_all(&releases).await.unwrap();
        let awkward = ["a b", "x'y", "*"];
        for name in awkward {
            tokio::fs::create_dir(releases.join(name)).await.unwrap();
        }
        // Enough newer dirs that the count alone would evict all three.
        let filler: Vec<String> = (0..=RELEASES_TO_KEEP).map(|i| format!("f{i}")).collect();
        for name in &filler {
            tokio::fs::create_dir(releases.join(name)).await.unwrap();
        }

        let pinned: PinnedReleases = awkward.iter().map(|s| (*s).to_string()).collect();
        gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
            .await
            .unwrap();

        for name in awkward {
            assert!(releases.join(name).exists(), "expected to survive: {name}");
        }
    }

    /// A pinned name matches a whole directory name, never a prefix of one.
    /// `case`-with-globbing or a `grep -F` without `-x` would keep `v0` and
    /// `v01` both because one contains the other, quietly widening the pinned
    /// set past what the database said.
    #[tokio::test]
    async fn gc_remote_releases_matches_whole_names_not_prefixes() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let names = releases_by_age(root, RELEASES_TO_KEEP + 3).await;

        // Pin the oldest by an exact name; its neighbours share the prefix.
        let pinned: PinnedReleases = [names[0].clone()].into_iter().collect();
        gc_remote_releases(&local_executor(), root.to_str().unwrap(), &pinned)
            .await
            .unwrap();

        let releases = root.join("releases");
        assert!(
            releases.join(&names[0]).exists(),
            "the pinned dir was evicted"
        );
        assert!(
            !releases.join(&names[1]).exists(),
            "a dir sharing the pinned name's prefix was treated as pinned"
        );
    }

    #[tokio::test]
    async fn gc_local_releases_noop_when_releases_dir_missing() {
        let tmp = tempfile::tempdir().unwrap();
        gc_local_releases(tmp.path(), &no_pins()).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 {
            platform: None,
            base_image: None,
            libc: None,
            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 placement = Placement::check(&node, &staged, None).expect("both sides silent");
        let result = deploy_node(&executor, placement, "0.0.1", "server", Some(&no_pins())).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 {
            platform: None,
            base_image: None,
            libc: None,
            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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.0.1",
            "server",
            Some(&no_pins()),
        )
        .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"
        );
    }

    // ---- ldd_guard_script: a binary this node cannot resolve fails closed ----

    /// A fake `ldd` on PATH that prints `body` and exits `code`, so the guard's
    /// three outcomes can be exercised without a binary that genuinely fails to
    /// link. The real `ldd` cannot be made to produce a `not found` on demand.
    async fn run_ldd_guard_with_fake(body: &str, code: i32) -> std::process::Output {
        let dir = tempfile::tempdir().unwrap();
        let fake = dir.path().join("ldd");
        std::fs::write(
            &fake,
            format!("#!/bin/sh\ncat <<'EOF'\n{body}\nEOF\nexit {code}\n"),
        )
        .unwrap();
        let mut perms = std::fs::metadata(&fake).unwrap().permissions();
        std::os::unix::fs::PermissionsExt::set_mode(&mut perms, 0o755);
        std::fs::set_permissions(&fake, perms).unwrap();
        let bin = dir.path().join("subject");
        std::fs::write(&bin, b"x").unwrap();
        Command::new("sh")
            .arg("-c")
            .arg(ldd_guard_script(&bin.to_string_lossy()))
            .env("PATH", format!("{}:/usr/bin:/bin", dir.path().display()))
            .output()
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn ldd_guard_fails_closed_on_an_unsatisfiable_symbol_version() {
        // The exact failure Bento's glibc_check used to catch at build time, and
        // the reason this guard exists: right arch, resolves every library, and
        // still cannot exec because the node's glibc is older than the build
        // host's.
        let out = run_ldd_guard_with_fake(
            "\tlinux-vdso.so.1 (0x00007fff)\n\
             \t/lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.40' not found (required by ./pom)\n\
             \tlibc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007f00)",
            0,
        )
        .await;
        assert!(
            !out.status.success(),
            "an unsatisfiable symbol version must fail before the symlink swap"
        );
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("GLIBC_2.40"),
            "the offending line must reach the operator, not just a verdict: {stderr}"
        );
    }

    #[tokio::test]
    async fn ldd_guard_fails_closed_on_a_missing_library() {
        let out = run_ldd_guard_with_fake("\tlibfoo.so.1 => not found", 0).await;
        assert!(!out.status.success(), "a missing library must fail closed");
    }

    #[tokio::test]
    async fn ldd_guard_passes_a_resolvable_binary() {
        let out = run_ldd_guard_with_fake(
            "\tlibc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007f00)",
            0,
        )
        .await;
        assert!(
            out.status.success(),
            "a fully resolved binary must pass: {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }

    #[tokio::test]
    async fn ldd_guard_passes_a_static_binary() {
        // ldd exits non-zero for these. Nothing to resolve is not a failure.
        let out = run_ldd_guard_with_fake("\tnot a dynamic executable", 1).await;
        assert!(
            out.status.success(),
            "a static binary has no dependencies to satisfy: {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }

    #[tokio::test]
    async fn ldd_guard_fails_when_ldd_errors_for_another_reason() {
        // Not the static case: ldd said something else and exited non-zero. We
        // do not know the binary is fine, so we do not say it is.
        let out = run_ldd_guard_with_fake("ldd: cannot read file", 1).await;
        assert!(
            !out.status.success(),
            "an unexplained ldd failure must not read as a pass"
        );
    }

    #[tokio::test]
    async fn ldd_guard_skips_when_the_node_has_no_ldd() {
        // Cannot verify is not known bad, matching arch_guard's unmapped-arch
        // call. PATH holds nothing, so `command -v ldd` finds none.
        let dir = tempfile::tempdir().unwrap();
        let bin = dir.path().join("subject");
        std::fs::write(&bin, b"x").unwrap();
        // Absolute path to the shell: PATH is what this test empties, so
        // resolving `sh` through it would fail before the script ever ran.
        let out = Command::new("/bin/sh")
            .arg("-c")
            .arg(ldd_guard_script(&bin.to_string_lossy()))
            .env("PATH", dir.path().display().to_string())
            .output()
            .await
            .unwrap();
        assert!(
            out.status.success(),
            "a node with no ldd must not fail the deploy: {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }

    #[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 {
            platform: None,
            base_image: None,
            libc: None,
            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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.0.1",
            "server",
            Some(&no_pins()),
        )
        .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 {
            platform: None,
            base_image: None,
            libc: None,
            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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .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:#?}"
        );
    }

    /// A companion is guarded on the same terms as the primary, and BEFORE the
    /// swap. It used to be checked by nothing at all, so the first thing that
    /// noticed a bad companion was its unit failing to start during the install
    /// loop, which runs after the server has already been restarted.
    #[tokio::test]
    async fn companions_are_guarded_before_the_swap() {
        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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .await
        .unwrap();

        let log = exec.log();
        // The companion's own arch and loader checks, named by its path so they
        // cannot be confused with the primary's.
        let guard = pos(&log, "companions/mnw-cli");
        let swap = pos(&log, "reload-or-restart");
        let install = pos(&log, "install-companion.sh");
        assert!(
            guard < swap && swap < install,
            "a companion must be guarded before the swap and installed after it: {log:#?}"
        );
        let companion_guards = log
            .iter()
            .filter(|c| c.contains("companions/mnw-cli") && !c.contains("install-companion.sh"))
            .count();
        assert_eq!(
            companion_guards, 2,
            "both guards must run against the companion, not just one: {log:#?}"
        );
    }

    /// And failing one of them fails the promote with the service intact, which
    /// is the whole point of moving the check ahead of the swap.
    #[tokio::test]
    async fn a_companion_failing_its_guard_aborts_before_the_swap() {
        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 mut exec = FakeExec::new();
        // Fails the first script naming the companion, which is its arch guard.
        // The primary's guards name the primary and are unaffected.
        exec.fail_run_matching = Some("companions/mnw-cli".into());
        let err = deploy_node(
            &exec,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .await
        .expect_err("a bad companion must fail the deploy");

        let msg = format!("{err:#}");
        assert!(
            msg.contains("mnw-cli"),
            "the refusal must name which companion: {msg}"
        );
        assert_eq!(
            stage_of(&err),
            Some(FailureStage::BeforeSwap),
            "a companion guard failing must leave the service intact: {msg}"
        );
        let log = exec.log();
        assert!(
            !log.iter().any(|c| c.contains("reload-or-restart")),
            "swap must not run after a failed companion guard: {log:#?}"
        );
        assert!(
            !log.iter().any(|c| c.contains("install-companion.sh")),
            "nothing should be installed after a failed companion guard: {log:#?}"
        );
    }

    /// The guards and the installer must read the same path. A guard checking a
    /// path the installer does not use is a check of nothing, and passes.
    #[test]
    fn the_guarded_companion_path_is_the_one_installed() {
        let release_dir = "/opt/mnw/releases/0.9.0";
        let src = companion_src(release_dir, "mnw-cli");
        assert_eq!(src, "/opt/mnw/releases/0.9.0/companions/mnw-cli");
        let cmd = install_companion_cmd(&src, "/opt/mnw-cli/mnw-cli", "mnw-cli.service");
        assert!(
            cmd.contains(&src),
            "the installer must read the path the guards checked: {cmd}"
        );
    }

    #[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,
            Placement::check(&node, &staged, None).unwrap(),
            "0.9.0",
            "makenotwork",
            Some(&no_pins()),
        )
        .await
        .unwrap();
        let log = exec.log();
        assert!(
            pos(&log, "reload-or-restart") < pos(&log, "install-companion.sh"),
            "companion install must follow the swap: {log:#?}"
        );
    }
}