diff --git a/services/flysim/crates/fly-session-types/src/media.rs b/services/flysim/crates/fly-session-types/src/media.rs index 5f2a1d8..71a0856 100644 --- a/services/flysim/crates/fly-session-types/src/media.rs +++ b/services/flysim/crates/fly-session-types/src/media.rs @@ -366,6 +366,163 @@ 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 and is not a + /// discontinuity, because nothing preceded it. + 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 + )); + } + if chunk.discontinuity != self.restored { + return err(if self.restored { + format!( + "AudioTimeline {}: the first chunk after a restore marks discontinuity", + self.stream_id + ) + } else { + format!( + "AudioTimeline {}: the first chunk at the episode's audio origin is not a 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> { let views = list(f, key, 0, MAX_VIEWS, ViewRef::from_json)?; diff --git a/services/flysim/crates/fly-session-types/tests/media_shapes.rs b/services/flysim/crates/fly-session-types/tests/media_shapes.rs new file mode 100644 index 0000000..29c0f81 --- /dev/null +++ b/services/flysim/crates/fly-session-types/tests/media_shapes.rs @@ -0,0 +1,390 @@ +//! 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); + let mut fresh = AudioTimeline::fresh(&descriptor, 0); + fresh + .accept(&audio_ref(&descriptor, 0, 800, true), &descriptor) + .expect_err("the episode's first chunk is not a discontinuity"); + 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"); + 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 = asset + .to_json() + .as_object() + .expect("an object") + .keys() + .cloned() + .collect(); + let artifact_keys: Vec = 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"); +} diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index 68a621b..aa47f5e 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -203,6 +203,8 @@ pub struct AgentConfig { pub incarnation_id: Id, pub tick_duration: RationalNs, pub warmup_ticks: u64, + /// Records every view this agent read, so a test can see which artifact reached it. + pub sensors: crate::media::SensorLog, pub faults: AgentFaults, } @@ -296,6 +298,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 { diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 3f31c87..def9d5d 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -14,6 +14,7 @@ use std::collections::{BTreeMap, BTreeSet}; use serde_json::{Map, Value, json}; use crate::clock::Pacing; +use crate::media::{self, AudioTimelines}; use crate::phase::{Phase, PhaseMachine}; use crate::rpc::{self, DomainReply, Serials, WorkerRef}; use crate::task::{ActionExecutor, Task}; @@ -182,6 +183,15 @@ pub struct Coordinator { views: BTreeMap, /// Holds on the boundary the world has just reached, before it is committed. pending_views: BTreeMap, + /// 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, + pending_audio: BTreeMap, + /// 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, serials: Serials, topics: Topics, pacing: Option, @@ -233,6 +243,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, @@ -276,6 +293,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 { + self.timelines.positions() + } + pub fn episode_request(&self) -> Option<&EpisodeRequest> { self.episode.as_ref() } @@ -476,6 +508,7 @@ impl Coordinator { }; let worker = self.environment.clone(); let scope = self.scope(0); + let want = self.media_names.clone(); let reply = self .call( &worker, @@ -483,7 +516,7 @@ impl Coordinator { Some(scope), object(params.to_json()), &[], - &["view.arena".to_owned()], + &want, ) .await?; let result: EnvironmentInitializeResult = @@ -516,7 +549,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())); @@ -1113,10 +1156,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 = 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?; @@ -1132,7 +1184,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; @@ -1420,7 +1474,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}")); let injected = self.injections.at_step == k; @@ -1518,7 +1572,9 @@ impl Coordinator { Ok(result) => result, Err(e) => return Err(self.fail_now(e, "advance")), }; - 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 @@ -1553,6 +1609,7 @@ impl Coordinator { None => None, }; self.pending_views.clear(); + self.pending_audio.clear(); let replay = self .resolve( &worker, @@ -1573,7 +1630,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) } @@ -1648,43 +1707,17 @@ impl Coordinator { "step-result", )); } - // 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 - // descriptors; requiring each declared view to be there at all is the coordinator's - // Phase C check, so it is made here. - for view in &descriptor.views { - let want = required_produced_step(view, result.observation.boundary); - let got = result - .observation - .sensory_views - .iter() - .find(|given| given.view_id == view.view_id); - match got { - Some(given) if given.produced_step == want => {} - Some(_) => { - 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", - )); - } - } + // A missing required sensory input is never silently replaced by an older frame, + // and neither is one produced further back than the declared delay allows. The + // contract's validator checks the views that are present against their descriptors; + // requiring each declared view to be there at all is this Phase C check. + if let Err(e) = media::check_required_views(descriptor, &result.observation) { + return Err(self.fail_now(e, "step-result")); + } + // Audio has no sensory role here, but its chunks still cannot overlap or go backwards + // inside an epoch, and a stale one must not reach presentation as current. + if let Err(e) = self.timelines.accept(descriptor, &result.observation) { + return Err(self.fail_now(e, "step-result")); } if let Err(e) = result.observation.validate_against(descriptor) { return Err(self.fail_now( @@ -1693,7 +1726,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 => {} _ => { @@ -1708,6 +1741,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(()) } @@ -2088,11 +2140,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::>(), }); - 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, diff --git a/services/flysim/crates/fly-session/src/environment.rs b/services/flysim/crates/fly-session/src/environment.rs index deb8d21..6b3b212 100644 --- a/services/flysim/crates/fly-session/src/environment.rs +++ b/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. -pub const VIEW_WIDTH: u64 = 4; -pub const VIEW_HEIGHT: u64 = 4; +/// The arena's one view: a small native RGBA8 image. +pub const VIEW_ID: &str = "arena"; +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, + /// 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, + /// Seal a frame one row short at this boundary, so its artifact length is not + /// `rowStride x height`. + pub truncated_view_at_boundary: Option, + /// Leave the audio chunk out of the result at this boundary. + pub omit_audio_at_boundary: Option, + /// Emit an audio chunk that starts before the previous chunk ended. + pub overlapping_audio_at_boundary: Option, } #[derive(Clone, Debug)] @@ -36,6 +58,10 @@ pub struct EnvironmentConfig { /// The world's fixed reduced step duration. 60 Hz is `1/60` s. pub step_duration: RationalNs, pub ports: Vec, + /// 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. + pub renders: RenderCounter, pub faults: EnvironmentFaults, } @@ -52,6 +78,10 @@ pub struct CounterEnvironment { world_time: RationalNs, advances: u64, batches: BTreeSet, + pipeline: Option, + audio: Option, + /// The frame served at the previous boundary, kept only so a fault can serve it again. + previous_view: Option<(ViewRef, flybus::Artifact)>, } impl CounterEnvironment { @@ -67,6 +97,9 @@ impl CounterEnvironment { world_time: RationalNs::ZERO, advances: 0, batches: BTreeSet::new(), + pipeline: None, + audio: None, + previous_view: None, config, } } @@ -101,15 +134,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, } } @@ -119,10 +162,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(), ), @@ -137,8 +181,10 @@ impl CounterEnvironment { }) .collect(), inspection_schema: inspection_schema(), - views: vec![CounterEnvironment::view_descriptor()], - audio: Vec::new(), + views: vec![CounterEnvironment::view_descriptor( + self.config.observation_delay_steps, + )], + audio: vec![CounterEnvironment::audio_descriptor()], recovery: Recovery::ExactCheckpoint, determinism: Determinism::FixedBuild, }; @@ -146,73 +192,76 @@ impl CounterEnvironment { Ok(descriptor) } - /// Seals one immutable native frame for the current counter and returns the handle. - async fn render( - &self, - 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)) - } - + /// Renders this boundary's native media and returns the observation with its owned + /// handles. The same immutable object serves the sensory and the broadcast view; nothing + /// is rendered twice and no second copy of the pixels exists. 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 (views, attachments) = if omit { - (Vec::new(), Vec::new()) + let boundary = self.boundary; + let counter = self.counter; + 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?; - let name = format!("view.{}", view.view_id); - (vec![view], vec![(name, artifact)]) - }; + pipeline.render(ctx.client, boundary, counter).await?; + } + 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)) } @@ -268,6 +317,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); diff --git a/services/flysim/crates/fly-session/src/harness.rs b/services/flysim/crates/fly-session/src/harness.rs index 429ed59..4547a01 100644 --- a/services/flysim/crates/fly-session/src/harness.rs +++ b/services/flysim/crates/fly-session/src/harness.rs @@ -18,6 +18,7 @@ use flybus::{ use crate::agent::{AgentConfig, AgentFaults, FakeAgentWorker, synthetic_profile}; use crate::coordinator::{AgentSlot, Coordinator}; use crate::environment::{CounterEnvironment, EnvironmentConfig, EnvironmentFaults}; +use crate::media::{RenderCounter, SensorLog}; use crate::rpc::WorkerRef; use crate::task::{ActionExecutor, CounterTask, IdentityExecutor, Terminal}; // `crate::types` is this crate's facade over the shared `fly-session-types` crate; the @@ -55,6 +56,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, } @@ -82,6 +85,7 @@ impl Default for HarnessConfig { tick_ms: 1, warmup_ticks: 10, terminal: Terminal::Never, + observation_delay_steps: 0, environment_faults: EnvironmentFaults::default(), } } @@ -152,6 +156,10 @@ pub struct SessionHarness { pub agents: BTreeMap, pub config: HarnessConfig, pub via: Via, + /// How many native frames the environment has actually rendered. + pub renders: RenderCounter, + /// What each agent read out of its sensory attachments. + pub sensors: BTreeMap, connector: Connector, observers: Mutex>, } @@ -207,6 +215,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 = config + .agents + .iter() + .map(|spec| (spec.agent_id.clone(), SensorLog::new())) + .collect(); // The environment first: it owns the world and the descriptor. let env_client = connector.client(ENV_CLIENT).await?; @@ -221,6 +235,8 @@ impl SessionHarness { incarnation_id: id("arena-inc-1"), step_duration, ports: config.agents.iter().map(|a| a.port_id.clone()).collect(), + observation_delay_steps: config.observation_delay_steps, + renders: renders.clone(), faults: config.environment_faults.clone(), }), ); @@ -242,6 +258,7 @@ impl SessionHarness { .expect("an agent id plus a suffix is an Id"), tick_duration, warmup_ticks: config.warmup_ticks, + sensors: sensors[&spec.agent_id].clone(), faults: spec.faults.clone(), }), ); @@ -280,6 +297,8 @@ impl SessionHarness { agents, config, via, + renders, + sensors, connector, observers: Mutex::new(Vec::new()), }) @@ -345,6 +364,7 @@ impl SessionHarness { incarnation_id, tick_duration, warmup_ticks: self.config.warmup_ticks, + sensors: self.sensor_log(agent_id), faults: spec.faults, }), ); @@ -352,6 +372,17 @@ impl SessionHarness { Ok(restarted) } + /// What one agent read out of its sensory attachments, in order. + pub fn sensor_log(&self, agent_id: &Id) -> SensorLog { + self.sensors.get(agent_id).cloned().unwrap_or_default() + } + + /// How many native frames the environment rendered. Forwarding one image to several + /// recipients does not render it again. + pub fn renders(&self) -> u64 { + self.renders.count() + } + /// The agent worker's progress counter, which is its fake model's mutation count. pub fn agent_mutations(&self, agent_id: &Id) -> u64 { self.agents.get(agent_id).map(WorkerHandle::progress_counter).unwrap_or_default() @@ -367,7 +398,8 @@ impl SessionHarness { /// Stops every worker and closes the router. pub async fn shutdown(self) { - let SessionHarness { coordinator, environment, agents, connector, observers, .. } = self; + let SessionHarness { coordinator, environment, agents, connector, observers, .. } = + self; drop(coordinator); environment.stop().await; for (_, handle) in agents { diff --git a/services/flysim/crates/fly-session/src/lib.rs b/services/flysim/crates/fly-session/src/lib.rs index 7594b52..b3885fd 100644 --- a/services/flysim/crates/fly-session/src/lib.rs +++ b/services/flysim/crates/fly-session/src/lib.rs @@ -27,6 +27,7 @@ pub mod coordinator; pub mod dedup; pub mod environment; pub mod harness; +pub mod media; pub mod phase; pub mod rpc; pub mod task; diff --git a/services/flysim/crates/fly-session/src/media.rs b/services/flysim/crates/fly-session/src/media.rs new file mode 100644 index 0000000..d2c603b --- /dev/null +++ b/services/flysim/crates/fly-session/src/media.rs @@ -0,0 +1,810 @@ +//! 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); + +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 { + 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() + } + + /// Whether `boundary` is still inside the declared pipeline delay, where the contract + /// allows `O[0]` to repeat. + pub fn is_bootstrap_repeat(&self, boundary: u64) -> bool { + boundary > 0 && boundary <= self.descriptor.observation_delay_steps + } +} + +/// 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 { + 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 { + 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 { + 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, +) -> ( + BTreeMap, + BTreeMap, +) { + 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 { + 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's chunk sequence, one timeline per stream and epoch. +#[derive(Clone, Debug, Default)] +pub struct AudioTimelines(BTreeMap); + +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. + pub fn restored( + descriptor: &EnvironmentDescriptor, + positions: &BTreeMap, + ) -> AudioTimelines { + AudioTimelines( + descriptor + .audio + .iter() + .map(|stream| { + let at = positions.get(&stream.stream_id).copied().unwrap_or_default(); + ( + stream.stream_id.clone(), + AudioTimeline::restored_at(stream, at), + ) + }) + .collect(), + ) + } + + /// 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 { + 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>>); + +/// 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 { + self.0.lock().expect("the sensor log is never poisoned").clone() + } + + /// The artifacts this agent read, in order. + pub fn artifact_ids(&self) -> Vec { + 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, + 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, + 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 { + 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 { + 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 { + 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 { + 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::().ok())?; + let media = payload.get("media")?; + let views = media.get("views")?.as_array()?; + let view = ViewRef::from_json(views.first()?).ok()?; + let audio = media + .get("audio")? + .as_array()? + .iter() + .filter_map(|v| AudioRef::from_json(v).ok()) + .filter_map(|chunk| { + let artifact = message.artifact(&audio_attachment(&chunk.stream_id)).ok()?; + Some((chunk, artifact)) + }) + .collect(); + 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)>, +} + +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) -> 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 { + 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) + } +} diff --git a/services/flysim/crates/fly-session/tests/media.rs b/services/flysim/crates/fly-session/tests/media.rs new file mode 100644 index 0000000..5f063fc --- /dev/null +++ b/services/flysim/crates/fly-session/tests/media.rs @@ -0,0 +1,527 @@ +//! 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::time::Duration; + +use serde_json::Map; + +use common::{default_fixture, fixture, fly_a, fly_b, within}; +use fly_session::environment::{ + AUDIO_STREAM_ID, CHANNELS, EnvironmentFaults, SAMPLE_RATE, VIEW_HEIGHT, VIEW_WIDTH, + synthetic_asset, +}; +use fly_session::harness::{HarnessConfig, Via}; +use fly_session::media::{ + AssetRegistry, AudioSource, 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::{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, +); + +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}"); +} + +fn boundaries(entries: &[SensedView]) -> Vec { + entries.iter().map(|e| e.boundary).collect() +} + +fn produced(entries: &[SensedView]) -> Vec { + 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!( + f.harness.renders(), + STEPS + 1, + "one native frame per boundary, whatever the number of recipients" + ); + + let a = f.harness.sensor_log(&fly_a()).entries(); + let b = f.harness.sensor_log(&fly_b()).entries(); + assert_eq!(boundaries(&a), (0..=STEPS).collect::>()); + assert_eq!(a, b, "both agents read the same artifact and the same bytes"); + assert_eq!(produced(&a), (0..=STEPS).collect::>(), "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 = f.harness.sensor_log(&fly_a()).entries(); + let b = f.harness.sensor_log(&fly_b()).entries(); + assert_eq!(boundaries(&a), (0..=2 * STEPS).collect::>()); + 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 = f.harness.sensor_log(&fly_a()).entries(); + assert_eq!(boundaries(&a), (0..=steps).collect::>()); + 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 = f.harness.sensor_log(&fly_a()).entries(); + 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 = f.harness.sensor_log(&fly_a()).entries(); + 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!(f.harness.renders(), 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 = f.harness.sensor_log(&fly_a()).entries(); + 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; +} + +/// 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 = f.harness.sensor_log(&agent).entries(); + assert_eq!( + boundaries(&entries), + (0..=steps).collect::>(), + "{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_eq!( + registry.resolve(&asset).expect("still installed"), + body.as_bytes(), + "a persistent asset outlives the bus objects imported from it" + ); + 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 = (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::(), 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" + ); +} diff --git a/services/flysim/crates/fly-session/tests/session.rs b/services/flysim/crates/fly-session/tests/session.rs index feb5d7f..209cc23 100644 --- a/services/flysim/crates/fly-session/tests/session.rs +++ b/services/flysim/crates/fly-session/tests/session.rs @@ -270,7 +270,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");