feat(observability): in-memory log ring buffer + /api/logs/live

core/observability/log_buffer.rs — new module.
  parking_lot::Mutex<VecDeque<LogEntry>> capped at 5000 rows with a
  lock-free AtomicU64 cursor. impl tracing_subscriber::Layer so every
  event goes into the ring alongside the existing stdout + file
  appenders. MessageVisitor joins the format-args body with any
  structured fields as k=v pairs; entries over 8 KB are truncated
  with a marker. snapshot(since_id, min_severity, limit) backs the
  HTTP tail endpoint. 5 unit tests.

adapter/http/logs.rs — add GET /api/logs/live.
  Query: since_id (cursor), limit (clamped 1..=2000), min_level.
  Response carries entries, next_id, total_buffered, and a
  dropped_oldest flag the UI uses to warn about gaps when the ring
  evicts rows between polls.

utils/logging.rs
  * Mount LogBufferLayer under the existing EnvFilter reload layer
    so level changes apply uniformly to stdout/file/ring.
  * extract_main_level strips per-target directives
    ("maxminddb=warn,debug" -> "debug") so the system log-level
    <select> binds to one of the five options instead of the raw
    EnvFilter string (UX-2 backend half). 2 unit tests.

Bundles the matching prior-session landings already in the working
tree:
  * adapter/websocket/fusion_websocket.rs — /ws/fusion channel
    (subscribes ThreatDetectedEvent, server-stamps ts) + 2 tests,
    fixing the F-4 gap where the event existed on the bus but not
    on the wire
  * model/event.rs — derive Serialize on ThreatDetectedEvent for
    WS serialization
  * adapter/websocket/{mod,routes}.rs, model/log/http.rs —
    wire fusion_websocket into the WS router with its HttpLog entry
  * adapter/persistence/repository.rs — accumulated refactor

cargo clippy --package net-guardia -- -D warnings: clean.
cargo test --bin net-guardia: 288 passing (281 prior + 7 new).

net-guardia-frontend

@@ -1 +1 @@
-Subproject commit 0e600c5f4956aa599beff09413011b088e848fe9
+Subproject commit 00d347c5eae3ed32f595b0a3553601f16bedfa7e

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

@@ -4,7 +4,9 @@ use std::path::Path;
 use std::time::UNIX_EPOCH;
 
 use actix_web::{HttpResponse, Scope, web};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
+
+use crate::core::observability::log_buffer::{self, LogEntry};
 
 /// Hardcoded log directory — not configurable via API to prevent directory traversal.
 const LOG_DIR: &str = "logs";
@@ -12,6 +14,15 @@ const LOG_DIR: &str = "logs";
 /// Maximum downloadable log file size (50 MB). Prevents OOM from reading huge files.
 const MAX_DOWNLOAD_SIZE: u64 = 50 * 1024 * 1024;
 
