transmux 0.28.0

//! Streaming RTP depayloader — RFC 6184 (H.264) / RFC 3640 (AAC).
//!
//! Stateful counterpart to [`crate::rtp::RtpDepacketiser`]: fed RTP packets
//! incrementally via [`RtpStreamDepacketiser::push`], it emits fully-timed
//! [`Sample`]s (real per-AU `duration` from RTP-timestamp deltas, `is_sync`
//! from IDR detection) carrying the real [`CodecConfig`] supplied at
//! construction (e.g. from [`crate::rtp_sdp`]).
//!
//! See [`transmux/docs/rtp/rtp-payload-formats.md`](../README.md)
//! for the RFC background and timing-model specification.
//!
//! # Timing model
//!
//! - Per track, the IR timescale is the RTP `clock_rate` (video: 90 kHz;
//!   AAC: the sample rate).
//! - A sample's `duration` is the RTP-timestamp delta to the *next* access
//!   unit's timestamp — so a sample can only be emitted once the following
//!   AU's timestamp is known (one-AU emission latency). `flush`
//!   emits the final pending AU using the last-computed duration (there is
//!   no "next" AU to measure against).
//! - The 32-bit wire RTP timestamp is unwrapped to a monotonic `u64` **once,
//!   here at the demux edge** (RFC 3550 §5.1), and carried into each sample's
//!   **absolute** `dts`/`pts` (media plane step 2c) — nothing downstream
//!   re-derives it.
//! - `is_sync` comes straight from the reassembled access unit
//!   (IDR detection for video; always `true` for audio).
//! - **Video `dts`** is derived, not copied: RTP carries only a presentation
//!   timestamp on the wire (RFC 6184 §5.1's sampling time) and delivers access
//!   units in decoding order, so with B-frames the wire timestamps step
//!   backward. `dts` is the presentation-ordered timestamp grid shifted by the
//!   codec's reorder depth, giving `dts` strictly increasing and `dts <= pts`
//!   on every sample; a stream without reorder keeps `dts == pts`. The rule,
//!   its spec basis (ITU-T H.264 §E.2.1 `max_num_reorder_frames`) and its
//!   latency are specified in [`crate::rtp_dts`]. The reorder itself is still
//!   reported — a wire timestamp earlier than a previous one raises
//!   [`RtpLossEvent::NonMonotonicTimestamp`] (audit r04-W29).
//! - Each track independently rebases its first unwrapped timestamp to
//!   `start_decode_time = 0` (via the caller building [`crate::media::Track`]
//!   from `track_specs` + emitted samples) **unless** an RTCP Sender Report
//!   has been fed for at least two tracks (issue #722): RTP timestamps alone
//!   carry no cross-stream relationship (RFC 3550 §5.1 — each SSRC's clock
//!   has an arbitrary random offset), so recovering true A/V sync needs the
//!   NTP-wallclock ↔ RTP-timestamp correlation each Sender Report carries
//!   (RFC 3550 §6.4.1). Feed reports via [`RtpStreamDepacketiser::push_sender_report`]
//!   / [`RtpStreamDepacketiser::push_rtcp`] as they arrive on a track's RTCP
//!   channel, then read [`RtpStreamDepacketiser::sync_start_decode_times`]
//!   once at least two tracks have an anchor: it maps every anchored
//!   track's first sample onto one common wallclock and returns each
//!   track's `start_decode_time` (in that track's own `clock_rate` ticks)
//!   relative to the earliest of them — preserving the real inter-track
//!   offset instead of discarding it. This is strictly additive: with no
//!   Sender Reports fed (or fewer than two anchored tracks), the method
//!   returns an empty `Vec` and callers keep the v1
//!   independent-rebase-to-0 behaviour unchanged.
//! - When an RFC 3640 `AAC-hbr` packet aggregates more than one access unit,
//!   all AUs in that packet share the RTP timestamp (§3.2.3.1: the timestamp
//!   is the first AU's, the rest follow back to back). Once the next packet's
//!   timestamp is known, the group's step is divided evenly across its AUs, so
//!   each gets its own `pts` and a non-zero `duration`.
//!
//! # Loss and reorder detection (issue #779)
//!
//! RTP runs over UDP: loss, reordering, and duplication are all normal, not
//! exceptional. Before this, [`push`](RtpStreamDepacketiser::push) decided access-unit
//! boundaries purely from the RTP timestamp and the marker bit, and never
//! looked at the sequence number at all — a dropped FU-A fragment was
//! concatenated with its neighbours into a malformed access unit and handed
//! downstream with no diagnostic trail.
//!
//! Each track now keeps RFC 3550 §5.1 sequence-number state, compared with
//! **wrapping** arithmetic (never `>` — the field is 16 bits and wraps every
//! 65536 packets, exactly like this module's own `unwrap_ts` for the
//! 32-bit timestamp). RFC 3550 §A.1's `update_seq` is the standard's
//! validity-check algorithm; the discipline it embodies (wrapping
//! comparison, SSRC-scoped state, treating a new SSRC as a new source) is
//! transcribed and cited in full at
//! `transmux/docs/rtp/rtp-sequence-validation.md`, along with a precise
//! account of where this implementation follows it and where it must
//! diverge (that RFC's algorithm classifies validity for RTCP statistics; it
//! never reorders, because nothing in plain RTCP loss accounting needs
//! packets delivered in order — H.264 FU-A reassembly does).
//!
//! - **Source validation (§A.1 `probation`)**: the first packet of a session
//!   takes the new-source branch and does *not* run `update_seq`, so the
//!   `MIN_SEQUENTIAL` = 2 sequential packets §A.1 requires complete on the
//!   **third** packet seen. No media is delayed for probation (see
//!   `SeqState::admit`), so this only means the source is declared valid one
//!   packet later than §A.1's own count would suggest.
//! - **In order**: reassembled immediately, as before.
//! - **Reordered within a small window**: held in a bounded buffer
//!   (`RtpStreamTrack::with_reorder_depth`, default [`DEFAULT_REORDER_DEPTH`])
//!   keyed by wire sequence number, and replayed in the correct order once
//!   the gap fills — reassembly then sees exactly the in-order byte stream.
//!   The buffer is a **hard bound**: this project has already shipped four
//!   unbounded-allocation vectors from RTP/TS input (`MAX_AU_BUFFER_BYTES`
//!   above is one of the fixes), so the reorder buffer never grows past its
//!   configured depth (momentarily `depth + 1` while a just-arrived packet
//!   is considered for the "closest to the hole" resume choice, then
//!   immediately collapsed back down — see `SeqState::force_resolve`).
//! - **Gap** (the buffer's window is exhausted, or end of stream forces a
//!   decision — [`RtpStreamDepacketiser::flush`]): the access unit under construction, if
//!   any, is dropped rather than reassembled from a run missing a fragment;
//!   [`RtpLossEvent::SequenceGap`] is recorded.
//! - **Duplicate or misordered** (behind what the buffer waits for, or already
//!   sitting in it): discarded silently, exactly as RFC 3550 §A.1's
//!   fall-through "duplicate or reordered packet" arm treats it. Two costs are
//!   inherent to that rule and worth stating:
//!   - a jump **at most `MAX_MISORDER` (100) behind** `max_seq` is *accepted* by
//!     §A.1 as misordered, and only then found to be behind `expected` and
//!     dropped by the reorder buffer — so a backward move of up to 100 sequence
//!     numbers is lost media with no [`RtpLossEvent`]. §A.1 treats it that way
//!     because for loss statistics a small backward step is indistinguishable
//!     from a reorder, and a real seek or restart moves much further;
//!   - the first packet of a post-seek run is the one that *arms* `bad_seq`, so
//!     it is discarded (§A.1 returns 0 for it). A run whose first packet was an
//!     FU-A start therefore loses that fragment, which damages the access unit
//!     it belonged to (`RtpLossEvent::DamagedAccessUnit`); the run itself
//!     continues from its second packet.
//! - **Wild jump needing confirmation** (§A.1: ahead of `max_seq` by
//!   `MAX_DROPOUT` or more, or behind by more than `MAX_MISORDER`): the packet
//!   is **discarded** and `bad_seq = (seq + 1) & (RTP_SEQ_MOD - 1)` is
//!   remembered. Only if the *next* packet carries `bad_seq` — §A.1's "two
//!   sequential packets ... the other side restarted without telling us" — is
//!   the source re-synced. One stray far-behind packet therefore cannot move
//!   the stream's baseline at all (the pre-fix bug), and a real seek or
//!   restart, which always brings a *run* of packets at the new numbering,
//!   is adopted on its second packet.
//! - **Resync** (the confirmed restart above, e.g. an RTSP seek or a source
//!   restart that kept its SSRC): the baseline moves onto the new numbering,
//!   the in-progress access unit is dropped (it straddles the seam) and the
//!   reorder buffer is cleared, and the track's timestamp origin is
//!   re-established — the new run's timeline does not inherit a delta measured
//!   across the discontinuity, because each source instance's RTP clock origin
//!   is random (§5.1). No [`RtpLossEvent`]: the sender renumbered, nothing was
//!   lost. Because §A.1's `init_seq` is all that carries over, the *next* run
//!   is an ordinary source from then on — nothing about the seam bounds it,
//!   so a run of any length (including one that wraps past the old numbering)
//!   is delivered in full, and a second seek resyncs the same way.
//!
//! ## Where the signal surfaces (design decision)
//!
//! Issue #778 (the MPEG-TS sibling of this issue — continuity-counter gaps
//! and `transport_error_indicator`) concluded that TR 101 290-style loss
//! belongs in the media plane's byte-level tap, *not* in the demux family's
//! own [`crate::ir::DemuxEvent`] vocabulary, because detecting a CC gap is a
//! pure re-scan of raw bytes: `media-doctor`, `dvb-conformance`, and
//! `ts-fix` each already implement it independently, entirely outside any
//! demuxer, over the same bytes a demuxer merely happens to also be
//! parsing. Widening `DemuxEvent` there would have duplicated a detector
//! that already exists three times over, in the wrong layer.
//!
//! None of that applies here, and the signal surfaces locally instead — a
//! new [`RtpLossEvent`], polled via [`RtpStreamDepacketiser::poll_loss_event`], scoped to
//! this module rather than folded into `DemuxEvent` (whose own doc comment
//! already draws its boundary at "the demux family's own vocabulary" —
//! `StreamingTsDemux` + `StreamingFlvDemux` — and this depacketiser is a
//! different spoke of the hub, not a member of that family):
//!
//! 1. **No detector to bridge.** Unlike #778, there is no existing RTP loss
//!    monitor anywhere in this workspace to wire in; it has to be written
//!    somewhere, and there is no "elsewhere" that already does this.
//! 2. **The signal and the corrective action share one piece of state.**
//!    A CC gap is diagnostic-only-by-construction: a tap can compute
//!    "sequence N is missing" from the wire bytes alone, without needing to
//!    know anything about framing, because nothing about *acting* on that
//!    gap requires framing state (TS payloads are independently
//!    re-synchronisable). An RTP sequence gap is different: knowing a
//!    packet is missing is only useful once paired with knowing *which*
//!    in-progress NAL it was fragmenting — and that is `TrackState`'s own
//!    `cur_pkts`/`cur_ts`, private reassembly state a separate tap could not
//!    see without independently re-implementing this entire module's FU-A
//!    tracking. The gap decision and the "drop this access unit" decision
//!    are the same decision, made from the same state, at the same moment —
//!    so they belong in the same place.
//!
//! A future consumer wanting this folded into a wider cross-container
//! vocabulary (mirroring `DemuxEvent`) remains free to do so additively —
//! [`RtpLossEvent`] is `#[non_exhaustive]` for exactly that reason — but
//! that is a decision for whoever builds that cross-container layer, not one
//! this module should presume.

use crate::TrackId;
use crate::error::{Error, Result};
use crate::pipeline::{CodecConfig, Sample, TrackSpec};
use crate::rtcp::SenderReport;
use crate::rtp::{RtpMediaKind, parse_rtp_header, reassemble_audio, reassemble_video};
use crate::rtp_dts::DtsDeriver;
use crate::slice_header::{AvcParamSets, AvcPicture};
use alloc::collections::{BTreeMap, VecDeque};
use alloc::vec::Vec;
use broadcast_common::{Demand, Parse, Stage, Timestamp};

