RTCApp

def __init__(
    self,
    async_event_loop: asyncio.AbstractEventLoop,
    encoder: str,
    stun_servers: Optional[List[str]] = None,
    turn_servers: Optional[List[str]] = None
) -> None:
    self.peer_connections: Dict[str, Any] = {}
    self.async_event_loop = async_event_loop
    self.stun_servers = stun_servers
    self.turn_servers = turn_servers
    self.encoder = encoder
    self.last_cursor_sent = None

    self.displays: Dict[str, Dict[str, Any]] = {}
    self.media_pipeline: Optional[MediaPipeline] = None
    self.start_display_media = self._default_start_display_media
    self.stop_display_media = self._default_stop_display_media
    self.get_encoder_for_display = lambda display_id: self.encoder
    self.get_fullcolor_for_display = lambda display_id: bool(app_settings.video_fullcolor[0])
    self.get_use_cpu_for_display = lambda display_id: bool(app_settings.use_cpu[0])

    self.on_data_open = lambda channel=None: logger.warning('unhandled on_data_open')
    self.on_data_close = lambda: logger.warning('unhandled on_data_close')
    self.on_data_error = lambda e=None: logger.warning('unhandled on_data_error')
    self.on_data_message = lambda msg, display_id='primary', conn_id=None: logger.warning('unhandled on_data_message')
    self.on_peer_gone = None

    self.on_ice = lambda ice, client_peer_id: logger.warning('unhandled ice event')
    self.on_sdp = lambda sdp_type, sdp, client_peer_id: logger.warning('unhandled sdp event')

    self.request_idr_frame = lambda display_id='primary': logger.warning('unhandled request_idr_frame')

    self.on_video_consumer_active = None
    self.on_consumers_changed = None

    self.provision_virtual_mic = None

async def set_sdp(self, sdp_type: str, sdp: str, client_peer_id: str) -> None:
    """Apply the remote SDP answer received from a peer over signaling.

    Answers for peers already in a closed/failed state are ignored rather
    than raised: teardown routinely races late signaling messages.

    Args:
        sdp_type: Must be "answer"; the server always offers.
        sdp: The remote session description text.
        client_peer_id: Registered peer the answer belongs to.

    Raises:
        RTCAppError: On a non-answer type, missing sdp/peer id, or an
            unknown peer.
    """
    if sdp_type != 'answer':
        raise RTCAppError('ERROR: sdp type is not "answer"')
    if sdp is None:
        raise RTCAppError("ERROR: sdp can't be None")
    if not client_peer_id:
        raise RTCAppError("ERROR: client_peer_id is required to set sdp")

    peer_obj = self.peer_connections.get(client_peer_id, None)
    if peer_obj is None:
        raise RTCAppError(f"ERROR: peer connection for client_peer_id: {client_peer_id} not found")

    peer_conn = peer_obj["peer_conn"]
    if peer_conn.connectionState in ["closed", "failed"]:
        logger.warning(
            f"Ignoring remote SDP: peer connection in {peer_conn.connectionState} state",
            extra={'client_peer_id': client_peer_id, 'client_type': peer_obj.get('client_type')}
        )
        return

    desc = RTCSessionDescription(sdp=sdp, type=sdp_type)
    await peer_conn.setRemoteDescription(desc)

async def set_ice(self, ice: Dict, client_peer_id: str) -> None:
    """Add an ICE candidate received from the signaling server.

    An empty candidate string is the end-of-candidates marker and is
    forwarded as None.

    Args:
        ice: Candidate dict with `candidate` and `sdpMid` or
            `sdpMLineIndex` keys, as sent by the browser.
        client_peer_id: Registered peer the candidate belongs to.

    Raises:
        RTCAppError: On a missing peer id, an unknown peer, or a
            candidate that fails to parse.
    """
    if not client_peer_id:
        raise RTCAppError("ERROR: client_peer_id is required to set sdp")

    peer_obj = self.peer_connections.get(client_peer_id, None)
    if peer_obj is None:
        raise RTCAppError(f"ERROR: peer connection for client_peer_id: {client_peer_id} not found")

    peer_conn = peer_obj["peer_conn"]
    if peer_conn.connectionState in ["closed", "failed"]:
        logger.warning(
            f"Ignoring adding ICE candidate: peer connection in {peer_conn.connectionState} state",
            extra={'client_peer_id': client_peer_id, 'client_type': peer_obj.get('client_type')}
        )
        return

    if ice.get('candidate') == "":
        await peer_conn.addIceCandidate(None)
        return

    obj = Candidate.from_sdp(ice.get('candidate'))
    icecandidate = candidate_from_aioice(obj)

    sdp_mid = ice.get('sdpMid')
    if sdp_mid is not None:
        icecandidate.sdpMid = sdp_mid
    else:
        icecandidate.sdpMLineIndex = ice.get('sdpMLineIndex')

    if isinstance(icecandidate, RTCIceCandidate):
        await peer_conn.addIceCandidate(icecandidate)
    else:
        raise RTCAppError("ERROR: ice candidate is not an instance of RTCIceCandidate")

