kglite 0.16.3

//! D1 — column-store ownership: divergence coverage and the ownership pins.
//!
//! # What "divergence" means here
//!
//! Before D1 Phase 3 a columnar type's `ColumnStore` was reachable through
//! **two** `Arc`s on a memory/mapped graph: a `DirGraph`-level master and the
//! handle inside every node's `PropertyStorage::Columnar`. Nothing stopped
//! those drifting apart — `Arc::make_mut` on either side forked — and deleting
//! one of them was the whole point of the programme. The backend is the sole
//! owner now; these tests keep asking the same questions of the surviving
//! route.
//!
//! This module forks them deliberately and then asks every public read surface
//! what it sees. Two classes of assertion live here:
//!
//! 1. **Cross-surface consistency** (`all_public_reads_agree_*`). Whatever a
//!    read resolves to, *every* surface must resolve to the same thing. This
//!    holds today (all of them read the node handle) and must still hold after
//!    Phase 3 (all of them will read the backend's store). It is
//!    phase-independent and is the real gate.
//! 2. **Which replica wins** (`*_today_*`). Pinned as an exact fact with an
//!    inversion instruction, in the style of `handle.rs`'s
//!    `held_reader_forces_a_whole_graph_copy`. Phase 3 flips these; a failure
//!    before then means an unintended ownership change.
//!
//! # Phase 2 — the mutation-proof gate
//!
//! Phase 1 re-routed every caller; Phase 2 makes that irreversible. Two layers:
//!
//! - **Compile-time.** A columnar node's store handle lives behind
//!   `ColumnarRow`, whose `store` field is private to `graph::storage`, and the
//!   `NodeData` property readers Phase 1 emptied are deleted. New code
//!   *cannot* express a direct-route read; it fails to compile. The two named
//!   escapes (`ColumnarRow::node_handle` / `::repoint`) are pinned site-for-site
//!   by `no_code_reaches_a_node_held_column_store_handle`, so the
//!   remaining direct-route set is an enumerated work list rather than a
//!   guess.
//! - **Runtime**, for what the compiler cannot see: a caller that *could* have
//!   used the accessors but reads a `NodeData` it already holds. `poison_*`
//!   makes the node handle and the backend's store disagree — exactly as
//!   Phase 3 will — and one named test per caller class asserts the class
//!   observes the authoritative value. Each was shown red by reverting that
//!   one call site; see the commit body.
//!
//! # Why the divergence tests do not just assert "the master wins"
//!
//! Through Phases 1-2 the node handle *was* the read route on memory/mapped,
//! and a re-point sweep pushed master writes back onto the nodes at
//! end-of-clause; asserting master-authority then would have been a
//! permanently red test, so the pins recorded the current answer instead.
//! Phase 3 removed the handle, and those pins are inverted in place rather
//! than deleted — each one names what it used to assert.

use std::collections::HashMap;
use std::sync::Arc;

use crate::datatypes::{DataFrame, Value};
use crate::graph::dir_graph::DirGraph;
use crate::graph::schema::{InternedKey, PropertyStorage};
use crate::graph::session::{execute_mut, execute_read, ExecuteOptions};
use crate::graph::storage::column_store::ColumnStore;
use crate::graph::storage::GraphRead;
use petgraph::graph::NodeIndex;

const N: i64 = 4;

fn run(graph: &mut DirGraph, query: &str) {
    let params = HashMap::new();
    let opts = ExecuteOptions::eager(&params);
    execute_mut(graph, query, &opts).unwrap_or_else(|e| panic!("setup query failed: {query}: {e}"));
}

fn read_one(graph: &DirGraph, query: &str) -> Value {
    let params = HashMap::new();
    let opts = ExecuteOptions::eager(&params);
    let out = execute_read(graph, query, &opts).unwrap_or_else(|e| panic!("{query}: {e}"));
    out.result
        .rows
        .first()
        .and_then(|r| r.first())
        .cloned()
        .unwrap_or(Value::Null)
}

/// `n` `Item` nodes with two ordinary properties each — **not** yet columnar.
fn sized_rows(n: i64) -> DirGraph {
    let mut g = DirGraph::new();
    let rows: Vec<Vec<Value>> = (1..=n)
        .map(|i| {
            vec![
                Value::Int64(i),
                Value::String(format!("t{i}")),
                Value::String(format!("c0-{i}")),
                Value::Int64(i * 10),
            ]
        })
        .collect();
    let df = DataFrame::from_cypher_rows(
        vec![
            "id".to_string(),
            "title".to_string(),
            "c0".to_string(),
            "c1".to_string(),
        ],
        rows,
    )
    .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut g,
        df,
        "Item".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();
    g
}

/// A row-storage fixture, for tests that drive `enable_columnar` themselves.
fn docs_fixture() -> DirGraph {
    sized_rows(N)
}

/// `n` `Item` nodes after `enable_columnar()` — the shape every graph takes the
/// moment it is saved.
fn sized_columnar(n: i64) -> DirGraph {
    let mut g = sized_rows(n);
    g.enable_columnar();
    assert!(
        g.column_store_count() > 0,
        "fixture must own a master column store, or every arm below is vacuous"
    );
    assert!(
        node_row_id(&g, node_of(&g, 1)).is_some(),
        "fixture nodes must be columnar rows, or the ownership arms are vacuous"
    );
    g
}

fn seeded_columnar() -> DirGraph {
    sized_columnar(N)
}

fn node_of(graph: &DirGraph, id: i64) -> NodeIndex {
    graph
        .graph
        .node_indices()
        .find(|&i| graph.graph.get_node_id(i) == Some(Value::Int64(id)))
        .unwrap_or_else(|| panic!("no Item with id {id}"))
}

fn node_row_id(graph: &DirGraph, idx: NodeIndex) -> Option<u32> {
    match graph.graph.node_weight(idx).map(|n| &n.properties) {
        Some(PropertyStorage::Columnar(row)) => Some(row.row_id()),
        _ => None,
    }
}

/// Does node `idx` still share the master's `Arc`?
/// Is the type's master store owned by the backend alone?
///
/// The D1 Phase 3 successor to `node_shares_master`: there is no node-held
/// handle to compare against any more, so the question that matters is whether
/// anything at all shares the store — which is what decides whether the next
/// write forks or mutates in place.
fn master_is_uniquely_owned(graph: &DirGraph) -> bool {
    graph
        .column_store("Item")
        .is_some_and(|master| Arc::strong_count(master) == 1)
}

