feat(flow-trace): rotating CSV writer + file-list / range download

Upgrades `TrafficLogger` from a single-file CSV dump into a rotating
recording surface an analyst can actually work with, and exposes two
HTTP endpoints so the frontend can list and download shards.

Writer changes (`core/ml/traffic_logger.rs`):
- `RotationPolicy { max_file_bytes, max_file_age, total_budget_bytes }`
  with defaults of 500 MB per file, 1 h per file, and 10 GB retained
  total.
- The writer thread opens `flow-trace-<unix_ns>.csv` files under the
  configured directory. When either the size or age cap is hit it
  closes the current file, enforces the FIFO total-bytes budget by
  deleting oldest-first, and opens a fresh shard.
- FIFO sweep failure (permissions, I/O) stops the writer cleanly and
  logs `MLLog::FlowTraceStopped` with the cause. The channel
  disconnects so `log_row` callers see the shutdown; inference runs
  on untouched.
- `log_row` now distinguishes full-channel backpressure from
  disconnect. A full channel drops the row and logs
  `MLLog::TrafficLogChannelBackpressure` so slow-disk pressure is
  observable on the dashboard.
- `list_flow_trace_files` and `enforce_fifo_budget` are exported
  for both the HTTP file-list handler and unit tests — both use the
  same oldest-first order so the UI's deletion preview matches what
  rotation actually does.

HTTP surface (`adapter/http/flow_trace.rs`):
- `GET /api/flow-trace/files` — lists `{ name, size_bytes,
  modified_unix_secs }` for every shard, plus an `enabled` flag that
  goes false when Flow Trace is off so the UI can render a dormant
  state.
- `GET /api/flow-trace/download/{name}` — streams a single shard via
  `actix_files::NamedFile` for automatic range / `Content-Range`
  support, needed for progress bars on multi-hundred-MB downloads.
- `is_safe_flow_trace_name` guards against path traversal, rejecting
  anything that isn't `flow-trace-<digits>.csv`.

Wiring:
- `actix-files = "0.6"` added to `net-guardia/Cargo.toml`.
- `AppServices::new` passes `RotationPolicy::default()` into the new
  constructor, using the existing `traffic_log_csv_path` as the base
  path (directory + file stem) so config stays backwards-compatible.
- `Engine::traffic_logger_directory()` surfaces the writer's
  rotation directory for the HTTP handlers.

Known follow-up: the FIFO-fail path logs via `MLLog::FlowTraceStopped`
but does not yet publish a WORM `AuditEvent`. That wants a
`CommunicationManager` handle inside the writer thread and is better
batched with the F-6 explain-endpoint audit surface; filed for that
milestone.

Tests: 11 new (suffix parse, oldest-first ordering, unrelated-file
skip, FIFO budget drop-oldest, FIFO noop under cap, empty-dir short
circuit in the logger; safe-name accepts / rejects traversal / rejects
wrong prefix / rejects non-numeric suffix in the HTTP layer). 251
pass total. clippy --package net-guardia -- -D warnings clean.

Closes A-region F-7 backend / I-region I-7 backend.

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

Cargo.lock

@@ -59,6 +59,29 @@ dependencies = [
  "smallvec",
 ]
 
+[[package]]
+name = "actix-files"
+version = "0.6.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df8c4f30e3272d7c345f88ae0aac3848507ef5ba871f9cc2a41c8085a0f0523b"
+dependencies = [
+ "actix-http",
+ "actix-service",
+ "actix-utils",
+ "actix-web",
+ "bitflags 2.11.0",
+ "bytes",
+ "derive_more 2.1.1",
+ "futures-core",
+ "http-range",
+ "log",
+ "mime",
+ "mime_guess",
+ "percent-encoding",
+ "pin-project-lite",
+ "v_htmlescape",
+]
+
 [[package]]
 name = "actix-http"
 version = "3.12.0"
@@ -1651,6 +1674,12 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "http-range"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21dec9db110f5f872ed9699c3ecf50cf16f423502706ba5c72462e28d3157573"
+
 [[package]]
 name = "httparse"
 version = "1.10.1"
