//! The **postgres column** — the suite's first SKIP-posture column:
//! the reference row set run against the fleet-capable-kv shape
//! (`kv = postgres` + `large = local`; feature `postgres` is
//! default-off, ADR-007, ADR-012 §5), extended with the
//! **durable-schedule sweep rows** and the dispatch rows (README
//! §Columns beyond the reference set; INV-7: every engine's rows
//! mirror the reference column).
//!
//! The SKIP posture (README §The SKIP posture) is a factory-owner
//! responsibility here: the connection settings ride the environment
//! (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` — host required; the
//! rest default to the harness server's values), the runners gate
//! every row behind a **reachability probe**, and when no server is
//! reachable the column skips cleanly — a skip is not a failure. The
//! acceptance run is green **against the harness server** (dockerized
//! `pglo-poc`, `postgres:16-alpine`, :15432, db `blobs`) — a skip
//! never substitutes for it. The same server hosts the sqlite and
//! local columns' neighbors' needs; nothing is skipped server-less
//! except this column's rows.
//!
//! The sweep rows are **schedule-relative on this column** (the
//! README §The schedule-relative reading for SQL-backed columns is
//! the recorded amendment, citing ADR-012 §6.7): the durable form
//! stages its candidates this sweep and executes them by the next,
//! so the exact counts here are asserted against the staged schedule.
//! The form-independent rows mirror verbatim ([`crate::rows`]: the
//! round-trip family, the stat-discipline row, the no-sources and
//! failed-protect abort shapes, and the kv-only rejection row against
//! the kv-only factory); the schedule-relative properties grew their
//! own rows below (dedup and stage hygiene observe their invariants
//! through sweep outcomes, so they read against the schedule too).
//! Mem-only never mirrors here: the mode is not constructible
//! without mem.
//!
//! The factories (README §The factory contract; the alkstore
//! pg-factory pattern — alkstore-postgres `contract_suite.rs` —
//! reused): `open()` mints a **fresh engine-owned schema per call** —
//! unique per call (seeded per tag/pid/time so two factory instances
//! never share a schema name, racing opens included), tracked
//! per-instance in the schema registry — and vends the concrete
//! [`alkblobs::Store`] facade over it plus a fresh private temp
//! directory for the large engine's root, so two opens never share
//! state even on the shared server. Teardown `DROP SCHEMA IF EXISTS
//! … CASCADE` over exactly the schemas this factory instance minted
//! (idempotent, owned — never the shared server's other schemas;
//! safe against an abandoned store: `CASCADE` removes the dependents
//! server-side while the store's pooled connections fail closed on
//! next use) plus the removal of every temp root it vended. The
//! sqlite column's instance-tracks-its-own-site pattern over schemas
//! (plus roots): every runner constructs its own factory, so
//! concurrent rows never touch each other's schemas.
//!
//! The dual-tier dispatch rows mirror the sqlite column's family with
//! the placement oracle adapted by medium again: kv entries live as
//! **rows in the pg schema's `blobs` table**, large entries under the
//! local root — the oracle distinguishes by medium (an independent
//! connection counting the schema's rows vs the root's committed
//! files), and watches the schedule itself (the schema's `gc_window`
//! rows).

#![allow(dead_code)]

use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use tokio_postgres::NoTls;

use crate::factory::StoreFactory;
use crate::rows::{
    self, committed_file_count, patterned, slice_digest, staging_file_count, Failing, LatchProtect,
    SetProtect, SetSource,
};
use alkblobs::{Error, Key, PgOpts, Store, StoreMode, DEFAULT_DISPATCH_THRESHOLD};

// ------------------------------------------------------------------
// The factories
// ------------------------------------------------------------------

/// The column's declared dual-tier threshold — small, so the shared
/// content matrix and the race fixtures cross it and both tiers'
/// bodies are exercised (sub-threshold rows land in the pg schema;
/// over-threshold rows stage-commit into the local root).
pub(crate) const TUNED_THRESHOLD: u64 = 1024;

/// The column's declared kv-only rejection threshold — small, so every
/// leg of the rejection row stays trivially materializable.
pub(crate) const KV_ONLY_THRESHOLD: u64 = 1024;

/// The **test tuning of the window delay** (ADR-008 §2's legal
/// degenerate tuning): `0` — execute-on-next-sweep. The durable
/// schedule's staging sweep stages the dead set; the *next* sweep is
/// its executor, which is exactly the schedule the sweep rows assert
/// against.
pub(crate) const WINDOW_DELAY: Duration = Duration::ZERO;

/// One vended backing site: the engine-owned schema this open minted
/// plus a fresh private temporary directory holding the large
/// engine's root.
#[derive(Clone)]
pub(crate) struct Site {
    /// The server's DSN — the oracle's independent connections and the
    /// kv tier's shared medium address. Two opens never share a
    /// schema; isolation is schema-scoped, not server-scoped.
    dsn: String,
    /// The engine-owned schema this open minted — the kv tier's
    /// medium (the dispatch oracle's kv-side counter rides the
    /// schema's table rows).
    schema: String,
    /// The whole private directory — what teardown removes (the local
    /// root's parent; kv-only sites leave it empty and still remove
    /// it).
    base: PathBuf,
    /// The local root — the large tier's medium (the dispatch
    /// oracle's file-side counter rides [`committed_file_count`]).
    root: PathBuf,
}

impl Site {
    /// The schema-qualified pool table's count via the oracle.
    pub(crate) fn schema(&self) -> &str {
        &self.schema
    }

    /// The local root path — for the file oracle's callers.
    pub(crate) fn root(&self) -> &Path {
        &self.root
    }

    fn remove_site(self) -> Result<(), Error> {
        match std::fs::remove_dir_all(&self.base) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(Error::Io(e.to_string())),
        }
    }
}

/// A fresh private temp directory for one open's local root.
fn fresh_root(tag: &str) -> PathBuf {
    static SEQ: AtomicU64 = AtomicU64::new(0);
    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let base = std::env::temp_dir().join(format!(
        "alkblobs-postgres-column-{tag}-{}-{:x}-{seq}",
        std::process::id(),
        nanos
    ));
    std::fs::create_dir_all(&base).expect("temp site created");
    base
}

/// The harness connection settings ride the environment
/// (`ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB` — the alkstore harness
/// convention). `ALKBLOBS_PG_HOST` set (non-empty) is the probe's
/// "configured" answer; port/user/password/db default to the harness
/// server's values (`pglo-poc`: :15432, `postgres`/`poc`, db
/// `blobs`). Absent → the caller skips.
pub(crate) fn harness_dsn() -> Option<String> {
    let host = std::env::var("ALKBLOBS_PG_HOST")
        .ok()
        .filter(|v| !v.is_empty())?;
    let port = std::env::var("ALKBLOBS_PG_PORT")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "15432".to_owned());
    let user = std::env::var("ALKBLOBS_PG_USER")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "postgres".to_owned());
    let password = std::env::var("ALKBLOBS_PG_PASSWORD")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "poc".to_owned());
    let db = std::env::var("ALKBLOBS_PG_DB")
        .ok()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| "blobs".to_owned());
    Some(format!(
        "host={host} port={port} user={user} password={password} dbname={db}"
    ))
}

