macrame-db 0.15.0

//! Property tests for the Doctrine VI invariant.
//!
//! `links_current` is derivative: it must always equal the latest-belief
//! projection of `links`. `audit_current` is the check that says so, and
//! `rebuild_current` is the repair. Both were previously covered only by
//! seeded unit tests — a handful of rows and three hand-chosen corruptions.
//!
//! That is not enough, and the pre-0.5.4 audit is the proof. It parsed as
//! `A EXCEPT A`, a constant zero, and it passed every seeded test *because a
//! constant zero is the right answer on clean data*. A check whose failure mode
//! is "always says fine" cannot be validated by examples that are fine.
//!
//! So these tests do not seed. They generate arbitrary histories, compute the
//! projection **independently in Rust**, and require the SQL and the model to
//! agree on the exact drift count. A degenerate query cannot survive that,
//! because the model disagrees the moment anything is actually wrong.
//!
//! # On the case count (R15)
//!
//! 32, not proptest's default 256, and the number is a workaround rather than a
//! judgement about coverage. libSQL faults intermittently with
//! STATUS_ACCESS_VIOLATION when local databases are opened concurrently in one
//! process — reproducible with no Macrame types, no proptest, just
//! open/migrate/drop in a loop, which puts it below the Doctrine I line. It is
//! not a stale-dependency problem: moving 0.6.0 → 0.9.30 left the rate
//! unchanged.
//!
//! Most of the suite is fixed by `RUST_TEST_THREADS = "1"` (.cargo/config.toml,
//! 0/30 bad runs). This binary is the residue that serialising does not reach,
//! because a property case needs a database of its own, so the only lever left
//! here is how many cases run.
//!
//! This does not weaken what the suite proves. The generator domains are tiny
//! on purpose, so 32 cases still saturate the interesting shapes, and any
//! failure proptest has ever found is replayed from `.proptest-regressions`
//! before a single new case is generated — the archive defect these tests
//! caught stays pinned no matter what this number is. Raise it deliberately
//! (`PROPTEST_CASES=512`) when changing the audit or the archive predicates,
//! and expect the occasional crash to be the engine, not the change.
//!
//! For the same reason **this binary is gated behind the `property-tests`
//! feature** and does not run under a plain `cargo test`. The gate is not a
//! demotion — these are the tests that found D-035 — it is an admission that a
//! suite failing for reasons unrelated to the code under test teaches
//! developers to ignore red. Run them with
//! `cargo test --features property-tests`, as their own step, so a genuine
//! failure is still a genuine failure.

#[path = "common/harness.rs"]
mod harness;

/// When the archive session ran, as distinct from the cutoff it used. Any
/// canonical stamp does; the point is that it is not the cutoff (Wave 4.5).
const ARCHIVED_AT: &str = "2026-07-30T12:00:00.000000Z";

use std::collections::BTreeSet;
use std::future::Future;

use harness::TestHarness;
use macrame::error::DbError;
use macrame::integrity::{audit_current, rebuild_current};
use proptest::prelude::*;

const SENTINEL: &str = "9999-12-31T23:59:59.999999Z";

/// Canonical (D-029) instants, in chronological *and* lexicographic order.
const TS: [&str; 5] = [
    "2026-01-01T00:00:00.000000Z",
    "2026-02-01T00:00:00.000000Z",
    "2026-03-01T00:00:00.000000Z",
    "2026-04-01T00:00:00.000000Z",
    "2026-05-01T00:00:00.000000Z",
];

/// Deliberately tiny domains. Bugs in a set-difference live at collisions —
/// same key, different payload; same partition, different `recorded_at` — and a
/// generator with room to spread out never produces one.
const NODES: [&str; 2] = ["c0", "c1"];
const TYPES: [&str; 2] = ["A", "B"];
/// Exactly representable, so the Rust model and SQLite's `EXCEPT` cannot
/// disagree over float rounding rather than over drift.
const WEIGHTS: [f64; 3] = [0.5, 1.0, 2.0];

/// One runtime for the whole binary. Building one per case is the obvious
/// shape and is one of the two things that provoke R15 — see the header note
/// on case counts.
static RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();

fn block_on<F: Future>(f: F) -> F::Output {
    RT.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap()
    })
    .block_on(f)
}

// ---------------------------------------------------------------------------
// The model
// ---------------------------------------------------------------------------

/// A row of `links` / `links_current`, comparable the way `EXCEPT` compares.
///
/// `weight` is carried as raw bits so the row can be `Ord` for a `BTreeSet`.
/// Both sides come from the same generated constants, so bit equality and
/// SQLite's numeric equality coincide.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
struct Row {
    source_id: String,
    target_id: String,
    edge_type: String,
    valid_from: String,
    valid_to: String,
    weight_bits: u64,
    properties: String,
    recorded_at: String,
}

