moq-mux 0.10.13

//! Tests for the MPEG-TS exporter.
//!
//! AAC audio is framed as ADTS (MP2/AC-3 pass through as whole frames); video
//! is normalized to length-prefixed NALU by
//! `ExportSource` and rewritten to Annex-B by the muxer (re-injecting the
//! parameter sets on keyframes). These build a synthetic broadcast, export to
//! TS, and re-parse with the `mpeg2ts` reader.

use std::io::Cursor;
use std::time::Duration;

use bytes::{Bytes, BytesMut};
use hang::catalog::{AAC, AudioCodec, AudioConfig, Container, H264, VideoConfig};
use mpeg2ts::es::StreamType;
use mpeg2ts::pes::{PesPacketReader, ReadPesPacket};
use mpeg2ts::ts::{ReadTsPacket, TsPacketReader, TsPayload};

use crate::catalog::hang::Container as HangContainer;
use crate::container::ts::export::PCR_INTERVAL;
use crate::container::ts::{Export, Stats, catalog as tscat};
use crate::container::{Frame, Kind, Producer};
use moq_net::Timestamp;

const SC: &[u8] = &[0, 0, 0, 1];
// Reusable H.264 parameter-set and slice NALs (NAL type = first byte & 0x1f).
const SPS: &[u8] = &[0x67, 0x42, 0xc0, 0x1f, 0xde];
const PPS: &[u8] = &[0x68, 0xce, 0x3c, 0x80];
// A second, distinct PPS (id 1): broadcast feeds often define more than one.
const PPS1: &[u8] = &[0x68, 0xce, 0x3c, 0x81];

// libklvanc public-sample SCTE-35 cue: splice_info_section, table_id 0xFC, 30 bytes.
const CUE: &[u8] = &[
	0xfc, 0x30, 0x1b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xff, 0xf0, 0x0a, 0x05, 0x00, 0x00, 0x2b, 0xb4, 0x7f,
	0xdf, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0xad, 0x25, 0xe8, 0x39,
];

/// Concatenate NALs into an Annex-B buffer (4-byte start code before each).
fn annexb(nals: &[&[u8]]) -> Bytes {
	let mut buf = BytesMut::new();
	for nal in nals {
		buf.extend_from_slice(SC);
		buf.extend_from_slice(nal);
	}
	buf.freeze()
}

/// Concatenate NALs into a length-prefixed (avc1/hvc1) buffer (4-byte big-endian
/// length before each), the wire shape of an out-of-band source.
fn length_prefixed(nals: &[&[u8]]) -> Bytes {
	let mut buf = BytesMut::new();
	for nal in nals {
		buf.extend_from_slice(&(nal.len() as u32).to_be_bytes());
		buf.extend_from_slice(nal);
	}
	buf.freeze()
}

/// Drive an exporter until it stops producing output, concatenating every chunk.
///
/// The broadcast producers stay alive so the exporter can subscribe to the
/// finished, retained tracks; that means it never reaches a hard end-of-stream,
/// so we pull until a `next()` blocks (`Pending`, surfaced as a timeout under
/// paused time) or the stream ends.
/// A drift budget no test timeline comes close to, so the exporter reads every group.
///
/// The media track's full retention window, so an exporter started after publishing
/// can still read every retained group. These tests write a whole broadcast up front
/// and only then export it, which the
/// exporter's default [`Duration::ZERO`] collapses to the
/// live edge: completeness has to be asked for, exactly as a real recorder does.
const RECORDING_MAX_AGE: Duration = Duration::from_secs(30);

async fn drain(consumer: moq_net::broadcast::Consumer) -> BytesMut {
	drain_with(Export::new(crate::source::announced(&consumer)).await.unwrap()).await
}

/// `drain` for an exporter built with an explicit catalog extension.
async fn drain_with<E: tscat::Catalog>(exporter: Export<E>) -> BytesMut {
	let mut exporter = exporter.with_max_age(RECORDING_MAX_AGE);
	let mut out = BytesMut::new();
	// `while let Ok` stops on the first timeout (`Pending`: no more output).
	while let Ok(res) = tokio::time::timeout(Duration::from_secs(1), exporter.next()).await {
		let Some(frame) = res.expect("exporter error") else {
			break;
		};
		out.extend_from_slice(&frame.payload);
	}
	out
}

/// An adaptation-field-only single-packet frame: the exporter's PCR carriage.
fn is_pcr_frame(frame: &Frame) -> bool {
	frame.payload.len() == 188 && frame.payload[3] & 0x30 == 0x20
}

fn assert_packet_aligned(ts: &[u8]) {
	assert!(!ts.is_empty(), "no TS output");
	assert_eq!(ts.len() % 188, 0, "output not a whole number of 188-byte packets");
	assert!(
		ts.chunks(188).all(|p| p[0] == 0x47),
		"every packet must start with the sync byte"
	);
}

#[tokio::test(start_paused = true)]
async fn export_aac_roundtrip() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().audio.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	// The last frame is > 184 bytes to force PES splitting across TS packets.
	let frames: Vec<Bytes> = vec![
		Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]),
		Bytes::from_static(&[0x10, 0x11, 0x12, 0x13, 0x14]),
		Bytes::from(vec![0x20u8; 200]),
	];
	for (i, payload) in frames.iter().enumerate() {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i as u64 * 20_000).unwrap(),
				duration: None,
				payload: payload.clone(),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	}
	producer.finish().unwrap();

	// The producers stay alive so the exporter can subscribe to the catalog and
	// the finished (retained) track; `drain` stops once all frames are emitted.
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// Pass 1: the program tables advertise exactly one ADTS AAC stream.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut saw_pat = false;
	let mut saw_pmt = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		match packet.payload {
			Some(TsPayload::Pat(_)) => saw_pat = true,
			Some(TsPayload::Pmt(pmt)) => {
				saw_pmt = true;
				assert_eq!(pmt.es_info.len(), 1);
				assert_eq!(pmt.es_info[0].stream_type, StreamType::AdtsAac);
			}
			_ => {}
		}
	}
	assert!(saw_pat, "missing PAT");
	assert!(saw_pmt, "missing PMT");

	// Pass 2: reassemble PES packets and recover the original raw AAC frames.
	let mut pes = PesPacketReader::new(TsPacketReader::new(Cursor::new(ts.as_ref())));
	let mut recovered: Vec<(u64, Vec<u8>)> = Vec::new();
	while let Some(packet) = pes.read_pes_packet().unwrap() {
		let pts = packet.header.pts.expect("PES carried no PTS").as_u64();
		// Strip the 7-byte ADTS header we added on export.
		assert!(packet.data.len() >= 7, "PES payload shorter than an ADTS header");
		recovered.push((pts, packet.data[7..].to_vec()));
	}

	assert_eq!(recovered.len(), frames.len());
	for (i, payload) in frames.iter().enumerate() {
		let (pts, raw) = &recovered[i];
		assert_eq!(*pts, i as u64 * 20 * 90, "PTS should be ms * 90 (90 kHz)");
		assert_eq!(raw.as_slice(), payload.as_ref(), "raw AAC payload mismatch");
	}
}

/// Collect PES presentation timestamps per elementary stream (video H.264, audio AAC),
/// keyed off the PMT's PID assignments.
fn collect_pes_pts(ts: &[u8]) -> (Vec<u64>, Vec<u64>) {
	let mut reader = TsPacketReader::new(Cursor::new(ts));
	let (mut video_pid, mut audio_pid) = (None, None);
	let (mut video, mut audio) = (Vec::new(), Vec::new());
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		match packet.payload {
			Some(TsPayload::Pmt(pmt)) => {
				for es in &pmt.es_info {
					match es.stream_type {
						StreamType::H264 => video_pid = Some(es.elementary_pid),
						StreamType::AdtsAac => audio_pid = Some(es.elementary_pid),
						_ => {}
					}
				}
			}
			Some(TsPayload::PesStart(pes)) => {
				if let Some(pts) = pes.header.pts {
					let pid = Some(packet.header.pid);
					if pid == video_pid {
						video.push(pts.as_u64());
					} else if pid == audio_pid {
						audio.push(pts.as_u64());
					}
				}
			}
			_ => {}
		}
	}
	(video, audio)
}

/// Build a broadcast whose audio begins before the first video keyframe (the shape a
/// mid-stream tune-in produces: the audio source is cached further back than the oldest
/// retained video keyframe), then export it to TS.
async fn export_lead_audio() -> BytesMut {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	// In-band avc3 video (SPS/PPS inline on keyframes; no out-of-band description).
	let vtrack = broadcast
		.create_track(
			broadcast.unique_name(".avc3"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: true,
		});
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.video
			.renditions
			.insert(vtrack.name().to_string(), cfg);
	}
	let mut video = Producer::new(vtrack, HangContainer::Legacy(crate::container::Kind::Data));

	let atrack = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(atrack.name().to_string(), cfg);
	}
	let mut audio = Producer::new(atrack, HangContainer::Legacy(crate::container::Kind::Data));

	let audio_frame = |ms: u64| Frame {
		timestamp: Timestamp::from_micros(ms * 1_000).unwrap(),
		duration: None,
		payload: Bytes::from(vec![0xAAu8; 16]),
		keyframe: true,
	};
	// Lead audio (0..80 ms) precedes the first video keyframe at 100 ms; both continue after.
	for ms in [0, 20, 40, 60, 80] {
		audio.write(audio_frame(ms)).unwrap();
		audio.cut(None).unwrap();
	}
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 200));
	video
		.write(Frame {
			timestamp: Timestamp::from_micros(100_000).unwrap(),
			duration: None,
			payload: annexb(&[SPS, PPS, &idr]),
			keyframe: true,
		})
		.unwrap();
	video.cut(None).unwrap();
	for ms in [100, 120, 140] {
		audio.write(audio_frame(ms)).unwrap();
		audio.cut(None).unwrap();
	}
	video.finish().unwrap();
	audio.finish().unwrap();

	let exporter = Export::new(crate::source::announced(&consumer)).await.unwrap();
	// The producers stay alive through the drain so the retained tracks are readable.
	drain_with(exporter).await
}

/// The exported stream must begin at the first video keyframe. On a mid-stream tune-in the
/// audio source can lead the first cached video keyframe by over a second; emitting that
/// audio first buries the in-band SPS/PPS behind an audio-only preamble, and a live decoder
/// probing the stream gives up before it ever configures video (RTMP/CMAF carry the codec
/// config out-of-band, so they don't hit this). The muxer drops the lead audio so the
/// keyframe leads. Audio from the keyframe onward is still carried.
#[tokio::test(start_paused = true)]
async fn export_starts_at_video_keyframe() {
	// 100 ms (the keyframe PTS) in 90 kHz ticks.
	const KEYFRAME_PTS: u64 = 100 * 90;

	let ts = export_lead_audio().await;
	assert_packet_aligned(&ts);
	let (video, audio) = collect_pes_pts(&ts);

	assert_eq!(
		video.first(),
		Some(&KEYFRAME_PTS),
		"the stream must begin at the video keyframe"
	);
	assert!(
		audio.iter().all(|&p| p >= KEYFRAME_PTS),
		"lead audio before the first keyframe must be dropped, got {audio:?}"
	);
	assert!(!audio.is_empty(), "audio from the keyframe onward is still carried");
}

/// Re-parse a TS byte stream: assert the single video stream type, that the
/// keyframe carries random-access + PCR in an unbounded PES, and return the
/// reassembled Annex-B elementary stream.
fn reassemble_video(ts: &[u8], expected_stream_type: StreamType) -> Vec<u8> {
	let mut reader = TsPacketReader::new(Cursor::new(ts));
	let mut video_pid = None;
	let mut saw_random_access = false;
	let mut saw_pcr = false;
	let mut reassembled: Vec<u8> = Vec::new();
	let mut unbounded = false;

	while let Some(packet) = reader.read_ts_packet().unwrap() {
		match packet.payload {
			Some(TsPayload::Pmt(pmt)) => {
				assert_eq!(pmt.es_info.len(), 1);
				assert_eq!(pmt.es_info[0].stream_type, expected_stream_type);
				video_pid = Some(pmt.es_info[0].elementary_pid);
			}
			Some(TsPayload::PesStart(pes)) => {
				// The first packet of a keyframe must signal random access.
				if let Some(af) = &packet.adaptation_field {
					saw_random_access |= af.random_access_indicator;
				}
				unbounded = pes.pes_packet_len == 0;
				reassembled.extend_from_slice(&pes.data);
			}
			Some(TsPayload::PesContinuation(bytes)) => reassembled.extend_from_slice(&bytes),
			// The clock rides adaptation-field-only packets on the PCR PID.
			None => saw_pcr |= packet.adaptation_field.as_ref().is_some_and(|af| af.pcr.is_some()),
			_ => {}
		}
	}

	assert!(video_pid.is_some(), "missing video PMT entry");
	assert!(saw_random_access, "keyframe should set random_access_indicator");
	assert!(saw_pcr, "PCR pid should carry the clock");
	assert!(unbounded, "video PES should be unbounded");
	reassembled
}

/// In-band avc3: SPS/PPS are inline in the bitstream. ExportSource strips them
/// into a synthesized avcC, and the muxer re-injects them on the keyframe.
#[tokio::test(start_paused = true)]
async fn export_avc3_in_band_reassembles() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc3"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: true,
		});
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().video.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	// IDR slice (NAL type 5), padded past 184 bytes to span multiple TS packets.
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	// Annex-B keyframe: inline SPS + PPS + IDR.
	producer
		.write(Frame {
			timestamp: Timestamp::from_micros(0).unwrap(),
			duration: None,
			payload: annexb(&[SPS, PPS, &idr]),
			keyframe: true,
		})
		.unwrap();
	producer.finish().unwrap();

	// Keep the producers alive (see `export_aac_roundtrip`).
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let reassembled = reassemble_video(&ts, StreamType::H264);
	// The parameter sets the muxer re-injected, followed by the slice, all Annex-B.
	assert_eq!(reassembled.as_slice(), annexb(&[SPS, PPS, &idr]).as_ref());
}

/// In-band avc3 carrying two distinct PPS (a real broadcast trait): both must
/// survive the round-trip, or slices referencing the dropped one stop decoding
/// (regression for non-existing PPS 0 referenced).
#[tokio::test(start_paused = true)]
async fn export_avc3_preserves_multiple_pps() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc3"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: true,
		});
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().video.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	// Annex-B keyframe: inline SPS + both PPS + IDR.
	producer
		.write(Frame {
			timestamp: Timestamp::from_millis(0).unwrap(),
			duration: None,
			payload: annexb(&[SPS, PPS, PPS1, &idr]),
			keyframe: true,
		})
		.unwrap();
	producer.finish().unwrap();

	// Keep the producers alive (see `export_aac_roundtrip`).
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let reassembled = reassemble_video(&ts, StreamType::H264);
	// Both PPS must be re-injected on the keyframe, in order, ahead of the slice.
	assert_eq!(reassembled.as_slice(), annexb(&[SPS, PPS, PPS1, &idr]).as_ref());
}

/// Out-of-band avc1 (e.g. from fmp4 import): length-prefixed NALs with the
/// SPS/PPS only in the catalog `description` (avcC). The muxer must parse the
/// avcC, prepend the parameter sets as Annex-B on the keyframe, and rewrite the
/// length prefixes to start codes.
#[tokio::test(start_paused = true)]
async fn export_avc1_out_of_band_reassembles() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		catalog.modify().unwrap().video.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	// IDR slice (NAL type 5), padded past 184 bytes to span multiple TS packets.
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	// Length-prefixed keyframe: just the slice, no inline parameter sets.
	producer
		.write(Frame {
			timestamp: Timestamp::from_micros(0).unwrap(),
			duration: None,
			payload: length_prefixed(&[&idr]),
			keyframe: true,
		})
		.unwrap();
	producer.finish().unwrap();

	// Keep the producers alive (see `export_aac_roundtrip`).
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let reassembled = reassemble_video(&ts, StreamType::H264);
	// SPS/PPS from the avcC must precede the slice, all converted to Annex-B.
	assert_eq!(reassembled.as_slice(), annexb(&[SPS, PPS, &idr]).as_ref());
}

/// H.265 suffix SEI must survive the container seam in both directions: the exporter
/// puts each access unit on the wire whole, and the importer's splitter gives it back
/// unchanged. A splitter that closed the access unit on a suffix SEI would hand it to
/// the next picture, and drop the last one entirely when the stream ends, which a unit
/// test on the splitter alone cannot see once the bytes cross a PES boundary.
#[tokio::test(start_paused = true)]
async fn export_import_h265_keeps_suffix_sei_on_its_picture() {
	use crate::codec::h265::fixtures::{PPS, SPS, VPS};

	// HEVC NAL headers: byte 0 = nal_unit_type << 1. Slices set
	// first_slice_segment_in_pic_flag (byte 2 high bit).
	const IDR: &[u8] = &[0x26, 0x01, 0x80, 0xaa]; // IdrWRadl (19)
	const TRAIL: &[u8] = &[0x02, 0x01, 0x80, 0x33]; // TrailR (1)
	const AUD: &[u8] = &[0x46, 0x01, 0x50]; // AudNut (35)
	const PREFIX_SEI: &[u8] = &[0x4e, 0x01, 0x01, 0x04, 0x80]; // PrefixSeiNut (39)
	const SUFFIX_SEI: &[u8] = &[0x50, 0x01, 0x84, 0x02, 0x80]; // SuffixSeiNut (40)
	const SUFFIX_SEI2: &[u8] = &[0x50, 0x01, 0x05, 0x03, 0x80]; // a second SuffixSeiNut

	// Three access units covering every shape: multiple suffix units, a prefix SEI
	// immediately after a suffix, consecutive pictures, and a suffix at end of stream.
	let units: [Vec<&[u8]>; 3] = [
		vec![VPS, SPS, PPS, IDR, SUFFIX_SEI, SUFFIX_SEI2],
		vec![PREFIX_SEI, TRAIL, SUFFIX_SEI],
		vec![AUD, TRAIL, SUFFIX_SEI],
	];

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".hev1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		// hev1: the config comes from the SPS the keyframe carries inline.
		let mut cfg = crate::codec::h265::config(&annexb(&units[0])).unwrap();
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().video.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	for (i, nals) in units.iter().enumerate() {
		producer
			.write(Frame {
				timestamp: Timestamp::from_millis(i as u64 * 40).unwrap(),
				duration: None,
				payload: annexb(nals),
				keyframe: i == 0,
			})
			.unwrap();
	}
	producer.finish().unwrap();

	// Keep the producers alive (see `export_aac_roundtrip`).
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// Export: the elementary stream is the three access units back to back, every SEI
	// still trailing the picture it was written with.
	let all: Vec<&[u8]> = units.iter().flatten().copied().collect();
	let reassembled = reassemble_video(&ts, StreamType::H265);
	assert_eq!(reassembled.as_slice(), annexb(&all).as_ref());

	// Import: the same TS back through the demuxer must rebuild the same access units.
	let mut imported = moq_net::broadcast::Info::new().produce();
	let imported_consumer = imported.consume();
	let import_catalog = crate::catalog::Producer::new(&mut imported, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(imported, import_catalog.reserve());
	import.decode(&ts).unwrap();
	import.finish().unwrap();

	let snapshot = import_catalog.snapshot();
	let (imported_name, video) = snapshot.video.renditions.iter().next().expect("an H.265 rendition");
	assert!(video.codec.to_string().starts_with("hev1"), "codec was {}", video.codec);

	let recovered = read_frames(&imported_consumer, imported_name, Kind::Video).await;
	assert_eq!(recovered.len(), units.len(), "access unit count");
	for (i, (got, nals)) in recovered.iter().zip(&units).enumerate() {
		assert_eq!(got.as_slice(), annexb(nals).as_ref(), "access unit {i}");
	}
}

/// A real broadcast contribution feed (Ateme Kyrion, H.264 1080i with ~86 B-frames)
/// must come out of the exporter with an authored decode timeline. The importer publishes
/// the reorder depth as the catalog `jitter`, and the exporter sizes its decode-clock reserve
/// from it, so the video PES carry a DTS that is both strictly increasing and never after the
/// PTS in decode order. Also assert the reorder was real (non-monotonic PTS in the source).
#[tokio::test(start_paused = true)]
async fn export_bframe_video_authors_dts() {
	let data = include_bytes!("test_data/scte35/kyrion_dirtystart.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	// `import` and `catalog` stay alive: retained tracks the exporter subscribes to.
	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// Collect (pts, dts) for the H.264 video PID in transport (decode) order.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut video_pid = None;
	let mut pts = Vec::new();
	let mut authored = 0usize;
	let mut effective = Vec::new();
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		match packet.payload {
			Some(TsPayload::Pmt(pmt)) if video_pid.is_none() => {
				video_pid = pmt
					.es_info
					.iter()
					.find(|e| e.stream_type == StreamType::H264)
					.map(|e| e.elementary_pid);
			}
			Some(TsPayload::PesStart(pes)) if Some(packet.header.pid) == video_pid => {
				let p = pes.header.pts.expect("video PES carried no PTS").as_u64();
				let d = pes.header.dts.map(|t| t.as_u64());
				if d.is_some() {
					authored += 1;
				}
				effective.push(d.unwrap_or(p));
				pts.push(p);
			}
			_ => {}
		}
	}

	assert!(video_pid.is_some(), "missing H.264 video PMT entry");
	assert!(pts.len() > 50, "expected the full feed, got {} frames", pts.len());
	// The source is genuinely reordered: PTS dips in decode order (B-frames).
	assert!(
		pts.windows(2).any(|w| w[1] < w[0]),
		"fixture must carry reordered B-frames"
	);
	// The exporter authored a decode timeline (the decode clock trails the PTS).
	assert!(authored > 0, "no DTS authored for a B-frame stream");
	// Strictly increasing (removes the `+igndts` requirement) and never after presentation
	// (the catalog jitter sized the reserve to the reorder depth).
	for (i, win) in effective.windows(2).enumerate() {
		assert!(win[1] > win[0], "DTS not strictly increasing at frame {i}: {win:?}");
	}
	for (i, (&d, &p)) in effective.iter().zip(pts.iter()).enumerate() {
		assert!(d <= p, "DTS {d} after PTS {p} at frame {i}");
	}
}

/// #2937: the PCR must be a uniform bounded-interval ramp, not a sample of the
/// per-unit decode clock. On a reordered (B-frame) capture the authored DTS is a
/// saw (reference frames leap a reorder span, B-frames nudge one tick), so a PCR
/// taken from it froze and jumped: most intervals landed within microseconds of
/// each other, the rest collected into gaps far over TR 101 290's 40 ms gate.
#[tokio::test(start_paused = true)]
async fn export_pcr_is_a_uniform_ramp() {
	let data = include_bytes!("test_data/scte35/kyrion_dirtystart.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// Walk the output in transport order: PCR values (90 kHz) as they appear, and
	// every PES unit's effective decode time against the clock preceding it.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut pcr_pid = None;
	let mut pcr_pids = Vec::new();
	let mut pcrs: Vec<u64> = Vec::new();
	let mut units = 0usize;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(pcr) = packet.adaptation_field.as_ref().and_then(|af| af.pcr) {
			// The clock leads the stream, so the first PCR precedes the PMT that
			// names its PID; collect and check at the end.
			pcr_pids.push(packet.header.pid);
			assert!(packet.payload.is_none(), "PCR rides adaptation-field-only packets");
			pcrs.push(pcr.as_u64() / 300);
		}
		match packet.payload {
			Some(TsPayload::Pmt(pmt)) if pcr_pid.is_none() => pcr_pid = pmt.pcr_pid,
			Some(TsPayload::PesStart(pes)) => {
				units += 1;
				let pts = pes.header.pts.expect("PES carried no PTS").as_u64();
				let decode = pes.header.dts.map(|t| t.as_u64()).unwrap_or(pts);
				if let Some(&pcr) = pcrs.last() {
					assert!(decode >= pcr, "unit decodes at {decode}, before the clock at {pcr}");
				}
			}
			_ => {}
		}
	}

	assert!(units > 50, "expected the full feed, got {units} PES units");
	assert!(pcrs.len() > 50, "expected a dense clock, got {} PCRs", pcrs.len());
	let pcr_pid = pcr_pid.expect("PMT must announce a PCR PID");
	assert!(
		pcr_pids.iter().all(|&pid| pid == pcr_pid),
		"PCR must ride the announced PID"
	);
	// One grid step apart, exactly: uniform, monotonic, and far under the 40 ms gate.
	let step = Duration::from_millis(25).as_micros() as u64 * 90 / 1_000;
	for (i, w) in pcrs.windows(2).enumerate() {
		assert_eq!(w[1] - w[0], step, "PCR interval off the grid at {i}: {w:?}");
	}
}

/// A timeline that starts inside the decode-clock reserve backs the PCR off
/// through the 33-bit wrap instead of saturating at zero: the grid step stays
/// uniform from the very first slot (saturation would emit 0 then 2234, and a
/// large catalog jitter would freeze several leading PCRs at zero).
#[tokio::test(start_paused = true)]
async fn export_pcr_wraps_below_the_reserve_at_start() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(track.name().to_string(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	for i in 0..4u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i * 20_000).unwrap(),
				duration: None,
				payload: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]),
				keyframe: true,
			})
			.unwrap();
	}
	producer.finish().unwrap();

	let mut exporter = Export::new(crate::source::announced(&consumer))
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	let frames = drain_frames(&mut exporter).await;

	// A slot's clock packets lead the frame carrying that slot's bytes, and the
	// frame is stamped at the slot boundary so the caller's pacer delivers the
	// clock at the time it asserts.
	let mut pcrs: Vec<u64> = Vec::new();
	for (i, frame) in frames.iter().enumerate() {
		assert_packet_aligned(&frame.payload);
		let mut head = None;
		for packet in frame.payload.chunks(188) {
			// adaptation-field-only packets (adaptation_field_control == 0b10) are the clock.
			if packet[3] & 0x30 != 0x20 {
				break;
			}
			assert_eq!(packet[5], 0x10, "PCR_flag alone, at {i}");
			// The six reserved bits between base and extension are ones (ISO 13818-1);
			// the crate's serializer writes zeros here, which is why the packet is
			// laid out by hand.
			assert_eq!(packet[10] & 0x7e, 0x7e, "reserved bits must be ones, at {i}");
			let base = (u64::from(packet[6]) << 25)
				| (u64::from(packet[7]) << 17)
				| (u64::from(packet[8]) << 9)
				| (u64::from(packet[9]) << 1)
				| u64::from(packet[10] >> 7);
			pcrs.push(base);
			head = Some(base);
		}
		// The frame is paced at the slot its last leading clock asserts (plus the
		// reserve), which is the newest slot that has begun by then.
		let Some(base) = head else { continue };
		let slot_ticks = frame.timestamp.as_micros() / 25_000 * 25_000 * 90 / 1_000;
		assert_eq!(
			base,
			(slot_ticks as u64).wrapping_sub(16) & WIRE,
			"pacing off value, at {i}"
		);
	}

	const WIRE: u64 = (1 << 33) - 1;
	assert!(pcrs.len() >= 2, "expected at least two grid slots, got {pcrs:?}");
	// Slot 0 minus the 16-tick default reserve, mod 2^33.
	assert_eq!(pcrs[0], WIRE - 15, "slot 0 backs off through the wrap: {pcrs:?}");
	// Every step is exactly one 25 ms slot (2250 ticks) in the circular clock.
	for (i, w) in pcrs.windows(2).enumerate() {
		assert_eq!(w[1].wrapping_sub(w[0]) & WIRE, 2250, "step off the grid at {i}: {w:?}");
	}
}

