topodb 0.0.13

use proptest::prelude::*;
use std::collections::{BTreeMap, HashSet};
use topodb::*;

/// Fixed, distinct vocabulary for `Intent::SetText` — deterministic (no
/// random strings), so BM25 postings are reproducible across replay.
const WORDS: [&str; 8] = [
    "alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf", "hotel",
];

/// Generate a small random-but-valid op sequence: create nodes, then a mix
/// of edges/prop-sets/closes/removes/text-sets referencing only created ids.
/// Abstract intents, lowered to concrete Ops inside the test where real ids
/// exist. Indices are taken modulo the live collection length, so every
/// generated script is valid by construction.
#[derive(Debug, Clone)]
enum Intent {
    Edge {
        from_ix: usize,
        to_ix: usize,
    },
    Close {
        edge_ix: usize,
    },
    SetProp {
        node_ix: usize,
        val: i64,
    },
    /// Sets the declared text prop (`"text"`) to a fixed word from `WORDS`
    /// (chosen via `word_ix % WORDS.len()`), exercising the FTS postings
    /// maintenance path under replay.
    SetText {
        node_ix: usize,
        word_ix: usize,
    },
    Embed {
        node_ix: usize,
    },
    Remove {
        node_ix: usize,
    },
}

fn scripts() -> impl Strategy<Value = (usize, usize, Vec<Intent>)> {
    let intent = prop_oneof![
        1 => (any::<usize>(), any::<usize>()).prop_map(|(f, t)| Intent::Edge {
            from_ix: f,
            to_ix: t
        }),
        1 => any::<usize>().prop_map(|i| Intent::Close { edge_ix: i }),
        1 => (any::<usize>(), any::<i64>()).prop_map(|(i, v)| Intent::SetProp { node_ix: i, val: v }),
        1 => (any::<usize>(), any::<usize>()).prop_map(|(i, w)| Intent::SetText {
            node_ix: i,
            word_ix: w
        }),
        // Embed is weighted 4x: with hnsw build_threshold 4 in the property
        // below, a built graph needs >= 4 DISTINCT embedded nodes in one
        // scoped cluster — at uniform 1/6 weight only a handful of the 64
        // cases ever crossed it, leaving the hnsw-table parity assertions
        // exercising mostly-unbuilt clusters. 4x makes built-graph replay
        // (insert/tombstone/inline-rebuild) a routine part of the sample.
        4 => any::<usize>().prop_map(|i| Intent::Embed { node_ix: i }),
        1 => any::<usize>().prop_map(|i| Intent::Remove { node_ix: i }),
    ];
    (
        3usize..10,
        0usize..4,
        proptest::collection::vec(intent, 0..20),
    )
}

