gunnar_server/

pairing.rs

//! **The pairing loop PID 1 runs itself** — `nordisk-pairing`, on this box.
//!
//! Until 2026-09-24 the laptop's xtask was the matchmaker of the estate: it
//! bootstrapped the primary, enrolled the principals, conferred the twin's
//! role and signed its certificate, one ssh command at a time, and every one
//! was a place the chain broke. The owner's order that day: the xtask does
//! not pair. It lays the machines; the boxes find the installer (the join
//! door on the front) and pair each other inside a tunnel. The rules live in
//! edda's `nordisk-pairing` (design: `edda/.nornir/pairing-design.md`); this
//! module is the box's side: the doors it opens and the loop it runs.
//!
//! # The seed is the MEDIUM's, not this box's
//!
//! One seed per medium (owner 2026-09-24), landed by the installer's data
//! stage at `<data>/pairing/seed.toml`: the estate, the tunnel kind, the join
//! door's fixed address on the utility network, the installer CA this box
//! pins, and the bootstrap client certificate it joins with — valid during
//! bootstrap ONLY. Nothing in it names this box, and nothing in it changes
//! from one lay of the estate to the next (the medium is CONSTANT, owner
//! 2026-09-24 evening): the front's tunnel key and endpoint are the door's to
//! FEED, as the `front` peer of the join reply (edda nordisk-pairing 0.2).
//!
//! # The loop, in order
//!
//! 1. **Join** over the utility network, mutual TLS: the bootstrap
//!    certificate the first time, with a signing request for a key this box
//!    minted; the door answers with this box's NAME, its tunnel ADDRESS, its
//!    own certificate, and the estate values it FEEDS. From then on the box
//!    speaks with its own certificate (kept under `<data>/pairing/`, so a
//!    reboot resumes as itself). While the other box has not joined, the
//!    election has not spoken and the box asks again.
//! 2. **Tunnel**: `wg0.conf` written with the assigned address, the front as
//!    the one dialled peer owning the overlay; raised at once
//!    ([`crate::wg::raise_from_config`]).
//! 3. **Pair** inside it. The PRIMARY installs the fed admin key as its first
//!    identity ([`gunnar_core::auth::AuthTable::bootstrap_admin`] — the window
//!    closes). The TWIN must still be FRESH when it becomes the twin, so it
//!    gets no admin: it runs BecomeTwin against the primary's overlay address
//!    and starts its follower; the certificate it follows with is the front's
//!    move, and the twin sees it took when its follower applies.
//!
//! The signal is written to `<data>/pairing/signal.toml` on every change and
//! said on the log — the same reading `cargo xtask estate pairing` prints,
//! reported by the box itself.

use crate::stores::ServerStores;
use nordisk_pairing::elect::Capacity;
use nordisk_pairing::join::{JoinReply, JoinRequest};
use nordisk_pairing::pair::{Event, Machine, Step};
use nordisk_pairing::seed::{Kind, Seed, TunnelKind};
use nordisk_pairing::signal::Signal;
use nordisk_pairing::tunnel::{Identity, Tunnel, TunnelError, TunnelStatus};
use nordisk_pairing::CONTROL_PORT;
use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

/// Under the data root.
pub const SEED_REL: &str = "pairing/seed.toml";
pub const SIGNAL_REL: &str = "pairing/signal.toml";
pub const JOINED_REL: &str = "pairing/joined.toml";
/// The first line of a `wg0.conf` this loop wrote — how the boot knows the
/// pairing loop, not an operator, owns the tunnel ([`crate::wg::boot_lane`]).
pub const CONF_MARK: &str = "# written by gunnar's pairing loop";
pub const OWN_CERT_REL: &str = "pairing/box.crt";
pub const OWN_KEY_REL: &str = "pairing/box.key";
/// The tunnel interface and the overlay prefix the estate uses.
pub const IFACE: &str = "wg0";
pub const OVERLAY_PREFIX: u8 = 24;
/// The fed value the primary installs as its first admin.
pub const FEED_ADMIN: &str = "admin_pubkey";
/// **The fed value that makes this box a RECHARGED main** (Skidbladnir
/// nordisk-pairing `Door::readmit`, `RECHARGE_FROM_FEED`): the twin's control
/// plane, `<utility>:50051`. The dead main of a closed estate was laid again
/// from the same medium and re-admitted as the primary it was; it must be
/// FILLED from its twin before it takes any identity of its own.
pub const FEED_RECHARGE_FROM: &str = nordisk_pairing::door::RECHARGE_FROM_FEED;
/// `<data>/pairing/recharge_from` — [`crate::twin::RECHARGE_REL`], which lives beside
/// `is_restoring` because a build without `wg-client` (no pairing loop) reads it too.
pub use crate::twin::RECHARGE_REL;

/// **Where a recharged main stands** ([`Doors::recharge_hold`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Hold {
    /// FRESH, fed `recharge_from`: the fed admin is HELD BACK (installing it
    /// would shut the bootstrap window and `Bootstrap.Restore` would refuse).
    Awaiting,
    /// A restore began and is not whole; the box's log head.
    Restoring { head: u64 },
    /// The restore finished in THIS process; the box restarts to open as the
    /// primary it was (the replayed secrets are sealed under the dead main's KEK).
    Restarting,
    /// Restored and restarted (or never fresh): the ordinary admin step.
    Restored,
}

/// The seed, if this box was installed with one. Absent is `None`; a seed
/// that is present and unusable IS an error — a declaration that lies must
/// not be quietly ignored.
///
/// **A release image carries no seed** (no operator machine in a sale, owner
/// 2026-10-08): its staged seed file holds only [`SEED_FROM_METADATA`], and the
/// box takes its seed — minted per order by the shop — from UpCloud's
/// metadata (`user_data`) on its first boot, and keeps it here (0600) as a
/// medium's own would be. The bootstrap key in it joins one slot once, as
/// any medium's (the installer's door marks the slot spent).
pub fn seed(root: &Path) -> anyhow::Result<Option<Seed>> {
    seed_with(root, &metadata_user_data, METADATA_TRIES, METADATA_EVERY)
}

/// The staged seed file of a release image: the seed is the metadata's.
pub const SEED_FROM_METADATA: &str = "nordisk-pairing-seed-from = \"upcloud-metadata\"";
/// UpCloud's metadata service, as every server with `metadata = yes` reaches it.
pub const METADATA_AT: &str = "169.254.169.254:80";
/// The server's `user_data` there.
pub const METADATA_USER_DATA: &str = "/metadata/v1/user_data";
/// The first boot asks this many times, this far apart (the network may come
/// up a little after the process).
pub const METADATA_TRIES: u32 = 36;
pub const METADATA_EVERY: Duration = Duration::from_secs(5);

/// [`seed`], asking `fetch` for the metadata's `user_data` (tests: a fake).
pub fn seed_with(root: &Path, fetch: &dyn Fn() -> Result<String, String>, tries: u32, every: Duration) -> anyhow::Result<Option<Seed>> {
    let p = root.join(SEED_REL);
    let text = match std::fs::read_to_string(&p) {
        Ok(t) => t,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => anyhow::bail!("{}: {e}", p.display()),
    };
    if is_metadata_marker(&text) {
        let mut last = String::new();
        for n in 0..tries.max(1) {
            match fetch().and_then(|body| seed_of_user_data(&body)) {
                Ok(seed) => {
                    let rendered = seed.render().map_err(|e| anyhow::anyhow!("the metadata's seed: {e}"))?;
                    write_secret(root, SEED_REL, &rendered)?;
                    tracing::info!(estate = %seed.estate, "pairing: this box's seed came from the metadata (a release image carries none); kept at {}", p.display());
                    return Ok(Some(seed));
                }
                Err(why) => last = why,
            }
            if n + 1 < tries {
                std::thread::sleep(every);
            }
        }
        anyhow::bail!("{}: a release image's seed is the metadata's, and the metadata gave none after {tries} tries: {last}", p.display());
    }
    Ok(Some(Seed::parse(&text).map_err(|e| anyhow::anyhow!("{}: {e}", p.display()))?))
}

/// Pure: whether a staged seed file is a release image's marker (its one line).
pub fn is_metadata_marker(text: &str) -> bool {
    let lines: Vec<&str> = text.lines().map(str::trim).filter(|l| !l.is_empty() && !l.starts_with('#')).collect();
    lines == [SEED_FROM_METADATA]
}

/// Pure: **the seed in a server's `user_data`** — the seed TOML itself, nothing
/// else (an empty `user_data` is a server laid without one).
pub fn seed_of_user_data(body: &str) -> Result<Seed, String> {
    if body.trim().is_empty() {
        return Err("the server's user_data is empty: it was laid without a seed".into());
    }
    Seed::parse(body).map_err(|e| format!("the server's user_data is not a pairing seed: {e}"))
}

/// `GET http://169.254.169.254/metadata/v1/user_data` — the body of a 200.
pub fn metadata_user_data() -> Result<String, String> {
    let at: std::net::SocketAddr = METADATA_AT.parse().map_err(|e| format!("{METADATA_AT}: {e}"))?;
    match http_get(at, "169.254.169.254", METADATA_USER_DATA)? {
        (200, body) => Ok(body),
        (st, body) => Err(format!("the metadata answered HTTP {st}: {}", body.chars().take(200).collect::<String>())),
    }
}

/// One plain HTTP/1.0 GET (the metadata service speaks no TLS): the status and the body.
pub fn http_get(at: std::net::SocketAddr, host: &str, path: &str) -> Result<(u16, String), String> {
    use std::io::{Read as _, Write as _};
    let mut tcp = std::net::TcpStream::connect_timeout(&at, Duration::from_secs(3)).map_err(|e| format!("{at}: {e}"))?;
    tcp.set_read_timeout(Some(Duration::from_secs(10))).map_err(|e| e.to_string())?;
    tcp.write_all(format!("GET {path} HTTP/1.0\r\nHost: {host}\r\nAccept: text/plain\r\nConnection: close\r\n\r\n").as_bytes()).map_err(|e| format!("{at}: {e}"))?;
    let mut raw = Vec::new();
    tcp.read_to_end(&mut raw).map_err(|e| format!("{at}: {e}"))?;
    let text = String::from_utf8_lossy(&raw);
    let (head, body) = text.split_once("\r\n\r\n").ok_or_else(|| format!("{at} answered no HTTP"))?;
    let status = head.split_whitespace().nth(1).and_then(|s| s.parse().ok()).ok_or_else(|| format!("{at} answered no status: {}", head.lines().next().unwrap_or("")))?;
    Ok((status, body.to_string()))
}