/// The clock backs off by the largest reserve of any track, not just the PCR
/// track's: a second rendition with a deeper reorder (catalog `jitter`) authors
/// its DTS further behind the PTS, and a clock respecting only the PCR track's
/// reserve would run ahead of those frames' decode times.
#[tokio::test(start_paused = true)]
async fn export_pcr_respects_every_renditions_reserve() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let mut make = |name: &str, jitter: Option<Duration>| {
		let track = broadcast
			.create_track(name, hang::container::track_info(hang::catalog::PRIORITY.video))
			.unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc.clone());
		cfg.jitter = jitter;
		catalog.modify().unwrap().video.renditions.insert(name.to_string(), cfg);
		Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data))
	};
	// "a" gets the lowest PID and so carries the PCR, with the tiny default
	// reserve; "b" declares a 100 ms reorder depth.
	let mut a = make("a.avc1", None);
	let mut b = make("b.avc1", Some(Duration::from_millis(100)));

	let idr = [0x65u8; 32];
	for i in 0..25u64 {
		let timestamp = Timestamp::from_millis(10_000 + i * 40).unwrap();
		for video in [&mut a, &mut b] {
			video
				.write(Frame {
					timestamp,
					duration: None,
					payload: length_prefixed(&[&idr]),
					keyframe: true,
				})
				.unwrap();
		}
	}
	a.finish().unwrap();
	b.finish().unwrap();

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// In transport order, every PES unit (either rendition) decodes at or after
	// the last PCR preceding it.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut last_pcr = None;
	let mut units = 0usize;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(pcr) = packet.adaptation_field.as_ref().and_then(|af| af.pcr) {
			last_pcr = Some(pcr.as_u64() / 300);
		}
		if let Some(TsPayload::PesStart(pes)) = packet.payload {
			units += 1;
			let pts = pes.header.pts.expect("PES carried no PTS").as_u64();
			let decode = pes.header.dts.map(|t| t.as_u64()).unwrap_or(pts);
			if let Some(pcr) = last_pcr {
				assert!(decode >= pcr, "unit decodes at {decode}, before the clock at {pcr}");
			}
		}
	}
	assert!(units >= 50, "expected both renditions' units, got {units}");
}

/// A frame cadence coarser than the grid backfills every missed slot: low-rate
/// video-only content (here 2.5 fps, 16 slots per frame) still asserts a uniform
/// 25 ms ramp. A tight backfill cap would skip slots on every frame and re-create
/// the clock jumps this grid exists to eliminate.
#[tokio::test(start_paused = true)]
async fn export_pcr_backfills_a_coarse_cadence() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		catalog
			.modify()
			.unwrap()
			.video
			.renditions
			.insert(track.name().to_string(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	let idr = [0x65u8; 32];
	for i in 0..10u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_millis(10_000 + i * 400).unwrap(),
				duration: None,
				payload: length_prefixed(&[&idr]),
				keyframe: true,
			})
			.unwrap();
	}
	producer.finish().unwrap();

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut pcrs: Vec<u64> = Vec::new();
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(pcr) = packet.adaptation_field.as_ref().and_then(|af| af.pcr) {
			pcrs.push(pcr.as_u64() / 300);
		}
	}

	// 9 inter-frame spans of 400 ms, 16 slots each: every one backfilled.
	assert!(
		pcrs.len() > 100,
		"expected a dense backfilled clock, got {}",
		pcrs.len()
	);
	for (i, w) in pcrs.windows(2).enumerate() {
		assert_eq!(w[1] - w[0], 2250, "PCR interval off the grid at {i}: {w:?}");
	}
}

/// Full SCTE-35 round-trip: import `bbb.ts` (real H.264 + AAC) into a broadcast
/// that also carries a `.scte35` cue track, export to TS, re-import, and assert
/// the splice_info_section came back byte-for-byte. The PMT must advertise the
/// SCTE-35 stream (0x86) and the program-level CUEI registration descriptor.
#[tokio::test(start_paused = true)]
async fn export_scte35_roundtrip() {
	let data = include_bytes!("test_data/bbb.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	// Create and write the SCTE-35 cue track BEFORE moving `broadcast` into
	// `Import` (which consumes it); the producer stays alive so the exporter can
	// subscribe to the retained track.
	let scte = broadcast
		.unique_track(".scte35", hang::container::track_info(hang::catalog::PRIORITY.video))
		.unwrap();
	let scte_name = scte.name().to_string();
	{
		let track = tscat::Track {
			pid: 0x102,
			descriptors: Vec::new(),
			verbatim: Some(tscat::Verbatim::new(0x86, tscat::Framing::Section)),
		};
		catalog
			.modify()
			.unwrap()
			.ext
			.mpegts
			.tracks
			.insert(scte_name.clone(), track);
	}
	let mut scte_producer = Producer::new(scte, HangContainer::Legacy(crate::container::Kind::Data));
	// bbb's first video keyframe is at 1.4 s; stamp the cue just after it so it survives
	// the tune-in alignment (a cue before the first keyframe is dropped with the lead).
	scte_producer
		.write(Frame {
			timestamp: Timestamp::from_millis(1410).unwrap(),
			duration: None,
			payload: Bytes::from_static(CUE),
			keyframe: true,
		})
		.unwrap();
	scte_producer.cut(None).unwrap();
	scte_producer.finish().unwrap();

	// Now add the real video/audio by importing bbb.ts (this moves `broadcast`).
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	// `import`, `catalog`, and `scte_producer` stay alive: retained tracks. The
	// exporter must carry the extension to see the mpegts section.
	let ts = drain_with(
		Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
			.await
			.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);

	// The first PMT advertises the SCTE-35 ES (0x86) and the CUEI descriptor.
	// Stop at it: the raw reader would choke on the SCTE section packets that
	// follow (the very reason the importer intercepts them).
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut saw_scte_es = false;
	let mut saw_cuei = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			saw_scte_es = pmt
				.es_info
				.iter()
				.any(|e| e.stream_type == StreamType::Dts8ChannelLosslessAudio);
			saw_cuei = pmt
				.program_info
				.iter()
				.any(|d| d.tag == 0x05 && d.data.len() >= 4 && &d.data[0..4] == b"CUEI");
			break;
		}
	}
	assert!(saw_scte_es, "PMT missing the SCTE-35 elementary stream (0x86)");
	assert!(saw_cuei, "PMT missing the program-level CUEI registration descriptor");

	// Re-import the exported TS and read the .scte35 frame back.
	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(
		&mut broadcast2,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let snapshot = catalog2.snapshot();
	let verbatim = snapshot
		.ext
		.mpegts
		.tracks
		.values()
		.filter(|t| t.verbatim.is_some())
		.count();
	assert_eq!(verbatim, 1, "round-trip lost the SCTE-35 track");
	let name = scte_track(&snapshot).expect("a scte35 track");

	let track = consumer2
		.track(&name)
		.unwrap()
		.subscribe(moq_net::track::Subscription::default().with_max_age(RECORDING_MAX_AGE))
		.await
		.unwrap();
	let mut scte_reader = crate::container::Consumer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let frame = scte_reader
		.read()
		.await
		.unwrap()
		.expect("no SCTE-35 frame after round-trip");
	assert_eq!(
		frame.payload.as_ref(),
		CUE,
		"SCTE-35 section did not round-trip byte-for-byte"
	);
}

/// PES-framed verbatim round-trip: import `bbb.ts` (real H.264 + AAC, whose video
/// supplies the media clock the exporter needs) alongside a private PES-framed
/// stream (stream_type 0x06) carried verbatim, export to TS, then re-import and
/// assert the PID, framing, stream_id, and payload all survive. Exercises the
/// exporter's PES re-emit path; `private_pes_carried_verbatim` only covers import.
#[tokio::test(start_paused = true)]
async fn export_pes_verbatim_roundtrip() {
	const DATA_PID: u16 = 0x104;
	const STREAM_ID: u8 = 0xc0;
	const PAYLOAD: &[u8] = &[0xde, 0xad, 0xbe, 0xef, 0x01, 0x02];

	let data = include_bytes!("test_data/bbb.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	// Build the verbatim PES track BEFORE moving `broadcast` into `Import`; the
	// producer stays alive so the exporter can subscribe to the retained track.
	let data_track = broadcast
		.unique_track(".data", hang::container::track_info(hang::catalog::PRIORITY.video))
		.unwrap();
	let data_name = data_track.name().to_string();
	{
		let mut verbatim = tscat::Verbatim::new(0x06, tscat::Framing::Pes);
		verbatim.stream_id = Some(STREAM_ID);
		let mut track = tscat::Track::new(DATA_PID);
		track.verbatim = Some(verbatim);
		catalog
			.modify()
			.unwrap()
			.ext
			.mpegts
			.tracks
			.insert(data_name.clone(), track);
	}
	let mut data_producer = Producer::new(data_track, HangContainer::Legacy(crate::container::Kind::Data));
	// bbb's first video keyframe is at 1.4 s; stamp the PES just after it so it survives
	// the tune-in alignment (content before the first keyframe is dropped with the lead).
	data_producer
		.write(Frame {
			timestamp: Timestamp::from_millis(1410).unwrap(),
			duration: None,
			payload: Bytes::from_static(PAYLOAD),
			keyframe: true,
		})
		.unwrap();
	data_producer.cut(None).unwrap();
	data_producer.finish().unwrap();

	// Real video/audio supplies the media clock (moves `broadcast`).
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	// `import`, `catalog`, and `data_producer` stay alive: retained tracks.
	let ts = drain_with(
		Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
			.await
			.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);

	// Re-import the exported TS and recover the verbatim PES stream.
	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(
		&mut broadcast2,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let snapshot = catalog2.snapshot();
	let (name, track) = snapshot
		.ext
		.mpegts
		.tracks
		.iter()
		.find(|(_, t)| t.verbatim.as_ref().is_some_and(|v| v.stream_type == 0x06))
		.expect("verbatim PES survived the round-trip");
	assert_eq!(track.pid, DATA_PID, "PES PID preserved");
	let verbatim = track.verbatim.as_ref().unwrap();
	assert_eq!(verbatim.framing, tscat::Framing::Pes, "PES framing preserved");
	assert_eq!(verbatim.stream_id, Some(STREAM_ID), "PES stream_id preserved");
	let name = name.clone();

	let track = consumer2
		.track(&name)
		.unwrap()
		.subscribe(moq_net::track::Subscription::default().with_max_age(RECORDING_MAX_AGE))
		.await
		.unwrap();
	let mut reader = crate::container::Consumer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let frame = reader
		.read()
		.await
		.unwrap()
		.expect("no verbatim PES frame after round-trip");
	assert_eq!(
		frame.payload.as_ref(),
		PAYLOAD,
		"verbatim PES payload round-trips byte-for-byte"
	);
}

// SCTE-35 cues are clocked on video, so the exporter rejects a cue program with no video
// track rather than emitting cues pinned to zero.
#[tokio::test(start_paused = true)]
async fn scte35_without_video_export_is_rejected() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	// A SCTE-35 cue track and nothing else.
	let scte = broadcast
		.unique_track(".scte35", hang::container::track_info(hang::catalog::PRIORITY.video))
		.unwrap();
	let scte_name = scte.name().to_string();
	{
		let track = tscat::Track {
			pid: 0x102,
			descriptors: Vec::new(),
			verbatim: Some(tscat::Verbatim::new(0x86, tscat::Framing::Section)),
		};
		catalog.modify().unwrap().ext.mpegts.tracks.insert(scte_name, track);
	}
	let mut producer = Producer::new(scte, HangContainer::Legacy(crate::container::Kind::Data));
	producer
		.write(Frame {
			timestamp: Timestamp::from_millis(0).unwrap(),
			duration: None,
			payload: Bytes::from_static(CUE),
			keyframe: true,
		})
		.unwrap();
	producer.cut(None).unwrap();
	producer.finish().unwrap();

	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap();
	let err = loop {
		match tokio::time::timeout(Duration::from_secs(1), exporter.next()).await {
			Ok(Ok(Some(_))) => continue,
			Ok(Ok(None)) => panic!("export completed; a cue program without video must be rejected"),
			Ok(Err(e)) => break e,
			Err(_) => panic!("export neither errored nor completed"),
		}
	};
	assert!(
		err.to_string().contains("requires a video track"),
		"expected a video-required rejection, got: {err}"
	);
}

/// Subscribe to a track and read every retained frame payload it holds.
async fn read_frames(consumer: &moq_net::broadcast::Consumer, name: &str, kind: Kind) -> Vec<Vec<u8>> {
	let track = consumer
		.track(name)
		.unwrap()
		.subscribe(moq_net::track::Subscription::default().with_max_age(RECORDING_MAX_AGE))
		.await
		.unwrap();
	let mut reader = crate::container::Consumer::new(track, HangContainer::Legacy(kind));
	let mut frames = Vec::new();
	while let Ok(res) = tokio::time::timeout(Duration::from_millis(50), reader.read()).await {
		let Some(frame) = res.unwrap() else { break };
		frames.push(frame.payload.to_vec());
	}
	frames
}

/// Both real Kyrion MP2 programs must survive TS -> MoQ -> TS byte-for-byte, and
/// the PMT must re-announce them as MPEG-1 audio (0x03): the capture is 48 kHz,
/// so the half-rate type (0x04) would be unfaithful. This capture is a dirty start
/// (begins mid-GOP), so the export's keyframe alignment drops the MP2 ahead of the
/// first video keyframe; what remains is a byte-exact suffix of each program.
#[tokio::test(start_paused = true)]
async fn mp2_kyrion_roundtrip_byte_exact() {
	let data = include_bytes!("test_data/scte35/kyrion_dirtystart.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let names: Vec<String> = catalog.snapshot().audio.renditions.keys().cloned().collect();
	assert_eq!(names.len(), 2, "both Kyrion MP2 programs");
	let mut ingested = Vec::new();
	for name in &names {
		let frames = read_frames(&consumer, name, Kind::Audio).await;
		assert!(!frames.is_empty(), "{name}: no MP2 frames");
		assert!(
			frames.iter().all(|f| f[0] == 0xFF && f[1] & 0xE0 == 0xE0),
			"{name}: whole-frame carriage starts at the Layer II sync word"
		);
		ingested.push(frames);
	}

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			let mp2 = pmt
				.es_info
				.iter()
				.filter(|e| e.stream_type == StreamType::Mpeg1Audio)
				.count();
			assert_eq!(mp2, 2, "PMT must re-announce both MP2 streams as 0x03");
			break;
		}
	}

	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(&mut broadcast2, crate::catalog::Config::default()).unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let names2: Vec<String> = catalog2.snapshot().audio.renditions.keys().cloned().collect();
	assert_eq!(names2.len(), 2, "round-trip lost an MP2 track");
	let mut roundtripped = Vec::new();
	for name in &names2 {
		roundtripped.push(read_frames(&consumer2, name, Kind::Audio).await);
	}

	// Keyframe alignment drops the MP2 ahead of the first video keyframe (the dirty-start
	// lead), so each program's surviving frames are a byte-exact suffix of what was
	// ingested. Track discovery order is not stable across imports, so match by content.
	for rt in &roundtripped {
		assert!(!rt.is_empty(), "a program lost all of its MP2 frames");
		assert!(
			ingested.iter().any(|ing| ing.ends_with(rt)),
			"round-tripped MP2 must be a byte-exact suffix of an ingested program"
		);
	}
}

/// The ffmpeg AC-3 fixture must survive TS -> MoQ -> TS byte-for-byte in an
/// audio-only program: the PCR falls to the audio track and the PMT re-announces
/// ATSC 0x81 with the 'AC-3' registration descriptor.
#[tokio::test(start_paused = true)]
async fn ac3_roundtrip_byte_exact() {
	let data = include_bytes!("test_data/ac3.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let name = catalog
		.snapshot()
		.audio
		.renditions
		.keys()
		.next()
		.expect("an AC-3 track")
		.clone();
	let ingested = read_frames(&consumer, &name, Kind::Audio).await;
	assert!(!ingested.is_empty(), "no AC-3 frames");
	assert!(
		ingested.iter().all(|f| f[0] == 0x0B && f[1] == 0x77),
		"whole-frame carriage starts at the AC-3 sync word"
	);

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut checked_pmt = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			assert_eq!(pmt.es_info.len(), 1);
			assert_eq!(pmt.es_info[0].stream_type, StreamType::DolbyDigitalUpToSixChannelAudio);
			assert!(
				pmt.es_info[0]
					.descriptors
					.iter()
					.any(|d| d.tag == 0x05 && d.data.as_slice() == b"AC-3"),
				"PMT missing the ES-level 'AC-3' registration descriptor"
			);
			checked_pmt = true;
			break;
		}
	}
	assert!(checked_pmt, "missing PMT");

	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(&mut broadcast2, crate::catalog::Config::default()).unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let name2 = catalog2
		.snapshot()
		.audio
		.renditions
		.keys()
		.next()
		.expect("round-trip lost the AC-3 track")
		.clone();
	let roundtripped = read_frames(&consumer2, &name2, Kind::Audio).await;
	assert_eq!(roundtripped, ingested, "AC-3 frames must survive byte-for-byte");
}

/// The first ADTS frame of the first AAC PES: its header and raw data block.
fn first_adts_frame(ts: &[u8]) -> (super::adts::Header, Vec<u8>) {
	let mut pes = PesPacketReader::new(TsPacketReader::new(Cursor::new(ts)));
	let packet = pes.read_pes_packet().unwrap().expect("an AAC PES");
	let header = super::adts::Header::parse(&packet.data).unwrap();
	(header, packet.data[header.header_len..header.frame_len].to_vec())
}

/// ffmpeg's quad AAC fixture has no channelConfiguration, so its layout rides in a program
/// config element. Import moves it into the description and export puts it back: channel_config
/// 0 in ADTS and the element leading the first raw data block, exactly as ffmpeg wrote it.
#[tokio::test(start_paused = true)]
async fn aac_program_config_roundtrip() {
	let data = include_bytes!("test_data/aac_quad.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let snapshot = catalog.snapshot();
	let (name, audio) = snapshot.audio.renditions.iter().next().expect("an AAC track");
	assert_eq!(audio.channel_count, 4);
	let ingested = read_frames(&consumer, name, Kind::Audio).await;
	assert!(!ingested.is_empty(), "no AAC frames");

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let (header, block) = first_adts_frame(&ts);
	assert_eq!(header.channel_config, 0, "the layout is not a channelConfiguration");
	assert_eq!(
		block,
		first_adts_frame(data).1,
		"the first raw data block, element and all"
	);

	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(&mut broadcast2, crate::catalog::Config::default()).unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let snapshot2 = catalog2.snapshot();
	let (name2, audio2) = snapshot2
		.audio
		.renditions
		.iter()
		.next()
		.expect("round-trip lost the AAC track");
	assert_eq!(audio2.channel_count, 4);
	assert_eq!(audio2.description, audio.description);
	let roundtripped = read_frames(&consumer2, name2, Kind::Audio).await;
	assert_eq!(roundtripped, ingested, "the element is written once, not per frame");
}

/// GStreamer 1.28 `fdkaacenc` output, 48 kHz stereo, remuxed to FLV by ffmpeg 9.0.1. Its
/// AudioSpecificConfig signals SBR (and PS for v2) explicitly: object type 5 or 29 over an LC core
/// at 24 kHz, stereo for v1 and mono for v2. ADTS has two bits for the object type, so export
/// labels the LC core at its own rate and layout and leaves SBR and PS to implicit signaling, the
/// header ffmpeg's ADTS muxer writes too. ffprobe reads the result back as HE-AAC or HE-AACv2 at
/// 48 kHz stereo, like the source.
#[tokio::test(start_paused = true)]
async fn aac_explicit_sbr_exports_its_lc_core() {
	let fixtures: [(&[u8], u8); 2] = [
		(include_bytes!("test_data/he_aac.flv"), 2),
		(include_bytes!("test_data/he_aac_v2.flv"), 1),
	];
	for (data, channel_config) in fixtures {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let consumer = broadcast.consume();
		let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
		let mut import = crate::container::flv::Import::new(broadcast, catalog.reserve());
		import.decode(data).unwrap();
		import.finish().unwrap();

		let snapshot = catalog.snapshot();
		let (name, _) = snapshot.audio.renditions.iter().next().expect("an AAC track");
		let ingested = read_frames(&consumer, name, Kind::Audio).await;
		assert!(!ingested.is_empty(), "no AAC frames");

		let ts = drain(consumer).await;
		assert_packet_aligned(&ts);

		let (header, block) = first_adts_frame(&ts);
		assert_eq!(header.object_type, 2, "the LC core, not a masked SBR or PS");
		assert_eq!(header.sample_rate, 24_000, "the core rate, not the output rate");
		assert_eq!(header.channel_config, channel_config);
		assert_eq!(block, ingested[0], "the raw data block is untouched");
	}
}

/// Without a description, the catalog is all ADTS has to label a track with. A channel count no
/// channelConfiguration names is refused rather than written as stereo, and HE-AAC, whose core
/// rate and layout only a description names, is refused rather than masked to AAC Main.
#[tokio::test(start_paused = true)]
async fn aac_export_refuses_what_adts_cannot_label() {
	for (profile, channel_count, refusal) in [(2, 7, "7 channels"), (5, 2, "audioObjectType 5")] {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let consumer = broadcast.consume();
		let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

		let track = broadcast
			.create_track(
				broadcast.unique_name(".aac"),
				hang::container::track_info(hang::catalog::PRIORITY.audio),
			)
			.unwrap();
		let mut cfg = AudioConfig::new(AAC { profile }, 48_000, channel_count);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(track.name().to_string(), cfg);

		let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
		producer
			.write(Frame {
				timestamp: Timestamp::from_millis(0).unwrap(),
				duration: None,
				payload: Bytes::from_static(&[0x01, 0x02]),
				keyframe: true,
			})
			.unwrap();
		producer.finish().unwrap();

		let mut exporter = Export::new(crate::source::announced(&consumer)).await.unwrap();
		let err = loop {
			match tokio::time::timeout(Duration::from_secs(1), exporter.next()).await {
				Ok(Ok(Some(_))) => continue,
				Ok(Ok(None)) => panic!("export completed; expected a refusal naming {refusal}"),
				Ok(Err(e)) => break e,
				Err(_) => panic!("export neither errored nor completed"),
			}
		};
		assert!(err.to_string().contains(refusal), "expected {refusal}, got: {err}");
	}
}

/// The ffmpeg E-AC-3 fixture must survive TS -> MoQ -> TS byte-for-byte in an
/// audio-only program; the PMT re-announces ATSC 0x87 with the 'EAC3'
/// registration descriptor.
#[tokio::test(start_paused = true)]
async fn eac3_roundtrip_byte_exact() {
	let data = include_bytes!("test_data/eac3.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let name = catalog
		.snapshot()
		.audio
		.renditions
		.keys()
		.next()
		.expect("an E-AC-3 track")
		.clone();
	let ingested = read_frames(&consumer, &name, Kind::Audio).await;
	assert!(!ingested.is_empty(), "no E-AC-3 frames");
	assert!(
		ingested.iter().all(|f| f[0] == 0x0B && f[1] == 0x77),
		"whole-frame carriage starts at the E-AC-3 sync word"
	);

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut checked_pmt = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			assert_eq!(pmt.es_info.len(), 1);
			assert_eq!(
				pmt.es_info[0].stream_type,
				StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc
			);
			assert!(
				pmt.es_info[0]
					.descriptors
					.iter()
					.any(|d| d.tag == 0x05 && d.data.as_slice() == b"EAC3"),
				"PMT missing the ES-level 'EAC3' registration descriptor"
			);
			checked_pmt = true;
			break;
		}
	}
	assert!(checked_pmt, "missing PMT");

	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(&mut broadcast2, crate::catalog::Config::default()).unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let name2 = catalog2
		.snapshot()
		.audio
		.renditions
		.keys()
		.next()
		.expect("round-trip lost the E-AC-3 track")
		.clone();
	let roundtripped = read_frames(&consumer2, &name2, Kind::Audio).await;
	assert_eq!(roundtripped, ingested, "E-AC-3 frames must survive byte-for-byte");
}

/// Read every audio rendition's retained frames, keyed by codec string.
async fn read_audio_by_codec(
	consumer: &moq_net::broadcast::Consumer,
	catalog: &crate::catalog::Producer,
) -> std::collections::BTreeMap<String, Vec<Vec<u8>>> {
	let mut out = std::collections::BTreeMap::new();
	for (name, config) in &catalog.snapshot().audio.renditions {
		out.insert(config.codec.to_string(), read_frames(consumer, name, Kind::Audio).await);
	}
	out
}

/// The ATSC-compliance Kyrion capture (MPEG-2 video + AC-3 + MP2) must round-trip
/// both real audio streams byte-for-byte. The video is clock-only, so the
/// re-exported program is audio-only with the PCR on an audio PID, and the PMT
/// re-announces 0x81 (with the 'AC-3' registration descriptor, which the Kyrion
/// itself also emits) and 0x03.
#[tokio::test(start_paused = true)]
async fn kyrion_ac3_mp2_roundtrip_byte_exact() {
	let data = include_bytes!("test_data/kyrion_mpeg2av_ac3.ts");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let ingested = read_audio_by_codec(&consumer, &catalog).await;
	assert_eq!(
		ingested.keys().cloned().collect::<Vec<_>>(),
		["ac-3", "mp2"],
		"both real audio codecs cataloged"
	);
	assert!(
		ingested["ac-3"].iter().all(|f| f[0] == 0x0B && f[1] == 0x77),
		"AC-3 whole-frame carriage"
	);
	assert!(
		ingested["mp2"].iter().all(|f| f[0] == 0xFF && f[1] & 0xE0 == 0xE0),
		"MP2 whole-frame carriage"
	);

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			assert_eq!(pmt.es_info.len(), 2, "audio-only program: AC-3 + MP2");
			let ac3 = pmt
				.es_info
				.iter()
				.find(|e| e.stream_type == StreamType::DolbyDigitalUpToSixChannelAudio)
				.expect("AC-3 ES re-announced as 0x81");
			assert!(
				ac3.descriptors
					.iter()
					.any(|d| d.tag == 0x05 && d.data.as_slice() == b"AC-3"),
				"AC-3 registration descriptor"
			);
			assert!(
				pmt.es_info.iter().any(|e| e.stream_type == StreamType::Mpeg1Audio),
				"MP2 re-announced as 0x03 (48 kHz is an MPEG-1 rate)"
			);
			break;
		}
	}

	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(&mut broadcast2, crate::catalog::Config::default()).unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let roundtripped = read_audio_by_codec(&consumer2, &catalog2).await;
	assert_eq!(roundtripped, ingested, "both audio streams survive byte-for-byte");
}