+/// Default page size for `/live` when the client does not specify `limit`.
+/// Chosen so a 2 s poll against a DEBUG-chatty deployment catches up in
+/// one round-trip without being absurd payload-wise.
+const LIVE_DEFAULT_LIMIT: usize = 500;
+
+/// Hard cap on `/live?limit=` — prevents pathological clients from asking
+/// for the entire buffer at once.
+const LIVE_MAX_LIMIT: usize = 2_000;
+
 /// Validate log filename: only alphanumeric, dots, underscores, hyphens.
 /// Prevents path traversal.
 fn is_valid_log_filename(name: &str) -> bool {
@@ -25,9 +36,52 @@ fn is_valid_log_filename(name: &str) -> bool {
 pub fn initialize() -> Scope {
     web::scope("/logs")
         .route("", web::get().to(list_logs))
+        .route("/live", web::get().to(live_logs))
         .route("/{filename}", web::get().to(download_log))
 }
 
+#[derive(Deserialize)]
+struct LiveQuery {
+    #[serde(default)]
+    since_id: Option<u64>,
+    #[serde(default)]
+    limit: Option<usize>,
+    #[serde(default)]
+    min_level: Option<String>,
+}
+
+#[derive(Serialize)]
+struct LiveResponse {
+    entries: Vec<LogEntry>,
+    next_id: u64,
+    total_buffered: usize,
+    dropped_oldest: bool,
+}
+
+async fn live_logs(query: web::Query<LiveQuery>) -> HttpResponse {
+    let since_id = query.since_id.unwrap_or(0);
+    let limit = query.limit.unwrap_or(LIVE_DEFAULT_LIMIT).clamp(1, LIVE_MAX_LIMIT);
+    let min_severity = query
+        .min_level
+        .as_deref()
+        .map(|s| log_buffer::level_severity(&s.to_ascii_uppercase()))
+        .unwrap_or(log_buffer::level_severity("TRACE"));
+
+    let snap = log_buffer::snapshot(since_id, min_severity, limit);
+    // Signal to the UI that it lagged enough for the ring to evict rows
+    // between polls. Frontend can warn "older entries dropped" without
+    // silently skipping a gap.
+    let dropped_oldest = since_id > 0 && snap.entries.first().is_some_and(|e| e.id > since_id + 1);
+    let next_id = snap.entries.last().map(|e| e.id).unwrap_or(snap.latest_id);
+
+    HttpResponse::Ok().json(LiveResponse {
+        entries: snap.entries,
+        next_id,
+        total_buffered: snap.total,
+        dropped_oldest,
+    })
+}
+
 #[derive(Serialize)]
 struct LogFileEntry {
     name: String,

net-guardia/src/adapter/persistence/repository.rs

@@ -47,7 +47,25 @@ impl r2d2::CustomizeConnection<rusqlite::Connection, rusqlite::Error> for Sqlite
             // Use a parameterised query to avoid SQL-injection via the key value.
             conn.pragma_update(None, "key", key)?;
         }
-        conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
+        // PRAGMA tuning notes:
+        // - `journal_mode=WAL`: many concurrent readers + one writer; the only
+        //   journal mode that survives crashes without losing committed rows.
+        // - `synchronous=NORMAL`: canonical pairing with WAL — `FULL` adds an
+        //   extra fsync per commit that buys no durability guarantees beyond
+        //   what WAL already provides for a power-loss event.
+        // - `busy_timeout=5000`: WAL still serializes writers (SOAR, audit,
+        //   drift, SQL hooks all share one DB), and the default 0ms returns
+        //   SQLITE_BUSY immediately on any contention. 5s gives the loser
+        //   enough time to wait out a normal commit (sub-ms) without masking
+        //   genuine deadlocks.
+        // - `foreign_keys=ON`: enforce FK constraints at the connection
+        //   level (SQLite's default is OFF for backwards compatibility).
+        conn.execute_batch(
+            "PRAGMA journal_mode=WAL; \
+             PRAGMA synchronous=NORMAL; \
+             PRAGMA busy_timeout=5000; \
+             PRAGMA foreign_keys=ON;",
+        )?;
         Ok(())
     }
 }

net-guardia/src/adapter/websocket/fusion_websocket.rs

