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.resource_monitor: Optional[resource_stats.ResourceMonitor] = 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]] = {}
    # Each page's reported CSS-to-remote scale, by display id (the primary
    # included), rebroadcast with the layout for cross-display drags.
    self._client_scales: Dict[str, float] = {}
    # Each page's stream box on the user's desktop, by display id: origin
    # and remote pixels per desktop pixel, with when it last changed.
    self._client_stream_boxes: Dict[
        str, Tuple[Tuple[float, float, float, float], float]] = {}
    self.display_pipelines: Dict[str, MediaPipelinePixel] = {}
    # The DPI each secondary's page asked for; the primary's is _last_applied_dpi.
    self._display_dpis: Dict[str, int] = {}
    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._congestion_steer: Dict[str, CongestionSteer] = {}
    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
    await self.rtc_app.open_ice_muxes()
    if self.rtc_app.ice_lite_enabled():
        logger.info("WebRTC ICE-lite: the server offers host candidates only and answers the client's checks; STUN/TURN serve the client side")
    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)

    # Keyed to the pipeline's render node so the GPU stats describe the encoding GPU.
    stats_gpu_id = parse_gpu_id(getattr(self.args, "gpu_id", ""))
    self.resource_monitor = resource_stats.ResourceMonitor(
        gpu_id=stats_gpu_id if (stats_gpu_id or 0) > 0 else 0,
        dri_node=getattr(self.args, "encode_dri", "") or "",
        metrics=self.metrics,
    )

    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}//localhost:{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",
    fullcolor_codecs: Optional[List[str]] = None,
) -> 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").
    """
    # From the registry, not SESSION_START: the signaling server that validated
    # the claim is this process, and the relay's fields are positional.
    peer = self.peer_manager.peers.get(session_peer_id) if self.peer_manager else None
    client_slot = getattr(peer, "client_slot", None) if peer else None
    logger.debug(
        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, client_slot,
            fullcolor_codecs=fullcolor_codecs)
        if self.args.enable_webrtc_statistics and self.metrics:
            await self.metrics.initialize_webrtc_csv_file(self.args.webrtc_statistics_dir)
        logger.info(f"Session started for peer {session_peer_id} ({client_type}, display '{display_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-color 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_encoder_demoted = (
        lambda encoder: asyncio.ensure_future(self._encoder_demoted("primary", encoder))
    )
    self.media_pipeline.on_stream_info = self._publish_stream_info
    self.media_pipeline.on_pipeline_started = self.send_current_cursor

    self.rtc_app.request_idr_frame = self.request_idr_for_display
    self.rtc_app.invalidate_reference = self.invalidate_reference_for_display
    self._invalidation_log: Dict[str, tuple] = {}
    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.on_video_codec_declined = self._video_codec_declined
    self.rtc_app.on_fullcolor_declined = self._fullcolor_declined
    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_audio_consumer_active = self.handle_audio_consumer_active
    self.rtc_app.on_consumers_changed = self.handle_consumers_changed
    # /api/tokens updates must reach live WebRTC peers too.
    sessions.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.input_handler.on_session_compositor_adopted = self._resync_wayland_session_scale

    self.resource_monitor.on_tick = self.handle_resource_tick
    self.resource_monitor.watched = lambda: bool(self.rtc_app and self.rtc_app.stats_displays())

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 mints a RandR monitor on
    demand; host capture is bounded by the host compositor's real output
    count (unknown until a pipeline start establishes the host session);
    and the self-composited Wayland backend rides the
    input handler's session-screen ladder: the session compositor's
    control socket grows a screen on demand, a spare screen the session
    already opened is arranged instead without one, and a session running
    directly on the capture compositor needs neither.

    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:
        return True, ""
    if (self.settings.wayland_host_display or "").strip():
        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, ""
    if self.input_handler is None:
        return False, "The input system is not up yet."
    return self.input_handler.session_screen_capability()

async def _refresh_second_screen_capacity(self) -> bool:
    """Re-read what bounds a second display on this backend.

    Host capture re-reads how many outputs the host exposes; the nested
    Wayland backend re-probes the session compositor's control socket.
    X11 has no bound.

    Returns:
        True when the answer changed, i.e. the second-screen availability
        that clients were told may have flipped.
    """
    if not IS_WAYLAND:
        return False
    if not (self.settings.wayland_host_display or "").strip():
        if self.input_handler is None:
            return False
        before = self.input_handler.session_screen_capability()[0]
        fresh = await self.input_handler.probe_session_screen_capability()
        return fresh[0] != before
    module = self._wayland_capture_handle()
    if module is None:
        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), and what the apps panel already has
    installed, which no browser's own storage can answer for a session
    opened somewhere else — absent until the runner has answered, because a
    client told nothing is installed clears its own record."""
    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)
    installed = self.input_handler.installed_apps() if self.input_handler else None
    if installed is not None:
        payload["settings"]["apps_installed"] = {"value": installed}
    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("Peer data channel open for input.")
    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
        )
        self.rtc_app.send_message_to_channel(
            channel, "display_config_update", self._display_config_payload()
        )
        peer = next((obj for obj in self.rtc_app.peer_connections.values()
                     if obj.get("data_channel") is channel), None)
        if peer is not None and peer.get("client_type") == ClientType.CONTROLLER \
                and (peer.get("display_id") or "primary") == "primary":
            for name, size in self.supervisor.pending_print_documents():
                self.rtc_app.send_print_document(name, size, channel)
        if peer is not None and peer.get("client_type") == ClientType.CONTROLLER:
            did = peer.get("display_id") or "primary"
            settled = getattr(getattr(self.display_pipelines.get(did), "stream_watch", None), "info", None)
            if settled:
                self.rtc_app.send_stream_info(did, settled, channel)
    else:
        self.rtc_app.send_media_data_over_channel(
            "server_settings", server_settings_payload
        )
        self._broadcast_display_config()
    self.send_current_cursor(channel)

