nordisk-pairing 0.5.29

//! **The VM's side of the installer: the bootstrap listener** (0.3.0).
//!
//! The owner, 2026-09-24: the installer runs FIRST and lives on the
//! operator's machine — which is behind NAT, so the direction is reversed.
//! The VMs do not dial a join service on the front; every VM of the medium
//! opens THIS listener from its first moment, and the installer connects OUT
//! to it, retrying until each VM answers.
//!
//! Mutual TLS both ways under the installer CA:
//!
//! * the VM shows the medium-wide bootstrap certificate (or, after a restart,
//!   the certificate the installer signed for it), named
//!   [`crate::mtls::VM_SERVER_NAME`];
//! * the installer shows its own client certificate, CN
//!   [`crate::mtls::INSTALLER_CN`], and the VM serves THAT ROLE ONLY — the
//!   bootstrap certificate on the medium, a box certificate, a stranger's:
//!   all refused by name (`installer-only`).
//!
//! Over it the installer, one request per connection (minimal HTTP/1.1 in
//! TLS 1.3, TOML bodies like the rest of the crate):
//!
//! | request | what the VM does |
//! |---|---|
//! | `GET /install/hello` | presents itself: the [`JoinRequest`] (kind, tunnel key, capacity, a signing request for a key it minted) |
//! | `GET /install/log?from=N` | its boot log from line `N` ([`crate::bootlog`]) |
//! | `POST /install/feed` | takes the [`JoinReply`]: its name, tunnel address, own certificate, role, peers, the estate values |
//! | `GET /install/status` | its pairing [`Signal`] |
//! | `GET /install/offer` | what the VM OFFERS the installer (0.3.1): public values only it can mint — the primary's ticket key, the twin's replication key and signing request, the certificate the primary signed for its twin |
//! | `GET /install/secret-offer` | what the VM HANDS OVER as a secret (0.5.0, [`SecretOffer`]): the front's certificate and key, for its successor — asked once, before the close; never fed on, never logged |
//! | `POST /install/close` | closes the listener for good: the bootstrap certificate is dead from then on |
//! | `POST /install/refeed` | AFTER the close only (0.5.8, [`Vm::refed`]): a later answer for the same VM — same estate, name, overlay and role, no certificate — when ONE box of the estate was re-admitted ([`crate::installer::readmit`]) and the others must learn its new address and key |
//!
//! **The offer is fed onward** (0.3.1): the installer feeds every VM's offer
//! to every VM of the run (the offering one too), each value under [`fed_offer_key`]
//! (`offer.<vm>.<key>`), and hands it to its [`crate::installer::Sink::offer`].
//! That is how two VMs that do not yet reach each other trade what only they
//! can mint, with the installer — already trusted both ways — as the carrier.
//!
//! What the VM does with a feed, how it keeps its log and how it remembers
//! that it was closed is the VM's ([`Vm`]); the wire and the one-role rule
//! are here, shared by gunnar and holger.

use crate::bootlog::{BootLog, MAX_READ};
use crate::door::{parse_request, Request, Response};

/// **How large one request to a VM's bootstrap listener may be: 16 MiB**
/// (0.5.1). A FEED carries vm1's own values since 0.5.0 — the front's
/// monetize plugins among them, ≈0.8 MB base64 each — and the door's 64 KiB
/// ([`crate::door::MAX_REQUEST_BYTES`], a JOIN) refused every feed to a paid
/// front: `request too large` (MEASURED 2026-09-25, lane B's bench). The
/// listener answers only over the installer's WireGuard link, under mutual TLS
/// on the installer CA, so the bound is against a runaway, not a stranger.
pub const MAX_FEED_BYTES: usize = 16 * 1024 * 1024;
use crate::join::{JoinReply, JoinRequest};
use crate::mtls;
use crate::signal::Signal;
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::time::Duration;

/// **The installer's mTLS endpoint on every VM.** The port the front's join
/// service used until 0.2 — the VM now listens on it, and the firewall opens
/// it to the operator's public address only.
pub const PORT: u16 = 8474;

pub const HELLO_PATH: &str = "/install/hello";
pub const LOG_PATH: &str = "/install/log";
pub const FEED_PATH: &str = "/install/feed";
pub const STATUS_PATH: &str = "/install/status";
pub const CLOSE_PATH: &str = "/install/close";
/// 0.3.1: what the VM offers the installer ([`Vm::offer`]).
pub const OFFER_PATH: &str = "/install/offer";

/// 0.5.0: what the VM hands over as a SECRET ([`Vm::secret_offer`]).
pub const SECRET_OFFER_PATH: &str = "/install/secret-offer";

/// 0.5.8: a CLOSED VM takes a later answer for itself ([`Vm::refed`]) — the
/// one install request its log door answers besides the log.
pub const REFEED_PATH: &str = "/install/refeed";
/// **The VM's own certificate, as it serves it** (0.5.12): PEM. A closed VM's, for its renewal.
pub const CERT_PATH: &str = "/install/cert";
/// **A renewed certificate for the VM's own key** (0.5.12): POST, the PEM; the VM keeps it
/// ([`Vm::renewed`]) once [`crate::mtls::renewal_refusal`] has no word against it.
pub const RENEW_PATH: &str = "/install/renew";
/// **A fresh bootstrap certificate and key for the VM's NEXT join** (0.5.13): POST, the pair as
/// TOML (`cert_pem`, `key_pem`). A box upgraded in place re-joins with the bootstrap its seed
/// keeps; the medium's ends with the material (180 days), so the installer refreshes it with
/// each renewal ([`Vm::rebootstrapped`]).
pub const BOOTSTRAP_PATH: &str = "/install/bootstrap";

/// **Pure: why a closed VM must refuse a re-feed**, or `None` (0.5.8). A
/// re-feed tells a VM what CHANGED around it — a re-admitted box's new
/// addresses and tunnel key, a re-laid peer — and never what the VM IS: the
/// estate, its name, its overlay address and its role are the install's, a
/// role is a bootstrap fact, and a closed VM keeps the certificate it was
/// signed (the bootstrap certificate is dead on it, so nothing could replace
/// it honestly).
pub fn refeed_refusal(kept: Option<&JoinReply>, new: &JoinReply) -> Option<String> {
    let Some(kept) = kept else {
        return Some("REFUSED [refeed-never-fed] this VM was never fed; a re-feed changes what an install fed, it is not an install".into());
    };
    let why = if new.estate != kept.estate {
        format!("the re-feed is for estate `{}`, this VM is of `{}`", new.estate, kept.estate)
    } else if new.me.name != kept.me.name {
        format!("the re-feed names this VM `{}`, the install named it `{}`", new.me.name, kept.me.name)
    } else if new.me.overlay != kept.me.overlay {
        format!("the re-feed moves `{}` to overlay {}, the install gave it {}", kept.me.name, new.me.overlay, kept.me.overlay)
    } else if new.role != kept.role {
        format!("the re-feed says role {:?}, the install elected {:?} — a role is a bootstrap fact", new.role, kept.role)
    } else if new.cert_pem.is_some() {
        "the re-feed carries a certificate; a closed VM keeps the one its install signed".to_string()
    } else {
        return None;
    };
    Some(format!("REFUSED [refeed-changes-the-vm] {why}"))
}

