gunnar_server/

twin.rs

//! **The twin**: the box that holds every transaction the primary was told, in
//! order, durably, and is never promoted.
//! `.nornir/twin-transaction-log.md` §"The twin's loop".
//!
//! **The twin only stores** (OWNER RULING 2026-09-22: "the twin shouldnt do
//! indexing or processing, only receive and save data … if the main is
//! recharged from the twin it could do the indexing then, at recharge time").
//! A twin appends every frame to its own log — the primary's numbers, the raw
//! pack bytes as pushed — and applies ONLY the control plane it itself runs on
//! ([`twin_applies`]: the KEK, control blobs and documents, sealed secrets —
//! the identity table that gates who may read it, and its host key). Packs,
//! refs, LFS, repository stores, append-only lines and framed files stay in
//! the log as bytes: no pack index, no oid tail, no commit graph, no explode
//! cache, no LFS store is ever built on a twin. They are built on the NEW main
//! when it is recharged from the twin ([`restore`], `gunnar control twin
//! replay`), by the same [`apply`] a primary's restore always ran.
//!
//! Three things live here:
//!
//! * [`stream_range`] — the primary's half of `Mirror.LogFetch`: entries
//!   `from..=to`, bounded by bytes, never fewer than one, each read off the
//!   segment in pieces and handed on as it is read (twin-streaming,
//!   2026-10-06); [`fetch_range`] is the same range gathered, for callers that
//!   want whole entries.
//! * **One entry is never held whole on either box.** The batch budget
//!   ([`DEFAULT_FETCH_BYTES`]) bounds how many entries one fetch carries, never
//!   the size of one: a 512 MiB pack used to be read whole on the primary, cut
//!   into a second whole copy of 1 MiB messages, and glued back into one `Vec`
//!   on the twin before it was appended. Now the primary reads it a piece at a
//!   time ([`Log::read_pieces`], the payload crc checked before the LAST piece
//!   leaves), and the twin writes each piece into its own log as it arrives
//!   ([`FrameSink`] into a [`txlog::Pending`]): the frame exists only once its
//!   last byte is in, and a cut, short or refused transfer leaves nothing.
//! * [`Follower`] — the loop over a [`LogSource`]: head, the gap rule, fetch,
//!   `append_at` into this box's OWN log (the primary's numbers, durable) and
//!   only then [`take`]: on a twin that keeps the entry (store-only), on a box
//!   being restored it is the full [`apply`]. A restart resumes at the own
//!   head; every apply is idempotent or a replayed CAS, so re-applying the
//!   tail after a crash is the same as applying it.
//! * [`restore`] — replay a twin's log into a fresh data root: the recharge,
//!   where every derived index is built.
//!
//! **The gap rule is the loop's shape.** The twin never appends `seq` unless
//! it holds `seq - 1`; `Log::append_at` refuses anything else. When the
//! primary no longer retains `own_head + 1`, the twin stops and says
//! `snapshot-needed` — a twin that kept going past a hole would be a copy that
//! looks complete and is not.
//!
//! **Through the unlogged stores.** The twin's `ServerStores` carry the same
//! tee a primary's do; applying through it would write a SECOND log with the
//! twin's own numbers. So `apply` goes through `raw_control_store`,
//! `raw_secrets` and `OpenRepo::unlogged`, and the twin's `<data-root>/txlog`
//! is exactly the primary's log, number for number.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use anyhow::{anyhow, Context};
use gunnar_core::control::BlobId;
use gunnar_core::txlog::{self, Entry, Kind, Log, Meta, RefEdit};
use gunnar_core::RepoRef;
use gunnar_store::git_storage_trait::{RefCas, RefRejection, RefUpdate};

use crate::stores::ServerStores;

/// `Mirror.LogFetch`'s ceiling when the request says 0.
pub const DEFAULT_FETCH_BYTES: u64 = 64 << 20;

/// **Entries one [`fetch_range`] may answer with, whatever the byte ceiling buys.**
/// The byte ceiling alone is not a bound on the WORK: a range of tiny frames costs one
/// `log.read` and one `Vec` slot each on the primary, and the gRPC layer turns each into
/// at least one `LogChunk`. The twin's loop asks again the moment it is level, so a cap
/// here costs a round trip and nothing else.
pub const MAX_FETCH_ENTRIES: u64 = 10_000;

/// **A twin's group commit** (perf twin-apply-fsync, 2026-10-06): at most this many entries are
/// written into its log between two `fdatasync`s. MEASURED before (oden, `tests/twin_apply_io.rs`):
/// 9.6 writes and 1.85 fsyncs per entry — one fsync per frame in the log and one per redb commit
/// — which on the UpCloud mockup's HDD-tier twin was ~2,800 write IOPS and a lag of ~93,000
/// entries under a storm of small pushes.
pub const GROUP_ENTRIES: usize = 2048;
/// …or this many bytes of frame.
pub const GROUP_BYTES: u64 = 32 << 20;
/// …or this long since the group's first entry: what a twin acknowledges (its durable head) is
/// never older than this while entries keep arriving.
pub const GROUP_AGE: Duration = Duration::from_millis(500);
/// The applied marker is moved at least this often even over a run with no control-plane write,
/// so a restart re-takes at most this many entries.
pub const MARK_EVERY: u64 = 4096;

/// What `Mirror.LogHead` says about a twin.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TwinStatus {
    /// The twin's own head — every entry up to here is in its log AND its
    /// stores.
    pub applied_seq: u64,
    /// The primary's head as of the last successful `LogHead`.
    pub primary_head_seq: u64,
    /// One word.
    pub state: State,
    /// Why, when not `Following`.
    pub why: String,
    /// Wall clock of the last `LogHead` the primary answered; 0 = never.
    pub last_answer_unix_ms: i64,
    /// **Wall clock of the last answer at which this twin was LEVEL** with its
    /// primary — `following` and `applied >= primary head` (2026-09-21, lane T4,
    /// the restore guard). 0 = never.
    ///
    /// Owned by [`set_status`]: whatever a caller passes is replaced, because
    /// this is a fact about the HISTORY of statuses — carried forward through
    /// every `unreachable`, `stalled` and `diverged` — and not something any
    /// one status knows by itself. Equal to `last_answer_unix_ms` exactly when
    /// the twin was level at its last word from the primary, which is the one
    /// fact a restore from a twin whose primary is dead has to be able to see.
    pub last_level_unix_ms: i64,
}

impl TwinStatus {
    /// The watermark: everything the primary has, the twin has applied.
    pub fn caught_up(&self) -> bool {
        self.state == State::Following && self.applied_seq == self.primary_head_seq
    }
}

/// The twin's state word.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
    /// Polling, fetching, applying.
    Following,
    /// The primary no longer retains what the twin needs next.
    SnapshotNeeded,
    /// The last poll could not reach the primary.
    Unreachable,
    /// **The poll got as far as the data and then failed** — the fetch died
    /// mid-range, the append would not go on the volume, an entry would not
    /// apply. The primary is not the problem; this copy stops where it stopped.
    Stalled,
    /// **The box at the other end is not the log this twin has been following.**
    /// Its head is BELOW what this twin has already applied, and a log's head
    /// only ever goes forward — so it is a re-imaged primary, a restore from
    /// somewhere else, or an address that now points at a different set.
    /// Stopped: an operator decides.
    Diverged,
    /// **This twin is growing its own data set** (T21, 2026-09-22, owner ruling:
    /// the twin grows EXACTLY like main — no restart). The grow machine has
    /// drained it: the follower applies nothing and the set may be unmounted.
    /// It still asks its primary for the head, so both ends see the word and
    /// the lag; it follows again the moment the set is mounted back.
    Maintenance,
    /// ★ **This twin is owed a growth before it may follow** (cov-13, 2026-10-06):
    /// main's data set grew while it was silent, and its volume cannot hold it —
    /// the primary refuses its fetches by name (`gunnar_grpc::TWIN_TOO_SMALL`)
    /// rather than let it fill part-way through main's data. It keeps polling
    /// (so it reads heard, and monetize's twin lane grows it) and follows the
    /// moment its report holds main's set. Nothing is appended meanwhile.
    Recovering,
    /// ★ **This twin is taking a full copy (a snapshot) from its primary** (twin-resync-space,
    /// 2026-10-06): it fell behind a hole main's retention trimmed. Its old copy is kept while
    /// the volume holds both; when it does not, the twin GIVES IT UP (`snapshot::DROPPED`) and the
    /// `why` says it holds NO copy — it serves nothing then, and a crash restarts the resync.
    Resyncing,
}

impl std::fmt::Display for State {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            State::Following => "following",
            State::SnapshotNeeded => "snapshot-needed",
            State::Unreachable => "unreachable",
            State::Stalled => "stalled",
            State::Diverged => "diverged",
            State::Maintenance => "maintenance",
            State::Recovering => "recovering",
            State::Resyncing => "resyncing",
        })
    }
}

/// **The twin status, per DATA ROOT** (T17, 2026-09-21). It was one process-wide slot: a box
/// runs one follower, so one slot was enough in production — but every test that ran a
/// follower wrote the same slot, and `set_status` carries `last_level_unix_ms` forward from
/// whatever was there before. Tests in parallel read each other's state and inherited each
/// other's level time (T15: `a_twin_keeps_the_primarys_last_head_and_time_when_the_primary_stops_answering`
/// failed once in a full run, 5/5 alone). Keyed by the root the follower's stores were opened
/// at, each box — and each test — has its own.
static STATUS: OnceLock<Mutex<HashMap<PathBuf, TwinStatus>>> = OnceLock::new();

fn status_slots() -> std::sync::MutexGuard<'static, HashMap<PathBuf, TwinStatus>> {
    STATUS
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .unwrap_or_else(|p| p.into_inner())
}

/// The twin status of the box whose data root is `root`, once its follower has polled.
pub fn status(root: &Path) -> Option<TwinStatus> {
    status_slots().get(root).cloned()
}

/// Record a status, carrying [`TwinStatus::last_level_unix_ms`] forward: it moves
/// only on a `following` answer at which the twin was level, and survives every
/// other state untouched.
pub(crate) fn set_status(root: &Path, mut s: TwinStatus) {
    let mut slots = status_slots();
    let before = slots.get(root).map(|p| p.last_level_unix_ms).unwrap_or(0);
    s.last_level_unix_ms = if s.state == State::Following
        && s.last_answer_unix_ms > 0
        && s.applied_seq >= s.primary_head_seq
    {
        s.last_answer_unix_ms
    } else {
        before
    };
    slots.insert(root.to_path_buf(), s);
}

// ── the primary's half ──────────────────────────────────────────────────────

/// Entries `from..=to` (`to == 0` = the head now), whole, stopping after the
/// entry that crosses `max_bytes` of TRANSACTION (`0` = [`DEFAULT_FETCH_BYTES`])
/// — never fewer than one. `from` below the oldest retained sequence is
/// [`txlog::Error::NotRetained`]; above the head, [`txlog::Error::BeyondHead`].
///
/// **The ceiling counts meta as well as blob, and the count of entries is capped
/// too** (2026-09-16, lane AUDIT-TWIN). It used to count `blob` alone, and most of
/// a forge's frames carry no blob at all: every `Refs`, `ControlDoc`, `LfsDelete`
/// and `RepoStore` is meta only. `bytes >= cap` was then never reached over such a
/// range, so this read the WHOLE range into one `Vec<Entry>` however small a ceiling
/// the caller asked for — and `service::mirror::chunks_of` built a second whole copy
/// of it beside this one. A twin catching up over a ref-heavy history, or any member
/// of group `twin` asking `LogFetch{from: 1, to: 0, max_bytes: 1}`, was an
/// out-of-memory on the PRIMARY. A `Refs` meta IS the transaction when the blob is
/// empty, so charging for it is what the word "bounded" was always supposed to mean.
pub fn fetch_range(log: &Log, from: u64, to: u64, max_bytes: u64) -> txlog::Result<Vec<Entry>> {
    let head = log.head_seq();
    let to = if to == 0 { head } else { to.min(head) };
    let cap = if max_bytes == 0 {
        DEFAULT_FETCH_BYTES
    } else {
        max_bytes
    };
    let mut out = Vec::new();
    let mut bytes = 0u64;
    let mut seq = from;
    while seq <= to {
        let e = log.read(seq)?;
        bytes += (e.blob.len() + e.meta.len()) as u64;
        out.push(e);
        seq += 1;
        if bytes >= cap || out.len() as u64 >= MAX_FETCH_ENTRIES {
            break;
        }
    }
    if out.is_empty() && from > head {
        // Nothing to give, and not because of retention: say so by the
        // same word the reader would.
        return Err(txlog::Error::BeyondHead { seq: from, head });
    }
    Ok(out)
}

/// **The most blob bytes one piece of a streamed entry carries** — the wire's `LogChunk` size,
/// and all a twin or a primary holds of an entry at once.
pub const PIECE_BYTES: usize = 1 << 20;

/// **The refusals [`fetch_range`] makes before it reads anything**, made up front: `from` above
/// the head is [`txlog::Error::BeyondHead`]; a non-empty range starting below the oldest
/// retained sequence is [`txlog::Error::NotRetained`]. What a streamed `Mirror.LogFetch` answers
/// as the call's status, before its first piece, exactly as the gathered one did.
pub fn check_range(log: &Log, from: u64, to: u64) -> txlog::Result<()> {
    let head = log.head_seq();
    if from > head {
        return Err(txlog::Error::BeyondHead { seq: from, head });
    }
    let to = if to == 0 { head } else { to.min(head) };
    let first = log.first_seq();
    if from <= to && from < first {
        return Err(txlog::Error::NotRetained { seq: from, first });
    }
    Ok(())
}

/// **[`fetch_range`], streamed** (twin-streaming, 2026-10-06): the same entries under the same
/// bounds — stopping after the entry that crosses `max_bytes` of transaction, or at
/// [`MAX_FETCH_ENTRIES`], never fewer than one — each handed to `into` as it is read off the
/// segment, [`PIECE_BYTES`] at a time ([`Log::read_pieces`]). `into.end()` for an entry is called
/// only after its payload crc checked out, so a reader that makes an entry visible at its end
/// never makes a corrupt one visible. Answers how many entries went out whole.
///
/// What the primary holds is ONE piece, whatever the entry's size; `fetch_range` held the whole
/// range and the gRPC layer a second copy of it.
pub fn stream_range(
    log: &Log,
    from: u64,
    to: u64,
    max_bytes: u64,
    into: &mut dyn FrameSink,
) -> anyhow::Result<u64> {
    stream_range_at(log, from, 0, to, max_bytes, into)
}

/// **[`stream_range`] with the FIRST entry started at byte `from_offset` of its blob**
/// (restore-large-entry, 2026-10-07): a restore resuming an entry the box being filled already
/// holds part of (`LogFetchRequest.from_offset`). Only the blob from the offset is read
/// ([`Log::read_pieces_from`]): a resumed entry of gigabytes costs this box its rest per call,
/// never a re-read of what went out before. Its stored payload crc goes out with the last piece
/// ([`FrameSink::checked`]) and the box that put the entry together checks it — the bytes below
/// the offset are not here to check it over. `into` is told where the entry starts
/// ([`FrameSink::begin_at`]). An offset at the blob's end is one empty last piece. An offset PAST
/// it (or on an erased frame) is not the entry the asker holds part of (one erased since, say):
/// the entry goes out whole from 0, checked, and the asker sees its length differ.
pub fn stream_range_at(
    log: &Log,
    from: u64,
    from_offset: u64,
    to: u64,
    max_bytes: u64,
    into: &mut dyn FrameSink,
) -> anyhow::Result<u64> {
    check_range(log, from, to)?;
    let head = log.head_seq();
    let to = if to == 0 { head } else { to.min(head) };
    let cap = if max_bytes == 0 {
        DEFAULT_FETCH_BYTES
    } else {
        max_bytes
    };
    let (mut bytes, mut sent, mut seq) = (0u64, 0u64, from);
    while seq <= to {
        let mut begun = false;
        let start = if seq == from && from_offset > 0 {
            let l = log.locate_blob(seq)?;
            if l.kind == Kind::Redacted as u16 || from_offset > l.blob_len {
                0
            } else {
                from_offset
            }
        } else {
            0
        };
        let at =
            log.read_pieces_from::<anyhow::Error>(seq, start, PIECE_BYTES, |l, piece, last| {
                if !begun {
                    into.begin_at(l.seq, l.kind, &l.meta, l.blob_len, start)?;
                    begun = true;
                }
                if last {
                    if let Some(crc) = l.payload_crc {
                        into.checked(crc)?;
                    }
                }
                if !piece.is_empty() {
                    into.piece(piece)?;
                }
                if last {
                    into.end()?;
                }
                Ok(())
            })?;
        bytes += at.blob_len + at.meta.len() as u64;
        sent += 1;
        seq += 1;
        if bytes >= cap || sent >= MAX_FETCH_ENTRIES {
            break;
        }
    }
    Ok(sent)
}

// ── streamed entries ────────────────────────────────────────────────────────

/// **Where a streamed range goes: entry by entry, piece by piece** (twin-streaming,
/// 2026-10-06). `begin` names an entry and the length its blob will have, `piece` carries the
/// next bytes of it in order, `end` says its last byte arrived and the source vouched for it.
/// An entry that never reaches `end` — the wire broke, the source refused, a check failed — was
/// never an entry: an implementation keeps nothing of it.
pub trait FrameSink {
    /// An entry begins.
    fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()>;
    /// The next piece of the open entry's blob.
    fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()>;
    /// The open entry is whole.
    fn end(&mut self) -> anyhow::Result<()>;
    /// **An entry begins part-way** — its pieces start at byte `offset` of the blob
    /// (restore-large-entry, 2026-10-07: [`stream_range_at`]). A sink that can only take whole
    /// entries refuses any offset but 0.
    fn begin_at(
        &mut self,
        seq: u64,
        kind: u16,
        meta: &[u8],
        blob_len: u64,
        offset: u64,
    ) -> anyhow::Result<()> {
        anyhow::ensure!(
            offset == 0,
            "seq {seq}: this sink takes whole entries, not one from byte {offset}"
        );
        self.begin(seq, kind, meta, blob_len)
    }
    /// The open entry's payload crc32c (meta ‖ blob), checked by the source against what it
    /// read, before its last piece: a sink that forwards entries vouches with it.
    fn checked(&mut self, _payload_crc: u32) -> anyhow::Result<()> {
        Ok(())
    }
}

/// **Whole entries out of a stream** — [`LogSource::fetch`] spelled over
/// [`LogSource::fetch_into`], for the callers that want them (tests, small ranges). Holds what it
/// gathers, which is the point of using it only where that is small — and
/// BOUNDED (stinky row 20, 2026-10-10): no entry over [`GATHER_MAX_ENTRY`], no
/// more than [`GATHER_MAX_TOTAL`] of blob in all, and no entry past the length
/// it declared. `reserve_for` bounds only the first allocation; these bound
/// what a peer that keeps sending can make this box hold.
#[derive(Debug)]
pub struct Gather {
    /// The entries, in order, once ended.
    pub out: Vec<Entry>,
    open: Option<Entry>,
    /// The open entry's declared blob length.
    declared: u64,
    /// Blob bytes held, ended entries and the open one.
    held: u64,
    max_entry: u64,
    max_total: u64,
}

impl Default for Gather {
    fn default() -> Self {
        Gather::bounded(GATHER_MAX_ENTRY, GATHER_MAX_TOTAL)
    }
}

/// The largest entry a [`Gather`] holds whole.
pub const GATHER_MAX_ENTRY: u64 = 256 << 20;
/// The most blob bytes a [`Gather`] holds in all.
pub const GATHER_MAX_TOTAL: u64 = 512 << 20;

impl Gather {
    /// A gather that holds no entry over `max_entry` bytes and no more than
    /// `max_total` bytes of blob in all.
    pub fn bounded(max_entry: u64, max_total: u64) -> Gather {
        Gather {
            out: Vec::new(),
            open: None,
            declared: 0,
            held: 0,
            max_entry,
            max_total,
        }
    }
}

impl FrameSink for Gather {
    fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
        anyhow::ensure!(self.open.is_none(), "seq {seq} began inside another entry");
        anyhow::ensure!(
            blob_len <= self.max_entry,
            "seq {seq} declares {blob_len} bytes, more than the {} one gathered entry may hold; it is streamed, not gathered",
            self.max_entry
        );
        anyhow::ensure!(
            self.held.saturating_add(blob_len) <= self.max_total,
            "seq {seq} ({blob_len} bytes) would take this gather past the {} bytes it may hold in all ({} held)",
            self.max_total,
            self.held
        );
        self.declared = blob_len;
        self.open = Some(Entry {
            seq,
            kind,
            meta: meta.to_vec(),
            blob: Vec::with_capacity(gunnar_proto::reserve_for(blob_len)),
        });
        Ok(())
    }
    fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
        let open = self
            .open
            .as_mut()
            .ok_or_else(|| anyhow!("a piece arrived outside an entry"))?;
        let after = open.blob.len() as u64 + bytes.len() as u64;
        anyhow::ensure!(
            after <= self.declared,
            "seq {}: {after} bytes arrived for an entry that declared {}",
            open.seq,
            self.declared
        );
        open.blob.extend_from_slice(bytes);
        self.held += bytes.len() as u64;
        Ok(())
    }
    fn end(&mut self) -> anyhow::Result<()> {
        let e = self
            .open
            .take()
            .ok_or_else(|| anyhow!("an entry ended that never began"))?;
        self.out.push(e);
        Ok(())
    }
}

/// **The `LogChunk` protocol, checked, for every reader of `Mirror.LogFetch`** — the twin's
/// follower and `gunnar control twin follow` read the same stream the same way. One entry's
/// pieces arrive in order with the same `seq`, `kind` and `blob_len`, `meta` on the first; its
/// bytes add up to exactly `blob_len` when `last` ends it (more is refused at the piece that
/// overflows, less at the end); a new `seq` before `last` and a stream that ends inside an
/// entry are both refused. Each piece goes to the [`FrameSink`] as it arrives and is dropped.
#[derive(Debug, Default)]
pub struct ChunkFeed {
    /// `(seq, kind, blob_len, bytes so far)` of the entry being received.
    open: Option<(u64, u16, u64, u64)>,
}

impl ChunkFeed {
    /// One chunk off the wire.
    pub fn take(
        &mut self,
        chunk: &gunnar_proto::LogChunk,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        let kind = u16::try_from(chunk.kind)
            .map_err(|_| anyhow!("seq {}: kind {} is not a frame kind", chunk.seq, chunk.kind))?;
        match self.open {
            Some((seq, k, len, _)) if seq == chunk.seq => anyhow::ensure!(
                k == kind && len == chunk.blob_len,
                "seq {seq}: a later piece says kind {kind} of {} bytes, the first said kind {k} of {len}",
                chunk.blob_len
            ),
            Some((seq, ..)) => anyhow::bail!("the primary moved to seq {} with seq {seq} unfinished", chunk.seq),
            None => {
                into.begin(chunk.seq, kind, &chunk.meta, chunk.blob_len)?;
                self.open = Some((chunk.seq, kind, chunk.blob_len, 0));
            }
        }
        let Some((seq, _, len, got)) = self.open.as_mut() else {
            unreachable!("an entry is open")
        };
        *got += chunk.blob.len() as u64;
        anyhow::ensure!(
            *got <= *len,
            "seq {seq}: more than the {len} blob bytes it declared arrived"
        );
        if !chunk.blob.is_empty() {
            into.piece(&chunk.blob)?;
        }
        if chunk.last {
            anyhow::ensure!(*got == *len, "seq {seq}: {got} of {len} blob bytes arrived");
            self.open = None;
            into.end()?;
        }
        Ok(())
    }

    /// The stream ended: refused if it ended inside an entry.
    pub fn finish(&self) -> anyhow::Result<()> {
        match self.open {
            Some((seq, _, len, got)) => Err(anyhow!(
                "the stream ended inside seq {seq}, {got} of {len} blob bytes in"
            )),
            None => Ok(()),
        }
    }
}

/// **A `Mirror.LogFetch` stream into a [`FrameSink`]**, each chunk handed on as it arrives,
/// none gathered: what both `GrpcSource` and `gunnar control twin follow` run (any stream of
/// chunks — tonic's `Streaming` is one). `idle` bounds the wait for every next chunk (the
/// follower's [`FETCH_IDLE_WITHIN`]); `said` turns the stream's error into the caller's.
pub async fn feed_log_stream<S, E>(
    stream: &mut S,
    idle: Option<(Duration, &(dyn Fn() -> anyhow::Error + Sync))>,
    said: &(dyn Fn(E) -> anyhow::Error + Sync),
    into: &mut dyn FrameSink,
) -> anyhow::Result<()>
where
    S: tokio_stream::Stream<Item = Result<gunnar_proto::LogChunk, E>> + Unpin,
{
    use tokio_stream::StreamExt as _;
    let mut feed = ChunkFeed::default();
    loop {
        let next = match idle {
            Some((within, quiet)) => tokio::time::timeout(within, stream.next())
                .await
                .map_err(|_| quiet())?,
            None => stream.next().await,
        };
        match next.transpose().map_err(said)? {
            Some(chunk) => feed.take(&chunk, into)?,
            None => return feed.finish(),
        }
    }
}

// ── the source ──────────────────────────────────────────────────────────────

/// What the twin reads from: `LogHead` and `LogFetch`, over gRPC in the
/// appliance and over a shared `Log` in tests and in `restore`.
pub trait LogSource: Send + Sync {
    /// `(first_seq, head_seq)` on the primary.
    fn head(&self) -> anyhow::Result<(u64, u64)>;
    /// Does the primary have another address to re-dial when this one keeps
    /// failing? (The tunnel's and the utility network's.) In-process sources
    /// have none.
    fn alternatives(&self) -> bool {
        false
    }
    /// **`(first_seq, head_seq, the source's log identity)`** — one call, because
    /// the identity has to come from the SAME answer as the numbers it
    /// qualifies. Asking twice would leave a window in which the numbers are
    /// one box's and the name is another's, which is the bug in a nicer shape.
    ///
    /// The default carries no identity, and that is the truthful answer for
    /// every in-process source: a bare `Arc<Log>`, a `LocalSource` over another
    /// data root in this process, a test's script. Nothing was fetched over a
    /// wire, so there is nothing to have been lied to about. Only
    /// [`GrpcSource`] overrides it. See [`LOG_ID`].
    fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
        let (first, head) = self.head()?;
        Ok((first, head, None))
    }
    /// **[`LogSource::head_and_identity`] and the source's GENERATION**
    /// ([`LOG_GENERATION`]), out of the same answer. 0 = none claimed: every
    /// in-process source, and a primary never restored or older than the field.
    fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
        let (first, head, id) = self.head_and_identity()?;
        Ok((first, head, id, 0))
    }
    /// Entries `from..=to`, at most about `max_bytes` of blob, never fewer
    /// than one when `from <= to`.
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>>;
    /// **[`LogSource::fetch`], streamed into `into`** — what the follower calls (twin-streaming,
    /// 2026-10-06). Every source that reads a real log overrides it ([`GrpcSource`] off the wire,
    /// a bare [`Arc<Log>`] and [`LocalSource`] off the segments), so an entry is never whole in
    /// memory however large it is. The default gathers through `fetch` and hands each blob over
    /// as one piece: right for a scripted source, whose entries are already in its hands.
    fn fetch_into(
        &self,
        from: u64,
        to: u64,
        max_bytes: u64,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        for e in self.fetch(from, to, max_bytes)? {
            into.begin(e.seq, e.kind, &e.meta, e.blob.len() as u64)?;
            if !e.blob.is_empty() {
                into.piece(&e.blob)?;
            }
            into.end()?;
        }
        Ok(())
    }
    /// A consistent snapshot of the source's stores, for a twin behind a retention hole, fed
    /// piece by piece `into` the twin's staging area as it arrives. `false` = this source
    /// cannot make one (a bare log): the twin stops with snapshot-needed.
    fn snapshot(&self, _into: &mut snapshot::Stager) -> anyhow::Result<bool> {
        Ok(false)
    }
}

impl LogSource for Arc<Log> {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        Ok((self.first_seq(), self.head_seq()))
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        Ok(fetch_range(self, from, to, max_bytes)?)
    }
    fn fetch_into(
        &self,
        from: u64,
        to: u64,
        max_bytes: u64,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        stream_range(self, from, to, max_bytes, into).map(drop)
    }
}

// ── apply ───────────────────────────────────────────────────────────────────

/// What one entry did on this box.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Applied {
    /// Stored, or already there.
    Done,
    /// A CAS the primary lost, lost here too — the design's replayed race.
    Lost(String),
    /// A kind this build does not know: kept in the log, not applied.
    Unknown(u16),
    /// **Kept in a twin's log and nothing else** (ruling 2026-09-22, "the twin only
    /// stores"): a data frame a twin does not apply. The recharge applies it on the new main.
    Stored,
}

