nordisk-pairing 0.5.2

//! **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
//!    listener goes, the installer is dropped as a peer, and the installer
//!    proves the VM no longer answers ([`Path::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."];
/// Whole feed keys only the front may hold (see [`FRONT_ONLY_PREFIXES`]).
pub const FRONT_ONLY_KEYS: &[&str] = &["console_admin_token"];

/// 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`)"));
        }
        reach(self.tunnel).map_err(|e| e.to_string())?;
        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) => {
                let f: LinkFile = toml::from_str(&t).map_err(|e| format!("REFUSED [installer-material] {}: {e}", dir.join(LINK_KEYS).display()))?;
                Some(LinkKeys { installer_private: b64_32(&f.installer_private, "installer_private")?, vm_public: b64_32(&f.vm_public, "vm_public")? })
            }
            Err(_) => None,
        };
        Ok(Material { ca, me, link })
    }

    /// 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(|_| ())
    }

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

/// **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 {
    cursor: u64,
    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>,
}

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

/// **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"));
        }
    }
    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.
fn peers_complete(d: &Door, tunnel: TunnelKind) -> bool {
    d.ledger.complete() && d.verdict.is_some() && (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();
    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())?
        }
        Err(e) => return Err(e.to_string()),
    };
    if *door != before {
        sink.ledger(&door);
    }
    Ok(reply)
}

fn store_lines(shared: &Shared, sink: &dyn Sink, name: &str, chunk: &Chunk) {
    let lines: Vec<String> = chunk.stored().iter().map(|l| crate::redact::clean(l).into_owned()).collect();
    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();
    }
}

/// 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 from = shared.progress.lock().unwrap_or_else(|p| p.into_inner()).get(name).map(|p| p.cursor).unwrap_or(0);
        let chunk = c.log(from)?;
        store_lines(shared, sink, name, &chunk);
        shared.progress.lock().unwrap_or_else(|p| p.into_inner()).entry(name.to_string()).or_default().cursor = chunk.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)?;
    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)?;
        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.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 on the installer's path (verified)" } else { "BUT the VM still answered after the close" }
    ));
    Ok(())
}

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 started = Instant::now();
    let budget = Duration::from_secs(plan.budget_s);
    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));
    }
    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) {
                    if shared.closing.load(Ordering::SeqCst) {
                        match close_one(shared, sink, &mut c, &vm.name) {
                            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;
                            }
                        }
                    }
                    match visit(shared, plan, sink, &mut c, &vm) {
                        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.
fn write_0600(path: &FsPath, text: &str) -> Result<(), String> {
    let tmp = path.with_extension("toml.part");
    let _ = std::fs::remove_file(&tmp);
    let mut o = std::fs::OpenOptions::new();
    o.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        o.mode(0o600);
    }
    let mut f = o.open(&tmp).map_err(|e| format!("{}: {e}", tmp.display()))?;
    f.write_all(text.as_bytes()).and_then(|_| f.sync_all()).map_err(|e| format!("{}: {e}", tmp.display()))?;
    drop(f);
    std::fs::rename(&tmp, path).map_err(|e| format!("{} → {}: {e}", tmp.display(), 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)));
        }
        let tmp = self.dir.join("status.toml.part");
        if std::fs::write(&tmp, out).is_ok() {
            let _ = std::fs::rename(tmp, self.dir.join("status.toml"));
        }
    }
}

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 tmp = self.dir.join("offers.toml.part");
        if std::fs::write(&tmp, toml::to_string(&*all).unwrap_or_default()).is_ok() {
            let _ = std::fs::rename(tmp, self.dir.join(OFFERS_FILE));
        }
    }

    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::*;
    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,
    }

    impl Vm for TestVm {
        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 == Kind::Gunnar).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.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))])
                }
                _ => 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<()>) {
        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 from the env");
        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),
        });
        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 || crate::bootstrap::serve(v.as_ref(), &log, &ca_pem, l, &|_| {}));
        (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");
        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.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"));
    }

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

    /// 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");
    }
}