DataStreamingServer

def __init__(self, supervisor: Optional[Any] = None) -> None:
    super().__init__("websockets")
    self.data_ws: Optional[web.WebSocketResponse] = (
        None
    )
    self.clients: set[web.WebSocketResponse] = set()
    self.app = None
    self.cli_args = settings
    self.is_secure_mode = False
    self.input_handler = None
    self._tasks_to_run = []
    self.RECONNECT_DEBOUNCE_MS = 500
    self.RECONNECT_GRACE_S = 3.0
    self._display_teardown_tasks = set()
    self.MAX_RECENT_CLIENTS = 1000
    self.last_connection_times = OrderedDict()
    self._latest_client_render_fps = 0.0
    self._last_time_client_ok = 0.0
    self._client_acknowledged_frame_id = -1
    self._last_client_acknowledged_frame_id_update_time = 0.0
    self._previous_ack_id_for_stall_check = -1
    self._previous_sent_id_for_stall_check = -1
    self._sent_frames_log = deque()
    self.rc_mode = RateControlMode.CRF
    self.config_gate = asyncio.Event()
    self.shutdown_event = asyncio.Event()
    self._shutdown_called = False
    self.supervisor = supervisor

    def get_initial_value(setting_name: str):
        """Get the correct initial integer/bool from a processed setting."""
        processed_value = getattr(self.cli_args, setting_name)
        setting_def = next((s for s in SETTING_DEFINITIONS if s['name'] == setting_name), None)
        if not setting_def: return None

        if setting_def['type'] == 'range':
            min_val, max_val = processed_value
            return min_val if min_val == max_val else setting_def.get('meta', {}).get('default_value')
        elif setting_def['type'] == 'bool':
            return processed_value[0]
        return processed_value

    self._initial_video_crf = get_initial_value('video_crf')
    self.video_crf = self._initial_video_crf
    self._initial_video_fullcolor = get_initial_value('video_fullcolor')
    self.video_fullcolor = self._initial_video_fullcolor
    self._initial_video_streaming_mode = get_initial_value('video_streaming_mode')
    self.video_streaming_mode = self._initial_video_streaming_mode
    self.capture_cursor = False
    self._initial_jpeg_quality = get_initial_value('jpeg_quality')
    self.jpeg_quality = self._initial_jpeg_quality
    self._initial_paint_over_jpeg_quality = get_initial_value('paint_over_jpeg_quality')
    self.paint_over_jpeg_quality = self._initial_paint_over_jpeg_quality
    self._initial_video_paintover_crf = get_initial_value('video_paintover_crf')
    self.video_paintover_crf = self._initial_video_paintover_crf
    self._initial_video_paintover_burst_frames = get_initial_value('video_paintover_burst_frames')
    self.video_paintover_burst_frames = self._initial_video_paintover_burst_frames
    self._initial_use_cpu = get_initial_value('use_cpu')
    self.use_cpu = self._initial_use_cpu
    self._initial_use_paint_over_quality = get_initial_value('use_paint_over_quality')
    self.use_paint_over_quality = self._initial_use_paint_over_quality
    self._initial_video_bitrate = get_initial_value('video_bitrate')
    self.video_bitrate = self._initial_video_bitrate

    self._resource_monitor: Optional[resource_stats.ResourceMonitor] = None
    self._stats_subscribers: Set[web.WebSocketResponse] = set()
    self._stats_sending: Set[web.WebSocketResponse] = set()
    self._stream_watches: Dict[str, stream_stats.StreamWatch] = {}
    self.uinput_mouse_socket = UINPUT_MOUSE_SOCKET
    self.js_socket_path = settings.js_socket_path
    self.enable_clipboard = settings.enable_clipboard
    self.enable_binary_clipboard = self.cli_args.enable_binary_clipboard[0]
    self.enable_cursors = ENABLE_CURSORS
    self.cursor_size = CURSOR_SIZE
    self.cursor_scale = 1.0
    self.cursor_debug = DEBUG_CURSORS
    self._last_adjustment_timestamp = 0.0
    self.client_settings_received = asyncio.Event()
    self._reconfigure_lock = asyncio.Lock()
    self._video_capture_lock = asyncio.Lock()
    self._is_reconfiguring = False
    self._reconfigure_pending = False
    self.last_start_video_request_times = {}
    self.last_viewer_keyframe_request_times = {}
    self.video_paused_clients = set()
    self._deferred_viewer_rejoins = {}
    self.allowed_desync_ms = BACKPRESSURE_ALLOWED_DESYNC_MS
    self.latency_threshold_for_adjustment_ms = BACKPRESSURE_LATENCY_THRESHOLD_MS
    self.backpressure_check_interval_s = BACKPRESSURE_CHECK_INTERVAL_S
    self.BACKPRESSURE_QUEUE_SIZE = getattr(settings, 'backpressure_queue_size', 120)
    self._last_client_frame_id_report_time = 0.0
    self.capture_loop = None
    # Displays whose live capture has delivered a frame, for the first-frame check.
    self._framed_displays: set = set()

    self.display_clients = {}
    self.video_relay_groups = {}
    self.capture_instances = {}
    self.display_layouts = {}
    self._persistent_capture_modules = {}
    self._wayland_ctl_module = None
    self._host_output_capacity = None
    self._last_keyframe_request = {}
    self._last_keyframe_log = {}
    self._keyframe_log_suppressed = {}
    self._last_lost_frame = {}
    self._invalidation_log = {}

    self.audio_device_name = self.cli_args.audio_device_name
    self.pcmflux_module = None
    self.is_pcmflux_capturing = False
    self.pcmflux_settings = None
    self._pcmflux_reported_failure = None
    self._pcmflux_last_restart = 0.0
    self.audio_redundancy_by_ws = {}
    self.audio_redundancy_enabled = bool(settings.audio_redundancy[0])
    self._active_audio_red_distance = 0
    # The vendored WebRTC RedOpusEncoder reads its depth from audio_config,
    # so one control (0 = plain Opus) drives both transports.
    _red_dist = getattr(settings, "audio_redundancy_distance", AUDIO_RED_DISTANCE)
    audio_config.set_red_distance(
        _red_dist if self.audio_redundancy_enabled else 0
    )
    self.pcmflux_callback = None
    self.pcmflux_audio_queue = None
    self.pcmflux_send_task = None
    self.pcmflux_capture_loop = None

    self._last_display_count = 0
    self._wm_swap = MultiMonitorWindowManager()

