gunnar-core 3.0.0

//! **The control plane's audit trail, made tamper-evident** — an append-only
//! hash chain over the decisions the gRPC gate makes.
//!
//! # What was there, and why it was not evidence
//!
//! Every authorised RPC already emitted one `tracing` record on
//! `gunnar::audit` — method, principal, key fingerprint, scope, level, decision.
//! That is a good *log* and it is not a *trail*: it goes wherever the
//! subscriber points it, usually a file or a journal that the same box's root
//! can rewrite, and nothing in it says whether it has been rewritten. An
//! operator reading it cannot tell a complete record from one a line was cut
//! out of, which is exactly the question an audit trail exists to answer. The
//! security review recorded it as GUN-044.
//!
//! # The shape: each entry names the one before it
//!
//! ```text
//!   entry.hash = SHA-256( "gunnar-audit-v1" ‖ prev ‖ len‖field … )
//! ```
//!
//! The first entry's `prev` is [`GENESIS`] (64 zeros). Every later entry's
//! `prev` is the previous entry's `hash`. So the tip hash commits to the whole
//! history: change one byte of one entry and that entry's hash changes, which
//! breaks the `prev` of the one after it, and so on to the end. Removing an
//! entry breaks the same link; so does reordering two. There is no way to edit
//! the middle of the file without rewriting every line after it, and a
//! recorded tip (an operator's note, a second box, a monitoring system that
//! reads the tip once a day) makes even that detectable.
//!
//! **This is not the transparency log.** [`crate::transparency`] is a Merkle
//! tree with signed heads and inclusion proofs, which is the right shape for a
//! log a *third party* audits without trusting the server. This is a linear
//! chain with no signature, which is the right shape for a trail an *operator*
//! checks on their own box, and it costs one hash per RPC and no key
//! management. Making the control-plane trail publicly verifiable is a
//! different, larger decision; this one closes the finding that was actually
//! reported — a trail that cannot say whether it is complete.
//!
//! # The fields are length-prefixed, and that is the load-bearing detail
//!
//! A chain whose fields were joined with a separator would let a value
//! containing that separator move the boundary: a `reason` of `"x\nallow"`
//! could hash identically to a different entry whose decision really was
//! `allow`. So each field goes in as its byte length (8 bytes, little-endian)
//! followed by its bytes, which is unambiguous whatever the field holds. The
//! same reasoning as [`crate::transparency`]'s domain-separating prefixes, one
//! layer down.
//!
//! # One file, appended, fsynced
//!
//! JSON Lines, one entry per line, beside the rest of the control state — the
//! same choice push attestation already made, for the same reason: a record
//! that outlives the process must be readable by something that is not gunnar.
//! Each append is durable before it returns. On the primary's REPLICATED trail
//! ([`AuditChain::open_framed`]) the durable copy is the `Append` frame in the
//! transaction log — group-committed with every frame staged beside it
//! (perf, 2026-10-06: concurrent decisions share one `fdatasync`, measured
//! 2.0 → 0.26 flushes per decision on eight request threads) — and the line in
//! this file follows it without an fsync of its own: the file is the view of the
//! log's frames that every start checks and rebuilds
//! (`gunnar_server::twin::rematerialise_audit_trail`). An unframed chain (a
//! twin's own trail) `sync_all`s each line as before.
//!
//! How decisions reach the chain is the gate's ledger's call
//! (`gunnar_server::grpc_gate`): every refusal, write, admin call, self-scope
//! call and override is one entry at once; repeated plain READS are aggregated —
//! the first of a 60 s window at once, then one entry with the count and the
//! first and last time (P3 of the load test, 2026-10-06, `READ_WINDOW`).

use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use sha2::Digest as _;

/// The file the chain lives in, under the control directory.
pub const AUDIT_FILE: &str = "audit-chain.jsonl";

/// Domain separation, so a hash from this chain is not a hash from anything
/// else in this repository that also runs SHA-256 over records.
const DOMAIN: &str = "gunnar-audit-v1";

/// What the first entry's `prev` is: 64 zeros. A literal rather than an empty
/// string, so every entry in the file has the same shape and a truncated-to-
/// nothing `prev` cannot be mistaken for the start of the chain.
pub const GENESIS: &str = "0000000000000000000000000000000000000000000000000000000000000000";

/// What went wrong with the chain.
#[derive(Debug, thiserror::Error)]
pub enum AuditError {
    /// The file could not be read, created or written.
    #[error("the audit chain at {path}: {source}")]
    Io {
        /// Which file.
        path: String,
        /// The underlying failure.
        source: std::io::Error,
    },
    /// A line in the file is not an entry.
    #[error("the audit chain entry on line {line} does not parse: {source}")]
    Malformed {
        /// 1-based line number.
        line: usize,
        /// What serde said.
        source: serde_json::Error,
    },
    /// The line could not be framed into the transaction log, so it was not
    /// written down at all. **Deliberately not best-effort**: a trail whose
    /// file and whose log disagree is worse than one missing entry, because
    /// the missing entry is announced and the disagreement is not.
    #[error("the audit chain at {path} could not be framed as a transaction: {detail}")]
    Frame {
        /// Which file.
        path: String,
        /// What the log said.
        detail: String,
    },
    /// This chain was asked to be the REPLICATED trail and cannot be
    /// ([`AuditChain::open_framed`]). Never silent: the caller says so out
    /// loud and opens an unframed chain instead.
    #[error("the audit chain at {path} cannot be the replicated trail: {why}")]
    Unframed {
        /// Which file.
        path: String,
        /// Why not, in a sentence an operator can act on.
        why: String,
    },
    /// The box FOLLOWS (a twin, or a box being restored): it originates nothing into the
    /// replicated trail it holds for its primary, so nothing was written ([`Recorded::NotOriginated`]
    /// through [`AuditChain::append`]). Not a fault: the caller's log line is the record.
    #[error("the audit chain at {path} is the primary's trail and this box follows it (a twin, or restoring): it originates nothing into it")]
    NotOriginated {
        /// Which file.
        path: String,
    },
    /// The chain does not hold. **This is the finding, not a fault.**
    #[error("the audit chain is broken at entry {seq} (line {line}): {what}")]
    Broken {
        /// The `seq` the entry claims.
        seq: u64,
        /// 1-based line number.
        line: usize,
        /// Which invariant failed, in a sentence an operator can act on.
        what: String,
    },
}

