gunnar-server 2.0.0

//! The attestation log, on the server's own store.
//!
//! At receive-pack time gunnar already knows who pushed, with which key, into
//! which repo, which references moved, which objects are new and what the pack
//! hashed to. The attestation rule says gunnar does exactly two things with that:
//!
//! 1. emits a `PushAttestation` on the event stream, and
//! 2. appends **one JSON line** to a log in its own store (this module).
//!
//! # The dependency edge that is deliberately absent
//!
//! **Nothing here links the Build Thing, nornir, skade, arrow or parquet.** The
//! log is a file of newline-delimited JSON and the event is a proto message;
//! anything that wants a warehouse row subscribes to `Events.Watch` and writes
//! the row itself. `tests/feature_split.rs` asserts the server half's resolved
//! dependency graph, and this module is why that assertion has to keep holding:
//! the moment a writer is linked in here, the appliance case (the no-OS server rule), where none
//! of those crates could be linked at all, stops building.
//!
//! # Why a failed append does not fail the push
//!
//! The push has already happened when this is called: the references moved and
//! the client has been told so. Returning an error here could only turn a
//! successful push into a reported failure, which is the worst outcome of all:
//! gunnar and the client would disagree about what is in the repository. A
//! failure is therefore logged at `error!` and nothing else, and the operator
//! sees a gap in the log rather than a phantom rejection.

use std::io::Write;
use std::path::{Path, PathBuf};

use gunnar_proto::PushAttestation;

/// The attestation log's file name, directly under the data root.
///
/// One file for the whole server rather than one per repository: an attestation
/// is a provenance record of a *push*, and the consumer of the attestation wants the whole
/// stream in arrival order. The repository is a field in every line.
pub const ATTESTATION_LOG: &str = "attestations.jsonl";

/// An append-only JSONL file.
#[derive(Debug, Clone)]
pub struct AttestationLog {
    path: PathBuf,
}

impl AttestationLog {
    /// The log under a data root.
    pub fn under(root: impl AsRef<Path>) -> Self {
        AttestationLog::at(root.as_ref().join(ATTESTATION_LOG))
    }

    /// The log at an explicit path.
    pub fn at(path: impl Into<PathBuf>) -> Self {
        AttestationLog { path: path.into() }
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Append one attestation as one line.
    ///
    /// The whole record, JSON plus its newline, is handed to a single
    /// `write_all` on a file opened `O_APPEND`, so two concurrent pushes cannot
    /// interleave halves of two lines. A reader may therefore split on `\n`
    /// without a parser, which is the only property JSONL actually promises.
    pub fn append(&self, attestation: &PushAttestation) -> anyhow::Result<()> {
        let line = gunnar_grpc::attestation_json_line(attestation)?;
        self.append_line(&line).map(|_| ())
    }

    /// **Append one record that is not a push**, and return its sequence
    /// number — the zero-based line index it now sits at, which is also its
    /// leaf index in the [`TransparencyLog`] over this file.
    ///
    /// The metered appliance's entitlement facts go through here
    /// (`crate::entitlement`): every plan change is then a line under the
    /// same signed tree heads as every push, so a service provider who edits
    /// their own plan history is caught by the same audit that catches a
    /// rewritten push. The line must be one JSON object with a top-level key
    /// that is not a push's — [`AttestationLog::read`] tells them apart by
    /// [`ENTITLEMENT_KEY`] and [`RECLAIM_KEY`] — and must not contain a
    /// newline.
    ///
    /// Counting the existing lines is a read of the whole file, and that is
    /// accepted: this is the rare path (a plan changes a few times a year per
    /// tenant), and a sequence number that could be wrong would be worse than
    /// one that costs a read. Two writers racing here can still interleave
    /// their counts; the caller serialises (`EntitlementSlot` holds a mutex
    /// across count-and-append).
    pub fn append_line(&self, line: &str) -> anyhow::Result<u64> {
        anyhow::ensure!(
            !line.contains('\n'),
            "an attestation record must be one line; this one holds a newline"
        );
        let seq = self.lines()?.len() as u64;
        let _framed = gunnar_core::txlog::sink::append_line(&self.path, line)?;
        let mut record = line.to_owned();
        record.push('\n');
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)?;
        file.write_all(record.as_bytes())?;
        // **Synced, like every other framed write under the data root.** The twin's side of the
        // same frame (`twin::append_line_once`) syncs; this did not, and
        // `tests/appliance_powercut_recovery.rs` is what an unsynced write under this root costs
        // on xfs — "rename journaled, data blocks not", the file back as zeroes. The frame above
        // is already durable and the twin already has the line, so without this the PRIMARY is
        // the only box that loses it, which is the wrong way round for the box that serves.
        file.sync_data()?;
        Ok(seq)
    }