def initialize(self) -> None:
    """Create the SelkiesStreamingApp and InputHandler and wire their callbacks.

    Must be called before run(). Also resolves secure vs legacy mode (a set
    master token closes the config gate until tokens are provisioned, and
    governs who holds input authority) and installs the WebRTC-dialect live
    verbs (`_arg_fps`, `vb`, `ab`, `_rc`, `_crf`) so both transports honor
    the same per-key tunables. With metrics enabled, the registry-global
    Prometheus gauges are fed server-side (ACK-derived client fps and
    smoothed RTT from the backpressure loop, GPU from the stats collector)
    and by the shared `_f,`/`_l,` verbs when a client reports directly.
    """
    self.is_secure_mode = bool(self.cli_args.master_token)
    if self.is_secure_mode:
        logger.debug("Secure Mode ENABLED (SELKIES_MASTER_TOKEN is set).")
    else:
        logger.debug("Legacy Mode ENABLED (SELKIES_MASTER_TOKEN is not set).")
        self.config_gate.set()

    global TARGET_FRAMERATE
    processed_framerate = settings.framerate
    min_fr, max_fr = processed_framerate
    if min_fr == max_fr:
        TARGET_FRAMERATE = min_fr
    else:
        fr_def = next((s for s in SETTING_DEFINITIONS if s['name'] == 'framerate'), None)
        TARGET_FRAMERATE = fr_def['meta']['default_value'] if fr_def else 60

    initial_encoder = settings.encoder

    logger.debug(f"Initializing DataStreamingServer with encoder: {initial_encoder}, Framerate: {TARGET_FRAMERATE}")

    event_loop = asyncio.get_running_loop()
    self.app = SelkiesStreamingApp(
        event_loop,
        framerate=TARGET_FRAMERATE,
        encoder=initial_encoder,
        mode="websockets",
    )
    self.app.server_enable_resize = ENABLE_RESIZE
    self.app.last_resize_success = True
    self.app.data_streaming_server = self
    logger.debug(
        f"SelkiesStreamingApp initialized: encoder={self.app.encoder}, display={self.app.display_width}x{self.app.display_height}"
    )

    if settings.enable_rate_control[0]:
        self.rc_mode = RateControlMode(settings.rate_control_mode)

    # The normalized policy string (true/false/in/out); the input handler gates
    # directions off it.
    clipboard_mode = settings.enable_clipboard

    self.input_handler = InputHandler(
        self.app,
        self.uinput_mouse_socket,
        self.js_socket_path,
        clipboard_mode,
        str(settings.enable_binary_clipboard[0]).lower(),
        self.enable_cursors,
        self.cursor_size,
        1.0,
        self.cursor_debug,
        data_server_instance=self,
        is_wayland=IS_WAYLAND,
        wayland_socket_index=settings.wayland_socket_index,
        app_wayland_display=(settings.app_wayland_display
                             or settings.wayland_host_display),
        uinput_gamepad=settings.uinput_gamepad,
    )

    self.input_handler.on_clipboard_read = self.app.send_ws_clipboard_data
    self.input_handler.on_set_fps = self._handle_opcode_fps
    self.input_handler.on_video_encoder_bit_rate = self._handle_opcode_video_bitrate
    self.input_handler.on_audio_encoder_bit_rate = self._handle_opcode_audio_bitrate
    self.input_handler.on_update_rate_control_mode = self._handle_opcode_rate_control
    self.input_handler.on_update_crf = self._handle_opcode_crf
    self.metrics = None
    if settings.enable_metrics_http[0]:
        self.metrics = Metrics()
        self.input_handler.on_client_fps = (
            lambda fps: self.metrics.set_fps(fps) if self.metrics else None
        )
        self.input_handler.on_client_latency = (
            lambda latency: self.metrics.set_latency(latency) if self.metrics else None
        )
    self.input_handler.on_mouse_pointer_visible = self.set_native_cursor_rendering
    self.input_handler.on_session_compositor_adopted = self._resync_wayland_session_scale
    self.input_handler.on_scaling_ratio = self._handle_scaling

    if ENABLE_RESIZE:
        self.input_handler.on_resize = self._handle_resize
    else:
        # Only the resolution is frozen: a DPI sync still scales the desktop.
        self.input_handler.on_resize = lambda res_str, display_id='primary': logger.warning("Resize disabled.")
    logger.debug("DataStreamingServer initialization complete.")

async def set_native_cursor_rendering(self, enabled: bool) -> None:
    """Compose the cursor into the captured video (vs the client-drawn overlay).

    Applies to every display's capture. Reached both from the
    SET_NATIVE_CURSOR_RENDERING message and the shared input protocol's
    pointer-visibility toggle ("p,N"), which map to the same tunable.
    """
    if self.capture_cursor == enabled:
        data_logger.debug(f"Native cursor rendering: value {enabled} is already set.")
        return
    self.capture_cursor = enabled
    if len(self.capture_instances) > 0:
        data_logger.info("Cursor rendering changed, triggering display reconfiguration.")
        await self.reconfigure_displays()

def _opcode_display_module(self, display_id: str) -> Optional[Any]:
    """The display's live ScreenCapture module, or None if not capturing."""
    inst = self.capture_instances.get(display_id)
    return inst.get('module') if inst else None

def _track_capture_settings(self, display_id: str, fresh: Optional[Any] = None,
                            **live_fields: Any) -> None:
    """Record what the display's running capture is actually configured with.

    Pass `fresh` after rebuilding the whole settings object, or individual
    fields after a targeted rate update.

    The tracked object is what _video_relay_budget sizes new relays from and
    what a layout-following reconfigure re-pushes to the module, so a live
    change that skipped it would be applied to the encoder and then silently
    reverted.
    """
    inst = self.capture_instances.get(display_id)
    if inst is None:
        return
    if fresh is not None:
        inst['settings'] = fresh
        return
    cs = inst.get('settings')
    if cs is None:
        return
    for name, value in live_fields.items():
        setattr(cs, name, value)