impl Row {
    /// The projection's partition key (§5.8): one row of current belief per
    /// interval, *not* per edge.
    fn key(&self) -> (&str, &str, &str, &str) {
        (
            &self.source_id,
            &self.target_id,
            &self.edge_type,
            &self.valid_from,
        )
    }
}

/// The latest-belief projection of `links`, computed in Rust.
///
/// This is the oracle. It is deliberately written as a fold rather than as
/// anything resembling the SQL, so that a mistake in the window function has
/// nowhere to hide: the two implementations share no code and no reasoning.
///
/// Well-defined without a tie-break because `links` has
/// `PRIMARY KEY (source_id, target_id, edge_type, valid_from, recorded_at)` —
/// two rows in one partition cannot share a `recorded_at`.
fn projection(links: &[Row]) -> BTreeSet<Row> {
    let mut best: Vec<Row> = Vec::new();
    for row in links {
        match best.iter_mut().find(|b| b.key() == row.key()) {
            Some(b) if row.recorded_at > b.recorded_at => *b = row.clone(),
            Some(_) => {}
            None => best.push(row.clone()),
        }
    }
    best.into_iter().collect()
}

/// Symmetric difference cardinality — what `audit_current` claims to return.
fn expected_drift(links: &[Row], current: &[Row]) -> usize {
    let projected = projection(links);
    let materialized: BTreeSet<Row> = current.iter().cloned().collect();
    materialized.difference(&projected).count() + projected.difference(&materialized).count()
}

async fn read_rows(conn: &libsql::Connection, table: &str) -> Vec<Row> {
    let mut rows = conn
        .query(
            &format!(
                "SELECT source_id, target_id, edge_type, valid_from, valid_to, \
                 weight, properties, recorded_at FROM {table}"
            ),
            (),
        )
        .await
        .unwrap();
    let mut out = Vec::new();
    while let Some(r) = rows.next().await.unwrap() {
        let weight: f64 = r.get(5).unwrap();
        out.push(Row {
            source_id: r.get(0).unwrap(),
            target_id: r.get(1).unwrap(),
            edge_type: r.get(2).unwrap(),
            valid_from: r.get(3).unwrap(),
            valid_to: r.get(4).unwrap(),
            weight_bits: weight.to_bits(),
            properties: r.get(6).unwrap(),
            recorded_at: r.get(7).unwrap(),
        });
    }
    out
}

/// `audit_current`'s answer as a plain number, so a property can compare it to
/// the model without caring which arm of the `Result` carried it.
async fn audit_count(conn: &libsql::Connection) -> usize {
    match audit_current(conn).await {
        Ok(n) => n,
        Err(DbError::CurrentDrift { n }) => n,
        Err(e) => panic!("audit failed for a reason other than drift: {e:?}"),
    }
}

// ---------------------------------------------------------------------------
// Generated operations
// ---------------------------------------------------------------------------

/// An assertion into `links`. Indices, not strings: the shrinker can then walk
/// a failing case down to the smallest indices rather than to arbitrary text.
#[derive(Debug, Clone, Copy)]
struct Assert {
    src: usize,
    tgt: usize,
    etype: usize,
    valid_from: usize,
    /// `None` is the open sentinel.
    valid_to: Option<usize>,
    weight: usize,
    recorded_at: usize,
}

fn assert_strategy() -> impl Strategy<Value = Assert> {
    (
        0..NODES.len(),
        0..NODES.len(),
        0..TYPES.len(),
        0..3usize,                   // valid_from  ∈ TS[0..3]
        prop::option::of(2..5usize), // valid_to ∈ TS[2..5] or open
        0..WEIGHTS.len(),
        0..TS.len(),
    )
        .prop_map(
            |(src, tgt, etype, valid_from, valid_to, weight, recorded_at)| Assert {
                src,
                tgt,
                etype,
                valid_from,
                valid_to,
                weight,
                recorded_at,
            },
        )
}

/// The concepts C1's properties seed. Four ids doing three different jobs, and
/// **every one of the three was forced by an injection the property failed to
/// catch without it** — the shape of this constant is a record of that.
///
/// * `c0`, `c1` — `NODES`. Sources and targets, as everywhere else in this file.
/// * `c2` — a **target-only** concept: [`archive_assert_strategy`] may point an
///   edge at it but never from it. Without this, "appears as a source" and
///   "appears at all" coincide over enough cases, and narrowing
///   `CONCEPTS_ARCHIVABLE`'s reachability check to `source_id` alone changes no
///   answer. With it, that injection fails.
/// * `c3` — **never referenced by any generated edge**, so it is archivable
///   whenever its own three clauses say so, on every case rather than by luck.
///
/// The first attempt seeded only `NODES`, and a history of up to twelve edges
/// over two nodes references both in nearly every case: `NOT EXISTS` was false
/// for both, the archivable set was empty on both sides, and three of five
/// injections passed. Adding `c2` fixed two of them. Dropping `retired = 1`
/// still survived, because `c2` is unreferenced only when no edge happens to
/// pick it — around one case in ten — which 32 cases cannot rely on. `c3`
/// removes the luck: the row-level clauses are under test in every case.
///
/// **The general shape is worth more than the fix.** A property test over a
/// conjunction is only as strong as its ability to reach each clause
/// *independently*, and a generator drawing every id from one domain
/// systematically prevents that. The predicate was correct throughout; what was
/// wrong was the evidence for it.
const ARCHIVE_NODES: [&str; 4] = ["c0", "c1", "c2", "c3"];

