fix(host/abr): stop pinning Automatic sessions, and tell the client when the rate moves

Two host-side halves of the same sweep.

**The cadence latch.** `cadence_degraded` — which makes the control task refuse
bitrate CLIMBS — was latched true for as long as the session was escalated
(adaptive capture depth or pipelined retrieve), independently of whether encode
was still missing deadlines. The client cannot tell that refusal apart from an
encoder's real ceiling: both arrive as a short `BitrateChanged`, and two
identical ones latch a cap. Escalation needs ~20 net behind-frames, which a
startup hitch supplies while the ABR is still in slow start at the 20 Mbps
default — so one transient pinned the whole session there, long after the
escalation had bought back the headroom it was for, and escaping cost +12.5 %
per 60 s. An escalated session is still judged strictly (ANY net behind-frame
keeps it flagged, where an unescalated one gets the full bucket), but being
escalated no longer flags it by itself: escalating exists so cadence CAN be
held, and once it is, refusing climbs refuses the thing that worked. The rule
moves into `encode_behind_cadence` so it is stateable and testable.

**The silent re-target.** `adopt_built_bitrate` publishes the rate a rebuilt
pipeline actually opened at — `build_pipeline` re-resolves an Automatic rate
whenever the source delivers a size the session did not negotiate, the
mirrored-panel case — and the encoder's own clamp can land below what the
control task already acked. Neither reached the client, whose controller keeps
its own copy of that number as its climb base. A 1080p client mirroring a 4K
panel therefore believed 20 Mbps while the host encoded 60, and its first climb
computed from the stale base asked for 40: a re-target DOWNWARD, paying an
encoder rebuild to get there. Both paths now push the applied rate to the
control task, which sends `BitrateChanged` — the existing 9-byte message, which
already means precisely this and which clients already handle arriving
unprompted. No wire-format change, no capability negotiation, old clients
unaffected.

2 host tests added.

crates/punktfunk-host/src/native.rs