/// **`wg0.conf` for the address the door assigned** — the shape `crate::wg`
/// already parses: this box's private key, its overlay address, the front as
/// the one dialled peer owning the whole overlay (the other box is reached
/// THROUGH the front's relay), a keepalive that keeps NATs open.
///
/// The front's key and endpoint are the DOOR's answer (the reply's `front`
/// peer), never the medium's: the relay is minted with each lay's front.
pub fn wg_conf(front: &nordisk_pairing::join::Peer, overlay: Ipv4Addr, private_b64: &str, psk: Option<&str>) -> Result<String, String> {
    // The PSK the installer fed (2026-10-06, the PQ transport lane): WireGuard's
    // handshake is X25519 alone, and a pre-shared key both ends hold — fed over
    // the installer's post-quantum TLS — is what a recording cannot be opened
    // without. Checked as a WireGuard key here, so a bad feed is named, never
    // written into a conf the tunnel lane would refuse later.
    let psk_line = match psk.map(str::trim) {
        Some(p) => {
            wg_appliance_core::decode_key(p).map_err(|_| "the installer fed a WireGuard pre-shared key that is not one (32 bytes, base64)".to_string())?;
            format!("PresharedKey = {p}\n")
        }
        None => String::new(),
    };
    let (net, bits) = nordisk_pairing::OVERLAY_NET;
    let endpoint = front.endpoint.as_deref().filter(|e| !e.trim().is_empty()).ok_or_else(|| format!("the door fed the front `{}` with no endpoint to dial", front.name))?;
    if front.tunnel_pubkey.trim().is_empty() {
        return Err(format!("the door fed the front `{}` with no tunnel key", front.name));
    }
    Ok(format!(
        "{CONF_MARK}: the address is the installer door's answer, the key was minted on this box\n\
         [Interface]\n\
         PrivateKey = {private_b64}\n\
         Address = {overlay}/{OVERLAY_PREFIX}\n\
         \n\
         # the front — the one box everybody dials (fed by the door)\n\
         [Peer]\n\
         PublicKey = {}\n\
         {psk_line}\
         AllowedIPs = {net}/{bits}\n\
         Endpoint = {endpoint}\n\
         PersistentKeepalive = 25\n",
        front.tunnel_pubkey.trim()
    ))
}

/// **The pre-shared key of this box's link to the front**, as the installer
/// fed it ([`nordisk_pairing::installer::FEED_WG_PSK`]) and the box keeps it:
/// the sealed `joined.toml`. `None` when nothing was fed (an estate laid
/// before PSKs, a test).
pub fn fed_psk(root: &Path) -> Option<String> {
    let text = crate::fed_at_rest::read(root, JOINED_REL)?;
    let reply = JoinReply::parse(&text).ok()?;
    reply.feed.get(nordisk_pairing::installer::FEED_WG_PSK).map(|p| p.trim().to_string()).filter(|p| !p.is_empty())
}

/// **Write `wg0.conf` when there is none, or when the one there is THIS
/// loop's and says something else** (the door fed another front: a lay whose
/// relay key changed under a box that kept its disk). A configuration an
/// operator wrote — no [`CONF_MARK`] — is never replaced. Pure over the text
/// on disk; answers whether to write.
pub fn conf_needs_writing(on_disk: Option<&str>, wanted: &str) -> bool {
    match on_disk {
        None => true,
        Some(text) if text.starts_with(CONF_MARK) => text != wanted,
        Some(_) => false,
    }
}

/// **The doors this loop opens**, as a trait so the loop is testable without
/// a network, a data root or a control plane.
pub trait Doors {
    /// The installer's door: join, as whoever this box is now.
    fn join(&self, req: &JoinRequest) -> Result<JoinReply, String>;
    /// Raise this box's tunnel on the assigned address, through the seed's
    /// [`Tunnel`] driver ([`tunnel_driver`]). `Ok(true)` when it is up.
    fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[nordisk_pairing::join::Peer]) -> Result<bool, String>;
    /// The primary's admin: install the fed key if the table is empty, then
    /// say whether an admin is there.
    fn admin(&self, fed: Option<&str>) -> Result<bool, String>;
    fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
    /// `(applied, head)` from the follower, or why not.
    fn follow_state(&self) -> Result<(u64, u64), String>;
    /// **A primary fed [`FEED_RECHARGE_FROM`]**: where its recharge stands.
    fn recharge_hold(&self, _from: &str) -> Result<Hold, String> {
        Ok(Hold::Restored)
    }
}

/// **The primary's utility address, when the door names the twin's primary
/// by its overlay** — an estate upgraded onto WireGuard-only keeps a twin that
/// was joined on that utility address ([`crate::twin::onto_tunnel`]). Only for
/// the SAME box: the reply's `primary_overlay` must be the `primary` the twin
/// is told, and its `primary_utility` another address.
pub fn same_primary_utility(reply: Option<&JoinReply>, primary: Ipv4Addr) -> Option<Ipv4Addr> {
    let reply = reply?;
    if reply.primary_overlay != Some(primary) {
        return None;
    }
    reply.primary_utility.filter(|u| *u != primary)
}

/// **Is this box FRESH** — the state `twin::prepare_replay` takes a restore
/// into: no identity, the bootstrap window open, no repository.
pub fn is_fresh(stores: &ServerStores) -> anyhow::Result<bool> {
    Ok(stores.auth().is_empty()? && stores.auth().bootstrap_window_open()? && stores.catalog().list(None)?.is_empty())
}

/// **This main's twin, as the door named it** (ASK-twin, resilience-gunnar 21:10): the
/// twin peer of the reply's peer list, at its control plane (`<it>:50051`) — its utility
/// address, or its overlay address on a WireGuard-only estate, the way a twin finds its
/// primary ([`crate::tunnel_addr::primary_control`]). `None` on a box that is not the main,
/// and when the door named no twin.
pub fn twin_of(reply: &JoinReply, wg_only: bool) -> Option<String> {
    if reply.role != Some(nordisk_pairing::Role::Primary) {
        return None;
    }
    let twin = reply.peers.iter().find(|p| p.role == nordisk_pairing::Role::Twin && p.name != reply.me.name)?;
    let at = crate::tunnel_addr::primary_control(twin.utility, Some(twin.overlay), wg_only)?;
    Some(format!("{at}:{CONTROL_PORT}"))
}

/// **The role the door elected for this box**, from `<data>/pairing/joined.toml`:
/// `Some(None)` while it has not elected one, `None` when no door ever answered.
pub fn door_role(root: &Path) -> Option<Option<String>> {
    let text = crate::fed_at_rest::read(root, JOINED_REL)?;
    Some(JoinReply::parse(&text).ok()?.role.map(|r| r.as_str().to_owned()))
}

/// [`twin_of`] the answer kept in `<data>/pairing/joined.toml`.
pub fn door_twin(root: &Path) -> Option<String> {
    let text = crate::fed_at_rest::read(root, JOINED_REL)?;
    twin_of(&JoinReply::parse(&text).ok()?, crate::tunnel_addr::wg_only())
}

/// **What `Bootstrap.Status.recharge_from` answers**: the twin this box
/// awaits, while it is a recharged main that has not been restored (fresh, or
/// a restore under way). `None` on every other box, and once it is restored.
pub fn recharge_awaited(stores: &ServerStores) -> Option<String> {
    let from = std::fs::read_to_string(stores.root().join(RECHARGE_REL)).ok()?.trim().to_string();
    if from.is_empty() {
        return None;
    }
    let root = stores.root();
    let open = crate::twin::is_restoring(root) || is_fresh(stores).unwrap_or(false);
    (open && !crate::twin::reseal_pending(root)).then_some(from)
}

/// Where a recharged main stands, off its own disk ([`Hold`]).
pub fn recharge_hold_of(stores: &ServerStores, from: &str) -> anyhow::Result<Hold> {
    let root = stores.root();
    if crate::twin::reseal_pending(root) {
        return Ok(Hold::Restarting);
    }
    if crate::twin::is_restoring(root) {
        return Ok(Hold::Restoring { head: stores.txlog().head_seq() });
    }
    if is_fresh(stores)? {
        if std::fs::read_to_string(root.join(RECHARGE_REL)).ok().map(|t| t.trim().to_string()).as_deref() != Some(from) {
            if let Some(dir) = root.join(RECHARGE_REL).parent() {
                std::fs::create_dir_all(dir)?;
            }
            crate::durable::write_atomic_unframed(root, RECHARGE_REL, format!("{from}\n").as_bytes())?;
            tracing::warn!(recharge_from = %from, "pairing: RECHARGE — this box is a re-admitted main and fresh; it awaits its twin, the admin is held back");
        }
        return Ok(Hold::Awaiting);
    }
    match std::fs::remove_file(root.join(RECHARGE_REL)) {
        Ok(()) => tracing::warn!(recharge_from = %from, "pairing: RECHARGE done — this box holds the identities its twin gave back; the ordinary admin step follows"),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e.into()),
    }
    Ok(Hold::Restored)
}

/// The real doors.
pub struct BoxDoors {
    stores: Arc<ServerStores>,
    seed: Seed,
    /// **What the installer fed** (since 2026-09-24 the installer connects to
    /// THIS box's bootstrap listener, `crate::installer_listener`): the loop's
    /// join is a read of it.
    installed: Arc<crate::installer_listener::Installed>,
    /// The seed's tunnel driver, or the refusal that this image carries none.
    /// WireGuard's is built at once (its key goes out in the FIRST
    /// presentation); a driver that needs the feed (mullvad, tailscale) is
    /// built from it at the tunnel step.
    tunnel: std::sync::Mutex<Option<Result<Box<dyn Tunnel>, TunnelError>>>,
    /// What the installer fed at the last read (a driver's credentials ride it).
    feed: std::sync::Mutex<std::collections::BTreeMap<String, String>>,
}

impl BoxDoors {
    fn new(stores: Arc<ServerStores>, seed: Seed, installed: Arc<crate::installer_listener::Installed>) -> BoxDoors {
        let root = stores.root().to_path_buf();
        let tunnel = (seed.tunnel == TunnelKind::Wireguard).then(|| tunnel_driver(&seed, &root, &Default::default()));
        BoxDoors { stores, seed, installed, tunnel: std::sync::Mutex::new(tunnel), feed: std::sync::Mutex::new(Default::default()) }
    }

    /// The key the presentation carries: the driver's [`Tunnel::public_key`]
    /// (known before `up`), empty while no driver is built or the image has none.
    fn tunnel_key(&self) -> String {
        match self.tunnel.lock().ok().as_deref() {
            Some(Some(Ok(t))) => t.public_key(),
            _ => String::new(),
        }
    }
}

pub fn write_secret(root: &Path, rel: &str, text: &str) -> anyhow::Result<()> {
    if let Some(parent) = root.join(rel).parent() {
        std::fs::create_dir_all(parent)?;
    }
    crate::durable::write_atomic_unframed(root, rel, text.as_bytes())?;
    use std::os::unix::fs::PermissionsExt as _;
    std::fs::set_permissions(root.join(rel), std::fs::Permissions::from_mode(0o600))?;
    Ok(())
}

