agentplane 0.26.0

//! What every journal store must do, checked against any implementation.
//!
//! # Why this exists rather than a second set of tests
//!
//! [`JournalStore`] states three guarantees and says they must hold
//! *atomically*: fencing, exactly-once, and chaining. They are not application
//! logic — they are storage invariants, deliberately, because application logic
//! can be bypassed by the next caller and a constraint cannot.
//!
//! A second backend is exactly where that stops being true. The embedded store
//! encodes all three; a Postgres store written from the same prose will encode
//! two of them and something *nearly* the third, and nothing will notice,
//! because the suite that proves the runtime correct runs against the embedded
//! store. The
//! new backend gets whatever tests its author remembered to write, and those
//! will be the ones they were already thinking about.
//!
//! So the invariants are written once, here, and every implementation is run
//! against the same battery. An embedder bringing their own store gets the same
//! benefit, which is why this ships rather than living in `tests/`.
//!
//! # It reports every failure, not the first
//!
//! Bringing up a backend is iterative. A battery that stops at the first
//! problem turns that into a sequence of full rebuilds, and — worse — hides
//! whether the second failure is a separate bug or a consequence of the first.

use std::sync::Arc;
use std::time::Duration;

use crate::core::{CaseId, Digest, EffectKey, RunId, StepId};
use crate::journal::{Append, JournalStore, Record, RecordKind};

/// What a store got wrong.
#[derive(Debug, Clone)]
pub struct Violation {
    /// The guarantee, in the words [`JournalStore`] uses.
    pub invariant: &'static str,
    /// What happened instead, and why it matters.
    pub detail: String,
}

/// The result of running the battery.
#[derive(Debug, Default)]
pub struct Report {
    pub violations: Vec<Violation>,
    pub checked: usize,
}

impl Report {
    pub(crate) fn record(&mut self, invariant: &'static str, detail: impl Into<String>) {
        self.violations.push(Violation {
            invariant,
            detail: detail.into(),
        });
    }

    /// Fail the test if anything was violated.
    ///
    /// # Panics
    ///
    /// If any invariant was violated, or if the battery somehow checked
    /// nothing — a battery that silently ran zero checks is worse than none,
    /// because it reports success.
    pub fn assert_conforms(&self, store: &str) {
        assert!(
            self.checked > 0,
            "the conformance battery ran no checks against {store}"
        );
        assert!(
            self.violations.is_empty(),
            "{store} does not satisfy its store contract:\n{}",
            self.violations
                .iter()
                .map(|v| format!("  • {}: {}", v.invariant, v.detail))
                .collect::<Vec<_>>()
                .join("\n")
        );
    }
}

/// Produces a store with no history in it.
///
/// A factory rather than one store, because several checks need a clean chain
/// and sharing one would let an earlier check's writes decide a later one's
/// result.
pub type Factory<'a> =
    &'a (dyn Fn() -> std::pin::Pin<Box<dyn Future<Output = Arc<dyn JournalStore>> + Send>> + Sync);

const LEASE: Duration = Duration::from_mins(5);
/// The shortest lease a store is obliged to honour.
///
/// A zero TTL is not usable: the embedded store clamps it to a second, and any backend is
/// free to, because a lease that expires the instant it is granted is not a
/// lease. So the fencing check takes the short one and *waits* — the only way to
/// reach an expired lease through the public API, and therefore the only way a
/// battery that must work against an unknown backend can get there.
const SHORT: Duration = Duration::from_secs(1);

