transmux 0.30.0

//! Classic HLS with MPEG-2 TS media segments — RFC 8216 + ISO/IEC 13818-1.
//!
//! Where [`HlsPackager`](crate::media::HlsPackager) emits **CMAF-HLS** (fMP4
//! `.m4s` segments described by an `#EXT-X-MAP`-bearing media playlist), this
//! module emits **classic HLS**: MPEG-2 Transport Stream `.ts` media segments
//! plus an RFC 8216 media playlist referencing them. Each `.ts` segment is a
//! self-contained MPEG-2 TS (its own PAT + PMT then a keyframe-aligned PES),
//! so it is independently decodable — there is no init/`#EXT-X-MAP` segment, as
//! that is a CMAF-only concept (RFC 8216 §4.3.2.5 applies only to fMP4 media).
//!
//! # Behaviour
//!
//! [`TsHlsPackager`] segments the hub [`Media`] IR at keyframe boundaries on the
//! **anchor track** (the first video track, else the first track whose samples
//! can advance a clock — never a section-carried track, whose `duration` *and*
//! `dts` are always `None`; see
//! [`crate::segmenter::choose_anchor`]), cutting a new segment
//! on the first sync sample at or past the target duration — mirroring
//! [`Segmenter`](crate::segmenter::Segmenter)'s CMAF rule so every video segment
//! begins on a random-access point. Every track's samples are partitioned across
//! the segments by decode time, so the concatenation of all segments carries the
//! full input (no sample dropped, duplicated, or reordered). Each segment is then
//! muxed via the shared [`crate::ts_mux`] machinery, re-emitting PAT + PMT at its
//! start (ISO/IEC 13818-1 §2.4.4 PSI repetition — a receiver joining mid-stream
//! must find the PSI at each segment boundary).
//!
//! The playlist is a VOD media playlist: `#EXTM3U`, `#EXT-X-VERSION`,
//! `#EXT-X-TARGETDURATION` (≥ every `#EXTINF`), `#EXT-X-MEDIA-SEQUENCE`, one
//! `#EXTINF` + `.ts` URI per segment, and a trailing `#EXT-X-ENDLIST`
//! (RFC 8216 §4.3.3). `#EXT-X-DISCONTINUITY` (RFC 8216 §4.3.2.3) and
//! `#EXT-X-DISCONTINUITY-SEQUENCE` (RFC 8216 §4.3.3.3) are forwarded from the
//! underlying [`MediaPlaylist`] when the [`MediaSegment::discontinuous`] flag
//! is set; this packager itself produces a single continuous timeline from one
//! contiguous IR, so it does not set the flag on any generated segment.
//!
//! # Cut policy
//!
//! [`StreamingTsHlsSegmenter::with_cut_policy`] selects a
//! [`SegmentCutPolicy`]: the default `AnyRap` cuts on
//! any sync sample; `IdrOnly` cuts only on an IDR picture and drops leading
//! access units until the first one (Apple HLS Authoring Spec 7.4,
//! `docs/hls-apple-conformance/`).
//!
//! # Streaming (live) input — [`StreamingTsHlsSegmenter`]
//!
//! [`TsHlsPackager::package`] is batch: it needs the whole [`Media`] up front.
//! For a live, unbounded feed there is no whole-input IR — [`StreamingTsHlsSegmenter`]
//! is the incremental analogue, mirroring [`Segmenter`](crate::segmenter::Segmenter)'s
//! CMAF push/flush model: [`StreamingTsHlsSegmenter::push`] buffers one coded
//! sample at a time and returns a finished `.ts` [`TsSegment`] whenever the
//! anchor track crosses a keyframe past the target duration;
//! [`StreamingTsHlsSegmenter::finish`] flushes the trailing partial segment;
//! [`StreamingTsHlsSegmenter::playlist`] renders a rolling media playlist over
//! a configurable sliding window, advancing `#EXT-X-MEDIA-SEQUENCE` and
//! `#EXT-X-DISCONTINUITY-SEQUENCE` as older segments roll off (RFC 8216
//! §6.2.1 defines a segment's Discontinuity Sequence Number as the header
//! value plus the `#EXT-X-DISCONTINUITY` tags still visible ahead of it in
//! the *current* playlist, so the header only needs to advance once a
//! discontinuous segment's own inline tag has rolled out of the window), and
//! omitting `#EXT-X-ENDLIST` until `finish` has been called. Both types share the same
//! anchor-selection, cut-decision, duration-accounting and sample-placement
//! logic — [`crate::segmenter::choose_anchor`] and
//! [`crate::segmenter::MediaClock`] (shared with the other three
//! segmenters), plus this module's private `is_cut_point`/
//! `segment_duration_secs`/`placement_secs` helpers — so they can never drift
//! apart.
//!
//! # Spec
//!
//! - **HLS media playlist**: RFC 8216 §4.1 / §4.3.2 (`#EXTINF`) / §4.3.3
//!   (`#EXT-X-TARGETDURATION`, `#EXT-X-MEDIA-SEQUENCE`, `#EXT-X-ENDLIST`).
//! - **TS segment constraints (PAT/PMT at each segment start)**:
//!   ISO/IEC 13818-1 §2.4.4 — PSI is periodically repeated so any access point
//!   is self-describing.

use crate::TrackId;
use alloc::collections::VecDeque;
use alloc::format;
use alloc::string::String;
use alloc::vec;
use alloc::vec::Vec;

use broadcast_common::{Demand, Package, Stage, Timestamp};
use broadcast_hls::{DecimalSeconds, MediaPlaylist, MediaSegment};

use crate::error::{Error, Result};
use crate::media::{Media, Track};
use crate::nal::{NalCodec, access_unit_is_idr};
use crate::pipeline::{CodecConfig, Sample, TrackSpec};
use crate::segment_cut::SegmentCutPolicy;
use crate::segmenter::{
    MAX_PENDING_SAMPLES_PER_TRACK, MediaClock, choose_anchor, no_sync_sample_error,
};
use crate::ts_mux::{TsContinuity, mux_tracks_at_continuing};

/// Outcome of [`StreamingTsHlsSegmenter::update_track`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum UpdateTrackOutcome {
    /// A discontinuity-triggering field differed; a discontinuity was armed.
    Changed,
    /// No discontinuity-triggering field differed; the stored spec was still
    /// refreshed.
    Unchanged,
    /// No track with that `track_id` exists; nothing was modified.
    Unknown,
}

/// Default `#EXT-X-VERSION` for a classic (TS-segment) media playlist. Version 3
/// is the floor for floating-point `#EXTINF` durations (RFC 8216 §7).
const DEFAULT_HLS_VERSION: u8 = 3;

// Test-only count of `MediaClock::tick` calls `TsHlsPackager::package` makes on
// the *current thread*.
//
// The batch packager must touch each sample a bounded number of times; before
// audit r05-W33 it re-walked every earlier sample of every track once per
// segment (`O(segments x samples)`). Instrumenting the real walk keeps the
// regression test timing-independent: it asserts a literal call count, not a
// duration.
//
// Thread-local rather than a process-wide static: the test harness runs tests
// on several threads, and another `#[test]` that happens to call `package`
// concurrently would otherwise inflate this counter and flake the `==`
// assertion.
#[cfg(test)]
std::thread_local! {
    pub(crate) static PACKAGE_TICK_CALLS: core::cell::Cell<usize> =
        const { core::cell::Cell::new(0) };
}

/// Increment the thread-local tick counter (a no-op outside tests).
#[cfg(test)]
#[inline]
fn count_tick() {
    PACKAGE_TICK_CALLS.with(|c| c.set(c.get() + 1));
}

/// The output of [`TsHlsPackager`]: the `.ts` media segments plus the media
/// playlist referencing them.
///
/// `segments[i]` is the whole-packet MPEG-2 TS bytes of the segment named by the
/// *i*-th `#EXTINF`/URI pair in `playlist`; the URIs are `{prefix}{i}.ts`.
#[derive(Debug, Clone)]
pub struct TsHlsOutput {
    /// The `.ts` media segments, in playlist order. Each is a whole number of
    /// 188-byte TS packets and opens with a PAT + PMT.
    pub segments: Vec<Vec<u8>>,
    /// The RFC 8216 media playlist (`#EXTM3U` … `#EXT-X-ENDLIST`) referencing
    /// `segments` by URI.
    pub playlist: String,
}

/// Package a hub [`Media`] IR into classic HLS: keyframe-aligned MPEG-2 TS
/// `.ts` media segments + an RFC 8216 media playlist.
///
/// Construct with [`TsHlsPackager::new`] or [`TsHlsPackager::default`], then call
/// [`Package::package`]. The target segment duration is a whole number of seconds
/// (integer arithmetic, `no_std`-friendly).
///
/// ```
/// use broadcast_common::{Package, Unpackage};
/// use transmux::{TsDemux, TsHlsPackager};
/// # fn ts_bytes() -> Vec<u8> { std::fs::read(test_fixtures::path("ts/h264-aac.ts")).unwrap() }
/// let ir = TsDemux::new().unpackage(&ts_bytes()).unwrap();
/// let out = TsHlsPackager::new(1).package(&ir).unwrap();
/// assert!(out.playlist.starts_with("#EXTM3U"));
/// assert!(out.segments.iter().all(|s| s.len() % 188 == 0));
/// ```
#[derive(Debug, Clone)]
pub struct TsHlsPackager {
    /// Target segment duration in whole seconds. Segments are cut on the first
    /// anchor-track keyframe at or past this many seconds of buffered anchor
    /// media (so the actual duration may be slightly longer, keyframe-aligned).
    pub target_secs: u32,
    /// `#EXT-X-VERSION` written to the playlist.
    pub version: u8,
    /// `#EXT-X-MEDIA-SEQUENCE` of the first segment.
    pub media_sequence: u64,
    /// URI prefix for the generated `.ts` segment entries (`{prefix}{i}.ts`).
    pub uri_prefix: String,
}

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

impl TsHlsPackager {
    /// Create a packager targeting `target_secs`-second segments (clamped to at
    /// least 1 second), with the default HLS version, media sequence 0, and the
    /// `"seg"` URI prefix.
    pub fn new(target_secs: u32) -> Self {
        Self {
            target_secs: target_secs.max(1),
            version: DEFAULT_HLS_VERSION,
            media_sequence: 0,
            uri_prefix: String::from("seg"),
        }
    }
}

/// One segment's per-track sample ranges, expressed as `[start, end)` indices
/// into each track's `samples`. `ranges[t]` is the half-open range for
/// `tracks[t]`; an empty range means that track contributes no sample here.
struct SegmentRanges {
    ranges: Vec<core::ops::Range<usize>>,
}

impl Package for TsHlsPackager {
    type Media = Media;
    type Output = TsHlsOutput;
    type Error = Error;

