nordisk-pairing 0.5.28

//! **A VM's boot log, kept for the installer** (0.3.0, owner 2026-09-24).
//!
//! The installer on the operator's machine collects each VM's FULL boot log
//! over the bootstrap connection — every line from the first one, including
//! the install phase the medium's installer wrote onto the disk. The VM keeps
//! the lines here until the installer has read them.
//!
//! **Bounded.** A VM that talks for an hour before the installer reaches it
//! must not grow without end: the evictable part holds at most `cap` lines
//! AND `max_bytes` bytes (each line cut at [`MAX_LINE`] bytes). When it is
//! full the OLDEST line goes and is counted, and a reader that asks for lines
//! no longer held gets ONE marker line first — `… N lines dropped …` — so a
//! log that lost its start never reads as a whole one.
//!
//! **The pinned head** (0.5.5). MEASURED 2026-09-28 on test2's main VM: the
//! installer's log door answered `… 2243 lines dropped`, the pairing lines
//! were gone after the close, and 8 502 of the 20 008 held lines read back as
//! `<<WITHHELD: a pubkey-class line …>>` — one ssh login or audited RPC per
//! line, each carrying a fingerprint; gunnar.rs's main held 7 642 after
//! 3.5 h. A log made [`BootLog::with_head`] keeps its first lines — the
//! install, the pairing, the first boot — for the log's whole life: they are
//! never evicted, whatever the runtime says afterwards. The head takes lines
//! until [`BootLog::seal_head`] (the install is over) or until its own bound
//! is reached; every later line goes to the evictable tail.
//!
//! **Redaction never multiplies lines** (0.5.5). Every line is cleaned HERE,
//! on the VM, before it is held ([`crate::redact::cleaned`]): a public key is
//! withheld INSIDE its line (the line is progress, the key is not), and a
//! line of a whole-line class (token, key, cmdline, …) is held as ONE marker
//! — and a run of them in a row is folded into that one marker with a count,
//! so a flood of secret-bearing lines costs the log one line, not one each.
//! A marker already handed to a reader is never rewritten; the next withheld
//! line of the run starts a new marker instead, so a reader never misses a
//! count it was shown.
//!
//! Every line has a sequence number, from 0 for the first line pushed. A
//! reader asks "everything from `n`" and is told where to continue
//! ([`Chunk::next`]), so a reader that reconnects neither loses nor repeats a
//! line.

use std::collections::VecDeque;
use std::sync::Mutex;

use crate::redact::{self, Cleaned};

/// A line longer than this is cut (and says so).
pub const MAX_LINE: usize = 4096;

/// The default bound of the evictable part: 20 000 lines.
pub const DEFAULT_CAP: usize = 20_000;

/// **The default byte bound of the evictable part** (0.5.5): 6 MiB. Before
/// 0.5.5 the log was bounded in lines only, so 20 000 lines of up to
/// [`MAX_LINE`] bytes was 80 MiB at the very worst, not "a few megabytes".
pub const DEFAULT_BYTES: usize = 6 << 20;

/// **The pinned head's default bounds** (0.5.5): the install-time lines a
/// process keeps for its whole life — 8 000 lines, 2 MiB. So a log made
/// with every default holds at most `HEAD_BYTES + DEFAULT_BYTES` = 8 MiB of
/// line text.
pub const HEAD_CAP: usize = 8_000;
pub const HEAD_BYTES: usize = 2 << 20;

/// The most lines one read hands out; the reader asks again from `next`.
pub const MAX_READ: usize = 2_000;

/// A run of withheld lines of one class, folded into the last held line.
#[derive(Debug)]
struct Run {
    class: &'static str,
    count: u64,
    bytes: u64,
}

#[derive(Debug)]
struct Ring {
    /// **The pinned head**: sequence numbers `0..head.len()`, never evicted.
    head: Vec<String>,
    head_bytes: usize,
    head_cap: usize,
    head_max_bytes: usize,
    /// No more lines go to the head.
    sealed: bool,
    /// The evictable tail, oldest first.
    lines: VecDeque<String>,
    bytes: usize,
    /// The sequence number of `lines[0]` (of the next line, when empty).
    first: u64,
    cap: usize,
    max_bytes: usize,
    /// The highest `next` any reader was handed: a line below it has been
    /// read and is never rewritten.
    served: u64,
    /// The last held line is a withheld marker for this run.
    run: Option<Run>,
    /// Lines withheld (held as, or folded into, a marker), ever.
    withheld: u64,
}

impl Ring {
    fn total(&self) -> u64 {
        self.first + self.lines.len() as u64
    }

    /// Fold one more withheld line of `class` into the last held line, when
    /// that line is this run's unread marker and the bound allows it.
    fn fold(&mut self, class: &'static str, bytes: usize) -> bool {
        let total = self.total();
        if total == 0 || total - 1 < self.served {
            return false;
        }
        let Some(run) = self.run.as_mut() else { return false };
        if run.class != class {
            return false;
        }
        let (count, sum) = (run.count + 1, run.bytes + bytes as u64);
        let text = redact::withheld_run(class, count, sum);
        if !self.sealed {
            // The run's marker is the head's last line.
            let Some(last) = self.head.last() else { return false };
            let grown = self.head_bytes - last.len() + text.len();
            if grown > self.head_max_bytes {
                return false;
            }
            self.head_bytes = grown;
            *self.head.last_mut().expect("checked") = text;
        } else {
            let Some(last) = self.lines.back() else { return false };
            let grown = self.bytes - last.len() + text.len();
            if grown > self.max_bytes {
                return false;
            }
            self.bytes = grown;
            *self.lines.back_mut().expect("checked") = text;
        }
        let run = self.run.as_mut().expect("checked");
        run.count = count;
        run.bytes = sum;
        true
    }