    /// The raw lines, in order, blank ones dropped. The line index is the
    /// record's sequence number and its leaf index.
    fn lines(&self) -> anyhow::Result<Vec<String>> {
        let text = match std::fs::read_to_string(&self.path) {
            Ok(text) => text,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(err) => return Err(err.into()),
        };
        Ok(text
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(str::to_owned)
            .collect())
    }

    /// Every push attestation in the log, in the order it was written.
    ///
    /// Exposed because a log nothing can read back is a log nobody can check.
    /// A record that is not a push (an entitlement line or a cache-reclaim
    /// line, see [`AttestationLog::append_line`]) is skipped, which is why the
    /// **position in this vector is not a sequence number** — use
    /// [`AttestationLog::read_indexed`] for anything that proves inclusion.
    pub fn read(&self) -> anyhow::Result<Vec<PushAttestation>> {
        Ok(self
            .read_indexed()?
            .into_iter()
            .map(|(_, record)| record)
            .collect())
    }

    /// [`AttestationLog::read`], with each push's line index — the number a
    /// [`TransparencyLog`] proof is made against. The two are one function so
    /// the index a verifier proves is the line the record was read from.
    pub fn read_indexed(&self) -> anyhow::Result<Vec<(usize, PushAttestation)>> {
        let mut out = Vec::new();
        for (index, line) in self.lines()?.into_iter().enumerate() {
            if is_non_push_line(&line) {
                continue;
            }
            if is_crash_tail(&line) {
                // The power cut's artifact, named rather than parsed. Kept as a LINE — the
                // indices of everything before it are what proofs were made against — and not
                // as a record. The line itself is not lost: `append_line` framed it before it
                // wrote it, so the txlog holds it and the twin applied it.
                tracing::warn!(
                    target: gunnar_core::observe::TARGET_TWIN,
                    path = %self.path.display(),
                    index,
                    bytes = line.len(),
                    "a line of NUL bytes at the end of this log is a power cut's tear, not a record; \
                     the records before it read as they did, and this box's copy of that one is torn \
                     (the txlog and the twin hold it)"
                );
                continue;
            }
            let record: PushAttestation = serde_json::from_str(&line)
                .map_err(|err| anyhow::anyhow!("attestation line {index} does not parse: {err}"))?;
            // **A JSON object is not a push.** `PushAttestation`'s serde impl tolerates unknown
            // keys and defaults everything else, so ANY object that is not a kind
            // [`is_entitlement_line`] names came back from the line above as a push with no id,
            // no repository, no principal and no pack hash — and `prove_pack`, `Repos.Chain` and
            // every event listing then carried an invented record that `gunnar verify` would
            // prove like a real one. An attestation always has an id; a record without one is a
            // record this build does not know, and that is an error a reader sees.
            anyhow::ensure!(
                !record.attestation_id.is_empty(),
                "attestation line {index} parses as a push with no attestation id — it is not a push. \
                 A record kind this build does not know must be named in `is_entitlement_line`'s \
                 predicate, never defaulted into a push that never happened"
            );
            out.push((index, record));
        }
        Ok(out)
    }
}

/// **The artifact of a power cut on an appended log**: a line that is nothing but NUL bytes.
///
/// `tests/appliance_powercut_recovery.rs` paid for this shape once already, on
/// `replication.json`: "rename journaled, data blocks not", and the file came back 0 bytes. An
/// `O_APPEND` log tears the same way — xfs journals the new size before the blocks reach the
/// platter, so the tail comes back as zeroes. It is not blank, so `lines()` keeps it; it is not
/// JSON, so every reader used to die on it, for ever, on a box with no shell to edit it with.
fn is_crash_tail(line: &str) -> bool {
    !line.is_empty() && line.bytes().all(|b| b == 0)
}

