alkblobs 0.1.0

//! The mem engine's conformance reference suite: the tests that run
//! in the default build anchoring the conformance contract (the
//! reference column). The suite spans every subject — constructors,
//! reads, puts, batches, the GC protocol's inline form — with one
//! shared helper set, so it rides the directory whole (the recorded
//! boundary note in the module doc); the engine/feature-gated
//! Construction/window/fleet sections ride their subject files.

use super::*;

use crate::backend::{slice_digest, Backend};
use crate::dispatch::DEFAULT_DISPATCH_THRESHOLD;
use crate::hashing::Algorithm;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

/// A fresh temp root per test (mirrors `local.rs`'s helper); Drop
/// removes the whole tree idempotently.
struct TempRoot(PathBuf);

fn temp_root(tag: &'static str) -> TempRoot {
    static SEQ: AtomicU64 = AtomicU64::new(0);
    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let dir = std::env::temp_dir().join(format!(
        "alkblobs-store-{tag}-{}-{:x}-{seq}",
        std::process::id(),
        nanos
    ));
    fs::create_dir_all(&dir).expect("temp root created");
    TempRoot(dir)
}

impl TempRoot {
    fn path(&self) -> &Path {
        &self.0
    }
}

impl Drop for TempRoot {
    fn drop(&mut self) {
        let _ = fs::remove_dir_all(&self.0);
    }
}

/// mem-only constructs and validates: the ephemeral mode's shape —
/// no large tier, no threshold in effect, kv routing for every
/// length (ADR-011 §4).
#[cfg(feature = "mem")]
#[test]
fn mem_only_constructs_and_validates() {
    let store = Store::mem_only();
    assert_eq!(store.mode(), MemOnly);
    assert_eq!(store.threshold(), None, "no dispatch threshold in effect");
    assert!(store.large().is_none(), "mem-only has no large tier");

    for len in [0u64, 1, DEFAULT_DISPATCH_THRESHOLD + 1, u64::MAX] {
        assert_eq!(
            store.tier_for(len),
            Tier::Kv,
            "mem-only routes every length to its sole tier"
        );
    }
}

/// kv-only constructs and validates: no large tier; the threshold
/// is the *rejection point's* carrier (ADR-012 §6.2), not a routing
/// answer — every length routes kv there, and the put seam rejects
/// known-length over-threshold puts immediately (its behavior).
#[cfg(feature = "mem")]
#[test]
fn kv_only_mem_constructs_and_validates() {
    let store = Store::kv_only_mem(1024);
    assert_eq!(store.mode(), KvOnly);
    assert_eq!(store.threshold(), Some(1024));
    assert!(store.large().is_none(), "kv-only has no large tier");

    for len in [0u64, 1023, 1024, 1025, u64::MAX] {
        assert_eq!(
            store.tier_for(len),
            Tier::Kv,
            "kv-only routes every length to its sole tier"
        );
    }
}

/// The ADR-012 §6.3 dispatch-coverage shape: `kv = mem` +
/// `large = local` dual-tier construction works, both tiers
/// load-bearing, dispatch threshold active — boundary values route
/// deterministically (at/at-below threshold kv, above large), and
/// tunability is proven against a non-default threshold.
#[cfg(feature = "mem")]
#[tokio::test]
async fn dual_tier_mem_local_constructs_and_routes() {
    let root = temp_root("dual-tier");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier mem+local constructs");
    assert_eq!(store.mode(), DualTier);
    assert_eq!(store.threshold(), Some(1024));
    assert!(store.large().is_some(), "both tiers present");

    for (len, tier) in [
        (0u64, Tier::Kv),
        (1023, Tier::Kv),
        (1024, Tier::Kv),
        (1025, Tier::Large),
        (u64::MAX, Tier::Large),
    ] {
        assert_eq!(
            store.tier_for(len),
            tier,
            "length {len} routes deterministically"
        );
        assert_eq!(
            store.tier_for(len),
            tier,
            "same length ⇒ same tier (the purity invariant, re-checked)"
        );
    }
}

/// The shipped default threshold is the constructor-tunable
/// baseline: dual-tier at `DEFAULT_DISPATCH_THRESHOLD` routes at
/// the 128 KiB boundary exactly as the pure function does.
#[cfg(feature = "mem")]
#[tokio::test]
async fn dual_tier_routes_at_the_default_threshold() {
    let root = temp_root("dual-tier-default");
    let store = Store::dual_tier_mem_local(root.path(), DEFAULT_DISPATCH_THRESHOLD)
        .await
        .expect("dual-tier constructs at the default threshold");

    for (len, tier) in [
        (131_071, Tier::Kv),
        (131_072, Tier::Kv),
        (131_073, Tier::Large),
    ] {
        assert_eq!(store.tier_for(len), tier, "length {len} at 128 KiB");
    }
}

/// The GC state home rides as a field of the store instance
/// (ADR-010 §2 — the store instance owns GC state) and is
/// reachable crate-internally: the gc-protocol task's machinery
/// lands against this handle.
#[cfg(feature = "mem")]
#[test]
fn gc_state_home_rides_on_the_foundation() {
    let store = Store::mem_only();
    let _ = store.gc();
}

/// Entry planters straight into the tier engines (bypassing the
/// put seam, which is a later task): opaque key bytes + value,
/// exactly what the engines' CAS tables accept. Planting
/// mismatching content under a key is how the read path's
/// verification postures are observed.
async fn plant_kv(store: &Store, key: &[u8], value: &[u8]) {
    match store.kv() {
        #[cfg(feature = "mem")]
        KvEngine::Mem(engine) => engine.put(key, value).await.expect("kv plant"),
        #[cfg(feature = "postgres")]
        KvEngine::Postgres(engine) => engine.put(key, value).await.expect("kv plant"),
        #[cfg(feature = "sqlite")]
        KvEngine::Sqlite(engine) => engine.put(key, value).await.expect("kv plant"),
    }
}

async fn plant_large(store: &Store, key: &[u8], value: &[u8]) {
    match store.large() {
        Some(LargeEngine::Local(engine)) => engine.put(key, value).await.expect("large plant"),
        #[cfg(feature = "pg-lo")]
        Some(LargeEngine::PgLo(_)) => panic!("these tests construct local-tier stores"),
        None => panic!("planting into a mode with no large tier"),
    }
}

fn key_for(content: &[u8]) -> Key {
    Key::derive(content, content.len() as u64)
}

fn sha1_key_for(content: &[u8]) -> Key {
    Key::from(Algorithm::GitBlobSha1.derive(content, content.len() as u64))
}

/// The both-tier-miss shapes every read op must resolve to
/// (store-api.md §Error model; ADR-012 §6.5 ruling 5) — pinned on
/// all three constructible modes, virgin-store reads as no-ops.
async fn assert_virgin_read_shapes(store: &Store) {
    let key = key_for(b"nothing ever landed").as_bytes();

    assert!(
        store.get(&key).await.unwrap().is_none(),
        "get's both-tier miss is Ok(None)"
    );
    assert_eq!(store.stat(&key).await.unwrap(), None, "stat misses as None");
    assert_eq!(
        store.read_range(&key, 0, 16).await.unwrap_err(),
        Error::Missing,
        "read_range's both-tier miss is Err(Missing) — the pinned shape difference vs get"
    );
    assert!(
        !store.has(&key).await.unwrap(),
        "has's both-tier miss is false"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "virgin-store list is empty"
    );
}

#[tokio::test]
async fn virgin_store_reads_are_no_ops_dual_tier() {
    let root = temp_root("read-virgin-dual");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");
    assert_virgin_read_shapes(&store).await;
}

#[tokio::test]
async fn virgin_store_reads_are_no_ops_mem_only() {
    let store = Store::mem_only();
    assert_virgin_read_shapes(&store).await;
}

#[tokio::test]
async fn virgin_store_reads_are_no_ops_kv_only() {
    let store = Store::kv_only_mem(64);
    assert_virgin_read_shapes(&store).await;
}

/// `get` returns the pinned `(len, stream)` shape from the kv tier:
/// the tuple carries the metadata length, the `Blob` serves
/// bounded windows (clamped at EOF) and the ADR-006 range seam.
#[tokio::test]
async fn get_returns_the_pinned_len_stream_shape_from_kv() {
    let store = Store::mem_only();
    let content = b"small-tier content";
    let key = key_for(content);
    plant_kv(&store, &key.as_bytes(), content).await;

    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(len, content.len() as u64, "tuple len is the entry length");
    assert_eq!(blob.len(), len);
    assert!(!blob.is_empty());

    let whole = blob.read_at(0, len).await.expect("full window");
    assert_eq!(whole, content, "full window matches");
    assert_eq!(
        blob.read_at(0, len + 17).await.expect("over-read"),
        content,
        "over-read clamps at the entry's end"
    );

    let (slice, digest) = blob.read_range(6, 6).await.expect("ranged window");
    assert_eq!(slice, &content[6..12]);
    assert_eq!(
        digest,
        slice_digest(&content[6..12]),
        "slice digest is SHA-256 over exactly the returned bytes"
    );

    let (full, full_digest) = blob.read_range(0, len).await.expect("full range");
    assert_eq!(full, content);
    assert_eq!(full_digest, slice_digest(content));
}