impl Doors for BoxDoors {
    /// **The join is what the installer fed** — nothing is dialled. Until the
    /// installer has reached this box's listener there is nothing to read,
    /// which is a transient: the loop asks again.
    fn join(&self, _req: &JoinRequest) -> Result<JoinReply, String> {
        let reply = self.installed.latest().ok_or_else(|| {
            "waiting for the installer: it connects to this VM's bootstrap listener and feeds it (name, tunnel address, role, peers)".to_string()
        })?;
        if let Ok(mut feed) = self.feed.lock() {
            feed.clone_from(&reply.feed);
        }
        Ok(reply)
    }

    fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[nordisk_pairing::join::Peer]) -> Result<bool, String> {
        let mut slot = self.tunnel.lock().map_err(|e| e.to_string())?;
        if slot.is_none() {
            let feed = self.feed.lock().map_err(|e| e.to_string())?.clone();
            *slot = Some(tunnel_driver(&self.seed, self.stores.root(), &feed));
        }
        let driver = slot.as_mut().expect("built above").as_mut().map_err(|e| e.to_string())?;
        let me = Identity { overlay, pubkey: driver.public_key() };
        driver.up(&me, peers).map_err(|e| e.to_string())?;
        Ok(driver.status().up)
    }

    fn admin(&self, fed: Option<&str>) -> Result<bool, String> {
        let auth = self.stores.auth();
        if let Some(key) = fed {
            auth.bootstrap_admin(key).map_err(|e| format!("{e:#}"))?;
        }
        let present = auth.is_empty().map(|empty| !empty).map_err(|e| format!("{e:#}"))?;
        if present {
            // The fed principals, right after the admin (the first identity
            // closed the window; these are ordinary administration now) —
            // from the LATEST feed, not the copy the join step kept: MEASURED
            // 2026-09-25 (lane B's bench), vm1's offers (certify_pubkey …)
            // reached the primary AFTER its join and BEFORE its admin; the
            // late-principal path skipped them (no admin yet) and this step
            // read the stale copy, so `front-certify` was never enrolled and
            // the twin could never be certified. The copy is refreshed too.
            let feed = match self.installed.latest() {
                Some(reply) => {
                    if let Ok(mut kept) = self.feed.lock() {
                        kept.clone_from(&reply.feed);
                    }
                    reply.feed
                }
                None => self.feed.lock().map_err(|e| e.to_string())?.clone(),
            };
            for line in crate::fed_principals::install(&self.stores, &feed)? {
                if !line.ends_with("already there") {
                    tracing::info!("pairing: fed principal {line}");
                }
            }
        }
        Ok(present)
    }

    fn recharge_hold(&self, from: &str) -> Result<Hold, String> {
        let hold = recharge_hold_of(&self.stores, from).map_err(|e| format!("{e:#}"))?;
        // ★ main rebuild (2026-10-12): the restore lane is retired; with auto-fix on, an
        // awaiting main fills itself from the twin the door named (`repair::on_awaiting`).
        #[cfg(feature = "replication")]
        if hold == Hold::Awaiting {
            if let Some(words) = crate::repair::on_awaiting(&self.stores, from) {
                tracing::info!("pairing: {words}");
            }
        }
        Ok(hold)
    }

    fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
        let at = format!("{primary}:{CONTROL_PORT}");
        if let Some(utility) = same_primary_utility(self.installed.latest().as_ref(), primary) {
            if crate::twin::onto_tunnel(&self.stores, &format!("{utility}:{CONTROL_PORT}"), &at).map_err(|e| format!("{e:#}"))? {
                tracing::info!(primary = %at, was = %format!("{utility}:{CONTROL_PORT}"), "pairing: this twin follows the same primary over the tunnel now (joined on its utility address before WireGuard-only)");
            }
        }
        crate::twin::become_twin(&self.stores, &at).map(|_| ()).map_err(|e| format!("{e:#}"))?;
        crate::twin::start_follower(Arc::clone(&self.stores));
        Ok(())
    }

    fn follow_state(&self) -> Result<(u64, u64), String> {
        match crate::twin::status(self.stores.root()) {
            Some(s) => Ok((s.applied_seq, s.primary_head_seq)),
            None => Err("the follower has not reported yet".into()),
        }
    }
}

/// **gunnar's WireGuard, behind the estate's [`Tunnel`] trait** — over
/// `crate::wg`'s config lane (userspace boringtun in this process, the lane
/// proven on the appliance). `up` writes `wg0.conf` for the assigned address
/// with the key [`ensure_tunnel_key`] minted (the key the first join carried)
/// and the FRONT as the one dialled peer owning the overlay, then raises it.
/// Every other box is reached THROUGH the front's relay, so `add_peer` has
/// nothing to add here.
pub struct GunnarWireGuard {
    root: PathBuf,
    /// Where the front was dialled, as the door fed it at the last `up`.
    front_endpoint: Option<String>,
}

impl GunnarWireGuard {
    pub fn new(root: &Path, _seed: &Seed) -> GunnarWireGuard {
        GunnarWireGuard { root: root.to_path_buf(), front_endpoint: None }
    }
}

impl Tunnel for GunnarWireGuard {
    fn kind(&self) -> TunnelKind {
        TunnelKind::Wireguard
    }

    /// `wg0.key`'s public half — minted (0600) here if absent, so the FIRST
    /// join carries it.
    fn public_key(&self) -> String {
        tunnel_pubkey(&self.root)
    }

    /// The front is the peer the door FED with role `front` — its key and
    /// endpoint belong to the lay, not the medium. A reply without one is a
    /// transient failure named as such: the loop asks the door again.
    fn up(&mut self, me: &Identity, peers: &[nordisk_pairing::join::Peer]) -> Result<(), TunnelError> {
        let failed = |e: anyhow::Error| TunnelError::Failed(format!("{e:#}"));
        let root = self.root.as_path();
        let front = peers.iter().find(|p| p.role == nordisk_pairing::Role::Front).ok_or_else(|| {
            TunnelError::Failed(format!(
                "{IFACE}: the door fed no front tunnel (its ledger names no [front]); nothing to dial"
            ))
        })?;
        let private_b64 = ensure_tunnel_key(root).map_err(failed)?;
        let psk = fed_psk(root);
        let wanted = wg_conf(front, me.overlay, &private_b64, psk.as_deref()).map_err(TunnelError::Failed)?;
        let on_disk = std::fs::read_to_string(root.join(crate::wg::CONFIG_FILE)).ok();
        if conf_needs_writing(on_disk.as_deref(), &wanted) {
            // 0600: the conf holds this box's private key and the link's PSK.
            write_secret(root, crate::wg::CONFIG_FILE, &wanted).map_err(failed)?;
            if psk.is_some() {
                tracing::info!(front = %front.name, "pairing: wg0 dials the front's relay with the pre-shared key the installer fed (post-quantum hardening of the X25519 handshake)");
            }
        }
        self.front_endpoint = front.endpoint.clone();
        match crate::wg::raise_from_config(root).map_err(failed)? {
            true => Ok(()),
            false => Err(TunnelError::Failed(format!("{IFACE}: no configuration to raise"))),
        }
    }

    fn add_peer(&mut self, _peer: &nordisk_pairing::join::Peer) -> Result<(), TunnelError> {
        Ok(())
    }

    fn status(&self) -> TunnelStatus {
        let shown = crate::wg::pairing_tunnel(&self.root);
        let up = shown.as_ref().map(|(_, _, up)| *up).unwrap_or(false) || crate::wg::config_tunnel_handshaken().is_some();
        let handshaken = crate::wg::config_tunnel_handshaken().unwrap_or(false);
        TunnelStatus {
            kind: TunnelKind::Wireguard,
            up,
            reachable: if handshaken { vec!["front".into()] } else { vec![] },
            detail: match shown.and_then(|(_, a, _)| a) {
                Some((a, p)) => format!(
                    "{IFACE} {a}/{p} (userspace WireGuard, the front's relay at {})",
                    self.front_endpoint.as_deref().unwrap_or("the endpoint the door fed")
                ),
                None => format!("{IFACE}: no address assigned yet"),
            },
        }
    }
}

/// **The seed's tunnel, as the driver this image carries** — or the named
/// refusal `tunnel-not-built`. WireGuard is gunnar's own driver above; every
/// other kind is `nordisk_pairing::tunnel::driver_for` with the door's `feed`
/// (the credentials a mullvad or tailscale driver needs ride it): mullvad and
/// tailscale are whatever nordisk-pairing was compiled with here — this
/// image enables neither feature, so both are REFUSED [tunnel-not-built].
/// `none` is REFUSED [installer-link-none] (edda nordisk-pairing 0.4.0, owner
/// 2026-09-25: WireGuard + mutual TLS ALWAYS) — there is no estate without
/// its tunnel.
pub fn tunnel_driver(seed: &Seed, root: &Path, feed: &std::collections::BTreeMap<String, String>) -> Result<Box<dyn Tunnel>, TunnelError> {
    match seed.tunnel {
        TunnelKind::Wireguard => Ok(Box::new(GunnarWireGuard::new(root, seed))),
        TunnelKind::None => Err(TunnelError::InstallerLinkNone),
        _ => nordisk_pairing::tunnel::driver_for(seed, feed),
    }
}

/// **This box's tunnel private key (base64), minted here — 0600 — if absent.**
/// The ONE minting of `wg0.key`: the first join carries its public half (the
/// door records it and the relay lists this box by it), and the tunnel lane
/// writes the same key into `wg0.conf`. Before 2026-09-24 the key was minted
/// only when the tunnel came up — AFTER the first join — so the door recorded
/// an empty key and the relay listed nobody.
pub fn ensure_tunnel_key(root: &Path) -> anyhow::Result<String> {
    if let Ok(k) = std::fs::read_to_string(root.join(crate::wg::KEY_FILE)) {
        let k = k.trim();
        if !k.is_empty() {
            return Ok(k.to_string());
        }
    }
    let (private, _) = crate::wg::generate_keypair();
    let b64 = wg_appliance_core::encode_key(&private);
    write_secret(root, crate::wg::KEY_FILE, &format!("{b64}\n"))?;
    Ok(b64)
}

/// This box's tunnel public key, off the private key on disk — minted first
/// if absent ([`ensure_tunnel_key`]), so the FIRST join already carries it.
pub fn tunnel_pubkey(root: &Path) -> String {
    match ensure_tunnel_key(root) {
        Ok(k) => wg_appliance_core::public_from_private(&k).unwrap_or_default(),
        Err(e) => {
            tracing::error!(error = %format!("{e:#}"), "pairing: this box cannot mint its tunnel key; the join carries none");
            String::new()
        }
    }
}

