media-plane 0.11.0

//! [`TsIngestSession`] — the shared MPEG-2 TS ingest session (plan step 5a).
//!
//! Every TS-carrying ingest source of a host — its TS-over-UDP source
//! (UDP datagrams), its TS-over-HTTP source (a chunked HTTP body),
//! its SRT source (SRT payloads) — differs *only* in its transport.
//! Once the bytes are in hand, all three do the identical thing: feed
//! [`transmux::StreamingTsDemux`] and translate its
//! [`transmux::DemuxEvent`]s into [`SessionEvent`]s. Before this port each
//! kept its own copy of that `while let Some(ev) = demux.poll_event()` drain
//! (three near-identical ~60-line blocks, which had already drifted — see
//! the differing `TrackAdded` log wording); this module is the single copy,
//! and the three transports keep only their socket/stream handling.
//!
//! This is exactly the consolidation `crate::ingress`'s module docs
//! anticipate ("this module's job is to ... own the feed/poll/deadline/
//! dispatch loop so no protocol has to reimplement it"): the plane owns the
//! driver loop, and this owns the one TS-specific translation the plane
//! cannot know about.
//!
//! # B5 + issue #781: mid-stream track changes reach the running segmenter
//!
//! Before this port, a PID declared only *after* `connect()`'s PMT wait
//! resolved was logged and silently dropped (issue #774). `ProgramTracker`
//! closed that by minting a new `ProgramId` for each mid-stream
//! `TrackAdded` — the track's samples were no longer dropped, but the new
//! program never got a segmenter (the brief's original defect: the segmenter
//! is built once at program-observation time from a one-shot track-spec
//! snapshot, issue #781).
//!
//! Now a mid-stream `TrackAdded` joins the **same** `ProgramId` as the
//! initial set — `track_program` maps it into the first-resolved program,
//! `track_specs` stores the spec, and the following `TracksResolved` emits a
//! `SessionEvent::TracksChanged` with the complete current track set. The
//! driver's existing `TracksChanged` handling applies it to the running
//! `Trunk` (bumping `track_generation`), and `drive_program_segmenters`
//! detects the generation change to admit the new track into (or rebuild)
//! the segmenter at the next segment boundary.
//!
//! # Why `Established` is queued at construction
//!
//! None of the three TS transports has a *media-level* handshake to
//! negotiate once its transport is up (UDP: a local bind; HTTP: the response
//! headers are already in; SRT: the SRT handshake completed in the socket
//! layer). They are the "purely local operation with nothing to negotiate"
//! case `crate::ingress`'s own `ScriptedSession` test precedent
//! documents, so [`SessionEvent::Established`] is queued up front rather
//! than gated on the PMT — establishment is a per-*connection* fact, and the
//! PMT resolving is a per-*program* one that `NewProgram` already carries
//! (see `SessionEvent::Established`'s "Why not called `TracksResolved`").

use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::convert::Infallible;
use transmux::{ProgramNumber, TrackId};

use crate::ingress::{IngestSession, ProgramId, SessionEvent};
use crate::trunk::RetentionClass;
use broadcast_common::{Demand, Stage, Timestamp};

use transmux::pipeline::{Sample, TrackSpec};
use transmux::{DemuxEvent, StreamingTsDemux};

/// Largest read a transport should hand [`TsIngestSession::feed`] at a time; the session's [`Demand`].
pub const MAX_TS_READ: usize = 65_536;

/// Most samples [`ProgramTracker`] holds for tracks the demux has announced
/// but whose programme is not yet announced (before the first
/// `TracksResolved`). See [`ProgramTracker`]'s "Pre-resolution backlog".
pub const MAX_HELD_SAMPLES: usize = 65_536;

/// Most sample payload bytes [`ProgramTracker`] holds in that pre-resolution
/// backlog (64 MiB). Whichever of the two caps is hit first applies.
pub const MAX_HELD_BYTES: usize = 64 * 1024 * 1024;

/// Translates [`transmux::DemuxEvent`]s into [`SessionEvent`]s and tracks
/// which [`ProgramId`] owns each `track_id` — a plain, byte-free state
/// machine (no socket, no demuxer), so the B5 mid-stream-`NewProgram`
/// behaviour is unit-testable by constructing [`DemuxEvent`]s directly (via
/// their own `#[non_exhaustive]` constructors), without needing a hand-built
/// MPTS byte stream.
///
/// # MPTS support (issue #906)
///
/// Tracks are grouped by `TrackSpec::program_number`: each distinct
/// `program_number` gets its own [`ProgramId`], so an MPTS (multi-programme
/// transport stream) produces N `ProgramId`s for N `program_number`s.
/// Non-TS sources (`program_number: None`) collapse into one `ProgramId`
/// (unchanged from the pre-906 single-program behaviour).
///
/// Mid-stream `TrackAdded` (issue #781) is preserved PER PROGRAM: a track
/// added mid-stream joins the program keyed by its `program_number`.
///
/// # Pre-resolution backlog
///
/// The demux promotes a track (`TrackAdded`) as soon as its codec config is
/// known and replays that track's probe backlog as `Sample`s right away -
/// which is *before* `TracksResolved`, the signal this tracker announces the
/// programme on. A [`SessionEvent::Sample`] must follow its programme's
/// `NewProgram`, so those samples are **held** (in arrival order, across
/// tracks) and flushed immediately after the `NewProgram` events, ahead of
/// any later sample: no loss, no duplicate, no reordering. Only samples of a
/// track that was already announced are held; a sample for a track the demux
/// never announced is still dropped.
///
/// The hold is bounded by [`MAX_HELD_SAMPLES`] / [`MAX_HELD_BYTES`]. On
/// overflow (a head that never resolves, with other tracks flowing) the
/// **oldest** held sample is dropped and counted in the
/// backlog-drop observer ([`ProgramTracker::with_backlog_drop_observer`]), so the newest media (the
/// live edge) survives.
pub struct ProgramTracker {
    /// Samples for announced-but-not-yet-programmed tracks, oldest first.
    held: VecDeque<(TrackId, Sample)>,
    held_bytes: usize,
    pending: VecDeque<SessionEvent>,
    /// Tracks buffered before the first `TracksResolved`, grouped by
    /// `program_number`. A `BTreeMap` so `ProgramId`s are minted (and
    /// `NewProgram` announced) in ascending `program_number` order: stable
    /// across processes, so a URL for a program survives a restart.
    resolving: BTreeMap<Option<ProgramNumber>, Vec<TrackSpec>>,
    resolved_once: bool,
    /// Resolved programs keyed by `program_number`.
    programs: BTreeMap<Option<ProgramNumber>, PerProgram>,
    /// Reverse map: `track_id → ProgramId` (for `Sample` routing).
    track_program: HashMap<TrackId, ProgramId>,
    /// Programs whose track set changed since last `TracksResolved`.
    changed: HashSet<ProgramId>,
    next_program_id: u32,
    /// Told how many held samples an overflow just dropped.
    backlog_drop_observer: Option<fn(u64)>,
}

struct PerProgram {
    program_id: ProgramId,
    /// Complete current track specs for this program, keyed by `track_id`.
    track_specs: BTreeMap<TrackId, TrackSpec>,
}

impl Default for ProgramTracker {
    fn default() -> Self {
        Self::new()
    }
}

impl ProgramTracker {
    /// Starts with [`SessionEvent::Established`] already queued — see the
    /// module doc.
    pub fn new() -> Self {
        ProgramTracker {
            held: VecDeque::new(),
            held_bytes: 0,
            pending: VecDeque::from(vec![SessionEvent::Established]),
            resolving: BTreeMap::new(),
            resolved_once: false,
            programs: BTreeMap::new(),
            track_program: HashMap::new(),
            changed: HashSet::new(),
            next_program_id: 0,
            backlog_drop_observer: None,
        }
    }