/// Index into [`ARCHIVE_NODES`] of the first concept no generated edge may name.
const UNREFERENCED: usize = 3;

/// A concept as generated (C1). `retired` and the two clocks are the three
/// things `CONCEPTS_ARCHIVABLE` reads off the row itself; the fourth clause —
/// reachability — is a function of `links`, and so is decided by `history`.
#[derive(Debug, Clone, Copy)]
struct ConceptSpec {
    retired: bool,
    /// `None` is the open sentinel.
    valid_to: Option<usize>,
    recorded_at: usize,
}

/// Edges for C1's properties, over three of the four [`ARCHIVE_NODES`]:
/// `c0`/`c1` may be either endpoint, `c2` may only be a **target**, and `c3` may
/// be neither. The reasoning for each is on `ARCHIVE_NODES`.
fn archive_assert_strategy() -> impl Strategy<Value = Assert> {
    (
        0..NODES.len(),  // sources: c0, c1
        0..UNREFERENCED, // targets: c0, c1, c2 — never c3
        0..TYPES.len(),
        0..3usize,
        prop::option::of(2..5usize),
        0..WEIGHTS.len(),
        0..TS.len(),
    )
        .prop_map(
            |(src, tgt, etype, valid_from, valid_to, weight, recorded_at)| Assert {
                src,
                tgt,
                etype,
                valid_from,
                valid_to,
                weight,
                recorded_at,
            },
        )
}

/// `valid_to` ranges over the whole of `TS` rather than the closed tail, so the
/// generator produces concepts whose validity closed *after* the cutoff as well
/// as before it. A predicate that ignored `valid_to` would still pass a
/// generator that only ever proposed archivable rows.
fn concept_strategy() -> impl Strategy<Value = ConceptSpec> {
    (any::<bool>(), prop::option::of(0..TS.len()), 0..TS.len()).prop_map(
        |(retired, valid_to, recorded_at)| ConceptSpec {
            retired,
            valid_to,
            recorded_at,
        },
    )
}

/// Deliberate damage to the derivative table. `audit_current` exists to notice
/// these; the property is that it notices *exactly* them.
#[derive(Debug, Clone, Copy)]
enum Corruption {
    /// Drop the nth row — a missed materialisation.
    Drop(usize),
    /// Rewrite the nth row's weight — right key, wrong payload, which is drift
    /// in *both* directions at once.
    Stale(usize),
    /// Insert a row the projection does not have — spurious materialisation.
    Ghost(usize),
}

fn corruption_strategy() -> impl Strategy<Value = Corruption> {
    prop_oneof![
        (0..8usize).prop_map(Corruption::Drop),
        (0..8usize).prop_map(Corruption::Stale),
        (0..3usize).prop_map(Corruption::Ghost),
    ]
}

/// Open a migrated database with both concepts present (`links` has a FK into
/// `concepts` and `PRAGMA foreign_keys` is on).
///
/// Returns the `libsql::Database` alongside the `Connection` because it **must**
/// outlive it. Returning the connection alone drops the Database at the end of
/// this function and leaves the caller holding a connection into freed state:
/// an intermittent `STATUS_ACCESS_VIOLATION` whose rate scales with how many
/// allocations follow, which is why it showed up under proptest and never in a
/// unit test. Rust cannot catch it — the relationship is inside the FFI, not in
/// the borrow checker — so the shape of the helper is the only guard.
async fn fresh(harness: &TestHarness) -> (libsql::Database, libsql::Connection) {
    let db = libsql::Builder::new_local(&harness.db_path)
        .build()
        .await
        .unwrap();
    let conn = db.connect().unwrap();
    macrame::schema::run_migrations(&conn).await.unwrap();
    for id in NODES {
        conn.execute(
            "INSERT INTO concepts (id, title, valid_from, recorded_at) VALUES (?1, 'N', ?2, ?2)",
            libsql::params![id, TS[0]],
        )
        .await
        .unwrap();
    }
    (db, conn)
}

