feat(flow-trace): WORM audit emit on recording stop

The writer thread stopped the Flow Trace recorder for three reasons —
initial file open failure, FIFO sweep failure, and mid-run rotation
failure — but each path only produced a tracing line. An auditor
reviewing the WORM chain later had no durable record that the
recording had gone dormant or why, leaving a silent gap that violates
the "auditor-visible state transitions" invariant the audit chain
exists to uphold.

The writer thread now also publishes `AuditEvent { actor="system",
action="flow_trace_stopped", detail={reason, directory} }` at each
stop site. CommunicationManager gains `publish_event_sync` so the
dedicated std::thread writer can hand events to the broadcast bus
without needing a tokio runtime — the channel send was always
synchronous under the hood, so this only exposes what was already
there.

AuditEvent registration moves ahead of AppServices::new in the
service factory so the broadcaster exists before the writer thread
can possibly fire its first audit during startup. Tests cover
publish_event_sync delivery and the stop-audit helper on both
comm-provided and comm-absent paths.

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

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

@@ -5,7 +5,10 @@
 //! recording session can't eat the disk.
 //!
 //! On FIFO failure (permissions, I/O error) the writer shuts down
-//! cleanly, leaves inference untouched, and logs through `MLLog`.
+//! cleanly, leaves inference untouched, logs through `MLLog`, and
+//! (when a `CommunicationManager` is wired in) also publishes a WORM
+//! `AuditEvent` so the chain records an auditor-visible reason the
+//! recording stopped, not just a tracing line that may be lost.
 //! Callers see the channel disconnect and stop sending rows.
 
 use std::fs::{File, OpenOptions};
@@ -18,7 +21,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use crossbeam::channel::{Receiver, Sender, TrySendError, bounded};
 use macros::log;
 
+use crate::infrastructure::communication_manager::CommunicationManager;
 use crate::model::error::ml::MLError;
+use crate::model::event::AuditEvent;
 use crate::model::log::ml::MLLog;
 
 /// Default per-file size cap. A single CSV file won't grow past this