/// Exercise the storage-level governed-memory contract against any backend.
///
/// Backends remain free to choose indexes and transaction mechanisms, but they
/// do not get to choose different semantics for versions, scopes, lineage, or
/// erasure.
///
/// # Panics
///
/// Panics when the backend violates the contract.
#[allow(clippy::too_many_lines)]
pub async fn memory(store: Arc<dyn crate::memory::MemoryStore>) {
    use crate::core::{Sensitivity, SourceId, Timestamp, Trust};
    use crate::memory::{MemoryItem, Recall, Selected};
    use serde_json::json;

    let at = |seconds| Timestamp::from_unix_timestamp(seconds).expect("representable time");
    let make = |id: &str, subject: &str, purpose: &str, content: serde_json::Value| MemoryItem {
        id: id.to_owned(),
        subject: subject.to_owned(),
        purpose: purpose.to_owned(),
        content,
        provenance: vec![SourceId::new("conformance")],
        sensitivity: Sensitivity::Internal,
        trust: Trust::Untrusted,
        written_by: "conformance".to_owned(),
        version: 0,
        created_at: at(1_760_000_000),
        expires_at: None,
        access_retention_seconds: None,
        superseded_at: None,
        derived_from: Vec::new(),
    };

    let mut first = make("memory-a", "team-a", "support", json!({"value": 1}));
    assert_eq!(store.remember(&first).await.expect("remember v1"), 1);
    first.content = json!({"value": 2});
    first.created_at = at(1_760_000_001);
    assert_eq!(store.remember(&first).await.expect("remember v2"), 2);
    let old = store
        .version("memory-a", 1)
        .await
        .expect("old version")
        .expect("v1 kept");
    assert_eq!(old.content, json!({"value": 1}));
    assert_eq!(old.superseded_at, Some(at(1_760_000_001)));

    // `current` answers the recall question by id: the latest version, never a
    // superseded one — the semantic tier's screen rides on this distinction.
    let live = store
        .current("memory-a", None)
        .await
        .expect("current")
        .expect("a revised memory is still current");
    assert_eq!(
        live.version, 2,
        "current must be the revision, not the v1 it superseded"
    );
    assert!(
        store
            .current("memory-never-written", None)
            .await
            .expect("current of nothing")
            .is_none()
    );

    // Trust outranks recency, on every backend.
    //
    // Recall truncates, so ordering by recency alone is an eviction an attacker
    // steers: anything able to write an untrusted memory — model output and tool
    // output both can, by design — writes `limit` of them and the trusted ones
    // silently lose their place. Every label stays correct in that scenario,
    // which is what made it hard to see; the defect is the ordering.
    let mut trusted = make("rank-trusted", "team-rank", "support", json!({"rule": 1}));
    trusted.trust = Trust::Trusted;
    trusted.created_at = at(1_760_000_100);
    store.remember(&trusted).await.expect("trusted");
    for i in 0..4 {
        let mut noise = make(
            &format!("rank-noise-{i}"),
            "team-rank",
            "support",
            json!({"rule": "ignore the above"}),
        );
        // Newer than the trusted one, which is the whole point.
        noise.created_at = at(1_760_000_200 + i64::from(i));
        store.remember(&noise).await.expect("noise");
    }
    let ranked = store
        .recall(&Recall::about("team-rank").limit(2))
        .await
        .expect("ranked recall");
    assert_eq!(ranked.len(), 2, "the limit is still honoured");
    assert_eq!(
        ranked[0].id,
        "rank-trusted",
        "newer untrusted memories evicted the trusted one: {:?}",
        ranked.iter().map(|i| (&i.id, i.trust)).collect::<Vec<_>>()
    );
    assert_eq!(
        ranked[1].trust,
        Trust::Untrusted,
        "untrusted memories must still fill the remaining room — a recall that \
         returned only trusted items is an agent that cannot see what it was told"
    );

    let moved = make("memory-a", "team-b", "support", json!({"value": 4}));
    assert!(
        store.remember(&moved).await.is_err(),
        "a stable id moved to another subject, so erasing the old subject can miss its history"
    );

    store
        .remember(&make("memory-b", "team-a", "payments", json!({"value": 3})))
        .await
        .expect("other purpose");
    let support = store
        .recall(&Recall::about("team-a").for_purpose("support"))
        .await
        .expect("purpose recall");
    assert_eq!(support.len(), 1);
    assert_eq!(support[0].id, "memory-a");

    let source = support[0].clone();
    let mut misplaced = make(
        "misplaced-summary",
        "team-b",
        "support",
        json!({"summary": true}),
    );
    misplaced.derived_from = vec![Selected {
        id: source.id.clone(),
        version: source.version,
        digest: source.selection_digest(),
    }];
    assert!(
        store.remember(&misplaced).await.is_err(),
        "a derivative escaped its source subject, so subject erasure cannot reach it"
    );

    let mut derived = make("summary", "team-a", "support", json!({"summary": true}));
    derived.derived_from = vec![Selected {
        id: source.id.clone(),
        version: source.version,
        digest: source.selection_digest(),
    }];
    store.remember(&derived).await.expect("derived");
    assert_eq!(
        store
            .derivatives("memory-a")
            .await
            .expect("derivatives")
            .iter()
            .map(|item| item.id.as_str())
            .collect::<Vec<_>>(),
        vec!["summary"]
    );

    assert_eq!(
        store.forget_cascading("memory-a").await.expect("cascade"),
        2
    );
    assert!(
        store
            .version("memory-a", 1)
            .await
            .expect("forgotten source")
            .is_none()
    );
    assert!(
        store
            .version("summary", 1)
            .await
            .expect("forgotten summary")
            .is_none()
    );
    // A cascade must pass **through** a tombstone. `forget` keeps a forgotten
    // memory's outgoing edges precisely so that a correction which later
    // becomes an erasure request can still find what was derived — and both
    // backends' traversals skipped a node with no current entry, which
    // defeated that provision exactly when it was needed: A → B → C with B
    // individually erased left C standing after a cascade from A. One-level
    // cascades cannot see this, which is why the chain here is three deep with
    // the middle link erased first.
    store
        .remember(&make("chain-a", "team-chain", "support", json!({"n": 1})))
        .await
        .expect("chain root");
    let chain_a = store
        .version("chain-a", 1)
        .await
        .expect("read root")
        .expect("root exists");
    let mut chain_b = make("chain-b", "team-chain", "support", json!({"n": 2}));
    chain_b.derived_from = vec![Selected {
        id: chain_a.id.clone(),
        version: chain_a.version,
        digest: chain_a.selection_digest(),
    }];
    store.remember(&chain_b).await.expect("chain middle");
    let chain_b = store
        .version("chain-b", 1)
        .await
        .expect("read middle")
        .expect("middle exists");
    let mut chain_c = make("chain-c", "team-chain", "support", json!({"n": 3}));
    chain_c.derived_from = vec![Selected {
        id: chain_b.id.clone(),
        version: chain_b.version,
        digest: chain_b.selection_digest(),
    }];
    store.remember(&chain_c).await.expect("chain leaf");

    store.forget("chain-b").await.expect("erase the middle");
    assert_eq!(
        store
            .forget_cascading("chain-a")
            .await
            .expect("cascade through the tombstone"),
        2,
        "the cascade must erase the root and the leaf — and count only what it \
         erased, not the tombstone it passed through"
    );
    assert!(
        store
            .version("chain-c", 1)
            .await
            .expect("leaf lookup")
            .is_none(),
        "a derivative reached only through an already-erased intermediate \
         survived the cascade — the poisoned source's summary of a summary is \
         still readable"
    );

    // The count answers an erasure *request*, so it may only report what this
    // call removed. Cascading from an id that is already a tombstone removes
    // nothing: the traversal has a root to start from and no state to destroy,
    // and answering `1` would tell a caller — a regulator, an erasure ledger —
    // that a memory was destroyed by a call that found nothing there. The
    // tombstone case above cannot see this: an already-erased *intermediate*
    // is reached version-wise and never counted either way, so the guard that
    // keeps the count honest is only exercised when the stateless node is the
    // one the request names.
    assert_eq!(
        store
            .forget_cascading("chain-a")
            .await
            .expect("cascading a second time is not an error"),
        0,
        "a cascade over an already-erased id reported an erasure it did not \
         perform — the count is what an erasure request is answered with"
    );

    // The same routing obligation for the *other* way a memory dies: expiry.
    // `sweep_expired` erases state, and it must leave the derivation edges in
    // both directions exactly as `forget` does — with U → E → D and E expired,
    // a later cascade from poisoned U still has to route through E's tombstone
    // to reach D. The sweep once deleted E's incoming edges, which severed
    // exactly that path; the cascade then stopped at the tombstone and the
    // summary-of-a-summary outlived the erasure.
    store
        .remember(&make(
            "exp-u",
            "team-expiry-chain",
            "support",
            json!({"n": 1}),
        ))
        .await
        .expect("expiry-chain root");
    let exp_u = store
        .version("exp-u", 1)
        .await
        .expect("read root")
        .expect("root exists");
    let mut exp_e = make("exp-e", "team-expiry-chain", "support", json!({"n": 2}));
    exp_e.expires_at = Some(at(1_760_000_150));
    exp_e.derived_from = vec![Selected {
        id: exp_u.id.clone(),
        version: exp_u.version,
        digest: exp_u.selection_digest(),
    }];
    store.remember(&exp_e).await.expect("expiring middle");
    let exp_e = store
        .version("exp-e", 1)
        .await
        .expect("read middle")
        .expect("middle exists");
    let mut exp_d = make("exp-d", "team-expiry-chain", "support", json!({"n": 3}));
    exp_d.derived_from = vec![Selected {
        id: exp_e.id.clone(),
        version: exp_e.version,
        digest: exp_e.selection_digest(),
    }];
    store.remember(&exp_d).await.expect("expiry-chain leaf");

    assert_eq!(
        store
            .sweep_expired(at(1_760_000_150))
            .await
            .expect("expire the middle"),
        1,
        "exactly the middle link expires"
    );
    assert_eq!(
        store
            .forget_cascading("exp-u")
            .await
            .expect("cascade through the expired tombstone"),
        2,
        "the cascade must erase the root and the leaf, counting only what it \
         erased — not the expired tombstone it passed through"
    );
    assert!(
        store
            .version("exp-d", 1)
            .await
            .expect("leaf lookup")
            .is_none(),
        "a derivative reached only through an *expired* intermediate survived \
         the cascade — the expiry sweep severed the lineage that a later \
         erasure of the poisoned source needed to route through"
    );

    assert_eq!(
        store
            .forget_subject("team-a")
            .await
            .expect("subject erasure"),
        1
    );
    assert!(
        store
            .recall(&Recall::about("team-a"))
            .await
            .expect("empty")
            .is_empty()
    );

    let mut expiring = make(
        "memory-expiring",
        "team-lifecycle",
        "support",
        json!({"temporary": true}),
    );
    expiring.expires_at = Some(at(1_760_000_100));
    store.remember(&expiring).await.expect("expiring memory");
    assert_eq!(
        store
            .recall(&Recall::about("team-lifecycle").at(at(1_760_000_099)))
            .await
            .expect("before expiry")
            .len(),
        1
    );
    assert!(
        store
            .recall(&Recall::about("team-lifecycle").at(at(1_760_000_100)))
            .await
            .expect("at expiry")
            .is_empty(),
        "expiry is inclusive"
    );
    assert!(
        store
            .version("memory-expiring", 1)
            .await
            .expect("exact replay version")
            .is_some(),
        "hiding an expired current item must not silently break replay"
    );
    // `current` applies the same inclusive cutoff recall does, and skips the
    // check entirely when no cutoff is given — the `version`/`current` split
    // is exactly replay-truth versus recall-truth.
    assert!(
        store
            .current("memory-expiring", Some(at(1_760_000_099)))
            .await
            .expect("current before expiry")
            .is_some()
    );
    assert!(
        store
            .current("memory-expiring", Some(at(1_760_000_100)))
            .await
            .expect("current at expiry")
            .is_none(),
        "current must apply the inclusive expiry cutoff"
    );
    assert!(
        store
            .current("memory-expiring", None)
            .await
            .expect("current without cutoff")
            .is_some(),
        "no cutoff means no expiry check, exactly as on recall"
    );

    store
        .set_legal_hold("memory-expiring", true)
        .await
        .expect("place legal hold");
    assert!(
        store
            .legal_hold("memory-expiring")
            .await
            .expect("read hold")
    );
    assert!(
        store.forget("memory-expiring").await.is_err(),
        "ordinary erasure bypassed legal hold"
    );
    assert_eq!(
        store
            .sweep_expired(at(1_760_000_101))
            .await
            .expect("held sweep"),
        0,
        "expiry sweep bypassed legal hold"
    );
    store
        .set_legal_hold("memory-expiring", false)
        .await
        .expect("release legal hold");
    assert_eq!(
        store
            .sweep_expired(at(1_760_000_101))
            .await
            .expect("expiry sweep"),
        1
    );
    assert!(
        store
            .version("memory-expiring", 1)
            .await
            .expect("swept version")
            .is_none()
    );
    assert!(
        store
            .current("memory-expiring", None)
            .await
            .expect("current after sweep")
            .is_none(),
        "a swept id has no current version under any cutoff"
    );

    // Sliding retention, alone: the window opens at the write. Initialized
    // lazily instead, an item with a window and no fixed expiry was immortal
    // until its first touch — opt-in garbage that never collects.
    let mut fading = make(
        "memory-fading",
        "team-fading",
        "support",
        json!({"fading": true}),
    );
    fading.access_retention_seconds = Some(60);
    store.remember(&fading).await.expect("fading memory");
    assert_eq!(
        store
            .recall(&Recall::about("team-fading").at(at(1_760_000_059)))
            .await
            .expect("inside the untouched window")
            .len(),
        1
    );
    assert!(
        store
            .recall(&Recall::about("team-fading").at(at(1_760_000_060)))
            .await
            .expect("after the untouched window")
            .is_empty(),
        "an untouched sliding-retention memory outlived its own window — the \
         window must open at the write, or an item nobody ever touches is \
         immortal"
    );
    assert_eq!(
        store
            .sweep_expired(at(1_760_000_060))
            .await
            .expect("sweep the untouched window"),
        1,
        "the sweep must collect an untouched sliding-retention memory"
    );

    // Sliding retention under a hard ceiling: the touch slides the window,
    // and the ceiling is immutable — the effective expiry is whichever comes
    // first, never a touched-up instant past `expires_at`.
    let mut sliding = make(
        "memory-sliding",
        "team-retention",
        "support",
        json!({"sliding": true}),
    );
    sliding.expires_at = Some(at(1_760_000_200));
    sliding.access_retention_seconds = Some(60);
    store.remember(&sliding).await.expect("sliding memory");
    store
        .touch(&["memory-sliding".to_owned()], at(1_760_000_040))
        .await
        .expect("touch sliding retention");
    assert_eq!(
        store
            .recall(&Recall::about("team-retention").at(at(1_760_000_099)))
            .await
            .expect("inside the touched window")
            .len(),
        1,
        "a journaled touch did not slide the window"
    );
    assert!(
        store
            .current("memory-sliding", Some(at(1_760_000_100)))
            .await
            .expect("current after the touched window")
            .is_none(),
        "current must honour the sliding window exactly as recall does"
    );
    store
        .touch(&["memory-sliding".to_owned()], at(1_760_000_190))
        .await
        .expect("touch near the ceiling");
    assert_eq!(
        store
            .recall(&Recall::about("team-retention").at(at(1_760_000_199)))
            .await
            .expect("just under the ceiling")
            .len(),
        1
    );
    assert!(
        store
            .recall(&Recall::about("team-retention").at(at(1_760_000_200)))
            .await
            .expect("at the ceiling")
            .is_empty(),
        "a touch extended a memory past its immutable `expires_at` — the \
         ceiling is the disposal date an operator can state, and sliding \
         retention may only shorten a life, never lengthen it"
    );
    assert_eq!(
        store
            .sweep_expired(at(1_760_000_200))
            .await
            .expect("sweep at the ceiling"),
        1,
        "the sweep must erase at the ceiling even though the touched window \
         reaches past it"
    );
    assert!(
        store
            .version("memory-sliding", 1)
            .await
            .expect("swept sliding version")
            .is_none()
    );

    // A lapsed-but-unswept window is resurrected by a touch, on every backend.
    // Expiry becomes fact at the sweep, not at the instant the window closes:
    // stores take max(existing, at + window), so a late touch opens a new
    // window rather than being refused — refusing would make the answer depend
    // on how recently the sweeper ran, an ambient race no journal records.
    let mut lazarus = make("memory-lazarus", "team-lazarus", "support", json!({"n": 1}));
    lazarus.access_retention_seconds = Some(60);
    store.remember(&lazarus).await.expect("lazarus memory");
    // The untouched window closed at 1_760_000_060; nobody swept. The touch
    // at 1_760_000_100 must reopen it to 1_760_000_160.
    store
        .touch(&["memory-lazarus".to_owned()], at(1_760_000_100))
        .await
        .expect("touch a lapsed window");
    assert_eq!(
        store
            .recall(&Recall::about("team-lazarus").at(at(1_760_000_150)))
            .await
            .expect("inside the reopened window")
            .len(),
        1,
        "a touch on a lapsed-but-unswept id must resurrect it — expiry \
         becomes fact at the sweep, and until then the id is alive to touch"
    );
    assert!(
        store
            .recall(&Recall::about("team-lazarus").at(at(1_760_000_160)))
            .await
            .expect("after the reopened window")
            .is_empty(),
        "the reopened window is a window, not immortality"
    );

    // The rolling-summary hole, closed: superseding a derivative does not
    // un-absorb its sources. S v1 summarises A; T summarises S v1; S is then
    // re-derived from B alone, so S v2 is clean — but S v1, which contains
    // A's content, stays readable through version(). A cascade from A must
    // erase S's superseded v1 and reach T *through* it, while sparing S v2:
    // an honest re-derivation is not destroyed by its own history.
    let sel = |item: &MemoryItem| Selected {
        id: item.id.clone(),
        version: item.version,
        digest: item.selection_digest(),
    };
    store
        .remember(&make("roll-a", "team-roll", "support", json!({"n": 1})))
        .await
        .expect("roll source a");
    store
        .remember(&make("roll-b", "team-roll", "support", json!({"n": 2})))
        .await
        .expect("roll source b");
    let roll_a = store
        .version("roll-a", 1)
        .await
        .expect("read a")
        .expect("a exists");
    let roll_b = store
        .version("roll-b", 1)
        .await
        .expect("read b")
        .expect("b exists");
    let mut summary = make("roll-s", "team-roll", "support", json!({"sum": "of a"}));
    summary.derived_from = vec![sel(&roll_a)];
    store.remember(&summary).await.expect("summary v1");
    let summary_v1 = store
        .version("roll-s", 1)
        .await
        .expect("read v1")
        .expect("v1 exists");
    let mut over = make("roll-t", "team-roll", "support", json!({"sum": "of s v1"}));
    over.derived_from = vec![sel(&summary_v1)];
    store.remember(&over).await.expect("summary of the summary");
    let mut summary2 = make("roll-s", "team-roll", "support", json!({"sum": "of b"}));
    summary2.created_at = at(1_760_000_001);
    summary2.derived_from = vec![sel(&roll_b)];
    store.remember(&summary2).await.expect("summary v2");

    // The positive half: before the cascade, the superseded version is
    // readable and still names A — this is exactly the copy the cascade must
    // reach.
    let v1 = store
        .version("roll-s", 1)
        .await
        .expect("v1 read")
        .expect("v1 kept before the cascade");
    assert_eq!(v1.derived_from[0].id, "roll-a");
    // And the *live* derivative view has moved on: repairs act on v2.
    assert!(
        store
            .derivatives("roll-a")
            .await
            .expect("derivatives of a")
            .is_empty(),
        "a re-derived summary is no longer a live derivative of its old source"
    );

    assert_eq!(
        store
            .forget_cascading("roll-a")
            .await
            .expect("cascade from the superseded source"),
        3,
        "the cascade must erase roll-a, roll-s's superseded v1 (counted once) \
         and roll-t, which was derived from exactly that version"
    );
    assert!(
        store
            .version("roll-a", 1)
            .await
            .expect("a lookup")
            .is_none()
    );
    assert!(
        store
            .version("roll-s", 1)
            .await
            .expect("superseded v1 lookup")
            .is_none(),
        "the superseded summary version that absorbed the erased source is \
         still readable — supersession sheltered absorbed content from the \
         cascade"
    );
    assert!(
        store
            .version("roll-t", 1)
            .await
            .expect("roll-t lookup")
            .is_none(),
        "a derivative of the superseded version survived — the cascade did \
         not route through the version it erased"
    );
    let kept = store
        .version("roll-s", 2)
        .await
        .expect("v2 read")
        .expect("the clean re-derivation survives");
    assert_eq!(kept.derived_from[0].id, "roll-b");
    assert!(
        store
            .recall(&Recall::about("team-roll"))
            .await
            .expect("recall after the cascade")
            .iter()
            .any(|item| item.id == "roll-s"),
        "the re-derived current summary must stay recallable"
    );
    assert!(
        store
            .version("roll-b", 1)
            .await
            .expect("b lookup")
            .is_some(),
        "the clean source is untouched"
    );

    // The purpose-less selection rule is global: most trusted first, then
    // newest **across purposes**, then id. An index that leads with the
    // purpose hands back whichever purpose sorts first, so an implementation
    // truncating in scan order returns every label correct and the newest
    // memory silently missing — which is why the rule is pinned here for
    // every backend rather than left as a ranking choice.
    let mut order_old = make("order-old", "team-order", "alpha", json!({"n": 1}));
    order_old.created_at = at(1_760_000_300);
    store
        .remember(&order_old)
        .await
        .expect("older, first purpose");
    let mut order_new = make("order-new", "team-order", "zeta", json!({"n": 2}));
    order_new.created_at = at(1_760_000_400);
    store
        .remember(&order_new)
        .await
        .expect("newest, last purpose");
    let mut order_trusted = make("order-trusted", "team-order", "middle", json!({"n": 3}));
    order_trusted.trust = Trust::Trusted;
    order_trusted.created_at = at(1_760_000_100);
    store
        .remember(&order_trusted)
        .await
        .expect("trusted, oldest");
    let picked = store
        .recall(&Recall::about("team-order").limit(2))
        .await
        .expect("ordered recall");
    assert_eq!(
        picked.iter().map(|i| i.id.as_str()).collect::<Vec<_>>(),
        vec!["order-trusted", "order-new"],
        "a purpose-less recall must rank trust first and then recency \
         globally across purposes — not whichever purpose sorts first"
    );
    let all = store
        .recall(&Recall::about("team-order").limit(3))
        .await
        .expect("full ordered recall");
    assert_eq!(
        all.iter().map(|i| i.id.as_str()).collect::<Vec<_>>(),
        vec!["order-trusted", "order-new", "order-old"]
    );

    // The erasure path's enumeration: every current id of the subject, in
    // stable order, with no page size to fall off.
    assert_eq!(
        store.subject_ids("team-order").await.expect("subject ids"),
        vec!["order-new", "order-old", "order-trusted"],
        "subject_ids must name every current id of the subject"
    );
    assert!(
        store
            .subject_ids("team-nobody")
            .await
            .expect("empty subject")
            .is_empty()
    );
}