/// The skip note — the SKIP posture's clean-skip print, exactly once
/// per skipped test.
fn skip_note() {
    eprintln!("skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB");
}

async fn admin_connect(dsn: &str) -> Result<tokio_postgres::Client, Error> {
    let (client, connection) = tokio_postgres::connect(dsn, NoTls)
        .await
        .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    Ok(client)
}

/// The server reachability probe: the SKIP posture's gate the runners
/// check before their rows run. One cheap admin connection per test;
/// `teardown` reconnects independently (no shared admin client
/// lifetime).
pub(crate) async fn harness_ready() -> bool {
    let Some(dsn) = harness_dsn() else {
        skip_note();
        return false;
    };
    match tokio::time::timeout(Duration::from_secs(3), admin_connect(&dsn)).await {
        Ok(Ok(_)) => true,
        Ok(Err(e)) => {
            eprintln!("skipping (harness server unreachable): {e}");
            false
        }
        Err(_) => {
            eprintln!("skipping (harness server unreachable: connect timeout)");
            false
        }
    }
}

/// A fresh engine-owned schema name — unique per call, seeded per
/// (tag, pid, nanos, factory-internal sequence) so two factory
/// instances never share a schema name, racing opens included.
fn fresh_schema_name(tag: &str, seq: &AtomicU64) -> String {
    format!(
        "alkblobs_suite_{}_{}_{}_{}",
        tag,
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0),
        seq.fetch_add(1, Ordering::Relaxed),
    )
}

/// The dispatch-coverage factory: `kv = postgres` + `large = local` at
/// the threshold this column declares and the test tuning of the
/// window delay (see [`TUNED_THRESHOLD`], [`WINDOW_DELAY`], and the
/// default-threshold sibling `DualTierPostgresFactory::new(
/// DEFAULT_DISPATCH_THRESHOLD)` the boundary row runs against).
///
/// **Why a fresh instance per runner (not a shared static):** teardown
/// drops the schemas *this instance* minted (and removes the roots it
/// vended); a shared instance's teardown would race concurrently
/// running rows' sites. Every runner constructs its own and owns its
/// lifecycle.
pub(crate) struct DualTierPostgresFactory {
    dsn: String,
    /// The declared dispatch threshold — the routing boundary this
    /// column's rows run at (constructor-tunable; ADR-003). Declared
    /// configuration, like the local column's (the dual-tier store
    /// never rejects, so there is nothing to probe).
    threshold: u64,
    /// The declared window delay — the durable schedule's cadence knob
    /// the sweep rows run at (the test tuning: execute-on-next-sweep).
    window_delay: Duration,
    schema_seq: AtomicU64,
    /// Every schema this instance minted, for its own teardown.
    schemas: Mutex<Vec<String>>,
    /// Every temp site this instance vended, for its own teardown.
    sites: Mutex<Vec<PathBuf>>,
}

impl DualTierPostgresFactory {
    pub(crate) fn new(threshold: u64) -> DualTierPostgresFactory {
        DualTierPostgresFactory {
            dsn: harness_dsn().expect("the harness server is configured for the column"),
            threshold,
            window_delay: WINDOW_DELAY,
            schema_seq: AtomicU64::new(
                (std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.subsec_nanos() as u64)
                    .unwrap_or(0))
                    ^ ((std::process::id() as u64) << 8),
            ),
            schemas: Mutex::new(Vec::new()),
            sites: Mutex::new(Vec::new()),
        }
    }

    /// Open + track the vended site: the shape the column's own rows
    /// use (they observe the backing medium, so they need the site
    /// alongside the store).
    pub(crate) async fn open_tracked(&self) -> Result<(Store, Site), Error> {
        let schema = fresh_schema_name("dual-tier", &self.schema_seq);
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(schema.clone());
        let base = fresh_root("dual-tier");
        let site = Site {
            dsn: self.dsn.clone(),
            schema: schema.clone(),
            root: base.join("local-root"),
            base,
        };
        let store = Store::dual_tier_postgres_local(
            &self.dsn,
            opts(schema),
            site.root.clone(),
            self.threshold,
            self.window_delay,
        )
        .await?;
        self.sites
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(site.base.clone());
        Ok((store, site))
    }

    /// The schemas this instance minted, for the structural tests'
    /// teardown assertions.
    pub(crate) fn tracked_schemas(&self) -> Vec<String> {
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// The DSN the factory opens against.
    pub(crate) fn dsn(&self) -> &str {
        &self.dsn
    }
}

/// The kv-only(postgres) factory (ADR-011 §4; ADR-012 §6.2): no large
/// tier; the threshold is the put-time rejection point the kv-only
/// row discovers by probing. The fleet host's kv-only shape.
pub(crate) struct KvOnlyPostgresFactory {
    dsn: String,
    /// The rejection point this column's configuration declares. The
    /// kv-only row discovers it behaviorally and requires it
    /// materializable for its boundary legs.
    pub(crate) threshold: u64,
    window_delay: Duration,
    schema_seq: AtomicU64,
    schemas: Mutex<Vec<String>>,
}

impl KvOnlyPostgresFactory {
    pub(crate) fn new(threshold: u64) -> KvOnlyPostgresFactory {
        KvOnlyPostgresFactory {
            dsn: harness_dsn().expect("the harness server is configured for the column"),
            threshold,
            window_delay: WINDOW_DELAY,
            schema_seq: AtomicU64::new(
                (std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map(|d| d.subsec_nanos() as u64)
                    .unwrap_or(0))
                    ^ ((std::process::id() as u64) << 8),
            ),
            schemas: Mutex::new(Vec::new()),
        }
    }
}

/// The factories' schema-name policy, in engine-opts form: the
/// constructor's `PgOpts` with exactly the factory's minted schema
/// and the ship defaults for the rest (pool size,
/// `synchronous_commit` — the alkstore factory's shape: only the
/// schema name is factory policy).
fn opts(schema: String) -> PgOpts {
    PgOpts {
        schema,
        ..PgOpts::default()
    }
}

impl StoreFactory for DualTierPostgresFactory {
    fn name(&self) -> &'static str {
        "dual-tier postgres+local"
    }

    async fn open(&self) -> Result<Store, Error> {
        self.open_tracked().await.map(|(store, _)| store)
    }

    async fn teardown(&self) -> Result<(), Error> {
        let schemas = self
            .schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        let sites = self
            .sites
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        if !schemas.is_empty() {
            let client = admin_connect(&self.dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                client
                    .batch_execute(&sql)
                    .await
                    .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
            }
        }
        for base in sites {
            match std::fs::remove_dir_all(&base) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(Error::Io(e.to_string())),
            }
        }
        Ok(())
    }
}

