clawhdf5/crates/clawhdf5-bench/src/bin/consolidation_efficiency.rs

//! Consolidation Efficiency Benchmark (Track 8.5)
//!
//! Measures how consolidation improves retrieval quality by removing low-importance
//! noise records and promoting high-importance signal records to durable memory tiers.
//!
//! # Test Design
//! 1. Insert 10 "signal" records with distinctive keywords (accessed 15+ times each)
//! 2. Insert 990 "noise" records with generic text
//! 3. Run 10 queries targeting signal records BEFORE consolidation → measure Hit@K
//! 4. Simulate consolidation (access pattern raises signal records' decay scores)
//! 5. Run consolidation cycle (working_capacity=100, noise gets evicted)
//! 6. Run same 10 queries AFTER consolidation → measure Hit@K
//! 7. Report improvement delta, eviction count, latency before/after
//!
//! Also measures consolidation cycle time at 100, 1K, 10K, 100K records.
//!
//! # Usage
//! ```
//! cargo run --release --bin consolidation_efficiency
//! ```

use std::time::Instant;

use clawhdf5_agent::bm25::BM25Index;
use clawhdf5_agent::consolidation::{
    ConsolidationConfig, ConsolidationEngine, TrustedSource, UntrustedSource,
};
use clawhdf5_agent::hybrid::hybrid_search;

const EMBEDDING_DIM: usize = 384;

// Signal records have these distinctive words in their content
const SIGNAL_KEYWORDS: &[&str] = &[
    "NEXUS_ALPHA_PROTOCOL",
    "QUANTUM_BEACON_ARRAY",
    "STELLAR_DRIFT_ANOMALY",
    "VORTEX_PRIME_SEQUENCE",
    "HELIX_OMEGA_FRAMEWORK",
    "PRISM_DELTA_SCHEMA",
    "APEX_NOVA_PIPELINE",
    "ZENITH_CORE_MATRIX",
    "AURORA_FLUX_TOPOLOGY",
    "CIPHER_SURGE_VECTOR",
];

const NOISE_TEMPLATE: &[&str] = &[
    "system architecture distributed",
    "memory vector embedding agent",
    "knowledge search retrieval temporal",
    "semantic episodic working consolidation",
    "importance activation cosine similarity",
    "hybrid keyword BM25 index session",
    "context token chunk overlap inference",
    "pipeline latency throughput benchmark",
    "performance Rust async parallel concurrent",
    "thread atomic signal noise data processing",
];

fn make_signal_content(keyword_idx: usize) -> String {
    format!(
        "Critical system record {} containing the identifier {} for \
         the advanced distributed processing protocol requiring immediate retrieval \
         and high-priority memory consolidation preservation.",
        keyword_idx, SIGNAL_KEYWORDS[keyword_idx]
    )
}

fn make_noise_content(idx: usize) -> String {
    let template = NOISE_TEMPLATE[idx % NOISE_TEMPLATE.len()];
    format!(
        "Record number {} in the memory store. Content: {} \
         with additional padding words for realistic text length.",
        idx, template
    )
}

fn make_embedding(seed: usize) -> Vec<f32> {
    (0..EMBEDDING_DIM)
        .map(|i| ((seed * 31 + i * 17) % 1000) as f32 / 1000.0 - 0.5)
        .collect()
}

// ---------------------------------------------------------------------------
// BM25 search helper (operates on ConsolidationEngine records)
// ---------------------------------------------------------------------------

fn search_engine(engine: &ConsolidationEngine, query: &str, k: usize) -> Vec<(usize, f32)> {
    let records = engine.records();
    if records.is_empty() {
        return Vec::new();
    }
    let docs: Vec<String> = records.iter().map(|r| r.chunk.clone()).collect();
    let vectors: Vec<Vec<f32>> = records.iter().map(|r| r.embedding.clone()).collect();
    let tombstones: Vec<u8> = vec![0u8; records.len()];
    let zero_emb = vec![0.0f32; EMBEDDING_DIM];

    let bm25 = BM25Index::build(&docs, &tombstones);
    hybrid_search(
        &zero_emb,
        query,
        &vectors,
        &docs,
        &tombstones,
        &bm25,
        0.0,
        1.0,
        k,
    )
}