/// Exercise the event buffer's erasure contract against any backend.
///
/// The buffer keeps a copy of every inbound payload — claimed rows for dedup,
/// dead-lettered rows for the operator — and this battery pins the three ways
/// that copy is allowed to die: shed at `unsubscribe` once the journal holds
/// the delivered copy, and removed by `erase_payload` for rows nobody claimed
/// or that dead-lettered. In every case the **row itself survives**, because
/// dedup and dead-letter accounting are identity, not content.
///
/// # Panics
///
/// Panics when the backend violates the contract.
#[allow(clippy::too_many_lines)]
pub async fn event_erasure(store: Arc<dyn crate::case::EventStore>) {
    use crate::core::{
        CorrelationKey, InboundEvent, Phase, RunId, StepId, Subscription, Timestamp,
    };
    use serde_json::json;

    let at = |seconds| Timestamp::from_unix_timestamp(seconds).expect("representable time");
    let event = |id: &str, payload: serde_json::Value| InboundEvent {
        source: "counterparty".to_owned(),
        id: id.to_owned(),
        kind: "reply".to_owned(),
        correlation: vec![CorrelationKey::new("order", id.to_owned())],
        payload,
    };
    let sub = |run: RunId, effect, id: &str| Subscription {
        run,
        case: None,
        effect,
        step: StepId(1),
        phase: Phase::Forward,
        kind: "reply".to_owned(),
        correlation: vec![CorrelationKey::new("order", id.to_owned())],
    };

    // ── A delivered payload is shed at unsubscribe, and dedup survives ─────
    let run = RunId::generate();
    let effect = key(201);
    let pii = json!({"pii": "the counterparty's content"});
    let first = event("A-1", pii.clone());
    assert!(store.buffer(&first, at(1_000)).await.expect("buffer"));
    store
        .subscribe(&sub(run, effect, "A-1"), at(1_001))
        .await
        .expect("subscribe");
    let claimed = store
        .claim_for(&sub(run, effect, "A-1"), at(1_002))
        .await
        .expect("claim")
        .expect("a buffered event matches");
    assert_eq!(
        claimed.event.payload, pii,
        "delivery hands over the payload"
    );

    // Crash recovery happens *before* the run acknowledges, so the buffer's
    // copy must survive until the unsubscribe — the journal may not hold the
    // payload yet, and recovery re-reads it from here.
    let recovered = store
        .claim_for(&sub(run, effect, "A-1"), at(1_003))
        .await
        .expect("re-claim")
        .expect("the run's own claim is returned to it");
    assert_eq!(
        recovered.event.payload, pii,
        "a payload stripped at the claim is lost to a run that crashed \
         between claim and resume — stripping must wait for the unsubscribe"
    );

    store.unsubscribe(run, effect).await.expect("unsubscribe");
    store
        .subscribe(&sub(run, effect, "A-1"), at(1_004))
        .await
        .expect("resubscribe");
    let after = store
        .claim_for(&sub(run, effect, "A-1"), at(1_005))
        .await
        .expect("claim after the strip")
        .expect("the row keeps its identity");
    assert_eq!(
        after.event.payload,
        serde_json::Value::Null,
        "a delivered payload outlived its delivery in the buffer — the \
         journal holds the delivered copy, and the buffer needs only the \
         (source, id) identity"
    );
    store.unsubscribe(run, effect).await.expect("clean up");
    assert!(
        !store.buffer(&first, at(1_006)).await.expect("replay"),
        "shedding the payload deleted the dedup row — a replay of the \
         delivered message was accepted as new, so erasure reopened the door \
         it was supposed to close"
    );

    // ── An unclaimed payload is erasable, and dead-lettering still counts ──
    let unclaimed = event("B-1", json!({"pii": "unclaimed"}));
    assert!(store.buffer(&unclaimed, at(2_000)).await.expect("buffer"));
    assert!(
        store
            .erase_payload("counterparty", "B-1")
            .await
            .expect("erase the unclaimed payload"),
        "the row exists, so the erasure must report having acted"
    );
    assert_eq!(
        store
            .sweep_unclaimed(at(2_100), "aged out")
            .await
            .expect("sweep"),
        1,
        "an erased event still dead-letters — erasure removes content, never \
         accounting"
    );
    let letters = store.dead_letters(10).await.expect("dead letters");
    let erased = letters
        .iter()
        .find(|letter| letter.event.id == "B-1")
        .expect("the erased event is still listed");
    assert_eq!(
        erased.event.payload,
        serde_json::Value::Null,
        "an erased unclaimed event still yielded its payload"
    );
    assert_eq!(erased.reason, "aged out");
    assert_eq!(
        erased.event.correlation,
        vec![CorrelationKey::new("order", "B-1")],
        "correlation keys are the identity the operator routes on, not content"
    );

    // ── A dead-lettered payload is erasable in place ───────────────────────
    let dead = event("C-1", json!({"pii": "dead"}));
    assert!(store.buffer(&dead, at(3_000)).await.expect("buffer"));
    assert_eq!(
        store
            .sweep_unclaimed(at(3_100), "nobody came")
            .await
            .expect("sweep"),
        1
    );
    // The positive half: a dead letter is readable until somebody erases it —
    // that list is how a wrong correlation key gets found.
    let letters = store.dead_letters(10).await.expect("dead letters");
    assert_eq!(
        letters
            .iter()
            .find(|letter| letter.event.id == "C-1")
            .expect("listed")
            .event
            .payload,
        json!({"pii": "dead"})
    );
    assert!(
        store
            .erase_payload("counterparty", "C-1")
            .await
            .expect("erase the dead letter")
    );
    let letters = store.dead_letters(10).await.expect("dead letters");
    assert_eq!(
        letters
            .iter()
            .find(|letter| letter.event.id == "C-1")
            .expect("still listed")
            .event
            .payload,
        serde_json::Value::Null,
        "an erased dead letter still yielded its payload"
    );

    // Erasing a message that never arrived reports that truthfully.
    assert!(
        !store
            .erase_payload("counterparty", "never-arrived")
            .await
            .expect("absent row")
    );
}

fn admitted_under(run: RunId, key: &str) -> Append {
    Append::new(
        run,
        RecordKind::RunAdmitted {
            capability: "conformance".into(),
            governed_by: None,
            input_label: crate::core::Label::trusted(),
            input: serde_json::Value::Null,
            policy_bundle: None,
            canon: crate::core::canon::VERSION,
            idempotency_key: Some(key.to_owned()),
        },
    )
}

fn admitted(run: RunId) -> Append {
    Append::new(
        run,
        RecordKind::RunAdmitted {
            capability: "conformance".into(),
            governed_by: None,
            input_label: crate::core::Label::trusted(),
            input: serde_json::Value::Null,
            policy_bundle: None,
            canon: crate::core::canon::VERSION,
            idempotency_key: None,
        },
    )
}

/// The conclusion record the executor writes — what feeds the outcome index.
fn concluded(run: RunId, outcome: &str) -> Append {
    Append::new(
        run,
        RecordKind::RunConcluded {
            outcome: outcome.to_owned(),
            reason: None,
            exhaustion: None,
            live_spend: crate::core::Spend::default(),
            chain_head: Digest::ZERO,
        },
    )
}

fn started(run: RunId, key: EffectKey) -> Append {
    Append::new(
        run,
        RecordKind::EffectStarted {
            descriptor: crate::core::EffectDescriptor::nullary("probe"),
            recovery: crate::core::Recovery::Retry,
            mutates: true,
            attempt: 1,
            backoff_ms: 0,
            outbound_label: None,
        },
    )
    .step(StepId(0))
    .effect(key)
}

fn key(n: u8) -> EffectKey {
    EffectKey::derive(
        StepId(0),
        crate::core::Phase::Forward,
        u32::from(n),
        1,
        "probe",
        &[n],
    )
}

