gunnar-grpc 2.0.0

//! **`Admin.Topology`: the estate's boxes, as THIS gunnar knows them**
//! (owner, 2026-10-04).
//!
//! The console's Machine card draws a picture of the estate, and the owner's
//! rule for the console is that it is a thin layer: it calls and it renders,
//! and it decides nothing. So every decision behind the picture is here — which
//! boxes there are, what each one runs, and whether it is alive — and the
//! console draws the answer as it comes.
//!
//! # Evidence, never inference
//!
//! A box is on the list because this gunnar holds EVIDENCE of it, and each
//! row says which (`sources`) and in one sentence how (`why`):
//!
//! * **`self`** — this box. It answered the call, so it is `up`.
//! * **`pairing`** — the installer's pairing feed (`joined.toml`): the front,
//!   the primary and the twins it named. A peer nothing here can ask is
//!   `unknown`, which is a different fact from `silent`.
//! * **`heartbeat`** — a twin's `Mirror.LogHead` polls (the primary half of
//!   the dual heartbeat), or, on a twin, when its primary last answered. Its
//!   state word (`ok`/`behind`/`silent`) is authoritative for that box.
//! * **`replication`** — a configured replication standby and its verdict.
//! * **`caller`** — a MACHINE principal that called this gunnar: a principal
//!   the pairing feed says runs on the front (the console, monetize, the
//!   guest, the certify key), or any machine that asked for this very picture
//!   (a console drawing it is a console).
//! * **`record`** — the ESTATE RECORD the installer fed (owner, 2026-10-04:
//!   "the installer has the full picture — at the end of the installation
//!   gunnar-server should have all the info"). The kit lays the boxes, so it
//!   feeds every box the estate as laid — each box's name, how it was laid
//!   (front/main/twin), its public and utility addresses, its hostname and
//!   title, its declared size, system disk and data disks — plus the variant
//!   and the commits, as the shared feed's `estate.*` keys ([`Record`], one
//!   encoding for both sides: [`Record::to_feed`] / [`Record::from_feed`]).
//!   The pairing persists the feed (`joined.toml`) and every re-admission
//!   renews it (upgrade, recharge, a twin laid again). With a record the
//!   answer is EXACTLY the laid boxes, every other piece of evidence merged
//!   onto them; without one (a solo gunnar, a box installed before the
//!   record) it is what the evidence alone shows.
//!
//! A solo gunnar is one row; a front and a server two; with a twin three, and
//! one more per twin. Nothing here walks a network, resolves a name or dials a
//! box: every input is something the server already recorded.

use std::collections::BTreeMap;
use std::sync::Mutex;

use gunnar_proto::{PrimaryHeartbeat, StandbyStatus, TopologyBox, TopologyResponse, TwinHeartbeat};

pub const ROLE_PRIMARY: &str = "primary";
pub const ROLE_TWIN: &str = "twin";
pub const ROLE_FRONT: &str = "front";
pub const ROLE_CONSOLE: &str = "console";
pub const ROLE_MONETIZE: &str = "monetize";
pub const ROLE_STANDBY: &str = "standby";

pub const STATE_UP: &str = "up";
pub const STATE_BEHIND: &str = "behind";
pub const STATE_SILENT: &str = "silent";
pub const STATE_UNKNOWN: &str = "unknown";

pub const SERVICE_GUNNAR: &str = "gunnar";
pub const SERVICE_CONSOLE: &str = "console";
pub const SERVICE_MONETIZE: &str = "monetize";

pub const SOURCE_SELF: &str = "self";
pub const SOURCE_PAIRING: &str = "pairing";
pub const SOURCE_HEARTBEAT: &str = "heartbeat";
pub const SOURCE_REPLICATION: &str = "replication";
pub const SOURCE_CALLER: &str = "caller";
pub const SOURCE_RECORD: &str = "record";

// ── the estate record's feed keys (the kit writes them, gunnar reads them) ──

/// The record's schema; a box fed a schema this build does not know keeps no
/// record (and says so in its answer: `recorded` false).
pub const FEED_RECORD: &str = "estate.record";
pub const RECORD_SCHEMA: &str = "1";
pub const FEED_VARIANT: &str = "estate.variant";
/// `<repository>=<commit>` pairs, comma-separated.
pub const FEED_VERSIONS: &str = "estate.versions";
/// The laid boxes' names, comma-separated, in lay order.
pub const FEED_BOXES: &str = "estate.boxes";
/// `estate.box.<name>.<field>`.
pub const FEED_BOX_PREFIX: &str = "estate.box.";

/// **One box as the installer laid it** — declared, never measured.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RecordBox {
    /// The pairing name (`front`, `main`, `twin`).
    pub name: String,
    /// How the kit laid it: `front` | `main` | `twin`.
    pub laid_as: String,
    pub hostname: String,
    pub title: String,
    pub public: String,
    pub utility: String,
    /// The provider's size name (`STARTER-4xCPU-16GB`).
    pub size: String,
    pub vcpu: u32,
    pub ram_mb: u64,
    pub system_disk_gb: u64,
    /// The data disks in words: `4 × 10 GB maxiops, md-raid0`.
    pub data: String,
}

/// **The estate as the installer laid it.**
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Record {
    pub variant: String,
    /// `(repository, commit)`, in the order fed.
    pub versions: Vec<(String, String)>,
    pub boxes: Vec<RecordBox>,
}

const BOX_FIELDS: [&str; 10] = ["laid_as", "hostname", "title", "public", "utility", "size", "vcpu", "ram_mb", "system_disk_gb", "data"];