    /// Call `observer` with the number of held samples dropped each time the
    /// pre-resolution backlog overflows (a counter in the host, say).
    #[must_use]
    pub fn with_backlog_drop_observer(mut self, observer: fn(u64)) -> Self {
        self.backlog_drop_observer = Some(observer);
        self
    }

    pub fn handle(&mut self, event: DemuxEvent) {
        match event {
            DemuxEvent::TrackAdded(spec) => self.track_added(spec),
            DemuxEvent::TracksResolved { .. } => self.tracks_resolved(),
            DemuxEvent::Sample {
                track_id, sample, ..
            } => self.sample(track_id, sample),
            DemuxEvent::TrackRemoved { track_id, .. } => self.track_removed(track_id),
            DemuxEvent::TrackUpdated(spec) => self.track_updated(spec),
            DemuxEvent::TrackAbandoned { .. } => {}
            _ => {}
        }
    }

    pub fn track_added(&mut self, spec: TrackSpec) {
        let pn = spec.program_number;
        if self.resolved_once {
            // Issue #781 + #906: a track declared after initial
            // resolution joins its program_number's program.
            let program = self.find_or_create_program(pn);
            self.track_specs_for_program(program)
                .insert(spec.track_id, spec.clone());
            self.track_program.insert(spec.track_id, program);
            self.changed.insert(program);
        } else {
            self.resolving.entry(pn).or_default().push(spec);
        }
    }

    pub fn tracks_resolved(&mut self) {
        if !self.resolved_once {
            if !self.resolving.is_empty() {
                self.resolved_once = true;
                let groups: Vec<_> = std::mem::take(&mut self.resolving).into_iter().collect();
                for (pn, tracks) in groups {
                    if tracks.is_empty() {
                        continue;
                    }
                    let program = ProgramId(self.next_program_id);
                    self.next_program_id += 1;
                    let mut track_specs = BTreeMap::new();
                    for spec in &tracks {
                        self.track_program.insert(spec.track_id, program);
                        track_specs.insert(spec.track_id, spec.clone());
                    }
                    self.programs.insert(
                        pn,
                        PerProgram {
                            program_id: program,
                            track_specs,
                        },
                    );
                    self.pending
                        .push_back(SessionEvent::NewProgram { program, tracks });
                }
                self.held_bytes = 0;
                for (track_id, sample) in std::mem::take(&mut self.held) {
                    if let Some(&program) = self.track_program.get(&track_id) {
                        self.pending.push_back(SessionEvent::Sample {
                            program,
                            track_id,
                            retention: RetentionClass::Timed,
                            sample,
                        });
                    }
                }
            }
        } else {
            // Mid-stream TracksResolved: only emit TracksChanged for
            // programs whose track set actually changed.
            let dirty = std::mem::take(&mut self.changed);
            for prog in self.programs.values() {
                if !dirty.contains(&prog.program_id) {
                    continue;
                }
                let tracks: Vec<TrackSpec> = prog.track_specs.values().cloned().collect();
                if !tracks.is_empty() {
                    self.pending.push_back(SessionEvent::TracksChanged {
                        program: prog.program_id,
                        tracks,
                    });
                }
            }
        }
    }

    /// A sample for a track never announced (or since removed) is dropped,
    /// mirroring the pre-5a `known_track_ids` check. Before the first
    /// `TracksResolved` the programme is not yet announced: the sample of an
    /// announced track is held (the demux's probe-backlog replay) until it is.
    /// See "Pre-resolution backlog".
    pub fn sample(&mut self, track_id: TrackId, sample: Sample) {
        if let Some(&program) = self.track_program.get(&track_id) {
            self.pending.push_back(SessionEvent::Sample {
                program,
                track_id,
                retention: RetentionClass::Timed,
                sample,
            });
        } else if !self.resolved_once
            && self
                .resolving
                .values()
                .any(|specs| specs.iter().any(|s| s.track_id == track_id))
        {
            self.hold(track_id, sample);
        }
    }

    /// A mid-stream PMT version bump dropped a previously-live PID
    /// (issue #774): stop routing samples for it.
    pub fn track_removed(&mut self, track_id: TrackId) {
        if let Some(program) = self.track_program.remove(&track_id) {
            if let Some(prog) = self.program_for_program_id_mut(program) {
                prog.track_specs.remove(&track_id);
            }
            self.changed.insert(program);
        }
    }

    /// A track's codec config changed mid-stream (a new in-band SPS/PPS/AAC
    /// config, or a changed PMT descriptor). Adopt the new spec and announce
    /// the program's track set, so the route rebuilds its init segment instead
    /// of describing a stream that is no longer being sent. Before the first
    /// `TracksResolved` the buffered spec is just replaced; an update for a
    /// track that was never announced is ignored (nothing routes on it).
    pub fn track_updated(&mut self, spec: TrackSpec) {
        let track_id = spec.track_id;
        if self.resolved_once {
            if let Some(&program) = self.track_program.get(&track_id)
                && let Some(prog) = self.program_for_program_id_mut(program)
            {
                prog.track_specs.insert(track_id, spec);
                let mut tracks: Vec<TrackSpec> = prog.track_specs.values().cloned().collect();
                tracks.sort_by_key(|t| t.track_id);
                self.pending
                    .push_back(SessionEvent::TracksChanged { program, tracks });
            }
        } else if let Some(buffered) = self
            .resolving
            .values_mut()
            .flat_map(|specs| specs.iter_mut())
            .find(|s| s.track_id == track_id)
        {
            *buffered = spec;
        }
    }

    pub fn poll(&mut self) -> Option<SessionEvent> {
        self.pending.pop_front()
    }

    /// Hold one pre-resolution sample, dropping the oldest held ones (counted)
    /// while either cap would be exceeded.
    fn hold(&mut self, track_id: TrackId, sample: Sample) {
        self.held_bytes += sample.data.len();
        self.held.push_back((track_id, sample));
        let mut dropped = 0u64;
        while self.held.len() > MAX_HELD_SAMPLES || self.held_bytes > MAX_HELD_BYTES {
            let Some((_, old)) = self.held.pop_front() else {
                break;
            };
            self.held_bytes -= old.data.len();
            dropped += 1;
        }
        if dropped > 0
            && let Some(observer) = self.backlog_drop_observer
        {
            observer(dropped);
        }
    }

    /// Find or create the `ProgramId` for a given `program_number`.
    fn find_or_create_program(&mut self, pn: Option<ProgramNumber>) -> ProgramId {
        if let Some(prog) = self.programs.get(&pn) {
            return prog.program_id;
        }
        let program = ProgramId(self.next_program_id);
        self.next_program_id += 1;
        self.programs.insert(
            pn,
            PerProgram {
                program_id: program,
                track_specs: BTreeMap::new(),
            },
        );
        // Mid-stream NewProgram — emitted before any TracksChanged so the
        // driver can mint a Trunk.
        self.pending.push_back(SessionEvent::NewProgram {
            program,
            tracks: Vec::new(),
        });
        program
    }

    fn track_specs_for_program(&mut self, program: ProgramId) -> &mut BTreeMap<TrackId, TrackSpec> {
        &mut self
            .program_for_program_id_mut(program)
            .expect("program must exist")
            .track_specs
    }

    fn program_for_program_id_mut(&mut self, program: ProgramId) -> Option<&mut PerProgram> {
        self.programs.values_mut().find(|p| p.program_id == program)
    }
}