/// Number of fractional-second units in the 32.32 fixed-point NTP timestamp
/// format an RTCP Sender Report's `ntp_msw`/`ntp_lsw` carry (RFC 3550 §6.4.1,
/// citing RFC 5905 §6 for the NTP timestamp format itself): `2^32`.
const NTP_FRACTION_SCALE: f64 = 4_294_967_296.0;

/// Hard cap, in bytes, on one track's in-progress access-unit buffer
/// (`TrackState::cur_pkts`, the raw RTP packets accumulated since the last
/// completed AU) — the same buffer that ultimately feeds
/// [`crate::rtp::reassemble_video`]'s FU-A `fu_buf`, so bounding it here
/// transitively bounds that too. A real access unit (even a 4K IDR frame)
/// is at most a few hundred KB; this is comfortably above that, but far
/// below what a malformed/hostile stream — a dropped final FU-A fragment
/// (`E=1` never seen) or a marker bit that's never set — would otherwise
/// accumulate for the life of the (indefinitely long, per the P0 reconnect
/// loop) session: an unbounded-memory DoS (audit-ingest #4). On overflow the
/// in-progress AU is dropped and [`RtpStreamDepacketiser::push`] returns a
/// recoverable [`Error::BufferCapExceeded`]; internal state is already reset
/// so the next packet starts a fresh AU (resync on the next timestamp change
/// or marker bit).
const MAX_AU_BUFFER_BYTES: usize = 4 * 1024 * 1024;

/// Default depth (in packets) of the bounded out-of-order reorder buffer
/// (see the module docs' "Loss and reorder detection" section and
/// `SeqState`), used unless a track is built with
/// [`RtpStreamTrack::with_reorder_depth`].
///
/// Chosen generously above the reorder distances a real RTSP/RTP session on
/// a managed network produces (typically a handful of packet positions at
/// most) while keeping the worst case tiny: `DEFAULT_REORDER_DEPTH + 1`
/// packets at the network MTU (a few KB) sit in the buffer at any instant,
/// nowhere near `MAX_AU_BUFFER_BYTES`. This is deliberate — this project
/// has already shipped four unbounded-allocation DoS vectors from RTP/TS
/// input (audit-ingest #4; `MAX_AU_BUFFER_BYTES` above is one of the fixes),
/// and RTP is untrusted remote input over UDP, so a fifth was not an option.
pub const DEFAULT_REORDER_DEPTH: usize = 16;

/// The 16-bit sequence-number space, RFC 3550 §A.1's `RTP_SEQ_MOD` (`#define
/// RTP_SEQ_MOD (1<<16)` in the RFC's own sample header).
const RTP_SEQ_MOD: u32 = 1 << 16;

/// `RTP_SEQ_MOD + 1`, the value RFC 3550 §A.1's `init_seq` stores in `bad_seq`
/// "so seq == bad_seq is false": it is one past the largest representable
/// sequence number, so no 16-bit `seq` can ever equal it.
const RTP_SEQ_MOD_PLUS_ONE: u32 = RTP_SEQ_MOD + 1;

/// RFC 3550 §A.1's `MAX_DROPOUT` (3000 there, "based on ... a maximum dropout
/// of 1 minute"). A jump *ahead* of `max_seq` smaller than this is treated as
/// an ordinary loss ("in order, with permissible gap"); at or above it the
/// number is a wild jump needing confirmation.
const MAX_DROPOUT: u16 = 3000;

/// RFC 3550 §A.1's `MAX_MISORDER` (100 there, "based on a maximum misordering
/// time of 2 seconds at 50 packets/second"). A number behind `max_seq` by at
/// most this much is a legal reorder and is accepted; further behind and it is
/// either a duplicate or part of a wild jump.
const MAX_MISORDER: u32 = 100;

/// RFC 3550 §A.1's `MIN_SEQUENTIAL`: how many sequential packets a source must
/// produce before it is declared valid. Two, because the RTCP header check is
/// strong and a larger number would delay validation when packets are lost.
const MIN_SEQUENTIAL: u8 = 2;

/// `RTP_SEQ_MOD - MAX_MISORDER`, narrowed to `u16` for the §A.1 comparison
/// `udelta <= RTP_SEQ_MOD - MAX_MISORDER`. `MAX_MISORDER` (100) is far below
/// `RTP_SEQ_MOD` (65536), so the subtraction cannot underflow (asserted in this
/// module's tests).
const MAX_WILD_JUMP_U16: u16 = (RTP_SEQ_MOD - MAX_MISORDER) as u16;

/// Loss/reorder signal from [`RtpStreamDepacketiser`] (RFC 3550 §5.1
/// sequence-number semantics; see the module docs' "Loss and reorder
/// detection" section and `transmux/docs/rtp/rtp-sequence-validation.md` for
/// the RFC 3550 §A.1 discipline this is adapted from, and why the signal
/// surfaces here rather than in [`crate::ir::DemuxEvent`]).
///
/// `#[non_exhaustive]`: a future variant (e.g. a distinct signal for a
/// changed SSRC) is additive.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RtpLossEvent {
    /// The sequence numbers strictly between `expected` and `got` were never
    /// recovered — either genuinely lost in transit, or reordered further
    /// than the bounded reorder window
    /// ([`RtpStreamTrack::with_reorder_depth`]) tolerates; from this
    /// depacketiser's point of view those are indistinguishable, and both
    /// require the same corrective action. Any access unit under
    /// construction at the moment the gap was detected was dropped rather
    /// than reassembled from a run missing a fragment.
    SequenceGap {
        /// The track this gap was observed on.
        track_id: TrackId,
        /// The SSRC (RFC 3550 §5.1) the gap was observed on.
        ssrc: u32,
        /// The sequence number that was expected next.
        expected: u16,
        /// The sequence number actually adopted as the new baseline.
        got: u16,
    },
    /// An access unit could not be reassembled after a [`Self::SequenceGap`]
    /// resync (e.g. the packet resumed on was itself an FU-A continuation
    /// fragment with no preceding start fragment in this run — RFC 6184
    /// §5.8) and was dropped. Always preceded, in the same
    /// [`RtpStreamDepacketiser::push`] call, by the `SequenceGap` that
    /// triggered the resync.
    DamagedAccessUnit {
        /// The track this occurred on.
        track_id: TrackId,
    },
    /// An access unit's RTP timestamp was **earlier** than the previous
    /// access unit's, i.e. the stream's presentation order is not wire order
    /// (a B-frame reorder).
    ///
    /// RTP carries exactly one timestamp per access unit and it is the
    /// *sampling/presentation* time (RFC 6184 §5.1 for H.264, RFC 3640 §3.3.1
    /// for AAC), so the wire alone cannot supply an independent decode time.
    /// The video `dts` is derived from the reorder depth ([`crate::rtp_dts`]),
    /// so a reordered source still yields a monotonic `dts`; this event reports
    /// that the wire order is not presentation order. Reported once per
    /// occurrence so a caller can count or route it.
    NonMonotonicTimestamp {
        /// The track this occurred on.
        track_id: TrackId,
        /// The previous access unit's unwrapped presentation timestamp, in the
        /// track's clock (the same unit as [`crate::ir::Sample::pts`]).
        previous_pts: i64,
        /// This access unit's, which is earlier.
        pts: i64,
    },
    /// An access unit's presentation time was earlier than a decode timestamp
    /// already issued, so the assumed reorder depth was too small (a declared
    /// `max_num_reorder_frames` the stream violates, or reordering deeper than
    /// the start-of-stream probe saw). The sample keeps a strictly increasing
    /// `dts`, which therefore exceeds its `pts`; the depth grows so later
    /// samples recover. See [`crate::rtp_dts`].
    ReorderDepthExceeded {
        /// The track this occurred on.
        track_id: TrackId,
        /// The reorder depth now in force, in frames.
        depth: usize,
        /// The offending sample's presentation timestamp, in the track's clock.
        pts: i64,
        /// The `dts` it was given.
        dts: i64,
    },
    /// A fully reassembled access unit was discarded without being emitted:
    /// the reorder window overflowed the 256-AU bound on AUs whose duration
    /// could not be resolved, or a sequence resync found held AUs for which no
    /// duration is known. One event per discarded AU, oldest first.
    DroppedAccessUnit {
        /// The track this occurred on.
        track_id: TrackId,
        /// The discarded AU's unwrapped presentation timestamp, in the track's
        /// clock (the same unit as [`crate::ir::Sample::pts`]).
        pts: i64,
    },
}

// `RtpLossEvent` is a data-carrying ADT (each variant is a distinct
// structured signal, not a flat spec code) — see this crate's
// `tests/label_coverage.rs` SKIP list (same category as `DemuxEvent`), so it
// is intentionally exempt from the #204 `name()`/`impl_spec_display!`
// convention.

/// One track's decode config for [`RtpStreamDepacketiser`].
#[non_exhaustive]
pub struct RtpStreamTrack {
    /// Track ID (matches the IR `track_id` this depayloader emits).
    pub track_id: TrackId,
    /// The payload format carried on this track's RTP stream.
    pub kind: RtpMediaKind,
    /// Real codec config (e.g. from [`crate::rtp_sdp`]).
    pub config: CodecConfig,
    /// RTP clock rate (Hz) — also used as the IR track timescale.
    pub clock_rate: u32,
    /// Bounded reorder-buffer depth (packets) — see [`DEFAULT_REORDER_DEPTH`].
    reorder_depth: usize,
}

impl RtpStreamTrack {
    /// Build a track config from its fields, with the default reorder-buffer
    /// depth ([`DEFAULT_REORDER_DEPTH`]).
    pub fn new(
        track_id: TrackId,
        kind: RtpMediaKind,
        config: CodecConfig,
        clock_rate: u32,
    ) -> Self {
        Self {
            track_id,
            kind,
            config,
            clock_rate,
            reorder_depth: DEFAULT_REORDER_DEPTH,
        }
    }

    /// Override the bounded reorder-buffer depth (packets) for this track —
    /// see [`DEFAULT_REORDER_DEPTH`] and the module docs' "Loss and reorder
    /// detection" section. `0` disables reordering entirely (any
    /// out-of-order arrival is immediately treated as a gap).
    pub fn with_reorder_depth(mut self, reorder_depth: usize) -> Self {
        self.reorder_depth = reorder_depth;
        self
    }
}

/// Per-(track, SSRC) RTP sequence-number tracking state.
///
/// Two layers, because RFC 3550 §A.1 and an H.264 FU-A reassembler want
/// different things from the same field:
///
/// 1. **[`SeqValidity`]** — a faithful transcription of §A.1's `update_seq`,
///    including `init_seq`, `probation`/`MIN_SEQUENTIAL`, `MAX_DROPOUT`,
///    `MAX_MISORDER`, `bad_seq` and `cycles`. It classifies each arriving
///    sequence number as in-order-with-tolerable-gap, duplicate/misordered, a
///    wild jump needing confirmation, or a confirmed restart, using only the
///    wrapping arithmetic §A.1 prescribes.
/// 2. **The reorder buffer** (`held`/`expected`) — §A.1 never reorders,
///    because RTCP loss accounting does not need packets *delivered* in order.
///    FU-A reassembly does (RFC 6184 §5.8: fragment order *is* the data), so
///    an accepted packet ahead of the hole is held briefly and replayed in
///    order once the hole fills or the bound is reached (a declared gap).
#[derive(Default)]
struct SeqState {
    /// The SSRC this state was initialised for. `None` until the first
    /// packet is seen. A *changed* SSRC (RFC 3550 §8.2: a new source, e.g. a
    /// stream restart) resets tracking rather than raising a gap.
    ssrc: Option<u32>,
    /// Next sequence number the reorder buffer is waiting for.
    expected: Option<u16>,
    /// Packets received ahead of `expected`, held (raw wire bytes, keyed by
    /// their own sequence number) so they can be replayed in order once the
    /// hole fills. Bounded: never holds more than `reorder_depth` entries at
    /// rest (see [`Self::admit`] / [`Self::force_resolve`]).
    held: Vec<(u16, Vec<u8>)>,
    /// RFC 3550 §A.1's per-source validity state (`max_seq`/`bad_seq`/
    /// `cycles`/`probation`), kept in one field so "re-initialise the source"
    /// is a single call and cannot half-reset.
    validity: SeqValidity,
}