/// The top-level key that marks an entitlement record in the push log. A
/// push attestation has no such key; a line that has it is skipped by
/// [`AttestationLog::read`] and written by `crate::entitlement`.
pub const ENTITLEMENT_KEY: &str = "entitlement";

/// The top-level key that marks a **cache-reclaim** record: the bytes of
/// derived cache a namespace gave back when its plan fell to Suspended or
/// Retention. Written by `crate::entitlement`, skipped by
/// [`AttestationLog::read`] for the same reason [`ENTITLEMENT_KEY`] is.
///
/// It is on this log and not a log of its own because the reclaim is a
/// *consequence of a plan change*: the fact that caused it is the line before
/// it, under the same signed tree heads, so an operator asking "what did
/// suspending this tenant actually give back" reads two adjacent lines and an
/// appliance that rewrote either is caught by the same audit.
pub const RECLAIM_KEY: &str = "cache_reclaim";

/// Is this line one of the records that is **not** a push?
///
/// Decided by the shape of the JSON, not by a parse failure: a push line that
/// does not parse must stay an error a reader sees, never be mistaken for a
/// different kind of record and skipped.
///
/// **Every new record kind has to be named here, and the failure mode of
/// forgetting is not the loud one it looks like.** The expectation was that an
/// unlisted key would make [`AttestationLog::read_indexed`] error — it does
/// hard-error on a line it cannot parse as a `PushAttestation`. That is not
/// what happens. MEASURED 2026-09-16 by removing `RECLAIM_KEY` from this
/// predicate and running
/// `a_record_that_is_not_a_push_is_skipped_and_its_index_still_belongs_to_it`:
/// `PushAttestation` tolerates unknown fields and defaults the rest, so the
/// reclaim line came back as a **phantom push** — `(2, "")`, an attestation
/// with no id, no repository, no principal and no pack hash — sitting between
/// the real ones. A verifier would have proved it and an event listing would
/// have shown it. The cost of a missing key here is an invented record, not an
/// error, which is why the guard asserts the exact line indices rather than
/// "it parses".
fn is_non_push_line(line: &str) -> bool {
    serde_json::from_str::<serde_json::Value>(line)
        .ok()
        .and_then(|value| {
            value
                .as_object()
                .map(|map| map.contains_key(ENTITLEMENT_KEY) || map.contains_key(RECLAIM_KEY))
        })
        .unwrap_or(false)
}

#[cfg(test)]
mod tests {
    use super::*;
    use gunnar_proto::{RefUpdate, RepoPath};

    fn sample(id: &str) -> PushAttestation {
        PushAttestation {
            attestation_id: id.to_owned(),
            repo: Some(RepoPath::new("team", "project")),
            principal_id: "tester".into(),
            principal_name: "tester".into(),
            key_fingerprint: "SHA256:zzz".into(),
            credential: "key:SHA256:zzz".into(),
            hash_kind: gunnar_proto::HashKind::Sha256 as i32,
            ref_updates: vec![RefUpdate {
                ref_name: "refs/heads/main".into(),
                old_oid: "0".repeat(64),
                new_oid: "a".repeat(64),
            }],
            new_object_oids: vec!["b".repeat(64)],
            new_object_oids_unknown: String::new(),
            new_object_count: 1,
            // Empty: this sample's object set is known, and the count above is
            // a count rather than an absence.
            new_object_bytes: 512,
            pack_hash: "c".repeat(64),
            received_at_unix_ms: 1,
            server_agent: gunnar_proto::AGENT.into(),
            transport: "ssh".into(),
        }
    }