/// The zero-length degenerate: a kv-tier empty entry get's as
/// `Some((0, blob))`, its windows read empty-never-error, and
/// `stat` carries exact zero.
#[tokio::test]
async fn zero_length_kv_entry_reads_as_empty_blob() {
    let store = Store::mem_only();
    let key = key_for(b"");
    plant_kv(&store, &key.as_bytes(), b"").await;

    let (len, blob) = store.get(&key.as_bytes()).await.unwrap().expect("present");
    assert_eq!(len, 0);
    assert!(blob.is_empty());
    assert_eq!(blob.read_at(0, 8).await.unwrap(), Vec::<u8>::new());

    assert_eq!(
        store.stat(&key.as_bytes()).await.unwrap(),
        Some(EntryMeta {
            len: 0,
            algorithm: key.algorithm()
        })
    );
}

/// `get`/`stat`/`read_range`/`has` fall through to the large tier
/// on a kv miss — consult order, not length routing: a
/// sub-threshold entry planted only in the large tier is still
/// found even though `tier_for` routes its length to kv.
#[tokio::test]
async fn read_ops_fall_through_kv_miss_to_large() {
    let root = temp_root("read-fall-through");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");
    let content = b"a sub-threshold entry planted directly in the large tier";
    let key = key_for(content);

    assert_eq!(
        store.tier_for(content.len() as u64),
        Tier::Kv,
        "the length *routes* kv — the read still lands via the large tier"
    );
    plant_large(&store, &key.as_bytes(), content).await;

    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("fall-through get succeeds")
        .expect("large-tier entry found via fall-through");
    assert_eq!(len, content.len() as u64);
    assert_eq!(blob.read_at(0, len).await.unwrap(), content);

    assert_eq!(
        store.stat(&key.as_bytes()).await.unwrap(),
        Some(EntryMeta {
            len: content.len() as u64,
            algorithm: key.algorithm()
        }),
        "stat falls through too"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "has falls through"
    );

    let (slice, digest) = store
        .read_range(&key.as_bytes(), 4, 8)
        .await
        .expect("read_range falls through");
    assert_eq!(slice, &content[4..12]);
    assert_eq!(digest, slice_digest(&slice));
}

/// The kv tier is the *first* consult on every key-addressed op:
/// the same key bytes planted with different (mismatching) content
/// in both tiers — get/read_range/stat/has must all answer from
/// the kv tier's entry.
#[tokio::test]
async fn kv_tier_is_the_first_consult_on_every_key_addressed_op() {
    let root = temp_root("read-kv-first");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");
    const KV_CONTENT: &[u8] = b"the kv tier's entry";
    let key = key_for(KV_CONTENT);
    plant_kv(&store, &key.as_bytes(), KV_CONTENT).await;
    plant_large(&store, &key.as_bytes(), b"decoy content in the large tier").await;

    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("kv hit wins");
    assert_eq!(
        blob.read_at(0, len).await.unwrap(),
        KV_CONTENT,
        "kv tier first — and its content verifies against the key"
    );

    let (slice, _) = store
        .read_range(&key.as_bytes(), 0, 4)
        .await
        .expect("ranged");
    assert_eq!(
        slice,
        &KV_CONTENT[0..4],
        "read_range also consults kv first"
    );

    assert!(store.has(&key.as_bytes()).await.unwrap());
    assert_eq!(
        store.stat(&key.as_bytes()).await.unwrap(),
        Some(EntryMeta {
            len: KV_CONTENT.len() as u64,
            algorithm: key.algorithm()
        })
    );
}

/// Verification-on-get, kv tier (ADR-006 decision 1): a
/// materialized get hash-checks against the key; mismatch is
/// `Err(Verification)` and never falls through further. `read_range`
/// by contrast carries no canonical check — the slice digest is its
/// whole verification story (ADR-006 decision 2).
#[tokio::test]
async fn kv_get_whole_blob_verifies_and_never_falls_through_past_mismatch() {
    let store = Store::mem_only();
    let key = key_for(b"the derived address");
    let bytes = key.as_bytes();
    plant_kv(&store, &bytes, b"content that is not the key").await;

    assert_eq!(
        store.get(&bytes).await.unwrap_err(),
        Error::Verification,
        "kv materialized reads hash-check at get time"
    );
    assert_eq!(
        store.read_range(&bytes, 0, 8).await.unwrap(),
        (b"content ".to_vec(), slice_digest(b"content ")),
        "read_range carries no canonical check — the slice-digest posture"
    );
    assert!(
        store.stat(&bytes).await.unwrap().is_some(),
        "stat is a metadata probe, not a verifier"
    );
}

/// Verification-on-get, large tier (the recorded posture's other
/// half): a streaming get opens metadata + handle only — it does
/// not re-derive the canonical digest at open. The tier's
/// integrity story is the put path's whole-blob verification pass
/// plus the per-range slice digests (ADR-006 decisions 1-2).
#[tokio::test]
async fn large_streaming_get_does_not_re_derive_at_open() {
    let root = temp_root("read-large-no-reverify");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let key = key_for(b"the derived address");
    plant_large(&store, &key.as_bytes(), b"stored bytes").await;

    assert!(
        store.get(&key.as_bytes()).await.unwrap().is_some(),
        "streaming open never re-derives (mismatching content and all)"
    );
}

/// `stat` carries len + algorithm for both algorithms (the
/// "type" at this layer is the hash algorithm, ADR-002) and is a
/// `size` ride — the fall-through consult, `None` when absent.
#[tokio::test]
async fn stat_carries_len_and_algorithm_from_size_probes() {
    let root = temp_root("read-stat");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");

    let sha256_key = key_for(b"canonical");
    plant_kv(&store, &sha256_key.as_bytes(), b"canonical").await;
    let legacy_key = sha1_key_for(b"legacy");
    plant_large(&store, &legacy_key.as_bytes(), b"legacy").await;

    assert_eq!(
        store.stat(&sha256_key.as_bytes()).await.unwrap(),
        Some(EntryMeta {
            len: 9,
            algorithm: Algorithm::GitBlobSha256
        })
    );
    assert_eq!(
        store
            .stat(&legacy_key.as_bytes())
            .await
            .unwrap()
            .expect("legacy entry present")
            .algorithm,
        Algorithm::GitBlobSha1,
        "the algorithm comes from the key"
    );
    assert_eq!(
        store.stat(&sha256_key.as_bytes()).await.unwrap(),
        store.stat(&sha256_key.as_bytes()).await.unwrap(),
        "stat is a pure probe"
    );
}

/// `list` returns the sorted, typed union of both tiers'
/// enumerations — both algorithms, dedup-free by key identity.
#[tokio::test]
async fn list_returns_the_sorted_typed_union_of_both_tiers() {
    let root = temp_root("read-list");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");

    let kv_keys = [
        key_for(b"kv entry one"),
        sha1_key_for(b"kv legacy"),
        key_for(b"kv entry two"),
    ];
    for key in &kv_keys {
        plant_kv(&store, &key.as_bytes(), b"kv bytes").await;
    }
    let large_keys = [key_for(b"large entry one"), key_for(b"large entry two")];
    for key in &large_keys {
        plant_large(&store, &key.as_bytes(), b"large bytes").await;
    }

    let mut expected: Vec<Key> = kv_keys.into_iter().chain(large_keys).collect();
    expected.sort();

    let listed = store.list().await.expect("list succeeds");
    assert_eq!(listed, expected, "complete, typed, both tiers");
    for window in listed.windows(2) {
        assert!(
            window[0] < window[1],
            "output is sorted in the keys' total order"
        );
    }
    for key in &listed {
        assert!(
            store.has(&key.as_bytes()).await.unwrap(),
            "every listed key addresses a real entry"
        );
    }
}

/// A malformed pool list is fatal (`Error::KeyInvalid`), never a
/// silent skip — ADR-003's malformed-list ruling applied at the
/// facade boundary; the engines own their own body-level checks.
#[tokio::test]
async fn list_of_malformed_pool_keys_is_fatal() {
    let store = Store::mem_only();
    plant_kv(&store, &[0x07; 33], b"stray in the table").await;

    assert_eq!(
        store.list().await.unwrap_err(),
        Error::KeyInvalid,
        "malformed pool bytes are fatal on list"
    );
}

