agentplane 0.46.0

//! Checking a plane's history without trusting the plane.
//!
//! # Why this is a deliverable and not a property
//!
//! Every mechanism underneath — the hash chain, per-record signatures, the
//! Merkle log — is only *checkable*. Somebody has to actually check it, and if
//! the only code that can is inside the runtime being audited, the claim
//! collapses: the party under examination is also the party running the
//! examination.
//!
//! So this module is deliberately shaped to run **against a store it did not
//! write**, with inputs an auditor holds rather than inputs the plane supplies:
//!
//! * a **prior checkpoint** they were given earlier — the one artifact that has
//!   to have left the operator's control;
//! * a **public key**, if they were told which workload should have signed.
//!
//! Neither is required, and what can be concluded shrinks accordingly. That
//! shrinkage is reported rather than hidden, because an audit that says "fine"
//! when it checked three things out of five is worse than one that checked
//! nothing.
//!
//! # What each input buys
//!
//! | Given | Answers |
//! |---|---|
//! | nothing | Is each run's chain internally consistent? |
//! | a public key | Who wrote each record? |
//! | a prior checkpoint | Has anything been **removed** since it was issued? |
//!
//! Only the third detects deletion, and only because the checkpoint came from
//! outside. That is the whole architecture of the thing in one row.

use std::sync::Arc;

use crate::core::{Digest, RunId, StoreError, Verifier, merkle};
use crate::journal::{Checkpoint, JournalStore, Record};

/// What an audit concluded, and what it could not look at.
///
/// `Serialize` is deliberate and load-bearing: the independent party this
/// report exists for should not have to link this crate to read it. The
/// findings render as the sentences they display as, because an auditor reads
/// prose and a machine that wants structure has the run ids beside it.
///
/// # What no report here can speak for: the runs that never started
///
/// A run refused **before it exists** — a policy denial on
/// [`ACTION_ADMIT`](crate::core::ACTION_ADMIT), a tenant ceiling, a standing
/// halt — has no run id and therefore no chain to append to. Nothing about it
/// is in this report, and no amount of reading the journal will find it.
///
/// Stated because *how often did policy stop a run from starting* is a question
/// an auditor asks, and a clean report answers it with silence rather than with
/// zero. The number lives in the `agentplane.policy.denials` metric and
/// `agentplane.policy.denied` telemetry, outside the hash chain.
///
/// It is not in [`not_checked`](Self::not_checked): that reports what *this*
/// audit could not check, and an entry in every report ever produced would train
/// a reader to skip the list.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct AuditReport {
    /// The checkpoint the store reports now.
    pub current: Checkpoint,
    /// The checkpoint this audit was held to, when it was given one.
    ///
    /// **A report that names its evidence, because the sibling reports do and
    /// this is the one where it matters most.** `RestoreReport` carries both
    /// the checkpoint it expected and the one it rebuilt; `VerifyReport`
    /// carries the file's. This carried only `current` — the store's own claim
    /// — so a clean report was indistinguishable from a clean report against
    /// an anchor, and the reader could not tell which history the append-only
    /// check had actually compared against.
    ///
    /// Empty says the deletion check did not run, which
    /// [`not_checked`](Self::not_checked) also says in words. Both, because
    /// this one is machine-readable and that one is for a person.
    ///
    /// **Every anchor, not the strongest one.** A reader has to be able to see
    /// which observations the store was held to, because the answer to *is
    /// this history forked* is only as strong as the set: one honest observer
    /// among several is enough to expose a fork, and a report naming one
    /// anchor cannot say whether that observer was consulted.
    ///
    /// Each carries the auditor's own label for where it came from. That is
    /// provenance this report did not check — an audit can verify a checkpoint
    /// *extends* and cannot verify who vouched for it — and it is recorded so
    /// a finding names the observer to go and ask.
    pub held_to: Vec<Anchor>,
    /// Runs whose chain, signatures and inclusion all checked out.
    ///
    /// An **open** run — one whose last conclusion does not seal, or which has
    /// no conclusion yet — appears here on chain and signatures alone: it has
    /// no Merkle leaf, so there is no inclusion to check, and [`not_checked`]
    /// says so once rather than a finding saying it per run. A run whose own
    /// records carry a *sealing* conclusion but which the log holds no leaf
    /// for is the opposite case, and that one is a finding.
    ///
    /// [`not_checked`]: Self::not_checked
    pub sound: Vec<RunId>,
    /// What went wrong, in the order found.
    ///
    /// Serialised as the rendered sentence rather than as a tagged variant: the
    /// consumer is a person or a SIEM, and a variant name is this crate's
    /// internal vocabulary. The run id each finding names is in the text.
    #[serde(serialize_with = "as_sentences")]
    pub findings: Vec<Finding>,
    /// Checks that were not performed, and why.
    ///
    /// Reported as loudly as failures. An audit that quietly skipped signature
    /// verification because no key was supplied, and then said "verified", is
    /// exactly the reassuring-but-empty artifact this crate exists to avoid.
    pub not_checked: Vec<String>,
    /// Every point at which a label was raised, in the order found.
    ///
    /// Not a finding — a release is a legitimate, authorized decision, and
    /// flagging it as a problem would train a reader to ignore the list. It is
    /// reported because it is the **only discretionary act in the system**: the
    /// chain, the signatures and the inclusion proofs all verify that history is
    /// intact, and none of them surfaces the moment somebody decided untrusted
    /// data could be treated as trusted. An auditor verifying integrity while
    /// never seeing that is checking the envelope and not the letter.
    ///
    /// Each entry answers the questions the decision was required to record:
    /// who, on what basis, toward what destination, over which fields, on what
    /// evidence.
    pub releases: Vec<ReleaseRecord>,
    /// What authorized each run: the declaration it ran under, and the policy
    /// bundle that governed it.
    ///
    /// Not a finding, for the same reason `releases` is not: an authorized run
    /// is the ordinary case. It is reported because **an audit that verifies
    /// history is intact and never says what warranted it has checked the
    /// letter and not the warrant** — the mirror of the argument one field up.
    ///
    /// The load-bearing entry is the one where `policy` is `None`. A run that
    /// executed with **no policy engine configured at all** verifies exactly as
    /// soundly as a governed one: its chain is intact, its signatures check, its
    /// leaf is included. Nothing in an integrity report distinguishes them, so
    /// an auditor reading `sound` would conclude a run was governed when the
    /// deployment had no gate wired. *Was policy switched on for this run* is an
    /// audit question the journal answers and this report did not surface.
    ///
    /// The digest is what makes the declaration half meaningful: a name and
    /// version identify a file that may since have been edited, and only the
    /// digest pins what it actually said — the system prompt included.
    ///
    /// Read it with [`unadmitted`](Self::unadmitted): every run whose chain
    /// verified appears in exactly one of the two.
    pub warrants: Vec<Warrant>,
    /// The runs that carry no admission record — the complement of
    /// [`warrants`](Self::warrants). See [`Unadmitted`].
    pub unadmitted: Vec<Unadmitted>,
    /// Per witness key, the time since it last saw the log, when freshness
    /// was judged. A suffix removed inside a window is undetectable: the
    /// window bounds *when* truncation could have happened, not *whether*.
    pub unwitnessed: Vec<UnwitnessedWindow>,
}