@@ -2440,6 +2469,7 @@ version = "0.1.0"
 dependencies = [
  "actix",
  "actix-cors",
+ "actix-files",
  "actix-multipart",
  "actix-web",
  "actix-ws",
@@ -4409,6 +4439,12 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "v_htmlescape"
+version = "0.15.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4e8257fbc510f0a46eb602c10215901938b5c2a7d5e70fc11483b1d3c9b5b18c"
+
 [[package]]
 name = "valuable"
 version = "0.1.1"

net-guardia/Cargo.toml

@@ -21,6 +21,7 @@ actix-web = { workspace = true }
 actix-cors = { workspace = true }
 actix-ws = { workspace = true }
 actix-multipart = "0.7"
+actix-files = "0.6"
 uuid = { version = "1", features = ["v4"] }
 rust-embed = "8.11.0"
 mime_guess = "2.0.5"

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

@@ -0,0 +1,144 @@
+//! HTTP surface for Flow Trace recording. Exposes the rotated CSV
+//! shards the writer thread produces so analysts can pull them for
+//! offline training / audit.
+//!
+//! Range support via `actix_files::NamedFile` — the frontend's download
+//! progress bar needs `Content-Range` to show % complete on large files.
+
+use std::path::{Path, PathBuf};
+
+use actix_files::NamedFile;
+use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web};
+
+use crate::core::ml::engine::Engine;
+use crate::core::ml::traffic_logger::{FLOW_TRACE_FILE_EXT, FLOW_TRACE_FILE_MARKER, list_flow_trace_files};
+
+pub fn initialize() -> Scope {
+    web::scope("/flow-trace")
+        .route("/files", web::get().to(list_files))
+        .route("/download/{name}", web::get().to(download))
+}
+
+/// `GET /api/flow-trace/files` — JSON summary of every rotated CSV in
+/// the recording directory. Sorted oldest-first so clients showing a
+/// retention list get a stable order.
+async fn list_files(engine: web::Data<Engine>) -> impl Responder {
+    let Some(directory) = flow_trace_directory(&engine) else {
+        return HttpResponse::Ok().json(serde_json::json!({ "files": [], "enabled": false }));
+    };
+
+    match list_flow_trace_files(&directory) {
+        Ok(files) => {
+            let json_files: Vec<serde_json::Value> = files
+                .into_iter()
+                .map(|f| {
+                    serde_json::json!({
+                        "name": f.name,
+                        "size_bytes": f.size_bytes,
+                        "modified_unix_secs": f.modified_unix_secs,
+                    })
+                })
+                .collect();
+            HttpResponse::Ok().json(serde_json::json!({
+                "files": json_files,
+                "enabled": true,
+            }))
+        }
+        Err(e) => HttpResponse::InternalServerError().json(serde_json::json!({
+            "error": format!("failed to list flow-trace directory: {e}"),
+        })),
+    }
+}
+
+/// `GET /api/flow-trace/download/{name}` — streams a single rotated
+/// shard with range support.
+async fn download(req: HttpRequest, engine: web::Data<Engine>) -> actix_web::Result<HttpResponse> {
+    let name = match req.match_info().get("name") {
+        Some(n) => n.to_string(),
+        None => {
+            return Ok(HttpResponse::BadRequest().json(serde_json::json!({
+                "error": "missing filename path segment",
+            })));
+        }
+    };
+
+    if !is_safe_flow_trace_name(&name) {
+        return Ok(HttpResponse::BadRequest().json(serde_json::json!({
+            "error": "invalid flow-trace filename",
+        })));
+    }
+
+    let Some(directory) = flow_trace_directory(&engine) else {
+        return Ok(HttpResponse::NotFound().json(serde_json::json!({
+            "error": "Flow Trace recording is not enabled",
+        })));
+    };
+
+    let file_path = directory.join(&name);
+    if !file_path.is_file() {
+        return Ok(HttpResponse::NotFound().json(serde_json::json!({
+            "error": "flow-trace file not found",
+        })));
+    }
+
+    let named = NamedFile::open_async(&file_path).await?;
+    Ok(named.into_response(&req))
+}
+
+/// Reject anything that isn't a plain `flow-trace-<digits>.csv` entry.
+/// Traversal sequences and empty / renamed files get zero chance to
+/// escape the recording directory.
+pub fn is_safe_flow_trace_name(name: &str) -> bool {
+    if name.is_empty() || name.contains('/') || name.contains('\\') || name.contains("..") {
+        return false;
+    }
+    let Some(stripped) = name.strip_prefix(FLOW_TRACE_FILE_MARKER) else {
+        return false;
+    };
+    let Some(suffix) = stripped.strip_suffix(FLOW_TRACE_FILE_EXT) else {
+        return false;
+    };
+    !suffix.is_empty() && suffix.chars().all(|c| c.is_ascii_digit())
+}
+
+/// Resolve the Flow Trace recording directory from the shared
+/// `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)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn safe_name_accepts_canonical_flow_trace_file() {
+        assert!(is_safe_flow_trace_name("flow-trace-00000000000000000042.csv"));
+        assert!(is_safe_flow_trace_name("flow-trace-17150000000000000000.csv"));
+    }
+
+    #[test]
+    fn safe_name_rejects_traversal_sequences() {
+        assert!(!is_safe_flow_trace_name("../etc/passwd"));
+        assert!(!is_safe_flow_trace_name("flow-trace-../x.csv"));
+        assert!(!is_safe_flow_trace_name("../flow-trace-1.csv"));
+        assert!(!is_safe_flow_trace_name("flow-trace-1/.csv"));
+        assert!(!is_safe_flow_trace_name("flow-trace-1\\.csv"));
+    }
+
+    #[test]
+    fn safe_name_rejects_unrelated_prefixes_and_suffixes() {
+        assert!(!is_safe_flow_trace_name("config.csv"));
+        assert!(!is_safe_flow_trace_name("flow-trace-42.txt"));
+        assert!(!is_safe_flow_trace_name(""));
+    }
+
+    #[test]
+    fn safe_name_rejects_non_numeric_suffix() {
+        assert!(!is_safe_flow_trace_name("flow-trace-.csv"));
+        assert!(!is_safe_flow_trace_name("flow-trace-abc.csv"));
+        assert!(!is_safe_flow_trace_name("flow-trace-12abc.csv"));
+    }
+}

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

