feat(pf-encode): wire split arbitration on Windows too

The last coverage gap, and only worth building once S1 proved it possible: the
Windows backend drives NV_ENC_DEVICE_TYPE_DIRECTX, and an in-place splitEncodeMode
change had never been tested there. It works (071358cb), so the arbiter is now
ungated from Linux-only to the union of both direct-SDK backends and wired into
windows/nvenc.rs: the submit stamp, the feed hook on AU completion,
apply_split_mode, split_key, arm_split_arbiter, and set_send_spread_us.

Same gates as Linux, and they are correctness conditions rather than preferences:
opt-in while it earns trust, an operator PUNKTFUNK_SPLIT_ENCODE pin always wins,
a cached verdict short-circuits, >=2 engines, never H.264, and the sub-frame
trade is only entered when the host has actually reported a send spread to price
it with. The one Windows-specific difference is that `async_rt` is a real
possibility here (opt-in two-thread retrieve) and the arbiter refuses it, because
under pipelined retrieve the submit->AU span includes queue depth and the
comparison would be noise.

⚠ Two more instances of the same item-level dead_code trap, caught by the Windows
run and not by reasoning -- that is now 4 and 5:
- `clear_split_verdicts` is called only by the Linux on-hw test, so it is dead on
  Windows; gated to `all(test, target_os = "linux")`.
- The arbiter methods first landed inside `impl Encoder` rather than the inherent
  impl (the anchor I used, supports_chunked_poll, is a trait method), which the
  compiler caught as "not a member of trait Encoder".

Verified .158 (RTX 4090 / Ada, driver 610.88, D3D11): clippy --features nvenc
--all-targets -D warnings clean, and 2 on-hardware NVENC tests green including S1
re-run with the arbitration code in place (engines=2 latched, DISABLE->TWO_FORCED
accepted, zero IDRs, reverse accepted). Verified .21: clippy clean with AND
without the nvenc feature, 65 unit tests, 25/25 NVENC on-hardware. fmt clean.

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

@@ -252,11 +252,11 @@ mod split_subframe_tests {
     }
 }
 
-// Split arbitration is wired into the Linux direct-SDK backend only for now, and
-// `nvenc_core` compiles on Windows too — so every item below is linux-gated or it trips
-// the item-level dead_code trap this file already carries a scar from (see
-// `subframe_env_forced`). Ungating is part of the Windows wiring, not a cleanup.
-#[cfg(target_os = "linux")]
+// Split arbitration now runs on BOTH direct-SDK backends, so these are gated to the union of
+// the two rather than to Linux. Kept gated at all because `nvenc_core` is also reachable from
+// builds where neither backend is compiled, and an ungated item there is the item-level
+// dead_code trap this file already carries three scars from (see `subframe_env_forced`).
+#[cfg(any(target_os = "linux", windows))]
 /// What the split arbiter wants the backend to do next.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(super) enum ArbAction {