    fn package(&mut self, media: &Media) -> Result<TsHlsOutput> {
        if media.tracks.is_empty() {
            return Err(Error::InvalidInput("cannot package a Media with no tracks"));
        }

        // Choose the anchor track: first video (AVC), else the first
        // anchor-capable track (one whose samples actually carry a
        // duration) — mirrors Segmenter's anchor selection, but errors
        // loudly rather than defaulting to track 0 when nothing qualifies
        // (issue B9).
        let anchor = choose_anchor(media.tracks.iter().map(|t| &t.spec.config))?;

        let target_ticks = self.anchor_target_ticks(&media.tracks[anchor]);
        let boundaries = anchor_segment_boundaries(&media.tracks[anchor].samples, target_ticks);
        let segments = partition_tracks(&media.tracks, anchor, &boundaries);

        // Mux each segment independently: each re-emits PAT/PMT then its PES.
        // One `TsContinuity` spans every segment: without an
        // `#EXT-X-DISCONTINUITY` between them the segments concatenate into one
        // transport stream, so each PID's continuity_counter must keep counting
        // (§2.4.3.3) instead of restarting at 0 (audit r05-W21).
        let mut continuity = TsContinuity::new();
        let mut ts_segments: Vec<Vec<u8>> = Vec::with_capacity(segments.len());
        let mut playlist_segments: Vec<MediaSegment> = Vec::with_capacity(segments.len());
        let mut target_duration: u32 = 0;
        // Running per-track elapsed media time, so each segment's base DTS is
        // the value the previous segment left off at rather than a fresh O(n)
        // rescans of every earlier sample. The ranges are contiguous and in
        // order (`partition_tracks`), so walking the cursor forward over the
        // samples each segment consumes yields exactly the same cumulative sum
        // the prefix walk did. `MediaClock` (duration, else dts delta) rather
        // than a duration sum, matching the streaming path's `base_decode` — a
        // `duration: Some(0)` stream would otherwise pin every segment's base
        // DTS at 0. Each clock is carried across boundaries (it is deliberately
        // not reset there), so the first sample of a segment can still take its
        // dts delta from the last sample of the previous one.
        let mut clocks: Vec<MediaClock> = alloc::vec![MediaClock::new(); media.tracks.len()];
        let mut base_dts: Vec<u64> = alloc::vec![0u64; media.tracks.len()];
        // Per-track cursor into `samples`: contiguous, monotonically advancing
        // segment ranges let one forward walk both feed the clock and advance
        // the next segment's base, so a sample is ticked exactly once for the
        // whole package (`O(samples)`, not `O(segments × samples)`).
        let mut cursor: Vec<usize> = alloc::vec![0usize; media.tracks.len()];
        for (i, seg) in segments.iter().enumerate() {
            // Fold any samples this segment's window starts past into the
            // running base *before* `base_dts` is handed to the muxer.
            // `partition_tracks` can emit a zero-width range for a track whose
            // samples all fall after an early window, and it can re-claim
            // samples an earlier segment already counted (its last range is
            // written from a `seg_start_idx` an earlier write may have passed),
            // so the cursor advances monotonically and never backwards — a
            // sample is ticked exactly once for the whole package.
            for (ti, track) in media.tracks.iter().enumerate() {
                let start = seg.ranges[ti].start.max(cursor[ti]);
                while cursor[ti] < start {
                    #[cfg(test)]
                    count_tick();
                    base_dts[ti] += clocks[ti].tick(&track.samples[cursor[ti]]);
                    cursor[ti] += 1;
                }
            }
            let sample_slices: Vec<&[Sample]> = media
                .tracks
                .iter()
                .zip(&seg.ranges)
                .map(|(t, r)| &t.samples[r.clone()])
                .collect();
            let bytes = mux_tracks_at_continuing(
                &media.tracks,
                &sample_slices,
                &base_dts,
                &mut continuity,
            )?;
            ts_segments.push(bytes);

            // Walk this segment's samples exactly once: the running clock
            // yields each sample's elapsed media time, which both advances the
            // track's base DTS for the next segment and gives the segment's own
            // buffered duration. `MediaClock` (duration, else dts delta) rather
            // than a duration sum, matching the streaming path's `base_decode`
            // — a `duration: Some(0)` stream would otherwise pin every
            // segment's base DTS at 0 — and the clock deliberately carries
            // across boundaries, so the first sample of a segment still takes
            // its dts delta from the last sample of the previous one.
            // Each track is walked at most once overall: `cursor[ti]` is the
            // first sample not yet ticked, and the loop ticks exactly the
            // samples this segment owns that the cursor has not passed. A
            // segment's own duration is the ticks of the samples in its range;
            // the running `base_dts` is the total ticked so far, which is what
            // the *next* segment's samples start from. `partition_tracks`
            // ranges are non-decreasing and, together, cover every sample, so
            // the cursor advances monotonically.
            let mut ticks: Vec<u64> = alloc::vec![0u64; media.tracks.len()];
            for (ti, track) in media.tracks.iter().enumerate() {
                let range = seg.ranges[ti].clone();
                // This segment's own duration is the ticks of the samples in
                // its range, minus any the cursor already passed (an
                // overlapping range re-claims samples an earlier segment
                // counted, and counting them twice would inflate #EXTINF).
                let mut segment_ticks: u64 = 0;
                for idx in cursor[ti].max(range.start)..range.end {
                    #[cfg(test)]
                    count_tick();
                    segment_ticks += clocks[ti].tick(&track.samples[idx]);
                }
                base_dts[ti] += segment_ticks;
                cursor[ti] = cursor[ti].max(range.end);
                ticks[ti] = segment_ticks;
            }
            let anchor_ticks = ticks[anchor];
            let ts_scale = media.tracks[anchor].spec.timescale.max(1) as u64;
            // #EXT-X-TARGETDURATION is an integer ≥ every #EXTINF (RFC 8216
            // §4.3.3.1: the rounded max segment duration).
            let (duration, ceil_secs) = segment_duration_secs(anchor_ticks, ts_scale);
            if ceil_secs > target_duration {
                target_duration = ceil_secs;
            }
            playlist_segments.push(MediaSegment {
                uri: format!("{}{}.ts", self.uri_prefix, i),
                // Same proof as the streaming path above: finite, >= 0.
                duration: DecimalSeconds::new(duration)
                    .expect("segment_duration_secs is finite, >= 0"),
                discontinuous: false,
                parts: vec![],
                ..Default::default()
            });
        }

        let playlist = MediaPlaylist {
            version: self.version,
            target_duration: target_duration.max(1),
            media_sequence: self.media_sequence,
            discontinuity_sequence: 0,
            segments: playlist_segments,
            endlist: true,
            extra_tags: vec![],
            low_latency: None,
            iframes_only: false,
            open_segment: None,
            ..Default::default()
        }
        .to_m3u8()?;

        Ok(TsHlsOutput {
            segments: ts_segments,
            playlist,
        })
    }
}

impl TsHlsPackager {
    /// Convert the configured `target_secs` into anchor-track timescale ticks
    /// (never zero, so a segment can always close).
    fn anchor_target_ticks(&self, anchor: &Track) -> u64 {
        target_ticks_for(self.target_secs, anchor.spec.timescale)
    }
}

// ── Shared segment-cutting logic (batch + streaming) ────────────────────────
//
// [`TsHlsPackager`] (batch) and [`StreamingTsHlsSegmenter`] (incremental) both
// cut segments on the same anchor-track keyframe rule and share the exact same
// duration/anchor-selection arithmetic below, so the two paths can never
// silently drift apart (issue #571).

// Anchor selection (`is_anchor_capable`/`choose_anchor`) and the anchor
// progress clock (`MediaClock`) are the *shared* implementations in
// [`crate::segmenter`] — this module used to carry its own copies, so the four
// segmenters could drift apart on which track may be the anchor. See
// [`choose_anchor`](crate::segmenter::choose_anchor) for the rule and why a
// section-only track set is a construction error rather than a silent stall.

/// Decode-start time of one sample of a **non-anchor** track, in seconds on
/// the segmentation timeline (`0.0` = the anchor's first sample), used to
/// decide which segment that sample belongs to.
///
/// Normally the running per-track accumulator `acc_ticks` (advanced by
/// [`MediaClock`]). A sample carrying **no `duration` of its own but an
/// absolute `dts`** is instead placed by that `dts`, relative to
/// `origin_secs` (the anchor's first sample's decode time in seconds): that is
/// exactly a section-carried sample (ISO/IEC 13818-1 §2.4.4 — an SCTE-35
/// `splice_info_section`, say), for which the accumulator can never advance,
/// so without this rule *every* section sample would be placed at time `0.0`
/// and land in segment 0 no matter how late it actually occurs.
///
/// Both the batch ([`partition_tracks`]) and streaming
/// ([`StreamingTsHlsSegmenter::cut_segment`]) partitions call this, so the two
/// place a given sample in the same segment (the equivalence this module's
/// docs claim, and `tests/it/streaming_tshls.rs` enforces).
///
/// A section sample carrying *neither* a duration nor a dts has no time
/// information anywhere in the IR; it keeps the accumulator's value, i.e. it
/// stays with the samples around it.
fn placement_secs(sample: &Sample, acc_ticks: u64, timescale: u64, origin_secs: f64) -> f64 {
    let timescale = timescale.max(1) as f64;
    match (sample.duration, sample.dts) {
        (None, Some(dts)) => (dts.max(0) as f64 / timescale) - origin_secs,
        _ => acc_ticks as f64 / timescale,
    }
}

/// The segmentation timeline's origin in seconds: the anchor track's first
/// sample's absolute decode time, or `0.0` when the anchor carries no `dts`
/// (segment start times are then relative to the anchor's first sample, which
/// is the same thing).
fn timeline_origin_secs(anchor_first_dts: Option<i64>, anchor_scale: u64) -> f64 {
    anchor_first_dts.map_or(0.0, |d| d.max(0) as f64 / anchor_scale.max(1) as f64)
}

/// Convert a whole-seconds target duration into anchor-track timescale ticks
/// (never zero, so a segment can always close).
fn target_ticks_for(target_secs: u32, ts_scale: u32) -> u64 {
    (target_secs.max(1) as u64) * (ts_scale.max(1) as u64)
}

/// Segment duration in seconds, plus the RFC 8216 `#EXT-X-TARGETDURATION`
/// ceiling (whole seconds, rounded up) for `anchor_ticks` of buffered anchor
/// media at `ts_scale` ticks/second.
fn segment_duration_secs(anchor_ticks: u64, ts_scale: u64) -> (f64, u32) {
    let ts_scale = ts_scale.max(1);
    let duration = anchor_ticks as f64 / ts_scale as f64;
    let ceil_secs = anchor_ticks.div_ceil(ts_scale) as u32;
    (duration, ceil_secs)
}

