haematite 0.7.0

//! §9 DISPATCH PINS — the test that would have caught the mixed-policy gap
//! (CHUNKING-POLICY.md §4.1, the r1 B3 finding; the Fable key-turn MUST-CLOSE).
//!
//! Every PRODUCTION `batch_mutate*` caller is driven here under a v2 policy and
//! asserted to produce v2-shaped output — the shape a hard-coded
//! `TreePolicy::V1_DEFAULT` at the call site could NOT produce. The
//! compile-time half of the guarantee (no policy-less mutation path exists) is
//! carried by the build itself: `active_detector()` is deleted and every entry
//! point takes an explicit `policy`, so this file only has to prove the RUNTIME
//! half — that each caller actually threads the v2 policy through to the tree.
//!
//! Callers covered (§4.1 inventory, all 8):
//! 1. public tree APIs (`batch_mutate` / `insert`)           — [`public_tree_api_*`]
//! 2. `WalBuffer::commit`                                     — [`wal_buffer_commit_*`]
//! 3. branch commit (`commit_branch`)                        — [`branch_commit_*`]
//! 4. branch merge (`merge_with_report`)                     — [`branch_merge_*`]
//! 5. sync merge (`merge_synced_roots`)                      — [`sync_merge_*`]
//! 6. sync handoff-merge (`merge_committed_union`)           — [`handoff_merge_*`]
//! 7. actor commit                                           — pinned by the
//!    Phase-2 stamp matrix (`db/config_stamp_tests.rs`: fresh create = v2, and
//!    `shard/actor.rs` mutates under `self.policy`); the mixed-policy pin below
//!    proves path-independence of the shape the actor emits.
//! 8. `browser_local` commit (wasm)                          — pinned by parity
//!    (`browser_local/codec.rs::decode_store_format` mirrors native
//!    `derive_policy`, itself pinned in `config_stamp_tests.rs`); the
//!    browser-runtime end-to-end pin needs the wasm harness (noted in the leg
//!    report — `haematite-wasm-probes`).

use std::error::Error;

use crate::branch::conflict::ConflictPolicy;
use crate::branch::handle::DEFAULT_SHARD_ID;
use crate::branch::lifecycle::create_branch;
use crate::branch::merge::merge_with_report;
use crate::branch::refstore::BranchRefStore;
use crate::branch::registry::BranchRegistry;
use crate::branch::{CommitDurability, CommitRequest, commit_branch};
use crate::store::MemoryStore;
use crate::sync::ballot::{Ballot, Stamp};
use crate::sync::topology::SyncNodeId;
use crate::sync::{SyncMergeRoots, merge_committed_union, merge_synced_roots};
use crate::tree::{Hash, LeafNode, Node, TreePolicy, batch_mutate, insert};
use crate::ttl::entry::encode_stamped;
use crate::wal::WalBuffer;

type TestResult = Result<(), Box<dyn Error>>;

/// A v2 policy whose targets are small enough that every entry in [`map`] is
/// byte-oversized (`entry_size` ≥ target) and therefore a DETERMINISTIC singleton
/// leaf under §2.1's oversized-isolation rule. That makes the v2 shape observably
/// different from v1's single-leaf packing, so a caller that silently fell back to
/// v1 would fail every equality below.
const fn v2_policy() -> TreePolicy {
    TreePolicy::v2(48, 48)
}

/// A 40-byte value. With a 3–4 byte key the entry weighs
/// `16 + |k| + 40 ≥ 48` = the v2 leaf target, so it isolates as a singleton.
fn oversized(seed: u8) -> Vec<u8> {
    vec![seed; 40]
}

/// The canonical logical map every pin drives, as owned put mutations.
fn map() -> Vec<(Vec<u8>, Option<Vec<u8>>)> {
    (0u8..4)
        .map(|n| (format!("k{n:02}").into_bytes(), Some(oversized(n))))
        .collect()
}

fn empty_root(store: &mut MemoryStore) -> Result<Hash, Box<dyn Error>> {
    Ok(store.put(&Node::Leaf(LeafNode::new(Vec::new())?)))
}