/// Lowers the abstract script into concrete `submit_at` calls against `db`.
///
/// - Batch 0 (t=0): creates `n_scoped` nodes in one scope plus `n_shared`
///   nodes in `Scope::Shared`.
/// - Each subsequent intent becomes at most one `submit_at` at
///   `t = 1 + intent index`.
/// - `Edge`: endpoints chosen via `ix % nodes.len()`. Skipped (nothing
///   submitted) if it would be a self-loop or would violate the cross-scope
///   rule (mismatched scopes with neither endpoint `Shared`) — those would
///   be rejected anyway, but we skip up front rather than relying on
///   tolerated rejection, per the brief.
/// - `Close`: edge chosen via `ix % edges.len()`, skipped if there are no
///   edges yet. If the chosen edge happens to already be closed (or was
///   removed via a cascading `RemoveNode`), we still submit — the resulting
///   `Rejected` is tolerated and ignored, since a rejected batch appends
///   nothing to the log and so cannot affect replay determinism. This is a
///   deliberate choice (over skipping) so the script generator continues to
///   exercise the "close an already-closed/missing edge" rejection path
///   inside `apply_batch` itself, not just at replay.
/// - `SetProp`: sets `props = {"v": Some(Int(val))}` on a live node chosen
///   via `ix % nodes.len()`.
/// - `Remove`: removes a live node chosen via `ix % nodes.len()`, skipped if
///   it would empty the local node list (so every later `ix % nodes.len()`
///   stays well-defined, and `RemoveNode` never targets an already-missing
///   node).
fn run_script(db: &Db, n_scoped: usize, n_shared: usize, intents: &[Intent]) -> ScopeId {
    let scope_id = ScopeId::new();
    let scope = Scope::Id(scope_id);

    let mut nodes: Vec<(NodeId, Scope)> = Vec::new();
    let mut create_ops = Vec::new();
    for _ in 0..n_scoped {
        let id = NodeId::new();
        nodes.push((id, scope));
        create_ops.push(Op::CreateNode {
            id,
            scope,
            // "M" (not "N"): must match the `open_with` spec's declared
            // equality/text index label so `SetProp`/`SetText` actually land
            // in the indexes under test.
            label: "M".into(),
            props: Default::default(),
        });
    }
    for _ in 0..n_shared {
        let id = NodeId::new();
        nodes.push((id, Scope::Shared));
        create_ops.push(Op::CreateNode {
            id,
            scope: Scope::Shared,
            label: "M".into(),
            props: Default::default(),
        });
    }
    db.submit_at(create_ops, 0).unwrap();

    let mut edges: Vec<EdgeId> = Vec::new();

    for (i, intent) in intents.iter().enumerate() {
        let t = 1 + i as i64;
        match *intent {
            Intent::Edge { from_ix, to_ix } => {
                if nodes.is_empty() {
                    continue;
                }
                let (from_id, from_scope) = nodes[from_ix % nodes.len()];
                let (to_id, to_scope) = nodes[to_ix % nodes.len()];
                if from_id == to_id {
                    continue; // self-loop
                }
                let cross_scope_violation = from_scope != to_scope
                    && from_scope != Scope::Shared
                    && to_scope != Scope::Shared;
                if cross_scope_violation {
                    continue;
                }
                let id = EdgeId::new();
                db.submit_at(
                    vec![Op::CreateEdge {
                        id,
                        scope: from_scope,
                        ty: "REL".into(),
                        from: from_id,
                        to: to_id,
                        props: Default::default(),
                        valid_from: None,
                    }],
                    t,
                )
                .unwrap();
                edges.push(id);
            }
            Intent::Close { edge_ix } => {
                if edges.is_empty() {
                    continue;
                }
                let id = edges[edge_ix % edges.len()];
                // Tolerated: already-closed (or cascaded-away) edges yield
                // `Rejected`, which appends nothing — harmless for replay.
                let _ = db.submit_at(vec![Op::CloseEdge { id, valid_to: None }], t);
            }
            Intent::SetProp { node_ix, val } => {
                if nodes.is_empty() {
                    continue;
                }
                let (id, _) = nodes[node_ix % nodes.len()];
                let mut props: BTreeMap<String, Option<PropValue>> = BTreeMap::new();
                props.insert("v".to_string(), Some(PropValue::Int(val)));
                db.submit_at(vec![Op::SetNodeProps { id, props }], t)
                    .unwrap();
            }
            Intent::SetText { node_ix, word_ix } => {
                if nodes.is_empty() {
                    continue;
                }
                let (id, _) = nodes[node_ix % nodes.len()];
                let mut props: BTreeMap<String, Option<PropValue>> = BTreeMap::new();
                props.insert(
                    "text".to_string(),
                    Some(PropValue::Str(WORDS[word_ix % WORDS.len()].into())),
                );
                db.submit_at(vec![Op::SetNodeProps { id, props }], t)
                    .unwrap();
            }
            Intent::Embed { node_ix } => {
                if nodes.is_empty() {
                    continue;
                }
                let (id, _) = nodes[node_ix % nodes.len()];
                // Deterministic payload derived from the index so replay
                // reproduces the same embedding; exercises the `SetEmbedding`
                // arm of both `apply_op` and `Snapshot::apply` under replay.
                db.submit_at(
                    vec![Op::SetEmbedding {
                        id,
                        model: "m".into(),
                        vector: vec![node_ix as f32],
                    }],
                    t,
                )
                .unwrap();
            }
            Intent::Remove { node_ix } => {
                if nodes.len() <= 1 {
                    continue; // would empty the live node list
                }
                let ix = node_ix % nodes.len();
                let (id, _) = nodes[ix];
                db.submit_at(vec![Op::RemoveNode { id }], t).unwrap();
                nodes.remove(ix);
            }
        }
    }
    scope_id
}

