nordisk-pairing 0.5.25

//! **A product box's side of the installer, shared** (0.5.4, 2026-09-27).
//!
//! [`crate::bootstrap`] is the wire; what a BOX keeps between the installer's
//! visits was gunnar-server's own code (`installer_listener.rs`): its own key,
//! the certificate the installer signed for it, what it was fed
//! (`joined.toml`), the closed marker, the fresh-install reset, and the thread
//! that raises the installer's link and serves the listener. holger follows
//! gunnar's shape now (owner 2026-09-27: vm1 front, vm2 primary, vm3 twin over
//! WireGuard + mTLS), so that part lives here once. What stays the product's
//! is what only it knows — its kind, tunnel key, capacity, what it offers, and
//! what it does when fed — the [`Product`] half.
//!
//! Every path is relative to the box's data root, and every write is atomic
//! (temporary file, fsync, rename, fsync of the directory — edda appliance-pid1 `durable`):
//! a box that loses power mid-feed keeps the old feed or the new one, never half of either.
//! What it was fed (`joined.toml`) is SEALED at rest ([`crate::fed_at_rest`], 0.5.17).

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

/// The feed bus, where a box publishes what it keeps (0.5.15).
pub use crate::feedbus;

use crate::bootstrap::Vm;
use crate::join::{JoinReply, JoinRequest};
use crate::mtls::{self, Issued};
use crate::seed::{Kind, Seed};
use crate::signal::Signal;

/// What the installer fed last.
pub const JOINED_REL: &str = "pairing/joined.toml";
/// This box's own certificate, signed by the installer CA.
pub const OWN_CERT_REL: &str = "pairing/own.pem";
/// This box's own key (minted once, never leaves the box).
pub const OWN_KEY_REL: &str = "pairing/own.key";
/// The pairing signal the product's loop writes.
pub const SIGNAL_REL: &str = "pairing/signal.toml";
/// The installer closed this box's listener on this installation.
pub const CLOSED_REL: &str = "pairing/closed";
/// The files a fresh installation must not carry.
pub const PAIRING_STATE: [&str; 5] = [JOINED_REL, OWN_CERT_REL, OWN_KEY_REL, SIGNAL_REL, CLOSED_REL];
/// The ESP marker a first boot writes once it has cleared the pairing state
/// (korp-installer lays a new ESP on every installation).
pub const ESP_CLEARED: &str = "korp/pairing.cleared";

/// **Write `rel` under `root` atomically**, 0600 for a secret: edda appliance-pid1's
/// `durable::write_atomic` (0.5.17, G4), the one writer a power cut leaves the old or the
/// new file of — staged under its own name, fsync, rename, fsync of the directory.
pub fn write_atomic(root: &Path, rel: &str, bytes: &[u8], secret: bool) -> std::io::Result<()> {
    appliance_pid1::durable::write_atomic(&root.join(rel), bytes, secret.then_some(0o600))
}

/// **Mark the ESP** (0.5.17, G4): the [`ESP_CLEARED`] marker (or any ESP file), its
/// directory made when missing, written whole and durably. A plain write could leave the
/// marker EMPTY after a power cut, and the next boot would read "already cleared" from a
/// file that never held a byte. The ESP is FAT: a rename, never a hard link. On the real
/// ESP (`/boot`, mounted read-only by korp-start) the write runs with the ESP flipped
/// read-write and straight back (lane D14, appliance-pid1 `with_esp_writable`).
pub fn mark_esp(path: &Path) -> bool {
    let mount = Path::new(appliance_pid1::stop_record::ESP_MOUNT);
    mark_esp_on(mount, path, &|m: &Path| appliance_pid1::stop_record::with_esp_writable(m, || write_marker_file(path)))
}

/// [`mark_esp`] over its seam: `writable` runs the write with `mount` writable. A marker outside
/// `mount` (a test's temporary ESP) is written as it is.
pub fn mark_esp_on(mount: &Path, path: &Path, writable: &dyn Fn(&Path) -> bool) -> bool {
    if path.starts_with(mount) {
        writable(mount)
    } else {
        write_marker_file(path)
    }
}

fn write_marker_file(path: &Path) -> bool {
    appliance_pid1::durable::write_atomic(path, b"the pairing state of the previous installation was cleared\n", None).is_ok()
}

/// **Clear the pairing state on the first boot after an installation** —
/// pure over two roots (`esp` is `/boot`, `data` the data root). The ESP
/// marker is written FIRST; only when it could be written is anything cleared
/// (a box that could not mark its ESP would clear its pairing on every boot).
/// `None` = not a first boot. MEASURED by gunnar 2026-09-24: a re-install onto
/// a kept data volume found its old `joined.toml` and never joined the new
/// estate.
pub fn reset_if_fresh_install(esp: &Path, data: &Path, state: &[&str], write_marker: &dyn Fn(&Path) -> bool) -> Option<Vec<String>> {
    if !esp.join("korp").is_dir() || esp.join(ESP_CLEARED).exists() {
        return None;
    }
    if !write_marker(&esp.join(ESP_CLEARED)) {
        tracing::error!("pairing: the first boot of this installation and the ESP marker could not be written — the pairing state is NOT cleared; the next boot asks again");
        return None;
    }
    Some(state.iter().filter(|rel| std::fs::remove_file(data.join(rel)).is_ok()).map(|rel| rel.to_string()).collect())
}