/// True when the anchor should be cut *before* buffering the incoming sample:
/// the current segment already has content (`has_pending`), the sample is a
/// sync sample, and the buffered duration has reached the target. Shared by
/// the batch boundary scan ([`anchor_segment_boundaries`]) and
/// [`StreamingTsHlsSegmenter::push`]'s incremental cut decision.
fn is_cut_point(has_pending: bool, is_sync: bool, buffered_ticks: u64, target_ticks: u64) -> bool {
    has_pending && is_sync && buffered_ticks >= target_ticks
}

/// Compute the anchor-track sample indices at which a new segment starts.
///
/// Always includes `0`. A cut is made *before* a sync sample once the buffered
/// anchor duration since the last cut has reached `target_ticks` — the same rule
/// as [`Segmenter`](crate::segmenter::Segmenter), so every segment begins on a
/// keyframe (random-access point). Returns the ascending list of start indices.
///
/// The buffered duration is accumulated by [`MediaClock`]: each sample's own
/// `duration` when that is a real, non-zero span, else the `dts` delta from
/// the previous sample. Requiring `duration` alone stalled this scan on the
/// `duration: Some(0)` samples live input legitimately produces (an RTMP
/// publish's first tag, or any two FLV tags sharing a timestamp), so **no**
/// boundary was ever found and the whole stream became one segment.
fn anchor_segment_boundaries(samples: &[Sample], target_ticks: u64) -> Vec<usize> {
    let mut starts = vec![0usize];
    if samples.is_empty() {
        return starts;
    }
    let mut clock = MediaClock::new();
    let mut buffered: u64 = 0;
    for (i, s) in samples.iter().enumerate() {
        // Cut before this sample when it is a keyframe past the target and it is
        // not the very first sample (the leading segment already starts at 0).
        if is_cut_point(i > 0, s.flags.is_sync, buffered, target_ticks) {
            starts.push(i);
            buffered = 0;
        }
        buffered += clock.tick(s);
    }
    starts
}

/// Partition every track's samples into per-segment index ranges.
///
/// The anchor track is split exactly at `anchor_boundaries`. Non-anchor tracks
/// (e.g. audio) are split by decode time: each sample is assigned to the segment
/// whose anchor-time window `[seg_start_time, next_seg_start_time)` contains the
/// sample's decode-start time — [`placement_secs`], the *same* rule
/// [`StreamingTsHlsSegmenter::cut_segment`] applies incrementally — computed in
/// seconds so tracks with different timescales align. Every sample lands in
/// exactly one segment, in order, so a concatenation of the segments reproduces
/// each track's full sample list.
fn partition_tracks(
    tracks: &[Track],
    anchor: usize,
    anchor_boundaries: &[usize],
) -> Vec<SegmentRanges> {
    let n_segs = anchor_boundaries.len();
    let anchor_samples = &tracks[anchor].samples;
    let anchor_scale = tracks[anchor].spec.timescale.max(1) as u64;
    let origin_secs =
        timeline_origin_secs(anchor_samples.first().and_then(|s| s.dts), anchor_scale);

    // Segment start times (seconds) on the anchor timeline; the last segment
    // runs to +∞ so it captures any trailing tail on longer tracks. Accumulated
    // by `MediaClock` (duration, else dts delta) for the same reason
    // `anchor_segment_boundaries` uses it: a plain duration sum freezes at 0 on
    // a `duration: Some(0)` stream and would place every non-anchor sample in
    // segment 0.
    let mut start_times: Vec<f64> = Vec::with_capacity(n_segs);
    {
        let mut clock = MediaClock::new();
        let mut acc: u64 = 0;
        let mut cursor = 0usize;
        for &b in anchor_boundaries {
            while cursor < b {
                acc += clock.tick(&anchor_samples[cursor]);
                cursor += 1;
            }
            start_times.push(acc as f64 / anchor_scale as f64);
        }
    }

    let mut out: Vec<SegmentRanges> = (0..n_segs)
        .map(|_| SegmentRanges {
            ranges: vec![0..0; tracks.len()],
        })
        .collect();

    for (t_idx, track) in tracks.iter().enumerate() {
        if t_idx == anchor {
            // Exact index split on the anchor.
            for (seg, &start) in anchor_boundaries.iter().enumerate() {
                let end = if seg + 1 < n_segs {
                    anchor_boundaries[seg + 1]
                } else {
                    anchor_samples.len()
                };
                out[seg].ranges[t_idx] = start..end;
            }
            continue;
        }

        // Time-based split for non-anchor tracks.
        let scale = track.spec.timescale.max(1) as u64;
        // For each segment, find the [start_idx, end_idx) of samples whose
        // decode-start time falls in this segment's window.
        let mut seg = 0usize;
        let mut seg_start_idx = 0usize;
        let mut acc_ticks: u64 = 0;
        let mut clock = MediaClock::new();
        for (i, s) in track.samples.iter().enumerate() {
            let start_time = placement_secs(s, acc_ticks, scale, origin_secs);
            // Advance to the segment whose window contains this sample. A sample
            // belongs to the last segment whose start_time ≤ this sample's time.
            while seg + 1 < n_segs && start_time >= start_times[seg + 1] {
                out[seg].ranges[t_idx] = seg_start_idx..i;
                seg += 1;
                seg_start_idx = i;
            }
            acc_ticks += clock.tick(s);
        }
        // Trailing samples belong to the current (last reached) segment.
        out[seg].ranges[t_idx] = seg_start_idx..track.samples.len();
    }

    out
}

// ── Streaming (incremental) TS-HLS segmentation for live input (issue #571) ─

/// One finished `.ts` media segment produced incrementally by
/// [`StreamingTsHlsSegmenter`] — the streaming analogue of one
/// [`TsHlsOutput::segments`] entry, plus the playlist metadata
/// [`StreamingTsHlsSegmenter::playlist`] needs to describe it.
#[derive(Debug, Clone)]
pub struct TsSegment {
    /// Whole-packet MPEG-2 TS bytes: PAT + PMT then the keyframe-aligned PES.
    /// Byte-identical to the corresponding segment [`TsHlsPackager::package`]
    /// would produce from the same samples in one batch call.
    pub bytes: Vec<u8>,
    /// Segment duration in seconds (the anchor track's buffered duration).
    pub duration: f64,
    /// `true` when this segment should be preceded by `#EXT-X-DISCONTINUITY`
    /// (RFC 8216 §4.3.2.3) — set via
    /// [`StreamingTsHlsSegmenter::mark_discontinuity`].
    pub discontinuous: bool,
    /// The playlist URI assigned to this segment (`"{uri_prefix}{sequence}.ts"`).
    pub uri: String,
    /// 0-based, ever-increasing sequence number of this segment. Stable for
    /// the life of the segmenter, independent of the rolling playlist window
    /// (unlike `#EXT-X-MEDIA-SEQUENCE`, which advances as segments roll off).
    pub sequence: u64,
}

/// Playlist-relevant metadata for one segment retained in the rolling window
/// — the segment bytes themselves are handed to the caller by
/// [`StreamingTsHlsSegmenter::push`]/[`finish`](StreamingTsHlsSegmenter::finish)
/// and are not kept here.
#[derive(Debug, Clone)]
struct WindowEntry {
    uri: String,
    duration: DecimalSeconds,
    discontinuous: bool,
}

/// Per-track accumulation state for [`StreamingTsHlsSegmenter`].
/// A cut the anchor triggered whose non-anchor audio has not yet caught up.
///
/// A live multiplex delivers a track's PES later than the video it
/// accompanies, by a jitter that varies per segment. Cutting the instant the
/// anchor reaches its next sync sample therefore closes each segment with
/// however much audio had arrived (zenith captures: 1.15 s to 1.34 s of audio
/// against 1.28 s of video), and Apple's `mediastreamvalidator` rejects the
/// playlist-vs-segment duration mismatch. The cut is held until every audio
/// track holds a sample at or past the boundary, bounded by
/// [`CATCH_UP_MAX_SECS`] of further anchor media.
#[derive(Debug, Clone, Copy)]
struct DeferredCut {
    /// Anchor samples (from the front of `pending`) that close the segment.
    anchor_split: usize,
    /// Anchor ticks those samples span: the segment's `EXTINF`.
    closed_anchor_dur: u64,
    /// The boundary: where the next segment starts, in seconds from the
    /// anchor's first sample.
    next_start_secs: f64,
}

/// Longest the streaming segmenter holds a cut for lagging audio, in seconds
/// of further anchor media. Beyond it the segment is cut with whatever audio
/// arrived, so a silent or stalled audio track cannot stop segments.
const CATCH_UP_MAX_SECS: u64 = 1;

struct StreamTrackState {
    spec: TrackSpec,
    /// Samples buffered so far that have not yet been flushed into a
    /// segment, in decode order.
    pending: Vec<Sample>,
    /// Decode time of `pending[0]`, in this track's media timescale ticks —
    /// the elapsed media time of every sample already flushed into an
    /// earlier segment. Mirrors `TrackState::base_decode` in
    /// [`Segmenter`](crate::segmenter::Segmenter).
    base_decode: u64,
    /// Advances `base_decode` past each sample as it is flushed, under the
    /// same duration-then-dts-delta rule as the anchor accumulator (see
    /// [`MediaClock`]). A plain `duration` sum would pin `base_decode` at 0
    /// forever on a `duration: Some(0)` stream, so every segment would be
    /// muxed at the same base DTS.
    flush_clock: MediaClock,
}

