haematite 0.7.0

//! Load-bearing correctness suite for the v2 byte-aware policy
//! (CHUNKING-POLICY.md §9).
//!
//! The centrepiece is the **windowed-mutation ≡ full-rebuild** property: a
//! localised mutation of an existing v2 tree must produce the byte-identical
//! root a fresh full rebuild of the mutated map would produce. This is the test
//! that fails if §2.1's both-edges locality argument is wrong — the highest-risk
//! deliverable. Around it sit the four named B1 regressions (the oversized-entry
//! window counterexamples), left-extension termination, the empty-result window,
//! min-2 termination under huge keys (collapse-all unreachable under v2), the
//! mode-boundary corpora, and the history-independence extension.

#![allow(clippy::unwrap_used)]
#![allow(clippy::panic)]

use std::collections::BTreeMap;

use proptest::prelude::*;

use super::mutate::batch_mutate;
use super::node::{Hash, LeafNode, Node};
use super::policy::TreePolicy;
use crate::store::MemoryStore;
use crate::tree::cursor::load_node;

/// Small byte targets so a few dozen entries make genuinely multi-leaf, multi
/// level trees. `leaf_target_bytes = 64` means an entry `16 + |k| + |v|` closes
/// a chunk with probability `entry_size/64`, and is oversized (a forced
/// singleton) once `|k| + |v| >= 48`.
const LEAF_T: u64 = 64;
const INTERNAL_T: u64 = 64;

fn v2() -> TreePolicy {
    TreePolicy::v2(LEAF_T, INTERNAL_T)
}

type Op = (Vec<u8>, Option<Vec<u8>>);

fn empty_tree() -> (MemoryStore, Hash) {
    let mut store = MemoryStore::new();
    let hash = store.put(&Node::Leaf(LeafNode::new(Vec::new()).unwrap()));
    (store, hash)
}

/// Build the canonical tree for `map` INTO `store` and return its root: a single
/// batch build from empty re-derives the whole tree, and `batch_mutate` is
/// history-independent, so this is THE canonical root for the logical map.
fn build_in(store: &mut MemoryStore, policy: TreePolicy, map: &BTreeMap<Vec<u8>, Vec<u8>>) -> Hash {
    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new()).unwrap()));
    let ops: Vec<Op> = map
        .iter()
        .map(|(k, v)| (k.clone(), Some(v.clone())))
        .collect();
    batch_mutate(store, empty, &ops, policy).unwrap()
}

/// The canonical root hash for `map`, built in a throwaway store — for pure hash
/// comparison against a windowed result (no node co-location needed).
fn canonical_hash(policy: TreePolicy, map: &BTreeMap<Vec<u8>, Vec<u8>>) -> Hash {
    let mut store = MemoryStore::new();
    build_in(&mut store, policy, map)
}

/// Read the whole logical map back by walking every leaf.
fn read_map(store: &MemoryStore, root: Hash) -> BTreeMap<Vec<u8>, Vec<u8>> {
    let mut map = BTreeMap::new();
    walk(store, root, &mut map);
    map
}

fn walk(store: &MemoryStore, hash: Hash, out: &mut BTreeMap<Vec<u8>, Vec<u8>>) {
    match &*load_node(store, hash).unwrap() {
        Node::Leaf(leaf) => {
            for (k, v) in leaf.entries() {
                out.insert(k.clone(), v.clone());
            }
        }
        Node::Internal(internal) => {
            for (_sep, child) in internal.children() {
                walk(store, *child, out);
            }
        }
    }
}

/// Height (internal levels above the leaves) and leaf count, for the structural
/// termination/shape assertions.
fn shape(store: &MemoryStore, root: Hash) -> (usize, usize) {
    fn go(store: &MemoryStore, hash: Hash, leaves: &mut usize) -> usize {
        match &*load_node(store, hash).unwrap() {
            Node::Leaf(_) => {
                *leaves += 1;
                0
            }
            Node::Internal(internal) => {
                let mut height = 0;
                for (_sep, child) in internal.children() {
                    height = go(store, *child, leaves);
                }
                height + 1
            }
        }
    }
    let mut leaves = 0;
    let height = go(store, root, &mut leaves);
    (height, leaves)
}

