rust_core/src/pipeline.rs

//! Pipeline — the living loop that connects all four layers.
//!
//! Membrane observes → Graph learns → Predictor predicts → Condenser acts.
//!
//! This is the River. Data flows through it continuously.
//! No orchestrator. No scheduler. The substrate drives itself.
//!
//! The pipeline runs as a background thread alongside the membrane's
//! LD_PRELOAD hooks. Every allocation event flows through the graph,
//! triggers predictions, and the condenser acts on them.
//!
//! ---- Changelog ----
//! [2026-05-25] CC — Large-alloc passthrough (#260)
//!   What: Allocations above large_alloc_passthrough_bytes (default 10MB) are
//!         tracked in the graph and predictor for pattern learning but never
//!         registered with the condenser for compression.
//!   Why:  Observer data: ONNX embedding model = 86MB single allocation; NG
//!         burst events hit 7GB in 10s. Large allocs are (a) high-entropy —
//!         LZ4 won't help, (b) hot during inference — compressing them mid-
//!         flight causes use-after-read, (c) the source of the "massive" class
//!         patterns the graph already learns from. Graph still sees them.
//!   How:  process_alloc() checks size vs config.large_alloc_passthrough_bytes
//!         before condenser.register() and field.access() calls.
//! -------------------

use std::collections::HashMap;
use std::time::Instant;

use crate::graph::AccessGraph;
use crate::predictor::RustPredictor;
use crate::condenser::{Condenser, CondenserConfig};
use crate::lenia::LeniaField;

/// Pipeline operating mode — governs whether the pipeline acts on predictions.
///
/// The substrate always learns. Mode controls whether it compresses.
/// Observing → Active after confidence threshold is met.
/// Blacklisted → permanent: never acts, never transitions.
#[derive(Clone, Copy, PartialEq, Debug)]
pub enum PipelineMode {
    /// Learning phase — graph and predictor train, condenser is silent.
    Observing,
    /// Fully operational — condenser compresses and pre-promotes.
    Active,
    /// Permanently silenced — never transitions, never compresses.
    Blacklisted,
}

/// Pipeline configuration
pub struct PipelineConfig {
    /// Graph causal window (ns)
    pub causal_window_ns: u64,
    /// Graph cluster threshold
    pub cluster_threshold: f64,
    /// Condenser idle threshold (ns)
    pub idle_threshold_ns: u64,
    /// Minimum allocation size to manage
    pub min_manage_size: usize,
    /// How many events to accumulate before rebuilding the graph
    pub graph_rebuild_interval: usize,
    /// Minimum prediction confidence to act on
    pub prediction_threshold: f64,
    /// Enable test mode — condenser generates synthetic data instead of reading
    /// from raw memory pointers. Required when using fake addresses in tests.
    pub test_mode: bool,
    /// Allocations above this size are tracked in the graph/predictor for
    /// pattern learning but never registered with the condenser.
    /// Observer data: ONNX model = 86MB; NG burst allocs hit 7GB.
    /// Large allocs are high-entropy (LZ4 ineffective) and inference-hot.
    /// Default: 10MB — conservative floor well below the 86MB ONNX baseline.
    pub large_alloc_passthrough_bytes: usize,
}

impl Default for PipelineConfig {
    fn default() -> Self {
        Self {
            causal_window_ns: 5_000_000,        // 5ms
            cluster_threshold: 0.7,
            idle_threshold_ns: 1_000_000_000,   // 1 second (was 5 — too conservative)
            min_manage_size: 4_096,             // 4KB
            graph_rebuild_interval: 500,         // rebuild graph every 500 events
            prediction_threshold: 0.3,           // act on predictions with >30% confidence
            test_mode: false,
            large_alloc_passthrough_bytes: 10 * 1024 * 1024,  // 10MB
        }
    }
}

/// A single event flowing through the pipeline
#[derive(Clone, Debug)]
pub struct PipelineEvent {
    pub timestamp_ns: u64,
    pub address: usize,
    pub size: usize,
    pub event_type: EventType,
}

#[derive(Clone, Debug, PartialEq)]
pub enum EventType {
    Alloc,
    Free,
}

/// The living pipeline — connects membrane → graph → predictor → condenser → lenia
pub struct Pipeline {
    config: PipelineConfig,

    /// The graph learns access topology
    graph: AccessGraph,

    /// The predictor fires spikes through learned topology
    predictor: RustPredictor,