/// One decision, as the control plane hands it over.
///
/// Borrowed, because every field of it is already a `&str` at the call site and
/// an audit record must not make the decision path allocate.
#[derive(Debug, Clone, Copy)]
pub struct AuditEvent<'a> {
    /// `Service.Method`, e.g. `Access.Grant`.
    pub method: &'a str,
    /// The ACL principal id the credential resolved to.
    pub principal: &'a str,
    /// `SHA256:…` of the key that was presented, empty under the development
    /// escape hatch.
    pub fingerprint: &'a str,
    /// What was asked about: `server`, `namespace:x`, `repo:x/y`.
    pub scope: &'a str,
    /// The grant level the method needs.
    pub level: &'a str,
    /// `required` or `allow-all-insecure`.
    pub auth_mode: &'a str,
    /// `allow`, `deny` or `narrow` — the same three words the `tracing` record
    /// carries, so the two can be read side by side.
    pub decision: &'a str,
    /// Why, for a decision that has a why. Empty otherwise.
    pub reason: &'a str,
}

/// One entry, as it sits on disk.
///
/// Field order is the serialisation order and the hashing order. It is
/// `deny_unknown_fields` so a file written by a newer gunnar is refused loudly
/// rather than verified against fields this build cannot see.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuditEntry {
    /// 0 for the first entry, then +1 each time. Redundant with the file's line
    /// numbering on purpose: a chain whose only ordering was its line order
    /// could be truncated to a shorter, still-valid chain without the entries
    /// themselves saying anything was missing.
    pub seq: u64,
    /// When the decision was made.
    pub at_unix_ms: i64,
    /// `Service.Method`.
    pub method: String,
    /// The ACL principal id.
    pub principal: String,
    /// The key fingerprint.
    pub fingerprint: String,
    /// The scope asked about.
    pub scope: String,
    /// The grant level.
    pub level: String,
    /// The authentication mode the listener was in.
    pub auth_mode: String,
    /// `allow`, `deny`, `narrow`.
    pub decision: String,
    /// Why, or empty.
    pub reason: String,
    /// The previous entry's [`AuditEntry::hash`], or [`GENESIS`].
    pub prev: String,
    /// This entry's own hash. Not hashed into itself; see [`AuditEntry::compute_hash`].
    pub hash: String,
}

impl AuditEntry {
    /// The hash this entry **should** carry, computed from everything else in
    /// it.
    ///
    /// Verification is `entry.hash == entry.compute_hash()` plus
    /// `entry.prev == previous.hash`. Both halves are needed: the first catches
    /// an edited field, the second catches a removed, reordered or inserted
    /// entry.
    pub fn compute_hash(&self) -> String {
        let mut hasher = sha2::Sha256::new();
        let mut field = |bytes: &[u8]| {
            // Length-prefixed, so no value can move a field boundary. See the
            // module docs.
            hasher.update((bytes.len() as u64).to_le_bytes());
            hasher.update(bytes);
        };
        field(DOMAIN.as_bytes());
        field(self.prev.as_bytes());
        field(&self.seq.to_le_bytes());
        field(&self.at_unix_ms.to_le_bytes());
        field(self.method.as_bytes());
        field(self.principal.as_bytes());
        field(self.fingerprint.as_bytes());
        field(self.scope.as_bytes());
        field(self.level.as_bytes());
        field(self.auth_mode.as_bytes());
        field(self.decision.as_bytes());
        field(self.reason.as_bytes());
        let digest = hasher.finalize();
        let mut out = String::with_capacity(64);
        for byte in digest {
            use std::fmt::Write as _;
            let _ = write!(out, "{byte:02x}");
        }
        out
    }
}

/// **The chain itself.** Open it once and append to it; it is `Send + Sync` and
/// holds its own lock, because one server writes one chain from many request
/// threads and two appends that interleaved would each claim the same `prev`.
#[derive(Debug)]
pub struct AuditChain {
    path: PathBuf,
    tip: Mutex<Tip>,
    /// **Is every entry also a transaction?** ([`AuditChain::open_framed`]).
    ///
    /// The primary's trail is: it goes into the box's transaction log as an
    /// `Append` frame, so the twin holds it and a restore brings it back. The
    /// twin's OWN trail is not, and neither is a test's — see
    /// [`crate::txlog::AUDIT_LOCAL_REL`].
    framed: bool,
}

/// What the next entry needs to know about the last one.
#[derive(Debug)]
struct Tip {
    next_seq: u64,
    hash: String,
    /// Lines computed while the log was RELEASED (the data set being grown and unmounted),
    /// oldest first, each already linked to the one before it: framed and written, in order,
    /// once the log is taken back ([`AuditChain::flush`]).
    queued: Vec<String>,
    /// The box followed its primary since this tip was read: the file may hold the primary's
    /// entries now, so the tip is read from the file again before the next write.
    stale: bool,
    /// **Lines framed and not yet in the file**, oldest first, each with its frame's sequence
    /// number (perf, 2026-10-06): a framed chain stages its frame, waits for the group's fsync
    /// with the tip unlocked — so concurrent decisions share one — and only then writes the line,
    /// in chain order, every line whose frame is durable.
    unfiled: Vec<(u64, String)>,
}

/// **What became of one decision** ([`AuditChain::record`]).
#[derive(Debug, Clone)]
pub enum Recorded {
    /// In the file (and, for a framed chain, in the transaction log).
    Written(AuditEntry),
    /// The log is RELEASED (the data set is being grown): the entry is linked and held in
    /// memory, in order, and is framed and written by the first [`AuditChain::record`] or
    /// [`AuditChain::flush`] after the remount. `pending` counts the held entries.
    Queued {
        /// The entry, as it will be written.
        entry: AuditEntry,
        /// How many entries wait.
        pending: usize,
    },
    /// The box FOLLOWS its primary (a twin, or restoring): a framed chain is the primary's
    /// trail, and this box originates nothing into it. Written nowhere.
    NotOriginated,
}

/// What one framed write came to.
enum Wrote {
    Yes,
    /// The log is released: nothing framed, nothing written.
    Released,
    /// The log follows: nothing framed, nothing written.
    Follower,
}

