The control loop stops believing its own bookkeeping (ABR overhaul Phase 2)

The four correctness seams from the 08-22 auto-bitrate review §2, chosen
options per the RFC (planning design/abr-stack-overhaul.md §3):

- §2.3: a failed bitrate-change encoder rebuild now snaps the client back
  (retarget_tx) — the control task acks BEFORE the apply, so the client's
  climb base, utilization and proven math tracked a rate the encoder
  never ran until some later event happened to correct them.
- §2.2: the ABR rebuild announces PipelineGap on success, like the
  mode-switch and topology rebuilds already do — a ~0.6 s host-local
  stall read as congestion killed slow start for the session (the 401 ms
  field case: minutes at ~15 Mbps on a clean link).
- §2.1: an accepted mode switch re-teaches the stream-shape cap —
  computed once from the Welcome mode, 1080p→4K kept a 1080p-sized climb
  ceiling and 4K→720p left an oversized one standing. A re-set
  set_stream_cap also rebinds the already-learned ceiling downward
  (set_ceiling deliberately never lowers); the FIRST set keeps the
  founding semantics, pinned by the existing stream-bound test.
- §2.4: the bitrate_ack slot becomes a queue drained in arrival order —
  latest-wins collapsed a full resolve ack + corrective short retarget
  landing in the same 750 ms window, and host-cap learning needs two
  CONSECUTIVE short acks.

punktfunk-core --features quic: 490 tests green natively, including the
new a_mode_switch_reteaches_the_stream_cap_both_ways.

crates/punktfunk-core/src/abr.rs

@@ -552,10 +552,27 @@ impl BitrateController {
     }
 
     /// Teach the controller what this session's mode and codec could plausibly use (see
-    /// [`stream_ceiling_kbps`]). Applied to LEARNED ceilings only, at the same funnel as the
+    /// [`stream_ceiling_kbps`]). Bounds future LEARNED ceilings at the same funnel as the
     /// operator's env cap.
+    ///
+    /// The FIRST set is the session's negotiated shape and keeps the founding semantics — a
+    /// negotiated start rate above it stands, the host resolved that number (pinned by
+    /// `the_stream_bound_clamps_a_learned_ceiling_only`). A RE-set is a mode switch, and
+    /// there a DROP in pixel rate also rebinds the already-standing ceiling: `set_ceiling`
+    /// clamps only at learn time and deliberately never lowers, so a 4K-learned ceiling
+    /// would otherwise stand over a 720p stream with only the reactive loss/decode signals
+    /// to bound the climb (review §2.1).
     pub(crate) fn set_stream_cap(&mut self, kbps: u32) {
+        let mode_switch = self.stream_cap_kbps.is_some();
         self.stream_cap_kbps = Some(kbps);
+        if mode_switch && self.enabled && self.ceiling_kbps > kbps {
+            tracing::info!(
+                ceiling_kbps = self.ceiling_kbps,
+                stream_cap_kbps = kbps,
+                "adaptive bitrate: ceiling rebound to the switched mode's stream shape"
+            );
+            self.ceiling_kbps = kbps;
+        }
     }
 
     /// Teach the controller this session's refresh rate, so the encode thresholds can be sized in
@@ -1551,6 +1568,48 @@ mod tests {
         );
     }
 