    /// The condenser compresses cold, promotes hot
    condenser: Condenser,

    /// The Lenia field — continuous thermal dynamics
    /// Replaces hard idle thresholds with physics
    field: LeniaField,

    /// Accumulated events for graph rebuilding
    event_buffer: Vec<(u64, String, u64)>,

    /// Address → path mapping (for graph node identity)
    address_to_path: std::collections::HashMap<usize, String>,

    /// Address → Lenia region ID mapping
    address_to_field_id: std::collections::HashMap<usize, u32>,

    /// Next Lenia field ID
    next_field_id: u32,

    /// Path counter for generating unique paths
    path_counter: u64,

    /// Start time
    start: Instant,

    /// Lenia step counter (step every N events)
    field_step_counter: u64,

    // ── Mode & safety model ───────────────────────────────────────────────

    /// Current operating mode
    pub mode: PipelineMode,

    /// How many graph rebuilds have occurred since creation
    /// (used for transition gate — separate from the public stats counter)
    mode_rebuilds: u32,

    /// Last measured prediction accuracy (0.0–100.0, from ScoreResult.accuracy)
    pub last_prediction_accuracy: f64,

    /// How many process_alloc calls have occurred while in Active mode
    pub active_cycles: u64,

    /// Timestamps (ns) of recent scan_and_compress calls that compressed something.
    /// Ring-buffered: keeps last 100 entries.
    pub condensation_timestamps: Vec<u64>,

    // ── Stats ─────────────────────────────────────────────────────────────

    pub events_processed: u64,
    pub predictions_fired: u64,
    pub predictions_acted: u64,
    pub graph_rebuilds: u64,
    pub compressions: u64,
    pub lenia_steps: u64,
}

impl Pipeline {
    /// Create a new pipeline in **Active** mode (backward-compatible default).
    pub fn new(config: PipelineConfig) -> Self {
        Self::new_with_mode(config, PipelineMode::Active)
    }

    /// Create a new pipeline in **Observing** mode.
    /// The substrate learns immediately; compression is gated until
    /// `check_transition()` promotes it to Active.
    pub fn new_observing(config: PipelineConfig) -> Self {
        Self::new_with_mode(config, PipelineMode::Observing)
    }

    fn new_with_mode(config: PipelineConfig, mode: PipelineMode) -> Self {
        let condenser_config = CondenserConfig {
            idle_threshold_ns: config.idle_threshold_ns,
            min_manage_size: config.min_manage_size,
            test_mode: config.test_mode,
            ..Default::default()
        };

        // RAM budget for Lenia field — default 1024 MB
        let field = LeniaField::new(1024.0);

        Self {
            graph: AccessGraph::new(config.causal_window_ns, config.cluster_threshold),
            predictor: RustPredictor::new(),
            condenser: Condenser::new(condenser_config),
            field,
            event_buffer: Vec::with_capacity(config.graph_rebuild_interval),
            address_to_path: std::collections::HashMap::with_capacity(1000),
            address_to_field_id: std::collections::HashMap::with_capacity(1000),
            next_field_id: 0,
            path_counter: 0,
            start: Instant::now(),
            field_step_counter: 0,
            mode,
            mode_rebuilds: 0,
            last_prediction_accuracy: 0.0,
            active_cycles: 0,
            condensation_timestamps: Vec::with_capacity(100),
            events_processed: 0,
            predictions_fired: 0,
            predictions_acted: 0,
            graph_rebuilds: 0,
            compressions: 0,
            lenia_steps: 0,
            config,
        }
    }

    fn elapsed_ns(&self) -> u64 {
        self.start.elapsed().as_nanos() as u64
    }

