doubleentry 0.7.0

//! The executable contract every backend must satisfy.
//!
//! A ledger is only as trustworthy as its weakest backend, and the ways a
//! backend can be subtly wrong — a read-then-write idempotency check that races,
//! a batch that half-lands, an index sequence with a gap — are exactly the ways
//! that produce no error and no symptom until an audit. Publishing the trait
//! without a way to check it would leave each implementor to guess.
//!
//! Run [`check_all`] against a fresh, empty store:
//!
//! ```
//! use doubleentry::LedgerId;
//! use doubleentry::storage::{MemoryStore, conformance};
//!
//! let store = MemoryStore::<2>::new(LedgerId::new("my-ledger")?);
//! let report = conformance::block_on(conformance::check_all(&store));
//! report.assert_passed();
//! # Ok::<(), Box<dyn std::error::Error>>(())
//! ```
//!
//! Every check builds its own accounts and entries, so a backend needs to
//! provide nothing but an empty store.

use std::future::Future;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Poll, Wake, Waker};

use time::macros::date;

use crate::account::{
    Account, AccountKind, AccountPath, AccountRecord, AccountRegistry, BalanceLimit,
};
use crate::balance::{BalanceKey, BalanceQuery};
use crate::clearing::{Clearing, PostingRef};
use crate::dimensions::{DimensionFilter, Dimensions, Label};
use crate::entry::{Draft, Entry, EntryId, IdempotencyKey, LedgerPolicy, SealContext};
use crate::merkle::MerkleLog;
use crate::money::{Amount, Currency};
use crate::period::PeriodCalendar;
use crate::posting::Layer;
use crate::seal::SealedBalanceOutcome;
use crate::storage::{Cursor, EntryBatch, LedgerStore, MAX_PAGE_SIZE, PostingCursor};
use crate::{AccountId, Balanced};

/// A ledger identifier for tests and examples.
#[must_use]
pub fn test_ledger() -> crate::period::LedgerId {
    crate::period::LedgerId::new("test").unwrap_or_else(|_| unreachable!("literal is valid"))
}

/// How many sequencing attempts a check makes before giving up.
///
/// Generous, because a pass that places nothing is not evidence that there is
/// nothing to place — see [`drain_sequencing`].
const SEQUENCING_ATTEMPTS: usize = 512;

/// Yields to the executor once, without depending on one.
///
/// The conformance suite has to let other tasks run — a backend whose watermark
/// is held back by an unrelated open transaction can only make progress once
/// that transaction ends — but it must not pick an async runtime on behalf of
/// the crate.
async fn yield_once() {
    struct YieldOnce(bool);
    impl Future for YieldOnce {
        type Output = ();
        fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
            if self.0 {
                Poll::Ready(())
            } else {
                self.0 = true;
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        }
    }
    YieldOnce(false).await;
}

/// Drives sequencing until every recorded entry has a position.
///
/// A single pass is not enough, and neither is "loop until a pass places
/// nothing". A backend advancing on a commit-order watermark declines to place
/// rows whose inserting transaction is still open, so a pass can legitimately
/// place zero and still have work outstanding. Sequencing is *eventually*
/// complete; a check that assumed otherwise would be testing a guarantee the
/// technique does not offer.
async fn drain_sequencing<const P: u8, S: LedgerStore<P>>(store: &S) -> Result<(), String> {
    let mut idle = 0u32;
    for _ in 0..SEQUENCING_ATTEMPTS {
        match store.sequence().await {
            Ok(0) => {
                idle = idle.saturating_add(1);
                // Several consecutive empty passes with nothing arriving is as
                // good a settling signal as this interface can give.
                if idle >= 8 {
                    return Ok(());
                }
            }
            Ok(_) => idle = 0,
            Err(e) => return Err(format!("sequence failed: {e}")),
        }
        yield_once().await;
    }
    Err("sequencing did not settle".to_owned())
}

/// Sequences until `id` has a position, or gives up.
async fn sequence_until_placed<const P: u8, S: LedgerStore<P>>(
    store: &S,
    id: EntryId,
) -> Result<Option<crate::journal::LogIndex>, String> {
    for _ in 0..SEQUENCING_ATTEMPTS {
        match store.get(id).await {
            Ok(Some(record)) if record.index.is_some() => return Ok(record.index),
            Ok(_) => {}
            Err(e) => return Err(format!("get failed: {e}")),
        }
        store
            .sequence()
            .await
            .map_err(|e| format!("sequence failed: {e}"))?;
        yield_once().await;
    }
    Ok(None)
}

/// Runs a future to completion on the current thread.
///
/// Provided so a backend can run the suite from an ordinary `#[test]` without
/// committing this crate — or the backend's test suite — to a particular async
/// runtime.
pub fn block_on<F: Future>(future: F) -> F::Output {
    struct Signal {
        woken: Mutex<bool>,
        ready: Condvar,
    }

    impl Wake for Signal {
        fn wake(self: Arc<Self>) {
            self.wake_by_ref();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            let mut woken = self.woken.lock().unwrap_or_else(|e| e.into_inner());
            *woken = true;
            self.ready.notify_one();
        }
    }

    let signal = Arc::new(Signal {
        woken: Mutex::new(false),
        ready: Condvar::new(),
    });
    let waker = Waker::from(Arc::clone(&signal));
    let mut context = Context::from_waker(&waker);
    let mut future = std::pin::pin!(future);

    loop {
        if let Poll::Ready(value) = future.as_mut().poll(&mut context) {
            return value;
        }
        let mut woken = signal.woken.lock().unwrap_or_else(|e| e.into_inner());
        while !*woken {
            woken = signal.ready.wait(woken).unwrap_or_else(|e| e.into_inner());
        }
        *woken = false;
    }
}

/// The outcome of one check.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CheckResult {
    /// What was checked.
    pub name: &'static str,
    /// `None` when it passed; the reason when it did not.
    pub failure: Option<String>,
}

impl CheckResult {
    fn pass(name: &'static str) -> Self {
        Self {
            name,
            failure: None,
        }
    }

    fn fail(name: &'static str, reason: impl Into<String>) -> Self {
        Self {
            name,
            failure: Some(reason.into()),
        }
    }

    /// True when the check passed.
    #[must_use]
    pub fn passed(&self) -> bool {
        self.failure.is_none()
    }
}

/// The result of a full conformance run.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Report {
    /// One entry per check, in the order they ran.
    pub checks: Vec<CheckResult>,
}

impl Report {
    /// True when every check passed.
    #[must_use]
    pub fn passed(&self) -> bool {
        self.checks.iter().all(CheckResult::passed)
    }

    /// The checks that failed.
    #[must_use]
    pub fn failures(&self) -> Vec<&CheckResult> {
        self.checks.iter().filter(|c| !c.passed()).collect()
    }

    /// Panics with a readable summary unless every check passed.
    ///
    /// # Panics
    ///
    /// Panics when any check failed.
    pub fn assert_passed(&self) {
        assert!(self.passed(), "{self}");
    }
}

impl std::fmt::Display for Report {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let failures = self.failures();
        if failures.is_empty() {
            return write!(f, "all {} conformance checks passed", self.checks.len());
        }
        writeln!(
            f,
            "{} of {} conformance checks failed:",
            failures.len(),
            self.checks.len()
        )?;
        for check in failures {
            writeln!(
                f,
                "  - {}: {}",
                check.name,
                check.failure.as_deref().unwrap_or("")
            )?;
        }
        Ok(())
    }
}

/// Accounts and policy the checks post against.
struct Fixture {
    accounts: AccountRegistry,
    calendar: PeriodCalendar,
    policy: LedgerPolicy,
    left: AccountId,
    right: AccountId,
}

impl Fixture {
    fn new() -> Self {
        let mut accounts = AccountRegistry::new();
        let left = accounts
            .register_path("Conformance:Left", date!(2000 - 01 - 01))
            .unwrap_or(AccountId::from_index(0));
        let right = accounts
            .register_path("Conformance:Right", date!(2000 - 01 - 01))
            .unwrap_or(AccountId::from_index(1));
        Self {
            accounts,
            calendar: PeriodCalendar::new(),
            policy: LedgerPolicy::default(),
            left,
            right,
        }
    }

    fn ctx(&self) -> SealContext<'_> {
        SealContext {
            accounts: &self.accounts,
            calendar: &self.calendar,
            policy: &self.policy,
        }
    }

    /// A balanced entry moving `minor` from right to left.
    fn entry<const P: u8>(&self, key: &[u8], minor: i64) -> Option<Entry<Balanced, P>> {
        self.entry_with_id(EntryId::generate(), key, minor)
    }

    /// A balanced entry moving `minor` from left to right — the settling side.
    fn reversed_entry<const P: u8>(&self, key: &[u8], minor: i64) -> Option<Entry<Balanced, P>> {
        Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 20),
        )
        .credit(self.left, Amount::<P>::from_minor(minor), Currency::EUR)
        .debit(self.right, Amount::<P>::from_minor(minor), Currency::EUR)
        .seal(&self.ctx())
        .ok()
    }

    /// A genuine reversal of `original`.
    fn reverse<const P: u8>(
        &self,
        original: &Entry<Balanced, P>,
        key: &[u8],
        on: time::Date,
    ) -> Option<Entry<Balanced, P>> {
        original
            .reverse(
                EntryId::generate(),
                IdempotencyKey::new(key.to_vec()).ok()?,
                on,
            )
            .seal(&self.ctx())
            .ok()
    }

    /// An entry that names `original` as reversed but posts something else.
    fn forged_reversal<const P: u8>(
        &self,
        original: &Entry<Balanced, P>,
        key: &[u8],
    ) -> Option<Entry<Balanced, P>> {
        Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 04 - 05),
        )
        .reversing(original.id(), original.booking_date())
        .debit(self.right, Amount::<P>::from_minor(1), Currency::EUR)
        .credit(self.left, Amount::<P>::from_minor(1), Currency::EUR)
        .seal(&self.ctx())
        .ok()
    }

    /// The settled EUR balance on the left account.
    fn key(&self) -> BalanceKey {
        BalanceKey {
            account: self.left,
            currency: Currency::EUR,
            layer: Layer::Settled,
        }
    }

    /// A clearing over the given postings.
    fn clearing<const P: u8>(&self, items: &[(PostingRef, i64)]) -> Clearing<P> {
        items.iter().fold(
            Clearing::new(
                crate::clearing::ClearingId::generate(),
                self.key(),
                date!(2026 - 04 - 20),
            ),
            |clearing, (posting, applied)| {
                clearing.apply(*posting, Amount::<P>::from_minor(*applied))
            },
        )
    }

    fn entry_with_id<const P: u8>(
        &self,
        id: EntryId,
        key: &[u8],
        minor: i64,
    ) -> Option<Entry<Balanced, P>> {
        Entry::<Draft, P>::new(
            id,
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 15),
        )
        .debit(self.left, Amount::<P>::from_minor(minor), Currency::EUR)
        .credit(self.right, Amount::<P>::from_minor(minor), Currency::EUR)
        .seal(&self.ctx())
        .ok()
    }

    /// An entry whose postings carry dimensions.
    fn dimensioned_entry<const P: u8>(&self, key: &[u8], minor: i64) -> Option<Entry<Balanced, P>> {
        let dims = crate::dimensions::Dimensions::none()
            .with(
                crate::Label::new("activity").ok()?,
                crate::Label::new("Network").ok()?,
            )
            .ok()?
            .with(
                crate::Label::new("segment").ok()?,
                crate::Label::new("Electricity").ok()?,
            )
            .ok()?;
        Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 15),
        )
        .post(
            crate::Posting::debit(self.left, Amount::<P>::from_minor(minor), Currency::EUR)
                .with_dimensions(dims.clone()),
        )
        .post(
            crate::Posting::credit(self.right, Amount::<P>::from_minor(minor), Currency::EUR)
                .with_dimensions(dims),
        )
        .seal(&self.ctx())
        .ok()
    }

    /// An entry carrying a caller-defined `kind` label.
    fn entry_with_kind<const P: u8>(
        &self,
        key: &[u8],
        minor: i64,
        kind: &str,
    ) -> Option<Entry<Balanced, P>> {
        Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 15),
        )
        .debit(self.left, Amount::<P>::from_minor(minor), Currency::EUR)
        .credit(self.right, Amount::<P>::from_minor(minor), Currency::EUR)
        .with_kind(crate::dimensions::Label::new(kind).ok()?)
        .seal(&self.ctx())
        .ok()
    }
}

