gunnar_server/

grpc_gate.rs

//! **The control plane's one transfer, on the same gate as git** — one
//! implementation, not a second gate.
//!
//! `gunnar-grpc` cannot name `gunnar_ssh::Admission` — it is on the
//! client/control graph, and a `gunnar-ssh` edge there would drag the whole SSH
//! server into every `gunnar --features control` client. So it reaches the gate
//! through the [`gunnar_grpc::TransferGate`] seam, and this is the one
//! implementation: a thin adapter over the **same** `Arc<Admission>` the SSH and
//! HTTP front doors hold ([`gunnar_ssh::BoundServer::admission_gate`]). No
//! admission logic lives here — the max-min share, the concurrency ceiling and
//! the memory ledger are all `gunnar_ssh::Admission`'s. This crate only
//! translates one `try_admit` call and its guard across the seam.
//!
//! It exists as its own module, and its constructor is public, because two
//! callers need the exact same wrapper and a second one would be the drift one
//! implementation forbids: [`crate::serve`] wires it onto the live gRPC control plane, and
//! `tests/grpc_admission.rs` wires it onto a gate whose ceiling it controls to
//! drive the refusal.

use std::sync::Arc;

use gunnar_grpc::auth::{AuditLedger, AuditRecord};
use gunnar_grpc::{TransferGate, TransferHold, TransferPermit, TransferRefused};
use gunnar_ssh::governor::OpClass;
use gunnar_ssh::{Admission, AdmissionGuard};

/// Wrap the SSH/HTTP admission gate so the gRPC control plane can put
/// `Repos.ArchiveOut` through it.
///
/// Hand it the very `Arc<Admission>` the git fronts hold
/// (`BoundServer::admission_gate()`), so an archive stream counts against the
/// same server-wide limit, the same per-principal share and the same memory
/// budget as a `git-upload-pack`. The returned trait object is what
/// [`gunnar_grpc::ServiceContext::with_transfer_gate`] takes.
pub fn control_transfer_gate(admission: Arc<Admission>) -> Arc<dyn TransferGate> {
    Arc::new(SshTransferGate { admission })
}

/// The seam implementation. Nothing but the gate and the translation.
struct SshTransferGate {
    admission: Arc<Admission>,
}

impl TransferGate for SshTransferGate {
    fn admit_transfer(
        &self,
        principal: &str,
        namespace: &str,
    ) -> Result<TransferPermit, TransferRefused> {
        // `OpClass::Clone`: an archive is the expensive class — a whole
        // repository's stored bytes read off disk and streamed out — so it
        // reserves and is metered as the heaviest git transfer is, not as a
        // cheap `ls-remote`.
        match self
            .admission
            .try_admit(principal, namespace, OpClass::Clone)
        {
            Ok(guard) => Ok(TransferPermit(Box::new(GuardHold(guard)))),
            Err(refusal) => {
                // The counters are read at the refusal so the audit line names
                // the state that refused — the same snapshot the SSH and HTTP
                // fronts log from.
                let stats = self.admission.stats();
                Err(TransferRefused {
                    message: refusal.client_message(),
                    in_flight: stats.in_flight,
                    peak_in_flight: stats.peak_in_flight,
                    limit: stats.limit,
                    reserved_bytes: stats.reserved_bytes,
                    refused_total: stats.refused(),
                })
            }
        }
    }
}

/// The held permit, across the seam. Dropping it drops the
/// [`AdmissionGuard`], which releases the slot, the per-principal count and the
/// memory reservation — the whole point of routing through the real gate.
struct GuardHold(AdmissionGuard);

impl TransferHold for GuardHold {
    fn entered_service(&self) {
        self.0.entered_service();
    }
}

// ───────────────────────────────────────────────────────────────────────────
// The audit trail (GUN-044)
// ───────────────────────────────────────────────────────────────────────────