async def send_clipboard_data(self, data: Union[str, bytes], mime_type: str = "text/plain",
                              reply_to: Optional[str] = None,
                              peer_id: Optional[str] = None) -> None:
    """Send clipboard data over the data channel, chunked when large.

    Chunk sends are paced against each channel's SCTP queue (see
    `drain_data_channel`) so a multi-MB clipboard neither buffers
    unboundedly in memory nor starves input/cursor/stats behind it on the
    one ordered stream, and the per-chunk gzip runs off the event loop.
    Each channel runs its own pipeline a chunk at a time, so a slow peer
    paces only its own transfer; one that cannot drain a chunk within the
    bulk timeout is left out of the rest of this payload (it keeps its
    stream, and the client discards the partial transfer at the next
    start) instead of wedging the others. Chunk payloads are built once
    and shared by every channel; an entry is dropped once the slowest
    channel has passed it, so the cache holds what is in flight, not the
    whole payload. An empty payload is sent only as a tagged reply,
    settling a client fetch against an empty server clipboard.

    Args:
        data: Clipboard payload; str is UTF-8 encoded before sending.
        mime_type: Payload MIME type; "text/plain" marks it as text.
        reply_to: Set to the requesting verb (e.g. "cr") when this send
            answers a client fetch rather than announcing a server-side
            clipboard change — the websockets `clipboard_reply,<verb>`
            contract, carried here as an extra field on the clipboard-msg
            / clipboard-msg-start payload so the client can treat the
            payload cache-only without time heuristics (old clients ignore
            the unknown field).
        peer_id: Peer that asked for this payload, addressed alone the way
            `send_system_action` addresses a requester: the other peers
            already hold the content, and a reply they did not ask for is
            read as their own fetch and cached rather than pasted. A
            requester whose channel has closed receives nothing.
    """
    if not data and not reply_to:
        return

    is_text = mime_type == "text/plain"
    data_bytes: bytes = data.encode() if isinstance(data, str) else data
    clipboard_chunk_size = get_adjusted_chunk_size(self.peer_connections)
    requester = None
    if peer_id is not None:
        peer_obj = self.peer_connections.get(peer_id)
        channel = peer_obj.get("data_channel") if peer_obj else None
        if channel is None or channel.readyState != "open":
            return
        requester = channel

    def send_typed(msg_type: str, payload: Any) -> None:
        if requester is not None:
            self.send_message_to_channel(requester, msg_type, payload)
        else:
            self.__send_data_channel_message(msg_type, payload)

    if len(data_bytes) <= clipboard_chunk_size:
        b64data = base64.b64encode(data_bytes).decode('utf-8')
        payload = {
            "content": b64data,
            "mime_type": mime_type,
            "is_binary_data": not is_text,
            "total_size": len(data_bytes)
        }
        if reply_to:
            payload["reply_to"] = reply_to
        send_typed("clipboard-msg", payload)
    else:
        start_payload = {
            "mime_type": mime_type,
            "is_binary_data": not is_text,
            "total_size": len(data_bytes),
        }
        if reply_to:
            start_payload["reply_to"] = reply_to
        channels = ([requester] if requester is not None
                    else list(self._iter_open_data_channels()))
        offsets = list(range(0, len(data_bytes), clipboard_chunk_size))
        prepared: dict = {}
        prepare_lock = asyncio.Lock()
        progress = {id(c): 0 for c in channels}

        async def chunk_for(offset: int, want_gz: bool) -> tuple:
            async with prepare_lock:
                entry = prepared.get(offset)
                if entry is None:
                    chunk = data_bytes[offset:offset + clipboard_chunk_size]
                    entry = (json.dumps({
                        "type": "clipboard-msg-data",
                        "data": {"content": base64.b64encode(chunk).decode("utf-8")},
                    }), None)
                if want_gz and entry[1] is None:
                    entry = (entry[0], await asyncio.to_thread(
                        gzip.compress, entry[0].encode("utf-8"), 6))
                prepared[offset] = entry
                return entry

        def passed(channel: Any, offset: int) -> None:
            progress[id(channel)] = offset + clipboard_chunk_size
            floor = min(progress.values())
            for done in [o for o in prepared if o + clipboard_chunk_size <= floor]:
                prepared.pop(done, None)

        async def deliver(channel: Any) -> None:
            want_gz = bool(getattr(channel, "_selkies_gz_tx", False))
            self.send_message_to_channel(channel, "clipboard-msg-start", start_payload)
            for offset in offsets:
                if channel.readyState != "open":
                    progress.pop(id(channel), None)
                    return
                payload, gz_payload = await chunk_for(offset, want_gz)
                self._send_prepared_to_channel(
                    channel, "clipboard-msg-data", payload, gz_payload)
                if not await drain_data_channel(channel):
                    logger.warning(
                        "Data channel did not drain a clipboard chunk within "
                        f"{BULK_DRAIN_TIMEOUT_S:.0f}s; leaving it out of this payload.")
                    progress.pop(id(channel), None)
                    return
                passed(channel, offset)
            self.send_message_to_channel(channel, "clipboard-msg-end", {})

        await asyncio.gather(*(deliver(c) for c in channels), return_exceptions=True)

    logger.info(f"Sent clipboard data of length {len(data_bytes)} with mime type {mime_type}")

def send_cursor_data(self, data: Any) -> None:
    """Broadcast a cursor update, remembering it for late-joining peers."""
    self.last_cursor_sent = data
    self.__send_data_channel_message(
        "cursor", data)

def send_gpu_stats(self, load: float, memory_total: int, memory_used: int) -> None:
    """Broadcast GPU stats (load fraction, memory in MiB) to all peers."""

    self.__send_data_channel_message("gpu_stats", {
        "gpu_percent": load * 100,
        "mem_total": memory_total * 1024 * 1024,
        "mem_used": memory_used * 1024 * 1024,
    })

def send_system_action(self, action: str, peer_id: Optional[str] = None) -> None:
    """Send a system action (e.g. ``command_error,<text>``) to clients.

    With a `peer_id` whose channel is still open, only that peer is
    addressed (requester-scoped feedback); otherwise — including a
    requester that reconnected under a new peer id — the action is
    broadcast, and shared-mode viewers suppress it client-side.
    """
    if peer_id is not None:
        peer_obj = self.peer_connections.get(peer_id)
        channel = peer_obj.get("data_channel") if peer_obj else None
        if channel is not None and channel.readyState == "open":
            self.send_message_to_channel(channel, "system", {"action": action})
            return
    self.__send_data_channel_message("system", {"action": action})

def send_framerate(self, framerate: int) -> None:
    """Broadcast the current framerate to all peers."""
    logger.info("sending framerate")
    self.__send_data_channel_message(
        "system", {"action": "videoFramerate," + str(framerate)})

