diff --git a/docs/design/session-framework/bus-conformance.md b/docs/design/session-framework/bus-conformance.md index 68fa08c..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` (its synchronisation was fixed on 2026-09-22; see "A flaky test and what it was measuring") | +| 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) | diff --git a/docs/design/session-framework/state-media-v1.md b/docs/design/session-framework/state-media-v1.md index 677f85d..2050221 100644 --- a/docs/design/session-framework/state-media-v1.md +++ b/docs/design/session-framework/state-media-v1.md @@ -78,7 +78,9 @@ because they are now enforced: 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. + 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 and + section 7's episode reset both establish a fresh timeline and publish one. The environment provides **native game output**. Sensor transformations belong to the agent profile. Resizing for viewers, overlays, composition, audio mixing/resampling, encoding, diff --git a/services/flysim/crates/fly-session-types/src/media.rs b/services/flysim/crates/fly-session-types/src/media.rs index 71a0856..5b5fbfd 100644 --- a/services/flysim/crates/fly-session-types/src/media.rs +++ b/services/flysim/crates/fly-session-types/src/media.rs @@ -389,8 +389,9 @@ pub struct AudioTimeline { } impl AudioTimeline { - /// A fresh episode: the first chunk starts at the configured audio origin and is not a - /// discontinuity, because nothing preceded it. + /// 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(), @@ -442,18 +443,15 @@ impl AudioTimeline { self.stream_id, self.start_sample, chunk.first_sample )); } - if chunk.discontinuity != self.restored { - return err(if self.restored { - format!( - "AudioTimeline {}: the first chunk after a restore marks discontinuity", - self.stream_id - ) - } else { - format!( - "AudioTimeline {}: the first chunk at the episode's audio origin is not a discontinuity", - self.stream_id - ) - }); + // 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 { diff --git a/services/flysim/crates/fly-session-types/tests/media_shapes.rs b/services/flysim/crates/fly-session-types/tests/media_shapes.rs index 29c0f81..f834989 100644 --- a/services/flysim/crates/fly-session-types/tests/media_shapes.rs +++ b/services/flysim/crates/fly-session-types/tests/media_shapes.rs @@ -297,10 +297,14 @@ fn chunks_cannot_overlap_or_go_backwards_within_an_epoch() { #[test] fn the_first_chunk_after_a_restore_marks_discontinuity() { let descriptor = audio_descriptor(48_000, 2); - let mut fresh = AudioTimeline::fresh(&descriptor, 0); - fresh + // 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_err("the episode's first chunk is not a discontinuity"); + .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"); @@ -310,6 +314,7 @@ fn the_first_chunk_after_a_restore_marks_discontinuity() { 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"); diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index def9d5d..4ef12ba 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -1716,6 +1716,9 @@ impl Coordinator { } // 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")); } diff --git a/services/flysim/crates/fly-session/src/media.rs b/services/flysim/crates/fly-session/src/media.rs index d2c603b..870ead9 100644 --- a/services/flysim/crates/fly-session/src/media.rs +++ b/services/flysim/crates/fly-session/src/media.rs @@ -216,11 +216,6 @@ impl ViewPipeline { self.renders.count() } - /// Whether `boundary` is still inside the declared pipeline delay, where the contract - /// allows `O[0]` to repeat. - pub fn is_bootstrap_repeat(&self, boundary: u64) -> bool { - boundary > 0 && boundary <= self.descriptor.observation_delay_steps - } } /// One audio stream's production: an exact sample budget and a deterministic waveform. @@ -444,6 +439,35 @@ pub fn check_required_views( 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); @@ -462,23 +486,28 @@ impl AudioTimelines { /// 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, - ) -> AudioTimelines { - AudioTimelines( - descriptor - .audio - .iter() - .map(|stream| { - let at = positions.get(&stream.stream_id).copied().unwrap_or_default(); - ( - stream.stream_id.clone(), - AudioTimeline::restored_at(stream, at), - ) - }) - .collect(), - ) + ) -> 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. @@ -681,16 +710,14 @@ pub fn snapshot_frame(message: &flybus::Message) -> Option { let media = payload.get("media")?; let views = media.get("views")?.as_array()?; let view = ViewRef::from_json(views.first()?).ok()?; - let audio = media - .get("audio")? - .as_array()? - .iter() - .filter_map(|v| AudioRef::from_json(v).ok()) - .filter_map(|chunk| { - let artifact = message.artifact(&audio_attachment(&chunk.stream_id)).ok()?; - Some((chunk, artifact)) - }) - .collect(); + // 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") diff --git a/services/flysim/crates/fly-session/tests/media.rs b/services/flysim/crates/fly-session/tests/media.rs index 80308f8..61925bf 100644 --- a/services/flysim/crates/fly-session/tests/media.rs +++ b/services/flysim/crates/fly-session/tests/media.rs @@ -8,6 +8,7 @@ mod common; +use std::collections::BTreeMap; use std::time::Duration; use serde_json::Map; @@ -19,8 +20,8 @@ use fly_session::environment::{ }; use fly_session::harness::{HarnessConfig, Via}; use fly_session::media::{ - AssetRegistry, AudioSource, SensedView, Spectator, SpectatorFrame, arena_frame, - audio_attachment, detach_frame, view_attachment, + AssetRegistry, AudioSource, AudioTimelines, SensedView, Spectator, SpectatorFrame, + arena_frame, audio_attachment, detach_frame, view_attachment, }; use fly_session::phase::Phase; use fly_session::types::*; @@ -39,6 +40,9 @@ both_transports!( 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, ); const STEPS: u64 = 3; @@ -385,6 +389,128 @@ async fn one_audio_chunk_per_boundary_with_an_exact_sample_budget(via: Via) { 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 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) { @@ -504,10 +630,13 @@ async fn a_restored_audio_source_resumes_and_marks_the_discontinuity(via: Via) { 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"); - let mut fresh = AudioTimeline::fresh(&descriptor, resumed_at); + // 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's first chunk is not a discontinuity"); + .expect_err("a fresh episode starts at its own origin, not a resumed position"); f.shutdown().await; }