@@ -0,0 +1,192 @@
+//! WebSocket bridge for post-fusion threat events.
+//!
+//! `/ws/fusion` subscribes to the `ThreatDetectedEvent` broadcast that the
+//! `DetectionOrchestrator` already publishes through `CommunicationManager`
+//! (the same stream SOAR consumes). Each event is wrapped with a server-side
+//! `ts` (unix seconds) so the dashboard can render relative timestamps
+//! without doing the conversion itself.
+//!
+//! Distinct from `/ws/alerts` (flow-level ML detections via `MLAlert`):
+//! this stream is the **fused, per-IP, multi-source** view that drives the
+//! Overview "Recent Threats" card and the sources-agreed chip. Treating
+//! them as one channel would conflate two bounded contexts — see
+//! `docs/strategy/DOMAIN_MAP.md` for the BC split rationale.
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use actix_web::rt::spawn;
+use actix_web::{HttpRequest, HttpResponse, Result, web};
+use actix_ws::{Message, MessageStream, Session, handle};
+use futures_util::StreamExt;
+use macros::log;
+use tokio::sync::broadcast;
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::infrastructure::communication_manager::CommunicationManager;
+use crate::model::error::http::HttpError;
+use crate::model::error::misc::MiscError;
+use crate::model::event::ThreatDetectedEvent;
+use crate::model::log::http::HttpLog;
+
+pub async fn websocket_fusion(
+    req: HttpRequest,
+    body: web::Payload,
+    comm: web::Data<CommunicationManager>,
+) -> Result<HttpResponse> {
+    let (response, session, msg_stream) = handle(&req, body)?;
+
+    let broadcast_rx = match comm.subscribe_event::<ThreatDetectedEvent>() {
+        Ok(rx) => rx,
+        Err(e) => {
+            log!(HttpLog::FusionSubscribeFailed(e.to_string()));
+            return Ok(HttpResponse::InternalServerError().json(serde_json::json!({
+                "error": "fusion event channel not registered",
+            })));
+        }
+    };
+
+    spawn(async move {
+        handle_fusion_connection(session, msg_stream, broadcast_rx).await;
+    });
+
+    Ok(response)
+}
+
+async fn handle_fusion_connection(
+    mut session: Session,
+    mut msg_stream: MessageStream,
+    mut broadcast_rx: broadcast::Receiver<ThreatDetectedEvent>,
+) {
+    loop {
+        tokio::select! {
+            msg_result = msg_stream.next() => {
+                if !handle_client_message(&mut session, msg_result).await {
+                    break;
+                }
+            },
+            broadcast_result = broadcast_rx.recv() => {
+                match broadcast_result {
+                    Ok(event) => {
+                        if !send_event(&mut session, &event).await {
+                            break;
+                        }
+                    }
+                    Err(RecvError::Lagged(skipped)) => {
+                        log!(HttpLog::WebSocketLagged(skipped));
+                        continue;
+                    }
+                    Err(RecvError::Closed) => {
+                        break;
+                    }
+                }
+            },
+        }
+    }
+
+    let _ = session.close(None).await;
+}
+
+async fn handle_client_message(
+    session: &mut Session,
+    msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
+) -> bool {
+    match msg_result {
+        Some(Ok(Message::Text(_))) => true,
+        Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
+        Some(Ok(Message::Close(reason))) => {
+            let _ = (session.clone()).close(reason).await;
+            false
+        }
+        Some(Err(err)) => {
+            log!(HttpError::WebSocketError(err));
+            false
+        }
+        None => false,
+        _ => true,
+    }
+}
+
+/// Wrap each event in `{ts, ...event_fields}`. The `ts` is a server-stamped
+/// unix-seconds value so the client can render "5s ago" without inferring
+/// the time from the audit chain. All declared fields of
+/// `ThreatDetectedEvent` flow through verbatim via the event's own
+/// `Serialize` derive — no field whitelist to drift out of date.
+async fn send_event(session: &mut Session, event: &ThreatDetectedEvent) -> bool {
+    let ts = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0);
+
+    let payload = match serde_json::to_value(event) {
+        Ok(serde_json::Value::Object(mut map)) => {
+            map.insert("ts".to_string(), serde_json::Value::from(ts));
+            serde_json::Value::Object(map)
+        }
+        // The derived Serialize on a struct always produces an Object —
+        // this branch only fires if the type changes shape in a future
+        // refactor. Falling back to the raw value keeps the stream alive.
+        Ok(other) => other,
+        Err(err) => {
+            log!(MiscError::SerializeError(err));
+            return false;
+        }
+    };
+
+    match serde_json::to_string(&payload) {
+        Ok(json) => session.text(json).await.is_ok(),
+        Err(err) => {
+            log!(MiscError::SerializeError(err));
+            false
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::model::event::DetectionSource;
+
+    fn sample_event() -> ThreatDetectedEvent {
+        ThreatDetectedEvent {
+            attack_type: "brute_force".to_string(),
+            confidence: 0.92,
+            source_ip: "203.0.113.10".to_string(),
+            dest_ip: "10.0.0.1".to_string(),
+            flow_count: 3,
+            packet_rate: 12.5,
+            protocol: 6,
+            geoip_country: Some("CN".to_string()),
+            is_repeat_offender: true,
+            sources: vec![DetectionSource::ML, DetectionSource::Suricata],
+            active_source_count: 2,
+            fused_confidence: 0.99,
+            ae_score: 0.0,
+            anomaly_score: 0.0,
+            c2_score: 0.0,
+        }
+    }
+
+    #[test]
+    fn event_serializes_with_canonical_source_strings() {
+        let event = sample_event();
+        let json = serde_json::to_value(&event).expect("serialize event");
+        let sources = json["sources"].as_array().expect("sources array");
+        assert_eq!(sources[0], "ML");
+        assert_eq!(sources[1], "Suricata");
+        assert_eq!(json["active_source_count"], 2);
+        assert_eq!(json["geoip_country"], "CN");
+    }
+
+    #[test]
+    fn envelope_adds_ts_field_to_event_object() {
+        // The `send_event` wire path inserts `ts` into the event's own
+        // serde object; mirror that here without an actix session so the
+        // wrapping logic stays covered when the orchestrator schema evolves.
+        let event = sample_event();
+        let mut value = serde_json::to_value(&event).expect("serialize event");
+        let object = value.as_object_mut().expect("expected object shape");
+        object.insert("ts".to_string(), serde_json::Value::from(1_700_000_000_u64));
+        assert_eq!(value["ts"], 1_700_000_000_u64);
+        assert_eq!(value["attack_type"], "brute_force");
+    }
+}

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

@@ -1,5 +1,6 @@
 pub mod alert_websocket;
 pub mod drop_websocket;
 pub mod flow_websocket;
+pub mod fusion_websocket;
 pub mod health_websocket;
 pub mod routes;

net-guardia/src/adapter/websocket/routes.rs

@@ -1,10 +1,11 @@
 use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web};
 use serde::Deserialize;
 