/// **A secret offer** (0.5.0, owner 2026-09-25): values a VM hands the
/// installer that must survive it — the front's TLS certificate AND ITS
/// PRIVATE KEY (`tls.<host>.cert`, `tls.<host>.key`), so the NEXT front of
/// the estate is fed them and does not spend one of Let's Encrypt's five
/// certificates a week. Marked secret by its own path and its own type:
///
/// * the installer asks for it at the END of a VM's install, right before the
///   close ([`crate::installer`]), and hands it to
///   [`crate::installer::Sink::secret_offer`] — ONLY there;
/// * it is never fed to another VM (no `offer.<vm>.<key>`), never in the
///   door's feed, the ledger, `offers.toml`, the narration or a log — the
///   narration names its keys, never a value;
/// * [`crate::installer::FolderSink`] keeps it in `<run>/secret-offers.toml`,
///   0600; the operator's next run feeds it to the successor VM with
///   [`crate::installer::successor_feed`] (`PlanVm::feed`, that VM only).
///
/// `Debug` prints the keys only.
#[derive(Clone, Default, PartialEq, Eq)]
pub struct SecretOffer(BTreeMap<String, String>);

impl SecretOffer {
    pub fn new(values: BTreeMap<String, String>) -> SecretOffer {
        SecretOffer(values)
    }
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
    pub fn len(&self) -> usize {
        self.0.len()
    }
    /// The keys — the only part that may be said.
    pub fn keys(&self) -> Vec<String> {
        self.0.keys().cloned().collect()
    }
    /// The values, for the one place that stores them.
    pub fn expose(&self) -> &BTreeMap<String, String> {
        &self.0
    }
    pub fn into_inner(self) -> BTreeMap<String, String> {
        self.0
    }
    /// The wire body (a TOML table of strings, like the public offer).
    pub fn render(&self) -> String {
        toml::to_string(&self.0).unwrap_or_default()
    }
    /// The wire body back. The error never quotes the body.
    pub fn parse(text: &str) -> Result<SecretOffer, String> {
        toml::from_str(text).map(SecretOffer).map_err(|_| "not a secret offer (a TOML table of strings); the body is not quoted".to_string())
    }
}

impl std::fmt::Debug for SecretOffer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "SecretOffer({:?} — values withheld)", self.keys())
    }
}

/// The feed key an offered value travels under to the other VMs:
/// `offer.<vm>.<key>` — `vm` the offering VM's name as the installer knows it.
pub fn fed_offer_key(vm: &str, key: &str) -> String {
    format!("offer.{vm}.{key}")
}

/// Every value `vm` offered, out of a feed ([`fed_offer_key`] undone).
pub fn offers_in(feed: &BTreeMap<String, String>, vm: &str) -> BTreeMap<String, String> {
    let prefix = format!("offer.{vm}.");
    feed.iter().filter_map(|(k, v)| k.strip_prefix(&prefix).map(|k| (k.to_string(), v.clone()))).collect()
}

/// An offer as its wire body (a TOML table of strings).
pub fn render_offer(offer: &BTreeMap<String, String>) -> String {
    toml::to_string(offer).unwrap_or_default()
}

/// The wire body back. An empty body is an empty offer.
pub fn parse_offer(text: &str) -> Result<BTreeMap<String, String>, String> {
    toml::from_str(text).map_err(|e| format!("not an offer: {e}"))
}

/// **What a VM does when the installer asks.** Implemented by gunnar-server
/// (PID 1) and by the fleet test's probe.
pub trait Vm: Send + Sync {
    /// Present this VM: estate, kind, tunnel key, capacity, and — while it
    /// has no certificate of its own — a signing request. `Err` = not ready
    /// yet (a transient: the installer asks again).
    fn hello(&self) -> Result<JoinRequest, String>;
    /// Take what the installer fed. Called again whenever the installer's
    /// answer changes (the role is decided, a peer's key arrives).
    fn fed(&self, reply: &JoinReply) -> Result<(), String>;
    /// Where this VM's pairing is.
    fn status(&self) -> Signal;
    /// Remember, durably, that the listener was closed — it never opens again
    /// on this installation.
    fn close(&self) -> Result<(), String>;
    /// The certificate this VM shows the installer NOW: the medium's
    /// bootstrap certificate until the installer has signed the VM's own.
    fn identity(&self) -> mtls::Issued;
    /// **What this VM offers the installer** (0.3.1): PUBLIC values only it
    /// can mint, as they come into being (empty until then). The installer
    /// hands them to its sink and feeds them to the other VMs under
    /// [`fed_offer_key`]. Never a private key: the offer is logged.
    fn offer(&self) -> BTreeMap<String, String> {
        BTreeMap::new()
    }
    /// **What this VM hands over as a SECRET** (0.5.0, [`SecretOffer`]):
    /// values its SUCCESSOR must be fed — the front's current certificate and
    /// key per host. Asked once, at the end of the install, right before the
    /// close; kept by the operator only. Empty by default (the boxes).
    fn secret_offer(&self) -> SecretOffer {
        SecretOffer::default()
    }
    /// **A later answer, after the close** (0.5.8, [`REFEED_PATH`]): take it
    /// exactly as [`Vm::fed`] would, once [`refeed_refusal`] found nothing.
    /// Refused by default: a VM that does not keep its feed cannot be re-fed.
    fn refed(&self, _reply: &JoinReply) -> Result<(), String> {
        Err("REFUSED [refeed-unsupported] this VM takes no re-feed after its close".into())
    }

    /// **A renewed certificate** (0.5.12): the installer re-signed this VM's OWN key for a new
    /// lifetime ([`crate::mtls::InstallerCa::renew_box`]); keep it and serve it from the next
    /// connection. The default refuses: a VM that cannot keep one says so by name.
    fn renewed(&self, _cert_pem: &str) -> Result<(), String> {
        Err("REFUSED [renew-not-kept] this VM keeps no renewed certificate".into())
    }

    /// **A fresh bootstrap pair for the next join** (0.5.13, [`BOOTSTRAP_PATH`]): keep it where
    /// the seed keeps its bootstrap. The default refuses by name.
    fn rebootstrapped(&self, _boot: &mtls::Issued) -> Result<(), String> {
        Err("REFUSED [rebootstrap-not-kept] this VM keeps no fresh bootstrap".into())
    }

    /// **What this VM runs** (live update, [`crate::live::RUNNING_PATH`]). The default refuses.
    fn live_running(&self) -> Result<crate::live::Running, String> {
        Err("REFUSED [live-unsupported] this VM takes no live update".into())
    }

    /// **A new app for this VM's other slot** (live update, [`crate::live::STAGE_PATH`]):
    /// verified and staged, then the VM re-execs into it; answers what it did. The default
    /// refuses.
    fn live_stage(&self, _stage: &crate::live::Stage) -> Result<String, String> {
        Err("REFUSED [live-unsupported] this VM takes no live update".into())
    }

    /// **A new app with what its release declares** (lane B5, [`crate::live::Declared`]). The
    /// default takes an undeclared stage as [`Vm::live_stage`] and refuses a declared one by
    /// name (`[live-declared-unsupported]`, which a front answers with a keep boot): a VM that
    /// cannot lay the declaration never drops it silently.
    fn live_stage_declared(&self, stage: &crate::live::Stage, declared: &crate::live::Declared) -> Result<String, String> {
        if declared.is_empty() {
            self.live_stage(stage)
        } else {
            Err(crate::live::declared_unsupported())
        }
    }
}

/// The one-role rule, as the refusal a VM answers with.
pub fn installer_only(cn: &str) -> String {
    format!(
        "REFUSED [installer-only] this is a VM's bootstrap listener; it serves the installer's own certificate (CN `{}`) and nothing else — `{cn}` was presented",
        mtls::INSTALLER_CN
    )
}