/// Run every check.
///
/// # Panics
///
/// Only if the factory itself fails; contract violations are collected into the
/// [`Report`] rather than raised, so bringing up a backend surfaces all of them
/// at once.
pub async fn check(fresh: Factory<'_>) -> Report {
    let mut r = Report::default();
    head_of_an_unwritten_run_is_genesis(fresh, &mut r).await;
    append_assigns_contiguous_seq(fresh, &mut r).await;
    append_links_each_record_to_the_previous(fresh, &mut r).await;
    a_sound_chain_verifies(fresh, &mut r).await;
    a_second_start_for_one_effect_is_rejected(fresh, &mut r).await;
    the_same_effect_key_in_another_run_is_allowed(fresh, &mut r).await;
    a_stale_epoch_is_fenced(fresh, &mut r).await;
    a_fabricated_future_epoch_is_fenced(fresh, &mut r).await;
    a_live_lease_is_not_stolen(fresh, &mut r).await;
    a_takeover_advances_the_epoch(fresh, &mut r).await;
    acquire_claims_and_never_renews(fresh, &mut r).await;
    a_renewal_extends_only_what_is_still_held(fresh, &mut r).await;
    a_batch_spans_one_run(fresh, &mut r).await;
    a_stale_epoch_cannot_seal(fresh, &mut r).await;
    a_rejected_batch_writes_nothing(fresh, &mut r).await;
    read_starts_where_it_is_told(fresh, &mut r).await;
    a_case_scan_selects_one_matter(fresh, &mut r).await;
    sealed_runs_are_findable_by_how_they_ended(fresh, &mut r).await;
    the_outcome_index_follows_the_last_conclusion(fresh, &mut r).await;
    a_sealed_run_refuses_appends(fresh, &mut r).await;
    a_stop_request_needs_no_lease(fresh, &mut r).await;
    the_first_asker_stays_on_the_record(fresh, &mut r).await;
    an_attestation_survives_the_round_trip(fresh, &mut r).await;
    the_log_only_grows(fresh, &mut r).await;
    a_released_lease_is_free_at_once(fresh, &mut r).await;
    a_release_does_not_forget_the_epoch(fresh, &mut r).await;
    a_fenced_caller_cannot_release_the_new_owners_lease(fresh, &mut r).await;
    the_discovery_index_pages_in_a_total_order(fresh, &mut r).await;
    abandonment_names_the_dead_not_the_finished(fresh, &mut r).await;
    an_admission_key_admits_one_run(fresh, &mut r).await;
    a_key_is_claimed_by_the_append_that_writes_the_run(fresh, &mut r).await;
    a_refused_admission_leaves_its_key_free(fresh, &mut r).await;
    an_unkeyed_admission_claims_nothing(fresh, &mut r).await;
    retiring_a_key_frees_it_and_leaves_the_run(fresh, &mut r).await;
    r
}

/// A retired key is free again, and the run it named is untouched.
///
/// Both halves. Retirement that did not free the key would be a no-op wearing a
/// row count; retirement that touched the journal would be a retention policy
/// editing history, and the key's authority is the `RunAdmitted` record.
async fn retiring_a_key_frees_it_and_leaves_the_run(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let key = "conformance/retired";

    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    if store
        .append(lease.epoch, vec![admitted_under(run, key)])
        .await
        .is_err()
    {
        return;
    }

    // A cutoff before the claim retires nothing: the window is a bound on age,
    // not a switch.
    match store
        .forget_admissions(crate::core::Timestamp::from_unix_timestamp(0).expect("epoch"))
        .await
    {
        Ok(0) => {}
        Ok(n) => r.record(
            "admission keys",
            format!("a cutoff older than every claim retired {n} of them"),
        ),
        Err(e) => {
            r.record("admission keys", format!("forget_admissions failed: {e}"));
            return;
        }
    }
    if !matches!(store.admitted_as(key).await, Ok(Some(_))) {
        r.record(
            "admission keys",
            "a key was released by a retirement that reported retiring nothing",
        );
    }

    match store
        .forget_admissions(
            crate::core::Timestamp::from_unix_timestamp(4_102_444_800).expect("year 2100"),
        )
        .await
    {
        Ok(1) => {}
        Ok(n) => r.record(
            "admission keys",
            format!("retiring one claim past its window reported {n}"),
        ),
        Err(e) => {
            r.record("admission keys", format!("forget_admissions failed: {e}"));
            return;
        }
    }
    match store.admitted_as(key).await {
        Ok(None) => {}
        Ok(Some(held)) => r.record(
            "admission keys",
            format!("a retired key is still held by {held}, so retirement reclaims nothing"),
        ),
        Err(e) => r.record("admission keys", format!("admitted_as failed: {e}")),
    }
    match store.read(run, 1).await {
        Ok(records) if !records.is_empty() => {}
        Ok(_) => r.record(
            "admission keys",
            "retiring a key erased the run it named — a retention policy must not \
             edit history, and the key's authority is the RunAdmitted record",
        ),
        Err(e) => r.record("admission keys", format!("the run became unreadable: {e}")),
    }
}

/// One admission key admits one run, and the loser is told which run won.
///
/// A backend answering the second append with success lets a redelivery start a
/// second run, and every single-threaded test written for it still passes.
///
/// The **holder** is checked, not merely the refusal: a conflict naming the
/// wrong run sends the caller to a journal answering a different question.
async fn an_admission_key_admits_one_run(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let key = "conformance/one-message";

    let first = RunId::generate();
    let Ok(lease) = store.acquire(first, "conformance", LEASE).await else {
        return;
    };
    if let Err(e) = store
        .append(lease.epoch, vec![admitted_under(first, key)])
        .await
    {
        r.record(
            "admission keys",
            format!("the first admission under a fresh key was refused: {e}"),
        );
        return;
    }

    let second = RunId::generate();
    let Ok(lease) = store.acquire(second, "conformance", LEASE).await else {
        return;
    };
    match store
        .append(lease.epoch, vec![admitted_under(second, key)])
        .await
    {
        Err(crate::core::StoreError::DuplicateAdmission { run, .. }) => {
            if run != first.to_string() {
                r.record(
                    "admission keys",
                    format!(
                        "the refusal named '{run}' as the key's holder, but {first} holds it — a caller sent there reads a run that answers a different question"
                    ),
                );
            }
        }
        Ok(_) => r.record(
            "admission keys",
            "a second run was admitted under a key already issued — an emitter's redelivery would start a second run, which is the whole failure this refuses"
                .to_owned(),
        ),
        Err(e) => r.record(
            "admission keys",
            format!("the duplicate was refused, but not as DuplicateAdmission: {e}"),
        ),
    }

    match store.admitted_as(key).await {
        Ok(Some(held)) if held == first => {}
        Ok(other) => r.record(
            "admission keys",
            format!("admitted_as must name {first}, said {other:?}"),
        ),
        Err(e) => r.record("admission keys", format!("admitted_as failed: {e}")),
    }
}

/// The key is taken by the append, so it is never taken while the run is not.
///
/// A ledger written before the run strands the key over a run that never
/// existed if the process dies between the two. The property is observable:
/// before the append, the key must be free.
async fn a_key_is_claimed_by_the_append_that_writes_the_run(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let key = "conformance/not-yet";

    match store.admitted_as(key).await {
        Ok(None) => {}
        Ok(Some(run)) => {
            r.record(
                "admission keys",
                format!("a key nothing has admitted under is held by {run}"),
            );
            return;
        }
        Err(e) => {
            r.record("admission keys", format!("admitted_as failed: {e}"));
            return;
        }
    }

    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    // Taking the lease is not admitting: a run that was claimed and never
    // written has not spent its key, or a crash between the two would burn it.
    if let Ok(Some(held)) = store.admitted_as(key).await {
        r.record(
            "admission keys",
            format!("taking a lease claimed the key for {held}; only the append may"),
        );
    }
    let _ = store
        .append(lease.epoch, vec![admitted_under(run, key)])
        .await;
    match store.admitted_as(key).await {
        Ok(Some(held)) if held == run => {}
        Ok(other) => r.record(
            "admission keys",
            format!("after the append the key must be held by {run}, said {other:?}"),
        ),
        Err(e) => r.record("admission keys", format!("admitted_as failed: {e}")),
    }
}

/// An append that was refused spends no key.
///
/// The whole batch commits or none of it does, and the key is part of the batch.
/// A backend that claimed it and rolled the records back would answer every
/// future redelivery with a run that does not exist.
async fn a_refused_admission_leaves_its_key_free(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let key = "conformance/refused";

    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    // Refused for spanning runs — the one rejection every backend must make,
    // so the check does not depend on which other refusals a backend has.
    let other = RunId::generate();
    let refused = store
        .append(lease.epoch, vec![admitted_under(run, key), admitted(other)])
        .await;
    if refused.is_ok() {
        r.record(
            "admission keys",
            "a batch spanning two runs was accepted; the rest of this check cannot mean anything"
                .to_owned(),
        );
        return;
    }

    match store.admitted_as(key).await {
        Ok(None) => {}
        Ok(Some(held)) => r.record(
            "admission keys",
            format!(
                "a refused append left the key held by {held} — the key committed without the run it names, so every future retry is answered with a run that has no journal"
            ),
        ),
        Err(e) => r.record("admission keys", format!("admitted_as failed: {e}")),
    }
}

/// A run admitted without a key claims none.
///
/// A backend filing unkeyed admissions under the empty key would refuse the
/// second ordinary run as a duplicate of the first — every plane broken by a
/// feature it never uses.
async fn an_unkeyed_admission_claims_nothing(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;

    for _ in 0..2 {
        let run = RunId::generate();
        let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
            return;
        };
        if let Err(e) = store.append(lease.epoch, vec![admitted(run)]).await {
            r.record(
                "admission keys",
                format!("an admission carrying no key was refused: {e}"),
            );
            return;
        }
    }

    if let Ok(Some(held)) = store.admitted_as("").await {
        r.record(
            "admission keys",
            format!("an unkeyed admission was filed under the empty key, held by {held}"),
        );
    }
}

/// `abandoned_runs` returns exactly the runs an instance died holding.
///
/// The recovery sweep resumes everything this method returns, so both error
/// directions are expensive: a lapsed-but-held lease it misses is a run
/// stranded forever, and a released lease it includes is a clean exit
/// "recovered" — an epoch bump and a replay — on every tick for the rest of
/// the store's life.
async fn abandonment_names_the_dead_not_the_finished(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;

    // One of each fate. The dead instance stops renewing without releasing;
    // the clean exit releases; the live one holds a lease that has not lapsed.
    let dead = RunId::generate();
    let finished = RunId::generate();
    let alive = RunId::generate();
    let Ok(_) = store.acquire(dead, "instance-a", SHORT).await else {
        return;
    };
    let Ok(f) = store.acquire(finished, "instance-b", SHORT).await else {
        return;
    };
    if store.release_lease(finished, f.epoch).await.is_err() {
        r.record("recovery", "release_lease of a held lease failed");
        return;
    }
    let Ok(_) = store.acquire(alive, "instance-c", LEASE).await else {
        return;
    };

    tokio::time::sleep(Duration::from_millis(1_500)).await;

    match store.abandoned_runs(10).await {
        Ok(listed) => {
            if !listed.contains(&dead) {
                r.record(
                    "recovery",
                    "a lease that expired while still naming an owner was not listed — \
                     the run its instance died holding is stranded forever",
                );
            }
            if listed.contains(&finished) {
                r.record(
                    "recovery",
                    "a released lease was listed as abandoned — every clean exit would \
                     be 'recovered' on every tick",
                );
            }
            if listed.contains(&alive) {
                r.record(
                    "recovery",
                    "a live lease was listed as abandoned — the sweep would take a run \
                     over from under its working owner's heartbeat",
                );
            }
        }
        Err(e) => r.record("recovery", format!("abandoned_runs failed: {e}")),
    }
}

