Merge feat/sf-media-01: native view and audio observations, the shared frame, and the spectator path

docs/design/session-framework/bus-conformance.md

@@ -119,7 +119,7 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means
 | Requirement | Status | Code | Test |
 | --- | --- | --- | --- |
 | `call-<U64>` with increasing serials per connected client; reused or retired ids are rejected, never executed again | conforms: a syntactically valid id advances the watermark even when admission is refused | `router/state.rs::op_call` (`call_watermark`) | `tests/sol_review_races.rs::rejected_call_id_still_advances_monotonic_watermark`, `tests/rpc.rs::raw_call_ids_and_forged_replies` (both) |
-| Reconnecting creates a new incarnation rather than reviving old calls | conforms | `router/state.rs::{hello, disconnect}` | `tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` |
+| Reconnecting creates a new incarnation rather than reviving old calls | conforms | `router/state.rs::{hello, disconnect}` | `tests/conformance_routing.rs::unpinned_call_after_incarnation_replacement_reaches_the_new_holder` (both), `tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` for the old calls half (its synchronisation was fixed on 2026-09-22; see "A flaky test and what it was measuring") |
 | An RPC targets one registered service, not a broadcast subject | conforms | `router/state.rs::op_call` | `tests/rpc.rs::request_reply_roundtrip` (both) |
 | First-dispatch FIFO per caller and service; responses may complete out of order and correlate by callId | conforms | `router/state.rs::{Svc::queue, dispatch_rpc}` | `tests/rpc.rs::fifo_dispatch_and_out_of_order_completion` (both), `tests/conformance_routing.rs::out_of_order_replies_correlate_across_concurrent_callers` (both) |
 | A service dispatcher can answer status concurrently with a long mutation | conforms | `router/state.rs::dispatch_rpc` (in-flight credits, not one-at-a-time) | `tests/bus_acceptance.rs::a_status_rpc_responds_while_another_handler_is_delayed` (both) |
@@ -412,6 +412,45 @@ What the numbers do and do not say:
   cores against 0.33 to 0.46), because the clients own the copies. A thread that exits between
   two samples takes its CPU with it, so the router figure is a floor.
 
+## A flaky test and what it was measuring, 2026-09-22 (MEDIA-01)
+
+`tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption`
+failed intermittently on `main` after the bus slice merged. Reproduced here at
+**37 failures in 240 runs** (four parallel loops of 60, debug, on the loaded dev VM), always
+on the same line and always the same way: `responder.reply(...)` returned `routed:true` where
+the test asserted `false`.
+
+The mechanism is a synchronisation gap in the test, not a routing defect.
+`router/state.rs::op_reply` returns `routed:false` only when `call.detached` is set, and for a
+disconnected caller that flag is set by `router/state.rs::disconnect`, which the router runs
+when **its** connection task reads EOF. `Client::close` documents what it waits for — "flushes
+queued releases, closes the connection and waits until the reader has stopped ... the router
+releases what they owned" — which is the client side only. So after `close()` returns, the
+router may not have torn the caller's connection down yet, and a reply that reaches it first is
+routed to a connection that is already closing. Nothing escapes: `disconnect` then releases
+that connection's roots along with the queued result, which is why the test's own later
+`settle` calls always passed. Only the `routed` flag, read one step too early, was wrong.
+
+The fix is in the test: it now waits for the teardown it is talking about
+(`e.settle("caller-a disconnected", |s| s.connections == 1)`) before asserting the
+reply-to-a-detached-call sentence of section 6. That is the same bounded
+poll-until-the-router-settles the rest of the file already uses for router-side consequences;
+no sleep, no timing constant, and the assertion now has the precondition its contract sentence
+names. **360 runs after the fix, 0 failures** (240 debug, 120 release).
+
+Two other intermittent failures were seen in the same sweep and are **not** fixed here, since
+they belong to the bus slice rather than to this one:
+
+- `tests/example_demo.rs::the_example_shows_a_counter_rpc_an_observer_and_a_held_frame`,
+  2 failures in 40 standalone runs plus 1 in 12 full-suite runs. It prints
+  "while the frame is held: 1 artifact(s), 2 root(s)" instead of 1 root: the producer's hold
+  release is queued on the control lane and had not been applied when the example read the
+  counts. The same shape of gap, in the guide deliverable's printed output.
+- `tests/integration.rs::unix_socket::session_over_one_router`, 1 failure in 12 full-suite
+  runs and 0 in 40 standalone runs, at the assertion that the deliberately slow consumer
+  skipped snapshots. Under load it kept up, so the assertion is a timing claim about the
+  machine.
+
 ## Contradictions
 
 Two, both inside bus-v1, both minor, neither resolved by changing code. Both were referred to

docs/design/session-framework/state-media-v1.md

@@ -65,6 +65,23 @@ the sample position relative to the episode's configured audio origin, with inte
 firstSample/sampleRate. Crash restore preserves sample position under a new epoch; first
 chunk marks discontinuity. Within an epoch, chunks cannot overlap or go backwards.
 
+**Amendment, 2026-09-22 (MEDIA-01).** Two readings of the paragraphs above, made explicit
+because they are now enforced:
+
+- The bootstrap window is exactly the boundaries where `max(0, boundary - observationDelaySteps)`
+  is zero, that is `boundary <= observationDelaySteps`. Inside it the repeated `O[0]` is the
+  **same artifact**, not a fresh render of the same scene; outside it the producing boundary
+  advances one per step, and a frame from any other boundary -- older or newer -- is a step
+  failure. A producer therefore keeps a queue of `observationDelaySteps + 1` frames and nothing
+  more, so there is no older frame available to substitute.
+- Within an epoch, `discontinuity` marks a range the stream actually skipped. The first chunk
+  after a restore marks it, and a later chunk may mark it when it starts past where the previous
+  chunk ended; a chunk that continues the previous one exactly is continuous by construction and
+  its flag is refused. Without that reading the restore rule is advisory, because a stream could
+  set the flag on every chunk and satisfy it by accident. The requirement is one-directional: a
+  fresh epoch's first chunk **may** mark a discontinuity, because section 6's recovery
+  establishes a fresh timeline and publishes one.
+
 The environment provides **native game output**. Sensor transformations belong to the agent
 profile. Resizing for viewers, overlays, composition, audio mixing/resampling, encoding,
 browser delivery and streaming belong to the application/presentation layer. No bus or

services/flysim/crates/fly-session-types/src/media.rs

@@ -366,6 +366,161 @@ impl DomainType for AudioRef {
     }
 }
 
+// ---------------------------------------------------------------------------------------------
+// Epoch audio sequencing (state-media-v1 section 2)
+
+/// One audio stream's chunk sequence within one epoch.
+///
+/// The contract's three sentences about sequencing are all here: `firstSample` identifies the
+/// sample position relative to the episode's configured audio origin; crash restore preserves
+/// that position under a new epoch and the first chunk marks `discontinuity`; within an epoch
+/// chunks cannot overlap or go backwards.
+///
+/// A timeline belongs to one epoch. A restore starts a new one with
+/// [`AudioTimeline::restored_at`], which is what makes the first chunk's discontinuity flag
+/// checkable rather than advisory.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AudioTimeline {
+    stream_id: String,
+    start_sample: u64,
+    restored: bool,
+    next_sample: u64,
+    accepted: u64,
+}
+
+impl AudioTimeline {
+    /// A fresh episode: the first chunk starts at the configured audio origin. Its
+    /// discontinuity flag is free, because a reset or a recovery establishes a fresh timeline
+    /// and does publish a discontinuity.
+    pub fn fresh(descriptor: &AudioDescriptor, origin: u64) -> AudioTimeline {
+        AudioTimeline {
+            stream_id: descriptor.stream_id.clone(),
+            start_sample: origin,
+            restored: false,
+            next_sample: origin,
+            accepted: 0,
+        }
+    }
+
+    /// A new epoch after a crash restore: the sample position is preserved, and the first
+    /// chunk of this epoch must mark `discontinuity`.
+    pub fn restored_at(descriptor: &AudioDescriptor, sample: u64) -> AudioTimeline {
+        AudioTimeline {
+            stream_id: descriptor.stream_id.clone(),
+            start_sample: sample,
+            restored: true,
+            next_sample: sample,
+            accepted: 0,
+        }
+    }
+
+    /// Where the next chunk may start. A chunk starting earlier overlaps or goes backwards.
+    pub fn next_sample(&self) -> u64 {
+        self.next_sample
+    }
+
+    /// How many chunks this epoch has accepted.
+    pub fn accepted(&self) -> u64 {
+        self.accepted
+    }
+
+    /// Validates one chunk's shape and its place in the sequence, then advances the timeline.
+    ///
+    /// A rejected chunk does not advance anything, so a caller that fails its step does not
+    /// leave the timeline believing the chunk was played.
+    pub fn accept(&mut self, chunk: &AudioRef, descriptor: &AudioDescriptor) -> Result<()> {
+        if chunk.stream_id != self.stream_id {
+            return err(format!(
+                "AudioTimeline {}: chunk names stream {:?}",
+                self.stream_id, chunk.stream_id
+            ));
+        }
+        chunk.validate_against(descriptor)?;
+        if self.accepted == 0 {
+            if chunk.first_sample != self.start_sample {
+                return err(format!(
+                    "AudioTimeline {}: the first chunk of this epoch must start at sample {}, not {}",
+                    self.stream_id, self.start_sample, chunk.first_sample
+                ));
+            }
+            // Only one direction is stated: the first chunk after a restore marks the
+            // discontinuity. A fresh epoch's first chunk may mark one too -- sections 6 and 7
+            // have recovery and episode reset publishing a discontinuity on a fresh timeline --
+            // so the flag is required after a restore and left free at an origin.
+            if self.restored && !chunk.discontinuity {
+                return err(format!(
+                    "AudioTimeline {}: the first chunk after a restore marks discontinuity",
+                    self.stream_id
+                ));
+            }
+        } else {
+            if chunk.first_sample < self.next_sample {
+                return err(format!(
+                    "AudioTimeline {}: firstSample {} overlaps or goes backwards; the previous chunk ends at {}",
+                    self.stream_id, chunk.first_sample, self.next_sample
+                ));
+            }
+            // Derived from the sentence above: within an epoch the only discontinuity a chunk
+            // can carry is a gap it actually skipped. A chunk that continues the previous one
+            // exactly is continuous by construction.
+            if chunk.discontinuity && chunk.first_sample == self.next_sample {
+                return err(format!(
+                    "AudioTimeline {}: a chunk continuing the previous one is not a discontinuity",
+                    self.stream_id
+                ));
+            }
+        }
+        self.next_sample = chunk
+            .first_sample
+            .checked_add(chunk.sample_frames)
+            .ok_or_else(|| {
+                crate::scalar::wire_err(format!(
+                    "AudioTimeline {}: firstSample + sampleFrames overflows U64",
+                    self.stream_id
+                ))
+            })?;
+        self.accepted += 1;
+        Ok(())
+    }
+}
+
+// ---------------------------------------------------------------------------------------------
+// Persistent assets against transient artifacts (state-media-v1 sections 1 and 3)
+
+/// Checks that a transient artifact carries the bytes of an installed asset.
+///
+/// Persistent [`AssetRef`](crate::workers::AssetRef) and transient
+/// [`ArtifactRef`] are different identities and never convert into one another: an asset names
+/// installed release content in a preprovisioned registry, while an artifact names live bytes
+/// in one store incarnation and resolves only through an owned handle. Importing an asset
+/// produces a **new** artifact identity, which is why this checks content rather than identity.
+///
+/// The digest is mandatory here: state-media-v1 section 1 makes content digests optional on
+/// transient live frames and mandatory on persistent asset import.
+pub fn check_imported_asset(
+    asset: &crate::workers::AssetRef,
+    imported: &ArtifactRef,
+) -> Result<()> {
+    asset.validate()?;
+    if imported.byte_length != asset.byte_length {
+        return err(format!(
+            "imported asset {}: the artifact is {} bytes, the asset is {}",
+            asset.id, imported.byte_length, asset.byte_length
+        ));
+    }
+    match &imported.digest {
+        Some(d) if *d == asset.digest => Ok(()),
+        Some(_) => err(format!(
+            "imported asset {}: the artifact's digest is not the asset's content",
+            asset.id
+        )),
+        None => err(format!(
+            "imported asset {}: a persistent asset import must carry a content digest",
+            asset.id
+        )),
+    }
+}
+
 /// Reads a bounded, unique-by-`viewId` list of view refs.
 pub fn view_list(f: &mut Fields<'_>, key: &'static str) -> Result<Vec<ViewRef>> {
     let views = list(f, key, 0, MAX_VIEWS, ViewRef::from_json)?;

services/flysim/crates/fly-session-types/tests/media_shapes.rs

@@ -0,0 +1,395 @@
+//! MEDIA-01 shape rules: every sentence of `state-media-v1` section 2 that a descriptor, a
+//! reference or a chunk sequence can be checked against on its own.
+//!
+//! These are the contract-level halves of the slice's acceptance bullets -- bad strides, bad
+//! lengths, bad producing times and the audio rules -- and the type-level distinction between
+//! a persistent `AssetRef` and a transient `ArtifactRef`.
+
+use fly_session_types::ArtifactRef;
+use fly_session_types::media::{
+    AudioDescriptor, AudioRef, AudioTimeline, ViewDescriptor, ViewRef, check_imported_asset,
+    require_finite_samples,
+};
+use fly_session_types::scalar::DomainType;
+use fly_session_types::workers::AssetRef;
+
+fn artifact(byte_length: u64, content_type: &str) -> ArtifactRef {
+    ArtifactRef {
+        store_id: "store-a".into(),
+        artifact_id: "art-1".into(),
+        generation: 1,
+        byte_length,
+        content_type: content_type.into(),
+        digest: None,
+    }
+}
+
+fn view_descriptor(width: u64, height: u64, delay: u64) -> ViewDescriptor {
+    ViewDescriptor {
+        view_id: "arena".into(),
+        width,
+        height,
+        row_stride: width * 4,
+        pixel_aspect_numerator: 1,
+        pixel_aspect_denominator: 1,
+        observation_delay_steps: delay,
+    }
+}
+
+fn view_ref(descriptor: &ViewDescriptor, produced_step: u64, bytes: u64) -> ViewRef {
+    ViewRef {
+        view_id: descriptor.view_id.clone(),
+        produced_step,
+        pixels: artifact(bytes, "image/x-rgba8"),
+    }
+}
+
+fn audio_descriptor(sample_rate: u64, channels: u64) -> AudioDescriptor {
+    AudioDescriptor {
+        stream_id: "arena".into(),
+        sample_rate,
+        channels,
+    }
+}
+
+fn audio_ref(
+    descriptor: &AudioDescriptor,
+    first_sample: u64,
+    frames: u64,
+    discontinuity: bool,
+) -> AudioRef {
+    AudioRef {
+        stream_id: descriptor.stream_id.clone(),
+        first_sample,
+        sample_frames: frames,
+        samples: artifact(frames * descriptor.channels * 4, "audio/x-f32le"),
+        discontinuity,
+    }
+}
+
+// -----------------------------------------------------------------------------------------
+// Bad strides
+
+/// `rowStride` is exactly `4 x width`; v1 has no padded rows.
+#[test]
+fn a_padded_row_stride_is_refused() {
+    let good = view_descriptor(32, 24, 0);
+    good.validate().expect("4 x width is the only stride");
+
+    let mut padded = good.clone();
+    padded.row_stride = 32 * 4 + 16;
+    padded.validate().expect_err("a padded row is not readable in v1");
+
+    let mut narrow = good.clone();
+    narrow.row_stride = 32 * 3;
+    narrow.validate().expect_err("a stride under 4 x width is refused");
+
+    // The same rule through the wire form, where a hand-written descriptor arrives.
+    let mut json = good.to_json();
+    json["rowStride"] = serde_json::json!(32 * 4 + 4);
+    ViewDescriptor::from_json(&json).expect_err("a padded stride is refused when read");
+}
+
+/// Dimensions are integers 1..=4096, and `pixelAspect` parts positive integers <=65535.
+#[test]
+fn dimensions_pixel_aspect_and_delay_have_stated_bounds() {
+    for (width, height) in [(0, 24), (32, 0), (4097, 24), (32, 4097)] {
+        let mut d = view_descriptor(32, 24, 0);
+        d.width = width;
+        d.height = height;
+        d.row_stride = width.max(1) * 4;
+        d.validate().expect_err("dimensions are 1..=4096");
+    }
+    view_descriptor(1, 1, 0).validate().expect("1x1 is inside the bounds");
+    view_descriptor(4096, 4096, 0)
+        .validate()
+        .expect("4096x4096 is inside the bounds");
+
+    for (numerator, denominator) in [(0, 1), (1, 0), (65_536, 1), (1, 65_536)] {
+        let mut d = view_descriptor(32, 24, 0);
+        d.pixel_aspect_numerator = numerator;
+        d.pixel_aspect_denominator = denominator;
+        d.validate().expect_err("pixelAspect parts are 1..=65535");
+    }
+
+    let mut d = view_descriptor(32, 24, 9);
+    d.validate().expect_err("observationDelaySteps is 0..=8");
+    d.observation_delay_steps = 8;
+    d.validate().expect("eight steps of delay are allowed");
+}
+
+/// Only top-left RGBA8 exists in v1; another format is a media-schema change.
+#[test]
+fn only_rgba8_is_a_readable_view_format() {
+    let descriptor = view_descriptor(32, 24, 0);
+    let mut json = descriptor.to_json();
+    json["format"] = serde_json::json!("rgb8");
+    ViewDescriptor::from_json(&json).expect_err("rgb8 is not a v1 format");
+    json["format"] = serde_json::json!("rgba8");
+    ViewDescriptor::from_json(&json).expect("rgba8 is the v1 format");
+}
+
+// -----------------------------------------------------------------------------------------
+// Bad lengths
+
+/// A frame's artifact is exactly `rowStride x height` bytes.
+#[test]
+fn a_frame_whose_length_is_not_stride_times_height_is_refused() {
+    let descriptor = view_descriptor(32, 24, 0);
+    let exact = descriptor.frame_bytes();
+    assert_eq!(exact, 32 * 4 * 24);
+
+    view_ref(&descriptor, 7, exact)
+        .validate_against(&descriptor, None)
+        .expect("the exact frame length is accepted");
+    for wrong in [exact - 1, exact + 1, exact - 32 * 4, exact * 2] {
+        view_ref(&descriptor, 7, wrong)
+            .validate_against(&descriptor, None)
+            .expect_err("only rowStride x height is the frame length");
+    }
+}
+
+// -----------------------------------------------------------------------------------------
+// Bad producing times
+
+/// A required sensory view is produced at exactly `max(0, boundary - observationDelaySteps)`.
+#[test]
+fn a_view_produced_at_the_wrong_boundary_is_refused() {
+    let descriptor = view_descriptor(32, 24, 2);
+    let bytes = descriptor.frame_bytes();
+    assert_eq!(descriptor.required_produced_step(10), 8);
+
+    view_ref(&descriptor, 8, bytes)
+        .validate_against(&descriptor, Some(10))
+        .expect("the declared delay is exactly two steps");
+    // One frame later than the delay allows, and one frame older: both are step failures,
+    // not an arbitrary latest frame.
+    view_ref(&descriptor, 9, bytes)
+        .validate_against(&descriptor, Some(10))
+        .expect_err("an under-delayed frame is refused");
+    view_ref(&descriptor, 7, bytes)
+        .validate_against(&descriptor, Some(10))
+        .expect_err("an extra-delayed frame is refused");
+}
+
+/// Bootstrap may repeat `O[0]` until the declared pipeline delay fills, and only until then.
+#[test]
+fn bootstrap_repeats_the_first_frame_until_the_pipeline_delay_fills() {
+    let descriptor = view_descriptor(32, 24, 3);
+    let bytes = descriptor.frame_bytes();
+    // Boundaries 0..=3 all require the frame produced at 0.
+    for boundary in 0..=3 {
+        assert_eq!(descriptor.required_produced_step(boundary), 0);
+        view_ref(&descriptor, 0, bytes)
+            .validate_against(&descriptor, Some(boundary))
+            .expect("O[0] repeats while the pipeline fills");
+    }
+    // From boundary 4 the pipeline is full and O[0] is a stale frame.
+    assert_eq!(descriptor.required_produced_step(4), 1);
+    view_ref(&descriptor, 0, bytes)
+        .validate_against(&descriptor, Some(4))
+        .expect_err("the repetition ends when the delay is filled");
+    view_ref(&descriptor, 1, bytes)
+        .validate_against(&descriptor, Some(4))
+        .expect("boundary 4 requires the frame produced at 1");
+}
+
+// -----------------------------------------------------------------------------------------
+// Audio shapes
+
+/// sampleRate is 8000..=192000, channels 1..=8 and sampleFrames 0..=192000.
+#[test]
+fn audio_rate_channels_and_frames_have_stated_bounds() {
+    audio_descriptor(8_000, 1).validate().expect("8 kHz mono is the floor");
+    audio_descriptor(192_000, 8).validate().expect("192 kHz 8ch is the ceiling");
+    audio_descriptor(7_999, 2).validate().expect_err("under 8 kHz is refused");
+    audio_descriptor(192_001, 2).validate().expect_err("over 192 kHz is refused");
+    audio_descriptor(48_000, 0).validate().expect_err("zero channels are refused");
+    audio_descriptor(48_000, 9).validate().expect_err("nine channels are refused");
+
+    let descriptor = audio_descriptor(48_000, 2);
+    let mut chunk = audio_ref(&descriptor, 0, 192_000, false);
+    chunk.validate().expect("192000 frames is the per-chunk ceiling");
+    chunk.sample_frames = 192_001;
+    chunk.validate().expect_err("over 192000 frames in one chunk is refused");
+}
+
+/// A chunk's artifact is exactly `sampleFrames x channels x 4` bytes.
+#[test]
+fn an_audio_chunk_length_is_frames_times_channels_times_four() {
+    let descriptor = audio_descriptor(48_000, 2);
+    let chunk = audio_ref(&descriptor, 0, 800, false);
+    assert_eq!(chunk.samples.byte_length, 800 * 2 * 4);
+    chunk
+        .validate_against(&descriptor)
+        .expect("the exact chunk length is accepted");
+
+    let mut wrong = chunk.clone();
+    wrong.samples = artifact(800 * 2 * 4 - 4, "audio/x-f32le");
+    wrong
+        .validate_against(&descriptor)
+        .expect_err("a short chunk is refused");
+
+    // The same frames at another channel count are a different number of bytes.
+    let mono = audio_descriptor(48_000, 1);
+    let mut wrong_channels = chunk.clone();
+    wrong_channels.stream_id = mono.stream_id.clone();
+    wrong_channels
+        .validate_against(&mono)
+        .expect_err("stereo bytes are not a mono chunk");
+}
+
+/// Samples are finite f32.
+#[test]
+fn a_non_finite_sample_is_refused() {
+    let mut bytes = Vec::new();
+    for value in [0.0f32, -0.5, 0.75, 1.0] {
+        bytes.extend_from_slice(&value.to_le_bytes());
+    }
+    require_finite_samples(&bytes).expect("finite samples are accepted");
+
+    for bad in [f32::NAN, f32::INFINITY, f32::NEG_INFINITY] {
+        let mut broken = bytes.clone();
+        broken.extend_from_slice(&bad.to_le_bytes());
+        require_finite_samples(&broken).expect_err("a non-finite sample is refused");
+    }
+    require_finite_samples(&bytes[..5]).expect_err("a partial sample is refused");
+}
+
+/// Within an epoch, chunks cannot overlap or go backwards.
+#[test]
+fn chunks_cannot_overlap_or_go_backwards_within_an_epoch() {
+    let descriptor = audio_descriptor(48_000, 2);
+    let mut timeline = AudioTimeline::fresh(&descriptor, 0);
+    timeline
+        .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor)
+        .expect("the first chunk starts at the origin");
+    assert_eq!(timeline.next_sample(), 800);
+    timeline
+        .accept(&audio_ref(&descriptor, 800, 800, false), &descriptor)
+        .expect("the second chunk continues the first");
+    assert_eq!(timeline.next_sample(), 1_600);
+
+    let mut overlapping = timeline.clone();
+    overlapping
+        .accept(&audio_ref(&descriptor, 1_599, 800, false), &descriptor)
+        .expect_err("a chunk that starts inside the previous one is refused");
+    let mut backwards = timeline.clone();
+    backwards
+        .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor)
+        .expect_err("a chunk that goes backwards is refused");
+    // A rejected chunk leaves the timeline where it was.
+    assert_eq!(overlapping.next_sample(), 1_600);
+    assert_eq!(overlapping.accepted(), 2);
+
+    // A gap is forward, so it is allowed; it is the one place a later chunk may mark a
+    // discontinuity.
+    timeline
+        .accept(&audio_ref(&descriptor, 2_000, 800, true), &descriptor)
+        .expect("a forward gap is not an overlap");
+    timeline
+        .accept(&audio_ref(&descriptor, 2_800, 800, true), &descriptor)
+        .expect_err("a chunk that continues the previous one is not a discontinuity");
+}
+
+/// Crash restore preserves the sample position under a new epoch, and its first chunk marks
+/// the discontinuity.
+#[test]
+fn the_first_chunk_after_a_restore_marks_discontinuity() {
+    let descriptor = audio_descriptor(48_000, 2);
+    // The requirement is one-directional. A fresh epoch's first chunk may mark a
+    // discontinuity -- a recovery or an episode reset establishes a fresh timeline and
+    // publishes one -- so both flags are accepted at an origin.
+    let mut reset = AudioTimeline::fresh(&descriptor, 0);
+    reset
+        .accept(&audio_ref(&descriptor, 0, 800, true), &descriptor)
+        .expect("a reset episode's first chunk may mark the discontinuity it published");
+    let mut fresh = AudioTimeline::fresh(&descriptor, 0);
+    fresh
+        .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor)
+        .expect("the episode's first chunk continues nothing");
+
+    // The restored epoch resumes at the preserved position.
+    let mut restored = AudioTimeline::restored_at(&descriptor, 800);
+    restored
+        .accept(&audio_ref(&descriptor, 800, 800, false), &descriptor)
+        .expect_err("the first chunk after a restore marks discontinuity");
+    let mut restored = AudioTimeline::restored_at(&descriptor, 800);
+    restored
+        .accept(&audio_ref(&descriptor, 0, 800, true), &descriptor)
+        .expect_err("the restored position is preserved, not reset");
+    restored
+        .accept(&audio_ref(&descriptor, 800, 800, true), &descriptor)
+        .expect("the restored epoch resumes at its preserved sample position");
+    assert_eq!(restored.next_sample(), 1_600);
+    restored
+        .accept(&audio_ref(&descriptor, 1_600, 800, false), &descriptor)
+        .expect("the chunks after it are ordinary");
+}
+
+// -----------------------------------------------------------------------------------------
+// Persistent assets against transient artifacts
+
+/// An `AssetRef` and an `ArtifactRef` are different identities with different fields, and
+/// neither is readable as the other.
+#[test]
+fn an_asset_ref_is_not_a_transient_artifact_ref() {
+    let asset = AssetRef {
+        id: "counter-arena-backend".into(),
+        digest: "a".repeat(64),
+        byte_length: 24,
+        format: "fly-config-v1".into(),
+    };
+    let imported = ArtifactRef {
+        store_id: "store-a".into(),
+        artifact_id: "art-9".into(),
+        generation: 1,
+        byte_length: 24,
+        content_type: "application/octet-stream".into(),
+        digest: Some("a".repeat(64)),
+    };
+
+    // Identity fields do not overlap: the asset has no store and the artifact has no format.
+    let asset_keys: Vec<String> = asset
+        .to_json()
+        .as_object()
+        .expect("an object")
+        .keys()
+        .cloned()
+        .collect();
+    let artifact_keys: Vec<String> = imported
+        .to_json()
+        .as_object()
+        .expect("an object")
+        .keys()
+        .cloned()
+        .collect();
+    assert_eq!(asset_keys, vec!["id", "digest", "byteLength", "format"]);
+    assert!(artifact_keys.contains(&"storeId".to_owned()));
+    assert!(artifact_keys.contains(&"artifactId".to_owned()));
+    assert!(!artifact_keys.contains(&"format".to_owned()));
+    assert!(!asset_keys.contains(&"storeId".to_owned()));
+
+    // Neither reads as the other: a view's pixels are an artifact, never an asset.
+    ArtifactRef::from_json(&asset.to_json()).expect_err("an asset is not an artifact reference");
+    AssetRef::from_json(&imported.to_json()).expect_err("an artifact is not an asset reference");
+    let pixels_as_asset = serde_json::json!({
+        "viewId": "arena",
+        "producedStep": "0",
+        "pixels": asset.to_json(),
+    });
+    ViewRef::from_json(&pixels_as_asset).expect_err("a view's pixels cannot be an asset");
+
+    // Importing an asset is a content check, not an identity conversion.
+    check_imported_asset(&asset, &imported).expect("the import carries the asset's content");
+    let mut no_digest = imported.clone();
+    no_digest.digest = None;
+    check_imported_asset(&asset, &no_digest)
+        .expect_err("a persistent asset import must carry a content digest");
+    let mut other_content = imported.clone();
+    other_content.digest = Some("b".repeat(64));
+    check_imported_asset(&asset, &other_content).expect_err("another digest is another content");
+    let mut short = imported.clone();
+    short.byte_length = 23;
+    check_imported_asset(&asset, &short).expect_err("the import must be the asset's length");
+}