/// Every seed node's full 1-hop neighborhood (both directions, unfiltered by
/// edge type, `as_of` pinned so the check is insensitive to wall-clock time),
/// as observed through the public `traverse` read path — which reads
/// OUT_ADJ/IN_ADJ directly. Keyed by seed node id in a `BTreeMap` for
/// order-independent comparison; each seed's node/edge id sets are sorted for
/// the same reason. Used to compare adjacency parity before vs. after
/// `rebuild_state_from_ops` — the disk-resident equivalent of the old
/// `Snapshot::debug_out`/`debug_inn` comparison, from before the in-memory
/// snapshot layer was deleted (the engine's read model is disk-resident now,
/// so there is no separate adjacency structure to fold/compare directly —
/// `traverse` IS the read path being guarded).
fn adjacency_fingerprint(
    db: &Db,
    scopes: &ScopeSet,
    seeds: &[NodeId],
) -> BTreeMap<NodeId, (Vec<NodeId>, Vec<EdgeId>)> {
    let mut out = BTreeMap::new();
    for &seed in seeds {
        let sub = db
            .traverse(&TraversalQuery {
                scopes: scopes.clone(),
                seeds: vec![seed],
                max_hops: 1,
                edge_types: None,
                direction: Direction::Both,
                as_of: Some(i64::MAX),
            })
            .unwrap();
        let mut node_ids: Vec<NodeId> = sub.nodes.iter().map(|n| n.id).collect();
        node_ids.sort();
        let mut edge_ids: Vec<EdgeId> = sub.edges.iter().map(|e| e.id).collect();
        edge_ids.sort();
        out.insert(seed, (node_ids, edge_ids));
    }
    out
}

/// The `IndexSpec` under which the proptest `Db` is opened: equality-indexes
/// `("M", "v")` (fed by `Intent::SetProp`) and text-indexes `("M", "text")`
/// (fed by `Intent::SetText`) — the two indexed recall paths this test guards
/// (`nodes_by_prop`, `search_text`), in addition to vector search (which
/// needs no spec declaration).
fn spec() -> IndexSpec {
    IndexSpec {
        equality: vec![PropIndex {
            label: "M".into(),
            prop: "v".into(),
        }],
        text: vec![PropIndex {
            label: "M".into(),
            prop: "text".into(),
        }],
    }
}