/// The discovery index is newest-first, bounded, and resumes without gap or
/// repeat.
///
/// This contract had **no battery entry at all** while the method was
/// unbounded, and the two facts belong together: a signature returning
/// everything has no page boundary to get wrong, so there was nothing to
/// check. Bounding it created the boundary, and a paged read has exactly two
/// silent failure modes — an item served twice, and an item served never —
/// and both look like working software from a single page.
///
/// The order must be **total**, which is why the tie-break on run id is part of
/// the contract rather than an implementation detail. Timestamps here have
/// whole-second granularity on both backends, so runs sharing one are ordinary;
/// without a tie-break the store may order them differently between two calls
/// and a cursor lands in the middle of the tie, dropping or duplicating
/// whichever moved.
async fn the_discovery_index_pages_in_a_total_order(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;

    // Four runs written back to back, so they very likely share a timestamp —
    // which is the case the tie-break exists for.
    let mut written = Vec::new();
    for _ in 0..4 {
        let run = RunId::generate();
        let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
            return;
        };
        if store
            .append(lease.epoch, vec![admitted(run)])
            .await
            .is_err()
        {
            return;
        }
        written.push(run);
    }

    let whole = match store.recent_runs(None, 100).await {
        Ok(rows) => rows,
        Err(e) => {
            r.record("chaining", format!("recent_runs failed: {e}"));
            return;
        }
    };
    if whole.len() != 4 {
        r.record(
            "chaining",
            format!("recent_runs returned {} rows for 4 runs", whole.len()),
        );
        return;
    }

    // Descending by (updated, run). A caller pages on this order, so a store
    // that returns its own order serves a cursor that means nothing.
    let mut sorted = whole.clone();
    sorted.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| b.0.cmp(&a.0)));
    if whole != sorted {
        r.record(
            "chaining",
            "recent_runs is not ordered by (updated_at, run) descending, so a \
             cursor into it cannot resume a page"
                .to_owned(),
        );
    }

    // The limit is a limit.
    match store.recent_runs(None, 2).await {
        Ok(rows) if rows.len() > 2 => r.record(
            "chaining",
            format!("recent_runs(limit=2) returned {} rows", rows.len()),
        ),
        Err(e) => r.record("chaining", format!("recent_runs(limit=2) failed: {e}")),
        Ok(_) => {}
    }

    // Paging two at a time must reassemble the whole index exactly: every run
    // once, in the same order. This is the assertion that catches an inclusive
    // cursor bound (one row served twice) and an off-by-one (one row skipped).
    let mut paged = Vec::new();
    let mut cursor: Option<(u64, RunId)> = None;
    for _ in 0..4 {
        let Ok(rows) = store.recent_runs(cursor, 2).await else {
            r.record("chaining", "a cursored recent_runs page failed".to_owned());
            return;
        };
        if rows.is_empty() {
            break;
        }
        cursor = rows.last().map(|(run, updated)| (*updated, *run));
        paged.extend(rows);
    }
    if paged != whole {
        r.record(
            "chaining",
            format!(
                "paging recent_runs two at a time did not reassemble the index: \
                 {} rows paged against {} whole — a page boundary either repeats \
                 a run or drops one, and both read as a healthy listing",
                paged.len(),
                whole.len()
            ),
        );
    }
}

/// A released lease is available immediately, without waiting out the TTL.
///
/// The counterpart to expiry, and the reason an owner can afford to be unique
/// per process. Without a release, a restart waits out the lease — and the
/// tempting fix is a constant owner string so the replacement "renews" instead,
/// which silently disables fencing, because two live instances then read each
/// other's lease as their own.
async fn a_released_lease_is_free_at_once(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(a) = store.acquire(run, "instance-a", LEASE).await else {
        return;
    };
    if let Err(e) = store.release_lease(run, a.epoch).await {
        r.record("fencing", format!("release() of a held lease failed: {e}"));
        return;
    }

    match store.acquire(run, "instance-b", LEASE).await {
        Ok(b) => {
            // The epoch *must* advance. An earlier version of this battery
            // asserted the opposite — "a handover is not a crash, so there is
            // nothing to fence" — which is wrong: releasing says the owner
            // intends to stop, not that it already has. An un-awaited task or a
            // crash between release and exit leaves an append in flight, and
            // the only thing that stops it is a bump. The property worth having
            // is that takeover is *immediate*, not that the epoch stands still.
            if b.epoch <= a.epoch {
                r.record(
                    "fencing",
                    format!(
                        "a released lease was reclaimed at epoch {} without advancing past \
                         {} — an append still in flight from the releasing owner would not \
                         be fenced",
                        b.epoch, a.epoch
                    ),
                );
            }
        }
        Err(e) => r.record(
            "fencing",
            format!(
                "a released lease was still held ({e}) — every restart then waits out the \
                 TTL, which is the pressure that makes deployments alias their owner strings"
            ),
        ),
    }

    // Releasing twice is releasing once.
    if let Err(e) = store.release_lease(run, a.epoch).await {
        r.record("fencing", format!("release() is not idempotent: {e}"));
    }
}

/// Releasing frees the lease without forgetting the epoch.
///
/// The epoch lives in the lease row, so a release that *deletes* the row throws
/// the fence away with it. Two things then go wrong at once, and both are
/// silent: `append` fences against the row it can no longer find, so a stale
/// writer is not stopped; and the next `acquire` sees nothing and starts again
/// at 1, so a previously fenced writer holding epoch 2 now **outranks** the new
/// legitimate owner. The mechanism does not merely stop working — it inverts.
async fn a_release_does_not_forget_the_epoch(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();

    // Take it, lose it to a takeover, so the epoch is above 1.
    let Ok(first) = store.acquire(run, "instance-a", SHORT).await else {
        return;
    };
    tokio::time::sleep(Duration::from_millis(1_200)).await;
    let Ok(second) = store.acquire(run, "instance-b", LEASE).await else {
        r.record("fencing", "an expired lease could not be taken over");
        return;
    };
    if second.epoch <= first.epoch {
        r.record("fencing", "a takeover did not advance the epoch");
        return;
    }

    // The current owner shuts down cleanly.
    if store.release_lease(run, second.epoch).await.is_err() {
        r.record("fencing", "release_lease of a held lease failed");
        return;
    }

    // The stale instance must still be fenced. If the epoch was forgotten there
    // is nothing left to compare against and this append succeeds.
    match store.append(first.epoch, vec![admitted(run)]).await {
        Err(crate::core::StoreError::Fenced { .. }) => {}
        Err(e) => r.record(
            "fencing",
            format!("unexpected error from a stale append: {e}"),
        ),
        Ok(_) => r.record(
            "fencing",
            "a fenced writer appended after the lease was released — releasing threw \
             away the epoch, so there was nothing left to fence against",
        ),
    }

    // And the next owner must rank *above* the fenced one, not restart at 1.
    match store.acquire(run, "instance-c", LEASE).await {
        Ok(third) => {
            if third.epoch <= first.epoch {
                r.record(
                    "fencing",
                    format!(
                        "after a release the epoch restarted at {} — a writer fenced at {} \
                         now outranks the legitimate owner",
                        third.epoch, first.epoch
                    ),
                );
            }
        }
        Err(e) => r.record(
            "fencing",
            format!("a released lease was not claimable: {e}"),
        ),
    }
}

/// A fenced caller cannot free the lease of whoever replaced it.
///
/// The safety half. A process that was taken over is still running and will
/// eventually shut down; if its release freed the *current* owner's lease, it
/// would hand the run to a third party while the rightful owner is mid-write —
/// manufacturing exactly the split-brain the epoch exists to prevent.
async fn a_fenced_caller_cannot_release_the_new_owners_lease(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(old) = store.acquire(run, "instance-a", SHORT).await else {
        return;
    };
    // Waited out rather than faked, because expiry is only reachable through
    // the public API — the same reason the fencing case does it this way.
    tokio::time::sleep(Duration::from_millis(1_200)).await;

    // Expired, so the takeover is legitimate and bumps the epoch.
    let Ok(new) = store.acquire(run, "instance-b", LEASE).await else {
        r.record("fencing", "an expired lease could not be taken over");
        return;
    };
    if new.epoch == old.epoch {
        r.record("fencing", "a takeover did not advance the epoch");
        return;
    }

    // The stale instance shuts down and releases. It must not succeed, and it
    // must not be an error either: being fenced is not a reason to fail an
    // orderly exit.
    if let Err(e) = store.release_lease(run, old.epoch).await {
        r.record(
            "fencing",
            format!("a fenced caller's release must be a no-op, not an error: {e}"),
        );
    }

    match store.acquire(run, "instance-c", LEASE).await {
        Err(crate::core::StoreError::LeaseHeld { .. }) => {}
        Err(e) => r.record(
            "fencing",
            format!("unexpected error after a stale release: {e}"),
        ),
        Ok(_) => r.record(
            "fencing",
            "a fenced caller released the lease of the instance that replaced it, handing \
             the run to a third party while its rightful owner is still writing",
        ),
    }
}

/// A sealed run enters the log, and the log only ever grows.
///
/// Two properties, and the second is what makes the first useful. An inclusion
/// proof says a run is committed to *now*; a consistency proof says every run
/// committed to *before* still is. Without the second, an auditor comparing two
/// checkpoints learns nothing — the root moves on every ordinary seal, so
/// legitimate growth and deletion-plus-growth look identical.
///
/// The earlier checkpoints are **kept as they were taken**, which is what an
/// auditor holds. Recomputing them from the store's current leaves would ask the
/// store to confirm its own ordering, and a bug in that ordering would agree
/// with itself.
async fn the_log_only_grows(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;

    let mut sealed = Vec::new();
    let mut history: Vec<(u64, Digest)> = Vec::new();

    for _ in 0..4u8 {
        let run = RunId::generate();
        let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
            r.record("fencing", "acquire() failed on a fresh run");
            return;
        };
        if store
            .append(lease.epoch, vec![admitted(run)])
            .await
            .is_err()
        {
            r.record("merkle-log", "append failed under a fresh lease");
            return;
        }
        if store.seal(run, lease.epoch, "succeeded").await.is_err() {
            r.record("merkle-log", "seal failed");
            return;
        }
        sealed.push(run);

        // A backend may legitimately maintain no log. What it must not do is
        // claim an empty one, which is why a failure here returns rather than
        // recording — and why the size check below is not skippable once a
        // checkpoint has been produced at all.
        let Ok(cp) = store.checkpoint().await else {
            return;
        };
        if cp.size != sealed.len() as u64 {
            r.record(
                "merkle-log",
                format!(
                    "{} runs are sealed and the checkpoint commits to {} — a run \
                     that sealed without entering the log is a run no checkpoint \
                     covers",
                    sealed.len(),
                    cp.size
                ),
            );
            return;
        }
        history.push((cp.size, cp.root));
    }

    let Ok(latest) = store.checkpoint().await else {
        return;
    };

    // Every sealed run proves its own inclusion against the current root.
    for (i, run) in sealed.iter().enumerate() {
        match store.inclusion_proof(*run).await {
            Ok(Some(inc)) => {
                let leaf = crate::core::merkle::leaf_hash(&inc.seal);
                if !crate::core::merkle::verify_inclusion(
                    leaf,
                    usize::try_from(inc.index).unwrap_or(usize::MAX),
                    usize::try_from(inc.size).unwrap_or(0),
                    &inc.proof,
                    &latest.root,
                ) {
                    r.record(
                        "merkle-log",
                        format!("run {i} is in the log and cannot prove it"),
                    );
                }
            }
            Ok(None) => r.record(
                "merkle-log",
                format!("run {i} sealed and the store has no position for it"),
            ),
            Err(e) => r.record("merkle-log", format!("inclusion_proof failed: {e}")),
        }
    }

    // And every checkpoint taken along the way proves the log only appended.
    for (size, root) in &history {
        let proof = match store.consistency_proof(*size).await {
            Ok(p) => p,
            Err(e) => {
                r.record(
                    "merkle-log",
                    format!("consistency_proof({size}) failed: {e}"),
                );
                continue;
            }
        };
        if !crate::core::merkle::verify_consistency(
            usize::try_from(*size).unwrap_or(0),
            root,
            usize::try_from(latest.size).unwrap_or(0),
            &latest.root,
            &proof,
        ) {
            r.record(
                "merkle-log",
                format!(
                    "the log could not prove it only appended between size {size} \
                     and {} — so a checkpoint published then proves nothing now",
                    latest.size
                ),
            );
        }
    }
}

