nordisk-pairing 0.5.28

//! **The pairing loop a product box runs itself, after the feed** (0.5.5,
//! 2026-09-27).
//!
//! [`crate::boxside`] keeps what the installer FEEDS a box; this is what the
//! box then DOES with it, until it can report PAIRED — the installer closes on
//! `CloseWhen::Paired`, so a box that never writes a paired signal never
//! finishes an install (measured on holger 2026-09-27: it wrote no signal,
//! brought no network, raised no wg0). The loop was gunnar-server's own
//! (`pairing.rs`: `turn()`, `BoxDoors`, the reporting thread); it lives here
//! once, and what only the product knows — how its primary is ready, how its
//! twin becomes the twin, is certified and follows — is the [`Steps`] half.
//!
//! ```text
//! the installer feeds (boxside) ──▶ Join: read the feed ──▶ BringTunnelUp: wg0 on the fed
//!   overlay address, the front's relay owning 10.44.0.0/24 ──▶ primary: Steps::primary_ready
//!                                                          └─▶ twin: become → certify → follow
//! every change ──▶ <root>/pairing/signal.toml ──▶ Vm::status ──▶ the installer: PAIRED → close
//! ```
//!
//! The join is a READ of what the installer fed ([`crate::boxside::Installed::latest`]):
//! 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.

use std::collections::BTreeMap;
use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::boxside::{write_atomic, Installed, Product, SIGNAL_REL};
use crate::join::{JoinReply, Peer};
use crate::pair::{Event, Machine, Step};
use crate::seed::{Seed, TunnelKind};
use crate::signal::Signal;
use crate::tunnel::{Identity, Tunnel, TunnelError};
use crate::Role;

/// **What only the product knows** — the steps after the tunnel is up.
/// gunnar's are its control plane (`bootstrap_admin`, `BecomeTwin`, its
/// follower's heartbeat); holger's are its role record and, once
/// holger-twin-mtls lands, its certified twin.
pub trait Steps: Send + Sync {
    /// PRIMARY: install what the feed hands the primary (gunnar: the fed admin
    /// key and principals), then say whether this box is ready as the primary.
    /// `Ok(false)` = not yet (a transient, asked again).
    fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String>;
    /// TWIN: become the twin of the primary at `primary` (its utility address
    /// when the installer fed one, else its overlay address). Idempotent.
    fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
    /// TWIN: `(applied, head)` of this box's follower, or why not.
    fn follow_state(&self) -> Result<(u64, u64), String>;
    /// TWIN: is this twin certified by the primary's issuer? gunnar's answer
    /// (the default): the certificate is the FRONT's move, and the twin sees it
    /// took when its follower has applied an entry.
    fn certified(&self, _primary: Ipv4Addr) -> Result<bool, String> {
        self.follow_state().map(|(applied, _)| applied > 0)
    }
}

/// **The doors the loop opens**, as a trait so [`turn`] is testable without a
/// network, a data root or a control plane. [`BoxDoors`] is the real one.
pub trait Doors {
    /// What the installer fed, or why there is nothing yet.
    fn join(&self) -> Result<JoinReply, String>;
    /// Raise this box's tunnel on `overlay` with these peers. `Ok(true)` = up.
    fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String>;
    fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String>;
    fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String>;
    fn certified(&self, primary: Ipv4Addr) -> Result<bool, String>;
    fn follow_state(&self) -> Result<(u64, u64), String>;
    /// **After the loop rests** (0.5.17): what the box still does for the life of the
    /// process, on the loop's thread (the tunnel driver is the loop's alone). [`BoxDoors`]
    /// follows the front every re-feed names ([`BoxDoors::follow_the_front`]); the default
    /// holds the doors, and the tunnel with them, and does nothing else.
    fn rest(&self, _say: &dyn Fn(&str)) {
        loop {
            std::thread::park();
        }
    }
}

/// The loop's memory between turns: what the feed said.
#[derive(Debug, Default, Clone)]
pub struct Learned {
    pub overlay: Option<Ipv4Addr>,
    /// The peers the feed named (handed to the tunnel driver).
    pub peers: Vec<Peer>,
    pub name: Option<String>,
    pub feed: BTreeMap<String, String>,
}

/// The refusals no retry changes: a spent or forged join, a medium composed
/// without the tunnel its seed names.
const FOR_GOOD: [&str; 7] = ["[join-spent]", "[join-forged]", "[join-unknown]", "[join-estate]", "[join-withdrawn]", "[tunnel-not-built]", "[installer-link-none]"];