fn apply_logical(map: &BTreeMap<Vec<u8>, Vec<u8>>, batch: &[Op]) -> BTreeMap<Vec<u8>, Vec<u8>> {
    let mut out = map.clone();
    for (k, v) in batch {
        match v {
            Some(value) => {
                out.insert(k.clone(), value.clone());
            }
            None => {
                out.remove(k);
            }
        }
    }
    out
}

/// The core equivalence check reused by the property test and every regression:
/// applying `batch` to the tree built from `map` yields the byte-identical root
/// a fresh rebuild of the mutated map yields, AND reads back the right values.
fn assert_windowed_matches_rebuild(map: &BTreeMap<Vec<u8>, Vec<u8>>, batch: &[Op]) {
    let policy = v2();
    let mut store = MemoryStore::new();
    let tree_root = build_in(&mut store, policy, map);

    let windowed = batch_mutate(&mut store, tree_root, batch, policy).unwrap();

    let mutated = apply_logical(map, batch);
    let rebuilt = canonical_hash(policy, &mutated);

    assert_eq!(
        windowed, rebuilt,
        "windowed mutation diverged from full rebuild\nmap={map:?}\nbatch={batch:?}"
    );
    assert_eq!(
        read_map(&store, windowed),
        mutated,
        "windowed root read back the wrong logical map"
    );
}

// ---------------------------------------------------------------------------
// Generators
// ---------------------------------------------------------------------------

/// A value length spanning tiny, mid, and OVERSIZED-at-target (>= 48 payload)
/// so oversized singletons are generated regularly.
fn value_strategy() -> impl Strategy<Value = Vec<u8>> {
    proptest::collection::vec(any::<u8>(), 0..90)
}

fn key_strategy() -> impl Strategy<Value = Vec<u8>> {
    proptest::collection::vec(any::<u8>(), 1..6)
}

fn map_strategy() -> impl Strategy<Value = BTreeMap<Vec<u8>, Vec<u8>>> {
    proptest::collection::btree_map(key_strategy(), value_strategy(), 1..70)
}

fn batch_strategy(keys: Vec<Vec<u8>>) -> impl Strategy<Value = Vec<Op>> {
    // Mix mutations on EXISTING keys (updates/deletes — the window-moving cases)
    // with brand-new keys (inserts). Sorted+deduped by the driver, so order here
    // is irrelevant to the result.
    let existing = if keys.is_empty() {
        Just(Vec::<u8>::new()).boxed()
    } else {
        prop::sample::select(keys).boxed()
    };
    let op = prop_oneof![
        existing.prop_flat_map(|k| value_strategy().prop_map(move |v| (k.clone(), Some(v)))),
        key_strategy().prop_flat_map(|k| value_strategy().prop_map(move |v| (k.clone(), Some(v)))),
        key_strategy().prop_map(|k| (k, None)),
    ];
    proptest::collection::vec(op, 1..8)
}

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

    /// THE load-bearing proof: windowed mutation == full rebuild under v2, across
    /// random maps (including oversized singletons) and random localised batches
    /// whose keys are drawn from the tree's OWN keys (so updates/deletes actually
    /// move windows) mixed with fresh inserts.
    #[test]
    fn windowed_mutation_equals_full_rebuild(
        (map, batch) in map_strategy().prop_flat_map(|map| {
            let keys: Vec<Vec<u8>> = map.keys().cloned().collect();
            (Just(map), batch_strategy(keys))
        }),
    ) {
        assert_windowed_matches_rebuild(&map, &batch);
    }

    /// History independence under v2: the same final map reached by two different
    /// insertion orders (one-at-a-time) hashes identically, including oversized
    /// entries and threshold-adjacent sizes.
    #[test]
    fn v2_history_independence(map in map_strategy(), seed in any::<u64>()) {
        let policy = v2();
        let ops_sorted: Vec<Op> = map.iter().map(|(k, v)| (k.clone(), Some(v.clone()))).collect();
        let mut shuffled = ops_sorted.clone();
        shuffle(&mut shuffled, seed);

        let (mut s1, e1) = empty_tree();
        let mut r1 = e1;
        for op in &ops_sorted {
            r1 = batch_mutate(&mut s1, r1, std::slice::from_ref(op), policy).unwrap();
        }
        let (mut s2, e2) = empty_tree();
        let mut r2 = e2;
        for op in &shuffled {
            r2 = batch_mutate(&mut s2, r2, std::slice::from_ref(op), policy).unwrap();
        }
        prop_assert_eq!(r1, r2, "v2 root depends on insertion order");
    }
}