    /// Hold one line: in the head while it takes lines, else in the tail,
    /// evicting the oldest tail lines until both bounds hold.
    fn hold(&mut self, line: String) {
        let len = line.len();
        if !self.sealed {
            if self.head.len() < self.head_cap && self.head_bytes + len <= self.head_max_bytes {
                // The tail is empty while the head takes lines, so the next
                // sequence number moves with the head.
                self.head_bytes += len;
                self.head.push(line);
                self.first += 1;
                return;
            }
            // Full: the head is what it is, and the rest is the tail.
            self.sealed = true;
        }
        if self.cap == 0 || len > self.max_bytes {
            self.first += 1;
            return;
        }
        while !self.lines.is_empty() && (self.lines.len() >= self.cap || self.bytes + len > self.max_bytes) {
            if let Some(old) = self.lines.pop_front() {
                self.bytes -= old.len();
            }
            self.first += 1;
        }
        self.bytes += len;
        self.lines.push_back(line);
    }
}

/// **The log.** `const`-constructible, so a process can keep it in a
/// `static` and push to it from its very first line.
#[derive(Debug)]
pub struct BootLog {
    ring: Mutex<Ring>,
    /// **This log's boot** (estate-box3): drawn at its first read, never 0.
    /// A restarted process is a new log with a new boot, so a reader whose
    /// cursor belongs to another boot reads this one from line 0 ([`resume`]).
    boot: std::sync::OnceLock<u64>,
}

/// A number no other log of this machine's processes draws: the clock's
/// nanoseconds, the process id and a per-process count, never 0.
fn draw_boot() -> u64 {
    static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(0);
    let id = nanos ^ (u64::from(std::process::id()) << 40) ^ N.fetch_add(1, std::sync::atomic::Ordering::Relaxed).wrapping_mul(0x9E37_79B9_7F4A_7C15);
    id.max(1)
}

/// **Where a reader stands in a VM's log**: the boot it was reading (0 when
/// the VM names none — an older build) and the next line to ask for.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Cursor {
    pub boot: u64,
    pub next: u64,
}

/// **After a read from `at`**, pure: where to ask next, and whether the VM's
/// log started over — a boot other than the cursor's, or (from a VM that
/// names no boot) fewer lines than the cursor. A new boot is read from line
/// 0: the chunk's lines, read from the old cursor, are NOT the new boot's
/// first lines and are not kept.
pub fn resume(at: Cursor, c: &Chunk) -> (Cursor, bool) {
    if at.boot != 0 && c.boot != 0 && c.boot != at.boot {
        return (Cursor { boot: c.boot, next: 0 }, true);
    }
    if c.next < at.next {
        return (Cursor { boot: c.boot, next: 0 }, true);
    }
    (Cursor { boot: if c.boot != 0 { c.boot } else { at.boot }, next: c.next }, false)
}

/// One read: the lines from `from`, and where the next read starts.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Chunk {
    /// How many lines between the asked-for `from` and the first line held
    /// were dropped (0 when none). The wire carries it as a marker line.
    pub dropped: u64,
    pub lines: Vec<String>,
    /// Ask from here next time.
    pub next: u64,
    /// The log's boot ([`BootLog`]); 0 from a VM that names none.
    pub boot: u64,
}

impl Default for BootLog {
    fn default() -> Self {
        BootLog::new(DEFAULT_CAP)
    }
}

impl BootLog {
    /// A log with no pinned head: at most `cap` lines and [`DEFAULT_BYTES`],
    /// the oldest dropped first.
    pub const fn new(cap: usize) -> BootLog {
        BootLog::with_head(cap, DEFAULT_BYTES, 0, 0)
    }

    /// **A log with a pinned head** (0.5.5): the first lines pushed — at most
    /// `head_cap` lines and `head_bytes` bytes, and none after
    /// [`BootLog::seal_head`] — are kept for the log's life; the rest is a
    /// ring of at most `cap` lines and `max_bytes` bytes. The whole log never
    /// holds more than `head_bytes + max_bytes` bytes of line text.
    pub const fn with_head(cap: usize, max_bytes: usize, head_cap: usize, head_bytes: usize) -> BootLog {
        BootLog {
            ring: Mutex::new(Ring {
                head: Vec::new(),
                head_bytes: 0,
                head_cap,
                head_max_bytes: head_bytes,
                sealed: head_cap == 0 || head_bytes == 0,
                lines: VecDeque::new(),
                bytes: 0,
                first: 0,
                cap,
                max_bytes,
                served: 0,
                run: None,
                withheld: 0,
            }),
            boot: std::sync::OnceLock::new(),
        }
    }

    /// **The log an installed process keeps** (0.5.23, Gunnar f21d8d09 shared): a pinned
    /// head and a tail at every default bound — [`HEAD_CAP`] / [`HEAD_BYTES`] for the
    /// install, [`DEFAULT_CAP`] / [`DEFAULT_BYTES`] for the runtime, 8 MiB of text at
    /// most. The listener seals the head when the installer closes the install, and at
    /// once on a restart that finds it closed
    /// ([`crate::bootstrap::serve_for_life_or_after_close_at`]), so a product only makes
    /// its `static` with this. A log made with [`BootLog::new`] has no head: a box that
    /// talks for hours loses its install and its pairing lines.
    pub const fn for_install() -> BootLog {
        BootLog::with_head(DEFAULT_CAP, DEFAULT_BYTES, HEAD_CAP, HEAD_BYTES)
    }