/// What this box measured about itself — the election's input.
pub fn capacity(root: &Path) -> Option<Capacity> {
    let m = crate::capacity::MachineFacts::measure();
    let cores = m.cpus.ok()? as u32;
    let ram_bytes = m.memory.ok()?.0;
    let data_bytes = rustix::fs::statvfs(root).map(|s| s.f_blocks * s.f_frsize).unwrap_or(0);
    let disk_mib_s = match crate::disktest::last() {
        Some(crate::disktest::Outcome::Done(r)) => (r.seq_write_bps / (1024.0 * 1024.0)) as u32,
        _ => 0,
    };
    Some(Capacity { cores, ram_bytes, data_bytes, disk_mib_s })
}

/// The join request, now: a signing request only while this box has no
/// certificate of its own, and the tunnel key the driver presents
/// ([`Tunnel::public_key`] — WireGuard's is minted before the FIRST join).
fn request(seed: &Seed, root: &Path, csr: &str, tunnel_pubkey: String) -> JoinRequest {
    JoinRequest {
        estate: seed.estate.clone(),
        kind: Kind::Gunnar,
        tunnel_pubkey,
        capacity: capacity(root),
        csr_pem: csr.to_string(),
    }
}

/// The loop's memory between turns: what the door said.
#[derive(Debug, Default, Clone)]
pub struct Learned {
    pub overlay: Option<Ipv4Addr>,
    /// The peers the door named (handed to the tunnel driver).
    pub peers: Vec<nordisk_pairing::join::Peer>,
    pub feed_admin: Option<String>,
    pub name: Option<String>,
    /// The feed the door answered with (`crate::feed`; published on the shared bus by
    /// [`take_learned_feed`]).
    pub feed: Option<nordisk_pairing::feedbus::Feed>,
    /// The twin this box is recharged from, when the door fed one.
    pub recharge_from: Option<String>,
}

/// **One turn of the loop**: do what the machine asks, tell it what happened.
/// Pure over the doors, so the fake doors below prove it.
pub fn turn(m: &mut Machine, learned: &mut Learned, seed: &Seed, req: &JoinRequest, doors: &dyn Doors) -> Step {
    let step = m.next();
    let event = match &step {
        Step::Join | Step::FrontAwaitPeers => match doors.join(req) {
            Ok(reply) if reply.estate != seed.estate => Event::JoinRefusedForGood(format!("the door is `{}`, this medium is `{}`", reply.estate, seed.estate)),
            Ok(reply) => {
                learned.overlay = Some(reply.me.overlay);
                learned.peers = reply.peers.clone();
                learned.name = Some(reply.me.name.clone());
                learned.feed_admin = reply.feed.get(FEED_ADMIN).cloned();
                learned.feed = Some(reply.feed.clone());
                // Fed as the twin's utility address (the installer is shared
                // with holger); a WireGuard-only box dials its overlay address.
                let pairs: Vec<(std::net::Ipv4Addr, std::net::Ipv4Addr)> = reply.peers.iter().filter_map(|p| p.utility.map(|u| (u, p.overlay))).collect();
                learned.recharge_from = reply
                    .feed
                    .get(FEED_RECHARGE_FROM)
                    .map(|t| t.trim().to_string())
                    .filter(|t| !t.is_empty())
                    .map(|t| if crate::tunnel_addr::wg_only() { crate::tunnel_addr::over_the_tunnel(&t, &pairs) } else { t });
                match reply.role {
                    None => Event::Transient(format!("admitted as `{}`; the election has not spoken yet", reply.me.name)),
                    Some(r) => Event::Joined { role: r, primary: crate::tunnel_addr::primary_control(reply.primary_utility, reply.primary_overlay, crate::tunnel_addr::wg_only()), peers_admitted: reply.peers.len() as u8, peers_expected: seed.peers_expected },
                }
            }
            Err(e) if ["[join-spent]", "[join-forged]", "[join-unknown]", "[join-estate]", "[join-withdrawn]"].iter().any(|w| e.contains(w)) => Event::JoinRefusedForGood(e),
            Err(e) => Event::Transient(e),
        },
        Step::BringTunnelUp => match learned.overlay {
            // A WireGuard tunnel dials the front the door FED; an answer that
            // carried none (a door laid before its relay) is asked again here,
            // or the step would retry the same stale answer forever.
            Some(_) if seed.tunnel == TunnelKind::Wireguard && !learned.peers.iter().any(|p| p.role == nordisk_pairing::Role::Front) => {
                match doors.join(req) {
                    Ok(reply) if reply.estate == seed.estate && reply.front().is_some() => {
                        learned.peers = reply.peers.clone();
                        Event::Transient("the door now feeds the front's tunnel; raising it next".into())
                    }
                    Ok(_) => Event::Transient("the door fed no front tunnel yet (its ledger names no [front])".into()),
                    Err(e) => Event::Transient(e),
                }
            }
            None => Event::Transient("no tunnel address yet".into()),
            Some(o) => match doors.tunnel_up(o, &learned.peers) {
                Ok(true) => {
                    // The fed clock is reached through this tunnel: ask it now, not after a pause.
                    crate::clock::tunnel_up();
                    Event::TunnelUp
                }
                Ok(false) => Event::Transient(format!("the {} tunnel is not up yet", seed.tunnel.as_str())),
                // The image carries no driver for the seed's tunnel: asking
                // again cannot build one — the medium must be composed with it.
                Err(e) if e.contains("[tunnel-not-built]") || e.contains("[installer-link-none]") => Event::JoinRefusedForGood(e),
                Err(e) => Event::Transient(e),
            },
        },
        Step::PrimaryCheckAdmin => {
            // A RECHARGED main (the installer re-admitted it with its twin
            // named) is filled from that twin FIRST: its first identity would
            // shut the bootstrap window and the restore would be refused.
            let hold = match learned.recharge_from.as_deref() {
                Some(from) => doors.recharge_hold(from),
                None => Ok(Hold::Restored),
            };
            match hold {
                Err(e) => Event::Transient(e),
                Ok(Hold::Awaiting) => Event::Transient(format!(
                    "RECHARGE: this box is the estate's re-admitted main and is FRESH — it awaits its twin at {} (auto-fix fills it when the estate opted in; else the console's [Recharge from twin] or a guardian order; the admin is held back so the bootstrap window stays open)",
                    learned.recharge_from.as_deref().unwrap_or("?")
                )),
                Ok(Hold::Restoring { head }) => Event::Transient(format!("RECHARGE: restoring from the twin at {} — the box's log is at seq {head}", learned.recharge_from.as_deref().unwrap_or("?"))),
                Ok(Hold::Restarting) => Event::Transient("RECHARGE: restored from the twin — the box restarts to open as the primary it was".into()),
                Ok(Hold::Restored) => match doors.admin(learned.feed_admin.as_deref()) {
                    Ok(true) => Event::AdminPresent,
                    Ok(false) => Event::Transient(format!("no admin on this box: the door fed no `{FEED_ADMIN}`")),
                    Err(e) => Event::Transient(e),
                },
            }
        }
        Step::TwinBecome { primary } => match doors.become_twin(*primary) {
            Ok(()) => Event::TwinConferred,
            Err(e) => Event::Transient(e),
        },
        // The certificate the twin follows with is the front's move (it holds
        // the admin key); this box learns it took when its follower applies.
        Step::TwinCertify { .. } | Step::TwinFollow { .. } => match doors.follow_state() {
            Ok((applied, _)) if matches!(step, Step::TwinCertify { .. }) && applied > 0 => Event::TwinCertified,
            Ok((applied, head)) if matches!(step, Step::TwinFollow { .. }) => Event::Following { applied, head },
            Ok((_, head)) => Event::Transient(format!("waiting for the front to certify this twin and for the first entry to apply (primary head {head})")),
            Err(e) => Event::Transient(e),
        },
        Step::Nothing => return step,
    };
    m.apply(event);
    step
}

/// What changed between the feed this process runs on and the door's latest.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct FeedChange {
    pub clock: bool,
    pub monetize: bool,
}

/// Pure: which fed values differ from what was published before.
pub fn feed_change(before: Option<&crate::feed::Fed>, now: &crate::feed::Fed) -> FeedChange {
    let empty = crate::feed::Fed::default();
    let before = before.unwrap_or(&empty);
    FeedChange {
        clock: before.clock_peer != now.clock_peer || before.clock_key != now.clock_key,
        monetize: before.monetize_pubkey != now.monetize_pubkey,
    }
}

/// Pure: **whether a feed starts the fed clock syncs** ([`crate::clock::fed_arrived`]). A NEW
/// clock always does. A REPEATED one does while this process has not set its clock right
/// (`last`: the outcome of its latest sync, `None` before any) — the in-place upgrade's
/// readmit (2026-10-09): the re-laid twin published the clock it kept at start, its boot sync
/// could not reach the front through a tunnel not yet up, and the installer's feed of the SAME
/// clock was "no change", so nothing asked the front again for six hours. The rule is edda
/// estate-clock's (`offer::feed_resyncs`); the change is Gunnar's [`FeedChange`].
pub fn feed_resyncs_the_clock(change: FeedChange, fed_clock: bool, last: Option<estate_clock::report::Outcome>) -> bool {
    estate_clock::offer::feed_resyncs(change.clock, fed_clock, last)
}

/// **Apply what the door fed, as far as a running process can.** The join
/// reply is already persisted to `joined.toml` ([`BoxDoors::join`]), so the
/// next start reads it whatever happens here. Now: publish it on the shared
/// feed bus (nordisk-pairing `feedbus`: every subscriber gets a CHANGED feed,
/// whichever key changed — the fleet door, the pay page, the IdP networks —
/// and nobody gets the same feed twice; the clock reads it on every sync, and
/// takes its first one at once — `clock::fed_arrived`), and hand the monetize
/// key to `feed::monetize_arrived`, which APPLIES it in place (owner
/// 2026-09-25: avoid restarts) or alarms why it cannot. Called by the pairing
/// loop and by every installer feed (`installer_listener`), so a key vm1
/// offers after this loop rested is still taken. Idempotent.
pub fn take_feed(feed: Option<&nordisk_pairing::feedbus::Feed>) {
    let Some(feed) = feed else { return };
    let fed = &crate::feed::Fed::from_feed(feed);
    let change = feed_change(crate::feed::Fed::current().as_ref(), fed);
    let resync = feed_resyncs_the_clock(change, fed.clock_peer.is_some(), crate::clock::last_outcome());
    nordisk_pairing::feedbus::publish(feed.clone());
    if change == FeedChange::default() && !resync {
        return;
    }
    if change.clock {
        tracing::info!(clock_peer = ?fed.clock_peer, clock_key = fed.clock_key.is_some(), "pairing: the door fed the clock — it applies from this sync on");
        crate::clock::fed_arrived();
    } else if resync {
        // ★ A re-laid box (an in-place upgrade's readmit) published the SAME clock at start
        // from the joined.toml it kept, and its boot sync could not reach the front yet.
        tracing::info!(clock_peer = ?fed.clock_peer, "pairing: the door fed the same clock again and this process has not set its clock right yet — a sync now");
        crate::clock::fed_arrived();
    }
    if change.monetize {
        // Applied, or alarmed by name (`feed::FEED_NOT_APPLIED`,
        // `feed::FEED_MISMATCH`) — never a restart. The key is in `joined.toml`
        // either way, so the next start reads it too.
        let _ = crate::feed::monetize_arrived(fed.monetize_pubkey.as_deref());
    }
}

