sim_lib_stream_core/
cassette.rs

1//! Golden-fixture record and replay for streams.
2//!
3//! A [`StreamCassette`] captures a deterministic trace of a stream -- its
4//! metadata, the ordered envelopes it produced, derived timing, accumulated
5//! diagnostics, and the final stats -- so that the same trace can be replayed
6//! as a fresh [`StreamValue`] or persisted as a golden fixture for tests. The
7//! kernel supplies the protocol types ([`Expr`], [`Symbol`], [`Error`]) used to
8//! serialize a cassette; this module supplies the concrete streaming-fabric
9//! behavior that records, redacts, validates, and round-trips those traces.
10
11use sim_kernel::{Error, Expr, Result, Symbol};
12use sim_value::access;
13
14#[path = "cassette/redaction.rs"]
15mod redaction;
16#[path = "cassette/stats.rs"]
17mod stats;
18
19use crate::buffer::{expr_kind, field, string_field, symbol_field};
20use crate::{
21    StreamCapability, StreamEnvelope, StreamItem, StreamMetadata, StreamPacket, StreamStats,
22    StreamValue, TransportProfile,
23};
24
25use redaction::{
26    envelope_has_host_device, is_host_device_symbol, metadata_has_host_device,
27    packet_has_private_payload, redact_envelope, redact_metadata, redact_symbol,
28};
29use stats::{stream_stats_expr, stream_stats_from_expr};
30
31/// Repository-relative root directory under which golden stream fixtures live.
32pub const STREAM_CASSETTE_FIXTURE_ROOT: &str = "fixtures/streams/golden";
33/// File extension (without the leading dot) for a persisted golden fixture.
34pub const STREAM_CASSETTE_EXTENSION: &str = "simcassette";
35
36/// Derived timing summary for a recorded cassette.
37///
38/// Captures the clock the stream ran on, how many packets were recorded, the
39/// sequence range of the first and last envelopes, and whether the trace is
40/// finite (a golden fixture must be finite).
41#[derive(Clone, Debug, PartialEq, Eq)]
42pub struct StreamCassetteTiming {
43    /// Clock-domain symbol the recorded stream advanced against.
44    pub clock: Symbol,
45    /// Number of envelopes captured in the cassette.
46    pub packet_count: usize,
47    /// Sequence number of the first envelope, or `None` when empty.
48    pub first_sequence: Option<u64>,
49    /// Sequence number of the last envelope, or `None` when empty.
50    pub last_sequence: Option<u64>,
51    /// Whether the recorded trace terminated; golden fixtures must be finite.
52    pub finite: bool,
53}
54
55/// A recorded, replayable trace of a single stream.
56///
57/// Holds the stream metadata, the ordered envelopes, derived [timing](StreamCassetteTiming),
58/// the deduplicated diagnostic symbols observed, and the final [`StreamStats`].
59/// A cassette can be rebuilt into a live [`StreamValue`] via
60/// [`replay_stream_value`](StreamCassette::replay_stream_value), serialized to
61/// an [`Expr`] map, and validated as a golden fixture.
62#[derive(Clone, Debug, PartialEq, Eq)]
63pub struct StreamCassette {
64    metadata: StreamMetadata,
65    envelopes: Vec<StreamEnvelope>,
66    timing: StreamCassetteTiming,
67    diagnostics: Vec<Symbol>,
68    final_stats: StreamStats,
69}
70
71/// Outcome of validating a cassette against the golden-fixture rules.
72///
73/// Returned by [`StreamCassette::validate_golden_fixture`] once a cassette
74/// passes every fixture invariant; records where the fixture lives, its format
75/// symbol, the packet count, and the final stats.
76#[derive(Clone, Debug, PartialEq, Eq)]
77pub struct StreamGoldenFixtureReport {
78    /// Validated fixture path under [`STREAM_CASSETTE_FIXTURE_ROOT`].
79    pub path: String,
80    /// Cassette format symbol the fixture was written with.
81    pub format: Symbol,
82    /// Number of envelopes the fixture contains.
83    pub packet_count: usize,
84    /// Final accumulated stats captured at the end of the trace.
85    pub final_stats: StreamStats,
86}
87
88impl StreamCassette {
89    /// Records a cassette by draining every packet from a live stream.
90    ///
91    /// Pulls packets until the stream is exhausted, snapshots its final stats,
92    /// and builds the cassette from the metadata, drained items, and the given
93    /// [`TransportProfile`].
94    pub fn from_stream_value(stream: &StreamValue, profile: TransportProfile) -> Result<Self> {
95        let mut items = Vec::new();
96        while let Some(item) = stream.next_packet()? {
97            items.push(item);
98        }
99        let final_stats = stream.stats()?;
100        Self::from_items(stream.metadata().clone(), items, profile, final_stats)
101    }
102
103    /// Records a cassette from already-drained stream items.
104    ///
105    /// Wraps each item in a sequenced [`StreamEnvelope`] under the given
106    /// [`TransportProfile`], then delegates to
107    /// [`from_envelopes`](StreamCassette::from_envelopes).
108    pub fn from_items(
109        metadata: StreamMetadata,
110        items: Vec<StreamItem>,
111        profile: TransportProfile,
112        final_stats: StreamStats,
113    ) -> Result<Self> {
114        let envelopes = items
115            .iter()
116            .enumerate()
117            .map(|(sequence, item)| {
118                StreamEnvelope::from_item_with_profile(
119                    &metadata,
120                    sequence as u64,
121                    item,
122                    profile.clone(),
123                )
124            })
125            .collect::<Result<Vec<_>>>()?;
126        Self::from_envelopes(metadata, envelopes, final_stats)
127    }
128
129    /// Builds a cassette directly from sequenced envelopes.
130    ///
131    /// Derives the [timing](StreamCassetteTiming) and the deduplicated
132    /// diagnostic set from the envelopes, pairing them with the supplied
133    /// metadata and final stats.
134    pub fn from_envelopes(
135        metadata: StreamMetadata,
136        envelopes: Vec<StreamEnvelope>,
137        final_stats: StreamStats,
138    ) -> Result<Self> {
139        let timing = timing_from_envelopes(&metadata, &envelopes);
140        let diagnostics = diagnostics_from_envelopes(&envelopes);
141        Ok(Self {
142            metadata,
143            envelopes,
144            timing,
145            diagnostics,
146            final_stats,
147        })
148    }
149
150    /// Returns the metadata of the recorded stream.
151    pub fn metadata(&self) -> &StreamMetadata {
152        &self.metadata
153    }
154
155    /// Returns the recorded envelopes in sequence order.
156    pub fn envelopes(&self) -> &[StreamEnvelope] {
157        &self.envelopes
158    }
159
160    /// Returns the derived timing summary for the trace.
161    pub fn timing(&self) -> &StreamCassetteTiming {
162        &self.timing
163    }
164
165    /// Returns the deduplicated diagnostic symbols observed during recording.
166    pub fn diagnostics(&self) -> &[Symbol] {
167        &self.diagnostics
168    }
169
170    /// Returns the final accumulated stats captured at end of trace.
171    pub fn final_stats(&self) -> &StreamStats {
172        &self.final_stats
173    }
174
175    /// Reconstructs the stream items from the recorded envelopes.
176    ///
177    /// Each item pairs an envelope's packet with its captured ticks, ready to
178    /// feed a replay stream.
179    pub fn items(&self) -> Result<Vec<StreamItem>> {
180        self.envelopes
181            .iter()
182            .map(|envelope| {
183                StreamItem::with_ticks(envelope.packet().clone(), envelope.ticks().to_vec())
184            })
185            .collect()
186    }
187
188    /// Rebuilds a live, pull-based [`StreamValue`] from the recorded trace.
189    pub fn replay_stream_value(&self) -> Result<StreamValue> {
190        Ok(StreamValue::pull(self.metadata.clone(), self.items()?))
191    }
192
193    /// Serializes the cassette to an [`Expr`] map keyed by field symbol.
194    ///
195    /// The map carries the format symbol, metadata table, timing, envelope
196    /// list, diagnostics, and final stats, suitable for persistence and
197    /// round-tripping through [`from_expr`](StreamCassette::from_expr).
198    pub fn to_expr(&self) -> Expr {
199        Expr::Map(vec![
200            (
201                Expr::Symbol(Symbol::new("cassette")),
202                Expr::Symbol(stream_cassette_format_symbol()),
203            ),
204            (
205                Expr::Symbol(Symbol::new("metadata")),
206                self.metadata.table_expr(),
207            ),
208            (Expr::Symbol(Symbol::new("timing")), self.timing.to_expr()),
209            (
210                Expr::Symbol(Symbol::new("envelopes")),
211                Expr::List(self.envelopes.iter().map(StreamEnvelope::to_expr).collect()),
212            ),
213            (
214                Expr::Symbol(Symbol::new("diagnostics")),
215                Expr::List(self.diagnostics.iter().cloned().map(Expr::Symbol).collect()),
216            ),
217            (
218                Expr::Symbol(Symbol::new("final-stats")),
219                stream_stats_expr(&self.final_stats),
220            ),
221        ])
222    }
223
224    /// Deserializes a cassette from an [`Expr`] map produced by
225    /// [`to_expr`](StreamCassette::to_expr).
226    ///
227    /// Validates the field set and format symbol, then reconstructs metadata,
228    /// envelopes, timing, diagnostics, and final stats. Fails closed on an
229    /// unknown format, missing or unexpected fields, or type mismatches.
230    pub fn from_expr(expr: &Expr) -> Result<Self> {
231        let Expr::Map(entries) = expr else {
232            return Err(Error::TypeMismatch {
233                expected: "stream cassette map",
234                found: expr_kind(expr),
235            });
236        };
237        ensure_fields(
238            entries,
239            &[
240                "cassette",
241                "metadata",
242                "timing",
243                "envelopes",
244                "diagnostics",
245                "final-stats",
246            ],
247        )?;
248        let format = symbol_field(entries, "cassette")?;
249        if *format != stream_cassette_format_symbol() {
250            return Err(Error::Eval(format!(
251                "unknown stream cassette format {}",
252                format.as_qualified_str()
253            )));
254        }
255        let metadata = StreamMetadata::from_table_expr(field(entries, "metadata")?)?;
256        let envelopes = list_field(entries, "envelopes")?
257            .iter()
258            .map(|expr| StreamEnvelope::try_from(expr.clone()))
259            .collect::<Result<Vec<_>>>()?;
260        let metadata = restore_metadata_id(metadata, &envelopes);
261        let timing = StreamCassetteTiming::from_expr(field(entries, "timing")?)?;
262        let diagnostics = symbol_list(entries, "diagnostics")?;
263        let final_stats = stream_stats_from_expr(field(entries, "final-stats")?)?;
264        Ok(Self {
265            metadata,
266            envelopes,
267            timing,
268            diagnostics,
269            final_stats,
270        })
271    }
272
273    /// Returns a copy with host-device names and private payloads redacted.
274    ///
275    /// Redacts the metadata, every envelope, and the diagnostic symbols so the
276    /// result is safe to persist as a golden fixture.
277    pub fn redacted(&self) -> Result<Self> {
278        let metadata = redact_metadata(&self.metadata);
279        let envelopes = self
280            .envelopes
281            .iter()
282            .map(redact_envelope)
283            .collect::<Result<Vec<_>>>()?;
284        let mut redacted = Self::from_envelopes(metadata, envelopes, self.final_stats.clone())?;
285        redacted.diagnostics = self.diagnostics.iter().map(redact_symbol).collect();
286        Ok(redacted)
287    }
288
289    /// Validates the cassette as a golden fixture at `path`.
290    ///
291    /// Checks the path lives under [`STREAM_CASSETTE_FIXTURE_ROOT`] with the
292    /// cassette extension, that the trace is finite, that envelope sequences
293    /// match their packet index, that each transport profile is replayable or
294    /// previewable but never realtime, and that no unredacted payload or
295    /// host-device name remains. Returns a [`StreamGoldenFixtureReport`] on
296    /// success, or an error describing the first failed invariant.
297    pub fn validate_golden_fixture(&self, path: &str) -> Result<StreamGoldenFixtureReport> {
298        validate_fixture_path(path)?;
299        if !self.timing.finite {
300            return Err(Error::Eval(
301                "golden stream fixture must be finite".to_owned(),
302            ));
303        }
304        for (index, envelope) in self.envelopes.iter().enumerate() {
305            if envelope.sequence() != index as u64 {
306                return Err(Error::Eval(format!(
307                    "golden stream fixture sequence {} is not packet index {index}",
308                    envelope.sequence()
309                )));
310            }
311            if !envelope
312                .profile()
313                .has_capability(StreamCapability::Replayable)
314                && !envelope.profile().has_capability(StreamCapability::Preview)
315            {
316                return Err(Error::Eval(format!(
317                    "golden stream fixture profile {} is not replayable or previewable",
318                    envelope.profile().name()
319                )));
320            }
321            if envelope
322                .profile()
323                .has_capability(StreamCapability::Realtime)
324            {
325                return Err(Error::Eval(
326                    "golden stream fixture cannot require realtime transport".to_owned(),
327                ));
328            }
329            if packet_has_private_payload(envelope.packet()) || envelope_has_host_device(envelope) {
330                return Err(Error::Eval(
331                    "golden stream fixture contains unredacted payload".to_owned(),
332                ));
333            }
334        }
335        if metadata_has_host_device(&self.metadata)
336            || is_host_device_symbol(&self.timing.clock)
337            || self.diagnostics.iter().any(is_host_device_symbol)
338        {
339            return Err(Error::Eval(
340                "golden stream fixture contains an unredacted host device name".to_owned(),
341            ));
342        }
343        Ok(StreamGoldenFixtureReport {
344            path: path.to_owned(),
345            format: stream_cassette_format_symbol(),
346            packet_count: self.envelopes.len(),
347            final_stats: self.final_stats.clone(),
348        })
349    }
350}
351
352impl StreamCassetteTiming {
353    /// Serializes the timing summary to an [`Expr`] map keyed by field symbol.
354    pub fn to_expr(&self) -> Expr {
355        Expr::Map(vec![
356            (
357                Expr::Symbol(Symbol::new("clock")),
358                Expr::Symbol(self.clock.clone()),
359            ),
360            (
361                Expr::Symbol(Symbol::new("packet-count")),
362                Expr::String(self.packet_count.to_string()),
363            ),
364            (
365                Expr::Symbol(Symbol::new("first-sequence")),
366                optional_u64_expr(self.first_sequence),
367            ),
368            (
369                Expr::Symbol(Symbol::new("last-sequence")),
370                optional_u64_expr(self.last_sequence),
371            ),
372            (Expr::Symbol(Symbol::new("finite")), Expr::Bool(self.finite)),
373        ])
374    }
375
376    /// Deserializes a timing summary from an [`Expr`] map produced by
377    /// [`to_expr`](StreamCassetteTiming::to_expr).
378    ///
379    /// Validates the field set and fails closed on missing or unexpected
380    /// fields or type mismatches.
381    pub fn from_expr(expr: &Expr) -> Result<Self> {
382        let Expr::Map(entries) = expr else {
383            return Err(Error::TypeMismatch {
384                expected: "stream cassette timing map",
385                found: expr_kind(expr),
386            });
387        };
388        ensure_fields(
389            entries,
390            &[
391                "clock",
392                "packet-count",
393                "first-sequence",
394                "last-sequence",
395                "finite",
396            ],
397        )?;
398        Ok(Self {
399            clock: symbol_field(entries, "clock")?.clone(),
400            packet_count: parse_usize(entries, "packet-count")?,
401            first_sequence: optional_u64(field(entries, "first-sequence")?)?,
402            last_sequence: optional_u64(field(entries, "last-sequence")?)?,
403            finite: bool_field(entries, "finite")?,
404        })
405    }
406}
407
408/// Returns the format symbol stamped into every serialized cassette.
409pub fn stream_cassette_format_symbol() -> Symbol {
410    Symbol::qualified("stream/cassette", "v1")
411}
412
413/// Returns the repository-relative root for golden stream fixtures.
414pub fn stream_cassette_golden_root() -> &'static str {
415    STREAM_CASSETTE_FIXTURE_ROOT
416}
417
418/// Returns the file extension (without leading dot) for golden fixtures.
419pub fn stream_cassette_golden_extension() -> &'static str {
420    STREAM_CASSETTE_EXTENSION
421}
422
423fn timing_from_envelopes(
424    metadata: &StreamMetadata,
425    envelopes: &[StreamEnvelope],
426) -> StreamCassetteTiming {
427    StreamCassetteTiming {
428        clock: metadata.clock().clone(),
429        packet_count: envelopes.len(),
430        first_sequence: envelopes.first().map(StreamEnvelope::sequence),
431        last_sequence: envelopes.last().map(StreamEnvelope::sequence),
432        finite: true,
433    }
434}
435
436fn restore_metadata_id(metadata: StreamMetadata, envelopes: &[StreamEnvelope]) -> StreamMetadata {
437    let Some(first) = envelopes.first() else {
438        return metadata;
439    };
440    if metadata.id().as_qualified_str() != first.stream_id().as_qualified_str() {
441        return metadata;
442    }
443    StreamMetadata::new(
444        first.stream_id().clone(),
445        metadata.media(),
446        metadata.direction(),
447        metadata.clock().clone(),
448        metadata.buffer().clone(),
449    )
450}
451
452fn diagnostics_from_envelopes(envelopes: &[StreamEnvelope]) -> Vec<Symbol> {
453    let mut diagnostics = Vec::new();
454    for envelope in envelopes {
455        for diagnostic in envelope.diagnostics() {
456            push_unique(&mut diagnostics, diagnostic.clone());
457        }
458        if let StreamPacket::Diagnostic(packet) = envelope.packet() {
459            push_unique(&mut diagnostics, packet.kind().clone());
460        }
461    }
462    diagnostics
463}
464
465fn push_unique(symbols: &mut Vec<Symbol>, symbol: Symbol) {
466    if !symbols.contains(&symbol) {
467        symbols.push(symbol);
468    }
469}
470
471fn validate_fixture_path(path: &str) -> Result<()> {
472    let Some(relative) = path.strip_prefix(STREAM_CASSETTE_FIXTURE_ROOT) else {
473        return Err(Error::Eval(format!(
474            "golden stream fixture path must live under {STREAM_CASSETTE_FIXTURE_ROOT}"
475        )));
476    };
477    if !relative.starts_with('/') || relative == "/" {
478        return Err(Error::Eval(format!(
479            "golden stream fixture path must live under {STREAM_CASSETTE_FIXTURE_ROOT}"
480        )));
481    }
482    let expected_extension = format!(".{STREAM_CASSETTE_EXTENSION}");
483    if !path.ends_with(&expected_extension) {
484        return Err(Error::Eval(format!(
485            "golden stream fixture path must end in .{STREAM_CASSETTE_EXTENSION}"
486        )));
487    }
488    Ok(())
489}
490
491fn ensure_fields(entries: &[(Expr, Expr)], allowed: &[&str]) -> Result<()> {
492    for (key, _) in entries {
493        let Expr::Symbol(symbol) = key else {
494            return Err(Error::TypeMismatch {
495                expected: "symbol stream cassette field",
496                found: expr_kind(key),
497            });
498        };
499        if symbol.namespace.is_none() && allowed.contains(&symbol.name.as_ref()) {
500            continue;
501        }
502        return Err(Error::Eval(format!(
503            "unknown stream cassette field {}",
504            symbol.as_qualified_str()
505        )));
506    }
507    Ok(())
508}
509
510fn list_field<'a>(entries: &'a [(Expr, Expr)], name: &str) -> Result<&'a [Expr]> {
511    access::entry_required_list(entries, name, "list field")
512}
513
514fn symbol_list(entries: &[(Expr, Expr)], name: &str) -> Result<Vec<Symbol>> {
515    list_field(entries, name)?
516        .iter()
517        .map(|expr| match expr {
518            Expr::Symbol(symbol) => Ok(symbol.clone()),
519            other => Err(Error::TypeMismatch {
520                expected: "symbol list item",
521                found: expr_kind(other),
522            }),
523        })
524        .collect()
525}
526
527fn parse_usize(entries: &[(Expr, Expr)], name: &str) -> Result<usize> {
528    string_field(entries, name)?
529        .parse::<usize>()
530        .map_err(|err| Error::Eval(format!("invalid stream cassette {name}: {err}")))
531}
532
533fn optional_u64(expr: &Expr) -> Result<Option<u64>> {
534    match expr {
535        Expr::Nil => Ok(None),
536        Expr::String(value) => value
537            .parse::<u64>()
538            .map(Some)
539            .map_err(|err| Error::Eval(format!("invalid stream cassette sequence: {err}"))),
540        other => Err(Error::TypeMismatch {
541            expected: "optional u64 string",
542            found: expr_kind(other),
543        }),
544    }
545}
546
547fn optional_u64_expr(value: Option<u64>) -> Expr {
548    value
549        .map(|value| Expr::String(value.to_string()))
550        .unwrap_or(Expr::Nil)
551}
552
553fn bool_field(entries: &[(Expr, Expr)], name: &str) -> Result<bool> {
554    access::entry_required_bool(entries, name, "bool field")
555}
sim_lib_stream_core/cassette.rs

sim_lib_stream_core/
cassette.rs