impl StoreFactory for KvOnlyPostgresFactory {
    fn name(&self) -> &'static str {
        "kv-only(postgres)"
    }

    async fn open(&self) -> Result<Store, Error> {
        let schema = fresh_schema_name("kv-only", &self.schema_seq);
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(schema.clone());
        let store =
            Store::kv_only_postgres(&self.dsn, opts(schema), self.threshold, self.window_delay)
                .await?;
        Ok(store)
    }

    async fn teardown(&self) -> Result<(), Error> {
        let schemas = self
            .schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        if !schemas.is_empty() {
            let client = admin_connect(&self.dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                client
                    .batch_execute(&sql)
                    .await
                    .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
            }
        }
        Ok(())
    }
}

// ------------------------------------------------------------------
// The medium oracle (the dispatch rows' instrument, adapted from the
// sqlite column's): both tiers are on the server's disk and the local
// root now, so the oracle distinguishes by medium — an independent
// connection counting the schema's tables vs the root's committed
// files.
// ------------------------------------------------------------------

/// Rows in one of the schema's tables — the kv tier's pool (`blobs`)
/// or the durable window's staged candidates (`gc_window`) — counted
/// from the oracle's own independent connection to the shared server.
async fn pg_table_rows(site: &Site, table: &str) -> usize {
    let client = admin_connect(&site.dsn)
        .await
        .expect("the oracle connection opens");
    let sql = format!("SELECT COUNT(*) FROM \"{}\".{table}", site.schema());
    let row = client
        .query_one(&sql, &[])
        .await
        .expect("the oracle query succeeds");
    let count: i64 = row.get(0);
    count.max(0) as usize
}

/// The pg kv tier's pool residency — exactly one row per committed
/// kv-tier entry (the CAS table's rows).
async fn pg_pool_rows(site: &Site) -> usize {
    pg_table_rows(site, "blobs").await
}

/// The durable window's staged-candidate residency — the schedule
/// observer the durable rows assert against (staged rows land here;
/// the executing sweep's take empties them).
async fn pg_window_rows(site: &Site) -> usize {
    pg_table_rows(site, "gc_window").await
}

/// Whether the engine-owned schema still exists server-side — the
/// structural teardown tests' instrument (`pg_namespace`, the same
/// probe the alkstore factory shape uses).
async fn schema_exists(dsn: &str, schema: &str) -> bool {
    let client = admin_connect(dsn)
        .await
        .expect("the oracle connection opens");
    let count: i64 = client
        .query_one(
            "SELECT count(*) FROM pg_namespace WHERE nspname = $1",
            &[&schema],
        )
        .await
        .expect("the oracle query succeeds")
        .get(0);
    count > 0
}

// ------------------------------------------------------------------
// The column's own rows — the durable-schedule sweep family, the
// schedule-relative readings of the sweep-counting INV rows, and the
// dispatch forms. Every row takes this column's concrete factory (it
// observes the backing medium), is ADR-stamped per the convention,
// and traces its family.
// ------------------------------------------------------------------

/// INV-2 on the durable schedule: identical content put twice —
/// through the same put path and across the two paths — is one pool
/// entry (one row in the schema), and the re-put's `Pin` protects
/// exactly as a fresh put's does. One-pool-entry is asserted through
/// durable-schedule outcomes (the INV-5 posture): the pins held stage
/// nothing; once unpinned, sweep 1 stages exactly the one entry and
/// removes nothing, and its executor sweep deletes exactly that
/// staged set — never a list comparison, and never a single-sweep
/// count (README §The schedule-relative reading for SQL-backed
/// columns).
///
/// 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); ADR-008 §2 (the
/// staged re-arbitrated delete); ADR-012 §6.7 (the executor is the
/// next sweep — the schedule these counts read against).
pub async fn dedup_identical_content_is_one_pool_entry(factory: &DualTierPostgresFactory) {
    let (store, site) = factory
        .open_tracked()
        .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"
    );
    assert_eq!(
        pg_pool_rows(&site).await,
        1,
        "one pool entry, both refcounts — one row in the schema"
    );
    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.candidates_staged, report.deleted),
        (1, 0),
        "exactly one pool entry all along — staged this sweep, not executed"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (report.deleted, report.cancelled_by_arbitration),
        (1, 0),
        "the executing sweep deleted exactly the staged entry"
    );
    assert!(!store.has(&canonical.as_bytes()).await.unwrap());
    assert_eq!(pg_pool_rows(&site).await, 0, "the executed delete landed");

    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.candidates_staged, report.deleted),
        (1, 0),
        "exactly one pool entry all along"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "executed exactly the staged set");
    assert!(!store.has(&cross_canonical.as_bytes()).await.unwrap());

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

/// INV-3 on the durable schedule: 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 — the schema's
/// rows, which the failures never touch — observed through
/// durable-schedule sweep outcomes: zero staged after every failure,
/// and a success staged exactly once, then executed exactly once.
///
/// 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);
/// ADR-012 §6.7 (the schedule-relative reading the success counts
/// follow).
pub async fn stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(
    factory: &DualTierPostgresFactory,
) {
    let (store, site) = factory
        .open_tracked()
        .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"
    );
    assert_eq!(
        pg_pool_rows(&site).await,
        0,
        "the pool table holds zero entries after every failure"
    );

    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, 0),
        "exactly one committed entry staged on success"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        report.deleted, 1,
        "the executing sweep removed exactly that one entry"
    );
    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, 0),
        "exactly one committed entry staged on the other put path too"
    );
    let report = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(report.deleted, 1, "and executed exactly that staged set");
    assert!(!store.has(&canonical.as_bytes()).await.unwrap());

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

