feat(android): live stats HUD + low-latency decode tuning

Stats HUD (mirrors the Apple client): the decode thread accumulates FPS, receive
throughput, and capture->client latency (p50/p95, skew-corrected) in Rust
(clients/android/native/src/stats.rs); nativeVideoStats drains a snapshot ~1 Hz
over JNI as a DoubleArray. StreamScreen renders a Compose overlay
(W*H@Hz / fps / Mb/s / latency, + dropped-under-loss), toggled by a Settings
switch (persisted, default on) or a 3-finger tap.

Performance (decode.rs):
- ANativeWindow_setFrameRate(refresh_hz): align display vsync to the stream rate
  (no 60-in-120 judder); safe since minSdk 31 >= API 30.
- Raise the decode thread toward URGENT_DISPLAY (best-effort setpriority) so
  background work can't preempt it under load.
- Codec low-latency hints KEY_PRIORITY=0 (realtime) + KEY_OPERATING_RATE.

Verified host-side: cargo build/clippy/fmt --workspace (the ungated stats + JNI
accessor). The android-gated decode.rs (NDK) and the Kotlin build only in CI
(android.yml: gradle + cargo-ndk) -- APIs verified against crate sources.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -2545,6 +2545,7 @@ version = "0.0.1"
 dependencies = [
  "android_logger",
  "jni",
+ "libc",
  "log",
  "ndk",
  "opus",

clients/android/app/src/main/kotlin/io/unom/punktfunk/Settings.kt

@@ -16,6 +16,8 @@ data class Settings(
     val compositor: Int = 0,
     val gamepad: Int = 0,
     val micEnabled: Boolean = false,
+    /** Show the live stats overlay (FPS / throughput / latency) during a stream. */
+    val statsHudEnabled: Boolean = true,
 )
 
 /** Loads/saves [Settings] in the app-private `punktfunk_settings` prefs. */
@@ -31,6 +33,7 @@ class SettingsStore(context: Context) {
         compositor = prefs.getInt(K_COMPOSITOR, 0),
         gamepad = prefs.getInt(K_GAMEPAD, 0),
         micEnabled = prefs.getBoolean(K_MIC, false),
+        statsHudEnabled = prefs.getBoolean(K_HUD, true),
     )
 
     fun save(s: Settings) {
@@ -42,6 +45,7 @@ class SettingsStore(context: Context) {
             .putInt(K_COMPOSITOR, s.compositor)
             .putInt(K_GAMEPAD, s.gamepad)
             .putBoolean(K_MIC, s.micEnabled)
+            .putBoolean(K_HUD, s.statsHudEnabled)
             .apply()
     }
 
@@ -53,6 +57,7 @@ class SettingsStore(context: Context) {
         const val K_COMPOSITOR = "compositor"
         const val K_GAMEPAD = "gamepad"
         const val K_MIC = "mic_enabled"
+        const val K_HUD = "stats_hud_enabled"
     }
 }
 

clients/android/app/src/main/kotlin/io/unom/punktfunk/SettingsScreen.kt

@@ -111,6 +111,22 @@ fun SettingsScreen(initial: Settings, onChange: (Settings) -> Unit, onBack: () -
                 },
             )
         }
+
+        // Live stats overlay (FPS / throughput / capture→client latency). A 3-finger tap in-stream
+        // toggles it without coming back here.
+        Row(modifier = Modifier.fillMaxWidth(), verticalAlignment = Alignment.CenterVertically) {
+            Column(Modifier.weight(1f)) {
+                Text("Stats overlay", style = MaterialTheme.typography.bodyLarge)
+                Text(
+                    "Show FPS, throughput and latency while streaming (3-finger tap toggles it live)",
+                    style = MaterialTheme.typography.bodySmall,
+                )
+            }
+            Switch(
+                checked = s.statsHudEnabled,
+                onCheckedChange = { on -> update(s.copy(statsHudEnabled = on)) },
+            )
+        }
     }
 }
 

clients/android/app/src/main/kotlin/io/unom/punktfunk/StreamScreen.kt

@@ -6,17 +6,31 @@ import android.view.SurfaceHolder
 import android.view.SurfaceView
 import android.view.WindowManager
 import androidx.activity.compose.BackHandler