/// How long one witness key has gone without seeing the log.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct UnwitnessedWindow {
    /// The cosigning key.
    pub key_id: crate::core::KeyId,
    /// Its latest signed timestamp, in seconds since the Unix epoch.
    pub since: u64,
    /// Seconds from `since` to the auditor's `now`; zero when `since` is ahead.
    pub seconds: u64,
}

/// The auditor's bound on how old a witness's latest timestamp may be.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Freshness {
    /// The auditor's clock — never the plane's.
    pub now: crate::core::Timestamp,
    /// The oldest a witness key's latest timestamp may be at `now`, and the
    /// furthest ahead of `now` one may be.
    pub max_age: std::time::Duration,
}

/// What authorized one run.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct Warrant {
    /// The run this describes.
    pub run: RunId,
    /// The agent declaration that governed it, if one did.
    ///
    /// `None` for a run started from code rather than from a manifest, which is
    /// a legitimate shape and a different one from a manifest-governed run.
    pub declaration: Option<crate::journal::AgentIdentity>,
    /// The complete policy bundle that governed it.
    ///
    /// `None` means **no engine was configured**. That is the entry an auditor
    /// most needs and the one an integrity-only report cannot show.
    pub policy: Option<crate::core::PolicyBundleIdentity>,
}

/// A run whose history carries no admission record.
///
/// Its existence is the other half of [`Warrant`], and the pair is what makes
/// *what authorized this* answerable for **every** run rather than for the ones
/// that happen to have an answer. A run with no `RunAdmitted` produces no
/// warrant, so listing warrants alone leaves a reader with two lists of
/// different lengths and nothing saying why — and the entry that goes missing is
/// the one [`Warrant`]'s own documentation calls load-bearing.
///
/// Not a finding on its own. Some runs are unadmitted by construction: this
/// plane writes a run of its own for the decisions a sweep takes without a
/// request, and the same shape is what a record *about* an agent this plane does
/// not execute would take. The outcome is carried so a reader can tell those
/// from a run that was supposed to be admitted and was not — the audit reports
/// the fact and leaves the judgement to somebody who knows which outcomes this
/// deployment writes.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct Unadmitted {
    /// The run this describes.
    pub run: RunId,
    /// The outcome it concluded under, when it has concluded.
    ///
    /// `None` for a run still open, which is a different statement from a run
    /// that ended without ever being admitted.
    pub outcome: Option<String>,
}

/// One journaled decision to improve a label.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct ReleaseRecord {
    pub run: RunId,
    /// The agent or operator the decision was recorded against.
    pub releaser: String,
    /// Why, in the releaser's own words.
    pub basis: String,
    /// Where the value was being released *to*. A release is always toward
    /// something; one with no destination is a permission with no boundary.
    pub destination: String,
    /// Which fields moved — `""` for the whole value.
    pub fields: Vec<String>,
    /// What was cited. An empty set is impossible: `Release::validate` refuses
    /// it, and this being non-empty is that rule observed from the outside.
    pub evidence: Vec<String>,
    /// The digest of the value that was released, so a reader can tie the
    /// decision to the bytes rather than to a description of them.
    pub value: Digest,
}