/// INV-4, the durable-schedule happy shape: the full protection set
/// spanning both tiers (a source-live kv entry, a pinned large-tier
/// entry, a protect-named large-tier entry) plus one dead entry per
/// tier — sweep 1 stages exactly the two dead candidates and deletes
/// nothing (staging is not executing: the pool and the medium hold
/// everything and the window holds the staged rows); sweep 2 executes
/// exactly that staged set (`deleted + cancelled == N` against the
/// staged set), and the protected entries stay byte-identical. The
/// exact staged candidates also make this the INV-5 posture's
/// both-tiers form: list correctness (complete and well-formed across
/// both tiers' enumerations) is observable only through these
/// outcomes. The medium oracle distinguishes the tiers (the schema's
/// pool rows vs the root's committed files) and watches the schedule
/// itself (the window's rows).
///
/// Contract stamp: store-api.md §Invariants 4 (exact counts with
/// correct liveness) and §Invariants 5 (list correctness through
/// sweep outcomes); ADR-005 (three liveness marks; delete windows
/// with arbitrate-at-delete-time); ADR-008 §2 (the staged
/// re-arbitrated delete; the constructor-tunable delay); ADR-012 §1
/// (sweeps are never errors — the report is the outcome record);
/// ADR-012 §6.7 (the executor is the next sweep — the schedule these
/// counts read against); ADR-007 §5 (postgres, the fleet host's kv
/// engine — the exact-count sweep-outcome gate against dockerized
/// postgres).
pub async fn two_sweep_exact_counts_over_the_full_protection_set(
    factory: &DualTierPostgresFactory,
) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .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 = patterned((threshold - 1) as usize, 191);
    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 = patterned(2 * threshold as usize + 7, 173);
    let (key_pinned, pin_pinned) = store
        .put(None, Cursor::new(pinned.as_slice()))
        .await
        .expect("the pinned entry publishes");

    let protected = patterned(2 * threshold as usize + 11, 167);
    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_kv = patterned((threshold - 3) as usize, 149);
    let (key_dead_kv, pin) = store
        .put(Some(dead_kv.len() as u64), Cursor::new(dead_kv.as_slice()))
        .await
        .expect("the kv-tier dead entry publishes");
    drop(pin);
    let dead_large = patterned(2 * threshold as usize + 13, 151);
    let (key_dead_large, pin) = store
        .put(None, Cursor::new(dead_large.as_slice()))
        .await
        .expect("the large-tier dead entry publishes");
    drop(pin);

    assert_eq!(
        (pg_pool_rows(&site).await, committed_file_count(site.root())),
        (2, 3),
        "before the sweep, both tiers hold their entries exactly \
         (two pool rows in the schema, three committed files \
         under the root)"
    );

    let first = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(first.aborted, None);
    assert_eq!(
        (
            first.candidates_staged,
            first.deleted,
            first.cancelled_by_arbitration
        ),
        (2, 0, 0),
        "sweep 1 staged exactly the two dead entries and deleted \
         nothing — staging is not executing"
    );
    assert_eq!(
        pg_window_rows(&site).await,
        2,
        "the staged set waits in the durable window"
    );
    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 staging sweep"
        );
    }
    assert!(
        store.has(&key_live.as_bytes()).await.unwrap()
            && store.has(&key_pinned.as_bytes()).await.unwrap()
            && store.has(&key_protected.as_bytes()).await.unwrap(),
        "the staged sweep deleted nothing: every entry intact"
    );

    let second = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(second.aborted, None);
    assert_eq!(
        second.candidates_staged, 0,
        "nothing newly dead since the staging sweep"
    );
    assert_eq!(
        second.deleted + second.cancelled_by_arbitration,
        2,
        "sweep 2 executed exactly the staged set"
    );
    assert_eq!(
        (second.deleted, second.cancelled_by_arbitration),
        (2, 0),
        "with correct liveness, every executed candidate deleted"
    );
    assert!(!store.has(&key_dead_kv.as_bytes()).await.unwrap());
    assert!(!store.has(&key_dead_large.as_bytes()).await.unwrap());
    assert_eq!(
        (pg_pool_rows(&site).await, committed_file_count(site.root())),
        (1, 2),
        "the executed deletes landed on both media exactly"
    );
    assert_eq!(
        pg_window_rows(&site).await,
        0,
        "the executed rows left the durable window"
    );

    drop(pin_pinned);
    let third = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (third.candidates_staged, third.deleted),
        (1, 0),
        "with the pin dropped, the formerly pinned entry stages"
    );
    let fourth = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        fourth.deleted, 1,
        "and the next sweep executes it — exactly one deletion"
    );
    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());
    assert_eq!(
        committed_file_count(site.root()),
        1,
        "exactly the protect-named entry's file remains"
    );

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

/// INV-4, the race, in its durable-schedule form: a put landing
/// *during the window* survives the executing sweep — the re-put's
/// pin is minted (pin-before-publish) between the staging sweep and
/// the executing sweep, so the execution-time re-arbitration cancels
/// the pinned candidate while the unpinned twin deletes; once the pin
/// drops, the twin re-stages and executes exactly once. Both put
/// paths race in-body over both tiers (a kv twin and a large twin).
///
/// Contract stamp: store-api.md §Invariants 4; ADR-005 (pin-before-
/// publish ordering; the delete window closes the sweep-vs-put race);
/// ADR-008 §2 (the staged re-arbitrated delete — the race holds
/// across the window delay); ADR-012 §6.7 (execution-time
/// re-arbitration is the executor sweep's posture).
pub async fn a_put_landing_during_the_window_survives_the_executing_sweep(
    factory: &DualTierPostgresFactory,
) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let rounds: [(String, Vec<u8>, bool, bool); 2] = [
        (
            "the kv twin (known-path re-put)".to_owned(),
            patterned((threshold - 1) as usize, 191),
            true,
            true,
        ),
        (
            "the large twin (unknown-path re-put)".to_owned(),
            patterned(2 * threshold as usize + 5, 173),
            false,
            false,
        ),
    ];
    for (index, (tag, content, original_known, reput_known)) in rounds.iter().enumerate() {
        let len = content.len() as u64;
        let twin = Key::derive(content, len);
        let witness = format!("the dead witness of round {index}").into_bytes();
        let witness_key = Key::derive(&witness, witness.len() as u64);

        let original = if *original_known { Some(len) } else { None };
        let (_, pin) = store
            .put(original, Cursor::new(content.as_slice()))
            .await
            .expect("the twin publishes");
        drop(pin);
        let (_, pin) = store
            .put(None, Cursor::new(witness.as_slice()))
            .await
            .expect("the witness publishes");
        drop(pin);

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            (
                staged.aborted,
                staged.candidates_staged,
                staged.deleted,
                staged.cancelled_by_arbitration
            ),
            (None, 2, 0, 0),
            "{tag}: the staging sweep stages exactly the two dead \
             twins and deletes nothing"
        );

        // The put lands during the window (the re-put dedups into the
        // existing twin and refcounts its pin): protected across the
        // delay, pin-before-publish.
        let reput = if *reput_known { Some(len) } else { None };
        let (_, pin) = store
            .put(reput, Cursor::new(content.as_slice()))
            .await
            .expect("the re-put publishes");
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(executed.aborted, None, "{tag}");
        assert_eq!(
            executed.candidates_staged, 0,
            "{tag}: nothing newly dead since the staging sweep"
        );
        assert_eq!(
            executed.deleted + executed.cancelled_by_arbitration,
            2,
            "{tag}: the executing sweep ran exactly the staged set"
        );
        assert_eq!(executed.deleted, 1, "{tag}: the unpinned witness deleted");
        assert_eq!(
            executed.cancelled_by_arbitration, 1,
            "{tag}: the pinned twin's candidate cancelled at execution time"
        );
        assert!(
            !store.has(&witness_key.as_bytes()).await.unwrap(),
            "{tag}: the unpinned twin is gone"
        );
        let (got_len, blob) = store
            .get(&twin.as_bytes())
            .await
            .expect("get succeeds")
            .expect("{tag}: the put landing during the window survived");
        assert_eq!(got_len, len, "{tag}");
        assert_eq!(
            blob.read_at(0, len).await.expect("the window reads"),
            content.as_slice(),
            "{tag}: the surviving twin is byte-identical"
        );
        drop(pin);

        let restaged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            (restaged.candidates_staged, restaged.deleted),
            (1, 0),
            "{tag}: the pin dropped — the twin re-stages"
        );
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            executed.deleted, 1,
            "{tag}: the re-staged candidate executes exactly once"
        );
        assert!(!store.has(&twin.as_bytes()).await.unwrap());
        assert_eq!(
            pg_window_rows(&site).await,
            0,
            "{tag}: the window holds nothing after the executed round"
        );
    }

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