/// **The frames a twin applies** — the control plane the twin itself runs on, and nothing
/// else (ruling 2026-09-22, "the twin only stores").
///
/// * `Kek` — without it the replicated sealed secrets are unreadable on this box.
/// * `ControlBlob` / `ControlDoc` — the identity table: who may read this twin's log
///   (`Mirror.LogFetch`/`SnapshotFetch`, `twin status`, `twin replay`) is decided against the
///   primary's principals and groups, so the twin must hold them to gate its own door.
/// * `Secret` — the host key the twin answers with, and the rest of the sealed rows (a row
///   write, no derived structure).
///
/// Everything else — `Pack`, `Refs`, `LfsPut`, `LfsDelete`, `RepoStore`, `Append`, `File` —
/// is the DATA a twin saves and does not process: the frame in the log IS the copy. Applying
/// a `Pack` is what builds the archive's oid tail, the pack index and (on read) the commit
/// graph and the explode pool; applying `Refs` needs that tail. None of it exists on a twin.
pub fn twin_applies(kind: u16) -> bool {
    // `Erase` too (owner ruling g-nonpay): the twin's log IS the copy, so an account
    // deleted on main is erased from it here — the one thing a twin does to data.
    // `Activity` too (2026-10-21): the last-seen stamps are control-plane rows, and a twin
    // promoted to main answers the admin's user list with them.
    matches!(
        Kind::from_u16(kind),
        Some(
            Kind::Kek
                | Kind::ControlBlob
                | Kind::ControlDoc
                | Kind::Secret
                | Kind::Erase
                | Kind::Activity
        )
    )
}

/// **What a twin applies**: [`twin_applies`]' kinds, and the identity core's
/// document (`identity.json`, a `File` frame): the core answers from those tables,
/// so a twin that only stored them would answer from the ones it loaded at its
/// boot and drift from its main (RA5-10).
pub(crate) fn twin_follows(kind: u16, meta: &[u8]) -> bool {
    twin_applies(kind)
        || (matches!(Kind::from_u16(kind), Some(Kind::File))
            && matches!(Meta::from_bytes(meta), Ok(Meta::File { ref path, .. }) if path == crate::identity::FILE))
}

/// **What a follower does with an entry that is durable in its log.** On a twin: the
/// [`twin_applies`] kinds are applied, every other kind is [`Applied::Stored`] (the log holds
/// it) — an unknown kind is still [`Applied::Unknown`]. On any other box (a fresh box being
/// restored — the recharge): the full [`apply`], which is where the indexes get built.
pub fn take(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
    take_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}

/// [`take`] over borrowed parts.
pub fn take_raw(
    stores: &ServerStores,
    seq: u64,
    kind: u16,
    meta: &[u8],
    blob: &[u8],
) -> anyhow::Result<Applied> {
    if stores.is_twin() && Kind::from_u16(kind).is_some() && !twin_follows(kind, meta) {
        // The meta is parsed — a frame that could never be applied is refused NOW, while the
        // primary is alive to be asked again, not discovered at the recharge. The bytes are
        // not touched.
        Meta::from_bytes(meta)
            .with_context(|| format!("seq {seq}: the frame's meta does not parse"))?;
        return Ok(Applied::Stored);
    }
    apply_raw(stores, seq, kind, meta, blob)
}

/// **[`take`] for an entry already durable in this box's log**, reading its blob back only when
/// the take needs it (twin-streaming, 2026-10-06). A twin keeps its data frames without touching
/// their bytes ([`take_raw`] parses the meta and stops), so a 512 MiB pack it just streamed to
/// disk is never read back into memory; the control plane it applies is small, and a box being
/// restored reads each frame whole because [`apply`] needs it whole.
pub fn take_logged(
    stores: &ServerStores,
    seq: u64,
    kind: u16,
    meta: &[u8],
) -> anyhow::Result<Applied> {
    if stores.is_twin() && !twin_follows(kind, meta) {
        return take_raw(stores, seq, kind, meta, &[]);
    }
    let e = stores.txlog().read(seq)?;
    take(stores, &e)
}

fn oid_bytes(hex_text: &str) -> anyhow::Result<Vec<u8>> {
    hex::decode(hex_text).map_err(|e| anyhow!("oid {hex_text}: {e}"))
}

fn repo_of(stores: &ServerStores, key: &str) -> anyhow::Result<gunnar_core::Repo> {
    let (ns, name) = key
        .split_once('/')
        .ok_or_else(|| anyhow!("repository key {key:?} is not namespace/name"))?;
    let path = RepoRef::new(ns, name).with_context(|| format!("repository key {key:?}"))?;
    stores
        .catalog()
        .get(&path)?
        .ok_or_else(|| anyhow!("the frame names {key}, which the catalog does not hold yet — the catalog row must precede the first pack"))
}

/// Apply one entry to `stores` through their UNLOGGED halves.
pub fn apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
    apply_raw(stores, e.seq, e.kind, &e.meta, &e.blob)
}

/// [`apply`] over borrowed parts — a snapshot item read back from the twin's staging file
/// (`seq` 0) is never an owned [`Entry`].
pub fn apply_raw(
    stores: &ServerStores,
    seq: u64,
    kind: u16,
    meta: &[u8],
    blob: &[u8],
) -> anyhow::Result<Applied> {
    struct E<'a> {
        seq: u64,
        blob: &'a [u8],
    }
    let e = E { seq, blob };
    let Some(kind) = Kind::from_u16(kind) else {
        // **A FOLLOWING twin keeps it and goes on; a box being RESTORED stops here**
        // (2026-09-16, lane AUDIT-TWIN). The two are not the same box:
        //
        // * a twin's log holds the frame byte for byte and the twin is never promoted, so a
        //   kind a newer primary invented costs it nothing — the restore that later reads this
        //   log gets the frame whole;
        // * a box being restored stops being a follower at [`restore_finish`] and starts
        //   ORIGINATING from that head. Skipping the frame there was SILENT: the replay
        //   answered "restored to seq N (N applied, 0 lost)" and the box came up as a primary
        //   whose stores were missing whatever those frames carried. Measured 2026-09-16: a log
        //   with one kind-99 frame restored green.
        //
        // Refusing it as an apply failure is the resumable shape the restore already has: the
        // entry is durable at the head, the stream stops, the box stays restoring, and
        // [`restore_begin`] re-applies the head — so re-running the replay under a build that
        // knows the kind finishes it instead of stepping over it.
        if is_restoring(stores.root()) {
            anyhow::bail!(
                "seq {seq}: frame kind {kind} is not one this build can apply — the log that is being \
                 restored was written by a NEWER gunnar. The frame is durable at this box's head and \
                 nothing is lost; restore again from a build that knows kind {kind}. This box stays \
                 restoring and originates nothing."
            );
        }
        return Ok(Applied::Unknown(kind));
    };
    // An erased frame's data is what was deleted: nothing to apply. Its meta is the
    // original frame's, so it is not matched below.
    if kind == Kind::Redacted {
        return Ok(Applied::Done);
    }
    let meta = Meta::from_bytes(meta).with_context(|| format!("seq {seq}"))?;
    match (kind, meta) {
        // **An account's data was deleted** (g-nonpay): every frame of these repositories
        // before this one is erased from THIS box's log — the twin's copy, or the log a
        // restore is rebuilding. The stores themselves went with their `RepoStore` frames.
        (Kind::Erase, Meta::Erase { repos, .. }) => {
            stores.txlog().erase(seq, &repos).with_context(|| {
                format!(
                    "seq {seq}: erasing {} repositor(ies) from the log",
                    repos.len()
                )
            })?;
            Ok(Applied::Done)
        }
        (Kind::Pack, Meta::Pack { repo, .. }) => {
            let r = repo_of(stores, &repo)?;
            let store = stores.repo_store(&r)?;
            store
                .unlogged()
                .put_pack(e.blob)
                .with_context(|| format!("seq {}: pack into {repo}", e.seq))?;
            Ok(Applied::Done)
        }
        (
            Kind::Refs,
            Meta::Refs {
                repo,
                atomic,
                edits,
            },
        ) => {
            let r = repo_of(stores, &repo)?;
            let store = stores.repo_store(&r)?;
            let git = store.unlogged();
            if atomic {
                let olds: Vec<Option<Vec<u8>>> = edits
                    .iter()
                    .map(|ed| ed.old.clone().flatten().map(|h| oid_bytes(&h)).transpose())
                    .collect::<anyhow::Result<_>>()?;
                let news: Vec<Option<Vec<u8>>> = edits
                    .iter()
                    .map(|ed| ed.new.as_deref().map(oid_bytes).transpose())
                    .collect::<anyhow::Result<_>>()?;
                let peeled: Vec<Option<Vec<u8>>> = edits
                    .iter()
                    .map(|ed| ed.peeled.as_deref().map(oid_bytes).transpose())
                    .collect::<anyhow::Result<_>>()?;
                let cas: Vec<RefCas<'_>> = edits
                    .iter()
                    .enumerate()
                    .map(|(i, ed)| RefCas {
                        name: ed.name.clone(),
                        old: olds[i].as_deref(),
                        new: news[i].as_deref(),
                        peeled: peeled[i].clone(),
                    })
                    .collect();
                match git.put_refs_cas(&cas) {
                    Ok(_) => Ok(Applied::Done),
                    Err(err) => match err.downcast_ref::<RefRejection>() {
                        Some(rej) => Ok(Applied::Lost(format!("{rej:?}"))),
                        None => {
                            Err(err).with_context(|| format!("seq {}: refs into {repo}", e.seq))
                        }
                    },
                }
            } else {
                let mut plain: Vec<RefUpdate> = Vec::new();
                for ed in &edits {
                    if ed.name == "HEAD" {
                        if let Some(target) = &ed.symref {
                            git.set_head(target)
                                .with_context(|| format!("seq {}: HEAD of {repo}", e.seq))?;
                            continue;
                        }
                    }
                    plain.push(update_of_edit(ed));
                }
                if !plain.is_empty() {
                    git.put_refs(&plain)
                        .with_context(|| format!("seq {}: refs into {repo}", e.seq))?;
                }
                Ok(Applied::Done)
            }
        }
        (Kind::ControlBlob, Meta::ControlBlob { id }) => {
            let got = stores.raw_control_store().put_blob(e.blob)?;
            let want = BlobId::from_hex(&id)
                .ok_or_else(|| anyhow!("seq {}: blob id {id:?} is not hex", e.seq))?;
            if got != want {
                return Err(anyhow!(
                    "seq {}: the blob hashes to {} but the frame says {id}",
                    e.seq,
                    hex::encode(got.as_bytes())
                ));
            }
            Ok(Applied::Done)
        }
        (
            Kind::ControlDoc,
            Meta::ControlDoc {
                name,
                expected,
                new,
            },
        ) => {
            let mut expected = expected
                .as_deref()
                .map(|h| {
                    BlobId::from_hex(h)
                        .ok_or_else(|| anyhow!("seq {}: expected {h:?} is not hex", e.seq))
                })
                .transpose()?;
            let new = BlobId::from_hex(&new)
                .ok_or_else(|| anyhow!("seq {}: new {new:?} is not hex", e.seq))?;
            if expected.is_none() && fresh_take(stores.root(), &name)? {
                // The fresh-box rule: this box wrote `name` itself before it
                // was reset; the primary's create is a create into empty.
                expected = stores.raw_control_store().read_doc(&name)?;
            }
            if stores
                .raw_control_store()
                .commit_doc(&name, expected, new)?
            {
                Ok(Applied::Done)
            } else {
                Ok(Applied::Lost(format!("document {name}")))
            }
        }
        (Kind::Secret, Meta::Secret { key, deleted }) => {
            use gunnar_store::SecretStore as _;
            if deleted {
                stores.raw_secrets().delete(&key)?;
            } else {
                stores.raw_secrets().put(&key, e.blob)?;
            }
            Ok(Applied::Done)
        }
        (Kind::LfsPut, Meta::Lfs { repo, oid, size }) => {
            use gunnar_lfs::LfsStore as _;
            use std::io::Write as _;
            let r = repo_of(stores, &repo)?;
            let store = stores.repo_store(&r)?;
            let lfs = gunnar_lfs::FsLfsStore::open_or_init(store.git_dir())?;
            let id = gunnar_lfs::Oid::parse(&oid)
                .map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
            if lfs.size_of(&id)?.is_some() {
                return Ok(Applied::Done);
            }
            let mut w = lfs.begin_write(&id, size)?;
            w.write_all(e.blob)?;
            w.commit()?;
            Ok(Applied::Done)
        }
        (Kind::Activity, Meta::Activity {}) => {
            let rows = gunnar_core::activity::decode_batch(e.blob)
                .with_context(|| format!("seq {}: last-seen stamps", e.seq))?;
            stores
                .activity()
                .absorb(rows)
                .with_context(|| format!("seq {}: storing last-seen stamps", e.seq))?;
            Ok(Applied::Done)
        }
        (Kind::Kek, Meta::Kek {}) => {
            write_kek(stores.control_dir(), e.blob)
                .with_context(|| format!("seq {}: the KEK", e.seq))?;
            stores
                .adopt_kek(e.blob)
                .with_context(|| format!("seq {}: holding the KEK", e.seq))?;
            Ok(Applied::Done)
        }
        (Kind::Append, Meta::Append { log }) => {
            let path = under_root(stores.root(), &log)?;
            append_line_once(&path, e.blob)
                .with_context(|| format!("seq {}: a line of {log}", e.seq))?;
            Ok(Applied::Done)
        }
        (Kind::File, Meta::File { path, deleted }) => {
            let full = under_root(stores.root(), &path)?;
            if deleted {
                match std::fs::remove_file(&full) {
                    Ok(()) => {}
                    Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
                    Err(err) => {
                        return Err(err).with_context(|| format!("seq {}: removing {path}", e.seq))
                    }
                }
            } else {
                let dir = full.parent().expect("under the root");
                std::fs::create_dir_all(dir)?;
                let name = full
                    .file_name()
                    .expect("a file")
                    .to_string_lossy()
                    .into_owned();
                crate::durable::write_atomic_unframed(dir, &name, e.blob)
                    .with_context(|| format!("seq {}: {path}", e.seq))?;
            }
            // The identity core holds its tables in memory: the main's document is read
            // again, so the twin's identity state follows the main (RA5-10).
            if path == crate::identity::FILE {
                if let Some(identity) = stores.identity() {
                    crate::identity::follow(stores.root(), &identity)
                        .map_err(|why| anyhow!("seq {}: {why}", e.seq))?;
                }
            }
            Ok(Applied::Done)
        }
        (Kind::LfsDelete, Meta::Lfs { repo, oid, .. }) => {
            use gunnar_lfs::LfsStore as _;
            let (ns, name) = repo.split_once('/').ok_or_else(|| {
                anyhow!(
                    "seq {}: repository key {repo:?} is not namespace/name",
                    e.seq
                )
            })?;
            let dir = stores.repo_dir(&RepoRef::new(ns, name)?)?;
            // No store, no object: the delete is already true here.
            if !dir.exists() {
                return Ok(Applied::Done);
            }
            let id = gunnar_lfs::Oid::parse(&oid)
                .map_err(|err| anyhow!("seq {}: {oid}: {err}", e.seq))?;
            gunnar_lfs::FsLfsStore::open_or_init(&dir)?
                .delete(&id)
                .with_context(|| format!("seq {}: deleting LFS {oid} of {repo}", e.seq))?;
            Ok(Applied::Done)
        }
        (Kind::RepoStore, Meta::RepoStore { repo, op, .. }) => {
            let (ns, name) = repo.split_once('/').ok_or_else(|| {
                anyhow!(
                    "seq {}: repository key {repo:?} is not namespace/name",
                    e.seq
                )
            })?;
            let path = RepoRef::new(ns, name)?;
            match op.as_str() {
                "purged" => {
                    stores
                        .reclaim_repo(&path)
                        .with_context(|| format!("seq {}: purging the store of {repo}", e.seq))?;
                }
                // The store opens on its first pack; a create frame, when one exists, opens it now.
                "created" => {
                    if let Some(r) = stores.catalog().get(&path)? {
                        stores.repo_store(&r)?;
                    }
                }
                other => {
                    return Err(anyhow!(
                        "seq {}: RepoStore op {other:?} is neither created nor purged",
                        e.seq
                    ))
                }
            }
            Ok(Applied::Done)
        }
        (kind, meta) => Err(anyhow!(
            "seq {}: kind {kind:?} with meta {meta:?} do not belong together",
            e.seq
        )),
    }
}

fn update_of_edit(ed: &RefEdit) -> RefUpdate {
    RefUpdate {
        name: ed.name.clone(),
        target: ed.new.clone(),
        peeled: ed.peeled.clone(),
        symref_target: ed.symref.clone(),
    }
}

// ── the box's role and the fresh-box rule ───────────────────────────────────

/// `<data-root>/twin.json`: this box is a twin of `primary`. Per box, never
/// framed (a twin's own facts are not the primary's transactions).
pub const TWIN_CONFIG: &str = "twin.json";
/// `<data-root>/twin.key`: the twin's own ed25519 key, minted at join — the
/// credential it presents to the primary's `Mirror.LogFetch`. Per box, never
/// framed; its public half is enrolled on the primary in group [`TWIN_GROUP`].
pub const TWIN_KEY: &str = "twin.key";
/// The group whose members may read the primary's log without being admins.
pub const TWIN_GROUP: &str = "twin";
/// `<data-root>/twin-owed.json`, on a PRIMARY: what each twin is owed — a full resync
/// after retention trimmed past it while it was silent, a growth after main's set grew
/// while it was silent (`gunnar_grpc::twin_owed`, cov-13). Per box, never framed.
pub const TWIN_OWED_FILE: &str = "twin-owed.json";
/// `<data-root>/txlog-fresh-docs.json`: the documents a fresh box wrote itself
/// before it was reset to take another log's numbers (see [`prepare_replay`]).
pub const FRESH_DOCS: &str = "txlog-fresh-docs.json";
/// `<data-root>/txlog-restoring`: a restore is in progress; the log stays
/// follower-only across a restart until [`restore_finish`].
pub const RESTORING: &str = "txlog-restoring";

/// What `twin.json` says.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TwinConfig {
    /// The primary's control plane, `host:port`.
    pub primary: String,
    /// When this box became a twin.
    pub joined_at_unix_ms: i64,
}

impl TwinConfig {
    /// `Ok(None)` = not a twin. A `twin.json` that does not parse is an error,
    /// never "not a twin": a twin that booted as a primary would frame its own
    /// KEK and fork the numbers.
    pub fn load(root: &Path) -> anyhow::Result<Option<Self>> {
        match std::fs::read(root.join(TWIN_CONFIG)) {
            Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes).with_context(|| {
                format!("{} does not parse", root.join(TWIN_CONFIG).display())
            })?)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => {
                Err(e).with_context(|| format!("reading {}", root.join(TWIN_CONFIG).display()))
            }
        }
    }
    /// Write it (unframed, atomic).
    pub fn save(&self, root: &Path) -> anyhow::Result<()> {
        crate::durable::write_atomic_unframed(root, TWIN_CONFIG, &serde_json::to_vec_pretty(self)?)
    }
}

/// Does `root` belong to a twin? Existence decides; [`TwinConfig::load`] parses.
pub fn is_twin_root(root: &Path) -> bool {
    root.join(TWIN_CONFIG).exists()
}

/// `<data>/pairing/recharge_from`: this box was re-admitted as a recharged
/// main and has not been restored yet — what `Bootstrap.Status.recharge_from`
/// answers, so the restore lane (monetize's) knows the box awaits its twin.
///
/// Declared HERE and re-exported by `pairing`: `stores` reads it in every build, and
/// `pairing` is compiled only with `wg-client` (2026-10-14: `--features
/// server,engine-znippy,metered` did not compile — "cannot find `pairing` in `crate`").
pub const RECHARGE_REL: &str = "pairing/recharge_from";

/// Is a restore in progress under `root`?
pub fn is_restoring(root: &Path) -> bool {
    root.join(RESTORING).exists()
}

// ── the log's identity ──────────────────────────────────────────────────────

/// `<data-root>/txlog-id`: **which log this is.** 128 random bits, hex, minted
/// once at a primary's first open and never derived from anything.
///
/// # Why a log needed a name at all
///
/// Until 2026-09-21 (lane T4) NOTHING on the wire said which log a
/// `Mirror.LogHead` answer was about. `LogHeadResponse` carried nine fields —
/// head, first, bytes, twin, applied, primary head, caught up, state, why —
/// and every one of them is a NUMBER ABOUT a log, not a name FOR one. A twin
/// compared sequence numbers against an address and called that following.
///
/// The lineage check added with `State::Diverged` catches a primary whose head
/// is BELOW the twin's, which is a re-imaged main caught early. It cannot catch
/// the same main once it has written past the twin's head: a fresh install at a
/// familiar address, left alone for an afternoon, reaches seq 20 000 and the
/// twin — sitting at 19 264 of a history that no longer exists anywhere else —
/// SPLICES ONE HISTORY ONTO ANOTHER, cleanly and silently, and the only copy of
/// the real one is gone. That is the single worst thing this repository can do,
/// and no number can catch it. Only a name can.
///
/// # Not the host key, and not anything else already on the box
///
/// The host key is replayed by a restore ON PURPOSE (T3, `T1h host key
/// identical`), and so are the KEK and every document — which is exactly right
/// for a restore and exactly wrong here, because a value that survives a restore
/// cannot by itself distinguish the two boxes we care about. The id is minted
/// from the system CSPRNG and from nothing else, so a fresh box CANNOT produce
/// one it has not been given.
///
/// # A restored main KEEPS the id, and that is the whole design
///
/// The id names the HISTORY, not the box. A restore replays the twin's log into
/// a fresh root at the same sequence numbers, frame for frame; if that is not
/// the same history then the word restore means nothing. So `restore_finish`
/// writes the id that came down the stream, and the twin the new main was built
/// FROM keeps following it without being touched.
///
/// The alternative — a restored main mints a new id — was rejected on what it
/// costs at the one moment that matters. It would put the twin into `Diverged`
/// against the box that had just been built out of it, so the estate would have
/// NO twin in the hour after it has just survived losing its main; and getting
/// one back is not cheap, because `become_twin` takes a FRESH box and the twin
/// is not fresh — it holds the whole history — so the twin would have to be
/// wiped and re-seeded from the new main before the estate was covered again.
/// Paying that, every restore, to detect a case the id is not the mechanism for
/// is the wrong trade.
///
/// **What it therefore does NOT catch, said plainly:** two boxes can hold the
/// same id — the restored main, and the old main if it ever comes back. That
/// split-brain is created by the restore itself, not by this file, and it is
/// the sequence rule that meets it: the old main returns BEHIND the restored
/// one, so its head is below the twin's and [`State::Diverged`] already stops
/// there. An id cannot order two boxes in time; it can only say whether they
/// are the same history, which is the question it is here to answer.
///
/// The old main can also come back AHEAD: it took writes the twin never saw
/// before it went dark, or it keeps taking them after it returns, until its
/// head passes the twin's and every number agrees again. That ordering is what
/// [`LOG_GENERATION`] gives (stinky row 18, 2026-10-10): a restore stamps a newer
/// generation, the twin pins the highest it has followed, and an answer from a
/// lower one is [`State::Diverged`] by name.
pub const LOG_ID: &str = "txlog-id";

/// `<data-root>/txlog-generation`: **which restore of its log this box is**
/// (stinky row 18, 2026-10-10). Absent = 0: never restored.
///
/// `restore_finish` writes the unix ms of the restore, kept strictly above the
/// generation the root held before, so a later restore is always a higher
/// number even across a clock that stepped back. It is time and not a counter
/// carried down the restore stream on purpose: the stream (and the pumps that
/// relay it) stay exactly as they are, and two restores of the same log are
/// still ordered. A primary names it in every `Mirror.LogHead` answer, inside
/// its signature.
///
/// What it fences: the TWIN. An old main that comes back after its replacement
/// was restored carries the lower generation, and the twin refuses it however
/// good its id, key and numbers are. And the OLD MAIN, from the first poll a twin
/// sends it: the twin relays the restored main's signed answer, and the old
/// main fences itself ([`LOG_FENCED`]). Until a twin reaches it, an old main
/// cut off from every twin still takes writes — knowing sooner would take a
/// lease or a quorum, which is a different design.
pub const LOG_GENERATION: &str = "txlog-generation";

/// `<data-root>/twin-primary-generation`: the highest [`LOG_GENERATION`] this
/// twin has followed. Absent = 0. It only ever moves up.
pub const PINNED_GENERATION: &str = "twin-primary-generation";

/// One generation file. NotFound = 0; anything else that cannot be read, or is
/// not a number, is an error by name — a corrupt fence must never read as "no
/// fence", which is exactly the hole it is here to close.
fn read_generation(root: &Path, name: &str) -> anyhow::Result<u64> {
    let path = root.join(name);
    match std::fs::read_to_string(&path) {
        Ok(raw) => raw.trim().parse::<u64>().map_err(|e| {
            anyhow::anyhow!(
                "{} holds {:?}, which is not a generation: {e}",
                path.display(),
                raw.trim()
            )
        }),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
        Err(e) => Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
    }
}

fn forget_generation(root: &Path, name: &str) -> anyhow::Result<()> {
    match std::fs::remove_file(root.join(name)) {
        Ok(()) => crate::durable::sync_dir(root),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => {
            Err(anyhow::Error::from(e).context(format!("removing {}", root.join(name).display())))
        }
    }
}

/// This box's [`LOG_GENERATION`].
pub fn log_generation(root: &Path) -> anyhow::Result<u64> {
    read_generation(root, LOG_GENERATION)
}

/// The generation this twin has pinned ([`PINNED_GENERATION`]).
pub fn pinned_generation(root: &Path) -> anyhow::Result<u64> {
    read_generation(root, PINNED_GENERATION)
}

/// **Stamp a restore**: a generation strictly above the one this root held.
fn stamp_generation(root: &Path) -> anyhow::Result<u64> {
    let before = log_generation(root)?;
    let generation = u64::try_from(now_ms())
        .unwrap_or(0)
        .max(before.saturating_add(1));
    crate::durable::write_atomic_unframed(
        root,
        LOG_GENERATION,
        format!("{generation}\n").as_bytes(),
    )?;
    Ok(generation)
}

/// Pin a primary's generation. Never moves the pin down.
fn pin_generation(root: &Path, generation: u64) -> anyhow::Result<()> {
    if generation <= pinned_generation(root)? {
        return Ok(());
    }
    crate::durable::write_atomic_unframed(
        root,
        PINNED_GENERATION,
        format!("{generation}\n").as_bytes(),
    )
}

/// `<data-root>/txlog-fenced`: **this box is an OLD generation of its log**
/// (stinky row 18, the main's half, 2026-10-10).
///
/// [`LOG_GENERATION`] fences the twin off an old main come back after a
/// restore. This fences the old main itself: the twin relays the restored
/// main's signed answer ([`GENERATION_PROOF`]) on every poll, and a box that
/// finds in it a NEWER generation of its OWN log, signed by its OWN host key, is
/// that old main — nobody else holds the key, and the restore replayed it onto
/// the new main. It writes this file (durably, before anything else), makes its
/// log follower-only, and from then on originates nothing and serves no
/// `Mirror.LogFetch`; [`ServerStores::open`] reads the file at every start, so a
/// restart does not undo it. It still answers `Mirror.LogHead`, with its own
/// (older) generation, so a twin that reaches it says why by name.
///
/// Only an operator removes it, deliberately: the box's writes since the
/// restore belong to no history, and the way back is to retire it or to re-join
/// it as a twin (`become_twin` forgets the fence along with the rest of its
/// own state).
pub const LOG_FENCED: &str = "txlog-fenced";

/// `<data-root>/twin-generation-proof`: the newest generation of its log this
/// twin has followed, **as its main signed it** — the fields of one verified
/// `Mirror.LogHead` answer that its `gunnar-log-head-v2` signature covers, and
/// the key that made it, one per line. Relayed as
/// `LogHeadRequest.generation_proof` on every poll; see [`LOG_FENCED`].
pub const GENERATION_PROOF: &str = "twin-generation-proof";

/// **Is this box fenced off as an old generation of its log?** ([`LOG_FENCED`])
/// A marker that cannot be looked at is a fence: the hole it closes is worse
/// than a box refusing writes it could have taken.
pub fn is_fenced(root: &Path) -> bool {
    match std::fs::symlink_metadata(root.join(LOG_FENCED)) {
        Ok(_) => true,
        Err(e) => e.kind() != std::io::ErrorKind::NotFound,
    }
}

/// What [`LOG_FENCED`] says, for the alarm and the refusals.
pub fn fenced_why(root: &Path) -> Option<String> {
    is_fenced(root).then(|| {
        let said = std::fs::read_to_string(root.join(LOG_FENCED)).unwrap_or_default();
        let said = said.trim();
        if said.is_empty() {
            format!("this box is fenced off as an OLD generation of its log ({LOG_FENCED} is on its data root)")
        } else {
            format!("{said} ({LOG_FENCED} is on its data root)")
        }
    })
}

/// The proof this twin relays on every poll ([`GENERATION_PROOF`]). `None` =
/// it has never followed a restored main. A file that is there and does not
/// read as one is an error by name.
pub fn generation_proof(root: &Path) -> anyhow::Result<Option<gunnar_proto::GenerationProof>> {
    let path = root.join(GENERATION_PROOF);
    let raw = match std::fs::read_to_string(&path) {
        Ok(raw) => raw,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(anyhow::Error::from(e).context(format!("reading {}", path.display()))),
    };
    let lines: Vec<&str> = raw.lines().collect();
    let [answer_key, answer_sig, nonce, log_id, first_seq, head_seq, generation] = lines[..] else {
        anyhow::bail!(
            "{} holds {} lines, not the 7 of one signed answer",
            path.display(),
            lines.len()
        );
    };
    let number = |what: &str, v: &str| {
        v.trim()
            .parse::<u64>()
            .map_err(|e| anyhow::anyhow!("{}: {what} {v:?} is not a number: {e}", path.display()))
    };
    Ok(Some(gunnar_proto::GenerationProof {
        answer_key: answer_key.to_owned(),
        answer_sig: answer_sig.to_owned(),
        nonce: nonce.to_owned(),
        log_id: log_id.to_owned(),
        first_seq: number("first_seq", first_seq)?,
        head_seq: number("head_seq", head_seq)?,
        generation: number("generation", generation)?,
    }))
}