/// Build [`map`] from empty under an explicit policy — the canonical reference
/// root for that policy.
fn canonical(store: &mut MemoryStore, policy: TreePolicy) -> Result<Hash, Box<dyn Error>> {
    let root = empty_root(store)?;
    Ok(batch_mutate(store, root, map().as_slice(), policy)?)
}

/// The v2 and v1 canonical roots MUST differ — otherwise "v2-shaped" is not
/// observable and every pin below would be vacuous. This is the fixture's own
/// sanity gate.
#[test]
fn fixture_v2_and_v1_roots_differ() -> TestResult {
    let mut store = MemoryStore::new();
    let v2_root = canonical(&mut store, v2_policy())?;
    let v1_root = canonical(&mut store, TreePolicy::V1_DEFAULT)?;
    assert_ne!(
        v2_root, v1_root,
        "v2 oversized-isolation must produce a different root than v1 packing"
    );
    Ok(())
}

// ── Caller 1: public tree APIs ──────────────────────────────────────────────

/// `batch_mutate` under v2 IS the canonical reference; `insert` applied
/// one-at-a-time under v2 must reach the SAME root (history independence holds
/// under the v2 detector), and neither equals the v1 root.
#[test]
fn public_tree_api_inserts_are_v2_shaped() -> TestResult {
    let mut store = MemoryStore::new();
    let canonical_v2 = canonical(&mut store, v2_policy())?;

    let mut root = empty_root(&mut store)?;
    for (key, value) in map() {
        root = insert(
            &mut store,
            root,
            &key,
            value.as_deref().unwrap_or_default(),
            v2_policy(),
        )?;
    }
    assert_eq!(root, canonical_v2, "insert-under-v2 must reach the v2 root");
    assert_ne!(root, canonical(&mut store, TreePolicy::V1_DEFAULT)?);
    Ok(())
}

// ── Caller 2: WalBuffer::commit ─────────────────────────────────────────────

#[test]
fn wal_buffer_commit_dispatches_v2() -> TestResult {
    let mut store = MemoryStore::new();
    let canonical_v2 = canonical(&mut store, v2_policy())?;
    let canonical_v1 = canonical(&mut store, TreePolicy::V1_DEFAULT)?;

    let base = empty_root(&mut store)?;
    let mut buffer = WalBuffer::new();
    for (key, value) in map() {
        buffer.put(&key, value.unwrap_or_default());
    }
    let v2_root = buffer.commit(base, &mut store, v2_policy())?;
    assert_eq!(v2_root, canonical_v2, "WalBuffer::commit must build v2");
    assert_ne!(v2_root, canonical_v1);

    // Control: the SAME buffer under v1 lands on the v1 root — the arg is live,
    // not ignored.
    let mut v1_buffer = WalBuffer::new();
    for (key, value) in map() {
        v1_buffer.put(&key, value.unwrap_or_default());
    }
    let v1_root = v1_buffer.commit(base, &mut store, TreePolicy::V1_DEFAULT)?;
    assert_eq!(v1_root, canonical_v1);
    Ok(())
}

// ── Caller 3: branch commit ─────────────────────────────────────────────────

fn commit_branch_root(policy: TreePolicy) -> Result<Hash, Box<dyn Error>> {
    let mut store = MemoryStore::new();
    let base = empty_root(&mut store)?;
    let registry = BranchRegistry::new();
    let dir = tempfile::tempdir()?;
    let mut refs = BranchRefStore::open(dir.path())?;
    let branch = create_branch("wip", [(DEFAULT_SHARD_ID, base)], &mut refs, &registry, 10)?;

    for (key, value) in map() {
        branch.put(DEFAULT_SHARD_ID, &key, value.unwrap_or_default())?;
    }
    let outcome = commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Volatile,
            extra_parents: &[],
            timestamp: 1,
        },
        policy,
    )?;
    outcome
        .heads
        .into_iter()
        .find(|(id, _)| *id == DEFAULT_SHARD_ID)
        .map(|(_, head)| head)
        .ok_or_else(|| "committed shard head absent".into())
}