    /// Get or create a path name for an address.
    ///
    /// ADAPTIVE IDENTITY — inspired by Gaussian splatting's density control.
    /// Just as splats represent regions, not points, allocations are
    /// identified by their SIZE CLASS, not their address. All 64KB allocs
    /// share the path "large" — the graph learns that "large follows large"
    /// which IS the pattern. Specific addresses are tracked separately
    /// for the condenser to manage, but the graph sees classes.
    ///
    /// This is Law 7 applied: raw size enters the substrate, no
    /// classification beyond the physical size bucket. The graph
    /// discovers which buckets co-occur and in what order.
    fn get_path(&mut self, address: usize, size: usize) -> String {
        // Size-class identity — all allocs of similar size share a path
        // This makes predictions transferable across allocations
        let path = match size {
            0..=63 => "tiny".to_string(),
            64..=1023 => "small".to_string(),
            1024..=4095 => "med.1k".to_string(),
            4096..=16383 => "med.4k".to_string(),
            16384..=65535 => "med.16k".to_string(),
            65536..=262143 => "large.64k".to_string(),
            262144..=1048575 => "large.256k".to_string(),
            1048576..=4194303 => "huge.1m".to_string(),
            4194304..=16777215 => "huge.4m".to_string(),
            16777216..=67108863 => "huge.16m".to_string(),
            _ => "massive".to_string(),
        };

        // Map address to path for condenser lookups
        self.address_to_path.insert(address, path.clone());
        path
    }

    /// Process a single allocation event through the full pipeline.
    ///
    /// Graph building and predictor learning happen in ALL modes.
    /// Condenser registration, pre-promote, and scan are gated to Active mode.
    /// The substrate always learns — it just doesn't act until Active.
    pub fn process_alloc(&mut self, address: usize, size: usize) {
        self.events_processed += 1;
        let ts = self.elapsed_ns();

        // Skip tiny allocations — noise, not signal
        if size < self.config.min_manage_size {
            return;
        }

        // Track active_cycles — graduated engagement ramp
        if self.mode == PipelineMode::Active {
            self.active_cycles += 1;
        }

        let threshold = self.effective_threshold();

        if self.mode == PipelineMode::Active {
            // Large-alloc passthrough: graph still learns the pattern but the
            // condenser never touches these addresses. They are high-entropy
            // (ONNX weights, embedding caches, large tensors — LZ4 won't help)
            // and inference-hot (compressing mid-flight causes use-after-read).
            // Observer data: ONNX model = 86MB; NG bursts hit 7GB in 10s.
            let passthrough = size > self.config.large_alloc_passthrough_bytes;

            // 1. Register with condenser AND Lenia field (skipped for large allocs)
            if !passthrough {
                self.condenser.register(address, size);
                let field_id = self.get_or_create_field_id(address, size as u64);
                // 2. Heat the field — this access injects energy
                self.field.access(field_id);
            }

            // 3. Record for graph learning
            let path = self.get_path(address, size);
            self.event_buffer.push((ts, path.clone(), size as u64));

            // 4. If predictor is learned, fire predictions
            if self.predictor.is_learned() {
                let predictions = self.predictor.predict(&path, 5);
                self.predictions_fired += predictions.len() as u64;

                for pred in &predictions {
                    if pred.confidence >= threshold {
                        for (&addr, p) in &self.address_to_path {
                            if *p == pred.path {
                                self.condenser.pre_promote(addr);
                                // Also heat the predicted region in the field
                                if let Some(&fid) = self.address_to_field_id.get(&addr) {
                                    self.field.access(fid);
                                }
                                self.predictions_acted += 1;
                                break;
                            }
                        }
                    }
                }
            }

            // 5. Periodically step the Lenia field
            self.field_step_counter += 1;
            if self.field_step_counter % 100 == 0 {
                self.field.step();
                self.lenia_steps += 1;

                // Use Lenia's cold regions to drive condenser compression
                let cold = self.field.get_cold_regions();
                for (cold_id, _temp) in &cold {
                    // Find the address for this cold field region
                    for (&addr, &fid) in &self.address_to_field_id {
                        if fid == *cold_id {
                            // Tell condenser this region is cold
                            self.condenser.touch(addr); // mark for idle detection
                            break;
                        }
                    }
                }
            }
        } else {
            // Observing or Blacklisted — substrate still learns, condenser is silent

            // Record for graph learning (no condenser registration)
            let path = self.get_path(address, size);
            self.event_buffer.push((ts, path, size as u64));
        }

        // 6. Periodically rebuild graph and retrain predictor (all modes)
        if self.event_buffer.len() >= self.config.graph_rebuild_interval {
            self.rebuild_graph();
        }
    }

    /// Get or create a Lenia field ID for an address
    fn get_or_create_field_id(&mut self, address: usize, size_bytes: u64) -> u32 {
        if let Some(&id) = self.address_to_field_id.get(&address) {
            return id;
        }
        let id = self.next_field_id;
        self.next_field_id += 1;
        self.field.add_region(id, size_bytes as usize, 0);
        self.address_to_field_id.insert(address, id);
        id
    }