async def announce_print_document(self, name: str, size: int) -> None:
    """Tell every controller peer a printed document waits in the spool."""
    if self.rtc_app:
        self.rtc_app.send_print_document(name, size)

async def sessions(self) -> List[Dict[str, Any]]:
    return await self.rtc_app.sessions() if self.rtc_app else []

async def disconnect_session(self, session_id: str) -> bool:
    return bool(self.rtc_app) and await self.rtc_app.disconnect_peer(session_id)

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"
    logger.debug(f"Resize message for display '{display_id}': {res}")
    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 screen is
    grown ahead of the capture restart, the restarted capture resizes the
    view, the compositor's realized geometry (it may even-mask or refuse
    the mode) is reconciled and pushed to the client, the screen is fitted
    to it, and a nested session's own screen is re-sized to the same
    geometry at the DPI in force.
    """
    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"Redundant resize request for primary to {target_w}x{target_h}. No action.")
            return
        logger.info(f"Resize requested for display 'primary' with resolution: {target_w}x{target_h}")

        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._size_wayland_screen(target_w, target_h, grow_only=True)
                await self.media_pipeline.restart_screen_capture()
                await self._push_wayland_realized_geometry("primary", self.media_pipeline)
                await self._size_wayland_screen(
                    self.media_pipeline.width, self.media_pipeline.height)
                # A nested session's screen is its own compositor's, not the
                # capture's: sizing only the capture leaves its applications
                # laid out for the size the last DPI change realized.
                if self.input_handler is not None:
                    await self.input_handler.realize_wayland_dpi(
                        getattr(self, "_last_applied_dpi", None)
                        or getattr(settings, "scaling_dpi", 96) or 96,
                        "primary",
                        (self.media_pipeline.width, self.media_pipeline.height))
            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.debug(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",
                                  unless_pending: bool = False) -> 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.

    Args:
        unless_pending: Hold the request only while that IDR has yet to be
            captured, whatever the floor says: a pacer's GOP reset needs a
            keyframe from after the reset, and one captured before it
            leaves the stream dead until the pacer's resurrect timeout.
    """
    display_id = display_id or "primary"
    pipeline = self.display_pipelines.get(display_id)
    if pipeline is None:
        return
    now = time.monotonic()
    if unless_pending:
        if pipeline.idr_pending:
            return
    elif now - self._last_idr_request_times.get(display_id, 0.0) < IDR_REQUEST_FLOOR_S:
        return
    self._last_idr_request_times[display_id] = now
    await pipeline.dynamic_idr_frame()