/// **The durable, tamper-evident half of the control plane's audit trail**, on
/// the same seam and for the same reason as the gate above: `gunnar-grpc`
/// cannot name `gunnar_core::AuditChain`, so it reaches one through
/// [`gunnar_grpc::auth::AuditLedger`] and this is the one implementation.
///
/// The chain lives beside the rest of the control state
/// (`<control dir>/audit-chain.jsonl`, [`gunnar_core::AUDIT_FILE`]) because
/// that is where everything else a server must not lose already lives, and
/// because a trail on a different disk from the thing it describes is a trail
/// somebody eventually forgets to copy.
///
/// **Opening it verifies it.** `AuditChain::open` reads the existing file and
/// checks every link before the first append, so a server refuses to extend a
/// chain that has already been broken rather than laying genuine records on top
/// of a forgery and making the break look later than it was. That refusal is
/// the caller's to handle: [`crate::serve`] logs it and starts without a
/// ledger, because a control plane that will not start is a worse outcome than
/// one whose audit trail is loudly known to be damaged.
pub fn control_audit_ledger(chain: Arc<gunnar_core::AuditChain>) -> Arc<dyn AuditLedger> {
    control_audit_ledger_with_window(chain, READ_WINDOW)
}

/// [`control_audit_ledger`] with the plain-read window given (tests use a short one).
pub fn control_audit_ledger_with_window(
    chain: Arc<gunnar_core::AuditChain>,
    window: std::time::Duration,
) -> Arc<dyn AuditLedger> {
    let every = window
        .min(std::time::Duration::from_secs(5))
        .max(std::time::Duration::from_millis(50));
    control_audit_ledger_with(chain, window, every)
}

/// [`control_audit_ledger`] with the plain-read window and the sweeper's interval given.
pub fn control_audit_ledger_with(
    chain: Arc<gunnar_core::AuditChain>,
    window: std::time::Duration,
    every: std::time::Duration,
) -> Arc<dyn AuditLedger> {
    ledger_on(chain, window, every, Arc::new(gunnar_core::now_ms))
}

/// **The gate's decisions, also read for "who changed something, and when"**
/// (2026-10-21): every allowed call whose method is not a read
/// ([`gunnar_grpc::auth::changes_state`]) stamps its caller's last operation in
/// `activity`, then the decision goes on to `trail` (the hash-chained ledger)
/// when there is one. The stamp is an in-memory move at most once a minute per
/// principal ([`gunnar_core::activity`]); the request pays no write for it.
///
/// Always installed, with or without a trail: a box whose audit chain would
/// not open still answers when its people last did something.
pub fn activity_ledger(
    trail: Option<Arc<dyn AuditLedger>>,
    activity: Arc<gunnar_core::PrincipalActivity>,
) -> Arc<dyn AuditLedger> {
    Arc::new(ActivityLedger { trail, activity })
}

struct ActivityLedger {
    trail: Option<Arc<dyn AuditLedger>>,
    activity: Arc<gunnar_core::PrincipalActivity>,
}

impl AuditLedger for ActivityLedger {
    fn record(&self, record: AuditRecord<'_>) {
        // An allowed change only: a refusal changed nothing, a narrowed
        // listing is a read, and so is every method `changes_state` says is.
        if record.decision == "allow" && gunnar_grpc::auth::changes_state(record.method) {
            self.activity
                .operated(record.principal, gunnar_core::now_ms());
        }
        if let Some(trail) = &self.trail {
            trail.record(record);
        }
    }

    /// RA4-8: a change is durable only when the trail kept it. With no trail (the chain would
    /// not open) it can only be a log line, so it is not durable.
    fn record_change(&self, record: AuditRecord<'_>) -> Result<(), String> {
        if record.decision == "allow" && gunnar_grpc::auth::changes_state(record.method) {
            self.activity
                .operated(record.principal, gunnar_core::now_ms());
        }
        match &self.trail {
            Some(trail) => trail.record_change(record),
            None => Err("this server keeps no audit trail (its chain did not open)".to_owned()),
        }
    }
}

