WebRTCService

def __init__(self, supervisor: CentralizedStreamServer) -> None:
    super().__init__("webrtc")
    self.settings: Optional[AppSettings] = settings
    self.tasks: List[asyncio.Task] = []
    self.shutdown_event = asyncio.Event()
    self._shutdown_called = False
    self.signaling_client: Optional[WebRTCSignalingClient] = None
    self.media_pipeline: Optional[MediaPipelinePixel] = None
    self.rtc_app: Optional[RTCApp] = None
    self.input_handler: Optional[WebRTCInput] = None
    self.system_monitor: Optional[SystemMonitor] = None
    self.gpu_monitor: Optional[GPUMonitor] = None
    self.metrics: Optional[Metrics] = None
    self.peer_id = 1
    self.args: Optional[SimpleNamespace] = None
    self.monitoring_utils_used: Dict[str, bool] = {}
    self.mon_hmac_turn: Optional[HMACRTCMonitor] = None
    self.mon_rest_api: Optional[RESTRTCMonitor] = None
    self.mon_rtc_config_file: Optional[RTCConfigFileMonitor] = None
    self.mon_cloudflare_turn: Optional[CloudflareRTCMonitor] = None
    self.peer_manager: Optional[WebRTCPeerManagement] = None
    self.supervisor = supervisor
    self.display_clients: Dict[str, Dict[str, Any]] = {}
    self.display_layouts: Dict[str, Dict[str, int]] = {}
    self.display_pipelines: Dict[str, MediaPipelinePixel] = {}
    self._last_idr_request_times: Dict[str, float] = {}
    self._display_lock = asyncio.Lock()
    self._primary_dims: Optional[Tuple[int, int]] = None
    self._wayland_ctl_module: Optional[Any] = None
    self._host_output_capacity: Optional[int] = None
    self._last_resize_request: Optional[Tuple[int, int]] = None
    self._wm_swap = MultiMonitorWindowManager()
    self.RECONNECT_GRACE_S = 3.0
    self._primary_stop_grace_task: Optional[asyncio.Task] = None

    self._mic_control: Optional[AudioControl] = None
    self._mic_module_index: Optional[int] = None
    self._mic_module_owned = False
    self._mic_provisioned = False
    self._mic_provision_lock = asyncio.Lock()

    self._init_default_settings()

def _init_default_settings(self) -> None:
    """Seed ``self.args`` from ``SETTING_DEFINITIONS`` and derive the manual
    startup geometry.

    Range settings pin to the single allowed value when the range is locked
    (min == max), otherwise take the definition's default; other types copy
    the configured value verbatim.
    """
    self.args = SimpleNamespace()
    try:
        for setting_def in SETTING_DEFINITIONS:
            name = setting_def["name"]
            stype = setting_def["type"]
            if stype == "bool":
                value = getattr(self.settings, name)[0]
            elif stype == "range":
                min, max = getattr(self.settings, name)
                value = (
                    min
                    if min == max
                    else setting_def.get("meta", {}).get("default_value", 0)
                )
            elif stype == "enum":
                value = getattr(self.settings, name)
            elif stype in ("int", "float", "str", "list"):
                value = getattr(self.settings, name)
            else:
                continue
            setattr(self.args, name, value)
    except Exception as e:
        logger.error(f"Error initializing default settings: {e}", exc_info=True)

    self._manual_dims: Optional[Tuple[int, int]] = None
    if getattr(self.args, "manual_resolution", False):
        width = int(getattr(self.args, "manual_width", 0) or 0)
        height = int(getattr(self.args, "manual_height", 0) or 0)
        if width > 0 and height > 0:
            self._manual_dims = (width - (width % 2), height - (height % 2))