/// Find the SCTE-35 verbatim stream (stream_type 0x86) in a catalog snapshot. A
/// clip may carry other undecoded streams verbatim, so select by type, not order.
fn scte_track(snap: &crate::catalog::hang::Catalog<tscat::Ext>) -> Option<String> {
	snap.ext
		.mpegts
		.tracks
		.iter()
		.find(|(_, t)| t.verbatim.as_ref().is_some_and(|v| v.stream_type == 0x86))
		.map(|(name, _)| name.clone())
}

/// Subscribe to a cue track and read every retained `splice_info_section` it holds.
async fn read_cues(consumer: &moq_net::broadcast::Consumer, name: &str) -> Vec<(Vec<u8>, Timestamp)> {
	let track = consumer
		.track(name)
		.unwrap()
		.subscribe(moq_net::track::Subscription::default().with_max_age(RECORDING_MAX_AGE))
		.await
		.unwrap();
	let mut reader = crate::container::Consumer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut cues = Vec::new();
	while let Ok(res) = tokio::time::timeout(Duration::from_millis(50), reader.read()).await {
		let Some(frame) = res.unwrap() else { break };
		cues.push((frame.payload.to_vec(), frame.timestamp));
	}
	cues
}

/// Full TS -> MoQ -> TS over fixtures carrying SCTE-35 cues; each section must survive the seam
/// byte-for-byte. Most are real-video clips with injected cues (regenerate via the `scte35_inject`
/// example); tsduck.ts is TSDuck-authored and kyrion_dirtystart.ts is a real-encoder capture. Add
/// a source by dropping a `.ts` in `test_data/scte35/` and listing it here.
///
/// The cues are independently valid SCTE-35: TSDuck (the authoritative toolkit) decodes every
/// section in every fixture with CRC32 OK. That decode is checked in next to each clip as
/// `<fixture>_tsduck.txt`; regenerate it via the `moq-tsduck` image (cue PID 0x21 for the injected
/// fixtures, 0x14d for the Kyrion capture):
/// `tsp -I file test_data/scte35/<fixture>.ts -P tables --pid <pid> -O drop > <fixture>_tsduck.txt`.
#[tokio::test(start_paused = true)]
async fn scte35_fixtures_survive_roundtrip() {
	// The corpus proves byte-exact survival across sources that each cover an axis no other does;
	// cue counts vary per fixture (five for the injected clips, ten on the wire for tsduck, six for
	// the Kyrion capture). For every cue we assert survival, a known splice_command_type, and that
	// the per-fixture distinct count holds (so a clip that lost variety to duplicates fails). Only
	// tsduck, whose cues we author, pins the exact command-type set.
	// (source, total cues, distinct cues, expected command-type set or empty, fixture bytes.)
	type Fixture = (&'static str, usize, usize, &'static [u8], &'static [u8]);
	let fixtures: &[Fixture] = &[
		// ffmpeg mpegts muxer, H.264 320x240 progressive, no audio: the baseline.
		("ffmpeg", 5, 5, &[], include_bytes!("test_data/scte35/ffmpeg.ts")),
		// GStreamer mpegtsmux, H.264 720x480 interlaced (480i) + AAC: a second muxer, SD
		// interlaced framing, and an audio track.
		("gst480i", 5, 5, &[], include_bytes!("test_data/scte35/gst480i.ts")),
		// Real BigBuckBunny frames, H.265 320x240 + Opus: a second video codec, real content,
		// and the WebCodec-friendly Opus path.
		("bbb5s", 5, 5, &[], include_bytes!("test_data/scte35/bbb5s.ts")),
		// TSDuck-authored: splice_null, splice_insert, time_signal, and a private_command (custom),
		// each re-sent with an advancing CC so the importer emits 5 distinct x2 = 10. The only
		// fixture covering section repetition, distinct from the byte-identical same-CC transport
		// duplicate the reassembler drops.
		(
			"tsduck",
			10,
			5,
			&[0x00, 0x05, 0x06, 0xff],
			include_bytes!("test_data/scte35/tsduck.ts"),
		),
		// Real Ateme Kyrion broadcast (H.264 1080i + MP2), captured mid-stream: a real
		// encoder's cues surviving the full round-trip, not a synthetic clip. Cues are external,
		// so the command-type set stays unpinned.
		(
			"kyrion_dirtystart",
			6,
			6,
			&[],
			include_bytes!("test_data/scte35/kyrion_dirtystart.ts"),
		),
	];

	// SCTE-35 splice_command_type lives at byte 13 of the splice_info_section.
	const KNOWN_SPLICE_COMMANDS: [u8; 6] = [0x00, 0x04, 0x05, 0x06, 0x07, 0xff];

	for (source, total, distinct, command_types, data) in fixtures {
		// Ingest the fixture.
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let consumer = broadcast.consume();
		let catalog = crate::catalog::Producer::new(
			&mut broadcast,
			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
		)
		.unwrap();
		let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
		import.decode(&BytesMut::from(&data[..])).unwrap();
		import.finish().unwrap();

		let snap = catalog.snapshot();
		assert!(!snap.video.renditions.is_empty(), "{source}: video track from the clip");
		// Select the SCTE-35 stream by stream_type (0x86); a clip may also carry other
		// undecoded streams verbatim (e.g. Opus as private PES in bbb5s).
		let name = scte_track(&snap).expect("a scte35 track");
		let ingested = read_cues(&consumer, &name).await;
		assert_eq!(ingested.len(), *total, "{source}: {total} cues on ingest");
		assert!(
			ingested.iter().all(|(b, _)| b.first() == Some(&0xfc)),
			"{source}: every cue is a splice_info_section (table_id 0xFC)"
		);
		let unique: std::collections::HashSet<&Vec<u8>> = ingested.iter().map(|(b, _)| b).collect();
		assert_eq!(
			unique.len(),
			*distinct,
			"{source}: {distinct} distinct cue sections, not dups"
		);
		// Structural validity: every cue's splice_command_type is a known SCTE-35 command.
		assert!(
			ingested
				.iter()
				.all(|(b, _)| b.get(13).is_some_and(|t| KNOWN_SPLICE_COMMANDS.contains(t))),
			"{source}: every cue carries a known splice_command_type"
		);
		// For fixtures we author (tsduck), pin the exact set of command types present.
		if !command_types.is_empty() {
			let mut got: Vec<u8> = ingested.iter().filter_map(|(b, _)| b.get(13).copied()).collect();
			got.sort_unstable();
			got.dedup();
			assert_eq!(got.as_slice(), *command_types, "{source}: splice_command_type set");
		}
		assert!(
			ingested.iter().all(|(_, ts)| *ts != Timestamp::ZERO),
			"{source}: cues stamped with the video PTS, not zero"
		);

		// Export and re-ingest.
		let ts = drain_with(
			Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
				.await
				.unwrap(),
		)
		.await;
		assert_packet_aligned(&ts);

		let mut broadcast2 = moq_net::broadcast::Info::new().produce();
		let consumer2 = broadcast2.consume();
		let catalog2 = crate::catalog::Producer::new(
			&mut broadcast2,
			crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
		)
		.unwrap();
		let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
		import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
		import2.finish().unwrap();
		let name2 = scte_track(&catalog2.snapshot()).expect("a scte35 track");
		let roundtripped = read_cues(&consumer2, &name2).await;

		let before: Vec<&Vec<u8>> = ingested.iter().map(|(b, _)| b).collect();
		let after: Vec<&Vec<u8>> = roundtripped.iter().map(|(b, _)| b).collect();
		assert_eq!(
			after, before,
			"{source}: every section survived TS -> MoQ -> TS byte-for-byte"
		);
	}
}

/// Build a well-formed long-form section: the full generic header (extension,
/// current version, `number` of `last`), then `body` and a valid CRC-32/MPEG-2.
/// The SI store buffers a sub-table until its generation completes, so the header
/// fields must be coherent.
fn make_long_section(table_id: u8, ext: u16, version: u8, number: u8, last: u8, body: &[u8]) -> Vec<u8> {
	let section_length = 5 + body.len() + 4;
	let mut s = vec![
		table_id,
		0xb0 | ((section_length >> 8) as u8 & 0x0f),
		(section_length & 0xff) as u8,
		(ext >> 8) as u8,
		(ext & 0xff) as u8,
		0xc0 | (version << 1) | 0x01,
		number,
		last,
	];
	s.extend_from_slice(body);
	let crc = crc::Crc::<u32>::new(&crc::CRC_32_MPEG_2).checksum(&s);
	s.extend_from_slice(&crc.to_be_bytes());
	s
}

/// Read an SI snapshot track from the start: every group, as its frames' payloads.
///
/// Starting at group 0 only asks the publisher to send from there; the default
/// [`Duration::ZERO`] budget then skips every group the live edge has
/// already passed, which is all of them once the importer has finished. A budget
/// wider than the test's wall-clock span (every cut lands within microseconds) is
/// what actually delivers the history, so a count of groups counts cuts.
async fn read_si_groups(consumer: &moq_net::broadcast::Consumer, name: &str) -> Vec<Vec<Bytes>> {
	let mut track = consumer
		.track(name)
		.unwrap()
		.subscribe(
			moq_net::track::Subscription::default()
				.with_start(moq_net::track::Position::group(0))
				.with_max_age(moq_net::track::DEFAULT_MAX_AGE),
		)
		.await
		.unwrap();
	let mut groups = Vec::new();
	while let Some(mut group) = track.recv_group().await.unwrap() {
		let mut frames = Vec::new();
		while let Some(frame) = kio::wait(|waiter| group.poll_read_frame(waiter)).await.unwrap() {
			frames.push(frame.payload);
		}
		groups.push(frames);
	}
	groups
}

/// The newest snapshot of an SI track, reduced to its sections.
async fn read_si_sections(consumer: &moq_net::broadcast::Consumer, name: &str) -> Vec<Bytes> {
	let groups = read_si_groups(consumer, name).await;
	let frames = groups.last().expect("at least one snapshot group");
	frames
		.iter()
		.flat_map(crate::container::ts::si::split_sections)
		.collect()
}

/// Wrap a complete section in one PUSI TS packet on `pid` (pointer_field 0), padded to 188.
fn si_packet(pid: u16, section: &[u8]) -> Vec<u8> {
	si_packet_cc(pid, section, 0)
}

/// [`si_packet`] with an explicit continuity counter. A repetition must vary the
/// counter to reach the SI store at all: a byte-identical packet is dropped as a TS
/// duplicate before reassembly.
fn si_packet_cc(pid: u16, section: &[u8], cc: u8) -> Vec<u8> {
	let mut p = vec![
		0x47,
		0x40 | ((pid >> 8) as u8 & 0x1f),
		(pid & 0xff) as u8,
		0x10 | (cc & 0x0f),
		0x00,
	];
	p.extend_from_slice(section);
	assert!(p.len() <= 188, "section overflows one TS packet");
	p.resize(188, 0xff);
	p
}

/// Split a complete section across several 188-byte TS packets on `pid`: a PUSI packet
/// (pointer_field 0) carrying the head, then continuation packets (PUSI clear, continuity
/// counter advancing) for the rest, the last padded with 0xff stuffing. Unlike `si_packet`
/// this reaches the multi-packet reassembly path (PUSI + continuity across packets).
fn si_packets_multi(pid: u16, section: &[u8]) -> Vec<u8> {
	let mut out = Vec::new();
	let mut cc = 0u8;
	// First packet: PUSI set, pointer_field 0, then as much of the section as fits.
	let mut first = vec![
		0x47,
		0x40 | ((pid >> 8) as u8 & 0x1f),
		(pid & 0xff) as u8,
		0x10 | cc,
		0x00,
	];
	let head = section.len().min(188 - first.len());
	first.extend_from_slice(&section[..head]);
	first.resize(188, 0xff);
	out.extend_from_slice(&first);

	// Continuation packets: PUSI clear, continuity counter incremented per payload packet.
	let mut pos = head;
	while pos < section.len() {
		cc = (cc + 1) & 0x0f;
		let mut p = vec![0x47, (pid >> 8) as u8 & 0x1f, (pid & 0xff) as u8, 0x10 | cc];
		let take = (section.len() - pos).min(188 - p.len());
		p.extend_from_slice(&section[pos..pos + take]);
		p.resize(188, 0xff);
		out.extend_from_slice(&p);
		pos += take;
	}
	out
}

/// Decode an SDT Actual section's first service: `(service_type, provider, name)` from the
/// service_descriptor (tag 0x48). Enough to prove the service identity survived, no more.
fn parse_sdt_service(sec: &[u8]) -> (u8, String, String) {
	// header(8) + first service loop entry: service_id(2), flags(1), running/free + desc_len(2).
	let desc_loop_len = (((sec[11 + 3] & 0x0f) as usize) << 8) | sec[11 + 4] as usize;
	let mut d = 11 + 5;
	let end = d + desc_loop_len;
	while d < end {
		let (tag, len) = (sec[d], sec[d + 1] as usize);
		let body = &sec[d + 2..d + 2 + len];
		if tag == 0x48 {
			let service_type = body[0];
			let prov_len = body[1] as usize;
			let provider = String::from_utf8_lossy(&body[2..2 + prov_len]).into_owned();
			let name_len = body[2 + prov_len] as usize;
			let name = String::from_utf8_lossy(&body[3 + prov_len..3 + prov_len + name_len]).into_owned();
			return (service_type, provider, name);
		}
		d += 2 + len;
	}
	panic!("SDT service_descriptor (0x48) not found");
}

/// The DVB service layer (SDT + NIT + transport/service identity) must survive
/// TS -> MoQ -> TS. `bbb.ts` carries a real ffmpeg SDT (service "Service01" / provider
/// "FFmpeg"); no fixture carries a NIT, so a synthetic one is injected on PID 0x0010.
/// After the round-trip the SDT and NIT are byte-identical and the identity is preserved.
#[tokio::test(start_paused = true)]
async fn service_layer_survives_roundtrip() {
	let data = include_bytes!("test_data/bbb.ts");
	let nit = make_long_section(0x40, 0x1234, 1, 0, 0, &[0xff, 0x01]);

	// Prepend a synthetic NIT Actual packet (0x0010); prepend keeps bbb's alignment.
	// Twice, because real SI repeats every few seconds: the repetition must collapse
	// into the same single committed section rather than cutting a second group. The
	// repeat varies the continuity counter so it reaches the store (an identical
	// packet would be dropped as a TS duplicate before reassembly).
	let mut input = si_packet(0x0010, &nit);
	input.extend_from_slice(&si_packet_cc(0x0010, &nit, 1));
	input.extend_from_slice(&data[..]);

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&input[..])).unwrap();
	import.finish().unwrap();

	let snapshot = catalog.snapshot();
	let program = snapshot.ext.mpegts.program.clone().expect("a program record");
	assert_eq!(program.transport_stream_id, 1, "TSID captured from the PAT");
	assert_eq!(program.program_number, 1, "program number captured from the PAT");
	assert_eq!(program.pmt_pid, 0x1000, "original PMT PID captured from the PAT");
	let si = snapshot.ext.mpegts.si.clone();
	let entry = |pid: u16, table_id: u8| {
		si.get(&pid)
			.and_then(|tables| tables.get(&table_id))
			.expect("an SI entry for the (PID, table_id)")
	};

	let sdt_entry = entry(0x0011, 0x42);
	assert_eq!(
		sdt_entry.interval,
		Some(Duration::from_secs(2)),
		"the DVB SDT interval was filled in"
	);
	let sdt_sections = read_si_sections(&consumer, &sdt_entry.track).await;
	assert_eq!(sdt_sections.len(), 1, "bbb.ts carries one SDT section");
	let sdt = sdt_sections[0].clone();
	assert_eq!(sdt.first(), Some(&0x42), "SDT Actual (table_id 0x42)");
	let (service_type, provider, name) = parse_sdt_service(&sdt);
	assert_eq!(
		(service_type, provider.as_str(), name.as_str()),
		(0x01, "FFmpeg", "Service01")
	);

	let nit_entry = entry(0x0010, 0x40);
	assert_eq!(
		nit_entry.interval,
		Some(Duration::from_secs(10)),
		"the DVB NIT interval was filled in"
	);
	let nit_groups = read_si_groups(&consumer, &nit_entry.track).await;
	assert_eq!(
		nit_groups.len(),
		1,
		"the repeated NIT committed once; a repetition cuts no group"
	);
	assert_eq!(
		nit_groups[0],
		vec![Bytes::from(nit.clone())],
		"the NIT snapshot is the section, byte-for-byte"
	);

	// `import` and `catalog` stay alive: retained tracks the exporter subscribes to.
	let ts = drain_with(
		Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
			.await
			.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);

	// The rebuilt PAT preserves the transport/service identity and PMT PID.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut checked_pat = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pat(pat)) = packet.payload {
			assert_eq!(pat.transport_stream_id, 1, "TSID preserved in the rebuilt PAT");
			assert_eq!(pat.table.len(), 1);
			assert_eq!(pat.table[0].program_num, 1, "service number preserved");
			assert_eq!(pat.table[0].program_map_pid.as_u16(), 0x1000, "PMT PID preserved");
			checked_pat = true;
			break;
		}
	}
	assert!(checked_pat, "missing PAT");

	// Re-import: the SDT and NIT must come back byte-for-byte.
	let mut broadcast2 = moq_net::broadcast::Info::new().produce();
	let consumer2 = broadcast2.consume();
	let catalog2 = crate::catalog::Producer::new(
		&mut broadcast2,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve());
	import2.decode(&BytesMut::from(ts.as_ref())).unwrap();
	import2.finish().unwrap();

	let snapshot2 = catalog2.snapshot();
	let program2 = snapshot2
		.ext
		.mpegts
		.program
		.clone()
		.expect("a program record after round-trip");
	assert_eq!(
		program2.transport_stream_id, program.transport_stream_id,
		"TSID survived"
	);
	assert_eq!(
		program2.program_number, program.program_number,
		"program number survived"
	);
	assert_eq!(program2.pmt_pid, program.pmt_pid, "PMT PID survived");
	assert_eq!(snapshot2.ext.mpegts.si, si, "every SI entry survived the round-trip");
	assert_eq!(
		read_si_sections(&consumer2, &snapshot2.ext.mpegts.si[&0x0011][&0x42].track).await,
		vec![sdt],
		"the SDT survived byte-for-byte"
	);
	assert_eq!(
		read_si_sections(&consumer2, &snapshot2.ext.mpegts.si[&0x0010][&0x40].track).await,
		vec![Bytes::from(nit)],
		"the NIT survived byte-for-byte"
	);
}

/// Each SI PID must be re-emitted on its own interval, independently of the PSI cadence
/// and of video keyframes. The fixtures are a fraction of a second long, so nothing else
/// distinguishes a correct interval from "emitted once and never again": this builds a
/// 12-second synthetic timeline where the SDT (2s) and NIT (10s) land a different number
/// of times, and neither matches the 13 keyframes that drive the PSI.
#[tokio::test(start_paused = true)]
async fn si_pids_are_re_emitted_on_their_own_interval() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();

	// SI snapshot tracks, one group each: an SDT (2s) and a NIT (10s).
	let mut sdt_track = broadcast.create_track("0x0011-0x42.si", None).unwrap();
	sdt_track
		.write_frame(
			Timestamp::ZERO,
			Bytes::from(make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8])),
		)
		.unwrap();
	let mut nit_track = broadcast.create_track("0x0010-0x40.si", None).unwrap();
	nit_track
		.write_frame(
			Timestamp::ZERO,
			Bytes::from(make_long_section(0x40, 1, 0, 0, 0, &[0xbb; 8])),
		)
		.unwrap();

	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name.clone(), cfg);

		// SDT every 2s, NIT every 10s: the DVB maxima import fills in.
		guard.ext.mpegts.si.entry(0x0011).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "0x0011-0x42.si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
		guard.ext.mpegts.si.entry(0x0010).or_default().insert(
			0x40,
			tscat::SiEntry {
				track: "0x0010-0x40.si".to_string(),
				interval: Some(Duration::from_secs(10)),
				..Default::default()
			},
		);
	}

	// One keyframe per second across 12s, so the PSI fires on every one of the 13.
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	for sec in 0..=12u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(),
				duration: None,
				payload: length_prefixed(&[&idr]),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	}
	producer.finish().unwrap();

	let ts = drain_with(
		Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
			.await
			.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);

	let count = |pid: u16| {
		ts.as_chunks::<188>()
			.0
			.iter()
			.filter(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == pid)
			.count()
	};

	// Control: the PSI rides every output frame here (each is a keyframe, and they are
	// further apart than PSI_INTERVAL either way). The SI PIDs must not follow it.
	assert_eq!(count(0x0000), 13, "PAT on every frame");
	// SDT at 0,2,4,6,8,10,12s.
	assert_eq!(count(0x0011), 7, "SDT re-emitted on its 2s interval");
	// NIT at 0 and 10s.
	assert_eq!(count(0x0010), 2, "NIT re-emitted on its 10s interval");
}

/// The catalog form every published moq-cli through 0.11 writes carries the SI
/// sections inline under the PID, with no snapshot track. Export must carry them
/// the same way it carries a track's snapshot, on the PID's interval, or every one
/// of those publishers loses its service layer (and, before the inline form was
/// read at all, the whole export).
#[tokio::test(start_paused = true)]
async fn inline_si_form_is_re_emitted() {
	use base64::Engine;

	let broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();

	// Hand-written rather than produced, since nothing writes this form any more.
	let mut catalog = crate::catalog::hang::Catalog::<tscat::Ext>::default();
	let mut cfg = VideoConfig::new(H264 {
		profile: 0x64,
		constraints: 0,
		level: 0x1f,
		inline: false,
	});
	cfg.container = Container::Legacy;
	cfg.description = Some(avcc);
	catalog.video.renditions.insert(name.clone(), cfg);
	let mut json = serde_json::to_value(&catalog).unwrap();
	let sdt = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8]);
	let inline = base64::engine::general_purpose::STANDARD.encode(&sdt);
	json["mpegts"] = serde_json::json!({"si": {"17": {"interval": 2000, "sections": [inline]}}});
	// Held open: dropping the producer ends the track before export subscribes.
	let mut catalog_track = broadcast
		.create_track(hang::Catalog::DEFAULT_NAME, hang::Catalog::default_track_info())
		.unwrap();
	catalog_track
		.write_frame(Timestamp::ZERO, Bytes::from(serde_json::to_vec(&json).unwrap()))
		.unwrap();

	// One keyframe per second across 12s, as in the snapshot-track test above.
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	for sec in 0..=12u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(),
				duration: None,
				payload: length_prefixed(&[&idr]),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	}
	producer.finish().unwrap();

	let ts = drain_with(
		Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
			.await
			.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);

	let count = |pid: u16| {
		ts.as_chunks::<188>()
			.0
			.iter()
			.filter(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == pid)
			.count()
	};
	assert_eq!(count(0x0000), 13, "PAT on every frame");
	// SDT at 0,2,4,6,8,10,12s, byte-for-byte the inline section.
	assert_eq!(count(0x0011), 7, "inline SDT re-emitted on its 2s interval");
	let packet = ts
		.as_chunks::<188>()
		.0
		.iter()
		.find(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == 0x0011)
		.unwrap();
	// The section is stuffed to the packet's tail, byte-for-byte the inline one.
	assert_eq!(
		&packet[188 - sdt.len()..],
		&sdt[..],
		"the inline SDT rides its PID: {packet:02x?}"
	);
}

/// Count payload-bearing TS packets on `pid`, excluding its standalone clock packets.
fn count_pid(frames: &[Frame], pid: u16) -> usize {
	frames
		.iter()
		.flat_map(|f| f.payload.as_chunks::<188>().0.iter())
		.filter(|p| p[3] & 0x10 != 0 && ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == pid)
		.count()
}

/// Count the TS packets whose adaptation field sets `discontinuity_indicator`.
fn count_discontinuity(frames: &[Frame]) -> usize {
	frames
		.iter()
		.flat_map(|f| f.payload.as_chunks::<188>().0.iter())
		.filter(|p| p[3] & 0x20 != 0 && p[4] > 0 && p[5] & 0x80 != 0)
		.count()
}

/// An SI PID that equals the PMT PID would interleave tables with the program map.
#[tokio::test(start_paused = true)]
async fn export_rejects_si_pid_on_the_pmt() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		let mut guard = catalog.modify().unwrap();
		guard.audio.renditions.insert(name, cfg);
		guard.ext.mpegts.program = Some(tscat::Program {
			transport_stream_id: 1,
			program_number: 1,
			pmt_pid: 0x1000,
			..Default::default()
		});
		guard.ext.mpegts.si.entry(0x1000).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	}
	let _producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	let err = exporter
		.next()
		.await
		.expect_err("an SI PID on the PMT must fail the export");
	assert!(
		err.to_string().contains("mpegts.si PID"),
		"expected a PID collision, got {err}"
	);
}

/// An SI PID that equals an elementary-stream PID would interleave tables with media.
#[tokio::test(start_paused = true)]
async fn export_rejects_si_pid_on_an_elementary_stream() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		let mut guard = catalog.modify().unwrap();
		guard.audio.renditions.insert(name.clone(), cfg);
		guard.ext.mpegts.tracks.insert(name, tscat::Track::new(0x101));
		guard.ext.mpegts.si.entry(0x101).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	}
	let _producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	let err = exporter
		.next()
		.await
		.expect_err("an SI PID on an elementary stream must fail the export");
	assert!(
		err.to_string().contains("mpegts.si PID"),
		"expected a PID collision, got {err}"
	);
}

