nordisk-pairing 0.5.26

//! **The installer — on the operator's machine, connecting OUT to every VM**
//! (0.3.0, owner 2026-09-24).
//!
//! Until 0.2 the installer was a join service on the front: the VMs dialled
//! it. The owner's decision: the installer runs FIRST and lives with the
//! operator (inside the xtask's run, as its own binary); the operator's
//! machine is behind NAT, so the direction is reversed — the installer
//! connects to each VM's bootstrap listener ([`crate::bootstrap`]), retrying
//! with backoff until each VM answers, over the path the seed's tunnel kind
//! names ([`crate::installer_link`]). For every VM it:
//!
//! 1. reads the VM's presentation (`hello`) and admits it into the
//!    [`Door`]'s ledger — by the address the installer DIALLED, which is what
//!    terraform said the VM is; the same ledger rules, election and feed as
//!    before, now in the installer's own process;
//! 2. FEEDS it: name, overlay address, its own certificate (signed from its
//!    request), the role once elected, the peers (the front's tunnel first)
//!    and the estate values — again whenever the answer changes;
//! 3. collects its FULL boot log ([`crate::bootlog`]), every line once,
//!    secrets withheld ([`crate::redact`]), and its pairing signal;
//! 4. when every VM is done ([`CloseWhen`]), tells each to CLOSE: the
//!    install goes — the VM keeps ONLY its log door for the installer
//!    ([`crate::bootstrap::serve_for_life`], 0.5.4) — and the installer
//!    proves the install is gone ([`Client::install_gone`]).
//!
//! The VMs then pair with each other over the tunnel exactly as before. What
//! the installer knows is handed to a [`Sink`] as it happens: the log lines,
//! each VM's signal and last line, and the ledger (which the front's relay
//! needs, to list each VM's tunnel key).

use crate::bootlog::{Chunk, MAX_READ};
use crate::door::{Door, Response};
use crate::installer_link::{reach, LinkKeys, Path, Reach};
#[cfg(test)]
use crate::installer_link::DirectPath;
use crate::join::{FrontTunnel, JoinError, JoinReply, JoinRequest, Ledger, Presented, Slot};
use crate::mtls::{self, InstallerCa, Issued};
use crate::seed::{Kind, TunnelKind};
use crate::signal::Signal;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddrV4};
use std::path::{Path as FsPath, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// When the installer is done with the VMs and closes them.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum CloseWhen {
    /// Every VM's pairing signal says PAIRED (the default: the boot logs are
    /// collected through the pairing).
    #[default]
    Paired,
    /// Every VM has been fed its role and every peer's tunnel key.
    Fed,
}

/// One VM the operator laid.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PlanVm {
    pub name: String,
    /// The address the installer dials (terraform's public address).
    pub public: Ipv4Addr,
    /// The VM's utility address — where the other boxes reach its control plane.
    pub utility: Ipv4Addr,
    pub overlay: Ipv4Addr,
    #[serde(default = "gunnar_kind")]
    pub kind: Kind,
    /// **The values THIS VM alone is fed** (0.5.0, owner 2026-09-25), over
    /// [`Plan::feed`] (its own wins on a clash). The front's TLS certificate
    /// and key, the console's admin token and plugin modules go HERE, on the
    /// front's VM — [`Plan::check`] refuses them in the shared feed and on a
    /// box. Never written to the ledger ([`Door::own_feed`]).
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub feed: BTreeMap<String, String>,
}

/// `Debug` names the VM's own feed by its KEYS only (0.5.0).
impl std::fmt::Debug for PlanVm {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PlanVm")
            .field("name", &self.name)
            .field("public", &self.public)
            .field("utility", &self.utility)
            .field("overlay", &self.overlay)
            .field("kind", &self.kind)
            .field("feed (keys)", &self.feed.keys().collect::<Vec<_>>())
            .finish()
    }
}

fn gunnar_kind() -> Kind {
    Kind::Gunnar
}

/// **Feed keys only the FRONT may hold** (0.5.0): the web certificate and its
/// private key (`tls.<host>.cert` / `.key`), the monetize plugins
/// (`plugin.<id>.*`, `plugin_root.<name>`) and the console's break-glass
/// bearer. Main and twin never need them, so they must never receive them:
/// such a key in [`Plan::feed`] or in a box's [`PlanVm::feed`] is refused by
/// name ([`Plan::check`]).
pub const FRONT_ONLY_PREFIXES: &[&str] = &["tls.", "plugin.", "plugin_root.", FEED_WG_PSK_PREFIX];
/// **The feed key a BOX holds its WireGuard pre-shared key under** (2026-10-06,
/// the PQ transport lane): the PSK of its link to the front's relay
/// (WireGuard's `PresharedKey`, base64, 32 bytes). WireGuard's handshake is
/// X25519 alone; a PSK both ends hold, that crossed only the installer's
/// post-quantum mutual TLS ([`crate::pqtls`]), is what a recording of the
/// tunnel cannot be opened without. In the box's OWN feed, never the shared
/// one ([`Plan::with_wg_psks`]).
pub const FEED_WG_PSK: &str = "wg_psk";
/// The prefix of the FRONT's feed keys holding each box's PSK:
/// `wg_psk.<box>` ([`feed_wg_psk_for`]). Front-only ([`FRONT_ONLY_PREFIXES`]).
pub const FEED_WG_PSK_PREFIX: &str = "wg_psk.";

/// The front's feed key for box `name`'s pre-shared key.
pub fn feed_wg_psk_for(name: &str) -> String {
    format!("{FEED_WG_PSK_PREFIX}{name}")
}

/// Whole feed keys only the front may hold (see [`FRONT_ONLY_PREFIXES`]).
pub const FRONT_ONLY_KEYS: &[&str] = &["console_admin_token", "monetize_backup_kek"];

/// Whether `key` is a front-only feed key.
pub fn front_only(key: &str) -> bool {
    FRONT_ONLY_KEYS.contains(&key) || FRONT_ONLY_PREFIXES.iter().any(|p| key.starts_with(p))
}

fn default_port() -> u16 {
    crate::bootstrap::PORT
}

fn default_link_port() -> u16 {
    crate::installer_link::LINK_PORT
}

/// **One run of the installer**: what the operator's tooling hands it (TOML).
/// No medium key: those are read from [`Material`]. **A plan whose front VM
/// carries its own feed ([`PlanVm::feed`]: the TLS key, the console token)
/// holds secrets** — keep it in memory; a rendering written to disk must be
/// 0600 like the run folder's secret store.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Plan {
    pub estate: String,
    pub tunnel: TunnelKind,
    pub vms: Vec<PlanVm>,
    /// The estate values every VM is fed.
    #[serde(default)]
    pub feed: BTreeMap<String, String>,
    /// The front's tunnel (relay key + endpoint), fed as the `front` peer.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub front: Option<FrontTunnel>,
    /// The bootstrap listener's port (direct mode).
    #[serde(default = "default_port")]
    pub port: u16,
    /// Each VM's link port (tunnel mode).
    #[serde(default = "default_link_port")]
    pub link_port: u16,
    pub budget_s: u64,
    #[serde(default)]
    pub close_when: CloseWhen,
    /// How long the box certificates are valid (default 24 h).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub box_cert_valid_s: Option<u64>,
    /// **What the FRONT must have been fed before anything closes** (0.5.2):
    /// offer keys of the ELECTED PRIMARY — e.g. `ticket_pubkey`, which the
    /// front's monetize needs as `appliance_key`. No VM is closed until the
    /// front's last feed carries `offer.<primary>.<key>` for each one: the
    /// primary offers its key only once it is elected, and a close that started
    /// first left the front without it for good.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub front_waits_for: Vec<String>,
}

impl Plan {
    pub fn parse(text: &str) -> Result<Plan, String> {
        let p: Plan = toml::from_str(text).map_err(|e| format!("REFUSED [installer-plan] not an installer plan: {e}"))?;
        p.check()?;
        Ok(p)
    }

    pub fn render(&self) -> Result<String, String> {
        self.check()?;
        toml::to_string(self).map_err(|e| e.to_string())
    }

    pub fn check(&self) -> Result<(), String> {
        let bad = |why: String| Err(format!("REFUSED [installer-plan] {why}"));
        if self.vms.is_empty() {
            return bad("no VM to install".into());
        }
        let mut seen = std::collections::BTreeSet::new();
        for v in &self.vms {
            if !seen.insert(v.public) {
                return bad(format!("two VMs at {} — the installer knows a VM by the address it dials", v.public));
            }
            if v.name == mtls::BOOTSTRAP_CN || v.name == mtls::INSTALLER_CN {
                return bad(format!("`{}` is not a VM name", v.name));
            }
            if v.kind != Kind::Front {
                if let Some(k) = v.feed.keys().find(|k| front_only(k)) {
                    return Err(format!("REFUSED [feed-front-only-on-a-box] `{k}` is the front's alone, and the plan feeds it to `{}` ({:?}); put it in the front VM's own feed", v.name, v.kind));
                }
            }
        }
        if let Some(k) = self.feed.keys().find(|k| front_only(k)) {
            return Err(format!("REFUSED [feed-front-only-shared] `{k}` is the front's alone, and the shared feed reaches every VM (main and twin too); put it in the front VM's own feed (`PlanVm::feed`)"));
        }
        self.check_wg_psks()?;
        reach(self.tunnel).map_err(|e| e.to_string())?;
        Ok(())
    }

    /// **The WireGuard pre-shared keys of the estate's links** (2026-10-06):
    /// `psks` maps each BOX's name to the PSK of its link to the front's relay
    /// (WireGuard base64, 32 bytes). Each box is fed its own as
    /// [`FEED_WG_PSK`], the front every box's as `wg_psk.<box>` — in the VMs'
    /// OWN feeds ([`PlanVm::feed`]), so no box ever holds another pair's key and
    /// the ledger never holds any. The caller keeps them (0600) so a later run
    /// — a re-admission, an upgrade — feeds both ends the same keys.
    pub fn with_wg_psks(mut self, psks: &BTreeMap<String, String>) -> Result<Plan, String> {
        let bad = |why: String| format!("REFUSED [installer-wg-psk] {why}");
        let front = self.vms.iter().position(|v| v.kind == Kind::Front).ok_or_else(|| bad("the plan has no front: a pre-shared key is for a box's link to the front's relay".into()))?;
        for (name, psk) in psks {
            let i = self.vms.iter().position(|v| &v.name == name).ok_or_else(|| bad(format!("`{name}` is not a VM of this plan")))?;
            if i == front {
                return Err(bad(format!("`{name}` is the front; a pre-shared key is a BOX's, for its link to the front")));
            }
            b64_32(psk, &format!("`{name}`'s pre-shared key")).map_err(bad)?;
            self.vms[i].feed.insert(FEED_WG_PSK.to_string(), psk.trim().to_string());
            self.vms[front].feed.insert(feed_wg_psk_for(name), psk.trim().to_string());
        }
        self.check()?;
        Ok(self)
    }

    /// The two ends of every link hold ONE pre-shared key, and only they do.
    fn check_wg_psks(&self) -> Result<(), String> {
        if self.feed.contains_key(FEED_WG_PSK) {
            return Err(format!("REFUSED [feed-wg-psk-shared] `{FEED_WG_PSK}` in the shared feed would give every box one key; each box's is its own (`Plan::with_wg_psks`)"));
        }
        let front = self.vms.iter().find(|v| v.kind == Kind::Front);
        for v in &self.vms {
            match (v.kind, v.feed.get(FEED_WG_PSK)) {
                (Kind::Front, Some(_)) => return Err(format!("REFUSED [feed-wg-psk-on-the-front] `{}` is the front and holds a box's `{FEED_WG_PSK}`; the front holds each box's as `{FEED_WG_PSK_PREFIX}<box>`", v.name)),
                (_, mine) => {
                    let theirs = front.and_then(|f| f.feed.get(&feed_wg_psk_for(&v.name)));
                    if v.kind != Kind::Front && mine != theirs {
                        return Err(format!(
                            "REFUSED [feed-wg-psk-split] the link of `{}` to the front: the box holds {} and the front {} — its two ends must hold one key, or neither any",
                            v.name,
                            if mine.is_some() { "a pre-shared key" } else { "none" },
                            if theirs.is_some() { "a pre-shared key" } else { "none" }
                        ));
                    }
                }
            }
        }
        if let Some(f) = front {
            if let Some(k) = f.feed.keys().filter_map(|k| k.strip_prefix(FEED_WG_PSK_PREFIX)).find(|n| !self.vms.iter().any(|v| v.name == *n && v.kind != Kind::Front)) {
                return Err(format!("REFUSED [feed-wg-psk-split] the front holds a pre-shared key for `{k}`, which is no box of this plan"));
            }
        }
        Ok(())
    }

    /// The door this plan lays: one slot per VM by its PUBLIC address, its
    /// utility address kept beside it.
    pub fn door(&self) -> Door {
        let mut l = Ledger::new(&self.estate);
        for v in &self.vms {
            l.lay(v.public, Slot { name: v.name.clone(), kind: v.kind, overlay: v.overlay, endpoint: None, utility: Some(v.utility) });
        }
        let mut d = Door::new(l);
        d.feed = self.feed.clone();
        d.own_feed = self.vms.iter().filter(|v| !v.feed.is_empty()).map(|v| (v.name.clone(), v.feed.clone())).collect();
        d.front = self.front.clone();
        d
    }
}

/// **The medium's installer material**, as files in one directory on the
/// operator's machine (0600): the CA, the installer's own certificate, and
/// the installer's half of the WireGuard link.
pub struct Material {
    pub ca: InstallerCa,
    pub me: Issued,
    pub link: Option<LinkKeys>,
}

pub const CA_CERT: &str = "ca.cert.pem";
pub const CA_KEY: &str = "ca.key.pem";
pub const INSTALLER_CERT: &str = "installer.cert.pem";
pub const INSTALLER_KEY: &str = "installer.key.pem";
/// `installer_private = "<b64>"` and `vm_public = "<b64>"`.
pub const LINK_KEYS: &str = "link.toml";

#[derive(Serialize, Deserialize)]
struct LinkFile {
    installer_private: String,
    vm_public: String,
}

fn b64_32(s: &str, what: &str) -> Result<[u8; 32], String> {
    let t = s.trim();
    let mut out = [0u8; 32];
    let alphabet = |c: u8| -> Option<u8> {
        Some(match c {
            b'A'..=b'Z' => c - b'A',
            b'a'..=b'z' => c - b'a' + 26,
            b'0'..=b'9' => c - b'0' + 52,
            b'+' => 62,
            b'/' => 63,
            _ => return None,
        })
    };
    let bytes = t.as_bytes();
    if bytes.len() != 44 || bytes[43] != b'=' {
        return Err(format!("{what} is not a 44-character WireGuard key"));
    }
    let mut acc: u32 = 0;
    let mut bits = 0;
    let mut n = 0;
    for &c in &bytes[..43] {
        let v = alphabet(c).ok_or_else(|| format!("{what} is not base64"))?;
        acc = (acc << 6) | v as u32;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            if n < 32 {
                out[n] = (acc >> bits) as u8;
            }
            n += 1;
        }
    }
    if n != 32 {
        return Err(format!("{what} decodes to {n} bytes, not 32"));
    }
    Ok(out)
}

impl Material {
    pub fn load(dir: &FsPath) -> Result<Material, String> {
        let read = |f: &str| std::fs::read_to_string(dir.join(f)).map_err(|e| format!("REFUSED [installer-material] {}: {e}", dir.join(f).display()));
        let ca = InstallerCa { cert_pem: read(CA_CERT)?, key_pem: read(CA_KEY)? };
        let me = Issued { cert_pem: read(INSTALLER_CERT)?, key_pem: read(INSTALLER_KEY)? };
        let link = match std::fs::read_to_string(dir.join(LINK_KEYS)) {
            Ok(t) => Some(Material::link_of(&t).map_err(|e| format!("REFUSED [installer-material] {}: {e}", dir.join(LINK_KEYS).display()))?),
            Err(_) => None,
        };
        Ok(Material { ca, me, link })
    }

    /// The link keys from the link file's text ([`LINK_KEYS`]).
    pub fn link_of(text: &str) -> Result<LinkKeys, String> {
        let f: LinkFile = toml::from_str(text).map_err(|e| e.to_string())?;
        Ok(LinkKeys { installer_private: b64_32(&f.installer_private, "installer_private")?, vm_public: b64_32(&f.vm_public, "vm_public")? })
    }

    /// The link file's text, for the caller that mints it (base64 keys).
    pub fn link_file(installer_private_b64: &str, vm_public_b64: &str) -> String {
        toml::to_string(&LinkFile { installer_private: installer_private_b64.to_string(), vm_public: vm_public_b64.to_string() }).unwrap_or_default()
    }
}