impl Record {
    /// The record as feed keys — what the kit puts in the shared feed.
    pub fn to_feed(&self) -> BTreeMap<String, String> {
        let mut f = BTreeMap::new();
        f.insert(FEED_RECORD.to_owned(), RECORD_SCHEMA.to_owned());
        f.insert(FEED_VARIANT.to_owned(), self.variant.clone());
        f.insert(FEED_VERSIONS.to_owned(), self.versions.iter().map(|(r, c)| format!("{r}={c}")).collect::<Vec<_>>().join(","));
        f.insert(FEED_BOXES.to_owned(), self.boxes.iter().map(|b| b.name.clone()).collect::<Vec<_>>().join(","));
        for b in &self.boxes {
            let values = [
                b.laid_as.clone(),
                b.hostname.clone(),
                b.title.clone(),
                b.public.clone(),
                b.utility.clone(),
                b.size.clone(),
                b.vcpu.to_string(),
                b.ram_mb.to_string(),
                b.system_disk_gb.to_string(),
                b.data.clone(),
            ];
            for (field, value) in BOX_FIELDS.iter().zip(values) {
                f.insert(format!("{FEED_BOX_PREFIX}{}.{field}", b.name), value);
            }
        }
        f
    }

    /// **The record a feed carries**, or `None` — no `estate.record` key, or a
    /// schema this build does not read. A box name the feed lists is a box
    /// even when its fields are missing (they read as not declared).
    pub fn from_feed(feed: &BTreeMap<String, String>) -> Option<Record> {
        if feed.get(FEED_RECORD).map(String::as_str) != Some(RECORD_SCHEMA) {
            return None;
        }
        let get = |k: &str| feed.get(k).cloned().unwrap_or_default();
        let versions = get(FEED_VERSIONS)
            .split(',')
            .filter_map(|p| p.split_once('=').map(|(r, c)| (r.trim().to_owned(), c.trim().to_owned())))
            .filter(|(r, c)| !r.is_empty() && !c.is_empty())
            .collect();
        let boxes = get(FEED_BOXES)
            .split(',')
            .map(str::trim)
            .filter(|n| !n.is_empty())
            .map(|name| {
                let f = |field: &str| get(&format!("{FEED_BOX_PREFIX}{name}.{field}"));
                RecordBox {
                    name: name.to_owned(),
                    laid_as: f("laid_as"),
                    hostname: f("hostname"),
                    title: f("title"),
                    public: f("public"),
                    utility: f("utility"),
                    size: f("size"),
                    vcpu: f("vcpu").parse().unwrap_or(0),
                    ram_mb: f("ram_mb").parse().unwrap_or(0),
                    system_disk_gb: f("system_disk_gb").parse().unwrap_or(0),
                    data: f("data"),
                }
            })
            .collect();
        Some(Record { variant: get(FEED_VARIANT), versions, boxes })
    }
}

/// **How long a caller's last call keeps its box `up`.** Longer than the
/// twin's poll bound on purpose: a twin polls every few seconds, a console
/// calls when a person does something, and ten quiet minutes on a console is
/// an idle evening, not an outage. The row's sentence always says how long ago.
pub const CALLER_SILENT_AFTER_MS: u64 = 10 * 60_000;

/// How many distinct machine principals the caller log keeps. Bounded, like
/// every roster on this listener: past it the oldest sighting goes.
pub const CALLERS_CAPACITY: usize = 64;

/// One machine principal's newest call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Sighting {
    pub principal: String,
    /// `ip:port` of the connection, empty when the transport gave none.
    pub address: String,
    pub last_unix_ms: i64,
    /// It has asked for `Admin.Topology` — a console drawing the picture.
    pub asked_topology: bool,
}

/// **The machine principals that call this listener**, newest call each.
/// Written by [`crate::ServiceContext::caller`] for every verified machine
/// caller, read by `Admin.Topology`. People are not recorded: a person's
/// laptop is not a box of the estate.
#[derive(Debug, Default)]
pub struct CallerLog {
    seen: Mutex<BTreeMap<String, Sighting>>,
}

impl CallerLog {
    /// Record one call. `topology` marks the caller as one that asked for the
    /// picture (it stays marked).
    pub fn note(&self, principal: &str, address: &str, now_ms: i64, topology: bool) {
        let Ok(mut seen) = self.seen.lock() else { return };
        if !seen.contains_key(principal) && seen.len() >= CALLERS_CAPACITY {
            if let Some(oldest) = seen.values().min_by_key(|s| s.last_unix_ms).map(|s| s.principal.clone()) {
                seen.remove(&oldest);
            }
        }
        let row = seen.entry(principal.to_owned()).or_insert_with(|| Sighting {
            principal: principal.to_owned(),
            address: String::new(),
            last_unix_ms: 0,
            asked_topology: false,
        });
        if now_ms >= row.last_unix_ms {
            row.last_unix_ms = now_ms;
            if !address.is_empty() {
                row.address = address.to_owned();
            }
        }
        row.asked_topology |= topology;
    }

    /// Every sighting, in principal order.
    pub fn sightings(&self) -> Vec<Sighting> {
        self.seen.lock().map(|s| s.values().cloned().collect()).unwrap_or_default()
    }
}

/// A peer the installer's pairing feed names.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Peer {
    pub name: String,
    /// `front` | `primary` | `twin`.
    pub role: String,
    /// Every address the feed gives it (utility first, then the tunnel), bare
    /// hosts — what a caller's or a heartbeat's address is matched against.
    pub hosts: Vec<String>,
}

/// **What the control plane knows and the service does not**: the pairing
/// feed, which principals run on the front, the plugin registration, this
/// box's own data volume. [`crate::seam::ControlPlane::topology_facts`].
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Facts {
    /// The estate's name from the pairing feed; empty when unpaired.
    pub estate: String,
    /// This box's own pairing name; empty when unpaired.
    pub me: String,
    /// Where this box is: its utility (or tunnel) address, or its bind.
    pub address: String,
    pub peers: Vec<Peer>,
    /// Principal id → the service it is on the front (`console`,
    /// `monetize`, …): the principals the pairing feed's front offered.
    pub front_principals: BTreeMap<String, String>,
    /// The estate record the installer fed, when there is one.
    pub record: Option<Record>,
    /// This box's data volume `(total, used)`, when measured.
    pub disk: Option<(u64, u64)>,
}

/// Everything [`assemble`] reads.
#[derive(Debug, Clone, Default)]
pub struct Input {
    pub facts: Facts,
    pub is_twin: bool,
    pub twins: Vec<TwinHeartbeat>,
    pub primary: Option<PrimaryHeartbeat>,
    pub standbys: Vec<StandbyStatus>,
    pub callers: Vec<Sighting>,
    pub now_ms: i64,
    /// The twin heartbeat's silence bound, reported on the answer.
    pub silent_after_ms: u64,
}