    /// ★ **A line written by an older gunnar must still read.** The log is
    /// append-only and outlives every release; the 1.4.0 server on gunnar.rs
    /// refused `Repos.Chain` on alice/linux with *"attestation line 0 does
    /// not parse: missing field `new_object_oids_unknown`"* — a field 1.3.x
    /// never wrote, because it did not exist. Every field the schema grows
    /// after a log has lines is absent from those lines, so the reader must
    /// take an absent field as its default. Here: a 1.3.0-shaped line, with
    /// no `new_object_oids_unknown`, read back through the real reader.
    #[test]
    fn a_line_from_before_a_field_existed_still_reads() {
        let dir = tempfile::tempdir().expect("temp dir");
        let log = AttestationLog::under(dir.path());
        let mut value = serde_json::to_value(sample("old")).expect("a sample serialises");
        value
            .as_object_mut()
            .expect("an object")
            .remove("new_object_oids_unknown")
            .expect("the sample carried the field the old server lacked");
        std::fs::write(log.path(), format!("{value}\n")).expect("write a 1.3.0 line");

        let back = log
            .read()
            .expect("a line from before the field existed must still parse");
        assert_eq!(back.len(), 1);
        assert_eq!(back[0].attestation_id, "old");
        assert_eq!(back[0].new_object_oids_unknown, "", "absent reads as the default");
    }

    #[test]
    fn two_appends_are_two_lines_and_read_back_in_order() {
        let dir = tempfile::tempdir().expect("temp dir");
        let log = AttestationLog::under(dir.path());
        assert!(
            log.read().expect("an absent log reads as empty").is_empty(),
            "a log that has never been written must read as empty, not as an error"
        );

        log.append(&sample("first")).expect("append the first");
        log.append(&sample("second")).expect("append the second");

        // Applied output: the bytes on disk, not a round-trip through the same
        // struct. A writer that overwrote instead of appending would still make
        // `read()` return something.
        let raw = std::fs::read_to_string(log.path()).expect("read the raw file");
        assert_eq!(
            raw.lines().count(),
            2,
            "an append-only log must have grown by one line: {raw}"
        );
        assert!(raw.ends_with('\n'), "every record is newline-terminated");

        let back = log.read().expect("read the log");
        assert_eq!(
            back.iter()
                .map(|a| a.attestation_id.as_str())
                .collect::<Vec<_>>(),
            ["first", "second"],
            "the log must preserve arrival order"
        );
        assert_eq!(back[0].ref_updates[0].ref_name, "refs/heads/main");
        assert_eq!(back[0].new_object_count, 1);
    }

    /// **Every non-push record kind has to be in [`is_non_push_line`], and
    /// what it costs to forget is an INVENTED PUSH.**
    ///
    /// Seen RED 2026-09-16, with `RECLAIM_KEY` taken back out of the
    /// predicate: the reclaim line did not error, it came back as
    /// `(2, "")` — a `PushAttestation` with every field defaulted, because
    /// prost's serde impl tolerates an unknown key and fills the rest in.
    /// A record with no id, no repository, no principal and no pack hash then
    /// sits in the middle of the real ones, and `gunnar verify` proves it like
    /// any other leaf.
    ///
    /// That is why this asserts the exact `(line index, id)` pairs. "It still
    /// parses" would have been green for the broken predicate.
    #[test]
    fn a_record_that_is_not_a_push_is_skipped_and_its_index_still_belongs_to_it() {
        let dir = tempfile::tempdir().expect("temp dir");
        let log = AttestationLog::under(dir.path());

        log.append(&sample("before")).expect("a push");
        let entitlement = serde_json::json!({ ENTITLEMENT_KEY: { "tenant": "acme" } });
        assert_eq!(log.append_line(&entitlement.to_string()).expect("a fact"), 1);
        let reclaim =
            serde_json::json!({ RECLAIM_KEY: { "namespace": "acme", "bytes_freed": 4096 } });
        assert_eq!(log.append_line(&reclaim.to_string()).expect("a reclaim"), 2);
        log.append(&sample("after")).expect("another push");

        let indexed = log.read_indexed().expect("the log still reads as pushes");
        assert_eq!(
            indexed
                .iter()
                .map(|(i, a)| (*i, a.attestation_id.as_str()))
                .collect::<Vec<_>>(),
            [(0, "before"), (3, "after")],
            "the two non-push lines are skipped, and the pushes keep the LINE index a \
             proof is made against — never their position in this vector"
        );
    }