+    /// Review §2.1: the stream-shape cap was computed once from the Welcome mode and never
+    /// again — 1080p→4K kept a 1080p-sized climb ceiling, 4K→720p left an oversized one
+    /// standing. A mode switch now re-teaches the cap: an upswitch opens room for the probe's
+    /// measurement to authorize more, a downswitch rebinds the already-learned ceiling.
+    #[test]
+    fn a_mode_switch_reteaches_the_stream_cap_both_ways() {
+        // 1080p session, probe measured a fat link: ceiling bound at the 1080p shape.
+        let mut c = BitrateController::new(20_000);
+        c.set_stream_cap(100_000);
+        c.set_ceiling(657_000);
+        assert_eq!(c.ceiling_kbps, 100_000);
+
+        // Switch UP to 4K: the new shape allows more, and the probe's measurement (already
+        // taken this session) may re-authorize up to it.
+        c.on_mode_switch();
+        c.set_stream_cap(400_000);
+        assert_eq!(
+            c.ceiling_kbps, 100_000,
+            "an upswitch alone raises nothing — authority still needs a measurement"
+        );
+        c.set_ceiling(657_000);
+        assert_eq!(
+            c.ceiling_kbps, 400_000,
+            "the 4K shape no longer pins the session to the 1080p bound"
+        );
+
+        // Switch DOWN to 720p: the learned 4K ceiling must not stand over the small stream —
+        // `set_ceiling` never lowers, so the re-taught cap is what rebinds it.
+        c.on_mode_switch();
+        c.set_stream_cap(42_000);
+        assert_eq!(
+            c.ceiling_kbps, 42_000,
+            "a downswitch rebinds the already-learned ceiling"
+        );
+
+        // A disabled controller (explicit bitrate) is untouched by all of it.
+        let mut d = BitrateController::new(0);
+        d.set_stream_cap(100_000);
+        d.set_stream_cap(42_000);
+        assert_eq!(d.ceiling_kbps, 0);
+    }
+
     #[test]
     fn owd_rise_alone_is_a_congestion_signal() {
         let mut c = BitrateController::new(20_000);

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

@@ -86,6 +86,10 @@ pub(super) async fn run_pump(args: WorkerArgs) {
     let clock_rtt_ns = negotiated.clock_rtt_ns;
     let resolved_bitrate_kbps = negotiated.bitrate_kbps;
     let negotiated_codec = negotiated.codec;
+    // Session constants a mode switch does not change — the pump recomputes the stream-shape
+    // cap from them for the switched geometry (review §2.1).
+    let bit_depth = negotiated.bit_depth;
+    let chroma_format = negotiated.chroma_format;
     // What this session's mode + codec could plausibly use — the bound the ABR holds its
     // probe-measured link ceiling to. Computed here because this is where the Welcome-resolved
     // geometry lives; the data pump stays codec-agnostic.
@@ -164,9 +168,14 @@ pub(super) async fn run_pump(args: WorkerArgs) {
         }
     });
 
-    // Adaptive bitrate ack slot: the control task parks the latest BitrateChanged here; the
-    // pump's controller drains it on its report tick (`take()` — an ack is consumed once).
-    let bitrate_ack: Arc<Mutex<Option<u32>>> = Arc::new(Mutex::new(None));
+    // Adaptive bitrate ack queue: the control task pushes every BitrateChanged; the pump's
+    // controller drains them in arrival order on its report tick. A QUEUE, not a latest-wins
+    // slot (review §2.4): a full resolve ack plus a corrective short retarget in the same
+    // 750 ms window used to collapse to whichever arrived last, and host-cap learning needs
+    // two CONSECUTIVE short acks — losing one delayed or prevented the cap and could
+    // reintroduce the encoder-overdrive sawtooth.
+    let bitrate_ack: Arc<Mutex<std::collections::VecDeque<u32>>> =
+        Arc::new(Mutex::new(std::collections::VecDeque::new()));
     // Decode-recovery keyframe asks (the ABR recovery signal): the control task counts every
     // outbound `CtrlRequest::Keyframe` — the one choke point all emitters funnel through — and
     // the pump drains the count per report window.