async def initialize_components(self) -> None:
    """Build every component: metrics, signaling, the primary media
    pipeline, the RTC app, the input handler, and the monitors, then wire
    the peer manager with the fetched RTC configuration.

    The settings are re-snapshotted first: the service is constructed once
    at boot, but a live transport switch lands here with the settings
    singleton already re-resolved for webrtc (encoder filter, rate-control
    default). Metrics backs both the Prometheus endpoint and the WebRTC
    CSV statistics, so it is built when either flag is on. A configured
    manual resolution is applied before the pipeline is sized: on X11 the
    screen is resized now and the pipeline takes what the X server
    realized (CVT cell alignment can widen the mode); on Wayland the
    dimensions are the resize, since the capture start sizes the compositor
    output from them, and the capture scale is seeded from the configured
    DPI so the first start honors it (`handle_scaling` updates it later).
    The interposer socket paths and the gamepad backend are process-wide
    state shared with the websockets service, so both transports read the
    same settings for them.
    """

    self._init_default_settings()

    if self.args.enable_metrics_http or self.args.enable_webrtc_statistics:
        webrtc_csv = self.args.enable_webrtc_statistics
        self.metrics = Metrics(using_webrtc_csv=webrtc_csv)

    self.signaling_client = self.create_signaling_client()

    # Surround (>2ch) rides Chromium's multiopus codec; the offered codec set
    # must be swapped before any peer connection builds its capabilities.
    if int(self.args.audio_channels) > 2:
        configure_multiopus(int(self.args.audio_channels))

    self.media_pipeline = MediaPipelinePixel(
        async_event_loop=asyncio.get_running_loop(),
        encoder=self.args.encoder,
        framerate=int(self.args.framerate),
        # kbps, as consumed by pixelflux.
        video_bitrate=int(self.args.video_bitrate),
        # Enum with a wider server-side value_range: an operator override can
        # arrive as an arbitrary numeric string.
        audio_bitrate=int(float(self.args.audio_bitrate)),
        audio_channels=int(self.args.audio_channels),
        audio_enabled=self.args.audio_enabled,
        audio_device_name=self.args.audio_device_name,
        crf=int(self.args.video_crf),
        video_fullcolor=bool(self.args.video_fullcolor),
        use_cpu=bool(self.args.use_cpu),
        video_streaming_mode=bool(self.args.video_streaming_mode),
        use_paint_over_quality=bool(self.args.use_paint_over_quality),
        video_paintover_crf=int(self.args.video_paintover_crf),
        video_paintover_burst_frames=int(self.args.video_paintover_burst_frames),
    )
    if self._manual_dims:
        if not IS_WAYLAND:
            realized = await resize_display(f"{self._manual_dims[0]}x{self._manual_dims[1]}")
            if realized:
                self._manual_dims = realized
        self.media_pipeline.width, self.media_pipeline.height = self._manual_dims
    if self.args.enable_rate_control:
        self.media_pipeline.rc_mode = RateControlMode(self.args.rate_control_mode)
    else:
        # Rate control disabled runs CRF on both transports.
        self.media_pipeline.rc_mode = RateControlMode.CRF

    (
        stun_servers,
        turn_servers,
        rtc_config,
        self.monitoring_utils_used,
    ) = await get_rtc_configuration(self.args)
    self.rtc_app = RTCApp(
        async_event_loop=asyncio.get_running_loop(),
        encoder=self.args.encoder,
        stun_servers=stun_servers,
        turn_servers=turn_servers,
    )
    self.rtc_app.media_pipeline = self.media_pipeline
    self.rtc_app.provision_virtual_mic = self._provision_webrtc_virtual_mic
    self.display_pipelines["primary"] = self.media_pipeline

    self.input_handler = WebRTCInput(
        rtc_app=self.rtc_app,
        uinput_mouse_socket_path=getattr(self.args, "uinput_mouse_socket", "") or "",
        js_socket_path_prefix=getattr(self.args, "js_socket_path", "/tmp"),
        enable_clipboard=self.args.enable_clipboard,
        enable_binary_clipboard="true"
        if self.args.enable_binary_clipboard
        else "false",
        enable_cursors=self.args.enable_cursors,
        cursor_size=self.args.cursor_size,
        cursor_scale=1.0,
        cursor_debug=self.args.debug_cursors,
        upload_dir=self.args.file_manager_path,
        is_wayland=IS_WAYLAND,
        app_wayland_display=(getattr(self.args, "app_wayland_display", "")
                             or getattr(self.args, "wayland_host_display", "")),
        uinput_gamepad=getattr(self.args, "uinput_gamepad", "auto"),
        # Duck-typed layout source: send_x11_mouse offsets a secondary
        # display's coordinates by display_layouts[display_id].
        data_server_instance=self,
    )
    self.input_handler.initialize_upload_dir()
    if IS_WAYLAND:
        self.media_pipeline.scale = await self.input_handler.realize_wayland_dpi(
            getattr(settings, "scaling_dpi", "96") or 96)

    self.system_monitor = SystemMonitor()
    # Always enabled: gpu_stats reports nothing without a supported GPU/tool.
    # Keyed to the pipeline's render node so stats describe the encoding GPU.
    stats_gpu_id = parse_gpu_id(getattr(self.args, "gpu_id", ""))
    self.gpu_monitor = GPUMonitor(
        gpu_id=stats_gpu_id if (stats_gpu_id or 0) > 0 else 0,
        enabled=True,
        dri_node=getattr(self.args, "encode_dri", "") or "",
    )

    self.create_peer_manager(rtc_config)

def create_signaling_client(self) -> WebRTCSignalingClient:
    """Create and configure signaling client."""
    using_https = self.args.enable_https
    using_basic_auth = self.args.enable_basic_auth
    ws_protocol = "wss:" if using_https else "ws:"

    prefix = self.settings.subfolder
    username = self.settings.basic_auth_user
    password = self.settings.basic_auth_password
    client = WebRTCSignalingClient(
        f"{ws_protocol}//127.0.0.1:{self.args.port}{prefix}/api/ws",
        enable_https=using_https,
        enable_basic_auth=using_basic_auth,
        basic_auth_user=username,
        basic_auth_password=password,
        server_token=getattr(self.settings, "master_token", None),
    )
    return client

async def handle_signaling_error(self, error: Exception) -> None:
    """Handle signaling errors."""
    logger.error(f"Signaling client error: {error}. Closing the pipelines")
    await self.handle_signaling_disconnect()

async def handle_signaling_disconnect(self) -> None:
    """Tear down every RTC connection once the signaling link drops."""
    logger.info("Signaling disconnected, cleaning up all resources")
    try:
        await self.rtc_app.stop_all_rtc_connections()
    except Exception as e:
        logger.error(
            f"Error during signaling disconnect cleanup: {e}", exc_info=True
        )

async def handle_session_start(
    self, session_peer_id: str, client_type: str, client_token: Optional[str] = None,
    display_id: str = "primary", display_position: str = "right",
) -> None:
    """Start an RTC connection for a joining peer.

    A secondary display's controller is gated on the effective second-screen
    availability first: the published setting can lag a host-side change, so
    the capacity is re-read and a refusal closes the peer's signaling socket
    with a fatal verdict (a bare return would leave the page on
    "Connecting..." forever).

    Args:
        session_peer_id: The signaling peer id of the joining client.
        client_type: "controller" or a viewer/shared role name.
        client_token: Optional per-client auth token (governs input role).
        display_id: The display this peer consumes ("primary" or a
            secondary id).
        display_position: Where a joining secondary sits relative to the
            primary ("right", "left", "up", "down").
    """
    logger.info(
        f"starting session for client peer id: {session_peer_id} of type: {client_type} (display '{display_id}')"
    )
    try:
        if display_id != "primary" and client_type == "controller":
            await self._refresh_second_screen_capacity()
            available, reason = self._second_screen_availability()
            if not available:
                logger.warning(
                    "Secondary display '%s' refused: %s", display_id, reason,
                )
                await self._close_peer_signaling_ws(
                    session_peer_id, 4000, reason.encode(),
                )
                return
            # Dimensions arrive through the client's first resize message.
            entry = self.display_clients.setdefault(display_id, {"width": 0, "height": 0})
            entry["position"] = display_position
            self._seed_display_settings(entry)
        await self.rtc_app.start_rtc_connection(session_peer_id, client_type, client_token, display_id)
        if self.args.enable_webrtc_statistics and self.metrics:
            await self.metrics.initialize_webrtc_csv_file(self.args.webrtc_statistics_dir)
        logger.info(f"started session for client peer id {session_peer_id}")
    except Exception as e:
        logger.error(
            f"Error starting session for client peer id {session_peer_id}: {e}",
            exc_info=True,
        )
        await self.rtc_app.stop_rtc_connection(session_peer_id, client_type)

