diff --git a/docs/design/session-framework/bus-conformance.md b/docs/design/session-framework/bus-conformance.md index cd90df6..00e72b3 100644 --- a/docs/design/session-framework/bus-conformance.md +++ b/docs/design/session-framework/bus-conformance.md @@ -119,7 +119,7 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means | Requirement | Status | Code | Test | | --- | --- | --- | --- | | `call-` with increasing serials per connected client; reused or retired ids are rejected, never executed again | conforms: a syntactically valid id advances the watermark even when admission is refused | `router/state.rs::op_call` (`call_watermark`) | `tests/sol_review_races.rs::rejected_call_id_still_advances_monotonic_watermark`, `tests/rpc.rs::raw_call_ids_and_forged_replies` (both) | -| Reconnecting creates a new incarnation rather than reviving old calls | conforms | `router/state.rs::{hello, disconnect}` | `tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` | +| Reconnecting creates a new incarnation rather than reviving old calls | conforms | `router/state.rs::{hello, disconnect}` | `tests/conformance_routing.rs::unpinned_call_after_incarnation_replacement_reaches_the_new_holder` (both), `tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` for the old calls half (its synchronisation was fixed on 2026-09-22; see "A flaky test and what it was measuring") | | An RPC targets one registered service, not a broadcast subject | conforms | `router/state.rs::op_call` | `tests/rpc.rs::request_reply_roundtrip` (both) | | First-dispatch FIFO per caller and service; responses may complete out of order and correlate by callId | conforms | `router/state.rs::{Svc::queue, dispatch_rpc}` | `tests/rpc.rs::fifo_dispatch_and_out_of_order_completion` (both), `tests/conformance_routing.rs::out_of_order_replies_correlate_across_concurrent_callers` (both) | | A service dispatcher can answer status concurrently with a long mutation | conforms | `router/state.rs::dispatch_rpc` (in-flight credits, not one-at-a-time) | `tests/bus_acceptance.rs::a_status_rpc_responds_while_another_handler_is_delayed` (both) | @@ -412,6 +412,45 @@ What the numbers do and do not say: cores against 0.33 to 0.46), because the clients own the copies. A thread that exits between two samples takes its CPU with it, so the router figure is a floor. +## A flaky test and what it was measuring, 2026-09-22 (MEDIA-01) + +`tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` +failed intermittently on `main` after the bus slice merged. Reproduced here at +**37 failures in 240 runs** (four parallel loops of 60, debug, on the loaded dev VM), always +on the same line and always the same way: `responder.reply(...)` returned `routed:true` where +the test asserted `false`. + +The mechanism is a synchronisation gap in the test, not a routing defect. +`router/state.rs::op_reply` returns `routed:false` only when `call.detached` is set, and for a +disconnected caller that flag is set by `router/state.rs::disconnect`, which the router runs +when **its** connection task reads EOF. `Client::close` documents what it waits for — "flushes +queued releases, closes the connection and waits until the reader has stopped ... the router +releases what they owned" — which is the client side only. So after `close()` returns, the +router may not have torn the caller's connection down yet, and a reply that reaches it first is +routed to a connection that is already closing. Nothing escapes: `disconnect` then releases +that connection's roots along with the queued result, which is why the test's own later +`settle` calls always passed. Only the `routed` flag, read one step too early, was wrong. + +The fix is in the test: it now waits for the teardown it is talking about +(`e.settle("caller-a disconnected", |s| s.connections == 1)`) before asserting the +reply-to-a-detached-call sentence of section 6. That is the same bounded +poll-until-the-router-settles the rest of the file already uses for router-side consequences; +no sleep, no timing constant, and the assertion now has the precondition its contract sentence +names. **360 runs after the fix, 0 failures** (240 debug, 120 release). + +Two other intermittent failures were seen in the same sweep and are **not** fixed here, since +they belong to the bus slice rather than to this one: + +- `tests/example_demo.rs::the_example_shows_a_counter_rpc_an_observer_and_a_held_frame`, + 2 failures in 40 standalone runs plus 1 in 12 full-suite runs. It prints + "while the frame is held: 1 artifact(s), 2 root(s)" instead of 1 root: the producer's hold + release is queued on the control lane and had not been applied when the example read the + counts. The same shape of gap, in the guide deliverable's printed output. +- `tests/integration.rs::unix_socket::session_over_one_router`, 1 failure in 12 full-suite + runs and 0 in 40 standalone runs, at the assertion that the deliberately slow consumer + skipped snapshots. Under load it kept up, so the assertion is a timing claim about the + machine. + ## Contradictions Two, both inside bus-v1, both minor, neither resolved by changing code. Both were referred to diff --git a/docs/design/session-framework/state-media-v1.md b/docs/design/session-framework/state-media-v1.md index a8c7a86..2669786 100644 --- a/docs/design/session-framework/state-media-v1.md +++ b/docs/design/session-framework/state-media-v1.md @@ -65,6 +65,23 @@ the sample position relative to the episode's configured audio origin, with inte firstSample/sampleRate. Crash restore preserves sample position under a new epoch; first chunk marks discontinuity. Within an epoch, chunks cannot overlap or go backwards. +**Amendment, 2026-09-22 (MEDIA-01).** Two readings of the paragraphs above, made explicit +because they are now enforced: + +- The bootstrap window is exactly the boundaries where `max(0, boundary - observationDelaySteps)` + is zero, that is `boundary <= observationDelaySteps`. Inside it the repeated `O[0]` is the + **same artifact**, not a fresh render of the same scene; outside it the producing boundary + advances one per step, and a frame from any other boundary -- older or newer -- is a step + failure. A producer therefore keeps a queue of `observationDelaySteps + 1` frames and nothing + more, so there is no older frame available to substitute. +- Within an epoch, `discontinuity` marks a range the stream actually skipped. The first chunk + after a restore marks it, and a later chunk may mark it when it starts past where the previous + chunk ended; a chunk that continues the previous one exactly is continuous by construction and + its flag is refused. Without that reading the restore rule is advisory, because a stream could + set the flag on every chunk and satisfy it by accident. The requirement is one-directional: a + fresh epoch's first chunk **may** mark a discontinuity, because section 6's recovery + establishes a fresh timeline and publishes one. + The environment provides **native game output**. Sensor transformations belong to the agent profile. Resizing for viewers, overlays, composition, audio mixing/resampling, encoding, browser delivery and streaming belong to the application/presentation layer. No bus or diff --git a/services/flysim/crates/fly-session-types/src/media.rs b/services/flysim/crates/fly-session-types/src/media.rs index 5f2a1d8..5b5fbfd 100644 --- a/services/flysim/crates/fly-session-types/src/media.rs +++ b/services/flysim/crates/fly-session-types/src/media.rs @@ -366,6 +366,161 @@ impl DomainType for AudioRef { } } +// --------------------------------------------------------------------------------------------- +// Epoch audio sequencing (state-media-v1 section 2) + +/// One audio stream's chunk sequence within one epoch. +/// +/// The contract's three sentences about sequencing are all here: `firstSample` identifies the +/// sample position relative to the episode's configured audio origin; crash restore preserves +/// that position under a new epoch and the first chunk marks `discontinuity`; within an epoch +/// chunks cannot overlap or go backwards. +/// +/// A timeline belongs to one epoch. A restore starts a new one with +/// [`AudioTimeline::restored_at`], which is what makes the first chunk's discontinuity flag +/// checkable rather than advisory. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AudioTimeline { + stream_id: String, + start_sample: u64, + restored: bool, + next_sample: u64, + accepted: u64, +} + +impl AudioTimeline { + /// A fresh episode: the first chunk starts at the configured audio origin. Its + /// discontinuity flag is free, because a reset or a recovery establishes a fresh timeline + /// and does publish a discontinuity. + pub fn fresh(descriptor: &AudioDescriptor, origin: u64) -> AudioTimeline { + AudioTimeline { + stream_id: descriptor.stream_id.clone(), + start_sample: origin, + restored: false, + next_sample: origin, + accepted: 0, + } + } + + /// A new epoch after a crash restore: the sample position is preserved, and the first + /// chunk of this epoch must mark `discontinuity`. + pub fn restored_at(descriptor: &AudioDescriptor, sample: u64) -> AudioTimeline { + AudioTimeline { + stream_id: descriptor.stream_id.clone(), + start_sample: sample, + restored: true, + next_sample: sample, + accepted: 0, + } + } + + /// Where the next chunk may start. A chunk starting earlier overlaps or goes backwards. + pub fn next_sample(&self) -> u64 { + self.next_sample + } + + /// How many chunks this epoch has accepted. + pub fn accepted(&self) -> u64 { + self.accepted + } + + /// Validates one chunk's shape and its place in the sequence, then advances the timeline. + /// + /// A rejected chunk does not advance anything, so a caller that fails its step does not + /// leave the timeline believing the chunk was played. + pub fn accept(&mut self, chunk: &AudioRef, descriptor: &AudioDescriptor) -> Result<()> { + if chunk.stream_id != self.stream_id { + return err(format!( + "AudioTimeline {}: chunk names stream {:?}", + self.stream_id, chunk.stream_id + )); + } + chunk.validate_against(descriptor)?; + if self.accepted == 0 { + if chunk.first_sample != self.start_sample { + return err(format!( + "AudioTimeline {}: the first chunk of this epoch must start at sample {}, not {}", + self.stream_id, self.start_sample, chunk.first_sample + )); + } + // Only one direction is stated: the first chunk after a restore marks the + // discontinuity. A fresh epoch's first chunk may mark one too -- sections 6 and 7 + // have recovery and episode reset publishing a discontinuity on a fresh timeline -- + // so the flag is required after a restore and left free at an origin. + if self.restored && !chunk.discontinuity { + return err(format!( + "AudioTimeline {}: the first chunk after a restore marks discontinuity", + self.stream_id + )); + } + } else { + if chunk.first_sample < self.next_sample { + return err(format!( + "AudioTimeline {}: firstSample {} overlaps or goes backwards; the previous chunk ends at {}", + self.stream_id, chunk.first_sample, self.next_sample + )); + } + // Derived from the sentence above: within an epoch the only discontinuity a chunk + // can carry is a gap it actually skipped. A chunk that continues the previous one + // exactly is continuous by construction. + if chunk.discontinuity && chunk.first_sample == self.next_sample { + return err(format!( + "AudioTimeline {}: a chunk continuing the previous one is not a discontinuity", + self.stream_id + )); + } + } + self.next_sample = chunk + .first_sample + .checked_add(chunk.sample_frames) + .ok_or_else(|| { + crate::scalar::wire_err(format!( + "AudioTimeline {}: firstSample + sampleFrames overflows U64", + self.stream_id + )) + })?; + self.accepted += 1; + Ok(()) + } +} + +// --------------------------------------------------------------------------------------------- +// Persistent assets against transient artifacts (state-media-v1 sections 1 and 3) + +/// Checks that a transient artifact carries the bytes of an installed asset. +/// +/// Persistent [`AssetRef`](crate::workers::AssetRef) and transient +/// [`ArtifactRef`] are different identities and never convert into one another: an asset names +/// installed release content in a preprovisioned registry, while an artifact names live bytes +/// in one store incarnation and resolves only through an owned handle. Importing an asset +/// produces a **new** artifact identity, which is why this checks content rather than identity. +/// +/// The digest is mandatory here: state-media-v1 section 1 makes content digests optional on +/// transient live frames and mandatory on persistent asset import. +pub fn check_imported_asset( + asset: &crate::workers::AssetRef, + imported: &ArtifactRef, +) -> Result<()> { + asset.validate()?; + if imported.byte_length != asset.byte_length { + return err(format!( + "imported asset {}: the artifact is {} bytes, the asset is {}", + asset.id, imported.byte_length, asset.byte_length + )); + } + match &imported.digest { + Some(d) if *d == asset.digest => Ok(()), + Some(_) => err(format!( + "imported asset {}: the artifact's digest is not the asset's content", + asset.id + )), + None => err(format!( + "imported asset {}: a persistent asset import must carry a content digest", + asset.id + )), + } +} + /// Reads a bounded, unique-by-`viewId` list of view refs. pub fn view_list(f: &mut Fields<'_>, key: &'static str) -> Result> { 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..f834989 --- /dev/null +++ b/services/flysim/crates/fly-session-types/tests/media_shapes.rs @@ -0,0 +1,395 @@ +//! MEDIA-01 shape rules: every sentence of `state-media-v1` section 2 that a descriptor, a +//! reference or a chunk sequence can be checked against on its own. +//! +//! These are the contract-level halves of the slice's acceptance bullets -- bad strides, bad +//! lengths, bad producing times and the audio rules -- and the type-level distinction between +//! a persistent `AssetRef` and a transient `ArtifactRef`. + +use fly_session_types::ArtifactRef; +use fly_session_types::media::{ + AudioDescriptor, AudioRef, AudioTimeline, ViewDescriptor, ViewRef, check_imported_asset, + require_finite_samples, +}; +use fly_session_types::scalar::DomainType; +use fly_session_types::workers::AssetRef; + +fn artifact(byte_length: u64, content_type: &str) -> ArtifactRef { + ArtifactRef { + store_id: "store-a".into(), + artifact_id: "art-1".into(), + generation: 1, + byte_length, + content_type: content_type.into(), + digest: None, + } +} + +fn view_descriptor(width: u64, height: u64, delay: u64) -> ViewDescriptor { + ViewDescriptor { + view_id: "arena".into(), + width, + height, + row_stride: width * 4, + pixel_aspect_numerator: 1, + pixel_aspect_denominator: 1, + observation_delay_steps: delay, + } +} + +fn view_ref(descriptor: &ViewDescriptor, produced_step: u64, bytes: u64) -> ViewRef { + ViewRef { + view_id: descriptor.view_id.clone(), + produced_step, + pixels: artifact(bytes, "image/x-rgba8"), + } +} + +fn audio_descriptor(sample_rate: u64, channels: u64) -> AudioDescriptor { + AudioDescriptor { + stream_id: "arena".into(), + sample_rate, + channels, + } +} + +fn audio_ref( + descriptor: &AudioDescriptor, + first_sample: u64, + frames: u64, + discontinuity: bool, +) -> AudioRef { + AudioRef { + stream_id: descriptor.stream_id.clone(), + first_sample, + sample_frames: frames, + samples: artifact(frames * descriptor.channels * 4, "audio/x-f32le"), + discontinuity, + } +} + +// ----------------------------------------------------------------------------------------- +// Bad strides + +/// `rowStride` is exactly `4 x width`; v1 has no padded rows. +#[test] +fn a_padded_row_stride_is_refused() { + let good = view_descriptor(32, 24, 0); + good.validate().expect("4 x width is the only stride"); + + let mut padded = good.clone(); + padded.row_stride = 32 * 4 + 16; + padded.validate().expect_err("a padded row is not readable in v1"); + + let mut narrow = good.clone(); + narrow.row_stride = 32 * 3; + narrow.validate().expect_err("a stride under 4 x width is refused"); + + // The same rule through the wire form, where a hand-written descriptor arrives. + let mut json = good.to_json(); + json["rowStride"] = serde_json::json!(32 * 4 + 4); + ViewDescriptor::from_json(&json).expect_err("a padded stride is refused when read"); +} + +/// Dimensions are integers 1..=4096, and `pixelAspect` parts positive integers <=65535. +#[test] +fn dimensions_pixel_aspect_and_delay_have_stated_bounds() { + for (width, height) in [(0, 24), (32, 0), (4097, 24), (32, 4097)] { + let mut d = view_descriptor(32, 24, 0); + d.width = width; + d.height = height; + d.row_stride = width.max(1) * 4; + d.validate().expect_err("dimensions are 1..=4096"); + } + view_descriptor(1, 1, 0).validate().expect("1x1 is inside the bounds"); + view_descriptor(4096, 4096, 0) + .validate() + .expect("4096x4096 is inside the bounds"); + + for (numerator, denominator) in [(0, 1), (1, 0), (65_536, 1), (1, 65_536)] { + let mut d = view_descriptor(32, 24, 0); + d.pixel_aspect_numerator = numerator; + d.pixel_aspect_denominator = denominator; + d.validate().expect_err("pixelAspect parts are 1..=65535"); + } + + let mut d = view_descriptor(32, 24, 9); + d.validate().expect_err("observationDelaySteps is 0..=8"); + d.observation_delay_steps = 8; + d.validate().expect("eight steps of delay are allowed"); +} + +/// Only top-left RGBA8 exists in v1; another format is a media-schema change. +#[test] +fn only_rgba8_is_a_readable_view_format() { + let descriptor = view_descriptor(32, 24, 0); + let mut json = descriptor.to_json(); + json["format"] = serde_json::json!("rgb8"); + ViewDescriptor::from_json(&json).expect_err("rgb8 is not a v1 format"); + json["format"] = serde_json::json!("rgba8"); + ViewDescriptor::from_json(&json).expect("rgba8 is the v1 format"); +} + +// ----------------------------------------------------------------------------------------- +// Bad lengths + +/// A frame's artifact is exactly `rowStride x height` bytes. +#[test] +fn a_frame_whose_length_is_not_stride_times_height_is_refused() { + let descriptor = view_descriptor(32, 24, 0); + let exact = descriptor.frame_bytes(); + assert_eq!(exact, 32 * 4 * 24); + + view_ref(&descriptor, 7, exact) + .validate_against(&descriptor, None) + .expect("the exact frame length is accepted"); + for wrong in [exact - 1, exact + 1, exact - 32 * 4, exact * 2] { + view_ref(&descriptor, 7, wrong) + .validate_against(&descriptor, None) + .expect_err("only rowStride x height is the frame length"); + } +} + +// ----------------------------------------------------------------------------------------- +// Bad producing times + +/// A required sensory view is produced at exactly `max(0, boundary - observationDelaySteps)`. +#[test] +fn a_view_produced_at_the_wrong_boundary_is_refused() { + let descriptor = view_descriptor(32, 24, 2); + let bytes = descriptor.frame_bytes(); + assert_eq!(descriptor.required_produced_step(10), 8); + + view_ref(&descriptor, 8, bytes) + .validate_against(&descriptor, Some(10)) + .expect("the declared delay is exactly two steps"); + // One frame later than the delay allows, and one frame older: both are step failures, + // not an arbitrary latest frame. + view_ref(&descriptor, 9, bytes) + .validate_against(&descriptor, Some(10)) + .expect_err("an under-delayed frame is refused"); + view_ref(&descriptor, 7, bytes) + .validate_against(&descriptor, Some(10)) + .expect_err("an extra-delayed frame is refused"); +} + +/// Bootstrap may repeat `O[0]` until the declared pipeline delay fills, and only until then. +#[test] +fn bootstrap_repeats_the_first_frame_until_the_pipeline_delay_fills() { + let descriptor = view_descriptor(32, 24, 3); + let bytes = descriptor.frame_bytes(); + // Boundaries 0..=3 all require the frame produced at 0. + for boundary in 0..=3 { + assert_eq!(descriptor.required_produced_step(boundary), 0); + view_ref(&descriptor, 0, bytes) + .validate_against(&descriptor, Some(boundary)) + .expect("O[0] repeats while the pipeline fills"); + } + // From boundary 4 the pipeline is full and O[0] is a stale frame. + assert_eq!(descriptor.required_produced_step(4), 1); + view_ref(&descriptor, 0, bytes) + .validate_against(&descriptor, Some(4)) + .expect_err("the repetition ends when the delay is filled"); + view_ref(&descriptor, 1, bytes) + .validate_against(&descriptor, Some(4)) + .expect("boundary 4 requires the frame produced at 1"); +} + +// ----------------------------------------------------------------------------------------- +// Audio shapes + +/// sampleRate is 8000..=192000, channels 1..=8 and sampleFrames 0..=192000. +#[test] +fn audio_rate_channels_and_frames_have_stated_bounds() { + audio_descriptor(8_000, 1).validate().expect("8 kHz mono is the floor"); + audio_descriptor(192_000, 8).validate().expect("192 kHz 8ch is the ceiling"); + audio_descriptor(7_999, 2).validate().expect_err("under 8 kHz is refused"); + audio_descriptor(192_001, 2).validate().expect_err("over 192 kHz is refused"); + audio_descriptor(48_000, 0).validate().expect_err("zero channels are refused"); + audio_descriptor(48_000, 9).validate().expect_err("nine channels are refused"); + + let descriptor = audio_descriptor(48_000, 2); + let mut chunk = audio_ref(&descriptor, 0, 192_000, false); + chunk.validate().expect("192000 frames is the per-chunk ceiling"); + chunk.sample_frames = 192_001; + chunk.validate().expect_err("over 192000 frames in one chunk is refused"); +} + +/// A chunk's artifact is exactly `sampleFrames x channels x 4` bytes. +#[test] +fn an_audio_chunk_length_is_frames_times_channels_times_four() { + let descriptor = audio_descriptor(48_000, 2); + let chunk = audio_ref(&descriptor, 0, 800, false); + assert_eq!(chunk.samples.byte_length, 800 * 2 * 4); + chunk + .validate_against(&descriptor) + .expect("the exact chunk length is accepted"); + + let mut wrong = chunk.clone(); + wrong.samples = artifact(800 * 2 * 4 - 4, "audio/x-f32le"); + wrong + .validate_against(&descriptor) + .expect_err("a short chunk is refused"); + + // The same frames at another channel count are a different number of bytes. + let mono = audio_descriptor(48_000, 1); + let mut wrong_channels = chunk.clone(); + wrong_channels.stream_id = mono.stream_id.clone(); + wrong_channels + .validate_against(&mono) + .expect_err("stereo bytes are not a mono chunk"); +} + +/// Samples are finite f32. +#[test] +fn a_non_finite_sample_is_refused() { + let mut bytes = Vec::new(); + for value in [0.0f32, -0.5, 0.75, 1.0] { + bytes.extend_from_slice(&value.to_le_bytes()); + } + require_finite_samples(&bytes).expect("finite samples are accepted"); + + for bad in [f32::NAN, f32::INFINITY, f32::NEG_INFINITY] { + let mut broken = bytes.clone(); + broken.extend_from_slice(&bad.to_le_bytes()); + require_finite_samples(&broken).expect_err("a non-finite sample is refused"); + } + require_finite_samples(&bytes[..5]).expect_err("a partial sample is refused"); +} + +/// Within an epoch, chunks cannot overlap or go backwards. +#[test] +fn chunks_cannot_overlap_or_go_backwards_within_an_epoch() { + let descriptor = audio_descriptor(48_000, 2); + let mut timeline = AudioTimeline::fresh(&descriptor, 0); + timeline + .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor) + .expect("the first chunk starts at the origin"); + assert_eq!(timeline.next_sample(), 800); + timeline + .accept(&audio_ref(&descriptor, 800, 800, false), &descriptor) + .expect("the second chunk continues the first"); + assert_eq!(timeline.next_sample(), 1_600); + + let mut overlapping = timeline.clone(); + overlapping + .accept(&audio_ref(&descriptor, 1_599, 800, false), &descriptor) + .expect_err("a chunk that starts inside the previous one is refused"); + let mut backwards = timeline.clone(); + backwards + .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor) + .expect_err("a chunk that goes backwards is refused"); + // A rejected chunk leaves the timeline where it was. + assert_eq!(overlapping.next_sample(), 1_600); + assert_eq!(overlapping.accepted(), 2); + + // A gap is forward, so it is allowed; it is the one place a later chunk may mark a + // discontinuity. + timeline + .accept(&audio_ref(&descriptor, 2_000, 800, true), &descriptor) + .expect("a forward gap is not an overlap"); + timeline + .accept(&audio_ref(&descriptor, 2_800, 800, true), &descriptor) + .expect_err("a chunk that continues the previous one is not a discontinuity"); +} + +/// Crash restore preserves the sample position under a new epoch, and its first chunk marks +/// the discontinuity. +#[test] +fn the_first_chunk_after_a_restore_marks_discontinuity() { + let descriptor = audio_descriptor(48_000, 2); + // The requirement is one-directional. A fresh epoch's first chunk may mark a + // discontinuity -- a recovery or an episode reset establishes a fresh timeline and + // publishes one -- so both flags are accepted at an origin. + let mut reset = AudioTimeline::fresh(&descriptor, 0); + reset + .accept(&audio_ref(&descriptor, 0, 800, true), &descriptor) + .expect("a reset episode's first chunk may mark the discontinuity it published"); + let mut fresh = AudioTimeline::fresh(&descriptor, 0); + fresh + .accept(&audio_ref(&descriptor, 0, 800, false), &descriptor) + .expect("the episode's first chunk continues nothing"); + + // The restored epoch resumes at the preserved position. + let mut restored = AudioTimeline::restored_at(&descriptor, 800); + restored + .accept(&audio_ref(&descriptor, 800, 800, false), &descriptor) + .expect_err("the first chunk after a restore marks discontinuity"); + let mut restored = AudioTimeline::restored_at(&descriptor, 800); + restored + .accept(&audio_ref(&descriptor, 0, 800, true), &descriptor) + .expect_err("the restored position is preserved, not reset"); + restored + .accept(&audio_ref(&descriptor, 800, 800, true), &descriptor) + .expect("the restored epoch resumes at its preserved sample position"); + assert_eq!(restored.next_sample(), 1_600); + restored + .accept(&audio_ref(&descriptor, 1_600, 800, false), &descriptor) + .expect("the chunks after it are ordinary"); +} + +// ----------------------------------------------------------------------------------------- +// Persistent assets against transient artifacts + +/// An `AssetRef` and an `ArtifactRef` are different identities with different fields, and +/// neither is readable as the other. +#[test] +fn an_asset_ref_is_not_a_transient_artifact_ref() { + let asset = AssetRef { + id: "counter-arena-backend".into(), + digest: "a".repeat(64), + byte_length: 24, + format: "fly-config-v1".into(), + }; + let imported = ArtifactRef { + store_id: "store-a".into(), + artifact_id: "art-9".into(), + generation: 1, + byte_length: 24, + content_type: "application/octet-stream".into(), + digest: Some("a".repeat(64)), + }; + + // Identity fields do not overlap: the asset has no store and the artifact has no format. + let asset_keys: Vec = 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 187e3b1..a52822f 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -206,6 +206,11 @@ pub struct AgentConfig { /// The thread allocation the launcher started this worker within. `workers-v1` requires /// `Agent.Initialize`'s `workerThreads` to lie inside it. pub worker_threads: usize, + /// Records every view this agent read, so a test can see which artifact reached it. + /// + /// It is this process's log: an agent with a process of its own writes to its own copy, + /// which the supervisor cannot read. `SessionHarness::sensor_log` says so with `None`. + pub sensors: crate::media::SensorLog, pub faults: AgentFaults, } @@ -299,6 +304,15 @@ impl FakeAgentWorker { format!("view {} is the wrong length", view.view_id), )); } + // What this agent read, from the bytes it read: the artifact it was given and the + // digest of its content. + self.config.sensors.record(crate::media::SensedView { + boundary: input.boundary, + view_id: view.view_id.clone(), + artifact_id: artifact.reference().artifact_id.clone(), + produced_step: view.produced_step, + digest: digest_of_bytes(&bytes), + }); total += i64::from(bytes.first().copied().unwrap_or_default()); } if let Some(structured) = &input.structured { diff --git a/services/flysim/crates/fly-session/src/cli.rs b/services/flysim/crates/fly-session/src/cli.rs index d492144..1f7bc68 100644 --- a/services/flysim/crates/fly-session/src/cli.rs +++ b/services/flysim/crates/fly-session/src/cli.rs @@ -21,7 +21,9 @@ use std::process::ExitCode; use crate::agent::AgentFaults; use crate::environment::EnvironmentFaults; -use crate::launcher::{AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, serve_one}; +use crate::launcher::{ + AgentLaunch, EnvironmentLaunch, ExecutionMode, Started, flags, serve_one, +}; use crate::types::*; const USAGE: &str = "\ @@ -69,9 +71,12 @@ pub fn main() -> ExitCode { let command = command.to_string_lossy().into_owned(); let rest: Vec = args.map(|a| a.to_string_lossy().into_owned()).collect(); let result = match command.as_str() { - "agent" | "environment" => Options::parse(&rest).and_then(|o| serve(&command, &o)), - "measure" => Options::parse(&rest).and_then(|o| measure(&o)), - "measure-row" => Options::parse(&rest).and_then(|o| measure_row(&o)), + "agent" => Options::parse(&rest, &[flags::COMMON, flags::AGENT_ONLY]) + .and_then(|o| serve(&command, &o)), + "environment" => Options::parse(&rest, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .and_then(|o| serve(&command, &o)), + "measure" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure(&o)), + "measure-row" => Options::parse(&rest, &[flags::MEASURE]).and_then(|o| measure_row(&o)), "--help" | "-h" | "help" => { print!("{USAGE}"); return ExitCode::SUCCESS; @@ -92,13 +97,22 @@ pub fn main() -> ExitCode { struct Options(BTreeMap); impl Options { - fn parse(args: &[String]) -> Result { + /// Reads the options of one command, refusing any flag that command does not have. + /// + /// `allowed` comes from [`crate::launcher::flags`], the same constants the launcher + /// writes the argv from. An unknown flag is an error naming it rather than a value that + /// is quietly ignored: a renamed option must fail the launch, not turn into a no-op that + /// no test notices. + fn parse(args: &[String], allowed: &[&[&str]]) -> Result { let mut out = BTreeMap::new(); let mut iter = args.iter(); while let Some(flag) = iter.next() { let Some(name) = flag.strip_prefix("--") else { return Err(format!("expected an option, found {flag:?}")); }; + if !allowed.iter().any(|set| set.contains(&name)) { + return Err(format!("unknown option --{name} for this command")); + } let value = iter .next() .ok_or_else(|| format!("option --{name} needs a value"))?; @@ -158,43 +172,53 @@ impl Options { /// Serves one worker until `Worker.Shutdown`, then exits. fn serve(role: &str, options: &Options) -> Result<(), String> { - let socket = options.path("socket")?; - let store_root = options.path("store-root")?; - let client_id = options.required("client-id")?.to_owned(); - let service = options.required("service")?.to_owned(); - let threads = options.usize("threads", 1)?; + let socket = options.path(flags::SOCKET)?; + let store_root = options.path(flags::STORE_ROOT)?; + let client_id = options.required(flags::CLIENT_ID)?.to_owned(); + let service = options.required(flags::SERVICE)?.to_owned(); + let threads = options.usize(flags::THREADS, 1)?; if threads == 0 { return Err("--threads must be at least 1".to_owned()); } - let session_id = options.id("session")?; - let incarnation_id = options.id("incarnation")?; + let session_id = options.id(flags::SESSION)?; + let incarnation_id = options.id(flags::INCARNATION)?; let what = match role { "agent" => Started::Agent(AgentLaunch { session_id, - agent_id: options.id("agent")?, - port_id: options.id("port")?, + agent_id: options.id(flags::AGENT)?, + port_id: options.id(flags::PORT)?, incarnation_id, - tick_duration: options.rational("tick-numerator", "tick-denominator")?, - warmup_ticks: options.u64("warmup-ticks", 0)?, + tick_duration: options.rational(flags::TICK_NUMERATOR, flags::TICK_DENOMINATOR)?, + warmup_ticks: options.u64(flags::WARMUP_TICKS, 0)?, worker_threads: threads, + // This process's own log. The supervisor reads what crosses the bus, not this. + sensors: crate::media::SensorLog::new(), faults: AgentFaults { - fail_commit_at_step: options.opt_u64("fail-commit-at-step")?, - prepare_delay_ms: options.u64("prepare-delay-ms", 0)?, - commit_delay_ms: options.u64("commit-delay-ms", 0)?, + fail_commit_at_step: options.opt_u64(flags::FAIL_COMMIT_AT_STEP)?, + prepare_delay_ms: options.u64(flags::PREPARE_DELAY_MS, 0)?, + commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?, }, client_id: client_id.clone(), service: service.clone(), }), _ => Started::Environment(EnvironmentLaunch { session_id, - worker_id: options.id("worker")?, + worker_id: options.id(flags::WORKER)?, incarnation_id, - step_duration: options.rational("step-numerator", "step-denominator")?, - ports: parse_ports(options.required("ports")?)?, + step_duration: options.rational(flags::STEP_NUMERATOR, flags::STEP_DENOMINATOR)?, + ports: parse_ports(options.required(flags::PORTS)?)?, worker_threads: threads, + observation_delay_steps: options.u64(flags::OBSERVATION_DELAY_STEPS, 0)?, + renders: crate::media::RenderCounter::new(), faults: EnvironmentFaults { - advance_delay_ms: options.u64("advance-delay-ms", 0)?, - omit_view_at_boundary: options.opt_u64("omit-view-at-boundary")?, + advance_delay_ms: options.u64(flags::ADVANCE_DELAY_MS, 0)?, + omit_view_at_boundary: options.opt_u64(flags::OMIT_VIEW_AT_BOUNDARY)?, + stale_view_at_boundary: options.opt_u64(flags::STALE_VIEW_AT_BOUNDARY)?, + truncated_view_at_boundary: options + .opt_u64(flags::TRUNCATED_VIEW_AT_BOUNDARY)?, + omit_audio_at_boundary: options.opt_u64(flags::OMIT_AUDIO_AT_BOUNDARY)?, + overlapping_audio_at_boundary: options + .opt_u64(flags::OVERLAPPING_AUDIO_AT_BOUNDARY)?, }, client_id: client_id.clone(), service: service.clone(), @@ -280,3 +304,34 @@ fn measure_row(options: &Options) -> Result<(), String> { println!("{}", row.to_json()); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + /// An unknown flag is refused by name. A renamed option must fail the launch rather than + /// be accepted and ignored, which would turn a fault or a delay into a no-op. + #[test] + fn an_unknown_option_is_refused_by_name() { + let args: Vec = ["--session", "demo", "--stale-view-at-boundry", "2"] + .iter() + .map(|s| (*s).to_owned()) + .collect(); + let error = Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .expect_err("an unknown option is refused"); + assert!(error.contains("--stale-view-at-boundry"), "{error}"); + } + + /// A flag that belongs to another command is refused too: an agent has no render delay. + #[test] + fn an_option_of_another_command_is_refused() { + let args: Vec = ["--observation-delay-steps", "2"] + .iter() + .map(|s| (*s).to_owned()) + .collect(); + Options::parse(&args, &[flags::COMMON, flags::AGENT_ONLY]) + .expect_err("the agent command has no render delay"); + Options::parse(&args, &[flags::COMMON, flags::ENVIRONMENT_ONLY]) + .expect("the environment command does"); + } +} diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 12a7938..e7e4bb7 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -15,6 +15,7 @@ use std::time::{Duration, Instant}; use serde_json::{Map, Value, json}; use crate::clock::Pacing; +use crate::media::{self, AudioTimelines}; use crate::metrics::Metrics; use crate::phase::{Phase, PhaseMachine}; use crate::rpc::{self, DomainReply, Serials, WorkerRef}; @@ -260,6 +261,15 @@ pub struct Coordinator { views: BTreeMap, /// 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, @@ -325,6 +335,13 @@ impl Coordinator { observation: None, views: BTreeMap::new(), pending_views: BTreeMap::new(), + audio: BTreeMap::new(), + pending_audio: BTreeMap::new(), + timelines: AudioTimelines::default(), + media_names: vec![ + media::view_attachment(crate::environment::VIEW_ID), + media::audio_attachment(crate::environment::AUDIO_STREAM_ID), + ], serials: Serials::default(), topics, pacing: None, @@ -374,6 +391,21 @@ impl Coordinator { self.observation.as_ref() } + /// The media handles this committed boundary holds: the one the agents were given and the + /// one presentation was published. They are the same objects, named by attachment. + pub fn media_handles(&self) -> Vec<(String, ArtifactRef)> { + self.views + .iter() + .chain(self.audio.iter()) + .map(|(name, artifact)| (name.clone(), artifact.reference().clone())) + .collect() + } + + /// Where each declared audio stream's next chunk may start. + pub fn audio_positions(&self) -> BTreeMap { + self.timelines.positions() + } + pub fn episode_request(&self) -> Option<&EpisodeRequest> { self.episode.as_ref() } @@ -632,6 +664,7 @@ impl Coordinator { }; let worker = self.environment.clone(); let scope = self.scope(0); + let want = self.media_names.clone(); let reply = self .call( &worker, @@ -639,7 +672,7 @@ impl Coordinator { Some(scope), object(params.to_json()), &[], - &["view.arena".to_owned()], + &want, ) .await?; let result: EnvironmentInitializeResult = @@ -672,7 +705,17 @@ impl Coordinator { "port-assignment", )); } - self.views = reply.artifacts; + let (views, audio) = media::split_attachments(reply.artifacts); + self.views = views; + self.audio = audio; + self.media_names = media::attachment_names(&result.descriptor); + self.timelines = AudioTimelines::fresh(&result.descriptor); + if let Err(e) = self.timelines.accept(&result.descriptor, &result.observation) { + return Err(self.fail_now(e, "observation-0")); + } + if let Err(e) = media::check_required_views(&result.descriptor, &result.observation) { + return Err(self.fail_now(e, "observation-0")); + } self.descriptor = Some(result.descriptor); self.observation = Some(result.observation); self.lifecycle_acks.push((worker, reply.request_id.clone())); @@ -1461,10 +1504,19 @@ impl Coordinator { .sensory_views .iter() .filter_map(|view| { - let name = format!("view.{}", view.view_id); + let name = media::view_attachment(&view.view_id); self.pending_views.get(&name).map(|a| (name, a.clone())) }) .collect(); + let new_audio: BTreeMap = step_result + .observation + .audio + .iter() + .filter_map(|chunk| { + let name = media::audio_attachment(&chunk.stream_id); + self.pending_audio.get(&name).map(|a| (name, a.clone())) + }) + .collect(); let commits = self .commit_all(k, &step_result.observation, &mut outcomes, &mut next_contexts, &new_views) .await?; @@ -1480,7 +1532,9 @@ impl Coordinator { } // The previous boundary's handles are no longer needed; the new ones take over. self.views = new_views; + self.audio = new_audio; self.pending_views.clear(); + self.pending_audio.clear(); self.observation = Some(step_result.observation.clone()); self.stats.advances += 1; @@ -1790,7 +1844,7 @@ impl Coordinator { let worker = self.environment.clone(); let request_id = self.serials.next(&worker.service); self.last_advance_request = Some(request_id.clone()); - let want = vec!["view.arena".to_owned()]; + let want = self.media_names.clone(); self.audit.push(format!("advance:{k}")); self.blame(Some(worker.worker_id.clone())); let advance_deadline = self.deadlines.probe; @@ -1912,7 +1966,9 @@ impl Coordinator { Err(e) => return Err(self.fail_now(e, "advance")), }; self.blame(None); - self.pending_views = reply.artifacts; + let (pending_views, pending_audio) = media::split_attachments(reply.artifacts); + self.pending_views = pending_views; + self.pending_audio = pending_audio; if injected && self.injections.altered_advance_controls { // The same request id with a different body: a conflict, never a second world @@ -1947,6 +2003,7 @@ impl Coordinator { None => None, }; self.pending_views.clear(); + self.pending_audio.clear(); let replay = self .resolve( &worker, @@ -1967,7 +2024,9 @@ impl Coordinator { code: None, identical: first.is_some() && first == again, }); - self.pending_views = replay.artifacts; + let (pending_views, pending_audio) = media::split_attachments(replay.artifacts); + self.pending_views = pending_views; + self.pending_audio = pending_audio; } Ok(result) } @@ -2042,43 +2101,20 @@ impl Coordinator { "step-result", )); } - // A missing required sensory input is never silently replaced by an older frame. - // 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) = media::check_required_audio(descriptor, &result.observation) { + return Err(self.fail_now(e, "step-result")); + } + 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( @@ -2087,7 +2123,7 @@ impl Coordinator { )); } for view in &result.observation.sensory_views { - let name = format!("view.{}", view.view_id); + let name = media::view_attachment(&view.view_id); match self.pending_views.get(&name) { Some(artifact) if artifact.reference() == &view.pixels => {} _ => { @@ -2102,6 +2138,25 @@ impl Coordinator { } } } + for chunk in &result.observation.audio { + let name = media::audio_attachment(&chunk.stream_id); + match self.pending_audio.get(&name) { + Some(artifact) if artifact.reference() == &chunk.samples => {} + _ => { + return Err(self.fail_now( + DomainError::new( + ErrorCode::BufferInvalid, + format!( + "audio chunk {} arrived without a live owned handle", + chunk.stream_id + ), + MutationCertainty::Unknown, + ), + "step-result", + )); + } + } + } Ok(()) } @@ -2528,11 +2583,13 @@ impl Coordinator { "progress": self.task.progress().to_json(), "media": json!({ "views": Value::Array(observation.broadcast_views.iter().map(DomainType::to_json).collect()), - "audio": [], + "audio": Value::Array(observation.audio.iter().map(DomainType::to_json).collect()), }), "eventIds": event_ids.iter().map(Id::as_str).collect::>(), }); - 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 cf72b21..8a2947b 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)] @@ -38,6 +60,12 @@ pub struct EnvironmentConfig { pub ports: Vec, /// The thread allocation the launcher started this worker within. pub worker_threads: usize, + /// The view's declared render delay, in steps. Zero is same-boundary output. + pub observation_delay_steps: u64, + /// Counts frames actually rendered, so a test can prove one image was not rendered twice. + /// + /// It counts in this process only: a world with a process of its own counts there. + pub renders: RenderCounter, pub faults: EnvironmentFaults, } @@ -54,6 +82,10 @@ pub struct CounterEnvironment { world_time: RationalNs, advances: u64, batches: BTreeSet, + 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 { @@ -69,6 +101,9 @@ impl CounterEnvironment { world_time: RationalNs::ZERO, advances: 0, batches: BTreeSet::new(), + pipeline: None, + audio: None, + previous_view: None, config, } } @@ -103,15 +138,25 @@ impl CounterEnvironment { } } - pub fn view_descriptor() -> ViewDescriptor { + /// The arena's native view, with the configured render delay. + pub fn view_descriptor(observation_delay_steps: u64) -> ViewDescriptor { ViewDescriptor { - view_id: id("arena"), + view_id: id(VIEW_ID), width: VIEW_WIDTH, height: VIEW_HEIGHT, row_stride: VIEW_WIDTH * 4, pixel_aspect_numerator: 1, pixel_aspect_denominator: 1, - observation_delay_steps: 0, + observation_delay_steps, + } + } + + /// The arena's native audio stream. + pub fn audio_descriptor() -> AudioDescriptor { + AudioDescriptor { + stream_id: id(AUDIO_STREAM_ID), + sample_rate: SAMPLE_RATE, + channels: CHANNELS, } } @@ -121,10 +166,11 @@ impl CounterEnvironment { content_digest: digest_of_bytes(b"counter-arena-content-v1"), configuration_digest: digest_of_bytes( format!( - "counter-arena-config-v1\nstep={}/{}\nports={}\n", + "counter-arena-config-v1\nstep={}/{}\nports={}\ndelay={}\n", self.config.step_duration.numerator, self.config.step_duration.denominator, - self.config.ports.len() + self.config.ports.len(), + self.config.observation_delay_steps ) .as_bytes(), ), @@ -139,8 +185,10 @@ impl CounterEnvironment { }) .collect(), inspection_schema: inspection_schema(), - views: vec![CounterEnvironment::view_descriptor()], - audio: Vec::new(), + views: vec![CounterEnvironment::view_descriptor( + self.config.observation_delay_steps, + )], + audio: vec![CounterEnvironment::audio_descriptor()], recovery: Recovery::ExactCheckpoint, determinism: Determinism::FixedBuild, }; @@ -148,73 +196,76 @@ impl CounterEnvironment { Ok(descriptor) } - /// Seals one immutable native frame for the current counter and returns the handle. - 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)) } @@ -270,6 +321,13 @@ impl CounterEnvironment { self.counter = 0; self.world_time = RationalNs::ZERO; self.batches.clear(); + self.pipeline = Some(ViewPipeline::new( + CounterEnvironment::view_descriptor(self.config.observation_delay_steps), + self.config.renders.clone(), + )); + // A fresh episode starts at audio origin zero; a restore would resume the preserved + // sample position instead, and its first chunk would mark the discontinuity. + self.audio = Some(AudioSource::new(CounterEnvironment::audio_descriptor(), 0)); self.descriptor = Some(descriptor.clone()); // The world is stopped when O[0] goes out and cannot free-run while the brains boot. self.status.set_state(WorkerState::Ready); diff --git a/services/flysim/crates/fly-session/src/harness.rs b/services/flysim/crates/fly-session/src/harness.rs index 6840f8c..c537a9b 100644 --- a/services/flysim/crates/fly-session/src/harness.rs +++ b/services/flysim/crates/fly-session/src/harness.rs @@ -22,6 +22,7 @@ use flybus::{Client, Grants, Pattern, Policy, Router, RouterConfig}; use crate::agent::{AgentFaults, synthetic_profile}; use crate::coordinator::{AgentSlot, Coordinator}; use crate::environment::EnvironmentFaults; +use crate::media::{RenderCounter, SensorLog}; use crate::launcher::{ AgentLaunch, EnvironmentLaunch, Launcher, ReapOutcome, SUPERVISOR_CLIENT, ThreadBudget, }; @@ -72,6 +73,8 @@ pub struct HarnessConfig { pub tick_ms: u64, pub warmup_ticks: u64, pub terminal: Terminal, + /// The view's declared render delay, in steps. Zero is same-boundary output. + pub observation_delay_steps: u64, pub environment_faults: EnvironmentFaults, /// Where each participant runs. pub mode: ExecutionMode, @@ -98,6 +101,7 @@ impl Default for HarnessConfig { tick_ms: 1, warmup_ticks: 10, terminal: Terminal::Never, + observation_delay_steps: 0, environment_faults: EnvironmentFaults::default(), mode: ExecutionMode::InProcess, thread_budget: None, @@ -160,6 +164,11 @@ pub struct SessionHarness { pub config: HarnessConfig, pub via: Via, pub mode: ExecutionMode, + /// The media instrumentation of the participants that live in this process. Both are + /// shared memory, so both are empty for a participant with a process of its own; the + /// accessors below return `None` there rather than zero. + renders: RenderCounter, + sensors: BTreeMap, /// The supervisor. It owns every participant's lifetime and thread allocation. pub launcher: Launcher, observers: Mutex>, @@ -226,6 +235,12 @@ impl SessionHarness { let step_duration = hz(config.step_hz).expect("a positive cadence"); let tick_duration = millis(config.tick_ms).expect("a positive tick"); + let renders = RenderCounter::new(); + let sensors: BTreeMap = config + .agents + .iter() + .map(|spec| (spec.agent_id.clone(), SensorLog::new())) + .collect(); // The environment first: it owns the world and the descriptor. let environment = launcher @@ -236,6 +251,8 @@ impl SessionHarness { step_duration, ports: config.agents.iter().map(|a| a.port_id.clone()).collect(), worker_threads: config.environment_threads, + observation_delay_steps: config.observation_delay_steps, + renders: renders.clone(), faults: config.environment_faults.clone(), client_id: ENV_CLIENT.to_owned(), service: ENV_SERVICE.to_owned(), @@ -259,6 +276,7 @@ impl SessionHarness { tick_duration, warmup_ticks: config.warmup_ticks, worker_threads: spec.worker_threads, + sensors: sensors[&spec.agent_id].clone(), faults: spec.faults.clone(), client_id: agent_client(&spec.agent_id), service: agent_service(&spec.agent_id), @@ -304,6 +322,8 @@ impl SessionHarness { config, via, mode: launcher.mode(), + renders, + sensors, launcher, observers: Mutex::new(Vec::new()), }) @@ -366,6 +386,9 @@ impl SessionHarness { tick_duration, warmup_ticks: self.config.warmup_ticks, worker_threads: spec.worker_threads, + // The same log: a replacement worker in this process keeps writing where its + // predecessor wrote, so a restore's sensory input is visible beside it. + sensors: self.sensors.get(agent_id).cloned().unwrap_or_default(), faults: spec.faults.clone(), client_id: format!("{}-r2", agent_client(agent_id)), service: agent_service(agent_id), @@ -390,6 +413,30 @@ impl SessionHarness { id(ENV_WORKER) } + /// What one agent read out of its sensory attachments, in order, when this process is + /// where that log lives. + /// + /// `None` means "not observable from here", not "nothing was read": an agent with a + /// process of its own records into its own copy. The media path itself crosses a process + /// boundary -- the frame is one artifact in the shared store, reached through owned + /// handles -- but this instrumentation does not, because it is shared memory. + pub fn sensor_log(&self, agent_id: &Id) -> Option { + match self.mode { + ExecutionMode::Process => None, + _ => self.sensors.get(agent_id).cloned(), + } + } + + /// How many native frames the environment rendered, when the world lives in this process. + /// + /// `None` for a world with a process of its own, for the same reason as above. + pub fn renders(&self) -> Option { + match self.mode { + ExecutionMode::Process => None, + _ => Some(self.renders.count()), + } + } + /// The agent worker's progress counter, which is its fake model's mutation count, when /// this process is where that counter lives. /// diff --git a/services/flysim/crates/fly-session/src/launcher.rs b/services/flysim/crates/fly-session/src/launcher.rs index 2537236..c80551e 100644 --- a/services/flysim/crates/fly-session/src/launcher.rs +++ b/services/flysim/crates/fly-session/src/launcher.rs @@ -227,6 +227,9 @@ pub struct AgentLaunch { pub warmup_ticks: u64, /// What the launcher asks the budget for. pub worker_threads: usize, + /// Where this agent records the views it reads. A participant with a process of its own + /// gets a fresh log in that process, which the supervisor cannot read. + pub sensors: crate::media::SensorLog, pub faults: AgentFaults, /// The configured client id. A replacement worker connects under its own. pub client_id: String, @@ -242,6 +245,10 @@ pub struct EnvironmentLaunch { pub step_duration: RationalNs, pub ports: Vec, pub worker_threads: usize, + /// The view's declared render delay, in steps. + pub observation_delay_steps: u64, + /// Where this world counts the frames it renders, with the same process caveat. + pub renders: crate::media::RenderCounter, pub faults: EnvironmentFaults, pub client_id: String, pub service: String, @@ -1197,6 +1204,97 @@ fn launch_error(worker_id: &Id, e: &flybus::BusError) -> DomainError { ) } + +// ------------------------------------------------------------------------------------------- +// The command line between a launcher and a participant in its own process + +/// Every `--flag` a launched participant or a measurement child accepts, named once. +/// +/// The launcher writes these and [`crate::cli`] reads them. Naming them in two places by +/// convention would let a rename turn an option into a silent no-op -- a fault that never +/// fires, a delay that is never applied -- so both sides use these constants and the parser +/// refuses any flag outside the set for the command it is parsing. +pub(crate) mod flags { + pub const SOCKET: &str = "socket"; + pub const STORE_ROOT: &str = "store-root"; + pub const CLIENT_ID: &str = "client-id"; + pub const SERVICE: &str = "service"; + pub const THREADS: &str = "threads"; + pub const SESSION: &str = "session"; + pub const INCARNATION: &str = "incarnation"; + + pub const AGENT: &str = "agent"; + pub const PORT: &str = "port"; + pub const TICK_NUMERATOR: &str = "tick-numerator"; + pub const TICK_DENOMINATOR: &str = "tick-denominator"; + pub const WARMUP_TICKS: &str = "warmup-ticks"; + pub const PREPARE_DELAY_MS: &str = "prepare-delay-ms"; + pub const COMMIT_DELAY_MS: &str = "commit-delay-ms"; + pub const FAIL_COMMIT_AT_STEP: &str = "fail-commit-at-step"; + + pub const WORKER: &str = "worker"; + pub const PORTS: &str = "ports"; + pub const STEP_NUMERATOR: &str = "step-numerator"; + pub const STEP_DENOMINATOR: &str = "step-denominator"; + pub const ADVANCE_DELAY_MS: &str = "advance-delay-ms"; + pub const OMIT_VIEW_AT_BOUNDARY: &str = "omit-view-at-boundary"; + pub const OBSERVATION_DELAY_STEPS: &str = "observation-delay-steps"; + pub const STALE_VIEW_AT_BOUNDARY: &str = "stale-view-at-boundary"; + pub const TRUNCATED_VIEW_AT_BOUNDARY: &str = "truncated-view-at-boundary"; + pub const OMIT_AUDIO_AT_BOUNDARY: &str = "omit-audio-at-boundary"; + pub const OVERLAPPING_AUDIO_AT_BOUNDARY: &str = "overlapping-audio-at-boundary"; + + pub const MODE: &str = "mode"; + pub const AGENTS: &str = "agents"; + pub const STEPS: &str = "steps"; + pub const WARMUP_STEPS: &str = "warmup-steps"; + pub const WORKER_THREADS: &str = "worker-threads"; + pub const MODES: &str = "modes"; + + /// What every launched worker is given. + pub const COMMON: &[&str] = &[ + SOCKET, + STORE_ROOT, + CLIENT_ID, + SERVICE, + THREADS, + SESSION, + INCARNATION, + ]; + /// What only an agent is given. + pub const AGENT_ONLY: &[&str] = &[ + AGENT, + PORT, + TICK_NUMERATOR, + TICK_DENOMINATOR, + WARMUP_TICKS, + PREPARE_DELAY_MS, + COMMIT_DELAY_MS, + FAIL_COMMIT_AT_STEP, + ]; + /// What only the environment is given, media options included. + pub const ENVIRONMENT_ONLY: &[&str] = &[ + WORKER, + PORTS, + STEP_NUMERATOR, + STEP_DENOMINATOR, + ADVANCE_DELAY_MS, + OMIT_VIEW_AT_BOUNDARY, + OBSERVATION_DELAY_STEPS, + STALE_VIEW_AT_BOUNDARY, + TRUNCATED_VIEW_AT_BOUNDARY, + OMIT_AUDIO_AT_BOUNDARY, + OVERLAPPING_AUDIO_AT_BOUNDARY, + ]; + /// What a measurement run or one of its row children is given. + pub const MEASURE: &[&str] = &[MODE, AGENTS, STEPS, WARMUP_STEPS, WORKER_THREADS, MODES]; +} + +/// One `--flag value` pair, so the flag name is written once and never spelled inline. +fn arg(name: &str, value: impl std::fmt::Display) -> (String, String) { + (format!("--{name}"), value.to_string()) +} + // ------------------------------------------------------------------------------------------- // What a participant is @@ -1220,48 +1318,50 @@ impl Started { match self { Started::Agent(spec) => { let mut args = vec![ - ("--session".to_owned(), spec.session_id.clone()), - ("--agent".to_owned(), spec.agent_id.clone()), - ("--port".to_owned(), spec.port_id.clone()), - ("--incarnation".to_owned(), spec.incarnation_id.clone()), - ("--tick-numerator".to_owned(), spec.tick_duration.numerator.to_string()), - ( - "--tick-denominator".to_owned(), - spec.tick_duration.denominator.to_string(), - ), - ("--warmup-ticks".to_owned(), spec.warmup_ticks.to_string()), - ( - "--prepare-delay-ms".to_owned(), - spec.faults.prepare_delay_ms.to_string(), - ), - ( - "--commit-delay-ms".to_owned(), - spec.faults.commit_delay_ms.to_string(), - ), + arg(flags::SESSION, &spec.session_id), + arg(flags::AGENT, &spec.agent_id), + arg(flags::PORT, &spec.port_id), + arg(flags::INCARNATION, &spec.incarnation_id), + arg(flags::TICK_NUMERATOR, spec.tick_duration.numerator), + arg(flags::TICK_DENOMINATOR, spec.tick_duration.denominator), + arg(flags::WARMUP_TICKS, spec.warmup_ticks), + arg(flags::PREPARE_DELAY_MS, spec.faults.prepare_delay_ms), + arg(flags::COMMIT_DELAY_MS, spec.faults.commit_delay_ms), ]; if let Some(step) = spec.faults.fail_commit_at_step { - args.push(("--fail-commit-at-step".to_owned(), step.to_string())); + args.push(arg(flags::FAIL_COMMIT_AT_STEP, step)); } args } Started::Environment(spec) => { let mut args = vec![ - ("--session".to_owned(), spec.session_id.clone()), - ("--worker".to_owned(), spec.worker_id.clone()), - ("--incarnation".to_owned(), spec.incarnation_id.clone()), - ("--step-numerator".to_owned(), spec.step_duration.numerator.to_string()), - ( - "--step-denominator".to_owned(), - spec.step_duration.denominator.to_string(), - ), - ("--ports".to_owned(), spec.ports.join(",")), - ( - "--advance-delay-ms".to_owned(), - spec.faults.advance_delay_ms.to_string(), - ), + arg(flags::SESSION, &spec.session_id), + arg(flags::WORKER, &spec.worker_id), + arg(flags::INCARNATION, &spec.incarnation_id), + arg(flags::STEP_NUMERATOR, spec.step_duration.numerator), + arg(flags::STEP_DENOMINATOR, spec.step_duration.denominator), + arg(flags::PORTS, spec.ports.join(",")), + arg(flags::ADVANCE_DELAY_MS, spec.faults.advance_delay_ms), + // The media options a world in another process needs to be exactly this + // world. Its render counter and its agents' sensor logs stay there. + arg(flags::OBSERVATION_DELAY_STEPS, spec.observation_delay_steps), ]; - if let Some(boundary) = spec.faults.omit_view_at_boundary { - args.push(("--omit-view-at-boundary".to_owned(), boundary.to_string())); + for (flag, boundary) in [ + (flags::OMIT_VIEW_AT_BOUNDARY, spec.faults.omit_view_at_boundary), + (flags::STALE_VIEW_AT_BOUNDARY, spec.faults.stale_view_at_boundary), + ( + flags::TRUNCATED_VIEW_AT_BOUNDARY, + spec.faults.truncated_view_at_boundary, + ), + (flags::OMIT_AUDIO_AT_BOUNDARY, spec.faults.omit_audio_at_boundary), + ( + flags::OVERLAPPING_AUDIO_AT_BOUNDARY, + spec.faults.overlapping_audio_at_boundary, + ), + ] { + if let Some(boundary) = boundary { + args.push(arg(flag, boundary)); + } } args } @@ -1277,6 +1377,7 @@ pub(crate) fn agent_config(spec: &AgentLaunch, worker_threads: usize) -> AgentCo tick_duration: spec.tick_duration, warmup_ticks: spec.warmup_ticks, worker_threads, + sensors: spec.sensors.clone(), faults: spec.faults.clone(), } } @@ -1289,6 +1390,8 @@ pub(crate) fn environment_config(spec: &EnvironmentLaunch) -> EnvironmentConfig step_duration: spec.step_duration, ports: spec.ports.clone(), worker_threads: spec.worker_threads, + observation_delay_steps: spec.observation_delay_steps, + renders: spec.renders.clone(), faults: spec.faults.clone(), } } @@ -1342,3 +1445,98 @@ pub(crate) async fn register_with_retry( } } } + +#[cfg(test)] +mod flag_tests { + use super::*; + + fn every_media_fault() -> EnvironmentFaults { + EnvironmentFaults { + advance_delay_ms: 3, + omit_view_at_boundary: Some(1), + stale_view_at_boundary: Some(2), + truncated_view_at_boundary: Some(3), + omit_audio_at_boundary: Some(4), + overlapping_audio_at_boundary: Some(5), + } + } + + fn environment_launch() -> EnvironmentLaunch { + EnvironmentLaunch { + session_id: id("demo"), + worker_id: id("arena"), + incarnation_id: id("arena-inc-1"), + step_duration: RationalNs::new(1, 60).expect("a cadence"), + ports: vec![id("p1"), id("p2")], + worker_threads: 1, + observation_delay_steps: 2, + renders: crate::media::RenderCounter::new(), + faults: every_media_fault(), + client_id: "environment".to_owned(), + service: "env.arena".to_owned(), + } + } + + fn agent_launch() -> AgentLaunch { + AgentLaunch { + session_id: id("demo"), + agent_id: id("fly-a"), + port_id: id("p1"), + incarnation_id: id("fly-a-inc-1"), + tick_duration: RationalNs::new(1, 1_000).expect("a tick"), + warmup_ticks: 10, + worker_threads: 1, + sensors: crate::media::SensorLog::new(), + faults: AgentFaults { + fail_commit_at_step: Some(2), + prepare_delay_ms: 1, + commit_delay_ms: 2, + }, + client_id: "worker-fly-a".to_owned(), + service: "agent.fly-a".to_owned(), + } + } + + /// Both halves of the command line name the same constants, and this proves it for every + /// argument a launch can produce: a flag the launcher writes that the parser does not + /// accept would be a silently ignored option, which is what the parser now refuses. + #[test] + fn every_flag_a_launch_writes_is_one_its_command_accepts() { + for (started, allowed) in [ + ( + Started::Environment(environment_launch()), + [flags::COMMON, flags::ENVIRONMENT_ONLY], + ), + (Started::Agent(agent_launch()), [flags::COMMON, flags::AGENT_ONLY]), + ] { + let arguments = started.arguments(); + assert!(!arguments.is_empty()); + for (flag, _value) in &arguments { + let name = flag.strip_prefix("--").expect("every argument is a --flag"); + assert!( + allowed.iter().any(|set| set.contains(&name)), + "{} writes --{name}, which its command does not accept", + started.subcommand() + ); + } + } + } + + /// Every media option reaches the argv when it is set, so a world in another process is + /// exactly the world the composition asked for. + #[test] + fn the_media_options_are_all_written_for_a_separate_process() { + let started = Started::Environment(environment_launch()); + let written: Vec = started.arguments().into_iter().map(|(f, _)| f).collect(); + for flag in [ + flags::OBSERVATION_DELAY_STEPS, + flags::OMIT_VIEW_AT_BOUNDARY, + flags::STALE_VIEW_AT_BOUNDARY, + flags::TRUNCATED_VIEW_AT_BOUNDARY, + flags::OMIT_AUDIO_AT_BOUNDARY, + flags::OVERLAPPING_AUDIO_AT_BOUNDARY, + ] { + assert!(written.contains(&format!("--{flag}")), "--{flag} is not written"); + } + } +} diff --git a/services/flysim/crates/fly-session/src/lib.rs b/services/flysim/crates/fly-session/src/lib.rs index 5d9aae1..15722b5 100644 --- a/services/flysim/crates/fly-session/src/lib.rs +++ b/services/flysim/crates/fly-session/src/lib.rs @@ -30,6 +30,7 @@ pub mod environment; pub mod harness; pub mod launcher; pub mod measure; +pub mod media; pub mod metrics; pub mod phase; pub mod rpc; diff --git a/services/flysim/crates/fly-session/src/measure.rs b/services/flysim/crates/fly-session/src/measure.rs index 963139f..ed79c6a 100644 --- a/services/flysim/crates/fly-session/src/measure.rs +++ b/services/flysim/crates/fly-session/src/measure.rs @@ -24,7 +24,7 @@ use serde_json::{Value, json}; use crate::coordinator::DispatchOrder; use crate::harness::{AgentSpec, HarnessConfig, SessionHarness, Via}; -use crate::launcher::ExecutionMode; +use crate::launcher::{ExecutionMode, flags}; use crate::metrics::{Percentiles, physical_cores}; /// What to compare. @@ -241,15 +241,15 @@ fn row_in_a_child( ) -> Result { let output = std::process::Command::new(program) .arg("measure-row") - .arg("--mode") + .arg(format!("--{}", flags::MODE)) .arg(mode.label()) - .arg("--agents") + .arg(format!("--{}", flags::AGENTS)) .arg(agents.to_string()) - .arg("--steps") + .arg(format!("--{}", flags::STEPS)) .arg(config.steps.to_string()) - .arg("--warmup-steps") + .arg(format!("--{}", flags::WARMUP_STEPS)) .arg(config.warmup_steps.to_string()) - .arg("--worker-threads") + .arg(format!("--{}", flags::WORKER_THREADS)) .arg(config.worker_threads.to_string()) .stdin(std::process::Stdio::null()) .output() 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..870ead9 --- /dev/null +++ b/services/flysim/crates/fly-session/src/media.rs @@ -0,0 +1,837 @@ +//! Native observations and the presentation handoff: the MEDIA-01 slice. +//! +//! Everything here sits on top of the bus `ArtifactRef` and its ownership rules. There is no +//! second buffer system: an environment allocates, writes and seals one immutable object per +//! boundary, the coordinator forwards that one owned handle to every agent and to publication, +//! and a spectator reads it through an ordinary latest subscription. +//! +//! The module owns four things: +//! +//! 1. Production. [`ViewPipeline`] renders one native frame per boundary and hands out the +//! frame a declared `observationDelaySteps` requires, so a pipeline delay is a real queue +//! rather than a number in a descriptor. [`AudioSource`] produces one chunk per boundary +//! with an exact rational sample budget. +//! 2. Acceptance. [`check_required_views`] and [`AudioTimelines`] are the coordinator's Phase C +//! media checks: a required sensory view must exist at exactly the producing boundary its +//! declared delay implies, and audio chunks cannot overlap or go backwards inside an epoch. +//! 3. Consumption. [`Spectator`] is a presentation-side consumer on a latest subscription with +//! finite credits, and [`detach_frame`] is the renderer that keeps its handle after the +//! message is gone. +//! 4. Identity. [`AssetRegistry`] holds installed persistent content named by `AssetRef`. +//! Importing an asset produces a *new* transient artifact; the two identities never convert. +//! +//! Resizing, overlays, compositing, mixing, encoding and streaming are not here and are not +//! anywhere else in this crate: they belong to the application's presentation layer. + +use std::collections::{BTreeMap, VecDeque}; +use std::io::Write; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use fly_session_types::media::AudioTimeline; + +// `crate::types` is this crate's facade over the shared `fly-session-types` crate; the glob +// keeps the contract's own names in sight instead of restating them. +use crate::types::*; + +/// The content type of a native RGBA8 frame. Top-left origin, no padded rows. +pub const FRAME_CONTENT_TYPE: &str = "image/x-rgba8"; + +/// The content type of a native audio chunk: interleaved little-endian f32. +pub const AUDIO_CONTENT_TYPE: &str = "audio/x-f32le"; + +/// The attachment name one view's pixels travel under. +pub fn view_attachment(view_id: &str) -> String { + format!("view.{view_id}") +} + +/// The attachment name one audio stream's samples travel under. +pub fn audio_attachment(stream_id: &str) -> String { + format!("audio.{stream_id}") +} + +fn store_error(what: &str, message: &str) -> DomainError { + DomainError::new( + ErrorCode::BackendFailure, + format!("{what}: {message}"), + MutationCertainty::Applied, + ) +} + +fn media_error(message: impl std::fmt::Display) -> DomainError { + // A world that advanced without usable media leaves the transition's certainty unknown: + // the mutation happened, the observation of it did not. + DomainError::new(ErrorCode::BufferInvalid, message, MutationCertainty::Unknown) +} + +// ---------------------------------------------------------------------------------------------- +// Production + +/// How many frames a producer has actually rendered. +/// +/// A shared counter, so a test can prove that forwarding one image to several recipients +/// renders it once. +#[derive(Clone, Debug, Default)] +pub struct RenderCounter(Arc); + +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() + } + +} + +/// 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 produces exactly one chunk per transition. +/// +/// The contract states the shape and the ordering of chunks, not whether one has to exist, so +/// this is MEDIA-01's choice and it is deliberate: a session that tolerates a silently missing +/// chunk cannot tell "this world produced no audio for this interval" from "the chunk was +/// lost", and the second is the case the retention rules care about. Boundary 0 has no +/// preceding interval and so carries no chunk. +pub fn check_required_audio( + descriptor: &EnvironmentDescriptor, + observation: &WorldObservation, +) -> DomainResult<()> { + if observation.boundary == 0 { + return Ok(()); + } + for stream in &descriptor.audio { + if !observation + .audio + .iter() + .any(|chunk| chunk.stream_id == stream.stream_id) + { + return Err(media_error(format!( + "declared audio stream {} produced no chunk for this transition", + stream.stream_id + ))); + } + } + Ok(()) +} + +/// Every declared audio stream's chunk sequence, one timeline per stream and epoch. +#[derive(Clone, Debug, Default)] +pub struct AudioTimelines(BTreeMap); + +impl AudioTimelines { + /// Fresh timelines for a new episode: every declared stream starts at origin zero. + pub fn fresh(descriptor: &EnvironmentDescriptor) -> AudioTimelines { + AudioTimelines( + descriptor + .audio + .iter() + .map(|stream| (stream.stream_id.clone(), AudioTimeline::fresh(stream, 0))) + .collect(), + ) + } + + /// Timelines for a new epoch after a restore: each stream resumes at its preserved sample + /// position, and each one's first chunk must mark a discontinuity. + /// + /// A declared stream with no recorded position is an error. Resuming it at sample zero + /// would restart the episode's audio clock silently, which is exactly the best-effort + /// policy the restore rules refuse: crash restore *preserves* the sample position. + pub fn restored( + descriptor: &EnvironmentDescriptor, + positions: &BTreeMap, + ) -> DomainResult { + let mut timelines = BTreeMap::new(); + for stream in &descriptor.audio { + let at = positions.get(&stream.stream_id).copied().ok_or_else(|| { + DomainError::before( + ErrorCode::IncompatibleState, + format!( + "audio stream {} has no restored sample position", + stream.stream_id + ), + ) + })?; + timelines.insert(stream.stream_id.clone(), AudioTimeline::restored_at(stream, at)); + } + Ok(AudioTimelines(timelines)) + } + + /// Accepts one observation's chunks. Unknown streams and out-of-sequence chunks fail. + pub fn accept( + &mut self, + descriptor: &EnvironmentDescriptor, + observation: &WorldObservation, + ) -> DomainResult<()> { + for chunk in &observation.audio { + let declared = descriptor + .audio_stream(&chunk.stream_id) + .ok_or_else(|| media_error(format!("audio stream {} is not declared", chunk.stream_id)))?; + let timeline = self + .0 + .get_mut(&chunk.stream_id) + .ok_or_else(|| media_error(format!("audio stream {} has no timeline", chunk.stream_id)))?; + timeline.accept(chunk, declared).map_err(media_error)?; + } + Ok(()) + } + + /// Where each stream's next chunk may start. + pub fn positions(&self) -> BTreeMap { + 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()?; + // An unreadable chunk, or one whose handle is not attached, makes the whole snapshot + // unreadable rather than a snapshot that quietly has less audio in it than was published. + let mut audio = Vec::new(); + for value in media.get("audio")?.as_array()? { + let chunk = AudioRef::from_json(value).ok()?; + let artifact = message.artifact(&audio_attachment(&chunk.stream_id)).ok()?; + audio.push((chunk, artifact)); + } + let artifact = message.artifact(&view_attachment(&view.view_id)).ok()?; + let agents = payload + .get("agents") + .and_then(serde_json::Value::as_array) + .map(|agents| { + agents + .iter() + .filter_map(|a| a.get("agentId").and_then(serde_json::Value::as_str)) + .map(str::to_owned) + .collect() + }) + .unwrap_or_default(); + Some(SpectatorFrame { + boundary, + agents, + sequence: message.topic_sequence(), + replaced: message.replaced(), + view, + artifact, + audio, + }) +} + +/// Takes the frame out of a message and drops the message, as a renderer that finishes later +/// does. The delivery stays alive because the extracted handle still owns it. +pub fn detach_frame(message: flybus::Message) -> Option<(ViewRef, flybus::Artifact)> { + let frame = snapshot_frame(&message)?; + drop(message); + Some((frame.view, frame.artifact)) +} + +// ---------------------------------------------------------------------------------------------- +// Persistent assets + +/// The preprovisioned local registry an `AssetRef` names. +/// +/// An asset is installed and verified before a run; it is not a path, a URL or something a +/// worker fetches. Nothing in this registry can be addressed by an `ArtifactRef`, and +/// [`AssetRegistry::import`] hands back a fresh transient artifact rather than turning the +/// asset into one. +#[derive(Clone, Debug, Default)] +pub struct AssetRegistry { + installed: BTreeMap)>, +} + +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..220221c --- /dev/null +++ b/services/flysim/crates/fly-session/tests/media.rs @@ -0,0 +1,862 @@ +//! MEDIA-01 acceptance: native observations on the bus artifact, and the presentation handoff. +//! +//! Every test here is one of the slice's acceptance bullets or one row of the domain retention +//! table in `state-media-v1` section 3. The shape rules themselves are proved against the +//! contract crate in `fly-session-types/tests/media_shapes.rs`; these prove the session's use +//! of them: one shared image, spectators that cannot touch sensory state, a renderer that +//! keeps its handle, and a persistent asset that is not a transient artifact. + +mod common; + +use std::collections::BTreeMap; +use std::time::Duration; + +use serde_json::Map; + +use common::{Fixture, default_fixture, fixture, fly_a, fly_b, mode_fixture, within}; +use fly_session::environment::{ + AUDIO_STREAM_ID, CHANNELS, EnvironmentFaults, SAMPLE_RATE, VIEW_HEIGHT, VIEW_WIDTH, + synthetic_asset, +}; +use fly_session::harness::{ExecutionMode, HarnessConfig, Via}; +use fly_session::media::{ + AssetRegistry, AudioSource, AudioTimelines, SensedView, Spectator, SpectatorFrame, + arena_frame, audio_attachment, detach_frame, view_attachment, +}; +use fly_session::phase::Phase; +use fly_session::types::*; +use fly_session_types::media::{AudioTimeline, check_imported_asset, require_finite_samples}; + +both_transports!( + one_shared_image_reaches_both_agents_through_owned_attachments, + a_spectator_cannot_corrupt_sensory_state, + a_slow_spectator_exhausts_only_its_own_credits, + delayed_rendering_retains_its_handle_after_the_message_drops, + a_declared_render_delay_repeats_o0_until_the_pipeline_fills, + an_extra_delayed_sensory_view_fails_the_step, + a_frame_of_the_wrong_length_fails_the_step, + overlapping_audio_fails_the_step, + one_audio_chunk_per_boundary_with_an_exact_sample_budget, + required_agent_input_is_never_coalesced_while_spectator_snapshots_are, + a_persistent_asset_and_a_transient_artifact_are_different_identities, + a_restored_audio_source_resumes_and_marks_the_discontinuity, + a_missing_audio_chunk_fails_the_step, + restored_timelines_need_every_declared_streams_position, + a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples, +); + +all_modes!( + the_media_path_works_in_every_execution_mode, + every_media_fault_fires_in_every_execution_mode, +); + +const STEPS: u64 = 3; + +/// Polls until `ok` holds, so a test never asserts a collection that has not happened yet. +async fn until(what: &str, mut ok: impl FnMut() -> bool) { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while !ok() { + assert!(std::time::Instant::now() < deadline, "{what}: never happened"); + tokio::time::sleep(Duration::from_millis(5)).await; + } +} + +/// Drains a latest subscription until the snapshot for `boundary` arrives. +/// +/// A latest subscription converges on the newest value rather than delivering every one, so a +/// test that wants a particular committed boundary reads until it gets there. +async fn frame_at(spectator: &mut Spectator, boundary: u64) -> SpectatorFrame { + for _ in 0..64 { + let frame = within("snapshot", spectator.take_frame()) + .await + .expect("a snapshot"); + if frame.boundary == boundary { + return frame; + } + } + panic!("the spectator never reached boundary {boundary}"); +} + +/// The views one agent read. +/// +/// The sensor log is shared memory, so it is readable only where that agent lives. These tests +/// run in the default in-process composition; the process-mode test below asserts the media +/// path over the bus instead, which is what crosses a process boundary. +fn sensed(f: &Fixture, agent_id: &Id) -> Vec { + f.harness + .sensor_log(agent_id) + .expect("this composition keeps its agents in this process") + .entries() +} + +/// How many native frames the world rendered, for a world in this process. +fn renders(f: &Fixture) -> u64 { + f.harness + .renders() + .expect("this composition keeps its world in this process") +} + +fn boundaries(entries: &[SensedView]) -> Vec { + 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!( + renders(&f), + STEPS + 1, + "one native frame per boundary, whatever the number of recipients" + ); + + let a = sensed(&f, &fly_a()); + let b = sensed(&f, &fly_b()); + assert_eq!(boundaries(&a), (0..=STEPS).collect::>()); + 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 = sensed(&f, &fly_a()); + let b = sensed(&f, &fly_b()); + 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 = sensed(&f, &fly_a()); + 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 = sensed(&f, &fly_a()); + let at_boundary = entries + .iter() + .find(|e| e.produced_step == view.produced_step) + .expect("the agents encoded this frame too"); + assert_eq!( + digest_of_bytes(&bytes), + at_boundary.digest, + "the retained handle still reads exactly the frame that was published" + ); + f.shutdown().await; +} + +/// A declared render delay repeats `O[0]` while the pipeline fills, then advances one frame +/// per boundary. The repetition is the same artifact, not a re-render. +async fn a_declared_render_delay_repeats_o0_until_the_pipeline_fills(via: Via) { + let config = HarnessConfig { + observation_delay_steps: 2, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(4)).await.unwrap(); + + let a = sensed(&f, &fly_a()); + assert_eq!(boundaries(&a), vec![0, 1, 2, 3, 4]); + assert_eq!( + produced(&a), + vec![0, 0, 0, 1, 2], + "max(0, boundary - 2) at every boundary" + ); + assert_eq!(a[0].artifact_id, a[1].artifact_id); + assert_eq!(a[0].artifact_id, a[2].artifact_id, "O[0] repeats while the delay fills"); + assert_ne!(a[2].artifact_id, a[3].artifact_id, "then the pipeline advances"); + assert_ne!(a[3].artifact_id, a[4].artifact_id); + // The world still renders once per boundary; the delay is a queue, not a missing frame. + assert_eq!(renders(&f), 5); + f.shutdown().await; +} + +/// Beyond the declared delay, an extra-delayed sensory view is a step failure, never an +/// arbitrary latest frame. +async fn an_extra_delayed_sensory_view_fails_the_step(via: Via) { + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + stale_view_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let failure = within("run", f.harness.coordinator.run(STEPS)) + .await + .expect_err("a stale frame fails the transition"); + assert_eq!(failure.error.code, ErrorCode::BufferInvalid); + assert_eq!(failure.error.mutation, MutationCertainty::Unknown); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert_eq!( + f.harness.coordinator.stats().advances, + 1, + "the transition that met the stale frame committed nothing" + ); + // The agents did not encode the stale frame. + let a = sensed(&f, &fly_a()); + assert_eq!(boundaries(&a), vec![0, 1]); + f.shutdown().await; +} + +/// A frame whose artifact is not `rowStride x height` bytes fails the step. +async fn a_frame_of_the_wrong_length_fails_the_step(via: Via) { + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + truncated_view_at_boundary: Some(1), + ..EnvironmentFaults::default() + }, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let failure = within("run", f.harness.coordinator.run(1)) + .await + .expect_err("a short frame fails the transition"); + assert_eq!(failure.error.code, ErrorCode::BufferInvalid); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert_eq!(f.harness.coordinator.stats().advances, 0); + f.shutdown().await; +} + +/// Within an epoch, an audio chunk that starts inside the previous one is refused. +async fn overlapping_audio_fails_the_step(via: Via) { + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + overlapping_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let failure = within("run", f.harness.coordinator.run(STEPS)) + .await + .expect_err("an overlapping chunk fails the transition"); + assert_eq!(failure.error.code, ErrorCode::BufferInvalid); + assert_eq!(f.harness.coordinator.stats().advances, 1); + f.shutdown().await; +} + +/// One chunk per boundary, with an exact sample budget, finite samples and the byte length the +/// descriptor implies. Audio is published for presentation and never enters sensory input. +async fn one_audio_chunk_per_boundary_with_an_exact_sample_budget(via: Via) { + let mut f = default_fixture(via).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let observer = f.harness.observer().await.unwrap(); + let topic = f.harness.coordinator.topics().snapshots.clone(); + let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap(); + within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); + + // 48 kHz at 60 Hz is exactly 800 frames a step, and the positions are contiguous. + let per_step = SAMPLE_RATE / f.harness.config.step_hz; + let positions = f.harness.coordinator.audio_positions(); + assert_eq!(positions[AUDIO_STREAM_ID], per_step * STEPS); + + let observation = f.harness.coordinator.observation().expect("an observation").clone(); + assert_eq!(observation.audio.len(), 1, "one chunk per boundary"); + let chunk = &observation.audio[0]; + assert_eq!(chunk.sample_frames, per_step); + assert_eq!(chunk.first_sample, per_step * (STEPS - 1)); + assert!(!chunk.discontinuity, "an uninterrupted epoch has no discontinuity"); + assert_eq!(chunk.samples.byte_length, per_step * CHANNELS * 4); + // Sensory input is pixels only: audio never becomes an agent's input. + assert!(observation.sensory_views.iter().all(|v| v.view_id == "arena")); + + let published = frame_at(&mut spectator, STEPS).await; + let (published_chunk, artifact) = published.audio.first().expect("the published chunk"); + assert_eq!(published_chunk.samples, chunk.samples); + assert_eq!( + artifact.reference().artifact_id, + chunk.samples.artifact_id, + "the same owned handle is published, not a copy" + ); + let bytes = artifact.read_all().await.expect("the published chunk"); + assert_eq!(bytes.len() as u64, per_step * CHANNELS * 4); + require_finite_samples(&bytes).expect("native samples are finite f32"); + assert_eq!( + published.audio.len(), + 1, + "the attachment list names every published chunk" + ); + assert_eq!(audio_attachment(AUDIO_STREAM_ID), "audio.arena"); + f.shutdown().await; +} + +/// A declared stream that produces no chunk for a transition is a step failure, not a silently +/// shorter epoch. +async fn a_missing_audio_chunk_fails_the_step(via: Via) { + let config = HarnessConfig { + environment_faults: EnvironmentFaults { + omit_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let failure = within("run", f.harness.coordinator.run(STEPS)) + .await + .expect_err("a missing chunk fails the transition"); + assert_eq!(failure.error.code, ErrorCode::BufferInvalid); + assert_eq!(f.harness.coordinator.stats().advances, 1); + // Boundary 0 has no preceding interval, so its empty audio list is not a missing chunk. + let mut clean = default_fixture(via).await; + within("bootstrap", clean.harness.coordinator.bootstrap()).await.unwrap(); + assert!(clean.harness.coordinator.observation().unwrap().audio.is_empty()); + clean.shutdown().await; + f.shutdown().await; +} + +/// Restored timelines resume every declared stream at its recorded position. A stream with no +/// recorded position is refused, not restarted at zero. +async fn restored_timelines_need_every_declared_streams_position(via: Via) { + let mut f = default_fixture(via).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + within("run", f.harness.coordinator.run(1)).await.unwrap(); + let descriptor = f.harness.coordinator.descriptor().expect("a descriptor").clone(); + let observation = f.harness.coordinator.observation().expect("an observation").clone(); + + // A fresh epoch accepts the transition's chunk and advances its position. + let mut fresh = AudioTimelines::fresh(&descriptor); + fresh + .accept(&descriptor, &observation) + .expect("the first chunk of a fresh epoch"); + assert_eq!(fresh.accepted(AUDIO_STREAM_ID), 1); + let chunk = observation.audio.first().expect("a chunk").clone(); + assert_eq!( + fresh.positions()[AUDIO_STREAM_ID], + chunk.first_sample + chunk.sample_frames + ); + + // A restore with nothing recorded for a declared stream is an error, not sample zero. + let missing = AudioTimelines::restored(&descriptor, &BTreeMap::new()) + .expect_err("a declared stream needs its preserved position"); + assert_eq!(missing.code, ErrorCode::IncompatibleState); + + // With the position recorded, the restored epoch resumes there and its first chunk must + // mark the discontinuity. + let positions = f.harness.coordinator.audio_positions(); + let mut resumed = observation.clone(); + let resumed_chunk = resumed.audio.first_mut().expect("a chunk"); + resumed_chunk.first_sample = positions[AUDIO_STREAM_ID]; + resumed_chunk.discontinuity = false; + let mut restored = AudioTimelines::restored(&descriptor, &positions).expect("positions"); + restored + .accept(&descriptor, &resumed) + .expect_err("the first chunk after a restore marks discontinuity"); + resumed.audio.first_mut().expect("a chunk").discontinuity = true; + let mut restored = AudioTimelines::restored(&descriptor, &positions).expect("positions"); + restored + .accept(&descriptor, &resumed) + .expect("the restored epoch resumes at the preserved position"); + assert_eq!(restored.accepted(AUDIO_STREAM_ID), 1); + f.shutdown().await; +} + +/// A world cadence that does not divide the sample rate still produces whole samples, with the +/// remainder carried rather than rounded: seven steps of a 7 Hz world are exactly one second. +async fn a_cadence_that_does_not_divide_the_sample_rate_still_lands_on_whole_samples(via: Via) { + let config = HarnessConfig { + step_hz: 7, + ..HarnessConfig::default() + }; + let mut f = fixture(via, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let observer = f.harness.observer().await.unwrap(); + let topic = f.harness.coordinator.topics().snapshots.clone(); + let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap(); + + // 48000 / 7 is 6857.14..., so no chunk can be the exact share of a second. + let mut frames = Vec::new(); + for step in 1..=7 { + within("step", f.harness.coordinator.run(1)).await.unwrap(); + let chunk = f + .harness + .coordinator + .observation() + .expect("an observation") + .audio + .first() + .expect("a chunk") + .clone(); + assert_eq!(chunk.first_sample, frames.iter().sum::()); + assert!( + chunk.sample_frames == 6_857 || chunk.sample_frames == 6_858, + "step {step} produced {} frames", + chunk.sample_frames + ); + assert_eq!(chunk.samples.byte_length, chunk.sample_frames * CHANNELS * 4); + frames.push(chunk.sample_frames); + } + assert_eq!( + frames.iter().sum::(), + SAMPLE_RATE, + "seven steps of a 7 Hz world are exactly one second of samples: {frames:?}" + ); + assert_eq!(f.harness.coordinator.audio_positions()[AUDIO_STREAM_ID], SAMPLE_RATE); + + // The published chunk is readable and finite whatever the cadence. + let published = frame_at(&mut spectator, 7).await; + let (chunk, artifact) = published.audio.first().expect("the published chunk"); + let bytes = artifact.read_all().await.expect("the published chunk"); + assert_eq!(bytes.len() as u64, chunk.sample_frames * CHANNELS * 4); + require_finite_samples(&bytes).expect("native samples are finite f32"); + f.shutdown().await; +} + +/// The media path itself is mode-agnostic: one native image per boundary, forwarded to every +/// agent as an owned attachment and published once for presentation, whether the participants +/// are tasks on one runtime, threads with their own runtimes, or separate processes. +/// +/// What a *test* can see differs by mode, and this test only asserts what crosses a process +/// boundary. An agent validates that each attachment is the artifact its payload names and +/// reads the pixels before it commits, so a session that commits every boundary in process +/// mode has carried one shared image across that boundary through the store, not through +/// shared memory. The sensor log and the render counter are shared memory, so they are +/// asserted where they exist and their absence is asserted where they do not. +async fn the_media_path_works_in_every_execution_mode(mode: ExecutionMode) { + // A declared render delay as well, so the option reaches a world in another process. + let config = HarnessConfig { + observation_delay_steps: 1, + ..HarnessConfig::default() + }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let observer = f.harness.observer().await.unwrap(); + let topic = f.harness.coordinator.topics().snapshots.clone(); + let mut spectator = Spectator::attach(&observer, &topic, 2).await.unwrap(); + within("run", f.harness.coordinator.run(STEPS)).await.unwrap(); + + // Every agent read its attachment and committed, in every mode. + assert_eq!(f.harness.coordinator.committed_boundary(), Some(STEPS)); + + // The coordinator holds one view handle and one audio handle for this boundary, and the + // observation names exactly those artifacts. + let observation = f.harness.coordinator.observation().expect("an observation").clone(); + let handles = f.harness.coordinator.media_handles(); + assert_eq!(handles.len(), 2, "one view and one chunk: {handles:?}"); + let view = observation.sensory_views.first().expect("a required view"); + assert_eq!( + view.produced_step, + STEPS - 1, + "the declared one-step delay reached the world in {mode:?} mode" + ); + assert!(handles.iter().any(|(name, r)| *name == view_attachment(&view.view_id) + && r.artifact_id == view.pixels.artifact_id)); + + // The same object is what presentation was published, and its bytes read back at the + // declared shape through an ordinary subscription. + let published = frame_at(&mut spectator, STEPS).await; + assert_eq!(published.view.pixels.artifact_id, view.pixels.artifact_id); + assert_eq!(published.view.produced_step, view.produced_step); + let pixels = published.artifact.read_all().await.expect("the published frame"); + assert_eq!(pixels.len() as u64, VIEW_WIDTH * VIEW_HEIGHT * 4); + let (chunk, audio) = published.audio.first().expect("the published chunk"); + let samples = audio.read_all().await.expect("the published chunk"); + assert_eq!(samples.len() as u64, chunk.sample_frames * CHANNELS * 4); + require_finite_samples(&samples).expect("native samples are finite f32"); + + // The shared-memory instrumentation exists exactly where the participants do. + match (mode, f.harness.sensor_log(&fly_a()), f.harness.renders()) { + (ExecutionMode::Process, sensors, renders) => { + assert!(sensors.is_none() && renders.is_none(), "not observable from here"); + } + (_, Some(sensors), Some(renders)) => { + let a = sensors.entries(); + let b = f.harness.sensor_log(&fly_b()).expect("in this process").entries(); + assert_eq!(a, b, "both agents read the same artifact and the same bytes"); + assert_eq!(boundaries(&a), (0..=STEPS).collect::>()); + assert_eq!( + digest_of_bytes(&pixels), + a.last().expect("an entry per boundary").digest, + "the published bytes are the bytes the agents encoded" + ); + assert_eq!(renders, STEPS + 1, "one render per boundary"); + } + (mode, _, _) => panic!("{mode:?} keeps its participants in this process"), + } + f.shutdown().await; +} + +/// Every media fault fires wherever the world runs, which is what proves the wiring rather +/// than assuming it. +/// +/// A fault reaches a world in another process only as a `--flag` on its command line, so a +/// renamed or dropped flag would make these faults silently stop firing. Here each one has to +/// fail its transition in every execution mode; the parser refuses an unknown flag, so a +/// mismatch fails the launch instead of turning into a no-op. +async fn every_media_fault_fires_in_every_execution_mode(mode: ExecutionMode) { + let cases: [(&str, EnvironmentFaults, u64); 4] = [ + ( + "an extra-delayed view", + EnvironmentFaults { + stale_view_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ( + "a frame of the wrong length", + EnvironmentFaults { + truncated_view_at_boundary: Some(1), + ..EnvironmentFaults::default() + }, + 1, + ), + ( + "a missing audio chunk", + EnvironmentFaults { + omit_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ( + "an overlapping audio chunk", + EnvironmentFaults { + overlapping_audio_at_boundary: Some(2), + ..EnvironmentFaults::default() + }, + 2, + ), + ]; + for (what, faults, steps) in cases { + let config = HarnessConfig { + environment_faults: faults, + ..HarnessConfig::default() + }; + let mut f = mode_fixture(mode, config).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let outcome = within("run", f.harness.coordinator.run(steps)).await; + let failure = match outcome { + Err(failure) => failure, + Ok(reports) => { + panic!("{what} did not fail the transition in {mode:?} mode: {reports:?}") + } + }; + assert_eq!( + failure.error.code, + ErrorCode::BufferInvalid, + "{what} in {mode:?} mode: {failure}" + ); + assert_eq!(f.harness.coordinator.phase(), Phase::Failed); + assert_eq!( + f.harness.coordinator.stats().advances, + steps - 1, + "{what} in {mode:?} mode committed the boundaries before it and no more" + ); + f.shutdown().await; + } +} + +/// The retention table: a required agent input is retained through encoding and Commit with no +/// coalescing, while a spectator's snapshots are a latest subscription with finite credits. +async fn required_agent_input_is_never_coalesced_while_spectator_snapshots_are(via: Via) { + let mut f = default_fixture(via).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let observer = f.harness.observer().await.unwrap(); + let topic = f.harness.coordinator.topics().snapshots.clone(); + let mut spectator = Spectator::attach(&observer, &topic, 1).await.unwrap(); + assert!(within("first snapshot", spectator.hold_one()).await); + + let steps = 5; + within("run", f.harness.coordinator.run(steps)).await.unwrap(); + + // The spectator's queue coalesced while it was not reading. + spectator.release(); + within("after release", spectator.hold_one()).await; + assert!(spectator.coalesced() > 0); + assert!( + spectator.seen() < steps + 1, + "a latest subscription does not deliver every boundary to a slow reader" + ); + + // Every agent's required input arrived once per boundary, in order, with nothing dropped + // or replaced. + for agent in [fly_a(), fly_b()] { + let entries = sensed(&f, &agent); + assert_eq!( + boundaries(&entries), + (0..=steps).collect::>(), + "{agent} encoded every boundary exactly once" + ); + } + f.shutdown().await; +} + +/// A persistent `AssetRef` names installed content; a transient `ArtifactRef` names live bytes +/// in one store. Importing the asset makes a new artifact, and collecting that artifact leaves +/// the asset installed. +async fn a_persistent_asset_and_a_transient_artifact_are_different_identities(via: Via) { + let mut f = default_fixture(via).await; + within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap(); + let client = f.harness.observer().await.unwrap(); + + let body = "counter-arena-backend-v1"; + let asset = synthetic_asset("counter-arena-backend", body); + let mut registry = AssetRegistry::new(); + registry + .install(asset.clone(), body.as_bytes().to_vec()) + .expect("installed content matches its reference"); + // Installing content that is not what the reference claims is refused. + let mut wrong = asset.clone(); + wrong.byte_length += 1; + registry + .install(wrong, body.as_bytes().to_vec()) + .expect_err("an asset reference is its content's identity"); + + let before = f.harness.router().stats().sealed_artifacts; + let artifact = registry.import(&client, &asset).await.expect("the import"); + assert_eq!( + f.harness.router().stats().sealed_artifacts, + before + 1, + "the import is a new object in the store" + ); + + // The identities are different, and the content is the same. + assert_ne!(artifact.reference().artifact_id, asset.id); + assert_eq!(artifact.reference().byte_length, asset.byte_length); + assert_eq!(artifact.reference().digest.as_deref(), Some(asset.digest.as_str())); + check_imported_asset(&asset, artifact.reference()).expect("the import carries the content"); + assert_eq!( + artifact.read_all().await.expect("readable"), + body.as_bytes(), + "the imported artifact is the installed bytes" + ); + + // The transient artifact is collected with its last handle; the asset is still installed. + drop(artifact); + let router = f.harness.router().clone(); + until("the imported artifact is collected", || { + router.stats().sealed_artifacts == before + }) + .await; + assert!(registry.contains(&asset)); + assert_eq!( + registry.resolve(&asset).expect("still installed"), + body.as_bytes(), + "a persistent asset outlives the bus objects imported from it" + ); + f.shutdown().await; +} + +/// A restored epoch resumes the preserved sample position, and its first chunk marks the +/// discontinuity that says so. The producer and the validator agree about which epoch it is. +async fn a_restored_audio_source_resumes_and_marks_the_discontinuity(via: Via) { + let f = default_fixture(via).await; + let client = f.harness.observer().await.unwrap(); + let descriptor = fly_session::environment::CounterEnvironment::audio_descriptor(); + let step = hz(f.harness.config.step_hz).expect("a positive cadence"); + let resumed_at = 2_400; + + let mut source = AudioSource::restored_at(descriptor.clone(), resumed_at); + let (first, _first_handle) = source + .produce(&client, &step, 3) + .await + .expect("the restored source produces"); + assert_eq!(first.first_sample, resumed_at, "the sample position is preserved"); + assert!(first.discontinuity, "the first chunk after a restore marks it"); + let (second, _second_handle) = source + .produce(&client, &step, 3) + .await + .expect("the next chunk"); + assert!(!second.discontinuity, "only the first chunk of the epoch marks it"); + assert_eq!(second.first_sample, resumed_at + first.sample_frames); + + // The validator accepts exactly this sequence under a restored timeline, and refuses it + // under a fresh one: the flag is what distinguishes the two epochs. + let mut restored = AudioTimeline::restored_at(&descriptor, resumed_at); + restored.accept(&first, &descriptor).expect("the restored epoch"); + restored.accept(&second, &descriptor).expect("and its next chunk"); + // A fresh episode at the audio origin refuses it: a restore preserves the sample + // position, and the position is what distinguishes the two epochs. The flag is required + // after a restore and free at an origin, so it cannot carry that distinction by itself. + let mut fresh = AudioTimeline::fresh(&descriptor, 0); + fresh + .accept(&first, &descriptor) + .expect_err("a fresh episode starts at its own origin, not a resumed position"); + f.shutdown().await; +} + +/// The sample budget is exact when the cadence does not divide the sample rate: 8 kHz at 60 Hz +/// is 133, 133, 134 and the total after three steps is exactly 400. +#[test] +fn the_sample_budget_is_exact_when_the_cadence_does_not_divide_the_rate() { + let descriptor = fly_session_types::media::AudioDescriptor { + stream_id: "arena".into(), + sample_rate: 8_000, + channels: 1, + }; + let mut source = AudioSource::new(descriptor, 0); + let step = hz(60).expect("a positive cadence"); + let frames: Vec = (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 752693b..48dc7d6 100644 --- a/services/flysim/crates/fly-session/tests/session.rs +++ b/services/flysim/crates/fly-session/tests/session.rs @@ -274,7 +274,11 @@ async fn the_committed_snapshot_names_the_boundary_that_just_ended(via: Via) { // The frame the snapshot names travels as an owned attachment. if step > 0 { let frame = message.artifact("view.arena").expect("the published frame"); - assert_eq!(frame.reference().byte_length, 4 * 4 * 4); + assert_eq!( + frame.reference().byte_length, + fly_session::environment::VIEW_WIDTH * fly_session::environment::VIEW_HEIGHT * 4, + "the published frame is the environment native frame" + ); } } assert_eq!(boundaries[0], (0, false), "boundary 0 has no decision or control"); diff --git a/services/flysim/crates/flybus/tests/sol_review_races.rs b/services/flysim/crates/flybus/tests/sol_review_races.rs index 5b93401..9d26b1b 100644 --- a/services/flysim/crates/flybus/tests/sol_review_races.rs +++ b/services/flysim/crates/flybus/tests/sol_review_races.rs @@ -714,6 +714,13 @@ async fn caller_disconnect_cleanup_works_before_and_after_consumption() { .await; caller.close().await; drop(pending); + // `Client::close` waits for this client to stop; the router marks the call detached when + // *it* observes the disconnect, in its own connection task. Asserting the reply routing + // before that is a race: under load the reply reaches the router first and is routed to a + // connection that is already closing. Nothing escapes -- teardown releases those roots -- + // but `routed` is then true. The contract sentence is about a reply to an already detached + // call, so the test waits for the teardown it is talking about. + e.settle("caller-a disconnected", |s| s.connections == 1).await; assert!(!responder.reply(obj(json!({})), &[]).await.unwrap()); drop(responder); e.settle("retained responder retired", |s| {