/// A stateful **streaming** classic-HLS segmenter — the incremental analogue
/// of [`TsHlsPackager`] for unbounded live input, mirroring how
/// [`Segmenter`](crate::segmenter::Segmenter) drives incremental CMAF
/// fragment production.
///
/// Build it from the same [`TrackSpec`]s (in the same order) as the source
/// `Media`'s tracks, `push` coded samples in decode order, and pull finished
/// `.ts` segments from `push`'s return value; `finish` emits the final
/// partial segment. [`Self::playlist`] renders the current rolling media
/// playlist over the configured window.
///
/// ```
/// use transmux::{CodecConfig, Sample, TrackSpec};
/// use transmux::ts_hls::StreamingTsHlsSegmenter;
/// # use transmux::{AVCConfigurationBox, AVCDecoderConfigurationRecord, AvcPps, AvcSps};
/// # fn spec() -> TrackSpec {
/// #     let record = AVCDecoderConfigurationRecord {
/// #         configuration_version: 1,
/// #         profile_indication: 66,
/// #         profile_compatibility: 0,
/// #         level_indication: 30,
/// #         length_size_minus_one: 3,
/// #         sps: vec![AvcSps(vec![0x67, 0x42, 0xc0, 0x1e, 0xd9, 0x00, 0x80, 0x1e, 0x24])],
/// #         pps: vec![AvcPps(vec![0x68, 0xce, 0x3c, 0x80])],
/// #         chroma_format: None,
/// #         bit_depth_luma_minus8: None,
/// #         bit_depth_chroma_minus8: None,
/// #         sps_ext: vec![],
/// #     };
/// #     TrackSpec::new(transmux::TrackId::new(1), 90_000, CodecConfig::Avc {
/// #         config: AVCConfigurationBox::new(record),
/// #         width: 16,
/// #         height: 16,
/// #     })
/// # }
/// # fn au(sync: bool) -> Sample {
/// #     use std::sync::atomic::{AtomicI64, Ordering};
/// #     static NEXT_DTS: AtomicI64 = AtomicI64::new(0);
/// #     let dts = NEXT_DTS.fetch_add(1000, Ordering::Relaxed);
/// #     Sample::new(vec![0u8; 4], Some(dts), Some(dts), Some(1000), sync)
/// # }
/// // 2 s target segments, keep the last 3 in the rolling playlist.
/// let mut seg = StreamingTsHlsSegmenter::new(vec![spec()], 2, 3).unwrap();
/// seg.push(transmux::TrackId::new(1), au(true)).unwrap();
/// seg.finish().unwrap();
/// let segments = seg.take_ready();          // write s.bytes for each
/// assert_eq!(segments.len(), 1);
/// assert_eq!(segments[0].bytes[0], 0x47);   // MPEG-TS sync byte
/// let playlist = seg.playlist().unwrap(); // rolling window, #EXT-X-ENDLIST after finish
/// assert!(playlist.contains("#EXT-X-ENDLIST"));
/// ```
pub struct StreamingTsHlsSegmenter {
    tracks: Vec<StreamTrackState>,
    /// Index into `tracks` of the segmentation anchor (keyframe cut boundary).
    anchor: usize,
    /// Target segment duration in whole seconds, as given to [`Self::new`] —
    /// retained (alongside the derived `target_ticks`) so
    /// [`Self::add_track`] can recompute `target_ticks` in the anchor-track's
    /// timescale if a late-added track becomes the new anchor (issue #624).
    target_secs: u32,
    /// Target segment duration in the *anchor track's* media timescale.
    target_ticks: u64,
    /// Buffered duration of the anchor's `pending` samples (media-timescale ticks).
    anchor_pending_dur: u64,
    /// A segment boundary the anchor reached while an audio track had not yet
    /// delivered the media up to it; see [`DeferredCut`].
    deferred_cut: Option<DeferredCut>,
    /// Anchor-progress clock: advances `anchor_pending_dur` from each anchor
    /// sample's `duration`, or from its `dts` delta when `duration` is absent
    /// or zero (see [`MediaClock`]).
    anchor_clock: MediaClock,
    /// The absolute decode time (seconds) of the anchor's **first** sample —
    /// the origin the time-based non-anchor split measures against, so a
    /// section sample's absolute `dts` can be compared with the
    /// relative-from-zero segment start times (see [`placement_secs`]).
    /// `None` until the first anchor sample is pushed.
    origin_secs: Option<f64>,
    /// Explicit discontinuity: when `true` the *next* cut is marked
    /// discontinuous. Reset to `false` after each cut.
    pending_discontinuity: bool,
    /// Which anchor pictures may open a segment ([`SegmentCutPolicy`]).
    cut_policy: SegmentCutPolicy,
    /// [`SegmentCutPolicy::IdrOnly`] only: no IDR has been seen since the
    /// session start or the last [`StreamingTsHlsSegmenter::mark_discontinuity`],
    /// so every sample is dropped until the anchor delivers one.
    awaiting_idr: bool,
    /// [`SegmentCutPolicy::IdrOnly`] only: the segment opened by the IDR that
    /// ends the current wait must be flagged discontinuous.
    discontinuity_at_resync: bool,
    /// Anchor access units dropped while awaiting an IDR.
    dropped_leading_aus: u64,
    /// Bound on the wait for the first IDR; `None` waits forever (strict).
    idr_wait: Option<crate::segment_cut::IdrWait>,
    /// [`SegmentCutPolicy::IdrOnly`] gave up waiting for an IDR and cuts on
    /// any sync sample for the rest of the stream.
    idr_fallback_active: bool,
    /// Set by [`Self::finish`]; makes [`Self::playlist`] emit `#EXT-X-ENDLIST`.
    finished: bool,
    /// `#EXT-X-VERSION` written to the playlist.
    pub version: u8,
    /// URI prefix for generated `.ts` segment entries (`"{prefix}{sequence}.ts"`).
    pub uri_prefix: String,
    /// Maximum number of segments retained in the rolling playlist window.
    window: usize,
    /// The segments currently in the rolling window, oldest first.
    window_segments: VecDeque<WindowEntry>,
    /// Cross-segment continuity-counter state (§2.4.3.3): the segments of one
    /// stream concatenate into a single transport stream, so each PID's
    /// `continuity_counter` must keep counting across cuts (audit r05-W21).
    continuity: TsContinuity,
    /// Total number of segments ever cut (also the sequence number of the
    /// next segment).
    total_segments: u64,
    /// `#EXT-X-DISCONTINUITY-SEQUENCE`: count of discontinuous segments that
    /// have already rolled out of the window. RFC 8216 §6.2.1 defines a
    /// segment's Discontinuity Sequence Number as this header value *plus*
    /// the number of `#EXT-X-DISCONTINUITY` tags still visible in the
    /// current playlist preceding that segment's URI line — so a
    /// discontinuous segment's own inline tag (rendered while it's still in
    /// the window) already accounts for its own boundary; the header must
    /// only advance once that segment (and its tag) has rolled off, or every
    /// segment still in the window would be double-counted.
    discontinuity_sequence: u64,
    /// Running max `#EXT-X-TARGETDURATION` ceiling ever produced (monotonic,
    /// per RFC 8216 §4.3.3.1 — never shrinks even as segments roll off).
    target_duration: u32,
    /// Segments cut but not yet handed to the caller — the **single** source
    /// of truth for every retrieval path: the inherent [`Self::take_ready`]
    /// drain, and [`Stage::poll`]. Every cut lands here and *only* here (issue
    /// R2); neither [`push`](Self::push) nor [`finish`](Self::finish) returns
    /// a segment inline any more, so there is no value a caller can silently
    /// drop.
    ///
    /// This used to also be handed back inline from `push`/`finish`
    /// (`Result<Option<TsSegment>>`), alongside this same queue. That was
    /// reachable data loss, not just a caller-discipline note: because
    /// `finish` is both an inherent method *and* the [`Stage::finish`] trait
    /// method with the same name and arity, an unqualified `seg.finish()` call
    /// always resolves to the inherent one (inherent-over-trait method
    /// resolution) regardless of whether the caller is otherwise driving this
    /// type purely through [`Stage`] (`Stage::feed` + `Stage::poll`). Such a
    /// caller has no reason to inspect the `Result<()>` a `Stage`-style
    /// `finish` implies, so the old inherent `finish`'s popped-and-returned
    /// trailing segment was dropped on the floor the moment that expression's
    /// value went unused — silently, with no error, and a subsequent
    /// `Stage::poll` finding the queue already empty. See
    /// `finish_bare_call_does_not_lose_the_trailing_segment_to_a_poll_driver`
    /// for the reproduction. Removing the inline return removes the value
    /// there was anything to drop: `push`/`finish` now only ever return
    /// `Result<()>`, and the *only* ways to retrieve a cut segment are
    /// [`Self::take_ready`] and [`Stage::poll`], both draining this one queue.
    ready: VecDeque<TsSegment>,
}

fn audio_discontinuity_trigger(old: &CodecConfig, new: &CodecConfig) -> bool {
    if core::mem::discriminant(old) != core::mem::discriminant(new) {
        return false;
    }
    if old.audio_sample_rate() != new.audio_sample_rate()
        || old.audio_channel_field() != new.audio_channel_field()
    {
        return true;
    }
    match (old, new) {
        (CodecConfig::Aac { esds: e1, .. }, CodecConfig::Aac { esds: e2, .. }) => {
            let asc = |e: &crate::mp4esds::EsdsBox| {
                e.es_descriptor
                    .decoder_config
                    .as_ref()
                    .and_then(|dc| dc.decoder_specific_info.as_ref())
                    .map(|d| d.data.clone())
            };
            asc(e1) != asc(e2)
        }
        _ => false,
    }
}

impl StreamingTsHlsSegmenter {
    /// Create a streaming segmenter for `tracks`, cutting segments roughly
    /// every `target_secs` seconds (clamped to at least 1) on the anchor
    /// track's keyframes, and keeping at most `window` segments in the
    /// rolling media playlist returned by [`Self::playlist`].
    ///
    /// The anchor is the first **video** track (any codec), else the first
    /// anchor-capable track — [`crate::segmenter::choose_anchor`],
    /// the same rule [`TsHlsPackager`] and the other three segmenters use.
    /// `tracks` must be given in the
    /// same order the caller will later `push` matching `track_id`s, and in
    /// the same order as the source `Media`'s tracks, so the muxed PID/PMT
    /// layout matches [`TsHlsPackager::package`] exactly (needed for the two
    /// paths to produce byte-identical segments from the same input).
    ///
    /// # Errors
    /// [`Error::InvalidInput`] if `tracks` is empty, has duplicate
    /// `track_id`s, `window` is `0`, or no track is anchor-capable (e.g.
    /// every track is a section-carried `CodecConfig::Data` track, whose
    /// samples are always `duration: None`).
    pub fn new(tracks: Vec<TrackSpec>, target_secs: u32, window: usize) -> Result<Self> {
        if tracks.is_empty() {
            return Err(Error::InvalidInput(
                "streaming ts-hls segmenter needs at least one track",
            ));
        }
        if window == 0 {
            return Err(Error::InvalidInput("window must be >= 1"));
        }
        for (i, a) in tracks.iter().enumerate() {
            if tracks[i + 1..].iter().any(|b| b.track_id == a.track_id) {
                return Err(Error::InvalidInput("duplicate track_id"));
            }
        }

        let anchor = choose_anchor(tracks.iter().map(|t| &t.config))?;
        let target_secs = target_secs.max(1);
        let target_ticks = target_ticks_for(target_secs, tracks[anchor].timescale);

        let tracks = tracks
            .into_iter()
            .map(|spec| StreamTrackState {
                spec,
                pending: Vec::new(),
                base_decode: 0,
                flush_clock: MediaClock::new(),
            })
            .collect();

        Ok(Self {
            tracks,
            anchor,
            target_secs,
            target_ticks,
            anchor_pending_dur: 0,
            deferred_cut: None,
            anchor_clock: MediaClock::new(),
            origin_secs: None,
            pending_discontinuity: false,
            cut_policy: SegmentCutPolicy::AnyRap,
            awaiting_idr: false,
            discontinuity_at_resync: false,
            dropped_leading_aus: 0,
            idr_wait: None,
            idr_fallback_active: false,
            finished: false,
            version: DEFAULT_HLS_VERSION,
            uri_prefix: String::from("seg"),
            window,
            window_segments: VecDeque::new(),
            continuity: TsContinuity::new(),
            total_segments: 0,
            discontinuity_sequence: 0,
            target_duration: 0,
            ready: VecDeque::new(),
        })
    }