/// `KeyInvalid` at the facade boundary on every key-addressed
/// op (typed conversion happens there and only there, ADR-002/003)
/// — malformed bytes reject on hit paths (a planted entry) and
/// miss paths alike, before any tier is consulted and never as a
/// `None`/`false` guess.
#[tokio::test]
async fn malformed_key_bytes_reject_keyinvalid_on_every_op() {
    let store = Store::mem_only();
    let anchor = key_for(b"an anchor entry");
    plant_kv(&store, &anchor.as_bytes(), b"anchor").await;

    let key_bytes = anchor.as_bytes();
    let malformed = [
        Vec::new(),                                // empty
        key_bytes[..key_bytes.len() - 1].to_vec(), // truncated digest
        [0x00u8; 33].to_vec(),                     // reserved tag, full width
        [0x03u8; 33].to_vec(),                     // unknown tag
        [0x01u8; 21].to_vec(),                     // right tag, wrong digest length
    ];
    for raw in &malformed {
        assert_eq!(
            store.get(raw).await.unwrap_err(),
            Error::KeyInvalid,
            "get rejects malformed key bytes: {raw:?}"
        );
        assert_eq!(
            store.stat(raw).await.unwrap_err(),
            Error::KeyInvalid,
            "stat rejects malformed key bytes: {raw:?}"
        );
        assert_eq!(
            store.read_range(raw, 0, 1).await.unwrap_err(),
            Error::KeyInvalid,
            "read_range rejects malformed key bytes: {raw:?}"
        );
        assert_eq!(
            store.has(raw).await.unwrap_err(),
            Error::KeyInvalid,
            "has rejects malformed key bytes: {raw:?}"
        );
    }
}

// ------------------------------------------------------------------
// The GC protocol (the gc-protocol work): pin registry, registered
// liveness sources, protect callback, delete windows, direct-delete
// refusal — ADR-005/012/013 on the facade.
// ------------------------------------------------------------------

/// A set-backed liveness source — shared-handle semantics, the POC
/// finding-7 shape: the store consults the same set this test
/// mutates, so its answers are live-shared, never snapshots.
#[derive(Default)]
struct SetSource {
    live: parking_lot::Mutex<BTreeSet<Key>>,
}

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

    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/refuse safe side).
#[derive(Default)]
struct SetProtect {
    set: parking_lot::Mutex<Option<Vec<Key>>>,
}

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

    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 panics — the sweep body's panic route
/// (review-001 N-6's plausible route).
struct PanickingProtect;

impl ProtectCallback for PanickingProtect {
    fn protect(&self) -> Option<Vec<Key>> {
        panic!("the protect consult panicked");
    }
}

/// A protect callback that blocks mid-consult (the sweep lease's
/// contended-window fixture): it announces its entry on a channel,
/// waits on a second, then answers with an empty protection set.
struct LatchProtect {
    entered: std::sync::mpsc::Sender<()>,
    release: parking_lot::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())
    }
}

/// `sweep()` with zero registered liveness sources is the safe
/// default: `Ok(SweepReport)` carrying `NoLivenessSources`, nothing
/// deleted, the pool intact (store-api.md §Lifecycle).
#[tokio::test]
async fn sweep_with_zero_sources_aborts_as_data_and_deletes_nothing() {
    let store = Store::mem_only();
    let key = key_for(b"unprotected content");
    plant_kv(&store, &key.as_bytes(), b"unprotected content").await;

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, Some(GcAbortCause::NoLivenessSources));
    assert_eq!(report.candidates_staged, 0);
    assert_eq!(report.deleted, 0);
    assert_eq!(report.cancelled_by_arbitration, 0);
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "an aborted sweep deletes nothing"
    );
}

/// With sources registered, an empty pool sweeps to a zero report
/// — no abort, exact zero counts.
#[tokio::test]
async fn sweep_with_sources_and_an_empty_pool_is_a_zero_report() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(report.candidates_staged, 0);
    assert_eq!(report.deleted, 0);
    assert_eq!(report.cancelled_by_arbitration, 0);
}

/// N-6's in-process half: a panic inside the sweep body (the protect
/// callback's panic route, consulted after the lease is taken) cannot
/// leave the single-node lease `true`-stuck — the lease guard ends it
/// on the unwind; the recovery sweep acquires cleanly (the panicking
/// consult unregistered by the correction flow).
#[tokio::test]
async fn sweep_panic_through_the_protect_consult_releases_the_in_process_lease() {
    let store = Arc::new(Store::mem_only());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    let protect = store.register_protect_callback(Arc::new(PanickingProtect));

    let sweeper = Arc::clone(&store);
    let joined = tokio::spawn(async move { sweeper.sweep().await })
        .await
        .unwrap_err();
    assert!(
        joined.is_panic(),
        "the sweep panicked through the protect consult (the recorded panic route)"
    );
    assert!(store.unregister_protect_callback(protect));

    let recovered = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        recovered.aborted, None,
        "the in-process lease released on the panic — no SweeperLock abort"
    );
}

/// Exact counts on a mixed pool (store-api.md invariant 4): the
/// registered source's key and a pinned key mark live, both tiers'
/// dead entries stage and delete — `candidates_staged == deleted +
/// cancelled_by_arbitration`.
#[tokio::test]
async fn sweep_counts_are_exact_across_both_tiers() {
    let root = temp_root("gc-exact-counts");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");
    let source = Arc::new(SetSource::default());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));

    let live = key_for(b"kept by the registered source");
    plant_kv(&store, &live.as_bytes(), b"kept by the registered source").await;
    source.add(live.clone());

    let pinned = key_for(b"kept by the pin");
    plant_kv(&store, &pinned.as_bytes(), b"kept by the pin").await;
    let pin = store.gc().acquire(&pinned).await;

    let kv_dead = key_for(b"dead kv entry");
    plant_kv(&store, &kv_dead.as_bytes(), b"dead kv entry").await;
    let large_dead = key_for(b"dead large entry with bulk");
    plant_large(
        &store,
        &large_dead.as_bytes(),
        b"dead large entry with bulk",
    )
    .await;

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(report.candidates_staged, 2);
    assert_eq!(report.deleted, 2, "both tiers' dead entries deleted");
    assert_eq!(report.cancelled_by_arbitration, 0);
    assert_eq!(
        report.candidates_staged,
        report.deleted + report.cancelled_by_arbitration,
        "the completed sweep's counts are exact over the staged set"
    );
    assert!(store.has(&live.as_bytes()).await.unwrap());
    assert!(store.has(&pinned.as_bytes()).await.unwrap());
    assert!(!store.has(&kv_dead.as_bytes()).await.unwrap());
    assert!(!store.has(&large_dead.as_bytes()).await.unwrap());

    drop(pin);
}

/// A dedup re-put's guard refcounts the existing entry: two pins
/// compose, each drop decrements, and only the last drop exposes
/// the entry to a sweep (ADR-005's dedup-refcount ruling).
#[tokio::test]
async fn dedup_reput_guards_refcount_under_the_arbitration_lock() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    let content = b"refcounted content";
    let key = key_for(content);
    plant_kv(&store, &key.as_bytes(), content).await;

    let first = store.gc().acquire(&key).await;
    let second = store.gc().acquire(&key).await;
    assert_eq!(store.gc().pin_count(&key), 2, "dedup re-put refcounts");
    drop(first);
    assert_eq!(store.gc().pin_count(&key), 1);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 0, "one refcount still protects");
    assert!(store.has(&key.as_bytes()).await.unwrap());

    drop(second);
    assert_eq!(store.gc().pin_count(&key), 0, "zero removes the entry row");
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 1);
    assert_eq!(report.deleted, 1);
    assert!(!store.has(&key.as_bytes()).await.unwrap());
}

/// Batch pins protect the whole group; batch drop releases every
/// member in one ordered step — no early single-pin release
/// (ADR-005 §Marks 2).
#[tokio::test]
async fn batch_pins_drop_together_in_one_ordered_step() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    let a = key_for(b"batch member a");
    let b = key_for(b"batch member b");
    plant_kv(&store, &a.as_bytes(), b"batch member a").await;
    plant_kv(&store, &b.as_bytes(), b"batch member b").await;

    let mut batch = store.gc().new_batch();
    batch.add(store.gc().acquire(&a).await);
    batch.add(store.gc().acquire(&b).await);
    assert_eq!(batch.len(), 2);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 0, "the batch protects both");
    assert!(store.has(&a.as_bytes()).await.unwrap());
    assert!(store.has(&b.as_bytes()).await.unwrap());

    drop(batch);
    assert_eq!(
        store.gc().pin_count(&a) + store.gc().pin_count(&b),
        0,
        "batch drop released every member refcount"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 2);
    assert_eq!(report.deleted, 2);
}