// ---------------------------------------------------------------------------
// Latency helper
// ---------------------------------------------------------------------------

fn percentile_us(latencies_ns: &mut [u64], p: usize) -> f64 {
    latencies_ns.sort_unstable();
    let idx = (p * latencies_ns.len() / 100).min(latencies_ns.len().saturating_sub(1));
    latencies_ns.get(idx).copied().unwrap_or(0) as f64 / 1000.0
}

// ---------------------------------------------------------------------------
// Part 1: Retrieval quality before/after consolidation
// ---------------------------------------------------------------------------

struct QualityResult {
    hit1: u32,
    hit5: u32,
    hit10: u32,
    mrr: f64,
    latency_ns: Vec<u64>,
    record_count: usize,
}

fn measure_retrieval_quality(engine: &ConsolidationEngine) -> QualityResult {
    let mut hit1 = 0u32;
    let mut hit5 = 0u32;
    let mut hit10 = 0u32;
    let mut mrr = 0.0f64;
    let mut latency_ns = Vec::new();

    for (qi, keyword) in SIGNAL_KEYWORDS.iter().enumerate() {
        // Find all signal record indices in the engine's current records
        let records = engine.records();
        let signal_indices: std::collections::HashSet<usize> = records
            .iter()
            .enumerate()
            .filter(|(_, r)| r.chunk.contains(keyword))
            .map(|(i, _)| i)
            .collect();

        if signal_indices.is_empty() {
            // Signal record was evicted — count as miss
            continue;
        }

        let query = format!("{keyword} critical system record {qi}");

        let t0 = Instant::now();
        let results = search_engine(engine, &query, 10);
        let elapsed_ns = t0.elapsed().as_nanos() as u64;
        latency_ns.push(elapsed_ns);

        for (rank, (idx, _)) in results.iter().enumerate() {
            if signal_indices.contains(idx) {
                hit10 += 1;
                if rank < 5 {
                    hit5 += 1;
                }
                if rank == 0 {
                    hit1 += 1;
                }
                mrr += 1.0 / (rank + 1) as f64;
                break;
            }
        }
    }

    QualityResult {
        hit1,
        hit5,
        hit10,
        mrr,
        latency_ns,
        record_count: engine.records().len(),
    }
}

fn print_quality(label: &str, q: &QualityResult, n_queries: usize) {
    let n = n_queries as f64;
    let mut lat = q.latency_ns.clone();
    println!("{label}:");
    println!("  Records in store:  {}", q.record_count);
    println!(
        "  Hit@1:  {:5.1}%  Hit@5:  {:5.1}%  Hit@10: {:5.1}%  MRR: {:.4}",
        q.hit1 as f64 / n * 100.0,
        q.hit5 as f64 / n * 100.0,
        q.hit10 as f64 / n * 100.0,
        q.mrr / n
    );
    println!(
        "  Latency (BM25 over {} records):  avg={:.1} µs  p50={:.1} µs  p95={:.1} µs",
        q.record_count,
        lat.iter().sum::<u64>() as f64 / lat.len().max(1) as f64 / 1000.0,
        percentile_us(&mut lat, 50),
        percentile_us(&mut lat, 95),
    );
}