    /// **The install is over**: every later line goes to the evictable tail.
    /// Called when the installer closes the install (or a restart finds it
    /// closed). Idempotent.
    pub fn seal_head(&self) {
        let mut r = self.ring.lock().unwrap_or_else(|p| p.into_inner());
        if !r.sealed {
            r.sealed = true;
            // A run folding into the head stops there; the tail starts afresh.
            r.run = None;
        }
    }

    /// Lines in the pinned head (sequence numbers `0..head_len`).
    pub fn head_len(&self) -> u64 {
        self.ring.lock().unwrap_or_else(|p| p.into_inner()).head.len() as u64
    }

    /// Bytes of line text held, head and tail together.
    pub fn held_bytes(&self) -> usize {
        let r = self.ring.lock().unwrap_or_else(|p| p.into_inner());
        r.head_bytes + r.bytes
    }

    /// Lines withheld so far (each is in a marker, alone or folded).
    pub fn withheld(&self) -> u64 {
        self.ring.lock().unwrap_or_else(|p| p.into_inner()).withheld
    }

    /// Push text; every `\n`-separated line is one entry (empty ones and a
    /// trailing `\r` are dropped). Each line is cleaned before it is held
    /// ([`crate::redact::cleaned`]); a withheld line right after a withheld
    /// line of the same class is folded into that line's marker.
    pub fn push(&self, text: &str) {
        let mut r = self.ring.lock().unwrap_or_else(|p| p.into_inner());
        for line in text.split('\n') {
            let line = line.strip_suffix('\r').unwrap_or(line);
            if line.trim().is_empty() {
                continue;
            }
            match redact::cleaned(line) {
                Cleaned::Withheld { class, bytes } => {
                    r.withheld += 1;
                    if r.fold(class, bytes) {
                        continue;
                    }
                    r.hold(redact::withheld(class, bytes));
                    r.run = Some(Run { class, count: 1, bytes: bytes as u64 });
                }
                Cleaned::Kept(kept) => {
                    let kept = if kept.len() > MAX_LINE {
                        let mut cut = MAX_LINE;
                        while !kept.is_char_boundary(cut) {
                            cut -= 1;
                        }
                        format!("{} … (cut at {MAX_LINE} bytes)", &kept[..cut])
                    } else {
                        kept.into_owned()
                    };
                    r.hold(kept);
                    r.run = None;
                }
            }
        }
    }

    /// Lines ever held (a folded run counts once), held or dropped.
    pub fn total(&self) -> u64 {
        self.ring.lock().unwrap_or_else(|p| p.into_inner()).total()
    }

    /// Lines dropped because the log was full (never a head line).
    pub fn dropped(&self) -> u64 {
        let r = self.ring.lock().unwrap_or_else(|p| p.into_inner());
        r.first - r.head.len() as u64
    }

    /// **Everything from `from`**, at most `max` lines. A read that starts in
    /// the head ends at the head's end when the tail lost lines after it, so
    /// the next read carries the dropped count exactly where the lines went.
    pub fn since(&self, from: u64, max: usize) -> Chunk {
        let boot = *self.boot.get_or_init(draw_boot);
        let mut r = self.ring.lock().unwrap_or_else(|p| p.into_inner());
        let h = r.head.len() as u64;
        let mut lines: Vec<String> = Vec::new();
        let mut from = from;
        if from < h {
            lines.extend(r.head.iter().skip(from as usize).take(max).cloned());
            let next = from + lines.len() as u64;
            if next < h || lines.len() == max || r.first > h {
                r.served = r.served.max(next);
                return Chunk { dropped: 0, lines, next, boot };
            }
            from = h;
        }
        let end = r.total();
        let start = from.max(r.first).min(end);
        // Head lines were handed out above only when no tail line was dropped
        // after the head (`first == h`), so a gap is reported only on a read
        // that starts in the tail.
        let dropped = if lines.is_empty() { start.saturating_sub(from.min(end)) } else { 0 };
        let skip = (start - r.first) as usize;
        let before = lines.len();
        lines.extend(r.lines.iter().skip(skip).take(max - before).cloned());
        let next = start + (lines.len() - before) as u64;
        r.served = r.served.max(next);
        Chunk { dropped, lines, next, boot }
    }
}

/// Where korp-start and the medium's installer wrote a box's story so far
/// (the ESP, mounted at `/boot`).
pub const ESP_BOOTLOG: &str = "/boot/BOOTLOG.TXT";

impl BootLog {
    /// **The box's story before the product** (0.5.5, gunnar-server's
    /// `installer_listener::capture`): the ESP's `BOOTLOG.TXT` — the medium's
    /// install phase, korp-start, this boot so far — between two marker lines
    /// naming `product`. Called once, first thing. `false` = no such file (a
    /// container, a dev box): nothing is pushed.
    pub fn capture_esp(&self, path: &std::path::Path, product: &str) -> bool {
        let Ok(text) = std::fs::read_to_string(path) else { return false };
        self.push(&format!("── {} on the ESP: the install and this boot before {product} ──", path.display()));
        self.push(&text);
        self.push(&format!("── {product} ──"));
        true
    }
}