    /// Build a streaming TS-HLS segmenter whose segment numbering starts
    /// from `next_sequence` (0-based, matching [`TsSegment::sequence`]'s own
    /// convention) rather than [`Self::new`]'s implicit `0` — the classic-TS
    /// analogue of [`crate::ll_hls::LlHlsSegmenter::with_part_target_at`]: a
    /// segmenter rebuilt after a track-set change, or built against a
    /// `Trunk` that already holds segments from an earlier `SegmentWriter`,
    /// must resume numbering from there rather than renumbering from 0
    /// again — a restarted segmenter that numbers from 0 is exactly what a
    /// `Trunk`'s monotonic-`sequence_number` guard rejects every publish
    /// for.
    ///
    /// The usual constructor, [`Self::new`], starts at `0`.
    #[doc(hidden)]
    pub fn with_start_sequence(
        tracks: Vec<TrackSpec>,
        target_secs: u32,
        window: usize,
        next_sequence: u64,
    ) -> Result<Self> {
        let mut seg = Self::new(tracks, target_secs, window)?;
        seg.total_segments = next_sequence;
        Ok(seg)
    }

    /// Select the [`SegmentCutPolicy`] (default [`SegmentCutPolicy::AnyRap`]).
    ///
    /// With [`SegmentCutPolicy::IdrOnly`] the segmenter drops every sample
    /// until the anchor track delivers its first IDR picture, so the first
    /// segment starts on an IDR (Apple HLS Authoring Spec 7.4); the dropped
    /// anchor access units are counted by [`Self::dropped_leading_aus`].
    /// Only AVC/HEVC/VVC anchors carry IDR pictures: any other anchor codec
    /// keeps the sync-flag rule under either policy.
    ///
    /// Call before the first [`Self::push`]; a later call re-arms the wait
    /// only if nothing has been buffered or cut yet.
    #[must_use]
    pub fn with_cut_policy(mut self, policy: SegmentCutPolicy) -> Self {
        self.cut_policy = policy;
        let untouched =
            self.total_segments == 0 && self.tracks.iter().all(|t| t.pending.is_empty());
        self.awaiting_idr = policy == SegmentCutPolicy::IdrOnly && untouched;
        self
    }

    /// Bound the [`SegmentCutPolicy::IdrOnly`] wait for the first IDR: once
    /// `bound` of anchor media has been dropped without one, the segmenter
    /// falls back to any sync sample for the rest of the stream
    /// ([`Self::idr_fallback_active`]). Without this call it waits forever.
    #[must_use]
    pub fn with_idr_fallback(mut self, bound: core::time::Duration) -> Self {
        self.idr_wait = Some(crate::segment_cut::IdrWait::new(bound));
        self
    }

    /// Stop requiring an IDR: cut on any sync sample from now on. For a caller
    /// that measured the wait itself.
    pub fn fall_back_to_any_rap(&mut self) {
        self.idr_fallback_active = true;
    }

    /// `true` once [`SegmentCutPolicy::IdrOnly`] has fallen back to any sync
    /// sample (see [`Self::with_idr_fallback`]).
    pub fn idr_fallback_active(&self) -> bool {
        self.idr_fallback_active
    }

    /// The configured [`SegmentCutPolicy`].
    pub fn cut_policy(&self) -> SegmentCutPolicy {
        self.cut_policy
    }

    /// Anchor access units dropped because [`SegmentCutPolicy::IdrOnly`] was
    /// waiting for an IDR (session start, or after
    /// [`Self::mark_discontinuity`]). Always `0` under
    /// [`SegmentCutPolicy::AnyRap`].
    pub fn dropped_leading_aus(&self) -> u64 {
        self.dropped_leading_aus
    }

    fn anchor_nal_codec(&self) -> Option<NalCodec> {
        match self.tracks[self.anchor].spec.config {
            CodecConfig::Avc { .. } => Some(NalCodec::Avc),
            CodecConfig::Hevc { .. } => Some(NalCodec::Hevc),
            CodecConfig::Vvc { .. } => Some(NalCodec::Vvc),
            _ => None,
        }
    }

    fn opens_segment(&self, sample: &Sample) -> bool {
        match (self.cut_policy, self.anchor_nal_codec()) {
            (SegmentCutPolicy::IdrOnly, Some(codec)) if !self.idr_fallback_active => {
                access_unit_is_idr(codec, &sample.data)
            }
            _ => sample.flags.is_sync,
        }
    }

    /// Push one coded sample for `track_id`, in decode order.
    ///
    /// Mirrors [`TsHlsPackager`]'s cut rule: when the anchor track reaches a
    /// sync sample past the target duration, the samples buffered so far are
    /// cut into a `.ts` segment (returned here) *before* this sample is
    /// buffered, so the new keyframe opens the next segment. Non-anchor
    /// tracks are split by decode time exactly as
    /// [`TsHlsPackager::package`]'s `partition_tracks` splits them: pushing
    /// every sample of a `Media` through this segmenter — interleaved so that
    /// no sample is pushed before an earlier-or-equal-decode-time sample on
    /// another track — then calling [`Self::finish`] reproduces the same
    /// segment boundaries and byte-identical segments as one
    /// [`TsHlsPackager::package`] call over the whole `Media`.
    ///
    /// Anchor progress is measured by [`MediaClock`]: each anchor sample's own
    /// `duration` when that is a real, non-zero span, else the `dts` delta
    /// from the previous anchor sample. Requiring a real `duration` used to be
    /// enforced here with a hard error, which rejected a shipped path outright
    /// — [`StreamingFlvDemux`](crate::flv_stream::StreamingFlvDemux) derives
    /// `duration` as the forward delta between FLV tag timestamps, so an RTMP
    /// publish's first sample (and any two tags sharing a timestamp) carries
    /// `Some(0)`, which the *other* three segmenters silently stalled on.
    ///
    /// # Errors
    /// [`Error::InvalidInput`] if `track_id` matches no track, the underlying
    /// mux fails while cutting, or this track already holds
    /// [`MAX_PENDING_SAMPLES_PER_TRACK`] un-cut samples (no anchor sync sample
    /// to cut on — call [`finish`](Self::finish) to close a trailing partial
    /// segment).
    ///
    /// Any segment this push cuts is queued, not returned inline (issue R2)
    /// — retrieve it via [`Self::take_ready`] or [`Stage::poll`].
    pub fn push(&mut self, track_id: TrackId, sample: Sample) -> Result<()> {
        self.push_inner(track_id, sample)
    }

    /// [`push`](Self::push)'s whole body, leaving any cut segment on
    /// [`ready`](Self::ready) rather than returning it — the shared core of
    /// the inherent and [`Stage::feed`] entry points, so neither can drift and
    /// neither needs a second queue.
    fn push_inner(&mut self, track_id: TrackId, sample: Sample) -> Result<()> {
        let idx = self
            .tracks
            .iter()
            .position(|t| t.spec.track_id == track_id)
            .ok_or(Error::InvalidInput("push: unknown track_id"))?;

        if self.awaiting_idr {
            if idx != self.anchor {
                return Ok(());
            }
            if !self.opens_segment(&sample) {
                self.dropped_leading_aus += 1;
                let timescale = self.tracks[self.anchor].spec.timescale;
                if let Some(wait) = self.idr_wait.as_mut()
                    && wait.observe_dropped(&sample, timescale)
                {
                    self.idr_fallback_active = true;
                }
                return Ok(());
            }
            self.awaiting_idr = false;
            self.complete_deferred()?;
            if !self.tracks[self.anchor].pending.is_empty() {
                self.cut_segment(false)?;
            }
            if self.discontinuity_at_resync {
                self.pending_discontinuity = self.total_segments > 0;
                self.discontinuity_at_resync = false;
            }
        }

        if idx == self.anchor
            && is_cut_point(
                !self.tracks[self.anchor].pending.is_empty(),
                self.opens_segment(&sample),
                self.anchor_pending_dur,
                self.target_ticks,
            )
        {
            self.cut_or_defer()?;
        }

        if self.tracks[idx].pending.len() >= MAX_PENDING_SAMPLES_PER_TRACK {
            return Err(no_sync_sample_error());
        }

        if idx == self.anchor {
            if self.origin_secs.is_none() {
                self.origin_secs = Some(timeline_origin_secs(
                    sample.dts,
                    self.tracks[self.anchor].spec.timescale.max(1) as u64,
                ));
            }
            self.anchor_pending_dur += self.anchor_clock.tick(&sample);
        }
        self.tracks[idx].pending.push(sample);
        self.try_complete_deferred()
    }

    /// Cut now, or hold the cut while an audio track still lags the boundary
    /// ([`DeferredCut`]).
    fn cut_or_defer(&mut self) -> Result<()> {
        self.complete_deferred()?;
        let anchor = self.anchor;
        let anchor_scale = self.tracks[anchor].spec.timescale.max(1) as u64;
        let next_start_ticks = self.tracks[anchor].base_decode + self.anchor_pending_dur;
        let next_start_secs = next_start_ticks as f64 / anchor_scale as f64;
        if !self.audio_lags(next_start_secs) {
            return self.cut_segment(false);
        }
        self.deferred_cut = Some(DeferredCut {
            anchor_split: self.tracks[anchor].pending.len(),
            closed_anchor_dur: self.anchor_pending_dur,
            next_start_secs,
        });
        self.anchor_pending_dur = 0;
        Ok(())
    }

    fn complete_deferred(&mut self) -> Result<()> {
        match self.deferred_cut.take() {
            Some(cut) => self.cut_segment_with(false, Some(cut)),
            None => Ok(()),
        }
    }

    fn try_complete_deferred(&mut self) -> Result<()> {
        let Some(cut) = self.deferred_cut else {
            return Ok(());
        };
        let scale = self.tracks[self.anchor].spec.timescale.max(1) as u64;
        let waited_out = self.anchor_pending_dur >= CATCH_UP_MAX_SECS.saturating_mul(scale);
        if waited_out || !self.audio_lags(cut.next_start_secs) {
            self.complete_deferred()?;
        }
        Ok(())
    }