async def handle_session_end(self, session_peer_id: str, client_type: str) -> None:
    """Handle end of a session initiated by a client.
    Stops the RTC connection and media pipeline for the given session peer id.
    """
    try:
        if self.rtc_app:
            await self.rtc_app.stop_rtc_connection(session_peer_id, client_type)
        logger.info(
            f"session ended for client peer id {session_peer_id} of type {client_type}"
        )
    except Exception as e:
        logger.error(
            f"Error handling session end for {session_peer_id}: {e}", exc_info=True
        )

def create_peer_manager(self, rtc_config: Any) -> None:
    """Build the signaling-side peer manager with the fetched RTC config
    and the TURN/STUN/sharing options the handshake needs."""
    options = argparse.Namespace(
        keepalive_timeout=30,
        rtc_config_file=self.args.rtc_config_json,
        turn_shared_secret=self.args.turn_shared_secret,
        rtc_config=rtc_config,
        turn_host=self.args.turn_host,
        turn_port=self.args.turn_port,
        turn_protocol=self.args.turn_protocol,
        turn_tls=self.args.turn_tls,
        turn_auth_header_name=self.args.turn_rest_username_auth_header,
        stun_host=self.args.stun_host,
        stun_port=self.args.stun_port,
        enable_sharing=self.args.enable_sharing,
        enable_shared=self.args.enable_shared,
        enable_player2=self.args.enable_player2,
        enable_player3=self.args.enable_player3,
        enable_player4=self.args.enable_player4,
    )
    self.peer_manager = WebRTCPeerManagement(options)
    self.peer_manager.on_client_presence = self.supervisor.set_clients_present

def setup_callbacks(self) -> None:
    """Wire signaling, RTC app, media pipeline, input handler and monitor
    callbacks to each other.

    Cursors come from pixelflux on both backends (Wayland compositor / X11
    XFixes monitor) and route through the input handler's transport
    callback, capped at its DPI-scaled cursor size. Offers resolve their
    codec and SDP munging per display, so displays can run different
    encoders and chroma formats and a live full-colour toggle reaches
    every later offer. DPI scaling is wired independently of
    `enable_resize`, which gates only the primary's dynamic resolution
    (in `on_resize_handler`): the websockets transport applies scaling
    through the SETTINGS payload regardless of the resize gate, and a
    secondary display's whole bring-up rides its resize message.
    """
    if not self.rtc_app or not self.media_pipeline or not self.input_handler:
        return

    self.signaling_client.on_error = self.handle_signaling_error
    self.signaling_client.on_disconnect = self.handle_signaling_disconnect
    self.signaling_client.on_session_start = self.handle_session_start
    self.signaling_client.on_session_end = self.handle_session_end
    self.signaling_client.on_sdp = self.rtc_app.set_sdp
    self.signaling_client.on_ice = self.rtc_app.set_ice

    self.media_pipeline.produce_data = self.rtc_app.consume_data
    self.media_pipeline.on_pipeline_started = self.send_current_cursor

    self.rtc_app.request_idr_frame = self.request_idr_for_display
    self.rtc_app.start_display_media = self.start_display_media
    self.rtc_app.stop_display_media = self.stop_display_media
    self.rtc_app.on_sdp = self.signaling_client.send_sdp
    self.rtc_app.on_ice = self.signaling_client.send_ice
    self.rtc_app.on_data_open = self.handle_data_channel_open
    self.rtc_app.on_data_close = lambda: logger.info("Data channel closed")
    self.rtc_app.on_data_error = lambda e: logger.error(f"Data channel error: {e}")
    self.rtc_app.on_data_message = self.input_handler.on_message
    self.rtc_app.on_peer_gone = self.handle_peer_gone
    self.input_handler.on_request_keyframe = self.request_idr_for_display

    self.input_handler.on_cursor_change = lambda data: (
        self.rtc_app.send_cursor_data(data)
    )
    self.media_pipeline.on_cursor_data = lambda data: (
        self.input_handler.on_cursor_change(data)
    )
    self.media_pipeline.get_cursor_size_cap = lambda: getattr(
        self.input_handler, "cursor_size_cap", 0
    )
    self.input_handler.on_video_encoder_bit_rate = self.handle_video_bitrate_change
    self.input_handler.on_audio_encoder_bit_rate = self.handle_audio_bitrate_change
    self.input_handler.on_mouse_pointer_visible = self.handle_pointer_visible
    self.input_handler.on_clipboard_read = lambda d, t: (
        self.rtc_app.send_clipboard_data(d, t)
    )
    self.input_handler.on_set_fps = self.handle_fps_change
    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_ping_response = lambda latency: (
        self.rtc_app.send_latency_time(latency)
    )
    self.input_handler.on_client_webrtc_stats = self.handle_client_werbtc_stats
    self.input_handler.on_update_settings = self.handle_update_settings
    self.input_handler.on_update_rate_control_mode = self.handle_rate_control_change
    self.input_handler.on_update_crf = self.handle_crf_change
    self.rtc_app.get_encoder_for_display = self._encoder_for_display
    self.rtc_app.get_fullcolor_for_display = self._fullcolor_for_display
    self.rtc_app.get_use_cpu_for_display = self._use_cpu_for_display
    self.rtc_app.on_video_consumer_active = self.handle_video_consumer_active
    self.rtc_app.on_consumers_changed = self.handle_consumers_changed
    # /api/tokens updates must reach live WebRTC peers too.
    selkies_module.webrtc_reconcile_hook = self.reconcile_webrtc_peers

    self.input_handler.on_scaling_ratio = self.handle_scaling
    self.input_handler.on_resize = self.on_resize_handler

    self.gpu_monitor.on_stats = self.handle_gpu_stats
    self.system_monitor.on_timer = self.handle_system_monitor