/// **Keep a verified answer as this twin's proof** ([`GENERATION_PROOF`]) — called
/// only after the answer passed [`GrpcSource`]'s signature check against `nonce`.
/// Kept when it is signed, names a generation, and that generation is above the
/// proof already kept; otherwise nothing changes.
pub fn remember_generation_proof(
    root: &Path,
    nonce: &str,
    head: &gunnar_proto::LogHeadResponse,
) -> anyhow::Result<()> {
    if head.generation == 0
        || head.answer_sig.trim().is_empty()
        || head.answer_key.trim().is_empty()
    {
        return Ok(());
    }
    if generation_proof(root)
        .ok()
        .flatten()
        .is_some_and(|kept| kept.generation >= head.generation)
    {
        return Ok(());
    }
    let one_line = |v: &str| !v.contains('\n') && !v.contains('\r');
    anyhow::ensure!(
        [
            head.answer_key.as_str(),
            head.answer_sig.as_str(),
            nonce,
            head.log_id.as_str()
        ]
        .into_iter()
        .all(one_line),
        "a signed answer whose key, signature, nonce or log id spans lines is not kept as a proof"
    );
    let text = format!(
        "{}\n{}\n{nonce}\n{}\n{}\n{}\n{}\n",
        head.answer_key.trim(),
        head.answer_sig.trim(),
        head.log_id,
        head.first_seq,
        head.head_seq,
        head.generation
    );
    crate::durable::write_atomic_unframed(root, GENERATION_PROOF, text.as_bytes())
}

/// **Fence this box when `proof` shows a newer generation of its own log**
/// ([`LOG_FENCED`]). `Some(generation)` = fenced by this call; `None` = not a
/// proof about this box (its own or an older generation, another log, another
/// key, a signature that does not verify), or nothing to fence (a twin, a box
/// mid-restore, a box already fenced). Only a failure to WRITE the fence is an
/// error, and the log is made follower-only even then.
pub fn fence_by_proof(
    stores: &ServerStores,
    proof: &gunnar_proto::GenerationProof,
) -> anyhow::Result<Option<u64>> {
    let root = stores.root();
    if is_twin_root(root) || stores.is_twin() || is_restoring(root) || is_fenced(root) {
        return Ok(None);
    }
    let own = log_generation(root)?;
    if proof.generation <= own {
        return Ok(None);
    }
    let Some(id) = log_id(root) else {
        return Ok(None);
    };
    if proof.log_id.trim() != id {
        tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, theirs = %proof.log_id, mine = %id, "a twin relayed a generation of another log; not this box's business");
        return Ok(None);
    }
    let Some(key) = answering_key(stores) else {
        return Ok(None);
    };
    let Ok(line) = key.public_key().to_openssh() else {
        return Ok(None);
    };
    if proof.answer_key.trim() != line.trim()
        || !gunnar_grpc::auth::verify_twin_answer_at(
            &line,
            &proof.answer_sig,
            &proof.nonce,
            &proof.log_id,
            proof.first_seq,
            proof.head_seq,
            proof.generation,
        )
    {
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, generation = proof.generation, "a twin relayed a generation of this log that this box's host key did not sign: ignored");
        return Ok(None);
    }
    let why = format!(
        "this box is generation {own} of log {id}; generation {} of the same log exists, restored at {} (unix ms) and signed by this \
         box's own host key. This is the OLD main come back after its replacement was restored: it originates nothing and serves no \
         log from here, because its writes since that restore belong to no history. Retire it, or re-join it as a twin.",
        proof.generation, proof.generation
    );
    stores.txlog().set_follower(true);
    crate::durable::write_atomic_unframed(root, LOG_FENCED, format!("{why}\n").as_bytes())
        .context("writing the fence")?;
    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, event = "twin.fenced_by_generation", own, newer = proof.generation, "{why}");
    Ok(Some(proof.generation))
}

fn forget_fence(root: &Path) -> anyhow::Result<()> {
    for name in [LOG_FENCED, GENERATION_PROOF] {
        match std::fs::remove_file(root.join(name)) {
            Ok(()) => crate::durable::sync_dir(root)?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(anyhow::Error::from(e)
                    .context(format!("removing {}", root.join(name).display())))
            }
        }
    }
    Ok(())
}

/// **The key this box signs its answers to a twin with** — its SSH host key,
/// read once per process out of the sealed store.
///
/// The host key and not a new one, deliberately. It is the key the box already
/// proves itself with on :2222, so an operator who has checked one has checked
/// the other; and it is REPLAYED BY A RESTORE, which is exactly the behaviour
/// this pairing wants — a main restored from the twin keeps signing with the
/// same key, so the twin keeps following it without being re-joined, matching
/// what [`LOG_ID`] does for the history. A fresh box at the same address mints
/// its own and is refused. (That same survive-a-restore property is why the
/// host key could NOT be the log identity: there it would have had to
/// distinguish things a restore is supposed to make identical.)
///
/// Cached because a twin polls every two seconds and a sealed-store read per
/// poll is a cost for nothing: a host key does not change under a running
/// process.
fn answering_key(stores: &ServerStores) -> Option<gunnar_core::ssh_key::PrivateKey> {
    static KEY: OnceLock<Option<gunnar_core::ssh_key::PrivateKey>> = OnceLock::new();
    KEY.get_or_init(|| {
        // A twin reads its (replicated) host key through the unframed store; a
        // primary through its own. Both are "the key this box answers as".
        let secrets = if stores.is_twin() { stores.secrets_unframed() } else { stores.secrets() };
        match secrets.and_then(|s| gunnar_core::ensure_host_key(&s).map_err(anyhow::Error::from)) {
            Ok(host) => Some(host.private_key().clone()),
            Err(err) => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box cannot read its host key, so it cannot sign its answers to a twin; a twin that has pinned a key will refuse them");
                None
            }
        }
    })
    .clone()
}

/// The `authorized_keys` line and one signature over this answer, for
/// `Mirror.LogHead`. `("", "")` when the caller sent no nonce (nothing to be
/// fresh about) or this box cannot read its key.
pub fn sign_head_answer(
    stores: &ServerStores,
    nonce: &str,
    log_id: &str,
    first_seq: u64,
    head_seq: u64,
    generation: u64,
) -> (String, String) {
    if nonce.trim().is_empty() {
        return (String::new(), String::new());
    }
    let Some(key) = answering_key(stores) else {
        return (String::new(), String::new());
    };
    let line = match key.public_key().to_openssh() {
        Ok(line) => line,
        Err(err) => {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "this box's host key does not re-encode");
            return (String::new(), String::new());
        }
    };
    match gunnar_grpc::auth::sign_twin_answer_at(
        &key, nonce, log_id, first_seq, head_seq, generation,
    ) {
        Ok(sig) => (line, sig),
        Err(err) => {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "signing this answer to a twin");
            (String::new(), String::new())
        }
    }
}

/// This data root's log identity, when it has one. `None` = never minted (a box
/// older than this change, or a twin that has not yet heard from its primary).
pub fn log_id(root: &Path) -> Option<String> {
    let raw = std::fs::read_to_string(root.join(LOG_ID)).ok()?;
    let id = raw.trim().to_owned();
    if id.is_empty() {
        None
    } else {
        Some(id)
    }
}

/// **Mint this log's identity, once.** Returns the existing one when there is
/// one, so it is safe to call on every open. 128 bits of CSPRNG, hex.
pub fn mint_log_id(root: &Path) -> anyhow::Result<String> {
    if let Some(id) = log_id(root) {
        return Ok(id);
    }
    use gunnar_core::ssh_key::rand_core::Rng as _;
    let mut bytes = [0u8; 16];
    gunnar_core::os_rng().fill_bytes(&mut bytes);
    let id: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
    crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())?;
    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %id, "this box minted its log's identity: from here a twin can tell this history from any other");
    Ok(id)
}

/// **Write the identity of the log this root now holds** — a twin learning its
/// primary's on the first poll, or a fresh box taking one off a restore stream.
/// Refuses to overwrite a DIFFERENT id: that is the splice this whole file
/// exists to prevent, and it must never happen quietly.
pub fn pin_log_id(root: &Path, id: &str) -> anyhow::Result<()> {
    let id = id.trim();
    anyhow::ensure!(!id.is_empty(), "a log identity cannot be empty");
    if let Some(mine) = log_id(root) {
        anyhow::ensure!(
            mine == id,
            "this data root already holds log {mine} and was asked to hold log {id}; two histories do not become \
             one by overwriting a file"
        );
        return Ok(());
    }
    crate::durable::write_atomic_unframed(root, LOG_ID, format!("{id}\n").as_bytes())
}

/// `<data-root>/twin-primary.pub`: **the key the box this twin follows answers
/// with** — one OpenSSH `authorized_keys` line, pinned on the first poll and
/// required on every one after (2026-09-21, lane T4).
///
/// # What was there before
///
/// Nothing. `GrpcSource::connect` dialled `client_signer::plain_channel` — a
/// bare `http://` endpoint, no TLS — and the twin verified NOTHING about who
/// answered. The signed credential on every call proved the TWIN to the
/// primary; not one byte ever went the other way. The trust was one-directional,
/// so whatever won the address won the twin: a DNS answer, a recycled lease, a
/// cable in the wrong port, a box rebuilt at the same IP.
///
/// # Trust on first use, and why there is no earlier moment
///
/// `Bootstrap.BecomeTwin` is answered by the TWIN. The primary is not in that
/// conversation at all — it does not learn it has a twin until that twin's key
/// is enrolled on it by hand, afterwards — so at join there is nothing to pin
/// except an address the operator typed. The first poll is the first time this
/// box has ever heard from the other one. That is exactly the position `ssh`
/// is in the first time it connects to a host, and `known_hosts` is the answer
/// there for the same reason it is the answer here: pinning at first sight
/// converts an open-ended exposure into a single one, and every later change is
/// loud.
///
/// # Its own file, not a field of `twin.json`
///
/// `twin.json` is what the operator wrote at join: an address and a time.
/// This is what the twin MEASURED afterwards. Keeping them apart means a hand
/// edit of the join record cannot silently re-pin the key, and it matches
/// [`LOG_ID`], which is a measured fact about the log in the same way. (It also
/// keeps a struct that four other test files construct by literal from growing
/// a field, one of which this lane must not touch.)
pub const PRIMARY_KEY: &str = "twin-primary.pub";

/// The pinned `authorized_keys` line, when this twin has one.
pub fn pinned_primary_key(root: &Path) -> Option<String> {
    let raw = std::fs::read_to_string(root.join(PRIMARY_KEY)).ok()?;
    let line = raw.trim().to_owned();
    if line.is_empty() {
        None
    } else {
        Some(line)
    }
}

/// **Pin the key the primary answers with**, on the first poll. Refuses to
/// replace a different one: that is the substitution this exists to catch.
pub fn pin_primary_key(root: &Path, line: &str) -> anyhow::Result<()> {
    let line = line.trim();
    anyhow::ensure!(!line.is_empty(), "a primary's key cannot be empty");
    if let Some(mine) = pinned_primary_key(root) {
        anyhow::ensure!(
            mine == line,
            "this twin already follows a box that answers with {mine}; it was offered {line}"
        );
        return Ok(());
    }
    crate::durable::write_atomic_unframed(root, PRIMARY_KEY, format!("{line}\n").as_bytes())?;
    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, key = %line, "this twin has pinned the key its primary answers with; every later answer must be signed by it");
    Ok(())
}

/// **Give up this box's own log identity**, with its numbers and its secrets: a
/// fresh box becoming a twin, or taking another box's log in a restore. Without
/// this it would compare the identity it minted at its own first open against
/// the one it is about to start following and refuse itself at the first poll.
/// **Drop this box's own control-plane audit trail** before it takes another
/// box's log (lane T11).
///
/// Not "best effort": a leftover line would make the restored chain fail
/// `AuditChain::open`, and a box that cannot open its trail serves without one.
/// The restore would then have destroyed the audit history it exists to carry —
/// the exact D6 failure, one layer in.
///
/// It is the replicated path that is removed, and only that: `prepare_replay`
/// runs before the `txlog-restoring` marker exists, so `audit_path()` still
/// names `log/audit-chain.jsonl`, which is precisely the file about to be
/// rebuilt from the stream.
fn forget_own_audit_trail(stores: &ServerStores) -> anyhow::Result<()> {
    let path = stores.audit_path();
    match std::fs::remove_file(&path) {
        Ok(()) => {
            tracing::info!(
                target: gunnar_core::observe::TARGET_TWIN,
                path = %path.display(),
                "dropped this fresh box's own audit trail: the one being restored starts at its own seq 0"
            );
            Ok(())
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e)
            .with_context(|| format!("dropping this box's own audit trail {}", path.display())),
    }
}

/// **Check the control-plane audit trail against this box's own log, on EVERY start**, and
/// rebuild what the file lost — before the gate opens it (lane T11, 2026-09-21; every start
/// since T17, owner's decision the same day).
///
/// **What this is NOT: the fix for T3's KVM `T1v`/`T2v` failure** (medium `ff2c7c9`). This
/// function was first written on the belief that the restored box's copy of the trail, on the
/// boot disk's P3, was lost in the hard stop between the replay and the restart. That diagnosis
/// was WRONG, and it is corrected here rather than left standing. T3 found the real cause:
/// `restore_finish` unlinked `txlog-restoring` with no directory fsync, the hard stop dropped
/// the unlink, and the box reopened STILL RESTORING — a follower that re-sealed nothing,
/// originated nothing, and answered `Admin.AuditTrail` from its own per-box trail. The log
/// says so: the twin polled the restored box every 2 s and saw its head stay at 262, where a
/// box that had re-sealed and served 16 audited calls would have framed far past it. And the
/// guard that decides whether THIS function runs is "not restoring", so it never ran. T3 fixed
/// the marker (`durable::remove_durable`, alarm `twin.restoring_at_open`).
///
/// **The window on a running primary** (T11 named it, T17 closes it): `AuditChain::append`
/// syncs the FRAME, then writes the line to the file. A crash between the two leaves the log one
/// entry ahead of the file; XFS log recovery on P3 after a hard stop can also bring the file
/// back shorter, or with a torn last line. The gate would then append to the SHORTER chain — a
/// second entry at an existing `seq`, a fork of the chain the twin
/// holds, whose copy stops verifying from that entry on. So the check is not a restore's: it
/// runs on every open of a box that originates its log (not a twin, not mid-restore), in
/// `ServerStores::open`, before `serve` opens the chain for the gate. A box that reopens still
/// RESTORING (T3: the marker's unlink lost, fixed by `durable::remove_durable` on
/// `agent/twin-restore-proof`) originates nothing and writes its per-box trail; it is not
/// checked here, and T3's alarm `twin.restoring_at_open` is what names it.
///
/// The trail is a materialised view of the log's frames for [`gunnar_core::txlog::AUDIT_REL`]
/// — `Append` lines, and a `File` frame where a snapshot carried the whole document (which
/// supersedes every earlier line). What the file lacks past its own tip is APPENDED, linked by
/// `prev`, checked in memory to verify, written atomically (temporary, fsync, rename, fsync of
/// the directory), and checked to open.
///
/// **It never discards a line the log does not hold.** The log may carry only a WINDOW of the
/// trail (the operator's retention door, [`Log::retain_from`]; a box upgraded from before the
/// trail was framed), so "the log's view" is never written over the file. A copy LONGER than
/// the log is left as found, and so is one whose tip the log contradicts, or one the log's
/// window cannot be joined to — each said out loud. The only bytes ever dropped are a torn
/// last line that does not parse as an entry.
///
/// Best effort in the sense the gate is: a failure is one `error!` naming the path and the box
/// still serves. But it is never silent — the one thing D6 was.
///
/// Cost: one header-and-meta pass over the log ([`Log::scan_meta`], one open per segment), then
/// blobs read newest-first only until the file's tip is met — ONE blob when the file is whole.
pub fn rematerialise_audit_trail(stores: &ServerStores, just_restored: bool) {
    let path = stores.audit_path();
    let when = if just_restored {
        "the first start after a restore"
    } else {
        "this start"
    };
    let t0 = std::time::Instant::now();
    let outcome = rebuild_audit_trail(stores.txlog(), &path);
    let micros = t0.elapsed().as_micros() as u64;
    match outcome {
        Ok(Rebuilt::Nothing) => {}
        Ok(Rebuilt::Intact(n)) => tracing::info!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            entries = n,
            micros,
            "the audit trail is whole at {when}: its tip is the log's newest audit entry"
        ),
        Ok(Rebuilt::Rebuilt { had, now }) => tracing::warn!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            had,
            now,
            micros,
            "the audit trail held {had} of the {now} entries its log carries at {when} — the \
             off-stripe copy lost its tail (a hard stop; XFS recovery on the boot disk's P3). \
             REBUILT from the log, which did not: the trail is whole again and the next entry \
             continues the chain the twin holds"
        ),
        Ok(Rebuilt::Longer { file, log }) => tracing::warn!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            file,
            log,
            "the audit trail holds MORE entries ({file}) than its log carries ({log}) at {when}; \
             left exactly as found — nothing is deleted from an audit trail"
        ),
        Ok(Rebuilt::Diverged { why }) => tracing::error!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            why = %why,
            "the audit trail and its log DISAGREE at {when}; left exactly as found — nothing is \
             guessed into an audit trail. The twin's copy is the one to compare against"
        ),
        Err(err) => tracing::error!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %path.display(),
            error = %format!("{err:#}"),
            "the audit trail could not be checked against its log at {when}; if the off-stripe \
             copy lost its tail, the gate appends to a shorter chain than the twin holds"
        ),
    }
}

/// What [`rebuild_audit_trail`] did.
#[derive(Debug, PartialEq, Eq)]
pub enum Rebuilt {
    /// The log carries no audit frame at all (a primary from before lane T11, or a box that has
    /// made no audited decision).
    Nothing,
    /// The file's tip is the log's newest audit entry: this many entries in the file.
    Intact(usize),
    /// The file held `had` whole entries; `now - had` were appended from the log.
    Rebuilt {
        /// Whole lines in the file before.
        had: usize,
        /// Lines in the file now.
        now: usize,
    },
    /// The file's tip is past the log's newest audit entry; left alone.
    Longer {
        /// Lines in the file.
        file: usize,
        /// Audit entries the log carries.
        log: usize,
    },
    /// The file and the log cannot be joined without guessing; left alone.
    Diverged {
        /// Which fact disagreed.
        why: String,
    },
}

/// One audit line of the log's view: where it came from, for a lazy read newest-first.
enum ViewLine {
    /// An `Append` frame's blob.
    Frame(u64),
    /// Line `i` of the snapshot's document.
    Snap(usize),
}

/// The work of [`rematerialise_audit_trail`], over a log and a path, so it can be proved without
/// a whole box.
pub fn rebuild_audit_trail(log: &Log, path: &Path) -> anyhow::Result<Rebuilt> {
    // 1. Which frames carry the trail: the newest snapshot of it, then every Append after it.
    let (append, file) = (Kind::Append as u16, Kind::File as u16);
    // The exact bytes this code writes for an audit line's meta: compared, not parsed — parsing the JSON
    // of every audit frame's meta made 90 000 more audit frames cost 263 ms more (MEASURED,
    // `audit_check_boot_cost`). Any other meta naming the trail is parsed in full.
    let audit_meta = Meta::Append {
        log: txlog::AUDIT_REL.into(),
    }
    .to_bytes();
    let names_trail = |meta: &[u8]| {
        meta.windows(txlog::AUDIT_REL.len())
            .any(|w| w == txlog::AUDIT_REL.as_bytes())
    };
    let mut frames: Vec<u64> = Vec::new();
    let mut snapshot: Option<(u64, bool)> = None;
    let mut bad_meta: Option<(u64, String)> = None;
    log.scan_meta(|seq, kind, meta| {
        if kind == append && meta == audit_meta.as_slice() {
            frames.push(seq);
            return;
        }
        if (kind != append && kind != file) || !names_trail(meta) {
            return;
        }
        match Meta::from_bytes(meta) {
            Ok(Meta::Append { log: rel }) if kind == append && rel == txlog::AUDIT_REL => {
                frames.push(seq)
            }
            Ok(Meta::File { path: rel, deleted }) if kind == file && rel == txlog::AUDIT_REL => {
                frames.clear();
                snapshot = Some((seq, deleted));
            }
            Ok(_) => {}
            Err(e) => bad_meta = bad_meta.take().or(Some((seq, e.to_string()))),
        }
    })
    .context("scanning the log's frame headers")?;
    if let Some((seq, e)) = bad_meta {
        return Err(anyhow!("frame {seq}'s meta does not parse: {e}"));
    }
    if frames.is_empty() && snapshot.is_none() {
        return Ok(Rebuilt::Nothing);
    }
    let snap_lines: Vec<Vec<u8>> = match snapshot {
        Some((seq, false)) => split_lines(&log.read(seq)?.blob).0,
        _ => Vec::new(),
    };
    let mut view: Vec<ViewLine> = (0..snap_lines.len()).map(ViewLine::Snap).collect();
    view.extend(frames.iter().map(|s| ViewLine::Frame(*s)));

    // 2. The file: whole lines, and a torn last one (no newline, and not an entry) set aside.
    let bytes = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
        Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
    };
    let (mut lines, rest) = split_lines(&bytes);
    let mut torn = false;
    if let Some(rest) = rest {
        if serde_json::from_slice::<gunnar_core::AuditEntry>(&rest).is_ok() {
            lines.push(rest);
        } else {
            torn = true;
        }
    }
    let had = lines.len();
    let tip: Option<gunnar_core::AuditEntry> = match lines.last() {
        Some(l) => Some(serde_json::from_slice(l).with_context(|| {
            format!(
                "the last whole line of {} is not an audit entry",
                path.display()
            )
        })?),
        None => None,
    };

    // 3. The log's view, newest first, until the file's tip is met: ONE blob when it is whole.
    let line_at = |v: &ViewLine| -> anyhow::Result<Vec<u8>> {
        Ok(match v {
            ViewLine::Frame(seq) => log.read(*seq)?.blob,
            ViewLine::Snap(i) => snap_lines[*i].clone(),
        })
    };
    let mut missing: Vec<Vec<u8>> = Vec::new();
    let mut met = false;
    for (k, v) in view.iter().enumerate().rev() {
        let line = line_at(v)?;
        let entry: gunnar_core::AuditEntry =
            serde_json::from_slice(&line).context("an audit line in the log is not an entry")?;
        match &tip {
            Some(t) if entry.seq == t.seq => {
                if entry.hash != t.hash {
                    return Ok(Rebuilt::Diverged {
                        why: format!(
                            "seq {}: the file's hash is {}, the log's {}",
                            t.seq, t.hash, entry.hash
                        ),
                    });
                }
                met = true;
                break;
            }
            Some(t) if entry.seq < t.seq => {
                if k + 1 == view.len() {
                    return Ok(Rebuilt::Longer {
                        file: lines.len(),
                        log: view.len(),
                    });
                }
                return Ok(Rebuilt::Diverged {
                    why: format!(
                        "the log skips from seq {} to a newer entry, past the file's tip {}",
                        entry.seq, t.seq
                    ),
                });
            }
            _ => missing.push(line),
        }
    }
    if missing.is_empty() && !torn {
        return Ok(Rebuilt::Intact(had));
    }
    missing.reverse();
    if !met {
        // The log's window starts after the file's tip, or the file is empty: only a window that
        // begins exactly one past the tip (or at seq 0 for an empty file) joins without a gap.
        let first: gunnar_core::AuditEntry = serde_json::from_slice(&missing[0])?;
        let want = tip.as_ref().map_or(0, |t| t.seq + 1);
        if first.seq != want {
            return Ok(Rebuilt::Diverged { why: format!("the log's audit entries begin at seq {}, the file needs {want} next — a retention hole nothing can bridge", first.seq) });
        }
    }

    // 4. The joined trail — the exact bytes about to be written — must verify in memory before
    //    one of them lands: a file that was already broken is not made to look whole by an
    //    append, and nothing is written that the gate would then refuse to open.
    let mut out =
        Vec::with_capacity(bytes.len() + missing.iter().map(|l| l.len() + 1).sum::<usize>());
    for l in lines.iter().chain(missing.iter()) {
        out.extend_from_slice(l);
        out.push(b'\n');
    }
    let entries: Vec<gunnar_core::AuditEntry> = out
        .split(|b| *b == b'\n')
        .filter(|l| !l.is_empty())
        .map(serde_json::from_slice)
        .collect::<Result<_, _>>()
        .context("parsing the joined trail")?;
    if let Err(e) = gunnar_core::audit::verify_entries(&entries) {
        return Ok(Rebuilt::Diverged {
            why: format!("the file joined to the log's tail does not verify: {e}"),
        });
    }
    let dir = path
        .parent()
        .ok_or_else(|| anyhow!("{} has no directory", path.display()))?;
    std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
    let name = path
        .file_name()
        .ok_or_else(|| anyhow!("{} names no file", path.display()))?
        .to_string_lossy()
        .into_owned();
    // Unframed: these lines ARE frames already — this box's log holds every one of them.
    crate::durable::write_atomic_unframed(dir, &name, &out)
        .with_context(|| format!("rewriting {}", path.display()))?;
    gunnar_core::AuditChain::open(path)
        .map_err(|e| anyhow!("the rebuilt trail does not open: {e}"))?;
    Ok(Rebuilt::Rebuilt {
        had,
        now: had + missing.len(),
    })
}

/// Whole `\n`-terminated non-empty lines, and what follows the last newline (a torn line), if any.
fn split_lines(bytes: &[u8]) -> (Vec<Vec<u8>>, Option<Vec<u8>>) {
    let (whole, rest) = match bytes.iter().rposition(|b| *b == b'\n') {
        Some(i) => (&bytes[..i], &bytes[i + 1..]),
        None => (&bytes[..0], bytes),
    };
    let lines = whole
        .split(|b| *b == b'\n')
        .filter(|l| !l.iter().all(u8::is_ascii_whitespace))
        .map(<[u8]>::to_vec)
        .collect();
    let rest = (!rest.iter().all(|b| b.is_ascii_whitespace() || *b == 0)).then(|| rest.to_vec());
    (lines, rest)
}

fn forget_log_id(root: &Path) -> anyhow::Result<()> {
    match std::fs::remove_file(root.join(LOG_ID)) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e).with_context(|| format!("dropping this box's own {LOG_ID}")),
    }
}

/// **Was the git listener bound in a twin's shape?** — set once by
/// `serve::start`, when it chose the bind (2026-09-21, lane T4).
static GIT_BOUND_AS_TWIN: AtomicBool = AtomicBool::new(false);

/// `serve::start` says which shape it bound the git listener in.
pub fn note_git_bind_is_twin_shaped(yes: bool) {
    GIT_BOUND_AS_TWIN.store(yes, Ordering::SeqCst);
}

/// **Does this box need a restart before its git listener is where a twin's
/// belongs?**
///
/// `serve::start` picks the git bind ONCE, from `stores.is_twin()` at that
/// instant: a twin gets `127.0.0.1:0`, everything else gets the operator's
/// address. A box conferred the twin role over the wire by
/// `Bootstrap.BecomeTwin` is past that point — and under the owner's
/// one-medium ruling that is the ONLY way a twin is ever made, because a twin
/// is installed from the same image as a main with no per-box parameter to say
/// which it is. MEASURED under KVM by lane T3, 2026-09-21: such a box kept
/// `0.0.0.0:2222` bound for the rest of its life and the
/// `this box is a TWIN: the git listener is bound to loopback only` warning
/// never printed.
///
/// The door refuses every git service on a twin from the instant the role lands
/// (`gunnar_ssh::Control::is_twin`), so nothing is SERVED through that socket —
/// but it is still bound on an address a twin's socket has no business being
/// on, and only a restart moves it. This is how that fact is told rather than
/// left for somebody to find with `ss`.
pub fn git_listener_needs_restart(stores: &ServerStores) -> bool {
    stores.is_twin() && !GIT_BOUND_AS_TWIN.load(Ordering::SeqCst)
}

