feat(fusion): explain endpoint + RBAC for fusion & flow-trace scopes

Surfaces the WORM audit evidence chain as a per-IP timeline so analysts
can answer "why was this IP blocked?" without grepping audit_log by
hand, and puts the fusion / flow-trace scopes under the existing
permission machinery so viewers can read evidence while writes stay
admin-only.

Repo + port:
- `AuditRepo::list_audit_logs_by_action(action, limit)` — indexed
  filter on the `action` column, newest-first, so the explain handler
  doesn't pull the whole chain to search for `fused_threat_emitted`
  entries.
- Same method lands on the `Database` impl with a straight
  `WHERE action = ?1 ORDER BY id DESC LIMIT ?2` query.

Handler (`adapter/http/fusion.rs`):
- `GET /api/fusion/explain/{src_ip}` scans up to 5 000 most-recent
  `fused_threat_emitted` rows, filters JSON `detail.src_ip ==
  {src_ip}`, and returns the first 200 matches oldest-first so the
  UI renders a chronological timeline. Response carries
  `match_count` + `truncated` so the client knows when the 200-entry
  cap clipped the history.
- `filter_fusion_evidence_for_ip` extracted as a free function so
  tests cover filter / ordering / cap / malformed-detail behaviour
  without an in-memory DB.

RBAC:
- Seed data in `seed_default_user_groups` adds `fusion:read` /
  `fusion:write` / `flow_trace:read` / `flow_trace:write` to the
  Administrator group, and `fusion:read` / `flow_trace:read` to the
  Viewer group. A viewer can now read explain timelines and flow-trace
  file listings without seeing the admin-only write actions that
  later milestones will add.
- `core/auth/middleware.rs::required_permission` maps `/api/fusion/`
  → `fusion` resource and `/api/flow-trace/` → `flow_trace`. Method
  still decides `:read` vs `:write`. This also closes an I-7 gap —
  flow-trace routes were previously falling through the permission
  match and relying on JWT presence alone.

Wiring:
- `http_server.rs` injects `Arc<dyn AuditRepo>` as Actix `app_data`
  on both the setup and main App builders (same pattern as `AppRepo`
  / `ApiKeyRepo`).
- Cleaned up a stale `let _ = engine;` placeholder in
  `adapter/http/flow_trace.rs` that shadowed the active use of the
  engine reference.

Tests: 6 new (only-matching-IP filter, oldest-first sort regardless
of input order, response cap trims oldest-take, malformed detail
dropped silently, non-matching IP returns empty, detail JSON
structure preserved end-to-end). 257 pass total. clippy --package
net-guardia -- -D warnings clean.

Closes A-region F-6 backend.

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

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

@@ -105,7 +105,6 @@ pub fn is_safe_flow_trace_name(name: &str) -> bool {
 /// `Engine` if the logger is active. Returns `None` when Flow Trace
 /// isn't enabled (Dormant state).
 fn flow_trace_directory(engine: &web::Data<Engine>) -> Option<PathBuf> {
-    let _ = engine; // placeholder until Engine exposes logger directory
     engine.traffic_logger_directory().map(Path::to_path_buf)
 }
 

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

@@ -1,16 +1,38 @@
-//! 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.
+//! HTTP surface for fusion-layer observability + incident explain.
+//! Metrics handlers read shared atomic counters maintained by the
+//! detection orchestrator — they never touch orchestrator state, so a
+//! hung dashboard cannot stall the detection pipeline. The explain
+//! handler reads the WORM audit chain populated by
+//! `publish_fusion_audit` and surfaces a per-IP evidence timeline so
+//! analysts can answer "why was this IP blocked?" without parsing
+//! logs by hand.
 
 use std::sync::Arc;
 
-use actix_web::{HttpResponse, Responder, Scope, web};
+use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web};
 
 use crate::core::detection::metrics::FusionMetrics;