/// INV-4, abort shape: the sweep is single-flight on the durable form
/// too — the in-process sweep lease (ADR-013 §2's single-node analog).
/// The contended second sweeper aborts with `SweeperLock` in a
/// completed `Ok(SweepReport)` — nothing executed, nothing staged,
/// nothing deleted (abort before staging: form-independent) — while
/// the lease holder's sweep completes by **staging only**: its staged
/// set waits in the window for the next sweep to execute (the
/// durable form's schedule-relative replacement for the reference
/// row's inline holder-completion leg).
///
/// Contract stamp: store-api.md §Invariants 4; ADR-012 §1 (abort is
/// report data, never an error); ADR-013 §2 (the named exclusive
/// sweep lease — one sweeper per pool, holdership ends at release);
/// ADR-012 §6.7 (the executor is the next sweep — the holder stages,
/// it does not execute).
pub async fn the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only(
    factory: &DualTierPostgresFactory,
) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .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: parking_lot::Mutex::new(Some(release_rx)),
    }));

    let content = patterned((threshold - 1) as usize, 167);
    let (key, pin) = store
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the dead witness 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(alkblobs::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 executes, stages, and deletes nothing"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "the contended sweep deleted nothing"
    );
    assert_eq!(
        pg_window_rows(&site).await,
        0,
        "the aborted sweep staged nothing into the window"
    );
    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),
        (1, 0),
        "the holder stages exactly the dead set — it does not execute \
         (the durable form's schedule)"
    );
    assert_eq!(
        pg_window_rows(&site).await,
        1,
        "the holder's staged set waits in the durable window"
    );
    assert!(
        store.has(&key.as_bytes()).await.unwrap(),
        "the holder staged, did not execute: the entry is intact"
    );

    let executed = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        executed.deleted, 1,
        "the next sweep is the executor of the staged set"
    );
    assert!(!store.has(&key.as_bytes()).await.unwrap());

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

/// INV-5 on the durable schedule: `list()` correctness is observable
/// only through GC — the enumeration is complete and well-formed in
/// both directions, proven by exact durable-schedule outcomes: the
/// staging sweep stages exactly the dead set (neither more nor fewer
/// candidates), the executing sweep deletes exactly that staged set,
/// and a liveness flip re-stages exactly the newly dead entry. Never
/// a direct list comparison.
///
/// 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()`); ADR-012 §6.7
/// (the executor is the next sweep — the schedule these counts read
/// against).
pub async fn list_correctness_is_observable_through_durable_schedule_outcomes(
    factory: &DualTierPostgresFactory,
) {
    let (store, site) = factory
        .open_tracked()
        .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 staging, entry three".to_vec(),
        b"dead at the first staging, entry four".to_vec(),
        b"dead at the first staging, 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 first = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(first.aborted, None);
    assert_eq!(
        (first.candidates_staged, first.deleted),
        (3, 0),
        "the enumeration staged exactly the three dead entries — \
         complete and well-formed in both directions"
    );
    for key in &keys {
        assert!(
            store.has(&key.as_bytes()).await.unwrap(),
            "staging deleted nothing: every enumerated entry is intact"
        );
    }
    assert_eq!(
        pg_window_rows(&site).await,
        3,
        "the staged schedule holds the three candidates"
    );

    let second = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (second.candidates_staged, second.deleted),
        (0, 3),
        "the executing sweep removed exactly the staged set"
    );
    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 third = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (third.candidates_staged, third.deleted),
        (1, 0),
        "the newly dead entry is staged exactly"
    );
    let fourth = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(fourth.deleted, 1, "and executed exactly by the next sweep");
    assert!(!store.has(&keys[1].as_bytes()).await.unwrap());
    assert!(store.has(&keys[0].as_bytes()).await.unwrap());

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

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

/// Dispatch row: **routing is a pure function of length** — at, below,
/// and over the declared threshold, the same content lands on exactly
/// the same tier through both put paths (same placement every time;
/// nothing moves, nothing duplicates), identical content through both
/// paths is one pool entry, and every read is byte-identical from
/// wherever that tier holds it.
///
/// Tier placement is observed through the backing medium (the module
/// doc's oracle, adapted): kv entries hold a row in the pg schema and
/// none under the root; large entries hold a committed file under the
/// root and no pool row — the facade cannot and must not expose tier
/// residency (it is engine-blind).
///
/// Contract stamp: store-api.md §Invariants 7 (facade-contract row);
/// ADR-003 §Dispatch policy (routing is a pure function of length;
/// no migration; CAS first-writer-wins makes re-puts one entry);
/// ADR-005 (re-put refcounts); ADR-007 §5 (postgres, the fleet host's
/// kv engine); ADR-012 §6.3 (the dual-tier dispatch-coverage shape);
/// ADR-012 §6.7 (the schedule-relative cleanup the dedup legs read
/// against).
pub async fn routing_is_a_pure_function_of_length_across_put_paths_and_re_puts(
    factory: &DualTierPostgresFactory,
) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    for content in [
        patterned((threshold - 1) as usize, 191),
        patterned(threshold as usize, 173),
        patterned((threshold + 1) as usize, 167),
        patterned(4 * threshold as usize + 9, 151),
    ] {
        let len = content.len() as u64;
        let canonical = Key::derive(&content, len);
        let expected_rows = if len > threshold { 0 } else { 1 };
        let expected_files = if len > threshold { 1 } else { 0 };

        for (path_tag, declared) in [("the known path", Some(len)), ("the unknown path", 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 on {path_tag}"
            );
            assert_eq!(
                pg_pool_rows(&site).await,
                expected_rows,
                "length {len} on {path_tag}: the placement is exactly the \
                 routing function's answer and never moves or duplicates \
                 (the pg medium holds {expected_rows} kv rows)"
            );
            assert_eq!(
                committed_file_count(site.root()),
                expected_files,
                "length {len} on {path_tag}: the large-tier medium holds \
                 {expected_files} committed files"
            );
            assert_eq!(
                staging_file_count(site.root()),
                0,
                "a published put leaves no stage file on {path_tag}"
            );
            let (got_len, blob) = store
                .get(&key.as_bytes())
                .await
                .expect("get succeeds")
                .expect("the entry is present");
            assert_eq!(
                got_len, len,
                "the read serves the entry from its placed tier"
            );
            assert_eq!(
                blob.read_at(0, len).await.expect("the whole window reads"),
                content,
                "the entry is byte-identical from its placed tier across paths"
            );
            drop(pin);
        }

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            staged.candidates_staged, 1,
            "identical content through both put paths is one pool entry"
        );
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            executed.deleted, 1,
            "the executing sweep removed exactly the one entry (durable \
             schedule: staged this sweep, executed by the next)"
        );
        assert_eq!(
            (pg_pool_rows(&site).await, committed_file_count(site.root())),
            (0, 0),
            "both media are back to zero"
        );
    }

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