/// Wall clock, ms since the epoch.
pub fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// **A fresh box gives up its own numbers.** Refused unless the box is
/// fresh: no key enrolled, the bootstrap window open, no repository. The
/// documents it DID write (its first-start issuer, its catalog seed — every
/// `ControlDoc` in its own log) are remembered in [`FRESH_DOCS`]; its log is
/// made follower-only and reset to empty. The first replayed frame for a
/// remembered document is then applied as if the document were absent — the
/// replay is into an empty store, and the fresh box's own value is not a
/// claim the primary ever made.
pub fn prepare_replay(stores: &ServerStores) -> anyhow::Result<()> {
    if !stores.auth().is_empty()? || !stores.auth().bootstrap_window_open()? {
        anyhow::bail!("this box has enrolled identities; only a fresh box (bootstrap window open, no key) can take another box's log");
    }
    if !stores.catalog().list(None)?.is_empty() {
        anyhow::bail!("this box holds repositories; only a fresh box can take another box's log");
    }
    let log = stores.txlog();
    let mut names = std::collections::BTreeSet::new();
    if log.head_seq() >= log.first_seq() {
        for e in log.range(log.first_seq(), log.head_seq())? {
            if let Ok(Meta::ControlDoc { name, .. }) = e.meta() {
                names.insert(name);
            }
        }
    }
    crate::durable::write_atomic_unframed(stores.root(), FRESH_DOCS, &serde_json::to_vec(&names)?)?;
    forget_own_secrets(stores)?;
    // **And its own log's NAME.** This box minted one at its first open, for a
    // history it is about to give up entirely. Keeping it would make the twin
    // refuse itself at its very first poll — its own id against the primary's —
    // and would make a restored box claim to be a history it only holds a copy
    // of. The id it ends up with is the one it is told: the primary's, on the
    // first poll, or the stream's, at `restore_finish`.
    forget_log_id(stores.root())?;
    // And any primary generation it pinned: the generation it ends up with is
    // the one `restore_finish` stamps (row 18). Its OWN `txlog-generation` stays,
    // so the stamp only ever moves up.
    forget_generation(stores.root(), PINNED_GENERATION)?;
    // And a fence from a life as an old main, and any proof it relayed (row 18).
    forget_fence(stores.root())?;
    // **And its own AUDIT TRAIL** (lane T11, 2026-09-21). The trail is a
    // replicated, hash-chained document now, so the frames arriving from the
    // twin are `Append`s of the DEAD primary's chain from its seq 0. A fresh
    // box that had written even one line of its own would take them on the end
    // of it, and the result would not verify: entry 0's `prev` is GENESIS, and
    // here it would follow somebody else's tip. The same reasoning as
    // `forget_own_secrets` and `forget_log_id` above — this box is giving up
    // its own short history entirely, not merging it with the one it is being
    // handed.
    forget_own_audit_trail(stores)?;
    log.set_follower(true);
    log.reset(u64::MAX)?;
    Ok(())
}

/// **A fresh box gives up its own SECRETS with its own numbers** (2026-09-17, lane
/// TWIN-COMPLETENESS).
///
/// Every secret a fresh box holds is sealed under the KEK it minted at first open, and the
/// replay's `Kek` frame — seq 1 of every primary's log — overwrites that key on disk before any
/// `Secret` frame lands. A row the replay then OVERWRITES is fine; a row it has no frame for is
/// left sealed under a key that no longer exists, and [`reseal_after_restore`] meets it with
/// neither key able to open it.
///
/// MEASURED: a fresh box holding one secret of its own refused to open after an otherwise
/// successful restore, every time, for ever. [`reseal_after_restore`] now drops such a row
/// loudly rather than bricking the box; this is the other half — the row never reaches it,
/// because a box that gives up its numbers and its documents has no business keeping the keys
/// it sealed under a KEK it is about to lose. The box is FRESH by the checks above (bootstrap
/// window open, no enrolled identity, no repository), so nothing here is anybody's data: a host
/// key it minted at first boot, an issuer it minted at first start. A twin re-mints what it
/// needs through its unlogged store at the next open, and the primary's replayed value
/// overwrites that in turn.
fn forget_own_secrets(stores: &ServerStores) -> anyhow::Result<()> {
    use gunnar_store::SecretStore as _;
    let rows = stores
        .control_db()
        .export()
        .context("reading this box's own secrets before it gives up its numbers")?
        .secrets;
    let mut gone = Vec::new();
    for (key, value) in &rows {
        if value.is_none() {
            continue;
        }
        stores
            .raw_secrets()
            .delete(key)
            .with_context(|| format!("dropping this box's own secret {key}"))?;
        gone.push(key.clone());
    }
    if !gone.is_empty() {
        tracing::warn!(
            target: gunnar_core::observe::TARGET_TWIN,
            secrets = ?gone,
            "this box gave up the secrets it minted for itself along with its own log numbers: they are sealed under a KEK the replay is about to replace, and what the log carries is the only identity this box keeps"
        );
    }
    Ok(())
}

/// The fresh-box rule, one document: `true` (and forgotten) when `name` was
/// written by this box before it was reset.
fn fresh_take(root: &Path, name: &str) -> anyhow::Result<bool> {
    let path = root.join(FRESH_DOCS);
    let bytes = match std::fs::read(&path) {
        Ok(b) => b,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(e) => return Err(e.into()),
    };
    let mut names: std::collections::BTreeSet<String> = serde_json::from_slice(&bytes)?;
    if !names.remove(name) {
        return Ok(false);
    }
    crate::durable::write_atomic_unframed(root, FRESH_DOCS, &serde_json::to_vec(&names)?)?;
    Ok(true)
}

/// A path from a frame, joined under `root` — never absolute, never `..`,
/// never a cache.
fn under_root(root: &Path, rel: &str) -> anyhow::Result<std::path::PathBuf> {
    let p = Path::new(rel);
    if p.is_absolute()
        || p.components()
            .any(|c| !matches!(c, std::path::Component::Normal(_)))
    {
        anyhow::bail!("the frame names {rel:?}, which is not a plain path under the data root");
    }
    if txlog::never_framed(rel) {
        anyhow::bail!("the frame names {rel:?}, a cache or per-box path — never applied on a twin");
    }
    Ok(root.join(p))
}

/// Append `line` + newline to `path` unless the file already ends with
/// exactly that line (a re-apply of the tail after a restart).
fn append_line_once(path: &Path, line: &[u8]) -> anyhow::Result<()> {
    use std::io::{Read as _, Seek as _, SeekFrom, Write as _};
    let mut want = line.to_vec();
    want.push(b'\n');
    if let Ok(mut f) = std::fs::File::open(path) {
        let len = f.metadata()?.len();
        if len >= want.len() as u64 {
            f.seek(SeekFrom::Start(len - want.len() as u64))?;
            let mut tail = vec![0u8; want.len()];
            f.read_exact(&mut tail)?;
            let boundary = len == want.len() as u64 || {
                f.seek(SeekFrom::Start(len - want.len() as u64 - 1))?;
                let mut b = [0u8; 1];
                f.read_exact(&mut b)?;
                b[0] == b'\n'
            };
            if tail == want && boundary {
                return Ok(());
            }
        }
    }
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let mut f = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)?;
    f.write_all(&want)?;
    f.sync_data()?;
    Ok(())
}

/// The primary's KEK, written where `Kek::for_control_dir` reads it (0600,
/// atomic). Takes effect at the next open — a restored box restarts.
fn write_kek(control_dir: &Path, bytes: &[u8]) -> anyhow::Result<()> {
    use std::io::Write as _;
    anyhow::ensure!(
        bytes.len() == 32,
        "a KEK frame carries {} bytes, not 32",
        bytes.len()
    );
    let path = gunnar_core::seal::kek_path(control_dir);
    if std::fs::read(&path).ok().as_deref() == Some(bytes) {
        return Ok(());
    }
    std::fs::create_dir_all(control_dir)?;
    let tmp = control_dir.join("secrets.kek.tmp");
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create(true).truncate(true);
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        options.mode(0o600);
    }
    let mut f = options.open(&tmp)?;
    f.write_all(bytes)?;
    f.sync_all()?;
    drop(f);
    std::fs::rename(&tmp, &path)?;
    // The KEK is written once its name is durable (stinky row 9).
    crate::durable::sync_dir(control_dir)
}

// ── the doors: join as a twin, restore a fresh box ──────────────────────────

/// **`Bootstrap.BecomeTwin`**: a fresh box becomes a twin of `primary`. Mints
/// the twin's key (per box, `twin.key`), writes `twin.json`, gives up its own
/// numbers ([`prepare_replay`]) and freezes its control plane. Returns the
/// public key line to enrol on the primary in group [`TWIN_GROUP`]. Idempotent
/// for the same primary (the same key comes back); another primary is refused.
pub fn become_twin(stores: &ServerStores, primary: &str) -> anyhow::Result<String> {
    let root = stores.root();
    if let Some(existing) = TwinConfig::load(root)? {
        anyhow::ensure!(
            existing.primary == primary,
            "this box is already the twin of {}; a twin is never re-pointed",
            existing.primary
        );
        return twin_public_key(root);
    }
    anyhow::ensure!(
        !primary.trim().is_empty(),
        "the primary's address is required"
    );
    prepare_replay(stores)?;
    if !root.join(TWIN_KEY).exists() {
        let key = gunnar_core::mint_ed25519()?;
        crate::durable::write_atomic_unframed(root, TWIN_KEY, &key)?;
        use std::os::unix::fs::PermissionsExt as _;
        std::fs::set_permissions(root.join(TWIN_KEY), std::fs::Permissions::from_mode(0o600))?;
    }
    TwinConfig {
        primary: primary.trim().to_string(),
        joined_at_unix_ms: now_ms(),
    }
    .save(root)?;
    stores.mark_twin();
    forget_mirrors(stores);
    stores.browse().store(true, Ordering::SeqCst);
    stores.writes_frozen().store(true, Ordering::SeqCst);
    twin_public_key(root)
}

/// **A twin keeps no per-box mirror.** The metadata mirror (`slow`'s `meta` dir) holds DERIVED
/// copies — the identity table, `replication.json`, `webhooks.json` — and a fresh box writes its
/// own at first open, before anybody can tell it to become a twin. Those copies are cache on a
/// twin (C7): removed here, and never written again once the box is marked. Best effort: a file
/// that is not there is the goal, and a removal that fails is one WARN naming the path.
pub fn forget_mirrors(stores: &ServerStores) {
    let meta = stores.slow().path_of(crate::slow::META_DIR);
    #[allow(unused_mut)]
    let mut names = vec![crate::stores::IDENTITY_MIRROR];
    // A build without the feature never wrote that mirror, so it has nothing to forget.
    #[cfg(feature = "replication")]
    names.push(crate::mirror::REPLICATION_CONFIG);
    #[cfg(feature = "webhooks")]
    names.push(crate::webhook::WEBHOOK_CONFIG);
    for name in names {
        let path = meta.join(name);
        match std::fs::remove_file(&path) {
            Ok(()) => {
                tracing::info!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), "removed a per-box mirror: a twin keeps none")
            }
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
            Err(err) => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, path = %path.display(), %err, "could not remove a per-box mirror on a twin")
            }
        }
    }
}

/// The twin's key.
pub fn twin_private_key(root: &Path) -> anyhow::Result<gunnar_core::ssh_key::PrivateKey> {
    let bytes = std::fs::read(root.join(TWIN_KEY))
        .with_context(|| format!("reading {}", root.join(TWIN_KEY).display()))?;
    gunnar_core::ssh_key::PrivateKey::from_openssh(&bytes).map_err(|e| anyhow!("{TWIN_KEY}: {e}"))
}

/// The twin's public key line.
pub fn twin_public_key(root: &Path) -> anyhow::Result<String> {
    let key = twin_private_key(root)?;
    let mut line = key.public_key().to_openssh().map_err(|e| anyhow!("{e}"))?;
    line.push_str(" gunnar-twin");
    Ok(line)
}

/// **The owner of an in-process restore** ([`restore`], [`restore_from`]): a caller that
/// holds the data root itself, and so needs no key to prove anything to this box.
pub const RESTORE_OWNER_LOCAL: &str = "local";

/// The refusal word for a resume by somebody who did not start the restore.
pub const RESTORE_NOT_YOURS: &str = "restore-not-yours";

/// **A resume by somebody else, refused** (h-v1-6a, 2026-09-28). Its own type so the
/// control plane answers it `PERMISSION_DENIED` — the caller is known and the answer is no —
/// rather than the `FAILED_PRECONDITION` of "this box is not fresh".
#[derive(Debug)]
pub struct RestoreNotYours {
    /// Who the [`RESTORING`] marker says started it.
    pub started_by: String,
    /// Who asked to resume it.
    pub asked_by: String,
}

impl std::fmt::Display for RestoreNotYours {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "REFUSED [{RESTORE_NOT_YOURS}] this box is mid-restore and the restore was started by {}; {} did not start it \
             and cannot resume it. Only the key that started a restore resumes it — run the same replay again with the \
             same key (monetize's restore lane, or `gunnar control twin replay -i <that key>`). A box whose starter is \
             gone for good is re-imaged, which makes it fresh again; nothing of the stream so far was touched",
            self.started_by, self.asked_by
        )
    }
}

impl std::error::Error for RestoreNotYours {}

/// The [`RESTORING`] marker's content: the word, then who started it.
fn restoring_marker(owner: &str) -> String {
    format!("restoring\nowner {owner}\n")
}

/// Who started the restore in progress, out of the [`RESTORING`] marker. `None` for a marker
/// with no owner line — one written before 2026-09-28, when the marker was the bare word.
fn restoring_owner(root: &Path) -> anyhow::Result<Option<String>> {
    let text = std::fs::read_to_string(root.join(RESTORING))
        .with_context(|| format!("reading {}", root.join(RESTORING).display()))?;
    Ok(text
        .lines()
        .find_map(|l| l.strip_prefix("owner "))
        .map(|o| o.trim().to_owned())
        .filter(|o| !o.is_empty()))
}

/// **One lock per data root around `restore_begin`'s check-then-mark** — so two streams
/// arriving together cannot both find the box fresh (or both find no owner) and both mark
/// it. Per ROOT rather than one for the process because a test process holds many boxes.
fn restore_begin_lock(root: &Path) -> Arc<std::sync::Mutex<()>> {
    static LOCKS: std::sync::OnceLock<
        std::sync::Mutex<std::collections::HashMap<PathBuf, Arc<std::sync::Mutex<()>>>>,
    > = std::sync::OnceLock::new();
    let mut locks = LOCKS
        .get_or_init(Default::default)
        .lock()
        .unwrap_or_else(|p| p.into_inner());
    Arc::clone(locks.entry(root.to_path_buf()).or_default())
}

/// **`Bootstrap.Restore`, the start**, for a caller holding the data root itself
/// ([`RESTORE_OWNER_LOCAL`]). See [`restore_begin_by`].
pub fn restore_begin(stores: &ServerStores) -> anyhow::Result<u64> {
    restore_begin_by(stores, RESTORE_OWNER_LOCAL)
}

/// **`Bootstrap.Restore`, the start**: the next sequence this box takes. A
/// box already restoring continues at its head + 1 — **only for the `owner`
/// that started it**; otherwise it must be fresh ([`prepare_replay`]), starts at
/// 1, and the marker records `owner`.
///
/// `owner` is who the caller PROVED to be — `Bootstrap.Restore` passes the
/// fingerprint of the key its credential verified under (h-v1-6a, 2026-09-28).
/// Until then the resume branch below asked nothing at all: once
/// `txlog-restoring` existed, anybody who reached :50051 could stream the rest
/// of the box's history into it, KEK and identity table included. A marker
/// written before that day names no owner; the first verified caller adopts it,
/// said out loud — the credential it presented is what keeps a stranger out.
///
/// The check and the mark are one step under [`restore_begin_lock`].
pub fn restore_begin_by(stores: &ServerStores, owner: &str) -> anyhow::Result<u64> {
    anyhow::ensure!(
        !owner.trim().is_empty() && !owner.contains('\n'),
        "a restore needs the name of who is restoring; {owner:?} is not one"
    );
    let lock = restore_begin_lock(stores.root());
    let _held = lock.lock().unwrap_or_else(|p| p.into_inner());
    if is_restoring(stores.root()) {
        match restoring_owner(stores.root())? {
            Some(started_by) if started_by != owner => {
                return Err(RestoreNotYours {
                    started_by,
                    asked_by: owner.to_owned(),
                }
                .into());
            }
            Some(_) => {}
            None => {
                tracing::warn!(
                    target: gunnar_core::observe::TARGET_TWIN,
                    owner,
                    "this box is mid-restore under a marker that names no owner (written before 2026-09-28); {owner} resumes it and owns it from now on"
                );
                crate::durable::write_atomic_unframed(
                    stores.root(),
                    RESTORING,
                    restoring_marker(owner).as_bytes(),
                )?;
            }
        }
    }
    // One stream, one staging area: whatever a stream that BROKE left staged is not part of
    // this one ([`snapshot::forget_restore_stager`]). After the owner check, so a refused
    // caller throws away nothing of the starter's.
    snapshot::forget_restore_stager(stores.root());
    if is_restoring(stores.root()) {
        stores.txlog().set_follower(true);
        // The stream that broke may have appended its last entry without applying it: apply
        // the head again before taking more (the follower's `reapply` rule).
        let head = stores.txlog().head_seq();
        if head > 0 && head >= stores.txlog().first_seq() && !snapshot::installing(stores.root()) {
            let e = stores.txlog().read(head)?;
            apply(stores, &e)
                .with_context(|| format!("re-applying seq {head} before the restore resumes"))?;
        }
        return Ok(stores.txlog().head_seq() + 1);
    }
    anyhow::ensure!(
        !stores.is_twin(),
        "a twin is never restored into; restore into a fresh primary"
    );
    prepare_replay(stores)?;
    // A fresh box holds no part of anybody's entry, nor of anybody's snapshot.
    intake::forget(stores.root())?;
    snapshot::forget_restore_snapshot(stores.root())?;
    crate::durable::write_atomic_unframed(
        stores.root(),
        RESTORING,
        restoring_marker(owner).as_bytes(),
    )?;
    Ok(1)
}

/// **What a restore costs on the box taking it, per byte the sender streams.**
///
/// A restore writes each frame TWICE: once into this box's own log ([`restore_apply`] does
/// `append_at` before `apply`, durable first) and once into the store the frame builds — a pack
/// into the archive, an LFS object into its file, a document into `control.redb`. So the floor
/// is 2. The margin above it is the engine's own overhead on ingest: the archive's oid index
/// (`objects.tail`, measured at ~0.8 % of a 200 MB pack in the note's twin-image table), the
/// ref log, redb's B-tree pages, and the filesystem's own metadata on tens of thousands of
/// small files.
///
/// **Deliberately an ESTIMATE, and it is named as one in the refusal.** A tight number would
/// need the sender to say what its stores measure rather than what its log measures, which is a
/// wire change this rule does not need in order to catch the case it exists for: a set that is
/// half the size it left with. It is a floor, not a promise — a restore that passes this can
/// still run out on a set that only just fits, and that failure is the ordinary mid-stream one
/// (the box stays `restoring`, resumable), not a silent short restore.
pub const RESTORE_BYTES_PER_LOG_BYTE: u64 = 2;

/// Room the box keeps for itself on top of the restore: the exploded read cache the engine
/// grows on the first clone, the crash and audit records, and the slack a full xfs needs to
/// keep allocating. 5 % of the volume, never less than 256 MiB.
fn reserve_bytes(total: u64) -> u64 {
    (total / 20).max(256 << 20)
}

/// **Does this box have room for the restore, asked BEFORE the first byte of it lands?**
///
/// The case this exists for is a restore onto DIFFERENT hardware: the data set left a machine
/// with four members and arrives at one with two, which may hold less total space than the set
/// being restored. Without this the stream fills the volume and the apply fails somewhere in
/// the middle — recoverable (the box stays `restoring` and nothing says "restored"), but the
/// operator learns it after an hour of streaming and learns it as an ENOSPC rather than as a
/// number. With it the box refuses at the first chunk and says how much is short.
///
/// `announced_bytes` of 0 means the sender did not say (an older `gunnar control twin replay`,
/// or a caller streaming a snapshot whose size it cannot know in advance): no refusal, because
/// a guess is not a measurement. The whole check is one `statvfs`.
pub fn room_for_restore(root: &Path, announced_bytes: u64) -> anyhow::Result<()> {
    if announced_bytes == 0 {
        return Ok(());
    }
    let (total, used) = volume_bytes(root);
    if total == 0 {
        // Unmeasurable is not "full": a box whose statvfs failed is one this must not refuse.
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "the data volume could not be measured; the restore's room is not checked");
        return Ok(());
    }
    let free = total.saturating_sub(used);
    let need = announced_bytes
        .saturating_mul(RESTORE_BYTES_PER_LOG_BYTE)
        .saturating_add(reserve_bytes(total));
    if free >= need {
        tracing::info!(
            target: gunnar_core::observe::TARGET_TWIN,
            announced_bytes, need, free, total,
            "room for the restore: the data set on this machine holds it"
        );
        return Ok(());
    }
    anyhow::bail!(
        "NOT ENOUGH ROOM for this restore, and nothing has been written: the sender streams \
         {announced} of log, which needs about {need} on this box ({factor}× the log — a restore \
         writes every frame into this box's own log AND into the store it builds — plus {reserve} \
         the box keeps for its caches and slack), and this data set has {free} free of {total}. \
         SHORT BY {short}. This box is being restored onto a SMALLER data set than the one the \
         twin's log came from: give it more or larger members and restore again. NOTHING of the \
         twin's log has landed — this box's log is still empty and it stays `restoring`, so the \
         same replay into a bigger set starts at seq 1 and loses nothing. The estimate is \
         DELIBERATELY conservative and a long-lived box's log is larger than the stores it \
         builds; if you have read these numbers and this set does hold it, re-run the replay \
         with `--ignore-room`.",
        announced = human(announced_bytes),
        need = human(need),
        factor = RESTORE_BYTES_PER_LOG_BYTE,
        reserve = human(reserve_bytes(total)),
        free = human(free),
        total = human(total),
        short = human(need - free),
    )
}

/// Bytes as an operator reads them. Small enough to live here rather than pull a crate in.
fn human(n: u64) -> String {
    const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
    let mut v = n as f64;
    let mut u = 0;
    while v >= 1024.0 && u + 1 < UNITS.len() {
        v /= 1024.0;
        u += 1;
    }
    if u == 0 {
        format!("{n} B")
    } else {
        format!("{v:.1} {}", UNITS[u])
    }
}

/// One entry of a restore: durable in the log at the twin's number, then applied.
pub fn restore_apply(stores: &ServerStores, e: &Entry) -> anyhow::Result<Applied> {
    stores.txlog().append_at(e.seq, e.kind, &e.meta, &e.blob)?;
    apply(stores, e)
}

/// The end of a restore: the marker and the fresh-docs list go. The log stays
/// follower-only in THIS process — it holds the fresh box's KEK in memory and
/// the replayed secrets are sealed under the primary's — so the box restarts,
/// and opens as the primary it was.
/// `announced_head` is the head the sender said it streams through (`LogChunk.restore_head`,
/// the largest seen; 0 = not announced): a box below it has taken a SHORT stream — one that
/// ended cleanly although the sender's fetch broke — and stays restoring, resumable by
/// resending from its head + 1.
pub fn restore_finish(
    stores: &ServerStores,
    announced_head: u64,
    log_id_of_source: &str,
) -> anyhow::Result<u64> {
    finish_restore(stores, announced_head, log_id_of_source, false)
}

/// **[`restore_finish`] of a restore streamed over the network** (`Bootstrap.Restore`): the stream
/// MUST name the log it restores, and so must every earlier stream of the same restore (stinky
/// row 7, 2026-10-10). The lineage check lets an empty id through on either side — right for a
/// local replay, which reads the id off the twin's disk — so a sender that named nothing could
/// finish a box on another history whenever the sequence numbers happened to fit. Every sender
/// in the tree names it (`LogHead.log_id`, copied onto every chunk).
pub fn restore_finish_streamed(
    stores: &ServerStores,
    announced_head: u64,
    log_id_of_source: &str,
) -> anyhow::Result<u64> {
    finish_restore(stores, announced_head, log_id_of_source, true)
}

fn finish_restore(
    stores: &ServerStores,
    announced_head: u64,
    log_id_of_source: &str,
    streamed: bool,
) -> anyhow::Result<u64> {
    // **What this restore was told it restores holds for every resend** (estate-box3
    // plain6): a resend whose sender's fetch failed before its first entry announces
    // nothing, and judged alone it would finish a box that an EARLIER stream of the
    // same restore announced to reach further. The largest head and the source's log
    // id ever announced are kept beside the marker until the restore finishes.
    let kept = restore_announced(stores.root())?;
    let now_id = log_id_of_source.trim();
    if streamed {
        anyhow::ensure!(
            !now_id.is_empty(),
            "this restore stream named no log (LogChunk.log_id is empty): a box is restored from one named history, never from whatever numbers fit — this box stays restoring; resend the twin's log with its log id"
        );
        if let Some(kept) = &kept {
            anyhow::ensure!(
                !kept.log_id.is_empty(),
                "an earlier stream of this restore named no log ({RESTORE_ANNOUNCED} says head {} and no log id): which history this box holds is unknown — this box stays restoring; start the restore again on a fresh box",
                kept.head
            );
        }
    }
    let Announced {
        head: kept_head,
        log_id: kept_id,
    } = kept.unwrap_or_default();
    anyhow::ensure!(
        kept_id.is_empty() || now_id.is_empty() || kept_id == now_id,
        "this restore began as log {kept_id} and a resend is log {now_id}: one restore is one history — this box stays restoring"
    );
    let announced_head = announced_head.max(kept_head);
    let log_id_of_source = if now_id.is_empty() {
        kept_id.as_str()
    } else {
        now_id
    };
    if announced_head > 0 || !log_id_of_source.is_empty() {
        crate::durable::write_atomic_unframed(
            stores.root(),
            RESTORE_ANNOUNCED,
            format!("{announced_head} {log_id_of_source}\n").as_bytes(),
        )?;
    }
    // A stream that ended with nothing in it is not a restore: a fetch that failed before its
    // first entry looks exactly like this from here. The box stays restoring, so the operator
    // resends the twin's log rather than restarting into an empty primary.
    anyhow::ensure!(
        stores.txlog().head_seq() > 0,
        "nothing was restored (the stream carried no entry); this box stays restoring — resend the twin's log"
    );
    let head = stores.txlog().head_seq();
    anyhow::ensure!(
        head >= announced_head,
        "the stream ended at seq {head} but the sender announced the twin's head {announced_head}; a short stream is not a restore — this box stays restoring, resend the twin's log"
    );
    // **This box IS the history it was just handed** (2026-09-21, lane T4).
    // `prepare_replay` dropped whatever id this fresh box had minted for itself;
    // the stream's id goes on now, BEFORE the restoring marker comes off, so no
    // open can ever see a restored box without the name of what it restored. A
    // sender older than this change announces nothing, and then the box mints
    // its own at the next open — the pre-2026-09-21 behaviour exactly.
    let announced_id = log_id_of_source.trim();
    if !announced_id.is_empty() {
        pin_log_id(stores.root(), announced_id)
            .context("taking the identity of the log this box was restored from")?;
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, log_id = %announced_id, "this box IS log {announced_id} now: the twin it was restored from keeps following it without being re-joined");
    }
    // **And a NEWER generation of it** (stinky row 18): the same name as the
    // main it replaces, ordered after it, so the twin can refuse that old main
    // if it ever comes back. Stamped before the restoring marker comes off, like
    // the id.
    let generation =
        stamp_generation(stores.root()).context("stamping this restore's generation of the log")?;
    forget_fence(stores.root())
        .context("a restored box is the newest generation: no fence from an older life")?;
    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation, "this restore is generation {generation} of its log: a twin refuses any older one from here");
    let _ = std::fs::remove_file(stores.root().join(FRESH_DOCS));
    // The next open re-seals every secret under a KEK this box owns ([`reseal_after_restore`]).
    // Written BEFORE the restoring marker goes, so no open can see "restored" without "re-seal".
    crate::durable::write_atomic_unframed(stores.root(), RESEAL, b"reseal\n")?;
    // DURABLY: the box is "restarted" next, and on the appliance that can be a stop that
    // loses the guest's page cache. An unlink that did not reach the disk reopens the box
    // mid-restore — read-only, forever (`crate::durable::remove_durable`, MEASURED).
    crate::durable::remove_durable(stores.root(), RESTORING)?;
    let _ = std::fs::remove_file(stores.root().join(RESTORE_ANNOUNCED));
    // Nothing staged outlives the restore (restore-large-entry): an entry staged is applied
    // before its staging goes, so whatever is left here is no part of this box's history.
    let _ = intake::forget(stores.root());
    snapshot::forget_restore_snapshot_or_warn(stores.root());
    Ok(stores.txlog().head_seq())
}

/// `<data-root>/txlog-restore-announced`: `<head> <log id>` — the largest head
/// and the source's log id any stream of the restore under way announced.
pub const RESTORE_ANNOUNCED: &str = "txlog-restore-announced";

/// What a restore under way was announced to be ([`RESTORE_ANNOUNCED`]).
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Announced {
    /// The largest head any stream announced; 0 = none.
    pub head: u64,
    /// The source's log id; empty = none named.
    pub log_id: String,
}