    /// Whether some audio track that has delivered media holds none at or
    /// past `next_start_secs`, so cutting there would close its segment short.
    fn audio_lags(&self, next_start_secs: f64) -> bool {
        let origin_secs = self.origin_secs.unwrap_or(0.0);
        self.tracks.iter().enumerate().any(|(i, t)| {
            if i == self.anchor
                || t.spec.config.audio_sample_rate().is_none()
                || (t.base_decode == 0 && t.pending.is_empty())
            {
                return false;
            }
            let scale = t.spec.timescale.max(1) as u64;
            let mut acc = t.base_decode;
            let mut clock = MediaClock::resumed_at(t.flush_clock.last_dts());
            for s in &t.pending {
                if placement_secs(s, acc, scale, origin_secs) >= next_start_secs {
                    return false;
                }
                acc += clock.tick(s);
            }
            true
        })
    }

    /// Finalize the trailing partial segment (call once at end-of-stream).
    /// `None` if nothing is buffered. Every track's remaining pending samples
    /// are flushed into this one final segment regardless of decode time —
    /// mirroring `partition_tracks`'s last segment, which runs to `+∞` and
    /// absorbs the full trailing tail of every track. Also marks the
    /// segmenter finished, so [`Self::playlist`] appends `#EXT-X-ENDLIST`.
    ///
    /// # Errors
    /// Propagates a mux failure while cutting the trailing segment.
    ///
    /// The trailing segment this cuts (if any) is queued, not returned inline
    /// (issue R2) — retrieve it via [`Self::take_ready`] or [`Stage::poll`].
    /// This matters even for a caller that never touches [`Stage`] directly:
    /// `finish` is also the [`Stage::finish`] trait method's name, so an
    /// inline return here would tempt a `Stage`-style driver's bare
    /// `seg.finish()` call (which always resolves to *this* inherent method,
    /// never the trait one, by inherent-over-trait precedence) into silently
    /// discarding it. See the `ready` field docs.
    pub fn finish(&mut self) -> Result<()> {
        self.finish_inner()
    }

    /// Drain every segment cut so far but not yet retrieved, in cut order —
    /// the inherent-API analogue of [`Stage::poll`], draining the same
    /// private `ready` queue (issue R2). Mirrors
    /// [`Segmenter::take_ready`](crate::segmenter::Segmenter::take_ready).
    pub fn take_ready(&mut self) -> Vec<TsSegment> {
        self.ready.drain(..).collect()
    }

    /// [`finish`](Self::finish)'s whole body, leaving the trailing segment on
    /// [`ready`](Self::ready) rather than returning it — shared with
    /// [`Stage::finish`] for the same reason as [`push_inner`](Self::push_inner).
    fn finish_inner(&mut self) -> Result<()> {
        self.finished = true;
        self.complete_deferred()?;
        if self.tracks.iter().any(|t| !t.pending.is_empty()) {
            self.cut_segment(true)?;
        }
        Ok(())
    }

    /// Mark the *next* segment cut as a media-timeline discontinuity
    /// (RFC 8216 §4.3.2.3) — call this when the caller detects a PID/PCR
    /// reset in the live source (e.g. an upstream PMT/PCR-PID change)
    /// between one push and the next.
    ///
    /// Under [`SegmentCutPolicy::IdrOnly`] the segmenter additionally drops
    /// samples until the anchor's next IDR, closes the pre-gap segment there,
    /// and flags the segment that IDR opens as discontinuous.
    pub fn mark_discontinuity(&mut self) {
        if self.cut_policy == SegmentCutPolicy::IdrOnly {
            self.awaiting_idr = true;
            self.discontinuity_at_resync = true;
        } else {
            self.pending_discontinuity = true;
        }
    }

    /// Register a track after construction (issue #624).
    ///
    /// A live [`StreamingTsDemux`](crate::ts_demux::StreamingTsDemux) resolves
    /// tracks incrementally — `DemuxEvent::TrackAdded` for one track can land
    /// after a segmenter has already been built and has already cut segments
    /// from an earlier-resolving track (e.g. video resolves and cuts a
    /// segment before audio's first frame parses). `add_track` lets that
    /// track join the segmenter mid-stream instead of forcing a rebuild that
    /// would silently drop it. Segments cut *before* `add_track` simply have
    /// no PES data for this track (spec-legal: a PMT may declare a track with
    /// zero samples in a given segment); every segment cut *after*
    /// `add_track` carries it correctly in both the PMT and the PES.
    ///
    /// # Anchor recomputation
    ///
    /// The anchor (the keyframe-cut-boundary track, normally "first video,
    /// else first track" — `choose_anchor`) was chosen from whatever track
    /// set existed at construction time and is deliberately **not**
    /// reconsidered on every `add_track`, or a video track added on segment
    /// 50 could retroactively invalidate the cut boundaries of the 49
    /// segments already emitted. The one case this method *does* recompute
    /// the anchor is: nothing has been cut yet (`total_segments == 0`), no
    /// track has any sample buffered yet (every `pending` is empty — so no
    /// cut/anchor decision has been made even provisionally), the current
    /// anchor is not already a video track (`CodecConfig::is_video`, internal), and
    /// the newly-added track *is* video. That exactly recovers the
    /// construction-time rule ("first video, else first track") for the case
    /// this issue targets: the segmenter had to be built audio-first because
    /// video hadn't resolved (or vice versa), so by the time
    /// `add_track(video)` runs, nothing has cut yet, hence it's safe. Every
    /// other case (anchor already video, new track isn't video, or anything
    /// has already been buffered/cut) keeps the existing anchor unchanged
    /// and simply adds the track for muxing.
    ///
    /// # PMT version and discontinuity
    ///
    /// Once any segment has been cut, the track set the stream carries
    /// changes, so the next segment's PMT is written with `version_number`
    /// incremented modulo 32 (ISO/IEC 13818-1 §2.4.4.9; issue #1699) and that
    /// segment is flagged discontinuous, because RFC 8216bis §4.4.4.3 requires
    /// `#EXT-X-DISCONTINUITY` when the number, type or identifiers of tracks
    /// change.
    ///
    /// # Errors
    /// [`Error::InvalidInput`] if `spec.track_id` collides with an existing
    /// track.
    pub fn add_track(&mut self, spec: TrackSpec) -> Result<()> {
        if self.tracks.iter().any(|t| t.spec.track_id == spec.track_id) {
            return Err(Error::InvalidInput("add_track: duplicate track_id"));
        }

        let nothing_cut_or_buffered =
            self.total_segments == 0 && self.tracks.iter().all(|t| t.pending.is_empty());
        let current_anchor_is_video = self.tracks[self.anchor].spec.config.is_video();
        let new_is_video = spec.config.is_video();

        let new_index = self.tracks.len();
        self.tracks.push(StreamTrackState {
            spec,
            pending: Vec::new(),
            base_decode: 0,
            flush_clock: MediaClock::new(),
        });

        if self.total_segments > 0 {
            self.pending_discontinuity = true;
        }

        if nothing_cut_or_buffered && new_is_video && !current_anchor_is_video {
            self.anchor = new_index;
            self.target_ticks =
                target_ticks_for(self.target_secs, self.tracks[self.anchor].spec.timescale);
        }

        Ok(())
    }

    /// Update an existing track's spec mid-stream (issue #1425; RFC 8216
    /// section 4.3.4.3, Apple HLS Authoring Spec 2.11 / 8.14).
    ///
    /// Arms a discontinuity only when audio `sample_rate`, `channel_count` or
    /// the AAC AudioSpecificConfig bytes change, or the video `timescale` or
    /// `display` changes. SPS or bitrate-only changes do not. The stored spec
    /// is always refreshed.
    ///
    /// An unknown `track_id` returns [`UpdateTrackOutcome::Unknown`]; use
    /// [`Self::add_track`] for a new track.
    pub fn update_track(&mut self, spec: TrackSpec) -> Result<UpdateTrackOutcome> {
        let Some(idx) = self
            .tracks
            .iter()
            .position(|t| t.spec.track_id == spec.track_id)
        else {
            return Ok(UpdateTrackOutcome::Unknown);
        };

        let old = &self.tracks[idx].spec;

        let discontinuity = old.timescale != spec.timescale
            || old.display != spec.display
            || audio_discontinuity_trigger(&old.config, &spec.config);

        let old_timescale = self.tracks[idx].spec.timescale;
        self.tracks[idx].spec = spec;
        if discontinuity {
            if idx == self.anchor && self.tracks[idx].spec.timescale != old_timescale {
                self.target_ticks =
                    target_ticks_for(self.target_secs, self.tracks[self.anchor].spec.timescale);
            }
            self.mark_discontinuity();
            Ok(UpdateTrackOutcome::Changed)
        } else {
            Ok(UpdateTrackOutcome::Unchanged)
        }
    }

    /// The current rolling media playlist: at most [`window`](Self::new)
    /// segments (the most recently cut), with `#EXT-X-MEDIA-SEQUENCE`
    /// advanced past every segment that has rolled out of the window and
    /// `#EXT-X-DISCONTINUITY-SEQUENCE` incremented once per discontinuous
    /// segment that has rolled *off* the window. RFC 8216 §6.2.1 defines a
    /// segment's Discontinuity Sequence Number as this header value plus the
    /// number of `#EXT-X-DISCONTINUITY` tags still visible in the current
    /// playlist preceding that segment's URI — so a discontinuous segment's
    /// own inline tag (rendered while it is still in the window) already
    /// accounts for its own boundary, and the header must only advance once
    /// that segment (and its tag) is gone, or every segment sharing the
    /// window with it would have its Discontinuity Sequence Number
    /// double-counted. No `#EXT-X-ENDLIST` until [`Self::finish`] has been
    /// called.
    pub fn playlist(&self) -> crate::Result<String> {
        let segments: Vec<MediaSegment> = self
            .window_segments
            .iter()
            .map(|e| MediaSegment {
                uri: e.uri.clone(),
                duration: e.duration,
                discontinuous: e.discontinuous,
                parts: vec![],
                ..Default::default()
            })
            .collect();

        let media_sequence = self.total_segments - self.window_segments.len() as u64;

        Ok(MediaPlaylist {
            version: self.version,
            target_duration: self.target_duration.max(1),
            media_sequence,
            discontinuity_sequence: self.discontinuity_sequence,
            segments,
            endlist: self.finished,
            extra_tags: vec![],
            low_latency: None,
            iframes_only: false,
            open_segment: None,
            ..Default::default()
        }
        .to_m3u8()?)
    }