/// Runs every conformance check against an **empty** store.
///
/// Checks are independent but share the store, so a backend that fails an early
/// check may cascade; read the first failure first.
pub async fn check_all<const P: u8, S: LedgerStore<P>>(store: &S) -> Report {
    let mut checks = Vec::new();
    checks.push(check_starts_empty(store).await);
    checks.push(check_append_assigns_dense_indices(store).await);
    checks.push(check_reads_are_stable(store).await);
    checks.push(check_idempotent_replay(store).await);
    checks.push(check_lookup_by_idempotency_key(store).await);
    checks.push(check_idempotency_conflict(store).await);
    checks.push(check_batch_is_atomic(store).await);
    checks.push(check_pagination_covers_the_log(store).await);
    checks.push(check_balances_match_the_log(store).await);
    checks.push(check_proofs_verify(store).await);
    checks.push(check_reversal_rules(store).await);
    checks.push(check_clearing_rules(store).await);
    checks.push(check_open_items_track_residuals(store).await);
    checks.push(check_account_bindings_survive_a_restart(store).await);
    checks.push(check_balance_limits_are_enforced(store).await);
    checks.push(check_a_posted_leaf_cannot_gain_a_child(store).await);
    checks.push(check_balances_slice_by_dimension(store).await);
    checks.push(check_statements_scope_to_a_period(store).await);
    checks.push(check_queries_fold_by_value_date(store).await);
    checks.push(check_kind_survives_a_round_trip(store).await);
    checks.push(check_dimensions_survive_a_round_trip(store).await);
    checks.push(check_balances_agree_across_readers(store).await);
    checks.push(check_statement_pages_do_not_repeat_or_skip(store).await);
    checks.push(check_checkpoints_round_trip(store).await);
    // Last: sealing changes what the calendar will accept, and every check
    // above books into 2026-03.
    checks.push(check_period_lifecycle_and_seals(store).await);
    Report { checks }
}

/// Posting dimensions survive a store round-trip.
///
/// Like `kind`, they are part of the content hash, so a backend that loses an
/// axis makes every dimensioned entry unreadable rather than merely
/// under-reported — [`get`](LedgerStore::get) rehydrates through
/// `adopt_verified`, which recomputes the hash and refuses a mismatch.
pub async fn check_dimensions_survive_a_round_trip<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "posting dimensions survive a round-trip";
    let f = Fixture::new();
    let Some(entry) = f.dimensioned_entry::<P>(b"dimensioned", 4242) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let id = entry.id();
    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }
    match store.get(id).await {
        Ok(Some(record)) => {
            let Some(posting) = record.entry.postings().first() else {
                return CheckResult::fail(NAME, "the entry came back with no postings");
            };
            let axes: Vec<(String, String)> = posting
                .dimensions
                .iter()
                .map(|(a, v)| (a.as_str().to_owned(), v.as_str().to_owned()))
                .collect();
            if axes
                == vec![
                    ("activity".to_owned(), "Network".to_owned()),
                    ("segment".to_owned(), "Electricity".to_owned()),
                ]
            {
                CheckResult::pass(NAME)
            } else {
                CheckResult::fail(NAME, format!("dimensions came back as {axes:?}"))
            }
        }
        Ok(None) => CheckResult::fail(NAME, "a dimensioned entry was not found"),
        Err(e) => CheckResult::fail(NAME, format!("get failed (hash mismatch?): {e}")),
    }
}

/// The three ways to read a balance agree with one another.
///
/// [`balance`](LedgerStore::balance), [`trial_balance`](LedgerStore::trial_balance)
/// and [`balances`](LedgerStore::balances) are three queries over the same
/// definition. A backend that optimises one of them — a materialised total, a
/// checkpoint short-circuit — and gets it subtly wrong shows no symptom until a
/// report disagrees with an account statement.
pub async fn check_balances_agree_across_readers<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "balance readers agree";
    let f = Fixture::new();
    let key = f.key();

    let single = match store.balance(key, BalanceQuery::all()).await {
        Ok(b) => b,
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    };
    match store.trial_balance(BalanceQuery::all()).await {
        Ok(tb) if tb.get_or_zero(&key) == single => {}
        Ok(tb) => {
            return CheckResult::fail(
                NAME,
                format!(
                    "trial_balance says {:?}, balance says {single:?}",
                    tb.get_or_zero(&key)
                ),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("trial_balance failed: {e}")),
    }
    match store
        .balances(
            &[f.left, f.right],
            Currency::EUR,
            Layer::Settled,
            BalanceQuery::all(),
        )
        .await
    {
        Ok(many) if many.get(&f.left).copied() == Some(single) => {}
        Ok(many) => {
            return CheckResult::fail(
                NAME,
                format!(
                    "balances says {:?}, balance says {single:?}",
                    many.get(&f.left)
                ),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("balances failed: {e}")),
    }

    // An account nobody posted to is absent, not zero: the caller knows what it
    // asked for, and inventing a row would hide a mis-typed handle.
    match store
        .balances(
            &[AccountId::from_index(u32::MAX)],
            Currency::EUR,
            Layer::Settled,
            BalanceQuery::all(),
        )
        .await
    {
        Ok(empty) if empty.is_empty() => CheckResult::pass(NAME),
        Ok(_) => CheckResult::fail(NAME, "an unposted account came back with a balance"),
        Err(e) => CheckResult::fail(NAME, format!("balances failed: {e}")),
    }
}

/// Paging a statement visits every line exactly once, with a running balance
/// that carries across page boundaries.
pub async fn check_statement_pages_do_not_repeat_or_skip<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "statement pagination is exact";
    let f = Fixture::new();
    let key = f.key();

    // Seed an entry that puts *three* postings on this one account — a split
    // receipt booked as three lines against one credit, which is an ordinary
    // entry. Without it every page boundary falls on an entry boundary, and a
    // cursor that addresses entries rather than postings passes by luck.
    let split = Entry::<Draft, P>::new(
        EntryId::generate(),
        match IdempotencyKey::new(b"conformance-split-statement".to_vec()) {
            Ok(k) => k,
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture key: {e}")),
        },
        date!(2026 - 03 - 21),
    )
    .debit(f.left, Amount::<P>::from_minor(11), Currency::EUR)
    .debit(f.left, Amount::<P>::from_minor(22), Currency::EUR)
    .debit(f.left, Amount::<P>::from_minor(33), Currency::EUR)
    .credit(f.right, Amount::<P>::from_minor(66), Currency::EUR)
    .seal(&f.ctx());
    match split {
        Ok(entry) => match EntryBatch::new(vec![entry]) {
            Ok(batch) => {
                if let Err(e) = store.append(&batch).await {
                    return CheckResult::fail(NAME, format!("append failed: {e}"));
                }
            }
            Err(e) => return CheckResult::fail(NAME, format!("batch failed: {e}")),
        },
        Err(e) => return CheckResult::fail(NAME, format!("fixture entry failed: {e}")),
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail(NAME, e);
    }

    let whole = match store
        .statement(
            key,
            BalanceQuery::all(),
            PostingCursor::start().with_limit(MAX_PAGE_SIZE),
        )
        .await
    {
        Ok(page) => page.lines,
        Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
    };
    if whole.is_empty() {
        return CheckResult::fail(NAME, "no statement lines to page over");
    }

    let mut paged = Vec::new();
    let mut cursor = Some(PostingCursor::start().with_limit(1));
    let mut guard = 0usize;
    while let Some(c) = cursor {
        guard = guard.saturating_add(1);
        if guard > whole.len().saturating_add(8) {
            return CheckResult::fail(NAME, "pagination did not terminate");
        }
        match store.statement(key, BalanceQuery::all(), c).await {
            Ok(page) => {
                if page.lines.is_empty() && page.next.is_some() {
                    return CheckResult::fail(NAME, "an empty page handed back another cursor");
                }
                paged.extend(page.lines);
                cursor = page.next;
            }
            Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
        }
    }

    // The seeding above must actually have produced a multi-posting entry, or
    // this check is back to only ever splitting on entry boundaries.
    let splits_inside_an_entry = whole
        .iter()
        .zip(whole.iter().skip(1))
        .any(|(a, b)| a.index == b.index && a.posting.index != b.posting.index);
    if !splits_inside_an_entry {
        return CheckResult::fail(
            NAME,
            "no entry contributed two lines, so page boundaries never fell inside one",
        );
    }

    if paged == whole {
        CheckResult::pass(NAME)
    } else {
        CheckResult::fail(
            NAME,
            format!(
                "paging produced {} lines against {} in one page; \
                 first divergence: {:?}",
                paged.len(),
                whole.len(),
                paged.iter().zip(whole.iter()).find(|(a, b)| a != b)
            ),
        )
    }
}

/// A checkpoint written is a checkpoint read.
///
/// A checkpoint is a cache for a definition, so it is only safe if what comes
/// back is what went in — including the tree head that pins it to one history.
/// A backend that drops the head returns a checkpoint that cannot be shown to be
/// stale, which is worse than no checkpoint at all.
pub async fn check_checkpoints_round_trip<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "checkpoints round-trip";
    let f = Fixture::new();
    let key = f.key();

    let (head, balance) = match (
        store.head().await,
        store.balance(key, BalanceQuery::all()).await,
    ) {
        (Ok(head), Ok(balance)) => (head, balance),
        (Err(e), _) | (_, Err(e)) => return CheckResult::fail(NAME, format!("read failed: {e}")),
    };
    let checkpoint = crate::checkpoint::Checkpoint::new(key, balance, head);

    if let Err(e) = store.save_checkpoint(&checkpoint).await {
        return CheckResult::fail(NAME, format!("save_checkpoint failed: {e}"));
    }
    match store.load_checkpoint(key).await {
        Ok(Some(loaded)) if loaded == checkpoint => {}
        Ok(Some(loaded)) => {
            return CheckResult::fail(
                NAME,
                format!("checkpoint came back as {loaded:?}, not {checkpoint:?}"),
            );
        }
        Ok(None) => return CheckResult::fail(NAME, "a saved checkpoint was not found"),
        Err(e) => return CheckResult::fail(NAME, format!("load_checkpoint failed: {e}")),
    }

    // Saving an *earlier* checkpoint must not replace a later one. A checkpoint
    // is a cache for a fold, so going backwards records nothing and loses work —
    // and with two writers it would make the stored value depend on arrival
    // order, which is how a caller that checkpoints from more than one place
    // ends up re-folding the whole log.
    if head.size > 0 {
        let earlier_size = head.size.saturating_sub(1);
        let earlier = match (
            store.head_at(earlier_size).await,
            store
                .balance(key, BalanceQuery::over_prefix(earlier_size))
                .await,
        ) {
            (Ok(h), Ok(b)) => crate::checkpoint::Checkpoint::new(key, b, h),
            (Err(e), _) | (_, Err(e)) => {
                return CheckResult::fail(NAME, format!("read failed: {e}"));
            }
        };
        if let Err(e) = store.save_checkpoint(&earlier).await {
            return CheckResult::fail(NAME, format!("save_checkpoint failed: {e}"));
        }
        match store.load_checkpoint(key).await {
            Ok(Some(loaded)) if loaded == checkpoint => {}
            Ok(other) => {
                return CheckResult::fail(
                    NAME,
                    format!(
                        "saving a checkpoint at {earlier_size} replaced the one at \
                         {}: got {other:?}",
                        head.size
                    ),
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("load_checkpoint failed: {e}")),
        }

        // And re-saving the stored one is a no-op rather than an error, so an
        // idempotent caller is not punished for retrying.
        if let Err(e) = store.save_checkpoint(&checkpoint).await {
            return CheckResult::fail(NAME, format!("re-saving a checkpoint failed: {e}"));
        }
    }

    // A key never checkpointed reads as absent rather than as a zero.
    let untouched = BalanceKey {
        account: AccountId::from_index(u32::MAX),
        currency: Currency::EUR,
        layer: Layer::Pending,
    };
    match store.load_checkpoint(untouched).await {
        Ok(None) => CheckResult::pass(NAME),
        Ok(Some(_)) => CheckResult::fail(NAME, "a checkpoint appeared that was never saved"),
        Err(e) => CheckResult::fail(NAME, format!("load_checkpoint failed: {e}")),
    }
}