/// Dispatch row: **threshold boundary lengths route deterministically**
/// — at/below/above the factory's declared threshold, on both put
/// paths, the placement is the routing function's answer every time
/// (the medium oracle distinguishes the schema's rows from the root's
/// committed files); and the default-threshold boundary re-routes
/// with the tuning (the constructor-tuned threshold deterministically
/// places the default boundary's length on the tier its own threshold
/// names). Run against two factories by the runner — the column's
/// tuned threshold and the shipped default — the same row shape at
/// two declared thresholds being the tunable re-route demonstration.
/// The durable cleanup: the staged sweep stages exactly the six
/// entries' dead set and the next sweep executes it.
///
/// Contract stamp: store-api.md §Invariants 7 (facade-contract row);
/// ADR-003 §Dispatch policy (default 128 KiB at-or-below routes kv;
/// constructor-tunable without API change); ADR-007 §5 (postgres, the
/// fleet host's kv engine); ADR-012 §6.3 (the dual-tier
/// dispatch-coverage shape); ADR-012 §6.7 (schedule-relative cleanup).
pub async fn threshold_boundary_lengths_route_deterministically(factory: &DualTierPostgresFactory) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    let boundary = [
        (threshold - 1, 191u32, false),
        (threshold, 173, false),
        (threshold, 167, false),
        (threshold + 1, 151, true),
        (threshold + 1, 139, true),
    ];
    let mut expected_rows = 0u64;
    let mut expected_files = 0u64;
    for (len, modulus, lands_large) in boundary {
        let content = patterned(len as usize, modulus);
        for declared in [Some(content.len() as u64), None] {
            let (key, pin) = store
                .put(declared, Cursor::new(content.as_slice()))
                .await
                .expect("put publishes");
            assert_eq!(
                key,
                Key::derive(&content, len),
                "the canonical derivation addresses the put"
            );
            assert_eq!(
                pg_pool_rows(&site).await as u64,
                expected_rows + if lands_large { 0 } else { 1 },
                "boundary length {len}: at/below the threshold lands kv \
                 (one row in the schema), over it lands large (a \
                 committed file instead), both put paths"
            );
            assert_eq!(
                committed_file_count(site.root()) as u64,
                expected_files + if lands_large { 1 } else { 0 },
                "boundary length {len}: the large-tier medium matches the \
                 routing answer"
            );
            drop(pin);
        }
        if lands_large {
            expected_files += 1;
        } else {
            expected_rows += 1;
        }
    }

    // The default boundary re-routes with the constructor's tuning:
    // the shipped default's length belongs to the tier this
    // configuration's own threshold puts it on.
    let at_default = patterned(DEFAULT_DISPATCH_THRESHOLD as usize, 199);
    let default_lands_large = DEFAULT_DISPATCH_THRESHOLD > threshold;
    let (key, pin) = store
        .put(
            Some(DEFAULT_DISPATCH_THRESHOLD),
            Cursor::new(at_default.as_slice()),
        )
        .await
        .expect("the default-boundary put publishes");
    assert_eq!(
        key,
        Key::derive(&at_default, DEFAULT_DISPATCH_THRESHOLD),
        "the canonical derivation addresses the put"
    );
    assert_eq!(
        pg_pool_rows(&site).await as u64,
        expected_rows + if default_lands_large { 0 } else { 1 },
        "the default boundary's length routes to the tier this \
         configuration's tuning names (the pg medium's answer)"
    );
    assert_eq!(
        committed_file_count(site.root()) as u64,
        expected_files + if default_lands_large { 1 } else { 0 },
        "the default boundary's length routes to the tier this \
         configuration's tuning names (the large medium's answer)"
    );
    drop(pin);

    let staged = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        staged.candidates_staged, 6,
        "each leg's content was exactly one pool entry"
    );
    let executed = store.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        executed.deleted, 6,
        "the executing sweep removed every entry, tier notwithstanding"
    );
    assert_eq!(
        (pg_pool_rows(&site).await, committed_file_count(site.root())),
        (0, 0),
        "both media are back to zero"
    );

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