+import androidx.compose.foundation.background
 import androidx.compose.foundation.gestures.awaitEachGesture
 import androidx.compose.foundation.gestures.awaitFirstDown
 import androidx.compose.foundation.layout.Box
+import androidx.compose.foundation.layout.Column
 import androidx.compose.foundation.layout.fillMaxSize
+import androidx.compose.foundation.layout.padding
+import androidx.compose.foundation.shape.RoundedCornerShape
+import androidx.compose.material3.Text
 import androidx.compose.runtime.Composable
 import androidx.compose.runtime.DisposableEffect
+import androidx.compose.runtime.LaunchedEffect
+import androidx.compose.runtime.getValue
+import androidx.compose.runtime.mutableStateOf
 import androidx.compose.runtime.remember
+import androidx.compose.runtime.setValue
+import androidx.compose.ui.Alignment
 import androidx.compose.ui.Modifier
+import androidx.compose.ui.graphics.Color
 import androidx.compose.ui.input.pointer.pointerInput
 import androidx.compose.ui.input.pointer.positionChange
 import androidx.compose.ui.platform.LocalContext
+import androidx.compose.ui.text.font.FontFamily
+import androidx.compose.ui.unit.dp
+import androidx.compose.ui.unit.sp
 import androidx.compose.ui.viewinterop.AndroidView
 import androidx.core.content.ContextCompat
 import androidx.core.view.WindowCompat
@@ -25,7 +39,9 @@ import androidx.core.view.WindowInsetsControllerCompat
 import io.unom.punktfunk.kit.Gamepad
 import io.unom.punktfunk.kit.GamepadFeedback
 import io.unom.punktfunk.kit.NativeBridge
+import kotlinx.coroutines.delay
 import kotlin.math.abs
+import kotlin.math.roundToInt
 
 @Composable
 fun StreamScreen(handle: Long, micEnabled: Boolean, onDisconnect: () -> Unit) {
@@ -42,6 +58,18 @@ fun StreamScreen(handle: Long, micEnabled: Boolean, onDisconnect: () -> Unit) {
         Manifest.permission.RECORD_AUDIO,
     ) == PackageManager.PERMISSION_GRANTED
 
+    // Live decode stats for the HUD. Poll once a second for the whole stream (cheap, and each call
+    // drains+resets the native window so it never grows unbounded even while the overlay is hidden);
+    // `showStats` only gates rendering. A 3-finger tap toggles it live; the default comes from Settings.
+    var stats by remember { mutableStateOf<DoubleArray?>(null) }
+    var showStats by remember { mutableStateOf(SettingsStore(context).load().statsHudEnabled) }
+    LaunchedEffect(handle) {
+        while (true) {
+            delay(1000)
+            stats = NativeBridge.nativeVideoStats(handle)
+        }
+    }
+
     DisposableEffect(handle) {
         window?.addFlags(WindowManager.LayoutParams.FLAG_KEEP_SCREEN_ON)
         controller?.let {
@@ -92,8 +120,14 @@ fun StreamScreen(handle: Long, micEnabled: Boolean, onDisconnect: () -> Unit) {
                 }
             },
         )
+        // Live stats HUD (FPS / throughput / capture→client latency), drawn over the video but
+        // BEFORE the transparent gesture layer below, so it shows through and never eats touches.
+        if (showStats) {
+            stats?.let { StatsOverlay(it, Modifier.align(Alignment.TopStart).padding(12.dp)) }
+        }
         // Touch virtual-trackpad overlay: 1-finger drag → relative mouse move; tap → left click;
-        // 2-finger drag → scroll. (Physical-mouse pointer capture comes in a later increment.)
+        // 2-finger drag → scroll; 3-finger tap → toggle the stats HUD. (Physical-mouse pointer
+        // capture comes in a later increment.)
         Box(
             Modifier.fillMaxSize().pointerInput(handle) {
                 awaitEachGesture {
@@ -124,9 +158,56 @@ fun StreamScreen(handle: Long, micEnabled: Boolean, onDisconnect: () -> Unit) {
                     if (!moved && maxFingers == 1) {
                         NativeBridge.nativeSendPointerButton(handle, 1, true)
                         NativeBridge.nativeSendPointerButton(handle, 1, false)
+                    } else if (!moved && maxFingers >= 3) {
+                        showStats = !showStats // quick in-stream HUD toggle
                     }
                 }
             },
         )
     }
 }