/// **[`RESTORE_ANNOUNCED`] could not be trusted** (stinky row 4, 2026-10-10): it used to be read
/// with `unwrap_or_default()` and its head parsed with `.ok().unwrap_or(0)`, so a marker that
/// could not be read, or did not parse, meant "nothing announced" — and a restore could finish
/// on a shorter head, as another log. Now the restore refuses by name and the box stays restoring.
#[derive(Debug, thiserror::Error)]
pub enum RestoreMarkerError {
    /// The marker is there and could not be read.
    #[error("the restore marker {path} could not be read ({source}): what this restore was announced to reach is unknown — this box stays restoring")]
    Unreadable {
        /// The marker.
        path: std::path::PathBuf,
        /// Why.
        source: std::io::Error,
    },
    /// The marker does not say `<head> [<log id>]`.
    #[error("the restore marker {path} is corrupt ({what}: {text:?}): what this restore was announced to reach is unknown — this box stays restoring")]
    Corrupt {
        /// The marker.
        path: std::path::PathBuf,
        /// What is wrong with it.
        what: &'static str,
        /// What it holds.
        text: String,
    },
}

/// What the restore under way was announced to be; `None` when no stream announced anything (no
/// marker). A marker that cannot be read or parsed is a [`RestoreMarkerError`], never "nothing".
pub fn restore_announced(root: &Path) -> Result<Option<Announced>, RestoreMarkerError> {
    let path = root.join(RESTORE_ANNOUNCED);
    let text = match std::fs::read_to_string(&path) {
        Ok(t) => t,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(source) => return Err(RestoreMarkerError::Unreadable { path, source }),
    };
    let corrupt = |what| RestoreMarkerError::Corrupt {
        path: path.clone(),
        what,
        text: text.clone(),
    };
    let mut words = text.split_whitespace();
    let head = words
        .next()
        .ok_or_else(|| corrupt("it is empty"))?
        .parse::<u64>()
        .map_err(|_| corrupt("its head is not a number"))?;
    let log_id = words.next().unwrap_or("").to_string();
    if words.next().is_some() {
        return Err(corrupt("it holds more than a head and a log id"));
    }
    Ok(Some(Announced { head, log_id }))
}

/// **What a box does once a restore FINISHED** (c-no-restart): the restored
/// box must restart to open as the primary it was (the replayed secrets are
/// sealed under the dead main's KEK, the log is follower-only in this
/// process). Nobody outside restarted it — monetize only logged
/// `needs_restart` — so the box stayed half-open for ever.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AfterRestore {
    /// The appliance's PID 1: sync and reboot itself, a few seconds after the
    /// answer went out.
    Reboot,
    /// Anything else (a container, a developer's process): say so loudly; the
    /// operator restarts it.
    Operator,
}

/// Pure: which [`AfterRestore`], for a process that is (or is not) PID 1 of
/// the appliance's own kernel.
pub fn after_restore(is_init: bool, appliance: bool) -> AfterRestore {
    if is_init && appliance {
        AfterRestore::Reboot
    } else {
        AfterRestore::Operator
    }
}

/// How long the restored box waits before it reboots: the `Bootstrap.Restore`
/// answer must reach the lane first.
pub const REBOOT_AFTER_RESTORE: std::time::Duration = std::time::Duration::from_secs(5);

/// **Restart the restored box** ([`after_restore`]): on the appliance, sync
/// and reboot on a thread of its own after [`REBOOT_AFTER_RESTORE`]; anywhere
/// else, say that the operator must. Returns what was decided.
pub fn restart_after_restore() -> AfterRestore {
    let how = after_restore(
        appliance_pid1::net::is_init(),
        appliance_pid1::net::booted_by_korp_start(),
    );
    match how {
        AfterRestore::Reboot => {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, after_s = REBOOT_AFTER_RESTORE.as_secs(), "RESTORED from the twin: this box reboots itself to open as the primary it was");
            // A28 M18, stays a thread: one wait, then the reboot; a start that fails is said.
            let spawned = std::thread::Builder::new()
                .name("restore-reboot".into())
                .spawn(|| {
                    std::thread::sleep(REBOOT_AFTER_RESTORE);
                    crate::narrate::reboot_now();
                });
            if let Err(e) = spawned {
                tracing::error!(target: gunnar_core::observe::TARGET_TWIN, error = %e, "the reboot after the restore could not be scheduled; restart this box");
            }
        }
        AfterRestore::Operator => {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, "RESTORED from the twin: RESTART this process — it opens as the primary it was (not the appliance's PID 1, so it does not reboot itself)");
        }
    }
    how
}

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

    #[test]
    fn only_the_appliances_pid_1_reboots_itself_after_a_restore() {
        assert_eq!(after_restore(true, true), AfterRestore::Reboot);
        assert_eq!(
            after_restore(true, false),
            AfterRestore::Operator,
            "a container's PID 1 never reboots the host"
        );
        assert_eq!(after_restore(false, true), AfterRestore::Operator);
        assert_eq!(after_restore(false, false), AfterRestore::Operator);
    }
}

/// `<data-root>/txlog-reseal`: a restore finished; the next open re-seals under a KEK of this
/// box's own. Per box, never framed.
pub const RESEAL: &str = "txlog-reseal";
/// `<control-dir>/secrets.kek.new`: the KEK being re-sealed to, durable before the first secret
/// moves, so a crash mid-way resumes with the same key.
const KEK_NEXT: &str = "secrets.kek.new";

/// Is a re-seal owed under `root`?
pub fn reseal_pending(root: &Path) -> bool {
    root.join(RESEAL).exists()
}

/// **A restored box takes a KEK of its own** (`.nornir/twin-transaction-log.md` §"The KEK after
/// a restore"). The replayed `secrets.kek` is the dead primary's; every sealed secret is opened
/// with it and sealed again under a new key (or `GUNNAR_SECRET_KEK` when the box is handed one),
/// and the old key leaves the disk. Each step is a frame first — a `Kek` frame with the new key,
/// then one `Secret` frame per re-sealed value — so the twin following this box holds the new
/// key and the re-sealed bytes, and a later restore from it opens them.
///
/// Crash-safe by construction: the new key is durable in `secrets.kek.new` before any secret
/// moves; a value that already opens under the new key is skipped; the rename over
/// `secrets.kek` and the marker's removal come last. Run once at open, before any sealed store
/// is built. Returns the KEK the box now uses.
pub fn reseal_after_restore(
    root: &Path,
    control_dir: &Path,
    log: &Log,
    rows: &std::collections::BTreeMap<String, Option<Vec<u8>>>,
    raw: &dyn gunnar_store::SecretStore,
) -> anyhow::Result<gunnar_core::Kek> {
    use gunnar_core::Kek;
    let path = gunnar_core::seal::kek_path(control_dir);
    let key32 = |bytes: Vec<u8>, what: &Path| -> anyhow::Result<Kek> {
        let arr: [u8; 32] = bytes
            .as_slice()
            .try_into()
            .map_err(|_| anyhow!("{} is {} bytes, not 32", what.display(), bytes.len()))?;
        Ok(Kek::from_bytes(arr))
    };
    let old = key32(
        std::fs::read(&path)
            .with_context(|| format!("reading the replayed KEK {}", path.display()))?,
        &path,
    )?;
    let next_path = control_dir.join(KEK_NEXT);
    let env = Kek::from_env()?;
    let new = match &env {
        Some(k) => k.clone(),
        None => match std::fs::read(&next_path) {
            Ok(bytes) => key32(bytes, &next_path)?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                let k = Kek::generate()?;
                write_key_file(control_dir, KEK_NEXT, &hex::decode(k.to_hex())?)?;
                k
            }
            Err(e) => return Err(e).with_context(|| format!("reading {}", next_path.display())),
        },
    };
    let mut moved = 0usize;
    let mut dropped = 0usize;
    if new.to_hex() != old.to_hex() {
        log.append(Kind::Kek, &Meta::Kek {}, &hex::decode(new.to_hex())?)
            .context("framing the restored box's own KEK")?;
        for (key, row) in rows {
            let Some(sealed) = row else { continue };
            // A legacy plaintext row is the migrating store's business, not a sealed value.
            if !gunnar_core::is_sealed(sealed) || gunnar_core::unseal(&new, key, sealed).is_ok() {
                continue;
            }
            // **A row this box wrote ITSELF, under a KEK that no longer exists** (MEASURED
            // 2026-09-17, lane TWIN-COMPLETENESS). A fresh box mints secrets of its own before
            // it is restored into — its host key at first boot, its issuer at first start, and
            // any key a request minted on first use. `prepare_replay` gives up its DOCUMENTS
            // (`FRESH_DOCS`) and its numbers, and since the same commit its secrets too; but a
            // box restored by an older build, or one whose fresh row the replay had no frame
            // for, reaches here with a row sealed under a KEK the `Kek` frame has already
            // overwritten on disk. That row opens under NEITHER key, and this used to be a `?`:
            // the restored box then REFUSED TO OPEN, every time, for ever —
            //
            //     re-sealing a restored box's secrets under its own KEK: opening fresh-only/key
            //     with the replayed KEK: … wrong key-encrypting key …
            //
            // — a box that had been restored successfully and could not start. The value is
            // unrecoverable by construction (its KEK is gone) and it is not the primary's, so
            // it is DROPPED, loudly and by name, rather than kept as the thing that bricks the
            // box. Nothing of the restore is lost: every row the twin's log carried opens under
            // the replayed KEK, which is the branch above.
            let plain = match gunnar_core::unseal(&old, key, sealed) {
                Ok(plain) => plain,
                Err(err) => {
                    tracing::warn!(
                        target: gunnar_core::observe::TARGET_TWIN,
                        key = %key,
                        %err,
                        "a secret that opens under NEITHER the replayed KEK nor this box's own: it \
                         was written by this box before the restore, under a key that is gone. \
                         DROPPED — it is not the restored primary's and it cannot be read by \
                         anyone. If something needs it, it will be minted again on first use"
                    );
                    raw.delete(key)
                        .with_context(|| format!("dropping the unopenable {key}"))?;
                    dropped += 1;
                    continue;
                }
            };
            let resealed = gunnar_core::seal(&new, key, &plain)?;
            log.append(
                Kind::Secret,
                &Meta::Secret {
                    key: key.clone(),
                    deleted: false,
                },
                &resealed,
            )
            .with_context(|| format!("framing the re-sealed {key}"))?;
            raw.put(key, &resealed)
                .with_context(|| format!("storing the re-sealed {key}"))?;
            moved += 1;
        }
    }
    if env.is_some() {
        // The box is handed its key; the dead primary's must not stay on the disk.
        match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                return Err(e)
                    .with_context(|| format!("removing the replayed KEK {}", path.display()))
            }
        }
    } else {
        std::fs::rename(&next_path, &path)
            .with_context(|| format!("installing the new KEK at {}", path.display()))?;
    }
    // The KEK's rename (or removal) is durable BEFORE the marker that owes it goes (stinky
    // row 9): a power cut between the two must find the re-seal still owed, never a box that
    // thinks it re-sealed with the replayed KEK back on its disk.
    crate::durable::sync_dir(control_dir).context("flushing the re-sealed KEK into place")?;
    crate::durable::remove_durable(root, RESEAL)?;
    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, moved, dropped, "a restored box re-sealed its secrets under a KEK of its own; the replayed KEK is gone from this disk");
    Ok(new)
}

/// A 0600 key file under `dir`, atomic and durable.
fn write_key_file(dir: &Path, name: &str, bytes: &[u8]) -> anyhow::Result<()> {
    use std::io::Write as _;
    std::fs::create_dir_all(dir)?;
    let tmp = dir.join(format!("{name}.tmp"));
    let mut options = std::fs::OpenOptions::new();
    options.write(true).create(true).truncate(true);
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        options.mode(0o600);
    }
    let mut f = options.open(&tmp)?;
    f.write_all(bytes)?;
    f.sync_all()?;
    drop(f);
    std::fs::rename(&tmp, dir.join(name))?;
    crate::durable::sync_dir(dir)
}

// ── the follower ────────────────────────────────────────────────────────────

/// What one [`Follower::step`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Progress {
    /// Applied this many entries; the twin's head is now `head`.
    Applied { count: u64, head: u64 },
    /// Nothing new on the primary.
    CaughtUp { head: u64 },
    /// The primary retains from `first`; the twin needs `wanted`. Stopped.
    SnapshotNeeded { wanted: u64, first: u64 },
    /// The hole was closed by a snapshot taken at `at`; the twin follows from `at + 1`.
    SnapshotInstalled { at: u64 },
    /// The primary's head is BELOW this twin's own — a different log. Stopped.
    Diverged {
        /// What this twin has applied.
        own: u64,
        /// What the box at the other end says its head is.
        primary_head: u64,
    },
}

/// Consecutive failed polls after which a follower whose primary has another
/// address hands back to the supervisor to re-dial.
pub const FAILOVER_AFTER: u32 = 5;

/// The twin's loop over one source into one set of stores.
pub struct Follower {
    stores: Arc<ServerStores>,
    source: Box<dyn LogSource>,
    /// Blob bytes per fetch.
    fetch_bytes: u64,
    /// **The log is one entry ahead of the stores.** The twin appends BEFORE it applies
    /// (durable first), so an apply that failed — or a process that died between the two —
    /// leaves the head entry unapplied. Set at construction (a restart does not know) and on
    /// every append; cleared once that entry applied. Measured 2026-09-15 under KVM: without
    /// it the next poll resumed at head + 1 and the twin reported `applied == head` with the
    /// failed entries' refs missing from its stores.
    reapply: AtomicBool,
    /// The run of failed polls being lived through — kept across the supervisor's re-dials
    /// ([`Follower::sharing_failures`]), so a primary that stays unreachable is said once.
    failures: Arc<Mutex<Option<Failing>>>,
    /// The peak resident set this follower last said ([`Follower::say_memory`]), bytes; 0 = never.
    said_peak: std::sync::atomic::AtomicU64,
    /// The applied marker ([`gunnar_core::TWIN_APPLIED`]) this follower last made durable.
    marked: std::sync::atomic::AtomicU64,
    /// The volume a snapshot is staged on ([`snapshot::Disk`]; a test caps one).
    volume: Arc<dyn snapshot::Volume>,
}

/// **This process's memory, from the kernel**: `(VmHWM, VmRSS)` of `/proc/self/status` in bytes —
/// the peak resident set since the process started, and the resident set now. `None` where the
/// file cannot be read (not Linux).
pub fn own_memory() -> Option<(u64, u64)> {
    let s = std::fs::read_to_string("/proc/self/status").ok()?;
    let kib = |key: &str| {
        s.lines()
            .find_map(|l| l.strip_prefix(key))
            .and_then(|v| v.split_whitespace().next())
            .and_then(|v| v.parse::<u64>().ok())
    };
    Some((kib("VmHWM:")? << 10, kib("VmRSS:")? << 10))
}

/// A following twin says its memory again once its PEAK rose this much above the figure it last
/// said — so a twin whose memory is bounded says it once, and one that keeps climbing says so in
/// steps, never more often than its memory grows.
pub const SAY_MEMORY_STEP: u64 = 32 << 20;

/// Whether a peak of `peak` bytes is news after `said` was said (0 = nothing said yet).
pub fn memory_is_news(said: u64, peak: u64) -> bool {
    said == 0 || peak >= said.saturating_add(SAY_MEMORY_STEP)
}

/// **A run of failed polls** (L1, estate-box check 3): what failed (its shape), since when,
/// how many, and when it was last said. A recharge makes a twin's primary unreachable, then
/// a blank box that does not prove itself (`BadSignature`) for minutes: every poll said
/// `poll failed; retrying` at WARN (30/min) and every hand-back `re-dialling` at WARN. The
/// first failure of a run is said at WARN; the same failure again only every [`RESAY_FAILING`];
/// the end of the run once, at INFO.
#[derive(Debug, Clone)]
pub struct Failing {
    shape: String,
    since: std::time::Instant,
    said: std::time::Instant,
    polls: u64,
}

/// While the same failure lasts, it is said again this often.
pub const RESAY_FAILING: Duration = Duration::from_secs(300);

/// An error's shape: its words without its numbers, so the same failure on another port or
/// with another nonce is the same failure.
/// Whether a poll failure is the primary refusing this twin's credential for clock SKEW —
/// [`gunnar_grpc::auth::CLOCK_SKEW_PHRASE`] in the refusal, which the gate's `Stale` status
/// carries — and not any other refusal or a transport failure.
pub(crate) fn refused_for_skew(err: &anyhow::Error) -> bool {
    format!("{err:#}").contains(gunnar_grpc::auth::CLOCK_SKEW_PHRASE)
}

fn failure_shape(err: &anyhow::Error) -> String {
    let text = format!("{err:#}");
    let mut out = String::with_capacity(text.len().min(200));
    let mut in_num = false;
    for c in text.chars().take(200) {
        if c.is_ascii_digit() {
            if !in_num {
                out.push('#');
            }
            in_num = true;
        } else {
            in_num = false;
            out.push(c);
        }
    }
    out
}

/// **The follower's end of a streamed range** (twin-streaming, 2026-10-06): each entry straight
/// into this box's OWN log as it arrives — a [`txlog::Pending`] at the primary's number, so the
/// gap rule holds piece by piece — and, once whole, [`take_logged`]. The frame does not exist
/// until its last byte is in: the length is held to the declared one in both directions
/// (`Pending::write` refuses more, `Pending::commit` refuses less), the primary hands over an
/// entry's last piece only after its payload crc checked, and an entry that never ends is cut off
/// the segment when the `Pending` drops. What this box holds of an entry in flight is the piece
/// in hand.
struct Appender<'a> {
    stores: &'a ServerStores,
    reapply: &'a AtomicBool,
    /// The entry being received: its frame in progress, its kind, meta and blob length.
    open: Option<(txlog::Pending<'a>, u16, Vec<u8>, u64)>,
    /// The sequence due next.
    next: u64,
    /// Entries appended and taken.
    count: u64,
    /// **The group commit of a store-only twin** (perf twin-apply-fsync, 2026-10-06); `None` on a
    /// box being restored, which keeps one durable frame and one full apply per entry.
    group: Option<Group>,
    /// The applied marker this follower last made durable.
    marked: &'a std::sync::atomic::AtomicU64,
}

/// Entries written into the twin's log and not yet durable, in order, with what their take needs.
struct Group {
    entries: Vec<(u64, u16, Vec<u8>)>,
    bytes: u64,
    since: std::time::Instant,
}

impl Group {
    fn new() -> Self {
        Group {
            entries: Vec::new(),
            bytes: 0,
            since: std::time::Instant::now(),
        }
    }
    fn full(&self) -> bool {
        self.entries.len() >= GROUP_ENTRIES
            || self.bytes >= GROUP_BYTES
            || self.since.elapsed() >= GROUP_AGE
    }
}

impl Appender<'_> {
    /// **End the group**: ONE `fdatasync` makes every entry of it durable in this twin's log —
    /// from here the head says so, and the head is what this twin acknowledges — then the
    /// control-plane entries among them are taken, in order, into ONE durable redb commit
    /// ([`take_group`]). A half-received entry is cut first. Nothing pending = nothing done.
    fn flush(&mut self) -> anyhow::Result<()> {
        drop(self.open.take());
        let Some(g) = self.group.as_mut() else {
            return Ok(());
        };
        if g.entries.is_empty() {
            return Ok(());
        }
        let entries = std::mem::take(&mut g.entries);
        g.bytes = 0;
        g.since = std::time::Instant::now();
        self.stores
            .txlog()
            .sync()
            .context("making a group of entries durable in this twin's log")?;
        self.reapply.store(true, Ordering::SeqCst);
        take_group(self.stores, &entries, self.marked)?;
        self.reapply.store(false, Ordering::SeqCst);
        Ok(())
    }
}

/// The name a `ControlDoc` frame's meta moves, if it is one.
fn doc_name(kind: u16, meta: &[u8]) -> Option<String> {
    if kind != Kind::ControlDoc as u16 {
        return None;
    }
    match Meta::from_bytes(meta) {
        Ok(Meta::ControlDoc { name, .. }) => Some(name),
        _ => None,
    }
}

/// The document names the fresh-box rule still holds ([`FRESH_DOCS`]); empty when none.
fn fresh_names(root: &Path) -> anyhow::Result<std::collections::BTreeSet<String>> {
    match std::fs::read(root.join(FRESH_DOCS)) {
        Ok(b) => Ok(serde_json::from_slice(&b)
            .with_context(|| format!("{} does not parse", root.join(FRESH_DOCS).display()))?),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Default::default()),
        Err(e) => Err(e.into()),
    }
}

/// **Take a group of entries that are durable in a twin's log, with ONE durable redb commit**
/// (perf twin-apply-fsync, 2026-10-06).
///
/// The control-plane writes of the whole group go into ONE redb write transaction
/// ([`gunnar_core::ControlDb::begin_group`]) and are committed once, with the applied marker
/// ([`gunnar_core::TWIN_APPLIED`]) at the group's last entry inside the same commit. A crash
/// anywhere leaves redb EXACTLY at its last durable marker — the group is in whole or not at all —
/// so the entries after the marker are re-taken in order onto exactly the state they were first
/// taken onto. That is what makes a re-take of more than one entry safe: a compare-and-swap the
/// primary LOST can only lose again when it is replayed onto the state it lost against. A write
/// that fails rolls the whole group back; the twin stops there and re-takes it next poll.
///
/// A `ControlDoc` the fresh-box rule still names is taken on its own (its `fresh_take` reads the
/// committed document and writes its own file durably): the group so far is committed first. A
/// run of data frames moves no redb page and is marked only every [`MARK_EVERY`] entries.
fn take_group(
    stores: &ServerStores,
    entries: &[(u64, u16, Vec<u8>)],
    marked: &std::sync::atomic::AtomicU64,
) -> anyhow::Result<()> {
    let db = stores.control_db();
    let mut fresh = fresh_names(stores.root())?;
    let mut done: Option<u64> = None;
    db.begin_group()?;
    // A panic inside the take must not leave the group's redb write transaction open for ever
    // (every other writer of this box's control database would wait on it, and the next take
    // would be refused "already open"): unwinding drops this, which rolls the group back. On
    // every ordinary path the group is already committed or aborted and this finds none.
    struct AbortOnDrop<'a>(&'a gunnar_core::ControlDb);
    impl Drop for AbortOnDrop<'_> {
        fn drop(&mut self) {
            self.0.abort_group();
        }
    }
    let _abort = AbortOnDrop(&db);
    let taken = (|| -> anyhow::Result<()> {
        for (seq, kind, meta) in entries {
            let (seq, kind) = (*seq, *kind);
            let fresh_doc = if fresh.is_empty() {
                None
            } else {
                doc_name(kind, meta).filter(|n| fresh.contains(n))
            };
            let applied = if let Some(name) = fresh_doc {
                if let Some(prev) = done {
                    db.commit_group(gunnar_core::TWIN_APPLIED, prev, false)?;
                } else {
                    db.abort_group();
                }
                let a = take_logged(stores, seq, kind, meta)?;
                db.set_mark(gunnar_core::TWIN_APPLIED, seq)?;
                marked.store(seq, Ordering::SeqCst);
                fresh.remove(&name);
                db.begin_group()?;
                a
            } else {
                take_logged(stores, seq, kind, meta)?
            };
            match applied {
                Applied::Done | Applied::Stored => {}
                Applied::Lost(what) => {
                    tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS")
                }
                Applied::Unknown(kind) => {
                    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply")
                }
            }
            done = Some(seq);
        }
        Ok(())
    })();
    match (taken, done) {
        (Ok(()), Some(last)) => {
            let force = last >= marked.load(Ordering::SeqCst).saturating_add(MARK_EVERY);
            // The group's one durable redb commit.
            if db
                .commit_group(gunnar_core::TWIN_APPLIED, last, force)
                .context("committing the group's control-plane writes")?
            {
                marked.store(last, Ordering::SeqCst);
            }
            Ok(())
        }
        (Ok(()), None) => {
            db.abort_group();
            Ok(())
        }
        (Err(err), _) => {
            db.abort_group();
            Err(err)
        }
    }
}

impl FrameSink for Appender<'_> {
    fn begin(&mut self, seq: u64, kind: u16, meta: &[u8], blob_len: u64) -> anyhow::Result<()> {
        if let Some((p, ..)) = &self.open {
            anyhow::bail!(
                "the primary began seq {seq} with seq {} unfinished",
                p.seq()
            );
        }
        if seq != self.next {
            return Err(anyhow!(
                "the primary sent seq {seq} where {} was due",
                self.next
            ));
        }
        // Into the twin's own log first — the primary's number.
        let pending = self.stores.txlog().begin(Some(seq), kind, meta, blob_len)?;
        self.open = Some((pending, kind, meta.to_vec(), blob_len));
        Ok(())
    }

    fn piece(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
        let (p, ..) = self
            .open
            .as_mut()
            .ok_or_else(|| anyhow!("the primary sent blob bytes outside an entry"))?;
        let seq = p.seq();
        p.write(bytes)
            .with_context(|| format!("seq {seq}: writing its blob into this box's log"))?;
        Ok(())
    }

    fn end(&mut self) -> anyhow::Result<()> {
        let (p, kind, meta, blob_len) = self
            .open
            .take()
            .ok_or_else(|| anyhow!("the primary ended an entry it never began"))?;
        if let Some(g) = self.group.as_mut() {
            // Whole in the log, durable with the rest of its group.
            let seq = p.commit_unsynced()?;
            if g.entries.is_empty() {
                g.since = std::time::Instant::now();
            }
            g.bytes += blob_len + meta.len() as u64;
            g.entries.push((seq, kind, meta));
            let full = g.full();
            self.next = seq + 1;
            self.count += 1;
            if full {
                self.flush()?;
            }
            return Ok(());
        }
        let seq = p.commit()?;
        self.reapply.store(true, Ordering::SeqCst);
        let applied = take_logged(self.stores, seq, kind, &meta)?;
        self.reapply.store(false, Ordering::SeqCst);
        match applied {
            Applied::Done => {}
            Applied::Stored => {}
            Applied::Lost(what) => {
                tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, seq, %what, "replayed a lost CAS")
            }
            Applied::Unknown(kind) => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, seq, kind, "kept a frame of a kind this build does not apply")
            }
        }
        self.next = seq + 1;
        self.count += 1;
        Ok(())
    }
}

/// **The resync's own words** for the status (twin-resync-space, 2026-10-06).
fn resyncing_why(first: u64, own: u64, dropped: Option<&snapshot::Dropped>) -> String {
    match dropped {
        Some(d) => format!("taking a full copy from its primary (it retains from {first}) — {}", d.sentence()),
        None => format!(
            "taking a full copy from its primary (it retains from {first}; this twin holds up to {own}) — its previous \
             copy is kept until the new one is whole, and it is NOT a current copy until then"
        ),
    }
}

/// **Give this twin's stale copy up for room** (twin-resync-space, 2026-10-06), called by the
/// stager ONCE when the volume runs short mid-resync. The [`snapshot::DROPPED`] marker is written
/// FIRST (durable): from then on the twin serves nothing and a crash restarts the resync. Then the
/// kept base and the log go ([`finish_giving_up`]); the head stays the twin's own. Said at WARN
/// as an ALERT, alarmed (`twin.resync_without_copy`), and written in the status. Answers the
/// bytes freed.
fn give_up_stale_copy(
    stores: &ServerStores,
    short: &snapshot::Short,
    first: u64,
    primary_head: u64,
    answered: i64,
) -> anyhow::Result<u64> {
    let root = stores.root();
    let log = stores.txlog();
    let head = log.head_seq();
    let base_at = snapshot::base_at(root).ok().flatten();
    let freed = snapshot::bytes_under(&root.join(snapshot::BASE))
        + snapshot::bytes_under(&root.join(format!("{}.old", snapshot::BASE)))
        + log.bytes();
    let d = snapshot::Dropped {
        at_unix_ms: now_ms(),
        head,
        base_at,
        freed,
    };
    snapshot::mark_dropped(root, &d)?;
    tracing::warn!(
        target: gunnar_core::observe::TARGET_TWIN,
        event = "twin.resync_without_copy",
        head,
        freed,
        staged = short.staged,
        free = short.free,
        total = short.total,
        "ALERT the full resync does not fit beside this twin's stale copy ({} staged, {} free of {}{}): {}",
        human(short.staged),
        human(short.free),
        human(short.total),
        if short.enospc { ", the volume said ENOSPC" } else { "" },
        d.sentence()
    );
    finish_giving_up(stores)?;
    set_status(
        root,
        TwinStatus {
            applied_seq: head,
            primary_head_seq: primary_head,
            state: State::Resyncing,
            why: resyncing_why(first, head, Some(&d)),
            last_answer_unix_ms: answered,
            last_level_unix_ms: 0,
        },
    );
    Ok(freed)
}

/// **The stale copy goes**: the kept base (and an `.old` one a crash left) and every entry of
/// the log — emptied at the twin's own head, so the numbers do not move. Idempotent: a give-up
/// a crash cut short is finished by calling it again.
fn finish_giving_up(stores: &ServerStores) -> anyhow::Result<()> {
    let root = stores.root();
    for dir in [
        root.join(snapshot::BASE),
        root.join(format!("{}.old", snapshot::BASE)),
    ] {
        match std::fs::remove_dir_all(&dir) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e).with_context(|| format!("giving up {}", dir.display())),
        }
    }
    let log = stores.txlog();
    let head = log.head_seq();
    if log.first_seq() <= head || log.bytes() > 0 {
        log.restart_after(head).context("giving up the stale log")?;
    }
    Ok(())
}

impl Follower {
    /// A follower of `source` into `stores`.
    pub fn new(stores: Arc<ServerStores>, source: Box<dyn LogSource>) -> Self {
        Self {
            stores,
            source,
            fetch_bytes: DEFAULT_FETCH_BYTES,
            reapply: AtomicBool::new(true),
            failures: Arc::new(Mutex::new(None)),
            said_peak: std::sync::atomic::AtomicU64::new(0),
            marked: std::sync::atomic::AtomicU64::new(0),
            volume: Arc::new(snapshot::Disk),
        }
    }