/// Periods persist, follow their lifecycle, and seal into a chain.
///
/// The calendar is store state, not caller state: a sealed period that came back
/// open after a restart would accept postings into books already committed to.
/// This check exercises the whole path — define, transition, seal, read back —
/// and the refusals that guard it.
pub async fn check_period_lifecycle_and_seals<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "periods persist and seal into a chain";

    let id = match crate::period::PeriodId::new("conformance-2026-03") {
        Ok(id) => id,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture identifier: {e}")),
    };
    let Ok(period) =
        crate::period::Period::new(id.clone(), date!(2026 - 03 - 01), date!(2026 - 03 - 31))
    else {
        return CheckResult::fail(NAME, "bad fixture range");
    };

    if let Err(e) = store.define_period(&period).await {
        return CheckResult::fail(NAME, format!("define_period failed: {e}"));
    }
    // Declaring the same period again is how a caller states its calendar on
    // every start-up, so it must not be an error.
    if let Err(e) = store.define_period(&period).await {
        return CheckResult::fail(NAME, format!("re-defining an identical period failed: {e}"));
    }

    // Sealing an open period is refused: stopping postings is a separate,
    // earlier decision, so verification runs against a set that cannot grow.
    if store.seal_period(&id).await.is_ok() {
        return CheckResult::fail(NAME, "an open period was sealed");
    }

    // An *earlier* period, defined and left open. A seal's closing balance is
    // cumulative through its period's last day, so sealing March while February
    // still accepts postings would let one ordinary February booking restate it
    // afterwards — with every seal, proof and chain still verifying. A backend
    // must refuse to seal out of date order.
    let Ok(earlier_id) = crate::period::PeriodId::new("conformance-2026-02") else {
        return CheckResult::fail(NAME, "bad fixture identifier");
    };
    let Ok(earlier) = crate::period::Period::new(
        earlier_id.clone(),
        date!(2026 - 02 - 01),
        date!(2026 - 02 - 28),
    ) else {
        return CheckResult::fail(NAME, "bad fixture range");
    };
    if let Err(e) = store.define_period(&earlier).await {
        return CheckResult::fail(NAME, format!("define_period failed: {e}"));
    }
    if store
        .transition_period(&id, crate::period::PeriodState::Sealed)
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "a period jumped straight from open to sealed");
    }

    if let Err(e) = store
        .transition_period(&id, crate::period::PeriodState::Closing)
        .await
    {
        return CheckResult::fail(NAME, format!("open to closing was refused: {e}"));
    }
    match store.periods().await {
        Ok(periods) => {
            let found = periods
                .iter()
                .find(|p| p.id == id)
                .map(|p| (p.state, p.start, p.end));
            if found
                != Some((
                    crate::period::PeriodState::Closing,
                    date!(2026 - 03 - 01),
                    date!(2026 - 03 - 31),
                ))
            {
                return CheckResult::fail(NAME, format!("period read back as {found:?}"));
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("periods failed: {e}")),
    }

    if store.seal_period(&id).await.is_ok() {
        return CheckResult::fail(
            NAME,
            "a period was sealed while an earlier period was still open, so its \
             closing balance can still be restated",
        );
    }

    // Seal the earlier one first, and the later one becomes sealable.
    if let Err(e) = store
        .transition_period(&earlier_id, crate::period::PeriodState::Closing)
        .await
    {
        return CheckResult::fail(NAME, format!("open to closing was refused: {e}"));
    }
    if let Err(e) = store.seal_period(&earlier_id).await {
        return CheckResult::fail(NAME, format!("the earlier period would not seal: {e}"));
    }

    let seal = match store.seal_period(&id).await {
        Ok(seal) => seal,
        Err(e) => return CheckResult::fail(NAME, format!("seal_period failed: {e}")),
    };
    if !seal.is_self_consistent() {
        return CheckResult::fail(NAME, "the seal does not hash its own contents");
    }
    if seal.ledger != *store.ledger() {
        return CheckResult::fail(
            NAME,
            format!(
                "the seal names ledger {}, not {}",
                seal.ledger,
                store.ledger()
            ),
        );
    }
    if store.seal_period(&id).await.is_ok() {
        return CheckResult::fail(NAME, "a sealed period was sealed again");
    }
    match store.periods().await {
        Ok(periods) => {
            if periods.iter().find(|p| p.id == id).map(|p| p.state)
                != Some(crate::period::PeriodState::Sealed)
            {
                return CheckResult::fail(NAME, "sealing did not advance the period's state");
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("periods failed: {e}")),
    }

    // The seal must commit to the registry its balances are keyed on. A backend
    // that computes the trial balance root but not this one leaves every account
    // handle in it floating: renumbering the accounts table afterwards would
    // keep the whole chain verifying while every balance meant something else.
    match store.accounts().await {
        Ok(records) => match AccountRegistry::from_records(records) {
            Ok(registry) if registry.commitment() == seal.accounts => {}
            Ok(_) => {
                return CheckResult::fail(
                    NAME,
                    "the seal's accounts head does not match the stored account bindings",
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("registry would not rebuild: {e}")),
        },
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    }

    // And the *chain* must say so against the registry the store holds, not only
    // the seal just taken. This is what catches renumbering after the fact: the
    // seals and the log are untouched, so every other check here passes while
    // each sealed balance refers to a different account.
    match store.seals().await {
        Ok(seals) => {
            let chain = match crate::seal::SealChain::from_seals(store.ledger().clone(), seals) {
                Ok(chain) => chain,
                Err(e) => {
                    return CheckResult::fail(NAME, format!("the stored seals do not chain: {e}"));
                }
            };
            match store.accounts().await {
                Ok(records) => match AccountRegistry::from_records(records) {
                    Ok(registry) => {
                        if let Err(e) = chain.verify_against_accounts(&registry) {
                            return CheckResult::fail(
                                NAME,
                                format!("the seal chain does not describe these accounts: {e}"),
                            );
                        }
                    }
                    Err(e) => {
                        return CheckResult::fail(NAME, format!("registry would not rebuild: {e}"));
                    }
                },
                Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("seals failed: {e}")),
    }

    // A sealed balance must stay provable and nameable after the books move on.
    // `Seal::accounts` is the registry commitment as of the seal, so a backend
    // proving against the *current* registry cannot name a balance once one more
    // account is onboarded — or once a routine `close` lands. Both are ordinary
    // operations, so failing here means failing in production.
    let onboarded = Account::new(
        match AccountPath::parse("Conformance:Later") {
            Ok(p) => p,
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
        },
        date!(2026 - 05 - 01),
    );
    let later_handle = AccountId::from_index(match u32::try_from(seal.accounts.size) {
        Ok(n) => n,
        Err(_) => return CheckResult::fail(NAME, "registry size does not fit a handle"),
    });
    if let Err(e) = store
        .register_account(&AccountRecord {
            id: later_handle,
            account: onboarded,
        })
        .await
    {
        return CheckResult::fail(NAME, format!("register_account failed: {e}"));
    }

    // Every row in the sealed closing balance must still prove and still name
    // its account — not just one of them, and not only before the registry
    // moved on. Which accounts those are is read off the store rather than
    // assumed, so this does not depend on the fixture's naming.
    let closing = match store
        .trial_balance(BalanceQuery::through(date!(2026 - 03 - 31)))
        .await
    {
        Ok(tb) => tb,
        Err(e) => {
            return CheckResult::fail(NAME, format!("trial_balance through a date failed: {e}"));
        }
    };
    if closing.is_empty() {
        return CheckResult::fail(NAME, "the period sealed with an empty closing balance");
    }
    let stored_paths = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    for (key, balance) in closing.iter() {
        match store.prove_sealed_balance(&id, *key).await {
            Ok(SealedBalanceOutcome::Proven(proven)) => {
                if !proven.verify() {
                    return CheckResult::fail(
                        NAME,
                        format!("the sealed balance for {} did not verify", key.account),
                    );
                }
                if proven.balance.balance != *balance {
                    return CheckResult::fail(
                        NAME,
                        format!("the proof for {} carries a different balance", key.account),
                    );
                }
                if proven.seal.seal_hash != seal.seal_hash {
                    return CheckResult::fail(NAME, "the proof came from a different seal");
                }
                let expected = stored_paths
                    .iter()
                    .find(|r| r.id == key.account)
                    .map(|r| &r.account.path);
                if expected != Some(proven.path()) {
                    return CheckResult::fail(
                        NAME,
                        format!(
                            "the sealed balance for {} named {}, not the stored path",
                            key.account,
                            proven.path()
                        ),
                    );
                }
            }
            Ok(absent) => {
                return CheckResult::fail(
                    NAME,
                    format!(
                        "{} is in the closing balance but came back as {absent:?}",
                        key.account
                    ),
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("prove_sealed_balance failed: {e}")),
        }
    }

    // The stored chain must reproduce what was returned, and verify.
    match store.seals().await {
        Ok(stored) => {
            if !stored.iter().any(|s| s.seal_hash == seal.seal_hash) {
                return CheckResult::fail(NAME, "the seal was not read back");
            }
            let chain = match crate::seal::SealChain::from_seals(store.ledger().clone(), stored) {
                Ok(chain) => chain,
                Err(e) => {
                    return CheckResult::fail(NAME, format!("seals do not chain in order: {e}"));
                }
            };
            if let Err(e) = chain.verify() {
                return CheckResult::fail(NAME, format!("the seal chain does not verify: {e}"));
            }

            // Stronger, and the one that ties evidence to books: every seal's
            // tree head has to be the head *this store's log* had at that size.
            // `verify` alone is satisfied by any internally consistent chain,
            // including one over a history the store does not hold — which is
            // exactly what a rebuilt log looks like.
            let mut leaves = Vec::new();
            let mut cursor = Some(Cursor::start());
            while let Some(c) = cursor {
                match store.page(c).await {
                    Ok(page) => {
                        leaves.extend(page.records.iter().map(|r| r.content_hash));
                        cursor = page.next;
                    }
                    Err(e) => return CheckResult::fail(NAME, format!("page failed: {e}")),
                }
            }
            match chain.verify_against_log(&MerkleLog::from_leaves(leaves)) {
                Ok(()) => CheckResult::pass(NAME),
                Err(e) => CheckResult::fail(
                    NAME,
                    format!("the seal chain does not describe this store's log: {e}"),
                ),
            }
        }
        Err(e) => CheckResult::fail(NAME, format!("seals failed: {e}")),
    }
}

/// An entry's `kind` label survives a store round-trip.
///
/// `kind` is part of the content hash, so a backend that fails to persist and
/// rehydrate it makes every kinded entry unreadable — [`get`](LedgerStore::get)
/// rehydrates through `adopt_verified`, which recomputes the hash and rejects a
/// mismatch. The check also asserts the statement line carries the kind, so a
/// caller can group by document type without a second lookup.
pub async fn check_kind_survives_a_round_trip<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "entry kind survives a round-trip";
    let f = Fixture::new();
    let Some(entry) = f.entry_with_kind::<P>(b"kinded", 5150, "INVOICE") else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let id = entry.id();
    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        // A hash mismatch here is exactly the "kind not persisted" failure.
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }
    // The statement half of this check reads the log, which a deferred backend
    // does not place the entry into until the sequencer runs. Without this the
    // check would silently only ever hold for inline sequencing.
    match sequence_until_placed(store, id).await {
        Ok(Some(_)) => {}
        Ok(None) => return CheckResult::fail(NAME, "the kinded entry was never sequenced"),
        Err(e) => return CheckResult::fail(NAME, e),
    }
    match store.get(id).await {
        Ok(Some(r)) => {
            let got = r.entry.kind().map(|k| k.as_str().to_owned());
            if got.as_deref() != Some("INVOICE") {
                return CheckResult::fail(
                    NAME,
                    format!("kind did not round-trip: expected Some(\"INVOICE\"), got {got:?}"),
                );
            }
        }
        Ok(None) => return CheckResult::fail(NAME, "a kinded entry was not found"),
        Err(e) => return CheckResult::fail(NAME, format!("get failed (hash mismatch?): {e}")),
    }

    // And the statement line exposes it.
    let key = BalanceKey {
        account: f.left,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    match store
        .statement(key, BalanceQuery::all(), PostingCursor::start())
        .await
    {
        Ok(page) => {
            let seen = page
                .lines
                .iter()
                .any(|l| l.kind.as_ref().map(|k| k.as_str()) == Some("INVOICE"));
            if seen {
                CheckResult::pass(NAME)
            } else {
                CheckResult::fail(NAME, "statement line did not carry the entry kind")
            }
        }
        Err(e) => CheckResult::fail(NAME, format!("statement failed: {e}")),
    }
}

/// Account handles read back exactly as they were written.
///
/// A handle is a position in registration order, and that position is written
/// into every posting row and into the trial balance leaves a seal commits to.
/// A backend that loses the binding — or reissues it from iteration order —
/// silently repoints history on the next restart, so the round trip is part of
/// the contract rather than a convenience.
pub async fn check_account_bindings_survive_a_restart<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "account bindings survive a restart";

    // Start from whatever the store already holds, so the check extends an
    // existing binding set rather than assuming it owns index zero.
    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };

    // Paths deliberately out of lexical order, so a backend that rebuilds by
    // sorting paths rather than by stored index fails here.
    let mut expected = Vec::new();
    for path in ["Zzconf:Late", "Aaconf:Early", "Mmconf:Middle"] {
        match registry.register(
            Account::new(
                match AccountPath::parse(path) {
                    Ok(p) => p,
                    Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
                },
                date!(2000 - 01 - 01),
            )
            .with_kind(AccountKind::Asset)
            .closing_on(date!(2030 - 12 - 31)),
        ) {
            Ok(id) => expected.push((id, path)),
            Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
        }
    }

    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }

    let stored = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };

    for (id, path) in &expected {
        let Some(found) = stored.iter().find(|r| r.id == *id) else {
            return CheckResult::fail(NAME, format!("handle {id} for {path} was not stored"));
        };
        if found.account.path.to_string() != *path {
            return CheckResult::fail(
                NAME,
                format!(
                    "handle {id} came back as {}, not {path}",
                    found.account.path
                ),
            );
        }
        // Classification and closing date are part of the binding: validation
        // reads them, so a backend that drops them changes what may be posted.
        if found.account.kind != Some(AccountKind::Asset) {
            return CheckResult::fail(NAME, format!("handle {id} lost its kind"));
        }
        if found.account.closed_on != Some(date!(2030 - 12 - 31)) {
            return CheckResult::fail(NAME, format!("handle {id} lost its closing date"));
        }
    }

    // Rebuilding from the stored records must reproduce the registry exactly,
    // handles included — that is what makes a restart safe.
    let rebuilt = match AccountRegistry::from_records(stored) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("registry would not rebuild: {e}")),
    };
    if rebuilt.commitment() != registry.commitment() {
        return CheckResult::fail(NAME, "rebuilt registry does not match the original");
    }

    CheckResult::pass(NAME)
}