    /// **A power cut leaves a NUL tail, and a NUL tail must not kill every read
    /// of the log for ever.**
    ///
    /// The same failure this repository has already paid for once:
    /// `tests/appliance_powercut_recovery.rs` — "`ReplicationConfig::save` wrote
    /// `replication.json` and renamed it with no fsync, so the power cut left it
    /// **0 bytes** on xfs (rename journaled, data blocks not)". [`AttestationLog::append_line`]
    /// is the same shape: `O_APPEND`, one `write_all`, and no `sync_data` — while
    /// its own twin-side sibling (`twin::append_line_once`) does sync. xfs journals
    /// the size before the data blocks reach the platter, so the artifact of a cut
    /// is a line of NUL bytes at the end of the file.
    ///
    /// That line is not blank, so `lines()` keeps it; it is not JSON, so
    /// [`AttestationLog::read`] and [`AttestationLog::read_indexed`] hard-error on
    /// it — and every caller of theirs dies with them: `prove_pack` (a verifier
    /// holding an artifact), `Repos.Chain`, the whole evidence door. For ever,
    /// on an appliance with no shell to edit the file with.
    ///
    /// The record itself is NOT lost: `append_line` frames the line before it
    /// writes it, so the txlog holds it and the twin applied it. Only this box's
    /// copy is torn. So the good records before the tear must keep reading, and
    /// the tear must be named rather than parsed.
    #[test]
    fn a_nul_tail_from_a_power_cut_does_not_kill_every_read_of_the_log() {
        use std::io::Write as _;
        let dir = tempfile::tempdir().expect("temp dir");
        let log = AttestationLog::under(dir.path());
        log.append(&sample("before")).expect("a push");
        log.append(&sample("after")).expect("another push");

        // What the cut leaves: the size is there, the bytes are not.
        let mut f = std::fs::OpenOptions::new().append(true).open(log.path()).expect("open");
        f.write_all(&[0u8; 96]).expect("the torn tail");
        f.write_all(b"\n").expect("and the newline xfs had already journalled");
        drop(f);

        let back = log.read().expect("the records written before the cut still read");
        assert_eq!(back.iter().map(|a| a.attestation_id.as_str()).collect::<Vec<_>>(), ["before", "after"]);

        let indexed = log.read_indexed().expect("and so do their line indices");
        assert_eq!(indexed.iter().map(|(i, a)| (*i, a.attestation_id.as_str())).collect::<Vec<_>>(), [(0, "before"), (1, "after")]);
    }

    /// **A record kind this build does not know must not be DEFAULTED into a push
    /// that never happened.**
    ///
    /// Found while pinning the power cut, and worse than it: `PushAttestation`'s
    /// serde impl tolerates unknown keys and defaults everything else, so ANY JSON
    /// object that [`is_non_push_line`] does not name came back out of
    /// [`AttestationLog::read_indexed`] as a push. MEASURED on master 2026-09-16:
    /// one line of `{"this": "..."}` produced
    /// `(1, PushAttestation { attestation_id: "", repo: None, principal_id: "",
    /// pack_hash: "", .. })` sitting between the real records — an invented push
    /// that `prove_pack` would prove, `Repos.Chain` would list and `gunnar verify`
    /// would accept like any other leaf.
    ///
    /// The cost of forgetting to name a new record kind is therefore not an error
    /// a reader sees; it is a phantom. An attestation always has an id, so a record
    /// without one is refused by name.
    #[test]
    fn a_json_object_that_is_not_a_push_is_refused_and_never_defaulted_into_one() {
        let dir = tempfile::tempdir().expect("temp dir");
        let log = AttestationLog::under(dir.path());
        log.append(&sample("before")).expect("a push");
        log.append_line("{\"some_later_record_kind\": {\"nobody\": \"named it here\"}}").expect("a line");

        let err = log.read_indexed().expect_err("a record kind this build does not know");
        let say = format!("{err:#}");
        assert!(say.contains("no attestation id") && say.contains("never happened"), "{say}");
    }
}

// ───────────────────────────────────────────────────────────────────────────
// Push attestation, second half: the log becomes evidence
// ───────────────────────────────────────────────────────────────────────────

/// Where the signed tree heads are kept, beside the log they cover.
///
/// A second append-only file rather than a column in the first, because the two
/// have different shapes: the log gains a line per push, and a head is issued
/// when somebody asks. Putting heads in the same file would make every head a
/// record that the next head has to cover, which is a log that can never be
/// caught up with.
pub const TRANSPARENCY_HEADS: &str = "transparency-heads.jsonl";