@@ -72,8 +77,15 @@ pub struct TrafficLogger {
 impl TrafficLogger {
     /// Build a rotating writer rooted at `base_path`'s parent. Any
     /// existing `flow-trace-*.csv` in that directory participates in
-    /// the FIFO budget.
-    pub fn new(base_path: &Path, header: Vec<String>, policy: RotationPolicy) -> Result<Self, io::Error> {
+    /// the FIFO budget. `comm` is optional so tests (and paths where
+    /// the bus isn't wired yet) can exercise the rotation logic without
+    /// the event-bus dependency; production code always passes `Some`.
+    pub fn new(
+        base_path: &Path,
+        header: Vec<String>,
+        policy: RotationPolicy,
+        comm: Option<Arc<CommunicationManager>>,
+    ) -> Result<Self, io::Error> {
         let directory = base_path
             .parent()
             .map(Path::to_path_buf)
@@ -88,7 +100,7 @@ impl TrafficLogger {
         thread::Builder::new()
             .name("traffic-logger".to_string())
             .spawn(move || {
-                writer_loop(receiver, writer_dir, writer_header, writer_policy);
+                writer_loop(receiver, writer_dir, writer_header, writer_policy, comm);
             })?;
 
         Ok(Self {
@@ -178,11 +190,19 @@ fn parse_timestamp_suffix(name: &str) -> Option<u64> {
     without_ext.parse::<u64>().ok()
 }
 
-fn writer_loop(receiver: Receiver<Vec<String>>, directory: PathBuf, header: Vec<String>, policy: RotationPolicy) {
+fn writer_loop(
+    receiver: Receiver<Vec<String>>,
+    directory: PathBuf,
+    header: Vec<String>,
+    policy: RotationPolicy,
+    comm: Option<Arc<CommunicationManager>>,
+) {
     let mut active = match open_new_file(&directory, &header) {
         Ok(a) => a,
         Err(e) => {
-            log!(MLLog::FlowTraceStopped(e.to_string()));
+            let reason = e.to_string();
+            log!(MLLog::FlowTraceStopped(reason.clone()));
+            emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory);
             return;
         }
     };
@@ -198,13 +218,17 @@ fn writer_loop(receiver: Receiver<Vec<String>>, directory: PathBuf, header: Vec<
                 // FIFO failure is the documented "stop Flow Trace, keep
                 // inference running" path. Drop the channel so callers
                 // see the disconnect and stop trying.
-                log!(MLLog::FlowTraceStopped(format!("FIFO sweep failed: {e}")));
+                let reason = format!("FIFO sweep failed: {e}");
+                log!(MLLog::FlowTraceStopped(reason.clone()));
+                emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory);
                 return;
             }
             active = match open_new_file(&directory, &header) {
                 Ok(a) => a,
                 Err(e) => {
-                    log!(MLLog::FlowTraceStopped(format!("rotate failed: {e}")));
+                    let reason = format!("rotate failed: {e}");
+                    log!(MLLog::FlowTraceStopped(reason.clone()));
+                    emit_flow_trace_stop_audit(comm.as_ref(), &reason, &directory);
                     return;
                 }
             };
@@ -223,6 +247,30 @@ fn writer_loop(receiver: Receiver<Vec<String>>, directory: PathBuf, header: Vec<
     }
 }
 
+/// Publish a WORM `flow_trace_stopped` audit entry so an auditor can
+/// later see why Flow Trace recording went dormant without grepping
+/// process logs. A `None` bus (tests, or pre-wiring paths) is a
+/// deliberate no-op — the sibling `MLLog::FlowTraceStopped` tracing
+/// line still fires in both cases.
+///
+/// Extracted as a free function so tests can cover the emit path
+/// without driving a full writer-loop + failing-disk fixture.
+fn emit_flow_trace_stop_audit(comm: Option<&Arc<CommunicationManager>>, reason: &str, directory: &Path) {
+    let Some(c) = comm else {
+        return;
+    };
+    let detail = serde_json::json!({
+        "reason": reason,
+        "directory": directory.display().to_string(),
+    })
+    .to_string();
+    let _ = c.publish_event_sync(AuditEvent {
+        actor: "system".to_string(),
+        action: "flow_trace_stopped".to_string(),
+        detail,
+    });
+}
+
 struct ActiveFile {
     writer: BufWriter<File>,
     opened_at: Instant,
@@ -357,4 +405,28 @@ mod tests {
         let missing = PathBuf::from("/nonexistent/flow-trace/dir");
         assert!(list_flow_trace_files(&missing).unwrap().is_empty());
     }
+
+    #[tokio::test]
+    async fn stop_audit_reaches_subscriber_when_comm_provided() {
+        let comm = Arc::new(CommunicationManager::new());
+        comm.register_event_type::<AuditEvent>();
+        let mut rx = comm.subscribe_event::<AuditEvent>().unwrap();
+
+        let dir = scratch_dir("audit-emit");
+        emit_flow_trace_stop_audit(Some(&comm), "FIFO sweep failed: perm denied", &dir);
+
+        let event = rx.recv().await.expect("audit event must be delivered");
+        assert_eq!(event.actor, "system");
+        assert_eq!(event.action, "flow_trace_stopped");
+        let parsed: serde_json::Value = serde_json::from_str(&event.detail).unwrap();
+        assert_eq!(parsed["reason"], "FIFO sweep failed: perm denied");
+        assert_eq!(parsed["directory"], dir.display().to_string());
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    #[test]
+    fn stop_audit_noop_when_comm_absent() {
+        // Passing None is the explicit test-mode path — must not panic.
+        emit_flow_trace_stop_audit(None, "any reason", Path::new("/tmp/anywhere"));
+    }
 }

net-guardia/src/infrastructure/app_services.rs

@@ -16,6 +16,7 @@ use crate::core::ml::manifest::ModelManifest;
 use crate::core::ml::model_loader::build_adapter;
 use crate::core::ml::traffic_logger::{RotationPolicy, TrafficLogger};
 use crate::infrastructure::app_config::AppConfig;
+use crate::infrastructure::communication_manager::CommunicationManager;
 use crate::infrastructure::health::SystemHealth;
 use crate::infrastructure::statistics::FlowStatistics;
 use crate::model::config::constants::{MANIFEST_FILENAME, MODELS_DIR};
@@ -50,6 +51,7 @@ impl AppServices {
         ml_manifest: Option<ModelManifest>,
         drift_detector: Arc<parking_lot::Mutex<DriftDetector>>,
         ebpf_health: Arc<parking_lot::RwLock<EbpfHealth>>,
+        comm: Arc<CommunicationManager>,
     ) -> Result<Self, Error> {
         let health = SystemHealth::new(app_config.clone(), ebpf_health)?;
 
@@ -100,7 +102,7 @@ impl AppServices {
             let mut header = FlowFeatures::all_feature_names_owned();
             header.push("Label".to_string());
             let base_path = PathBuf::from(&csv_path);
-            let logger = TrafficLogger::new(&base_path, header, RotationPolicy::default())
+            let logger = TrafficLogger::new(&base_path, header, RotationPolicy::default(), Some(comm.clone()))
                 .map_err(|e| MiscError::TrafficLogCreateError(csv_path.clone(), e.to_string()))?;
             log!(SystemLog::TrafficLoggingEnabled(csv_path));
             Some(Arc::new(logger))

net-guardia/src/infrastructure/communication_manager.rs

@@ -134,6 +134,21 @@ impl CommunicationManager {
             .ok_or(MiscError::TypeNotRegistered)?;
         broadcaster.broadcast_event(Box::new(event))
     }
+
+    /// Synchronous counterpart for callers that live outside the tokio
+    /// runtime — in particular, the Flow Trace writer thread, which
+    /// runs on a dedicated `std::thread` and can't `.await`. The
+    /// internal broadcast channel is already non-blocking, so the
+    /// `async fn` sibling never actually yields; this variant exposes
+    /// the same work without the ceremony.
+    pub fn publish_event_sync<E: Event + 'static>(&self, event: E) -> Result<(), Error> {
+        let type_id = TypeId::of::<E>();
+        let broadcaster = self
+            .event_broadcasters
+            .get(&type_id)
+            .ok_or(MiscError::TypeNotRegistered)?;
+        broadcaster.broadcast_event(Box::new(event))
+    }
 }
 
 /// Fluent builder for registering a service's command/query/event handlers.
@@ -295,6 +310,27 @@ mod tests {
         assert!(result.is_err());
     }
 
+    #[tokio::test]
+    async fn publish_event_sync_delivers_to_subscriber() {
+        let comm = CommunicationManager::new();
+        comm.register_event_type::<TestEvent>();
+        let mut rx = comm.subscribe_event::<TestEvent>().unwrap();
+
+        comm.publish_event_sync(TestEvent { message: "sync".into() }).unwrap();
+
+        let event = rx.recv().await.unwrap();
+        assert_eq!(event.message, "sync");
+    }
+
+    #[test]
+    fn publish_event_sync_errors_when_type_unregistered() {
+        let comm = CommunicationManager::new();
+        let result = comm.publish_event_sync(TestEvent {
+            message: "dropped".into(),
+        });
+        assert!(result.is_err());
+    }
+
     #[tokio::test]
     async fn test_event_multiple_subscribers() {
         let comm = CommunicationManager::new();

net-guardia/src/infrastructure/service_factory.rs

@@ -185,14 +185,6 @@ impl ServiceFactory {
             Duration::from_secs(drift_window_secs),
         )));
 
-        let app_services = Arc::new(AppServices::new(
-            app_config.clone(),
-            inference_config.clone(),
-            ml_manifest.clone(),
-            drift_detector.clone(),
-            ebpf_health.clone(),
-        )?);
-
         // Create AtomicU8 enforce-level cache (Monitor=0, MlOnly=1, Enforce=2)
         let enforce_level_cache = Arc::new(AtomicU8::new({
             use crate::infrastructure::enforce_mode_handler::enforce_mode_to_u8;
@@ -200,8 +192,25 @@ impl ServiceFactory {
             enforce_mode_to_u8(&mode_str)
         }));
 
-        // Create CommunicationManager and register enforce-mode handler
+        // CommunicationManager and its event channels must exist before
+        // AppServices spins up the TrafficLogger: the writer thread can
+        // publish `flow_trace_stopped` audit events the moment it tries
+        // to open its first rotated file, and an unregistered channel
+        // would silently drop that evidence.
         let comm = Arc::new(CommunicationManager::new());
+        comm.register_event_type::<ThreatDetectedEvent>();
+        comm.register_event_type::<DriftDetectedEvent>();
+        comm.register_event_type::<AuditEvent>();
+
+        let app_services = Arc::new(AppServices::new(
+            app_config.clone(),
+            inference_config.clone(),
+            ml_manifest.clone(),
+            drift_detector.clone(),
+            ebpf_health.clone(),
+            comm.clone(),
+        )?);
+
         let enforce_handler = Arc::new(EnforceModeHandler::new(
             db.clone() as Arc<dyn AppRepo>,
             comm.clone(),
@@ -214,11 +223,6 @@ impl ServiceFactory {
             .query::<GetEnforceModeQuery>()
             .build();
 
-        // Register event type channels
-        comm.register_event_type::<ThreatDetectedEvent>();
-        comm.register_event_type::<DriftDetectedEvent>();
-        comm.register_event_type::<AuditEvent>();
-
         // Seed default SOAR playbooks if empty
         (db.as_ref() as &dyn SoarRepo).seed_default_playbooks()?;