+use crate::interface::port::audit::AuditRepo;
+
+/// Maximum audit rows scanned per explain request. Caps DB work in
+/// case the audit chain grows large enough that a naive full-table
+/// scan would be noticeable.
+const FUSION_EXPLAIN_SCAN_LIMIT: i64 = 5_000;
+
+/// Upper cap on entries returned to the client per explain request.
+/// Guards against a UI rendering path that chokes on enormous JSON.
+const FUSION_EXPLAIN_RESPONSE_CAP: usize = 200;
+
+/// Stable audit action string the fusion engine emits — kept in sync
+/// with `core::detection::orchestrator::FUSION_AUDIT_ACTION`. If that
+/// constant changes, the explain endpoint silently returns nothing, so
+/// keep this updated at the same time.
+const FUSION_AUDIT_ACTION: &str = "fused_threat_emitted";
 
 pub fn initialize() -> Scope {
-    web::scope("/fusion").route("/metrics", web::get().to(get_metrics))
+    web::scope("/fusion")
+        .route("/metrics", web::get().to(get_metrics))
+        .route("/explain/{src_ip}", web::get().to(explain_ip))
 }
 
 /// `GET /api/fusion/metrics` — lock-free snapshot of fusion counters and
@@ -19,3 +41,166 @@ pub fn initialize() -> Scope {
 async fn get_metrics(metrics: web::Data<Arc<FusionMetrics>>) -> impl Responder {
     HttpResponse::Ok().json(metrics.snapshot())
 }
+
+/// `GET /api/fusion/explain/{src_ip}` — per-IP fusion evidence timeline.
+/// Scans the WORM audit chain for `fused_threat_emitted` entries that
+/// match `src_ip`, returning them oldest-first so the UI can render a
+/// chronological "why was this IP blocked" view.
+async fn explain_ip(req: HttpRequest, audit: web::Data<Arc<dyn AuditRepo>>) -> impl Responder {
+    let src_ip = match req.match_info().get("src_ip") {
+        Some(ip) => ip.to_string(),
+        None => {
+            return HttpResponse::BadRequest().json(serde_json::json!({
+                "error": "missing src_ip path segment",
+            }));
+        }
+    };
+
+    let entries = match audit.list_audit_logs_by_action(FUSION_AUDIT_ACTION, FUSION_EXPLAIN_SCAN_LIMIT) {
+        Ok(e) => e,
+        Err(e) => {
+            return HttpResponse::InternalServerError().json(serde_json::json!({
+                "error": format!("audit store unavailable: {e}"),
+            }));
+        }
+    };
+
+    let matches = filter_fusion_evidence_for_ip(&entries, &src_ip, FUSION_EXPLAIN_RESPONSE_CAP);
+    HttpResponse::Ok().json(serde_json::json!({
+        "src_ip": src_ip,
+        "match_count": matches.len(),
+        "truncated": matches.len() >= FUSION_EXPLAIN_RESPONSE_CAP,
+        "entries": matches,
+    }))
+}
+
+/// Filter audit entries down to the ones whose JSON detail's `src_ip`
+/// matches `target_ip`, ordered oldest-first (ascending id). Entries
+/// with unparseable detail are dropped silently — the chain is
+/// append-only, so a malformed row is an integrity concern for the
+/// audit-verify endpoint to surface, not this handler.
+///
+/// Extracted as a free function so tests can cover the filter /
+/// ordering / cap behaviour without an in-memory DB.
+pub fn filter_fusion_evidence_for_ip(
+    entries: &[crate::interface::port::audit::AuditLogEntry],
+    target_ip: &str,
+    cap: usize,
+) -> Vec<serde_json::Value> {
+    let mut filtered: Vec<&crate::interface::port::audit::AuditLogEntry> = entries
+        .iter()
+        .filter(|entry| detail_matches_src_ip(&entry.detail, target_ip))
+        .collect();
+    filtered.sort_by_key(|entry| entry.id);
+    filtered
+        .into_iter()
+        .take(cap)
+        .map(|entry| {
+            let detail: serde_json::Value = serde_json::from_str(&entry.detail).unwrap_or(serde_json::Value::Null);
+            serde_json::json!({
+                "id": entry.id,
+                "actor": entry.actor,
+                "action": entry.action,
+                "created_at": entry.created_at,
+                "detail": detail,
+            })
+        })
+        .collect()
+}
+
+fn detail_matches_src_ip(detail_json: &str, target_ip: &str) -> bool {
+    let parsed: serde_json::Value = match serde_json::from_str(detail_json) {
+        Ok(v) => v,
+        Err(_) => return false,
+    };
+    parsed.get("src_ip").and_then(|v| v.as_str()) == Some(target_ip)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::interface::port::audit::AuditLogEntry;
+
+    fn entry(id: i64, src_ip: &str, attack: &str) -> AuditLogEntry {
+        let detail = serde_json::json!({
+            "src_ip": src_ip,
+            "attack_type": attack,
+            "fused_confidence": 0.9,
+            "per_source": [{"source": "Suricata", "confidence": 0.9, "local_attack_type": "brute-force"}],
+        })
+        .to_string();
+        AuditLogEntry {
+            id,
+            actor: "FusionEngine".to_string(),
+            action: "fused_threat_emitted".to_string(),
+            detail,
+            created_at: format!("2026-04-18T10:00:{:02}Z", id),
+        }
+    }
+
+    #[test]
+    fn filter_returns_only_matching_src_ip() {
+        let entries = [
+            entry(1, "1.2.3.4", "brute_force"),
+            entry(2, "10.0.0.5", "port_scan"),
+            entry(3, "1.2.3.4", "exploit"),
+        ];
+        let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
+        assert_eq!(got.len(), 2);
+        assert_eq!(got[0]["id"], 1);
+        assert_eq!(got[1]["id"], 3);
+    }
+
+    #[test]
+    fn filter_sorts_oldest_first_even_when_input_is_reversed() {
+        // Real repo query returns DESC; filter must still hand back ASC.
+        let entries = [
+            entry(30, "1.1.1.1", "a"),
+            entry(10, "1.1.1.1", "b"),
+            entry(20, "1.1.1.1", "c"),
+        ];
+        let got = filter_fusion_evidence_for_ip(&entries, "1.1.1.1", 100);
+        let ids: Vec<i64> = got.iter().map(|v| v["id"].as_i64().unwrap()).collect();
+        assert_eq!(ids, vec![10, 20, 30]);
+    }
+
+    #[test]
+    fn filter_applies_response_cap() {
+        let entries: Vec<AuditLogEntry> = (1..=10).map(|i| entry(i, "9.9.9.9", "x")).collect();
+        let got = filter_fusion_evidence_for_ip(&entries, "9.9.9.9", 3);
+        assert_eq!(got.len(), 3);
+        let ids: Vec<i64> = got.iter().map(|v| v["id"].as_i64().unwrap()).collect();
+        assert_eq!(ids, vec![1, 2, 3], "cap takes oldest, not newest");
+    }
+
+    #[test]
+    fn filter_drops_rows_with_unparseable_detail() {
+        let good = entry(1, "1.2.3.4", "brute_force");
+        let bad = AuditLogEntry {
+            id: 2,
+            actor: "FusionEngine".into(),
+            action: "fused_threat_emitted".into(),
+            detail: "{{not json".into(),
+            created_at: "2026-04-18T10:00:02Z".into(),
+        };
+        let got = filter_fusion_evidence_for_ip(&[good, bad], "1.2.3.4", 100);
+        assert_eq!(got.len(), 1);
+        assert_eq!(got[0]["id"], 1);
+    }
+
+    #[test]
+    fn filter_nonmatching_ip_returns_empty() {
+        let entries = [entry(1, "1.2.3.4", "brute_force")];
+        assert!(filter_fusion_evidence_for_ip(&entries, "5.6.7.8", 100).is_empty());
+    }
+
+    #[test]
+    fn filter_preserves_detail_structure_in_response() {
+        let entries = [entry(1, "1.2.3.4", "brute_force")];
+        let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100);
+        assert_eq!(got.len(), 1);
+        let detail = &got[0]["detail"];
+        assert_eq!(detail["attack_type"], "brute_force");
+        assert_eq!(detail["per_source"][0]["source"], "Suricata");
+    }
+}

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