/// `10.0.0.5:4242` → `10.0.0.5`; `[::1]:50051` → `::1`;
/// `https://host:8443/x` → `host`; a bare host stays itself.
pub fn host_of(address: &str) -> String {
    let a = address.trim();
    let a = a.split_once("://").map(|(_, rest)| rest).unwrap_or(a);
    let a = a.split('/').next().unwrap_or(a);
    let a = a.rsplit_once('@').map(|(_, h)| h).unwrap_or(a);
    if let Some(rest) = a.strip_prefix('[') {
        return rest.split(']').next().unwrap_or(rest).to_owned();
    }
    match a.rsplit_once(':') {
        // One colon is host:port; more is a bare IPv6 address.
        Some((host, port)) if !host.contains(':') && port.chars().all(|c| c.is_ascii_digit()) => host.to_owned(),
        _ => a.to_owned(),
    }
}

/// One piece of evidence about a box.
#[derive(Debug, Clone)]
struct Evidence {
    state: &'static str,
    at_ms: i64,
    why: String,
    /// A heartbeat's word decides the box's state over everything else.
    authoritative: bool,
    /// The state in a few words, as this evidence says it ([`TopologyBox::headline`]).
    headline: String,
}

#[derive(Debug, Clone)]
struct Draft {
    row: TopologyBox,
    hosts: Vec<String>,
    evidence: Vec<Evidence>,
}

impl Draft {
    fn new(name: &str, role: &str, address: &str, source: &str) -> Draft {
        let host = host_of(address);
        Draft {
            row: TopologyBox {
                name: name.to_owned(),
                role: role.to_owned(),
                address: address.to_owned(),
                sources: vec![source.to_owned()],
                ..Default::default()
            },
            hosts: if host.is_empty() { Vec::new() } else { vec![host] },
            evidence: Vec::new(),
        }
    }

    fn has_host(&self, host: &str) -> bool {
        !host.is_empty() && self.hosts.iter().any(|h| h == host)
    }

    fn source(&mut self, source: &str) {
        if !self.row.sources.iter().any(|s| s == source) {
            self.row.sources.push(source.to_owned());
        }
    }

    fn service(&mut self, service: &str) {
        if !service.is_empty() && !self.row.services.iter().any(|s| s == service) {
            self.row.services.push(service.to_owned());
        }
    }

    /// What the installer laid, onto this row.
    fn record(&mut self, b: &RecordBox) {
        self.source(SOURCE_RECORD);
        let r = &mut self.row;
        r.laid_as = b.laid_as.clone();
        r.public_address = b.public.clone();
        r.utility_address = b.utility.clone();
        r.hostname = b.hostname.clone();
        r.title = b.title.clone();
        r.provider_size = b.size.clone();
        r.declared_vcpu = b.vcpu;
        r.declared_ram_mb = b.ram_mb;
        r.declared_system_disk_gb = b.system_disk_gb;
        r.declared_data = b.data.clone();
        for a in [&b.utility, &b.public] {
            if !a.is_empty() {
                self.at(a);
            }
        }
    }

    fn at(&mut self, address: &str) {
        let host = host_of(address);
        if self.row.address.is_empty() {
            self.row.address = address.to_owned();
        }
        if !host.is_empty() && !self.has_host(&host) {
            self.hosts.push(host);
        }
    }
}

/// **How long, coarsely** — whole minutes once past one, whole hours past
/// two — so a headline changes with the state and not with the clock.
fn coarse(ms: u64) -> String {
    let s = ms / 1000;
    if s < 60 {
        "under a minute".to_owned()
    } else if s < 7200 {
        format!("{} min", s / 60)
    } else {
        format!("{} h", s / 3600)
    }
}

/// A heartbeat's headline: the state word, the lag, and the twin's OWN word
/// when it is not plain following.
fn heartbeat_headline(state: &str, last_seen_ms: i64, ago_ms: u64, lag: u64, own: &str, ok_word: &str) -> String {
    let head = match state {
        STATE_UP => ok_word.to_owned(),
        STATE_BEHIND => format!("behind {lag} entries"),
        _ if last_seen_ms == 0 => "never heard".to_owned(),
        _ => format!("silent for {}", coarse(ago_ms)),
    };
    let own = own.trim();
    if state != STATE_SILENT && !own.is_empty() && own != "following" && own != "idle" {
        format!("{head} · {own}")
    } else {
        head
    }
}

fn ago_words(ms: u64) -> String {
    let s = ms / 1000;
    if s < 120 {
        format!("{s} s ago")
    } else if s < 7200 {
        format!("{} min ago", s / 60)
    } else {
        format!("{} h ago", s / 3600)
    }
}

/// A standby's verdict word as a box state.
fn standby_state(verdict: &str, never: bool) -> &'static str {
    match verdict {
        "NEVER" | "STALE" => STATE_SILENT,
        "BEHIND" => STATE_BEHIND,
        "syncing" | "idle" => STATE_UP,
        _ if never => STATE_SILENT,
        _ => STATE_UNKNOWN,
    }
}

fn role_rank(role: &str) -> u8 {
    match role {
        ROLE_PRIMARY => 0,
        ROLE_TWIN => 1,
        ROLE_STANDBY => 2,
        ROLE_FRONT => 3,
        ROLE_CONSOLE => 4,
        ROLE_MONETIZE => 5,
        _ => 6,
    }
}