/// The time the ledger's windows are measured in: the wall clock, or a test's.
type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;

/// The duty that writes the read windows that close while nobody reads (A28 M18).
pub const READ_SWEEPER: &str = "audit-reads";

/// [`control_audit_ledger_with`] on `clock`. A test moves its own clock, so a window closes when
/// the test says and not when a loaded box's fsync happens to outlast it (R5.1, 2026-10-07: a
/// 50 ms window on the wall clock closed between a read and its repeat while the first read's
/// append waited on the disk, and the test counted 83 entries for 50 first reads).
fn ledger_on(
    chain: Arc<gunnar_core::AuditChain>,
    window: std::time::Duration,
    every: std::time::Duration,
    clock: Clock,
) -> Arc<dyn AuditLedger> {
    let ledger = Arc::new(ChainLedger {
        chain,
        window_ms: i64::try_from(window.as_millis()).unwrap_or(i64::MAX).max(1),
        reads: std::sync::Mutex::new(Reads::default()),
        clock,
    });
    // The windows that close while nobody reads are written by a sweeper that lives as long as
    // the ledger does (it holds only a `Weak`).
    // A jera job (A28 M18, A32-14): listed, a sweep that panics is said and a FAILED row and the
    // next one runs. Its first round only waits, as the thread's first sleep did.
    let weak = Arc::downgrade(&ledger);
    let mut first = true;
    let say: gunnar_common::jobs::Say = Arc::new(|l: &str| tracing::error!("{l}"));
    let started = gunnar_common::jobs::start(READ_SWEEPER, every, say, move || {
        if std::mem::take(&mut first) {
            return gunnar_common::jobs::Round::Again;
        }
        match weak.upgrade() {
            Some(ledger) => {
                ledger.sweep((ledger.clock)(), false);
                gunnar_common::jobs::Round::Again
            }
            None => gunnar_common::jobs::Round::Done,
        }
    });
    if let Err(e) = started {
        tracing::error!("ALERT {READ_SWEEPER} no sweeper ({e}): a read window that closes while nobody reads is written at the next read only");
    }
    ledger
}

/// **Repeated plain reads are AGGREGATED, never dropped** (P3 of the load test, 2026-10-06).
///
/// "Who read what" is an audit promise (GUN-044; the enterprise audit row cov-03), so every plain
/// read — allowed or narrowed at level `read`, no override — still reaches the chain. What made
/// them expensive was their number: the console, the sign-in pages and monetize's usage poller
/// made every read a replicated frame and two flushes on main (measured: 300 reads = 300 frames,
/// 600 fsyncs, 250,880 B), each the twin's to store too.
///
/// Per (method, principal, key, scope, level, mode, decision, reason) the FIRST read of a window
/// is written at once, as it always was; the repeats inside the window ([`READ_WINDOW`], 60 s from
/// that first read) are counted, and when the window closes ONE more entry says how many there
/// were and when the first and last of them came: `reason` = `aggregated: <n> more calls, first
/// <ms>, last <ms>, window <s> s` (followed by the read's own reason, if it had one), `at` = the
/// last one. The chain's format is unchanged, so every reader of it (verify, the export, a
/// restore) reads it as before and still answers who read what, and when to the window.
///
/// Said plainly: the COUNT of a window still open when the box dies is not in the trail — its
/// first read is. Refusals (reads included), writes, admin and self-scope calls and overrides
/// are one entry each, at once, as before.
pub const READ_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);

/// The words that open an aggregated entry's `reason`.
pub const AGGREGATED: &str = "aggregated: ";

/// Is this a plain read ([`READ_WINDOW`])?
fn plain_read(record: &AuditRecord<'_>) -> bool {
    record.level == gunnar_grpc::auth::level_name(gunnar_proto::GrantLevel::Read)
        && matches!(record.decision, "allow" | "narrow")
        && !record.reason.starts_with("OVERRIDE:")
}