/// Equality-index and vector-search parity, checked against whatever the
/// current live state is (called both before and after
/// `rebuild_state_from_ops` — the caller supplies the same `db`/`scopes`
/// each time, so a rebuild that silently drops `prop_index` entries or
/// vector slab rows fails the *second* call even though the *first* passed).
///
/// Falsifiable: comment out `apply_op`'s `prop_index::index_node`/
/// `unindex_node` calls, or its `vector_store::put_vector`/`remove_vector`
/// calls (`storage.rs`) — both of which `rebuild_state_from_ops` drives via
/// the SAME `apply_op` replayed per logged op, there being no separate
/// snapshot-seeding step to disable — and the post-rebuild call here goes
/// from "finds it" to "doesn't" while the pre-rebuild call still passes —
/// exactly the drop this guards.
fn assert_equality_and_vector_parity(db: &Db, scopes: &ScopeSet) {
    let dump = db.debug_dump_nodes();

    // --- Equality-index parity for the declared ("M", "v") key. ---
    let mut present_v: HashSet<i64> = HashSet::new();
    for n in &dump {
        if let Some(PropValue::Int(v)) = n.props.get("v") {
            present_v.insert(*v);
            let hits = db
                .nodes_by_prop(scopes, "M", "v", &PropValue::Int(*v))
                .unwrap();
            assert!(
                hits.iter().any(|h| h.id == n.id),
                "nodes_by_prop(\"M\",\"v\",{v}) must find node {:?}",
                n.id
            );
        }
    }
    // A value no live node carries must yield nothing. `present_v` is small
    // (bounded by the script length), so a short linear probe from 0 always
    // terminates fast in practice.
    let mut absent = 0i64;
    while present_v.contains(&absent) {
        absent = absent.wrapping_add(1);
    }
    let empty_hits = db
        .nodes_by_prop(scopes, "M", "v", &PropValue::Int(absent))
        .unwrap();
    assert!(
        empty_hits.is_empty(),
        "nodes_by_prop must find nothing for unused value {absent}"
    );

    // --- Vector parity: every embedded node's own vector must retrieve it. ---
    // k = live node count is an upper bound on any tie group (see below), so
    // asking for the whole node count guarantees the self-match can't be
    // pushed out of the top-k by ties.
    let k = dump.len().max(1);
    for n in &dump {
        if let Some((model, vector)) = &n.embedding {
            // Skip exactly-zero vectors: `vector.rs::cosine` returns `None`
            // for a zero-norm operand by design (cosine similarity is
            // undefined there), so a zero vector never scores — even against
            // itself. That is correct `search_vector` behavior, not a
            // rebuild bug, so asserting self-retrieval here would be a false
            // failure unrelated to what this test is guarding.
            if vector.iter().all(|x| *x == 0.0) {
                continue;
            }
            let hits = db
                .search_vector(&VectorQuery {
                    scopes: scopes.clone(),
                    model: model.clone(),
                    vector: vector.clone(),
                    k,
                    candidates: None,
                })
                .unwrap();
            assert!(
                hits.iter().any(|(rec, _)| rec.id == n.id),
                "search_vector must find node {:?} via its own embedding under model {model:?}",
                n.id
            );
        }
    }
}

/// Sorted result-id set for `search_text(scopes, word, k)` — used to compare
/// FTS parity before vs. after rebuild (postings must reindex to the exact
/// same set, not merely a non-empty one).
fn fts_hit_ids(db: &Db, scopes: &ScopeSet, word: &str, k: usize) -> Vec<NodeId> {
    let mut ids: Vec<NodeId> = db
        .search_text(scopes, word, k)
        .unwrap()
        .into_iter()
        .map(|(n, _)| n.id)
        .collect();
    ids.sort();
    ids
}

