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>
2026-07-22 23:43:23 +02:00
parent 97d20e1f0a
commit 4a01bc4463
29 changed files with 894 additions and 13 deletions
@@ -42,8 +42,8 @@ pub use self::rumble::{ActuatorQuirks, RumbleCommand};
 use self::control::{CtrlRequest, Negotiated};
 use self::frame_channel::{DecodeLatAcc, FrameChannel, FramePop};
 use self::planes::{
-    RumbleUpdate, AUDIO_QUEUE, CLIP_EVENT_QUEUE, HDR_META_QUEUE, HIDOUT_QUEUE, HOST_TIMING_QUEUE,
-    RUMBLE_QUEUE,
+    RumbleUpdate, AUDIO_QUEUE, CLIP_EVENT_QUEUE, CURSOR_SHAPE_QUEUE, CURSOR_STATE_QUEUE,
+    HDR_META_QUEUE, HIDOUT_QUEUE, HOST_TIMING_QUEUE, RUMBLE_QUEUE,
 };
 use self::probe::ProbeState;
 use self::pump::run_pump;
@@ -93,6 +93,12 @@ pub struct NativeClient {
    /// Inbound per-AU host capture→send timings — 0xCF datagrams (the client always advertises
    /// [`quic::VIDEO_CAP_HOST_TIMING`]; an older host simply never sends any).
    host_timing: Mutex<Receiver<crate::quic::HostTiming>>,
+    /// Inbound cursor shapes (control-stream [`crate::quic::CursorShape`]) — only a session
+    /// that advertised [`quic::CLIENT_CAP_CURSOR`] against a [`quic::HOST_CAP_CURSOR`] host
+    /// ever receives any.
+    cursor_shape: Mutex<Receiver<crate::quic::CursorShape>>,
+    /// Inbound per-frame cursor state — `0xD0` datagrams (same negotiation gate as shapes).
+    cursor_state: Mutex<Receiver<crate::quic::CursorState>>,
    input_tx: tokio::sync::mpsc::UnboundedSender<InputEvent>,
    /// Outbound mic frames `(seq, pts_ns, opus)` → encoded as 0xCB datagrams by the worker.
    /// Bounded ([`MIC_QUEUE`]): a wedged worker drops fresh frames (logged) instead of queueing
@@ -316,6 +322,12 @@ impl NativeClient {
        // display's EDID so host apps tone-map to the client's real panel; `None` = unknown/SDR
        // (the host keeps its built-in EDID defaults). See [`crate::quic::Hello::display_hdr`].
        display_hdr: Option<HdrMeta>,
+        // Non-video client capabilities ([`crate::quic::Hello::client_caps`]) — set
+        // [`crate::quic::CLIENT_CAP_CURSOR`] ONLY if this embedder actually renders the host
+        // cursor locally (shape + state planes): the host stops compositing the pointer into
+        // the video for a session that advertises it, so a non-rendering embedder that sets it
+        // streams with NO visible cursor at all. `0` = today's composited behavior.
+        client_caps: u8,
        launch: Option<String>,
        pin: Option<[u8; 32]>,
        identity: Option<(String, String)>,
@@ -337,6 +349,10 @@ impl NativeClient {
        let (clip_event_tx, clip_event_rx) =
            std::sync::mpsc::sync_channel::<ClipEventCore>(CLIP_EVENT_QUEUE);
        let (clip_cmd_tx, clip_cmd_rx) = tokio::sync::mpsc::unbounded_channel::<ClipCommand>();
+        let (cursor_shape_tx, cursor_shape_rx) =
+            std::sync::mpsc::sync_channel::<crate::quic::CursorShape>(CURSOR_SHAPE_QUEUE);
+        let (cursor_state_tx, cursor_state_rx) =
+            std::sync::mpsc::sync_channel::<crate::quic::CursorState>(CURSOR_STATE_QUEUE);
        let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<Negotiated>>();
        let shutdown = Arc::new(AtomicBool::new(false));
        let quit = Arc::new(AtomicBool::new(false));
@@ -390,6 +406,7 @@ impl NativeClient {
                    video_codecs,
                    preferred_codec,
                    display_hdr,
+                    client_caps,
                    launch,
                    pin,
                    identity,
@@ -401,6 +418,8 @@ impl NativeClient {
                    hidout_tx,
                    hdr_meta_tx,
                    host_timing_tx,
+                    cursor_shape_tx,
+                    cursor_state_tx,
                    input_rx,
                    mic_rx,
                    rich_input_rx,
@@ -445,6 +464,8 @@ impl NativeClient {
            hidout: Mutex::new(hidout_rx),
            hdr_meta: Mutex::new(hdr_meta_rx),
            host_timing: Mutex::new(host_timing_rx),
+            cursor_shape: Mutex::new(cursor_shape_rx),
+            cursor_state: Mutex::new(cursor_state_rx),
            input_tx,
            mic_tx,
            rich_input_tx,
@@ -892,6 +913,32 @@ impl NativeClient {
        }
    }

+    /// Pull the next host cursor shape (design/remote-desktop-sweep.md M2): RGBA bitmap +
+    /// hotspot, sent on pointer-bitmap change over the reliable control stream. The embedder
+    /// caches by `serial` and builds an OS cursor from it; [`NativeClient::next_cursor_state`]
+    /// references shapes by serial. Only a session that advertised
+    /// [`crate::quic::CLIENT_CAP_CURSOR`] against a capable host receives any. Same
+    /// timeout/closed semantics as [`NativeClient::next_hidout`].
+    pub fn next_cursor_shape(&self, timeout: Duration) -> Result<crate::quic::CursorShape> {
+        match self.cursor_shape.lock().unwrap().recv_timeout(timeout) {
+            Ok(s) => Ok(s),
+            Err(RecvTimeoutError::Timeout) => Err(PunktfunkError::NoFrame),
+            Err(RecvTimeoutError::Disconnected) => Err(PunktfunkError::Closed),
+        }
+    }
+
+    /// Pull the next per-frame cursor state (`0xD0`): position, visibility and the M3
+    /// relative-mode hint, referencing a shape by serial. Latest-wins — an embedder should
+    /// drain the queue and apply only the newest. Same negotiation gate and timeout/closed
+    /// semantics as [`NativeClient::next_cursor_shape`].
+    pub fn next_cursor_state(&self, timeout: Duration) -> Result<crate::quic::CursorState> {
+        match self.cursor_state.lock().unwrap().recv_timeout(timeout) {
+            Ok(s) => Ok(s),
+            Err(RecvTimeoutError::Timeout) => Err(PunktfunkError::NoFrame),
+            Err(RecvTimeoutError::Disconnected) => Err(PunktfunkError::Closed),
+        }
+    }
+
    /// Pull the next per-AU host timing (0xCF): the host's capture→sent duration for one access
    /// unit, correlated to the AU by `pts_ns`. Feeds the unified stats HUD's `host` / `network`
    /// split (`network = (received + clock_offset − pts) − host_us`); a stats consumer should
@@ -35,6 +35,16 @@ pub(crate) const HOST_TIMING_QUEUE: usize = 512;
 /// a dropped fetch-request makes the serving stream time out and reset cleanly.
 pub(crate) const CLIP_EVENT_QUEUE: usize = 32;

+/// Cursor-shape plane depth (control-stream [`crate::quic::CursorShape`], one per pointer-bitmap
+/// change — human-paced). Overflow drops the newest (try_send); the next shape change or a
+/// serial mismatch against `0xD0` state heals it visually within a shape-change period.
+pub(crate) const CURSOR_SHAPE_QUEUE: usize = 8;
+
+/// Cursor-state plane depth (`0xD0`, one datagram per captured frame). Latest-wins state — the
+/// embedder drains per present; a tiny ring only bridges scheduling jitter. Overflow drops the
+/// newest (try_send), healed by the very next frame's datagram.
+pub(crate) const CURSOR_STATE_QUEUE: usize = 8;
+
 /// One Opus packet from the host's audio datagram stream (48 kHz stereo, 5 ms frames).
 #[derive(Clone, Debug)]
 pub struct AudioPacket {
@@ -51,6 +51,8 @@ pub(super) async fn run_pump(args: WorkerArgs) {
        hidout_tx,
        hdr_meta_tx,
        host_timing_tx,
+        cursor_shape_tx,
+        cursor_state_tx,
        input_rx,
        mut mic_rx,
        mut rich_input_rx,
@@ -123,6 +125,7 @@ pub(super) async fn run_pump(args: WorkerArgs) {
            clock_offset: clock_offset.clone(),
            clock_gen: clock_gen.clone(),
            clip_event_tx: clip_event_tx.clone(),
+            cursor_shape_tx,
        }
        .run(),
    );
@@ -136,6 +139,7 @@ pub(super) async fn run_pump(args: WorkerArgs) {
        hidout_tx,
        hdr_meta_tx,
        host_timing_tx,
+        cursor_state_tx,
    ));

    // Clipboard task: the fetch-stream accept loop (host pulls what we offered) + outbound fetches
@@ -21,6 +21,9 @@ pub(super) struct ControlTask {
    /// Clipboard metadata events (ClipState/ClipOffer) feed the same event plane the
    /// clipboard task uses for fetch data.
    pub(super) clip_event_tx: std::sync::mpsc::SyncSender<ClipEventCore>,
+    /// Host cursor shapes ([`CursorShape`], sent on pointer-bitmap change) → the embedder's
+    /// shape plane ([`NativeClient::next_cursor_shape`]).
+    pub(super) cursor_shape_tx: std::sync::mpsc::SyncSender<crate::quic::CursorShape>,
 }

 impl ControlTask {
@@ -36,6 +39,7 @@ impl ControlTask {
            clock_offset,
            clock_gen,
            clip_event_tx,
+            cursor_shape_tx,
        } = self;
        // Mid-stream clock re-sync (see [`ClockResync`]): a batch runs every
        // CLOCK_RESYNC_INTERVAL and whenever the pump asks (CtrlRequest::ClockResync after
@@ -167,6 +171,10 @@ impl ControlTask {
                            seq: offer.seq,
                            kinds: offer.kinds,
                        });
+                    } else if let Ok(shape) = crate::quic::CursorShape::decode(&msg) {
+                        // Pointer bitmap changed (cursor channel, only when negotiated). try_send:
+                        // an overflowing ring drops the newest shape — the next change resends.
+                        let _ = cursor_shape_tx.try_send(shape);
                    } else {
                        tracing::warn!(
                            tag = ?msg.first(),
@@ -3,6 +3,9 @@

 use super::*;

+// One parameter per demuxed plane — grouping them into a struct would just move the field
+// list one hop away from the single call site.
+#[allow(clippy::too_many_arguments)]
 pub(super) async fn run(
    conn: quinn::Connection,
    audio_tx: std::sync::mpsc::SyncSender<AudioPacket>,
@@ -11,6 +14,7 @@ pub(super) async fn run(
    hidout_tx: std::sync::mpsc::SyncSender<crate::quic::HidOutput>,
    hdr_meta_tx: std::sync::mpsc::SyncSender<crate::quic::HdrMeta>,
    host_timing_tx: std::sync::mpsc::SyncSender<crate::quic::HostTiming>,
+    cursor_state_tx: std::sync::mpsc::SyncSender<crate::quic::CursorState>,
 ) {
    // Per-pad reorder gate for v2 rumble envelopes (the seq analog of the host's gamepad-state
    // gate): a datagram the network reordered must not roll a stopped motor back on. Legacy v1
@@ -73,6 +77,11 @@ pub(super) async fn run(
                    let _ = host_timing_tx.try_send(t);
                }
            }
+            Some(&crate::quic::CURSOR_STATE_MAGIC) => {
+                if let Some(s) = crate::quic::decode_cursor_state_datagram(&d) {
+                    let _ = cursor_state_tx.try_send(s);
+                }
+            }
            _ => {} // unknown tag — a newer host; ignore
        }
    }
@@ -143,6 +143,10 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result<Handshake
                // The client display's HDR volume → the host's virtual-display EDID (host apps
                // tone-map to the client's real panel). `None` = unknown/SDR.
                display_hdr,
+                // NOT unconditional like HOST_TIMING above: CLIENT_CAP_CURSOR makes the host
+                // stop compositing the pointer, so only an embedder that actually renders the
+                // cursor locally may set it (the embedder decides, we pass through).
+                client_caps: args.client_caps,
            }
            .encode(),
        )
@@ -22,6 +22,7 @@ pub(crate) struct WorkerArgs {
    pub(crate) video_codecs: u8,
    pub(crate) preferred_codec: u8,
    pub(crate) display_hdr: Option<HdrMeta>,
+    pub(crate) client_caps: u8,
    pub(crate) launch: Option<String>,
    pub(crate) pin: Option<[u8; 32]>,
    pub(crate) identity: Option<(String, String)>,
@@ -38,6 +39,8 @@ pub(crate) struct WorkerArgs {
    pub(crate) hidout_tx: SyncSender<HidOutput>,
    pub(crate) hdr_meta_tx: SyncSender<HdrMeta>,
    pub(crate) host_timing_tx: SyncSender<crate::quic::HostTiming>,
+    pub(crate) cursor_shape_tx: SyncSender<crate::quic::CursorShape>,
+    pub(crate) cursor_state_tx: SyncSender<crate::quic::CursorState>,
    pub(crate) input_rx: tokio::sync::mpsc::UnboundedReceiver<InputEvent>,
    pub(crate) mic_rx: tokio::sync::mpsc::Receiver<(u32, u64, Vec<u8>)>,
    pub(crate) rich_input_rx: tokio::sync::mpsc::UnboundedReceiver<RichInput>,