/// A `mpegts`-extension exporter over an announced broadcast.
async fn export_of(consumer: &moq_net::broadcast::Consumer) -> Export<tscat::Ext> {
	Export::with_ts(crate::source::announced(consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE)
}

fn publish_sdt(
	broadcast: &mut moq_net::broadcast::Producer,
	catalog: &mut crate::catalog::Producer<tscat::Ext>,
) -> moq_net::track::Producer {
	let mut track = broadcast.create_track("0x0011-0x42.si", None).unwrap();
	track
		.write_frame(
			Timestamp::ZERO,
			Bytes::from(make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8])),
		)
		.unwrap();
	catalog
		.modify()
		.unwrap()
		.ext
		.mpegts
		.si
		.entry(0x0011)
		.or_default()
		.insert(
			0x42,
			tscat::SiEntry {
				track: track.name().to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	track
}

/// A declared marker restarts the program clock and table cadence. An audio-only
/// program is the worst case: the PSI has no keyframe to recover at, and the PCR-only
/// packet is the only adaptation field it ever writes.
#[tokio::test(start_paused = true)]
async fn discontinuity_re_emits_tables_and_resumes_the_clock() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		guard.audio.renditions.insert(name, cfg);
	}
	let _sdt = publish_sdt(&mut broadcast, &mut catalog);

	// 100ms frames in one-second groups.
	let write = |producer: &mut Producer<HangContainer>, count: u64, offset: u64| {
		for i in 0..count {
			producer
				.write(Frame {
					timestamp: Timestamp::from_micros(offset + i * 100_000).unwrap(),
					duration: None,
					payload: Bytes::from_iter((0..180u16).map(|b| (b ^ i as u16) as u8)),
					keyframe: i % 10 == 0,
				})
				.unwrap();
			if i % 10 == 9 {
				producer.cut(None).unwrap();
			}
		}
	};

	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Audio));
	let mut export = export_of(&consumer).await;

	// A ten-minute offset exercises the long rewind from the controlled stimulus
	// campaign. Drain five seconds before restarting at zero.
	write(&mut producer, 50, 600_000_000);
	let before = drain_frames(&mut export).await;
	assert_eq!(export.discontinuity(), 0, "no rewind yet");
	assert_eq!(count_pid(&before, 0x0000), 10, "PAT on its 500ms cadence");
	assert_eq!(count_pid(&before, 0x0011), 3, "SDT at 0, 2 and 4s");

	// A forward marker must flush the last old frame instead of discarding it.
	producer.discontinuity().unwrap();
	write(&mut producer, 20, 605_000_000);
	producer.cut(None).unwrap();
	let marked = drain_frames(&mut export).await;
	let tail: Vec<_> = before
		.iter()
		.chain(&marked)
		.flat_map(|f| f.payload.iter().copied())
		.collect();
	let (_, audio_pts) = collect_pes_pts(&tail);
	// The carried tail may precede the refreshed PMT, so inspect all PES starts.
	let mut reader = TsPacketReader::new(Cursor::new(tail));
	let mut preserved = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::PesStart(pes)) = packet.payload {
			preserved |= pes.header.pts.unwrap().as_u64() == 604_900_000 * 90 / 1000;
		}
	}
	assert!(preserved, "forward marker discarded the pre-boundary tail");
	assert!(!audio_pts.is_empty());
	assert_eq!(export.discontinuity(), 1, "the marker was observed");

	let resume = marked.iter().position(|f| f.timestamp.as_micros() >= 605_000_000);
	let resume = resume.unwrap_or_else(|| {
		panic!(
			"resumed media missing, marked timestamps {:?}",
			marked.iter().map(|f| f.timestamp.as_micros()).collect::<Vec<_>>()
		)
	});
	assert_eq!(
		marked[resume].payload[3] & 0x30,
		0x20,
		"the new timeline leads with its clock"
	);
	let pcrs = collect_pcrs(&marked[resume..]);
	assert!(!pcrs.is_empty(), "clock resumed promptly");
	for pair in pcrs.windows(2) {
		assert_eq!(pair[1].1.wrapping_sub(pair[0].1) & ((1 << 33) - 1), 2250);
	}

	// Tables come back on cadence rather than waiting for the old 10-minute clock.
	assert!(
		count_pid(&marked[resume..], 0x0000) >= 1,
		"PAT re-emitted after the marker"
	);
	assert_eq!(count_discontinuity(&before), 0);
	assert_eq!(count_discontinuity(&marked), 1, "the break is flagged exactly once");
}

/// The counter-case, and why the fix keys on the discontinuity counter rather than on a
/// backwards slot: video is emitted in decode order, so a reordered (B-frame) PTS steps
/// backwards constantly. Re-emitting on every backwards step was measured at 25x the
/// intended PSI rate, and none of those steps is a rewind.
#[tokio::test(start_paused = true)]
async fn reordered_video_keeps_the_table_cadence() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name, cfg);
	}
	let _sdt = publish_sdt(&mut broadcast, &mut catalog);

	// One group, one keyframe: the PSI has no keyframe boundary to hide behind, so what
	// is counted below is the interval cadence alone. 125 frames at 40ms displayed,
	// emitted in decode order as IPBB quads, so every fourth timestamp jumps 120ms ahead
	// and the next two step back behind it.
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	let mut slice = vec![0x41u8];
	slice.extend(std::iter::repeat_n(0xCD, 200));
	for quad in 0..32u64 {
		for offset in [0, 3, 1, 2] {
			let index = quad * 4 + offset;
			if index >= 125 {
				continue;
			}
			let keyframe = index == 0;
			producer
				.write(Frame {
					timestamp: Timestamp::from_micros(index * 40_000).unwrap(),
					duration: None,
					payload: length_prefixed(&[if keyframe { idr.as_slice() } else { slice.as_slice() }]),
					keyframe,
				})
				.unwrap();
		}
	}

	let mut export = export_of(&consumer).await;
	let out = drain_frames(&mut export).await;

	// 125 frames span 0..4.96s: ten 500ms slots and three 2s slots, one emission each.
	assert_eq!(export.discontinuity(), 0, "a reorder is not a rewind");
	assert_eq!(count_pid(&out, 0x0000), 10, "PAT once per 500ms slot, not per reorder");
	assert_eq!(count_pid(&out, 0x0011), 3, "SDT once per 2s slot, not per reorder");
	// The SPS declares no reorder depth, so the first B-frame grows the reserve once, after
	// the first PCR: that one step back is a new time base, the reorders after it are not.
	assert_eq!(
		count_discontinuity(&out),
		1,
		"only the reserve's growth restarts the clock"
	);
}

/// A program with more than one track marks the break once, not once per track. A
/// declared marker joins every rendition; no track is fenced across it.
#[tokio::test(start_paused = true)]
async fn discontinuity_flags_the_break_once_across_tracks() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let video_track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let audio_track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	{
		let mut guard = catalog.modify().unwrap();
		let mut video = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		video.container = Container::Legacy;
		video.description = Some(avcc);
		guard.video.renditions.insert(video_track.name().to_string(), video);

		let mut audio = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		audio.container = Container::Legacy;
		guard.audio.renditions.insert(audio_track.name().to_string(), audio);
	}

	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 300));
	let mut video = Producer::new(video_track, HangContainer::Legacy(crate::container::Kind::Video));
	let mut audio = Producer::new(audio_track, HangContainer::Legacy(crate::container::Kind::Audio));

	// One keyframe-led second per group on video, 100ms audio frames alongside it.
	let write = |video: &mut Producer<HangContainer>, audio: &mut Producer<HangContainer>, start: u64, seconds: u64| {
		for sec in start..start + seconds {
			video
				.write(Frame {
					timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(),
					duration: None,
					payload: length_prefixed(&[idr.as_slice()]),
					keyframe: true,
				})
				.unwrap();
			video.cut(None).unwrap();
			for tenth in 0..10u64 {
				audio
					.write(Frame {
						timestamp: Timestamp::from_micros(sec * 1_000_000 + tenth * 100_000).unwrap(),
						duration: None,
						payload: Bytes::from_iter((0..180u16).map(|b| (b ^ tenth as u16) as u8)),
						keyframe: tenth == 0,
					})
					.unwrap();
			}
			audio.cut(None).unwrap();
		}
	};

	let mut export = export_of(&consumer).await;
	write(&mut video, &mut audio, 0, 4);
	let before = drain_frames(&mut export).await;
	assert_eq!(export.discontinuity(), 0);
	assert_eq!(count_discontinuity(&before), 0);

	// Independent counters can differ before the shared rewind. Forward markers
	// affect only video; audio must not be fenced out of a continuous timeline.
	for _ in 0..3 {
		video.discontinuity().unwrap();
	}
	video
		.write(Frame {
			timestamp: Timestamp::from_micros(4_000_000).unwrap(),
			duration: None,
			payload: length_prefixed(&[idr.as_slice()]),
			keyframe: true,
		})
		.unwrap();
	video.cut(None).unwrap();
	video
		.write(Frame {
			timestamp: Timestamp::from_micros(4_100_000).unwrap(),
			duration: None,
			payload: length_prefixed(&[idr.as_slice()]),
			keyframe: true,
		})
		.unwrap();
	video.cut(None).unwrap();
	// Audio stays quiet through the marker, so video waits for it and then goes
	// out around it once the wait lapses. The rewind restarts that wait for the new
	// generation, so it lapses twice.
	let mut marked = drain_frames(&mut export).await;
	for _ in 0..2 {
		tokio::time::sleep(RECORDING_MAX_AGE).await;
		marked.extend(drain_frames(&mut export).await);
	}
	assert_eq!(export.discontinuity(), 1, "local marker counts are not program epochs");
	let epoch = export.discontinuity();

	// Both tracks declare the same break and continue forward. Neither is fenced.
	video.discontinuity().unwrap();
	audio.discontinuity().unwrap();
	write(&mut video, &mut audio, 5, 2);
	let after = drain_frames(&mut export).await;

	assert_eq!(
		export.discontinuity(),
		epoch + 1,
		"one break, however many tracks saw it"
	);
	assert_eq!(
		after[0].payload[3] & 0x30,
		0x20,
		"the new timeline leads with its clock"
	);
	assert_eq!(count_discontinuity(&after), 1, "the break is flagged exactly once");
	let bytes: Vec<_> = after.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let (video_pts, audio_pts) = collect_pes_pts(&bytes);
	assert!(!video_pts.is_empty(), "video resumed");
	assert!(!audio_pts.is_empty(), "audio resumed");
	let mut reader = TsPacketReader::new(Cursor::new(bytes));
	let mut video_pid = None;
	let mut video_frames = 0;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		match packet.payload {
			Some(TsPayload::Pmt(pmt)) => {
				video_pid = pmt
					.es_info
					.iter()
					.find(|es| es.stream_type == StreamType::H264)
					.map(|es| es.elementary_pid.as_u16());
			}
			Some(TsPayload::PesStart(pes)) if Some(packet.header.pid.as_u16()) == video_pid => {
				let pts = pes.header.pts.unwrap().as_u64();
				assert!(
					pes.header.dts.is_none_or(|dts| dts.as_u64() <= pts),
					"DTS retained the old epoch"
				);
				video_frames += 1;
			}
			_ => {}
		}
	}
	assert!(video_frames > 0);
	let video_pid = video_pid.expect("video PID in PMT");

	// #3533: a content join on a continuous timeline, audio stepping only a
	// sub-frame (already re-anchored forward). Both tracks keep emitting; no fence.
	write(&mut video, &mut audio, 7, 2);
	let again = drain_frames(&mut export).await;
	assert!(!again.is_empty(), "both tracks kept emitting across the join");
	assert!(count_pid(&again, video_pid) > 0, "video was not fenced");
	assert_eq!(count_discontinuity(&again), 0, "a forward join is not a PCR break");
	let mut counters = std::collections::HashMap::new();
	for frame in before.iter().chain(&marked).chain(&after).chain(&again) {
		for packet in frame.payload.as_chunks::<188>().0.iter() {
			let pid = u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]);
			let cc = packet[3] & 15;
			if let Some(prev) = counters.insert(pid, cc) {
				let expected = (prev + u8::from(packet[3] & 0x10 != 0)) & 15;
				assert_eq!(cc, expected, "counter gap on PID {pid}");
			}
		}
	}
}

/// A DVB SI section larger than one TS packet must be reassembled and captured verbatim.
/// `si_packet` only covers a single-packet section; a real SDT with several services (or a
/// NIT) spans packets, exercising the `SectionReassembler` PUSI + continuity path that
/// feeds the SI store. The body is arbitrary here (capture is verbatim, not parsed).
#[tokio::test(start_paused = true)]
async fn multi_packet_si_section_is_captured() {
	// 400-byte body forces the SDT Actual across three TS packets (183 + 184 + rest).
	let body: Vec<u8> = (0..400u16).map(|i| i as u8).collect();
	let sdt = make_long_section(0x42, 1, 0, 0, 0, &body);
	let input = si_packets_multi(0x0011, &sdt);
	assert!(input.len() > 188, "the SDT must span more than one TS packet");

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&input[..])).unwrap();
	import.finish().unwrap();

	let si = catalog.snapshot().ext.mpegts.si.clone();
	let entry = si
		.get(&0x0011)
		.and_then(|tables| tables.get(&0x42))
		.expect("an SDT entry");
	assert_eq!(
		read_si_sections(&consumer, &entry.track).await,
		vec![Bytes::from(sdt)],
		"the multi-packet SDT was reassembled and captured byte-for-byte"
	);
}

/// Import rig for SI-only fixtures: importer, catalog, and a consumer, all kept
/// alive so the SI tracks stay readable after `finish`.
struct SiRig {
	import: crate::container::ts::Import<tscat::Ext>,
	catalog: crate::catalog::Producer<tscat::Ext>,
	consumer: moq_net::broadcast::Consumer,
}

fn si_rig() -> SiRig {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	SiRig {
		import,
		catalog,
		consumer,
	}
}

/// #2881: a multi-section table revised one section at a time must never publish a
/// torn set. The store buffers the incoming generation and commits it atomically,
/// so every snapshot group holds a single version.
#[tokio::test(start_paused = true)]
async fn torn_transition_is_never_published() {
	let sdt = |version: u8, number: u8, fill: u8| make_long_section(0x42, 1, version, number, 1, &[fill; 4]);
	let mut rig = si_rig();

	// Version 5 arrives complete in one batch.
	let mut input = si_packet(0x0011, &sdt(5, 0, 0xaa));
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt(5, 1, 0xab), 1));
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	// Version 6 arrives torn across two batches; the intermediate state (section 0
	// at v6, section 1 still v5) must never hit the track.
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0011, &sdt(6, 0, 0xba), 2)[..]))
		.unwrap();
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0011, &sdt(6, 1, 0xbb), 3)[..]))
		.unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	let track = &si[&0x0011][&0x42].track;
	let groups = read_si_groups(&rig.consumer, track).await;
	assert!(!groups.is_empty(), "at least one snapshot group");
	for (i, frames) in groups.iter().enumerate() {
		let versions: Vec<u8> = frames
			.iter()
			.flat_map(crate::container::ts::si::split_sections)
			.map(|section| (section[5] >> 1) & 0x1f)
			.collect();
		assert!(
			versions.windows(2).all(|w| w[0] == w[1]),
			"group {i} mixes versions: {versions:?}"
		);
	}
	assert_eq!(
		read_si_sections(&rig.consumer, track).await,
		vec![Bytes::from(sdt(6, 0, 0xba)), Bytes::from(sdt(6, 1, 0xbb))],
		"the newest snapshot is version 6, complete"
	);
}

/// EIT rides per-table entries (#2800): now/next (0x4E) and schedule (0x50) each
/// get their own track and their own cadence, and the schedule's deliberately
/// sparse section numbering (segments skip unused numbers) commits on cycle wrap
/// rather than waiting for a contiguity that never comes.
#[tokio::test(start_paused = true)]
async fn eit_now_next_and_schedule_are_captured() {
	// Body layout past the generic header: TSID(2), ONID(2), then filler.
	let pf0 = make_long_section(0x4E, 1, 0, 0, 1, &[0x00, 0x01, 0x00, 0x02, 0x01, 0x4E, 0xaa]);
	let pf1 = make_long_section(0x4E, 1, 0, 1, 1, &[0x00, 0x01, 0x00, 0x02, 0x01, 0x4E, 0xbb]);
	let sc0 = make_long_section(0x50, 1, 0, 0, 8, &[0x00, 0x01, 0x00, 0x02, 0x08, 0x50, 0xcc]);
	let sc8 = make_long_section(0x50, 1, 0, 8, 8, &[0x00, 0x01, 0x00, 0x02, 0x08, 0x50, 0xdd]);

	let mut input = Vec::new();
	for (cc, section) in [&pf0, &pf1, &sc0, &sc8, &sc0].into_iter().enumerate() {
		input.extend_from_slice(&si_packet_cc(0x0012, section, cc as u8));
	}
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	let eit = si.get(&0x0012).expect("EIT entries");

	let pf = eit.get(&0x4E).expect("a now/next entry");
	assert_eq!(pf.interval, Some(Duration::from_secs(2)), "now/next actual: 2s");
	assert_eq!(
		read_si_sections(&rig.consumer, &pf.track).await,
		vec![Bytes::from(pf0), Bytes::from(pf1)],
		"now/next committed on contiguity"
	);

	let sched = eit.get(&0x50).expect("a schedule entry");
	assert_eq!(sched.interval, Some(Duration::from_secs(10)), "schedule actual: 10s");
	assert_eq!(
		read_si_sections(&rig.consumer, &sched.track).await,
		vec![Bytes::from(sc0), Bytes::from(sc8)],
		"the sparse schedule committed on cycle wrap"
	);
}

/// [`si_rig`] importing only `program`.
fn si_rig_selecting(program: u16) -> SiRig {
	let SiRig {
		import,
		catalog,
		consumer,
	} = si_rig();
	SiRig {
		import: import.with_program(program),
		catalog,
		consumer,
	}
}

/// An SDT service loop entry for `service`, running, with a service_descriptor naming it.
fn sdt_service_entry(service: u16, name: &[u8]) -> Vec<u8> {
	let mut descriptor = vec![0x48, 0, 0x01, 1, b'P', name.len() as u8];
	descriptor.extend_from_slice(name);
	descriptor[1] = (descriptor.len() - 2) as u8;
	let mut entry = service.to_be_bytes().to_vec();
	entry.push(0xfd);
	entry.extend_from_slice(&[0x80, descriptor.len() as u8]);
	entry.extend_from_slice(&descriptor);
	entry
}

/// SDT actual section `number` of `last` for TSID 1 on ONID 2, listing `services`.
fn sdt_actual(version: u8, number: u8, last: u8, services: &[(u16, &[u8])]) -> Vec<u8> {
	let mut body = vec![0x00, 0x02, 0xff];
	for &(service, name) in services {
		body.extend(sdt_service_entry(service, name));
	}
	make_long_section(0x42, 1, version, number, last, &body)
}

/// EIT present/following actual section `number` of 1 for `service` on TSID 1, ONID 2.
fn eit_pf(service: u16, number: u8) -> Bytes {
	let body = [0x00, 0x01, 0x00, 0x02, 0x01, 0x4E, service as u8, number];
	Bytes::from(make_long_section(0x4E, service, 0, number, 1, &body))
}

fn nit() -> Bytes {
	Bytes::from(make_long_section(0x40, 1, 0, 0, 0, &[0xbb; 4]))
}

/// SI for services 1 and 2: an SDT listing service 1 in section 0 and service 2 in
/// section 1, EIT present/following for each, and a NIT.
fn two_service_si() -> Vec<u8> {
	let mut out = si_packet(0x0011, &sdt_actual(0, 0, 1, &[(1, b"One")]));
	out.extend(si_packet_cc(0x0011, &sdt_actual(0, 1, 1, &[(2, b"Two")]), 1));
	for (cc, section) in [eit_pf(1, 0), eit_pf(1, 1), eit_pf(2, 0), eit_pf(2, 1)]
		.iter()
		.enumerate()
	{
		out.extend(si_packet_cc(0x0012, section, cc as u8));
	}
	out.extend(si_packet(0x0010, &nit()));
	out
}

/// `section` is one whole SDT actual for TSID 1 on ONID 2 at `version`, listing only
/// `service` named `name`, under a valid CRC.
fn assert_sdt_lists_only(section: &[u8], version: u8, service: u16, name: &[u8]) {
	let crc = crc::Crc::<u32>::new(&crc::CRC_32_MPEG_2);
	assert_eq!(crc.checksum(section), 0, "a valid CRC-32/MPEG-2");
	let section_length = (usize::from(section[1] & 0x0f) << 8) | usize::from(section[2]);
	assert_eq!(section.len(), 3 + section_length, "section_length covers the section");
	assert_eq!(section[0], 0x42, "SDT actual");
	assert_eq!(
		u16::from_be_bytes([section[3], section[4]]),
		1,
		"transport_stream_id kept"
	);
	assert_eq!((section[5] >> 1) & 0x1f, version, "version_number kept");
	assert_eq!((section[6], section[7]), (0, 0), "section 0 of 0");
	assert_eq!(
		u16::from_be_bytes([section[8], section[9]]),
		2,
		"original_network_id kept"
	);
	assert_eq!(
		&section[11..section.len() - 4],
		&sdt_service_entry(service, name)[..],
		"the service loop holds the selected service alone"
	);
}

/// A selected program's SI describes its own service alone: one SDT actual section
/// listing it, wherever it sat in the source, and only its own EIT. Network-wide
/// tables pass through.
#[tokio::test(start_paused = true)]
async fn a_selected_program_carries_only_its_own_si() {
	for (program, name) in [(1u16, &b"One"[..]), (2, b"Two")] {
		let mut input = crate::container::ts::import::test::two_programs();
		input.extend(two_service_si());
		let mut rig = si_rig_selecting(program);
		rig.import.decode(&BytesMut::from(&input[..])).unwrap();
		rig.import.finish().unwrap();

		let si = rig.catalog.snapshot().ext.mpegts.si.clone();
		let sdt = read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await;
		assert_eq!(sdt.len(), 1, "program {program}: one SDT actual section");
		assert_sdt_lists_only(&sdt[0], 0, program, name);
		assert_eq!(
			read_si_sections(&rig.consumer, &si[&0x0012][&0x4E].track).await,
			vec![eit_pf(program, 0), eit_pf(program, 1)],
			"program {program}: only its own EIT"
		);
		assert_eq!(
			read_si_sections(&rig.consumer, &si[&0x0010][&0x40].track).await,
			vec![nit()],
			"the NIT passes through"
		);
	}
}

/// Without a selection nothing is filtered by program: every section is kept verbatim.
#[tokio::test(start_paused = true)]
async fn an_unselected_import_keeps_every_service() {
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&two_service_si()[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert_eq!(
		read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await,
		vec![
			Bytes::from(sdt_actual(0, 0, 1, &[(1, b"One")])),
			Bytes::from(sdt_actual(0, 1, 1, &[(2, b"Two")])),
		],
		"the SDT verbatim"
	);
	assert_eq!(
		read_si_sections(&rig.consumer, &si[&0x0012][&0x4E].track).await,
		vec![eit_pf(1, 0), eit_pf(1, 1), eit_pf(2, 0), eit_pf(2, 1)],
		"every service's EIT"
	);
}

/// A selected service the SDT does not list gets no SDT actual rather than a
/// fabricated one, and a table filtered to nothing gets no catalog entry.
#[tokio::test(start_paused = true)]
async fn a_selection_missing_from_the_sdt_carries_none() {
	let mut rig = si_rig_selecting(3);
	rig.import.decode(&BytesMut::from(&two_service_si()[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert!(si.contains_key(&0x0010), "the NIT was captured (control)");
	assert!(!si.contains_key(&0x0011), "no SDT entry: {si:?}");
	assert!(!si.contains_key(&0x0012), "no EIT entry: {si:?}");
}

/// An SDT revision that drops the selected service retires the SDT captured before
/// it, catalog entry and all; a later revision listing it again is captured normally.
/// Driven on the capture itself so each revision cuts without waiting out the debounce.
#[tokio::test(start_paused = true)]
async fn an_sdt_revision_dropping_the_service_retires_it() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone());
	capture.select(1);
	let mut revise = |version: u8, services: &[(u16, &[u8])]| {
		capture.section(0x0011, sdt_actual(version, 0, 0, services)).unwrap();
		capture.flush(Timestamp::ZERO, true).unwrap();
		catalog
			.snapshot()
			.ext
			.mpegts
			.si
			.get(&0x0011)
			.map(|tables| tables[&0x42].track.clone())
	};

	let track = revise(0, &[(1, b"One"), (2, b"Two")]).expect("the SDT is advertised");
	assert_eq!(revise(1, &[(2, b"Two")]), None, "the revision retired the SDT entry");
	assert_eq!(
		revise(2, &[(1, b"One"), (2, b"Two")]),
		Some(track.clone()),
		"listed again, on the same track"
	);
	capture.finish(Timestamp::ZERO).unwrap();

	let groups = read_si_groups(&consumer, &track).await;
	assert_eq!(groups.len(), 3, "one snapshot per revision");
	assert_sdt_lists_only(&groups[0][0], 0, 1, b"One");
	assert!(groups[1].is_empty(), "the retiring snapshot carries no SDT");
	assert_sdt_lists_only(&groups[2][0], 2, 1, b"One");
}

/// A corrupted SDT actual is not rebuilt under a fresh, valid CRC: the section is
/// dropped and the last good snapshot stays in force, rather than the corruption
/// reading as a revision (or as the service leaving). An intact revision after it
/// is captured normally.
#[tokio::test(start_paused = true)]
async fn a_corrupt_sdt_keeps_the_last_good_snapshot() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone());
	capture.select(1);
	let mut feed = |section: Vec<u8>| {
		capture.section(0x0011, section).unwrap();
		capture.flush(Timestamp::ZERO, true).unwrap();
		catalog
			.snapshot()
			.ext
			.mpegts
			.si
			.get(&0x0011)
			.map(|tables| tables[&0x42].track.clone())
	};

	let track = feed(sdt_actual(0, 0, 0, &[(1, b"One")])).expect("the SDT is advertised");
	// One unflagged bit flip turns the service name "One" into "Nne" under the source CRC.
	let mut corrupt = sdt_actual(1, 0, 0, &[(1, b"One")]);
	let name = corrupt.len() - 4 - 3;
	corrupt[name] ^= 0x01;
	assert_eq!(feed(corrupt), Some(track.clone()), "the corrupt revision kept the SDT");
	assert_eq!(feed(sdt_actual(2, 0, 0, &[(1, b"Uno")])), Some(track.clone()));
	capture.finish(Timestamp::ZERO).unwrap();

	let groups = read_si_groups(&consumer, &track).await;
	assert_eq!(groups.len(), 2, "no snapshot for the corrupt revision");
	assert_sdt_lists_only(&groups[0][0], 0, 1, b"One");
	assert_sdt_lists_only(&groups[1][0], 2, 1, b"Uno");
}