async def _handle_resize(self, res_str: str, display_id: str = 'primary') -> None:
    """Route a client resize once the display it names has been laid out.

    The layout comes from the connection's initial SETTINGS, which this
    transport's own message loop processes, so the wait is bounded: a
    client that sends `r,` first would otherwise deadlock it.
    """
    try:
        await asyncio.wait_for(self.client_settings_received.wait(), timeout=15.0)
    except asyncio.TimeoutError:
        data_logger.warning("Ignoring resize request received before initial SETTINGS.")
        return
    await on_resize_handler(res_str, self.app, self, display_id)

async def _handle_scaling(self, dpi_value: float, display_id: str = 'primary') -> None:
    """Apply a client DPI sync to the desktop (WebRTC `handle_scaling` parity).

    Fractional DPI is legal on the shared verb; the desktop property itself
    is integral and bounded by the declared span, and an operator-set DPI
    governs the desktop over any client's. X11 takes one DPI for the
    session; on Wayland the display's own screen takes it, while the cursor
    cap and size stay the primary's. The DPI is stored where SETTINGS
    stores it, or a later partial SETTINGS re-applies one the desktop has
    moved off. The wait is the one `_handle_resize` documents.
    """
    try:
        await asyncio.wait_for(self.client_settings_received.wait(), timeout=15.0)
    except asyncio.TimeoutError:
        data_logger.warning("Ignoring DPI sync received before initial SETTINGS.")
        return
    try:
        dpi_value = min(SCALING_DPI_MAX,
                        max(SCALING_DPI_MIN, int(round(float(dpi_value)))))
    except (TypeError, ValueError, OverflowError):
        data_logger.error(f"Invalid DPI value: {dpi_value!r}")
        return
    if app_settings._overridden.get("scaling_dpi", False):
        data_logger.info("Ignoring client DPI sync: scaling_dpi is operator-overridden.")
        return
    data_logger.debug(f"Received DPI setting from client: {dpi_value}")
    try:
        if not IS_WAYLAND:
            if await set_dpi(dpi_value):
                data_logger.info(f"Successfully set DPI to {dpi_value}")
            else:
                data_logger.error(f"Failed to set DPI to {dpi_value}")
            self._update_cursor_cap(dpi_value)
        else:
            # Before the restart, which reads the cap through CaptureSettings.
            if display_id in (None, 'primary'):
                self._update_cursor_cap(dpi_value)
            await self._realize_wayland_display_dpi(display_id, dpi_value)

        dpi_state = self.display_clients.get(display_id)
        if dpi_state is not None:
            dpi_state["scaling_dpi"] = dpi_value

        if CURSOR_SIZE is not None:
            if IS_WAYLAND:
                if display_id in (None, 'primary'):
                    await self._apply_wayland_cursor_size(dpi_value)
            else:
                new_cursor_size = cursor_size_for_dpi(dpi_value, CURSOR_SIZE)
                data_logger.debug(f"Attempting to set cursor size to: {new_cursor_size} (based on DPI {dpi_value})")
                if await set_cursor_size(new_cursor_size):
                    data_logger.debug(f"Successfully set cursor size to {new_cursor_size}")
                else:
                    data_logger.error(f"Failed to set cursor size to {new_cursor_size}")
    except Exception as e_dpi:
        data_logger.error(f"Error applying DPI {dpi_value}: {e_dpi}", exc_info=True)

async def _handle_opcode_fps(self, fps: Any, display_id: str = 'primary') -> None:
    """Live framerate for the shared '_arg_fps' verb (WebRTC-mode parity):
    sanitize against the server range, store, and live-update the display's
    capture; a stopped display applies the new rate at its next START_VIDEO."""
    sanitized = sanitize_client_setting("framerate", fps, self.cli_args, data_logger)
    if sanitized is None:
        return
    if display_id == 'primary':
        # Only the primary controller moves the session default later displays seed from.
        self.app.set_framerate(sanitized)
        data_logger.debug(f"Session default framerate updated to {int(sanitized)} for new displays.")
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["framerate"] = sanitized
    module = self._opcode_display_module(display_id)
    if module is not None:
        try:
            module.update_framerate(float(sanitized))
            self._track_capture_settings(display_id, target_fps=float(sanitized))
            data_logger.info(f"Applied framerate live via '_arg_fps': {sanitized} fps for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live framerate update failed for '{display_id}' ({e}).")

async def _handle_opcode_video_bitrate(self, bitrate: Any, display_id: str = "primary") -> None:
    """Live video bitrate (kbps) for the 'vb' verb, sanitized exactly like
    the SETTINGS path so locked server ranges cannot be bypassed."""
    sanitized = sanitize_client_setting("video_bitrate", bitrate, self.cli_args, data_logger)
    if sanitized is None:
        return
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["video_bitrate"] = sanitized
    if display_id == 'primary':
        self._initial_video_bitrate = sanitized
        data_logger.debug(f"Session default video_bitrate updated to {int(sanitized)} kbps for new displays.")
    module = self._opcode_display_module(display_id)
    if module is not None:
        kbps = int(round(float(sanitized)))
        try:
            module.update_video_bitrate(kbps)
            self._track_capture_settings(display_id, video_bitrate_kbps=kbps)
            data_logger.info(f"Applied video bitrate live via 'vb': {kbps} kbps for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live bitrate update failed for '{display_id}' ({e}).")

async def _handle_opcode_audio_bitrate(self, bitrate: Any) -> None:
    """Live Opus bitrate (bps) for the 'ab' verb; same live-retarget with
    restart fallback as the SETTINGS path."""
    sanitized = sanitize_client_setting("audio_bitrate", bitrate, self.cli_args, data_logger)
    if sanitized is None:
        return
    self.app.audio_bitrate = sanitized
    for display_state in self.display_clients.values():
        display_state["audio_bitrate"] = self.app.audio_bitrate
    if self.is_pcmflux_capturing and self.pcmflux_module:
        try:
            self.pcmflux_module.update_audio_bitrate(int(self.app.audio_bitrate))
            data_logger.info(f"Applied audio bitrate live: {self.app.audio_bitrate} bps")
        except Exception as e:
            data_logger.warning(f"Live audio bitrate update failed ({e}); restarting audio pipeline.")
            # Under the guard like every audio start/stop, and re-checked there:
            # a concurrent guarded op must not orphan a second AudioCapture.
            async with self._reconfigure_guard():
                if self.is_pcmflux_capturing:
                    await self._stop_pcmflux_pipeline()
                    await self._start_pcmflux_pipeline()