impl AuditReport {
    /// Whether every check that ran, passed.
    ///
    /// Note the qualifier. A report with findings is a failure; a report with
    /// *no* findings and a long `not_checked` is not a pass, and callers are
    /// expected to look. [`Self::assert_complete`] is the strict form.
    #[must_use]
    pub fn is_sound(&self) -> bool {
        self.findings.is_empty()
    }

    /// Panic unless everything passed **and** everything was checkable.
    ///
    /// # Panics
    ///
    /// If anything failed, or if any check was skipped.
    pub fn assert_complete(&self) {
        assert!(
            self.findings.is_empty(),
            "the audit found {} problem(s):\n{}",
            self.findings.len(),
            self.findings
                .iter()
                .map(|f| format!("  • {f}"))
                .collect::<Vec<_>>()
                .join("\n")
        );
        assert!(
            self.not_checked.is_empty(),
            "the audit passed but could not check everything:\n{}",
            self.not_checked
                .iter()
                .map(|s| format!("  • {s}"))
                .collect::<Vec<_>>()
                .join("\n")
        );
    }
}

/// Render findings as the sentences they display as.
fn as_sentences<S: serde::Serializer>(f: &[Finding], s: S) -> Result<S::Ok, S::Error> {
    use serde::ser::SerializeSeq;
    let mut seq = s.serialize_seq(Some(f.len()))?;
    for finding in f {
        seq.serialize_element(&finding.to_string())?;
    }
    seq.end()
}

/// One thing wrong with a plane's history.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum Finding {
    #[error("run {run}: {detail}")]
    Chain { run: RunId, detail: String },

    #[error("run {run} is sealed but is not in the log — no checkpoint covers it")]
    NotInLog { run: RunId },

    #[error("run {run} claims a position the log's own root does not support")]
    BadInclusion { run: RunId },

    /// The chain the store *served* is not the chain the log committed to.
    ///
    /// The one that catches a truncated-but-internally-consistent record set:
    /// a prefix of a chain verifies on its own, and the leaf the log holds is
    /// the terminal hash of the *whole* run — so an audit that verified the
    /// records and then checked the store-supplied leaf against the tree,
    /// without ever holding the two to each other, was verifying two halves of
    /// two different claims.
    #[error(
        "run {run}: the log's leaf is not the verified chain's head — records were \
         removed or replaced after sealing, and the served history is not the one \
         the checkpoint commits to"
    )]
    LeafMismatch { run: RunId },

    /// The sealing record's own claim disagrees with the chain it sits in.
    ///
    /// `RunSealed.chain_head` is the head the conclusion was drawn over — by
    /// construction, the record's own `prev_hash`. A mismatch means the
    /// conclusion was composed against a different history than the one it was
    /// appended to, which no honest writer produces.
    #[error(
        "run {run}: the sealing record claims a chain head that is not the head it \
         sits on — the conclusion was drawn over a different history"
    )]
    SealClaim { run: RunId },

    /// A sealed conclusion over an undecided transactional unit.
    ///
    /// `GroupOpened`/`GroupSettled` bracket several effects that take together
    /// or not at all, and the settlement is the most consequential thing a
    /// group does. A run still open with a group unsettled is the ordinary
    /// crash shape — the resume re-walks the members and settles, and the run
    /// itself sits in a findable backlog until it does. A **sealed** run is
    /// the state no honest writer produces: nothing may resume it, so nothing
    /// will ever settle the group, and whether its members were taken or taken
    /// back is permanently unanswerable from a history that claims to be
    /// complete.
    #[error(
        "run {run} is sealed but group '{group}' was opened and never settled — \
         nothing may resume a sealed run, so whether the group's members were \
         taken or taken back is permanently undecided"
    )]
    GroupUnsettled { run: RunId, group: String },

    /// A sealed conclusion over an effect whose outcome was never established.
    ///
    /// The finding that outlives the run. A quarantine is a *status*, and a
    /// status is something a later action overwrites: abandoning the run takes
    /// it off the quarantine backlog, which is the only listing that carried
    /// it. What the run left in the world does not go away with the listing, so
    /// the record of it is derived from the journal instead — where nothing an
    /// operator does can take it off.
    ///
    /// Under a **sealing** conclusion only, and for the reason
    /// [`GroupUnsettled`](Self::GroupUnsettled) is: an open run with an
    /// undecided effect is the ordinary crash shape, healed by a resume or
    /// answered by a person, and flagging it would teach the reader this
    /// finding is weather.
    ///
    /// Mutating effects only. A read that never came back is safe to repeat and
    /// changed nothing, so there is nothing here for an auditor to act on.
    #[error(
        "run {run} is sealed but effect {effect} (step {step}, {doubt}) never reached a known \
         outcome — nothing may resume a sealed run, so whether the call changed the outside \
         world is permanently undecided"
    )]
    EffectUndecided {
        run: RunId,
        step: crate::core::StepId,
        effect: crate::core::EffectKey,
        /// Whether the runtime never heard back, or heard back and was told
        /// nothing — two different places for an investigator to start.
        doubt: &'static str,
    },

    /// The one that needs an outside artifact.
    ///
    /// Names **which** anchor it failed against, because an audit is held to
    /// every checkpoint an auditor brought rather than to one: a fork is
    /// visible from whichever observer saw the history it diverged from, and
    /// that observer is the first thing an investigator has to go and ask.
    #[error(
        "the log cannot prove it only grew since the checkpoint of size {old_size} held by \
         {obtained_from} — something committed to earlier is no longer committed to now"
    )]
    NotAppendOnly {
        old_size: u64,
        obtained_from: String,
    },

    #[error("the checkpoint held by {obtained_from} names log '{theirs}', this store is '{ours}'")]
    WrongLog {
        theirs: String,
        ours: String,
        obtained_from: String,
    },

    #[error(
        "the checkpoint held by {obtained_from} is larger ({old_size}) than this log ({now}) — \
         a log cannot shrink, so runs were removed or this is a different plane"
    )]
    Shrunk {
        old_size: u64,
        now: u64,
        obtained_from: String,
    },

    /// A witness key has not seen the log for longer than the auditor allows.
    /// Not attributed: plane silence, a failed submission and a witness
    /// outage look the same from outside.
    #[error(
        "witness key {key_id} ({obtained_from}) last saw this log at {timestamp}, \
         {age_secs}s ago — older than the maximum age, so anything appended and \
         removed since is invisible to it"
    )]
    StaleWitness {
        key_id: crate::core::KeyId,
        obtained_from: String,
        timestamp: u64,
        age_secs: u64,
    },

    /// A witness key's signed time is ahead of the auditor's clock by more
    /// than the maximum age, so it cannot be read as fresh.
    #[error(
        "witness key {key_id} ({obtained_from}) signed time {timestamp}, {ahead_secs}s \
         ahead of this audit's clock — more than the maximum age, so it is not read as fresh"
    )]
    WitnessTimeAhead {
        key_id: crate::core::KeyId,
        obtained_from: String,
        timestamp: u64,
        ahead_secs: u64,
    },
}