/// Write `value` straight into the type's master store.
///
/// **Named for the pre-Phase-3 world and kept for continuity of the tests that
/// call it, but it no longer diverges anything.** It used to fork the master
/// away from the node-held handles, and `Arc::make_mut` succeeded at that
/// precisely because the nodes held strong handles (D1 §1.2). Phase 3 deleted
/// those handles, so the store is uniquely owned (see
/// `master_is_uniquely_owned`) and `make_mut` now mutates it in place. The
/// callers are consequently asserting "a read returns what the backend's store
/// holds", not "the surfaces agree despite a divergence" — which is the
/// strongest statement still expressible, since the divergence it was built to
/// create is no longer constructible.
///
/// Returns the interned key written.
fn diverge_master(graph: &mut DirGraph, idx: NodeIndex, key: &str, value: Value) -> InternedKey {
    let row_id = node_row_id(graph, idx).expect("columnar node");
    let ikey = graph.interner.get_or_intern(key);
    let master = Arc::make_mut(graph.column_store_mut("Item").expect("master store"));
    assert!(
        master.set(row_id, ikey, &value, None),
        "master write must land"
    );
    ikey
}

/// Pull `c0` for the node with `id: 1` out of a D3-JSON export.
/// `Value::Null` when the key is absent (which is what a REMOVE produces).
fn extract_json_c0(json: &str) -> Value {
    let obj = json
        .split('{')
        .find(|chunk| chunk.contains("\"id\":1,"))
        .unwrap_or("");
    match obj.split("\"c0\":").nth(1) {
        Some(rest) => {
            let raw = rest
                .split([',', '}'])
                .next()
                .unwrap_or("")
                .trim()
                .trim_matches('"');
            Value::String(raw.to_string())
        }
        None => Value::Null,
    }
}

/// The value each public read surface resolves for `Item{id:1}.c0`.
fn all_read_surfaces(
    graph: &mut DirGraph,
    idx: NodeIndex,
    ikey: InternedKey,
) -> Vec<(&'static str, Value)> {
    // `read_indexed` is the funnel every index / constraint build reads
    // through; its `PropertyReader` needs `&mut` only to intern the key.
    let reader = graph.property_reader("Item", "c0");
    let graph = &*graph;
    vec![
        (
            "GraphRead::node_view",
            graph
                .node_view(idx)
                .and_then(|v| v.get_property("c0"))
                .map(|c| c.into_owned())
                .unwrap_or(Value::Null),
        ),
        (
            "GraphRead::get_node_property",
            graph
                .graph
                .get_node_property(idx, ikey)
                .unwrap_or(Value::Null),
        ),
        (
            "GraphRead::node_row_properties",
            graph
                .graph
                .node_row_properties(idx)
                .into_iter()
                .find(|(k, _)| *k == ikey)
                .map(|(_, v)| v)
                .unwrap_or(Value::Null),
        ),
        (
            "DirGraph::read_indexed (index build funnel)",
            graph.read_indexed(&reader, idx).unwrap_or(Value::Null),
        ),
        (
            "Cypher RETURN n.c0",
            read_one(graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        ),
        (
            "Cypher RETURN n (whole-node projection)",
            match read_one(graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n") {
                Value::Node(nv) => nv.properties.get("c0").cloned().unwrap_or(Value::Null),
                other => panic!("expected a node value, got {other:?}"),
            },
        ),
        (
            "Cypher properties(n)",
            match read_one(graph, "MATCH (n:Item) WHERE n.id = 1 RETURN properties(n)") {
                Value::Map(m) => m.get("c0").cloned().unwrap_or(Value::Null),
                other => panic!("expected a map, got {other:?}"),
            },
        ),
        ("D3-JSON export", {
            let json = crate::graph::io::export::to_d3_json(graph, None).unwrap();
            extract_json_c0(&json)
        }),
    ]
}

// ── 1. Cross-surface consistency — phase-independent ───────────────────────

/// Without divergence, every surface must see the stored value. Without this
/// arm the consistency test below would pass on a build where every surface
/// returned `Null`.
#[test]
fn all_public_reads_agree_without_divergence() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = InternedKey::from_str("c0");
    let stored = Value::String("c0-1".into());
    for (surface, got) in all_read_surfaces(&mut graph, idx, ikey) {
        assert_eq!(got, stored, "{surface} disagreed with the stored value");
    }
}

/// **The gate.** Under master/node divergence every public read surface must
/// still agree with every other one. Which replica wins is pinned separately;
/// what must never happen is two surfaces answering differently, because that
/// is a user-visible inconsistency no matter which side is authoritative.
#[test]
fn all_public_reads_agree_under_master_node_divergence() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));

    let surfaces = all_read_surfaces(&mut graph, idx, ikey);
    let (first_name, first) = surfaces[0].clone();
    for (surface, got) in &surfaces[1..] {
        assert_eq!(
            got, &first,
            "{surface} resolved {got:?} but {first_name} resolved {first:?} — \
             two public reads of the same property must never disagree"
        );
    }
    // Agreement alone is satisfied by every surface returning `Null`, which is
    // exactly what a storeless columnar node produces — unanimously, and
    // wrongly. Pin the value the store actually holds so this cannot pass by
    // agreeing on nothing.
    assert_eq!(
        first,
        Value::String("MASTER".into()),
        "{first_name} agreed with the others on {first:?}; all surfaces \
         returning Null is agreement without a read"
    );
}

// ── 2. Which replica wins — pinned, inverted by Phase 3 ────────────────────

/// **Inverted by D1 Phase 3** (was `today_the_node_handle_wins_over_the_master`).
///
/// Before Phase 3 a master-only write was invisible: every read resolved
/// through the node's own `Arc`, and this test asserted it read the *stale*
/// value with an instruction to flip when ownership moved. Ownership has moved.
/// The backend's store is now the only store, so a write into it is what every
/// read returns.
#[test]
fn the_backend_store_is_the_only_read_route() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));
    assert_eq!(
        graph.graph.get_node_property(idx, ikey),
        Some(Value::String("MASTER".into())),
        "a write into the backend's store must be what a read returns — there \
         is no second replica left to shadow it"
    );
}

// ── 3. Writes reconverge the two replicas ──────────────────────────────────

/// A columnar `SET` writes through the master and then re-points every node of
/// the type. Post-Phase-3 there are no node handles to re-point: the write goes
/// into the store the backend owns, the journal releases its pre-image at
/// commit, and every surface reads the new value.
#[test]
fn set_leaves_the_master_uniquely_owned() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));

    run(
        &mut graph,
        "MATCH (n:Item) WHERE n.id = 1 SET n.c0 = 'WRITTEN'",
    );

    assert!(
        master_is_uniquely_owned(&graph),
        "a committed columnar SET must leave the master uniquely owned — the \
         journal's pre-image is released at commit"
    );
    let want = Value::String("WRITTEN".into());
    for (surface, got) in all_read_surfaces(&mut graph, idx, ikey) {
        assert_eq!(got, want, "{surface} did not observe the SET");
    }
}

