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

sim_lib_stream_core/
cassette.rs