/// The protect callback adds externally-known hashes (they stay
/// alive) and aborts as report data on failure — nothing deleted,
/// `ProtectFailed` in `Ok(SweepReport)` (ADR-012 §1).
#[tokio::test]
async fn protect_callback_adds_keys_and_aborts_as_data() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    let protect = Arc::new(SetProtect::default());
    store.register_protect_callback(protect.clone());

    let named = key_for(b"protected by the callback");
    plant_kv(&store, &named.as_bytes(), b"protected by the callback").await;
    let dead = key_for(b"dead while the callback answers");
    plant_kv(&store, &dead.as_bytes(), b"dead while the callback answers").await;
    protect.set_keys(vec![named.clone()]);

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, None);
    assert_eq!(report.candidates_staged, 1, "the named key never staged");
    assert_eq!(report.deleted, 1);
    assert!(store.has(&named.as_bytes()).await.unwrap());
    assert!(!store.has(&dead.as_bytes()).await.unwrap());

    protect.fail();
    let another_dead = key_for(b"dead while the callback fails");
    plant_kv(
        &store,
        &another_dead.as_bytes(),
        b"dead while the callback fails",
    )
    .await;

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, Some(GcAbortCause::ProtectFailed));
    assert_eq!(
        report.deleted, 0,
        "a failed protection source deletes nothing"
    );
    assert!(store.has(&another_dead.as_bytes()).await.unwrap());
    assert!(store.has(&named.as_bytes()).await.unwrap());
}

/// The live-shared seam: registration never snapshots — content
/// added to the set after registering is seen by the next sweep;
/// content removed exposes the entry again (POC #1 finding 7).
#[tokio::test]
async fn liveness_sources_are_consulted_live_not_snapshotted() {
    let store = Store::mem_only();
    let source = Arc::new(SetSource::default());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));

    let key = key_for(b"live-shared content");
    plant_kv(&store, &key.as_bytes(), b"live-shared content").await;

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "unprotected at mark, deleted");

    plant_kv(&store, &key.as_bytes(), b"live-shared content").await;
    source.add(key.clone());
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.candidates_staged, 0,
        "the registration after registration is consulted live"
    );
    assert!(store.has(&key.as_bytes()).await.unwrap());

    source.remove(&key);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "deregistration re-exposes the entry");
    assert!(!store.has(&key.as_bytes()).await.unwrap());
}

/// Unregistration consumes the token (returning `false` the second
/// time) and the corrected zero-source posture aborts sweeps again
/// — ADR-012 §6.4's correction-flow prerequisite.
#[tokio::test]
async fn unregistration_consumes_the_token_and_restores_the_safe_default() {
    let store = Store::mem_only();
    let registration = store
        .register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )))
        .expect("the in-process registration never errors in practice");

    assert!(
        store.unregister_liveness_source(registration),
        "the token deregisters once"
    );
    assert!(
        !store.unregister_liveness_source(registration),
        "the token does not deregister twice"
    );

    let protect_token = store.register_protect_callback(Arc::new(SetProtect::default()));
    assert!(
        !store.unregister_liveness_source(protect_token),
        "a protect token is not a source token"
    );
    assert!(store.unregister_protect_callback(protect_token));

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.aborted, Some(GcAbortCause::NoLivenessSources));
}

/// Direct `delete` refuses the full protection set with the typed
/// `Error::GcRefuse` — pinned, registered-source-live,
/// protect-callback-named (ADR-012 §6.4) — and deletes unprotected
/// keys through the fall-through dispatch; absent keys carry
/// `Err(Missing)`; malformed bytes `Err(KeyInvalid)`.
#[tokio::test]
async fn direct_delete_refuses_the_full_protection_set() {
    let root = temp_root("gc-direct-delete");
    let store = Store::dual_tier_mem_local(root.path(), 1024)
        .await
        .expect("dual-tier constructs");
    let source = Arc::new(SetSource::default());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));
    let protect = Arc::new(SetProtect::default());
    store.register_protect_callback(protect.clone());

    let pinned = key_for(b"pinned refusal");
    plant_kv(&store, &pinned.as_bytes(), b"pinned refusal").await;
    let _pin = store.gc().acquire(&pinned).await;

    let source_live = key_for(b"source-live refusal");
    plant_large(&store, &source_live.as_bytes(), b"source-live refusal").await;
    source.add(source_live.clone());

    let protect_named = key_for(b"protect-named refusal");
    plant_kv(&store, &protect_named.as_bytes(), b"protect-named refusal").await;
    protect.set_keys(vec![protect_named.clone()]);

    let unprotected_kv = key_for(b"unprotected kv deletes");
    plant_kv(
        &store,
        &unprotected_kv.as_bytes(),
        b"unprotected kv deletes",
    )
    .await;
    let unprotected_large = key_for(b"unprotected large deletes");
    plant_large(
        &store,
        &unprotected_large.as_bytes(),
        b"unprotected large deletes",
    )
    .await;

    assert_eq!(
        store.delete(&pinned.as_bytes()).await.unwrap_err(),
        Error::GcRefuse,
        "pinned keys refuse"
    );
    assert!(
        store.has(&pinned.as_bytes()).await.unwrap(),
        "nothing deleted on refusal"
    );
    assert_eq!(
        store.delete(&source_live.as_bytes()).await.unwrap_err(),
        Error::GcRefuse,
        "registered-source-live keys refuse"
    );
    assert!(store.has(&source_live.as_bytes()).await.unwrap());
    assert_eq!(
        store.delete(&protect_named.as_bytes()).await.unwrap_err(),
        Error::GcRefuse,
        "protect-callback-named keys refuse"
    );
    assert!(store.has(&protect_named.as_bytes()).await.unwrap());

    store
        .delete(&unprotected_kv.as_bytes())
        .await
        .expect("unprotected kv keys delete");
    assert!(!store.has(&unprotected_kv.as_bytes()).await.unwrap());
    store
        .delete(&unprotected_large.as_bytes())
        .await
        .expect("unprotected large keys delete through the fall-through");
    assert!(!store.has(&unprotected_large.as_bytes()).await.unwrap());

    let absent = key_for(b"never planted");
    assert_eq!(
        store.delete(&absent.as_bytes()).await.unwrap_err(),
        Error::Missing,
        "absent keys carry Missing"
    );
    assert_eq!(
        store.delete(&[0x03; 33]).await.unwrap_err(),
        Error::KeyInvalid,
        "malformed key bytes reject at the boundary"
    );
}

/// `delete_everywhere`'s partial failure reports which homes moved
/// (review-001 N-7; ADR-012 §6.5's altitude discipline): a degenerate
/// duplicate (the same key in the kv home and the large home) whose
/// large-home delete is forced to fail mid-delete returns an `Err`
/// that wraps the media failure in an `Error::Io` payload naming both
/// homes' outcomes — while the kv home's deletion actually stands and
/// the large home still holds the divergent residue. The wrapper form
/// is the recorded reporting shape (the family is ADR-pinned; no
/// payload variant, no report field); the self-healing re-mark posture
/// is unchanged — the observability is the fix.
#[tokio::test]
async fn delete_everywhere_partial_failure_reports_which_homes_moved() {
    let root = temp_root("gc-partial-delete");
    let mut store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content = b"degenerate duplicate";
    let key = key_for(content);
    plant_kv(&store, &key.as_bytes(), content).await;
    plant_large(&store, &key.as_bytes(), content).await;

    match &mut store.large {
        Some(LargeEngine::Local(engine)) => engine.inject_fail_delete(),
        #[cfg(feature = "pg-lo")]
        Some(LargeEngine::PgLo(_)) => panic!("these tests construct local-tier stores"),
        None => panic!("the tested mode carries a large tier"),
    }

    let err = store.delete(&key.as_bytes()).await.unwrap_err();
    assert!(
        matches!(err, Error::Io(_)),
        "the report rides the Io family, not a new variant"
    );
    let msg = err.to_string();
    assert!(
        msg.contains("large home delete failed"),
        "the failed home is named: {msg}"
    );
    assert!(
        msg.contains("kv home deleted"),
        "the kv home's move is in the report: {msg}"
    );
    assert!(
        msg.contains("injected"),
        "the underlying media failure rides the payload: {msg}"
    );

    assert!(
        !store.kv().has(&key.as_bytes()).await.unwrap(),
        "the kv home's deletion really moved"
    );
    assert!(
        store
            .large()
            .expect("the tested mode carries a large tier")
            .has(&key.as_bytes())
            .await
            .unwrap(),
        "the large home still holds the divergent residue"
    );
}