async def _handle_opcode_rate_control(self, mode: Any, display_id: str = 'primary') -> None:
    """Rate-control switch for the '_rc' verb: structural like the SETTINGS
    path (the encoder session must be rebuilt), honoring the server's
    enable_rate_control lock and a stopped display's start gating."""
    enable_rate_control, _ = self.cli_args.enable_rate_control
    if not enable_rate_control:
        data_logger.debug("Server has rate control disabled. Ignoring '_rc' change.")
        return
    # Resolved by value: a duplicate enum class from another module must not
    # leave a stray 'RateControlMode.CBR' repr.
    mode_str = (mode.value if hasattr(mode, "value") else str(mode)).split(".")[-1].lower()
    sanitized = sanitize_client_setting("rate_control_mode", mode_str, self.cli_args, data_logger)
    if sanitized not in ("cbr", "crf"):
        return
    display_state = self.display_clients.get(display_id)
    if display_state is None:
        return
    if display_state.get("rate_control_mode") == sanitized:
        return
    display_state["rate_control_mode"] = sanitized
    if display_id == 'primary':
        self.rc_mode = RateControlMode(sanitized)
        data_logger.debug(f"Session default rate_control_mode updated to {sanitized} for new displays.")
    if not display_state.get('video_active', True):
        return
    layout = self.display_layouts.get(display_id)
    if layout is None:
        return
    restart_ok = False
    async with self._reconfigure_guard():
        if display_state.get('video_active', True):
            data_logger.info(f"Applied rate-control via '_rc': {sanitized} for '{display_id}'. Restarting its capture stream.")
            await self._stop_capture_for_display(display_id)
            await self._start_capture_for_display(
                display_id=display_id,
                width=layout['w'], height=layout['h'],
                x_offset=layout['x'], y_offset=layout['y']
            )
            await self._start_backpressure_task_if_needed(display_id)
            self._schedule_idr_for_display(display_id)
            await self._broadcast_live_server_settings(display_id)
            if IS_WAYLAND:
                await self._sync_wayland_realized_geometry(display_id)
            restart_ok = self._opcode_display_module(display_id) is not None
    if not restart_ok:
        data_logger.warning(f"Rate-control restart failed for '{display_id}'; falling back to full reconfiguration.")
        await self.reconfigure_displays()

async def _handle_opcode_crf(self, crf: Any, display_id: str = 'primary') -> None:
    """Live CRF for the '_crf' verb; rides the tunables path like SETTINGS."""
    sanitized = sanitize_client_setting("video_crf", crf, self.cli_args, data_logger)
    if sanitized is None:
        return
    display_state = self.display_clients.get(display_id)
    if display_state is not None:
        display_state["video_crf"] = sanitized
    module = self._opcode_display_module(display_id)
    layout = self.display_layouts.get(display_id)
    if module is not None and layout is not None:
        try:
            fresh = self._get_capture_settings(
                display_id, layout['w'], layout['h'], layout['x'], layout['y']
            )
            module.update_tunables(fresh)
            self._track_capture_settings(display_id, fresh=fresh)
            data_logger.info(f"Applied CRF live via '_crf': {sanitized} for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live CRF update failed for '{display_id}' ({e}).")

async def set_client_stream_box(self, display_id: str, origin_x: float,
                                origin_y: float, scale_x: float,
                                scale_y: float) -> None:
    """Record where a display's page draws its stream on the user's desktop.

    Rebroadcast with the layout, since a page maps a drag that crossed onto
    a neighbor through the neighbor's box rather than off its own edge.
    Only the browser knows those origins, and they are the only thing
    relating two viewports whose monitors, window chrome and device pixel
    ratios all differ. Ignored for an unknown display or an impossible box.
    """
    display_state = self.display_clients.get(display_id)
    if display_state is None or display_id not in self.display_layouts:
        return
    if not (0.05 <= scale_x <= 100.0 and 0.05 <= scale_y <= 100.0):
        return
    if not (abs(origin_x) <= 100000.0 and abs(origin_y) <= 100000.0):
        return
    box = (origin_x, origin_y, scale_x, scale_y)
    if display_state.get("client_stream_box") == box:
        return
    # One box a page publishes is one broadcast to every client; a page that
    # alternated two would otherwise amplify at whatever rate it sent. The
    # page republishes what the layout comes back missing, so a dropped
    # update is not a lost one.
    now = time.monotonic()
    if now - display_state.get("client_stream_box_at", 0.0) < 0.2:
        return
    display_state["client_stream_box_at"] = now
    display_state["client_stream_box"] = box
    await self.broadcast_display_config()

def _display_config_payload(self) -> dict:
    """DISPLAY_CONFIG_UPDATE body: the display roster, the backend, plus
    each laid-out display's rectangle, its client's reported CSS-to-remote
    scale and the desktop box that client draws it in, so a page can map a
    cross-display drag into its neighbor's region and, on X11, a secondary
    can follow the primary's density."""
    payload = {
        "type": "display_config_update",
        "displays": list(self.display_clients.keys()),
        "wayland": IS_WAYLAND,
    }
    layouts = {}
    for did, rect in (self.display_layouts or {}).items():
        entry = dict(rect)
        client = self.display_clients.get(did) or {}
        scale = client.get("client_scale")
        if scale:
            entry["scale"] = scale
        box = client.get("client_stream_box")
        if box:
            (entry["originX"], entry["originY"],
             entry["scaleX"], entry["scaleY"]) = box
        layouts[did] = entry
    if layouts:
        payload["layouts"] = layouts
    return payload

async def broadcast_display_config(self) -> None:
    """Broadcast the current display roster to all clients."""
    if not self.clients:
        return

    payload = self._display_config_payload()
    message_str = f"DISPLAY_CONFIG_UPDATE,{json.dumps(payload)}"

    data_logger.debug(f"Broadcasting display config update: {message_str}")
    # Bounded: callers hold _reconfigure_lock.
    await _broadcast_to_clients(self.clients, message_str, per_client_timeout=2.0)