/// **The boot-log tap** (0.5.5): a writer that hands every whole line to the
/// [`BootLog`] the installer collects, and passes the bytes on to `inner`
/// unchanged (stderr, the console). Any logger that writes to a `Write` takes
/// it — `env_logger::Target::Pipe`, a `tracing_subscriber` writer — so the
/// product's own log lines stream to the installer with no logger of ours.
/// A line is pushed when its `\n` arrives; a partial one waits (bounded at
/// [`MAX_LINE`]).
pub struct Tap<W: std::io::Write> {
    log: &'static BootLog,
    inner: W,
    partial: Vec<u8>,
}

impl<W: std::io::Write> Tap<W> {
    pub fn new(log: &'static BootLog, inner: W) -> Tap<W> {
        Tap { log, inner, partial: Vec::new() }
    }
}

impl<W: std::io::Write> std::io::Write for Tap<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        for &b in &buf[..n] {
            if b == b'\n' || self.partial.len() >= MAX_LINE {
                self.log.push(&String::from_utf8_lossy(&self.partial));
                self.partial.clear();
                if b == b'\n' {
                    continue;
                }
            }
            self.partial.push(b);
        }
        Ok(n)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

/// **fd 2 into the boot log** (feature `stderr-capture`, owner 2026-09-28: "the
/// monetize log visible through the log door"). A library that logs with
/// `eprintln!` (monetize-server embedded in a product's front: 87 of them and no
/// logger) writes fd 2 — on PID 1 the serial console, which the installer's log
/// door never reads. After this call fd 2 is a pipe, and a thread drains it line
/// by line into `log` AND onto the console fd 2 was, so nothing is lost on the
/// serial line either.
///
/// Returns that console (a duplicate of the old fd 2). Hand IT to a [`Tap`] as
/// the inner writer — a Tap over `std::io::stderr()` would now write into the
/// pipe and every tracing line would be pushed twice. A line longer than
/// [`MAX_LINE`] is cut into pieces, as the Tap cuts it. On an error nothing has
/// changed: fd 2 is still the console.
#[cfg(all(unix, feature = "stderr-capture"))]
pub fn capture_stderr(log: &'static BootLog) -> std::io::Result<std::fs::File> {
    use std::io::{BufRead, Read, Write};
    use std::os::fd::AsFd;
    let console = std::fs::File::from(std::io::stderr().as_fd().try_clone_to_owned()?);
    let mut drain_console = console.try_clone()?;
    let (reader, writer) = std::io::pipe()?;
    rustix::stdio::dup2_stderr(&writer)?;
    drop(writer);
    // A28 M18, stays a thread: the pipe's one reader for the process's life (it ends with fd 2); its start failure is answered.
    std::thread::Builder::new().name("stderr-to-bootlog".into()).spawn(move || {
        let mut r = std::io::BufReader::new(reader);
        let mut line = Vec::with_capacity(256);
        loop {
            line.clear();
            match r.by_ref().take(MAX_LINE as u64 + 1).read_until(b'\n', &mut line) {
                Ok(0) | Err(_) => return,
                Ok(_) => {
                    let _ = drain_console.write_all(&line);
                    let text = String::from_utf8_lossy(line.strip_suffix(b"\n").unwrap_or(&line));
                    log.push(text.strip_suffix('\r').unwrap_or(&text));
                }
            }
        }
    })?;
    Ok(console)
}

/// The marker line a reader stores in place of lines that were dropped.
pub fn dropped_marker(n: u64) -> String {
    format!("… {n} lines dropped (the VM's log holds its install-time head and its newest lines) …")
}

impl Chunk {
    /// The wire: `next = N` and `dropped = D`, a blank line, then the lines.
    pub fn render(&self) -> String {
        let mut out = format!("next = {}\ndropped = {}\n", self.next, self.dropped);
        if self.boot != 0 {
            out.push_str(&format!("boot = {}\n", self.boot));
        }
        out.push('\n');
        for l in &self.lines {
            out.push_str(l);
            out.push('\n');
        }
        out
    }

    pub fn parse(text: &str) -> Result<Chunk, String> {
        let (head, body) = text.split_once("\n\n").ok_or("a log chunk without its header")?;
        let mut next = None;
        let mut dropped = 0;
        let mut boot = 0;
        for l in head.lines() {
            match l.split_once('=').map(|(k, v)| (k.trim(), v.trim())) {
                Some(("next", v)) => next = v.parse().ok(),
                Some(("dropped", v)) => dropped = v.parse().map_err(|_| format!("`{l}` is not a count"))?,
                Some(("boot", v)) => boot = v.parse().map_err(|_| format!("`{l}` is not a boot"))?,
                _ => {}
            }
        }
        let next = next.ok_or("a log chunk without `next`")?;
        Ok(Chunk { dropped, lines: body.lines().map(str::to_string).collect(), next, boot })
    }

    /// The lines as they are stored: the dropped marker first, when any were.
    pub fn stored(&self) -> Vec<String> {
        let mut v = Vec::with_capacity(self.lines.len() + 1);
        if self.dropped > 0 {
            v.push(dropped_marker(self.dropped));
        }
        v.extend(self.lines.iter().cloned());
        v
    }
}

// ── the box's own copy, on its disk (RUNTIME LOGS, 2026-09-30) ────────────

/// **The bound of a box's own log file.** Two files are kept — the current
/// and the previous one (`<path>.1`) — so the box never holds more than
/// `2 × KEEP_BYTES` of its own log on its disk, whatever it says.
pub const KEEP_BYTES: u64 = 4 << 20;

/// How many of the previous boot's last lines the next boot carries into its
/// log (and so through the log door), at most [`CARRY_BYTES`] of them.
pub const CARRY_LINES: usize = 400;
pub const CARRY_BYTES: usize = 128 << 10;