    /// Stage snapshots on `volume` instead of the box's own filesystem (the tests' capped
    /// volume, twin-resync-space 2026-10-06).
    pub fn with_volume(mut self, volume: Arc<dyn snapshot::Volume>) -> Self {
        self.volume = volume;
        self
    }

    /// Share the run of failures with the follower before this one (the supervisor's
    /// re-dial starts a new follower on the same primary: the run goes on).
    pub fn sharing_failures(mut self, failures: Arc<Mutex<Option<Failing>>>) -> Self {
        self.failures = failures;
        self
    }

    /// **The twin says its own memory** (twin-streaming, 2026-10-06), at INFO, after a poll that
    /// applied something — the first time, and again whenever the peak rose
    /// [`SAY_MEMORY_STEP`] above what it last said. A twin has no shell: this line, on its log
    /// door, is how an operator (and the UpCloud mockup) sees whether following a fill of large
    /// packs costs it the packs' size. The test2 twin that went silent on 1 GB under 256 MiB
    /// packs said nothing of the kind.
    fn say_memory(&self, head: u64) {
        let Some((peak, now)) = own_memory() else {
            return;
        };
        let said = self.said_peak.load(Ordering::Relaxed);
        if !memory_is_news(said, peak) {
            return;
        }
        self.said_peak.store(peak, Ordering::Relaxed);
        tracing::info!(
            target: gunnar_core::observe::TARGET_TWIN,
            peak_mib = peak >> 20,
            rss_mib = now >> 20,
            head,
            "twin memory: peak resident set {} MiB, {} MiB now, following through seq {head}",
            peak >> 20,
            now >> 20
        );
    }

    /// A poll answered: the end of a run of failures, said once.
    fn failures_ended(&self) {
        if let Some(f) = self
            .failures
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .take()
        {
            tracing::info!(target: gunnar_core::observe::TARGET_TWIN, polls = f.polls, seconds = f.since.elapsed().as_secs(), "the primary answers again: following on after {} failed poll(s) over {} s", f.polls, f.since.elapsed().as_secs());
        }
    }

    /// A poll failed: said at WARN when it starts a run (or every [`RESAY_FAILING`] of the
    /// same run), at DEBUG otherwise.
    fn failed(&self, err: &anyhow::Error) {
        // ★ This twin's OWN credential refused for skew: its clock or the primary's is wrong, and
        // the front is the judge — the same trigger a refusal on this box's own gate pulls,
        // rate-limited, carrying nothing from the refusal (test5's in-place upgrade, 2026-10-09:
        // the re-laid twin's clock stayed two hours off while every poll was refused).
        if refused_for_skew(err) {
            estate_clock::trigger::skew_refused(0);
        }
        let shape = failure_shape(err);
        let now = std::time::Instant::now();
        let mut g = self.failures.lock().unwrap_or_else(|p| p.into_inner());
        match g.as_mut() {
            Some(f) if f.shape == shape => {
                f.polls += 1;
                if now.duration_since(f.said) >= RESAY_FAILING {
                    f.said = now;
                    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, seconds = now.duration_since(f.since).as_secs(), "poll still failing the same way; this twin keeps asking");
                } else {
                    tracing::debug!(target: gunnar_core::observe::TARGET_TWIN, %err, polls = f.polls, "poll failed (said already)");
                }
            }
            _ => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "poll failed; retrying (said once while it fails this way)");
                *g = Some(Failing {
                    shape,
                    since: now,
                    said: now,
                    polls: 1,
                });
            }
        }
    }

    /// How many blob bytes one fetch asks for (tests ask for tiny fetches
    /// to exercise the restart-mid-range path).
    pub fn with_fetch_bytes(mut self, bytes: u64) -> Self {
        self.fetch_bytes = bytes.max(1);
        self
    }

    /// The twin's own head.
    pub fn applied_seq(&self) -> u64 {
        self.stores.txlog().head_seq()
    }

    /// Apply the log's head entry again when it may not have landed (see `reapply`).
    /// Harmless when it had: every apply is idempotent or a replayed CAS that loses the same
    /// way. An entry that still does not apply stops the twin HERE, loudly, instead of
    /// being skipped.
    fn reapply_head(&self) -> anyhow::Result<()> {
        if !self.reapply.load(Ordering::SeqCst) {
            return Ok(());
        }
        if self.stores.is_twin() {
            return self.retake_from_marker();
        }
        let log = self.stores.txlog();
        let head = log.head_seq();
        if head > 0 && head >= log.first_seq() {
            // The header and meta only: a twin re-takes a data frame without reading its bytes.
            let at = log.locate_blob(head)?;
            take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
        }
        self.reapply.store(false, Ordering::SeqCst);
        Ok(())
    }

    /// **A store-only twin re-takes what its log holds past its applied marker** (perf
    /// twin-apply-fsync, 2026-10-06) — in order, from exactly the state the marker vouches for
    /// ([`take_group`]): the log is durable before anything is taken, so after a crash the log
    /// may be a whole group ahead of redb, never behind. A twin from before the marker has none:
    /// it keeps the old rule (its head was the only entry that could be missing) and gets one.
    fn retake_from_marker(&self) -> anyhow::Result<()> {
        let log = self.stores.txlog();
        let db = self.stores.control_db();
        let head = log.head_seq();
        match db.mark(gunnar_core::TWIN_APPLIED)? {
            Some(m) if m >= head => {
                if m > head {
                    // The log was cut below what redb holds (it cannot be by this code: redb is
                    // only ever written after the log's fsync). Say so; follow from the log.
                    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, marker = m, head, "the applied marker is past this twin's log head; following on from the head");
                    db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
                }
                self.marked.store(head, Ordering::SeqCst);
            }
            Some(m) => {
                self.marked.store(m, Ordering::SeqCst);
                let from = (m + 1).max(log.first_seq());
                let mut entries =
                    Vec::with_capacity((head + 1).saturating_sub(from).min(1 << 16) as usize);
                for seq in from..=head {
                    let at = log.locate_blob(seq)?;
                    entries.push((seq, at.kind, at.meta));
                }
                take_group(&self.stores, &entries, &self.marked).with_context(|| format!("re-taking seq {from}..={head}, past this twin's applied marker {m}, before fetching anything newer"))?;
                if self.marked.load(Ordering::SeqCst) < head {
                    db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
                    self.marked.store(head, Ordering::SeqCst);
                }
            }
            None => {
                if head > 0 && head >= log.first_seq() {
                    let at = log.locate_blob(head)?;
                    take_logged(&self.stores, head, at.kind, &at.meta).with_context(|| format!("re-applying seq {head}, the head of this twin's log, before fetching anything newer"))?;
                }
                db.set_mark(gunnar_core::TWIN_APPLIED, head)?;
                self.marked.store(head, Ordering::SeqCst);
            }
        }
        self.reapply.store(false, Ordering::SeqCst);
        Ok(())
    }

    /// One poll: head, the gap rule, fetch-append-apply until level.
    ///
    /// **A poll that reached the primary and then failed says so in the status**
    /// (2026-09-21, lane T4). Only the `head()` call used to set a status on
    /// failure. Everything after it — `append_at` on a full volume, an `apply`
    /// that will not go in, a fetch that dies mid-range — propagated with `?`,
    /// and the STATUS WAS LEFT EXACTLY AS THE LAST GOOD POLL SET IT: `following`,
    /// often `caught_up`, with an empty `why`. `run` logged one `warn` and
    /// polled again. So the shape a twin's life actually ends in — its data
    /// volume filling — read as a healthy, current copy on both `twin status`
    /// and the primary's heartbeat, for as long as the box stayed up. A twin's
    /// whole job is to be trustworthy at the moment nobody has time to check.
    pub fn step(&self) -> anyhow::Result<Progress> {
        // `step_inner` sets this when it has already written a status of its
        // own for the failure it is returning (`unreachable` from `head()`,
        // `snapshot-needed` from a half-installed snapshot). Every OTHER error
        // path is one this twin used to keep quiet about.
        let mut said_why = false;
        // What THIS poll's answer said, when it got one: the primary's head and
        // when. A poll that reached the primary and then failed must record the
        // answer it got — it used to record the PREVIOUS poll's, so a twin that
        // stalled behind a primary still taking writes reported an older head
        // and an older time than the primary had actually just given it.
        let mut heard: Option<(u64, i64)> = None;
        let r = self.step_inner(&mut said_why, &mut heard);
        if let (Err(err), false) = (&r, said_why) {
            let last = status(self.stores.root());
            let (primary_head_seq, last_answer_unix_ms) = heard.unwrap_or((
                last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
                last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
            ));
            // ★ cov-13: the primary refused to let this twin follow into a volume smaller than
            // main's grown set. Not a stall: a debt, said in the primary's own words.
            let said = format!("{err:#}");
            // ★ twin-resync-space: a twin that gave its stale copy up says so on every failure
            // after it, until the resync lands.
            let no_copy = snapshot::dropped(self.stores.root())
                .map(|d| format!(" — {}", d.sentence()))
                .unwrap_or_default();
            if let Some(at) = said.find(gunnar_grpc::TWIN_TOO_SMALL) {
                set_status(
                    self.stores.root(),
                    TwinStatus {
                        applied_seq: self.applied_seq(),
                        primary_head_seq,
                        state: State::Recovering,
                        why: format!("{}{no_copy}", &said[at..]),
                        last_answer_unix_ms,
                        last_level_unix_ms: 0,
                    },
                );
                return r;
            }
            let holds = if no_copy.is_empty() {
                "what it holds stops here".to_owned()
            } else {
                format!("it holds NO copy{no_copy}")
            };
            set_status(self.stores.root(), TwinStatus {
                applied_seq: self.applied_seq(),
                primary_head_seq,
                state: State::Stalled,
                why: format!("this twin could not take what its primary sent — it is NOT a current copy, and {holds}: {err:#}"),
                last_answer_unix_ms,
                last_level_unix_ms: 0,
            });
        }
        r
    }

    fn step_inner(
        &self,
        said_why: &mut bool,
        heard: &mut Option<(u64, i64)>,
    ) -> anyhow::Result<Progress> {
        // A crash mid-install: the stores are half a snapshot; only a whole one fixes that.
        let installing = snapshot::installing(self.stores.root());
        if !installing {
            self.reapply_head()?;
        }
        let (first, primary_head, their_id, their_generation) = match self
            .source
            .head_identity_generation()
        {
            Ok(h) => h,
            Err(err) => {
                // **An unanswered poll keeps what the last answer said** — the
                // primary's head AND the time (2026-09-21, lane T4). This used
                // to write `primary_head_seq: 0`, so the moment a primary died
                // the twin that had followed it to 221 reported "a primary head
                // it last saw at 0" (MEASURED under KVM by T3) — and a restore
                // guard reading that could not tell a complete copy from an
                // empty one. The primary saying nothing now is not the primary
                // having had nothing then.
                let last = status(self.stores.root());
                set_status(
                    self.stores.root(),
                    TwinStatus {
                        applied_seq: self.applied_seq(),
                        primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
                        state: State::Unreachable,
                        why: err.to_string(),
                        last_answer_unix_ms: last
                            .as_ref()
                            .map(|s| s.last_answer_unix_ms)
                            .unwrap_or(0),
                        last_level_unix_ms: 0,
                    },
                );
                *said_why = true;
                return Err(err);
            }
        };
        let answered = now_ms();
        let own = self.applied_seq();
        *heard = Some((primary_head, answered));
        // ── the identity check (2026-09-21, lane T4) ────────────────────────
        //
        // **Before any number is believed, the box that gave it is asked which
        // log it is.** The sequence rule below catches a primary that is BEHIND
        // this twin. It cannot catch one that is AHEAD of it for the wrong
        // reason: a re-imaged main at the same address, left running for an
        // afternoon, passes the twin's head, and from then on every number
        // agrees — `first <= wanted`, `wanted <= primary_head` — and the twin
        // splices one history onto another and applies a stranger's frames over
        // the only surviving copy of its own. No arithmetic can see that. Only
        // a name can, so [`LOG_ID`] gives the log one.
        //
        // Three cases, and the middle one is the join:
        //   * both named and different — DIVERGED, and it stays stopped,
        //   * this root unnamed — pin what the source says (the first poll
        //     after `become_twin`, which dropped this box's own id),
        //   * the source unnamed — a primary older than this change. Nothing is
        //     claimed and nothing is refused; the numbers are all there is, as
        //     they were before. Saying so beats inventing a verdict.
        if let Some(theirs) = their_id.as_deref().map(str::trim).filter(|s| !s.is_empty()) {
            match log_id(self.stores.root()) {
                Some(mine) if mine != theirs => {
                    let why = format!(
                        "the box at the other end is log {theirs}; this twin holds log {mine}. Two different histories, \
                         whatever their sequence numbers say — this is a re-imaged or reinstalled primary at a familiar \
                         address, or an address that now points at another set. NOTHING is applied on top of it, and \
                         retrying cannot fix it: either point this twin at the box that holds {mine}, or — if {mine} is \
                         gone — restore a primary FROM this twin, which keeps the id, and it will follow again."
                    );
                    set_status(
                        self.stores.root(),
                        TwinStatus {
                            applied_seq: own,
                            primary_head_seq: primary_head,
                            state: State::Diverged,
                            why,
                            last_answer_unix_ms: answered,
                            last_level_unix_ms: 0,
                        },
                    );
                    return Ok(Progress::Diverged { own, primary_head });
                }
                Some(_) => {}
                None => {
                    pin_log_id(self.stores.root(), theirs)
                        .context("pinning the primary's log identity on this twin")?;
                    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, log_id = %theirs, "this twin now knows which log it follows");
                }
            }
        }
        // ── the generation fence (stinky row 18, 2026-10-10) ───────────────
        //
        // **A restored main keeps the id and the key, so the OLD main is the
        // same name too.** If it comes back after its replacement was restored
        // from this twin, and it took writes the twin never saw (or takes them
        // now), its head passes this twin's and every check above and below
        // agrees: the twin would splice the dead main's lost writes onto the
        // restored history. A restore stamps a newer [`LOG_GENERATION`]; this
        // twin pins the highest it has followed and refuses a lower one. A
        // fence that cannot be read refuses the poll rather than reading as 0.
        let pinned = pinned_generation(self.stores.root())?;
        if their_generation < pinned {
            let why = format!(
                "the box at the other end is generation {their_generation} of this log; this twin already follows generation \
                 {pinned}, a later restore of it. This is the OLD main come back after its replacement was restored: the same \
                 log id and the same key, so nothing else can tell them apart. NOTHING is applied from it — its writes since \
                 the restore belong to no history; this poll handed it the proof, so it fences itself ({LOG_FENCED}). Point \
                 this twin back at the restored main, and retire the old box."
            );
            set_status(
                self.stores.root(),
                TwinStatus {
                    applied_seq: own,
                    primary_head_seq: primary_head,
                    state: State::Diverged,
                    why,
                    last_answer_unix_ms: answered,
                    last_level_unix_ms: 0,
                },
            );
            return Ok(Progress::Diverged { own, primary_head });
        }
        // ── the lineage check (2026-09-21, lane T4) ─────────────────────────
        // **A log's head only ever goes forward.** Retention moves `first`; a
        // restore continues the same numbers; nothing in gunnar rewinds a head.
        // So a primary answering a head BELOW what this twin has already applied
        // is not the log this twin has been following — a re-imaged box, a
        // restore from somewhere else, or an address that now points at a
        // different set.
        //
        // MEASURED on master 2026-09-21, before this: that case fell straight
        // through. `wanted <= primary_head` is false, so the snapshot branch and
        // the snapshot-needed refusal are both skipped, the fetch loop's
        // condition is false, and the poll ended at the bottom of `step` setting
        // `State::Following` with an EMPTY `why`. On the primary's side the lag
        // is `head_seq.saturating_sub(applied)` = 0, so the heartbeat read `ok`.
        // BOTH ENDS reported a healthy pairing, for ever, while not one entry
        // would ever be applied again and the twin's contents belonged to
        // another box. A restore leaning on that twin is the worst shape this
        // repo knows — and re-imaging a main is a thing this estate does.
        if !installing && primary_head < own {
            let why = format!(
                "the primary answers head {primary_head}, BELOW the {own} this twin has already applied. A log's head never goes \
                 backwards, so this is NOT the log this twin was following (a re-imaged or restored primary, or an address that now \
                 points at a different set). Nothing is applied on top of it; an operator decides whether to re-join from scratch."
            );
            set_status(
                self.stores.root(),
                TwinStatus {
                    applied_seq: own,
                    primary_head_seq: primary_head,
                    state: State::Diverged,
                    why,
                    last_answer_unix_ms: answered,
                    last_level_unix_ms: 0,
                },
            );
            return Ok(Progress::Diverged { own, primary_head });
        }
        // A newer generation that passed every check above is this twin's
        // primary from now on; the pin only ever moves up.
        if their_generation > pinned {
            pin_generation(self.stores.root(), their_generation)
                .context("pinning the primary's log generation on this twin")?;
            tracing::info!(target: gunnar_core::observe::TARGET_TWIN, generation = their_generation, "this twin now follows generation {their_generation} of its log and refuses any older one");
        }
        let wanted = own + 1;
        if installing || (wanted <= primary_head && first > wanted) {
            let root = self.stores.root();
            // ★ twin-resync-space (2026-10-06): a twin makes ROOM for the resync instead of
            // filling its volume part-way (see `snapshot`'s "Room on a twin's volume").
            let twin = self.stores.is_twin();
            let mut given_up = false;
            if twin {
                // A resync this volume could not hold even alone is not asked for again until
                // the volume is larger: every snapshot holds main's writes for its whole stream.
                if let Some(was) = snapshot::too_small_total(root) {
                    if self
                        .volume
                        .space(root)
                        .is_some_and(|(total, _)| total <= was)
                    {
                        let no_copy = snapshot::dropped(root)
                            .map(|d| format!(" — {}", d.sentence()))
                            .unwrap_or_default();
                        anyhow::bail!(
                            "{} this twin's last full resync ran out of room on its {} volume with its stale copy given up; it asks main for another only once its volume is larger (main holds its writes for every snapshot stream){no_copy}",
                            gunnar_grpc::TWIN_TOO_SMALL,
                            human(was)
                        );
                    }
                }
                // A give-up a crash cut short is finished before anything is staged: the marker
                // says the copy is gone, so all of it goes.
                if snapshot::dropped(root).is_some() {
                    finish_giving_up(&self.stores)?;
                    given_up = true;
                }
                set_status(
                    root,
                    TwinStatus {
                        applied_seq: own,
                        primary_head_seq: primary_head,
                        state: State::Resyncing,
                        why: resyncing_why(first, own, snapshot::dropped(root).as_ref()),
                        last_answer_unix_ms: answered,
                        last_level_unix_ms: 0,
                    },
                );
            }
            let mut stager = snapshot::Stager::begin_on(root, Arc::clone(&self.volume))?;
            if twin {
                let stores = Arc::clone(&self.stores);
                stager = stager.making_room(
                    Box::new(move |short: &snapshot::Short| {
                        give_up_stale_copy(&stores, short, first, primary_head, answered)
                    }),
                    given_up,
                );
            }
            let fetched = self.source.snapshot(&mut stager);
            if let Err(err) = &fetched {
                if let Some(t) = err.downcast_ref::<snapshot::TooSmall>() {
                    snapshot::note_too_small(root, t)?;
                }
            }
            if fetched.context("fetching a snapshot to close the hole")? {
                let snap = stager.finish()?;
                let was_given_up = snapshot::dropped(root).is_some();
                snapshot::install(&self.stores, &snap)?;
                // The installed state IS the state through `at`.
                self.stores
                    .control_db()
                    .set_mark(gunnar_core::TWIN_APPLIED, snap.at)?;
                self.marked.store(snap.at, Ordering::SeqCst);
                self.reapply.store(false, Ordering::SeqCst);
                let whole_again = if was_given_up {
                    "; the stale copy given up for it is replaced — this twin holds a whole copy again"
                } else {
                    ""
                };
                set_status(root, TwinStatus {
                    applied_seq: snap.at,
                    primary_head_seq: primary_head.max(snap.at),
                    state: State::Following,
                    why: format!("installed a snapshot at {} (the source retained from {first}){whole_again}", snap.at),
                    last_answer_unix_ms: answered,
                    last_level_unix_ms: 0,
                });
                return Ok(Progress::SnapshotInstalled { at: snap.at });
            }
        }
        // **The marker outlives the attempt.** `snapshot()` answering `false` — a source that
        // cannot make one — used to fall straight through to the fetch loop below, because the
        // `snapshot-needed` refusal after this is guarded by the RETENTION test and a source
        // that retains everything passes it. MEASURED 2026-09-16: a twin with the install
        // marker on its disk and a bare-log source applied 29 entries onto stores `install` had
        // already wiped, and said nothing. Half a snapshot is not a state to follow on from,
        // and only a whole one replaces it.
        if installing {
            let why = match snapshot::dropped(self.stores.root()) {
                Some(d) => format!(
                    "{} — and the source cannot make a snapshot to replace it; nothing is applied on \
                     top of an empty copy",
                    d.sentence()
                ),
                None => format!(
                    "this box crashed inside a snapshot install ({}/{}) and the source cannot make a \
                     snapshot to replace it; half a snapshot is not a state to follow on from, so \
                     nothing is applied on top of it",
                    self.stores.root().display(),
                    snapshot::INSTALLING
                ),
            };
            set_status(
                self.stores.root(),
                TwinStatus {
                    applied_seq: own,
                    primary_head_seq: primary_head,
                    state: State::SnapshotNeeded,
                    why: why.clone(),
                    last_answer_unix_ms: answered,
                    last_level_unix_ms: 0,
                },
            );
            *said_why = true;
            return Err(anyhow!(why));
        }
        if wanted <= primary_head && first > wanted {
            set_status(self.stores.root(), TwinStatus {
                applied_seq: own,
                primary_head_seq: primary_head,
                state: State::SnapshotNeeded,
                why: format!("the primary retains from {first}; this twin holds up to {own} and needs {wanted} next"),
                last_answer_unix_ms: answered,
                last_level_unix_ms: 0,
            });
            return Ok(Progress::SnapshotNeeded { wanted, first });
        }
        let mut count = 0u64;
        let mut next = wanted;
        while next <= primary_head {
            // **Streamed, never gathered** (twin-streaming, 2026-10-06): each entry goes into this
            // box's own log piece by piece as it arrives and exists once its last byte is in; an
            // entry the fetch did not finish is cut off the segment when `into` drops.
            let group = self.stores.is_twin().then(Group::new);
            let mut into = Appender {
                stores: &self.stores,
                reapply: &self.reapply,
                open: None,
                next,
                count: 0,
                group,
                marked: &self.marked,
            };
            let fetched = self
                .source
                .fetch_into(next, primary_head, self.fetch_bytes, &mut into);
            // Whatever arrived whole is made durable and taken, whether or not the fetch ended well.
            let flushed = into.flush();
            let (taken, after) = (into.count, into.next);
            drop(into);
            count += taken;
            fetched?;
            flushed?;
            if after == next {
                return Err(anyhow!(
                    "the primary answered an empty range for {next}..={primary_head}"
                ));
            }
            next = after;
        }
        let head = self.applied_seq();
        set_status(
            self.stores.root(),
            TwinStatus {
                applied_seq: head,
                primary_head_seq: primary_head,
                state: State::Following,
                why: String::new(),
                last_answer_unix_ms: answered,
                last_level_unix_ms: 0,
            },
        );
        Ok(if count == 0 {
            Progress::CaughtUp { head }
        } else {
            Progress::Applied { count, head }
        })
    }

    /// Poll until `stop`; an unreachable primary is retried every `poll`, a
    /// snapshot-needed stops the loop and leaves the status saying so. What it
    /// returns says why it returned ([`Ended`]).
    pub fn run(&self, poll: Duration, stop: &AtomicBool) -> Ended {
        let hold = maintenance::hold_of(self.stores.root());
        let mut failed_in_a_row = 0u32;
        // M2: the primary answered `Unavailable [primary-growing]` — said once, waited on.
        let mut primary_growing = false;
        // ★ cov-13: the primary refused this twin as too small for main's grown set — said
        // once, waited on at the same walking pace, never re-dialled.
        let mut recovering = false;
        while !stop.load(Ordering::Relaxed) {
            // A step runs under `busy`, so `maintenance::hold` returns only once no
            // fetch/append/apply is in flight — the grow machine unmounts after that.
            let busy = hold.busy.lock().unwrap_or_else(|p| p.into_inner());
            if hold.held.load(Ordering::SeqCst) {
                drop(busy);
                self.maintenance_beat();
                nap(poll, stop);
                continue;
            }
            let stepped = self.step();
            drop(busy);
            // T1 (estate-box check 2): a poll that APPLIED something asks again at once.
            // Its `LogHead` carried the report taken BEFORE the fetch, so the primary's
            // heartbeat reads this twin one poll behind until the next ask; and a primary
            // still taking writes has more to give. Only a caught-up (or failed) poll
            // waits the interval — progress is bounded by the primary's own writes.
            let mut again = false;
            match stepped {
                Ok(Progress::SnapshotNeeded { wanted, first }) => {
                    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, wanted, first, "snapshot needed — the twin stops rather than append past a hole");
                    return Ended::SnapshotNeeded;
                }
                Ok(Progress::Diverged { own, primary_head }) => {
                    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, own, primary_head, "DIVERGED — the primary's head is below this twin's; the twin stops rather than pretend to follow a different log");
                    return Ended::Diverged;
                }
                Ok(p) => {
                    failed_in_a_row = 0;
                    self.failures_ended();
                    again = matches!(
                        p,
                        Progress::Applied { .. } | Progress::SnapshotInstalled { .. }
                    );
                    if let Progress::Applied { head, .. } = p {
                        self.say_memory(head);
                    }
                    if primary_growing {
                        primary_growing = false;
                        tracing::info!(target: gunnar_core::observe::TARGET_TWIN, "the primary's data set is back: following on");
                    }
                    if recovering {
                        recovering = false;
                        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, "this twin's volume holds main's set again: following on");
                    }
                }
                // M2 (release-3 test2 mock 2026-09-29): a primary growing its own set is not
                // a fault and not a reason to re-dial: said ONCE at INFO, waited on at a
                // walking pace, never handed back to the supervisor.
                Err(err) if format!("{err:#}").contains(gunnar_grpc::PRIMARY_GROWING) => {
                    failed_in_a_row = 0;
                    if !primary_growing {
                        primary_growing = true;
                        tracing::info!(target: gunnar_core::observe::TARGET_TWIN, "the primary is growing its data set: this twin waits and follows on after the remount");
                    }
                    for _ in 0..GROWING_WAIT_POLLS {
                        if stop.load(Ordering::Relaxed) {
                            break;
                        }
                        nap(poll, stop);
                    }
                    continue;
                }
                // ★ cov-13: owed a growth before it may follow. The primary is fine and so is
                // the connection: wait, polling (the head and this twin's report go with every
                // poll, so the primary and monetize see it heard and growable).
                Err(err) if format!("{err:#}").contains(gunnar_grpc::TWIN_TOO_SMALL) => {
                    failed_in_a_row = 0;
                    if !recovering {
                        recovering = true;
                        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "recovering: the primary will not let this twin follow until its volume holds main's grown set; it keeps polling and follows once grown");
                    }
                    for _ in 0..GROWING_WAIT_POLLS {
                        if stop.load(Ordering::Relaxed) {
                            break;
                        }
                        nap(poll, stop);
                    }
                    continue;
                }
                // T1: the connection went quiet under the call (a path that died beneath
                // it: vm1's relay restarted, the TCP back-off holding the next segment for
                // seconds). The connection is the broken thing, so it is dropped AT ONCE:
                // the supervisor dials afresh, the first address that answers.
                Err(err) if format!("{err:#}").contains(STALLED) => {
                    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, %err, "the primary said nothing on this connection: re-dialling it now");
                    return Ended::Redial;
                }
                Err(err) => {
                    failed_in_a_row += 1;
                    self.failed(&err);
                    // The primary has another address (tunnel / utility): hand
                    // back to the supervisor, which re-dials the first that answers.
                    if self.source.alternatives() && failed_in_a_row >= FAILOVER_AFTER {
                        tracing::info!(target: gunnar_core::observe::TARGET_TWIN, failed_in_a_row, "re-dialling the primary on the address that answers (tunnel or utility network)");
                        return Ended::Redial;
                    }
                }
            }
            if !again {
                nap(poll, stop);
            }
        }
        Ended::Stopped
    }

    /// **One poll while this twin grows its own set**: the status says
    /// `maintenance` FIRST (so the report on this very poll carries the word), then
    /// the primary is asked for its head only — nothing is fetched, appended or
    /// applied, so nothing touches the unmounted set.
    pub fn maintenance_beat(&self) {
        let root = self.stores.root();
        let last = status(root);
        let mut s = TwinStatus {
            applied_seq: self.applied_seq(),
            primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
            state: State::Maintenance,
            why: maintenance::WHY.to_owned(),
            last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
            last_level_unix_ms: 0,
        };
        set_status(root, s.clone());
        if let Ok((_, head)) = self.source.head() {
            s.primary_head_seq = head;
            s.last_answer_unix_ms = now_ms();
            set_status(root, s);
        }
    }
}