def send_video_bitrate(self, bitrate: int) -> None:
    """Broadcast the current video bitrate to all peers."""
    logger.info("sending video bitrate")
    self.__send_data_channel_message(
        "system", {"action": "video_bitrate," + str(bitrate)})

def send_audio_bitrate(self, bitrate: int) -> None:
    """Broadcast the current audio bitrate to all peers."""
    logger.info("sending audio bitrate")
    self.__send_data_channel_message(
        "system", {"action": "audio_bitrate,%d" % bitrate})

def send_encoder(self, encoder: str) -> None:
    """Broadcast the active encoder name to all peers."""
    logger.info("sending encoder: " + encoder)
    self.__send_data_channel_message(
        "system", {"action": "encoder,%s" % encoder})

def send_resize_enabled(self, resize_enabled: bool) -> None:
    """Broadcast the current resize-enabled state to all peers."""
    logger.info("sending resize enabled state")
    self.__send_data_channel_message(
        "system", {"action": "resize," + str(resize_enabled)})

def send_remote_resolution(self, res: str, display_id: str = "primary") -> None:
    """Send the realized remote resolution to the clients of `display_id`.

    Display-scoped: the websockets transport tags its stream_resolution
    with the display id and addresses only that page, so a secondary's
    realized size must never rescale the primary page.
    """
    logger.info("sending remote resolution of: " + res)
    sent = False
    for peer_obj in self.peer_connections.values():
        if (peer_obj.get("display_id") or "primary") != display_id:
            continue
        peer_conn = peer_obj.get("peer_conn")
        channel = peer_obj.get("data_channel")
        if (
            peer_conn is not None
            and channel is not None
            and peer_conn.connectionState == "connected"
            and channel.readyState == "open"
        ):
            self.send_message_to_channel(
                channel, "system", {"action": "resolution," + res})
            sent = True
    if not sent:
        logger.info("skipping remote resolution because no data channel is ready")

def send_ping(self, t: float) -> None:
    """Send a ping request to the PRIMARY controller only.

    Latency is measured against one shared ping_start, and the websockets
    transport likewise derives its reported latency from the primary
    client.
    """
    state, data_channel = self.get_data_channel()
    if not state:
        return
    self.send_message_to_channel(
        data_channel, "ping", {"start_time": float("%.3f" % t)})

def send_latency_time(self, latency: float) -> None:
    """Broadcast the measured latency response time in milliseconds."""
    self.__send_data_channel_message(
        "latency_measurement", {"latency_ms": latency})

def send_system_stats(self, cpu_percent: float, mem_total: int, mem_used: int) -> None:
    """Broadcast CPU and memory stats to all peers."""
    self.__send_data_channel_message(
        "system_stats", {
            "cpu_percent": cpu_percent,
            "mem_total": mem_total,
            "mem_used": mem_used,
        })

def get_data_channel(self) -> Tuple[bool, Optional[RTCDataChannel]]:
    """Return the controller's data channel and whether it is usable.

    Returns:
        A `(ready, channel)` pair: ready is True only when the controller's
        connection is connected and its channel open; channel is None when
        no controller exists.
    """
    state = False
    peer_obj = self.get_controller_instance()
    if not peer_obj:
        return state, None

    conn_state = peer_obj.get("peer_conn").connectionState
    data_channel_state = peer_obj.get("data_channel").readyState
    return conn_state == "connected" and data_channel_state == "open", peer_obj.get("data_channel")

def _iter_open_data_channels(self) -> Iterator[RTCDataChannel]:
    """Yield every connected peer's open data channel — controllers and
    viewers, all displays."""
    for peer_obj in self.peer_connections.values():
        peer_conn = peer_obj.get("peer_conn")
        channel = peer_obj.get("data_channel")
        if (
            peer_conn is not None
            and channel is not None
            and peer_conn.connectionState == "connected"
            and channel.readyState == "open"
        ):
            yield channel

def send_message_to_channel(self, channel: RTCDataChannel, msg_type: str,
                            data: Any) -> None:
    """Send one typed message to one specific peer's data channel.

    Payloads of 512 bytes or more (cursor PNGs, settings, clipboard, stats)
    are gzipped for a channel that completed the `_gz` handshake; smaller
    ones are not worth the CPU or the risk to input latency.
    """
    payload = json.dumps({"type": msg_type, "data": data})
    gz_payload = None
    if getattr(channel, "_selkies_gz_tx", False) and len(payload) >= 512:
        gz_payload = gzip.compress(payload.encode("utf-8"), 6)
    self._send_prepared_to_channel(channel, msg_type, payload, gz_payload)

def _send_prepared_to_channel(self, channel: RTCDataChannel, msg_type: str,
                              payload: str,
                              gz_payload: Optional[bytes]) -> None:
    """Guarded raw send of an already-serialized (and possibly
    pre-compressed) message; bulk senders reuse one compression across
    channels instead of re-gzipping per peer.

    A message oversized for the peer's negotiated max-message-size
    (`ValueError`) is dropped and logged, which beats the peer
    hard-closing the channel; one that finds the channel no longer open
    (`InvalidStateError`, a close racing the sender) is dropped like any
    not-ready channel's.
    """
    try:
        if gz_payload is not None and getattr(channel, "_selkies_gz_tx", False):
            channel.send(gz_payload)
        else:
            channel.send(payload)
    except ValueError as e:
        logger.error("dropping oversized data channel message '%s': %s", msg_type, e)
    except InvalidStateError:
        logger.info("skipping message because data channel closed mid-send: %s" % msg_type)

def __send_data_channel_message(self, msg_type: str, data: Any) -> None:
    """Broadcast a typed message to every connected peer.

    All display controllers and viewers receive it — the websockets
    transport broadcasts cursor, stats, and clipboard to all of its
    clients, so the channel path must too. Channels that are not open are
    skipped.
    """
    if not self.peer_connections:
        return
    sent = False
    for channel in self._iter_open_data_channels():
        self.send_message_to_channel(channel, msg_type, data)
        sent = True
    if not sent:
        logger.info("skipping message because no data channel is ready: %s" % msg_type)