/// RFC 3550 §A.1's per-source validity state — a transcription of the RFC's
/// `source` fields that `update_seq` reads and writes (`base_seq`/`received`/
/// `expected_prior` are RTCP-statistics bookkeeping this depacketiser does not
/// keep, since it emits its own [`RtpLossEvent`] signal instead).
struct SeqValidity {
    /// §A.1 `max_seq`: the highest sequence number accepted so far (or the
    /// most recently seen one while in probation).
    max_seq: u16,
    /// §A.1 `bad_seq`: the number that, if the next packet carries it, means
    /// "two sequential packets" arrived at a wild jump and the source has
    /// restarted. `RTP_SEQ_MOD + 1` (i.e. not representable) after
    /// `init_seq`, "so seq == bad_seq is false".
    bad_seq: u32,
    /// §A.1 `cycles`: `RTP_SEQ_MOD` per wrapped 64k cycle. Counted and
    /// exposed for completeness, though this depacketiser's own ordering comes
    /// from the reorder buffer rather than the cycle count.
    cycles: u32,
    /// §A.1 `probation`: non-zero until `MIN_SEQUENTIAL` consecutive packets
    /// have been seen, so a stray packet cannot declare a source valid.
    probation: u8,
}

impl Default for SeqValidity {
    fn default() -> Self {
        Self {
            max_seq: 0,
            bad_seq: RTP_SEQ_MOD_PLUS_ONE,
            cycles: 0,
            probation: 0,
        }
    }
}

impl SeqValidity {
    /// §A.1's `init_seq(s, seq)`: reset the per-source statistics to a known
    /// starting point. `bad_seq` is left at the "cannot match" sentinel, since
    /// that is what `init_seq` does.
    fn init_seq(&mut self, seq: u16) {
        self.max_seq = seq;
        self.bad_seq = RTP_SEQ_MOD_PLUS_ONE;
        self.cycles = 0;
        self.probation = 0;
    }

    /// §A.1's `update_seq(s, seq)`, transcribed arm for arm. Returns
    /// [`SeqVerdict::Valid`] when §A.1 returns 1 (and a restart when the
    /// "two sequential packets" rule fired), [`SeqVerdict::WildJumpConsumed`]
    /// when §A.1 returns 0 having stored `bad_seq` (the packet is discarded but
    /// arms the confirmation), and [`SeqVerdict::WildJumpIgnored`] when
    /// §A.1 returns 0 from the probation branch.
    fn update_seq(&mut self, seq: u16) -> SeqVerdict {
        // u_int16 udelta = seq - s->max_seq
        let udelta = seq.wrapping_sub(self.max_seq);

        // "Source is not valid until MIN_SEQUENTIAL packets with sequential
        // sequence numbers have been received."
        if self.probation != 0 {
            if seq == self.max_seq.wrapping_add(1) {
                self.probation -= 1;
                self.max_seq = seq;
                if self.probation == 0 {
                    self.init_seq(seq);
                    return SeqVerdict::Valid { restarted: false };
                }
            } else {
                self.probation = MIN_SEQUENTIAL - 1;
                self.max_seq = seq;
            }
            return SeqVerdict::Probation;
        }

        if udelta < MAX_DROPOUT {
            // "in order, with permissible gap"
            if seq < self.max_seq {
                // "Sequence number wrapped - count another 64K cycle."
                self.cycles = self.cycles.wrapping_add(RTP_SEQ_MOD);
            }
            self.max_seq = seq;
            SeqVerdict::Valid { restarted: false }
        } else if udelta <= MAX_WILD_JUMP_U16 {
            // "the sequence number made a very large jump"
            if u32::from(seq) == self.bad_seq {
                // "Two sequential packets -- assume that the other side
                // restarted without telling us so just re-sync (i.e., pretend
                // this was the first packet)."
                self.init_seq(seq);
                SeqVerdict::Valid { restarted: true }
            } else {
                self.bad_seq = u32::from(seq.wrapping_add(1)) & (RTP_SEQ_MOD - 1);
                SeqVerdict::WildJumpArmed
            }
        } else {
            // "duplicate or reordered packet" — accepted, no state change.
            SeqVerdict::Valid { restarted: false }
        }
    }
}

/// The outcome of one RFC 3550 §A.1 `update_seq` call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SeqVerdict {
    /// §A.1 returned 1: the number is valid. `restarted` is set when the
    /// "two sequential packets at a wild jump" rule re-synced the source.
    Valid { restarted: bool },
    /// §A.1 returned 0 from the probation branch — the source is not yet
    /// declared valid and the packet is not counted.
    Probation,
    /// §A.1 returned 0 having stored `bad_seq` for a wild jump: the packet is
    /// discarded, and the *next* packet must carry `bad_seq` to re-sync.
    WildJumpArmed,
}

/// One [`SeqState::admit`] call's outcome.
struct Admission {
    /// `Some((expected, got))` if a gap was declared as a result — the
    /// caller must drop any access unit under construction.
    gap: Option<(u16, u16)>,
    /// Whether the sequence numbering itself moved (a source restart, RFC 3550
    /// §A.1). There is no missing media to report, but any access unit under
    /// construction is still known-broken across the seam and must be
    /// dropped, and the depacketiser's timeline assumptions must be
    /// re-established.
    resynced: bool,
    /// Packets released for reassembly this call, in sequence order (may be
    /// empty — held for reorder, or a discarded duplicate).
    released: Vec<Vec<u8>>,
}

impl SeqState {
    /// Feed one packet's SSRC + sequence number through the gate.
    ///
    /// Order of decisions:
    ///
    /// 1. A changed/unknown SSRC re-initialises the source (§8.2).
    /// 2. §A.1's `update_seq` classifies the number. A rejected number
    ///    (`WildJumpArmed`) is discarded outright; the *next* wild number may
    ///    re-sync (`Valid { restarted: true }`), which also resets the reorder
    ///    buffer.
    /// 3. A number at or behind `expected` is a duplicate/late arrival for the
    ///    reorder buffer and is dropped silently (it is either already
    ///    delivered or already given up on).
    /// 4. Otherwise the packet is released, or held if it is ahead of a hole
    ///    that the bounded reorder buffer can still wait out.
    fn admit(&mut self, reorder_depth: usize, ssrc: u32, seq: u16, packet: &[u8]) -> Admission {
        // A new source (first packet ever, or a changed SSRC — RFC 3550
        // §8.2) resyncs without a gap: there is nothing to have lost yet.
        if self.ssrc != Some(ssrc) || self.expected.is_none() {
            self.ssrc = Some(ssrc);
            self.expected = Some(seq.wrapping_add(1));
            self.held.clear();
            // §A.1: "s->max_seq = seq - 1; s->probation = MIN_SEQUENTIAL".
            self.validity.max_seq = seq.wrapping_sub(1);
            self.validity.bad_seq = RTP_SEQ_MOD_PLUS_ONE;
            self.validity.cycles = 0;
            self.validity.probation = MIN_SEQUENTIAL;
            return Admission {
                gap: None,
                resynced: false,
                released: alloc::vec![packet.to_vec()],
            };
        }

        match self.validity.update_seq(seq) {
            SeqVerdict::Probation => {
                // §A.1 keeps a source on probation until `MIN_SEQUENTIAL`
                // sequential packets have been seen, and an implementation may
                // discard or *delay* those packets. This depacketiser delays
                // none of them: the caller's RTSP `SETUP`/SDP negotiation has
                // already established the session and its SSRC, so there is no
                // unknown source to screen out, and dropping the first packets
                // of a stream would lose media for no benefit. §A.1's own
                // rejection applies to the *statistics*, not to the media; the
                // reorder buffer still gets its normal treatment below, and
                // probation's `max_seq` bookkeeping is what stops a stray
                // number from being read as a restart.
            }
            SeqVerdict::WildJumpArmed => {
                // §A.1 returned 0 having armed `bad_seq`: the packet is
                // discarded, and the reorder buffer is not touched, so a stray
                // far-behind number cannot drag the live run backwards (the
                // pre-fix bug: one late packet 101 behind resynced
                // immediately).
                return Admission {
                    gap: None,
                    resynced: false,
                    released: Vec::new(),
                };
            }
            SeqVerdict::Valid { restarted } => {
                if restarted {
                    // §A.1's "two sequential packets ... just re-sync". The
                    // new run starts here: adopt it and drop everything the
                    // old one left, including a hole the buffer was waiting
                    // on. Nothing about the seam persists (only `init_seq`'s
                    // state carries over), so the new run is an ordinary
                    // source from here on however far it climbs — a run that
                    // passes the old numbering, or the 16-bit wrap, is
                    // delivered in full, and a later seek resyncs the same
                    // way.
                    self.expected = Some(seq.wrapping_add(1));
                    self.held.clear();
                    return Admission {
                        gap: None,
                        resynced: true,
                        released: alloc::vec![packet.to_vec()],
                    };
                }
            }
        }

        let expected = self.expected.expect("checked above");
        if seq == expected {
            self.expected = Some(seq.wrapping_add(1));
            let mut released = alloc::vec![packet.to_vec()];
            self.drain_contiguous(&mut released);
            return Admission {
                gap: None,
                resynced: false,
                released,
            };
        }
        if (seq.wrapping_sub(expected) as i16) < 0 {
            // Behind what the reorder buffer is waiting for: a duplicate of a
            // packet already delivered, or a very late arrival for one already
            // declared lost. RFC 3550 §A.1's fall-through "duplicate or
            // reordered packet" arm takes no corrective action.
            return Admission {
                gap: None,
                resynced: false,
                released: Vec::new(),
            };
        }
        if self.held.iter().any(|(s, _)| *s == seq) {
            // Duplicate of an already-held future packet.
            return Admission {
                gap: None,
                resynced: false,
                released: Vec::new(),
            };
        }
        self.held.push((seq, packet.to_vec()));
        if self.held.len() <= reorder_depth {
            // Still within the bounded window: wait.
            return Admission {
                gap: None,
                resynced: false,
                released: Vec::new(),
            };
        }
        // The buffer's bound is reached (momentarily `reorder_depth + 1`,
        // collapsed back down by `force_resolve` below, never sustained):
        // the hole cannot be waited out any longer. Force a decision.
        let (expected, got, released) = self
            .force_resolve()
            .expect("held is non-empty: just pushed to it");
        Admission {
            gap: Some((expected, got)),
            resynced: false,
            released,
        }
    }

    /// Force a decision when the buffer can wait no longer (its bound was
    /// reached mid-stream, or end of stream — [`RtpStreamDepacketiser::flush`]
    /// — forces one): pick the held packet with the smallest wrapping
    /// distance from `expected` (RFC 3550 §A.1's wrapping-distance
    /// discipline, applied to choosing a resume point rather than
    /// classifying validity), declare the sequence numbers strictly between
    /// `expected` and it lost, adopt it as the new baseline, and drain
    /// whatever else in `held` is now consecutively reachable from there —
    /// so a run of packets that arrived correctly *after* the one genuinely
    /// lost packet is never discarded along with it. Returns `None` if
    /// `held` is empty (nothing to resolve).
    fn force_resolve(&mut self) -> Option<(u16, u16, Vec<Vec<u8>>)> {
        let expected = self.expected?;
        if self.held.is_empty() {
            return None;
        }
        let idx = self
            .held
            .iter()
            .enumerate()
            .min_by_key(|(_, (s, _))| s.wrapping_sub(expected))
            .map(|(i, _)| i)?;
        let (got, bytes) = self.held.remove(idx);
        self.expected = Some(got.wrapping_add(1));
        let mut released = alloc::vec![bytes];
        self.drain_contiguous(&mut released);
        Some((expected, got, released))
    }

    /// Drain any packets in `held` that are now consecutively next after
    /// `self.expected`, advancing `self.expected` and appending each to
    /// `out` — the mechanism that reassembles a reordered run into the
    /// original wire order once the hole it was waiting on fills.
    fn drain_contiguous(&mut self, out: &mut Vec<Vec<u8>>) {
        // `self.expected` is re-read on every iteration because the body
        // advances it, so `while let` is equivalent to the previous
        // `loop { let … else break }` and satisfies `clippy::while_let_loop`.
        while let Some(expected) = self.expected {
            let Some(pos) = self.held.iter().position(|(s, _)| *s == expected) else {
                break;
            };
            let (_, bytes) = self.held.remove(pos);
            out.push(bytes);
            self.expected = Some(expected.wrapping_add(1));
        }
    }
}

