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);
@@ -760,6 +769,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 +810,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

@@ -450,6 +450,7 @@ fn parsing() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: None,
+                replay: None,
             },
         ),
     ];
@@ -1628,6 +1629,7 @@ fn update() {
                 hotreload: Some(true),
                 scan_interval: Some(Duration::from_secs(30)),
                 rate_limit: None,
+                replay: None,
             },
             FactConfig {
                 paths: Some(vec![PathBuf::from("/etc")]),
@@ -1659,6 +1661,7 @@ fn update() {
                 hotreload: Some(false),
                 scan_interval: Some(Duration::from_secs(60)),
                 rate_limit: None,
+                replay: None,
             },
         ),
     ];

fact/src/event/mod.rs

@@ -11,7 +11,7 @@ use std::{
 use globset::GlobSet;
 #[cfg(feature = "otel")]
 use opentelemetry::logs::AnyValue;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
 use fact_ebpf::{
     PATH_MAX, XATTR_NAME_MAX_LEN, event_t, file_activity_type_t, inode_key_t, monitored_t,
@@ -74,9 +74,10 @@ pub(crate) enum EventTestData {
     Rename(PathBuf),
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Event {
     timestamp: u64,
+    #[serde(skip_deserializing)]
     hostname: &'static str,
     process: Process,
     file: FileData,
@@ -391,7 +392,7 @@ impl PartialEq for Event {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum FileData {
     Open(BaseFileData),
     Creation(BaseFileData),
@@ -607,7 +608,7 @@ impl PartialEq for FileData {
     }
 }
 
-#[derive(Debug, Clone, Serialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct BaseFileData {
     pub filename: PathBuf,
     host_file: PathBuf,
@@ -665,7 +666,7 @@ impl From<BaseFileData> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ChmodFileData {
     inner: BaseFileData,
     new_mode: u16,
@@ -711,7 +712,7 @@ impl PartialEq for ChmodFileData {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ChownFileData {
     inner: BaseFileData,
     new_uid: u32,
@@ -767,7 +768,7 @@ impl From<ChownFileData> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RenameFileData {
     new: BaseFileData,
     old: BaseFileData,
@@ -795,7 +796,7 @@ impl From<RenameFileData> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
 pub enum AclTag {
     UserObj,
     User,
@@ -842,7 +843,7 @@ impl AclTag {
     }
 }
 
-#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AclEntry {
     tag: AclTag,
     perm: u16,
@@ -876,7 +877,7 @@ impl From<AclEntry> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
 pub enum AclType {
     Access,
     Default,
@@ -892,7 +893,7 @@ impl From<AclType> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AclSetFileData {
     inner: BaseFileData,
     acl_type: AclType,
@@ -968,7 +969,7 @@ impl PartialEq for RenameFileData {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct XattrFileData {
     inner: BaseFileData,
     xattr_name: String,

fact/src/event/process.rs

@@ -5,14 +5,14 @@ use std::{ffi::CStr, path::PathBuf};
 use fact_ebpf::{lineage_t, process_t};
 #[cfg(feature = "otel")]
 use opentelemetry::logs::AnyValue;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
 use crate::host_info;
 
 use super::{sanitize_d_path, slice_to_string};
 
-#[derive(Debug, Clone, Default, PartialEq, Serialize)]
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
 pub struct Lineage {
     uid: u32,
     exe_path: PathBuf,
@@ -55,13 +55,14 @@ impl From<Lineage> for opentelemetry::logs::AnyValue {
     }
 }
 
-#[derive(Debug, Clone, Default, Serialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct Process {
     comm: String,
     args: Vec<String>,
     exe_path: PathBuf,
     container_id: Option<String>,
     uid: u32,
+    #[serde(skip_deserializing)]
     username: &'static str,
     gid: u32,
     login_uid: u32,