fix(core): jump-to-live survives a mid-session clock step — disarm on no-op flushes

The clock-based jump-to-live detector compares wall-clock receive time against
the CONNECT-TIME skew offset. A wall-clock step on either end (NTP mid-session,
resume-from-sleep correction) shifts every future frame's apparent latency by a
constant: past the 400 ms bound the detector fires forever — one backlog flush +
recovery IDR every 2 s cooldown, and the bitrate controller rides the repeated
"flushed" bad windows down to its floor. A stream that was perfectly live turns
into a periodic quality pulse with no recovery path.

The tell is in the flush itself: a genuine 400 ms backlog is ≥~170 datagrams
even at the 5 Mbps bitrate floor, but a clock-step flush finds nothing to
discard. So: two consecutive clock-triggered flushes that discarded <64
datagrams and zero queued AUs disarm the clock detector for the session (logged).
This also covers upstream router bufferbloat — delay standing in a queue a local
flush can't drain, where the OWD signal to the bitrate controller is the actual
remedy and a 2 s IDR cadence only feeds the congestion. The clock-free
queue-depth detector stays armed either way; it measures the local queue
directly and can't be fooled by a clock.

Rode along: the 11-field `Negotiated` tuple is now a documented struct — the
connect/worker plumbing reads as named fields instead of positional magic.

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

crates/punktfunk-core/src/client.rs

@@ -55,26 +55,32 @@ enum CtrlRequest {
     SetBitrate(u32),
 }
 
-/// What the worker reports to [`NativeClient::connect`] once the handshake lands: the negotiated
+/// What the worker reports to [`NativeClient::connect`] once the handshake lands: the
-/// mode, the host-resolved compositor backend, the host-resolved gamepad backend, the host's
+/// [`Welcome`]-resolved session parameters (mode, backends, encode/colour/audio geometry) plus the
-/// certificate fingerprint, the resolved encoder bitrate (kbps), and the host↔client clock offset
+/// host certificate fingerprint and the connect-time clock offset. Mirrored one-to-one onto the
-/// (ns, host minus client; 0 = no skew correction / an old host that didn't answer the handshake).
+/// public `NativeClient` fields of the same names.
-/// The trailing `u8`s are the resolved encode bit depth (8/10), the chroma `chroma_format_idc`
+#[derive(Clone, Copy)]
-/// (1 = 4:2:0, 3 = 4:4:4), the resolved audio channel count (2/6/8), and the resolved video codec
+struct Negotiated {
-/// (`quic::CODEC_*`), with [`ColorInfo`] the resolved colour signalling — all from the [`Welcome`].
+    mode: Mode,
-type Negotiated = (
+    compositor: CompositorPref,
-    Mode,
+    gamepad: GamepadPref,
-    CompositorPref,
+    /// SHA-256 of the certificate the host actually presented (TOFU callers persist this).
-    GamepadPref,
+    host_fingerprint: [u8; 32],
-    [u8; 32],
+    /// The encoder bitrate the host actually configured (kbps); `0` = an older host.
-    u32,
+    bitrate_kbps: u32,
-    i64,
+    /// Host clock minus client clock (ns); `0` = no skew handshake (old host / synced clocks).
-    u8,
+    clock_offset_ns: i64,
-    ColorInfo,
+    /// Resolved encode bit depth: `8`, or `10` for a Main10 / HDR session.
-    u8,
+    bit_depth: u8,
-    u8,
+    /// Resolved CICP colour signalling.
-    u8,
+    color: ColorInfo,
-);
+    /// Resolved chroma subsampling as the HEVC `chroma_format_idc` (1 = 4:2:0, 3 = 4:4:4).
+    chroma_format: u8,
+    /// Resolved audio channel count (2/6/8) — what the Opus decoders must be built from.
+    audio_channels: u8,
+    /// The single codec the host will emit (`quic::CODEC_*`).
+    codec: u8,
+}
 
 /// Accumulated state of an in-flight / finished speed test. The data-plane pump mirrors the
 /// session's packet-level receive counters here; the control task finalizes the delivered figure
@@ -177,6 +183,22 @@ const FLUSH_AFTER_FRAMES: u32 = 30;
 /// warning instead of a continuous flush/keyframe storm.
 const FLUSH_COOLDOWN: Duration = Duration::from_secs(2);
 