/// The shared MPEG-2 TS [`IngestSession`]: no socket, no I/O — just a
/// [`StreamingTsDemux`] plus a `ProgramTracker`. Each transport module
/// (`ts_udp`/`ts_http`/`srt`) owns the real socket/stream and feeds this.
pub struct TsIngestSession {
    demux: StreamingTsDemux,
    tracker: ProgramTracker,
    first_fed_at: Option<Timestamp>,
    silent_heads_released: bool,
}

/// Arrival time after the first byte of input past which a PID the PMT
/// declares but that has sent no payload packet is treated as absent rather
/// than late ([`StreamingTsDemux::release_silent_heads`]). Without this bound
/// such a PID holds every track, and with them `NewProgram`, back until
/// 4 MiB of input has arrived (`STUCK_HEAD_INPUT_PACKETS`): never, on a
/// low-bitrate SPTS. Bounded by time, not bytes, so the wait does not scale
/// with the bitrate. A silent PID that does send later is announced by a
/// later `TrackAdded`, so nothing is lost; one second is the same order as
/// the settle `broadcast-client` allows a freshly joined stream.
pub const SILENT_PID_SETTLE: std::time::Duration = std::time::Duration::from_secs(1);

impl TsIngestSession {
    /// Call `observer` with the number of samples dropped when the
    /// pre-resolution backlog overflows; see
    /// [`ProgramTracker::with_backlog_drop_observer`].
    #[must_use]
    pub fn with_backlog_drop_observer(mut self, observer: fn(u64)) -> Self {
        self.tracker = self.tracker.with_backlog_drop_observer(observer);
        self
    }

    /// Construct a fresh session — performs no I/O.
    pub fn new() -> Self {
        TsIngestSession {
            demux: StreamingTsDemux::new(),
            tracker: ProgramTracker::new(),
            first_fed_at: None,
            silent_heads_released: false,
        }
    }

    fn drain_demux(&mut self) {
        while let Some(event) = self.demux.poll_event() {
            self.tracker.handle(event);
        }
    }
}

impl Default for TsIngestSession {
    fn default() -> Self {
        Self::new()
    }
}

impl Stage for TsIngestSession {
    type In<'a> = &'a [u8];
    type Out = SessionEvent;
    /// TS demuxing itself cannot fail (`StreamingTsDemux::feed` is
    /// infallible — a corrupt packet is resynced past, not surfaced as an
    /// error). Every failure mode of a TS ingest route (a dead socket, a
    /// read stall, an HTTP status) lives at the I/O layer in the transport
    /// module, outside this sans-IO session entirely.
    type Error = Infallible;

    fn feed(&mut self, input: &[u8], now: Timestamp) -> Result<(), Infallible> {
        self.demux.feed(input);
        self.drain_demux();
        let first = *self.first_fed_at.get_or_insert(now);
        if !self.silent_heads_released && now.saturating_sub(first) >= SILENT_PID_SETTLE {
            self.silent_heads_released = true;
            self.demux.release_silent_heads();
            self.drain_demux();
        }
        Ok(())
    }

    fn poll(&mut self) -> Option<SessionEvent> {
        self.tracker.poll()
    }

    /// End of input: flush the demux (`StreamingTsDemux::finish`) so a declared
    /// PID that never carried data cannot hold `TracksResolved` — and with it
    /// `NewProgram` and every parked/probing track's backlog — hostage, then
    /// translate what it emitted. Idempotent: the demux's second flush emits
    /// nothing, so a repeated call queues no duplicate event.
    fn finish(&mut self) -> Result<(), Infallible> {
        self.demux.finish();
        while let Some(event) = self.demux.poll_event() {
            self.tracker.handle(event);
        }
        Ok(())
    }

    fn demand(&self) -> Demand {
        Demand::new(MAX_TS_READ)
    }
}

/// Nothing to send back on any of the three TS transports — takes the
/// default `poll_transmit`. `Request = Bytes` (media-plane round 3: every
/// `IngestSession` now names its own request type; a byte-stream source
/// always names `Bytes`, whether or not it ever actually sends one).
impl IngestSession for TsIngestSession {
    type Request = bytes::Bytes;
}

/// Shared test fixtures for the three TS transports — a real `TsMux`-muxed
/// byte stream plus the `TrunkConfig`/`HandshakePolicy` each transport's
/// loopback test needs. Lives here (rather than being copy-pasted into
/// `ts_udp`/`ts_http`/`srt`'s own `mod tests`) for the same
/// one-copy reason the session itself does.
/// Real `TsMux`-muxed TS bytes and the configs a TS transport's test needs.
#[cfg(any(test, feature = "test-support"))]
pub mod test_support {
    use crate::ingress::HandshakePolicy;
    use crate::trunk::TrunkConfig;
    use transmux::TrackId;

    use broadcast_common::Timestamp;
    use std::num::NonZeroUsize;
    use transmux::TsMux;
    use transmux::media::Track;
    use transmux::pipeline::{CodecConfig, Sample, TrackSpec};

    fn nz(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("test capacity must be non-zero")
    }

    /// A small `TrunkConfig`.
    pub fn trunk_config() -> TrunkConfig {
        TrunkConfig::default()
            .with_timed_capacity(nz(64))
            .with_sparse_capacity(nz(16))
            .with_segment_capacity(nz(8))
            .with_event_capacity(nz(8))
            .with_part_capacity(nz(8))
    }

    /// A handshake deadline far enough out that it never fires — these
    /// transports' `Established` is queued at construction.
    /// A handshake deadline that never fires.
    pub fn handshake() -> HandshakePolicy {
        HandshakePolicy::establish_by(Timestamp::from_nanos(u64::MAX))
    }

    /// An AVC track spec.
    pub fn track_spec(track_id: TrackId) -> TrackSpec {
        let avc = transmux::avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==")
            .expect("valid sprop");
        TrackSpec::new(
            track_id,
            90_000,
            CodecConfig::Avc {
                config: avc,
                width: 0,
                height: 0,
            },
        )
    }

    /// Builds a real (not hand-faked) single-track muxed TS byte stream via
    /// the workspace's own `transmux::TsMux` packager — the "real fixture,
    /// not inline bytes" discipline this project requires, since a
    /// hand-built TS risks missing real PSI/PES framing quirks.
    /// A muxed TS of `count` frames of one track.
    pub fn build_ts_bytes(track_id: TrackId, nal_byte: u8, count: u32) -> Vec<u8> {
        build_ts_bytes_with_gop(track_id, nal_byte, count, u32::MAX)
    }

    /// [`build_ts_bytes`] with a sync sample every `gop` frames, so the
    /// segmenter closes a segment per `gop` frames.
    pub fn build_ts_bytes_with_gop(
        track_id: TrackId,
        nal_byte: u8,
        count: u32,
        gop: u32,
    ) -> Vec<u8> {
        use broadcast_common::Package;
        let spec = track_spec(track_id);
        let frame_dur = 90_000 / 30;
        let samples: Vec<Sample> = (0..count)
            .map(|i| {
                let nal = [0x65u8, nal_byte, (i % 256) as u8];
                let mut data = (nal.len() as u32).to_be_bytes().to_vec();
                data.extend_from_slice(&nal);
                Sample::new(
                    data,
                    Some(i64::from(i) * i64::from(frame_dur)),
                    Some(i64::from(i) * i64::from(frame_dur)),
                    Some(frame_dur),
                    i % gop == 0,
                )
            })
            .collect();
        let track = Track::new(spec, samples);
        let media = transmux::media::Media::new(vec![track], 90_000);
        TsMux::default().package(&media).expect("mux to TS")
    }
}