/// How often the keeper writes (and syncs) what the log took since.
pub const KEEP_EVERY: std::time::Duration = std::time::Duration::from_secs(1);
/// The duty that keeps the log on disk ([`BootLog::keep_on_disk`], A28 M18).
pub const KEEP_DUTY: &str = "bootlog-keep";

/// The prefix of a carried line: the log door serves it, and no reader
/// takes it for a line of this boot (a log gate parses no level after it).
pub const CARRIED: &str = "(previous boot) ";

/// The previous file beside `path`: `<path>.1`.
pub fn rotated(path: &std::path::Path) -> std::path::PathBuf {
    let mut s = path.as_os_str().to_os_string();
    s.push(".1");
    std::path::PathBuf::from(s)
}

/// Pure: **the last lines** of `older` then `newer` (one log split over two
/// files), at most `lines` of them and at most `bytes` of text, in order.
pub fn last_lines(older: &str, newer: &str, lines: usize, bytes: usize) -> Vec<String> {
    let mut out: VecDeque<String> = VecDeque::new();
    let mut held = 0usize;
    for l in newer.lines().rev().chain(older.lines().rev()) {
        if l.trim().is_empty() {
            continue;
        }
        if out.len() >= lines || held + l.len() + 1 > bytes {
            break;
        }
        held += l.len() + 1;
        out.push_front(l.to_string());
    }
    out.into()
}

/// **Append `text` to `path`, bounded**: when the file would pass `max_bytes`
/// it becomes `<path>.1` first (the older `.1` goes). Open, append, sync,
/// close — no descriptor is held between writes, so nothing keeps the
/// filesystem busy.
pub fn append_bounded(path: &std::path::Path, text: &str, max_bytes: u64) -> std::io::Result<()> {
    use std::io::Write as _;
    if text.is_empty() {
        return Ok(());
    }
    if let Some(d) = path.parent() {
        std::fs::create_dir_all(d)?;
    }
    let have = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
    if have > 0 && have + text.len() as u64 > max_bytes {
        std::fs::rename(path, rotated(path))?;
    }
    // A single batch larger than the bound keeps its newest part.
    let text = if text.len() as u64 > max_bytes {
        let mut cut = text.len() - max_bytes as usize;
        while !text.is_char_boundary(cut) {
            cut += 1;
        }
        let t = &text[cut..];
        t.find('\n').map(|n| &t[n + 1..]).unwrap_or(t)
    } else {
        text
    };
    let mut f = std::fs::OpenOptions::new().create(true).append(true).open(path)?;
    f.write_all(text.as_bytes())?;
    f.sync_data()
}

impl BootLog {
    /// **The previous boot's last lines, carried into this log.** Reads what
    /// the last boot kept (`<path>.1`, then `path`), pushes its last `lines`
    /// (at most `bytes`) between two marker lines, each line prefixed
    /// [`CARRIED`], and moves `path` to `<path>.1`: this boot keeps a fresh
    /// file. Answers how many lines were carried (0: nothing was kept).
    pub fn carry_previous(&self, path: &std::path::Path, lines: usize, bytes: usize) -> usize {
        let newer = std::fs::read_to_string(path).unwrap_or_default();
        let older = std::fs::read_to_string(rotated(path)).unwrap_or_default();
        let last = last_lines(&older, &newer, lines, bytes);
        if !newer.is_empty() {
            let _ = std::fs::rename(path, rotated(path));
        }
        if last.is_empty() {
            return 0;
        }
        self.push(&format!("── the previous boot's last {} line(s), kept on {} (a box that restarted or crashed left them there) ──", last.len(), path.display()));
        for l in &last {
            self.push(&format!("{CARRIED}{l}"));
        }
        self.push("── end of the previous boot ──");
        last.len()
    }

    /// **One write of the box's own copy**: every line from `from` on
    /// (a dropped count as its marker) appended to `path` ([`append_bounded`]).
    /// Answers where the next write starts; a write that failed is tried
    /// again from the same place.
    pub fn keep_once(&self, path: &std::path::Path, from: u64, max_bytes: u64) -> u64 {
        let mut at = from;
        loop {
            let c = self.since(at, MAX_READ);
            if c.lines.is_empty() && c.dropped == 0 {
                return c.next.max(at).min(self.total());
            }
            let mut text = String::new();
            for l in c.stored() {
                text.push_str(&l);
                text.push('\n');
            }
            if append_bounded(path, &text, max_bytes).is_err() {
                return at;
            }
            let full = c.lines.len() >= MAX_READ;
            at = c.next;
            if !full {
                return at;
            }
        }
    }