impl AuditChain {
    /// Open — or start — the chain at `path`.
    ///
    /// The existing file is **read and verified** at open, not merely counted.
    /// A server that appended to a chain it had not checked would extend
    /// somebody's forgery with genuine entries and make the break look like it
    /// happened later than it did.
    pub fn open(path: impl Into<PathBuf>) -> Result<AuditChain, AuditError> {
        let path = path.into();
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).map_err(|source| AuditError::Io {
                path: parent.display().to_string(),
                source,
            })?;
        }
        let entries = read_entries(&path)?;
        verify_entries(&entries)?;
        let tip = match entries.last() {
            Some(last) => Tip {
                next_seq: last.seq + 1,
                hash: last.hash.clone(),
                queued: Vec::new(),
                stale: false,
                unfiled: Vec::new(),
            },
            None => Tip {
                next_seq: 0,
                hash: GENESIS.to_owned(),
                queued: Vec::new(),
                stale: false,
                unfiled: Vec::new(),
            },
        };
        Ok(AuditChain {
            path,
            tip: Mutex::new(tip),
            framed: false,
        })
    }

    /// **Open the chain as the REPLICATED trail** — every append is an
    /// `Append` transaction before it is a line in a file, so the twin holds
    /// the trail and `gunnar control twin replay` brings it back.
    ///
    /// This is the D6 fix (lane T11, 2026-09-21). Before it, `log/` was a
    /// per-box path: the twin never held the audit trail and a restore came up
    /// with none, reporting success. The restore happens on the worst day,
    /// which is the day somebody has to reconstruct what happened.
    ///
    /// **It refuses rather than falling back.** The chain must sit at
    /// [`crate::txlog::AUDIT_REL`] under a data root the sink knows, or this
    /// returns [`AuditError::Unframed`] naming why. A chain that quietly
    /// wrote unframed lines would reproduce the exact bug — a trail the
    /// restore drops without saying so. The caller's job is to say it out
    /// loud and open an ordinary chain with [`AuditChain::open`].
    pub fn open_framed(path: impl Into<PathBuf>) -> Result<AuditChain, AuditError> {
        let path = path.into();
        match crate::txlog::sink::log_for(&path) {
            None => {
                return Err(AuditError::Unframed {
                    path: path.display().to_string(),
                    why: "it is under no data root that has a transaction log — the layout put \
                          the `log` directory outside the data root, or this box originates \
                          nothing (a twin, a box being restored)"
                        .to_owned(),
                })
            }
            Some((_, rel)) if rel != crate::txlog::AUDIT_REL => {
                return Err(AuditError::Unframed {
                    path: path.display().to_string(),
                    why: format!(
                        "the log would name it {rel:?}, and a restore resolves only {:?}",
                        crate::txlog::AUDIT_REL
                    ),
                })
            }
            Some(_) => {}
        }
        let mut chain = AuditChain::open(path)?;
        chain.framed = true;
        Ok(chain)
    }

    /// Does every append of this chain also become a transaction? See
    /// [`AuditChain::open_framed`].
    pub fn is_framed(&self) -> bool {
        self.framed
    }

    /// Where the chain lives.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// **Append one decision.** Returns the entry that was written (or, while the log is
    /// released, queued — [`Recorded::Queued`]), whose `hash` is the new tip an operator can
    /// record. A box that follows its primary answers [`AuditError::NotOriginated`].
    pub fn append(&self, event: AuditEvent<'_>, at_unix_ms: i64) -> Result<AuditEntry, AuditError> {
        match self.record(event, at_unix_ms)? {
            Recorded::Written(entry) | Recorded::Queued { entry, .. } => Ok(entry),
            Recorded::NotOriginated => Err(AuditError::NotOriginated {
                path: self.path.display().to_string(),
            }),
        }
    }

    /// The log this chain frames into, when it is framed and the sink knows its root.
    fn framing_log(&self) -> Option<std::sync::Arc<crate::txlog::Log>> {
        if self.framed {
            crate::txlog::sink::log_for(&self.path).map(|(log, _)| log)
        } else {
            None
        }
    }

    /// **Record one decision** — [`Recorded`] says what became of it.
    ///
    /// G + G2 (release-3 test2 mock 2026-09-29): a chain opened framed on a box that was
    /// then told `BecomeTwin` at runtime, and a primary whose log was released for its own
    /// growth, answered every decision with an error (`txlog: this box is a twin …`, `txlog:
    /// released …`) that the gate logged at ERROR — 13 and 4 lines per mock — and the grow's
    /// decisions were lost. A follower now originates nothing and says so without an error; a
    /// released log holds the trail in memory, in order, and writes it after the remount.
    pub fn record(&self, event: AuditEvent<'_>, at_unix_ms: i64) -> Result<Recorded, AuditError> {
        // A framed decision is a mutation of the replicated trail, and its guard is taken BEFORE
        // the chain's own lock and held until the file has the line: a snapshot's freeze waits
        // for every guard in flight, and a thread that held the chain while it waited on the
        // freeze would stop the very thread the freeze waits for (that one files its line under
        // the chain's lock after its frame's fsync). The guards the frame write takes inside are
        // nested and count once.
        let _mutation = self.framing_log().map(|log| log.mutation());
        // Poisoning is not a reason to stop writing the audit trail: a panic in
        // some other request's append leaves the tip behind, and refusing every
        // later record would turn one bug into a gap in the evidence.
        let mut tip = self
            .tip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let log = self.framing_log();
        if log.as_ref().is_some_and(|l| l.is_follower()) {
            // The file is the PRIMARY's trail now (the twin applies its frames into it): a
            // local line would splice a second writer into a one-writer chain. Anything held
            // from before is the primary role's, which ended.
            tip.queued.clear();
            tip.unfiled.clear();
            tip.stale = true;
            return Ok(Recorded::NotOriginated);
        }
        if tip.stale {
            // Promoted again: continue from what the file holds NOW, not from the tip this
            // box remembered before it followed.
            let entries = read_entries(&self.path)?;
            match entries.last() {
                Some(last) => {
                    tip.next_seq = last.seq + 1;
                    tip.hash.clone_from(&last.hash);
                }
                None => {
                    tip.next_seq = 0;
                    tip.hash = GENESIS.to_owned();
                }
            }
            tip.stale = false;
        }
        let released = log.as_ref().is_some_and(|l| l.is_released());
        if !released {
            self.flush_locked(&mut tip)?;
        }

        let mut entry = AuditEntry {
            seq: tip.next_seq,
            at_unix_ms,
            method: event.method.to_owned(),
            principal: event.principal.to_owned(),
            fingerprint: event.fingerprint.to_owned(),
            scope: event.scope.to_owned(),
            level: event.level.to_owned(),
            auth_mode: event.auth_mode.to_owned(),
            decision: event.decision.to_owned(),
            reason: event.reason.to_owned(),
            prev: tip.hash.clone(),
            hash: String::new(),
        };
        entry.hash = entry.compute_hash();

        let line = serde_json::to_string(&entry).expect("an audit entry serialises");
        debug_assert!(!line.contains('\n'), "a JSONL record holds no newline");

        if released || !tip.queued.is_empty() {
            return Ok(Self::hold(&mut tip, entry, line));
        }
        if self.framed {
            // ── the group-committed path (perf, 2026-10-06) ──
            match crate::txlog::sink::stage_line(&self.path, &line) {
                Ok(Some(staged)) => {
                    tip.next_seq = entry.seq + 1;
                    tip.hash.clone_from(&entry.hash);
                    tip.unfiled.push((staged.seq(), line));
                    // The fsync is waited for with the tip UNLOCKED: decisions made meanwhile
                    // link onto this one and stage their frames behind it, and one leader's
                    // fdatasync makes them all durable.
                    drop(tip);
                    staged.durable().map_err(|e| AuditError::Frame {
                        path: self.path.display().to_string(),
                        detail: e.to_string(),
                    })?;
                    let mut tip = self
                        .tip
                        .lock()
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                    self.file_durable(&mut tip, staged.log())?;
                    drop(tip);
                    // The mutation guard goes only now: the file has the line.
                    drop(staged);
                    return Ok(Recorded::Written(entry));
                }
                Ok(None) => {} // under no registered root any more: the unframed write below
                Err(crate::txlog::Error::Released(_)) => {
                    return Ok(Self::hold(&mut tip, entry, line))
                }
                Err(crate::txlog::Error::Follower) => {
                    tip.stale = true;
                    return Ok(Recorded::NotOriginated);
                }
                Err(source) => {
                    return Err(AuditError::Frame {
                        path: self.path.display().to_string(),
                        detail: source.to_string(),
                    });
                }
            }
        }
        match self.write_line(&line)? {
            Wrote::Yes => {}
            Wrote::Released => return Ok(Self::hold(&mut tip, entry, line)),
            Wrote::Follower => {
                tip.stale = true;
                return Ok(Recorded::NotOriginated);
            }
        }
        tip.next_seq = entry.seq + 1;
        tip.hash.clone_from(&entry.hash);
        Ok(Recorded::Written(entry))
    }

    /// Hold a linked entry until the log is taken back.
    fn hold(tip: &mut Tip, entry: AuditEntry, line: String) -> Recorded {
        tip.queued.push(line);
        tip.next_seq = entry.seq + 1;
        tip.hash.clone_from(&entry.hash);
        Recorded::Queued {
            entry,
            pending: tip.queued.len(),
        }
    }

    /// How many entries wait for the log to be taken back ([`Recorded::Queued`]).
    pub fn queued(&self) -> usize {
        self.tip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .queued
            .len()
    }

    /// **Write what waits** — every held entry, in order, once the log is taken back (the
    /// remount after a growth). Answers how many were written; a log still released writes
    /// none and keeps them.
    pub fn flush(&self) -> Result<usize, AuditError> {
        let mut tip = self
            .tip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if self
            .framing_log()
            .is_some_and(|l| l.is_released() || l.is_follower())
        {
            return Ok(0);
        }
        self.flush_locked(&mut tip)
    }

    /// **Write every framed line whose frame is durable into the file**, oldest first, as ONE
    /// write and WITHOUT an fsync: the frame is the durable copy, and the file is the view of the
    /// log's frames that every start checks and rebuilds (`gunnar_server::twin::
    /// rematerialise_audit_trail`, T17) — an fsync per line here bought nothing the log does not
    /// already hold, at the cost of a second flush per decision. A line whose write fails stays
    /// unfiled and goes with the next.
    fn file_durable(&self, tip: &mut Tip, log: &crate::txlog::Log) -> Result<usize, AuditError> {
        let head = log.head_seq();
        let n = tip
            .unfiled
            .iter()
            .take_while(|(seq, _)| *seq <= head)
            .count();
        if n == 0 {
            return Ok(0);
        }
        let mut bytes = Vec::with_capacity(tip.unfiled[..n].iter().map(|(_, l)| l.len() + 1).sum());
        for (_, l) in &tip.unfiled[..n] {
            bytes.extend_from_slice(l.as_bytes());
            bytes.push(b'\n');
        }
        let io = |source: std::io::Error| AuditError::Io {
            path: self.path.display().to_string(),
            source,
        };
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(io)?;
        file.write_all(&bytes).map_err(io)?;
        tip.unfiled.drain(..n);
        Ok(n)
    }

    fn flush_locked(&self, tip: &mut Tip) -> Result<usize, AuditError> {
        // Framed lines whose frames are durable go into the file first — they precede anything
        // held. Their fsync is NOT forced here (that would hold the tip across an fsync and undo
        // the group commit): the thread that staged each one files it. Only held lines waiting
        // behind them (a release just ended) are worth an fsync of their own.
        if !tip.unfiled.is_empty() {
            if let Some(log) = self.framing_log() {
                if !tip.queued.is_empty() {
                    let last = tip.unfiled.last().map(|(s, _)| *s).unwrap_or(0);
                    log.sync_through(last).map_err(|e| AuditError::Frame {
                        path: self.path.display().to_string(),
                        detail: e.to_string(),
                    })?;
                }
                self.file_durable(tip, &log)?;
            }
            if !tip.unfiled.is_empty() && !tip.queued.is_empty() {
                return Ok(0);
            }
        }
        let mut n = 0;
        while let Some(line) = tip.queued.first().cloned() {
            match self.write_line(&line)? {
                Wrote::Yes => {
                    tip.queued.remove(0);
                    n += 1;
                }
                Wrote::Released | Wrote::Follower => break,
            }
        }
        Ok(n)
    }

    /// Frame `line` (a framed chain) and append it to the file, durable.
    fn write_line(&self, line: &str) -> Result<Wrote, AuditError> {
        // **The trail is a transaction before it is a line** (lane T11). Framed FIRST and held
        // until the file has the write, exactly as `attest.rs` frames an attestation: the
        // mutation gate keeps a snapshot from cutting between the frame and the byte, so the
        // twin can never hold a trail that is ahead of the file or behind it.
        //
        // NOT best effort. The gate's contract is that a failure here is one `error!` and a
        // served RPC, and that contract is kept by the CALLER — what must not happen is a line
        // in the file with no frame behind it, because the restore would then bring back a
        // chain with a hole in the middle and `verify` would report the break as the
        // primary's, months later, to somebody who cannot know it was ours.
        let _framed = if self.framed {
            match crate::txlog::sink::append_line(&self.path, line) {
                Ok(f) => Some(f),
                Err(crate::txlog::Error::Released(_)) => return Ok(Wrote::Released),
                Err(crate::txlog::Error::Follower) => return Ok(Wrote::Follower),
                Err(source) => {
                    return Err(AuditError::Frame {
                        path: self.path.display().to_string(),
                        detail: source.to_string(),
                    })
                }
            }
        } else {
            None
        };
        // G3: an UNFRAMED chain (the twin's own trail) has no log to tell it the set is off;
        // the directory it lives in is gone while a growth has /data unmounted. That is the
        // same state as a released log: hold the line, write it after the remount — and never
        // create the directory on the root filesystem under the empty mountpoint.
        if _framed.is_none() && self.path.parent().is_some_and(|p| !p.exists()) {
            return Ok(Wrote::Released);
        }
        let io = |source: std::io::Error| AuditError::Io {
            path: self.path.display().to_string(),
            source,
        };
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(io)?;
        file.write_all(format!("{line}\n").as_bytes()).map_err(io)?;
        // Durable before the caller is told it landed; see the module docs.
        file.sync_all().map_err(io)?;
        Ok(Wrote::Yes)
    }

    /// Every entry, oldest first.
    pub fn entries(&self) -> Result<Vec<AuditEntry>, AuditError> {
        read_entries(&self.path)
    }

    /// **Check the chain end to end.** `Ok(n)` is "n entries, and they hold".
    ///
    /// This is the function the finding asks for: an operator (or
    /// `cargo test`) can ask the file whether anything in it has been changed
    /// or removed, and get an answer that names the entry rather than a
    /// shrug.
    pub fn verify(&self) -> Result<usize, AuditError> {
        let entries = read_entries(&self.path)?;
        verify_entries(&entries)?;
        Ok(entries.len())
    }

    /// The current tip hash — what a whole history commits to.
    ///
    /// Worth writing down somewhere the box cannot reach: with a recorded tip,
    /// even a rewrite of the entire file is detected.
    pub fn tip(&self) -> String {
        self.tip
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .hash
            .clone()
    }
}

