feat(core+host+client): cursor channel — remote-desktop sweep M2a+M2b

The host cursor stops riding the video and becomes a real OS cursor on
the client (the Parsec/RDP model): pointer feel no longer pays the
capture→encode→network→decode→present round trip.

Wire (M2a):
- Hello grows a client_caps trailing byte (CLIENT_CAP_CURSOR) after the
  fixed display_hdr block — presence disambiguated by remaining length,
  which caps the post-HDR tail at 27 bytes (documented); Welcome answers
  HOST_CAP_CURSOR (capable-and-asked, the 444/clipboard precedent).
- CursorShape (0x50, control stream): serial + dims + hotspot + straight
  RGBA, ≤120px/side so the u16 frame always fits (128² would overshoot);
  client caches by serial — re-showing a known shape costs 14 bytes, not
  a bitmap (RDP pointer-cache for free).
- CursorState (0xD0 datagram): serial + visible/relative_hint flags +
  position, sent once per encode-loop tick — latest-wins, self-healing
  under loss, no refresh timer. relative_hint is reserved for M3.
- Client core: two new planes (control-task + datagram-task arms) →
  next_cursor_shape/next_cursor_state; connect() grows client_caps
  (C ABI passes 0 until the v11 cursor poll fns exist).

Host (M2b, Linux portal only):
- handshake::cursor_forward is THE predicate (client asked ∧ Linux ∧
  compositor ≠ gamescope) — Welcome bit and session wiring both read it.
- SessionPlan.cursor_blend goes false for a forwarding session; the
  encode loop ticks a CursorForwarder every iteration: shape-serial diff
  → control-task bridge (mirrors probe_result), state datagram → conn.
- CursorOverlay/capture CursorState carry the hotspot through
  (nearest-neighbor downscale backstop for XL cursors, unit-tested).

Presenter:
- CursorChannel drains both planes per loop iteration; shapes become
  SDL color cursors (from_surface + hotspot), applied while the desktop
  mouse model is engaged; visibility follows the host; capture/released
  hands back the system cursor. Sessions advertise the cap when they
  START in desktop mode.

Verified on .21: fmt + clippy -D warnings (7 crates) + tests green
(core 218 incl. new wire roundtrips, host 245 incl. e2e + forwarder
downscale tests).

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

crates/punktfunk-host/src/native.rs

@@ -76,6 +76,8 @@ mod handshake;
 /// The mid-stream control task (plan §W1); `serve_session` spawns `control::run` after the
 /// handshake to multiplex renegotiation / speed-test control messages onto the data-plane channels.
 mod control;
+/// Cursor-forward channel (M2): the encode loop's shape/state emission.
+mod cursor_fwd;
 
 /// The capture→encode→send data plane (plan §W1); `serve_session` dispatches the synthetic or
 /// virtual source here (`synthetic_stream` / `virtual_stream`) and hands the latter a
@@ -943,6 +945,15 @@ async fn serve_session(
     // accepted ack as "the active mode is now X" and fixes itself; old clients just log it.
     let (reconfig_result_tx, reconfig_result_rx) =
         tokio::sync::mpsc::unbounded_channel::<Reconfigured>();
+    // Cursor-forward bridge (M2): the encode loop diffs each frame's cursor serial and hands
+    // changed SHAPES here; the control task (the control stream's sole writer) sends them.
+    // Same shape as `probe_result_tx`. Wired even when the channel wasn't negotiated — it
+    // just never fires then.
+    let (cursor_shape_tx, cursor_shape_rx) =
+        tokio::sync::mpsc::unbounded_channel::<punktfunk_core::quic::CursorShape>();
+    // Negotiated cursor forwarding: MUST match the HOST_CAP_CURSOR bit the Welcome advertised
+    // (handshake::cursor_forward is the single predicate both read).
+    let cursor_forward = handshake::cursor_forward(hello.client_caps, compositor);
     // Adaptive FEC: the control task maps each client LossReport to a recovery percent and publishes
     // it here; the data-plane send loop reads + applies it per frame. Disabled (pinned) when
     // PUNKTFUNK_FEC_PCT is set. Seeded with the session's starting FEC so it's a no-op until a report.
@@ -978,6 +989,7 @@ async fn serve_session(
         probe_tx,
         probe_result_rx,
         reconfig_result_rx,
+        cursor_shape_rx,
         clip_enabled,
         clip,
     ));