/// Per-track mutable depacketisation state.
struct TrackState {
    kind: RtpMediaKind,
    config: CodecConfig,
    clock_rate: u32,
    /// RTP timestamp of the packets currently buffered in `cur_pkts` (the
    /// access unit still being assembled).
    cur_ts: Option<u32>,
    /// Packets accumulated for the current (not-yet-complete) RTP timestamp.
    cur_pkts: Vec<Vec<u8>>,
    /// Total bytes across `cur_pkts` — tracked incrementally so
    /// [`MAX_AU_BUFFER_BYTES`] can be enforced without re-summing the vec on
    /// every packet.
    cur_bytes: usize,
    /// Unwrapped 64-bit form of the most recently completed AU's timestamp.
    last_unwrapped: Option<u64>,
    /// Unwrapped 64-bit form of this track's very first completed AU's
    /// timestamp — the anchor [`RtpStreamDepacketiser::sync_start_decode_times`]
    /// measures this track's RTCP SR offset against.
    first_unwrapped: Option<u64>,
    /// AUs awaiting their duration (filled in once the next AU's timestamp is
    /// known). More than one only for an audio packet that aggregates several
    /// AUs under one RTP timestamp (RFC 3640 §3.2.3.1).
    pending: Vec<PendingAu>,
    /// Last computed duration, reused for the final AU emitted by `flush`.
    /// Always non-zero once set; `0` means no real duration has been observed.
    last_duration: u32,
    /// Per-AU duration (in `clock_rate` ticks) declared by the stream's own
    /// codec config: H.264 SPS VUI timing with `fixed_frame_rate_flag`, or the
    /// AAC `AudioSpecificConfig` frame length. `None` when not declared.
    declared_duration: Option<u32>,
    /// Set once a wire timestamp has stepped backward (a B-frame reorder). From
    /// then on `pending` is a reorder window resolved by
    /// [`RtpStreamDepacketiser::resolve_window`].
    reordered: bool,
    /// Presentation timestamps of the reorder window and of recently resolved
    /// AUs, so an AU's presentation successor is found even when it was sent
    /// (and resolved) before it. Pruned below the oldest unresolved AU.
    seen_pts: BTreeMap<u64, u64>,
    /// Wire-order index the next AU of this track takes.
    next_seq: u64,
    /// Largest number of wire-order positions by which an AU's presentation
    /// predecessor preceded it so far: the reorder window the track needs.
    au_window: usize,
    /// Smallest positive step seen between consecutive wire timestamps — the
    /// frame-period estimate used as the duration fallback when a step goes
    /// backward (a presentation reorder has no forward delta). `None` until
    /// such a step has been seen.
    min_positive_step: Option<i64>,
    /// This track's most recent RTCP Sender Report anchor, if any has been
    /// fed via [`RtpStreamDepacketiser::push_sender_report`].
    sr_anchor: Option<SrAnchor>,
    /// Bounded reorder-buffer depth for this track (packets) — see
    /// [`DEFAULT_REORDER_DEPTH`] / [`RtpStreamTrack::with_reorder_depth`].
    reorder_depth: usize,
    /// RFC 3550 §5.1 sequence-number tracking state (issue #779) — see the
    /// module docs' "Loss and reorder detection" section.
    seq: SeqState,
    /// Decode-timestamp derivation for video (see [`crate::rtp_dts`]).
    dts: DtsDeriver,
    /// SPS/PPS known for this track (config plus in-band), to read
    /// `field_pic_flag` from slice headers.
    param_sets: AvcParamSets,
}

/// One track's RTCP Sender Report wallclock anchor (RFC 3550 §6.4.1): the NTP
/// wallclock instant at which the sender's RTP clock read `raw_rtp_ts`.
///
/// `raw_rtp_ts` is kept in wire (32-bit) form rather than eagerly unwrapped:
/// an SR can be fed before, interleaved with, or after the access units it
/// anchors, so it is unwrapped lazily in [`wall_seconds`] against whichever
/// AU's own unwrapped timestamp needs a wallclock value — always valid
/// because a real SR's RTP timestamp is at most a few RTCP-interval seconds
/// from the AUs it anchors, far under half the 32-bit wrap range.
struct SrAnchor {
    /// NTP wallclock time, in seconds (32.32 fixed-point `ntp_msw`/`ntp_lsw`
    /// converted to `f64`).
    ntp_seconds: f64,
    /// The wire-form (32-bit) RTP timestamp corresponding to `ntp_seconds`.
    raw_rtp_ts: u32,
}

/// Resolve one access unit's NTP wallclock instant from a track's SR anchor:
/// `wall(au) = anchor.ntp_seconds + (unwrapped_au_ts − unwrapped_anchor_ts) /
/// clock_rate` (RFC 3550 §6.4.1's NTP/RTP correlation, generalised from the
/// SR's own instant to any AU on the same RTP clock).
fn wall_seconds(anchor: &SrAnchor, clock_rate: u32, au_unwrapped_ts: u64) -> f64 {
    let anchor_unwrapped = unwrap_ts(Some(au_unwrapped_ts), anchor.raw_rtp_ts);
    let delta_ticks = au_unwrapped_ts as i128 - i128::from(anchor_unwrapped);
    anchor.ntp_seconds + (delta_ticks as f64) / f64::from(clock_rate)
}

/// An access unit whose duration is not yet known (waiting on the next AU's
/// timestamp).
struct PendingAu {
    unwrapped_ts: u64,
    is_sync: bool,
    data: Vec<u8>,
}

/// Stateful, timing- and config-aware RTP depayloader (see module docs).
pub struct RtpStreamDepacketiser {
    tracks: Vec<(TrackId, TrackState)>,
    /// Output buffered for the [`Stage`] adapter (media plane step 2e), which
    /// — unlike [`push`](Self::push)/[`flush`](Self::flush) — cannot return
    /// samples directly (drained via [`Stage::poll`] instead). Unused by the
    /// inherent API.
    stage_ready: VecDeque<Sample>,
    /// Loss/reorder signals raised by [`Self::push`]/[`Self::flush`],
    /// drained via [`Self::poll_loss_event`] — see the module docs' "Loss
    /// and reorder detection" section.
    loss_events: VecDeque<RtpLossEvent>,
}

/// Unwrap a 32-bit wire RTP timestamp against the last unwrapped value.
///
/// RTP timestamps are a 32-bit field (RFC 3550 §5.1) that wraps every
/// `2^32` ticks. Given the previous unwrapped 64-bit value, this takes its
/// low 32 bits and computes
/// the wire-to-wire delta via a wrapping subtraction reinterpreted as a
/// signed 32-bit integer — the standard idiom for unwrapping any monotonic
/// counter that wraps modulo `2^N` (the same trick used for TCP sequence
/// numbers). That signed delta (forward *or* backward) is then added to the
/// full-width previous value.
///
/// This is exact for any single step whose true magnitude is within half the
/// wrap range (~13.6 hours at a 90 kHz clock) — i.e. any real packet
/// arrival, in order or mildly reordered, between two AUs. It cannot tell a
/// giant forward jump from an equally giant backward jump (an inherent
/// ambiguity of wraparound counters), but that distinction only matters at
/// magnitudes no real RTP stream produces between consecutive AUs.
fn unwrap_ts(prev: Option<u64>, ts: u32) -> u64 {
    let Some(prev) = prev else {
        return u64::from(ts);
    };
    let prev_low = prev as u32;
    let delta = ts.wrapping_sub(prev_low) as i32;
    if delta >= 0 {
        prev + u64::from(delta as u32)
    } else {
        prev.saturating_sub(u64::from(delta.unsigned_abs()))
    }
}

/// Narrow an unwrapped RTP media-clock value into the `i64` range
/// [`Sample::dts`]/[`Sample::pts`] carry (media plane step 2c). The unwrapped
/// clock is a `u64` accumulating whole `2^32` windows, so this only clamps at
/// values no real stream reaches (`i64::MAX` ticks is ~3 million years at
/// 90 kHz) — it makes the narrowing checked rather than a silent truncation.
fn to_ticks(uw: u64) -> i64 {
    uw.min(i64::MAX as u64) as i64
}

/// Cap on how many access units a reordered video track holds waiting for the
/// presentation successor of its oldest AU, and on the learned `au_window`
/// (the largest wire-order distance observed between an AU and its
/// presentation predecessor, used while no frame period is known); a hostile
/// timestamp series cannot grow either past this.
pub(crate) const MAX_REORDER_DEPTH: usize = 16;

/// Hard bound on a reorder window whose oldest AU cannot be given a truthful
/// duration (no later timestamp, no known frame period): beyond it the oldest
/// is dropped rather than letting the window grow with the stream.
const MAX_REORDER_WINDOW: usize = 256;

/// Per-AU duration, in `clock_rate` ticks, that the stream's own config
/// declares. H.264: the SPS VUI `timing_info` frame period
/// `2 * num_units_in_tick / time_scale` (ITU-T H.264 §E.2.1); with
/// `fixed_frame_rate_flag` 0 it is the stream's nominal period rather than a
/// constraint, which is still its own declaration and is used only after every
/// observed duration. AAC: `AudioSpecificConfig::samples_per_frame` scaled from the
/// ASC sampling rate to the RTP clock rate. `None` when nothing is declared.
fn declared_au_duration(kind: RtpMediaKind, config: &CodecConfig, clock_rate: u32) -> Option<u32> {
    let ticks = match (kind, config) {
        (RtpMediaKind::H264, CodecConfig::Avc { config, .. }) => {
            let sps = config.config.sps.first()?;
            let timing = crate::sps::decode_avc_vui_timing(&sps.0).ok().flatten()?;
            if timing.num_units_in_tick == 0 || timing.time_scale == 0 {
                return None;
            }
            let num =
                u64::from(clock_rate) * VUI_UNITS_PER_FRAME * u64::from(timing.num_units_in_tick);
            let den = u64::from(timing.time_scale);
            (num + den / 2) / den
        }
        (RtpMediaKind::Aac, CodecConfig::Aac { esds, .. }) => {
            let asc =
                crate::aac_asc::AudioSpecificConfig::parse(esds.decoder_specific_info_data()?)
                    .ok()?;
            let rate = u64::from(asc.effective_sampling_frequency()?);
            if rate == 0 {
                return None;
            }
            let num = u64::from(asc.samples_per_frame()?) * u64::from(clock_rate);
            (num + rate / 2) / rate
        }
        _ => return None,
    };
    u32::try_from(ticks).ok().filter(|&d| d > 0)
}

/// Reorder depth, in frames, that the stream's own H.264 SPS fixes
/// (`max_num_reorder_frames`, or the ITU-T H.264 §E.2.1 / §A.2.1 inference), or
/// `None` when it must be observed. Audio never reorders.
fn declared_reorder_depth(kind: RtpMediaKind, config: &CodecConfig) -> Option<usize> {
    match (kind, config) {
        (RtpMediaKind::H264, CodecConfig::Avc { config, .. }) => {
            let sps = config.config.sps.first()?;
            let depth = crate::sps::decode_avc_reorder_depth(&sps.0).ok()?.depth()?;
            usize::try_from(depth).ok()
        }
        _ => None,
    }
}

fn seed_param_sets(param_sets: &mut AvcParamSets, config: &CodecConfig) {
    if let CodecConfig::Avc { config, .. } = config {
        for nal in config
            .config
            .sps
            .iter()
            .map(|n| &n.0)
            .chain(config.config.pps.iter().map(|n| &n.0))
        {
            param_sets.observe_nal(nal);
        }
    }
}

fn sample_is_field_picture(param_sets: &mut AvcParamSets, sample: &Sample) -> bool {
    crate::annexb::iter_length_prefixed_nals(&sample.data)
        .map(|nals| matches!(param_sets.classify_nals(nals), AvcPicture::Field(_)))
        .unwrap_or(false)
}

/// Run a track's freshly emitted samples through its decode-timestamp
/// derivation (video only; see [`crate::rtp_dts`]).
fn derive_dts(
    st: &mut TrackState,
    loss_events: &mut VecDeque<RtpLossEvent>,
    track_id: TrackId,
    raw: Vec<Sample>,
    finish: bool,
) -> Vec<Sample> {
    if st.kind != RtpMediaKind::H264 {
        return raw;
    }
    let period = known_duration(st).map(i64::from);
    let mut out = Vec::with_capacity(raw.len());
    let mut exceeded = Vec::new();
    for s in raw {
        if sample_is_field_picture(&mut st.param_sets, &s) {
            st.dts.declare_field_coded();
        }
        st.dts.feed(s, period, &mut out, &mut exceeded);
    }
    if finish {
        st.dts.finish(period, &mut out, &mut exceeded);
    }
    for e in exceeded {
        loss_events.push_back(RtpLossEvent::ReorderDepthExceeded {
            track_id,
            depth: e.depth,
            pts: e.pts,
            dts: e.dts,
        });
    }
    out
}

