feat(fusion): expose observability metrics via HTTP

Three rates the operator actually needs to judge whether fusion is
doing its job: how often sources agree, how often windows get dropped
under LRU pressure, and which source dominates the firing mix.

- `core/detection/metrics.rs` holds a lock-free `FusionMetrics` struct
  with relaxed-order `AtomicU64` counters for total emits, multi-source
  emits, LRU-pressure evictions, and per-source ingress counts. Shared
  via `Arc` — readers never block the orchestrator task.
- `FusionMetricsSnapshot` precomputes `agreed_rate`,
  `window_drop_rate`, and a `PerSourceRate` view so the frontend never
  re-implements the formulas and drifts.
- Orchestrator instruments three call sites:
  - `handle_detection` → `record_fire(source)` before canonicalization
    so the per-source counter reflects raw ingress.
  - `emit_fused` → `record_emit(active_source_count)` so multi-vs-solo
    is tagged by the actual count used in the fused event.
  - Path B dedup `put` → when the LRU is at capacity and the key is
    absent, bump `windows_evicted` and log
    `DetectionLog::FusionWindowEvicted` with the evicted key. Wires
    up the log variant that was previously declared but unused.
- `AppServices` owns the `Arc<FusionMetrics>`. `system.rs` threads it
  into `DetectionOrchestrator::new`; `http_server.rs` injects it as
  Actix `app_data`.
- New `adapter/http/fusion.rs` scope at `/api/fusion/metrics` returning
  the snapshot JSON. Future `/api/fusion/explain/{ip}` lands in the
  same scope.

Tests: 5 new (zero-traffic snapshot is all-zeros, agreed_rate reflects
multi/total ratio, window_drop_rate isolates evictions from emits,
per-source rates sum to 1.0, record_fire routes to the correct
bucket). 224 pass total. clippy --package net-guardia -- -D warnings
clean.

Closes B-region Q-6.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

net-guardia/src/adapter/http/fusion.rs

@@ -0,0 +1,21 @@
+//! HTTP surface for fusion-layer observability. Reads shared atomic
+//! counters maintained by the detection orchestrator — handlers here
+//! never touch orchestrator state, so a hung dashboard cannot stall
+//! the detection pipeline.
+
+use std::sync::Arc;
+
+use actix_web::{HttpResponse, Responder, Scope, web};
+
+use crate::core::detection::metrics::FusionMetrics;
+
+pub fn initialize() -> Scope {
+    web::scope("/fusion").route("/metrics", web::get().to(get_metrics))
+}
+
+/// `GET /api/fusion/metrics` — lock-free snapshot of fusion counters and
+/// derived rates. Drives the operator dashboard's "how well is fusion
+/// working on my network?" view.
+async fn get_metrics(metrics: web::Data<Arc<FusionMetrics>>) -> impl Responder {
+    HttpResponse::Ok().json(metrics.snapshot())
+}

net-guardia/src/adapter/http/mod.rs

@@ -4,6 +4,7 @@ pub mod audit;
 pub mod auth;
 pub mod default;
 pub mod filter;
+pub mod fusion;
 pub mod health;
 pub mod logs;
 pub mod ml;

net-guardia/src/core/detection/metrics.rs