def _second_screen_availability(self) -> Tuple[bool, str]:
    """Whether this session can actually attach a second display, and the
    reason when it cannot (websockets-mode parity).

    The admin flag gates first; past it, X11 and the self-composited
    Wayland backend mint another output on demand, while host capture is
    bounded by the host compositor's real output count (unknown until a
    pipeline start establishes the host session).

    Returns:
        A ``(available, reason)`` pair; ``reason`` is empty when available.
    """
    enabled, _ = self.settings.second_screen
    if not enabled:
        return False, "Second screens are disabled on this server."
    if not IS_WAYLAND or not (self.settings.wayland_host_display or "").strip():
        return True, ""
    capacity = self._host_output_capacity
    if capacity is None or capacity < 0:
        return False, "The host compositor's outputs are not known yet."
    if capacity < 2:
        return False, "The host compositor has a single output, so a second display has nothing to capture."
    return True, ""

async def _refresh_second_screen_capacity(self) -> bool:
    """Host-capture mode only: re-read how many outputs the host exposes.

    Returns:
        True when the answer changed, i.e. the second-screen availability
        that clients were told may have flipped.
    """
    if not IS_WAYLAND or not (self.settings.wayland_host_display or "").strip():
        return False
    module = self._wayland_capture_handle()
    if module is None or not hasattr(module, "output_capacity"):
        return False
    try:
        capacity = int(await asyncio.to_thread(module.output_capacity))
    except Exception as e:
        logger.warning(f"Wayland output capacity query failed: {e}")
        return False
    changed = capacity != self._host_output_capacity
    self._host_output_capacity = capacity
    return changed

def _server_settings_payload(self) -> Dict[str, Any]:
    """get_server_settings with second_screen and ui_sidebar_show_apps
    published as EFFECTIVE availability — the admin flag AND what the
    backend can actually do — so dashboards never offer a second display
    the server would immediately refuse, nor an apps panel whose every
    button would fail. Adds the terminal the apps panel launches in, chosen
    by the session's windowing system (absent when none is installed: the
    client keeps its default)."""
    payload = get_server_settings()
    available, _ = self._second_screen_availability()
    entry = payload.get("settings", {}).get("second_screen")
    if isinstance(entry, dict) and entry.get("value") and not available:
        payload["settings"]["second_screen"] = dict(entry, value=False)
    apps = payload.get("settings", {}).get("ui_sidebar_show_apps")
    if (isinstance(apps, dict) and apps.get("value")
            and self.input_handler and not self.input_handler.apps_available()):
        payload["settings"]["ui_sidebar_show_apps"] = dict(apps, value=False)
    terminal = self.input_handler.app_terminal() if self.input_handler else None
    if terminal:
        payload["settings"]["app_terminal"] = {"value": terminal}
    return payload

def handle_data_channel_open(self, channel: Optional[Any] = None) -> None:
    """Greet the peer that just joined: every display page and viewer needs
    the server settings for conditional UI, the current cursor, and the
    display roster. Sent on ITS channel when given; without one, falls back
    to broadcasting."""
    logger.info("opened peer data channel for user input to X11")
    server_settings_payload = self._server_settings_payload()
    if channel is not None:
        self.rtc_app.send_message_to_channel(
            channel, "server_settings", server_settings_payload
        )
        displays = ["primary"] + [d for d in self.display_clients.keys() if d != "primary"]
        self.rtc_app.send_message_to_channel(
            channel, "display_config_update", {"displays": displays}
        )
    else:
        self.rtc_app.send_media_data_over_channel(
            "server_settings", server_settings_payload
        )
        self._broadcast_display_config()
    self.send_current_cursor(channel)

def send_current_cursor(self, channel: Optional[Any] = None) -> None:
    """Resend the current cursor (on channel open / video restart): to one
    peer's channel when given, otherwise to every connected peer.

    Idempotent; a slept/woken tab clears its cursor canvas and needs it back.
    """
    if not self.rtc_app:
        return
    cursor_data = None
    if self.input_handler:
        try:
            cursor_data = self.input_handler.get_current_cursor_data()
        except Exception as e:
            logger.warning(f"Failed to fetch current cursor data: {e}")
    if cursor_data is None:
        cursor_data = self.rtc_app.last_cursor_sent
    if not cursor_data:
        return
    try:
        if channel is not None:
            self.rtc_app.send_message_to_channel(channel, "cursor", cursor_data)
        else:
            self.rtc_app.send_cursor_data(cursor_data)
    except Exception as e:
        logger.warning(f"Failed to send current cursor to client: {e}")

async def handle_pointer_visible(self, visible: bool) -> None:
    """Compose the cursor into the captured video, on every display's capture
    (the websockets capture_cursor tunable is likewise global)."""
    for pipeline in list(self.display_pipelines.values()):
        if pipeline is not None:
            await pipeline.set_pointer_visible(visible)

async def handle_video_bitrate_change(self, bitrate: int, display_id: str = "primary") -> None:
    """Video bitrate change for the display whose page sent it; sanitized
    against the server's configured range like the SETTINGS path, so the
    opcode cannot bypass a locked/narrowed range."""
    sanitized = sanitize_client_setting("video_bitrate", bitrate, self.settings, logger)
    if sanitized is None or sanitized == self._display_setting(display_id, "video_bitrate"):
        return
    await self._apply_display_setting(display_id or "primary", "video_bitrate", sanitized)

async def handle_audio_bitrate_change(self, bitrate: int) -> None:
    """Handle audio bitrate change request (bps; sanitized like SETTINGS)."""
    sanitized = sanitize_client_setting("audio_bitrate", bitrate, self.settings, logger)
    if sanitized is None or sanitized == getattr(self.args, "audio_bitrate", None):
        return
    if self.media_pipeline:
        await self.media_pipeline.set_audio_bitrate(int(sanitized))
    self.args.audio_bitrate = sanitized

async def handle_fps_change(self, fps: int, display_id: str = "primary") -> None:
    """Framerate change for the display whose page sent it; sanitized against
    the server's configured range like the SETTINGS path."""
    sanitized = sanitize_client_setting("framerate", fps, self.settings, logger)
    if sanitized is None or sanitized == self._display_setting(display_id, "framerate"):
        return
    await self._apply_display_setting(display_id or "primary", "framerate", sanitized)