@@ -1363,6 +1375,8 @@ async fn serve_session(
                         fec_target: fec_target_dp,
                         conn: conn_stream,
                         timing_conn,
+                        cursor_forward,
+                        cursor_shape_tx,
                         probe_seq,
                         streamed_au,
                         stats: stats_dp,
@@ -2029,6 +2043,7 @@ mod tests {
             0,    // video_codecs (HEVC-only)
             0,    // preferred_codec
             None, // display_hdr
+            0,    // client_caps
             None, // launch
             None, // pin (TOFU)
             None, // identity (host doesn't require pairing)
@@ -2199,6 +2214,7 @@ mod tests {
             0,    // video_codecs (0 → HEVC-only)
             0,    // preferred_codec (auto)
             None, // display_hdr
+            0,    // client_caps
             None, // launch
             None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client
             Some((cert, key)),
@@ -2266,6 +2282,7 @@ mod tests {
                 0,    // video_codecs
                 0,    // preferred_codec
                 None, // display_hdr
+                0,    // client_caps
                 None, // launch
                 None,
                 None,
@@ -2295,6 +2312,7 @@ mod tests {
             0,    // video_codecs
             0,    // preferred_codec
             None, // display_hdr
+            0,    // client_caps
             None, // launch
             Some(host_fp),
             Some((cert.clone(), key.clone())),

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

@@ -30,6 +30,7 @@ pub(super) async fn run(
     probe_tx: std::sync::mpsc::Sender<ProbeRequest>,
     mut probe_result_rx: tokio::sync::mpsc::UnboundedReceiver<ProbeResult>,
     mut reconfig_result_rx: tokio::sync::mpsc::UnboundedReceiver<Reconfigured>,
+    mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver<punktfunk_core::quic::CursorShape>,
     clip_enabled: Arc<AtomicBool>,
     clip: pf_clipboard::ClipCoord,
 ) {
@@ -262,6 +263,15 @@ pub(super) async fn run(
                     break;
                 }
             }
+            shape = cursor_shape_rx.recv() => {
+                // Cursor-forward bridge (M2): the encode loop diffed a new pointer bitmap.
+                // Rare (shape changes are human-paced); ≤ ~58 KiB fits the u16 frame by
+                // construction (cursor_fwd downscales).
+                let Some(shape) = shape else { break }; // data plane gone
+                if io::write_msg(&mut ctrl_send, &shape.encode()).await.is_err() {
+                    break;
+                }
+            }
             offer = clip_offer_rx.recv(), if !clip_offer_closed => {
                 // Host copied → the coordinator minted a `ClipOffer`; forward it to the client
                 // (only while sync is on — a race with a just-received disable would otherwise

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

@@ -0,0 +1,144 @@
+//! Cursor-forward channel, host side (design/remote-desktop-sweep.md M2).
+//!
+//! When the session negotiated the cursor channel (client `CLIENT_CAP_CURSOR` met our
+//! `HOST_CAP_CURSOR`), the encoder stops blending the pointer into the video
+//! (`SessionPlan::cursor_blend = false`) and the encode loop forwards it out-of-band instead:
+//! the SHAPE (bitmap + hotspot, rare) rides the reliable control stream via the control-task
+//! bridge, per-tick STATE (position/visibility, 14 B) rides a lossy `0xD0` datagram — resent
+//! every iteration so loss self-heals with no refresh timer.
+
+use punktfunk_core::quic::{
+    encode_cursor_state_datagram, CursorShape, CursorState, CURSOR_SHAPE_MAX_SIDE, CURSOR_VISIBLE,
+};
+
+/// Per-session forward state, owned by the encode loop (the thread that binds frames).
+pub(super) struct CursorForwarder {
+    /// Serial of the last shape handed to the control-task bridge (`None` = none yet).
+    sent_serial: Option<u64>,
+    /// Last visible pointer position (hotspot point, frame px) — held across hidden spans so
+    /// a hide still states WHERE the pointer was (the M3 reappear position).
+    last_pos: (i32, i32),
+}
+
+impl CursorForwarder {
+    pub(super) fn new() -> CursorForwarder {
+        CursorForwarder {
+            sent_serial: None,
+            last_pos: (0, 0),
+        }
+    }
+
+    /// Called once per encode-loop iteration with the bound frame's overlay (also on repeat
+    /// iterations — the state datagram is the plane's loss heal, so it goes out every tick).
+    /// `None` overlay = hidden pointer (or no bitmap yet): state only, `visible` clear.
+    pub(super) fn tick(
+        &mut self,
+        cursor: Option<&pf_frame::CursorOverlay>,
+        conn: &quinn::Connection,
+        shape_tx: &tokio::sync::mpsc::UnboundedSender<CursorShape>,
+    ) {
+        let flags = match cursor {
+            Some(ov) => {
+                if self.sent_serial != Some(ov.serial) {
+                    if let Some(shape) = shape_from_overlay(ov) {
+                        // Bridge full ⇒ control task gone ⇒ session is tearing down anyway.
+                        let _ = shape_tx.send(shape);
+                        self.sent_serial = Some(ov.serial);
+                    }
+                }
+                self.last_pos = (ov.x + ov.hot_x as i32, ov.y + ov.hot_y as i32);
+                CURSOR_VISIBLE
+            }
+            None => 0,
+        };
+        let state = CursorState {
+            serial: self.sent_serial.unwrap_or(0) as u32,
+            flags,
+            x: self.last_pos.0,
+            y: self.last_pos.1,
+        };
+        let _ = conn.send_datagram(encode_cursor_state_datagram(&state).into());
+    }
+}
+
+/// Build the wire shape from a capture overlay, integer-downscaling (nearest-neighbor) anything
+/// over [`CURSOR_SHAPE_MAX_SIDE`] so the message always fits the u16-length control frame.
+/// Real cursors are far under the cap — the scale path is a correctness backstop for XL
+/// accessibility cursors, not a quality path. `None` on a malformed overlay (short buffer).
+fn shape_from_overlay(ov: &pf_frame::CursorOverlay) -> Option<CursorShape> {
+    let px = (ov.w as usize).checked_mul(ov.h as usize)?.checked_mul(4)?;
+    if ov.w == 0 || ov.h == 0 || ov.rgba.len() < px {
+        return None;
+    }
+    let max = CURSOR_SHAPE_MAX_SIDE as u32;
+    let f = ov.w.max(ov.h).div_ceil(max).max(1);
+    let (w, h) = (ov.w.div_ceil(f), ov.h.div_ceil(f));
+    let rgba = if f == 1 {
+        ov.rgba.as_ref().clone()
+    } else {
+        let mut out = Vec::with_capacity((w * h * 4) as usize);
+        for y in 0..h {
+            for x in 0..w {
+                let (sx, sy) = ((x * f).min(ov.w - 1), (y * f).min(ov.h - 1));
+                let o = ((sy * ov.w + sx) * 4) as usize;
+                out.extend_from_slice(&ov.rgba[o..o + 4]);
+            }
+        }
+        out
+    };
+    Some(CursorShape {
+        serial: ov.serial as u32,
+        w: w as u16,
+        h: h as u16,
+        hot_x: (ov.hot_x / f).min(w - 1) as u16,
+        hot_y: (ov.hot_y / f).min(h - 1) as u16,
+        rgba,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+
+    fn overlay(w: u32, h: u32, hot: (u32, u32)) -> pf_frame::CursorOverlay {
+        pf_frame::CursorOverlay {
+            x: 10,
+            y: 20,
+            w,
+            h,
+            rgba: Arc::new((0..w * h * 4).map(|i| i as u8).collect()),
+            serial: 3,
+            hot_x: hot.0,
+            hot_y: hot.1,
+        }
+    }
+
+    #[test]
+    fn small_shape_passes_through() {
+        let s = shape_from_overlay(&overlay(32, 32, (4, 5))).unwrap();
+        assert_eq!((s.w, s.h, s.hot_x, s.hot_y, s.serial), (32, 32, 4, 5, 3));
+        assert_eq!(s.rgba.len(), 32 * 32 * 4);
+        // Encodes within the u16 control-frame cap.
+        assert!(s.encode().len() <= u16::MAX as usize);
+    }
+
+    #[test]
+    fn oversize_shape_downscales_with_hotspot() {
+        // 256² → f = ceil(256/120) = 3 → 86² (256.div_ceil(3)), hotspot scales with it.
+        let s = shape_from_overlay(&overlay(256, 256, (255, 0))).unwrap();
+        assert!(s.w <= CURSOR_SHAPE_MAX_SIDE && s.h <= CURSOR_SHAPE_MAX_SIDE);
+        assert_eq!(s.rgba.len(), s.w as usize * s.h as usize * 4);
+        assert!(s.hot_x < s.w && s.hot_y < s.h);
+        assert!(s.encode().len() <= u16::MAX as usize);
+        // The scaled message must decode (dims within the cap).
+        assert_eq!(CursorShape::decode(&s.encode()).unwrap(), s);
+    }
+
+    #[test]
+    fn short_buffer_rejected() {
+        let mut ov = overlay(8, 8, (0, 0));
+        ov.rgba = Arc::new(vec![0; 8]);
+        assert!(shape_from_overlay(&ov).is_none());
+    }
+}

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

@@ -8,6 +8,22 @@
 
 use super::*;
 
+/// Whether this session forwards the cursor out-of-band (design/remote-desktop-sweep.md M2):
+/// the client asked ([`CLIENT_CAP_CURSOR`](punktfunk_core::quic::CLIENT_CAP_CURSOR)) AND the
+/// capture path can deliver cursor metadata separately from the frame — today that is the
+/// Linux portal `SPA_META_Cursor` path only: not gamescope (its capture paints no cursor at
+/// all), not Windows (DWM composites into the IDD frame — M2c). THE single predicate: the
+/// Welcome's `HOST_CAP_CURSOR` bit and the session's forwarding/blend-off wiring both read it,
+/// so they can never disagree.
+pub(super) fn cursor_forward(
+    client_caps: u8,
+    compositor: Option<crate::vdisplay::Compositor>,
+) -> bool {
+    cfg!(target_os = "linux")
+        && client_caps & punktfunk_core::quic::CLIENT_CAP_CURSOR != 0
+        && compositor.is_some_and(|c| c != crate::vdisplay::Compositor::Gamescope)
+}
+
 /// Run the Hello→Welcome→Start negotiation. Borrows the control streams (the caller keeps them for
 /// mid-stream renegotiation afterwards). `first` is the already-read first control message.
 #[allow(clippy::type_complexity, clippy::too_many_arguments)]
@@ -444,6 +460,14 @@ pub(super) async fn negotiate(
                 punktfunk_core::quic::HOST_CAP_CLIPBOARD
             } else {
                 0
+            }
+            // Cursor channel granted (client asked + this capture path can deliver cursor
+            // metadata out of the frame) — the client turns its local renderer on ONLY when
+            // it sees this bit, and serve_session wires forwarding from the same predicate.
+            | if cursor_forward(hello.client_caps, compositor) {
+                punktfunk_core::quic::HOST_CAP_CURSOR
+            } else {
+                0
             },
         // The negotiated session AEAD (resolved above) + its 32-byte key toward a ChaCha
         // client; toward everyone else cipher 0 keeps the Welcome byte-identical to the

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

@@ -938,6 +938,14 @@ pub(super) struct SessionContext {
     /// thread emits one 0xCF datagram per AU (capture→sent µs) on it, so the client can split its
     /// `host+network` latency stage. `None` = older client, no emission.
     pub(super) timing_conn: Option<quinn::Connection>,
+    /// The session negotiated the cursor channel (design/remote-desktop-sweep.md M2 —
+    /// `handshake::cursor_forward`): the encoder does NOT blend the pointer into the video;
+    /// the encode loop forwards shape (via `cursor_shape_tx`) + per-tick `0xD0` state instead.
+    pub(super) cursor_forward: bool,
+    /// SHAPE bridge to the control task (the control stream's sole writer) — mirrors
+    /// `probe_result_tx`. Inert when `cursor_forward` is false.
+    pub(super) cursor_shape_tx:
+        tokio::sync::mpsc::UnboundedSender<punktfunk_core::quic::CursorShape>,
     /// The client advertised [`punktfunk_core::quic::VIDEO_CAP_PROBE_SEQ`]: speed-test bursts may
     /// run mid-session in the probe index space (its reassembler keeps a separate probe window).
     /// `false` = older client whose single-window reassembler would drop probe-space frames as
@@ -987,7 +995,10 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         ctx.bit_depth,
         ctx.chroma,
         ctx.codec,
-        ctx.compositor != pf_vdisplay::Compositor::Gamescope,
+        // Blend the pointer into the video only where the capture HAS one (not gamescope) AND
+        // the client is not drawing it locally (the M2 cursor channel — blending too would
+        // show it twice).
+        ctx.compositor != pf_vdisplay::Compositor::Gamescope && !ctx.cursor_forward,
     );
     // PyroWave rides the datagram-aligned wire mode (§4.4): every encoder this session opens
     // packetizes at the negotiated shard payload, so a lost datagram costs blocks, not frames.
@@ -1021,6 +1032,8 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         fec_target,
         conn,
         timing_conn,
+        cursor_forward,
+        cursor_shape_tx,
         probe_seq,
         streamed_au,
         stats,
@@ -1034,6 +1047,13 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
     // reverts to whole-AU sends without touching the encoder's slicing knobs). The third gate —
     // whether the ENCODER actually chunks — is dynamic (`supports_chunked_poll`, per AU).
     let streamed_wire = streamed_au && std::env::var("PUNKTFUNK_STREAMED_AU").as_deref() != Ok("0");
+    // Cursor-forward state (M2): shape-serial diffing + the per-tick 0xD0 state send. The
+    // encoder was told not to blend (SessionPlan above), so from the first frame the client's
+    // locally-drawn cursor is the only one.
+    let mut cursor_fwd = cursor_forward.then(super::cursor_fwd::CursorForwarder::new);
+    if cursor_forward {
+        tracing::info!("cursor channel negotiated — forwarding shape/state, encoder blend off");
+    }
     if streamed_wire {
         tracing::info!(
             "client accepts streamed AUs (VIDEO_CAP_STREAMED_AU) — chunked encoder output \
@@ -1985,6 +2005,12 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                 );
             }
         }
+        // Cursor channel (M2): every iteration — new frame OR repeat — states the pointer
+        // (self-healing under datagram loss) and forwards a changed shape via the control
+        // bridge. `frame` is the newest bound frame either way.
+        if let Some(fwd) = cursor_fwd.as_mut() {
+            fwd.tick(frame.cursor.as_ref(), &conn, &cursor_shape_tx);
+        }
         if perf && diag_at.elapsed() >= std::time::Duration::from_secs(2) {
             let secs = diag_at.elapsed().as_secs_f64();
             tracing::info!(