@@ -366,7 +366,11 @@ impl Database {
                 "system:admin",
                 "users:read",
                 "users:write",
-                "users:admin"
+                "users:admin",
+                "fusion:read",
+                "fusion:write",
+                "flow_trace:read",
+                "flow_trace:write"
             ])
             .to_string();
             let viewer_permissions = serde_json::json!([
@@ -380,7 +384,9 @@ impl Database {
                 "dns_filter:read",
                 "rate_limit:read",
                 "protocol_filter:read",
-                "system:read"
+                "system:read",
+                "fusion:read",
+                "flow_trace:read"
             ])
             .to_string();
 
@@ -1652,6 +1658,28 @@ impl Database {
         Ok(rows)
     }
 
+    /// Read audit entries whose `action` matches exactly, newest-first,
+    /// capped at `limit`. Drives the fusion explain endpoint.
+    pub fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result<Vec<AuditLogEntry>, Error> {
+        let conn = self.conn()?;
+        let mut stmt = conn.prepare(
+            "SELECT id, actor, action, detail, ts FROM audit_log WHERE action = ?1 ORDER BY id DESC LIMIT ?2",
+        )?;
+        let rows = stmt
+            .query_map(params![action, limit], |row| {
+                Ok(AuditLogEntry {
+                    id: row.get(0)?,
+                    actor: row.get(1)?,
+                    action: row.get(2)?,
+                    detail: row.get(3)?,
+                    created_at: row.get(4)?,
+                })
+            })?
+            .filter_map(|r| r.ok())
+            .collect();
+        Ok(rows)
+    }
+
     /// Walk the entire audit_log in id order and verify the hash chain.
     /// Returns `Ok(count)` on success; returns `Err` at the first mismatch,
     /// naming the offending row id and the kind of mismatch.
@@ -2067,6 +2095,9 @@ impl AuditRepo for Database {
     fn list_audit_logs(&self) -> Result<Vec<AuditLogEntry>, Error> {
         self.list_audit_logs()
     }
+    fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result<Vec<AuditLogEntry>, Error> {
+        self.list_audit_logs_by_action(action, limit)
+    }
     fn verify_audit_log_chain(&self) -> Result<usize, Error> {
         self.verify_audit_log_chain()
     }

net-guardia/src/core/auth/middleware.rs

@@ -49,6 +49,10 @@ fn required_permission(path: &str, method: &Method) -> Option<String> {
         "dashboard"
     } else if path.starts_with("/api/ml/") {
         "ai_detection"
+    } else if path.starts_with("/api/fusion/") {
+        "fusion"
+    } else if path.starts_with("/api/flow-trace/") {
+        "flow_trace"
     } else if path.starts_with("/api/acl/geo/") {
         "geo_block"
     } else if path.starts_with("/api/acl/") {

net-guardia/src/infrastructure/http_server.rs

@@ -35,6 +35,7 @@ use crate::infrastructure::suricata_manager::SuricataManager;
 use crate::infrastructure::system::ShutdownHandle;
 use crate::interface::port::api_key::ApiKeyRepo;
 use crate::interface::port::app_repo::AppRepo;
+use crate::interface::port::audit::AuditRepo;
 use crate::model::config::constants::HTTP_FALLBACK_PORT;
 use crate::model::error::Error;
 use crate::model::error::http::HttpError;
@@ -155,6 +156,7 @@ pub fn start_setup_server(
             .wrap(cors(vec![]))
             .app_data(web::Data::from(db.clone() as Arc<dyn AppRepo>))
             .app_data(web::Data::from(db.clone() as Arc<dyn ApiKeyRepo>))
+            .app_data(web::Data::from(db.clone() as Arc<dyn AuditRepo>))
             .app_data(web::Data::from(db.clone()))
             .app_data(web::Data::from(secret_store.clone()))
             .app_data(web::Data::from(jwt_service.clone()))
@@ -255,6 +257,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
             .app_data(web::Data::from(drop_monitor.clone()))
             .app_data(web::Data::from(db.clone() as Arc<dyn AppRepo>))
             .app_data(web::Data::from(db.clone() as Arc<dyn ApiKeyRepo>))
+            .app_data(web::Data::from(db.clone() as Arc<dyn AuditRepo>))
             .app_data(web::Data::from(db.clone()))
             .app_data(web::Data::from(secret_store.clone()))
             .app_data(web::Data::from(jwt_service.clone()))

net-guardia/src/interface/port/audit.rs

@@ -23,6 +23,12 @@ pub trait AuditRepo: Send + Sync {
     /// Read all audit entries ordered by id ASC.
     fn list_audit_logs(&self) -> Result<Vec<AuditLogEntry>, Error>;
 
+    /// Read audit entries whose `action` exactly matches, newest first,
+    /// capped at `limit`. Drives the fusion explain endpoint, which
+    /// filters on `fused_threat_emitted` rather than walking the full
+    /// chain for every request.
+    fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result<Vec<AuditLogEntry>, Error>;
+
     /// Walk the full chain and verify every `row_hash` matches
     /// `H(ts || actor || action || detail || prev_hash)`. Returns the number
     /// of entries verified. Errors on the first broken link.