/// The same for `REMOVE`, which takes a different master path
/// (`Arc::make_mut(master).set(.., Null, ..)`).
#[test]
fn remove_leaves_the_master_uniquely_owned() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));

    run(&mut graph, "MATCH (n:Item) WHERE n.id = 1 REMOVE n.c0");

    assert!(
        master_is_uniquely_owned(&graph),
        "a committed columnar REMOVE must leave the master uniquely owned"
    );
    for (surface, got) in all_read_surfaces(&mut graph, idx, ikey) {
        assert_eq!(got, Value::Null, "{surface} still sees a removed property");
    }
}

/// A `MERGE` key read must resolve the same value the read surfaces do —
/// otherwise `MERGE` would create a duplicate for a row that already matches
/// (or match a row that does not).
#[test]
fn merge_key_read_matches_the_public_read() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let _ = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));

    let before = graph.graph.node_count();
    let observed = read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0");
    let observed_str = match &observed {
        Value::String(s) => s.clone(),
        other => panic!("expected a string, got {other:?}"),
    };
    run(
        &mut graph,
        &format!("MERGE (n:Item {{id: 1, c0: '{observed_str}'}})"),
    );
    assert_eq!(
        graph.graph.node_count(),
        before,
        "MERGE on the value the public read reports must match the existing row, not create one"
    );
}

/// A rolled-back statement must leave every surface on the pre-statement value.
/// The columnar SET path emits no `NodeWeight` undo entry — its only signal is
/// the per-cell `UndoEntry::ColumnarCell` — so this is the arm that proves the
/// journal covers the master write at all.
#[test]
fn rollback_restores_every_read_surface() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let ikey = InternedKey::from_str("c0");
    let before = graph.graph.get_node_property(idx, ikey);

    // Two patterns: the first commits its SET, the second is rejected, so the
    // whole statement rolls back.
    let params = HashMap::new();
    let opts = ExecuteOptions::eager(&params);
    let err = execute_mut(
        &mut graph,
        "MATCH (n:Item) WHERE n.id = 1 SET n.c0 = 'DOOMED', \
         n.c1 = duration({months: 2147483648})",
        &opts,
    );
    assert!(err.is_err(), "the fixture statement must fail to roll back");

    assert_eq!(
        graph.graph.get_node_property(idx, ikey),
        before,
        "a rolled-back columnar SET must restore the pre-statement value"
    );
    for (surface, got) in all_read_surfaces(&mut graph, idx, ikey) {
        assert_eq!(
            Some(got),
            before.clone(),
            "{surface} kept a rolled-back value"
        );
    }
}

/// Save + reload must round-trip whatever the read surfaces report — a
/// divergence that only the writer can see is a data-loss bug, not a caching
/// one.
#[test]
fn save_and_reload_round_trips_the_observed_value() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let _ = diverge_master(&mut graph, idx, "c0", Value::String("MASTER".into()));
    run(
        &mut graph,
        "MATCH (n:Item) WHERE n.id = 1 SET n.c0 = 'PERSISTED'",
    );

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("g.kgl");
    let mut arc = Arc::new(graph);
    crate::graph::io::file::prepare_save(&mut arc);
    Arc::make_mut(&mut arc).enable_columnar();
    crate::graph::io::file::write_kgl(&arc, path.to_str().unwrap()).unwrap();

    let loaded = crate::graph::io::file::load_file(path.to_str().unwrap()).unwrap();
    assert_eq!(
        read_one(&loaded, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        Value::String("PERSISTED".into()),
        "the saved file must carry the value the reads reported"
    );
}

// ── 4. Defect 2 — `maybe_spill_columns` reclaims nothing ───────────────────

/// **D1 defect 2 — CLOSED, and inverted here** (was
/// `spill_forks_the_master_and_reclaims_nothing_today`).
///
/// `maybe_spill_columns` calls `Arc::make_mut` on the type's store and then
/// `materialize_to_files`. Before Phase 3 every node held a strong handle, so
/// `make_mut` *forked*: the master became the file-backed copy while all N
/// nodes kept the pre-spill in-heap store alive, and — unlike the SET path —
/// no sweep re-pointed them. Reads stayed correct; the memory the spill exists
/// to reclaim was never reclaimed.
///
/// With the backend the sole owner the store is uniquely owned, `make_mut`
/// mutates in place, and the spilled store *is* the one every read resolves.
/// The two assertions the old test made are inverted verbatim.
#[test]
fn spill_reclaims_the_heap_it_materialises() {
    let mut graph = seeded_columnar();
    let dir = tempfile::tempdir().unwrap();
    graph.spill_dir = Some(dir.path().to_path_buf());
    // Any limit below the store's heap footprint forces a spill.
    graph.memory_limit = Some(0);

    assert!(
        master_is_uniquely_owned(&graph),
        "precondition: nothing but the backend owns the store before the spill"
    );
    let heap_before = graph
        .column_store("Item")
        .expect("master store")
        .heap_bytes();
    assert!(heap_before > 0, "precondition: the store holds heap data");

    graph.maybe_spill_columns();

    let master = graph.column_store("Item").expect("master store");
    assert!(
        master.is_mapped(),
        "the spill must have materialised the master to files, or this test proves nothing"
    );
    // INVERTED (was: the node keeps an unmapped pre-spill copy alive).
    assert!(
        master.heap_bytes() < heap_before,
        "the spill must reclaim heap: got {} bytes, was {heap_before}. Before D1 \
         Phase 3 `make_mut` forked and this number never moved.",
        master.heap_bytes()
    );
    assert!(
        master_is_uniquely_owned(&graph),
        "and the spilled store must still be the uniquely-owned one — a fork \
         here would mean the reclaimed copy is not what reads resolve"
    );

    // The user-visible contract is unchanged: reads still resolve.
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        Value::String("c0-1".into()),
        "a spill must never change what a read returns"
    );
}

// ── 5. Defect 1 — columnar enumeration completeness ────────────────────────

/// `describe()`'s per-type property block and node samples read through the
/// accessors now; before D1 Phase 1 they went through `NodeData::property_iter`
/// and enumerated **nothing** for a saved graph.
#[test]
fn describe_reports_columnar_properties() {
    use crate::graph::introspection::{ConnectionDetail, CypherDetail, FluentDetail};
    let graph = seeded_columnar();
    let xml = crate::graph::introspection::describe::compute_description(
        &graph,
        None,
        &ConnectionDetail::Off,
        &CypherDetail::Off,
        &FluentDetail::Off,
        None,
        None,
        None,
    )
    .unwrap();
    assert!(
        xml.contains("c0"),
        "describe() lost a columnar property: {xml}"
    );
    assert!(
        xml.contains("c0-1"),
        "describe()'s node sample lost a columnar property value: {xml}"
    );
}