pub use crate::journal::{Anchor, WitnessTime};

/// What an auditor brought with them.
///
/// Hand-written `Debug` because a `Verifier` is a trait object with no useful
/// rendering, and deriving would demand one.
#[derive(Default)]
pub struct Evidence<'a> {
    /// Every checkpoint issued earlier, from outside this store.
    ///
    /// **All of them, and that is the point.** Each is an independent
    /// observation of this log, so each is an independent constraint: the
    /// store must prove it extends every one. Holding the store to a single
    /// anchor — the largest, say — is exactly the shape an equivocating
    /// operator wants, because the fork it feeds a fresh observer is the
    /// longest history anybody holds, and the observer that saw the honest
    /// prefix is the one whose answer gets dropped.
    pub anchors: &'a [Anchor],
    /// The key the records should carry.
    pub verifier: Option<&'a dyn Verifier>,
    /// Whether an unsigned record is a failure.
    ///
    /// Off by default: history written before signing was configured is
    /// legitimately unsigned, and an auditor who does not know that would read a
    /// wall of failures for a plane that is fine.
    pub require_signatures: bool,
    /// The auditor's clock and maximum age, against which each witness key's
    /// latest signed timestamp is judged. `None` judges nothing.
    pub freshness: Option<Freshness>,
}

impl std::fmt::Debug for Evidence<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Evidence")
            .field("anchors", &self.anchors)
            .field("verifier", &self.verifier.is_some())
            .field("require_signatures", &self.require_signatures)
            .field("freshness", &self.freshness)
            .finish()
    }
}

/// Where one run stands relative to the Merkle log.
enum Placement {
    /// In the log, at a position the root supports.
    Sound,
    /// Not in the log because it has not concluded with a sealing outcome —
    /// a state, not a defect. Chain and signatures were verified upstream.
    Open,
    /// The run's own records carry a sealing conclusion, and the log holds no
    /// leaf for it: history the log no longer commits to.
    NotInLog,
    /// The log holds a leaf that is not the verified chain's head: the served
    /// records are not the history the log committed to.
    LeafMismatch,
    /// In the log by its own claim, at a position the root does not support.
    BadInclusion,
    /// The log grew faster than the audit could catch its checkpoint up, so
    /// nothing ties this run's proof to one root. Honest, and rare: it takes a
    /// seal landing between two adjacent store calls, twice.
    Unpinned,
}

