feat(core/client): all-intra streams drain the frame channel to the newest AU

The pre-decode FrameChannel is strictly FIFO because H.26x reference
chains forbid mid-stream drops — falling behind is only recoverable via
the coarse jump-to-live thresholds (depth ≥6 sustained, or 400 ms behind
the capture clock) at the cost of a keyframe round-trip.

PyroWave has no reference chains: every AU decodes independently, so a
slow consumer can skip straight to the newest queued AU with zero
recovery cost. The pump now flags the channel all-intra for PyroWave
sessions and pop() drains to the newest, which caps any standing
pre-decode queue at ~1 frame structurally — the 2026-07 field report's
780M client could otherwise ratchet a real multi-frame backlog (and with
it the OSD latency) between those coarse thresholds.

Skips are counted separately from losses (they were delivered, just
superseded) and surfaced at debug on the report tick — the OSD 'lost'
line keeps meaning wire loss.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/punktfunk-core/src/client/frame_channel.rs

@@ -248,7 +248,16 @@ pub(crate) struct EncodeLatAcc {
 /// reference-chain break the loss counters never saw). A transient burst fills it briefly and drains on
 /// its own, so a clump never costs a keyframe.
 ///
+/// **All-intra exception** ([`set_all_intra`], PyroWave): every AU is independently decodable, so
+/// the reference-chain reasoning above does not apply — a consumer that falls behind can skip
+/// straight to the newest queued AU with zero recovery cost (no keyframe round-trip, no corrupt
+/// dependents). [`pop`] then drains to the newest instead of returning the oldest, which caps any
+/// standing queue at ~1 frame structurally; the 2026-07 field report's 780M client otherwise
+/// ratcheted a genuine multi-frame backlog between the coarse jump-to-live thresholds.
+///
 /// [`clear`]: FrameChannel::clear
+/// [`set_all_intra`]: FrameChannel::set_all_intra
+/// [`pop`]: FrameChannel::pop
 pub(crate) struct FrameChannel {
     inner: Mutex<FrameQueue>,
     ready: Condvar,
@@ -259,6 +268,11 @@ struct FrameQueue {
     /// Set when the pump exits so a blocked [`FrameChannel::pop`] reports the stream ended
     /// ([`PunktfunkError::Closed`]) rather than a spurious timeout (the old mpsc did this on sender drop).
     closed: bool,
+    /// Every AU decodes independently (PyroWave): [`FrameChannel::pop`] drains to the newest.
+    all_intra: bool,
+    /// AUs skipped by the all-intra drain since the last [`FrameChannel::take_skipped`] — NOT
+    /// losses (the wire delivered them); the pump surfaces them at debug on its report tick.
+    skipped_total: u64,
 }
 
 /// Outcome of [`FrameChannel::pop`] — mirrors the old `recv_timeout` results so `next_frame`'s
@@ -275,11 +289,25 @@ impl FrameChannel {
             inner: Mutex::new(FrameQueue {
                 q: VecDeque::new(),
                 closed: false,
+                all_intra: false,
+                skipped_total: 0,
             }),
             ready: Condvar::new(),
         }
     }
 
+    /// Pump side, once at session start: mark the stream all-intra (every AU independently
+    /// decodable) — [`Self::pop`] then drains to the newest queued AU instead of strict FIFO.
+    pub(crate) fn set_all_intra(&self, all_intra: bool) {
+        self.inner.lock().unwrap().all_intra = all_intra;
+    }
+
+    /// Pump side: AUs skipped by the all-intra drain since the last call (reset on read).
+    pub(crate) fn take_skipped(&self) -> u64 {
+        let mut st = self.inner.lock().unwrap();
+        std::mem::take(&mut st.skipped_total)
+    }
+
     /// Pump side: append a completed AU and wake a blocked consumer. Enforces the memory backstop
     /// ([`FRAME_QUEUE_HARD_CAP`]) by dropping the oldest (see its doc — a jump-to-live keyframe is
     /// already in flight by the time this can bite).
@@ -312,12 +340,21 @@ impl FrameChannel {
         self.ready.notify_all();
     }
 
-    /// Consumer side: pop the oldest AU, waiting up to `timeout` for one to arrive.
+    /// Consumer side: pop the oldest AU, waiting up to `timeout` for one to arrive. On an
+    /// all-intra stream ([`Self::set_all_intra`]) a multi-deep queue drains to the NEWEST AU
+    /// instead — the skipped ones are already superseded and decode independently, so showing
+    /// them only adds latency.
     pub(crate) fn pop(&self, timeout: Duration) -> FramePop {
         let mut st = self.inner.lock().unwrap();
         if st.q.is_empty() && !st.closed {
             st = self.ready.wait_timeout(st, timeout).unwrap().0;
         }
+        if st.all_intra && st.q.len() > 1 {
+            st.skipped_total += (st.q.len() - 1) as u64;
+            let newest = st.q.pop_back().expect("len > 1");
+            st.q.clear();
+            return FramePop::Frame(newest);
+        }
         if let Some(f) = st.q.pop_front() {
             FramePop::Frame(f)
         } else if st.closed {
@@ -364,6 +401,25 @@ mod frame_channel_tests {
         assert_eq!(ch.depth(), 0);
     }
 
+    /// The all-intra exception: a multi-deep queue drains to the NEWEST AU (every frame
+    /// decodes independently — older queued ones are superseded), the skips are counted
+    /// separately from losses, and a single-deep queue behaves exactly like FIFO.
+    #[test]
+    fn all_intra_drains_to_newest_and_counts_skips() {
+        let ch = FrameChannel::new();
+        ch.set_all_intra(true);
+        for i in 1..=3 {
+            ch.push(frame(i));
+        }
+        assert_eq!(popped(&ch), Some(3));
+        assert_eq!(ch.depth(), 0);
+        assert_eq!(ch.take_skipped(), 2);
+        assert_eq!(ch.take_skipped(), 0); // reset on read
+        ch.push(frame(4));
+        assert_eq!(popped(&ch), Some(4)); // depth 1 = plain FIFO, no skip accounting
+        assert_eq!(ch.take_skipped(), 0);
+    }
+
     #[test]
     fn empty_pop_times_out_not_closed() {
         let ch = FrameChannel::new();

crates/punktfunk-core/src/client/pump/data.rs

@@ -87,6 +87,10 @@ impl DataPump {
         // above its floor, and the climb probe's VBV reasoning doesn't apply to hard
         // per-frame CBR — controller and capacity probe stay off (0 = permanently off).
         let rate_pinned = negotiated_codec == crate::quic::CODEC_PYROWAVE;
+        // All-intra streams have no reference chains: the frame channel drains to the newest
+        // AU instead of strict FIFO (see `FrameChannel::set_all_intra`), so a slow consumer
+        // caps its standing queue at ~1 frame with zero recovery cost.
+        frames.set_all_intra(negotiated_codec == crate::quic::CODEC_PYROWAVE);
         let mut abr = BitrateController::new(if bitrate_kbps == 0 && !rate_pinned {
             resolved_bitrate_kbps
         } else {
@@ -289,6 +293,13 @@ impl DataPump {
                     resync_wanted = false;
                     let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
                 }
+                // All-intra drain-to-newest skips are NOT losses (the wire delivered them) —
+                // surface them at debug so a slow consumer is visible without alarming the
+                // OSD loss counters.
+                let skipped = frames.take_skipped();
+                if skipped > 0 {
+                    tracing::debug!(skipped, "all-intra frame channel drained to newest");
+                }
                 let window_dropped = st.frames_dropped.wrapping_sub(last_dropped);
                 let loss_ppm = window_loss_ppm(
                     st.fec_recovered_shards.wrapping_sub(last_recovered),