/// **The twin's own data-set growth** (T21, 2026-09-22). Owner ruling: the twin,
/// built from the same medium, behaves EXACTLY like main and never restarts on a
/// growth; `maintenance` is the word it reports meanwhile. The grow machine calls
/// [`hold`] where main pauses its admission gate and [`release`] where main
/// unpauses it, so the twin drains (no step in flight), unmounts, grows and
/// follows again — the SAME engine, no stop, no start.
pub mod maintenance {
    use super::*;

    /// The `why` a held twin reports.
    pub const WHY: &str = "this twin is growing its own data set (the same runtime growth as main: drained, \
                           unmounted, grown, mounted back — never restarted); it applies nothing until the set \
                           is back, then follows on from its own head";

    pub(super) struct Hold {
        pub(super) held: AtomicBool,
        pub(super) busy: Mutex<()>,
    }

    static HOLDS: OnceLock<Mutex<HashMap<PathBuf, Arc<Hold>>>> = OnceLock::new();

    pub(super) fn hold_of(root: &Path) -> Arc<Hold> {
        let mut m = HOLDS
            .get_or_init(|| Mutex::new(HashMap::new()))
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        Arc::clone(m.entry(root.to_path_buf()).or_insert_with(|| {
            Arc::new(Hold {
                held: AtomicBool::new(false),
                busy: Mutex::new(()),
            })
        }))
    }

    /// **Hold the follower of `root` for a growth** and return once no step is in
    /// flight. The status says `maintenance` from this call on, follower running or not.
    pub fn hold(root: &Path) {
        let h = hold_of(root);
        h.held.store(true, Ordering::SeqCst);
        drop(h.busy.lock().unwrap_or_else(|p| p.into_inner()));
        let last = status(root);
        set_status(
            root,
            TwinStatus {
                applied_seq: last.as_ref().map(|s| s.applied_seq).unwrap_or(0),
                primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
                state: State::Maintenance,
                why: WHY.to_owned(),
                last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
                last_level_unix_ms: 0,
            },
        );
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "maintenance: the twin's follower is held for its own data-set growth");
    }

    /// The set is back: the next poll follows again.
    pub fn release(root: &Path) {
        hold_of(root).held.store(false, Ordering::SeqCst);
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, root = %root.display(), "maintenance over: the twin follows again");
    }

    /// Is the twin at `root` held for a growth?
    pub fn is_held(root: &Path) -> bool {
        hold_of(root).held.load(Ordering::SeqCst)
    }

    #[cfg(test)]
    mod tests;
}

// ── following over the wire ─────────────────────────────────────────────────

/// **A tokio runtime that is DROPPED ON A THREAD OF ITS OWN.**
///
/// MEASURED 2026-09-23 (T19r, the live gunnar.rs estate). The twin came up,
/// printed
///
/// ```text
/// gunnar::twin: following the primary primary=10.13.7.223:50051
/// gunnar::twin: this twin's client certificate is good for 364 more day(s)
/// ```
///
/// and then died:
///
/// ```text
/// gunnar PANIC (exit 101): panicked at tokio/src/runtime/blocking/shutdown.rs:51:21:
/// Cannot drop a runtime in a context where blocking is not allowed.
/// This happens when a runtime is dropped from within an asynchronous context.
/// Kernel panic - not syncing: Attempted to kill init! exitcode=0x00006500
/// ```
///
/// [`GrpcSource`] builds its own current-thread runtime and keeps it, because
/// the follower's `head`/`fetch`/`snapshot` are blocking calls on a blocking
/// loop. Dropping that runtime is a BLOCKING operation — it joins the runtime's
/// threads — and tokio refuses it from inside another runtime's async context.
/// The follower is constructed and dropped from the server, which is async, so
/// every path that drops the source panicked. On the appliance gunnar is PID 1,
/// so the panic took the kernel with it: a twin that could not be followed was
/// a box that could not stay alive.
///
/// The fix is the documented one and the smallest: the runtime is moved out in
/// `Drop` and dropped on a thread created for that purpose, where blocking is
/// allowed. `shutdown_background` was the other candidate and is not used — it
/// may leak the runtime's worker threads, and a follower that is rebuilt on
/// every reconnect would leak one set per attempt.
struct OffThreadRuntime(Option<tokio::runtime::Runtime>);

impl OffThreadRuntime {
    fn new(rt: tokio::runtime::Runtime) -> Self {
        OffThreadRuntime(Some(rt))
    }

    /// The runtime, which is present for the whole of this value's life: only
    /// [`Drop`] takes it, and nothing can call this afterwards.
    fn get(&self) -> &tokio::runtime::Runtime {
        self.0
            .as_ref()
            .expect("the follower's runtime is taken only by Drop")
    }

    fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
        self.get().block_on(f)
    }
}

impl Drop for OffThreadRuntime {
    fn drop(&mut self) {
        let Some(rt) = self.0.take() else { return };
        // The thread is not joined: joining it here would block in exactly the
        // context this exists to avoid.
        // A28 M18, stays a thread: a runtime dropped off the async context, deliberately not joined (see below).
        match std::thread::Builder::new()
            .name("twin-rt-drop".to_string())
            .spawn(move || drop(rt))
        {
            Ok(_) => {}
            Err(e) => {
                // A box that cannot spawn a thread is in no state to follow
                // anything. Say so and LEAK the runtime: panicking in a
                // destructor aborts the process, which is the failure this
                // whole type exists to stop.
                tracing::error!(
                    target: gunnar_core::observe::TARGET_TWIN, error = %e,
                    "the follower's runtime could not be dropped on a thread of its own; it is leaked rather than dropped here"
                );
            }
        }
    }
}

/// `Mirror.LogHead` / `Mirror.LogFetch` against the primary, signed with the
/// twin's own key by the one client signer (`gunnar_grpc::client_signer`).
pub struct GrpcSource {
    rt: OffThreadRuntime,
    client: Mutex<
        gunnar_proto::mirror_client::MirrorClient<gunnar_grpc::client_signer::ControlChannel>,
    >,
    signer: gunnar_core::ssh_key::PrivateKey,
    /// The twin's half of the dual heartbeat, sent on every `LogHead`.
    reporter: Option<Box<dyn Fn() -> gunnar_proto::TwinReport + Send + Sync>>,
    /// This twin's data root — where [`PRIMARY_KEY`] is pinned and read back.
    root: std::path::PathBuf,
    /// The primary named more than one replication host (`<tunnel>,<utility>`):
    /// a follower that keeps failing on this one should re-dial.
    alternatives: bool,
}

fn say(status: impl std::fmt::Display) -> anyhow::Error {
    anyhow!("{status}")
}

impl GrpcSource {
    /// A source for the primary at `at` (`host:port`), as `signer`, for the twin
    /// whose data root is `root`.
    ///
    /// **`root` is not decoration**: it is where the key this primary answers
    /// with is pinned on the first poll and read back on every one after
    /// ([`PRIMARY_KEY`]). A source that could not say which twin it belongs to
    /// could not enforce anything, so it is a parameter rather than an optional
    /// builder — there is no such thing here as a follower without a data root.
    ///
    /// # Mutual TLS whenever the primary is another machine
    ///
    /// The `root` pinning above answers **who** and **which history** — the
    /// primary's sshsig over `gunnar-log-head-v1`, TOFU-pinned, and the
    /// `log_id` carried on every poll. Those are settled and nothing here
    /// re-solves them. What they do not answer is **confidentiality**: this
    /// channel carries the primary's transaction log, and *seq 1 of that log is
    /// the KEK*. A signed answer proves the bytes are genuine; it does not stop
    /// anyone on the path from reading every sealed secret in the estate as
    /// they go by. Until 2026-09-21 they went by in clear text.
    ///
    /// So the channel is encrypted too, and the rule has no flag:
    ///
    /// * a **loopback** primary (the test harness, a solo box following itself)
    ///   stays plaintext — the bytes never reach a wire;
    /// * any **other** primary needs [`ClientTrust`](gunnar_grpc::client_signer::ClientTrust)
    ///   from `GUNNAR_CLIENT_CERT`/`GUNNAR_CLIENT_KEY`/`GUNNAR_CA`, and a
    ///   follower without it REFUSES to start rather than following in the
    ///   clear. `plain_channel` refuses the same target for the same reason;
    ///   this refusal exists so the twin says which three files are missing
    ///   instead of failing at the transport with the address alone.
    ///
    /// The channel is built **once** and reused for every poll: `connect_lazy`
    /// hands back a `Channel` that opens its HTTP/2 connection on first use and
    /// keeps it, so the 2-second poll loop pays one TLS handshake per
    /// connection and not one per poll. Measured at ~0.5 ms, a handshake per
    /// poll would still be under a tenth of a percent of the poll budget — but
    /// it would also be far more than the per-poll signature verification the
    /// liveness half already does, and there is no reason to spend it.
    pub fn connect(
        at: &str,
        signer: gunnar_core::ssh_key::PrivateKey,
        root: &Path,
    ) -> anyhow::Result<Self> {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
        let loopback = gunnar_grpc::client_signer::is_loopback_target(at);
        // **Where the credential comes from, in order.** The PAIRED one first —
        // minted at `twin certify`, in this box's data root, the only shape the
        // one medium allows — then the operator's environment. Paired, the twin
        // dials the primary's MUTUAL REPLICATION LISTENER (its host, the port it
        // reported) and checks the fixed service name, even on loopback: the
        // credential is there, and a lane that encrypts only sometimes is a lane
        // whose tests prove the other branch.
        let mut alternatives = false;
        let (trust, dial, domain) = match crate::twin_tls::load(root)? {
            Some((trust, port)) => {
                // The hosts the primary REPORTED at `twin certify` win over the
                // host this twin was joined to: `<tunnel>,<utility>` while its
                // replication listener also answers on `wg0`. The overlay is
                // preferred; the first that answers is dialled, so a tunnel that
                // is down falls back to the utility network (2026-09-24).
                let candidates = crate::twin_tls::load_host(root)
                    .map(|r| crate::twin_tls::hosts(&r))
                    .filter(|h| !h.is_empty())
                    .unwrap_or_else(|| vec![gunnar_grpc::client_signer::tls_domain_for(at)]);
                alternatives = candidates.len() > 1;
                let dial = pick_replication_dial(&candidates, port, DIAL_PROBE);
                (
                    Some(trust),
                    dial,
                    gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
                )
            }
            None => (
                gunnar_grpc::client_signer::ClientTrust::from_env().map_err(|e| anyhow!(e))?,
                at.to_owned(),
                gunnar_grpc::client_signer::tls_domain_for(at),
            ),
        };
        let channel = {
            let _g = rt.enter();
            match (loopback, &trust) {
                (_, Some(trust)) => {
                    // **Before the first poll, not after the lapse.** This box
                    // has no shell and its only door is the plane this
                    // certificate opens, so a lapse is not a degraded service:
                    // it is a box nobody can repair. The grade refuses while
                    // the certificate is STILL VALID, which is the only moment
                    // at which anyone can act on it.
                    let now = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map(|d| d.as_millis() as i64)
                        .unwrap_or(0);
                    match gunnar_core::expiry::grade_pem(now, &trust.cert) {
                        Ok(expiry) => {
                            let line = expiry.line("this twin's client certificate");
                            if expiry.is_refusal() {
                                anyhow::bail!("{line}");
                            }
                            match expiry.grade {
                                gunnar_core::expiry::Grade::Renew => {
                                    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, days_left = expiry.days_left, "{line}")
                                }
                                _ => {
                                    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, days_left = expiry.days_left, "{line}")
                                }
                            }
                        }
                        // Unreadable is NOT waved through: bytes that are not a
                        // certificate are the same operational fact as a
                        // certificate that cannot be used.
                        Err(err) => anyhow::bail!(
                            "this twin's client certificate ({}) could not be read, so nothing \
                             can say how long it is good for: {err}",
                            gunnar_grpc::client_signer::CLIENT_CERT_ENV
                        ),
                    }
                    tracing::info!(
                        target: gunnar_core::observe::TARGET_TWIN,
                        %at, %dial, %domain, suite = gunnar_grpc::preferred_bulk_suite(),
                        "following the primary over MUTUAL TLS 1.3"
                    );
                    // **THE BOUNDARY A TWIN DIES AT.** MEASURED 2026-09-23:
                    // the estate's twin took its role, held this certificate,
                    // started following, and panicked — and nothing on the
                    // narration wire had said any of it, because the wire went
                    // quiet at `end of hand-off` six seconds earlier. This is
                    // the line that makes the death legible: WHICH primary,
                    // over WHICH name, with a certificate graded above.
                    crate::narrate::at(
                        crate::narrate::Stage::Follow,
                        &format!(
                            "primary {at} dialled as {dial} (server name {domain}) over mutual \
                             TLS 1.3; this twin's client certificate is held and in grade"
                        ),
                    );
                    gunnar_grpc::client_signer::mutual_channel_with(
                        &dial,
                        trust,
                        &domain,
                        kept_alive(),
                    )
                    .map_err(|e| anyhow!(e))?
                }
                (true, None) => {
                    tracing::info!(
                        target: gunnar_core::observe::TARGET_TWIN, %at,
                        "following a LOOPBACK primary in the clear: the bytes never leave this box"
                    );
                    gunnar_grpc::client_signer::plain_channel_with(at, kept_alive())
                        .map_err(|e| anyhow!(e))?
                }
                (false, None) => {
                    return Err(anyhow::Error::new(AwaitingCertificate).context(format!(
                        "this twin's primary is {at}, which is another machine, and no client \
                     certificate is configured on this box — so the only way to follow it would \
                     be in the clear. It will not. The log this stream carries holds the \
                     estate's sealed secrets and the KEK as seq 1.\n\n\
                     Set all three of {}, {} and {} to the files this box was enrolled with, and \
                     make sure the primary is reached over the WireGuard address its listener \
                     bound (the certificate carries that address, not the tunnel's name).",
                        gunnar_grpc::client_signer::CLIENT_CERT_ENV,
                        gunnar_grpc::client_signer::CLIENT_KEY_ENV,
                        gunnar_grpc::client_signer::CA_ENV,
                    )))
                }
            }
        };
        Ok(Self {
            rt: OffThreadRuntime::new(rt),
            client: Mutex::new(gunnar_proto::mirror_client::MirrorClient::new(channel)),
            signer,
            reporter: None,
            root: root.to_path_buf(),
            alternatives,
        })
    }
}

/// **How long a `Mirror.LogHead` may take** before the call is STALLED (T1,
/// estate-box check 2). A head is a few hundred bytes; on a healthy link it
/// answers in milliseconds. A connection whose path died under it (vm1's relay
/// restarted beneath the tunnel) answered after 8–57 s of TCP back-off instead.
pub const HEAD_WITHIN: Duration = Duration::from_secs(3);

/// **How long a `Mirror.LogFetch` stream may go without its next piece.**
pub const FETCH_IDLE_WITHIN: Duration = Duration::from_secs(15);

/// **How long a `Mirror.SnapshotFetch` stream may go without its next piece**
/// (r5.2, the r5.1 integration test: a twin whose primary died or restarted
/// mid-snapshot hung for minutes on a half-open connection). Past it the call
/// is STALLED: the loop cuts the connection and re-dials.
///
/// **60 s, not 15** (wiring tracer B 2026-10-08): a LIVE primary sends nothing
/// while it walks a repository with no reachability index or takes its cut,
/// and past 15 s the twin re-dialled and asked a NEW cut — on such a repository
/// it never installed. A dead path is the channel's to find, by its HTTP/2
/// keepalive ([`KEEPALIVE_EVERY`], [`KEEPALIVE_WITHIN`]): about 30 s, and the
/// next step's `Mirror.LogHead` ([`HEAD_WITHIN`]) then re-dials. The primary
/// waits 120 s on the twin (`SNAPSHOT_STALL_TIMEOUT`).
pub const SNAPSHOT_IDLE_WITHIN: Duration = Duration::from_secs(60);

/// **HTTP/2 keepalive on the twin's channels to its primary**: a PING every
/// [`KEEPALIVE_EVERY`], idle or not, and the connection is dropped when one is
/// not answered within [`KEEPALIVE_WITHIN`] — a dead path is found by the
/// channel itself, not by a call that waits on it.
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(10);
pub const KEEPALIVE_WITHIN: Duration = Duration::from_secs(20);

/// The word a stalled call's error carries: the loop re-dials on it at once.
pub const STALLED: &str = "[twin-primary-stalled]";

/// **The twin's channel to its primary, kept alive** ([`KEEPALIVE_EVERY`],
/// [`KEEPALIVE_WITHIN`], idle too): a path that died under the connection is
/// found by an unanswered PING and the connection dropped, so the next call
/// dials afresh instead of writing into a half-open socket.
fn kept_alive(
) -> impl FnOnce(gunnar_grpc::client_signer::Endpoint) -> gunnar_grpc::client_signer::Endpoint {
    gunnar_grpc::client_signer::keep_alive(KEEPALIVE_EVERY, KEEPALIVE_WITHIN)
}

/// How long a replication host gets to accept a TCP connection before the
/// next one is tried.
pub const DIAL_PROBE: Duration = Duration::from_secs(3);

/// `host:port` for a host (bracketing IPv6).
fn dial_of(host: &str, port: u16) -> String {
    if host.contains(':') {
        format!("[{host}]:{port}")
    } else {
        format!("{host}:{port}")
    }
}

/// **The replication host to dial**: the first of `candidates` (in the
/// primary's order of preference — the overlay first) that accepts a TCP
/// connection within `probe`; the first one when none does (the follower
/// retries, and says why).
pub fn pick_replication_dial(candidates: &[String], port: u16, probe: Duration) -> String {
    for host in candidates {
        let dial = dial_of(host, port);
        let Ok(addr) = dial.parse::<std::net::SocketAddr>() else {
            continue;
        };
        if std::net::TcpStream::connect_timeout(&addr, probe).is_ok() {
            return dial;
        }
        tracing::info!(target: gunnar_core::observe::TARGET_TWIN, %dial, "the primary's replication door does not answer here; trying its next address");
    }
    dial_of(
        candidates
            .first()
            .map(String::as_str)
            .unwrap_or("127.0.0.1"),
        port,
    )
}

impl GrpcSource {
    /// Carry `report()` on every `LogHead` (the dual heartbeat).
    pub fn with_reporter(
        mut self,
        report: impl Fn() -> gunnar_proto::TwinReport + Send + Sync + 'static,
    ) -> Self {
        self.reporter = Some(Box::new(report));
        self
    }

    /// **Did the box this twin agreed to follow sign this answer?** — the whole
    /// of the twin's half of the pairing, run before a single number out of it
    /// is believed.
    ///
    /// Three states, and only one of them is new trust:
    ///
    /// * **pinned, and the answer is signed by it** — the ordinary case;
    /// * **pinned, and it is not** — refused, by name. A different key, a
    ///   signature that does not verify, a replayed answer from an earlier poll
    ///   (the nonce is inside the signature), or no signature at all. The poll
    ///   fails, so the status goes `stalled` with this reason and the twin
    ///   applies nothing;
    /// * **not pinned yet** — trust on first use. The key is written down and
    ///   every answer after this one is measured against it. An UNSIGNED answer
    ///   here is left unpinned rather than pinned as "no key": a primary too old
    ///   to sign must keep working, and a twin that recorded "this primary has
    ///   no key" would never start requiring one.
    fn check_answer(
        &self,
        nonce: &str,
        head: &gunnar_proto::LogHeadResponse,
    ) -> anyhow::Result<()> {
        let offered = head.answer_key.trim();
        match pinned_primary_key(&self.root) {
            Some(pinned) => {
                anyhow::ensure!(
                    !head.answer_sig.trim().is_empty(),
                    "the box at this address answered UNSIGNED, and this twin has pinned the key its primary answers with \
                     ({pinned}). An answer nobody signed is an answer from nobody: this twin applies nothing on it."
                );
                anyhow::ensure!(
                    offered == pinned,
                    "the box at this address now answers with key {offered}; this twin follows the box that answers with \
                     {pinned}. The address is the same and the box is not: nothing is applied. If the primary was rebuilt \
                     rather than restored, it is a different history too — re-join this twin deliberately, or restore a \
                     primary FROM it."
                );
                anyhow::ensure!(
                    gunnar_grpc::auth::verify_twin_answer_at(&pinned, &head.answer_sig, nonce, &head.log_id, head.first_seq, head.head_seq, head.generation),
                    "the answer from this address does not verify against the key this twin pinned ({pinned}) — the signature \
                     is not over this poll's challenge and these numbers (and this generation). A replayed or edited answer is not an answer."
                );
                Ok(())
            }
            None if offered.is_empty() => Ok(()),
            None => {
                anyhow::ensure!(
                    gunnar_grpc::auth::verify_twin_answer_at(offered, &head.answer_sig, nonce, &head.log_id, head.first_seq, head.head_seq, head.generation),
                    "the box at this address offered key {offered} with a signature that does not verify against it; this twin \
                     pins nothing it has not seen prove itself once"
                );
                pin_primary_key(&self.root, offered)
            }
        }
    }
}

// ── the log's size alarm (2026-09-22, the owner's ruling) ───────────────────

/// The alarm's `event` name on `Admin.Health` and in the log line.
pub const TXLOG_SIZE_ALARM: &str = "txlog.size_over_cap";

/// `GUNNAR_TXLOG_ALARM_BYTES`: the cap, in bytes, over which [`TXLOG_SIZE_ALARM`] is raised.
/// Unset = [`TXLOG_ALARM_DEFAULT_PERCENT`] of the data volume.
pub const TXLOG_ALARM_ENV: &str = "GUNNAR_TXLOG_ALARM_BYTES";

/// **The default cap: a quarter of the data volume.** Chosen from the restore's own arithmetic,
/// not a round number: `room_for_restore` asks for [`RESTORE_BYTES_PER_LOG_BYTE`] × the log free
/// on the box being restored. At a quarter of the volume, a restore onto the SAME hardware needs
/// half its disk free for the log alone — the last point at which "restore onto a box like this
/// one" is still comfortably true. Past it the operator is losing that without being told, which
/// is what this alarm exists to say. Relative, so a 64 GB lab box and a 20 TB appliance are both
/// told at the same fraction.
pub const TXLOG_ALARM_DEFAULT_PERCENT: u64 = 25;

/// The cap in force for the data root at `root`, and how it was decided (for the alarm's
/// sentence). `env` is `GUNNAR_TXLOG_ALARM_BYTES` as read (passed in so a test does not race the
/// process environment). `None` = no cap can be decided (the variable unparseable is refused
/// loudly by the caller; an unmeasurable volume with no variable has no default to apply).
pub fn txlog_alarm_cap(
    env: Option<&str>,
    volume_total: u64,
) -> Result<Option<(u64, String)>, String> {
    match env.map(str::trim).filter(|v| !v.is_empty()) {
        Some(v) => v
            .parse::<u64>()
            .map(|b| Some((b, format!("{TXLOG_ALARM_ENV}={b}"))))
            .map_err(|_| format!("{TXLOG_ALARM_ENV}={v:?} is not a number of bytes; the log's size is NOT watched")),
        None if volume_total == 0 => Ok(None),
        None => Ok(Some((
            volume_total / 100 * TXLOG_ALARM_DEFAULT_PERCENT,
            format!("the default, {TXLOG_ALARM_DEFAULT_PERCENT}% of the {volume_total}-byte data volume"),
        ))),
    }
}

/// **[`TXLOG_SIZE_ALARM`] when the log's bytes on disk pass `cap`** — a warning, never an action:
/// nothing here deletes, rolls or retains a frame (the operator's `Mirror.LogRetain` is the one
/// door that does, and it is guarded by the twin's position).
pub fn txlog_size_alarm(log_bytes: u64, cap: u64, how: &str) -> Option<gunnar_proto::HealthAlarm> {
    (log_bytes > cap).then(|| gunnar_proto::HealthAlarm {
        event: TXLOG_SIZE_ALARM.into(),
        count: 1,
        since_unix_ms: 0,
        detail: format!(
            "the transaction log holds {log_bytes} bytes on disk, over the cap of {cap} ({how}). \
             Nothing has been deleted and nothing will be: let every twin catch up, then retain \
             (`gunnar control twin retain --at <primary>`), or raise {TXLOG_ALARM_ENV}. A log this \
             size needs {} bytes free on any box restored from it",
            log_bytes.saturating_mul(RESTORE_BYTES_PER_LOG_BYTE)
        ),
    })
}

/// **What this twin tells its primary on each poll**: its applied head, its
/// data volume (one `statvfs` of the data root — used = total − free-to-root,
/// so the reserved blocks count as used, the same view `df` gives), and the
/// follower's state word. A `statvfs` that fails reports 0 / 0, which the
/// primary shows as fill 0 of an unknown volume, never as a full one.
pub fn measure_report(stores: &ServerStores) -> gunnar_proto::TwinReport {
    // Held for its own growth, the data root may be the guard tmpfs over an unmounted set:
    // its size is not the twin's volume, so the report says "unknown" (0 / 0), never a size.
    let (total, used) = if maintenance::is_held(stores.root()) {
        (0, 0)
    } else {
        volume_bytes(stores.root())
    };
    let (cpu_millicores_busy, cpu_millicores_total) = own_load();
    gunnar_proto::TwinReport {
        applied_seq: stores.txlog().head_seq(),
        disk_total_bytes: total,
        disk_used_bytes: used,
        state: status(stores.root())
            .map(|s| s.state.to_string())
            .unwrap_or_default(),
        cpu_millicores_busy,
        cpu_millicores_total,
    }
}

/// **This box's own `Mirror.LogHead`**, unsigned: its log's numbers and, on a twin, what
/// its follower last saw of the primary. `Mirror.LogHead` signs it; a twin's Fleet report
/// reviews it (G12).
pub fn own_head(stores: &ServerStores) -> gunnar_proto::LogHeadResponse {
    let log = stores.txlog();
    let mut head = gunnar_proto::LogHeadResponse {
        head_seq: log.head_seq(),
        first_seq: log.first_seq(),
        bytes: log.bytes(),
        // **Which log these numbers are about** (2026-09-21, lane T4).
        // Answered on every poll, by a twin as well as a primary: a twin's
        // log IS its primary's, so it carries the same id, and that is what
        // lets `control twin replay` hand it to the fresh box it restores.
        log_id: crate::twin::log_id(stores.root()).unwrap_or_default(),
        ..Default::default()
    };
    // **A box with a `twin.json` is a twin, whatever its follower is doing**
    // (2026-09-21, lane T4). This used to read `twin` off the follower's
    // in-process status, which is `None` until the first successful poll —
    // so a twin that had just restarted, or whose follower could not start,
    // answered `twin = false`, and every reader took it for a primary:
    // `control twin replay` skipped its staleness warning entirely and
    // restored from it without a word, and `twin catchup`'s not-a-primary
    // refusal did not fire.
    if crate::twin::is_twin_root(stores.root()) {
        head.twin = true;
        // restore-after-snapshot: the base this twin keeps — everything up to its `at` is
        // there and its log starts at `at + 1`; a restore streams it first. Said only while
        // the copy is whole (a twin mid-resync holds no base to restore from).
        if crate::twin::snapshot::not_whole(stores.root()).is_none() {
            if let Ok(Some(at)) = crate::twin::snapshot::base_at(stores.root()) {
                head.snapshot_at = at;
                head.snapshot_bytes = crate::twin::snapshot::base_bytes(stores.root());
            }
        }
        match crate::twin::status(stores.root()) {
            Some(twin) => {
                head.applied_seq = twin.applied_seq;
                head.primary_head_seq = twin.primary_head_seq;
                head.caught_up = twin.caught_up();
                head.state = twin.state.to_string();
                head.why = twin.why;
                head.last_answer_unix_ms = twin.last_answer_unix_ms;
                head.last_level_unix_ms = twin.last_level_unix_ms;
                head.answer_ago_ms = if twin.last_answer_unix_ms > 0 {
                    crate::twin::now_ms()
                        .saturating_sub(twin.last_answer_unix_ms)
                        .max(0) as u64
                } else {
                    0
                };
            }
            None => {
                // Nothing has polled yet in this process. The applied head
                // is real (it is on disk); everything about the PRIMARY is
                // unknown, and saying so is the whole point.
                head.applied_seq = log.head_seq();
                head.state = "idle".to_owned();
                head.why = "this box is a twin (twin.json is on its data root) but its follower has not \
                            reported since this process started: the primary's head, the lag and the age \
                            of this copy are all UNKNOWN"
                    .to_owned();
            }
        }
    }
    head
}

/// **The twin's own load, for its poll** (2026-10-05): `/proc/loadavg`'s
/// 1-minute load ×1000 and the cores this process may run on ×1000 — the
/// readers `Admin.Capacity` uses (`capacity::machine_fields`), the same
/// approximation it states. Each absent when it could not be read, never 0.
pub fn own_load() -> (Option<u64>, Option<u64>) {
    let busy = crate::dashboard::probe::load1_value()
        .ok()
        .map(|l| (l * 1000.0).round().max(0.0) as u64);
    let total = std::thread::available_parallelism()
        .ok()
        .map(|n| n.get() as u64 * 1000);
    (busy, total)
}

