blaze-host: opt-in raw frame capture + auditable sanitizer

Evidence infrastructure, not protocol functionality. Built before gates 7-10
because those sessions cannot be reproduced -- a later run is a different
session, and the migration-validation runs happen once. Gates 5-6 already went
past without their bytes being recorded.

TWO LAYERS

  live FIFA traffic
    ├── raw capture      exact RX/TX bytes, mode 0600, gitignored
    └── blaze-sanitize   → repository-safe, replayable fixtures

CAPTURE. Off unless OPENFUT_BLAZE_CAPTURE names a file. Deterministic
big-endian container: 20-byte file header, then per-frame records carrying
connection id, a global monotonic sequence, timestamp, direction and the EXACT
frame bytes. RX is recorded as received; TX only AFTER a successful write, so a
record means the bytes were sent rather than intended.

Component/command/msgNum/msgType/payload length are deliberately NOT stored
beside the frame: they are already in its 16-byte header, and a redundant copy
can disagree with the bytes, leaving a reader unable to tell which is true.
Record::header() derives them, so every field the requirements name is
available without duplicating it.

SANITIZER. Redacts only the named tags in SENSITIVE_TAGS (KEY, AUTH, SESS,
MAIL, PML) and reports every substitution with path, kind and length.
Replacement is LENGTH-PRESERVING, so the TDF varint, payload length and Fire2
header are unchanged and the sanitized frame is exactly the size of the
captured one -- asserted per frame, failing rather than emitting a subtly
different conversation. Frames with nothing sensitive keep their exact wire
bytes. Payloads that will not decode are passed through and REPORTED, so a
reader knows they were never inspected rather than assuming they were checked.

TESTS. 39 in this crate. All nine required cases: capture disabled produces no
artefact; RX and TX captured exactly; ordering preserved; fragmented input
(one byte at a time) reconstructs the same frames as a single write; coalesced
input is captured as separate frames, not per-read; capture does not alter wire
output; sanitization removes a real session key from a real captured login;
malformed/truncated/wrong-version captures fail clearly; every listed sensitive
tag is provably reachable.

MUTATION TESTED. Dropping TX capture, truncating captured frames to their
header, and removing KEY from the sensitive list were each verified to turn the
suite red. One mutation was NOT caught: moving the TX capture above the write.
It is indistinguishable while writes succeed and only diverges when one fails.
That invariant is held by code placement and a comment saying so, not by a
test, and the code says as much rather than implying coverage it does not have.

Python oracle unchanged.

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

.gitignore

@@ -31,3 +31,8 @@ Thumbs.db
 # Frozen baseline archives / inspects / manifests
 /docker-backups/
 gate-evidence/
+
+# Raw Fire2 frame captures — forensic evidence, may contain session material.
+# Sanitize with `blaze-sanitize` before anything leaves this machine.
+*.ofcap
+captures/

openfut-blaze-host/Cargo.toml

@@ -28,3 +28,7 @@ path = "src/main.rs"
 [[bin]]
 name = "blaze-probe"
 path = "src/bin/blaze-probe.rs"
+
+[[bin]]
+name = "blaze-sanitize"
+path = "src/bin/blaze-sanitize.rs"

openfut-blaze-host/src/bin/blaze-sanitize.rs

@@ -0,0 +1,170 @@
+//! Read a raw capture and emit a repository-safe, replayable fixture.
+//!
+//! ```text
+//!   blaze-sanitize <capture.ofcap> [-o out.jsonl] [--report report.txt]
+//! ```
+//!
+//! Prints an audit of exactly what was replaced. Nothing is redacted silently:
+//! every substitution appears in the report with its path, kind and length, and
+//! untouched frames keep their exact wire bytes.
+//!
+//! The output is a JSONL conversation that replays against either backend —
+//! see `blaze-probe --capture`.
+
+use std::io::Write;
+
+use openfut_blaze_host::capture::{self, Direction};
+use openfut_blaze_host::sanitize::{self, SENSITIVE_TAGS};
+use openfut_protocol_blaze::fire2::Header;
+
+fn usage() -> ! {
+    eprintln!("usage: blaze-sanitize <capture.ofcap> [-o out.jsonl] [--report report.txt]");
+    eprintln!();
+    eprintln!("Reads a raw frame capture and writes a sanitized, replayable JSONL");
+    eprintln!("conversation plus an audit of every redaction made.");
+    std::process::exit(2);
+}
+
+fn arg_after(args: &[String], flag: &str) -> Option<String> {
+    args.windows(2).find(|w| w[0] == flag).map(|w| w[1].clone())
+}
+
+fn hex(b: &[u8]) -> String {
+    b.iter().map(|x| format!("{x:02x}")).collect()
+}
+
+fn json_escape(s: &str) -> String {
+    s.chars()
+        .flat_map(|c| match c {
+            '"' => "\\\"".chars().collect::<Vec<_>>(),
+            '\\' => "\\\\".chars().collect(),
+            '\n' => "\\n".chars().collect(),
+            c => vec![c],
+        })
+        .collect()
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    if args.is_empty() || args[0].starts_with('-') {
+        usage();
+    }
+    let input = args[0].clone();
+    let out_path = arg_after(&args, "-o").unwrap_or_else(|| format!("{input}.sanitized.jsonl"));
+    let report_path = arg_after(&args, "--report");
+
+    let records = match capture::read_file(&input) {
+        Ok(r) => r,
+        Err(e) => {
+            eprintln!("blaze-sanitize: cannot read {input}: {e}");
+            std::process::exit(1);
+        }
+    };
+    eprintln!("read {} frame(s) from {input}", records.len());
+
+    let (out, report) = match sanitize::sanitize(&records) {
+        Ok(v) => v,
+        Err(e) => {
+            eprintln!("blaze-sanitize: {e}");
+            std::process::exit(1);
+        }
+    };
+
+    // ------------------------------------------------------------- fixture
+    let mut f = match std::fs::File::create(&out_path) {
+        Ok(f) => f,
+        Err(e) => {
+            eprintln!("blaze-sanitize: cannot write {out_path}: {e}");
+            std::process::exit(1);
+        }
+    };
+
+    let _ = writeln!(
+        f,
+        "{{\"kind\":\"capture\",\"source\":\"{}\",\"frames\":{},\"redacted_frames\":{},\
+         \"sensitive_tags\":[{}],\"note\":\"sanitized; redacted values are \
+         length-preserving fillers, all other bytes are exact\"}}",
+        json_escape(&input),
+        report.frames_out,
+        report.frames_redacted,
+        SENSITIVE_TAGS
+            .iter()
+            .map(|t| format!("\"{t}\""))
+            .collect::<Vec<_>>()
+            .join(",")
+    );
+
+    for (rec, bytes) in &out {
+        let h = Header::parse(bytes).ok();
+        let (comp, cmd, mtype, mnum, uidx, plen) = match h {
+            Some(h) => (
+                h.component,
+                h.command,
+                h.msg_type.as_bits(),
+                h.msg_num,
+                h.user_index,
+                h.payload_len,
+            ),
+            None => (0, 0, 0, 0, 0, 0),
+        };
+        let _ = writeln!(
+            f,
+            "{{\"kind\":\"frame\",\"conn\":{},\"seq\":{},\"dir\":\"{}\",\
+             \"component\":{},\"command\":{},\"msg_type\":{},\"msg_num\":{},\
+             \"user_index\":{},\"payload_len\":{},\"frame_hex\":\"{}\"}}",
+            rec.conn_id,
+            rec.seq,
+            rec.dir.label(),
+            comp,
+            cmd,
+            mtype,
+            mnum,
+            uidx,
+            plen,
+            hex(bytes)
+        );
+    }
+    eprintln!("wrote {out_path}");
+
+    // -------------------------------------------------------------- report
+    let mut lines = Vec::new();
+    lines.push(format!("sanitization report for {input}"));
+    lines.push(format!("  frames in:       {}", report.frames_in));
+    lines.push(format!("  frames out:      {}", report.frames_out));
+    lines.push(format!("  frames redacted: {}", report.frames_redacted));
+    lines.push(format!("  redactions:      {}", report.redactions.len()));
+    lines.push(format!("  sensitive tags:  {}", SENSITIVE_TAGS.join(", ")));
+    if !report.undecodable.is_empty() {
+        lines.push(format!(
+            "  NOT decodable as TDF (passed through unchanged, NOT inspected): seq {:?}",
+            report.undecodable
+        ));
+    }
+    lines.push(String::new());
+    if report.redactions.is_empty() {
+        lines.push("  no sensitive values found".into());
+    } else {
+        lines.push("  seq   conn  path                 kind    bytes".into());
+        for r in &report.redactions {
+            lines.push(format!(
+                "  {:<5} {:<5} {:<20} {:<7} {} -> {}",
+                r.seq, r.conn_id, r.path, r.kind, r.original_len, r.replacement_len
+            ));
+        }
+    }
+
+    // Direction/route summary, so a reviewer can see the conversation shape.
+    lines.push(String::new());
+    lines.push("  conversation shape:".into());
+    let rx = out.iter().filter(|(r, _)| r.dir == Direction::Rx).count();
+    let tx = out.len() - rx;
+    lines.push(format!("    {rx} RX, {tx} TX"));
+
+    let text = lines.join("\n") + "\n";
+    print!("{text}");
+    if let Some(p) = report_path {
+        if std::fs::write(&p, &text).is_ok() {
+            eprintln!("wrote {p}");
+        }
+    }
+}