def send_media_data_over_channel(self, msg_type: str, data: Any) -> None:
    """Broadcast a media-related message to all peers."""
    self.__send_data_channel_message(msg_type, data)

async def close_display_peers(self, display_id: str) -> None:
    """Close every peer attached to `display_id`.

    Each close is finished by the connection-state handler (channel
    consumers, media graph, display registration), which runs as its own
    task.
    """
    for peer_obj in [obj for obj in self.peer_connections.values()
                     if (obj.get("display_id") or "primary") == display_id]:
        peer_conn = peer_obj.get("peer_conn")
        if peer_conn is not None:
            try:
                await peer_conn.close()
            except Exception as e:
                logger.warning(f"Error closing a '{display_id}' peer: {e}")

def get_controller_instance(self) -> Optional[Dict[str, Any]]:
    """Return the peer entry for the controller client, if one exists.

    With multiple display controllers connected, the PRIMARY display's
    controller is the authoritative one (latency pings and
    controller-directed replies).
    """
    controllers = [
        obj for obj in self.peer_connections.values()
        if obj.get("client_type") == ClientType.CONTROLLER
    ]
    if not controllers:
        return None
    return next(
        (obj for obj in controllers if obj.get("display_id", "primary") == "primary"),
        controllers[0],
    )

def munge_sdp(self, sdp: str, encoder: Optional[str] = None,
              fullcolor: Optional[bool] = None,
              use_cpu: Optional[bool] = None) -> str:
    """Rewrite the local offer SDP for optimal streaming behavior.

    Injects a 125 ms rtx-time, `sps-pps-idr-in-keyframe=1` for H.264/H.265,
    the Opus ptime, and generous video bandwidth ceilings
    (`_munge_video_bandwidth`). Displays can run different encoders,
    chroma formats and software-encoding flags; the caller passes the ones
    this offer's display is using (defaults: the primary/global encoder and
    the configured full-colour and software-encoding settings).

    Full colour is a 4:4:4 bitstream, so the H.264 profile-level-id is
    rewritten to High 4:4:4 (`f4001f`) rather than handing the decoder a
    4:2:0 baseline profile that cannot match what it receives; 4:2:0 keeps
    `42e01f`, the profile Firefox negotiates. A session known to encode on
    OpenH264 (the software encoder of a GPL-free pixelflux build, forced
    onto the CPU) is excluded: it always emits limited-range 4:2:0, and a
    4:4:4 profile makes decoders misread its color range (visibly darker
    output).

    The Opus ptime advertises the real frame duration pcmflux emits
    (`audio_frame_duration_ms`) so the client keys its minptime munge off
    it, rounded to whole milliseconds for browser SDP parsers.

    Args:
        sdp: The local offer SDP text.
        encoder: Encoder the display is running; None means the global one.
        fullcolor: Whether the display emits a 4:4:4 bitstream; None reads
            the configured setting.
        use_cpu: Whether the display forces software encoding; None reads
            the configured setting.

    Returns:
        The munged SDP text.
    """
    encoder = encoder or self.encoder
    if fullcolor is None:
        fullcolor = bool(app_settings.video_fullcolor[0])
    if use_cpu is None:
        use_cpu = bool(app_settings.use_cpu[0])
    software_path = software_h264_path(
        encoder, use_cpu or str(getattr(app_settings, "gpu_id", "")).strip() == "-1")
    sdp_text = sdp
    if 'rtx-time' not in sdp_text:
        logger.warning("injecting rtx-time to SDP")
        sdp_text = re.sub(r'(apt=\d+)', r'\1;rtx-time=125', sdp_text)
    elif 'rtx-time=125' not in sdp_text:
        logger.warning("injecting modified rtx-time to SDP")
        sdp_text = re.sub(r'rtx-time=\d+', r'rtx-time=125', sdp_text)
    if "h264" in encoder or "x264" in encoder or "h265" in encoder or "x265" in encoder:
        if 'sps-pps-idr-in-keyframe' not in sdp_text:
            logger.warning("injecting sps-pps-idr-in-keyframe to SDP")
            sdp_text = sdp_text.replace('packetization-mode=', 'sps-pps-idr-in-keyframe=1;packetization-mode=')
        elif 'sps-pps-idr-in-keyframe=1' not in sdp_text:
            logger.warning("injecting modified sps-pps-idr-in-keyframe to SDP")
            sdp_text = re.sub(r'sps-pps-idr-in-keyframe=\d+', r'sps-pps-idr-in-keyframe=1', sdp_text)
        if ("h264" in encoder or "x264" in encoder) and fullcolor \
                and not (software_path and software_h264_encoder() == "openh264"):
            sdp_text = re.sub(r'profile-level-id=[0-9A-Fa-f]{6}',
                              'profile-level-id=f4001f', sdp_text)
    if "opus/" in sdp_text.lower():
        frame_ms = float(getattr(app_settings, 'audio_frame_duration_ms', '10') or 10)
        # A 2.5 ms frame advertises 3; pcmflux keeps the real frame.
        ptime = int(frame_ms + 0.5)
        # a=ptime is media-level: it must sit in the audio m-section, not after
        # the video sprop lines, and audio-less offers must skip it.
        if f"a=ptime:{ptime}" not in sdp_text:
            sections = re.split(r'(?m)(?=^m=)', sdp_text)
            for i, section in enumerate(sections):
                if section.startswith('m=audio'):
                    lines = section.split('\r\n')
                    lines.insert(1, f'a=ptime:{ptime}')
                    sections[i] = '\r\n'.join(lines)
                    break
            sdp_text = ''.join(sections)

    sdp_text = self._munge_video_bandwidth(sdp_text)

    return sdp_text