/// One read's identity, owned.
#[derive(Clone, PartialEq, Eq, Hash)]
struct ReadKey {
    method: String,
    principal: String,
    fingerprint: String,
    scope: String,
    level: String,
    auth_mode: String,
    decision: String,
    reason: String,
}

impl ReadKey {
    fn of(r: &AuditRecord<'_>) -> Self {
        ReadKey {
            method: r.method.to_owned(),
            principal: r.principal.to_owned(),
            fingerprint: r.fingerprint.to_owned(),
            scope: r.scope.to_owned(),
            level: r.level.to_owned(),
            auth_mode: r.auth_mode.to_owned(),
            decision: r.decision.to_owned(),
            reason: r.reason.to_owned(),
        }
    }
}

/// A window: its first read (written), the last, and how many came after the first.
#[derive(Clone, Copy)]
struct Window {
    first_ms: i64,
    last_ms: i64,
    more: u64,
}

#[derive(Default)]
struct Reads {
    open: std::collections::HashMap<ReadKey, Window>,
    /// Windows with repeats that closed when their key was read again: written by the sweeper,
    /// never by the request that noticed (r5 stall review, 2026-10-07).
    closed: Vec<(ReadKey, Window)>,
}

/// The seam implementation. The chain does the arithmetic; this translates.
struct ChainLedger {
    chain: Arc<gunnar_core::AuditChain>,
    window_ms: i64,
    reads: std::sync::Mutex<Reads>,
    /// [`gunnar_core::now_ms`] outside tests.
    clock: Clock,
}

impl ChainLedger {
    /// Close every window older than the window length (`all`: every window), writing the
    /// count of the ones that had repeats. In order of their first read.
    ///
    /// **Only the sweeper thread and the ledger's drop call this** — each window it closes is a
    /// durable audit append, and a request thread must not pay other readers' fsyncs.
    fn sweep(&self, now_ms: i64, all: bool) {
        let mut due: Vec<(ReadKey, Window)> = {
            let mut reads = self.reads.lock().unwrap_or_else(|p| p.into_inner());
            let window_ms = self.window_ms;
            let closed: Vec<ReadKey> = reads
                .open
                .iter()
                .filter(|(_, w)| all || now_ms - w.first_ms >= window_ms)
                .map(|(k, _)| k.clone())
                .collect();
            let mut due: Vec<(ReadKey, Window)> = std::mem::take(&mut reads.closed);
            due.extend(
                closed
                    .into_iter()
                    .filter_map(|k| reads.open.remove(&k).map(|w| (k, w)))
                    .filter(|(_, w)| w.more > 0),
            );
            due
        };
        due.sort_by_key(|(_, w)| w.first_ms);
        for (k, w) in due {
            self.write_window(&k, w);
        }
    }

    /// The one entry a closed window with repeats leaves.
    fn write_window(&self, k: &ReadKey, w: Window) {
        let mut reason = format!(
            "{AGGREGATED}{} more calls, first {}, last {}, window {} s",
            w.more,
            w.first_ms,
            w.last_ms,
            self.window_ms / 1000
        );
        if !k.reason.is_empty() {
            reason.push_str("; ");
            reason.push_str(&k.reason);
        }
        self.write(
            AuditRecord {
                method: &k.method,
                principal: &k.principal,
                fingerprint: &k.fingerprint,
                scope: &k.scope,
                level: &k.level,
                auth_mode: &k.auth_mode,
                decision: &k.decision,
                reason: &reason,
            },
            w.last_ms,
        );
    }
}

impl Drop for ChainLedger {
    fn drop(&mut self) {
        // The counts of the windows still open go into the trail rather than with the process.
        self.sweep((self.clock)(), true);
    }
}

impl AuditLedger for ChainLedger {
    /// RA4-8: a change goes to the chain at once (it is never a plain read) and says whether it
    /// is there.
    fn record_change(&self, record: AuditRecord<'_>) -> Result<(), String> {
        let now = (self.clock)();
        self.append(record, now)
    }

