//! The suite's **reference row set** — the INV-1…6 invariant families
//! and the two constructor-mode rows, landed by the mem column as the
//! required set every other engine's rows mirror (ADR-011 §2's
//! contract-reference role; store-api.md §Invariants 7;
//! tests/conformance/README.md §The reference row set is the recorded
//! inventory).
//!
//! Row anatomy (README §Row anatomy): one named, independently-runnable
//! property per row — an async function taking a [`StoreFactory`]
//! (never a store), asserting by panicking on violation (a failing row
//! is a test failure naming the property); the content matrix and both
//! put paths loop in-body. Each row owns its factory lifecycle:
//! `open()` at the top (a fresh, isolated store per call — factory.rs'
//! contract), the assertions, then `teardown()` before returning; a
//! failing assertion panics first and skips teardown, which is exactly
//! why `teardown()` is contract-required to be idempotent and safe
//! against an abandoned instance (the mem column's structural tests
//! exercise that path).
//!
//! The rows are engine-uniform by construction — none branches on the
//! factory or any engine. The one deliberate exception is the two
//! constructor-mode rows' **precondition** (`store.mode()`): a
//! mode-scoped property runs against the configuration it names, and
//! `mode()` is the facade contract's own introspection (ADR-011 §4),
//! not an engine branch.
//!
//! Every row is ADR-stamped ([`crate::stamp`] is the convention's
//! anchor; `grep -rn "Contract stamp:" tests/conformance/` is the
//! inventory) and traces to a backlog entry ([`crate::backlog`]).

#![allow(dead_code)]

use std::collections::BTreeSet;
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::pin::Pin as StdPin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;

use parking_lot::Mutex;
use sha2::{Digest as _, Sha256};
use tokio::io::{AsyncRead, ReadBuf};

use crate::factory::StoreFactory;
use alkblobs::{
    Algorithm, EntryMeta, Error, GcAbortCause, Key, LivenessSource, ProtectCallback, Store,
    StoreMode, DEFAULT_DISPATCH_THRESHOLD,
};

/// The shared fixtures the rows run on — set-backed liveness sources,
/// a set-backed protect callback, and the stream fixtures for the
/// failure legs. None of them knows an engine.
///
/// A set-backed liveness source — the live-shared seam's test shape:
/// the store consults the same set the row mutates, so answers are
/// live-shared, never registration snapshots. `pub(crate)`: the
/// suite's other columns reuse these fixtures.
#[derive(Default)]
pub(crate) struct SetSource {
    live: Mutex<BTreeSet<Key>>,
}

impl SetSource {
    pub(crate) fn register(store: &Store) -> Arc<SetSource> {
        let source = Arc::new(SetSource::default());
        store.register_liveness_source(source.clone());
        source
    }

    pub(crate) fn add(&self, key: Key) {
        self.live.lock().insert(key);
    }

    pub(crate) fn remove(&self, key: &Key) {
        self.live.lock().remove(key);
    }
}

impl LivenessSource for SetSource {
    fn live_keys(&self) -> Vec<Key> {
        self.live.lock().iter().cloned().collect()
    }

    fn contains(&self, key: &Key) -> bool {
        self.live.lock().contains(key)
    }
}

/// A set-backed protect callback; `None` models the failed-to-answer
/// case (abort-as-data on the safe side).
#[derive(Default)]
pub(crate) struct SetProtect {
    set: Mutex<Option<Vec<Key>>>,
}

impl SetProtect {
    pub(crate) fn set_keys(&self, keys: Vec<Key>) {
        *self.set.lock() = Some(keys);
    }

    pub(crate) fn fail(&self) {
        *self.set.lock() = None;
    }
}

impl ProtectCallback for SetProtect {
    fn protect(&self) -> Option<Vec<Key>> {
        self.set.lock().clone()
    }
}

/// A protect callback that blocks mid-consult — the fixture for the
/// sweep-lease contended window: it announces its entry on a channel,
/// waits on a second, then answers with an empty protection set.
/// `pub(crate)`: the suite's other columns reuse it.
pub(crate) struct LatchProtect {
    pub(crate) entered: std::sync::mpsc::Sender<()>,
    pub(crate) release: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
}

impl ProtectCallback for LatchProtect {
    fn protect(&self) -> Option<Vec<Key>> {
        self.entered.send(()).ok();
        let receiver = self.release.lock().take();
        if let Some(receiver) = receiver {
            let _ = receiver.recv();
        }
        Some(Vec::new())
    }
}

/// A stream cut at controlled chunk boundaries (`sizes`, in order;
/// tail data falls out whole) — makes multi-read loops, mid-stream
/// overflow, and staged continuation observable past the put loop's
/// own chunking. `pub(crate)`: the suite's other columns reuse it.
pub(crate) struct Chunked {
    data: Vec<u8>,
    cursor: usize,
    sizes: std::collections::VecDeque<usize>,
}