/// `compute_property_stats` accumulates from the row, not just from the
/// `type_schemas` pre-seed: a columnar property must report a non-zero
/// non-null count and real sample values.
#[test]
fn property_stats_count_columnar_rows() {
    let graph = seeded_columnar();
    let stats = crate::graph::introspection::schema_overview::compute_property_stats(
        &graph, "Item", 32, None,
    )
    .expect("property stats");
    let c0 = stats
        .iter()
        .find(|p| p.property_name == "c0")
        .expect("c0 must appear in the property stats");
    assert_eq!(
        c0.non_null, N as usize,
        "columnar rows contributed no values to the property stats"
    );
    assert_eq!(
        c0.unique, N as usize,
        "columnar rows contributed no distinct values"
    );
}

/// `property_ndv` — the planner's selectivity input — must see columnar rows.
/// It bypasses `read_indexed` and reads the node directly, so it is one of the
/// callers the inventory flagged as "a reader would assume it is covered".
#[test]
fn property_ndv_counts_columnar_rows() {
    let graph = seeded_columnar();
    assert_eq!(
        graph.property_ndv("Item", "c0"),
        Some(N as usize),
        "property_ndv must see a columnar type's distinct values"
    );
}

// ══════════════════════════════════════════════════════════════════════════
// Phase 2 — the mutation-proof gate
// ══════════════════════════════════════════════════════════════════════════

// ── The poison primitive ──────────────────────────────────────────────────

/// Install a **different** store for `node_type`, with `edit` applied.
///
/// # What this proves after D1 Phase 3
///
/// In Phase 2 the poison had to fabricate a disagreement between two replicas —
/// a stale copy on every node and the truth in the master — because both routes
/// pointed at the same object and a gate that cannot tell them apart is not a
/// gate. Phase 3 deleted the node-held replica outright, so the disagreement is
/// no longer expressible: `column_store(type)` *is* the read route, and the
/// compile-time gate (`ColumnarRow` carries a row id and nothing else) is what
/// now rules out the class the thread-local hook used to catch.
///
/// What survives here is still worth having: a caller that captured an `Arc` of
/// the store earlier — a cache, a snapshot taken across a write — keeps reading
/// the old object, and every named class test below re-reads through the
/// backend after this swap. The mechanism got simpler because the ownership
/// got simpler.
fn poison_row(
    graph: &mut DirGraph,
    node_type: &str,
    edit: impl FnOnce(&mut ColumnStore),
) -> PoisonGuard {
    let mut replacement: ColumnStore = (**graph
        .column_store(node_type)
        .expect("type must be columnar, or the poison is a no-op"))
    .clone();
    edit(&mut replacement);
    graph.install_column_store(node_type, Arc::new(replacement));
    PoisonGuard
}

/// Kept as a unit so the call sites read unchanged across Phase 2 → 3; the
/// swap is permanent for the graph under test, which is built per test.
struct PoisonGuard;

/// Poison one row's property column.
fn poison_property(
    graph: &mut DirGraph,
    node_type: &str,
    row_id: u32,
    key: &str,
    value: Value,
) -> PoisonGuard {
    let ikey = graph.interner.get_or_intern(key);
    poison_row(graph, node_type, move |store| {
        assert!(
            store.set(row_id, ikey, &value, None),
            "poison write must land, or the swap proves nothing"
        );
    })
}

/// Poison one row's `__title__` column.
fn poison_title(graph: &mut DirGraph, node_type: &str, row_id: u32, value: Value) -> PoisonGuard {
    poison_row(graph, node_type, move |store| {
        assert!(
            store.set_title(row_id, &value),
            "poison title write must land, or the swap proves nothing"
        );
    })
}

/// Fixture: a saved graph with row 0 (`id: 1`) poisoned so its authoritative
/// `c0` is `TRUTH` while its node handle still says `c0-1`.
fn poisoned_fixture() -> (DirGraph, NodeIndex, PoisonGuard) {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).expect("columnar node");
    let guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("TRUTH".into()),
    );
    (graph, idx, guard)
}

/// **Non-vacuity for the class tests below.** The swap must install a genuinely
/// different allocation and the read route must resolve it — otherwise every
/// named test below would pass by accident.
#[test]
fn poison_installs_a_distinct_store_that_reads_resolve() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let before = Arc::as_ptr(graph.column_store("Item").expect("master"));

    let row_id = node_row_id(&graph, idx).unwrap();
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("TRUTH".into()),
    );

    let after = Arc::as_ptr(graph.column_store("Item").expect("master"));
    assert!(
        !std::ptr::eq(before, after),
        "the poison must install a distinct allocation, or a caller holding the \
         old one would be indistinguishable from one reading the new"
    );
    assert_eq!(
        graph.node_view(idx).unwrap().get_property_value("c0"),
        Some(Value::String("TRUTH".into())),
        "and the read route must resolve the newly installed store"
    );
    assert_eq!(
        node_row_id(&graph, idx),
        Some(row_id),
        "the node's row identity must be untouched — the swap is of the store, \
         not of the node"
    );
}

// ── One named test per caller class ───────────────────────────────────────

/// **R1 — pattern matcher filter.** `MATCH (n:Item {c0: …})` resolves the
/// authoritative value, so the inline-property filter finds the poisoned row
/// and not the stale one.
#[test]
fn r1_matcher_property_filter_reads_the_authoritative_store() {
    let (graph, _idx, _guard) = poisoned_fixture();
    assert_eq!(
        read_one(&graph, "MATCH (n:Item {c0: 'TRUTH'}) RETURN n.id"),
        Value::Int64(1),
        "the matcher's property filter must see the authoritative value"
    );
    assert_eq!(
        read_one(&graph, "MATCH (n:Item {c0: 'c0-1'}) RETURN n.id"),
        Value::Null,
        "the matcher must not match the stale replica"
    );
}

/// **R3 — WHERE / expression resolution.**
#[test]
fn r3_where_clause_reads_the_authoritative_store() {
    let (graph, _idx, _guard) = poisoned_fixture();
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.c0 = 'TRUTH' RETURN n.id"),
        Value::Int64(1)
    );
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        Value::String("TRUTH".into())
    );
}

/// **R4 — projection / whole-node materialisation.**
#[test]
fn r4_whole_node_projection_reads_the_authoritative_store() {
    let (graph, _idx, _guard) = poisoned_fixture();
    match read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n") {
        Value::Node(nv) => assert_eq!(
            nv.properties.get("c0"),
            Some(&Value::String("TRUTH".into())),
            "RETURN n must carry the authoritative value"
        ),
        other => panic!("expected a node value, got {other:?}"),
    }
}