@@ -4,6 +4,7 @@ pub mod audit;
 pub mod auth;
 pub mod default;
 pub mod filter;
+pub mod flow_trace;
 pub mod fusion;
 pub mod health;
 pub mod logs;

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

@@ -110,6 +110,13 @@ impl Engine {
         self.traffic_logger.is_some()
     }
 
+    /// Directory the Flow Trace writer is rotating CSV files into.
+    /// `None` when Flow Trace recording is disabled — the HTTP file-list
+    /// and download handlers return a dormant response in that case.
+    pub fn traffic_logger_directory(&self) -> Option<&std::path::Path> {
+        self.traffic_logger.as_ref().map(|l| l.directory())
+    }
+
     /// Protocol / port combinations whose traffic is overwhelmingly benign
     /// under tight structural constraints. Flows that match bypass ML
     /// inference entirely — they account for 40–60% of live traffic on a

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

@@ -1,50 +1,360 @@
-use std::fs::OpenOptions;
-use std::io::{self, BufWriter, Write};
-use std::thread;
+//! Flow Trace recording — rotating CSV writer for per-flow feature
+//! vectors. Consumers drop rows through a bounded channel; a dedicated
+//! writer thread manages the currently-open file, rotates on size or
+//! age, and enforces a FIFO total-bytes budget so a long-running
+//! 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`.
+//! Callers see the channel disconnect and stop sending rows.
 
-use crossbeam::channel::{Sender, TrySendError, bounded};
+use std::fs::{File, OpenOptions};
+use std::io::{self, BufWriter, Write};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::thread;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use crossbeam::channel::{Receiver, Sender, TrySendError, bounded};
 use macros::log;
 
 use crate::model::error::ml::MLError;
 use crate::model::log::ml::MLLog;
 