impl Chunked {
    pub(crate) fn new(data: &[u8], sizes: &[usize]) -> Chunked {
        Chunked {
            data: data.to_vec(),
            cursor: 0,
            sizes: sizes.iter().copied().collect(),
        }
    }
}

impl AsyncRead for Chunked {
    fn poll_read(
        mut self: StdPin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        if self.cursor >= self.data.len() || buf.remaining() == 0 {
            return Poll::Ready(Ok(()));
        }
        let remaining = self.data.len() - self.cursor;
        let wanted = self
            .sizes
            .front()
            .copied()
            .unwrap_or(remaining)
            .min(remaining)
            .min(buf.remaining());
        buf.put_slice(&self.data[self.cursor..self.cursor + wanted]);
        self.cursor += wanted;
        if let Some(front) = self.sizes.front_mut() {
            *front -= wanted;
            if *front == 0 {
                self.sizes.pop_front();
            }
        }
        Poll::Ready(Ok(()))
    }
}

/// A stream that delivers `fail_after` bytes then errors mid-stream —
/// the non-restartable source's failure (the failure-discard path's
/// injectable shape). `pub(crate)`: the suite's other columns reuse it.
pub(crate) struct Failing {
    data: Vec<u8>,
    cursor: usize,
    fail_after: usize,
}

impl Failing {
    pub(crate) fn new(data: &[u8], fail_after: usize) -> Failing {
        Failing {
            data: data.to_vec(),
            cursor: 0,
            fail_after,
        }
    }
}

impl AsyncRead for Failing {
    fn poll_read(
        mut self: StdPin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let message = "stream failure (conformance fixture)";
        if self.cursor >= self.fail_after {
            return Poll::Ready(Err(std::io::Error::other(message)));
        }
        let wanted = (self.data.len() - self.cursor)
            .min(self.fail_after - self.cursor)
            .min(buf.remaining());
        if wanted == 0 {
            return Poll::Ready(Err(std::io::Error::other(message)));
        }
        buf.put_slice(&self.data[self.cursor..self.cursor + wanted]);
        self.cursor += wanted;
        Poll::Ready(Ok(()))
    }
}

/// A stream nobody may ever read — the fixture proving kv-only's
/// known-length rejection happens before any byte flows.
struct Unreadable;

impl AsyncRead for Unreadable {
    fn poll_read(
        self: StdPin<&mut Self>,
        _cx: &mut Context<'_>,
        _buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        Poll::Ready(Err(std::io::Error::other(
            "the Unreadable stream must never be read (conformance fixture)",
        )))
    }
}

/// The content matrix the round-trip family loops: the degenerate
/// empty, one byte, small text, and patterned binaries spanning the
/// put loop's multi-read range. Sizes are content-shape choices, not
/// engine or dispatch-internals knowledge.
fn matrix() -> Vec<Vec<u8>> {
    vec![
        Vec::new(),
        b"x".to_vec(),
        b"the conformance round-trip content".to_vec(),
        (0..137u32).map(|i| (i % 251) as u8).collect(),
        (0..4096u32).map(|i| (i % 199) as u8).collect(),
        (0..65537u32).map(|i| (i % 173) as u8).collect(),
    ]
}

/// A content-shape generator: `len` bytes cycling against `modulus` —
/// deterministic, size-tunable content. `pub(crate)`: the suite's
/// other columns reuse it.
pub(crate) fn patterned(len: usize, modulus: u32) -> Vec<u8> {
    (0..len as u32).map(|i| (i % modulus) as u8).collect()
}

/// One leg of the exact-count sweep assertions: a completed report's
/// counts are exact over the staged set (`candidates_staged ==
/// deleted + cancelled_by_arbitration`). `pub(crate)`: the suite's
/// other columns reuse it.
pub(crate) fn assert_exact_counts(report: &alkblobs::SweepReport) {
    assert_eq!(
        report.candidates_staged,
        report.deleted + report.cancelled_by_arbitration,
        "the completed sweep's counts are exact over the staged set"
    );
}