fn query_from(path: &str) -> u64 {
    path.split_once('?')
        .map(|(_, q)| q)
        .unwrap_or("")
        .split('&')
        .find_map(|kv| kv.strip_prefix("from="))
        .and_then(|v| v.parse().ok())
        .unwrap_or(0)
}

/// **One request, answered** — pure over the VM, its log and the request.
/// The second value says whether the listener was closed by it.
pub fn respond(vm: &dyn Vm, log: &BootLog, req: &Request) -> (Response, bool) {
    let path = req.path.split('?').next().unwrap_or("");
    let ok = |body: String| Response { status: 200, body };
    match (req.method.as_str(), path) {
        ("GET", HELLO_PATH) => match vm.hello().and_then(|h| h.render().map_err(|e| e.to_string())) {
            Ok(t) => (ok(t), false),
            Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
        },
        ("GET", LOG_PATH) => (ok(log.since(query_from(&req.path), MAX_READ).render()), false),
        ("POST", FEED_PATH) => match JoinReply::parse(&String::from_utf8_lossy(&req.body)) {
            Err(e) => (Response { status: 400, body: format!("{e}\n") }, false),
            Ok(reply) => match vm.fed(&reply) {
                Ok(()) => (ok(format!("fed `{}`\n", reply.me.name)), false),
                Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
            },
        },
        ("GET", STATUS_PATH) => (ok(vm.status().render().unwrap_or_default()), false),
        ("GET", OFFER_PATH) => (ok(render_offer(&vm.offer())), false),
        ("GET", SECRET_OFFER_PATH) => (ok(vm.secret_offer().render()), false),
        ("POST", CLOSE_PATH) => match vm.close() {
            // The install is over: what the VM said up to here is the log's head, kept
            // for life (Gunnar f21d8d09, shared 0.5.23). Sealed before the answer, so
            // nothing the installer does after its close lands in the head.
            Ok(()) => {
                log.seal_head();
                (ok("closed: the bootstrap certificate is dead on this VM\n".into()), true)
            }
            Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
        },
        (_, HELLO_PATH | LOG_PATH | FEED_PATH | STATUS_PATH | OFFER_PATH | SECRET_OFFER_PATH | CLOSE_PATH) => (Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() }, false),
        _ => (Response { status: 404, body: format!("REFUSED [install-path] this listener has {HELLO_PATH}, {LOG_PATH}, {FEED_PATH}, {STATUS_PATH}, {OFFER_PATH}, {SECRET_OFFER_PATH} and {CLOSE_PATH}\n") }, false),
    }
}

/// **The pairing steps the listener says itself**, each ONCE per listener
/// life (owner 2026-09-30: every pairing step is its own line on the right
/// box): the hello it answered and the secret offer it handed over. What a VM
/// does with a feed, and the close, are said where they happen (the product's
/// `Vm::fed`, the loops below).
#[derive(Debug, Default)]
pub struct StepsSaid(std::sync::Mutex<std::collections::BTreeSet<&'static str>>);

impl StepsSaid {
    /// The line to say for an answered request, the first time only.
    pub fn step(&self, method: &str, path: &str, status: u16) -> Option<&'static str> {
        if status != 200 {
            return None;
        }
        let path = path.split('?').next().unwrap_or("");
        let line = match (method, path) {
            ("GET", HELLO_PATH) => "hello — this VM presented itself to the installer (its kind, tunnel key, capacity and a signing request for its own key)",
            ("GET", SECRET_OFFER_PATH) => "successor-offer — this VM handed the installer what its successor takes (the values are never logged)",
            _ => return None,
        };
        self.0.lock().unwrap_or_else(|p| p.into_inner()).insert(line).then_some(line)
    }
}

/// [`respond`], saying each pairing step once ([`StepsSaid`]).
fn respond_saying(vm: &dyn Vm, log: &BootLog, req: &Request, said: &StepsSaid, say: &dyn Fn(Tone, &str)) -> (Response, bool) {
    let (r, closed) = respond(vm, log, req);
    if let Some(l) = said.step(&req.method, &req.path, r.status) {
        say(Tone::Info, l);
    }
    (r, closed)
}

/// **How loud one line of the listener is** (the `_at` loops): a refusal is a
/// `Warn`; the listener doing its job — the close, the log door opening, a
/// reader going away — is `Info`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Tone {
    Info,
    Warn,
}

/// One socket read's timeout while a request is awaited. A timeout is "not
/// yet" (EAGAIN, os error 11), read again until [`REQUEST_BUDGET`].
#[cfg(not(test))]
pub const REQUEST_READ_TIMEOUT: Duration = Duration::from_secs(10);
#[cfg(test)]
pub const REQUEST_READ_TIMEOUT: Duration = Duration::from_millis(300);
/// How long a connection may take in all to send its request.
pub const REQUEST_BUDGET: Duration = Duration::from_secs(60);

/// Why one connection ended without an answer.
#[derive(Debug)]
enum ConnError {
    /// The reader went away (reset, closed, silent past the budget): not a refusal.
    Gone(String),
    /// The listener refused it (its certificate, its request).
    Refused(String),
}

impl ConnError {
    fn io(from: &str, e: std::io::Error) -> ConnError {
        use std::io::ErrorKind::*;
        let line = format!("{from}: {e}");
        match e.kind() {
            ConnectionReset | ConnectionAborted | BrokenPipe | UnexpectedEof | NotConnected | WouldBlock | TimedOut => ConnError::Gone(line),
            _ => ConnError::Refused(line),
        }
    }

    fn line(self) -> String {
        match self {
            ConnError::Gone(l) | ConnError::Refused(l) => l,
        }
    }
}

/// **The line a VM says ONCE, the first time the installer's own certificate
/// reached its listener** (DEV-NARRATE, 2026-09-30): the proof, in the box's
/// own log, that it got past its WireGuard link and mutual TLS. The installer's
/// narration (nordisk-narrate) reads a box log WITHOUT it as "stopped before
/// WireGuard" and names the box's last stage.
pub const REACHED: &str = "the installer reached this VM over its link (mutual TLS, the installer's own certificate)";

/// **Serve one connection**: the handshake (a client certificate from the
/// installer CA REQUIRED), the one-role check, one request, one answer.
/// Returns whether the listener was closed by it.
pub fn serve_one(vm: &dyn Vm, log: &BootLog, cfg: Arc<rustls::ServerConfig>, stream: TcpStream) -> Result<bool, String> {
    serve_one_with(cfg, stream, &|req| respond(vm, log, req)).map_err(ConnError::line)
}