/// A stop request must not be fenced.
///
/// Every other write here requires the lease, and for a stop request that would
/// be exactly backwards: the operator asking is not the owner, holds no epoch,
/// and is usually asking *because* the owner is busy doing the thing they want
/// stopped. A backend that reuses the append path for this — an easy mistake,
/// since every other write goes through it — makes the only party who can cancel
/// a running agent the process running it.
async fn a_stop_request_needs_no_lease(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();

    // Somebody else owns the run, at an epoch this caller does not have.
    let Ok(lease) = store.acquire(run, "the-owner", LEASE).await else {
        r.record("fencing", "acquire() failed on a fresh run");
        return;
    };
    if store
        .append(lease.epoch, vec![admitted(run)])
        .await
        .is_err()
    {
        r.record("chaining", "append failed under a fresh lease");
        return;
    }

    match store.request_cancel(run, "operator", "stop it").await {
        Ok(true) => {}
        Ok(false) => r.record(
            "cancellation",
            "the first stop request on a run reported that one already existed",
        ),
        Err(e) => r.record(
            "cancellation",
            format!(
                "a stop request from somebody who does not hold the lease was refused ({e}) — \
                 then the only party who can stop a running agent is the process running it"
            ),
        ),
    }

    match store.cancellation(run).await {
        Ok(Some(c)) if c.actor == "operator" && c.reason == "stop it" => {}
        Ok(Some(c)) => r.record(
            "cancellation",
            format!("the stop request read back as {c:?}, which is not what was written"),
        ),
        Ok(None) => r.record(
            "cancellation",
            "a recorded stop request was not readable — an operator's intervention \
             that the owner cannot see is not an intervention",
        ),
        Err(e) => r.record("cancellation", format!("cancellation() failed: {e}")),
    }

    // And a run nobody asked about has none.
    match store.cancellation(RunId::generate()).await {
        Ok(None) => {}
        Ok(Some(_)) => r.record(
            "cancellation",
            "an untouched run reported a stop request — every run would unwind",
        ),
        Err(e) => r.record("cancellation", format!("cancellation() failed: {e}")),
    }
}

/// A second request must not rewrite who intervened.
///
/// The obvious implementation is an upsert, and it silently reassigns the
/// intervention to whoever asked last. Six weeks later "who stopped this run?"
/// has the wrong answer, and it looks authoritative.
async fn the_first_asker_stays_on_the_record(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();

    if !matches!(store.request_cancel(run, "alice", "first").await, Ok(true)) {
        r.record("cancellation", "the first stop request was not recorded");
        return;
    }
    match store.request_cancel(run, "bob", "second").await {
        Ok(false) => {}
        Ok(true) => r.record(
            "cancellation",
            "a second stop request reported itself as the first — a retry would \
             read as a fresh intervention",
        ),
        Err(e) => r.record(
            "cancellation",
            format!("a repeated stop request failed: {e}"),
        ),
    }
    match store.cancellation(run).await {
        Ok(Some(c)) if c.actor == "alice" && c.reason == "first" => {}
        Ok(other) => r.record(
            "cancellation",
            format!(
                "the stop request now reads {other:?} — the second asker overwrote the \
                 first, so the permanent record names the wrong person"
            ),
        ),
        Err(e) => r.record("cancellation", format!("cancellation() failed: {e}")),
    }
}

/// A signature written must be a signature read back.
///
/// The easy way to get this wrong is to store the record and quietly drop the
/// two extra columns — nothing fails, the chain still verifies, and the loss is
/// invisible until an auditor asks who wrote something and every record answers
/// "nobody". A backend that hashes but silently discards authorship is worse
/// than one that never claimed to keep it.
async fn an_attestation_survives_the_round_trip(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        r.record("fencing", "acquire() failed on a fresh run");
        return;
    };

    let Ok(written) = store.append(lease.epoch, vec![admitted(run)]).await else {
        r.record("attestation", "append failed under a fresh lease");
        return;
    };

    // Only stores configured with a signer produce one. An unsigned store is a
    // legitimate configuration, so the check is conditional on the write having
    // carried an attestation at all — what it must never do is *lose* one.
    let Some(expected) = written.first().and_then(|w| w.attestation.clone()) else {
        return;
    };

    match store.read(run, 1).await {
        Ok(back) => match back.first().and_then(|b| b.attestation.clone()) {
            Some(got) if got == expected => {}
            Some(got) => r.record(
                "attestation",
                format!(
                    "the signature read back is not the one written: wrote {:?}, read {:?}",
                    expected.key_id, got.key_id
                ),
            ),
            None => r.record(
                "attestation",
                "a signed record read back unsigned — the chain still verifies, and \
                 authorship is gone without anything reporting it",
            ),
        },
        Err(e) => r.record("attestation", format!("read failed: {e}")),
    }
}

async fn head_of_an_unwritten_run_is_genesis(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    match store.head(RunId::generate()).await {
        Ok(head) if head.seq == 0 && head.hash == Digest::ZERO => {}
        Ok(head) => r.record(
            "chaining",
            format!(
                "a run with no records must start at genesis (seq 0, zero hash), got seq {} — \
                 a chain that starts anywhere else cannot be verified from the beginning",
                head.seq
            ),
        ),
        Err(e) => r.record(
            "chaining",
            format!("head() of an unwritten run must succeed at genesis, got {e}"),
        ),
    }
}

async fn append_assigns_contiguous_seq(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        r.record("fencing", "acquire() failed on a fresh run");
        return;
    };

    for i in 0..3u8 {
        if let Err(e) = store.append(lease.epoch, vec![admitted(run)]).await {
            r.record("chaining", format!("append {i} failed: {e}"));
            return;
        }
    }
    match store.read(run, 1).await {
        Ok(records) => {
            let seqs: Vec<u64> = records.iter().map(Record::seq).collect();
            if seqs != vec![1, 2, 3] {
                r.record(
                    "chaining",
                    format!(
                        "seq must be contiguous from 1, got {seqs:?} — a gap means a record was \
                         lost and verification cannot tell that from tampering"
                    ),
                );
            }
        }
        Err(e) => r.record("chaining", format!("read failed: {e}")),
    }
}

async fn append_links_each_record_to_the_previous(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    for _ in 0..3 {
        let _ = store.append(lease.epoch, vec![admitted(run)]).await;
    }
    let Ok(records) = store.read(run, 1).await else {
        return;
    };

    let mut prev = Digest::ZERO;
    for rec in &records {
        if rec.prev_hash != prev {
            r.record(
                "chaining",
                format!(
                    "record {} does not link to its predecessor — the chain is what makes \
                     deletion and reordering detectable, and an unlinked record is neither",
                    rec.seq()
                ),
            );
            return;
        }
        prev = rec.hash;
    }
}

async fn a_sound_chain_verifies(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    for _ in 0..4 {
        let _ = store.append(lease.epoch, vec![admitted(run)]).await;
    }
    if let Err(e) = store.verify(run).await {
        r.record(
            "chaining",
            format!("a chain this store wrote itself must verify, got {e}"),
        );
    }
}

async fn a_second_start_for_one_effect_is_rejected(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    let k = key(1);
    if let Err(e) = store.append(lease.epoch, vec![started(run, k)]).await {
        r.record("exactly-once", format!("the first start must succeed: {e}"));
        return;
    }
    match store.append(lease.epoch, vec![started(run, k)]).await {
        Err(crate::core::StoreError::DuplicateEffect(_)) => {}
        Err(e) => r.record(
            "exactly-once",
            format!("a second start must be rejected as DuplicateEffect, got {e}"),
        ),
        Ok(_) => r.record(
            "exactly-once",
            "a second EffectStarted for one effect key was accepted. Exactly-once is a \
             storage constraint here, not a code path — replay reads a completed effect \
             back, and this is what stops anything that gets past it from performing the \
             call twice",
        ),
    }
}

async fn the_same_effect_key_in_another_run_is_allowed(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let k = key(2);
    for _ in 0..2 {
        let run = RunId::generate();
        let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
            return;
        };
        if let Err(e) = store
            .append(lease.epoch, vec![started(run, k)].clone())
            .await
        {
            r.record(
                "exactly-once",
                format!(
                    "the constraint is per-run: two runs performing the same effect are two \
                     performances, and rejecting the second would make a batch of identical \
                     items unrunnable. Got {e}"
                ),
            );
            return;
        }
    }
}

async fn a_stale_epoch_is_fenced(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();

    // A lease that is already expired, so the takeover below is legitimate
    // rather than theft. This is the shape of the real failure: an instance
    // that was paused past its TTL and does not know it.
    let Ok(first) = store.acquire(run, "instance-a", SHORT).await else {
        r.record("fencing", "acquire() failed on a fresh run");
        return;
    };
    tokio::time::sleep(Duration::from_millis(1_200)).await;

    let Ok(second) = store.acquire(run, "instance-b", LEASE).await else {
        r.record(
            "fencing",
            "an expired lease must be claimable by another instance, or a crashed \
             owner strands its run forever",
        );
        return;
    };
    if second.epoch <= first.epoch {
        r.record(
            "fencing",
            "a takeover must advance the epoch, or the previous owner cannot be fenced",
        );
        return;
    }
    match store.append(first.epoch, vec![admitted(run)]).await {
        Err(crate::core::StoreError::Fenced { .. }) => {}
        Err(e) => r.record("fencing", format!("expected Fenced, got {e}")),
        Ok(_) => r.record(
            "fencing",
            "a superseded owner's write was accepted. The check must happen inside the same \
             transaction that writes: a read-then-write leaves a window a paused instance \
             wakes up into, which is split-brain",
        ),
    }
}

/// A fencing token is proof returned by `acquire`, not a sequence number a
/// caller may predict.
///
/// Checking only `writer < current` rejects zombies but accepts a fabricated
/// future epoch. The real owner can then append under the actual epoch as well,
/// so the store has admitted two writers while still claiming the lease is
/// exclusive. Equality is the ownership check.
async fn a_fabricated_future_epoch_is_fenced(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "instance-a", LEASE).await else {
        r.record("fencing", "acquire() failed on a fresh run");
        return;
    };

    let invented = lease.epoch.saturating_add(1);
    match store.append(invented, vec![admitted(run)]).await {
        Err(crate::core::StoreError::Fenced { .. }) => {}
        Err(e) => r.record("fencing", format!("expected Fenced, got {e}")),
        Ok(_) => r.record(
            "fencing",
            "an epoch the store never issued was accepted. A fencing token is proof of a \
             lease, not a number a caller may outrank by adding one",
        ),
    }
}

/// A *live* lease is not stealable, and saying so is not the same as fencing.
///
/// The two errors call for opposite responses. A fenced writer has been taken
/// over and must drop the run; a writer refused a live lease is not stale at all
/// and should wait for expiry. A backend that returns one for the other sends an
/// operator — or a retry loop — in exactly the wrong direction.
async fn a_live_lease_is_not_stolen(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(_held) = store.acquire(run, "instance-a", LEASE).await else {
        return;
    };
    match store.acquire(run, "instance-b", LEASE).await {
        Err(crate::core::StoreError::LeaseHeld { .. }) => {}
        Err(e) => r.record(
            "fencing",
            format!(
                "taking a live lease must fail as LeaseHeld, not {e} — a writer that is \
                 merely early must not be told it has been superseded"
            ),
        ),
        Ok(_) => r.record(
            "fencing",
            "a live lease was handed to a second instance. Two owners writing one chain is \
             the split-brain the epoch exists to prevent",
        ),
    }
}

async fn a_takeover_advances_the_epoch(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(a) = store.acquire(run, "instance-a", SHORT).await else {
        return;
    };
    tokio::time::sleep(Duration::from_millis(1_500)).await;
    let Ok(again) = store.acquire(run, "instance-b", LEASE).await else {
        r.record("fencing", "an expired lease could not be claimed");
        return;
    };
    if again.epoch <= a.epoch {
        r.record(
            "fencing",
            format!(
                "claiming an expired lease must advance the epoch ({} did not move past \
                 {}) — without the bump the dead owner's writes are never fenced",
                again.epoch, a.epoch
            ),
        );
    }
}

/// `acquire` is a pure claim: a live lease is refused, **the caller's own
/// included**.
///
/// The same-owner case is the one a backend author "helpfully" allows, and it
/// re-opens two closed defects at once: a heartbeat racing its run's
/// conclusion re-takes the lease the conclusion just released, and a second
/// entry point on one instance drives a second execution of a run the
/// instance is already executing — under the *same epoch*, which fencing
/// cannot see. Renewal is `renew`, checked below.
async fn acquire_claims_and_never_renews(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(_) = store.acquire(run, "instance-a", LEASE).await else {
        return;
    };
    match store.acquire(run, "instance-a", LEASE).await {
        Err(crate::core::StoreError::LeaseHeld { .. }) => {}
        Err(e) => r.record(
            "fencing",
            format!("a same-owner acquire of a held lease must fail as LeaseHeld, not {e}"),
        ),
        Ok(lease) => r.record(
            "fencing",
            format!(
                "a same-owner acquire of a held lease succeeded at epoch {} — acquire \
                 renewed instead of claiming, so two executors on one instance share \
                 one epoch and fencing cannot tell them apart",
                lease.epoch
            ),
        ),
    }
}