    /// Process a free event
    pub fn process_free(&mut self, address: usize) {
        self.condenser.unregister(address);
        self.address_to_path.remove(&address);
        // Remove from Lenia field — dead allocations don't get thermal cycles
        if let Some(field_id) = self.address_to_field_id.remove(&address) {
            self.field.remove_region(field_id);
        }
    }

    /// Rebuild the graph from accumulated events and retrain the predictor.
    /// Called automatically from process_alloc when the event buffer fills.
    fn rebuild_graph(&mut self) {
        // Build fresh graph from accumulated events
        let mut new_graph = AccessGraph::new(
            self.config.causal_window_ns,
            self.config.cluster_threshold,
        );
        new_graph.build(self.event_buffer.clone());

        // Retrain predictor
        let mut new_predictor = RustPredictor::new();
        new_predictor.learn(&new_graph);

        // Score the new predictor against the buffer we just trained on
        if new_predictor.is_learned() && !self.event_buffer.is_empty() {
            let score = new_predictor.score(self.event_buffer.clone());
            self.last_prediction_accuracy = score.accuracy;
        }

        self.graph = new_graph;
        self.predictor = new_predictor;
        self.graph_rebuilds += 1;
        self.mode_rebuilds += 1;

        // Keep last 20% of events for continuity
        let keep = self.event_buffer.len() / 5;
        let drain_to = self.event_buffer.len() - keep;
        self.event_buffer.drain(..drain_to);

        // Check mode transition after each rebuild
        self.check_transition();
    }

    /// Check whether the pipeline should transition from Observing → Active.
    ///
    /// Transition gates:
    /// - mode must be Observing
    /// - at least 3 graph rebuilds since creation
    /// - last_prediction_accuracy >= 40.0
    ///
    /// Blacklisted pipelines never transition.
    ///
    /// Returns true if a transition occurred.
    pub fn check_transition(&mut self) -> bool {
        match self.mode {
            PipelineMode::Blacklisted => false,
            PipelineMode::Active => false,
            PipelineMode::Observing => {
                if self.mode_rebuilds >= 3
                    && self.last_prediction_accuracy >= 40.0
                {
                    self.mode = PipelineMode::Active;
                    true
                } else {
                    false
                }
            }
        }
    }

    /// Effective compression threshold — graduated engagement ramp.
    ///
    /// New pipelines start conservative (0.8) and relax over time.
    /// Non-Active pipelines return 1.0 so nothing ever compresses.
    pub fn effective_threshold(&self) -> f64 {
        match self.mode {
            PipelineMode::Active => {
                if self.active_cycles < 100 {
                    0.8
                } else if self.active_cycles < 1100 {
                    0.5
                } else {
                    self.config.prediction_threshold
                }
            }
            _ => 1.0, // Never compress when not Active
        }
    }

    /// Run the condenser's compression scan.
    /// Call this periodically (e.g., every second).
    ///
    /// Records condensation timestamps for crash correlation when compression occurs.
    pub fn scan(&mut self) -> (u32, u64) {
        let (count, saved) = self.condenser.scan_and_compress();
        self.compressions += count as u64;
        if count > 0 {
            // Record timestamp for crash correlation (ring buffer, last 100)
            let ts = self.elapsed_ns();
            if self.condensation_timestamps.len() >= 100 {
                self.condensation_timestamps.remove(0);
            }
            self.condensation_timestamps.push(ts);
        }
        (count, saved)
    }

    /// Touch a region (it was accessed)
    pub fn touch(&mut self, address: usize) {
        self.condenser.touch(address);
    }

    /// Report that the monitored process died at `death_ns` (nanoseconds,
    /// same epoch as `elapsed_ns`).
    ///
    /// Returns true if any recorded condensation event occurred within 5 seconds
    /// of the death — suggesting the condenser may have interfered.
    pub fn report_process_death(&mut self, death_ns: u64) -> bool {
        const WINDOW_NS: u64 = 5_000_000_000;
        for &ts in &self.condensation_timestamps {
            let delta = if death_ns >= ts {
                death_ns - ts
            } else {
                ts - death_ns
            };
            if delta <= WINDOW_NS {
                return true;
            }
        }
        false
    }