fn shuffle(ops: &mut [Op], seed: u64) {
    let mut state = seed.wrapping_mul(0x9e37_79b9_7f4a_7c15).wrapping_add(1);
    for i in (1..ops.len()).rev() {
        state = state
            .wrapping_mul(0x5851_f42d_4c95_7f2d)
            .wrapping_add(0x1405_7b7e_f767_814f);
        let j = (state >> 33) as usize % (i + 1);
        ops.swap(i, j);
    }
}

// ---------------------------------------------------------------------------
// The four named B1 regressions (§2.1 / §9) — the oversized-entry window
// counterexamples the r1 right-edge-only rule got wrong.
// ---------------------------------------------------------------------------

/// A value guaranteed oversized at `LEAF_T` (payload >= 48 ⇒ `entry_size` >= 64).
fn oversized_value() -> Vec<u8> {
    vec![0xab; 60]
}

fn small_map(pairs: &[(&[u8], Vec<u8>)]) -> BTreeMap<Vec<u8>, Vec<u8>> {
    pairs.iter().map(|(k, v)| (k.to_vec(), v.clone())).collect()
}

#[test]
fn b1_delete_oversized_first_merges_leftward() {
    // [a][X oversized][b c ...] — deleting X must let the canonical tree merge
    // its neighbours; the window's left edge rested on X's unstable boundary.
    let map = small_map(&[
        (b"a", vec![1]),
        (b"m", oversized_value()),
        (b"p", vec![2]),
        (b"q", vec![3]),
        (b"r", vec![4]),
    ]);
    assert_windowed_matches_rebuild(&map, &[(b"m".to_vec(), None)]);
}

#[test]
fn b1_shrink_oversized_first_below_target() {
    // Shrink the oversized entry below target: its boundary_before vanishes and
    // it must re-merge with its neighbours.
    let map = small_map(&[
        (b"a", vec![1]),
        (b"m", oversized_value()),
        (b"p", vec![2]),
        (b"q", vec![3]),
    ]);
    assert_windowed_matches_rebuild(&map, &[(b"m".to_vec(), Some(vec![9]))]);
}

#[test]
fn b1_insert_oversized_mid_window() {
    // Insert a NEW oversized entry in the middle of a window: it must split its
    // leaf into a stable singleton isolated from both neighbours.
    let map = small_map(&[
        (b"a", vec![1]),
        (b"b", vec![2]),
        (b"c", vec![3]),
        (b"d", vec![4]),
    ]);
    assert_windowed_matches_rebuild(&map, &[(b"bb".to_vec(), Some(oversized_value()))]);
}

#[test]
fn b1_size_flip_at_left_edge() {
    // Grow an entry at the window's left edge across the oversized threshold,
    // flipping its boundary bits — the left edge must extend to a stable edge.
    let map = small_map(&[
        (b"a", vec![1]),
        (b"b", vec![2]),
        (b"c", vec![3]),
        (b"d", vec![4]),
        (b"e", vec![5]),
    ]);
    assert_windowed_matches_rebuild(&map, &[(b"c".to_vec(), Some(oversized_value()))]);
}