/// **R8 — index build funnel (`read_indexed`).** A property index built after
/// the poison buckets the row under its authoritative value.
#[test]
fn r8_property_index_build_reads_the_authoritative_store() {
    let (mut graph, idx, _guard) = poisoned_fixture();
    graph.create_index("Item", "c0");
    let bucket = graph
        .property_indices
        .get(&("Item".to_string(), "c0".to_string()))
        .expect("index must exist");
    assert_eq!(
        bucket.get(&Value::String("TRUTH".into())),
        Some(&vec![idx]),
        "the built index must bucket the row under its authoritative value"
    );
    assert!(
        !bucket.contains_key(&Value::String("c0-1".into())),
        "the built index must not carry the stale replica's value"
    );
}

/// **R9 — incremental index maintenance.** The incremental updater
/// (`update_property_indices_for_add`) reads through `read_indexed` for exactly
/// this reason; it gets its own arm because it is a separate call path from the
/// rebuild above, and the two must file a row identically.
#[test]
fn r9_incremental_index_maintenance_reads_the_authoritative_store() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).unwrap();
    // Build the index *before* the poison, from the pre-poison values.
    graph.create_index("Item", "c0");
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("TRUTH".into()),
    );

    graph.update_property_indices_for_add("Item", idx);

    let bucket = graph
        .property_indices
        .get(&("Item".to_string(), "c0".to_string()))
        .expect("index must exist");
    assert!(
        bucket
            .get(&Value::String("TRUTH".into()))
            .is_some_and(|members| members.contains(&idx)),
        "incremental maintenance must file the row under its authoritative \
         value, or it disagrees with a rebuilt index"
    );
}

/// **R11 — constraint gates.** Declaring a unique constraint validates the
/// existing rows through `read_indexed`; with two rows sharing an
/// authoritative `c0`, the declaration must be rejected.
#[test]
fn r11_unique_constraint_gate_reads_the_authoritative_store() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).unwrap();
    // Collide row 0 with row 1's value (`c0-2`) in the master only.
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("c0-2".into()),
    );

    let params = HashMap::new();
    let opts = ExecuteOptions::eager(&params);
    let result = execute_mut(
        &mut graph,
        "CREATE CONSTRAINT FOR (i:Item) REQUIRE i.c0 IS UNIQUE",
        &opts,
    );
    assert!(
        result.is_err(),
        "the constraint gate must see the authoritative duplicate and reject; \
         reading the stale node handles would show four distinct values"
    );
}

/// **R12 — planner statistics.** `property_ndv` bypasses `read_indexed`, so it
/// gets its own arm: the collision above must drop the distinct count.
#[test]
fn r12_property_ndv_reads_the_authoritative_store() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).unwrap();
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("c0-2".into()),
    );
    assert_eq!(
        graph.property_ndv("Item", "c0"),
        Some(N as usize - 1),
        "property_ndv must count the authoritative values; reading the stale \
         node handles would still report {N} distinct"
    );
}

/// **R13a — export.** The D3-JSON exporter enumerates a node's properties.
#[test]
fn r13_export_reads_the_authoritative_store() {
    let (graph, _idx, _guard) = poisoned_fixture();
    let json = crate::graph::io::export::to_d3_json(&graph, None).unwrap();
    assert_eq!(
        extract_json_c0(&json),
        Value::String("TRUTH".into()),
        "D3-JSON export must carry the authoritative value"
    );
}

/// **R13b — introspection statistics.** Asserted on `compute_property_stats`
/// directly rather than on the rendered `describe()` XML: the XML mentions a
/// value in several places, so a substring check there cannot tell which
/// producer supplied it, and a mutation of the stats accumulator left it green.
#[test]
fn r13_property_stats_read_the_authoritative_store() {
    let (graph, _idx, _guard) = poisoned_fixture();
    let stats = crate::graph::introspection::schema_overview::compute_property_stats(
        &graph, "Item", 32, None,
    )
    .expect("property stats");
    let c0 = stats
        .iter()
        .find(|p| p.property_name == "c0")
        .expect("c0 must appear in the property stats");
    let values = c0.values.as_ref().expect("small-cardinality values");
    assert!(
        values.contains(&Value::String("TRUTH".into())),
        "property stats must observe the authoritative value; got {values:?}"
    );
    assert!(
        !values.contains(&Value::String("c0-1".into())),
        "property stats must not observe the stale replica; got {values:?}"
    );
}

/// **R14 — binding-layer readers.** `session::resolve_noderefs` is public API,
/// runs after the executor returns and holds only a `&GraphBackend`.
#[test]
fn r14_resolve_noderefs_reads_the_authoritative_store() {
    let mut graph = seeded_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).unwrap();
    let _guard = poison_title(
        &mut graph,
        "Item",
        row_id,
        Value::String("TRUE-TITLE".into()),
    );

    let mut rows = vec![vec![Value::NodeRef(idx.index() as u32)]];
    crate::graph::session::resolve_noderefs(&graph.graph, &mut rows);
    assert_eq!(
        rows[0][0],
        Value::String("TRUE-TITLE".into()),
        "resolve_noderefs must resolve the authoritative title"
    );
}

// ── The compile-time gate's enumerated escape list ────────────────────────

/// **D1 Phase 3 landed: the escapes are gone.**
///
/// `ColumnarRow::node_handle` and `::repoint` were the only two ways to reach a
/// node's own `Arc<ColumnStore>` outside `graph::storage`, and Phase 2 pinned
/// their call sites file-by-file as the Phase-3 work list. Phase 3 deleted the
/// field they exposed, so both methods and every one of their 13 call sites are
/// gone — the expected set is empty.
///
/// The test is kept rather than deleted because an empty expectation is the
/// strongest form of the gate: re-introducing either name anywhere in the crate
/// fails it. If a future phase legitimately needs a node-held handle again, it
/// has to say so here.
const NODE_HANDLE_ESCAPE_SITES: &[(&str, usize)] = &[];