/// **Where the installer reports** as it works.
pub trait Sink: Send + Sync {
    /// Log lines of `vm`, already filtered, in order, each exactly once.
    fn lines(&self, vm: &str, lines: &[String]);
    /// A VM's pairing signal and its last log line changed.
    fn status(&self, vm: &str, signal: Option<&Signal>, last_line: &str);
    /// The ledger changed (an admission, a verdict, a late tunnel key).
    fn ledger(&self, door: &Door);
    /// One line of the installer's own narration.
    fn say(&self, line: &str);
    /// A VM's offer changed (0.3.1, [`crate::bootstrap::Vm::offer`]): the
    /// whole offer as it stands — e.g. the primary's `ticket_pubkey`, which
    /// the operator's tooling carries on to monetize. The installer has
    /// already fed it to the other VMs.
    fn offer(&self, _vm: &str, _offer: &BTreeMap<String, String>) {}
    /// **What a VM handed over as a SECRET** (0.5.0,
    /// [`crate::bootstrap::SecretOffer`]), once, at the end of its install
    /// (right before the close) — the front's certificate and key per host,
    /// for its SUCCESSOR's feed. Store it (0600) and nowhere else: never
    /// print, log or ledger a value. REQUIRED, with no default, on purpose: a
    /// sink that wraps another must pass it on, or the next rebuild has no
    /// certificate to carry over (Let's Encrypt's 5 a week).
    /// An `Err` (the store could not be written) fails the close, which is
    /// tried again: a close that lost the certificate is not a close.
    fn secret_offer(&self, vm: &str, offer: &crate::bootstrap::SecretOffer) -> Result<(), String>;
}

/// What became of one VM.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct VmOutcome {
    pub name: String,
    pub admitted: bool,
    pub role: Option<crate::Role>,
    pub log_lines: u64,
    pub last_line: String,
    pub signal: Option<String>,
    pub closed: bool,
    /// After the close, the VM no longer answered on the installer's path.
    pub gone: bool,
    pub last_error: Option<String>,
    /// The feed keys (non-blank) of the LAST reply this VM was fed (0.5.2) —
    /// so the caller judges per VM what each one holds (e.g. did the front
    /// get `offer.<primary>.ticket_pubkey`), not merely that someone offered it.
    pub fed_keys: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outcome {
    pub vms: Vec<VmOutcome>,
    pub took: Duration,
}

// ── one request over mutual TLS ────────────────────────────────────────────

fn call(path: &mut dyn Path, cfg: &Arc<rustls::ClientConfig>, method: &str, p: &str, body: &str, timeout: Duration) -> Result<(Response, Presented), String> {
    let wire = path.connect(timeout)?;
    let conn = rustls::ClientConnection::new(cfg.clone(), mtls::vm_name()).map_err(|e| e.to_string())?;
    let mut tls = rustls::StreamOwned::new(conn, wire);
    let raw = format!("{method} {p} HTTP/1.1\r\nhost: {}\r\ncontent-type: application/toml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", mtls::VM_SERVER_NAME, body.len());
    tls.write_all(raw.as_bytes()).map_err(|e| format!("could not be reached: TLS: {e}"))?;
    let _ = tls.flush();
    let mut buf = Vec::new();
    match tls.read_to_end(&mut buf) {
        Ok(_) => {}
        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && !buf.is_empty() => {}
        Err(e) => return Err(format!("could not be reached: reading: {e}")),
    }
    let who = mtls::server_cn(&tls.conn).map(|cn| Presented::from_cn(&cn)).ok_or("the VM showed no certificate")?;
    Ok((Response::parse(&buf)?, who))
}

fn ok_body(r: Response) -> Result<String, String> {
    if r.status == 200 {
        Ok(r.body)
    } else {
        Err(r.body.trim().to_string())
    }
}

/// **The installer's client of ONE VM** — the five requests of
/// [`crate::bootstrap`], over the VM's path.
pub struct Client {
    path: Box<dyn Path>,
    cfg: Arc<rustls::ClientConfig>,
    timeout: Duration,
}

impl Client {
    pub fn new(path: Box<dyn Path>, ca_pem: &str, me: &Issued, timeout: Duration) -> Result<Client, String> {
        Ok(Client { path, cfg: mtls::client_config(ca_pem, me).map_err(|e| e.to_string())?, timeout })
    }

    pub fn describe(&self) -> String {
        self.path.describe()
    }

    pub fn hello(&mut self) -> Result<(JoinRequest, Presented), String> {
        let (r, who) = call(self.path.as_mut(), &self.cfg, "GET", crate::bootstrap::HELLO_PATH, "", self.timeout)?;
        Ok((JoinRequest::parse(&ok_body(r)?).map_err(|e| e.to_string())?, who))
    }

    pub fn log(&mut self, from: u64) -> Result<Chunk, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", &format!("{}?from={from}", crate::bootstrap::LOG_PATH), "", self.timeout)?;
        Chunk::parse(&ok_body(r)?)
    }

    pub fn feed(&mut self, reply: &JoinReply) -> Result<(), String> {
        let body = reply.render().map_err(|e| e.to_string())?;
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::bootstrap::FEED_PATH, &body, self.timeout)?;
        ok_body(r).map(|_| ())
    }

    pub fn status(&mut self) -> Result<Signal, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", crate::bootstrap::STATUS_PATH, "", self.timeout)?;
        Signal::parse(&ok_body(r)?)
    }

    /// What the VM offers (0.3.1). A VM from before 0.3.1 has no such path
    /// (404): that is an empty offer, not an error.
    pub fn offer(&mut self) -> Result<BTreeMap<String, String>, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", crate::bootstrap::OFFER_PATH, "", self.timeout)?;
        if r.status == 404 {
            return Ok(BTreeMap::new());
        }
        crate::bootstrap::parse_offer(&ok_body(r)?)
    }

    /// What the VM hands over as a secret (0.5.0). A VM from before 0.5.0 has
    /// no such path (404): nothing to hand over. An error names the status,
    /// never the body.
    pub fn secret_offer(&mut self) -> Result<crate::bootstrap::SecretOffer, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", crate::bootstrap::SECRET_OFFER_PATH, "", self.timeout)?;
        match r.status {
            404 => Ok(crate::bootstrap::SecretOffer::default()),
            200 => crate::bootstrap::SecretOffer::parse(&r.body),
            s => Err(format!("the secret offer was answered {s}")),
        }
    }

    pub fn close(&mut self) -> Result<(), String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::bootstrap::CLOSE_PATH, "", self.timeout)?;
        ok_body(r).map(|_| ())
    }

    /// **The certificate a CLOSED VM serves** (0.5.12, [`crate::bootstrap::CERT_PATH`]).
    pub fn cert(&mut self) -> Result<String, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", crate::bootstrap::CERT_PATH, "", self.timeout)?;
        ok_body(r)
    }

    /// **Hand a CLOSED VM its renewed certificate** (0.5.12, [`crate::bootstrap::RENEW_PATH`]).
    pub fn renew(&mut self, pem: &str) -> Result<(), String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::bootstrap::RENEW_PATH, pem, self.timeout)?;
        ok_body(r).map(|_| ())
    }

    /// **Hand a CLOSED VM a fresh bootstrap pair** (0.5.13, [`crate::bootstrap::BOOTSTRAP_PATH`]).
    pub fn rebootstrap(&mut self, boot: &Issued) -> Result<(), String> {
        #[derive(Serialize)]
        struct Pair<'a> {
            cert_pem: &'a str,
            key_pem: &'a str,
        }
        let body = toml::to_string(&Pair { cert_pem: &boot.cert_pem, key_pem: &boot.key_pem }).map_err(|e| e.to_string())?;
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::bootstrap::BOOTSTRAP_PATH, &body, self.timeout)?;
        ok_body(r).map(|_| ())
    }

    /// **A later answer to a CLOSED VM** (0.5.8, [`crate::bootstrap::REFEED_PATH`]).
    pub fn refeed(&mut self, reply: &JoinReply) -> Result<(), String> {
        let body = reply.render().map_err(|e| e.to_string())?;
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::bootstrap::REFEED_PATH, &body, self.timeout)?;
        ok_body(r).map(|_| ())
    }

    /// **What a CLOSED VM runs** (live update, [`crate::live::RUNNING_PATH`]).
    pub fn live_running(&mut self) -> Result<crate::live::Running, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "GET", crate::live::RUNNING_PATH, "", self.timeout)?;
        crate::live::Running::parse(&ok_body(r)?)
    }

    /// **A new app for a CLOSED VM** (live update, [`crate::live::STAGE_PATH`]): the VM's
    /// answer, or its refusal by name.
    pub fn live_stage(&mut self, stage: &crate::live::Stage) -> Result<String, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::live::STAGE_PATH, &stage.encode(), self.timeout)?;
        ok_body(r)
    }

    /// [`Client::live_stage`] with what the release declares (lane B5): the VM's answer, or its
    /// refusal by name ([`crate::live::falls_to_keep`] names the ones a keep boot answers).
    pub fn live_stage_declared(&mut self, stage: &crate::live::Stage, declared: &crate::live::Declared) -> Result<String, String> {
        let (r, _) = call(self.path.as_mut(), &self.cfg, "POST", crate::live::STAGE_PATH, &stage.encode_declared(declared), self.timeout)?;
        ok_body(r)
    }

    pub fn gone(&mut self, wait: Duration) -> Result<bool, String> {
        self.path.gone(wait)
    }

    /// **After the close: is the INSTALL gone?** (0.5.4) Yes when the VM
    /// refuses the install by name (`[install-closed]`: its log door is all
    /// that is left, [`crate::bootstrap::respond_closed`]) or answers nothing
    /// at all on the path (a VM that drops its listener with the close).
    pub fn install_gone(&mut self, wait: Duration) -> Result<bool, String> {
        let t0 = std::time::Instant::now();
        while t0.elapsed() < wait {
            match self.hello() {
                Err(e) if e.contains("[install-closed]") => return Ok(true),
                Err(_) => return self.path.gone(wait.saturating_sub(t0.elapsed()).max(Duration::from_secs(1))),
                Ok(_) => std::thread::sleep(Duration::from_millis(250)),
            }
        }
        Ok(false)
    }
}

/// **A VM's log, read by the installer — during the install and for the
/// VM's whole life after it** (0.5.4, owner 2026-09-27: "THE INSTALLER WILL
/// RECEIVE THE LOGS", "standard for all using the installer"). One read from
/// `from`, over `path`, with the installer's own certificate; every line
/// through [`crate::redact`] exactly as [`run`] stores it. A VM that keeps
/// [`crate::bootstrap::serve_for_life`] answers it after the close too.
pub fn read_log(path: Box<dyn Path>, m: &Material, from: u64) -> Result<Chunk, String> {
    let mut c = Client::new(path, &m.ca.cert_pem, &m.me, Duration::from_secs(20))?;
    let mut chunk = c.log(from)?;
    chunk.lines = crate::redact::clean_lines(chunk.lines.iter().map(String::as_str));
    Ok(chunk)
}

/// [`read_log`] for the plan's VM `vm`, over the installer's link.
pub fn log_of(plan: &Plan, m: &Material, vm: &str, from: u64) -> Result<Chunk, String> {
    let v = plan.vms.iter().find(|v| v.name == vm).ok_or_else(|| format!("REFUSED [installer-no-such-vm] the plan has {}, not `{vm}`", plan.vms.iter().map(|v| v.name.as_str()).collect::<Vec<_>>().join(", ")))?;
    read_log(path_to(plan, v, m.link.as_ref())?, m, from)
}

/// **The path to a VM**, by the plan's tunnel kind.
pub fn path_to(plan: &Plan, vm: &PlanVm, link: Option<&LinkKeys>) -> Result<Box<dyn Path>, String> {
    match reach(plan.tunnel).map_err(|e| e.to_string())? {
        Reach::Tunnel => {
            let keys = link.ok_or_else(|| format!("REFUSED [installer-link-keys] a `{}` estate is reached through the installer's link, and the medium's link keys are missing", plan.tunnel.as_str()))?;
            tunnel_path(SocketAddrV4::new(vm.public, plan.link_port), keys.clone())
        }
    }
}

#[cfg(feature = "installer-wireguard")]
fn tunnel_path(endpoint: SocketAddrV4, keys: LinkKeys) -> Result<Box<dyn Path>, String> {
    Ok(Box::new(crate::installer_link::WgPath::new(endpoint, keys)?))
}

#[cfg(not(feature = "installer-wireguard"))]
fn tunnel_path(_endpoint: SocketAddrV4, _keys: LinkKeys) -> Result<Box<dyn Path>, String> {
    Err("REFUSED [installer-link-not-built] this installer was built without its WireGuard link (feature `installer-wireguard`)".into())
}

// ── the run ────────────────────────────────────────────────────────────────

#[derive(Default)]
struct Progress {
    /// Where the installer stands in the VM's log: its boot and next line.
    cursor: crate::bootlog::Cursor,
    log_lines: u64,
    last_line: String,
    signal: Option<Signal>,
    /// The last reply fed, without its certificate.
    fed: Option<String>,
    /// The feed keys (non-blank) of that reply.
    fed_keys: std::collections::BTreeSet<String>,
    role: Option<crate::Role>,
    admitted: bool,
    ready: bool,
    closed: bool,
    gone: bool,
    last_error: Option<String>,
    /// When this VM was first fed ([`LATE_FEED`]'s window runs from it).
    first_fed: Option<Instant>,
    /// Whether its [`LATE_FEED`] keys were said held / released.
    late_said: (bool, bool),
}

fn reply_key(r: &JoinReply) -> String {
    let mut r = r.clone();
    r.cert_pem = None;
    r.render().unwrap_or_default()
}

/// **The FRONT's own feed key that holds the close for its offers** (0.5.9):
/// a comma-separated list of the front's offer keys (e.g. `monetize_pubkey`)
/// that every product box must have been FED (`offer.<front>.<key>`) before
/// anything closes. A front laid again offers its RESTORED monetize key only
/// once it took its state back — a close that began first left every metered
/// box without the key, enforcing no plan. In the front's own feed (not a
/// [`Plan`] field) so every kit's plan stays as it is; the front ignores it.
pub const HOLD_FOR_OFFER_FEED: &str = "installer.boxes_wait_for";

/// **Feed keys held back from every VM's FIRST feed** (2026-10-20, the bench's `fleet-feed-late`):
/// a comma-separated list in the VM's feed (shared or its own). A VM is fed without them on its
/// first visit, and until [`LATE_FEED_S`] seconds after that first feed; its next visit after
/// that feeds them (its answer changed), and nothing closes before it holds them ([`waiting`]
/// compares each VM's last feed with the door's whole answer). It reproduces a box whose pull
/// channel arrives in a LATER feed than its first — the box must take it then, not only at start.
pub const LATE_FEED: &str = "installer.late_feed";
/// The seconds [`LATE_FEED`] is held from a VM's first feed (absent or 0: its second visit).
pub const LATE_FEED_S: &str = "installer.late_feed_s";

/// Pure over the clock: `reply` without its [`LATE_FEED`] keys while they are held for this VM
/// (first fed at `first_fed`; `None` = not fed yet), and whether anything was held.
pub fn late_held(mut reply: JoinReply, first_fed: Option<Instant>, now: Instant) -> (JoinReply, bool) {
    let keys: Vec<String> = reply.feed.get(LATE_FEED).map(|v| v.split(',').map(|k| k.trim().to_string()).filter(|k| !k.is_empty()).collect()).unwrap_or_default();
    if keys.is_empty() {
        return (reply, false);
    }
    let secs = reply.feed.get(LATE_FEED_S).and_then(|v| v.trim().parse::<u64>().ok()).unwrap_or(0);
    let held = match first_fed {
        None => true,
        Some(t) => now.duration_since(t) < Duration::from_secs(secs),
    };
    if !held {
        return (reply, false);
    }
    let mut any = false;
    for k in &keys {
        any |= reply.feed.remove(k).is_some();
    }
    (reply, any)
}

/// **Why the close must wait although every VM says ready**, or `None`
/// (0.5.2) — judged PER VM:
///
/// * a VM whose last feed is not what the door would feed it now (a later
///   offer from another VM) is fed again first — before, the feeding stopped
///   the moment the close began, whatever arrived after the VM's last visit;
/// * the front must have been FED each [`Plan::front_waits_for`] key as the
///   elected primary's offer (`offer.<primary>.<key>`).
fn waiting(plan: &Plan, door: &Door, p: &BTreeMap<String, Progress>, names: &[String]) -> Option<String> {
    for n in names {
        let Some(fed) = p.get(n).and_then(|e| e.fed.as_deref()) else { continue };
        if door.ledger.admitted.contains_key(n) && fed != reply_key(&door.reply_for(n)) {
            return Some(format!("`{n}` was last fed before a later offer — it is fed again first"));
        }
    }
    // 0.5.9: what the FRONT's own feed says every product box must hold of its offer.
    for front in plan.vms.iter().filter(|v| v.kind == Kind::Front) {
        let Some(keys) = front.feed.get(HOLD_FOR_OFFER_FEED) else { continue };
        for k in keys.split(',').map(str::trim).filter(|k| !k.is_empty()) {
            let key = crate::bootstrap::fed_offer_key(&front.name, k);
            for b in plan.vms.iter().filter(|v| v.kind != Kind::Front && names.contains(&v.name)) {
                if !p.get(&b.name).is_some_and(|e| e.fed_keys.contains(&key)) {
                    return Some(format!("`{}` has not been fed `{key}` (the front's offer) yet", b.name));
                }
            }
        }
    }
    if plan.front_waits_for.is_empty() {
        return None;
    }
    let Some(primary) = door.verdict.as_ref().map(|v| v.primary.clone()) else {
        return Some("no primary is elected yet".into());
    };
    for front in plan.vms.iter().filter(|v| v.kind == Kind::Front) {
        let fed = p.get(&front.name).map(|e| &e.fed_keys);
        for k in &plan.front_waits_for {
            let key = crate::bootstrap::fed_offer_key(&primary, k);
            if !fed.is_some_and(|f| f.contains(&key)) {
                return Some(format!("the front `{}` has not been fed `{key}` (the primary's offer) yet", front.name));
            }
        }
    }
    None
}

