gunnar-server 3.0.0

//! **The console dashboard**: what a human sees on the appliance's VGA screen
//! (`/dev/tty0`) when the box is otherwise dark — a Talos-style status band
//! and a rolling list of what the server is doing.
//!
//! ```text
//!  gunnar v1.1.3-145-g1eb7083   appliance   up 3d 04:12:07   12:04:11 UTC
//!  host  gunnar-appliance   eth0 198.51.100.155/24   eth1 10.20.6.152/22
//!  cpu   2x AMD EPYC 7542 32-Core   12%   load 0.10   ram 1.2G/16.0G  7%
//!  disk  /data 3.1G/300G  1%   ssh 0.0.0.0:2222   grpc 0.0.0.0:50051
//!  ─────────────────────────────────────────────────────────────────────
//!  12:03:58  dhcp    eth0 198.51.100.155/24 via 198.51.100.1  lease 3600s
//!  12:04:11  login   alice  SHA256:ab12…  10.0.2.2
//!  12:04:12  push    ns/repo  alice  ok  812 ms
//!  12:04:40  admin   Identity.AddKey  alice  allow
//! ```
//!
//! # Where it lives, and why here
//!
//! On the appliance there is no init system: `korp-start` (edda) verifies
//! `gunnar-server` and `fexecve`s it as **PID 1** — `appliance_pid1` (edda) is the
//! operating system this process carries. A Skidbladnir service unit would be
//! a unit for a systemd that does not exist on that disk, and a separate
//! `gunnar-dashboard` binary would need a supervisor (this process) to start
//! it, a socket to hear events on, and a second copy of the netlink and
//! `/proc` readers. Fewest moving parts: one thread in `gunnar serve`, fed by
//! the log tap [`gunnar_common::logtap`] every crate already writes into, and
//! the same `open_gfx_console` the banner mirror uses. On a host with an OS
//! (`gunnar-service` under systemd, the container) it does not run at all —
//! see the gate below — because `/dev/tty0` there is somebody's real console.
//!
//! # The gate, and the promise
//!
//! [`spawn`] draws only when `appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start()`
//! (the banner mirror's own gate: PID 1 of THIS kernel, booted by korp-start),
//! or when `GUNNAR_DASHBOARD=<path>` names a device explicitly (the seam a
//! developer uses with a pty). If `/dev/tty0` cannot be opened — a kernel
//! without VT, a headless build — it returns quietly. Every write is
//! best-effort; nothing here has a path to a panic on PID 1 (`catch_unwind`
//! around the whole loop, no `unwrap` on I/O) and the thread never holds a
//! lock the server needs. The boot does not wait for it.
//!
//! # What is shown
//!
//! [`Screen`] is the whole state: the header facts, sampled every
//! [`TICK`], and a ring of the last [`EVENT_ROWS`] events. It renders to a
//! `String` — `render_frame` — which the tests assert on and the loop writes
//! to the device in one `write_all` (a 25×80 frame is 2 KB). The frame homes
//! the cursor and repaints every row padded to full width instead of clearing
//! the screen, so it does not flicker and so anything else that lands on
//! tty0 (a kernel `printk` — `console=tty0` is on the command line, the
//! panic hook's last word) is overpainted on the next tick rather than left
//! to corrupt the layout. The console loglevel is lowered to `1` on start
//! (`/proc/sys/kernel/printk`, best-effort) so only emergencies interleave;
//! the serial console, the installer's log door and — on a DEV build — the
//! sealed DEV-NARRATE stream (`crate::dev_narrate`, which follows
//! `/dev/kmsg` itself) keep the full log. (The clear-text `korp.log_udp`
//! narration this line used to name was retired 2026-09-24.)
//!
//! [`describe`] is the vocabulary: which logged lines become a row and how
//! they read. It keys on the structured `event = "…"` names the crates
//! already emit (`ssh.auth.accepted`, `git.transfer.finished`, `grpc.call`,
//! `repo.created_on_first_push`, …) plus the DHCP client's lines, and on
//! `WARN`/`ERROR` from any `gunnar::` target — everything else is left to the
//! log. A new event a lane adds shows up on the screen by adding one arm
//! here, and the test `every_shown_event_reads_as_one_short_row` pins the
//! shapes.

pub mod probe;

use std::io::Write;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::mpsc::{Receiver, RecvTimeoutError};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use gunnar_common::logtap::{self, LogLine};
use tracing::Level;

/// Header re-sample and repaint period.
pub const TICK: Duration = Duration::from_secs(2);
/// Rows the event list keeps; a VGA text console is 25 rows and the band
/// (4 lines + the rule) takes 5.
pub const EVENT_ROWS: usize = 20;
/// The VGA text mode the kernel's VT gives an appliance: 80×25. The loop
/// asks the device (`TIOCGWINSZ`) and falls back to this.
pub const DEFAULT_COLS: usize = 80;
pub const DEFAULT_ROWS: usize = 25;
/// The env seam: a device path to draw on regardless of the PID-1 gate.
pub const DEVICE_ENV: &str = "GUNNAR_DASHBOARD";
const DEVICE: &str = "/dev/tty0";

/// The facts `serve` knows and `/proc` does not.
#[derive(Debug, Clone)]
pub struct Facts {
    pub ssh_bind: SocketAddr,
    /// `None` = `--no-grpc`.
    pub grpc_bind: Option<SocketAddr>,
    pub data_root: PathBuf,
    /// `gunnar_common::BUILD_REF` — the build, not just the crate version.
    pub build: String,
    /// The engine, so the tick can ask it for index progress. `None` = no
    /// engine to ask (a render test).
    pub engine: Option<crate::storage::Engine>,
    /// **Where replication is read from.**
    ///
    /// A [`ReplicationView`](crate::replication_view::ReplicationView) rather
    /// than a replicator, because the band starts BEFORE the server does and
    /// the replicator does not exist yet. Unfilled it reads "nothing
    /// configured", and the band draws no row — which is right for a box with
    /// no standby and is not the same as drawing nothing because the wiring is
    /// broken.
    pub replication: crate::replication_view::ReplicationView,
    /// **Where the tenants row is read from.** See [`TenantView`]: unfilled
    /// on a community build, filled by a metered server at start.
    pub tenants: TenantView,
    /// **Where the FLOW row is read from.** See [`FlowView`]. Unlike
    /// [`Facts::tenants`] this is filled on EVERY build, community included:
    /// counting what moved is not the metered appliance's capability, it is
    /// every appliance's, exactly as `Usage` is.
    pub flow: FlowView,
}

/// **The counts the tenants row and `Admin.Stats` show**:
/// `tenants  paid N  grace N  suspended N`.
///
/// Defined here, engine- and feature-neutral, because two surfaces draw it
/// (the band and `Admin.Stats`) and the thing that FILLS it is behind
/// `metered`. Three integers in a struct are not plan enforcement; what
/// produces them is, and that stays where the doctrine puts it.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TenantCounts {
    pub paid: u64,
    pub grace: u64,
    /// Suspended AND Retention: both are "not paying, writes closed", which
    /// is the number a provider watches.
    pub suspended: u64,
}

/// **Where the tenants row is read from** — the `ReplicationView` shape, for
/// the same reason: the band starts before the server, and the entitlement
/// slot (if this build has one) does not exist yet. Unfilled it reads `None`
/// and the band draws no row; a metered server fills it once, at start.
#[derive(Clone, Default)]
pub struct TenantView {
    source: std::sync::Arc<std::sync::OnceLock<Box<dyn Fn() -> TenantCounts + Send + Sync>>>,
}

impl TenantView {
    pub fn new() -> Self {
        Self::default()
    }

    /// Fill it. The second call is ignored, as for `ReplicationView::fill`.
    pub fn fill(&self, source: impl Fn() -> TenantCounts + Send + Sync + 'static) {
        let _ = self.source.set(Box::new(source));
    }

    /// The counts now, or `None` when nothing fills this view — which on a
    /// community build is always, and is drawn as nothing.
    pub fn counts(&self) -> Option<TenantCounts> {
        self.source.get().map(|source| source())
    }
}

impl std::fmt::Debug for TenantView {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TenantView")
            .field("filled", &self.source.get().is_some())
            .finish()
    }
}

/// **How long a namespace must have neither pushed nor fetched before the band
/// calls it silent: 14 days.**
///
/// Two weeks, so an ordinary holiday is not an alarm and a customer who has
/// actually stopped is one. The same number the console's SILENT list uses, and
/// it is a constant rather than a setting because a threshold an operator can
/// tune is a threshold two surfaces will disagree about.
pub const SILENT_AFTER_DAYS: i64 = 14;

/// The closure a [`FlowView`] reads the last fleet fold through.
type FleetSource = Box<dyn Fn() -> Option<gunnar_core::Fleet> + Send + Sync>;

/// **Where the flow row is read from** — [`TenantView`]'s shape, for the same
/// reason: the band starts before the stores are open.
///
/// The closure behind it must be CHEAP. The fold it reports
/// (`gunnar_core::FlowLedger::fleet`) walks every namespace and reads the
/// control document, which is far too much for a two-second tick, so the flush
/// task computes it once a minute and this view hands back the last result. The
/// dashboard's rule — *reads a few atomics and nothing else* — is kept.
#[derive(Clone, Default)]
pub struct FlowView {
    source: std::sync::Arc<std::sync::OnceLock<FleetSource>>,
}

impl FlowView {
    pub fn new() -> Self {
        Self::default()
    }

    /// Fill it. The second call is ignored, as for [`TenantView::fill`].
    pub fn fill(&self, source: impl Fn() -> Option<gunnar_core::Fleet> + Send + Sync + 'static) {
        let _ = self.source.set(Box::new(source));
    }

    /// The fold now, or `None` when nothing fills this view or the first flush
    /// has not run yet — drawn as nothing, because "no row" and "a row of
    /// zeroes" are different claims and only the first is true before the first
    /// fold.
    pub fn fleet(&self) -> Option<gunnar_core::Fleet> {
        self.source.get().and_then(|source| source())
    }
}

impl std::fmt::Debug for FlowView {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FlowView")
            .field("filled", &self.source.get().is_some())
            .finish()
    }
}
/// One row of the event list.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Row {
    /// Seconds since the Unix epoch, printed as `HH:MM:SS` UTC.
    pub at: u64,
    pub kind: &'static str,
    pub text: String,
    pub warn: bool,
}