fn run_quality_benchmark() {
    println!("## Part 1: Retrieval Quality Before vs. After Consolidation");
    println!();
    println!("Setup:");
    println!(
        "  Signal records: {} (distinctive keywords, accessed 15x each)",
        SIGNAL_KEYWORDS.len()
    );
    println!("  Noise records:  990 (generic text, zero accesses)");
    println!("  Total initial:  1000");
    println!("  Working capacity: 100  (triggers eviction of 900 lowest-decay records)");
    println!("  Episodic threshold: 0.6  (high-importance records promoted)");
    println!();

    let config = ConsolidationConfig {
        working_capacity: 100,
        episodic_capacity: 10_000,
        working_to_episodic_threshold: 0.5, // easier to promote
        ..ConsolidationConfig::default()
    };
    let mut engine = ConsolidationEngine::new(config);
    let now = 1_000_000.0f64;

    // Insert signal records using Correction source (highest importance)
    let mut signal_ids: Vec<u64> = Vec::new();
    for i in 0..SIGNAL_KEYWORDS.len() {
        let chunk = make_signal_content(i);
        let embedding = make_embedding(i * 1000);
        let id = engine.add_trusted_memory(chunk, embedding, TrustedSource::Correction, now);
        signal_ids.push(id);
    }

    // Insert noise records
    for i in 0..990 {
        let chunk = make_noise_content(i);
        let embedding = make_embedding(i + 100);
        engine.add_trusted_memory(
            chunk,
            embedding,
            TrustedSource::System,
            now + i as f64 * 0.1,
        );
    }

    println!("  → Inserted {} records total", engine.records().len());
    println!();

    // Measure BEFORE consolidation
    let before = measure_retrieval_quality(&engine);
    print_quality("BEFORE consolidation", &before, SIGNAL_KEYWORDS.len());
    println!();

    // Simulate access: bump signal records many times (promotes via access_count)
    let access_time = now + 10_000.0;
    for &id in &signal_ids {
        for _ in 0..15 {
            engine.access_memory(id, access_time);
        }
    }

    // Run consolidation
    let t0 = Instant::now();
    engine.consolidate(now + 20_000.0);
    let consolidation_ms = t0.elapsed().as_secs_f64() * 1000.0;

    let stats = engine.get_stats();

    println!("Consolidation cycle:");
    println!("  Time:       {:.2} ms", consolidation_ms);
    println!(
        "  Remaining:  {} records (was 1000)",
        engine.records().len()
    );
    println!("  Evictions:  {}", stats.total_evictions);
    println!(
        "  Promotions: {} (Working→Episodic/Semantic)",
        stats.total_promotions
    );
    println!(
        "  Tier dist:  Working={}, Episodic={}, Semantic={}",
        stats.working_count, stats.episodic_count, stats.semantic_count
    );
    println!();

    // Measure AFTER consolidation
    let after = measure_retrieval_quality(&engine);
    print_quality("AFTER consolidation", &after, SIGNAL_KEYWORDS.len());
    println!();

    // Delta report
    let n = SIGNAL_KEYWORDS.len() as f64;
    let delta_hit1 = (after.hit1 as f64 - before.hit1 as f64) / n * 100.0;
    let delta_hit5 = (after.hit5 as f64 - before.hit5 as f64) / n * 100.0;
    let delta_mrr = after.mrr / n - before.mrr / n;
    let speedup = before.latency_ns.iter().sum::<u64>() as f64
        / after.latency_ns.iter().sum::<u64>().max(1) as f64;

    println!("Delta (after - before):");
    println!(
        "  Hit@1:  {:+.1}%   Hit@5:  {:+.1}%   MRR:  {:+.4}",
        delta_hit1, delta_hit5, delta_mrr
    );
    println!(
        "  Search speedup: {:.1}x faster ({} records → {} records)",
        speedup, before.record_count, after.record_count
    );
    println!();
}

// ---------------------------------------------------------------------------
// Part 2: Consolidation cycle time at various scales
// ---------------------------------------------------------------------------

fn run_cycle_time_benchmark() {
    println!("## Part 2: Consolidation Cycle Time at Various Scales");
    println!();
    println!(
        "{:>8}  {:>12}  {:>14}  {:>14}",
        "Records", "Cycle Time", "Evictions", "Promotions"
    );
    println!("{}", "-".repeat(54));

    for &n in &[100usize, 1_000, 10_000, 100_000] {
        let config = ConsolidationConfig {
            working_capacity: (n / 2).max(50),
            episodic_capacity: n * 10,
            ..ConsolidationConfig::default()
        };
        let mut engine = ConsolidationEngine::new(config);
        let now = 1_000_000.0f64;

        for i in 0..n {
            let chunk = make_noise_content(i);
            let embedding = make_embedding(i);
            engine.add_memory(chunk, embedding, UntrustedSource::User, now + i as f64);
        }

        // Warmup
        engine.consolidate(now + 1_000_000.0);
        let stats_before = engine.get_stats();

        // Re-fill
        for i in n..(n * 2) {
            let chunk = make_noise_content(i);
            let embedding = make_embedding(i);
            engine.add_memory(chunk, embedding, UntrustedSource::User, now + i as f64);
        }

        // Timed consolidation
        let t0 = Instant::now();
        engine.consolidate(now + 2_000_000.0);
        let elapsed_us = t0.elapsed().as_micros();

        let stats = engine.get_stats();
        let evictions = stats.total_evictions - stats_before.total_evictions;
        let promotions = stats.total_promotions - stats_before.total_promotions;

        let n_label = match n {
            100 => "100".to_owned(),
            1_000 => "1K".to_owned(),
            10_000 => "10K".to_owned(),
            100_000 => "100K".to_owned(),
            _ => n.to_string(),
        };

        let time_str = if elapsed_us >= 1000 {
            format!("{:.2} ms", elapsed_us as f64 / 1000.0)
        } else {
            format!("{} µs", elapsed_us)
        };

        println!(
            "{:>8}  {:>12}  {:>14}  {:>14}",
            n_label, time_str, evictions, promotions
        );
    }
    println!();
}