/// Every regular file under `root` outside any `staging` directory —
/// the committed-tree counter of the medium-backed columns' on-disk
/// placement oracle. With a kv tier that never touches this disk it is
/// exactly one per committed large-tier entry; a SQL-backed kv tier
/// distinguishes by medium instead (its entries live in the database
/// file, so this count is still exactly the large tier's residency).
/// `pub(crate)`: the suite's other columns reuse it.
pub(crate) fn committed_file_count(root: &Path) -> usize {
    fn walk(dir: &Path, out: &mut Vec<PathBuf>) {
        let Ok(entries) = std::fs::read_dir(dir) else {
            return;
        };
        for entry in entries.flatten() {
            let Ok(kind) = entry.file_type() else {
                continue;
            };
            let path = entry.path();
            if kind.is_dir() {
                if path.file_name().and_then(|n| n.to_str()) == Some("staging") {
                    continue;
                }
                walk(&path, out);
            } else if kind.is_file() {
                out.push(path);
            }
        }
    }
    let mut files = Vec::new();
    walk(root, &mut files);
    files.len()
}

/// Files in the large tier's staging area — the residue counter of the
/// on-disk stage-hygiene exact-count form (zero after every failure
/// form and every success form; a staged artifact lives there only
/// mid-put). `pub(crate)`: the suite's other columns reuse it.
pub(crate) fn staging_file_count(root: &Path) -> usize {
    match std::fs::read_dir(root.join("staging")) {
        Ok(entries) => entries.count(),
        Err(_) => 0,
    }
}