/// Every admitted gunnar box has a tunnel key and the verdict is in. An
/// estate of ONE — a front and no product box (the monetize-admin box) — has
/// nothing to elect: it is complete once admitted.
fn peers_complete(d: &Door, tunnel: TunnelKind) -> bool {
    let boxes = d.ledger.slots.values().any(|s| s.kind != Kind::Front);
    d.ledger.complete() && (d.verdict.is_some() || !boxes) && (tunnel == TunnelKind::None || d.ledger.admitted.values().all(|a| !a.tunnel_pubkey.is_empty()))
}

struct Shared {
    door: Mutex<Door>,
    progress: Mutex<BTreeMap<String, Progress>>,
    closing: AtomicBool,
    over: AtomicBool,
}

/// **Admit and feed**: one visit's door work for a presentation.
fn admit(shared: &Shared, sink: &dyn Sink, vm: &PlanVm, req: &JoinRequest, who: &Presented) -> Result<JoinReply, String> {
    let mut door = shared.door.lock().unwrap_or_else(|p| p.into_inner());
    let before = door.clone();
    // Was this VM's slot admitted before this hello? (The ledger's slots are
    // keyed by the address the installer dials.)
    let first = !before.ledger.slots.get(&vm.public).is_some_and(|s| before.ledger.admitted.contains_key(&s.name));
    let reply = match door.join(req, who, vm.public) {
        Ok(r) => r,
        // The installer DIALLED this address itself, so a VM that speaks the
        // bootstrap certificate on an admitted slot is that VM, re-installed
        // or restarted before it kept what it was fed: admitted again.
        Err(JoinError::Spent { .. } | JoinError::Withdrawn) if *who == Presented::Bootstrap && door.ledger.slots.contains_key(&vm.public) => {
            let name = door.ledger.slots[&vm.public].name.clone();
            sink.say(&format!("{name}: speaks the bootstrap certificate again (re-installed, or it lost its feed) — admitted afresh"));
            door.forget(&name).map_err(|e| e.to_string())?;
            door.join(req, who, vm.public).map_err(|e| e.to_string())?
        }
        // 0.5.26: a box of an IN-PLACE re-admission that shows its own certificate — fed by an
        // earlier run of it that ended before the close (gunnar.rs twin, 2026-10-25) — is that
        // box: admitted again as itself ([`Door::rejoin_in_place`]).
        Err(JoinError::Forged(why)) => match who {
            Presented::Box(cn) if door.in_place.contains(cn) => {
                let r = door.rejoin_in_place(req, cn, vm.public).map_err(|e| format!("{e} ({why})"))?;
                sink.say(&format!("{cn}: shows its own certificate from an earlier run of this re-admission that never closed — admitted again as itself, tunnel key {}…", req.tunnel_pubkey.chars().take(8).collect::<String>()));
                r
            }
            _ => return Err(JoinError::Forged(why).to_string()),
        },
        Err(e) => return Err(e.to_string()),
    };
    if *door != before {
        sink.ledger(&door);
    }
    // Owner 2026-09-30: every pairing step is its own line in the installer's
    // log — the hello and the admission, once per VM; the election, once.
    if first && door.ledger.admitted.contains_key(&reply.me.name) {
        let cap = req.capacity.as_ref().map(|c| format!(", {} core(s), {} MiB/s disk", c.cores, c.disk_mib_s)).unwrap_or_default();
        sink.say(&format!(
            "{}: hello — a {:?} box{cap}, tunnel key {}…; admitted into slot `{}` from {} on the {}",
            vm.name,
            req.kind,
            req.tunnel_pubkey.chars().take(8).collect::<String>(),
            reply.me.name,
            vm.public,
            match who {
                Presented::Bootstrap => "bootstrap certificate",
                _ => "box's own certificate",
            }
        ));
    }
    if let (None, Some(v)) = (&before.verdict, &door.verdict) {
        sink.say(&format!("the election: primary `{}`, twin `{}` — {}", v.primary, v.twin, v.why));
    }
    Ok(reply)
}

fn store_lines(shared: &Shared, sink: &dyn Sink, name: &str, chunk: &Chunk) {
    let lines: Vec<String> = crate::redact::clean_lines(chunk.stored().iter().map(String::as_str));
    if lines.is_empty() {
        return;
    }
    sink.lines(name, &lines);
    let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let e = p.entry(name.to_string()).or_default();
    e.log_lines += lines.len() as u64;
    if let Some(l) = lines.last() {
        e.last_line = l.clone();
    }
}

/// The line stored where a VM's log started over (it restarted): the
/// reader reads the new boot from its first line.
pub const STARTED_OVER: &str = "(the VM's log started over — a new boot; reading it from its first line)";

/// Everything the VM has logged since the cursor.
fn drain_log(shared: &Shared, sink: &dyn Sink, c: &mut Client, name: &str) -> Result<(), String> {
    loop {
        let at = shared.progress.lock().unwrap_or_else(|p| p.into_inner()).get(name).map(|p| p.cursor).unwrap_or_default();
        let chunk = c.log(at.next)?;
        let (next, restarted) = crate::bootlog::resume(at, &chunk);
        if restarted {
            // A new boot (a restart mid-install): its lines from line 0, the
            // chunk read from the old cursor is not its start.
            sink.lines(name, &[STARTED_OVER.to_string()]);
            shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(name.to_string()).or_default().cursor = next;
            continue;
        }
        store_lines(shared, sink, name, &chunk);
        shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(name.to_string()).or_default().cursor = next;
        if chunk.lines.len() < MAX_READ {
            return Ok(());
        }
    }
}

/// **The VM's offer, fed onward** (0.3.1): every offered value goes into the
/// door's feed as `offer.<vm>.<key>` — so the next answer to every other VM
/// differs and is fed again — and the sink hears the offer when it changed.
fn take_offer(shared: &Shared, sink: &dyn Sink, c: &mut Client, name: &str) -> Result<(), String> {
    let offer = c.offer()?;
    if offer.is_empty() {
        return Ok(());
    }
    let mut door = shared.door.lock().unwrap_or_else(|p| p.into_inner());
    let mut changed = false;
    for (k, v) in &offer {
        let key = crate::bootstrap::fed_offer_key(name, k);
        if door.feed.get(&key) != Some(v) {
            door.feed.insert(key, v.clone());
            changed = true;
        }
    }
    if changed {
        sink.ledger(&door);
        drop(door);
        sink.say(&format!("{name}: offers {} — fed to the other VMs", offer.keys().cloned().collect::<Vec<_>>().join(", ")));
        sink.offer(name, &offer);
    }
    Ok(())
}

/// One turn with one VM: present, admit, feed, logs, offer, signal.
fn visit(shared: &Shared, plan: &Plan, sink: &dyn Sink, c: &mut Client, vm: &PlanVm) -> Result<(), String> {
    let (req, who) = c.hello()?;
    if req.estate != plan.estate {
        return Err(format!("REFUSED [installer-estate] the VM at {} is of `{}`, this run installs `{}`", vm.public, req.estate, plan.estate));
    }
    let reply = admit(shared, sink, vm, &req, &who)?;
    // The late keys held back from this VM's first feed (the bench's `fleet-feed-late`).
    let first_fed = shared.progress.lock().unwrap_or_else(|p| p.into_inner()).get(&vm.name).and_then(|p| p.first_fed);
    let (reply, held) = late_held(reply, first_fed, Instant::now());
    {
        let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
        let e = p.entry(vm.name.clone()).or_default();
        let keys = reply.feed.get(LATE_FEED).cloned().unwrap_or_default();
        if held && !e.late_said.0 {
            e.late_said.0 = true;
            drop(p);
            sink.say(&format!("{}: fed WITHOUT {keys} this time ({LATE_FEED}: a later feed carries them)", vm.name));
        } else if !held && e.late_said.0 && !e.late_said.1 {
            e.late_said.1 = true;
            drop(p);
            sink.say(&format!("{}: {keys} released — fed in this later feed", vm.name));
        }
    }
    let key = reply_key(&reply);
    let must = reply.cert_pem.is_some() || shared.progress.lock().unwrap_or_else(|p| p.into_inner()).get(&vm.name).and_then(|p| p.fed.clone()).as_deref() != Some(key.as_str());
    if must {
        c.feed(&reply)?;
        shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(vm.name.clone()).or_default().first_fed.get_or_insert_with(Instant::now);
        sink.say(&format!(
            "{}: fed — `{}` at {}, role {}, {} peer(s), feed {}",
            vm.name,
            reply.me.name,
            reply.me.overlay,
            reply.role.map(|r| r.as_str()).unwrap_or("not elected yet"),
            reply.peers.len(),
            reply.feed.keys().cloned().collect::<Vec<_>>().join(", ")
        ));
    }
    drain_log(shared, sink, c, &vm.name)?;
    take_offer(shared, sink, c, &vm.name)?;
    let signal = c.status().ok();
    let complete = peers_complete(&shared.door.lock().unwrap_or_else(|p| p.into_inner()), plan.tunnel);
    let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let e = p.entry(vm.name.clone()).or_default();
    e.admitted = true;
    e.fed = Some(key);
    e.fed_keys = reply.feed.iter().filter(|(_, v)| !v.trim().is_empty()).map(|(k, _)| k.clone()).collect();
    e.role = reply.role;
    let changed = e.signal.as_ref().map(|s| s.line()) != signal.as_ref().map(|s| s.line());
    e.signal = signal;
    e.ready = match plan.close_when {
        CloseWhen::Paired => e.signal.as_ref().map(|s| s.paired).unwrap_or(false),
        CloseWhen::Fed => reply.role.is_some() && complete,
    };
    e.last_error = None;
    if changed {
        sink.status(&vm.name, e.signal.as_ref(), &e.last_line);
    }
    Ok(())
}

/// The close, with the last lines first and the proof after.
fn close_one(shared: &Shared, sink: &dyn Sink, c: &mut Client, name: &str) -> Result<(), String> {
    drain_log(shared, sink, c, name)?;
    // 0.5.0: the hand-over — at the END of the install, before the close
    // (after it the listener is gone for good). To the sink's store ONLY:
    // not the door's feed, not the ledger, not a log; the narration names
    // the keys. A failure here fails the close, which is tried again.
    let secret = c.secret_offer()?;
    if !secret.is_empty() {
        sink.secret_offer(name, &secret).map_err(|e| format!("REFUSED [installer-secret-store] {name} handed over {} but it could not be stored: {e}", secret.keys().join(", ")))?;
        sink.say(&format!("{name}: handed over {} secret value(s) for its successor ({}) — kept in the operator's store, never fed on, never logged", secret.len(), secret.keys().join(", ")));
    }
    c.close()?;
    let gone = c.install_gone(Duration::from_secs(8)).unwrap_or(false);
    let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let e = p.entry(name.to_string()).or_default();
    e.closed = true;
    e.gone = gone;
    sink.say(&format!(
        "{name}: closed — the bootstrap certificate is dead there; {}",
        if gone { "the VM no longer answers the install — its log door is all that is left (verified)" } else { "BUT the VM still answered the install after the close" }
    ));
    Ok(())
}

/// **One turn with a CLOSED VM** (0.5.8, [`readmit`]): it is re-fed the
/// door's answer for it whenever that changed (a re-admitted box's new
/// address and key, a new offer), and its log is read on. Nothing else of the
/// install is asked: its listener is the log door.
fn refeed_visit(shared: &Shared, sink: &dyn Sink, c: &mut Client, name: &str) -> Result<(), String> {
    let reply = shared.door.lock().unwrap_or_else(|p| p.into_inner()).reply_for(name);
    let key = reply_key(&reply);
    let last = shared.progress.lock().unwrap_or_else(|p| p.into_inner()).get(name).and_then(|p| p.fed.clone());
    if last.as_deref() != Some(key.as_str()) {
        c.refeed(&reply)?;
        sink.say(&format!(
            "{name}: re-fed (its install was closed before) — role {}, primary at {}, {} peer(s)",
            reply.role.map(|r| r.as_str()).unwrap_or("none"),
            reply.primary_utility.map(|a| a.to_string()).unwrap_or_else(|| "-".into()),
            reply.peers.len()
        ));
    }
    drain_log(shared, sink, c, name)?;
    let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let e = p.entry(name.to_string()).or_default();
    e.admitted = true;
    e.fed = Some(key);
    e.fed_keys = reply.feed.iter().filter(|(_, v)| !v.trim().is_empty()).map(|(k, _)| k.clone()).collect();
    e.role = reply.role;
    e.ready = true;
    e.last_error = None;
    Ok(())
}

/// The end of a re-admission for a CLOSED VM: it holds the last answer, and
/// its install is still gone (proved like a close).
fn settle_refed(shared: &Shared, sink: &dyn Sink, c: &mut Client, name: &str) -> Result<(), String> {
    refeed_visit(shared, sink, c, name)?;
    let gone = c.install_gone(Duration::from_secs(8)).unwrap_or(false);
    let mut p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let e = p.entry(name.to_string()).or_default();
    e.closed = true;
    e.gone = gone;
    sink.say(&format!("{name}: holds the estate's last answer; {}", if gone { "its install stays closed (verified)" } else { "BUT it answered the install" }));
    Ok(())
}

/// **Re-admit ONE box into a CLOSED estate** (0.5.8) — a box that died and was
/// laid again from the same medium: the dead main of gunnar or holger, whose
/// twin holds everything.
///
/// `prior` is the door the install left (its run folder's `ledger.toml`,
/// [`Door::load`]): the ledger, the verdict and the offers. `plan` is the
/// estate's plan as the install ran it, with `name` at its NEW addresses and
/// every other VM where it was (anything else moving is refused by name), and
/// `paths` reach every VM of the plan. Then:
///
/// 1. [`Door::readmit`] moves `name`'s slot, forgets the dead admission and
///    keeps the verdict; a re-admitted primary is fed
///    [`crate::door::RECHARGE_FROM_FEED`] (where its twin is);
/// 2. `name` is INSTALLED like any VM of a run: presented, admitted on the
///    bootstrap certificate, fed its role back, its log collected, its
///    offers fed on, and closed when ready ([`Plan::close_when`]);
/// 3. every OTHER VM — closed by the install — is RE-FED
///    ([`Client::refeed`]) whenever its answer changes: the new box's
///    address and tunnel key (the front's relay lists it, its routes dial it),
///    the new box's offers.
///
/// Returns what the door did and the run's outcome.
pub fn readmit(plan: &Plan, prior: Door, m: &Material, name: &str, paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink) -> Result<(crate::door::Readmitted, Outcome), String> {
    let (mut r, out) = readmit_how(plan, prior, m, &[name.to_string()], paths, sink, false)?;
    Ok((r.remove(0), out))
}

/// **Renew the boxes' own certificates that are due** (0.5.12): for each CLOSED VM on its path,
/// read the certificate it serves; at two thirds of its life ([`crate::mtls::renewal_due`] at
/// `now_unix`) re-sign ITS key for `valid_for` and hand it over. Per VM: `Ok(true)` renewed,
/// `Ok(false)` not due yet, `Err` what refused it (the others are still tried).
pub fn renew(m: &Material, paths: Vec<(String, Box<dyn Path>)>, valid_for: Duration, now_unix: i64) -> Vec<(String, Result<bool, String>)> {
    paths
        .into_iter()
        .map(|(name, path)| {
            let r = (|| {
                let mut c = Client::new(path, &m.ca.cert_pem, &m.me, Duration::from_secs(20))?;
                let current = c.cert()?;
                if !crate::mtls::renewal_due(&current, now_unix).map_err(|e| e.to_string())? {
                    return Ok(false);
                }
                let renewed = m.ca.renew_box(&current, valid_for).map_err(|e| e.to_string())?;
                c.renew(&renewed)?;
                // The seed's bootstrap ends with the medium's material: a box upgraded in place
                // re-joins with it, so every renewal hands a fresh one (0.5.13).
                let boot = m.ca.issue_bootstrap(valid_for).map_err(|e| e.to_string())?;
                c.rebootstrap(&boot).map_err(|e| format!("renewed, but the fresh bootstrap for its next join was refused: {e}"))?;
                Ok(true)
            })();
            (name, r)
        })
        .collect()
}

/// **[`readmit`] for a box UPGRADED IN PLACE** (0.5.9, [`crate::door::Door::readmit_kept`]):
/// the plan names it at the address it has, its data was kept, a primary is
/// filled from nobody.
pub fn readmit_kept(plan: &Plan, prior: Door, m: &Material, name: &str, paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink) -> Result<(crate::door::Readmitted, Outcome), String> {
    let (mut r, out) = readmit_how(plan, prior, m, &[name.to_string()], paths, sink, true)?;
    Ok((r.remove(0), out))
}

/// **[`readmit_kept`] for SEVERAL boxes upgraded in place, in ONE run** (0.5.26, gunnar.rs
/// 2026-10-25): every box named is installed again (presented, admitted, fed, closed) and only
/// the others are RE-FED. A re-admission of one box re-feeds every other one, and a box whose
/// own keep boot was never re-admitted answers a re-feed with its OPEN install listener
/// (`[install-path]`) — so the re-admission of the twin waited for the main until its budget
/// (`REFUSED [installer-incomplete] … main: never answered (last: REFUSED [install-path] …)`).
/// Two boxes in that state are re-admitted together. A named box that shows its own
/// certificate from an earlier run of this re-admission is admitted as itself
/// ([`crate::door::Door::rejoin_in_place`]).
pub fn readmit_kept_together(plan: &Plan, prior: Door, m: &Material, names: &[String], paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink) -> Result<(Vec<crate::door::Readmitted>, Outcome), String> {
    readmit_how(plan, prior, m, names, paths, sink, true)
}