/// A multi-section SDT revision with the selected service's section corrupt does not
/// commit as complete-as-observed when the other section repeats: that would read as
/// the service leaving. The revision commits once the section arrives intact.
#[tokio::test(start_paused = true)]
async fn a_partial_sdt_revision_keeps_the_last_good_snapshot() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone());
	capture.select(1);
	let mut feed = |section: Vec<u8>| {
		capture.section(0x0011, section).unwrap();
		capture.flush(Timestamp::ZERO, true).unwrap();
		catalog
			.snapshot()
			.ext
			.mpegts
			.si
			.get(&0x0011)
			.map(|tables| tables[&0x42].track.clone())
	};

	feed(sdt_actual(0, 0, 1, &[(1, b"One")]));
	let track = feed(sdt_actual(0, 1, 1, &[(2, b"Two")])).expect("the SDT is advertised");
	let mut corrupt = sdt_actual(1, 0, 1, &[(1, b"One")]);
	let name = corrupt.len() - 4 - 3;
	corrupt[name] ^= 0x01;
	assert_eq!(feed(corrupt), Some(track.clone()));
	assert_eq!(feed(sdt_actual(1, 1, 1, &[(2, b"Two")])), Some(track.clone()));
	assert_eq!(
		feed(sdt_actual(1, 1, 1, &[(2, b"Two")])),
		Some(track.clone()),
		"the repeated section did not commit the partial revision"
	);
	assert_eq!(feed(sdt_actual(1, 0, 1, &[(1, b"Uno")])), Some(track.clone()));
	capture.finish(Timestamp::ZERO).unwrap();

	let groups = read_si_groups(&consumer, &track).await;
	assert_eq!(groups.len(), 2, "no snapshot for the partial revision");
	assert_sdt_lists_only(&groups[0][0], 0, 1, b"One");
	assert_sdt_lists_only(&groups[1][0], 1, 1, b"Uno");
}

/// A selection filters EIT schedule actual (0x50..=0x5F) by service, as it does
/// present/following.
#[tokio::test(start_paused = true)]
async fn a_selection_filters_eit_schedule() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut capture = crate::container::ts::si::Capture::new(broadcast, catalog.clone());
	capture.select(1);
	let schedule = |service: u16| {
		let body = [0x00, 0x01, 0x00, 0x02, 0x00, 0x50, service as u8];
		make_long_section(0x50, service, 0, 0, 0, &body)
	};
	for service in [1, 2] {
		capture.section(0x0012, schedule(service)).unwrap();
	}
	capture.finish(Timestamp::ZERO).unwrap();

	let si = catalog.snapshot().ext.mpegts.si.clone();
	assert_eq!(
		read_si_sections(&consumer, &si[&0x0012][&0x50].track).await,
		vec![Bytes::from(schedule(1))],
		"only the selected service's schedule"
	);
}

/// A selected program's reduced SI survives export: the TS parses, and importing it
/// again finds the same single-service SDT and the same EIT.
#[tokio::test(start_paused = true)]
async fn a_selected_programs_si_survives_export() {
	let mut input = crate::container::ts::import::test::two_programs();
	input.extend(two_service_si());
	let mut rig = si_rig_selecting(2);
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let ts = drain_with(
		Export::with_ts(
			crate::source::announced(&rig.consumer),
			crate::catalog::CatalogFormat::Hang,
		)
		.await
		.unwrap(),
	)
	.await;
	assert_packet_aligned(&ts);
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	while reader.read_ts_packet().unwrap().is_some() {}

	let mut again = si_rig();
	again.import.decode(&BytesMut::from(ts.as_ref())).unwrap();
	again.import.finish().unwrap();
	let si = again.catalog.snapshot().ext.mpegts.si.clone();
	let sdt = read_si_sections(&again.consumer, &si[&0x0011][&0x42].track).await;
	assert_eq!(sdt.len(), 1, "one SDT actual section");
	assert_sdt_lists_only(&sdt[0], 0, 2, b"Two");
	assert_eq!(
		read_si_sections(&again.consumer, &si[&0x0012][&0x4E].track).await,
		vec![eit_pf(2, 0), eit_pf(2, 1)],
		"only program 2's EIT"
	);
}

/// #2842: SDT other sections from two networks that reuse a transport_stream_id
/// must not collide. The identity reads original_network_id (bytes 8..10) for
/// table_id 0x46, so both survive as separate sub-tables; a revision within one
/// network still replaces in place.
#[tokio::test(start_paused = true)]
async fn sdt_other_networks_do_not_collide() {
	// Same TSID (the extension), different ONID leading the body.
	let net1 = make_long_section(0x46, 7, 0, 0, 0, &[0x00, 0x01, 0xaa]);
	let net2 = make_long_section(0x46, 7, 0, 0, 0, &[0x00, 0x02, 0xbb]);
	let net1v2 = make_long_section(0x46, 7, 1, 0, 0, &[0x00, 0x01, 0xcc]);

	let mut input = si_packet(0x0011, &net1);
	input.extend_from_slice(&si_packet_cc(0x0011, &net2, 1));
	input.extend_from_slice(&si_packet_cc(0x0011, &net1v2, 2));
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	let entry = &si[&0x0011][&0x46];
	assert_eq!(
		read_si_sections(&rig.consumer, &entry.track).await,
		vec![Bytes::from(net1v2), Bytes::from(net2)],
		"both networks survive; the revision replaced only its own network"
	);
}

/// A next-version section (current_next_indicator clear) describes a future state
/// and is dropped: only what is currently in force is carried. The current NIT is
/// the positive control proving the pipeline ran; a lone dropped packet would pass
/// vacuously, since sync lock needs a second packet before anything routes.
#[tokio::test(start_paused = true)]
async fn next_version_sections_are_dropped() {
	let mut next = make_long_section(0x42, 1, 3, 0, 0, &[0xaa; 4]);
	next[5] &= !0x01;
	let nit = make_long_section(0x40, 1, 0, 0, 0, &[0xbb; 4]);

	let mut input = si_packet(0x0011, &next);
	input.extend_from_slice(&si_packet_cc(0x0011, &next, 1));
	input.extend_from_slice(&si_packet(0x0010, &nit));
	input.extend_from_slice(&si_packet_cc(0x0010, &nit, 1));
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert!(si.contains_key(&0x0010), "the current NIT was captured (control)");
	assert!(!si.contains_key(&0x0011), "a next-version section creates no entry");
}

/// TDT/TOT (0x0014) is proxied as a latest-value slot (#2914): each tick replaces
/// the last, so the newest snapshot is the source's most recent time, never an
/// accumulation of stale ones. The SDT is the positive control proving the
/// pipeline ran.
#[tokio::test(start_paused = true)]
async fn tdt_round_trips_as_latest_value() {
	let tick = |mjd: u8| make_short_section(0x70, &[0xc0, mjd, 0x12, 0x34, 0x56]);
	let sdt = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 4]);

	let mut rig = si_rig();
	let mut input = si_packet(0x0014, &tick(1));
	input.extend_from_slice(&si_packet_cc(0x0014, &tick(1), 1));
	input.extend_from_slice(&si_packet(0x0011, &sdt));
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt, 1));
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	// The clock ticks: the new section replaces the old slot.
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0014, &tick(2), 2)[..]))
		.unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert!(si.contains_key(&0x0011), "the SDT was captured (control)");
	let entry = si
		.get(&0x0014)
		.and_then(|tables| tables.get(&0x70))
		.expect("a TDT entry");
	assert_eq!(entry.interval, Some(Duration::from_secs(30)), "the TDT/TOT 30s maximum");
	assert_eq!(
		read_si_sections(&rig.consumer, &entry.track).await,
		vec![Bytes::from(tick(2))],
		"the newest snapshot is the latest tick alone"
	);
	let groups = read_si_groups(&rig.consumer, &entry.track).await;
	assert_eq!(groups.len(), 2, "one group per tick, no accumulation");
}

/// Build a well-formed short-form section (syntax indicator clear): header + body,
/// no extension, versioning, or CRC.
fn make_short_section(table_id: u8, body: &[u8]) -> Vec<u8> {
	let mut s = vec![
		table_id,
		0x30 | ((body.len() >> 8) as u8 & 0x0f),
		(body.len() & 0xff) as u8,
	];
	s.extend_from_slice(body);
	s
}

/// Aborting the importer must remove the advertised SI entries from the catalog:
/// a map naming aborted tracks would strand every later exporter on subscriptions
/// that can never deliver.
#[tokio::test(start_paused = true)]
async fn abort_removes_si_catalog_entries() {
	let sdt = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 4]);
	// Twice: sync lock needs a second packet before the first routes at all.
	let mut input = si_packet(0x0011, &sdt);
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt, 1));
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	assert!(
		!rig.catalog.snapshot().ext.mpegts.si.is_empty(),
		"the SDT entry was advertised"
	);

	rig.import.abort(moq_net::Error::Cancel);
	assert!(
		rig.catalog.snapshot().ext.mpegts.si.is_empty(),
		"abort removed the advertised entries"
	);
}

/// A byte-identical short-form repetition is not a change: the latest-value slot
/// only cuts a group when the bytes actually differ.
#[tokio::test(start_paused = true)]
async fn short_form_repetition_cuts_no_group() {
	let tdt = make_short_section(0x70, &[0xc0, 0x79, 0x12, 0x34, 0x56]);

	let mut rig = si_rig();
	let mut input = si_packet(0x0014, &tdt);
	input.extend_from_slice(&si_packet_cc(0x0014, &tdt, 1));
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0014, &tdt, 2)[..]))
		.unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	let groups = read_si_groups(&rig.consumer, &si[&0x0014][&0x70].track).await;
	assert_eq!(groups.len(), 1, "a repetition cut no further group");
}

/// SI never gates output: an advertised entry whose track resolves but never
/// delivers a snapshot (a stale announce) must not hold the programme dark.
/// Media flows immediately and the entry simply emits nothing.
#[tokio::test(start_paused = true)]
async fn stale_si_entry_does_not_block_output() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	// A track that exists (the subscription resolves) but never produces a group.
	let ghost = broadcast.create_track("ghost.si", None).unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name.clone(), cfg);
		guard.ext.mpegts.si.entry(0x0011).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "ghost.si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	}

	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 64));
	producer
		.write(Frame {
			timestamp: Timestamp::ZERO,
			duration: None,
			payload: length_prefixed(&[&idr]),
			keyframe: true,
		})
		.unwrap();
	producer.cut(None).unwrap();
	producer.finish().unwrap();

	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap();
	// The timeout distinguishes "produced output promptly" from "held dark";
	// under paused time a wedged exporter would hit it instantly.
	let frame = tokio::time::timeout(Duration::from_secs(1), exporter.next())
		.await
		.expect("a stale SI entry must not hold output dark")
		.unwrap()
		.expect("a muxed frame");
	assert!(!frame.payload.is_empty());
	assert!(
		!frame
			.payload
			.as_chunks::<188>()
			.0
			.iter()
			.any(|p| ((((p[1] & 0x1f) as u16) << 8) | p[2] as u16) == 0x0011),
		"nothing was emitted for the undelivered entry"
	);
	drop(ghost);
}

/// A raw Opus packet: a one-byte TOC (config 1 = SILK NB 20 ms, stereo, code 0) plus
/// `len` filler bytes, so it parses as one 20 ms frame.
fn opus_packet(fill: u8, len: usize) -> Bytes {
	let mut p = vec![(1 << 3) | (1 << 2)]; // TOC: config=1, s=1 (stereo), c=0
	p.extend(std::iter::repeat_n(fill, len));
	Bytes::from(p)
}

/// Strip the Opus-in-TS control header from a PES payload, returning the raw packets it
/// carries. Assumes no trim / no control extension (what the exporter emits).
fn strip_opus_control(mut data: &[u8]) -> Vec<Vec<u8>> {
	let mut packets = Vec::new();
	while !data.is_empty() {
		assert_eq!(data[0], 0x7f, "control header sync byte 0");
		assert_eq!(data[1] & 0xe0, 0xe0, "control header sync byte 1");
		assert_eq!(data[1] & 0x1c, 0x00, "exporter emits no trim / extension flags");
		let mut pos = 2;
		let mut size = 0usize;
		loop {
			let b = data[pos];
			pos += 1;
			size += b as usize;
			if b != 0xff {
				break;
			}
		}
		packets.push(data[pos..pos + size].to_vec());
		data = &data[pos + size..];
	}
	packets
}

/// Export an Opus broadcast and assert the program tables advertise a private-data
/// (0x06) stream carrying the 'Opus' registration + DVB extension descriptors, and that
/// the control-header-wrapped PES recovers the raw Opus packets with the right PTS.
#[tokio::test(start_paused = true)]
async fn export_opus_roundtrip() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".opus"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AudioCodec::Opus, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().audio.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	// The last packet is > 184 bytes to force PES splitting across TS packets.
	let packets: Vec<Bytes> = vec![opus_packet(0x01, 4), opus_packet(0x10, 8), opus_packet(0x20, 200)];
	for (i, payload) in packets.iter().enumerate() {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i as u64 * 20_000).unwrap(),
				duration: None,
				payload: payload.clone(),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	}
	producer.finish().unwrap();

	let ts = drain(consumer).await;
	assert_packet_aligned(&ts);

	// Pass 1: one private-data stream with the Opus registration + extension descriptors.
	let mut reader = TsPacketReader::new(Cursor::new(ts.as_ref()));
	let mut saw_pmt = false;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::Pmt(pmt)) = packet.payload {
			saw_pmt = true;
			assert_eq!(pmt.es_info.len(), 1);
			let es = &pmt.es_info[0];
			assert_eq!(es.stream_type, StreamType::from_u8(0x06).unwrap());
			let reg = es.descriptors.iter().find(|d| d.tag == 0x05).expect("registration");
			assert_eq!(reg.data, b"Opus");
			let ext = es.descriptors.iter().find(|d| d.tag == 0x7f).expect("extension");
			assert_eq!(ext.data, vec![0x80, 0x02], "ext tag 0x80 + stereo channel_config_code");
		}
	}
	assert!(saw_pmt, "missing PMT");

	// Pass 2: reassemble PES packets and recover the raw Opus packets.
	let mut pes = PesPacketReader::new(TsPacketReader::new(Cursor::new(ts.as_ref())));
	let mut recovered: Vec<(u64, Vec<u8>)> = Vec::new();
	while let Some(packet) = pes.read_pes_packet().unwrap() {
		assert_eq!(
			packet.header.stream_id.as_u8(),
			mpeg2ts::es::StreamId::PRIVATE_STREAM_1,
			"Opus rides private_stream_1"
		);
		let pts = packet.header.pts.expect("PES carried no PTS").as_u64();
		for raw in strip_opus_control(&packet.data) {
			recovered.push((pts, raw));
		}
	}

	assert_eq!(recovered.len(), packets.len());
	for (i, payload) in packets.iter().enumerate() {
		let (pts, raw) = &recovered[i];
		assert_eq!(*pts, i as u64 * 20 * 90, "PTS should be ms * 90 (90 kHz)");
		assert_eq!(raw.as_slice(), payload.as_ref(), "raw Opus payload mismatch");
	}
}

/// Round-trip an Opus broadcast through TS and back: export, re-import, and confirm the
/// catalog surfaces one 48 kHz Opus track whose frames recover the original packets.
#[tokio::test(start_paused = true)]
async fn opus_export_import_roundtrip() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".opus"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AudioCodec::Opus, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().audio.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));

	let packets: Vec<Bytes> = (0..4).map(|i| opus_packet(0x40 + i as u8, 24)).collect();
	for (i, payload) in packets.iter().enumerate() {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i as u64 * 20_000).unwrap(),
				duration: None,
				payload: payload.clone(),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	}
	producer.finish().unwrap();

	let ts = drain(consumer).await;

	// Re-import the TS we just produced.
	let mut imported = moq_net::broadcast::Info::new().produce();
	let imported_consumer = imported.consume();
	let import_catalog = crate::catalog::Producer::new(&mut imported, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(imported, import_catalog.reserve());
	import.decode(&ts).unwrap();
	import.finish().unwrap();

	let snapshot = import_catalog.snapshot();
	assert_eq!(snapshot.audio.renditions.len(), 1, "expected one Opus track");
	let (opus_name, audio) = snapshot.audio.renditions.iter().next().unwrap();
	assert_eq!(audio.codec.to_string(), "opus");
	assert_eq!(audio.sample_rate, 48_000);
	assert_eq!(audio.channel_count, 2);

	// The imported packets must match what we published.
	let recovered = read_frames(&imported_consumer, opus_name, Kind::Audio).await;
	assert_eq!(recovered.len(), packets.len(), "frame count");
	for (orig, got) in packets.iter().zip(&recovered) {
		assert_eq!(got.as_slice(), orig.as_ref(), "Opus packet survived the round-trip");
	}
}

// Two exporters of one broadcast, started at different times, must render the same packets
// from the moment they overlap. That is what a redundant (SMPTE ST 2022-7) pair compares, and
// it is what lets a leg be restarted without the merge at the far end seeing the two disagree.
// See moq-dev/moq#2779.

/// 25 fps video.
const VIDEO_US: u64 = 40_000;
/// 48 kHz AAC, 1024 samples per frame.
const AUDIO_US: u64 = 21_333;
/// Video frames per group. Deliberately not a whole number of PSI intervals, so a table
/// cadence anchored anywhere but the media timeline drifts against the keyframes.
const GOP: u64 = 15;
/// Audio frames per group, roughly matching the video group duration.
const AUDIO_GROUP: u64 = 28;
/// Video-frame ticks to produce, and the tick the second exporter joins at.
const TICKS: u64 = 150;
const JOIN: u64 = 75;

/// Produce one broadcast and export it twice, the second exporter joining partway in, and
/// return what each rendered.
///
/// Both exporters are drained after every write, so neither skips a group and the two see the
/// same arrival order. That isolates the question under test (does the *rendering* depend on
/// when the process started) from the separate question of whether two legs received the same
/// groups in the same order. The program carries an SDT, whose unchanged repeats have to land on
/// the same media times in both legs too (#3948).
async fn export_twice(with_video: bool) -> (Vec<Frame>, Vec<Frame>) {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let _sdt = publish_sdt(&mut broadcast, &mut catalog);

	let mut video = with_video.then(|| {
		let track = broadcast
			.create_track(
				broadcast.unique_name(".avc1"),
				hang::container::track_info(hang::catalog::PRIORITY.video),
			)
			.unwrap();
		// Out-of-band parameter sets (avc1), so the export source takes the catalog
		// description as-is instead of parsing them out of the bitstream.
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description =
			Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap());
		catalog
			.modify()
			.unwrap()
			.video
			.renditions
			.insert(track.name().to_string(), cfg);
		Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data))
	});

	let mut audio = {
		let track = broadcast
			.create_track(
				broadcast.unique_name(".aac"),
				hang::container::track_info(hang::catalog::PRIORITY.audio),
			)
			.unwrap();
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(track.name().to_string(), cfg);
		Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data))
	};

	let source = crate::source::announced(&consumer);
	let export = || Export::with_ts(source.clone(), crate::catalog::CatalogFormat::Hang);
	let mut a = export().await.unwrap();
	let mut b = None;
	let (mut out_a, mut out_b) = (Vec::new(), Vec::new());

	let mut audio_index = 0;
	for tick in 0..TICKS {
		if let Some(video) = video.as_mut() {
			let keyframe = tick % GOP == 0;
			let slice = if keyframe {
				vec![0x65u8; 3_000]
			} else {
				vec![0x41u8; 400]
			};
			video
				.write(Frame {
					timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(),
					duration: None,
					payload: length_prefixed(&[&slice]),
					keyframe,
				})
				.unwrap();
		}
		// Every audio frame that starts before the next video tick.
		while audio_index * AUDIO_US < (tick + 1) * VIDEO_US {
			audio
				.write(Frame {
					timestamp: Timestamp::from_micros(audio_index * AUDIO_US).unwrap(),
					duration: None,
					payload: Bytes::from_iter((0..180u16).map(|i| (i ^ audio_index as u16) as u8)),
					keyframe: audio_index % AUDIO_GROUP == 0,
				})
				.unwrap();
			audio_index += 1;
		}

		out_a.extend(drain_frames(&mut a).await);
		if let Some(b) = b.as_mut() {
			out_b.extend(drain_frames(b).await);
		}
		if tick + 1 == JOIN {
			b = Some(export().await.unwrap());
		}
	}

	(out_a, out_b)
}

/// Pull every frame an exporter can render right now, like `drain` but keeping the frames
/// whole (this compares them one by one, not as one byte stream).
async fn drain_frames<E: tscat::Catalog>(export: &mut Export<E>) -> Vec<Frame> {
	let mut out = Vec::new();
	while let Ok(res) = tokio::time::timeout(Duration::from_secs(1), export.next()).await {
		match res.expect("exporter error") {
			Some(frame) => out.push(frame),
			None => break,
		}
	}
	out
}

/// Compare the overlapping output of two exporters, starting at `from`, and assert the only
/// bytes that disagree are continuity counters.
///
/// The counter is the known exception: it is numbered from process state, so two legs are
/// offset by a constant. Fixing that needs the emitted packet count per group to be a function
/// of the broadcast, which is a much larger change than this guards. Everything else has to
/// match exactly, so this fails if any new field starts being minted per process.
fn assert_only_continuity_differs(a: &[Frame], b: &[Frame], from: Timestamp) {
	let a: Vec<&Frame> = a.iter().filter(|f| f.timestamp >= from).collect();
	let b: Vec<&Frame> = b.iter().filter(|f| f.timestamp >= from).collect();
	assert!(b.len() > 20, "not enough overlap to be worth comparing: {}", b.len());
	assert_eq!(a.len(), b.len(), "exporters rendered a different number of frames");

	for (a, b) in a.iter().zip(b.iter()) {
		assert_eq!(a.timestamp, b.timestamp, "compared frames must be the same frame");
		assert_eq!(
			a.payload.len(),
			b.payload.len(),
			"same frame rendered to a different size"
		);
		assert_packet_aligned(&a.payload);

		for (offset, (x, y)) in a.payload.iter().zip(b.payload.iter()).enumerate() {
			if x == y {
				continue;
			}
			// Byte 3 of a TS packet is `transport_scrambling_control | adaptation_field_control |
			// continuity_counter`, and only the low nibble is the counter. A difference anywhere
			// else is a value the exporter minted from its own state rather than from the broadcast.
			assert_eq!(
				(offset % 188, (x ^ y) & 0xf0),
				(3, 0),
				"frame at {:?} differs outside the continuity counter: offset {offset}, {x:#04x} vs {y:#04x}",
				a.timestamp,
			);
		}
	}
}

#[tokio::test(start_paused = true)]
async fn late_join_matches_a_running_exporter() {
	let (a, b) = export_twice(true).await;

	// Skip to the joiner's second keyframe: its first group covers tune-in, where the two legs
	// legitimately differ because only the joiner has to lead with the program tables.
	let keyframes: Vec<Timestamp> = b.iter().filter(|f| f.keyframe).map(|f| f.timestamp).collect();
	assert_only_continuity_differs(&a, &b, keyframes[1]);
}

/// The same property for a program with no video track. Worth its own case because the program
/// tables are re-emitted at every video keyframe, a boundary both legs share, which hides a
/// drifting cadence. Audio-only has no such boundary, so the cadence has to come from the media
/// timeline on its own.
#[tokio::test(start_paused = true)]
async fn late_join_matches_a_running_exporter_without_video() {
	let (a, b) = export_twice(false).await;

	// The joiner's first span leads with PAT/PMT, while the running exporter has no
	// reason to repeat them there. The span can cover several PCR slices, so compare
	// from the next PAT/PMT refresh, which both exporters anchor to the media grid.
	let from = b
		.iter()
		.filter(|frame| count_pid(std::slice::from_ref(*frame), 0) > 0)
		.nth(1)
		.expect("a periodic PAT/PMT refresh")
		.timestamp;
	assert_only_continuity_differs(&a, &b, from);
}

// The interleave is a function of the media, not of arrival (moq-dev/moq#2829): the
// earliest pending frame waits for every track to show it cannot be preceded, bounded
// by the exporter's `max_age`.

/// A video and an audio rendition written frame by frame, on the late-join fixture's
/// cadence, so a test decides exactly what each exporter has seen when it polls.
struct Interleave {
	_broadcast: moq_net::broadcast::Producer,
	_catalog: crate::catalog::Producer,
	video: Producer<HangContainer>,
	audio: Producer<HangContainer>,
	source: crate::Source,
	/// The next audio frame to write.
	audio_index: u64,
}

impl Interleave {
	fn new() -> Self {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let consumer = broadcast.consume();
		let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

		let track = broadcast
			.create_track(
				broadcast.unique_name(".avc1"),
				hang::container::track_info(hang::catalog::PRIORITY.video),
			)
			.unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description =
			Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap());
		catalog
			.modify()
			.unwrap()
			.video
			.renditions
			.insert(track.name().to_string(), cfg);
		let video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Video));

		let track = broadcast
			.create_track(
				broadcast.unique_name(".aac"),
				hang::container::track_info(hang::catalog::PRIORITY.audio),
			)
			.unwrap();
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(track.name().to_string(), cfg);
		let audio = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Audio));

		Self {
			source: crate::source::announced(&consumer),
			_broadcast: broadcast,
			_catalog: catalog,
			video,
			audio,
			audio_index: 0,
		}
	}

	/// An exporter with its subscriptions resolved, so later polls need no runtime.
	async fn export(&self, max_age: Duration) -> Export {
		let mut export = Export::new(self.source.clone()).await.unwrap().with_max_age(max_age);
		assert!(drain_frames(&mut export).await.is_empty());
		export
	}

	fn video(&mut self, tick: u64) {
		let keyframe = tick.is_multiple_of(GOP);
		let slice = if keyframe {
			vec![0x65u8; 3_000]
		} else {
			vec![0x41u8; 400]
		};
		self.video
			.write(Frame {
				timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(),
				duration: None,
				payload: length_prefixed(&[&slice]),
				keyframe,
			})
			.unwrap();
	}

	/// Write every audio frame that starts before `until` microseconds, one at a time,
	/// polling `export` after each.
	fn audio_until(&mut self, until: u64, export: &mut Export, out: &mut Vec<Frame>) {
		while self.audio_index * AUDIO_US < until {
			let index = self.audio_index;
			self.audio
				.write(Frame {
					timestamp: Timestamp::from_micros(index * AUDIO_US).unwrap(),
					duration: None,
					payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)),
					keyframe: index.is_multiple_of(AUDIO_GROUP),
				})
				.unwrap();
			self.audio_index += 1;
			out.extend(poll_frames(export));
		}
	}

	fn finish(&mut self) {
		self.video.finish().unwrap();
		self.audio.finish().unwrap();
	}
}