#[cfg(test)]
mod tests {
    use super::test_support::track_spec;
    use super::*;
    use transmux::pipeline::Sample;

    fn sample_at(nal: u8) -> Sample {
        Sample::new(vec![0x65, nal], Some(0), Some(0), Some(3000), true)
    }

    /// Determinism: an MPTS announces programs in ascending `program_number`
    /// order with `ProgramId`s 0,1,2 in that order, every time. A fresh
    /// tracker per iteration gets a fresh `RandomState`, so a `HashMap`
    /// iteration order would differ between iterations (each of the 6
    /// orders is equally likely: 200 iterations make a pass by luck
    /// vanishingly improbable).
    #[test]
    fn mpts_programs_announced_in_ascending_program_number_order() {
        for _ in 0..200 {
            let mut tracker = ProgramTracker::new();
            // Arrival order deliberately not sorted.
            for (pn, tid) in [(300u16, 31u32), (100, 11), (200, 21), (100, 12)] {
                tracker.handle(DemuxEvent::TrackAdded(
                    track_spec(TrackId::new(tid)).with_program(ProgramNumber::new(pn)),
                ));
            }
            tracker.handle(DemuxEvent::tracks_resolved(0));
            assert!(matches!(tracker.poll(), Some(SessionEvent::Established)));
            let mut got = Vec::new();
            while let Some(ev) = tracker.poll() {
                match ev {
                    SessionEvent::NewProgram { program, tracks } => {
                        got.push((
                            program,
                            tracks.iter().map(|t| t.track_id).collect::<Vec<_>>(),
                        ));
                    }
                    other => panic!("unexpected {other:?}"),
                }
            }
            assert_eq!(
                got,
                vec![
                    (ProgramId(0), vec![TrackId::new(11), TrackId::new(12)]),
                    (ProgramId(1), vec![TrackId::new(21)]),
                    (ProgramId(2), vec![TrackId::new(31)]),
                ]
            );
        }
    }

    /// Determinism of the mid-stream path: `TracksChanged` is emitted in
    /// ascending `ProgramId`/`program_number` order, each with its tracks
    /// sorted by `track_id`.
    #[test]
    fn mid_stream_tracks_changed_is_ordered() {
        for _ in 0..200 {
            let mut tracker = ProgramTracker::new();
            for (pn, tid) in [(100u16, 11u32), (200, 21), (300, 31)] {
                tracker.handle(DemuxEvent::TrackAdded(
                    track_spec(TrackId::new(tid)).with_program(ProgramNumber::new(pn)),
                ));
            }
            tracker.handle(DemuxEvent::tracks_resolved(0));
            while tracker.poll().is_some() {}
            // Dirty every program, several tracks each.
            for (pn, tid) in [(300u16, 39u32), (100, 19), (200, 29), (100, 15), (300, 33)] {
                tracker.handle(DemuxEvent::TrackAdded(
                    track_spec(TrackId::new(tid)).with_program(ProgramNumber::new(pn)),
                ));
            }
            tracker.handle(DemuxEvent::tracks_resolved(1));
            let mut got = Vec::new();
            while let Some(ev) = tracker.poll() {
                match ev {
                    SessionEvent::TracksChanged { program, tracks } => {
                        got.push((
                            program,
                            tracks
                                .iter()
                                .map(|t| t.track_id.value())
                                .collect::<Vec<_>>(),
                        ));
                    }
                    other => panic!("unexpected {other:?}"),
                }
            }
            assert_eq!(
                got,
                vec![
                    (ProgramId(0), vec![11, 15, 19]),
                    (ProgramId(1), vec![21, 29]),
                    (ProgramId(2), vec![31, 33, 39]),
                ]
            );
        }
    }