/// [`fresh`], but with the two concepts carrying generated `retired` and clock
/// values instead of the defaults (C1).
///
/// Separate from `fresh` rather than a parameter on it, because every property
/// above this one depends on the concepts being unremarkable — they exist only
/// to satisfy the `links` foreign key — and threading a generator through them
/// would widen their search space for no gain.
async fn fresh_with_concepts(
    harness: &TestHarness,
    specs: &[ConceptSpec; ARCHIVE_NODES.len()],
) -> (libsql::Database, libsql::Connection) {
    let db = libsql::Builder::new_local(&harness.db_path)
        .build()
        .await
        .unwrap();
    let conn = db.connect().unwrap();
    macrame::schema::run_migrations(&conn).await.unwrap();
    for (id, spec) in ARCHIVE_NODES.iter().zip(specs) {
        conn.execute(
            "INSERT INTO concepts (id, title, valid_from, valid_to, recorded_at, retired) \
             VALUES (?1, 'N', ?2, ?3, ?4, ?5)",
            libsql::params![
                *id,
                TS[0],
                spec.valid_to.map_or(SENTINEL, |i| TS[i]),
                TS[spec.recorded_at],
                i64::from(spec.retired),
            ],
        )
        .await
        .unwrap();
    }
    (db, conn)
}

/// The archivability predicate, computed in Rust from the state the database
/// actually holds.
///
/// This is the oracle, and it is written as four independent tests against the
/// row rather than as anything resembling the SQL — the same discipline
/// [`projection`] follows, for the same reason: two implementations that share
/// reasoning cannot disagree, and a predicate that cannot disagree with its
/// oracle is not being checked by it.
///
/// `links` is passed in as it stands in the hot table, not as the generated
/// history, because the generator is free to propose inserts the schema
/// refuses — reachability is a claim about the rows that exist.
fn archivable_model(
    specs: &[ConceptSpec; ARCHIVE_NODES.len()],
    links: &[Row],
    cutoff: &str,
) -> BTreeSet<String> {
    let mut out = BTreeSet::new();
    for (id, spec) in ARCHIVE_NODES.iter().zip(specs) {
        let valid_to = spec.valid_to.map_or(SENTINEL, |i| TS[i]);
        let referenced = links
            .iter()
            .any(|l| l.source_id == *id || l.target_id == *id);
        if spec.retired && TS[spec.recorded_at] < cutoff && valid_to < cutoff && !referenced {
            out.insert((*id).to_string());
        }
    }
    out
}

/// Apply a generated history. Insert failures are *expected and skipped*: the
/// generator is free to propose a second open interval (blocked by
/// `trg_links_single_open`) or a duplicate primary key. Those are the schema
/// working, not the property failing — what matters is the state that results.
async fn apply(conn: &libsql::Connection, history: &[Assert]) {
    for a in history {
        let _ = conn
            .execute(
                "INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, \
                 weight, properties, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, '{}', ?7)",
                libsql::params![
                    NODES[a.src],
                    NODES[a.tgt],
                    TYPES[a.etype],
                    TS[a.valid_from],
                    a.valid_to.map_or(SENTINEL, |i| TS[i]),
                    WEIGHTS[a.weight],
                    TS[a.recorded_at],
                ],
            )
            .await;
    }
}

/// [`apply`], resolving endpoint indices against [`ARCHIVE_NODES`] rather than
/// `NODES`, so a generated edge can name the third concept.
///
/// `ARCHIVE_NODES` starts with `NODES`, so index `0` and `1` mean the same
/// concept in both — only index `2` is new, and only [`archive_assert_strategy`]
/// ever produces it.
async fn apply_over_archive_nodes(conn: &libsql::Connection, history: &[Assert]) {
    for a in history {
        let _ = conn
            .execute(
                "INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, \
                 weight, properties, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, '{}', ?7)",
                libsql::params![
                    ARCHIVE_NODES[a.src],
                    ARCHIVE_NODES[a.tgt],
                    TYPES[a.etype],
                    TS[a.valid_from],
                    a.valid_to.map_or(SENTINEL, |i| TS[i]),
                    WEIGHTS[a.weight],
                    TS[a.recorded_at],
                ],
            )
            .await;
    }
}