/// Pull every frame an exporter can render without letting time pass, so a `max_age`
/// wait lapses only where a test advances the clock itself.
fn poll_frames<E: tscat::Catalog>(export: &mut Export<E>) -> Vec<Frame> {
	let waiter = kio::Waiter::noop();
	let mut out = Vec::new();
	while let std::task::Poll::Ready(frame) = export.poll_next(&waiter) {
		match frame.expect("exporter error") {
			Some(frame) => out.push(frame),
			None => break,
		}
	}
	out
}

/// Every PES start's PTS in the order the stream carries them, across PIDs.
fn pes_pts_in_order(frames: &[Frame]) -> Vec<u64> {
	let bytes: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let mut reader = TsPacketReader::new(Cursor::new(bytes));
	let mut out = Vec::new();
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(TsPayload::PesStart(pes)) = packet.payload
			&& let Some(pts) = pes.header.pts
		{
			out.push(pts.as_u64());
		}
	}
	out
}

/// Audio lagging video by a frame: each video frame lands before the audio that
/// precedes it. Returns the PTS order the exporter emitted.
async fn lagging_audio(max_age: Duration) -> Vec<u64> {
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();
	for tick in 0..TICKS / 2 {
		rig.video(tick);
		out.extend(poll_frames(&mut export));
		rig.audio_until(tick * VIDEO_US, &mut export, &mut out);
	}
	rig.finish();
	out.extend(drain_frames(&mut export).await);
	pes_pts_in_order(&out)
}

#[tokio::test(start_paused = true)]
async fn late_audio_still_leads_the_interleave() {
	let pts = lagging_audio(Duration::from_millis(500)).await;
	assert!(pts.len() > 100, "too little output to judge: {}", pts.len());
	assert!(pts.is_sorted(), "a frame went out ahead of an earlier one: {pts:?}");
}

/// Zero `max_age` waits for nothing, so the interleave stays in arrival order.
#[tokio::test(start_paused = true)]
async fn zero_max_age_keeps_arrival_order() {
	let pts = lagging_audio(Duration::ZERO).await;
	assert!(
		!pts.is_sorted(),
		"video should have led the audio that arrived after it"
	);
}

/// Two exporters that see the same frames arrive in different orders render the same
/// bytes: one polls after every frame (video ahead of its audio), the other only once
/// each tick's frames have all landed.
#[tokio::test(start_paused = true)]
async fn arrival_order_does_not_change_the_output() {
	let mut rig = Interleave::new();
	let max_age = Duration::from_millis(500);
	let (mut eager, mut batched) = (rig.export(max_age).await, rig.export(max_age).await);
	let (mut out_eager, mut out_batched) = (Vec::new(), Vec::new());
	for tick in 0..TICKS / 2 {
		rig.video(tick);
		out_eager.extend(poll_frames(&mut eager));
		rig.audio_until(tick * VIDEO_US, &mut eager, &mut out_eager);
		out_batched.extend(poll_frames(&mut batched));
	}
	rig.finish();
	out_eager.extend(drain_frames(&mut eager).await);
	out_batched.extend(drain_frames(&mut batched).await);

	let eager: Vec<u8> = out_eager.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let batched: Vec<u8> = out_batched.iter().flat_map(|f| f.payload.iter().copied()).collect();
	assert!(eager.len() > 100 * 188, "too little output to compare: {}", eager.len());
	assert!(eager == batched, "arrival order changed the rendering");
}

/// A track quiet past `max_age` is emitted around, and once it catches up the
/// interleave waits for it again.
#[tokio::test(start_paused = true)]
async fn quiet_track_is_emitted_around_then_rejoins() {
	let max_age = Duration::from_millis(500);
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();

	rig.video(0);
	rig.audio_until(1, &mut export, &mut out);

	// Audio goes quiet while video runs on. Video waits for it, then goes around it.
	for tick in 1..=10 {
		rig.video(tick);
	}
	assert!(poll_frames(&mut export).is_empty(), "video waits for the quiet audio");
	tokio::time::advance(max_age).await;
	out.extend(poll_frames(&mut export));
	assert!(
		pes_pts_in_order(&out)
			.iter()
			.any(|&pts| pts >= 9 * VIDEO_US * 90 / 1_000),
		"video went out around the quiet audio"
	);

	// Audio catches up with its backlog, then keeps lagging by a frame: the order
	// resumes without the clock moving.
	rig.audio_until(10 * VIDEO_US, &mut export, &mut out);
	for tick in 11..TICKS / 2 {
		rig.video(tick);
		out.extend(poll_frames(&mut export));
		rig.audio_until(tick * VIDEO_US, &mut export, &mut out);
	}
	rig.finish();
	out.extend(drain_frames(&mut export).await);

	// Everything from the first frame written after the catch-up on is in order.
	let mut pts = pes_pts_in_order(&out);
	let resumed = pts.iter().position(|&pts| pts >= 11 * VIDEO_US * 90 / 1_000).unwrap();
	let pts = pts.split_off(resumed);
	assert!(pts.len() > 100, "too little output to judge: {}", pts.len());
	assert!(pts.is_sorted(), "the interleave did not resume: {pts:?}");
}

/// Audio dropped for leading the first keyframe does not stall the interleave: the
/// audio already cached past the keyframe shows where the track resumes.
#[tokio::test(start_paused = true)]
async fn tune_in_does_not_wait_on_dropped_audio() {
	let mut rig = Interleave::new();
	let mut export = rig.export(Duration::from_millis(500)).await;
	let mut out = Vec::new();
	rig.audio_until(3 * GOP * VIDEO_US, &mut export, &mut out);
	for tick in GOP..2 * GOP {
		rig.video(tick);
	}
	out.extend(poll_frames(&mut export));
	assert!(!out.is_empty(), "the tune-in waited on audio it had dropped");
}

/// A rewind taken while going around a quiet track keeps going around it: the
/// expired hold carries into the new generation rather than starting afresh.
#[tokio::test(start_paused = true)]
async fn rewind_keeps_an_expired_stall() {
	let max_age = Duration::from_millis(500);
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();

	rig.video(0);
	rig.audio_until(1, &mut export, &mut out);
	for tick in 1..=5 {
		rig.video(tick);
	}
	assert!(poll_frames(&mut export).is_empty(), "video waits for the quiet audio");
	tokio::time::advance(max_age).await;
	out.extend(poll_frames(&mut export));
	assert!(!out.is_empty(), "video went out around the quiet audio");

	rig.video.discontinuity().unwrap();
	for tick in 0..=5 {
		rig.video(GOP + tick);
	}
	out.extend(poll_frames(&mut export));
	let after = pes_pts_in_order(&out)
		.into_iter()
		.filter(|&pts| pts >= GOP * VIDEO_US * 90 / 1_000)
		.count();
	assert!(after > 0, "the new generation waited on the audio again");
	assert_eq!(export.discontinuity(), 1);
}

/// A frame held across a rewind keeps the wait it has already served.
#[tokio::test(start_paused = true)]
async fn rewind_keeps_a_held_frame_waiting_time() {
	let max_age = Duration::from_millis(500);
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();

	rig.video(0);
	rig.audio_until(1, &mut export, &mut out);
	for tick in 1..=5 {
		rig.video(tick);
	}
	// Audio jumps past the coming break, so it is held waiting on the video.
	rig.audio_index = (GOP + 3) * VIDEO_US / AUDIO_US;
	rig.audio_until(rig.audio_index * AUDIO_US + 1, &mut export, &mut out);
	tokio::time::advance(max_age).await;

	rig.video.discontinuity().unwrap();
	rig.video(GOP);
	out.extend(poll_frames(&mut export));
	assert_eq!(export.discontinuity(), 1, "the held audio waited a second budget");
}

/// A rewind part way through a hold neither ends it nor renews it: the new generation
/// waits out only what is left of the budget.
#[tokio::test(start_paused = true)]
async fn rewind_keeps_a_partial_stall() {
	let max_age = Duration::from_millis(500);
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();

	rig.video(0);
	rig.audio_until(1, &mut export, &mut out);
	for tick in 1..=5 {
		rig.video(tick);
	}
	assert!(poll_frames(&mut export).is_empty(), "video waits for the quiet audio");
	tokio::time::advance(max_age / 2).await;
	assert!(poll_frames(&mut export).is_empty(), "the hold lapsed early");

	rig.video.discontinuity().unwrap();
	for tick in 0..=5 {
		rig.video(GOP + tick);
	}
	assert!(poll_frames(&mut export).is_empty(), "the rewind ended the hold");

	tokio::time::advance(max_age / 2).await;
	assert!(!poll_frames(&mut export).is_empty(), "the rewind renewed the hold");
	assert_eq!(export.discontinuity(), 1);
}

/// Under loss every source skip is a rewind. However many arrive while a sparse
/// track stays quiet, the interleave holds for one budget, not one per rewind:
/// renewing it at each would delay every source by the budget they skip on,
/// and the feed would collapse into alternating holds and skips.
#[tokio::test(start_paused = true)]
async fn repeated_rewinds_hold_once() {
	let max_age = Duration::from_millis(500);
	let mut rig = Interleave::new();
	let mut export = rig.export(max_age).await;
	let mut out = Vec::new();

	rig.video(0);
	rig.audio_until(1, &mut export, &mut out);
	for tick in 1..=5 {
		rig.video(tick);
	}
	assert!(poll_frames(&mut export).is_empty(), "video waits for the quiet audio");
	tokio::time::advance(max_age).await;
	out.extend(poll_frames(&mut export));

	for generation in 1..=8 {
		rig.video.discontinuity().unwrap();
		for tick in 0..=5 {
			rig.video(generation * GOP + tick);
		}
		out.extend(poll_frames(&mut export));
		assert_eq!(
			export.discontinuity(),
			generation,
			"rewind {generation} held the feed for a fresh budget"
		);
	}
}

/// A section lost before the cycle wraps commits an observed subset; the next
/// cycle must *converge* to the full set rather than flip-flop between subsets.
/// The repetition fast-path skips sections already active, so without the
/// same-version merge in `commit` the stragglers would replace the subset instead
/// of completing it, oscillating forever.
#[tokio::test(start_paused = true)]
async fn lost_dense_section_recovers_on_the_next_cycle() {
	let sdt = |number: u8, fill: u8| make_long_section(0x42, 1, 0, number, 2, &[fill; 4]);
	let s0 = sdt(0, 0xa0);
	let s1 = sdt(1, 0xa1);
	let s2 = sdt(2, 0xa2);

	// Cycle 1 loses section 1; the repeat of section 0 wraps and commits {0, 2}.
	let mut input = si_packet(0x0011, &s0);
	input.extend_from_slice(&si_packet_cc(0x0011, &s2, 1));
	input.extend_from_slice(&si_packet_cc(0x0011, &s0, 2));
	// Cycle 2 supplies section 1; sections 0 and 2 short-circuit as repetitions,
	// so the wrap carries only the straggler, which must merge, not replace.
	input.extend_from_slice(&si_packet_cc(0x0011, &s1, 3));
	input.extend_from_slice(&si_packet_cc(0x0011, &s1, 4));
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert_eq!(
		read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await,
		vec![Bytes::from(s0), Bytes::from(s1), Bytes::from(s2)],
		"the lost section joined the committed generation instead of replacing it"
	);
}

/// A catalog update that keeps the `(PID, table_id)` key but repoints it at a new
/// track (a restarted publisher) must rebuild the subscription: staying attached
/// to the old track would repeat its stale sections forever.
#[tokio::test(start_paused = true)]
async fn repointed_si_entry_resubscribes() {
	let sdt_a = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8]);
	let sdt_b = make_long_section(0x42, 1, 1, 0, 0, &[0xbb; 8]);

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let mut track_a = broadcast.create_track("a.si", None).unwrap();
	track_a
		.write_frame(Timestamp::ZERO, Bytes::from(sdt_a.clone()))
		.unwrap();
	let mut track_b = broadcast.create_track("b.si", None).unwrap();
	track_b
		.write_frame(Timestamp::ZERO, Bytes::from(sdt_b.clone()))
		.unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name.clone(), cfg);
		guard.ext.mpegts.si.entry(0x0011).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "a.si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	}

	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 64));
	let write_key = |producer: &mut Producer<HangContainer>, sec: u64| {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(),
				duration: None,
				payload: length_prefixed(&[&idr]),
				keyframe: true,
			})
			.unwrap();
		producer.cut(None).unwrap();
	};

	// Writes both GOPs up front and only then reads, so it needs a replay window:
	// the real-time default would take the live edge and skip the first.
	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	let mut before = BytesMut::new();
	write_key(&mut producer, 0);
	write_key(&mut producer, 1);
	for frame in drain_frames(&mut exporter).await {
		before.extend_from_slice(&frame.payload);
	}

	// Repoint the entry at the replacement track.
	catalog
		.modify()
		.unwrap()
		.ext
		.mpegts
		.si
		.get_mut(&0x0011)
		.unwrap()
		.insert(
			0x42,
			tscat::SiEntry {
				track: "b.si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);

	// The repointed entry resubscribes through the origin, whose driver mints and
	// splices the new track between polls: give it the scheduler before the frames
	// that should carry the new sections, or the exporter drains them all first.
	let mut after = BytesMut::new();
	write_key(&mut producer, 2);
	let frame = tokio::time::timeout(Duration::from_secs(1), exporter.next())
		.await
		.expect("a frame after the switch")
		.unwrap()
		.unwrap();
	after.extend_from_slice(&frame.payload);
	for _ in 0..100 {
		tokio::task::yield_now().await;
	}
	write_key(&mut producer, 3);
	write_key(&mut producer, 4);
	producer.finish().unwrap();
	while let Ok(res) = tokio::time::timeout(Duration::from_secs(1), exporter.next()).await {
		let Some(frame) = res.unwrap() else { break };
		after.extend_from_slice(&frame.payload);
	}

	let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle);
	assert!(
		contains(&before, &sdt_a),
		"the original track's SDT was emitted (control)"
	);
	assert!(
		contains(&after, &sdt_b),
		"the replacement track's SDT is emitted after the repoint"
	);
}

/// An SI revision that lands after the final media frame still reaches the TS:
/// emission rides media frames, so end of stream flushes every entry's current
/// sections in one trailing frame before yielding `None`.
#[tokio::test(start_paused = true)]
async fn si_revision_after_final_media_frame_is_flushed() {
	let sdt_v1 = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8]);
	let sdt_v2 = make_long_section(0x42, 1, 1, 0, 0, &[0xbb; 8]);

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let mut si_track = broadcast.create_track("0x0011-0x42.si", None).unwrap();
	si_track
		.write_frame(Timestamp::ZERO, Bytes::from(sdt_v1.clone()))
		.unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name.clone(), cfg);
		guard.ext.mpegts.si.entry(0x0011).or_default().insert(
			0x42,
			tscat::SiEntry {
				track: "0x0011-0x42.si".to_string(),
				interval: Some(Duration::from_secs(2)),
				..Default::default()
			},
		);
	}

	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut idr = vec![0x65u8];
	idr.extend(std::iter::repeat_n(0xAB, 64));
	producer
		.write(Frame {
			timestamp: Timestamp::ZERO,
			duration: None,
			payload: length_prefixed(&[&idr]),
			keyframe: true,
		})
		.unwrap();
	producer.cut(None).unwrap();
	producer.finish().unwrap();

	let mut exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap();
	let first = loop {
		let frame = tokio::time::timeout(Duration::from_secs(1), exporter.next())
			.await
			.expect("the only media frame")
			.unwrap()
			.unwrap();
		if !is_pcr_frame(&frame) {
			break frame;
		}
	};
	let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle);
	assert!(contains(&first.payload, &sdt_v1), "v1 rode the media frame (control)");

	// The revision lands after the last media frame; only the trailing flush can
	// carry it. Closing the catalog lets the exporter reach end of stream.
	si_track
		.write_frame(Timestamp::ZERO, Bytes::from(sdt_v2.clone()))
		.unwrap();
	catalog.finish().unwrap();

	let tail = tokio::time::timeout(Duration::from_secs(1), exporter.next())
		.await
		.expect("a trailing SI frame rather than an immediate end")
		.unwrap()
		.expect("the trailing SI frame");
	assert_packet_aligned(&tail.payload);
	assert!(
		contains(&tail.payload, &sdt_v2),
		"the trailing flush carries the revision"
	);
	let end = tokio::time::timeout(Duration::from_secs(1), exporter.next())
		.await
		.expect("the stream ends after the flush")
		.unwrap();
	assert!(end.is_none(), "end of stream after the trailing flush");
}

/// A broadcast with one avc1 video track and one SI entry, for driving SI emission
/// frame by frame: `media(millis, pid)` writes one video frame and reports how
/// many packets of `pid` ride the output it produced.
struct SiCadenceRig {
	started: bool,
	producer: Producer<HangContainer>,
	si_track: moq_net::track::Producer,
	exporter: Export<tscat::Ext>,
	catalog: crate::catalog::Producer<tscat::Ext>,
}

async fn si_cadence_rig(pid: u16, table_id: u8, interval: Duration) -> SiCadenceRig {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let si_name = format!("{pid:#06x}-{table_id:#04x}.si");
	let si_track = broadcast.create_track(si_name.as_str(), None).unwrap();

	let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap();
	let track = broadcast
		.create_track(
			broadcast.unique_name(".avc1"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut guard = catalog.modify().unwrap();
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x64,
			constraints: 0,
			level: 0x1f,
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description = Some(avcc);
		guard.video.renditions.insert(name, cfg);
		guard.ext.mpegts.si.entry(pid).or_default().insert(
			table_id,
			tscat::SiEntry {
				track: si_name,
				interval: Some(interval),
				..Default::default()
			},
		);
	}

	let producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let exporter = Export::with_ts(crate::source::announced(&consumer), crate::catalog::CatalogFormat::Hang)
		.await
		.unwrap();
	SiCadenceRig {
		started: false,
		producer,
		si_track,
		exporter,
		catalog,
	}
}

impl SiCadenceRig {
	async fn media_frames(&mut self, millis: u64) -> Vec<Frame> {
		// Reordered pictures stay in one group; a newer group going backwards is a rewind.
		let mut nal = vec![if self.started { 0x41u8 } else { 0x65u8 }];
		nal.extend(std::iter::repeat_n(0xAB, 64));
		self.producer
			.write(Frame {
				timestamp: Timestamp::from_millis(millis).unwrap(),
				duration: None,
				payload: length_prefixed(&[&nal]),
				keyframe: !self.started,
			})
			.unwrap();
		self.started = true;

		let frames = drain_frames(&mut self.exporter).await;
		for frame in &frames {
			assert_packet_aligned(&frame.payload);
		}
		frames
	}

	async fn media(&mut self, millis: u64, pid: u16) -> usize {
		count_pid(&self.media_frames(millis).await, pid)
	}
}

/// #2934: a revised SI snapshot goes out on the revision floor instead of waiting
/// out the repetition interval (here the floor has long elapsed, so it rides the
/// very next frame). For a clock table (TDT/TOT) the old interval-grid hold
/// delivered the asserted time up to a whole 30 s slot late, and a source ticking
/// slower than the grid had its stale value re-sent, stepping receivers backwards.
#[tokio::test(start_paused = true)]
async fn si_revision_does_not_wait_for_the_interval() {
	let tick = |mjd: u8| make_short_section(0x70, &[0xc0, mjd, 0x12, 0x34, 0x56]);
	let mut rig = si_cadence_rig(0x0014, 0x70, Duration::from_secs(30)).await;

	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(tick(1))).unwrap();
	assert_eq!(rig.media(0, 0x0014).await, 0, "the first span stays buffered");
	assert_eq!(
		rig.media(1_000, 0x0014).await,
		1,
		"the next span closes the lead emission"
	);
	assert_eq!(rig.media(2_000, 0x0014).await, 0, "unchanged: still held");

	// The clock ticks. Nothing about the 30 s interval has elapsed, but the value
	// changed: it must go out with the very next frame.
	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(tick(2))).unwrap();
	assert_eq!(rig.media(3_000, 0x0014).await, 0, "the revision enters the open span");
	assert_eq!(rig.media(4_000, 0x0014).await, 1, "the next span closes the revision");
}

/// #3948: an unchanged snapshot repeats on the absolute media-time grid, like the
/// PSI, so exporters that started at different moments repeat it at the same
/// instants. A revision still goes out on its floor between grid slots (#2934).
#[tokio::test(start_paused = true)]
async fn si_repeats_ride_the_media_grid() {
	let sdt_v1 = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8]);
	let sdt_v2 = make_long_section(0x42, 1, 1, 0, 0, &[0xbb; 8]);
	let mut rig = si_cadence_rig(0x0011, 0x42, Duration::from_secs(2)).await;

	// The lead emission lands wherever this exporter joined.
	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(sdt_v1)).unwrap();
	assert_eq!(rig.media(1_500, 0x0011).await, 0, "the first span stays buffered");
	assert_eq!(
		rig.media(1_900, 0x0011).await,
		1,
		"the next span closes the lead emission"
	);
	// The 2s boundary is 0.5s after the lead: the grid, not a floor from the join.
	assert_eq!(rig.media(2_000, 0x0011).await, 0, "the repeat enters the open span");
	assert_eq!(rig.media(2_500, 0x0011).await, 1, "the next span closes the repeat");

	// A revision lands mid-slot: it waits only for the 1s revision floor.
	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(sdt_v2)).unwrap();
	assert_eq!(rig.media(2_900, 0x0011).await, 0, "within the floor of the repeat");
	assert_eq!(
		rig.media(3_000, 0x0011).await,
		0,
		"the revision enters the span once its floor lapses"
	);
	assert_eq!(rig.media(3_500, 0x0011).await, 1, "the next span closes the revision");

	// Repeats stay on the grid after it.
	assert_eq!(rig.media(4_000, 0x0011).await, 0, "the 4s repeat enters the open span");
	assert_eq!(rig.media(5_000, 0x0011).await, 1, "the next span closes it");
	assert_eq!(rig.media(5_900, 0x0011).await, 0, "nothing between grid slots");
	assert_eq!(rig.media(6_000, 0x0011).await, 0, "the 6s repeat enters the open span");
	assert_eq!(rig.media(6_100, 0x0011).await, 1, "the next span closes it");
}

/// A clock table (TDT/TOT) never repeats an unchanged snapshot: a repeat re-asserts
/// a time already sent and steps a receiver backwards (#2934). Every new value is a
/// revision, so it still goes out promptly.
#[tokio::test(start_paused = true)]
async fn si_clock_tables_do_not_repeat_unchanged() {
	let tick = |mjd: u8| make_short_section(0x70, &[0xc0, mjd, 0x12, 0x34, 0x56]);
	let mut rig = si_cadence_rig(0x0014, 0x70, Duration::from_secs(2)).await;

	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(tick(1))).unwrap();
	assert_eq!(rig.media(0, 0x0014).await, 0, "the first span stays buffered");
	assert_eq!(
		rig.media(1_000, 0x0014).await,
		1,
		"the next span closes the lead emission"
	);
	let mut repeats = 0;
	for millis in (2_000..=9_000).step_by(1_000) {
		repeats += rig.media(millis, 0x0014).await;
	}
	assert_eq!(repeats, 0, "an unchanged time is never re-sent");

	rig.si_track.write_frame(Timestamp::ZERO, Bytes::from(tick(2))).unwrap();
	assert_eq!(rig.media(10_000, 0x0014).await, 0, "the new time enters the open span");
	assert_eq!(rig.media(11_000, 0x0014).await, 1, "the next span closes it");
}

/// A reordered (B-frame) timestamp below the emission anchor earns no credit:
/// `si_due` saturates elapsed time, so a revision arriving there is deferred, the
/// anchor never moves backwards, and neither revisions nor repeats can fire early
/// off the reorder span.
#[tokio::test(start_paused = true)]
async fn si_reordered_frames_earn_no_emission_credit() {
	let section = |version: u8| make_long_section(0x42, 1, version, 0, 0, &[version; 8]);
	let mut rig = si_cadence_rig(0x0011, 0x42, Duration::from_secs(10)).await;

	rig.si_track
		.write_frame(Timestamp::ZERO, Bytes::from(section(0)))
		.unwrap();
	assert_eq!(rig.media(1_000, 0x0011).await, 0, "the first span stays buffered");

	rig.si_track
		.write_frame(Timestamp::ZERO, Bytes::from(section(1)))
		.unwrap();
	assert_eq!(rig.media(2_500, 0x0011).await, 1, "the lead emission closes");

	// The next revision arrives on a frame stepping back behind the 2.5s anchor,
	// like a B-frame emitted in decode order: no elapsed time, no emission.
	rig.si_track
		.write_frame(Timestamp::ZERO, Bytes::from(section(2)))
		.unwrap();
	assert_eq!(rig.media(2_400, 0x0011).await, 0, "a reordered frame earns no credit");

	// The floor measures from the 2.5s anchor: not due at 3.4s, due at 3.5s.
	assert_eq!(rig.media(3_400, 0x0011).await, 1, "only the prior revision closes");
	assert_eq!(
		rig.media(3_500, 0x0011).await,
		0,
		"the deferred revision enters the span at the floor"
	);
	assert_eq!(rig.media(3_600, 0x0011).await, 1, "the next span closes the revision");
	assert_eq!(rig.exporter.discontinuity(), 0, "B-frame reordering is not a rewind");
}

/// The emission anchor never moves backwards, even where a zero-interval entry
/// emits on a reordered (B-frame) timestamp below it: a catalog update can raise
/// the interval afterwards, and a regressed anchor would then sit in an earlier
/// grid slot and repeat early. (Non-zero intervals cannot regress the anchor on
/// their own, since only a timestamp past it is due; the zero-to-nonzero
/// transition is the one reachable path.)
#[tokio::test(start_paused = true)]
async fn si_anchor_survives_a_zero_interval_reorder() {
	let section = |version: u8| make_long_section(0x42, 1, version, 0, 0, &[version; 8]);
	// Zero interval: the table rides every frame, whatever its timestamp.
	let mut rig = si_cadence_rig(0x0011, 0x42, Duration::ZERO).await;

	rig.si_track
		.write_frame(Timestamp::ZERO, Bytes::from(section(0)))
		.unwrap();
	assert_eq!(rig.media(1_000, 0x0011).await, 0, "the first span stays buffered");
	assert_eq!(rig.media(3_000, 0x0011).await, 1, "the anchor advances to 3s");
	// A reordered frame steps back behind the anchor, into the previous 1s slot;
	// zero interval still emits.
	assert_eq!(rig.media(2_900, 0x0011).await, 0, "the reordered span stays open");

	// The catalog raises the interval to 1s. The grid slot must be the 3s anchor's,
	// not the reordered 2.9s emission's.
	rig.catalog
		.modify()
		.unwrap()
		.ext
		.mpegts
		.si
		.get_mut(&0x0011)
		.unwrap()
		.get_mut(&0x42)
		.unwrap()
		.interval = Some(Duration::from_secs(1));
	assert_eq!(
		rig.media(3_400, 0x0011).await,
		2,
		"both zero-interval frames close together"
	);
	assert_eq!(rig.media(3_900, 0x0011).await, 0, "still in the anchor's slot");
	assert_eq!(rig.media(4_000, 0x0011).await, 0, "the due table enters the open span");
	assert_eq!(rig.media(4_100, 0x0011).await, 1, "the next span closes the due table");
	assert_eq!(rig.exporter.discontinuity(), 0, "B-frame reordering is not a rewind");
}

