feat: model hot-reload with ArcSwap + file watcher

Zero-downtime ONNX model swap:
  - Inference holds ArcSwap<MLModels> instead of Arc<MLModels>
  - Lock-free readers: inference continues unblocked during reload
  - swap_models() atomically replaces both AE and classifier as one unit
  - Circuit breaker resets on successful model swap

Model watcher (new core/ml/model_watcher.rs):
  - notify crate watches models/ directory for .onnx file changes
  - 5-second debounce handles both models landing simultaneously
  - Background load → tract into_optimized → into_runnable → ArcSwap store
  - On failure: old model stays active, error logged
  - Started in system.rs alongside other background tasks

Dependencies: arc-swap 1, notify 7

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

net-guardia/Cargo.toml

@@ -53,6 +53,8 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus
 # Architecture
 async-trait = "0.1"
 dashmap = "6"
+arc-swap = "1"
+notify = { version = "7", default-features = false, features = ["macos_kqueue"] }
 
 # Utilities
 parking_lot = { workspace = true }

net-guardia/src/core/ml/engine.rs

@@ -72,6 +72,10 @@ impl Engine {
         }
     }
 
+    pub fn inference_pipeline(&self) -> &Arc<Inference> {
+        &self.inference_pipeline
+    }
+
     /// xsk_manager calls this per queue_id; with symmetric hash each queue has its own tracker.
     pub fn tracker(&self, queue_id: u32) -> &ThreadTracker {
         &self.trackers[queue_id as usize % self.trackers.len()]

net-guardia/src/core/ml/inference.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
 
+use arc_swap::ArcSwap;
 use macros::log;
 use tract_onnx::prelude::*;
 
@@ -23,15 +24,12 @@ const CIRCUIT_BREAKER_WINDOW_SECS: u64 = 60;
 const CIRCUIT_BREAKER_COOLDOWN_SECS: u64 = 120;
 
 pub struct Inference {
-    pub models: Arc<MLModels>,
+    models: ArcSwap<MLModels>,
     pub config: Arc<MLInferenceConfig>,
     c2_class_idx: Option<usize>,
     normal_class_idx: Option<usize>,
-    /// Consecutive failure count within the current window.
     failure_count: AtomicU32,
-    /// Timestamp (epoch secs) of first failure in current window. 0 = no failures.
     failure_window_start: AtomicU64,
-    /// Timestamp (epoch secs) when circuit breaker tripped. 0 = not tripped.
     circuit_open_since: AtomicU64,
 }
 
@@ -40,7 +38,7 @@ impl Inference {
         let c2_class_idx = Self::find_label_index(&config, "C2 Communication");
         let normal_class_idx = Self::find_label_index(&config, "Normal");
         Self {
-            models,
+            models: ArcSwap::from(models),
             config,
             c2_class_idx,
             normal_class_idx,
@@ -50,6 +48,15 @@ impl Inference {
         }
     }
 
+    /// Atomically swap to new models. Old models are dropped when the last reader finishes.
+    pub fn swap_models(&self, new_models: Arc<MLModels>) {
+        self.models.store(new_models);
+        // Reset circuit breaker on successful model swap
+        self.failure_count.store(0, Ordering::Relaxed);
+        self.failure_window_start.store(0, Ordering::Relaxed);
+        self.circuit_open_since.store(0, Ordering::Relaxed);
+    }
+
     fn find_label_index(config: &MLInferenceConfig, label: &str) -> Option<usize> {
         config
             .attack_labels
@@ -83,8 +90,9 @@ impl Inference {
     }
 
     fn infer_batch_inner(&self, flows: &[FlowData]) -> Vec<DetectionResult> {
+        let models = self.models.load();
         let n = flows.len();
-        let batch_size = self.models.batch_size;
+        let batch_size = models.batch_size;
         let n_ae = self.config.num_ae_features();
         let n_cls = self.config.num_classifier_features();
 
@@ -106,7 +114,7 @@ impl Inference {
                 }
             });
 
-            match self.run_ae_batch(&ae_input, actual, n_ae) {
+            match Self::run_ae_batch(&models, &ae_input, actual, n_ae) {
                 Ok(scores) => ae_scores.extend_from_slice(&scores),
                 Err(e) => {
                     log!(MLLog::InferenceFailed("DeepAutoEncoder".to_string(), e.to_string()));
@@ -138,7 +146,7 @@ impl Inference {
                 }
             });
 
-            match self.run_classifier_batch(&cls_input, actual) {
+            match Self::run_classifier_batch(&models, &cls_input, actual) {
                 Ok((anomaly, class_probs, c2)) => {
                     all_anomaly.extend_from_slice(&anomaly);
                     all_class_probs.extend(class_probs);
@@ -182,12 +190,12 @@ impl Inference {
 
     /// Run AE on a padded batch, return MSE scores for the first `actual` rows.
     fn run_ae_batch(
-        &self,
+        models: &MLModels,
         input: &tract_ndarray::Array2<f32>,
         actual: usize,
         n_features: usize,
     ) -> TractResult<Vec<f32>> {
-        let result = self.models.deep_autoencoder.run(tvec![input.clone().into_tensor().into()])?;
+        let result = models.deep_autoencoder.run(tvec![input.clone().into_tensor().into()])?;
 
         let output = result[0]
             .to_array_view::<f32>()?
@@ -205,14 +213,13 @@ impl Inference {
         Ok(scores)
     }
 
-    /// Run multi-task classifier on a padded batch, return (anomaly, class_probs, c2) for first `actual` rows.
     /// Returns (anomaly_scores, per_class_probs, c2_scores) for the first `actual` rows.
     fn run_classifier_batch(
-        &self,
+        models: &MLModels,
         input: &tract_ndarray::Array2<f32>,
         actual: usize,
     ) -> TractResult<ClassifierBatchOutput> {
-        let result = self.models.classifier.run(tvec![input.clone().into_tensor().into()])?;
+        let result = models.classifier.run(tvec![input.clone().into_tensor().into()])?;
 
         // Output 0: anomaly (batch_size, 1)
         let anomaly_view = result[0].to_array_view::<f32>()?;

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

@@ -7,4 +7,5 @@ pub mod feature_extractor;
 pub mod flow_tracker;
 pub mod inference;
 pub mod model_loader;
+pub mod model_watcher;
 pub mod traffic_logger;

net-guardia/src/core/ml/model_watcher.rs

@@ -0,0 +1,122 @@
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+use macros::log;
+use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
+use tokio::sync::mpsc;
+
+use super::inference::Inference;
+use super::model_loader::MLModels;
+use crate::infrastructure::app_config::AppConfig;
+use crate::model::log::ml::MLLog;
+use crate::model::system::config::MLInferenceConfig;
+
+/// Debounce window: wait for both AE + classifier to land before reloading.
+const DEBOUNCE_SECS: u64 = 5;
+
+/// Watch the models/ directory for .onnx file changes and hot-reload into the inference pipeline.
+pub struct ModelWatcher {
+    inference: Arc<Inference>,
+    app_config: Arc<AppConfig>,
+    inference_config: Arc<MLInferenceConfig>,
+}
+
+impl ModelWatcher {
+    pub fn new(
+        inference: Arc<Inference>,
+        app_config: Arc<AppConfig>,
+        inference_config: Arc<MLInferenceConfig>,
+    ) -> Self {
+        Self {
+            inference,
+            app_config,
+            inference_config,
+        }
+    }
+
+    /// Start watching in a background task.
+    pub fn start(self) {
+        tokio::spawn(async move {
+            if let Err(e) = self.run().await {
+                log!(MLLog::InferenceFailed(
+                    "ModelWatcher".to_string(),
+                    format!("watcher failed to start: {e}"),
+                ));
+            }
+        });
+    }
+
+    async fn run(self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let models_dir = PathBuf::from("models");
+        if !models_dir.exists() {
+            log!(MLLog::InferenceFailed(
+                "ModelWatcher".to_string(),
+                "models/ directory does not exist".to_string(),
+            ));
+            return Ok(());
+        }
+
+        let (tx, mut rx) = mpsc::channel::<()>(16);
+
+        // notify watcher runs on a blocking thread — forward events via channel
+        let _watcher = Self::spawn_watcher(models_dir, tx)?;
+
+        log!(MLLog::ModelWatcherStarted);
+
+        loop {
+            // Wait for first event
+            if rx.recv().await.is_none() {
+                break;
+            }
+
+            // Debounce: drain any additional events within the window
+            tokio::time::sleep(Duration::from_secs(DEBOUNCE_SECS)).await;
+            while rx.try_recv().is_ok() {}
+
+            // Attempt reload
+            self.try_reload();
+        }
+
+        Ok(())
+    }
+
+    fn spawn_watcher(
+        models_dir: PathBuf,
+        tx: mpsc::Sender<()>,
+    ) -> Result<RecommendedWatcher, notify::Error> {
+        let mut watcher = notify::recommended_watcher(move |res: Result<Event, notify::Error>| {
+            if let Ok(event) = res {
+                let dominated = matches!(
+                    event.kind,
+                    EventKind::Create(_) | EventKind::Modify(_)
+                );
+                let has_onnx = event.paths.iter().any(|p| {
+                    p.extension().is_some_and(|ext| ext == "onnx")
+                });
+                if dominated && has_onnx {
+                    let _ = tx.blocking_send(());
+                }
+            }
+        })?;
+
+        watcher.watch(&models_dir, RecursiveMode::NonRecursive)?;
+        Ok(watcher)
+    }
+
+    fn try_reload(&self) {
+        let batch_size = self.app_config.inference.inference_batch_size;
+
+        log!(MLLog::ModelReloadStarting);
+
+        match MLModels::load_models(&self.app_config, &self.inference_config, batch_size) {
+            Ok(new_models) => {
+                self.inference.swap_models(Arc::new(new_models));
+                log!(MLLog::ModelReloadSuccess);
+            }
+            Err(e) => {
+                log!(MLLog::ModelReloadFailed(e.to_string()));
+            }
+        }
+    }
+}

net-guardia/src/core/system.rs

@@ -219,6 +219,14 @@ impl System {
             Self::bridge_ml_to_detection(ml_alert_rx, detection_tx).await;
         });
 
+        // Start model hot-reload watcher (monitors models/ for .onnx changes)
+        let model_watcher = crate::core::ml::model_watcher::ModelWatcher::new(
+            self.app_services.ml_engine.inference_pipeline().clone(),
+            self.app_config.clone(),
+            self.inference_config.clone(),
+        );
+        model_watcher.start();
+
         // Initialize force_https flag from DB setting
         let force_https = Arc::new(std::sync::atomic::AtomicBool::new(
             self.db

net-guardia/src/model/log/ml.rs

@@ -44,5 +44,17 @@ loggable! {
 
         #[error("ML circuit breaker RESET: inference re-enabled after {cooldown_secs}s cooldown")]
         CircuitBreakerReset { cooldown_secs: u64 } => tracing::Level::WARN,
+
+        #[error("Model watcher started, monitoring models/ for .onnx changes")]
+        ModelWatcherStarted => tracing::Level::INFO,
+
+        #[error("Model reload triggered, loading new ONNX models...")]
+        ModelReloadStarting => tracing::Level::INFO,
+
+        #[error("Model reload successful, inference pipeline updated")]
+        ModelReloadSuccess => tracing::Level::INFO,
+
+        #[error("Model reload failed, keeping current models: {error}")]
+        ModelReloadFailed { error: String } => tracing::Level::ERROR,
     }
 }