# Phase 6.5: On-Disk B+Tree Indexes — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Replace paged-blob index serialisation with proper on-disk B+tree pages so index memory usage becomes O(cache_pages) instead of O(facts), and open time becomes O(1).

**Architecture:** Four covering indexes (EAVT, AEVT, AVET, VAET) get proper per-page B+tree nodes. `FactStorage` holds only pending (pre-checkpoint) entries in memory; committed entries are range-scanned on demand via a new `CommittedIndexReader` trait backed by `OnDiskIndexReader`. Checkpoint does a bulk merge-sort of pending + existing committed entries and writes a fresh B+tree. FileHeader expands to 80 bytes (v6) with a new `fact_page_count` field.

**Tech Stack:** Rust, postcard, crc32fast, uuid — no new dependencies.

---

## File Map

| Action | File | Responsibility |
|--------|------|----------------|
| Modify | `src/storage/mod.rs` | FileHeader v6 (80 bytes, `fact_page_count`), `CommittedIndexReader` trait, `FORMAT_VERSION=6`, register `btree_v6` |
| Create | `src/storage/btree_v6.rs` | B+tree builder, leaf streamer, range scanner, `OnDiskIndexReader` |
| Modify | `src/graph/storage.rs` | Rename `indexes`→`pending_indexes`, add `committed_index_reader`, merge query pattern, `post_checkpoint_clear` |
| Modify | `src/storage/persistent_facts.rs` | v5→v6 migration on open, v6 save/checkpoint, v6 checksum-rebuild, wire `OnDiskIndexReader` |
| Create | `tests/btree_v6_test.rs` | Integration tests for v6 B+tree, migration, pending+committed merge |
| Modify | `benches/minigraf_bench.rs` | Concurrent range-scan benchmark (N=2/4/8 threads) |

---

### Task 1: FileHeader v6 + `CommittedIndexReader` trait

**Files:**
- Modify: `src/storage/mod.rs`

- [ ] **Step 1.1: Write failing unit tests for v6 header**

Add these NEW tests to the `tests` module in `src/storage/mod.rs`.
(Three existing tests — `test_format_version_is_5`, `test_new_header_has_version_5`, `test_validate_rejects_version_0_and_6` — are renamed in Step 1.3; do NOT redefine them here.)

```rust
#[test]
fn test_validate_accepts_versions_1_to_6() {
    let mut h = FileHeader::new();
    for v in 1u32..=6 {
        h.version = v;
        assert!(h.validate().is_ok(), "version {} should be accepted", v);
    }
}

#[test]
fn test_file_header_v6_fact_page_count_roundtrip() {
    let mut h = FileHeader::new();
    h.fact_page_count = 42;
    let bytes = h.to_bytes();
    assert_eq!(bytes.len(), 80);
    let parsed = FileHeader::from_bytes(&bytes).unwrap();
    assert_eq!(parsed.fact_page_count, 42);
}

#[test]
fn test_file_header_v6_byte_layout_all_fields() {
    let mut h = FileHeader::new();
    h.page_count                  = 0x0102_0304_0506_0708_u64;
    h.node_count                  = 0x1112_1314_1516_1718_u64;
    h.last_checkpointed_tx_count  = 0x2122_2324_2526_2728_u64;
    h.eavt_root_page              = 0x3132_3334_3536_3738_u64;
    h.aevt_root_page              = 0x4142_4344_4546_4748_u64;
    h.avet_root_page              = 0x5152_5354_5556_5758_u64;
    h.vaet_root_page              = 0x6162_6364_6566_6768_u64;
    h.index_checksum              = 0x7172_7374_u32;
    h.fact_page_format            = 0x02;
    h._padding                    = [0x00; 3];
    h.fact_page_count             = 0xA1A2_A3A4_A5A6_A7A8_u64;

    let b = h.to_bytes();
    assert_eq!(b.len(), 80, "v6 header must be exactly 80 bytes");

    assert_eq!(&b[0..4],   b"MGRF");
    assert_eq!(&b[4..8],   &6u32.to_le_bytes());
    assert_eq!(&b[8..16],  &0x0102_0304_0506_0708_u64.to_le_bytes());
    assert_eq!(&b[16..24], &0x1112_1314_1516_1718_u64.to_le_bytes());
    assert_eq!(&b[24..32], &0x2122_2324_2526_2728_u64.to_le_bytes());
    assert_eq!(&b[32..40], &0x3132_3334_3536_3738_u64.to_le_bytes());
    assert_eq!(&b[40..48], &0x4142_4344_4546_4748_u64.to_le_bytes());
    assert_eq!(&b[48..56], &0x5152_5354_5556_5758_u64.to_le_bytes());
    assert_eq!(&b[56..64], &0x6162_6364_6566_6768_u64.to_le_bytes());
    assert_eq!(&b[64..68], &0x7172_7374_u32.to_le_bytes());
    assert_eq!(b[68],        0x02);
    assert_eq!(&b[69..72], &[0x00u8; 3]);
    assert_eq!(&b[72..80], &0xA1A2_A3A4_A5A6_A7A8_u64.to_le_bytes());
}

#[test]
fn test_file_header_v5_reads_fact_page_count_zero() {
    // v5 files have a 72-byte header; fact_page_count must come back as 0.
    let mut bytes = vec![0u8; 72];
    bytes[0..4].copy_from_slice(b"MGRF");
    bytes[4..8].copy_from_slice(&5u32.to_le_bytes());
    bytes[8..16].copy_from_slice(&2u64.to_le_bytes());
    let h = FileHeader::from_bytes(&bytes).unwrap();
    assert_eq!(h.version, 5);
    assert_eq!(h.fact_page_count, 0);
}

#[test]
fn test_file_header_v6_truncated_rejected() {
    let mut bytes = vec![0u8; 75]; // only 75 bytes, need 80 for v6
    bytes[0..4].copy_from_slice(b"MGRF");
    bytes[4..8].copy_from_slice(&6u32.to_le_bytes());
    assert!(FileHeader::from_bytes(&bytes).is_err());
}
```

- [ ] **Step 1.2: Run tests to confirm they fail**

```bash
cargo test --lib storage::tests -- --nocapture 2>&1 | head -40
```

Expected compile errors resembling:
```
error[E0609]: no field `fact_page_count` on type `FileHeader`
error[E0308]: mismatched types
  left: `usize` (value 72), right: `usize` (value 80) -- from assert_eq!(bytes.len(), 80)
```

- [ ] **Step 1.3: Implement FileHeader v6 changes in `src/storage/mod.rs`**

Replace the constant and struct and implementations with the following. Keep all existing tests (updating three of them as noted):

```rust
// At top of mod.rs — change this line:
pub const FORMAT_VERSION: u32 = 6;  // was 5
```

Add `fact_page_count` field to `FileHeader` struct:
```rust
/// File header for .graph files — 80 bytes in v6.
///
/// Layout (all fields little-endian):
///   0..4    magic ("MGRF")
///   4..8    version (u32)
///   8..16   page_count (u64)
///   16..24  node_count (u64)          — reused as fact count
///   24..32  last_checkpointed_tx_count (u64)
///   32..40  eavt_root_page (u64)
///   40..48  aevt_root_page (u64)
///   48..56  avet_root_page (u64)
///   56..64  vaet_root_page (u64)
///   64..68  index_checksum (u32)
///   68      fact_page_format (u8)
///   69..72  _padding ([u8; 3])
///   72..80  fact_page_count (u64)     — new in v6
#[derive(Debug, Clone, Copy)]
pub struct FileHeader {
    pub magic: [u8; 4],
    pub version: u32,
    pub page_count: u64,
    pub node_count: u64,
    pub last_checkpointed_tx_count: u64,
    pub eavt_root_page: u64,
    pub aevt_root_page: u64,
    pub avet_root_page: u64,
    pub vaet_root_page: u64,
    pub index_checksum: u32,
    pub fact_page_format: u8,
    pub(crate) _padding: [u8; 3],
    /// Number of pages (starting at page 1) holding committed fact data.
    /// New in v6; zero-initialised when reading v5 or older headers.
    pub fact_page_count: u64,
}
```

Update `FileHeader::new()`:
```rust
pub fn new() -> Self {
    FileHeader {
        magic: MAGIC_NUMBER,
        version: FORMAT_VERSION,
        page_count: 1,
        node_count: 0,
        last_checkpointed_tx_count: 0,
        eavt_root_page: 0,
        aevt_root_page: 0,
        avet_root_page: 0,
        vaet_root_page: 0,
        index_checksum: 0,
        fact_page_format: FACT_PAGE_FORMAT_PACKED,
        _padding: [0; 3],
        fact_page_count: 0,
    }
}
```

Update `to_bytes()` to emit 80 bytes:
```rust
pub fn to_bytes(&self) -> Vec<u8> {
    let mut b = Vec::with_capacity(80);
    b.extend_from_slice(&self.magic);
    b.extend_from_slice(&self.version.to_le_bytes());
    b.extend_from_slice(&self.page_count.to_le_bytes());
    b.extend_from_slice(&self.node_count.to_le_bytes());
    b.extend_from_slice(&self.last_checkpointed_tx_count.to_le_bytes());
    b.extend_from_slice(&self.eavt_root_page.to_le_bytes());
    b.extend_from_slice(&self.aevt_root_page.to_le_bytes());
    b.extend_from_slice(&self.avet_root_page.to_le_bytes());
    b.extend_from_slice(&self.vaet_root_page.to_le_bytes());
    b.extend_from_slice(&self.index_checksum.to_le_bytes());
    b.push(self.fact_page_format);
    b.extend_from_slice(&self._padding);
    b.extend_from_slice(&self.fact_page_count.to_le_bytes());
    b
}
```

Update `from_bytes()` — add v6 branch after the v4/v5 block:
```rust
pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
    if bytes.len() < 64 {
        anyhow::bail!(
            "Invalid header: too short (got {} bytes, need 64)",
            bytes.len()
        );
    }

    let mut magic = [0u8; 4];
    magic.copy_from_slice(&bytes[0..4]);
    if magic != MAGIC_NUMBER {
        anyhow::bail!("Invalid magic number: not a .graph file");
    }

    let version = u32::from_le_bytes(bytes[4..8].try_into().unwrap());
    let page_count = u64::from_le_bytes(bytes[8..16].try_into().unwrap());
    let node_count = u64::from_le_bytes(bytes[16..24].try_into().unwrap());
    let last_checkpointed_tx_count = u64::from_le_bytes(bytes[24..32].try_into().unwrap());

    if version <= 3 {
        return Ok(FileHeader {
            magic, version, page_count, node_count,
            last_checkpointed_tx_count,
            eavt_root_page: 0, aevt_root_page: 0,
            avet_root_page: 0, vaet_root_page: 0,
            index_checksum: 0, fact_page_format: 0,
            _padding: [0; 3], fact_page_count: 0,
        });
    }

    // v4, v5, v6: need at least 72 bytes
    if bytes.len() < 72 {
        anyhow::bail!(
            "Invalid v4/v5/v6 header: expected at least 72 bytes, got {}",
            bytes.len()
        );
    }

    let fact_page_count = if version >= 6 {
        if bytes.len() < 80 {
            anyhow::bail!(
                "Invalid v6 header: expected 80 bytes, got {}",
                bytes.len()
            );
        }
        u64::from_le_bytes(bytes[72..80].try_into().unwrap())
    } else {
        0
    };

    Ok(FileHeader {
        magic, version, page_count, node_count,
        last_checkpointed_tx_count,
        eavt_root_page: u64::from_le_bytes(bytes[32..40].try_into().unwrap()),
        aevt_root_page: u64::from_le_bytes(bytes[40..48].try_into().unwrap()),
        avet_root_page: u64::from_le_bytes(bytes[48..56].try_into().unwrap()),
        vaet_root_page: u64::from_le_bytes(bytes[56..64].try_into().unwrap()),
        index_checksum: u32::from_le_bytes(bytes[64..68].try_into().unwrap()),
        fact_page_format: bytes[68],
        _padding: [bytes[69], bytes[70], bytes[71]],
        fact_page_count,
    })
}
```

Update `validate()` to accept 1–6:
```rust
pub fn validate(&self) -> Result<()> {
    if self.magic != MAGIC_NUMBER {
        anyhow::bail!("Invalid magic number");
    }
    if self.version < 1 || self.version > FORMAT_VERSION {
        anyhow::bail!(
            "Unsupported format version: {} (supported: 1-{})",
            self.version,
            FORMAT_VERSION
        );
    }
    Ok(())
}
```