#[test]
fn left_extension_terminates_and_empty_window_is_canonical() {
    // Delete EVERY entry of a multi-leaf tree in one batch: the window spans the
    // whole tree, left extension runs to the start, and the result is the empty
    // leaf — the empty-result window case.
    let map: BTreeMap<Vec<u8>, Vec<u8>> = (0u16..40)
        .map(|i| (i.to_be_bytes().to_vec(), vec![i as u8; 3]))
        .collect();
    let batch: Vec<Op> = map.keys().map(|k| (k.clone(), None)).collect();
    assert_windowed_matches_rebuild(&map, &batch);

    let policy = v2();
    let (mut store, empty) = empty_tree();
    let root = build_in(&mut store, policy, &map);
    let emptied = batch_mutate(&mut store, root, &batch, policy).unwrap();
    assert_eq!(emptied, canonical_hash(policy, &BTreeMap::new()));
    assert_eq!(
        emptied, empty,
        "emptied tree must equal the canonical empty leaf"
    );
}

// ---------------------------------------------------------------------------
// Min-2 internal grouping (§2.2): huge keys make EVERY internal separator a
// boundary, so the level is all-boundary; min-2 must still halve it (collapse
// all unreachable under v2), giving O(log M) height and a terminating build.
// ---------------------------------------------------------------------------

#[test]
fn min2_terminates_and_halves_under_huge_keys() {
    // 200-byte keys ⇒ internal separators 8+200+32 = 240 >> INTERNAL_T, so every
    // internal boundary fires. Under min-2 each level still halves.
    let policy = v2();
    let (mut store, empty) = empty_tree();
    let mut root = empty;
    let count = 300usize;
    for i in 0..count {
        let mut key = vec![0u8; 200];
        key[..8].copy_from_slice(&(i as u64).to_be_bytes());
        root = batch_mutate(&mut store, root, &[(key, Some(vec![1]))], policy).unwrap();
    }
    let (height, leaves) = shape(&store, root);
    // Every huge entry is its own oversized leaf, so leaves == count. With
    // unconditional halving the height is logarithmic, never linear.
    assert_eq!(
        leaves, count,
        "each huge entry should be an oversized singleton leaf"
    );
    let ceiling = (usize::BITS - (count - 1).leading_zeros()) as usize + 2;
    assert!(
        height <= ceiling,
        "min-2 height {height} exceeded the O(log M) ceiling {ceiling} for {count} leaves"
    );

    // And it reads back every key.
    assert_eq!(read_map(&store, root).len(), count);
}

#[test]
fn reduce_by_one_shape_stays_logarithmic() {
    // The r5 counterexample family: a persistent non-boundary separator ahead of
    // an all-boundary tail must NOT lose one entry per level (linear height).
    // Huge keys (all-boundary) plus a small-key prefix approximate the shape; the
    // assertion is the logarithmic height bound the uniform min-2 rule guarantees.
    let policy = v2();
    let (mut store, empty) = empty_tree();
    let mut root = empty;
    let mut expected = 0usize;
    // One small key first, then many huge (all-boundary) keys.
    root = batch_mutate(&mut store, root, &[(vec![0u8, 1], Some(vec![7]))], policy).unwrap();
    expected += 1;
    for i in 1..200u64 {
        let mut key = vec![0xffu8; 180];
        key[..8].copy_from_slice(&i.to_be_bytes());
        root = batch_mutate(&mut store, root, &[(key, Some(vec![1]))], policy).unwrap();
        expected += 1;
    }
    let (height, _leaves) = shape(&store, root);
    let ceiling = (usize::BITS - (expected - 1).leading_zeros()) as usize + 2;
    assert!(
        height <= ceiling,
        "reduce-by-one shape gave height {height} > O(log M) ceiling {ceiling}"
    );
    assert_eq!(read_map(&store, root).len(), expected);
}

// ---------------------------------------------------------------------------
// Mode-boundary corpora (§2.2 / §9): entry sizes at 0.49T / 0.50T / 0.51T /
// 0.75T of the LEAF target pin the near-target regime. Each corpus must build
// deterministically and windowed-mutate exactly like a rebuild.
// ---------------------------------------------------------------------------