/// Decide one run's [`Placement`], catching `current` up if the log grew.
///
/// The catch-up is the part that earns a comment: the log can grow while the
/// audit walks it, and a proof computed against a larger tree than the
/// checkpoint in hand fails for a reason that is time, not tampering — a false
/// integrity alarm teaches the reader to ignore the true one. One refresh
/// covers the realistic race; a plane sealing continuously lands in
/// [`Placement::Unpinned`] rather than in a finding.
async fn placement(
    store: &Arc<dyn JournalStore>,
    run: RunId,
    records: &[Record],
    head: Digest,
    current: &mut Checkpoint,
) -> Result<Placement, StoreError> {
    let Some(inc) = store.inclusion_proof(run).await? else {
        // The store holds no leaf, and whether that is a finding is decided by
        // the run's own records rather than assumed. An **open** run — failed,
        // exhausted, still executing — was never in the log; reporting it as a
        // defect would flag every healthy resumable run and teach the reader
        // to skim past the flag that matters.
        return Ok(if has_sealing_conclusion(records) {
            Placement::NotInLog
        } else {
            Placement::Open
        });
    };

    // The leaf must be the verified chain's own head, checked **before** the
    // tree math and independent of any checkpoint race. `inc.seal` is
    // store-supplied; the head was recomputed from the served bytes. Verifying
    // each without holding them to each other let a store serve a truncated —
    // but internally consistent — prefix of a sealed run and have both halves
    // pass: the prefix's chain verifies, and the log's genuine leaf verifies
    // against the genuine tree.
    if inc.seal != head {
        return Ok(Placement::LeafMismatch);
    }

    if inc.size != current.size {
        *current = store.checkpoint().await?;
    }
    if inc.size != current.size {
        return Ok(Placement::Unpinned);
    }
    let leaf = merkle::leaf_hash(&inc.seal);
    let ok = merkle::verify_inclusion(
        leaf,
        usize::try_from(inc.index).unwrap_or(usize::MAX),
        usize::try_from(inc.size).unwrap_or(0),
        &inc.proof,
        &current.root,
    );
    Ok(if ok {
        Placement::Sound
    } else {
        Placement::BadInclusion
    })
}

/// Whether the run's last conclusion is one nothing may resume.
///
/// The outcome list is the library's own ([`crate::runtime::SEALED_OUTCOMES`]),
/// not a re-spelling of it: two copies of *which conclusions close* is the
/// duplicate-rule shape, and the copy in an offline checker is the one that
/// drifts.
pub(crate) fn has_sealing_conclusion(records: &[Record]) -> bool {
    concluded_outcome(records).is_some_and(|o| crate::runtime::SEALED_OUTCOMES.contains(&o))
}

/// The outcome a run concluded under, if it has concluded.
///
/// One traversal shared by the seal question and the unadmitted report, because
/// two copies of *which record carries the outcome* is the duplicate-rule shape
/// and the copy nobody edits is the one that drifts.
fn concluded_outcome(records: &[Record]) -> Option<&str> {
    records.iter().rev().find_map(|r| match r.kind() {
        crate::journal::RecordKind::RunConcluded { outcome, .. } => Some(outcome.as_str()),
        _ => None,
    })
}

/// What this run left permanently undecided, if it may never resume.
///
/// Both findings share a scope and an argument, so they share a function. An
/// **open** run's unsettled group or in-doubt effect is the ordinary crash
/// shape — a resume settles the one, a person answers the other, and the run
/// sits in a backlog somebody drains until then. Flagging those would teach the
/// reader this finding is weather. Under a **sealing** conclusion nothing may
/// resume, so whether the work was taken or taken back is unanswerable from a
/// history that claims to be complete.
///
/// The doubt half reads
/// [`undecided_effects`](crate::journal::undecided_effects) rather than
/// restating it, so an offline audit and a live runtime cannot disagree about
/// what is in doubt.
fn permanently_undecided(run: RunId, records: &[Record]) -> Vec<Finding> {
    if !has_sealing_conclusion(records) {
        return Vec::new();
    }
    unsettled_groups(records)
        .into_iter()
        .map(|group| Finding::GroupUnsettled { run, group })
        .chain(
            crate::journal::undecided_effects(records)
                .into_iter()
                .map(|u| Finding::EffectUndecided {
                    run,
                    step: u.step,
                    effect: u.effect,
                    doubt: u.doubt.as_str(),
                }),
        )
        .collect()
}

/// Groups opened and never settled, in the order opened.
///
/// Counted per writing step, phase and name — two *distinct* groups may
/// legitimately share a name within one step (opened, settled, opened again),
/// so each settlement excuses exactly one opening, which is the same
/// arithmetic the executor's own resume bookkeeping uses.
fn unsettled_groups(records: &[Record]) -> Vec<String> {
    use std::collections::BTreeMap;
    type Key<'a> = (Option<crate::core::StepId>, crate::core::Phase, &'a str);
    let mut open: BTreeMap<Key<'_>, u64> = BTreeMap::new();
    let mut order: Vec<Key<'_>> = Vec::new();
    for r in records {
        match r.kind() {
            crate::journal::RecordKind::GroupOpened { group, .. } => {
                let key = (r.body.step, r.body.phase, group.as_str());
                *open.entry(key).or_insert(0) += 1;
                order.push(key);
            }
            crate::journal::RecordKind::GroupSettled { group, .. } => {
                if let Some(n) = open.get_mut(&(r.body.step, r.body.phase, group.as_str())) {
                    *n = n.saturating_sub(1);
                }
            }
            _ => {}
        }
    }
    // Whatever count a settlement did not excuse is unsettled. Leftovers are
    // attributed newest-first, because a settlement pairs with the most recent
    // opening of its name still standing.
    let mut out = Vec::new();
    for key in order.into_iter().rev() {
        if let Some(n) = open.get_mut(&key)
            && *n > 0
        {
            *n -= 1;
            out.push(key.2.to_owned());
        }
    }
    out.reverse();
    out
}