/// **The answer.** Pure: the same input always gives the same rows.
pub fn assemble(input: &Input) -> TopologyResponse {
    let facts = &input.facts;
    let now = input.now_ms;
    let mut boxes: Vec<Draft> = Vec::new();

    // ── 1. this box ────────────────────────────────────────────────────────
    let mut me = Draft::new(
        if facts.me.is_empty() { "this box" } else { &facts.me },
        if input.is_twin { ROLE_TWIN } else { ROLE_PRIMARY },
        &facts.address,
        SOURCE_SELF,
    );
    me.row.this_box = true;
    me.service(SERVICE_GUNNAR);
    if let Some((total, used)) = facts.disk {
        me.row.disk_measured = true;
        me.row.disk_total_bytes = total;
        me.row.disk_used_bytes = used;
    }
    me.evidence.push(Evidence { state: STATE_UP, at_ms: now, why: "this box answered this call".into(), authoritative: true, headline: "answering".into() });
    boxes.push(me);

    // ── 1b. the estate record: every box the installer laid ────────────────
    if let Some(rec) = &facts.record {
        for b in &rec.boxes {
            let i = if !facts.me.is_empty() && b.name == facts.me {
                0
            } else {
                // How it was laid; the pairing's election (step 2) has the
                // last word on primary and twin.
                let role = match b.laid_as.as_str() {
                    "front" => ROLE_FRONT,
                    "twin" => ROLE_TWIN,
                    _ => ROLE_PRIMARY,
                };
                boxes.push(Draft::new(&b.name, role, "", SOURCE_RECORD));
                boxes.len() - 1
            };
            let d = &mut boxes[i];
            d.record(b);
            if b.laid_as != ROLE_FRONT {
                d.service(SERVICE_GUNNAR);
            }
        }
    }

    // ── 2. the pairing feed's peers ────────────────────────────────────────
    for p in &facts.peers {
        if !facts.me.is_empty() && p.name == facts.me {
            continue;
        }
        // A box the record named is the same box: the pairing's elected role
        // and its addresses merge onto it.
        let i = match boxes.iter().position(|b| !b.row.this_box && b.row.name == p.name) {
            Some(i) => {
                boxes[i].row.role = p.role.clone();
                boxes[i].source(SOURCE_PAIRING);
                i
            }
            None => {
                boxes.push(Draft::new(&p.name, &p.role, p.hosts.first().map(String::as_str).unwrap_or(""), SOURCE_PAIRING));
                boxes.len() - 1
            }
        };
        let d = &mut boxes[i];
        for h in &p.hosts {
            d.at(h);
        }
        if p.role == ROLE_PRIMARY || p.role == ROLE_TWIN {
            d.service(SERVICE_GUNNAR);
        }
    }

    // ── 3. the twins this primary hears ────────────────────────────────────
    for t in &input.twins {
        let host = host_of(&t.address);
        let found = boxes.iter().position(|b| !b.row.this_box && b.row.role == ROLE_TWIN && b.has_host(&host)).or_else(|| {
            // A twin never heard from has no address. If the feed names
            // exactly one twin nothing else has claimed, it is that one.
            if !host.is_empty() {
                return None;
            }
            let open: Vec<usize> = boxes
                .iter()
                .enumerate()
                .filter(|(_, b)| !b.row.this_box && b.row.role == ROLE_TWIN && !b.row.sources.iter().any(|s| s == SOURCE_HEARTBEAT))
                .map(|(i, _)| i)
                .collect();
            (open.len() == 1).then(|| open[0])
        });
        let i = match found {
            Some(i) => i,
            None => {
                boxes.push(Draft::new(&t.principal, ROLE_TWIN, &t.address, SOURCE_HEARTBEAT));
                boxes.len() - 1
            }
        };
        let d = &mut boxes[i];
        d.source(SOURCE_HEARTBEAT);
        d.service(SERVICE_GUNNAR);
        d.at(&t.address);
        if t.disk_total_bytes > 0 {
            d.row.disk_measured = true;
            d.row.disk_total_bytes = t.disk_total_bytes;
            d.row.disk_used_bytes = t.disk_used_bytes;
        }
        let state = match t.state.as_str() {
            "ok" => STATE_UP,
            "behind" => STATE_BEHIND,
            _ => STATE_SILENT,
        };
        let why = if t.last_seen_unix_ms == 0 {
            format!("twin `{}` has not polled this primary since it started: {}", t.principal, t.why)
        } else {
            format!("twin `{}` polled {} (lag {} entries): {}", t.principal, ago_words(t.seen_ago_ms), t.lag_entries, t.why)
        };
        let headline = heartbeat_headline(state, t.last_seen_unix_ms, t.seen_ago_ms, t.lag_entries, &t.twin_state, "following");
        d.evidence.push(Evidence { state, at_ms: t.last_seen_unix_ms, why, authoritative: true, headline });
    }

    // ── 4. on a twin: its primary ──────────────────────────────────────────
    if let Some(p) = &input.primary {
        let host = host_of(&p.address);
        let i = match boxes.iter().position(|b| !b.row.this_box && b.row.role == ROLE_PRIMARY && (b.has_host(&host) || host.is_empty())) {
            Some(i) => i,
            None => {
                boxes.push(Draft::new(if p.address.is_empty() { "primary" } else { &host }, ROLE_PRIMARY, &p.address, SOURCE_HEARTBEAT));
                boxes.len() - 1
            }
        };
        let d = &mut boxes[i];
        d.source(SOURCE_HEARTBEAT);
        d.service(SERVICE_GUNNAR);
        d.at(&p.address);
        let state = if p.state == "ok" { STATE_UP } else { STATE_SILENT };
        let why = if p.last_answer_unix_ms == 0 {
            format!("the primary has not answered this twin since it started: {}", p.why)
        } else {
            format!("the primary answered {}: {}", ago_words(p.answer_ago_ms), p.why)
        };
        let headline = heartbeat_headline(state, p.last_answer_unix_ms, p.answer_ago_ms, 0, "", "answering");
        d.evidence.push(Evidence { state, at_ms: p.last_answer_unix_ms, why, authoritative: true, headline });
    }

    // ── 5. replication standbys ────────────────────────────────────────────
    for s in &input.standbys {
        let host = host_of(&s.remote);
        let i = match boxes.iter().position(|b| !b.row.this_box && b.has_host(&host)) {
            Some(i) => i,
            None => {
                boxes.push(Draft::new(&s.name, ROLE_STANDBY, &s.remote, SOURCE_REPLICATION));
                boxes.len() - 1
            }
        };
        let d = &mut boxes[i];
        d.source(SOURCE_REPLICATION);
        d.service(SERVICE_GUNNAR);
        let state = standby_state(&s.verdict, s.last_success_unix_ms == 0);
        let why = format!(
            "replication standby `{}`: {}{}",
            s.name,
            if s.verdict.is_empty() { "no verdict" } else { &s.verdict },
            if s.last_error.is_empty() { String::new() } else { format!(" — {}", s.last_error) }
        );
        let headline = match s.verdict.as_str() {
            "NEVER" => "never synced".to_owned(),
            "" => "no verdict".to_owned(),
            v => v.to_lowercase(),
        };
        d.evidence.push(Evidence { state, at_ms: s.last_success_unix_ms, why, authoritative: false, headline });
    }

    // ── 6. the machines that call this gunnar ──────────────────────────────
    let front = boxes.iter().position(|b| !b.row.this_box && b.row.role == ROLE_FRONT);
    for c in &input.callers {
        let service = match facts.front_principals.get(&c.principal) {
            Some(service) => service.clone(),
            None if c.asked_topology => SERVICE_CONSOLE.to_owned(),
            None => continue,
        };
        let host = host_of(&c.address);
        let on_front = facts.front_principals.contains_key(&c.principal);
        let i = match (on_front, front) {
            (true, Some(i)) => i,
            _ => match boxes.iter().position(|b| !b.row.this_box && matches!(b.row.role.as_str(), ROLE_FRONT | ROLE_CONSOLE | ROLE_MONETIZE) && b.has_host(&host)) {
                Some(i) => i,
                None => {
                    let role = if service == SERVICE_MONETIZE { ROLE_MONETIZE } else { ROLE_CONSOLE };
                    boxes.push(Draft::new(&c.principal, role, &c.address, SOURCE_CALLER));
                    boxes.len() - 1
                }
            },
        };
        let d = &mut boxes[i];
        d.source(SOURCE_CALLER);
        if service != ROLE_FRONT {
            d.service(&service);
        }
        d.at(&c.address);
        let ago = now.saturating_sub(c.last_unix_ms).max(0) as u64;
        let state = if ago <= CALLER_SILENT_AFTER_MS { STATE_UP } else { STATE_SILENT };
        let from = if c.address.is_empty() { String::new() } else { format!(" from {}", c.address) };
        d.evidence.push(Evidence {
            state,
            at_ms: c.last_unix_ms,
            why: format!("`{}` called this gunnar {}{from}", c.principal, ago_words(ago)),
            authoritative: false,
            headline: if state == STATE_UP { "calling".to_owned() } else { format!("silent for {}", coarse(ago)) },
        });
    }

    // ── the verdicts ───────────────────────────────────────────────────────
    let mut rows: Vec<TopologyBox> = boxes
        .into_iter()
        .map(|mut d| {
            let authoritative: Vec<&Evidence> = d.evidence.iter().filter(|e| e.authoritative).collect();
            let state = if let Some(e) = authoritative.first() {
                e.state
            } else if d.evidence.iter().any(|e| e.state == STATE_UP) {
                STATE_UP
            } else if d.evidence.iter().any(|e| e.state == STATE_BEHIND) {
                STATE_BEHIND
            } else if d.evidence.iter().any(|e| e.state == STATE_SILENT) {
                STATE_SILENT
            } else {
                STATE_UNKNOWN
            };
            d.row.state = state.to_owned();
            // The words of the evidence that decided the state (the first
            // authoritative one, else the first saying that state).
            d.row.headline = authoritative
                .first()
                .map(|e| e.headline.clone())
                .or_else(|| d.evidence.iter().find(|e| e.state == state).map(|e| e.headline.clone()))
                .unwrap_or_else(|| "not asked".to_owned());
            let last = d.evidence.iter().map(|e| e.at_ms).max().unwrap_or(0);
            d.row.last_seen_unix_ms = last;
            d.row.seen_ago_ms = if last > 0 { now.saturating_sub(last).max(0) as u64 } else { 0 };
            d.row.why = if d.evidence.is_empty() {
                "laid by the installer (its record and pairing feed); nothing on this box can ask it whether it is up".to_owned()
            } else {
                d.evidence.iter().map(|e| e.why.as_str()).collect::<Vec<_>>().join("; ")
            };
            d.row
        })
        .collect();
    let this = rows.remove(0);
    rows.sort_by(|a, b| (role_rank(&a.role), &a.name).cmp(&(role_rank(&b.role), &b.name)));
    rows.insert(0, this);

    let (recorded, variant, versions) = match &facts.record {
        Some(r) => (true, r.variant.clone(), r.versions.iter().map(|(repo, commit)| format!("{repo} {commit}")).collect()),
        None => (false, String::new(), Vec::new()),
    };
    TopologyResponse { boxes: rows, server_unix_ms: now, silent_after_ms: input.silent_after_ms, estate: facts.estate.clone(), recorded, variant, versions }
}

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

    const NOW: i64 = 1_800_000_000_000;

    fn twin(principal: &str, address: &str, state: &str, seen_ago_ms: u64) -> TwinHeartbeat {
        TwinHeartbeat {
            principal: principal.into(),
            address: address.into(),
            last_seen_unix_ms: if address.is_empty() { 0 } else { NOW - seen_ago_ms as i64 },
            seen_ago_ms,
            state: state.into(),
            why: format!("{state} (test)"),
            disk_total_bytes: 20 << 30,
            disk_used_bytes: 1 << 30,
            ..Default::default()
        }
    }

    fn caller(principal: &str, address: &str, ago_ms: i64, topology: bool) -> Sighting {
        Sighting { principal: principal.into(), address: address.into(), last_unix_ms: NOW - ago_ms, asked_topology: topology }
    }

    fn input() -> Input {
        Input { now_ms: NOW, silent_after_ms: 60_000, ..Default::default() }
    }

    fn shape(r: &TopologyResponse) -> Vec<(String, String, String)> {
        r.boxes.iter().map(|b| (b.name.clone(), b.role.clone(), b.state.clone())).collect()
    }

    /// ★ **A solo gunnar is ONE box** — itself, up, the primary, first.
    #[test]
    fn a_solo_gunnar_is_one_box_and_it_is_up() {
        let mut i = input();
        i.facts.address = "127.0.0.1:50051".into();
        i.facts.disk = Some((100, 40));
        let r = assemble(&i);
        assert_eq!(shape(&r), [("this box".into(), "primary".into(), "up".into())]);
        let b = &r.boxes[0];
        assert!(b.this_box && b.disk_measured && b.disk_total_bytes == 100 && b.disk_used_bytes == 40);
        assert_eq!(b.services, ["gunnar"]);
        assert_eq!(b.sources, ["self"]);
        assert_eq!(r.server_unix_ms, NOW);
    }

    /// ★ **A console that asks for the picture is a box of its own** — a front
    /// and a server are TWO boxes. A machine that never asked (a builder's
    /// key, a twin's principal) is not drawn as a console, and a person is
    /// never in the caller log at all.
    #[test]
    fn a_console_that_asks_is_the_second_box_and_other_machines_are_not() {
        let mut i = input();
        i.callers = vec![caller("uitest-console", "127.0.0.1:41000", 0, true), caller("dwarves-oden", "10.0.0.9:5000", 1000, false)];
        let r = assemble(&i);
        assert_eq!(
            shape(&r),
            [("this box".into(), "primary".into(), "up".into()), ("uitest-console".into(), "console".into(), "up".into())]
        );
        assert_eq!(r.boxes[1].services, ["console"]);
        assert_eq!(r.boxes[1].address, "127.0.0.1:41000");
        assert!(r.boxes[1].why.contains("uitest-console") && r.boxes[1].why.contains("127.0.0.1:41000"), "{}", r.boxes[1].why);
    }

    /// ★ **With a twin it is three boxes**, and one more per twin. The twin's
    /// own heartbeat word decides its state: `behind` stays `behind`, a twin
    /// never heard from is `silent`, never `up`.
    #[test]
    fn each_twin_is_one_more_box_with_its_heartbeat_state() {
        let mut i = input();
        i.callers = vec![caller("gunnar-ui", "10.13.8.10:40000", 5_000, true)];
        i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
        assert_eq!(assemble(&i).boxes.len(), 3, "front + server + twin");
        i.twins.push(twin("gunnar-twin-2", "10.13.8.123:5222", "behind", 2_000));
        i.twins.push(twin("gunnar-twin-3", "", "silent", 0));
        let r = assemble(&i);
        assert_eq!(r.boxes.len(), 5);
        let state = |name: &str| r.boxes.iter().find(|b| b.name == name).map(|b| b.state.clone()).unwrap();
        assert_eq!(state("gunnar-twin"), "up");
        assert_eq!(state("gunnar-twin-2"), "behind");
        assert_eq!(state("gunnar-twin-3"), "silent");
        let t = r.boxes.iter().find(|b| b.name == "gunnar-twin").unwrap();
        assert!(t.disk_measured && t.disk_total_bytes == 20 << 30, "the twin's own statvfs travels");
        assert_eq!(r.boxes.iter().map(|b| b.role.as_str()).collect::<Vec<_>>(), ["primary", "twin", "twin", "twin", "console"], "this box first, then by role");
    }

    /// ★ **The estate as the installer paired it**: the front's principals
    /// (console, monetize) and the plugin registration land ON the front
    /// peer — one front box, not three — and a twin's heartbeat lands on the
    /// twin peer the feed named, matched by address.
    #[test]
    fn the_paired_estate_is_front_main_twin_and_the_fronts_services_merge() {
        let mut i = input();
        i.facts.estate = "test2.gunnar.rs".into();
        i.facts.me = "vm2".into();
        i.facts.address = "10.13.8.121".into();
        i.facts.peers = vec![
            Peer { name: "front".into(), role: "front".into(), hosts: vec!["10.99.0.1".into()] },
            Peer { name: "vm2".into(), role: "primary".into(), hosts: vec!["10.13.8.121".into()] },
            Peer { name: "vm3".into(), role: "twin".into(), hosts: vec!["10.13.8.122".into(), "10.99.0.3".into()] },
        ];
        i.facts.front_principals = [("gunnar-ui".to_string(), "console".to_string()), ("monetize".to_string(), "monetize".to_string())].into_iter().collect();
        i.callers = vec![caller("gunnar-ui", "10.13.8.120:40000", 1_000, true), caller("monetize", "10.13.8.120:40001", 20_000, false)];
        i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
        let r = assemble(&i);
        assert_eq!(r.estate, "test2.gunnar.rs");
        assert_eq!(
            shape(&r),
            [
                ("vm2".into(), "primary".into(), "up".into()),
                ("vm3".into(), "twin".into(), "up".into()),
                ("front".into(), "front".into(), "up".into()),
            ]
        );
        let front = &r.boxes[2];
        assert_eq!(front.services, ["console", "monetize"]);
        for s in ["pairing", "caller"] {
            assert!(front.sources.iter().any(|x| x == s), "{s} in {:?}", front.sources);
        }
        assert!(r.boxes[1].sources.iter().any(|s| s == "heartbeat"), "the twin's poll landed on vm3");
    }

    /// ★ **A peer nothing here can ask is `unknown`, not `silent` and not
    /// `up`**; a console that has not called for longer than the bound is
    /// `silent`, with how long ago in its sentence.
    #[test]
    fn unknown_is_not_silent_and_an_old_call_is_silent() {
        let mut i = input();
        i.facts.peers = vec![Peer { name: "front".into(), role: "front".into(), hosts: vec![] }];
        assert_eq!(assemble(&i).boxes[1].state, "unknown");
        let mut i = input();
        i.callers = vec![caller("old-console", "10.0.0.7:1", (CALLER_SILENT_AFTER_MS + 60_000) as i64, true)];
        let r = assemble(&i);
        assert_eq!(r.boxes[1].state, "silent");
        assert!(r.boxes[1].why.contains("min ago"), "{}", r.boxes[1].why);
    }

    /// ★ **On a twin: its primary is a box**, from when it last answered.
    #[test]
    fn a_twin_draws_its_primary_from_the_primarys_answers() {
        let mut i = input();
        i.is_twin = true;
        i.primary = Some(PrimaryHeartbeat { address: "10.13.8.121:50051".into(), last_answer_unix_ms: NOW - 3_000, answer_ago_ms: 3_000, state: "ok".into(), why: "ok".into(), ..Default::default() });
        let r = assemble(&i);
        assert_eq!(shape(&r), [("this box".into(), "twin".into(), "up".into()), ("10.13.8.121".into(), "primary".into(), "up".into())]);
        i.primary.as_mut().unwrap().state = "silent".into();
        assert_eq!(assemble(&i).boxes[1].state, "silent");
    }

    /// A replication standby with no other evidence is a box of its own; a
    /// standby that has never taken a byte is `silent`.
    #[test]
    fn a_standby_is_a_box_and_never_is_silent() {
        let mut i = input();
        i.standbys = vec![StandbyStatus { name: "dr".into(), remote: "ssh://git@10.0.0.40:2222".into(), verdict: "NEVER".into(), ..Default::default() }];
        let r = assemble(&i);
        assert_eq!(shape(&r)[1], ("dr".into(), "standby".into(), "silent".into()));
    }

    #[test]
    fn hosts_are_read_out_of_every_address_shape() {
        assert_eq!(host_of("10.0.0.5:4242"), "10.0.0.5");
        assert_eq!(host_of("[::1]:50051"), "::1");
        assert_eq!(host_of("https://host.example:8443/x"), "host.example");
        assert_eq!(host_of("ssh://git@10.0.0.40:2222"), "10.0.0.40");
        assert_eq!(host_of("10.0.0.5"), "10.0.0.5");
        assert_eq!(host_of("fe80::1"), "fe80::1");
        assert_eq!(host_of(""), "");
    }

    /// The caller log keeps the NEWEST call per principal, keeps the
    /// topology mark once set, and is bounded.
    #[test]
    fn the_caller_log_is_newest_per_principal_and_bounded() {
        let log = CallerLog::default();
        log.note("c", "1.1.1.1:1", 10, true);
        log.note("c", "1.1.1.1:2", 20, false);
        log.note("c", "1.1.1.1:3", 15, false);
        let s = log.sightings();
        assert_eq!(s.len(), 1);
        assert_eq!((s[0].address.as_str(), s[0].last_unix_ms, s[0].asked_topology), ("1.1.1.1:2", 20, true));
        for n in 0..(CALLERS_CAPACITY + 5) {
            log.note(&format!("m{n}"), "", 100 + n as i64, false);
        }
        let s = log.sightings();
        assert_eq!(s.len(), CALLERS_CAPACITY);
        assert!(!s.iter().any(|x| x.principal == "c"), "the oldest went first");
    }

    fn laid() -> Record {
        let b = |name: &str, laid_as: &str, public: &str, utility: &str, size: &str, vcpu: u32, ram_mb: u64, data: &str| RecordBox {
            name: name.into(),
            laid_as: laid_as.into(),
            hostname: format!("gunnar-{name}"),
            title: format!("the {name}"),
            public: public.into(),
            utility: utility.into(),
            size: size.into(),
            vcpu,
            ram_mb,
            system_disk_gb: 40,
            data: data.into(),
        };
        Record {
            variant: "gunnar-twin-one".into(),
            versions: vec![("gunnar".into(), "abc123".into()), ("gunnar-ui".into(), "def456".into())],
            boxes: vec![
                b("front", "front", "203.0.113.9", "10.13.8.99", "STARTER-4xCPU-8GB", 4, 8192, ""),
                b("main", "main", "203.0.113.10", "10.13.8.101", "STARTER-4xCPU-16GB", 4, 16384, "4 × 10 GB maxiops, md-raid0"),
                b("twin", "twin", "203.0.113.11", "10.13.8.122", "STARTER-1xCPU-2GB", 1, 2048, "1 × 20 GB hdd, xfs"),
            ],
        }
    }

    /// ★ **The record is one encoding for both sides**: what the kit feeds is
    /// what gunnar reads, field for field; a feed with no record, or a schema
    /// this build does not read, is no record (never a half one).
    #[test]
    fn the_estate_record_round_trips_through_the_feed() {
        let r = laid();
        let feed = r.to_feed();
        assert_eq!(feed.get(FEED_RECORD).map(String::as_str), Some(RECORD_SCHEMA));
        assert_eq!(feed.get("estate.box.main.data").map(String::as_str), Some("4 × 10 GB maxiops, md-raid0"));
        assert!(feed.keys().all(|k| k.starts_with("estate.")), "the record stays in its own namespace of the shared feed");
        assert_eq!(Record::from_feed(&feed), Some(r));
        assert_eq!(Record::from_feed(&BTreeMap::new()), None);
        let mut newer = feed.clone();
        newer.insert(FEED_RECORD.into(), "2".into());
        assert_eq!(Record::from_feed(&newer), None, "a schema this build does not read is no record");
    }

    /// ★ **With the installer's record, the topology is EXACTLY the laid
    /// boxes** (owner, 2026-10-04): the front, main and the twin — with what
    /// the kit declared (addresses, size, disks), the pairing's elected roles,
    /// and the live evidence merged on: the console and the plugin
    /// registration on the front, the twin's poll on the twin. No extra box.
    #[test]
    fn with_the_record_the_topology_is_exactly_the_laid_boxes() {
        let mut i = input();
        i.facts.estate = "test2.gunnar.rs".into();
        i.facts.me = "main".into();
        i.facts.address = "10.13.8.101".into();
        i.facts.record = Some(laid());
        i.facts.peers = vec![
            Peer { name: "front".into(), role: "front".into(), hosts: vec!["10.13.8.99".into(), "10.99.0.1".into()] },
            Peer { name: "twin".into(), role: "twin".into(), hosts: vec!["10.13.8.122".into(), "10.99.0.3".into()] },
        ];
        i.facts.front_principals = [("gunnar-ui".to_string(), "console".to_string()), ("monetize".to_string(), "monetize".to_string())].into_iter().collect();
        i.callers = vec![caller("gunnar-ui", "10.13.8.99:40000", 1_000, true)];
        i.twins = vec![twin("gunnar-twin", "10.13.8.122:5222", "ok", 2_000)];
        let r = assemble(&i);
        assert!(r.recorded && r.variant == "gunnar-twin-one");
        assert_eq!(r.versions, ["gunnar abc123", "gunnar-ui def456"]);
        assert_eq!(
            shape(&r),
            [
                ("main".into(), "primary".into(), "up".into()),
                ("twin".into(), "twin".into(), "up".into()),
                ("front".into(), "front".into(), "up".into()),
            ],
            "exactly the laid boxes"
        );
        let main = &r.boxes[0];
        assert!(main.this_box && main.laid_as == "main" && main.declared_vcpu == 4 && main.declared_ram_mb == 16384);
        assert_eq!((main.public_address.as_str(), main.utility_address.as_str()), ("203.0.113.10", "10.13.8.101"));
        assert_eq!(main.declared_data, "4 × 10 GB maxiops, md-raid0");
        let twin = &r.boxes[1];
        assert!(twin.sources.iter().any(|s| s == "record") && twin.sources.iter().any(|s| s == "heartbeat"), "{:?}", twin.sources);
        assert_eq!(twin.provider_size, "STARTER-1xCPU-2GB");
        let front = &r.boxes[2];
        // The front's monetize is seen when it CALLS this box (the register fact that also
        // named it went with self_register, v2.0 step 1); here only gunnar-ui called.
        assert_eq!(front.services, ["console"]);
        assert_eq!(front.hostname, "gunnar-front");
    }

    /// The pairing's ELECTION decides primary and twin; how the kit laid a
    /// box stays on the row as `laid_as`. A laid box no evidence reaches is
    /// still drawn — `unknown`, never dropped.
    #[test]
    fn the_election_names_the_role_and_a_silent_laid_box_stays_on_the_picture() {
        let mut i = input();
        i.facts.me = "twin".into();
        i.is_twin = false; // this box was ELECTED primary although laid as twin
        i.facts.record = Some(laid());
        i.facts.peers = vec![Peer { name: "main".into(), role: "twin".into(), hosts: vec!["10.13.8.101".into()] }];
        let r = assemble(&i);
        let row = |n: &str| r.boxes.iter().find(|b| b.name == n).unwrap().clone();
        assert_eq!((row("twin").role.as_str(), row("twin").laid_as.as_str()), ("primary", "twin"));
        assert_eq!((row("main").role.as_str(), row("main").laid_as.as_str()), ("twin", "main"));
        assert_eq!(row("front").state, "unknown", "laid, and nothing here can ask it");
        assert_eq!(r.boxes.len(), 3);
    }

    fn headline(r: &TopologyResponse, name: &str) -> String {
        r.boxes.iter().find(|b| b.name == name).map(|b| b.headline.clone()).unwrap_or_else(|| panic!("no box {name}"))
    }

    /// ★ **gunnar says the state in a few words** — the words a live picture
    /// draws, so the console composes none: this box answering, a following
    /// twin, a twin behind by N entries, a twin silent for whole minutes (coarse,
    /// so the answer changes when the state does), one never heard.
    #[test]
    fn every_box_carries_gunnars_headline() {
        let mut i = input();
        i.twins = vec![
            twin("t-ok", "10.0.0.1:1", "ok", 2_000),
            TwinHeartbeat { lag_entries: 120, ..twin("t-behind", "10.0.0.2:1", "behind", 2_000) },
            twin("t-dead", "10.0.0.3:1", "silent", 4 * 60_000 + 30_000),
            twin("t-fresh", "10.0.0.4:1", "silent", 45_000),
            twin("t-never", "", "silent", 0),
        ];
        i.callers = vec![caller("console", "10.0.0.9:1", 1_000, true)];
        let r = assemble(&i);
        assert_eq!(r.boxes[0].headline, "answering");
        assert_eq!(headline(&r, "t-ok"), "following");
        assert_eq!(headline(&r, "t-behind"), "behind 120 entries");
        assert_eq!(headline(&r, "t-dead"), "silent for 4 min");
        assert_eq!(headline(&r, "t-fresh"), "silent for under a minute");
        assert_eq!(headline(&r, "t-never"), "never heard");
        assert_eq!(headline(&r, "console"), "calling");
        // Ten seconds later a dead twin's words are the same: the answer
        // changes with the state, not with the clock's seconds.
        let mut later = i.clone();
        later.now_ms += 10_000;
        later.twins[2].seen_ago_ms += 10_000;
        assert_eq!(headline(&assemble(&later), "t-dead"), "silent for 4 min");
    }

    /// The twin's OWN word rides the headline when it is not plain following —
    /// a twin being refilled after a recharge says so (`snapshot-needed`), as
    /// gunnar heard it on the poll; nothing is invented for it.
    #[test]
    fn the_twins_own_word_rides_the_headline() {
        let mut i = input();
        i.twins = vec![TwinHeartbeat { twin_state: "snapshot-needed".into(), lag_entries: 900, ..twin("t", "10.0.0.1:1", "behind", 1_000) }];
        assert_eq!(headline(&assemble(&i), "t"), "behind 900 entries · snapshot-needed");
        i.twins[0].state = "ok".into();
        assert_eq!(headline(&assemble(&i), "t"), "following · snapshot-needed");
    }

    /// ★ **After a grow the measured disk wins** (no re-feed needed): the twin's
    /// own statvfs on its poll is the row's disk, whatever the installer's
    /// record declared at the lay; the declaration stays what it was, as laid.
    #[test]
    fn after_a_grow_the_measured_disk_wins_over_the_record() {
        let mut i = input();
        i.facts.me = "main".into();
        i.facts.record = Some(laid());
        i.facts.peers = vec![Peer { name: "twin".into(), role: "twin".into(), hosts: vec!["10.13.8.122".into()] }];
        i.twins = vec![TwinHeartbeat { disk_total_bytes: 40 << 30, disk_used_bytes: 2 << 30, ..twin("gunnar-twin", "10.13.8.122:5222", "ok", 1_000) }];
        let r = assemble(&i);
        let t = r.boxes.iter().find(|b| b.name == "twin").unwrap();
        assert!(t.disk_measured && t.disk_total_bytes == 40 << 30, "the grown volume is the row's disk: {t:?}");
        assert_eq!(t.declared_data, "1 × 20 GB hdd, xfs", "the record stays what was laid");
    }
}