/// The homes report's other arm (the wave-9 gate's coverage review):
/// the large home's failure with *nothing moved in the kv home* — the
/// kv delete answered `Missing` (the entry was large-only) — reports
/// the kv-absent form, not the kv-deleted divergence.
#[tokio::test]
async fn delete_everywhere_partial_failure_reports_the_kv_absent_arm() {
    let root = temp_root("gc-partial-delete-kv-absent");
    let mut store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content = b"large only";
    let key = key_for(content);
    plant_large(&store, &key.as_bytes(), content).await;

    match &mut store.large {
        Some(LargeEngine::Local(engine)) => engine.inject_fail_delete(),
        #[cfg(feature = "pg-lo")]
        Some(LargeEngine::PgLo(_)) => panic!("these tests construct local-tier stores"),
        None => panic!("the tested mode carries a large tier"),
    }

    let err = store.delete(&key.as_bytes()).await.unwrap_err();
    let msg = err.to_string();
    assert!(
        msg.contains("large home delete failed"),
        "the failed home is named: {msg}"
    );
    assert!(
        msg.contains("kv home absent"),
        "the kv home's absence is in the report: {msg}"
    );
    assert!(
        !store.kv().has(&key.as_bytes()).await.unwrap(),
        "the kv home never held the entry"
    );
    assert!(
        store
            .large()
            .expect("the tested mode carries a large tier")
            .has(&key.as_bytes())
            .await
            .unwrap(),
        "the large home still holds the residue"
    );
}

/// Sequential evidence of the conversion ordering primitive: after
/// the ordered step, the pin refcount is released **and** the
/// registration is visible — under one arbitration-lock section
/// (ADR-005's pinned conversion ordering).
#[tokio::test]
async fn pin_conversion_releases_and_registers_in_one_ordered_step() {
    let store = Store::mem_only();
    let source = Arc::new(SetSource::default());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));

    let key = key_for(b"converted protection");
    let pin = store.gc().acquire(&key).await;
    assert_eq!(store.gc().pin_count(&key), 1);

    store.gc().convert(pin, || source.add(key.clone()));

    assert_eq!(
        store.gc().pin_count(&key),
        0,
        "the pin's release is one half of the ordered step"
    );
    assert!(
        source.contains(&key),
        "the registration is the other half — the same section"
    );

    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.candidates_staged, 0,
        "the converted registration protects through the source"
    );
}

/// The ordered step under race: a pin converted into a source is
/// never an unprotected instant — a concurrent direct delete must
/// observe the pin or the registration at every schedule and
/// refuse, never delete (ADR-005's conversion-ordering invariant,
/// raced).
#[tokio::test]
async fn conversion_raced_against_direct_delete_never_deletes() {
    let store = Store::mem_only();
    let source = Arc::new(SetSource::default());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));

    for round in 0..64 {
        let content = format!("converted protection round {round}");
        let key = key_for(content.as_bytes());
        let bytes = key.as_bytes();
        plant_kv(&store, &bytes, content.as_bytes()).await;

        let pin = store.gc().acquire(&key).await;
        let (result, ()) = tokio::join!(store.delete(&bytes), async {
            store.gc().convert(pin, {
                let source = Arc::clone(&source);
                let key = key.clone();
                move || source.add(key)
            })
        },);
        assert_eq!(
            result,
            Err(Error::GcRefuse),
            "round {round}: no schedule leaves the key unprotected "
        );
        assert!(
            store.has(&bytes).await.unwrap(),
            "nothing deleted while the protection converted"
        );

        source.remove(&key);
        store
            .delete(&bytes)
            .await
            .expect("unprotected now, deletes");
    }
}

/// The test-asserted property (ADR-005): a sweep racing an
/// in-flight pinned put never deletes the protected entry — the
/// pin-before-publish ordering holds from the sweep side, and the
/// delete-then-recover arm lands byte-identical. Counts stay exact
/// in every schedule, and the surviving pin keeps protecting.
#[tokio::test]
async fn sweep_racing_a_pinned_put_never_deletes_the_protected_entry() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    for round in 0..64 {
        let content = format!("race round {round}");
        let key = key_for(content.as_bytes());
        let bytes = key.as_bytes();

        plant_kv(&store, &bytes, content.as_bytes()).await;

        let (report, pin) = tokio::join!(store.sweep(), async {
            let pin = store.gc().acquire(&key).await;
            plant_kv(&store, &bytes, content.as_bytes()).await;
            pin
        },);

        let report = report.expect("sweeps are never errors");
        assert_eq!(report.aborted, None, "round {round}");
        assert_eq!(
            report.candidates_staged,
            report.deleted + report.cancelled_by_arbitration,
            "round {round}: exact counts in every schedule"
        );

        assert!(
            store.has(&bytes).await.unwrap(),
            "round {round}: the entry survives (canceled, or delete-then-recovered)"
        );
        let (_, blob) = store
            .get(&bytes)
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(
            blob.read_at(0, content.len() as u64).await.unwrap(),
            content.as_bytes(),
            "round {round}: the recovered re-put is byte-identical"
        );

        drop(pin);
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            report.deleted, 1,
            "round {round}: with the pin dropped the entry is dead — exactly one deletion"
        );
        assert!(!store.has(&bytes).await.unwrap());
    }
}

/// An in-flight put whose publish lands after the sweep's mark is
/// never enumerated by it — the sweep started before the commit
/// and cannot delete what its mark never saw (ADR-005's
/// pin-before-publish invariant, the fresh-put arm).
#[tokio::test]
async fn put_publishing_after_a_sweep_started_is_never_deleted() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let content = b"published after the mark";
    let key = key_for(content);
    let bytes = key.as_bytes();
    let dead = key_for(b"a dead witness for the concurrent sweep");
    plant_kv(
        &store,
        &dead.as_bytes(),
        b"a dead witness for the concurrent sweep",
    )
    .await;

    let (report, pin) = tokio::join!(store.sweep(), async {
        let pin = store.gc().acquire(&key).await;
        tokio::task::yield_now().await;
        tokio::task::yield_now().await;
        plant_kv(&store, &bytes, content).await;
        pin
    },);

    let report = report.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "the dead witness deleted");
    assert_eq!(
        report.candidates_staged, 1,
        "the in-flight put's entry never listed — nothing staged for it"
    );
    assert!(
        store.has(&bytes).await.unwrap(),
        "the published entry survived"
    );
    assert_eq!(store.gc().pin_count(&key), 1, "the pin still protects");

    drop(pin);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "now unprotected, deleted");
}

/// `has`/`get` during a delete window observe no torn state: each
/// observation is either the untouched full content or clean
/// absence — existence flips atomically per key, content never
/// partial (the pool's immutable-entry posture).
#[tokio::test]
async fn reads_during_a_delete_window_observe_no_torn_state() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let content = b"whole-content visibility under the window";
    let key = key_for(content);
    let bytes = key.as_bytes();
    plant_kv(&store, &bytes, content).await;

    let observer = {
        let store = &store;
        let bytes = bytes.clone();
        let content_len = content.len() as u64;
        async move {
            let mut seen_true = false;
            let mut seen_absence = false;
            for _ in 0..200 {
                if store.has(&bytes).await.unwrap() {
                    seen_true = true;
                    assert!(
                        !seen_absence,
                        "no re-appearance: existence flips once per key"
                    );
                    let (_, blob) = store
                        .get(&bytes)
                        .await
                        .expect("get succeeds")
                        .expect("has said true");
                    let whole = blob.read_at(0, content_len).await.unwrap();
                    assert_eq!(
                        whole, content,
                        "an observed entry presents its exact, whole content"
                    );
                } else {
                    seen_absence = true;
                }
                tokio::task::yield_now().await;
            }
            (seen_true, seen_absence)
        }
    };

    let ((seen_true, seen_absence), report) = tokio::join!(observer, store.sweep());
    let report = report.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1);
    assert!(seen_true || seen_absence, "the observer ran alongside");
    assert!(
        !store.has(&bytes).await.unwrap(),
        "after the window the entry is gone"
    );
}

/// The single-node sweep is single-flight — the in-process analog
/// of the named exclusive sweep lease (ADR-013 §2): a second
/// sweeper contending behind the first aborts as report data
/// (`SweeperLock`), nothing deleted.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_sweeps_single_flight_and_the_second_aborts_as_data() {
    let store = Arc::new(Store::mem_only());
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    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: parking_lot::Mutex::new(Some(release_rx)),
    }));

    let dead = key_for(b"dead entry under the contended sweeps");
    plant_kv(
        &store,
        &dead.as_bytes(),
        b"dead entry under the contended sweeps",
    )
    .await;

    let first = {
        let store = Arc::clone(&store);
        tokio::spawn(async move { store.sweep().await })
    };

    while entered_rx
        .recv_timeout(std::time::Duration::from_millis(500))
        .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.deleted, 0);
    assert!(
        store.has(&dead.as_bytes()).await.unwrap(),
        "the contended sweep deleted nothing"
    );

    release_tx.send(()).ok();
    let first = first
        .await
        .expect("first sweep task joined")
        .expect("first sweep returns a report");
    assert_eq!(first.aborted, None);
    assert_eq!(first.deleted, 1, "the lease holder sweeps");
    assert!(!store.has(&dead.as_bytes()).await.unwrap());
}