fn readmit_how(plan: &Plan, prior: Door, m: &Material, names: &[String], paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink, kept: bool) -> Result<(Vec<crate::door::Readmitted>, Outcome), String> {
    plan.check()?;
    let bad = |why: String| Err(format!("REFUSED [installer-readmit] {why}"));
    if prior.ledger.estate != plan.estate {
        return bad(format!("the kept door is of `{}`, the plan installs `{}`", prior.ledger.estate, plan.estate));
    }
    if names.is_empty() {
        return bad("no box is named to re-admit".into());
    }
    let named = |n: &str| names.iter().any(|x| x == n);
    for name in names {
        if !plan.vms.iter().any(|v| v.name == *name) {
            return bad(format!("the plan has {}, not `{name}`", plan.vms.iter().map(|v| v.name.as_str()).collect::<Vec<_>>().join(", ")));
        }
    }
    if !kept && names.len() > 1 {
        return bad(format!("a re-admission at new addresses moves ONE box, not {}", names.join(" and ")));
    }
    let one = names.join(" and ");
    for v in plan.vms.iter().filter(|v| !named(&v.name)) {
        match prior.ledger.slot_of(&v.name) {
            Some((at, _)) if *at == v.public && prior.ledger.admitted.contains_key(&v.name) => {}
            Some((at, _)) if *at != v.public => return bad(format!("`{}` was installed at {at} and the plan has it at {}; a re-admission moves ONE box (`{one}`)", v.name, v.public)),
            Some(_) => return bad(format!("`{}` never joined this estate; a re-admission is for a CLOSED estate", v.name)),
            None => return bad(format!("`{}` is not a box of the kept door", v.name)),
        }
    }
    let mut door = prior.with_ca(m.ca.clone());
    if let Some(s) = plan.box_cert_valid_s {
        door = door.with_box_cert_valid(Duration::from_secs(s));
    }
    door.own_feed = plan.vms.iter().filter(|v| !v.feed.is_empty()).map(|v| (v.name.clone(), v.feed.clone())).collect();
    door.feed.extend(plan.feed.iter().map(|(k, v)| (k.clone(), v.clone())));
    if plan.front.is_some() {
        door.front = plan.front.clone();
    }
    let mut done = Vec::new();
    for name in names {
        let vm = plan.vms.iter().find(|v| v.name == *name).expect("checked above");
        let r = if kept {
            if door.ledger.slot_of(name).map(|(at, _)| *at) != Some(vm.public) {
                return bad(format!("`{name}` was upgraded IN PLACE and the plan moves it to {}: an upgrade keeps the server and its address", vm.public));
            }
            door.readmit_kept(name)?
        } else {
            door.readmit(name, vm.public, vm.utility)?
        };
        sink.say(&format!(
            "{name}: re-admitted — {} (utility {}) → {} (utility {}), its role kept{}",
            r.was.0,
            r.was.1,
            r.now.0,
            r.now.1,
            r.recharge_from.as_deref().map(|t| format!(", fed {} = {t}", crate::door::RECHARGE_FROM_FEED)).unwrap_or_default()
        ));
        done.push(r);
    }
    let refed: std::collections::BTreeSet<String> = plan.vms.iter().filter(|v| !named(&v.name)).map(|v| v.name.clone()).collect();
    let out = run_on(plan, door, m, paths, sink, &refed)?;
    Ok((done, out))
}

fn backoff(tries: u32) -> Duration {
    Duration::from_secs([1u64, 2, 3, 5, 8, 10][tries.min(5) as usize])
}

/// **Run the installer** against every VM of the plan, until each is closed
/// or the budget ends. `paths` gives each VM's path (by name) — the caller's
/// ([`path_to`] in production, loopback in tests).
pub fn run(plan: &Plan, m: &Material, paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink) -> Result<Outcome, String> {
    plan.check()?;
    let mut door = plan.door().with_ca(m.ca.clone());
    if let Some(s) = plan.box_cert_valid_s {
        door = door.with_box_cert_valid(Duration::from_secs(s));
    }
    run_on(plan, door, m, paths, sink, &std::collections::BTreeSet::new())
}

/// The run over `door`: every VM installed, except those named in `refed` —
/// VMs CLOSED by an earlier install, which are only RE-FED ([`readmit`]).
fn run_on(plan: &Plan, door: Door, m: &Material, paths: Vec<(PlanVm, Box<dyn Path>)>, sink: &dyn Sink, refed: &std::collections::BTreeSet<String>) -> Result<Outcome, String> {
    let started = Instant::now();
    let budget = Duration::from_secs(plan.budget_s);
    let shared = Shared { door: Mutex::new(door), progress: Mutex::new(BTreeMap::new()), closing: AtomicBool::new(false), over: AtomicBool::new(false) };
    sink.ledger(&shared.door.lock().unwrap_or_else(|p| p.into_inner()));
    let names: Vec<String> = paths.iter().map(|(v, _)| v.name.clone()).collect();
    std::thread::scope(|s| {
        for (vm, path) in paths {
            let shared = &shared;
            s.spawn(move || {
                let mut c = match Client::new(path, &m.ca.cert_pem, &m.me, Duration::from_secs(20)) {
                    Ok(c) => c,
                    Err(e) => {
                        shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(vm.name.clone()).or_default().last_error = Some(e);
                        return;
                    }
                };
                sink.say(&format!("{}: reaching {}", vm.name, c.describe()));
                let mut tries = 0u32;
                let mut said_waiting = Instant::now() - Duration::from_secs(3600);
                while !shared.over.load(Ordering::SeqCst) {
                    let refeed = refed.contains(&vm.name);
                    if shared.closing.load(Ordering::SeqCst) {
                        let closed = if refeed { settle_refed(shared, sink, &mut c, &vm.name) } else { close_one(shared, sink, &mut c, &vm.name) };
                        match closed {
                            Ok(()) => return,
                            Err(e) => {
                                shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(vm.name.clone()).or_default().last_error = Some(e);
                                tries += 1;
                                std::thread::sleep(backoff(tries));
                                continue;
                            }
                        }
                    }
                    let visited = if refeed { refeed_visit(shared, sink, &mut c, &vm.name) } else { visit(shared, plan, sink, &mut c, &vm) };
                    match visited {
                        Ok(()) => {
                            tries = 0;
                            std::thread::sleep(Duration::from_secs(2));
                        }
                        Err(e) => {
                            if said_waiting.elapsed() > Duration::from_secs(30) {
                                sink.say(&format!("{}: not yet — {}", vm.name, e.lines().next().unwrap_or("")));
                                said_waiting = Instant::now();
                            }
                            shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(vm.name.clone()).or_default().last_error = Some(e);
                            tries += 1;
                            std::thread::sleep(backoff(tries));
                        }
                    }
                }
            });
        }
        // The judge: everybody ready → close; everybody closed → done; the
        // budget → over.
        let mut last_wait: Option<String> = None;
        loop {
            std::thread::sleep(Duration::from_millis(250));
            let p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
            let all = |f: &dyn Fn(&Progress) -> bool| names.iter().all(|n| p.get(n).map(f).unwrap_or(false));
            if all(&|e| e.closed) {
                shared.over.store(true, Ordering::SeqCst);
                break;
            }
            if !shared.closing.load(Ordering::SeqCst) && all(&|e| e.ready) {
                drop(p);
                // Judged PER VM under the door's lock (0.5.2), so no offer can
                // land between the judgement and the close: every VM holds what
                // it would be fed now, and the front what `front_waits_for` names.
                let door = shared.door.lock().unwrap_or_else(|p| p.into_inner());
                let p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
                if let Some(why) = waiting(plan, &door, &p, &names) {
                    drop(p);
                    drop(door);
                    if last_wait.as_deref() != Some(why.as_str()) {
                        sink.say(&format!("not closing yet: {why}"));
                        last_wait = Some(why);
                    }
                    // Falls through to the budget: a wait is never for ever.
                } else {
                    shared.closing.store(true, Ordering::SeqCst);
                    drop(p);
                    drop(door);
                    sink.say(&format!("every VM is {} — closing the bootstrap listeners", match plan.close_when { CloseWhen::Paired => "paired", CloseWhen::Fed => "fed" }));
                    continue;
                }
            }
            if started.elapsed() > budget {
                shared.over.store(true, Ordering::SeqCst);
                break;
            }
        }
    });
    let p = shared.progress.lock().unwrap_or_else(|p| p.into_inner());
    let vms: Vec<VmOutcome> = names
        .iter()
        .map(|n| {
            let e = p.get(n);
            VmOutcome {
                name: n.clone(),
                admitted: e.map(|e| e.admitted).unwrap_or(false),
                role: e.and_then(|e| e.role),
                log_lines: e.map(|e| e.log_lines).unwrap_or(0),
                last_line: e.map(|e| e.last_line.clone()).unwrap_or_default(),
                signal: e.and_then(|e| e.signal.as_ref().map(|s| s.line().trim().to_string())),
                closed: e.map(|e| e.closed).unwrap_or(false),
                gone: e.map(|e| e.gone).unwrap_or(false),
                last_error: e.and_then(|e| e.last_error.clone()),
                fed_keys: e.map(|e| e.fed_keys.iter().cloned().collect()).unwrap_or_default(),
            }
        })
        .collect();
    let out = Outcome { vms, took: started.elapsed() };
    if out.vms.iter().all(|v| v.closed) {
        Ok(out)
    } else {
        Err(format!(
            "REFUSED [installer-incomplete] after {} s: {}",
            out.took.as_secs(),
            out.vms
                .iter()
                .map(|v| format!(
                    "{}: {}{}",
                    v.name,
                    if !v.admitted { "never answered".to_string() } else if v.closed { "closed".to_string() } else { v.signal.clone().unwrap_or_else(|| "no signal".into()) },
                    v.last_error.as_deref().map(|e| format!(" (last: {})", e.lines().next().unwrap_or(""))).unwrap_or_default()
                ))
                .collect::<Vec<_>>()
                .join("; ")
        ))
    }
}

/// A [`Sink`] that writes a run folder: `<dir>/logs/<vm>.log` (appended),
/// `<dir>/status.toml` (each VM's signal and last line), `<dir>/ledger.toml`
/// (the door, whole, renamed into place), `<dir>/offers.toml` (each VM's
/// offer, 0.3.1 — [`read_offers`]), `<dir>/secret-offers.toml` (what each VM
/// handed over as a secret, 0.5.0, **0600** — [`read_secret_offers`],
/// [`successor_feed`]) and `<dir>/installer.log`.
pub struct FolderSink {
    dir: PathBuf,
    status: Mutex<BTreeMap<String, (String, String)>>,
    offers: Mutex<BTreeMap<String, BTreeMap<String, String>>>,
    secrets: Mutex<BTreeMap<String, BTreeMap<String, String>>>,
    echo: bool,
}

/// The file [`FolderSink`] keeps every VM's offer in.
pub const OFFERS_FILE: &str = "offers.toml";

/// **The run folder's secret store** (0.5.0): what each VM handed over as a
/// secret, `[<vm>] <key> = "<value>"`, mode 0600, written whole and renamed.
/// The only file of the run that holds a private key.
pub const SECRET_OFFERS_FILE: &str = "secret-offers.toml";