/// A leaf that has been posted to cannot acquire a child.
///
/// The rule "only leaves are postable" is checked when an entry is validated,
/// which settles the question in one direction only: an account that is a leaf
/// today can be turned into an aggregation node tomorrow by registering
/// something beneath it. Every entry already recorded against it would then
/// violate the rule it was accepted under, and the log is append-only, so
/// nothing can be taken back.
///
/// A registry cannot catch this — it holds no postings. A store can, and must,
/// or the invariant holds only until the next master-data change.
pub async fn check_a_posted_leaf_cannot_gain_a_child<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "a posted-to leaf cannot gain a child";

    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };

    // A fresh pair, so the check does not depend on what earlier checks booked.
    let mut ids = Vec::new();
    for path in ["Leafconf:Bucket", "Leafconf:Other"] {
        let parsed = match AccountPath::parse(path) {
            Ok(p) => p,
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
        };
        match registry.register(Account::new(parsed, date!(2000 - 01 - 01))) {
            Ok(id) => ids.push(id),
            Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
        }
    }
    let (bucket, other) = match ids.as_slice() {
        [bucket, other] => (*bucket, *other),
        _ => return CheckResult::fail(NAME, "fixture registration returned the wrong shape"),
    };
    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }

    let calendar = PeriodCalendar::new();
    let policy = LedgerPolicy::default();
    let ctx = SealContext {
        accounts: &registry,
        calendar: &calendar,
        policy: &policy,
    };
    let Ok(entry) = Entry::<Draft, P>::new(
        EntryId::generate(),
        match IdempotencyKey::new(b"leaf-rule".to_vec()) {
            Ok(k) => k,
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture key: {e}")),
        },
        date!(2026 - 03 - 15),
    )
    .debit(bucket, Amount::<P>::from_minor(500), Currency::EUR)
    .credit(other, Amount::<P>::from_minor(500), Currency::EUR)
    .seal(&ctx) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    // Now the bucket has postings. A child under it must be refused.
    let child = match AccountPath::parse("Leafconf:Bucket:Petty") {
        Ok(p) => p,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
    };
    let Ok(child_id) = registry.register(Account::new(child, date!(2000 - 01 - 01))) else {
        return CheckResult::fail(NAME, "fixture registration failed");
    };
    let Some(record) = registry.records().into_iter().find(|r| r.id == child_id) else {
        return CheckResult::fail(NAME, "the child record went missing");
    };
    if store.register_account(&record).await.is_ok() {
        return CheckResult::fail(
            NAME,
            "a child was registered under an account that has already been posted to",
        );
    }

    // And the refusal wrote nothing.
    match store.accounts().await {
        Ok(stored) if stored.iter().any(|r| r.id == child_id) => {
            CheckResult::fail(NAME, "a refused registration still wrote the binding")
        }
        Ok(_) => CheckResult::pass(NAME),
        Err(e) => CheckResult::fail(NAME, format!("accounts failed: {e}")),
    }
}