/// The ADR-006 slice digest, computed independently of the crate —
/// SHA-256 over exactly the returned bytes (the out-of-band
/// verification the convention requires; callers verify the same way).
/// `pub(crate)`: the suite's other columns reuse it.
pub(crate) fn slice_digest(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

/// INV-1: whole-blob put/get round-trips byte-identical under the
/// canonical derivation — the content matrix loops both put paths
/// (known-length, unknown-length) in-body; every put's key is the
/// canonical derivation over the content (the git-interop test
/// remains the derivation's source of truth), and `get` returns the
/// pinned `(len, stream)` shape whole and windowed.
///
/// Contract stamp: store-api.md §Invariants 1; ADR-002 (canonical
/// derivation; interop with real git is the source of truth); ADR-003
/// (the two put paths behind one signature).
pub async fn round_trip_puts_and_gets_are_byte_identical<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");

    for content in matrix() {
        let len = content.len() as u64;
        let canonical = Key::derive(&content, len);
        for declared in [Some(len), None] {
            let (key, pin) = store
                .put(declared, Cursor::new(content.as_slice()))
                .await
                .expect("put publishes");
            assert_eq!(key, canonical, "the canonical derivation addresses the put");
            let (got_len, blob) = store
                .get(&key.as_bytes())
                .await
                .expect("get succeeds")
                .expect("the entry is present");
            assert_eq!(got_len, len, "the tuple carries the entry length");
            assert_eq!(blob.len(), len, "the stream's metadata length agrees");
            assert_eq!(
                blob.read_at(0, len).await.expect("the whole window reads"),
                content,
                "whole-blob round-trip is byte-identical"
            );
            if len > 2 {
                assert_eq!(
                    blob.read_at(1, len - 2)
                        .await
                        .expect("the inner window reads"),
                    content[1..(len - 2) as usize + 1],
                    "a bounded window reads the matching inner slice"
                );
            }
            drop(pin);
        }
    }

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-2: identical content put twice — through the same put path and
/// across the two paths — is one pool entry, and the re-put's `Pin`
/// protects exactly as a fresh put's does. One-pool-entry is asserted
/// through sweep outcomes (the INV-5 posture), never by comparing
/// enumerated lists.
///
/// Contract stamp: store-api.md §Invariants 2; ADR-003 (whole-value
/// put is CAS first-writer-wins); ADR-005 (a re-put refcounts the
/// existing entry under the same arbitration lock).
pub async fn dedup_identical_content_is_one_pool_entry<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let content = b"deduplicated content".to_vec();
    let canonical = Key::derive(&content, content.len() as u64);

    let (key, first) = store
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fresh put publishes");
    let (reput, second) = store
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the same-path re-put publishes");
    assert_eq!(key, canonical);
    assert_eq!(reput, canonical, "one derivation, one address");
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(
        report.candidates_staged, 0,
        "one pool entry, both refcounts composing under the same lock"
    );
    drop(first);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.candidates_staged, 0,
        "the re-put's pin protects exactly as a fresh put's"
    );
    drop(second);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "exactly one pool entry all along");
    assert!(!store.has(&canonical.as_bytes()).await.unwrap());

    let cross = b"deduplicated across the put paths".to_vec();
    let cross_canonical = Key::derive(&cross, cross.len() as u64);
    let (key, first) = store
        .put(Some(cross.len() as u64), Cursor::new(cross.as_slice()))
        .await
        .expect("the known-length put publishes");
    let (reput, second) = store
        .put(None, Cursor::new(cross.as_slice()))
        .await
        .expect("the unknown-length re-put publishes");
    assert_eq!(key, cross_canonical);
    assert_eq!(reput, cross_canonical, "one pool entry across the paths");
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 0, "both refcounts compose");
    drop(second);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.candidates_staged, 0,
        "the re-put's pin protects exactly as a fresh put's (other drop order)"
    );
    drop(first);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "exactly one pool entry all along");
    assert!(!store.has(&cross_canonical.as_bytes()).await.unwrap());

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-3: stage hygiene — every put-path failure leaves zero residue
/// and every success leaves exactly one committed entry, over both
/// put paths and both injectable failure modes (a mid-stream stream
/// failure and a declared-length mismatch). The kv tier's stage form
/// is degenerate (nothing stages outside a flush), so residue is the
/// pool itself, observed through exact-count sweep outcomes.
///
/// Contract stamp: store-api.md §Invariants 3; ADR-010 §1 (the
/// staged-put contract: commit publishes, every failure path
/// discards); ADR-003 (stage-then-commit, the kv form degenerate).
pub async fn stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success<F: StoreFactory>(
    factory: &F,
) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let content = patterned(200, 167);
    let canonical = Key::derive(&content, content.len() as u64);

    let err = store
        .put(Some(content.len() as u64), Failing::new(&content, 40))
        .await
        .expect_err("the failed mid-stream put rejects");
    assert!(
        matches!(err, Error::Io(_)),
        "the stream failure surfaces as Io: {err}"
    );
    let err = store
        .put(None, Failing::new(&content, 40))
        .await
        .expect_err("the failed unknown-length put rejects");
    assert!(
        matches!(err, Error::Io(_)),
        "the stream failure surfaces as Io: {err}"
    );
    let err = store
        .put(Some(100), Cursor::new(content.as_slice()))
        .await
        .expect_err("the over-fed mismatch rejects");
    assert_eq!(err, Error::Verification, "declared-len mismatch rejects");
    let err = store
        .put(Some(400), Cursor::new(content.as_slice()))
        .await
        .expect_err("the under-fed mismatch rejects");
    assert_eq!(err, Error::Verification, "declared-len mismatch rejects");

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "zero residue after every failure, both put paths, both failure modes"
    );

    let (key, pin) = store
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the known-length put publishes");
    assert_eq!(key, canonical);
    drop(pin);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (report.candidates_staged, report.deleted),
        (1, 1),
        "exactly one committed entry on success"
    );
    assert!(!store.has(&canonical.as_bytes()).await.unwrap());

    let (key, pin) = store
        .put(None, Cursor::new(content.as_slice()))
        .await
        .expect("the unknown-length put publishes");
    assert_eq!(key, canonical);
    drop(pin);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (report.candidates_staged, report.deleted),
        (1, 1),
        "exactly one committed entry on the other put path too"
    );
    assert!(!store.has(&canonical.as_bytes()).await.unwrap());

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, happy shape: with liveness correctly registered on a mixed
/// pool — a source-live entry, a pinned entry, a protect-callback-
/// named entry, and two dead entries (one per put path) — the sweep's
/// counts are exact and only the dead entries delete. The protected
/// entries stay byte-identical.
///
/// Contract stamp: store-api.md §Invariants 4; ADR-005 (three liveness
/// marks; delete windows with arbitrate-at-delete-time); ADR-012 §1
/// (sweeps are never errors — the report is the outcome record).
pub async fn sweep_counts_are_exact_across_the_full_protection_set<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    let source = SetSource::register(&store);
    let protect = Arc::new(SetProtect::default());
    store.register_protect_callback(protect.clone());

    let source_live = b"kept by the registered source".to_vec();
    let (key_live, _) = store
        .put(
            Some(source_live.len() as u64),
            Cursor::new(source_live.as_slice()),
        )
        .await
        .expect("the source-live entry publishes");
    source.add(key_live.clone());

    let pinned = b"kept by the pin".to_vec();
    let (key_pinned, pin_pinned) = store
        .put(None, Cursor::new(pinned.as_slice()))
        .await
        .expect("the pinned entry publishes");

    let protected = b"named by the protect callback".to_vec();
    let (key_protected, _) = store
        .put(
            Some(protected.len() as u64),
            Cursor::new(protected.as_slice()),
        )
        .await
        .expect("the protect-named entry publishes");
    protect.set_keys(vec![key_protected.clone()]);

    let dead_one = b"a dead witness from the known-length path".to_vec();
    let (_, pin) = store
        .put(
            Some(dead_one.len() as u64),
            Cursor::new(dead_one.as_slice()),
        )
        .await
        .expect("the first dead entry publishes");
    drop(pin);
    let dead_two = b"a dead witness from the unknown-length path".to_vec();
    let (_, pin) = store
        .put(None, Cursor::new(dead_two.as_slice()))
        .await
        .expect("the second dead entry publishes");
    drop(pin);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None, "correct liveness completes the sweep");
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (2, 2, 0),
        "only the two dead entries stage and delete"
    );
    assert_exact_counts(&report);

    assert!(store.has(&key_live.as_bytes()).await.unwrap());
    assert!(store.has(&key_pinned.as_bytes()).await.unwrap());
    assert!(store.has(&key_protected.as_bytes()).await.unwrap());
    for (key, content) in [
        (&key_live, &source_live),
        (&key_pinned, &pinned),
        (&key_protected, &protected),
    ] {
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("the protected entry is present");
        assert_eq!(
            blob.read_at(0, len).await.expect("the window reads"),
            content.as_slice(),
            "the protected entry is byte-identical after the sweep"
        );
    }

    drop(pin_pinned);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(
        (report.candidates_staged, report.deleted),
        (1, 1),
        "with the pin dropped exactly the formerly pinned entry dies"
    );
    assert!(!store.has(&key_pinned.as_bytes()).await.unwrap());
    assert!(store.has(&key_live.as_bytes()).await.unwrap());
    assert!(store.has(&key_protected.as_bytes()).await.unwrap());

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, abort shape: the safe default — a sweep invoked with zero
/// registered liveness sources aborts with `NoLivenessSources` in a
/// completed `Ok(SweepReport)` and deletes nothing.
///
/// Contract stamp: store-api.md §Invariants 4; store-api.md §Lifecycle
/// (the safe default); ADR-012 §1 (every abort is report data — never
/// an error, nothing deleted).
pub async fn sweep_without_liveness_sources_aborts_and_deletes_nothing<F: StoreFactory>(
    factory: &F,
) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    let content = b"unprotected while no source is registered".to_vec();
    let (key, pin) = store
        .put(None, Cursor::new(content.as_slice()))
        .await
        .expect("the entry publishes");
    drop(pin);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.aborted,
        Some(GcAbortCause::NoLivenessSources),
        "the safe default aborts as report data"
    );
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "an aborted sweep stages and deletes nothing"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "the entry survives the abort"
    );
    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the entry is present");
    assert_eq!(
        blob.read_at(0, len).await.expect("the window reads"),
        content
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, abort shape: a protection source that fails to answer
/// aborts the sweep with `ProtectFailed` in a completed
/// `Ok(SweepReport)` — the flaky source skips the sweep rather than
/// risk deletion, and nothing is deleted.
///
/// Contract stamp: store-api.md §Invariants 4; ADR-005 §Marks 3 (a
/// failed protection source takes the safe side); ADR-012 §1 (abort
/// is report data, never an error).
pub async fn sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing<F: StoreFactory>(
    factory: &F,
) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);
    let protect = Arc::new(SetProtect::default());
    store.register_protect_callback(protect.clone());
    protect.fail();

    let content = b"dead while the protection source fails to answer".to_vec();
    let (key, pin) = store
        .put(None, Cursor::new(content.as_slice()))
        .await
        .expect("the entry publishes");
    drop(pin);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.aborted,
        Some(GcAbortCause::ProtectFailed),
        "the failed protection source aborts as report data"
    );
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "an aborted sweep stages and deletes nothing"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "the entry survives the abort"
    );
    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the entry is present");
    assert_eq!(
        blob.read_at(0, len).await.expect("the window reads"),
        content
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, abort shape: the single-node sweep is single-flight — the
/// in-process analog of the sweep lease (ADR-013 §2). A second
/// sweeper contending behind a lease holder aborts with
/// `SweeperLock` in a completed `Ok(SweepReport)`, deleting nothing;
/// the holder's own sweep completes with exact counts.
///
/// Contract stamp: store-api.md §Invariants 4; ADR-012 §1 (abort is
/// report data); ADR-013 §2 (the named exclusive sweep lease — one
/// sweeper per pool, holdership ends at release); ADR-012 §5 (the
/// in-process single-flighter is the non-SQL form of that lease).
pub async fn sweep_is_single_flight_the_contended_second_aborts_as_data<F: StoreFactory>(
    factory: &F,
) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    store.register_protect_callback(Arc::new(LatchProtect {
        entered: entered_tx,
        release: Mutex::new(Some(release_rx)),
    }));

    let content = b"dead under the contended sweeps".to_vec();
    let (key, pin) = store
        .put(None, Cursor::new(content.as_slice()))
        .await
        .expect("the entry publishes");
    drop(pin);

    // The lease holder sweeps on its own task (its protect consult
    // blocks that task's thread mid-callback); this task contends and
    // then releases the holder.
    let store = Arc::new(store);
    let holder = tokio::spawn({
        let store = Arc::clone(&store);
        async move { store.sweep().await }
    });

    while entered_rx.recv_timeout(Duration::from_millis(50)).is_err() {
        tokio::task::yield_now().await;
    }

    let second = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        second.aborted,
        Some(GcAbortCause::SweeperLock),
        "the contended sweeper aborts as report data"
    );
    assert_eq!(
        (
            second.candidates_staged,
            second.deleted,
            second.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "the contended sweep stages and deletes nothing"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "the contended sweep deleted nothing"
    );
    release_tx.send(()).ok();

    let report = holder
        .await
        .expect("the holder task joined")
        .expect("sweeps are never errors");
    assert_eq!(report.aborted, None, "the lease holder completes");
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (1, 1, 0),
        "the lease holder's counts are exact"
    );
    assert_exact_counts(&report);
    assert!(
        !store.has(&key.as_bytes()).await.unwrap(),
        "the holder's deletion landed after the release"
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, the race: a sweep racing an in-flight pinned put never
/// deletes the protected entry — the pin is minted before the entry
/// can appear in any `list()`, so every schedule keeps the counts
/// exact and the published entry whole and byte-identical; the
/// surviving pin keeps protecting, and dropping it re-exposes the
/// entry to exactly one deletion. Both put paths race in-body.
///
/// Contract stamp: store-api.md §Invariants 4; ADR-005 (pin-before-
/// publish ordering; the delete window closes the sweep-vs-put race).
pub async fn sweep_racing_a_pinned_put_never_deletes_the_protected_entry<F: StoreFactory>(
    factory: &F,
) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let witness = b"the dead witness the first racing sweep deletes".to_vec();
    let (_, pin) = store
        .put(None, Cursor::new(witness.as_slice()))
        .await
        .expect("the witness publishes");
    drop(pin);

    let rounds: [(Vec<u8>, bool); 4] = [
        (b"racing round a, known path".to_vec(), true),
        (b"racing round b, unknown path".to_vec(), false),
        (patterned(4096, 191), true),
        (patterned(4096, 173), false),
    ];
    for (round, (content, known)) in rounds.iter().enumerate() {
        let declared = if *known {
            Some(content.len() as u64)
        } else {
            None
        };
        let (report, put) = tokio::join!(store.sweep(), async {
            store.put(declared, Cursor::new(content.as_slice())).await
        });
        let report = report.expect("sweeps are never errors");
        let (key, pin) = put.expect("the racing put publishes");
        assert_eq!(
            key,
            Key::derive(content, content.len() as u64),
            "round {round}: the canonical derivation addresses the racing put"
        );
        assert_eq!(report.aborted, None, "round {round}: the sweep completed");
        assert_exact_counts(&report);
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("the racing put's entry survived");
        assert_eq!(len, content.len() as u64, "round {round}");
        assert_eq!(
            blob.read_at(0, len).await.expect("the window reads"),
            content.as_slice(),
            "round {round}: the published entry is byte-identical"
        );
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            report.candidates_staged, 0,
            "round {round}: the minted pin still protects"
        );
        drop(pin);
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            (report.candidates_staged, report.deleted),
            (1, 1),
            "round {round}: unpinned, exactly one deletion"
        );
        assert!(!store.has(&key.as_bytes()).await.unwrap());
    }

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-5: `list()` correctness is observable only through GC — the
/// enumeration is complete and well-formed in both directions, proven
/// by exact sweep outcomes (neither more nor fewer candidates staged),
/// never by comparing enumerated lists.
///
/// Contract stamp: store-api.md §Invariants 5; ADR-003 (`list()` is
/// complete and well-formed by contract — load-bearing for GC);
/// ADR-005 (the sweep's live-set diff rides `list()`).
pub async fn list_correctness_is_observable_through_sweep_outcomes<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    let source = SetSource::register(&store);

    let mut keys = Vec::new();
    for (index, content) in [
        b"kept live by the source, entry one".to_vec(),
        b"kept live by the source, entry two".to_vec(),
        b"dead at the first sweep, entry three".to_vec(),
        b"dead at the first sweep, entry four".to_vec(),
        b"dead at the first sweep, entry five".to_vec(),
    ]
    .into_iter()
    .enumerate()
    {
        let declared = if index % 2 == 0 {
            Some(content.len() as u64)
        } else {
            None
        };
        let (key, pin) = store
            .put(declared, Cursor::new(content.as_slice()))
            .await
            .expect("the entry publishes");
        drop(pin);
        keys.push(key);
    }
    source.add(keys[0].clone());
    source.add(keys[1].clone());

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (3, 3, 0),
        "the enumeration staged exactly the three dead entries — complete and \
         well-formed in both directions"
    );
    assert_exact_counts(&report);
    assert!(store.has(&keys[0].as_bytes()).await.unwrap());
    assert!(store.has(&keys[1].as_bytes()).await.unwrap());
    for key in &keys[2..] {
        assert!(!store.has(&key.as_bytes()).await.unwrap());
    }

    source.remove(&keys[1]);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (report.candidates_staged, report.deleted),
        (1, 1),
        "the newly dead entry is found exactly"
    );
    assert!(store.has(&keys[0].as_bytes()).await.unwrap());

    source.remove(&keys[0]);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (report.candidates_staged, report.deleted),
        (1, 1),
        "the last entry is found exactly"
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-6: `stat` returns the entry's length and hash algorithm
/// without fetching content — the cheap probe riding the trait's
/// `size` seam — over the content matrix on both put paths, pure
/// across repeated calls, and `None` on a miss.
///
/// The non-fetching discipline's engine half (the size probe is a
/// metadata probe, never a value load) is asserted per engine by the
/// tier probes it rides; the facade-observable slice is the shape
/// this row owns.
///
/// Contract stamp: store-api.md §Invariants 6; ADR-008 §1 (the trait
/// `size` probe — stat is metadata, never a content load).
pub async fn stat_reports_len_and_algorithm_without_fetching_content<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");

    for content in matrix() {
        let len = content.len() as u64;
        let canonical = Key::derive(&content, len);
        for declared in [Some(len), None] {
            let (key, pin) = store
                .put(declared, Cursor::new(content.as_slice()))
                .await
                .expect("put publishes");
            drop(pin);
            assert_eq!(key, canonical);
            let meta = store
                .stat(&key.as_bytes())
                .await
                .expect("stat succeeds")
                .expect("the entry is present");
            assert_eq!(
                meta,
                EntryMeta {
                    len,
                    algorithm: Algorithm::GitBlobSha256,
                },
                "stat carries the entry length and the key's algorithm"
            );
            assert_eq!(
                store
                    .stat(&key.as_bytes())
                    .await
                    .expect("stat succeeds again"),
                Some(meta),
                "stat is a pure probe"
            );
        }
    }

    let absent = b"stat never sees this content".to_vec();
    let absent = Key::derive(&absent, absent.len() as u64);
    assert_eq!(
        store.stat(&absent.as_bytes()).await.expect("stat succeeds"),
        None,
        "a both-tier miss is None, never an error"
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// The kv-only rejection carrier's message marker (ADR-012 §6.2's
/// put-time rejection rides `Error::Io(String)` with a descriptive
/// message — the pinned family holds no dedicated variant).
const KV_ONLY_REJECTION: &str = "kv-only mode rejects an over-threshold put";

/// The largest threshold the boundary legs of
/// [`kv_only_mode_rejects_over_threshold_puts_at_put_time`]
/// materialize (the at-threshold legs stream exactly `threshold`
/// bytes and the kv tier holds the whole value) — the configuration
/// contract a column's kv-only factory keeps.
const MAX_BOUNDARY_MATERIALIZATION: u64 = 8 * 1024 * 1024;

/// One rejection point probe: a known-length put of `len` bytes whose
/// stream nobody may read. Over the threshold the put rejects before
/// the stream flows (`true`, the mode's message); at-or-below it the
/// put path reads the stream and fails there instead (`false`). Either
/// way nothing publishes and zero residue lands, so probing is free.
async fn probe_rejects(store: &Store, len: u64) -> bool {
    match store.put(Some(len), Unreadable).await {
        Err(err) => err.to_string().contains(KV_ONLY_REJECTION),
        Ok(_) => panic!("the Unreadable probe stream can never publish"),
    }
}

/// The constructor mode row, kv-only: over-threshold puts reject at
/// put time (ADR-012 §6.2) — a known-length put rejects immediately
/// (its stream is never read), an unknown-length put rejects at
/// mid-stream threshold overflow — with zero residue on either
/// rejection, while at-threshold puts land on both paths. The row
/// discovers the configuration's rejection point by probing
/// (rejections read nothing, so probing is free) and requires it
/// materializable (see [`MAX_BOUNDARY_MATERIALIZATION`]).
///
/// Contract stamp: store-api.md §Invariants 7 (facade-contract row of
/// the reference set); ADR-011 §4 (kv-only mode); ADR-012 §6.2
/// (known-length immediate, unknown-length mid-stream); ADR-012 §6.5
/// (the `Io(String)` rejection carrier).
pub async fn kv_only_mode_rejects_over_threshold_puts_at_put_time<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    assert_eq!(
        store.mode(),
        StoreMode::KvOnly,
        "the row runs against a kv-only factory (ADR-011 §4's mode vocabulary)"
    );
    SetSource::register(&store);

    // The rejection point: a known-length put over the threshold
    // rejects before its stream is read, so an Unreadable probe
    // stream distinguishes rejection (never read) from acceptance
    // (the probe stream errors instead). Binary-search the smallest
    // rejecting length.
    assert!(
        probe_rejects(&store, u64::MAX).await,
        "the kv-only factory must declare a threshold below u64::MAX for this \
         row to apply"
    );
    let mut low = 0u64;
    let mut high = u64::MAX;
    while high - low > 1 {
        let mid = low + (high - low) / 2;
        if probe_rejects(&store, mid).await {
            high = mid;
        } else {
            low = mid;
        }
    }
    let rejecting = high;
    let threshold = rejecting - 1;
    assert!(
        threshold <= MAX_BOUNDARY_MATERIALIZATION,
        "the kv-only factory's threshold must keep the boundary legs \
         materializable (≤ {MAX_BOUNDARY_MATERIALIZATION}); got {threshold}"
    );

    // Known-length, over-threshold: immediate — the stream never
    // flows, and zero residue lands.
    let err = store
        .put(Some(u64::MAX), Unreadable)
        .await
        .expect_err("the over-threshold known-length put rejects");
    assert!(
        err.to_string().contains(KV_ONLY_REJECTION),
        "the immediate rejection carries the mode's message: {err}"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "zero residue after the immediate rejection"
    );

    // Unknown-length, over-threshold: rejects at mid-stream threshold
    // overflow — the mode's rejection, not a stream error — and zero
    // residue.
    let content: Vec<u8> = (0..threshold)
        .map(|i| (i % 199) as u8)
        .chain(std::iter::once(7))
        .collect();
    assert_eq!(content.len() as u64, rejecting);
    let err = store
        .put(None, Chunked::new(&content, &[7; 64]))
        .await
        .expect_err("the over-threshold unknown-length put rejects at mid-stream");
    assert!(
        err.to_string().contains(KV_ONLY_REJECTION) && err.to_string().contains("streamed"),
        "the overflow rejection is the mode's, mid-stream: {err}"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (
            report.candidates_staged,
            report.deleted,
            report.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "zero residue after the mid-stream rejection"
    );

    // Exactly-at-threshold lands on both put paths (≤ threshold routes
    // kv; the rejection is strictly past it).
    let at = (0..threshold).map(|i| (i % 251) as u8).collect::<Vec<u8>>();
    let at_canonical = Key::derive(&at, threshold);
    let (key, pin) = store
        .put(Some(threshold), Cursor::new(at.as_slice()))
        .await
        .expect("the at-threshold known-length put lands");
    assert_eq!(key, at_canonical);
    drop(pin);
    assert!(store.has(&at_canonical.as_bytes()).await.unwrap());

    let other: Vec<u8> = (0..threshold).map(|i| (i % 173) as u8).collect();
    let other_canonical = Key::derive(&other, threshold);
    let (key, pin) = store
        .put(None, Chunked::new(&other, &[11; 512]))
        .await
        .expect("the at-threshold unknown-length put lands");
    assert_eq!(key, other_canonical);
    drop(pin);
    assert!(store.has(&other_canonical.as_bytes()).await.unwrap());

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}

/// The constructor mode row, mem-only: the ephemeral mode reports its
/// mode (ADR-011 §4's vocabulary) and carries **no dispatch threshold
/// in effect** — every length is accepted on both put paths,
/// including content well past the shipped default threshold, and
/// round-trips byte-identical from the sole engine.
///
/// Contract stamp: store-api.md §Invariants 7 (facade-contract row of
/// the reference set); ADR-011 §4 (mem-only mode — no dispatch
/// threshold in effect, never production).
pub async fn mem_only_mode_accepts_every_length_on_both_put_paths<F: StoreFactory>(factory: &F) {
    let store = factory
        .open()
        .await
        .expect("open vends a fresh, isolated store");
    assert_eq!(
        store.mode(),
        StoreMode::MemOnly,
        "the row runs against a mem-only factory (ADR-011 §4's mode vocabulary)"
    );

    let content = patterned(DEFAULT_DISPATCH_THRESHOLD as usize + 7, 191);
    let canonical = Key::derive(&content, content.len() as u64);
    for declared in [Some(content.len() as u64), None] {
        let (key, pin) = store
            .put(declared, Cursor::new(content.as_slice()))
            .await
            .expect("mem-only accepts the over-default-threshold put");
        assert_eq!(key, canonical, "one canonical entry across both paths");
        drop(pin);
    }

    let (len, blob) = store
        .get(&canonical.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the entry is present");
    assert_eq!(len, content.len() as u64);
    assert_eq!(
        blob.read_at(0, len).await.expect("the window reads"),
        content,
        "the sole engine holds the whole artifact byte-identical"
    );

    drop(store);
    factory.teardown().await.expect("teardown is idempotent");
}