services/flysim/crates/fly-session/src/agent.rs

@@ -206,6 +206,11 @@ pub struct AgentConfig {
     /// The thread allocation the launcher started this worker within. `workers-v1` requires
     /// `Agent.Initialize`'s `workerThreads` to lie inside it.
     pub worker_threads: usize,
+    /// Records every view this agent read, so a test can see which artifact reached it.
+    ///
+    /// It is this process's log: an agent with a process of its own writes to its own copy,
+    /// which the supervisor cannot read. `SessionHarness::sensor_log` says so with `None`.
+    pub sensors: crate::media::SensorLog,
     pub faults: AgentFaults,
 }
 
@@ -299,6 +304,15 @@ impl FakeAgentWorker {
                     format!("view {} is the wrong length", view.view_id),
                 ));
             }
+            // What this agent read, from the bytes it read: the artifact it was given and the
+            // digest of its content.
+            self.config.sensors.record(crate::media::SensedView {
+                boundary: input.boundary,
+                view_id: view.view_id.clone(),
+                artifact_id: artifact.reference().artifact_id.clone(),
+                produced_step: view.produced_step,
+                digest: digest_of_bytes(&bytes),
+            });
             total += i64::from(bytes.first().copied().unwrap_or_default());
         }
         if let Some(structured) = &input.structured {

services/flysim/crates/fly-session/src/cli.rs

@@ -21,7 +21,9 @@ use std::process::ExitCode;
 
 use crate::agent::AgentFaults;
 use crate::environment::EnvironmentFaults;
-use crate::launcher::{AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, serve_one};
+use crate::launcher::{
+    AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, flags, serve_one,
+};
 use crate::types::*;
 
 const USAGE: &str = "\
@@ -69,9 +71,12 @@ pub fn main() -> ExitCode {
     let command = command.to_string_lossy().into_owned();
     let rest: Vec<String> = args.map(|a| a.to_string_lossy().into_owned()).collect();
     let result = match command.as_str() {
-        "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)),
+        "agent" => Options::parse(&rest, &[flags::COMMON, flags::AGENT_ONLY])
-        "measure" => Options::parse(&rest).and_then(|o| measure(&o)),
+            .and_then(|o| serve(&command, &o)),
-        "measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)),
+        "environment" => Options::parse(&rest, &[flags::COMMON, flags::ENVIRONMENT_ONLY])
+            .and_then(|o| serve(&command, &o)),
+        "measure" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure(&o)),
+        "measure-row" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure_row(&o)),
         "--help" | "-h" | "help" => {
             print!("{USAGE}");
             return ExitCode::SUCCESS;
@@ -92,13 +97,22 @@ pub fn main() -> ExitCode {
 struct Options(BTreeMap<String, String>);
 
 impl Options {
-    fn parse(args: &[String]) -> Result<Options, String> {
+    /// Reads the options of one command, refusing any flag that command does not have.
+    ///
+    /// `allowed` comes from [`crate::launcher::flags`], the same constants the launcher
+    /// writes the argv from. An unknown flag is an error naming it rather than a value that
+    /// is quietly ignored: a renamed option must fail the launch, not turn into a no-op that
+    /// no test notices.
+    fn parse(args: &[String], allowed: &[&[&str]]) -> Result<Options, String> {
         let mut out = BTreeMap::new();
         let mut iter = args.iter();
         while let Some(flag) = iter.next() {
             let Some(name) = flag.strip_prefix("--") else {
                 return Err(format!("expected an option, found {flag:?}"));
             };
+            if !allowed.iter().any(|set| set.contains(&name)) {
+                return Err(format!("unknown option --{name} for this command"));
+            }
             let value = iter
                 .next()
                 .ok_or_else(|| format!("option --{name} needs a value"))?;
@@ -158,43 +172,53 @@ impl Options {
 
 /// Serves one worker until `Worker.Shutdown`, then exits.
 fn serve(role: &str, options: &Options) -> Result<(), String> {
-    let socket = options.path("socket")?;
+    let socket = options.path(flags::SOCKET)?;
-    let store_root = options.path("store-root")?;
+    let store_root = options.path(flags::STORE_ROOT)?;
-    let client_id = options.required("client-id")?.to_owned();
+    let client_id = options.required(flags::CLIENT_ID)?.to_owned();
-    let service = options.required("service")?.to_owned();
+    let service = options.required(flags::SERVICE)?.to_owned();
-    let threads = options.usize("threads", 1)?;
+    let threads = options.usize(flags::THREADS, 1)?;
     if threads == 0 {
         return Err("--threads must be at least 1".to_owned());
     }
-    let session_id = options.id("session")?;
+    let session_id = options.id(flags::SESSION)?;
-    let incarnation_id = options.id("incarnation")?;
+    let incarnation_id = options.id(flags::INCARNATION)?;
     let what = match role {
         "agent" => Started::Agent(AgentLaunch {
             session_id,
-            agent_id: options.id("agent")?,
+            agent_id: options.id(flags::AGENT)?,
-            port_id: options.id("port")?,
+            port_id: options.id(flags::PORT)?,
             incarnation_id,
-            tick_duration: options.rational("tick-numerator", "tick-denominator")?,
+            tick_duration: options.rational(flags::TICK_NUMERATOR, flags::TICK_DENOMINATOR)?,
-            warmup_ticks: options.u64("warmup-ticks", 0)?,
+            warmup_ticks: options.u64(flags::WARMUP_TICKS, 0)?,
             worker_threads: threads,
+            // This process's own log. The supervisor reads what crosses the bus, not this.
+            sensors: crate::media::SensorLog::new(),
             faults: AgentFaults {
-                fail_commit_at_step: options.opt_u64("fail-commit-at-step")?,
+                fail_commit_at_step: options.opt_u64(flags::FAIL_COMMIT_AT_STEP)?,
-                prepare_delay_ms: options.u64("prepare-delay-ms", 0)?,
+                prepare_delay_ms: options.u64(flags::PREPARE_DELAY_MS, 0)?,
-                commit_delay_ms: options.u64("commit-delay-ms", 0)?,
+                commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?,
             },
             client_id: client_id.clone(),
             service: service.clone(),
         }),
         _ => Started::Environment(EnvironmentLaunch {
             session_id,
-            worker_id: options.id("worker")?,
+            worker_id: options.id(flags::WORKER)?,
             incarnation_id,
-            step_duration: options.rational("step-numerator", "step-denominator")?,
+            step_duration: options.rational(flags::STEP_NUMERATOR, flags::STEP_DENOMINATOR)?,
-            ports: parse_ports(options.required("ports")?)?,
+            ports: parse_ports(options.required(flags::PORTS)?)?,
             worker_threads: threads,
+            observation_delay_steps: options.u64(flags::OBSERVATION_DELAY_STEPS, 0)?,
+            renders: crate::media::RenderCounter::new(),
             faults: EnvironmentFaults {
-                advance_delay_ms: options.u64("advance-delay-ms", 0)?,
+                advance_delay_ms: options.u64(flags::ADVANCE_DELAY_MS, 0)?,
-                omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?,
+                omit_view_at_boundary: options.opt_u64(flags::OMIT_VIEW_AT_BOUNDARY)?,
+                stale_view_at_boundary: options.opt_u64(flags::STALE_VIEW_AT_BOUNDARY)?,
+                truncated_view_at_boundary: options
+                    .opt_u64(flags::TRUNCATED_VIEW_AT_BOUNDARY)?,
+                omit_audio_at_boundary: options.opt_u64(flags::OMIT_AUDIO_AT_BOUNDARY)?,
+                overlapping_audio_at_boundary: options
+                    .opt_u64(flags::OVERLAPPING_AUDIO_AT_BOUNDARY)?,
             },
             client_id: client_id.clone(),
             service: service.clone(),
@@ -280,3 +304,34 @@ fn measure_row(options: &Options) -> Result<(), String> {
     println!("{}", row.to_json());
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// An unknown flag is refused by name. A renamed option must fail the launch rather than
+    /// be accepted and ignored, which would turn a fault or a delay into a no-op.
+    #[test]
+    fn an_unknown_option_is_refused_by_name() {
+        let args: Vec<String> = ["--session", "demo", "--stale-view-at-boundry", "2"]
+            .iter()
+            .map(|s| (*s).to_owned())
+            .collect();
+        let error = Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY])
+            .expect_err("an unknown option is refused");
+        assert!(error.contains("--stale-view-at-boundry"), "{error}");
+    }
+
+    /// A flag that belongs to another command is refused too: an agent has no render delay.
+    #[test]
+    fn an_option_of_another_command_is_refused() {
+        let args: Vec<String> = ["--observation-delay-steps", "2"]
+            .iter()
+            .map(|s| (*s).to_owned())
+            .collect();
+        Options::parse(&args, &[flags::COMMON, flags::AGENT_ONLY])
+            .expect_err("the agent command has no render delay");
+        Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY])
+            .expect("the environment command does");
+    }
+}

services/flysim/crates/fly-session/src/coordinator.rs

@@ -15,6 +15,7 @@ use std::time::{Duration, Instant};
 use serde_json::{Map, Value, json};
 
 use crate::clock::Pacing;
+use crate::media::{self, AudioTimelines};
 use crate::metrics::Metrics;
 use crate::phase::{Phase, PhaseMachine};
 use crate::rpc::{self, DomainReply, Serials, WorkerRef};
@@ -260,6 +261,15 @@ pub struct Coordinator {
     views: BTreeMap<String, flybus::Artifact>,
     /// Holds on the boundary the world has just reached, before it is committed.
     pending_views: BTreeMap<String, flybus::Artifact>,
+    /// The same handles for this boundary's audio chunks. Audio is presentation data: it is
+    /// published and never attached to an agent's sensory input.
+    audio: BTreeMap<String, flybus::Artifact>,
+    pending_audio: BTreeMap<String, flybus::Artifact>,
+    /// One chunk sequence per declared audio stream, for this epoch.
+    timelines: AudioTimelines,
+    /// The attachment names this session's native media travels under. The composition
+    /// supplies them for Initialize; afterwards they come from the environment descriptor.
+    media_names: Vec<String>,
     serials: Serials,
     topics: Topics,
     pacing: Option<Pacing>,
@@ -325,6 +335,13 @@ impl Coordinator {
             observation: None,
             views: BTreeMap::new(),
             pending_views: BTreeMap::new(),
+            audio: BTreeMap::new(),
+            pending_audio: BTreeMap::new(),
+            timelines: AudioTimelines::default(),
+            media_names: vec![
+                media::view_attachment(crate::environment::VIEW_ID),
+                media::audio_attachment(crate::environment::AUDIO_STREAM_ID),
+            ],
             serials: Serials::default(),
             topics,
             pacing: None,
@@ -374,6 +391,21 @@ impl Coordinator {
         self.observation.as_ref()
     }
 
+    /// The media handles this committed boundary holds: the one the agents were given and the
+    /// one presentation was published. They are the same objects, named by attachment.
+    pub fn media_handles(&self) -> Vec<(String, ArtifactRef)> {
+        self.views
+            .iter()
+            .chain(self.audio.iter())
+            .map(|(name, artifact)| (name.clone(), artifact.reference().clone()))
+            .collect()
+    }
+
+    /// Where each declared audio stream's next chunk may start.
+    pub fn audio_positions(&self) -> BTreeMap<String, u64> {
+        self.timelines.positions()
+    }
+
     pub fn episode_request(&self) -> Option<&EpisodeRequest> {
         self.episode.as_ref()
     }
@@ -632,6 +664,7 @@ impl Coordinator {
         };
         let worker = self.environment.clone();
         let scope = self.scope(0);
+        let want = self.media_names.clone();
         let reply = self
             .call(
                 &worker,
@@ -639,7 +672,7 @@ impl Coordinator {
                 Some(scope),
                 object(params.to_json()),
                 &[],
-                &["view.arena".to_owned()],
+                &want,
             )
             .await?;
         let result: EnvironmentInitializeResult =
@@ -672,7 +705,17 @@ impl Coordinator {
                 "port-assignment",
             ));
         }
-        self.views = reply.artifacts;
+        let (views, audio) = media::split_attachments(reply.artifacts);
+        self.views = views;
+        self.audio = audio;
+        self.media_names = media::attachment_names(&result.descriptor);
+        self.timelines = AudioTimelines::fresh(&result.descriptor);
+        if let Err(e) = self.timelines.accept(&result.descriptor, &result.observation) {
+            return Err(self.fail_now(e, "observation-0"));
+        }
+        if let Err(e) = media::check_required_views(&result.descriptor, &result.observation) {
+            return Err(self.fail_now(e, "observation-0"));
+        }
         self.descriptor = Some(result.descriptor);
         self.observation = Some(result.observation);
         self.lifecycle_acks.push((worker, reply.request_id.clone()));
@@ -1461,10 +1504,19 @@ impl Coordinator {
             .sensory_views
             .iter()
             .filter_map(|view| {
-                let name = format!("view.{}", view.view_id);
+                let name = media::view_attachment(&view.view_id);
                 self.pending_views.get(&name).map(|a| (name, a.clone()))
             })
             .collect();
+        let new_audio: BTreeMap<String, flybus::Artifact> = step_result
+            .observation
+            .audio
+            .iter()
+            .filter_map(|chunk| {
+                let name = media::audio_attachment(&chunk.stream_id);
+                self.pending_audio.get(&name).map(|a| (name, a.clone()))
+            })
+            .collect();
         let commits = self
             .commit_all(k, &step_result.observation, &mut outcomes, &mut next_contexts, &new_views)
             .await?;
@@ -1480,7 +1532,9 @@ impl Coordinator {
         }
         // The previous boundary's handles are no longer needed; the new ones take over.
         self.views = new_views;
+        self.audio = new_audio;
         self.pending_views.clear();
+        self.pending_audio.clear();
         self.observation = Some(step_result.observation.clone());
         self.stats.advances += 1;
 
@@ -1790,7 +1844,7 @@ impl Coordinator {
         let worker = self.environment.clone();
         let request_id = self.serials.next(&worker.service);
         self.last_advance_request = Some(request_id.clone());
-        let want = vec!["view.arena".to_owned()];
+        let want = self.media_names.clone();
         self.audit.push(format!("advance:{k}"));
         self.blame(Some(worker.worker_id.clone()));
         let advance_deadline = self.deadlines.probe;
@@ -1912,7 +1966,9 @@ impl Coordinator {
             Err(e) => return Err(self.fail_now(e, "advance")),
         };
         self.blame(None);
-        self.pending_views = reply.artifacts;
+        let (pending_views, pending_audio) = media::split_attachments(reply.artifacts);
+        self.pending_views = pending_views;
+        self.pending_audio = pending_audio;
 
         if injected && self.injections.altered_advance_controls {
             // The same request id with a different body: a conflict, never a second world
@@ -1947,6 +2003,7 @@ impl Coordinator {
                 None => None,
             };
             self.pending_views.clear();
+            self.pending_audio.clear();
             let replay = self
                 .resolve(
                     &worker,
@@ -1967,7 +2024,9 @@ impl Coordinator {
                 code: None,
                 identical: first.is_some() && first == again,
             });
-            self.pending_views = replay.artifacts;
+            let (pending_views, pending_audio) = media::split_attachments(replay.artifacts);
+            self.pending_views = pending_views;
+            self.pending_audio = pending_audio;
         }
         Ok(result)
     }
@@ -2042,43 +2101,20 @@ impl Coordinator {
                 "step-result",
             ));
         }
-        // A missing required sensory input is never silently replaced by an older frame.
+        // A missing required sensory input is never silently replaced by an older frame,
-        // The contract's validator checks the views that are present against their
+        // and neither is one produced further back than the declared delay allows. The
-        // descriptors; requiring each declared view to be there at all is the coordinator's
+        // contract's validator checks the views that are present against their descriptors;
-        // Phase C check, so it is made here.
+        // requiring each declared view to be there at all is this Phase C check.
-        for view in &descriptor.views {
+        if let Err(e) = media::check_required_views(descriptor, &result.observation) {
-            let want = required_produced_step(view, result.observation.boundary);
+            return Err(self.fail_now(e, "step-result"));
-            let got = result
+        }
-                .observation
+        // Audio has no sensory role here, but its chunks still cannot overlap or go backwards
-                .sensory_views
+        // inside an epoch, and a stale one must not reach presentation as current.
-                .iter()
+        if let Err(e) = media::check_required_audio(descriptor, &result.observation) {
-                .find(|given| given.view_id == view.view_id);
+            return Err(self.fail_now(e, "step-result"));
-            match got {
+        }
-                Some(given) if given.produced_step == want => {}
+        if let Err(e) = self.timelines.accept(descriptor, &result.observation) {
-                Some(_) => {
+            return Err(self.fail_now(e, "step-result"));
-                    return Err(self.fail_now(
-                        DomainError::new(
-                            ErrorCode::BufferInvalid,
-                            format!(
-                                "view {} did not come from the boundary its declared delay requires",
-                                view.view_id
-                            ),
-                            MutationCertainty::Unknown,
-                        ),
-                        "step-result",
-                    ));
-                }
-                None => {
-                    return Err(self.fail_now(
-                        DomainError::new(
-                            ErrorCode::BufferInvalid,
-                            format!("required sensory view {} is missing", view.view_id),
-                            MutationCertainty::Unknown,
-                        ),
-                        "step-result",
-                    ));
-                }
-            }
         }
         if let Err(e) = result.observation.validate_against(descriptor) {
             return Err(self.fail_now(
@@ -2087,7 +2123,7 @@ impl Coordinator {
             ));
         }
         for view in &result.observation.sensory_views {
-            let name = format!("view.{}", view.view_id);
+            let name = media::view_attachment(&view.view_id);
             match self.pending_views.get(&name) {
                 Some(artifact) if artifact.reference() == &view.pixels => {}
                 _ => {
@@ -2102,6 +2138,25 @@ impl Coordinator {
                 }
             }
         }
+        for chunk in &result.observation.audio {
+            let name = media::audio_attachment(&chunk.stream_id);
+            match self.pending_audio.get(&name) {
+                Some(artifact) if artifact.reference() == &chunk.samples => {}
+                _ => {
+                    return Err(self.fail_now(
+                        DomainError::new(
+                            ErrorCode::BufferInvalid,
+                            format!(
+                                "audio chunk {} arrived without a live owned handle",
+                                chunk.stream_id
+                            ),
+                            MutationCertainty::Unknown,
+                        ),
+                        "step-result",
+                    ));
+                }
+            }
+        }
         Ok(())
     }
 
@@ -2528,11 +2583,13 @@ impl Coordinator {
             "progress": self.task.progress().to_json(),
             "media": json!({
                 "views": Value::Array(observation.broadcast_views.iter().map(DomainType::to_json).collect()),
-                "audio": [],
+                "audio": Value::Array(observation.audio.iter().map(DomainType::to_json).collect()),
             }),
             "eventIds": event_ids.iter().map(Id::as_str).collect::<Vec<_>>(),
         });
-        let attachments = self.view_attachments();
+        // The same owned handles the agents were given, published once for presentation.
+        let mut attachments = self.view_attachments();
+        attachments.extend(self.audio.iter().map(|(n, a)| (n.clone(), a.clone())));
         let topic = self.topics.snapshots.clone();
         self.publish(
             &topic,

services/flysim/crates/fly-session/src/environment.rs

@@ -4,19 +4,31 @@
 //! one interval, and returns boundary `k+1` with its world time advanced by `stepDuration`. It
 //! never advances while waiting for the next request, and it does not free-run during agent
 //! initialization.
+//!
+//! Its native output is real: one immutable RGBA8 frame per boundary through a
+//! [`ViewPipeline`](crate::media::ViewPipeline) that honours the declared
+//! `observationDelaySteps`, and one audio chunk per transition with an exact sample budget.
+//! Nothing here resizes, mixes, composites or encodes anything; that is the presentation
+//! layer's work.
 
 use std::collections::BTreeSet;
-use std::io::Write;
 
+use crate::media::{self, AudioSource, RenderCounter, ViewPipeline};
 use crate::task::{controller_schema_ref, inspection, inspection_schema};
 // `crate::types` is this crate's facade over the shared `fly-session-types` crate; the
 // glob keeps the contract's own names in sight instead of restating them.
 use crate::types::*;
 use crate::worker::{BoxFuture, HandlerCtx, HandlerReply, StatusCell, WorkerEndpoint};
 
-/// The arena's view: a 4x4 RGBA8 tile whose bytes carry the counter.
+/// The arena's one view: a small native RGBA8 image.
-pub const VIEW_WIDTH: u64 = 4;
+pub const VIEW_ID: &str = "arena";
-pub const VIEW_HEIGHT: u64 = 4;
+pub const VIEW_WIDTH: u64 = 32;
+pub const VIEW_HEIGHT: u64 = 24;
+
+/// The arena's one audio stream. 48 kHz stereo is a native rate, not a presentation choice.
+pub const AUDIO_STREAM_ID: &str = "arena";
+pub const SAMPLE_RATE: u64 = 48_000;
+pub const CHANNELS: u64 = 2;
 
 /// Deliberate faults a test can ask the environment for.
 #[derive(Clone, Debug, Default)]
@@ -26,6 +38,16 @@ pub struct EnvironmentFaults {
     /// Drop the required sensory view from the result at this boundary, so the coordinator
     /// meets a world that advanced with no usable sensory data.
     pub omit_view_at_boundary: Option<u64>,
+    /// Serve the previous boundary's frame at this boundary: an extra-delayed sensory input,
+    /// which is a step failure rather than an acceptable latest frame.
+    pub stale_view_at_boundary: Option<u64>,
+    /// Seal a frame one row short at this boundary, so its artifact length is not
+    /// `rowStride x height`.
+    pub truncated_view_at_boundary: Option<u64>,
+    /// Leave the audio chunk out of the result at this boundary.
+    pub omit_audio_at_boundary: Option<u64>,
+    /// Emit an audio chunk that starts before the previous chunk ended.
+    pub overlapping_audio_at_boundary: Option<u64>,
 }
 
 #[derive(Clone, Debug)]
@@ -38,6 +60,12 @@ pub struct EnvironmentConfig {
     pub ports: Vec<Id>,
     /// The thread allocation the launcher started this worker within.
     pub worker_threads: usize,
+    /// The view's declared render delay, in steps. Zero is same-boundary output.
+    pub observation_delay_steps: u64,
+    /// Counts frames actually rendered, so a test can prove one image was not rendered twice.
+    ///
+    /// It counts in this process only: a world with a process of its own counts there.
+    pub renders: RenderCounter,
     pub faults: EnvironmentFaults,
 }
 
@@ -54,6 +82,10 @@ pub struct CounterEnvironment {
     world_time: RationalNs,
     advances: u64,
     batches: BTreeSet<Id>,
+    pipeline: Option<ViewPipeline>,
+    audio: Option<AudioSource>,
+    /// The frame served at the previous boundary, kept only so a fault can serve it again.
+    previous_view: Option<(ViewRef, flybus::Artifact)>,
 }
 
 impl CounterEnvironment {
@@ -69,6 +101,9 @@ impl CounterEnvironment {
             world_time: RationalNs::ZERO,
             advances: 0,
             batches: BTreeSet::new(),
+            pipeline: None,
+            audio: None,
+            previous_view: None,
             config,
         }
     }
@@ -103,15 +138,25 @@ impl CounterEnvironment {
         }
     }
 
-    pub fn view_descriptor() -> ViewDescriptor {
+    /// The arena's native view, with the configured render delay.
+    pub fn view_descriptor(observation_delay_steps: u64) -> ViewDescriptor {
         ViewDescriptor {
-            view_id: id("arena"),
+            view_id: id(VIEW_ID),
             width: VIEW_WIDTH,
             height: VIEW_HEIGHT,
             row_stride: VIEW_WIDTH * 4,
             pixel_aspect_numerator: 1,
             pixel_aspect_denominator: 1,
-            observation_delay_steps: 0,
+            observation_delay_steps,
+        }
+    }
+
+    /// The arena's native audio stream.
+    pub fn audio_descriptor() -> AudioDescriptor {
+        AudioDescriptor {
+            stream_id: id(AUDIO_STREAM_ID),
+            sample_rate: SAMPLE_RATE,
+            channels: CHANNELS,
         }
     }
 
@@ -121,10 +166,11 @@ impl CounterEnvironment {
             content_digest: digest_of_bytes(b"counter-arena-content-v1"),
             configuration_digest: digest_of_bytes(
                 format!(
-                    "counter-arena-config-v1\nstep={}/{}\nports={}\n",
+                    "counter-arena-config-v1\nstep={}/{}\nports={}\ndelay={}\n",
                     self.config.step_duration.numerator,
                     self.config.step_duration.denominator,
-                    self.config.ports.len()
+                    self.config.ports.len(),
+                    self.config.observation_delay_steps
                 )
                 .as_bytes(),
             ),
@@ -139,8 +185,10 @@ impl CounterEnvironment {
                 })
                 .collect(),
             inspection_schema: inspection_schema(),
-            views: vec![CounterEnvironment::view_descriptor()],
+            views: vec![CounterEnvironment::view_descriptor(
-            audio: Vec::new(),
+                self.config.observation_delay_steps,
+            )],
+            audio: vec![CounterEnvironment::audio_descriptor()],
             recovery: Recovery::ExactCheckpoint,
             determinism: Determinism::FixedBuild,
         };
@@ -148,73 +196,76 @@ impl CounterEnvironment {
         Ok(descriptor)
     }
 
-    /// Seals one immutable native frame for the current counter and returns the handle.
+    /// Renders this boundary's native media and returns the observation with its owned
-    async fn render(
+    /// handles. The same immutable object serves the sensory and the broadcast view; nothing
-        &self,
+    /// is rendered twice and no second copy of the pixels exists.
-        ctx: &HandlerCtx<'_>,
-    ) -> DomainResult<(ViewRef, flybus::Artifact)> {
-        let descriptor = CounterEnvironment::view_descriptor();
-        let len = descriptor.byte_length();
-        let mut writer = ctx
-            .client
-            .artifacts()
-            .allocate(len, "image/x-rgba")
-            .await
-            .map_err(|e| {
-                DomainError::new(
-                    ErrorCode::BackendFailure,
-                    format!("frame allocation failed: {}", e.message),
-                    MutationCertainty::Applied,
-                )
-            })?;
-        // Every pixel carries the counter's low byte, so an agent reading the frame reads the
-        // world rather than a constant.
-        let byte = (self.counter & 0xff) as u8;
-        writer
-            .write_all(&vec![byte; len as usize])
-            .map_err(|e| {
-                DomainError::new(
-                    ErrorCode::BackendFailure,
-                    format!("frame write failed: {e}"),
-                    MutationCertainty::Applied,
-                )
-            })?;
-        let artifact = writer.seal().await.map_err(|e| {
-            DomainError::new(
-                ErrorCode::BackendFailure,
-                format!("frame seal failed: {}", e.message),
-                MutationCertainty::Applied,
-            )
-        })?;
-        let view = ViewRef {
-            view_id: descriptor.view_id.clone(),
-            produced_step: descriptor.required_produced_step(self.boundary),
-            pixels: artifact.reference().clone(),
-        };
-        Ok((view, artifact))
-    }
-
     async fn observation(
-        &self,
+        &mut self,
         ctx: &HandlerCtx<'_>,
     ) -> DomainResult<(WorldObservation, Vec<(String, flybus::Artifact)>)> {
-        let omit = self.config.faults.omit_view_at_boundary == Some(self.boundary);
+        let boundary = self.boundary;
-        let (views, attachments) = if omit {
+        let counter = self.counter;
-            (Vec::new(), Vec::new())
+        let pipeline = self
+            .pipeline
+            .as_mut()
+            .ok_or_else(|| DomainError::before(ErrorCode::InvalidPhase, "no view pipeline"))?;
+        if self.config.faults.truncated_view_at_boundary == Some(boundary) {
+            pipeline
+                .render_truncated(ctx.client, boundary, counter)
+                .await?;
         } else {
-            let (view, artifact) = self.render(ctx).await?;
+            pipeline.render(ctx.client, boundary, counter).await?;
-            let name = format!("view.{}", view.view_id);
+        }
-            (vec![view], vec![(name, artifact)])
+        let produced = pipeline.at(boundary);
-        };
+
+        let mut attachments = Vec::new();
+        let mut views = Vec::new();
+        if self.config.faults.omit_view_at_boundary == Some(boundary) {
+            // A world that advanced with no usable sensory data.
+        } else if self.config.faults.stale_view_at_boundary == Some(boundary) {
+            if let Some((view, artifact)) = self.previous_view.clone() {
+                attachments.push((media::view_attachment(&view.view_id), artifact));
+                views.push(view);
+            }
+        } else if let Some((view, artifact)) = produced.clone() {
+            attachments.push((media::view_attachment(&view.view_id), artifact));
+            views.push(view);
+        } else {
+            return Err(DomainError::new(
+                ErrorCode::BackendFailure,
+                "the view pipeline has no frame for this boundary",
+                MutationCertainty::Applied,
+            ));
+        }
+        self.previous_view = produced;
+
+        let mut audio = Vec::new();
+        if boundary > 0 && self.config.faults.omit_audio_at_boundary != Some(boundary) {
+            let step = self.config.step_duration;
+            let overlap = self.config.faults.overlapping_audio_at_boundary == Some(boundary);
+            let source = self
+                .audio
+                .as_mut()
+                .ok_or_else(|| DomainError::before(ErrorCode::InvalidPhase, "no audio source"))?;
+            let (mut chunk, artifact) = source.produce(ctx.client, &step, counter).await?;
+            if overlap {
+                // A chunk that starts inside the previous one: the timeline refuses it rather
+                // than playing the same samples twice.
+                chunk.first_sample = chunk.first_sample.saturating_sub(1);
+            }
+            attachments.push((media::audio_attachment(&chunk.stream_id), artifact));
+            audio.push(chunk);
+        }
+
         let observation = WorldObservation {
-            boundary: self.boundary,
+            boundary,
             world_time: self.world_time,
-            engine_frame: Some(self.boundary.to_string()),
+            engine_frame: Some(boundary.to_string()),
             sensory_views: views.clone(),
-            inspection: inspection(self.counter, self.boundary),
+            inspection: inspection(counter, boundary),
             // The same immutable object serves the broadcast view; nothing is rendered twice.
             broadcast_views: views,
-            audio: Vec::new(),
+            audio,
         };
         Ok((observation, attachments))
     }
@@ -270,6 +321,13 @@ impl CounterEnvironment {
         self.counter = 0;
         self.world_time = RationalNs::ZERO;
         self.batches.clear();
+        self.pipeline = Some(ViewPipeline::new(
+            CounterEnvironment::view_descriptor(self.config.observation_delay_steps),
+            self.config.renders.clone(),
+        ));
+        // A fresh episode starts at audio origin zero; a restore would resume the preserved
+        // sample position instead, and its first chunk would mark the discontinuity.
+        self.audio = Some(AudioSource::new(CounterEnvironment::audio_descriptor(), 0));
         self.descriptor = Some(descriptor.clone());
         // The world is stopped when O[0] goes out and cannot free-run while the brains boot.
         self.status.set_state(WorkerState::Ready);

services/flysim/crates/fly-session/src/harness.rs

@@ -22,6 +22,7 @@ use flybus::{Client, Grants, Pattern, Policy, Router, RouterConfig};
 use crate::agent::{AgentFaults, synthetic_profile};
 use crate::coordinator::{AgentSlot, Coordinator};
 use crate::environment::EnvironmentFaults;
+use crate::media::{RenderCounter, SensorLog};
 use crate::launcher::{
     AgentLaunch, EnvironmentLaunch, Launcher, ReapOutcome, SUPERVISOR_CLIENT, ThreadBudget,
 };
@@ -72,6 +73,8 @@ pub struct HarnessConfig {
     pub tick_ms: u64,
     pub warmup_ticks: u64,
     pub terminal: Terminal,
+    /// The view's declared render delay, in steps. Zero is same-boundary output.
+    pub observation_delay_steps: u64,
     pub environment_faults: EnvironmentFaults,
     /// Where each participant runs.
     pub mode: ExecutionMode,
@@ -98,6 +101,7 @@ impl Default for HarnessConfig {
             tick_ms: 1,
             warmup_ticks: 10,
             terminal: Terminal::Never,
+            observation_delay_steps: 0,
             environment_faults: EnvironmentFaults::default(),
             mode: ExecutionMode::InProcess,
             thread_budget: None,
@@ -160,6 +164,11 @@ pub struct SessionHarness {
     pub config: HarnessConfig,
     pub via: Via,
     pub mode: ExecutionMode,
+    /// The media instrumentation of the participants that live in this process. Both are
+    /// shared memory, so both are empty for a participant with a process of its own; the
+    /// accessors below return `None` there rather than zero.
+    renders: RenderCounter,
+    sensors: BTreeMap<Id, SensorLog>,
     /// The supervisor. It owns every participant's lifetime and thread allocation.
     pub launcher: Launcher,
     observers: Mutex<Vec<Client>>,
@@ -226,6 +235,12 @@ impl SessionHarness {
 
         let step_duration = hz(config.step_hz).expect("a positive cadence");
         let tick_duration = millis(config.tick_ms).expect("a positive tick");
+        let renders = RenderCounter::new();
+        let sensors: BTreeMap<Id, SensorLog> = config
+            .agents
+            .iter()
+            .map(|spec| (spec.agent_id.clone(), SensorLog::new()))
+            .collect();
 
         // The environment first: it owns the world and the descriptor.
         let environment = launcher
@@ -236,6 +251,8 @@ impl SessionHarness {
                 step_duration,
                 ports: config.agents.iter().map(|a| a.port_id.clone()).collect(),
                 worker_threads: config.environment_threads,
+                observation_delay_steps: config.observation_delay_steps,
+                renders: renders.clone(),
                 faults: config.environment_faults.clone(),
                 client_id: ENV_CLIENT.to_owned(),
                 service: ENV_SERVICE.to_owned(),
@@ -259,6 +276,7 @@ impl SessionHarness {
                     tick_duration,
                     warmup_ticks: config.warmup_ticks,
                     worker_threads: spec.worker_threads,
+                    sensors: sensors[&spec.agent_id].clone(),
                     faults: spec.faults.clone(),
                     client_id: agent_client(&spec.agent_id),
                     service: agent_service(&spec.agent_id),
@@ -304,6 +322,8 @@ impl SessionHarness {
             config,
             via,
             mode: launcher.mode(),
+            renders,
+            sensors,
             launcher,
             observers: Mutex::new(Vec::new()),
         })
@@ -366,6 +386,9 @@ impl SessionHarness {
                 tick_duration,
                 warmup_ticks: self.config.warmup_ticks,
                 worker_threads: spec.worker_threads,
+                // The same log: a replacement worker in this process keeps writing where its
+                // predecessor wrote, so a restore's sensory input is visible beside it.
+                sensors: self.sensors.get(agent_id).cloned().unwrap_or_default(),
                 faults: spec.faults.clone(),
                 client_id: format!("{}-r2", agent_client(agent_id)),
                 service: agent_service(agent_id),
@@ -390,6 +413,30 @@ impl SessionHarness {
         id(ENV_WORKER)
     }
 
+    /// What one agent read out of its sensory attachments, in order, when this process is
+    /// where that log lives.
+    ///
+    /// `None` means "not observable from here", not "nothing was read": an agent with a
+    /// process of its own records into its own copy. The media path itself crosses a process
+    /// boundary -- the frame is one artifact in the shared store, reached through owned
+    /// handles -- but this instrumentation does not, because it is shared memory.
+    pub fn sensor_log(&self, agent_id: &Id) -> Option<SensorLog> {
+        match self.mode {
+            ExecutionMode::Process => None,
+            _ => self.sensors.get(agent_id).cloned(),
+        }
+    }
+
+    /// How many native frames the environment rendered, when the world lives in this process.
+    ///
+    /// `None` for a world with a process of its own, for the same reason as above.
+    pub fn renders(&self) -> Option<u64> {
+        match self.mode {
+            ExecutionMode::Process => None,
+            _ => Some(self.renders.count()),
+        }
+    }
+
     /// The agent worker's progress counter, which is its fake model's mutation count, when
     /// this process is where that counter lives.
     ///

services/flysim/crates/fly-session/src/launcher.rs

@@ -227,6 +227,9 @@ pub struct AgentLaunch {
     pub warmup_ticks: u64,
     /// What the launcher asks the budget for.
     pub worker_threads: usize,
+    /// Where this agent records the views it reads. A participant with a process of its own
+    /// gets a fresh log in that process, which the supervisor cannot read.
+    pub sensors: crate::media::SensorLog,
     pub faults: AgentFaults,
     /// The configured client id. A replacement worker connects under its own.
     pub client_id: String,
@@ -242,6 +245,10 @@ pub struct EnvironmentLaunch {
     pub step_duration: RationalNs,
     pub ports: Vec<Id>,
     pub worker_threads: usize,
+    /// The view's declared render delay, in steps.
+    pub observation_delay_steps: u64,
+    /// Where this world counts the frames it renders, with the same process caveat.
+    pub renders: crate::media::RenderCounter,
     pub faults: EnvironmentFaults,
     pub client_id: String,
     pub service: String,
@@ -1197,6 +1204,97 @@ fn launch_error(worker_id: &Id, e: &flybus::BusError) -> DomainError {
     )
 }
 
+
+// -------------------------------------------------------------------------------------------
+// The command line between a launcher and a participant in its own process
+
+/// Every `--flag` a launched participant or a measurement child accepts, named once.
+///
+/// The launcher writes these and [`crate::cli`] reads them. Naming them in two places by
+/// convention would let a rename turn an option into a silent no-op -- a fault that never
+/// fires, a delay that is never applied -- so both sides use these constants and the parser
+/// refuses any flag outside the set for the command it is parsing.
+pub(crate) mod flags {
+    pub const SOCKET: &str = "socket";
+    pub const STORE_ROOT: &str = "store-root";
+    pub const CLIENT_ID: &str = "client-id";
+    pub const SERVICE: &str = "service";
+    pub const THREADS: &str = "threads";
+    pub const SESSION: &str = "session";
+    pub const INCARNATION: &str = "incarnation";
+
+    pub const AGENT: &str = "agent";
+    pub const PORT: &str = "port";
+    pub const TICK_NUMERATOR: &str = "tick-numerator";
+    pub const TICK_DENOMINATOR: &str = "tick-denominator";
+    pub const WARMUP_TICKS: &str = "warmup-ticks";
+    pub const PREPARE_DELAY_MS: &str = "prepare-delay-ms";
+    pub const COMMIT_DELAY_MS: &str = "commit-delay-ms";
+    pub const FAIL_COMMIT_AT_STEP: &str = "fail-commit-at-step";
+
+    pub const WORKER: &str = "worker";
+    pub const PORTS: &str = "ports";
+    pub const STEP_NUMERATOR: &str = "step-numerator";
+    pub const STEP_DENOMINATOR: &str = "step-denominator";
+    pub const ADVANCE_DELAY_MS: &str = "advance-delay-ms";
+    pub const OMIT_VIEW_AT_BOUNDARY: &str = "omit-view-at-boundary";
+    pub const OBSERVATION_DELAY_STEPS: &str = "observation-delay-steps";
+    pub const STALE_VIEW_AT_BOUNDARY: &str = "stale-view-at-boundary";
+    pub const TRUNCATED_VIEW_AT_BOUNDARY: &str = "truncated-view-at-boundary";
+    pub const OMIT_AUDIO_AT_BOUNDARY: &str = "omit-audio-at-boundary";
+    pub const OVERLAPPING_AUDIO_AT_BOUNDARY: &str = "overlapping-audio-at-boundary";
+
+    pub const MODE: &str = "mode";
+    pub const AGENTS: &str = "agents";
+    pub const STEPS: &str = "steps";
+    pub const WARMUP_STEPS: &str = "warmup-steps";
+    pub const WORKER_THREADS: &str = "worker-threads";
+    pub const MODES: &str = "modes";
+
+    /// What every launched worker is given.
+    pub const COMMON: &[&str] = &[
+        SOCKET,
+        STORE_ROOT,
+        CLIENT_ID,
+        SERVICE,
+        THREADS,
+        SESSION,
+        INCARNATION,
+    ];
+    /// What only an agent is given.
+    pub const AGENT_ONLY: &[&str] = &[
+        AGENT,
+        PORT,
+        TICK_NUMERATOR,
+        TICK_DENOMINATOR,
+        WARMUP_TICKS,
+        PREPARE_DELAY_MS,
+        COMMIT_DELAY_MS,
+        FAIL_COMMIT_AT_STEP,
+    ];
+    /// What only the environment is given, media options included.
+    pub const ENVIRONMENT_ONLY: &[&str] = &[
+        WORKER,
+        PORTS,
+        STEP_NUMERATOR,
+        STEP_DENOMINATOR,
+        ADVANCE_DELAY_MS,
+        OMIT_VIEW_AT_BOUNDARY,
+        OBSERVATION_DELAY_STEPS,
+        STALE_VIEW_AT_BOUNDARY,
+        TRUNCATED_VIEW_AT_BOUNDARY,
+        OMIT_AUDIO_AT_BOUNDARY,
+        OVERLAPPING_AUDIO_AT_BOUNDARY,
+    ];
+    /// What a measurement run or one of its row children is given.
+    pub const MEASURE: &[&str] = &[MODE, AGENTS, STEPS, WARMUP_STEPS, WORKER_THREADS, MODES];
+}
+
+/// One `--flag value` pair, so the flag name is written once and never spelled inline.
+fn arg(name: &str, value: impl std::fmt::Display) -> (String, String) {
+    (format!("--{name}"), value.to_string())
+}
+
 // -------------------------------------------------------------------------------------------
 // What a participant is
 
@@ -1220,48 +1318,50 @@ impl Started {
         match self {
             Started::Agent(spec) => {
                 let mut args = vec![
-                    ("--session".to_owned(), spec.session_id.clone()),
+                    arg(flags::SESSION, &spec.session_id),
-                    ("--agent".to_owned(), spec.agent_id.clone()),
+                    arg(flags::AGENT, &spec.agent_id),
-                    ("--port".to_owned(), spec.port_id.clone()),
+                    arg(flags::PORT, &spec.port_id),
-                    ("--incarnation".to_owned(), spec.incarnation_id.clone()),
+                    arg(flags::INCARNATION, &spec.incarnation_id),
-                    ("--tick-numerator".to_owned(), spec.tick_duration.numerator.to_string()),
+                    arg(flags::TICK_NUMERATOR, spec.tick_duration.numerator),
-                    (
+                    arg(flags::TICK_DENOMINATOR, spec.tick_duration.denominator),
-                        "--tick-denominator".to_owned(),
+                    arg(flags::WARMUP_TICKS, spec.warmup_ticks),
-                        spec.tick_duration.denominator.to_string(),
+                    arg(flags::PREPARE_DELAY_MS, spec.faults.prepare_delay_ms),
-                    ),
+                    arg(flags::COMMIT_DELAY_MS, spec.faults.commit_delay_ms),
-                    ("--warmup-ticks".to_owned(), spec.warmup_ticks.to_string()),
-                    (
-                        "--prepare-delay-ms".to_owned(),
-                        spec.faults.prepare_delay_ms.to_string(),
-                    ),
-                    (
-                        "--commit-delay-ms".to_owned(),
-                        spec.faults.commit_delay_ms.to_string(),
-                    ),
                 ];
                 if let Some(step) = spec.faults.fail_commit_at_step {
-                    args.push(("--fail-commit-at-step".to_owned(), step.to_string()));
+                    args.push(arg(flags::FAIL_COMMIT_AT_STEP, step));
                 }
                 args
             }
             Started::Environment(spec) => {
                 let mut args = vec![
-                    ("--session".to_owned(), spec.session_id.clone()),
+                    arg(flags::SESSION, &spec.session_id),
-                    ("--worker".to_owned(), spec.worker_id.clone()),
+                    arg(flags::WORKER, &spec.worker_id),
-                    ("--incarnation".to_owned(), spec.incarnation_id.clone()),
+                    arg(flags::INCARNATION, &spec.incarnation_id),
-                    ("--step-numerator".to_owned(), spec.step_duration.numerator.to_string()),
+                    arg(flags::STEP_NUMERATOR, spec.step_duration.numerator),
-                    (
+                    arg(flags::STEP_DENOMINATOR, spec.step_duration.denominator),
-                        "--step-denominator".to_owned(),
+                    arg(flags::PORTS, spec.ports.join(",")),
-                        spec.step_duration.denominator.to_string(),
+                    arg(flags::ADVANCE_DELAY_MS, spec.faults.advance_delay_ms),
-                    ),
+                    // The media options a world in another process needs to be exactly this
-                    ("--ports".to_owned(), spec.ports.join(",")),
+                    // world. Its render counter and its agents' sensor logs stay there.
-                    (
+                    arg(flags::OBSERVATION_DELAY_STEPS, spec.observation_delay_steps),
-                        "--advance-delay-ms".to_owned(),
-                        spec.faults.advance_delay_ms.to_string(),
-                    ),
                 ];
-                if let Some(boundary) = spec.faults.omit_view_at_boundary {
+                for (flag, boundary) in [
-                    args.push(("--omit-view-at-boundary".to_owned(), boundary.to_string()));
+                    (flags::OMIT_VIEW_AT_BOUNDARY, spec.faults.omit_view_at_boundary),
+                    (flags::STALE_VIEW_AT_BOUNDARY, spec.faults.stale_view_at_boundary),
+                    (
+                        flags::TRUNCATED_VIEW_AT_BOUNDARY,
+                        spec.faults.truncated_view_at_boundary,
+                    ),
+                    (flags::OMIT_AUDIO_AT_BOUNDARY, spec.faults.omit_audio_at_boundary),
+                    (
+                        flags::OVERLAPPING_AUDIO_AT_BOUNDARY,
+                        spec.faults.overlapping_audio_at_boundary,
+                    ),
+                ] {
+                    if let Some(boundary) = boundary {
+                        args.push(arg(flag, boundary));
+                    }
                 }
                 args
             }
@@ -1277,6 +1377,7 @@ pub(crate) fn agent_config(spec: &AgentLaunch, worker_threads: usize) -> AgentCo
         tick_duration: spec.tick_duration,
         warmup_ticks: spec.warmup_ticks,
         worker_threads,
+        sensors: spec.sensors.clone(),
         faults: spec.faults.clone(),
     }
 }
@@ -1289,6 +1390,8 @@ pub(crate) fn environment_config(spec: &EnvironmentLaunch) -> EnvironmentConfig
         step_duration: spec.step_duration,
         ports: spec.ports.clone(),
         worker_threads: spec.worker_threads,
+        observation_delay_steps: spec.observation_delay_steps,
+        renders: spec.renders.clone(),
         faults: spec.faults.clone(),
     }
 }
@@ -1342,3 +1445,98 @@ pub(crate) async fn register_with_retry(
         }
     }
 }
+
+#[cfg(test)]
+mod flag_tests {
+    use super::*;
+
+    fn every_media_fault() -> EnvironmentFaults {
+        EnvironmentFaults {
+            advance_delay_ms: 3,
+            omit_view_at_boundary: Some(1),
+            stale_view_at_boundary: Some(2),
+            truncated_view_at_boundary: Some(3),
+            omit_audio_at_boundary: Some(4),
+            overlapping_audio_at_boundary: Some(5),
+        }
+    }
+
+    fn environment_launch() -> EnvironmentLaunch {
+        EnvironmentLaunch {
+            session_id: id("demo"),
+            worker_id: id("arena"),
+            incarnation_id: id("arena-inc-1"),
+            step_duration: RationalNs::new(1, 60).expect("a cadence"),
+            ports: vec![id("p1"), id("p2")],
+            worker_threads: 1,
+            observation_delay_steps: 2,
+            renders: crate::media::RenderCounter::new(),
+            faults: every_media_fault(),
+            client_id: "environment".to_owned(),
+            service: "env.arena".to_owned(),
+        }
+    }
+
+    fn agent_launch() -> AgentLaunch {
+        AgentLaunch {
+            session_id: id("demo"),
+            agent_id: id("fly-a"),
+            port_id: id("p1"),
+            incarnation_id: id("fly-a-inc-1"),
+            tick_duration: RationalNs::new(1, 1_000).expect("a tick"),
+            warmup_ticks: 10,
+            worker_threads: 1,
+            sensors: crate::media::SensorLog::new(),
+            faults: AgentFaults {
+                fail_commit_at_step: Some(2),
+                prepare_delay_ms: 1,
+                commit_delay_ms: 2,
+            },
+            client_id: "worker-fly-a".to_owned(),
+            service: "agent.fly-a".to_owned(),
+        }
+    }
+
+    /// Both halves of the command line name the same constants, and this proves it for every
+    /// argument a launch can produce: a flag the launcher writes that the parser does not
+    /// accept would be a silently ignored option, which is what the parser now refuses.
+    #[test]
+    fn every_flag_a_launch_writes_is_one_its_command_accepts() {
+        for (started, allowed) in [
+            (
+                Started::Environment(environment_launch()),
+                [flags::COMMON, flags::ENVIRONMENT_ONLY],
+            ),
+            (Started::Agent(agent_launch()), [flags::COMMON, flags::AGENT_ONLY]),
+        ] {
+            let arguments = started.arguments();
+            assert!(!arguments.is_empty());
+            for (flag, _value) in &arguments {
+                let name = flag.strip_prefix("--").expect("every argument is a --flag");
+                assert!(
+                    allowed.iter().any(|set| set.contains(&name)),
+                    "{} writes --{name}, which its command does not accept",
+                    started.subcommand()
+                );
+            }
+        }
+    }
+
+    /// Every media option reaches the argv when it is set, so a world in another process is
+    /// exactly the world the composition asked for.
+    #[test]
+    fn the_media_options_are_all_written_for_a_separate_process() {
+        let started = Started::Environment(environment_launch());
+        let written: Vec<String> = started.arguments().into_iter().map(|(f, _)| f).collect();
+        for flag in [
+            flags::OBSERVATION_DELAY_STEPS,
+            flags::OMIT_VIEW_AT_BOUNDARY,
+            flags::STALE_VIEW_AT_BOUNDARY,
+            flags::TRUNCATED_VIEW_AT_BOUNDARY,
+            flags::OMIT_AUDIO_AT_BOUNDARY,
+            flags::OVERLAPPING_AUDIO_AT_BOUNDARY,
+        ] {
+            assert!(written.contains(&format!("--{flag}")), "--{flag} is not written");
+        }
+    }
+}

services/flysim/crates/fly-session/src/lib.rs

@@ -30,6 +30,7 @@ pub mod environment;
 pub mod harness;
 pub mod launcher;
 pub mod measure;
+pub mod media;
 pub mod metrics;
 pub mod phase;
 pub mod rpc;

services/flysim/crates/fly-session/src/measure.rs

@@ -24,7 +24,7 @@ use serde_json::{Value, json};
 
 use crate::coordinator::DispatchOrder;
 use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via};
-use crate::launcher::ExecutionMode;
+use crate::launcher::{ExecutionMode, flags};
 use crate::metrics::{Percentiles, physical_cores};
 
 /// What to compare.
@@ -241,15 +241,15 @@ fn row_in_a_child(
 ) -> Result<Row, String> {
     let output = std::process::Command::new(program)
         .arg("measure-row")
-        .arg("--mode")
+        .arg(format!("--{}", flags::MODE))
         .arg(mode.label())
-        .arg("--agents")
+        .arg(format!("--{}", flags::AGENTS))
         .arg(agents.to_string())
-        .arg("--steps")
+        .arg(format!("--{}", flags::STEPS))
         .arg(config.steps.to_string())
-        .arg("--warmup-steps")
+        .arg(format!("--{}", flags::WARMUP_STEPS))
         .arg(config.warmup_steps.to_string())
-        .arg("--worker-threads")
+        .arg(format!("--{}", flags::WORKER_THREADS))
         .arg(config.worker_threads.to_string())
         .stdin(std::process::Stdio::null())
         .output()

services/flysim/crates/fly-session/src/media.rs

@@ -0,0 +1,837 @@
+//! Native observations and the presentation handoff: the MEDIA-01 slice.
+//!
+//! Everything here sits on top of the bus `ArtifactRef` and its ownership rules. There is no
+//! second buffer system: an environment allocates, writes and seals one immutable object per
+//! boundary, the coordinator forwards that one owned handle to every agent and to publication,
+//! and a spectator reads it through an ordinary latest subscription.
+//!
+//! The module owns four things:
+//!
+//! 1. Production. [`ViewPipeline`] renders one native frame per boundary and hands out the
+//!    frame a declared `observationDelaySteps` requires, so a pipeline delay is a real queue
+//!    rather than a number in a descriptor. [`AudioSource`] produces one chunk per boundary
+//!    with an exact rational sample budget.
+//! 2. Acceptance. [`check_required_views`] and [`AudioTimelines`] are the coordinator's Phase C
+//!    media checks: a required sensory view must exist at exactly the producing boundary its
+//!    declared delay implies, and audio chunks cannot overlap or go backwards inside an epoch.
+//! 3. Consumption. [`Spectator`] is a presentation-side consumer on a latest subscription with
+//!    finite credits, and [`detach_frame`] is the renderer that keeps its handle after the
+//!    message is gone.
+//! 4. Identity. [`AssetRegistry`] holds installed persistent content named by `AssetRef`.
+//!    Importing an asset produces a *new* transient artifact; the two identities never convert.
+//!
+//! Resizing, overlays, compositing, mixing, encoding and streaming are not here and are not
+//! anywhere else in this crate: they belong to the application's presentation layer.
+
+use std::collections::{BTreeMap, VecDeque};
+use std::io::Write;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use fly_session_types::media::AudioTimeline;
+
+// `crate::types` is this crate's facade over the shared `fly-session-types` crate; the glob
+// keeps the contract's own names in sight instead of restating them.
+use crate::types::*;
+
+/// The content type of a native RGBA8 frame. Top-left origin, no padded rows.
+pub const FRAME_CONTENT_TYPE: &str = "image/x-rgba8";
+
+/// The content type of a native audio chunk: interleaved little-endian f32.
+pub const AUDIO_CONTENT_TYPE: &str = "audio/x-f32le";
+
+/// The attachment name one view's pixels travel under.
+pub fn view_attachment(view_id: &str) -> String {
+    format!("view.{view_id}")
+}
+
+/// The attachment name one audio stream's samples travel under.
+pub fn audio_attachment(stream_id: &str) -> String {
+    format!("audio.{stream_id}")
+}
+
+fn store_error(what: &str, message: &str) -> DomainError {
+    DomainError::new(
+        ErrorCode::BackendFailure,
+        format!("{what}: {message}"),
+        MutationCertainty::Applied,
+    )
+}
+
+fn media_error(message: impl std::fmt::Display) -> DomainError {
+    // A world that advanced without usable media leaves the transition's certainty unknown:
+    // the mutation happened, the observation of it did not.
+    DomainError::new(ErrorCode::BufferInvalid, message, MutationCertainty::Unknown)
+}
+
+// ----------------------------------------------------------------------------------------------
+// Production
+
+/// How many frames a producer has actually rendered.
+///
+/// A shared counter, so a test can prove that forwarding one image to several recipients
+/// renders it once.
+#[derive(Clone, Debug, Default)]
+pub struct RenderCounter(Arc<AtomicU64>);
+
+impl RenderCounter {
+    pub fn new() -> RenderCounter {
+        RenderCounter::default()
+    }
+
+    pub fn bump(&self) {
+        self.0.fetch_add(1, Ordering::SeqCst);
+    }
+
+    pub fn count(&self) -> u64 {
+        self.0.load(Ordering::SeqCst)
+    }
+}
+
+/// One native frame of the counter arena: a real synthetic pattern, not a constant fill.
+///
+/// Top-left RGBA8 with `rowStride` exactly `4 x width` and no padded rows, which is the only
+/// pixel layout v1 has. The red channel carries the world counter, so a reader that samples
+/// one pixel still reads the world; green is a horizontal ramp and blue a vertical ramp with
+/// a one-column bar that walks with the boundary, so consecutive frames differ.
+pub fn arena_frame(descriptor: &ViewDescriptor, counter: i64, boundary: u64) -> Vec<u8> {
+    let width = descriptor.width;
+    let height = descriptor.height;
+    let stride = descriptor.row_stride as usize;
+    let mut out = vec![0u8; stride * height as usize];
+    let counter_byte = (counter & 0xff) as u8;
+    let bar = boundary % width;
+    for y in 0..height {
+        let row = y as usize * stride;
+        for x in 0..width {
+            let p = row + x as usize * 4;
+            let ramp_x = if width > 1 {
+                (x * 255 / (width - 1)) as u8
+            } else {
+                0
+            };
+            let ramp_y = if height > 1 {
+                (y * 255 / (height - 1)) as u8
+            } else {
+                0
+            };
+            out[p] = counter_byte;
+            out[p + 1] = ramp_x;
+            out[p + 2] = if x == bar { 255 } else { ramp_y };
+            out[p + 3] = 255;
+        }
+    }
+    out
+}
+
+/// One view's production pipeline: render at every boundary, deliver with the declared delay.
+///
+/// `observationDelaySteps` is a real queue here. At boundary `b` the required frame is the one
+/// produced at `max(0, b - delay)`, so a declared delay of two repeats `O[0]` at boundaries 0,
+/// 1 and 2 -- the bootstrap repetition the contract allows -- and then advances one frame per
+/// boundary. Nothing beyond that is retained: an older frame is dropped, so a later boundary
+/// cannot be served an arbitrary stale image.
+pub struct ViewPipeline {
+    descriptor: ViewDescriptor,
+    frames: VecDeque<(u64, flybus::Artifact)>,
+    renders: RenderCounter,
+}
+
+impl ViewPipeline {
+    pub fn new(descriptor: ViewDescriptor, renders: RenderCounter) -> ViewPipeline {
+        ViewPipeline {
+            descriptor,
+            frames: VecDeque::new(),
+            renders,
+        }
+    }
+
+    pub fn descriptor(&self) -> &ViewDescriptor {
+        &self.descriptor
+    }
+
+    /// Seals one immutable frame for `boundary` and files it under its producing boundary.
+    pub async fn render(
+        &mut self,
+        client: &flybus::Client,
+        boundary: u64,
+        counter: i64,
+    ) -> DomainResult<()> {
+        let bytes = arena_frame(&self.descriptor, counter, boundary);
+        let artifact = seal(client, FRAME_CONTENT_TYPE, &bytes).await?;
+        self.renders.bump();
+        self.frames.push_back((boundary, artifact));
+        // Keep exactly the frames a declared delay can still require.
+        while self.frames.len() > self.descriptor.observation_delay_steps as usize + 1 {
+            self.frames.pop_front();
+        }
+        Ok(())
+    }
+
+    /// Seals a frame of the wrong length, which is what a broken backend produces. The
+    /// reference it returns describes the artifact honestly, so the shape check is the thing
+    /// under test rather than a lie in the payload.
+    pub async fn render_truncated(
+        &mut self,
+        client: &flybus::Client,
+        boundary: u64,
+        counter: i64,
+    ) -> DomainResult<()> {
+        let mut bytes = arena_frame(&self.descriptor, counter, boundary);
+        bytes.truncate(bytes.len() - self.descriptor.row_stride as usize);
+        let artifact = seal(client, FRAME_CONTENT_TYPE, &bytes).await?;
+        self.renders.bump();
+        self.frames.push_back((boundary, artifact));
+        while self.frames.len() > self.descriptor.observation_delay_steps as usize + 2 {
+            self.frames.pop_front();
+        }
+        Ok(())
+    }
+
+    /// The view reference and the owned handle a required sensory view has at `boundary`.
+    pub fn at(&self, boundary: u64) -> Option<(ViewRef, flybus::Artifact)> {
+        let produced = self.descriptor.required_produced_step(boundary);
+        self.frame_produced_at(produced)
+    }
+
+    /// The frame produced at exactly `produced`, if it is still retained.
+    pub fn frame_produced_at(&self, produced: u64) -> Option<(ViewRef, flybus::Artifact)> {
+        self.frames
+            .iter()
+            .find(|(step, _)| *step == produced)
+            .map(|(step, artifact)| {
+                (
+                    ViewRef {
+                        view_id: self.descriptor.view_id.clone(),
+                        produced_step: *step,
+                        pixels: artifact.reference().clone(),
+                    },
+                    artifact.clone(),
+                )
+            })
+    }
+
+    /// How many boundaries this pipeline has rendered.
+    pub fn renders(&self) -> u64 {
+        self.renders.count()
+    }
+
+}
+
+/// One audio stream's production: an exact sample budget and a deterministic waveform.
+///
+/// The number of frames in a step is `sampleRate x stepDuration`, accumulated as a rational so
+/// a cadence that does not divide the sample rate never drifts: 8 kHz at 60 Hz produces
+/// 133, 133, 134, ... and the sum is exact at every boundary. The waveform is integer-phase
+/// arithmetic only, so a `fixed-build` environment produces the same bytes on every run.
+pub struct AudioSource {
+    descriptor: AudioDescriptor,
+    /// The unconsumed fraction of a frame, over `denominator`.
+    accumulator: u128,
+    denominator: u128,
+    next_sample: u64,
+    phase: u64,
+    chunks: u64,
+    discontinuous: bool,
+}
+
+impl AudioSource {
+    /// A fresh episode, whose first chunk starts at the configured audio origin.
+    pub fn new(descriptor: AudioDescriptor, origin: u64) -> AudioSource {
+        AudioSource {
+            descriptor,
+            accumulator: 0,
+            denominator: 1,
+            next_sample: origin,
+            phase: 0,
+            chunks: 0,
+            discontinuous: false,
+        }
+    }
+
+    /// A new epoch after a restore: the sample position is preserved and the first chunk of
+    /// this epoch marks a discontinuity.
+    pub fn restored_at(descriptor: AudioDescriptor, sample: u64) -> AudioSource {
+        let mut source = AudioSource::new(descriptor, sample);
+        source.discontinuous = true;
+        source
+    }
+
+    pub fn descriptor(&self) -> &AudioDescriptor {
+        &self.descriptor
+    }
+
+    pub fn next_sample(&self) -> u64 {
+        self.next_sample
+    }
+
+    pub fn chunks(&self) -> u64 {
+        self.chunks
+    }
+
+    /// The exact number of sample frames one step of `step` nanoseconds contains.
+    ///
+    /// The remainder is kept, never rounded: the accumulator is integer arithmetic over the
+    /// common denominator `stepDenominator x 1e9`.
+    pub fn frames_for_step(&mut self, step: &RationalNs) -> DomainResult<u64> {
+        let denominator = u128::from(step.denominator)
+            .checked_mul(1_000_000_000)
+            .ok_or_else(|| DomainError::invalid("audio: the step denominator overflows"))?;
+        if self.denominator != denominator {
+            // A cadence change would need a new epoch; carrying a remainder across one would
+            // be a silent resample.
+            if self.chunks > 0 {
+                return Err(DomainError::invalid(
+                    "audio: the cadence changed inside an epoch",
+                ));
+            }
+            self.denominator = denominator;
+        }
+        let per_step = u128::from(self.descriptor.sample_rate)
+            .checked_mul(u128::from(step.numerator))
+            .ok_or_else(|| DomainError::invalid("audio: the sample budget overflows"))?;
+        self.accumulator = self
+            .accumulator
+            .checked_add(per_step)
+            .ok_or_else(|| DomainError::invalid("audio: the sample accumulator overflows"))?;
+        let frames = self.accumulator / self.denominator;
+        self.accumulator %= self.denominator;
+        u64::try_from(frames).map_err(|_| DomainError::invalid("audio: too many frames in a step"))
+    }
+
+    /// Produces one chunk covering exactly one step of world time.
+    pub async fn produce(
+        &mut self,
+        client: &flybus::Client,
+        step: &RationalNs,
+        counter: i64,
+    ) -> DomainResult<(AudioRef, flybus::Artifact)> {
+        let frames = self.frames_for_step(step)?;
+        let bytes = self.samples(frames, counter);
+        let artifact = seal(client, AUDIO_CONTENT_TYPE, &bytes).await?;
+        let chunk = AudioRef {
+            stream_id: self.descriptor.stream_id.clone(),
+            first_sample: self.next_sample,
+            sample_frames: frames,
+            samples: artifact.reference().clone(),
+            discontinuity: self.discontinuous && self.chunks == 0,
+        };
+        self.next_sample = self
+            .next_sample
+            .checked_add(frames)
+            .ok_or_else(|| DomainError::invalid("audio: the sample position overflows"))?;
+        self.chunks += 1;
+        Ok((chunk, artifact))
+    }
+
+    /// A deterministic triangle wave whose pitch follows the world counter, interleaved across
+    /// the declared channels. Every sample is finite by construction.
+    fn samples(&mut self, frames: u64, counter: i64) -> Vec<u8> {
+        let rate = self.descriptor.sample_rate;
+        let channels = self.descriptor.channels;
+        let step = 220 + (counter.rem_euclid(8) as u64) * 55;
+        let mut out = Vec::with_capacity((frames * channels * 4) as usize);
+        for _ in 0..frames {
+            self.phase = (self.phase + step) % rate;
+            let position = self.phase as f32 / rate as f32;
+            // 1 - 2|2p - 1| is a triangle in [-1, 1] built from exact IEEE operations.
+            let value = 1.0 - 2.0 * (2.0 * position - 1.0).abs();
+            for channel in 0..channels {
+                let scaled = value * 0.25 / (channel + 1) as f32;
+                out.extend_from_slice(&scaled.to_le_bytes());
+            }
+        }
+        out
+    }
+}
+
+/// Allocates, writes and seals one immutable artifact.
+async fn seal(
+    client: &flybus::Client,
+    content_type: &str,
+    bytes: &[u8],
+) -> DomainResult<flybus::Artifact> {
+    let mut writer = client
+        .artifacts()
+        .allocate(bytes.len() as u64, content_type)
+        .await
+        .map_err(|e| store_error("allocate", &e.message))?;
+    writer
+        .write_all(bytes)
+        .map_err(|e| store_error("write", &e.to_string()))?;
+    writer
+        .seal()
+        .await
+        .map_err(|e| store_error("seal", &e.message))
+}
+
+/// Splits one reply's attachments into the view handles and the audio handles.
+///
+/// Views are sensory data forwarded to agents; audio is presentation data that is published
+/// and never attached to a sensory input.
+pub fn split_attachments(
+    artifacts: BTreeMap<String, flybus::Artifact>,
+) -> (
+    BTreeMap<String, flybus::Artifact>,
+    BTreeMap<String, flybus::Artifact>,
+) {
+    let mut views = BTreeMap::new();
+    let mut audio = BTreeMap::new();
+    for (name, artifact) in artifacts {
+        if name.starts_with("audio.") {
+            audio.insert(name, artifact);
+        } else {
+            views.insert(name, artifact);
+        }
+    }
+    (views, audio)
+}
+
+/// The attachment names one environment's declared media travel under.
+pub fn attachment_names(descriptor: &EnvironmentDescriptor) -> Vec<String> {
+    descriptor
+        .views
+        .iter()
+        .map(|view| view_attachment(&view.view_id))
+        .chain(
+            descriptor
+                .audio
+                .iter()
+                .map(|stream| audio_attachment(&stream.stream_id)),
+        )
+        .collect()
+}
+
+// ----------------------------------------------------------------------------------------------
+// Acceptance
+
+/// Every declared view must be present at exactly the producing boundary its delay implies.
+///
+/// A missing spectator frame is tolerable; a missing required sensory input is not. Neither is
+/// one that arrived from an older boundary than the declared delay allows: the transition
+/// fails instead of the session substituting whatever frame it happens to hold.
+pub fn check_required_views(
+    descriptor: &EnvironmentDescriptor,
+    observation: &WorldObservation,
+) -> DomainResult<()> {
+    for view in &descriptor.views {
+        let want = required_produced_step(view, observation.boundary);
+        match observation
+            .sensory_views
+            .iter()
+            .find(|given| given.view_id == view.view_id)
+        {
+            Some(given) if given.produced_step == want => {}
+            Some(given) => {
+                return Err(media_error(format!(
+                    "view {} came from boundary {}, and its declared delay of {} requires {want}",
+                    view.view_id, given.produced_step, view.observation_delay_steps
+                )));
+            }
+            None => {
+                return Err(media_error(format!(
+                    "required sensory view {} is missing",
+                    view.view_id
+                )));
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Every declared audio stream produces exactly one chunk per transition.
+///
+/// The contract states the shape and the ordering of chunks, not whether one has to exist, so
+/// this is MEDIA-01's choice and it is deliberate: a session that tolerates a silently missing
+/// chunk cannot tell "this world produced no audio for this interval" from "the chunk was
+/// lost", and the second is the case the retention rules care about. Boundary 0 has no
+/// preceding interval and so carries no chunk.
+pub fn check_required_audio(
+    descriptor: &EnvironmentDescriptor,
+    observation: &WorldObservation,
+) -> DomainResult<()> {
+    if observation.boundary == 0 {
+        return Ok(());
+    }
+    for stream in &descriptor.audio {
+        if !observation
+            .audio
+            .iter()
+            .any(|chunk| chunk.stream_id == stream.stream_id)
+        {
+            return Err(media_error(format!(
+                "declared audio stream {} produced no chunk for this transition",
+                stream.stream_id
+            )));
+        }
+    }
+    Ok(())
+}
+
+/// Every declared audio stream's chunk sequence, one timeline per stream and epoch.
+#[derive(Clone, Debug, Default)]
+pub struct AudioTimelines(BTreeMap<String, AudioTimeline>);
+
+impl AudioTimelines {
+    /// Fresh timelines for a new episode: every declared stream starts at origin zero.
+    pub fn fresh(descriptor: &EnvironmentDescriptor) -> AudioTimelines {
+        AudioTimelines(
+            descriptor
+                .audio
+                .iter()
+                .map(|stream| (stream.stream_id.clone(), AudioTimeline::fresh(stream, 0)))
+                .collect(),
+        )
+    }
+
+    /// Timelines for a new epoch after a restore: each stream resumes at its preserved sample
+    /// position, and each one's first chunk must mark a discontinuity.
+    ///
+    /// A declared stream with no recorded position is an error. Resuming it at sample zero
+    /// would restart the episode's audio clock silently, which is exactly the best-effort
+    /// policy the restore rules refuse: crash restore *preserves* the sample position.
+    pub fn restored(
+        descriptor: &EnvironmentDescriptor,
+        positions: &BTreeMap<String, u64>,
+    ) -> DomainResult<AudioTimelines> {
+        let mut timelines = BTreeMap::new();
+        for stream in &descriptor.audio {
+            let at = positions.get(&stream.stream_id).copied().ok_or_else(|| {
+                DomainError::before(
+                    ErrorCode::IncompatibleState,
+                    format!(
+                        "audio stream {} has no restored sample position",
+                        stream.stream_id
+                    ),
+                )
+            })?;
+            timelines.insert(stream.stream_id.clone(), AudioTimeline::restored_at(stream, at));
+        }
+        Ok(AudioTimelines(timelines))
+    }
+
+    /// Accepts one observation's chunks. Unknown streams and out-of-sequence chunks fail.
+    pub fn accept(
+        &mut self,
+        descriptor: &EnvironmentDescriptor,
+        observation: &WorldObservation,
+    ) -> DomainResult<()> {
+        for chunk in &observation.audio {
+            let declared = descriptor
+                .audio_stream(&chunk.stream_id)
+                .ok_or_else(|| media_error(format!("audio stream {} is not declared", chunk.stream_id)))?;
+            let timeline = self
+                .0
+                .get_mut(&chunk.stream_id)
+                .ok_or_else(|| media_error(format!("audio stream {} has no timeline", chunk.stream_id)))?;
+            timeline.accept(chunk, declared).map_err(media_error)?;
+        }
+        Ok(())
+    }
+
+    /// Where each stream's next chunk may start.
+    pub fn positions(&self) -> BTreeMap<String, u64> {
+        self.0
+            .iter()
+            .map(|(id, timeline)| (id.clone(), timeline.next_sample()))
+            .collect()
+    }
+
+    pub fn accepted(&self, stream_id: &str) -> u64 {
+        self.0.get(stream_id).map_or(0, AudioTimeline::accepted)
+    }
+}
+
+// ----------------------------------------------------------------------------------------------
+// Consumption
+
+/// What one agent actually sensed, recorded where a test can read it.
+///
+/// The fake agent is the only thing that reads the pixels, so this is how "one shared image
+/// reached both agents" is proved from the agents' side rather than from the producer's.
+#[derive(Clone, Debug, Default)]
+pub struct SensorLog(Arc<std::sync::Mutex<Vec<SensedView>>>);
+
+/// One view an agent read: which boundary it was consumed at, which artifact it was, and the
+/// digest of the bytes the agent actually read.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SensedView {
+    pub boundary: u64,
+    pub view_id: Id,
+    pub artifact_id: String,
+    pub produced_step: u64,
+    pub digest: Digest,
+}
+
+impl SensorLog {
+    pub fn new() -> SensorLog {
+        SensorLog::default()
+    }
+
+    pub fn record(&self, view: SensedView) {
+        self.0.lock().expect("the sensor log is never poisoned").push(view);
+    }
+
+    pub fn entries(&self) -> Vec<SensedView> {
+        self.0.lock().expect("the sensor log is never poisoned").clone()
+    }
+
+    /// The artifacts this agent read, in order.
+    pub fn artifact_ids(&self) -> Vec<String> {
+        self.entries().into_iter().map(|v| v.artifact_id).collect()
+    }
+}
+
+/// One frame a spectator took off its subscription.
+pub struct SpectatorFrame {
+    pub boundary: u64,
+    /// Every agent the committed snapshot carries, in publication order. A presentation
+    /// consumer is a multi-agent consumer: one snapshot holds the whole session.
+    pub agents: Vec<Id>,
+    pub sequence: u64,
+    /// How many undelivered snapshots were coalesced into this one.
+    pub replaced: u64,
+    pub view: ViewRef,
+    pub artifact: flybus::Artifact,
+    /// Each published chunk with the handle it travelled on.
+    pub audio: Vec<(AudioRef, flybus::Artifact)>,
+}
+
+/// A presentation-side consumer of committed snapshots.
+///
+/// It subscribes `latest` with finite credits, which is the spectator row of the domain
+/// retention table: new snapshots replace its queued value, it never blocks the session, and
+/// the only thing a slow one exhausts is its own credits.
+pub struct Spectator {
+    subscription: flybus::Subscription,
+    held: Vec<flybus::Message>,
+    seen: u64,
+    coalesced: u64,
+}
+
+impl Spectator {
+    /// Subscribes to `topic` in latest mode with `credits` in flight.
+    ///
+    /// A latest subscription always has exactly one queued value; `credits` is its in-flight
+    /// bound, which the router limits (two by default). Finite credits are the spectator row
+    /// of the domain retention table: they are the only thing a slow viewer exhausts.
+    pub async fn attach(
+        client: &flybus::Client,
+        topic: &str,
+        credits: u32,
+    ) -> Result<Spectator, flybus::BusError> {
+        let subscription = client
+            .subscribe(
+                topic,
+                flybus::SubscriptionConfig::latest().in_flight(credits).replay(true),
+            )
+            .await?;
+        Ok(Spectator {
+            subscription,
+            held: Vec::new(),
+            seen: 0,
+            coalesced: 0,
+        })
+    }
+
+    /// Takes the next snapshot, reads its frame and releases the delivery.
+    pub async fn take_frame(&mut self) -> Option<SpectatorFrame> {
+        let message = self.subscription.next().await?;
+        self.seen += 1;
+        self.coalesced += message.replaced();
+        let frame = snapshot_frame(&message);
+        drop(message);
+        frame
+    }
+
+    /// Takes the next snapshot message itself, for a renderer that keeps its own handle
+    /// after the message is gone.
+    pub async fn next_message(&mut self) -> Option<flybus::Message> {
+        let message = self.subscription.next().await?;
+        self.seen += 1;
+        self.coalesced += message.replaced();
+        Some(message)
+    }
+
+    /// Takes a snapshot without consuming it, which is what a viewer that stops rendering
+    /// does. Its credits run out and nothing else in the session notices.
+    pub async fn hold_one(&mut self) -> bool {
+        match self.subscription.next().await {
+            Some(message) => {
+                self.seen += 1;
+                self.coalesced += message.replaced();
+                self.held.push(message);
+                true
+            }
+            None => false,
+        }
+    }
+
+    /// Takes a snapshot without consuming it if one is queued right now.
+    pub fn try_hold_one(&mut self) -> bool {
+        match self.subscription.try_next() {
+            Some(message) => {
+                self.seen += 1;
+                self.coalesced += message.replaced();
+                self.held.push(message);
+                true
+            }
+            None => false,
+        }
+    }
+
+    /// Releases everything this spectator was holding, returning its credits.
+    pub fn release(&mut self) {
+        self.held.clear();
+    }
+
+    pub fn held(&self) -> usize {
+        self.held.len()
+    }
+
+    pub fn seen(&self) -> u64 {
+        self.seen
+    }
+
+    /// How many snapshots were replaced in this spectator's queue while it was busy.
+    pub fn coalesced(&self) -> u64 {
+        self.coalesced
+    }
+}
+
+/// Reads one committed snapshot's first view and its handle.
+pub fn snapshot_frame(message: &flybus::Message) -> Option<SpectatorFrame> {
+    let payload = message.payload();
+    let boundary = payload
+        .get("scope")
+        .and_then(|s| s.get("step"))
+        .and_then(serde_json::Value::as_str)
+        .and_then(|s| s.parse::<u64>().ok())?;
+    let media = payload.get("media")?;
+    let views = media.get("views")?.as_array()?;
+    let view = ViewRef::from_json(views.first()?).ok()?;
+    // An unreadable chunk, or one whose handle is not attached, makes the whole snapshot
+    // unreadable rather than a snapshot that quietly has less audio in it than was published.
+    let mut audio = Vec::new();
+    for value in media.get("audio")?.as_array()? {
+        let chunk = AudioRef::from_json(value).ok()?;
+        let artifact = message.artifact(&audio_attachment(&chunk.stream_id)).ok()?;
+        audio.push((chunk, artifact));
+    }
+    let artifact = message.artifact(&view_attachment(&view.view_id)).ok()?;
+    let agents = payload
+        .get("agents")
+        .and_then(serde_json::Value::as_array)
+        .map(|agents| {
+            agents
+                .iter()
+                .filter_map(|a| a.get("agentId").and_then(serde_json::Value::as_str))
+                .map(str::to_owned)
+                .collect()
+        })
+        .unwrap_or_default();
+    Some(SpectatorFrame {
+        boundary,
+        agents,
+        sequence: message.topic_sequence(),
+        replaced: message.replaced(),
+        view,
+        artifact,
+        audio,
+    })
+}
+
+/// Takes the frame out of a message and drops the message, as a renderer that finishes later
+/// does. The delivery stays alive because the extracted handle still owns it.
+pub fn detach_frame(message: flybus::Message) -> Option<(ViewRef, flybus::Artifact)> {
+    let frame = snapshot_frame(&message)?;
+    drop(message);
+    Some((frame.view, frame.artifact))
+}
+
+// ----------------------------------------------------------------------------------------------
+// Persistent assets
+
+/// The preprovisioned local registry an `AssetRef` names.
+///
+/// An asset is installed and verified before a run; it is not a path, a URL or something a
+/// worker fetches. Nothing in this registry can be addressed by an `ArtifactRef`, and
+/// [`AssetRegistry::import`] hands back a fresh transient artifact rather than turning the
+/// asset into one.
+#[derive(Clone, Debug, Default)]
+pub struct AssetRegistry {
+    installed: BTreeMap<Id, (AssetRef, Vec<u8>)>,
+}
+
+impl AssetRegistry {
+    pub fn new() -> AssetRegistry {
+        AssetRegistry::default()
+    }
+
+    /// Installs content, verifying that it is the content the reference claims.
+    pub fn install(&mut self, asset: AssetRef, bytes: Vec<u8>) -> DomainResult<()> {
+        asset.validate().map_err(DomainError::invalid)?;
+        if asset.byte_length != bytes.len() as u64 {
+            return Err(DomainError::before(
+                ErrorCode::IdentityMismatch,
+                format!("asset {}: byteLength is not the installed length", asset.id),
+            ));
+        }
+        if asset.digest != digest_of_bytes(&bytes) {
+            return Err(DomainError::before(
+                ErrorCode::IdentityMismatch,
+                format!("asset {}: digest is not the installed content", asset.id),
+            ));
+        }
+        self.installed.insert(asset.id.clone(), (asset, bytes));
+        Ok(())
+    }
+
+    /// Resolves installed content. Identity and digest must both match: the same id with
+    /// another digest is a different asset, not an upgrade.
+    pub fn resolve(&self, asset: &AssetRef) -> DomainResult<&[u8]> {
+        match self.installed.get(&asset.id) {
+            Some((installed, bytes)) if installed == asset => Ok(bytes),
+            Some(_) => Err(DomainError::before(
+                ErrorCode::IdentityMismatch,
+                format!("asset {} is installed with another identity", asset.id),
+            )),
+            None => Err(DomainError::before(
+                ErrorCode::IdentityMismatch,
+                format!("asset {} is not installed", asset.id),
+            )),
+        }
+    }
+
+    pub fn contains(&self, asset: &AssetRef) -> bool {
+        self.resolve(asset).is_ok()
+    }
+
+    /// Imports installed content into the bus as a fresh immutable artifact.
+    ///
+    /// The artifact is sealed against the asset's digest, which is mandatory for a persistent
+    /// asset import, and its identity belongs to the current store incarnation. The asset
+    /// reference is unchanged and outlives it.
+    pub async fn import(
+        &self,
+        client: &flybus::Client,
+        asset: &AssetRef,
+    ) -> DomainResult<flybus::Artifact> {
+        let bytes = self.resolve(asset)?;
+        let mut writer = client
+            .artifacts()
+            .allocate(asset.byte_length, "application/octet-stream")
+            .await
+            .map_err(|e| store_error("allocate", &e.message))?;
+        writer
+            .write_all(bytes)
+            .map_err(|e| store_error("write", &e.to_string()))?;
+        let artifact = writer
+            .seal_with_digest(Some(asset.digest.clone()))
+            .await
+            .map_err(|e| store_error("seal", &e.message))?;
+        fly_session_types::media::check_imported_asset(asset, artifact.reference())
+            .map_err(|e| DomainError::before(ErrorCode::IdentityMismatch, e))?;
+        Ok(artifact)
+    }
+}

services/flysim/crates/fly-session/tests/media.rs

@@ -0,0 +1,862 @@
+//! MEDIA-01 acceptance: native observations on the bus artifact, and the presentation handoff.
+//!
+//! Every test here is one of the slice's acceptance bullets or one row of the domain retention
+//! table in `state-media-v1` section 3. The shape rules themselves are proved against the
+//! contract crate in `fly-session-types/tests/media_shapes.rs`; these prove the session's use
+//! of them: one shared image, spectators that cannot touch sensory state, a renderer that
+//! keeps its handle, and a persistent asset that is not a transient artifact.
+
+mod common;
+
+use std::collections::BTreeMap;
+use std::time::Duration;
+
+use serde_json::Map;
+
+use common::{Fixture, default_fixture, fixture, fly_a, fly_b, mode_fixture, within};
+use fly_session::environment::{
+    AUDIO_STREAM_ID, CHANNELS, EnvironmentFaults, SAMPLE_RATE, VIEW_HEIGHT, VIEW_WIDTH,
+    synthetic_asset,
+};
+use fly_session::harness::{ExecutionMode, HarnessConfig, Via};
+use fly_session::media::{
+    AssetRegistry, AudioSource, AudioTimelines, SensedView, Spectator, SpectatorFrame,
+    arena_frame, audio_attachment, detach_frame, view_attachment,
+};
+use fly_session::phase::Phase;
+use fly_session::types::*;
+use fly_session_types::media::{AudioTimeline, check_imported_asset, require_finite_samples};
+
+both_transports!(
+    one_shared_image_reaches_both_agents_through_owned_attachments,
+    a_spectator_cannot_corrupt_sensory_state,
+    a_slow_spectator_exhausts_only_its_own_credits,
+    delayed_rendering_retains_its_handle_after_the_message_drops,
+    a_declared_render_delay_repeats_o0_until_the_pipeline_fills,
+    an_extra_delayed_sensory_view_fails_the_step,
+    a_frame_of_the_wrong_length_fails_the_step,
+    overlapping_audio_fails_the_step,
+    one_audio_chunk_per_boundary_with_an_exact_sample_budget,
+    required_agent_input_is_never_coalesced_while_spectator_snapshots_are,
+    a_persistent_asset_and_a_transient_artifact_are_different_identities,
+    a_restored_audio_source_resumes_and_marks_the_discontinuity,
+    a_missing_audio_chunk_fails_the_step,
+    restored_timelines_need_every_declared_streams_position,
+    a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples,
+);
+
+all_modes!(
+    the_media_path_works_in_every_execution_mode,
+    every_media_fault_fires_in_every_execution_mode,
+);
+
+const STEPS: u64 = 3;
+
+/// Polls until `ok` holds, so a test never asserts a collection that has not happened yet.
+async fn until(what: &str, mut ok: impl FnMut() -> bool) {
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    while !ok() {
+        assert!(std::time::Instant::now() < deadline, "{what}: never happened");
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+}
+
+/// Drains a latest subscription until the snapshot for `boundary` arrives.
+///
+/// A latest subscription converges on the newest value rather than delivering every one, so a
+/// test that wants a particular committed boundary reads until it gets there.
+async fn frame_at(spectator: &mut Spectator, boundary: u64) -> SpectatorFrame {
+    for _ in 0..64 {
+        let frame = within("snapshot", spectator.take_frame())
+            .await
+            .expect("a snapshot");
+        if frame.boundary == boundary {
+            return frame;
+        }
+    }
+    panic!("the spectator never reached boundary {boundary}");
+}
+
+/// The views one agent read.
+///
+/// The sensor log is shared memory, so it is readable only where that agent lives. These tests
+/// run in the default in-process composition; the process-mode test below asserts the media
+/// path over the bus instead, which is what crosses a process boundary.
+fn sensed(f: &Fixture, agent_id: &Id) -> Vec<SensedView> {
+    f.harness
+        .sensor_log(agent_id)
+        .expect("this composition keeps its agents in this process")
+        .entries()
+}
+
+/// How many native frames the world rendered, for a world in this process.
+fn renders(f: &Fixture) -> u64 {
+    f.harness
+        .renders()
+        .expect("this composition keeps its world in this process")
+}
+
+fn boundaries(entries: &[SensedView]) -> Vec<u64> {
+    entries.iter().map(|e| e.boundary).collect()
+}
+
+fn produced(entries: &[SensedView]) -> Vec<u64> {
+    entries.iter().map(|e| e.produced_step).collect()
+}
+
+/// One image per boundary reaches both agents as an owned attachment, and the same handle is
+/// published for presentation. There is no second copy of the pixels anywhere.
+async fn one_shared_image_reaches_both_agents_through_owned_attachments(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+    within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+
+    // One render per boundary: forwarding the handle to two agents and to publication does not
+    // render or copy it again.
+    assert_eq!(
+        renders(&f),
+        STEPS + 1,
+        "one native frame per boundary, whatever the number of recipients"
+    );
+
+    let a = sensed(&f, &fly_a());
+    let b = sensed(&f, &fly_b());
+    assert_eq!(boundaries(&a), (0..=STEPS).collect::<Vec<_>>());
+    assert_eq!(a, b, "both agents read the same artifact and the same bytes");
+    assert_eq!(produced(&a), (0..=STEPS).collect::<Vec<_>>(), "no declared delay");
+
+    // The handle the agents were given is the handle presentation was published.
+    let published = frame_at(&mut spectator, STEPS).await;
+    assert_eq!(
+        published.agents,
+        vec![fly_a(), fly_b()],
+        "one snapshot carries the whole multi-agent session"
+    );
+    let last = a.last().expect("an entry per boundary");
+    assert_eq!(published.view.pixels.artifact_id, last.artifact_id);
+    assert_eq!(published.view.produced_step, last.produced_step);
+    let bytes = published.artifact.read_all().await.expect("the published frame");
+    assert_eq!(bytes.len() as u64, VIEW_WIDTH * VIEW_HEIGHT * 4);
+    assert_eq!(digest_of_bytes(&bytes), last.digest, "the bytes both agents read");
+
+    // The coordinator holds exactly one handle per declared view and stream at this boundary.
+    let handles = f.harness.coordinator.media_handles();
+    assert_eq!(handles.len(), 2, "one view and one audio chunk: {handles:?}");
+    assert!(handles.iter().any(|(name, r)| *name == view_attachment("arena")
+        && r.artifact_id == last.artifact_id));
+    f.shutdown().await;
+}
+
+/// A spectator reads committed snapshots and cannot touch what the agents sense: it has no
+/// authority over the environment, and its own reads leave sensory state exactly as produced.
+async fn a_spectator_cannot_corrupt_sensory_state(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+
+    // Naming the environment is not authority to drive it: a subscriber cannot call it.
+    let refused = observer
+        .call("env.arena", None, "Environment.Advance", Map::new(), &[])
+        .await;
+    assert!(
+        refused.is_err(),
+        "a spectator must not be able to call the environment"
+    );
+
+    within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+    // The spectator reads the snapshot and releases it while the session keeps stepping.
+    let frame = within("snapshot", spectator.take_frame()).await.expect("a snapshot");
+    let seen = frame.artifact.read_all().await.expect("readable");
+    drop(frame);
+    within("run more", f.harness.coordinator.run(STEPS)).await.unwrap();
+
+    let a = sensed(&f, &fly_a());
+    let b = sensed(&f, &fly_b());
+    assert_eq!(boundaries(&a), (0..=2 * STEPS).collect::<Vec<_>>());
+    assert_eq!(a, b);
+    assert_eq!(
+        f.harness.coordinator.committed_boundary(),
+        Some(2 * STEPS),
+        "the spectator changed nothing about the session's progress"
+    );
+    // What the spectator read was one of the frames the agents encoded, unchanged.
+    let digest = digest_of_bytes(&seen);
+    assert!(
+        a.iter().any(|entry| entry.digest == digest),
+        "a spectator sees the committed frame, and only reads it"
+    );
+    f.shutdown().await;
+}
+
+/// A slow spectator exhausts its own credits. New snapshots replace its queued value; the
+/// world never waits for it.
+async fn a_slow_spectator_exhausts_only_its_own_credits(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    // One credit and one queued value: the smallest spectator the bus allows.
+    let mut slow = Spectator::attach(&observer, &topic, 1).await.unwrap();
+    let mut fast = Spectator::attach(&observer, &topic, 2).await.unwrap();
+
+    // The slow one takes one snapshot and stops rendering, holding its only credit.
+    assert!(within("first snapshot", slow.hold_one()).await);
+    assert_eq!(slow.held(), 1);
+
+    let steps = 5;
+    within("run", f.harness.coordinator.run(steps)).await.unwrap();
+    assert_eq!(f.harness.coordinator.stats().advances, steps);
+
+    // With its credit in use it receives nothing more, and its queued value is replaced.
+    assert!(!slow.try_hold_one(), "a spectator out of credits gets nothing more");
+    assert_eq!(slow.seen(), 1);
+    slow.release();
+    let next = within("after release", slow.hold_one()).await;
+    assert!(next, "releasing its own delivery returns its own credit");
+    assert!(
+        slow.coalesced() > 0,
+        "new snapshots replaced the value queued for a spectator that was not reading"
+    );
+
+    // The session and the other spectator are untouched.
+    let latest = frame_at(&mut fast, steps).await;
+    assert_eq!(
+        latest.boundary, steps,
+        "the reading spectator reaches the latest boundary"
+    );
+    let a = sensed(&f, &fly_a());
+    assert_eq!(boundaries(&a), (0..=steps).collect::<Vec<_>>());
+    f.shutdown().await;
+}
+
+/// A renderer keeps its extracted handle after the message is gone, and after the boundary it
+/// came from has been replaced everywhere else.
+async fn delayed_rendering_retains_its_handle_after_the_message_drops(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+    within("step", f.harness.coordinator.run(1)).await.unwrap();
+
+    // Take the message, keep only the frame, and drop the message itself.
+    let message = within("snapshot", spectator.next_message()).await.expect("a snapshot");
+    let (view, artifact) = detach_frame(message).expect("a frame in the snapshot");
+
+    // Everything else moves on: the coordinator drops that boundary's handles, and the topic's
+    // retained value is replaced twice.
+    within("more steps", f.harness.coordinator.run(2)).await.unwrap();
+    let handles = f.harness.coordinator.media_handles();
+    assert!(
+        !handles.iter().any(|(_, r)| r.artifact_id == view.pixels.artifact_id),
+        "the session no longer holds the frame the renderer is still using"
+    );
+
+    // The rendering finishes now, long after its message is gone.
+    let bytes = artifact.read_all().await.expect("the guard kept the bytes alive");
+    assert_eq!(bytes.len() as u64, VIEW_WIDTH * VIEW_HEIGHT * 4);
+    let entries = sensed(&f, &fly_a());
+    let at_boundary = entries
+        .iter()
+        .find(|e| e.produced_step == view.produced_step)
+        .expect("the agents encoded this frame too");
+    assert_eq!(
+        digest_of_bytes(&bytes),
+        at_boundary.digest,
+        "the retained handle still reads exactly the frame that was published"
+    );
+    f.shutdown().await;
+}
+
+/// A declared render delay repeats `O[0]` while the pipeline fills, then advances one frame
+/// per boundary. The repetition is the same artifact, not a re-render.
+async fn a_declared_render_delay_repeats_o0_until_the_pipeline_fills(via: Via) {
+    let config = HarnessConfig {
+        observation_delay_steps: 2,
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("run", f.harness.coordinator.run(4)).await.unwrap();
+
+    let a = sensed(&f, &fly_a());
+    assert_eq!(boundaries(&a), vec![0, 1, 2, 3, 4]);
+    assert_eq!(
+        produced(&a),
+        vec![0, 0, 0, 1, 2],
+        "max(0, boundary - 2) at every boundary"
+    );
+    assert_eq!(a[0].artifact_id, a[1].artifact_id);
+    assert_eq!(a[0].artifact_id, a[2].artifact_id, "O[0] repeats while the delay fills");
+    assert_ne!(a[2].artifact_id, a[3].artifact_id, "then the pipeline advances");
+    assert_ne!(a[3].artifact_id, a[4].artifact_id);
+    // The world still renders once per boundary; the delay is a queue, not a missing frame.
+    assert_eq!(renders(&f), 5);
+    f.shutdown().await;
+}
+
+/// Beyond the declared delay, an extra-delayed sensory view is a step failure, never an
+/// arbitrary latest frame.
+async fn an_extra_delayed_sensory_view_fails_the_step(via: Via) {
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            stale_view_at_boundary: Some(2),
+            ..EnvironmentFaults::default()
+        },
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let failure = within("run", f.harness.coordinator.run(STEPS))
+        .await
+        .expect_err("a stale frame fails the transition");
+    assert_eq!(failure.error.code, ErrorCode::BufferInvalid);
+    assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert_eq!(
+        f.harness.coordinator.stats().advances,
+        1,
+        "the transition that met the stale frame committed nothing"
+    );
+    // The agents did not encode the stale frame.
+    let a = sensed(&f, &fly_a());
+    assert_eq!(boundaries(&a), vec![0, 1]);
+    f.shutdown().await;
+}
+
+/// A frame whose artifact is not `rowStride x height` bytes fails the step.
+async fn a_frame_of_the_wrong_length_fails_the_step(via: Via) {
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            truncated_view_at_boundary: Some(1),
+            ..EnvironmentFaults::default()
+        },
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let failure = within("run", f.harness.coordinator.run(1))
+        .await
+        .expect_err("a short frame fails the transition");
+    assert_eq!(failure.error.code, ErrorCode::BufferInvalid);
+    assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+    assert_eq!(f.harness.coordinator.stats().advances, 0);
+    f.shutdown().await;
+}
+
+/// Within an epoch, an audio chunk that starts inside the previous one is refused.
+async fn overlapping_audio_fails_the_step(via: Via) {
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            overlapping_audio_at_boundary: Some(2),
+            ..EnvironmentFaults::default()
+        },
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let failure = within("run", f.harness.coordinator.run(STEPS))
+        .await
+        .expect_err("an overlapping chunk fails the transition");
+    assert_eq!(failure.error.code, ErrorCode::BufferInvalid);
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    f.shutdown().await;
+}
+
+/// One chunk per boundary, with an exact sample budget, finite samples and the byte length the
+/// descriptor implies. Audio is published for presentation and never enters sensory input.
+async fn one_audio_chunk_per_boundary_with_an_exact_sample_budget(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+    within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+
+    // 48 kHz at 60 Hz is exactly 800 frames a step, and the positions are contiguous.
+    let per_step = SAMPLE_RATE / f.harness.config.step_hz;
+    let positions = f.harness.coordinator.audio_positions();
+    assert_eq!(positions[AUDIO_STREAM_ID], per_step * STEPS);
+
+    let observation = f.harness.coordinator.observation().expect("an observation").clone();
+    assert_eq!(observation.audio.len(), 1, "one chunk per boundary");
+    let chunk = &observation.audio[0];
+    assert_eq!(chunk.sample_frames, per_step);
+    assert_eq!(chunk.first_sample, per_step * (STEPS - 1));
+    assert!(!chunk.discontinuity, "an uninterrupted epoch has no discontinuity");
+    assert_eq!(chunk.samples.byte_length, per_step * CHANNELS * 4);
+    // Sensory input is pixels only: audio never becomes an agent's input.
+    assert!(observation.sensory_views.iter().all(|v| v.view_id == "arena"));
+
+    let published = frame_at(&mut spectator, STEPS).await;
+    let (published_chunk, artifact) = published.audio.first().expect("the published chunk");
+    assert_eq!(published_chunk.samples, chunk.samples);
+    assert_eq!(
+        artifact.reference().artifact_id,
+        chunk.samples.artifact_id,
+        "the same owned handle is published, not a copy"
+    );
+    let bytes = artifact.read_all().await.expect("the published chunk");
+    assert_eq!(bytes.len() as u64, per_step * CHANNELS * 4);
+    require_finite_samples(&bytes).expect("native samples are finite f32");
+    assert_eq!(
+        published.audio.len(),
+        1,
+        "the attachment list names every published chunk"
+    );
+    assert_eq!(audio_attachment(AUDIO_STREAM_ID), "audio.arena");
+    f.shutdown().await;
+}
+
+/// A declared stream that produces no chunk for a transition is a step failure, not a silently
+/// shorter epoch.
+async fn a_missing_audio_chunk_fails_the_step(via: Via) {
+    let config = HarnessConfig {
+        environment_faults: EnvironmentFaults {
+            omit_audio_at_boundary: Some(2),
+            ..EnvironmentFaults::default()
+        },
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let failure = within("run", f.harness.coordinator.run(STEPS))
+        .await
+        .expect_err("a missing chunk fails the transition");
+    assert_eq!(failure.error.code, ErrorCode::BufferInvalid);
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    // Boundary 0 has no preceding interval, so its empty audio list is not a missing chunk.
+    let mut clean = default_fixture(via).await;
+    within("bootstrap", clean.harness.coordinator.bootstrap()).await.unwrap();
+    assert!(clean.harness.coordinator.observation().unwrap().audio.is_empty());
+    clean.shutdown().await;
+    f.shutdown().await;
+}
+
+/// Restored timelines resume every declared stream at its recorded position. A stream with no
+/// recorded position is refused, not restarted at zero.
+async fn restored_timelines_need_every_declared_streams_position(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    within("run", f.harness.coordinator.run(1)).await.unwrap();
+    let descriptor = f.harness.coordinator.descriptor().expect("a descriptor").clone();
+    let observation = f.harness.coordinator.observation().expect("an observation").clone();
+
+    // A fresh epoch accepts the transition's chunk and advances its position.
+    let mut fresh = AudioTimelines::fresh(&descriptor);
+    fresh
+        .accept(&descriptor, &observation)
+        .expect("the first chunk of a fresh epoch");
+    assert_eq!(fresh.accepted(AUDIO_STREAM_ID), 1);
+    let chunk = observation.audio.first().expect("a chunk").clone();
+    assert_eq!(
+        fresh.positions()[AUDIO_STREAM_ID],
+        chunk.first_sample + chunk.sample_frames
+    );
+
+    // A restore with nothing recorded for a declared stream is an error, not sample zero.
+    let missing = AudioTimelines::restored(&descriptor, &BTreeMap::new())
+        .expect_err("a declared stream needs its preserved position");
+    assert_eq!(missing.code, ErrorCode::IncompatibleState);
+
+    // With the position recorded, the restored epoch resumes there and its first chunk must
+    // mark the discontinuity.
+    let positions = f.harness.coordinator.audio_positions();
+    let mut resumed = observation.clone();
+    let resumed_chunk = resumed.audio.first_mut().expect("a chunk");
+    resumed_chunk.first_sample = positions[AUDIO_STREAM_ID];
+    resumed_chunk.discontinuity = false;
+    let mut restored = AudioTimelines::restored(&descriptor, &positions).expect("positions");
+    restored
+        .accept(&descriptor, &resumed)
+        .expect_err("the first chunk after a restore marks discontinuity");
+    resumed.audio.first_mut().expect("a chunk").discontinuity = true;
+    let mut restored = AudioTimelines::restored(&descriptor, &positions).expect("positions");
+    restored
+        .accept(&descriptor, &resumed)
+        .expect("the restored epoch resumes at the preserved position");
+    assert_eq!(restored.accepted(AUDIO_STREAM_ID), 1);
+    f.shutdown().await;
+}
+
+/// A world cadence that does not divide the sample rate still produces whole samples, with the
+/// remainder carried rather than rounded: seven steps of a 7 Hz world are exactly one second.
+async fn a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples(via: Via) {
+    let config = HarnessConfig {
+        step_hz: 7,
+        ..HarnessConfig::default()
+    };
+    let mut f = fixture(via, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+
+    // 48000 / 7 is 6857.14..., so no chunk can be the exact share of a second.
+    let mut frames = Vec::new();
+    for step in 1..=7 {
+        within("step", f.harness.coordinator.run(1)).await.unwrap();
+        let chunk = f
+            .harness
+            .coordinator
+            .observation()
+            .expect("an observation")
+            .audio
+            .first()
+            .expect("a chunk")
+            .clone();
+        assert_eq!(chunk.first_sample, frames.iter().sum::<u64>());
+        assert!(
+            chunk.sample_frames == 6_857 || chunk.sample_frames == 6_858,
+            "step {step} produced {} frames",
+            chunk.sample_frames
+        );
+        assert_eq!(chunk.samples.byte_length, chunk.sample_frames * CHANNELS * 4);
+        frames.push(chunk.sample_frames);
+    }
+    assert_eq!(
+        frames.iter().sum::<u64>(),
+        SAMPLE_RATE,
+        "seven steps of a 7 Hz world are exactly one second of samples: {frames:?}"
+    );
+    assert_eq!(f.harness.coordinator.audio_positions()[AUDIO_STREAM_ID], SAMPLE_RATE);
+
+    // The published chunk is readable and finite whatever the cadence.
+    let published = frame_at(&mut spectator, 7).await;
+    let (chunk, artifact) = published.audio.first().expect("the published chunk");
+    let bytes = artifact.read_all().await.expect("the published chunk");
+    assert_eq!(bytes.len() as u64, chunk.sample_frames * CHANNELS * 4);
+    require_finite_samples(&bytes).expect("native samples are finite f32");
+    f.shutdown().await;
+}
+
+/// The media path itself is mode-agnostic: one native image per boundary, forwarded to every
+/// agent as an owned attachment and published once for presentation, whether the participants
+/// are tasks on one runtime, threads with their own runtimes, or separate processes.
+///
+/// What a *test* can see differs by mode, and this test only asserts what crosses a process
+/// boundary. An agent validates that each attachment is the artifact its payload names and
+/// reads the pixels before it commits, so a session that commits every boundary in process
+/// mode has carried one shared image across that boundary through the store, not through
+/// shared memory. The sensor log and the render counter are shared memory, so they are
+/// asserted where they exist and their absence is asserted where they do not.
+async fn the_media_path_works_in_every_execution_mode(mode: ExecutionMode) {
+    // A declared render delay as well, so the option reaches a world in another process.
+    let config = HarnessConfig {
+        observation_delay_steps: 1,
+        ..HarnessConfig::default()
+    };
+    let mut f = mode_fixture(mode, config).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap();
+    within("run", f.harness.coordinator.run(STEPS)).await.unwrap();
+
+    // Every agent read its attachment and committed, in every mode.
+    assert_eq!(f.harness.coordinator.committed_boundary(), Some(STEPS));
+
+    // The coordinator holds one view handle and one audio handle for this boundary, and the
+    // observation names exactly those artifacts.
+    let observation = f.harness.coordinator.observation().expect("an observation").clone();
+    let handles = f.harness.coordinator.media_handles();
+    assert_eq!(handles.len(), 2, "one view and one chunk: {handles:?}");
+    let view = observation.sensory_views.first().expect("a required view");
+    assert_eq!(
+        view.produced_step,
+        STEPS - 1,
+        "the declared one-step delay reached the world in {mode:?} mode"
+    );
+    assert!(handles.iter().any(|(name, r)| *name == view_attachment(&view.view_id)
+        && r.artifact_id == view.pixels.artifact_id));
+
+    // The same object is what presentation was published, and its bytes read back at the
+    // declared shape through an ordinary subscription.
+    let published = frame_at(&mut spectator, STEPS).await;
+    assert_eq!(published.view.pixels.artifact_id, view.pixels.artifact_id);
+    assert_eq!(published.view.produced_step, view.produced_step);
+    let pixels = published.artifact.read_all().await.expect("the published frame");
+    assert_eq!(pixels.len() as u64, VIEW_WIDTH * VIEW_HEIGHT * 4);
+    let (chunk, audio) = published.audio.first().expect("the published chunk");
+    let samples = audio.read_all().await.expect("the published chunk");
+    assert_eq!(samples.len() as u64, chunk.sample_frames * CHANNELS * 4);
+    require_finite_samples(&samples).expect("native samples are finite f32");
+
+    // The shared-memory instrumentation exists exactly where the participants do.
+    match (mode, f.harness.sensor_log(&fly_a()), f.harness.renders()) {
+        (ExecutionMode::Process, sensors, renders) => {
+            assert!(sensors.is_none() && renders.is_none(), "not observable from here");
+        }
+        (_, Some(sensors), Some(renders)) => {
+            let a = sensors.entries();
+            let b = f.harness.sensor_log(&fly_b()).expect("in this process").entries();
+            assert_eq!(a, b, "both agents read the same artifact and the same bytes");
+            assert_eq!(boundaries(&a), (0..=STEPS).collect::<Vec<_>>());
+            assert_eq!(
+                digest_of_bytes(&pixels),
+                a.last().expect("an entry per boundary").digest,
+                "the published bytes are the bytes the agents encoded"
+            );
+            assert_eq!(renders, STEPS + 1, "one render per boundary");
+        }
+        (mode, _, _) => panic!("{mode:?} keeps its participants in this process"),
+    }
+    f.shutdown().await;
+}
+
+/// Every media fault fires wherever the world runs, which is what proves the wiring rather
+/// than assuming it.
+///
+/// A fault reaches a world in another process only as a `--flag` on its command line, so a
+/// renamed or dropped flag would make these faults silently stop firing. Here each one has to
+/// fail its transition in every execution mode; the parser refuses an unknown flag, so a
+/// mismatch fails the launch instead of turning into a no-op.
+async fn every_media_fault_fires_in_every_execution_mode(mode: ExecutionMode) {
+    let cases: [(&str, EnvironmentFaults, u64); 4] = [
+        (
+            "an extra-delayed view",
+            EnvironmentFaults {
+                stale_view_at_boundary: Some(2),
+                ..EnvironmentFaults::default()
+            },
+            2,
+        ),
+        (
+            "a frame of the wrong length",
+            EnvironmentFaults {
+                truncated_view_at_boundary: Some(1),
+                ..EnvironmentFaults::default()
+            },
+            1,
+        ),
+        (
+            "a missing audio chunk",
+            EnvironmentFaults {
+                omit_audio_at_boundary: Some(2),
+                ..EnvironmentFaults::default()
+            },
+            2,
+        ),
+        (
+            "an overlapping audio chunk",
+            EnvironmentFaults {
+                overlapping_audio_at_boundary: Some(2),
+                ..EnvironmentFaults::default()
+            },
+            2,
+        ),
+    ];
+    for (what, faults, steps) in cases {
+        let config = HarnessConfig {
+            environment_faults: faults,
+            ..HarnessConfig::default()
+        };
+        let mut f = mode_fixture(mode, config).await;
+        within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+        let outcome = within("run", f.harness.coordinator.run(steps)).await;
+        let failure = match outcome {
+            Err(failure) => failure,
+            Ok(reports) => {
+                panic!("{what} did not fail the transition in {mode:?} mode: {reports:?}")
+            }
+        };
+        assert_eq!(
+            failure.error.code,
+            ErrorCode::BufferInvalid,
+            "{what} in {mode:?} mode: {failure}"
+        );
+        assert_eq!(f.harness.coordinator.phase(), Phase::Failed);
+        assert_eq!(
+            f.harness.coordinator.stats().advances,
+            steps - 1,
+            "{what} in {mode:?} mode committed the boundaries before it and no more"
+        );
+        f.shutdown().await;
+    }
+}
+
+/// The retention table: a required agent input is retained through encoding and Commit with no
+/// coalescing, while a spectator's snapshots are a latest subscription with finite credits.
+async fn required_agent_input_is_never_coalesced_while_spectator_snapshots_are(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let observer = f.harness.observer().await.unwrap();
+    let topic = f.harness.coordinator.topics().snapshots.clone();
+    let mut spectator = Spectator::attach(&observer, &topic, 1).await.unwrap();
+    assert!(within("first snapshot", spectator.hold_one()).await);
+
+    let steps = 5;
+    within("run", f.harness.coordinator.run(steps)).await.unwrap();
+
+    // The spectator's queue coalesced while it was not reading.
+    spectator.release();
+    within("after release", spectator.hold_one()).await;
+    assert!(spectator.coalesced() > 0);
+    assert!(
+        spectator.seen() < steps + 1,
+        "a latest subscription does not deliver every boundary to a slow reader"
+    );
+
+    // Every agent's required input arrived once per boundary, in order, with nothing dropped
+    // or replaced.
+    for agent in [fly_a(), fly_b()] {
+        let entries = sensed(&f, &agent);
+        assert_eq!(
+            boundaries(&entries),
+            (0..=steps).collect::<Vec<_>>(),
+            "{agent} encoded every boundary exactly once"
+        );
+    }
+    f.shutdown().await;
+}
+
+/// A persistent `AssetRef` names installed content; a transient `ArtifactRef` names live bytes
+/// in one store. Importing the asset makes a new artifact, and collecting that artifact leaves
+/// the asset installed.
+async fn a_persistent_asset_and_a_transient_artifact_are_different_identities(via: Via) {
+    let mut f = default_fixture(via).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let client = f.harness.observer().await.unwrap();
+
+    let body = "counter-arena-backend-v1";
+    let asset = synthetic_asset("counter-arena-backend", body);
+    let mut registry = AssetRegistry::new();
+    registry
+        .install(asset.clone(), body.as_bytes().to_vec())
+        .expect("installed content matches its reference");
+    // Installing content that is not what the reference claims is refused.
+    let mut wrong = asset.clone();
+    wrong.byte_length += 1;
+    registry
+        .install(wrong, body.as_bytes().to_vec())
+        .expect_err("an asset reference is its content's identity");
+
+    let before = f.harness.router().stats().sealed_artifacts;
+    let artifact = registry.import(&client, &asset).await.expect("the import");
+    assert_eq!(
+        f.harness.router().stats().sealed_artifacts,
+        before + 1,
+        "the import is a new object in the store"
+    );
+
+    // The identities are different, and the content is the same.
+    assert_ne!(artifact.reference().artifact_id, asset.id);
+    assert_eq!(artifact.reference().byte_length, asset.byte_length);
+    assert_eq!(artifact.reference().digest.as_deref(), Some(asset.digest.as_str()));
+    check_imported_asset(&asset, artifact.reference()).expect("the import carries the content");
+    assert_eq!(
+        artifact.read_all().await.expect("readable"),
+        body.as_bytes(),
+        "the imported artifact is the installed bytes"
+    );
+
+    // The transient artifact is collected with its last handle; the asset is still installed.
+    drop(artifact);
+    let router = f.harness.router().clone();
+    until("the imported artifact is collected", || {
+        router.stats().sealed_artifacts == before
+    })
+    .await;
+    assert!(registry.contains(&asset));
+    assert_eq!(
+        registry.resolve(&asset).expect("still installed"),
+        body.as_bytes(),
+        "a persistent asset outlives the bus objects imported from it"
+    );
+    f.shutdown().await;
+}
+
+/// A restored epoch resumes the preserved sample position, and its first chunk marks the
+/// discontinuity that says so. The producer and the validator agree about which epoch it is.
+async fn a_restored_audio_source_resumes_and_marks_the_discontinuity(via: Via) {
+    let f = default_fixture(via).await;
+    let client = f.harness.observer().await.unwrap();
+    let descriptor = fly_session::environment::CounterEnvironment::audio_descriptor();
+    let step = hz(f.harness.config.step_hz).expect("a positive cadence");
+    let resumed_at = 2_400;
+
+    let mut source = AudioSource::restored_at(descriptor.clone(), resumed_at);
+    let (first, _first_handle) = source
+        .produce(&client, &step, 3)
+        .await
+        .expect("the restored source produces");
+    assert_eq!(first.first_sample, resumed_at, "the sample position is preserved");
+    assert!(first.discontinuity, "the first chunk after a restore marks it");
+    let (second, _second_handle) = source
+        .produce(&client, &step, 3)
+        .await
+        .expect("the next chunk");
+    assert!(!second.discontinuity, "only the first chunk of the epoch marks it");
+    assert_eq!(second.first_sample, resumed_at + first.sample_frames);
+
+    // The validator accepts exactly this sequence under a restored timeline, and refuses it
+    // under a fresh one: the flag is what distinguishes the two epochs.
+    let mut restored = AudioTimeline::restored_at(&descriptor, resumed_at);
+    restored.accept(&first, &descriptor).expect("the restored epoch");
+    restored.accept(&second, &descriptor).expect("and its next chunk");
+    // A fresh episode at the audio origin refuses it: a restore preserves the sample
+    // position, and the position is what distinguishes the two epochs. The flag is required
+    // after a restore and free at an origin, so it cannot carry that distinction by itself.
+    let mut fresh = AudioTimeline::fresh(&descriptor, 0);
+    fresh
+        .accept(&first, &descriptor)
+        .expect_err("a fresh episode starts at its own origin, not a resumed position");
+    f.shutdown().await;
+}
+
+/// The sample budget is exact when the cadence does not divide the sample rate: 8 kHz at 60 Hz
+/// is 133, 133, 134 and the total after three steps is exactly 400.
+#[test]
+fn the_sample_budget_is_exact_when_the_cadence_does_not_divide_the_rate() {
+    let descriptor = fly_session_types::media::AudioDescriptor {
+        stream_id: "arena".into(),
+        sample_rate: 8_000,
+        channels: 1,
+    };
+    let mut source = AudioSource::new(descriptor, 0);
+    let step = hz(60).expect("a positive cadence");
+    let frames: Vec<u64> = (0..6)
+        .map(|_| source.frames_for_step(&step).expect("an exact budget"))
+        .collect();
+    assert_eq!(frames, vec![133, 133, 134, 133, 133, 134]);
+    assert_eq!(frames.iter().sum::<u64>(), 800, "8000 samples in a tenth of a second");
+
+    // 48 kHz at 60 Hz divides exactly.
+    let exact = fly_session_types::media::AudioDescriptor {
+        stream_id: "arena".into(),
+        sample_rate: SAMPLE_RATE,
+        channels: CHANNELS,
+    };
+    let mut source = AudioSource::new(exact, 0);
+    assert_eq!(source.frames_for_step(&step).unwrap(), 800);
+}
+
+/// The native frame is a real pattern with no padded rows: `rowStride` is `4 x width`, the
+/// rows are top-left first, and consecutive boundaries differ.
+#[test]
+fn the_native_frame_is_top_left_rgba8_with_no_padding() {
+    let descriptor = fly_session::environment::CounterEnvironment::view_descriptor(0);
+    let frame = arena_frame(&descriptor, 5, 3);
+    assert_eq!(frame.len() as u64, descriptor.row_stride * descriptor.height);
+    assert_eq!(descriptor.row_stride, descriptor.width * 4);
+    // Every pixel is opaque and carries the world counter in its red channel.
+    for pixel in frame.chunks_exact(4) {
+        assert_eq!(pixel[0], 5);
+        assert_eq!(pixel[3], 255);
+    }
+    assert_ne!(
+        arena_frame(&descriptor, 5, 3),
+        arena_frame(&descriptor, 5, 4),
+        "consecutive boundaries are different images"
+    );
+    assert_ne!(
+        arena_frame(&descriptor, 5, 3),
+        arena_frame(&descriptor, 6, 3),
+        "the counter changes the image"
+    );
+}

services/flysim/crates/fly-session/tests/session.rs

@@ -274,7 +274,11 @@ async fn the_committed_snapshot_names_the_boundary_that_just_ended(via: Via) {
         // The frame the snapshot names travels as an owned attachment.
         if step > 0 {
             let frame = message.artifact("view.arena").expect("the published frame");
-            assert_eq!(frame.reference().byte_length, 4 * 4 * 4);
+            assert_eq!(
+                frame.reference().byte_length,
+                fly_session::environment::VIEW_WIDTH * fly_session::environment::VIEW_HEIGHT * 4,
+                "the published frame is the environment native frame"
+            );
         }
     }
     assert_eq!(boundaries[0], (0, false), "boundary 0 has no decision or control");

services/flysim/crates/flybus/tests/sol_review_races.rs

@@ -714,6 +714,13 @@ async fn caller_disconnect_cleanup_works_before_and_after_consumption() {
     .await;
     caller.close().await;
     drop(pending);
+    // `Client::close` waits for this client to stop; the router marks the call detached when
+    // *it* observes the disconnect, in its own connection task. Asserting the reply routing
+    // before that is a race: under load the reply reaches the router first and is routed to a
+    // connection that is already closing. Nothing escapes -- teardown releases those roots --
+    // but `routed` is then true. The contract sentence is about a reply to an already detached
+    // call, so the test waits for the teardown it is talking about.
+    e.settle("caller-a disconnected", |s| s.connections == 1).await;
     assert!(!responder.reply(obj(json!({})), &[]).await.unwrap());
     drop(responder);
     e.settle("retained responder retired", |s| {