/// Ticks of the VUI `time_scale` clock per frame: a frame spans two field
/// ticks (ITU-T H.264 §E.2.1, `time_scale / (2 * num_units_in_tick)`).
const VUI_UNITS_PER_FRAME: u64 = 2;

/// The truthful per-AU duration for a track, in order: the last real
/// per-AU duration, the smallest positive timestamp step seen, then the
/// stream's declared frame timing (AAC prefers the declared frame length to a
/// step, which spans a whole aggregated packet). `None` when none of these exist; the
/// caller must then not invent one.
fn known_duration(st: &TrackState) -> Option<u32> {
    if st.last_duration != 0 {
        return Some(st.last_duration);
    }
    if st.kind == RtpMediaKind::Aac && st.declared_duration.is_some() {
        return st.declared_duration;
    }
    if let Some(p) = st.min_positive_step
        && let Ok(p) = u32::try_from(p)
        && p > 0
    {
        return Some(p);
    }
    st.declared_duration
}

impl RtpStreamDepacketiser {
    /// Build a depayloader for the given tracks (kind, config, clock rate).
    pub fn new(tracks: Vec<RtpStreamTrack>) -> Self {
        let tracks = tracks
            .into_iter()
            .map(|t| {
                let declared_duration = declared_au_duration(t.kind, &t.config, t.clock_rate);
                let reorder_depth = declared_reorder_depth(t.kind, &t.config);
                let mut param_sets = AvcParamSets::new();
                seed_param_sets(&mut param_sets, &t.config);
                (
                    t.track_id,
                    TrackState {
                        kind: t.kind,
                        config: t.config,
                        clock_rate: t.clock_rate,
                        cur_ts: None,
                        cur_pkts: Vec::new(),
                        cur_bytes: 0,
                        last_unwrapped: None,
                        first_unwrapped: None,
                        pending: Vec::new(),
                        last_duration: 0,
                        declared_duration,
                        reordered: false,
                        seen_pts: BTreeMap::new(),
                        next_seq: 0,
                        au_window: 0,
                        min_positive_step: None,
                        sr_anchor: None,
                        reorder_depth: t.reorder_depth,
                        seq: SeqState::default(),
                        dts: DtsDeriver::new(reorder_depth),
                        param_sets,
                    },
                )
            })
            .collect();
        Self {
            tracks,
            stage_ready: VecDeque::new(),
            loss_events: VecDeque::new(),
        }
    }

    /// Build the [`TrackSpec`]s (timescale = `clock_rate`) for init-segment
    /// construction.
    pub fn track_specs(&self) -> Vec<TrackSpec> {
        self.tracks
            .iter()
            .map(|(id, st)| TrackSpec::new(*id, st.clock_rate, st.config.clone()))
            .collect()
    }

    /// Replace the codec configuration carried for one track (used by callers
    /// that learn dimensions from an in-band parameter set after SETUP).
    pub fn set_track_config(&mut self, track_id: TrackId, config: CodecConfig) {
        if let Some((_, st)) = self.tracks.iter_mut().find(|(id, _)| *id == track_id) {
            st.declared_duration = declared_au_duration(st.kind, &config, st.clock_rate);
            st.dts
                .set_declared(declared_reorder_depth(st.kind, &config));
            seed_param_sets(&mut st.param_sets, &config);
            st.config = config;
        }
    }

    fn state(&mut self, track_id: TrackId) -> Option<&mut TrackState> {
        self.tracks
            .iter_mut()
            .find(|(id, _)| *id == track_id)
            .map(|(_, st)| st)
    }

    /// Drain the next pending loss/reorder signal, if any (FIFO across every
    /// track) — see the module docs' "Loss and reorder detection" section.
    /// A clean stream never produces one.
    pub fn poll_loss_event(&mut self) -> Option<RtpLossEvent> {
        self.loss_events.pop_front()
    }

    /// Feed one RTCP Sender Report (RFC 3550 §6.4.1) for `track_id`, anchoring
    /// this track's wallclock for [`Self::sync_start_decode_times`]. Replaces
    /// any previous anchor for the track (the most recent SR wins); unknown
    /// `track_id`s are silently ignored, matching [`Self::push`].
    pub fn push_sender_report(&mut self, track_id: TrackId, sr: SenderReport) {
        if let Some(st) = self.state(track_id) {
            let ntp_seconds = f64::from(sr.ntp_msw) + f64::from(sr.ntp_lsw) / NTP_FRACTION_SCALE;
            st.sr_anchor = Some(SrAnchor {
                ntp_seconds,
                raw_rtp_ts: sr.rtp_timestamp,
            });
        }
    }

    /// Parse `bytes` as a single RTCP Sender Report (RFC 3550 §6.4.1) and feed
    /// it to [`Self::push_sender_report`]. Anything that isn't a parseable SR
    /// (a Receiver Report, a different compound-packet member, malformed
    /// bytes) is silently ignored — this is a convenience wrapper for
    /// callers holding raw RTCP bytes off the wire, not a general RTCP
    /// dispatcher.
    pub fn push_rtcp(&mut self, track_id: TrackId, bytes: &[u8]) {
        if let Ok(sr) = SenderReport::parse(bytes) {
            self.push_sender_report(track_id, sr);
        }
    }

    /// Compute the `start_decode_time` (in each track's own `clock_rate`
    /// ticks) that rebases every SR-anchored track onto one common wallclock,
    /// preserving their real inter-track offset (issue #722; RFC 3550
    /// §6.4.1). The earliest anchored track's first sample becomes the
    /// origin (`start_decode_time = 0`); every other anchored track's
    /// `start_decode_time` is its first sample's wallclock distance from
    /// that origin, converted to its own clock rate.
    ///
    /// Returns an empty `Vec` — the v1 opt-out — unless at least two tracks
    /// have both received a Sender Report ([`Self::push_sender_report`]) and
    /// emitted a first sample. A track absent from the returned `Vec` (no
    /// anchor, or fewer than two anchored tracks overall) keeps the existing
    /// independent-rebase-to-0 behaviour: the caller should only apply the
    /// returned `start_decode_time` to tracks it names.
    pub fn sync_start_decode_times(&self) -> Vec<(TrackId, u64)> {
        let anchored: Vec<(TrackId, f64, u32)> = self
            .tracks
            .iter()
            .filter_map(|(id, st)| {
                let anchor = st.sr_anchor.as_ref()?;
                let first_ts = st.first_unwrapped?;
                Some((
                    *id,
                    wall_seconds(anchor, st.clock_rate, first_ts),
                    st.clock_rate,
                ))
            })
            .collect();
        if anchored.len() < 2 {
            return Vec::new();
        }
        let origin = anchored
            .iter()
            .map(|(_, wall, _)| *wall)
            .fold(f64::INFINITY, f64::min);
        anchored
            .into_iter()
            .map(|(id, wall, clock_rate)| {
                let raw_ticks = ((wall - origin) * f64::from(clock_rate)).max(0.0);
                // `f64::round` is a `std`-only inherent method (it needs
                // libm), unavailable in this crate's `no_std` core build —
                // round-half-up via a truncating cast instead, which is
                // exact for `round()`'s behaviour given `raw_ticks >= 0.0`.
                let ticks = (raw_ticks + 0.5) as u64;
                (id, ticks)
            })
            .collect()
    }

    /// Feed one RTP packet for `track_id`. Returns any [`Sample`]s that
    /// became fully timed as a result (zero, one — the AU this packet
    /// completed — the previous AU emitted with its now-known duration; or
    /// several, when a reordered run finishes draining — see the module
    /// docs' "Loss and reorder detection" section). Unknown `track_id`s are
    /// silently ignored (return an empty `Vec`).
    ///
    /// The sequence number is checked before anything else: an in-order or
    /// legally-duplicate/reordered-within-window packet is handled as
    /// before; a genuine gap drops any access unit under construction and
    /// records [`RtpLossEvent::SequenceGap`] (drained via
    /// [`Self::poll_loss_event`]) instead of silently reassembling a run
    /// missing a fragment.
    ///
    /// A **resync** (RFC 3550 §A.1: the source's sequence numbering moved,
    /// e.g. an RTSP seek or a restart that kept its SSRC) first emits every
    /// complete AU the reorder window still holds (those whose duration cannot
    /// be resolved are reported as [`RtpLossEvent::DroppedAccessUnit`]), then
    /// drops the in-progress AU and re-establishes the timestamp origin and
    /// the reorder state, because the
    /// RTP timestamp clock is unrelated across the seam (§5.1 — the origin
    /// is random per source instance). No `SequenceGap` is raised: nothing
    /// was lost, the sender simply renumbered.
    pub fn push(&mut self, track_id: TrackId, rtp_packet: &[u8]) -> Result<Vec<Sample>> {
        let hdr = parse_rtp_header(rtp_packet)?;
        let RtpStreamDepacketiser {
            tracks,
            loss_events,
            ..
        } = self;
        let Some((_, st)) = tracks.iter_mut().find(|(id, _)| *id == track_id) else {
            return Ok(Vec::new());
        };

        let adm = st
            .seq
            .admit(st.reorder_depth, hdr.ssrc, hdr.sequence, rtp_packet);
        let mut out = Vec::new();
        if adm.resynced {
            // The AU under construction straddles the seam (its later
            // fragments belong to a different numbering run), so it is
            // known-broken: drop it rather than hand a merged, malformed
            // sample downstream. The timestamp origin is re-established
            // from this packet, so the new run's samples start a fresh,
            // continuous timeline instead of inheriting a delta computed
            // across the discontinuity (`unwrap_ts` cannot know the two
            // runs' clocks are unrelated — RFC 3550 §5.1).
            let mut held = Vec::new();
            Self::drain_pending(st, loss_events, track_id, &mut held);
            for p in st.pending.drain(..) {
                loss_events.push_back(RtpLossEvent::DroppedAccessUnit {
                    track_id,
                    pts: to_ticks(p.unwrapped_ts),
                });
            }
            st.cur_pkts.clear();
            st.cur_bytes = 0;
            st.cur_ts = None;
            st.last_unwrapped = None;
            st.first_unwrapped = None;
            st.reordered = false;
            st.seen_pts.clear();
            st.next_seq = 0;
            st.au_window = 0;
            st.min_positive_step = None;
            let mut flushed = derive_dts(st, loss_events, track_id, held, true);
            for pkt in adm.released {
                Self::push_one(st, loss_events, track_id, pkt, &mut out)?;
            }
            flushed.extend(derive_dts(st, loss_events, track_id, out, false));
            return Ok(flushed);
        }
        if let Some((expected, got)) = adm.gap {
            // The AU under construction, if any, is now known-incomplete:
            // drop it rather than hand a merged, malformed sample
            // downstream (issue #779).
            st.cur_pkts.clear();
            st.cur_bytes = 0;
            st.cur_ts = None;
            loss_events.push_back(RtpLossEvent::SequenceGap {
                track_id,
                ssrc: hdr.ssrc,
                expected,
                got,
            });
        }
        for pkt in adm.released {
            Self::push_one(st, loss_events, track_id, pkt, &mut out)?;
        }
        Ok(derive_dts(st, loss_events, track_id, out, false))
    }

    /// The original single-packet accumulation logic (RTP timestamp /
    /// marker-bit AU boundaries, [`MAX_AU_BUFFER_BYTES`]) — factored out so
    /// [`Self::push`] can feed it zero, one, or many packets released by
    /// [`SeqState::admit`] in a single call (a drained reordered run). Only
    /// [`Error::BufferCapExceeded`] escapes as a hard error; a reassembly
    /// failure is caught by [`Self::drain_complete_or_discard`] instead, so
    /// one bad packet in a released batch never aborts the packets after it.
    fn push_one(
        st: &mut TrackState,
        loss_events: &mut VecDeque<RtpLossEvent>,
        track_id: TrackId,
        rtp_packet: Vec<u8>,
        out: &mut Vec<Sample>,
    ) -> Result<()> {
        let hdr = parse_rtp_header(&rtp_packet)?;
        let ts = hdr.timestamp;
        // `hdr` borrows the packet, which is moved into the AU buffer below.
        let marker = hdr.marker;

        // A timestamp change while packets are buffered means the previous
        // timestamp's packets already form a complete AU (defensive: covers
        // a dropped/missing marker bit).
        if let Some(cur) = st.cur_ts
            && cur != ts
            && !st.cur_pkts.is_empty()
        {
            Self::drain_complete_or_discard(st, loss_events, track_id, out);
        }
        st.cur_ts = Some(ts);
        st.cur_bytes += rtp_packet.len();
        st.cur_pkts.push(rtp_packet);

        // Runaway AU — a dropped final FU-A fragment or a marker bit that
        // never arrives would otherwise grow `cur_pkts` forever (see
        // `MAX_AU_BUFFER_BYTES`). Drop the partial unit and resync: the next
        // packet starts a fresh AU exactly as if this were the first packet
        // ever seen for this track.
        if st.cur_bytes > MAX_AU_BUFFER_BYTES {
            st.cur_pkts.clear();
            st.cur_bytes = 0;
            st.cur_ts = None;
            return Err(Error::BufferCapExceeded {
                what: "RTP access-unit reassembly",
                cap: MAX_AU_BUFFER_BYTES,
            });
        }

        // The video marker bit ends an AU immediately (RFC 6184 §5.1).
        if matches!(st.kind, RtpMediaKind::H264) && marker {
            Self::drain_complete_or_discard(st, loss_events, track_id, out);
            st.cur_ts = None;
        }
        Ok(())
    }