/// `renew` extends exactly what is still held, and never claims.
///
/// Four refusals and one success, and the refusals carry the weight: a renewal
/// that re-takes a released lease resurrects ownership over a run that
/// concluded — which the recovery sweep then "recovers" when the lease lapses
/// — and one that accepts a stale epoch renews the fenced past over the
/// current owner.
async fn a_renewal_extends_only_what_is_still_held(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "instance-a", LEASE).await else {
        return;
    };

    // Held, unexpired, right owner and epoch: the one case that succeeds, and
    // the epoch must not move — bumping it fences the owner against itself.
    match store.renew(run, "instance-a", lease.epoch, LEASE).await {
        Ok(renewed) if renewed.epoch == lease.epoch => {}
        Ok(renewed) => r.record(
            "fencing",
            format!(
                "a renewal moved the epoch ({} became {}) — the owner is fenced \
                 against its own in-flight writes",
                lease.epoch, renewed.epoch
            ),
        ),
        Err(e) => r.record("fencing", format!("renewing a held lease failed: {e}")),
    }

    // The wrong owner and the wrong epoch are both refusals, not claims.
    if store
        .renew(run, "instance-b", lease.epoch, LEASE)
        .await
        .is_ok()
    {
        r.record(
            "fencing",
            "another owner renewed a lease it does not hold — renewal became theft",
        );
    }
    if store
        .renew(run, "instance-a", lease.epoch + 1, LEASE)
        .await
        .is_ok()
    {
        r.record(
            "fencing",
            "a renewal with a fabricated epoch succeeded — a caller invented \
             authority it never acquired",
        );
    }

    // Released is gone. This is the heartbeat-races-conclusion shape: the run
    // concluded and handed its lease back, and a late renewal must fail
    // rather than resurrect it — a live lease over a concluded run is a run
    // the recovery sweep re-executes forever once it lapses.
    if store.release_lease(run, lease.epoch).await.is_err() {
        r.record("fencing", "release_lease of a held lease failed");
        return;
    }
    match store.renew(run, "instance-a", lease.epoch, LEASE).await {
        Err(_) => {}
        Ok(_) => r.record(
            "fencing",
            "a renewal re-took a lease the owner had already released — the \
             heartbeat racing its own run's conclusion leaves a live lease over \
             a concluded run, and the sweep recovers a clean exit forever",
        ),
    }
    // And the failed renewal left nothing behind: the run must not surface as
    // abandoned, now or after any TTL, because it exited cleanly.
    match store.abandoned_runs(10).await {
        Ok(listed) if listed.contains(&run) => r.record(
            "recovery",
            "a released run appears abandoned after a refused renewal — the \
             renewal wrote something it had no claim to",
        ),
        _ => {}
    }
}

/// A batch is one run's atomic unit, on every backend.
///
/// Everything an append does — the fence, the seal check, the head read, the
/// chain each record links into — is scoped to one run, so a record naming
/// another run would be sealed into the wrong chain under the wrong fence.
/// The embedded store refused this loudly; the shared one keyed the whole
/// batch under its first record's run and wrote the rest into it silently,
/// which is worse than either accepting or refusing on purpose.
async fn a_batch_spans_one_run(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let a = RunId::generate();
    let b = RunId::generate();
    let Ok(lease) = store.acquire(a, "conformance", LEASE).await else {
        return;
    };
    if store
        .append(lease.epoch, vec![admitted(a), admitted(b)])
        .await
        .is_ok()
    {
        r.record(
            "atomicity",
            "a batch spanning two runs was accepted — the second run's record was \
             sealed into the first run's chain under the first run's fence",
        );
        return;
    }
    for run in [a, b] {
        if store.read(run, 1).await.map_or(0, |v| v.len()) != 0 {
            r.record(
                "atomicity",
                "a refused cross-run batch still left records behind",
            );
        }
    }
}

/// A seal is fenced like any other write.
///
/// A seal freezes a chain forever and publishes its head as a Merkle leaf. A
/// backend that seals without checking the epoch lets a fenced zombie freeze a
/// run its replacement is still extending — every later checkpoint then
/// attests a head the true history has moved past.
async fn a_stale_epoch_cannot_seal(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    if store
        .append(lease.epoch, vec![admitted(run)])
        .await
        .is_err()
    {
        return;
    }
    if store.seal(run, lease.epoch + 7, "succeeded").await.is_ok() {
        r.record(
            "fencing",
            "a seal under a fabricated epoch was accepted — a fenced writer froze \
             a chain it does not own",
        );
    }
    if let Err(e) = store.seal(run, lease.epoch, "succeeded").await {
        r.record(
            "fencing",
            format!("a seal under the current epoch was refused: {e}"),
        );
    }
}

async fn a_rejected_batch_writes_nothing(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    let k = key(3);
    let _ = store.append(lease.epoch, vec![started(run, k)]).await;
    let before = store.read(run, 1).await.map_or(0, |v| v.len());

    // A batch whose *second* record violates exactly-once. The first is
    // perfectly legal, and must not survive.
    let batch = vec![admitted(run), started(run, k)];
    if store.append(lease.epoch, batch).await.is_ok() {
        r.record(
            "atomicity",
            "a batch containing a duplicate effect start was accepted",
        );
        return;
    }
    let after = store.read(run, 1).await.map_or(0, |v| v.len());
    if after != before {
        r.record(
            "atomicity",
            format!(
                "a rejected batch left {} record(s) behind. The whole batch commits or none \
                 of it does — a partially written step describes something that never \
                 happened",
                after - before
            ),
        );
    }
}

/// A quarantined run is findable by the person who has to deal with it.
///
/// Without this, the most serious conclusion this runtime reaches would
/// produce only a status, a log line and a counter — none of which can be
/// queried, and a run started with `spawn` returns before the status exists.
/// Every other backlog is findable by whoever must clear it, and this is the
/// shape production studies of agent runtimes report as the most common
/// failure: not an undetected fault, but a detected one whose signal never
/// reaches a human.
///
/// Both directions, because either alone passes for the wrong reason: a scan
/// returning nothing satisfies "no other outcome", and one returning everything
/// satisfies "finds its own".
async fn sealed_runs_are_findable_by_how_they_ended(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;

    let mut quarantined = Vec::new();
    for (outcome, count) in [("quarantined", 2usize), ("succeeded", 1)] {
        for _ in 0..count {
            let run = RunId::generate();
            let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
                return;
            };
            // The index derives from the chain's conclusion record, not from
            // the seal — the executor writes the record, then seals.
            let _ = store
                .append(lease.epoch, vec![admitted(run), concluded(run, outcome)])
                .await;
            if store.seal(run, lease.epoch, outcome).await.is_err() {
                return;
            }
            if outcome == "quarantined" {
                quarantined.push(run);
            }
        }
    }

    match store.runs_by_outcome("quarantined", 100).await {
        Ok(found) => {
            if found.len() != 2 {
                r.record(
                    "outcome index",
                    format!(
                        "runs_by_outcome must find both quarantined runs, got {}",
                        found.len()
                    ),
                );
            }
            if found.iter().any(|run| !quarantined.contains(run)) {
                r.record(
                    "outcome index",
                    "runs_by_outcome returned a run that ended some other way".to_owned(),
                );
            }
        }
        Err(e) => r.record("outcome index", format!("runs_by_outcome failed: {e}")),
    }

    // The bound is applied rather than advisory — **and it keeps the newest**.
    //
    // Checking only the count is a check that proves nothing: a page limit in
    // ascending order returns the right *number* of runs and the wrong ones,
    // so a plane whose backlog exceeds one page shows the same list forever and
    // the quarantine that just happened never appears. `quarantined` is in
    // creation order, so the last element is the one a bounded query must keep.
    let newest = quarantined.last().copied();
    match store.runs_by_outcome("quarantined", 1).await {
        Ok(found) if found.len() == 1 && Some(found[0]) == newest => {}
        Ok(found) if found.len() == 1 => r.record(
            "outcome index",
            "runs_by_outcome(limit=1) kept the oldest run — a bounded query in \
             ascending order never surfaces the quarantine that just happened"
                .to_owned(),
        ),
        Ok(found) => r.record(
            "outcome index",
            format!("runs_by_outcome(limit=1) returned {} runs", found.len()),
        ),
        Err(e) => r.record(
            "outcome index",
            format!("runs_by_outcome(limit=1) failed: {e}"),
        ),
    }
}

/// The outcome index follows the *last* conclusion, not the first.
///
/// A failed run is open: its conclusion is in the chain, so it must be
/// findable by whoever clears failures — and when it is resumed and succeeds,
/// it must *move*. An index that keeps the first conclusion lists a run as
/// failed for the rest of its life, and a backlog page that never drains is
/// worse than no page, because a wrong answer reads exactly like a right one.
async fn the_outcome_index_follows_the_last_conclusion(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };

    // A failed conclusion is indexed without any seal: failure leaves the run
    // open and resumable, and openness must not cost findability.
    if let Err(e) = store
        .append(lease.epoch, vec![admitted(run), concluded(run, "failed")])
        .await
    {
        r.record(
            "outcome index",
            format!("appending a conclusion failed: {e}"),
        );
        return;
    }
    match store.runs_by_outcome("failed", 10).await {
        Ok(found) if found.contains(&run) => {}
        Ok(_) => r.record(
            "outcome index",
            "a failed conclusion was not indexed — an unsealed run's failure is \
             a finding nobody can find"
                .to_owned(),
        ),
        Err(e) => r.record("outcome index", format!("runs_by_outcome failed: {e}")),
    }

    // The run resumes and succeeds: it must leave the failed listing and enter
    // the succeeded one — one run, one row, the latest answer.
    if let Err(e) = store
        .append(lease.epoch, vec![concluded(run, "succeeded")])
        .await
    {
        r.record("outcome index", format!("re-concluding failed: {e}"));
        return;
    }
    let _ = store.seal(run, lease.epoch, "succeeded").await;

    match store.runs_by_outcome("failed", 10).await {
        Ok(found) if found.contains(&run) => r.record(
            "outcome index",
            "a run that later succeeded is still listed as failed — the backlog \
             page never drains, and a wrong answer reads exactly like a right one"
                .to_owned(),
        ),
        Ok(_) => {}
        Err(e) => r.record("outcome index", format!("runs_by_outcome failed: {e}")),
    }
    match store.runs_by_outcome("succeeded", 10).await {
        Ok(found) if found.contains(&run) => {}
        Ok(_) => r.record(
            "outcome index",
            "the re-conclusion did not move the run into the succeeded listing".to_owned(),
        ),
        Err(e) => r.record("outcome index", format!("runs_by_outcome failed: {e}")),
    }
}