/// Everything the frame is drawn from.
#[derive(Debug, Clone)]
pub struct Screen {
    pub facts: Facts,
    pub hostname: Option<String>,
    pub addresses: Vec<probe::IfAddr>,
    /// What `/etc/resolv.conf` names, read each tick: `None` = no file.
    /// On the host line as `dns a, b` / `dns none`, because a box that can
    /// spell an address and not a name looks healthy from every other row
    /// (measured 2026-09-08: two re-images to learn `github.com` did not
    /// resolve).
    pub dns: Option<Vec<std::net::Ipv4Addr>>,
    pub cpu_model: Option<String>,
    pub cpus: usize,
    pub cpu_percent: Option<u8>,
    pub load1: Option<String>,
    /// `(total, available)` bytes.
    pub memory: Option<(u64, u64)>,
    pub disk: Option<(u64, u64)>,
    /// **The exploded read cache, summed over every open repository**:
    /// `(used, budget)`. Named on the band because it is the one number on the
    /// box that grows with READS rather than with pushes, and because an
    /// operator looking at a full `/data` needs to know which part of it the
    /// server will give back on its own. `None` until the first tick with an
    /// engine.
    pub explode: Option<(u64, u64)>,
    /// What the data root IS (`4×75G raid0 xfs`); `None` = say nothing.
    pub disk_kind: Option<probe::DiskKind>,
    /// The writer arm the stores append through (`uring` / `safe`), after
    /// the first store opened; `None` = say nothing.
    pub writer: Option<&'static str>,
    pub uptime_secs: Option<u64>,
    pub now_unix: u64,
    pub rows: std::collections::VecDeque<Row>,
    /// Lines the tap could not buffer (`logtap::dropped`), shown when > 0.
    pub dropped: u64,
    ticks: Option<probe::CpuTicks>,
    /// **The `index` rows**: one per repository with a pack being resolved,
    /// re-sampled every tick and drawn between the band and the event list —
    /// `index  alice/linux 34% 4.7M/13.9M resolve 2.1 GB/s`. Information,
    /// never a gate: the tick reads a few atomics and nothing else.
    pub index: Vec<Row>,
    /// Last tick's `(git_dir, bytes_done, at)` per active pack, for the rate.
    index_last: Vec<(PathBuf, u64, Instant)>,
    /// **The `standby` rows**, one per configured replication target, drawn
    /// with the `index` rows between the band and the event list:
    ///
    /// ```text
    ///  standby  standby  synced 105  failed 40  last 00:00:03 ago
    /// !standby  standby  synced 0  failed 8  last successful replication: NEVER
    /// ```
    ///
    /// The `!` is the staleness alarm. Nothing on this box said anything at all
    /// while every configured standby received NOTHING for weeks; this row is
    /// the answer to that, and the *never* wording is deliberately its own
    /// sentence — see [`crate::replication_view::StandbyRow::describe`].
    pub standby: Vec<Row>,
    /// **The `tenants` row** (metered only): `tenants  paid N  grace N
    /// suspended N`, re-sampled every tick like the standby rows and drawn
    /// with them. Empty when nothing fills [`Facts::tenants`].
    pub tenants: Vec<Row>,
    /// **The `flow` row**: `flow  in 4.2 GiB  out 1.1 GiB  today  ·  silent 3
    /// of 12`. The other half of the tenants band, and the half that exists on
    /// every build — what the tenants are DOING, beside what they are paying.
    ///
    /// Its own `Vec` rather than a second entry in [`Screen::tenants`] so that
    /// the two can be sampled from two different sources and either can be
    /// absent alone: a community box draws this row and no tenants row, and a
    /// metered box before the first flush draws the tenants row and not this
    /// one.
    pub flow: Vec<Row>,
}

impl Screen {
    pub fn new(facts: Facts) -> Self {
        Screen {
            facts,
            hostname: None,
            addresses: Vec::new(),
            dns: None,
            cpu_model: None,
            cpus: 0,
            cpu_percent: None,
            load1: None,
            memory: None,
            disk: None,
            explode: None,
            disk_kind: None,
            writer: None,
            uptime_secs: None,
            now_unix: 0,
            rows: std::collections::VecDeque::with_capacity(EVENT_ROWS + 1),
            dropped: 0,
            ticks: None,
            index: Vec::new(),
            index_last: Vec::new(),
            standby: Vec::new(),
            tenants: Vec::new(),
            flow: Vec::new(),
        }
    }

    /// **Re-derive the `tenants` row.** Public so the render test can feed it
    /// counts without an entitlement slot. `None` draws nothing — the
    /// community band must not show a row of zeros on a box that cannot know.
    /// Any suspended tenant makes it a WARN row: that is revenue not arriving.
    pub fn sample_tenants(&mut self, counts: Option<TenantCounts>) {
        self.tenants = counts
            .map(|c| Row {
                at: self.now_unix,
                kind: "tenants",
                text: format!(
                    "paid {}  grace {}  suspended {}",
                    c.paid, c.grace, c.suspended
                ),
                warn: c.suspended > 0,
            })
            .into_iter()
            .collect();
    }

    /// **Re-derive the `flow` row.** Public so the render test can feed it a
    /// fold without a store. `None` draws nothing — before the first flush has
    /// folded anything there is no number, and a row of zeroes would say the
    /// box served nobody today, which is a claim and not an absence.
    ///
    /// A silent tenant is a WARN row for the same reason a suspended one is:
    /// it is revenue that has stopped arriving, and it is the one an operator
    /// can still do something about.
    pub fn sample_flow(&mut self, fleet: Option<gunnar_core::Fleet>) {
        self.flow = fleet
            .map(|f| Row {
                at: self.now_unix,
                kind: "flow",
                text: format!(
                    "in {}  out {}  today  silent {} of {}",
                    short_bytes(f.bytes_in_today),
                    short_bytes(f.bytes_out_today),
                    f.silent,
                    f.namespaces
                ),
                warn: f.silent > 0,
            })
            .into_iter()
            .collect();
    }

    /// **Re-derive the `standby` rows, and raise the staleness alarm.**
    ///
    /// Public so a render test can feed it a report without a replicator —
    /// which is the only way the *never* row can be asserted at all, since a
    /// standby that has never synced is precisely a standby nobody can stand
    /// up in a test.
    pub fn sample_replication(&mut self, report: &crate::replication_view::ReplicationReport) {
        self.standby = report
            .standbys
            .iter()
            .map(|row| Row {
                at: self.now_unix,
                kind: "standby",
                text: row.describe(report.synced, report.failed, self.now_unix as i64 * 1000),
                // A stale standby — including one that has NEVER synced — is a
                // WARN row, marked `!` like any other. An alarm that looked
                // like an ordinary row would be the log line nobody read,
                // moved to a screen.
                warn: row.stale,
            })
            .collect();
        // Configured, and no rows: a replicator that is running with targets
        // must produce a row per target, so this state is a bug and says so
        // rather than drawing nothing.
        if report.configured && report.standbys.is_empty() {
            self.standby.push(Row {
                at: self.now_unix,
                kind: "standby",
                text: "replication is configured and reports NO standby rows".to_owned(),
                warn: true,
            });
        }
    }

    /// Re-derive the `index` rows from the engine's counters. Public so a
    /// test can feed it facts without an engine.
    pub fn sample_index(&mut self, facts: &[probe::IndexFacts], now: Instant) {
        let mut next_last = Vec::with_capacity(facts.len());
        let mut rows = Vec::new();
        for f in facts {
            if !f.snap.active {
                continue;
            }
            let prev = self.index_last.iter().find(|(d, _, _)| *d == f.git_dir);
            let rate = prev.and_then(|(_, bytes, at)| {
                let secs = now.saturating_duration_since(*at).as_secs_f64();
                (secs > 0.0 && f.snap.bytes_done >= *bytes)
                    .then(|| ((f.snap.bytes_done - bytes) as f64 / secs) as u64)
            });
            next_last.push((f.git_dir.clone(), f.snap.bytes_done, now));
            rows.push(Row {
                at: self.now_unix,
                kind: "index",
                text: index_row_text(
                    &repo_label(&f.git_dir, &self.facts.data_root),
                    &f.snap,
                    rate,
                ),
                warn: f.snap.phase == "paused",
            });
        }
        self.index_last = next_last;
        self.index = rows;
    }

    /// Re-read every header fact from the kernel.
    pub fn sample(&mut self) {
        self.hostname = probe::hostname();
        self.addresses = probe::addresses();
        self.dns = appliance_pid1::resolver::published();
        if self.cpu_model.is_none() {
            self.cpu_model = probe::cpu_model();
        }
        let (ticks, cpus) = probe::cpu_ticks();
        if cpus > 0 {
            self.cpus = cpus;
        }
        if let (Some(now), Some(then)) = (ticks, self.ticks) {
            if let Some(p) = now.busy_percent_since(then) {
                self.cpu_percent = Some(p);
            }
        }
        if ticks.is_some() {
            self.ticks = ticks;
        }
        self.load1 = probe::load1();
        self.memory = probe::memory();
        self.disk = probe::disk(&self.facts.data_root);
        if self.disk_kind.is_none() {
            self.disk_kind = probe::disk_kind(&self.facts.data_root);
        }
        if self.writer.is_none() {
            self.writer = crate::storage::writer_arm();
        }
        self.uptime_secs = probe::uptime_secs();
        self.now_unix = now_unix();
        self.dropped = logtap::dropped();
        if let Some(engine) = self.facts.engine {
            let facts = probe::index_progress(engine);
            self.sample_index(&facts, Instant::now());
            let bands = engine.explode_use();
            self.explode = (!bands.is_empty()).then(|| {
                bands.iter().fold((0u64, 0u64), |(u, b), (_, band)| {
                    (u + band.used(), b + band.budget)
                })
            });
        }
        // Replication, every tick. A few atomics and a mutex the workers hold
        // for microseconds; never a gate, exactly like the index rows.
        self.sample_replication(&self.facts.replication.report());
        self.sample_tenants(self.facts.tenants.counts());
        self.sample_flow(self.facts.flow.fleet());
    }

    /// Append a row, dropping the oldest past [`EVENT_ROWS`].
    pub fn push(&mut self, row: Row) {
        self.rows.push_back(row);
        while self.rows.len() > EVENT_ROWS {
            self.rows.pop_front();
        }
    }

    /// Take one logged line in, if it is one the screen shows.
    pub fn take(&mut self, line: &LogLine) {
        if let Some((kind, text, warn)) = describe(line) {
            self.push(Row {
                at: now_unix(),
                kind,
                text,
                warn,
            });
        }
    }