/// Read the file, or `[]` if it is not there yet.
fn read_entries(path: &Path) -> Result<Vec<AuditEntry>, AuditError> {
    let text = match std::fs::read_to_string(path) {
        Ok(text) => text,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(source) => {
            return Err(AuditError::Io {
                path: path.display().to_string(),
                source,
            })
        }
    };
    let mut out = Vec::new();
    for (index, line) in text.lines().enumerate() {
        if line.trim().is_empty() {
            continue;
        }
        let entry = serde_json::from_str(line).map_err(|source| AuditError::Malformed {
            line: index + 1,
            source,
        })?;
        out.push(entry);
    }
    Ok(out)
}

/// The arithmetic, over entries already parsed.
///
/// Separate from the file so a caller with entries in hand — a test, a future
/// reader that streams — checks them with the same code the server does (one
/// implementation).
pub fn verify_entries(entries: &[AuditEntry]) -> Result<(), AuditError> {
    let mut expected_prev = GENESIS.to_owned();
    for (index, entry) in entries.iter().enumerate() {
        let line = index + 1;
        let broken = |what: String| AuditError::Broken {
            seq: entry.seq,
            line,
            what,
        };
        if entry.seq != index as u64 {
            return Err(broken(format!(
                "it is the {}th entry in the file and claims seq {}; an entry has been removed \
                 or reordered",
                index, entry.seq
            )));
        }
        if entry.prev != expected_prev {
            return Err(broken(format!(
                "it names {} as the entry before it and the entry before it hashes to {}; \
                 the chain has been cut",
                short(&entry.prev),
                short(&expected_prev)
            )));
        }
        let computed = entry.compute_hash();
        if entry.hash != computed {
            return Err(broken(format!(
                "its contents hash to {} and it carries {}; this entry has been edited",
                short(&computed),
                short(&entry.hash)
            )));
        }
        expected_prev = entry.hash.clone();
    }
    Ok(())
}