/// A sealed run's journal is frozen — an append after the seal is refused.
///
/// The Merkle leaf is the chain head *at seal time*. The fence answers "who
/// owns this run", and the caller that sealed it is exactly who still holds
/// the current epoch — so without this check, that caller can advance the true
/// head past the leaf every checkpoint attests, and an inclusion proof then
/// vouches for a prefix of a history that kept growing. The executor's own
/// refusal to resume a closed run is application logic a future caller can
/// bypass; the store's refusal is the constraint that cannot be.
async fn a_sealed_run_refuses_appends(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    // Positive half: the same batch shape is accepted before the seal, so a
    // store that refuses everything cannot pass by refusing this too.
    if let Err(e) = store.append(lease.epoch, vec![admitted(run)]).await {
        r.record("seal", format!("append before seal failed: {e}"));
        return;
    }
    if let Err(e) = store.seal(run, lease.epoch, "succeeded").await {
        r.record("seal", format!("seal failed: {e}"));
        return;
    }
    let Ok(frozen) = store.head(run).await else {
        r.record("seal", "head() failed after seal".to_owned());
        return;
    };

    match store
        .append(lease.epoch, vec![started(run, key(201))])
        .await
    {
        Ok(_) => r.record(
            "seal",
            "a sealed run accepted an append under the current epoch — the true \
             head has moved past the leaf every checkpoint attests"
                .to_owned(),
        ),
        Err(crate::core::StoreError::RunSealed { .. }) => {}
        Err(e) => r.record(
            "seal",
            format!(
                "an append after seal was refused, but as '{e}' rather than \
                 RunSealed — a caller cannot tell a frozen run from a store fault"
            ),
        ),
    }

    // And nothing was written: the refusal must leave the chain head exactly
    // where the seal froze it, or the checkpoint still lies by one record.
    match store.head(run).await {
        Ok(after) if after.hash == frozen.hash && after.seq == frozen.seq => {}
        Ok(_) => r.record(
            "seal",
            "the refused append still moved the chain head".to_owned(),
        ),
        Err(e) => r.record("seal", format!("head() failed after refusal: {e}")),
    }
}

/// A case's history is a scan over one matter, and only that matter.
///
/// The question is *show me everything about this matter*. Answering it by
/// listing the case's runs and reading each is a join that also **misses**
/// every record written by a run the case does not own — which is exactly what
/// a sweep is, since one tick may act on several cases and belongs to none.
///
/// Both halves are checked, because either alone passes for the wrong reason: a
/// scan that returns nothing satisfies "no foreign records", and a scan that
/// returns everything satisfies "finds its own".
async fn a_case_scan_selects_one_matter(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let mine = CaseId::generate();
    let theirs = CaseId::generate();

    // Two records for one matter and one for another, written by runs that
    // belong to neither — the shape a sweep produces.
    for (case, count) in [(mine, 2), (theirs, 1)] {
        let run = RunId::generate();
        let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
            return;
        };
        for _ in 0..count {
            let _ = store
                .append(lease.epoch, vec![admitted(run).case(case)])
                .await;
        }
    }

    match store.case_history(mine, 100).await {
        Ok(records) => {
            if records.len() != 2 {
                r.record(
                    "case history",
                    format!(
                        "case_history must return this matter's records, got {} of 2",
                        records.len()
                    ),
                );
            }
            if records.iter().any(|rec| rec.body.case != Some(mine)) {
                r.record(
                    "case history",
                    "case_history returned a record belonging to another matter".to_owned(),
                );
            }
        }
        Err(e) => r.record("case history", format!("case_history failed: {e}")),
    }

    // The bound is applied rather than advisory.
    match store.case_history(mine, 1).await {
        Ok(records) if records.len() == 1 => {}
        Ok(records) => r.record(
            "case history",
            format!("case_history(limit=1) returned {} records", records.len()),
        ),
        Err(e) => r.record("case history", format!("case_history(limit=1) failed: {e}")),
    }
}

async fn read_starts_where_it_is_told(fresh: Factory<'_>, r: &mut Report) {
    r.checked += 1;
    let store = fresh().await;
    let run = RunId::generate();
    let Ok(lease) = store.acquire(run, "conformance", LEASE).await else {
        return;
    };
    for _ in 0..4 {
        let _ = store.append(lease.epoch, vec![admitted(run)]).await;
    }
    match store.read(run, 3).await {
        Ok(records) => {
            let seqs: Vec<u64> = records.iter().map(Record::seq).collect();
            if seqs != vec![3, 4] {
                r.record(
                    "chaining",
                    format!("read(from=3) must return seq 3 onward, got {seqs:?}"),
                );
            }
        }
        Err(e) => r.record("chaining", format!("read(from=3) failed: {e}")),
    }
}

/// The standing-authority contract, applied to any backend.
///
/// One battery for both stores, because the guarantees here are the sort that
/// hold on a single-writer embedded store almost by accident and have to be
/// *built* on a shared one. Running it against redb proves the semantics;
/// running it against `PostgreSQL` proves the arbitration.
///
/// Panics with the property that failed, so a backend under development gets a
/// sentence rather than a diff of two structs.
///
/// # Panics
///
/// If the store violates any part of the contract.
#[allow(clippy::too_many_lines)]
pub async fn authority(store: Arc<dyn crate::authority::AuthorityStore>) {
    use crate::authority::{AuthorityError, AuthorityId, StandingAuthority};
    use crate::core::{EffectKey, Spend, Timestamp};

    let key = |n: u8| EffectKey::from_hex(&format!("{n:064x}")).expect("32 bytes of hex");
    let at = Timestamp::UNIX_EPOCH;

    // ── Draws accumulate, and the ceiling is cumulative ──────────────────────
    let id = AuthorityId::new("conformance-accumulate");
    store
        .issue(&StandingAuthority::new(
            "conformance-accumulate",
            "conformance",
            Spend::money(1_000),
        ))
        .await
        .expect("issue");

    let first = store
        .draw(&id, key(1), Spend::money(600), at)
        .await
        .expect("a draw within the ceiling");
    assert_eq!(
        first.remaining,
        Spend::money(400),
        "the receipt must report what is left after this draw"
    );

    let refused = store
        .draw(&id, key(2), Spend::money(500), at)
        .await
        .expect_err("the ceiling is cumulative across draws, not per draw");
    assert!(
        matches!(refused, AuthorityError::Exhausted { .. }),
        "an over-draw must be Exhausted, not {refused:?}"
    );

    // A refusal consumes nothing, or probing the ceiling would drain it.
    store
        .draw(&id, key(3), Spend::money(400), at)
        .await
        .expect("exactly the remainder is still available after a refusal");

    // ── A retry under one dispatch key consumes once ─────────────────────────
    let id = AuthorityId::new("conformance-retry");
    store
        .issue(&StandingAuthority::new(
            "conformance-retry",
            "conformance",
            Spend::money(1_000),
        ))
        .await
        .expect("issue");

    let original = store
        .draw(&id, key(10), Spend::money(300), at)
        .await
        .expect("draw");
    let repeat = store
        .draw(&id, key(10), Spend::money(300), at)
        .await
        .expect("a retry is not a second draw");
    assert_eq!(
        original, repeat,
        "a retried draw must return its original receipt"
    );
    let state = store.state(&id).await.expect("state").expect("issued");
    assert_eq!(
        state.drawn,
        Spend::money(300),
        "a retry spent the authority twice"
    );
    assert_eq!(state.draws, 1, "a retry counted as a second draw");

    // ── Revocation refuses new draws and preserves landed ones ───────────────
    store
        .revoke(&id, "conformance withdrew it", at)
        .await
        .expect("revoke");
    let after = store
        .draw(&id, key(10), Spend::money(300), at)
        .await
        .expect("a draw that already landed must stay landed on retry");
    assert_eq!(after, original);

    let refused = store
        .draw(&id, key(11), Spend::money(1), at)
        .await
        .expect_err("a new draw against a revoked authority");
    assert!(
        matches!(refused, AuthorityError::Revoked { .. }),
        "revoked must be distinguishable from exhausted, got {refused:?}"
    );

    // Idempotent, and the first reason stands.
    store
        .revoke(&id, "a second, different reason", at)
        .await
        .expect("revoking twice is a retry");
    let state = store.state(&id).await.expect("state").expect("issued");
    let revocation = state.revoked.expect("revoked");
    assert_eq!(
        revocation.reason, "conformance withdrew it",
        "the first reason must stand; overwriting loses why it was withdrawn"
    );

    // ── Terms are immutable ─────────────────────────────────────────────────
    let terms = StandingAuthority::new("conformance-terms", "conformance", Spend::money(500));
    store.issue(&terms).await.expect("issue");
    store
        .issue(&terms)
        .await
        .expect("an identical re-issue is a retried deploy");
    let conflict = store
        .issue(&StandingAuthority::new(
            "conformance-terms",
            "conformance",
            Spend::money(999_999),
        ))
        .await
        .expect_err("a ceiling must not be editable under whoever agreed to it");
    assert!(
        matches!(conflict, AuthorityError::AlreadyIssued(_)),
        "got {conflict:?}"
    );

    // ── Expiry is evaluated against the instant handed in ───────────────────
    store
        .issue(
            &StandingAuthority::new("conformance-expiry", "conformance", Spend::money(500))
                .expires_at(Timestamp::from_unix_timestamp(1_000).expect("representable")),
        )
        .await
        .expect("issue");
    let id = AuthorityId::new("conformance-expiry");
    store
        .draw(
            &id,
            key(20),
            Spend::money(1),
            Timestamp::from_unix_timestamp(999).expect("representable"),
        )
        .await
        .expect("before expiry the authority stands");
    let expired = store
        .draw(
            &id,
            key(21),
            Spend::money(1),
            Timestamp::from_unix_timestamp(2_000).expect("representable"),
        )
        .await
        .expect_err("past its expiry");
    assert!(
        matches!(expired, AuthorityError::Expired { .. }),
        "expiry must read the caller's instant, not a store clock; got {expired:?}"
    );

    // ── A draw ceiling bounds separately from a spend ceiling ───────────────
    store
        .issue(
            &StandingAuthority::new("conformance-draws", "conformance", Spend::money(10_000))
                .max_draws(1),
        )
        .await
        .expect("issue");
    let id = AuthorityId::new("conformance-draws");
    store
        .draw(&id, key(30), Spend::money(1), at)
        .await
        .expect("first");
    let spent = store
        .draw(&id, key(31), Spend::money(1), at)
        .await
        .expect_err("one draw was all it permitted");
    assert!(
        matches!(spent, AuthorityError::DrawsSpent { .. }),
        "a draw ceiling must refuse with money still left; got {spent:?}"
    );

    // ── An unissued authority is Unknown, not a silent allow ────────────────
    let unknown = store
        .draw(
            &AuthorityId::new("conformance-never-issued"),
            key(40),
            Spend::money(1),
            at,
        )
        .await
        .expect_err("never issued");
    assert!(
        matches!(unknown, AuthorityError::Unknown(_)),
        "got {unknown:?}"
    );
    assert!(
        store
            .state(&AuthorityId::new("conformance-never-issued"))
            .await
            .expect("state")
            .is_none()
    );

    // ── Concurrent carriers of ONE dispatch key all get the one receipt ──────
    //
    // The sequential retry case above cannot catch the racing form: a store
    // that checks the receipt before taking its lock lets two carriers of the
    // same key both pass the check, and the loser then re-evaluates the ceiling
    // *after* the winner spent it — refusing `Exhausted` for a draw that
    // stands, or tripping the receipt table's unique key. The ceiling here
    // covers exactly one such draw, so any double-spend or spurious refusal is
    // observable rather than absorbed by slack.
    //
    // The runtime's fencing cannot produce this race, which is exactly why the
    // battery must: the trait promises idempotence by key without qualifying
    // who carries it.
    let id = AuthorityId::new("conformance-racing-retry");
    store
        .issue(&StandingAuthority::new(
            "conformance-racing-retry",
            "conformance",
            Spend::money(500),
        ))
        .await
        .expect("issue");

    let carriers: Vec<_> = (0..8)
        .map(|_| {
            let store = Arc::clone(&store);
            let id = id.clone();
            tokio::spawn(async move { store.draw(&id, key(50), Spend::money(500), at).await })
        })
        .collect();
    let mut receipts = Vec::new();
    for carrier in carriers {
        receipts.push(
            carrier
                .await
                .expect("a carrier panicked")
                .expect("every carrier of the one key must get the receipt, not a refusal"),
        );
    }
    for r in &receipts {
        assert_eq!(
            *r, receipts[0],
            "two carriers of one dispatch key were given different receipts"
        );
    }
    let state = store.state(&id).await.expect("state").expect("issued");
    assert_eq!(
        state.drawn,
        Spend::money(500),
        "racing retries spent the authority more than once"
    );
    assert_eq!(state.draws, 1, "racing retries counted as several draws");
}