feat: add --replay mode for JSONL event replay without eBPF

Add a --replay <file> CLI flag that reads pre-recorded events from a
JSONL file instead of loading eBPF programs. This allows profiling the
entire userspace pipeline with tools like valgrind DHAT that don't
support BPF syscalls.

In replay mode, the BPF loader and HostScanner are bypassed entirely.
Events flow directly from the JSONL reader through the RateLimiter and
Output stages.

Changes:
- Add Deserialize impls for inode_key_t and monitored_t (fact-ebpf)
- Add Deserialize derives to Event, FileData, Process and related types
- Make kernel metrics optional in Exporter (None in replay mode)
- Migrate task handles to a shared JoinSet for cleaner lifecycle mgmt
- Add replay module with async JSONL file reader
- Add --replay flag to CLI and config

Assisted-by: `agent-mode` <noreply@opencode.ai>

Author: Molter73

fact-ebpf/src/lib.rs

@@ -4,7 +4,11 @@ use std::{error::Error, ffi::c_char, fmt::Display, hash::Hash, path::PathBuf};
 
 use aya::{maps::lpm_trie, Pod};
 use libc::memcpy;
-use serde::{ser::SerializeStruct, Serialize};
+use serde::{
+    de::{self, MapAccess, Visitor},
+    ser::SerializeStruct,
+    Deserialize, Serialize,
+};
 
 include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
 
@@ -101,6 +105,51 @@ impl Serialize for inode_key_t {
     }
 }
 
+impl<'de> Deserialize<'de> for inode_key_t {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        struct InodeKeyVisitor;
+
+        impl<'de> Visitor<'de> for InodeKeyVisitor {
+            type Value = inode_key_t;
+
+            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+                f.write_str("a struct with inode and dev fields")
+            }
+
+            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
+            where
+                A: MapAccess<'de>,
+            {
+                let mut inode = None;
+                let mut dev = None;
+                while let Some(key) = map.next_key::<String>()? {
+                    match key.as_str() {
+                        "inode" => inode = Some(map.next_value()?),
+                        "dev" => dev = Some(map.next_value()?),
+                        _ => {
+                            map.next_value::<de::IgnoredAny>()?;
+                        }
+                    }
+                }
+
+                let Some(inode) = inode else {
+                    return Err(de::Error::missing_field("inode"));
+                };
+                let Some(dev) = dev else {
+                    return Err(de::Error::missing_field("dev"));
+                };
+
+                Ok(inode_key_t { inode, dev })
+            }
+        }
+
+        deserializer.deserialize_struct("inode_key_t", &["inode", "dev"], InodeKeyVisitor)
+    }
+}
+
 unsafe impl Pod for inode_key_t {}
 
 impl Default for monitored_t {
@@ -124,6 +173,25 @@ impl Serialize for monitored_t {
     }
 }
 