/// Balances can be sliced by a reporting axis, and by a date range.
///
/// Dimensions exist to be reported on. A backend that stores and rehydrates them
/// but cannot filter by them has a write-only feature, and the index the
/// reference schema ships for exactly this would never be used.
///
/// Three things are checked, and the third is the one a naive implementation
/// gets wrong: an equality clause, an **absence** clause, and that the two
/// together partition the account. Slice by the values an axis takes and the
/// totals come up short, because postings carrying no value land in none of the
/// slices — silently, which is the whole reason the absence clause exists.
pub async fn check_balances_slice_by_dimension<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "balances slice by dimension and date";
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    let f = Fixture::new();

    // A fresh account, so the totals below are this check's own. Every other
    // check books against the shared pair.
    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };
    let path = match AccountPath::parse("Conformance:Sliced") {
        Ok(p) => p,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
    };
    let sliced = match registry.register(Account::new(path, date!(2000 - 01 - 01))) {
        Ok(id) => id,
        Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
    };
    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }
    let ctx = SealContext {
        accounts: &registry,
        calendar: &f.calendar,
        policy: &f.policy,
    };

    let Ok(axis) = Label::new("conformance-segment") else {
        return CheckResult::fail(NAME, "bad fixture axis");
    };
    let slice = |value: &str| {
        Label::new(value)
            .ok()
            .and_then(|v| Dimensions::none().with(axis.clone(), v).ok())
    };

    // Three postings on one account: two attributed, one not, on two dates.
    let rows: [(&[u8], i64, Option<&str>, time::Date); 3] = [
        (b"dim-retail", 100, Some("Retail"), date!(2026 - 03 - 10)),
        (b"dim-trade", 200, Some("Trade"), date!(2026 - 03 - 20)),
        (b"dim-none", 400, None, date!(2026 - 04 - 05)),
    ];
    for (key, minor, value, on) in rows {
        let amount = Amount::<P>::from_minor(minor);
        let (mut debit, mut credit) = (
            crate::Posting::debit(sliced, amount, Currency::EUR),
            crate::Posting::credit(f.right, amount, Currency::EUR),
        );
        if let Some(value) = value {
            let Some(dims) = slice(value) else {
                return CheckResult::fail(NAME, "bad fixture dimension");
            };
            debit = debit.with_dimensions(dims.clone());
            credit = credit.with_dimensions(dims);
        }
        let Ok(idem) = IdempotencyKey::new(key.to_vec()) else {
            return CheckResult::fail(NAME, "bad fixture key");
        };
        let Ok(entry) = Entry::<Draft, P>::new(EntryId::generate(), idem, on)
            .post(debit)
            .post(credit)
            .seal(&ctx)
        else {
            return CheckResult::fail(NAME, "fixture entry failed to seal");
        };
        if let Err(e) = store.append(&EntryBatch::single(entry)).await {
            return CheckResult::fail(NAME, format!("append failed: {e}"));
        }
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }

    let key = BalanceKey {
        account: sliced,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    let debits = |b: crate::Balance<P>| b.debits.to_minor();

    // The values the axis takes — absences are not among them.
    match store.dimension_values(axis.as_str()).await {
        Ok(values) => {
            let found: Vec<&str> = values.iter().map(Label::as_str).collect();
            if found != vec!["Retail", "Trade"] {
                return CheckResult::fail(
                    NAME,
                    format!("dimension_values returned {found:?}, not [Retail, Trade]"),
                );
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("dimension_values failed: {e}")),
    }

    let Ok(retail_value) = Label::new("Retail") else {
        return CheckResult::fail(NAME, "bad fixture value");
    };
    let retail = DimensionFilter::any().matching(axis.clone(), retail_value);
    let unattributed = DimensionFilter::any().missing(axis.clone());

    // An equality clause.
    match store
        .balance(key, BalanceQuery::all().matching(&retail))
        .await
    {
        Ok(b) if debits(b) == 100 => {}
        Ok(b) => {
            return CheckResult::fail(
                NAME,
                format!("the Retail slice came back as {}, not 100", debits(b)),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    }

    // An absence clause — and that the slices plus the absences are the whole.
    let whole = match store.balance(key, BalanceQuery::all()).await {
        Ok(b) => debits(b),
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    };
    let rest = match store
        .balance(key, BalanceQuery::all().matching(&unattributed))
        .await
    {
        Ok(b) => debits(b),
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    };
    let mut attributed = 0i64;
    for value in ["Retail", "Trade"] {
        let Ok(value) = Label::new(value) else {
            return CheckResult::fail(NAME, "bad fixture value");
        };
        let filter = DimensionFilter::any().matching(axis.clone(), value);
        match store
            .balance(key, BalanceQuery::all().matching(&filter))
            .await
        {
            Ok(b) => attributed = attributed.saturating_add(debits(b)),
            Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
        }
    }
    if attributed.saturating_add(rest) != whole {
        return CheckResult::fail(
            NAME,
            format!("slices ({attributed}) plus unattributed ({rest}) is not the whole ({whole})"),
        );
    }

    // A date range is activity, not the cumulative position.
    let march = BalanceQuery::between(date!(2026 - 03 - 01), date!(2026 - 03 - 31));
    match store.balance(key, march).await {
        Ok(b) if debits(b) == 300 => {}
        Ok(b) => {
            return CheckResult::fail(
                NAME,
                format!("March's activity came back as {}, not 300", debits(b)),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    }

    // And the two narrowings compose, in a trial balance as well as a balance.
    match store.trial_balance(march.matching(&retail)).await {
        Ok(tb) if tb.get_or_zero(&key).debits.to_minor() == 100 => {}
        Ok(tb) => {
            return CheckResult::fail(
                NAME,
                format!(
                    "Retail in March came back as {}, not 100",
                    tb.get_or_zero(&key).debits.to_minor()
                ),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("trial_balance failed: {e}")),
    }

    CheckResult::pass(NAME)
}

/// A statement can be scoped to a period, and opens at the right figure.
///
/// The rule a backend gets wrong by being literal: an opening balance is folded
/// over everything the query narrows to that was booked **before** its window.
/// Apply the start date to both the lines and the opening and a March statement
/// opens at zero, which is not an opening balance — it is the absence of one.
///
/// The rule a backend gets wrong by being clever: bounding the carry set by log
/// position instead. Entries are appended in recording order, so this fixture
/// books **out of date order** on purpose — April first, February second, March
/// last. A position bound folds April into the opening and drops February.
///
/// Also checks the identity that makes a statement a statement: `opening` plus
/// the page's movements is the last line's `running`, page by page.
pub async fn check_statements_scope_to_a_period<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "statements scope to a period";
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    let f = Fixture::new();

    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };
    let path = match AccountPath::parse("Conformance:Scoped") {
        Ok(p) => p,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
    };
    let scoped = match registry.register(Account::new(path, date!(2000 - 01 - 01))) {
        Ok(id) => id,
        Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
    };
    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }
    let ctx = SealContext {
        accounts: &registry,
        calendar: &f.calendar,
        policy: &f.policy,
    };

    // 700 before the window, 300 inside it, 900 after — recorded in an order
    // that shares nothing with their booking dates, so a backend that bounds the
    // opening balance by log position rather than by date gets a different and
    // wrong answer for every figure below.
    let rows: [(&[u8], i64, time::Date); 3] = [
        (b"scoped-after", 900, date!(2026 - 04 - 10)),
        (b"scoped-before", 700, date!(2026 - 02 - 10)),
        (b"scoped-inside", 300, date!(2026 - 03 - 10)),
    ];
    for (key, minor, on) in rows {
        let amount = Amount::<P>::from_minor(minor);
        let Ok(idem) = IdempotencyKey::new(key.to_vec()) else {
            return CheckResult::fail(NAME, "bad fixture key");
        };
        let Ok(entry) = Entry::<Draft, P>::new(EntryId::generate(), idem, on)
            .debit(scoped, amount, Currency::EUR)
            .credit(f.right, amount, Currency::EUR)
            .seal(&ctx)
        else {
            return CheckResult::fail(NAME, "fixture entry failed to seal");
        };
        if let Err(e) = store.append(&EntryBatch::single(entry)).await {
            return CheckResult::fail(NAME, format!("append failed: {e}"));
        }
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }

    let key = BalanceKey {
        account: scoped,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    let march = BalanceQuery::between(date!(2026 - 03 - 01), date!(2026 - 03 - 31));
    let page = match store.statement(key, march, PostingCursor::start()).await {
        Ok(page) => page,
        Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
    };

    if page.lines.len() != 1 {
        return CheckResult::fail(
            NAME,
            format!(
                "March holds one movement, the statement listed {}",
                page.lines.len()
            ),
        );
    }
    if page.opening.debits.to_minor() != 700 {
        return CheckResult::fail(
            NAME,
            format!(
                "March opened at {}, not the 700 carried in from February",
                page.opening.debits.to_minor()
            ),
        );
    }
    let Some(last) = page.lines.last() else {
        return CheckResult::fail(NAME, "the page lost its line");
    };
    if last.running.debits.to_minor() != 1000 {
        return CheckResult::fail(
            NAME,
            format!(
                "the running balance closed at {}, not opening plus the movement",
                last.running.debits.to_minor()
            ),
        );
    }
    if last.booking_date != date!(2026 - 03 - 10) {
        return CheckResult::fail(NAME, "the statement listed a movement outside the window");
    }

    // Unscoped, the same account lists everything and opens at nothing.
    let all = match store
        .statement(key, BalanceQuery::all(), PostingCursor::start())
        .await
    {
        Ok(all) => all,
        Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
    };
    if all.lines.len() != 3 || all.opening != crate::Balance::ZERO {
        return CheckResult::fail(
            NAME,
            format!(
                "an unscoped statement listed {} lines opening at {:?}",
                all.lines.len(),
                all.opening
            ),
        );
    }

    // Paging must not move the figures. Reading the unscoped statement one line
    // at a time has to reproduce exactly the running balances the whole one
    // showed, which is only true if each page's opening carries the pages before
    // it — see `StatementPage::opening`.
    let mut cursor = PostingCursor::start().with_limit(1);
    for expected in &all.lines {
        let page = match store.statement(key, BalanceQuery::all(), cursor).await {
            Ok(page) => page,
            Err(e) => return CheckResult::fail(NAME, format!("paged statement failed: {e}")),
        };
        let Some(line) = page.lines.first() else {
            return CheckResult::fail(NAME, "paging the statement lost a line");
        };
        if line.running != expected.running {
            return CheckResult::fail(
                NAME,
                format!(
                    "paged, the line at {} ran to {:?} rather than the {:?} the \
                     whole statement showed",
                    line.index, line.running, expected.running
                ),
            );
        }
        let Ok(closing) = page.opening.checked_add(&crate::Balance {
            debits: if line.direction.is_debit() {
                line.amount
            } else {
                crate::Amount::ZERO
            },
            credits: if line.direction.is_credit() {
                line.amount
            } else {
                crate::Amount::ZERO
            },
        }) else {
            return CheckResult::fail(NAME, "a page's opening plus its movement overflowed");
        };
        if closing != line.running {
            return CheckResult::fail(
                NAME,
                format!(
                    "a page opened at {:?}, moved by {}, and closed at {:?}",
                    page.opening, line.amount, line.running
                ),
            );
        }
        match page.next {
            Some(next) => cursor = next.with_limit(1),
            None => break,
        }
    }

    CheckResult::pass(NAME)
}

/// A date-bounded query can fold by **value date** instead of booking date.
///
/// The fixture is the ordinary straddling entry: booked at the end of a month,
/// settling at the start of the next, so it belongs to the earlier month's books
/// and the later month's cash. A backend that stores `value_date` but filters on
/// `booking_date` reports the same figure for both.
pub async fn check_queries_fold_by_value_date<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "queries fold by value date";
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    let f = Fixture::new();

    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };
    let path = match AccountPath::parse("Conformance:Valued") {
        Ok(p) => p,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
    };
    let valued = match registry.register(Account::new(path, date!(2000 - 01 - 01))) {
        Ok(id) => id,
        Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
    };
    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }
    let ctx = SealContext {
        accounts: &registry,
        calendar: &f.calendar,
        policy: &f.policy,
    };

    // 100 wholly inside March; 400 booked in March, settling in April.
    let rows: [(&[u8], i64, time::Date, time::Date); 2] = [
        (
            b"valued-inside",
            100,
            date!(2026 - 03 - 10),
            date!(2026 - 03 - 10),
        ),
        (
            b"valued-straddle",
            400,
            date!(2026 - 03 - 28),
            date!(2026 - 04 - 02),
        ),
    ];
    for (key, minor, booked, valued_on) in rows {
        let amount = Amount::<P>::from_minor(minor);
        let Ok(idem) = IdempotencyKey::new(key.to_vec()) else {
            return CheckResult::fail(NAME, "bad fixture key");
        };
        let Ok(entry) = Entry::<Draft, P>::new(EntryId::generate(), idem, booked)
            .with_value_date(valued_on)
            .debit(valued, amount, Currency::EUR)
            .credit(f.right, amount, Currency::EUR)
            .seal(&ctx)
        else {
            return CheckResult::fail(NAME, "fixture entry failed to seal");
        };
        if let Err(e) = store.append(&EntryBatch::single(entry)).await {
            return CheckResult::fail(NAME, format!("append failed: {e}"));
        }
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }

    let key = BalanceKey {
        account: valued,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    let march = BalanceQuery::between(date!(2026 - 03 - 01), date!(2026 - 03 - 31));
    let april = BalanceQuery::between(date!(2026 - 04 - 01), date!(2026 - 04 - 30));

    for (label, query, expected) in [
        ("March, booked", march, 500),
        ("March, valued", march.by_value_date(), 100),
        ("April, booked", april, 0),
        ("April, valued", april.by_value_date(), 400),
    ] {
        match store.balance(key, query).await {
            Ok(balance) if balance.debits.to_minor() == expected => {}
            Ok(balance) => {
                return CheckResult::fail(
                    NAME,
                    format!(
                        "{label} came back as {}, not {expected}",
                        balance.debits.to_minor()
                    ),
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
        }
    }

    // A statement takes the same basis, and its opening balance with it.
    let page = match store
        .statement(key, april.by_value_date(), PostingCursor::start())
        .await
    {
        Ok(page) => page,
        Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
    };
    if page.lines.len() != 1 || page.opening.debits.to_minor() != 100 {
        return CheckResult::fail(
            NAME,
            format!(
                "a value-dated April statement listed {} lines opening at {}",
                page.lines.len(),
                page.opening.debits.to_minor()
            ),
        );
    }
    let Some(line) = page.lines.first() else {
        return CheckResult::fail(NAME, "the page lost its line");
    };
    if line.booking_date != date!(2026 - 03 - 28) || line.value_date != date!(2026 - 04 - 02) {
        return CheckResult::fail(
            NAME,
            format!(
                "the line came back booked {} valued {}",
                line.booking_date, line.value_date
            ),
        );
    }

    CheckResult::pass(NAME)
}

/// A balance limit is enforced by the store, and master data can be updated.
///
/// Two guarantees that only make sense together. A limit is worth nothing if a
/// store cannot record a change to it, and a store that could only ever *insert*
/// an account could not close one either — [`AccountRegistry`] treats the
/// classification, the open window and the limit as mutable master data, so a
/// backend that ignores a re-registration silently diverges from the engine the
/// first time an account is closed.
///
/// The limit itself has to be enforced in the write path rather than checked by
/// the caller beforehand: a read-then-write races, and two concurrent appends
/// that each stay within the limit can together breach it.
pub async fn check_balance_limits_are_enforced<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "balance limits are enforced";
    let f = Fixture::new();

    // A fresh account, so the check owns its balance outright.
    let existing = match store.accounts().await {
        Ok(records) => records,
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    };
    let mut registry = match AccountRegistry::from_records(existing) {
        Ok(r) => r,
        Err(e) => return CheckResult::fail(NAME, format!("stored bindings are unusable: {e}")),
    };
    let path = match AccountPath::parse("Conformance:Limited") {
        Ok(p) => p,
        Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
    };
    let limited = match registry.register(Account::new(path, date!(2000 - 01 - 01))) {
        Ok(id) => id,
        Err(e) => return CheckResult::fail(NAME, format!("fixture registration failed: {e}")),
    };
    for record in registry.records() {
        if let Err(e) = store.register_account(&record).await {
            return CheckResult::fail(NAME, format!("register_account failed: {e}"));
        }
    }

    // Master data moves: the account is registered unconstrained, then limited.
    if let Err(e) = registry.set_limit(limited, BalanceLimit::NoCreditBalance) {
        return CheckResult::fail(NAME, format!("set_limit failed: {e}"));
    }
    let Some(record) = registry.records().into_iter().find(|r| r.id == limited) else {
        return CheckResult::fail(NAME, "the limited account vanished from the registry");
    };
    if let Err(e) = store.register_account(&record).await {
        return CheckResult::fail(NAME, format!("updating master data failed: {e}"));
    }
    match store.accounts().await {
        Ok(stored) => match stored.iter().find(|r| r.id == limited) {
            Some(found) if found.account.limit == BalanceLimit::NoCreditBalance => {}
            Some(found) => {
                return CheckResult::fail(
                    NAME,
                    format!(
                        "the limit came back as {}, not no credit",
                        found.account.limit
                    ),
                );
            }
            None => return CheckResult::fail(NAME, "the limited account was not stored"),
        },
        Err(e) => return CheckResult::fail(NAME, format!("accounts failed: {e}")),
    }

    let ctx = SealContext {
        accounts: &registry,
        calendar: &f.calendar,
        policy: &f.policy,
    };
    let movement = |key: &[u8], minor: i64, into: bool| {
        let draft = Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 21),
        );
        let amount = Amount::<P>::from_minor(minor);
        let draft = if into {
            draft
                .debit(limited, amount, Currency::EUR)
                .credit(f.right, amount, Currency::EUR)
        } else {
            draft
                .credit(limited, amount, Currency::EUR)
                .debit(f.right, amount, Currency::EUR)
        };
        draft.seal(&ctx).ok()
    };

    let Some(funding) = movement(b"limit-funding", 1000, true) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(funding)).await {
        return CheckResult::fail(NAME, format!("funding was refused: {e}"));
    }

    // Exactly to zero is on the permitted side.
    let Some(drain) = movement(b"limit-drain", 1000, false) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(drain)).await {
        return CheckResult::fail(NAME, format!("draining to exactly zero was refused: {e}"));
    }

    // One minor unit past it is not.
    let Some(overdraw) = movement(b"limit-overdraw", 1, false) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let before = store.len().await;
    if store.append(&EntryBatch::single(overdraw)).await.is_ok() {
        return CheckResult::fail(NAME, "an entry breaching a balance limit was accepted");
    }
    match (before, store.len().await) {
        (Ok(before), Ok(after)) if before == after => {}
        (Ok(before), Ok(after)) => {
            return CheckResult::fail(
                NAME,
                format!("a refused entry still appended: {before} -> {after}"),
            );
        }
        (Err(e), _) | (_, Err(e)) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    }

    // A reservation draws on the same funds as a settled movement. The account
    // is back at zero, so there is nothing to hold — a backend that checked the
    // pending layer against its own zero-based balance would accept this, and a
    // limit that a reservation can step around is not a limit.
    let reserve = |key: &[u8], minor: i64| {
        let amount = Amount::<P>::from_minor(minor);
        Entry::<Draft, P>::new(
            EntryId::generate(),
            IdempotencyKey::new(key.to_vec()).ok()?,
            date!(2026 - 03 - 22),
        )
        .post(crate::Posting::credit(limited, amount, Currency::EUR).in_layer(Layer::Pending))
        .post(crate::Posting::debit(f.right, amount, Currency::EUR).in_layer(Layer::Pending))
        .seal(&ctx)
        .ok()
    };
    let Some(unfunded_hold) = reserve(b"limit-hold-empty", 1) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if store
        .append(&EntryBatch::single(unfunded_hold))
        .await
        .is_ok()
    {
        return CheckResult::fail(
            NAME,
            "a reservation was accepted against an account with nothing in it",
        );
    }

    // Fund it again, and a reservation within the funds is accepted — which the
    // opposite mistake, checking each layer against its own zero, would refuse:
    // an account that cannot reserve an outflow cannot use the pending layer at
    // all.
    let Some(refunding) = movement(b"limit-refunding", 500, true) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(refunding)).await {
        return CheckResult::fail(NAME, format!("refunding was refused: {e}"));
    }
    let Some(hold) = reserve(b"limit-hold", 400) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(hold)).await {
        return CheckResult::fail(
            NAME,
            format!("a reservation within the funds was refused: {e}"),
        );
    }

    // And the outstanding hold consumes what a settled draw can still take.
    let Some(too_much) = movement(b"limit-after-hold", 101, false) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if store.append(&EntryBatch::single(too_much)).await.is_ok() {
        return CheckResult::fail(
            NAME,
            "a settled draw ignored the reservation already outstanding",
        );
    }

    // Rebinding a handle to a different path is refused outright, because every
    // posting row that names it would silently repoint.
    let rebound = AccountRecord {
        id: limited,
        account: match AccountPath::parse("Conformance:Elsewhere") {
            Ok(p) => Account::new(p, date!(2000 - 01 - 01)),
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture path: {e}")),
        },
    };
    if store.register_account(&rebound).await.is_ok() {
        return CheckResult::fail(NAME, "a handle was rebound to a different path");
    }

    CheckResult::pass(NAME)
}