#[test]
fn no_code_reaches_a_node_held_column_store_handle() {
    use std::collections::BTreeMap;

    let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
    let mut found: BTreeMap<String, usize> = BTreeMap::new();

    // Split so this detector does not match its own source line — otherwise
    // the file's count would include the scanner and drift every time this
    // function is edited.
    let read_escape = concat!(".node_", "handle()");
    let write_escape = concat!(".re", "point(");

    fn walk(
        dir: &std::path::Path,
        root: &std::path::Path,
        needles: (&str, &str),
        found: &mut BTreeMap<String, usize>,
    ) {
        for entry in std::fs::read_dir(dir).expect("readable source dir") {
            let path = entry.expect("dir entry").path();
            if path.is_dir() {
                walk(&path, root, needles, found);
            } else if path.extension().is_some_and(|e| e == "rs") {
                let text = std::fs::read_to_string(&path).expect("readable source file");
                let hits = text.matches(needles.0).count() + text.matches(needles.1).count();
                if hits > 0 {
                    let rel = path
                        .strip_prefix(root)
                        .expect("under src")
                        .to_string_lossy()
                        .replace('\\', "/");
                    *found.entry(rel).or_insert(0) += hits;
                }
            }
        }
    }
    walk(&src, &src, (read_escape, write_escape), &mut found);

    let expected: BTreeMap<String, usize> = NODE_HANDLE_ESCAPE_SITES
        .iter()
        .map(|(f, n)| ((*f).to_string(), *n))
        .collect();

    assert_eq!(
        found, expected,
        "\nA node-held column-store handle is reachable again.\n\
         D1 Phase 3 made the storage backend the sole owner: a node carries a \
         row id, and the store is resolved by `GraphRead::column_store`. Read \
         through `NodeView` / `GraphRead` and write through \
         `GraphWrite::set_node_property` instead of re-introducing a per-node \
         handle.\n"
    );
}

// ── The save fast path (D1 risk 1) ────────────────────────────────────────

/// Saving a freshly built graph must **not** rebuild the stores — and neither
/// must saving it again.
///
/// The first half is what the always-columnar flip buys: a graph is built in
/// the shape it is saved in, so the consolidation pass has nothing to
/// consolidate and `save()` no longer changes the write regime. Before the
/// flip the first `enable_columnar` rebuilt every store (the assertion here
/// used to *require* that, as the only way to tell a skipped rebuild from a
/// graph that had never been columnar).
///
/// The second half is the idempotence guard whose replacement reasoning D1
/// risk 1 flagged as *believed* sufficient — counted rather than trusted,
/// because losing the fast path costs a full O(N) rebuild on every save
/// (~257 s at wiki100m).
#[test]
fn saving_a_freshly_built_graph_skips_the_rebuild() {
    use crate::graph::dir_graph::COLUMNAR_REBUILDS;
    let rebuilds = || COLUMNAR_REBUILDS.with(|c| c.get());

    let mut graph = docs_fixture();
    assert!(
        graph.column_store_count() > 0,
        "construction must already be columnar, or this test measures nothing"
    );

    let before = rebuilds();
    graph.enable_columnar();
    graph.enable_columnar();
    assert_eq!(
        rebuilds(),
        before,
        "a save of an unmodified graph must take the fast path; a rebuild here \
         means construction and the saved shape disagree, or the idempotence \
         guard regressed, and every save pays O(N)"
    );

    // And the guard must still *fire* when it should — otherwise "no rebuild"
    // would be trivially true and the assertion above would be vacuous. A
    // delete is the drift that survives the flip: it leaves the node's row
    // behind, and consolidation is what keeps that row out of the saved file.
    run(&mut graph, "MATCH (n:Item) WHERE n.id = 1 DELETE n");
    graph.enable_columnar();
    assert_eq!(
        rebuilds(),
        before + 1,
        "an orphaned row must still be detected as drift and rebuild"
    );
}

/// A create that reuses a freed petgraph slot puts the type's rows out of
/// ascending-node-index order, and the consolidation pass must notice.
///
/// Row order is part of the `.kgl` format: the column section is positional and
/// the load path binds row k of a type to that type's k-th node in ascending
/// index order. `rebuild_column_stores` sorts by node index, which is what makes
/// the two orders agree — so the drift check has to be what decides the rebuild
/// happens. It did not have to be, while a fresh graph rebuilt on every save
/// regardless; the moment the fast path became the normal case, a
/// delete-then-create pair started serializing every row against the wrong node
/// (`test_runtime_write_bugs.py::test_recreate_after_delete_is_fresh`, which
/// saw it as edges connecting different nodes after a reload).
#[test]
fn a_row_appended_out_of_index_order_is_detected_as_drift() {
    use crate::graph::dir_graph::COLUMNAR_REBUILDS;
    use crate::graph::schema::PropertyStorage;
    let rebuilds = || COLUMNAR_REBUILDS.with(|c| c.get());

    let mut graph = docs_fixture();
    // Free slot 0, then consolidate so the store is dense and in order again.
    run(&mut graph, "MATCH (n:Item) WHERE n.id = 1 DETACH DELETE n");
    graph.enable_columnar();
    let settled = rebuilds();
    graph.enable_columnar();
    assert_eq!(
        rebuilds(),
        settled,
        "precondition: the graph must be settled"
    );

    // The create takes the freed slot 0 but the appended row is last.
    run(&mut graph, "CREATE (:Item {id: 99, title: 'late'})");
    let out_of_order = graph
        .graph
        .node_indices()
        .filter_map(|idx| graph.graph.node_weight(idx))
        .filter_map(|node| match &node.properties {
            PropertyStorage::Columnar(row) => Some(row.row_id()),
            _ => None,
        })
        .enumerate()
        .any(|(position, row_id)| row_id as usize != position);
    assert!(
        out_of_order,
        "precondition: the create must have produced an out-of-order row, or \
         this test cannot see whether the drift check catches one"
    );

    graph.enable_columnar();
    assert_eq!(
        rebuilds(),
        settled + 1,
        "an out-of-order row must be detected as drift; without the rebuild the \
         save writes every row against the wrong node"
    );
    let ordered = graph
        .graph
        .node_indices()
        .filter_map(|idx| graph.graph.node_weight(idx))
        .filter_map(|node| match &node.properties {
            PropertyStorage::Columnar(row) => Some(row.row_id()),
            _ => None,
        })
        .enumerate()
        .all(|(position, row_id)| row_id as usize == position);
    assert!(ordered, "the rebuild must restore ascending row order");
}

/// A one-row `SET` on a saved type must touch one row, whatever N is.
///
/// The structural half of the perf claim (the timing half waits for Phase 5's
/// release measurement): the deleted sweep was O(N_type) per clause, so the
/// observable is that a graph of 200 nodes and a graph of 20 nodes both leave
/// every *other* row untouched and the master uniquely owned.
#[test]
fn a_one_row_columnar_set_leaves_every_other_row_untouched() {
    for n in [20i64, 200] {
        let mut graph = sized_columnar(n);
        let before: Vec<Option<Value>> = (0..n)
            .map(|i| {
                graph
                    .column_store("Item")
                    .unwrap()
                    .get(i as u32, InternedKey::from_str("c0"))
            })
            .collect();

        run(&mut graph, "MATCH (n:Item) WHERE n.id = 1 SET n.c0 = 'ONE'");

        let after: Vec<Option<Value>> = (0..n)
            .map(|i| {
                graph
                    .column_store("Item")
                    .unwrap()
                    .get(i as u32, InternedKey::from_str("c0"))
            })
            .collect();

        let changed: Vec<usize> = before
            .iter()
            .zip(&after)
            .enumerate()
            .filter(|(_, (b, a))| b != a)
            .map(|(i, _)| i)
            .collect();
        assert_eq!(
            changed.len(),
            1,
            "N={n}: a one-row SET must change exactly one row, changed {changed:?}"
        );
        assert!(
            master_is_uniquely_owned(&graph),
            "N={n}: and must leave the master uniquely owned, so the next write \
             mutates in place rather than copying the store"
        );
    }
}