def _munge_video_bandwidth(self, sdp_text: str) -> str:
    """Raise the bandwidth ceiling of every video m-section in the SDP.

    A generous `b=AS` keeps the browser's REMB from throttling a
    high-bitrate desktop stream (it is a cap hint, not a target), and
    `x-google-max-bitrate` mirrors it on the Chrome receive side. Both are
    scoped to each video m-section: the `b=AS` presence check looks only
    inside that section (a `b=AS` elsewhere says nothing about this
    section's ceiling), and the line goes after the section's own `c=`,
    or right after the `m=video` line when the section inherits the
    session-level `c=` (RFC 4566). The x-google hints go on the fmtp of
    every video rtpmap payload type except RTX, so VP8/VP9 get them too;
    a codec without an fmtp gets one carrying just the hints.
    """
    XGOOGLE = "x-google-max-bitrate=300000;x-google-min-bitrate=0"
    lines = sdp_text.split("\r\n")
    out: List[str] = []
    i = 0
    n = len(lines)
    while i < n:
        line = lines[i]
        if not line.startswith("m=video"):
            out.append(line)
            i += 1
            continue

        section = [line]
        i += 1
        while i < n and not lines[i].startswith("m="):
            section.append(lines[i])
            i += 1

        if not any(s.startswith("b=AS:") for s in section):
            c_idx = next(
                (idx for idx, s in enumerate(section) if s.startswith("c=")),
                None,
            )
            insert_at = (c_idx + 1) if c_idx is not None else 1
            section.insert(insert_at, "b=AS:300000")

        video_pts = []
        for s in section:
            m = re.match(r'a=rtpmap:(\d+)\s+(\S+)', s)
            if m and not m.group(2).lower().startswith("rtx/"):
                video_pts.append(m.group(1))

        for pt in video_pts:
            fmtp_idx = next(
                (idx for idx, s in enumerate(section)
                 if s.startswith("a=fmtp:{} ".format(pt))),
                None,
            )
            if fmtp_idx is not None:
                if "x-google-max-bitrate" not in section[fmtp_idx]:
                    section[fmtp_idx] = re.sub(
                        r'^(a=fmtp:{} )'.format(pt),
                        r'\g<1>' + XGOOGLE + ';',
                        section[fmtp_idx],
                    )
            else:
                rtpmap_idx = next(
                    (idx for idx, s in enumerate(section)
                     if s.startswith("a=rtpmap:{} ".format(pt))),
                    None,
                )
                if rtpmap_idx is not None:
                    section.insert(rtpmap_idx + 1, "a=fmtp:{} {}".format(pt, XGOOGLE))

        out.extend(section)

    return "\r\n".join(out)

def consume_data(self, buf: Any, pts: Optional[int], kind: str,
                 display_id: str = "primary") -> None:
    """Feed one encoded frame from the capture side into a display's bridge.

    Synchronous: scheduled via `loop.call_soon_threadsafe` from the capture
    thread, since `set_data` does not await — no per-frame Future/Task.
    `EncodedPacket` references the encoder buffer without copying and the
    packers walk it as a memoryview, so no per-frame FFmpeg object is
    allocated and no whole-frame copy is taken.

    Args:
        buf: Buffer-protocol object holding the encoded sample.
        pts: Presentation timestamp in the stream's clock, or None.
        kind: "video" or "audio".
        display_id: Display whose media graph receives the sample.
    """
    graph = self.displays.get(display_id or "primary")
    if graph is None:
        return
    if kind == "video":
        if buf:
            try:
                RTP_VIDEO_CLOCK_RATE = 90000
                packet = EncodedPacket(buf, pts, Fraction(1, RTP_VIDEO_CLOCK_RATE))
                bridge = graph.get("video_bridge")
                if bridge is not None:
                    bridge.set_data(packet)
            except Exception as e:
                logger.error(f"error processing video sample: {e}")
    elif kind == "audio":
        if buf:
            try:
                packet = EncodedPacket(buf, pts, Fraction(1, 48000))
                bridge = graph.get("audio_bridge")
                if bridge is not None:
                    bridge.set_data(packet)
            except Exception as e:
                logger.error(f"error processing audio sample: {e}")

def update_rtc_config(self, stun_servers: List[str], turn_servers: List[str]) -> None:
    """Update the STUN/TURN servers used for every NEW peer connection.

    get_rtc_config() reads these at peer-creation time, so a refresh (typically
    rotated TURN REST credentials) takes effect for every subsequent connection.
    Live sessions deliberately keep their established ICE: their TURN allocations
    stay valid, and forcing an ICE restart on refresh would drop working streams.
    """
    changed = (stun_servers, turn_servers) != (self.stun_servers, self.turn_servers)
    self.stun_servers = stun_servers
    self.turn_servers = turn_servers
    if changed:
        logger.debug(
            "RTC ICE servers updated; applies to new connections "
            "(established sessions keep their current ICE)."
        )

def format_turn_servers(self, turn_servers: List[str]) -> List[Dict[str, Optional[str]]]:
    """Parse turn:// or turns:// URL strings into ICE server dicts.

    Non-TURN or unparsable entries are skipped; missing ports fall back to
    the scheme default, bare IPv6 hosts are bracketed, and URL-encoded
    credentials are decoded.

    Returns:
        One dict per valid server with `urls` and, when the URL carried
        them, `username` / `credential` keys.
    """
    formatted_servers: List[Dict[str, Optional[str]]] = []
    for server in turn_servers or []:
        if not isinstance(server, str):
            continue

        lower_server = server.lower()
        if not (lower_server.startswith("turn://") or lower_server.startswith("turns://")):
            continue

        parsed = urllib.parse.urlparse(server)
        if not parsed.hostname:
            continue

        scheme = 'turns' if parsed.scheme.lower() == 'turns' else 'turn'
        try:
            port = parsed.port or (443 if scheme == 'turns' else 3478)
        except ValueError:
            port = 443 if scheme == 'turns' else 3478

        host = parsed.hostname
        if host and ":" in host and not (host.startswith("[") and host.endswith("]")):
            host = f"[{host}]"

        query = f"?{parsed.query}" if parsed.query else ""
        turn_entry: Dict[str, Optional[str]] = {
            'urls': f'{scheme}:{host}:{port}{query}'
        }

        if parsed.username is not None and parsed.password is not None:
            turn_entry['username'] = urllib.parse.unquote(parsed.username)
            turn_entry['credential'] = urllib.parse.unquote(parsed.password)

        formatted_servers.append(turn_entry)
    return formatted_servers