def refresh_cursor_cache(self) -> Optional[dict]:
    """Refresh and return the cached cursor payload for late-joining clients."""
    if not self.app:
        return None

    cursor_data = None
    if self.input_handler and hasattr(self.input_handler, "get_current_cursor_data"):
        cursor_data = self.input_handler.get_current_cursor_data()

    if cursor_data is not None:
        self.app.last_cursor_sent = cursor_data

    return self.app.last_cursor_sent

async def send_current_cursor(self, websocket: web.WebSocketResponse, raddr: Any) -> None:
    """Send the current cursor image to one client (used at connect/resume)."""
    cursor_data = self.refresh_cursor_cache()
    if not cursor_data:
        return

    data_logger.debug(f"Sending current cursor to client {raddr}")
    try:
        msg_str = json.dumps(cursor_data)
        await websocket.send_str(f"cursor,{msg_str}")
    except Exception as e:
        data_logger.warning(f"Failed to send current cursor to client {raddr}: {e}")

def _pcmflux_audio_callback(self, frame: Any) -> None:
    """Queue one encoded audio frame for broadcast.

    Called from pcmflux's capture thread with an AudioFrame, so it never
    touches asyncio state directly: the enqueue is scheduled onto the loop
    with call_soon_threadsafe, and loop/queue references are snapshotted
    because teardown can null them concurrently.
    """
    if self.is_pcmflux_capturing and frame is not None and self.pcmflux_audio_queue is not None:
        if len(frame) > 0:
            loop = self.pcmflux_capture_loop
            q = self.pcmflux_audio_queue
            if loop is None or q is None or loop.is_closed():
                return
            # Zero-copy: the AudioFrame owns the buffer (header included) and
            # frees it once the queue item drops.
            item = {'data': memoryview(frame), 'owner': frame}
            def _do_put():
                try:
                    q.put_nowait(item)
                except asyncio.QueueFull:
                    pass
            # The loop can close between the check above and here; the
            # RuntimeError would surface in pcmflux's C thread.
            try:
                loop.call_soon_threadsafe(_do_put)
            except RuntimeError:
                pass

def _check_pcmflux_health(self) -> None:
    """Restart the audio pipeline when its pcmflux worker died after start.

    pcmflux's start handshake answers while the worker is still starting,
    so a backend that gives up afterwards (its retry ladder spent, a
    mid-run reconnect budget exhausted) surfaces only through
    `last_error`; the broadcast loop asks here whenever its queue stays
    silent. The failure is logged once per run, and the restart — the same
    stop/start the audio toggles use — runs as its own task (the stop
    cancels the broadcast loop) no more often than the restart floor.
    """
    module = self.pcmflux_module
    if module is None or not self.is_pcmflux_capturing:
        return
    error = getattr(module, "last_error", None)
    if not error:
        return
    failure = (id(module), str(error))
    if self._pcmflux_reported_failure != failure:
        self._pcmflux_reported_failure = failure
        data_logger.error(f"pcmflux audio capture failed after start: {error}")
    now = time.monotonic()
    if now - self._pcmflux_last_restart < PCMFLUX_RESTART_FLOOR_SECONDS:
        return
    self._pcmflux_last_restart = now
    _spawn_background_task(self._restart_failed_pcmflux(module), name="pcmflux-restart")

async def _restart_failed_pcmflux(self, failed_module: Any) -> None:
    """Stop and start the audio pipeline under the reconfigure guard, unless
    the failed capture was already replaced or stopped meanwhile."""
    async with self._reconfigure_guard():
        if self.pcmflux_module is not failed_module or not self.is_pcmflux_capturing:
            return
        data_logger.info("Restarting the audio pipeline after its capture failed.")
        await self._stop_pcmflux_pipeline()
        await self._start_pcmflux_pipeline()

async def _pcmflux_send_audio_chunks(self) -> None:
    """Broadcast queued Opus audio chunks to the primary-viewer sockets.

    Runs as a long-lived task. Secondary-display sockets are excluded (they
    render video only; audio rides the primary connection), and sends are
    bounded so one stalled socket cannot freeze the shared stream. A queue
    that stays silent past the health interval is the cue to ask pcmflux
    whether the capture worker died (_check_pcmflux_health).
    """
    data_logger.debug("pcmflux audio chunk broadcasting task started.")
    try:
        while True:
            try:
                item = await asyncio.wait_for(
                    self.pcmflux_audio_queue.get(), timeout=PCMFLUX_HEALTH_INTERVAL_SECONDS)
            except asyncio.TimeoutError:
                self._check_pcmflux_health()
                continue

            secondary_websockets = {
                client_info.get('ws')
                for did, client_info in self.display_clients.items()
                if did != 'primary' and client_info.get('ws')
            }
            primary_viewers = self.clients - secondary_websockets

            if not primary_viewers:
                self.pcmflux_audio_queue.task_done()
                continue

            # A zero-copy view over the AudioFrame, header included; sent as-is.
            message_to_send = item['data']
            dropped = await _broadcast_to_clients(
                primary_viewers, message_to_send,
                per_client_timeout=SHARED_STREAM_SEND_TIMEOUT_SECONDS,
            )
            if dropped:
                # primary_viewers is a per-chunk temporary; the drop must reach the registry.
                self.clients -= dropped

            self.pcmflux_audio_queue.task_done()
    except asyncio.CancelledError:
        data_logger.debug("pcmflux audio chunk broadcasting task canceled.")
    finally:
        data_logger.debug("pcmflux audio chunk broadcasting task finished.")

def _compute_audio_red_distance(self) -> int:
    """RED distance for the shared audio broadcast.

    WS is TCP, but the sender still drops frames under backpressure
    (pcmflux delivery ring drop-oldest, and this server's audio queue drops
    on overflow), and RED lets the client recover those within the
    redundancy distance.

    Returns:
        The configured distance only when the server allows it AND there is
        at least one client AND every connected client advertised
        audioRedundancy; otherwise 0 (plain frames, which decode
        everywhere) — a single non-capable or legacy (field-absent) client
        falls the whole stream back.
    """
    if not self.audio_redundancy_enabled:
        return 0
    if not self.clients:
        return 0
    for ws in self.clients:
        if not self.audio_redundancy_by_ws.get(ws):
            return 0
    return getattr(settings, "audio_redundancy_distance", AUDIO_RED_DISTANCE)