/// Sweeps surface operational failures as errors (ADR-012 §1's
/// error half): a malformed pool list is fatal `KeyInvalid`, never
/// a silent skip — through the sweep's own mark enumeration too.
#[tokio::test]
async fn sweep_of_a_malformed_pool_list_is_an_operational_error() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    plant_kv(&store, &[0x07; 33], b"stray in the table").await;

    assert_eq!(
        store.sweep().await.unwrap_err(),
        Error::KeyInvalid,
        "a malformed pool list is fatal through the sweep"
    );
}

// ------------------------------------------------------------------
// The put seam (store-put-paths): the two internal paths behind one
// signature, the three numbered unknown-length rules, pin-before-
// publish with dedup refcounting, the Batch seam, kv-only rejection
// (ADR-003/005/010/012 §6.2) — on the facade's write half.
// ------------------------------------------------------------------

use std::collections::VecDeque;
use std::io::Cursor;
use std::pin::Pin as TestPin;
use std::task::{Context, Poll};
use tokio::io::ReadBuf;

/// A stream cut at controlled chunk boundaries (`sizes`, in order;
/// tail data falls out whole) — the fixture that makes mid-stream
/// overflow, flush ordering, and staged continuation observable
/// past the put loop's own chunking.
struct Chunked {
    data: Vec<u8>,
    cursor: usize,
    sizes: VecDeque<usize>,
}

impl Chunked {
    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: TestPin<&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 (rule 3's shape).
struct Failing {
    data: Vec<u8>,
    cursor: usize,
    fail_after: usize,
}

impl Failing {
    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: TestPin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        let message = "stream failure (test 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 *immediately*, before content
/// flows (ADR-012 §6.2).
struct Unreadable;

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

/// Files currently in the local tier's staging area — the residue
/// counter of the store-level stage-hygiene assertions.
fn store_stage_file_count(root: &Path) -> usize {
    match fs::read_dir(root.join("staging")) {
        Ok(entries) => entries.count(),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => 0,
        Err(e) => panic!("staging dir unreadable: {e}"),
    }
}

/// INV-1, kv tier, known-length path: `put(Some(len), stream)`
/// round-trips byte-identical, lands in the routed tier (kv), and
/// the pin rides out with the put.
#[tokio::test]
async fn known_length_put_round_trips_on_the_kv_tier() {
    let root = temp_root("put-known-kv");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let content = b"a known-length small blob";
    let canonical = key_for(content);

    let (key, pin) = store
        .put(Some(content.len() as u64), Cursor::new(&content[..]))
        .await
        .expect("kv put publishes");
    assert_eq!(key, canonical, "the canonical derivation addresses the put");
    assert_eq!(store.gc().pin_count(&canonical), 1, "the pin rode out");
    assert!(
        store.kv().has(&canonical.as_bytes()).await.unwrap(),
        "landed in the routed tier (kv)"
    );
    assert!(
        !store
            .large()
            .unwrap()
            .has(&canonical.as_bytes())
            .await
            .unwrap(),
        "no large-tier copy on the kv route"
    );
    let (len, blob) = store
        .get(&canonical.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(len, content.len() as u64);
    assert_eq!(blob.read_at(0, len).await.unwrap(), &content[..]);

    drop(pin);
}

/// INV-1, large tier, known-length path: over-threshold known
/// content streams into the staged put with hashing in the same
/// pass, the derivation completing at stream end; commit-rename
/// publishes; zero stage files remain.
#[tokio::test]
async fn known_length_put_round_trips_on_the_large_tier_through_the_staged_put() {
    let root = temp_root("put-known-large");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let content = vec![42u8; 300];
    let canonical = key_for(&content);

    let (key, pin) = store
        .put(Some(300), Cursor::new(&content[..]))
        .await
        .expect("large staged put publishes");
    assert_eq!(key, canonical);
    assert_eq!(store.gc().pin_count(&canonical), 1);
    assert!(
        store
            .large()
            .unwrap()
            .has(&canonical.as_bytes())
            .await
            .unwrap(),
        "landed in the routed tier (large, commit-rename)"
    );
    assert!(
        !store.kv().has(&canonical.as_bytes()).await.unwrap(),
        "no kv copy on the large route"
    );
    let (len, blob) = store
        .get(&canonical.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(len, 300);
    assert_eq!(blob.read_at(0, len).await.unwrap(), content);
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "commit-rename left zero stage files"
    );

    drop(pin);
}

/// INV-1, both tiers, unknown-length path: sub-threshold content
/// commits through the buffered kv form; over-threshold content
/// flushed-through staging — both round-trip byte-identical and
/// both arrive under the canonical derivation over the eventual
/// stored artifact as a whole (rule 2 asserted by key equality:
/// anything but preamble + the wholeArtifact derivation would
/// derive a different key).
#[tokio::test]
async fn unknown_length_put_round_trips_on_both_tiers() {
    let root = temp_root("put-unknown");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");

    let small = b"small unknown-length content";
    let (key, pin) = store
        .put(None, Cursor::new(&small[..]))
        .await
        .expect("buffered put publishes");
    assert_eq!(key, key_for(small), "rule 2, buffer form");
    assert!(store.kv().has(&key.as_bytes()).await.unwrap());
    drop(pin);

    let big: Vec<u8> = (0..200u32).map(|i| (i % 251) as u8).collect();
    let (key, pin) = store
        .put(None, Chunked::new(&big, &[7; 30]))
        .await
        .expect("staged put publishes");
    assert_eq!(key, key_for(&big), "rule 2, staged-bytes form");
    assert!(
        store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
        "the over-threshold unknown put landed large"
    );
    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(len, big.len() as u64);
    assert_eq!(blob.read_at(0, len).await.unwrap(), big);
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "zero stage files on success"
    );
    drop(pin);
}

/// Rule 1 (ADR-003 option (d)): threshold exceeded mid-stream — the
/// buffered prefix flushes into a fresh stage file and the stream
/// *continues appending into it*; no restart anywhere, nothing
/// migrates tiers, and the flushed artifact derives canonically.
#[tokio::test]
async fn mid_stream_overflow_flushes_into_a_fresh_stage_and_the_stream_continues() {
    let root = temp_root("put-rule1-flush");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content: Vec<u8> = (0..40u32).map(|i| (i % 199) as u8).collect();
    // Chunks of 3: two fill the buffer to 6; the third (would-be 9)
    // crosses the threshold mid-stream and triggers the flush; the
    // rest stream on into the stage.
    let chunked = Chunked::new(&content, &[3; 14]);

    let (key, pin) = store
        .put(None, chunked)
        .await
        .expect("mid-stream overflow flushes; the stream continues");
    assert_eq!(
        key,
        key_for(&content),
        "the flushed artifact derives canonically"
    );
    assert!(
        !store.kv().has(&key.as_bytes()).await.unwrap(),
        "nothing migrated to the kv tier on a large-route put"
    );
    assert!(
        store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
        "the stage holds the whole streamed artifact"
    );
    let (_, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(
        blob.read_at(0, content.len() as u64).await.unwrap(),
        content,
        "the flushed prefix and the continued stream are one artifact"
    );
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "the stage published by rename; zero residue"
    );
    drop(pin);
}

/// The dispatch seam's boundary through the put path: an
/// unknown-length stream ending exactly at the threshold stays kv
/// (≤ threshold routes kv; the overflow is strictly past it).
#[tokio::test]
async fn unknown_length_exactly_at_threshold_stays_kv() {
    let root = temp_root("put-boundary");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content = b"01234567"; // exactly 8 bytes
    let chunked = Chunked::new(content, &[4, 4]);

    let (key, pin) = store
        .put(None, chunked)
        .await
        .expect("at-threshold stream commits through the buffer");
    assert_eq!(key, key_for(content));
    assert!(
        store.kv().has(&key.as_bytes()).await.unwrap(),
        "exactly-at-threshold stays kv"
    );
    assert!(
        !store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
        "no large-tier copy on the boundary"
    );
    drop(pin);
}

/// Rule 3 (ADR-003): a stream error before commit leaves no entry
/// and no stage file — on the buffered kv form, on the flushed
/// staged form, and on both known-length tiers; the stage-hygiene
/// invariant extended to the buffered case.
#[tokio::test]
async fn stream_errors_before_commit_leave_no_entry_and_no_stage_file() {
    let root = temp_root("put-rule3-error");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content: Vec<u8> = (0..40u32).map(|i| (i % 173) as u8).collect();

    // Flushed staged form: the overflow flush ran, then the stream
    // errored — the stage discards.
    let err = store
        .put(None, Failing::new(&content, 12))
        .await
        .unwrap_err();
    assert!(
        matches!(err, Error::Io(_)),
        "stream failures surface as Io: {err}"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "no entry landed from the errored staging put"
    );
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "no stage file remains"
    );

    // Buffered kv form: error before the threshold — the buffer
    // discards exactly like a stage.
    let err = store
        .put(None, Failing::new(&content, 5))
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Io(_)));
    assert_eq!(store.list().await.unwrap(), Vec::<Key>::new());
    assert_eq!(store_stage_file_count(root.path()), 0);

