test(trace): add legacy characterization replay fixtures

src/trace/characterization.rs

@@ -0,0 +1,497 @@
+// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use super::reader::{TraceCaptureV1, TraceHeaderV1};
+use super::v1::{Direction, Event, RateRequestReason, Record};
+use std::mem;
+use thiserror::Error;
+
+#[derive(Clone, Debug)]
+pub(super) struct CharacterizationV1 {
+    pub(super) header: TraceHeaderV1,
+    pub(super) initialization: Record,
+    pub(super) startup_rates: Record,
+    pub(super) evaluations: Vec<EvaluationCaseV1>,
+    pub(super) trailing_ping_replies: Vec<Record>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct EvaluationCaseV1 {
+    pub(super) ping_replies_before: Vec<Record>,
+    pub(super) boundary: Record,
+    pub(super) download_calculation: Option<Record>,
+    pub(super) upload_calculation: Option<Record>,
+    pub(super) random_choices: Vec<Record>,
+    pub(super) expected_rates: Record,
+    pub(super) deferred_ping_replies: Vec<Record>,
+}
+
+#[derive(Debug, Error)]
+pub(super) enum CharacterizationErrorV1 {
+    #[error("trace has no controller initialization")]
+    MissingInitialization,
+    #[error("trace has no startup requested-rates record")]
+    MissingStartupRates,
+    #[error("record {seq} is structurally unexpected: {detail}")]
+    UnexpectedRecord { seq: u64, detail: &'static str },
+    #[error("evaluation {evaluation_id} is incomplete at EOF")]
+    IncompleteEvaluation { evaluation_id: u64 },
+    #[error("evaluation {evaluation_id} has invalid structure: {detail}")]
+    InvalidEvaluation {
+        evaluation_id: u64,
+        detail: &'static str,
+    },
+}
+
+struct OpenEvaluation {
+    evaluation_id: u64,
+    ping_replies_before: Vec<Record>,
+    boundary: Record,
+    download_calculation: Option<Record>,
+    upload_calculation: Option<Record>,
+    random_choices: Vec<Record>,
+    deferred_ping_replies: Vec<Record>,
+}
+
+pub(super) fn build_characterization_v1(
+    capture: &TraceCaptureV1,
+) -> Result<CharacterizationV1, CharacterizationErrorV1> {
+    let mut initialization = None;
+    let mut startup_rates = None;
+    let mut evaluations = Vec::new();
+    let mut pending_ping_replies = Vec::new();
+    let mut open = None::<OpenEvaluation>;
+
+    for record in &capture.records {
+        match &record.event {
+            Event::PingReply { .. } => {
+                if let Some(evaluation) = &mut open {
+                    evaluation.deferred_ping_replies.push(record.clone());
+                } else {
+                    pending_ping_replies.push(record.clone());
+                }
+            }
+            Event::ControllerInitialized { .. } => {
+                if initialization.is_some() || open.is_some() || !evaluations.is_empty() {
+                    return unexpected(record, "duplicate or late controller initialization");
+                }
+                initialization = Some(record.clone());
+            }
+            Event::ControlEvaluation { evaluation_id, .. } => {
+                if initialization.is_none() || startup_rates.is_none() {
+                    return unexpected(record, "evaluation precedes controller startup");
+                }
+                if open.is_some() {
+                    return unexpected(record, "evaluation begins before the previous one ends");
+                }
+                if evaluations
+                    .last()
+                    .is_none_or(|previous: &EvaluationCaseV1| {
+                        evaluation_id_of(&previous.boundary).wrapping_add(1) != *evaluation_id
+                    })
+                    && (!evaluations.is_empty() || *evaluation_id != 0)
+                {
+                    return invalid(*evaluation_id, "evaluation ids are not contiguous");
+                }
+                open = Some(OpenEvaluation {
+                    evaluation_id: *evaluation_id,
+                    ping_replies_before: mem::take(&mut pending_ping_replies),
+                    boundary: record.clone(),
+                    download_calculation: None,
+                    upload_calculation: None,
+                    random_choices: Vec::new(),
+                    deferred_ping_replies: Vec::new(),
+                });
+            }
+            Event::RateCalculation {
+                evaluation_id,
+                direction,
+            } => {
+                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
+                let calculation = match direction {
+                    Direction::Download => &mut evaluation.download_calculation,
+                    Direction::Upload => &mut evaluation.upload_calculation,
+                };
+                if calculation.replace(record.clone()).is_some() {
+                    return invalid(*evaluation_id, "duplicate directional calculation");
+                }
+            }
+            Event::RandomSafeRateChoice {
+                evaluation_id,
+                direction,
+                ..
+            } => {
+                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
+                let has_calculation = match direction {
+                    Direction::Download => evaluation.download_calculation.is_some(),
+                    Direction::Upload => evaluation.upload_calculation.is_some(),
+                };
+                if !has_calculation {
+                    return invalid(*evaluation_id, "random choice precedes its calculation");
+                }
+                evaluation.random_choices.push(record.clone());
+            }
+            Event::RequestedRates {
+                evaluation_id: None,
+                reason: RateRequestReason::Startup,
+                ..
+            } => {
+                if startup_rates.is_some() || open.is_some() || !evaluations.is_empty() {
+                    return unexpected(record, "duplicate or late startup rates");
+                }
+                startup_rates = Some(record.clone());
+            }
+            Event::RequestedRates {
+                evaluation_id: Some(evaluation_id),
+                reason,
+                ..
+            } => {
+                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
+                match reason {
+                    RateRequestReason::Control
+                        if evaluation.download_calculation.is_none()
+                            || evaluation.upload_calculation.is_none() =>
+                    {
+                        return invalid(*evaluation_id, "control evaluation lacks a calculation");
+                    }
+                    RateRequestReason::NoReflectorData
+                        if evaluation.download_calculation.is_some()
+                            || evaluation.upload_calculation.is_some()
+                            || !evaluation.random_choices.is_empty() =>
+                    {
+                        return invalid(
+                            *evaluation_id,
+                            "no-reflector evaluation contains calculation metadata",
+                        );
+                    }
+                    RateRequestReason::Startup => {
+                        return invalid(*evaluation_id, "startup rates have an evaluation id");
+                    }
+                    _ => {}
+                }
+                let evaluation = open.take().expect("matching evaluation checked above");
+                evaluations.push(EvaluationCaseV1 {
+                    ping_replies_before: evaluation.ping_replies_before,
+                    boundary: evaluation.boundary,
+                    download_calculation: evaluation.download_calculation,
+                    upload_calculation: evaluation.upload_calculation,
+                    random_choices: evaluation.random_choices,
+                    expected_rates: record.clone(),
+                    deferred_ping_replies: evaluation.deferred_ping_replies,
+                });
+            }
+            Event::RequestedRates { .. } => {
+                return unexpected(record, "requested-rates reason and evaluation id disagree");
+            }
+            Event::Header { .. } | Event::End { .. } => {
+                return unexpected(record, "header or end leaked from the trace reader");
+            }
+        }
+    }
+
+    if let Some(evaluation) = open {
+        return Err(CharacterizationErrorV1::IncompleteEvaluation {
+            evaluation_id: evaluation.evaluation_id,
+        });
+    }
+
+    Ok(CharacterizationV1 {
+        header: capture.header.clone(),
+        initialization: initialization.ok_or(CharacterizationErrorV1::MissingInitialization)?,
+        startup_rates: startup_rates.ok_or(CharacterizationErrorV1::MissingStartupRates)?,
+        evaluations,
+        trailing_ping_replies: pending_ping_replies,
+    })
+}
+
+fn matching_open<'a>(
+    open: &'a mut Option<OpenEvaluation>,
+    record: &Record,
+    evaluation_id: u64,
+) -> Result<&'a mut OpenEvaluation, CharacterizationErrorV1> {
+    let Some(evaluation) = open else {
+        return unexpected(record, "evaluation-scoped record has no open evaluation");
+    };
+    if evaluation.evaluation_id != evaluation_id {
+        return invalid(
+            evaluation_id,
+            "record refers to a different open evaluation",
+        );
+    }
+    Ok(evaluation)
+}
+
+fn evaluation_id_of(record: &Record) -> u64 {
+    let Event::ControlEvaluation { evaluation_id, .. } = record.event else {
+        unreachable!("builder only stores control-evaluation boundaries")
+    };
+    evaluation_id
+}
+
+fn unexpected<T>(record: &Record, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
+    Err(CharacterizationErrorV1::UnexpectedRecord {
+        seq: record.seq,
+        detail,
+    })
+}
+
+fn invalid<T>(evaluation_id: u64, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
+    Err(CharacterizationErrorV1::InvalidEvaluation {
+        evaluation_id,
+        detail,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::trace::reader::{CaptureCompletionV1, read_trace_v1};
+    use crate::trace::v1::{RateRequestReason, Timestamp};
+    use std::collections::HashSet;
+    use std::io::Cursor;
+
+    const LIVE_FIXTURE: &str = include_str!("../../tests/fixtures/trace-v1/speedtests.jsonl");
+
+    fn record(seq: u64, event: Event) -> Record {
+        Record {
+            seq,
+            timestamp: Timestamp {
+                mono_ns: seq,
+                unix_us: seq as i64,
+            },
+            event,
+        }
+    }
+
+    fn jsonl(records: &[Record]) -> String {
+        records
+            .iter()
+            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
+            .collect()
+    }
+
+    fn initialization(seq: u64) -> Record {
+        record(
+            seq,
+            Event::ControllerInitialized {
+                previous_rx_bytes: 10,
+                previous_tx_bytes: 20,
+                download_prev_mono_ns: 1,
+                upload_prev_mono_ns: 2,
+                download_history_index: 0,
+                upload_history_index: 0,
+                download_safe_rates_kbit: vec![100.0],
+                upload_safe_rates_kbit: vec![200.0],
+            },
+        )
+    }
+
+    fn requested(seq: u64, evaluation_id: Option<u64>, reason: RateRequestReason) -> Record {
+        record(
+            seq,
+            Event::RequestedRates {
+                evaluation_id,
+                reason,
+                download_kbit: 100,
+                upload_kbit: 200,
+                download_requested: true,
+                upload_requested: true,
+            },
+        )
+    }
+
+    fn ping(seq: u64, reflector: &str) -> Record {
+        record(
+            seq,
+            Event::PingReply {
+                reflector: reflector.parse().unwrap(),
+                down_time_ms: 1.0,
+                up_time_ms: 2.0,
+            },
+        )
+    }
+
+    fn boundary(seq: u64, evaluation_id: u64) -> Record {
+        record(
+            seq,
+            Event::ControlEvaluation {
+                evaluation_id,
+                loop_mono_ns: seq,
+                rx_bytes: 30,
+                tx_bytes: 40,
+                active_reflectors: vec!["192.0.2.1".parse().unwrap()],
+            },
+        )
+    }
+
+    fn calculation(seq: u64, evaluation_id: u64, direction: Direction) -> Record {
+        record(
+            seq,
+            Event::RateCalculation {
+                evaluation_id,
+                direction,
+            },
+        )
+    }
+
+    fn synthetic_capture(records: &[Record]) -> TraceCaptureV1 {
+        let mut all = vec![record(0, crate::trace::v1::tests::header_event())];
+        all.extend_from_slice(records);
+        let input = jsonl(&all);
+        read_trace_v1(Cursor::new(input)).unwrap()
+    }
+
+    #[test]
+    fn freezes_evaluation_before_interleaved_ping_replies() {
+        let capture = synthetic_capture(&[
+            initialization(1),
+            requested(2, None, RateRequestReason::Startup),
+            ping(3, "192.0.2.1"),
+            boundary(4, 0),
+            calculation(5, 0, Direction::Download),
+            ping(6, "192.0.2.2"),
+            calculation(7, 0, Direction::Upload),
+            requested(8, Some(0), RateRequestReason::Control),
+            ping(9, "192.0.2.3"),
+            boundary(10, 1),
+            calculation(11, 1, Direction::Download),
+            calculation(12, 1, Direction::Upload),
+            requested(13, Some(1), RateRequestReason::Control),
+        ]);
+
+        let characterization = build_characterization_v1(&capture).unwrap();
+
+        assert_eq!(characterization.header.seq, 0);
+        assert!(characterization.trailing_ping_replies.is_empty());
+        assert_eq!(
+            characterization.evaluations[0].ping_replies_before[0].seq,
+            3
+        );
+        assert_eq!(
+            characterization.evaluations[0].deferred_ping_replies[0].seq,
+            6
+        );
+        assert_eq!(
+            characterization.evaluations[1].ping_replies_before[0].seq,
+            9
+        );
+        assert!(
+            characterization.evaluations[1]
+                .deferred_ping_replies
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn live_fixture_parses_as_an_incomplete_observed_prefix() {
+        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
+
+        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
+        assert_eq!(capture.header.program_version, "0.4.1");
+        assert_eq!(capture.records.last().unwrap().seq, 1515);
+    }
+
+    #[test]
+    fn live_fixture_covers_useful_legacy_decisions() {
+        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
+        let characterization = build_characterization_v1(&capture).unwrap();
+
+        assert!(matches!(
+            characterization.initialization.event,
+            Event::ControllerInitialized { .. }
+        ));
+        assert!(matches!(
+            characterization.startup_rates.event,
+            Event::RequestedRates {
+                reason: RateRequestReason::Startup,
+                ..
+            }
+        ));
+        assert!(characterization.evaluations.iter().any(|evaluation| {
+            let Event::ControlEvaluation {
+                active_reflectors, ..
+            } = &evaluation.boundary.event
+            else {
+                unreachable!();
+            };
+            active_reflectors.len() > 5
+        }));
+
+        let stochastic_index = characterization
+            .evaluations
+            .iter()
+            .position(|evaluation| !evaluation.random_choices.is_empty())
+            .expect("fixture must exercise the legacy stochastic branch");
+        let stochastic = &characterization.evaluations[stochastic_index];
+        assert_eq!(evaluation_id_of(&stochastic.boundary), 150);
+        assert!(matches!(
+            stochastic.random_choices.as_slice(),
+            [Record {
+                event: Event::RandomSafeRateChoice {
+                    direction: Direction::Download,
+                    index: 92,
+                    ..
+                },
+                ..
+            }]
+        ));
+        let Event::RequestedRates {
+            download_kbit,
+            download_requested,
+            ..
+        } = stochastic.expected_rates.event
+        else {
+            unreachable!();
+        };
+        assert_eq!(download_kbit, 256_374);
+        assert!(download_requested);
+        let Event::RequestedRates {
+            download_kbit: previous_download_kbit,
+            ..
+        } = characterization.evaluations[stochastic_index - 1]
+            .expected_rates
+            .event
+        else {
+            unreachable!();
+        };
+        assert!(download_kbit < previous_download_kbit);
+
+        let distinct_rates = characterization
+            .evaluations
+            .iter()
+            .map(|evaluation| match evaluation.expected_rates.event {
+                Event::RequestedRates {
+                    download_kbit,
+                    upload_kbit,
+                    ..
+                } => (download_kbit, upload_kbit),
+                _ => unreachable!(),
+            })
+            .collect::<HashSet<_>>();
+        assert!(distinct_rates.len() > 3);
+    }
+
+    #[test]
+    fn live_expected_rates_have_one_unambiguous_evaluation() {
+        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
+        let characterization = build_characterization_v1(&capture).unwrap();
+        let mut ids = HashSet::new();
+
+        for evaluation in &characterization.evaluations {
+            let evaluation_id = evaluation_id_of(&evaluation.boundary);
+            let Event::RequestedRates {
+                evaluation_id: Some(expected_id),
+                reason: RateRequestReason::Control,
+                ..
+            } = evaluation.expected_rates.event
+            else {
+                panic!("fixture evaluation must have normal control output");
+            };
+            assert_eq!(expected_id, evaluation_id);
+            assert!(ids.insert(evaluation_id));
+            assert!(evaluation.download_calculation.is_some());
+            assert!(evaluation.upload_calculation.is_some());
+        }
+    }
+}

src/trace/mod.rs

@@ -4,6 +4,11 @@
 
 pub(crate) mod v1;
 
+#[cfg(all(feature = "trace", test))]
+mod characterization;
+#[cfg(all(feature = "trace", test))]
+mod reader;
+
 #[cfg(feature = "trace")]
 mod writer;
 
@@ -374,7 +379,7 @@ mod tests {
             let contents = std::fs::read_to_string(&path).unwrap();
             let records = contents
                 .lines()
-                .map(|line| serde_json::from_str::<v1::Record>(line).unwrap())
+                .map(|line| super::reader::parse_record_v1(line).unwrap())
                 .collect::<Vec<_>>();
 
             assert_eq!(records.len(), 102);

src/trace/reader.rs

@@ -0,0 +1,343 @@
+// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
+//
+// SPDX-License-Identifier: MPL-2.0
+
+use super::v1::{self, ControlConfig, Event, Record, Timestamp};
+use std::io::{self, BufRead};
+use thiserror::Error;
+
+#[derive(Clone, Debug, PartialEq)]
+pub(super) struct TraceHeaderV1 {
+    pub(super) seq: u64,
+    pub(super) timestamp: Timestamp,
+    pub(super) program_version: String,
+    pub(super) control: ControlConfig,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(super) enum CaptureCompletionV1 {
+    Complete,
+    IncompleteEof,
+    IncompleteEnd,
+    TruncatedFinalLine { line: usize },
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub(super) struct TraceCaptureV1 {
+    pub(super) header: TraceHeaderV1,
+    pub(super) records: Vec<Record>,
+    pub(super) completion: CaptureCompletionV1,
+}
+
+#[derive(Debug, Error)]
+pub(super) enum TraceReadErrorV1 {
+    #[error("failed to read trace line {line}: {source}")]
+    Io {
+        line: usize,
+        #[source]
+        source: io::Error,
+    },
+    #[error("trace line {line} is not a valid v1 record: {source}")]
+    MalformedRecord {
+        line: usize,
+        #[source]
+        source: serde_json::Error,
+    },
+    #[error("trace is missing its v1 header")]
+    MissingHeader,
+    #[error("first trace record must be the v1 header")]
+    FirstRecordNotHeader,
+    #[error("trace header sequence must be 0, got {actual}")]
+    InvalidHeaderSequence { actual: u64 },
+    #[error("unsupported trace format {actual:?}")]
+    InvalidFormat { actual: String },
+    #[error("unsupported trace version {actual}")]
+    InvalidVersion { actual: u16 },
+    #[error("trace line {line} has sequence {actual}, expected {expected}")]
+    InvalidSequence {
+        line: usize,
+        expected: u64,
+        actual: u64,
+    },
+    #[error("trace line {line} contains another header")]
+    UnexpectedHeader { line: usize },
+    #[error("trace line {line} follows an end record")]
+    RecordAfterEnd { line: usize },
+}
+
+pub(super) fn read_trace_v1(mut reader: impl BufRead) -> Result<TraceCaptureV1, TraceReadErrorV1> {
+    let mut header = None;
+    let mut records = Vec::new();
+    let mut next_seq = 0;
+    let mut completion = None;
+    let mut line_number = 0;
+
+    loop {
+        let mut line = String::new();
+        line_number += 1;
+        let bytes_read = reader
+            .read_line(&mut line)
+            .map_err(|source| TraceReadErrorV1::Io {
+                line: line_number,
+                source,
+            })?;
+        if bytes_read == 0 {
+            break;
+        }
+
+        if completion.is_some() {
+            return Err(TraceReadErrorV1::RecordAfterEnd { line: line_number });
+        }
+
+        let terminated = line.ends_with('\n');
+        let json = line.strip_suffix('\n').unwrap_or(&line);
+        let json = json.strip_suffix('\r').unwrap_or(json);
+        let record = match parse_record_v1(json) {
+            Ok(record) => record,
+            Err(source)
+                if !terminated
+                    && !json.trim().is_empty()
+                    && source.is_eof()
+                    && header.is_some() =>
+            {
+                return Ok(TraceCaptureV1 {
+                    header: header.expect("header checked above"),
+                    records,
+                    completion: CaptureCompletionV1::TruncatedFinalLine { line: line_number },
+                });
+            }
+            Err(source) => {
+                return Err(TraceReadErrorV1::MalformedRecord {
+                    line: line_number,
+                    source,
+                });
+            }
+        };
+
+        if record.seq != next_seq {
+            if header.is_none() && next_seq == 0 {
+                return Err(TraceReadErrorV1::InvalidHeaderSequence { actual: record.seq });
+            }
+            return Err(TraceReadErrorV1::InvalidSequence {
+                line: line_number,
+                expected: next_seq,
+                actual: record.seq,
+            });
+        }
+        next_seq = next_seq.saturating_add(1);
+
+        if header.is_none() {
+            let Event::Header {
+                format,
+                version,
+                program_version,
+                control,
+            } = record.event
+            else {
+                return Err(TraceReadErrorV1::FirstRecordNotHeader);
+            };
+            if format != v1::FORMAT_NAME {
+                return Err(TraceReadErrorV1::InvalidFormat { actual: format });
+            }
+            if version != v1::FORMAT_VERSION {
+                return Err(TraceReadErrorV1::InvalidVersion { actual: version });
+            }
+            header = Some(TraceHeaderV1 {
+                seq: record.seq,
+                timestamp: record.timestamp,
+                program_version,
+                control,
+            });
+            continue;
+        }
+
+        match record.event {
+            Event::Header { .. } => {
+                return Err(TraceReadErrorV1::UnexpectedHeader { line: line_number });
+            }
+            Event::End { clean } => {
+                completion = Some(if clean {
+                    CaptureCompletionV1::Complete
+                } else {
+                    CaptureCompletionV1::IncompleteEnd
+                });
+            }
+            _ => records.push(record),
+        }
+    }
+
+    let header = header.ok_or(TraceReadErrorV1::MissingHeader)?;
+    Ok(TraceCaptureV1 {
+        header,
+        records,
+        completion: completion.unwrap_or(CaptureCompletionV1::IncompleteEof),
+    })
+}
+
+pub(super) fn parse_record_v1(json: &str) -> Result<Record, serde_json::Error> {
+    serde_json::from_str(json)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Cursor;
+
+    fn record(seq: u64, event: Event) -> Record {
+        Record {
+            seq,
+            timestamp: Timestamp {
+                mono_ns: seq,
+                unix_us: seq as i64,
+            },
+            event,
+        }
+    }
+
+    fn jsonl(records: &[Record]) -> String {
+        records
+            .iter()
+            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
+            .collect()
+    }
+
+    fn header(seq: u64) -> Record {
+        record(seq, v1::tests::header_event())
+    }
+
+    fn ping(seq: u64) -> Record {
+        record(
+            seq,
+            Event::PingReply {
+                reflector: "192.0.2.1".parse().unwrap(),
+                down_time_ms: 1.0,
+                up_time_ms: 2.0,
+            },
+        )
+    }
+
+    #[test]
+    fn parses_clean_v1_trace() {
+        let input = jsonl(&[header(0), ping(1), record(2, Event::End { clean: true })]);
+
+        let capture = read_trace_v1(Cursor::new(input)).unwrap();
+
+        assert_eq!(capture.header.seq, 0);
+        assert_eq!(capture.records, vec![ping(1)]);
+        assert_eq!(capture.completion, CaptureCompletionV1::Complete);
+    }
+
+    #[test]
+    fn retains_records_before_truncated_final_line() {
+        let input = format!(
+            "{}{{\"seq\":2,\"timestamp\":{{",
+            jsonl(&[header(0), ping(1)])
+        );
+
+        let capture = read_trace_v1(Cursor::new(input)).unwrap();
+
+        assert_eq!(capture.records, vec![ping(1)]);
+        assert_eq!(
+            capture.completion,
+            CaptureCompletionV1::TruncatedFinalLine { line: 3 }
+        );
+    }
+
+    #[test]
+    fn marks_valid_eof_without_end_incomplete() {
+        let capture = read_trace_v1(Cursor::new(jsonl(&[header(0), ping(1)]))).unwrap();
+
+        assert_eq!(capture.records, vec![ping(1)]);
+        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
+    }
+
+    #[test]
+    fn rejects_malformed_middle_line() {
+        let input = format!("{}not-json\n{}", jsonl(&[header(0)]), jsonl(&[ping(1)]));
+
+        assert!(matches!(
+            read_trace_v1(Cursor::new(input)),
+            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
+        ));
+    }
+
+    #[test]
+    fn rejects_sequence_gap_duplicate_and_out_of_order() {
+        let cases = [
+            (vec![header(0), ping(2)], 2, 1, 2),
+            (vec![header(0), ping(1), ping(1)], 3, 2, 1),
+            (vec![header(0), ping(1), ping(2), ping(1)], 4, 3, 1),
+        ];
+
+        for (records, line, expected, actual) in cases {
+            let input = jsonl(&records);
+            assert!(matches!(
+                read_trace_v1(Cursor::new(input)),
+                Err(TraceReadErrorV1::InvalidSequence {
+                    line: seen_line,
+                    expected: seen_expected,
+                    actual: seen,
+                }) if seen_line == line && seen_expected == expected && seen == actual
+            ));
+        }
+    }
+
+    #[test]
+    fn rejects_non_truncation_error_on_final_line() {
+        let input = format!("{}{{\"seq\":1,]", jsonl(&[header(0)]));
+
+        assert!(matches!(
+            read_trace_v1(Cursor::new(input)),
+            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
+        ));
+    }
+
+    #[test]
+    fn rejects_records_after_clean_end() {
+        let input = jsonl(&[header(0), record(1, Event::End { clean: true }), ping(2)]);
+
+        assert!(matches!(
+            read_trace_v1(Cursor::new(input)),
+            Err(TraceReadErrorV1::RecordAfterEnd { line: 3 })
+        ));
+    }
+
+    #[test]
+    fn validates_format_name_and_version() {
+        let Event::Header {
+            program_version,
+            control,
+            ..
+        } = v1::tests::header_event()
+        else {
+            unreachable!();
+        };
+        let invalid_format = record(
+            0,
+            Event::Header {
+                format: "other".to_owned(),
+                version: v1::FORMAT_VERSION,
+                program_version: program_version.clone(),
+                control: control.clone(),
+            },
+        );
+        let invalid_version = record(
+            0,
+            Event::Header {
+                format: v1::FORMAT_NAME.to_owned(),
+                version: v1::FORMAT_VERSION + 1,
+                program_version,
+                control,
+            },
+        );
+
+        assert!(matches!(
+            read_trace_v1(Cursor::new(jsonl(&[invalid_format]))),
+            Err(TraceReadErrorV1::InvalidFormat { .. })
+        ));
+        assert!(matches!(
+            read_trace_v1(Cursor::new(jsonl(&[invalid_version]))),
+            Err(TraceReadErrorV1::InvalidVersion { .. })
+        ));
+    }
+}

src/trace/v1.rs

@@ -2,7 +2,9 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
+use serde::de::{Deserializer, Visitor};
 use serde::{Deserialize, Serialize};
+use std::fmt;
 use std::net::IpAddr;
 
 #[cfg(feature = "trace")]
@@ -51,7 +53,9 @@ pub(crate) enum Event {
         up_time_ms: f64,
     },
     ControllerInitialized {
+        #[serde(deserialize_with = "deserialize_i128")]
         previous_rx_bytes: i128,
+        #[serde(deserialize_with = "deserialize_i128")]
         previous_tx_bytes: i128,
         download_prev_mono_ns: u64,
         upload_prev_mono_ns: u64,
@@ -63,7 +67,9 @@ pub(crate) enum Event {
     ControlEvaluation {
         evaluation_id: u64,
         loop_mono_ns: u64,
+        #[serde(deserialize_with = "deserialize_i128")]
         rx_bytes: i128,
+        #[serde(deserialize_with = "deserialize_i128")]
         tx_bytes: i128,
         active_reflectors: Vec<IpAddr>,
     },
@@ -90,6 +96,31 @@ pub(crate) enum Event {
     },
 }
 
+fn deserialize_i128<'de, D>(deserializer: D) -> Result<i128, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    struct I128Visitor;
+
+    impl Visitor<'_> for I128Visitor {
+        type Value = i128;
+
+        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+            formatter.write_str("a signed or unsigned integer")
+        }
+
+        fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
+            Ok(i128::from(value))
+        }
+
+        fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
+            Ok(i128::from(value))
+        }
+    }
+
+    deserializer.deserialize_any(I128Visitor)
+}
+
 #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub(crate) enum Direction {
@@ -218,7 +249,11 @@ pub(super) mod tests {
         assert_eq!(json["format"], FORMAT_NAME);
         assert_eq!(json["version"], FORMAT_VERSION);
         assert_eq!(json["timestamp"]["mono_ns"], 0);
-        assert_eq!(serde_json::from_value::<Record>(json).unwrap(), record);
+        assert_eq!(
+            crate::trace::reader::parse_record_v1(&serde_json::to_string(&record).unwrap())
+                .unwrap(),
+            record
+        );
     }
 
     #[test]

tests/fixtures/trace-v1/README.md

@@ -0,0 +1,7 @@
+# Trace v1 characterization fixture
+
+`speedtests.jsonl` is an observed prefix of a live legacy-controller capture
+recorded during speed and LibreQoS testing. It preserves the original records,
+sequence values, and order through evaluation 150. The prefix intentionally has
+no synthetic `end` record and may include undesirable legacy behavior; it is an
+extraction-characterization artifact, not a correctness oracle.