async def _regate_audio_redundancy(self) -> None:
    """Recompute the RED gate for the shared audio stream and, if it flipped
    while capturing, restart the pipeline so the new red_distance takes
    effect. Callers hold the reconfigure guard (pipeline start/stop must be
    serialized against reconfigure_displays). A missing app means teardown
    (the last client leaving drops RED to 0, and the disconnect path stops
    the pipeline itself), so no restart is attempted then."""
    desired = self._compute_audio_red_distance()
    if desired == self._active_audio_red_distance:
        return
    if not self.is_pcmflux_capturing:
        return
    if not self.app:
        return
    data_logger.info(
        f"Audio RED gate changed ({self._active_audio_red_distance} -> {desired}); "
        "restarting audio pipeline."
    )
    await self._stop_pcmflux_pipeline()
    await self._start_pcmflux_pipeline()

async def _apply_initial_audio_policy(self, websocket: web.WebSocketResponse,
                                      display_id: str) -> None:
    """Bring the audio capture in line with the session's start policy once a
    page's first SETTINGS is in: started for a primary page that starts with
    audio on, stopped when nobody left listens to a capture that policy
    keeps off, and otherwise re-gated for the client set that just grew."""
    async with self._reconfigure_guard():
        audio_is_active = self.is_pcmflux_capturing
        if not pipeline_starts_on('audio', display_id):
            if audio_is_active and not self._audio_listeners(exclude=websocket):
                data_logger.debug("Initial setup: audio starts off for this session; stopping the idle audio capture.")
                await self._stop_pcmflux_pipeline()
        elif not audio_is_active and PCMFLUX_AVAILABLE and display_id == 'primary':
            data_logger.debug("Initial setup: Primary client connected, audio not active, attempting start.")
            await self._start_pcmflux_pipeline()
        elif not PCMFLUX_AVAILABLE and not audio_is_active:
            data_logger.warning("Initial setup: Audio pipeline (server-to-client) cannot be started (pcmflux not available).")
        else:
            await self._regate_audio_redundancy()

async def _start_pcmflux_pipeline(self) -> bool:
    """Start the pcmflux audio capture and the shared broadcast task.

    Resolves the RED distance for the current client set at start, so a
    gate change while running requires a restart (see
    _regate_audio_redundancy). Callers serialize via the reconfigure guard.

    Returns:
        True when capturing afterwards (already-running counts); False when
        audio is disabled, pcmflux is unavailable, or the start failed (a
        partial start is cleaned up).
    """
    if not settings.audio_enabled[0]:
        data_logger.debug("Audio is disabled by server settings. Not starting pipeline.")
        return False
    if not PCMFLUX_AVAILABLE:
        data_logger.error("Cannot start audio pipeline: pcmflux library not available.")
        return False
    if self.is_pcmflux_capturing:
        data_logger.debug("pcmflux audio pipeline is already capturing.")
        return True
    if self.pcmflux_module is not None:
        # A start canceled between the capture's open and its bookkeeping
        # (its page left) is retired before another module replaces it.
        await self._stop_pcmflux_pipeline()
    if not self.app:
        data_logger.error("Cannot start pcmflux: self.app (SelkiesStreamingApp instance) is not available.")
        return False

    self.pcmflux_capture_loop = self.capture_loop or asyncio.get_running_loop()
    if not self.pcmflux_capture_loop:
        data_logger.error("Cannot start pcmflux: asyncio event loop not found.")
        return False

    await ensure_capture_sink(self.audio_device_name)
    try:
        frame_ms = float(getattr(settings, 'audio_frame_duration_ms', '20') or 20)
        capture_settings = opus_capture_settings(self.audio_device_name, self.app.audio_channels,
                                                 int(self.app.audio_bitrate), frame_ms)
        capture_settings.debug_logging = self.cli_args.debug[0]
        # pcmflux's native [0x01,0x00] header goes on the wire; no Python prepend/copy.
        capture_settings.omit_audio_header = False
        red_distance = self._compute_audio_red_distance()
        capture_settings.red_distance = red_distance
        self._active_audio_red_distance = red_distance
        self.pcmflux_settings = capture_settings

        data_logger.info(f"Starting pcmflux audio pipeline: device '{self.audio_device_name}', "
                         f"{capture_settings.opus_bitrate} bps, {capture_settings.channels} ch, "
                         f"RED distance {red_distance}.")

        self.pcmflux_callback = self._pcmflux_audio_callback
        self.pcmflux_module = AudioCapture()
        self.pcmflux_audio_queue = asyncio.Queue(maxsize=getattr(self, 'BACKPRESSURE_QUEUE_SIZE', 120))

        await self.pcmflux_capture_loop.run_in_executor(
            None, self.pcmflux_module.start_capture, self.pcmflux_settings, self.pcmflux_callback
        )

        # The start handshake answers before the worker is up; a run that
        # already died reports through last_error.
        state = getattr(self.pcmflux_module, "state", "running")
        error = getattr(self.pcmflux_module, "last_error", None)
        if state == "failed" or error:
            raise RuntimeError(f"capture {state}: {error}")
        self.is_pcmflux_capturing = True
        if self.pcmflux_send_task is None or self.pcmflux_send_task.done():
            self.pcmflux_send_task = asyncio.create_task(self._pcmflux_send_audio_chunks())

        data_logger.debug(f"pcmflux audio capture state: {state}.")
        return True
    except Exception as e:
        data_logger.error(f"Failed to start pcmflux audio pipeline: {e}", exc_info=True)
        await self._stop_pcmflux_pipeline()
        return False

async def _stop_pcmflux_pipeline(self) -> bool:
    """Stop the audio capture and broadcast task; idempotent.

    The capturing flag is cleared first so the capture-thread callback
    stops queueing chunks before the queue is dropped.
    """
    if not self.is_pcmflux_capturing and not self.pcmflux_module:
        return True

    data_logger.info("Stopping pcmflux audio pipeline...")
    self.is_pcmflux_capturing = False

    if self.pcmflux_send_task:
        self.pcmflux_send_task.cancel()
        try:
            await self.pcmflux_send_task
        except asyncio.CancelledError:
            pass
        self.pcmflux_send_task = None

    if self.pcmflux_module:
        try:
            if self.pcmflux_capture_loop:
                await self.pcmflux_capture_loop.run_in_executor(
                    None, self.pcmflux_module.stop_capture
                )
        except Exception as e:
            data_logger.error(f"Error during pcmflux stop_capture: {e}")
        finally:
            del self.pcmflux_module
            self.pcmflux_module = None

    self.pcmflux_audio_queue = None
    data_logger.debug("pcmflux audio pipeline stopped.")
    return True