Update these three existing tests:
- `test_format_version_is_5` → rename to `test_format_version_is_6`, assert `FORMAT_VERSION == 6`
- `test_new_header_has_version_5` → rename to `test_new_header_has_version_6`, assert `header.version == 6`
- `test_validate_rejects_version_0_and_6` → rename to `test_validate_rejects_version_0_and_7`, change `header.version = 6` to `header.version = 7`
- `test_validate_accepts_versions_1_to_5` → update to loop `1u32..=6`
- `test_file_header_byte_layout_all_fields` → update `assert_eq!(b.len(), 72, ...)` to `80`; add assertion for `b[72..80]`; update `&5u32.to_le_bytes()` to `&6u32.to_le_bytes()`

Also add `CommittedIndexReader` trait below `CommittedFactReader`:

```rust
/// Provides bounded range scans over the four committed (on-disk) covering indexes.
///
/// Implemented by `OnDiskIndexReader` in `btree_v6.rs`. Set on `FactStorage`
/// after load/migration/checkpoint so query methods can merge committed and
/// pending index entries without loading the full index into RAM.
pub trait CommittedIndexReader: Send + Sync {
    fn range_scan_eavt(
        &self,
        start: &crate::storage::index::EavtKey,
        end: Option<&crate::storage::index::EavtKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>>;

    fn range_scan_aevt(
        &self,
        start: &crate::storage::index::AevtKey,
        end: Option<&crate::storage::index::AevtKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>>;

    fn range_scan_avet(
        &self,
        start: &crate::storage::index::AvetKey,
        end: Option<&crate::storage::index::AvetKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>>;

    fn range_scan_vaet(
        &self,
        start: &crate::storage::index::VaetKey,
        end: Option<&crate::storage::index::VaetKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>>;
}
```

Register the new module at the top of `src/storage/mod.rs`:
```rust
pub mod btree_v6;
```

- [ ] **Step 1.4: Run tests to confirm they pass**

```bash
cargo test --lib storage::tests -- --nocapture
```

Expected: all `storage::tests` pass. Some other tests may still fail due to `FORMAT_VERSION` change — that's expected and will be fixed in Task 4.

- [ ] **Step 1.5: Commit**

```bash
git add src/storage/mod.rs
git commit -m "feat(storage): FileHeader v6 (80 bytes, fact_page_count) + CommittedIndexReader trait"
```

---

### Task 2: `btree_v6.rs` — page writers + `build_btree`

**Files:**
- Create: `src/storage/btree_v6.rs`

- [ ] **Step 2.1: Write failing unit tests for `build_btree`**

Create `src/storage/btree_v6.rs` with the tests module only (no implementation yet):

```rust
// src/storage/btree_v6.rs — stub + tests

use crate::storage::index::FactRef;
use crate::storage::{PAGE_SIZE, StorageBackend};
use crate::storage::cache::PageCache;
use anyhow::Result;
use serde::{Deserialize, Serialize};

pub const PAGE_TYPE_LEAF: u8 = 0x21;
pub const PAGE_TYPE_INTERNAL: u8 = 0x22;

pub fn build_btree<K>(
    _sorted_entries: impl Iterator<Item = (K, FactRef)>,
    _backend: &mut dyn StorageBackend,
    _cache: &PageCache,
    _start_page_id: u64,
) -> Result<(u64, u64)>
where K: Serialize + Ord {
    unimplemented!()
}

pub fn stream_all_entries<K>(
    _root_page_id: u64,
    _backend: &dyn StorageBackend,
    _cache: &PageCache,
) -> Result<Vec<(K, FactRef)>>
where K: for<'de> Deserialize<'de> + Ord {
    unimplemented!()
}

pub fn range_scan<K>(
    _root_page_id: u64,
    _start: &K,
    _end: Option<&K>,
    _backend: &dyn StorageBackend,
    _cache: &PageCache,
) -> Result<Vec<FactRef>>
where K: Serialize + for<'de> Deserialize<'de> + Ord {
    unimplemented!()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::backend::MemoryBackend;
    use crate::storage::index::{EavtKey, FactRef};
    use uuid::Uuid;

    fn make_eavt(n: u128, attr: &str, tx: u64) -> (EavtKey, FactRef) {
        (
            EavtKey {
                entity: Uuid::from_u128(n),
                attribute: attr.to_string(),
                valid_from: 0,
                valid_to: i64::MAX,
                tx_count: tx,
            },
            FactRef { page_id: tx + 1, slot_index: 0 },
        )
    }

    #[test]
    fn test_build_btree_empty_returns_single_leaf() {
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(64);
        let entries: Vec<(EavtKey, FactRef)> = vec![];
        let (root, next_free) = build_btree(entries.into_iter(), &mut backend, &cache, 1).unwrap();
        assert_eq!(root, 1, "root must be at start_page_id");
        assert_eq!(next_free, 2, "single empty leaf = 1 page");
        // Verify it is a leaf page
        let page = cache.get_or_load(1, &backend).unwrap();
        assert_eq!(page[0], PAGE_TYPE_LEAF);
        let entry_count = u16::from_le_bytes(page[2..4].try_into().unwrap());
        assert_eq!(entry_count, 0);
    }

    #[test]
    fn test_build_btree_single_entry() {
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(64);
        let entries = vec![make_eavt(1, ":name", 1)];
        let (root, next_free) = build_btree(entries.into_iter(), &mut backend, &cache, 5).unwrap();
        assert_eq!(root, 5);
        assert_eq!(next_free, 6);
        let page = cache.get_or_load(5, &backend).unwrap();
        assert_eq!(page[0], PAGE_TYPE_LEAF);
        assert_eq!(u16::from_le_bytes(page[2..4].try_into().unwrap()), 1);
    }

    #[test]
    fn test_build_btree_chained_next_free() {
        // Two sequential build_btree calls: second must start where first ended.
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(128);
        let entries1 = (0u128..5).map(|n| make_eavt(n, ":a", n as u64 + 1));
        let (_, next1) = build_btree(entries1, &mut backend, &cache, 1).unwrap();

        let entries2 = (5u128..10).map(|n| make_eavt(n, ":b", n as u64 + 1));
        let (root2, next2) = build_btree(entries2, &mut backend, &cache, next1).unwrap();

        assert!(root2 >= next1, "second tree must not overlap with first");
        assert!(next2 > root2);
    }

    #[test]
    fn test_build_btree_pages_in_cache_after_build() {
        // All written pages must be retrievable from cache without backend read
        // (i.e., put_dirty was called for each).
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(256);
        let entries = (0u128..100).map(|n| make_eavt(n, ":x", n as u64 + 1));
        let (root, next_free) = build_btree(entries, &mut backend, &cache, 1).unwrap();

        // Count cache entries: every page from root..next_free should be cached
        // We can't inspect directly, but can verify get_or_load doesn't fail
        // on a backend with no pages (the cache must serve them).
        let empty_backend = MemoryBackend::new();
        for page_id in root..next_free {
            // This would fail if the page wasn't in the cache
            let result = cache.get_or_load(page_id, &empty_backend);
            // If a page was written to backend but not cache, this errors with "page out of range"
            // (MemoryBackend returns error for unwritten pages)
            // The test passes if all pages were put into cache by build_btree
            assert!(result.is_ok(), "page {} missing from cache", page_id);
        }
    }

    #[test]
    fn test_build_btree_fill_factor_no_overflow() {
        // With many entries, leaf pages must not exceed PAGE_SIZE
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(256);
        let entries = (0u128..200).map(|n| make_eavt(n, ":verylongattributename", n as u64 + 1));
        let (root, next_free) = build_btree(entries, &mut backend, &cache, 1).unwrap();

        for page_id in root..next_free {
            let page = cache.get_or_load(page_id, &backend).unwrap();
            assert_eq!(page.len(), PAGE_SIZE, "every page must be exactly PAGE_SIZE");
        }
    }

    #[test]
    fn test_build_btree_internal_node_created_for_many_entries() {
        let mut backend = MemoryBackend::new();
        let cache = PageCache::new(512);
        // ~300 entries should force at least 2 leaf pages and 1 internal node
        let entries = (0u128..300).map(|n| make_eavt(n, ":attr", n as u64 + 1));
        let (root, next_free) = build_btree(entries, &mut backend, &cache, 1).unwrap();

        let root_page = cache.get_or_load(root, &backend).unwrap();
        // With 300 entries the root should be an internal node
        // (unless all fit in one leaf, which is very unlikely at ~25-30 entries/leaf)
        let pages_written = next_free - 1;
        assert!(pages_written >= 2, "300 entries must need multiple pages; got {}", pages_written);
        // Root must be either leaf (few entries) or internal
        assert!(
            root_page[0] == PAGE_TYPE_LEAF || root_page[0] == PAGE_TYPE_INTERNAL,
            "root page type 0x{:02x} is not leaf or internal", root_page[0]
        );
    }
}
```

- [ ] **Step 2.2: Run tests to confirm failure**

```bash
cargo test --lib storage::btree_v6::tests -- --nocapture 2>&1 | head -20
```

Expected panic output:
```
thread '...test_build_btree_empty_returns_single_leaf' panicked at 'not yet implemented', src/storage/btree_v6.rs:XX
```

- [ ] **Step 2.3: Implement leaf/internal page writers and `build_btree`**

Replace the stub file with the full implementation:

