WebRTCInput

def __init__(
    self,
    rtc_app: Any,
    uinput_mouse_socket_path: str = "",
    js_socket_path_prefix: str = "/tmp",
    enable_clipboard: str = "",
    enable_binary_clipboard: str = "",
    enable_cursors: bool = True,
    cursor_size: int = 16,
    cursor_scale: float = 1.0,
    cursor_debug: bool = False,
    max_cursor_size: int = 32,
    data_server_instance: Any = None,
    upload_dir: Optional[str] = None,
    is_wayland: bool = False,
    wayland_socket_index: int = 0,
    app_wayland_display: str = "",
    uinput_gamepad: str = "auto",
) -> None:
    self.wayland_socket_index = wayland_socket_index
    self.app_wayland_display = app_wayland_display
    self._app_wl_display_cached = None
    self._x_reconnect_thread = None
    self._x11_monitor_build_lock = asyncio.Lock()
    self._x_event_wake = None
    self._x_watcher_fd = None
    self._app_wl_negcache = None
    self._app_wl_negcache_at = 0.0
    self._app_wl_is_separate = False
    self.active_shortcut_modifiers = set()
    self.SHORTCUT_MODIFIER_XKEY_NAMES = {
        'Control_L', 'Control_R', 
        'Alt_L', 'Alt_R', 
        'Super_L', 'Super_R',
        'Meta_L', 'Meta_R'
    }
    self.active_modifiers = set()
    self.atomically_typed_keys = set()
    self.translated_keys = set()
    self.ACTION_MODIFIER_KEYSYMS = {65507, 65508, 65513, 65514, 65511, 65512,
                                    65515, 65516, 65517, 65518}
    self.LEVEL_MODIFIER_KEYSYMS = frozenset({0xFFE1, 0xFFE2, 0xFE03, 0xFF7E})
    self.MODIFIER_KEYSYMS = {
        # Shift_L, Shift_R
        65505, 65506,
        # Control_L, Control_R
        65507, 65508,
        # Alt_L, Alt_R
        65513, 65514,
        # ISO_Level3_Shift (AltGr)
        65027,
        # Meta_L, Meta_R
        65511, 65512,
        # Super_L, Super_R
        65515, 65516,
        # Hyper_L, Hyper_R
        65517, 65518,
    }
    self.rtc_app = rtc_app
    self.loop = asyncio.get_running_loop()
    self.js_socket_path_prefix = js_socket_path_prefix
    self.num_gamepads = 4
    self.gamepad_instances = {}
    self.client_gamepad_associations = {}
    self.gamepad_heartbeats = {}
    self.uinput_gamepads = uinput_gamepads_enabled(uinput_gamepad)

    self.clipboard_running = False
    self._binary_clipboard_lock = asyncio.Lock()
    self._clipboard_monitor_active = False
    self.uinput_mouse_socket_path = uinput_mouse_socket_path
    self.uinput_mouse_socket = None
    self.enable_clipboard = enable_clipboard
    self.enable_binary_clipboard = enable_binary_clipboard
    self._apps_runner_ok: Optional[bool] = None
    self.enable_cursors = enable_cursors
    self.cursors_running = False
    self.cursor_scale = cursor_scale
    self.cursor_size = cursor_size
    self.cursor_debug = cursor_debug
    # An explicit cursor_size raises the capture cap so the requested size
    # survives the transport instead of being resized down.
    if isinstance(cursor_size, int) and cursor_size > 0:
        max_cursor_size = max(max_cursor_size, cursor_size)
    self.max_cursor_size = max_cursor_size
    self.system_dpi = 96.0
    self.cursor_size_cap = max_cursor_size
    self._cursor_msg_cache = None
    self.keyboard = None
    self.mouse = None
    self.xdisplay = None
    self.button_mask = 0
    self.last_x = -1
    self.last_y = -1
    self.tracked_position_stale = False
    self.ping_start = None

    self.upload_dir = upload_dir
    self.upload_dir_path = None

    async def _unhandled_video_bitrate(bitrate, display_id="primary"):
        logger_webrtc_input.warning(f"unhandled on_video_encoder_bit_rate: {bitrate}")
    self.on_video_encoder_bit_rate = _unhandled_video_bitrate
    async def _unhandled_audio_bitrate(bitrate):
        logger_webrtc_input.warning(f"unhandled on_audio_encoder_bit_rate: {bitrate}")
    self.on_audio_encoder_bit_rate = _unhandled_audio_bitrate
    async def _unhandled_mouse_pointer(visible):
        logger_webrtc_input.warning(f"unhandled on_mouse_pointer_visible: {visible}")
    self.on_mouse_pointer_visible = _unhandled_mouse_pointer
    self.on_clipboard_read = self._on_clipboard_read
    self.on_set_fps = lambda fps, display_id="primary": logger_webrtc_input.warning("unhandled on_set_fps")
    self.on_request_keyframe = lambda display_id="primary": logger_webrtc_input.warning("unhandled on_request_keyframe")
    self.on_set_enable_resize = lambda enable_resize, res: logger_webrtc_input.warning("unhandled on_set_enable_resize")
    self.on_client_fps = lambda fps: logger_webrtc_input.warning("unhandled on_client_fps")
    self.on_client_latency = lambda latency: logger_webrtc_input.warning("unhandled on_client_latency")
    self.on_resize = lambda res, display_id="primary": logger_webrtc_input.warning("unhandled on_resize")
    self.on_scaling_ratio = lambda res: logger_webrtc_input.warning("unhandled on_scaling_ratio")
    self.on_ping_response = lambda latency: logger_webrtc_input.warning("unhandled on_ping_response")
    self.on_cursor_change = self._on_cursor_change
    async def _unhandled_webrtc_stats(webrtc_stat_type, webrtc_stats):
        logger_webrtc_input.debug(f"unhandled on_client_webrtc_stats: {webrtc_stat_type}")
    self.on_client_webrtc_stats = _unhandled_webrtc_stats
    self.clipboard_monitor_task = None
    self.multipart_clipboard_buffer = None
    self.multipart_clipboard_mime_type = "text/plain"
    self.multipart_clipboard_total_size = 0
    self.multipart_clipboard_in_progress = False
    self.multipart_clipboard_id = None
    self.multipart_clipboard_kind = None
    self.data_server_instance = data_server_instance
    self.on_update_settings = lambda settings_json, display_id="primary": logger_webrtc_input.warning("unhandled update_settings")
    self.is_wayland = is_wayland
    self.wayland_input = None
    self._client_kb_layout = None
    self._wl_seat_client_layout = None
    self._wl_keymap_owner = None
    self._wl_keymap_stale = False
    self._wl_keymap_owner_lock = asyncio.Lock()
    self._wl_keymap_retry_at = 0.0
    self._wl_typer_lock = asyncio.Lock()
    self._wl_typer_retry_at = 0.0
    self._clipboard_inject_lock = asyncio.Lock()
    self._clipboard_inject_active = False
    self._clipboard_last_bytes = None
    self._x11_clipboard_monitor = None
    self._x11_monitor_retry_at = 0.0
    self._x11_monitor_unavail_logged = False
    self._app_watch_failure = None
    self._app_clip_read_failure = None
    self._wl_native_arm_failure = None
    self._session_env_cache = {}
    self._session_env_empty_at = {}
    self._session_env_negcache_ttl = 5.0
    self._xclip_missing_warned = False
    self._last_clipboard_request_ts = {}
    self._clipboard_request_debounce = 0.25
    self._bg_tasks = set()
    self.keyboard_queue = asyncio.Queue(maxsize=4096)
    self.keyboard_worker_task = None
    self._wl_text_routed = {}
    self.pressed_keys = {}
    self.reaped_atomic_keys = set()
    self.max_pressed_keys = 1024
    self.key_stale_window = 2.0
    self.key_sweep_interval = 0.1
    self.key_sweep_task = None
    self.key_repeat_enabled = not self.is_wayland
    self.key_repeat_delay = 0.5
    self.key_repeat_interval = 0.04
    self.key_repeat_tick = 0.02
    self.key_repeat_heartbeat_grace = 0.3
    self.key_repeat_state = {}
    self.key_repeat_task = None
    self.keymap_watch_task = None
    self.on_update_rate_control_mode = lambda mode, display_id="primary": logger_webrtc_input.warning("unhandled on_update_rate_control_mode")
    self.on_update_crf = lambda value, display_id="primary": logger_webrtc_input.warning("unhandled on_update_crf")

    if self.is_wayland:
        try:
            if ScreenCapture is None:
                raise RuntimeError("pixelflux is not installed")
            self.wayland_input = ScreenCapture()
            logger_webrtc_input.info("Wayland input injection initialized.")
            missing = [m for m in (
                "clipboard_write_app", "clipboard_unwatch_app",
                "list_outputs", "create_output", "set_keymap_overlay",
                "hold_spare_app_screens", "set_app_output_scale",
                "set_app_screen_geometry",
                "set_app_wayland_display", "type_text_wayland",
                "get_keyboard_state",
            ) if not hasattr(self.wayland_input, m)]
            if missing:
                logger_webrtc_input.warning(
                    "Installed pixelflux is missing APIs this build "
                    "expects; Wayland features that depend on them "
                    "degrade or stay off. Update pixelflux to a "
                    "matching build.")
                logger_webrtc_input.debug(
                    f"pixelflux methods absent: {', '.join(missing)}")
        except Exception as e:
            logger_webrtc_input.error(f"Failed to initialize Wayland input: {e}")