    // Known-length large path: mid-stage failure discards.
    let err = store
        .put(Some(40), Failing::new(&content, 12))
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Io(_)));
    assert_eq!(store.list().await.unwrap(), Vec::<Key>::new());
    assert_eq!(store_stage_file_count(root.path()), 0);

    // Known-length kv path: never published.
    let err = store
        .put(Some(40), Failing::new(&content, 5))
        .await
        .unwrap_err();
    assert!(matches!(err, Error::Io(_)));
    assert_eq!(store.list().await.unwrap(), Vec::<Key>::new());
    assert_eq!(store_stage_file_count(root.path()), 0);
}

/// The staged form's mid-write failure through the real engine's
/// injection (the store-core arms of review-001's D-2 sweep, driven
/// against the `local` large tier): a known-length large put's append
/// failing mid-stream, an unknown-length put's staged continuation
/// failing, and both flush-phase forms (the buffered-prefix append and
/// the overflowing chunk's append) — every one discards the stage, fails
/// the put, and leaves the pool exact (no entries, no stage files).
#[tokio::test]
async fn injected_append_failures_discard_the_staged_put_zero_residue() {
    // Known-length large put: the first stage append fails mid-stream.
    let root = temp_root("put-append-fail-known");
    let mut store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    match &mut store.large {
        Some(LargeEngine::Local(backend)) => backend.inject_fail_append(0),
        _ => panic!("the mem+local dual tier is required for the local injection"),
    }
    let content: Vec<u8> = (0..40u32).map(|i| (i % 173) as u8).collect();
    let err = store
        .put(Some(content.len() as u64), Cursor::new(&content[..]))
        .await
        .unwrap_err();
    assert!(
        err.to_string().contains("injected"),
        "the staged append failure surfaces: {err}"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "no entry landed from the failed stage"
    );
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "zero stage files after the known-length failure"
    );

    // Unknown-length staged form, two shapes of the flush phase and
    // the staged continuation, over piecemeal reads so the buffered
    // prefix is non-empty and every append is observable.
    for (after, stage_of_failure) in [
        (0, "the buffered prefix's first flush append"),
        (1, "the overflowing chunk's flush append"),
        (2, "the staged continuation's append"),
    ] {
        let root = temp_root("put-append-fail-unknown");
        let mut store = Store::dual_tier_mem_local(root.path(), 8)
            .await
            .expect("dual-tier constructs");
        match &mut store.large {
            Some(LargeEngine::Local(backend)) => backend.inject_fail_append(after),
            _ => panic!("the mem+local dual tier is required for the local injection"),
        }
        let stream = Chunked::new(&content, &[5, 5, 5, 5, 5, 5, 5, 5]);
        let err = store.put(None, stream).await.unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "{stage_of_failure} failed the put: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "{stage_of_failure}: no entry landed"
        );
        assert_eq!(
            store_stage_file_count(root.path()),
            0,
            "{stage_of_failure}: zero stage files"
        );
    }
}

/// Pin-before-publish (ADR-005) on both put paths and both tiers:
/// a sweep running alongside an in-flight put never deletes the
/// protected entry — the pin is minted before the entry can appear
/// in any `list()` — and every completed schedule keeps exact
/// counts. The surviving pin protects; dropping it re-exposes.
#[tokio::test]
async fn pin_before_publish_holds_on_both_put_paths_and_both_tiers() {
    let root = temp_root("put-pin");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let big: Vec<u8> = (0..200u32).map(|i| (i % 241) as u8).collect();
    let big_other: Vec<u8> = (0..288u32).map(|i| (i % 223) as u8).collect();
    let rounds: [(Vec<u8>, Option<u64>); 4] = [
        (b"pinned known small".to_vec(), Some(18)),
        (b"still-pinned unknown small content".to_vec(), None),
        (big, Some(200)),
        (big_other, None),
    ];

    for (round, (content, len)) in rounds.iter().enumerate() {
        let expected = key_for(content);
        let (report, put) = tokio::join!(store.sweep(), async {
            store.put(*len, Cursor::new(&content[..])).await
        });
        let report = report.expect("sweeps are never errors");
        let (key, pin) = put.expect("put published");
        assert_eq!(key, expected, "round {round}");
        assert_eq!(report.aborted, None, "round {round}");
        assert_eq!(
            report.candidates_staged,
            report.deleted + report.cancelled_by_arbitration,
            "round {round}: exact counts in every schedule"
        );
        assert!(
            store.has(&key.as_bytes()).await.unwrap(),
            "round {round}: the entry survived the race"
        );
        assert_eq!(
            store.gc().pin_count(&expected),
            1,
            "round {round}: the minted pin still protects"
        );
        let (_, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(
            blob.read_at(0, content.len() as u64).await.unwrap(),
            &content[..],
            "round {round}: the published entry is byte-identical"
        );

        drop(pin);
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(report.deleted, 1, "round {round}: unpinned, deleted");
        assert!(!store.has(&key.as_bytes()).await.unwrap());
    }
}

/// Dedup (store-api.md invariant 2; ADR-005): a re-put of identical
/// content acquires no new pool entry and refcounts the existing
/// one — across put paths (known vs unknown) and across both tiers;
/// the refcounts compose under the same lock and drop in sequence.
#[tokio::test]
async fn dedup_reputs_refcount_the_existing_entry_from_any_path() {
    let root = temp_root("put-dedup");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let content = b"deduplicated content";
    let (known_key, first) = store
        .put(Some(content.len() as u64), Cursor::new(&content[..]))
        .await
        .expect("known-path put");
    let (unknown_key, second) = store
        .put(None, Cursor::new(&content[..]))
        .await
        .expect("unknown-path re-put");
    assert_eq!(known_key, unknown_key, "one derivation, one address");
    assert_eq!(
        store.list().await.unwrap(),
        vec![known_key.clone()],
        "one pool entry"
    );
    assert_eq!(
        store.gc().pin_count(&known_key),
        2,
        "the re-put refcounted the existing entry"
    );

    drop(first);
    assert_eq!(store.gc().pin_count(&known_key), 1);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.candidates_staged, 0, "one refcount still protects");

    drop(second);
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1);
    assert!(!store.has(&known_key.as_bytes()).await.unwrap());

    // The same composition across the large tier: two staged puts
    // of identical content — one entry, two refcounts.
    let big = [91u8; 200];
    let (first_key, first) = store
        .put(Some(200), Cursor::new(&big[..]))
        .await
        .expect("known staged put");
    let (_, second) = store
        .put(None, Cursor::new(&big[..]))
        .await
        .expect("unknown staged re-put");
    assert_eq!(
        store.list().await.unwrap(),
        vec![first_key.clone()],
        "one pool entry across both staged paths"
    );
    assert_eq!(store.gc().pin_count(&first_key), 2);
    drop(first);
    drop(second);
}

/// `Batch` (store-api.md §Write path; ADR-005 §Marks 2): multi-put
/// writes group under one batch-scoped pin — members publish pin-
/// before-publish exactly as lone puts do, both tiers, and the
/// group's pins drop together on batch commit.
#[tokio::test]
async fn batch_groups_multi_put_writes_and_members_drop_together() {
    let root = temp_root("put-batch");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let small_a = b"manifest member a";
    let small_b = b"manifest member b";
    let big = vec![77u8; 200];

    let mut batch = store.batch();
    assert!(batch.is_empty(), "a fresh batch protects nothing");
    let key_a = store
        .put_in_batch(&mut batch, Some(17), Cursor::new(&small_a[..]))
        .await
        .expect("batch member a");
    let key_b = store
        .put_in_batch(&mut batch, None, Cursor::new(&small_b[..]))
        .await
        .expect("batch member b");
    let key_big = store
        .put_in_batch(&mut batch, Some(200), Cursor::new(&big[..]))
        .await
        .expect("batch member large");
    assert_eq!(batch.len(), 3);
    assert_eq!(key_a, key_for(small_a));
    assert_eq!(key_b, key_for(small_b));
    assert_eq!(key_big, key_for(&big));

    for key in [&key_a, &key_b, &key_big] {
        assert_eq!(
            store.gc().pin_count(key),
            1,
            "each member's pin rides in the group"
        );
    }
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.candidates_staged, 0,
        "the batch-scoped pin protects the whole group"
    );
    assert!(store.has(&key_a.as_bytes()).await.unwrap());
    assert!(store.has(&key_big.as_bytes()).await.unwrap());

    batch.commit();
    for key in [&key_a, &key_b, &key_big] {
        assert_eq!(
            store.gc().pin_count(key),
            0,
            "no early single-pin release and no late release either"
        );
    }
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 3, "unprotected after the batch ends");
    assert!(!store.has(&key_a.as_bytes()).await.unwrap());
    assert!(!store.has(&key_b.as_bytes()).await.unwrap());
    assert!(!store.has(&key_big.as_bytes()).await.unwrap());
}