+/// A clock-triggered jump-to-live that discarded fewer datagrams than this (and no queued AUs)
+/// found NO local backlog: the frames read as late, but nothing here was actually behind. Two
+/// causes, and flushing helps neither: a **wall-clock step** (NTP mid-session on either end)
+/// shifted the skew-corrected latency by a constant — every future frame reads over-bound and the
+/// detector would fire forever, one flush + recovery IDR per cooldown, dragging the bitrate
+/// controller to its floor; or the delay is standing in an **upstream queue** (router bufferbloat),
+/// which a local flush can't drain — the OWD signal already feeds the bitrate controller, the
+/// actual remedy. Even at the 5 Mbps bitrate floor a genuine 400 ms backlog is ~170 datagrams, so
+/// 64 cleanly separates "empty" from "real". See `NOOP_CLOCK_FLUSHES_TO_DISARM`.
+const NOOP_FLUSH_DATAGRAMS: u64 = 64;
+
+/// Consecutive no-op clock-triggered flushes (see [`NOOP_FLUSH_DATAGRAMS`]) before the clock-based
+/// detector is disarmed for the rest of the session. The clock-free standing-queue detector stays
+/// armed — it measures the local queue directly and can't be fooled by a clock step.
+const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2;
+
 /// The pre-decode video hand-off from the data-plane pump to the embedder. Unlike the side planes
 /// (self-contained samples that drop the newest on overflow), video AUs are reference-chained under the
 /// host's infinite GOP: dropping ANY frame mid-stream corrupts every dependent frame until the next
@@ -581,19 +603,7 @@ impl NativeClient {
             })
             .map_err(PunktfunkError::Io)?;
 
-        let (
+        let negotiated = match ready_rx.recv_timeout(timeout) {
-            negotiated,
-            resolved_compositor,
-            resolved_gamepad,
-            fingerprint,
-            resolved_bitrate_kbps,
-            clock_offset_ns,
-            bit_depth,
-            color,
-            chroma_format,
-            audio_channels,
-            codec,
-        ) = match ready_rx.recv_timeout(timeout) {
             Ok(Ok(t)) => t,
             Ok(Err(e)) => return Err(e),
             Err(_) => {
@@ -601,7 +611,7 @@ impl NativeClient {
                 return Err(PunktfunkError::Timeout);
             }
         };
-        *mode_slot.lock().unwrap() = negotiated;
+        *mode_slot.lock().unwrap() = negotiated.mode;
         Ok(NativeClient {
             frames: frame_chan,
             audio: Mutex::new(audio_rx),
@@ -621,16 +631,16 @@ impl NativeClient {
             fec_recovered,
             hot_tids,
             mode: mode_slot,
-            host_fingerprint: fingerprint,
+            host_fingerprint: negotiated.host_fingerprint,
-            resolved_compositor,
+            resolved_compositor: negotiated.compositor,
-            resolved_gamepad,
+            resolved_gamepad: negotiated.gamepad,
-            resolved_bitrate_kbps,
+            resolved_bitrate_kbps: negotiated.bitrate_kbps,
-            clock_offset_ns,
+            clock_offset_ns: negotiated.clock_offset_ns,
-            bit_depth,
+            bit_depth: negotiated.bit_depth,
-            color,
+            color: negotiated.color,
-            chroma_format,
+            chroma_format: negotiated.chroma_format,
-            audio_channels,
+            audio_channels: negotiated.audio_channels,
-            codec,
+            codec: negotiated.codec,
         })
     }
 
@@ -1231,58 +1241,35 @@ async fn worker_main(args: WorkerArgs) {
             session,
             send,
             recv,
-            welcome.mode,
+            Negotiated {
-            welcome.compositor,
+                mode: welcome.mode,
-            welcome.gamepad,
+                compositor: welcome.compositor,
-            fingerprint,
+                gamepad: welcome.gamepad,
-            welcome.bitrate_kbps,
+                host_fingerprint: fingerprint,
-            clock_offset_ns,
+                bitrate_kbps: welcome.bitrate_kbps,
-            welcome.bit_depth,
+                clock_offset_ns,
-            welcome.color,
+                bit_depth: welcome.bit_depth,
-            welcome.chroma_format,
+                color: welcome.color,
-            welcome.audio_channels,
+                chroma_format: welcome.chroma_format,
-            welcome.codec,
+                audio_channels: welcome.audio_channels,
+                codec: welcome.codec,
+            },
             welcome.host_caps,
         ))
     };
 