/// Every label-raising decision in one run's history.
fn releases_in(run: RunId, records: &[Record]) -> Vec<ReleaseRecord> {
    records
        .iter()
        .filter_map(|record| match record.kind() {
            crate::journal::RecordKind::Released {
                releaser,
                release,
                value,
                ..
            } => Some(ReleaseRecord {
                run,
                releaser: releaser.clone(),
                basis: release.basis().to_owned(),
                destination: release.destination().to_owned(),
                fields: release.fields_scope().iter().cloned().collect(),
                evidence: release.evidence().iter().cloned().collect(),
                value: *value,
            }),
            _ => None,
        })
        .collect()
}

/// What authorized one run, from the record that opened it.
///
/// `RunAdmitted` carries both, and carries them once: the declaration because
/// *which manifest governed this* must be answerable years later, and the bundle
/// because *was policy switched on* must be too. Reading them here rather than
/// re-deriving from today's wiring is the whole point — an audit runs against a
/// store it did not write, on a machine that may have no engine configured at
/// all.
fn warrant_in(run: RunId, records: &[Record]) -> Option<Warrant> {
    records.iter().find_map(|record| match record.kind() {
        crate::journal::RecordKind::RunAdmitted {
            governed_by,
            policy_bundle,
            ..
        } => Some(Warrant {
            run,
            declaration: governed_by.as_deref().cloned(),
            policy: policy_bundle.as_deref().cloned(),
        }),
        _ => None,
    })
}

/// Each witness key's latest signed timestamp, judged on its own against the
/// auditor's clock. Keys are never compared with each other.
fn judge_freshness(
    evidence: &Evidence<'_>,
    findings: &mut Vec<Finding>,
    not_checked: &mut Vec<String>,
) -> Vec<UnwitnessedWindow> {
    let mut latest: std::collections::BTreeMap<&str, (u64, &str)> =
        std::collections::BTreeMap::default();
    for held in evidence.anchors {
        for time in &held.witnessed {
            let entry = latest
                .entry(time.key_id.as_str())
                .or_insert((time.timestamp, held.obtained_from.as_str()));
            if time.timestamp > entry.0 {
                *entry = (time.timestamp, held.obtained_from.as_str());
            }
        }
    }
    let Some(freshness) = evidence.freshness else {
        if !latest.is_empty() {
            not_checked.push(
                "freshness — the anchors carry witness times and no maximum age was \
                 supplied, so how long each witness has gone without seeing this log \
                 was not judged"
                    .to_owned(),
            );
        }
        return Vec::new();
    };
    if latest.is_empty() {
        not_checked.push(
            "freshness — a maximum age was supplied and no anchor came from a witness, \
             so there is no signed time to judge"
                .to_owned(),
        );
        return Vec::new();
    }
    let now = u64::try_from(freshness.now.unix_timestamp()).unwrap_or(0);
    let max_age = freshness.max_age.as_secs();
    let mut windows = Vec::new();
    for (key_id, (timestamp, obtained_from)) in latest {
        let age = now.saturating_sub(timestamp);
        let ahead = timestamp.saturating_sub(now);
        if ahead > max_age {
            findings.push(Finding::WitnessTimeAhead {
                key_id: key_id.to_owned(),
                obtained_from: obtained_from.to_owned(),
                timestamp,
                ahead_secs: ahead,
            });
        } else if age > max_age {
            findings.push(Finding::StaleWitness {
                key_id: key_id.to_owned(),
                obtained_from: obtained_from.to_owned(),
                timestamp,
                age_secs: age,
            });
        }
        windows.push(UnwitnessedWindow {
            key_id: key_id.to_owned(),
            since: timestamp,
            seconds: age,
        });
    }
    not_checked.push(
        "truncation inside the unwitnessed window — a suffix appended and removed after \
         each witness key's latest time is undetectable; the window bounds when \
         truncation could have happened, not whether"
            .to_owned(),
    );
    windows
}

/// What an audit cannot conclude from the evidence it was given.
///
/// Gathered before any check runs, so a report is never missing a line because
/// the walk stopped early. Why it is reported at all is on
/// [`AuditReport::not_checked`], which is where a reader meets it.
fn missing_evidence(evidence: &Evidence<'_>) -> Vec<String> {
    let mut out = Vec::new();
    if evidence.verifier.is_none() {
        out.push(
            "signatures — no public key was supplied, so this audit cannot say who \
             wrote anything"
                .to_owned(),
        );
    }
    if evidence.anchors.is_empty() {
        out.push(
            "deletion — no earlier checkpoint was supplied, so this audit cannot \
             detect a run that was removed. Every check below passes over a store \
             somebody emptied, because the records, the leaves and the root all \
             come from the party being audited. Pass the checkpoint an earlier \
             audit printed, or the one this plane's witnesses hold"
                .to_owned(),
        );
    } else if evidence.anchors.len() == 1 {
        out.push(format!(
            "equivocation — one anchor was supplied ({}), so this audit says the store \
             extends that observer's history and nothing about whether a second observer \
             holds a different one. A fork is visible only from an observer that saw the \
             history it diverged from, so bring every checkpoint you can obtain",
            evidence.anchors[0].obtained_from
        ));
    }
    out
}