/// A hash, shortened for a message a person reads. The full value is in the
/// file.
fn short(hash: &str) -> &str {
    hash.get(..12).unwrap_or(hash)
}

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

    fn event<'a>(method: &'a str, decision: &'a str) -> AuditEvent<'a> {
        AuditEvent {
            method,
            principal: "alice",
            fingerprint: "SHA256:abc",
            scope: "repo:team/app",
            level: "admin",
            auth_mode: "required",
            decision,
            reason: "",
        }
    }

    fn chain_of(path: &Path, count: usize) -> AuditChain {
        let chain = AuditChain::open(path).expect("open");
        for index in 0..count {
            chain
                .append(
                    event("Access.Grant", "allow"),
                    1_700_000_000_000 + index as i64,
                )
                .expect("append");
        }
        chain
    }

    /// The accept side, and the shape. Without it every tamper assertion below
    /// would pass against a `verify` that refused everything.
    #[test]
    fn a_chain_links_each_entry_to_the_one_before_it_and_verifies() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(AUDIT_FILE);
        let chain = chain_of(&path, 3);
        assert_eq!(chain.verify().expect("a fresh chain verifies"), 3);

        let entries = chain.entries().expect("read");
        assert_eq!(entries[0].prev, GENESIS, "the first entry starts the chain");
        assert_eq!(entries[1].prev, entries[0].hash);
        assert_eq!(entries[2].prev, entries[1].hash);
        assert_eq!(chain.tip(), entries[2].hash);
        assert_eq!((entries[0].seq, entries[2].seq), (0, 2));

        // Reopening continues the chain rather than starting a second one: the
        // tip is read back off the file, which is what makes a restart
        // invisible to a verifier.
        let reopened = AuditChain::open(&path).expect("reopen");
        assert_eq!(reopened.tip(), entries[2].hash);
        reopened
            .append(event("Repos.Delete", "deny"), 1_700_000_009_000)
            .expect("append after reopen");
        assert_eq!(reopened.verify().expect("still holds"), 4);
        let entries = reopened.entries().expect("read");
        assert_eq!(entries[3].prev, entries[2].hash);
    }

    /// **A modified entry is detected.** The whole point: `tracing` could not
    /// say this, and this is the sentence the operator gets.
    #[test]
    fn an_edited_entry_is_named() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(AUDIT_FILE);
        let _chain = chain_of(&path, 3);

        // Rewrite the middle record's decision, the edit somebody covering a
        // refusal would make, leaving everything else — the hash included —
        // exactly as gunnar wrote it.
        let text = std::fs::read_to_string(&path).expect("read");
        let mut lines: Vec<String> = text.lines().map(str::to_owned).collect();
        lines[1] = lines[1].replace("\"decision\":\"allow\"", "\"decision\":\"deny\"");
        std::fs::write(&path, lines.join("\n") + "\n").expect("rewrite");

        let err = AuditChain::open(&path).expect_err("a rewritten chain must not open");
        match err {
            AuditError::Broken {
                seq,
                line,
                ref what,
            } => {
                assert_eq!((seq, line), (1, 2), "the break must be named: {what}");
                assert!(what.contains("edited"), "{what}");
            }
            other => panic!("the edit was not detected: {other}"),
        }
    }

    /// **A removed entry is detected** — the tamper `tracing` is least able to
    /// show, because a deleted line leaves nothing behind.
    #[test]
    fn a_removed_entry_is_named() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(AUDIT_FILE);
        let chain = chain_of(&path, 4);
        assert_eq!(chain.verify().expect("green first"), 4);

        let text = std::fs::read_to_string(&path).expect("read");
        let lines: Vec<&str> = text.lines().collect();
        // Drop the second of four.
        let cut: Vec<&str> = lines
            .iter()
            .enumerate()
            .filter(|(index, _)| *index != 1)
            .map(|(_, line)| *line)
            .collect();
        std::fs::write(&path, cut.join("\n") + "\n").expect("rewrite");

        let err = AuditChain::open(&path).expect_err("a cut chain must not open");
        assert!(
            matches!(
                err,
                AuditError::Broken {
                    seq: 2,
                    line: 2,
                    ..
                }
            ),
            "the removal was not detected where it happened: {err}"
        );
    }

    /// **Truncating the tail is detected too — but only against a recorded
    /// tip**, and the difference is stated rather than assumed.
    ///
    /// Cutting the LAST entries leaves a chain that is internally consistent:
    /// that is what a hash chain is, and claiming otherwise would be the kind
    /// of overclaim this file is here to replace. What catches it is the tip an
    /// operator wrote down, so the test asserts exactly that — the shortened
    /// file verifies, and its tip is not the one that was recorded.
    #[test]
    fn a_truncated_tail_verifies_but_does_not_match_the_recorded_tip() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(AUDIT_FILE);
        let chain = chain_of(&path, 4);
        let recorded = chain.tip();

        let text = std::fs::read_to_string(&path).expect("read");
        let kept: Vec<&str> = text.lines().take(2).collect();
        std::fs::write(&path, kept.join("\n") + "\n").expect("rewrite");

        let shortened = AuditChain::open(&path).expect("a truncated chain is self-consistent");
        assert_eq!(shortened.verify().expect("and verifies"), 2);
        assert_ne!(
            shortened.tip(),
            recorded,
            "a truncated chain reproduced the tip of the full one"
        );
    }

    /// A field cannot be moved across a boundary into another field, because
    /// each one is length-prefixed. Without the prefixes these two entries hash
    /// the same.
    #[test]
    fn a_field_holding_a_separator_cannot_impersonate_two_fields() {
        let base = AuditEntry {
            seq: 0,
            at_unix_ms: 1,
            method: "Access.Grant".into(),
            principal: "alice".into(),
            fingerprint: "SHA256:abc".into(),
            scope: "server".into(),
            level: "admin".into(),
            auth_mode: "required".into(),
            decision: "deny".into(),
            reason: String::new(),
            prev: GENESIS.into(),
            hash: String::new(),
        };
        // The pair that COLLIDES under a newline-joined concatenation: both
        // sides render as `alice\nSHA256:abc\n`, and only the length prefixes
        // tell them apart.
        let mut shifted = base.clone();
        shifted.principal = "alice\nSHA256:abc".into();
        shifted.fingerprint = String::new();
        // The naive concatenation of the two differs only in where the
        // boundary is drawn.
        assert_ne!(
            base.compute_hash(),
            shifted.compute_hash(),
            "two entries with different field boundaries hash the same"
        );
    }

    // ── the trail survives a restore (lane T11, 2026-09-21; the D6 finding) ──

    /// A data root with a transaction log registered on it, and the audit chain where the
    /// layout puts it: `<root>/log/audit-chain.jsonl`.
    fn framed_root() -> (
        tempfile::TempDir,
        std::sync::Arc<crate::txlog::Log>,
        PathBuf,
    ) {
        let dir = tempfile::tempdir().expect("tempdir");
        let log =
            std::sync::Arc::new(crate::txlog::Log::open(dir.path().join("txlog")).expect("a log"));
        crate::txlog::sink::register(dir.path(), &log);
        let path = dir.path().join(crate::txlog::AUDIT_REL);
        (dir, log, path)
    }

    fn a_decision(method: &str) -> AuditEvent<'_> {
        AuditEvent {
            method,
            principal: "alice",
            fingerprint: "SHA256:abc",
            scope: "server",
            level: "admin",
            auth_mode: "required",
            decision: "allow",
            reason: "",
        }
    }

    /// **Every entry of a framed chain is a transaction**, so the twin holds the audit trail
    /// and `gunnar control twin replay` brings it back. That is the whole D6 fix: before it,
    /// `log/` was a per-box path and a restore came up with no audit history, silently.
    ///
    /// Proven the way a restore would see it: the frames are read back out of the log and the
    /// file is rebuilt from them alone, then opened and verified.
    ///
    /// **Fails before the fix**: make `AuditChain::open_framed` return the chain that
    /// `AuditChain::open` returns (drop the `chain.framed = true;` line) and the log holds
    /// nothing — `head_seq()` is 0 and the rebuilt file is empty.
    #[test]
    fn a_framed_chain_puts_every_entry_in_the_transaction_log() {
        let (_dir, log, path) = framed_root();
        let chain = AuditChain::open_framed(&path).expect("the trail is framable where it lives");
        assert!(chain.is_framed());
        for m in ["Repos.Create", "Access.Grant", "Policy.SetProtection"] {
            chain.append(a_decision(m), 1).expect("append");
        }
        assert_eq!(chain.verify().expect("verify"), 3);

        // What the twin holds: three `Append` frames, all naming the ONE canonical path, in
        // the order the chain wrote them.
        assert_eq!(log.head_seq(), 3, "one frame per decision");
        let mut rebuilt = String::new();
        for seq in 1..=3 {
            let e = log.read(seq).expect("a frame");
            assert_eq!(e.kind(), Some(crate::txlog::Kind::Append));
            match e.meta().expect("meta") {
                crate::txlog::Meta::Append { log: rel } => assert_eq!(rel, crate::txlog::AUDIT_REL),
                other => panic!("seq {seq} is {other:?}, not an Append of the audit trail"),
            }
            rebuilt.push_str(std::str::from_utf8(&e.blob).expect("utf8"));
            rebuilt.push('\n');
        }

        // A restored box rebuilds the file from the frames alone — and it verifies, with the
        // same tip. That is the property the owner asked for in one assertion.
        let elsewhere = tempfile::tempdir().expect("tempdir");
        let restored_path = elsewhere.path().join("audit-chain.jsonl");
        std::fs::write(&restored_path, rebuilt).expect("write");
        let restored = AuditChain::open(&restored_path).expect("a restored trail opens");
        assert_eq!(restored.verify().expect("verify"), 3);
        assert_eq!(restored.tip(), chain.tip());

        // And the restored box CONTINUES the chain rather than starting a new one: its next
        // entry's `prev` is the dead primary's tip, so the history is unbroken across the
        // disaster.
        let next = restored
            .append(a_decision("Repos.List"), 2)
            .expect("append");
        assert_eq!(next.prev, chain.tip());
        assert_eq!(next.seq, 3);
    }

    /// **A chain that cannot be replicated REFUSES to pretend it is**, rather than quietly
    /// writing unframed lines — which is the exact shape of the bug being fixed: a record the
    /// restore drops without saying so.
    ///
    /// Two ways it can happen: the `log` directory resolved outside any data root (no sink),
    /// and the chain sitting somewhere other than the one path a restore resolves.
    ///
    /// **Fails before the fix**: delete the `match crate::txlog::sink::log_for(&path)` block
    /// from `AuditChain::open_framed` and both halves return `Ok`.
    #[test]
    fn an_unreplicable_chain_refuses_to_be_the_replicated_trail() {
        let outside = tempfile::tempdir().expect("tempdir");
        let err = AuditChain::open_framed(outside.path().join("audit-chain.jsonl"))
            .expect_err("a chain under no data root is not the replicated trail");
        assert!(matches!(err, AuditError::Unframed { .. }), "{err}");
        assert!(format!("{err}").contains("under no data root"), "{err}");

        let (dir, _log, _path) = framed_root();
        let misplaced = dir.path().join("somewhere-else.jsonl");
        let err = AuditChain::open_framed(&misplaced).expect_err("a misplaced chain is refused");
        assert!(matches!(err, AuditError::Unframed { .. }), "{err}");
        assert!(format!("{err}").contains(crate::txlog::AUDIT_REL), "{err}");
    }

    /// **An unframed chain frames nothing** — the twin's own trail
    /// ([`crate::txlog::AUDIT_LOCAL_REL`]) and every test harness. A twin originates nothing,
    /// and a twin that put its own decisions into the log it is holding for the primary would
    /// be forging the primary's trail.
    ///
    /// **Fails before the fix**: make `AuditChain::append` frame unconditionally (drop the
    /// `if self.framed`) and the log holds the twin's own decision at seq 1.
    #[test]
    fn a_twins_own_trail_is_never_a_transaction() {
        let (dir, log, _path) = framed_root();
        let local = dir.path().join(crate::txlog::AUDIT_LOCAL_REL);
        let chain = AuditChain::open(&local).expect("the twin's own trail opens");
        assert!(!chain.is_framed());
        chain
            .append(a_decision("Mirror.LogHead"), 1)
            .expect("append");
        assert_eq!(
            log.head_seq(),
            0,
            "a twin originates nothing into the primary's log"
        );
        assert_eq!(chain.verify().expect("verify"), 1);
        let _ = log;
    }

    /// **G (release-3 test2 mock 2026-09-29): a box that FOLLOWS originates nothing into the
    /// replicated trail, and says so without an error.** The twin opened its chain framed (it
    /// started as a primary) and was then told `Bootstrap.BecomeTwin` at runtime: its log turned
    /// follower, and every later decision (`Admin.GrowStatus` polls from monetize, 13 of them)
    /// was an `ERROR txlog: this box is a twin (or restoring); it originates no transactions`.
    ///
    /// A follower's decision is NOT written — not into the log (the primary's numbers), and not
    /// into the file (the primary's chain, which the twin holds) — and the answer is a
    /// [`Recorded::NotOriginated`], not an error. A box promoted again continues from the tip the
    /// file holds THEN, not from the one it remembered before it followed.
    #[test]
    fn a_framed_chain_on_a_box_that_follows_originates_nothing_and_is_no_error() {
        let (_dir, log, path) = framed_root();
        let chain = AuditChain::open_framed(&path).expect("framed");
        chain
            .append(a_decision("Repos.Create"), 1)
            .expect("the primary's own decision");
        log.set_follower(true);
        match chain
            .record(a_decision("Admin.GrowStatus"), 2)
            .expect("a follower's decision is not an error")
        {
            Recorded::NotOriginated => {}
            other => panic!("a follower originates nothing: {other:?}"),
        }
        assert!(matches!(
            chain.append(a_decision("Admin.GrowStatus"), 3),
            Err(AuditError::NotOriginated { .. })
        ));
        assert_eq!(
            log.head_seq(),
            1,
            "nothing framed into the primary's numbers"
        );
        assert_eq!(
            chain.verify().expect("verify"),
            1,
            "nothing written into the primary's chain"
        );

        // The primary's own frames reach the file while it follows (the twin applies them) …
        let other = AuditChain::open(&path).expect("the applier's view");
        let applied = other
            .append(a_decision("Access.Grant"), 4)
            .expect("applied");
        // … and a promoted box continues from THAT tip.
        log.set_follower(false);
        let next = chain
            .append(a_decision("Repos.List"), 5)
            .expect("originates again");
        assert_eq!(
            (next.seq, next.prev.clone()),
            (applied.seq + 1, applied.hash.clone())
        );
        assert_eq!(chain.verify().expect("verify"), 3);
    }

    /// **G2: while the data set is grown the trail WAITS, in order, and is written after the
    /// remount — never an error.** MEASURED main on the same mock: 4 `ERROR txlog: released — the
    /// files under /data/txlog are given up while the data set is grown` on `Admin.Capacity` /
    /// `Admin.GrowStatus` during its growth, and those decisions were never in the trail.
    #[test]
    fn a_released_log_queues_the_trail_and_the_first_append_after_the_remount_writes_it_in_order() {
        let (_dir, log, path) = framed_root();
        let chain = AuditChain::open_framed(&path).expect("framed");
        chain.append(a_decision("Repos.Create"), 1).expect("append");
        log.release().expect("the grow releases the log");
        for (i, m) in ["Admin.Capacity", "Admin.GrowStatus"]
            .into_iter()
            .enumerate()
        {
            match chain
                .record(a_decision(m), 2 + i as i64)
                .expect("a decision during the grow is not an error")
            {
                Recorded::Queued { pending, .. } => assert_eq!(pending, i + 1),
                other => panic!("released: queued, not {other:?}"),
            }
        }
        assert_eq!(chain.queued(), 2);
        assert_eq!(
            log.head_seq(),
            1,
            "nothing framed while the files are given up"
        );
        assert_eq!(
            chain.verify().expect("verify"),
            1,
            "nothing written under the unmounted set"
        );
        log.reacquire().expect("the remount");
        match chain
            .record(a_decision("Admin.GrowStatus"), 9)
            .expect("after the remount")
        {
            Recorded::Written(e) => assert_eq!(e.seq, 3),
            other => panic!("written after the remount: {other:?}"),
        }
        assert_eq!(chain.queued(), 0);
        assert_eq!(chain.verify().expect("the chain holds across the grow"), 4);
        assert_eq!(log.head_seq(), 4, "every decision framed, in order");
        let methods: Vec<String> = chain
            .entries()
            .unwrap()
            .into_iter()
            .map(|e| e.method)
            .collect();
        assert_eq!(
            methods,
            [
                "Repos.Create",
                "Admin.Capacity",
                "Admin.GrowStatus",
                "Admin.GrowStatus"
            ]
        );
        // `flush` writes a queue without waiting for the next decision.
        log.release().unwrap();
        chain.record(a_decision("Admin.Capacity"), 10).unwrap();
        log.reacquire().unwrap();
        assert_eq!(chain.flush().expect("flush"), 1);
        assert_eq!(
            (chain.queued(), chain.verify().unwrap(), log.head_seq()),
            (0, 5, 5)
        );
    }

    /// **G3 (d-upg3 2026-10-02, the KVM mock of the full release upgrade at ops fa23a53): an
    /// UNFRAMED chain whose set is unmounted for a growth WAITS too — never an error.** With
    /// `[machines.twin]` fed to vm1, monetize grows the TWIN; the twin's own trail
    /// ([`crate::txlog::AUDIT_LOCAL_REL`]) is not framed, so nothing told it the set was off, and
    /// monetize's own `Admin.GrowStatus` poll during the growth answered `ERROR … the audit chain
    /// at /data/log/audit-chain.local.jsonl: No such file or directory` — the log gate's FAIL.
    /// The directory the chain lives in is GONE while /data is unmounted: the line is held, in
    /// order, and written by the first decision after the remount (or `flush`).
    #[test]
    fn an_unframed_chain_whose_set_is_unmounted_queues_and_writes_after_the_remount() {
        let dir = tempfile::tempdir().unwrap();
        let data = dir.path().join("data");
        let local = data.join(crate::txlog::AUDIT_LOCAL_REL);
        let chain = AuditChain::open(&local).expect("the twin's own trail opens");
        chain
            .append(a_decision("Mirror.LogHead"), 1)
            .expect("before the grow");
        // The growth unmounts /data: the directory the chain is in is not there.
        let off = dir.path().join("data-off");
        std::fs::rename(&data, &off).unwrap();
        match chain
            .record(a_decision("Admin.GrowStatus"), 2)
            .expect("a decision while the set is off is not an error")
        {
            Recorded::Queued { pending, .. } => assert_eq!(pending, 1),
            other => panic!("the set is off: queued, not {other:?}"),
        }
        assert!(
            !data.exists(),
            "nothing is created where the set is not mounted"
        );
        std::fs::rename(&off, &data).unwrap();
        match chain
            .record(a_decision("Admin.GrowStatus"), 3)
            .expect("after the remount")
        {
            Recorded::Written(e) => assert_eq!(e.seq, 2),
            other => panic!("written after the remount: {other:?}"),
        }
        assert_eq!(
            (
                chain.queued(),
                chain.verify().expect("the chain holds across the grow")
            ),
            (0, 3)
        );
    }

    /// **Decisions group-committed from many threads, with a snapshot's freeze cutting in,
    /// never deadlock and never fork the chain** (perf, 2026-10-06). Eight threads record while
    /// another freezes the log over and over: every decision lands once, the chain verifies, the
    /// file holds exactly the lines the log's frames carry, in the same order — and the whole run
    /// ends (a thread that held the chain's lock while it waited on the freeze would stop the
    /// thread the freeze waits for; the guard is taken before the lock).
    #[test]
    fn concurrent_framed_decisions_with_freezes_land_once_in_order_and_never_deadlock() {
        let (_dir, log, path) = framed_root();
        let chain = std::sync::Arc::new(AuditChain::open_framed(&path).expect("framed"));
        let (tx, rx) = std::sync::mpsc::channel();
        {
            let (chain, log) = (std::sync::Arc::clone(&chain), std::sync::Arc::clone(&log));
            std::thread::spawn(move || {
                let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
                let freezer = {
                    let (log, stop) = (std::sync::Arc::clone(&log), std::sync::Arc::clone(&stop));
                    std::thread::spawn(move || {
                        let mut n = 0u64;
                        while !stop.load(std::sync::atomic::Ordering::SeqCst) {
                            let g = log.freeze();
                            std::thread::sleep(std::time::Duration::from_micros(200));
                            drop(g);
                            n += 1;
                        }
                        n
                    })
                };
                std::thread::scope(|s| {
                    for t in 0..8 {
                        let chain = std::sync::Arc::clone(&chain);
                        s.spawn(move || {
                            for i in 0..40 {
                                chain
                                    .append(
                                        a_decision(if (t + i) % 2 == 0 {
                                            "Repos.Create"
                                        } else {
                                            "Access.Grant"
                                        }),
                                        i,
                                    )
                                    .expect("a decision lands");
                            }
                        });
                    }
                });
                stop.store(true, std::sync::atomic::Ordering::SeqCst);
                let _ = tx.send(freezer.join().unwrap());
            });
        }
        let freezes = rx
            .recv_timeout(std::time::Duration::from_secs(120))
            .expect("320 decisions under freezes end — no deadlock");
        assert!(freezes > 0, "the freezer cut in");
        assert_eq!(chain.verify().expect("the chain holds"), 320);
        let file: Vec<String> = std::fs::read_to_string(&path)
            .unwrap()
            .lines()
            .map(str::to_owned)
            .collect();
        let framed: Vec<String> = (1..=log.head_seq())
            .map(|s| String::from_utf8(log.read(s).unwrap().blob).unwrap())
            .collect();
        assert_eq!(
            file, framed,
            "the file is exactly the log's frames, in the log's order"
        );
    }
}