async def shutdown_pipelines(self) -> None:
    """Stop all capture pipelines; the ONLY way pipelines are stopped programmatically.

    Deadlock-proof by construction: reconfigure_displays() self-acquires
    the reconfigure lock, so it runs first and outside the guard; the
    audio/backpressure teardown then runs under the guard (a
    disconnect/connect race could otherwise tear down audio a new client
    just started), and none of the awaited teardowns re-acquire the lock.
    """
    logger.debug("Initiating unified pipeline shutdown...")
    await self.reconfigure_displays()
    async with self._reconfigure_guard():
        await self._stop_pcmflux_pipeline()
        if self.display_clients:
            stop_bp_tasks = [
                self._ensure_backpressure_task_is_stopped(disp_id)
                for disp_id in self.display_clients.keys()
            ]
            await asyncio.gather(*stop_bp_tasks, return_exceptions=True)
        if self.pcmflux_send_task and not self.pcmflux_send_task.done():
            self.pcmflux_send_task.cancel()
            try:
                await self.pcmflux_send_task
            except asyncio.CancelledError:
                pass
    logger.debug("Unified pipeline shutdown complete.")

async def _ensure_backpressure_task_is_stopped(self, display_id: str, notify: bool = True) -> bool:
    """Cancel and clean up the backpressure task for a specific display.

    Args:
        display_id: The display whose task to stop.
        notify: When True and a task was actually running, reset the frame
            ids and notify the client(s).

    Returns:
        Whether the pipeline-reset notification was sent, so callers that
        must guarantee a reset (capture stop) can send it exactly once
        themselves when no task was running.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        return False

    task_was_running = False
    task = display_state.get('backpressure_task')
    if task and not task.done():
        data_logger.debug(f"Ensuring frame backpressure task for '{display_id}' is stopped.")
        task.cancel()
        try:
            await task
            task_was_running = True
        except asyncio.CancelledError:
            data_logger.debug(f"Backpressure task for '{display_id}' canceled successfully.")
            task_was_running = True
        except Exception as e_cancel:
            data_logger.error(f"Error awaiting cancellation for '{display_id}' backpressure task: {e_cancel}")
        display_state['backpressure_task'] = None

    display_state['backpressure_enabled'] = True

    if task_was_running and notify:
        data_logger.debug(f"Backpressure task for '{display_id}' was stopped. Resetting its frame IDs.")
        await self._reset_frame_ids_and_notify(display_id)
        return True
    return False

async def _reset_frame_ids_and_notify(self, display_id: str) -> None:
    """Reset a display's frame-id state and send PIPELINE_RESETTING.

    For the primary display the reset is broadcast to ALL clients (shared
    viewers decode the same stream); a secondary notifies only its own
    socket. Every id-keyed artifact (send stamps, RTT samples, the fps
    estimator's baseline) resets with the numbering, since a stale stamp
    matched by a NEW id of the same value manufactures a giant RTT sample
    that poisons the smoothed estimate. The notify sends are bounded: this
    runs under _video_capture_lock, so one stalled client must not wedge
    every display's start/stop; a timed-out socket is dropped and closed
    rather than reused, and the id state was already reset either way.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        return

    data_logger.debug(f"Resetting frame IDs for display '{display_id}'.")
    display_state['last_sent_frame_id'] = 0
    display_state['has_sent_any_frame'] = False
    display_state['acknowledged_frame_id'] = -1
    display_state['acked_sent_at'] = None
    sent_ts = display_state.get('sent_timestamps')
    if sent_ts is not None:
        sent_ts.clear()
    rtt_samples = display_state.get('rtt_samples')
    if rtt_samples is not None:
        rtt_samples.clear()
    display_state['smoothed_rtt'] = 0.0
    display_state.pop('_fps_sample_acked', None)
    display_state.pop('_fps_sample_time', None)

    message = f"PIPELINE_RESETTING {display_id}"

    if display_id == 'primary' and self.clients:
        data_logger.debug(f"Broadcasting primary pipeline reset to all {len(self.clients)} clients: {message}")
        await _broadcast_to_clients(self.clients, message, per_client_timeout=2.0)
    else:
        websocket = display_state.get('ws')
        if websocket:
            try:
                await asyncio.wait_for(websocket.send_str(message), timeout=2.0)
            except asyncio.TimeoutError:
                data_logger.warning(f"Timed out notifying client for '{display_id}' of reset; dropping socket.")
                self.clients.discard(websocket)
                _close_abandoned_ws(websocket)
            except (ConnectionResetError, OSError, RuntimeError):
                data_logger.warning(f"Could not notify client for '{display_id}' of reset; connection closed.")

    display_state['backpressure_enabled'] = True
    display_state['unacked_since'] = None
    display_state['stall_gated_at'] = None

async def _start_backpressure_task_if_needed(self, display_id: str) -> None:
    """Start the backpressure task for a specific display if not already running.

    Backend-agnostic: frame ids, ACKs, and RTT flow identically on Wayland.
    A task restart is not a pipeline death, so it never notifies: the
    capture (re)start handles stream freshness, and the client reset
    belongs to the capture-stop path alone.
    """
    display_state = self.display_clients.get(display_id)
    if not display_state:
        data_logger.error(f"Cannot start backpressure task: display '{display_id}' not found.")
        return

    await self._ensure_backpressure_task_is_stopped(display_id, notify=False)

    task = display_state.get('backpressure_task')
    if not task or task.done():
        new_task = asyncio.create_task(self._run_frame_backpressure_logic(display_id))
        display_state['backpressure_task'] = new_task
        data_logger.debug(f"New frame backpressure task started for display '{display_id}'.")
    else:
        data_logger.warning(f"Backpressure task for '{display_id}' was already running. Not starting a new one.")

def _active_primary_consumers(self, exclude: Optional[web.WebSocketResponse] = None) -> set:
    """Sockets still consuming the primary broadcast.

    Every client except the secondary displays' owners, minus the paused
    ones. A paused viewer with a deferred rejoin pending counts as active —
    it keeps the capture alive under a waking viewer, while genuinely
    hidden ones let an all-tabs-hidden session stop encoding.

    Args:
        exclude: Drops the socket whose STOP_VIDEO is in flight.
    """
    secondary_ws = {
        info.get('ws')
        for did, info in self.display_clients.items()
        if did != 'primary'
    }
    waking = set(self._deferred_viewer_rejoins)
    consumers = self.clients - secondary_ws - (self.video_paused_clients - waking)
    if exclude is not None:
        consumers.discard(exclude)
    return consumers

def _audio_listeners(self, exclude: Optional[web.WebSocketResponse] = None) -> set:
    """Sockets the audio fan-out serves: every client but the secondary
    displays' owners, minus `exclude`."""
    secondary_ws = {
        info.get('ws')
        for did, info in self.display_clients.items()
        if did != 'primary'
    }
    listeners = self.clients - secondary_ws
    if exclude is not None:
        listeners.discard(exclude)
    return listeners

def _video_start_state(self, websocket: web.WebSocketResponse, display_id: str) -> bool:
    """The `video_active` a session owner's fresh page starts with.

    The start policy names it. An off state pauses this socket (the
    STOP_VIDEO rule): it leaves the primary fan-out until its START_VIDEO,
    so a capture a shared viewer starts meanwhile is not delivered to it,
    and while viewers consume the capture it keeps running for them.
    """
    if pipeline_starts_on('video', display_id):
        return True
    if display_id == 'primary':
        self.video_paused_clients.add(websocket)
        if self._active_primary_consumers(exclude=websocket):
            return True
    data_logger.info(f"Display '{display_id}' starts with video off; its capture waits for START_VIDEO.")
    return False

def _primary_reconnect_pending(self) -> bool:
    """Whether the primary display entry is being held for a socket that is
    already gone: the reconnect grace keeps the capture warm so a reloading
    page resumes on it instead of paying a full pipeline rebuild."""
    entry = self.display_clients.get('primary')
    return entry is not None and entry.get('ws') not in self.clients

async def _stop_stats_collectors(self) -> None:
    """Ends the resource monitor with the last client, nulling the ref so
    a fast reconnect restarts it and never reads a dead one's figures."""
    monitor, self._resource_monitor = self._resource_monitor, None
    if monitor is not None:
        await monitor.stop()

def _controller_socket(self, display_id: str) -> Optional[web.WebSocketResponse]:
    """The live socket registered for a display, unless a viewer holds it."""
    ws = (self.display_clients.get(display_id) or {}).get('ws')
    if ws is None or ws not in self.clients:
        return None
    if client_permissions.get(ws, {}).get("role") == "viewer":
        return None
    return ws

async def _send_stream_message(self, ws: web.WebSocketResponse, message: Dict[str, Any]) -> None:
    """One `stream_info` or `stream_stats` message to one socket; a socket
    still busy with the last one is skipped rather than queued behind."""
    if ws in self._stats_sending:
        return
    self._stats_sending.add(ws)
    try:
        await asyncio.wait_for(ws.send_str(json.dumps(message)), timeout=2.0)
    except (asyncio.TimeoutError, ConnectionResetError, OSError, RuntimeError):
        pass
    finally:
        self._stats_sending.discard(ws)

async def _publish_stream_info(self, display_id: str, info: Dict[str, Any]) -> None:
    """Tell a display's controller what its capture streams and how."""
    ws = self._controller_socket(display_id)
    if ws is not None:
        await self._send_stream_message(
            ws, {"type": "stream_info", "displayId": display_id, "info": info})

def _watch_stream(self, display_id: str, module: Any) -> None:
    """Follow a fresh capture's description for its display's controller."""
    watch = self._stream_watches.get(display_id)
    if watch is None:
        watch = self._stream_watches[display_id] = stream_stats.StreamWatch(
            display_id, self._publish_stream_info)
    entry = self.display_clients.get(display_id) or {}
    watch.follow(module, entry.get('encoder') or self.app.encoder, bool(entry.get('use_cpu')))

async def _send_stream_stats(self, _now: float) -> None:
    """Resource-monitor tick: one `stream_stats` to every subscribed controller,
    with its own display's encode figures, round trip and throttle state."""
    if not self._stats_subscribers:
        return
    host = stream_stats.host_stats(self._resource_monitor)
    for display_id, state in list(self.display_clients.items()):
        ws = self._controller_socket(display_id)
        if ws is None or ws not in self._stats_subscribers:
            continue
        stats = dict(host)
        watch = self._stream_watches.get(display_id)
        if watch is not None:
            stats.update(watch.rates())
        stats["rtt_ms"] = round(state.get('smoothed_rtt', 0.0), 1)
        stats["throttled"] = not state.get('backpressure_enabled', True)
        asyncio.ensure_future(self._send_stream_message(
            ws, {"type": "stream_stats", "displayId": display_id, "stats": stats}))

async def _stop_primary_if_unconsumed(self, reason: str) -> None:
    """Stop the primary capture once nothing decodes it — the last unpaused
    consumer hid its tab or disconnected. Hiding and disconnecting take the
    same verdict here; only a pending reconnect grace keeps the capture warm.
    A resume restarts it (START_VIDEO from the display owner, or the viewer
    path's capture ensure)."""
    if 'primary' not in self.capture_instances:
        return
    if self._active_primary_consumers() or self._primary_reconnect_pending():
        return
    data_logger.info(f"{reason} Stopping the 'primary' capture.")
    primary_entry = self.display_clients.get('primary')
    if primary_entry is not None:
        primary_entry['video_active'] = False
    await self._stop_capture_for_display('primary')

def _cancel_deferred_rejoin(self, websocket: web.WebSocketResponse) -> None:
    """Drop a pending deferred rejoin for this socket. A STOP_VIDEO (or a
    disconnect) arriving after a throttled resume supersedes it: the rejoin
    would otherwise un-pause a tab that is hidden again, and the socket would
    keep counting as a live consumer until it fired."""
    rejoin_task = self._deferred_viewer_rejoins.pop(websocket, None)
    if rejoin_task is not None:
        rejoin_task.cancel()