+
+/**
+ * The live stats overlay — mirrors the Apple client's HUD. Reads the 10-double layout from
+ * [NativeBridge.nativeVideoStats]:
+ * `[fps, mbps, latP50Ms, latP95Ms, latValid, skew, w, h, hz, dropped]`.
+ */
+@Composable
+private fun StatsOverlay(s: DoubleArray, modifier: Modifier = Modifier) {
+    if (s.size < 10) return
+    val w = s[6].toInt()
+    val h = s[7].toInt()
+    val hz = s[8].toInt()
+    val latValid = s[4] != 0.0
+    val skew = s[5] != 0.0
+    val dropped = s[9].toLong()
+    Column(
+        modifier = modifier
+            .background(Color.Black.copy(alpha = 0.45f), RoundedCornerShape(6.dp))
+            .padding(horizontal = 8.dp, vertical = 4.dp),
+    ) {
+        Text(
+            "$w×$h@$hz   ${s[0].roundToInt()} fps   ${"%.1f".format(s[1])} Mb/s",
+            color = Color.White,
+            fontFamily = FontFamily.Monospace,
+            fontSize = 12.sp,
+        )
+        if (latValid) {
+            val tag = if (skew) "" else " (same-host)"
+            Text(
+                "capture→client ${"%.1f".format(s[2])}/${"%.1f".format(s[3])} ms p50/p95$tag",
+                color = Color.White,
+                fontFamily = FontFamily.Monospace,
+                fontSize = 12.sp,
+            )
+        }
+        if (dropped > 0) {
+            Text(
+                "dropped $dropped",
+                color = Color(0xFFFFB0B0),
+                fontFamily = FontFamily.Monospace,
+                fontSize = 12.sp,
+            )
+        }
+    }
+}

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/NativeBridge.kt