/// **The pairing loop's turn at [`take_feed`]**: what the loop learned at its
/// last join (`learned`), with `taken` what it handed over before — handed
/// over when it is NEW, never again on every turn (cov-A r3 on oden
/// 2026-10-06). Between the loop's joins the installer feeds this box through
/// its listener (`installer_listener::Installed::fed`), which publishes the
/// newer feed itself; a loop that handed its join answer over on every turn
/// put that older answer back over it — the twin lost the front's clock
/// offer that way, for the life of the process.
pub fn take_learned_feed(taken: &mut Option<nordisk_pairing::feedbus::Feed>, learned: Option<&nordisk_pairing::feedbus::Feed>) {
    if learned.is_some() && learned != taken.as_ref() {
        take_feed(learned);
        *taken = learned.cloned();
    }
}

/// The signal, written and said.
fn report(root: &Path, seed: &Seed, name: &str, m: &Machine) {
    let s = Signal::of(name, seed.tunnel, m);
    tracing::info!("{}", s.line().trim());
    if let Some(text) = s.render() {
        let _ = crate::durable::write_atomic_unframed(root, SIGNAL_REL, text.as_bytes());
    }
}

/// **Start the installer's listener and the loop.** Returns at once.
///
/// In order: a fresh installation clears the pairing state it finds on the
/// data volume (`installer_listener::reset_if_fresh_install`); the installer's
/// view of this box is opened (its own key, what was fed before); the
/// installer's link and listener open — the install, or the log door alone
/// once the installer closed the install on this installation (for the life
/// of the VM either way); and the loop runs on its own thread until the box is paired
/// or refused for good, reporting every change.
pub fn start(stores: Arc<ServerStores>, seed: Seed) {
    let root = stores.root().to_path_buf();
    if let Some(cleared) = crate::installer_listener::reset_if_fresh_install(Path::new("/boot"), &root, &|p| crate::narrate::write_esp_file(p, b"the pairing state of the previous installation was cleared\n")) {
        tracing::info!(cleared = ?cleared, "pairing: the first boot of this installation — it starts UNPAIRED (the data volume's pairing state is cleared)");
    }
    let installed = match crate::installer_listener::Installed::open(&root, seed.clone()) {
        Ok(i) => Arc::new(i.with_stores(Arc::clone(&stores))),
        Err(e) => {
            tracing::error!(error = %format!("{e:#}"), "pairing: this box cannot prepare its own key — it will not pair");
            return;
        }
    };
    crate::installer_listener::start(Arc::clone(&installed));
    // Live update: an app started on trial confirms once it serves, or is rolled back.
    #[cfg(all(feature = "live-app", target_os = "linux"))]
    crate::live_update::spawn_confirm(crate::readiness::is_ready);
    // The metered primary registers its OWN plugin into vm1's monetize (once, and
    // again only when it changes) — `self_register`.
    #[cfg(feature = "metered")]
    crate::self_register::start(Arc::clone(&installed), Arc::clone(&stores));
    std::thread::Builder::new()
        .name("gunnar-pairing".into())
        .spawn(move || {
            // Built on this thread: the tunnel driver is the loop's alone.
            let doors = BoxDoors::new(stores, seed.clone(), installed);
            let mut m = Machine::default();
            let mut learned = Learned::default();
            let mut taken = None;
            loop {
                let req = request(&seed, &root, "", doors.tunnel_key());
                let before = m.clone();
                let step = turn(&mut m, &mut learned, &seed, &req, &doors);
                take_learned_feed(&mut taken, learned.feed.as_ref());
                let name = learned.name.clone().unwrap_or_else(|| "?".into());
                if m != before {
                    report(&root, &seed, &name, &m);
                }
                if step == Step::Nothing {
                    if m.paired() {
                        tracing::info!(name = %name, "pairing: PAIRED — the loop rests");
                    } else {
                        tracing::error!(name = %name, "pairing: REFUSED for good — re-seed this box");
                    }
                    return;
                }
                std::thread::sleep(Duration::from_secs(m.backoff_s()));
            }
        })
        .expect("spawning the pairing thread");
}

#[cfg(test)]
mod tests {
    use super::*;
    use nordisk_pairing::mtls;
    use nordisk_pairing::join::{Assigned, Peer};
    use nordisk_pairing::Role;
    use nordisk_pairing::seed::Front;
    use std::cell::{Cell, RefCell};

    fn a_seed() -> Seed {
        let ca = mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        Seed {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            front: Front { door: "10.13.8.179:8474".into() },
            kind: Kind::Gunnar,
            installer_ca_pem: ca.cert_pem,
            bootstrap_cert_pem: boot.cert_pem,
            bootstrap_key_pem: boot.key_pem,
            peers_expected: 3,
            mullvad: None,
            tailscale: None,
            installer_link: None,
        }
    }