/// [`serve_one`] over any answer: the TLS, the one-role rule, one request.
///
/// H (release-3 mocks 2026-09-29): a read timeout is `WouldBlock` — "not yet",
/// read again until [`REQUEST_BUDGET`]; it was `Resource temporarily
/// unavailable (os error 11)`, a refusal, for a reader slower than ONE timeout.
fn serve_one_with(cfg: Arc<rustls::ServerConfig>, stream: TcpStream, answer: &dyn Fn(&Request) -> (Response, bool)) -> Result<bool, ConnError> {
    let from = stream.peer_addr().map(|a| a.to_string()).unwrap_or_else(|_| "?".into());
    let _ = stream.set_read_timeout(Some(REQUEST_READ_TIMEOUT));
    let _ = stream.set_write_timeout(Some(Duration::from_secs(20)));
    let conn = rustls::ServerConnection::new(cfg).map_err(|e| ConnError::Refused(e.to_string()))?;
    let mut tls = rustls::StreamOwned::new(conn, stream);
    let mut bytes = Vec::new();
    let mut chunk = [0u8; 4096];
    let started = std::time::Instant::now();
    let req = loop {
        let n = match tls.read(&mut chunk) {
            Ok(n) => n,
            Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut | std::io::ErrorKind::Interrupted) && started.elapsed() < REQUEST_BUDGET => continue,
            Err(e) => return Err(ConnError::io(&from, e)),
        };
        if n == 0 {
            return Err(ConnError::Gone(format!("{from}: closed before a request")));
        }
        bytes.extend_from_slice(&chunk[..n]);
        // A live update's app is larger than any feed (crate::live::MAX_STAGE_BYTES).
        let bound = if bytes.starts_with(format!("POST {} ", crate::live::STAGE_PATH).as_bytes()) { crate::live::MAX_STAGE_BYTES } else { MAX_FEED_BYTES };
        if bytes.len() > bound {
            return Err(ConnError::Refused(format!("{from}: request too large")));
        }
        if let Some(r) = parse_request(&bytes) {
            break r.map_err(ConnError::Refused)?;
        }
    };
    let cn = mtls::peer_cn(&tls.conn).unwrap_or_default();
    let (resp, closed) = if cn == mtls::INSTALLER_CN {
        answer(&req)
    } else {
        (Response { status: 403, body: format!("{}\n", installer_only(&cn)) }, false)
    };
    tls.write_all(&resp.bytes()).map_err(|e| ConnError::io(&from, e))?;
    tls.conn.send_close_notify();
    let _ = tls.flush();
    if cn != mtls::INSTALLER_CN {
        return Err(ConnError::Refused(format!("{from}: {}", installer_only(&cn))));
    }
    Ok(closed)
}

/// **After the close: the log door** (0.5.4, owner 2026-09-27 — "THE
/// INSTALLER WILL RECEIVE THE LOGS", "standard for all using the installer").
/// The same listener, the same link, the same one-role rule, but ONE request
/// left: `GET /install/log?from=N`. Everything the install needed answers
/// `410 REFUSED [install-closed]` — hello, feed, status, the offers, close.
/// So the operator reads a box's log for its whole life, with the installer's
/// own certificate, and nobody else ever can.
pub fn respond_closed(log: &BootLog, req: &Request) -> Response {
    let path = req.path.split('?').next().unwrap_or("");
    match (req.method.as_str(), path) {
        ("GET", LOG_PATH) => Response { status: 200, body: log.since(query_from(&req.path), MAX_READ).render() },
        (_, LOG_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
        (_, HELLO_PATH | FEED_PATH | STATUS_PATH | OFFER_PATH | SECRET_OFFER_PATH | CLOSE_PATH) => {
            Response { status: 410, body: format!("REFUSED [install-closed] this VM's install is closed; only {LOG_PATH} is left\n") }
        }
        _ => Response { status: 404, body: format!("REFUSED [install-path] this closed listener has {LOG_PATH} only\n") },
    }
}

/// **The log door of `vm`** (0.5.8): [`respond_closed`], and a re-feed
/// ([`REFEED_PATH`], [`Vm::refed`]) — the one change a closed VM takes, from
/// the installer's own certificate only (the one-role rule of the listener).
pub fn respond_closed_to(vm: &dyn Vm, log: &BootLog, req: &Request) -> Response {
    let path = req.path.split('?').next().unwrap_or("");
    match (req.method.as_str(), path) {
        ("POST", REFEED_PATH) => match JoinReply::parse(&String::from_utf8_lossy(&req.body)) {
            Err(e) => Response { status: 400, body: format!("{e}\n") },
            Ok(reply) => match vm.refed(&reply) {
                Ok(()) => Response { status: 200, body: format!("re-fed `{}`\n", reply.me.name) },
                Err(e) => Response { status: 409, body: format!("{e}\n") },
            },
        },
        (_, REFEED_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
        ("GET", CERT_PATH) => Response { status: 200, body: vm.identity().cert_pem },
        ("POST", RENEW_PATH) => {
            let pem = String::from_utf8_lossy(&req.body).to_string();
            if let Some(why) = mtls::renewal_refusal(&vm.identity().cert_pem, &pem) {
                return Response { status: 409, body: format!("{why}\n") };
            }
            match vm.renewed(&pem) {
                Ok(()) => Response { status: 200, body: "renewed\n".into() },
                Err(e) => Response { status: 409, body: format!("{e}\n") },
            }
        }
        ("POST", BOOTSTRAP_PATH) => {
            #[derive(serde::Deserialize)]
            struct Pair {
                cert_pem: String,
                key_pem: String,
            }
            let boot = match toml::from_str::<Pair>(&String::from_utf8_lossy(&req.body)) {
                Ok(p) => mtls::Issued { cert_pem: p.cert_pem, key_pem: p.key_pem },
                Err(e) => return Response { status: 400, body: format!("REFUSED [rebootstrap-unreadable] not a bootstrap pair: {e}\n") },
            };
            if let Some(why) = mtls::bootstrap_refusal(&boot) {
                return Response { status: 409, body: format!("{why}\n") };
            }
            match vm.rebootstrapped(&boot) {
                Ok(()) => Response { status: 200, body: "bootstrap kept\n".into() },
                Err(e) => Response { status: 409, body: format!("{e}\n") },
            }
        }
        ("GET", crate::live::RUNNING_PATH) => match vm.live_running() {
            Ok(r) => Response { status: 200, body: r.render() },
            Err(e) => Response { status: 409, body: format!("{e}\n") },
        },
        ("POST", crate::live::STAGE_PATH) => match crate::live::Stage::decode_declared(&String::from_utf8_lossy(&req.body)) {
            Err(e) => Response { status: 400, body: format!("REFUSED {e}\n") },
            Ok((stage, declared)) => match vm.live_stage_declared(&stage, &declared) {
                Ok(said) => Response { status: 200, body: format!("{said}\n") },
                Err(e) => Response { status: 409, body: format!("{e}\n") },
            },
        },
        (_, crate::live::RUNNING_PATH | crate::live::STAGE_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
        (_, CERT_PATH | RENEW_PATH | BOOTSTRAP_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
        _ => respond_closed(log, req),
    }
}

/// The certificate the listener shows follows `Vm::identity` (the bootstrap
/// one until the installer signs the VM's own), rebuilt only when it changes.
fn config_for(vm: &dyn Vm, ca_pem: &str, current: &mut Option<(String, Arc<rustls::ServerConfig>)>, say: &dyn Fn(&str)) -> Option<Arc<rustls::ServerConfig>> {
    let id = vm.identity();
    match current {
        Some((pem, cfg)) if *pem == id.cert_pem => Some(cfg.clone()),
        _ => match mtls::server_config(ca_pem, &id) {
            Ok(cfg) => {
                *current = Some((id.cert_pem.clone(), cfg.clone()));
                Some(cfg)
            }
            Err(e) => {
                say(&format!("the bootstrap listener cannot show its certificate: {e}"));
                None
            }
        },
    }
}

/// **The listener's loop**: one connection at a time until the installer
/// closes it. Returns once closed. `say` hears every refused connection, in
/// one line. A VM keeps the listener with [`serve_for_life`] instead.
pub fn serve(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
    serve_until_closed(vm, log, ca_pem, &listener, &|_, l| say(l))
}

fn serve_until_closed(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str)) {
    // "this is `main` again", not "the bootstrap certificate again".
    let mut current = None;
    let said = StepsSaid::default();
    let mut reached = false;
    for stream in listener.incoming() {
        let Ok(s) = stream else { continue };
        let Some(cfg) = config_for(vm, ca_pem, &mut current, &|l| say(Tone::Warn, l)) else { continue };
        let served = serve_one_with(cfg, s, &|req| respond_saying(vm, log, req, &said, say));
        if served.is_ok() && !reached {
            reached = true;
            say(Tone::Info, REACHED);
        }
        match served {
            Ok(true) => {
                say(Tone::Info, "the installer closed the bootstrap listener: the bootstrap certificate is dead on this VM");
                return;
            }
            Ok(false) => {}
            Err(ConnError::Gone(e)) => say(Tone::Info, &format!("a reader of the bootstrap listener went away before its answer: {e}")),
            Err(ConnError::Refused(e)) => say(Tone::Warn, &format!("a connection to the bootstrap listener was refused: {e}")),
        }
    }
}

/// **The log door's loop** ([`respond_closed`]): for as long as the listener
/// lives. Never returns while it accepts.
pub fn serve_log_door(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(&str)) {
    serve_log_door_at(vm, log, ca_pem, listener, &|_, l| say(l))
}

/// [`serve_log_door`], each line with its [`Tone`].
///
/// I (release-3 mocks 2026-09-29): the installer's reader going away
/// (`Connection reset by peer (os error 104)`, `unexpected end of file`, a
/// reader silent past [`REQUEST_BUDGET`]) was a WARN "refused" line for the
/// box's whole life. It is not a refusal: it is said ONCE, at INFO. A
/// certificate that is not the installer's stays a WARN refusal, every time.
pub fn serve_log_door_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str)) {
    serve_log_door_said(vm, log, ca_pem, listener, say, false)
}

/// [`serve_log_door_at`], told whether [`REACHED`] was said already (by the
/// install before it, on this listener).
fn serve_log_door_said(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str), mut reached: bool) {
    let mut current = None;
    let mut gone_said = false;
    for stream in listener.incoming() {
        let Ok(s) = stream else { continue };
        let Some(cfg) = config_for(vm, ca_pem, &mut current, &|l| say(Tone::Warn, l)) else { continue };
        let served = serve_one_with(cfg, s, &|req| (respond_closed_to(vm, log, req), false));
        if served.is_ok() && !reached {
            reached = true;
            say(Tone::Info, REACHED);
        }
        match served {
            Ok(_) => {}
            Err(ConnError::Gone(e)) => {
                if !gone_said {
                    gone_said = true;
                    say(Tone::Info, &format!("a reader of the log door went away before its answer ({e}); the installer reads again, and this is said once"));
                }
            }
            Err(ConnError::Refused(e)) => say(Tone::Warn, &format!("a connection to the log door was refused: {e}")),
        }
    }
}