async fn corrupt(conn: &libsql::Connection, damage: &[Corruption]) {
    for c in damage {
        match *c {
            Corruption::Drop(n) | Corruption::Stale(n) => {
                let current = read_rows(conn, "links_current").await;
                if current.is_empty() {
                    continue;
                }
                let victim = &current[n % current.len()];
                let sql = match c {
                    Corruption::Drop(_) => {
                        "DELETE FROM links_current WHERE source_id = ?1 AND target_id = ?2 \
                         AND edge_type = ?3 AND valid_from = ?4"
                    }
                    _ => {
                        "UPDATE links_current SET weight = 99.0 WHERE source_id = ?1 \
                         AND target_id = ?2 AND edge_type = ?3 AND valid_from = ?4"
                    }
                };
                conn.execute(
                    sql,
                    libsql::params![
                        victim.source_id.clone(),
                        victim.target_id.clone(),
                        victim.edge_type.clone(),
                        victim.valid_from.clone()
                    ],
                )
                .await
                .unwrap();
            }
            Corruption::Ghost(n) => {
                // May collide with a real row's primary key; that is a no-op,
                // not damage, and the model sees the same thing either way.
                let _ = conn
                    .execute(
                        "INSERT OR IGNORE INTO links_current (source_id, target_id, edge_type, \
                         valid_from, valid_to, weight, properties, recorded_at) \
                         VALUES ('c0', 'c1', 'GHOST', ?1, ?2, 7.0, '{}', ?2)",
                        libsql::params![TS[n % 3], TS[0]],
                    )
                    .await;
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Properties
// ---------------------------------------------------------------------------

// A fresh SQLite file per case, so the case count is a runtime budget rather
// than a coverage target. 96 cases over domains this small saturates the
// interesting shapes; the shrinker does the rest.
proptest! {
    #![proptest_config(ProptestConfig { cases: 32, ..ProptestConfig::default() })]

    /// **The trigger maintains the invariant.** For any history the schema
    /// accepts, `links_current` is already the projection — no rebuild needed.
    ///
    /// This is the property the old seeded tests thought they were checking.
    /// It is also the one a constant-zero audit passes trivially, which is why
    /// it is worthless on its own and why the next property exists.
    #[test]
    fn the_sync_trigger_keeps_current_equal_to_the_projection(
        history in prop::collection::vec(assert_strategy(), 0..12)
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh(&harness).await;
            apply(&conn, &history).await;

            let links = read_rows(&conn, "links").await;
            let current = read_rows(&conn, "links_current").await;
            prop_assert_eq!(
                expected_drift(&links, &current), 0,
                "the AFTER INSERT trigger did not maintain Doctrine VI"
            );
            prop_assert_eq!(audit_count(&conn).await, 0);
            Ok(())
        })?;
    }

    /// **The audit is exact.** Not "detects corruption" — *equals the model*.
    ///
    /// An audit that over-reports is as broken as one that under-reports: it
    /// makes `rebuild_current` fail its own post-check and turns a healthy
    /// database into an unopenable one. Comparing counts, not just
    /// `is_err()`, is what makes this a check on the query rather than on the
    /// existence of a query.
    #[test]
    fn the_audit_count_equals_the_model(
        history in prop::collection::vec(assert_strategy(), 0..12),
        damage in prop::collection::vec(corruption_strategy(), 0..5),
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh(&harness).await;
            apply(&conn, &history).await;
            corrupt(&conn, &damage).await;

            let links = read_rows(&conn, "links").await;
            let current = read_rows(&conn, "links_current").await;
            prop_assert_eq!(
                audit_count(&conn).await,
                expected_drift(&links, &current),
                "audit disagreed with the model\nlinks: {:#?}\ncurrent: {:#?}",
                links, current
            );
            Ok(())
        })?;
    }

    /// **Rebuild is a fixpoint, and reaches it from anywhere.**
    ///
    /// Three claims in one: rebuilding produces exactly the projection, the
    /// resulting drift is zero, and rebuilding again changes nothing. The third
    /// is what catches a repair that is only correct on the state it was
    /// written against.
    #[test]
    fn rebuild_restores_the_projection_from_any_damage(
        history in prop::collection::vec(assert_strategy(), 0..12),
        damage in prop::collection::vec(corruption_strategy(), 0..5),
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh(&harness).await;
            apply(&conn, &history).await;
            corrupt(&conn, &damage).await;

            let links = read_rows(&conn, "links").await;
            let report = rebuild_current(&conn).await.unwrap();

            let rebuilt = read_rows(&conn, "links_current").await;
            let expected = projection(&links);
            prop_assert_eq!(rebuilt.iter().cloned().collect::<BTreeSet<_>>(), expected.clone());
            prop_assert_eq!(report.rows_rebuilt, expected.len());
            prop_assert_eq!(report.drift_after, 0);
            prop_assert_eq!(audit_count(&conn).await, 0);

            // Idempotence: the second pass must be a no-op, not a re-derivation
            // that happens to land somewhere else.
            let second = rebuild_current(&conn).await.unwrap();
            prop_assert_eq!(second, report);
            prop_assert_eq!(read_rows(&conn, "links_current").await, rebuilt);
            Ok(())
        })?;
    }

    /// **Archiving preserves the invariant.**
    ///
    /// `archive()` deletes from `links` under `LINKS_ARCHIVABLE` and separately
    /// deletes from `links_current` to compensate. Those are two different
    /// predicates over two different clocks, and nothing else in the suite
    /// exercises them jointly. Doctrine VI does not pause for an archive: if
    /// the compensation is not the exact image of the deletion, the ledger is
    /// left permanently unauditable.
    #[test]
    fn archiving_leaves_the_ledger_auditable(
        history in prop::collection::vec(assert_strategy(), 0..12),
        cutoff in 0..TS.len(),
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh(&harness).await;
            apply(&conn, &history).await;
            prop_assume!(audit_count(&conn).await == 0);

            let cold = harness.temp_dir.path().join("cold.db");
            macrame::temporal::archive(&conn, TS[cutoff], ARCHIVED_AT, &cold).await.unwrap();

            let links = read_rows(&conn, "links").await;
            let current = read_rows(&conn, "links_current").await;
            prop_assert_eq!(
                audit_count(&conn).await, 0,
                "archive left drift behind\ncutoff: {}\nlinks: {:#?}\ncurrent: {:#?}",
                TS[cutoff], links, current
            );
            Ok(())
        })?;
    }

    /// **The archivability predicate agrees with its model, exactly** (C1,
    /// D-128).
    ///
    /// `CONCEPTS_ARCHIVABLE` is four conjoined clauses, three read off the
    /// concept row and one a `NOT EXISTS` over `links` in both directions. Every
    /// one of them is a way to be wrong in a direction no seeded test would
    /// notice: dropping `retired` archives live concepts, dropping either clock
    /// archives too early, and — the one that matters — getting the `NOT EXISTS`
    /// backwards or checking only `source_id` produces a predicate that says
    /// *yes* for a concept a surviving hot link still points at. That is an
    /// unsatisfiable foreign key at the moment C2 acts on it, and it is exactly
    /// the shape a fixture built from an intuition about archival would miss.
    ///
    /// Set equality in both directions, not a count: a predicate that admits one
    /// wrong concept and rejects one right one has the correct cardinality.
    #[test]
    fn the_archivability_predicate_matches_its_model(
        history in prop::collection::vec(archive_assert_strategy(), 0..12),
        specs in prop::array::uniform4(concept_strategy()),
        cutoff in 0..TS.len(),
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh_with_concepts(&harness, &specs).await;
            apply_over_archive_nodes(&conn, &history).await;

            let links = read_rows(&conn, "links").await;
            let expected = archivable_model(&specs, &links, TS[cutoff]);
            let actual: BTreeSet<String> = macrame::temporal::archivable_concepts(&conn, TS[cutoff])
                .await
                .unwrap()
                .into_iter()
                .collect();

            prop_assert_eq!(
                &actual, &expected,
                "predicate disagreed with the model\ncutoff: {}\nspecs: {:#?}\nlinks: {:#?}",
                TS[cutoff], specs, links
            );
            Ok(())
        })?;
    }

    /// **Archiving links can only enlarge the archivable set, never shrink it**
    /// (C1, D-128).
    ///
    /// This is the downstream relationship stated as something executable.
    /// `archive()` removes rows from hot `links` and never adds any, so the
    /// `NOT EXISTS` clause can only go from false to true — which is the whole
    /// argument for evaluating the predicate *after* the link delete inside a
    /// session rather than before it.
    ///
    /// It is worth a property rather than a comment because the monotonicity is
    /// a joint claim about two predicates written independently: if
    /// `LINKS_ARCHIVABLE` ever deleted a row it should not, or `archive()`'s
    /// `links_current` re-derivation ever wrote back into `links`, this is what
    /// would say so.
    ///
    /// **Restated for C2, which is what this docstring said would happen and
    /// then went unnoticed for three items.** The paragraph below used to end
    /// *"the stronger half — every archived concept is unreferenced — cannot be
    /// asserted until C2 archives one"*, and once C2 did, the property as
    /// written became **false**: `archive()` moves archivable concepts out of
    /// the hot table, so a concept that was archivable before a session is
    /// legitimately absent from `archivable_concepts` after it. It was absent
    /// because it had gone cold, which is success, and the test called it a
    /// withdrawal.
    ///
    /// The claim that survives is the one the union makes: **a concept leaves
    /// the archivable set only by being archived.** `before ⊆ after ∪ cold`
    /// keeps the original monotonicity — link archival cannot withdraw anything
    /// — while admitting the one legitimate way out. And the stronger half is
    /// now assertable and asserted: nothing in `cold.concepts` is referenced by
    /// a surviving hot link.
    ///
    /// **How it went unnoticed is worth more than the fix.** This test is behind
    /// the `property-tests` feature, and C2, C3 and C4 each ran their gate on
    /// the default feature set — 27 targets, green every time, with this one not
    /// among them. A feature-gated target is not covered by "the suite passes"
    /// unless the suite is run with the feature, and three consecutive items
    /// reported green against a gate that was not running.
    #[test]
    fn a_concept_leaves_the_archivable_set_only_by_being_archived(
        history in prop::collection::vec(archive_assert_strategy(), 0..12),
        specs in prop::array::uniform4(concept_strategy()),
        cutoff in 0..TS.len(),
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh_with_concepts(&harness, &specs).await;
            apply_over_archive_nodes(&conn, &history).await;
            prop_assume!(audit_count(&conn).await == 0);

            let before: BTreeSet<String> =
                macrame::temporal::archivable_concepts(&conn, TS[cutoff]).await.unwrap()
                    .into_iter().collect();

            let cold = harness.temp_dir.path().join("cold.db");
            macrame::temporal::archive(&conn, TS[cutoff], ARCHIVED_AT, &cold).await.unwrap();

            let after: BTreeSet<String> =
                macrame::temporal::archivable_concepts(&conn, TS[cutoff]).await.unwrap()
                    .into_iter().collect();

            // What went cold in this session. Read by ATTACH rather than by
            // opening the file a second time: a second open of a second database
            // in one process is R15's exact shape, and a property test runs it
            // hundreds of times.
            conn.execute(
                "ATTACH DATABASE ?1 AS cold",
                libsql::params![cold.to_string_lossy().as_ref()],
            ).await.unwrap();
            let archived: BTreeSet<String> = {
                let mut rows = conn.query("SELECT id FROM cold.concepts", ()).await.unwrap();
                let mut out = BTreeSet::new();
                while let Some(r) = rows.next().await.unwrap() {
                    out.insert(r.get::<String>(0).unwrap());
                }
                out
            };
            conn.execute("DETACH DATABASE cold", ()).await.unwrap();

            let accounted: BTreeSet<String> = after.union(&archived).cloned().collect();
            prop_assert!(
                before.is_subset(&accounted),
                "a concept left the archivable set without being archived\n\
                 cutoff: {}\nbefore: {:?}\nafter: {:?}\narchived: {:?}",
                TS[cutoff], before, after, archived
            );

            // The stronger half, assertable now that C2 archives concepts:
            // nothing that went cold — and nothing the predicate still admits —
            // is referenced by a surviving hot link.
            let surviving = read_rows(&conn, "links").await;
            for id in accounted.iter() {
                prop_assert!(
                    !surviving.iter().any(|l| l.source_id == *id || l.target_id == *id),
                    "{} is archivable or archived while a hot link still references it\nlinks: {:#?}",
                    id, surviving
                );
            }
            Ok(())
        })?;
    }
}