/// Rewrite `Item {id: 1}` through `add_nodes` replace-mode with a batch that
/// carries `c0` and nothing else — so `c1` is the property the caller expects
/// to be gone afterwards.
fn replace_item_one_with_c0_only(graph: &mut DirGraph) {
    let df = DataFrame::from_cypher_rows(
        vec!["id".to_string(), "c0".to_string()],
        vec![vec![Value::Int64(1), Value::String("replaced".into())]],
    )
    .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        graph,
        df,
        "Item".to_string(),
        "id".to_string(),
        None,
        Some("replace".to_string()),
    )
    .unwrap();
}

/// `replace_node_properties` is replace-**all** on every backend, including the
/// overlay a held result view forks the writer into.
///
/// `add_nodes(conflict_handling="replace")` promises the row is rewritten, not
/// merged: a property the batch omits is gone afterwards. `ForkedGraph`'s
/// columnar arm wrote the incoming pairs over the row and left every other cell
/// exactly where it was, where both heap backends null the row first
/// (`impl_heap_column_writes!`). So holding a view turned a replace into an
/// update — a wrong read, not a slow one — and only on a saved (columnar)
/// graph.
///
/// The unforked run above the fork is the control: it is the semantics the
/// forked run has to match, and it passed on both sides of the fix, so a
/// failure there means the *reference* moved rather than the overlay.
#[test]
fn a_forked_columnar_replace_drops_the_properties_it_omits() {
    use crate::graph::handle::make_dir_graph_mut;

    const C1: &str = "MATCH (n:Item) WHERE n.id = 1 RETURN n.c1";
    const C0: &str = "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0";

    let mut control = seeded_columnar();
    assert_eq!(
        read_one(&control, C1),
        Value::Int64(10),
        "precondition: the fixture row carries the property the batch will omit"
    );
    replace_item_one_with_c0_only(&mut control);
    assert_eq!(
        read_one(&control, C0),
        Value::String("replaced".into()),
        "control: the batch's own property is written"
    );
    assert_eq!(
        read_one(&control, C1),
        Value::Null,
        "control: replace-mode drops a property the batch omits"
    );

    let mut writer = Arc::new(seeded_columnar());
    let reader = Arc::clone(&writer);
    let graph = make_dir_graph_mut(&mut writer);
    assert!(
        graph.graph.is_forked(),
        "precondition: a held view must fork the writer, or this is a second \
         run of the control"
    );
    replace_item_one_with_c0_only(graph);
    assert!(
        graph.graph.is_forked(),
        "precondition: the replace must land on the overlay, not after a flatten"
    );
    assert_eq!(
        read_one(graph, C0),
        Value::String("replaced".into()),
        "the batch's own property is written on the overlay too"
    );
    assert_eq!(
        read_one(graph, C1),
        Value::Null,
        "a replace on a forked columnar row must drop the properties the batch \
         omits, exactly as the unforked control does"
    );

    assert_eq!(
        read_one(&reader, C1),
        Value::Int64(10),
        "and the held view keeps the row it was forked with"
    );
}

// ── Backend arms: the same invariants on Mapped and on Disk ────────────────
//
// Every fixture above is `DirGraph::new()`, i.e. `Memory`. That was the gap
// filed as Track E after the 0.15.9 release review: the read-route and
// ownership pins were only ever asked of one backend, so a backend that
// resolved a property through some other replica — or that kept a second
// handle on the store across a write — would have gone unnoticed here.
//
// The arms below re-ask the two load-bearing questions on the other two
// backends. They are deliberately the *questions*, not the whole file: the
// caller-class matrix (`r1_` … `r14_`) exercises reader code that sits above
// the backend and cannot differ per backend, whereas "which store does a read
// resolve" and "who holds it after a write" are exactly the backend's business.

/// `n` `Item` nodes on `mode`, consolidated — the mapped/disk counterpart of
/// [`sized_columnar`].
///
/// The preconditions are asserted rather than assumed: a fixture that silently
/// landed on the wrong backend, or that carried no column store, would make
/// every arm below pass without testing anything.
fn sized_columnar_in_mode(
    n: i64,
    mode: crate::graph::storage::mode::StorageMode,
    path: Option<&std::path::Path>,
) -> DirGraph {
    let mut g = crate::graph::storage::mode::new_dir_graph_in_mode(mode, path)
        .expect("fixture backend must be constructible");
    let rows: Vec<Vec<Value>> = (1..=n)
        .map(|i| {
            vec![
                Value::Int64(i),
                Value::String(format!("t{i}")),
                Value::String(format!("c0-{i}")),
                Value::Int64(i * 10),
            ]
        })
        .collect();
    let df = DataFrame::from_cypher_rows(
        vec![
            "id".to_string(),
            "title".to_string(),
            "c0".to_string(),
            "c1".to_string(),
        ],
        rows,
    )
    .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut g,
        df,
        "Item".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();
    g.enable_columnar();
    assert_eq!(
        crate::graph::storage::mode::live_storage_mode(&g),
        mode,
        "fixture must be on the backend it names, or the arm tests Memory twice"
    );
    assert!(
        g.column_store_count() > 0,
        "fixture must own a master column store, or every arm below is vacuous"
    );
    g
}

fn mapped_columnar() -> DirGraph {
    sized_columnar_in_mode(N, crate::graph::storage::mode::StorageMode::Mapped, None)
}

/// A read must resolve the store the **backend** owns, on `Mapped` too.
///
/// Red-first: writing the poison into a *clone* of the store instead of
/// installing it turns this assertion red on `c0-1`.
#[test]
fn the_backend_store_is_the_only_read_route_on_mapped() {
    let mut graph = mapped_columnar();
    let idx = node_of(&graph, 1);
    let row_id = node_row_id(&graph, idx).expect("columnar node");
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("TRUTH".into()),
    );
    assert_eq!(
        graph.node_view(idx).unwrap().get_property_value("c0"),
        Some(Value::String("TRUTH".into())),
        "a mapped graph must read the store its backend owns"
    );
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        Value::String("TRUTH".into()),
        "and the Cypher read route must resolve the same store"
    );
}