@@ -266,7 +266,7 @@ pub(super) enum ArbAction {
     Settled(u32),
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum ArbState {
     MeasuringIncumbent,
@@ -275,7 +275,7 @@ enum ArbState {
     Done,
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 /// Picks the faster of two NVENC split modes **on the live session**, by measuring both.
 ///
 /// This exists because the alternative — predicting the right mode at open — cannot work: the
@@ -311,19 +311,19 @@ pub(super) struct SplitArbiter {
 }
 
 /// Frames discarded after a switch before the challenger is judged (measured — see the struct doc).
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 const SETTLE_FRAMES: u32 = 16;
 /// Frames measured per arm. Long enough to median out content variation, short enough that the
 /// whole arbitration is over in well under a second at 60 fps.
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 const SAMPLE_FRAMES: usize = 24;
 /// The challenger must beat the incumbent by this much to win. Switching is not free (a
 /// reconfigure, and for HEVC it costs sub-frame readback), so a coin-flip difference should leave
 /// the session where it already is.
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 const WIN_MARGIN_PCT: u64 = 10;
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 impl SplitArbiter {
     /// `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`].
@@ -405,7 +405,7 @@ impl SplitArbiter {
     }
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 fn median(v: &mut [u64]) -> u64 {
     v.sort_unstable();
     v[v.len() / 2]
@@ -455,7 +455,7 @@ pub(super) fn store_ceiling(key: CeilingKey, bps: u64) {
     ceilings().lock().unwrap().insert(key, bps);
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 /// A config's identity for the split-arbitration verdict cache — [`CeilingKey`] **minus
 /// `split_mode`**, because the split mode is the thing being decided. Including it would key each
 /// verdict under the arm that produced it and the cache could never answer "which arm should this
@@ -471,14 +471,14 @@ pub(super) struct SplitKey {
     pub chroma_444: bool,
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 fn split_verdicts() -> &'static std::sync::Mutex<std::collections::HashMap<SplitKey, u32>> {
     static V: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<SplitKey, u32>>> =
         std::sync::OnceLock::new();
     V.get_or_init(Default::default)
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 /// The split mode a previous arbitration found fastest for `key` this process lifetime.
 ///
 /// Process-lifetime and advisory, exactly like [`cached_ceiling`]: a session that reads a verdict
@@ -489,17 +489,19 @@ pub(super) fn cached_split_verdict(key: &SplitKey) -> Option<u32> {
     split_verdicts().lock().unwrap().get(key).copied()
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 /// Record an arbitration result for `key`.
 pub(super) fn store_split_verdict(key: SplitKey, mode: u32) {
     split_verdicts().lock().unwrap().insert(key, mode);
 }
 
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 /// Drop every cached verdict. Test-only: the cache is process-global, so an on-hardware test that
 /// runs an arbitration would otherwise leak its verdict into every later test that opens the same
 /// config with `PUNKTFUNK_SPLIT_ENCODE` unset — which is exactly the shape the D5 legs use.
-#[cfg(test)]
+// Linux-only: its sole caller is `nvenc_cuda`'s arbitration on-hw test. Ungated it is dead
+// code on Windows — the same item-level trap, now four times over.
+#[cfg(all(test, target_os = "linux"))]
 pub(super) fn clear_split_verdicts() {
     split_verdicts().lock().unwrap().clear();
 }
@@ -1184,7 +1186,7 @@ pub(super) unsafe fn apply_low_latency_config(cfg: &mut nv::NV_ENC_CONFIG, c: Lo
     }
 }
 
-#[cfg(all(test, target_os = "linux"))]
+#[cfg(all(test, any(target_os = "linux", windows)))]
 mod arbiter_tests {
     use super::{ArbAction, SplitArbiter, SETTLE_FRAMES};
 

crates/pf-encode/src/enc/windows/nvenc.rs

@@ -49,8 +49,11 @@ use super::nvenc_core::{
 };
 // Moved to `codec.rs` (WP4) so the libav path, which builds without the `nvenc` feature, can share
 // one split policy instead of keeping the copy that had already drifted.
+use super::nvenc_core::{
+    cached_split_verdict, store_split_verdict, ArbAction, SplitArbiter, SplitKey,
+};
 use super::nvenc_status;
-use super::resolve_split_mode;
+use super::{max_forced_split_mode, resolve_split_mode};
 use super::{AuChunk, ChromaFormat, Codec, EncodedFrame, Encoder, EncoderCaps};
 use anyhow::{anyhow, bail, Context, Result};
 use pf_frame::{CapturedFrame, FramePayload, PixelFormat};
@@ -600,6 +603,16 @@ pub struct NvencD3d11Encoder {
     /// the driver accepts a split wider than the hardware and silently encodes narrower, so this
     /// is the only honest source for how wide we may go (see `codec::max_forced_split_mode`).
     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. `0` = never reported, which keeps
+    /// the arbiter out of the sub-frame trade it cannot otherwise price.
+    send_spread_us: u32,
+    /// Sub-frame state the session was OPENED able to run, so a return to a non-forced split can
+    /// restore it 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.
+    arbiter: Option<SplitArbiter>,
     /// (bitstream, mapped input resource to unmap after retrieval, pts_ns, recovery-anchor) per
     /// in-flight encode. The fourth field tags the first frame encoded after a successful
     /// [`invalidate_ref_frames`](Encoder::invalidate_ref_frames) — the clean re-anchor P-frame the
@@ -762,6 +775,10 @@ impl NvencD3d11Encoder {
             async_supported: false,
             subframe_cap: false,
             encoder_engines: 0,
+            last_submit_at: None,
+            send_spread_us: 0,
+            subframe_opened_with: false,
+            arbiter: None,
             pending: VecDeque::new(),
             frame_idx: 0,
             force_kf: false,
@@ -1048,6 +1065,126 @@ impl NvencD3d11Encoder {
         Ok(cfg)
     }
 
+    /// The config identity this session's split verdict is cached under.
+    fn split_key(&self) -> SplitKey {
+        // Same GPU identity as `ceiling_key`: the selected render adapter's LUID, `0` when
+        // unresolved. Advisory either way.
+        let gpu = pf_gpu::resolve_render_adapter_luid()
+            .map(|l| ((l.HighPart as u32 as u64) << 32) | l.LowPart as u64)
+            .unwrap_or(0);
+        SplitKey {
+            gpu,
+            codec: self.codec,
+            width: self.width,
+            height: self.height,
+            fps: self.fps,
+            bit_depth: self.bit_depth,
+            chroma_444: self.chroma_444,
+        }
+    }
+
+    /// Move the LIVE session to `mode` without an IDR. Windows twin of the Linux method; S1 on
+    /// D3D11 proved `nvEncReconfigureEncoder` takes a changed `splitEncodeMode` with
+    /// `resetEncoder=0` and emits no keyframe on this device type too.
+    fn apply_split_mode(&mut self, mode: u32) -> bool {
+        let (prev_mode, prev_sub) = (self.split_mode, self.subframe_on);
+        let (mode, subframe) = resolve_split_subframe(
+            self.codec,
+            mode,
+            self.subframe_opened_with,
+            subframe_env_forced(),
+        );
+        self.split_mode = mode;
+        self.subframe_on = subframe;
+        if self.reconfigure_bitrate(self.bitrate_bps) {
+            true
+        } else {
+            tracing::warn!(
+                from = prev_mode,
+                to = mode,
+                "NVENC split arbitration: driver refused the in-place split change — staying put"
+            );
+            self.split_mode = prev_mode;
+            self.subframe_on = prev_sub;
+            false
+        }
+    }
+
+    /// Feed one frame's encode cost to the split arbiter and act on its verdict.
+    fn feed_split_arbiter(&mut self, encode_us: u64) {
+        let Some(arb) = self.arbiter.as_mut() else {
+            return;
+        };
+        let action = arb.on_frame(encode_us);
+        let done = arb.is_done();
+        match action {
+            Some(ArbAction::SwitchTo(mode)) => {
+                if !self.apply_split_mode(mode) {
+                    self.arbiter = None;
+                    return;
+                }
+            }
+            Some(ArbAction::Settled(mode)) => store_split_verdict(self.split_key(), mode),
+            None => {}
+        }
+        if done {
+            store_split_verdict(self.split_key(), self.split_mode);
+            self.arbiter = None;
+        }
+    }
+
+    /// Decide whether this session may run a live split experiment. Same gates as the Linux
+    /// backend — see its `arm_split_arbiter` for why each one is a correctness condition rather
+    /// than a preference; the only Windows difference is that `async_rt` is a real possibility
+    /// here (opt-in two-thread retrieve), and under it the submit→AU span includes queue depth,
+    /// so the comparison would be noise.
+    fn arm_split_arbiter(&mut self) {
+        if !matches!(
+            std::env::var("PUNKTFUNK_NVENC_SPLIT_ARBITRATE").as_deref(),
+            Ok("1")
+        ) {
+            return;
+        }
+        if std::env::var_os("PUNKTFUNK_SPLIT_ENCODE").is_some()
+            || cached_split_verdict(&self.split_key()).is_some()
+            || self.async_rt.is_some()
+            || self.encoder_engines < 2
+            || self.codec == Codec::H264
+        {
+            return;
+        }
+        let handicap_us = if self.subframe_on && self.codec != Codec::Av1 {
+            if self.send_spread_us == 0 || self.slices < 2 {
+                return;
+            }
+            let slices = self.slices as u64;
+            self.send_spread_us as u64 * (slices - 1) / slices
+        } else {
+            0
+        };
+        let disable = nv::NV_ENC_SPLIT_ENCODE_MODE::NV_ENC_SPLIT_DISABLE_MODE as u32;
+        let widest = max_forced_split_mode(self.encoder_engines);
+        let challenger = if self.split_mode == widest {
+            disable
+        } else {
+            widest
+        };
+        if challenger == self.split_mode {
+            return;
+        }
+        tracing::info!(
+            incumbent = self.split_mode,
+            challenger,
+            handicap_us,
+            "NVENC split arbitration armed (Windows) — measuring both arms live (no IDR)"
+        );
+        self.arbiter = Some(SplitArbiter::with_handicap(
+            self.split_mode,
+            challenger,
+            handicap_us,
+        ));
+    }
+
     /// This session config's identity in the process-lifetime bitrate-ceiling cache
     /// (`nvenc_core::{cached_ceiling, store_ceiling}`). GPU identity is the selected render
     /// adapter's LUID — the adapter the capturer's device (and so this session) lives on; `0`
@@ -1433,6 +1570,8 @@ impl NvencD3d11Encoder {
                 self.bitrate_bps / 1_000_000,
                 self.codec_guid
             );
+            self.subframe_opened_with = self.subframe_on;
+            self.arm_split_arbiter();
             Ok(())
         }
     }
@@ -1776,6 +1915,9 @@ impl Encoder for NvencD3d11Encoder {
                 anchor,
                 idr_hint,
             ));
+            // Split-arbiter cost stamp; only meaningful on the sync depth-1 path, which is the
+            // only path `arm_split_arbiter` allows an experiment on.
+            self.last_submit_at = Some(std::time::Instant::now());
             // Async: hand the in-flight encode to the retrieve thread (channel capacity = POOL ≥
             // in-flight, so this send never blocks). The pending entry above pairs with its
             // completion FIFO in `absorb_done`.
@@ -1959,6 +2101,13 @@ impl Encoder for NvencD3d11Encoder {
             if !map.is_null() {
                 let _ = (api().unmap_input_resource)(self.encoder, map);
             }
+            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(EncodedFrame {
                 data,
                 pts_ns,
@@ -2218,6 +2367,10 @@ impl Encoder for NvencD3d11Encoder {
         }
     }
 
+    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.