feat(audit): audit runs and attestations are journaled (P8)

A claim that someone checked the books is evidence the CPA handoff rests on,
so it is a journal fact and replays: audit.run (a programme run through a
day at the books' figure head, with the hashes that make it reproducible and
the two filed workpapers), receipt.attested, entry.reviewed and
period.attested. Behind the journal feature audits, schema 20; no figure
moves. The figure head is the last line that bears on a figure: filing a
workpaper, a run or an attestation never makes an audit stale, an entry,
document, close, settings or role change does, so a period is attested only
on an audit that still describes the books. A receipt reading that differs
from the entry is recorded as disagreeing and never counts; whether a claim
counts is derived against the bar when asked, never stored.

The store also reads what a ledger cannot: the chain, the database against
a replay, stored and extracted files, the access log, duplicate files, and
the backups' standing handed in by the server. Fixture old-books/p8_before
was written by the version before this one and opens to its figures.
period.attested was the tests' placeholder for an unknown event type; they
use ledger.not_yet_written now.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HS8WZziDnyxunC23MUxrLp

crates/ledger-store/src/audit.rs

@@ -0,0 +1,795 @@
+//! The audit's journal and the integrity checks (plan P8; rules.md B-45):
+//! `audit.run`, `receipt.attested`, `entry.reviewed` and `period.attested`,
+//! what "the current head" means, and the findings only the store can see.
+//!
+//! ```text
+//! GOAL: an audit run is a journaled fact keyed by the books' figure
+//!       head, so the same books give the same run once, and an
+//!       attestation rests on a head that has not moved.
+//! SOURCE: IRS Pub. 583 (records); Rev. Proc. 98-25 §5.01 (the audit
+//!         trail); docs/spec/rules.md B-45; DECISION D-9 (the bar), D-53
+//!         (an audit is a run, not a job), IZ-1k (the head).
+//! INVARIANTS:
+//!   1. The figure head is the seq of the last journal line that bears on
+//!      a figure: everything but the audit and attestation events, the
+//!      evidence filings (an attachment, a link, a statement, a
+//!      reconciliation, extraction, annotations) and the format
+//!      declaration. An audit's own filing never moves it.
+//!   2. A run is idempotent on (through, figure head, scope, program
+//!      version): asking again returns the run that exists and journals
+//!      nothing.
+//!   3. A period is attested only on a run at the current figure head
+//!      that covers it (its through date is on or after the period's
+//!      end).
+//!   4. A receipt attestation whose amount or date differs from the
+//!      entry's is recorded, marked `disagrees`, and never counts.
+//!   5. `counted` is never stored: it is read against
+//!      `policy.attestation_bar` as it stands when asked.
+//!   6. The method is the client's own claim; the books keep it and say
+//!      so.
+//!   7. Every write waits for the `audits` journal feature and journals
+//!      nothing when refused.
+//! SCENARIOS:
+//!   1. IF the same audit is asked twice at one head: SHOULD return the
+//!      first run both times.
+//!   2. IF an entry posts after a run: SHOULD refuse attesting a period
+//!      on that run.
+//! ```
+
+use std::collections::BTreeSet;
+
+use chrono::NaiveDate;
+use ledger_core::audit::{Finding, Method, Scope, Severity, Subject};
+use rusqlite::params;
+use rust_decimal::Decimal;
+use serde_json::Value;
+
+use super::{Store, apply_line_to_sqlite};
+use crate::error::StoreError;
+
+/// The audit programme's version: the crate's, and a revision of its
+/// checks. A run at another version is another run.
+pub const PROGRAM_VERSION: &str = concat!("audit-", env!("CARGO_PKG_VERSION"), "-1");
+
+/// Journal lines that do not bear on a figure (INVARIANT 1).
+pub const NOT_FIGURES: &[&str] = &[
+    "audit.run",
+    "receipt.attested",
+    "entry.reviewed",
+    "period.attested",
+    "store.created",
+    "journal.format_set",
+    "journal.tail_repaired",
+    "attachment.linked",
+    "attachment.unlinked",
+    "attachment.withdrawn",
+    "attachment.text_extracted",
+    "attachment.annotated",
+    "receipt.attached",
+    "receipt.linked",
+    "receipt.unlinked",
+    "document.filed",
+    "statement.recorded",
+    "statement.voided",
+    "reconciliation.completed",
+    "reconciliation.voided",
+    "decision.recorded",
+    "open_item.raised",
+    "open_item.resolved",
+    "entry.settles",
+];
+
+/// The integrity checks' codes, severities and predicates (the
+/// statements checks are `ledger_core::audit::CODES`).
+pub const CODES: &[ledger_core::audit::CodeRow] = &[
+    ledger_core::audit::CodeRow {
+        code: "journal_damaged",
+        severity: Severity::Failure,
+        predicate: "the journal's hash chain does not verify",
+    },
+    ledger_core::audit::CodeRow {
+        code: "projection_differs",
+        severity: Severity::Failure,
+        predicate: "the live books or the database differ from a replay of the journal",
+    },
+    ledger_core::audit::CodeRow {
+        code: "blob_missing",
+        severity: Severity::Failure,
+        predicate: "a stored file the books name is not on disk",
+    },
+    ledger_core::audit::CodeRow {
+        code: "blob_damaged",
+        severity: Severity::Failure,
+        predicate: "a stored file's bytes do not match their SHA-256",
+    },
+    ledger_core::audit::CodeRow {
+        code: "access_log_damaged",
+        severity: Severity::Failure,
+        predicate: "the access log's chain does not verify",
+    },
+    ledger_core::audit::CodeRow {
+        code: "extraction_blob_missing",
+        severity: Severity::Warning,
+        predicate: "a file's extracted text is not on disk",
+    },
+    ledger_core::audit::CodeRow {
+        code: "duplicate_blob",
+        severity: Severity::Warning,
+        predicate: "two filed attachments, neither withdrawn, hold the same bytes",
+    },
+    ledger_core::audit::CodeRow {
+        code: "backup_stale",
+        severity: Severity::Warning,
+        predicate: "a backup target has not reached the books' head within a day",
+    },
+    ledger_core::audit::CodeRow {
+        code: "drill_failed",
+        severity: Severity::Failure,
+        predicate: "a backup failed, or failed its restore drill",
+    },
+];
+
+fn found(code: &'static str, subject: Subject, message: String) -> Finding {
+    let severity = CODES
+        .iter()
+        .find(|r| r.code == code)
+        .map_or(Severity::Failure, |r| r.severity);
+    Finding {
+        code,
+        severity,
+        subject,
+        message,
+    }
+}
+
+/// What the backups' standing says, one target at a time, for the audit
+/// (the server reads the standing; the store does not know the backups).
+#[derive(Clone, Debug)]
+pub struct BackupStanding {
+    /// The target's kind and name, as the backups name it.
+    pub target: String,
+    /// It has not reached the head within a day.
+    pub stale: bool,
+    /// Its last run's error, if it failed.
+    pub last_error: Option<String>,
+    /// Its last restore drill's error, if it failed.
+    pub drill_error: Option<String>,
+}
+
+/// The backup findings: a target stale, failing, or failing its drill.
+/// Books with no target set up have none (that is the owner's standing
+/// choice, which `books.start` already says; an audit reports what the
+/// backups it can see do).
+pub fn backup_findings(targets: &[BackupStanding]) -> Vec<Finding> {
+    let mut out = Vec::new();
+    for t in targets {
+        if t.stale {
+            out.push(found(
+                "backup_stale",
+                Subject::Named(t.target.clone()),
+                format!("{} has not reached the books' head within a day", t.target),
+            ));
+        }
+        if let Some(e) = &t.last_error {
+            out.push(found(
+                "drill_failed",
+                Subject::Named(t.target.clone()),
+                format!("{}'s last backup failed: {e}", t.target),
+            ));
+        }
+        if let Some(e) = &t.drill_error {
+            out.push(found(
+                "drill_failed",
+                Subject::Named(t.target.clone()),
+                format!("{}'s restore drill failed: {e}", t.target),
+            ));
+        }
+    }
+    out
+}
+
+/// An audit run as the journal records it (`audit.run`).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AuditRun {
+    /// Its id (1 for the first).
+    pub id: u64,
+    /// The audit covers the books through this date.
+    pub through: NaiveDate,
+    /// The figure head it was made at ([`Store::figure_head`]).
+    pub through_seq: u64,
+    /// What it covers.
+    pub scope: Scope,
+    /// The programme's version ([`PROGRAM_VERSION`]).
+    pub program_version: String,
+    /// SHA-256 of the trial balance through the date.
+    pub tb_hash: String,
+    /// SHA-256 of the findings.
+    pub findings_sha256: String,
+    /// The workpaper's JSON, filed as a document.
+    pub workpaper_attachment: Option<u64>,
+    /// The workpaper's PDF, filed as a document.
+    pub workpaper_pdf: Option<u64>,
+    /// The journal line that recorded it.
+    pub seq: u64,
+}
+
+/// What a run is made from, before it has an id or a seq.
+#[derive(Clone, Debug)]
+pub struct NewAuditRun {
+    /// Through this date.
+    pub through: NaiveDate,
+    /// The figure head the findings were computed at.
+    pub through_seq: u64,
+    /// What it covers.
+    pub scope: Scope,
+    /// SHA-256 of the trial balance.
+    pub tb_hash: String,
+    /// SHA-256 of the findings.
+    pub findings_sha256: String,
+    /// The filed workpaper JSON.
+    pub workpaper_attachment: Option<u64>,
+    /// The filed workpaper PDF.
+    pub workpaper_pdf: Option<u64>,
+}
+
+/// What an attestation says it checked.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum Attested {
+    /// A receipt against an entry.
+    Receipt {
+        /// The filed receipt.
+        attachment_id: u64,
+        /// The entry it supports.
+        entry_id: u64,
+        /// The amount the checker read off it.
+        amount_checked: Decimal,
+        /// The date the checker read off it.
+        date_checked: NaiveDate,
+        /// The checker's reading differs from the entry's.
+        disagrees: bool,
+    },
+    /// An entry, reviewed whole.
+    Entry {
+        /// The entry.
+        entry_id: u64,
+    },
+    /// A period, on an audit run.
+    Period {
+        /// The period.
+        period_id: u64,
+        /// The run it rests on.
+        audit_run_id: u64,
+    },
+}
+
+/// A claim that someone checked, as the journal records it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Attestation {
+    /// The journal line.
+    pub seq: u64,
+    /// Who claims it: the caller's token name.
+    pub by: String,
+    /// When, UTC.
+    pub at: String,
+    /// How, as the client says.
+    pub method: Method,
+    /// A free note.
+    pub note: Option<String>,
+    /// What was checked.
+    pub what: Attested,
+}
+
+impl Attestation {
+    /// Its kind's word: `receipt`, `entry` or `period`.
+    pub fn kind(&self) -> &'static str {
+        match self.what {
+            Attested::Receipt { .. } => "receipt",
+            Attested::Entry { .. } => "entry",
+            Attested::Period { .. } => "period",
+        }
+    }
+}
+
+/// Who claims an attestation, how, and what they add.
+#[derive(Clone, Debug)]
+pub struct Claim {
+    /// The caller's token name.
+    pub by: String,
+    /// How, as the client says.
+    pub method: Method,
+    /// A free note.
+    pub note: Option<String>,
+}
+
+impl Claim {
+    fn into_attestation(self, what: Attested) -> Attestation {
+        Attestation {
+            seq: 0,
+            by: self.by,
+            at: now(),
+            method: self.method,
+            note: self
+                .note
+                .map(|n| n.trim().to_string())
+                .filter(|n| !n.is_empty()),
+            what,
+        }
+    }
+}
+
+fn refused(why: impl Into<String>) -> StoreError {
+    StoreError::AuditRefused(why.into())
+}
+
+fn now() -> String {
+    chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()
+}
+
+impl Store {
+    /// The seq of the last journal line that bears on a figure
+    /// (INVARIANT 1); zero for books with none.
+    pub fn figure_head(&self) -> Result<u64, StoreError> {
+        let quoted: Vec<String> = NOT_FIGURES.iter().map(|t| format!("'{t}'")).collect();
+        let sql = format!(
+            "SELECT COALESCE(MAX(seq), 0) FROM events WHERE type NOT IN ({})",
+            quoted.join(",")
+        );
+        let n: i64 = self.conn.query_row(&sql, [], |r| r.get(0))?;
+        Ok(n as u64)
+    }
+
+    /// The entries with at least one receipt filed on them or linked to
+    /// them, not withdrawn.
+    pub fn entries_with_receipts(&self) -> Result<BTreeSet<u64>, StoreError> {
+        let mut stmt = self.conn.prepare(
+            "SELECT entry_id FROM attachments WHERE withdrawn_at IS NULL AND entry_id IS NOT NULL
+             UNION
+             SELECT l.subject_id FROM attachment_links l JOIN attachments a ON a.id = l.attachment_id
+              WHERE l.subject_kind = 'entry' AND a.withdrawn_at IS NULL",
+        )?;
+        let ids = stmt
+            .query_map([], |r| r.get::<_, i64>(0))?
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(ids.into_iter().map(|i| i as u64).collect())
+    }
+
+    /// What only the store can see: the chain, the database against a
+    /// replay, the stored files, the access log, extraction text and
+    /// duplicate files. Reads; changes nothing.
+    pub fn integrity_findings(&self) -> Vec<Finding> {
+        let mut out = Vec::new();
+        match self.verify() {
+            Ok(()) => {}
+            Err(StoreError::VerifyMismatch(why)) => out.push(found(
+                "projection_differs",
+                Subject::Books,
+                format!("the books differ from a replay of the journal: {why}"),
+            )),
+            Err(e) => out.push(found(
+                "journal_damaged",
+                Subject::Books,
+                format!("the journal does not verify: {e}"),
+            )),
+        }
+        let shas = |sql: &str| -> Vec<String> {
+            self.conn
+                .prepare(sql)
+                .and_then(|mut s| {
+                    s.query_map([], |r| r.get::<_, String>(0))?
+                        .collect::<Result<Vec<_>, _>>()
+                })
+                .unwrap_or_default()
+        };
+        let mut blobs = shas("SELECT sha256 FROM blobs ORDER BY sha256");
+        blobs.dedup();
+        for sha in blobs {
+            if !self.blobs.exists(&sha) {
+                out.push(found(
+                    "blob_missing",
+                    Subject::Named(sha.chars().take(12).collect()),
+                    format!("the stored file {sha} is not on disk"),
+                ));
+            } else if let Err(StoreError::CorruptBlob(_)) = self.blobs.get(&sha) {
+                out.push(found(
+                    "blob_damaged",
+                    Subject::Named(sha.chars().take(12).collect()),
+                    format!("the stored file {sha} does not match its SHA-256"),
+                ));
+            }
+        }
+        if let Err(e) = crate::access::read(&self.data_dir) {
+            out.push(found(
+                "access_log_damaged",
+                Subject::Named("access log".into()),
+                format!("the access log does not verify: {e}"),
+            ));
+        }
+        for sha in shas(
+            "SELECT DISTINCT text_sha256 FROM extractions WHERE text_sha256 IS NOT NULL ORDER BY text_sha256",
+        ) {
+            if !self.blobs.exists(&sha) {
+                out.push(found(
+                    "extraction_blob_missing",
+                    Subject::Named(sha.chars().take(12).collect()),
+                    format!("the extracted text {sha} is not on disk"),
+                ));
+            }
+        }
+        let dups: Vec<(String, String)> = self
+            .conn
+            .prepare(
+                "SELECT blob_sha, GROUP_CONCAT(id) FROM
+                   (SELECT blob_sha, id FROM attachments WHERE withdrawn_at IS NULL ORDER BY id)
+                 GROUP BY blob_sha HAVING COUNT(*) > 1 ORDER BY blob_sha",
+            )
+            .and_then(|mut s| {
+                s.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
+                    .collect::<Result<Vec<_>, _>>()
+            })
+            .unwrap_or_default();
+        for (sha, ids) in dups {
+            out.push(found(
+                "duplicate_blob",
+                Subject::Named(sha.chars().take(12).collect()),
+                format!("attachments {ids} hold the same bytes ({sha})"),
+            ));
+        }
+        out
+    }
+
+    /// Every audit run, oldest first.
+    pub fn audit_runs(&self) -> Result<Vec<AuditRun>, StoreError> {
+        let mut stmt = self.conn.prepare(
+            "SELECT id, through, through_seq, scope, program_version, tb_hash, findings_sha256,
+                    workpaper_attachment, workpaper_pdf, seq FROM audits ORDER BY id",
+        )?;
+        let rows = stmt.query_map([], |r| {
+            Ok((
+                r.get::<_, i64>(0)?,
+                r.get::<_, String>(1)?,
+                r.get::<_, i64>(2)?,
+                r.get::<_, String>(3)?,
+                r.get::<_, String>(4)?,
+                r.get::<_, String>(5)?,
+                r.get::<_, String>(6)?,
+                r.get::<_, Option<i64>>(7)?,
+                r.get::<_, Option<i64>>(8)?,
+                r.get::<_, i64>(9)?,
+            ))
+        })?;
+        let mut out = Vec::new();
+        for row in rows {
+            let (id, through, seq_at, scope, version, tb, fs, wp, pdf, seq) = row?;
+            out.push(AuditRun {
+                id: id as u64,
+                through: NaiveDate::parse_from_str(&through, "%Y-%m-%d")
+                    .map_err(|_| StoreError::JournalCorrupt("audits.through".into()))?,
+                through_seq: seq_at as u64,
+                scope: Scope::parse(&scope)
+                    .ok_or_else(|| StoreError::JournalCorrupt("audits.scope".into()))?,
+                program_version: version,
+                tb_hash: tb,
+                findings_sha256: fs,
+                workpaper_attachment: wp.map(|v| v as u64),
+                workpaper_pdf: pdf.map(|v| v as u64),
+                seq: seq as u64,
+            });
+        }
+        Ok(out)
+    }
+
+    /// One run.
+    pub fn audit_run(&self, id: u64) -> Result<Option<AuditRun>, StoreError> {
+        Ok(self.audit_runs()?.into_iter().find(|r| r.id == id))
+    }
+
+    /// The run that already answers an audit of `through` at `head`
+    /// (INVARIANT 2).
+    pub fn find_audit_run(
+        &self,
+        through: NaiveDate,
+        head: u64,
+        scope: Scope,
+    ) -> Result<Option<AuditRun>, StoreError> {
+        Ok(self.audit_runs()?.into_iter().find(|r| {
+            r.through == through
+                && r.through_seq == head
+                && r.scope == scope
+                && r.program_version == PROGRAM_VERSION
+        }))
+    }
+
+    /// Whether the books' figures are where `run` left them
+    /// (IZ-1k): nothing bearing on a figure has been journaled since.
+    pub fn audit_is_current(&self, run: &AuditRun) -> Result<bool, StoreError> {
+        Ok(run.through_seq == self.figure_head()?)
+    }
+
+    /// Record an audit run (`audit.run`). Waits for the `audits` feature.
+    pub fn record_audit_run(&mut self, new: NewAuditRun) -> Result<AuditRun, StoreError> {
+        self.ensure_feature(crate::format::AUDITS)?;
+        if let Some(existing) = self.find_audit_run(new.through, new.through_seq, new.scope)? {
+            return Ok(existing);
+        }
+        for a in [new.workpaper_attachment, new.workpaper_pdf]
+            .into_iter()
+            .flatten()
+        {
+            self.attachment_state(a)?;
+        }
+        let id = self.audit_runs()?.last().map_or(1, |r| r.id + 1);
+        let mut run = AuditRun {
+            id,
+            through: new.through,
+            through_seq: new.through_seq,
+            scope: new.scope,
+            program_version: PROGRAM_VERSION.into(),
+            tb_hash: new.tb_hash,
+            findings_sha256: new.findings_sha256,
+            workpaper_attachment: new.workpaper_attachment,
+            workpaper_pdf: new.workpaper_pdf,
+            seq: 0,
+        };
+        let line = self.append_journal("audit.run", crate::meta::audit_run(&run))?;
+        run.seq = line.seq;
+        apply_line_to_sqlite(&self.conn, &line)?;
+        Ok(run)
+    }
+
+    /// Every attestation, oldest first.
+    pub fn attestations(&self) -> Result<Vec<Attestation>, StoreError> {
+        let mut stmt = self.conn.prepare(
+            "SELECT seq, kind, attachment_id, entry_id, period_id, audit_run_id, by_who, at_ts,
+                    method, amount_checked, date_checked, disagrees, note
+               FROM attestations ORDER BY seq",
+        )?;
+        let rows = stmt.query_map([], |r| {
+            Ok((
+                r.get::<_, i64>(0)?,
+                r.get::<_, String>(1)?,
+                r.get::<_, Option<i64>>(2)?,
+                r.get::<_, Option<i64>>(3)?,
+                r.get::<_, Option<i64>>(4)?,
+                r.get::<_, Option<i64>>(5)?,
+                r.get::<_, String>(6)?,
+                r.get::<_, String>(7)?,
+                r.get::<_, String>(8)?,
+                r.get::<_, Option<String>>(9)?,
+                r.get::<_, Option<String>>(10)?,
+                r.get::<_, Option<i64>>(11)?,
+                r.get::<_, Option<String>>(12)?,
+            ))
+        })?;
+        let bad = |w: &str| StoreError::JournalCorrupt(format!("attestations.{w}"));
+        let mut out = Vec::new();
+        for row in rows {
+            let (seq, kind, att, entry, period, run, by, at, method, amount, date, dis, note) =
+                row?;
+            let method: Value = serde_json::from_str(&method)?;
+            let what = match kind.as_str() {
+                "receipt" => Attested::Receipt {
+                    attachment_id: att.ok_or_else(|| bad("attachment_id"))? as u64,
+                    entry_id: entry.ok_or_else(|| bad("entry_id"))? as u64,
+                    amount_checked: amount
+                        .and_then(|a| a.parse().ok())
+                        .ok_or_else(|| bad("amount_checked"))?,
+                    date_checked: date
+                        .and_then(|d| NaiveDate::parse_from_str(&d, "%Y-%m-%d").ok())
+                        .ok_or_else(|| bad("date_checked"))?,
+                    disagrees: dis.unwrap_or(0) != 0,
+                },
+                "entry" => Attested::Entry {
+                    entry_id: entry.ok_or_else(|| bad("entry_id"))? as u64,
+                },
+                "period" => Attested::Period {
+                    period_id: period.ok_or_else(|| bad("period_id"))? as u64,
+                    audit_run_id: run.ok_or_else(|| bad("audit_run_id"))? as u64,
+                },
+                _ => return Err(bad("kind")),
+            };
+            out.push(Attestation {
+                seq: seq as u64,
+                by,
+                at,
+                method: crate::meta::method_from_value(&method)?,
+                note,
+                what,
+            });
+        }
+        Ok(out)
+    }
+
+    fn journal_attestation(
+        &mut self,
+        line_type: &str,
+        a: Attestation,
+    ) -> Result<Attestation, StoreError> {
+        let (line_type, payload) = (line_type, crate::meta::attestation(&a));
+        let line = self.append_journal(line_type, payload)?;
+        apply_line_to_sqlite(&self.conn, &line)?;
+        Ok(Attestation { seq: line.seq, ..a })
+    }
+
+    /// Attest that a receipt supports an entry (`receipt.attested`): the
+    /// checker's reading of its amount and date against the entry's. A
+    /// reading that differs is recorded as disagreeing and never counts
+    /// (INVARIANT 4).
+    pub fn attest_receipt(
+        &mut self,
+        attachment_id: u64,
+        entry_id: u64,
+        amount: Decimal,
+        date: NaiveDate,
+        claim: Claim,
+    ) -> Result<Attestation, StoreError> {
+        self.ensure_feature(crate::format::AUDITS)?;
+        let entry = self
+            .ledger
+            .get_entry(entry_id)
+            .ok_or(ledger_core::LedgerError::UnknownEntry(entry_id))?;
+        if entry.status() != ledger_core::EntryStatus::Posted {
+            return Err(refused(format!("entry {entry_id} is not posted")));
+        }
+        let (_, withdrawn) = self.attachment_state(attachment_id)?;
+        if withdrawn {
+            return Err(StoreError::AttachmentWithdrawn(attachment_id));
+        }
+        if !self
+            .attachment_links(attachment_id)?
+            .contains(&crate::store::Subject::Entry(entry_id))
+        {
+            return Err(refused(format!(
+                "attachment {attachment_id} is not filed on or linked to entry {entry_id}"
+            )));
+        }
+        let debits = entry
+            .lines()
+            .iter()
+            .filter(|l| l.side == ledger_core::Side::Debit)
+            .try_fold(Decimal::ZERO, |s, l| s.checked_add(l.amount.amount));
+        let amount_agrees =
+            debits == Some(amount) || entry.lines().iter().any(|l| l.amount.amount == amount);
+        let disagrees = !amount_agrees || date != entry.date();
+        self.journal_attestation(
+            "receipt.attested",
+            claim.into_attestation(Attested::Receipt {
+                attachment_id,
+                entry_id,
+                amount_checked: amount,
+                date_checked: date,
+                disagrees,
+            }),
+        )
+    }
+
+    /// Record that an entry was reviewed whole (`entry.reviewed`).
+    pub fn review_entry(&mut self, entry_id: u64, claim: Claim) -> Result<Attestation, StoreError> {
+        self.ensure_feature(crate::format::AUDITS)?;
+        let entry = self
+            .ledger
+            .get_entry(entry_id)
+            .ok_or(ledger_core::LedgerError::UnknownEntry(entry_id))?;
+        if entry.status() != ledger_core::EntryStatus::Posted {
+            return Err(refused(format!("entry {entry_id} is not posted")));
+        }
+        self.journal_attestation(
+            "entry.reviewed",
+            claim.into_attestation(Attested::Entry { entry_id }),
+        )
+    }
+
+    /// The run a period can be attested on: the latest at the current
+    /// figure head whose through date is on or after the period's end
+    /// (INVARIANT 3).
+    pub fn run_for_period(&self, period_id: u64) -> Result<AuditRun, StoreError> {
+        let period = self
+            .ledger
+            .get_period(period_id)
+            .ok_or_else(|| refused(format!("no period {period_id}")))?;
+        let head = self.figure_head()?;
+        let runs = self.audit_runs()?;
+        runs.into_iter()
+            .rev()
+            .find(|r| r.through_seq == head && r.through >= period.end)
+            .ok_or_else(|| {
+                refused(format!(
+                    "no audit through {} or later stands at the books' current head: run books.audit first (an entry, a document, a close or a settings change since the last one moves the head)",
+                    period.end
+                ))
+            })
+    }
+
+    /// Attest a period on an audit run at the current head
+    /// (`period.attested`).
+    pub fn attest_period(
+        &mut self,
+        period_id: u64,
+        claim: Claim,
+    ) -> Result<(Attestation, AuditRun), StoreError> {
+        self.ensure_feature(crate::format::AUDITS)?;
+        let run = self.run_for_period(period_id)?;
+        let a = self.journal_attestation(
+            "period.attested",
+            claim.into_attestation(Attested::Period {
+                period_id,
+                audit_run_id: run.id,
+            }),
+        )?;
+        Ok((a, run))
+    }
+}
+
+/// Read an `audits` row for a replayed line.
+pub(crate) fn insert_run(conn: &rusqlite::Connection, run: &AuditRun) -> Result<(), StoreError> {
+    conn.execute(
+        "INSERT INTO audits(id, through, through_seq, scope, program_version, tb_hash, findings_sha256,
+                            workpaper_attachment, workpaper_pdf, seq)
+         VALUES (?,?,?,?,?,?,?,?,?,?)",
+        params![
+            run.id as i64,
+            run.through.to_string(),
+            run.through_seq as i64,
+            run.scope.key(),
+            run.program_version,
+            run.tb_hash,
+            run.findings_sha256,
+            run.workpaper_attachment.map(|v| v as i64),
+            run.workpaper_pdf.map(|v| v as i64),
+            run.seq as i64,
+        ],
+    )?;
+    Ok(())
+}
+
+/// Write an `attestations` row for a replayed line.
+pub(crate) fn insert_attestation(
+    conn: &rusqlite::Connection,
+    a: &Attestation,
+) -> Result<(), StoreError> {
+    let (mut att, mut entry, mut period, mut run) = (None, None, None, None);
+    let (mut amount, mut date, mut dis) = (None, None, None);
+    match &a.what {
+        Attested::Receipt {
+            attachment_id,
+            entry_id,
+            amount_checked,
+            date_checked,
+            disagrees,
+        } => {
+            att = Some(*attachment_id as i64);
+            entry = Some(*entry_id as i64);
+            amount = Some(amount_checked.to_string());
+            date = Some(date_checked.to_string());
+            dis = Some(i64::from(*disagrees));
+        }
+        Attested::Entry { entry_id } => entry = Some(*entry_id as i64),
+        Attested::Period {
+            period_id,
+            audit_run_id,
+        } => {
+            period = Some(*period_id as i64);
+            run = Some(*audit_run_id as i64);
+        }
+    }
+    conn.execute(
+        "INSERT INTO attestations(seq, kind, attachment_id, entry_id, period_id, audit_run_id, by_who, at_ts,
+                                  method, amount_checked, date_checked, disagrees, note)
+         VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        params![
+            a.seq as i64,
+            a.kind(),
+            att,
+            entry,
+            period,
+            run,
+            a.by,
+            a.at,
+            crate::meta::method_json(&a.method).to_string(),
+            amount,
+            date,
+            dis,
+            a.note,
+        ],
+    )?;
+    Ok(())
+}