    #[test]
    fn first_tracks_resolved_mints_program_zero() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        assert!(matches!(tracker.poll(), Some(SessionEvent::Established)));
        match tracker.poll() {
            Some(SessionEvent::NewProgram { program, tracks }) => {
                assert_eq!(program, ProgramId(0));
                assert_eq!(tracks.len(), 1);
            }
            other => panic!("expected NewProgram(0), got {other:?}"),
        }
    }

    /// **Issue #781:** a `TrackAdded` arriving *after* the initial program
    /// resolved goes to the **same** `ProgramId` — the track's samples join
    /// the running segmenter, rather than being siloed in a new, unserved
    /// program (the exact defect the brief describes).
    ///
    /// The `TrackAdded` itself emits nothing; the following `TracksResolved`
    /// emits `TracksChanged` with the complete current set.
    ///
    /// MUTATION-CHECKED: change `self.first_program.expect(...)` in the
    /// mid-stream `TrackAdded` arm back to minting a new `ProgramId` (the
    /// pre-781 behaviour) and this test's `assert_eq!(program, first)` fails:
    /// the sample routes to a new program, not the first one.
    #[test]
    fn late_track_added_joins_the_same_program_not_a_new_one() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let first = match tracker.poll() {
            Some(SessionEvent::NewProgram { program, .. }) => program,
            other => panic!("expected NewProgram, got {other:?}"),
        };

        // Mid-stream TrackAdded: emitted only when TracksResolved follows.
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(9))));
        assert!(
            tracker.poll().is_none(),
            "TrackAdded alone must not emit anything — TracksChanged waits for TracksResolved"
        );

        // TracksResolved after the addition: emits TracksChanged.
        tracker.handle(DemuxEvent::tracks_resolved(1));
        match tracker.poll() {
            Some(SessionEvent::TracksChanged { program, tracks }) => {
                assert_eq!(program, first, "TracksChanged must go to the same program");
                assert_eq!(tracks.len(), 2, "complete current set: both tracks");
                let ids: Vec<TrackId> = tracks.iter().map(|t| t.track_id).collect();
                assert!(ids.contains(&TrackId::new(1)));
                assert!(ids.contains(&TrackId::new(9)));
            }
            other => panic!("expected TracksChanged, got {other:?}"),
        }

        // Sample for the new track routes to the same program.
        tracker.handle(DemuxEvent::sample(TrackId::new(9), sample_at(0xAA)));
        match tracker.poll() {
            Some(SessionEvent::Sample {
                program, track_id, ..
            }) => {
                assert_eq!(program, first);
                assert_eq!(track_id, TrackId::new(9));
            }
            other => panic!("expected Sample routed to the first program, got {other:?}"),
        }
    }

    /// A `Sample` for a track never announced (or already `TrackRemoved`) is
    /// dropped, not panicked on or misrouted.
    #[test]
    fn sample_for_unannounced_track_is_dropped() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::sample(TrackId::new(42), sample_at(0x01)));
        assert!(matches!(tracker.poll(), Some(SessionEvent::Established)));
        assert!(
            tracker.poll().is_none(),
            "no event for an unannounced track's sample"
        );
    }

    /// A `TrackRemoved` mid-stream stops routing samples for that track and
    /// removes its spec from the complete set — the following `TracksResolved`
    /// emits `TracksChanged` without the removed track.
    #[test]
    fn removed_track_vanishes_from_tracks_changed() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(2))));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let _new_program = tracker.poll();

        // Remove track 2.
        tracker.handle(DemuxEvent::track_removed(
            TrackId::new(2),
            transmux::EventProvenance::default(),
        ));
        // TracksResolved after the removal.
        tracker.handle(DemuxEvent::tracks_resolved(1));
        match tracker.poll() {
            Some(SessionEvent::TracksChanged { tracks, .. }) => {
                assert_eq!(tracks.len(), 1);
                assert_eq!(tracks[0].track_id, TrackId::new(1));
            }
            other => panic!("expected TracksChanged with track 1 only, got {other:?}"),
        }

        // Sample for the removed track is dropped.
        tracker.handle(DemuxEvent::sample(TrackId::new(2), sample_at(0xBB)));
        assert!(
            tracker.poll().is_none(),
            "sample for removed track must be dropped"
        );
    }

    /// **The segmenter gap, closed.** Step 5a round 1 reported that no
    /// producer could obtain a second `Trunk` write handle, so the
    /// `LlHlsSegmenter`/movie-timescale path the pre-5a `run_pipeline`
    /// owned was structurally impossible to express: `Trunk::writer()` was
    /// single-take for the *whole* `Trunk`, and `IngestDriver` claims it
    /// per-program to publish samples.
    ///
    /// `Trunk::segment_writer()` (the ring-group split) fixes exactly that,
    /// and this test is the proof — not an assertion in a doc comment. It
    /// drives the **real** chain: a real `TsMux`-muxed TS stream → the real
    /// `TsIngestSession` → a real `IngestDriver` (holding the `TrunkWriter`)
    /// → a real `SampleCursor` → a real `LlHlsSegmenter` → a
    /// `SegmentWriter` **taken from the same `Trunk` the driver is already
    /// writing samples into** → `Trunk::part_bytes`/`last_closed_segment`.
    ///
    /// Both handles are held live simultaneously across the whole loop; if
    /// the split had not landed, `segment_writer()` would return `None` and
    /// the `expect` below would fail — which is what makes this a real
    /// check rather than a restatement.
    ///
    /// This is deliberately a *test*, not production wiring: production
    /// wiring is `crate::source::segment::ProgramSegmenter`/
    /// `drive_program_segmenters` (issue #805 task 2b), which every
    /// driver-backed `run_*` entry point calls; it uses
    /// `transmux::VIDEO_CLOCK_RATE` for the same movie timescale this test
    /// hardcodes, and takes `target_duration_secs`/`part_target_ms` off the
    /// route exactly as this test's literal `1.0, 250` stand in for here.
    #[test]
    fn a_segmenter_can_hold_a_segment_writer_while_ingest_holds_the_sample_writer() {
        use super::test_support::{build_ts_bytes, handshake, track_spec, trunk_config};
        use crate::ingress::{IngestDriver, ProgramId};
        use crate::trunk::{PartEntry, SampleCursorItem, SegmentEntry};
        use broadcast_common::Timestamp;
        use std::time::Duration as StdDuration;
        use transmux::ll_hls::LlHlsSegmenter;
        use transmux::segmenter::SegmentMeta;

        let mut driver = IngestDriver::new(
            TsIngestSession::new(),
            crate::IngestConfig::new(handshake()).with_trunk(trunk_config()),
        );

        // Feed a real muxed TS stream; the driver mints program 0's Trunk
        // and takes its TrunkWriter internally.
        let ts_bytes = build_ts_bytes(TrackId::new(1), 0xAB, 90);
        driver.feed(&ts_bytes, Timestamp::ZERO);
        let trunk = driver
            .trunk(ProgramId(0))
            .cloned()
            .expect("program 0 resolved from the muxed TS");

        // The sample/event writer is genuinely already taken — this is what
        // made the segmenter impossible before the ring-group split, and
        // asserting it here is what stops this test from silently degrading
        // into "a Trunk nobody else was writing to".
        assert!(
            trunk.writer().is_none(),
            "the IngestDriver must already hold this Trunk's sample writer"
        );

        // THE POINT: a second, independent write handle on the SAME Trunk
        // the driver is already publishing samples into.
        let segment_writer = trunk
            .segment_writer()
            .expect("the segments+parts ring group has its own single-take writer");
        assert!(
            trunk.segment_writer().is_none(),
            "the segment writer is itself single-take, on its own flag"
        );

        // A segmenter consumes samples off a cursor and produces
        // segments/parts. One cursor, per `Trunk::subscribe`'s
        // single-digit-readers-by-design contract.
        let mut cursor = trunk.subscribe();
        let mut seg = LlHlsSegmenter::with_part_target(
            vec![track_spec(TrackId::new(1))],
            transmux::VIDEO_CLOCK_RATE,
            1.0,
            250,
        )
        .expect("segmenter builds from the resolved track spec");

        // Drive more media through so the cursor (which starts from *now*)
        // actually observes samples.
        let more = build_ts_bytes(TrackId::new(1), 0xCD, 90);
        driver.feed(&more, Timestamp::from_nanos(1));

        let mut pushed = 0usize;
        while let Some(item) = cursor.poll() {
            if let SampleCursorItem::Timed { track_id, sample } = item {
                seg.push(track_id, sample)
                    .expect("segmenter accepts sample");
                pushed += 1;
            }
        }
        assert!(pushed > 0, "the cursor must have observed real samples");
        seg.flush().expect("flush the trailing partial segment");

        let mut parts_published = 0usize;
        for part in seg.take_ready_parts() {
            segment_writer.publish_part(PartEntry::new(
                part.bytes,
                crate::ids::SegmentSeq::new(part.segment_seq),
                crate::ids::PartIndex::new(part.part_index),
                StdDuration::from_secs_f64(part.duration),
                part.independent,
            ));
            parts_published += 1;
        }
        let mut segments_published = 0usize;
        for segment in seg.take_ready_segments() {
            segment_writer
                .publish_segment(SegmentEntry::new(
                    segment.bytes,
                    crate::ids::SegmentSeq::new(segment.segment_seq),
                    StdDuration::from_secs_f64(segment.duration),
                    Timestamp::ZERO,
                    SegmentMeta {
                        discontinuous: false,
                    },
                ))
                .unwrap();
            segments_published += 1;
        }

        assert!(
            segments_published > 0,
            "the segmenter must have produced at least one closed segment"
        );
        assert!(
            trunk.last_closed_segment().is_some(),
            "a published segment must be visible via Trunk::last_closed_segment"
        );
        assert_eq!(
            trunk.segment_len(),
            segments_published,
            "every published segment must be in the Trunk's segment log"
        );
        if parts_published > 0 {
            assert!(
                trunk.part_len() > 0,
                "published parts must be visible in the Trunk's live-part log"
            );
        }
    }

    /// A real `TsMux` stream: a PMT-first non-video head `(0x04` private
    /// audio-type PES that never carries a frame sync, so it can never
    /// resolve) with 540 access units, followed by two healthy data PIDs.
    fn stalled_head_ts_bytes() -> Vec<u8> {
        stalled_head_ts_bytes_n(540)
    }

    /// [`stalled_head_ts_bytes`] with `head_aus` access units on the stalled head.
    fn stalled_head_ts_bytes_n(head_aus: u32) -> Vec<u8> {
        use broadcast_common::Package;
        use transmux::media::{Media, Track};
        use transmux::pipeline::{CodecConfig, DataCarriage};
        let dur = 3000u32;
        let samples = |n: u32, len: usize, sync: bool| -> Vec<Sample> {
            (0..n)
                .map(|i| {
                    let t = i64::from(i) * i64::from(dur);
                    Sample::new(vec![0xBBu8; len], Some(t), Some(t), Some(dur), sync)
                })
                .collect()
        };
        let data = |stream_type: u8| CodecConfig::Data {
            stream_type,
            descriptors: Vec::new(),
            carriage: DataCarriage::Pes,
        };
        let tracks = vec![
            Track::new(
                TrackSpec::new(TrackId::new(1), 90_000, data(0x04)),
                samples(head_aus, 3, true),
            ),
            Track::new(
                TrackSpec::new(TrackId::new(2), 90_000, data(0x7F)),
                samples(600, 32, true),
            ),
            Track::new(
                TrackSpec::new(TrackId::new(3), 90_000, data(0x7E)),
                samples(600, 32, true),
            ),
        ];
        transmux::TsMux::default()
            .package(&Media::new(tracks, 90_000))
            .expect("mux to TS")
    }

    /// **Round-3 audit A, end to end.** A stalled non-video head (never
    /// resolves) with two healthy PIDs behind it: the demuxer steps over the
    /// head after 512 access units and, because the stepped-over head counts
    /// toward `TracksResolved`, `ProgramTracker` announces the programme
    /// (`NewProgram` with the two healthy tracks) and their samples reach the
    /// session. Before the fix `TracksResolved` never fired, so no
    /// `NewProgram` was emitted and every sample was dropped by the
    /// `track_program` guard.
    #[test]
    fn stalled_head_does_not_hide_the_programme_from_the_session() {
        use crate::ingress::IngestDriver;
        use test_support::{handshake, trunk_config};

        let ts = stalled_head_ts_bytes();
        let mut session = TsIngestSession::new();
        session
            .feed(&ts, Timestamp::ZERO)
            .expect("TS feed is infallible");
        let mut programs = Vec::new();
        let mut samples = std::collections::BTreeSet::new();
        while let Some(ev) = session.poll() {
            match ev {
                SessionEvent::NewProgram { tracks, .. } => programs.push(tracks),
                SessionEvent::Sample { track_id, .. } => {
                    samples.insert(track_id);
                }
                _ => {}
            }
        }
        assert_eq!(programs.len(), 1, "the programme is announced");
        let mut ids: Vec<TrackId> = programs[0].iter().map(|t| t.track_id).collect();
        ids.sort();
        assert_eq!(ids, vec![TrackId::new(1), TrackId::new(2)]);
        assert_eq!(
            samples,
            [TrackId::new(1), TrackId::new(2)].into_iter().collect(),
            "both healthy tracks' samples flow"
        );

        // And through the driver: the programme's Trunk exists.
        let mut driver = IngestDriver::new(
            TsIngestSession::new(),
            crate::IngestConfig::new(handshake()).with_trunk(trunk_config()),
        );
        driver.feed(&ts, Timestamp::ZERO);
        assert!(
            driver.trunk(ProgramId(0)).is_some(),
            "the driver minted the programme's Trunk"
        );
    }

    /// Per track: `(bytes, dts, pts)` of each sample, in emission order.
    type TrackSamples =
        std::collections::BTreeMap<TrackId, Vec<(Vec<u8>, Option<i64>, Option<i64>)>>;

    /// Per-track `(count, first sample bytes, timestamps)` from the raw
    /// demux output, in emission order, plus how many samples the demux
    /// emitted *before* its first `TracksResolved`.
    fn raw_demux_samples(ts: &[u8], finish: bool) -> (TrackSamples, usize) {
        let mut demux = StreamingTsDemux::new();
        demux.feed(ts);
        if finish {
            demux.finish();
        }
        let mut by_track: std::collections::BTreeMap<_, Vec<_>> = Default::default();
        let (mut resolved, mut before) = (false, 0usize);
        while let Some(ev) = demux.poll_event() {
            match ev {
                DemuxEvent::TracksResolved { .. } => resolved = true,
                DemuxEvent::Sample {
                    track_id, sample, ..
                } => {
                    if !resolved {
                        before += 1;
                    }
                    by_track.entry(track_id).or_default().push((
                        sample.data.to_vec(),
                        sample.dts,
                        sample.pts,
                    ));
                }
                _ => {}
            }
        }
        (by_track, before)
    }

    /// What the session publishes, per track, asserting the contract on the
    /// way: a `Sample` never precedes its program's `NewProgram`, and a
    /// sample's track is one the program announced.
    fn session_samples(ts: &[u8], finish: bool) -> TrackSamples {
        let mut session = TsIngestSession::new();
        session.feed(ts, Timestamp::ZERO).expect("infallible");
        if finish {
            session.finish().expect("infallible");
        }
        let mut announced: HashMap<ProgramId, HashSet<TrackId>> = HashMap::new();
        let mut by_track: std::collections::BTreeMap<_, Vec<_>> = Default::default();
        while let Some(ev) = session.poll() {
            match ev {
                SessionEvent::NewProgram { program, tracks } => {
                    announced.insert(program, tracks.iter().map(|t| t.track_id).collect());
                }
                SessionEvent::Sample {
                    program,
                    track_id,
                    sample,
                    ..
                } => {
                    assert!(
                        announced
                            .get(&program)
                            .is_some_and(|t| t.contains(&track_id)),
                        "a sample for {track_id:?} preceded its programme's announcement"
                    );
                    by_track.entry(track_id).or_default().push((
                        sample.data.to_vec(),
                        sample.dts,
                        sample.pts,
                    ));
                }
                _ => {}
            }
        }
        by_track
    }

    fn assert_session_equals_raw_demux(ts: &[u8]) {
        assert_session_equals_raw_demux_with(ts, false);
    }

    fn assert_session_equals_raw_demux_with(ts: &[u8], finish: bool) {
        let (raw, before_resolved) = raw_demux_samples(ts, finish);
        let published = session_samples(ts, finish);
        assert!(!raw.is_empty(), "fixture demuxed no samples");
        println!("samples demuxed before TracksResolved: {before_resolved}");
        for (track, expect) in &raw {
            let got = published.get(track).map_or(&[][..], |v| v.as_slice());
            assert_eq!(
                got.len(),
                expect.len(),
                "{track:?}: the session lost {} of {} samples (first-AU data loss at \
                 stream start; {before_resolved} samples were demuxed before TracksResolved)",
                expect.len() - got.len().min(expect.len()),
                expect.len()
            );
            assert_eq!(
                got,
                expect.as_slice(),
                "{track:?}: bytes/timestamps/order differ"
            );
        }
        assert_eq!(published.len(), raw.len(), "a track went missing");
    }

    /// **Stream-start data loss.** Samples the demux replays from its probe
    /// backlog when a track is promoted arrive *before* `TracksResolved`;
    /// the programme is only announced at `TracksResolved`, so the
    /// `track_program` guard dropped every one of them. Synthetic stream
    /// (a stalled head holds `TracksResolved` back for 512 access units
    /// while two healthy PIDs are promoted): every healthy track's samples
    /// must reach the session byte-identical, with the raw demux's
    /// timestamps and order.
    #[test]
    fn samples_before_tracks_resolved_reach_the_session_synthetic() {
        let ts = stalled_head_ts_bytes();
        let (_, before) = raw_demux_samples(&ts, false);
        assert!(before > 0, "premise: samples precede TracksResolved");
        assert_session_equals_raw_demux(&ts);
    }

    /// The same on the private France TNT multiplex capture (several tracks
    /// promoted ahead of the resolved signal).
    #[test]
    fn samples_before_tracks_resolved_reach_the_session_france_tnt() {
        let Some(path) =
            test_fixtures::require("ts/france-tnt-dvbt-20s.ts", "multimux backlog replay")
        else {
            return;
        };
        assert_session_equals_raw_demux(&std::fs::read(path).expect("read fixture"));
    }

    /// **End-of-input flush.** A stalled head with too few access units to be
    /// stepped over (100 < 512) never resolves mid-stream, so `TracksResolved`
    /// (and with it `NewProgram` and every sample) waits for the demux's
    /// `finish()`. `TsIngestSession::finish` used to be a no-op, so nothing was
    /// ever announced. After `finish()` the session must publish exactly what
    /// the raw demux emits after its own `finish()`.
    #[test]
    fn finish_announces_programme_for_a_silent_declared_pid_synthetic() {
        let ts = stalled_head_ts_bytes_n(100);
        // Before finish nothing is announced (the premise of the bug).
        let mut session = TsIngestSession::new();
        session.feed(&ts, Timestamp::ZERO).expect("infallible");
        let mut announced_before = false;
        while let Some(ev) = session.poll() {
            announced_before |= matches!(ev, SessionEvent::NewProgram { .. });
        }
        assert!(!announced_before, "premise: no NewProgram before finish()");
        let (raw, _) = raw_demux_samples(&ts, true);
        assert!(
            !raw.is_empty(),
            "premise: the demux emits media at finish()"
        );
        assert_session_equals_raw_demux_with(&ts, true);
    }

    /// `finish()` is idempotent: a second call queues no further event.
    #[test]
    fn finish_twice_emits_nothing_more() {
        let ts = stalled_head_ts_bytes_n(100);
        let mut session = TsIngestSession::new();
        session.feed(&ts, Timestamp::ZERO).expect("infallible");
        session.finish().expect("infallible");
        let mut first = 0usize;
        while session.poll().is_some() {
            first += 1;
        }
        assert!(first > 1, "finish() produced events");
        session.finish().expect("infallible");
        assert!(
            session.poll().is_none(),
            "a second finish() must not duplicate events"
        );
    }

    /// The real private capture whose declared PID 0x8D never carries data:
    /// `TracksResolved` never fires mid-stream, so everything waits for
    /// `finish()`.
    #[test]
    fn finish_announces_programme_for_the_zenith_capture() {
        let Some(path) = test_fixtures::require(
            "ts/france-tnt-m6-eac3-zenith.ts",
            "multimux end-of-input demux finish",
        ) else {
            return;
        };
        let ts = std::fs::read(path).expect("read fixture");
        assert_session_equals_raw_demux_with(&ts, true);
    }

    /// Unit form: samples for buffered (not yet announced) tracks are held
    /// and flushed in arrival order, interleaved across tracks as received,
    /// right after `NewProgram`; none is emitted before it; a later sample
    /// follows them. Samples for a never-announced track are still dropped.
    #[test]
    fn pre_resolution_samples_flush_in_order_after_new_program() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::sample(TrackId::new(1), sample_at(0x01)));
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(2))));
        tracker.handle(DemuxEvent::sample(TrackId::new(2), sample_at(0x02)));
        tracker.handle(DemuxEvent::sample(TrackId::new(1), sample_at(0x03)));
        tracker.handle(DemuxEvent::sample(TrackId::new(9), sample_at(0x09)));
        assert!(matches!(tracker.poll(), Some(SessionEvent::Established)));
        assert!(tracker.poll().is_none(), "nothing before the programme");
        tracker.handle(DemuxEvent::tracks_resolved(0));
        assert!(matches!(
            tracker.poll(),
            Some(SessionEvent::NewProgram { .. })
        ));
        let mut got = Vec::new();
        while let Some(ev) = tracker.poll() {
            match ev {
                SessionEvent::Sample {
                    track_id, sample, ..
                } => got.push((track_id, sample.data.to_vec())),
                other => panic!("unexpected {other:?}"),
            }
        }
        assert_eq!(
            got,
            vec![
                (TrackId::new(1), vec![0x65, 0x01]),
                (TrackId::new(2), vec![0x65, 0x02]),
                (TrackId::new(1), vec![0x65, 0x03]),
            ]
        );
        tracker.handle(DemuxEvent::sample(TrackId::new(2), sample_at(0x04)));
        assert!(matches!(tracker.poll(), Some(SessionEvent::Sample { .. })));
    }

    fn flushed_dts(tracker: &mut ProgramTracker) -> Vec<i64> {
        let mut out = Vec::new();
        while let Some(ev) = tracker.poll() {
            if let SessionEvent::Sample { sample, .. } = ev {
                out.push(sample.dts.expect("dts"));
            }
        }
        out
    }

    /// The hold is bounded: past `MAX_HELD_SAMPLES` the OLDEST samples are
    /// dropped and the newest survive, in order.
    #[test]
    fn held_backlog_is_bounded_by_count_dropping_oldest() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        let total = MAX_HELD_SAMPLES + 5;
        for i in 0..total {
            let s = Sample::new(vec![0x65], Some(i as i64), Some(i as i64), Some(1), true);
            tracker.handle(DemuxEvent::sample(TrackId::new(1), s));
        }
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let got = flushed_dts(&mut tracker);
        assert_eq!(got.len(), MAX_HELD_SAMPLES);
        assert_eq!(got[0], 5, "the 5 oldest were dropped");
        assert_eq!(*got.last().unwrap(), total as i64 - 1, "newest kept");
        assert!(got.windows(2).all(|w| w[0] + 1 == w[1]), "order preserved");
    }

    /// The byte cap bounds the hold independently of the sample count.
    #[test]
    fn held_backlog_is_bounded_by_bytes_dropping_oldest() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        const MIB: usize = 1024 * 1024;
        let total = MAX_HELD_BYTES / MIB + 3;
        for i in 0..total {
            let s = Sample::new(
                vec![0u8; MIB],
                Some(i as i64),
                Some(i as i64),
                Some(1),
                true,
            );
            tracker.handle(DemuxEvent::sample(TrackId::new(1), s));
        }
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let got = flushed_dts(&mut tracker);
        assert_eq!(got.len(), MAX_HELD_BYTES / MIB, "capped at the byte budget");
        assert_eq!(got[0], 3, "the 3 oldest were dropped");
        assert_eq!(*got.last().unwrap(), total as i64 - 1);
    }

    /// A copy of [`track_spec`] with a different coded size, so a config
    /// change is observable in the announced spec.
    fn track_spec_sized(track_id: TrackId, width: u16, height: u16) -> TrackSpec {
        let avc = transmux::avc_config_from_sprop("Z0IAKeKQFAe2AtwEBAaQeJEV,aM48gA==")
            .expect("valid sprop");
        TrackSpec::new(
            track_id,
            90_000,
            transmux::pipeline::CodecConfig::Avc {
                config: avc,
                width,
                height,
            },
        )
    }

    fn video_size(spec: &TrackSpec) -> (u16, u16) {
        match &spec.config {
            transmux::pipeline::CodecConfig::Avc { width, height, .. } => (*width, *height),
            other => panic!("expected an AVC track, got {other:?}"),
        }
    }

    /// Deferred item (#1083): a mid-stream config change on a live track
    /// (`DemuxEvent::TrackUpdated`) was ignored, so the route's init segment
    /// kept describing a stream that was no longer being sent. It now
    /// announces the program's complete, updated track set.
    #[test]
    fn track_updated_announces_the_new_config() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec_sized(
            TrackId::new(1),
            640,
            360,
        )));
        tracker.handle(DemuxEvent::TrackAdded(track_spec_sized(
            TrackId::new(2),
            0,
            0,
        )));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let _new_program = tracker.poll();

        tracker.handle(DemuxEvent::TrackUpdated(track_spec_sized(
            TrackId::new(1),
            1920,
            1080,
        )));
        match tracker.poll() {
            Some(SessionEvent::TracksChanged { program, tracks }) => {
                assert_eq!(program, ProgramId(0));
                assert_eq!(
                    tracks.iter().map(|t| t.track_id).collect::<Vec<_>>(),
                    vec![TrackId::new(1), TrackId::new(2)],
                    "the complete track set, in track-id order"
                );
                assert_eq!(video_size(&tracks[0]), (1920, 1080));
                assert_eq!(
                    video_size(&tracks[1]),
                    (0, 0),
                    "the other track is untouched"
                );
            }
            other => panic!("expected TracksChanged, got {other:?}"),
        }
        assert!(tracker.poll().is_none());

        // An update for a track nothing announced is ignored.
        tracker.handle(DemuxEvent::TrackUpdated(track_spec_sized(
            TrackId::new(9),
            1,
            1,
        )));
        assert!(tracker.poll().is_none());
    }

    /// Before the first `TracksResolved` the buffered spec is replaced, so
    /// the `NewProgram` carries the latest config.
    #[test]
    fn track_updated_before_resolution_replaces_the_buffered_spec() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec_sized(
            TrackId::new(1),
            640,
            360,
        )));
        tracker.handle(DemuxEvent::TrackUpdated(track_spec_sized(
            TrackId::new(1),
            1280,
            720,
        )));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        match tracker.poll() {
            Some(SessionEvent::NewProgram { tracks, .. }) => {
                assert_eq!(tracks.len(), 1);
                assert_eq!(video_size(&tracks[0]), (1280, 720));
            }
            other => panic!("expected NewProgram, got {other:?}"),
        }
    }

    /// A `TrackRemoved` (mid-stream PMT version bump, issue #774) stops
    /// routing that track's samples rather than forwarding stale media.
    ///
    /// MUTATION-CHECKED: delete the `self.track_program.remove(&track_id)`
    /// line and this test's final assertion fails — a `Sample` for the
    /// removed track would still be routed.
    #[test]
    fn removed_track_stops_routing_its_samples() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let _new_program = tracker.poll();

        // Before removal: the sample routes.
        tracker.handle(DemuxEvent::sample(TrackId::new(1), sample_at(0xAA)));
        assert!(matches!(tracker.poll(), Some(SessionEvent::Sample { .. })));

        tracker.handle(DemuxEvent::track_removed(
            TrackId::new(1),
            Default::default(),
        ));
        tracker.handle(DemuxEvent::sample(TrackId::new(1), sample_at(0xBB)));
        assert!(
            tracker.poll().is_none(),
            "a removed track's samples must no longer be routed"
        );
    }

    // --- MPTS tests (issue #906) ---

    /// Two distinct `program_number`s produce two distinct `ProgramId`s.
    #[test]
    fn mpts_two_programmes_produce_two_program_ids() {
        let mut tracker = ProgramTracker::new();
        // Program 1: track 1
        let mut spec1 = track_spec(TrackId::new(1));
        spec1.program_number = Some(ProgramNumber::new(100));
        tracker.handle(DemuxEvent::TrackAdded(spec1));
        // Program 2: track 2
        let mut spec2 = track_spec(TrackId::new(2));
        spec2.program_number = Some(ProgramNumber::new(200));
        tracker.handle(DemuxEvent::TrackAdded(spec2));
        // Resolve
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let mut new_programs = Vec::new();
        while let Some(event) = tracker.poll() {
            if let SessionEvent::NewProgram { program, tracks } = event {
                new_programs.push((program, tracks.len()));
            }
        }
        assert_eq!(new_programs.len(), 2, "MPTS: two distinct programmes");
        assert_ne!(new_programs[0].0, new_programs[1].0, "distinct ProgramIds");
        assert_eq!(new_programs[0].1, 1, "each programme has one track");
        assert_eq!(new_programs[1].1, 1, "each programme has one track");
    }

    /// A mid-stream `TrackAdded` with a specific `program_number` joins
    /// that programme's `TracksChanged`, not a new one.
    #[test]
    fn mpts_mid_stream_track_joins_correct_programme() {
        let mut tracker = ProgramTracker::new();
        // Program 100: track 1
        let mut spec1 = track_spec(TrackId::new(1));
        spec1.program_number = Some(ProgramNumber::new(100));
        tracker.handle(DemuxEvent::TrackAdded(spec1));
        // Program 200: track 2
        let mut spec2 = track_spec(TrackId::new(2));
        spec2.program_number = Some(ProgramNumber::new(200));
        tracker.handle(DemuxEvent::TrackAdded(spec2));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        let mut programs: HashMap<ProgramNumber, ProgramId> = HashMap::new();
        while let Some(event) = tracker.poll() {
            if let SessionEvent::NewProgram { program, tracks } = event
                && let Some(t) = tracks.first()
                && let Some(pn) = t.program_number
            {
                programs.insert(pn, program);
            }
        }

        // Mid-stream: add track 3 to programme 100
        let mut spec3 = track_spec(TrackId::new(3));
        spec3.program_number = Some(ProgramNumber::new(100));
        tracker.handle(DemuxEvent::TrackAdded(spec3));
        tracker.handle(DemuxEvent::tracks_resolved(1));

        // TracksChanged for programme 100 must include both tracks;
        // programme 200 must NOT get a spurious TracksChanged.
        let prog100 = programs[&ProgramNumber::new(100)];
        let prog200 = programs[&ProgramNumber::new(200)];
        let mut saw_tracks_changed_100 = false;
        while let Some(event) = tracker.poll() {
            if let SessionEvent::TracksChanged { program, tracks } = event {
                assert_ne!(
                    program, prog200,
                    "programme 200 must NOT get a spurious TracksChanged"
                );
                if program == prog100 {
                    assert_eq!(tracks.len(), 2, "both tracks in programme 100");
                    let ids: Vec<TrackId> = tracks.iter().map(|t| t.track_id).collect();
                    assert!(ids.contains(&TrackId::new(1)));
                    assert!(ids.contains(&TrackId::new(3)));
                    saw_tracks_changed_100 = true;
                }
            }
        }
        assert!(
            saw_tracks_changed_100,
            "must see TracksChanged for programme 100"
        );

        // Sample for track 3 routes to programme 100
        tracker.handle(DemuxEvent::sample(TrackId::new(3), sample_at(0xCC)));
        let sample_event = tracker.poll();
        match sample_event {
            Some(SessionEvent::Sample {
                program, track_id, ..
            }) => {
                assert_eq!(program, prog100);
                assert_eq!(track_id, TrackId::new(3));
            }
            other => panic!("expected Sample routed to programme 100, got {other:?}"),
        }
    }

    /// SPTS (`program_number: None`) works as before — all tracks
    /// collapse into one `ProgramId`.
    #[test]
    fn spts_none_programme_works_as_before() {
        let mut tracker = ProgramTracker::new();
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(1))));
        tracker.handle(DemuxEvent::TrackAdded(track_spec(TrackId::new(2))));
        tracker.handle(DemuxEvent::tracks_resolved(0));
        let _established = tracker.poll();
        match tracker.poll() {
            Some(SessionEvent::NewProgram { program, tracks }) => {
                assert_eq!(program, ProgramId(0));
                assert_eq!(tracks.len(), 2);
            }
            other => panic!("expected single NewProgram with 2 tracks, got {other:?}"),
        }

        // Mid-stream addition joins the same None programme
        let mut spec3 = track_spec(TrackId::new(3));
        spec3.program_number = None;
        tracker.handle(DemuxEvent::TrackAdded(spec3));
        tracker.handle(DemuxEvent::tracks_resolved(1));
        match tracker.poll() {
            Some(SessionEvent::TracksChanged { program, tracks }) => {
                assert_eq!(program, ProgramId(0));
                assert_eq!(tracks.len(), 3, "all 3 tracks in TracksChanged");
            }
            other => panic!("expected TracksChanged for SPTS, got {other:?}"),
        }
    }
}