    /// ★ **A release image's box takes its seed from the metadata** (no operator machine in a
    /// sale, owner 2026-10-08): the staged marker → the `user_data` seed, kept 0600 and read
    /// back as a medium's own on the next start; asked again while the metadata has none, and
    /// refused by name when it never does; a medium's own seed is read as before.
    #[test]
    fn a_release_image_takes_its_seed_from_the_metadata() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::create_dir_all(root.join("pairing")).unwrap();
        std::fs::write(root.join(SEED_REL), format!("# a release image: no seed\n{SEED_FROM_METADATA}\n")).unwrap();
        // A release image's seed, as the shop mints it per order: WireGuard, with the
        // installer's link (a wireguard seed without one is refused by its own check).
        let want = Seed {
            installer_link: Some(nordisk_pairing::seed::LinkSeed {
                installer_pubkey: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=".into(),
                vm_private: "ICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj8=".into(),
                port: nordisk_pairing::installer_link::LINK_PORT,
            }),
            ..a_seed()
        };
        let user_data = want.render().unwrap();
        let asked = Cell::new(0);
        let fetch = || -> Result<String, String> {
            asked.set(asked.get() + 1);
            if asked.get() < 3 { Err("the network is not up yet".into()) } else { Ok(user_data.clone()) }
        };
        let got = seed_with(root, &fetch, 5, Duration::from_millis(1)).unwrap().expect("a seed");
        assert_eq!((got.estate.as_str(), got.bootstrap_key_pem.as_str()), (want.estate.as_str(), want.bootstrap_key_pem.as_str()));
        assert_eq!(asked.get(), 3, "asked again while the metadata had none");
        assert_eq!(std::fs::metadata(root.join(SEED_REL)).unwrap().permissions().mode() & 0o777, 0o600);
        let never = || -> Result<String, String> { panic!("a kept seed is not asked for again") };
        assert_eq!(seed_with(root, &never, 1, Duration::ZERO).unwrap().unwrap().estate, want.estate, "the next start reads the kept seed");
        // The metadata never gives one: refused by name, nothing written.
        std::fs::write(root.join(SEED_REL), SEED_FROM_METADATA).unwrap();
        let empty = || -> Result<String, String> { Ok(String::new()) };
        let e = seed_with(root, &empty, 2, Duration::ZERO).unwrap_err().to_string();
        assert!(e.contains("gave none after 2 tries") && e.contains("laid without a seed"), "{e}");
        assert!(is_metadata_marker(&std::fs::read_to_string(root.join(SEED_REL)).unwrap()));
        assert!(seed_of_user_data("#cloud-config\nruncmd: [echo hi]\n").unwrap_err().contains("not a pairing seed"));
        assert!(!is_metadata_marker(&user_data), "a seed is not the marker");
    }

    /// The metadata is read with one plain GET: the body of a 200, anything else refused.
    #[test]
    fn the_metadata_is_read_with_one_plain_get() {
        use std::io::{Read as _, Write as _};
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let at = l.local_addr().unwrap();
        let t = std::thread::spawn(move || {
            for answer in ["HTTP/1.0 200 OK\r\nContent-Type: text/plain\r\n\r\nSEED TEXT", "HTTP/1.0 404 Not Found\r\n\r\nnone"] {
                let (mut c, _) = l.accept().unwrap();
                let mut buf = [0u8; 512];
                let n = c.read(&mut buf).unwrap();
                assert!(String::from_utf8_lossy(&buf[..n]).starts_with("GET /metadata/v1/user_data HTTP/1.0\r\nHost: 169.254.169.254"));
                c.write_all(answer.as_bytes()).unwrap();
            }
        });
        assert_eq!(http_get(at, "169.254.169.254", METADATA_USER_DATA).unwrap(), (200, "SEED TEXT".to_string()));
        assert_eq!(http_get(at, "169.254.169.254", METADATA_USER_DATA).unwrap().0, 404);
        t.join().unwrap();
    }

    fn req(seed: &Seed) -> JoinRequest {
        JoinRequest { estate: seed.estate.clone(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: None, csr_pem: String::new() }
    }

    /// Fix 2026-09-24: the FIRST join already carries this box's tunnel key —
    /// minted 0600 before any join, the same key the tunnel lane writes later.
    #[test]
    fn the_first_join_request_carries_a_non_empty_tunnel_key_minted_0600() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        assert!(!root.join(crate::wg::KEY_FILE).exists());
        let driver = tunnel_driver(&a_seed(), root, &Default::default()).unwrap();
        let first = request(&a_seed(), root, "csr", driver.public_key());
        assert!(!first.tunnel_pubkey.is_empty(), "the first join must carry a key: {first:?}");
        let mode = std::fs::metadata(root.join(crate::wg::KEY_FILE)).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600);
        let private = ensure_tunnel_key(root).unwrap();
        assert_eq!(wg_appliance_core::public_from_private(&private).unwrap(), first.tunnel_pubkey, "the join's key is the key on disk");
        let again = tunnel_driver(&a_seed(), root, &Default::default()).unwrap();
        assert_eq!(request(&a_seed(), root, "", again.public_key()).tunnel_pubkey, first.tunnel_pubkey, "a later join (a reboot) asks with the same key");
        let conf = wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, &private, None).unwrap();
        assert!(conf.contains(&format!("PrivateKey = {private}")), "the tunnel is raised with the key the door recorded");
    }

    /// The front as the door feeds it (the reply's `front` peer).
    fn fed_front() -> Peer {
        nordisk_pairing::join::FrontTunnel { tunnel_pubkey: "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=".into(), endpoint: "95.111.211.189:51820".into() }.as_peer()
    }

    #[test]
    fn the_config_written_for_the_assigned_address_is_the_shape_the_tunnel_lane_parses() {
        let text = wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, "gEi6mvh6b5rKDg7UkleGpsJLlA9ueOKyWhjqJqFjWHg=", None).unwrap();
        assert!(text.contains("Address = 10.44.0.2/24"), "{text}");
        assert!(text.contains("AllowedIPs = 10.44.0.0/24") && text.contains("Endpoint = 95.111.211.189:51820"), "{text}");
        assert!(text.contains("PublicKey = xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg="), "the front's key is the FED one: {text}");
        assert!(!text.contains("ListenPort"), "a gunnar box only dials");
        let no_endpoint = Peer { endpoint: None, ..fed_front() };
        assert!(wg_conf(&no_endpoint, nordisk_pairing::OVERLAY_BOX_A, "k", None).unwrap_err().contains("no endpoint"));
    }

    const PSK: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";

    /// **The link to the front carries the fed pre-shared key** (2026-10-06,
    /// the PQ transport lane): `PresharedKey` in the front's `[Peer]`, read
    /// back by the tunnel lane as that peer's 32 bytes. A fed value that is
    /// not a WireGuard key is refused by name, never written.
    #[test]
    fn the_conf_carries_the_fed_preshared_key_and_the_tunnel_lane_reads_it() {
        let private = "gEi6mvh6b5rKDg7UkleGpsJLlA9ueOKyWhjqJqFjWHg=";
        let text = wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, private, Some(PSK)).unwrap();
        assert!(text.contains(&format!("PresharedKey = {PSK}")), "{text}");
        let parsed = crate::wg::WgConfig::parse("wg0", &text).unwrap();
        assert_eq!(parsed.peers.len(), 1);
        assert_eq!(parsed.peers[0].preshared_key, Some(core::array::from_fn(|i| i as u8)), "the front's peer carries it");
        let plain = wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, private, None).unwrap();
        assert!(!plain.contains("PresharedKey"), "none fed, none written: {plain}");
        assert!(wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, private, Some("not-a-key")).unwrap_err().contains("pre-shared key"));
    }

    /// **The driver takes the PSK from what the installer fed** (the sealed
    /// `joined.toml`) and writes `wg0.conf` 0600 — it holds the private key
    /// and the PSK. Raising the tunnel needs CAP_NET_ADMIN, which a test does
    /// not have; the conf is written before that.
    #[test]
    fn the_driver_writes_wg0_conf_0600_with_the_psk_the_installer_fed() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let reply = JoinReply {
            estate: "gunnar.rs".into(),
            me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
            cert_pem: None,
            role: Some(Role::Primary),
            primary_overlay: None,
            primary_utility: None,
            peers: vec![fed_front()],
            feed: std::collections::BTreeMap::from([(nordisk_pairing::installer::FEED_WG_PSK.to_string(), PSK.to_string())]),
        };
        crate::fed_at_rest::write(root, JOINED_REL, &reply.render().unwrap()).unwrap();
        let mut driver = tunnel_driver(&a_seed(), root, &Default::default()).unwrap();
        let _ = driver.up(&Identity { overlay: nordisk_pairing::OVERLAY_BOX_A, pubkey: String::new() }, &reply.peers);
        let conf = std::fs::read_to_string(root.join(crate::wg::CONFIG_FILE)).unwrap();
        assert!(conf.contains(&format!("PresharedKey = {PSK}")), "{conf}");
        let mode = std::fs::metadata(root.join(crate::wg::CONFIG_FILE)).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "wg0.conf holds the private key and the PSK");
    }

    /// The medium is constant, the relay is not: a conf THIS loop wrote for an
    /// older front is replaced; an operator's conf never is.
    #[test]
    fn our_conf_is_rewritten_for_another_front_and_an_operators_never() {
        let now = wg_conf(&fed_front(), nordisk_pairing::OVERLAY_BOX_A, "k", None).unwrap();
        let other = Peer { tunnel_pubkey: "AAAABBBBCCCCDDDDEEEEFFFFGGGGHHHHIIIIJJJJKKK=".into(), ..fed_front() };
        let before = wg_conf(&other, nordisk_pairing::OVERLAY_BOX_A, "k", None).unwrap();
        assert!(conf_needs_writing(None, &now), "none yet");
        assert!(!conf_needs_writing(Some(&now), &now), "the same: left alone");
        assert!(conf_needs_writing(Some(&before), &now), "ours, for another front: rewritten");
        assert!(!conf_needs_writing(Some("[Interface]\nPrivateKey = x\n"), &now), "an operator's: never");
    }

    /// The tunnel step of a WireGuard box whose last answer carried no front
    /// asks the door again, instead of retrying a stale answer forever.
    #[test]
    fn a_wireguard_box_with_no_fed_front_asks_the_door_again_before_raising_the_tunnel() {
        struct NoFrontThenFront(Cell<u32>);
        impl Doors for NoFrontThenFront {
            fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
                let n = self.0.get() + 1;
                self.0.set(n);
                let peers = if n == 1 { vec![] } else { vec![fed_front()] };
                Ok(JoinReply { estate: "gunnar.rs".into(), me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A }, cert_pem: None, role: Some(Role::Primary), primary_overlay: None, primary_utility: None, peers, feed: Default::default() })
            }
            fn tunnel_up(&self, _: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
                assert!(peers.iter().any(|p| p.role == Role::Front), "the tunnel is raised only with the fed front");
                Ok(true)
            }
            fn admin(&self, _: Option<&str>) -> Result<bool, String> {
                Ok(true)
            }
            fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
                unreachable!()
            }
            fn follow_state(&self) -> Result<(u64, u64), String> {
                unreachable!()
            }
        }
        let seed = a_seed();
        let doors = NoFrontThenFront(Cell::new(0));
        let (mut m, mut learned) = (Machine::default(), Learned::default());
        let mut steps = Vec::new();
        for _ in 0..10 {
            let s = turn(&mut m, &mut learned, &seed, &req(&seed), &doors);
            if s == Step::Nothing {
                break;
            }
            steps.push(s.name());
        }
        assert!(m.paired(), "{m:?}");
        assert_eq!(doors.0.get(), 2, "joined, then asked again for the front");
        assert_eq!(steps, ["join", "tunnel-up", "tunnel-up", "check-admin"], "{steps:?}");
    }

    /// Fake doors: the election needs one extra ask; the tunnel is up at once;
    /// the primary is not listening for the first two BecomeTwin tries; the
    /// follower catches up on its third poll.
    struct Fake {
        joins: Cell<u32>,
        becomes: Cell<u32>,
        polls: Cell<u32>,
        role: Role,
        admin_fed: RefCell<Option<String>>,
        tunnel_for: RefCell<Option<Ipv4Addr>>,
    }

    impl Fake {
        fn new(role: Role) -> Fake {
            Fake { joins: Cell::new(0), becomes: Cell::new(0), polls: Cell::new(0), role, admin_fed: RefCell::new(None), tunnel_for: RefCell::new(None) }
        }
    }

    impl Doors for Fake {
        fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
            let n = self.joins.get() + 1;
            self.joins.set(n);
            let (me, overlay) = match self.role {
                Role::Twin => ("main", nordisk_pairing::OVERLAY_BOX_A),
                _ => ("twin", nordisk_pairing::OVERLAY_BOX_B),
            };
            let mut feed = std::collections::BTreeMap::new();
            feed.insert(FEED_ADMIN.to_string(), "ssh-ed25519 AAAAconsole".to_string());
            Ok(JoinReply {
                estate: "gunnar.rs".into(),
                me: Assigned { name: me.into(), overlay },
                cert_pem: None,
                role: if n == 1 { None } else { Some(self.role) },
                primary_overlay: Some(nordisk_pairing::OVERLAY_BOX_B),
                primary_utility: None,
                peers: vec![Peer { name: "front".into(), role: Role::Front, tunnel_pubkey: String::new(), overlay: nordisk_pairing::OVERLAY_FRONT, endpoint: None, utility: None }],
                feed,
            })
        }
        fn tunnel_up(&self, overlay: Ipv4Addr, _: &[Peer]) -> Result<bool, String> {
            *self.tunnel_for.borrow_mut() = Some(overlay);
            Ok(true)
        }
        fn admin(&self, fed: Option<&str>) -> Result<bool, String> {
            *self.admin_fed.borrow_mut() = fed.map(str::to_string);
            Ok(fed.is_some())
        }
        fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
            assert_eq!(primary, nordisk_pairing::OVERLAY_BOX_B);
            let n = self.becomes.get() + 1;
            self.becomes.set(n);
            if n < 3 { Err("could not be reached".into()) } else { Ok(()) }
        }
        fn follow_state(&self) -> Result<(u64, u64), String> {
            let n = self.polls.get() + 1;
            self.polls.set(n);
            Ok(match n {
                1 => (0, 12),
                2 => (5, 12),
                _ => (12, 12),
            })
        }
    }

    fn run(doors: &Fake) -> (Machine, Learned, Vec<&'static str>) {
        let seed = a_seed();
        let r = req(&seed);
        let (mut m, mut learned) = (Machine::default(), Learned::default());
        let mut steps = Vec::new();
        for _ in 0..20 {
            let s = turn(&mut m, &mut learned, &seed, &r, doors);
            if s == Step::Nothing {
                break;
            }
            steps.push(s.name());
        }
        (m, learned, steps)
    }

    #[test]
    fn a_twin_joins_then_raises_the_tunnel_on_its_assigned_address_then_becomes_twin_and_follows_with_no_admin() {
        let doors = Fake::new(Role::Twin);
        let (m, learned, steps) = run(&doors);
        assert!(m.paired(), "{m:?}");
        assert_eq!(steps, ["join", "join", "tunnel-up", "become-twin", "become-twin", "become-twin", "certify", "certify", "follow"], "{steps:?}");
        assert_eq!(*doors.tunnel_for.borrow(), Some(nordisk_pairing::OVERLAY_BOX_A), "the address is the door's answer");
        assert_eq!(learned.name.as_deref(), Some("main"));
        assert!(doors.admin_fed.borrow().is_none(), "a twin must be fresh for BecomeTwin: it installs no admin");
    }

    /// **MEASURED 2026-09-25 (lane B's bench)**: vm1's offers reach the
    /// primary AFTER its join and BEFORE its admin. The admin step must take
    /// them from the LATEST feed — the join step's copy predates them — or
    /// `front-certify` is never enrolled and the twin is never certified.
    #[test]
    fn a_primary_enrols_the_fronts_offers_that_arrived_after_its_join_and_before_its_admin() {
        use nordisk_pairing::bootstrap::Vm as _;
        let data = tempfile::tempdir().unwrap();
        let stores = Arc::new(ServerStores::open(data.path()).unwrap());
        let installed = Arc::new(crate::installer_listener::Installed::open(data.path(), a_seed()).unwrap().with_stores(Arc::clone(&stores)));
        let doors = BoxDoors::new(Arc::clone(&stores), a_seed(), Arc::clone(&installed));
        let key = || {
            let pem = gunnar_core::mint_ed25519().unwrap();
            gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).unwrap().public_key().to_openssh().unwrap()
        };
        let (op, certify) = (key(), key());
        let reply = |feed: Vec<(String, String)>| JoinReply {
            estate: "gunnar.rs".into(),
            me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
            cert_pem: None,
            role: Some(Role::Primary),
            primary_overlay: None,
            primary_utility: None,
            peers: vec![],
            feed: feed.into_iter().collect(),
        };
        // 1. fed and joined with the admin only (vm1 has offered nothing yet)
        installed.fed(&reply(vec![(FEED_ADMIN.into(), op.clone())])).unwrap();
        doors.join(&req(&a_seed())).unwrap();
        // 2. vm1's offer arrives — the primary has no admin yet, so the late path waits
        installed.fed(&reply(vec![(FEED_ADMIN.into(), op.clone()), (format!("offer.front.{}", gunnar_pairing_door::front::OFFER_CERTIFY_PUBKEY), certify)])).unwrap();
        assert!(stores.auth().principal(gunnar_pairing_door::front::CERTIFY_PRINCIPAL).unwrap().is_none());
        // 3. the admin step — from the LATEST feed
        assert!(doors.admin(Some(&op)).unwrap());
        assert!(stores.auth().principal(gunnar_pairing_door::front::CERTIFY_PRINCIPAL).unwrap().is_some(), "front-certify enrolled from the late offer");
    }

    /// ★ **The WireGuard-only upgrade drill (2026-10-05)**: an existing twin
    /// was joined on its primary's utility address; upgraded, the door names the
    /// same primary by its overlay address. The pairing reads the utility address
    /// it moves from ONLY from that door's answer, and only for that same box.
    #[test]
    fn an_upgraded_twin_moves_from_its_primarys_utility_address_only_for_the_same_box() {
        let (u, o) = (Ipv4Addr::new(10, 13, 12, 118), nordisk_pairing::OVERLAY_BOX_B);
        let reply = |overlay: Option<Ipv4Addr>, utility: Option<Ipv4Addr>| JoinReply {
            estate: "gunnar.rs".into(),
            me: Assigned { name: "vm2".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
            cert_pem: None,
            role: Some(Role::Twin),
            primary_overlay: overlay,
            primary_utility: utility,
            peers: vec![],
            feed: Default::default(),
        };
        assert_eq!(same_primary_utility(Some(&reply(Some(o), Some(u))), o), Some(u), "the same primary: its utility address");
        assert_eq!(same_primary_utility(Some(&reply(Some(o), Some(u))), u), None, "told the utility address itself: nothing moves");
        assert_eq!(same_primary_utility(Some(&reply(Some(Ipv4Addr::new(10, 200, 0, 9)), Some(u))), o), None, "another box's overlay: nothing moves");
        assert_eq!(same_primary_utility(Some(&reply(Some(o), None)), o), None, "no utility address named");
        assert_eq!(same_primary_utility(None, o), None, "no answer from the door yet");
    }

    #[test]
    fn a_primary_installs_the_fed_admin_key_and_is_paired() {
        let doors = Fake::new(Role::Primary);
        let (m, _, steps) = run(&doors);
        assert!(m.paired(), "{m:?}");
        assert_eq!(steps, ["join", "join", "tunnel-up", "check-admin"]);
        assert_eq!(doors.admin_fed.borrow().as_deref(), Some("ssh-ed25519 AAAAconsole"), "the admin key came from the door, not the medium");
        assert_eq!(doors.becomes.get(), 0, "a primary never becomes a twin");
    }

    /// **c-paired-restore.** A primary the installer RE-ADMITTED into its closed
    /// estate (nordisk-pairing `Door::readmit`: fed `recharge_from`, the twin's
    /// control plane) must NOT install its fed admin while it is fresh — the
    /// first identity shuts the bootstrap window and `Bootstrap.Restore`
    /// (`twin::prepare_replay`) refuses every box whose window is shut. It holds
    /// the admin back while it awaits its twin, while the restore runs and until
    /// it has restarted; then it takes the ordinary admin step and is PAIRED.
    #[test]
    fn a_recharged_primary_holds_its_admin_back_until_it_is_restored_and_restarted() {
        struct Recharged {
            holds: RefCell<Vec<Hold>>,
            admin_calls: Cell<u32>,
            asked_from: RefCell<Option<String>>,
        }
        impl Doors for Recharged {
            fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
                let mut feed = std::collections::BTreeMap::new();
                feed.insert(FEED_ADMIN.to_string(), "ssh-ed25519 AAAAconsole".to_string());
                feed.insert(FEED_RECHARGE_FROM.to_string(), "10.13.8.3:50051".to_string());
                Ok(JoinReply {
                    estate: "gunnar.rs".into(),
                    me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
                    cert_pem: None,
                    role: Some(Role::Primary),
                    primary_overlay: None,
                    primary_utility: None,
                    peers: vec![fed_front()],
                    feed,
                })
            }
            fn tunnel_up(&self, _: Ipv4Addr, _: &[Peer]) -> Result<bool, String> {
                Ok(true)
            }
            fn admin(&self, _: Option<&str>) -> Result<bool, String> {
                self.admin_calls.set(self.admin_calls.get() + 1);
                Ok(true)
            }
            fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
                unreachable!("a recharged main is the primary it was")
            }
            fn follow_state(&self) -> Result<(u64, u64), String> {
                unreachable!()
            }
            fn recharge_hold(&self, from: &str) -> Result<Hold, String> {
                *self.asked_from.borrow_mut() = Some(from.to_string());
                let mut h = self.holds.borrow_mut();
                Ok(if h.is_empty() { Hold::Restored } else { h.remove(0) })
            }
        }
        let doors = Recharged {
            holds: RefCell::new(vec![Hold::Awaiting, Hold::Awaiting, Hold::Restoring { head: 7 }, Hold::Restarting, Hold::Restored]),
            admin_calls: Cell::new(0),
            asked_from: RefCell::new(None),
        };
        let seed = a_seed();
        let (mut m, mut learned) = (Machine::default(), Learned::default());
        let mut steps = Vec::new();
        for _ in 0..20 {
            let s = turn(&mut m, &mut learned, &seed, &req(&seed), &doors);
            if s == Step::Nothing {
                break;
            }
            steps.push(s.name());
            if s.name() == "check-admin" && doors.admin_calls.get() == 0 {
                assert!(!m.paired(), "not paired while the recharge holds: {m:?}");
            }
        }
        assert!(m.paired(), "{m:?} {steps:?}");
        assert_eq!(doors.admin_calls.get(), 1, "the admin step ran once, after the restore and the restart");
        assert_eq!(doors.asked_from.borrow().as_deref(), Some("10.13.8.3:50051"));
        assert_eq!(learned.recharge_from.as_deref(), Some("10.13.8.3:50051"));
        assert_eq!(steps.iter().filter(|s| **s == "check-admin").count(), 5, "{steps:?}");
    }

    /// **The real doors of a recharged main**: fresh and fed `recharge_from`,
    /// it says AWAITING, keeps its bootstrap window OPEN and writes the marker
    /// `Bootstrap.Status.recharge_from` answers from; once it holds identities
    /// (restored, restarted) it is RESTORED and the marker is gone.
    #[test]
    fn a_fresh_box_fed_recharge_from_awaits_with_its_window_open_and_a_restored_one_does_not() {
        use nordisk_pairing::bootstrap::Vm as _;
        let data = tempfile::tempdir().unwrap();
        let stores = Arc::new(ServerStores::open(data.path()).unwrap());
        let installed = Arc::new(crate::installer_listener::Installed::open(data.path(), a_seed()).unwrap().with_stores(Arc::clone(&stores)));
        let doors = BoxDoors::new(Arc::clone(&stores), a_seed(), Arc::clone(&installed));
        let pem = gunnar_core::mint_ed25519().unwrap();
        let op = gunnar_core::ssh_key::PrivateKey::from_openssh(&pem).unwrap().public_key().to_openssh().unwrap();
        let feed: std::collections::BTreeMap<String, String> = [(FEED_ADMIN.to_string(), op.clone()), (FEED_RECHARGE_FROM.to_string(), "10.13.8.3:50051".to_string())].into_iter().collect();
        installed
            .fed(&JoinReply { estate: "gunnar.rs".into(), me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A }, cert_pem: None, role: Some(Role::Primary), primary_overlay: None, primary_utility: None, peers: vec![], feed })
            .unwrap();
        doors.join(&req(&a_seed())).unwrap();
        assert_eq!(doors.recharge_hold("10.13.8.3:50051").unwrap(), Hold::Awaiting);
        assert!(stores.auth().bootstrap_window_open().unwrap(), "the window stays open for the restore");
        assert_eq!(std::fs::read_to_string(data.path().join(RECHARGE_REL)).unwrap().trim(), "10.13.8.3:50051");
        assert_eq!(recharge_awaited(&stores).as_deref(), Some("10.13.8.3:50051"), "what Bootstrap.Status answers");
        stores.auth().bootstrap_admin(&op).unwrap();
        assert_eq!(doors.recharge_hold("10.13.8.3:50051").unwrap(), Hold::Restored);
        assert!(!data.path().join(RECHARGE_REL).exists(), "the marker goes once the box holds its identities");
        assert_eq!(recharge_awaited(&stores), None);
    }

    /// The loop's tunnel goes through the seed's driver: WireGuard is gunnar's
    /// own driver; `none` is REFUSED [installer-link-none] (0.4.0: WireGuard +
    /// mutual TLS always) and writes no wg0; a kind this image carries no
    /// driver for is the named refusal. Both END the loop.
    #[test]
    fn the_tunnel_is_chosen_by_the_seed_kind_through_the_trait() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let mut seed = a_seed();
        assert_eq!(tunnel_driver(&seed, root, &Default::default()).unwrap().kind(), TunnelKind::Wireguard);

        seed.tunnel = TunnelKind::None;
        let none = tunnel_driver(&seed, root, &Default::default()).err().expect("`none` is refused");
        assert!(none.to_string().starts_with("REFUSED [installer-link-none]"), "{none}");
        assert!(!root.join(crate::wg::CONFIG_FILE).exists(), "`none` raises no wg0");

        for kind in [TunnelKind::Tailscale, TunnelKind::Mullvad] {
            seed.tunnel = kind;
            if let Err(e) = tunnel_driver(&seed, root, &Default::default()) {
                assert!(e.to_string().starts_with("REFUSED [tunnel-not-built]"), "{e}");
            }
        }

        /// Doors whose tunnel is the real driver selection for the seed.
        struct Driven(std::cell::RefCell<Result<Box<dyn Tunnel>, TunnelError>>);
        impl Doors for Driven {
            fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
                Ok(JoinReply {
                    estate: "gunnar.rs".into(),
                    me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
                    cert_pem: None,
                    role: Some(Role::Primary),
                    primary_overlay: Some(nordisk_pairing::OVERLAY_BOX_A),
                    primary_utility: None,
                    peers: vec![],
                    feed: Default::default(),
                })
            }
            fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
                let mut t = self.0.borrow_mut();
                let t = t.as_mut().map_err(|e| e.to_string())?;
                t.up(&Identity { overlay, pubkey: String::new() }, peers).map_err(|e| e.to_string())?;
                Ok(t.status().up)
            }
            fn admin(&self, _: Option<&str>) -> Result<bool, String> {
                Ok(true)
            }
            fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
                unreachable!()
            }
            fn follow_state(&self) -> Result<(u64, u64), String> {
                unreachable!()
            }
        }
        let run_with = |seed: &Seed, driver: Result<Box<dyn Tunnel>, TunnelError>| {
            let doors = Driven(std::cell::RefCell::new(driver));
            let (mut m, mut learned) = (Machine::default(), Learned::default());
            for _ in 0..10 {
                if turn(&mut m, &mut learned, seed, &req(seed), &doors) == Step::Nothing {
                    break;
                }
            }
            m
        };
        seed.tunnel = TunnelKind::None;
        let refused = run_with(&seed, tunnel_driver(&seed, root, &Default::default()));
        assert!(!refused.paired() && refused.next() == Step::Nothing, "`none` never pairs: {refused:?}");
        assert!(Signal::of("main", TunnelKind::None, &refused).line().contains("[installer-link-none]"));
        seed.tunnel = TunnelKind::Tailscale;
        let refused = run_with(&seed, Err(TunnelError::NotBuilt(TunnelKind::Tailscale)));
        assert!(!refused.paired() && refused.next() == Step::Nothing, "{refused:?}");
        assert!(Signal::of("main", TunnelKind::Tailscale, &refused).line().contains("[tunnel-not-built]"));
    }

    struct Withdrawn;
    impl Doors for Withdrawn {
        fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
            Err("REFUSED [join-withdrawn] every box has joined; the bootstrap certificate is valid during bootstrap only".into())
        }
        fn tunnel_up(&self, _: Ipv4Addr, _: &[nordisk_pairing::join::Peer]) -> Result<bool, String> {
            unreachable!()
        }
        fn admin(&self, _: Option<&str>) -> Result<bool, String> {
            unreachable!()
        }
        fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
            unreachable!()
        }
        fn follow_state(&self) -> Result<(u64, u64), String> {
            unreachable!()
        }
    }

    fn a_twin_peer(utility: Ipv4Addr, overlay: Ipv4Addr) -> Peer {
        Peer { name: "twin".into(), role: Role::Twin, tunnel_pubkey: String::new(), overlay, endpoint: None, utility: Some(utility) }
    }

    fn a_mains_reply(peers: Vec<Peer>) -> JoinReply {
        JoinReply {
            estate: "gunnar.rs".into(),
            me: Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
            cert_pem: None,
            role: Some(Role::Primary),
            primary_overlay: None,
            primary_utility: None,
            peers,
            feed: std::collections::BTreeMap::new(),
        }
    }

    /// ★ **ASK-twin** (resilience-gunnar 21:10): a sold main fed neither `--repair-twin` nor
    /// `recharge_from` refused a paid step `no_twin` — yet the door names every box of the
    /// estate to it, the twin included. The main's twin is that peer's control plane: its
    /// utility address, or its overlay on a WireGuard-only estate.
    #[test]
    fn a_main_finds_its_twins_control_plane_in_the_doors_peer_list() {
        let twin = a_twin_peer(Ipv4Addr::new(10, 13, 8, 3), Ipv4Addr::new(10, 44, 0, 3));
        let reply = a_mains_reply(vec![fed_front(), twin.clone()]);
        assert_eq!(twin_of(&reply, false).as_deref(), Some("10.13.8.3:50051"));
        assert_eq!(twin_of(&reply, true).as_deref(), Some("10.44.0.3:50051"), "WireGuard-only: over the tunnel");
        let no_utility = Peer { utility: None, ..twin.clone() };
        assert_eq!(twin_of(&a_mains_reply(vec![no_utility]), false).as_deref(), Some("10.44.0.3:50051"), "a ledger without utility: the overlay");
        assert_eq!(twin_of(&a_mains_reply(vec![fed_front()]), false), None, "no twin named, none known");
        assert_eq!(twin_of(&JoinReply { role: Some(Role::Twin), ..reply.clone() }, false), None, "only a main has a twin to wait for");
        assert_eq!(twin_of(&JoinReply { role: None, ..reply.clone() }, false), None, "not elected yet");

        let dir = tempfile::tempdir().unwrap();
        assert_eq!(door_twin(dir.path()), None, "nothing kept, nothing known");
        crate::fed_at_rest::write(dir.path(), JOINED_REL, &reply.render().unwrap()).unwrap();
        assert_eq!(door_twin(dir.path()), twin_of(&reply, crate::tunnel_addr::wg_only()), "the kept answer");
        assert!(door_twin(dir.path()).is_some());
    }

    #[test]
    fn a_withdrawn_bootstrap_ends_the_loop_by_name() {
        let seed = a_seed();
        let (mut m, mut learned) = (Machine::default(), Learned::default());
        turn(&mut m, &mut learned, &seed, &req(&seed), &Withdrawn);
        assert_eq!(m.next(), Step::Nothing);
        assert!(Signal::of("?", TunnelKind::Wireguard, &m).line().contains("REFUSED for good: REFUSED [join-withdrawn]"));
    }
}