async def _on_clipboard_read(self, data: Union[str, bytes],
                             mime_type: str = "text/plain") -> None:
    await self.send_clipboard_data(data, mime_type)

def _on_cursor_change(self, data: dict) -> None: self.send_cursor_data(data)

async def send_clipboard_data(self, data: Union[str, bytes],
                              mime_type: str = "text/plain",
                              reply_to: Optional[str] = None,
                              conn_id: Any = None) -> None:
    """Route clipboard content to the transport's own chunked sender.

    Each transport owns one (SelkiesStreamingApp.send_ws_clipboard_data /
    RTCApp.send_clipboard_data) and addresses the requester by the
    identity its own connections carry.
    """
    if self._ws_transport():
        await self.rtc_app.send_ws_clipboard_data(data, mime_type, reply_to=reply_to,
                                                  conn_id=conn_id)
    else:
        await self.rtc_app.send_clipboard_data(data, mime_type, reply_to=reply_to,
                                               peer_id=conn_id)

def _ws_transport(self) -> bool:
    """Whether the owning app is the websockets transport (it carries a
    `mode`); the WebRTC app has no such attribute and routes the other way."""
    return getattr(self.rtc_app, "mode", None) == "websockets"

def send_cursor_data(self, data: dict) -> None:
    if self._ws_transport(): self.rtc_app.send_ws_cursor_data(data)
    else: self.rtc_app.send_cursor_data(data)

def send_command_status(self, action: str, conn_id: Optional[str] = None) -> None:
    """Route a command notice (``command_error``/``command_done``) to the
    transport (see send_cursor_data).

    Each transport carries the system action on its own wire format. Over
    WebRTC ``conn_id`` is the requesting peer's id, so the notice targets
    that peer's channel; the websockets transport notifies its requesting
    socket in its own cmd branch and only broadcasts here.
    """
    try:
        if self._ws_transport():
            self.rtc_app.send_system_action(action)
        else:
            self.rtc_app.send_system_action(action, peer_id=conn_id)
    except Exception:
        logger_webrtc_input.debug("command status notify failed", exc_info=True)

def __keyboard_connect(self) -> None: self.keyboard = _XTestKeyboard(self.xdisplay) if self.xdisplay else None

def _apply_input_x_reply_bound(self) -> None:
    """Bound the wait for an X REPLY on the shared input connection so an
    unresponsive server (driver hang, a foreign client's server grab) raises
    ConnectionClosedError instead of freezing the event loop forever. XTEST
    injection (mouse motion/buttons, key press) is a no-reply request and is
    unaffected — only the modifier query (query_keymap) and the cursor-image
    fetch block, and those recover via _reconnect_xdisplay(). Event waits stay
    unbounded (a quiet server sending no cursor events is not an error)."""
    if self.xdisplay is None:
        return
    try:
        self.xdisplay.display.blocking_timeout = INPUT_X_REPLY_TIMEOUT_S
    except Exception:
        pass

def _arm_x_event_watcher(self) -> None:
    """(Re)register the event-loop reader that wakes X consumers when the
    input connection's socket goes readable. Idempotent; re-arms when the fd
    changes under us (reconnect)."""
    if self._x_event_wake is None:
        return
    fd = None
    if self.xdisplay is not None:
        try:
            fd = self.xdisplay.fileno()
        except Exception:
            fd = None
    if fd is not None and fd < 0:
        fd = None
    if fd == self._x_watcher_fd:
        return
    if self._x_watcher_fd is not None:
        try:
            self.loop.remove_reader(self._x_watcher_fd)
        except Exception:
            pass
    self._x_watcher_fd = fd
    if fd is not None:
        try:
            self.loop.add_reader(fd, self._x_event_wake.set)
        except Exception as e:
            logger_webrtc_input.debug(f"X event watcher unavailable ({e}); consumers poll.")
            self._x_watcher_fd = None

