feat(pf-encode,host): price the HEVC sub-frame trade so arbitration can cover it

The named next step after WP3's first increment. That increment deliberately
REFUSED to arbitrate HEVC-with-sub-frame -- the fleet default, and the reported
field case -- because engaging split there gives up sub-frame readback, whose
whole value is that the send overlaps the encode. An encoder measuring only
encode time would see split as ~2x faster, take it, and make end-to-end latency
worse while reporting a win. This supplies the missing number.

The real comparison is encode_1eng + send_of_last_slice against
encode_2eng + send_of_whole_AU, so the challenger owes roughly
spread x (slices-1)/slices. Split across the two sides that can each see half:

- Host: new `Encoder::set_send_spread_us` (defaulted, forwarded by
  TrackedEncoder -- same trap class as set_wire_chunking, and unforwarded it
  would fail SILENTLY IN THE SAFE DIRECTION, which is the hardest kind to
  notice). The send thread is the only place a paced send is observed and the
  encode loop the only place the encoder can be touched, so it goes over an
  AtomicU32 like encoder_ceiling_kbps, EWMA-smoothed 3:1 per completed AU: one
  content spike must not flip a verdict that then gets cached.
- Encoder: turns the raw spread into the handicap, because only it knows
  `slices`. SplitArbiter::with_handicap charges it to the challenger before the
  comparison. A unit test runs identical encode numbers with a cheap and an
  expensive send and asserts the verdict REVERSES -- with an expensive send the
  arm that looks twice as fast is a loss end to end, and the incumbent must
  hold. That is precisely the regression an encode-only arbiter ships.

Gate now opens for HEVC+sub-frame only when a spread has actually been reported
(and slices >= 2); with no hint it still refuses, so behaviour is unchanged until
the host feeds it.

Two mechanics this needed:
- apply_split_mode became a PAIR flip (split + sub-frame), routed through
  resolve_split_subframe and restoring from `subframe_opened_with` so a session
  that never had sub-frame can never gain it. It also recomputes
  `subframe_chunks`, which reconfigure_bitrate does NOT -- spike S1c's finding;
  leave it stale and supports_chunked_poll keeps saying yes while numSlices never
  advances, so poll_chunk busy-polls its whole budget every AU.
- The arbiter is now fed from BOTH completion points. A sub-frame session
  finishes through poll_chunk, so the incumbent arm of an HEVC experiment would
  otherwise never deliver a sample -- only the challenger, with sub-frame
  dropped, comes through poll.

Verified .21: clippy -D warnings clean for pf-encode AND punktfunk-host with
nvenc, 63 unit tests (1 new), 23/23 NVENC on-hardware green. Verified .133:
Windows clippy -D warnings clean, zero dead_code. fmt clean.

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