/// A `SET` must mutate the mapped master **in place** and leave it uniquely
/// owned, exactly as it does on `Memory`.
///
/// Two assertions, and the pointer one is the load-bearing half.
/// "Uniquely owned afterwards" is satisfied trivially by a store that forked:
/// `Arc::make_mut` installs a *fresh* allocation in the map, so the map's entry
/// has one holder either way and only the old allocation is shared. Pinning the
/// allocation identity across the statement is what distinguishes "mutated one
/// cell" from "deep-copied every column of the type", which is the whole
/// mechanism this file exists to guard.
///
/// Red-first: holding a second `Arc` on the master across the statement — what
/// the pre-Phase-2 `ColumnarHandles` journal capture did — forces `make_mut` to
/// fork and turns the pointer assertion red. Verified by doing exactly that.
#[test]
fn set_leaves_the_master_uniquely_owned_on_mapped() {
    let mut graph = mapped_columnar();
    let idx = node_of(&graph, 1);
    let ikey = graph.interner.get_or_intern("c0");
    let before = Arc::as_ptr(graph.column_store("Item").expect("master"));

    run(
        &mut graph,
        "MATCH (n:Item) WHERE n.id = 1 SET n.c0 = 'WRITTEN'",
    );

    assert!(
        std::ptr::eq(
            before,
            Arc::as_ptr(graph.column_store("Item").expect("master"))
        ),
        "a columnar SET on Mapped must mutate the master in place — a changed \
         allocation means the statement deep-copied the type's columns"
    );
    assert!(
        master_is_uniquely_owned(&graph),
        "a committed columnar SET on Mapped must leave the master uniquely owned"
    );
    for (surface, got) in all_read_surfaces(&mut graph, idx, ikey) {
        assert_eq!(
            got,
            Value::String("WRITTEN".into()),
            "{surface} did not observe the SET on Mapped"
        );
    }
}

/// The same read-route question on `Disk`.
///
/// `DiskGraph` reaches a type's columns through its own arena rather than a
/// heap map, so "the backend owns the store" is a different claim there and is
/// worth asking separately. The write-ownership arm above has no disk
/// counterpart on purpose: a disk graph takes the whole-graph rollback
/// checkpoint (`supports_undo_journal() == false`), so nothing about the
/// journal's handle on a store applies to it.
///
/// Red-first: dropping the `install_column_store` call inside `poison_row`
/// leaves this reading `c0-1`.
#[test]
fn the_backend_store_is_the_only_read_route_on_disk() {
    let dir = tempfile::tempdir().expect("temp dir");
    let mut graph = sized_columnar_in_mode(
        N,
        crate::graph::storage::mode::StorageMode::Disk,
        Some(dir.path()),
    );
    // Every direct node read on a disk graph must sit inside an open query
    // guard (the arena SAFETY protocol in `disk/graph.rs`); the Cypher route
    // opens its own. Hence the scoping — the guard borrows the graph the
    // poison needs mutably.
    let (idx, row_id) = {
        let _query = graph.graph.begin_query();
        let idx = node_of(&graph, 1);
        let row_id = node_row_id(&graph, idx).expect("columnar node");
        (idx, row_id)
    };
    let _guard = poison_property(
        &mut graph,
        "Item",
        row_id,
        "c0",
        Value::String("TRUTH".into()),
    );
    {
        let _query = graph.graph.begin_query();
        assert_eq!(
            graph.node_view(idx).unwrap().get_property_value("c0"),
            Some(Value::String("TRUTH".into())),
            "a disk graph must read the store its backend owns"
        );
    }
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.c0"),
        Value::String("TRUTH".into()),
        "and the Cypher read route must resolve the same store"
    );
}

// ── 4b. P4 — the spill pass converges, and still fires when it must ────────

/// The spill trigger must reach a fixed point.
///
/// `heap_bytes` counts what a spill cannot move — the tombstone `Vec<bool>`,
/// the `Str` `relocated` overlay, `Mixed` columns, the overflow bag — so
/// comparing *it* against the limit left `StorageMode::Mapped`
/// (`memory_limit = Some(0)`) permanently over: the pass re-ran its whole
/// per-type loop (Vec + sort + a `create_dir_all` syscall per type) after every
/// mutating statement and spilled nothing each time. 245 us per single-row
/// `SET` at 100 types.
///
/// The convergence property is the pair of readings below: after one spill the
/// *spillable* total is zero while the total is not.
#[test]
fn the_spill_trigger_converges_on_the_unspillable_floor() {
    let mut graph = seeded_columnar();
    let dir = tempfile::tempdir().unwrap();
    graph.spill_dir = Some(dir.path().to_path_buf());
    graph.memory_limit = Some(0);

    graph.maybe_spill_columns();

    let master = graph.column_store("Item").expect("master store");
    assert!(
        master.is_mapped(),
        "precondition: the store must actually have spilled"
    );
    assert_eq!(
        master.spillable_heap_bytes(),
        0,
        "everything a spill can move is file-backed, so the trigger must now \
         read zero against its zero limit"
    );
    assert!(
        master.heap_bytes() > 0,
        "precondition: the unspillable floor is still there — without it this \
         test cannot distinguish convergence from an empty store"
    );
    assert!(
        !master.may_have_grown_spillable_heap(),
        "and a completed spill must clear the growth flag, or every later \
         statement re-walks every type to rediscover the floor"
    );
}

/// ...and the flag that buys that skip must not swallow a real spill.
///
/// A `SET` of a property the type has never carried appends a whole column of
/// heap behind the limit. That is the one write shape the growth flag has to
/// catch: break `append_column_typed`'s flag and this goes red (the fresh
/// column stays on the heap because nothing re-runs the pass), while
/// `the_spill_trigger_converges_on_the_unspillable_floor` above stays green.
#[test]
fn a_statement_that_grows_spillable_heap_still_triggers_the_spill() {
    let mut graph = seeded_columnar();
    let dir = tempfile::tempdir().unwrap();
    graph.spill_dir = Some(dir.path().to_path_buf());
    graph.memory_limit = Some(0);
    graph.maybe_spill_columns();
    assert_eq!(
        graph
            .column_store("Item")
            .expect("master store")
            .spillable_heap_bytes(),
        0,
        "precondition: the fixture starts converged"
    );

    run(&mut graph, "MATCH (n:Item) SET n.fresh = 7");

    let master = graph.column_store("Item").expect("master store");
    assert_eq!(
        master.spillable_heap_bytes(),
        0,
        "the appended `fresh` column is spillable heap over a zero limit; the \
         statement that created it must have re-run the spill pass"
    );
    assert_eq!(
        read_one(&graph, "MATCH (n:Item) WHERE n.id = 1 RETURN n.fresh"),
        Value::Int64(7),
        "and the value must survive the spill it triggered"
    );
}