/// A publisher revising its snapshot before every frame must not drive the mux at
/// that rate: revisions coalesce onto the revision floor, newest snapshot wins.
/// The import side debounces its own cuts, but export consumes any catalog-named
/// snapshot track, so the bound has to hold here too.
#[tokio::test(start_paused = true)]
async fn si_rapid_revisions_are_rate_bounded() {
	let section = |version: u8| make_long_section(0x42, 1, version, 0, 0, &[version; 8]);
	let mut rig = si_cadence_rig(0x0011, 0x42, Duration::from_secs(30)).await;

	let mut emitted = 0;
	for i in 0..13u8 {
		rig.si_track
			.write_frame(Timestamp::ZERO, Bytes::from(section(i)))
			.unwrap();
		let frames = rig.media_frames(u64::from(i) * 250).await;
		let count = count_pid(&frames, 0x0011);
		emitted += count;
		if i == 5 {
			// The 1s floor lapses here; the emission carries the newest revision,
			// not the revisions that arrived after it while its span was buffered.
			assert_eq!(count, 1, "the next span closes the 1s emission");
			let needle = section(4);
			assert!(
				frames
					.iter()
					.any(|f| f.payload.windows(needle.len()).any(|w| w == needle)),
				"the floored emission carries the newest revision"
			);
		}
	}
	emitted += rig.media(3_250, 0x0011).await;
	assert_eq!(emitted, 4, "lead plus one per second, not one per revision");
}

/// Two captures overlapping on one broadcast (a supervisor restarting its
/// importer before the old one is dropped) contend for the same `(PID, table_id)`
/// key: the newer one wins the catalog mapping under a fallback track name, and
/// the older one's teardown must not strip it, since the survivor never
/// re-advertises.
#[tokio::test(start_paused = true)]
async fn overlapping_capture_teardown_keeps_the_survivors_mapping() {
	let sdt = make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 4]);
	let mut input = si_packet(0x0011, &sdt);
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt, 1));

	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let _consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut old = crate::container::ts::Import::new(broadcast.clone(), catalog.reserve());
	old.decode(&BytesMut::from(&input[..])).unwrap();

	// The replacement importer captures the same table; its deterministic track
	// name is taken, so it advertises under a fallback name, overwriting the key.
	let mut new = crate::container::ts::Import::new(broadcast, catalog.reserve());
	new.decode(&BytesMut::from(&input[..])).unwrap();
	let survivor = catalog.snapshot().ext.mpegts.si[&0x0011][&0x42].track.clone();
	assert_ne!(survivor, "0x0011-0x42.si", "the replacement fell back to a unique name");

	drop(old);
	assert_eq!(
		catalog.snapshot().ext.mpegts.si[&0x0011][&0x42].track,
		survivor,
		"the old capture's teardown left the survivor's mapping in place"
	);
}

/// A contiguous commit replaces even a same-version active generation: versions
/// are five bits, so a reception gap can bring the same value back with fewer
/// sections, and merging would resurrect the removed one forever.
#[tokio::test(start_paused = true)]
async fn contiguous_same_version_commit_replaces_stale_sections() {
	let sdt = |number: u8, last: u8, fill: u8| make_long_section(0x42, 1, 0, number, last, &[fill; 4]);

	// Three sections at v0, committed contiguously.
	let mut input = si_packet(0x0011, &sdt(0, 2, 0xa0));
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt(1, 2, 0xa1), 1));
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt(2, 2, 0xa2), 2));
	// After a gap the table comes back at v0 again (wrapped), now two sections.
	let b0 = sdt(0, 1, 0xb0);
	let b1 = sdt(1, 1, 0xb1);
	input.extend_from_slice(&si_packet_cc(0x0011, &b0, 3));
	input.extend_from_slice(&si_packet_cc(0x0011, &b1, 4));
	let mut rig = si_rig();
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();
	rig.import.finish().unwrap();

	let si = rig.catalog.snapshot().ext.mpegts.si.clone();
	assert_eq!(
		read_si_sections(&rig.consumer, &si[&0x0011][&0x42].track).await,
		vec![Bytes::from(b0), Bytes::from(b1)],
		"the complete new generation retired the stale third section"
	);
}

/// The cut debounce runs on the host clock, so a revision publishes even when no
/// media ever advances a PTS (an audio-only or SI-only input). This test spends
/// real wall time on the debounce window; media timestamps stay pinned at zero
/// throughout, which is exactly the case a media-clock debounce wedges on.
#[tokio::test(start_paused = true)]
async fn debounce_opens_without_a_media_clock() {
	let sdt = |version: u8, fill: u8| make_long_section(0x42, 1, version, 0, 0, &[fill; 4]);
	let mut rig = si_rig();

	let mut input = si_packet(0x0011, &sdt(0, 0xaa));
	input.extend_from_slice(&si_packet_cc(0x0011, &sdt(0, 0xaa), 1));
	rig.import.decode(&BytesMut::from(&input[..])).unwrap();

	let name = rig.catalog.snapshot().ext.mpegts.si[&0x0011][&0x42].track.clone();
	let track = rig.consumer.track(&name).unwrap().subscribe(None).await.unwrap();
	assert_eq!(track.latest(), Some(0), "the first snapshot cut immediately");

	// A revision inside the window is coalesced...
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0011, &sdt(1, 0xbb), 2)[..]))
		.unwrap();
	assert_eq!(track.latest(), Some(0), "a revision inside the window is held");

	// ...and publishes once the window passes in *real* time, no finish, no PTS.
	std::thread::sleep(Duration::from_millis(1200));
	rig.import
		.decode(&BytesMut::from(&si_packet_cc(0x0011, &sdt(1, 0xbb), 3)[..]))
		.unwrap();
	assert_eq!(track.latest(), Some(1), "the held revision cut after the window");
}

/// Every PCR packet in transport order: its packet index in the stream, its value
/// (90 kHz), and the media timestamp of the frame carrying it.
fn collect_pcrs(frames: &[Frame]) -> Vec<(usize, u64, u128)> {
	let mut at = 0;
	let mut pcrs = Vec::new();
	for frame in frames {
		for packet in frame.payload.chunks(188) {
			// adaptation-field-only packets (adaptation_field_control == 0b10) are the clock.
			if packet[3] & 0x30 == 0x20 && packet[5] & 0x10 != 0 {
				let base = (u64::from(packet[6]) << 25)
					| (u64::from(packet[7]) << 17)
					| (u64::from(packet[8]) << 9)
					| (u64::from(packet[9]) << 1)
					| u64::from(packet[10] >> 7);
				pcrs.push((at, base, frame.timestamp.as_micros()));
			}
			at += 1;
		}
	}
	pcrs
}

/// A 4 s CBR-ish single-rendition H.264 feed at 25 fps, one second per group: the
/// shape of a broadcast contribution capture, and the one #3334 measured.
async fn export_cbr_video() -> Vec<Frame> {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".h264"),
			hang::container::track_info(hang::catalog::PRIORITY.video),
		)
		.unwrap();
	{
		let mut cfg = VideoConfig::new(H264 {
			profile: 0x42,
			constraints: 0xc0,
			level: 0x1f,
			inline: true,
		});
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.video
			.renditions
			.insert(track.name().to_string(), cfg);
	}
	let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	for i in 0..100u64 {
		let keyframe = i % 25 == 0;
		let mut nal = vec![if keyframe { 0x65u8 } else { 0x41 }];
		nal.extend(std::iter::repeat_n(0xAB, 7_000));
		if keyframe && i > 0 {
			video.cut(None).unwrap();
		}
		video
			.write(Frame {
				timestamp: Timestamp::from_micros(i * 40_000).unwrap(),
				duration: None,
				payload: if keyframe {
					annexb(&[SPS, PPS, &nal])
				} else {
					annexb(&[&nal])
				},
				keyframe,
			})
			.unwrap();
	}
	video.finish().unwrap();

	let mut exporter = Export::new(crate::source::announced(&consumer))
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	drain_frames(&mut exporter).await
}

/// #3334: the clock a receiver recovers from *byte position* has to agree with the
/// values, because a byte stream carries no other timing. Emitting each media frame
/// whole put every PCR between two frames instead of among the bytes it labels, so
/// the grid #2967 made exact could not be read off the wire: consecutive clock
/// packets sat one packet apart with the media they described heaped between the
/// clusters, and a downstream stage re-deriving PCR from position regenerated the
/// original clustered distribution.
#[tokio::test(start_paused = true)]
async fn pcr_rides_the_bytes_it_labels() {
	let frames = export_cbr_video().await;
	let pcrs = collect_pcrs(&frames);
	assert!(pcrs.len() > 100, "expected the full feed, got {} PCRs", pcrs.len());

	// Every step is one grid slot, so every gap should carry the same span of media
	// and therefore a comparable number of packets.
	let step = PCR_INTERVAL.as_micros() as u64 * 90 / 1_000;
	for (i, w) in pcrs.windows(2).enumerate() {
		assert_eq!(w[1].1.wrapping_sub(w[0].1) & ((1 << 33) - 1), step, "value step at {i}");
	}

	let gaps: Vec<usize> = pcrs.windows(2).map(|w| w[1].0 - w[0].0).collect();
	assert_eq!(
		gaps.iter().filter(|&&gap| gap == 1).count(),
		0,
		"no clock packet may sit adjacent to the previous one: {gaps:?}"
	);
	let mut sorted = gaps.clone();
	sorted.sort_unstable();
	let median = sorted[sorted.len() / 2];
	// The leading group carries the parameter sets and program tables on top of its
	// media, so allow generous headroom; what this rules out is the bimodal
	// distribution (one packet, then hundreds) that made the clock unrecoverable.
	assert!(
		sorted[0] * 3 >= median && sorted[sorted.len() - 1] <= median * 3,
		"packet gaps must track the interval the values assert, got {sorted:?}"
	);
}

/// The other half of #3334: a PCR's *release* has to track the interval its own
/// value asserts. The exporter only stamps; the caller paces on the stamps (see
/// [`moq_mux::Pacer`]), so the property here is that consecutive clock packets are
/// stamped exactly one grid interval apart. Frames are emitted whole and a slot's
/// bytes are only all in hand once media past it has arrived, so a stamp that
/// tracked frame arrival was already in the past and its sleep was a no-op.
#[tokio::test(start_paused = true)]
async fn pcr_stamps_step_by_the_grid() {
	let frames = export_cbr_video().await;
	let pcrs = collect_pcrs(&frames);

	let steps: Vec<i128> = pcrs.windows(2).map(|w| w[1].2 as i128 - w[0].2 as i128).collect();
	let interval = PCR_INTERVAL.as_micros() as i128;
	assert!(
		steps.iter().all(|&step| step == interval),
		"every clock packet must be stamped one grid interval past the last: {steps:?}"
	);
}

/// The same positional property on a real reordered (B-frame) capture with a second
/// rendition, where the exporter has no uniform cadence to lean on: frames arrive in
/// decode order, and the two tracks advance the media clock at different rates. The
/// clock still lands among the bytes it labels rather than clustering, though the
/// gaps are no longer uniform: a span is however long it took the next timestamp to
/// arrive, which for interleaved tracks is nothing like the media a frame's bytes
/// represent. Evening that out needs the muxer to hold a byte buffer and drain it at
/// a measured rate, which this does not do.
#[tokio::test(start_paused = true)]
async fn pcr_stays_among_the_bytes_across_reordered_tracks() {
	let data = include_bytes!("test_data/scte35/kyrion_dirtystart.ts");
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(&BytesMut::from(&data[..])).unwrap();
	import.finish().unwrap();

	let mut exporter = Export::new(crate::source::announced(&consumer))
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	let frames = drain_frames(&mut exporter).await;
	let pcrs = collect_pcrs(&frames);
	assert!(pcrs.len() > 50, "expected the full feed, got {} PCRs", pcrs.len());

	let gaps: Vec<usize> = pcrs.windows(2).map(|w| w[1].0 - w[0].0).collect();
	assert_eq!(
		gaps.iter().filter(|&&gap| gap == 1).count(),
		0,
		"no clock packet may sit adjacent to the previous one: {gaps:?}"
	);

	// Stamps still step by exactly one slot, apart from the first interval, which
	// covers the leading span rather than a whole slot.
	let interval = PCR_INTERVAL.as_micros() as i128;
	let off = pcrs
		.windows(2)
		.filter(|w| w[1].2 as i128 - w[0].2 as i128 != interval)
		.count();
	assert!(off <= 1, "{off} clock packets are stamped off the grid");
}

/// A clock packet carries no payload, so ISO 13818-1 2.4.3.3 says it must repeat
/// the continuity counter of whatever preceded it on its PID rather than advance
/// it. The clock rides a PID that also carries media, and slicing on the grid puts
/// clock packets *inside* a frame's packet run, whose counters were assigned when
/// the frame was muxed rather than when the bytes go out. Numbering the clock
/// packet from the counter's current value there lands it a whole frame ahead, and
/// an analyzer reports a discontinuity on it and another on the payload packet
/// after it.
#[tokio::test(start_paused = true)]
async fn payload_less_clock_packets_repeat_the_counter() {
	let frames = export_cbr_video().await;
	let ts: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
	assert_packet_aligned(&ts);

	let mut last: std::collections::HashMap<u16, u8> = std::collections::HashMap::new();
	let mut advanced = 0;
	let mut discontinuities = 0;
	for (i, packet) in ts.chunks(188).enumerate() {
		let pid = u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]);
		let cc = packet[3] & 0x0f;
		let payload = packet[3] & 0x10 != 0;
		if let Some(&prev) = last.get(&pid) {
			if payload && cc != (prev + 1) & 0x0f {
				discontinuities += 1;
				eprintln!("discontinuity at packet {i} on pid {pid}: {prev} -> {cc}");
			} else if !payload && cc != prev {
				advanced += 1;
				eprintln!("payload-less packet {i} on pid {pid} advanced: {prev} -> {cc}");
			}
		}
		last.insert(pid, cc);
	}
	assert_eq!(advanced, 0, "payload-less packets must repeat the counter");
	assert_eq!(discontinuities, 0, "the counter must be continuous on every PID");
}

/// What a constant-rate receiver would measure of `ts`: the packets between its
/// first and last clock packet over the PCR ticks they span, how much of that was
/// null stuffing, and the widest PCR gap.
struct Clocked {
	packets: usize,
	ticks: u64,
	nulls: usize,
	max_gap: u64,
}

impl Clocked {
	fn of(ts: &[u8]) -> Self {
		const WRAP: u64 = (1 << 33) * 300;
		let mut first = None;
		let mut last = None;
		let mut nulls = 0;
		let mut max_gap = 0;
		let mut span = 0;
		for (index, packet) in ts.chunks(188).enumerate() {
			let pid = (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]);
			if pid == 0x1fff {
				nulls += 1;
			}
			if packet[3] & 0x20 != 0 && packet[4] >= 7 && packet[5] & 0x10 != 0 {
				let base = (u64::from(packet[6]) << 25)
					| (u64::from(packet[7]) << 17)
					| (u64::from(packet[8]) << 9)
					| (u64::from(packet[9]) << 1)
					| (u64::from(packet[10]) >> 7);
				let ticks = base * 300 + ((u64::from(packet[10] & 1) << 8) | u64::from(packet[11]));
				// The exporter backs the clock off through the 33-bit wrap at the start.
				if let Some(prev) = last.replace((index, ticks)) {
					let gap = (ticks + WRAP - prev.1) % WRAP;
					max_gap = max_gap.max(gap);
					span += gap;
				}
				first.get_or_insert(index);
			}
		}
		let (first, last) = (first.expect("no PCR"), last.expect("no PCR"));
		Self {
			packets: last.0 - first,
			ticks: span,
			nulls,
			max_gap,
		}
	}

	/// The multiplex rate in bits per second.
	fn rate(&self) -> u64 {
		(self.packets as u128 * 188 * 8 * 27_000_000 / u128::from(self.ticks)) as u64
	}
}

/// Import a fixture into a broadcast whose catalog carries the `mpegts` section,
/// then export it. The producers live until the export drains.
async fn export_fixture(data: &[u8], mux_rate: Option<u64>) -> BytesMut {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();
	let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve());
	import.decode(data).unwrap();
	import.finish().unwrap();

	let mut export = export_of(&consumer).await;
	if let Some(mux_rate) = mux_rate {
		export = export.with_mux_rate(mux_rate);
	}
	let ts = drain_with(export).await;
	assert_packet_aligned(&ts);
	ts
}

/// `bbb_cbr.ts` is a 400 kb/s multiplex carrying 167 kb/s of media, which import
/// records as such, so export pads the media back to the recorded rate.
#[tokio::test(start_paused = true)]
async fn export_pads_to_the_recorded_mux_rate() {
	let ts = export_fixture(include_bytes!("test_data/bbb_cbr.ts"), None).await;
	let clocked = Clocked::of(&ts);
	assert!(clocked.nulls > 0, "no null stuffing was emitted");
	let rate = clocked.rate();
	assert!(
		rate.abs_diff(400_000) * 100 <= 400_000,
		"output runs at {rate} b/s, not the recorded 400 kb/s"
	);
	assert!(
		clocked.max_gap <= 40 * 27_000,
		"PCR gap of {} ticks exceeds 40 ms",
		clocked.max_gap
	);
}

/// The builder override wins over the catalog's recorded rate.
#[tokio::test(start_paused = true)]
async fn export_mux_rate_override_beats_the_catalog() {
	let ts = export_fixture(include_bytes!("test_data/bbb_cbr.ts"), Some(1_000_000)).await;
	let rate = Clocked::of(&ts).rate();
	assert!(
		rate.abs_diff(1_000_000) * 100 <= 1_000_000,
		"output runs at {rate} b/s, not the 1 Mb/s override"
	);
}

/// Zero and absurd override rates are refused, leaving the output unpadded: zero
/// pads nothing, and an unbounded rate would allocate unbounded nulls per slot.
/// An explicit override wins even when refused, so the catalog rate is not used.
#[tokio::test(start_paused = true)]
async fn export_mux_rate_override_is_bounded() {
	for rate in [0, i64::MAX as u64] {
		let ts = export_fixture(include_bytes!("test_data/bbb_cbr.ts"), Some(rate)).await;
		assert_eq!(
			Clocked::of(&ts).nulls,
			0,
			"null packets in an export with a refused override rate {rate}"
		);
	}
}

/// An absurd catalog multiplex rate is refused, not padded to: the catalog is
/// untrusted input, and padding to it would allocate unbounded nulls per slot.
#[tokio::test(start_paused = true)]
async fn export_refuses_an_absurd_catalog_mux_rate() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(
		&mut broadcast,
		crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::<tscat::Ext>::default()),
	)
	.unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	catalog.modify().unwrap().audio.renditions.insert(name.clone(), {
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		cfg
	});
	catalog.modify().unwrap().ext.mpegts.mux_rate = Some(i64::MAX as u64);

	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	// Ten seconds of small frames: the media alone never approaches the refused rate.
	for i in 0..500u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i * 20_000).unwrap(),
				duration: None,
				payload: Bytes::from(vec![i as u8; 16]),
				keyframe: i % 50 == 0,
			})
			.unwrap();
	}
	producer.finish().unwrap();

	let ts = drain_with(export_of(&consumer).await).await;
	assert_packet_aligned(&ts);
	assert_eq!(
		Clocked::of(&ts).nulls,
		0,
		"null packets in an export with a refused catalog mux rate"
	);
}

/// A VBR source records no rate, and without one nothing is padded.
#[tokio::test(start_paused = true)]
async fn export_without_a_mux_rate_is_unpadded() {
	let ts = export_fixture(include_bytes!("test_data/scte35/bbb5s.ts"), None).await;
	assert_eq!(Clocked::of(&ts).nulls, 0, "null packets in an unpadded export");
}

/// 1 Mb/s is 16.62 packets per 25 ms slot. The count between the first and last
/// clock packet must be the floor of the exact allowance, which only holds when
/// the fractional remainder carries across slots instead of rounding each one.
/// Also the override supplying a rate to a catalog that has none.
#[tokio::test(start_paused = true)]
async fn export_stuffing_keeps_the_fractional_remainder() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track(
			broadcast.unique_name(".aac"),
			hang::container::track_info(hang::catalog::PRIORITY.audio),
		)
		.unwrap();
	let name = track.name().to_string();
	{
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog.modify().unwrap().audio.renditions.insert(name.clone(), cfg);
	}
	let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	// Ten seconds of small frames: a few kb/s of media under a 1 Mb/s rate.
	for i in 0..500u64 {
		producer
			.write(Frame {
				timestamp: Timestamp::from_micros(i * 20_000).unwrap(),
				duration: None,
				payload: Bytes::from(vec![i as u8; 16]),
				keyframe: i % 50 == 0,
			})
			.unwrap();
	}
	producer.finish().unwrap();

	let export = Export::new(crate::source::announced(&consumer))
		.await
		.unwrap()
		.with_mux_rate(1_000_000);
	let ts = drain_with(export).await;
	assert_packet_aligned(&ts);

	let clocked = Clocked::of(&ts);
	let slots = clocked.ticks / (PCR_INTERVAL.as_nanos() as u64 * 27 / 1000);
	assert!(slots > 300, "expected a long run of slots, got {slots}");
	// Bits allowed over the run, in whole packets.
	let expected = (slots as u128 * 1_000_000 * PCR_INTERVAL.as_nanos() / (1_000_000_000 * 188 * 8)) as u64;
	assert_eq!(
		clocked.packets as u64, expected,
		"{slots} slots at 1 Mb/s carry {expected} packets"
	);
	assert!(clocked.nulls > 0, "no null stuffing was emitted");
}

// The decode clock follows the stream's reordering: its reserve comes from the catalog `jitter`,
// else the depth the SPS declares, else the reordering muxed so far, and it keeps following all
// three after the program tables are written.

/// Display offsets within a group, in decode order: a closed GOP with one B-frame per reference.
/// A B-frame lands one frame below the high-water mark.
const IPB: [u64; GOP as usize] = [0, 2, 1, 4, 3, 6, 5, 8, 7, 10, 9, 12, 11, 14, 13];
/// A three-B pyramid: the outer B-frames sit under the middle one, so one lands three frames
/// below the high-water mark while the SPS can declare a depth of two.
const PYRAMID: [u64; GOP as usize] = [0, 4, 2, 1, 3, 8, 6, 5, 7, 12, 10, 9, 11, 14, 13];
/// Where the [`Reordered`] timeline starts, in frames: whole groups, and far enough in that
/// the PCR's back-off never wraps below zero.
const REORDERED_BASE: u64 = 20 * GOP;
/// One 25 fps frame in 90 kHz ticks.
const FRAME_TICKS: u64 = 3_600;

/// The start of video frame `frame` of a [`Reordered`] broadcast, in display order.
fn reordered_at(frame: u64) -> Timestamp {
	Timestamp::from_micros((REORDERED_BASE + frame) * VIDEO_US).unwrap()
}

/// A reordered H.264 rendition and an AAC one, on the late-join fixture's cadence.
struct Reordered {
	_broadcast: moq_net::broadcast::Producer,
	catalog: crate::catalog::Producer,
	video: Producer<HangContainer>,
	audio: Producer<HangContainer>,
	source: crate::Source,
	name: String,
	/// Groups laid out as [`PYRAMID`]; every other group is [`IPB`].
	pyramid: &'static [u64],
	audio_index: u64,
}

impl Reordered {
	fn new(sps: &'static [u8], pps: &'static [u8], pyramid: &'static [u64]) -> Self {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let consumer = broadcast.consume();
		let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

		let track = broadcast
			.create_track(
				broadcast.unique_name(".avc1"),
				hang::container::track_info(hang::catalog::PRIORITY.video),
			)
			.unwrap();
		let name = track.name().to_string();
		let mut cfg = VideoConfig::new(H264 {
			profile: sps[1],
			constraints: sps[2],
			level: sps[3],
			inline: false,
		});
		cfg.container = Container::Legacy;
		cfg.description =
			Some(crate::codec::h264::build_avcc(&[Bytes::from_static(sps)], &[Bytes::from_static(pps)]).unwrap());
		catalog.modify().unwrap().video.renditions.insert(name.clone(), cfg);
		let video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Video));

		let track = broadcast
			.create_track(
				broadcast.unique_name(".aac"),
				hang::container::track_info(hang::catalog::PRIORITY.audio),
			)
			.unwrap();
		let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
		cfg.container = Container::Legacy;
		catalog
			.modify()
			.unwrap()
			.audio
			.renditions
			.insert(track.name().to_string(), cfg);
		let audio = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Audio));

		Self {
			source: crate::source::announced(&consumer),
			_broadcast: broadcast,
			catalog,
			video,
			audio,
			name,
			pyramid,
			audio_index: 0,
		}
	}

	/// An exporter with its subscriptions resolved, so later polls need no runtime.
	async fn export(&self, max_age: Duration) -> Export {
		let mut export = Export::new(self.source.clone()).await.unwrap().with_max_age(max_age);
		assert!(drain_frames(&mut export).await.is_empty());
		export
	}

	/// Write the video frame `tick`-th in decode order.
	fn video(&mut self, tick: u64) {
		let (group, at) = (tick / GOP, (tick % GOP) as usize);
		let layout = if self.pyramid.contains(&group) { &PYRAMID } else { &IPB };
		let keyframe = at == 0;
		let slice = if keyframe {
			vec![0x65u8; 3_000]
		} else {
			vec![0x41u8; 400]
		};
		self.video
			.write(Frame {
				timestamp: reordered_at(group * GOP + layout[at]),
				duration: None,
				payload: length_prefixed(&[&slice]),
				keyframe,
			})
			.unwrap();
	}

	/// Write the next audio frame if it starts before video frame `frame`.
	fn audio_before(&mut self, frame: u64) -> bool {
		let index = self.audio_index;
		let timestamp = Timestamp::from_micros(REORDERED_BASE * VIDEO_US + index * AUDIO_US).unwrap();
		if timestamp >= reordered_at(frame) {
			return false;
		}
		self.audio
			.write(Frame {
				timestamp,
				duration: None,
				payload: Bytes::from_iter((0..180u16).map(|i| (i ^ index as u16) as u8)),
				keyframe: index.is_multiple_of(AUDIO_GROUP),
			})
			.unwrap();
		self.audio_index += 1;
		true
	}

	fn publish_jitter(&mut self, jitter: Duration) {
		let mut catalog = self.catalog.modify().unwrap();
		catalog.video.renditions.get_mut(&self.name).unwrap().jitter = Some(jitter);
	}

	fn finish(&mut self) {
		self.video.finish().unwrap();
		self.audio.finish().unwrap();
	}
}

