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.HIDDEN_OWNER_HANDOFF_S = 1.0
    self.IDLE_OWNER_HANDOFF_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_10bit = get_initial_value('video_10bit')
    self.video_10bit = self._initial_video_10bit
    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_cpu_encode_threading = get_initial_value('cpu_encode_threading')
    self.cpu_encode_threading = self._initial_cpu_encode_threading
    # Unset, a display's threads follow its encoder (`effective_cpu_encode_threading`).
    self._cpu_encode_threading_provided = bool(app_settings.was_provided('cpu_encode_threading'))
    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 = socket_dir(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._audio_lock = asyncio.Lock()
    self._video_capture_lock = asyncio.Lock()
    self._is_reconfiguring = False
    self._reconfigure_pending = False
    self.last_start_video_request_times = {}
    # When each page that does not own its display last had a repair of its
    # stream taken, by kind (`_repair_taken`).
    self._repair_times: Dict[Tuple[Any, str], float] = {}
    self.video_paused_clients = set()
    # The audio each controller beside the primary's owner asked for, its own:
    # the owner's page decides it for itself and the viewers, None until it
    # says, which leaves it to the start policy (`_hears_audio`).
    self._beside_audio: Dict[Any, bool] = {}
    self._session_audio: Optional[bool] = None
    # When each controller page that reports its visibility (`PAGE_VISIBILITY`)
    # last hid, None while it shows, and when it last showed: a page that
    # never reported is in neither (`_settle_display_owner`).
    self._page_hidden_at: Dict[Any, Optional[float]] = {}
    self._page_shown_at: Dict[Any, float] = {}
    # When each controller tab's user last acted through it (`page_put_to_use`),
    # or it took the primary; by tab, so a reconnect keeps it.
    self._tab_used_at: Dict[str, float] = {}
    self._owner_settle_task: Optional[asyncio.Task] = None
    self._deferred_viewer_rejoins = {}
    self.allowed_desync_ms = BACKPRESSURE_ALLOWED_DESYNC_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 = {}
    # Controllers of a display beside its owner, oldest first: a page of
    # another tab takes the stream and input with it, where the owner keeps
    # the display's size and encoding, and the oldest owns it once the owner
    # is gone for good (`_promote_co_controller`), the one that showed last
    # once the owner's tab is hidden (`_settle_display_owner`).
    self.co_controllers: Dict[str, "OrderedDict[Any, Optional[str]]"] = {}
    # Each controller page's last SETTINGS: a later one applies the stream
    # settings it changed, not the ones it repeats (`_settings_changed_by`).
    self._page_settings: Dict[Any, dict] = {}
    self.video_relay_groups = {}
    self.common_frames: Dict[Tuple[str, bool], CommonFrames] = {}
    self.capture_instances = {}
    # A display's capture while its start is awaited: its module takes rate,
    # tunable and key-frame requests meanwhile, and the settings registered
    # once it runs carry what was applied (`_opcode_display_module`).
    self._starting_captures: Dict[str, Dict[str, Any]] = {}
    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._red_unsettled = set()
    # When each socket's page last sent anything (pings and pongs are the browser's own).
    self._heard_at: Dict[Any, float] = {}
    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_session_screens_changed = self._republish_second_screen
    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 ScreenCapture module, running or starting, or None if
    not capturing: pixelflux takes rate, tunable and key-frame requests while
    a capture starts."""
    inst = self.capture_instances.get(display_id) or self._starting_captures.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) or self._starting_captures.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)

def _joins_beside_owner(self, websocket: Any, tab_id: Optional[str]) -> bool:
    """Whether a controller's SETTINGS for the primary makes it a controller
    beside the display's owner rather than its owner: with sharing on, a
    page of another tab than the live owner's, or one already beside it. A
    page that names no tab (an older client), or the owner's own tab
    reloading, takes the display over as before.
    """
    if websocket in self.co_controllers.get('primary', ()):
        return True
    if not getattr(self.cli_args, 'enable_sharing', (True,))[0]:
        return False
    owner = self.display_clients.get('primary')
    owner_ws = owner.get('ws') if owner else None
    return (owner_ws is not None and owner_ws is not websocket and not owner_ws.closed
            and bool(tab_id) and bool(owner.get('tab_id')) and tab_id != owner.get('tab_id'))

async def _promote_co_controller(self, display_id: str) -> bool:
    """Hand a display whose owner is gone for good to the oldest controller
    beside it that holds full permissions, else to the oldest one: told
    `DISPLAY_OWNER <display>`, the page sends its settings again, and that
    SETTINGS takes the entry over, with full control of the display, its own
    changes now applying. False when none is left to tell."""
    joiners = self.co_controllers.get(display_id)
    while joiners:
        ws = next((w for w in joiners if not w.closed and self._holds_full_control(w)), next(iter(joiners)))
        joiners.pop(ws)
        if ws.closed:
            continue
        try:
            await asyncio.wait_for(ws.send_str(f"DISPLAY_OWNER {display_id}"), timeout=2.0)
        except (asyncio.TimeoutError, ConnectionResetError, OSError, RuntimeError):
            continue
        data_logger.info(f"The controller beside '{display_id}' owns it now.")
        return True
    return False

def _seat_display_page(self, display_id: str, display_state: dict, websocket: Any,
                       tab_id: Optional[str]) -> None:
    """Point a display's entry at the page that owns it now: its decode pace,
    frame acknowledgements and round trips start over, and its stream waits
    for a key frame."""
    display_state['ws'] = websocket
    display_state['tab_id'] = tab_id
    self._apply_decode_pace(display_id, display_state, None)
    display_state['acknowledged_frame_id'] = -1
    display_state['acked_sent_at'] = None
    display_state['unacked_since'] = None
    display_state['stall_gated_at'] = None
    _expect_key_frame(display_state)
    display_state['sent_timestamps'].clear()
    display_state['rtt_samples'].clear()
    display_state['smoothed_rtt'] = 0.0

async def _settle_display_owner(self) -> None:
    """Hand the primary to the controller beside its owner that showed or was
    used last (`_hand_display_to`), once the owner's page has been hidden
    for HIDDEN_OWNER_HANDOFF_S: a phone put away, the display left to
    another device. Only pages that report their visibility and hold full
    permissions take it; a hidden owner with none beside it keeps the
    display, as does one whose socket closed, which the reconnect grace
    answers instead."""
    entry = self.display_clients.get('primary')
    owner = entry.get('ws') if entry else None
    hidden_at = self._page_hidden_at.get(owner) if owner is not None else None
    if owner is None or owner.closed or hidden_at is None:
        return
    joiners = self.co_controllers.get('primary') or {}
    visible = [ws for ws in joiners if not ws.closed and ws in self._page_hidden_at
               and self._page_hidden_at[ws] is None and self._holds_full_control(ws)]
    if not visible:
        return
    wait = hidden_at + self.HIDDEN_OWNER_HANDOFF_S - time.monotonic()
    if wait > 0:
        if self._owner_settle_task is None or self._owner_settle_task.done():
            async def _settle_later() -> None:
                await asyncio.sleep(wait)
                self._owner_settle_task = None
                await self._settle_display_owner()
            self._owner_settle_task = asyncio.create_task(_settle_later())
        return
    heir = max(visible, key=lambda ws: max(self._page_shown_at.get(ws, 0.0),
                                           self._tab_used_at.get(joiners[ws] or '', 0.0)))
    await self._hand_display_to(heir, "hid its page: the visible controller beside it")

async def _take_display_in_use(self, websocket: Any) -> None:
    """Hand the primary to the controller beside its owner whose user just
    acted through it, once the owner has gone IDLE_OWNER_HANDOFF_S without
    its own user's: one person moving between devices takes the display
    along, and nobody is disconnected. Only between pages of the same
    credential (the same token, or none), so a page of another token, a
    collaborator's, stays beside the owner; and only between pages that
    report their visibility, the one acting shown and holding full
    permissions."""
    entry = self.display_clients.get('primary')
    owner = entry.get('ws') if entry else None
    if (owner is None or owner.closed or websocket not in self.co_controllers.get('primary', ())
            or owner not in self._page_hidden_at or websocket not in self._page_hidden_at
            or self._page_hidden_at[websocket] is not None
            or time.monotonic() - self._tab_used_at.get(entry.get('tab_id') or '', 0.0)
            < self.IDLE_OWNER_HANDOFF_S
            or (client_permissions.get(websocket) or {}).get('token')
            != (client_permissions.get(owner) or {}).get('token')
            or not self._holds_full_control(websocket)):
        return
    await self._hand_display_to(websocket, "went unused: the controller beside it in use")

def _capture_for_page_beside(self, websocket: Any) -> None:
    """Start the primary capture a hidden owner stopped for a page that
    joined beside it and is still there once the owner's hand-off had its
    time (`_settle_display_owner`): an older page, or one without full
    control, never takes the display, and would wait on the owner. The
    owner's own feed stays paused until its START_VIDEO."""
    async def _later() -> None:
        await asyncio.sleep(self.HIDDEN_OWNER_HANDOFF_S + 0.5)
        owner = (self.display_clients.get('primary') or {}).get('ws')
        if (websocket.closed or websocket not in self.co_controllers.get('primary', ())
                or 'primary' in self.capture_instances or owner is None or owner.closed
                or not pipeline_starts_on('video', 'primary')):
            return
        self.video_paused_clients.add(owner)
        data_logger.info("Starting the 'primary' capture its hidden owner stopped, for the page beside it.")
        if await self._ensure_viewer_capture():
            try:
                await websocket.send_str("PIPELINE_RESETTING primary")
            except (ConnectionResetError, OSError, RuntimeError):
                pass
            self._schedule_idr_for_display('primary')
    _spawn_background_task(_later(), name="capture-for-page-beside")

async def _hand_display_to(self, heir: Any, why: str) -> None:
    """Make `heir`, a controller beside the primary's owner, its owner: told
    `DISPLAY_OWNER primary`, it sends its settings and its size, which lay
    the display out, and the owner stays beside it, paused while hidden."""
    entry = self.display_clients['primary']
    owner = entry['ws']
    joiners = self.co_controllers['primary']
    heir_tab = joiners.pop(heir)
    owner_tab = entry.get('tab_id')
    joiners[owner] = owner_tab
    self._seat_display_page('primary', entry, heir, heir_tab)
    _forget_path(entry)
    if heir_tab:
        self._tab_used_at[heir_tab] = time.monotonic()
    # Each page keeps the audio it asked for: the new owner's is the session's now.
    heir_audio = self._beside_audio.pop(heir, None)
    self._beside_audio[owner] = (self._session_audio if self._session_audio is not None
                                 else pipeline_starts_on('audio', 'primary'))
    if heir_audio is not None:
        self._session_audio = heir_audio
    if self._page_hidden_at.get(owner) is not None:
        self.video_paused_clients.add(owner)
    data_logger.info(
        f"The owner of 'primary' {why} owns the display now "
        f"(tab {(heir_tab or 'none')[:8]}, the old owner's {(owner_tab or 'none')[:8]}).")
    try:
        await asyncio.wait_for(heir.send_str("DISPLAY_OWNER primary"), timeout=2.0)
    except (asyncio.TimeoutError, ConnectionResetError, OSError, RuntimeError):
        pass

def _owns_display(self, websocket: Any, display_id: Optional[str]) -> bool:
    """Whether `websocket` is the page that owns `display_id`, the primary for none."""
    return (self.display_clients.get(display_id or 'primary') or {}).get('ws') is websocket

def _repair_taken(self, websocket: Any, display_id: Optional[str], kind: str) -> bool:
    """Whether a page's request to repair its display's stream (`kind`: a
    keyframe or a lost frame) is taken: always from the page that owns the
    display, at most once a second from any other, a viewer or a controller
    beside the owner, since any number of them share the owner's stream."""
    if self._owns_display(websocket, display_id):
        return True
    now = time.monotonic()
    if now - self._repair_times.get((websocket, kind), 0.0) < 1.0:
        return False
    self._repair_times[(websocket, kind)] = now
    return True

def _hears_audio(self, websocket: Any) -> bool:
    """Whether the audio fan-out serves a page: a controller beside the
    primary's owner by its own START_AUDIO, every other page by the owner's."""
    if websocket in self.co_controllers.get('primary', ()):
        return self._beside_audio.get(websocket, False)
    if self._session_audio is not None:
        return self._session_audio
    return pipeline_starts_on('audio', 'primary')

def _audio_wanted(self) -> bool:
    """Whether a page the audio fan-out serves wants the capture running."""
    return any(self._hears_audio(ws) for ws in self._audio_listeners())

def _holds_full_control(self, websocket: Any) -> bool:
    """Whether a page holds a display's full permissions: a controller that
    may drive keyboard and mouse, not a viewer, which only watches, nor a
    gamepad player."""
    perms = client_permissions.get(websocket) if websocket is not None else None
    return bool(perms) and perms.get("role") == "controller" and self._holds_input_authority(websocket)

def _cpu_encode_threading_for(self, encoder: str, requested: Optional[bool]) -> bool:
    """Whether a display on `encoder` threads its software encode: the client's request,
    the operator's value, or what the encoder defaults to (`effective_cpu_encode_threading`)."""
    return effective_cpu_encode_threading(encoder, requested, bool(self._initial_cpu_encode_threading),
                                          self._cpu_encode_threading_provided)

def _display_settings(self, display_id: str) -> Dict[str, Any]:
    """What a display streams with, keyed as SETTINGS carries it: the
    `use_cpu` the display asks for rather than the one a CPU-only encoder
    implies, and the session's audio bitrate."""
    state = self.display_clients.get(display_id) or {}
    values = {key: state.get(key) for key in STREAM_SETTINGS}
    requested = state.get("use_cpu_requested")
    values["use_cpu"] = bool(self._initial_use_cpu) if requested is None else bool(requested)
    values["audio_bitrate"] = self.app.audio_bitrate
    return {key: value for key, value in values.items() if value is not None}

async def _tell_display_settings(self, display_id: str, sockets: Any) -> None:
    """Tell pages what `display_id` streams with (`display_settings`): a
    page beside the display's owner holds it for its tab, as does the owner
    once another page's pick changed it, so each shows and asks for what
    the display does."""
    sockets = {ws for ws in sockets if ws is not None and ws in self.clients}
    if not sockets:
        return
    message = json.dumps({"type": "display_settings", "displayId": display_id,
                          "settings": self._display_settings(display_id)})
    for ws in await _broadcast_to_clients(sockets, message, watched=True):
        self.clients.discard(ws)

def _settings_changed_by(self, websocket: Any, settings: dict) -> dict:
    """A page's later SETTINGS without the stream settings it repeats from
    its last one: a page beside its owner may have changed them since. An
    encoder whose fallback mark changed counts as changed."""
    last = self._page_settings.get(websocket) or {}
    self._page_settings[websocket] = settings
    changed = {key for key in STREAM_SETTINGS if settings.get(key) != last.get(key)}
    if settings.get("encoderFallback") != last.get("encoderFallback"):
        changed.add("encoder")
    return {key: value for key, value in settings.items() if key not in STREAM_SETTINGS or key in changed}

async def _take_pick_beside_owner(self, websocket: Any, settings: dict, client_role: str) -> None:
    """A later SETTINGS from a controller beside the primary's owner. What
    its user picked of the stream (`picked`) applies to the display where
    both pages hold full permissions, and the display's pages are told; the
    rest, its own size and density and what its page changed on its own,
    follows the owner, and the page is told what that is."""
    picks = {key: settings[key] for key in (settings.get("picked") or ())
             if key in STREAM_SETTINGS and settings.get(key) is not None}
    owner_ws = (self.display_clients.get('primary') or {}).get('ws')
    if not (picks and self._holds_full_control(websocket) and self._holds_full_control(owner_ws)):
        await self._tell_display_settings('primary', {websocket})
        return
    data_logger.info(f"Applying {', '.join(sorted(picks))} picked by the controller beside the owner of 'primary'.")
    # Every other key as a payload that does not carry it: what the owner's
    # page sends of the display's geometry stays as it was.
    applied = dict.fromkeys(settings)
    applied.update(picks, displayId='primary')
    if "encoder" in picks:
        applied["encoderFallback"] = settings.get("encoderFallback")
    await self._apply_client_settings(websocket, applied, False, client_role)
    await self._tell_display_settings('primary', {owner_ws, *self.co_controllers.get('primary', {})})

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

    The layout comes from an initial SETTINGS, and a client's own carries
    the geometry, so a resize ahead of it is left to it: waited for here,
    it would hold the message loop that has to read that SETTINGS.
    """
    if not self.client_settings_received.is_set():
        data_logger.debug("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. A sync ahead of the initial SETTINGS, which carries the DPI,
    is left to it, as `_handle_resize` leaves a resize.
    """
    if not self.client_settings_received.is_set():
        data_logger.debug("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)

def _capture_fps(self, display_state: dict) -> float:
    """The rate a display's capture runs at: the rate its client chose, or
    the session's, held to what that client's decoder keeps up with
    (`DECODE_PACE`)."""
    rate = float(display_state.get('framerate') or self.app.framerate)
    pace = display_state.get('decode_pace')
    return min(rate, pace) if pace else rate

def _apply_decode_pace(self, display_id: str, display_state: dict, pace: Optional[float]) -> None:
    """Hold a display's capture to `pace`, the rate its client's decoder
    keeps up with, or lift the hold (None), live where the capture runs. The
    pace lives in the display's entry: a new connection taking the entry
    over starts without it, and the rate the client chose stays stored."""
    before = self._capture_fps(display_state)
    display_state['decode_pace'] = pace
    after = self._capture_fps(display_state)
    if after == before:
        return
    data_logger.info(
        f"Display '{display_id}': its client's decoder keeps up with {fps_label(pace)} fps; capturing at that."
        if pace else f"Display '{display_id}': capturing at the chosen {fps_label(after)} fps again.")
    module = self.capture_instances.get(display_id, {}).get('module')
    if module is not None:
        try:
            module.update_framerate(after)
            self._track_capture_settings(display_id, target_fps=after)
        except Exception as e:
            data_logger.warning(f"Live framerate update failed for '{display_id}' ({e}).")

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 {fps_label(sanitized)} fps 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:
            live = self._capture_fps(display_state) if display_state is not None else float(sanitized)
            module.update_framerate(live)
            self._track_capture_settings(display_id, target_fps=live)
            data_logger.info(f"Applied framerate live via '_arg_fps': {fps_label(sanitized)} fps for '{display_id}'")
        except Exception as e:
            data_logger.warning(f"Live framerate update failed for '{display_id}' ({e}).")
    await self._refresh_display_mode()

async def _refresh_display_mode(self) -> None:
    """Keep the display's refresh at the rule its resizes follow, after a
    frame rate changed without one: never below the fastest display of the
    layout (`display_utils.refresh_output_mode`, no capture restarts)."""
    if IS_WAYLAND or not self.display_layouts:
        return
    await refresh_output_mode(max(
        float((self.display_clients.get(did) or {}).get('framerate') or self.app.framerate)
        for did in self.display_layouts))

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(self._video_bitrate_kbps(display_state) if display_state else 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 audio lock like every audio start/stop, and re-checked
            # there: a concurrent one must not orphan a second AudioCapture.
            async with self._audio_lock:
                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 and whether
    it takes a touchpad's scroll as a finger's, 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,
        "finger_scroll": bool(self.input_handler and self.input_handler.finger_scroll_available()),
    }
    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, watched=True)

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 audio lock, unless the
    failed capture was already replaced or stopped meanwhile."""
    async with self._audio_lock:
        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()