@@ -443,6 +443,20 @@ pub trait Encoder: Send {
     /// flagged [`EncodedFrame::chunk_aligned`] and the session marks them on the wire.
     /// Default: no-op (the H.26x backends' bitstreams cannot be cut losslessly).
     fn set_wire_chunking(&mut self, _shard_payload: usize) {}
+    /// How long a whole AU's packets currently take to leave the socket (µs, smoothed) — the
+    /// host's paced-send `spread_us`.
+    ///
+    /// Exists for ONE decision, and only the host can supply it. The Linux direct-NVENC split
+    /// arbitration compares single-engine against split, but on HEVC engaging split costs
+    /// sub-frame readback, and sub-frame's whole value is that the send overlaps the encode. So
+    /// the real comparison is `encode_1eng + send_of_last_slice` against
+    /// `encode_2eng + send_of_whole_AU`, and an encoder that measures only encode time would
+    /// reliably pick split and make end-to-end latency WORSE. The backend turns this number into
+    /// that handicap (it knows its own slice count); the host just reports what it observes.
+    ///
+    /// Optional by design: a backend that ignores it simply never arbitrates the sub-frame trade,
+    /// which is the safe direction. `0` = unknown / not reported yet.
+    fn set_send_spread_us(&mut self, _us: u32) {}
     /// How many frames the CAPTURER guarantees the encoder may hold in flight before it starts
     /// reusing an input texture (`Capturer::pipeline_depth`). Backends that encode the capturer's
     /// textures IN PLACE — no `CopyResource` — must not pipeline deeper than this: the capturer

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

@@ -830,6 +830,15 @@ pub struct NvencCudaEncoder {
     encoder_engines: u32,
     /// Submit stamp for the split arbiter's per-frame cost (sync depth-1 path only).
     last_submit_at: Option<std::time::Instant>,
+    /// Whole-AU paced-send time (µs) the host last reported, via
+    /// [`Encoder::set_send_spread_us`]. `0` = never reported, which keeps the arbiter out of the
+    /// sub-frame trade entirely (it cannot price what it cannot see).
+    send_spread_us: u32,
+    /// Sub-frame state the session was OPENED able to run — what `resolve_subframe` decided from
+    /// the caps probe and the env. `subframe_on` moves as the arbiter flips arms; this does not,
+    /// so a return to a non-forced split can restore sub-frame without re-deriving it (and
+    /// without ever turning it on for a session that never had it).
+    subframe_opened_with: bool,
     /// The live split-mode experiment, when one is running. `None` = not arbitrating (gated off,
     /// already decided this process, or the config is one we refuse to arbitrate).
     arbiter: Option<SplitArbiter>,
@@ -923,6 +932,8 @@ impl NvencCudaEncoder {
             subframe_chunks: false,
             encoder_engines: 0,
             last_submit_at: None,
+            send_spread_us: 0,
+            subframe_opened_with: false,
             arbiter: None,
             chunk: None,
         })
@@ -1383,6 +1394,7 @@ impl NvencCudaEncoder {
                 self.subframe_forced,
             );
             self.subframe_on = subframe_on;
+            self.subframe_opened_with = subframe_on;
             const CLAMP_TOL_BPS: u64 = 20_000_000;
 
             // Ceiling cache (process lifetime, `nvenc_core`): a prior clamp search already found
@@ -1724,13 +1736,25 @@ impl NvencCudaEncoder {
         {
             return;
         }
-        if self.subframe_on && self.codec != Codec::Av1 {
-            tracing::debug!(
-                "NVENC split arbitration skipped: sub-frame readback is on and this codec cannot \
-                 keep it while split, so the trade costs send overlap the encoder cannot measure"
-            );
-            return;
-        }
+        // Losing sub-frame costs the send/encode overlap: without it the AU's last byte waits for
+        // the WHOLE send instead of just the final slice, so the challenger owes roughly
+        // `spread × (slices−1)/slices`. Priced here because only the encoder knows `slices`; the
+        // host reports the raw spread.
+        let handicap_us = if self.subframe_on && self.codec != Codec::Av1 {
+            if self.send_spread_us == 0 || self.slices < 2 {
+                tracing::debug!(
+                    "NVENC split arbitration skipped: engaging split would cost sub-frame readback \
+                     and no send-spread has been reported, so the trade cannot be priced — an \
+                     encode-only comparison would take the arm that looks fastest and lose \
+                     end-to-end"
+                );
+                return;
+            }
+            let slices = self.slices as u64;
+            self.send_spread_us as u64 * (slices - 1) / slices
+        } else {
+            0
+        };
         // Pick the challenger that tests the question worth asking: "are we leaving engines idle?"
         // So anything that is not already the widest forced split is challenged BY the widest, and
         // only a session already there is challenged by single-engine ("is splitting even helping
@@ -1752,9 +1776,15 @@ impl NvencCudaEncoder {
         tracing::info!(
             incumbent = self.split_mode,
             challenger,
+            handicap_us,
+            send_spread_us = self.send_spread_us,
             "NVENC split arbitration armed — measuring both arms on the live session (no IDR)"
         );
-        self.arbiter = Some(SplitArbiter::new(self.split_mode, challenger));
+        self.arbiter = Some(SplitArbiter::with_handicap(
+            self.split_mode,
+            challenger,
+            handicap_us,
+        ));
     }
 
     /// The config identity this session's split verdict is cached under.
@@ -1776,17 +1806,34 @@ impl NvencCudaEncoder {
     /// moves. Returns whether the driver accepted it; on refusal the field is restored so the
     /// encoder's idea of its own session stays truthful.
     fn apply_split_mode(&mut self, mode: u32) -> bool {
-        let previous = self.split_mode;
+        let (prev_mode, prev_sub, prev_chunks) =
+            (self.split_mode, self.subframe_on, self.subframe_chunks);
+        // Sub-frame rides along: HEVC cannot hold both, so a forced split must drop it and a
+        // return to non-forced may take it back (only up to what the session was opened able to
+        // do — `subframe_cap`/`resolve_subframe` decided that once, at open).
+        let (mode, subframe) = resolve_split_subframe(
+            self.codec,
+            mode,
+            self.subframe_opened_with,
+            self.subframe_forced,
+        );
         self.split_mode = mode;
+        self.subframe_on = subframe;
+        // ⚠ The latch `reconfigure_bitrate` does NOT recompute (spike S1c): leave it stale and
+        // `supports_chunked_poll` keeps saying yes while `numSlices` never advances, so
+        // `poll_chunk` busy-polls its entire budget every AU.
+        self.subframe_chunks = self.slices >= 2 && subframe && self.async_rt.is_none();
         if self.reconfigure_bitrate(self.bitrate_bps) {
             true
         } else {
             tracing::warn!(
-                from = previous,
+                from = prev_mode,
                 to = mode,
                 "NVENC split arbitration: driver refused the in-place split change — staying put"
             );
-            self.split_mode = previous;
+            self.split_mode = prev_mode;
+            self.subframe_on = prev_sub;
+            self.subframe_chunks = prev_chunks;
             false
         }
     }
@@ -2551,6 +2598,17 @@ impl Encoder for NvencCudaEncoder {
                     "NVENC chunked poll: picture type diverged from the submit-time prediction"
                 );
             }
+            // The AU is complete here too — the chunked path is how a sub-frame session finishes,
+            // so the arbiter has to be fed from BOTH completion points or it would never see a
+            // frame on the incumbent arm of an HEVC sub-frame experiment (that arm is chunked;
+            // only the challenger, with sub-frame dropped, comes through `poll`).
+            let encode_us = self
+                .last_submit_at
+                .take()
+                .map(|t| t.elapsed().as_micros() as u64);
+            if let Some(us) = encode_us {
+                self.feed_split_arbiter(us);
+            }
             Ok(Some(AuChunk {
                 data,
                 pts_ns,
@@ -2634,6 +2692,10 @@ impl Encoder for NvencCudaEncoder {
         }
     }
 
+    fn set_send_spread_us(&mut self, us: u32) {
+        self.send_spread_us = us;
+    }
+
     fn applied_bitrate_bps(&self) -> Option<u64> {
         // `bitrate_bps` is the post-clamp truth: the open path's ceiling search and the
         // reconfigure path's cache clamp both write what the session ACTUALLY targets.

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

@@ -402,6 +402,13 @@ pub(super) struct SplitArbiter {
     samples: Vec<u64>,
     incumbent_us: u64,
     settle_left: u32,
+    /// Latency the challenger COSTS beyond its encode time, added to its measured result before
+    /// the comparison. Non-zero only when winning the split means giving up sub-frame readback:
+    /// sub-frame lets the send overlap the encode, so losing it pushes the AU's last byte out by
+    /// roughly `send_spread × (slices−1)/slices`. Without this term the arbiter compares encode
+    /// against encode, always prefers split on HEVC, and makes end-to-end latency worse while
+    /// reporting a win.
+    challenger_handicap_us: u64,
 }
 
 /// Frames discarded after a switch before the challenger is judged (measured — see the struct doc).
@@ -419,7 +426,9 @@ const WIN_MARGIN_PCT: u64 = 10;
 
 #[cfg(target_os = "linux")]
 impl SplitArbiter {
-    pub(super) fn new(incumbent: u32, challenger: u32) -> Self {
+    /// `handicap_us` is what the challenger costs OUTSIDE the encode it is measured on — pass `0`
+    /// when it gives up nothing. See [`Self::challenger_handicap_us`].
+    pub(super) fn with_handicap(incumbent: u32, challenger: u32, handicap_us: u64) -> Self {
         Self {
             state: ArbState::MeasuringIncumbent,
             incumbent,
@@ -427,6 +436,7 @@ impl SplitArbiter {
             samples: Vec::with_capacity(SAMPLE_FRAMES),
             incumbent_us: 0,
             settle_left: 0,
+            challenger_handicap_us: handicap_us,
         }
     }
 
@@ -457,7 +467,9 @@ impl SplitArbiter {
                 if self.samples.len() < SAMPLE_FRAMES {
                     return None;
                 }
-                let challenger_us = median(&mut self.samples);
+                // Compare TOTAL cost, not encode cost: whatever the challenger gives up outside
+                // the encode (on HEVC, the sub-frame send overlap) is charged to it here.
+                let challenger_us = median(&mut self.samples) + self.challenger_handicap_us;
                 self.state = ArbState::Done;
                 // Strictly better by the margin, or the incumbent keeps the session. Equal-ish is
                 // deliberately a win for the incumbent: we are already there.
@@ -1241,15 +1253,57 @@ pub(super) unsafe fn apply_low_latency_config(cfg: &mut nv::NV_ENC_CONFIG, c: Lo
 #[cfg(all(test, target_os = "linux"))]
 mod arbiter_tests {
     use super::{ArbAction, SplitArbiter, SETTLE_FRAMES};
+
     use nvidia_video_codec_sdk::sys::nvEncodeAPI::NV_ENC_SPLIT_ENCODE_MODE as M;
 
+    /// THE SUB-FRAME TRADE, which is the whole reason `set_send_spread_us` exists. Same encode
+    /// numbers both times; only the handicap differs.
+    ///
+    /// A 4K HEVC session where split halves the encode (5000 → 2400 µs) but costs sub-frame
+    /// readback. With a cheap send there is headroom and split wins. With an expensive send the
+    /// lost overlap outweighs the encode saving, and the arbiter must REFUSE the arm that looks
+    /// twice as fast — which is exactly the mistake an encode-only comparison makes.
+    #[test]
+    fn handicap_can_reverse_the_verdict() {
+        let (inc, chal) = (
+            M::NV_ENC_SPLIT_DISABLE_MODE as u32,
+            M::NV_ENC_SPLIT_TWO_FORCED_MODE as u32,
+        );
+        let run = |handicap: u64| {
+            let mut arb = SplitArbiter::with_handicap(inc, chal, handicap);
+            let mut live = inc;
+            for _ in 0..500 {
+                if arb.is_done() {
+                    break;
+                }
+                let us = if live == inc { 5000 } else { 2400 };
+                if let Some(a) = arb.on_frame(us) {
+                    match a {
+                        ArbAction::SwitchTo(m) | ArbAction::Settled(m) => live = m,
+                    }
+                }
+            }
+            live
+        };
+        // Cheap send: the 2600 µs encode saving is real, split wins.
+        assert_eq!(run(500), chal, "with a cheap send, split should win");
+        // Expensive send: 2400 + 3000 = 5400 against 5000 — the "twice as fast" arm is a LOSS
+        // end to end, and an encode-only comparison would have taken it.
+        assert_eq!(
+            run(3000),
+            inc,
+            "when losing sub-frame costs more than split saves, the incumbent must hold — this is \
+             the regression an encode-only arbiter would ship"
+        );
+    }
+
     /// Drive an arbiter with a fixed cost per arm and return every action it emitted.
     fn drive(incumbent_us: u64, challenger_us: u64) -> (Vec<ArbAction>, u32) {
         let (inc, chal) = (
             M::NV_ENC_SPLIT_DISABLE_MODE as u32,
             M::NV_ENC_SPLIT_TWO_FORCED_MODE as u32,
         );
-        let mut arb = SplitArbiter::new(inc, chal);
+        let mut arb = SplitArbiter::with_handicap(inc, chal, 0);
         let mut actions = Vec::new();
         // Whatever the session is currently running; the harness follows the arbiter's switches
         // so the cost it reports matches the arm actually in effect.
@@ -1323,7 +1377,7 @@ mod arbiter_tests {
             M::NV_ENC_SPLIT_DISABLE_MODE as u32,
             M::NV_ENC_SPLIT_TWO_FORCED_MODE as u32,
         );
-        let mut arb = SplitArbiter::new(inc, chal);
+        let mut arb = SplitArbiter::with_handicap(inc, chal, 0);
         let mut switched_at = None;
         let mut frame = 0usize;
         let mut outcome = None;

crates/pf-encode/src/lib.rs

@@ -287,6 +287,12 @@ impl Encoder for TrackedEncoder {
     fn set_wire_chunking(&mut self, shard_payload: usize) {
         self.inner.set_wire_chunking(shard_payload)
     }
+    // Same trap class again: unforwarded, the default no-op would leave the split arbitration
+    // permanently blind to send cost and it would never arbitrate the sub-frame trade — failing
+    // silently in the safe direction, which is the hardest kind to notice.
+    fn set_send_spread_us(&mut self, us: u32) {
+        self.inner.set_send_spread_us(us)
+    }
     // Forwarded for the same reason as `set_wire_chunking` above — an unforwarded default here
     // would silently leave the in-place backends pipelining past the capturer's ring.
     fn set_input_ring_depth(&mut self, depth: usize) {

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

@@ -746,6 +746,11 @@ fn send_loop(
     probe_result_tx: tokio::sync::mpsc::UnboundedSender<ProbeResult>,
     stop: Arc<AtomicBool>,
     perf: bool,
+    // Smoothed whole-AU paced-send time (µs) published for the ENCODE loop, which hands it to
+    // `Encoder::set_send_spread_us`. The split arbiter needs it to price what engaging split
+    // costs on HEVC (sub-frame readback, and with it the send/encode overlap) — a number the
+    // encoder cannot observe. Written here because this is the only thread that sees a send.
+    send_spread_us: Arc<AtomicU32>,
     // Streamed AUs go out as slice-granularity blocks ([`USER_FLAG_SLICE_STREAM`]'s contract)
     // instead of the legacy full-FEC-block shape.
     slice_wire: bool,
@@ -902,6 +907,19 @@ fn send_loop(
                                 );
                             }
                         }
+                        // Smooth before publishing: a single AU's spread swings with content and
+                        // FEC shape, and the arbiter turns this into a latency handicap that
+                        // decides an arm. EWMA (3:1) over completed AUs is enough to stop one
+                        // spike flipping a verdict.
+                        {
+                            let prev = send_spread_us.load(Ordering::Relaxed);
+                            let next = if prev == 0 {
+                                stat.spread_us
+                            } else {
+                                ((prev as u64 * 3 + stat.spread_us as u64) / 4) as u32
+                            };
+                            send_spread_us.store(next, Ordering::Relaxed);
+                        }
                         if perf || stats.rec.is_armed() {
                             // `encode_us`/`pace_us`/fps are valid for every frame (always measured),
                             // including the Windows relay + tail-drain frames. The cap/submit/wait splits
@@ -1770,6 +1788,10 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
     let force_idr = Arc::new(AtomicBool::new(false));
     // The send thread emits the web-console stats sample (it owns `session.stats()`); clone the
     // recorder so the capture loop keeps its own handle for the per-frame `is_armed()` gate.
+    // Shared with the send thread: it is the only place a paced send is observed, and the encode
+    // loop is the only place the encoder can be touched.
+    let send_spread_us = Arc::new(AtomicU32::new(0));
+    let send_spread_send = Arc::clone(&send_spread_us);
     let send_stats = SendStats {
         rec: stats.clone(),
         mode: live_mode.clone(),
@@ -1791,6 +1813,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                     probe_result_tx,
                     stop,
                     perf,
+                    send_spread_send,
                     slice_wire,
                     burst_cap,
                     fec_target,
@@ -2320,6 +2343,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         // off the driver into a full rebuild + IDR (~0.6 s each, four in one logged minute).
         // (The control task clamps its acks from the same atomic; this covers requests already
         // in flight when the ceiling was discovered.)
+        // Give the encoder the send cost it cannot measure. Cheap, and it is what lets the split
+        // arbiter price the sub-frame trade instead of refusing to arbitrate HEVC at all.
+        enc.set_send_spread_us(send_spread_us.load(Ordering::Relaxed));
         if let Some(k) = want_kbps.as_mut() {
             let ceiling = encoder_ceiling_kbps.load(Ordering::Relaxed);
             if ceiling != 0 && *k > ceiling {