perf(latency): T1.1 frame-driven encode trigger + T1.4 time-based flush thresholds

design/latency-reduction-2026-07.md tier 1, remaining halves:

- T1.1: the native encode loop wakes on the capture's ACTUAL arrival instead of
  sampling at a free-running tick — deletes the sample-and-hold (~half a frame
  interval on average, a full one worst-case: ~8ms avg @60fps). New
  Capturer::supports_arrival_wait/wait_arrival pair (IDD-push waits its
  frame-ready event against the shared-header token; the PipeWire portal blocks
  its channel with a pending stash); backends without an arrival signal — and
  PUNKTFUNK_FRAME_DRIVEN=0 — keep the legacy tick bit-identically. A
  0.9x-interval rate floor caps encode at ~1.11x target when the compositor
  outruns the session; a +0.5x-interval keepalive keeps static desktops
  re-encoding at 1.5x-interval cadence. Pacing deadlines re-anchor to the
  actual submit so they can't drift against the arrival clock. GameStream
  plane untouched.
- T1.4: the jump-to-live detectors run on WALL-CLOCK now (STANDING_TIME /
  FLUSH_AFTER = 250ms) instead of 30-frame counts whose meaning scaled with
  fps (500ms @60 but 125ms @240 — and stretching further under T1.1's slower
  static-scene repeats). The queue trip also requires depth still >= high, so
  a hysteresis-band hover can't fire on elapsed time alone.

Validated: .21 Linux 185 core + 177 host + pf-capture tests, clippy
-D warnings; .133 Windows cargo check of pf-capture + punktfunk-host green.

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

crates/pf-capture/src/lib.rs

@@ -32,6 +32,22 @@ pub trait Capturer: Send {
         self.next_frame().map(Some)
     }
 
+    /// Whether this backend can block until a frame ARRIVES ([`wait_arrival`]
+    /// (Self::wait_arrival)) — the frame-driven encode trigger (latency plan T1.1). `false`
+    /// (the default) keeps the encode loop on its legacy fixed-cadence tick for this backend.
+    fn supports_arrival_wait(&self) -> bool {
+        false
+    }
+
+    /// Block until a FRESH frame is available via [`try_latest`](Self::try_latest) or
+    /// `deadline` passes — the encode loop's frame-driven wait (latency plan T1.1): waking on
+    /// the compositor's publish instead of sampling at a free-running tick deletes the
+    /// sample-and-hold (~half a frame interval on average). Must NOT consume the frame (the
+    /// loop's `try_latest` call does); backends buffer internally where the arrival channel
+    /// can't be peeked. Only called when [`supports_arrival_wait`](Self::supports_arrival_wait)
+    /// is `true`; errors surface at the following `try_latest`.
+    fn wait_arrival(&mut self, _deadline: std::time::Instant) {}
+
     /// Gate expensive per-frame work so the capturer can be kept alive (reused) between
     /// streams without burning CPU. The portal capturer skips the de-pad copy while inactive;
     /// the default is a no-op (synthetic sources are produced on demand). Set `true` for the

crates/pf-capture/src/linux/mod.rs