crates/ledger-store/src/error.rs

@@ -162,6 +162,12 @@ pub enum StoreError {
     )]
     WriteOffNeedsEvidence,
 
+    /// An audit or an attestation the books refuse (plan P8; rules.md
+    /// B-45): a stale head, an entry not posted, a receipt not on the
+    /// entry. The words say what to do.
+    #[error("{0}")]
+    AuditRefused(String),
+
     /// The requested base currency differs from the store's.
     #[error("base currency mismatch: store is {0}, requested {1}")]
     CurrencyMismatch(String, String),

crates/ledger-store/src/format.rs

@@ -66,6 +66,7 @@ pub const KNOWN_FEATURES: &[&str] = &[
     ATTACHMENT_TEXT,
     DRAFTS_V2,
     SOURCE_REVISIONS,
+    AUDITS,
 ];
 
 /// Feature: `period.closed` closes to members with `"rounding":
@@ -105,6 +106,9 @@ pub(crate) fn setting_feature(key: &str) -> Option<&'static str> {
         | "bills.discount_warn_days"
         | "company.email"
         | "company.website" => Some(DOCUMENTS_V2),
+        // P8: what counts as an attestation, and whether the CPA package
+        // waits for an audit.
+        "policy.attestation_bar" | "policy.cpa_package_requires_audit" => Some(AUDITS),
         _ => None,
     }
 }