    fn record(&self, record: AuditRecord<'_>) {
        let now = (self.clock)();
        if !plain_read(&record) {
            self.write(record, now);
            return;
        }
        let first = {
            let mut reads = self.reads.lock().unwrap_or_else(|p| p.into_inner());
            let key = ReadKey::of(&record);
            let window_ms = self.window_ms;
            match reads.open.get(&key).copied() {
                Some(w) if now - w.first_ms < window_ms => {
                    if let Some(w) = reads.open.get_mut(&key) {
                        w.more += 1;
                        w.last_ms = now;
                    }
                    false
                }
                Some(w) => {
                    // This key's window closed: its count goes to the sweeper, in order.
                    if w.more > 0 {
                        reads.closed.push((key.clone(), w));
                    }
                    reads.open.insert(
                        key,
                        Window {
                            first_ms: now,
                            last_ms: now,
                            more: 0,
                        },
                    );
                    true
                }
                None => {
                    reads.open.insert(
                        key,
                        Window {
                            first_ms: now,
                            last_ms: now,
                            more: 0,
                        },
                    );
                    true
                }
            }
        };
        // The request writes its OWN first read and nothing else; closed windows are the
        // sweeper's ([`ChainLedger::sweep`]).
        if first {
            self.write(record, now);
        }
    }
}

impl ChainLedger {
    /// Best effort ([`AuditLedger::record`]): a failure is the ERROR line [`Self::append`] says.
    fn write(&self, record: AuditRecord<'_>, at_ms: i64) {
        let _ = self.append(record, at_ms);
    }

