wm_dispatch/
flight.rs

1//! Q35b flight recorder — opt-in dispatch payload capture (JSONL sidecar).
2//!
3//! Design: `planning/Q35B_FLIGHT_RECORDER_DESIGN.md`. The write-audit
4//! journal stays digest-only (always-on, security-clean); the flight
5//! sidecar holds the replay payloads, OFF by default, captured at the
6//! same point the `args_digest` is computed so sidecar args always hash
7//! to the journal digest (the replay identity gate).
8
9use std::fs::{self, File, OpenOptions};
10use std::io::{self, BufRead, BufReader, Write};
11use std::path::{Path, PathBuf};
12use std::sync::atomic::{AtomicU64, Ordering};
13
14use serde::{Deserialize, Serialize};
15use serde_json::Value;
16
17/// One captured dispatch — the replay payload unit.
18#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
19pub struct FlightEntry {
20    /// Monotonic capture sequence (capture order).
21    pub seq: u64,
22    /// Unix timestamp (seconds) at capture.
23    pub ts: u64,
24    /// Tool that was dispatched.
25    pub tool: String,
26    /// The dispatch arguments as captured (post-gate, digest-aligned).
27    pub args: Value,
28}
29
30/// Append-only JSONL sidecar. Single-writer per file (matches the
31/// journal's best-effort window note for concurrent dispatches).
32pub struct FlightRecorder {
33    path: PathBuf,
34    next_seq: AtomicU64,
35}
36
37impl FlightRecorder {
38    /// Create (or append to) a flight log at `path`.
39    pub fn new(path: impl Into<PathBuf>) -> io::Result<Self> {
40        let path = path.into();
41        if let Some(parent) = path.parent() {
42            fs::create_dir_all(parent)?;
43        }
44        let next = Self::scan_last_seq(&path)?.map_or(0, |s| s + 1);
45        Ok(Self {
46            path,
47            next_seq: AtomicU64::new(next),
48        })
49    }
50
51    /// Capture one dispatch. Returns the assigned seq.
52    pub fn record(&self, tool: &str, args: &Value) -> io::Result<u64> {
53        let seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
54        let entry = FlightEntry {
55            seq,
56            ts: wm_core::time::now_unix_secs(),
57            tool: tool.to_string(),
58            args: args.clone(),
59        };
60        let mut file = OpenOptions::new()
61            .create(true)
62            .append(true)
63            .open(&self.path)?;
64        writeln!(
65            file,
66            "{}",
67            serde_json::to_string(&entry).map_err(|e| {
68                io::Error::new(io::ErrorKind::InvalidData, format!("serialize: {e}"))
69            })?
70        )?;
71        file.flush()?;
72        Ok(seq)
73    }
74
75    /// Last seq in an existing log (`None` = new/empty file).
76    fn scan_last_seq(path: &Path) -> io::Result<Option<u64>> {
77        if !path.exists() {
78            return Ok(None);
79        }
80        let last = BufReader::new(File::open(path)?)
81            .lines()
82            .map_while(Result::ok)
83            .filter_map(|l| serde_json::from_str::<FlightEntry>(&l).ok())
84            .map(|e| e.seq)
85            .max();
86        Ok(last)
87    }
88
89    /// Read all entries from a flight log (skips malformed tail lines
90    /// loudly via count — a torn last line after a crash must not
91    /// silently truncate the replay set).
92    pub fn read_entries(path: impl AsRef<Path>) -> io::Result<(Vec<FlightEntry>, usize)> {
93        let file = File::open(path.as_ref())?;
94        let mut entries = Vec::new();
95        let mut malformed = 0usize;
96        for line in BufReader::new(file).lines() {
97            let line = line?;
98            match serde_json::from_str::<FlightEntry>(&line) {
99                Ok(e) => entries.push(e),
100                Err(_) => malformed += 1,
101            }
102        }
103        Ok((entries, malformed))
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110    use serde_json::json;
111
112    fn tmp_path(tag: &str) -> PathBuf {
113        let dir = tempfile::tempdir().unwrap();
114        // Leak the tempdir for the test lifetime via Box::leak pattern:
115        // simpler — write into std::env::temp_dir with a unique suffix.
116        drop(dir);
117        std::env::temp_dir().join(format!("wm-flight-test-{tag}-{}", std::process::id()))
118    }
119
120    #[test]
121    fn record_then_read_roundtrip_and_append_seq() {
122        let path = tmp_path("roundtrip");
123        let rec = FlightRecorder::new(&path).unwrap();
124        let s0 = rec
125            .record("test.writer", &json!({"key": "a", "n": 1}))
126            .unwrap();
127        let s1 = rec.record("test.reader", &json!({"id": "x"})).unwrap();
128        assert_eq!((s0, s1), (0, 1));
129
130        // Reopen: append must continue from the last seq, not restart.
131        let rec2 = FlightRecorder::new(&path).unwrap();
132        let s2 = rec2
133            .record("test.writer", &json!({"key": "b", "n": 2}))
134            .unwrap();
135        assert_eq!(s2, 2, "reopen continues the sequence");
136
137        let (entries, malformed) = FlightRecorder::read_entries(&path).unwrap();
138        assert_eq!(malformed, 0);
139        assert_eq!(entries.len(), 3);
140        assert_eq!(entries[0].tool, "test.writer");
141        assert_eq!(entries[1].args["id"], "x");
142        let _ = fs::remove_file(&path);
143    }
144
145    #[test]
146    fn malformed_tail_is_counted_not_silent() {
147        let path = tmp_path("torn");
148        let rec = FlightRecorder::new(&path).unwrap();
149        rec.record("test.writer", &json!({"key": "a"})).unwrap();
150        // Simulate a torn write (crash mid-line).
151        let mut f = OpenOptions::new().append(true).open(&path).unwrap();
152        f.write_all(b"{\"seq\":1,\"tool\":\"te").unwrap();
153        drop(f);
154        let (entries, malformed) = FlightRecorder::read_entries(&path).unwrap();
155        assert_eq!(entries.len(), 1);
156        assert_eq!(malformed, 1, "torn line must be loud");
157        let _ = fs::remove_file(&path);
158    }
159}
wm_dispatch/flight.rs

wm_dispatch/
flight.rs