def format_stun_servers(self, stun_servers: List[str]) -> List[str]:
    """Strip the URL scheme separator from each STUN server string."""
    formatted_servers: List[str] = []
    for stun in stun_servers:
        server = stun.split("//")
        formatted_servers.append("".join(server))
    return formatted_servers

def get_rtc_config(self) -> RTCConfiguration:
    """Build the RTCConfiguration for a new peer from the current servers.

    Operator-configured public addresses (`webrtc_public_ip`, comma- or
    space-separated IPv4/IPv6) are advertised in host ICE candidates for
    hosts behind static 1:1 NAT; each family maps to its own host
    candidates.
    """
    formatted_turn_servers = self.format_turn_servers(self.turn_servers)
    formatted_stun_servers = self.format_stun_servers(self.stun_servers)
    logger.debug(f"stun servers: {formatted_stun_servers}")
    logger.debug(f"turn servers: {formatted_turn_servers}")

    ice_servers = []
    if self.stun_servers:
        ice_servers.append(RTCIceServer(urls=formatted_stun_servers))
    for turn in formatted_turn_servers:
        turn_kwargs: Dict[str, Any] = {
            'urls': turn.get('urls', [])
        }
        if turn.get('username') is not None:
            turn_kwargs['username'] = turn.get('username')
        if turn.get('credential') is not None:
            turn_kwargs['credential'] = turn.get('credential')
        ice_servers.append(RTCIceServer(**turn_kwargs))
    public_ips = (
        getattr(app_settings, "webrtc_public_ip", "") or ""
    ).replace(",", " ").split()
    config = RTCConfiguration(
        iceServers=ice_servers,
        bundlePolicy=RTCBundlePolicy.MAX_BUNDLE,
        iceHostPublicIps=public_ips or None,
    )
    return config

def force_codec(self, pc: RTCPeerConnection, sender: RTCRtpSender,
                forced_codec_mime: str) -> None:
    """Restrict a sender's codec preferences to one MIME type plus RTX.

    Every codec matching the MIME type stays eligible — H.264 appears once
    per advertised profile. FlexFEC rides along when the receiver supports
    it (Chrome family); a receiver without it answers without the codec and
    the sender emits no repair stream.

    Args:
        pc: Peer connection owning the sender's transceiver.
        sender: RTP sender whose transceiver is being restricted.
        forced_codec_mime: MIME type (e.g. "video/H264") to force.

    Raises:
        ValueError: When the codec or its RTX companion is not in the
            sender capabilities.
    """
    kind = sender.track.kind
    capabilities = RTCRtpSender.getCapabilities(kind)
    logger.debug(f"Current capabilities for {kind}: {capabilities}")

    chosen_codec = []
    for codec in capabilities.codecs:
        if codec.mimeType == forced_codec_mime:
            chosen_codec.append(codec)

    if not chosen_codec:
        raise ValueError(f"Codec {forced_codec_mime} not found in capabilities")

    rtx_codec = None
    for codec in capabilities.codecs:
        if codec.mimeType.lower() == f"{kind}/rtx":
            rtx_codec = codec
            break

    if not rtx_codec:
        raise ValueError(f"RTX codec for {forced_codec_mime} not found")

    flexfec_codec = next(
        (
            codec
            for codec in capabilities.codecs
            if codec.mimeType.lower() == f"{kind}/flexfec-03"
        ),
        None,
    )
    preferences = [*chosen_codec, rtx_codec]
    if flexfec_codec is not None:
        preferences.append(flexfec_codec)

    transceiver = next(t for t in pc.getTransceivers() if t.sender == sender)
    logger.debug(f"Forcing codec preferences to: {preferences}")
    transceiver.setCodecPreferences(preferences)

async def _drain_channel_queue(self, queue: asyncio.Queue,
                               handler: Callable[[Any], Any],
                               label: str) -> None:
    """Single consumer that dispatches queued messages strictly in order.

    Running one awaited handler at a time is what guarantees ordering: if
    each message spawned its own task, handlers that await mid-dispatch
    could complete out of order (e.g. a key-up finishing before its
    key-down, sticking the key).
    """
    while True:
        msg = await queue.get()
        try:
            result = handler(msg)
            if inspect.isawaitable(result):
                await result
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.error("Error handling message on channel %s: %s", label, e)

def _serialize_channel(self, channel: RTCDataChannel,
                       handler: Callable[[Any], Any],
                       max_queue: int = 512) -> "asyncio.Task":
    """Wire a channel's messages through a bounded per-channel queue drained
    by a single consumer task, so dispatch stays in arrival order.

    The message handler only enqueues (drop+log on overflow); the consumer
    is cancelled when the channel closes. handler is called late-bound so
    reassigning the target callback still takes effect.

    Returns:
        The consumer task, so teardown paths can cancel it for channels
        that never emit a close event.
    """
    queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue)

    def _enqueue(msg: Any) -> None:
        try:
            queue.put_nowait(msg)
        except asyncio.QueueFull:
            logger.warning("Data channel %s input queue full, dropping message", channel.label)

    consumer = self.async_event_loop.create_task(
        self._drain_channel_queue(queue, handler, channel.label)
    )
    channel.on("message", _enqueue)
    channel.on("close", lambda: consumer.cancel())
    return consumer

