feat(host/encode): de-escalate the latency escalation once cadence holds clean

Escalate-and-hold's missing half. The contention escalation (capture depth,
then the NVENC pipelined retrieve) was permanent: one sustained overrun — even
one CAUSED by the ABR overdrive's rebuild storms — cost the session its
depth-1 latency and its sub-frame streaming forever, and `encode_us` reported
queue depth instead of ASIC time for the rest of the session.

- The leaky bucket now keeps scoring after escalation. A sustained
  every-frame-on-cadence run (~5 s at 120 fps) winds back one stage in reverse
  order: pipelined retrieve first (its rebuild restores the IO-stream binding
  and sub-frame chunked streaming), then capture depth back to 1. Attempts are
  paced by an exponential backoff (1 → 5 → 25 min, capped) — a workload that
  truly needs the escalation converges to keeping it, but never a permanent
  latch.

- NVENC (Linux) implements `set_pipelined(false)`: a `want_sync` latch handled
  at the same drained safe point as the engage side (`maybe_disengage_async`
  mirrors `maybe_engage_async`); the lazy sync re-init re-arms everything and
  opens on an IDR. The stream loop polls until the switch lands, then re-runs
  the escalation warmup so the wind-back's own stall can't re-escalate it.
  `PUNKTFUNK_NVENC_ASYNC=1` (operator-pinned async) refuses the wind-back;
  the trait doc now specifies the two-way contract.

- While escalated, `cadence_degraded` stays latched (bitrate climbs refused)
  even with the bucket drained: the headroom is spent, and climbs resuming
  mid-escalation would saw against it and starve the clean run.

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

crates/pf-encode/src/enc/codec.rs