/// The sealing record's own claim, held to the chain it sits in.
///
/// `RunSealed.chain_head` is the head the conclusion was drawn over, which is
/// by construction its own record's `prev_hash` — checkable only after the
/// chain has verified, so `prev_hash` is evidence rather than input. A run
/// with no conclusion has made no claim, and holds vacuously.
fn seal_claim_holds(records: &[Record]) -> bool {
    records
        .iter()
        .rev()
        .find_map(|r| match r.kind() {
            crate::journal::RecordKind::RunConcluded { chain_head, .. } => {
                Some(*chain_head == r.prev_hash)
            }
            _ => None,
        })
        .unwrap_or(true)
}

/// The deletion check, against the checkpoint the auditor brought.
///
/// The proof must be paired with the checkpoint it is verified against, and
/// the log can grow between the two store calls — the same race `placement`
/// refreshes for, handled the same way, because two halves of one audit
/// reporting one race differently teaches the reader that findings are
/// weather. A proof computed over a log larger than the checkpoint in hand
/// fails for a reason that is time, not tampering, and a false `NotAppendOnly`
/// is the alarm this whole module exists to make believable. One refresh
/// covers the realistic race; a plane sealing continuously lands in
/// `not_checked` rather than in a finding.
///
/// What this does NOT cover: it decides nothing about tampering on the
/// unpinned path — a store that really did rewrite history and also keeps
/// growing is only caught by re-running against a quiesced store, which the
/// entry says in words.
async fn check_append_only(
    store: &Arc<dyn JournalStore>,
    anchor: &Anchor,
    current: &mut Checkpoint,
    findings: &mut Vec<Finding>,
    not_checked: &mut Vec<String>,
) -> Result<(), StoreError> {
    let prior = &anchor.checkpoint;
    if prior.origin != current.origin {
        findings.push(Finding::WrongLog {
            theirs: prior.origin.clone(),
            ours: current.origin.clone(),
            obtained_from: anchor.obtained_from.clone(),
        });
        return Ok(());
    }
    if prior.size > current.size {
        findings.push(Finding::Shrunk {
            old_size: prior.size,
            now: current.size,
            obtained_from: anchor.obtained_from.clone(),
        });
        return Ok(());
    }
    let mut proof = store.consistency_proof(prior.size).await?;
    let mut latest = store.checkpoint().await?;
    if latest.size != current.size {
        *current = latest;
        proof = store.consistency_proof(prior.size).await?;
        latest = store.checkpoint().await?;
    }
    if latest.size == current.size {
        let ok = merkle::verify_consistency(
            usize::try_from(prior.size).unwrap_or(0),
            &prior.root,
            usize::try_from(current.size).unwrap_or(0),
            &current.root,
            &proof,
        );
        if !ok {
            findings.push(Finding::NotAppendOnly {
                old_size: prior.size,
                obtained_from: anchor.obtained_from.clone(),
            });
        }
    } else {
        *current = latest;
        not_checked.push(format!(
            "append-only consistency against {}: the log grew throughout the audit, so \
             the proof could not be pinned to one checkpoint — re-run against a quiesced \
             store",
            anchor.obtained_from
        ));
    }
    Ok(())
}