-use super::{alert_websocket, drop_websocket, flow_websocket, health_websocket};
+use super::{alert_websocket, drop_websocket, flow_websocket, fusion_websocket, health_websocket};
 use crate::adapter::ebpf::drop_monitor::DropMonitor;
 use crate::core::auth::jwt::JwtService;
 use crate::core::ml::alert::MLAlert;
+use crate::infrastructure::communication_manager::CommunicationManager;
 use crate::infrastructure::health::SystemHealth;
 use crate::infrastructure::statistics::FlowStatistics;
 
@@ -17,6 +18,7 @@ pub fn initialize() -> Scope {
     web::scope("/ws")
         .route("/health", web::get().to(health_ws))
         .route("/alerts", web::get().to(alerts_ws))
+        .route("/fusion", web::get().to(fusion_ws))
         .route("/flows", web::get().to(flows_ws))
         .route("/drops", web::get().to(drops_ws))
 }
@@ -84,6 +86,24 @@ async fn alerts_ws(
     }
 }
 
+async fn fusion_ws(
+    req: HttpRequest,
+    stream: web::Payload,
+    comm: web::Data<CommunicationManager>,
+    query: web::Query<WsQuery>,
+    jwt: web::Data<JwtService>,
+) -> impl Responder {
+    if let Err(resp) = validate_ws_token(&req, &query, &jwt) {
+        return resp;
+    }
+    match fusion_websocket::websocket_fusion(req, stream, comm).await {
+        Ok(response) => response,
+        Err(err) => {
+            HttpResponse::InternalServerError().json(serde_json::json!({"error": format!("WebSocket error: {}", err)}))
+        }
+    }
+}
+
 async fn flows_ws(
     req: HttpRequest,
     stream: web::Payload,

net-guardia/src/core/mod.rs

@@ -7,6 +7,7 @@ pub mod dns_filter_service;
 pub mod email;
 pub mod ml;
 pub mod notification_service;
+pub mod observability;
 pub mod playbook_service;
 pub mod rate_limit_service;
 pub mod report;

net-guardia/src/core/observability/log_buffer.rs

@@ -0,0 +1,304 @@
+use std::collections::VecDeque;
+use std::fmt::{Arguments, Debug, Write as _};
+use std::sync::OnceLock;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use parking_lot::Mutex;
+use serde::Serialize;
+use tracing::field::{Field, Visit};
+use tracing::{Event, Level, Subscriber};
+use tracing_subscriber::Layer;
+use tracing_subscriber::layer::Context;
+
+/// Ring-buffer capacity. Tuned for ~30 min of INFO traffic on a small SOC
+/// deployment; DEBUG floods will churn faster.
+const DEFAULT_CAPACITY: usize = 5_000;
+
+/// Per-entry payload cap. Guards against pathological debug logs from
+/// bursting the buffer.
+const MAX_MESSAGE_BYTES: usize = 8_192;
+
+/// Monotonic id allocator. Clients use `since_id` to resume tailing.
+/// u64 never wraps in practice (2^64 events at 1 µs/event ≈ 584 000 years).
+static NEXT_ID: AtomicU64 = AtomicU64::new(1);
+
+static BUFFER: OnceLock<LogRingBuffer> = OnceLock::new();
+
+#[derive(Clone, Debug, Serialize)]
+pub struct LogEntry {
+    pub id: u64,
+    pub ts_ms: u64,
+    pub level: &'static str,
+    pub target: String,
+    pub message: String,
+}
+
+struct LogRingBuffer {
+    entries: Mutex<VecDeque<LogEntry>>,
+    capacity: usize,
+}
+
+impl LogRingBuffer {
+    fn new(capacity: usize) -> Self {
+        Self {
+            entries: Mutex::new(VecDeque::with_capacity(capacity)),
+            capacity,
+        }
+    }
+
+    fn push(&self, entry: LogEntry) {
+        let mut guard = self.entries.lock();
+        if guard.len() >= self.capacity {
+            guard.pop_front();
+        }
+        guard.push_back(entry);
+    }
+
+    fn snapshot(&self, since_id: u64, min_severity: u8, limit: usize) -> Snapshot {
+        let guard = self.entries.lock();
+        let total = guard.len();
+        let latest_id = guard.back().map(|e| e.id).unwrap_or(0);
+        let entries: Vec<LogEntry> = guard
+            .iter()
+            .filter(|e| e.id > since_id && level_severity(e.level) <= min_severity)
+            .take(limit)
+            .cloned()
+            .collect();
+        Snapshot {
+            entries,
+            latest_id,
+            total,
+        }
+    }
+}
+
+pub struct Snapshot {
+    pub entries: Vec<LogEntry>,
+    pub latest_id: u64,
+    pub total: usize,
+}
+
+/// Get a snapshot for the `/api/logs/live` endpoint.
+///
+/// `min_severity` follows tracing level numeric ordering (ERROR=1…TRACE=5);
+/// an entry at level L is included when `level_severity(L) <= min_severity`.
+/// Returns an empty snapshot when the buffer has not been installed yet
+/// (tests, dry-runs).
+pub fn snapshot(since_id: u64, min_severity: u8, limit: usize) -> Snapshot {
+    match BUFFER.get() {
+        Some(buf) => buf.snapshot(since_id, min_severity, limit),
+        None => Snapshot {
+            entries: Vec::new(),
+            latest_id: 0,
+            total: 0,
+        },
+    }
+}
+
+/// Map a level string to severity rank. Unknown strings sort as TRACE so
+/// they are only visible when the caller asks for everything.
+pub fn level_severity(level: &str) -> u8 {
+    match level {
+        "ERROR" => 1,
+        "WARN" => 2,
+        "INFO" => 3,
+        "DEBUG" => 4,
+        _ => 5,
+    }
+}
+
+fn level_str(level: &Level) -> &'static str {
+    match *level {
+        Level::ERROR => "ERROR",
+        Level::WARN => "WARN",
+        Level::INFO => "INFO",
+        Level::DEBUG => "DEBUG",
+        Level::TRACE => "TRACE",
+    }
+}
+
+fn now_unix_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+/// [`Layer`] that appends each formatted event into the in-memory ring
+/// buffer so the UI can tail logs without round-tripping the filesystem.
+pub struct LogBufferLayer;
+
+impl LogBufferLayer {
+    pub fn new() -> Self {
+        let _ = BUFFER.set(LogRingBuffer::new(DEFAULT_CAPACITY));
+        Self
+    }
+}
+
+impl Default for LogBufferLayer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<S: Subscriber> Layer<S> for LogBufferLayer {
+    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+        let Some(buf) = BUFFER.get() else {
+            return;
+        };
+        let metadata = event.metadata();
+        let mut visitor = MessageVisitor::default();
+        event.record(&mut visitor);
+        let mut message = visitor.into_message();
+        if message.len() > MAX_MESSAGE_BYTES {
+            message.truncate(MAX_MESSAGE_BYTES);
+            message.push_str("…[truncated]");
+        }
+        let entry = LogEntry {
+            id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
+            ts_ms: now_unix_ms(),
+            level: level_str(metadata.level()),
+            target: metadata.target().to_string(),
+            message,
+        };
+        buf.push(entry);
+    }
+}
+
+/// Collects `message` plus remaining fields as `key=value` pairs. `tracing`
+/// macros emit the format-args body under the `message` field; structured
+/// fields come through [`Visit::record_*`] for the respective primitive.
+#[derive(Default)]
+struct MessageVisitor {
+    message: String,
+    extra: String,
+}
+
+impl MessageVisitor {
+    fn into_message(mut self) -> String {
+        if self.extra.is_empty() {
+            self.message
+        } else if self.message.is_empty() {
+            self.extra
+        } else {
+            self.message.push(' ');
+            self.message.push_str(&self.extra);
+            self.message
+        }
+    }
+
+    fn push_extra(&mut self, name: &str, value: Arguments<'_>) {
+        if !self.extra.is_empty() {
+            self.extra.push(' ');
+        }
+        let _ = write!(self.extra, "{}={}", name, value);
+    }
+}
+
+impl Visit for MessageVisitor {
+    fn record_debug(&mut self, field: &Field, value: &dyn Debug) {
+        if field.name() == "message" {
+            let _ = write!(self.message, "{:?}", value);
+        } else {
+            self.push_extra(field.name(), format_args!("{:?}", value));
+        }
+    }
+
+    fn record_str(&mut self, field: &Field, value: &str) {
+        if field.name() == "message" {
+            self.message.push_str(value);
+        } else {
+            self.push_extra(field.name(), format_args!("{}", value));
+        }
+    }
+
+    fn record_i64(&mut self, field: &Field, value: i64) {
+        self.push_extra(field.name(), format_args!("{}", value));
+    }
+
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        self.push_extra(field.name(), format_args!("{}", value));
+    }
+
+    fn record_f64(&mut self, field: &Field, value: f64) {
+        self.push_extra(field.name(), format_args!("{}", value));
+    }
+
+    fn record_bool(&mut self, field: &Field, value: bool) {
+        self.push_extra(field.name(), format_args!("{}", value));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make_entry(id: u64, level: &'static str, message: &str) -> LogEntry {
+        LogEntry {
+            id,
+            ts_ms: 0,
+            level,
+            target: "test".into(),
+            message: message.into(),
+        }
+    }
+
+    #[test]
+    fn severity_ordering() {
+        assert!(level_severity("ERROR") < level_severity("WARN"));
+        assert!(level_severity("WARN") < level_severity("INFO"));
+        assert!(level_severity("INFO") < level_severity("DEBUG"));
+        assert!(level_severity("DEBUG") < level_severity("TRACE"));
+        assert_eq!(level_severity("unknown"), 5);
+    }
+
+    #[test]
+    fn ring_buffer_drops_oldest_at_capacity() {
+        let buf = LogRingBuffer::new(3);
+        for id in 1..=5 {
+            buf.push(make_entry(id, "INFO", "m"));
+        }
+        let snap = buf.snapshot(0, level_severity("TRACE"), 100);
+        let ids: Vec<u64> = snap.entries.iter().map(|e| e.id).collect();
+        assert_eq!(ids, vec![3, 4, 5]);
+        assert_eq!(snap.latest_id, 5);
+        assert_eq!(snap.total, 3);
+    }
+
+    #[test]
+    fn snapshot_filters_since_id_and_severity() {
+        let buf = LogRingBuffer::new(16);
+        buf.push(make_entry(1, "INFO", "first"));
+        buf.push(make_entry(2, "DEBUG", "noisy"));
+        buf.push(make_entry(3, "ERROR", "boom"));
+
+        let snap = buf.snapshot(1, level_severity("INFO"), 100);
+        let levels: Vec<&str> = snap.entries.iter().map(|e| e.level).collect();
+        assert_eq!(levels, vec!["ERROR"]);
+        assert_eq!(snap.latest_id, 3);
+    }
+
+    #[test]
+    fn snapshot_respects_limit() {
+        let buf = LogRingBuffer::new(16);
+        for id in 1..=10 {
+            buf.push(make_entry(id, "INFO", "m"));
+        }
+        let snap = buf.snapshot(0, level_severity("TRACE"), 4);
+        assert_eq!(snap.entries.len(), 4);
+        assert_eq!(snap.latest_id, 10);
+    }
+
+    #[test]
+    fn visitor_concatenates_message_and_structured_fields() {
+        let mut v = MessageVisitor::default();
+        v.push_extra("count", format_args!("{}", 42u64));
+        v.push_extra("ok", format_args!("{}", true));
+        v.message.push_str("hello");
+        let out = v.into_message();
+        assert!(out.contains("hello"));
+        assert!(out.contains("count=42"));
+        assert!(out.contains("ok=true"));
+    }
+}

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