    /// [`Self::drain_complete`], but a reassembly failure — e.g. the packet
    /// resumed on after a [`RtpLossEvent::SequenceGap`] turns out to be an
    /// FU-A continuation fragment with no preceding start fragment in this
    /// run (RFC 6184 §5.8) — is caught and recorded as
    /// [`RtpLossEvent::DamagedAccessUnit`] instead of propagated:
    /// `drain_complete` already resets the accumulator
    /// (`core::mem::take`) before attempting reassembly, so there is no
    /// inconsistent state to clean up, and a stream that is actively
    /// recovering from loss must not have that recovery itself abort the
    /// caller's processing of any packets still to come.
    fn drain_complete_or_discard(
        st: &mut TrackState,
        loss_events: &mut VecDeque<RtpLossEvent>,
        track_id: TrackId,
        out: &mut Vec<Sample>,
    ) {
        if Self::drain_complete(st, loss_events, track_id, out).is_err() {
            loss_events.push_back(RtpLossEvent::DamagedAccessUnit { track_id });
        }
    }

    /// Flush a track at end-of-stream: reassemble any buffered packets, and
    /// emit the final pending AU using the last-known duration (there is no
    /// following AU to measure a real delta against).
    ///
    /// First forces a decision on anything still sitting in the reorder
    /// buffer awaiting a hole that never filled — end of stream is exactly
    /// as final as the buffer's bound being reached mid-stream (see
    /// `SeqState::force_resolve`); without this, a fragment run missing
    /// its still-buffered continuation would sit invisibly in the reorder
    /// buffer forever while `cur_pkts` (missing that fragment) got wrongly
    /// flushed below as if it were complete.
    pub fn flush(&mut self, track_id: TrackId) -> Result<Vec<Sample>> {
        let RtpStreamDepacketiser {
            tracks,
            loss_events,
            ..
        } = self;
        let Some((_, st)) = tracks.iter_mut().find(|(id, _)| *id == track_id) else {
            return Ok(Vec::new());
        };
        let mut out = Vec::new();

        if let Some((expected, got, released)) = st.seq.force_resolve() {
            let ssrc = st.seq.ssrc.unwrap_or(0);
            st.cur_pkts.clear();
            st.cur_bytes = 0;
            st.cur_ts = None;
            loss_events.push_back(RtpLossEvent::SequenceGap {
                track_id,
                ssrc,
                expected,
                got,
            });
            for pkt in released {
                Self::push_one(st, loss_events, track_id, pkt, &mut out)?;
            }
        }

        if !st.cur_pkts.is_empty() {
            Self::drain_complete_or_discard(st, loss_events, track_id, &mut out);
            st.cur_ts = None;
        }
        Self::drain_pending(st, loss_events, track_id, &mut out);
        Ok(derive_dts(st, loss_events, track_id, out, true))
    }

    fn drain_pending(
        st: &mut TrackState,
        loss_events: &mut VecDeque<RtpLossEvent>,
        track_id: TrackId,
        out: &mut Vec<Sample>,
    ) {
        if st.reordered && st.kind == RtpMediaKind::H264 {
            Self::resolve_window(st, loss_events, track_id, out, true);
        } else if let Some(duration) = known_duration(st) {
            for (index, p) in st.pending.drain(..).enumerate() {
                let offset = u64::from(duration) * index as u64;
                let ts = to_ticks(p.unwrapped_ts.saturating_add(offset));
                out.push(Sample::new(
                    p.data,
                    Some(ts),
                    Some(ts),
                    Some(duration),
                    p.is_sync,
                ));
            }
        }
    }

    /// Resolve the oldest AUs of a reordered video track's window.
    ///
    /// RTP carries presentation time only, so under B-frame reorder the step to
    /// the next AU *in wire order* is not a frame duration (it can be zero or
    /// negative). The duration of an AU is its step to the next presentation
    /// instant, which is the smallest later timestamp seen; an AU
    /// with no later timestamp takes the frame period (declared, else the
    /// smallest positive step seen). With neither, the AU stays in the window
    /// rather than being given an invented duration.
    ///
    /// The head AU is resolved as soon as its presentation successor is exactly
    /// one frame period away (nothing can sit between them), so the hold is the
    /// AU's real reorder distance and a stream without reorder holds nothing. A
    /// successor that is further away may still be in flight: the head waits
    /// for it, at most `MAX_REORDER_DEPTH` AUs. With no frame period known yet
    /// the learned `au_window` decides. `drain_all` (flush) resolves every AU
    /// the window can resolve.
    fn resolve_window(
        st: &mut TrackState,
        loss_events: &mut VecDeque<RtpLossEvent>,
        track_id: TrackId,
        out: &mut Vec<Sample>,
        drain_all: bool,
    ) {
        while !st.pending.is_empty() {
            let head_ts = st.pending[0].unwrapped_ts;
            let successor = st
                .seen_pts
                .range(head_ts.saturating_add(1)..)
                .next()
                .map(|(&t, _)| t - head_ts);
            let period = st.declared_duration.map(u64::from).or_else(|| {
                st.min_positive_step
                    .and_then(|p| u64::try_from(p).ok())
                    .filter(|&p| p > 0)
            });
            let ready = drain_all
                || match period {
                    Some(p) => {
                        successor.is_some_and(|gap| gap <= p)
                            || st.pending.len() > MAX_REORDER_DEPTH
                    }
                    None => st.pending.len() > st.au_window,
                };
            if !ready {
                break;
            }
            let Some(duration) = successor
                .or(period)
                .and_then(|d| u32::try_from(d).ok())
                .filter(|&d| d > 0)
            else {
                break;
            };
            let p = st.pending.remove(0);
            let ts = to_ticks(p.unwrapped_ts);
            st.last_duration = duration;
            if let Some(oldest) = st.pending.iter().map(|p| p.unwrapped_ts).min() {
                while st
                    .seen_pts
                    .first_key_value()
                    .is_some_and(|(&t, _)| t < oldest)
                {
                    st.seen_pts.pop_first();
                }
            }
            out.push(Sample::new(
                p.data,
                Some(ts),
                Some(ts),
                Some(duration),
                p.is_sync,
            ));
        }
        if st.pending.len() > MAX_REORDER_WINDOW {
            let excess = st.pending.len() - MAX_REORDER_WINDOW;
            for p in st.pending.drain(..excess) {
                loss_events.push_back(RtpLossEvent::DroppedAccessUnit {
                    track_id,
                    pts: to_ticks(p.unwrapped_ts),
                });
            }
        }
    }

    /// Reassemble the buffered packets into AUs, then for each: unwrap its
    /// timestamp and emit the previously-pending AU with `duration` = the
    /// delta to this AU's timestamp.
    fn drain_complete(
        st: &mut TrackState,
        loss_events: &mut VecDeque<RtpLossEvent>,
        track_id: TrackId,
        out: &mut Vec<Sample>,
    ) -> Result<()> {
        let pkts = core::mem::take(&mut st.cur_pkts);
        st.cur_bytes = 0;
        let aus = match st.kind {
            RtpMediaKind::H264 => reassemble_video(&pkts)?,
            RtpMediaKind::Aac => reassemble_audio(&pkts)?,
        };
        for au in aus {
            let unwrapped = unwrap_ts(st.last_unwrapped, au.timestamp);
            // The wire timestamp is the *presentation* time (RFC 6184 §5.1 /
            // RFC 3640 §3.3.1), and RTP carries no second timestamp from which
            // a decode instant could be recovered. So a step backward means
            // the stream's presentation order is not its wire (decode) order —
            // a B-frame reorder — and `dts == pts` would be an unstated
            // assumption rather than a fact: report it (the samples themselves
            // are intact). See `RtpLossEvent::NonMonotonicTimestamp`.
            if let Some(prev) = st.last_unwrapped
                && unwrapped < prev
            {
                loss_events.push_back(RtpLossEvent::NonMonotonicTimestamp {
                    track_id,
                    previous_pts: to_ticks(prev),
                    pts: to_ticks(unwrapped),
                });
            }
            // Track the smallest positive step between consecutive wire
            // timestamps: the frame-period estimate used when a step goes
            // backward (see the duration fallback below).
            if let Some(prev) = st.last_unwrapped {
                let step = unwrapped as i64 - prev as i64;
                if step > 0 && st.min_positive_step.is_none_or(|p| step < p) {
                    st.min_positive_step = Some(step);
                }
            }
            st.last_unwrapped = Some(unwrapped);
            if st.first_unwrapped.is_none() {
                st.first_unwrapped = Some(unwrapped);
            }
            if st.kind == RtpMediaKind::H264
                && (st.reordered
                    || st
                        .pending
                        .last()
                        .is_some_and(|p| unwrapped < p.unwrapped_ts))
            {
                if !st.reordered {
                    let first = st.next_seq - st.pending.len() as u64;
                    for (i, p) in st.pending.iter().enumerate() {
                        st.seen_pts.insert(p.unwrapped_ts, first + i as u64);
                    }
                }
                st.reordered = true;
                if let Some((_, &pred)) = st.seen_pts.range(..unwrapped).next_back() {
                    let lag = usize::try_from(st.next_seq - pred).unwrap_or(usize::MAX);
                    st.au_window = st.au_window.max(lag.min(MAX_REORDER_DEPTH));
                }
                st.seen_pts.insert(unwrapped, st.next_seq);
                st.next_seq += 1;
                st.pending.push(PendingAu {
                    unwrapped_ts: unwrapped,
                    is_sync: au.is_sync,
                    data: au.data,
                });
                Self::resolve_window(st, loss_events, track_id, out, false);
                continue;
            }
            let same_packet_group = st.kind == RtpMediaKind::Aac
                && st
                    .pending
                    .last()
                    .is_some_and(|p| p.unwrapped_ts == unwrapped);
            if !same_packet_group && !st.pending.is_empty() {
                let prev_ts = st.pending[0].unwrapped_ts;
                // Duration = the *forward* step to this AU's presentation
                // time. On a stream whose presentation order is its wire order
                // that is exactly the frame period. On a reordered one it is
                // the distance to the next presentation instant, which is not
                // a per-frame duration at all — and no online rule can do
                // better, because RTP carries no decode time and the frame
                // period is only recoverable once a consecutive presentation
                // pair has been seen (which a reorder pattern may not show for
                // a whole GOP). This is reported through
                // [`RtpLossEvent::NonMonotonicTimestamp`]; a consumer that
                // needs exact per-frame durations should take the batch
                // [`crate::rtp::RtpDepacketiser`] path, which sees the whole
                // stream and can derive the period exactly.
                let forward = unwrapped.saturating_sub(prev_ts);
                let count = st.pending.len() as u64;
                // A step backward (presentation reorder) or a repeated
                // timestamp has no forward delta; fall back to the smallest
                // positive step seen, then to the declared frame timing. The
                // running minimum can be negative on a hostile series, so it is
                // converted with a checked conversion, not an `as` cast.
                let delta = if forward > 0 {
                    Some(forward)
                } else {
                    st.min_positive_step
                        .and_then(|p| u64::try_from(p).ok())
                        .filter(|&p| p > 0)
                        .or_else(|| st.declared_duration.map(|d| u64::from(d) * count))
                };
                let group: Vec<PendingAu> = st.pending.drain(..).collect();
                // Per-AU share of an aggregated group. When the group's delta
                // is smaller than its AU count the timestamps cannot resolve
                // the AUs (a share would be 0 ticks), so each AU takes the last
                // real per-AU duration or the declared one instead; with
                // neither the group has no truthful duration and is not
                // emitted.
                let share = delta.map(|d| d / count).filter(|&e| e > 0);
                match (delta, share) {
                    (Some(delta), Some(each)) => {
                        for (index, p) in group.into_iter().enumerate() {
                            let offset = each * index as u64;
                            let duration = if index as u64 + 1 == count {
                                delta - offset
                            } else {
                                each
                            };
                            let duration = u32::try_from(duration).unwrap_or(u32::MAX);
                            st.last_duration = duration;
                            let ts = to_ticks(p.unwrapped_ts.saturating_add(offset));
                            out.push(Sample::new(
                                p.data,
                                Some(ts),
                                Some(ts),
                                Some(duration),
                                p.is_sync,
                            ));
                        }
                    }
                    _ => {
                        let per_au = (st.last_duration != 0)
                            .then_some(st.last_duration)
                            .or(st.declared_duration);
                        if let Some(each) = per_au {
                            for (index, p) in group.into_iter().enumerate() {
                                let offset = u64::from(each) * index as u64;
                                let ts = to_ticks(p.unwrapped_ts.saturating_add(offset));
                                out.push(Sample::new(
                                    p.data,
                                    Some(ts),
                                    Some(ts),
                                    Some(each),
                                    p.is_sync,
                                ));
                            }
                        }
                    }
                }
            }
            st.pending.push(PendingAu {
                unwrapped_ts: unwrapped,
                is_sync: au.is_sync,
                data: au.data,
            });
            st.next_seq += 1;
        }
        Ok(())
    }
}