/// **A keep boot of the SAME estate keeps the pairing state** (0.5.17, A7 row 6; Gunnar
/// ff58dad1 `vm::first_boot`) — [`reset_if_fresh_install`], asked with the seed this boot
/// runs with. The shop keep-boots a box or a front from its release's keep medium: a new
/// ESP, so this IS a first boot, but the data volume is kept and the installer stages nothing
/// onto a kept volume, so the seed is the kept one. When the kept, READABLE feed names that
/// same estate (and is a front's reply on a front seed, a box's on a product seed), it is
/// still this box's: only the ESP is marked (`Some(vec![])`, decided once) and the feed, the
/// box's own key and its certificate stay — clearing them left a keep-booted front unfed for
/// ever and every keep fallback unpaired the box. A feed that cannot be read, or names another
/// estate, is cleared as before (MEASURED 2026-09-24).
pub fn reset_if_fresh_install_of(esp: &Path, data: &Path, state: &[&str], seed: &Seed, write_marker: &dyn Fn(&Path) -> bool) -> Option<Vec<String>> {
    if !esp.join("korp").is_dir() || esp.join(ESP_CLEARED).exists() {
        return None;
    }
    let kept = match crate::fed_at_rest::read_checked(data, JOINED_REL) {
        Ok(t) => t.and_then(|t| JoinReply::parse(&t).ok()),
        Err(e) => {
            tracing::warn!(file = JOINED_REL, error = %e, "pairing: the first boot of this installation — the kept feed cannot be read, so the pairing state is cleared");
            None
        }
    };
    let ours = kept.as_ref().is_some_and(|r| r.estate == seed.estate && ((seed.kind == Kind::Front) == (r.role == Some(crate::Role::Front))));
    if ours {
        // Marked or not, the state is kept: a box that could not mark its ESP asks again on
        // the next boot and finds the same feed.
        if !write_marker(&esp.join(ESP_CLEARED)) {
            tracing::error!("pairing: a keep boot of this estate and the ESP marker could not be written — the pairing state is KEPT");
        }
        tracing::info!(estate = %seed.estate, "pairing: a keep boot of the same estate onto its kept data — the installer's feed, this box's key and certificate are KEPT");
        return Some(Vec::new());
    }
    reset_if_fresh_install(esp, data, state, write_marker)
}

/// **What only the product knows.**
pub trait Product: Send + Sync {
    /// Its kind: `Gunnar` or `Holger`.
    fn kind(&self) -> Kind;
    /// Its tunnel public key (WireGuard base64), empty for `tunnel = none`.
    fn tunnel_pubkey(&self) -> String;
    /// What the election compares. `Err` = not measured yet, ask again.
    fn capacity(&self) -> Result<Option<crate::elect::Capacity>, String>;
    /// What it offers the installer on every visit (`offer.<vm>.<key>`).
    fn offer(&self, fed: Option<&JoinReply>) -> BTreeMap<String, String>;
    /// Called after a feed was KEPT (idempotent; apply live, never restart).
    fn fed(&self, reply: &JoinReply);
    /// Whether this medium may be the front. A product medium may not.
    fn may_be_front(&self) -> bool {
        false
    }
    /// **What this box runs** (0.5.14, live update, [`crate::live::RUNNING_PATH`]): a product
    /// whose box carries a live slot answers it; the default refuses by name.
    fn live_running(&self) -> Result<crate::live::Running, String> {
        Err("REFUSED [live-unsupported] this VM takes no live update".into())
    }
    /// **A new app for this box's other slot** (0.5.14, [`crate::live::STAGE_PATH`]).
    fn live_stage(&self, _stage: &crate::live::Stage) -> Result<String, String> {
        Err("REFUSED [live-unsupported] this VM takes no live update".into())
    }
    /// **A new app with what its release declares** (lane B5): the default takes an
    /// undeclared stage and refuses a declared one by name ([`crate::live::declared_unsupported`]).
    fn live_stage_declared(&self, stage: &crate::live::Stage, declared: &crate::live::Declared) -> Result<String, String> {
        if declared.is_empty() {
            self.live_stage(stage)
        } else {
            Err(crate::live::declared_unsupported())
        }
    }
}

/// **This box, as the installer sees it** — the shared half of a product's
/// [`Vm`].
pub struct Installed<P: Product> {
    root: PathBuf,
    seed: Seed,
    own_key: String,
    fed: Mutex<Option<JoinReply>>,
    /// Every kept reply, published (0.5.15): [`Installed::feed_bus`].
    bus: Arc<crate::feedbus::ReplyBus>,
    pub product: P,
}

impl<P: Product> Installed<P> {
    /// Opens (or mints) this box's own key and reads what it was fed — only a
    /// feed of THIS estate counts.
    pub fn open(root: &Path, seed: Seed, product: P) -> Result<Installed<P>, String> {
        let own_key = match std::fs::read_to_string(root.join(OWN_KEY_REL)) {
            Ok(k) if k.contains("PRIVATE KEY") => k,
            _ => {
                let (_, key) = mtls::box_request().map_err(|e| e.to_string())?;
                write_atomic(root, OWN_KEY_REL, key.as_bytes(), true).map_err(|e| format!("{}: {e}", root.join(OWN_KEY_REL).display()))?;
                key
            }
        };
        let fed = crate::fed_at_rest::read(root, JOINED_REL).and_then(|t| JoinReply::parse(&t).ok()).filter(|r| r.estate == seed.estate);
        let bus = crate::feedbus::Bus::new();
        if let Some(kept) = &fed {
            bus.publish(kept.clone());
        }
        Ok(Installed { root: root.to_path_buf(), seed, own_key, fed: Mutex::new(fed), bus, product })
    }