async def handle_rate_control_change(self, mode: Any, display_id: str = "primary") -> None:
    """Rate-control switch for the display whose page sent it; honors the
    server's enable_rate_control lock like the SETTINGS path."""
    if self.args.enable_rate_control is False:
        logger.debug("Server has rate control disabled. Ignoring rate-control change.")
        return
    # Store the plain value: str(<str-Enum>) formats as the member name on
    # some supported Python versions, which would corrupt later comparisons.
    mode_str = mode.value if isinstance(mode, RateControlMode) else str(mode)
    await self._apply_display_setting(display_id or "primary", "rate_control_mode", mode_str)

async def handle_crf_change(self, crf: int, display_id: str = "primary") -> None:
    """CRF change for the display whose page sent it; sanitized against the
    server's configured range like the SETTINGS path."""
    sanitized = sanitize_client_setting("video_crf", int(crf), self.settings, logger)
    if sanitized is None or sanitized == self._display_setting(display_id, "video_crf"):
        return
    await self._apply_display_setting(display_id or "primary", "video_crf", sanitized)

async def handle_client_werbtc_stats(
    self, webrtc_stat_type: str, webrtc_stats: str
) -> None:
    """Ingest client-reported WebRTC stats, gated on the Metrics object
    itself (built for metrics-http AND/OR the CSV statistics flag) so
    CSV-only configs actually ingest the stats they enabled."""
    if self.metrics:
        await self.metrics.set_webrtc_stats(webrtc_stat_type, webrtc_stats)

async def on_resize_handler(self, res: str, display_id: str = "primary") -> None:
    """Route a client resolution to its display: the primary resizes the real
    display directly while it is alone; once a secondary display is connected
    (or for any secondary), the resolution feeds the extended-desktop layout
    instead (websockets parity).

    The layout path honors an admin manual-resolution lock the way the
    single-display path does: every display follows the server's geometry,
    so a second screen cannot be the way a client escapes the lock, and a
    locked size is the server's own, beyond a client alignment toggle.
    """
    display_id = display_id or "primary"
    if display_id == "primary" and not self.args.enable_resize:
        logger.warning(f"remote resizing disabled, skipping resize to {res}")
        return
    if display_id != "primary" or self.display_clients:
        locked_dims = self._server_locked_dims()
        if locked_dims is not None:
            logger.warning(
                f"Client attempted to resize to {res} but server is in manual resolution mode. "
                f"Using the configured {locked_dims[0]}x{locked_dims[1]} instead."
            )
            w, h = locked_dims
        else:
            dims = parse_resize_dims(res)
            if dims is None:
                logger.error(f"Invalid resize request: {res}")
                return
            w, h = dims
            if self._display_setting(display_id, "force_aligned_resolution"):
                w, h = align_dims_16(w, h)
        if display_id == "primary":
            self._primary_dims = (w, h)
        else:
            entry = self.display_clients.get(display_id)
            if entry is None:
                logger.warning(f"Resize for unknown display '{display_id}' ignored.")
                return
            entry["width"], entry["height"] = w, h
        await self.reconfigure_displays()
        return
    self._primary_dims = None
    await self._resize_primary_display(res)

def _server_locked_dims(self) -> Optional[Tuple[int, int]]:
    """The geometry an admin-configured manual-resolution lock pins the desktop
    to, or None when the server sets no lock. Derived on every read from the
    server settings (never from the client-writable args, whose manual trio is
    the client's own manual/auto toggle) and from the dimensions startup
    realized, so the lock cannot drift with client state. The settings layer
    guarantees positive manual dimensions while the lock is on; its own
    defaults stand in should one be unusable, so a locked server never
    falls back to honoring the client's request."""
    server_is_manual, _ = self.settings.manual_resolution
    if not server_is_manual:
        return None
    if self._manual_dims:
        return self._manual_dims
    width = int(getattr(self.settings, "manual_width", 0) or 0)
    height = int(getattr(self.settings, "manual_height", 0) or 0)
    if width <= 0:
        width = 1024
    if height <= 0:
        height = 768
    return (width - (width % 2), height - (height % 2))

async def _resize_primary_display(self, res: str) -> None:
    """Resize the single (primary-only) display to a client-requested
    resolution, keep the capture dimensions in sync with what was realized,
    and tell the client the realized size when it differs.

    Only an admin-configured manual-resolution lock blocks client resizes
    (websockets parity); the client's own manual/auto toggle in `args`
    must not gate here, since in client manual mode the chosen resolution
    arrives through this same path. Idempotent: clients re-assert their
    resolution on reconnects and settings broadcasts, and re-applying the
    current size would churn RandR (X11) or restart the capture (Wayland)
    for nothing; the last request counts as applied too, or a request the
    realized size differs from (CVT cell alignment) would read as pending
    forever. On Wayland there is no X server to resize: the compositor
    output follows the capture dimensions, so a running capture restarts
    and the compositor's realized geometry (it may even-mask or refuse the
    mode) is reconciled and pushed to the client.
    """
    if self._server_locked_dims() is not None:
        logger.warning(
            f"Client attempted to resize to {res} but server is in manual resolution mode. Request ignored."
        )
        return
    try:
        dims = parse_resize_dims(res)
        if dims is None:
            logger.error(f"Invalid resize request: {res}. Ignoring")
            if self.media_pipeline:
                self.media_pipeline.last_resize_success = False
            return
        target_w, target_h = dims
        if getattr(self.args, "force_aligned_resolution", False):
            target_w, target_h = align_dims_16(target_w, target_h)

        if (
            self.media_pipeline
            and self.media_pipeline.last_resize_success
            and (
                (self.media_pipeline.width == target_w
                 and self.media_pipeline.height == target_h)
                or self._last_resize_request == (target_w, target_h)
            )
        ):
            logger.debug(f"Resolution already {target_w}x{target_h}; skipping re-apply.")
            return

        if IS_WAYLAND:
            self.media_pipeline.width = target_w
            self.media_pipeline.height = target_h
            # The capture, not the whole pipeline: the session's first request
            # can land while audio is still coming up. An unstarted capture
            # reads the new size when it starts.
            if self.media_pipeline.is_screen_capturing():
                await self.media_pipeline.restart_screen_capture()
                await self._push_wayland_realized_geometry("primary", self.media_pipeline)
            self.media_pipeline.last_resize_success = True
            self._last_resize_request = (target_w, target_h)
            logger.info(
                f"Wayland capture resized to {self.media_pipeline.width}x{self.media_pipeline.height}"
                f" (requested {target_w}x{target_h})"
            )
            return

        realized = await resize_display(f"{target_w}x{target_h}")
        if realized:
            realized_w, realized_h = realized
            if (realized_w, realized_h) != (target_w, target_h):
                logger.info(
                    f"resize_display realized {realized_w}x{realized_h} for request {target_w}x{target_h}"
                )
            else:
                logger.info(f"resize_display('{target_w}x{target_h}') reported success")
            # A zero-size region re-reads the live root now and keeps root-follow;
            # the auto-adjust poll trails ~30 frames, leaving new bands out of frame.
            capture_module = getattr(self.media_pipeline, "capture_module", None)
            if capture_module is not None:
                try:
                    await asyncio.to_thread(
                        capture_module.update_capture_region, 0, 0, 0, 0
                    )
                except Exception as e:
                    logger.warning(f"Capture re-follow after resize failed: {e}")
            self.media_pipeline.width = realized_w
            self.media_pipeline.height = realized_h
            self.media_pipeline.last_resize_success = True
            self._last_resize_request = (target_w, target_h)
            if self.rtc_app is not None:
                self.rtc_app.send_remote_resolution(f"{realized_w}x{realized_h}", "primary")
        else:
            logger.error(
                f"resize_display('{target_w}x{target_h}') reported failure"
            )
            self.media_pipeline.last_resize_success = False

    except Exception as e:
        logger.error(
            f"Error during resize handling for '{res}': {e}", exc_info=True
        )
        if self.media_pipeline:
            self.media_pipeline.last_resize_success = False