@@ -0,0 +1 @@
+pub mod log_buffer;

net-guardia/src/model/event.rs

@@ -1,13 +1,20 @@
 use std::fmt;
 use std::str::FromStr;
 
+use serde::Serialize;
+
 use crate::interface::communication::event::Event;
 
 // -- Detection Source ---------------------------------------------------------
 
 /// Identifies which detection subsystem produced a detection.
 /// Used for attribution tracking and cross-source deduplication.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+///
+/// Serialized as the canonical `Display` form ("ML", "Suricata", "Beaconing",
+/// "Correlation") so the WebSocket wire matches the SOAR `SingleSourceHigh`
+/// `value` field — frontend rendering and playbook authoring share one
+/// vocabulary.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
 pub enum DetectionSource {
     ML,
     Correlation,
@@ -66,8 +73,11 @@ pub struct DetectionEvent {
 // -- Threat Events ------------------------------------------------------------
 
 /// Fired when the DetectionOrchestrator emits a deduplicated, enriched threat.
-/// Consumed by the SOAR engine to trigger automated responses.
-#[derive(Debug, Clone)]
+/// Consumed by the SOAR engine to trigger automated responses, and broadcast
+/// to the dashboard over `/ws/fusion` so the operator's "Recent Threats"
+/// stream surfaces post-fusion (multi-source) detections rather than raw
+/// per-flow ML alerts.
+#[derive(Debug, Clone, Serialize)]
 pub struct ThreatDetectedEvent {
     pub attack_type: String,
     pub confidence: f32,

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

@@ -11,5 +11,8 @@ loggable! {
 
         #[error("Setup HTTP server error: {error}")]
         SetupServerError { error: String } => tracing::Level::ERROR,
+
+        #[error("Fusion WebSocket failed to subscribe to ThreatDetectedEvent: {err}")]
+        FusionSubscribeFailed { err: String } => tracing::Level::ERROR,
     }
 }

net-guardia/src/utils/logging.rs

@@ -10,6 +10,7 @@ use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::reload;
 use tracing_subscriber::util::SubscriberInitExt;
 
+use crate::core::observability::log_buffer::LogBufferLayer;
 use crate::model::error::Error;
 use crate::model::error::io::IOError;
 
@@ -78,6 +79,7 @@ impl Logging {
             .with(filter_layer)
             .with(stdout_layer)
             .with(file_layer)
+            .with(LogBufferLayer::new())
             .init();
 
         // Store type-erased handle for runtime log level changes
@@ -102,14 +104,47 @@ impl Logging {
 
         handle.reload_filter(new_filter)?;
 
-        Ok(parsed_level.to_string())
+        Ok(parsed_level.to_string().to_lowercase())
     }
 
-    /// Get the current log level filter string.
+    /// Get the current global log level as a bare lowercase directive —
+    /// e.g. `"info"`, not the full `"maxminddb=warn,info"` EnvFilter string.
+    /// Per-target overrides (like `maxminddb=warn`) are internal tuning and
+    /// would break the frontend `<select>` that only knows five options.
     pub fn current_level() -> String {
         FILTER_HANDLE
             .get()
-            .map(|h| h.current_filter())
+            .map(|h| extract_main_level(&h.current_filter()))
             .unwrap_or_else(|| "unknown".to_string())
     }
 }
+
+/// Strip per-target directives out of an EnvFilter string and return the
+/// bare level directive in lowercase. Falls back to the raw string if no
+/// bare directive is present.
+fn extract_main_level(raw: &str) -> String {
+    raw.split(',')
+        .map(str::trim)
+        .find(|d| !d.is_empty() && !d.contains('='))
+        .unwrap_or(raw)
+        .to_lowercase()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn strips_per_target_directives() {
+        assert_eq!(extract_main_level("maxminddb=warn,debug"), "debug");
+        assert_eq!(extract_main_level("info,maxminddb=warn"), "info");
+        assert_eq!(extract_main_level("DEBUG"), "debug");
+    }
+
+    #[test]
+    fn falls_back_when_no_bare_level() {
+        // Only per-target directives → return lowercased raw so the UI at
+        // least shows *something* rather than silently misleading.
+        assert_eq!(extract_main_level("maxminddb=warn"), "maxminddb=warn");
+    }
+}