    /// **The feed bus of this box** (0.5.15, [`crate::feedbus`]): the reply kept at open,
    /// then every feed and re-feed the box keeps, each NEW or CHANGED one handed to every
    /// subscriber. A part of the product that applies a fed value subscribes here
    /// ([`crate::feedbus::Bus::on_change`] for one key) and takes a late value live.
    pub fn feed_bus(&self) -> &Arc<crate::feedbus::ReplyBus> {
        &self.bus
    }

    /// What the installer fed last.
    pub fn latest(&self) -> Option<JoinReply> {
        self.fed.lock().ok().and_then(|f| f.clone())
    }

    pub fn closed(&self) -> bool {
        self.root.join(CLOSED_REL).exists()
    }

    pub fn seed(&self) -> &Seed {
        &self.seed
    }

    /// The installer CA (PEM): what the front's doors chain to.
    pub fn installer_ca_pem(&self) -> &str {
        &self.seed.installer_ca_pem
    }

    /// This box's OWN certificate, once the installer signed one.
    pub fn own_cert(&self) -> Option<String> {
        std::fs::read_to_string(self.root.join(OWN_CERT_REL)).ok().filter(|c| c.contains("BEGIN CERTIFICATE"))
    }

    /// This box's own identity — `None` while it holds only the medium's
    /// bootstrap certificate (which is nobody's).
    pub fn own_identity(&self) -> Option<Issued> {
        self.own_cert().map(|cert_pem| Issued { cert_pem, key_pem: self.own_key.clone() })
    }
}

/// Pure: a feed a product medium must refuse, by name.
pub fn refusal(seed: &Seed, reply: &JoinReply, may_be_front: bool) -> Option<String> {
    if reply.estate != seed.estate {
        return Some(format!("REFUSED [installer-estate] this VM is of `{}`, the installer fed `{}`", seed.estate, reply.estate));
    }
    if !may_be_front && reply.role == Some(crate::Role::Front) {
        return Some(format!(
            "REFUSED [product-medium-is-not-the-front] the installer fed `{}` role `front`, and this VM booted a PRODUCT medium (primary and twin only); vm1 boots the front medium",
            reply.me.name
        ));
    }
    None
}

impl<P: Product> Vm for Installed<P> {
    fn live_running(&self) -> Result<crate::live::Running, String> {
        self.product.live_running()
    }

    fn live_stage(&self, stage: &crate::live::Stage) -> Result<String, String> {
        self.product.live_stage(stage)
    }

    fn live_stage_declared(&self, stage: &crate::live::Stage, declared: &crate::live::Declared) -> Result<String, String> {
        self.product.live_stage_declared(stage, declared)
    }

    fn hello(&self) -> Result<JoinRequest, String> {
        let capacity = self.product.capacity()?;
        let csr = if self.own_cert().is_some() { String::new() } else { mtls::box_request_for(&self.own_key).map_err(|e| e.to_string())?.0 };
        Ok(JoinRequest { estate: self.seed.estate.clone(), kind: self.product.kind(), tunnel_pubkey: self.product.tunnel_pubkey(), capacity, csr_pem: csr })
    }

    fn fed(&self, reply: &JoinReply) -> Result<(), String> {
        if let Some(why) = refusal(&self.seed, reply, self.product.may_be_front()) {
            return Err(why);
        }
        if let Some(cert) = &reply.cert_pem {
            write_atomic(&self.root, OWN_CERT_REL, cert.as_bytes(), true).map_err(|e| e.to_string())?;
        }
        let text = reply.render().map_err(|e| e.to_string())?;
        crate::fed_at_rest::write(&self.root, JOINED_REL, &text)?;
        *self.fed.lock().map_err(|e| e.to_string())? = Some(reply.clone());
        self.product.fed(reply);
        self.bus.publish(reply.clone());
        Ok(())
    }

    fn status(&self) -> Signal {
        std::fs::read_to_string(self.root.join(SIGNAL_REL))
            .ok()
            .and_then(|t| Signal::parse(&t).ok())
            .unwrap_or_else(|| Signal::of(self.latest().map(|r| r.me.name).as_deref().unwrap_or("?"), self.seed.tunnel, &crate::pair::Machine::default()))
    }

    fn close(&self) -> Result<(), String> {
        write_atomic(&self.root, CLOSED_REL, b"closed by the installer\n", false).map_err(|e| e.to_string())
    }

    fn identity(&self) -> Issued {
        self.own_identity().unwrap_or_else(|| Issued { cert_pem: self.seed.bootstrap_cert_pem.clone(), key_pem: self.seed.bootstrap_key_pem.clone() })
    }

    fn offer(&self) -> BTreeMap<String, String> {
        self.product.offer(self.latest().as_ref())
    }

    /// 0.5.8: after the close, a later answer for this same box — kept and
    /// handed to the product exactly like a feed, once
    /// [`crate::bootstrap::refeed_refusal`] found nothing that changes the box.
    fn refed(&self, reply: &JoinReply) -> Result<(), String> {
        if let Some(why) = refusal(&self.seed, reply, self.product.may_be_front()).or_else(|| crate::bootstrap::refeed_refusal(self.latest().as_ref(), reply)) {
            return Err(why);
        }
        self.fed(reply)
    }
}