/// Export a [`Reordered`] broadcast twice, like [`export_twice`]: the second exporter joins at
/// [`JOIN`], and both are drained after every tick. `jitter` is published at its tick when given.
async fn export_reordered(
	sps: &'static [u8],
	pps: &'static [u8],
	pyramid: &'static [u64],
	jitter: Option<(u64, Duration)>,
) -> (Vec<Frame>, Vec<Frame>) {
	let mut rig = Reordered::new(sps, pps, pyramid);
	let mut a = Export::new(rig.source.clone()).await.unwrap();
	let mut b = None;
	let (mut out_a, mut out_b) = (Vec::new(), Vec::new());
	for tick in 0..TICKS {
		if let Some((at, jitter)) = jitter
			&& at == tick
		{
			rig.publish_jitter(jitter);
		}
		rig.video(tick);
		while rig.audio_before(tick + 1) {}

		out_a.extend(drain_frames(&mut a).await);
		if let Some(b) = b.as_mut() {
			out_b.extend(drain_frames(b).await);
		}
		if tick + 1 == JOIN {
			b = Some(Export::new(rig.source.clone()).await.unwrap());
		}
	}
	rig.finish();
	let mut b = b.unwrap();
	out_a.extend(drain_frames(&mut a).await);
	out_b.extend(drain_frames(&mut b).await);
	(out_a, out_b)
}

/// `(PTS, DTS)` of every video PES start in the frames stamped within `range`, in transport
/// order. The whole stream is read so the program tables resolve every PID.
fn video_timing(frames: &[Frame], range: impl std::ops::RangeBounds<Timestamp>) -> Vec<(u64, Option<u64>)> {
	let stamps: Vec<Timestamp> = frames
		.iter()
		.flat_map(|f| std::iter::repeat_n(f.timestamp, f.payload.len() / 188))
		.collect();
	let bytes: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let mut reader = TsPacketReader::new(Cursor::new(bytes));
	let mut out = Vec::new();
	let mut index = 0;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		let stamp = stamps[index];
		index += 1;
		if !range.contains(&stamp) {
			continue;
		}
		if let Some(TsPayload::PesStart(pes)) = packet.payload
			&& (mpeg2ts::es::StreamId::VIDEO_MIN..=mpeg2ts::es::StreamId::VIDEO_MAX)
				.contains(&pes.header.stream_id.as_u8())
		{
			let pts = pes.header.pts.expect("video PES carried no PTS").as_u64();
			out.push((pts, pes.header.dts.map(|t| t.as_u64())));
		}
	}
	out
}

/// Every video frame decodes at or before it is presented.
fn assert_decodes_before_presenting(timing: &[(u64, Option<u64>)]) {
	assert!(timing.len() > 20, "too few video frames to judge: {}", timing.len());
	for (i, &(pts, dts)) in timing.iter().enumerate() {
		let dts = dts.unwrap_or(pts);
		assert!(dts <= pts, "frame {i} decodes at {dts}, after it is presented at {pts}");
	}
}

/// In transport order, every PES unit decodes at or after the last PCR preceding it, and the
/// clock only steps back where a PCR signals a new time base.
fn assert_decodes_after_the_clock(frames: &[Frame]) {
	let bytes: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let mut reader = TsPacketReader::new(Cursor::new(bytes));
	let mut last_pcr = None;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if let Some(af) = packet.adaptation_field.as_ref()
			&& let Some(pcr) = af.pcr
		{
			let pcr = pcr.as_u64() / 300;
			if let Some(last) = last_pcr {
				assert!(
					pcr > last || af.discontinuity_indicator,
					"the clock steps back from {last} to {pcr} without a discontinuity"
				);
			}
			last_pcr = Some(pcr);
		}
		if let Some(TsPayload::PesStart(pes)) = packet.payload
			&& let Some(pcr) = last_pcr
		{
			let pts = pes.header.pts.expect("PES carried no PTS").as_u64();
			let decode = pes.header.dts.map(|t| t.as_u64()).unwrap_or(pts);
			assert!(decode >= pcr, "unit decodes at {decode}, before the clock at {pcr}");
		}
	}
}

/// With no `jitter`, an SPS that declares its reorder depth at a fixed frame rate sizes the
/// reserve from the first keyframe: every B-frame decodes at or before it is presented from the
/// start, and an exporter whose B-frames arrive behind the audio covering them renders the same
/// bytes as one that sees each tick whole.
#[tokio::test(start_paused = true)]
async fn declared_reorder_sizes_the_decode_clock_from_the_first_frame() {
	let mut rig = Reordered::new(
		crate::codec::h264::fixtures::SPS_IPB,
		crate::codec::h264::fixtures::PPS,
		&[],
	);
	let max_age = Duration::from_millis(500);
	let (mut late, mut whole) = (rig.export(max_age).await, rig.export(max_age).await);
	let (mut out_late, mut out_whole) = (Vec::new(), Vec::new());
	for tick in 0..TICKS / 2 {
		while rig.audio_before(tick + 2) {
			out_late.extend(poll_frames(&mut late));
		}
		rig.video(tick);
		out_late.extend(poll_frames(&mut late));
		out_whole.extend(poll_frames(&mut whole));
	}
	rig.finish();
	out_late.extend(drain_frames(&mut late).await);
	out_whole.extend(drain_frames(&mut whole).await);

	let late: Vec<u8> = out_late.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let whole: Vec<u8> = out_whole.iter().flat_map(|f| f.payload.iter().copied()).collect();
	assert!(late.len() > 100 * 188, "too little output to compare: {}", late.len());
	assert!(late == whole, "arrival order changed the rendering");

	let timing = video_timing(&out_late, ..);
	assert_decodes_before_presenting(&timing);
	// Depth 1 at 25 fps, and one tick so the B-frame decodes after the P-frame above it.
	let (pts, dts) = timing[1];
	assert_eq!(
		pts,
		(REORDERED_BASE + 2) * FRAME_TICKS,
		"the first P-frame follows the keyframe"
	);
	assert_eq!(
		dts,
		Some(pts - FRAME_TICKS - 1),
		"the first P-frame already runs on the declared reserve"
	);
	assert_decodes_after_the_clock(&out_late);
}

/// With nothing declared, the reserve grows to the deepest reordering muxed so far. This is the
/// one path where it depends on when an exporter joined: a joiner that has muxed only shallow
/// groups runs a shallower clock than a runner that saw a deep one before the join, until the
/// deep structure recurs. From then on the two render the same bytes.
#[tokio::test(start_paused = true)]
async fn undeclared_reorder_converges_after_its_deepest_reorder() {
	// The runner sees a pyramid in group 2; the joiner, arriving at group 5, first in group 7.
	// Group 9 repeats it on the settled reserve.
	let (runner, joiner) = export_reordered(SPS, PPS, &[2, 7, 9], None).await;
	let keyframes: Vec<Timestamp> = joiner.iter().filter(|f| f.keyframe).map(|f| f.timestamp).collect();
	let deepest = reordered_at(8 * GOP);
	assert!(
		keyframes[1] < deepest,
		"the joiner must render shallow groups before the deep one"
	);

	assert_ne!(
		video_timing(&runner, keyframes[1]..deepest),
		video_timing(&joiner, keyframes[1]..deepest),
		"a joiner that has not muxed the deepest reordering runs the runner's clock"
	);
	assert_only_continuity_differs(&runner, &joiner, deepest);

	for frames in [&runner, &joiner] {
		assert_decodes_before_presenting(&video_timing(frames, deepest..));
		// Growth steps the reserve up mid-stream; the clock must still never overtake a decode.
		assert_decodes_after_the_clock(frames);
	}
}

/// A `jitter` first published after the program tables sizes the decode clock from then on,
/// so the running exporter renders what a joiner that found it in its first catalog renders.
#[tokio::test(start_paused = true)]
async fn jitter_published_after_the_tables_raises_the_reserve() {
	let jitter = Duration::from_millis(80);
	let (runner, joiner) = export_reordered(SPS, PPS, &[], Some((2 * GOP, jitter))).await;
	let keyframes: Vec<Timestamp> = joiner.iter().filter(|f| f.keyframe).map(|f| f.timestamp).collect();
	assert_only_continuity_differs(&runner, &joiner, keyframes[1]);

	// Every keyframe after the publish decodes a full `jitter` ahead of its presentation.
	let ticks = 80 * 90;
	let keyframe_timing: Vec<(u64, Option<u64>)> = video_timing(&runner, reordered_at(3 * GOP)..)
		.into_iter()
		.filter(|(pts, _)| (pts / FRAME_TICKS).is_multiple_of(GOP))
		.collect();
	assert!(
		keyframe_timing.len() > 3,
		"too few keyframes to judge: {keyframe_timing:?}"
	);
	for (pts, dts) in keyframe_timing {
		assert_eq!(dts, Some(pts - ticks), "keyframe at {pts} ignores the published jitter");
	}
	assert_decodes_after_the_clock(&runner);
}

/// Publish a Legacy AAC rendition named `name`.
fn aac_rendition(
	broadcast: &mut moq_net::broadcast::Producer,
	catalog: &mut crate::catalog::Producer,
	name: &str,
) -> Producer<HangContainer> {
	let track = broadcast
		.create_track(name, hang::container::track_info(hang::catalog::PRIORITY.audio))
		.unwrap();
	let mut cfg = AudioConfig::new(AAC { profile: 2 }, 48_000, 2);
	cfg.container = Container::Legacy;
	catalog.modify().unwrap().audio.renditions.insert(name.to_string(), cfg);
	Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data))
}

/// Write one AAC frame at `ms`, in a group of its own.
fn write_aac(producer: &mut Producer<HangContainer>, ms: u64) {
	producer
		.write(Frame {
			timestamp: Timestamp::from_millis(ms).unwrap(),
			duration: None,
			payload: Bytes::from_static(&[0x21, 0x10, 0x04, 0x60]),
			keyframe: true,
		})
		.unwrap();
	producer.cut(None).unwrap();
}

/// Count the PES packets across `frames`.
fn pes_count(frames: &[Frame]) -> usize {
	let bytes: Vec<u8> = frames.iter().flat_map(|f| f.payload.iter().copied()).collect();
	let mut reader = TsPacketReader::new(Cursor::new(bytes));
	let mut count = 0;
	while let Some(packet) = reader.read_ts_packet().unwrap() {
		if matches!(packet.payload, Some(TsPayload::PesStart(_))) {
			count += 1;
		}
	}
	count
}

/// Pull frames until the export ends, returning them and how it ended.
async fn drain_to_end<E: tscat::Catalog>(export: &mut Export<E>) -> (Vec<Frame>, crate::Result<()>) {
	let mut out = Vec::new();
	loop {
		match tokio::time::timeout(Duration::from_secs(5), export.next())
			.await
			.expect("the export ends")
		{
			Ok(Some(frame)) => out.push(frame),
			Ok(None) => return (out, Ok(())),
			Err(err) => return (out, Err(err)),
		}
	}
}

/// A track that leaves the catalog after the PMT is read to its own end rather than
/// refused as a layout change. A publisher retires a rendition as its track ends, and
/// that catalog update can land before the track's last frames and its finish do, so
/// this retires the rendition first: the order in which the old check misread a clean
/// end as a removed track.
#[tokio::test(start_paused = true)]
async fn a_track_leaving_the_catalog_is_read_to_its_end() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut kept = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let mut leaving = aac_rendition(&mut broadcast, &mut catalog, "b.aac");

	let mut export = Export::new(crate::source::announced(&consumer))
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	for ms in (0..200).step_by(20) {
		write_aac(&mut kept, ms);
		write_aac(&mut leaving, ms);
	}
	let mut frames = drain_frames(&mut export).await;
	assert!(!frames.is_empty(), "the program started");

	catalog.modify().unwrap().audio.renditions.remove("b.aac");
	for ms in (200..300).step_by(20) {
		write_aac(&mut kept, ms);
		write_aac(&mut leaving, ms);
	}
	leaving.finish().unwrap();
	for ms in (300..400).step_by(20) {
		write_aac(&mut kept, ms);
	}
	kept.finish().unwrap();
	catalog.finish().unwrap();

	let (rest, end) = drain_to_end(&mut export).await;
	end.expect("a track leaving the catalog is not a layout change");
	frames.extend(rest);
	assert_eq!(pes_count(&frames), 20 + 15, "every frame of both tracks went out");
}

/// A broadcast that ends and is published again carries on in the same export: the
/// returned catalog resubscribes the program's tracks, the clock break is flagged once,
/// and the PAT/PMT go out again for a receiver re-acquiring after the gap. Both ways a
/// broadcast ends, a clean finish and a drop, resume the same way.
async fn resume_after(finish: bool) {
	let origin = crate::source::produce_origin();
	let source = crate::Source::new(origin.consume(), "live");
	let publish = || {
		let mut broadcast = origin.publish("live", Default::default()).unwrap();
		let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
		(broadcast, catalog)
	};

	let (mut broadcast, mut catalog) = publish();
	let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let ended = source.broadcast().await.unwrap();
	let mut export = Export::new(source.clone())
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	for ms in (0..200).step_by(20) {
		write_aac(&mut track, ms);
	}
	let mut before = drain_frames(&mut export).await;
	if finish {
		track.finish().unwrap();
		catalog.finish().unwrap();
	}
	drop((broadcast, catalog, track));
	let (rest, end) = drain_to_end(&mut export).await;
	before.extend(rest);
	assert_eq!(end.is_ok(), finish, "a finish ends cleanly and a drop fails: {end:?}");

	// The returned catalog lists the track only in its second snapshot, and the restarted
	// publisher's clock starts over.
	let (mut broadcast, mut catalog) = publish();
	std::ops::DerefMut::deref_mut(&mut catalog.modify().unwrap());
	source.returned(&ended).await.unwrap();
	export.resume().await.unwrap();
	let mut after = drain_frames(&mut export).await;
	assert!(after.is_empty(), "nothing to carry before the track is listed");
	let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	for ms in (0..200).step_by(20) {
		write_aac(&mut track, ms);
	}
	after.extend(drain_frames(&mut export).await);
	track.finish().unwrap();
	catalog.finish().unwrap();
	let (rest, end) = drain_to_end(&mut export).await;
	after.extend(rest);
	end.unwrap();

	assert_eq!(export.discontinuity(), 1, "the return is one break");
	assert_eq!(count_discontinuity(&before), 0);
	assert_eq!(count_discontinuity(&after), 1, "the break is flagged once");
	assert!(count_pid(&after, 0x0000) >= 1, "PAT re-emitted after the return");
	assert_eq!(pes_count(&after), 10, "the returned broadcast's frames all went out");
}

/// A replacement broadcast gets a fresh interleave budget: a hold that expired on the
/// broadcast before it doesn't let the new one go around a track still within its own.
#[tokio::test(start_paused = true)]
async fn resume_does_not_inherit_an_expired_stall() {
	let max_age = Duration::from_secs(2);
	let origin = crate::source::produce_origin();
	let source = crate::Source::new(origin.consume(), "live");
	let publish = || {
		let mut broadcast = origin.publish("live", Default::default()).unwrap();
		let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
		(broadcast, catalog)
	};
	// Output that needs no more than a tenth of the budget to pass.
	async fn quick(export: &mut Export, max_age: Duration) -> Vec<Frame> {
		let mut out = Vec::new();
		while let Ok(frame) = tokio::time::timeout(max_age / 10, export.next()).await {
			out.extend(frame.expect("exporter error"));
		}
		out
	}

	let (mut broadcast, mut catalog) = publish();
	let mut leading = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let mut quiet = aac_rendition(&mut broadcast, &mut catalog, "b.aac");
	let ended = source.broadcast().await.unwrap();
	let mut export = Export::new(source.clone()).await.unwrap().with_max_age(max_age);
	write_aac(&mut quiet, 0);
	for ms in (0..200).step_by(20) {
		write_aac(&mut leading, ms);
	}
	quick(&mut export, max_age).await;
	tokio::time::advance(max_age).await;
	assert!(
		!quick(&mut export, max_age).await.is_empty(),
		"the leading track went around the quiet one"
	);
	drop((broadcast, catalog, leading, quiet));
	let (_, end) = drain_to_end(&mut export).await;
	assert!(end.is_err(), "a drop fails the export");

	let (mut broadcast, mut catalog) = publish();
	let mut leading = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let mut quiet = aac_rendition(&mut broadcast, &mut catalog, "b.aac");
	source.returned(&ended).await.unwrap();
	export.resume().await.unwrap();
	for ms in (0..200).step_by(20) {
		write_aac(&mut leading, ms);
	}
	assert_eq!(
		pes_count(&quick(&mut export, max_age).await),
		0,
		"the replacement went around its quiet track on the old broadcast's budget"
	);
	write_aac(&mut quiet, 200);
	assert!(
		!quick(&mut export, max_age).await.is_empty(),
		"the replacement went out once both tracks showed"
	);
}

#[tokio::test(start_paused = true)]
async fn resume_after_a_finish() {
	resume_after(true).await;
}

#[tokio::test(start_paused = true)]
async fn resume_after_a_drop() {
	resume_after(false).await;
}

/// The stats count only output that was returned. A frame the muxer refuses fails the export
/// with the span before it queued but never returned, and the resume discards it.
#[tokio::test(start_paused = true)]
async fn export_stats_skip_output_a_failure_discards() {
	let origin = crate::source::produce_origin();
	let source = crate::Source::new(origin.consume(), "live");
	let publish = || {
		let mut broadcast = origin.publish("live", Default::default()).unwrap();
		let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
		(broadcast, catalog)
	};
	let units = |stats: Stats| stats.streams.values().map(|row| row.units).sum::<u64>() as usize;

	let (mut broadcast, mut catalog) = publish();
	let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let ended = source.broadcast().await.unwrap();
	let mut export = Export::new(source.clone())
		.await
		.unwrap()
		.with_max_age(RECORDING_MAX_AGE);
	for ms in (0..200).step_by(20) {
		write_aac(&mut track, ms);
	}
	let mut frames = drain_frames(&mut export).await;
	// One byte past what an ADTS header can frame.
	track
		.write(Frame {
			timestamp: Timestamp::from_millis(200).unwrap(),
			duration: None,
			payload: Bytes::from(vec![0; 8185]),
			keyframe: true,
		})
		.unwrap();
	let (rest, end) = drain_to_end(&mut export).await;
	frames.extend(rest);
	assert!(end.is_err(), "an unframeable AAC frame fails the export");
	assert!(pes_count(&frames) < 10, "the span before the failure stayed queued");
	assert_eq!(units(export.stats()), pes_count(&frames));

	drop((broadcast, catalog, track));
	let (mut broadcast, mut catalog) = publish();
	let mut track = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	source.returned(&ended).await.unwrap();
	export.resume().await.unwrap();
	for ms in (0..200).step_by(20) {
		write_aac(&mut track, ms);
	}
	frames.extend(drain_frames(&mut export).await);
	track.finish().unwrap();
	catalog.finish().unwrap();
	let (rest, end) = drain_to_end(&mut export).await;
	frames.extend(rest);
	end.unwrap();
	assert_eq!(
		units(export.stats()),
		pes_count(&frames),
		"the discarded span never counts"
	);
}

/// Export 25 fps video and two AAC tracks for [`TICKS`] video frames, the video and the first
/// audio track stopping after `stop` while the second carries on. Returns the stats sampled
/// after `sample` and at the end, and the frames rendered in between.
///
/// Drained after every write, like [`export_twice`], so the export reads every frame and a row
/// stops only because its track did.
async fn export_liveness(sample: u64, stop: u64) -> (Stats, Stats, Vec<Frame>) {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();

	let track = broadcast
		.create_track("video.avc1", hang::container::track_info(hang::catalog::PRIORITY.video))
		.unwrap();
	let mut cfg = VideoConfig::new(H264 {
		profile: 0x42,
		constraints: 0xc0,
		level: 0x1f,
		inline: false,
	});
	cfg.container = Container::Legacy;
	cfg.description =
		Some(crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap());
	catalog
		.modify()
		.unwrap()
		.video
		.renditions
		.insert(track.name().to_string(), cfg);
	let mut video = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data));
	let mut audio = [
		aac_rendition(&mut broadcast, &mut catalog, "primary.aac"),
		aac_rendition(&mut broadcast, &mut catalog, "secondary.aac"),
	];

	let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap();
	assert!(export.stats().is_empty(), "no row before the program tables");
	let (mut sampled, mut frames) = (None, Vec::new());
	let mut audio_index = 0;
	for tick in 0..TICKS {
		if tick < stop {
			let keyframe = tick % GOP == 0;
			let slice = if keyframe {
				vec![0x65u8; 3_000]
			} else {
				vec![0x41u8; 400]
			};
			video
				.write(Frame {
					timestamp: Timestamp::from_micros(tick * VIDEO_US).unwrap(),
					duration: None,
					payload: length_prefixed(&[&slice]),
					keyframe,
				})
				.unwrap();
		}
		while audio_index * AUDIO_US < (tick + 1) * VIDEO_US {
			let running = if tick < stop { &mut audio[..] } else { &mut audio[1..] };
			for track in running {
				track
					.write(Frame {
						timestamp: Timestamp::from_micros(audio_index * AUDIO_US).unwrap(),
						duration: None,
						payload: Bytes::from_iter((0..180u16).map(|i| (i ^ audio_index as u16) as u8)),
						keyframe: audio_index % AUDIO_GROUP == 0,
					})
					.unwrap();
			}
			audio_index += 1;
		}

		let out = drain_frames(&mut export).await;
		if sampled.is_some() {
			frames.extend(out);
		}
		if tick + 1 == sample {
			sampled = Some(export.stats());
		}
	}
	(sampled.expect("sampled mid-run"), export.stats(), frames)
}

/// A healthy export advances every elementary stream's row, each quiet for no longer than
/// its own frame spacing plus the mux buffer, and leaves the import-only counters at zero.
#[tokio::test(start_paused = true)]
async fn export_stats_advance_every_stream() {
	let (mid, end, _) = export_liveness(TICKS / 2, TICKS).await;

	let tracks: Vec<&str> = end.streams.values().map(|row| row.track).collect();
	assert_eq!(tracks, [".aac", ".aac", ".avc3"], "one row per elementary stream");
	for (pid, row) in &end.streams {
		let before = &mid.streams[pid];
		assert!(before.units > 0, "PID {pid:#x} wrote nothing by mid-run");
		assert!(
			row.units > before.units,
			"PID {pid:#x} stopped advancing: {before:?} -> {row:?}"
		);
		let quiet = row.quiet.expect("the output carries a PCR");
		assert!(quiet < Duration::from_millis(200), "PID {pid:#x} quiet for {quiet:?}");
		assert_eq!((row.resyncs, row.discarded, row.unconfirmed), (0, 0, 0));
	}
}

/// The #3533 shape: video and the primary audio stop reaching the exporter while the
/// secondary audio carries on. Their rows freeze and their quiet time grows on the output's
/// own PCR, which keeps running on the surviving track, as do the PSI and that track's row.
#[tokio::test(start_paused = true)]
async fn export_stats_catch_a_track_that_stops() {
	let stop = TICKS / 2;
	// Sampled once the mux buffer has written out what the stopped tracks sent last.
	let (sampled, end, after) = export_liveness(stop + 5, stop).await;
	let stalled = Duration::from_micros((TICKS - stop) * VIDEO_US);

	let advanced = |pid: &u16| end.streams[pid].units > sampled.streams[pid].units;
	let (running, stopped): (Vec<u16>, Vec<u16>) = end.streams.keys().copied().partition(advanced);
	let [running] = running[..] else {
		panic!("expected one stream to keep advancing, got {running:?}: {end:?}");
	};
	assert_eq!(end.streams[&running].track, ".aac");
	assert_eq!(stopped.len(), 2, "video and the primary audio stopped");

	let quiet = |pid: u16| end.streams[&pid].quiet.expect("the output carries a PCR");
	assert!(quiet(running) < Duration::from_millis(200), "{:?}", quiet(running));
	for pid in stopped {
		// The mux buffer holds the last span back, so the output's clock trails the media.
		assert!(
			quiet(pid) > stalled - Duration::from_millis(200),
			"PID {pid:#x} quiet for only {:?} of a {stalled:?} stall",
			quiet(pid)
		);
	}
	assert!(
		count_pid(&after, 0x0000) >= 3,
		"the PAT kept repeating through the stall"
	);
	assert!(count_pid(&after, running) > 0, "the surviving track kept flowing");
}

/// A stopped track's silence counts the whole of a media gap on the surviving track, not just
/// the one second of clock the exporter backfills: the output's PCR jumps across the rest
/// unflagged, and that jump is time the stopped track was silent.
#[tokio::test(start_paused = true)]
async fn export_stats_count_a_gap_longer_than_the_backfill() {
	let mut broadcast = moq_net::broadcast::Info::new().produce();
	let consumer = broadcast.consume();
	let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap();
	let mut stopped = aac_rendition(&mut broadcast, &mut catalog, "a.aac");
	let mut running = aac_rendition(&mut broadcast, &mut catalog, "b.aac");

	let mut export = Export::new(crate::source::announced(&consumer)).await.unwrap();
	for ms in (0..2_000).step_by(20) {
		write_aac(&mut stopped, ms);
		write_aac(&mut running, ms);
		drain_frames(&mut export).await;
	}
	// Five seconds with nothing on either track, then only the second resumes.
	for ms in (7_000..8_000).step_by(20) {
		write_aac(&mut running, ms);
		drain_frames(&mut export).await;
	}

	let stats = export.stats();
	let mut rows: Vec<_> = stats.streams.values().collect();
	rows.sort_by_key(|row| row.units);
	let [stopped, running] = rows[..] else {
		panic!("expected two rows: {stats:?}");
	};
	assert!(stopped.units < running.units, "{stats:?}");
	let quiet = stopped.quiet.expect("the output carries a PCR");
	assert!(
		quiet > Duration::from_millis(5_800),
		"stopped at 2 s, quiet for only {quiet:?} at 8 s"
	);
	let quiet = running.quiet.expect("the output carries a PCR");
	assert!(quiet < Duration::from_millis(200), "{quiet:?}");
}