/// Check a plane's history.
///
/// `runs` is what to look at — an auditor sampling, or everything they were
/// given. The log-level checks do not depend on it.
///
/// # Errors
///
/// [`StoreError`] only when the store cannot be read at all. A *finding* is a
/// result, not an error: an audit that stopped at the first problem would report
/// one defect in a store with forty.
pub async fn audit(
    store: &Arc<dyn JournalStore>,
    runs: &[RunId],
    evidence: &Evidence<'_>,
) -> Result<AuditReport, StoreError> {
    let mut current = store.checkpoint().await?;
    let mut findings = Vec::new();
    let mut not_checked = missing_evidence(evidence);
    let mut sound = Vec::new();
    let mut releases = Vec::new();
    let mut warrants = Vec::new();
    let mut unadmitted = Vec::new();
    let mut open_runs = 0usize;

    // ── Per run ────────────────────────────────────────────────────────────
    for &run in runs {
        let records = store.read(run, 1).await?;
        // A run the store returns nothing for is unchecked, never sound.
        // Both backends answer an unknown run with an empty read rather than
        // an error, and every check downstream holds vacuously over nothing:
        // `verify_chain(&[])` passes, the seal claim is absent, and a missing
        // leaf reads as an ordinary open run — so a mistyped or deleted run id
        // audited as "chain and signatures verified" without one record ever
        // being looked at. Not a finding either: an unknown id and a run whose
        // records are genuinely gone cannot be told apart from an empty read,
        // and deletion is the prior-checkpoint check's question, which answers
        // it with evidence rather than a guess.
        if records.is_empty() {
            not_checked.push(format!(
                "run {run}: the store returned no records, so nothing about it was \
                 verified — an empty history holds every check vacuously, which is a \
                 different statement from sound"
            ));
            continue;
        }
        let chain = match evidence.verifier {
            Some(v) => {
                Record::verify_signed(&records, Digest::ZERO, v, evidence.require_signatures)
            }
            None => Record::verify_chain(&records, Digest::ZERO),
        };
        // The head is kept, not merely the verdict: it is the one value that
        // ties the verified bytes to the log's leaf below, and discarding it
        // was what let the two halves of this audit verify two different
        // histories.
        let head = match chain {
            Ok(head) => head,
            Err(e) => {
                findings.push(Finding::Chain {
                    run,
                    detail: e.to_string(),
                });
                continue;
            }
        };

        // Collected as soon as the chain verifies, and **before** any finding
        // this run may carry. The gate is integrity, not innocence: a reader
        // must never be shown a decision drawn from records whose chain did not
        // hold — a forged `RunAdmitted` naming a policy bundle nobody
        // configured is exactly the claim an auditor must not be handed — but a
        // run that verified and then failed a check is the one an investigator
        // most needs the warrant and the releases for. Withholding them with
        // the finding answers *this run left the world in an unknown state*
        // and declines to say what it was authorized to do.
        releases.extend(releases_in(run, &records));
        // Total over the runs that verified: an absent row is indistinguishable
        // from a row nobody looked for. See `Unadmitted`.
        match warrant_in(run, &records) {
            Some(warrant) => warrants.push(warrant),
            None => unadmitted.push(Unadmitted {
                run,
                outcome: concluded_outcome(&records).map(str::to_owned),
            }),
        }

        // Every fault this run has, not the first one. A sealed run may both
        // claim a foreign chain head and be missing from the log, and the two
        // send an investigator to different places.
        let mut faults = Vec::new();
        if !seal_claim_holds(&records) {
            faults.push(Finding::SealClaim { run });
        }
        // Only under a sealing conclusion — an open run's unsettled group
        // is the crash shape a resume repairs, and flagging it would teach
        // the reader this finding is weather. `Finding::GroupUnsettled`
        // carries the argument.
        faults.extend(permanently_undecided(run, &records));

        match placement(store, run, &records, head, &mut current).await? {
            Placement::Sound => {}
            Placement::Open => open_runs += 1,
            Placement::NotInLog => faults.push(Finding::NotInLog { run }),
            Placement::LeafMismatch => faults.push(Finding::LeafMismatch { run }),
            Placement::BadInclusion => faults.push(Finding::BadInclusion { run }),
            Placement::Unpinned => not_checked.push(format!(
                "run {run}: the log grew throughout the audit, so this run's inclusion \
                 could not be pinned to one checkpoint — re-run against a quiesced store"
            )),
        }

        if faults.is_empty() {
            sound.push(run);
        } else {
            findings.extend(faults);
        }
    }

    // Said once rather than per run, and in `not_checked` rather than as a
    // finding: an open run's tail has no leaf to pin it, so truncating it is
    // undetectable until it seals — a limit of what an open run *is*, reported
    // so a clean audit over open runs is not read as more than it proved.
    if open_runs > 0 {
        not_checked.push(format!(
            "{open_runs} open run(s): an open run has no Merkle leaf, so nothing pins its \
             tail — chain and signatures verified, and a truncated tail is undetectable \
             until the run seals"
        ));
    }

    // The audit's scope, stated rather than implied. `runs` is whatever the
    // caller sampled, and the log commits to `current.size` sealed runs — a
    // clean report over three runs of a three-thousand-run log is a true
    // statement about three runs, and nothing in the findings list would ever
    // say so. Stated as unchecked coverage, because that is what it is. What
    // this does NOT cover: it counts the runs *named*, not the runs verified —
    // duplicates in the sample, open runs, and runs the store returned nothing
    // for all inflate the count, so it is an upper bound on coverage and the
    // per-run entries above are the exact record.
    let examined = u64::try_from(runs.len()).unwrap_or(u64::MAX);
    if examined < current.size {
        not_checked.push(format!(
            "scope — this audit examined {examined} named run(s) and the log commits to \
             {} sealed run(s); the remainder was not looked at, and a clean report speaks \
             only for the runs it names",
            current.size
        ));
    }

    // ── Against what the auditor brought ───────────────────────────────────
    //
    // Every anchor, each on its own. They are independent observations, so a
    // store that extends one and not another is forked — and which one it
    // failed is the finding, because that observer is who an investigator asks
    // for the history the store no longer has.
    for anchor in evidence.anchors {
        check_append_only(store, anchor, &mut current, &mut findings, &mut not_checked).await?;
    }
    let unwitnessed = judge_freshness(evidence, &mut findings, &mut not_checked);

    Ok(AuditReport {
        current,
        held_to: evidence.anchors.to_vec(),
        sound,
        findings,
        not_checked,
        releases,
        warrants,
        unadmitted,
        unwitnessed,
    })
}