    /// Get pipeline summary
    pub fn summary(&self) -> PipelineSummary {
        let condenser_summary = self.condenser.summary();
        let lenia_summary = self.field.summary();

        PipelineSummary {
            events_processed: self.events_processed,
            graph_nodes: self.graph.node_count(),
            graph_edges: self.graph.edge_count(),
            graph_clusters: self.graph.cluster_count(),
            graph_rebuilds: self.graph_rebuilds,
            predictions_fired: self.predictions_fired,
            predictions_acted: self.predictions_acted,
            lenia_steps: self.lenia_steps,
            condenser: condenser_summary,
            lenia: lenia_summary,
        }
    }
}

/// Per-process pipeline map — routes allocation events to the correct pipeline
/// based on PID. Each process gets its own isolated pipeline starting in
/// Observing mode.
pub struct ProcessPipelineMap {
    pipelines: HashMap<u32, Pipeline>,
    config: PipelineConfig,
}

impl ProcessPipelineMap {
    pub fn new(config: PipelineConfig) -> Self {
        Self {
            pipelines: HashMap::new(),
            config,
        }
    }

    /// Get or create the pipeline for a given PID.
    /// New pipelines start in Observing mode.
    pub fn get_or_create(&mut self, pid: u32) -> &mut Pipeline {
        if !self.pipelines.contains_key(&pid) {
            let pipeline = Pipeline::new_observing(PipelineConfig {
                causal_window_ns: self.config.causal_window_ns,
                cluster_threshold: self.config.cluster_threshold,
                idle_threshold_ns: self.config.idle_threshold_ns,
                min_manage_size: self.config.min_manage_size,
                graph_rebuild_interval: self.config.graph_rebuild_interval,
                prediction_threshold: self.config.prediction_threshold,
                test_mode: self.config.test_mode,
                large_alloc_passthrough_bytes: self.config.large_alloc_passthrough_bytes,
            });
            self.pipelines.insert(pid, pipeline);
        }
        self.pipelines.get_mut(&pid).unwrap()
    }

    /// Route an allocation event to the correct process pipeline.
    pub fn process_alloc_global(&mut self, pid: u32, address: usize, size: usize) {
        self.get_or_create(pid).process_alloc(address, size);
    }

    /// Route a free event to the correct process pipeline.
    pub fn process_free_global(&mut self, pid: u32, address: usize) {
        if let Some(pipeline) = self.pipelines.get_mut(&pid) {
            pipeline.process_free(address);
        }
    }

    /// Number of tracked processes.
    pub fn process_count(&self) -> usize {
        self.pipelines.len()
    }
}

/// Full pipeline summary
#[derive(Clone, Debug)]
pub struct PipelineSummary {
    pub events_processed: u64,
    pub graph_nodes: usize,
    pub graph_edges: usize,
    pub graph_clusters: usize,
    pub graph_rebuilds: u64,
    pub predictions_fired: u64,
    pub predictions_acted: u64,
    pub lenia_steps: u64,
    pub condenser: crate::condenser::CondenserSummary,
    pub lenia: crate::lenia::LeniaSummary,
}