@@ -1073,6 +1073,17 @@ async fn serve_session(
     // accepted ack as "the active mode is now X" and fixes itself; old clients just log it.
     let (reconfig_result_tx, reconfig_result_rx) =
         tokio::sync::mpsc::unbounded_channel::<Reconfigured>();
+    // Unsolicited bitrate re-target, data plane → control task (the `reconfig_result_tx` pattern
+    // again, for the same reason). A pipeline rebuild can RE-RESOLVE an Automatic rate — most
+    // visibly when the source delivers a different size than the session negotiated, e.g. a
+    // client that asked for 1080p mirroring a 4K panel — and that number is what everything
+    // downstream reasons about: the send pacer, the console, and the base a `SetBitrate` ack is
+    // measured against. The client's copy only ever moved on an ack, so it stayed on the
+    // negotiated rate while the host encoded at another one, and the ABR's first climb computed
+    // from that stale base asked for LESS than the host was already sending — a re-target
+    // downward, with the rebuild it costs. Tell the client instead; `BitrateChanged` already
+    // means exactly this and old clients already handle one arriving unprompted.
+    let (retarget_tx, retarget_rx) = tokio::sync::mpsc::unbounded_channel::<u32>();
     // Cursor-forward bridge (M2): the encode loop diffs each frame's cursor serial and hands
     // changed SHAPES here; the control task (the control stream's sole writer) sends them.
     // Same shape as `probe_result_tx`. Wired even when the channel wasn't negotiated — it
@@ -1133,6 +1144,7 @@ async fn serve_session(
         probe_tx,
         probe_result_rx,
         reconfig_result_rx,
+        retarget_rx,
         cursor_shape_rx,
         cursor_client_draws,
         clip_enabled,
@@ -1579,6 +1591,7 @@ async fn serve_session(
                         probe_rx,
                         probe_result_tx,
                         reconfig_result_tx,
+                        retarget_tx,
                         fec_target: fec_target_dp,
                         phase: phase_ctl,
                         conn: conn_stream,

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

@@ -40,6 +40,9 @@ pub(super) async fn run(
     probe_tx: std::sync::mpsc::Sender<ProbeRequest>,
     mut probe_result_rx: tokio::sync::mpsc::UnboundedReceiver<ProbeResult>,
     mut reconfig_result_rx: tokio::sync::mpsc::UnboundedReceiver<Reconfigured>,
+    // Host-initiated bitrate re-target (a rebuild re-resolved an Automatic rate): forwarded to
+    // the client as a `BitrateChanged` so its controller's climb base tracks the real encoder.
+    mut retarget_rx: tokio::sync::mpsc::UnboundedReceiver<u32>,
     mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver<punktfunk_core::quic::CursorShape>,
     cursor_client_draws: Arc<AtomicBool>,
     clip_enabled: Arc<AtomicBool>,
@@ -338,6 +341,27 @@ pub(super) async fn run(
                     None => clip_offer_closed = true,
                 }
             }
+            retarget = retarget_rx.recv() => {
+                // A pipeline rebuild re-resolved the Automatic rate (see `retarget_tx`). Same
+                // message the `SetBitrate` path answers with — the client's controller treats
+                // any `BitrateChanged` as authoritative for what the encoder now targets, which
+                // is exactly right here: it IS what the encoder now targets, we just weren't
+                // asked. PyroWave reaches this too, and should: its rate is pinned against
+                // mid-stream RETARGETS, but a mode switch legitimately re-resolves the pin
+                // (~1.6 bpp for the new pixel rate) and the client's live-rate display is
+                // otherwise stuck on the old one. Its controller is off, so nothing acts on it.
+                let Some(kbps) = retarget else { break }; // data plane gone
+                tracing::info!(
+                    kbps,
+                    "encoder re-targeted by a pipeline rebuild — telling the client"
+                );
+                if io::write_msg(&mut ctrl_send, &BitrateChanged { bitrate_kbps: kbps }.encode())
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+            }
             correction = reconfig_result_rx.recv() => {
                 // H2 rollback/correction ack: the data plane reports the mode ACTUALLY live
                 // after a rebuild that failed (stayed at the old mode) or that the backend

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

@@ -1214,6 +1214,9 @@ pub(super) struct SessionContext {
     /// `Reconfigured { accepted: true, mode: <actually live> }` when a rebuild failed (stayed at
     /// the old mode) or the backend honored a different refresh than requested.
     pub(super) reconfig_result_tx: tokio::sync::mpsc::UnboundedSender<Reconfigured>,
+    /// Host-initiated bitrate re-target → control task → the client's `BitrateChanged`. Fired
+    /// by [`adopt_built_bitrate`] when a rebuild lands on a rate the client wasn't told about.
+    pub(super) retarget_tx: tokio::sync::mpsc::UnboundedSender<u32>,
     /// Adaptive-FEC target the control task updates from the client's loss reports.
     pub(super) fec_target: Arc<AtomicU8>,
     /// The QUIC control connection (carries host→client 0xCE source-HDR metadata mid-stream).
@@ -1397,6 +1400,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         probe_rx,
         probe_result_tx,
         reconfig_result_tx,
+        retarget_tx,
         fec_target,
         conn,
         timing_conn,
@@ -1597,7 +1601,12 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
     ) = pipe;
     // The encoder may have opened at a re-resolved rate (a mirrored head delivering a size this
     // session never negotiated). Adopt it before anything downstream reads `bitrate_kbps`.
-    adopt_built_bitrate(&mut bitrate_kbps, built_bitrate, &live_bitrate);
+    adopt_built_bitrate(
+        &mut bitrate_kbps,
+        built_bitrate,
+        &live_bitrate,
+        &retarget_tx,
+    );
 
     // Capture is live — launch the requested title so it renders onto the streamed output and
     // grabs focus. Windows spawns the library id into the interactive user session; Linux spawns
@@ -2037,7 +2046,12 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                         // The new compositor may deliver a different size than the old one did (a
                         // Game→Desktop switch onto a mirrored 4K panel is exactly that), so adopt
                         // the rate the rebuilt encoder actually opened at.
-                        adopt_built_bitrate(&mut bitrate_kbps, new_bitrate, &live_bitrate);
+                        adopt_built_bitrate(
+                            &mut bitrate_kbps,
+                            new_bitrate,
+                            &live_bitrate,
+                            &retarget_tx,
+                        );
                         vd = new_vd;
                         compositor = sw.compositor;
                         next = std::time::Instant::now();
@@ -2177,7 +2191,12 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                     }
                 };
             if rebuilt {
-                adopt_built_bitrate(&mut bitrate_kbps, built_bitrate, &live_bitrate);
+                adopt_built_bitrate(
+                    &mut bitrate_kbps,
+                    built_bitrate,
+                    &live_bitrate,
+                    &retarget_tx,
+                );
                 cur_mode = new_mode;
                 next = std::time::Instant::now();
                 // H2/H3: the backend may have honored a different mode than requested — KWin caps
@@ -2306,6 +2325,11 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 );
                 if applied_kbps < new_kbps {
                     encoder_ceiling_kbps.store(applied_kbps, Ordering::Relaxed);
+                    // The control task already acked the client with its own resolve, which was
+                    // higher than what the encoder took. Correct it, or the controller climbs
+                    // from a rate the encoder never ran at until its NEXT request happens to be
+                    // pre-clamped by the ceiling we just stored.
+                    let _ = retarget_tx.send(applied_kbps);
                 }
                 if applied_kbps < bitrate_kbps {
                     // Down-step: the behind-cadence backlog was scored against the old,
@@ -2356,6 +2380,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                         enc = new_enc;
                         if applied_kbps < new_kbps {
                             encoder_ceiling_kbps.store(applied_kbps, Ordering::Relaxed);
+                            // As in the in-place arm: the ack the client already has promises
+                            // more than the fresh encoder accepted — correct it.
+                            let _ = retarget_tx.send(applied_kbps);
                         }
                         bitrate_kbps = applied_kbps;
                         live_bitrate.store(applied_kbps, Ordering::Relaxed);
@@ -2797,7 +2824,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 // A capture-loss rebuild can land on a different source than it lost (this loop
                 // re-detects the session every cycle, precisely so it can follow a switch), so the
                 // delivered size — and with it an Automatic rate — may have changed under us.
-                adopt_built_bitrate(&mut bitrate_kbps, new_bitrate, &live_bitrate);
+                adopt_built_bitrate(&mut bitrate_kbps, new_bitrate, &live_bitrate, &retarget_tx);
                 enc.request_keyframe(); // belt-and-suspenders; a fresh encoder opens on an IDR anyway
                 last_forced_idr = Some(std::time::Instant::now()); // anchor the IDR cooldown from the rebuild
                 next = std::time::Instant::now();
@@ -3397,11 +3424,22 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 };
                 let escalated = cur_depth > 1 || pipelined_active || deescalating;
                 // Export "encode can't hold cadence" for the control task's climb refusal.
-                // An escalated session stays flagged even with the bucket drained: its climb
-                // headroom is spent, and letting climbs resume would saw against the
+                // An escalated session is held to a stricter standard — ANY net behind-frame
+                // keeps it flagged, where an unescalated one is given the full bucket — because
+                // its climb headroom really is partly spent and a climb would saw against the
                 // escalation and starve the de-escalation clean run below.
+                //
+                // But being escalated cannot flag it BY ITSELF, which is what this used to do.
+                // The client can't tell a transient refusal from an encoder's real ceiling: two
+                // identical short acks latch a cap, so a session that escalated once — the
+                // bucket needs ~20 net misses, which a startup hitch supplies while the ABR is
+                // still in slow start at the 20 Mbps default — got pinned there, and stayed
+                // pinned long after the escalation had bought back the headroom it was for.
+                // Escalating exists precisely so cadence CAN be held; once it is (bucket
+                // drained, every frame on time), refusing climbs is refusing the thing that
+                // worked.
                 cadence_degraded.store(
-                    escalated || behind_score >= DEPTH_DEGRADE,
+                    encode_behind_cadence(escalated, behind_score, DEPTH_DEGRADE),
                     Ordering::Relaxed,
                 );
                 if deescalating {
@@ -4068,13 +4106,42 @@ impl PaceBudget {
     }
 }
 
+/// Does the encoder currently fail to hold the frame cadence? Exported to the control task, which
+/// refuses bitrate CLIMBS while it is true (descents always pass — they are the cure).
+///
+/// `escalated` = the session has already spent an adaptive-depth / pipelined-retrieve step to buy
+/// headroom; `behind_score` is the leaky bucket of frames whose work overran the cadence deadline.
+/// An escalated session is judged strictly — ANY net behind-frame keeps it flagged — but being
+/// escalated does not flag it on its own. That distinction is the whole point: the client cannot
+/// tell a transient refusal from an encoder's hard ceiling (two identical short acks latch a cap),
+/// so "escalated ⇒ degraded, permanently" pinned Automatic sessions at whatever rate they happened
+/// to hold when a startup hitch escalated them — routinely the 20 Mbps default, while slow start
+/// had barely begun. Escalation exists so cadence CAN be held; once it is, refusing climbs refuses
+/// the thing that worked.
+fn encode_behind_cadence(escalated: bool, behind_score: u32, degrade_at: u32) -> bool {
+    behind_score >= degrade_at || (escalated && behind_score > 0)
+}
+
 /// Adopt the rate a freshly built pipeline's encoder was actually opened at.
 ///
 /// The session's own `bitrate_kbps` is the number every later decision reads — the ABR controller's
 /// climb base, the console's sample, what a `SetBitrate` ack is measured against — so letting it
 /// disagree with the live encoder means each of those reasons about a stream that doesn't exist.
 /// Silent when nothing changed, which is the overwhelmingly common case.
-fn adopt_built_bitrate(current: &mut u32, built: u32, live: &Arc<AtomicU32>) {
+///
+/// The client keeps its OWN copy of that number, and it used to move only on an ack — so a
+/// rebuild that re-resolved an Automatic rate (`build_pipeline` does, whenever the source
+/// delivers a size the session did not negotiate) left the two disagreeing for the rest of the
+/// session. The ABR's next climb then computed from the stale base and asked for a rate BELOW
+/// what the host was already sending: a re-target downward, paying an encoder rebuild to get
+/// there. So tell the client too — `BitrateChanged` is the same message the `SetBitrate` path
+/// answers with, and means the same thing arriving unprompted.
+fn adopt_built_bitrate(
+    current: &mut u32,
+    built: u32,
+    live: &Arc<AtomicU32>,
+    retarget: &tokio::sync::mpsc::UnboundedSender<u32>,
+) {
     if built == *current {
         return;
     }
@@ -4085,6 +4152,7 @@ fn adopt_built_bitrate(current: &mut u32, built: u32, live: &Arc<AtomicU32>) {
     );
     *current = built;
     live.store(built, Ordering::Relaxed);
+    let _ = retarget.send(built); // control task gone ⇒ the session is ending anyway
 }
 
 /// Encode-stall recovery: rebuild the encoder in place (keeping capture + the session up) and
@@ -4329,6 +4397,38 @@ fn build_pipeline(
 mod tests {
     use super::*;
 
+    #[test]
+    fn an_escalated_but_caught_up_encoder_stops_refusing_climbs() {
+        const DEGRADE: u32 = 10;
+        // Not escalated: the full bucket is allowed before climbs are refused.
+        assert!(!encode_behind_cadence(false, 0, DEGRADE));
+        assert!(!encode_behind_cadence(false, 9, DEGRADE));
+        assert!(encode_behind_cadence(false, 10, DEGRADE));
+        // Escalated and still missing deadlines: strict — one net behind-frame is enough.
+        assert!(encode_behind_cadence(true, 1, DEGRADE));
+        // Escalated, bucket fully drained: cadence is being HELD, which is what escalating was
+        // for. This is the case that used to stay latched for the rest of the session and pin an
+        // Automatic client at its slow-start rate.
+        assert!(!encode_behind_cadence(true, 0, DEGRADE));
+    }
+
+    #[test]
+    fn adopting_a_rebuilt_rate_tells_the_client() {
+        let live = Arc::new(AtomicU32::new(20_000));
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<u32>();
+        let mut current = 20_000;
+        // The overwhelmingly common case: the rebuild landed on the same rate — silent.
+        adopt_built_bitrate(&mut current, 20_000, &live, &tx);
+        assert_eq!(rx.try_recv().ok(), None);
+        // A re-resolve (the client asked 1080p, the source delivers a mirrored 4K panel): the
+        // host's rate moves, so the client has to hear about it — its controller's climb base is
+        // its own copy of this number, and a stale one makes the next "climb" a cut.
+        adopt_built_bitrate(&mut current, 60_000, &live, &tx);
+        assert_eq!(current, 60_000);
+        assert_eq!(live.load(Ordering::Relaxed), 60_000);
+        assert_eq!(rx.try_recv().ok(), Some(60_000));
+    }
+
     #[test]
     fn pacing_never_exceeds_the_session_rate_or_the_display() {
         // Backend honored the request exactly (the multiplier off): pace at it.