    /// Cut samples into one `.ts` segment.
    ///
    /// When `final_cut` is `false` (a keyframe-triggered cut from
    /// [`Self::push`]), the anchor's whole pending buffer closes out, and a
    /// non-anchor track's samples that decode-start at or after the new
    /// segment's start time stay pending for the next segment — the
    /// incremental form of `partition_tracks`'s time-based split for
    /// non-anchor tracks. When `final_cut` is `true` (from [`Self::finish`]),
    /// every track's entire pending buffer is flushed regardless of decode
    /// time, mirroring `partition_tracks`'s last segment (which runs to `+∞`
    /// and absorbs the full trailing tail of every track).
    ///
    /// The cut segment is pushed onto [`ready`](Self::ready), the single queue
    /// both the inherent API and [`Stage::poll`] drain.
    fn cut_segment(&mut self, final_cut: bool) -> Result<()> {
        self.cut_segment_with(final_cut, None)
    }

    fn cut_segment_with(&mut self, final_cut: bool, deferred: Option<DeferredCut>) -> Result<()> {
        let closed_anchor_dur = deferred.map_or(self.anchor_pending_dur, |d| d.closed_anchor_dur);
        let anchor = self.anchor;
        let anchor_scale = self.tracks[anchor].spec.timescale.max(1) as u64;
        let origin_secs = self.origin_secs.unwrap_or(0.0);
        // The new segment's start time = this track's cumulative decode time
        // so far (already-flushed ticks + the pending anchor buffer) — the
        // streaming equivalent of `partition_tracks`'s `start_times[seg + 1]`.
        let next_start_ticks = self.tracks[anchor].base_decode + closed_anchor_dur;
        let next_start_secs = next_start_ticks as f64 / anchor_scale as f64;

        // Per-track split point: everything pending for the anchor (and for
        // every track on the final cut); time-partitioned for non-anchor
        // tracks on a regular keyframe-triggered cut.
        let split_at: Vec<usize> = self
            .tracks
            .iter()
            .enumerate()
            .map(|(i, t)| {
                if final_cut {
                    return t.pending.len();
                }
                if i == anchor {
                    return deferred.map_or(t.pending.len(), |d| d.anchor_split);
                }
                let scale = t.spec.timescale.max(1) as u64;
                let mut acc = t.base_decode;
                // Same `MediaClock`/`placement_secs` pair the batch
                // `partition_tracks` uses, so both paths place a given sample
                // — including a timestamped, duration-less section sample — in
                // the same segment.
                let mut clock = MediaClock::resumed_at(t.flush_clock.last_dts());
                for (j, s) in t.pending.iter().enumerate() {
                    let start_secs = placement_secs(s, acc, scale, origin_secs);
                    if start_secs >= next_start_secs {
                        return j;
                    }
                    acc += clock.tick(s);
                }
                t.pending.len()
            })
            .collect();

        // Ephemeral `Track`s carrying only the spec (the mux reads
        // `track.spec` for PID/stream_type planning; samples are passed
        // separately as borrowed slices below).
        let mux_tracks: Vec<Track> = self
            .tracks
            .iter()
            .map(|t| Track::new(t.spec.clone(), Vec::new()))
            .collect();
        let sample_slices: Vec<&[Sample]> = self
            .tracks
            .iter()
            .zip(&split_at)
            .map(|(t, &n)| &t.pending[..n])
            .collect();
        let base_dts: Vec<u64> = self.tracks.iter().map(|t| t.base_decode).collect();
        let bytes =
            mux_tracks_at_continuing(&mux_tracks, &sample_slices, &base_dts, &mut self.continuity)?;

        let (duration, ceil_secs) = segment_duration_secs(closed_anchor_dur, anchor_scale);
        if ceil_secs > self.target_duration {
            self.target_duration = ceil_secs;
        }

        let discontinuous = self.pending_discontinuity;
        self.pending_discontinuity = false;

        // Drop the flushed prefix of each track's pending buffer and advance
        // its base_decode past it.
        for (t, &n) in self.tracks.iter_mut().zip(&split_at) {
            let clock = &mut t.flush_clock;
            let dur: u64 = t.pending[..n].iter().map(|s| clock.tick(s)).sum();
            t.base_decode += dur;
            t.pending.drain(..n);
        }
        if deferred.is_none() {
            self.anchor_pending_dur = 0;
        }

        let sequence = self.total_segments;
        let uri = format!("{}{}.ts", self.uri_prefix, sequence);
        self.window_segments.push_back(WindowEntry {
            uri: uri.clone(),
            // `segment_duration_secs` divides a `u64` tick count by a
            // `u64.max(1)` timescale (issue #1140): always finite and
            // non-negative.
            duration: DecimalSeconds::new(duration).expect("segment_duration_secs is finite, >= 0"),
            discontinuous,
        });
        self.total_segments += 1;
        while self.window_segments.len() > self.window {
            if let Some(dropped) = self.window_segments.pop_front()
                && dropped.discontinuous
            {
                self.discontinuity_sequence += 1;
            }
        }

        self.ready.push_back(TsSegment {
            bytes,
            duration,
            discontinuous,
            uri,
            sequence,
        });
        Ok(())
    }
}

/// [`Stage`] adoption (media plane step 2e-2): `In = (u32, Sample)`, same
/// reasoning as the other three segmenters' impls. `Out = TsSegment` — this
/// segmenter's own natural output, not unified with any other segmenter's
/// `Out`.
///
/// Like the other three segmenters (issue R2), [`push`](StreamingTsHlsSegmenter::push)/
/// [`finish`](StreamingTsHlsSegmenter::finish) never return a cut segment
/// inline — every cut lands on the same `ready` queue this impl's
/// [`poll`](Stage::poll) drains, and the inherent
/// [`take_ready`](StreamingTsHlsSegmenter::take_ready) drains it too, so a
/// segment is retrievable exactly once no matter which API — inherent,
/// `Stage`, or a mix of both on the same instance — the caller uses.
impl Stage for StreamingTsHlsSegmenter {
    type In<'a> = (TrackId, Sample);
    type Out = TsSegment;
    type Error = Error;

    fn feed(&mut self, (track_id, sample): Self::In<'_>, _now: Timestamp) -> Result<()> {
        self.push_inner(track_id, sample)
    }

    fn poll(&mut self) -> Option<Self::Out> {
        self.ready.pop_front()
    }

    /// Delegates to the private `finish_inner`, **not** to `self.finish()`:
    /// the inherent [`finish`](StreamingTsHlsSegmenter::finish) and this trait
    /// method share a name *and* an arity, so bare method-call syntax picks
    /// the inherent one only by inherent-over-trait precedence. Renaming the
    /// inherent method would silently retarget `self.finish()` at this very
    /// method — infinite recursion that still compiles.
    fn finish(&mut self) -> Result<()> {
        self.finish_inner()
    }