-    let (
+    let (conn, mut session, mut ctrl_send, mut ctrl_recv, negotiated, host_caps) = match setup.await
-        conn,
+    {
-        mut session,
-        mut ctrl_send,
-        mut ctrl_recv,
-        negotiated,
-        resolved_compositor,
-        resolved_gamepad,
-        fingerprint,
-        resolved_bitrate_kbps,
-        clock_offset_ns,
-        bit_depth,
-        color,
-        chroma_format,
-        audio_channels,
-        codec,
-        host_caps,
-    ) = match setup.await {
         Ok(t) => t,
         Err(e) => {
             let _ = ready_tx.send(Err(e));
             return;
         }
     };
-    let _ = ready_tx.send(Ok((
+    // Copies the pump needs after `negotiated` is handed over to `connect`.
-        negotiated,
+    let clock_offset_ns = negotiated.clock_offset_ns;
-        resolved_compositor,
+    let resolved_bitrate_kbps = negotiated.bitrate_kbps;
-        resolved_gamepad,
+    let _ = ready_tx.send(Ok(negotiated));
-        fingerprint,
-        resolved_bitrate_kbps,
-        clock_offset_ns,
-        bit_depth,
-        color,
-        chroma_format,
-        audio_channels,
-        codec,
-    )));
 
     // Input task: embedder events → QUIC datagrams. Toward a host that advertised
     // HOST_CAP_GAMEPAD_STATE, the per-transition gamepad events every embedder still emits are
@@ -1515,6 +1502,11 @@ async fn worker_main(args: WorkerArgs) {
         let mut stale_frames: u32 = 0;
         let mut standing_frames: u32 = 0;
         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
+        // detector off for the session (a clock step / upstream queue it can't fix).
+        let mut noop_clock_flushes: u32 = 0;
+        let mut clock_detector_armed = true;
         while !pump_shutdown.load(Ordering::SeqCst) {
             // Mirror the reassembler's unrecoverable-drop count for the client's keyframe-recovery
             // loop, and (during a speed test) the packet-level receive counters for the throughput
@@ -1605,7 +1597,10 @@ async fn worker_main(args: WorkerArgs) {
                             owd_sum_ns += lat_ns;
                             owd_frames += 1;
                         }
-                        if clock_offset_ns != 0 && lat_ns > FLUSH_LATENCY.as_nanos() as i128 {
+                        if clock_detector_armed
+                            && clock_offset_ns != 0
+                            && lat_ns > FLUSH_LATENCY.as_nanos() as i128
+                        {
                             stale_frames += 1;
                         } else {
                             stale_frames = 0;
@@ -1635,6 +1630,27 @@ async fn worker_main(args: WorkerArgs) {
                                 dropped_frames = dropped,
                                 "receive backlog stopped draining — jumped to live (flush + keyframe)"
                             );
+                            // Clock-detector health check: a clock-only trigger whose flush found
+                            // no local backlog is a false "behind" reading (a wall-clock step, or
+                            // an upstream queue a local flush can't drain) — repeated, it would
+                            // cost a recovery IDR every cooldown forever. Disarm after two in a
+                            // row; the clock-free queue detector keeps covering real backlogs.
+                            if clock_behind && !queue_behind
+                                && flushed < NOOP_FLUSH_DATAGRAMS
+                                && dropped == 0
+                            {
+                                noop_clock_flushes += 1;
+                                if noop_clock_flushes >= NOOP_CLOCK_FLUSHES_TO_DISARM {
+                                    clock_detector_armed = false;
+                                    tracing::warn!(
+                                        "clock-based jump-to-live disarmed — its flushes found no \
+                                         local backlog (clock step or upstream queueing suspected); \
+                                         the queue-depth detector stays armed"
+                                    );
+                                }
+                            } else {
+                                noop_clock_flushes = 0;
+                            }
                             continue; // this frame is part of the stale past — don't render it
                         }
                     }