alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

media: no best-effort defaults in the audio path

Browse source 
Review round two.

A restored timeline refuses a declared stream with no recorded sample
position instead of resuming it at zero, which would have restarted the
episode's audio clock silently.

A declared stream that produces no chunk for a transition now fails the
step. The contract does not say a chunk must exist; this slice requires
one, because a silently missing chunk cannot be told from a lost one.
Boundary 0 is exempt: no interval precedes it.

The fresh-epoch discontinuity refusal is dropped rather than written into
the amendment. Only the restore direction is stated, and sections 6 and 7
both have a fresh timeline publishing a discontinuity after a recovery or
a reset, so refusing the flag at an origin contradicted them. The dated
amendment now says the requirement is one-directional.

A snapshot with an unreadable chunk, or one whose handle is not attached,
is now unreadable as a whole rather than quietly carrying less audio than
was published. ViewPipeline::is_bootstrap_repeat is removed.

New tests: a missing chunk fails the step, restored timelines need every
declared stream's position, and a 7 Hz world's seven chunks sum to exactly
one second of 48 kHz samples. The bus-conformance reconnect row also cites
the test that actually replaces an incarnation.
This commit is contained in:
acamilo 2026-09-22 15:15:06 +00:00
parent 903629db00
commit 079842f818
7 changed files with 216 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs/design/session-framework/bus-conformance.md
View file

@ -119,7 +119,7 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means
| Requirement | Status | Code | Test |
| --- | --- | --- | --- |
| `call-<U64>` with increasing serials per connected client; reused or retired ids are rejected, never executed again | conforms: a syntactically valid id advances the watermark even when admission is refused | `router/state.rs::op_call` (`call_watermark`) | `tests/sol_review_races.rs::rejected_call_id_still_advances_monotonic_watermark`, `tests/rpc.rs::raw_call_ids_and_forged_replies` (both) |
| Reconnecting creates a new incarnation rather than reviving old calls | conforms | `router/state.rs::{hello, disconnect}` | `tests/sol_review_races.rs::caller_disconnect_cleanup_works_before_and_after_consumption` (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) |

4
docs/design/session-framework/state-media-v1.md
View file

@ -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,

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

@ -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 {

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

@ -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");

3
services/flysim/crates/fly-session/src/coordinator.rs
View file

@ -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"));
}

85
services/flysim/crates/fly-session/src/media.rs
View file

@ -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<String, AudioTimeline>);
@ -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<String, u64>,
) -> 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<AudioTimelines> {
let mut timelines = BTreeMap::new();
for stream in &descriptor.audio {
let at = positions.get(&stream.stream_id).copied().ok_or_else(|| {
DomainError::before(
ErrorCode::IncompatibleState,
format!(
"audio stream {} has no restored sample position",
stream.stream_id
),
)
})?;
timelines.insert(stream.stream_id.clone(), AudioTimeline::restored_at(stream, at));
}
Ok(AudioTimelines(timelines))
}
/// Accepts one observation's chunks. Unknown streams and out-of-sequence chunks fail.
@ -681,16 +710,14 @@ pub fn snapshot_frame(message: &flybus::Message) -> Option<SpectatorFrame> {
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")

137
services/flysim/crates/fly-session/tests/media.rs
View file

@ -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::<u64>());
assert!(
chunk.sample_frames == 6_857 || chunk.sample_frames == 6_858,
"step {step} produced {} frames",
chunk.sample_frames
);
assert_eq!(chunk.samples.byte_length, chunk.sample_frames * CHANNELS * 4);
frames.push(chunk.sample_frames);
}
assert_eq!(
frames.iter().sum::<u64>(),
SAMPLE_RATE,
"seven steps of a 7 Hz world are exactly one second of samples: {frames:?}"
);
assert_eq!(f.harness.coordinator.audio_positions()[AUDIO_STREAM_ID], SAMPLE_RATE);
// The published chunk is readable and finite whatever the cadence.
let published = frame_at(&mut spectator, 7).await;
let (chunk, artifact) = published.audio.first().expect("the published chunk");
let bytes = artifact.read_all().await.expect("the published chunk");
assert_eq!(bytes.len() as u64, chunk.sample_frames * CHANNELS * 4);
require_finite_samples(&bytes).expect("native samples are finite f32");
f.shutdown().await;
}
/// The 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;
}