/// **The standard for every VM of an installer** (0.5.4): [`serve`] until the
/// installer closes, then [`serve_log_door`] on the same listener for life.
/// Keep the installer's link up with it — the log door is reached through it.
pub fn serve_for_life(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
    serve_for_life_at(vm, log, ca_pem, listener, &|_, l| say(l))
}

/// [`serve_for_life`], each line with its [`Tone`].
pub fn serve_for_life_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(Tone, &str)) {
    serve_until_closed(vm, log, ca_pem, &listener, say);
    say(Tone::Info, "the log door is open: the installer reads this VM's log over its link, for life");
    // The close was a request the installer made: it reached this VM already.
    serve_log_door_said(vm, log, ca_pem, &listener, say, true);
}

/// **A VM that was closed on an EARLIER start** (a restart after the install):
/// the log door only, at once. The install is not served again — `hello`,
/// `feed`, the offers and `close` answer `410 [install-closed]` — and the
/// installer still reads the whole log of this start. Without it a restarted
/// VM either opens no listener (the log is lost for the rest of its life) or
/// serves the whole install again until someone closes it a second time.
pub fn serve_after_close(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
    serve_after_close_at(vm, log, ca_pem, listener, &|_, l| say(l))
}

/// [`serve_after_close`], each line with its [`Tone`].
pub fn serve_after_close_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(Tone, &str)) {
    // A restart after the close: what this boot said before its door opened (the ESP's
    // story, the start-up) is its head; the rest is runtime (Gunnar f21d8d09).
    log.seal_head();
    say(Tone::Info, "the install closed on an earlier start: only the log door is open, the installer reads this VM's log over its link");
    serve_log_door_at(vm, log, ca_pem, &listener, say);
}

/// [`serve_for_life`] or [`serve_after_close`], by what the VM kept: `closed`
/// is the product's own record that the installer closed it (its kept file).
pub fn serve_for_life_or_after_close(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, closed: bool, say: &dyn Fn(&str)) {
    serve_for_life_or_after_close_at(vm, log, ca_pem, listener, closed, &|_, l| say(l))
}

/// [`serve_for_life_or_after_close`], each line with its [`Tone`].
pub fn serve_for_life_or_after_close_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, closed: bool, say: &dyn Fn(Tone, &str)) {
    if closed {
        serve_after_close_at(vm, log, ca_pem, listener, say)
    } else {
        serve_for_life_at(vm, log, ca_pem, listener, say)
    }
}

