Pin the media FEC and impairment rules that mutation left standing

Adds the policy fields that must be non-zero, the inclusive shard limit at
the decoder for video and audio parity, admission statistics, seeded jitter
bounds, exact seeded loss, and arrival order for datagrams due together.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/cultnet-rs/tests/impairment_policy.rs

@@ -156,3 +156,64 @@ fn a_malformed_profile_is_refused_with_its_line() {
 delay_ms = x
 ").contains(":3:"));
 }
+
+#[test]
+fn admission_counts_what_it_did_to_each_datagram() {
+    let mut scheduler = Scheduler::new(
+        Profile { duplicate_every: 2, reorder_every: 3, reorder_delay_ms: 10, ..Default::default() },
+        1,
+    );
+    for value in 0..6 {
+        scheduler.admit(0, Direction::ClientToUpstream, &[value], client());
+    }
+    // Datagrams 2, 4 and 6 are duplicated, 3 and 6 are reordered (6 is both).
+    assert_eq!(scheduler.stats.received, 6);
+    assert_eq!(scheduler.stats.duplicated, 3);
+    assert_eq!(scheduler.stats.reordered, 2);
+    assert_eq!(scheduler.queue.len(), 9);
+
+    let mut stalled = Scheduler::new(Profile { stall_at_ms: 0, stall_for_ms: 10, ..Default::default() }, 1);
+    stalled.admit(5, Direction::UpstreamToClient, b"x", client());
+    assert_eq!((stalled.stats.stalled, stalled.stats.dropped, stalled.queue.len()), (1, 1, 0));
+}
+
+#[test]
+fn jitter_is_seeded_uniform_within_its_bound() {
+    let profile = Profile { delay_ms: 7, jitter_ms: 5, ..Default::default() };
+    let mut scheduler = Scheduler::new(profile, 9);
+    let delays: Vec<u64> = (0..300).map(|_| scheduler.decide(0).delay_ms).collect();
+    assert!(delays.iter().all(|delay| (7..=12).contains(delay)), "{delays:?}");
+    for delay in 7..=12 {
+        assert!(delays.contains(&delay), "delay {delay} never drawn");
+    }
+}
+
+#[test]
+fn the_loss_ceiling_is_inclusive() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("all.toml");
+    std::fs::write(&path, "loss_basis_points = 10000\n").unwrap();
+    let profile = Profile::load(&path).unwrap();
+    let mut scheduler = Scheduler::new(profile, 1);
+    assert!((0..50).all(|_| scheduler.decide(0).drop), "10000 basis points drops everything");
+}
+
+/// The seeded stream is part of the contract: the same seed must drop the same
+/// datagrams on every run, on every host. 100,000 draws of xorshift64 from seed
+/// 42, taken modulo 10,000 and compared against 100, give 1,008.
+#[test]
+fn seeded_loss_drops_the_exact_datagrams_the_stream_dictates() {
+    let mut scheduler = Scheduler::new(Profile { loss_basis_points: 100, ..Default::default() }, 42);
+    let dropped = (0..100_000).filter(|_| scheduler.decide(0).drop).count();
+    assert_eq!(dropped, 1_008);
+}
+
+#[test]
+fn datagrams_due_together_leave_in_arrival_order() {
+    let mut scheduler = Scheduler::new(Profile { delay_ms: 5, ..Default::default() }, 1);
+    for value in 0..4_u8 {
+        scheduler.admit(0, Direction::ClientToUpstream, &[value], client());
+    }
+    let order: Vec<u8> = std::iter::from_fn(|| scheduler.pop_due(5)).map(|item| item.bytes[0]).collect();
+    assert_eq!(order, vec![0, 1, 2, 3]);
+}

packages/cultnet-rs/tests/media_fec.rs

@@ -768,6 +768,12 @@ fn a_custom_policy_sizes_blocks_by_its_own_fields() {
 fn a_policy_that_cannot_form_a_block_is_refused() {
     let zero = MediaFecPolicy { video_min_parity_shards: 0, ..STANDARD };
     assert!(protect_video_frame(&frame(1, 2, 10, 10), &zero).is_err());
+    let no_divisor = MediaFecPolicy { video_parity_divisor: 0, ..STANDARD };
+    assert!(protect_video_frame(&frame(1, 2, 10, 10), &no_divisor).is_err());
+    let no_audio = MediaFecPolicy { audio_data_shards: 0, ..STANDARD };
+    assert!(protect_audio_block(&audio_block(1, 10), &no_audio).is_err());
+    let no_audio_parity = MediaFecPolicy { audio_parity_shards: 0, ..STANDARD };
+    assert!(protect_audio_block(&audio_block(1, 10), &no_audio_parity).is_err());
     let too_wide = MediaFecPolicy { video_max_block_data_shards: 255, ..STANDARD };
     assert!(protect_video_frame(&frame(1, 2, 10, 10), &too_wide).is_err());
     let audio_too_wide = MediaFecPolicy { audio_data_shards: 255, audio_parity_shards: 2, ..STANDARD };

packages/cultnet-rs/tests/media_stream_wire.rs

@@ -383,3 +383,19 @@ fn a_malformed_audio_parity_record_is_refused_with_its_reason() {
         assert!(error.contains(expected), "expected {expected:?} in {error:?}");
     }
 }
+
+/// The shard limit is inclusive: a shard of exactly the largest size is
+/// admitted, one byte more is not.
+#[test]
+fn a_parity_shard_at_the_size_limit_is_admitted() {
+    use cultnet_rs::MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES as MAX;
+    let mut video = video_parity();
+    video.shard_payload_bytes = MAX as u32;
+    video.last_chunk_payload_bytes = MAX as u32;
+    video.payload = vec![0; MAX];
+    assert!(decodes(GameCultMediaWireRecord::VideoParity(video)).is_ok());
+    let mut audio = audio_parity();
+    audio.shard_payload_bytes = MAX as u32;
+    audio.payload = vec![0; MAX];
+    assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok());
+}