// ---------------------------------------------------------------------------
// T0.1 — the two traversal CTE forms must agree
// ---------------------------------------------------------------------------

/// The form shipped through 0.5.6: `UNION ALL`, a `path` column of visited ids,
/// and an `INSTR` cycle check restricting the walk to **simple paths**.
///
/// Kept here as the oracle, and nowhere else in the tree. It is the definition
/// T0.1's rewrite claims to preserve, so preserving it has to be checkable after
/// the implementation stops containing it.
const OLD_WALK: &str = r#"
WITH RECURSIVE walk(node_id, depth, path) AS (
    SELECT ?1, 0, '/' || CAST(?1 AS BLOB) || '/'
    UNION ALL
    SELECT l.target_id, w.depth + 1, w.path || CAST(l.target_id AS BLOB) || '/'
    FROM walk w
    JOIN links_current l ON l.source_id = w.node_id
    WHERE w.depth < ?2
      AND l.valid_from <= ?3 AND ?3 < l.valid_to
      AND l.weight >= ?4
      AND INSTR(w.path, '/' || CAST(l.target_id AS BLOB) || '/') = 0
)"#;

/// The 0.6.0 form: `UNION` dedupes `(node_id, depth)` on entry; no path column.
const NEW_WALK: &str = r#"
WITH RECURSIVE walk(node_id, depth) AS (
    SELECT ?1, 0
    UNION
    SELECT l.target_id, w.depth + 1
    FROM walk w
    JOIN links_current l ON l.source_id = w.node_id
    WHERE w.depth < ?2
      AND l.valid_from <= ?3 AND ?3 < l.valid_to
      AND l.weight >= ?4
)"#;