/// The address the listener binds: every interface, [`PORT`] — the firewall
/// decides who reaches it (the operator's public address only).
pub fn bind(port: u16) -> std::io::Result<TcpListener> {
    TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::elect::Capacity;
    use crate::seed::{Kind, TunnelKind};
    use std::sync::Mutex;

    struct FakeVm {
        fed: Mutex<Vec<JoinReply>>,
        closed: Mutex<bool>,
    }

    impl Vm for FakeVm {
        fn hello(&self) -> Result<JoinRequest, String> {
            Ok(JoinRequest { estate: "e".into(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: Some(Capacity { cores: 2, ram_bytes: 1, data_bytes: 1, disk_mib_s: 1 }), csr_pem: String::new() })
        }
        fn fed(&self, reply: &JoinReply) -> Result<(), String> {
            self.fed.lock().unwrap().push(reply.clone());
            Ok(())
        }
        fn status(&self) -> Signal {
            Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
        }
        fn close(&self) -> Result<(), String> {
            *self.closed.lock().unwrap() = true;
            Ok(())
        }
        fn identity(&self) -> crate::mtls::Issued {
            crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
        }
        fn offer(&self) -> BTreeMap<String, String> {
            BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())])
        }
    }

    fn get(path: &str) -> Request {
        Request { method: "GET".into(), path: path.into(), body: Vec::new() }
    }

    #[test]
    fn the_listener_answers_hello_log_feed_status_and_close_and_nothing_else() {
        let vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
        let log = BootLog::new(10);
        log.push("korp-installer: pass began\ngunnar: listening");
        let (r, c) = respond(&vm, &log, &get(HELLO_PATH));
        assert_eq!((r.status, c), (200, false));
        assert_eq!(JoinRequest::parse(&r.body).unwrap().tunnel_pubkey, "K");
        let (r, _) = respond(&vm, &log, &get(&format!("{LOG_PATH}?from=1")));
        let chunk = crate::bootlog::Chunk::parse(&r.body).unwrap();
        assert_eq!((chunk.lines, chunk.next), (vec!["gunnar: listening".to_string()], 2));
        let reply = JoinReply { estate: "e".into(), me: crate::join::Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_A }, cert_pem: None, role: None, primary_overlay: None, primary_utility: None, peers: vec![], feed: Default::default() };
        let (r, _) = respond(&vm, &log, &Request { method: "POST".into(), path: FEED_PATH.into(), body: reply.render().unwrap().into_bytes() });
        assert_eq!(r.status, 200, "{}", r.body);
        assert_eq!(vm.fed.lock().unwrap().len(), 1);
        let (r, _) = respond(&vm, &log, &get(STATUS_PATH));
        assert!(Signal::parse(&r.body).is_ok(), "{}", r.body);
        let (r, _) = respond(&vm, &log, &get(OFFER_PATH));
        assert_eq!(parse_offer(&r.body).unwrap()["ticket_pubkey"], "ab12");
        assert_eq!(parse_offer("").unwrap(), BTreeMap::new(), "an empty body is an empty offer");
        let feed = BTreeMap::from([(fed_offer_key("main", "ticket_pubkey"), "ab12".to_string()), (fed_offer_key("twin", "twin_pubkey"), "ssh-ed25519 X".to_string()), ("monetize_pubkey".to_string(), "m".to_string())]);
        assert_eq!(offers_in(&feed, "main"), BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())]));
        assert_eq!(offers_in(&feed, "twin")["twin_pubkey"], "ssh-ed25519 X");
        assert_eq!(respond(&vm, &log, &get(CLOSE_PATH)).0.status, 405);
        assert_eq!(respond(&vm, &log, &get("/pair/join")).0.status, 404);
        let (r, closed) = respond(&vm, &log, &Request { method: "POST".into(), path: CLOSE_PATH.into(), body: Vec::new() });
        assert!(r.status == 200 && closed && *vm.closed.lock().unwrap());
    }

    /// **The box says each pairing step it answers, once** (owner 2026-09-30:
    /// every pairing step is its own line on the right box): a hello answered
    /// is said once however often the installer asks, the secret offer once;
    /// a refused hello and the other paths are the product's to say.
    #[test]
    fn the_listener_says_the_hello_and_the_secret_offer_once() {
        let said = StepsSaid::default();
        let mut lines = Vec::new();
        for (m, p, st) in [("GET", HELLO_PATH, 200), ("GET", HELLO_PATH, 200), ("GET", LOG_PATH, 200), ("GET", HELLO_PATH, 503), ("GET", SECRET_OFFER_PATH, 200), ("GET", SECRET_OFFER_PATH, 200), ("POST", FEED_PATH, 200)] {
            if let Some(l) = said.step(m, p, st) {
                lines.push(l);
            }
        }
        assert_eq!(lines.len(), 2, "{lines:?}");
        assert!(lines[0].starts_with("hello — "), "{lines:?}");
        assert!(lines[1].starts_with("successor-offer — "), "{lines:?}");
        // ★ estate-box3 plain3: the step's own line reached the installer as
        // "<<WITHHELD: a token-class line>>" — the word "secret" is the
        // token class. A step line that is progress must survive the filter.
        for l in &lines {
            assert_eq!(crate::redact::secret_class(l), None, "a step line the redaction withholds whole: {l}");
        }
    }

    /// 0.5.0: the secret offer has its own path, a VM that has none answers
    /// an empty one, and its `Debug` never shows a value.
    #[test]
    fn the_secret_offer_is_its_own_path_and_never_shows_a_value() {
        struct Front;
        impl Vm for Front {
            fn hello(&self) -> Result<JoinRequest, String> {
                Err("n/a".into())
            }
            fn fed(&self, _r: &JoinReply) -> Result<(), String> {
                Ok(())
            }
            fn status(&self) -> Signal {
                Signal::of("front", TunnelKind::Wireguard, &crate::pair::Machine::default())
            }
            fn close(&self) -> Result<(), String> {
                Ok(())
            }
            fn identity(&self) -> crate::mtls::Issued {
                crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
            }
            fn secret_offer(&self) -> SecretOffer {
                SecretOffer::new(BTreeMap::from([("tls.gunnar.rs.key".to_string(), "-----BEGIN PRIVATE KEY-----\nK\n".to_string())]))
            }
        }
        let log = BootLog::new(4);
        let (r, _) = respond(&Front, &log, &get(SECRET_OFFER_PATH));
        assert_eq!(r.status, 200);
        let s = SecretOffer::parse(&r.body).unwrap();
        assert_eq!(s.expose()["tls.gunnar.rs.key"], "-----BEGIN PRIVATE KEY-----\nK\n");
        assert_eq!(format!("{s:?}"), "SecretOffer([\"tls.gunnar.rs.key\"] — values withheld)");
        assert!(parse_offer(&respond(&Front, &log, &get(OFFER_PATH)).0.body).unwrap().is_empty(), "the public offer does not carry it");
        let box_vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
        assert!(SecretOffer::parse(&respond(&box_vm, &log, &get(SECRET_OFFER_PATH)).0.body).unwrap().is_empty(), "a box hands over nothing");
        assert!(!SecretOffer::parse("= nope").unwrap_err().contains("nope"), "the error never quotes the body");
    }

    /// 0.5.4: after the close, the log and nothing else.
    #[test]
    fn a_closed_listener_answers_the_log_and_refuses_the_install_by_name() {
        let log = BootLog::new(10);
        log.push("a\nb\nc");
        let r = respond_closed(&log, &get(&format!("{LOG_PATH}?from=1")));
        assert_eq!(r.status, 200);
        let chunk = crate::bootlog::Chunk::parse(&r.body).unwrap();
        assert_eq!((chunk.lines, chunk.next), (vec!["b".to_string(), "c".to_string()], 3));
        for p in [HELLO_PATH, STATUS_PATH, OFFER_PATH, SECRET_OFFER_PATH] {
            let r = respond_closed(&log, &get(p));
            assert!(r.status == 410 && r.body.starts_with("REFUSED [install-closed]"), "{p}: {} {}", r.status, r.body);
        }
        for p in [FEED_PATH, CLOSE_PATH] {
            assert_eq!(respond_closed(&log, &Request { method: "POST".into(), path: p.into(), body: Vec::new() }).status, 410, "{p}");
        }
        assert_eq!(respond_closed(&log, &Request { method: "POST".into(), path: LOG_PATH.into(), body: Vec::new() }).status, 405);
        assert_eq!(respond_closed(&log, &get("/pair/join")).status, 404);
    }

    fn reply_as(name: &str, role: Option<crate::Role>) -> JoinReply {
        JoinReply { estate: "e".into(), me: crate::join::Assigned { name: name.into(), overlay: crate::OVERLAY_BOX_A }, cert_pem: None, role, primary_overlay: None, primary_utility: None, peers: vec![], feed: Default::default() }
    }

    /// 0.5.8: a closed VM's log door takes a RE-FEED — only through
    /// [`Vm::refed`], refused by default — and nothing else of the install.
    #[test]
    fn a_closed_vm_takes_a_refeed_through_its_log_door_and_refuses_it_by_default() {
        let log = BootLog::new(10);
        let post = |p: &str, r: &JoinReply| Request { method: "POST".into(), path: p.into(), body: r.render().unwrap().into_bytes() };
        let vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(true) };
        let r = respond_closed_to(&vm, &log, &post(REFEED_PATH, &reply_as("main", Some(crate::Role::Primary))));
        assert!(r.status == 409 && r.body.contains("[refeed-unsupported]"), "{} {}", r.status, r.body);
        assert!(vm.fed.lock().unwrap().is_empty());
        assert_eq!(respond_closed_to(&vm, &log, &get(REFEED_PATH)).status, 405);
        assert_eq!(respond_closed_to(&vm, &log, &Request { method: "POST".into(), path: REFEED_PATH.into(), body: b"not toml [".to_vec() }).status, 400);
        assert_eq!(respond_closed_to(&vm, &log, &post(FEED_PATH, &reply_as("main", None))).status, 410, "the install's feed stays closed");
        assert_eq!(respond_closed_to(&vm, &log, &get(LOG_PATH)).status, 200, "the log door is unchanged");
    }

    #[test]
    fn a_refeed_never_changes_what_the_vm_is() {
        let kept = reply_as("main", Some(crate::Role::Primary));
        assert!(refeed_refusal(None, &kept).unwrap().contains("[refeed-never-fed]"));
        assert_eq!(refeed_refusal(Some(&kept), &JoinReply { primary_utility: Some([10, 0, 0, 9].into()), ..kept.clone() }), None, "addresses around it may change");
        for (bad, says) in [
            (JoinReply { estate: "other".into(), ..kept.clone() }, "estate"),
            (reply_as("twin", Some(crate::Role::Primary)), "names this VM"),
            (JoinReply { me: crate::join::Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_B }, ..kept.clone() }, "overlay"),
            (reply_as("main", Some(crate::Role::Twin)), "bootstrap fact"),
            (JoinReply { cert_pem: Some("C".into()), ..kept.clone() }, "certificate"),
        ] {
            let e = refeed_refusal(Some(&kept), &bad).expect(says);
            assert!(e.starts_with("REFUSED [refeed-changes-the-vm]") && e.contains(says), "{e}");
        }
    }

    #[test]
    fn the_from_query_is_read_and_a_missing_one_is_the_start() {
        assert_eq!(query_from("/install/log?from=42"), 42);
        assert_eq!(query_from("/install/log?x=1&from=7"), 7);
        assert_eq!(query_from("/install/log"), 0);
        assert!(installer_only("bootstrap").starts_with("REFUSED [installer-only]"));
    }

    /// A path that connects, then says nothing for `quiet` before its TLS begins — the
    /// installer's reader over a busy link.
    struct QuietPath {
        at: std::net::SocketAddrV4,
        quiet: Duration,
    }

    impl crate::installer_link::Path for QuietPath {
        fn connect(&mut self, timeout: Duration) -> Result<Box<dyn crate::installer_link::Wire + '_>, String> {
            let s = TcpStream::connect_timeout(&SocketAddr::V4(self.at), timeout).map_err(|e| e.to_string())?;
            let _ = s.set_read_timeout(Some(timeout));
            std::thread::sleep(self.quiet);
            Ok(Box::new(s))
        }
        fn gone(&mut self, _wait: Duration) -> Result<bool, String> {
            Ok(false)
        }
        fn describe(&self) -> String {
            format!("{} (quiet)", self.at)
        }
    }

    struct DoorVm(crate::mtls::Issued);

    impl Vm for DoorVm {
        fn hello(&self) -> Result<JoinRequest, String> {
            Err("closed".into())
        }
        fn fed(&self, _r: &JoinReply) -> Result<(), String> {
            Ok(())
        }
        fn status(&self) -> Signal {
            Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
        }
        fn close(&self) -> Result<(), String> {
            Ok(())
        }
        fn identity(&self) -> crate::mtls::Issued {
            self.0.clone()
        }
    }

    /// **H + I + J (release-3 mocks 2026-09-29): the log door after the close.** Every box's
    /// log carried `WARN installer: a connection to the log door was refused: …: Connection
    /// reset by peer (os error 104)` (and `unexpected end of file`, and `Resource temporarily
    /// unavailable (os error 11)`) again and again for the box's whole life: the installer's
    /// reader going away mid-connection, and a reader slower than ONE read timeout, said as
    /// refusals. A reader that is merely slow is served (EAGAIN is "not yet"); one that goes
    /// away is said ONCE at INFO; the close and the door opening are INFO; a certificate that
    /// is not the installer's is still a WARN refusal.
    #[test]
    fn the_log_door_serves_a_slow_reader_and_says_a_reader_going_away_once_at_info() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::new(10));
        log.push("gunnar: serving");
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let said: Arc<std::sync::Mutex<Vec<(Tone, String)>>> = Arc::default();
        let (s2, log2, ca_pem) = (said.clone(), log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log2, &ca_pem, l, &|t, line| s2.lock().unwrap().push((t, line.to_string()))));

        // Readers that go away: three connections dropped before a request.
        for _ in 0..3 {
            drop(TcpStream::connect(at).unwrap());
        }
        // A reader quieter than one read timeout is SERVED.
        let quiet = QuietPath { at, quiet: REQUEST_READ_TIMEOUT * 2 };
        let chunk = crate::installer::Client::new(Box::new(quiet), &ca.cert_pem, &me, Duration::from_secs(20)).unwrap().log(0).expect("a slow reader is served, not refused");
        assert_eq!(chunk.lines, vec!["gunnar: serving".to_string()]);
        // A stranger's certificate is still a refusal.
        let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().log(1).is_err());
        std::thread::sleep(Duration::from_millis(300));

        let said = said.lock().unwrap().clone();
        let warns: Vec<_> = said.iter().filter(|(t, _)| *t == Tone::Warn).collect();
        assert_eq!(warns.len(), 1, "only the stranger is a WARN refusal: {said:?}");
        assert!(warns[0].1.contains("installer-only"), "{said:?}");
        let gone: Vec<_> = said.iter().filter(|(_, l)| l.contains("went away")).collect();
        assert_eq!(gone.len(), 1, "a reader going away is said once: {said:?}");
        assert_eq!(gone[0].0, Tone::Info, "{said:?}");
        assert!(said.iter().filter(|(_, l)| l.contains("only the log door is open")).all(|(t, _)| *t == Tone::Info), "{said:?}");
    }

    /// **DEV-NARRATE (2026-09-30): the listener says ONCE, at INFO, that the
    /// installer reached it** — the first request its OWN certificate made over
    /// the link. The installer's narration (nordisk-narrate, sealed UDP) reads a
    /// box log without this line as "stopped BEFORE its WireGuard link", and
    /// fails the install naming the box and its last stage. A stranger's
    /// certificate is not the installer and says nothing of the kind.
    #[test]
    fn the_listener_says_once_that_the_installer_reached_it() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::new(10));
        log.push("gunnar: listening");
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let said: Arc<std::sync::Mutex<Vec<(Tone, String)>>> = Arc::default();
        let (s2, log2, ca_pem) = (said.clone(), log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_for_life_at(&DoorVm(boot), &log2, &ca_pem, l, &|t, line| s2.lock().unwrap().push((t, line.to_string()))));
        let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().log(0).is_err());
        std::thread::sleep(Duration::from_millis(200));
        assert!(!said.lock().unwrap().iter().any(|(_, l)| l == REACHED), "a stranger is not the installer: {:?}", said.lock().unwrap());
        for _ in 0..3 {
            crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(5)).unwrap().log(0).unwrap();
        }
        std::thread::sleep(Duration::from_millis(200));
        let said = said.lock().unwrap().clone();
        let reached: Vec<_> = said.iter().filter(|(_, l)| l == REACHED).collect();
        assert_eq!(reached.len(), 1, "said once: {said:?}");
        assert_eq!(reached[0].0, Tone::Info);
    }

    /// ★ Gunnar f21d8d09 (2026-09-28), shared (lane G10): **the install's close seals the
    /// log's head**, in the listener every product serves, so no product has to remember
    /// to (Gunnar sealed in its own `Vm::close` and on a restart's log door; Holger never
    /// did). The close is the end of the install: the lines before it are kept for life,
    /// the runtime after it is the evictable tail.
    #[test]
    fn the_installers_close_seals_the_head_of_the_log() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::with_head(4, 10_000, 50, 10_000));
        log.push("install: laid\npairing: PAIRED");
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let (log2, ca_pem) = (log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_for_life_at(&DoorVm(boot), &log2, &ca_pem, l, &|_, _| {}));
        let client = || crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
        client().close().expect("the installer closes the install");
        for i in 0..20 {
            log.push(&format!("runtime {i}"));
        }
        assert_eq!(log.head_len(), 2, "the close sealed the head: the runtime went to the tail");
        let c = client().log(0).unwrap();
        assert_eq!(&c.lines[..2], &["install: laid", "pairing: PAIRED"]);
    }

    /// ★ Gunnar f21d8d09: **a restart that finds the install closed seals at once** —
    /// what this boot said before its log door opened is its head.
    #[test]
    fn a_restart_after_the_close_seals_the_head_before_the_log_door_opens() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::with_head(4, 10_000, 50, 10_000));
        log.push("korp-start: booted");
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let (log2, ca_pem) = (log.clone(), ca.cert_pem.clone());
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log2, &ca_pem, l, &|_, _| { let _ = tx.send(()); }));
        rx.recv_timeout(Duration::from_secs(10)).expect("the log door opened");
        for i in 0..20 {
            log.push(&format!("runtime {i}"));
        }
        assert_eq!(log.head_len(), 1);
    }

    struct LiveVm(crate::mtls::Issued, std::sync::Mutex<Vec<String>>);

    impl Vm for LiveVm {
        fn hello(&self) -> Result<JoinRequest, String> {
            Err("closed".into())
        }
        fn fed(&self, _r: &JoinReply) -> Result<(), String> {
            Ok(())
        }
        fn status(&self) -> Signal {
            Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
        }
        fn close(&self) -> Result<(), String> {
            Ok(())
        }
        fn identity(&self) -> crate::mtls::Issued {
            self.0.clone()
        }
        fn live_running(&self) -> Result<crate::live::Running, String> {
            Ok(crate::live::Running { base: "b".repeat(64), app: "a".repeat(64), slot: "a".into(), live_update: true })
        }
        fn live_stage(&self, stage: &crate::live::Stage) -> Result<String, String> {
            if stage.base != "b".repeat(64) {
                return Err("REFUSED [live-base-mismatch] other base".into());
            }
            self.1.lock().unwrap().push(stage.pin_hex.clone());
            Ok(format!("staged into slot b: {} bytes", stage.archive.len()))
        }
        fn live_stage_declared(&self, stage: &crate::live::Stage, d: &crate::live::Declared) -> Result<String, String> {
            let said = self.live_stage(stage)?;
            let files = d.release.as_ref().map(|(b, _)| b.len()).unwrap_or(0);
            Ok(format!("{said}; declaration {files} bytes, env {:?}", d.env))
        }
    }

    /// ★ **Live update over the log door** (owner 2026-10-12): the installer's own
    /// certificate reads what a closed VM runs and hands it a new app — a payload larger than a
    /// feed included; the VM's refusal comes back by name; a VM without live update refuses
    /// by name; a stranger's certificate is refused.
    #[test]
    fn the_log_door_takes_a_live_update_from_the_installer_only() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::new(10));
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let vm = Arc::new(LiveVm(boot.clone(), Default::default()));
        let (vm2, log2, ca_pem) = (vm.clone(), log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_after_close_at(&*vm2, &log2, &ca_pem, l, &|_, _| {}));
        let client = || crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(60)).unwrap();
        let r = client().live_running().unwrap();
        assert_eq!((r.slot.as_str(), r.live_update), ("a", true));
        let big = crate::live::Stage { pin_hex: "c".repeat(64), signature: vec![9; 72], base: "b".repeat(64), archive: vec![7u8; MAX_FEED_BYTES + 1024] };
        let said = client().live_stage(&big).expect("a payload larger than a feed is taken");
        assert!(said.contains("staged into slot b"), "{said}");
        assert_eq!(vm.1.lock().unwrap().as_slice(), &["c".repeat(64)]);
        let other = crate::live::Stage { base: "f".repeat(64), archive: vec![1; 10], ..big.clone() };
        assert!(client().live_stage(&other).unwrap_err().contains("[live-base-mismatch]"));
        let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().live_running().is_err());
        // A VM without live update refuses by name.
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at2) = l.local_addr().unwrap() else { panic!() };
        let (log3, ca_pem) = (log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log3, &ca_pem, l, &|_, _| {}));
        let mut plain = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at: at2 }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
        assert!(plain.live_running().unwrap_err().contains("[live-unsupported]"));
    }

    /// ★ **The declaration reaches the VM over the door** (lane B5): a stage with a release
    /// declaration and env arrives whole at the VM's `live_stage_declared`; a VM that takes
    /// only the plain stage refuses a declared one by name (`[live-declared-unsupported]`,
    /// which a front answers with a keep boot) and still takes an undeclared one.
    #[test]
    fn the_log_door_hands_the_declaration_to_the_vm() {
        let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
        let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
        let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
        let log = Arc::new(BootLog::new(10));
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
        let vm = Arc::new(LiveVm(boot.clone(), Default::default()));
        let (vm2, log2, ca_pem) = (vm.clone(), log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_after_close_at(&*vm2, &log2, &ca_pem, l, &|_, _| {}));
        let mut client = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(60)).unwrap();
        let st = crate::live::Stage { pin_hex: "c".repeat(64), signature: vec![9; 72], base: "b".repeat(64), archive: vec![7u8; 1000] };
        let d = crate::live::Declared { release: Some((vec![1; 33], vec![2; 72])), env: Some(vec!["korp.env.A=1".into()]) };
        let said = client.live_stage_declared(&st, &d).unwrap();
        assert!(said.contains("declaration 33 bytes") && said.contains("korp.env.A=1"), "{said}");
        // A VM that takes only the plain stage.
        struct PlainVm(LiveVm);
        impl Vm for PlainVm {
            fn hello(&self) -> Result<JoinRequest, String> { self.0.hello() }
            fn fed(&self, r: &JoinReply) -> Result<(), String> { self.0.fed(r) }
            fn status(&self) -> Signal { self.0.status() }
            fn close(&self) -> Result<(), String> { self.0.close() }
            fn identity(&self) -> crate::mtls::Issued { self.0.identity() }
            fn live_stage(&self, stage: &crate::live::Stage) -> Result<String, String> { self.0.live_stage(stage) }
        }
        let l = TcpListener::bind("127.0.0.1:0").unwrap();
        let std::net::SocketAddr::V4(at2) = l.local_addr().unwrap() else { panic!() };
        let (log3, ca_pem) = (log.clone(), ca.cert_pem.clone());
        std::thread::spawn(move || serve_after_close_at(&PlainVm(LiveVm(boot, Default::default())), &log3, &ca_pem, l, &|_, _| {}));
        let mut plain = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at: at2 }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
        let e = plain.live_stage_declared(&st, &d).unwrap_err();
        assert!(e.contains("[live-declared-unsupported]") && crate::live::falls_to_keep(&e), "{e}");
        assert!(plain.live_stage_declared(&st, &crate::live::Declared::default()).unwrap().contains("staged into slot b"));
    }
}