    /// `saturated` once any track holds [`MAX_PENDING_SAMPLES_PER_TRACK`]
    /// un-cut samples — the same bound (and reasoning) as
    /// [`Segmenter`](crate::segmenter::Segmenter)'s: past it
    /// [`feed`](Stage::feed) errors rather than buffering a stream that never
    /// produces the sync sample a `.ts` segment must open on.
    fn demand(&self) -> Demand {
        Demand::saturated_if(
            self.tracks
                .iter()
                .any(|t| t.pending.len() >= MAX_PENDING_SAMPLES_PER_TRACK),
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pipeline::{CodecConfig, DataCarriage};

    fn sample(dur: u32, sync: bool) -> Sample {
        Sample {
            data: vec![0u8; 4].into(),
            // Boundary selection is duration-driven; absolute time is not
            // what this helper exercises.
            dts: None,
            pts: None,
            duration: Some(dur),
            flags: crate::ir::SampleFlags::new(sync),
            provenance: None,
            display: None,
        }
    }

    #[test]
    fn boundaries_cut_on_keyframe_past_target() {
        // Durations 1 each, sync at 0,2,4,6; target 2 ticks.
        let s: Vec<Sample> = (0..8).map(|i| sample(1, i % 2 == 0)).collect();
        let b = anchor_segment_boundaries(&s, 2);
        // Segment 0: [0,2) (buffered reaches 2 at idx2 sync). Then [2,4), [4,6), [6,8).
        assert_eq!(b, vec![0, 2, 4, 6]);
    }

    #[test]
    fn boundaries_single_when_target_exceeds_stream() {
        let s: Vec<Sample> = (0..4).map(|i| sample(1, i == 0)).collect();
        let b = anchor_segment_boundaries(&s, 1000);
        assert_eq!(b, vec![0], "one segment when target dwarfs the stream");
    }

    // ── Issue B9: a section-carried anchor must not stall the segmenter ────
    //
    // A section-carried track (e.g. SCTE-35 `splice_info_section`, carried
    // directly on its own PID with no PES header — ISO/IEC 13818-1 §2.4.4)
    // always has `duration: None` (media plane step 2c: never fabricated).
    // Before this fix, `choose_anchor` fell back to track index 0 whenever
    // no video track was present, so a track set like "SCTE-35 first, audio
    // second" picked the SCTE-35 track as the anchor: its buffered duration
    // could never advance, `push` never returned a segment, and every
    // track's `pending` buffer grew without bound.

    use crate::mp4esds::{
        DecoderConfigDescriptor, DecoderSpecificInfo, ESDescriptor, EsdsBox, SLConfigDescriptor,
    };
    use broadcast_common::Unpackage;

    fn dummy_esds() -> EsdsBox {
        EsdsBox::new(ESDescriptor::new(
            1,
            0,
            Some(DecoderConfigDescriptor::new(
                0x40,
                0x05,
                false,
                0,
                0,
                0,
                Some(DecoderSpecificInfo::new(vec![0x12, 0x10])),
            )),
            Some(SLConfigDescriptor::predefined_two()),
        ))
    }

    /// A section-carried SCTE-35 track spec (ISO/IEC 13818-1 Table 2-34
    /// stream_type 0x86, ANSI/SCTE 35): permanently `duration: None`, so it
    /// must never be picked as the segmentation anchor.
    fn scte35_track(track_id: TrackId) -> TrackSpec {
        TrackSpec::new(
            track_id,
            90_000,
            CodecConfig::Data {
                stream_type: 0x86,
                descriptors: Vec::new(),
                carriage: DataCarriage::Sections,
            },
        )
    }

    /// A `splice_info_section`-shaped placeholder sample: no timestamp, no
    /// duration — exactly what `TsDemux`/`StreamingTsDemux` emit for a
    /// section-carried track (never fabricated).
    fn section_sample() -> Sample {
        Sample {
            data: vec![0xFCu8, 0x30, 0x11].into(),
            dts: None,
            pts: None,
            duration: None,
            flags: crate::ir::SampleFlags::SYNC,
            provenance: None,
            display: None,
        }
    }

    fn aac_track(track_id: TrackId, timescale: u32) -> TrackSpec {
        TrackSpec::new(
            track_id,
            timescale,
            CodecConfig::Aac {
                esds: dummy_esds(),
                channel_count: 2,
                sample_rate: timescale,
                sample_size: 16,
            },
        )
    }

    /// An AAC-shaped audio sample: always a sync sample (mirrors
    /// `Sample::from_raw`), with a real, caller-supplied duration.
    fn audio_sample(duration: u32) -> Sample {
        Sample::from_raw(vec![0u8; 4], None, None, Some(duration))
    }

    #[test]
    fn streaming_segmenter_advances_past_a_section_track_anchor_and_cuts_real_segments() {
        const SCTE_ID: u32 = 10;
        const AUDIO_ID: u32 = 20;
        const TIMESCALE: u32 = 1000;
        const TARGET_SECS: u32 = 1; // target_ticks = 1000

        // Section track first, no video — the exact routine DVB/ad-insertion
        // shape that used to stall forever.
        let mut seg = StreamingTsHlsSegmenter::new(
            vec![
                scte35_track(TrackId::new(SCTE_ID)),
                aac_track(TrackId::new(AUDIO_ID), TIMESCALE),
            ],
            TARGET_SECS,
            usize::MAX,
        )
        .expect("construct: audio is anchor-capable even though it isn't first");

        // A couple of section samples up front (duration: None) must not
        // prevent the audio anchor from advancing.
        seg.push(TrackId::new(SCTE_ID), section_sample())
            .expect("push section");
        seg.push(TrackId::new(SCTE_ID), section_sample())
            .expect("push section");

        // 12 audio samples of 200 ticks each: a cut fires once buffered
        // reaches the 1000-tick target on the 6th and 11th pushes. Cuts are
        // queued, not returned inline (issue R2) — drained via `take_ready`.
        for _ in 0..12 {
            seg.push(TrackId::new(AUDIO_ID), audio_sample(200))
                .expect("push audio");
        }
        let cuts = seg.take_ready();

        assert_eq!(
            cuts.len(),
            2,
            "the audio anchor must have advanced past the 1000-tick target twice \
             — before the fix this was 0 (the segmenter stalled forever)"
        );
        for (i, c) in cuts.iter().enumerate() {
            assert!(
                !c.bytes.is_empty(),
                "cut segment {i} must carry real TS bytes"
            );
            assert!(
                c.duration > 0.0,
                "cut segment {i} must have a positive duration"
            );
        }

        // Every emitted segment is a byte-oracle-verifiable real TS mux —
        // decode the first one and confirm it actually carries the AAC PES.
        let demuxed = crate::ts_demux::TsDemux::new()
            .unpackage(&cuts[0].bytes)
            .expect("cut segment must be a valid TS stream");
        assert!(
            demuxed
                .tracks
                .iter()
                .any(|t| matches!(t.spec.config, CodecConfig::Aac { .. }) && !t.samples.is_empty()),
            "cut segment must carry the audio track's samples, not just PSI"
        );

        // The pending buffer stayed bounded (2 audio samples left over after
        // 12 pushes and 2 cuts) — before the fix this would have grown to
        // 12 (audio) with the section track's backlog also never draining.
        let audio_track_state = seg
            .tracks
            .iter()
            .find(|t| t.spec.track_id == TrackId::new(AUDIO_ID))
            .expect("audio track state");
        assert_eq!(
            audio_track_state.pending.len(),
            2,
            "pending must be bounded by the cut cadence, not grow without bound"
        );
    }

    #[test]
    fn streaming_segmenter_construction_errors_loudly_with_no_anchorable_track() {
        // Only section/opaque tracks, no video/audio: there is no track
        // whose duration can ever advance a cut boundary. Prefer a loud
        // construction error over a segmenter that silently never cuts.
        let result = StreamingTsHlsSegmenter::new(
            vec![
                scte35_track(TrackId::new(10)),
                scte35_track(TrackId::new(11)),
            ],
            1,
            usize::MAX,
        );
        match result {
            Err(Error::InvalidInput(_)) => {}
            Err(other) => panic!("expected InvalidInput, got a different error: {other:?}"),
            Ok(_) => panic!("a track set of only section-carried tracks must not construct"),
        }
    }

    /// Reproduces the R2 hazard directly: a driver that pushes exclusively
    /// through [`Stage::feed`]/[`Stage::poll`] (never touching the inherent
    /// API) but closes the stream with a bare, unqualified `seg.finish()`
    /// call. Rust's inherent-over-trait method resolution means that bare
    /// call always picks [`StreamingTsHlsSegmenter::finish`] — the inherent
    /// method — never [`Stage::finish`], regardless of the fact that every
    /// other call in this test goes through `Stage`. A `Stage`-only driver
    /// has no reason to inspect a `finish()` call's return value as anything
    /// but `Result<()>`, so before the R2 fix (when the inherent `finish`
    /// popped-and-returned the trailing segment as `Result<Option<TsSegment>>`)
    /// that value was simply dropped, unread, and the trailing segment was
    /// gone: a following `Stage::poll` found the queue already empty. Since
    /// the R2 fix, `finish` only ever returns `Result<()>`, so there is no
    /// value to drop — the segment is retrieved by the following `poll` no
    /// matter what.
    #[test]
    fn finish_bare_call_does_not_lose_the_trailing_segment_to_a_poll_driver() {
        const AUDIO_ID: u32 = 1;
        const TIMESCALE: u32 = 1000;

        let mut seg = StreamingTsHlsSegmenter::new(
            vec![aac_track(TrackId::new(AUDIO_ID), TIMESCALE)],
            1000,
            usize::MAX,
        )
        .expect("construct");

        // One sample only: not enough to cross the (huge) cut target, so
        // nothing is cut yet and the only segment `Stage::poll` will ever see
        // is the trailing one `finish` cuts below.
        Stage::feed(
            &mut seg,
            (TrackId::new(AUDIO_ID), audio_sample(200)),
            Timestamp::ZERO,
        )
        .expect("feed via Stage");
        assert!(
            Stage::poll(&mut seg).is_none(),
            "no cut yet: nothing should be ready before finish"
        );

        // The hazard: an unqualified bare call, exactly what a caller who
        // believes they are only using the `Stage` trait would naturally
        // write to close the stream out.
        seg.finish().expect("finish");

        assert!(
            Stage::poll(&mut seg).is_some(),
            "the trailing segment cut by a bare finish() call must still be retrievable via \
             Stage::poll — it must not have been silently handed back inline and dropped"
        );
    }

    /// [`StreamingTsHlsSegmenter::with_start_sequence`] must number the
    /// first segment it cuts from the seed, not from `0` — the seam
    /// `multimux::source::segment` uses to resume TS numbering across an
    /// `add_track` failure/rebuild instead of renumbering from `0` again
    /// (which a `Trunk`'s monotonic `sequence_number` guard would then
    /// reject forever). `Self::new` (seed implicitly `0`) is exercised by
    /// every other test in this module; this is the seeded constructor's
    /// own coverage.
    #[test]
    fn with_start_sequence_numbers_the_first_cut_segment_from_the_seed() {
        const AUDIO_ID: u32 = 1;
        const TIMESCALE: u32 = 1000;
        const SEED: u64 = 41;

        let mut seg = StreamingTsHlsSegmenter::with_start_sequence(
            vec![aac_track(TrackId::new(AUDIO_ID), TIMESCALE)],
            1,
            usize::MAX,
            SEED,
        )
        .expect("construct");

        // Two 1000-tick (1s) samples cross the 1s target on the second push.
        seg.push(TrackId::new(AUDIO_ID), audio_sample(1000))
            .expect("push 1");
        seg.push(TrackId::new(AUDIO_ID), audio_sample(1000))
            .expect("push 2");
        seg.finish().expect("finish");

        let cut: Vec<TsSegment> = seg.take_ready().into_iter().collect();
        assert_eq!(
            cut.iter().map(|s| s.sequence).collect::<Vec<_>>(),
            vec![SEED, SEED + 1],
            "segments cut by a seeded segmenter must number from the seed, not from 0"
        );
    }
    // ── Audit r05-W33: the batch packager is O(samples), not O(segments ×
    //    samples) ────────────────────────────────────────────────────────────

    /// A two-track media (an AVC anchor plus an audio track) of `count`
    /// one-tick samples each, sync every `gop` on the anchor, so `package`
    /// cuts `count / gop` segments and both tracks contribute samples.
    fn countable_media(count: usize, gop: usize) -> Media {
        use crate::nalu_types::{AvcPps, AvcSps};
        let record = crate::avc_config::AVCDecoderConfigurationRecord {
            configuration_version: 1,
            profile_indication: 0x42,
            profile_compatibility: 0,
            level_indication: 0x1E,
            length_size_minus_one: 3,
            sps: vec![AvcSps(vec![0x67, 0x42, 0xc0, 0x1e, 0xd9])],
            pps: vec![AvcPps(vec![0x68, 0xce, 0x3c, 0x80])],
            chroma_format: None,
            bit_depth_luma_minus8: None,
            bit_depth_chroma_minus8: None,
            sps_ext: vec![],
        };
        let video_spec = TrackSpec::new(
            TrackId::new(1),
            90_000,
            CodecConfig::Avc {
                config: crate::avc_config::AVCConfigurationBox::new(record),
                width: 320,
                height: 240,
            },
        );
        let audio_spec = aac_track(TrackId::new(2), 90_000);
        // One second per sample in a 90 kHz timescale, so a 1-second target
        // cuts `count / gop` segments.
        let video: Vec<Sample> = (0..count).map(|i| sample(90_000, i % gop == 0)).collect();
        let audio: Vec<Sample> = (0..count).map(|i| sample(90_000, i % gop == 0)).collect();
        Media::new(
            vec![Track::new(video_spec, video), Track::new(audio_spec, audio)],
            1000,
        )
    }

    /// The batch packager must call `MediaClock::tick` a number of times
    /// proportional to the *sample count*, however many segments it cuts. On
    /// the pre-fix code each segment re-walked every earlier sample of every
    /// track, so 400 anchor samples in 100 segments cost `Σ 4k` = 20 200 anchor
    /// visits alone.
    #[test]
    fn batch_package_ticks_each_sample_once() {
        const COUNT: usize = 400;
        const GOP: usize = 4;

        let media = countable_media(COUNT, GOP);
        let mut packager = TsHlsPackager::new(1);
        PACKAGE_TICK_CALLS.with(|c| c.set(0));
        let out = packager.package(&media).expect("package");
        let visits = PACKAGE_TICK_CALLS.with(|c| c.get());

        assert_eq!(
            out.segments.len(),
            COUNT / GOP,
            "the fixture must cut many segments for the bound to bite"
        );
        assert_eq!(
            visits,
            COUNT * 2,
            "each of the two tracks' {COUNT} samples must be ticked exactly once 
             (the pre-fix code re-walked every earlier sample per segment)"
        );
    }
}