#[cfg(test)]
mod feed_tests {
    use super::*;
    use nordisk_pairing::mtls;
    use crate::feed::Fed;

    fn fed(m: Option<&str>, peer: Option<&str>, key: Option<&str>) -> Fed {
        Fed { monetize_pubkey: m.map(Into::into), clock_peer: peer.map(Into::into), clock_key: key.map(Into::into), ..Fed::default() }
    }

    /// ★ **A re-laid box re-fed the SAME clock syncs it when this process has not set its clock
    /// right** (test5's in-place upgrade on the KVM mock, 2026-10-09: the twin kept /data, so
    /// `load_at_start` published the old feed; its boot sync could not reach the front — the
    /// tunnel was not up — and the installer's readmit feed carried the same peer and key, so
    /// it was "no change", no fed sync ran, and the twin's every credential was refused
    /// `-7198470 ms away` for the hour the readmit waited). A clock already set right in this
    /// process is not asked again by a repeat feed; a new clock always is.
    #[test]
    fn a_repeated_clock_feed_syncs_while_this_process_has_not_set_its_clock_right() {
        use estate_clock::report::Outcome;
        let peer = fed(None, Some("peer://10.44.0.1:50123"), Some("ssh-ed25519 AAAAk"));
        let same = feed_change(Some(&peer), &peer);
        assert_eq!(same, FeedChange::default());
        for unset in [None, Some(Outcome::NoQuorum), Some(Outcome::Disagree), Some(Outcome::NotConfigured), Some(Outcome::ApplyFailed)] {
            assert!(feed_resyncs_the_clock(same, true, unset), "a repeat feed after {unset:?} must sync");
        }
        for right in [Outcome::Stepped, Outcome::Slewed, Outcome::AlreadyRight] {
            assert!(!feed_resyncs_the_clock(same, true, Some(right)), "a clock set right ({right:?}) is not asked again by a repeat");
        }
        assert!(!feed_resyncs_the_clock(same, false, Some(Outcome::NoQuorum)), "a feed with no clock starts no sync");
        assert!(feed_resyncs_the_clock(FeedChange { clock: true, monetize: false }, true, Some(Outcome::AlreadyRight)), "a NEW clock always syncs");
    }