/// **One turn of the loop**: do what the machine asks, tell it what happened.
/// Pure over the doors — gunnar-server's `pairing::turn`, product-neutral.
pub fn turn(m: &mut Machine, learned: &mut Learned, seed: &Seed, doors: &dyn Doors) -> Step {
    let step = m.next();
    let event = match &step {
        Step::Join | Step::FrontAwaitPeers => match doors.join() {
            Ok(reply) if reply.estate != seed.estate => Event::JoinRefusedForGood(format!("the installer fed `{}`, 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 = reply.feed.clone();
                match reply.role {
                    None => Event::Transient(format!("fed as `{}`; the election has not spoken yet", reply.me.name)),
                    Some(r) => Event::Joined { role: r, primary: reply.primary_utility.or(reply.primary_overlay), peers_admitted: reply.peers.len() as u8, peers_expected: seed.peers_expected },
                }
            }
            Err(e) if FOR_GOOD.iter().any(|w| e.contains(w)) => Event::JoinRefusedForGood(e),
            Err(e) => Event::Transient(e),
        },
        Step::BringTunnelUp => match learned.overlay {
            // A WireGuard box dials the front it was FED; a feed that carried
            // none yet (a front laid after the box) is read 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 == Role::Front) => match doors.join() {
                Ok(reply) if reply.estate == seed.estate && reply.front().is_some() => {
                    learned.peers = reply.peers.clone();
                    Event::Transient("the installer now feeds the front's tunnel; raising it next".into())
                }
                Ok(_) => Event::Transient("the installer fed no front tunnel yet".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) => Event::TunnelUp,
                Ok(false) => Event::Transient(format!("the {} tunnel is not up yet", seed.tunnel.as_str())),
                Err(e) if FOR_GOOD.iter().any(|w| e.contains(w)) => Event::JoinRefusedForGood(e),
                Err(e) => Event::Transient(e),
            },
        },
        Step::PrimaryCheckAdmin => match doors.join().map(|r| r.feed).map(|f| {
            // The LATEST feed, not the copy the join kept: gunnar MEASURED
            // 2026-09-25 that the front's offers reach the primary after its
            // join and before its admin.
            learned.feed = f;
            doors.primary_ready(&learned.feed)
        }) {
            Ok(Ok(true)) => Event::AdminPresent,
            Ok(Ok(false)) => Event::Transient("the primary is not ready yet (the product's own step)".into()),
            Ok(Err(e)) | Err(e) => Event::Transient(e),
        },
        Step::TwinBecome { primary } => match doors.become_twin(*primary) {
            Ok(()) => Event::TwinConferred,
            Err(e) => Event::Transient(e),
        },
        Step::TwinCertify { primary } => match doors.certified(*primary) {
            Ok(true) => Event::TwinCertified,
            Ok(false) => Event::Transient("waiting for this twin's certificate (the front certifies it against the primary)".into()),
            Err(e) => Event::Transient(e),
        },
        Step::TwinFollow { .. } => match doors.follow_state() {
            Ok((applied, head)) => Event::Following { applied, head },
            Err(e) => Event::Transient(e),
        },
        Step::Nothing => return step,
    };
    m.apply(event);
    step
}

/// **The signal, written where [`crate::boxside::Installed`]'s `status` reads
/// it** — what the installer sees, and closes on when it says PAIRED.
pub fn report(root: &Path, seed: &Seed, name: &str, m: &Machine) -> Signal {
    let s = Signal::of(name, seed.tunnel, m);
    if let Some(text) = s.render() {
        let _ = write_atomic(root, SIGNAL_REL, text.as_bytes(), false);
    }
    s
}

/// **Run the loop until PAIRED or refused for good**, reporting every change
/// (`say` gets the signal line). `pause` is the backoff's sleep (the tests
/// pass a short one). Returns the machine where it rested.
pub fn run(root: &Path, seed: &Seed, doors: &dyn Doors, say: &dyn Fn(&str), pause: &dyn Fn(Duration)) -> Machine {
    let mut m = Machine::default();
    let mut learned = Learned::default();
    report(root, seed, "?", &m);
    loop {
        let before = m.clone();
        let step = turn(&mut m, &mut learned, seed, doors);
        let name = learned.name.clone().unwrap_or_else(|| "?".into());
        if m != before {
            say(report(root, seed, &name, &m).line().trim());
        }
        if step == Step::Nothing {
            say(&if m.paired() { format!("pairing: {name} PAIRED — the loop rests") } else { format!("pairing: {name} REFUSED for good — re-seed this box") });
            return m;
        }
        pause(Duration::from_secs(m.backoff_s()));
    }
}

/// **Start the loop on its own thread.** `make` builds the doors ON that
/// thread (a tunnel driver is the loop's alone, and need not be `Send`), and
/// the thread keeps them — the tunnel with them — for the life of the process
/// once the loop rests: dropping the driver would take wg0 down.
pub fn start<D, F>(root: PathBuf, seed: Seed, make: F, say: Arc<dyn Fn(&str) + Send + Sync>) -> std::io::Result<()>
where
    D: Doors + 'static,
    F: FnOnce() -> D + Send + 'static,
{
    start_again_after(root, seed, make, say, AGAIN_AFTER_PANIC)
}

/// The duty of the pairing loop (A28 M18).
pub const DUTY: &str = "box-pairing";
/// How long a loop that panicked waits before it runs again.
pub const AGAIN_AFTER_PANIC: Duration = Duration::from_secs(30);

/// [`start`], a loop that panicked run again after `again` (a test's seam).
pub fn start_again_after<D, F>(root: PathBuf, seed: Seed, make: F, say: Arc<dyn Fn(&str) + Send + Sync>, again: Duration) -> std::io::Result<()>
where
    D: Doors + 'static,
    F: FnOnce() -> D + Send + 'static,
{
    // A duty (A28 M18): the doors are built on its thread and kept there, the tunnel with them;
    // a loop or a rest that panics is said and runs again after `again`.
    crate::duty::start_with(DUTY, again, Arc::clone(&say), move || {
        let doors = make();
        let (mut ran, mut rested) = (false, false);
        move || {
            if !ran {
                let _ = run(&root, &seed, &doors, &|l| say(l), &std::thread::sleep);
                ran = true;
            }
            if !rested {
                doors.rest(&|l| say(l));
                rested = true;
            }
            crate::duty::Round::Again
        }
    })?;
    Ok(())
}