@@ -35,6 +35,10 @@ use std::time::Duration;
 /// and no second session to conflict with).
 pub struct PortalCapturer {
     frames: Receiver<CapturedFrame>,
+    /// A frame [`wait_arrival`](Capturer::wait_arrival) received while blocking (the channel
+    /// can't be peeked, so the wait must consume) — always the FIRST candidate the next
+    /// `try_latest`/`next_frame` considers, before draining anything newer off the channel.
+    pending: Option<CapturedFrame>,
     active: Arc<AtomicBool>,
     /// Set true once the PipeWire stream agrees a video format. Read in [`next_frame`]'s timeout
     /// branch to tell "format never negotiated" (modifier/format mismatch) apart from "negotiated
@@ -166,6 +170,7 @@ impl PwHandles {
     fn into_capturer(self, node_id: u32, keepalive: Option<Box<dyn Send>>) -> PortalCapturer {
         PortalCapturer {
             frames: self.frames,
+            pending: None,
             active: self.active,
             negotiated: self.negotiated,
             streaming: self.streaming,
@@ -266,6 +271,9 @@ impl Capturer for PortalCapturer {
                     self.node_id
                 ));
             }
+            if let Some(f) = self.pending.take() {
+                return Ok(f); // a wait_arrival stash outranks the channel (it's older)
+            }
             let slice = Duration::from_millis(500)
                 .min(deadline.saturating_duration_since(std::time::Instant::now()));
             match self.frames.recv_timeout(slice) {
@@ -276,6 +284,27 @@ impl Capturer for PortalCapturer {
         }
     }
 
+    fn supports_arrival_wait(&self) -> bool {
+        true
+    }
+
+    fn wait_arrival(&mut self, deadline: std::time::Instant) {
+        // Frame-driven trigger (latency plan T1.1): block on the PipeWire channel until the
+        // compositor delivers a frame or the deadline passes. The channel can't be peeked, so
+        // the received frame is stashed in `pending` — `try_latest` starts from it and still
+        // drains anything newer. A broken/ended stream just returns; the following
+        // `try_latest` surfaces the error through its existing paths.
+        if self.pending.is_some() || self.broken.load(Ordering::Relaxed) {
+            return;
+        }
+        let Some(left) = deadline.checked_duration_since(std::time::Instant::now()) else {
+            return;
+        };
+        if let Ok(f) = self.frames.recv_timeout(left) {
+            self.pending = Some(f);
+        }
+    }
+
     fn try_latest(&mut self) -> Result<Option<CapturedFrame>> {
         if self.broken.load(Ordering::Relaxed) {
             return Err(anyhow!(
@@ -285,8 +314,9 @@ impl Capturer for PortalCapturer {
             ));
         }
         // Drain to the newest queued frame without blocking; `None` means the compositor
-        // hasn't produced a new frame since last call (static/idle desktop).
+        // hasn't produced a new frame since last call (static/idle desktop). A frame
-        let mut latest = None;
+        // `wait_arrival` stashed is the oldest candidate — anything on the channel is newer.
+        let mut latest = self.pending.take();
         loop {
             match self.frames.try_recv() {
                 Ok(frame) => latest = Some(frame),

crates/pf-capture/src/windows/idd_push.rs

@@ -1625,6 +1625,32 @@ impl Capturer for IddPushCapturer {
         self.try_consume()
     }
 
+    fn supports_arrival_wait(&self) -> bool {
+        true
+    }
+
+    fn wait_arrival(&mut self, deadline: Instant) {
+        // Frame-driven trigger (latency plan T1.1): wake the encode loop the moment the driver
+        // publishes a frame we haven't consumed. The shared-header token is the truth (state,
+        // not edge — the auto-reset event may have been consumed by an earlier wait); the event
+        // is just the wakeup, waited in ≤16 ms slices exactly like `next_frame`'s bringup wait.
+        loop {
+            let tok = frame::FrameToken::unpack(self.latest());
+            if tok.generation == self.generation && u64::from(tok.seq) != self.last_seq {
+                return; // a fresh publish exists — `try_latest` will consume it
+            }
+            let Some(left) = deadline.checked_duration_since(Instant::now()) else {
+                return;
+            };
+            let ms = (left.as_millis() as u32).clamp(1, 16);
+            // SAFETY: `self.event` is the live frame-ready `OwnedHandle` this capturer owns; its
+            // raw value (borrowed for the call, so it outlives this synchronous wait) is a valid
+            // auto-reset event handle. `WaitForSingleObject` only reads the handle; the bounded
+            // timeout keeps the wait sliced.
+            let _ = unsafe { WaitForSingleObject(HANDLE(self.event.as_raw_handle()), ms) };
+        }
+    }
+
     fn hdr_meta(&self) -> Option<punktfunk_core::quic::HdrMeta> {
         // While the display is HDR we emit BT.2020 PQ (Rgb10a2) → the encoder forces HEVC Main10 + the
         // PQ VUI; pair that with a mastering-display SEI so any decoder tone-maps from a real grade. The

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

@@ -15,13 +15,16 @@ pub(crate) const QUEUE_HIGH: usize = 6;
 /// A true standing queue never falls back to this; a clump does within a few frames.
 pub(crate) const QUEUE_LOW: usize = 2;
 
-/// Consecutive frames the hand-off queue must sit ≥ [`QUEUE_HIGH`] (never dropping to [`QUEUE_LOW`])
+/// How long the hand-off queue must sit ≥ [`QUEUE_HIGH`] (never dropping to [`QUEUE_LOW`], and
-/// before the pump declares a standing backlog and jumps to live. ~0.5 s at 60 fps — long enough that
+/// still ≥ high at the trip) before the pump declares a standing backlog and jumps to live.
-/// a burst/clump (which drains in a few frames) never reaches it.
+/// WALL-CLOCK (latency plan T1.4) — the old 30-FRAME count made the accepted backlog scale with
-pub(crate) const STANDING_FRAMES: u32 = 30;
+/// fps (500 ms @60 but 125 ms @240) and stretched further under the frame-driven host's slower
+/// static-scene repeat cadence. 250 ms sits comfortably above a Wi-Fi clump's drain time (a
+/// clump empties in a few frames, ≤ ~100 ms) at any rate.
+pub(crate) const STANDING_TIME: Duration = Duration::from_millis(250);
 
 /// Memory backstop on the pre-decode hand-off queue. The standing-queue detector jumps to live long
-/// before this (typically ≤ QUEUE_HIGH + STANDING_FRAMES deep), and a jump already requested a
+/// before this (typically ≤ QUEUE_HIGH + a [`STANDING_TIME`] run deep), and a jump already requested a
 /// keyframe, so on the rare path that outruns it (a wedged consumer during the flush cooldown) dropping
 /// the OLDEST queued AU is safe — the pending IDR re-anchors decode regardless. Purely bounds memory.
 const FRAME_QUEUE_HARD_CAP: usize = 90;
@@ -31,14 +34,15 @@ const FRAME_QUEUE_HARD_CAP: usize = 90;
 /// AUs and requests a keyframe) instead of playing that far behind forever. Deliberately generous — an
 /// interactive stream is unusable well before 400 ms, but the bound must sit safely above the skew
 /// handshake's own error (≈ RTT/2) plus normal delivery jitter so a healthy stream can never trip it.
-/// This is the CLOCK-BASED detector; the clock-free [`QUEUE_HIGH`]/[`STANDING_FRAMES`] detector covers
+/// This is the CLOCK-BASED detector; the clock-free [`QUEUE_HIGH`]/[`STANDING_TIME`] detector covers
 /// same-clock and no-handshake sessions (where `clock_offset_ns == 0` disarms this one).
 pub(crate) const FLUSH_LATENCY: Duration = Duration::from_millis(400);
 
-/// How many CONSECUTIVE over-bound frames arm the clock-based jump (~0.5 s at 60 fps). A genuine
+/// How long frames must run CONTINUOUSLY over the [`FLUSH_LATENCY`] bound before the clock-based
-/// standing queue puts EVERY frame over the bound; a one-off burst (an IDR, a Wi-Fi scan blip) clears
+/// jump fires. WALL-CLOCK (latency plan T1.4, was a 30-frame count — same fps-scaling problem as
-/// within a frame or two and never reaches the count.
+/// the standing-queue detector). A genuine standing queue puts EVERY frame over the bound; a
-pub(crate) const FLUSH_AFTER_FRAMES: u32 = 30;
+/// one-off burst (an IDR, a Wi-Fi scan blip) clears within a frame or two and resets the run.
+pub(crate) const FLUSH_AFTER: Duration = Duration::from_millis(250);
 
 /// Minimum spacing between jump-to-live events, so a bottleneck that instantly rebuilds the queue (a
 /// link/consumer that can't sustain the bitrate at all) degrades into a periodic skip + a logged

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

@@ -1,8 +1,8 @@
 //! The client worker: QUIC handshake + control/input/datagram tasks + the blocking data-plane pump.
 
 use super::frame_channel::{
-    CLOCK_RESYNC_INTERVAL, FLUSH_AFTER_FRAMES, FLUSH_COOLDOWN, FLUSH_LATENCY,
+    CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, FLUSH_LATENCY,
-    NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_FRAMES,
+    NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_TIME,
 };
 use super::worker::reject_from_close;
 use super::*;
@@ -711,12 +711,13 @@ pub(super) async fn run_pump(args: WorkerArgs) {
         let mut capacity_probe_deadline: Option<Instant> = None;
         let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32);
         let mut flush_in_window = false;
-        // Jump-to-live state (see the guard in the loop below): the clock-based over-bound run
+        // Jump-to-live state (see the guard in the loop below): when the clock-based over-bound
-        // (`stale_frames`, armed only when the skew handshake succeeded so the clocks are comparable),
+        // run began (`stale_since`, armed only when the skew handshake succeeded so the clocks
-        // the clock-free non-draining-queue run (`standing_frames`), and the last-jump instant for the
+        // are comparable), when the clock-free non-draining-queue run began (`standing_since`),
-        // shared cooldown.
+        // and the last-jump instant for the shared cooldown. Wall-clock runs (T1.4), not frame
-        let mut stale_frames: u32 = 0;
+        // counts — the detectors' sensitivity must not scale with fps or repeat cadence.
-        let mut standing_frames: u32 = 0;
+        let mut stale_since: Option<Instant> = None;
+        let mut standing_since: Option<Instant> = None;
         let mut last_flush: Option<Instant> = None;
         // Clock-detector health: consecutive clock-triggered flushes that found no local backlog
         // (see NOOP_FLUSH_DATAGRAMS). Reaching NOOP_CLOCK_FLUSHES_TO_DISARM turns the clock-based
@@ -737,7 +738,7 @@ pub(super) async fn run_pump(args: WorkerArgs) {
             let gen = pump_clock_gen.load(Ordering::Relaxed);
             if gen != seen_clock_gen {
                 seen_clock_gen = gen;
-                stale_frames = 0;
+                stale_since = None;
                 noop_clock_flushes = 0;
                 if !clock_detector_armed {
                     clock_detector_armed = true;
@@ -956,18 +957,18 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                     // FLUSH_COOLDOWN, both suspended during a speed test (the probe MEASURES a saturated
                     // queue; flushing would corrupt its counters):
                     //  * clock-based — completed frames sit > FLUSH_LATENCY behind the skew-corrected
-                    //    capture clock for FLUSH_AFTER_FRAMES straight. Needs the skew handshake, and
+                    //    capture clock continuously for FLUSH_AFTER. Needs the skew handshake, and
                     //    also catches kernel/reassembler backlog the hand-off queue hasn't reached yet.
                     //  * clock-free — the pre-decode hand-off queue stopped draining: its depth stayed
-                    //    ≥ QUEUE_HIGH (never falling to QUEUE_LOW) for STANDING_FRAMES straight. Works
+                    //    ≥ QUEUE_HIGH (never falling to QUEUE_LOW, still high at the trip) for
-                    //    with no handshake / a same-clock session (where the clock path is disarmed),
+                    //    STANDING_TIME. Works with no handshake / a same-clock session (where the
-                    //    and is the direct signal that the embedder can't keep up. A transient Wi-Fi
+                    //    clock path is disarmed), and is the direct signal that the embedder can't
-                    //    clump drains in a few frames and never reaches the count.
+                    //    keep up. A transient Wi-Fi clump drains within ~100 ms and never trips it.
                     if probe_active {
                         // Keep both detectors disarmed across a speed test so its (deliberately)
-                        // saturated queue doesn't leave a primed count that fires the moment it ends.
+                        // saturated queue doesn't leave a primed run that fires the moment it ends.
-                        stale_frames = 0;
+                        stale_since = None;
-                        standing_frames = 0;
+                        standing_since = None;
                     } else {
                         let lat_ns = if clock_offset_ns != 0 {
                             now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128
@@ -985,23 +986,27 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                             && clock_offset_ns != 0
                             && lat_ns > FLUSH_LATENCY.as_nanos() as i128
                         {
-                            stale_frames += 1;
+                            stale_since.get_or_insert_with(Instant::now);
                         } else {
-                            stale_frames = 0;
+                            stale_since = None;
                         }
                         let depth = frames.depth();
                         if depth >= QUEUE_HIGH {
-                            standing_frames += 1;
+                            standing_since.get_or_insert_with(Instant::now);
                         } else if depth <= QUEUE_LOW {
-                            standing_frames = 0;
+                            standing_since = None;
                         }
-                        let clock_behind = stale_frames >= FLUSH_AFTER_FRAMES;
+                        // The queue trip additionally requires the depth to still be high NOW, so
-                        let queue_behind = standing_frames >= STANDING_FRAMES;
+                        // a run that started ≥ high but is hovering in the hysteresis band (a
+                        // clump mid-drain) never fires on elapsed time alone.
+                        let clock_behind = stale_since.is_some_and(|t| t.elapsed() >= FLUSH_AFTER);
+                        let queue_behind = depth >= QUEUE_HIGH
+                            && standing_since.is_some_and(|t| t.elapsed() >= STANDING_TIME);
                         if (clock_behind || queue_behind)
                             && last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN)
                         {
-                            stale_frames = 0;
+                            stale_since = None;
-                            standing_frames = 0;
+                            standing_since = None;
                             last_flush = Some(Instant::now());
                             flush_in_window = true; // strongest "link can't hold the rate" signal
                             let flushed = session.flush_backlog().unwrap_or(0);

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

@@ -193,6 +193,14 @@ fn service_probes(
     }
 }
 
+/// T1.1 frame-driven encode trigger (latency plan): `PUNKTFUNK_FRAME_DRIVEN=0` restores the
+/// legacy fixed-cadence tick everywhere (backends without an arrival wait keep it regardless —
+/// see [`pf_capture::Capturer::supports_arrival_wait`]).
+fn frame_driven_enabled() -> bool {
+    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
+    *ON.get_or_init(|| std::env::var("PUNKTFUNK_FRAME_DRIVEN").as_deref() != Ok("0"))
+}
+
 /// Seal one access unit and send it with MICROBURST pacing (the shared
 /// [`send_pacing`](crate::send_pacing) policy, native parameterization): the first `burst_cap`
 /// bytes go out immediately (one absorbed burst the NIC / socket tx-buffer can swallow), and
@@ -1784,7 +1792,14 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         }
         // This frame's pacing deadline (the next frame's due time); the send thread spreads a big frame
         // up to here. Each in-flight frame carries its own (capture_ns, deadline) for when it's polled.
-        next += interval;
+        // Frame-driven mode (T1.1) re-anchors to the ACTUAL submit — arrivals are the clock, and a
+        // fixed `+= interval` grid would drift against them and squeeze the pacing budget; the
+        // legacy tick keeps its fixed grid (with the catch-up reset in the tail).
+        next = if frame_driven_enabled() && capturer.supports_arrival_wait() {
+            std::time::Instant::now() + interval
+        } else {
+            next + interval
+        };
         inflight.push_back((capture_ns, submit_ns, next));
         // Drain the OLDEST in-flight frames, keeping at most depth-1 deferred. At depth 1 this polls
         // immediately after every submit (synchronous); at depth 2 it polls N right after submitting N+1,
@@ -1919,11 +1934,26 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 "encode stall detected — encoder rebuilt in place, forcing an IDR");
             last_au_at = std::time::Instant::now();
         }
+        if frame_driven_enabled() && capturer.supports_arrival_wait() {
+            // T1.1 frame-driven trigger: instead of sleeping out the whole tick and then
+            // SAMPLING (which holds a frame that arrived just after the previous sample for up
+            // to a full interval — ~half on average), sleep only to the rate floor and then
+            // wake on the capture's actual arrival. The 0.9×interval floor caps the encode
+            // rate at ~1.11× target when the source runs faster (compositor Hz > session fps);
+            // the +0.5×interval keepalive keeps a static desktop re-encoding (bitrate shape,
+            // client liveness) at 1.5×interval cadence and bounds control-servicing latency.
+            let earliest = next - interval.mul_f32(0.1);
+            if let Some(d) = earliest.checked_duration_since(std::time::Instant::now()) {
+                std::thread::sleep(d);
+            }
+            capturer.wait_arrival(next + interval.mul_f32(0.5));
+        } else {
             match next.checked_duration_since(std::time::Instant::now()) {
                 Some(d) => std::thread::sleep(d),
                 None => next = std::time::Instant::now(),
             }
         }
+    }
     // Drain the in-flight tail (the depth-1 frames submitted but not yet polled) so the last frames still
     // reach the client instead of being dropped on the way out.
     while let Some((cap_ns, sub_ns, deadline)) = inflight.pop_front() {