#[test]
fn branch_commit_dispatches_v2() -> TestResult {
    // The materialised head equals a from-empty v2 rebuild of the same map, and
    // differs from the v1 rebuild: branch commit threads the policy to
    // `batch_mutate_owned_with_source`.
    let mut reference = MemoryStore::new();
    let canonical_v2 = canonical(&mut reference, v2_policy())?;
    let canonical_v1 = canonical(&mut reference, TreePolicy::V1_DEFAULT)?;

    assert_eq!(commit_branch_root(v2_policy())?, canonical_v2);
    assert_eq!(commit_branch_root(TreePolicy::V1_DEFAULT)?, canonical_v1);
    assert_ne!(canonical_v2, canonical_v1);
    Ok(())
}

// ── Callers 4 & 5: branch merge + sync merge (sync delegates to branch) ──────

/// A clean 3-way merge whose resolved additions are built by `batch_mutate`
/// under `tree_policy`. `ancestor == parent == {k00,k01}`, `branch =
/// {k00..k03}`, so the merge applies k02,k03 to the parent and the merged root
/// must equal a from-empty v2 rebuild of the full map.
fn merge_fixture(store: &mut MemoryStore) -> Result<(Hash, Hash, Hash, Hash), Box<dyn Error>> {
    let base = empty_root(store)?;
    let head: Vec<_> = map().into_iter().take(2).collect();
    let full = map();

    let parent = batch_mutate(store, base, head.as_slice(), v2_policy())?;
    let ancestor = parent;
    let branch = batch_mutate(store, base, full.as_slice(), v2_policy())?;
    let canonical_v2 = canonical(store, v2_policy())?;
    Ok((ancestor, parent, branch, canonical_v2))
}

#[test]
fn branch_merge_dispatches_v2() -> TestResult {
    let mut store = MemoryStore::new();
    let (ancestor, parent, branch, canonical_v2) = merge_fixture(&mut store)?;

    let report_v2 = merge_with_report(
        &mut store,
        parent,
        branch,
        ancestor,
        &ConflictPolicy::Lww,
        v2_policy(),
    )?;
    assert_eq!(
        report_v2.merged_root, canonical_v2,
        "branch merge must build additions under v2"
    );

    let report_v1 = merge_with_report(
        &mut store,
        parent,
        branch,
        ancestor,
        &ConflictPolicy::Lww,
        TreePolicy::V1_DEFAULT,
    )?;
    assert_ne!(
        report_v1.merged_root, canonical_v2,
        "under v1 the merge must NOT reach the v2 root"
    );
    Ok(())
}

#[test]
fn sync_merge_dispatches_v2() -> TestResult {
    let mut store = MemoryStore::new();
    let (ancestor, parent, branch, canonical_v2) = merge_fixture(&mut store)?;

    let result = merge_synced_roots(
        &mut store,
        DEFAULT_SHARD_ID,
        SyncMergeRoots::new(parent, branch, ancestor),
        &ConflictPolicy::Lww,
        v2_policy(),
    )?;
    assert_eq!(
        result.merged_root, canonical_v2,
        "sync merge must thread v2 through to the branch merge engine"
    );
    Ok(())
}

// ── Caller 6: sync handoff-merge ────────────────────────────────────────────

fn stamp(counter: u64, node: &str, seq: u64) -> Stamp {
    Stamp::new(Ballot::new(counter, SyncNodeId::new(node)), seq)
}

/// Build a committed tree of stamped VALUE entries under `policy`.
fn stamped_tree(
    store: &mut MemoryStore,
    entries: &[(&[u8], Vec<u8>)],
    policy: TreePolicy,
) -> Result<Hash, Box<dyn Error>> {
    let base = empty_root(store)?;
    let mutations: Vec<(Vec<u8>, Option<Vec<u8>>)> = entries
        .iter()
        .map(|(key, bytes)| (key.to_vec(), Some(bytes.clone())))
        .collect();
    Ok(batch_mutate(store, base, mutations.as_slice(), policy)?)
}