/// **Serve the installer from this box** (feature `wireguard`): raise the
/// installer's link from the seed, bind the bootstrap port on the link address
/// only, serve `vm` under mutual TLS until the installer closes it — and then,
/// the standard for every product (0.5.4, owner 2026-09-27: "the installer will
/// receive the logs"), keep answering `GET /install/log` on the same listener
/// for the VM's whole life ([`crate::bootstrap::serve_for_life`]), with the
/// installer's link kept UP (its peer is not dropped). On its own thread;
/// returns at once. `say` gets every line worth a journal.
#[cfg(feature = "wireguard")]
pub fn serve_installer(vm: std::sync::Arc<dyn Vm>, seed: Seed, log: &'static crate::bootlog::BootLog, say: std::sync::Arc<dyn Fn(&str) + Send + Sync>) -> std::io::Result<()> {
    serve_installer_as(vm, seed, log, false, std::sync::Arc::new(move |_, l: &str| say(l)))
}

/// Where a toned serve says its lines: each with its [`crate::bootstrap::Tone`].
#[cfg(feature = "wireguard")]
pub type Say = std::sync::Arc<dyn Fn(crate::bootstrap::Tone, &str) + Send + Sync>;

/// [`serve_installer`], each line with its [`crate::bootstrap::Tone`]: the
/// ordinary steps of a pairing are `Info`, a refusal or a link that cannot be
/// raised `Warn` (stinkies madmin-pairing-warn-level: a product that logs
/// every line at one level said 6 WARN per install).
#[cfg(feature = "wireguard")]
pub fn serve_installer_toned(vm: std::sync::Arc<dyn Vm>, seed: Seed, log: &'static crate::bootlog::BootLog, say: Say) -> std::io::Result<()> {
    serve_installer_as(vm, seed, log, false, say)
}

/// [`serve_installer_after_close`], each line with its tone.
#[cfg(feature = "wireguard")]
pub fn serve_installer_after_close_toned(vm: std::sync::Arc<dyn Vm>, seed: Seed, log: &'static crate::bootlog::BootLog, say: Say) -> std::io::Result<()> {
    serve_installer_as(vm, seed, log, true, say)
}

/// **A box or front that the installer closed on an earlier start** (a
/// restart after the install): the same link and listener as
/// [`serve_installer`], but ONLY the log door, for life
/// ([`crate::bootstrap::serve_after_close`]). A product calls this where it
/// used to open nothing once its kept `closed` file exists — so the operator's
/// `estate logs` still reads the box after a reboot, a recharge or a grow.
#[cfg(feature = "wireguard")]
pub fn serve_installer_after_close(vm: std::sync::Arc<dyn Vm>, seed: Seed, log: &'static crate::bootlog::BootLog, say: std::sync::Arc<dyn Fn(&str) + Send + Sync>) -> std::io::Result<()> {
    serve_installer_as(vm, seed, log, true, std::sync::Arc::new(move |_, l: &str| say(l)))
}

/// The wait before bind attempt `n` (1-based) is asked again: 1 s doubling to 30 s.
pub fn bind_backoff(n: u32) -> std::time::Duration {
    std::time::Duration::from_secs((1u64 << n.saturating_sub(1).min(5)).min(30))
}

/// **Ask `attempt` again until it opens** (lane B4, A9 row 20): each failure is said through
/// `say` with the wait before the next, [`bind_backoff`], slept through `sleep`. Never gives
/// up: the listener is the box's only way to its installer (and its log door) for life.
pub fn bind_until_open<T>(mut attempt: impl FnMut() -> Result<T, String>, mut sleep: impl FnMut(std::time::Duration), mut say: impl FnMut(&str)) -> T {
    let mut n = 0u32;
    loop {
        match attempt() {
            Ok(t) => return t,
            Err(why) => {
                n = n.saturating_add(1);
                let wait = bind_backoff(n);
                say(&format!("installer: {why} — asked again in {} s (attempt {n})", wait.as_secs()));
                sleep(wait);
            }
        }
    }
}

#[cfg(feature = "wireguard")]
fn serve_installer_as(vm: std::sync::Arc<dyn Vm>, seed: Seed, log: &'static crate::bootlog::BootLog, closed: bool, say: Say) -> std::io::Result<()> {
    use crate::bootstrap::Tone;
    std::thread::Builder::new().name("installer-listener".into()).spawn(move || {
        // The link and its bind are asked again until they open (lane B4): a link that is not
        // up yet, or an address not on it yet, is a wait, never the end of the listener.
        let (link, listener) = bind_until_open(
            || {
                let link = crate::installer_link::vm_link(&seed).map_err(|e| format!("the installer's link cannot be raised — the installer cannot reach this VM yet: {e}"))?;
                let listener = std::net::TcpListener::bind((link.bind, crate::bootstrap::PORT)).map_err(|e| format!("the bootstrap listener cannot bind {}:{}: {e}", link.bind, crate::bootstrap::PORT))?;
                Ok((link, listener))
            },
            std::thread::sleep,
            |l| say(Tone::Warn, l),
        );
        say(Tone::Info, &format!("installer: the bootstrap listener is open on {}:{} (the installer's link only; mutual TLS, the installer's certificate only)", link.bind, crate::bootstrap::PORT));
        // Never returns: after the install closes, the same listener is the log
        // door for life, and the link it answers on (owned by this thread) stays up.
        crate::bootstrap::serve_for_life_or_after_close_at(vm.as_ref(), log, &seed.installer_ca_pem, listener, closed, &|t, l| say(t, l));
        drop(link);
    })?;
    Ok(())
}

#[cfg(test)]
mod tests {