def _send_collab_state(self, channel: RTCDataChannel,
                       client_type: ClientType,
                       client_token: Optional[str]) -> None:
    """Send a peer its mk-token input verdict over the data channel.

    Delivered as a `system` action (the channel is JSON-typed):
    `mk_access,1` attaches the client's input context, `mk_access,0`
    detaches it. Sent to controllers as well as viewers (websockets
    MK_ACCESS parity) — an mk handoff strips a controller's input
    authority too, and without the verdict its page keeps a live input UI
    whose messages the server drops. No-op outside secure mode. Gated on
    secure mode itself, NOT on an mk token existing — a handoff that
    clears the token must still push the 0 that detaches the previous
    holder.
    """
    if not app_settings.master_token:
        return
    granted = (
        self._viewer_is_collaborator(client_token)
        if client_type == ClientType.VIEWER
        else self._mk_input_authorized(client_token)
    )
    try:
        channel.send(json.dumps(
            {"type": "system", "data": {"action": f"mk_access,{1 if granted else 0}"}}))
    except Exception:
        logger.debug("collab-state send failed (channel closing)", exc_info=True)

def _send_auth_success(self, channel: RTCDataChannel,
                       client_type: ClientType,
                       client_token: Optional[str]) -> None:
    """Tell the client its effective role/slot (websockets AUTH_SUCCESS
    parity): role coercion is decided server-side, so the page must learn
    the verdict to degrade its own UI instead of driving a controller UI
    whose input is all dropped."""
    try:
        role = "controller" if client_type == ClientType.CONTROLLER else "viewer"
        slot = None
        if client_token:
            perms = current_session_tokens()[0].get(client_token)
            if perms:
                slot = perms.get("slot")
        verdict = json.dumps({"role": role, "slot": slot})
        channel.send(json.dumps(
            {"type": "system", "data": {"action": f"auth_success,{verdict}"}}))
    except Exception:
        logger.debug("auth_success send failed (channel closing)", exc_info=True)

def _viewer_is_collaborator(self, client_token: Optional[str]) -> bool:
    """A viewer holding the active mk (mouse+keyboard) token is a read-write
    collaborator — mirrors the WS mk-token path — but only while enable_collab
    is on. Fail-safe: any missing piece means not a collaborator (stays
    read-only)."""
    if not client_token:
        return False
    if not bool(app_settings.enable_collab[0]):
        return False
    _, mk = current_session_tokens()
    return mk is not None and client_token == mk

def _mk_input_authorized(self, client_token: Optional[str]) -> bool:
    """Token-level input authority in secure mode: the active mk-token holder,
    or a controller-role token while no mk token is provisioned."""
    tokens, mk = current_session_tokens()
    if mk is not None:
        return bool(client_token) and client_token == mk
    perms = tokens.get(client_token) if client_token else None
    return bool(perms) and perms.get("role") == "controller"

def peer_holds_input_authority(self, peer: Optional[Dict[str, Any]]) -> bool:
    """Whether a peer entry may drive keyboard/mouse input, composing the two
    gates on_data_message applies: a viewer needs to be a read-write
    collaborator, and in secure mode every peer is additionally held to the
    token check. Used for session-wide input cleanup (held keys and pointer
    buttons are one global desktop state), so it is deliberately fail-safe:
    an unknown peer holds nothing."""
    if not peer:
        return False
    client_token = peer.get("client_token")
    if peer.get("client_type") == ClientType.VIEWER and not self._viewer_is_collaborator(client_token):
        return False
    if not app_settings.master_token:
        return True
    return self._mk_input_authorized(client_token)

def _secure_input_denied(self, msg: str, client_token: Optional[str]) -> bool:
    """Secure-mode (master token configured) input authority, mirroring the WS
    gate: cmd and the keyboard/mouse/clipboard set are admitted only from the
    active mk-token holder, or from a controller-role token when no mk-token is
    provisioned. client_type is self-asserted over signaling, so a peer that
    merely claims 'controller' is still held to the token here. `co`
    (composed-text typing, `co,end,<text>`) is keyboard input like kd/ku.
    The bare `cr` clipboard read-back is exempt like every clipboard read
    on the websockets transport: the handler itself direction-gates it
    (enable_clipboard "out") and it is sent at connect, before the peer
    can hold input authority.

    Returns:
        True when the message must be dropped.
    """
    if not app_settings.master_token:
        return False
    if msg.split(",", 1)[0] in ("cr",):
        return False
    if msg.split(",", 1)[0] not in ("cmd", "co") and not msg.startswith(VIEWER_COLLAB_EXTRA_PREFIXES):
        return False
    authorized = self._mk_input_authorized(client_token)
    if not authorized:
        logger.warning("Dropping unauthorized secure-mode input: %s", msg[:32])
    return not authorized

def _secure_gamepad_denied(self, msg: str, client_token: Optional[str]) -> bool:
    """Secure-mode gamepad authority, mirroring the WS gate: a client may only
    drive its own assigned slot (js index == slot - 1), so a viewer or
    collaborator can't spoof another player's controller. No slot on the
    token means no gamepad.

    Returns:
        True when the gamepad message must be dropped.
    """
    if not app_settings.master_token or not msg.startswith("js,"):
        return False
    tokens, _ = current_session_tokens()
    perms = tokens.get(client_token) if client_token else None
    slot = perms.get("slot") if perms else None
    if slot is None:
        return True
    try:
        index = int(msg.split(",", 3)[2])
    except (IndexError, ValueError):
        return True
    return int(slot) - 1 != index