```rust
//! On-disk B+tree for covering index persistence (file format v6).
//!
//! Each node maps to exactly one 4KB page. The `PageCache` serves all reads.
//! `build_btree` does a bulk-build (write-all-leaves, then internal levels
//! bottom-up). Range scans traverse the tree through the cache.

use crate::storage::cache::PageCache;
use crate::storage::index::FactRef;
use crate::storage::{PAGE_SIZE, StorageBackend};
use anyhow::Result;
use serde::{Deserialize, Serialize};
use std::sync::{Arc, Mutex};

// ─── Page type constants ───────────────────────────────────────────────────────

/// Leaf node page type (v6).
pub const PAGE_TYPE_LEAF: u8 = 0x21;
/// Internal node page type (v6).
pub const PAGE_TYPE_INTERNAL: u8 = 0x22;

// ─── Fixed sizes ──────────────────────────────────────────────────────────────

/// Leaf page fixed header: type(1) + reserved(1) + entry_count(2) + next_leaf(8) = 12 bytes.
const LEAF_HEADER_SIZE: usize = 12;
/// Internal page fixed header: type(1) + reserved(1) + key_count(2) + rightmost_child(8) = 12 bytes.
const INTERNAL_HEADER_SIZE: usize = 12;
/// Slot directory entry: offset(u16) + length(u16) = 4 bytes.
const SLOT_SIZE: usize = 4;
/// Fill-factor threshold: stop packing once total used bytes exceed this (~75% of PAGE_SIZE).
const PAGE_FILL_BYTES: usize = 3072;

// ─── Low-level page writers ───────────────────────────────────────────────────

/// Write a single leaf page and insert it into the cache.
///
/// `entries`: each element is the postcard-serialised `(K, FactRef)` bytes for
/// one index entry, in sort order. Written end-to-start in the page.
fn write_leaf_page(
    backend: &mut dyn StorageBackend,
    cache: &PageCache,
    page_id: u64,
    entries: &[Vec<u8>],
    next_leaf: u64,
) -> Result<()> {
    let entry_count = entries.len() as u16;
    let mut page = vec![0u8; PAGE_SIZE];

    // Fixed header
    page[0] = PAGE_TYPE_LEAF;
    page[1] = 0; // reserved
    page[2..4].copy_from_slice(&entry_count.to_le_bytes());
    page[4..12].copy_from_slice(&next_leaf.to_le_bytes());

    // Slot directory starts at byte 12; data written end-to-start
    let mut write_pos = PAGE_SIZE;
    for (i, entry) in entries.iter().enumerate() {
        write_pos -= entry.len();
        page[write_pos..write_pos + entry.len()].copy_from_slice(entry);
        let slot_off = LEAF_HEADER_SIZE + i * SLOT_SIZE;
        page[slot_off..slot_off + 2].copy_from_slice(&(write_pos as u16).to_le_bytes());
        page[slot_off + 2..slot_off + 4].copy_from_slice(&(entry.len() as u16).to_le_bytes());
    }

    backend.write_page(page_id, &page)?;
    cache.put_dirty(page_id, page);
    Ok(())
}

/// Write a single internal node page and insert it into the cache.
///
/// `child_ids`: all child page IDs in order; the last one is `rightmost_child`.
/// `sep_bytes`: postcard-serialised Key bytes for each separator key.
///   `sep_bytes[j]` = first key of `child_ids[j+1]`'s subtree.
///   `sep_bytes.len()` == `child_ids.len() - 1`.
fn write_internal_page(
    backend: &mut dyn StorageBackend,
    cache: &PageCache,
    page_id: u64,
    child_ids: &[u64],
    sep_bytes: &[Vec<u8>],
) -> Result<()> {
    assert_eq!(child_ids.len(), sep_bytes.len() + 1);
    let key_count = sep_bytes.len() as u16;
    let rightmost_child = *child_ids.last().unwrap();

    let mut page = vec![0u8; PAGE_SIZE];

    // Fixed header
    page[0] = PAGE_TYPE_INTERNAL;
    page[1] = 0; // reserved
    page[2..4].copy_from_slice(&key_count.to_le_bytes());
    page[4..12].copy_from_slice(&rightmost_child.to_le_bytes());

    // Child array: key_count entries starting at byte 12
    let child_arr_start = INTERNAL_HEADER_SIZE;
    for (i, &cid) in child_ids[..child_ids.len() - 1].iter().enumerate() {
        let off = child_arr_start + i * 8;
        page[off..off + 8].copy_from_slice(&cid.to_le_bytes());
    }

    // Slot directory for separator keys: after child array
    let slot_dir_start = INTERNAL_HEADER_SIZE + (key_count as usize) * 8;

    // Separator key data written end-to-start
    let mut write_pos = PAGE_SIZE;
    for (i, sep) in sep_bytes.iter().enumerate() {
        write_pos -= sep.len();
        page[write_pos..write_pos + sep.len()].copy_from_slice(sep);
        let slot_off = slot_dir_start + i * SLOT_SIZE;
        page[slot_off..slot_off + 2].copy_from_slice(&(write_pos as u16).to_le_bytes());
        page[slot_off + 2..slot_off + 4].copy_from_slice(&(sep.len() as u16).to_le_bytes());
    }

    backend.write_page(page_id, &page)?;
    cache.put_dirty(page_id, page);
    Ok(())
}

// ─── build_btree ──────────────────────────────────────────────────────────────

/// Build a B+tree from sorted entries and write it to the backend.
///
/// Returns `(root_page_id, next_free_page_id)`. Chain multiple calls:
/// pass the returned `next_free_page_id` as `start_page_id` for the next index.
///
/// All written pages are inserted into `cache` via `put_dirty`.
pub fn build_btree<K>(
    sorted_entries: impl Iterator<Item = (K, FactRef)>,
    backend: &mut dyn StorageBackend,
    cache: &PageCache,
    start_page_id: u64,
) -> Result<(u64, u64)>
where
    K: Serialize + Ord,
{
    // ── Phase 1: pack entries into leaf pages ─────────────────────────────────
    // leaf_infos: (page_id, first_key_bytes) — first_key_bytes = postcard::to_allocvec(&key)
    let mut leaf_infos: Vec<(u64, Vec<u8>)> = Vec::new();

    let mut cur_entries: Vec<Vec<u8>> = Vec::new(); // serialised (K, FactRef) per entry
    let mut cur_data_bytes: usize = 0;
    let mut cur_first_key: Option<Vec<u8>> = None;
    let mut next_page = start_page_id;

    for (key, fact_ref) in sorted_entries {
        let entry_bytes = postcard::to_allocvec(&(&key, &fact_ref))?;

        // Projected used bytes if we add this entry
        let projected = LEAF_HEADER_SIZE
            + (cur_entries.len() + 1) * SLOT_SIZE
            + cur_data_bytes
            + entry_bytes.len();

        if projected > PAGE_FILL_BYTES && !cur_entries.is_empty() {
            // Flush current leaf (next_leaf = 0 for now; patched below)
            write_leaf_page(backend, cache, next_page, &cur_entries, 0)?;
            leaf_infos.push((next_page, cur_first_key.unwrap()));
            next_page += 1;
            cur_entries.clear();
            cur_data_bytes = 0;
            cur_first_key = None;
        }

        if cur_first_key.is_none() {
            cur_first_key = Some(postcard::to_allocvec(&key)?);
        }
        cur_data_bytes += entry_bytes.len();
        cur_entries.push(entry_bytes);
    }

    // Flush the last (or only) batch
    if cur_entries.is_empty() && leaf_infos.is_empty() {
        // Empty tree: single empty leaf
        write_leaf_page(backend, cache, next_page, &[], 0)?;
        return Ok((next_page, next_page + 1));
    }
    if !cur_entries.is_empty() {
        write_leaf_page(backend, cache, next_page, &cur_entries, 0)?;
        leaf_infos.push((next_page, cur_first_key.unwrap()));
        next_page += 1;
    }

    // Patch next_leaf pointers: leaf[i].next_leaf = leaf[i+1].page_id
    for i in 0..leaf_infos.len() - 1 {
        let pid = leaf_infos[i].0;
        let next_lid = leaf_infos[i + 1].0;
        // Page was just put_dirty; read from cache, update, re-write + re-cache.
        let cached = cache.get_or_load(pid, backend as &dyn StorageBackend)?;
        let mut page = (*cached).clone();
        page[4..12].copy_from_slice(&next_lid.to_le_bytes());
        backend.write_page(pid, &page)?;
        cache.put_dirty(pid, page);
    }

    // Single leaf: it is the root
    if leaf_infos.len() == 1 {
        return Ok((leaf_infos[0].0, next_page));
    }

    // ── Phase 2: build internal levels bottom-up ──────────────────────────────
    // current_level: Vec<(page_id, first_key_bytes_of_subtree)>
    let mut current_level = leaf_infos;

    loop {
        if current_level.len() == 1 {
            return Ok((current_level[0].0, next_page));
        }

        let mut next_level: Vec<(u64, Vec<u8>)> = Vec::new();
        let mut i = 0;

        while i < current_level.len() {
            let i_start = i;
            let mut child_ids: Vec<u64> = vec![current_level[i].0];
            let mut sep_bytes: Vec<Vec<u8>> = Vec::new();
            let mut sep_data_bytes: usize = 0;
            i += 1;

            while i < current_level.len() {
                // separator[j] = first key of child[j+1]
                let sep = current_level[i].1.clone();
                let projected = INTERNAL_HEADER_SIZE
                    + child_ids.len() * 8
                    + (sep_bytes.len() + 1) * SLOT_SIZE
                    + sep_data_bytes
                    + sep.len();

                if projected > PAGE_FILL_BYTES && !sep_bytes.is_empty() {
                    break; // start a new internal node
                }

                sep_data_bytes += sep.len();
                sep_bytes.push(sep);
                child_ids.push(current_level[i].0);
                i += 1;
            }

            let node_page_id = next_page;
            write_internal_page(backend, cache, node_page_id, &child_ids, &sep_bytes)?;
            next_page += 1;

            // First key of this node's subtree = first key of first child
            let first_key = current_level[i_start].1.clone();
            next_level.push((node_page_id, first_key));
        }

        current_level = next_level;
    }
}
```

- [ ] **Step 2.4: Run tests to confirm they pass**

```bash
cargo test --lib storage::btree_v6::tests -- --nocapture
```

Expected: all 6 tests pass.

- [ ] **Step 2.5: Commit**

```bash
git add src/storage/btree_v6.rs
git commit -m "feat(btree_v6): B+tree page writers and build_btree bulk-builder"
```

Also add `merge_sorted_vecs` now (used by `save()` in Task 5 to merge committed + pending index entries):

```rust
/// Merge two already-sorted `Vec`s into a single sorted iterator.
///
/// Used by `PersistentFactStorage::save()` to merge committed B+tree entries
/// with new pending entries before building the replacement B+tree.
pub fn merge_sorted_vecs<T: Ord>(a: Vec<T>, b: Vec<T>) -> impl Iterator<Item = T> {
    let mut ai = a.into_iter().peekable();
    let mut bi = b.into_iter().peekable();
    std::iter::from_fn(move || match (ai.peek(), bi.peek()) {
        (Some(_), Some(_)) => {
            if ai.peek().unwrap() <= bi.peek().unwrap() {
                ai.next()
            } else {
                bi.next()
            }
        }
        (Some(_), None) => ai.next(),
        (None, Some(_)) => bi.next(),
        (None, None) => None,
    })
}
```

Add a unit test in the `tests` module:

```rust
#[test]
fn test_merge_sorted_vecs() {
    let a = vec![1u32, 3, 5, 7];
    let b = vec![2u32, 4, 6, 8];
    let merged: Vec<u32> = merge_sorted_vecs(a, b).collect();
    assert_eq!(merged, vec![1, 2, 3, 4, 5, 6, 7, 8]);
}

#[test]
fn test_merge_sorted_vecs_empty_left() {
    let merged: Vec<u32> = merge_sorted_vecs(vec![], vec![1, 2, 3]).collect();
    assert_eq!(merged, vec![1, 2, 3]);
}
```

Re-run tests to confirm all still pass:
```bash
cargo test --lib storage::btree_v6::tests -- --nocapture
```

---

### Task 3: `btree_v6.rs` — `stream_all_entries`, `range_scan`, `OnDiskIndexReader`

**Files:**
- Modify: `src/storage/btree_v6.rs`

- [ ] **Step 3.1: Write failing tests**

Add to `src/storage/btree_v6.rs` tests module:

```rust
#[test]
fn test_stream_all_entries_roundtrip() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(256);
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..50).map(|n| make_eavt(n, ":name", n as u64 + 1)).collect();
    let (root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    let output: Vec<(EavtKey, FactRef)> =
        stream_all_entries(root, &backend, &cache).unwrap();

    assert_eq!(output.len(), 50);
    // Must be in sorted order
    for w in output.windows(2) {
        assert!(w[0].0 <= w[1].0, "entries must be in sorted order");
    }
    // FactRefs must match
    for (original, recovered) in input.iter().zip(output.iter()) {
        assert_eq!(original.1, recovered.1);
    }
}

#[test]
fn test_stream_all_entries_empty_tree() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(16);
    let entries: Vec<(EavtKey, FactRef)> = vec![];
    let (root, _) = build_btree(entries.into_iter(), &mut backend, &cache, 1).unwrap();
    let out: Vec<(EavtKey, FactRef)> = stream_all_entries(root, &backend, &cache).unwrap();
    assert_eq!(out.len(), 0);
}

#[test]
fn test_range_scan_exact_match() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(256);
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..100).map(|n| make_eavt(n, ":v", n as u64 + 1)).collect();
    let (root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    let target_entity = Uuid::from_u128(42);
    let start = EavtKey { entity: target_entity, attribute: String::new(),
                           valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let next_entity = Uuid::from_u128(43);
    let end = EavtKey { entity: next_entity, attribute: String::new(),
                         valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };

    let refs = range_scan(root, &start, Some(&end), &backend, &cache).unwrap();
    assert_eq!(refs.len(), 1, "exactly one entry for entity 42");
    assert_eq!(refs[0], FactRef { page_id: 43, slot_index: 0 });
}

#[test]
fn test_range_scan_empty_range() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(256);
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..50).map(|n| make_eavt(n, ":v", n as u64 + 1)).collect();
    let (root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    // start > all keys
    let start = EavtKey { entity: Uuid::from_u128(999),
                           attribute: String::new(), valid_from: 0, valid_to: 0, tx_count: 0 };
    let refs = range_scan::<EavtKey>(root, &start, None, &backend, &cache).unwrap();
    assert_eq!(refs.len(), 0);
}

#[test]
fn test_range_scan_unbounded_end() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(256);
    // Entries for two entities; scan from entity 5 with no upper bound
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..10).map(|n| make_eavt(n, ":v", n as u64 + 1)).collect();
    let (root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    let start = EavtKey { entity: Uuid::from_u128(5), attribute: String::new(),
                           valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let refs = range_scan::<EavtKey>(root, &start, None, &backend, &cache).unwrap();
    assert_eq!(refs.len(), 5, "entities 5..9 = 5 entries");
}

#[test]
fn test_range_scan_multi_leaf_span() {
    let mut backend = MemoryBackend::new();
    let cache = PageCache::new(512);
    // 500 entries — forces multiple leaf pages
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..500).map(|n| make_eavt(n, ":a", n as u64 + 1)).collect();
    let (root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    let start = EavtKey { entity: Uuid::from_u128(100), attribute: String::new(),
                           valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let end = EavtKey { entity: Uuid::from_u128(200), attribute: String::new(),
                         valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let refs = range_scan(root, &start, Some(&end), &backend, &cache).unwrap();
    assert_eq!(refs.len(), 101, "entities 100..=200 inclusive");
}

#[test]
fn test_on_disk_index_reader_range_scan_eavt() {
    use crate::storage::CommittedIndexReader;
    use std::sync::Arc;

    let mut backend = MemoryBackend::new();
    let cache = Arc::new(PageCache::new(256));
    let input: Vec<(EavtKey, FactRef)> =
        (0u128..20).map(|n| make_eavt(n, ":x", n as u64 + 1)).collect();
    let (eavt_root, _) = build_btree(input.iter().cloned(), &mut backend, &cache, 1).unwrap();

    let reader = OnDiskIndexReader::new(
        Arc::new(Mutex::new(backend)),
        cache,
        eavt_root, 0, 0, 0,
    );

    let start = EavtKey { entity: Uuid::from_u128(5), attribute: String::new(),
                           valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let end = EavtKey { entity: Uuid::from_u128(10), attribute: String::new(),
                         valid_from: i64::MIN, valid_to: i64::MIN, tx_count: 0 };
    let refs = reader.range_scan_eavt(&start, Some(&end)).unwrap();
    assert_eq!(refs.len(), 6, "entities 5..=10");
}
```