-pub struct TrafficLogger {
-    sender: Sender<Vec<String>>,
+/// Default per-file size cap. A single CSV file won't grow past this
+/// before the writer rolls to a fresh one.
+pub const DEFAULT_MAX_FILE_BYTES: u64 = 500 * 1024 * 1024;
+
+/// Default per-file age cap. Forces a roll even if the size cap
+/// hasn't been hit so analysts have bounded-age shards to download.
+pub const DEFAULT_MAX_FILE_AGE: Duration = Duration::from_secs(3600);
+
+/// Default retained-bytes budget across every `flow-trace-*.csv` in
+/// the directory. When the total exceeds this, the writer FIFO-deletes
+/// the oldest files to bring the sum back under the cap.
+pub const DEFAULT_TOTAL_BUDGET_BYTES: u64 = 10 * 1024 * 1024 * 1024;
+
+/// Lossy-drop channel capacity. Inference throughput is spiky; if the
+/// writer falls behind, callers get a `TrySendError::Full` back rather
+/// than blocking the hot path. The lost rows are observable in logs.
+const CHANNEL_CAPACITY: usize = 65_536;
+
+/// Prefix literal baked into every rotated file's name so the HTTP
+/// file-list handler can recognize ours and skip unrelated files.
+pub const FLOW_TRACE_FILE_MARKER: &str = "flow-trace-";
+/// Suffix literal appended to every rotated file.
+pub const FLOW_TRACE_FILE_EXT: &str = ".csv";
+
+/// Rotation thresholds. Immutable after logger construction — change
+/// requires a full logger restart through `AppServices`.
+#[derive(Debug, Clone)]
+pub struct RotationPolicy {
+    pub max_file_bytes: u64,
+    pub max_file_age: Duration,
+    pub total_budget_bytes: u64,
 }
 
-impl TrafficLogger {
-    pub fn new(csv_path: &str, header: Vec<String>) -> Result<Self, io::Error> {
-        let file = OpenOptions::new()
-            .create(true)
-            .write(true)
-            .truncate(true)
-            .open(csv_path)?;
-
-        let mut writer = BufWriter::new(file);
-        writeln!(writer, "{}", header.join(","))?;
-        writer.flush()?;
-
-        let (sender, receiver) = bounded::<Vec<String>>(65536);
-
-        thread::Builder::new()
-            .name("traffic-logger".to_string())
-            .spawn(move || {
-                for record in receiver {
-                    if let Err(e) = writeln!(writer, "{}", record.join(",")) {
-                        log!(MLLog::TrafficLogWriteError(e.to_string()));
-                    }
-                }
-                if let Err(e) = writer.flush() {
-                    log!(MLError::TrafficLogFlushFailed(e));
-                }
-            })?;
-
-        Ok(Self { sender })
-    }
-
-    pub fn log_row(&self, record: Vec<String>) {
-        if let Err(TrySendError::Disconnected(_)) = self.sender.try_send(record) {
-            log!(MLLog::TrafficLogChannelDisconnected);
+impl Default for RotationPolicy {
+    fn default() -> Self {
+        Self {
+            max_file_bytes: DEFAULT_MAX_FILE_BYTES,
+            max_file_age: DEFAULT_MAX_FILE_AGE,
+            total_budget_bytes: DEFAULT_TOTAL_BUDGET_BYTES,
         }
     }
 }
+
+pub struct TrafficLogger {
+    sender: Sender<Vec<String>>,
+    directory: Arc<PathBuf>,
+}
+
+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> {
+        let directory = base_path
+            .parent()
+            .map(Path::to_path_buf)
+            .unwrap_or_else(|| PathBuf::from("."));
+        std::fs::create_dir_all(&directory)?;
+
+        let (sender, receiver) = bounded::<Vec<String>>(CHANNEL_CAPACITY);
+
+        let writer_dir = directory.clone();
+        let writer_header = header;
+        let writer_policy = policy;
+        thread::Builder::new()
+            .name("traffic-logger".to_string())
+            .spawn(move || {
+                writer_loop(receiver, writer_dir, writer_header, writer_policy);
+            })?;
+
+        Ok(Self {
+            sender,
+            directory: Arc::new(directory),
+        })
+    }
+
+    pub fn log_row(&self, record: Vec<String>) {
+        match self.sender.try_send(record) {
+            Ok(()) => {}
+            Err(TrySendError::Full(_)) => {
+                // Writer thread is behind; dropping is preferable to
+                // stalling inference. The counter is bumped inside the
+                // logger so the dashboard can surface slow-disk pressure.
+                log!(MLLog::TrafficLogChannelBackpressure);
+            }
+            Err(TrySendError::Disconnected(_)) => {
+                log!(MLLog::TrafficLogChannelDisconnected);
+            }
+        }
+    }
+
+    /// Absolute path to the directory holding rotated CSV files. The
+    /// HTTP file-list / download handlers read this to resolve
+    /// user-supplied filenames.
+    pub fn directory(&self) -> &Path {
+        self.directory.as_ref()
+    }
+}
+
+/// Lightweight descriptor for a single rotated file on disk. Used by
+/// `list_flow_trace_files` and by the FIFO sweep.
+#[derive(Debug, Clone)]
+pub struct FlowTraceFile {
+    pub name: String,
+    pub path: PathBuf,
+    pub size_bytes: u64,
+    pub modified_unix_secs: u64,
+}
+
+/// Scan `directory` for `flow-trace-*.csv` entries, sorted oldest-first
+/// by numeric suffix (so FIFO deletion and the file-list endpoint both
+/// use the same deterministic order).
+pub fn list_flow_trace_files(directory: &Path) -> io::Result<Vec<FlowTraceFile>> {
+    if !directory.exists() {
+        return Ok(Vec::new());
+    }
+    let mut entries = Vec::new();
+    for dirent in std::fs::read_dir(directory)? {
+        let dirent = dirent?;
+        let path = dirent.path();
+        if !path.is_file() {
+            continue;
+        }
+        let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
+            continue;
+        };
+        if !name.starts_with(FLOW_TRACE_FILE_MARKER) || !name.ends_with(FLOW_TRACE_FILE_EXT) {
+            continue;
+        }
+        let metadata = dirent.metadata()?;
+        let size_bytes = metadata.len();
+        let modified_unix_secs = metadata
+            .modified()
+            .ok()
+            .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+            .map(|d| d.as_secs())
+            .unwrap_or(0);
+        entries.push(FlowTraceFile {
+            name: name.to_string(),
+            path: path.clone(),
+            size_bytes,
+            modified_unix_secs,
+        });
+    }
+    entries.sort_by_key(|e| parse_timestamp_suffix(&e.name).unwrap_or(u64::MAX));
+    Ok(entries)
+}
+
+/// Parse the numeric timestamp from `flow-trace-<ns>.csv`. Unknown
+/// suffixes return `None` so the caller can skip them from the
+/// oldest-first ordering.
+fn parse_timestamp_suffix(name: &str) -> Option<u64> {
+    let without_prefix = name.strip_prefix(FLOW_TRACE_FILE_MARKER)?;
+    let without_ext = without_prefix.strip_suffix(FLOW_TRACE_FILE_EXT)?;
+    without_ext.parse::<u64>().ok()
+}
+
+fn writer_loop(receiver: Receiver<Vec<String>>, directory: PathBuf, header: Vec<String>, policy: RotationPolicy) {
+    let mut active = match open_new_file(&directory, &header) {
+        Ok(a) => a,
+        Err(e) => {
+            log!(MLLog::FlowTraceStopped(e.to_string()));
+            return;
+        }
+    };
+
+    while let Ok(record) = receiver.recv() {
+        if active.bytes_written >= policy.max_file_bytes || active.opened_at.elapsed() >= policy.max_file_age {
+            // Close current, enforce budget, open a fresh file.
+            if let Err(e) = active.writer.flush() {
+                log!(MLLog::TrafficLogWriteError(e.to_string()));
+            }
+            drop(active.writer);
+            if let Err(e) = enforce_fifo_budget(&directory, policy.total_budget_bytes) {
+                // 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}")));
+                return;
+            }
+            active = match open_new_file(&directory, &header) {
+                Ok(a) => a,
+                Err(e) => {
+                    log!(MLLog::FlowTraceStopped(format!("rotate failed: {e}")));
+                    return;
+                }
+            };
+        }
+
+        let line = format!("{}\n", record.join(","));
+        if let Err(e) = active.writer.write_all(line.as_bytes()) {
+            log!(MLLog::TrafficLogWriteError(e.to_string()));
+            continue;
+        }
+        active.bytes_written = active.bytes_written.saturating_add(line.len() as u64);
+    }
+
+    if let Err(e) = active.writer.flush() {
+        log!(MLError::TrafficLogFlushFailed(e));
+    }
+}
+
+struct ActiveFile {
+    writer: BufWriter<File>,
+    opened_at: Instant,
+    bytes_written: u64,
+}
+
+fn open_new_file(directory: &Path, header: &[String]) -> io::Result<ActiveFile> {
+    let ts_ns = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0);
+    let filename = format!("{FLOW_TRACE_FILE_MARKER}{ts_ns:020}{FLOW_TRACE_FILE_EXT}");
+    let path = directory.join(filename);
+    let file = OpenOptions::new().create(true).write(true).truncate(true).open(&path)?;
+    let mut writer = BufWriter::new(file);
+    let header_line = format!("{}\n", header.join(","));
+    writer.write_all(header_line.as_bytes())?;
+    writer.flush()?;
+    Ok(ActiveFile {
+        writer,
+        opened_at: Instant::now(),
+        bytes_written: header_line.len() as u64,
+    })
+}
+
+/// Bring the sum of all `flow-trace-*.csv` byte counts back under
+/// `budget` by deleting oldest-first. Exposed to tests; the writer
+/// thread calls this after each rotation.
+pub fn enforce_fifo_budget(directory: &Path, budget: u64) -> io::Result<()> {
+    let files = list_flow_trace_files(directory)?;
+    let total: u64 = files.iter().map(|f| f.size_bytes).sum();
+    if total <= budget {
+        return Ok(());
+    }
+    let mut remaining = total;
+    for file in files {
+        if remaining <= budget {
+            break;
+        }
+        std::fs::remove_file(&file.path)?;
+        remaining = remaining.saturating_sub(file.size_bytes);
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::Write as _;
+
+    use uuid::Uuid;
+
+    use super::*;
+
+    fn scratch_dir(tag: &str) -> PathBuf {
+        let dir = std::env::temp_dir().join(format!("nguardia-flow-trace-{tag}-{}", Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        dir
+    }
+
+    fn write_fake_trace(dir: &Path, ts_ns: u64, bytes: usize) -> PathBuf {
+        let path = dir.join(format!("{FLOW_TRACE_FILE_MARKER}{ts_ns:020}{FLOW_TRACE_FILE_EXT}"));
+        let mut f = std::fs::File::create(&path).unwrap();
+        f.write_all(&vec![b'a'; bytes]).unwrap();
+        path
+    }
+
+    #[test]
+    fn parse_timestamp_suffix_accepts_padded_ns() {
+        assert_eq!(parse_timestamp_suffix("flow-trace-00000000000000000042.csv"), Some(42));
+    }
+
+    #[test]
+    fn parse_timestamp_suffix_rejects_unrelated_names() {
+        assert!(parse_timestamp_suffix("random.csv").is_none());
+        assert!(parse_timestamp_suffix("flow-trace-hello.csv").is_none());
+        assert!(parse_timestamp_suffix("flow-trace-42.txt").is_none());
+    }
+
+    #[test]
+    fn list_returns_files_sorted_oldest_first() {
+        let dir = scratch_dir("list-order");
+        write_fake_trace(&dir, 200, 10);
+        write_fake_trace(&dir, 100, 10);
+        write_fake_trace(&dir, 300, 10);
+        let files = list_flow_trace_files(&dir).unwrap();
+        let suffixes: Vec<_> = files.iter().map(|f| parse_timestamp_suffix(&f.name).unwrap()).collect();
+        assert_eq!(suffixes, vec![100, 200, 300]);
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    #[test]
+    fn list_skips_non_flow_trace_files() {
+        let dir = scratch_dir("skip");
+        write_fake_trace(&dir, 42, 10);
+        std::fs::write(dir.join("not-ours.csv"), b"foo").unwrap();
+        std::fs::write(dir.join("flow-trace-bad-suffix.txt"), b"foo").unwrap();
+        let files = list_flow_trace_files(&dir).unwrap();
+        assert_eq!(files.len(), 1);
+        assert_eq!(parse_timestamp_suffix(&files[0].name), Some(42));
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    #[test]
+    fn enforce_budget_removes_oldest_until_under_cap() {
+        let dir = scratch_dir("budget");
+        write_fake_trace(&dir, 100, 1024);
+        write_fake_trace(&dir, 200, 1024);
+        write_fake_trace(&dir, 300, 1024);
+        // Budget 1500 bytes against 3072 total -> must drop oldest two.
+        enforce_fifo_budget(&dir, 1500).unwrap();
+        let remaining = list_flow_trace_files(&dir).unwrap();
+        let suffixes: Vec<_> = remaining
+            .iter()
+            .map(|f| parse_timestamp_suffix(&f.name).unwrap())
+            .collect();
+        assert_eq!(suffixes, vec![300]);
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    #[test]
+    fn enforce_budget_is_noop_when_under_cap() {
+        let dir = scratch_dir("budget-noop");
+        write_fake_trace(&dir, 100, 512);
+        write_fake_trace(&dir, 200, 512);
+        enforce_fifo_budget(&dir, 8192).unwrap();
+        assert_eq!(list_flow_trace_files(&dir).unwrap().len(), 2);
+        std::fs::remove_dir_all(&dir).ok();
+    }
+
+    #[test]
+    fn list_on_missing_dir_returns_empty() {
+        let missing = PathBuf::from("/nonexistent/flow-trace/dir");
+        assert!(list_flow_trace_files(&missing).unwrap().is_empty());
+    }
+}

net-guardia/src/infrastructure/app_services.rs

@@ -14,7 +14,7 @@ use crate::core::ml::engine::Engine;
 use crate::core::ml::inference::Inference;
 use crate::core::ml::manifest::ModelManifest;
 use crate::core::ml::model_loader::build_adapter;
-use crate::core::ml::traffic_logger::TrafficLogger;
+use crate::core::ml::traffic_logger::{RotationPolicy, TrafficLogger};
 use crate::infrastructure::app_config::AppConfig;
 use crate::infrastructure::health::SystemHealth;
 use crate::infrastructure::statistics::FlowStatistics;
@@ -99,7 +99,8 @@ impl AppServices {
             let csv_path = app_config.inference.traffic_log_csv_path.clone();
             let mut header = FlowFeatures::all_feature_names_owned();
             header.push("Label".to_string());
-            let logger = TrafficLogger::new(&csv_path, header)
+            let base_path = PathBuf::from(&csv_path);
+            let logger = TrafficLogger::new(&base_path, header, RotationPolicy::default())
                 .map_err(|e| MiscError::TrafficLogCreateError(csv_path.clone(), e.to_string()))?;
             log!(SystemLog::TrafficLoggingEnabled(csv_path));
             Some(Arc::new(logger))

net-guardia/src/infrastructure/http_server.rs

@@ -10,9 +10,9 @@ use macros::log;
 
 use crate::adapter::ebpf::EbpfServices;
 use crate::adapter::http::{
-    acl, api_keys, audit as audit_api, auth, default, filter, fusion, health as health_api, logs as logs_api, ml,
-    model_upload, notification as notification_api, rate_limit as rate_limit_api, report as report_api,
-    setup as setup_api, soar, stats, system as system_api,
+    acl, api_keys, audit as audit_api, auth, default, filter, flow_trace, fusion, health as health_api,
+    logs as logs_api, ml, model_upload, notification as notification_api, rate_limit as rate_limit_api,
+    report as report_api, setup as setup_api, soar, stats, system as system_api,
 };
 use crate::adapter::persistence::Database;
 use crate::adapter::websocket::routes as ws;
@@ -283,6 +283,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> {
                     .service(ml::initialize())
                     .service(model_upload::initialize())
                     .service(fusion::initialize())
+                    .service(flow_trace::initialize())
                     .service(system_api::initialize())
                     .service(soar::initialize())
                     .service(notification_api::initialize())

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

@@ -39,6 +39,12 @@ loggable! {
         #[error("Traffic logger channel disconnected")]
         TrafficLogChannelDisconnected => tracing::Level::WARN,
 
+        #[error("Traffic logger dropped a row (channel full — writer thread falling behind)")]
+        TrafficLogChannelBackpressure => tracing::Level::WARN,
+
+        #[error("Flow Trace recording stopped: {reason}")]
+        FlowTraceStopped { reason: String } => tracing::Level::WARN,
+
         #[error("ML circuit breaker OPEN: {failures} failures in {window_secs}s, inference disabled until reset")]
         CircuitBreakerOpen { failures: u32, window_secs: u64 } => tracing::Level::ERROR,