fix: flow WS — server-side throughput, no-clear tracker, useFlowData hook

Backend:
- FlowTracker: ML uses get_uninferred_flows() instead of take_snapshot()
  (tracker never cleared, WS always has data, LRU manages memory)
- FlowSummary: server-side bps calculation per direction + ip_version
- flow_websocket: pushes {summary, flows} instead of bare FlowStatsEntry[]

Frontend (submodule updated):
- New useFlowData hook replaces 300 lines of duplicated throttle logic
- BehaviorSubject → Subject for flows channel (no stale replay)
- Dashboard: shows bytes/sec rate, IPv6 working, formatRate with /s
- Update frequency controls backend WS interval directly

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

net-guardia-frontend

@@ -1 +1 @@
-Subproject commit c38d20e10bfb09896743bc62a55b1b314c1c8028
+Subproject commit 4066e24b0412c10da8caa8ec4e82e580322646f1

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

@@ -18,6 +18,12 @@ fn default_subscription() -> FlowSubscription {
     }
 }
 
+/// Push { summary, flows } payload to the client.
+fn push_payload(stats: &FlowStatistics, sub: &FlowSubscription) -> Option<String> {
+    let payload = stats.get_flow_payload(sub);
+    serde_json::to_string(&payload).ok()
+}
+
 pub async fn flow_stats_ws(
     req: HttpRequest,
     body: web::Payload,
@@ -34,8 +40,7 @@ pub async fn flow_stats_ws(
         loop {
             tokio::select! {
                 _ = ticker.tick() => {
-                    let flows = stats.get_filtered_flows(&subscription);
-                    if let Ok(json) = serde_json::to_string(&flows)
+                    if let Some(json) = push_payload(&stats, &subscription)
                         && session.text(json).await.is_err() {
                             break;
                     }
@@ -50,8 +55,7 @@ pub async fn flow_stats_ws(
                                     subscription.interval_secs = Some(new_interval);
                                     ticker = interval(Duration::from_secs(new_interval));
 
-                                    let flows = stats.get_filtered_flows(&subscription);
-                                    if let Ok(json) = serde_json::to_string(&flows)
+                                    if let Some(json) = push_payload(&stats, &subscription)
                                         && session.text(json).await.is_err() {
                                             break;
                                     }

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

@@ -103,23 +103,20 @@ impl Engine {
     }
 
     fn run_inference_tick(&self) {
-        let mut all_snapshots = Vec::new();
+        let mut all_flows = Vec::new();
         let mut total_count = 0;
 
-        // Phase 1: O(1) lock per tracker — just swap
+        // Phase 1: short lock per tracker — clone uninferred flows, mark as inferred
         for tracker in &self.trackers {
             let mut t = tracker.lock();
             total_count += t.flow_count();
-            all_snapshots.push(t.take_snapshot());
+            all_flows.extend(
+                t.get_uninferred_flows().into_iter()
+                    .filter(|flow| flow.packet_count() >= self.min_packets)
+            );
             // lock released here
         }
 
-        // Phase 2: filter outside all locks — O(flows) but non-blocking
-        let all_flows: Vec<FlowData> = all_snapshots.into_iter()
-            .flat_map(|map| map.into_values())
-            .filter(|flow| flow.packet_count() >= self.min_packets)
-            .collect();
-
         log!(MLLog::FlowStats(
             total_count,
             all_flows.len(),

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

@@ -34,6 +34,9 @@ pub struct FlowData {
     pub bwd_bulk_state: BulkState,
     pub act_data_pkt_fwd: u32,
     is_first_packet: bool,
+    /// Timestamp (us) when this flow was last sent to ML inference.
+    /// 0 means never inferred. Used to avoid re-inferring unchanged flows.
+    pub last_inferred_us: u64,
 }
 
 impl FlowData {
@@ -66,6 +69,7 @@ impl FlowData {
             bwd_bulk_state: BulkState::default(),
             act_data_pkt_fwd: 0,
             is_first_packet: true,
+            last_inferred_us: 0,
         }
     }
 
@@ -186,14 +190,6 @@ impl FlowTracker {
         }
     }
 
-    /// Swap active flows with an empty map and return the old one.
-    /// This is O(1) — the caller filters outside the lock.
-    pub fn take_snapshot(&mut self) -> HashMap<FlowKey, FlowData> {
-        let mut snapshot = HashMap::with_capacity(self.active.capacity());
-        std::mem::swap(&mut self.active, &mut snapshot);
-        snapshot
-    }
-
     pub fn process_packet(&mut self, mut packet: UserPacket, is_ingress: bool) {
         let packet_key = FlowKey::from_packet(&packet);
         let reversed_key = packet_key.clone().reverse();
@@ -247,11 +243,24 @@ impl FlowTracker {
         }
     }
 
-    /// Get a snapshot without draining.
+    /// Get all active flows (clone, no drain). Used by WebSocket.
     pub fn get_flows(&self) -> Vec<FlowData> {
         self.active.values().cloned().collect()
     }
 
+    /// Get flows that received new packets since their last inference,
+    /// and mark them as inferred. Used by ML engine.
+    pub fn get_uninferred_flows(&mut self) -> Vec<FlowData> {
+        let mut result = Vec::new();
+        for flow in self.active.values_mut() {
+            if flow.last_time_us > flow.last_inferred_us {
+                result.push(flow.clone());
+                flow.last_inferred_us = flow.last_time_us;
+            }
+        }
+        result
+    }
+
     pub fn flow_count(&self) -> usize {
         self.active.len()
     }

net-guardia/src/infrastructure/statistics.rs

@@ -3,7 +3,8 @@ use std::time;
 
 use crate::core::ml::engine::Engine;
 use crate::core::ml::flow_tracker::FlowData;
-use crate::model::flow_stats::{FlowStatsEntry, FlowSubscription, StatsSummary};
+use crate::model::direction::Direction;
+use crate::model::flow_stats::{FlowPushPayload, FlowStatsEntry, FlowSubscription, FlowSummary, StatsSummary};
 
 /// Conversion from core::ml::FlowData to model::FlowStatsEntry.
 /// Placed here (core layer) to maintain dependency rule: model/ must not import core/.
@@ -11,6 +12,7 @@ impl From<&FlowData> for FlowStatsEntry {
     fn from(flow: &FlowData) -> Self {
         Self {
             direction: flow.direction,
+            ip_version: flow.flow_key.ip_version,
             src_ip: flow.flow_key.src_ip_string(),
             dst_ip: flow.flow_key.dst_ip_string(),
             src_port: flow.flow_key.src_port,
@@ -81,6 +83,43 @@ impl FlowStatistics {
         })
     }
 
+    pub fn get_flow_payload(&self, sub: &FlowSubscription) -> FlowPushPayload {
+        let flows = self.get_filtered_flows(sub);
+        let window_secs = sub.window_secs.unwrap_or(60).max(1);
+
+        let mut ingress_bytes_v4: u64 = 0;
+        let mut egress_bytes_v4: u64 = 0;
+        let mut ingress_bytes_v6: u64 = 0;
+        let mut egress_bytes_v6: u64 = 0;
+
+        for f in &flows {
+            let bytes = f.fwd_bytes + f.bwd_bytes;
+            match (f.direction, f.ip_version) {
+                (Direction::Ingress, 6) => ingress_bytes_v6 += bytes,
+                (Direction::Ingress, _) => ingress_bytes_v4 += bytes,
+                (Direction::Egress, 6) => egress_bytes_v6 += bytes,
+                (Direction::Egress, _) => egress_bytes_v4 += bytes,
+            }
+        }
+
+        let now_ms = time::SystemTime::now()
+            .duration_since(time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+
+        let summary = FlowSummary {
+            ingress_bps_v4: ingress_bytes_v4 / window_secs,
+            egress_bps_v4: egress_bytes_v4 / window_secs,
+            ingress_bps_v6: ingress_bytes_v6 / window_secs,
+            egress_bps_v6: egress_bytes_v6 / window_secs,
+            total_flows: flows.len(),
+            window_secs,
+            timestamp_ms: now_ms,
+        };
+
+        FlowPushPayload { summary, flows }
+    }
+
     pub fn get_summary(&self) -> StatsSummary {
         let flows = self.get_all_flows();
         let total_flows = flows.len();

net-guardia/src/model/monitoring/flow_stats.rs

@@ -5,6 +5,7 @@ use crate::model::direction::Direction;
 #[derive(Debug, Clone, Serialize)]
 pub struct FlowStatsEntry {
     pub direction: Direction,
+    pub ip_version: u8,
     pub src_ip: String,
     pub dst_ip: String,
     pub src_port: u16,
@@ -28,6 +29,26 @@ pub struct StatsSummary {
     pub total_packets: usize,
 }
 
+/// Real-time throughput summary pushed via WebSocket alongside flow data.
+/// Rates are calculated server-side using actual elapsed time.
+#[derive(Debug, Clone, Serialize)]
+pub struct FlowSummary {
+    pub ingress_bps_v4: u64,
+    pub egress_bps_v4: u64,
+    pub ingress_bps_v6: u64,
+    pub egress_bps_v6: u64,
+    pub total_flows: usize,
+    pub window_secs: u64,
+    pub timestamp_ms: u64,
+}
+
+/// Combined payload for the flow WebSocket: summary + individual flows.
+#[derive(Debug, Clone, Serialize)]
+pub struct FlowPushPayload {
+    pub summary: FlowSummary,
+    pub flows: Vec<FlowStatsEntry>,
+}
+
 /// Client subscription filter for WebSocket flow stats
 #[derive(Debug, Clone, Deserialize)]
 pub struct FlowSubscription {