def _disarm_x_event_watcher(self) -> None:
    if self._x_watcher_fd is not None:
        try:
            self.loop.remove_reader(self._x_watcher_fd)
        except Exception:
            pass
        self._x_watcher_fd = None

async def _wait_x_event(self, timeout: float = 1.0) -> None:
    """Sleep until the X socket signals readability or the failsafe elapses.
    Callers clear the Event BEFORE re-checking event availability, so an
    event arriving between the check and the wait is never lost. With no
    reader armed (add_reader unsupported or failed) nothing will ever set the
    Event, so fall back to a short poll rather than idling out the failsafe."""
    wake = self._x_event_wake
    if wake is None:
        await asyncio.sleep(timeout)
        return
    if self._x_watcher_fd is None:
        await asyncio.sleep(min(timeout, INPUT_X_EVENT_POLL_S))
        return
    try:
        await asyncio.wait_for(wake.wait(), timeout)
    except asyncio.TimeoutError:
        pass

def _reconnect_xdisplay(self) -> None:
    """Rebuild the input X connection after a bounded reply-wait closed it.

    Fire-and-forget: the rebuild runs on a worker thread and installs via
    the loop (_install_reconnected_xdisplay), so xdisplay stays None — and
    input degraded to the xdotool fallbacks — until the attempt lands;
    callers just continue and the next X failure retries. The thread is
    needed because every caller sits on the event loop, and against a
    server hung under another client's grab — the main condition this
    reconnect exists to survive — close() and the connection setup would
    freeze the loop for the whole outage. The handshake itself is bounded
    so a permanently dead server cannot pin the thread forever.
    """
    self._disarm_x_event_watcher()
    if self._x_reconnect_thread is not None and self._x_reconnect_thread.is_alive():
        return
    old = self.xdisplay
    self.xdisplay = None
    self.keyboard = None
    self.mouse = None

    def _attempt():
        try:
            if old is not None:
                try:
                    old.close()
                except Exception:
                    pass
            disp = display.Display(blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
        except Exception as e:
            logger_webrtc_input.error(f"Could not reconnect input X display: {e}")
            return
        try:
            self.loop.call_soon_threadsafe(self._install_reconnected_xdisplay, disp)
        except RuntimeError:
            try:
                disp.close()
            except Exception:
                pass

    self._x_reconnect_thread = threading.Thread(
        target=_attempt, name="x-input-reconnect", daemon=True
    )
    self._x_reconnect_thread.start()

def _install_reconnected_xdisplay(self, disp: Any) -> None:
    """Loop-side half of _reconnect_xdisplay: wire the fresh connection into
    every consumer in one step so nothing observes a half-initialized display."""
    if self.xdisplay is not None:
        # An earlier attempt already landed; drop this connection.
        try:
            disp.close()
        except Exception:
            pass
        return
    self.xdisplay = disp
    self._apply_input_x_reply_bound()
    self._arm_x_event_watcher()
    self.__keyboard_connect()
    if not self.is_wayland:
        self.mouse = _XTestMouse(self.xdisplay)
    if self.cursors_running:
        try:
            screen = self.xdisplay.screen()
            self.xdisplay.xfixes_select_cursor_input(
                screen.root, xfixes.XFixesDisplayCursorNotifyMask
            )
        except Exception as e:
            logger_webrtc_input.warning(f"Could not re-arm cursor monitor after reconnect: {e}")
    logger_webrtc_input.warning("Input X connection was unresponsive; reconnected.")

def _is_x_conn_closed(self, exc: BaseException) -> bool:
    """True if `exc` is the connection-closed error the reply bound raises."""
    return xlib_error is not None and isinstance(exc, xlib_error.ConnectionClosedError)

async def _load_server_autorepeat_rate(self) -> None:
    """Best-effort: adopt the X server's configured autorepeat delay/rate for our
    synthetic server-side repeat, so a held key feels native. python-xlib in some
    builds lacks the XKB controls API, so read it once at connect from `xset q`
    (one-time, not per-event). Any failure or out-of-range value keeps the sane
    defaults set in __init__."""
    if self.is_wayland:
        return
    try:
        process = await subprocess.create_subprocess_exec(
            "xset", "q", stdout=subprocess.PIPE, stderr=subprocess.PIPE
        )
        out, _err = await self._communicate_or_kill(process, 0.5, "xset q autorepeat")
        if process.returncode != 0 or not out:
            return
        text = out.decode("utf-8", "replace") if isinstance(out, (bytes, bytearray)) else str(out)
        m = re.search(r"auto repeat delay:\s*(\d+)\s*repeat rate:\s*(\d+)", text)
        if not m:
            return
        delay_ms = int(m.group(1))
        rate_hz = int(m.group(2))
        # Reject nonsense (a 0) that would busy-repeat or never repeat.
        if 100 <= delay_ms <= 2000:
            self.key_repeat_delay = delay_ms / 1000.0
        if 1 <= rate_hz <= 100:
            self.key_repeat_interval = 1.0 / rate_hz
        logger_webrtc_input.info(
            f"Server autorepeat: delay {self.key_repeat_delay:.3f}s, "
            f"interval {self.key_repeat_interval:.3f}s."
        )
    except Exception as e:
        logger_webrtc_input.debug(f"Could not read server autorepeat rate (using defaults): {e}")

def __mouse_connect(self) -> None:
    if self.uinput_mouse_socket_path:
        logger_webrtc_input.info(f"Connecting to uinput mouse socket: {self.uinput_mouse_socket_path}")
        self.uinput_mouse_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    if not self.is_wayland and self.xdisplay:
        self.mouse = _XTestMouse(self.xdisplay)

def __mouse_disconnect(self) -> None:
    if self.mouse: del self.mouse; self.mouse = None
    if self.uinput_mouse_socket is not None:
        try:
            self.uinput_mouse_socket.close()
        except OSError:
            pass
        self.uinput_mouse_socket = None

def __mouse_emit(self, *args: Any, **kwargs: Any) -> None:
    """Forward one msgpack-encoded mouse event to the uinput helper socket."""
    if self.uinput_mouse_socket_path:
        cmd = {"args": args, "kwargs": kwargs}
        data = msgpack.packb(cmd, use_bin_type=True)
        self.uinput_mouse_socket.sendto(data, self.uinput_mouse_socket_path)

async def __gamepad_connect(self, gamepad_idx: int, client_name: str,
                            client_num_btns: int, client_num_axes: int,
                            conn_id: Any = None) -> None:
    """Associate a client controller with a persistent gamepad slot.

    A fresh association starts with no heartbeat, so the previous client's
    last beat cannot date-stamp it into an immediate sweep, and the kernel
    device is brought up before the first input so applications see a
    plug event rather than a controller appearing mid-press.
    """
    if not (0 <= gamepad_idx < self.num_gamepads):
        logger_webrtc_input.error(f"Client association: Gamepad index {gamepad_idx} out of range (0-{self.num_gamepads-1}).")
        return

    if gamepad_idx not in self.gamepad_instances:
        logger_webrtc_input.error(
            f"Client association: No persistent gamepad instance found for index {gamepad_idx}. "
            f"This should not happen if _initialize_persistent_gamepads ran correctly."
        )
        return

    logger_webrtc_input.info(
        f"Client controller '{client_name}' ({client_num_btns}b, {client_num_axes}a) "
        f"is now associated with persistent virtual gamepad slot {gamepad_idx}."
    )

    self.client_gamepad_associations[gamepad_idx] = {
        "client_name": client_name,
        "client_num_btns": client_num_btns,
        "client_num_axes": client_num_axes,
        "association_time": time.time(),
        "conn_id": conn_id,
    }

    self.gamepad_heartbeats.pop(gamepad_idx, None)

    self.gamepad_instances[gamepad_idx].ensure_uinput()

async def release_gamepads_for_conn(self, conn_id: Any) -> None:
    """Disassociate (and neutralize, via reset_state) every gamepad slot whose
    association was made by this transport connection. This is the ungraceful
    path — a tab that dies mid-press never sends 'js,d', and only the transport
    knows the connection is gone."""
    if conn_id is None:
        return
    for idx, info in list(self.client_gamepad_associations.items()):
        if info.get("conn_id") == conn_id:
            await self.__gamepad_disconnect(idx)

async def __gamepad_disconnect(self, gamepad_idx: Optional[int] = None) -> None:
    """Disassociate one slot (or all, with None), releasing anything held.

    The release matters for an ungraceful client drop (a tab closed
    mid-press), which would otherwise leave the in-desktop app with a
    stuck button or deflected stick until a new client re-sends state.
    """
    if gamepad_idx is None:
        indices_to_disassociate = list(self.client_gamepad_associations.keys())
        logger_webrtc_input.info("Disassociating all client gamepads from persistent slots.")
    elif not (0 <= gamepad_idx < self.num_gamepads):
        logger_webrtc_input.error(f"Client disassociation: Gamepad index {gamepad_idx} out of range.")
        return
    else:
        indices_to_disassociate = [gamepad_idx]

    for idx in indices_to_disassociate:
        self.gamepad_heartbeats.pop(idx, None)
        if idx in self.client_gamepad_associations:
            associated_info = self.client_gamepad_associations.pop(idx)
            gamepad = self.gamepad_instances.get(idx)
            if gamepad is not None:
                gamepad.reset_state()
            logger_webrtc_input.info(
                f"Client controller '{associated_info.get('client_name', 'Unknown')}' "
                f"disassociated from persistent virtual gamepad slot {idx}."
            )
        elif gamepad_idx is not None:
             logger_webrtc_input.warning(
                f"Client disassociation: No active client association found for gamepad slot {idx} to disassociate."
            )

def __gamepad_emit_btn(self, gamepad_idx: int, client_btn_num: int,
                       client_btn_val: float) -> None:
    gamepad = self.gamepad_instances.get(gamepad_idx)
    if gamepad:
        gamepad.send_event(client_btn_num, client_btn_val, is_button_event=True)

def __gamepad_emit_axis(self, gamepad_idx: int, client_axis_num: int,
                        client_axis_val: float) -> None:
    gamepad = self.gamepad_instances.get(gamepad_idx)
    if gamepad:
        gamepad.send_event(client_axis_num, client_axis_val, is_button_event=False)

async def connect(self) -> None:
    """Bring the input backends up: X/Wayland connections, DPI detection,
    keyboard reset, persistent gamepads, and the background key tasks."""
    if not self.is_wayland and X11_LIBS_AVAILABLE:
        # Bounded handshake: a server hung under another client's grab must
        # surface as a failure, not freeze the loop.
        try: self.xdisplay = display.Display(blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
        except Exception as e: logger_webrtc_input.error(f"Failed to connect to X display: {e}"); self.xdisplay = None
        self._apply_input_x_reply_bound()
        if self._x_event_wake is None:
            self._x_event_wake = asyncio.Event()
        self._arm_x_event_watcher()
    if self.xdisplay:
        try:
            screen = self.xdisplay.screen()
            width_mm = screen.width_in_mms
            height_mm = screen.height_in_mms
            if width_mm > 0 and height_mm > 0:
                dpi_x = (screen.width_in_pixels * 25.4) / width_mm
                dpi_y = (screen.height_in_pixels * 25.4) / height_mm
                self.system_dpi = (dpi_x + dpi_y) / 2.0
            dpi_scale_factor = self.system_dpi / 96.0
            self.cursor_size_cap = int(self.max_cursor_size * dpi_scale_factor)
            logger_webrtc_input.info(
                f"System DPI detected as ~{self.system_dpi:.0f}. "
                f"Cursor size cap set to {self.cursor_size_cap}x{self.cursor_size_cap}px."
            )
        except Exception as e:
            logger_webrtc_input.warning(f"Could not determine system DPI, using default 96. Error: {e}")
    if not self.is_wayland and X11_LIBS_AVAILABLE:
        self.__keyboard_connect()
    if self.xdisplay:
        await self._load_server_autorepeat_rate()
        await self.reset_keyboard()
    self.__mouse_connect()

    await self._initialize_persistent_gamepads()

    if self.is_wayland:
        self.keyboard_worker_task = asyncio.create_task(self._keyboard_worker())
        await self._push_wayland_base_layout()
        # After the worker starts: a Lock left engaged by a previous session
        # inverts every letter, and the reset rides the fresh worker.
        await self.reset_keyboard()
    if self.key_sweep_task is None:
        self.key_sweep_task = asyncio.create_task(self._key_stale_sweep())
    if self.key_repeat_enabled and self.key_repeat_task is None:
        self.key_repeat_task = asyncio.create_task(self._key_repeat_loop())
    if self.xdisplay is not None and self.keymap_watch_task is None:
        self.keymap_watch_task = asyncio.create_task(self._keymap_watch_loop())

async def _initialize_persistent_gamepads(self) -> None:
    """Adopt live process-wide gamepad instances or create and start new ones.

    A live instance's sockets are what already-running apps hold open, so
    a service restart (transport mode switch) reuses them; rebinding would
    orphan those apps' fds.
    """
    logger_webrtc_input.info(f"Initializing {self.num_gamepads} persistent gamepad instances...")
    if not os.path.exists(self.js_socket_path_prefix):
        try:
            os.makedirs(self.js_socket_path_prefix, exist_ok=True)
            logger_webrtc_input.info(f"Created directory for gamepad sockets: {self.js_socket_path_prefix}")
        except OSError as e:
            logger_webrtc_input.error(f"Failed to create directory {self.js_socket_path_prefix} for gamepad sockets: {e}")
            return

    for i in range(self.num_gamepads):
        if i in self.gamepad_instances:
            logger_webrtc_input.warning(f"Gamepad instance for index {i} already exists. Skipping re-initialization.")
            continue

        existing = _persistent_gamepads.get(i)
        if existing is not None and existing.running:
            self.gamepad_instances[i] = existing
            logger_webrtc_input.info(
                f"Adopted live persistent gamepad instance for index {i} (JS: {existing.js_sock_path})."
            )
            continue

        js_ip_sock_path = os.path.join(self.js_socket_path_prefix, f"selkies_js{i}.sock")
        evdev_ip_sock_path = os.path.join(self.js_socket_path_prefix, f"selkies_event{1000+i}.sock")

        gamepad = SelkiesGamepad(
            js_ip_sock_path, evdev_ip_sock_path, self.loop,
            uinput_enabled=self.uinput_gamepads,
        )

        gamepad_name_for_interposer = STANDARD_XPAD_CONFIG.get("name", f"Selkies Virtual Gamepad {i}")
        std_num_btns = len(STANDARD_XPAD_CONFIG["btn_map"])
        std_num_axes = len(STANDARD_XPAD_CONFIG["axes_map"])

        gamepad.set_config(gamepad_name_for_interposer, std_num_btns, std_num_axes)

        self._spawn_task(gamepad.run_servers())
        _persistent_gamepads[i] = gamepad
        self.gamepad_instances[i] = gamepad
        logger_webrtc_input.info(f"Initialized and started persistent gamepad instance for index {i} (Name: '{gamepad_name_for_interposer}', JS: {js_ip_sock_path}, EVDEV: {evdev_ip_sock_path}).")

async def disconnect(self) -> None:
    """Tear down this handler's own resources; persistent gamepads stay up.

    Only the per-session client-to-slot associations are this handler's
    (see SelkiesGamepad). The input X connection is closed, not just
    dropped: the keyboard shim holds the same Display and python-xlib's
    root-window back-reference makes the graph a cycle with no finalizer,
    so a dropped connection would keep its X client slot until a cyclic
    collection happens to run — and every transport switch builds a new
    handler.
    """
    logger_webrtc_input.info("Releasing gamepad associations (persistent instances stay up).")
    await self.__gamepad_disconnect()
    self.gamepad_instances = {}
    self.gamepad_heartbeats.clear()
    # Before the pointer backends go away, or a held button stays pressed for good.
    await self.release_mouse_buttons()
    self.__mouse_disconnect()
    self._disarm_x_event_watcher()
    if self.keyboard is not None:
        self.keyboard.release_group_lock()
    old_display, self.xdisplay, self.keyboard = self.xdisplay, None, None
    if old_display is not None:
        try:
            await asyncio.to_thread(old_display.close)
        except Exception as e:
            logger_webrtc_input.debug(f"closing the input X connection failed: {e}")

    if self.keyboard_worker_task:
        self.keyboard_worker_task.cancel()
        self.keyboard_worker_task = None
    if self.key_sweep_task:
        self.key_sweep_task.cancel()
        self.key_sweep_task = None
    if self.key_repeat_task:
        self.key_repeat_task.cancel()
        self.key_repeat_task = None
    if self.keymap_watch_task:
        self.keymap_watch_task.cancel()
        self.keymap_watch_task = None
    self.pressed_keys.clear()
    self.reaped_atomic_keys.clear()
    self.key_repeat_state.clear()
    self._reset_multipart_clipboard()

async def _key_stale_sweep(self) -> None:
    """Auto-release keys and neutralize gamepads whose heartbeats stopped,
    so no input stays stuck held when a key-up is lost to congestion or the
    client vanishes without a transport close.

    On Wayland a release rides the serialized keyboard queue. X11
    injection is not queue-serialized, so the modifier/atomic state
    discard is deferred until after the release and a concurrent kd is
    checked for on both sides of the await: the keysym is popped first, so
    a non-None entry means a kd re-pressed it and already injected its own
    keydown, and the sweep then abandons its release (a second down would
    double-press) and leaves that kd's state intact.
    """
    try:
        while True:
            await asyncio.sleep(self.key_sweep_interval)
            if not self.pressed_keys and not self.gamepad_heartbeats:
                continue
            now = time.monotonic()
            stale = [k for k, seen in self.pressed_keys.items() if now - seen > self.key_stale_window]
            for keysym in stale:
                # A heartbeat or re-press during a prior await may have
                # refreshed this key.
                seen = self.pressed_keys.get(keysym)
                if seen is None or time.monotonic() - seen <= self.key_stale_window:
                    continue
                was_atomic = keysym in self.atomically_typed_keys
                self.pressed_keys.pop(keysym, None)
                self.key_repeat_state.pop(keysym, None)
                logger_webrtc_input.warning(f"Auto-releasing key {keysym} (heartbeat lost).")
                # An atomically-typed key was never physically held on X11.
                if was_atomic and not self.is_wayland:
                    self.atomically_typed_keys.discard(keysym)
                    if len(self.reaped_atomic_keys) < self.max_pressed_keys:
                        self.reaped_atomic_keys.add(keysym)
                    continue
                try:
                    if self.is_wayland:
                        self.active_modifiers.discard(keysym)
                        self.atomically_typed_keys.discard(keysym)
                        self._keyboard_enqueue(("ku", keysym))
                    else:
                        if self.pressed_keys.get(keysym) is not None:
                            # A kd raced us: it owns the key now.
                            continue
                        await self.send_x11_keypress(keysym, down=False)
                        if self.pressed_keys.get(keysym) is not None:
                            # A kd raced the keyup await: leave its state intact.
                            continue
                        self.active_modifiers.discard(keysym)
                        self.atomically_typed_keys.discard(keysym)
                except Exception as e:
                    logger_webrtc_input.warning(f"Failed to auto-release key {keysym}: {e}")
            for idx, seen in list(self.gamepad_heartbeats.items()):
                if now - seen <= self.key_stale_window:
                    continue
                self.gamepad_heartbeats.pop(idx, None)
                gamepad = self.gamepad_instances.get(idx)
                if gamepad is not None and gamepad._held_controls:
                    logger_webrtc_input.warning(
                        f"Neutralizing gamepad slot {idx} (heartbeat lost).")
                    gamepad.reset_state()
    except asyncio.CancelledError:
        pass

async def _key_repeat_loop(self) -> None:
    """X11 server-side key auto-repeat for held keys.

    XTEST/xdotool synthetic presses don't trigger the X server's native
    auto-repeat, so without this a held key types a single character. We re-emit the
    most-recently-pressed still-held repeatable key at key_repeat_interval after an
    initial key_repeat_delay -- exactly like a physical keyboard, where only the last
    key pressed repeats and releasing it resumes the previously-held one. Modifiers
    are never armed, so the repeated key carries whatever modifiers are currently
    held (Shift+Arrow selection, Ctrl+Backspace word-delete, Ctrl+Z, etc. all repeat
    like native -- no special shortcut suppression); atomically-typed keys
    (digits/punctuation) are armed and repeat through the atomic path below.
    Repeats are KeyPress-only (no synthetic KeyRelease),
    matching X11 detectable auto-repeat, so state-based games keep the key held with
    no movement stutter and ignore the extra presses. Wayland is excluded (the
    focused app repeats held virtual-keyboard keys itself via wl_keyboard
    repeat_info, so a server-side repeat would double it).

    key_repeat_state is insertion-ordered and arming moves a key to the
    end, so its last entry is the newest held key. An atomic key repeats
    as a self-contained XTEST press+release at its shift level (the 'ku'
    path injects no key-up for atomic keys, so a lone press would stick),
    falling to the co,end path when it has no keycode in the layout.
    Repeat pauses while the key's heartbeats have stopped (stalled stream,
    hidden tab), bounding run-on to the grace rather than the stale window.
    """
    try:
        while True:
            await asyncio.sleep(self.key_repeat_tick)
            if not self.key_repeat_state:
                continue
            # Keys released/reaped since arming: never inject a down after the key-up.
            for k in [k for k in self.key_repeat_state if k not in self.pressed_keys]:
                self.key_repeat_state.pop(k, None)
            if not self.key_repeat_state:
                continue
            keysym = next(reversed(self.key_repeat_state))
            now = time.monotonic()
            if now < self.key_repeat_state[keysym]:
                continue
            last_seen = self.pressed_keys.get(keysym)
            if last_seen is None or (now - last_seen) > self.key_repeat_heartbeat_grace:
                continue
            try:
                if keysym in self.atomically_typed_keys:
                    injected = False
                    if self.keyboard is not None:
                        try:
                            self.keyboard.press(
                                keysym,
                                held_keysyms=(self.active_modifiers
                                              & self.LEVEL_MODIFIER_KEYSYMS))
                            self.keyboard.release(keysym)
                            injected = True
                        except Exception as e:
                            logger_webrtc_input.debug(
                                f"XTEST atomic repeat failed for keysym {keysym}; falling back: {e}"
                            )
                            if self._is_x_conn_closed(e):
                                self._reconnect_xdisplay()
                    if not injected:
                        unicode_codepoint = (keysym & 0x00FFFFFF
                                             if (keysym & 0xFF000000) == 0x01000000 else keysym)
                        char_to_type = chr(unicode_codepoint)
                        await self.on_message(f"co,end,{char_to_type}")
                else:
                    await self.send_x11_keypress(keysym, down=True)
            except Exception as e:
                logger_webrtc_input.warning(f"Key auto-repeat failed for {keysym}: {e}")
                self.key_repeat_state.pop(keysym, None)
                continue
            # A 'ku' during the await released the key; the extra down is
            # healed by the real key-up / stale sweep.
            if keysym in self.pressed_keys:
                self.key_repeat_state[keysym] = time.monotonic() + self.key_repeat_interval
            else:
                self.key_repeat_state.pop(keysym, None)
    except asyncio.CancelledError:
        pass

async def reset_keyboard(self) -> None:
    """Release every held key/modifier and normalize a stuck Caps Lock.

    Runs on client 'kr' (blur/visibility loss), at connect, and when a
    client holding input departs, on both backends: the client resolves
    letter case itself, so an engaged Lock modifier on the server would
    invert every letter it types. On Wayland every key goes through the
    serialized keyboard worker, so the reset is queued behind the key work
    in flight like a client 'kr' is (a press still queued behind a text
    batch would otherwise land after the reset, untracked and held for
    good); it runs in place only when no worker is draining the queue.

    On X11 every still-held key is released, not just the modifier and
    hotkey list: once pressed_keys is cleared the stale sweep (the only
    other releaser) never runs for them, so a held 'w' in a game would
    stay pressed forever. Atomically-typed keys were never physically
    held and are skipped to avoid a spurious keyup; translated (Cyrillic)
    keys go through send_x11_keypress's translation so the injected QWERTY
    key comes up. Caps Lock is not forwarded from the browser, but a prior
    session or the desktop's own startup can leave Lock engaged, so it is
    toggled back off.
    """
    if self.is_wayland:
        worker = self.keyboard_worker_task
        if (worker is None or worker.done()
                or asyncio.current_task() is worker):
            await self._reset_keyboard_wayland()
            return
        done = asyncio.get_running_loop().create_future()
        self._keyboard_enqueue(("kr", done))
        await asyncio.wait({done, worker}, timeout=self._WL_RESET_WAIT_S,
                           return_when=asyncio.FIRST_COMPLETED)
        if done.done():
            return
        if worker.done():
            # The worker died with the reset queued; nothing else injects now.
            await self._reset_keyboard_wayland()
        else:
            logger_webrtc_input.warning(
                "Keyboard reset still queued behind pending key work.")
        return

    if not self.keyboard or not self.xdisplay :
        logger_webrtc_input.warning("Cannot reset keyboard, X display or keyboard controller not available.")
        return
    logger_webrtc_input.info("Resetting keyboard modifiers.")
    lctrl, lshift, lalt, altgr = 65507, 65505, 65513, 65027
    rctrl, rshift, ralt = 65508, 65506, 65514
    lmeta, rmeta, keyf, keyF, keym, keyM, escape = 65511, 65512, 102, 70, 109, 77, 65307
    # Super/Hyper included: the client maps the Meta/Windows key to Super.
    lsuper, rsuper, lhyper, rhyper = 65515, 65516, 65517, 65518
    for k in [lctrl, lshift, lalt, altgr, rctrl, rshift, ralt, lmeta, rmeta,
              lsuper, rsuper, lhyper, rhyper, keyf, keyF, keym, keyM, escape]:
        try: await self.send_x11_keypress(k, down=False)
        except Exception as e: logger_webrtc_input.warning(f"Error resetting key {k}: {e}")
    for keysym in list(self.pressed_keys):
        if keysym in self.atomically_typed_keys:
            continue
        try: await self.send_x11_keypress(keysym, down=False)
        except Exception as e: logger_webrtc_input.warning(f"Error releasing held key {keysym}: {e}")
    for k in list(self.translated_keys):
        try: await self.send_x11_keypress(k, down=False)
        except Exception as e: logger_webrtc_input.warning(f"Error releasing translated key {k}: {e}")
    try:
        if self.xdisplay.screen().root.query_pointer().mask & Xlib.X.LockMask:
            caps_kc = self.xdisplay.keysym_to_keycode(0xffe5)
            if caps_kc:
                xtest.fake_input(self.xdisplay, Xlib.X.KeyPress, caps_kc)
                xtest.fake_input(self.xdisplay, Xlib.X.KeyRelease, caps_kc)
                self.xdisplay.flush()
    except Exception as e:
        logger_webrtc_input.warning(f"Could not normalize Lock modifier: {e}")
    # Every key is up, so a group lock of ours goes back at once, not after its linger.
    if self.keyboard is not None:
        self.keyboard.release_group_lock()
    # After the release loops, which consume this state.
    self.active_modifiers.clear()
    self.active_shortcut_modifiers.clear()
    self.atomically_typed_keys.clear()
    self.translated_keys.clear()
    # Or the stale sweep would auto-release keys the reset just cleared.
    self.pressed_keys.clear()
    self.reaped_atomic_keys.clear()
    self.key_repeat_state.clear()

async def _reset_keyboard_wayland(self) -> None:
    """The Wayland reset itself — see reset_keyboard for when it runs and
    why every held and translated key is released before the maps clear.

    Caps Lock is read back through get_keyboard_state — a compositor
    round-trip answered by the thread that also renders, so it waits off
    the loop — and toggled off with a virtual Caps_Lock press+release.
    """
    if self.wayland_input:
        # Ctrl, Shift, Alt, AltGr, Meta, Super, Hyper (the client maps the
        # Meta/Windows key to Super).
        modifiers = [65507, 65505, 65513, 65508, 65506, 65514, 65027,
                     65511, 65512, 65515, 65516, 65517, 65518]
        # A stale owner (rebuild backing off) still knows what is down.
        owner = await self._ensure_wayland_keymap_owner() or self._wl_keymap_owner
        if owner is not None:
            for k in modifiers:
                try:
                    owner.release(k)
                except Exception:
                    pass
            try:
                owner.reset()
            except Exception:
                pass
    for k in list(self.translated_keys):
        try: await self.send_x11_keypress(k, down=False)
        except Exception: pass
    for keysym in list(self.pressed_keys):
        if keysym in self.atomically_typed_keys:
            continue
        try: await self.send_x11_keypress(keysym, down=False)
        except Exception as e: logger_webrtc_input.warning(f"Error releasing held key {keysym}: {e}")
    self.active_modifiers.clear()
    self.active_shortcut_modifiers.clear()
    self.atomically_typed_keys.clear()
    self.translated_keys.clear()
    # Or the stale sweep would auto-release keys the reset just cleared.
    self.pressed_keys.clear()
    self.reaped_atomic_keys.clear()
    try:
        if self.wayland_input and hasattr(self.wayland_input, 'get_keyboard_state'):
            _pressed, mods = await asyncio.to_thread(self.wayland_input.get_keyboard_state)
            # mods bit 4 is caps_lock.
            if mods & 0x10:
                keymap_owner = await self._ensure_wayland_keymap_owner()
                if keymap_owner is not None:
                    keymap_owner.press(0xffe5)
                    await asyncio.sleep(0.02)
                    keymap_owner.release(0xffe5)
    except Exception as e:
        logger_webrtc_input.debug(f"Wayland caps-lock normalization skipped: {e}")

async def release_mouse_buttons(self) -> None:
    """Release every pointer button still held server-side.

    The ungraceful pointer path, mirroring release_gamepads_for_conn and the
    key stale-sweep: a client that dies mid-drag never sends the mask with the
    button cleared, so the desktop would keep the drag or selection alive until
    some future client happens to diff the bit away. The releases ride the
    normal mask-diff loop, so X11 and Wayland both heal; the zero-delta
    relative move leaves the pointer exactly where it is (and injects no motion
    at all on Wayland)."""
    if not self.button_mask:
        return
    try:
        await self.send_x11_mouse(0, 0, 0, 0, relative=True)
    except Exception as e:
        logger_webrtc_input.warning(f"Failed to release held mouse buttons: {e}")

def send_mouse(self, action: int, data: Any) -> None:
    """Route one MOUSE_* action to the uinput socket or XTEST backend.

    Args:
        action: A MOUSE_* constant.
        data: Action-dependent — an (x, y) pair for position/move, a
            (press/release, button-id) pair for MOUSE_BUTTON, None for
            scroll actions.
    """
    if action == MOUSE_POSITION:
        if self.mouse: self.mouse.position = data
    elif action == MOUSE_MOVE:
        x, y = data
        if self.uinput_mouse_socket_path:
            self.__mouse_emit(UINPUT_REL_X, x, syn=False)
            self.__mouse_emit(UINPUT_REL_Y, y)
        elif self.xdisplay:
            xtest.fake_input(self.xdisplay, Xlib.X.MotionNotify, detail=True, root=Xlib.X.NONE, x=x, y=y)
            # flush(), not sync(): XTEST needs no reply and sync() is a round trip per move.
            self.xdisplay.flush()
    elif action == MOUSE_SCROLL_UP:
        # MOUSE_SCROLL_* are named for the client button, not the physical
        # direction: this is wheel-down (X button 5), so REL_WHEEL is -1.
        if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_WHEEL, -1)
        elif self.mouse: self.mouse.scroll(0, -1)
    elif action == MOUSE_SCROLL_DOWN:
        if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_WHEEL, 1)
        elif self.mouse: self.mouse.scroll(0, 1)
    elif action == MOUSE_SCROLL_LEFT:
        # REL_HWHEEL is signed as the client names it (negative = left), like
        # X buttons 6/7, so no flip here.
        if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_HWHEEL, -1)
        elif self.mouse: self.mouse.scroll(-1, 0)
    elif action == MOUSE_SCROLL_RIGHT:
        if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_HWHEEL, 1)
        elif self.mouse: self.mouse.scroll(1, 0)
    elif action == MOUSE_BUTTON: 
        btn_map_key = "uinput" if self.uinput_mouse_socket_path else "x11"
        btn_uinput_or_x11 = MOUSE_BUTTON_MAP[data[1]][btn_map_key]
        if data[0] == MOUSE_BUTTON_PRESS:
            if self.uinput_mouse_socket_path: self.__mouse_emit(btn_uinput_or_x11, 1)
            elif self.mouse: self.mouse.press(btn_uinput_or_x11)
        else:
            if self.uinput_mouse_socket_path: self.__mouse_emit(btn_uinput_or_x11, 0)
            elif self.mouse: self.mouse.release(btn_uinput_or_x11)