fn corpus_at_fraction(
    numerator: u64,
    denominator: u64,
    count: usize,
) -> BTreeMap<Vec<u8>, Vec<u8>> {
    // entry_size target fraction: choose |value| so 16 + |k| + |v| ≈ frac*LEAF_T.
    let target_entry = (LEAF_T * numerator) / denominator;
    let value_len = target_entry.saturating_sub(16 + 4).max(1) as usize;
    (0u32..count as u32)
        .map(|i| (i.to_be_bytes().to_vec(), vec![i as u8; value_len]))
        .collect()
}

#[test]
fn mode_boundary_corpora_build_and_mutate_canonically() {
    for (num, den) in [(49u64, 100u64), (50, 100), (51, 100), (75, 100)] {
        let map = corpus_at_fraction(num, den, 40);
        // A representative localised mutation: overwrite one mid key, delete one,
        // insert one — all near-target sized.
        let mid: Vec<u8> = 20u32.to_be_bytes().to_vec();
        let del: Vec<u8> = 10u32.to_be_bytes().to_vec();
        let target_entry = (LEAF_T * num) / den;
        let vlen = target_entry.saturating_sub(20).max(1) as usize;
        let batch: Vec<Op> = vec![
            (mid, Some(vec![7u8; vlen])),
            (del, None),
            (b"new-key".to_vec(), Some(vec![3u8; vlen])),
        ];
        assert_windowed_matches_rebuild(&map, &batch);
    }
}

// ---------------------------------------------------------------------------
// Formula pins (§2.1 / §9) — threshold equality, oversized isolation, and
// native/wasm-portable u128 arithmetic (the compare is target-neutral by
// construction: BLAKE3 + u128, no platform types).
// ---------------------------------------------------------------------------

/// True when some leaf under `root` is a singleton holding exactly `target`.
fn contains_singleton_leaf(store: &MemoryStore, hash: Hash, target: &[u8]) -> bool {
    match &*load_node(store, hash).unwrap() {
        Node::Leaf(leaf) => leaf.entries().len() == 1 && leaf.entries()[0].0 == target,
        Node::Internal(internal) => internal
            .children()
            .iter()
            .any(|(_s, c)| contains_singleton_leaf(store, *c, target)),
    }
}

#[test]
fn threshold_equality_forces_singleton() {
    // entry_size exactly == target ⇒ boundary_after AND boundary_before both fire.
    let policy = TreePolicy::v2(20, 64); // 16 + 4-byte key + 0 value == 20
    assert!(policy.leaf_boundary_after(b"abcd", 0));
    assert!(policy.leaf_boundary_before(4, 0));

    // In a real tree the equal-size entry is isolated as its own leaf.
    let map = small_map(&[(b"aa", vec![]), (b"abcd", vec![]), (b"zz", vec![])]);
    let (mut store, _e) = empty_tree();
    let root = build_in(&mut store, policy, &map);
    assert!(
        contains_singleton_leaf(&store, root, b"abcd"),
        "the at-target entry must be an isolated singleton leaf"
    );
}

#[test]
fn oversized_isolation_is_total() {
    // A single oversized value among small entries always sits alone, and its
    // neighbours never rewrite it (windowed mutation of a neighbour keeps its
    // hash).
    let map = small_map(&[
        (b"a", vec![1]),
        (b"big", oversized_value()),
        (b"c", vec![3]),
    ]);
    let policy = v2();
    let (mut store, _e) = empty_tree();
    let root = build_in(&mut store, policy, &map);

    // The leaf holding "big" before the neighbour edit.
    let before = read_map(&store, root);
    // Mutate a neighbour only.
    let after_root =
        batch_mutate(&mut store, root, &[(b"a".to_vec(), Some(vec![2]))], policy).unwrap();
    let after = read_map(&store, after_root);
    assert_eq!(before.get(b"big".as_slice()), after.get(b"big".as_slice()));
    // And the whole thing still matches a rebuild.
    assert_windowed_matches_rebuild(&map, &[(b"a".to_vec(), Some(vec![2]))]);
}