/// Dispatch row: **a kv-tier miss falls through to the large tier on
/// every key-addressed op** — content the routing function placed on
/// the large tier (one committed file under the root; the pg medium
/// holds no pool row — the kv tier misses) serves `get`, `stat`,
/// `has`, and `read_range` whole, exact, and digest-correct; the
/// mirror-position leg proves the kv tier is the *first* consult
/// (sub-threshold content, one schema row and no committed file,
/// reads still served); and both-tier misses resolve per the error
/// model's per-op shapes (reached here by the durable schedule's
/// staged-then-executed deletion of the subject entry).
///
/// Contract stamp: store-api.md §Read path (fall-through consult
/// order); ADR-012 §6.1 (every key-addressed op consults kv first,
/// then large; the general rule); ADR-012 §6.5 ruling 5 (both-tier
/// miss shapes: `None` on get/stat, `Err(Missing)` on read_range);
/// ADR-006 (the slice digest); ADR-007 §5 (postgres, the fleet host's
/// kv engine); ADR-012 §6.3 (the dispatch-coverage shape).
pub async fn kv_tier_misses_fall_through_to_the_large_tier_per_op(
    factory: &DualTierPostgresFactory,
) {
    let threshold = factory.threshold;
    let (store, site) = factory
        .open_tracked()
        .await
        .expect("open vends a fresh, isolated store");
    SetSource::register(&store);

    // Over-threshold content — the kv tier misses, the large tier
    // serves every op. Both put paths hold the shape.
    for (content, declared) in [
        (
            patterned(3 * threshold as usize + 41, 181),
            Some(3 * threshold + 41),
        ),
        (patterned(3 * threshold as usize + 43, 197), None),
    ] {
        let len = content.len() as u64;
        let (key, pin) = store
            .put(declared, Cursor::new(content.as_slice()))
            .await
            .expect("put publishes");
        assert_eq!(
            (pg_pool_rows(&site).await, committed_file_count(site.root())),
            (0, 1),
            "the content lives on exactly the large tier — the kv tier \
             misses"
        );

        let (got_len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("the large tier served the kv miss");
        assert_eq!(got_len, len);
        assert_eq!(
            blob.read_at(0, len).await.expect("the whole window reads"),
            content,
            "the fall-through read is byte-identical"
        );
        assert_eq!(
            store.stat(&key.as_bytes()).await.expect("stat succeeds"),
            Some(alkblobs::EntryMeta {
                len,
                algorithm: key.algorithm(),
            }),
            "stat falls through and carries the length plus algorithm"
        );
        assert!(
            store.has(&key.as_bytes()).await.unwrap(),
            "has falls through"
        );
        let mid = len / 2;
        let (slice, digest) = store
            .read_range(&key.as_bytes(), mid, 17)
            .await
            .expect("read_range falls through");
        assert_eq!(
            slice,
            content[(mid as usize)..(mid as usize + 17)],
            "the range slice is exact"
        );
        assert_eq!(
            digest,
            slice_digest(&slice),
            "the slice digest is SHA-256 over exactly the returned bytes"
        );

        // Sweep the entry away (the pin dropped): the miss shapes now
        // resolve per op — the large tier misses too. The durable
        // schedule: staged this sweep, executed by the next.
        drop(pin);
        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!((staged.candidates_staged, staged.deleted), (1, 0));
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            executed.deleted, 1,
            "exactly the one entry died across the two sweeps"
        );
        assert!(
            store
                .get(&key.as_bytes())
                .await
                .expect("get succeeds")
                .is_none(),
            "a both-tier miss on get is Ok(None)"
        );
        assert_eq!(
            store.stat(&key.as_bytes()).await.expect("stat succeeds"),
            None,
            "a both-tier miss on stat is Ok(None)"
        );
        assert!(
            !store.has(&key.as_bytes()).await.unwrap(),
            "a both-tier miss on has is false"
        );
        assert_eq!(
            store
                .read_range(&key.as_bytes(), 0, 4)
                .await
                .expect_err("the both-tier miss errs"),
            Error::Missing,
            "a both-tier miss on read_range is Err(Missing)"
        );
    }

    // The mirror-position leg: sub-threshold content — the kv tier is
    // the first consult and serves before the large tier is reached
    // (one row in the schema; no committed file ever lands).
    let small = patterned((threshold - 1) as usize, 149);
    let (key, pin) = store
        .put(Some(small.len() as u64), Cursor::new(small.as_slice()))
        .await
        .expect("the sub-threshold put publishes");
    assert_eq!(
        (pg_pool_rows(&site).await, committed_file_count(site.root())),
        (1, 0),
        "the kv tier holds the sub-threshold entry — the large tier was \
         never consulted on the write, never reached on the read"
    );
    let (len, blob) = store
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the kv tier served");
    assert_eq!(len, small.len() as u64);
    assert_eq!(
        blob.read_at(0, len).await.expect("the whole window reads"),
        small
    );

    drop(pin);

    // Both-tier miss shapes for a key that never existed at all, per
    // op, over the fresh both-tier store.
    let absent = b"never put here".to_vec();
    let absent = Key::derive(&absent, absent.len() as u64);
    assert!(
        store
            .get(&absent.as_bytes())
            .await
            .expect("get succeeds")
            .is_none(),
        "never-put: get is Ok(None)"
    );
    assert_eq!(
        store.stat(&absent.as_bytes()).await.expect("stat succeeds"),
        None,
        "never-put: stat is Ok(None)"
    );
    assert!(
        !store.has(&absent.as_bytes()).await.unwrap(),
        "never-put: has is false"
    );
    assert_eq!(
        store
            .read_range(&absent.as_bytes(), 0, 4)
            .await
            .expect_err("never-put read_range errs"),
        Error::Missing,
        "never-put: read_range is Err(Missing)"
    );

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

/// Virgin-store row: **reads are no-ops on a fresh schema** — an
/// empty `list`, absent reads per op (`None` on get/stat, `false` on
/// has, `Err(Missing)` on read_range), and nothing on the fresh
/// medium itself (zero pool rows, zero window rows, zero committed
/// files). The constructor always ensures the schema (the engine's
/// recorded posture — `ensure_gc_schema` runs at construction, both
/// GC tables in the fresh schema), so the facade's fresh-schema form
/// of the tolerance is this row's assertions.
///
/// Contract stamp: store-api.md §Error model (virgin-store semantics:
/// reads on a fresh store see absent/empty, never errors); ADR-012
/// §6.5 ruling 5 (the per-op miss shapes); ADR-007 §5 (the postgres
/// engine — the schema-ensuring construction posture and the
/// exact-count gate's host).
pub async fn virgin_store_reads_are_no_ops_on_a_fresh_schema(factory: &DualTierPostgresFactory) {
    let (store, site) = factory
        .open_tracked()
        .await
        .expect("open vends a fresh, isolated store");
    assert_eq!(
        store.mode(),
        StoreMode::DualTier,
        "the column's factory is the dispatch-coverage shape"
    );

    let absent = b"nothing was ever put".to_vec();
    let absent = Key::derive(&absent, absent.len() as u64);

    assert_eq!(
        store.list().await.expect("list succeeds"),
        Vec::<Key>::new(),
        "the virgin pool lists empty"
    );
    assert!(
        store
            .get(&absent.as_bytes())
            .await
            .expect("get succeeds")
            .is_none(),
        "absent get is Ok(None)"
    );
    assert_eq!(
        store.stat(&absent.as_bytes()).await.expect("stat succeeds"),
        None,
        "absent stat is Ok(None)"
    );
    assert!(
        !store.has(&absent.as_bytes()).await.unwrap(),
        "absent has is false"
    );
    assert_eq!(
        store
            .read_range(&absent.as_bytes(), 0, 4)
            .await
            .expect_err("absent read_range errs"),
        Error::Missing,
        "absent read_range is Err(Missing)"
    );
    assert_eq!(
        (
            pg_pool_rows(&site).await,
            pg_window_rows(&site).await,
            committed_file_count(site.root())
        ),
        (0, 0, 0),
        "the fresh medium holds nothing (pool, durable window, and \
         committed tree all empty)"
    );

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

// ------------------------------------------------------------------
// The runners — the column's tests: the form-independent reference
// rows mirrored verbatim against the dual-tier factory (INV-1, INV-6,
// and the two no-op abort shapes), the durable-schedule rows above,
// the kv-only rejection row against the kv-only factory (unchanged —
// threshold-agnostic by discovery), and the factory contract's
// structural shapes. Every runner gates behind the reachability
// probe — the SKIP posture: server-less runs skip cleanly, a skip is
// not a failure; acceptance is the green run against the harness
// server.
// ------------------------------------------------------------------

#[tokio::test]
async fn postgres_inv1_round_trip_puts_and_gets_are_byte_identical() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    rows::round_trip_puts_and_gets_are_byte_identical(&factory).await;
}

#[tokio::test]
async fn postgres_inv2_dedup_identical_content_is_one_pool_entry() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    dedup_identical_content_is_one_pool_entry(&factory).await;
}

#[tokio::test]
async fn postgres_inv3_stage_hygiene_zero_residue_one_entry_on_success() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(&factory).await;
}

#[tokio::test]
async fn postgres_inv4_two_sweep_exact_counts_over_the_full_protection_set() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    two_sweep_exact_counts_over_the_full_protection_set(&factory).await;
}

#[tokio::test]
async fn postgres_inv4_a_put_landing_during_the_window_survives_the_executing_sweep() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    a_put_landing_during_the_window_survives_the_executing_sweep(&factory).await;
}

#[tokio::test]
async fn postgres_inv4_sweep_without_liveness_sources_aborts_and_deletes_nothing() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    rows::sweep_without_liveness_sources_aborts_and_deletes_nothing(&factory).await;
}

#[tokio::test]
async fn postgres_inv4_sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    rows::sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing(&factory).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn postgres_inv4_the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    the_contended_second_sweeper_aborts_as_data_and_the_holder_stages_only(&factory).await;
}

#[tokio::test]
async fn postgres_inv5_list_correctness_through_durable_schedule_outcomes() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    list_correctness_is_observable_through_durable_schedule_outcomes(&factory).await;
}

#[tokio::test]
async fn postgres_inv6_stat_reports_len_and_algorithm_without_fetching_content() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    rows::stat_reports_len_and_algorithm_without_fetching_content(&factory).await;
}