    /// ★ **The installer's listener binds again until it can** (lane B4, A9 row 20): a link or
    /// bind that fails (the link not up yet) is said and asked again, 1 s doubling to 30 s,
    /// never given up — a box whose listener never opens can never be reached for life.
    #[test]
    fn the_listener_bind_is_asked_again_until_it_opens() {
        assert_eq!([1, 2, 3, 4, 5, 6, 9].map(|n| bind_backoff(n).as_secs()), [1, 2, 4, 8, 16, 30, 30]);
        let mut tries = 0;
        let (mut slept, mut said) = (Vec::new(), Vec::new());
        let got = bind_until_open(
            || {
                tries += 1;
                if tries < 4 { Err(format!("cannot bind 10.44.0.2:8474: Cannot assign requested address (try {tries})")) } else { Ok(tries) }
            },
            |d| slept.push(d.as_secs()),
            |l| said.push(l.to_string()),
        );
        assert_eq!(got, 4);
        assert_eq!(slept, [1, 2, 4]);
        assert_eq!(said.len(), 3, "{said:?}");
        assert!(said[0].contains("try 1") && said[0].contains("again in 1 s"), "{said:?}");
    }
    use super::*;
    use crate::join::Assigned;

    struct P;
    impl Product for P {
        fn kind(&self) -> Kind {
            Kind::Holger
        }
        fn tunnel_pubkey(&self) -> String {
            "KH".into()
        }
        fn capacity(&self) -> Result<Option<crate::elect::Capacity>, String> {
            Ok(Some(crate::elect::Capacity { cores: 2, ram_bytes: 1 << 30, data_bytes: 1 << 30, disk_mib_s: 100 }))
        }
        fn offer(&self, fed: Option<&JoinReply>) -> BTreeMap<String, String> {
            BTreeMap::from([("version".into(), fed.map(|r| r.me.name.clone()).unwrap_or_default())])
        }
        fn fed(&self, _: &JoinReply) {}
    }

    fn seed(ca: &mtls::InstallerCa) -> Seed {
        let boot = ca.issue_bootstrap(std::time::Duration::from_secs(3600)).unwrap();
        Seed {
            estate: "holger.test".into(),
            tunnel: Default::default(),
            front: Default::default(),
            kind: Kind::Holger,
            installer_ca_pem: ca.cert_pem.clone(),
            bootstrap_cert_pem: boot.cert_pem,
            bootstrap_key_pem: boot.key_pem,
            peers_expected: 3,
            mullvad: None,
            tailscale: None,
            installer_link: None,
        }
    }