/// Builds the seed's tunnel driver from the seed and the latest feed (the
/// credentials a mullvad driver needs ride the feed).
pub type TunnelFactory = Box<dyn Fn(&Seed, &BTreeMap<String, String>) -> Result<Box<dyn Tunnel>, TunnelError> + Send>;

/// **The real doors of a product box**: the join is what the installer fed
/// ([`Installed::latest`]), the tunnel is the seed's driver (built at the
/// tunnel step), the rest is the product's [`Steps`].
pub struct BoxDoors<P: Product, S: Steps> {
    installed: Arc<Installed<P>>,
    steps: S,
    make_tunnel: TunnelFactory,
    tunnel: Mutex<Option<Result<Box<dyn Tunnel>, TunnelError>>>,
    /// What the box's feed bus says the front is (0.5.17): every kept feed and re-feed whose
    /// front, address or PSK differ, for [`Doors::rest`] to follow.
    fronts: Mutex<std::sync::mpsc::Receiver<JoinReply>>,
    /// What the raised tunnel dials now (0.5.17): a re-feed naming the same is no change.
    dialled: Mutex<Option<Dial>>,
}

/// This box's address, the front it dials, the link's PSK.
type Dial = (Ipv4Addr, Option<Peer>, Option<String>);

/// The PSK of this box's link, as the installer fed it ([`crate::installer::FEED_WG_PSK`]),
/// trimmed. A fed but BLANK key stays `Some("")` (lane D14, decision c1) and is refused.
fn fed_psk(reply: &JoinReply) -> Option<String> {
    reply.feed.get(crate::installer::FEED_WG_PSK).map(|p| p.trim().to_string())
}

/// A fed but blank PSK refuses the tunnel by name, whatever the driver (lane D14, decision c1).
fn refuse_blank(psk: Option<&str>) -> Result<(), String> {
    match psk {
        Some("") => Err("REFUSED [wg-psk-blank] the installer fed this box a BLANK WireGuard pre-shared key; its tunnel is never raised without its key".into()),
        _ => Ok(()),
    }
}

/// The front a dial names, for a log line: `<key, 8 characters>@<endpoint>` (Gunnar's
/// `peers_said`), `-` for none. The PSK is never said.
fn dial_said(d: Option<&Dial>) -> String {
    match d.and_then(|d| d.1.as_ref()) {
        Some(f) => format!("{}@{}", f.tunnel_pubkey.trim().chars().take(8).collect::<String>(), f.endpoint.as_deref().unwrap_or("-")),
        None => "-".into(),
    }
}

/// What the tunnel dials, as a reply names it: this box's address, the front, the PSK.
fn dial_of(reply: &JoinReply) -> Dial {
    (reply.me.overlay, reply.front().cloned(), fed_psk(reply))
}

/// How often the rested loop looks for the first handshake after a re-point (0.5.18).
#[cfg(not(test))]
pub const WATCH_EVERY: Duration = Duration::from_secs(3);
#[cfg(test)]
pub const WATCH_EVERY: Duration = Duration::from_millis(20);
/// How many looks before the rested loop says that nobody answered (0.5.18).
pub const WATCH_ATTEMPTS: u32 = 20;

impl<P: Product, S: Steps> BoxDoors<P, S> {
    pub fn new(installed: Arc<Installed<P>>, steps: S, make_tunnel: TunnelFactory) -> BoxDoors<P, S> {
        let (tx, rx) = std::sync::mpsc::channel();
        let tx = Mutex::new(tx);
        let last: Mutex<Option<Dial>> = Mutex::new(None);
        installed.feed_bus().on_published("boxloop-follow-the-front", move |r: &JoinReply| {
            let now = dial_of(r);
            let mut seen = last.lock().unwrap_or_else(|p| p.into_inner());
            if seen.as_ref() != Some(&now) {
                *seen = Some(now);
                let _ = tx.lock().unwrap_or_else(|p| p.into_inner()).send(r.clone());
            }
        });
        BoxDoors { installed, steps, make_tunnel, tunnel: Mutex::new(None), fronts: Mutex::new(rx), dialled: Mutex::new(None) }
    }

    /// **Follow the front a re-feed names** (0.5.17, A7 row 7; Gunnar 907f4f99
    /// `Installed::follow_the_front`): a FRONT recharged onto new addresses, with a new relay
    /// key, re-feeds main and twin; the loop's own driver is re-pointed at the relay as the
    /// re-feed names it ([`Tunnel::repoint`], with the re-feed's PSK), at once — before, a
    /// closed box kept the dead relay until its next start, and the recharged front could not
    /// reach it. `Ok(Some(words))` when the dial changed; `Ok(None)` when it did not, or when
    /// no tunnel is raised yet (the tunnel step takes the latest feed) or the estate runs none.
    pub fn follow_the_front(&self, reply: &JoinReply) -> Result<Option<String>, String> {
        if self.installed.seed().tunnel == TunnelKind::None || reply.estate != self.installed.seed().estate {
            return Ok(None);
        }
        let Some(front) = reply.front() else { return Ok(None) };
        let mut slot = self.tunnel.lock().map_err(|e| e.to_string())?;
        let Some(Ok(driver)) = slot.as_mut() else { return Ok(None) };
        let mut dialled = self.dialled.lock().map_err(|e| e.to_string())?;
        let now = dial_of(reply);
        if dialled.is_none() || dialled.as_ref() == Some(&now) {
            // No tunnel raised yet (the tunnel step takes the latest feed), or the same dial.
            return Ok(None);
        }
        refuse_blank(now.2.as_deref())?;
        let psk = driver.set_preshared_key(now.2.as_deref()).map_err(|e| e.to_string())?;
        let me = Identity { overlay: reply.me.overlay, pubkey: driver.public_key() };
        let moved = driver.repoint(&me, &reply.peers).map_err(|e| e.to_string())?;
        let from = dial_said(dialled.as_ref());
        let to = dial_said(Some(&now));
        *dialled = Some(now);
        Ok((moved || psk).then(|| {
            format!(
                "pairing: re-fed — {} RE-POINTED from [{from}] to [{to}]: it dials the front `{}` at {} as the re-feed names it, without a restart",
                self.installed.seed().tunnel.as_str(),
                front.name,
                front.endpoint.as_deref().unwrap_or("-")
            )
        }))
    }