@@ -352,8 +352,15 @@ pub trait Encoder: Send {
     /// encoder analog of the capturer depth escalation: AUs ride ~one loop tick behind (`poll`
     /// may return `None` while an encode is in flight) in exchange for capture/submit no longer
     /// serializing on the encode wait. Returns whether pipelined retrieve is (now) active; the
-    /// switch may be deferred to the next safe point internally. `false` from the default impl =
+    /// switch may be deferred to the next safe point internally. `set_pipelined(true)` returning
-    /// unsupported — the session loop stops asking. De-escalation is a v2 item everywhere.
+    /// `false` (the default impl) = unsupported — the session loop stops asking.
+    ///
+    /// `set_pipelined(false)` requests the wind-back (de-escalation, latency recovery): the
+    /// backend restores its sync-retrieve mode — and the latency features that mode carries
+    /// (IO-stream binding, sub-frame chunking) — at its next safe point, usually via a session
+    /// rebuild whose first frame is an IDR. The return is still "is pipelined retrieve active":
+    /// the caller polls until it reads `false`. Backends that never escalate return `false`
+    /// trivially. An operator pin (`PUNKTFUNK_NVENC_ASYNC=1`) refuses the wind-back.
     fn set_pipelined(&mut self, _on: bool) -> bool {
         false
     }

crates/pf-encode/src/enc/linux/nvenc_cuda.rs

@@ -596,6 +596,12 @@ pub struct NvencCudaEncoder {
     /// (escalate-and-hold, like the depth escalation); the switch itself happens at the next
     /// safe point via [`maybe_engage_async`](Self::maybe_engage_async).
     want_async: bool,
+    /// A de-escalation request ([`Encoder::set_pipelined(false)`]) waiting for its safe point:
+    /// the next drained moment tears the session down and lazily re-inits SYNC (IO-stream
+    /// binding and sub-frame chunking re-arm at that re-init). Distinct from `!want_async` —
+    /// an operator-forced async session (`PUNKTFUNK_NVENC_ASYNC=1`) also has `want_async`
+    /// false, and a de-escalation must never tear THAT down.
+    want_sync: bool,
     /// Boxed `CUstream` the session's IO-stream binding points at (`NvEncSetIOCudaStreams` takes
     /// POINTERS to `CUstream`, and this struct moves — the pointee needs a stable heap address for
     /// the session's lifetime). Null when stream-ordering is off; freed in `teardown` AFTER the
@@ -703,6 +709,7 @@ impl NvencCudaEncoder {
             last_rfi_range: None,
             async_rt: None,
             want_async: false,
+            want_sync: false,
             io_stream: ptr::null_mut(),
             stream_ordered: false,
             slices: 1,
@@ -736,6 +743,29 @@ impl NvencCudaEncoder {
         }
     }
 
+    /// [`maybe_engage_async`](Self::maybe_engage_async)'s inverse — wind the escalated pipelined
+    /// retrieve back at a safe point: nothing in flight, then a clean session rebuild whose lazy
+    /// SYNC re-init restores the IO-stream binding and re-arms sub-frame chunking (the two
+    /// latency features the escalation traded away). No-op until
+    /// [`want_sync`](Self::want_sync) is set and `pending` drains.
+    fn maybe_disengage_async(&mut self) {
+        if !self.want_sync || self.async_rt.is_none() || !self.pending.is_empty() {
+            return;
+        }
+        self.want_sync = false;
+        if self.inited {
+            // SAFETY: encode thread, `pending` empty ⇒ no encode in flight (and nothing queued
+            // to the retrieve thread); `teardown` joins the retrieve thread and handles exactly
+            // this live-session state — the next submit lazily re-inits sync.
+            unsafe { self.teardown() };
+            tracing::info!(
+                "NVENC pipelined-retrieve de-escalation: rebuilding the session with the sync \
+                 retrieve (IO-stream binding and sub-frame chunking restored); next frame opens \
+                 with an IDR"
+            );
+        }
+    }
+
     /// Tear down the encode session + pooled resources. Reused on a size change and at Drop.
     unsafe fn teardown(&mut self) {
         if self.encoder.is_null() {
@@ -1471,9 +1501,10 @@ impl Encoder for NvencCudaEncoder {
                 "Linux direct-NVENC needs a CUDA frame (FramePayload::Cuda); got a CPU/dmabuf frame"
             ),
         };
-        // A pending pipelined-retrieve escalation engages here, at the submit-side safe point
+        // A pending pipelined-retrieve escalation — or de-escalation — engages here, at the
-        // (nothing in flight after the previous poll drained).
+        // submit-side safe point (nothing in flight after the previous poll drained).
         self.maybe_engage_async();
+        self.maybe_disengage_async();
         // Re-init on a size change (the capturer can return at a different resolution after a mode
         // switch). Format changes (NV12↔YUV444) likewise re-init.
         let new_fmt = buffer_format(buf);
@@ -1781,12 +1812,25 @@ impl Encoder for NvencCudaEncoder {
 
     fn set_pipelined(&mut self, on: bool) -> bool {
         if !on {
-            // v1 is escalate-and-hold (no de-escalation), mirroring the depth escalation.
+            // De-escalation (the v2 of escalate-and-hold): latch the wind-back intent; the
+            // switch itself happens at the next drained safe point
+            // ([`maybe_disengage_async`](Self::maybe_disengage_async)) — the caller polls
+            // this same method until it reports inactive.
+            if async_retrieve_env() == Some(true) {
+                // Operator pinned async on — de-escalation must not undo an explicit choice.
+                return self.want_async || self.async_rt.is_some();
+            }
+            if self.want_async || self.async_rt.is_some() {
+                self.want_async = false;
+                self.want_sync = true;
+                self.maybe_disengage_async();
+            }
             return self.want_async || self.async_rt.is_some();
         }
         if async_retrieve_env() == Some(false) {
             return false; // operator veto: PUNKTFUNK_NVENC_ASYNC=0 means NEVER
         }
+        self.want_sync = false; // latest intent wins — cancel a pending wind-back
         if !self.want_async && self.async_rt.is_none() {
             self.want_async = true;
             self.maybe_engage_async();

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

@@ -1457,6 +1457,22 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
     // network, is the bottleneck" evidence — enough to flag `cadence_degraded` (the control task
     // then refuses bitrate CLIMBS) well before the session pays a latency escalation for it.
     const DEPTH_DEGRADE: u32 = 10;
+    // De-escalation (the escalate-and-hold v1's missing half): a sustained clean run at the
+    // escalated setting (~5 s at 120 fps, every frame on cadence) earns ONE attempt at winding
+    // back — reverse order of the escalation, pipelined retrieve first (its rebuild restores
+    // sub-frame streaming and the IO-stream binding), then capture depth back to 1. Each
+    // attempt costs the wind-back rebuild's IDR, so attempts are paced by an exponential
+    // backoff (1 → 5 → 25 min, capped) — a workload that genuinely needs the escalation
+    // converges to keeping it, but NEVER a permanent latch: a latch plus the ABR sawtooth
+    // pinned sessions at the floor with the escalation stuck.
+    const DEESCALATE_CLEAN_FRAMES: u32 = 600;
+    const DEESCALATE_BACKOFF_START: std::time::Duration = std::time::Duration::from_secs(60);
+    const DEESCALATE_BACKOFF_MAX: std::time::Duration = std::time::Duration::from_secs(25 * 60);
+    let mut pipelined_active = false;
+    let mut deescalating = false;
+    let mut ahead_run: u32 = 0;
+    let mut deescalate_not_before: Option<std::time::Instant> = None;
+    let mut deescalate_backoff = DEESCALATE_BACKOFF_START;
     while !stop.load(Ordering::SeqCst) && std::time::Instant::now() < deadline {
         // Mid-stream session switch (the box flipped Gaming↔Desktop): rebuild the WHOLE backend in
         // place — a different compositor at the SAME client mode — keeping the Session + send thread
@@ -1794,6 +1810,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                         // escalation — clean slate + re-run the warmup before judging again.
                         behind_score = 0;
                         depth_frames = 0;
+                        ahead_run = 0;
                     }
                     Err(e) => {
                         tracing::warn!(error = %format!("{e:#}"), to_kbps = new_kbps,
@@ -2618,7 +2635,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         // two-thread lock moves the encode wait off this loop so capture/submit keep cadence,
         // at ~one tick of AU latency. `enc.set_pipelined` may decline (unsupported backend or
         // an explicit PUNKTFUNK_NVENC_ASYNC=0); either way it is asked exactly once.
-        if idd_adaptive_enabled() && (cur_depth < max_depth || !pipeline_asked) {
+        if idd_adaptive_enabled() {
             depth_frames += 1;
             if depth_frames > DEPTH_WARMUP_FRAMES {
                 let behind = std::time::Instant::now() >= next;
@@ -2627,34 +2644,89 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 } else {
                     behind_score.saturating_sub(1)
                 };
+                let escalated = cur_depth > 1 || pipelined_active || deescalating;
                 // Export "encode can't hold cadence" for the control task's climb refusal.
-                // Stored BEFORE the escalate check: the firing iteration writes `true`, and a
+                // An escalated session stays flagged even with the bucket drained: its climb
-                // final-stage escalation freezes this whole block (the guard above goes false)
+                // headroom is spent, and letting climbs resume would saw against the
-                // — so an escalated session stays flagged, which is exactly right: its climb
+                // escalation and starve the de-escalation clean run below.
-                // headroom is spent until something (a down-step, de-escalation) changes.
+                cadence_degraded.store(
-                cadence_degraded.store(behind_score >= DEPTH_DEGRADE, Ordering::Relaxed);
+                    escalated || behind_score >= DEPTH_DEGRADE,
-                if behind_score >= DEPTH_ESCALATE {
+                    Ordering::Relaxed,
+                );
+                if deescalating {
+                    // A requested wind-back completes at the encoder's drained safe point —
+                    // poll it (the call is a cheap latch check until then).
+                    if !enc.set_pipelined(false) {
+                        deescalating = false;
+                        pipelined_active = false;
+                        // Re-arm the ask: a future sustained overrun may escalate again (the
+                        // backoff below paces how soon another wind-back may follow it).
+                        pipeline_asked = false;
+                        tracing::info!(
+                            "encoder pipelined retrieve de-escalated — sync retrieve (and \
+                             sub-frame streaming, where armed) restored; re-monitoring cadence"
+                        );
+                        // The wind-back rebuild's own stall must not re-escalate on the spot.
+                        behind_score = 0;
+                        depth_frames = 0;
+                        ahead_run = 0;
+                    }
+                } else if behind_score >= DEPTH_ESCALATE
+                    && (cur_depth < max_depth || !pipeline_asked)
+                {
                     if cur_depth < max_depth {
                         cur_depth = max_depth;
                         tracing::info!(
                             depth = cur_depth,
                             "IDD pipeline depth escalated — encode can't hold cadence at depth-1 \
-                             (GPU contention); pipelining for the rest of the session (latency \
+                             (GPU contention); pipelining until cadence holds clean (latency \
                              trade for throughput)"
                         );
                     } else {
                         pipeline_asked = true;
-                        if enc.set_pipelined(true) {
+                        pipelined_active = enc.set_pipelined(true);
+                        if pipelined_active {
                             tracing::info!(
                                 "encoder pipelined retrieve escalated — encode can't hold \
                                  cadence and the capturer has no depth to give; the encode wait \
-                                 moves off the loop for the rest of the session (latency trade \
+                                 moves off the loop until cadence holds clean (latency trade \
                                  for throughput)"
                             );
                         }
                     }
                     // Give the action time to take effect before judging again.
                     behind_score = 0;
+                    ahead_run = 0;
+                } else if escalated {
+                    // De-escalation: a sustained every-frame-on-cadence run at the escalated
+                    // setting is the evidence the contention passed (a lower ABR rate, the
+                    // game scene lightened) — wind back in reverse order, paced by the
+                    // exponential backoff (see the consts above).
+                    ahead_run = if behind { 0 } else { ahead_run + 1 };
+                    if ahead_run >= DEESCALATE_CLEAN_FRAMES
+                        && deescalate_not_before.is_none_or(|t| std::time::Instant::now() >= t)
+                    {
+                        ahead_run = 0;
+                        deescalate_not_before =
+                            Some(std::time::Instant::now() + deescalate_backoff);
+                        deescalate_backoff = (deescalate_backoff * 5).min(DEESCALATE_BACKOFF_MAX);
+                        if pipelined_active {
+                            tracing::info!(
+                                "cadence held clean while escalated — winding the pipelined \
+                                 retrieve back (latency recovery; costs one IDR)"
+                            );
+                            deescalating = true;
+                        } else if cur_depth > 1 {
+                            cur_depth = 1;
+                            tracing::info!(
+                                depth = cur_depth,
+                                "IDD pipeline depth de-escalated — cadence held clean at the \
+                                 escalated depth (latency recovery)"
+                            );
+                            behind_score = 0;
+                            depth_frames = 0;
+                        }
+                    }
                 }
             }
         }