const IDS_PROJECTION: &str = r#"
SELECT DISTINCT w.node_id
FROM walk w JOIN concepts c ON c.id = w.node_id
WHERE c.retired = 0
ORDER BY w.node_id;"#;

const EDGES_PROJECTION: &str = r#"
SELECT DISTINCT l.source_id, l.target_id, l.edge_type
FROM walk w
JOIN links_current l ON l.source_id = w.node_id
WHERE l.valid_from <= ?3 AND ?3 < l.valid_to
  AND l.weight >= ?4
ORDER BY l.source_id, l.target_id, l.edge_type;"#;

/// Five nodes, because the shapes that separate the two forms — a diamond, a
/// cycle, two routes of different length to one node — do not exist in two.
const TNODES: [&str; 5] = ["t0", "t1", "t2", "t3", "t4"];

/// An edge as generated: endpoints, type, and whether it is still live at the
/// instant queried. Expired edges are generated because the temporal predicate
/// sits inside the recursion, so a rewrite could preserve reachability and quietly
/// change *when* an edge counts.
#[derive(Debug, Clone)]
struct TEdge {
    src: usize,
    tgt: usize,
    etype: usize,
    expired: bool,
}

fn tedge_strategy() -> impl Strategy<Value = TEdge> {
    (
        0..TNODES.len(),
        0..TNODES.len(),
        0..TYPES.len(),
        prop::bool::weighted(0.2),
    )
        .prop_map(|(src, tgt, etype, expired)| TEdge {
            src,
            tgt,
            etype,
            expired,
        })
}