/// The foreign-Batch refusal (review-001 N-5): a batch minted by one
/// store is rejected in `put_in_batch` on another store — the
/// `debug_assert!(Arc::ptr_eq)` on the batch's registry Arc against
/// the store's fires before any pin is minted (debug builds run it;
/// the recorded point over a named error — the error family is
/// ADR-pinned, ADR-012 §6.5).
#[tokio::test]
#[should_panic(expected = "foreign Store")]
async fn put_in_batch_refuses_a_foreign_batch_in_debug_builds() {
    let store = Store::mem_only();
    let mut foreign_batch = Store::mem_only().batch();
    let content = b"cross-store mis-use";
    let _ = store
        .put_in_batch(
            &mut foreign_batch,
            Some(content.len() as u64),
            Cursor::new(&content[..]),
        )
        .await;
}

/// Batch drop is the failure path: a member put error leaves the
/// landed members batch-protected and contributes nothing to the
/// group; member liveness never outlives the batch.
#[tokio::test]
async fn batch_member_put_failure_does_not_endanger_landed_members() {
    let root = temp_root("put-batch-failure");
    let store = Store::dual_tier_mem_local(root.path(), 64)
        .await
        .expect("dual-tier constructs");
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let landed = b"the member that landed";
    let mut batch = store.batch();
    let key = store
        .put_in_batch(&mut batch, Some(22), Cursor::new(&landed[..]))
        .await
        .expect("landed member published");
    assert_eq!(store.gc().pin_count(&key), 1);

    let content: Vec<u8> = (0..40u32).map(|i| (i % 157) as u8).collect();
    let err = store
        .put_in_batch(&mut batch, None, Failing::new(&content, 12))
        .await
        .unwrap_err();
    assert!(
        matches!(err, Error::Io(_)),
        "the failed member surfaces its error"
    );
    assert_eq!(batch.len(), 1, "the failed member joined nothing");
    assert_eq!(
        store.gc().pin_count(&key),
        1,
        "landed members stay protected"
    );
    assert_eq!(
        store.list().await.unwrap(),
        vec![key.clone()],
        "the failed member published no entry"
    );

    drop(batch);
    assert_eq!(
        store.gc().pin_count(&key),
        0,
        "batch drop released every member together"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1);
}

/// kv-only mode's put-time rejection (ADR-012 §6.2): known-length
/// over-threshold rejects *immediately* (the stream is never read);
/// unknown-length rejects at mid-stream threshold overflow — zero
/// residue on either rejection, and at/exactly-threshold puts land.
#[tokio::test]
async fn kv_only_mode_rejects_over_threshold_puts_at_the_two_points() {
    let store = Store::kv_only_mem(64);
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));

    let err = store.put(Some(65), Unreadable).await.unwrap_err();
    assert!(
        err.to_string().contains("kv-only mode rejects"),
        "the immediate rejection carries its carrier message: {err}"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "zero residue"
    );

    let at_threshold = vec![9u8; 64];
    let (key, pin) = store
        .put(Some(64), Cursor::new(&at_threshold[..]))
        .await
        .expect("exactly-at-threshold known put lands");
    assert_eq!(key, key_for(&at_threshold));
    assert!(store.has(&key.as_bytes()).await.unwrap());
    drop(pin);

    let streamed = vec![3u8; 100];
    let err = store
        .put(None, Chunked::new(&streamed, &[10; 10]))
        .await
        .unwrap_err();
    assert!(
        err.to_string()
            .contains("kv-only mode rejects an over-threshold put"),
        "the overflow rejection is the mode's, not the stream's: {err}"
    );
    assert_eq!(
        store.list().await.unwrap(),
        vec![key_for(&at_threshold)],
        "zero residue"
    );

    let just_under = vec![5u8; 64];
    let (key, pin) = store
        .put(None, Chunked::new(&just_under, &[8; 8]))
        .await
        .expect("at-threshold unknown stream lands through the buffer");
    assert_eq!(key, key_for(&just_under));
    assert!(store.has(&key.as_bytes()).await.unwrap());
    drop(pin);
}

/// The declared-len mismatch (the put seam's named design point):
/// a known-length stream yielding fewer or more bytes than `len`
/// rejects with `Error::Verification` on either tier — zero residue
/// either way (nothing published, no stage file left).
#[tokio::test]
async fn declared_len_mismatch_rejects_verification_with_zero_residue() {
    let root = temp_root("put-mismatch");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let content = b"0123456789"; // 10 bytes

    assert_eq!(
        store
            .put(Some(5), Cursor::new(&content[..]))
            .await
            .unwrap_err(),
        Error::Verification,
        "more bytes than declared (kv route)"
    );
    assert_eq!(
        store
            .put(Some(20), Cursor::new(&content[..]))
            .await
            .unwrap_err(),
        Error::Verification,
        "fewer bytes than declared (kv route)"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "zero residue on the kv mismatches"
    );

    let big = [5u8; 32];
    assert_eq!(
        store
            .put(Some(16), Cursor::new(&big[..]))
            .await
            .unwrap_err(),
        Error::Verification,
        "more bytes than declared (large route)"
    );
    assert_eq!(
        store
            .put(Some(64), Cursor::new(&big[..]))
            .await
            .unwrap_err(),
        Error::Verification,
        "fewer bytes than declared (large route)"
    );
    assert_eq!(
        store.list().await.unwrap(),
        Vec::<Key>::new(),
        "zero residue on the staged mismatches"
    );
    assert_eq!(
        store_stage_file_count(root.path()),
        0,
        "the discarded stages left zero stage files"
    );
}

/// mem-only accepts every length on both put paths — no dispatch
/// threshold in effect (ADR-011 §4), nothing to reject; the sole
/// mem engine holds even over-default-threshold content.
#[tokio::test]
async fn mem_only_accepts_every_length_on_both_put_paths() {
    let store = Store::mem_only();
    let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
        SetSource::default(),
    )));
    assert_eq!(store.threshold(), None, "no threshold in effect");

    let small = b"mem-only content";
    let (key, pin) = store
        .put(Some(16), Cursor::new(&small[..]))
        .await
        .expect("known put");
    assert_eq!(key, key_for(small));
    drop(pin);

    let big = vec![1u8; (DEFAULT_DISPATCH_THRESHOLD + 1) as usize];
    let (key, pin) = store
        .put(None, Cursor::new(&big[..]))
        .await
        .expect("over-default-threshold unknown put routes to the sole kv engine");
    assert_eq!(key, key_for(&big));
    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("entry present");
    assert_eq!(len, big.len() as u64);
    assert_eq!(blob.read_at(0, len).await.unwrap(), big);
    drop(pin);
}

/// The zero-length degenerate on both put paths: empty content
/// round-trips through its canonical key (one entry, both paths
/// dedup into it), reading empty-never-error.
#[tokio::test]
async fn zero_length_puts_round_trip_on_both_paths() {
    let root = temp_root("put-empty");
    let store = Store::dual_tier_mem_local(root.path(), 8)
        .await
        .expect("dual-tier constructs");
    let empty = b"";

    let (known, pin) = store
        .put(Some(0), Cursor::new(&empty[..]))
        .await
        .expect("known empty put");
    assert_eq!(known, key_for(empty));
    assert!(store.kv().has(&known.as_bytes()).await.unwrap());
    drop(pin);

    let (unknown, pin) = store
        .put(None, Cursor::new(&empty[..]))
        .await
        .expect("unknown empty put");
    assert_eq!(unknown, known, "empty content dedups into one entry");
    assert_eq!(store.list().await.unwrap(), vec![known.clone()]);
    drop(pin);

    let (len, blob) = store
        .get(&known.as_bytes())
        .await
        .unwrap()
        .expect("entry present");
    assert_eq!(len, 0);
    assert!(blob.is_empty());
    assert_eq!(blob.read_at(0, 8).await.unwrap(), Vec::<u8>::new());
}