    /// **Keep this log on the box's disk, for the life of the process**: the
    /// previous boot carried in ([`BootLog::carry_previous`]), then a thread
    /// that writes every later line to `path` every [`KEEP_EVERY`], bounded
    /// by `max_bytes` per file. A crash loses at most the last second.
    /// Answers the lines carried.
    pub fn keep_on_disk(&'static self, path: std::path::PathBuf, max_bytes: u64) -> usize {
        let carried = self.carry_previous(&path, CARRY_LINES, CARRY_BYTES);
        let from = self.total();
        // A duty (A28 M18): listed, a panicked round said and the keeping goes on, stopped at the stop.
        let mut at = from;
        let say: crate::duty::Say = std::sync::Arc::new(|l: &str| eprintln!("{l}"));
        if let Err(e) = crate::duty::start(KEEP_DUTY, KEEP_EVERY, say, move || {
            at = self.keep_once(&path, at, max_bytes);
            crate::duty::Round::Again
        }) {
            eprintln!("ALERT {KEEP_DUTY} the log is NOT kept on disk: {e}");
        }
        carried
    }
}

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

    /// ★ **The log kept on disk is a listed duty** (A28 M18): the raw thread was nowhere to be
    /// seen, and a panic in it stopped the keeping for the box's life without a word.
    #[test]
    fn the_kept_log_is_a_listed_duty() {
        let _one = crate::duty::tests::ONE.lock().unwrap_or_else(|p| p.into_inner());
        let log: &'static BootLog = Box::leak(Box::new(BootLog::new(16)));
        let dir = std::env::temp_dir().join(format!("np-bootlog-keep-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        log.keep_on_disk(dir.join("boot.log"), 1 << 20);
        let d = crate::duty::find(KEEP_DUTY).expect("the keeping is a running duty");
        assert!(d.stop(std::time::Duration::from_secs(5)), "it stops at the box's stop");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// ★ estate-box3 (a restart the reader never saw): the reader's cursor
    /// is a line number and nothing else, so a VM that restarted and logged
    /// MORE lines than the cursor before the next read is read on from the
    /// cursor — the new boot's first lines (its hello, an early ERROR) are
    /// skipped and no "started over" is ever said; the installer's own drain
    /// during the install does not even look for a shorter log. A log names
    /// its boot, and a reader that sees a new one reads it from line 0.
    #[test]
    fn a_reader_notices_a_new_boot_even_when_it_already_holds_more_lines_than_the_cursor() {
        let old = BootLog::new(100);
        for i in 0..5 {
            old.push(&format!("old {i}"));
        }
        let c = Chunk::parse(&old.since(0, 100).render()).unwrap();
        let (at, restarted) = resume(Cursor::default(), &c);
        assert_eq!((at.next, restarted), (5, false));
        // The VM restarts and logs 9 lines before the next read.
        let new = BootLog::new(100);
        for i in 0..9 {
            new.push(&format!("new {i}"));
        }
        let c = Chunk::parse(&new.since(at.next, 100).render()).unwrap();
        let (again, restarted) = resume(at, &c);
        assert!(restarted, "the new boot is noticed although it holds more lines than the cursor");
        assert_eq!(again.next, 0, "and read from its first line");
        let c = Chunk::parse(&new.since(again.next, 100).render()).unwrap();
        assert_eq!(c.lines.first().map(String::as_str), Some("new 0"));
        assert_eq!(resume(again, &c), (Cursor { boot: again.boot, next: 9 }, false));
        // A box that names no boot (an older build): a shorter log is still a restart.
        let legacy = Chunk { dropped: 0, lines: vec![], next: 2, boot: 0 };
        assert!(resume(Cursor { boot: 0, next: 5 }, &legacy).1);
    }

    #[test]
    fn a_reader_gets_every_line_once_and_continues_where_it_left_off() {
        let log = BootLog::new(10);
        log.push("one\ntwo\r\n\nthree");
        let c = log.since(0, 2);
        assert_eq!((c.lines.clone(), c.next, c.dropped), (vec!["one".to_string(), "two".to_string()], 2, 0));
        let c = log.since(c.next, 100);
        assert_eq!((c.lines, c.next), (vec!["three".to_string()], 3));
        assert_eq!(log.since(3, 100).lines.len(), 0, "nothing new");
        assert_eq!(log.since(99, 100).next, 3, "a reader ahead of the log is brought back to its end");
    }

    #[test]
    fn a_full_log_drops_the_oldest_and_counts_them_and_the_reader_is_told() {
        let log = BootLog::new(3);
        for i in 0..8 {
            log.push(&format!("line {i}"));
        }
        assert_eq!((log.total(), log.dropped()), (8, 5));
        let c = log.since(0, 100);
        assert_eq!(c.dropped, 5);
        assert_eq!(c.lines, vec!["line 5", "line 6", "line 7"]);
        assert_eq!(c.stored()[0], dropped_marker(5));
        assert!(c.stored()[0].contains("5 lines dropped"));
        let later = log.since(6, 100);
        assert_eq!((later.dropped, later.lines.len()), (0, 2), "a reader that kept up loses nothing");
    }

    #[test]
    fn the_tap_streams_whole_lines_to_the_log_and_passes_every_byte_on() {
        use std::io::Write as _;
        static LOG: BootLog = BootLog::new(100);
        let mut out = Vec::new();
        {
            let mut tap = Tap::new(&LOG, &mut out);
            write!(tap, "[INFO holger] pairing: vm2 PA").unwrap();
            assert_eq!(LOG.total(), 0, "a partial line waits for its newline");
            writeln!(tap, "IRED\n[WARN holger] second").unwrap();
        }
        assert_eq!(LOG.since(0, 10).lines, vec!["[INFO holger] pairing: vm2 PAIRED", "[WARN holger] second"]);
        assert_eq!(String::from_utf8(out).unwrap(), "[INFO holger] pairing: vm2 PAIRED\n[WARN holger] second\n", "the console gets every byte");
        let esp = std::env::temp_dir().join(format!("bootlog-esp-{}.txt", std::process::id()));
        std::fs::write(&esp, "korp-installer: INSTALL-OK\nkorp-start: exec holger\n").unwrap();
        let log = BootLog::new(100);
        assert!(log.capture_esp(&esp, "holger"));
        let lines = log.since(0, 10).lines;
        assert!(lines[0].contains("BOOTLOG") || lines[0].contains("on the ESP"), "{lines:?}");
        assert_eq!((lines[1].as_str(), lines.last().unwrap().as_str()), ("korp-installer: INSTALL-OK", "── holger ──"));
        assert!(!log.capture_esp(std::path::Path::new("/nonexistent/BOOTLOG.TXT"), "holger"));
        let _ = std::fs::remove_file(&esp);
    }

    #[test]
    fn a_long_line_is_cut_on_a_character_boundary_and_the_wire_round_trips() {
        let log = BootLog::new(4);
        log.push(&"å".repeat(MAX_LINE));
        let c = log.since(0, 10);
        assert!(c.lines[0].len() < MAX_LINE + 40 && c.lines[0].ends_with("(cut at 4096 bytes)"));
        let wire = c.render();
        assert_eq!(Chunk::parse(&wire).unwrap(), c);
        assert!(Chunk::parse("garbage").is_err());
    }

    /// Drain a log the way the installer does: from 0, `next` after `next`,
    /// the dropped marker stored where the lines went.
    fn read_all(log: &BootLog, max: usize) -> Vec<String> {
        let mut out = Vec::new();
        let mut from = 0;
        loop {
            let c = log.since(from, max);
            if c.lines.is_empty() && c.dropped == 0 {
                return out;
            }
            out.extend(c.stored());
            from = c.next;
        }
    }

    #[test]
    fn the_pinned_head_outlives_any_flood_and_the_reader_sees_head_marker_tail() {
        let log = BootLog::with_head(50, 10_000, 5, 1_000);
        log.push("install: laid\npairing: admitted as main\npairing: PAIRED");
        log.seal_head();
        for i in 0..10_000 {
            log.push(&format!("ssh.auth.accepted fingerprint=SHA256:{i:043} publickey accepted"));
        }
        assert_eq!(log.head_len(), 3);
        let all = read_all(&log, 7);
        assert_eq!(&all[..3], &["install: laid", "pairing: admitted as main", "pairing: PAIRED"]);
        assert!(all[3].contains("lines dropped"), "{:?}", all[3]);
        assert!(!all.iter().any(|l| l.contains("SHA256:0")), "no fingerprint is held");
        assert!(all.last().unwrap().contains("ssh.auth.accepted fingerprint=") && all.last().unwrap().contains(redact::VALUE_WITHHELD), "{:?}", all.last());
        assert_eq!(log.total(), 10_003);
        assert_eq!(log.dropped() + (all.len() as u64 - 4) + 3, 10_003, "every line is held or counted");
    }

    /// ★ Gunnar f21d8d09 (2026-09-28), shared (lane G10): **an installed process's log
    /// keeps its install for life.** `BootLog::new(DEFAULT_CAP)` has no head, so a box that
    /// talks for hours lost its pairing lines (test2: `… 2243 lines dropped`; test3's front
    /// lost its boot lines on 2026-10-20). [`BootLog::for_install`] is Gunnar's `boot_log()`:
    /// the head and the tail at every default bound, `const`, so it lives in a `static`.
    #[test]
    fn the_install_log_keeps_its_head_for_life() {
        static LOG: BootLog = BootLog::for_install();
        LOG.push("install: laid\npairing: PAIRED");
        LOG.seal_head();
        for i in 0..(DEFAULT_CAP + 50) {
            LOG.push(&format!("runtime line {i}"));
        }
        assert_eq!(LOG.head_len(), 2);
        let c = LOG.since(0, 3);
        assert_eq!(&c.lines[..2], &["install: laid", "pairing: PAIRED"]);
        assert!(c.dropped == 0 && LOG.dropped() == 50, "the tail evicts, the head stays: {}", LOG.dropped());
        // Unsealed, the head stops at its own bound and the rest is the tail.
        let log = BootLog::for_install();
        for i in 0..(HEAD_CAP + 10) {
            log.push(&format!("l{i}"));
        }
        assert_eq!(log.head_len(), HEAD_CAP as u64);
        assert!(log.held_bytes() <= HEAD_BYTES + DEFAULT_BYTES);
    }

    #[test]
    fn the_byte_bound_holds_head_and_tail_and_a_full_head_seals_itself() {
        let (cap, max_bytes, head_cap, head_bytes) = (1_000_000, 8_192, 1_000, 2_048);
        let log = BootLog::with_head(cap, max_bytes, head_cap, head_bytes);
        let line = "x ".repeat(50);
        for _ in 0..5_000 {
            log.push(&line);
            assert!(log.held_bytes() <= max_bytes + head_bytes, "{}", log.held_bytes());
        }
        assert_eq!(log.head_len(), 20, "the head took 2 048 bytes' worth and sealed itself");
        // Lines over MAX_LINE are cut, and the bound still holds.
        log.push(&"y ".repeat(3 * MAX_LINE));
        assert!(log.held_bytes() <= max_bytes + head_bytes);
        let plain = BootLog::new(DEFAULT_CAP);
        for _ in 0..DEFAULT_CAP {
            plain.push(&"z ".repeat(MAX_LINE));
        }
        assert!(plain.held_bytes() <= DEFAULT_BYTES, "{}", plain.held_bytes());
        assert_eq!(plain.head_len(), 0);
    }

    #[test]
    fn a_head_read_with_no_gap_runs_straight_on_into_the_tail() {
        let log = BootLog::with_head(10, 10_000, 2, 1_000);
        log.push("h0\nh1\nt0\nt1");
        let c = log.since(0, 100);
        assert_eq!((c.dropped, c.lines.len(), c.next), (0, 4, 4));
        let c = log.since(1, 2);
        assert_eq!((c.lines.clone(), c.next), (vec!["h1".to_string(), "t0".to_string()], 3));
    }

    #[test]
    fn a_run_of_withheld_lines_is_one_counted_marker_and_a_read_marker_is_never_rewritten() {
        let log = BootLog::new(100);
        log.push("boot");
        for _ in 0..1_000 {
            log.push("reading GUNNAR_UI_ADMIN_TOKEN=a1 from the env");
        }
        log.push("served");
        let c = log.since(0, 100);
        assert_eq!(c.lines.len(), 3, "{:?}", c.lines);
        assert_eq!(c.lines[1], redact::withheld_run("token", 1_000, 1_000 * 45));
        assert!(!c.lines[1].contains("GUNNAR_UI_ADMIN"));
        assert_eq!(log.withheld(), 1_000);
        // A marker a reader has seen stays as it was shown; the run goes on
        // in a new marker.
        log.push("reading GUNNAR_UI_ADMIN_TOKEN=a1 from the env");
        let seen = log.since(0, 100);
        log.push("reading GUNNAR_UI_ADMIN_TOKEN=a1 from the env");
        let again = log.since(0, 100);
        assert_eq!(&again.lines[..seen.lines.len()], &seen.lines[..], "nothing a reader was handed changed");
        assert_eq!(again.lines.len(), 5);
        // Another class starts its own marker.
        log.push("reading /etc/monetize/monetize.env");
        assert!(log.since(5, 10).lines[0].contains("env-class"));
    }

    fn scratch(name: &str) -> std::path::PathBuf {
        let d = std::env::temp_dir().join(format!("bootlog-keep-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&d);
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// **RUNTIME LOGS gap 3 (2026-09-30): a box keeps its own log on its disk.**
    /// vm1 has no narrator and no data volume: its log lived in RAM only, so a
    /// crash left no trace. The box writes every held line to a BOUNDED file
    /// (two files of at most `max` bytes: the current and `<path>.1`), and the
    /// next boot carries the previous boot's LAST lines into its log — marked
    /// `(previous boot) `, so the installer's log door serves them and no gate
    /// judges yesterday's lines as today's.
    #[test]
    fn a_box_keeps_its_log_on_disk_bounded_and_the_next_boot_carries_its_last_lines() {
        let d = scratch("carry");
        let file = d.join("log").join("door.log");
        // Boot 1: lines kept, the file bounded (rotated to .1 past `max`).
        let boot1 = BootLog::new(1000);
        boot1.push("2026-09-30T10:00:00.000000Z  INFO gunnar_front: the front starts");
        let mut at = boot1.keep_once(&file, 0, 200);
        for i in 0..20 {
            boot1.push(&format!("2026-09-30T10:00:{i:02}.000000Z  INFO gunnar_front::web: GET /logproof-{i} 200"));
            at = boot1.keep_once(&file, at, 200);
        }
        assert_eq!(at, 21, "every line written once");
        let cur = std::fs::read_to_string(&file).unwrap();
        let old = std::fs::read_to_string(rotated(&file)).unwrap();
        assert!(cur.len() as u64 <= 200 && old.len() as u64 <= 200, "bounded: {} + {}", cur.len(), old.len());
        assert!(cur.contains("/logproof-19 200"), "the newest line is on disk at once: {cur}");
        // A crash: nothing more. Boot 2 carries the last lines, marked.
        let boot2 = BootLog::new(1000);
        boot2.push("── BOOTLOG.TXT on the ESP ──");
        let carried = boot2.carry_previous(&file, 3, 1 << 20);
        assert_eq!(carried, 3);
        let lines = boot2.since(0, 100).lines;
        assert!(lines[1].contains("the previous boot's last 3 line(s)") && lines[1].contains("door.log"), "{lines:?}");
        assert_eq!(lines[2], "(previous boot) 2026-09-30T10:00:17.000000Z  INFO gunnar_front::web: GET /logproof-17 200");
        assert_eq!(lines[4], "(previous boot) 2026-09-30T10:00:19.000000Z  INFO gunnar_front::web: GET /logproof-19 200");
        assert!(lines[5].contains("end of the previous boot"), "{lines:?}");
        // The previous boot's file is now `.1`; this boot writes a fresh one
        // from where it starts keeping (the carried lines are not kept again).
        assert!(!file.exists() || std::fs::read_to_string(&file).unwrap().is_empty());
        assert!(std::fs::read_to_string(rotated(&file)).unwrap().contains("/logproof-19"));
        let from = boot2.total();
        boot2.push("2026-09-30T10:05:00.000000Z  INFO gunnar_front: the front starts");
        assert_eq!(boot2.keep_once(&file, from, 200), from + 1);
        assert_eq!(std::fs::read_to_string(&file).unwrap(), "2026-09-30T10:05:00.000000Z  INFO gunnar_front: the front starts\n");
        // Nothing kept before: nothing carried, nothing said.
        let fresh = BootLog::new(10);
        assert_eq!(fresh.carry_previous(&d.join("none").join("door.log"), 3, 1 << 20), 0);
        assert_eq!(fresh.total(), 0);
        // The bytes bound on the carry holds too.
        assert_eq!(last_lines("", "aaaa\nbbbb\ncccc\n", 10, 10), vec!["bbbb".to_string(), "cccc".to_string()], "5 bytes a line (with its newline): two fit in 10");
        assert_eq!(last_lines("a\nb\n", "c\n", 2, 100), vec!["b".to_string(), "c".to_string()], "the older file first, then the newer");
        let _ = std::fs::remove_dir_all(&d);
    }
}