@@ -279,6 +288,8 @@ pub(super) async fn run_pump(args: WorkerArgs) {
         bitrate_kbps,
         resolved_bitrate_kbps,
         negotiated_codec,
+        bit_depth,
+        chroma_format,
         stream_cap_kbps,
         refresh_hz,
         mode_slot: mode_slot_pump,

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

@@ -15,7 +15,7 @@ pub(super) struct ControlTask {
     pub(super) mode_slot: Arc<Mutex<Mode>>,
     pub(super) probe: Arc<Mutex<ProbeState>>,
     /// The latest host `BitrateChanged` ack, drained by the pump's ABR on its report tick.
-    pub(super) bitrate_ack: Arc<Mutex<Option<u32>>>,
+    pub(super) bitrate_ack: Arc<Mutex<std::collections::VecDeque<u32>>>,
     /// The live encoder-target mirror ([`NativeClient::current_bitrate_kbps`]): unlike the
     /// drain-once ack slot above, this one always holds the latest acked rate for stats HUDs.
     pub(super) live_bitrate: Arc<AtomicU32>,
@@ -200,7 +200,7 @@ impl ControlTask {
                         if ack.bitrate_kbps > 0 {
                             live_bitrate.store(ack.bitrate_kbps, Ordering::Relaxed);
                         }
-                        *bitrate_ack.lock().unwrap() = Some(ack.bitrate_kbps);
+                        bitrate_ack.lock().unwrap().push_back(ack.bitrate_kbps);
                     } else if let Ok(gap) = crate::quic::PipelineGap::decode(&msg) {
                         // The host rebuilt its capture ring + encoder in place and nothing flowed
                         // while it did. Park it for the pump, which discards the report window in

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

@@ -27,7 +27,10 @@ pub(super) struct DataPump {
     pub(super) mode_gen: Arc<AtomicU32>,
     pub(super) frames_dropped: Arc<std::sync::atomic::AtomicU64>,
     pub(super) fec_recovered: Arc<std::sync::atomic::AtomicU64>,
-    pub(super) bitrate_ack: Arc<Mutex<Option<u32>>>,
+    /// Host `BitrateChanged` acks since the last report tick, drained in arrival order — a
+    /// queue so a corrective short retarget can't be clobbered by a full resolve ack in the
+    /// same window (review §2.4; host-cap learning needs two CONSECUTIVE short acks).
+    pub(super) bitrate_ack: Arc<Mutex<std::collections::VecDeque<u32>>>,
     /// Outbound decode-recovery keyframe asks, counted by the control task at its send choke
     /// point; drained per report window as the ABR's recovery signal.
     pub(super) recovery_kf: Arc<AtomicU32>,
@@ -40,9 +43,15 @@ pub(super) struct DataPump {
     /// The rate the host actually configured (echoed in Welcome).
     pub(super) resolved_bitrate_kbps: u32,
     pub(super) negotiated_codec: u8,
+    /// The negotiated encode bit depth and chroma wire byte — session constants a mode switch
+    /// does NOT change, carried so the stream-shape cap can be recomputed for a new geometry
+    /// (review §2.1).
+    pub(super) bit_depth: u8,
+    pub(super) chroma_format: u8,
     /// What this session's mode + codec could plausibly use (see
     /// [`crate::abr::stream_ceiling_kbps`]) — the bound the probe-measured link ceiling is held
-    /// to. Computed where the negotiated geometry lives, so this module stays codec-agnostic.
+    /// to. Computed where the negotiated geometry lives; recomputed here on an accepted mode
+    /// switch (review §2.1).
     pub(super) stream_cap_kbps: u32,
     /// The negotiated refresh, which sets the frame budget the ABR sizes its host-encode
     /// thresholds against (see [`crate::abr::BitrateController::set_frame_budget`]).
@@ -74,6 +83,8 @@ impl DataPump {
             bitrate_kbps,
             resolved_bitrate_kbps,
             negotiated_codec,
+            bit_depth,
+            chroma_format,
             stream_cap_kbps,
             refresh_hz,
             mode_slot: pump_mode_slot,
@@ -550,12 +561,26 @@ impl DataPump {
                 if mg != seen_mode_gen {
                     seen_mode_gen = mg;
                     abr.on_mode_switch();
+                    let m = *pump_mode_slot.lock().unwrap();
                     // The frame budget is a property of the MODE: a switch that changes the
                     // refresh changes what one frame of encode time costs, and the encode
                     // thresholds are sized in those.
-                    abr.set_frame_budget(pump_mode_slot.lock().unwrap().refresh_hz);
+                    abr.set_frame_budget(m.refresh_hz);
+                    // So is the stream-shape cap (review §2.1): computed once from the
+                    // Welcome mode, 1080p→4K kept a 1080p-sized climb ceiling (under-running
+                    // quality on a fat link) and 4K→720p left an oversized cap standing with
+                    // only the reactive loss/decode signals to bound the climb.
+                    // `set_stream_cap` also rebinds an already-learned ceiling downward.
+                    abr.set_stream_cap(crate::abr::stream_ceiling_kbps(
+                        m.width,
+                        m.height,
+                        m.refresh_hz,
+                        negotiated_codec,
+                        bit_depth,
+                        chroma_format,
+                    ));
                 }
-                if let Some(acked) = bitrate_ack.lock().unwrap().take() {
+                for acked in bitrate_ack.lock().unwrap().drain(..) {
                     abr.on_ack(acked);
                 }
                 let owd_mean_us =
@@ -1028,7 +1053,7 @@ mod tests {
                     refresh_hz: 60,
                 })),
                 probe: Arc::new(Mutex::new(ProbeState::default())),
-                bitrate_ack: Arc::new(Mutex::new(None)),
+                bitrate_ack: Arc::new(Mutex::new(std::collections::VecDeque::new())),
                 live_bitrate: Arc::new(AtomicU32::new(0)),
                 recovery_kf: Arc::new(AtomicU32::new(0)),
                 pipeline_gap: pipeline_gap.clone(),
@@ -1064,12 +1089,14 @@ mod tests {
             mode_gen: Arc::new(AtomicU32::new(0)),
             frames_dropped: Arc::new(std::sync::atomic::AtomicU64::new(0)),
             fec_recovered: Arc::new(std::sync::atomic::AtomicU64::new(0)),
-            bitrate_ack: Arc::new(Mutex::new(None)),
+            bitrate_ack: Arc::new(Mutex::new(std::collections::VecDeque::new())),
             recovery_kf: Arc::new(AtomicU32::new(0)),
             pipeline_gap: pipeline_gap.clone(),
             bitrate_kbps: 20_000,
             resolved_bitrate_kbps: 20_000,
             negotiated_codec: crate::quic::CODEC_HEVC,
+            bit_depth: 8,
+            chroma_format: 0,
             stream_cap_kbps: 100_000,
             refresh_hz: 60,
             mode_slot: Arc::new(Mutex::new(crate::config::Mode {

crates/punktfunk-host/src/native/stream.rs

@@ -2798,6 +2798,11 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 // `interval` was built as 1/effective_hz, so the round-trip recovers the integer
                 // rate.
                 let hz = interval_hz(interval);
+                // Timed for the `PipelineGap` below: the rebuild stalls capture for ~0.6 s,
+                // and a client that isn't told discards its starved windows as congestion
+                // (the 401 ms field case: slow start killed, minutes at ~15 Mbps on a clean
+                // link — review §2.2). The mode-switch and topology rebuilds already announce.
+                let rebuild_t0 = std::time::Instant::now();
                 match crate::encode::open_video(
                     plan.codec,
                     frame.format,
@@ -2853,10 +2858,24 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                         behind_score = 0;
                         depth_frames = 0;
                         ahead_run = 0;
+                        // …and it must not feed the CLIENT's controller either: announce the
+                        // host-local gap so the starved window is discarded, exactly as a
+                        // mode-switch rebuild does (review §2.2 — this arm was the one rebuild
+                        // that never told the client).
+                        announce_pipeline_gap(
+                            &gap_tx,
+                            rebuild_t0.elapsed().as_millis().min(u32::MAX as u128) as u32,
+                        );
                     }
                     Err(e) => {
                         tracing::warn!(error = %format!("{e:#}"), to_kbps = new_kbps,
                             "bitrate-change encoder rebuild failed — keeping the current rate");
+                        // The control task acked the resolved rate BEFORE this apply — with
+                        // the rebuild failed, the client's controller now tracks a rate the
+                        // encoder never ran: its climb base, utilization and proven math all
+                        // drift from a phantom number (review §2.3). Snap it back, same
+                        // channel as the short-apply correction above.
+                        let _ = retarget_tx.send(bitrate_kbps);
                     }
                 }
             }