    /// The whole frame: cursor home, every row padded to `cols - 1`, closed
    /// with `ESC[K` (erase to end of line) and joined with `\r\n`, exactly
    /// `rows` rows. `cols - 1` and not `cols`: a full-width row leaves the
    /// VT's deferred-wrap flag set and the next `\n` would then eat a line.
    /// `ESC[K` is what clears the LAST column the padding never reaches:
    /// MEASURED on UpCloud 2026-08-29, column 80 kept stale glyphs of the
    /// serve log the band overpainted. Erase-to-EOL clears it without
    /// writing a character there, so the wrap flag stays clear.
    pub fn render_frame(&self, cols: usize, rows: usize) -> String {
        let width = cols.saturating_sub(1).max(20);
        let mut lines = self.header_lines(width);
        lines.push("\u{2500}".repeat(width));
        // Live rows first: they are re-sampled, not appended, so they sit
        // above the scrolling event list and never age out of it.
        for row in self
            .index
            .iter()
            .chain(self.standby.iter())
            .chain(self.tenants.iter())
            .chain(self.flow.iter())
        {
            let mark = if row.warn { "!" } else { " " };
            lines.push(format!(
                "{mark}{}  {:<7} {}",
                clock(row.at),
                row.kind,
                row.text
            ));
        }
        let body_rows = rows.saturating_sub(lines.len());
        let shown = self.rows.iter().rev().take(body_rows).collect::<Vec<_>>();
        for row in shown.into_iter().rev() {
            let mark = if row.warn { "!" } else { " " };
            lines.push(format!(
                "{mark}{}  {:<7} {}",
                clock(row.at),
                row.kind,
                row.text
            ));
        }
        while lines.len() < rows {
            lines.push(String::new());
        }
        lines.truncate(rows);
        let mut out = String::with_capacity(rows * (cols + 8) + 8);
        out.push_str("\x1b[H");
        for (i, line) in lines.iter().enumerate() {
            if i > 0 {
                out.push_str("\r\n");
            }
            out.push_str(&pad(line, width));
            out.push_str("\x1b[K");
        }
        out
    }

    fn header_lines(&self, width: usize) -> Vec<String> {
        let up = self
            .uptime_secs
            .map(fmt_uptime)
            .unwrap_or_else(|| "-".into());
        let dropped = if self.dropped > 0 {
            format!("  (log tap dropped {})", self.dropped)
        } else {
            String::new()
        };
        let line1 = format!(
            " gunnar {}   appliance   up {}   {} UTC{}",
            self.facts.build,
            up,
            clock(self.now_unix),
            dropped
        );
        let mut line2 = format!(" host  {}", self.hostname.as_deref().unwrap_or("(none)"));
        if self.addresses.is_empty() {
            line2.push_str("   no address");
        }
        for a in &self.addresses {
            let piece = format!("   {a}");
            if line2.len() + piece.len() > width {
                break;
            }
            line2.push_str(&piece);
        }
        let dns = match &self.dns {
            None => "   dns none".to_string(),
            Some(v) if v.is_empty() => "   dns (file names none)".to_string(),
            Some(v) => format!(
                "   dns {}",
                v.iter()
                    .map(|s| s.to_string())
                    .collect::<Vec<_>>()
                    .join(",")
            ),
        };
        if line2.len() + dns.len() <= width {
            line2.push_str(&dns);
        }
        let cpu = match (&self.cpu_model, self.cpus) {
            (Some(m), n) if n > 0 => format!("{n}x {m}"),
            (Some(m), _) => m.clone(),
            (None, n) if n > 0 => format!("{n} cpu"),
            (None, _) => "-".into(),
        };
        // Capped: "AMD Ryzen Threadripper PRO 3975WX 32-Cores" alone pushed
        // `ram` off a 79-column row (measured on the reference desktop). The count and the
        // first words are what identify a box; the tail is the sales name.
        let cpu = clip(&cpu, CPU_MODEL_COLS);
        let pct = self
            .cpu_percent
            .map(|p| format!("{p:>3}%"))
            .unwrap_or_else(|| "  -%".into());
        let ram = match self.memory {
            Some((total, avail)) => {
                let used = total.saturating_sub(avail);
                format!(
                    "{}/{} {:>2}%",
                    fmt_bytes(used),
                    fmt_bytes(total),
                    pct_of(used, total)
                )
            }
            None => "-".into(),
        };
        let line3 = format!(
            " cpu   {cpu}   {pct}   load {}   ram {ram}",
            self.load1.as_deref().unwrap_or("-")
        );
        let disk = match self.disk {
            Some((total, avail)) => {
                let used = total.saturating_sub(avail);
                format!(
                    "{}/{} {:>2}%",
                    fmt_bytes(used),
                    fmt_bytes(total),
                    pct_of(used, total)
                )
            }
            None => "-".into(),
        };
        let kind = self
            .disk_kind
            .as_ref()
            .map(|k| format!("  {k}"))
            .unwrap_or_default();
        let kind = match self.writer {
            Some(w) if kind.is_empty() => format!("  {w}"),
            Some(w) => format!("{kind} {w}"),
            None => kind,
        };
        // `explode 3.1G/8G` — the read cache and its budget. Only when there
        // is one: a box with no repository open has nothing to report, and a
        // cell reading `0/0` would say something false about a store it has
        // not looked at.
        let explode = match self.explode {
            Some((used, budget)) if budget > 0 => {
                format!("   explode {}/{}", fmt_bytes(used), fmt_bytes(budget))
            }
            Some((used, _)) => format!("   explode {} unlimited", fmt_bytes(used)),
            None => String::new(),
        };
        let grpc = self
            .facts
            .grpc_bind
            .map(|g| format!("   grpc {}", fmt_bind(g)))
            .unwrap_or_default();
        let line4 = format!(
            " disk  {} {disk}{kind}{explode}   ssh {}{grpc}",
            clip(&self.facts.data_root.display().to_string(), CPU_MODEL_COLS),
            fmt_bind(self.facts.ssh_bind)
        );
        vec![line1, line2, line3, line4]
    }
}