async fn fresh_traversal(harness: &TestHarness) -> (libsql::Database, libsql::Connection) {
    let db = libsql::Builder::new_local(&harness.db_path)
        .build()
        .await
        .unwrap();
    let conn = db.connect().unwrap();
    macrame::schema::run_migrations(&conn).await.unwrap();
    for id in TNODES {
        conn.execute(
            "INSERT INTO concepts (id, title, valid_from, recorded_at) VALUES (?1, 'N', ?2, ?2)",
            libsql::params![id, TS[0]],
        )
        .await
        .unwrap();
    }
    (db, conn)
}

async fn apply_edges(conn: &libsql::Connection, edges: &[TEdge]) {
    for e in edges {
        // Errors are ignored the way `apply` ignores them: a generated duplicate
        // trips `trg_links_single_open`, which is the schema working.
        let _ = conn
            .execute(
                "INSERT INTO links (source_id, target_id, edge_type, valid_from, valid_to, \
                 weight, properties, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, 1.0, '{}', ?4)",
                libsql::params![
                    TNODES[e.src],
                    TNODES[e.tgt],
                    TYPES[e.etype],
                    TS[0],
                    if e.expired { TS[2] } else { SENTINEL },
                ],
            )
            .await;
    }
}

async fn rows_of(conn: &libsql::Connection, sql: &str, depth: usize) -> Vec<Vec<String>> {
    let mut rows = conn
        .query(
            sql,
            libsql::params![TNODES[0], depth as i64, TS[3], f64::NEG_INFINITY],
        )
        .await
        .unwrap();
    let mut out = Vec::new();
    while let Some(r) = rows.next().await.unwrap() {
        let mut cols = Vec::new();
        for i in 0.. {
            match r.get::<String>(i) {
                Ok(v) => cols.push(v),
                Err(_) => break,
            }
        }
        out.push(cols);
    }
    out
}

proptest! {
    #![proptest_config(ProptestConfig { cases: 32, ..ProptestConfig::default() })]

    /// **T0.1: the rewrite preserves the answer.** Node sets and projected edge
    /// sets are identical between the simple-path form and the walk form, at
    /// every depth, over generated graphs.
    ///
    /// The equivalence is provable — if a walk of length `k ≤ D` reaches `X`,
    /// excising its cycles gives a simple path of length `≤ k` that also reaches
    /// `X`, so the two reachability relations coincide — and an argument is
    /// exactly what this project does not accept on its own for a query. The
    /// generator produces cycles, self-loops, diamonds and expired edges,
    /// which are the four shapes the proof steps over.
    ///
    /// The shipped `TraversalBuilder::build_sql()` is compared too, so this is a
    /// test of the query that runs rather than of a copy of it kept in step by
    /// hand.
    #[test]
    fn the_traversal_rewrite_returns_what_the_simple_path_form_returned(
        edges in prop::collection::vec(tedge_strategy(), 0..10),
        depth in 1usize..5,
    ) {
        block_on(async {
            let harness = TestHarness::new();
            let (_db, conn) = fresh_traversal(&harness).await;
            apply_edges(&conn, &edges).await;

            let old_ids = rows_of(&conn, &format!("{OLD_WALK}{IDS_PROJECTION}"), depth).await;
            let new_ids = rows_of(&conn, &format!("{NEW_WALK}{IDS_PROJECTION}"), depth).await;
            prop_assert_eq!(&old_ids, &new_ids, "node sets diverged at depth {}", depth);

            let old_edges = rows_of(&conn, &format!("{OLD_WALK}{EDGES_PROJECTION}"), depth).await;
            let new_edges = rows_of(&conn, &format!("{NEW_WALK}{EDGES_PROJECTION}"), depth).await;
            prop_assert_eq!(&old_edges, &new_edges, "edge sets diverged at depth {}", depth);

            // And the query the crate actually issues agrees with both.
            let shipped = macrame::graph::TraversalBuilder::new(TNODES[0])
                .max_depth(depth)
                .min_weight(f64::NEG_INFINITY)
                .build_sql();
            let shipped_ids = rows_of(&conn, &shipped, depth).await;
            prop_assert_eq!(&old_ids, &shipped_ids, "build_sql diverged at depth {}", depth);

            Ok(())
        })?;
    }
}