/// Where signed build records are kept.
///
/// Its own file rather than lines in the attestation log, because the two are
/// different records with different readers: `attestations.jsonl` parses as
/// `PushAttestation` end to end, and a build record among them would break
/// every reader of it. They get the same Merkle machinery — [`TransparencyLog`]
/// is over *lines* — and their own heads.
pub const BUILD_LOG: &str = "builds.jsonl";

/// The signed tree heads over [`BUILD_LOG`].
pub const BUILD_HEADS: &str = "build-heads.jsonl";

/// Where signed forge events are kept — issues, reviews, merges.
///
/// **One file for the whole server**, like the attestation log and for the same
/// reason: the consumer wants the stream in arrival order, and the repository
/// is a field in every line. A file per repository would make "what happened on
/// this server today" a directory walk.
pub const FORGE_LOG: &str = "forge.jsonl";

/// The signed tree heads over [`FORGE_LOG`].
pub const FORGE_HEADS: &str = "forge-heads.jsonl";

/// The signed group-membership log: one line per add/remove, each an `sshsig`
/// the acting admin made under `gunnar-membership`. One file, in arrival order,
/// with the group in every line, for the same reason [`FORGE_LOG`] is one file.
pub const MEMBERSHIP_LOG: &str = "membership.jsonl";

/// The signed tree heads over [`MEMBERSHIP_LOG`].
pub const MEMBERSHIP_HEADS: &str = "membership-heads.jsonl";

/// **The transparency log**: the push-attestation log, plus the evidence that it
/// has not been rewritten.
///
/// # What it adds, and what it deliberately does not
///
/// It adds nothing to the attestation format. The Merkle tree is a **pure
/// function of the lines already on disk** — leaf *i* is
/// `SHA-256(0x00 ‖ line_i)` over the exact bytes of line *i*, no
/// canonicalisation, no re-serialisation — so there is no second encoding to
/// keep in step and no chance of hashing something other than what is stored.
///
/// A signed tree head at size N pins every line up to N. Rewriting line 3
/// changes the root, so **every head issued since** stops verifying against the
/// file: the tampering is detected by any holder of any later head, including
/// one who never saw the original line. That is the property, and
/// [`TransparencyLog::audit`] is the check.
///
/// What it does not add is a *consistency* proof between two heads (RFC 6962's
/// second proof type). A verifier with two heads can today detect that the file
/// changed under them; it cannot yet be shown that head M is a prefix-extension
/// of head N without recomputing from the records. That is a real gap and it is
/// named rather than papered over.
#[derive(Debug, Clone)]
pub struct TransparencyLog {
    records: PathBuf,
    heads: PathBuf,
}

/// One record's place in the log, and the proof that it is there.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Inclusion {
    /// Where in the log the record sits, counting from zero.
    pub index: usize,
    /// How many records the tree this proof is against holds.
    pub size: usize,
    /// The record's leaf hash, hex.
    pub leaf: String,
    /// The root the proof reconstructs, hex.
    pub root: String,
    /// The sibling hashes, hex, leaf-upwards.
    pub proof: Vec<String>,
}

/// What an [`TransparencyLog::audit`] found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AuditVerdict {
    /// Every stored head verifies, and every head's root matches the records
    /// the log holds at that size.
    Consistent { records: usize, heads: usize },
    /// A head does not verify against the key it names.
    HeadNotSigned { size: u64, why: String },
    /// A head verifies, and the records do not hash to the root it swore to.
    /// **This is tampering**: the head is authentic and the file no longer is.
    RecordsRewritten {
        size: u64,
        signed_root: String,
        actual_root: String,
    },
    /// A head covers more records than the log holds. Lines were removed.
    RecordsTruncated { size: u64, records: usize },
}

impl AuditVerdict {
    /// `true` only for [`AuditVerdict::Consistent`]. A caller that treats every
    /// other verdict as "some problem" is right to.
    pub fn is_consistent(&self) -> bool {
        matches!(self, AuditVerdict::Consistent { .. })
    }
}