/// [`Stage`] adoption (media plane step 2e) — with a real restriction, stated
/// plainly rather than papered over: `Stage::feed` carries no track key, but
/// every [`push`](Self::push)/[`flush`](Self::flush) call on this type
/// requires an explicit `track_id`, because `RtpStreamDepacketiser` is
/// inherently multi-track (the whole point of
/// [`sync_start_decode_times`](Self::sync_start_decode_times) is correlating
/// *several* tracks' RTCP Sender Reports against one shared wallclock — that
/// needs their state alive together in one instance, not one instance per
/// track). So this impl only works for a depacketiser constructed with
/// **exactly one** track: `feed`/`finish` target that track directly, and
/// return [`Error::InvalidInput`] otherwise. Multi-track depacketisation
/// keeps driving [`push`](Self::push)/[`flush`](Self::flush) with an explicit
/// `track_id` directly — that inherent API is unchanged and still the right
/// tool for that case.
impl Stage for RtpStreamDepacketiser {
    type In<'a> = &'a [u8];
    type Out = Sample;
    type Error = Error;

    fn feed(&mut self, input: &[u8], _now: Timestamp) -> Result<()> {
        let [(track_id, _)] = self.tracks.as_slice() else {
            return Err(Error::InvalidInput(
                "Stage::feed requires RtpStreamDepacketiser to be constructed with exactly one track",
            ));
        };
        let track_id = *track_id;
        let samples = self.push(track_id, input)?;
        self.stage_ready.extend(samples);
        Ok(())
    }

    fn poll(&mut self) -> Option<Sample> {
        self.stage_ready.pop_front()
    }

    fn finish(&mut self) -> Result<()> {
        let [(track_id, _)] = self.tracks.as_slice() else {
            return Err(Error::InvalidInput(
                "Stage::finish requires RtpStreamDepacketiser to be constructed with exactly one track",
            ));
        };
        let track_id = *track_id;
        let samples = self.flush(track_id)?;
        self.stage_ready.extend(samples);
        Ok(())
    }

    fn next_deadline(&self) -> Option<Timestamp> {
        // RTP packets only produce output in reaction to feed/finish; no
        // rate-scheduled or timeout work needs a deadline.
        None
    }

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