async def request_idr_for_display(self, display_id: str = "primary") -> None:
    """Schedule a dynamic IDR frame on the display's encoder, throttled by a
    per-display floor (websockets REQUEST_KEYFRAME parity).

    Any number of viewers share one encoder, and an unthrottled
    data-channel request or PLI storm would let a single client force
    keyframe bursts for every consumer. A request landing inside the floor
    is satisfied by the IDR the previous request already scheduled.
    """
    display_id = display_id or "primary"
    now = time.monotonic()
    if now - self._last_idr_request_times.get(display_id, 0.0) < 0.25:
        return
    self._last_idr_request_times[display_id] = now
    pipeline = self.display_pipelines.get(display_id)
    if pipeline is not None:
        await pipeline.dynamic_idr_frame()

async def _provision_webrtc_virtual_mic(self) -> None:
    """Bring up the SelkiesVirtualMic once for the WebRTC transport (shared
    provisioning with the websockets 0x02 path). Called from the per-peer mic
    playback start on the first mic packet; the lock + flag make concurrent
    first-packet calls across peers provision exactly once, and the shared
    helper reuses a source the websockets path already loaded rather than
    double-loading it."""
    if self._mic_provisioned:
        return
    async with self._mic_provision_lock:
        if self._mic_provisioned:
            return
        if self._mic_control is None:
            self._mic_control = AudioControl("selkies-webrtc-mic")
        audio_device_name = getattr(self.media_pipeline, "audio_device_name", None)
        is_capturing = bool(getattr(self.media_pipeline, "_is_pcmflux_capturing", False))
        self._mic_module_index, self._mic_module_owned = (
            await self._mic_control.ensure_virtual_microphone(audio_device_name, is_capturing)
        )
        self._mic_provisioned = self._mic_module_index is not None

async def _teardown_webrtc_virtual_mic(self) -> None:
    """Unload the virtual-source module (only if this path loaded it) and
    release the mic control connection on shutdown."""
    control = self._mic_control
    self._mic_control = None
    if control is None:
        return
    if self._mic_module_index is not None and self._mic_module_owned:
        logger.info(f"Unloading WebRTC virtual mic module {self._mic_module_index}.")
        await control.unload_module(self._mic_module_index)
    self._mic_module_index = None
    self._mic_module_owned = False
    self._mic_provisioned = False
    await control.aclose()

async def start_display_media(self, display_id: str) -> None:
    """A display's consumer connected: the primary starts its pipeline right
    away; a secondary waits for its dimensions (the client's first resize
    message), which trigger the layout pass that creates its pipeline.

    A consumer reclaiming the primary cancels a pending grace stop, and
    `start_media_pipeline` is idempotent, so a controller tab reload that
    reconnects inside the grace reuses the still-warm capture. A Wayland
    start only enqueues a compositor command, so its real outcome is read
    back through the same barrier the secondaries use: a pipeline that
    believes it is running with no live capture would leave the page
    waiting on frames that never arrive, so it is stopped and logged and
    the next consumer retries (this also surfaces host death, where
    `reap_dead_host` flips `is_capturing` off). The pipeline start is what
    establishes the host session in host-capture mode, so the host's
    output count — and the second-screen availability announced on
    channel open — can first become known here and is re-published.
    """
    if display_id == "primary" and self.media_pipeline:
        self._cancel_primary_stop_grace()
        await self.media_pipeline.start_media_pipeline()
        if (IS_WAYLAND and self.media_pipeline.is_media_pipeline_running()
                and not await self._wayland_capture_live("primary", self.media_pipeline)):
            last_error = self._wayland_capture_last_error(self.media_pipeline, "primary")
            logger.error(
                "Primary Wayland capture is not live after start"
                + (f": {last_error}." if last_error else "."))
            await self.media_pipeline.stop_media_pipeline()
            return
        caveat = (self._wayland_capture_last_error(self.media_pipeline, "primary")
                  if IS_WAYLAND else None)
        if caveat:
            logger.warning(f"Primary Wayland capture started with a caveat: {caveat}")
        if await self._refresh_second_screen_capacity() and self.rtc_app:
            self.rtc_app.send_media_data_over_channel(
                "server_settings", self._server_settings_payload()
            )