/// Corrections follow the rules: at most one reversal, never of a reversal,
/// never a claim that does not actually invert.
pub async fn check_reversal_rules<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "reversal rules are enforced";
    let f = Fixture::new();

    let Some(original) = f.entry::<P>(b"rev-original", 1000) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(original.clone())).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    let Some(reversal) = f.reverse(&original, b"rev-first", date!(2026 - 04 - 01)) else {
        return CheckResult::fail(NAME, "reversal failed to seal");
    };
    let reversal_clone = reversal.clone();
    if let Err(e) = store.append(&EntryBatch::single(reversal)).await {
        return CheckResult::fail(NAME, format!("a valid reversal was rejected: {e}"));
    }

    // A second reversal of the same entry.
    let Some(second) = f.reverse(&original, b"rev-second", date!(2026 - 04 - 02)) else {
        return CheckResult::fail(NAME, "reversal failed to seal");
    };
    if store.append(&EntryBatch::single(second)).await.is_ok() {
        return CheckResult::fail(NAME, "an entry was reversed twice");
    }

    // A reversal of a reversal.
    let Some(chained) = f.reverse(&reversal_clone, b"rev-chained", date!(2026 - 04 - 03)) else {
        return CheckResult::fail(NAME, "reversal failed to seal");
    };
    if store.append(&EntryBatch::single(chained)).await.is_ok() {
        return CheckResult::fail(NAME, "a reversal was itself reversed");
    }

    // A claim that does not invert. Uses a *fresh* original that has not been
    // reversed, so this cannot pass on the at-most-once rule instead.
    let Some(untouched) = f.entry::<P>(b"rev-untouched", 500) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(untouched.clone())).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }
    let Some(forged) = f.forged_reversal::<P>(&untouched, b"rev-forged") else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if store.append(&EntryBatch::single(forged)).await.is_ok() {
        return CheckResult::fail(
            NAME,
            "an entry claiming a reversal it does not perform was accepted",
        );
    }

    // Reversing an entry that is not in the store at all.
    let Some(absent) = f.entry::<P>(b"rev-absent", 700) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let Some(orphan) = f.reverse(&absent, b"rev-orphan", date!(2026 - 04 - 06)) else {
        return CheckResult::fail(NAME, "reversal failed to seal");
    };
    if store.append(&EntryBatch::single(orphan)).await.is_ok() {
        return CheckResult::fail(NAME, "a reversal of an unknown entry was accepted");
    }

    CheckResult::pass(NAME)
}

/// Clearing validates its own rules and never moves money.
pub async fn check_clearing_rules<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "clearing rules are enforced";
    let f = Fixture::new();

    let Some(invoice) = f.entry::<P>(b"clr-invoice", 1000) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let invoice_id = invoice.id();
    if let Err(e) = store.append(&EntryBatch::single(invoice)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    let Some(payment) = f.reversed_entry::<P>(b"clr-payment", 400) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let payment_id = payment.id();
    if let Err(e) = store.append(&EntryBatch::single(payment)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    let key = BalanceKey {
        account: f.left,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    let before = match store.trial_balance(BalanceQuery::all()).await {
        Ok(tb) => tb,
        Err(e) => return CheckResult::fail(NAME, format!("trial_balance failed: {e}")),
    };

    let invoice_ref = PostingRef::new(invoice_id, 0);
    let payment_ref = PostingRef::new(payment_id, 0);

    // Applying more than a posting has open.
    if store
        .clear(f.clearing::<P>(&[(invoice_ref, 9_000), (payment_ref, 9_000)]))
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "an over-application was accepted");
    }

    // Sides that do not match.
    if store
        .clear(f.clearing::<P>(&[(invoice_ref, 400), (payment_ref, 300)]))
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "an unbalanced clearing was accepted");
    }

    // The same posting twice in one clearing.
    if store
        .clear(f.clearing::<P>(&[(invoice_ref, 200), (invoice_ref, 200)]))
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "a duplicated item was accepted");
    }

    // A single item cannot clear anything.
    if store
        .clear(f.clearing::<P>(&[(invoice_ref, 400)]))
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "a one-sided clearing was accepted");
    }

    // A valid partial application.
    let valid = f.clearing::<P>(&[(invoice_ref, 400), (payment_ref, 400)]);
    let clearing_id = valid.id;
    if let Err(e) = store.clear(valid).await {
        return CheckResult::fail(NAME, format!("a valid clearing was rejected: {e}"));
    }

    // The same identifier again.
    let mut duplicate = f.clearing::<P>(&[(invoice_ref, 100), (payment_ref, 100)]);
    duplicate.id = clearing_id;
    if store.clear(duplicate).await.is_ok() {
        return CheckResult::fail(NAME, "a duplicate clearing identifier was accepted");
    }

    // Clearing is an assignment, never a movement.
    match store.trial_balance(BalanceQuery::all()).await {
        Ok(after) if after == before => {}
        Ok(_) => return CheckResult::fail(NAME, "clearing changed a balance"),
        Err(e) => return CheckResult::fail(NAME, format!("trial_balance failed: {e}")),
    }

    // Releasing it, then releasing it again.
    if let Err(e) = store
        .reset_clearing(clearing_id, date!(2026 - 05 - 01))
        .await
    {
        return CheckResult::fail(NAME, format!("a valid reset was rejected: {e}"));
    }
    if store
        .reset_clearing(clearing_id, date!(2026 - 05 - 02))
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "a clearing was reset twice");
    }
    if store
        .reset_clearing(
            crate::clearing::ClearingId::generate(),
            date!(2026 - 05 - 03),
        )
        .await
        .is_ok()
    {
        return CheckResult::fail(NAME, "an unknown clearing was reset");
    }

    let _ = key;
    CheckResult::pass(NAME)
}