    /// The tunnel's status, once a driver is built.
    pub fn tunnel_status(&self) -> Option<crate::tunnel::TunnelStatus> {
        match self.tunnel.lock().ok()?.as_ref()? {
            Ok(t) => Some(t.status()),
            Err(_) => None,
        }
    }
}

impl<P: Product, S: Steps> Doors for BoxDoors<P, S> {
    fn join(&self) -> Result<JoinReply, String> {
        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())
    }

    fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
        let mut slot = self.tunnel.lock().map_err(|e| e.to_string())?;
        if slot.is_none() {
            let feed = self.installed.latest().map(|r| r.feed).unwrap_or_default();
            *slot = Some((self.make_tunnel)(self.installed.seed(), &feed));
        }
        let driver = slot.as_mut().expect("built above").as_mut().map_err(|e| e.to_string())?;
        // The link's PSK, as fed (0.5.17, A7 row 13): set before the raise.
        let psk = self.installed.latest().as_ref().and_then(fed_psk);
        refuse_blank(psk.as_deref())?;
        driver.set_preshared_key(psk.as_deref()).map_err(|e| e.to_string())?;
        let me = Identity { overlay, pubkey: driver.public_key() };
        driver.up(&me, peers).map_err(|e| e.to_string())?;
        if let Ok(mut d) = self.dialled.lock() {
            *d = Some((overlay, peers.iter().find(|p| p.role == Role::Front).cloned(), psk));
        }
        Ok(driver.status().up)
    }

    fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
        self.steps.primary_ready(feed)
    }

    fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
        self.steps.become_twin(primary)
    }

    fn certified(&self, primary: Ipv4Addr) -> Result<bool, String> {
        self.steps.certified(primary)
    }

    fn follow_state(&self) -> Result<(u64, u64), String> {
        self.steps.follow_state()
    }

    /// The rested loop follows the front every re-feed names, for the life of the process,
    /// and watches a re-pointed tunnel for its first handshake (0.5.18, Gunnar's
    /// `handshake_watch`): said ONCE, that the front answered or that nobody did.
    fn rest(&self, say: &dyn Fn(&str)) {
        use std::sync::mpsc::RecvTimeoutError;
        // After a re-point: the front's name and the looks left.
        let mut watch: Option<(String, u32)> = None;
        loop {
            let next = {
                let rx = self.fronts.lock().unwrap_or_else(|p| p.into_inner());
                if watch.is_some() { rx.recv_timeout(WATCH_EVERY) } else { rx.recv().map_err(|_| RecvTimeoutError::Disconnected) }
            };
            match next {
                Ok(reply) => match self.follow_the_front(&reply) {
                    Ok(Some(words)) => {
                        say(&words);
                        watch = reply.front().map(|f| (f.name.clone(), WATCH_ATTEMPTS));
                    }
                    Ok(None) => {}
                    Err(e) => say(&format!("pairing: re-fed — the tunnel could not follow the front now ({e}); the next start raises it from the feed")),
                },
                Err(RecvTimeoutError::Timeout) => {
                    let Some((front, left)) = watch.take() else { continue };
                    let reached = self.tunnel_status().is_some_and(|s| s.reachable.contains(&front));
                    if reached {
                        say(&format!("pairing: re-pointed — WireGuard HANDSHAKE COMPLETED with the front `{front}`: this box is reachable over the tunnel again"));
                    } else if left <= 1 {
                        say(&format!(
                            "pairing: re-pointed — NO handshake with the front `{front}` within {:?}: the tunnel dials it and nothing answers; check the front's relay endpoint, its key, and that UDP reaches it",
                            WATCH_EVERY * WATCH_ATTEMPTS
                        ));
                    } else {
                        watch = Some((front, left - 1));
                    }
                }
                Err(RecvTimeoutError::Disconnected) => break,
            }
        }
        loop {
            std::thread::park();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::join::Assigned;
    use crate::pair::State;
    use std::cell::RefCell;

    fn seed() -> Seed {
        Seed {
            estate: "holger.test".into(),
            tunnel: TunnelKind::Wireguard,
            front: Default::default(),
            kind: crate::seed::Kind::Holger,
            installer_ca_pem: String::new(),
            bootstrap_cert_pem: String::new(),
            bootstrap_key_pem: String::new(),
            peers_expected: 3,
            mullvad: None,
            tailscale: None,
            installer_link: None,
        }
    }

    fn front() -> Peer {
        Peer { name: "front".into(), role: Role::Front, tunnel_pubkey: "F".into(), overlay: crate::OVERLAY_FRONT, endpoint: Some("10.13.8.1:51820".into()), utility: None }
    }

    fn reply(role: Option<Role>, with_front: bool) -> JoinReply {
        JoinReply {
            estate: "holger.test".into(),
            me: Assigned { name: if role == Some(Role::Twin) { "vm3".into() } else { "vm2".into() }, overlay: crate::OVERLAY_BOX_A },
            cert_pem: None,
            role,
            primary_overlay: Some(crate::OVERLAY_BOX_A),
            primary_utility: Some(Ipv4Addr::new(10, 13, 8, 2)),
            peers: if with_front { vec![front()] } else { Vec::new() },
            feed: BTreeMap::from([("k".to_string(), "v".to_string())]),
        }
    }

    /// Fake doors: the feeds in order (the last one repeats), and a record of
    /// every door asked.
    struct Fake {
        feeds: RefCell<Vec<Result<JoinReply, String>>>,
        asked: RefCell<Vec<String>>,
        applied: RefCell<u64>,
    }

    impl Fake {
        fn new(feeds: Vec<Result<JoinReply, String>>) -> Fake {
            Fake { feeds: RefCell::new(feeds), asked: RefCell::new(Vec::new()), applied: RefCell::new(0) }
        }
    }

    impl Doors for Fake {
        fn join(&self) -> Result<JoinReply, String> {
            self.asked.borrow_mut().push("join".into());
            let mut f = self.feeds.borrow_mut();
            if f.len() > 1 {
                f.remove(0)
            } else {
                f[0].clone()
            }
        }
        fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
            self.asked.borrow_mut().push(format!("tunnel {overlay} {}", peers.len()));
            Ok(true)
        }
        fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
            self.asked.borrow_mut().push(format!("primary {}", feed.len()));
            Ok(true)
        }
        fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
            self.asked.borrow_mut().push(format!("become {primary}"));
            Ok(())
        }
        fn certified(&self, _: Ipv4Addr) -> Result<bool, String> {
            self.asked.borrow_mut().push("certified".into());
            *self.applied.borrow_mut() += 1;
            Ok(*self.applied.borrow() > 1)
        }
        fn follow_state(&self) -> Result<(u64, u64), String> {
            Ok((*self.applied.borrow(), 2))
        }
    }

    /// Doors whose first join PANICS, then pair as a primary; `rest` says it rested.
    struct Breaks {
        fake: Fake,
        joins: std::sync::atomic::AtomicU32,
        rested: Arc<std::sync::atomic::AtomicBool>,
    }

    impl Doors for Breaks {
        fn join(&self) -> Result<JoinReply, String> {
            if self.joins.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
                panic!("the first join breaks");
            }
            self.fake.join()
        }
        fn tunnel_up(&self, overlay: Ipv4Addr, peers: &[Peer]) -> Result<bool, String> {
            self.fake.tunnel_up(overlay, peers)
        }
        fn primary_ready(&self, feed: &BTreeMap<String, String>) -> Result<bool, String> {
            self.fake.primary_ready(feed)
        }
        fn become_twin(&self, primary: Ipv4Addr) -> Result<(), String> {
            self.fake.become_twin(primary)
        }
        fn certified(&self, primary: Ipv4Addr) -> Result<bool, String> {
            self.fake.certified(primary)
        }
        fn follow_state(&self) -> Result<(u64, u64), String> {
            self.fake.follow_state()
        }
        fn rest(&self, _say: &dyn Fn(&str)) {
            self.rested.store(true, std::sync::atomic::Ordering::SeqCst);
            loop {
                std::thread::park();
            }
        }
    }

    /// ★ **A pairing loop that panics is said and runs again** (A28 M18): the raw thread ended
    /// at the panic, its tunnel with it, and the box never paired, without a word.
    #[test]
    fn a_pairing_loop_that_panics_is_said_and_runs_again() {
        let _one = crate::duty::tests::ONE.lock().unwrap_or_else(|p| p.into_inner());
        let root = std::env::temp_dir().join(format!("np-boxloop-{}", std::process::id()));
        std::fs::create_dir_all(&root).unwrap();
        let rested = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let lines = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
        let (r, l) = (Arc::clone(&rested), Arc::clone(&lines));
        let say: Arc<dyn Fn(&str) + Send + Sync> = Arc::new(move |s: &str| l.lock().unwrap().push(s.to_string()));
        start_again_after(root.clone(), seed(), move || Breaks { fake: Fake::new(vec![Ok(reply(Some(Role::Primary), true))]), joins: Default::default(), rested: r }, say, Duration::from_millis(10)).unwrap();
        let t = std::time::Instant::now();
        while !rested.load(std::sync::atomic::Ordering::SeqCst) && t.elapsed() < Duration::from_secs(10) {
            std::thread::sleep(Duration::from_millis(5));
        }
        assert!(rested.load(std::sync::atomic::Ordering::SeqCst), "the loop ran again and rests: {:?}", lines.lock().unwrap());
        assert!(lines.lock().unwrap().iter().any(|l| l.contains("ALERT box-pairing") && l.contains("the first join breaks")), "the panic is said: {:?}", lines.lock().unwrap());
        assert!(crate::duty::find(DUTY).is_some(), "the loop is a listed duty");
        let _ = std::fs::remove_dir_all(&root);
    }

    fn walk(doors: &Fake, max: usize) -> (Machine, Vec<&'static str>) {
        let (mut m, mut l, s) = (Machine::default(), Learned::default(), seed());
        let mut steps = Vec::new();
        for _ in 0..max {
            let step = turn(&mut m, &mut l, &s, doors);
            steps.push(step.name());
            if step == Step::Nothing {
                break;
            }
        }
        (m, steps)
    }

    #[test]
    fn a_primary_waits_for_the_feed_raises_its_tunnel_on_the_fed_address_and_is_paired() {
        let doors = Fake::new(vec![Err("waiting for the installer".into()), Ok(reply(None, true)), Ok(reply(Some(Role::Primary), true))]);
        let (m, steps) = walk(&doors, 20);
        assert_eq!(m.state, State::Paired { role: Role::Primary });
        assert_eq!(steps, ["join", "join", "join", "tunnel-up", "check-admin", "nothing"], "no feed and no role are both transients");
        assert!(doors.asked.borrow().contains(&format!("tunnel {} 1", crate::OVERLAY_BOX_A)), "{:?}", doors.asked.borrow());
    }

    #[test]
    fn a_twin_becomes_the_twin_of_the_primarys_utility_address_is_certified_and_follows() {
        let doors = Fake::new(vec![Ok(reply(Some(Role::Twin), true))]);
        let (m, steps) = walk(&doors, 20);
        assert_eq!(m.state, State::Paired { role: Role::Twin });
        assert_eq!(steps, ["join", "tunnel-up", "become-twin", "certify", "certify", "follow", "nothing"]);
        assert!(doors.asked.borrow().contains(&"become 10.13.8.2".to_string()), "the utility address wins: {:?}", doors.asked.borrow());
    }

    #[test]
    fn a_wireguard_feed_without_the_front_is_read_again_before_the_tunnel_and_another_estate_is_refused() {
        let doors = Fake::new(vec![Ok(reply(Some(Role::Primary), false)), Ok(reply(Some(Role::Primary), false)), Ok(reply(Some(Role::Primary), true))]);
        let (m, steps) = walk(&doors, 20);
        assert!(m.paired());
        assert_eq!(&steps[..4], ["join", "tunnel-up", "tunnel-up", "tunnel-up"], "no front: the feed is read again, never a tunnel to nobody");
        let other = Fake::new(vec![Ok(JoinReply { estate: "other".into(), ..reply(Some(Role::Primary), true) })]);
        let (m, _) = walk(&other, 5);
        assert!(matches!(m.state, State::Refused { .. }));
        let unbuilt = Fake::new(vec![Err("REFUSED [tunnel-not-built] …".into())]);
        assert!(matches!(walk(&unbuilt, 5).0.state, State::Refused { .. }), "a medium without its driver is refused for good");
    }

    #[test]
    fn run_writes_the_signal_the_installer_reads_and_rests_paired() {
        let root = std::env::temp_dir().join(format!("boxloop-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let doors = Fake::new(vec![Ok(reply(Some(Role::Primary), true))]);
        let said = RefCell::new(Vec::new());
        let m = run(&root, &seed(), &doors, &|l| said.borrow_mut().push(l.to_string()), &|_| {});
        assert!(m.paired());
        let s = Signal::parse(&std::fs::read_to_string(root.join(SIGNAL_REL)).unwrap()).unwrap();
        assert!(s.paired && s.name == "vm2" && s.role == Some(Role::Primary), "{s:?}");
        assert!(said.borrow().iter().any(|l| l.ends_with("PAIRED")), "{:?}", said.borrow());
        let _ = std::fs::remove_dir_all(&root);
    }

    /// A tunnel driver that only says what it was asked (the real one needs CAP_NET_ADMIN).
    struct Said(Arc<Mutex<Vec<String>>>);
    impl Tunnel for Said {
        fn kind(&self) -> TunnelKind {
            TunnelKind::Wireguard
        }
        fn public_key(&self) -> String {
            "KB".into()
        }
        fn up(&mut self, me: &Identity, peers: &[Peer]) -> Result<(), TunnelError> {
            let front = peers.iter().find(|p| p.role == Role::Front).and_then(|p| p.endpoint.clone()).unwrap_or_default();
            self.0.lock().unwrap().push(format!("up {} {front}", me.overlay));
            Ok(())
        }
        fn repoint(&mut self, me: &Identity, peers: &[Peer]) -> Result<bool, TunnelError> {
            let front = peers.iter().find(|p| p.role == Role::Front).and_then(|p| p.endpoint.clone()).unwrap_or_default();
            self.0.lock().unwrap().push(format!("repoint {} {front}", me.overlay));
            Ok(true)
        }
        fn set_preshared_key(&mut self, psk: Option<&str>) -> Result<bool, TunnelError> {
            self.0.lock().unwrap().push(format!("psk {}", psk.unwrap_or("-")));
            Ok(psk.is_some())
        }
        fn add_peer(&mut self, _: &Peer) -> Result<(), TunnelError> {
            Ok(())
        }
        /// The front is reachable once the test logs `handshake` (a real pump's first handshake).
        fn status(&self) -> crate::tunnel::TunnelStatus {
            let reachable = if self.0.lock().unwrap().iter().any(|l| l == "handshake") { vec!["front".to_string()] } else { Vec::new() };
            crate::tunnel::TunnelStatus { kind: TunnelKind::Wireguard, up: true, reachable, detail: String::new() }
        }
    }

    struct Box2;
    impl Product for Box2 {
        fn kind(&self) -> crate::seed::Kind {
            crate::seed::Kind::Holger
        }
        fn tunnel_pubkey(&self) -> String {
            "KB".into()
        }
        fn capacity(&self) -> Result<Option<crate::elect::Capacity>, String> {
            Ok(None)
        }
        fn offer(&self, _: Option<&JoinReply>) -> BTreeMap<String, String> {
            BTreeMap::new()
        }
        fn fed(&self, _: &JoinReply) {}
    }

    struct NoSteps;
    impl Steps for NoSteps {
        fn primary_ready(&self, _: &BTreeMap<String, String>) -> Result<bool, String> {
            Ok(true)
        }
        fn become_twin(&self, _: Ipv4Addr) -> Result<(), String> {
            Ok(())
        }
        fn follow_state(&self) -> Result<(u64, u64), String> {
            Ok((1, 1))
        }
    }

    const PSK: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";

    fn fed_box(tag: &str, front_at: &str) -> (Arc<Installed<Box2>>, JoinReply, PathBuf) {
        let root = std::env::temp_dir().join(format!("boxloop-{tag}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let installed = Arc::new(Installed::open(&root, seed(), Box2).unwrap());
        let mut r = reply(Some(Role::Primary), true);
        r.peers[0].endpoint = Some(front_at.into());
        r.feed.insert(crate::installer::FEED_WG_PSK.into(), PSK.into());
        use crate::bootstrap::Vm as _;
        installed.fed(&r).unwrap();
        installed.close().unwrap();
        (installed, r, root)
    }

    fn said_doors(installed: &Arc<Installed<Box2>>, log: &Arc<Mutex<Vec<String>>>) -> BoxDoors<Box2, NoSteps> {
        let l = Arc::clone(log);
        BoxDoors::new(Arc::clone(installed), NoSteps, Box::new(move |_, _| Ok(Box::new(Said(Arc::clone(&l))) as Box<dyn Tunnel>)))
    }

    /// ★ 0.5.17 (A7 row 13, Gunnar 356e8dd2): the box's link to the front carries the PSK the
    /// installer fed it (`wg_psk`), set on the driver before the tunnel is raised.
    #[test]
    fn the_box_raises_its_link_with_the_psk_the_installer_fed() {
        let (installed, r, root) = fed_box("psk", "10.13.8.1:51820");
        let log = Arc::new(Mutex::new(Vec::new()));
        let doors = said_doors(&installed, &log);
        assert!(doors.tunnel_up(r.me.overlay, &r.peers).unwrap());
        assert_eq!(*log.lock().unwrap(), vec![format!("psk {PSK}"), format!("up {} 10.13.8.1:51820", r.me.overlay)]);
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ Lane D14, decision c1: a `wg_psk` that was fed BLANK refuses the box's tunnel by name;
    /// it is never raised without a key (a silent downgrade on a security gateway).
    #[test]
    fn a_blank_fed_psk_refuses_the_boxs_tunnel_instead_of_raising_it_without_one() {
        let (installed, mut r, root) = fed_box("psk-blank", "10.13.8.1:51820");
        r.feed.insert(crate::installer::FEED_WG_PSK.into(), "  ".into());
        use crate::bootstrap::Vm as _;
        let _ = installed.fed(&r);
        let log = Arc::new(Mutex::new(Vec::new()));
        let doors = said_doors(&installed, &log);
        let e = doors.tunnel_up(r.me.overlay, &r.peers).unwrap_err();
        assert!(e.contains("[wg-psk-blank]"), "{e}");
        assert!(!log.lock().unwrap().iter().any(|l| l.starts_with("up ")), "never raised: {:?}", log.lock().unwrap());
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ 0.5.17 (A7 row 7, Gunnar 907f4f99): a closed box RE-FED after a FRONT recharge (the
    /// front on a new address with a new relay key) re-points its tunnel at the new relay at
    /// once, without a restart: through the loop's own driver, from the rested loop.
    #[test]
    fn a_closed_box_refed_after_a_front_recharge_dials_the_new_relay() {
        let (installed, r, root) = fed_box("refeed", "127.86.33.51:51821");
        let log = Arc::new(Mutex::new(Vec::new()));
        let (i, l) = (Arc::clone(&installed), Arc::clone(&log));
        let overlay = r.me.overlay;
        std::thread::spawn(move || {
            let doors = said_doors(&i, &l);
            doors.tunnel_up(overlay, &i.latest().unwrap().peers).unwrap();
            doors.rest(&|_| {});
        });
        let wait_for = |want: &str| {
            let deadline = std::time::Instant::now() + Duration::from_secs(10);
            while std::time::Instant::now() < deadline && !log.lock().unwrap().iter().any(|l| l == want) {
                std::thread::sleep(Duration::from_millis(20));
            }
            log.lock().unwrap().clone()
        };
        let up = wait_for(&format!("up {overlay} 127.86.33.51:51821"));
        assert!(up.contains(&format!("up {overlay} 127.86.33.51:51821")), "{up:?}");
        // The front is recharged onto a new address with a new relay key.
        let mut moved = r.clone();
        moved.peers[0].endpoint = Some("127.86.33.10:51821".into());
        moved.peers[0].tunnel_pubkey = "F2".into();
        use crate::bootstrap::Vm as _;
        installed.refed(&moved).expect("the re-feed is taken");
        let after = wait_for(&format!("repoint {overlay} 127.86.33.10:51821"));
        assert!(after.contains(&format!("repoint {overlay} 127.86.33.10:51821")), "the box still dials the dead relay: {after:?}");
        let _ = std::fs::remove_dir_all(&root);
    }

    /// 0.5.17: a re-feed that names the SAME front changes nothing; one before any tunnel is
    /// raised is the tunnel step's to take.
    #[test]
    fn a_refeed_naming_the_same_front_or_before_the_tunnel_repoints_nothing() {
        let (installed, r, root) = fed_box("same", "10.13.8.1:51820");
        let log = Arc::new(Mutex::new(Vec::new()));
        let doors = said_doors(&installed, &log);
        assert_eq!(doors.follow_the_front(&r).unwrap(), None, "no tunnel yet: nothing to re-point");
        doors.tunnel_up(r.me.overlay, &r.peers).unwrap();
        assert_eq!(doors.follow_the_front(&r).unwrap(), None, "the same front: nothing to do");
        let mut moved = r.clone();
        moved.peers[0].endpoint = Some("10.13.8.9:51820".into());
        assert!(doors.follow_the_front(&moved).unwrap().is_some_and(|w| w.contains("10.13.8.9:51820")));
        assert!(log.lock().unwrap().iter().any(|l| l.starts_with("repoint ")));
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ 0.5.18 (lane G7, Gunnar 7e33c0f7 `wg::raise_from_config`): the re-point SAYS the old
    /// and the new peer, `<key>@<endpoint>` each, as Gunnar's `WireGuard RE-POINTED` line does.
    #[test]
    fn a_repoint_says_the_old_and_the_new_peer() {
        let (installed, r, root) = fed_box("said-peers", "10.13.8.1:51820");
        let log = Arc::new(Mutex::new(Vec::new()));
        let doors = said_doors(&installed, &log);
        doors.tunnel_up(r.me.overlay, &r.peers).unwrap();
        let mut moved = r.clone();
        moved.peers[0].endpoint = Some("10.13.8.9:51820".into());
        moved.peers[0].tunnel_pubkey = "F2NEWRELAYKEY".into();
        let words = doors.follow_the_front(&moved).unwrap().expect("re-pointed");
        assert!(words.contains("RE-POINTED") && words.contains("from [F@10.13.8.1:51820] to [F2NEWREL@10.13.8.9:51820]"), "{words}");
        let _ = std::fs::remove_dir_all(&root);
    }

    type Lines = Arc<Mutex<Vec<String>>>;

    fn rested(tag: &str) -> (Arc<Installed<Box2>>, JoinReply, PathBuf, Lines, Lines) {
        let (installed, r, root) = fed_box(tag, "127.86.33.51:51821");
        let log = Arc::new(Mutex::new(Vec::new()));
        let said = Arc::new(Mutex::new(Vec::new()));
        let (i, l, s) = (Arc::clone(&installed), Arc::clone(&log), Arc::clone(&said));
        let overlay = r.me.overlay;
        std::thread::spawn(move || {
            let doors = said_doors(&i, &l);
            doors.tunnel_up(overlay, &i.latest().unwrap().peers).unwrap();
            doors.rest(&|line| s.lock().unwrap().push(line.to_string()));
        });
        (installed, r, root, log, said)
    }

    fn wait_said(said: &Arc<Mutex<Vec<String>>>, want: &str) -> Vec<String> {
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while std::time::Instant::now() < deadline && !said.lock().unwrap().iter().any(|l| l.contains(want)) {
            std::thread::sleep(Duration::from_millis(10));
        }
        said.lock().unwrap().clone()
    }

    fn recharge(installed: &Arc<Installed<Box2>>, r: &JoinReply) {
        let mut moved = r.clone();
        moved.peers[0].endpoint = Some("127.86.33.10:51821".into());
        moved.peers[0].tunnel_pubkey = "F2".into();
        use crate::bootstrap::Vm as _;
        installed.refed(&moved).expect("the re-feed is taken");
    }

    /// ★ 0.5.18 (lane G7, Gunnar 7e33c0f7 `wg::handshake_watch` after a re-point): the rested
    /// loop watches the re-pointed tunnel and says ONCE when the front answered.
    #[test]
    fn the_rested_loop_says_the_handshake_after_a_repoint() {
        let (installed, r, root, log, said) = rested("watch-ok");
        while !log.lock().unwrap().iter().any(|l| l.starts_with("up ")) {
            std::thread::sleep(Duration::from_millis(5));
        }
        recharge(&installed, &r);
        assert!(wait_said(&said, "RE-POINTED").iter().any(|l| l.contains("RE-POINTED")));
        log.lock().unwrap().push("handshake".into());
        let heard = wait_said(&said, "HANDSHAKE COMPLETED");
        assert!(heard.iter().any(|l| l.contains("HANDSHAKE COMPLETED") && l.contains("`front`")), "{heard:?}");
        std::thread::sleep(WATCH_EVERY * 3);
        assert_eq!(said.lock().unwrap().iter().filter(|l| l.contains("HANDSHAKE")).count(), 1, "said once");
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ 0.5.18 (lane G7): a re-pointed tunnel nobody answers is said ONCE, with a verdict,
    /// after the watch's budget — never a line per poll, never silence.
    #[test]
    fn the_rested_loop_says_no_handshake_once_after_its_budget() {
        let (installed, r, root, log, said) = rested("watch-none");
        while !log.lock().unwrap().iter().any(|l| l.starts_with("up ")) {
            std::thread::sleep(Duration::from_millis(5));
        }
        recharge(&installed, &r);
        let heard = wait_said(&said, "NO handshake");
        assert!(heard.iter().any(|l| l.contains("NO handshake") && l.contains("`front`")), "{heard:?}");
        std::thread::sleep(WATCH_EVERY * 3);
        assert_eq!(said.lock().unwrap().iter().filter(|l| l.contains("handshake")).count(), 1, "said once: {:?}", said.lock().unwrap());
        let _ = std::fs::remove_dir_all(&root);
    }
}