async def stop_display_media(self, display_id: str) -> None:
    """Release a display's pipeline: the primary's stop is deferred by a
    reconnect grace (a controller tab reload reconnects within a second or
    two and reuses the warm capture, and viewers/display2 keep streaming
    throughout — websockets _teardown_if_unclaimed parity); a secondary is
    fully unregistered and the desktop re-laid-out at once."""
    if display_id == "primary":
        self._schedule_primary_stop_grace()
        return
    async with self._display_lock:
        pipeline = self.display_pipelines.pop(display_id, None)
        self.display_clients.pop(display_id, None)
        self.display_layouts.pop(display_id, None)
        if pipeline is not None:
            await pipeline.stop_media_pipeline()
    await self.reconfigure_displays()

def _cancel_primary_stop_grace(self) -> None:
    """Drop a pending deferred primary-capture stop: a consumer reclaimed
    the display before the grace elapsed."""
    task = self._primary_stop_grace_task
    self._primary_stop_grace_task = None
    if task is not None and not task.done():
        task.cancel()

def _schedule_primary_stop_grace(self) -> None:
    """Stop the primary capture after RECONNECT_GRACE_S unless a consumer
    reclaims it first. A page reload drops and re-adds its peer within the
    window, so tearing the capture down immediately would black out a
    reconnecting controller (and stall the audio fan-out the viewers share)
    for no reason; if nobody reclaims the primary, the stop runs after the
    grace. At most one grace is pending at a time."""
    if self._primary_stop_grace_task is not None and not self._primary_stop_grace_task.done():
        return
    if self.media_pipeline is None or not self.media_pipeline.is_media_pipeline_running():
        return

    async def _stop_after_grace() -> None:
        """Stop the primary unless reclaimed. No lock: the event loop
        serializes start/stop with this coroutine's resume."""
        try:
            await asyncio.sleep(self.RECONNECT_GRACE_S)
        except asyncio.CancelledError:
            return
        self._primary_stop_grace_task = None
        if self._primary_display_has_consumer():
            logger.info("Primary reclaimed within the grace; capture kept.")
            return
        if self.media_pipeline is not None:
            logger.info("Primary unclaimed after the grace; stopping its capture.")
            await self.media_pipeline.stop_media_pipeline()

    self._primary_stop_grace_task = asyncio.create_task(_stop_after_grace())

def _primary_display_has_consumer(self) -> bool:
    """Whether any peer (controller or viewer) still consumes the primary."""
    if self.rtc_app is None:
        return False
    return any(
        (p.get("display_id") or "primary") == "primary"
        for p in self.rtc_app.peer_connections.values()
    )

def _wayland_capture_handle(self) -> Optional[Any]:
    """A pixelflux handle for compositor output management (any ScreenCapture
    reaches the shared Wayland backend); prefers the primary pipeline's live
    capture module."""
    module = getattr(self.media_pipeline, "capture_module", None)
    if module is not None:
        return module
    if PixelfluxScreenCapture is None:
        return None
    if self._wayland_ctl_module is None:
        self._wayland_ctl_module = PixelfluxScreenCapture()
    return self._wayland_ctl_module

async def _destroy_wayland_secondary_outputs(self, keep_oid: Optional[int] = None) -> None:
    """Retire every secondary compositor output except `keep_oid` (the
    primary, output 0, always persists)."""
    module = self._wayland_capture_handle()
    if module is None:
        return
    try:
        for out in await asyncio.to_thread(module.list_outputs):
            if out[0] != 0 and out[0] != keep_oid:
                await asyncio.to_thread(module.destroy_output, out[0])
    except Exception as e:
        logger.warning(f"Wayland output teardown failed: {e}")

async def _apply_wayland_extension(self, did: str, layouts: Dict[str, Dict[str, int]]) -> bool:
    """Realize the extended layout as compositor outputs, BEFORE the
    secondary's pipeline binds a capture — the Wayland counterpart of
    apply_extended_layout.

    The primary (output 0) MOVES to its layout offset ('left'/'up' place it
    off-origin); a secondary reposition is a destroy + recreate (its
    capture rebinds on the pipeline restart that follows), destroyed before
    the primary moves so the rectangles never overlap.

    Returns:
        False when the output cannot be created or the primary cannot move
        (the caller drops the display).
    """
    module = self._wayland_capture_handle()
    if module is None:
        return False
    oid = wayland_output_id(did)
    s = layouts[did]
    scale = float(getattr(self.media_pipeline, "scale", 1.0) or 1.0)
    try:
        outputs = {o[0]: o for o in await asyncio.to_thread(module.list_outputs)}
    except Exception as e:
        logger.error(f"Wayland list_outputs failed: {e}")
        outputs = {}
    await self._destroy_wayland_secondary_outputs(keep_oid=oid)
    existing = outputs.get(oid)
    if existing is not None and (existing[1], existing[2]) != (s["x"], s["y"]):
        logger.info(f"Wayland output {oid} moves to +{s['x']}+{s['y']}; recreating it.")
        await asyncio.to_thread(module.destroy_output, oid)
        existing = None
    p = layouts.get("primary") or {"x": 0, "y": 0}
    existing0 = outputs.get(0)
    current0 = (existing0[1], existing0[2]) if existing0 is not None else (0, 0)
    if (p["x"], p["y"]) != current0:
        if not await wayland_reposition_primary(module, p["x"], p["y"]):
            return False
    if existing is not None:
        return True
    try:
        created = bool(await asyncio.to_thread(
            module.create_output, oid, s["w"], s["h"], s["x"], s["y"], scale))
    except Exception as e:
        logger.error(f"Wayland create_output {oid} failed: {e}")
        return False
    if created and self.input_handler:
        # Which of the session's own screens a capture drives changed.
        self.input_handler.resync_session_screens()
    return created

async def _wayland_capture_live(self, did: str, pipeline: MediaPipelinePixel) -> bool:
    """Whether the display's capture really runs in the compositor. The
    geometry read is a barrier: it is answered only after the queued capture
    start finished, so is_capturing is authoritative afterwards."""
    module = getattr(pipeline, "capture_module", None)
    if module is None:
        return False
    try:
        await asyncio.to_thread(module.get_realized_geometry, wayland_output_id(did))
        return bool(module.is_capturing)
    except Exception:
        return False