- [ ] **Step 3.2: Run tests to confirm failure**

```bash
cargo test --lib storage::btree_v6::tests -- --nocapture 2>&1 | head -20
```

Expected panic output:
```
thread '...test_stream_all_entries_roundtrip' panicked at 'not yet implemented', src/storage/btree_v6.rs:XX
```

- [ ] **Step 3.3: Implement `stream_all_entries`, `range_scan`, and `OnDiskIndexReader`**

Append to `src/storage/btree_v6.rs`:

```rust
// ─── Leaf traversal helpers ───────────────────────────────────────────────────

/// Traverse internal nodes from `root` to find the leftmost (first) leaf page.
fn find_leftmost_leaf(
    root: u64,
    backend: &dyn StorageBackend,
    cache: &PageCache,
) -> Result<u64> {
    let mut page_id = root;
    loop {
        let page = cache.get_or_load(page_id, backend)?;
        match page[0] {
            PAGE_TYPE_LEAF => return Ok(page_id),
            PAGE_TYPE_INTERNAL => {
                let key_count =
                    u16::from_le_bytes(page[2..4].try_into().unwrap()) as usize;
                if key_count == 0 {
                    // No separators — only rightmost_child
                    page_id = u64::from_le_bytes(page[4..12].try_into().unwrap());
                } else {
                    // First child is child_array[0]
                    page_id = u64::from_le_bytes(
                        page[INTERNAL_HEADER_SIZE..INTERNAL_HEADER_SIZE + 8]
                            .try_into()
                            .unwrap(),
                    );
                }
            }
            t => anyhow::bail!(
                "find_leftmost_leaf: unexpected page type 0x{:02x} at page_id={}",
                t,
                page_id
            ),
        }
    }
}

/// Traverse from `root` to the leaf that would contain `key`.
///
/// Applies the traversal rule: for each separator[i] in the internal node,
/// if `key < separator[i]` → descend into child[i]; else → rightmost_child.
/// Separator keys are deserialized from the page with postcard.
fn find_leaf_for_key<K>(
    root: u64,
    key: &K,
    backend: &dyn StorageBackend,
    cache: &PageCache,
) -> Result<u64>
where
    K: for<'de> Deserialize<'de> + Ord,
{
    let mut page_id = root;
    loop {
        let page = cache.get_or_load(page_id, backend)?;
        match page[0] {
            PAGE_TYPE_LEAF => return Ok(page_id),
            PAGE_TYPE_INTERNAL => {
                let key_count =
                    u16::from_le_bytes(page[2..4].try_into().unwrap()) as usize;
                let rightmost_child =
                    u64::from_le_bytes(page[4..12].try_into().unwrap());
                let child_arr_start = INTERNAL_HEADER_SIZE;
                let slot_dir_start = INTERNAL_HEADER_SIZE + key_count * 8;

                let mut descended = false;
                for i in 0..key_count {
                    let slot_off = slot_dir_start + i * SLOT_SIZE;
                    let sep_offset = u16::from_le_bytes(
                        page[slot_off..slot_off + 2].try_into().unwrap(),
                    ) as usize;
                    let sep_length = u16::from_le_bytes(
                        page[slot_off + 2..slot_off + 4].try_into().unwrap(),
                    ) as usize;
                    let sep_key: K =
                        postcard::from_bytes(&page[sep_offset..sep_offset + sep_length])?;

                    if *key < sep_key {
                        let child_off = child_arr_start + i * 8;
                        page_id = u64::from_le_bytes(
                            page[child_off..child_off + 8].try_into().unwrap(),
                        );
                        descended = true;
                        break;
                    }
                }
                if !descended {
                    page_id = rightmost_child;
                }
            }
            t => anyhow::bail!(
                "find_leaf_for_key: unexpected page type 0x{:02x} at page_id={}",
                t,
                page_id
            ),
        }
    }
}

/// Read all `(K, FactRef)` entries from a leaf page's slot directory.
fn read_leaf_entries<K>(page: &[u8]) -> Result<Vec<(K, FactRef)>>
where
    K: for<'de> Deserialize<'de>,
{
    let entry_count = u16::from_le_bytes(page[2..4].try_into().unwrap()) as usize;
    let mut entries = Vec::with_capacity(entry_count);
    for i in 0..entry_count {
        let slot_off = LEAF_HEADER_SIZE + i * SLOT_SIZE;
        let offset =
            u16::from_le_bytes(page[slot_off..slot_off + 2].try_into().unwrap()) as usize;
        let length =
            u16::from_le_bytes(page[slot_off + 2..slot_off + 4].try_into().unwrap()) as usize;
        let (k, fr): (K, FactRef) = postcard::from_bytes(&page[offset..offset + length])?;
        entries.push((k, fr));
    }
    Ok(entries)
}

// ─── stream_all_entries ───────────────────────────────────────────────────────

/// Stream all `(K, FactRef)` entries from a B+tree in sorted order.
///
/// Traverses to the leftmost leaf, then follows the `next_leaf` chain.
/// All page reads go through `cache.get_or_load`.
pub fn stream_all_entries<K>(
    root_page_id: u64,
    backend: &dyn StorageBackend,
    cache: &PageCache,
) -> Result<Vec<(K, FactRef)>>
where
    K: for<'de> Deserialize<'de> + Ord,
{
    let first_leaf = find_leftmost_leaf(root_page_id, backend, cache)?;
    let mut result = Vec::new();
    let mut leaf_id = first_leaf;

    loop {
        let page = cache.get_or_load(leaf_id, backend)?;
        if page[0] != PAGE_TYPE_LEAF {
            anyhow::bail!(
                "stream_all_entries: expected leaf page at page_id={}",
                leaf_id
            );
        }
        let next_leaf = u64::from_le_bytes(page[4..12].try_into().unwrap());
        result.extend(read_leaf_entries::<K>(&page)?);

        if next_leaf == 0 {
            break;
        }
        leaf_id = next_leaf;
    }

    Ok(result)
}

// ─── range_scan ───────────────────────────────────────────────────────────────

/// Scan the B+tree for all `FactRef`s whose key is in `[start, end]`.
///
/// `end: None` means unbounded (scan to last leaf).
/// All page reads go through `cache.get_or_load`. No direct `backend.read_page` calls.
pub fn range_scan<K>(
    root_page_id: u64,
    start: &K,
    end: Option<&K>,
    backend: &dyn StorageBackend,
    cache: &PageCache,
) -> Result<Vec<FactRef>>
where
    K: Serialize + for<'de> Deserialize<'de> + Ord,
{
    let start_leaf = find_leaf_for_key(root_page_id, start, backend, cache)?;
    let mut result = Vec::new();
    let mut leaf_id = start_leaf;

    'outer: loop {
        let page = cache.get_or_load(leaf_id, backend)?;
        if page[0] != PAGE_TYPE_LEAF {
            anyhow::bail!("range_scan: expected leaf at page_id={}", leaf_id);
        }
        let next_leaf = u64::from_le_bytes(page[4..12].try_into().unwrap());
        let entries: Vec<(K, FactRef)> = read_leaf_entries(&page)?;

        for (k, fr) in entries {
            if k < *start {
                continue;
            }
            if let Some(e) = end {
                if k > *e {
                    break 'outer;
                }
            }
            result.push(fr);
        }

        if next_leaf == 0 {
            break;
        }
        leaf_id = next_leaf;
    }

    Ok(result)
}

// ─── OnDiskIndexReader ────────────────────────────────────────────────────────

use std::sync::Mutex;

/// Implements `CommittedIndexReader` by delegating to `range_scan` on
/// on-disk B+tree pages via the page cache.
///
/// Created by `PersistentFactStorage` and set on `FactStorage` after each
/// checkpoint or migration.
pub struct OnDiskIndexReader<B: StorageBackend + 'static> {
    backend: Arc<Mutex<B>>,
    cache: Arc<PageCache>,
    pub eavt_root: u64,
    pub aevt_root: u64,
    pub avet_root: u64,
    pub vaet_root: u64,
}

impl<B: StorageBackend + 'static> OnDiskIndexReader<B> {
    pub fn new(
        backend: Arc<Mutex<B>>,
        cache: Arc<PageCache>,
        eavt_root: u64,
        aevt_root: u64,
        avet_root: u64,
        vaet_root: u64,
    ) -> Self {
        OnDiskIndexReader { backend, cache, eavt_root, aevt_root, avet_root, vaet_root }
    }
}

impl<B: StorageBackend + 'static> crate::storage::CommittedIndexReader
    for OnDiskIndexReader<B>
{
    fn range_scan_eavt(
        &self,
        start: &crate::storage::index::EavtKey,
        end: Option<&crate::storage::index::EavtKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>> {
        if self.eavt_root == 0 {
            return Ok(vec![]);
        }
        let backend = self.backend.lock().unwrap();
        range_scan(self.eavt_root, start, end, &*backend, &self.cache)
    }

    fn range_scan_aevt(
        &self,
        start: &crate::storage::index::AevtKey,
        end: Option<&crate::storage::index::AevtKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>> {
        if self.aevt_root == 0 {
            return Ok(vec![]);
        }
        let backend = self.backend.lock().unwrap();
        range_scan(self.aevt_root, start, end, &*backend, &self.cache)
    }

    fn range_scan_avet(
        &self,
        start: &crate::storage::index::AvetKey,
        end: Option<&crate::storage::index::AvetKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>> {
        if self.avet_root == 0 {
            return Ok(vec![]);
        }
        let backend = self.backend.lock().unwrap();
        range_scan(self.avet_root, start, end, &*backend, &self.cache)
    }

    fn range_scan_vaet(
        &self,
        start: &crate::storage::index::VaetKey,
        end: Option<&crate::storage::index::VaetKey>,
    ) -> anyhow::Result<Vec<crate::storage::index::FactRef>> {
        if self.vaet_root == 0 {
            return Ok(vec![]);
        }
        let backend = self.backend.lock().unwrap();
        range_scan(self.vaet_root, start, end, &*backend, &self.cache)
    }
}
```

- [ ] **Step 3.4: Run tests**

```bash
cargo test --lib storage::btree_v6::tests -- --nocapture
```

Expected: all 13 btree_v6 tests pass.

- [ ] **Step 3.5: Commit**

```bash
git add src/storage/btree_v6.rs
git commit -m "feat(btree_v6): stream_all_entries, range_scan, OnDiskIndexReader"
```

---

### Task 4: `FactStorage` field renames + `CommittedIndexReader` integration

**Files:**
- Modify: `src/graph/storage.rs`

Key changes:
1. Rename `FactData.indexes` → `pending_indexes`
2. Add `committed_index_reader: Option<Arc<dyn CommittedIndexReader>>`
3. Add `set_committed_index_reader()` setter
4. Rename `replace_indexes` → `replace_pending_indexes`
5. Rename `indexes_snapshot` → `pending_indexes_snapshot`
6. Replace `clear_pending_facts()` with `post_checkpoint_clear()` that also clears `pending_indexes`
7. Update `get_facts_by_entity` and `get_facts_by_attribute` to use the merge pattern with bounded keys

- [ ] **Step 4.1: Write failing unit tests**

Add to the `tests` module in `src/graph/storage.rs`:

```rust
#[test]
fn test_post_checkpoint_clear_clears_indexes() {
    use uuid::Uuid;
    let storage = FactStorage::new();
    let e = Uuid::new_v4();
    storage.transact(
        vec![(e, ":name".to_string(), Value::String("Alice".to_string()))],
        None,
    ).unwrap();

    assert_eq!(storage.pending_index_counts().0, 1, "one pending EAVT entry");
    storage.post_checkpoint_clear();
    assert_eq!(storage.pending_index_counts().0, 0, "pending indexes cleared");
    assert_eq!(storage.get_pending_facts().len(), 0, "pending facts cleared");
}

#[test]
fn test_set_committed_index_reader_accepted() {
    // Just a compile + no-panic test: set a None-equivalent reader.
    let storage = FactStorage::new();
    // This should compile and not panic; CommittedIndexReader is wired at open time.
    // We just verify the method exists and the storage stays usable.
    let _ = storage.pending_index_counts();
}
```