    #[test]
    fn the_first_answer_changes_what_it_carries_and_a_repeat_changes_nothing() {
        let first = fed(Some("aa"), Some("peer://10.13.8.179:50123"), Some("ssh-ed25519 AAAAk"));
        assert_eq!(feed_change(None, &first), FeedChange { clock: true, monetize: true });
        assert_eq!(feed_change(Some(&first), &first), FeedChange::default(), "a re-join with the same feed re-applies nothing");
        assert_eq!(feed_change(None, &fed(None, Some("peer://10.13.8.179:50123"), None)), FeedChange { clock: true, monetize: false });
        assert_eq!(feed_change(Some(&first), &fed(Some("bb"), Some("peer://10.13.8.179:50123"), Some("ssh-ed25519 AAAAk"))), FeedChange { clock: false, monetize: true });
    }

    /// The join step records the fed configuration values beside the admin key.
    #[test]
    fn a_join_learns_the_fed_values() {
        struct Feeds;
        impl Doors for Feeds {
            fn join(&self, _: &JoinRequest) -> Result<JoinReply, String> {
                let mut feed = std::collections::BTreeMap::new();
                feed.insert(FEED_ADMIN.to_string(), "ssh-ed25519 AAAAconsole".to_string());
                feed.insert(crate::feed::FEED_MONETIZE.to_string(), "ab".repeat(32));
                feed.insert(crate::feed::FEED_CLOCK_PEER.to_string(), "peer://10.13.8.179:50123".to_string());
                feed.insert(crate::feed::FEED_CLOCK_KEY.to_string(), "ssh-ed25519 AAAAk".to_string());
                Ok(JoinReply {
                    estate: "gunnar.rs".into(),
                    me: nordisk_pairing::join::Assigned { name: "main".into(), overlay: nordisk_pairing::OVERLAY_BOX_A },
                    cert_pem: None,
                    role: None,
                    primary_overlay: None,
                    primary_utility: None,
                    peers: vec![],
                    feed,
                })
            }
            fn tunnel_up(&self, _: Ipv4Addr, _: &[nordisk_pairing::join::Peer]) -> Result<bool, String> { unreachable!() }
            fn admin(&self, _: Option<&str>) -> Result<bool, String> { unreachable!() }
            fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> { unreachable!() }
            fn follow_state(&self) -> Result<(u64, u64), String> { unreachable!() }
        }
        let seed = tests_seed();
        let req = JoinRequest { estate: seed.estate.clone(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: None, csr_pem: String::new() };
        let (mut m, mut learned) = (Machine::default(), Learned::default());
        turn(&mut m, &mut learned, &seed, &req, &Feeds);
        assert_eq!(learned.feed.as_ref().map(Fed::from_feed), Some(fed(Some(&"ab".repeat(32)), Some("peer://10.13.8.179:50123"), Some("ssh-ed25519 AAAAk"))));
    }

    fn tests_seed() -> Seed {
        let ca = mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        Seed {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            front: nordisk_pairing::seed::Front { door: "10.13.8.179:8474".into() },
            kind: Kind::Gunnar,
            installer_ca_pem: ca.cert_pem,
            bootstrap_cert_pem: boot.cert_pem,
            bootstrap_key_pem: boot.key_pem,
            peers_expected: 3,
            mullvad: None,
            tailscale: None,
            installer_link: None,
        }
    }
}