def _wayland_capture_last_error(
    self, pipeline: Optional[MediaPipelinePixel], did: str
) -> Optional[str]:
    """The reason a display's Wayland capture failed, or a caveat a live one came up
    with (encoder fell back to CPU, host connect refused), or None. Read straight from
    ``capture_state`` (no barrier); the caller ensures ordering. None on an older
    pixelflux without the readback."""
    module = getattr(pipeline, "capture_module", None) if pipeline is not None else None
    getter = getattr(module, "capture_state", None) if module is not None else None
    if getter is None:
        return None
    try:
        _state, last_error = getter(wayland_output_id(did))
        return last_error
    except Exception:
        return None

async def _realized_wayland_dims(self, did: str) -> Optional[Tuple[int, int]]:
    """The ``(width, height)`` the compositor currently has for this
    display's output, or None when it cannot be read."""
    if not IS_WAYLAND:
        return None
    pipeline = (self.media_pipeline if did == "primary"
                else self.display_pipelines.get(did))
    module = getattr(pipeline, "capture_module", None)
    if module is None or not hasattr(module, "get_realized_geometry"):
        return None
    try:
        geom = await asyncio.to_thread(
            module.get_realized_geometry, wayland_output_id(did))
    except Exception as e:
        logger.warning(f"Wayland realized-geometry read failed for '{did}': {e}")
        return None
    if geom is None:
        logger.warning(f"Wayland realized-geometry read for '{did}' timed out; size unknown.")
        return None
    w, h, _scale = geom
    return (w, h) if w > 0 and h > 0 else None

async def _push_wayland_realized_geometry(
    self, did: str, pipeline: Optional[MediaPipelinePixel]
) -> None:
    """Read what the pixelflux compositor actually realized on this
    display's output after a capture (re)start (it may even-mask dimensions
    or keep the old mode on a GBM allocation failure), fold it into the
    pipeline/layout state the input math offsets against, and push the
    corrected size to the clients over the existing system resolution
    message — the WR counterpart of the WS realized clamp + broadcast. The
    stream itself re-negotiates through the encoder (the track's intrinsic
    size IS the realized resolution); this closes the control-plane loop.
    The read is also a barrier: the compositor answers only after the
    queued capture (re)start finished. The push is unconditional
    (idempotent, WS-broadcast parity: the client's request may have been
    snapped by sanitization before the pipeline saw it) and scoped to this
    display's channels so a secondary's size never rescales the primary
    page."""
    if not IS_WAYLAND or pipeline is None:
        return
    module = getattr(pipeline, "capture_module", None)
    if module is None or not hasattr(module, "get_realized_geometry"):
        return
    try:
        geom = await asyncio.to_thread(
            module.get_realized_geometry, wayland_output_id(did))
    except Exception as e:
        logger.warning(f"Wayland realized-geometry read failed for '{did}': {e}")
        return
    if geom is None:
        # A timeout is unknown geometry, not "nothing to reconcile".
        logger.warning(f"Wayland realized-geometry read for '{did}' timed out; state left unreconciled.")
        return
    w, h, scale = geom
    if w <= 0 or h <= 0:
        return
    pipeline.width, pipeline.height = w, h
    if did == "primary":
        if self._primary_dims is not None:
            self._primary_dims = (w, h)
    else:
        entry = self.display_clients.get(did)
        if entry is not None:
            entry["width"], entry["height"] = w, h
    layout = self.display_layouts.get(did)
    if layout is not None:
        layout["w"], layout["h"] = w, h
    logger.info(f"Wayland realized geometry for '{did}': {w}x{h} @ scale {scale}")
    if self.rtc_app is not None:
        self.rtc_app.send_remote_resolution(f"{w}x{h}", did)

async def _apply_wayland_cursor_size(self, dpi_value: float) -> None:
    """Wayland counterpart of the X11 per-DPI cursor resize: the compositor
    reloads its theme cursor (composited overlay and named-cursor delivery
    both re-render) at the DPI-scaled size, live, no capture restart."""
    if CURSOR_SIZE is None:
        return
    module = self._wayland_capture_handle()
    setter = getattr(module, "set_cursor_size", None) if module else None
    if setter is None:
        logger.warning("Wayland cursor resize unavailable (no set_cursor_size).")
        return
    size = cursor_size_for_dpi(dpi_value, CURSOR_SIZE)
    try:
        if await asyncio.to_thread(setter, size):
            logger.info(f"Wayland cursor size set to {size} (DPI {dpi_value}).")
        else:
            logger.warning(f"Wayland compositor refused cursor size {size}.")
    except Exception as e:
        logger.warning(f"Wayland cursor resize failed: {e}")

def _display_consumers(self, display_id: str) -> List[Dict[str, Any]]:
    """Registered peers (controller + viewers) consuming this display's
    stream."""
    if self.rtc_app is None:
        return []
    return [
        p for p in self.rtc_app.peer_connections.values()
        if (p.get("display_id") or "primary") == display_id
    ]

async def handle_peer_gone(
    self, peer_id: str, peer: Optional[Dict[str, Any]] = None
) -> None:
    """Release the input a departing peer may still be holding, matching the
    websockets disconnect cleanup. Gamepad slots are per-connection, so they
    are always released. Held keys and pointer buttons are one global desktop
    state instead: they are force-released only when the departing peer could
    drive input and no input-capable peer is left, so a viewer (or a second
    display's peer) leaving never drops the controller's held keys or its
    in-flight drag. A controller that vanishes while others remain is covered
    by the input handler's heartbeat stale-sweep."""
    if self.input_handler is None:
        return
    try:
        await self.input_handler.release_gamepads_for_conn(peer_id)
    except Exception as e:
        logger.warning(f"Gamepad release for departed peer {peer_id} failed: {e}")
    if self.rtc_app is None or not self.rtc_app.peer_holds_input_authority(peer):
        return
    if (peer.get("display_id") or "primary") != "primary":
        # The departing peer is already out of peer_connections: survivors only.
        for survivor in list(self.rtc_app.peer_connections.values()):
            if self.rtc_app.peer_holds_input_authority(survivor):
                return
    for release in (self.input_handler.release_mouse_buttons,
                    self.input_handler.reset_keyboard):
        try:
            await release()
        except Exception as e:
            logger.warning(f"Input release for departed peer {peer_id} failed: {e}")