openfut-blaze-host/src/capture.rs

@@ -0,0 +1,430 @@
+//! Opt-in raw Fire2 frame capture.
+//!
+//! **Evidence infrastructure, not protocol functionality.** Disabled unless
+//! `OPENFUT_BLAZE_CAPTURE` names a file. Nothing in the dispatch path consults
+//! it, and it never changes a byte, an ordering, or a session.
+//!
+//! # Why this exists
+//!
+//! The wire captures every RE finding cites are gitignored and gone from disk,
+//! and a complete live FIFA Blaze session ran past during gates 5–6 without its
+//! bytes being recorded. Those sessions cannot be reproduced: a later run is a
+//! different session, and the migration-validation runs in particular happen
+//! once.
+//!
+//! # Two layers
+//!
+//! ```text
+//!   live FIFA traffic
+//!         ├── raw capture     exact RX/TX bytes, gitignored, mode 0600
+//!         └── blaze-sanitize  → repository-safe, auditable fixtures
+//! ```
+//!
+//! Raw captures are forensic evidence and may carry session material. They are
+//! never committed and never printed to the ordinary log.
+//!
+//! # Format
+//!
+//! Deterministic and self-describing, big-endian throughout to match the
+//! protocol.
+//!
+//! ```text
+//!   file header (20 bytes)
+//!     magic    8   "OFUTCAP1"
+//!     version  u16
+//!     flags    u16
+//!     created  u64  unix seconds
+//!
+//!   record (repeated)
+//!     len       u32   bytes after this field
+//!     conn_id   u64
+//!     seq       u64   global, monotonic — total order across connections
+//!     ts_ms     u64
+//!     dir       u8    0 = RX (from client), 1 = TX (to client)
+//!     pad       u8×3
+//!     frame_len u32
+//!     frame     [frame_len]   EXACT bytes, one whole Fire2 frame
+//! ```
+//!
+//! Component, command, message number, message type and payload length are
+//! deliberately **not** stored alongside the frame: they are already in its
+//! 16-byte header, and a redundant copy can disagree with the bytes, leaving a
+//! reader unable to tell which is true. The frame is authoritative and readers
+//! derive those fields from it — [`Record::header`] does exactly that, so every
+//! field named in the capture requirements is available without duplicating it.
+
+use std::fs::File;
+use std::io::{self, Read, Write};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Mutex;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use openfut_protocol_blaze::fire2::Header;
+
+pub const MAGIC: &[u8; 8] = b"OFUTCAP1";
+pub const VERSION: u16 = 1;
+pub const FILE_HEADER_LEN: usize = 20;
+/// Bytes of a record after its own length field, excluding the frame.
+const RECORD_FIXED: usize = 8 + 8 + 8 + 1 + 3 + 4;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Direction {
+    /// Received from the client.
+    Rx,
+    /// Sent to the client.
+    Tx,
+}
+
+impl Direction {
+    pub fn as_byte(self) -> u8 {
+        match self {
+            Direction::Rx => 0,
+            Direction::Tx => 1,
+        }
+    }
+    pub fn from_byte(b: u8) -> Option<Direction> {
+        match b {
+            0 => Some(Direction::Rx),
+            1 => Some(Direction::Tx),
+            _ => None,
+        }
+    }
+    pub fn label(self) -> &'static str {
+        match self {
+            Direction::Rx => "RX",
+            Direction::Tx => "TX",
+        }
+    }
+}
+
+/// A capture sink. Cheap and inert when disabled.
+pub struct Capture {
+    sink: Option<Mutex<File>>,
+    seq: AtomicU64,
+}
+
+fn now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+impl Capture {
+    /// Disabled sink.
+    pub fn disabled() -> Capture {
+        Capture {
+            sink: None,
+            seq: AtomicU64::new(0),
+        }
+    }
+
+    /// Create a capture file, replacing any existing one.
+    ///
+    /// Created mode 0600: these bytes are forensic evidence and may contain
+    /// session material.
+    pub fn create(path: &str) -> io::Result<Capture> {
+        let mut opts = File::options();
+        opts.write(true).create(true).truncate(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            opts.mode(0o600);
+        }
+        let mut file = opts.open(path)?;
+
+        let mut hdr = Vec::with_capacity(FILE_HEADER_LEN);
+        hdr.extend_from_slice(MAGIC);
+        hdr.extend_from_slice(&VERSION.to_be_bytes());
+        hdr.extend_from_slice(&0u16.to_be_bytes()); // flags
+        hdr.extend_from_slice(
+            &SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .map(|d| d.as_secs())
+                .unwrap_or(0)
+                .to_be_bytes(),
+        );
+        file.write_all(&hdr)?;
+        file.flush()?;
+
+        Ok(Capture {
+            sink: Some(Mutex::new(file)),
+            seq: AtomicU64::new(0),
+        })
+    }
+
+    pub fn enabled(&self) -> bool {
+        self.sink.is_some()
+    }
+
+    /// Record one whole frame.
+    ///
+    /// Callers pass the exact bytes that crossed the socket, and call this
+    /// AFTER the I/O has happened — so a TX record means "these bytes were
+    /// written", not "these bytes were intended". Errors are swallowed
+    /// deliberately: a capture problem must never take down a live session that
+    /// is otherwise healthy.
+    pub fn record(&self, conn_id: u64, dir: Direction, frame: &[u8]) {
+        let Some(sink) = &self.sink else { return };
+        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
+
+        let mut buf = Vec::with_capacity(4 + RECORD_FIXED + frame.len());
+        buf.extend_from_slice(&((RECORD_FIXED + frame.len()) as u32).to_be_bytes());
+        buf.extend_from_slice(&conn_id.to_be_bytes());
+        buf.extend_from_slice(&seq.to_be_bytes());
+        buf.extend_from_slice(&now_ms().to_be_bytes());
+        buf.push(dir.as_byte());
+        buf.extend_from_slice(&[0u8; 3]);
+        buf.extend_from_slice(&(frame.len() as u32).to_be_bytes());
+        buf.extend_from_slice(frame);
+
+        if let Ok(mut f) = sink.lock() {
+            let _ = f.write_all(&buf);
+            // Flushed per record: a crash mid-session must not cost the
+            // evidence collected so far. Volume is a few dozen frames per FIFA
+            // session, so this is not a hot path.
+            let _ = f.flush();
+        }
+    }
+}
+
+/// One captured frame.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Record {
+    pub conn_id: u64,
+    pub seq: u64,
+    pub ts_ms: u64,
+    pub dir: Direction,
+    /// Exact bytes as they crossed the socket.
+    pub frame: Vec<u8>,
+}
+
+impl Record {
+    /// Parse the Fire2 header from the captured bytes.
+    ///
+    /// Component, command, msgNum, msgType, userIndex and payload length all
+    /// come from here rather than from stored copies.
+    pub fn header(&self) -> Option<Header> {
+        Header::parse(&self.frame).ok()
+    }
+}
+
+#[derive(Debug)]
+pub enum ReadError {
+    NotACapture,
+    UnsupportedVersion(u16),
+    Truncated { at: usize, want: usize, have: usize },
+    Io(String),
+}
+
+impl std::fmt::Display for ReadError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ReadError::NotACapture => write!(f, "not an OpenFUT capture (bad magic)"),
+            ReadError::UnsupportedVersion(v) => write!(f, "unsupported capture version {v}"),
+            ReadError::Truncated { at, want, have } => {
+                write!(
+                    f,
+                    "truncated capture at offset {at}: want {want} bytes, have {have}"
+                )
+            }
+            ReadError::Io(e) => write!(f, "io error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for ReadError {}
+
+/// Read a whole capture. Fails clearly rather than returning partial data.
+pub fn read_file(path: &str) -> Result<Vec<Record>, ReadError> {
+    let mut buf = Vec::new();
+    File::open(path)
+        .and_then(|mut f| f.read_to_end(&mut buf))
+        .map_err(|e| ReadError::Io(e.to_string()))?;
+    parse(&buf)
+}
+
+pub fn parse(buf: &[u8]) -> Result<Vec<Record>, ReadError> {
+    if buf.len() < FILE_HEADER_LEN || &buf[..8] != MAGIC {
+        return Err(ReadError::NotACapture);
+    }
+    let version = u16::from_be_bytes([buf[8], buf[9]]);
+    if version != VERSION {
+        return Err(ReadError::UnsupportedVersion(version));
+    }
+
+    let mut out = Vec::new();
+    let mut i = FILE_HEADER_LEN;
+    while i < buf.len() {
+        if i + 4 > buf.len() {
+            return Err(ReadError::Truncated {
+                at: i,
+                want: 4,
+                have: buf.len() - i,
+            });
+        }
+        let len = u32::from_be_bytes([buf[i], buf[i + 1], buf[i + 2], buf[i + 3]]) as usize;
+        i += 4;
+        if len < RECORD_FIXED || i + len > buf.len() {
+            return Err(ReadError::Truncated {
+                at: i,
+                want: len,
+                have: buf.len().saturating_sub(i),
+            });
+        }
+        let rec = &buf[i..i + len];
+        let conn_id = u64::from_be_bytes(rec[0..8].try_into().unwrap());
+        let seq = u64::from_be_bytes(rec[8..16].try_into().unwrap());
+        let ts_ms = u64::from_be_bytes(rec[16..24].try_into().unwrap());
+        let dir = Direction::from_byte(rec[24]).ok_or(ReadError::Truncated {
+            at: i + 24,
+            want: 1,
+            have: 1,
+        })?;
+        let frame_len = u32::from_be_bytes(rec[28..32].try_into().unwrap()) as usize;
+        if 32 + frame_len != len {
+            return Err(ReadError::Truncated {
+                at: i + 32,
+                want: frame_len,
+                have: len.saturating_sub(32),
+            });
+        }
+        out.push(Record {
+            conn_id,
+            seq,
+            ts_ms,
+            dir,
+            frame: rec[32..32 + frame_len].to_vec(),
+        });
+        i += len;
+    }
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use openfut_protocol_blaze::fire2::{Frame, MsgType};
+
+    fn tmp(name: &str) -> String {
+        let dir = std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into());
+        format!("{dir}/ofcap-test-{}-{name}.ofcap", std::process::id())
+    }
+
+    #[test]
+    fn disabled_capture_writes_nothing() {
+        let c = Capture::disabled();
+        assert!(!c.enabled());
+        c.record(1, Direction::Rx, &[1, 2, 3]); // must not panic or create anything
+    }
+
+    #[test]
+    fn round_trips_frames_exactly_and_in_order() {
+        let path = tmp("roundtrip");
+        let c = Capture::create(&path).unwrap();
+
+        let a = Frame::new(0x0009, 0x0007, 1, MsgType::Message, vec![1, 2, 3]).encode();
+        let b = Frame::new(0x0009, 0x0007, 1, MsgType::Reply, vec![4, 5]).encode();
+        c.record(7, Direction::Rx, &a);
+        c.record(7, Direction::Tx, &b);
+        drop(c);
+
+        let recs = read_file(&path).unwrap();
+        assert_eq!(recs.len(), 2);
+        assert_eq!(recs[0].frame, a, "RX bytes must be exact");
+        assert_eq!(recs[1].frame, b, "TX bytes must be exact");
+        assert_eq!(recs[0].dir, Direction::Rx);
+        assert_eq!(recs[1].dir, Direction::Tx);
+        assert_eq!(recs[0].conn_id, 7);
+        assert!(recs[0].seq < recs[1].seq, "sequence preserves order");
+
+        // Metadata is derivable rather than stored.
+        let h = recs[0].header().unwrap();
+        assert_eq!(h.component, 0x0009);
+        assert_eq!(h.command, 0x0007);
+        assert_eq!(h.msg_type, MsgType::Message);
+        assert_eq!(h.payload_len, 3);
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn interleaved_connections_keep_a_total_order() {
+        let path = tmp("interleaved");
+        let c = Capture::create(&path).unwrap();
+        for i in 0..6u64 {
+            let f = Frame::new(9, 2, i as u32, MsgType::Message, vec![i as u8]).encode();
+            c.record(i % 2, Direction::Rx, &f);
+        }
+        drop(c);
+
+        let recs = read_file(&path).unwrap();
+        assert_eq!(recs.len(), 6);
+        let seqs: Vec<u64> = recs.iter().map(|r| r.seq).collect();
+        assert_eq!(seqs, (0..6).collect::<Vec<_>>());
+        // And per-connection order is recoverable from the same total order.
+        let conn0: Vec<u8> = recs
+            .iter()
+            .filter(|r| r.conn_id == 0)
+            .map(|r| r.frame[16])
+            .collect();
+        assert_eq!(conn0, vec![0, 2, 4]);
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn capture_files_are_owner_only() {
+        use std::os::unix::fs::PermissionsExt;
+        let path = tmp("perms");
+        let _c = Capture::create(&path).unwrap();
+        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
+        assert_eq!(mode, 0o600, "captures may contain session material");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    #[test]
+    fn rejects_a_file_that_is_not_a_capture() {
+        assert!(matches!(
+            parse(b"hello world padding.."),
+            Err(ReadError::NotACapture)
+        ));
+        assert!(matches!(parse(b""), Err(ReadError::NotACapture)));
+    }
+
+    #[test]
+    fn rejects_an_unsupported_version() {
+        let mut buf = MAGIC.to_vec();
+        buf.extend_from_slice(&99u16.to_be_bytes());
+        buf.extend_from_slice(&0u16.to_be_bytes());
+        buf.extend_from_slice(&0u64.to_be_bytes());
+        assert!(matches!(
+            parse(&buf),
+            Err(ReadError::UnsupportedVersion(99))
+        ));
+    }
+
+    #[test]
+    fn truncation_fails_clearly_rather_than_returning_partial_data() {
+        let path = tmp("truncated");
+        let c = Capture::create(&path).unwrap();
+        c.record(
+            1,
+            Direction::Rx,
+            &Frame::new(9, 2, 1, MsgType::Message, vec![7; 40]).encode(),
+        );
+        drop(c);
+
+        let full = std::fs::read(&path).unwrap();
+        for cut in [FILE_HEADER_LEN + 2, FILE_HEADER_LEN + 10, full.len() - 5] {
+            match parse(&full[..cut]) {
+                Err(ReadError::Truncated { .. }) => {}
+                other => panic!("cut at {cut} should be Truncated, got {other:?}"),
+            }
+        }
+        // The whole file still parses.
+        assert_eq!(parse(&full).unwrap().len(), 1);
+        let _ = std::fs::remove_file(&path);
+    }
+}

openfut-blaze-host/src/config.rs

@@ -38,6 +38,8 @@ pub struct HostConfig {
     pub max_payload_bytes: u32,
     /// Optional path for the normalized structural trace.
     pub trace_path: Option<String>,
+    /// Optional path for the raw frame capture. Opt-in; forensic evidence.
+    pub capture_path: Option<String>,
     /// What the adapter answers with.
     pub adapter: AdapterConfig,
 }
@@ -103,6 +105,9 @@ impl HostConfig {
             trace_path: env::var("OPENFUT_BLAZE_TRACE")
                 .ok()
                 .filter(|v| !v.trim().is_empty()),
+            capture_path: env::var("OPENFUT_BLAZE_CAPTURE")
+                .ok()
+                .filter(|v| !v.trim().is_empty()),
             adapter: AdapterConfig {
                 identity: Identity::default(),
                 endpoints,

openfut-blaze-host/src/conn.rs

@@ -17,6 +17,7 @@ use openfut_adapter_fifa17::blaze::{Adapter, Session};
 use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN};
 use openfut_protocol_blaze::heat2::{self, Struct};
 
+use crate::capture::{Capture, Direction};
 use crate::config::HostConfig;
 use crate::trace::{self, Tracer};
 use crate::Hooks;
@@ -81,6 +82,7 @@ pub fn handle(
     adapter: &Adapter,
     tracer: &Tracer,
     hooks: &Hooks,
+    capture: &Capture,
 ) -> CloseReason {
     let peer = stream
         .peer_addr()
@@ -157,6 +159,11 @@ pub fn handle(
         buf.drain(..used);
 
         frame_no += 1;
+        // Capture the exact bytes as received, before anything interprets them.
+        // `&buf[..total]` was already consumed above, so re-encode from the
+        // parsed frame — which is byte-identical by construction and covered by
+        // the protocol crate's round-trip tests.
+        capture.record(conn_id, Direction::Rx, &frame.encode());
         trace::log(&format!(
             "conn-{conn_id:04} RX #{frame_no} {}",
             trace::describe(&frame.header)
@@ -189,6 +196,16 @@ pub fn handle(
                 trace::log(&format!("conn-{conn_id:04} TX #{frame_no}.{k} FAILED: {e}"));
                 break;
             }
+            // AFTER the write: a TX record means these bytes were sent, not
+            // merely intended. Capturing here also keeps the client's latency
+            // untouched — it already has the bytes.
+            //
+            // NOT COVERED BY A TEST: moving this above the write is
+            // indistinguishable while writes succeed, and only diverges when
+            // one fails (it would record a frame that never reached the
+            // client). Mutation-tested and confirmed undetected. The invariant
+            // is held by this placement and this comment; do not move it.
+            capture.record(conn_id, Direction::Tx, &bytes);
             trace::log(&format!(
                 "conn-{conn_id:04} TX #{frame_no}.{k} {} ({}B total)",
                 trace::describe(&resp.header),

openfut-blaze-host/src/lib.rs

@@ -35,8 +35,10 @@
 //! silently collide with the working container. See the crate README for the
 //! A/B procedure and the gate list.
 
+pub mod capture;
 pub mod config;
 pub mod conn;
+pub mod sanitize;
 pub mod trace;
 
 pub use config::HostConfig;
@@ -118,6 +120,7 @@ pub struct Server {
     adapter: Arc<Adapter>,
     tracer: Arc<trace::Tracer>,
     hooks: Arc<Hooks>,
+    capture: Arc<capture::Capture>,
 }
 
 /// Bind without accepting, so a caller can learn the real port first (useful
@@ -126,6 +129,10 @@ pub fn bind(cfg: HostConfig, hooks: Hooks) -> io::Result<Server> {
     let listener = TcpListener::bind(cfg.listen_on())?;
     let local_addr = listener.local_addr()?;
     let tracer = Arc::new(trace::Tracer::new(cfg.trace_path.as_deref())?);
+    let capture = Arc::new(match cfg.capture_path.as_deref() {
+        Some(p) => capture::Capture::create(p)?,
+        None => capture::Capture::disabled(),
+    });
     let adapter = Arc::new(Adapter::new(cfg.adapter.clone()));
     Ok(Server {
         local_addr,
@@ -134,6 +141,7 @@ pub fn bind(cfg: HostConfig, hooks: Hooks) -> io::Result<Server> {
         adapter,
         tracer,
         hooks: Arc::new(hooks),
+        capture,
     })
 }
 
@@ -164,6 +172,13 @@ impl Server {
                 self.cfg.trace_path.as_deref().unwrap_or("")
             ));
         }
+        if self.capture.enabled() {
+            trace::log(&format!(
+                "RAW FRAME CAPTURE -> {} (forensic evidence: never commit, \
+                 sanitize with blaze-sanitize before sharing)",
+                self.cfg.capture_path.as_deref().unwrap_or("")
+            ));
+        }
 
         let counter = AtomicU64::new(0);
         for incoming in self.listener.incoming() {
@@ -174,8 +189,9 @@ impl Server {
                     let adapter = self.adapter.clone();
                     let tracer = self.tracer.clone();
                     let hooks = self.hooks.clone();
+                    let capture = self.capture.clone();
                     std::thread::spawn(move || {
-                        conn::handle(stream, id, &cfg, &adapter, &tracer, &hooks);
+                        conn::handle(stream, id, &cfg, &adapter, &tracer, &hooks, &capture);
                     });
                 }
                 Err(e) => trace::log(&format!("accept failed: {e}")),

openfut-blaze-host/src/sanitize.rs

@@ -0,0 +1,359 @@
+//! Turn a raw capture into a repository-safe fixture, auditably.
+//!
+//! Raw captures are forensic evidence and may carry session material. This
+//! produces something committable from one, and the governing rule is that
+//! sanitization is **explicit and reported**, never a silent byte substitution
+//! followed by a claim of equivalence.
+//!
+//! # What is preserved
+//!
+//! Everything except identified sensitive values: Fire2 headers, component and
+//! command IDs, message types and numbers, user index, frame ordering,
+//! connection identity, payload structure, and every non-sensitive protocol
+//! value.
+//!
+//! # What is replaced, and how
+//!
+//! Only the tags in [`SENSITIVE_TAGS`], and only their values. Replacement is
+//! **length-preserving**: a redacted string occupies exactly as many bytes as
+//! the original, so the TDF length varint, the payload length and the Fire2
+//! header are all unchanged and the sanitized frame is the same size as the
+//! captured one. That is asserted per frame — a size change means the
+//! sanitizer is wrong, and it fails rather than emitting a subtly different
+//! conversation.
+//!
+//! # What this costs
+//!
+//! A sanitized frame is NOT byte-identical to the wire: that is the entire
+//! point, and it is why the raw capture is kept separately. Every difference is
+//! enumerated in the report.
+
+use openfut_protocol_blaze::fire2::Frame;
+use openfut_protocol_blaze::heat2::{self, Struct, Value};
+
+use crate::capture::Record;
+
+/// Tags whose values are session-specific or credential-shaped.
+///
+/// A deliberately small, named list. Anything not here is preserved, so a
+/// reviewer can see exactly what was touched — and a new sensitive field shows
+/// up as un-redacted rather than being silently caught by a broad pattern.
+pub const SENSITIVE_TAGS: &[&str] = &[
+    "KEY",  // Blaze session key (LoginResponse.SESS.KEY, UserAuthenticated.KEY)
+    "AUTH", // auth code / token (LoginRequest.AUTH, GetAuthTokenResponse.AUTH)
+    "SESS", // telemetry session echo, when carried as a string
+    "MAIL", // account email
+    "PML",  // parental email
+];
+
+/// One replacement that was made.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Redaction {
+    pub seq: u64,
+    pub conn_id: u64,
+    /// Dotted path to the member, e.g. `SESS.KEY`.
+    pub path: String,
+    pub kind: &'static str,
+    pub original_len: usize,
+    pub replacement_len: usize,
+}
+
+#[derive(Debug, Default)]
+pub struct Report {
+    pub frames_in: usize,
+    pub frames_out: usize,
+    pub frames_redacted: usize,
+    pub redactions: Vec<Redaction>,
+    /// Frames whose payload did not decode as TDF and were passed through.
+    pub undecodable: Vec<u64>,
+}
+
+/// A length-preserving, obviously-synthetic replacement.
+fn filler(len: usize) -> String {
+    const PATTERN: &[u8] = b"REDACTED_";
+    (0..len)
+        .map(|i| PATTERN[i % PATTERN.len()] as char)
+        .collect()
+}
+
+fn redact_struct(s: &Struct, prefix: &str, rec: &Record, out: &mut Vec<Redaction>) -> Struct {
+    let mut fields = Vec::with_capacity(s.len());
+    for (tag, value) in s.iter() {
+        let label = tag.to_label();
+        let path = if prefix.is_empty() {
+            label.clone()
+        } else {
+            format!("{prefix}.{label}")
+        };
+
+        let new_value = match value {
+            Value::Struct(inner) => Value::Struct(redact_struct(inner, &path, rec, out)),
+            Value::List { elem, items } => Value::List {
+                elem: *elem,
+                items: items
+                    .iter()
+                    .map(|it| match it {
+                        Value::Struct(inner) => {
+                            Value::Struct(redact_struct(inner, &path, rec, out))
+                        }
+                        other => other.clone(),
+                    })
+                    .collect(),
+            },
+            Value::String(v) if SENSITIVE_TAGS.contains(&label.as_str()) && !v.is_empty() => {
+                // Byte length, not char count: the wire length is what must be
+                // preserved, and the filler is ASCII so the two coincide.
+                let n = v.len();
+                out.push(Redaction {
+                    seq: rec.seq,
+                    conn_id: rec.conn_id,
+                    path: path.clone(),
+                    kind: "string",
+                    original_len: n,
+                    replacement_len: n,
+                });
+                Value::String(filler(n))
+            }
+            other => other.clone(),
+        };
+        fields.push((*tag, new_value));
+    }
+    Struct { fields }
+}
+
+/// Sanitize one captured frame. Returns the new frame bytes.
+pub fn sanitize_record(rec: &Record, report: &mut Report) -> Result<Vec<u8>, String> {
+    let (frame, _) = Frame::parse(&rec.frame).map_err(|e| format!("seq {}: {e}", rec.seq))?;
+
+    if frame.payload.is_empty() {
+        return Ok(rec.frame.clone());
+    }
+    let body = match heat2::decode(&frame.payload) {
+        Ok(b) => b,
+        Err(_) => {
+            // Not decodable as TDF. Passed through unchanged and reported, so a
+            // reader knows it was never inspected rather than assuming it was
+            // checked and found clean.
+            report.undecodable.push(rec.seq);
+            return Ok(rec.frame.clone());
+        }
+    };
+
+    let before = report.redactions.len();
+    let redacted = redact_struct(&body, "", rec, &mut report.redactions);
+    if report.redactions.len() == before {
+        // Nothing sensitive: emit the original bytes, not a re-encoding, so
+        // untouched frames stay byte-identical to the wire.
+        return Ok(rec.frame.clone());
+    }
+    report.frames_redacted += 1;
+
+    let mut out = Frame::new(
+        frame.header.component,
+        frame.header.command,
+        frame.header.msg_num,
+        frame.header.msg_type,
+        heat2::encode(&redacted),
+    );
+    out.header.user_index = frame.header.user_index;
+    out.header.options = frame.header.options;
+    let bytes = out.encode();
+
+    // Length preservation is the contract that keeps a sanitized capture a
+    // faithful conversation. If it does not hold, something is wrong with the
+    // replacement and emitting the result would be worse than failing.
+    if bytes.len() != rec.frame.len() {
+        return Err(format!(
+            "seq {}: sanitized frame is {} bytes, original {} — replacement was not \
+             length-preserving",
+            rec.seq,
+            bytes.len(),
+            rec.frame.len()
+        ));
+    }
+    Ok(bytes)
+}
+
+/// A captured record paired with its sanitized frame bytes.
+pub type Sanitized = (Record, Vec<u8>);
+
+/// Sanitize a whole capture.
+pub fn sanitize(records: &[Record]) -> Result<(Vec<Sanitized>, Report), String> {
+    let mut report = Report {
+        frames_in: records.len(),
+        ..Default::default()
+    };
+    let mut out = Vec::with_capacity(records.len());
+    for rec in records {
+        let bytes = sanitize_record(rec, &mut report)?;
+        out.push((rec.clone(), bytes));
+    }
+    report.frames_out = out.len();
+    Ok((out, report))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::capture::Direction;
+    use openfut_protocol_blaze::fire2::MsgType;
+
+    fn rec(seq: u64, body: &Struct) -> Record {
+        Record {
+            conn_id: 1,
+            seq,
+            ts_ms: 0,
+            dir: Direction::Tx,
+            frame: Frame::new(0x0001, 0x000A, 7, MsgType::Reply, heat2::encode(body)).encode(),
+        }
+    }
+
+    #[test]
+    fn redacts_a_session_key_without_changing_frame_size() {
+        let body = Struct::new()
+            .with(
+                "KEY",
+                Value::String("a-real-looking-session-key-0123456789".into()),
+            )
+            .with("UID", Value::Int(33068179));
+        let r = rec(1, &body);
+
+        let mut report = Report::default();
+        let out = sanitize_record(&r, &mut report).unwrap();
+
+        assert_eq!(out.len(), r.frame.len(), "framing must be preserved");
+        assert_eq!(report.redactions.len(), 1);
+        assert_eq!(report.redactions[0].path, "KEY");
+        assert_eq!(
+            report.redactions[0].original_len,
+            report.redactions[0].replacement_len
+        );
+
+        let (f, _) = Frame::parse(&out).unwrap();
+        let decoded = heat2::decode(&f.payload).unwrap();
+        let key = decoded.get("KEY").and_then(Value::as_str).unwrap();
+        assert!(!key.contains("real-looking"), "secret must be gone");
+        assert!(
+            key.starts_with("REDACTED"),
+            "and obviously synthetic: {key}"
+        );
+        // Non-sensitive values survive untouched.
+        assert_eq!(decoded.get("UID").and_then(Value::as_int), Some(33068179));
+    }
+
+    #[test]
+    fn redacts_nested_and_reports_the_path() {
+        let inner = Struct::new()
+            .with("KEY", Value::String("secret-key-value".into()))
+            .with("MAIL", Value::String("someone@example.com".into()));
+        let body = Struct::new()
+            .with("SESS", Value::Struct(inner))
+            .with("ANON", Value::Int(0));
+
+        let mut report = Report::default();
+        let out = sanitize_record(&rec(2, &body), &mut report).unwrap();
+
+        let paths: Vec<&str> = report.redactions.iter().map(|r| r.path.as_str()).collect();
+        assert!(paths.contains(&"SESS.KEY"), "{paths:?}");
+        assert!(paths.contains(&"SESS.MAIL"), "{paths:?}");
+        assert!(!String::from_utf8_lossy(&out).contains("someone@example.com"));
+        assert!(!String::from_utf8_lossy(&out).contains("secret-key-value"));
+    }
+
+    #[test]
+    fn redacts_inside_lists() {
+        let body = Struct::new().with(
+            "NLST",
+            Value::List {
+                elem: openfut_protocol_blaze::heat2::TypeId::Struct,
+                items: vec![Value::Struct(
+                    Struct::new().with("AUTH", Value::String("token-abcdef".into())),
+                )],
+            },
+        );
+        let mut report = Report::default();
+        let out = sanitize_record(&rec(3, &body), &mut report).unwrap();
+        assert_eq!(report.redactions.len(), 1, "list members must be reached");
+        assert!(!String::from_utf8_lossy(&out).contains("token-abcdef"));
+    }
+
+    #[test]
+    fn frames_with_nothing_sensitive_stay_byte_identical() {
+        let body = Struct::new()
+            .with("UID", Value::Int(1))
+            .with("CO", Value::String("US".into()));
+        let r = rec(4, &body);
+        let mut report = Report::default();
+        let out = sanitize_record(&r, &mut report).unwrap();
+        assert_eq!(out, r.frame, "untouched frames keep the exact wire bytes");
+        assert!(report.redactions.is_empty());
+        assert_eq!(report.frames_redacted, 0);
+    }
+
+    #[test]
+    fn empty_payloads_pass_through() {
+        let r = Record {
+            conn_id: 1,
+            seq: 5,
+            ts_ms: 0,
+            dir: Direction::Rx,
+            frame: Frame::new(9, 2, 1, MsgType::Message, vec![]).encode(),
+        };
+        let mut report = Report::default();
+        assert_eq!(sanitize_record(&r, &mut report).unwrap(), r.frame);
+    }
+
+    #[test]
+    fn undecodable_payloads_are_passed_through_and_flagged() {
+        let r = Record {
+            conn_id: 1,
+            seq: 6,
+            ts_ms: 0,
+            dir: Direction::Rx,
+            frame: Frame::new(9, 2, 1, MsgType::Message, vec![0xFF; 8]).encode(),
+        };
+        let mut report = Report::default();
+        let out = sanitize_record(&r, &mut report).unwrap();
+        assert_eq!(out, r.frame);
+        assert_eq!(
+            report.undecodable,
+            vec![6],
+            "must be reported, not silently kept"
+        );
+    }
+
+    #[test]
+    fn ordering_and_identity_survive_a_whole_capture() {
+        let recs: Vec<Record> = (0..5)
+            .map(|i| {
+                rec(
+                    i,
+                    &Struct::new().with("KEY", Value::String(format!("key-{i}-padding"))),
+                )
+            })
+            .collect();
+        let (out, report) = sanitize(&recs).unwrap();
+        assert_eq!(report.frames_in, 5);
+        assert_eq!(report.frames_out, 5);
+        assert_eq!(report.frames_redacted, 5);
+        for (i, (r, _)) in out.iter().enumerate() {
+            assert_eq!(r.seq, i as u64, "sequence order preserved");
+            assert_eq!(r.conn_id, 1);
+        }
+    }
+
+    #[test]
+    fn every_sensitive_tag_is_actually_reachable() {
+        // Guards against a tag being listed but never matched — e.g. a typo.
+        for tag in SENSITIVE_TAGS {
+            let body = Struct::new().with(tag, Value::String("sensitive-value".into()));
+            let mut report = Report::default();
+            let out = sanitize_record(&rec(9, &body), &mut report).unwrap();
+            assert_eq!(report.redactions.len(), 1, "tag {tag} was not redacted");
+            assert!(
+                !String::from_utf8_lossy(&out).contains("sensitive-value"),
+                "tag {tag} leaked"
+            );
+        }
+    }
+}

openfut-blaze-host/tests/live_transport.rs

@@ -21,7 +21,7 @@ use std::sync::Mutex;
 use std::time::Duration;
 
 use openfut_adapter_fifa17::blaze::{AdapterConfig, Endpoints, Identity};
-use openfut_blaze_host::{bind, config::HostConfig, Hooks};
+use openfut_blaze_host::{bind, capture, config::HostConfig, sanitize, Hooks};
 use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN};
 use serde_json::Value as J;
 
@@ -68,6 +68,10 @@ struct Harness {
 /// Nth connection this test opens gets the Nth recorded key. That is why the
 /// connection order below must match the fixture's session order.
 fn start() -> Harness {
+    start_with_capture(None)
+}
+
+fn start_with_capture(capture_path: Option<String>) -> Harness {
     let records = records();
     let cfg_rec = records
         .iter()
@@ -114,6 +118,7 @@ fn start() -> Harness {
         idle_timeout_secs: 10,
         max_payload_bytes: 4 * 1024 * 1024,
         trace_path: None,
+        capture_path,
         adapter: adapter_cfg,
     };
 
@@ -470,3 +475,276 @@ fn an_undecodable_body_still_gets_a_reply() {
     assert_eq!(reply.header.command, 0x0002);
     assert!(!reply.payload.is_empty(), "ping still answers with STIM");
 }
+
+// ───────────────────────────── raw frame capture ─────────────────────────────
+//
+// Capture is evidence infrastructure. These tests exist because the value of a
+// capture is entirely in its fidelity: a capture that quietly drops, reorders
+// or alters frames is worse than none, since it would be trusted.
+
+fn cap_path(name: &str) -> String {
+    let dir = std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into());
+    format!("{dir}/ofcap-it-{}-{name}.ofcap", std::process::id())
+}
+
+/// Drive a conversation over TCP and return (captured records, wire responses).
+fn converse(h: &Harness, take: usize) -> Vec<Vec<u8>> {
+    let sid = session_ids(&h.records).into_iter().next().unwrap();
+    let mut stream = connect(&h.addr);
+    let mut buf = Vec::new();
+    let mut got = Vec::new();
+    for tx in transactions(&h.records, &sid).into_iter().take(take) {
+        stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap();
+        stream.flush().unwrap();
+        for _ in 0..tx["responses"].as_array().unwrap().len() {
+            got.push(read_frame(&mut stream, &mut buf).expect("frame").encode());
+        }
+    }
+    got
+}
+
+#[test]
+fn capture_disabled_produces_no_artefact() {
+    let path = cap_path("disabled");
+    let _ = std::fs::remove_file(&path);
+
+    let h = start(); // capture_path: None
+    converse(&h, 4);
+
+    assert!(
+        !std::path::Path::new(&path).exists(),
+        "no capture file may appear when capture is off"
+    );
+}
+
+#[test]
+fn capture_records_rx_and_tx_frames_exactly_and_in_order() {
+    let path = cap_path("exact");
+    let _ = std::fs::remove_file(&path);
+
+    let h = start_with_capture(Some(path.clone()));
+    let sid = session_ids(&h.records).into_iter().next().unwrap();
+    let txs: Vec<&J> = transactions(&h.records, &sid).into_iter().take(6).collect();
+
+    let mut stream = connect(&h.addr);
+    let mut buf = Vec::new();
+    let mut sent = Vec::new();
+    let mut received = Vec::new();
+    for tx in &txs {
+        let req = unhex(&st(tx, "request_hex"));
+        stream.write_all(&req).unwrap();
+        stream.flush().unwrap();
+        sent.push(req);
+        for _ in 0..tx["responses"].as_array().unwrap().len() {
+            received.push(read_frame(&mut stream, &mut buf).expect("frame").encode());
+        }
+    }
+    drop(stream);
+    std::thread::sleep(std::time::Duration::from_millis(200));
+
+    let recs = capture::read_file(&path).expect("capture parses");
+
+    let rx: Vec<Vec<u8>> = recs
+        .iter()
+        .filter(|r| r.dir == capture::Direction::Rx)
+        .map(|r| r.frame.clone())
+        .collect();
+    let tx_caps: Vec<Vec<u8>> = recs
+        .iter()
+        .filter(|r| r.dir == capture::Direction::Tx)
+        .map(|r| r.frame.clone())
+        .collect();
+
+    assert_eq!(
+        rx, sent,
+        "captured RX must equal the exact bytes sent by the client"
+    );
+    assert_eq!(
+        tx_caps, received,
+        "captured TX must equal the exact bytes the client received"
+    );
+
+    // Total order is monotonic, and RX precedes its own TX.
+    let seqs: Vec<u64> = recs.iter().map(|r| r.seq).collect();
+    let mut sorted = seqs.clone();
+    sorted.sort();
+    assert_eq!(seqs, sorted, "records are written in order");
+    assert_eq!(recs[0].dir, capture::Direction::Rx);
+
+    let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn capture_does_not_alter_the_wire_output() {
+    // The same conversation with and without capture must produce identical
+    // response bytes. Capture must be observation, not participation.
+    let without = converse(&start(), 6);
+
+    let path = cap_path("noalter");
+    let _ = std::fs::remove_file(&path);
+    let with = converse(&start_with_capture(Some(path.clone())), 6);
+
+    assert_eq!(
+        with.iter().map(|f| hex(f)).collect::<Vec<_>>(),
+        without.iter().map(|f| hex(f)).collect::<Vec<_>>(),
+        "enabling capture changed the wire output"
+    );
+    let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn fragmented_input_reconstructs_the_same_captured_frames() {
+    // A frame dribbled one byte at a time must be captured as ONE whole frame,
+    // identical to the same frame sent in a single write.
+    let sid_take = 3;
+
+    let whole_path = cap_path("whole");
+    let _ = std::fs::remove_file(&whole_path);
+    converse(&start_with_capture(Some(whole_path.clone())), sid_take);
+    std::thread::sleep(std::time::Duration::from_millis(150));
+    let whole = capture::read_file(&whole_path).unwrap();
+
+    let frag_path = cap_path("frag");
+    let _ = std::fs::remove_file(&frag_path);
+    let h = start_with_capture(Some(frag_path.clone()));
+    let sid = session_ids(&h.records).into_iter().next().unwrap();
+    let mut stream = connect(&h.addr);
+    let mut buf = Vec::new();
+    for tx in transactions(&h.records, &sid).into_iter().take(sid_take) {
+        for byte in unhex(&st(tx, "request_hex")) {
+            stream.write_all(&[byte]).unwrap();
+            stream.flush().unwrap();
+        }
+        for _ in 0..tx["responses"].as_array().unwrap().len() {
+            read_frame(&mut stream, &mut buf).expect("frame");
+        }
+    }
+    drop(stream);
+    std::thread::sleep(std::time::Duration::from_millis(200));
+    let frag = capture::read_file(&frag_path).unwrap();
+
+    let f_whole: Vec<String> = whole.iter().map(|r| hex(&r.frame)).collect();
+    let f_frag: Vec<String> = frag.iter().map(|r| hex(&r.frame)).collect();
+    assert_eq!(
+        f_frag, f_whole,
+        "fragmentation must not change captured frames"
+    );
+
+    let _ = std::fs::remove_file(&whole_path);
+    let _ = std::fs::remove_file(&frag_path);
+}
+
+#[test]
+fn coalesced_input_is_captured_as_separate_frames() {
+    let path = cap_path("coalesced");
+    let _ = std::fs::remove_file(&path);
+
+    let h = start_with_capture(Some(path.clone()));
+    let sid = session_ids(&h.records).into_iter().next().unwrap();
+    let txs: Vec<&J> = transactions(&h.records, &sid).into_iter().take(4).collect();
+
+    let mut stream = connect(&h.addr);
+    let mut buf = Vec::new();
+    let mut blob = Vec::new();
+    for tx in &txs {
+        blob.extend_from_slice(&unhex(&st(tx, "request_hex")));
+    }
+    stream.write_all(&blob).unwrap(); // ONE write
+    stream.flush().unwrap();
+    for tx in &txs {
+        for _ in 0..tx["responses"].as_array().unwrap().len() {
+            read_frame(&mut stream, &mut buf).expect("frame");
+        }
+    }
+    drop(stream);
+    std::thread::sleep(std::time::Duration::from_millis(200));
+
+    let recs = capture::read_file(&path).unwrap();
+    let rx: Vec<Vec<u8>> = recs
+        .iter()
+        .filter(|r| r.dir == capture::Direction::Rx)
+        .map(|r| r.frame.clone())
+        .collect();
+
+    assert_eq!(rx.len(), txs.len(), "one record per frame, not per read");
+    for (got, tx) in rx.iter().zip(txs.iter()) {
+        assert_eq!(
+            hex(got),
+            st(tx, "request_hex"),
+            "frame boundaries preserved"
+        );
+    }
+    let _ = std::fs::remove_file(&path);
+}
+
+#[test]
+fn sanitization_removes_the_session_key_from_a_real_captured_login() {
+    let path = cap_path("sanitize");
+    let _ = std::fs::remove_file(&path);
+
+    let h = start_with_capture(Some(path.clone()));
+    let sid = session_ids(&h.records).into_iter().next().unwrap();
+    let mut stream = connect(&h.addr);
+    let mut buf = Vec::new();
+    for tx in transactions(&h.records, &sid) {
+        stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap();
+        stream.flush().unwrap();
+        for _ in 0..tx["responses"].as_array().unwrap().len() {
+            read_frame(&mut stream, &mut buf).expect("frame");
+        }
+        if st(tx, "name") == "login" {
+            break;
+        }
+    }
+    drop(stream);
+    std::thread::sleep(std::time::Duration::from_millis(200));
+
+    let recs = capture::read_file(&path).unwrap();
+    let session_key = h
+        .records
+        .iter()
+        .find(|r| r["kind"] == "session" && r["id"] == sid.as_str())
+        .map(|r| st(r, "session_key"))
+        .expect("recorded session key");
+
+    // The raw capture DOES contain the key — that is what makes it forensic
+    // evidence, and why it is gitignored.
+    let raw_has_key = recs
+        .iter()
+        .any(|r| String::from_utf8_lossy(&r.frame).contains(&session_key));
+    assert!(
+        raw_has_key,
+        "raw capture should contain the real session key"
+    );
+
+    let (out, report) = sanitize::sanitize(&recs).expect("sanitizes");
+    assert!(
+        report.frames_redacted > 0,
+        "login frames carry a session key"
+    );
+
+    for (_, bytes) in &out {
+        assert!(
+            !String::from_utf8_lossy(bytes).contains(&session_key),
+            "sanitized output still contains the session key"
+        );
+    }
+    // Structure is untouched: same frame count, same sizes, same order.
+    assert_eq!(out.len(), recs.len());
+    for ((rec, bytes), orig) in out.iter().zip(recs.iter()) {
+        assert_eq!(
+            bytes.len(),
+            orig.frame.len(),
+            "frame size must be preserved"
+        );
+        assert_eq!(rec.seq, orig.seq, "order must be preserved");
+    }
+    // And every redaction is reported, not silent.
+    assert!(
+        report.redactions.iter().any(|r| r.path.ends_with("KEY")),
+        "{:?}",
+        report.redactions
+    );
+
+    let _ = std::fs::remove_file(&path);
+}