/// `(total, used)` bytes of the filesystem holding `path`; `(0, 0)` when unmeasurable.
pub fn volume_bytes(path: &Path) -> (u64, u64) {
    match rustix::fs::statvfs(path) {
        Ok(v) => {
            let frsize = if v.f_frsize == 0 {
                v.f_bsize
            } else {
                v.f_frsize
            };
            let total = v.f_blocks.saturating_mul(frsize);
            let free = v.f_bfree.saturating_mul(frsize);
            (total, total.saturating_sub(free))
        }
        Err(_) => (0, 0),
    }
}

impl LogSource for GrpcSource {
    fn alternatives(&self) -> bool {
        self.alternatives
    }
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        let (first, head, _) = self.head_and_identity()?;
        Ok((first, head))
    }

    /// The numbers AND the name, out of one answer — see
    /// [`LogSource::head_and_identity`]. This is the only override, because this
    /// is the only source that reached across a wire to a box that could be
    /// anything.
    fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
        let (first, head, id, _) = self.head_identity_generation()?;
        Ok((first, head, id))
    }

    fn head_identity_generation(&self) -> anyhow::Result<(u64, u64, Option<String>, u64)> {
        let mut client = self.client.lock().expect("twin client").clone();
        let report = self.reporter.as_ref().map(|r| r());
        // The newest generation this twin has followed, relayed on every poll: an
        // old main that answers it fences itself ([`LOG_FENCED`]).
        let generation_proof = generation_proof(&self.root).unwrap_or_else(|err| {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "the kept generation proof does not read; relaying none");
            None
        });
        // **A fresh challenge per poll.** The primary signs it back, so this
        // answer is bound to this poll: a recorded one cannot be replayed later,
        // however convincing its numbers.
        let nonce = {
            use gunnar_core::ssh_key::rand_core::Rng as _;
            let mut bytes = [0u8; 16];
            gunnar_core::os_rng().fill_bytes(&mut bytes);
            bytes.iter().map(|b| format!("{b:02x}")).collect::<String>()
        };
        let req = gunnar_grpc::client_signer::authed(
            &self.signer,
            gunnar_proto::LogHeadRequest {
                report,
                nonce: nonce.clone(),
                generation_proof,
            },
        );
        let head = self
            .rt
            .block_on(async { tokio::time::timeout(HEAD_WITHIN, client.log_head(req)).await })
            .map_err(|_| anyhow!("{STALLED} the primary said nothing to Mirror.LogHead within {} s on this connection", HEAD_WITHIN.as_secs()))?
            .map_err(say)?
            .into_inner();
        self.check_answer(&nonce, &head)?;
        // Signed by the box this twin follows: if it names a newer generation, it
        // is the proof an old main is fenced with (stinky row 18, the main's half).
        if let Err(err) = remember_generation_proof(&self.root, &nonce, &head) {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "keeping the restored main's signed generation to relay it");
        }
        let id = Some(head.log_id).filter(|s| !s.trim().is_empty());
        Ok((head.first_seq, head.head_seq, id, head.generation))
    }

    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        let mut whole = Gather::default();
        self.fetch_into(from, to, max_bytes, &mut whole)?;
        Ok(whole.out)
    }

    /// **Each chunk into `into` as it arrives** (twin-streaming, 2026-10-06). This used to glue
    /// every entry's pieces back into one `Vec` — a 256 MiB pack was 256 MiB on the twin before a
    /// byte of it reached the disk, and a 1 GiB twin under such packs went silent. Now the twin
    /// holds the piece in hand; HTTP/2 flow control holds the primary back while the twin writes.
    fn fetch_into(
        &self,
        from: u64,
        to: u64,
        max_bytes: u64,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        let mut client = self.client.lock().expect("twin client").clone();
        let req = gunnar_grpc::client_signer::authed(
            &self.signer,
            gunnar_proto::LogFetchRequest {
                from,
                to,
                max_bytes,
                from_offset: 0,
            },
        );
        let quiet = || {
            anyhow!("{STALLED} the primary sent nothing of Mirror.LogFetch {from}..={to} for {} s on this connection", FETCH_IDLE_WITHIN.as_secs())
        };
        self.rt.block_on(async {
            let mut stream = tokio::time::timeout(FETCH_IDLE_WITHIN, client.log_fetch(req))
                .await
                .map_err(|_| quiet())?
                .map_err(say)?
                .into_inner();
            feed_log_stream(
                &mut stream,
                Some((FETCH_IDLE_WITHIN, &quiet)),
                &|status| say(status),
                into,
            )
            .await
        })
    }

    fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
        let mut client = self.client.lock().expect("twin client").clone();
        let req = gunnar_grpc::client_signer::authed(
            &self.signer,
            gunnar_proto::SnapshotFetchRequest::default(),
        );
        // ★ No piece for SNAPSHOT_IDLE_WITHIN is a primary that died or restarted under the
        // stream (a half-open connection): STALLED, so the loop cuts it and re-dials (r5.2).
        let quiet = || {
            anyhow!("{STALLED} the primary sent nothing of Mirror.SnapshotFetch for {} s on this connection", SNAPSHOT_IDLE_WITHIN.as_secs())
        };
        self.rt.block_on(async {
            let mut stream = tokio::time::timeout(SNAPSHOT_IDLE_WITHIN, client.snapshot_fetch(req))
                .await
                .map_err(|_| quiet())?
                .map_err(say)?
                .into_inner();
            // Each piece goes to the staging file as it arrives and is dropped: the twin holds
            // one piece of the snapshot at a time, and pulls the next only when it has written
            // this one (HTTP/2 flow control is the backpressure). The bound is on the WAIT for
            // a piece, never on the twin's own write of it.
            while let Some(chunk) = tokio::time::timeout(SNAPSHOT_IDLE_WITHIN, stream.message())
                .await
                .map_err(|_| quiet())?
                .map_err(say)?
            {
                anyhow::ensure!(
                    chunk.seq == 0,
                    "the primary's snapshot stream carried seq {}",
                    chunk.seq
                );
                let kind = u16::try_from(chunk.kind)
                    .map_err(|_| anyhow!("snapshot piece of kind {}", chunk.kind))?;
                into.feed(snapshot::Piece {
                    kind,
                    meta: &chunk.meta,
                    blob_len: chunk.blob_len,
                    blob: &chunk.blob,
                    last: chunk.last,
                })?;
            }
            Ok::<_, anyhow::Error>(())
        })?;
        anyhow::ensure!(
            into.terminated().is_some(),
            "the primary's snapshot stream carried no SnapshotAt terminator"
        );
        Ok(true)
    }
}

/// **The poll's wait, cut short by `stop`**: slept in slices, so a stop (a re-point,
/// the wind-down) is heard within [`NAP_SLICE`] instead of a whole poll later.
fn nap(poll: Duration, stop: &AtomicBool) {
    let until = std::time::Instant::now() + poll;
    while !stop.load(Ordering::Relaxed) {
        let left = until.saturating_duration_since(std::time::Instant::now());
        if left.is_zero() {
            return;
        }
        std::thread::sleep(left.min(NAP_SLICE));
    }
}

/// How often a sleeping poll loop looks at its stop flag.
pub const NAP_SLICE: Duration = Duration::from_millis(100);

/// The refusal a twin gives while its primary is on another machine and it holds no client
/// certificate yet. Right after the pair is formed this is the PAIRING still in flight (the
/// front's `twin certify` lands a few seconds after the twin's role does — mock gate chain 14,
/// 2026-10-24: role twin at 22:12:21, the follower's first dial at 22:12:24, the certificate
/// after it), so for [`CERTIFY_GRACE`] the follower says it at INFO and waits; past that it is
/// the ERROR it always was.
#[derive(Debug, Clone, Copy)]
pub struct AwaitingCertificate;

impl std::fmt::Display for AwaitingCertificate {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("no client certificate is configured on this box yet")
    }
}

impl std::error::Error for AwaitingCertificate {}

/// How long a fresh follower waits for the pairing's `twin certify` before a missing client
/// certificate is an ERROR.
pub const CERTIFY_GRACE: Duration = Duration::from_secs(180);

/// Whether `err` is the twin waiting for its certificate, inside the grace window since the
/// follower started.
pub(crate) fn awaiting_certify(err: &anyhow::Error, since_start: Duration) -> bool {
    since_start < CERTIFY_GRACE && err.downcast_ref::<AwaitingCertificate>().is_some()
}

/// **Why [`Follower::run`] returned.**
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Ended {
    /// Asked to stop.
    Stopped,
    /// A hole the primary no longer retains: an operator's `twin catchup`.
    SnapshotNeeded,
    /// The primary's head is below this twin's.
    Diverged,
    /// The primary kept failing on this address and has another: re-dial it (the supervisor
    /// starts the loop again at once — a hand-back, not a fault).
    Redial,
}

/// While the primary grows its set, the twin asks again every this many polls (2 s × 5 =
/// 10 s): its growth takes minutes, and every ask is a line in the primary's log.
pub const GROWING_WAIT_POLLS: u32 = 5;

static FOLLOWING: AtomicBool = AtomicBool::new(false);
static STOP_FOLLOWING: AtomicBool = AtomicBool::new(false);
/// Set by [`repoint`]: the stop that follows is a re-dial, not a wind-down.
static REPOINTED: AtomicBool = AtomicBool::new(false);

/// **How long the supervisor waits before starting the follower again.**
///
/// The same order as the poll itself: a twin that cannot start is retried at
/// walking pace, not in a spin. There is no back-off ladder on purpose — the
/// failures this retries (a key not yet enrolled, a primary that is down, a
/// data root that is not mounted yet) are all closed by something OUTSIDE this
/// process, and an exponential wait would mean an estate that is fixed in
/// minute three stays dark until minute forty.
pub const FOLLOWER_RETRY: Duration = Duration::from_secs(5);

/// A re-dial is dialled at once at most this often (T1); otherwise after [`FOLLOWER_RETRY`].
pub const QUICK_REDIAL_EVERY: Duration = Duration::from_secs(30);

/// **Start this twin's loop**, once per process, on its own thread: read
/// `twin.json` and `twin.key`, follow the primary every two seconds. An
/// unreachable primary, or a key not yet enrolled there, is retried; a
/// snapshot-needed stops the loop and the status says so.
pub fn start_follower(stores: Arc<ServerStores>) {
    start_follower_every(stores, FOLLOWER_RETRY);
}

/// [`start_follower`] with the supervisor's retry wait named — the tests drive
/// it in milliseconds.
///
/// **Supervised since 2026-09-21 (STINK S1).** It used to call `run()` ONCE:
/// every failure before the poll loop — no `twin.json` yet, an unreadable
/// `twin.key`, a runtime that would not build, an address that would not parse
/// — killed the thread for the life of the process, and nothing retried it. A
/// twin whose follower is dead can never catch up, and `flushing` refuses to
/// unmount a set whose twin is behind, so ONE dead thread makes every growth
/// the estate will ever order fail — while the only thing anybody sees is a
/// lag number that never moves (measured on the live twin: `applied 0 of
/// 19264`). Two things changed here and both matter:
///
/// * the run is **retried for ever**, every [`FOLLOWER_RETRY`], and
/// * a follower that is NOT running now says so **in the status**, with the
///   attempt count and the error, so the twin's state word on the heartbeat is
///   `unreachable` (printed loud) instead of an empty string beside a lag.
///
/// A `SnapshotNeeded` is the one stop that stays a stop: the twin needs an
/// operator's `twin catchup` and re-running the same step would only write the
/// same refusal every five seconds. The status says `snapshot-needed`, which
/// is the loud word for exactly that.
/// The duty that follows the primary (A28 M18).
pub const TWIN_FOLLOWER: &str = "gunnar-twin";

pub fn start_follower_every(stores: Arc<ServerStores>, retry: Duration) {
    if FOLLOWING.swap(true, Ordering::SeqCst) {
        return;
    }
    STOP_FOLLOWING.store(false, Ordering::SeqCst);
    // A duty (A28 M18): listed, a round that panics (a dial, a poll) is said and the follower
    // goes on after `retry`; one that cannot start is said, never a crash (`expect`).
    let say: nordisk_pairing::duty::Say =
        Arc::new(|l: &str| tracing::error!(target: gunnar_core::observe::TARGET_TWIN, "{l}"));
    // A32-14, stays a thread duty: the follower holds its dialled doors across rounds, which are not Send, so it cannot run on a jera blocking pool.
    let started = nordisk_pairing::duty::start_with(TWIN_FOLLOWER, retry, say, move || {
        // One run of failures across the re-dials (L1): said once, not once per dial.
        let failures: Arc<Mutex<Option<Failing>>> = Arc::new(Mutex::new(None));
        let mut attempt: u64 = 0;
        let started = std::time::Instant::now();
        // T1: a hand-back for a re-dial is dialled again AT ONCE — once per
        // [`QUICK_REDIAL_EVERY`]; a primary that stays quiet waits `retry` as before.
        let mut quick: Option<std::time::Instant> = None;
        move || {
            let run = || -> anyhow::Result<Ended> {
                let cfg = TwinConfig::load(stores.root())?
                    .ok_or_else(|| anyhow!("no {TWIN_CONFIG}: not a twin"))?;
                let key = twin_private_key(stores.root())?;
                tracing::info!(target: gunnar_core::observe::TARGET_TWIN, primary = %cfg.primary, "following the primary");
                let reporting = Arc::clone(&stores);
                let source = GrpcSource::connect(&cfg.primary, key, stores.root())?
                    .with_reporter(move || measure_report(&reporting));
                Ok(Follower::new(Arc::clone(&stores), Box::new(source))
                    .sharing_failures(Arc::clone(&failures))
                    .run(Duration::from_secs(2), &STOP_FOLLOWING))
            };
            if STOP_FOLLOWING.load(Ordering::SeqCst) {
                // A re-point stops the follower to have it dial the new
                // address; anything else is the wind-down.
                if !REPOINTED.swap(false, Ordering::SeqCst) {
                    FOLLOWING.store(false, Ordering::SeqCst);
                    return nordisk_pairing::duty::Round::Done;
                }
                STOP_FOLLOWING.store(false, Ordering::SeqCst);
                attempt = 0;
                return nordisk_pairing::duty::Round::After(Duration::ZERO);
            }
            attempt += 1;
            match run() {
                Err(err) if awaiting_certify(&err, started.elapsed()) => {
                    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin waits for its client certificate from the pairing (`twin certify`); dialling again");
                    stalled(&stores, attempt, &format!("{err:#}"), retry);
                }
                Err(err) => {
                    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, %err, attempt, "the twin cannot follow; starting it again");
                    stalled(&stores, attempt, &format!("{err:#}"), retry);
                }
                // M2 (release-3 test2 mock 2026-09-29): a hand-back for a re-dial is the
                // loop doing its job — said at INFO and started again, never the ERROR
                // below.
                Ok(Ended::Redial) => {
                    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin's poll loop handed back to re-dial its primary; following again");
                    attempt = 0;
                    if quick.is_none_or(|t| t.elapsed() >= QUICK_REDIAL_EVERY) {
                        quick = Some(std::time::Instant::now());
                        return nordisk_pairing::duty::Round::After(Duration::ZERO);
                    }
                }
                Ok(_) => {
                    // `Follower::run` returns on the stop flag or on a
                    // snapshot-needed, and the second one left the status
                    // saying so.
                    if STOP_FOLLOWING.load(Ordering::SeqCst) {
                        return nordisk_pairing::duty::Round::After(Duration::ZERO);
                    }
                    // Diverged and snapshot-needed are the two stops that
                    // STAY stopped: both need an operator, and re-running
                    // the same step would write the same refusal every five
                    // seconds.
                    if status(stores.root()).map(|s| s.state) == Some(State::Diverged) {
                        tracing::error!(target: gunnar_core::observe::TARGET_TWIN, why = %status(stores.root()).map(|s| s.why).unwrap_or_default(), "the twin stopped: the box it is pointed at is a different log");
                        FOLLOWING.store(false, Ordering::SeqCst);
                        return nordisk_pairing::duty::Round::Done;
                    }
                    if status(stores.root()).map(|s| s.state) == Some(State::SnapshotNeeded) {
                        tracing::error!(target: gunnar_core::observe::TARGET_TWIN, "the twin stopped at a hole its primary no longer retains: it needs `gunnar control twin catchup`");
                        FOLLOWING.store(false, Ordering::SeqCst);
                        return nordisk_pairing::duty::Round::Done;
                    }
                    tracing::error!(target: gunnar_core::observe::TARGET_TWIN, attempt, "the twin's poll loop returned without being asked to stop; starting it again");
                    stalled(
                        &stores,
                        attempt,
                        "the poll loop returned without being asked to stop",
                        retry,
                    );
                }
            }
            nordisk_pairing::duty::Round::After(retry)
        }
    });
    if let Err(e) = started {
        FOLLOWING.store(false, Ordering::SeqCst);
        tracing::error!(target: gunnar_core::observe::TARGET_TWIN, error = %e, "ALERT {TWIN_FOLLOWER} the follower could not start: this twin does NOT follow its primary");
    }
}

/// **The SAME primary, now reached over its tunnel** (the WireGuard-only
/// upgrade drill, 2026-10-05): a twin joined on the line before WireGuard-only
/// holds its primary's UTILITY address (`twin.json`); upgraded in place (its
/// `/data` kept), the door names that same primary by its OVERLAY address and
/// `become_twin` refused it as a re-point. `utility` is the address the door
/// names for the primary (`JoinReply.primary_utility`), `overlay` the one it
/// is joined at now. Only a twin that follows exactly `utility` moves; any
/// other address is left for `become_twin` to refuse. `Ok(true)` when it moved.
pub fn onto_tunnel(stores: &ServerStores, utility: &str, overlay: &str) -> anyhow::Result<bool> {
    let root = stores.root();
    let (utility, overlay) = (utility.trim(), overlay.trim());
    anyhow::ensure!(
        !overlay.is_empty(),
        "the primary's overlay address is required"
    );
    let Some(mut cfg) = TwinConfig::load(root)? else {
        return Ok(false);
    };
    if utility.is_empty() || cfg.primary != utility || utility == overlay {
        return Ok(false);
    }
    cfg.primary = overlay.to_string();
    cfg.save(root)?;
    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, was = %utility, now = %overlay, "this twin follows the SAME primary over the tunnel now: joined on its utility address before WireGuard-only, its follower dials the overlay address from its next attempt");
    // The follower dials the address it read when it started: stop it and let
    // the supervisor start it again on the overlay address (as `repoint`).
    if FOLLOWING.load(Ordering::SeqCst) {
        REPOINTED.store(true, Ordering::SeqCst);
        STOP_FOLLOWING.store(true, Ordering::SeqCst);
    }
    Ok(true)
}

/// **Re-point this twin at its primary's NEW control plane** — the one move a
/// twin makes when its primary DIED and was recharged from it at another
/// address (the installer re-feeds the closed twin: Skidbladnir nordisk-pairing
/// `installer::readmit`). The restored main IS the same log and signs with the
/// same host key, so the pins hold; only the address moves. `Ok(true)` when it
/// moved (the follower re-dials it), `Ok(false)` when it already follows there.
pub fn repoint(stores: &ServerStores, primary: &str) -> anyhow::Result<bool> {
    let root = stores.root();
    let primary = primary.trim();
    anyhow::ensure!(!primary.is_empty(), "the primary's address is required");
    let mut cfg = TwinConfig::load(root)?.ok_or_else(|| {
        anyhow!("no {TWIN_CONFIG}: this box is not a twin, there is nothing to re-point")
    })?;
    if cfg.primary == primary {
        return Ok(false);
    }
    let was = std::mem::replace(&mut cfg.primary, primary.to_string());
    cfg.save(root)?;
    tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, was = %was, now = %primary, "RE-POINTED: this twin's primary was recharged at another address; the follower dials it from its next attempt");
    // The follower dials the address it read when it started: stop it and let
    // the supervisor start it again on the new one.
    if FOLLOWING.load(Ordering::SeqCst) {
        REPOINTED.store(true, Ordering::SeqCst);
        STOP_FOLLOWING.store(true, Ordering::SeqCst);
    }
    Ok(true)
}

/// **Ask the follower to stop and let the supervisor wind down.** For the
/// tests, and for a box that is no longer a twin.
pub fn stop_follower() {
    STOP_FOLLOWING.store(true, Ordering::SeqCst);
}

/// **The words a `why` starts with when THIS twin's follower is not running** —
/// as opposed to its primary not answering. Both read `unreachable`, and a restore
/// guard must tell them apart: the first says nothing about the primary at all.
/// Written again, verbatim, in `gunnar-cli`'s `control/twin.rs`; the CLI's test
/// `the_follower_not_running_wording_is_the_servers` holds the two together.
pub const FOLLOWER_NOT_RUNNING: &str = "the follower is NOT running";

/// **Say in the STATUS that this twin is not following.** The heartbeat carries
/// the state word, so a primary printing `twin <p> UNREACHABLE` is what an
/// operator sees instead of a lag number that never moves. The applied head is
/// real; the primary's head is whatever the last successful poll saw, or 0 when
/// there has never been one.
fn stalled(stores: &ServerStores, attempt: u64, why: &str, retry: Duration) {
    // A twin held for its own growth keeps saying so: its follower failing to start meanwhile
    // is not news about its primary.
    if maintenance::is_held(stores.root()) {
        return;
    }
    let last = status(stores.root());
    set_status(
        stores.root(),
        TwinStatus {
            applied_seq: stores.txlog().head_seq(),
            primary_head_seq: last.as_ref().map(|s| s.primary_head_seq).unwrap_or(0),
            state: State::Unreachable,
            why: format!(
                "{FOLLOWER_NOT_RUNNING} (attempt {attempt}, starting again in {} s): {why}",
                retry.as_secs_f32()
            ),
            last_answer_unix_ms: last.as_ref().map(|s| s.last_answer_unix_ms).unwrap_or(0),
            last_level_unix_ms: 0,
        },
    );
}

// ── restore ─────────────────────────────────────────────────────────────────

/// Replay `from` (a twin's log) into `into` (a fresh data root's stores):
/// the restored primary's own log then holds the same numbers and continues
/// from the twin's head. Returns the head reached.
pub fn restore(
    from: Arc<Log>,
    into: Arc<ServerStores>,
    source_log_id: &str,
) -> anyhow::Result<u64> {
    restore_from(Box::new(from), into, source_log_id)
}

/// [`restore`] from any source — one that can make a snapshot closes a hole with it (a twin
/// whose own log starts after a snapshot it installed).
pub fn restore_from(
    source: Box<dyn LogSource>,
    into: Arc<ServerStores>,
    source_log_id: &str,
) -> anyhow::Result<u64> {
    restore_begin(&into)?;
    let f = Follower::new(Arc::clone(&into), source);
    loop {
        match f.step()? {
            Progress::Applied { .. } | Progress::SnapshotInstalled { .. } => continue,
            Progress::CaughtUp { .. } => return restore_finish(&into, 0, source_log_id),
            Progress::SnapshotNeeded { wanted, first } => {
                return Err(anyhow!("the source starts at {first}; a restore needs {wanted}"))
            }
            // A restore's target is a FRESH box (`restore_begin` refuses
            // otherwise), so its own head is 0 and nothing can be below it.
            // Reachable only if that ever stops being true, and then it is the
            // same refusal: the source is not this box's log.
            Progress::Diverged { own, primary_head } => {
                return Err(anyhow!(
                    "the source's head is {primary_head}, below the {own} already on this box — it is not this box's log"
                ))
            }
        }
    }
}

pub mod snapshot;

/// The restore's intake: an entry larger than one call, staged on disk (restore-large-entry).
pub mod intake;

/// **A source over another box's stores in this process** — its log for ranges, and a snapshot
/// of its stores for a hole. What a restore from a twin's data root reads, and what the tests
/// follow a primary through.
pub struct LocalSource {
    /// The box read from.
    pub stores: Arc<ServerStores>,
    /// Pretend retention starts here (tests: a hole without rolling segments). 0 = the log's own.
    pub retained_from: u64,
}

impl LocalSource {
    /// A twin that holds no whole copy (twin-resync-space) gives out no log either.
    fn whole(&self) -> anyhow::Result<()> {
        match self
            .stores
            .is_twin()
            .then(|| snapshot::not_whole(self.stores.root()))
            .flatten()
        {
            Some(why) => anyhow::bail!("{why}. Nothing is read from it"),
            None => Ok(()),
        }
    }
}

impl LogSource for LocalSource {
    fn head(&self) -> anyhow::Result<(u64, u64)> {
        let log = self.stores.txlog();
        Ok((log.first_seq().max(self.retained_from), log.head_seq()))
    }
    fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
        self.whole()?;
        let first = self.head()?.0;
        if from < first {
            return Err(txlog::Error::NotRetained { seq: from, first }.into());
        }
        Ok(fetch_range(self.stores.txlog(), from, to, max_bytes)?)
    }
    fn fetch_into(
        &self,
        from: u64,
        to: u64,
        max_bytes: u64,
        into: &mut dyn FrameSink,
    ) -> anyhow::Result<()> {
        self.whole()?;
        let first = self.head()?.0;
        if from < first {
            return Err(txlog::Error::NotRetained { seq: from, first }.into());
        }
        stream_range(self.stores.txlog(), from, to, max_bytes, into).map(drop)
    }
    fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
        snapshot::serve(&self.stores, &mut |p| into.feed(p))?;
        Ok(true)
    }
}

#[cfg(test)]
mod tests;

/// **A twin that was following before the store-only ruling says so, once, at open.** Its
/// `repos/` tree is what an older build applied; nothing reads it now and nothing updates it,
/// so it is stale bytes taking room on the twin's volume. It is NOT deleted here: deleting a
/// box's data on an upgrade is not a thing an open does. The operator reclaims it (or re-joins
/// the twin, which starts a fresh root).
pub fn say_if_stores_were_applied(stores: &ServerStores) {
    let repos = stores.root().join(crate::stores::REPOS_DIR);
    let stale = std::fs::read_dir(&repos)
        .map(|d| d.flatten().count())
        .unwrap_or(0);
    if stale > 0 {
        tracing::warn!(
            target: gunnar_core::observe::TARGET_TWIN,
            path = %repos.display(),
            namespaces = stale,
            "this twin holds repository stores an OLDER build applied; a twin only stores (2026-09-22) and nothing reads or updates them — \
             they are stale bytes on the twin's volume, kept rather than deleted by an open"
        );
    }
}

#[cfg(test)]
mod skew_tests {
    /// ★ **The twin's OWN credential refused for skew is a reason to ask the front** (test5's
    /// in-place upgrade, 2026-10-09: the primary answered every poll "the credential is
    /// -7198470 ms away from this server's clock" and the twin, whose clock was the wrong one,
    /// asked nobody — the skew trigger fired only on credentials THIS box refused). The
    /// primary's exact refusal is recognised; any other poll failure is not.
    #[test]
    fn a_poll_refused_for_skew_is_told_from_any_other_failure() {
        let skew = anyhow::anyhow!("status: Unauthenticated, message: \"the credential is -7198470 ms away from this server's clock; the window is ±300000 ms\"");
        assert!(super::refused_for_skew(&skew));
        let stale = gunnar_grpc::auth::AuthFailure::Stale {
            skew_ms: -7_198_470,
        }
        .status();
        assert!(
            super::refused_for_skew(&anyhow::anyhow!("{stale}")),
            "the gate's own Stale status: {stale}"
        );
        for other in [
            "status: Unauthenticated, message: \"this credential has already been used\"",
            "transport error: connection refused",
            "the primary is not answering",
        ] {
            assert!(
                !super::refused_for_skew(&anyhow::anyhow!("{other}")),
                "{other}"
            );
        }
    }
}

#[cfg(test)]
mod own_load_tests {
    /// ★ **A twin reports its own load on every poll** (2026-10-05): the
    /// cores it may run on and how busy the box is, read here — on Linux
    /// both are there; a reader that fails is absent, never 0.
    #[test]
    fn a_twin_reports_its_own_cores_and_load() {
        let (busy, total) = super::own_load();
        assert!(
            total.is_some_and(|t| t >= 1_000 && t % 1_000 == 0),
            "{total:?}"
        );
        assert!(busy.is_some(), "/proc/loadavg is readable here");
    }
}

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

    #[test]
    fn a_missing_certificate_is_awaited_only_inside_the_grace_window() {
        let err = anyhow::Error::new(AwaitingCertificate)
            .context("this twin\x27s primary is 10.44.0.2:50051, which is another machine");
        assert!(awaiting_certify(&err, Duration::from_secs(3)));
        assert!(!awaiting_certify(&err, CERTIFY_GRACE));
        assert!(format!("{err:#}").contains("no client certificate is configured on this box yet"));
    }

    #[test]
    fn any_other_failure_stays_an_error() {
        let err = anyhow::anyhow!("connection refused");
        assert!(!awaiting_certify(&err, Duration::from_secs(1)));
    }

    #[test]
    fn a_remote_primary_with_no_certificate_refuses_as_awaiting() {
        let dir = tempfile::tempdir().unwrap();
        let key =
            gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
                .unwrap();
        let err = GrpcSource::connect("10.44.0.2:50051", key, dir.path())
            .err()
            .expect("refused");
        assert!(
            err.downcast_ref::<AwaitingCertificate>().is_some(),
            "{err:#}"
        );
        assert!(format!("{err:#}").contains("It will not"), "{err:#}");
    }
}