impl std::fmt::Display for AuditVerdict {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AuditVerdict::Consistent { records, heads } => write!(
                f,
                "consistent: {records} record(s) under {heads} signed head(s)"
            ),
            AuditVerdict::HeadNotSigned { size, why } => {
                write!(f, "the head at size {size} does not verify: {why}")
            }
            AuditVerdict::RecordsRewritten {
                size,
                signed_root,
                actual_root,
            } => write!(
                f,
                "THE LOG HAS BEEN REWRITTEN: the head at size {size} was signed over root \
                 {signed_root} and the records now hash to {actual_root}"
            ),
            AuditVerdict::RecordsTruncated { size, records } => write!(
                f,
                "THE LOG HAS BEEN TRUNCATED: a head covers {size} record(s) and the log holds \
                 {records}"
            ),
        }
    }
}

impl TransparencyLog {
    /// The push-attestation log under a data root.
    pub fn under(root: impl AsRef<Path>) -> Self {
        let root = root.as_ref();
        TransparencyLog::over(root.join(ATTESTATION_LOG), root.join(TRANSPARENCY_HEADS))
    }

    /// **Any append-only JSONL file**, with its own heads beside it.
    ///
    /// The Merkle machinery cares about *lines*, not about what is in them, so
    /// the same log covers push attestations and build provenance without a
    /// second implementation (one implementation). What is record-specific is exactly two
    /// methods — [`Self::records`] and [`Self::prove_pack`] — and they say so.
    pub fn over(records: impl Into<PathBuf>, heads: impl Into<PathBuf>) -> Self {
        TransparencyLog {
            records: records.into(),
            heads: heads.into(),
        }
    }

    /// This log's records, **read as push attestations**.
    ///
    /// Only meaningful for the push log; a build log's lines are a different
    /// shape and this would refuse them. Everything else on this type is
    /// record-agnostic.
    pub fn records(&self) -> AttestationLog {
        AttestationLog::at(self.records.clone())
    }

    /// Where the records are.
    pub fn records_path(&self) -> &Path {
        &self.records
    }

    /// Where the signed heads are.
    pub fn heads_path(&self) -> &Path {
        &self.heads
    }

    /// The raw record lines, exactly as they are on disk.
    ///
    /// **The bytes, not the parsed attestations.** A leaf is the hash of what
    /// was written; re-serialising a parsed record could produce different
    /// bytes for the same push — a different field order, a different escape —
    /// and the root would then depend on the serialiser rather than on the log.
    fn lines(&self) -> anyhow::Result<Vec<String>> {
        let text = match std::fs::read_to_string(&self.records) {
            Ok(text) => text,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(err) => return Err(err.into()),
        };
        Ok(text
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(str::to_owned)
            .collect())
    }

    /// The record lines, for a caller that decodes them itself.
    ///
    /// `pub` because the build log's records are not `PushAttestation`s and
    /// have to be parsed by whoever knows their shape.
    pub fn lines_of(&self) -> anyhow::Result<Vec<String>> {
        self.lines()
    }

    fn leaves(&self) -> anyhow::Result<Vec<gunnar_core::transparency::Hash>> {
        Ok(self
            .lines()?
            .iter()
            .map(|line| gunnar_core::transparency::leaf_hash(line.as_bytes()))
            .collect())
    }

    /// How many records the log holds.
    pub fn len(&self) -> anyhow::Result<usize> {
        Ok(self.lines()?.len())
    }

    pub fn is_empty(&self) -> anyhow::Result<bool> {
        Ok(self.len()? == 0)
    }

    /// The current Merkle root, hex.
    pub fn root(&self) -> anyhow::Result<String> {
        Ok(gunnar_core::transparency::hex(
            &gunnar_core::transparency::root(&self.leaves()?),
        ))
    }