#[tokio::test]
async fn postgres_virgin_store_reads_are_no_ops_on_a_fresh_schema() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    virgin_store_reads_are_no_ops_on_a_fresh_schema(&factory).await;
}

#[tokio::test]
async fn postgres_kv_only_mode_rejects_over_threshold_puts_at_put_time() {
    if !harness_ready().await {
        return;
    }
    let factory = KvOnlyPostgresFactory::new(KV_ONLY_THRESHOLD);
    rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&factory).await;
}

#[tokio::test]
async fn dispatch_routing_is_a_pure_function_of_length_across_put_paths_and_re_puts() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    routing_is_a_pure_function_of_length_across_put_paths_and_re_puts(&factory).await;
}

#[tokio::test]
async fn dispatch_threshold_boundary_lengths_route_deterministically() {
    if !harness_ready().await {
        return;
    }
    let tuned = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    threshold_boundary_lengths_route_deterministically(&tuned).await;
    let default = DualTierPostgresFactory::new(DEFAULT_DISPATCH_THRESHOLD);
    threshold_boundary_lengths_route_deterministically(&default).await;
}

#[tokio::test]
async fn dispatch_kv_tier_misses_fall_through_to_the_large_tier_per_op() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);
    kv_tier_misses_fall_through_to_the_large_tier_per_op(&factory).await;
}

// ------------------------------------------------------------------
// The factory contract's structural shapes (the mem and sqlite
// columns' pattern, carried to the first SKIP-posture factory):
// fresh isolated opens (a fresh schema per open, nothing shared — two
// opens never observe one another on the shared server) and teardown
// against a panicking row (the schemas really dropped, idempotent).
// Not conformance rows — no backlog entry — the factory contract's
// own wiring-level assertions.
// ------------------------------------------------------------------

/// Every `open()` mints a fresh, isolated schema: no shared state
/// between islands, observable on the facade *and* on the fresh media
/// (each open's schema and local root are private and start empty).
#[tokio::test]
async fn factory_open_vends_fresh_isolated_postgres_stores() {
    if !harness_ready().await {
        return;
    }
    let factory = DualTierPostgresFactory::new(TUNED_THRESHOLD);

    let (a, site_a) = factory.open_tracked().await.expect("open vends a store");
    let (b, site_b) = factory.open_tracked().await.expect("open vends another");
    assert_ne!(
        site_a.schema(),
        site_b.schema(),
        "two opens never share a schema"
    );
    assert_ne!(
        site_a.root(),
        site_b.root(),
        "two opens never share a local root"
    );

    let content_a = patterned(2 * TUNED_THRESHOLD as usize + 5, 167);
    let content_b = patterned(TUNED_THRESHOLD as usize - 5, 191);
    let (key_a, _) = a
        .put(
            Some(content_a.len() as u64),
            Cursor::new(content_a.as_slice()),
        )
        .await
        .expect("island a's put publishes");
    let (key_b, _) = b
        .put(None, Cursor::new(content_b.as_slice()))
        .await
        .expect("island b's put publishes");

    assert!(a.has(&key_a.as_bytes()).await.unwrap());
    assert!(
        !a.has(&key_b.as_bytes()).await.unwrap(),
        "island a sees nothing of island b"
    );
    assert!(b.has(&key_b.as_bytes()).await.unwrap());
    assert!(
        !b.has(&key_a.as_bytes()).await.unwrap(),
        "island b sees nothing of island a"
    );
    assert_eq!(
        (
            pg_pool_rows(&site_a).await,
            committed_file_count(site_a.root())
        ),
        (0, 1),
        "island a's media hold exactly its own entry (a's content routes large)"
    );
    assert_eq!(
        (
            pg_pool_rows(&site_b).await,
            committed_file_count(site_b.root())
        ),
        (1, 0),
        "island b's media hold exactly its own entry (b's content routes kv)"
    );

    let (c, site_c) = factory.open_tracked().await.expect("a third vends fresh");
    assert_eq!(
        (
            pg_pool_rows(&site_c).await,
            committed_file_count(site_c.root())
        ),
        (0, 0),
        "a third open's media are virgin"
    );
    assert!(
        !c.has(&key_a.as_bytes()).await.unwrap(),
        "a third open shares nothing with the first"
    );
    assert!(
        !c.has(&key_b.as_bytes()).await.unwrap(),
        "a third open shares nothing with the second"
    );

    let schemas = factory.tracked_schemas();
    assert_eq!(schemas.len(), 3, "one minted schema per open");
    assert_eq!(
        schemas.len(),
        schemas
            .iter()
            .collect::<std::collections::BTreeSet<_>>()
            .len(),
        "no two opens share a schema name"
    );

    factory
        .teardown()
        .await
        .expect("teardown drops the schemas and removes the roots");
    factory.teardown().await.expect("teardown repeats cleanly");
    for schema in &schemas {
        assert!(
            !schema_exists(factory.dsn(), schema).await,
            "teardown dropped the schema: {schema}"
        );
    }
}

/// A failing assertion panics the row before teardown runs — and
/// teardown must make sense over the abandoned instance anyway: it is
/// idempotent against the panicked row's path, the minted schemas are
/// really gone server-side (`DROP SCHEMA … CASCADE` over exactly the
/// registry's list; `CASCADE` removes the dependents while the
/// abandoned store's pooled connections fail closed on next use), and
/// the next open mints a fresh schema on a fresh (empty) medium
/// (README §The factory contract).
#[tokio::test]
async fn factory_teardown_is_safe_against_a_panicking_row() {
    if !harness_ready().await {
        return;
    }
    let content = patterned(TUNED_THRESHOLD as usize - 7, 173);
    let abandoned = Key::derive(&content, content.len() as u64);
    let factory = Arc::new(DualTierPostgresFactory::new(TUNED_THRESHOLD));

    let task = {
        let factory = Arc::clone(&factory);
        tokio::spawn(async move {
            let (store, _) = factory.open_tracked().await.expect("open vends the store");
            let (_, pin) = store
                .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
                .await
                .expect("the put publishes");
            drop(pin);
            panic!("a failing assertion panics the row (the factory contract's early exit)");
        })
    };
    assert!(
        task.await.is_err(),
        "the row's panic surfaced as its failure"
    );

    let schemas = factory.tracked_schemas();
    assert!(!schemas.is_empty(), "the panicked row minted a schema");

    factory
        .teardown()
        .await
        .expect("teardown makes sense over the abandoned instance");
    factory.teardown().await.expect("teardown is idempotent");
    for schema in &schemas {
        assert!(
            !schema_exists(factory.dsn(), schema).await,
            "teardown dropped the panicked row's schema: {schema}"
        );
    }

    let (store, site) = factory
        .open_tracked()
        .await
        .expect("the next open still vends a store");
    assert!(
        !store.has(&abandoned.as_bytes()).await.unwrap(),
        "no residue from the panicked row — isolation is factory-owned"
    );
    assert_eq!(
        (pg_pool_rows(&site).await, committed_file_count(site.root())),
        (0, 0),
        "and its media carry nothing of the abandoned run"
    );
    factory
        .teardown()
        .await
        .expect("teardown drops the last schema");
}