/// **What every VM of a run handed over as a secret**, as its folder keeps it.
/// A folder with none is an empty map. Errors never quote the file.
pub fn read_secret_offers(dir: &FsPath) -> Result<BTreeMap<String, crate::bootstrap::SecretOffer>, String> {
    let path = dir.join(SECRET_OFFERS_FILE);
    match std::fs::read_to_string(&path) {
        Ok(t) => {
            let all: BTreeMap<String, BTreeMap<String, String>> = toml::from_str(&t).map_err(|_| format!("{}: not a secret store (its content is not quoted)", path.display()))?;
            Ok(all.into_iter().map(|(vm, m)| (vm, crate::bootstrap::SecretOffer::new(m))).collect())
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(BTreeMap::new()),
        Err(e) => Err(format!("{}: {e}", path.display())),
    }
}

/// **What the successor of `vm` is fed** (0.5.0): everything `vm` handed
/// over as a secret in the run kept at `previous_run` — e.g. the front's
/// `tls.<host>.cert` + `tls.<host>.key`. Put it in the new run's
/// [`PlanVm::feed`] of the VM that takes `vm`'s place (never [`Plan::feed`]:
/// that reaches every VM). Empty when that run kept nothing for `vm`.
pub fn successor_feed(previous_run: &FsPath, vm: &str) -> Result<BTreeMap<String, String>, String> {
    Ok(read_secret_offers(previous_run)?.remove(vm).map(|s| s.into_inner()).unwrap_or_default())
}

/// Write `text` to `path` whole, created 0600 BEFORE a byte is written, then
/// renamed into place, durably (0.5.17, G4: edda appliance-pid1 `durable::write_atomic`).
fn write_0600(path: &FsPath, text: &str) -> Result<(), String> {
    appliance_pid1::durable::write_atomic(path, text.as_bytes(), Some(0o600)).map_err(|e| format!("{}: {e}", path.display()))
}

/// **Every VM's offer, as a run folder holds it** (`[<vm>] <key> = "<value>"`)
/// — e.g. `read_offers(dir)?["main"]["ticket_pubkey"]`. A folder with no
/// offer yet is an empty map.
pub fn read_offers(dir: &FsPath) -> Result<BTreeMap<String, BTreeMap<String, String>>, String> {
    match std::fs::read_to_string(dir.join(OFFERS_FILE)) {
        Ok(t) => toml::from_str(&t).map_err(|e| format!("{}: {e}", dir.join(OFFERS_FILE).display())),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(BTreeMap::new()),
        Err(e) => Err(format!("{}: {e}", dir.join(OFFERS_FILE).display())),
    }
}

impl FolderSink {
    pub fn new(dir: &FsPath, echo: bool) -> Result<FolderSink, String> {
        std::fs::create_dir_all(dir.join("logs")).map_err(|e| format!("{}: {e}", dir.display()))?;
        Ok(FolderSink { dir: dir.to_path_buf(), status: Mutex::new(BTreeMap::new()), offers: Mutex::new(BTreeMap::new()), secrets: Mutex::new(BTreeMap::new()), echo })
    }

    fn append(&self, file: PathBuf, text: &str) {
        if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(&file) {
            let _ = f.write_all(text.as_bytes());
        }
    }

    fn write_status(&self) {
        let s = self.status.lock().unwrap_or_else(|p| p.into_inner());
        let mut out = String::new();
        for (vm, (signal, last)) in s.iter() {
            out.push_str(&format!("[{vm}]\nsignal = {}\nlast_line = {}\n\n", toml_str(signal), toml_str(last)));
        }
        // Whole and durably (0.5.17, G4), as the ledger is.
        let _ = appliance_pid1::durable::write_atomic(&self.dir.join("status.toml"), out.as_bytes(), None);
    }
}

fn toml_str(s: &str) -> String {
    toml::Value::String(s.to_string()).to_string()
}

impl Sink for FolderSink {
    fn lines(&self, vm: &str, lines: &[String]) {
        let mut text = lines.join("\n");
        text.push('\n');
        self.append(self.dir.join("logs").join(format!("{vm}.log")), &text);
        if let Some(l) = lines.last() {
            let mut s = self.status.lock().unwrap_or_else(|p| p.into_inner());
            s.entry(vm.to_string()).or_default().1 = l.clone();
        }
        self.write_status();
    }

    fn status(&self, vm: &str, signal: Option<&Signal>, last_line: &str) {
        {
            let mut s = self.status.lock().unwrap_or_else(|p| p.into_inner());
            let e = s.entry(vm.to_string()).or_default();
            e.0 = signal.map(|s| s.line().trim().to_string()).unwrap_or_default();
            e.1 = last_line.to_string();
        }
        self.write_status();
        if let Some(sig) = signal {
            self.say(&format!("{vm}: {}", sig.line().trim()));
        }
    }

    fn ledger(&self, door: &Door) {
        let _ = door.save(&self.dir.join("ledger.toml"));
    }

    fn offer(&self, vm: &str, offer: &BTreeMap<String, String>) {
        let mut all = self.offers.lock().unwrap_or_else(|p| p.into_inner());
        all.insert(vm.to_string(), offer.clone());
        let _ = appliance_pid1::durable::write_atomic(&self.dir.join(OFFERS_FILE), toml::to_string(&*all).unwrap_or_default().as_bytes(), None);
    }

    fn secret_offer(&self, vm: &str, offer: &crate::bootstrap::SecretOffer) -> Result<(), String> {
        let mut all = self.secrets.lock().unwrap_or_else(|p| p.into_inner());
        all.insert(vm.to_string(), offer.expose().clone());
        let text = toml::to_string(&*all).map_err(|_| "the secret store could not be rendered".to_string())?;
        write_0600(&self.dir.join(SECRET_OFFERS_FILE), &text)
    }

    fn say(&self, line: &str) {
        if self.echo {
            println!("   installer  {line}");
            let _ = std::io::stdout().flush();
        }
        self.append(self.dir.join("installer.log"), &format!("{line}\n"));
    }
}

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

    /// **An estate of ONE — a front and no product box — is complete when it
    /// is admitted** (the monetize-admin box, 2026-09-29): no box means no
    /// election, so no verdict can ever come, and waiting for one kept the
    /// install open until its budget (MEASURED on the mock: `step await-peers
    /// · not paired`, REFUSED [installer-incomplete] after 1201 s). An estate
    /// with a box still waits for the verdict.
    #[test]
    fn a_front_alone_is_complete_once_admitted_and_a_box_still_waits_for_the_verdict() {
        let slot = |name: &str, kind: Kind| crate::join::Slot { name: name.into(), kind, overlay: crate::OVERLAY_FRONT, endpoint: None, utility: None };
        let admitted = |n: u8| crate::join::Admitted { from: std::net::Ipv4Addr::new(10, 0, 0, n), tunnel_pubkey: "K".into(), capacity: None };
        let mut alone = crate::join::Ledger::new("e");
        alone.lay(std::net::Ipv4Addr::new(10, 0, 0, 1), slot("admin", Kind::Front));
        alone.admitted.insert("admin".into(), admitted(1));
        assert!(peers_complete(&Door::new(alone), TunnelKind::Wireguard), "a front alone: nothing to elect");
        let mut pair = crate::join::Ledger::new("e");
        pair.lay(std::net::Ipv4Addr::new(10, 0, 0, 1), slot("front", Kind::Front));
        pair.lay(std::net::Ipv4Addr::new(10, 0, 0, 2), slot("main", Kind::Gunnar));
        pair.admitted.insert("front".into(), admitted(1));
        pair.admitted.insert("main".into(), admitted(2));
        assert!(!peers_complete(&Door::new(pair), TunnelKind::Wireguard), "a box: the verdict is waited for");
    }
    use crate::bootlog::BootLog;
    use crate::bootstrap::Vm;
    use crate::elect::Capacity;
    use crate::pair::{Event, Machine};
    use std::net::TcpListener;

    const H: Duration = Duration::from_secs(3600);

    /// A VM in-process: its own key, a log, a pairing machine that walks to
    /// PAIRED once it is fed a role.
    struct TestVm {
        estate: String,
        own_key: String,
        cores: u32,
        own_cert: Mutex<Option<String>>,
        machine: Mutex<Machine>,
        closed: AtomicBool,
        log: Arc<BootLog>,
        role: Mutex<Option<crate::Role>>,
        boot: Issued,
        /// What it presents: a gunnar box, or the front (0.5.0 tests).
        kind: Kind,
        /// Every feed it was ever given, whole.
        feeds: Mutex<Vec<BTreeMap<String, String>>>,
        /// What it hands over at the end (the front's certificate and key).
        secret: crate::bootstrap::SecretOffer,
        /// A primary offers its ticket key only after this many offer reads
        /// (0.5.2 tests: the late offer that raced the close).
        offer_after: std::sync::atomic::AtomicU32,
        /// A FRONT offers `monetize_pubkey` only after this many offer reads;
        /// `u32::MAX` = never (0.5.9 tests: the front's late key).
        front_offer_after: std::sync::atomic::AtomicU32,
        /// Every reply it took, whole (fed or re-fed, 0.5.8 tests).
        replies: Mutex<Vec<JoinReply>>,
        /// The fresh bootstrap pair it was handed (0.5.13 tests).
        boot_now: Mutex<Option<Issued>>,
    }

    impl Vm for TestVm {
        fn renewed(&self, cert_pem: &str) -> Result<(), String> {
            *self.own_cert.lock().unwrap() = Some(cert_pem.to_string());
            Ok(())
        }
        fn rebootstrapped(&self, boot: &Issued) -> Result<(), String> {
            *self.boot_now.lock().unwrap() = Some(boot.clone());
            Ok(())
        }
        fn refed(&self, reply: &JoinReply) -> Result<(), String> {
            if let Some(why) = crate::bootstrap::refeed_refusal(self.replies.lock().unwrap().last(), reply) {
                return Err(why);
            }
            self.replies.lock().unwrap().push(reply.clone());
            self.feeds.lock().unwrap().push(reply.feed.clone());
            Ok(())
        }
        fn hello(&self) -> Result<JoinRequest, String> {
            let csr = if self.own_cert.lock().unwrap().is_some() { String::new() } else { mtls::box_request_for(&self.own_key).unwrap().0 };
            let capacity = self.kind.is_product().then_some(Capacity { cores: self.cores, ram_bytes: 1 << 30, data_bytes: 1 << 30, disk_mib_s: 100 });
            Ok(JoinRequest { estate: self.estate.clone(), kind: self.kind, tunnel_pubkey: format!("key-{}", self.cores), capacity, csr_pem: csr })
        }
        fn fed(&self, reply: &JoinReply) -> Result<(), String> {
            if let Some(c) = &reply.cert_pem {
                *self.own_cert.lock().unwrap() = Some(c.clone());
            }
            self.feeds.lock().unwrap().push(reply.feed.clone());
            self.replies.lock().unwrap().push(reply.clone());
            self.log.push(&format!("fed as `{}` role {:?}", reply.me.name, reply.role));
            let mut m = self.machine.lock().unwrap();
            if let (Some(role), crate::pair::State::Booted) = (reply.role, &m.state) {
                m.apply(Event::Joined { role, primary: reply.primary_utility, peers_admitted: 2, peers_expected: 2 });
                m.apply(Event::TunnelUp);
                m.apply(match role {
                    crate::Role::Primary => Event::AdminPresent,
                    _ => Event::TwinConferred,
                });
            }
            // 0.3.1: the twin is certified only with what the PRIMARY offered,
            // carried by the installer (`offer.main.ticket_pubkey`).
            if matches!(m.state, crate::pair::State::TwinConferred { .. }) && crate::bootstrap::offers_in(&reply.feed, "main").contains_key("ticket_pubkey") {
                m.apply(Event::TwinCertified);
                m.apply(Event::Following { applied: 1, head: 1 });
            }
            *self.role.lock().unwrap() = reply.role;
            Ok(())
        }
        fn status(&self) -> Signal {
            Signal::of("vm", crate::seed::TunnelKind::None, &self.machine.lock().unwrap())
        }
        fn close(&self) -> Result<(), String> {
            self.closed.store(true, Ordering::SeqCst);
            Ok(())
        }
        fn identity(&self) -> Issued {
            match self.own_cert.lock().unwrap().clone() {
                Some(c) => Issued { cert_pem: c, key_pem: self.own_key.clone() },
                None => self.boot.clone(),
            }
        }
        fn offer(&self) -> BTreeMap<String, String> {
            match *self.role.lock().unwrap() {
                Some(crate::Role::Primary) => {
                    if self.offer_after.load(Ordering::SeqCst) > 0 {
                        self.offer_after.fetch_sub(1, Ordering::SeqCst);
                        return BTreeMap::new();
                    }
                    BTreeMap::from([("ticket_pubkey".to_string(), format!("ticket-{}", self.cores))])
                }
                Some(crate::Role::Front) => match self.front_offer_after.load(Ordering::SeqCst) {
                    u32::MAX => BTreeMap::new(),
                    0 => BTreeMap::from([("monetize_pubkey".to_string(), "mk-front".to_string())]),
                    _ => {
                        self.front_offer_after.fetch_sub(1, Ordering::SeqCst);
                        BTreeMap::new()
                    }
                },
                _ => BTreeMap::new(),
            }
        }
        fn secret_offer(&self) -> crate::bootstrap::SecretOffer {
            self.secret.clone()
        }
    }

    struct Collect(Mutex<Vec<String>>, Mutex<BTreeMap<String, Vec<String>>>, Mutex<BTreeMap<String, BTreeMap<String, String>>>);
    impl Sink for Collect {
        fn secret_offer(&self, _vm: &str, _offer: &crate::bootstrap::SecretOffer) -> Result<(), String> {
            Ok(())
        }
        fn offer(&self, vm: &str, offer: &BTreeMap<String, String>) {
            self.2.lock().unwrap().insert(vm.into(), offer.clone());
        }
        fn lines(&self, vm: &str, lines: &[String]) {
            self.1.lock().unwrap().entry(vm.into()).or_default().extend(lines.iter().cloned());
        }
        fn status(&self, _vm: &str, _s: Option<&Signal>, _l: &str) {}
        fn ledger(&self, _d: &Door) {}
        fn say(&self, line: &str) {
            self.0.lock().unwrap().push(line.to_string());
        }
    }

    /// A loopback "public address" per VM: the listener on 127.0.0.1:<eph>,
    /// the plan's public address a distinct 127.0.0.x so the ledger keys differ.
    fn spawn_vm(ca: &InstallerCa, boot: &Issued, cores: u32) -> (Arc<TestVm>, SocketAddrV4, std::thread::JoinHandle<()>) {
        spawn_vm_as(ca, boot, cores, Kind::Gunnar, crate::bootstrap::SecretOffer::default())
    }

    fn spawn_vm_as(ca: &InstallerCa, boot: &Issued, cores: u32, kind: Kind, secret: crate::bootstrap::SecretOffer) -> (Arc<TestVm>, SocketAddrV4, std::thread::JoinHandle<()>) {
        spawn_vm_life(ca, boot, cores, kind, secret, false)
    }

    fn spawn_vm_life(ca: &InstallerCa, boot: &Issued, cores: u32, kind: Kind, secret: crate::bootstrap::SecretOffer, for_life: bool) -> (Arc<TestVm>, SocketAddrV4, std::thread::JoinHandle<()>) {
        spawn_vm_mode(ca, boot, cores, kind, secret, if for_life { Mode::ForLife } else { Mode::UntilClosed })
    }

    #[derive(Clone, Copy)]
    enum Mode {
        UntilClosed,
        ForLife,
        AfterClose,
    }

    fn spawn_vm_mode(ca: &InstallerCa, boot: &Issued, cores: u32, kind: Kind, secret: crate::bootstrap::SecretOffer, mode: Mode) -> (Arc<TestVm>, SocketAddrV4, std::thread::JoinHandle<()>) {
        let log = Arc::new(BootLog::new(100));
        log.push("korp-installer: pass began\nkorp-installer: INSTALL-OK\ngunnar: first boot");
        log.push("reading GUNNAR_UI_ADMIN_TOKEN=a1 from the env");
        spawn_vm_on(ca, boot, cores, kind, secret, mode, log)
    }

    fn spawn_vm_on(ca: &InstallerCa, boot: &Issued, cores: u32, kind: Kind, secret: crate::bootstrap::SecretOffer, mode: Mode, log: Arc<BootLog>) -> (Arc<TestVm>, SocketAddrV4, std::thread::JoinHandle<()>) {
        let vm = Arc::new(TestVm {
            estate: "gunnar.rs".into(),
            own_key: mtls::box_request().unwrap().1,
            cores,
            own_cert: Mutex::new(None),
            machine: Mutex::new(Machine::default()),
            closed: AtomicBool::new(false),
            log: log.clone(),
            boot: boot.clone(),
            role: Mutex::new(None),
            kind,
            feeds: Mutex::new(Vec::new()),
            secret,
            offer_after: std::sync::atomic::AtomicU32::new(0),
            front_offer_after: std::sync::atomic::AtomicU32::new(u32::MAX),
            replies: Mutex::new(Vec::new()),
            boot_now: Mutex::new(None),
        });
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let ca_pem = ca.cert_pem.clone();
        let v = vm.clone();
        let h = std::thread::spawn(move || match mode {
            Mode::ForLife => crate::bootstrap::serve_for_life(v.as_ref(), &log, &ca_pem, l, &|_| {}),
            Mode::UntilClosed => crate::bootstrap::serve(v.as_ref(), &log, &ca_pem, l, &|_| {}),
            Mode::AfterClose => crate::bootstrap::serve_for_life_or_after_close(v.as_ref(), &log, &ca_pem, l, true, &|_| {}),
        });
        (vm, at, h)
    }

    #[test]
    fn the_installer_admits_feeds_elects_collects_and_closes_every_vm_and_a_stranger_is_refused() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (main_vm, main_at, h1) = spawn_vm(&ca, &boot, 4);
        let (twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        // The medium's bootstrap certificate is NOT the installer: refused.
        let mut stranger = Client::new(Box::new(DirectPath { at: main_at }), &ca.cert_pem, &boot, Duration::from_secs(5)).unwrap();
        let e = stranger.hello().unwrap_err();
        assert!(e.contains("REFUSED [installer-only]"), "{e}");
        // Another CA's installer never gets past the handshake.
        let other = InstallerCa::mint("other", H).unwrap();
        let mut foreign = Client::new(Box::new(DirectPath { at: main_at }), &ca.cert_pem, &other.issue_installer(H).unwrap(), Duration::from_secs(5)).unwrap();
        assert!(foreign.hello().is_err());

        let vm = |name: &str, public: [u8; 4], o: Ipv4Addr| PlanVm { name: name.into(), public: Ipv4Addr::from(public), utility: Ipv4Addr::new(10, 13, 9, public[3]), overlay: o, kind: Kind::Gunnar, feed: BTreeMap::new() };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("main", [127, 0, 0, 1], crate::OVERLAY_BOX_A), vm("twin", [127, 0, 0, 2], crate::OVERLAY_BOX_B)],
            feed: BTreeMap::from([("monetize_pubkey".to_string(), "eb16".to_string())]),
            front: Some(FrontTunnel { tunnel_pubkey: "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=".into(), endpoint: "95.111.211.189:51820".into() }),
            port: 0,
            link_port: 0,
            budget_s: 60,
            close_when: CloseWhen::Paired,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        };
        assert_eq!(Plan::parse(&plan.render().unwrap()).unwrap(), plan);
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![(plan.vms[0].clone(), Box::new(DirectPath { at: main_at })), (plan.vms[1].clone(), Box::new(DirectPath { at: twin_at }))];
        let sink = Collect(Mutex::new(Vec::new()), Mutex::new(BTreeMap::new()), Mutex::new(BTreeMap::new()));
        // The ledger keys are the plan's public addresses; loopback connects
        // from 127.0.0.1 whatever the plan says, so the door is told by the
        // address the installer DIALLED — the plan's — not the socket's.
        let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:?}", sink.0.lock().unwrap()));
        assert!(out.vms.iter().all(|v| v.closed && v.gone && v.admitted), "{out:?}");
        let roles: BTreeMap<_, _> = out.vms.iter().map(|v| (v.name.clone(), v.role)).collect();
        assert_eq!(roles["twin"], Some(crate::Role::Twin), "the one with fewer cores");
        assert_eq!(roles["main"], Some(crate::Role::Primary));
        assert_eq!(sink.2.lock().unwrap()["main"]["ticket_pubkey"], "ticket-4", "the primary's offer reached the sink");
        assert!(!sink.2.lock().unwrap().contains_key("twin"), "the twin offered nothing");
        // Owner 2026-09-30: EVERY pairing step is its own line in the
        // installer's log — the hello and the admission of each VM, and the
        // election, once.
        let said = sink.0.lock().unwrap().clone();
        for vm in ["main", "twin"] {
            assert_eq!(said.iter().filter(|l| l.starts_with(&format!("{vm}: hello — "))).count(), 1, "{vm}'s hello and admission, said once: {said:#?}");
        }
        let elections: Vec<&String> = said.iter().filter(|l| l.starts_with("the election: ")).collect();
        assert_eq!(elections.len(), 1, "the election said once: {said:#?}");
        assert!(elections[0].contains("primary `main`") && elections[0].contains("twin `twin`"), "{}", elections[0]);
        assert!(main_vm.closed.load(Ordering::SeqCst) && twin_vm.closed.load(Ordering::SeqCst));
        assert!(main_vm.own_cert.lock().unwrap().as_deref().unwrap_or("").contains("BEGIN CERTIFICATE"), "each VM got its own certificate");
        let logs = sink.1.lock().unwrap();
        let main_log = &logs["main"];
        assert_eq!(main_log[0], "korp-installer: pass began", "the log from its first line");
        assert!(main_log.iter().any(|l| l.starts_with("<<WITHHELD: a token-class line")), "{main_log:?}");
        assert!(main_log.iter().any(|l| l.contains("fed as `main`")), "{main_log:?}");
        assert_eq!(main_log.iter().filter(|l| l.as_str() == "gunnar: first boot").count(), 1, "every line exactly once");
        h1.join().unwrap();
        h2.join().unwrap();
        // After the close nothing answers.
        assert!(Client::new(Box::new(DirectPath { at: main_at }), &ca.cert_pem, &m.me, Duration::from_secs(2)).unwrap().hello().is_err());
    }

    /// 0.5.5 (MEASURED 2026-09-28, test2's main VM: `… 2243 lines dropped`,
    /// the pairing lines gone after the close, 8 502 of 20 008 held lines a
    /// pubkey-class marker): a flood of ssh logins and secret-bearing lines
    /// after the close leaves the install and pairing lines readable through
    /// `read_log`; every login reads as a login with its fingerprint withheld
    /// in place; a run of withheld lines is ONE counted marker; and the log's
    /// byte bound holds throughout.
    #[test]
    fn a_flood_of_pubkey_lines_leaves_the_pairing_lines_readable_through_read_log() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (tail_cap, tail_bytes, head_cap, head_bytes) = (2_000, 64 * 1024, 100, 16 * 1024);
        let log = Arc::new(BootLog::with_head(tail_cap, tail_bytes, head_cap, head_bytes));
        log.push("korp-installer: pass began\nkorp-installer: INSTALL-OK\ngunnar: first boot");
        log.push("pairing: admitted as `main`, role primary\npairing: PAIRED with twin");
        log.seal_head();
        for i in 0..20_000u32 {
            log.push(&format!("WARN gunnar::audit: event=\"ssh.auth.accepted\" fingerprint=SHA256:{i:043} principal=admin"));
            log.push("reading GUNNAR_UI_ADMIN_TOKEN=a1 from the env\nreading GUNNAR_UI_ADMIN_TOKEN=a1 from the env");
            assert!(log.held_bytes() <= tail_bytes + head_bytes, "{}", log.held_bytes());
        }
        assert_eq!(log.withheld(), 40_000);
        let (_vm, at, _h) = spawn_vm_on(&ca, &boot, 4, Kind::Gunnar, crate::bootstrap::SecretOffer::default(), Mode::ForLife, log.clone());
        let m = Material { ca, me, link: None };
        let mut all: Vec<String> = Vec::new();
        let mut from = 0;
        loop {
            let c = read_log(Box::new(DirectPath { at }), &m, from).unwrap();
            if c.lines.is_empty() && c.dropped == 0 {
                break;
            }
            all.extend(c.stored());
            from = c.next;
        }
        assert_eq!(
            &all[..5],
            &["korp-installer: pass began", "korp-installer: INSTALL-OK", "gunnar: first boot", "pairing: admitted as `main`, role primary", "pairing: PAIRED with twin"]
        );
        assert!(all[5].contains("lines dropped"), "{:?}", all[5]);
        let logins: Vec<&String> = all.iter().filter(|l| l.contains("ssh.auth.accepted")).collect();
        assert!(!logins.is_empty() && logins.iter().all(|l| l.contains("fingerprint=<withheld>") && !l.contains("SHA256:")), "{:?}", logins.first());
        let markers: Vec<&String> = all.iter().filter(|l| l.starts_with("<<WITHHELD")).collect();
        assert!(!markers.is_empty() && markers.iter().all(|l| l.contains("2 token-class lines in a row")), "{:?}", markers.first());
        // The tail may start on a marker whose login was evicted.
        assert!(markers.len() - logins.len() <= 1, "one marker per run of two, never one per withheld line: {} markers, {} logins", markers.len(), logins.len());
        assert_eq!(log.total(), 5 + 40_000, "a login and a folded run per round");
    }

    /// 0.5.4: the standard — after the close the install is gone (proved),
    /// and the installer still reads the VM's log, new lines included, with
    /// its own certificate only; the medium's bootstrap certificate cannot.
    #[test]
    fn after_the_close_the_installer_still_reads_the_log_and_nothing_else() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (main_vm, main_at, _for_life) = spawn_vm_life(&ca, &boot, 4, Kind::Gunnar, crate::bootstrap::SecretOffer::default(), true);
        let (_twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        let vm = |name: &str, public: [u8; 4], o: Ipv4Addr| PlanVm { name: name.into(), public: Ipv4Addr::from(public), utility: Ipv4Addr::new(10, 13, 9, public[3]), overlay: o, kind: Kind::Gunnar, feed: BTreeMap::new() };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("main", [127, 0, 0, 1], crate::OVERLAY_BOX_A), vm("twin", [127, 0, 0, 2], crate::OVERLAY_BOX_B)],
            feed: BTreeMap::new(),
            front: Some(FrontTunnel { tunnel_pubkey: "xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=".into(), endpoint: "95.111.211.189:51820".into() }),
            port: 0,
            link_port: 0,
            budget_s: 60,
            close_when: CloseWhen::Paired,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![(plan.vms[0].clone(), Box::new(DirectPath { at: main_at })), (plan.vms[1].clone(), Box::new(DirectPath { at: twin_at }))];
        let sink = Collect(Mutex::new(Vec::new()), Mutex::new(BTreeMap::new()), Mutex::new(BTreeMap::new()));
        let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:?}", sink.0.lock().unwrap()));
        assert!(out.vms.iter().all(|v| v.closed && v.gone), "the log door is not the install: both proved gone {out:?}");
        h2.join().unwrap();
        // Hours later, the box logs on ...
        main_vm.log.push("grow: draining\nreading GUNNAR_UI_ADMIN_TOKEN=abc from the env");
        let all = read_log(Box::new(DirectPath { at: main_at }), &m, 0).unwrap();
        assert_eq!(all.lines[0], "korp-installer: pass began", "from its first line");
        assert!(all.lines.iter().any(|l| l == "grow: draining"), "a line logged after the close: {:?}", all.lines);
        assert!(!all.lines.iter().any(|l| l.contains("abc")), "secrets withheld as in the run: {:?}", all.lines);
        let next = read_log(Box::new(DirectPath { at: main_at }), &m, all.next).unwrap();
        assert!(next.lines.is_empty() && next.next == all.next, "follow: nothing new, same cursor");
        // ... the install itself is refused by name, and nobody else reads.
        let e = Client::new(Box::new(DirectPath { at: main_at }), &ca.cert_pem, &m.me, Duration::from_secs(2)).unwrap().hello().unwrap_err();
        assert!(e.contains("[install-closed]"), "{e}");
        let e = Client::new(Box::new(DirectPath { at: main_at }), &ca.cert_pem, &boot, Duration::from_secs(2)).unwrap().log(0).unwrap_err();
        assert!(e.contains("[installer-only]"), "{e}");
        assert!(log_of(&plan, &m, "nope", 0).unwrap_err().contains("[installer-no-such-vm]"));
    }

    /// A VM the installer closed on an EARLIER start (a restart after the
    /// install) opens the log door at once and nothing else: the installer
    /// reads the whole log of this start, the install answers
    /// `[install-closed]`, and the medium's bootstrap certificate reads nothing.
    #[test]
    fn a_restarted_closed_vm_serves_the_log_door_only() {
        let ca = InstallerCa::mint("holger.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (vm, at, _life) = spawn_vm_mode(&ca, &boot, 2, Kind::Holger, crate::bootstrap::SecretOffer::default(), Mode::AfterClose);
        vm.log.push("holger: restarted after the install\nreading GUNNAR_UI_ADMIN_TOKEN=abc from the env");
        let m = Material { ca: ca.clone(), me, link: None };
        let all = read_log(Box::new(DirectPath { at }), &m, 0).unwrap();
        assert!(all.lines.iter().any(|l| l == "holger: restarted after the install"), "{:?}", all.lines);
        assert!(!all.lines.iter().any(|l| l.contains("abc")), "secrets withheld: {:?}", all.lines);
        let e = Client::new(Box::new(DirectPath { at }), &ca.cert_pem, &m.me, Duration::from_secs(2)).unwrap().hello().unwrap_err();
        assert!(e.contains("[install-closed]"), "the install is not served again: {e}");
        let e = Client::new(Box::new(DirectPath { at }), &ca.cert_pem, &boot, Duration::from_secs(2)).unwrap().log(0).unwrap_err();
        assert!(e.contains("[installer-only]"), "{e}");
    }

    /// 0.4.2: the front a plan's boxes are fed is vm1 at its UTILITY
    /// address, port 51820 — never the public one the installer dials.
    #[test]
    fn the_boxes_are_fed_the_front_on_its_utility_address() {
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 80, 99, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed: BTreeMap::new(),
            front: None,
            port: 8474,
            link_port: 51821,
            budget_s: 60,
            close_when: CloseWhen::Paired,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        };
        let mut d = plan.door().with_ca(InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap());
        let req = |k: &str| JoinRequest { estate: "gunnar.rs".into(), kind: Kind::Gunnar, tunnel_pubkey: k.into(), capacity: None, csr_pem: mtls::box_request().unwrap().0 };
        d.join(&req("xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg="), &Presented::Bootstrap, Ipv4Addr::new(127, 80, 99, 1)).unwrap();
        let main = d.join(&req("KM"), &Presented::Bootstrap, Ipv4Addr::new(127, 80, 99, 2)).unwrap();
        assert_eq!(main.front().expect("fed").endpoint.as_deref(), Some("10.13.8.1:51820"));
    }

    /// P3a: an untagged VM keeps today's default (a gunnar box); a VM tagged with a product's
    /// own kind is read as that box kind.
    #[test]
    fn an_untagged_vm_is_still_a_gunnar_box_and_a_tagged_one_its_own_kind() {
        let untagged: PlanVm = toml::from_str("name = \"main\"\npublic = \"1.2.3.4\"\nutility = \"10.0.0.1\"\noverlay = \"10.44.0.2\"\n").unwrap();
        assert_eq!(untagged.kind, Kind::Gunnar);
        let tagged: PlanVm = toml::from_str("name = \"fe\"\npublic = \"1.2.3.4\"\nutility = \"10.0.0.1\"\noverlay = \"10.44.0.2\"\nkind = \"fe\"\n").unwrap();
        assert_eq!(tagged.kind, Kind::named("fe").unwrap());
        assert_eq!(toml::from_str::<PlanVm>(&toml::to_string(&tagged).unwrap()).unwrap(), tagged);
    }

    #[test]
    fn a_plan_is_checked_by_name() {
        let base = Plan { estate: "e".into(), tunnel: TunnelKind::Wireguard, vms: vec![], feed: BTreeMap::new(), front: None, port: 1, link_port: 1, budget_s: 1, close_when: CloseWhen::Fed, box_cert_valid_s: None, front_waits_for: Vec::new() };
        assert!(base.check().unwrap_err().contains("no VM"));
        let v = PlanVm { name: "main".into(), public: Ipv4Addr::new(1, 2, 3, 4), utility: Ipv4Addr::new(10, 0, 0, 1), overlay: crate::OVERLAY_BOX_A, kind: Kind::Gunnar, feed: BTreeMap::new() };
        let none = Plan { vms: vec![v.clone()], tunnel: TunnelKind::None, ..base.clone() };
        assert!(none.check().unwrap_err().contains("REFUSED [installer-link-none]"), "WireGuard + mutual TLS always");
        let two = Plan { vms: vec![v.clone(), PlanVm { name: "twin".into(), ..v.clone() }], ..base.clone() };
        assert!(two.check().unwrap_err().contains("two VMs at 1.2.3.4"));
        let ts = Plan { vms: vec![v.clone()], tunnel: TunnelKind::Tailscale, ..base.clone() };
        assert!(ts.check().unwrap_err().contains("[tunnel-tailscale-unimplemented]"));
        let wg = Plan { vms: vec![v.clone()], tunnel: TunnelKind::Wireguard, ..base.clone() };
        assert!(path_to(&wg, &v, None).err().unwrap().contains("[installer-link-keys]"));
        let d = Plan { vms: vec![v.clone()], ..base }.door();
        assert_eq!(d.ledger.slots[&v.public].utility, Some(v.utility), "keyed by the dialled address, the utility kept");
        assert_eq!(b64_32("xTIBA5rboUvnH4htodjb6e697QjLERt1NAB4mZqp8Dg=", "k").unwrap().len(), 32);
        assert!(b64_32("short", "k").is_err());
    }

    /// 0.5.0: what only the front may hold is refused in the shared feed and
    /// on a box, and taken on the front's own feed.
    #[test]
    fn front_only_keys_are_refused_in_the_shared_feed_and_on_a_box() {
        let v = |name: &str, last: u8, kind: Kind, feed: &[(&str, &str)]| PlanVm {
            name: name.into(),
            public: Ipv4Addr::new(127, 80, 99, last),
            utility: Ipv4Addr::new(10, 13, 8, last),
            overlay: crate::OVERLAY_BOX_A,
            kind,
            feed: feed.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(),
        };
        let base = Plan { estate: "e".into(), tunnel: TunnelKind::Wireguard, vms: vec![v("front", 1, Kind::Front, &[]), v("main", 2, Kind::Gunnar, &[])], feed: BTreeMap::new(), front: None, port: 1, link_port: 1, budget_s: 1, close_when: CloseWhen::Fed, box_cert_valid_s: None, front_waits_for: Vec::new() };
        base.check().unwrap();
        for key in ["tls.gunnar.rs.key", "tls.gunnar.rs.cert", "console_admin_token", "monetize_backup_kek", "plugin.gunnar.wasm", "plugin_root.vetra"] {
            assert!(front_only(key), "{key}");
            let shared = Plan { feed: BTreeMap::from([(key.to_string(), "x".to_string())]), ..base.clone() };
            assert!(shared.check().unwrap_err().contains("REFUSED [feed-front-only-shared]"), "{key}");
            let on_box = Plan { vms: vec![v("front", 1, Kind::Front, &[]), v("main", 2, Kind::Gunnar, &[(key, "x")])], ..base.clone() };
            assert!(on_box.check().unwrap_err().contains("REFUSED [feed-front-only-on-a-box]"), "{key}");
            let on_front = Plan { vms: vec![v("front", 1, Kind::Front, &[(key, "x")]), v("main", 2, Kind::Gunnar, &[])], ..base.clone() };
            on_front.check().unwrap();
        }
        for key in ["admin_pubkey", "principal.op", "web_hosts", "monetize_pubkey"] {
            assert!(!front_only(key), "{key} may be shared");
        }
        // A VM's own feed survives the plan's TOML, and its Debug shows keys only.
        let p = Plan { vms: vec![v("front", 1, Kind::Front, &[("tls.gunnar.rs.key", "PRIVATE-KEY-BYTES")]), v("main", 2, Kind::Gunnar, &[])], ..base };
        assert_eq!(Plan::parse(&p.render().unwrap()).unwrap(), p);
        let dbg = format!("{p:?} {:?}", p.door());
        assert!(dbg.contains("tls.gunnar.rs.key") && !dbg.contains("PRIVATE-KEY-BYTES"), "{dbg}");
    }

    const PSK_MAIN: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
    const PSK_TWIN: &str = "ICEiIyQlJicoKSorLC0uLzAxMjM0NTY3ODk6Ozw9Pj8=";

    fn three_vm_plan() -> Plan {
        let v = |name: &str, last: u8, kind: Kind, overlay: Ipv4Addr| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 80, 99, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        Plan {
            estate: "e".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![v("vm1", 1, Kind::Front, crate::OVERLAY_FRONT), v("vm2", 2, Kind::Gunnar, crate::OVERLAY_BOX_A), v("vm3", 3, Kind::Gunnar, crate::OVERLAY_BOX_B)],
            feed: BTreeMap::from([("admin_pubkey".to_string(), "ssh-ed25519 AAAA".to_string())]),
            front: None,
            port: 1,
            link_port: 1,
            budget_s: 1,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        }
    }

    /// **The pre-shared keys go to the two ends of each link and nowhere else**
    /// (2026-10-06, the PQ transport lane): each box is fed its own PSK, the
    /// front every box's under its name; no box holds another's, the shared feed
    /// holds none, the ledger file holds none, and no Debug prints one.
    #[test]
    fn each_box_is_fed_its_psk_the_front_every_boxs_and_nobody_else_holds_one() {
        let psks = BTreeMap::from([("vm2".to_string(), PSK_MAIN.to_string()), ("vm3".to_string(), PSK_TWIN.to_string())]);
        let plan = three_vm_plan().with_wg_psks(&psks).unwrap();
        plan.check().unwrap();
        let d = plan.door();
        let main = d.feed_for("vm2");
        let twin = d.feed_for("vm3");
        let front = d.feed_for("vm1");
        assert_eq!(main.get(FEED_WG_PSK).map(String::as_str), Some(PSK_MAIN));
        assert_eq!(twin.get(FEED_WG_PSK).map(String::as_str), Some(PSK_TWIN));
        assert!(!main.keys().chain(twin.keys()).any(|k| k.starts_with(FEED_WG_PSK_PREFIX)), "a box holds no other box's PSK: {:?} {:?}", main.keys(), twin.keys());
        assert_eq!(front.get(&feed_wg_psk_for("vm2")).map(String::as_str), Some(PSK_MAIN));
        assert_eq!(front.get(&feed_wg_psk_for("vm3")).map(String::as_str), Some(PSK_TWIN));
        assert!(!front.contains_key(FEED_WG_PSK), "the front has no link to a front");
        assert!(!d.feed.keys().any(|k| k.starts_with(FEED_WG_PSK)), "the shared feed holds none");
        let ledger = toml::to_string(&d).unwrap();
        assert!(!ledger.contains(PSK_MAIN) && !ledger.contains(PSK_TWIN), "the ledger file holds no PSK: {ledger}");
        let dbg = format!("{plan:?} {d:?}");
        assert!(!dbg.contains(PSK_MAIN) && !dbg.contains(PSK_TWIN), "{dbg}");
        assert_eq!(Plan::parse(&plan.render().unwrap()).unwrap(), plan, "a rendered plan keeps them (it is kept 0600)");
    }

    /// **A PSK where it does not belong is refused by name**: in the shared
    /// feed, a box holding the front's `wg_psk.<box>`, the front holding a box's
    /// `wg_psk`, a key that is not 32 bytes, a PSK for the front or for a VM the
    /// plan does not have, and two ends of one link holding different keys.
    #[test]
    fn a_psk_where_it_does_not_belong_is_refused_by_name() {
        let base = three_vm_plan();
        let one = |name: &str, psk: &str| BTreeMap::from([(name.to_string(), psk.to_string())]);
        assert!(base.clone().with_wg_psks(&one("vm2", "short")).unwrap_err().contains("[installer-wg-psk]"));
        assert!(base.clone().with_wg_psks(&one("vm1", PSK_MAIN)).unwrap_err().contains("[installer-wg-psk]"), "the front has no link to itself");
        assert!(base.clone().with_wg_psks(&one("vm9", PSK_MAIN)).unwrap_err().contains("[installer-wg-psk]"), "no such VM");
        let shared = Plan { feed: BTreeMap::from([(FEED_WG_PSK.to_string(), PSK_MAIN.to_string())]), ..base.clone() };
        assert!(shared.check().unwrap_err().contains("[feed-wg-psk-shared]"), "one PSK for every box is no pair's key");
        let mut on_box = base.clone();
        on_box.vms[1].feed.insert(feed_wg_psk_for("vm3"), PSK_TWIN.into());
        assert!(on_box.check().unwrap_err().contains("[feed-front-only-on-a-box]"));
        let mut on_front = base.clone();
        on_front.vms[0].feed.insert(FEED_WG_PSK.into(), PSK_MAIN.into());
        assert!(on_front.check().unwrap_err().contains("[feed-wg-psk-on-the-front]"));
        let mut split = base.clone().with_wg_psks(&one("vm2", PSK_MAIN)).unwrap();
        split.vms[0].feed.insert(feed_wg_psk_for("vm2"), PSK_TWIN.into());
        assert!(split.check().unwrap_err().contains("[feed-wg-psk-split]"), "the two ends of a link hold one key");
        let mut half = base.clone().with_wg_psks(&one("vm2", PSK_MAIN)).unwrap();
        half.vms[0].feed.remove(&feed_wg_psk_for("vm2"));
        assert!(half.check().unwrap_err().contains("[feed-wg-psk-split]"), "a box with a PSK the front does not hold could never pair");
    }

    /// Every file of a run folder, recursively, as text.
    fn every_file(dir: &FsPath) -> Vec<(PathBuf, String)> {
        let mut out = Vec::new();
        for e in std::fs::read_dir(dir).unwrap() {
            let p = e.unwrap().path();
            if p.is_dir() {
                out.extend(every_file(&p));
            } else {
                out.push((p.clone(), String::from_utf8_lossy(&std::fs::read(&p).unwrap()).into_owned()));
            }
        }
        out
    }

    /// **0.5.0, the owner's two rules in one run** (front + main + twin):
    ///
    /// 1. main and twin are NEVER fed `tls.*`, `plugin.*` or
    ///    `console_admin_token` — the front's own feed reaches the front only
    ///    — while the shared feed and the relayed offers reach everyone as
    ///    before;
    /// 2. at the end, the front HANDS OVER its certificate and key; they land
    ///    in the run folder's 0600 secret store and NOWHERE else (not the
    ///    ledger, the offers, the narration, a log, another VM's feed), and
    ///    [`successor_feed`] gives them to the next front's own feed.
    #[test]
    fn main_and_twin_never_get_the_fronts_secrets_and_the_front_hands_its_certificate_to_its_successor() {
        const FED_KEY: &str = "-----BEGIN PRIVATE KEY-----\nFED-TLS-KEY-a1b2c3\n-----END PRIVATE KEY-----\n";
        const FED_TOKEN: &str = "console-admin-token-0123456789abcdef0123456789";
        const HANDED_KEY: &str = "-----BEGIN PRIVATE KEY-----\nHANDED-OVER-KEY-d4e5f6\n-----END PRIVATE KEY-----\n";
        const HANDED_CERT: &str = "-----BEGIN CERTIFICATE-----\nHANDED-OVER-CERT\n-----END CERTIFICATE-----\n";
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let handed = crate::bootstrap::SecretOffer::new(BTreeMap::from([("tls.gunnar.rs.cert".to_string(), HANDED_CERT.to_string()), ("tls.gunnar.rs.key".to_string(), HANDED_KEY.to_string())]));
        let (front_vm, front_at, h0) = spawn_vm_as(&ca, &boot, 0, Kind::Front, handed.clone());
        let (main_vm, main_at, h1) = spawn_vm(&ca, &boot, 4);
        let (twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        let own = BTreeMap::from([
            ("tls.gunnar.rs.cert".to_string(), "-----BEGIN CERTIFICATE-----\nFED-CERT\n-----END CERTIFICATE-----\n".to_string()),
            ("tls.gunnar.rs.key".to_string(), FED_KEY.to_string()),
            ("console_admin_token".to_string(), FED_TOKEN.to_string()),
            ("plugin.gunnar.wasm".to_string(), "AGFzbQEAAAA=".to_string()),
            // A PLUGIN-SIZED value (0.5.1): a real signed module is ≈0.8 MB base64,
            // and the listener's old 64 KiB refused every feed that carried one.
            ("plugin.big.wasm".to_string(), "A".repeat(900 * 1024)),
        ]);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind, feed: BTreeMap<String, String>| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front, own.clone()), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar, BTreeMap::new()), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar, BTreeMap::new())],
            feed: BTreeMap::from([("admin_pubkey".to_string(), "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOP operator".to_string())]),
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 60,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![
            (plan.vms[0].clone(), Box::new(DirectPath { at: front_at })),
            (plan.vms[1].clone(), Box::new(DirectPath { at: main_at })),
            (plan.vms[2].clone(), Box::new(DirectPath { at: twin_at })),
        ];
        let dir = std::env::temp_dir().join(format!("nordisk-pairing-handover-{}-{}", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        let sink = FolderSink::new(&dir, false).unwrap();
        run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{}", std::fs::read_to_string(dir.join("installer.log")).unwrap_or_default()));
        for h in [h0, h1, h2] {
            h.join().unwrap();
        }

        // 1. The boxes: fed, never with the front's own values.
        for (name, box_vm) in [("main", &main_vm), ("twin", &twin_vm)] {
            let feeds = box_vm.feeds.lock().unwrap();
            assert!(!feeds.is_empty(), "{name} was fed");
            for f in feeds.iter() {
                assert!(f.keys().all(|k| !front_only(k) && !k.starts_with("tls.") && !k.starts_with("plugin.") && k != "console_admin_token"), "{name} was fed {:?}", f.keys().collect::<Vec<_>>());
                assert!(f.values().all(|v| !v.contains("FED-TLS-KEY") && !v.contains(FED_TOKEN) && !v.contains("HANDED-OVER")), "{name} got a front secret");
                assert!(f.contains_key("admin_pubkey"), "the shared feed still reaches {name}");
            }
        }
        // Offers are relayed as before: into the SHARED feed (the ledger keeps it), so every VM's next answer carries them.
        let ledger = Door::load(&dir.join("ledger.toml")).unwrap();
        assert_eq!(ledger.feed.get("offer.main.ticket_pubkey").map(String::as_str), Some("ticket-4"), "offers are relayed as before");
        assert!(ledger.own_feed.is_empty(), "no VM's own feed is in the ledger");
        // The front: its own values AND the shared ones.
        let last = front_vm.feeds.lock().unwrap().last().cloned().unwrap();
        assert_eq!(last.get("tls.gunnar.rs.key").map(String::as_str), Some(FED_KEY));
        assert_eq!(last.get("plugin.big.wasm").map(String::len), Some(900 * 1024), "a plugin-sized feed reaches the front whole");
        assert_eq!(last.get("console_admin_token").map(String::as_str), Some(FED_TOKEN));
        assert!(last.contains_key("plugin.gunnar.wasm") && last.contains_key("admin_pubkey"));

        // 2. The hand-over: in the 0600 store, and nowhere else.
        let store = dir.join(SECRET_OFFERS_FILE);
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            assert_eq!(std::fs::metadata(&store).unwrap().permissions().mode() & 0o777, 0o600, "the secret store is 0600");
        }
        assert_eq!(read_secret_offers(&dir).unwrap()["front"], handed);
        assert!(!read_secret_offers(&dir).unwrap().contains_key("main"), "a box hands over nothing");
        for (path, text) in every_file(&dir) {
            if path == store {
                assert!(text.contains("HANDED-OVER-KEY"));
                continue;
            }
            for secret in ["HANDED-OVER", "FED-TLS-KEY", FED_TOKEN] {
                assert!(!text.contains(secret), "{} holds `{secret}`", path.display());
            }
        }
        let narration = std::fs::read_to_string(dir.join("installer.log")).unwrap();
        assert!(narration.contains("front: handed over 2 secret value(s) for its successor (tls.gunnar.rs.cert, tls.gunnar.rs.key)"), "{narration}");
        assert!(read_offers(&dir).unwrap().get("front").is_none_or(|o| !o.contains_key("tls.gunnar.rs.key")), "a secret is not an offer");
        for f in main_vm.feeds.lock().unwrap().iter().chain(twin_vm.feeds.lock().unwrap().iter()) {
            assert!(f.keys().all(|k| !k.starts_with("offer.front.tls")), "a secret is never fed on");
        }

        // The next rebuild: the successor front is fed what its predecessor handed over.
        let carried = successor_feed(&dir, "front").unwrap();
        assert_eq!(carried, handed.clone().into_inner());
        assert!(successor_feed(&dir, "main").unwrap().is_empty());
        let next = Plan { vms: vec![PlanVm { feed: carried, ..plan.vms[0].clone() }, plan.vms[1].clone(), plan.vms[2].clone()], ..plan.clone() };
        next.check().unwrap();
        let d = next.door();
        assert_eq!(d.feed_for("front").get("tls.gunnar.rs.key").map(String::as_str), Some(HANDED_KEY));
        assert!(!d.feed_for("main").contains_key("tls.gunnar.rs.key") && !d.feed_for("twin").contains_key("tls.gunnar.rs.cert"));
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// **0.5.2 — the ticket key's race, and the front's certificate at 25 h.**
    ///
    /// The primary offers its ticket key LATE (only on its fifth offer read,
    /// long after every VM is fed and ready): with `front_waits_for =
    /// ["ticket_pubkey"]` nothing closes until the front's last feed carries
    /// `offer.main.ticket_pubkey`, and the outcome says per VM what each was
    /// fed. The plan's `box_cert_valid_s` makes the front's own certificate —
    /// the one its plugin door serves — outlive a day; the default does not.
    #[test]
    fn nothing_closes_until_the_front_is_fed_the_primarys_late_ticket_key_and_its_certificate_outlives_a_day() {
        let ca = InstallerCa::mint("gunnar.rs", 400 * 24 * H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (front_vm, front_at, h0) = spawn_vm_as(&ca, &boot, 0, Kind::Front, crate::bootstrap::SecretOffer::default());
        let (main_vm, main_at, h1) = spawn_vm(&ca, &boot, 4);
        let (_twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        main_vm.offer_after.store(4, Ordering::SeqCst);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        const LONG: u64 = 180 * 24 * 3600;
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed: BTreeMap::new(),
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 90,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: Some(LONG),
            front_waits_for: vec!["ticket_pubkey".into()],
        };
        assert_eq!(Plan::parse(&plan.render().unwrap()).unwrap(), plan, "the new key round-trips");
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![
            (plan.vms[0].clone(), Box::new(DirectPath { at: front_at })),
            (plan.vms[1].clone(), Box::new(DirectPath { at: main_at })),
            (plan.vms[2].clone(), Box::new(DirectPath { at: twin_at })),
        ];
        let sink = Collect(Mutex::new(Vec::new()), Mutex::new(BTreeMap::new()), Mutex::new(BTreeMap::new()));
        let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:#?}", sink.0.lock().unwrap()));
        for h in [h0, h1, h2] {
            h.join().unwrap();
        }
        let said = sink.0.lock().unwrap().join("\n");
        assert!(said.contains("not closing yet: the front `front` has not been fed `offer.main.ticket_pubkey`"), "the wait was said: {said}");
        let front = out.vms.iter().find(|v| v.name == "front").unwrap();
        assert!(front.closed && front.fed_keys.contains(&"offer.main.ticket_pubkey".to_string()), "{front:?}");
        assert_eq!(front_vm.feeds.lock().unwrap().last().unwrap().get("offer.main.ticket_pubkey").map(String::as_str), Some("ticket-4"), "the front's LAST feed carries the primary's key");

        // The front's own certificate, on a clock 25 h on.
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() as i64;
        let at_25h = now + 25 * 3600;
        let cert = front_vm.own_cert.lock().unwrap().clone().expect("the front got its own certificate");
        assert!(mtls::cert_left(&cert, at_25h).unwrap() > Duration::from_secs(170 * 24 * 3600), "the plan's validity reached the certificate");
        // The default (no `box_cert_valid_s`) is dead by then — the bug the plan's value closes.
        let (csr, _) = mtls::box_request().unwrap();
        let short = ca.sign_box(&csr, "front", crate::door::BOX_CERT_VALID).unwrap();
        assert_eq!(mtls::cert_left(&short, at_25h).unwrap(), Duration::ZERO, "the 24 h default has ended at 25 h");
    }

    /// **0.5.9 — the boxes wait for the FRONT's late offer.** A front laid
    /// again offers its (restored) `monetize_pubkey` only once it took its
    /// state back; a close that started first left every box without it, and
    /// a metered box without the front's key enforces no plan. With
    /// [`HOLD_FOR_OFFER_FEED`] = `monetize_pubkey` in the front's OWN feed,
    /// nothing closes until every product box was fed `offer.front.monetize_pubkey`.
    #[test]
    fn nothing_closes_until_every_box_is_fed_the_fronts_late_key() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (front_vm, front_at, h0) = spawn_vm_as(&ca, &boot, 0, Kind::Front, crate::bootstrap::SecretOffer::default());
        let (main_vm, main_at, h1) = spawn_vm(&ca, &boot, 4);
        let (twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        front_vm.front_offer_after.store(6, Ordering::SeqCst);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        let mut front = vm("front", 1, crate::OVERLAY_FRONT, Kind::Front);
        front.feed.insert(HOLD_FOR_OFFER_FEED.to_string(), "monetize_pubkey".to_string());
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![front, vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed: BTreeMap::new(),
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 90,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: vec!["ticket_pubkey".into()],
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![
            (plan.vms[0].clone(), Box::new(DirectPath { at: front_at })),
            (plan.vms[1].clone(), Box::new(DirectPath { at: main_at })),
            (plan.vms[2].clone(), Box::new(DirectPath { at: twin_at })),
        ];
        let sink = Collect(Mutex::new(Vec::new()), Mutex::new(BTreeMap::new()), Mutex::new(BTreeMap::new()));
        let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:#?}", sink.0.lock().unwrap()));
        for h in [h0, h1, h2] {
            h.join().unwrap();
        }
        let said = sink.0.lock().unwrap().join("\n");
        assert!(said.contains("has not been fed `offer.front.monetize_pubkey` (the front's offer)"), "the wait was said: {said}");
        for (name, v) in [("main", &main_vm), ("twin", &twin_vm)] {
            assert_eq!(v.feeds.lock().unwrap().last().unwrap().get("offer.front.monetize_pubkey").map(String::as_str), Some("mk-front"), "{name}'s LAST feed carries the front's key");
            assert!(out.vms.iter().find(|x| x.name == name).unwrap().closed);
        }
    }

    /// **The late keys (2026-10-20)** ([`LATE_FEED`]): every VM's FIRST feed lacks them, a later
    /// feed carries them, and nothing closes before each VM holds them.
    #[test]
    fn late_feed_keys_are_held_from_the_first_feed_and_fed_later_before_the_close() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let (front_vm, front_at, h0) = spawn_vm_as(&ca, &boot, 0, Kind::Front, crate::bootstrap::SecretOffer::default());
        let (main_vm, main_at, h1) = spawn_vm(&ca, &boot, 4);
        let (twin_vm, twin_at, h2) = spawn_vm(&ca, &boot, 1);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        let feed = BTreeMap::from([
            ("fleet".to_string(), "https://one.example".to_string()),
            ("fleet_at".to_string(), "10.13.12.5:443".to_string()),
            (LATE_FEED.to_string(), "fleet,fleet_at".to_string()),
            (LATE_FEED_S.to_string(), "3".to_string()),
        ]);
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed,
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 90,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: Vec::new(),
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let paths: Vec<(PlanVm, Box<dyn Path>)> = vec![
            (plan.vms[0].clone(), Box::new(DirectPath { at: front_at })),
            (plan.vms[1].clone(), Box::new(DirectPath { at: main_at })),
            (plan.vms[2].clone(), Box::new(DirectPath { at: twin_at })),
        ];
        let sink = Collect(Mutex::new(Vec::new()), Mutex::new(BTreeMap::new()), Mutex::new(BTreeMap::new()));
        let out = run(&plan, &m, paths, &sink).unwrap_or_else(|e| panic!("{e}\n{:#?}", sink.0.lock().unwrap()));
        for h in [h0, h1, h2] {
            h.join().unwrap();
        }
        let said = sink.0.lock().unwrap().join("\n");
        for (name, v) in [("front", &front_vm), ("main", &main_vm), ("twin", &twin_vm)] {
            let feeds = v.feeds.lock().unwrap();
            assert!(!feeds.first().unwrap().contains_key("fleet") && !feeds.first().unwrap().contains_key("fleet_at"), "{name}'s FIRST feed lacks the late keys");
            assert_eq!(feeds.last().unwrap().get("fleet").map(String::as_str), Some("https://one.example"), "{name}'s LAST feed carries them");
            assert!(out.vms.iter().find(|x| x.name == name).unwrap().closed);
            assert!(said.contains(&format!("{name}: fed WITHOUT fleet,fleet_at")), "{said}");
        }
    }

    /// **0.5.8 — the dead main, re-admitted into the closed estate.** An
    /// install of front + main + twin closes all three (each keeps its log
    /// door); the main dies and a FRESH one is laid at new addresses. The
    /// re-admission installs the new main alone — admitted on the bootstrap
    /// certificate into the old slot, the primary again, fed where its twin
    /// is — and RE-FEEDS the closed front and twin with its new address and
    /// tunnel key. A plan that moves a second box is refused before anything
    /// is dialled.
    #[test]
    fn a_dead_main_is_readmitted_into_the_closed_estate_and_the_closed_front_and_twin_are_re_fed() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let life = |cores, kind| spawn_vm_mode(&ca, &boot, cores, kind, crate::bootstrap::SecretOffer::default(), Mode::ForLife);
        let (front_vm, front_at, _h0) = life(0, Kind::Front);
        let (_dead_main, main_at, _h1) = life(4, Kind::Gunnar);
        let (twin_vm, twin_at, _h2) = life(1, Kind::Gunnar);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed: BTreeMap::new(),
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 60,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: vec!["ticket_pubkey".into()],
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let dir = std::env::temp_dir().join(format!("nordisk-pairing-readmit-{}-{}", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        let sink = FolderSink::new(&dir, false).unwrap();
        let direct = |at: SocketAddrV4| -> Box<dyn Path> { Box::new(DirectPath { at }) };
        run(&plan, &m, vec![(plan.vms[0].clone(), direct(front_at)), (plan.vms[1].clone(), direct(main_at)), (plan.vms[2].clone(), direct(twin_at))], &sink)
            .unwrap_or_else(|e| panic!("{e}\n{}", std::fs::read_to_string(dir.join("installer.log")).unwrap_or_default()));
        assert!(front_vm.closed.load(Ordering::SeqCst) && twin_vm.closed.load(Ordering::SeqCst));
        let prior = Door::load(&dir.join("ledger.toml")).unwrap();
        assert_eq!(prior.feed.get("offer.main.ticket_pubkey").map(String::as_str), Some("ticket-4"), "the dead main's offer is in the kept door");

        // The main dies; a fresh one (weaker, another tunnel key) is laid at a new address.
        let (new_main, new_at, h_new) = spawn_vm_mode(&ca, &boot, 2, Kind::Gunnar, crate::bootstrap::SecretOffer::default(), Mode::UntilClosed);
        let moved = Plan { vms: vec![plan.vms[0].clone(), vm("main", 9, crate::OVERLAY_BOX_A, Kind::Gunnar), plan.vms[2].clone()], ..plan.clone() };
        // Moving a second box is refused before anything is dialled.
        let two = Plan { vms: vec![plan.vms[0].clone(), vm("main", 9, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 8, crate::OVERLAY_BOX_B, Kind::Gunnar)], ..plan.clone() };
        let e = readmit(&two, prior.clone(), &m, "main", Vec::new(), &sink).unwrap_err();
        assert!(e.contains("[installer-readmit]") && e.contains("moves ONE box"), "{e}");
        assert!(readmit(&moved, { let mut o = prior.clone(); o.ledger.estate = "other.rs".into(); o }, &m, "main", Vec::new(), &sink).unwrap_err().contains("[installer-readmit]"));

        let (r, out) = readmit(&moved, prior, &m, "main", vec![(moved.vms[0].clone(), direct(front_at)), (moved.vms[1].clone(), direct(new_at)), (moved.vms[2].clone(), direct(twin_at))], &sink)
            .unwrap_or_else(|e| panic!("{e}\n{}", std::fs::read_to_string(dir.join("installer.log")).unwrap_or_default()));
        h_new.join().unwrap();
        let twin_ctl = format!("10.13.8.3:{}", crate::CONTROL_PORT);
        assert_eq!(r.recharge_from.as_deref(), Some(twin_ctl.as_str()));
        assert!(out.vms.iter().all(|v| v.closed && v.gone), "{out:?}");
        // The new main: the primary again, its own certificate, told where its twin is.
        let fed = new_main.replies.lock().unwrap().last().cloned().expect("the new main was fed");
        assert_eq!(fed.role, Some(crate::Role::Primary));
        assert_eq!(fed.feed.get(crate::door::RECHARGE_FROM_FEED), Some(&twin_ctl));
        assert!(new_main.own_cert.lock().unwrap().as_deref().unwrap_or("").contains("BEGIN CERTIFICATE"));
        assert!(new_main.closed.load(Ordering::SeqCst));
        // The closed front and twin were RE-FED the new main's address and key.
        let new_util = Ipv4Addr::new(10, 13, 8, 9);
        for (who, v) in [("front", &front_vm), ("twin", &twin_vm)] {
            let last = v.replies.lock().unwrap().last().cloned().unwrap();
            let p = last.peers.iter().find(|p| p.name == "main").unwrap_or_else(|| panic!("{who} lists main"));
            assert_eq!((p.utility, p.tunnel_pubkey.as_str()), (Some(new_util), "key-2"), "{who} was re-fed the new main");
            assert!(!last.feed.contains_key(crate::door::RECHARGE_FROM_FEED), "{who} is not told to recharge");
        }
        assert_eq!(twin_vm.replies.lock().unwrap().last().unwrap().primary_utility, Some(new_util), "the twin's answer names the new primary");
        assert_eq!(front_vm.replies.lock().unwrap().last().unwrap().feed.get("offer.main.ticket_pubkey").map(String::as_str), Some("ticket-2"), "the front holds the NEW main's offer, never the dead one's");
        let said = std::fs::read_to_string(dir.join("installer.log")).unwrap();
        assert!(said.contains("main: re-admitted") && said.contains("front: re-fed") && said.contains("twin: re-fed"), "{said}");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// ★ **0.5.26 — two boxes upgraded in place, re-admitted TOGETHER** (gunnar.rs, 2026-10-25).
    /// An install of front + main + twin closes all three; then both boxes run their keep
    /// medium (in place: the same addresses, a fresh system each). The twin's re-admission alone
    /// re-feeds the main, whose open install listener has no re-feed — the live failure, word
    /// for word: `REFUSED [installer-incomplete] … main: never answered (last: REFUSED
    /// [install-path] …)`, the twin fed and paired by a run that never closed. Re-admitted
    /// together, both are installed again — the main on the bootstrap certificate, the twin as
    /// itself with the certificate that run gave it — the front is re-fed both new tunnel keys,
    /// and everything closes.
    #[test]
    fn two_boxes_upgraded_in_place_are_readmitted_together_and_the_closed_front_is_re_fed() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let me = ca.issue_installer(H).unwrap();
        let life = |cores, kind| spawn_vm_mode(&ca, &boot, cores, kind, crate::bootstrap::SecretOffer::default(), Mode::ForLife);
        let (front_vm, front_at, _h0) = life(0, Kind::Front);
        let (_old_main, main_at, _h1) = life(4, Kind::Gunnar);
        let (_old_twin, twin_at, _h2) = life(1, Kind::Gunnar);
        let vm = |name: &str, last: u8, overlay: Ipv4Addr, kind: Kind| PlanVm { name: name.into(), public: Ipv4Addr::new(127, 0, 0, last), utility: Ipv4Addr::new(10, 13, 8, last), overlay, kind, feed: BTreeMap::new() };
        let plan = Plan {
            estate: "gunnar.rs".into(),
            tunnel: TunnelKind::Wireguard,
            vms: vec![vm("front", 1, crate::OVERLAY_FRONT, Kind::Front), vm("main", 2, crate::OVERLAY_BOX_A, Kind::Gunnar), vm("twin", 3, crate::OVERLAY_BOX_B, Kind::Gunnar)],
            feed: BTreeMap::new(),
            front: None,
            port: 0,
            link_port: 0,
            budget_s: 60,
            close_when: CloseWhen::Fed,
            box_cert_valid_s: None,
            front_waits_for: vec!["ticket_pubkey".into()],
        };
        let m = Material { ca: ca.clone(), me, link: None };
        let dir = std::env::temp_dir().join(format!("nordisk-pairing-together-{}-{}", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
        let sink = FolderSink::new(&dir, false).unwrap();
        let direct = |at: SocketAddrV4| -> Box<dyn Path> { Box::new(DirectPath { at }) };
        let log = || std::fs::read_to_string(dir.join("installer.log")).unwrap_or_default();
        run(&plan, &m, vec![(plan.vms[0].clone(), direct(front_at)), (plan.vms[1].clone(), direct(main_at)), (plan.vms[2].clone(), direct(twin_at))], &sink).unwrap_or_else(|e| panic!("{e}\n{}", log()));
        let prior = Door::load(&dir.join("ledger.toml")).unwrap();

        // Both boxes ran their keep medium in place: a fresh system each, another tunnel key.
        let (new_main, new_main_at, _h3) = life(6, Kind::Gunnar);
        let (new_twin, new_twin_at, _h4) = life(2, Kind::Gunnar);
        let paths = || vec![(plan.vms[0].clone(), direct(front_at)), (plan.vms[1].clone(), direct(new_main_at)), (plan.vms[2].clone(), direct(new_twin_at))];
        // The twin's re-admission ALONE waits for the main: the live words.
        let short = Plan { budget_s: 6, ..plan.clone() };
        let e = readmit_kept(&short, prior.clone(), &m, "twin", paths(), &sink).unwrap_err();
        assert!(e.starts_with("REFUSED [installer-incomplete]") && e.contains("main: never answered (last: REFUSED [install-path] this listener has /install/hello"), "{e}");
        assert!(new_twin.own_cert.lock().unwrap().is_some(), "that run fed the twin its own certificate, and never closed it");
        assert!(!new_twin.closed.load(Ordering::SeqCst));

        // Together: both installed again, the closed front re-fed.
        let (r, out) = readmit_kept_together(&plan, prior.clone(), &m, &["main".to_string(), "twin".to_string()], paths(), &sink).unwrap_or_else(|e| panic!("{e}\n{}", log()));
        assert_eq!(r.iter().map(|r| r.name.as_str()).collect::<Vec<_>>(), vec!["main", "twin"]);
        assert!(out.vms.iter().all(|v| v.closed), "{out:?}");
        assert!(new_main.closed.load(Ordering::SeqCst) && new_twin.closed.load(Ordering::SeqCst));
        assert_eq!(new_main.replies.lock().unwrap().last().unwrap().role, Some(crate::Role::Primary), "the main is the primary again");
        assert_eq!(new_twin.replies.lock().unwrap().last().unwrap().role, Some(crate::Role::Twin));
        let front_last = front_vm.replies.lock().unwrap().last().cloned().unwrap();
        for (name, key) in [("main", "key-6"), ("twin", "key-2")] {
            assert_eq!(front_last.peers.iter().find(|p| p.name == name).map(|p| p.tunnel_pubkey.as_str()), Some(key), "the front was re-fed {name}'s new tunnel key");
        }
        assert!(log().contains("twin: shows its own certificate from an earlier run of this re-admission"), "{}", log());
        // A re-admission at new addresses still moves ONE box.
        let e = readmit_how(&plan, prior, &m, &["main".to_string(), "twin".to_string()], Vec::new(), &sink, false).unwrap_err();
        assert!(e.contains("[installer-readmit]") && e.contains("moves ONE box"), "{e}");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// ★ **A closed VM's own certificate is renewed by its installer** (0.5.12, F4 at a sold
    /// Gunnar: the material lives in the estate, the box certificates last 180 days): the
    /// installer reads what the VM serves, re-signs ITS key with the CA for a new lifetime, and
    /// the VM keeps it — the same key, the same name, a later end; a certificate for another
    /// key is refused, and a VM that keeps none says so.
    #[test]
    fn a_closed_vms_certificate_is_renewed_for_its_own_key_and_nothing_else() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let (vm, at, _h) = spawn_vm_mode(&ca, &boot, 4, Kind::Gunnar, crate::bootstrap::SecretOffer::default(), Mode::AfterClose);
        // Installed and closed: its own certificate, signed for an hour.
        let (csr, _) = mtls::box_request_for(&vm.own_key).unwrap();
        let first = ca.sign_box(&csr, "main", H).unwrap();
        *vm.own_cert.lock().unwrap() = Some(first.clone());
        vm.closed.store(true, Ordering::SeqCst);
        let me = ca.issue_installer(H).unwrap();
        let mut c = Client::new(Box::new(DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
        assert_eq!(c.cert().unwrap().trim(), first.trim(), "the installer reads what the VM serves");
        let renewed = ca.renew_box(&first, 180 * 24 * H).unwrap();
        c.renew(&renewed).unwrap();
        let kept = vm.own_cert.lock().unwrap().clone().unwrap();
        assert_eq!(kept, renewed, "the VM keeps the renewed certificate");
        assert_eq!(mtls::cert_cn(&kept).unwrap(), "main");
        assert!(mtls::cert_left(&kept, 0).unwrap() > mtls::cert_left(&first, 0).unwrap(), "a later end");
        // Another key's certificate is no renewal.
        let (other_csr, _) = mtls::box_request().unwrap();
        let other = ca.sign_box(&other_csr, "main", 180 * 24 * H).unwrap();
        assert!(c.renew(&other).unwrap_err().contains("renew"), "refused by name");
        assert_eq!(vm.own_cert.lock().unwrap().clone().unwrap(), renewed, "nothing changed");
        // The estate-wide pass: not due now, due at an advanced clock.
        let m = Material { ca: ca.clone(), me: me.clone(), link: None };
        let now = mtls::now_unix();
        let r = renew(&m, vec![("main".into(), Box::new(DirectPath { at }) as Box<dyn Path>)], 180 * 24 * H, now);
        assert!(matches!(r[0].1, Ok(false)), "not due: {:?}", r[0].1);
        let r = renew(&m, vec![("main".into(), Box::new(DirectPath { at }) as Box<dyn Path>)], 180 * 24 * H, now + 121 * 24 * 3600);
        assert!(matches!(r[0].1, Ok(true)), "due at day 121 of 180: {:?}", r[0].1);
        assert_ne!(vm.own_cert.lock().unwrap().clone().unwrap(), renewed, "renewed again");
    }

    /// ★ **Renewal is due at two thirds of a certificate's life** — read off the certificate,
    /// at a clock the caller advances.
    #[test]
    fn renewal_is_due_at_two_thirds_of_a_certificates_life() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let (csr, _) = mtls::box_request().unwrap();
        let cert = ca.sign_box(&csr, "twin", Duration::from_secs(90 * 24 * 3600)).unwrap();
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() as i64;
        assert!(!mtls::renewal_due(&cert, now).unwrap(), "fresh");
        assert!(!mtls::renewal_due(&cert, now + 59 * 24 * 3600).unwrap(), "day 59 of 90");
        assert!(mtls::renewal_due(&cert, now + 61 * 24 * 3600).unwrap(), "day 61 of 90: renew");
        assert!(mtls::renewal_due(&cert, now + 120 * 24 * 3600).unwrap(), "expired: overdue");
    }

    /// ★ **A renewal hands the VM a FRESH bootstrap pair for its next join** (0.5.13, the 180-day
    /// gap: a box upgraded in place re-joins with the bootstrap its seed keeps, and the medium's
    /// ends with the material): a bootstrap certificate from the CA, its key with it; a pair whose
    /// key is not the certificate's, or a certificate that is not a bootstrap one, is refused.
    #[test]
    fn a_renewal_hands_a_closed_vm_a_fresh_bootstrap_for_its_next_join() {
        let ca = InstallerCa::mint("gunnar.rs", H).unwrap();
        let boot = ca.issue_bootstrap(H).unwrap();
        let (vm, at, _h) = spawn_vm_mode(&ca, &boot, 4, Kind::Gunnar, crate::bootstrap::SecretOffer::default(), Mode::AfterClose);
        let (csr, _) = mtls::box_request_for(&vm.own_key).unwrap();
        *vm.own_cert.lock().unwrap() = Some(ca.sign_box(&csr, "main", 180 * 24 * H).unwrap());
        vm.closed.store(true, Ordering::SeqCst);
        let me = ca.issue_installer(H).unwrap();
        let m = Material { ca: ca.clone(), me: me.clone(), link: None };
        let due = mtls::now_unix() + 121 * 24 * 3600;
        let r = renew(&m, vec![("main".into(), Box::new(DirectPath { at }) as Box<dyn Path>)], 180 * 24 * H, due);
        assert!(matches!(r[0].1, Ok(true)), "{:?}", r[0].1);
        let fresh = vm.boot_now.lock().unwrap().clone().expect("a fresh bootstrap pair was handed over");
        assert_eq!(mtls::cert_cn(&fresh.cert_pem).unwrap(), mtls::BOOTSTRAP_CN);
        assert_ne!(fresh.key_pem, boot.key_pem, "its own new key");
        assert!(mtls::cert_left(&fresh.cert_pem, 0).unwrap() > mtls::cert_left(&boot.cert_pem, 0).unwrap(), "a later end");
        // A mismatched pair, and a certificate that is not a bootstrap one, are refused.
        let mut c = Client::new(Box::new(DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
        let other = ca.issue_bootstrap(H).unwrap();
        let mixed = Issued { cert_pem: fresh.cert_pem.clone(), key_pem: other.key_pem };
        assert!(c.rebootstrap(&mixed).unwrap_err().contains("bootstrap"), "a key that is not the certificate's");
        let not_boot = Issued { cert_pem: ca.issue_installer(H).unwrap().cert_pem, key_pem: me.key_pem.clone() };
        assert!(c.rebootstrap(&not_boot).unwrap_err().contains("bootstrap"), "not a bootstrap certificate");
        assert_eq!(vm.boot_now.lock().unwrap().clone().unwrap(), fresh, "nothing changed");
    }

}