/// Residuals reflect exactly what has been applied.
pub async fn check_open_items_track_residuals<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "open items track residuals";
    let f = Fixture::new();

    let Some(invoice) = f.entry::<P>(b"oi-invoice", 1000) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let invoice_id = invoice.id();
    if let Err(e) = store.append(&EntryBatch::single(invoice)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }
    let Some(payment) = f.reversed_entry::<P>(b"oi-payment", 250) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let payment_id = payment.id();
    if let Err(e) = store.append(&EntryBatch::single(payment)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    // Two more invoices, left open, with identifiers that **descend**. Without
    // them every entry here carries an `EntryId::generate()` value — UUIDv7, so
    // time-ordered — and sorting by identifier would coincide with log order,
    // letting an implementation that sorts by the wrong key pass by luck.
    let mut descending = Vec::new();
    for (n, high) in [0xeeee_u128, 0xdddd_u128].into_iter().enumerate() {
        let id = EntryId::from_uuid(uuid::Uuid::from_u128(high << 112));
        let Some(entry) = f.entry_with_id::<P>(id, format!("oi-descending-{n}").as_bytes(), 400)
        else {
            return CheckResult::fail(NAME, "fixture entry failed to seal");
        };
        if let Err(e) = store.append(&EntryBatch::single(entry)).await {
            return CheckResult::fail(NAME, format!("append failed: {e}"));
        }
        descending.push(PostingRef::new(id, 0));
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail(NAME, e);
    }
    let mut by_identifier = descending.clone();
    by_identifier.sort();
    if by_identifier == descending {
        return CheckResult::fail(
            NAME,
            "the fixture must distinguish identifier order from log order",
        );
    }

    // And one entry contributing *two* open postings on this account, so a page
    // boundary can fall inside an entry. Without it every open item sits on its
    // own entry and an entry-addressed cursor pages correctly by accident.
    let split_id = EntryId::generate();
    let split = Entry::<Draft, P>::new(
        split_id,
        match IdempotencyKey::new(b"oi-split".to_vec()) {
            Ok(k) => k,
            Err(e) => return CheckResult::fail(NAME, format!("bad fixture key: {e}")),
        },
        date!(2026 - 03 - 22),
    )
    .debit(f.left, Amount::<P>::from_minor(70), Currency::EUR)
    .debit(f.left, Amount::<P>::from_minor(80), Currency::EUR)
    .credit(f.right, Amount::<P>::from_minor(150), Currency::EUR)
    .seal(&f.ctx());
    match split {
        Ok(entry) => {
            if let Err(e) = store.append(&EntryBatch::single(entry)).await {
                return CheckResult::fail(NAME, format!("append failed: {e}"));
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("fixture entry failed: {e}")),
    }
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail(NAME, e);
    }

    let key = BalanceKey {
        account: f.left,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };
    let invoice_ref = PostingRef::new(invoice_id, 0);
    let payment_ref = PostingRef::new(payment_id, 0);

    let clearing = f.clearing::<P>(&[(invoice_ref, 250), (payment_ref, 250)]);
    let clearing_id = clearing.id;
    if let Err(e) = store.clear(clearing).await {
        return CheckResult::fail(NAME, format!("a valid clearing was rejected: {e}"));
    }

    let open = match store
        .open_items(key, PostingCursor::start().with_limit(MAX_PAGE_SIZE))
        .await
    {
        Ok(page) => page.items,
        Err(e) => return CheckResult::fail(NAME, format!("open_items failed: {e}")),
    };
    let Some(item) = open.iter().find(|i| i.posting == invoice_ref) else {
        return CheckResult::fail(NAME, "the partly-settled invoice is not open");
    };
    if item.applied != Amount::<P>::from_minor(250) {
        return CheckResult::fail(NAME, format!("applied is {}, expected 250", item.applied));
    }
    if item.residual != Amount::<P>::from_minor(750) {
        return CheckResult::fail(NAME, format!("residual is {}, expected 750", item.residual));
    }
    if open.iter().any(|i| i.posting == payment_ref) {
        return CheckResult::fail(NAME, "a fully applied posting is still open");
    }

    // Oldest first. Open items exist to be cleared, and FIFO — apply this
    // payment to the oldest open invoice — is the workflow; "oldest" is log
    // order, because that is the crate's answer to ordering everywhere else.
    // It must not be entry-*identifier* order: identifiers are caller-supplied,
    // so that ordering is chronological only when the caller happens to use
    // this crate's own time-ordered generator, and silently is not otherwise.
    let statement = match store
        .statement(
            key,
            BalanceQuery::all(),
            PostingCursor::start().with_limit(MAX_PAGE_SIZE),
        )
        .await
    {
        Ok(page) => page.lines,
        Err(e) => return CheckResult::fail(NAME, format!("statement failed: {e}")),
    };
    let log_rank = |reference: PostingRef| {
        statement
            .iter()
            .position(|l| l.posting == reference)
            .unwrap_or(usize::MAX)
    };
    let mut previous = 0usize;
    for item in &open {
        let rank = log_rank(item.posting);
        if rank < previous {
            return CheckResult::fail(
                NAME,
                format!(
                    "open items are not in log order; {} came too late",
                    item.posting
                ),
            );
        }
        previous = rank;
    }

    // The seeding above must actually have left two open items on one entry, or
    // paging is back to only ever splitting on entry boundaries.
    let splits_inside_an_entry = open
        .iter()
        .zip(open.iter().skip(1))
        .any(|(a, b)| a.position.index == b.position.index);
    if !splits_inside_an_entry {
        return CheckResult::fail(
            NAME,
            "no entry left two open items, so page boundaries never fell inside one",
        );
    }

    // Paging must reproduce the one-page answer exactly, at every page size.
    for limit in 1..=3usize {
        let mut paged = Vec::new();
        let mut cursor = Some(PostingCursor::start().with_limit(limit));
        let mut guard = 0usize;
        while let Some(c) = cursor {
            guard = guard.saturating_add(1);
            if guard > open.len().saturating_add(8) {
                return CheckResult::fail(NAME, "open-item pagination did not terminate");
            }
            match store.open_items(key, c).await {
                Ok(page) => {
                    if page.items.is_empty() && page.next.is_some() {
                        return CheckResult::fail(NAME, "an empty page handed back another cursor");
                    }
                    paged.extend(page.items);
                    cursor = page.next;
                }
                Err(e) => return CheckResult::fail(NAME, format!("open_items failed: {e}")),
            }
        }
        if paged != open {
            return CheckResult::fail(
                NAME,
                format!(
                    "paging open items at limit {limit} produced {} of {}",
                    paged.len(),
                    open.len()
                ),
            );
        }
    }

    // Releasing reopens both.
    if let Err(e) = store
        .reset_clearing(clearing_id, date!(2026 - 05 - 10))
        .await
    {
        return CheckResult::fail(NAME, format!("reset failed: {e}"));
    }
    match store
        .open_items(key, PostingCursor::start().with_limit(MAX_PAGE_SIZE))
        .await
    {
        Ok(page) => {
            let after = page.items;
            let invoice_back = after
                .iter()
                .any(|i| i.posting == invoice_ref && i.residual == Amount::<P>::from_minor(1000));
            let payment_back = after.iter().any(|i| i.posting == payment_ref);
            if invoice_back && payment_back {
                CheckResult::pass(NAME)
            } else {
                CheckResult::fail(NAME, "a reset did not reopen both items")
            }
        }
        Err(e) => CheckResult::fail(NAME, format!("open_items failed: {e}")),
    }
}

/// A fresh store holds nothing and commits to the empty root.
pub async fn check_starts_empty<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "starts empty";
    match store.len().await {
        Ok(0) => {}
        Ok(n) => return CheckResult::fail(NAME, format!("expected an empty store, found {n}")),
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    }
    match store.head().await {
        Ok(head) if head.size == 0 && head.root == crate::merkle::empty_root() => {
            CheckResult::pass(NAME)
        }
        Ok(head) => CheckResult::fail(NAME, format!("empty store has head {head:?}")),
        Err(e) => CheckResult::fail(NAME, format!("head failed: {e}")),
    }
}

/// Indices start at zero and increase by one, in append order.
pub async fn check_append_assigns_dense_indices<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "append assigns dense, ordered indices";
    let f = Fixture::new();
    let start = match store.len().await {
        Ok(n) => n,
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    };

    for i in 0..3u64 {
        let key = format!("dense-{i}");
        let Some(entry) = f.entry::<P>(key.as_bytes(), 100) else {
            return CheckResult::fail(NAME, "fixture entry failed to seal");
        };
        let recorded = match store.append(&EntryBatch::single(entry)).await {
            Ok(r) => r,
            Err(e) => return CheckResult::fail(NAME, format!("append failed: {e}")),
        };
        let Some(first) = recorded.first() else {
            return CheckResult::fail(NAME, "append returned no outcome");
        };
        if !first.is_new {
            return CheckResult::fail(NAME, "a fresh key was reported as a replay");
        }

        // A deferred backend has no position yet; sequencing must produce one.
        let placed = if first.index.is_some() {
            first.index
        } else {
            match sequence_until_placed(store, first.id).await {
                Ok(index) => index,
                Err(e) => return CheckResult::fail(NAME, e),
            }
        };

        let expected = start.saturating_add(i);
        match placed {
            Some(index) if index.get() == expected => {}
            Some(index) => {
                return CheckResult::fail(
                    NAME,
                    format!("expected index {expected}, got {}", index.get()),
                );
            }
            None => {
                return CheckResult::fail(NAME, "an entry has no position after sequencing");
            }
        }
    }
    CheckResult::pass(NAME)
}

/// The same record read twice is identical, by identifier and by page.
pub async fn check_reads_are_stable<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "reads are stable";
    let f = Fixture::new();
    let Some(entry) = f.entry::<P>(b"stable", 4242) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let id = entry.id();
    let expected_hash = entry.content_hash();

    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    let first = match store.get(id).await {
        Ok(Some(r)) => r,
        Ok(None) => return CheckResult::fail(NAME, "an appended entry was not found"),
        Err(e) => return CheckResult::fail(NAME, format!("get failed: {e}")),
    };
    if first.content_hash != expected_hash {
        return CheckResult::fail(NAME, "stored content hash differs from the entry's own");
    }

    match store.get(id).await {
        Ok(Some(second)) if second == first => CheckResult::pass(NAME),
        Ok(Some(_)) => CheckResult::fail(NAME, "two reads of one record disagreed"),
        Ok(None) => CheckResult::fail(NAME, "a record disappeared between reads"),
        Err(e) => CheckResult::fail(NAME, format!("get failed: {e}")),
    }
}

/// Re-appending identical content under the same key changes nothing.
pub async fn check_idempotent_replay<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "replay is idempotent";
    let f = Fixture::new();
    let Some(first) = f.entry::<P>(b"replay", 777) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let original = match store.append(&EntryBatch::single(first)).await {
        Ok(r) => match r.first().copied() {
            Some(r) => r,
            None => return CheckResult::fail(NAME, "append returned no outcome"),
        },
        Err(e) => return CheckResult::fail(NAME, format!("append failed: {e}")),
    };

    let before = match store.len().await {
        Ok(n) => n,
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    };

    // A different entry identifier, but the same logical transaction.
    let Some(again) = f.entry::<P>(b"replay", 777) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    match store.append(&EntryBatch::single(again)).await {
        Ok(r) => {
            let Some(replay) = r.first() else {
                return CheckResult::fail(NAME, "append returned no outcome");
            };
            if replay.is_new {
                return CheckResult::fail(NAME, "a replay was reported as a new entry");
            }
            if replay.index != original.index || replay.id != original.id {
                return CheckResult::fail(NAME, "a replay returned a different record");
            }
        }
        Err(e) => return CheckResult::fail(NAME, format!("a safe replay was rejected: {e}")),
    }

    match store.len().await {
        Ok(after) if after == before => CheckResult::pass(NAME),
        Ok(after) => CheckResult::fail(
            NAME,
            format!("a replay appended: {before} entries became {after}"),
        ),
        Err(e) => CheckResult::fail(NAME, format!("len failed: {e}")),
    }
}