proptest! {
    #![proptest_config(ProptestConfig::with_cases(64))]
    #[test]
    fn state_from_replay_equals_state_from_execution(script in scripts()) {
        let (n_scoped, n_shared, intents) = script;
        let dir = tempfile::tempdir().unwrap();
        // `build_threshold: 4` (default is 1024) so the 3-10-node scripts
        // generated here genuinely cross the HNSW build/tombstone/rebuild
        // thresholds in many cases, rather than staying perpetually
        // brute-force-only — see the module doc for the zero-norm/threshold
        // interaction this has with `Intent::Embed`'s 1-dim vector recipe.
        let options = DbOptions {
            hnsw_params: Some(HnswParams {
                build_threshold: 4,
                ..Default::default()
            }),
            ..Default::default()
        };
        let db = Db::open_with_options(dir.path().join("t.redb"), spec(), options).unwrap();
        let scope_id = run_script(&db, n_scoped, n_shared, &intents);
        // Covers both script scopes: the one generated scope plus Shared —
        // every read path under test (`nodes_by_prop`, `search_text`,
        // `search_vector`) is scoped, so this must see everything the script
        // created.
        let scopes = ScopeSet::of(&[scope_id]).with_shared();

        let live_nodes = db.debug_dump_nodes();
        let live_edges = db.debug_dump_edges();
        let seeds: Vec<NodeId> = live_nodes.iter().map(|n| n.id).collect();
        // Captured through the public disk-backed read path (`traverse`) —
        // the engine's read model is disk-resident, so there is no separate
        // in-memory snapshot for a reader to observe instead of storage.
        let adj_before = adjacency_fingerprint(&db, &scopes, &seeds);
        // Raw OUT_ADJ/IN_ADJ chunk contents, open AND closed entries. The
        // traverse fingerprint above cannot observe closed edges (its
        // temporal filter excludes every entry with `valid_to = Some(_)` at
        // any `as_of`), so on its own it would miss a rebuild bug that
        // corrupts a closed edge's `valid_to` — or drops its adjacency entry
        // outright — while leaving the separate EDGES-table copy (checked
        // via `debug_dump_edges` below) intact. This raw dump is the
        // disk-resident equivalent of the old `Snapshot::debug_out`/
        // `debug_inn` comparison, closed entries included.
        let adj_raw_before = db.debug_dump_adjacency().unwrap();

        // Raw v4-table dumps, mirroring `adj_raw_before` above: entry-for-
        // entry byte parity for the recall-layer tables `rebuild_state_from_ops`
        // drains and repopulates, not just the "finds it" parity
        // `assert_equality_and_vector_parity`/`fts_hit_ids` check below. A
        // rebuild that reproduced the same *search results* while silently
        // reordering postings chunks, dropping a stale `embedding_ref`
        // pointer, or losing a `vector_dims` pin would sail through the
        // "finds it" checks but fail here.
        let postings_raw_before = db.debug_dump_postings().unwrap();
        let vectors_raw_before = db.debug_dump_vectors().unwrap();
        let embedding_ref_raw_before = db.debug_dump_embedding_ref().unwrap();
        let vector_dims_raw_before = db.debug_dump_vector_dims().unwrap();
        // LABEL_INDEX (F9-11 Task 7): derived purely from NODES, so this
        // dump's before/after parity below is the raw-table-level proof (to
        // go with the `nodes_by_label`-based explicit rebuild test) that
        // `rebuild_state_from_ops` repopulates it byte-for-byte, not just
        // "close enough" to keep query results looking right.
        let label_index_raw_before = db.debug_dump_label_index().unwrap();
        // HNSW cluster tables (F8 Task 6): entry-for-entry parity, the same
        // load-bearing check as the other raw v4/v7 dumps above — a rebuild
        // that reproduced identical `search_vector` results while silently
        // reordering HNSW_LINKS neighbor lists, dropping a tombstone flag, or
        // losing a cluster's HNSW_META entry-point/level/`stale` bookkeeping
        // would sail through the "finds it" checks but fail here.
        let hnsw_meta_raw_before = db.debug_dump_hnsw_meta().unwrap();
        let hnsw_links_raw_before = db.debug_dump_hnsw_links().unwrap();

        // Recall-layer parity, BEFORE rebuild. `text_k` is computed from the
        // pre-rebuild live count (rebuild never changes the live set — that's
        // exactly what `live_nodes == db.debug_dump_nodes()` below asserts —
        // so the same bound stays valid for the post-rebuild calls too).
        let text_k = live_nodes.len().max(1);
        assert_equality_and_vector_parity(&db, &scopes);
        let fts_before: Vec<Vec<NodeId>> =
            WORDS.iter().map(|w| fts_hit_ids(&db, &scopes, w, text_k)).collect();

        db.rebuild_state_from_ops().unwrap();

        prop_assert_eq!(live_nodes, db.debug_dump_nodes());
        prop_assert_eq!(live_edges, db.debug_dump_edges());

        // `rebuild_state_from_ops` drains and repopulates NODES/EDGES/
        // OUT_ADJ/IN_ADJ from the op log in one write transaction. Two
        // adjacency parity checks against the pre-rebuild captures:
        //
        // 1. The raw OUT_ADJ/IN_ADJ dump must be entry-for-entry identical —
        //    including CLOSED entries and their exact `valid_to` — pinning
        //    that replay reproduces byte-equal adjacency content. This is
        //    the load-bearing check (the old `debug_out`/`debug_inn`
        //    comparison's true equivalent); a rebuild that mangled a closed
        //    edge's `valid_to` in the chunks fails HERE and nowhere else.
        // 2. The `traverse`-based fingerprint must also be unchanged —
        //    cheap, and exercises the public read path over the same tables
        //    (open entries only, by `traverse`'s temporal contract).
        //
        // There is no more separate snapshot to check swap identity on: the
        // read model IS storage now, so "did the rebuild repopulate the
        // adjacency tables correctly" is exactly what these content
        // comparisons (with the node/edge dumps above) verify.
        let adj_raw_after = db.debug_dump_adjacency().unwrap();
        prop_assert_eq!(adj_raw_before, adj_raw_after);
        let adj_after = adjacency_fingerprint(&db, &scopes, &seeds);
        prop_assert_eq!(adj_before, adj_after);

        // Raw v4-table parity, AFTER rebuild — entry-for-entry equality
        // against the pre-rebuild captures above, the same load-bearing
        // check `adj_raw_before`/`adj_raw_after` performs for adjacency: a
        // rebuild that mangled a POSTINGS chunk, a VECTORS row, an
        // EMBEDDING_REF pointer, or a VECTOR_DIMS pin fails HERE even if
        // every "finds it" query below still happens to pass.
        let postings_raw_after = db.debug_dump_postings().unwrap();
        prop_assert_eq!(postings_raw_before, postings_raw_after);
        let vectors_raw_after = db.debug_dump_vectors().unwrap();
        prop_assert_eq!(vectors_raw_before, vectors_raw_after);
        let embedding_ref_raw_after = db.debug_dump_embedding_ref().unwrap();
        prop_assert_eq!(embedding_ref_raw_before, embedding_ref_raw_after);
        let vector_dims_raw_after = db.debug_dump_vector_dims().unwrap();
        prop_assert_eq!(vector_dims_raw_before, vector_dims_raw_after);
        let label_index_raw_after = db.debug_dump_label_index().unwrap();
        prop_assert_eq!(label_index_raw_before, label_index_raw_after);
        let hnsw_meta_raw_after = db.debug_dump_hnsw_meta().unwrap();
        prop_assert_eq!(hnsw_meta_raw_before, hnsw_meta_raw_after);
        let hnsw_links_raw_after = db.debug_dump_hnsw_links().unwrap();
        prop_assert_eq!(hnsw_links_raw_before, hnsw_links_raw_after);

        // Recall-layer parity, AFTER rebuild. Equality/vector re-assert the
        // same "finds it" property against the rebuilt state; FTS asserts the
        // *exact same* result-id set as before rebuild — this is the one
        // group compared by equality rather than just re-affirmed, since the
        // brief calls for identical sets, not merely non-empty ones.
        assert_equality_and_vector_parity(&db, &scopes);
        let fts_after: Vec<Vec<NodeId>> =
            WORDS.iter().map(|w| fts_hit_ids(&db, &scopes, w, text_k)).collect();
        prop_assert_eq!(fts_before, fts_after);
    }
}