async def send_x11_keypress(self, keysym: int, down: bool = True,
                            neutralize: Optional[bool] = None) -> None:
    """Inject one key transition on whichever backend this session uses.

    Despite the name this is the shared key injector: Wayland routes
    through the seat keymap owner (virtual-keyboard/clipboard on error),
    X11 through XTEST with xdotool fallbacks. Cyrillic keysyms chorded
    with an action modifier are translated to the QWERTY keysym on the
    same physical key so shortcuts (Ctrl+C on a ЙЦУКЕН layout) reach the
    application as the app expects.

    On X11 a keysym with a keycode in the current keymap is injected
    through XTEST on the already-open display, which spares a ~15 ms
    xdotool fork per shortcut, arrow or function key; a keysym the layout
    lacks is overlay-bound once by the shim and reused, and one only a
    later layout group carries is injected under that group's lock —
    never a per-key xdotool fork, whose transient rebind floods
    MappingNotify and lags the whole input queue behind real typing.
    xdotool remains for a keysym with no keycode at all that it can still
    synthesize.

    Args:
        neutralize: Whether a conflicting held Shift/AltGr is lifted around
            the key. None derives it from the client's held modifiers:
            lifted around plain keystrokes only, since while a chord
            modifier (Ctrl/Alt/Super/...) is down every held modifier is
            part of the chord. A server-synthesized chord passes False,
            since a Shift it pressed through this injector is not in
            active_modifiers and would be lifted for the very key it
            modifies.
    """
    if down:
        if (self.active_modifiers & self.ACTION_MODIFIER_KEYSYMS) and keysym in CYRILLIC_TO_QWERTY_KEYSYM:
            self.translated_keys.add(keysym)
            keysym = CYRILLIC_TO_QWERTY_KEYSYM[keysym]
    else:
        if keysym in self.translated_keys:
            self.translated_keys.discard(keysym)
            keysym = CYRILLIC_TO_QWERTY_KEYSYM[keysym]

    if neutralize is None:
        neutralize = (keysym not in self.MODIFIER_KEYSYMS
                      and not (self.active_modifiers & self.ACTION_MODIFIER_KEYSYMS))
    held_level_mods = frozenset(self.active_modifiers & self.LEVEL_MODIFIER_KEYSYMS)

    if self.is_wayland and self.wayland_input:
        owner = await self._ensure_wayland_keymap_owner()
        if owner is not None:
            try:
                if down:
                    owner.press(keysym, neutralize=neutralize)
                else:
                    owner.release(keysym)
                return
            except Exception as e:
                logger_webrtc_input.warning(
                    f"Wayland keymap injection failed for keysym {keysym}; falling back: {e}"
                )
        await self._type_keysym_fallback(keysym, down)
        return

    is_printable = (0x20 <= keysym <= 0xFF) or ((keysym & 0xFF000000) == 0x01000000)
    action = "keydown" if down else "keyup"
    command = None
    use_keyboard_for_printable = False
    allow_xtest = False
    if is_printable:
        unicode_codepoint = keysym & 0x00FFFFFF if (keysym & 0xFF000000) == 0x01000000 else keysym
        try:
            char = chr(unicode_codepoint)
            if char.isalpha():
                use_keyboard_for_printable = True
            else:
                xdotool_arg = f"U{unicode_codepoint:04X}"
                if not self.active_shortcut_modifiers:
                    use_keyboard_for_printable = True
                else:
                    command = ["xdotool", action, xdotool_arg]
                    allow_xtest = True
        except ValueError:
            use_keyboard_for_printable = True

    else:
        map_entry = X11_KEYSYM_MAP.get(keysym)
        if map_entry:
            xdotool_arg = map_entry.get('xkey_name')
            if xdotool_arg:
                command = ["xdotool", action, xdotool_arg]
                allow_xtest = True
                if xdotool_arg in self.SHORTCUT_MODIFIER_XKEY_NAMES:
                    if down:
                        self.active_shortcut_modifiers.add(xdotool_arg)
                    else:
                        self.active_shortcut_modifiers.discard(xdotool_arg)

    if command:
        if allow_xtest and xtest is not None and self.xdisplay is not None:
            try:
                keycode = self.xdisplay.keysym_to_keycode(keysym)
                if self.keyboard and self.keyboard.outside_base_group(keysym):
                    # A bare keycode would type the group-1 glyph; the shim
                    # locks the group around the injection instead.
                    keycode = 0
                if keycode:
                    xtest.fake_input(
                        self.xdisplay,
                        X.KeyPress if down else X.KeyRelease,
                        keycode,
                    )
                    self.xdisplay.flush()
                    return
                if self.keyboard:
                    if down:
                        self.keyboard.press(keysym, neutralize=neutralize,
                                            held_keysyms=held_level_mods)
                    else:
                        self.keyboard.release(keysym)
                    return
            except Exception as e:
                logger_webrtc_input.debug(
                    f"XTEST inject failed for keysym {keysym}; falling back to xdotool: {e}"
                )
        try:
            process = await subprocess.create_subprocess_exec(
                *command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
            )
            await self._communicate_or_kill(process, 0.5, "xdotool key")
            if process.returncode == 0:
                return
            logger_webrtc_input.warning(
                f"xdotool {action} failed (rc={process.returncode}) for keysym {keysym}")
        except Exception:
            pass

    if use_keyboard_for_printable or not command:
        try:
            if not self.keyboard:
                await self._type_keysym_fallback(keysym, down)
                return

            if down:
                self.keyboard.press(keysym, neutralize=neutralize,
                                    held_keysyms=held_level_mods)
            else:
                self.keyboard.release(keysym)
        except Exception as e:
            if self._is_x_conn_closed(e):
                self._reconnect_xdisplay()
            await self._type_keysym_fallback(keysym, down)