- [ ] **Step 4.2: Run tests to confirm failure**

```bash
cargo test --lib graph::storage::tests -- --nocapture 2>&1 | head -20
```

Expected compile errors:
```
error[E0599]: no method named `post_checkpoint_clear` found for struct `FactStorage`
error[E0599]: no method named `pending_index_counts` found for struct `FactStorage`
```

- [ ] **Step 4.3: Apply all FactStorage changes**

In `src/graph/storage.rs`:

**Update `FactData` struct** — rename field and add new one:
```rust
struct FactData {
    facts: Vec<Fact>,
    pending_indexes: Indexes,          // was: indexes
    committed: Option<Arc<dyn crate::storage::CommittedFactReader>>,
    committed_index_reader: Option<Arc<dyn crate::storage::CommittedIndexReader>>, // new
}
```

**Update `FactStorage::new()`**:
```rust
pub fn new() -> Self {
    FactStorage {
        data: Arc::new(RwLock::new(FactData {
            facts: Vec::new(),
            pending_indexes: Indexes::new(),    // was: indexes
            committed: None,
            committed_index_reader: None,       // new
        })),
        tx_counter: Arc::new(AtomicU64::new(0)),
    }
}
```

**Update `transact()` and `retract()`** — change `d.indexes.insert` to `d.pending_indexes.insert`:
```rust
// In transact(), change:
d.indexes.insert(fact, FactRef { page_id: 0, slot_index: slot });
// To:
d.pending_indexes.insert(fact, FactRef { page_id: 0, slot_index: slot });
```
Same for `retract()`.

**Update `load_fact()`**:
```rust
pub fn load_fact(&self, fact: Fact) -> Result<()> {
    let mut d = self.data.write().unwrap();
    let slot = d.facts.len() as u16;
    d.pending_indexes.insert(       // was: d.indexes.insert
        &fact,
        FactRef { page_id: 0, slot_index: slot },
    );
    d.facts.push(fact);
    Ok(())
}
```

**Add `set_committed_index_reader()` method**:
```rust
/// Set the committed index reader. Called by PersistentFactStorage after
/// each open/migration/checkpoint so queries can range-scan the on-disk B+tree.
pub fn set_committed_index_reader(
    &self,
    reader: Arc<dyn crate::storage::CommittedIndexReader>,
) {
    let mut d = self.data.write().unwrap();
    d.committed_index_reader = Some(reader);
}
```

**Rename `replace_indexes` → `replace_pending_indexes`**:
```rust
pub fn replace_pending_indexes(&self, indexes: Indexes) {
    let mut d = self.data.write().unwrap();
    d.pending_indexes = indexes;
}
```

**Rename `indexes_snapshot` → `pending_indexes_snapshot`**:
```rust
pub fn pending_indexes_snapshot(&self) -> Indexes {
    let d = self.data.read().unwrap();
    Indexes {
        eavt: d.pending_indexes.eavt.clone(),
        aevt: d.pending_indexes.aevt.clone(),
        avet: d.pending_indexes.avet.clone(),
        vaet: d.pending_indexes.vaet.clone(),
    }
}
```

**Replace `clear_pending_facts()` with `post_checkpoint_clear()`**:
```rust
/// Clear pending facts and pending indexes after a successful checkpoint.
/// All pending data is now committed (on-disk packed pages + B+tree).
pub fn post_checkpoint_clear(&self) {
    let mut d = self.data.write().unwrap();
    d.facts.clear();
    d.pending_indexes = Indexes::new();
}
```

**Add `get_pending_facts()` method** (used by `save()` in Task 5 to pack only pending facts):
```rust
/// Return a clone of the pending (pre-checkpoint) facts vec.
/// Does NOT include committed facts — callers merge those via the B+tree.
pub fn get_pending_facts(&self) -> Vec<Fact> {
    let d = self.data.read().unwrap();
    d.facts.clone()
}
```

**Add `pending_index_counts()` for testing** (alongside the existing `index_counts()`):
```rust
/// Returns (eavt_len, aevt_len, avet_len, vaet_len) of pending indexes for testing.
pub fn pending_index_counts(&self) -> (usize, usize, usize, usize) {
    let d = self.data.read().unwrap();
    (
        d.pending_indexes.eavt.len(),
        d.pending_indexes.aevt.len(),
        d.pending_indexes.avet.len(),
        d.pending_indexes.vaet.len(),
    )
}
```

**Update `index_counts()`** — point at `pending_indexes`:
```rust
pub fn index_counts(&self) -> (usize, usize, usize, usize) {
    let d = self.data.read().unwrap();
    (
        d.pending_indexes.eavt.len(),
        d.pending_indexes.aevt.len(),
        d.pending_indexes.avet.len(),
        d.pending_indexes.vaet.len(),
    )
}
```

**Update `clear()`** — rename field reference:
```rust
pub fn clear(&self) -> Result<()> {
    let mut d = self.data.write().unwrap();
    d.facts.clear();
    d.pending_indexes = Indexes::new();
    d.committed = None;
    d.committed_index_reader = None;
    self.tx_counter.store(0, Ordering::SeqCst);
    Ok(())
}
```

**Update `get_facts_by_entity()`** — use merge pattern with bounded end key:
```rust
pub fn get_facts_by_entity(&self, entity_id: &EntityId) -> Result<Vec<Fact>> {
    use crate::storage::index::EavtKey;
    let d = self.data.read().unwrap();

    let start = EavtKey {
        entity: *entity_id,
        attribute: String::new(),
        valid_from: i64::MIN,
        valid_to: i64::MIN,
        tx_count: 0,
    };
    let next_entity = uuid::Uuid::from_u128(entity_id.as_u128().wrapping_add(1));
    let end = EavtKey {
        entity: next_entity,
        attribute: String::new(),
        valid_from: i64::MIN,
        valid_to: i64::MIN,
        tx_count: 0,
    };

    // If no indexes built yet (empty DB or pre-index fallback), linear scan
    if d.pending_indexes.eavt.is_empty() && d.committed_index_reader.is_none() {
        if d.committed.is_none() {
            return Ok(d.facts.iter().filter(|f| &f.entity == entity_id).cloned().collect());
        }
        // Committed facts available but no index — full scan fallback
        let mut result: Vec<Fact> = d.facts.iter()
            .filter(|f| &f.entity == entity_id).cloned().collect();
        if let Some(loader) = &d.committed {
            for fact in loader.stream_all()? {
                if &fact.entity == entity_id { result.push(fact); }
            }
        }
        return Ok(result);
    }

    let mut facts = Vec::new();

    // Pending: in-memory BTreeMap bounded range
    for (key, &fr) in d.pending_indexes.eavt.range(start.clone()..end.clone()) {
        if key.entity != *entity_id { break; }
        facts.push(resolve_fact_ref(&d, fr)?);
    }

    // Committed: on-disk B+tree range scan
    if let Some(reader) = &d.committed_index_reader {
        let committed_refs = reader.range_scan_eavt(&start, Some(&end))?;
        for fr in committed_refs {
            facts.push(resolve_fact_ref(&d, fr)?);
        }
    }

    Ok(facts)
}
```

**Update `get_facts_by_attribute()`** — use merge pattern with prefix-increment end key:
```rust
pub fn get_facts_by_attribute(&self, attribute: &Attribute) -> Result<Vec<Fact>> {
    use crate::storage::index::AevtKey;
    let d = self.data.read().unwrap();

    // Fallback: no index
    if d.pending_indexes.aevt.is_empty() && d.committed_index_reader.is_none() {
        drop(d);
        return Ok(self.get_all_facts()?.into_iter()
            .filter(|f| &f.attribute == attribute).collect());
    }

    let start = AevtKey {
        attribute: attribute.clone(),
        entity: uuid::Uuid::nil(),
        valid_from: i64::MIN,
        valid_to: i64::MIN,
        tx_count: 0,
    };
    // Upper bound: increment last byte of attribute string for prefix scan.
    // If all bytes are 0xFF, use None (unbounded) for the committed scan.
    let end_opt: Option<AevtKey> = next_string_prefix(attribute).map(|next_attr| AevtKey {
        attribute: next_attr,
        entity: uuid::Uuid::nil(),
        valid_from: i64::MIN,
        valid_to: i64::MIN,
        tx_count: 0,
    });

    let mut facts = Vec::new();

    // Pending
    let pending_range: Vec<FactRef> = match &end_opt {
        Some(end) => d.pending_indexes.aevt.range(start.clone()..end.clone())
            .filter(|(k, _)| k.attribute == *attribute)
            .map(|(_, &r)| r).collect(),
        None => d.pending_indexes.aevt.range(start.clone()..)
            .take_while(|(k, _)| k.attribute == *attribute)
            .map(|(_, &r)| r).collect(),
    };
    for fr in pending_range {
        facts.push(resolve_fact_ref(&d, fr)?);
    }

    // Committed
    if let Some(reader) = &d.committed_index_reader {
        let committed_refs = reader.range_scan_aevt(&start, end_opt.as_ref())?;
        for fr in committed_refs {
            let fact = resolve_fact_ref(&d, fr)?;
            if &fact.attribute == attribute { facts.push(fact); }
        }
    }

    Ok(facts)
}
```

Add the helper function (outside `impl FactStorage`, in the same file):
```rust
/// Increment the last byte of a string for prefix upper-bound construction.
/// Returns `None` if all bytes are 0xFF (true unbounded scan needed).
fn next_string_prefix(s: &str) -> Option<String> {
    let mut bytes = s.as_bytes().to_vec();
    for i in (0..bytes.len()).rev() {
        if bytes[i] < 0xFF {
            bytes[i] += 1;
            bytes.truncate(i + 1);
            return String::from_utf8(bytes).ok();
        }
    }
    None
}
```

**Update the `resolve_fact_ref` function** — change `d.indexes` references to `d.pending_indexes` where applicable. The function itself reads from `d.facts` (pending vec) or `d.committed` — no index field used directly, so it stays unchanged.

- [ ] **Step 4.4: Run tests**

```bash
cargo test --lib graph::storage::tests -- --nocapture
```