    /// One decision into the chain; `Err` when the chain could not take it (RA4-8: a change
    /// that must be recorded before it is applied is then refused). A twin's line kept in this
    /// log only, and a decision held for the remount, are what the box was told to do: `Ok`.
    fn append(&self, record: AuditRecord<'_>, at_ms: i64) -> Result<(), String> {
        let entry = gunnar_core::AuditEvent {
            method: record.method,
            principal: record.principal,
            fingerprint: record.fingerprint,
            scope: record.scope,
            level: record.level,
            auth_mode: record.auth_mode,
            decision: record.decision,
            reason: record.reason,
        };
        // **Best effort, by the seam's contract.** A full disk must not turn
        // into a control plane that refuses to serve; it turns into this line,
        // on the audit target, where an operator is already looking.
        //
        // A box doing what it was told is not that line (G + G2, release-3
        // test2 mock 2026-09-29): a twin (or a box being restored) originates
        // nothing into the trail it holds for its primary, and a growth
        // releases the log for the remount — the first is said at INFO and the
        // decision lives in this log only, the second waits and is written
        // after the remount.
        match self.chain.record(entry, at_ms) {
            Ok(gunnar_core::Recorded::Written(_)) => {}
            Ok(gunnar_core::Recorded::Queued { pending, .. }) => tracing::info!(
                target: gunnar_grpc::AUDIT_TARGET,
                method = record.method,
                principal = record.principal,
                decision = record.decision,
                pending,
                "the data set is being grown: this decision waits for the audit chain and is \
                 written after the remount"
            ),
            Ok(gunnar_core::Recorded::NotOriginated) => tracing::info!(
                target: gunnar_grpc::AUDIT_TARGET,
                method = record.method,
                principal = record.principal,
                decision = record.decision,
                "this box follows its primary (a twin, or restoring): it originates nothing into \
                 the replicated trail, so this decision is in this log only"
            ),
            Err(err) => {
                tracing::error!(
                    target: gunnar_grpc::AUDIT_TARGET,
                    error = %err,
                    method = record.method,
                    principal = record.principal,
                    "the control-plane audit chain could not be appended to; this decision is in \
                     the log and NOT in the trail"
                );
                return Err(err.to_string());
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {

    /// ★ **The read sweeper is a jera job, and a sweep that panics is said, a FAILED row, and the
    /// next one runs** (A28 M18, A32-14): the raw `audit-reads` thread ended at a panic, and the
    /// windows that close while nobody reads were never written again — an audit promise (GUN-044)
    /// kept by nothing.
    #[test]
    fn the_read_sweeper_is_a_jera_job_that_outlives_a_panicking_sweep() {
        let dir = tempfile::tempdir().unwrap();
        let chain =
            Arc::new(gunnar_core::AuditChain::open(dir.path().join("audit-chain.jsonl")).unwrap());
        let calls = Arc::new(std::sync::atomic::AtomicU32::new(0));
        let clock: Clock = {
            let calls = Arc::clone(&calls);
            Arc::new(move || {
                if calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
                    panic!("the clock broke once");
                }
                1_000_000
            })
        };
        let ledger = ledger_on(
            chain,
            std::time::Duration::from_millis(50),
            std::time::Duration::from_millis(10),
            clock,
        );
        let t = std::time::Instant::now();
        let seen = loop {
            let d = gunnar_common::jobs::running().into_iter().find(|d| {
                d.kind() == READ_SWEEPER
                    && d.failures() >= 1
                    && d.looks() >= 4
                    && d.queue().stats().failed >= 1
            });
            if d.is_some() || t.elapsed() > std::time::Duration::from_secs(10) {
                break d;
            }
            std::thread::sleep(std::time::Duration::from_millis(5));
        };
        let d = seen.expect(
            "the sweeper is a jera job that went on after its panic, the panic a FAILED row",
        );
        assert!(d
            .last_failure()
            .unwrap_or_default()
            .contains("the clock broke once"));
        drop(ledger);
        assert!(d.stop(std::time::Duration::from_secs(5)));
    }

    use super::*;
    use std::sync::Mutex;

    /// Every line the ledger says, at its level.
    #[derive(Clone, Default)]
    struct Heard(Arc<Mutex<Vec<u8>>>);

    impl std::io::Write for Heard {
        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(b);
            Ok(b.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    /// An admin decision: one entry each (a plain read is aggregated — [`READ_WINDOW`]).
    fn decision(method: &str) -> AuditRecord<'_> {
        AuditRecord {
            method,
            principal: "monetize",
            fingerprint: "SHA256:x",
            scope: "server",
            level: "admin",
            auth_mode: "required",
            decision: "allow",
            reason: "",
        }
    }

    /// **A plain read writes at most its own entry on the request thread** (r5 stall review,
    /// 2026-10-07). The windows of OTHER readers that closed meanwhile used to be written by
    /// whichever request read next — one durable audit append (an fsync) per window, on a tokio
    /// worker, while the RPC waited: 50 closed windows = 50 fsyncs inside one `Usage.Read`. They
    /// belong to the sweeper thread; the request writes its own first read and nothing else.
    #[test]
    fn a_plain_read_writes_at_most_its_own_entry_on_the_request_thread() {
        let dir = tempfile::tempdir().unwrap();
        let chain =
            Arc::new(gunnar_core::AuditChain::open(dir.path().join("audit-chain.jsonl")).unwrap());
        // A 50 ms window on a clock THIS TEST moves, and a sweeper that does not run during the
        // test: every count below is of entries, whatever the disk's fsyncs cost under load.
        let now = Arc::new(std::sync::atomic::AtomicI64::new(1_000_000));
        let clock: Clock = {
            let now = Arc::clone(&now);
            Arc::new(move || now.load(std::sync::atomic::Ordering::SeqCst))
        };
        let ledger = ledger_on(
            Arc::clone(&chain),
            std::time::Duration::from_millis(50),
            std::time::Duration::from_secs(3600),
            clock,
        );
        let scopes: Vec<String> = (0..50).map(|i| format!("namespace:t{i}")).collect();
        fn read(scope: &str) -> AuditRecord<'_> {
            AuditRecord {
                method: "Usage.Read",
                principal: "monetize",
                fingerprint: "SHA256:m",
                scope,
                level: "read",
                auth_mode: "required",
                decision: "allow",
                reason: "",
            }
        }
        for scope in &scopes {
            ledger.record(read(scope));
            ledger.record(read(scope));
        }
        assert_eq!(
            chain.entries().unwrap().len(),
            50,
            "50 first reads, the repeats counted"
        );
        // Every window closes.
        now.fetch_add(1200, std::sync::atomic::Ordering::SeqCst);
        let before = chain.entries().unwrap().len();
        ledger.record(read("namespace:someone-else"));
        let inline = chain.entries().unwrap().len() - before;
        assert_eq!(
            inline,
            1,
            "the request wrote {inline} entries inline: its own first read and {} other windows",
            inline.saturating_sub(1)
        );
        // A key read again after its window closed: its own new first read, and its old window's
        // count is the sweeper's, not this request's.
        let before = chain.entries().unwrap().len();
        ledger.record(read(&scopes[0]));
        let inline = chain.entries().unwrap().len() - before;
        assert_eq!(
            inline,
            1,
            "a re-read wrote {inline} entries inline: its own first read and {} closed windows",
            inline.saturating_sub(1)
        );
        // Nothing is lost: the closed windows are written by the sweeper or at the ledger's end.
        drop(ledger);
        let entries = chain.entries().unwrap();
        let counted: u64 = entries
            .iter()
            .map(|e| match e.reason.strip_prefix(AGGREGATED) {
                Some(rest) => rest
                    .split_whitespace()
                    .next()
                    .and_then(|n| n.parse().ok())
                    .unwrap_or(0),
                None => 1,
            })
            .sum();
        assert_eq!(counted, 102, "every read is in the trail");
    }

    /// **G + G2 (release-3 test2 mock 2026-09-29), the lines the log gate reads.** The twin was
    /// told `BecomeTwin` at runtime with its chain opened framed, and main's log was released
    /// for its growth: every decision after was `ERROR gunnar::audit: the control-plane audit
    /// chain could not be appended to` (twin 13x, main 4x) — a box doing what it was told, at
    /// ERROR. A follower's decision is said at INFO and written nowhere; a decision during the
    /// grow waits and is written after the remount. Neither is ever an ERROR.
    #[test]
    fn a_twin_or_a_growing_box_says_its_decisions_at_info_never_error() {
        let dir = tempfile::tempdir().unwrap();
        let log = Arc::new(gunnar_core::txlog::Log::open(dir.path().join("txlog")).unwrap());
        gunnar_core::txlog::sink::register(dir.path(), &log);
        let chain = Arc::new(
            gunnar_core::AuditChain::open_framed(dir.path().join(gunnar_core::txlog::AUDIT_REL))
                .unwrap(),
        );
        let ledger = control_audit_ledger(Arc::clone(&chain));
        let heard = Heard::default();
        let h = heard.clone();
        let sub = tracing_subscriber::fmt()
            .with_writer(move || h.clone())
            .with_ansi(false)
            .with_max_level(tracing::Level::TRACE)
            .finish();
        tracing::subscriber::with_default(sub, || {
            ledger.record(decision("Repos.Create"));
            log.release().unwrap();
            ledger.record(decision("Admin.Capacity"));
            ledger.record(decision("Admin.GrowStatus"));
            log.reacquire().unwrap();
            ledger.record(decision("Admin.GrowStatus"));
            log.set_follower(true);
            ledger.record(decision("Admin.GrowStart"));
        });
        let said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
        assert!(
            !said.contains("ERROR"),
            "a twin or a growing box is no ERROR:\n{said}"
        );
        assert_eq!(
            chain.verify().unwrap(),
            4,
            "the grow's two decisions are in the trail after the remount:\n{said}"
        );
        assert!(
            said.contains("INFO") && said.contains("originates nothing"),
            "the follower's decision is said at INFO:\n{said}"
        );
    }
}