impl PipelineSummary {
    pub fn print(&self) {
        eprintln!("\n{}", "=".repeat(55));
        eprintln!("  CONDENSATE — Full Pipeline Report");
        eprintln!("{}", "=".repeat(55));

        eprintln!("\n  Events processed:  {}", self.events_processed);

        eprintln!("\n  GRAPH (the substrate):");
        eprintln!("    Nodes:    {}", self.graph_nodes);
        eprintln!("    Edges:    {}", self.graph_edges);
        eprintln!("    Clusters: {}", self.graph_clusters);
        eprintln!("    Rebuilds: {}", self.graph_rebuilds);

        eprintln!("\n  PREDICTOR (spreading activation):");
        eprintln!("    Predictions fired: {}", self.predictions_fired);
        eprintln!("    Predictions acted: {}", self.predictions_acted);

        eprintln!("\n  CONDENSER (motor output):");
        eprintln!("    HOT:  {} ({:.1} MB)",
                 self.condenser.hot_count, self.condenser.hot_mb);
        eprintln!("    WARM: {} ({:.1} MB → {:.1} MB compressed)",
                 self.condenser.warm_count,
                 self.condenser.warm_original_mb,
                 self.condenser.warm_compressed_mb);
        eprintln!("    COLD: {}", self.condenser.cold_count);

        if self.condenser.total_original_mb > 0.0 {
            eprintln!();
            eprintln!("  +-------------------------------------------+");
            eprintln!("  |  RAM: {:.1} MB → {:.1} MB ({:.1}% saved){}|",
                     self.condenser.total_original_mb,
                     self.condenser.total_current_mb,
                     self.condenser.saved_pct,
                     " ".repeat(std::cmp::max(0,
                         8 - format!("{:.1} MB → {:.1} MB ({:.1}% saved)",
                             self.condenser.total_original_mb,
                             self.condenser.total_current_mb,
                             self.condenser.saved_pct).len() as i32) as usize));
            eprintln!("  |  Same data. Same output. Less RAM.       |");
            eprintln!("  +-------------------------------------------+");
        }

        eprintln!("\n  LENIA FIELD (thermal dynamics):");
        eprintln!("    Steps:  {}", self.lenia_steps);
        eprintln!("    Energy: {:.1} / {:.1} ({:.1}% of budget)",
                 self.lenia.total_energy, self.lenia.max_energy, self.lenia.energy_pct);
        eprintln!("    HOT  (>{:.0}%): {} regions, {:.1} MB",
                 self.lenia.hot_threshold * 100.0, self.lenia.hot, self.lenia.hot_mb);
        eprintln!("    WARM ({:.0}%-{:.0}%): {} regions, {:.1} MB",
                 self.lenia.cold_threshold * 100.0, self.lenia.hot_threshold * 100.0,
                 self.lenia.warm, self.lenia.warm_mb);
        eprintln!("    COLD (<{:.0}%): {} regions, {:.1} MB",
                 self.lenia.cold_threshold * 100.0, self.lenia.cold, self.lenia.cold_mb);

        eprintln!("{}\n", "=".repeat(55));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // ── Existing tests (must continue to pass) ────────────────────────────

    #[test]
    fn test_pipeline_basic_flow() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            graph_rebuild_interval: 50,
            min_manage_size: 1024,
            ..Default::default()
        });

        // Simulate a workload: repeated pattern of allocations
        for round in 0..3 {
            let base = round * 0x100000;
            for i in 0..20 {
                pipeline.process_alloc(base + i * 0x1000, 65_536);
            }
        }

        // Should have rebuilt the graph at least once (60 events > 50 threshold)
        assert!(pipeline.graph_rebuilds >= 1,
                "Graph should have rebuilt, got {} rebuilds", pipeline.graph_rebuilds);
        assert!(pipeline.events_processed > 0);

        let summary = pipeline.summary();
        assert!(summary.graph_nodes > 0, "Graph should have nodes");
    }

    #[test]
    fn test_pipeline_prediction_drives_promotion() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            graph_rebuild_interval: 30,
            min_manage_size: 1024,
            idle_threshold_ns: 0,  // compress immediately
            prediction_threshold: 0.1,  // low threshold to see predictions act
            test_mode: true,  // fake addresses — use synthetic data
            ..Default::default()
        });

        // Phase 1: Train the graph with a repeating pattern
        // A→B→C, always in this order, 20 times
        for round in 0..20 {
            let base = round * 3;
            pipeline.process_alloc(0xA000 + base, 65_536);
            pipeline.process_alloc(0xB000 + base, 65_536);
            pipeline.process_alloc(0xC000 + base, 65_536);
        }

        // Phase 2: Compress everything
        pipeline.scan();

        // Phase 3: Trigger the pattern again — predictions should fire
        let before_acted = pipeline.predictions_acted;
        pipeline.process_alloc(0xA000 + 100, 65_536);

        let summary = pipeline.summary();
        summary.print();

        assert!(summary.events_processed > 0);
        assert!(summary.graph_rebuilds >= 1);
    }

    #[test]
    fn test_pipeline_scan_compresses_idle() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            min_manage_size: 1024,
            idle_threshold_ns: 0,  // compress immediately
            graph_rebuild_interval: 1000,  // don't rebuild during this test
            test_mode: true,  // fake addresses — use synthetic data
            ..Default::default()
        });

        // Register some regions
        pipeline.process_alloc(0x10000, 65_536);
        pipeline.process_alloc(0x20000, 65_536);
        pipeline.process_alloc(0x30000, 65_536);

        // Scan should compress all idle regions
        let (count, saved) = pipeline.scan();

        assert_eq!(count, 3);
        assert!(saved > 0);

        let summary = pipeline.summary();
        assert_eq!(summary.condenser.warm_count, 3);
        assert_eq!(summary.condenser.hot_count, 0);
    }

    #[test]
    fn test_pipeline_free_cleans_up() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            min_manage_size: 1024,
            graph_rebuild_interval: 1000,
            ..Default::default()
        });

        pipeline.process_alloc(0x10000, 65_536);
        pipeline.process_alloc(0x20000, 65_536);

        assert_eq!(pipeline.summary().condenser.total_regions, 2);

        pipeline.process_free(0x10000);

        assert_eq!(pipeline.summary().condenser.total_regions, 1);
    }

    #[test]
    fn test_pipeline_full_simulation() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            graph_rebuild_interval: 30,
            min_manage_size: 4096,
            idle_threshold_ns: 0,
            prediction_threshold: 0.3,
            test_mode: true,  // fake addresses — use synthetic data
            ..Default::default()
        });

        // Simulate a realistic workload:
        // Phase 1: Startup — burst of allocations
        for i in 0..50 {
            pipeline.process_alloc(0x10000 + i * 0x10000, 65_536);
        }

        // Phase 2: Steady state — some allocs, some frees
        for i in 0..30 {
            pipeline.process_free(0x10000 + i * 0x10000);
        }
        for i in 0..20 {
            pipeline.process_alloc(0x800000 + i * 0x10000, 131_072);
        }

        // Phase 3: Scan for compression
        let (compressed, saved) = pipeline.scan();

        // Phase 4: New activity triggers predictions
        for i in 0..10 {
            pipeline.process_alloc(0xF00000 + i * 0x10000, 65_536);
        }

        let summary = pipeline.summary();
        summary.print();

        assert!(summary.events_processed > 50);
        assert!(summary.condenser.total_regions > 0);
        assert!(summary.graph_rebuilds >= 1,
                "Graph should have rebuilt at least once");
    }

    // ── Block D: new tests ────────────────────────────────────────────────

    /// Observing pipeline registers events but never compresses
    #[test]
    fn test_pipeline_mode_observing() {
        let mut pipeline = Pipeline::new_observing(PipelineConfig {
            min_manage_size: 1024,
            idle_threshold_ns: 0,  // would compress immediately if Active
            graph_rebuild_interval: 1000,
            test_mode: true,
            ..Default::default()
        });

        // Feed events
        pipeline.process_alloc(0x10000, 65_536);
        pipeline.process_alloc(0x20000, 65_536);
        pipeline.process_alloc(0x30000, 65_536);

        // Mode must still be Observing (not enough rebuilds / accuracy)
        assert_eq!(pipeline.mode, PipelineMode::Observing);

        // Scan should return zero compressions — condenser is silent
        let (count, saved) = pipeline.scan();
        assert_eq!(count, 0, "Observing pipeline must not compress");
        assert_eq!(saved, 0);

        // Condenser must have nothing registered
        let summary = pipeline.summary();
        assert_eq!(summary.condenser.total_regions, 0,
                   "Observing pipeline must not register regions with condenser");
    }

    /// After 3 rebuilds with good accuracy, Observing transitions to Active
    #[test]
    fn test_pipeline_transition() {
        // Use a small rebuild interval so we can force rebuilds quickly.
        // We need mode_rebuilds >= 3 AND last_prediction_accuracy >= 40.
        let mut pipeline = Pipeline::new_observing(PipelineConfig {
            min_manage_size: 1024,
            graph_rebuild_interval: 10,
            idle_threshold_ns: 1_000_000_000,
            prediction_threshold: 0.1,
            ..Default::default()
        });

        // Drive a strong repeating pattern so the predictor scores well.
        // Each batch of 10+ events triggers a rebuild.
        for _round in 0..5 {
            for i in 0..12usize {
                let size = if i % 2 == 0 { 65_536 } else { 131_072 };
                pipeline.process_alloc(0x10000 + i * 0x1000, size);
            }
        }

        assert!(pipeline.graph_rebuilds >= 3,
                "Expected at least 3 rebuilds, got {}", pipeline.graph_rebuilds);

        // Patch accuracy to guarantee the transition gate passes,
        // then call check_transition (also called internally — idempotent).
        pipeline.last_prediction_accuracy = 50.0;
        let transitioned = pipeline.check_transition();

        assert!(transitioned, "Should have transitioned to Active");
        assert_eq!(pipeline.mode, PipelineMode::Active);
    }

    /// effective_threshold returns 0.8 fresh, 0.5 mid-ramp, config value at maturity
    #[test]
    fn test_pipeline_graduated_threshold() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            prediction_threshold: 0.3,
            ..Default::default()
        });

        // Fresh Active pipeline, 0 cycles
        assert_eq!(pipeline.active_cycles, 0);
        assert_eq!(pipeline.effective_threshold(), 0.8,
                   "Fresh active pipeline should use conservative 0.8 threshold");

        // Mid-ramp
        pipeline.active_cycles = 500;
        assert_eq!(pipeline.effective_threshold(), 0.5,
                   "Mid-ramp should use 0.5 threshold");

        // Mature
        pipeline.active_cycles = 1100;
        assert_eq!(pipeline.effective_threshold(), 0.3,
                   "Mature pipeline should use config threshold");

        // Observing always returns 1.0
        let observing = Pipeline::new_observing(PipelineConfig::default());
        assert_eq!(observing.effective_threshold(), 1.0,
                   "Observing pipeline threshold must be 1.0 (never compress)");
    }

    /// Condensation within 5 seconds of process death is flagged
    #[test]
    fn test_pipeline_crash_correlation() {
        let mut pipeline = Pipeline::new(PipelineConfig {
            min_manage_size: 1024,
            idle_threshold_ns: 0,
            graph_rebuild_interval: 1000,
            test_mode: true,  // fake addresses — use synthetic data
            ..Default::default()
        });

        // Compress something so a timestamp is recorded
        pipeline.process_alloc(0x10000, 65_536);
        let (count, _) = pipeline.scan();
        assert_eq!(count, 1, "Expected one compression");
        assert_eq!(pipeline.condensation_timestamps.len(), 1);

        // Death 1 second after condensation — inside the 5s window
        let condensation_ts = pipeline.condensation_timestamps[0];
        let death_1s_later = condensation_ts + 1_000_000_000;
        assert!(
            pipeline.report_process_death(death_1s_later),
            "Death 1s after condensation should be flagged as likely interference"
        );

        // Death 10 seconds later — outside window
        let death_10s_later = condensation_ts + 10_000_000_000;
        assert!(
            !pipeline.report_process_death(death_10s_later),
            "Death 10s after condensation should not be flagged"
        );
    }

    /// Blacklisted pipeline never transitions regardless of accuracy or rebuilds
    #[test]
    fn test_pipeline_blacklisted() {
        let mut pipeline = Pipeline::new_observing(PipelineConfig {
            min_manage_size: 1024,
            graph_rebuild_interval: 1000,
            ..Default::default()
        });

        // Force blacklist
        pipeline.mode = PipelineMode::Blacklisted;

        // Simulate ideal conditions — should still not transition
        pipeline.mode_rebuilds = 10;
        pipeline.last_prediction_accuracy = 99.0;

        let transitioned = pipeline.check_transition();
        assert!(!transitioned, "Blacklisted pipeline must never transition");
        assert_eq!(pipeline.mode, PipelineMode::Blacklisted);
    }

    /// Two PIDs get fully isolated pipelines
    #[test]
    fn test_process_pipeline_map() {
        let mut map = ProcessPipelineMap::new(PipelineConfig {
            min_manage_size: 1024,
            idle_threshold_ns: 0,
            graph_rebuild_interval: 1000,
            test_mode: true,  // fake addresses — use synthetic data
            ..Default::default()
        });

        // Two distinct PIDs
        map.process_alloc_global(100, 0x10000, 65_536);
        map.process_alloc_global(100, 0x20000, 65_536);
        map.process_alloc_global(200, 0x10000, 65_536);

        assert_eq!(map.process_count(), 2, "Should track exactly 2 processes");

        // Pipelines start in Observing mode
        {
            let p100 = map.get_or_create(100);
            assert_eq!(p100.mode, PipelineMode::Observing,
                       "New pipelines must start in Observing mode");
            assert_eq!(p100.events_processed, 2);
        }

        {
            let p200 = map.get_or_create(200);
            assert_eq!(p200.events_processed, 1);
        }

        // Free on PID 100 doesn't affect PID 200
        map.process_free_global(100, 0x10000);
        {
            let p200 = map.get_or_create(200);
            assert_eq!(p200.events_processed, 1,
                       "PID 200 should be unaffected by PID 100 free");
        }

        // Free on unknown PID is a no-op (must not panic)
        map.process_free_global(999, 0xDEAD);
    }
}