/// Which logged lines the screen shows, and how: `(kind, text, is_warning)`.
///
/// `None` = stays in the log. Field values arrive as `fmt` would print them
/// (see `logtap::Collect`), so a `?`-recorded `Option` reads `Some(x)` — the
/// DHCP arms unwrap that spelling.
pub fn describe(line: &LogLine) -> Option<(&'static str, String, bool)> {
    let warn = line.level <= Level::WARN;
    let get = |k: &str| line.get(k).unwrap_or("-").to_owned();
    let short_fp = |fp: &str| {
        // `SHA256:` + 43 base64 chars; 12 are plenty to recognise a key.
        let body = fp.strip_prefix("SHA256:").unwrap_or(fp);
        format!("{}\u{2026}", &body[..body.len().min(12)])
    };
    if let Some(event) = line.event() {
        return match event {
            "ssh.auth.accepted" => Some((
                "login",
                format!(
                    "{}  {}  {}",
                    get("principal"),
                    short_fp(&get("fingerprint")),
                    get("peer")
                ),
                false,
            )),
            "ssh.auth.anonymous" => {
                Some(("anon", format!("{}  public read only", get("peer")), false))
            }
            "ssh.auth.refused" | "ssh.auth.none.refused" => Some((
                "REFUSED",
                format!("login {}  {}", get("peer"), get("reason")),
                true,
            )),
            "git.transfer.finished" => {
                let verb = match line.get("service") {
                    Some("git-receive-pack") => "push",
                    Some("git-upload-pack") => "fetch",
                    _ => "git",
                };
                Some((
                    verb,
                    format!(
                        "{}  {}  {}  {} ms",
                        get("repo"),
                        get("principal"),
                        get("status"),
                        get("elapsed_ms")
                    ),
                    warn,
                ))
            }
            // Transfers are waiting for a slot rather than being refused: one
            // AGGREGATE row per second (gunnar-ssh rate-limits it), because a
            // row per waiting client is the whole console under load.
            "git.transfer.waited" => Some((
                "waiting",
                format!(
                    "{} waited  mean {} ms  worst {} ms  limit {}",
                    get("waited_total"),
                    get("mean_wait_ms"),
                    get("peak_wait_ms"),
                    get("limit")
                ),
                false,
            )),
            "git.transfer.refused" | "http.transfer.refused" => Some((
                "REFUSED",
                format!("{}  {}  {}", get("repo"), get("principal"), line.message),
                true,
            )),
            "policy.gate.refused" => Some((
                "REFUSED",
                format!("push {}  {}", get("repo"), get("principal")),
                true,
            )),
            "repo.created_on_first_push" => Some((
                "repo+",
                format!("{}  {}", get("repo"), get("principal")),
                false,
            )),
            "grpc.call" => {
                let decision = line.get("decision").unwrap_or("-");
                Some((
                    if decision == "allow" {
                        "admin"
                    } else {
                        "DENIED"
                    },
                    format!(
                        "{}  {}  {}",
                        get("method"),
                        get("principal"),
                        get("reason").replace('-', "")
                    ),
                    decision != "allow",
                ))
            }
            "store.maintained" => Some((
                "gc",
                format!(
                    "{}  {} -> {} B",
                    get("repo"),
                    get("bytes_before"),
                    get("bytes_after")
                ),
                false,
            )),
            // `writer  uring` / `writer  safe (io_uring unavailable: ENOSYS)` — the
            // arm every store appends through, and the refusal when `uring` was
            // asked for and could not be built (storage::writer).
            "store.writer_selected" => Some((
                "writer",
                match line.get("fallback").filter(|f| *f != "-" && !f.is_empty()) {
                    Some(why) => format!("{} ({why})", get("writer")),
                    None => get("writer"),
                },
                false,
            )),
            "replication.synced" | "replication.failed" => {
                Some(("mirror", format!("{}  {}", get("repo"), line.message), warn))
            }
            "http.listening" | "metrics.listening" => Some((
                "listen",
                format!("{}  {}", event.split('.').next().unwrap_or(""), get("addr")),
                false,
            )),
            // korp-installer's record off the ESP: did /data survive the last
            // install? `upgrade` = kept (users, keys, repos, the closed
            // bootstrap all still here); `fresh` = a new, empty /data.
            // `kept=korp-data` names the P3 span the boot disk kept;
            // `kept=data-set` is a multi-disk box whose /data lives on the
            // members (the record carries no span) — the set's shape comes
            // from the md sysfs, and an unknown shape prints as nothing.
            "install.mode" => Some((
                "install",
                match (line.get("mode"), line.get("kept")) {
                    (Some("upgrade"), Some("data-set")) => {
                        match line.get("kept_set").filter(|s| *s != "-" && !s.is_empty()) {
                            Some(set) => format!("upgrade  /data kept (data set {set})"),
                            None => "upgrade  /data kept (data set)".to_owned(),
                        }
                    }
                    (Some("upgrade"), _) => match line
                        .get("kept_partno")
                        .filter(|p| *p != "-" && !p.is_empty())
                    {
                        Some(partno) => {
                            format!("upgrade  /data kept (p{partno} {}G)", get("kept_gib"))
                        }
                        None => "upgrade  /data kept".to_owned(),
                    },
                    (Some(m), _) => format!("{m}  /data new"),
                    (None, _) => "-  /data new".to_string(),
                },
                false,
            )),
            // The DECLARED budgets (`crate::budget`), once, at start, after any
            // operator flag — so the row is what is in force. The three facts
            // in brackets are the point: the concurrency ceiling used to be 19
            // and nobody could tie that to anything about the machine, because
            // it was not a decision but a load-shedder's descent.
            "limits.budget" => Some((
                "limits",
                format!(
                    "transfers {}  repos {}  git mem {}  ({}c {}fd {})",
                    get("transfers"),
                    get("open_repos"),
                    crate::budget::human(get("git_memory_bytes").parse().unwrap_or(0)),
                    get("cores"),
                    get("fds"),
                    crate::budget::human(get("mem_available").parse().unwrap_or(0)),
                ),
                false,
            )),
            // The BOOT WINDOW (`crate::readiness`): the socket answers before
            // the stores are usable, and between these two rows every git
            // transfer is refused with `starting` and EX_TEMPFAIL rather than
            // a sentence that means the repository is gone.
            "serve.starting" => Some((
                "starting",
                format!("git refused (retry) until {} finishes", get("waiting_for")),
                true,
            )),
            "serve.ready" => Some((
                "serving",
                format!("git: {} took {} ms", get("waited_for"), get("ms")),
                false,
            )),
            // The handle cache giving memory back under pressure (the 2 GB
            // 2 GB collapse; `ServerStores::reclaim_under_pressure`).
            "limits.handles.reclaimed" => Some((
                "limits",
                format!(
                    "repos {} -> {}  ({} dropped, {} available)",
                    get("was"),
                    get("now"),
                    get("dropped"),
                    crate::budget::human(get("mem_available").parse().unwrap_or(0)),
                ),
                true,
            )),
            // The start-up disk test (`crate::disktest`): one row, the
            // numbers preformatted by its own formatter.
            "disktest" => Some((
                "disktest",
                format!("{}  {}", clip(&get("path"), CPU_MODEL_COLS), line.message),
                warn,
            )),
            // **The layout config** (`crate::slow`): one row, and it is a WARN
            // row whenever the copies that are supposed to outlive losing a
            // data disk are not on another one. Same shape as `disktest` above;
            // the sentence is the module's, not this file's, so the console and
            // `Admin.StorageInfo` cannot word the same verdict differently.
            "layout" => Some((
                "layout",
                format!("{}  {}", clip(&get("path"), CPU_MODEL_COLS), line.message),
                warn,
            )),
            // The post-ack indexer (znippy-plugin-git `memguard` / `absorb`):
            // the memory watchdog's pause and resume, and the per-pack audit
            // line when a pack is fully indexed. The live progress row is not
            // an event — `Screen::sample_index` re-derives it every tick.
            "index.paused" => Some((
                "index",
                format!(
                    "paused: {} available < {} budget  {}",
                    fmt_bytes(
                        line.get("available")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    fmt_bytes(line.get("budget").and_then(|v| v.parse().ok()).unwrap_or(0)),
                    short_repo(&get("repo"))
                ),
                true,
            )),
            "index.resumed" => Some((
                "index",
                format!(
                    "resumed after {} ms: {} available  {}",
                    get("paused_ms"),
                    fmt_bytes(
                        line.get("available")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    short_repo(&get("repo"))
                ),
                false,
            )),
            // **The give-up.** A pause that ends without the memory coming back
            // is not a resume and must not read like one: the rig parked
            // 31m44s and 1h45m41s and every row on the band said "paused",
            // which is true and useless. This one says the pack was refused and
            // why (`deadline` / `nothing-to-give`).
            "index.gave_up" => Some((
                "index",
                format!(
                    "GAVE UP ({}) after {} ms: {} available, {} needed  {}",
                    get("reason"),
                    get("paused_ms"),
                    fmt_bytes(
                        line.get("available")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    fmt_bytes(line.get("budget").and_then(|v| v.parse().ok()).unwrap_or(0)),
                    short_repo(&get("repo"))
                ),
                true,
            )),
            // The LIVE row, every 10 s while a pack is being absorbed
            // (`znippy_plugin_git::progress::ProgressTicker`). `peak_rss` and
            // the re-inflation count used to ride only on `index.finished`, so
            // a pack that never finished — which is the case they exist for —
            // printed neither.
            "index.progress" => Some((
                "index",
                format!(
                    "{} {}% {}/{} {} {}/s  {} re-inflations {} pauses  rss {} avail {}",
                    short_repo(&get("repo")),
                    get("percent"),
                    fmt_count(
                        line.get("objects_done")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    fmt_count(
                        line.get("objects_total")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    get("phase"),
                    fmt_count(
                        line.get("objects_per_s")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    get("reinflations"),
                    get("pauses"),
                    fmt_bytes(
                        line.get("peak_rss")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    fmt_bytes(
                        line.get("mem_available")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                ),
                false,
            )),
            "index.finished" => Some((
                "index",
                format!(
                    "{} done: {} objects in {}  {} re-inflations {} pauses  peak rss {}",
                    short_repo(&get("repo")),
                    fmt_count(
                        line.get("objects")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    fmt_ms(
                        line.get("total_ms")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    ),
                    get("reinflations"),
                    get("pauses"),
                    fmt_bytes(
                        line.get("peak_rss")
                            .and_then(|v| v.parse().ok())
                            .unwrap_or(0)
                    )
                ),
                false,
            )),
            "repo.purged" => Some((
                "delete",
                format!(
                    "{}  {} reclaimed",
                    get("repo"),
                    fmt_bytes(
                        line.get("bytes_before")
                            .and_then(|v| v.parse::<u64>().ok())
                            .unwrap_or(0)
                            .saturating_sub(
                                line.get("bytes_after")
                                    .and_then(|v| v.parse::<u64>().ok())
                                    .unwrap_or(0)
                            )
                    )
                ),
                false,
            )),
            _ => {
                if warn {
                    Some(("WARN", format!("{event}  {}", line.message), true))
                } else {
                    None
                }
            }
        };
    }
    let msg = line.message.as_str();
    if msg.starts_with("DHCP")
        || msg.contains("lease") && line.target.contains("dhcp")
        || line.target.contains("pid1") && msg.contains("DHCP")
    {
        let some = |v: String| {
            v.trim_start_matches("Some(")
                .trim_end_matches(')')
                .to_owned()
        };
        let text = if msg.starts_with("DHCP ACK") {
            format!(
                "{} {}/{} via {}  lease {}s",
                get("interface"),
                get("address"),
                get("prefix"),
                some(get("gateway")),
                get("lease_secs")
            )
        } else if msg.contains("renewed") {
            format!(
                "{} renewed via {}  next in {}s",
                get("interface"),
                get("via"),
                get("next_renew_in_secs")
            )
        } else if msg.contains("CHANGED") {
            format!(
                "{} NEW ADDRESS {}/{} via {}",
                get("interface"),
                get("address"),
                get("prefix"),
                some(get("gateway"))
            )
        } else {
            format!("{}  {}", get("interface"), first_clause(msg))
        };
        return Some(("dhcp", text, warn));
    }
    if msg.starts_with("gunnar ssh listener") && line.get("addr").is_some() {
        return Some(("listen", format!("ssh {}", get("addr")), false));
    }
    if warn && (line.target.starts_with("gunnar") || line.target.starts_with("appliance_pid1")) {
        return Some((
            if line.level == Level::ERROR {
                "ERROR"
            } else {
                "WARN"
            },
            first_clause(msg),
            true,
        ));
    }
    None
}

/// `index alice/linux 34% 4.7M/13.9M resolve 2.1 GB/s` — the live row's
/// text. `resumed` goes in front of the percent after a restart, so the row
/// says why it started from zero.
pub fn index_row_text(
    repo: &str,
    s: &crate::storage::IndexSnapshot,
    rate_bytes_per_s: Option<u64>,
) -> String {
    let mut t = format!(
        "{repo} {}{}% {}/{} {}",
        if s.resumed { "resumed " } else { "" },
        s.percent(),
        fmt_count(s.objects_done),
        fmt_count(s.objects_total),
        s.phase
    );
    if let Some(r) = rate_bytes_per_s {
        t.push_str(&format!(" {}/s", fmt_bytes(r)));
    }
    if s.pauses > 0 {
        t.push_str(&format!(" {}p", s.pauses));
    }
    t
}

/// `4.7M` / `13.9M` / `812` — object counts the width of a band.
pub fn fmt_count(n: u64) -> String {
    if n >= 1_000_000 {
        format!("{:.1}M", n as f64 / 1e6)
    } else if n >= 10_000 {
        format!("{}k", n / 1000)
    } else {
        n.to_string()
    }
}

/// `42m` / `3.2s` / `812 ms`.
pub fn fmt_ms(ms: u64) -> String {
    if ms >= 3_600_000 {
        format!("{:.1}h", ms as f64 / 3_600_000.0)
    } else if ms >= 60_000 {
        format!("{}m", ms / 60_000)
    } else if ms >= 1000 {
        format!("{:.1}s", ms as f64 / 1000.0)
    } else {
        format!("{ms} ms")
    }
}

/// `<data_root>/repos/<ns>/<name>.git` → `ns/name`; anything else is shown
/// as its last two path components.
pub fn repo_label(git_dir: &std::path::Path, data_root: &std::path::Path) -> String {
    let rel = git_dir.strip_prefix(data_root).unwrap_or(git_dir);
    let parts: Vec<String> = rel
        .components()
        .filter_map(|c| match c {
            std::path::Component::Normal(s) => Some(s.to_string_lossy().into_owned()),
            _ => None,
        })
        .collect();
    let n = parts.len();
    if n >= 2 {
        let name = parts[n - 1].strip_suffix(".git").unwrap_or(&parts[n - 1]);
        format!("{}/{name}", parts[n - 2])
    } else {
        rel.display().to_string()
    }
}

/// A store root as the plugin logs it (`…/repos/ns/name.git/gunnar-git.znippy`)
/// → `ns/name`.
fn short_repo(root: &str) -> String {
    let p = std::path::Path::new(root);
    let dir = if p.file_name().is_some_and(|f| f == "gunnar-git.znippy") {
        p.parent().unwrap_or(p)
    } else {
        p
    };
    repo_label(dir, std::path::Path::new("/"))
}

/// Up to the first `:` or `.` — the sentence's head, which is what fits.
fn first_clause(msg: &str) -> String {
    let end = msg.find([':', '.']).unwrap_or(msg.len());
    msg[..end].trim().to_owned()
}

/// The CPU cell: enough for `32x AMD EPYC 7542 32-Core`.
const CPU_MODEL_COLS: usize = 26;

/// A listener as the band spells it: `*:2222` when bound to every address
/// (the `host` row already says which those are), the full socket otherwise.
/// Measured 2026-08-29: `ssh 0.0.0.0:2222   grpc 0.0.0.0:50051` plus what
/// /data IS did not fit a 79-column row.
fn fmt_bind(addr: SocketAddr) -> String {
    if addr.ip().is_unspecified() {
        format!("*:{}", addr.port())
    } else {
        addr.to_string()
    }
}

/// At most `n` characters, `…` in the last slot when it was cut.
fn clip(s: &str, n: usize) -> String {
    if s.chars().count() <= n {
        return s.to_owned();
    }
    let mut out: String = s.chars().take(n.saturating_sub(1)).collect();
    out.push('\u{2026}');
    out
}

fn pad(line: &str, width: usize) -> String {
    let mut s: String = line.chars().take(width).collect();
    let n = s.chars().count();
    s.extend(std::iter::repeat_n(' ', width.saturating_sub(n)));
    s
}

fn pct_of(part: u64, whole: u64) -> u64 {
    if whole == 0 {
        0
    } else {
        (part as u128 * 100 / whole as u128) as u64
    }
}

/// `3.1G`, `812M`, `12K`, `900B` — one decimal above 10 units, none below.
pub fn fmt_bytes(n: u64) -> String {
    const UNITS: [&str; 5] = ["B", "K", "M", "G", "T"];
    let mut v = n as f64;
    let mut u = 0;
    while v >= 1000.0 && u < UNITS.len() - 1 {
        v /= 1024.0;
        u += 1;
    }
    if u == 0 {
        format!("{n}B")
    } else if v < 10.0 {
        format!("{v:.1}{}", UNITS[u])
    } else {
        format!("{v:.0}{}", UNITS[u])
    }
}

/// `3d 04:12:07`, or `04:12:07` under a day.
pub fn fmt_uptime(secs: u64) -> String {
    let d = secs / 86_400;
    let h = secs % 86_400 / 3600;
    let m = secs % 3600 / 60;
    let s = secs % 60;
    if d > 0 {
        format!("{d}d {h:02}:{m:02}:{s:02}")
    } else {
        format!("{h:02}:{m:02}:{s:02}")
    }
}

/// `HH:MM:SS` UTC of a Unix time.
pub fn clock(unix: u64) -> String {
    let t = unix % 86_400;
    format!("{:02}:{:02}:{:02}", t / 3600, t % 3600 / 60, t % 60)
}

/// **Bytes, short enough for an 80-column band**: `4.2G`, `917M`, `0`.
///
/// Not `gunnar_cli::human_bytes` and not `monetize_product::bytes`, and the
/// difference is the width budget rather than a second opinion about
/// arithmetic. Those two print `4.2 GiB` with the raw figure beside it, which
/// is right where a reader can compare numbers and wrong on a row that has
/// eighteen columns of clock and kind before it even starts. One unit letter,
/// one decimal, truncated — a band is a glance, and the exact number is in
/// `gunnar control usage`.
fn short_bytes(n: u64) -> String {
    const K: u64 = 1 << 10;
    for (unit, per) in [("T", 1u64 << 40), ("G", 1 << 30), ("M", 1 << 20), ("K", K)] {
        if n >= per {
            return format!("{}.{}{unit}", n / per, ((n % per) * 10) / per);
        }
    }
    n.to_string()
}

fn now_unix() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

// ── the loop ────────────────────────────────────────────────────────────────

/// Start the dashboard thread if this process should draw one; say why not
/// otherwise (one INFO line, never an error).
///
/// Returns whether it started.
pub fn spawn(facts: Facts) -> bool {
    let explicit = std::env::var(DEVICE_ENV).ok().filter(|p| !p.is_empty());
    let gated = appliance_pid1::net::is_init() && appliance_pid1::net::booted_by_korp_start();
    if explicit.is_none() && !gated {
        tracing::debug!(
            "console dashboard: not the appliance's PID 1 and {DEVICE_ENV} unset; not drawing"
        );
        return false;
    }
    let device = match &explicit {
        Some(path) => std::fs::OpenOptions::new().write(true).open(path).ok(),
        None => appliance_pid1::console::open_gfx_console(),
    };
    let Some(device) = device else {
        tracing::info!(
            device = explicit.as_deref().unwrap_or(DEVICE),
            "console dashboard: the device cannot be opened (no VT in this kernel?); not drawing"
        );
        return false;
    };
    let Some(rx) = logtap::take_receiver() else {
        tracing::info!(
            "console dashboard: the log tap was already taken or never installed; not drawing"
        );
        return false;
    };
    if explicit.is_none() {
        quiet_printk();
    }
    tracing::info!(
        device = explicit.as_deref().unwrap_or(DEVICE),
        "console dashboard drawing"
    );
    // A28 M18, stays a thread: the screen's drawer for PID 1's life, its panic caught and said, its start answered.
    std::thread::Builder::new()
        .name("gunnar-dashboard".into())
        .spawn(move || {
            // A panic here must never be PID 1's panic: it is caught, logged
            // once, and the screen simply stops updating.
            let result =
                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run(device, rx, facts)));
            if result.is_err() {
                tracing::error!(
                    "console dashboard thread panicked; the screen is frozen, serving continues"
                );
            }
        })
        .is_ok()
}

/// Console loglevel 1: only `KERN_EMERG` still prints on the VT. Serial and
/// the DEV-NARRATE kernel follower are unaffected (`/dev/kmsg` is the log, not
/// tty0).
fn quiet_printk() {
    let _ = std::fs::write("/proc/sys/kernel/printk", "1 4 1 7\n");
}

fn run(mut device: std::fs::File, rx: Receiver<LogLine>, facts: Facts) {
    let (cols, rows) = winsize(&device).unwrap_or((DEFAULT_COLS, DEFAULT_ROWS));
    let mut screen = Screen::new(facts);
    // Hide the cursor, clear once; from here on every frame overpaints.
    let _ = device.write_all(b"\x1b[?25l\x1b[2J");
    screen.sample();
    let mut next_tick = Instant::now();
    loop {
        // Drain what arrived, then wait for the next line or the tick,
        // whichever comes first.
        let now = Instant::now();
        if now >= next_tick {
            screen.sample();
            next_tick = now + TICK;
        }
        let wait = next_tick.saturating_duration_since(now);
        match rx.recv_timeout(wait) {
            Ok(line) => {
                screen.take(&line);
                while let Ok(more) = rx.try_recv() {
                    screen.take(&more);
                }
            }
            Err(RecvTimeoutError::Timeout) => {}
            Err(RecvTimeoutError::Disconnected) => break,
        }
        if device
            .write_all(screen.render_frame(cols, rows).as_bytes())
            .is_err()
        {
            // The device went away (VT switched off, pty closed): stop
            // quietly rather than spin on EIO.
            break;
        }
        let _ = device.flush();
    }
}

fn winsize(device: &std::fs::File) -> Option<(usize, usize)> {
    let ws = rustix::termios::tcgetwinsize(device).ok()?;
    if ws.ws_col == 0 || ws.ws_row == 0 {
        return None;
    }
    Some((ws.ws_col as usize, ws.ws_row as usize))
}

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

    fn line(
        level: Level,
        target: &'static str,
        event: Option<&str>,
        msg: &str,
        fields: &[(&'static str, &str)],
        spans: &[(&'static str, &str)],
    ) -> LogLine {
        let mut f: Vec<(&'static str, String)> =
            fields.iter().map(|(k, v)| (*k, v.to_string())).collect();
        if let Some(e) = event {
            f.insert(0, ("event", e.to_owned()));
        }
        LogLine {
            level,
            target,
            message: msg.to_owned(),
            fields: f,
            span_fields: spans.iter().map(|(k, v)| (*k, v.to_string())).collect(),
        }
    }

    #[test]
    fn every_shown_event_reads_as_one_short_row() {
        let conn = [
            ("peer", "10.0.2.2:5123"),
            ("principal", "alice"),
            ("repo", "ns/repo"),
            ("service", "git-receive-pack"),
        ];
        let cases: Vec<(LogLine, (&str, &str, bool))> = vec![
            (
                line(
                    Level::INFO,
                    "gunnar::ssh",
                    Some("ssh.auth.accepted"),
                    "publickey accepted",
                    &[(
                        "fingerprint",
                        "SHA256:abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG",
                    )],
                    &conn,
                ),
                ("login", "alice  abcdefghijkl\u{2026}  10.0.2.2:5123", false),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::git",
                    Some("git.transfer.finished"),
                    "transfer finished",
                    &[("status", "ok"), ("elapsed_ms", "812")],
                    &conn,
                ),
                ("push", "ns/repo  alice  ok  812 ms", false),
            ),
            (
                line(
                    Level::WARN,
                    "gunnar::audit",
                    Some("grpc.call"),
                    "refused",
                    &[
                        ("method", "Admin.X"),
                        ("principal", "eve"),
                        ("decision", "deny"),
                        ("reason", "not admin"),
                    ],
                    &[],
                ),
                ("DENIED", "Admin.X  eve  not admin", true),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::audit",
                    Some("grpc.call"),
                    "call",
                    &[
                        ("method", "Identity.AddKey"),
                        ("principal", "alice"),
                        ("decision", "allow"),
                    ],
                    &[],
                ),
                ("admin", "Identity.AddKey  alice  ", false),
            ),
            (
                line(
                    Level::INFO,
                    "appliance_pid1::dhcp",
                    None,
                    "DHCP ACK: lease taken. The renewal loop keeps it alive",
                    &[
                        ("interface", "eth0"),
                        ("address", "10.0.2.15"),
                        ("prefix", "24"),
                        ("gateway", "Some(10.0.2.2)"),
                        ("lease_secs", "3600"),
                    ],
                    &[],
                ),
                ("dhcp", "eth0 10.0.2.15/24 via 10.0.2.2  lease 3600s", false),
            ),
            (
                line(
                    Level::WARN,
                    "appliance_pid1::runtime",
                    None,
                    "DHCP renewal CHANGED THE ADDRESS mid-life — a big event",
                    &[
                        ("interface", "eth0"),
                        ("address", "10.0.2.99"),
                        ("prefix", "24"),
                        ("gateway", "Some(10.0.2.2)"),
                    ],
                    &[],
                ),
                ("dhcp", "eth0 NEW ADDRESS 10.0.2.99/24 via 10.0.2.2", true),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::git",
                    Some("repo.created_on_first_push"),
                    "created",
                    &[],
                    &conn,
                ),
                ("repo+", "ns/repo  alice", false),
            ),
            (
                line(
                    Level::WARN,
                    "gunnar_server::wg",
                    None,
                    "the tunnel refused: no address. details follow",
                    &[],
                    &[],
                ),
                ("WARN", "the tunnel refused", true),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar_server::boot",
                    Some("install.mode"),
                    "install record read off /boot/korp/install",
                    &[
                        ("mode", "upgrade"),
                        ("kept", "korp-data"),
                        ("kept_partno", "3"),
                        ("kept_gib", "225.0"),
                    ],
                    &[],
                ),
                ("install", "upgrade  /data kept (p3 225.0G)", false),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar_server::boot",
                    Some("install.mode"),
                    "install record read off /boot/korp/install",
                    &[
                        ("mode", "fresh"),
                        ("kept", "-"),
                        ("kept_partno", "-"),
                        ("kept_gib", "0.0"),
                    ],
                    &[],
                ),
                ("install", "fresh  /data new", false),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar_server::boot",
                    Some("install.mode"),
                    "install record read off /boot/korp/install",
                    &[
                        ("mode", "upgrade"),
                        ("kept", "data-set"),
                        ("kept_set", "4\u{d7}75G"),
                    ],
                    &[],
                ),
                (
                    "install",
                    "upgrade  /data kept (data set 4\u{d7}75G)",
                    false,
                ),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar_server::boot",
                    Some("install.mode"),
                    "install record read off /boot/korp/install",
                    &[("mode", "upgrade"), ("kept", "data-set"), ("kept_set", "-")],
                    &[],
                ),
                ("install", "upgrade  /data kept (data set)", false),
            ),
            (
                // A record with no kept_* lines at all must never print `p-`.
                line(
                    Level::INFO,
                    "gunnar_server::boot",
                    Some("install.mode"),
                    "install record read off /boot/korp/install",
                    &[("mode", "upgrade"), ("kept", "-")],
                    &[],
                ),
                ("install", "upgrade  /data kept", false),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar_server::disktest",
                    Some("disktest"),
                    "seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s  (1.6 s)",
                    &[("path", "/data")],
                    &[],
                ),
                (
                    "disktest",
                    "/data  seq w 812 MB/s \u{b7} r 1.9 GB/s \u{b7} 4k sync 3100 ops/s  (1.6 s)",
                    false,
                ),
            ),
            // **The layout config, off-stripe.** An informational row: the
            // copies that must outlive a data disk are on another one — and
            // the authenticity is in the same row, in the installer's own two
            // words, because a config nobody signed is the estate's reality.
            (
                line(
                    Level::INFO,
                    "gunnar::slow",
                    Some("layout"),
                    "INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
                    &[("path", "/boot/korp/layout.conf"), ("claim", "off-stripe")],
                    &[],
                ),
                (
                    "layout",
                    "/boot/korp/layout.conf  INTEGRITY ONLY \u{b7} off-stripe: p3 /dev/vda3",
                    false,
                ),
            ),
            // **And the row that matters.** A same-device layout is a WARN on
            // the console, because the mirror, the crash records and the audit
            // trail die with the stripe they were meant to outlive — and an
            // operator who sees a quiet row will never know.
            (
                line(
                    Level::WARN,
                    "gunnar::slow",
                    Some("layout"),
                    "INTEGRITY ONLY \u{b7} SAME DEVICE as /data \u{2014} not off-stripe \
                     (fallback /data)",
                    &[("path", "/boot/korp/layout.conf"), ("claim", "same-device")],
                    &[],
                ),
                (
                    "layout",
                    "/boot/korp/layout.conf  INTEGRITY ONLY \u{b7} SAME DEVICE as /data \
                     \u{2014} not off-stripe (fallback /data)",
                    true,
                ),
            ),
            // The DECLARED budget, with the three facts it came from.
            // The row exists because "why 64?" must be answerable from the
            // screen; the ceiling it replaces was 19 and was answerable from
            // nowhere.
            (
                line(
                    Level::INFO,
                    "gunnar::limits",
                    Some("limits.budget"),
                    "transfers 64 \u{b7} open repos 204",
                    &[
                        ("transfers", "64"),
                        ("open_repos", "204"),
                        ("git_memory_bytes", "4289783800"),
                        ("cores", "4"),
                        ("fds", "1048576"),
                        ("mem_available", "8579567616"),
                    ],
                    &[],
                ),
                (
                    "limits",
                    "transfers 64  repos 204  git mem 4.0G  (4c 1048576fd 8.0G)",
                    false,
                ),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::git",
                    Some("git.transfer.waited"),
                    "transfers are WAITING",
                    &[
                        ("waited_total", "412"),
                        ("mean_wait_ms", "180"),
                        ("peak_wait_ms", "1974"),
                        ("limit", "64"),
                    ],
                    &[],
                ),
                (
                    "waiting",
                    "412 waited  mean 180 ms  worst 1974 ms  limit 64",
                    false,
                ),
            ),
            // The boot window, both ends.
            (
                line(
                    Level::INFO,
                    "gunnar::limits",
                    Some("serve.starting"),
                    "git transfers are refused",
                    &[("waiting_for", "the start-up disk test")],
                    &[],
                ),
                (
                    "starting",
                    "git refused (retry) until the start-up disk test finishes",
                    true,
                ),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::limits",
                    Some("serve.ready"),
                    "serving git",
                    &[("waited_for", "the start-up disk test"), ("ms", "710")],
                    &[],
                ),
                ("serving", "git: the start-up disk test took 710 ms", false),
            ),
            // The handle cache giving memory back (the 2 GB box).
            (
                line(
                    Level::WARN,
                    "gunnar::limits",
                    Some("limits.handles.reclaimed"),
                    "memory is short",
                    &[
                        ("was", "50"),
                        ("now", "25"),
                        ("dropped", "25"),
                        ("mem_available", "432013312"),
                        ("mem_total", "2147483648"),
                    ],
                    &[],
                ),
                (
                    "limits",
                    "repos 50 -> 25  (25 dropped, 412M available)",
                    true,
                ),
            ),
            (
                line(
                    Level::WARN,
                    "gunnar_server::disktest",
                    Some("disktest"),
                    "failed: write: No space left on device",
                    &[("path", "/data")],
                    &[],
                ),
                (
                    "disktest",
                    "/data  failed: write: No space left on device",
                    true,
                ),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::store",
                    Some("store.writer_selected"),
                    "the writer arm",
                    &[
                        ("writer", "uring"),
                        ("source", "cmdline"),
                        ("fallback", "-"),
                    ],
                    &[],
                ),
                ("writer", "uring", false),
            ),
            (
                line(
                    Level::INFO,
                    "gunnar::store",
                    Some("store.writer_selected"),
                    "the writer arm",
                    &[
                        ("writer", "safe"),
                        ("source", "cmdline"),
                        ("fallback", "io_uring unavailable: ENOSYS"),
                    ],
                    &[],
                ),
                ("writer", "safe (io_uring unavailable: ENOSYS)", false),
            ),
        ];
        for (l, (kind, text, warn)) in cases {
            let got = describe(&l).unwrap_or_else(|| panic!("{:?} must be shown", l.event()));
            assert_eq!((got.0, got.1.as_str(), got.2), (kind, text, warn));
        }
    }

    #[test]
    fn the_noise_stays_in_the_log() {
        let quiet = [
            line(
                Level::INFO,
                "gunnar::git",
                Some("git.upload_pack.advertised"),
                "served the advertisement",
                &[],
                &[],
            ),
            line(
                Level::INFO,
                "gunnar_server::serve",
                None,
                "issuer ready (persisted)",
                &[],
                &[],
            ),
            line(
                Level::INFO,
                "gunnar::audit",
                Some("grpc.request.done"),
                "done",
                &[],
                &[],
            ),
            line(
                Level::WARN,
                "hyper::proto",
                None,
                "some third-party warning",
                &[],
                &[],
            ),
        ];
        for l in &quiet {
            assert_eq!(describe(l), None, "{:?} / {:?}", l.event(), l.message);
        }
    }

    #[test]
    fn numbers_format_for_a_narrow_cell() {
        assert_eq!(fmt_bytes(900), "900B");
        assert_eq!(fmt_bytes(12 * 1024), "12K");
        assert_eq!(fmt_bytes(3_300_000_000), "3.1G");
        assert_eq!(fmt_bytes(300 * 1024 * 1024 * 1024), "300G");
        assert_eq!(
            fmt_uptime(3 * 86_400 + 4 * 3600 + 12 * 60 + 7),
            "3d 04:12:07"
        );
        assert_eq!(fmt_uptime(59), "00:00:59");
        assert_eq!(clock(1_700_000_000), "22:13:20");
        assert_eq!(pad("ab", 4), "ab  ");
        assert_eq!(pad("abcdef", 4), "abcd");
        assert_eq!(
            pad("åäö", 4),
            "åäö ",
            "padding counts characters, not bytes"
        );
        assert_eq!(
            clip(
                "32x AMD Ryzen Threadripper PRO 3975WX 32-Cores",
                CPU_MODEL_COLS
            ),
            "32x AMD Ryzen Threadrippe\u{2026}"
        );
        assert_eq!(
            clip("2x AMD EPYC 7542 32-Core", CPU_MODEL_COLS),
            "2x AMD EPYC 7542 32-Core"
        );
        assert_eq!(fmt_bind("0.0.0.0:2222".parse().unwrap()), "*:2222");
        assert_eq!(fmt_bind("[::]:50051".parse().unwrap()), "*:50051");
        assert_eq!(
            fmt_bind("127.0.0.1:50051".parse().unwrap()),
            "127.0.0.1:50051"
        );
    }

    /// **The `index` rows and their event arms, pinned.** The live row is
    /// derived from the engine's snapshot (no engine here — the facts are
    /// handed in), the rate from two ticks, and `resumed` rides in front of
    /// the percent after a restart. The three `index.*` events read as one
    /// row each; the pause is a warning row, the other two are not.
    #[test]
    fn the_index_row_says_what_the_indexer_is_doing_and_the_events_read_as_rows() {
        use crate::storage::IndexSnapshot;
        let snap = |done: u64, phase: &'static str, resumed: bool, pauses: u64| IndexSnapshot {
            active: true,
            resumed,
            pack_id: 3,
            objects_done: done,
            objects_total: 13_948_244,
            bytes_done: done * 600,
            bytes_total: 8_700_000_000,
            phase,
            started_at_ms: 1,
            reinflations: 0,
            pauses,
        };
        assert_eq!(
            index_row_text(
                "alice/linux",
                &snap(4_700_000, "resolve", false, 0),
                Some(2_100_000_000)
            ),
            "alice/linux 33% 4.7M/13.9M resolve 2.0G/s"
        );
        assert_eq!(
            index_row_text("alice/linux", &snap(0, "walk", true, 0), None),
            "alice/linux resumed 0% 0/13.9M walk"
        );
        assert_eq!(
            index_row_text("a/b", &snap(13_948_244, "fold", false, 2), None),
            "a/b 100% 13.9M/13.9M fold 2p"
        );

        // The screen: two ticks, one active pack → one row with a rate; an
        // inactive counter → no row.
        let mut s = Screen::new(facts());
        s.now_unix = 1_700_000_000;
        let dir = PathBuf::from("/data/repos/alice/linux.git");
        let t0 = Instant::now();
        let f1 = probe::IndexFacts {
            git_dir: dir.clone(),
            snap: snap(1_000_000, "resolve", false, 0),
        };
        s.sample_index(&[f1], t0);
        assert_eq!(s.index.len(), 1);
        assert_eq!(s.index[0].kind, "index");
        assert_eq!(s.index[0].text, "alice/linux 7% 1.0M/13.9M resolve");
        assert!(!s.index[0].warn);
        let f2 = probe::IndexFacts {
            git_dir: dir.clone(),
            snap: snap(3_000_000, "paused", false, 1),
        };
        s.sample_index(&[f2], t0 + Duration::from_secs(2));
        assert_eq!(
            s.index[0].text,
            "alice/linux 21% 3.0M/13.9M paused 572M/s 1p"
        );
        assert!(s.index[0].warn, "a paused pack is a warning row");
        let mut idle = snap(3_000_000, "done", false, 1);
        idle.active = false;
        s.sample_index(
            &[probe::IndexFacts {
                git_dir: dir,
                snap: idle,
            }],
            t0 + Duration::from_secs(4),
        );
        assert!(s.index.is_empty(), "an idle counter draws no row");
        // The live rows sit above the event list, under the rule.
        s.sample_index(
            &[probe::IndexFacts {
                git_dir: PathBuf::from("/data/repos/a/b.git"),
                snap: snap(5, "walk", false, 0),
            }],
            t0,
        );
        s.push(Row {
            at: 1_700_000_000,
            kind: "login",
            text: "x".into(),
            warn: false,
        });
        let frame = s.render_frame(80, 25);
        let lines: Vec<&str> = frame.split("\r\n").collect();
        assert!(
            lines[5].contains("index   a/b 0% 5/13.9M walk"),
            "row 5: {:?}",
            lines[5]
        );
        assert!(lines[6].contains("login   x"), "row 6: {:?}", lines[6]);

        // The events.
        let paused = line(
            Level::WARN,
            "gunnar::index",
            Some("index.paused"),
            "index paused",
            &[
                ("available", "3113851289"),
                ("budget", "3435973836"),
                ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
            ],
            &[],
        );
        assert_eq!(
            describe(&paused),
            Some((
                "index",
                "paused: 2.9G available < 3.2G budget  alice/linux".to_string(),
                true
            ))
        );
        let resumed = line(
            Level::INFO,
            "gunnar::index",
            Some("index.resumed"),
            "index resumed",
            &[
                ("available", "5368709120"),
                ("budget", "3435973836"),
                ("paused_ms", "8120"),
                ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
            ],
            &[],
        );
        assert_eq!(
            describe(&resumed),
            Some((
                "index",
                "resumed after 8120 ms: 5.0G available  alice/linux".to_string(),
                false
            ))
        );
        let finished = line(
            Level::INFO,
            "gunnar::index",
            Some("index.finished"),
            "indexed",
            &[
                ("objects", "13948244"),
                ("total_ms", "2520000"),
                ("reinflations", "1234"),
                ("pauses", "2"),
                ("peak_rss", "9126805504"),
                ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
            ],
            &[],
        );
        assert_eq!(describe(&finished), Some(("index", "alice/linux done: 13.9M objects in 42m  1234 re-inflations 2 pauses  peak rss 8.5G".to_string(), false)));
        // The two rows the console could not show. `index.progress` is the live one
        // and `index.gave_up` is the one that says a pause ended badly.
        let progress = line(
            Level::INFO,
            "gunnar::index",
            Some("index.progress"),
            "index",
            &[
                ("percent", "78"),
                ("objects_done", "11000000"),
                ("objects_total", "13948244"),
                ("phase", "resolve"),
                ("objects_per_s", "744000"),
                ("reinflations", "12"),
                ("pauses", "1"),
                ("peak_rss", "13958643712"),
                ("mem_available", "3286511616"),
                ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
            ],
            &[],
        );
        assert_eq!(
            describe(&progress),
            Some((
                "index",
                "alice/linux 78% 11.0M/13.9M resolve 744k/s  12 re-inflations 1 pauses  rss 13G avail 3.1G".to_string(),
                false
            ))
        );
        let gave_up = line(
            Level::WARN,
            "gunnar::index",
            Some("index.gave_up"),
            "gave up",
            &[
                ("reason", "nothing-to-give"),
                ("paused_ms", "31200"),
                ("available", "3113851289"),
                ("budget", "3435973836"),
                ("repo", "/data/repos/alice/linux.git/gunnar-git.znippy"),
            ],
            &[],
        );
        assert_eq!(
            describe(&gave_up),
            Some((
                "index",
                "GAVE UP (nothing-to-give) after 31200 ms: 2.9G available, 3.2G needed  alice/linux".to_string(),
                true
            ))
        );
        let purged = line(
            Level::INFO,
            "gunnar::store",
            Some("repo.purged"),
            "reclaimed",
            &[
                ("repo", "alice/linux"),
                ("bytes_before", "26843545600"),
                ("bytes_after", "0"),
            ],
            &[],
        );
        assert_eq!(
            describe(&purged),
            Some(("delete", "alice/linux  25G reclaimed".to_string(), false))
        );
        assert_eq!(fmt_count(812), "812");
        assert_eq!(fmt_count(45_000), "45k");
        assert_eq!(fmt_ms(3_700_000), "1.0h");
        assert_eq!(
            repo_label(
                std::path::Path::new("/data/repos/ns/x.git"),
                std::path::Path::new("/data")
            ),
            "ns/x"
        );
    }

    fn facts() -> Facts {
        Facts {
            ssh_bind: "0.0.0.0:2222".parse().unwrap(),
            grpc_bind: Some("0.0.0.0:50051".parse().unwrap()),
            data_root: PathBuf::from("/data"),
            build: "v1.1.3-145-g1eb7083".into(),
            engine: None,
            replication: crate::replication_view::ReplicationView::new(),
            tenants: TenantView::new(),
            flow: FlowView::new(),
        }
    }

    /// **The tenants row: `tenants  paid N  grace N  suspended N`** — the line
    /// a service provider reads their month off, and the one the license
    /// usage line is computed from (the metered plan's per-tenant enforcement). Drawn only
    /// when a metered server filled the view; the community band, whose view
    /// is never filled, draws NOTHING — not a row of zeros, which would read
    /// as "no tenants owe anything" on a box that cannot know.
    #[test]
    fn the_tenants_row_is_drawn_when_filled_and_absent_when_not() {
        let mut s = Screen::new(facts());
        s.now_unix = 1_000;
        s.sample_tenants(None);
        let frame = s.render_frame(80, 24);
        assert!(
            !frame.contains("tenants"),
            "an unfilled view must draw no tenants row:\n{frame}"
        );
        s.sample_tenants(Some(TenantCounts {
            paid: 118,
            grace: 4,
            suspended: 3,
        }));
        let frame = s.render_frame(80, 24);
        assert!(
            frame.contains("tenants paid 118  grace 4  suspended 3"),
            "the row names all three counts:\n{frame}"
        );
        // Suspended tenants are the alarm: the row is a WARN row when any exist.
        assert!(
            frame.contains("!00:16:40  tenants"),
            "suspended > 0 is a `!` row:\n{frame}"
        );
        s.sample_tenants(Some(TenantCounts {
            paid: 1,
            grace: 0,
            suspended: 0,
        }));
        let frame = s.render_frame(80, 24);
        assert!(
            frame.contains(" 00:16:40  tenants paid 1  grace 0  suspended 0"),
            "nobody suspended is an ordinary row:\n{frame}"
        );
    }

    /// **The row that would have caught weeks of silence.**
    ///
    /// A standby that has never taken a byte must appear on the band, marked
    /// `!`, with the word NEVER in it. Asserted on the rendered FRAME — the
    /// pixels an operator actually reads — and not on the `Screen`'s fields,
    /// which is the difference between proving the row exists and proving it
    /// is drawn.
    #[test]
    fn a_standby_that_has_never_synced_is_a_warn_row_on_the_band() {
        use crate::replication_view::{ReplicationReport, StandbyRow};
        let mut s = Screen::new(facts());
        s.now_unix = 1_000;
        s.sample_replication(&ReplicationReport {
            configured: true,
            failed: 8,
            standbys: vec![StandbyRow {
                name: "standby".into(),
                remote: "ssh://git@standby:2222".into(),
                last_success_unix_ms: 0,
                stale: true,
                ..StandbyRow::default()
            }],
            ..ReplicationReport::default()
        });
        let frame = s.render_frame(100, 25);
        assert!(
            frame.contains("last successful replication: NEVER"),
            "the band must say it in those words:\n{frame}"
        );
        assert!(
            frame.contains("!00:16:40  standby"),
            "a stale standby must be a WARN row (`!`), not an ordinary one:\n{frame}"
        );

        // …and a healthy one is neither marked nor worded that way, or the
        // alarm is noise and the identity value would pass this test.
        let mut s = Screen::new(facts());
        s.now_unix = 1_000;
        s.sample_replication(&ReplicationReport {
            configured: true,
            synced: 105,
            failed: 40,
            standbys: vec![StandbyRow {
                name: "standby".into(),
                last_success_unix_ms: 997_000,
                stale: false,
                ..StandbyRow::default()
            }],
            ..ReplicationReport::default()
        });
        let frame = s.render_frame(100, 25);
        assert!(!frame.contains("NEVER"), "{frame}");
        assert!(frame.contains(" 00:16:40  standby"), "{frame}");
        assert!(
            frame.contains("synced 105  failed 40  last 00:00:03 ago"),
            "{frame}"
        );

        // Configured with no rows at all is a BUG, and the band says so
        // rather than drawing nothing — which is what "sending nothing"
        // looked like for weeks.
        let mut s = Screen::new(facts());
        s.now_unix = 1_000;
        s.sample_replication(&ReplicationReport {
            configured: true,
            ..ReplicationReport::default()
        });
        assert!(s.render_frame(100, 25).contains("reports NO standby rows"));

        // Nothing configured draws NOTHING. A box with no standby must not
        // grow a row that looks like a fault.
        let mut s = Screen::new(facts());
        s.now_unix = 1_000;
        s.sample_replication(&ReplicationReport::default());
        assert!(s.standby.is_empty());
    }

    #[test]
    fn a_frame_is_exactly_rows_lines_of_width_minus_one_and_the_newest_events_win() {
        let mut s = Screen::new(facts());
        s.hostname = Some("gunnar-appliance".into());
        s.addresses = vec![probe::IfAddr {
            interface: "eth0".into(),
            address: "198.51.100.155".parse().unwrap(),
            prefix: 24,
        }];
        s.memory = Some((16 * 1024 * 1024 * 1024, 15 * 1024 * 1024 * 1024));
        s.disk = Some((300 * 1024 * 1024 * 1024, 297 * 1024 * 1024 * 1024));
        s.uptime_secs = Some(3 * 86_400 + 4 * 3600 + 12 * 60 + 7);
        s.cpus = 4;
        s.cpu_model = Some("AMD EPYC 7542".into());
        s.disk_kind = Some(probe::DiskKind {
            fs: "xfs".into(),
            set: Some(probe::SetFacts {
                level: "raid0".into(),
                members: 4,
                member_bytes: 75 * 1024 * 1024 * 1024,
                chunk_bytes: 524288,
            }),
        });
        s.cpu_percent = Some(12);
        s.load1 = Some("0.10".into());
        s.now_unix = 1_700_000_000;
        for i in 0..(EVENT_ROWS + 5) {
            s.push(Row {
                at: i as u64,
                kind: "login",
                text: format!("user{i}"),
                warn: i % 7 == 0,
            });
        }
        let frame = s.render_frame(80, 25);
        assert!(
            frame.starts_with("\x1b[H"),
            "homes the cursor, never clears"
        );
        let raw: Vec<&str> = frame["\x1b[H".len()..].split("\r\n").collect();
        assert_eq!(raw.len(), 25);
        assert!(
            raw.iter().all(|l| l.ends_with("\x1b[K")),
            "every row ends with erase-to-EOL so column 80 is cleared: {:?}",
            raw.iter()
                .map(|l| &l[l.len().saturating_sub(4)..])
                .collect::<Vec<_>>()
        );
        let lines: Vec<&str> = raw
            .iter()
            .map(|l| l.strip_suffix("\x1b[K").unwrap())
            .collect();
        assert!(
            lines.iter().all(|l| l.chars().count() == 79),
            "every row is padded to cols-1: {:?}",
            lines.iter().map(|l| l.chars().count()).collect::<Vec<_>>()
        );
        assert!(
            lines[0].contains("gunnar v1.1.3-145-g1eb7083")
                && lines[0].contains("up 3d 04:12:07")
                && lines[0].contains("22:13:20 UTC"),
            "{}",
            lines[0]
        );
        assert!(
            lines[1].contains("host  gunnar-appliance   eth0 198.51.100.155/24"),
            "{}",
            lines[1]
        );
        assert!(
            lines[2].contains("cpu   4x AMD EPYC 7542    12%   load 0.10   ram 1.0G/16G  6%"),
            "{}",
            lines[2]
        );
        assert!(
            lines[3].contains(
                "disk  /data 3.0G/300G  1%  4\u{d7}75G raid0 xfs   ssh *:2222   grpc *:50051"
            ),
            "{}",
            lines[3]
        );
        // The writer arm rides the disk cell once a store has been built.
        s.writer = Some("uring");
        let frame = s.render_frame(80, 25);
        let l3 = frame["\x1b[H".len()..].split("\r\n").nth(3).unwrap();
        assert!(
            l3.contains("4\u{d7}75G raid0 xfs uring   ssh *:2222   grpc *:50051"),
            "{l3}"
        );
        assert_eq!(
            l3.strip_suffix("\x1b[K").unwrap().chars().count(),
            79,
            "still fits: {l3}"
        );
        s.writer = None;
        assert!(lines[4].starts_with("\u{2500}\u{2500}\u{2500}"));
        assert_eq!(s.rows.len(), EVENT_ROWS, "the ring holds EVENT_ROWS");
        assert!(
            lines[5].contains("user5"),
            "the oldest kept row is the first body row: {}",
            lines[5]
        );
        assert!(
            lines[24].contains("user24"),
            "the newest row is the last: {}",
            lines[24]
        );
        assert!(
            lines[7].starts_with('!'),
            "a warning row is marked: {}",
            lines[7]
        );
        assert!(
            lines[5].starts_with(' '),
            "a plain row is not: {}",
            lines[5]
        );
    }

    #[test]
    fn a_dark_screen_says_dash_not_zero() {
        // Nothing sampled: every unknown reads as a dash or (none), never as
        // a fabricated 0 (the identity-value trap: measure, never assert).
        let s = Screen::new(facts());
        let frame = s.render_frame(80, 25);
        assert!(
            frame.contains("up -")
                && frame.contains("host  (none)   no address")
                && frame.contains("cpu   -     -%   load -   ram -")
                && frame.contains("disk  /data -   ssh *:2222"),
            "{frame}"
        );
    }

    #[test]
    fn the_disk_cell_says_what_data_is_from_fixture_mounts_and_sysfs() {
        // /proc/mounts as the appliance writes it: the md node korp-start
        // mknod'ed, xfs; plus a FAT single-disk box, and a non-mount root.
        const MOUNTS: &str = "rootfs / rootfs rw 0 0\nproc /proc proc rw 0 0\n/dev/md/korp-data /data xfs rw,nosuid,nodev,noexec,relatime,attr2,inode64 0 0\n/dev/vda3 /mnt/fat\\040box vfat rw,dirsync 0 0\n";
        let md = probe::parse_mounts(MOUNTS, "/data").unwrap();
        assert_eq!(
            md,
            probe::MountFacts {
                source: "/dev/md/korp-data".into(),
                fstype: "xfs".into()
            }
        );
        let fat = probe::parse_mounts(MOUNTS, "/mnt/fat box").unwrap();
        assert_eq!(fat.fstype, "vfat");
        assert_eq!(
            probe::parse_mounts(MOUNTS, "/tmp/x"),
            None,
            "a plain directory is no mount"
        );

        // A sysfs tree as the kernel lays it under /sys/block/md127 for a
        // 4×75 GiB raid0 (dev-*/size is KiB).
        let dir = tempfile::tempdir().unwrap();
        let mdir = dir.path().join("md");
        std::fs::create_dir_all(&mdir).unwrap();
        std::fs::write(mdir.join("level"), "raid0\n").unwrap();
        std::fs::write(mdir.join("raid_disks"), "4\n").unwrap();
        std::fs::write(mdir.join("chunk_size"), "524288\n").unwrap();
        for m in ["vdb1", "vdc1", "vdd1", "vde1"] {
            let d = mdir.join(format!("dev-{m}"));
            std::fs::create_dir_all(&d).unwrap();
            std::fs::write(d.join("size"), "78643200\n").unwrap();
        }
        let set = probe::read_set_facts(dir.path()).unwrap();
        assert_eq!(
            set,
            probe::SetFacts {
                level: "raid0".into(),
                members: 4,
                member_bytes: 75 * 1024 * 1024 * 1024,
                chunk_bytes: 524288
            }
        );
        assert_eq!(
            probe::DiskKind::from_mount(&md, Some(set)).to_string(),
            "4\u{d7}75G raid0 xfs"
        );
        assert_eq!(probe::DiskKind::from_mount(&fat, None).to_string(), "fat");
        assert_eq!(
            probe::DiskKind::from_mount(&md, None).to_string(),
            "xfs",
            "a plain xfs disk names only the filesystem"
        );

        // Half a tree: level but no members → nothing, never `0×0G`.
        let half = tempfile::tempdir().unwrap();
        std::fs::create_dir_all(half.path().join("md")).unwrap();
        std::fs::write(half.path().join("md/level"), "raid0\n").unwrap();
        std::fs::write(half.path().join("md/raid_disks"), "4\n").unwrap();
        assert_eq!(probe::read_set_facts(half.path()), None);
        assert_eq!(probe::read_set_facts(&dir.path().join("nowhere")), None);
    }

    #[test]
    fn the_address_line_does_not_overflow_the_width() {
        let mut s = Screen::new(facts());
        for i in 0..12u8 {
            s.addresses.push(probe::IfAddr {
                interface: format!("eth{i}"),
                address: format!("10.0.{i}.15").parse().unwrap(),
                prefix: 24,
            });
        }
        let frame = s.render_frame(80, 25);
        let line2 = frame.split("\r\n").nth(1).unwrap();
        assert!(
            line2.contains("eth0 10.0.0.15/24") && !line2.contains("eth11"),
            "{line2}"
        );
    }
}