/// An entry can be found by the key it was recorded under.
///
/// The lookup the idempotency key exists for. A backend that stores the key only
/// to enforce uniqueness makes the callers derive one from a source transaction
/// and then gives them no way to ask about it.
pub async fn check_lookup_by_idempotency_key<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    const NAME: &str = "entries are findable by key";
    let f = Fixture::new();
    let Ok(key) = IdempotencyKey::new(b"findable".to_vec()) else {
        return CheckResult::fail(NAME, "bad fixture key");
    };

    // Absent before it is written — and absent is `None`, not an error.
    match store.get_by_key(&key).await {
        Ok(None) => {}
        Ok(Some(_)) => return CheckResult::fail(NAME, "an unwritten key was found"),
        Err(e) => return CheckResult::fail(NAME, format!("get_by_key failed: {e}")),
    }

    let Some(entry) = f.entry::<P>(b"findable", 4242) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let expected = entry.id();
    let hash = entry.content_hash();
    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    match store.get_by_key(&key).await {
        Ok(Some(found)) => {
            if found.entry.id() != expected {
                return CheckResult::fail(NAME, "the key returned a different entry");
            }
            if found.content_hash != hash {
                return CheckResult::fail(NAME, "the key returned a different content hash");
            }
            // And it agrees with the identifier-keyed lookup, including its
            // position — two lookups of one row must not disagree.
            match store.get(expected).await {
                Ok(Some(by_id)) if by_id.index == found.index => CheckResult::pass(NAME),
                Ok(Some(_)) => CheckResult::fail(NAME, "the two lookups disagree on the position"),
                Ok(None) => CheckResult::fail(NAME, "the entry is findable by key but not by id"),
                Err(e) => CheckResult::fail(NAME, format!("get failed: {e}")),
            }
        }
        Ok(None) => CheckResult::fail(NAME, "a recorded key was not found"),
        Err(e) => CheckResult::fail(NAME, format!("get_by_key failed: {e}")),
    }
}

/// The same key with different content is refused, not overwritten.
pub async fn check_idempotency_conflict<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "conflicting key is refused";
    let f = Fixture::new();
    let Some(first) = f.entry::<P>(b"conflict", 100) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(first)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }
    let before = match store.len().await {
        Ok(n) => n,
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    };

    let Some(different) = f.entry::<P>(b"conflict", 999) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if store.append(&EntryBatch::single(different)).await.is_ok() {
        return CheckResult::fail(NAME, "a conflicting key was accepted");
    }

    match store.len().await {
        Ok(after) if after == before => CheckResult::pass(NAME),
        Ok(after) => CheckResult::fail(
            NAME,
            format!("a refused append still wrote: {before} became {after}"),
        ),
        Err(e) => CheckResult::fail(NAME, format!("len failed: {e}")),
    }
}

/// A batch that cannot land in full lands not at all.
pub async fn check_batch_is_atomic<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    const NAME: &str = "batches are atomic";
    let f = Fixture::new();

    // Poison the batch: the second entry reuses the first's key with different
    // content, so the batch must be refused as a whole.
    let Some(good) = f.entry::<P>(b"atomic-a", 100) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let Some(poison_first) = f.entry::<P>(b"atomic-b", 200) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let Some(poison_second) = f.entry::<P>(b"atomic-b", 300) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    let good_id = good.id();

    let Ok(batch) = EntryBatch::new(vec![good, poison_first, poison_second]) else {
        return CheckResult::fail(NAME, "batch construction failed");
    };

    let before = match store.len().await {
        Ok(n) => n,
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    };

    if store.append(&batch).await.is_ok() {
        return CheckResult::fail(NAME, "a batch with a conflicting entry was accepted");
    }

    match store.len().await {
        Ok(after) if after != before => {
            return CheckResult::fail(
                NAME,
                format!("a refused batch partly landed: {before} became {after}"),
            );
        }
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
        Ok(_) => {}
    }

    match store.get(good_id).await {
        Ok(None) => CheckResult::pass(NAME),
        Ok(Some(_)) => CheckResult::fail(NAME, "the valid part of a refused batch was kept"),
        Err(e) => CheckResult::fail(NAME, format!("get failed: {e}")),
    }
}

/// Paging visits every record exactly once, in order.
pub async fn check_pagination_covers_the_log<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "pagination covers the log";
    let expected = match store.len().await {
        Ok(n) => n,
        Err(e) => return CheckResult::fail(NAME, format!("len failed: {e}")),
    };

    let mut seen: Vec<u64> = Vec::new();
    let mut cursor = Some(Cursor::start().with_limit(2));
    let mut guard = 0u32;

    while let Some(c) = cursor {
        guard = guard.saturating_add(1);
        if u64::from(guard) > expected.saturating_add(16) {
            return CheckResult::fail(NAME, "pagination did not terminate");
        }
        match store.page(c).await {
            Ok(page) => {
                for record in &page.records {
                    match record.require_index() {
                        Ok(index) => seen.push(index.get()),
                        Err(e) => {
                            return CheckResult::fail(
                                NAME,
                                format!("a page returned an unsequenced entry: {e}"),
                            );
                        }
                    }
                }
                cursor = page.next;
            }
            Err(e) => return CheckResult::fail(NAME, format!("page failed: {e}")),
        }
    }

    let wanted: Vec<u64> = (0..expected).collect();
    if seen == wanted {
        CheckResult::pass(NAME)
    } else {
        CheckResult::fail(NAME, format!("expected indices {wanted:?}, paged {seen:?}"))
    }
}

/// Balances agree with a fold over the paged log.
pub async fn check_balances_match_the_log<const P: u8, S: LedgerStore<P>>(
    store: &S,
) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "balances match the log";
    let f = Fixture::new();
    let key = BalanceKey {
        account: f.left,
        currency: Currency::EUR,
        layer: Layer::Settled,
    };

    let reported = match store.balance(key, BalanceQuery::all()).await {
        Ok(b) => b,
        Err(e) => return CheckResult::fail(NAME, format!("balance failed: {e}")),
    };

    // Recompute from the log itself.
    let mut folded = crate::balance::Balance::<P>::ZERO;
    let mut cursor = Some(Cursor::start());
    while let Some(c) = cursor {
        match store.page(c).await {
            Ok(page) => {
                for record in &page.records {
                    for posting in record.entry.postings() {
                        if posting.account == key.account
                            && posting.currency == key.currency
                            && posting.layer == key.layer
                            && folded.add(posting.direction, posting.amount).is_err()
                        {
                            return CheckResult::fail(NAME, "folding the log overflowed");
                        }
                    }
                }
                cursor = page.next;
            }
            Err(e) => return CheckResult::fail(NAME, format!("page failed: {e}")),
        }
    }

    if folded == reported {
        CheckResult::pass(NAME)
    } else {
        CheckResult::fail(
            NAME,
            format!("store reported {reported:?}, the log folds to {folded:?}"),
        )
    }
}

/// Every record is provably included, and growth is provably append-only.
pub async fn check_proofs_verify<const P: u8, S: LedgerStore<P>>(store: &S) -> CheckResult {
    if let Err(e) = drain_sequencing(store).await {
        return CheckResult::fail("sequencing", e);
    }
    const NAME: &str = "proofs verify";
    let head = match store.head().await {
        Ok(h) => h,
        Err(e) => return CheckResult::fail(NAME, format!("head failed: {e}")),
    };
    if head.size == 0 {
        return CheckResult::fail(NAME, "no entries to prove");
    }

    // Kept for the archived-head checks below, which need an entry that was
    // already in the log when this head was published.
    let mut archived_entry = None;
    let mut leaves = Vec::new();
    let mut cursor = Some(Cursor::start());
    while let Some(c) = cursor {
        match store.page(c).await {
            Ok(page) => {
                for record in &page.records {
                    let Ok(index) = record.require_index() else {
                        return CheckResult::fail(NAME, "a page returned an unsequenced entry");
                    };
                    if archived_entry.is_none() {
                        archived_entry = Some((index, record.content_hash));
                    }
                    leaves.push(record.content_hash);
                    match store.prove_inclusion(index).await {
                        Ok(proof) => {
                            if !proof.verify(&record.content_hash, &head) {
                                return CheckResult::fail(
                                    NAME,
                                    format!("inclusion proof failed for index {index}"),
                                );
                            }
                        }
                        Err(e) => {
                            return CheckResult::fail(NAME, format!("prove_inclusion failed: {e}"));
                        }
                    }
                }
                cursor = page.next;
            }
            Err(e) => return CheckResult::fail(NAME, format!("page failed: {e}")),
        }
    }

    // Append one more, then prove the earlier head was a prefix.
    let f = Fixture::new();
    let Some(entry) = f.entry::<P>(b"proof-growth", 55) else {
        return CheckResult::fail(NAME, "fixture entry failed to seal");
    };
    if let Err(e) = store.append(&EntryBatch::single(entry)).await {
        return CheckResult::fail(NAME, format!("append failed: {e}"));
    }

    let grown = match store.head().await {
        Ok(h) => h,
        Err(e) => return CheckResult::fail(NAME, format!("head failed: {e}")),
    };
    match store.prove_consistency(head.size).await {
        Ok(proof) if proof.verify(&head, &grown) => {}
        Ok(_) => return CheckResult::fail(NAME, "consistency proof did not verify"),
        Err(e) => return CheckResult::fail(NAME, format!("prove_consistency failed: {e}")),
    }

    // The archived-head case: an auditor holding an earlier head must be able to
    // check it, so every historical head the store reports has to equal the one
    // a log built from the same prefix produces. A backend that answers this
    // from a stored column rather than a replay is exactly where the two can
    // silently drift apart.
    for size in 0..=leaves.len() {
        let Some(prefix) = leaves.get(..size) else {
            return CheckResult::fail(NAME, "prefix beyond the collected log");
        };
        let expected = MerkleLog::from_leaves(prefix.to_vec()).head();
        match store.head_at(expected.size).await {
            Ok(recalled) if recalled == expected => {}
            Ok(other) => {
                return CheckResult::fail(
                    NAME,
                    format!("head_at({size}) returned {other:?}, not the head as it stood"),
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("head_at({size}) failed: {e}")),
        }
    }

    if let Some((index, content_hash)) = archived_entry {
        // Against the archived head, not the current one — the log has grown a
        // row since, so a proof built against `grown` would not verify here.
        match store.prove_inclusion_at(index, head.size).await {
            Ok(proof) if proof.verify(&content_hash, &head) => {}
            Ok(_) => {
                return CheckResult::fail(
                    NAME,
                    "an inclusion proof against the archived head did not verify",
                );
            }
            Err(e) => return CheckResult::fail(NAME, format!("prove_inclusion_at failed: {e}")),
        }
    }

    match store.prove_consistency_between(head.size, grown.size).await {
        Ok(proof) if proof.verify(&head, &grown) => CheckResult::pass(NAME),
        Ok(_) => CheckResult::fail(NAME, "a proof between two archived heads did not verify"),
        Err(e) => CheckResult::fail(NAME, format!("prove_consistency_between failed: {e}")),
    }
}

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

    #[test]
    fn the_memory_store_conforms() {
        let report = block_on(check_all(&MemoryStore::<2>::new(test_ledger())));
        report.assert_passed();
        // Pinned so a check cannot be dropped from `check_all` unnoticed: a
        // suite that silently shrinks still passes.
        assert_eq!(report.checks.len(), 25);
    }

    #[test]
    fn block_on_drives_a_future_to_completion() {
        assert_eq!(block_on(async { 1 + 1 }), 2);
    }

    #[test]
    fn a_report_renders_its_failures() {
        let report = Report {
            checks: vec![
                CheckResult::pass("fine"),
                CheckResult::fail("broken", "it did the wrong thing"),
            ],
        };
        assert!(!report.passed());
        assert_eq!(report.failures().len(), 1);
        let rendered = report.to_string();
        assert!(rendered.contains("broken"), "{rendered}");
        assert!(rendered.contains("it did the wrong thing"), "{rendered}");
    }

    #[test]
    fn a_passing_report_says_so() {
        let report = Report {
            checks: vec![CheckResult::pass("fine")],
        };
        assert_eq!(report.to_string(), "all 1 conformance checks passed");
    }
}