#[test]
fn handoff_merge_dispatches_v2() -> TestResult {
    // Two disjoint stamped roots; the max-stamp union rebuilds from empty under
    // the passed policy. Stamped values are large, so the union isolates per-key
    // under v2 — a different root than the v1 union, and deterministic.
    let a = [(
        b"k00".as_slice(),
        encode_stamped(oversized(0), stamp(1, "a", 1), None),
    )];
    let b = [(
        b"k01".as_slice(),
        encode_stamped(oversized(1), stamp(1, "b", 1), None),
    )];

    let mut store = MemoryStore::new();
    let root_a = stamped_tree(&mut store, &a, v2_policy())?;
    let root_b = stamped_tree(&mut store, &b, v2_policy())?;

    let present = || -> Box<dyn Error> { "union of two present roots".into() };
    let union_v2 = merge_committed_union(Some(root_a), Some(root_b), &mut store, v2_policy())?
        .ok_or_else(present)?;
    let union_v1 = merge_committed_union(
        Some(root_a),
        Some(root_b),
        &mut store,
        TreePolicy::V1_DEFAULT,
    )?
    .ok_or_else(present)?;
    assert_ne!(
        union_v2, union_v1,
        "handoff union must build under the passed policy, not a v1 literal"
    );

    // Determinism: re-running the v2 union reaches the identical root (HI).
    let union_v2_again =
        merge_committed_union(Some(root_a), Some(root_b), &mut store, v2_policy())?
            .ok_or_else(present)?;
    assert_eq!(union_v2, union_v2_again);
    Ok(())
}

// ── Mixed-policy regression pin: same map ⇒ same root regardless of path ─────

/// The load-bearing canonicity pin (§4.1 mixed-policy prohibition): the SAME
/// logical map built via three independent v2-dispatched production paths — the
/// public batch API, the WAL buffer, and branch commit — converges on ONE root.
/// If any caller had smuggled in a v1 literal, its path would diverge here.
#[test]
fn mixed_policy_regression_same_map_same_root_across_paths() -> TestResult {
    let mut store = MemoryStore::new();
    let via_batch = canonical(&mut store, v2_policy())?;

    let base = empty_root(&mut store)?;
    let mut buffer = WalBuffer::new();
    for (key, value) in map() {
        buffer.put(&key, value.unwrap_or_default());
    }
    let via_wal = buffer.commit(base, &mut store, v2_policy())?;

    let via_branch = commit_branch_root(v2_policy())?;

    assert_eq!(
        via_batch, via_wal,
        "batch vs WAL-buffer path diverged under v2"
    );
    assert_eq!(
        via_batch, via_branch,
        "batch vs branch-commit path diverged under v2"
    );
    Ok(())
}

// ── u128 formula pins (native; wasm parity by shared code) ───────────────────

/// Threshold equality (§2.1): an entry whose weight EQUALS the leaf target is a
/// deterministic singleton — `boundary_before` fires (isolation) and
/// `boundary_after` is capped at P=1. The compare is the exact `u128` arithmetic
/// shared byte-for-byte with the wasm build (no `cfg`-split in `tree/policy.rs`),
/// so this native assertion also stands as the wasm-parity witness (precedent:
/// `wasm/transport.rs` BLAKE3 parity).
#[test]
fn formula_threshold_equality_isolates_at_target() {
    let target = 48u64;
    let policy = TreePolicy::v2(target, target);
    // entry_size = 16 + |k| + |v| == target  ⇒  |k| + |v| == 32.
    let key = b"kk"; // 2
    let value_len = 30usize; // 16 + 2 + 30 == 48
    assert!(
        policy.leaf_boundary_before(key.len(), value_len),
        "an at-target entry must isolate (boundary_before)"
    );
    assert!(
        policy.leaf_boundary_after(key, value_len),
        "an at-target entry closes its chunk (boundary_after, P=1)"
    );

    // One byte under target: no forced isolation.
    assert!(
        !policy.leaf_boundary_before(key.len(), value_len - 1),
        "a sub-target entry must NOT be force-isolated"
    );
}

/// Saturating-size consistency (§2.1): a value length near `usize::MAX` weighs
/// far above any target and must isolate WITHOUT overflow panic — the detector's
/// checked/saturating length arithmetic in action.
#[test]
fn formula_saturating_size_isolates_without_panic() {
    let policy = TreePolicy::v2(64 * 1024, 48 * 1024);
    assert!(policy.leaf_boundary_before(usize::MAX, usize::MAX));
    assert!(policy.leaf_boundary_after(b"k", usize::MAX));
}