@@ -75,6 +75,14 @@ object NativeBridge {
     /** Stop + join the decode thread without closing the session. No-op on `0`. */
     external fun nativeStopVideo(handle: Long)
 
+    /**
+     * Drain ~1 s of live decode stats for the on-stream HUD, or `null` when no decode thread runs.
+     * Returns 10 doubles:
+     * `[fps, mbps, latP50Ms, latP95Ms, latValid, skewCorrected, width, height, refreshHz, framesDropped]`
+     * (the two flags are 1.0/0.0). Poll ~1 Hz; each call resets the measurement window.
+     */
+    external fun nativeVideoStats(handle: Long): DoubleArray?
+
     /**
      * Start host→client audio: Opus decode → jitter ring → AAudio (LowLatency), all in Rust. No-op
      * if already started. Best-effort — a failure leaves video streaming.

clients/android/native/Cargo.toml

@@ -29,6 +29,8 @@ android_logger = "0.14"
 # Pure-Rust FFI to libmediandk/libnativewindow/libaaudio — no C++/libc++_shared to bundle. Decode +
 # audio run entirely in Rust on native threads (the "no async on the hot path" invariant).
 ndk = { version = "0.9", features = ["media", "audio"] }
+# setpriority/gettid to raise the decode thread toward URGENT_DISPLAY (see decode::boost_thread_priority).
+libc = "0.2"
 # Opus decode for the host→client audio plane (0xC9: 48 kHz stereo, 5 ms frames). Same crate the
 # host + Linux client use. audiopus_sys vendors libopus (pure C) and builds it static via cmake —
 # the cargo-ndk build sets LIBOPUS_STATIC=1/LIBOPUS_NO_PKG=1 so it links the bundled lib, not the host's.

clients/android/native/src/decode.rs

@@ -10,7 +10,7 @@ use ndk::media::media_codec::{
     DequeuedInputBufferResult, DequeuedOutputBufferInfoResult, MediaCodec, MediaCodecDirection,
 };
 use ndk::media::media_format::MediaFormat;
-use ndk::native_window::NativeWindow;
+use ndk::native_window::{FrameRateCompatibility, NativeWindow};
 use punktfunk_core::client::NativeClient;
 use punktfunk_core::error::PunktfunkError;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -18,7 +18,13 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 /// The decode loop. Runs on the `pf-decode` thread until `shutdown` is set or the session closes.
-pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<AtomicBool>) {
+pub fn run(
+    client: Arc<NativeClient>,
+    window: NativeWindow,
+    shutdown: Arc<AtomicBool>,
+    stats: Arc<crate::stats::VideoStats>,
+) {
+    boost_thread_priority();
     let mode = client.mode();
     let codec = match MediaCodec::from_decoder_type("video/hevc") {
         Some(c) => c,
@@ -39,6 +45,11 @@ pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<Atomic
     );
     // Ask for the low-latency decode path where the decoder supports it (no reordering buffer).
     format.set_i32("low-latency", 1);
+    // Advisory low-latency hints (KEY_PRIORITY / KEY_OPERATING_RATE), ignored where unsupported:
+    // realtime priority + the target frame rate, so vendor decoders (e.g. Qualcomm) run at full
+    // clocks instead of a power-saving cadence that adds dequeue latency.
+    format.set_i32("priority", 0); // 0 = realtime
+    format.set_i32("operating-rate", mode.refresh_hz as i32);
 
     if let Err(e) = codec.configure(&format, Some(&window), MediaCodecDirection::Decoder) {
         log::error!("decode: configure failed: {e}");
@@ -53,6 +64,15 @@ pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<Atomic
         mode.width,
         mode.height
     );
+    // Tell the display the stream's refresh so Android can pick a matching display mode and align
+    // vsync (no 60-in-120 judder on high-refresh panels). minSdk 31 ≥ API 30, so the underlying
+    // ANativeWindow_setFrameRate is always present; non-fatal if the platform declines.
+    if let Err(e) = window.set_frame_rate(mode.refresh_hz as f32, FrameRateCompatibility::Default) {
+        log::warn!(
+            "decode: set_frame_rate({} Hz) failed (non-fatal): {e}",
+            mode.refresh_hz
+        );
+    }
 
     let mut fed: u64 = 0;
     let mut rendered: u64 = 0;
@@ -60,6 +80,9 @@ pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<Atomic
     // climbs.
     let mut last_dropped = client.frames_dropped();
     let mut last_kf_req: Option<Instant> = None;
+    // Capture→client-receipt latency uses the negotiated host-minus-client clock offset (0 if the
+    // host didn't answer the skew handshake — then the HUD flags it "same-host").
+    let clock_offset = client.clock_offset_ns;
     while !shutdown.load(Ordering::Relaxed) {
         match client.next_frame(Duration::from_millis(5)) {
             Ok(frame) => {
@@ -72,6 +95,11 @@ pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<Atomic
                     );
                 }
                 fed += 1;
+                // HUD stat: capture→client-receipt latency = client_now + (host−client) − capture_pts.
+                let lat_ns = now_realtime_ns() + clock_offset as i128 - frame.pts_ns as i128;
+                let lat_us =
+                    (lat_ns > 0 && lat_ns < 10_000_000_000).then_some((lat_ns / 1000) as u64);
+                stats.note(frame.data.len(), lat_us, clock_offset != 0);
                 feed(&codec, &frame.data, frame.pts_ns / 1000);
             }
             Err(PunktfunkError::NoFrame) => {} // timeout — still drain output below
@@ -105,6 +133,33 @@ pub fn run(client: Arc<NativeClient>, window: NativeWindow, shutdown: Arc<Atomic
     log::info!("decode: stopped (fed={fed} rendered={rendered})");
 }
 
+/// Wall-clock now in nanoseconds (CLOCK_REALTIME basis), to compare against the host-stamped
+/// capture `pts_ns` after the skew offset is applied.
+fn now_realtime_ns() -> i128 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos() as i128)
+        .unwrap_or(0)
+}
+
+/// Best-effort: raise the decode thread toward Android's URGENT_DISPLAY band so background work
+/// can't preempt it under load (which shows up as late/dropped frames). Non-fatal if the platform
+/// refuses (foreground apps may set their own threads; the exact floor is policy-dependent).
+fn boost_thread_priority() {
+    // SAFETY: `gettid`/`setpriority` on the calling thread are always-safe syscalls. PRIO_PROCESS
+    // with a TID targets that one task on Linux — the same idiom `Process.setThreadPriority` uses.
+    unsafe {
+        let tid = libc::gettid();
+        if libc::setpriority(libc::PRIO_PROCESS, tid as libc::id_t, -10) != 0 {
+            log::warn!(
+                "decode: setpriority(-10) failed (non-fatal): {}",
+                std::io::Error::last_os_error()
+            );
+        }
+    }
+}
+
 /// Copy one access unit into a codec input buffer and queue it.
 fn feed(codec: &MediaCodec, au: &[u8], pts_us: u64) {
     match codec.dequeue_input_buffer(Duration::from_millis(10)) {

clients/android/native/src/lib.rs

@@ -29,6 +29,7 @@ mod feedback;
 #[cfg(target_os = "android")]
 mod mic;
 mod session;
+mod stats;
 
 /// Initialize `android_logger` once when the JVM loads the library. Logs land in logcat under the
 /// `punktfunk` tag. Android-only — there is no JVM (and no logcat) on the host build.

clients/android/native/src/session.rs

@@ -14,7 +14,7 @@
 //! renegotiation. Port the remaining orchestration from `clients/linux`.
 
 use jni::objects::{JObject, JString};
-use jni::sys::{jboolean, jint, jlong};
+use jni::sys::{jboolean, jdoubleArray, jint, jlong, jsize};
 use jni::JNIEnv;
 use punktfunk_core::client::NativeClient;
 use punktfunk_core::config::{CompositorPref, GamepadPref, Mode};
@@ -40,6 +40,8 @@ pub(crate) struct SessionHandle {
 struct VideoThread {
     shutdown: Arc<AtomicBool>,
     join: Option<JoinHandle<()>>,
+    /// Live decode stats, written by the decode thread and drained ~1 Hz by `nativeVideoStats`.
+    stats: Arc<crate::stats::VideoStats>,
 }
 
 impl SessionHandle {
@@ -331,13 +333,19 @@ pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeStartVideo(
         }
     };
     let shutdown = Arc::new(AtomicBool::new(false));
+    let stats = Arc::new(crate::stats::VideoStats::new());
     let client = h.client.clone();
     let sd = shutdown.clone();
+    let st = stats.clone();
     let join = std::thread::Builder::new()
         .name("pf-decode".into())
-        .spawn(move || crate::decode::run(client, window, sd))
+        .spawn(move || crate::decode::run(client, window, sd, st))
         .ok();
-    *guard = Some(VideoThread { shutdown, join });
+    *guard = Some(VideoThread {
+        shutdown,
+        join,
+        stats,
+    });
 }
 
 /// `NativeBridge.nativeStopVideo(handle)` — stop + join the decode thread (without closing the
@@ -355,6 +363,50 @@ pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeStopVideo(
     }
 }
 
+/// `NativeBridge.nativeVideoStats(handle): DoubleArray?` — drain ~1 s of decode stats for the HUD.
+/// Returns 10 doubles
+/// `[fps, mbps, latP50Ms, latP95Ms, latValid, skewCorrected, width, height, refreshHz, framesDropped]`
+/// (the two flags are 1.0/0.0), or `null` when no decode thread is running. Poll ~1 Hz from the UI;
+/// each call resets the measurement window. Not android-gated — pure `jni` + connector reads, so it
+/// links on the host build too (Kotlin only ever calls it on device).
+#[no_mangle]
+pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeVideoStats(
+    env: JNIEnv,
+    _this: JObject,
+    handle: jlong,
+) -> jdoubleArray {
+    if handle == 0 {
+        return std::ptr::null_mut();
+    }
+    // SAFETY: live handle per the nativeConnect/nativeClose contract.
+    let h = unsafe { &*(handle as *const SessionHandle) };
+    let snap = match h.video.lock().unwrap().as_ref() {
+        Some(vt) => vt.stats.drain(),
+        None => return std::ptr::null_mut(), // not streaming → no stats
+    };
+    let mode = h.client.mode();
+    let buf: [f64; 10] = [
+        snap.fps,
+        snap.mbps,
+        snap.lat_p50_ms,
+        snap.lat_p95_ms,
+        if snap.lat_valid { 1.0 } else { 0.0 },
+        if snap.skew_corrected { 1.0 } else { 0.0 },
+        mode.width as f64,
+        mode.height as f64,
+        mode.refresh_hz as f64,
+        h.client.frames_dropped() as f64,
+    ];
+    let arr = match env.new_double_array(buf.len() as jsize) {
+        Ok(a) => a,
+        Err(_) => return std::ptr::null_mut(),
+    };
+    if env.set_double_array_region(&arr, 0, &buf).is_err() {
+        return std::ptr::null_mut();
+    }
+    arr.into_raw()
+}
+
 /// `NativeBridge.nativeStartAudio(handle)` — start the Opus→AAudio playback thread. No-op if already
 /// started or on a `0` handle. Best-effort: a failure leaves video streaming.
 #[cfg(target_os = "android")]

clients/android/native/src/stats.rs

@@ -0,0 +1,93 @@
+//! Live decode stats for the on-stream HUD (mirrors the Apple client's stats overlay): FPS,
+//! receive throughput, and capture→client-receipt latency (p50/p95). The decode thread is the sole
+//! writer (`note` per access unit); the JNI accessor `nativeVideoStats` drains a snapshot ~1 Hz and
+//! resets the window. Pure `std` so it compiles on the host build too (the decode thread is
+//! android-only, but `VideoThread` holds the shared handle unconditionally).
+
+use std::sync::Mutex;
+use std::time::Instant;
+
+/// Rolling per-window accumulator. Rates are computed over the actual elapsed wall-time at drain
+/// (robust to poll jitter), so a poll that lands at 0.9 s or 1.1 s still reports the right FPS.
+pub struct VideoStats {
+    inner: Mutex<Inner>,
+}
+
+struct Inner {
+    window_start: Instant,
+    frames: u64,
+    bytes: u64,
+    /// capture→client-receipt latency samples for this window, in microseconds.
+    lat_us: Vec<u64>,
+    /// Whether the host answered the clock-skew handshake (latency is cross-machine valid).
+    skew_corrected: bool,
+}
+
+/// A drained, computed view of one window. `lat_valid` is false when no in-range latency sample
+/// landed (then p50/p95 are 0 and the HUD hides the latency line, exactly like the Apple client).
+pub struct Snapshot {
+    pub fps: f64,
+    pub mbps: f64,
+    pub lat_p50_ms: f64,
+    pub lat_p95_ms: f64,
+    pub lat_valid: bool,
+    pub skew_corrected: bool,
+}
+
+impl VideoStats {
+    // `new`/`note` are driven only by the android-only decode thread; `drain` (the JNI accessor) is
+    // ungated, so on the host build these two are unreferenced — that's expected, not dead code.
+    #[cfg_attr(not(target_os = "android"), allow(dead_code))]
+    pub fn new() -> VideoStats {
+        VideoStats {
+            inner: Mutex::new(Inner {
+                window_start: Instant::now(),
+                frames: 0,
+                bytes: 0,
+                lat_us: Vec::with_capacity(256),
+                skew_corrected: false,
+            }),
+        }
+    }
+
+    /// Record one decoded access unit: its wire size and (if in range) its capture→client latency.
+    #[cfg_attr(not(target_os = "android"), allow(dead_code))]
+    pub fn note(&self, bytes: usize, lat_us: Option<u64>, skew_corrected: bool) {
+        let mut g = self.inner.lock().unwrap();
+        g.frames += 1;
+        g.bytes += bytes as u64;
+        g.skew_corrected = skew_corrected;
+        if let Some(l) = lat_us {
+            g.lat_us.push(l);
+        }
+    }
+
+    /// Compute the window's rates + latency percentiles, then reset for the next window.
+    pub fn drain(&self) -> Snapshot {
+        let mut g = self.inner.lock().unwrap();
+        let elapsed = g.window_start.elapsed().as_secs_f64().max(1e-3);
+        let fps = g.frames as f64 / elapsed;
+        let mbps = g.bytes as f64 * 8.0 / 1_000_000.0 / elapsed;
+        let (p50, p95, valid) = if g.lat_us.is_empty() {
+            (0.0, 0.0, false)
+        } else {
+            g.lat_us.sort_unstable();
+            let n = g.lat_us.len();
+            let at = |p: f64| g.lat_us[((n as f64 * p) as usize).min(n - 1)] as f64 / 1000.0;
+            (at(0.50), at(0.95), true)
+        };
+        let skew = g.skew_corrected;
+        g.window_start = Instant::now();
+        g.frames = 0;
+        g.bytes = 0;
+        g.lat_us.clear();
+        Snapshot {
+            fps,
+            mbps,
+            lat_p50_ms: p50,
+            lat_p95_ms: p95,
+            lat_valid: valid,
+            skew_corrected: skew,
+        }
+    }
+}