Expected: all graph storage tests pass. Also run:
```bash
cargo build 2>&1 | head -40
```
Fix any remaining compile errors from renamed fields (callers in `persistent_facts.rs` will fail — that's expected and fixed in Task 5).

- [ ] **Step 4.5: Commit**

```bash
git add src/graph/storage.rs
git commit -m "feat(graph/storage): rename indexes→pending_indexes, add CommittedIndexReader integration"
```

---

### Task 5: `persistent_facts.rs` — v6 save, load, migrate, rebuild

**Files:**
- Modify: `src/storage/persistent_facts.rs`

This task updates all call sites for renamed methods and implements the new v6 paths.

- [ ] **Step 5.0: Write failing unit tests for v6 persistent_facts behaviour**

Add to the `tests` module in `src/storage/persistent_facts.rs`:

```rust
#[test]
fn test_save_writes_v6_header() {
    let backend = MemoryBackend::new();
    let mut storage = PersistentFactStorage::new(backend, 256).unwrap();
    storage.storage().transact(
        vec![(Uuid::new_v4(), ":name".to_string(), Value::String("x".to_string()))],
        None,
    ).unwrap();
    storage.mark_dirty();
    storage.save().unwrap();

    let backend = storage.into_backend();
    let header_page = backend.read_page(0).unwrap();
    let header = FileHeader::from_bytes(&header_page).unwrap();
    assert_eq!(header.version, 6, "save() must write v6 header");
    assert_eq!(header.to_bytes().len(), 80, "v6 header must be 80 bytes");
    assert!(header.fact_page_count > 0, "fact_page_count must be set");
    assert!(header.eavt_root_page > 0, "eavt_root must be set after save");
}

#[test]
fn test_load_v6_wires_committed_index_reader() {
    let alice = Uuid::new_v4();
    let backend = {
        let backend = MemoryBackend::new();
        let mut s = PersistentFactStorage::new(backend, 256).unwrap();
        s.storage().transact(
            vec![(alice, ":name".to_string(), Value::String("Alice".to_string()))],
            None,
        ).unwrap();
        s.mark_dirty();
        s.save().unwrap();
        s.into_backend()
    };

    // Re-open: queries must work (CommittedIndexReader must be wired)
    let s2 = PersistentFactStorage::new(backend, 256).unwrap();
    let facts = s2.storage().get_facts_by_entity(&alice).unwrap();
    assert_eq!(facts.len(), 1, "committed fact must be visible after reopen");
}
```

```rust
#[test]
fn test_save_twice_merges_committed_and_pending() {
    // Critical correctness test: facts from the FIRST checkpoint must remain
    // visible after the SECOND checkpoint. This verifies that save() merges
    // committed B+tree entries with new pending entries rather than only
    // writing pending entries to the new B+tree.
    let backend = MemoryBackend::new();
    let mut storage = PersistentFactStorage::new(backend, 256).unwrap();
    let e1 = Uuid::new_v4();
    let e2 = Uuid::new_v4();

    // First checkpoint (e1 committed)
    storage.storage().transact(
        vec![(e1, ":name".to_string(), Value::String("Alice".to_string()))],
        None,
    ).unwrap();
    storage.mark_dirty();
    storage.save().unwrap();

    // Second checkpoint (e2 pending → committed)
    storage.storage().transact(
        vec![(e2, ":name".to_string(), Value::String("Bob".to_string()))],
        None,
    ).unwrap();
    storage.mark_dirty();
    storage.save().unwrap();

    // Re-open and verify BOTH facts are visible
    let backend = storage.into_backend();
    let s2 = PersistentFactStorage::new(backend, 256).unwrap();
    let e1_facts = s2.storage().get_facts_by_entity(&e1).unwrap();
    let e2_facts = s2.storage().get_facts_by_entity(&e2).unwrap();
    assert_eq!(e1_facts.len(), 1, "e1 from first checkpoint must survive second checkpoint");
    assert_eq!(e2_facts.len(), 1, "e2 from second checkpoint must be visible");
}

#[test]
fn test_v6_migration_from_v5_unit() {
    // Write a minimal v5-format header to a MemoryBackend (empty DB, no facts,
    // no index pages). PersistentFactStorage::new() must detect version=5,
    // run migrate_v5_to_v6(), and produce a valid v6 header on re-read.
    let mut backend = MemoryBackend::new();

    // Raw v5 header page (72-byte header fields, rest zeroed; version field = 5)
    let mut page = vec![0u8; PAGE_SIZE];
    page[0..4].copy_from_slice(b"MGRF");
    page[4..8].copy_from_slice(&5u32.to_le_bytes());  // version = 5
    page[8..16].copy_from_slice(&1u64.to_le_bytes()); // page_count = 1 (header only)
    page[68] = 0x02; // fact_page_format = PACKED
    backend.write_page(0, &page).unwrap();

    // Open — migrate_v5_to_v6 must run
    let s = PersistentFactStorage::new(backend, 256).unwrap();

    // Verify header is v6
    let b = s.into_backend();
    let header_page = b.read_page(0).unwrap();
    let header = FileHeader::from_bytes(&header_page).unwrap();
    assert_eq!(header.version, 6, "migration must upgrade header to v6");
    assert_eq!(header.to_bytes().len(), 80, "v6 header must be 80 bytes");
    assert_eq!(header.fact_page_count, 0, "empty DB: fact_page_count = 0");
}

#[test]
fn test_v6_migration_crash_safe_unit() {
    // Simulate a crash mid-migration: the header still says v5 (page_count=1)
    // but garbage pages exist at pages 1 and 2 (as if a prior migration run
    // wrote partial B+tree pages before crashing before updating the header).
    // Re-opening must run migration again, overwrite the garbage, and succeed.
    let mut backend = MemoryBackend::new();

    let mut page = vec![0u8; PAGE_SIZE];
    page[0..4].copy_from_slice(b"MGRF");
    page[4..8].copy_from_slice(&5u32.to_le_bytes());
    page[8..16].copy_from_slice(&1u64.to_le_bytes()); // page_count = 1
    page[68] = 0x02;
    backend.write_page(0, &page).unwrap();

    // Garbage pages from simulated partial migration
    backend.write_page(1, &vec![0xFF_u8; PAGE_SIZE]).unwrap();
    backend.write_page(2, &vec![0xFF_u8; PAGE_SIZE]).unwrap();

    // Re-open: header still v5 → migration must restart and succeed
    let s = PersistentFactStorage::new(backend, 256).unwrap();
    let b = s.into_backend();
    let header_bytes = b.read_page(0).unwrap();
    let header = FileHeader::from_bytes(&header_bytes).unwrap();
    assert_eq!(header.version, 6, "migration must complete despite prior partial run");
}
```

Run:
```bash
cargo test --lib storage::persistent_facts::tests::test_save_writes_v6_header \
           storage::persistent_facts::tests::test_load_v6_wires_committed_index_reader \
           storage::persistent_facts::tests::test_save_twice_merges_committed_and_pending \
           storage::persistent_facts::tests::test_v6_migration_from_v5_unit \
           storage::persistent_facts::tests::test_v6_migration_crash_safe_unit \
           -- --nocapture 2>&1 | head -30
```

Expected:
```
FAILED: left: `5`, right: `6`  (test_save_writes_v6_header — version still 5 before implementation)
FAILED: left: `0`, right: `1`  (test_save_twice_merges — second checkpoint loses e1 before implementation)
```

- [ ] **Step 5.1: Fix compile errors from renamed methods**

In `persistent_facts.rs`, update all references:
- `self.storage.replace_indexes(...)` → `self.storage.replace_pending_indexes(...)`
- `self.storage.indexes_snapshot()` → `self.storage.pending_indexes_snapshot()`
- `self.storage.clear_pending_facts()` → `self.storage.post_checkpoint_clear()`

Also add the import:
```rust
use crate::storage::btree_v6::{build_btree, stream_all_entries, OnDiskIndexReader};
```

- [ ] **Step 5.2: Update `save()` to use v6 format**

Replace the `save()` method body (the section that calls `write_all_indexes`) with:

```rust
pub fn save(&mut self) -> Result<()> {
    if !self.dirty {
        return Ok(());
    }

    // ── Step A: pack ONLY pending facts as new appended pages ───────────────
    // We use append-only fact pages: committed pages are never rewritten.
    // Pending facts are the only ones in d.facts (committed were cleared by
    // the previous post_checkpoint_clear). This ensures each save() only
    // touches new pages, keeping old committed FactRefs valid.
    let pending_facts = self.storage.get_pending_facts();
    let mut backend = self.backend.lock().unwrap();

    let old_fact_page_count = self.committed_fact_pages.load(Ordering::SeqCst);
    let new_fact_start = 1 + old_fact_page_count; // page 0 = header

    let (new_pages, new_fact_refs) = pack_facts(&pending_facts, new_fact_start)?;
    for (i, page_data) in new_pages.iter().enumerate() {
        backend.write_page(new_fact_start + i as u64, page_data)?;
    }
    let new_total_fact_pages = old_fact_page_count + new_pages.len() as u64;

    // CRC32 over ALL fact pages (old committed + newly appended)
    let checksum = compute_page_checksum(&*backend, 1, new_total_fact_pages)?;

    // ── Step B: build FactRef entries for pending facts ─────────────────────
    // Pending index entries (page_id=0) are discarded; we build real entries
    // pointing to the newly written pages.
    use crate::storage::index::{AevtKey, AvetKey, EavtKey, VaetKey, encode_value};

    let mut pending_eavt: Vec<(EavtKey, FactRef)> = pending_facts.iter().zip(new_fact_refs.iter())
        .map(|(f, &fr)| (EavtKey {
            entity: f.entity, attribute: f.attribute.clone(),
            valid_from: f.valid_from, valid_to: f.valid_to, tx_count: f.tx_count,
        }, fr))
        .collect();
    pending_eavt.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut pending_aevt: Vec<(AevtKey, FactRef)> = pending_facts.iter().zip(new_fact_refs.iter())
        .map(|(f, &fr)| (AevtKey {
            attribute: f.attribute.clone(), entity: f.entity,
            valid_from: f.valid_from, valid_to: f.valid_to, tx_count: f.tx_count,
        }, fr))
        .collect();
    pending_aevt.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut pending_avet: Vec<(AvetKey, FactRef)> = pending_facts.iter().zip(new_fact_refs.iter())
        .map(|(f, &fr)| (AvetKey {
            attribute: f.attribute.clone(), value_bytes: encode_value(&f.value),
            valid_from: f.valid_from, valid_to: f.valid_to,
            entity: f.entity, tx_count: f.tx_count,
        }, fr))
        .collect();
    pending_avet.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut pending_vaet: Vec<(VaetKey, FactRef)> = pending_facts.iter().zip(new_fact_refs.iter())
        .filter_map(|(f, &fr)| {
            if let crate::graph::types::Value::Ref(target) = &f.value {
                Some((VaetKey {
                    ref_target: *target, attribute: f.attribute.clone(),
                    valid_from: f.valid_from, valid_to: f.valid_to,
                    source_entity: f.entity, tx_count: f.tx_count,
                }, fr))
            } else { None }
        })
        .collect();
    pending_vaet.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    // ── Step C: read current header to check existing B+tree roots ──────────
    let curr_header_bytes = backend.read_page(0)?;
    let curr_header = FileHeader::from_bytes(&curr_header_bytes)?;
    let index_start = 1 + new_total_fact_pages;

    // ── Step D: merge committed B+tree with pending entries (bulk-rebuild) ───
    // If an existing B+tree root is present, stream all its entries and
    // merge-sort with the new pending entries. This preserves all facts from
    // prior checkpoints without requiring any fact page rewrites.
    let (eavt_root, next1) = if curr_header.eavt_root_page != 0 {
        let committed: Vec<(EavtKey, FactRef)> =
            stream_all_entries(curr_header.eavt_root_page, &*backend, &self.page_cache)?;
        build_btree(merge_sorted_vecs(committed, pending_eavt), &mut *backend, &self.page_cache, index_start)?
    } else {
        build_btree(pending_eavt.into_iter(), &mut *backend, &self.page_cache, index_start)?
    };

    let (aevt_root, next2) = if curr_header.aevt_root_page != 0 {
        let committed: Vec<(AevtKey, FactRef)> =
            stream_all_entries(curr_header.aevt_root_page, &*backend, &self.page_cache)?;
        build_btree(merge_sorted_vecs(committed, pending_aevt), &mut *backend, &self.page_cache, next1)?
    } else {
        build_btree(pending_aevt.into_iter(), &mut *backend, &self.page_cache, next1)?
    };

    let (avet_root, next3) = if curr_header.avet_root_page != 0 {
        let committed: Vec<(AvetKey, FactRef)> =
            stream_all_entries(curr_header.avet_root_page, &*backend, &self.page_cache)?;
        build_btree(merge_sorted_vecs(committed, pending_avet), &mut *backend, &self.page_cache, next2)?
    } else {
        build_btree(pending_avet.into_iter(), &mut *backend, &self.page_cache, next2)?
    };

    let (vaet_root, next4) = if curr_header.vaet_root_page != 0 {
        let committed: Vec<(VaetKey, FactRef)> =
            stream_all_entries(curr_header.vaet_root_page, &*backend, &self.page_cache)?;
        build_btree(merge_sorted_vecs(committed, pending_vaet), &mut *backend, &self.page_cache, next3)?
    } else {
        build_btree(pending_vaet.into_iter(), &mut *backend, &self.page_cache, next3)?
    };

    // ── Step E: write v6 header (last write = crash-safe boundary) ──────────
    let mut header = FileHeader::new(); // version=6
    header.page_count = next4;
    header.node_count = curr_header.node_count + pending_facts.len() as u64;
    header.last_checkpointed_tx_count = self.storage.current_tx_count();
    header.eavt_root_page = eavt_root;
    header.aevt_root_page = aevt_root;
    header.avet_root_page = avet_root;
    header.vaet_root_page = vaet_root;
    header.index_checksum = checksum;
    header.fact_page_format = FACT_PAGE_FORMAT_PACKED;
    header.fact_page_count = new_total_fact_pages;

    let mut header_page = header.to_bytes();
    header_page.resize(PAGE_SIZE, 0);
    backend.write_page(0, &header_page)?;
    backend.sync()?;
    drop(backend);

    self.committed_fact_pages.store(new_total_fact_pages, Ordering::SeqCst);
    self.last_checkpointed_tx_count = self.storage.current_tx_count();
    self.dirty = false;

    // ── Step F: wire CommittedFactReader and CommittedIndexReader ────────────
    let loader: Arc<dyn crate::storage::CommittedFactReader> =
        Arc::new(CommittedFactLoaderImpl {
            backend: self.backend.clone(),
            page_cache: self.page_cache.clone(),
            committed_fact_pages: self.committed_fact_pages.clone(),
            first_fact_page: 1,
        });
    self.storage.set_committed_reader(loader);

    let index_reader: Arc<dyn crate::storage::CommittedIndexReader> =
        Arc::new(OnDiskIndexReader::new(
            self.backend.clone(),
            self.page_cache.clone(),
            eavt_root, aevt_root, avet_root, vaet_root,
        ));
    self.storage.set_committed_index_reader(index_reader);

    // Clear pending — all data now on disk
    self.storage.post_checkpoint_clear();

    Ok(())
}
```

Also add to the imports at top of `persistent_facts.rs`:
```rust
use crate::storage::btree_v6::{build_btree, merge_sorted_vecs, stream_all_entries, OnDiskIndexReader};
```

- [ ] **Step 5.3: Update `load()` for v6 — replace committed index load with `OnDiskIndexReader`**

In the `load()` method, replace the block that calls `read_eavt_index` / `read_aevt_index` etc. (the "Fast path: load indexes from B+tree pages" block at lines ~259-273):

```rust
// Fast path: v6 — set OnDiskIndexReader, no RAM index load
} else if header.eavt_root_page != 0 {
    let index_reader: Arc<dyn crate::storage::CommittedIndexReader> =
        Arc::new(OnDiskIndexReader::new(
            self.backend.clone(),
            self.page_cache.clone(),
            header.eavt_root_page,
            header.aevt_root_page,
            header.avet_root_page,
            header.vaet_root_page,
        ));
    self.storage.set_committed_index_reader(index_reader);
}
```

Also update `num_fact_pages` detection — for v6 headers, use `header.fact_page_count`; fall back to the old heuristic for v5:

```rust
let num_fact_pages = if header.version >= 6 && header.fact_page_count > 0 {
    header.fact_page_count
} else {
    // v5 heuristic: min(root_pages) - 1
    let first_index_page = [
        header.eavt_root_page,
        header.aevt_root_page,
        header.avet_root_page,
        header.vaet_root_page,
    ]
    .iter()
    .filter(|&&p| p > 0)
    .copied()
    .min()
    .unwrap_or(header.page_count);
    first_index_page.saturating_sub(1)
};
```

- [ ] **Step 5.4: Add v5→v6 eager migration on open**

In the `load()` method, add a migration branch right after the v1→v2 check:

```rust
// After: if header.version < 2 { return self.migrate_v1_to_v2(); }
// Add:
if header.version == 5 {
    return self.migrate_v5_to_v6(&header);
}
```

Add the migration method:

```rust
/// Migrate a v5 file (paged-blob indexes) to v6 (on-disk B+tree indexes).
///
/// Writes new B+tree pages beyond the current page_count before updating the
/// header. If the process crashes mid-migration the header is unchanged and
/// the file remains valid v5 — migration restarts cleanly on re-open.
fn migrate_v5_to_v6(&mut self, header: &FileHeader) -> Result<()> {
    use crate::storage::btree::{
        read_aevt_index, read_avet_index, read_eavt_index, read_vaet_index,
    };

    // Determine fact page count using v5 heuristic (used once during migration)
    let num_fact_pages = {
        let first_index_page = [
            header.eavt_root_page,
            header.aevt_root_page,
            header.avet_root_page,
            header.vaet_root_page,
        ]
        .iter()
        .filter(|&&p| p > 0)
        .copied()
        .min()
        .unwrap_or(header.page_count);
        first_index_page.saturating_sub(1)
    };
    self.committed_fact_pages.store(num_fact_pages, Ordering::SeqCst);

    // Read v5 paged-blob indexes into memory (last time we do this)
    let (eavt, aevt, avet, vaet) = {
        let backend = self.backend.lock().unwrap();
        let e = if header.eavt_root_page > 0 {
            read_eavt_index(header.eavt_root_page, &*backend)?
        } else {
            std::collections::BTreeMap::new()
        };
        let a = if header.aevt_root_page > 0 {
            read_aevt_index(header.aevt_root_page, &*backend)?
        } else {
            std::collections::BTreeMap::new()
        };
        let av = if header.avet_root_page > 0 {
            read_avet_index(header.avet_root_page, &*backend)?
        } else {
            std::collections::BTreeMap::new()
        };
        let v = if header.vaet_root_page > 0 {
            read_vaet_index(header.vaet_root_page, &*backend)?
        } else {
            std::collections::BTreeMap::new()
        };
        (e, a, av, v)
    };

    // Write new B+tree pages beyond current header.page_count (crash-safe: header not yet updated)
    let mut backend = self.backend.lock().unwrap();
    let mut next_free = header.page_count; // start after all existing content

    let (eavt_root, next_free2) =
        build_btree(eavt.into_iter(), &mut *backend, &self.page_cache, next_free)?;
    let (aevt_root, next_free3) =
        build_btree(aevt.into_iter(), &mut *backend, &self.page_cache, next_free2)?;
    let (avet_root, next_free4) =
        build_btree(avet.into_iter(), &mut *backend, &self.page_cache, next_free3)?;
    let (vaet_root, final_next_free) =
        build_btree(vaet.into_iter(), &mut *backend, &self.page_cache, next_free4)?;

    // Write v6 header (atomically upgrades version on-disk)
    let mut new_header = FileHeader::new(); // version=6
    new_header.page_count = final_next_free;
    new_header.node_count = header.node_count;
    new_header.last_checkpointed_tx_count = header.last_checkpointed_tx_count;
    new_header.eavt_root_page = eavt_root;
    new_header.aevt_root_page = aevt_root;
    new_header.avet_root_page = avet_root;
    new_header.vaet_root_page = vaet_root;
    new_header.index_checksum = header.index_checksum; // unchanged
    new_header.fact_page_format = header.fact_page_format;
    new_header.fact_page_count = num_fact_pages;

    let mut header_page = new_header.to_bytes();
    header_page.resize(PAGE_SIZE, 0);
    backend.write_page(0, &header_page)?;
    backend.sync()?;
    drop(backend);

    // Store last_checkpointed_tx_count
    self.last_checkpointed_tx_count = header.last_checkpointed_tx_count;

    // Wire CommittedFactReader
    let loader: Arc<dyn crate::storage::CommittedFactReader> =
        Arc::new(CommittedFactLoaderImpl {
            backend: self.backend.clone(),
            page_cache: self.page_cache.clone(),
            committed_fact_pages: self.committed_fact_pages.clone(),
            first_fact_page: 1,
        });
    self.storage.set_committed_reader(loader);

    // Wire CommittedIndexReader (using new B+tree roots)
    let index_reader: Arc<dyn crate::storage::CommittedIndexReader> =
        Arc::new(OnDiskIndexReader::new(
            self.backend.clone(),
            self.page_cache.clone(),
            eavt_root, aevt_root, avet_root, vaet_root,
        ));
    self.storage.set_committed_index_reader(index_reader);

    // Restore tx_counter from header
    self.storage.restore_tx_counter_from(header.last_checkpointed_tx_count);
    self.dirty = false;
    Ok(())
}
```

- [ ] **Step 5.5: Update v6 checksum-mismatch rebuild path**

Replace the `if needs_rebuild` block in `load()` with the v6 rebuild path:

```rust
if needs_rebuild {
    // v6 rebuild: stream committed facts, build sorted (Key, FactRef) iterators
    // directly — no full in-memory BTreeMap (that is what we're eliminating).
    let all_facts = {
        let backend = self.backend.lock().unwrap();
        crate::storage::packed_pages::read_all_from_pages(&*backend, 1, num_fact_pages)?
    };
    let (_, real_refs) = pack_facts(&all_facts, 1)?;

    use crate::storage::index::{AevtKey, AvetKey, EavtKey, VaetKey, encode_value};

    let mut eavt: Vec<(EavtKey, FactRef)> = all_facts.iter().zip(real_refs.iter())
        .map(|(f, &fr)| (EavtKey {
            entity: f.entity, attribute: f.attribute.clone(),
            valid_from: f.valid_from, valid_to: f.valid_to, tx_count: f.tx_count,
        }, fr))
        .collect();
    eavt.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut aevt: Vec<(AevtKey, FactRef)> = all_facts.iter().zip(real_refs.iter())
        .map(|(f, &fr)| (AevtKey {
            attribute: f.attribute.clone(), entity: f.entity,
            valid_from: f.valid_from, valid_to: f.valid_to, tx_count: f.tx_count,
        }, fr))
        .collect();
    aevt.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut avet: Vec<(AvetKey, FactRef)> = all_facts.iter().zip(real_refs.iter())
        .map(|(f, &fr)| (AvetKey {
            attribute: f.attribute.clone(), value_bytes: encode_value(&f.value),
            valid_from: f.valid_from, valid_to: f.valid_to,
            entity: f.entity, tx_count: f.tx_count,
        }, fr))
        .collect();
    avet.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    let mut vaet: Vec<(VaetKey, FactRef)> = all_facts.iter().zip(real_refs.iter())
        .filter_map(|(f, &fr)| {
            if let crate::graph::types::Value::Ref(target) = &f.value {
                Some((VaetKey {
                    ref_target: *target, attribute: f.attribute.clone(),
                    valid_from: f.valid_from, valid_to: f.valid_to,
                    source_entity: f.entity, tx_count: f.tx_count,
                }, fr))
            } else { None }
        })
        .collect();
    vaet.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));

    // Build B+tree from scratch (no RAM index created)
    let next_free_start = 1 + num_fact_pages;
    let mut backend = self.backend.lock().unwrap();
    let (eavt_root, nf1) = build_btree(eavt.into_iter(), &mut *backend, &self.page_cache, next_free_start)?;
    let (aevt_root, nf2) = build_btree(aevt.into_iter(), &mut *backend, &self.page_cache, nf1)?;
    let (avet_root, nf3) = build_btree(avet.into_iter(), &mut *backend, &self.page_cache, nf2)?;
    let (vaet_root, final_nf) = build_btree(vaet.into_iter(), &mut *backend, &self.page_cache, nf3)?;

    // Recompute checksum from pages
    let new_checksum = compute_page_checksum(&*backend, 1, num_fact_pages)?;

    // Write updated header
    let mut new_header = FileHeader::new();
    new_header.page_count = final_nf;
    new_header.node_count = all_facts.len() as u64;
    new_header.last_checkpointed_tx_count = self.last_checkpointed_tx_count;
    new_header.eavt_root_page = eavt_root;
    new_header.aevt_root_page = aevt_root;
    new_header.avet_root_page = avet_root;
    new_header.vaet_root_page = vaet_root;
    new_header.index_checksum = new_checksum;
    new_header.fact_page_format = FACT_PAGE_FORMAT_PACKED;
    new_header.fact_page_count = num_fact_pages;
    let mut header_page = new_header.to_bytes();
    header_page.resize(PAGE_SIZE, 0);
    backend.write_page(0, &header_page)?;
    backend.sync()?;
    drop(backend);

    let index_reader: Arc<dyn crate::storage::CommittedIndexReader> =
        Arc::new(OnDiskIndexReader::new(
            self.backend.clone(), self.page_cache.clone(),
            eavt_root, aevt_root, avet_root, vaet_root,
        ));
    self.storage.set_committed_index_reader(index_reader);
    self.storage.restore_tx_counter_from(self.last_checkpointed_tx_count);
} else if header.eavt_root_page != 0 {
    // Fast path: wire OnDiskIndexReader from header roots
    let index_reader: Arc<dyn crate::storage::CommittedIndexReader> =
        Arc::new(OnDiskIndexReader::new(
            self.backend.clone(), self.page_cache.clone(),
            header.eavt_root_page, header.aevt_root_page,
            header.avet_root_page, header.vaet_root_page,
        ));
    self.storage.set_committed_index_reader(index_reader);
}
```

- [ ] **Step 5.6: Build and run all tests**

```bash
cargo build 2>&1 | head -40
cargo test -- --nocapture 2>&1 | tail -20
```

Expected: all 301+ existing tests pass plus any new tests. Fix any remaining compile errors.

- [ ] **Step 5.7: Commit**

```bash
git add src/storage/persistent_facts.rs
git commit -m "feat(persistent_facts): v6 save/load/migrate/rebuild with on-disk B+tree indexes"
```

---

### Task 6: Integration tests

**Files:**
- Create: `tests/btree_v6_test.rs`

- [ ] **Step 6.1: Write integration tests**

```rust
//! Integration tests for Phase 6.5: on-disk B+tree indexes (file format v6).

use minigraf::OpenOptions;
use tempfile::NamedTempFile;

fn tmp_path() -> (NamedTempFile, String) {
    let f = NamedTempFile::new().unwrap();
    let p = f.path().to_str().unwrap().to_string();
    (f, p)
}

/// Create and checkpoint a file DB with N facts about entity `:eN` with attribute `:val`.
fn populate_and_checkpoint(n: usize, path: &str) {
    let db = OpenOptions::new().path(path).open().unwrap();
    let cmd: String = {
        let mut s = String::from("(transact [");
        for i in 0..n {
            s.push_str(&format!("[:e{} :val {}]", i, i));
        }
        s.push_str("])");
        s
    };
    db.execute(&cmd).unwrap();
    db.checkpoint().unwrap();
}

#[test]
fn test_v6_roundtrip_basic() {
    let (_tmp, path) = tmp_path();
    populate_and_checkpoint(10, &path);

    let db = OpenOptions::new().path(&path).open().unwrap();
    let result = db.execute("(query [:find ?v :where [:e0 :val ?v]])").unwrap();
    assert!(result.contains('0'), "entity e0 should have val 0; got: {}", result);
}

#[test]
fn test_v6_range_scan_across_leaves() {
    // 500 facts force multiple leaf pages per index
    let (_tmp, path) = tmp_path();
    populate_and_checkpoint(500, &path);

    let db = OpenOptions::new().path(&path).open().unwrap();
    let result = db.execute("(query [:find ?v :where [:e100 :val ?v]])").unwrap();
    assert!(result.contains("100"), "entity e100 should have val 100; got: {}", result);
}

#[test]
fn test_v6_pending_plus_committed_merge() {
    let (_tmp, path) = tmp_path();
    // Checkpoint 10 facts (committed)
    populate_and_checkpoint(10, &path);

    // Add 5 more (pending, in WAL)
    let db = OpenOptions::new().path(&path).open().unwrap();
    db.execute("(transact [[:e10 :val 10][:e11 :val 11][:e12 :val 12]])").unwrap();

    // Query must see both committed (e0) and pending (e10)
    let r0 = db.execute("(query [:find ?v :where [:e0 :val ?v]])").unwrap();
    assert!(r0.contains('0'), "committed e0 missing; got: {}", r0);

    let r10 = db.execute("(query [:find ?v :where [:e10 :val ?v]])").unwrap();
    assert!(r10.contains("10"), "pending e10 missing; got: {}", r10);
}

#[test]
fn test_v6_migration_from_v5_eager() {
    // Write a minimal v5 header directly to the .graph file (raw bytes),
    // then open with v6 code — migrate_v5_to_v6 must run and produce a v6 file.
    // We cannot use OpenOptions to produce a v5 file (v6 code always writes v6),
    // so we write the raw 72-byte v5 header at offset 0.
    let (_tmp, path) = tmp_path();

    // Create an empty v5 .graph file with only a header (no facts, no index pages)
    {
        let mut f = std::fs::OpenOptions::new().create(true).write(true).open(&path).unwrap();
        let mut page = vec![0u8; 4096]; // PAGE_SIZE
        page[0..4].copy_from_slice(b"MGRF");
        page[4..8].copy_from_slice(&5u32.to_le_bytes()); // version = 5
        page[8..16].copy_from_slice(&1u64.to_le_bytes()); // page_count = 1
        page[68] = 0x02; // fact_page_format = PACKED
        use std::io::Write;
        f.write_all(&page).unwrap();
    }

    // Open with v6 code — migration must run
    let db = OpenOptions::new().path(&path).open().unwrap();

    // Query on empty DB must return empty result (not an error)
    let result = db.execute("(query [:find ?e :where [?e :any :any]])").unwrap();
    drop(db);

    // Re-open and verify header was upgraded to v6
    let mut f = std::fs::File::open(&path).unwrap();
    let mut header_bytes = vec![0u8; 4096];
    use std::io::Read;
    f.read_exact(&mut header_bytes).unwrap();
    let version = u32::from_le_bytes(header_bytes[4..8].try_into().unwrap());
    assert_eq!(version, 6, "header must be upgraded from v5 to v6; got version={}", version);
}

#[test]
fn test_v6_explicit_checkpoint_clears_wal() {
    // After writing facts and calling checkpoint(), the WAL sidecar file must
    // be absent or empty, and subsequent queries must still work.
    //
    // Note: auto-checkpoint (wal_checkpoint_threshold) is out of scope for
    // Phase 6.5 and will be a separate feature. This test covers the existing
    // explicit checkpoint path to ensure it still works with v6 files.
    use std::path::Path;
    let (_tmp, path) = tmp_path();

    let db = OpenOptions::new().path(&path).open().unwrap();

    // Write 10 facts then explicitly checkpoint
    for i in 0..10 {
        let mut tx = db.begin_write().unwrap();
        tx.execute(&format!("(transact [[:e{} :val {}]])", i, i)).unwrap();
        tx.commit().unwrap();
    }
    db.checkpoint().unwrap();

    // WAL sidecar is named <db_path>.wal (per CLAUDE.md "WAL sidecar <db>.wal")
    let wal_path = format!("{}.wal", path);
    let wal_absent = !Path::new(&wal_path).exists();
    let wal_empty = std::fs::metadata(&wal_path).map(|m| m.len() == 0).unwrap_or(true);
    assert!(
        wal_absent || wal_empty,
        "WAL must be absent or empty after explicit checkpoint; path={} size={}",
        wal_path,
        std::fs::metadata(&wal_path).map(|m| m.len()).unwrap_or(0),
    );

    // Queries still return correct results
    let result = db.execute("(query [:find ?v :where [:e0 :val ?v]])").unwrap();
    assert!(result.contains('0'), "query must work after checkpoint; got: {}", result);
}

#[test]
fn test_v6_dead_pages_queries_correct_after_two_checkpoints() {
    let (_tmp, path) = tmp_path();

    // First checkpoint
    populate_and_checkpoint(20, &path);

    // Second checkpoint (adds new facts + new B+tree; old index pages are dead)
    let db = OpenOptions::new().path(&path).open().unwrap();
    db.execute("(transact [[:e20 :val 20][:e21 :val 21]])").unwrap();
    db.checkpoint().unwrap();

    // Re-open and verify queries are correct
    let db2 = OpenOptions::new().path(&path).open().unwrap();
    let r = db2.execute("(query [:find ?v :where [:e20 :val ?v]])").unwrap();
    assert!(r.contains("20"), "e20 should be queryable after second checkpoint; got: {}", r);
    let r0 = db2.execute("(query [:find ?v :where [:e0 :val ?v]])").unwrap();
    assert!(r0.contains('0'), "original e0 should still be visible; got: {}", r0);
}

#[test]
fn test_v6_checksum_mismatch_triggers_rebuild() {
    use std::io::{Read, Seek, SeekFrom, Write};

    let (_tmp, path) = tmp_path();
    populate_and_checkpoint(30, &path);

    // Corrupt the index_checksum field in the header (bytes 64..68)
    {
        let mut f = std::fs::OpenOptions::new().read(true).write(true).open(&path).unwrap();
        f.seek(SeekFrom::Start(64)).unwrap();
        f.write_all(&[0xFF, 0xFF, 0xFF, 0xFF]).unwrap();
    }

    // Re-open: checksum mismatch must trigger v6 rebuild path
    let db = OpenOptions::new().path(&path).open().unwrap();
    let result = db.execute("(query [:find ?v :where [:e0 :val ?v]])").unwrap();
    assert!(result.contains('0'), "queries must work after rebuild; got: {}", result);
}

#[test]
fn test_v6_reopen_close_reopen() {
    let (_tmp, path) = tmp_path();
    populate_and_checkpoint(100, &path);

    // Open, query, close, reopen, query again
    {
        let db = OpenOptions::new().path(&path).open().unwrap();
        let r = db.execute("(query [:find ?v :where [:e50 :val ?v]])").unwrap();
        assert!(r.contains("50"));
    }
    {
        let db = OpenOptions::new().path(&path).open().unwrap();
        let r = db.execute("(query [:find ?v :where [:e50 :val ?v]])").unwrap();
        assert!(r.contains("50"), "after reopen; got: {}", r);
    }
}
```

- [ ] **Step 6.2: Run integration tests**

```bash
cargo test --test btree_v6_test -- --nocapture
```

Expected: all 8 tests pass. Fix any failures before proceeding.

- [ ] **Step 6.3: Run full test suite**

```bash
cargo test -- --nocapture 2>&1 | tail -30
```

Expected: all 301+ existing tests plus 8 new integration tests pass (≥309 total). Fix any regressions.

- [ ] **Step 6.4: Commit**

```bash
git add tests/btree_v6_test.rs
git commit -m "test(btree_v6): integration tests for v6 roundtrip, migration, pending+committed merge"
```

---

### Task 7: Concurrent range-scan benchmark

**Files:**
- Modify: `benches/minigraf_bench.rs`

- [ ] **Step 7.1: Add concurrent range-scan benchmark**

Append before `criterion_group!` in `benches/minigraf_bench.rs`:

```rust
// ── Task 6.5: concurrent_btree_scan/ ─────────────────────────────────────────
// Measures throughput of simultaneous EAVT range scans against committed B+tree.
// A near-linear drop in per-thread throughput as N grows signals backend mutex
// contention and would trigger the per-page locking revisit noted in the spec.

fn bench_concurrent_btree_scan(c: &mut Criterion) {
    use std::sync::{Arc as StdArc, Barrier};
    use std::time::Instant;
    use tempfile::NamedTempFile;

    let mut group = c.benchmark_group("concurrent_btree_scan");
    group.sample_size(10);

    for &(label, n_threads) in &[("2", 2usize), ("4", 4), ("8", 8)] {
        // Pre-populate and checkpoint file DB so all facts are in committed B+tree.
        let tmp = Box::new(NamedTempFile::new().unwrap());
        let path = tmp.path().to_str().unwrap().to_string();
        helpers::populate_file(10_000, &path);
        // Open a handle shared by all threads
        let db = helpers::open_file_no_checkpoint(&path);
        let db = StdArc::clone(&db);

        group.bench_with_input(
            BenchmarkId::from_parameter(label),
            &n_threads,
            |b, &n_threads| {
                b.iter_custom(|iters| {
                    let barrier = StdArc::new(Barrier::new(n_threads + 1));
                    let mut handles = Vec::new();
                    for _ in 0..n_threads {
                        let db = StdArc::clone(&db);
                        let barrier = StdArc::clone(&barrier);
                        handles.push(std::thread::spawn(move || {
                            barrier.wait();
                            let start = Instant::now();
                            for _ in 0..iters {
                                // EAVT range scan: entity :e0 with attribute :val
                                db.execute("(query [:find ?v :where [:e0 :val ?v]])")
                                    .unwrap();
                            }
                            start.elapsed()
                        }));
                    }
                    barrier.wait();
                    handles
                        .into_iter()
                        .map(|h| h.join().unwrap())
                        .max()
                        .unwrap()
                });
            },
        );
        drop(tmp);
    }
    group.finish();
}
```

Add to `criterion_group!`:
```rust
criterion_group!(
    benches,
    bench_insert,
    bench_insert_file,
    bench_query,
    bench_time_travel,
    bench_recursion,
    bench_open,
    bench_checkpoint,
    bench_concurrent,
    bench_concurrent_file,
    bench_concurrent_btree_scan,  // new
);
```

- [ ] **Step 7.2: Verify benchmark compiles**

```bash
cargo bench --bench minigraf_bench -- --list 2>&1 | grep btree
```

Expected: `concurrent_btree_scan/2`, `concurrent_btree_scan/4`, `concurrent_btree_scan/8` listed.

- [ ] **Step 7.3: Commit**

```bash
git add benches/minigraf_bench.rs
git commit -m "bench: concurrent B+tree range-scan benchmark to detect mutex contention"
```

---

### Task 8: Final regression check and cleanup

**Files:**
- No new files

- [ ] **Step 8.1: Run full test suite**

```bash
cargo test 2>&1 | tail -10
```

Expected: all tests pass, test count ≥ 309. Note exact count.

- [ ] **Step 8.2: Run clippy**

```bash
cargo clippy -- -D warnings 2>&1 | head -40
```

Fix any warnings.

- [ ] **Step 8.3: Build docs**

```bash
cargo doc --no-deps 2>&1 | grep -i warning | head -20
```

Fix any documentation warnings.

- [ ] **Step 8.4: Final commit**

```bash
git add -A
git commit -m "chore: clippy and doc fixes for Phase 6.5"
```

- [ ] **Step 8.5: Invoke finishing skill**

Use `superpowers:finishing-a-development-branch` to decide whether to merge, PR, or tag.