// ---------------------------------------------------------------------------
// Part 3: Memory reduction (records evicted vs. retained)
// ---------------------------------------------------------------------------

fn run_memory_reduction_benchmark() {
    println!("## Part 3: Memory Reduction After Consolidation");
    println!();
    println!("Working capacity = 20% of initial records. Signal records accessed 15x.");
    println!();
    println!(
        "{:>8}  {:>10}  {:>10}  {:>10}  {:>12}",
        "Initial", "Remaining", "Eviction%", "Signal OK?", "Records ÷"
    );
    println!("{}", "-".repeat(58));

    for &n in &[100usize, 1_000, 10_000] {
        let signal_count = 5.min(n / 10);
        let noise_count = n - signal_count;

        let config = ConsolidationConfig {
            working_capacity: (n / 5).max(10),
            episodic_capacity: n * 10,
            working_to_episodic_threshold: 0.5,
            ..ConsolidationConfig::default()
        };
        let mut engine = ConsolidationEngine::new(config);
        let now = 1_000_000.0f64;

        let mut signal_ids = Vec::new();
        for i in 0..signal_count {
            let chunk = make_signal_content(i % SIGNAL_KEYWORDS.len());
            let emb = make_embedding(i * 999);
            let id = engine.add_trusted_memory(chunk, emb, TrustedSource::Correction, now);
            signal_ids.push(id);
        }
        for i in 0..noise_count {
            let chunk = make_noise_content(i);
            let emb = make_embedding(i + 200);
            engine.add_trusted_memory(chunk, emb, TrustedSource::System, now + i as f64 * 0.1);
        }

        // Access signal records heavily
        for &id in &signal_ids {
            for _ in 0..15 {
                engine.access_memory(id, now + 10_000.0);
            }
        }

        let before_count = engine.records().len();
        engine.consolidate(now + 20_000.0);
        let after_count = engine.records().len();

        // Check all signal records survived
        let signal_survived = signal_ids.iter().all(|&id| engine.get_by_id(id).is_some());

        // How many times fewer records there are. Not a measured speedup —
        // Part 1 measures search latency before and after.
        let speedup = before_count as f64 / after_count.max(1) as f64;

        println!(
            "{:>8}  {:>10}  {:>9.1}%  {:>10}  {:>9.1}x",
            n,
            after_count,
            (1.0 - after_count as f64 / before_count.max(1) as f64) * 100.0,
            if signal_survived { "YES ✓" } else { "NO ✗" },
            speedup
        );
    }
    println!();
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

fn main() {
    println!("=================================================================");
    println!("  Consolidation Efficiency Benchmark");
    println!("=================================================================");
    println!();

    run_quality_benchmark();
    run_cycle_time_benchmark();
    run_memory_reduction_benchmark();

    println!("=================================================================");
    println!("  Summary");
    println!("=================================================================");
    println!();
    println!("Consolidation improves retrieval by:");
    println!("  1. Evicting low-importance noise records → smaller BM25 search space");
    println!("  2. Promoting high-importance/frequently-accessed records to Episodic/Semantic");
    println!("     (these tiers are never evicted, guaranteeing durable recall)");
    println!("  3. Reducing search latency proportional to record reduction");
    println!();
    println!(
        "Cycle time grows a little faster than linearly: 100 records ~microseconds, 100K records ~tens of ms."
    );
    println!("Signal records with Correction source + high access_count survive eviction.");
}