@@ -0,0 +1,223 @@
+//! Lock-free fusion observability counters. The orchestrator bumps these
+//! on ingress / emit / eviction; HTTP handlers (and eventually the admin
+//! dashboard) read atomic snapshots without touching orchestrator state.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use serde::Serialize;
+
+use crate::model::event::DetectionSource;
+
+/// Relaxed ordering is enough for counters: readers tolerate arbitrary
+/// interleaving, and no counter's value gates access to another memory
+/// location. Anything stronger would just waste fence instructions on the
+/// hot packet path without adding any real invariant.
+const ORDER: Ordering = Ordering::Relaxed;
+
+/// Atomic counters maintained by the detection orchestrator. Shared via
+/// `Arc` with the HTTP metrics handler so dashboards can read without
+/// blocking the orchestrator task.
+#[derive(Debug, Default)]
+pub struct FusionMetrics {
+    /// Total fused emits (single-source + multi-source combined).
+    total_emits: AtomicU64,
+    /// Fused emits whose final `active_source_count` was ≥ 2 (i.e. fusion
+    /// actually fired across multiple sources rather than single-source solo).
+    multi_source_emits: AtomicU64,
+    /// Fusion windows evicted under LRU pressure before they could emit.
+    /// Kept separate from shutdown-drain drops — those are legitimate.
+    windows_evicted: AtomicU64,
+    /// Per-source event count. Increments once per ingress detection
+    /// regardless of whether the event fires a fused emit downstream.
+    ml_fires: AtomicU64,
+    suricata_fires: AtomicU64,
+    beaconing_fires: AtomicU64,
+    correlation_fires: AtomicU64,
+}
+
+impl FusionMetrics {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Record an ingress detection from the given source. Called before
+    /// fusion-window bookkeeping so per-source counters reflect raw
+    /// volume, not what survives dedup.
+    pub fn record_fire(&self, source: DetectionSource) {
+        let counter = match source {
+            DetectionSource::ML => &self.ml_fires,
+            DetectionSource::Suricata => &self.suricata_fires,
+            DetectionSource::Beaconing => &self.beaconing_fires,
+            DetectionSource::Correlation => &self.correlation_fires,
+        };
+        counter.fetch_add(1, ORDER);
+    }
+
+    /// Record a fused emit. `source_count` is the number of distinct
+    /// sources that contributed to this emit — 1 for single-source,
+    /// 2..=4 when fusion actually agreed.
+    pub fn record_emit(&self, source_count: usize) {
+        self.total_emits.fetch_add(1, ORDER);
+        if source_count >= 2 {
+            self.multi_source_emits.fetch_add(1, ORDER);
+        }
+    }
+
+    /// Record a fusion-window eviction that happened before the window
+    /// could emit. Shutdown-drain drops are not counted here.
+    pub fn record_eviction(&self) {
+        self.windows_evicted.fetch_add(1, ORDER);
+    }
+
+    /// Take an atomic snapshot of every counter and derive the three
+    /// rate figures the dashboard surfaces.
+    pub fn snapshot(&self) -> FusionMetricsSnapshot {
+        let total_emits = self.total_emits.load(ORDER);
+        let multi_source_emits = self.multi_source_emits.load(ORDER);
+        let windows_evicted = self.windows_evicted.load(ORDER);
+
+        let ml = self.ml_fires.load(ORDER);
+        let suricata = self.suricata_fires.load(ORDER);
+        let beaconing = self.beaconing_fires.load(ORDER);
+        let correlation = self.correlation_fires.load(ORDER);
+
+        let agreed_rate = ratio(multi_source_emits, total_emits);
+        let drop_denominator = total_emits + windows_evicted;
+        let window_drop_rate = ratio(windows_evicted, drop_denominator);
+
+        let total_fires = ml + suricata + beaconing + correlation;
+        let per_source_fire_rate = PerSourceRate {
+            ml: ratio(ml, total_fires),
+            suricata: ratio(suricata, total_fires),
+            beaconing: ratio(beaconing, total_fires),
+            correlation: ratio(correlation, total_fires),
+        };
+
+        FusionMetricsSnapshot {
+            total_emits,
+            multi_source_emits,
+            agreed_rate,
+            windows_evicted,
+            window_drop_rate,
+            per_source_fires: PerSourceCount {
+                ml,
+                suricata,
+                beaconing,
+                correlation,
+            },
+            per_source_fire_rate,
+        }
+    }
+}
+
+/// Return `numerator / denominator` as `f64`, or `0.0` when the
+/// denominator is zero. Saves every rate caller from an `if denom == 0`
+/// rewrite of the same guard.
+fn ratio(numerator: u64, denominator: u64) -> f64 {
+    if denominator == 0 {
+        0.0
+    } else {
+        numerator as f64 / denominator as f64
+    }
+}
+
+/// Wire-format snapshot consumed by `GET /api/fusion/metrics`. Derived
+/// fields (`agreed_rate`, `window_drop_rate`, `per_source_fire_rate`)
+/// are precomputed server-side so the UI doesn't have to re-implement
+/// the formulas and drift.
+#[derive(Debug, Clone, Serialize)]
+pub struct FusionMetricsSnapshot {
+    pub total_emits: u64,
+    pub multi_source_emits: u64,
+    /// `multi_source_emits / total_emits`.
+    pub agreed_rate: f64,
+    pub windows_evicted: u64,
+    /// `windows_evicted / (windows_evicted + total_emits)`.
+    pub window_drop_rate: f64,
+    pub per_source_fires: PerSourceCount,
+    pub per_source_fire_rate: PerSourceRate,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct PerSourceCount {
+    pub ml: u64,
+    pub suricata: u64,
+    pub beaconing: u64,
+    pub correlation: u64,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct PerSourceRate {
+    pub ml: f64,
+    pub suricata: f64,
+    pub beaconing: f64,
+    pub correlation: f64,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn zero_traffic_snapshot_reports_zero_rates() {
+        let m = FusionMetrics::new();
+        let s = m.snapshot();
+        assert_eq!(s.total_emits, 0);
+        assert_eq!(s.agreed_rate, 0.0);
+        assert_eq!(s.window_drop_rate, 0.0);
+        assert_eq!(s.per_source_fire_rate.ml, 0.0);
+    }
+
+    #[test]
+    fn agreed_rate_reflects_multi_source_ratio() {
+        let m = FusionMetrics::new();
+        m.record_emit(1); // single-source
+        m.record_emit(2); // multi
+        m.record_emit(3); // multi
+        m.record_emit(1); // single
+        let s = m.snapshot();
+        assert_eq!(s.total_emits, 4);
+        assert_eq!(s.multi_source_emits, 2);
+        assert!((s.agreed_rate - 0.5).abs() < 1e-9);
+    }
+
+    #[test]
+    fn window_drop_rate_isolates_evictions_from_emits() {
+        let m = FusionMetrics::new();
+        for _ in 0..9 {
+            m.record_emit(1);
+        }
+        m.record_eviction(); // 1 evicted / 10 total tracked
+        let s = m.snapshot();
+        assert_eq!(s.windows_evicted, 1);
+        assert!((s.window_drop_rate - 0.1).abs() < 1e-9);
+    }
+
+    #[test]
+    fn per_source_fire_rate_sums_to_one_when_nonzero() {
+        let m = FusionMetrics::new();
+        m.record_fire(DetectionSource::ML);
+        m.record_fire(DetectionSource::ML);
+        m.record_fire(DetectionSource::Suricata);
+        m.record_fire(DetectionSource::Beaconing);
+        let s = m.snapshot();
+        let sum = s.per_source_fire_rate.ml
+            + s.per_source_fire_rate.suricata
+            + s.per_source_fire_rate.beaconing
+            + s.per_source_fire_rate.correlation;
+        assert!((sum - 1.0).abs() < 1e-9, "per-source rates must sum to 1, got {sum}");
+        assert!((s.per_source_fire_rate.ml - 0.5).abs() < 1e-9);
+    }
+
+    #[test]
+    fn record_fire_routes_to_correct_source_bucket() {
+        let m = FusionMetrics::new();
+        m.record_fire(DetectionSource::Suricata);
+        m.record_fire(DetectionSource::Correlation);
+        let s = m.snapshot();
+        assert_eq!(s.per_source_fires.suricata, 1);
+        assert_eq!(s.per_source_fires.correlation, 1);
+        assert_eq!(s.per_source_fires.ml, 0);
+        assert_eq!(s.per_source_fires.beaconing, 0);
+    }
+}

net-guardia/src/core/detection/mod.rs

@@ -1,3 +1,4 @@
 pub mod beaconing;
 pub mod fusion_math;
+pub mod metrics;
 pub mod orchestrator;

net-guardia/src/core/detection/orchestrator.rs

@@ -8,6 +8,7 @@ use tokio::sync::mpsc;
 use tokio::time::interval;
 
 use super::fusion_math::{FusionWindowLengths, fused_confidence};
+use super::metrics::FusionMetrics;
 use crate::infrastructure::communication_manager::CommunicationManager;
 use crate::infrastructure::geoip::GeoIpService;
 use crate::model::detection::attack_type::translate;
@@ -67,6 +68,7 @@ pub struct DetectionOrchestrator {
     rx: mpsc::Receiver<DetectionEvent>,
     comm: Arc<CommunicationManager>,
     geoip: Option<Arc<GeoIpService>>,
+    metrics: Arc<FusionMetrics>,
     // Enrichment state
     src_ip_counts: lru::LruCache<String, u32>,
     repeat_tracker: lru::LruCache<String, Instant>,
@@ -83,11 +85,13 @@ impl DetectionOrchestrator {
         rx: mpsc::Receiver<DetectionEvent>,
         comm: Arc<CommunicationManager>,
         geoip: Option<Arc<GeoIpService>>,
+        metrics: Arc<FusionMetrics>,
     ) -> Self {
         Self {
             rx,
             comm,
             geoip,
+            metrics,
             // SAFETY: NonZero::new on a non-zero literal is infallible.
             src_ip_counts: LruCache::new(NonZero::new(10_000).unwrap()),
             repeat_tracker: LruCache::new(NonZero::new(5_000).unwrap()),
@@ -122,6 +126,10 @@ impl DetectionOrchestrator {
     }
 
     async fn handle_detection(&mut self, mut event: DetectionEvent) {
+        // Count every ingress event per-source before dedup — this is the
+        // raw firing rate, independent of whether the event survives to emit.
+        self.metrics.record_fire(event.source);
+
         // Canonicalize the raw attack_type so Suricata's "brute-force"
         // classtype and ML's "Brute Force" class name land on the same dedup
         // key — the precondition for cross-source fusion. Keep the original
@@ -187,6 +195,22 @@ impl DetectionOrchestrator {
         // Path B: brand-new key (or expired dedup). Emit single-source,
         // open a fusion window sized by this source.
         let fusion_window = Duration::from_secs(self.fusion_windows.for_source(event.source));
+
+        // Detect LRU-pressure eviction: if the dedup map is already at capacity
+        // and this key wasn't present, inserting will evict the least-recently-
+        // used entry silently. That's a real lost-signal event; count it and
+        // warn so the operator sees sustained-attack saturation.
+        let cap = self.dedup.cap().get();
+        let was_full = self.dedup.len() >= cap;
+        let key_was_absent = self.dedup.peek(&key).is_none();
+        if was_full && key_was_absent {
+            self.metrics.record_eviction();
+            log!(DetectionLog::FusionWindowEvicted {
+                key_src: key.0.clone(),
+                key_type: key.1.clone(),
+            });
+        }
+
         self.dedup.put(
             key.clone(),
             DedupEntry {
@@ -233,6 +257,8 @@ impl DetectionOrchestrator {
             threat_event.active_source_count,
         ));
 
+        self.metrics.record_emit(threat_event.active_source_count);
+
         self.publish_fusion_audit(trigger_event, fused, &per_source_samples)
             .await;
 

net-guardia/src/infrastructure/app_services.rs

@@ -6,6 +6,7 @@ use crossbeam::queue::SegQueue;
 use macros::log;
 use tokio::sync::oneshot;
 
+use crate::core::detection::metrics::FusionMetrics;
 use crate::core::ml::adapter::ModelSourceState;
 use crate::core::ml::alert::MLAlert;
 use crate::core::ml::drift_detector::DriftDetector;
@@ -38,6 +39,7 @@ pub struct AppServices {
     pub ml_inference: Arc<Inference>,
     pub ml_engine: Arc<Engine>,
     pub flow_statistics: Arc<FlowStatistics>,
+    pub fusion_metrics: Arc<FusionMetrics>,
     shutdowns: SegQueue<oneshot::Sender<()>>,
 }
 
@@ -123,6 +125,7 @@ impl AppServices {
         ));
 
         let flow_statistics = Arc::new(FlowStatistics::new(ml_engine.clone()));
+        let fusion_metrics = Arc::new(FusionMetrics::new());
 
         Ok(Self {
             health: Arc::new(health),
@@ -130,6 +133,7 @@ impl AppServices {
             ml_inference,
             ml_engine,
             flow_statistics,
+            fusion_metrics,
             shutdowns: SegQueue::new(),
         })
     }

net-guardia/src/infrastructure/http_server.rs

@@ -10,7 +10,7 @@ use macros::log;
 
 use crate::adapter::ebpf::EbpfServices;
 use crate::adapter::http::{
-    acl, api_keys, audit as audit_api, auth, default, filter, health as health_api, logs as logs_api, ml,
+    acl, api_keys, audit as audit_api, auth, default, filter, fusion, health as health_api, logs as logs_api, ml,
     notification as notification_api, rate_limit as rate_limit_api, report as report_api, setup as setup_api, soar,
     stats, system as system_api,
 };
@@ -209,6 +209,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
     let ml_alert = params.app_services.ml_alert.clone();
     let ml_engine = params.app_services.ml_engine.clone();
     let ml_inference = params.app_services.ml_inference.clone();
+    let fusion_metrics = params.app_services.fusion_metrics.clone();
     let flow_statistics = params.app_services.flow_statistics.clone();
     let drop_monitor = params.ebpf_services.drop_monitor.clone();
     let app_config = params.app_config;
@@ -248,6 +249,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
             .app_data(web::Data::from(ml_alert.clone()))
             .app_data(web::Data::from(ml_engine.clone()))
             .app_data(web::Data::from(ml_inference.clone()))
+            .app_data(web::Data::from(fusion_metrics.clone()))
             .app_data(web::Data::from(flow_statistics.clone()))
             .app_data(web::Data::from(drop_monitor.clone()))
             .app_data(web::Data::from(db.clone() as Arc<dyn AppRepo>))
@@ -277,6 +279,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
                     .service(stats::initialize())
                     .service(health_api::initialize())
                     .service(ml::initialize())
+                    .service(fusion::initialize())
                     .service(system_api::initialize())
                     .service(soar::initialize())
                     .service(notification_api::initialize())

net-guardia/src/infrastructure/system.rs

@@ -246,7 +246,12 @@ impl System {
 
         // Start detection orchestrator (dedup + enrichment + source attribution)
         let (detection_tx, detection_rx) = mpsc::channel::<DetectionEvent>(1024);
-        let orchestrator = DetectionOrchestrator::new(detection_rx, self.comm.clone(), self.geoip.clone());
+        let orchestrator = DetectionOrchestrator::new(
+            detection_rx,
+            self.comm.clone(),
+            self.geoip.clone(),
+            self.app_services.fusion_metrics.clone(),
+        );
         orchestrator.start();
 
         // Clone detection_tx for correlation engine and beaconing detector