    /// A product whose box carries a live slot (its own A/B, korp-installer `live`).
    struct LiveP;
    impl Product for LiveP {
        fn kind(&self) -> Kind {
            Kind::Holger
        }
        fn tunnel_pubkey(&self) -> String {
            String::new()
        }
        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) {}
        fn live_running(&self) -> Result<crate::live::Running, String> {
            Ok(crate::live::Running { base: "b".into(), app: "a".into(), slot: "a".into(), live_update: true })
        }
        fn live_stage(&self, s: &crate::live::Stage) -> Result<String, String> {
            Ok(format!("staged {}", s.pin_hex))
        }
    }

    /// ★ A box answers the installer's live door with its product's slot; a product without
    /// one refuses by name (0.5.14).
    #[test]
    fn a_box_answers_the_live_door_with_its_products_slot_or_refuses_by_name() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let root = std::env::temp_dir().join(format!("boxside-live-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let live = Installed::open(&root, seed(&ca), LiveP).unwrap();
        assert_eq!(live.live_running().unwrap().app, "a");
        let st = crate::live::Stage { pin_hex: "c".repeat(64), signature: vec![1], base: "b".into(), archive: vec![2] };
        assert_eq!(live.live_stage(&st).unwrap(), format!("staged {}", "c".repeat(64)));
        let plain = Installed::open(&root, seed(&ca), P).unwrap();
        assert!(plain.live_running().unwrap_err().contains("[live-unsupported]"));
        assert!(plain.live_stage(&st).unwrap_err().contains("[live-unsupported]"));
        let _ = std::fs::remove_dir_all(&root);
    }

    fn reply(estate: &str, role: Option<crate::Role>, cert: Option<String>) -> JoinReply {
        JoinReply {
            estate: estate.into(),
            me: Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_A },
            cert_pem: cert,
            role,
            primary_overlay: None,
            primary_utility: None,
            peers: Vec::new(),
            feed: BTreeMap::new(),
        }
    }

    #[test]
    fn a_box_presents_its_kind_keeps_its_feed_and_its_own_certificate_and_refuses_the_front() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let root = std::env::temp_dir().join(format!("boxside-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let b = Installed::open(&root, seed(&ca), P).unwrap();
        let hello = b.hello().unwrap();
        assert_eq!((hello.kind, hello.tunnel_pubkey.as_str()), (Kind::Holger, "KH"));
        assert!(!hello.csr_pem.is_empty(), "no certificate yet: it asks for one");
        assert!(b.own_identity().is_none(), "the bootstrap certificate is nobody's");
        let cert = ca.sign_box(&hello.csr_pem, "main", std::time::Duration::from_secs(3600)).unwrap();

        assert!(b.fed(&reply("other.test", Some(crate::Role::Primary), None)).unwrap_err().contains("[installer-estate]"));
        assert!(b.fed(&reply("holger.test", Some(crate::Role::Front), None)).unwrap_err().contains("[product-medium-is-not-the-front]"));
        b.fed(&reply("holger.test", Some(crate::Role::Primary), Some(cert.clone()))).unwrap();
        assert_eq!(b.latest().and_then(|r| r.role), Some(crate::Role::Primary));
        assert_eq!(b.own_identity().map(|i| i.cert_pem), Some(cert));
        assert!(b.hello().unwrap().csr_pem.is_empty(), "a box with its own certificate asks for no other");
        assert_eq!(b.offer()["version"], "main");

        // Reopened (a restart): the key, the certificate and the feed are kept.
        let again = Installed::open(&root, seed(&ca), P).unwrap();
        assert!(again.own_identity().is_some() && again.latest().is_some());
        // Another estate's medium does not read this feed as its own.
        let other = Installed::open(&root, Seed { estate: "other.test".into(), ..seed(&ca) }, P).unwrap();
        assert!(other.latest().is_none());
        assert!(!again.closed());
        again.close().unwrap();
        assert!(again.closed());
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ 0.5.15 (A7 row 8): the box's feed bus carries the reply kept at open, and every
    /// later feed and re-feed: a subscriber added at the start takes a value the installer
    /// feeds LATE (the front's `offer.front.fleet`), without a restart.
    #[test]
    fn a_late_feed_reaches_every_subscriber_of_the_boxs_bus() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let root = std::env::temp_dir().join(format!("boxside-bus-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let b = Installed::open(&root, seed(&ca), P).unwrap();
        assert!(b.feed_bus().current().is_none(), "nothing fed yet");
        let fleet = Arc::new(Mutex::new(Vec::new()));
        let f = Arc::clone(&fleet);
        b.feed_bus().on_change("fleet", |r: &JoinReply| crate::feedbus::offered(&r.feed, "fleet"), move |v: &Option<String>| f.lock().unwrap().push(v.clone()));
        b.fed(&reply("holger.test", Some(crate::Role::Primary), None)).unwrap();
        b.close().unwrap();
        let mut late = reply("holger.test", Some(crate::Role::Primary), None);
        late.feed.insert("offer.front.fleet".into(), "https://one.example".into());
        b.refed(&late).unwrap();
        b.refed(&late).unwrap();
        assert_eq!(*fleet.lock().unwrap(), vec![None, Some("https://one.example".to_string())]);
        // A restart publishes the kept reply at open: the subscriber gets it at once.
        let again = Installed::open(&root, seed(&ca), P).unwrap();
        assert_eq!(again.feed_bus().current().map(|r| r.feed.get("offer.front.fleet").cloned()), Some(Some("https://one.example".into())));
        let _ = std::fs::remove_dir_all(&root);
    }

    /// 0.5.8: a closed box is RE-FED when a peer was re-admitted — kept like a
    /// feed, the product told — and a re-feed that would change the box itself
    /// (its role, its name, a new certificate) is refused and nothing is kept.
    #[test]
    fn a_closed_box_keeps_a_refeed_about_its_peers_and_refuses_one_that_changes_it() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let root = std::env::temp_dir().join(format!("boxside-refeed-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let b = Installed::open(&root, seed(&ca), P).unwrap();
        assert!(b.refed(&reply("holger.test", Some(crate::Role::Twin), None)).unwrap_err().contains("[refeed-never-fed]"));
        b.fed(&reply("holger.test", Some(crate::Role::Twin), None)).unwrap();
        b.close().unwrap();
        let moved = JoinReply { primary_utility: Some([10, 13, 8, 77].into()), ..reply("holger.test", Some(crate::Role::Twin), None) };
        b.refed(&moved).unwrap();
        assert_eq!(b.latest().and_then(|r| r.primary_utility), Some([10, 13, 8, 77].into()));
        let kept = Installed::open(&root, seed(&ca), P).unwrap();
        assert_eq!(kept.latest().and_then(|r| r.primary_utility), Some([10, 13, 8, 77].into()), "kept across a restart");
        for bad in [reply("holger.test", Some(crate::Role::Primary), None), reply("holger.test", Some(crate::Role::Twin), Some("C".into())), reply("other.test", Some(crate::Role::Twin), None)] {
            assert!(b.refed(&bad).unwrap_err().contains("REFUSED ["));
        }
        assert_eq!(b.latest(), Some(moved), "a refused re-feed kept nothing");
        let _ = std::fs::remove_dir_all(&root);
    }

    #[test]
    fn the_first_boot_after_an_installation_clears_the_pairing_state_once() {
        let esp = std::env::temp_dir().join(format!("boxside-esp-{}", std::process::id()));
        let data = std::env::temp_dir().join(format!("boxside-data-{}", std::process::id()));
        let _ = (std::fs::remove_dir_all(&esp), std::fs::remove_dir_all(&data));
        std::fs::create_dir_all(esp.join("korp")).unwrap();
        for rel in PAIRING_STATE {
            write_atomic(&data, rel, b"old", false).unwrap();
        }
        let mark = |p: &Path| std::fs::write(p, "x").is_ok();
        assert_eq!(reset_if_fresh_install(&esp, &data, &PAIRING_STATE, &mark).map(|v| v.len()), Some(5));
        assert_eq!(reset_if_fresh_install(&esp, &data, &PAIRING_STATE, &mark), None, "once");
        write_atomic(&data, JOINED_REL, b"new", false).unwrap();
        let _ = (std::fs::remove_dir_all(&esp), std::fs::remove_dir_all(&data));
        // An ESP that cannot be marked keeps the state.
        let esp2 = std::env::temp_dir().join(format!("boxside-esp2-{}", std::process::id()));
        std::fs::create_dir_all(esp2.join("korp")).unwrap();
        assert_eq!(reset_if_fresh_install(&esp2, &data, &PAIRING_STATE, &|_| false), None);
        let _ = std::fs::remove_dir_all(&esp2);
    }

    fn fed_with_secret(estate: &str) -> JoinReply {
        let mut r = reply(estate, Some(crate::Role::Primary), None);
        r.feed.insert("console_admin_token".into(), "t0k3n-NEVER-on-disk-in-the-clear".into());
        r
    }

    /// ★ 0.5.17 (A7 row 21): what a box was fed is SEALED at rest, 0600, and reads back.
    #[test]
    fn a_box_keeps_its_feed_sealed_at_rest_and_reads_it_back_after_a_restart() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let root = std::env::temp_dir().join(format!("boxside-sealed-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        let b = Installed::open(&root, seed(&ca), P).unwrap();
        b.fed(&fed_with_secret("holger.test")).unwrap();
        let raw = std::fs::read(root.join(JOINED_REL)).unwrap();
        assert!(crate::fed_at_rest::is_sealed(&raw), "joined.toml is sealed");
        assert!(!String::from_utf8_lossy(&raw).contains("t0k3n-NEVER"), "the fed token is not readable on disk");
        let again = Installed::open(&root, seed(&ca), P).unwrap();
        assert_eq!(again.latest().and_then(|r| r.feed.get("console_admin_token").cloned()).as_deref(), Some("t0k3n-NEVER-on-disk-in-the-clear"));
        let _ = std::fs::remove_dir_all(&root);
    }

    /// ★ 0.5.17 (A7 row 6, Gunnar ff58dad1): a KEEP boot (a new ESP onto the kept data) of
    /// the SAME estate keeps the feed, the box's own key and its certificate; it is marked
    /// once. A feed of another estate, or one that cannot be read, is cleared as before.
    #[test]
    fn a_keep_boot_of_the_same_estate_keeps_the_feed_the_key_and_the_certificate() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let tag = std::process::id();
        let data = std::env::temp_dir().join(format!("boxside-keep-data-{tag}"));
        let _ = std::fs::remove_dir_all(&data);
        let s = seed(&ca);
        let b = Installed::open(&data, s.clone(), P).unwrap();
        let cert = ca.sign_box(&b.hello().unwrap().csr_pem, "main", std::time::Duration::from_secs(3600)).unwrap();
        b.fed(&reply("holger.test", Some(crate::Role::Primary), Some(cert.clone()))).unwrap();
        let own_key = std::fs::read_to_string(data.join(OWN_KEY_REL)).unwrap();
        drop(b);
        let mark = |p: &Path| std::fs::write(p, "x").is_ok();
        let esp = std::env::temp_dir().join(format!("boxside-keep-esp-{tag}"));
        let _ = std::fs::remove_dir_all(&esp);
        std::fs::create_dir_all(esp.join("korp")).unwrap();
        let cleared = reset_if_fresh_install_of(&esp, &data, &PAIRING_STATE, &s, &mark);
        assert!(cleared.as_ref().is_none_or(|c| c.is_empty()), "a keep boot of the same estate clears nothing: {cleared:?}");
        assert!(esp.join(ESP_CLEARED).exists(), "the boot is marked, so it is decided once");
        let after = Installed::open(&data, s.clone(), P).unwrap();
        assert!(after.latest().is_some(), "still fed");
        assert_eq!(after.own_identity().map(|i| i.cert_pem), Some(cert), "its certificate kept");
        assert_eq!(std::fs::read_to_string(data.join(OWN_KEY_REL)).unwrap(), own_key, "its own key kept");
        // Gunnar's other half: installed for ANOTHER estate onto the kept data: unpaired.
        let esp2 = std::env::temp_dir().join(format!("boxside-keep-esp2-{tag}"));
        let _ = std::fs::remove_dir_all(&esp2);
        std::fs::create_dir_all(esp2.join("korp")).unwrap();
        let other = Seed { estate: "other.test".into(), ..s.clone() };
        let cleared = reset_if_fresh_install_of(&esp2, &data, &PAIRING_STATE, &other, &mark).expect("a first boot");
        assert!(cleared.contains(&JOINED_REL.to_string()) && cleared.contains(&OWN_KEY_REL.to_string()), "{cleared:?}");
        // A feed that cannot be read is cleared too.
        write_atomic(&data, JOINED_REL, b"not a reply", true).unwrap();
        let esp3 = std::env::temp_dir().join(format!("boxside-keep-esp3-{tag}"));
        let _ = std::fs::remove_dir_all(&esp3);
        std::fs::create_dir_all(esp3.join("korp")).unwrap();
        assert!(reset_if_fresh_install_of(&esp3, &data, &PAIRING_STATE, &s, &mark).expect("a first boot").contains(&JOINED_REL.to_string()));
        // A front's feed is not a box's: a product seed clears it.
        let front = Seed { kind: Kind::Front, ..s.clone() };
        let esp4 = std::env::temp_dir().join(format!("boxside-keep-esp4-{tag}"));
        let _ = std::fs::remove_dir_all(&esp4);
        std::fs::create_dir_all(esp4.join("korp")).unwrap();
        crate::fed_at_rest::write(&data, JOINED_REL, &reply("holger.test", Some(crate::Role::Primary), None).render().unwrap()).unwrap();
        assert!(reset_if_fresh_install_of(&esp4, &data, &PAIRING_STATE, &front, &mark).expect("a first boot").contains(&JOINED_REL.to_string()), "a box's feed under a front seed is not the front's");
        for d in [&data, &esp, &esp2, &esp3, &esp4] {
            let _ = std::fs::remove_dir_all(d);
        }
    }

    /// ★ Lane D14 (rickard-81): the CLEARING path says a marker it could not write, the way the
    /// keep path does (ERROR): nothing is cleared, and the next boot asks again.
    #[test]
    fn the_clearing_first_boot_says_a_marker_it_could_not_write() {
        let tag = format!("clear-said-{}", std::process::id());
        let (data, esp) = (std::env::temp_dir().join(format!("boxside-{tag}-data")), std::env::temp_dir().join(format!("boxside-{tag}-esp")));
        let _ = std::fs::remove_dir_all(&esp);
        std::fs::create_dir_all(esp.join("korp")).unwrap();
        let (cleared, said) = crate::heard(|| reset_if_fresh_install(&esp, &data, &PAIRING_STATE, &|_| false));
        assert_eq!(cleared, None, "nothing cleared without the marker");
        assert!(said.contains("ERROR") && said.contains("the ESP marker could not be written") && said.contains("NOT cleared"), "{said}");
        let _ = std::fs::remove_dir_all(&esp);
    }

    /// ★ Lane D14 (rickard-81): the marker on a READ-ONLY /boot is written with the ESP flipped
    /// read-write for the write and back (appliance-pid1 `with_esp_writable`); one outside the
    /// ESP mount is written as it is.
    #[test]
    fn the_esp_marker_on_a_read_only_boot_is_written_with_the_esp_flipped_read_write_and_back() {
        use std::cell::RefCell;
        let esp = std::env::temp_dir().join(format!("boxside-ro-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&esp);
        let marker = esp.join(ESP_CLEARED);
        let flips = RefCell::new(Vec::new());
        let ok = mark_esp_on(&esp, &marker, &|m: &Path| {
            appliance_pid1::stop_record::with_esp_writable_by(m, &|_| true, &|_, ro| { flips.borrow_mut().push(ro); Ok(()) }, || { assert_eq!(*flips.borrow(), vec![false], "written while read-write"); write_marker_file(&marker) })
        });
        assert!(ok && marker.exists());
        assert_eq!(*flips.borrow(), vec![false, true], "flipped read-write, then back");
        let other = std::env::temp_dir().join(format!("boxside-ro-other-{}", std::process::id())).join(ESP_CLEARED);
        assert!(mark_esp_on(&esp, &other, &|_| panic!("not on the ESP mount: no flip")));
        let _ = std::fs::remove_dir_all(&esp);
        let _ = std::fs::remove_dir_all(other.parent().unwrap().parent().unwrap());
    }

    /// 0.5.17 (G4): the ESP marker is written whole (staged, synced, renamed), its directory
    /// made when missing, and no staging file is left beside it.
    #[test]
    fn the_esp_marker_is_written_whole_and_leaves_nothing_beside_it() {
        let esp = std::env::temp_dir().join(format!("boxside-mark-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&esp);
        assert!(mark_esp(&esp.join(ESP_CLEARED)));
        assert!(std::fs::read_to_string(esp.join(ESP_CLEARED)).unwrap().contains("cleared"));
        let beside: Vec<_> = std::fs::read_dir(esp.join("korp")).unwrap().map(|e| e.unwrap().file_name()).collect();
        assert_eq!(beside.len(), 1, "{beside:?}");
        let _ = std::fs::remove_dir_all(&esp);
    }

    /// ★ 0.5.18 (lane G7, Gunnar 7e33c0f7 `vm::first_boot`): the first boot SAYS what it
    /// decided — a keep boot of the same estate (INFO, the feed KEPT), a marker that could not
    /// be written on a keep boot (ERROR), a kept feed that cannot be read and is cleared (WARN).
    #[test]
    fn the_first_boot_says_a_keep_boot_a_marker_it_could_not_write_and_a_feed_it_could_not_read() {
        let ca = mtls::InstallerCa::mint("holger.test", std::time::Duration::from_secs(3600)).unwrap();
        let tag = format!("said-{}", std::process::id());
        let data = std::env::temp_dir().join(format!("boxside-{tag}-data"));
        let _ = std::fs::remove_dir_all(&data);
        let s = seed(&ca);
        crate::fed_at_rest::write(&data, JOINED_REL, &reply("holger.test", Some(crate::Role::Primary), None).render().unwrap()).unwrap();
        let esp = |n: u8| {
            let e = std::env::temp_dir().join(format!("boxside-{tag}-esp{n}"));
            let _ = std::fs::remove_dir_all(&e);
            std::fs::create_dir_all(e.join("korp")).unwrap();
            e
        };
        let (e1, e2, e3) = (esp(1), esp(2), esp(3));
        let (kept, said) = crate::heard(|| reset_if_fresh_install_of(&e1, &data, &PAIRING_STATE, &s, &|p| mark_esp(p)));
        assert_eq!(kept, Some(Vec::new()));
        assert!(said.contains("INFO") && said.contains("a keep boot of the same estate") && said.contains("KEPT"), "{said}");
        let (kept, said) = crate::heard(|| reset_if_fresh_install_of(&e2, &data, &PAIRING_STATE, &s, &|_| false));
        assert_eq!(kept, Some(Vec::new()));
        assert!(said.contains("ERROR") && said.contains("the ESP marker could not be written"), "{said}");
        let mut junk = crate::fed_at_rest::SEAL_MAGIC.to_vec();
        junk.extend_from_slice(&[9u8; 64]);
        write_atomic(&data, JOINED_REL, &junk, true).unwrap();
        let (cleared, said) = crate::heard(|| reset_if_fresh_install_of(&e3, &data, &PAIRING_STATE, &s, &|p| mark_esp(p)));
        assert!(cleared.expect("a first boot").contains(&JOINED_REL.to_string()));
        assert!(said.contains("WARN") && said.contains("the kept feed cannot be read") && said.contains("cleared"), "{said}");
        for d in [&data, &e1, &e2, &e3] {
            let _ = std::fs::remove_dir_all(d);
        }
    }
}