@@ -228,6 +232,17 @@ pub const DRAFTS_V2: &str = "drafts_v2";
 /// declaration instead (rules.md B-27).
 pub const SOURCE_REVISIONS: &str = "source_revisions";
 
+/// Feature: `audits` (plan P8): `audit.run` (a programme of checks run
+/// through a date, with the hash of its findings and its workpaper),
+/// `receipt.attested`, `entry.reviewed` and `period.attested` (claims by
+/// an authenticated client that it checked), and the two settings that
+/// say what counts (`policy.attestation_bar`) and whether the CPA package
+/// waits for an audit (`policy.cpa_package_requires_audit`). An older
+/// version doesn't know those event types or settings; it stops at the
+/// declaration instead of dropping the claims a handoff rests on
+/// (rules.md B-27, B-45).
+pub const AUDITS: &str = "audits";
+
 /// The line type that declares the journal format.
 pub const FORMAT_SET: &str = "journal.format_set";
 
@@ -253,6 +268,7 @@ pub const EVENT_TYPES: &[&str] = &[
     "attachment.text_extracted",
     "attachment.unlinked",
     "attachment.withdrawn",
+    "audit.run",
     "bill.created",
     "bill.paid",
     "class.added",
@@ -275,6 +291,7 @@ pub const EVENT_TYPES: &[&str] = &[
     "entry.party_set",
     "entry.posted",
     "entry.reversed",
+    "entry.reviewed",
     "entry.settles",
     "entry.vendor_set",
     "fx.rate_set",
@@ -298,10 +315,12 @@ pub const EVENT_TYPES: &[&str] = &[
     "party.w9_set",
     "payroll.run",
     "payroll.taxes_remitted",
+    "period.attested",
     "period.closed",
     "period.created",
     "project.set",
     "receipt.attached",
+    "receipt.attested",
     "receipt.linked",
     "receipt.unlinked",
     "reconciliation.completed",
@@ -1035,6 +1054,39 @@ mod tests {
         assert_eq!(up.adds(), vec!["the journal feature source_revisions"]);
     }
 
+    /// The version before P8 stops at a declaration of `audits` as newer,
+    /// never dropping an attestation a handoff rests on; books that
+    /// declared every earlier feature still wait for it (B-27, D-39).
+    #[test]
+    fn audits_is_its_own_feature() {
+        assert!(KNOWN_FEATURES.contains(&AUDITS));
+        let before: Vec<&str> = KNOWN_FEATURES
+            .iter()
+            .copied()
+            .filter(|f| *f != AUDITS)
+            .collect();
+        let l = line(
+            FORMAT_SET,
+            json!({"format": 2, "features": [DRAFTS_V2, AUDITS]}),
+        );
+        match declared_knowing(&l, &before) {
+            Err(StoreError::NewerJournal { need, .. }) => {
+                assert_eq!(need, "journal features audits")
+            }
+            other => panic!("expected NewerJournal, got {other:?}"),
+        }
+        let older = JournalFormat {
+            format: 2,
+            features: before.iter().map(|f| f.to_string()).collect(),
+        };
+        assert!(matches!(
+            older.require_feature(AUDITS),
+            Err(StoreError::FormatUpgradePending { .. })
+        ));
+        let up = older.upgrade().expect("waits for it");
+        assert_eq!(up.adds(), vec!["the journal feature audits"]);
+    }
+
     /// This version's format: the newest it reads, every feature it
     /// knows, each once, sorted (what new books and an upgrade declare).
     #[test]

crates/ledger-store/src/lib.rs

@@ -47,6 +47,7 @@ pub use error::StoreError;
 pub use format::{FormatUpgrade, JournalFormat, SUPPORTED_JOURNAL_FORMAT};
 pub use journal::{GENESIS_PREV, JournalLine, JournalWriter};
 pub use snapshot::{DiskCheck, Snapshot};
+pub use store::audit;
 pub use store::{
     Attachment, BillPreview, DepreciationRun, DisposalPreview, EntryRow, EventRow, Extraction,
     ExtractionStatus, InvoicePreview, JOURNAL_FILE, LEGACY_DB_FILE, LOCK_FILE, MAX_RECURRING_RUN,

crates/ledger-store/src/meta.rs

@@ -729,6 +729,198 @@ pub fn draft_discarded_of(p: &Value) -> Result<(u64, String), StoreError> {
     Ok((u64_of(p, E, "id")?, str_of(p, E, "reason")?))
 }
 
+// ---------------------------------------------------------------------------
+// Audits and attestations (plan P8, feature `audits`)
+// ---------------------------------------------------------------------------
+
+/// A method as journaled: only the parts the client gave.
+pub fn method_json(m: &ledger_core::audit::Method) -> Value {
+    let mut o = Map::new();
+    put(&mut o, "model", m.model.clone());
+    put(
+        &mut o,
+        "reasoning_effort",
+        m.reasoning_effort.map(|e| e.key()),
+    );
+    put(&mut o, "adversarial", m.adversarial);
+    put(&mut o, "verifier_agents", m.verifier_agents);
+    Value::Object(o)
+}
+
+/// Read a method. A key or an effort this version doesn't know is newer
+/// when a line is at hand (B-27), else corrupt.
+fn method_of(
+    line: Option<&crate::journal::JournalLine>,
+    m: &Value,
+    event: &str,
+) -> Result<ledger_core::audit::Method, StoreError> {
+    let o = m.as_object().ok_or_else(|| corrupt(event, "method"))?;
+    let unknown = |what: String| match line {
+        Some(l) => crate::format::newer_content(l, what),
+        None => corrupt(event, "method"),
+    };
+    let mut out = ledger_core::audit::Method::default();
+    for (k, v) in o {
+        match k.as_str() {
+            "model" => {
+                out.model = Some(
+                    v.as_str()
+                        .ok_or_else(|| corrupt(event, "method.model"))?
+                        .to_string(),
+                )
+            }
+            "reasoning_effort" => {
+                let s = v
+                    .as_str()
+                    .ok_or_else(|| corrupt(event, "method.reasoning_effort"))?;
+                out.reasoning_effort = Some(
+                    ledger_core::audit::Effort::parse(s)
+                        .ok_or_else(|| unknown(format!("the reasoning effort {s:?}")))?,
+                );
+            }
+            "adversarial" => {
+                out.adversarial = Some(
+                    v.as_bool()
+                        .ok_or_else(|| corrupt(event, "method.adversarial"))?,
+                )
+            }
+            "verifier_agents" => {
+                out.verifier_agents = Some(
+                    v.as_u64()
+                        .and_then(|n| u32::try_from(n).ok())
+                        .ok_or_else(|| corrupt(event, "method.verifier_agents"))?,
+                )
+            }
+            other => return Err(unknown(format!("the method key {other:?}"))),
+        }
+    }
+    Ok(out)
+}
+
+/// A method read back from the projection.
+pub fn method_from_value(m: &Value) -> Result<ledger_core::audit::Method, StoreError> {
+    method_of(None, m, "attestations")
+}
+
+/// `audit.run {id, through, through_seq, scope, program_version, tb_hash,
+/// findings_sha256, workpaper_attachment?, workpaper_pdf?}`.
+pub fn audit_run(r: &crate::audit::AuditRun) -> Value {
+    let mut o = Map::new();
+    o.insert("id".into(), json!(r.id));
+    o.insert("through".into(), json!(r.through.to_string()));
+    o.insert("through_seq".into(), json!(r.through_seq));
+    o.insert("scope".into(), json!(r.scope.key()));
+    o.insert("program_version".into(), json!(r.program_version));
+    o.insert("tb_hash".into(), json!(r.tb_hash));
+    o.insert("findings_sha256".into(), json!(r.findings_sha256));
+    put(&mut o, "workpaper_attachment", r.workpaper_attachment);
+    put(&mut o, "workpaper_pdf", r.workpaper_pdf);
+    Value::Object(o)
+}
+
+/// Read an `audit.run`. A scope this version doesn't know is newer.
+pub fn audit_run_of(
+    line: &crate::journal::JournalLine,
+) -> Result<crate::audit::AuditRun, StoreError> {
+    const E: &str = "audit.run";
+    let p = &line.payload;
+    let scope_key = str_of(p, E, "scope")?;
+    let scope = ledger_core::audit::Scope::parse(&scope_key).ok_or_else(|| {
+        crate::format::newer_content(line, format!("the audit scope {scope_key:?}"))
+    })?;
+    Ok(crate::audit::AuditRun {
+        id: u64_of(p, E, "id")?,
+        through: date_of(&str_of(p, E, "through")?, E, "through")?,
+        through_seq: u64_of(p, E, "through_seq")?,
+        scope,
+        program_version: str_of(p, E, "program_version")?,
+        tb_hash: str_of(p, E, "tb_hash")?,
+        findings_sha256: str_of(p, E, "findings_sha256")?,
+        workpaper_attachment: opt_u64(p, E, "workpaper_attachment")?,
+        workpaper_pdf: opt_u64(p, E, "workpaper_pdf")?,
+        seq: line.seq,
+    })
+}
+
+/// The payload of `receipt.attested`, `entry.reviewed` or `period.attested`.
+pub fn attestation(a: &crate::audit::Attestation) -> Value {
+    use crate::audit::Attested;
+    let mut o = Map::new();
+    match &a.what {
+        Attested::Receipt {
+            attachment_id,
+            entry_id,
+            amount_checked,
+            date_checked,
+            disagrees,
+        } => {
+            o.insert("attachment_id".into(), json!(attachment_id));
+            o.insert("entry_id".into(), json!(entry_id));
+            o.insert("amount_checked".into(), json!(amount_checked.to_string()));
+            o.insert("date_checked".into(), json!(date_checked.to_string()));
+            o.insert("disagrees".into(), json!(disagrees));
+        }
+        Attested::Entry { entry_id } => {
+            o.insert("entry_id".into(), json!(entry_id));
+        }
+        Attested::Period {
+            period_id,
+            audit_run_id,
+        } => {
+            o.insert("period_id".into(), json!(period_id));
+            o.insert("audit_run_id".into(), json!(audit_run_id));
+        }
+    }
+    o.insert("by".into(), json!(a.by));
+    o.insert("at".into(), json!(a.at));
+    o.insert("method".into(), method_json(&a.method));
+    put(&mut o, "note", a.note.clone());
+    Value::Object(o)
+}
+
+/// Read one of the three attestation events by its line.
+pub fn attestation_of(
+    line: &crate::journal::JournalLine,
+) -> Result<crate::audit::Attestation, StoreError> {
+    use crate::audit::Attested;
+    let p = &line.payload;
+    let e = line.line_type.as_str();
+    let what = match e {
+        "receipt.attested" => Attested::Receipt {
+            attachment_id: u64_of(p, e, "attachment_id")?,
+            entry_id: u64_of(p, e, "entry_id")?,
+            amount_checked: str_of(p, e, "amount_checked")?
+                .parse()
+                .map_err(|_| corrupt(e, "amount_checked"))?,
+            date_checked: date_of(&str_of(p, e, "date_checked")?, e, "date_checked")?,
+            disagrees: p
+                .get("disagrees")
+                .and_then(Value::as_bool)
+                .ok_or_else(|| corrupt(e, "disagrees"))?,
+        },
+        "entry.reviewed" => Attested::Entry {
+            entry_id: u64_of(p, e, "entry_id")?,
+        },
+        "period.attested" => Attested::Period {
+            period_id: u64_of(p, e, "period_id")?,
+            audit_run_id: u64_of(p, e, "audit_run_id")?,
+        },
+        _ => return Err(crate::format::unknown_type(line)),
+    };
+    Ok(crate::audit::Attestation {
+        seq: line.seq,
+        by: str_of(p, e, "by")?,
+        at: str_of(p, e, "at")?,
+        method: method_of(
+            Some(line),
+            p.get("method").ok_or_else(|| corrupt(e, "method"))?,
+            e,
+        )?,
+        note: opt_str(p, e, "note")?,
+        what,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ledger-store/src/schema.rs

@@ -5,7 +5,7 @@
 //! is the current version.
 
 /// Current schema version.
-pub const SCHEMA_VERSION: u32 = 19;
+pub const SCHEMA_VERSION: u32 = 20;
 
 /// Migrations, in order. `MIGRATIONS[i]` migrates from version `i` to `i+1`.
 pub const MIGRATIONS: &[&str] = &[
@@ -485,5 +485,41 @@ CREATE TABLE IF NOT EXISTS draft_states(
   state TEXT NOT NULL,
   note TEXT NULL
 );
+"#,
+    r#"
+-- v20: audits and attestations (plan P8, feature `audits`). An audit run
+-- is a row per `audit.run`: the date it covers, the figure head it was
+-- made at, the hashes that make it reproducible and the filed workpaper.
+-- An attestation is a row per `receipt.attested`, `entry.reviewed` or
+-- `period.attested`; whether one counts is derived against
+-- `policy.attestation_bar` when asked and never stored. The method is the
+-- client's own claim, kept as canonical JSON. Amounts stay TEXT.
+CREATE TABLE IF NOT EXISTS audits(
+  id INTEGER PRIMARY KEY,
+  through TEXT NOT NULL,
+  through_seq INTEGER NOT NULL,
+  scope TEXT NOT NULL,
+  program_version TEXT NOT NULL,
+  tb_hash TEXT NOT NULL,
+  findings_sha256 TEXT NOT NULL,
+  workpaper_attachment INTEGER NULL,
+  workpaper_pdf INTEGER NULL,
+  seq INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS attestations(
+  seq INTEGER PRIMARY KEY,
+  kind TEXT NOT NULL,
+  attachment_id INTEGER NULL,
+  entry_id INTEGER NULL,
+  period_id INTEGER NULL,
+  audit_run_id INTEGER NULL,
+  by_who TEXT NOT NULL,
+  at_ts TEXT NOT NULL,
+  method TEXT NOT NULL,
+  amount_checked TEXT NULL,
+  date_checked TEXT NULL,
+  disagrees INTEGER NULL,
+  note TEXT NULL
+);
 "#,
 ];

crates/ledger-store/src/store.rs

@@ -51,6 +51,9 @@ use rust_decimal::Decimal;
 use serde_json::{Value, json};
 use std::path::{Path, PathBuf};
 
+#[path = "audit.rs"]
+pub mod audit;
+
 /// File names inside the data directory.
 pub const JOURNAL_FILE: &str = "journal.jsonl";
 
@@ -7594,6 +7597,21 @@ fn replay_line(
             let a = crate::meta::annotated_of(line)?;
             ledger.replay_annotate_attachment(a)?;
         }
+        "audit.run" => {
+            // Not accounting state; the projection records it. Checked
+            // for shape here (an unknown scope is newer, rules.md B-27).
+            crate::meta::audit_run_of(line)?;
+        }
+        // A claim that someone checked; no figure moves.
+        "receipt.attested" => {
+            crate::meta::attestation_of(line)?;
+        }
+        "entry.reviewed" => {
+            crate::meta::attestation_of(line)?;
+        }
+        "period.attested" => {
+            crate::meta::attestation_of(line)?;
+        }
         "entry.settles" => {
             let (entry, settles) = crate::meta::entry_settles_of(p)?;
             ledger.replay_set_entry_settles(entry, settles)?;
@@ -8809,6 +8827,22 @@ fn apply_line_rows(conn: &Connection, line: &JournalLine) -> Result<(), StoreErr
                 ],
             )?;
         }
+        "audit.run" => {
+            let run = crate::meta::audit_run_of(line)?;
+            audit::insert_run(conn, &run)?;
+        }
+        "receipt.attested" => {
+            let a = crate::meta::attestation_of(line)?;
+            audit::insert_attestation(conn, &a)?;
+        }
+        "entry.reviewed" => {
+            let a = crate::meta::attestation_of(line)?;
+            audit::insert_attestation(conn, &a)?;
+        }
+        "period.attested" => {
+            let a = crate::meta::attestation_of(line)?;
+            audit::insert_attestation(conn, &a)?;
+        }
         "entry.settles" => {
             // Rebuilt into the ledger by replay; no report rows.
         }