transmux 0.20.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.4.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.
//!
//! # 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 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 crate::error::{Error, Result};
use crate::hls::{MediaPlaylist, MediaSegment};
use crate::media::{Media, Track};
use crate::pipeline::{Sample, TrackSpec};
use crate::segmenter::{
    MAX_PENDING_SAMPLES_PER_TRACK, MediaClock, choose_anchor, no_sync_sample_error,
};
use crate::ts_mux::mux_tracks_at;

/// 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;

/// 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(concat!(env!("CARGO_MANIFEST_DIR"), "/../fixtures/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.
        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;
        for (i, seg) in segments.iter().enumerate() {
            let sample_slices: Vec<&[Sample]> = media
                .tracks
                .iter()
                .zip(&seg.ranges)
                .map(|(t, r)| &t.samples[r.clone()])
                .collect();
            // Base DTS per track = cumulative elapsed media time of all samples
            // before this segment's start, so the segment continues the previous
            // timeline and the concatenation forms one monotonic DTS/PTS
            // timeline. `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.
            let base_dts: Vec<u64> = media
                .tracks
                .iter()
                .zip(&seg.ranges)
                .map(|(t, r)| {
                    let mut clock = MediaClock::new();
                    t.samples[..r.start].iter().map(|s| clock.tick(s)).sum()
                })
                .collect();
            let bytes = mux_tracks_at(&media.tracks, &sample_slices, &base_dts)?;
            ts_segments.push(bytes);

            // Segment duration = the anchor track's buffered duration (seconds),
            // on the same `MediaClock` rule as the cut boundaries above.
            let anchor_ticks: u64 = {
                let anchor_samples = &media.tracks[anchor].samples;
                let mut clock = MediaClock::new();
                for s in &anchor_samples[..seg.ranges[anchor].start] {
                    clock.tick(s);
                }
                anchor_samples[seg.ranges[anchor].clone()]
                    .iter()
                    .map(|s| clock.tick(s))
                    .sum()
            };
            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),
                duration,
                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/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.4.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: f64,
    discontinuous: bool,
}

/// Per-track accumulation state for [`StreamingTsHlsSegmenter`].
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(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(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(); // 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,
    /// 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,
    /// 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>,
    /// 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>,
}

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,
            anchor_clock: MediaClock::new(),
            origin_secs: None,
            pending_discontinuity: false,
            finished: false,
            version: DEFAULT_HLS_VERSION,
            uri_prefix: String::from("seg"),
            window,
            window_segments: VecDeque::new(),
            total_segments: 0,
            discontinuity_sequence: 0,
            target_duration: 0,
            ready: VecDeque::new(),
        })
    }

    /// 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: u32, 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: u32, 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 idx == self.anchor
            && is_cut_point(
                !self.tracks[self.anchor].pending.is_empty(),
                sample.flags.is_sync,
                self.anchor_pending_dur,
                self.target_ticks,
            )
        {
            self.cut_segment(false)?;
        }

        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);
        Ok(())
    }

    /// 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;
        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.4.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.
    pub fn mark_discontinuity(&mut self) {
        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.
    ///
    /// # 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 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(())
    }

    /// 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) -> 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;

        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<()> {
        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 + self.anchor_pending_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 || i == anchor {
                    return t.pending.len();
                }
                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 (mux_tracks_at 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(&mux_tracks, &sample_slices, &base_dts)?;

        let (duration, ceil_secs) = segment_duration_secs(self.anchor_pending_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);
        }
        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(),
            duration,
            discontinuous,
        });
        self.total_segments += 1;
        while self.window_segments.len() > self.window {
            if let Some(dropped) = self.window_segments.pop_front() {
                if 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> = (u32, 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()
    }

    fn next_deadline(&self) -> Option<Timestamp> {
        // Segments are only cut in reaction to `push`/`finish` — no
        // rate-scheduled or timeout work.
        None
    }

    fn on_deadline(&mut self, _now: Timestamp) {}

    /// `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 {
        if self
            .tracks
            .iter()
            .any(|t| t.pending.len() >= MAX_PENDING_SAMPLES_PER_TRACK)
        {
            Demand::saturated()
        } else {
            Demand::default()
        }
    }
}

#[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,
        }
    }

    #[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, ObjectTypeIndication,
        SLConfigDescriptor, StreamType,
    };
    use broadcast_common::Unpackage;

    fn dummy_esds() -> EsdsBox {
        EsdsBox::new(ESDescriptor {
            es_id: 1,
            stream_dependence_flag: false,
            url_flag: false,
            ocr_stream_flag: false,
            stream_priority: 0,
            depends_on_es_id: None,
            url: None,
            ocr_es_id: None,
            decoder_config: Some(DecoderConfigDescriptor {
                object_type_indication: ObjectTypeIndication(0x40),
                stream_type: StreamType(0x05),
                up_stream: false,
                buffer_size_db: 0,
                max_bitrate: 0,
                avg_bitrate: 0,
                decoder_specific_info: Some(DecoderSpecificInfo {
                    data: vec![0x12, 0x10],
                }),
            }),
            sl_config: Some(SLConfigDescriptor { body: vec![0x02] }),
        })
    }

    /// 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: u32) -> 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,
        }
    }

    fn aac_track(track_id: u32, 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(SCTE_ID), aac_track(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(SCTE_ID, section_sample()).expect("push section");
        seg.push(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(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 == 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(10), scte35_track(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(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, (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"
        );
    }
}