/// F9-11 Task 7: `nodes_by_label` results (still a full `NODES` scan on this
/// branch — Task 8 rewires the read path onto `LABEL_INDEX`) must be
/// identical before and after `rebuild_state_from_ops`, over a corpus that
/// mixes two labels, two scopes, and a create-then-remove that leaves one
/// node dead — proving `LABEL_INDEX` (via `debug_dump_label_index`'s exact
/// byte parity above) AND the still-full-scan read path agree, rather than
/// merely proving the two never diverge by construction. Falsifiable: skip
/// `apply_op`'s `label_index.insert`/`.remove` calls, or drop `label_index`
/// from the tables `rebuild_state_from_ops` clears/repopulates, and this
/// still passes (nothing here reads `LABEL_INDEX` directly) — the
/// `label_index_raw_before`/`label_index_raw_after` proptest assertion above
/// is what catches that. This test instead pins the observable contract:
/// whatever `LABEL_INDEX` is FOR, the plain node-scan read it will one day
/// replace must keep agreeing with it.
#[test]
fn label_reads_are_identical_before_and_after_rebuild() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open_with(dir.path().join("t.redb"), spec()).unwrap();

    let scope_a = ScopeId::new();
    let scope_b = ScopeId::new();
    let (e1, e2, e3, m1, doomed) = (
        NodeId::new(),
        NodeId::new(),
        NodeId::new(),
        NodeId::new(),
        NodeId::new(),
    );
    db.submit(vec![
        Op::CreateNode {
            id: e1,
            scope: Scope::Id(scope_a),
            label: "Entity".into(),
            props: Default::default(),
        },
        Op::CreateNode {
            id: e2,
            scope: Scope::Id(scope_a),
            label: "Entity".into(),
            props: Default::default(),
        },
        Op::CreateNode {
            id: e3,
            scope: Scope::Id(scope_b),
            label: "Entity".into(),
            props: Default::default(),
        },
        Op::CreateNode {
            id: m1,
            scope: Scope::Id(scope_a),
            label: "M".into(),
            props: Default::default(),
        },
        Op::CreateNode {
            id: doomed,
            scope: Scope::Id(scope_a),
            label: "Entity".into(),
            props: Default::default(),
        },
    ])
    .unwrap();
    db.submit(vec![Op::RemoveNode { id: doomed }]).unwrap();

    let scopes = ScopeSet::of(&[scope_a, scope_b]);
    let mut before_entity: Vec<NodeId> = db
        .nodes_by_label(&scopes, "Entity")
        .iter()
        .map(|n| n.id)
        .collect();
    let mut before_m: Vec<NodeId> = db
        .nodes_by_label(&scopes, "M")
        .iter()
        .map(|n| n.id)
        .collect();
    before_entity.sort();
    before_m.sort();
    // Sanity on the corpus itself: `doomed` must not survive as an "Entity" hit.
    assert_eq!(before_entity, {
        let mut v = vec![e1, e2, e3];
        v.sort();
        v
    });
    assert_eq!(before_m, vec![m1]);

    db.rebuild_state_from_ops().unwrap();

    let mut after_entity: Vec<NodeId> = db
        .nodes_by_label(&scopes, "Entity")
        .iter()
        .map(|n| n.id)
        .collect();
    let mut after_m: Vec<NodeId> = db
        .nodes_by_label(&scopes, "M")
        .iter()
        .map(|n| n.id)
        .collect();
    after_entity.sort();
    after_m.sort();
    assert_eq!(
        before_entity, after_entity,
        "Entity hits must survive rebuild unchanged"
    );
    assert_eq!(before_m, after_m, "M hits must survive rebuild unchanged");
}