def _on_input_channel_message(self, msg: Any,
                              channel: Optional[RTCDataChannel] = None,
                              client_type: Optional[ClientType] = None,
                              client_token: Optional[str] = None,
                              display_id: str = "primary",
                              peer_id: Optional[str] = None) -> Any:
    """Pre-filter one data-channel message before the input dispatcher.

    In order: a gzip'd payload is inflated with a bound (the channel's
    negotiated max-message-size caps only the compressed size, websockets
    0x05 parity); the `_gz,1` handshake marks this channel for gzip'd
    sends and is echoed; a viewer's SETTINGS snapshot is connection sync
    only, never applied (the websockets transport likewise ignores viewer
    payloads); a viewer may send only the allow-listed messages, and a
    read-write collaborator (mk token plus enable_collab) additionally the
    keyboard/mouse/clipboard set — the same two tiers as the websockets
    gate, so a collaborator still cannot send cmd — with blur/visibility
    noise dropped silently and the collaborator check reached only for
    otherwise-disallowed input so normal viewer traffic pays nothing; the
    secure-mode input and gamepad gates apply; STOP_VIDEO / START_VIDEO
    pause this peer only (viewer-allowed, so a hidden viewer pauses its
    own feed); STOP_AUDIO / START_AUDIO are ignored because audio is
    negotiated per peer over SDP here, so the websockets global toggle
    would be a no-op for late joiners or cut audio for peers that never
    asked — each page mutes its `<video>` locally, which stops Opus
    playback without stopping RTP. Everything else reaches the late-bound
    `on_data_message` with the peer id as the connection id, so
    per-connection input state (gamepad associations) traces to the peer.

    Returns:
        Whatever the dispatched handler returns (possibly an awaitable,
        which the channel's queue consumer awaits), or None when the
        message was consumed or dropped here.
    """
    if isinstance(msg, (bytes, bytearray)) and bytes(msg[:2]) == b"\x1f\x8b":
        try:
            msg = inflate_gz_bounded(msg)
        except Exception:
            logger.warning("Dropping undecodable compressed data channel message")
            return
    if msg == "_gz,1":
        if channel is not None:
            channel._selkies_gz_tx = True
            try:
                channel.send("_gz,1")
            except Exception as e:
                logger.warning("Failed to ack compression handshake: %s", e)
        return
    if client_type == ClientType.VIEWER and isinstance(msg, str) and msg.startswith("SETTINGS,"):
        logger.debug("Ignoring SETTINGS payload from a viewer (display '%s')", display_id)
        return
    if client_type == ClientType.VIEWER and isinstance(msg, str):
        if not msg.startswith(VIEWER_ALLOWED_PREFIXES) and not (
            msg.startswith(VIEWER_COLLAB_EXTRA_PREFIXES)
            and self._viewer_is_collaborator(client_token)
        ):
            if not msg.startswith(VIEWER_SILENT_DROP_PREFIXES):
                logger.warning("Dropping unauthorized viewer input: %s", msg[:32])
            return
    if isinstance(msg, str) and self._secure_input_denied(msg, client_token):
        return
    if isinstance(msg, str) and self._secure_gamepad_denied(msg, client_token):
        logger.warning("Dropping gamepad input for unassigned slot: %s", msg[:32])
        return
    if msg in ("STOP_VIDEO", "START_VIDEO") and self.on_video_consumer_active is not None:
        return self.on_video_consumer_active(
            peer_id, display_id or "primary", msg == "START_VIDEO")
    if msg in ("STOP_AUDIO", "START_AUDIO"):
        logger.debug("Ignoring %s over WebRTC: audio is per-peer (SDP), not global.", msg)
        return
    return self.on_data_message(msg, display_id or "primary", conn_id=peer_id)

async def on_peer_connection_established(self, client_peer_id: str, client_type: ClientType, display_id: str = "primary") -> None:
    """Start the display's capture when a peer finishes connecting.

    Every consumer asks, not just the controller: a lone viewer must get
    the desktop it joined for (websockets parity), and the start is
    idempotent on a display whose capture is already running.
    """
    await self.start_display_media(display_id)
    logger.info(f"Media pipeline start requested for {client_peer_id} (display '{display_id}')")

async def _default_start_display_media(self, display_id: str) -> None:
    """Single-display default: only the primary pipeline is started."""
    if display_id == "primary" and self.media_pipeline:
        await self.media_pipeline.start_media_pipeline()

async def _default_stop_display_media(self, display_id: str) -> None:
    """Single-display default: only the primary pipeline is stopped."""
    if display_id == "primary" and self.media_pipeline:
        await self.media_pipeline.stop_media_pipeline()

async def on_connectionstatechange(self, client_peer_id: str) -> None:
    """React to a peer connection's state changes.

    The "closed" branch is the teardown point for a peer whose client
    vanished without a session end (ICE failure, the browser going away):
    it deregisters the peer and reaps what the peer owned (consumer tasks,
    mic playback, the display's media while nothing else consumes it, the
    on_peer_gone hook). An explicit stop deregisters the peer before
    closing it, so for that peer this handler finds no entry and the stop
    reaps it itself.
    """
    peer_conn = None
    peer_obj = None
    if client_peer_id:
        peer_obj = self.peer_connections.get(client_peer_id, None)
        if peer_obj:
            peer_conn = peer_obj.get("peer_conn")

    if peer_conn is None:
        logger.debug("No peer connection found for connectionstatechange")
        return

    state = peer_conn.connectionState
    client_type = peer_obj.get('client_type') if peer_obj else ''
    display_id = (peer_obj.get('display_id') if peer_obj else None) or 'primary'
    if state == "failed":
        await peer_conn.close()
    elif state == "disconnected":
        logger.warning("Peer connection disconnected", extra={'client_peer_id': client_peer_id, 'client_type': client_type})
    elif state == "connected":
        await self.on_peer_connection_established(client_peer_id, client_type, display_id)
        logger.info("Peer connection established", extra={'client_peer_id': client_peer_id, 'client_type': client_type})
    elif state == "closed":
        self.peer_connections.pop(client_peer_id, None)
        await self._reap_peer(client_peer_id, peer_obj)
        logger.info("Peer connection closed", extra={'client_peer_id': client_peer_id, 'client_type': client_type})
    elif state == "connecting":
        logger.info("Peer connection is connecting", extra={'client_peer_id': client_peer_id, 'client_type': client_type})
    else:
        logger.debug(f"Unhandled peer connection state: {state}", extra={'client_peer_id': client_peer_id, 'client_type': client_type})