def invalidate_reference_for_display(self, display_id: str, frame_id: int) -> None:
    """Tell the display's encoder a peer lost `frame_id`, so the frames after it stop
    predicting from it (websockets LOST_FRAME parity). Logged once per display per
    five seconds with the count of the rest, since loss comes in bursts."""
    display_id = display_id or "primary"
    pipeline = self.display_pipelines.get(display_id)
    if pipeline is None:
        return
    pipeline.invalidate_reference(frame_id)
    now = time.monotonic()
    last, more = self._invalidation_log.get(display_id, (0.0, 0))
    if now - last >= 5.0:
        suffix = f" (+{more} more in the last 5 s)" if more else ""
        logger.info(f"Display '{display_id}': frame {frame_id} lost by a peer; the encoder predicts past it.{suffix}")
        self._invalidation_log[display_id] = (now, 0)
    else:
        self._invalidation_log[display_id] = (last, more + 1)

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. The
    pipeline starts only the captures its consumers receive: a peer whose
    session policy starts video or audio off is registered paused, and a
    warm pipeline is settled against the consumer set afterwards, so a
    paused reconnect stops what it no longer receives. 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()
        consumers = self._display_consumers("primary")
        video_wanted = any(not p.get("video_paused", False) for p in consumers) or not consumers
        audio_wanted = any(not p.get("audio_paused", False) for p in consumers) or not consumers
        await self.media_pipeline.start_media_pipeline(video=video_wanted, audio=audio_wanted)
        await self._settle_primary_consumers()
        if (IS_WAYLAND and video_wanted 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_dpis.pop(display_id, None)
        self.display_layouts.pop(display_id, None)
        self._client_scales.pop(display_id, None)
        self._client_stream_boxes.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 _size_wayland_screen(self, width: int, height: int,
                               grow_only: bool = False) -> None:
    """Size the primary's screen (output 0) to its display rectangle.

    The primary's capture binds to the view covering that screen, and a
    capture start sizes the view alone, which the screen must already hold:
    so the screen is grown ahead of the capture restart and fitted to the
    rectangle once the capture carries it -- an early shrink is refused by
    the compositor, which leaves no view hanging outside its screen."""
    module = self._wayland_capture_handle()
    if module is None or width <= 0 or height <= 0:
        return
    scale = float(getattr(self.media_pipeline, "scale", 1.0) or 1.0)
    try:
        if grow_only:
            outputs = {o[0]: o for o in await asyncio.to_thread(module.list_outputs)}
            screen = outputs.get(WAYLAND_SCREEN_OUTPUT_ID)
            if screen:
                width, height = max(width, screen[3]), max(height, screen[4])
        ok = await asyncio.to_thread(
            module.resize_output, WAYLAND_SCREEN_OUTPUT_ID, width, height, scale)
        if not ok:
            logger.warning(f"Wayland screen resize to {width}x{height} refused.")
    except Exception as e:
        logger.error(f"Wayland resize_output failed: {e}")

async def _destroy_wayland_secondary_outputs(self, keep_did: Optional[str] = None) -> None:
    """Retire every secondary display's compositor screen except `keep_did`'s.

    The primary's screen (output 0) and the view its capture binds to
    always persist -- the primary is the session, not an extension of it.
    """
    module = self._wayland_capture_handle()
    if module is None:
        return
    keep = {WAYLAND_SCREEN_OUTPUT_ID, wayland_output_id("primary")}
    if keep_did:
        keep.add(wayland_output_id(keep_did))
    try:
        for out in await asyncio.to_thread(module.list_outputs):
            if out[0] not in keep:
                await asyncio.to_thread(module.destroy_output, out[0])
    except Exception as e:
        logger.warning(f"Wayland output teardown failed: {e}")
    if self.input_handler:
        # The session keeps a screen per surviving display: the kept
        # secondary's while it is being (re)placed, only the primary's on
        # teardown, where a removed screen's windows return to the primary.
        await self.input_handler.ensure_session_screens(
            [keep_did] if keep_did else [])
        # Which of the session's own screens a capture drives changed.
        self.input_handler.resync_session_screens()

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

    Every display is a screen of the session compositor's own, so it lays
    its windows and panels out per monitor. 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. A secondary that keeps its origin but shrinks
    gives the room up in place ahead of the move, and grows into the room
    the move leaves on the same restart; a retained output the compositor
    still finds in the primary's way is recreated after all.

    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]
    dpi = self._display_dpi(did)
    scale = float(dpi) / 96.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_did=did)
    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
    elif (existing is not None
          and not await wayland_shrink_output(module, existing, s["w"], s["h"])):
        logger.info(f"Wayland output {oid} cannot shrink to {s['w']}x{s['h']}; recreating it.")
        await asyncio.to_thread(module.destroy_output, oid)
        existing = None
    p = layouts.get("primary") or {"x": 0, "y": 0}
    existing0 = outputs.get(WAYLAND_SCREEN_OUTPUT_ID)
    current0 = (existing0[1], existing0[2]) if existing0 is not None else (0, 0)
    if (p["x"], p["y"]) != current0:
        moved = await wayland_reposition_primary(module, p["x"], p["y"])
        if not moved and existing is not None:
            logger.info(f"Wayland output {oid} blocks the primary's move; recreating it.")
            await asyncio.to_thread(module.destroy_output, oid)
            existing = None
            moved = await wayland_reposition_primary(module, p["x"], p["y"])
        if not moved:
            return False
    if existing is not None:
        return True
    pw, ph = p.get("w"), p.get("h")
    if (existing0 is not None and pw and ph
            and (pw < existing0[3] or ph < existing0[4])):
        # The secondary drops into the room a shrinking primary frees, and the
        # compositor refuses an output overlapping the primary's rectangle, as
        # it refuses a screen too small for the live view its capture holds:
        # the capture takes the new rectangle, then the screen gives the room
        # up, before the secondary is placed.
        if (self.media_pipeline.width, self.media_pipeline.height) != (pw, ph):
            await self.media_pipeline.update_capture_region(p["x"], p["y"], pw, ph)
            await self._push_wayland_realized_geometry("primary", self.media_pipeline)
        if not await asyncio.to_thread(
                module.resize_output, WAYLAND_SCREEN_OUTPUT_ID, pw, ph, existing0[5]):
            logger.warning(f"Wayland screen shrink to {pw}x{ph} refused.")
    if self.input_handler:
        # The screen this display owns, grown just ahead of the output
        # that adopts its host window, then given the display's own DPI;
        # what the session leaves is this output's capture scale.
        await self.input_handler.ensure_session_screen(
            did, size=(s["w"], s["h"]), scale=scale)
        scale = await self.input_handler.realize_wayland_dpi(dpi, did, (s["w"], s["h"]))
    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