/// F8 Task 6: HNSW's observable contract — `search_vector` results, not just
/// the raw `HNSW_META`/`HNSW_LINKS` byte parity the proptest above pins —
/// must be identical before and after `rebuild_state_from_ops`, over a corpus
/// large enough (12 embeddings at `build_threshold: 4`) to actually cross the
/// build threshold, followed by two `RemoveNode`s (tombstoning graph entries,
/// and — depending on how far under the ratio that pushes the live count —
/// possibly triggering an in-band rebuild) before the replay-rebuild under
/// test. Companion to `label_reads_are_identical_before_and_after_rebuild`:
/// that test pins the read-path/index-table agreement for LABEL_INDEX, this
/// one pins it for the HNSW cluster tables. Multi-dimensional vectors are
/// fine here (unlike the proptest's frozen `vec![node_ix as f32]`
/// single-dimension vocabulary in `Intent::Embed` above) since this test owns
/// its own embedding recipe independently.
#[test]
fn search_vector_is_identical_before_and_after_rebuild() {
    let dir = tempfile::tempdir().unwrap();
    let options = DbOptions {
        hnsw_params: Some(HnswParams {
            build_threshold: 4,
            ..Default::default()
        }),
        ..Default::default()
    };
    let db = Db::open_with_options(dir.path().join("t.redb"), spec(), options).unwrap();

    let scope_id = ScopeId::new();
    let scope = Scope::Id(scope_id);
    let ids: Vec<NodeId> = (0..12).map(|_| NodeId::new()).collect();
    let create_ops: Vec<Op> = ids
        .iter()
        .map(|&id| Op::CreateNode {
            id,
            scope,
            label: "M".into(),
            props: Default::default(),
        })
        .collect();
    db.submit(create_ops).unwrap();

    // Deterministic 3-dim vectors, one non-zero component walked across the
    // corpus so every node's embedding is distinct and none is zero-norm.
    for (i, &id) in ids.iter().enumerate() {
        let vector = vec![(i as f32) + 1.0, ((i * 3) % 7) as f32, (i % 5) as f32];
        db.submit(vec![Op::SetEmbedding {
            id,
            model: "m".into(),
            vector,
        }])
        .unwrap();
    }

    let scopes = ScopeSet::of(&[scope_id]).with_shared();
    let queries: Vec<Vec<f32>> = vec![
        vec![1.0, 0.0, 0.0],
        vec![5.0, 2.0, 4.0],
        vec![10.0, 1.0, 3.0],
    ];
    let k = ids.len();

    // Two removals: tombstones graph entries, and — depending on live count
    // vs. `rebuild_num`/`rebuild_den` (3/10 default) against `build_threshold`
    // (4) — may cross the ratio and trigger an in-band rebuild too.
    db.submit(vec![Op::RemoveNode { id: ids[2] }]).unwrap();
    db.submit(vec![Op::RemoveNode { id: ids[7] }]).unwrap();

    let search_before_removal_rebuild: Vec<Vec<(NodeId, f32)>> = queries
        .iter()
        .map(|q| {
            db.search_vector(&VectorQuery {
                scopes: scopes.clone(),
                model: "m".into(),
                vector: q.clone(),
                k,
                candidates: None,
            })
            .unwrap()
            .into_iter()
            .map(|(rec, score)| (rec.id, score))
            .collect()
        })
        .collect();
    let hnsw_meta_before = db.debug_dump_hnsw_meta().unwrap();
    let hnsw_links_before = db.debug_dump_hnsw_links().unwrap();

    db.rebuild_state_from_ops().unwrap();

    let search_after: Vec<Vec<(NodeId, f32)>> = queries
        .iter()
        .map(|q| {
            db.search_vector(&VectorQuery {
                scopes: scopes.clone(),
                model: "m".into(),
                vector: q.clone(),
                k,
                candidates: None,
            })
            .unwrap()
            .into_iter()
            .map(|(rec, score)| (rec.id, score))
            .collect()
        })
        .collect();
    let hnsw_meta_after = db.debug_dump_hnsw_meta().unwrap();
    let hnsw_links_after = db.debug_dump_hnsw_links().unwrap();

    assert_eq!(
        search_before_removal_rebuild, search_after,
        "search_vector results must survive rebuild_state_from_ops unchanged"
    );
    assert_eq!(
        hnsw_meta_before, hnsw_meta_after,
        "HNSW_META must be byte-identical before/after rebuild_state_from_ops"
    );
    assert_eq!(
        hnsw_links_before, hnsw_links_after,
        "HNSW_LINKS must be byte-identical before/after rebuild_state_from_ops"
    );
}