    /// Issue a signed tree head over everything the log currently holds, and
    /// append it.
    ///
    /// Returns the head. Issuing one over an **empty** log is allowed and
    /// meaningful: it is how a verifier learns the log exists and held nothing
    /// at that moment, which is the only way "this server has never accepted a
    /// push" can be evidence rather than an absence.
    pub fn issue_head(
        &self,
        key: &gunnar_core::ssh_key::PrivateKey,
        at_unix_ms: i64,
    ) -> anyhow::Result<gunnar_core::SignedTreeHead> {
        let leaves = self.leaves()?;
        let head = gunnar_core::TreeHead {
            size: leaves.len() as u64,
            root: gunnar_core::transparency::hex(&gunnar_core::transparency::root(&leaves)),
            at_unix_ms,
        };
        let signed = gunnar_core::transparency::sign_tree_head(key, head)?;
        let mut line = serde_json::to_string(&signed)?;
        line.push('\n');
        let _framed = gunnar_core::txlog::sink::append_line(&self.heads, line.trim_end_matches('\n'))?;
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.heads)?;
        file.write_all(line.as_bytes())?;
        Ok(signed)
    }

    /// Every signed head that has been issued, oldest first.
    pub fn heads(&self) -> anyhow::Result<Vec<gunnar_core::SignedTreeHead>> {
        let text = match std::fs::read_to_string(&self.heads) {
            Ok(text) => text,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(err) => return Err(err.into()),
        };
        text.lines()
            .filter(|line| !line.trim().is_empty())
            .map(|line| serde_json::from_str(line).map_err(anyhow::Error::from))
            .collect()
    }

    /// The inclusion proof for the record at `index`.
    ///
    /// `None` when the log does not hold that many records — a proof for a
    /// record that is not there is not an empty proof.
    pub fn prove(&self, index: usize) -> anyhow::Result<Option<Inclusion>> {
        let leaves = self.leaves()?;
        let Some(proof) = gunnar_core::transparency::inclusion_proof(&leaves, index) else {
            return Ok(None);
        };
        Ok(Some(Inclusion {
            index,
            size: leaves.len(),
            leaf: gunnar_core::transparency::hex(&leaves[index]),
            root: gunnar_core::transparency::hex(&gunnar_core::transparency::root(&leaves)),
            proof: proof.iter().map(gunnar_core::transparency::hex).collect(),
        }))
    }

    /// Find the record for a pack hash, and prove it.
    ///
    /// The lookup a verifier actually makes: it holds an artifact, the artifact
    /// names a pack, and it wants to know the push that landed it. Scans, which
    /// is honest about the cost — an index over the log is a later problem and
    /// pretending one exists would be worse than the scan.
    pub fn prove_pack(&self, pack_hash: &str) -> anyhow::Result<Option<Inclusion>> {
        // `read_indexed`, not `read` + `position`: an entitlement line between
        // two pushes is a leaf too, so a push's vector position is not its
        // line index, and a proof made against it would not verify.
        let records = self.records().read_indexed()?;
        let Some((index, _)) = records.iter().find(|(_, r)| r.pack_hash == pack_hash) else {
            return Ok(None);
        };
        self.prove(*index)
    }

    /// **Check every signed head against the records on disk.**
    ///
    /// This is what the log is for. It answers three different questions, and
    /// keeping them apart is the point:
    ///
    /// * a head that does not verify means somebody wrote a head they could not
    ///   sign — or the key rotated and the verifier was not told;
    /// * a head that *does* verify over a root the records no longer produce
    ///   means the **records** were changed after they were signed for, which is
    ///   tampering and nothing else;
    /// * a head covering more records than exist means lines were removed.
    ///
    /// The first verdict wins and is returned; an operator fixes one thing at a
    /// time.
    pub fn audit(&self, key: &gunnar_core::ssh_key::PublicKey) -> anyhow::Result<AuditVerdict> {
        let leaves = self.leaves()?;
        let heads = self.heads()?;
        for head in &heads {
            if let Err(problem) = gunnar_core::transparency::verify_tree_head(head, key) {
                return Ok(AuditVerdict::HeadNotSigned {
                    size: head.head.size,
                    why: problem.to_string(),
                });
            }
            let size = head.head.size as usize;
            if size > leaves.len() {
                return Ok(AuditVerdict::RecordsTruncated {
                    size: head.head.size,
                    records: leaves.len(),
                });
            }
            // The root over the FIRST `size` records: a head is a claim about a
            // prefix, and checking it against the whole file would fail on
            // every head but the newest.
            let actual =
                gunnar_core::transparency::hex(&gunnar_core::transparency::root(&leaves[..size]));
            if actual != head.head.root {
                return Ok(AuditVerdict::RecordsRewritten {
                    size: head.head.size,
                    signed_root: head.head.root.clone(),
                    actual_root: actual,
                });
            }
        }
        Ok(AuditVerdict::Consistent {
            records: leaves.len(),
            heads: heads.len(),
        })
    }
}