+impl<'de> Deserialize<'de> for monitored_t {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        match s.as_str() {
+            "not monitored" => Ok(monitored_t::NOT_MONITORED),
+            "by inode" => Ok(monitored_t::MONITORED_BY_INODE),
+            "by path" => Ok(monitored_t::MONITORED_BY_PATH),
+            "by parent" => Ok(monitored_t::MONITORED_BY_PARENT),
+            _ => Err(de::Error::unknown_variant(
+                &s,
+                &["not monitored", "by inode", "by path", "by parent"],
+            )),
+        }
+    }
+}
+
 impl metrics_by_hook_t {
     fn accumulate(mut self, other: &metrics_by_hook_t) -> metrics_by_hook_t {
         self.total += other.total;

fact/src/bpf/mod.rs

@@ -13,7 +13,7 @@ use log::{error, info, warn};
 use tokio::{
     io::unix::AsyncFd,
     sync::{mpsc, watch},
-    task::JoinHandle,
+    task::JoinSet,
 };
 
 use crate::{config::BpfConfig, event::Event, host_info, metrics::EventCounter};
@@ -225,12 +225,13 @@ impl Bpf {
     // Gather events from the ring buffer and print them out.
     pub fn start(
         mut self,
+        task_set: &mut JoinSet<anyhow::Result<()>>,
         mut running: watch::Receiver<bool>,
         event_counter: EventCounter,
-    ) -> JoinHandle<anyhow::Result<()>> {
+    ) {
         info!("Starting BPF worker...");
 
-        tokio::spawn(async move {
+        task_set.spawn(async move {
             let rb = self.take_ringbuffer()?;
             let mut fd = AsyncFd::new(rb)?;
 
@@ -283,7 +284,7 @@ impl Bpf {
             }
 
             Ok(())
-        })
+        });
     }
 }
 
@@ -322,9 +323,10 @@ mod bpf_tests {
             Bpf::new(reloader.paths(), &reloader.config().bpf).expect("Failed to load BPF code");
         let (run_tx, run_rx) = watch::channel(true);
         // Create a metrics exporter, but don't start it
-        let exporter = Exporter::new(bpf.take_metrics().unwrap());
+        let exporter = Exporter::new(Some(bpf.take_metrics().unwrap()));
+        let mut task_set = JoinSet::new();
 
-        let handle = bpf.start(run_rx, exporter.metrics.bpf_worker.clone());
+        bpf.start(&mut task_set, run_rx, exporter.metrics.bpf_worker.clone());
 
         tokio::time::sleep(Duration::from_millis(500)).await;
 
@@ -412,7 +414,7 @@ mod bpf_tests {
 
         tokio::select! {
             res = wait => res.unwrap(),
-            res = handle => res.unwrap().unwrap(),
+            res = task_set.join_next() => res.unwrap().unwrap().unwrap(),
         }
 
         run_tx.send(false).unwrap();

fact/src/config/mod.rs

@@ -42,6 +42,7 @@ pub struct FactConfig {
     hotreload: Option<bool>,
     scan_interval: Option<Duration>,
     rate_limit: Option<u64>,
+    replay: Option<PathBuf>,
 }
 
 impl FactConfig {
@@ -108,6 +109,10 @@ impl FactConfig {
         if let Some(rate_limit) = from.rate_limit {
             self.rate_limit = Some(rate_limit);
         }
+
+        if let Some(replay) = from.replay.as_deref() {
+            self.replay = Some(replay.to_path_buf());
+        }
     }
 
     pub fn paths(&self) -> &[PathBuf] {
@@ -134,6 +139,10 @@ impl FactConfig {
         self.rate_limit.unwrap_or(0)
     }
 
+    pub fn replay(&self) -> Option<&Path> {
+        self.replay.as_deref()
+    }
+
     #[cfg(test)]
     pub fn set_paths(&mut self, paths: Vec<PathBuf>) {
         self.paths = Some(paths);
@@ -247,6 +256,12 @@ impl TryFrom<Vec<Yaml>> for FactConfig {
                     }
                     config.rate_limit = Some(rate_limit as u64);
                 }
+                "replay" => {
+                    let Some(replay) = v.as_str() else {
+                        bail!("replay field has incorrect type: {v:?}")
+                    };
+                    config.replay = Some(PathBuf::from(replay));
+                }
                 name => bail!("Invalid field '{name}' with value: {v:?}"),
             }
         }
@@ -760,6 +775,13 @@ pub struct FactCli {
     /// Default value is 0 (unlimited)
     #[arg(long, short = 'l', env = "FACT_RATE_LIMIT")]
     rate_limit: Option<u64>,
+
+    /// Replay events from a JSONL file instead of loading eBPF programs.
+    ///
+    /// This mode skips BPF loading and HostScanner, reading pre-recorded
+    /// events for profiling purposes (e.g. valgrind, DHAT).
+    #[arg(long, env = "FACT_REPLAY")]
+    replay: Option<PathBuf>,
 }
 
 impl FactCli {
@@ -794,6 +816,7 @@ impl FactCli {
             hotreload: resolve_bool_arg(self.hotreload, self.no_hotreload),
             scan_interval: self.scan_interval,
             rate_limit: self.rate_limit,
+            replay: self.replay.clone(),
         }
     }
 }

fact/src/config/reloader.rs

@@ -5,7 +5,6 @@ use std::{
 use log::{debug, info, warn};
 use tokio::{
     sync::{Notify, watch},
-    task::JoinHandle,
     time::interval,
 };
 
@@ -34,13 +33,13 @@ impl Reloader {
     ///
     /// If hotreload is disabled on startup the task will not be
     /// spawned.
-    pub fn start(mut self, mut running: watch::Receiver<bool>) -> Option<JoinHandle<()>> {
+    pub fn start(mut self, mut running: watch::Receiver<bool>) {
         if !self.config.hotreload() {
             info!("Configuration hotreload is disabled, changes will require a restart.");
-            return None;
+            return;
         }
 
-        let handle = tokio::spawn(async move {
+        tokio::spawn(async move {
             let mut ticker = interval(Duration::from_secs(10));
             loop {
                 tokio::select! {
@@ -55,7 +54,6 @@ impl Reloader {
                 }
             }
         });
-        Some(handle)
     }
 
     pub fn config(&self) -> &FactConfig {

fact/src/config/tests.rs

@@ -393,6 +393,13 @@ fn parsing() {
                 ..Default::default()
             },
         ),
+        (
+            "replay: /some/path",
+            FactConfig {
+                replay: Some(PathBuf::from("/some/path")),
+                ..Default::default()
+            },
+        ),
         (
             r#"
             paths:
@@ -420,6 +427,7 @@ fn parsing() {
             hotreload: false
             scan_interval: 60
             rate_limit: 50000
+            replay: /some/path.jsonl
             "#,
             FactConfig {
                 paths: Some(vec![PathBuf::from("/etc")]),
@@ -451,6 +459,7 @@ fn parsing() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: Some(50000),
+                replay: Some(PathBuf::from("/some/path.jsonl")),
             },
         ),
     ];
@@ -790,6 +799,10 @@ paths:
             "rate_limit field has incorrect type: Real(\"1000.0\")",
         ),
         ("rate_limit: -1000", "invalid rate_limit: -1000"),
+        (
+            "replay: true",
+            "replay field has incorrect type: Boolean(true)",
+        ),
         ("unknown:", "Invalid field 'unknown' with value: Null"),
     ];
     for (input, expected) in tests {
@@ -1600,6 +1613,25 @@ fn update() {
                 ..Default::default()
             },
         ),
+        (
+            "replay: /some/path.jsonl",
+            FactConfig::default(),
+            FactConfig {
+                replay: Some("/some/path.jsonl".into()),
+                ..Default::default()
+            },
+        ),
+        (
+            "replay: /some/path.jsonl",
+            FactConfig {
+                replay: Some("/initial/path.jsonl".into()),
+                ..Default::default()
+            },
+            FactConfig {
+                replay: Some("/some/path.jsonl".into()),
+                ..Default::default()
+            },
+        ),
         (
             r#"
             paths:
@@ -1658,6 +1690,7 @@ fn update() {
                 hotreload: Some(true),
                 scan_interval: Some(Duration::from_secs(30)),
                 rate_limit: Some(5000),
+                replay: None,
             },
             FactConfig {
                 paths: Some(vec![PathBuf::from("/etc")]),
@@ -1689,6 +1722,7 @@ fn update() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: Some(1000),
+                replay: None,
             },
         ),
     ];
@@ -1728,6 +1762,7 @@ fn defaults() {
     assert_eq!(config.otel.endpoint(), None);
     assert_eq!(config.scan_interval(), Duration::from_secs(30));
     assert_eq!(config.rate_limit(), 0);
+    assert!(config.replay().is_none());
 }
 
 static ENV_MUTEX: Mutex<()> = Mutex::new(());