    /// Honest against the one real bound this type enforces:
    /// `MAX_AU_BUFFER_BYTES`, the in-progress access-unit buffer a runaway
    /// (never-terminated) AU would otherwise grow without limit.
    fn demand(&self) -> Demand {
        let saturated = self
            .tracks
            .first()
            .map(|(_, st)| st.cur_bytes >= MAX_AU_BUFFER_BYTES)
            .unwrap_or(false);
        if saturated {
            Demand::saturated()
        } else {
            Demand::default()
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::avc_config::{AVCConfigurationBox, AVCDecoderConfigurationRecord};

    fn dummy_avc() -> CodecConfig {
        CodecConfig::Avc {
            config: AVCConfigurationBox::new(AVCDecoderConfigurationRecord {
                configuration_version: 1,
                profile_indication: 0x42,
                profile_compatibility: 0,
                level_indication: 0x1E,
                length_size_minus_one: 3,
                sps: alloc::vec![],
                pps: alloc::vec![],
                chroma_format: None,
                bit_depth_luma_minus8: None,
                bit_depth_chroma_minus8: None,
                sps_ext: alloc::vec![],
            }),
            width: 0,
            height: 0,
        }
    }

    const SPS_25FPS_FIXED: &[u8] = &[
        0x67, 0x64, 0x00, 0x0D, 0xAC, 0xD9, 0x41, 0x41, 0xFB, 0x01, 0x10, 0x00, 0x00, 0x03, 0x00,
        0x10, 0x00, 0x00, 0x03, 0x03, 0x20, 0xF1, 0x42, 0x99, 0x60,
    ];
    const SPS_NO_TIMING: &[u8] = &[
        0x67, 0x64, 0x00, 0x0D, 0xAD, 0xC8, 0xBF, 0xFE, 0x03, 0xC1, 0x41, 0xF9,
    ];
    const ASC_LC_48K_STEREO_1024: [u8; 2] = [0x11, 0x90];
    const ASC_LC_48K_STEREO_960: [u8; 2] = [0x11, 0x94];

    fn avc_with_sps(sps: &[u8]) -> CodecConfig {
        let CodecConfig::Avc {
            mut config,
            width,
            height,
        } = dummy_avc()
        else {
            unreachable!()
        };
        config.config.sps = alloc::vec![crate::nalu_types::AvcSps(sps.to_vec())];
        CodecConfig::Avc {
            config,
            width,
            height,
        }
    }

    fn declare_reorder_depth(d: &mut RtpStreamDepacketiser, depth: usize) {
        for (_, st) in &mut d.tracks {
            st.dts = DtsDeriver::new(Some(depth));
        }
    }

    fn single_track(kind: RtpMediaKind, config: CodecConfig, clock: u32) -> RtpStreamDepacketiser {
        RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(1),
            kind,
            config,
            clock,
        )])
    }

    #[test]
    fn test_sps_fixtures_declare_what_the_tests_assume() {
        let t = crate::sps::decode_avc_vui_timing(SPS_25FPS_FIXED)
            .unwrap()
            .unwrap();
        assert!(t.num_units_in_tick == 1 && t.time_scale == 50);
        assert!(
            crate::sps::decode_avc_vui_timing(SPS_NO_TIMING)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn flush_after_first_h264_au_uses_sps_vui_frame_period() {
        let id = TrackId::new(1);
        let mut d = single_track(RtpMediaKind::H264, avc_with_sps(SPS_25FPS_FIXED), 90_000);
        assert!(
            d.push(id, &vpkt(1, 1000, true, &[0x65, 0xAA]))
                .unwrap()
                .is_empty()
        );
        let out = d.flush(id).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].duration, Some(90_000 / 25));
    }

    #[test]
    fn set_track_config_refreshes_the_declared_duration() {
        let id = TrackId::new(1);
        let mut d = single_track(RtpMediaKind::H264, avc_with_sps(SPS_NO_TIMING), 90_000);
        d.set_track_config(id, avc_with_sps(SPS_25FPS_FIXED));
        assert!(
            d.push(id, &vpkt(1, 1000, true, &[0x65, 0xAA]))
                .unwrap()
                .is_empty()
        );
        assert_eq!(d.flush(id).unwrap()[0].duration, Some(3600));
    }

    #[test]
    fn flush_without_any_known_duration_holds_instead_of_inventing_one() {
        let id = TrackId::new(1);
        let mut d = single_track(RtpMediaKind::H264, avc_with_sps(SPS_NO_TIMING), 90_000);
        assert!(
            d.push(id, &vpkt(1, 1000, true, &[0x65, 0xAA]))
                .unwrap()
                .is_empty()
        );
        assert!(d.flush(id).unwrap().is_empty());
        let out = d.push(id, &vpkt(2, 4000, true, &[0x41, 0xBB])).unwrap();
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].duration, Some(3000));
    }

    #[test]
    fn flush_after_first_aac_au_uses_asc_frame_length() {
        let id = TrackId::new(1);
        for (asc, expected) in [(ASC_LC_48K_STEREO_1024, 1024), (ASC_LC_48K_STEREO_960, 960)] {
            let cfg = crate::rtp_sdp::aac_config_from_asc_bytes(asc.to_vec()).unwrap();
            let mut d = single_track(RtpMediaKind::Aac, cfg, 48_000);
            assert!(
                d.push(id, &aac_packet(1, 5000, &[&[1, 2]]))
                    .unwrap()
                    .is_empty()
            );
            let out = d.flush(id).unwrap();
            assert_eq!(out.len(), 1);
            assert_eq!(out[0].duration, Some(expected));
        }
    }

    #[test]
    fn aggregated_group_with_delta_below_au_count_uses_declared_au_duration() {
        let id = TrackId::new(1);
        let cfg =
            crate::rtp_sdp::aac_config_from_asc_bytes(ASC_LC_48K_STEREO_1024.to_vec()).unwrap();
        let mut d = single_track(RtpMediaKind::Aac, cfg, 48_000);
        let aus: [&[u8]; 4] = [&[1], &[2], &[3], &[4]];
        assert!(d.push(id, &aac_packet(1, 10_000, &aus)).unwrap().is_empty());
        assert!(
            d.push(id, &aac_packet(2, 10_002, &[&[5]]))
                .unwrap()
                .is_empty()
        );
        let out = d.push(id, &aac_packet(3, 12_000, &[&[6]])).unwrap();
        let got: Vec<_> = out.iter().map(|s| (s.pts, s.duration)).collect();
        assert_eq!(
            got,
            alloc::vec![
                (Some(10_000), Some(1024)),
                (Some(11_024), Some(1024)),
                (Some(12_048), Some(1024)),
                (Some(13_072), Some(1024)),
            ]
        );
    }

    #[test]
    fn aggregated_group_with_delta_below_au_count_and_no_known_duration_is_not_emitted() {
        let id = TrackId::new(1);
        let mut d = single_track(RtpMediaKind::Aac, dummy_avc(), 48_000);
        let aus: [&[u8]; 4] = [&[1], &[2], &[3], &[4]];
        assert!(d.push(id, &aac_packet(1, 10_000, &aus)).unwrap().is_empty());
        assert!(
            d.push(id, &aac_packet(2, 10_002, &[&[5]]))
                .unwrap()
                .is_empty()
        );
        let out = d.push(id, &aac_packet(3, 12_000, &[&[6]])).unwrap();
        assert!(out.is_empty());
    }

    fn aac_packet(seq: u16, ts: u32, aus: &[&[u8]]) -> Vec<u8> {
        let mut p = alloc::vec![0x80u8, 96];
        p.extend_from_slice(&seq.to_be_bytes());
        p.extend_from_slice(&ts.to_be_bytes());
        p.extend_from_slice(&[0, 0, 0, 0]);
        let headers_bits = (aus.len() * 16) as u16;
        p.extend_from_slice(&headers_bits.to_be_bytes());
        for au in aus {
            p.extend_from_slice(&((au.len() as u16) << 3).to_be_bytes());
        }
        for au in aus {
            p.extend_from_slice(au);
        }
        p
    }

    #[test]
    fn aggregated_aac_packet_spreads_its_timestamp_over_its_aus() {
        let id = TrackId::new(2);
        let mut d = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            id,
            RtpMediaKind::Aac,
            dummy_avc(),
            48_000,
        )]);
        let aus: [&[u8]; 4] = [&[1, 1], &[2, 2], &[3, 3], &[4, 4]];
        assert!(d.push(id, &aac_packet(1, 10_000, &aus)).unwrap().is_empty());
        assert!(
            d.push(id, &aac_packet(2, 14_096, &[&[5, 5]]))
                .unwrap()
                .is_empty()
        );
        let out = d.push(id, &aac_packet(3, 15_120, &[&[6, 6]])).unwrap();
        let got: Vec<_> = out.iter().map(|s| (s.pts, s.duration)).collect();
        assert_eq!(
            got,
            alloc::vec![
                (Some(10_000), Some(1024)),
                (Some(11_024), Some(1024)),
                (Some(12_048), Some(1024)),
                (Some(13_072), Some(1024)),
            ]
        );
    }

    fn vpkt(seq: u16, ts: u32, marker: bool, nal: &[u8]) -> Vec<u8> {
        let mut p = alloc::vec![0x80u8, if marker { 0x80 | 96 } else { 96 }];
        p.extend_from_slice(&seq.to_be_bytes());
        p.extend_from_slice(&ts.to_be_bytes());
        p.extend_from_slice(&[0, 0, 0, 0]);
        p.extend_from_slice(nal);
        p
    }

    #[test]
    fn video_stream_recovers_durations_and_sync() {
        let mut d = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(1),
            RtpMediaKind::H264,
            dummy_avc(),
            90_000,
        )]);
        declare_reorder_depth(&mut d, 0);

        // AU0 @1000 (IDR), AU1 @4000 (non-IDR), AU2 @7000 (non-IDR). 3000-tick spacing.
        let idr = [0x65u8, 0xAA];
        let non = [0x41u8, 0xBB];
        // AU0: emits nothing yet (duration needs AU1).
        assert!(
            d.push(TrackId::new(1), &vpkt(1, 1000, true, &idr))
                .unwrap()
                .is_empty()
        );
        // AU1 arrives → AU0 emitted with duration 3000, is_sync=true.
        let s0 = d.push(TrackId::new(1), &vpkt(2, 4000, true, &non)).unwrap();
        assert_eq!(s0.len(), 1);
        assert_eq!(s0[0].duration, Some(3000));
        assert!(s0[0].flags.is_sync);
        assert_eq!(s0[0].composition_offset(), 0);
        // AU2 arrives → AU1 emitted, duration 3000, is_sync=false.
        let s1 = d.push(TrackId::new(1), &vpkt(3, 7000, true, &non)).unwrap();
        assert_eq!(s1.len(), 1);
        assert_eq!(s1[0].duration, Some(3000));
        assert!(!s1[0].flags.is_sync);
        // flush → AU2 emitted with the last-known duration (3000).
        let s2 = d.flush(TrackId::new(1)).unwrap();
        assert_eq!(s2.len(), 1);
        assert_eq!(s2[0].duration, Some(3000));
    }

    /// [`Stage::feed`]/[`Stage::poll`]/[`Stage::finish`] on a single-track
    /// depacketiser must reproduce exactly what
    /// [`RtpStreamDepacketiser::push`]/[`RtpStreamDepacketiser::flush`]
    /// already return — the trait impl is a delegation, not a second
    /// implementation that could drift from the first.
    #[test]
    fn stage_feed_poll_finish_matches_push_flush() {
        let idr = [0x65u8, 0xAA];
        let non = [0x41u8, 0xBB];
        let packets = [
            vpkt(1, 1000, true, &idr),
            vpkt(2, 4000, true, &non),
            vpkt(3, 7000, true, &non),
        ];

        // Oracle: drive the inherent push/flush API directly.
        let mut oracle = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(1),
            RtpMediaKind::H264,
            dummy_avc(),
            90_000,
        )]);
        let mut oracle_samples: Vec<Sample> = Vec::new();
        for pkt in &packets {
            oracle_samples.extend(oracle.push(TrackId::new(1), pkt).unwrap());
        }
        oracle_samples.extend(oracle.flush(TrackId::new(1)).unwrap());

        // Same input, driven through Stage instead.
        let mut staged = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(1),
            RtpMediaKind::H264,
            dummy_avc(),
            90_000,
        )]);
        let mut staged_samples: Vec<Sample> = Vec::new();
        for pkt in &packets {
            Stage::feed(&mut staged, pkt, Timestamp::ZERO).unwrap();
            while let Some(s) = Stage::poll(&mut staged) {
                staged_samples.push(s);
            }
        }
        Stage::finish(&mut staged).unwrap();
        while let Some(s) = Stage::poll(&mut staged) {
            staged_samples.push(s);
        }

        assert_eq!(staged_samples.len(), oracle_samples.len());
        for (s, o) in staged_samples.iter().zip(oracle_samples.iter()) {
            assert_eq!(s.dts, o.dts);
            assert_eq!(s.pts, o.pts);
            assert_eq!(s.duration, o.duration);
            assert_eq!(s.flags.is_sync, o.flags.is_sync);
            assert_eq!(s.data, o.data);
        }
    }

    /// `Stage::feed`/`Stage::finish` on a depacketiser constructed with more
    /// than one track must error cleanly (no track key to target), not
    /// silently pick one — see the `Stage` impl's own doc comment.
    #[test]
    fn stage_feed_errors_on_multi_track_construction() {
        let mut d = RtpStreamDepacketiser::new(alloc::vec![
            RtpStreamTrack::new(TrackId::new(1), RtpMediaKind::H264, dummy_avc(), 90_000),
            RtpStreamTrack::new(TrackId::new(2), RtpMediaKind::H264, dummy_avc(), 90_000),
        ]);
        let err = Stage::feed(&mut d, &vpkt(1, 1000, true, &[0x65]), Timestamp::ZERO).unwrap_err();
        assert!(matches!(err, Error::InvalidInput(_)));
        let err2 = Stage::finish(&mut d).unwrap_err();
        assert!(matches!(err2, Error::InvalidInput(_)));
    }

    /// Builds one FU-A (RFC 6184 §5.8) fragment payload: `fu_indicator` +
    /// `fu_header` (start bit only on the first fragment, end bit **never**
    /// set) + `extra_len` bytes of filler — the "dropped final fragment"
    /// scenario audit-ingest #4 flags.
    fn fu_a_fragment(start: bool, extra_len: usize) -> Vec<u8> {
        const NAL_TYPE_FU_A: u8 = 28;
        const FU_START: u8 = 0x80;
        const ORIG_TYPE_IDR: u8 = 5;
        let fu_header = if start {
            FU_START | ORIG_TYPE_IDR
        } else {
            ORIG_TYPE_IDR
        };
        let mut payload = alloc::vec![NAL_TYPE_FU_A, fu_header];
        payload.extend(core::iter::repeat_n(0xABu8, extra_len));
        payload
    }

    /// A never-terminating FU-A run (end bit never set, marker bit never
    /// set, same RTP timestamp throughout — exactly a dropped/corrupted
    /// final fragment or a hostile encoder) must not grow `cur_pkts`
    /// without bound: [`MAX_AU_BUFFER_BYTES`] must trip, dropping the
    /// partial AU, and the depacketiser must keep working normally
    /// afterward (resync proof) rather than being wedged or OOMing.
    #[test]
    fn runaway_fu_a_without_end_bit_is_bounded_not_unbounded() {
        let mut d = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(1),
            RtpMediaKind::H264,
            dummy_avc(),
            90_000,
        )]);
        declare_reorder_depth(&mut d, 0);

        // Each fragment carries ~2 KiB of filler; MAX_AU_BUFFER_BYTES (4 MiB)
        // must trip well before we'd reach an unreasonable iteration count —
        // bounding proves the cap, not exhausting real memory.
        const FRAGMENT_FILLER: usize = 2048;
        let mut hit_cap = false;
        let mut resume_seq: u16 = 0;
        for i in 0..4096u16 {
            match d.push(
                TrackId::new(1),
                &vpkt(i, 1000, false, &fu_a_fragment(i == 0, FRAGMENT_FILLER)),
            ) {
                Ok(samples) => assert!(
                    samples.is_empty(),
                    "a never-completing AU must not emit a sample"
                ),
                Err(e) => {
                    assert!(
                        matches!(e, crate::error::Error::BufferCapExceeded { .. }),
                        "unexpected error variant: {e:?}"
                    );
                    hit_cap = true;
                    // The sequence gate (issue #779) already accepted packet
                    // `i` as in-order before `push_one` hit the byte cap, so
                    // `i + 1` is the real next-expected sequence number —
                    // resuming from a low/reused one (as this test did
                    // pre-#779, back when sequence numbers were ignored)
                    // would now look like a very late duplicate and be
                    // discarded, not a fresh start.
                    resume_seq = i.wrapping_add(1);
                    break;
                }
            }
        }
        assert!(
            hit_cap,
            "expected MAX_AU_BUFFER_BYTES to trip well within {} fragments \
             (never grow unbounded)",
            4096
        );

        // Resync proof: normal AUs at fresh timestamps process exactly as if
        // nothing had gone wrong — the overflow reset internal state cleanly.
        let idr = [0x65u8, 0xAA];
        let non = [0x41u8, 0xBB];
        assert!(
            d.push(TrackId::new(1), &vpkt(resume_seq, 4000, true, &idr))
                .unwrap()
                .is_empty()
        );
        let s = d
            .push(
                TrackId::new(1),
                &vpkt(resume_seq.wrapping_add(1), 7000, true, &non),
            )
            .unwrap();
        assert_eq!(s.len(), 1);
        assert_eq!(s[0].duration, Some(3000));
        assert!(s[0].flags.is_sync);
        assert!(
            d.poll_loss_event().is_none(),
            "no loss event expected: the byte-cap overflow is its own \
             recorded corrective action, and the sequence numbers this \
             resync uses are contiguous with the ones already accepted"
        );
    }

    #[test]
    fn track_specs_use_clock_rate_as_timescale() {
        let d = RtpStreamDepacketiser::new(alloc::vec![RtpStreamTrack::new(
            TrackId::new(7),
            RtpMediaKind::H264,
            dummy_avc(),
            90_000,
        )]);
        let specs = d.track_specs();
        assert_eq!(specs.len(), 1);
        assert_eq!(specs[0].track_id, TrackId::new(7));
        assert_eq!(specs[0].timescale, 90_000);
    }
    fn max_hold_in_aus(
        presentation_order_of_wire: &[u32],
        depth: usize,
    ) -> (usize, Vec<(u64, u32)>) {
        let id = TrackId::new(1);
        let mut d = single_track(RtpMediaKind::H264, avc_with_sps(SPS_25FPS_FIXED), 90_000);
        declare_reorder_depth(&mut d, depth);
        let mut emitted = Vec::new();
        let mut max_hold = 0;
        for (n, &frame) in presentation_order_of_wire.iter().enumerate() {
            let nal: &[u8] = if n == 0 { &[0x65, 0xAA] } else { &[0x41, 0xBB] };
            let ts = 1000 + frame * 3600;
            for s in d.push(id, &vpkt(n as u16, ts, true, nal)).unwrap() {
                emitted.push((s.pts.unwrap() as u64, s.duration.unwrap()));
            }
            max_hold = max_hold.max(n + 1 - emitted.len());
        }
        (max_hold, emitted)
    }

    #[test]
    fn reorder_window_tracks_the_observed_depth_not_a_fixed_size() {
        // The bounds are the duration window plus, once at the start of the
        // run, the `depth` samples the DTS derivation holds for its anchor.
        let one_b: Vec<u32> = [0, 2, 1, 4, 3, 6, 5, 8, 7, 10, 9, 12, 11, 14, 13].to_vec();
        let (hold, emitted) = max_hold_in_aus(&one_b, 1);
        assert!(hold <= 3 + 1, "one B frame held {hold} AUs");
        assert!(emitted.iter().skip(3).all(|&(_, dur)| dur == 3600));

        let two_b: Vec<u32> = [
            0, 3, 1, 2, 6, 4, 5, 9, 7, 8, 12, 10, 11, 15, 13, 14, 18, 16, 17,
        ]
        .to_vec();
        let (hold, emitted) = max_hold_in_aus(&two_b, 2);
        assert!(hold <= 4 + 2, "two B frames held {hold} AUs");
        assert!(emitted.iter().skip(4).all(|&(_, dur)| dur == 3600));

        let none: Vec<u32> = (0..12).collect();
        assert_eq!(max_hold_in_aus(&none, 0).0, 1);
    }

    #[test]
    fn reorder_window_is_capped() {
        let mut wire = alloc::vec![0u32];
        for k in (1..40u32).rev() {
            wire.push(k);
        }
        let (hold, _) = max_hold_in_aus(&wire, 0);
        assert!(hold <= MAX_REORDER_DEPTH + 1, "held {hold}");
    }
}