_XTestKeyboard

def __init__(self, xdisplay: Any) -> None:
    self._d = xdisplay
    self._xkb = open_xkb_link(xdisplay) if open_xkb_link is not None else None
    self._group_hold = None
    self._group_restore = None
    self._shift_kc = xdisplay.keysym_to_keycode(0xffe1)
    self._shift_r_kc = xdisplay.keysym_to_keycode(0xffe2)
    self._altgr_kc = (xdisplay.keysym_to_keycode(0xfe03)
                      or xdisplay.keysym_to_keycode(0xff7e))
    self._effective_mod_keycodes = self._read_effective_modifiers()
    self._synth_mods = {}
    self._spare_keycodes = None
    self._spare_set = frozenset()
    self._overlay = {}
    self._overlay_value_kc = {}
    self._overlay_order = []
    self._pressed_kc = {}
    self._dirty_spares = set()

def _find_spare_keycodes(self) -> list:
    """Every keycode free to repurpose for the overlay.

    Spare means all levels NoSymbol, or carrying a previous overlay bind —
    every level the SAME Unicode-plane keysym, a shape no real layout
    produces (the server echoes a two-sym bind back expanded across all
    levels). Overlay binds persist on the X server across a handler
    restart, so without reclaiming them each restart would shrink the pool
    until typing beyond the layout starves. Modifier-mapped keycodes are
    never spare: pressing one would toggle its modifier under the typed
    char. The full range is scanned (not a fixed cap): more slots make
    recycling — the only case where a slow app can mistranslate a rebound
    keycode — rare.
    """
    info = self._d.display.info
    lo, hi = info.min_keycode, info.max_keycode
    mapping = self._d.get_keyboard_mapping(lo, hi - lo + 1)
    try:
        mod_keycodes = {kc for row in self._d.get_modifier_mapping()
                        for kc in row if kc}
    except Exception:
        mod_keycodes = set()
    spares = []
    for i, syms in enumerate(mapping):
        kc = lo + i
        if kc in mod_keycodes:
            continue
        bound = {s for s in syms if s}
        if not bound:
            spares.append(kc)
        elif len(bound) == 1:
            sym = next(iter(bound))
            if (sym & 0xFF000000) == 0x01000000:
                # Clients may translate it by its old value until the
                # rebind's MappingNotify lands: first bind settles like a recycle.
                spares.append(kc)
                self._dirty_spares.add(kc)
    self._spare_set = frozenset(spares)
    return spares

def _free_spares(self) -> list:
    """Spare keycodes not currently bound, in pool order."""
    if self._spare_keycodes is None:
        self._spare_keycodes = self._find_spare_keycodes()
    used = set(self._overlay.values())
    return [kc for kc in self._spare_keycodes if kc not in used]

def _layout_keycode(self, keysym: int) -> int:
    """keysym_to_keycode, distrusting hits on spare-pool keycodes: the
    display's cached lookup can name a keycode whose bind belongs to a
    previous handler — or to a DIFFERENT keysym after this pool re-purposed
    it — so on a pool keycode only this shim's own live bind carrying
    exactly this value counts. Only keysym forms an overlay bind can carry
    (Latin-1 high half, Unicode plane) can hit the pool, so everything
    else skips the discovery."""
    kc = self._d.keysym_to_keycode(keysym)
    if not kc:
        return 0
    if (keysym & 0xFF000000) == 0x01000000 or 0xA0 <= keysym <= 0xFF:
        if self._spare_keycodes is None:
            self._spare_keycodes = self._find_spare_keycodes()
        if kc in self._spare_set:
            return self._overlay_value_kc.get(keysym, 0)
    return kc

def _alloc_overlay_keycode(self, keysym: int) -> tuple:
    """Reserve a spare keycode for keysym and record the binding.

    Recycles the oldest binding when the pool is full. The mapping request
    itself is the caller's (single vs batched).

    Returns:
        (keycode, needs_settle) — needs_settle is True when clients may
        still hold a previous mapping for the keycode (an in-process
        recycle, or a reclaimed spare's first bind).
    """
    free = self._free_spares()
    if free:
        kc = free[0]
        needs_settle = kc in self._dirty_spares
        self._dirty_spares.discard(kc)
    else:
        oldest = self._overlay_order.pop(0)
        kc = self._overlay.pop(oldest)
        self._overlay_value_kc.pop(overlay_bind_keysym(oldest), None)
        needs_settle = True
    self._overlay[keysym] = kc
    self._overlay_value_kc[overlay_bind_keysym(keysym)] = kc
    self._overlay_order.append(keysym)
    return kc, needs_settle

def _overlay_keycode(self, keysym: int) -> Optional[int]:
    """Bind an unmapped keysym to a spare keycode (recycling the oldest) and
    return it, or None if no spare keycode exists."""
    if keysym in self._overlay:
        return self._overlay[keysym]
    if self._spare_keycodes is None:
        self._spare_keycodes = self._find_spare_keycodes()
    if not self._spare_keycodes:
        return None
    kc, needs_settle = self._alloc_overlay_keycode(keysym)
    # Bound at levels 0 and 1 so an accidental Shift cannot change it.
    bind_value = overlay_bind_keysym(keysym)
    self._d.change_keyboard_mapping(kc, [[bind_value, bind_value]])
    self._d.sync()
    if needs_settle:
        time.sleep(self._RECYCLE_SETTLE_S)
    return kc

def prebind(self, keysyms: Iterable[int]) -> bool:
    """Overlay-bind every unmapped keysym in as few requests as possible.

    One ChangeKeyboardMapping per contiguous spare-keycode run, one sync —
    so a CJK composition commit broadcasts O(1) MappingNotify events
    instead of one per new char; the longest free runs are taken first to
    keep that count down.

    Returns:
        False (nothing bound) when more new keysyms than slots exist, since
        filling would recycle bindings made earlier in the same batch and
        corrupt the text; the caller then falls back without partial
        typing. True otherwise.
    """
    d = self._d
    missing = []
    for ks in dict.fromkeys(keysyms):
        if ks not in self._overlay and not self._layout_keycode(ks):
            missing.append(ks)
    if not missing:
        return True
    if self._spare_keycodes is None:
        self._spare_keycodes = self._find_spare_keycodes()
    if len(missing) > len(self._spare_keycodes):
        return False
    free = self._free_spares()
    runs = []
    i = 0
    while i < len(free):
        j = i
        while j + 1 < len(free) and free[j + 1] == free[j] + 1:
            j += 1
        runs.append(free[i:j + 1])
        i = j + 1
    runs.sort(key=len, reverse=True)
    picked = []
    for run in runs:
        if len(picked) >= len(missing):
            break
        picked.extend(run[:len(missing) - len(picked)])
    recycled_any = any(kc in self._dirty_spares for kc in picked)
    self._dirty_spares.difference_update(picked)
    while len(picked) < len(missing):
        oldest = self._overlay_order.pop(0)
        picked.append(self._overlay.pop(oldest))
        self._overlay_value_kc.pop(overlay_bind_keysym(oldest), None)
        recycled_any = True
    assigns = []
    for ks, kc in zip(missing, picked):
        self._overlay[ks] = kc
        self._overlay_value_kc[overlay_bind_keysym(ks)] = kc
        self._overlay_order.append(ks)
        assigns.append((kc, ks))
    assigns.sort()
    i = 0
    while i < len(assigns):
        j = i
        while j + 1 < len(assigns) and assigns[j + 1][0] == assigns[j][0] + 1:
            j += 1
        d.change_keyboard_mapping(
            assigns[i][0],
            [[overlay_bind_keysym(ks)] * 2 for _kc, ks in assigns[i:j + 1]])
        i = j + 1
    d.sync()
    if recycled_any:
        time.sleep(self._RECYCLE_SETTLE_S)
    return True

def bindings_intact(self) -> bool:
    """True when every overlay binding still resolves to its keysym in the
    server's map. Distinguishes our own MappingNotify from a foreign layout
    change by SEMANTICS: servers vary in how many notifies one
    ChangeKeyboardMapping emits and report the full keycode range, so
    neither counting nor range matching works — but a self-bind leaves the
    bindings intact and a foreign change wipes them."""
    if not self._overlay:
        return True
    try:
        for ks, kc in self._overlay.items():
            syms = self._d.get_keyboard_mapping(kc, 1)[0]
            if not len(syms) or syms[0] != overlay_bind_keysym(ks):
                return False
        return True
    except Exception:
        return False

def refresh_modifier_keycodes(self) -> None:
    """Re-resolve the synth-modifier keycodes from the (already refreshed)
    cache — a modifier remap moves them without touching the overlay."""
    d = self._d
    self._shift_kc = d.keysym_to_keycode(0xffe1)
    self._shift_r_kc = d.keysym_to_keycode(0xffe2)
    self._altgr_kc = (d.keysym_to_keycode(0xfe03)
                      or d.keysym_to_keycode(0xff7e))
    self._effective_mod_keycodes = self._read_effective_modifiers()

def _read_effective_modifiers(self) -> Optional[set]:
    """Keycodes the server actually treats as modifiers.

    Holding a key only selects a shifted level when that keycode is bound in
    the modifier map. A keymap can carry the Shift keysym without binding it
    (a bare Xvfb with no keymap is the common case), and injecting Shift there
    types the level-0 glyph instead: every capital arrives lowercase.
    """
    try:
        return {kc for row in self._d.get_modifier_mapping() for kc in row if kc}
    except Exception as e:
        logger_webrtc_input.debug(f"modifier map unreadable ({e}); assuming it binds what it names")
        return None

def invalidate_mapping(self) -> None:
    """A foreign keymap change (setxkbmap, desktop layout switcher) wiped
    our overlay bindings and may have moved modifier keycodes: drop the
    overlay bookkeeping, rediscover spares lazily and re-resolve the
    modifier keycodes. Held keys are kept: release replays the exact
    press-time keycode."""
    self._overlay.clear()
    self._overlay_value_kc.clear()
    self._overlay_order.clear()
    self._spare_keycodes = None
    self._spare_set = frozenset()
    self._dirty_spares.clear()
    if self._xkb is not None:
        self._xkb.invalidate()
    self.refresh_modifier_keycodes()

def note_mapping_change(self, first_keycode: int, count: int) -> None:
    """A keyboard MappingNotify arrived for this keycode range: refetch the
    XKB placement on the next lookup unless the range lies inside the spare
    pool, where only overlay binds live and the placement never trusts
    them anyway — so this shim's own binds cost no refetch."""
    if self._xkb is None:
        return
    if self._spare_keycodes is not None and all(
            kc in self._spare_set for kc in range(first_keycode, first_keycode + count)):
        return
    self._xkb.invalidate()

def keyboard_replaced(self, event: Any) -> Optional[tuple]:
    """`(min_keycode, max_keycode)` when the event announces a replaced
    keyboard on the XKB link, else None."""
    if self._xkb is None:
        return None
    return self._xkb.replaced_keyboard(event)

def outside_base_group(self, keysym: int) -> bool:
    """Whether this keysym must inject through press()/release() rather
    than a bare keycode: it is down under a group lock of this shim's, or
    the layout carries it only in a group past the first."""
    if keysym in self._pressed_kc and self._group_hold is not None:
        return True
    if self._xkb is None:
        return False
    try:
        placed = self._xkb.locate(keysym)
    except Exception as e:
        logger_webrtc_input.debug(f"XKB placement lookup failed ({e}); core keymap only")
        return False
    return placed is not None and placed[1] != 0

def layout_carries(self, keysym: int) -> bool:
    """Whether the layout itself (not an overlay bind) carries the keysym."""
    try:
        if self._xkb is not None:
            return self._placement(keysym) is not None
    except Exception as e:
        logger_webrtc_input.debug(f"XKB placement lookup failed ({e}); core keymap only")
    return bool(self._layout_keycode(keysym))

def _placement(self, keysym: int) -> Optional[tuple]:
    """`(keycode, group, level)` from the XKB map, with the spare-pool
    distrust of _layout_keycode applied; None when XKB is unavailable or
    the keymap lacks the keysym."""
    if self._xkb is None:
        return None
    placed = self._xkb.locate(keysym)
    if placed is None:
        return None
    kc, group, level = placed
    if (keysym & 0xFF000000) == 0x01000000 or 0xA0 <= keysym <= 0xFF:
        if self._spare_keycodes is None:
            self._spare_keycodes = self._find_spare_keycodes()
        if kc in self._spare_set:
            kc = self._overlay_value_kc.get(keysym, 0)
            if not kc:
                return None
            return kc, 0, 0
    return kc, group, level

def _resolve(self, keysym: int) -> tuple:
    """Return (keycode, modifier_keycodes, group) to inject this keysym.

    The modifiers are the Shift / AltGr keycodes whose held state selects
    the keymap level the keysym sits at, so a glyph bound above the Shift
    level (e.g. AltGr '@') types correctly instead of falling through to
    its level-0 glyph. The group is the layout group the keycode carries
    the keysym in; None for an overlay keycode, which carries one group
    and so types the same under any lock, and when XKB is unavailable and
    the core keymap's flattened columns are all there is.

    A keysym the layout lacks binds to a spare keycode in-process (no
    xdotool fork); overlay keysyms sit at level 0 and never need
    modifiers. The same bind is used when the level a glyph sits at is
    unreachable because its modifier keycode is not in the modifier map:
    the glyph then carries its own case instead of depending on a
    modifier the server will not act on.

    Raises:
        ValueError: No keycode exists and no spare keycode can be bound;
            the caller falls back to xdotool.
    """
    d = self._d
    group = None
    placed = None
    xkb_answered = self._xkb is not None
    if xkb_answered:
        try:
            placed = self._placement(keysym)
        except Exception as e:
            logger_webrtc_input.debug(f"XKB placement lookup failed ({e}); core keymap only")
            xkb_answered = False
    if placed is not None:
        kc, group, level = placed
    elif xkb_answered:
        kc = 0
    else:
        kc = self._layout_keycode(keysym)
        # Lowest column carrying this glyph: 0 base, 1 Shift, 2 AltGr, 3 Shift+AltGr.
        level = next((lvl for lvl in range(4)
                      if d.keycode_to_keysym(kc, lvl) == keysym), 0)
    if not kc:
        kc = self._overlay_keycode(keysym)
        if not kc:
            raise ValueError("no keycode for keysym %r" % (keysym,))
        return kc, (), None
    mods = []
    if level & 1 and self._shift_kc:
        mods.append(self._shift_kc)
    if level & 2 and self._altgr_kc:
        mods.append(self._altgr_kc)
    if mods and not self._modifiers_engage(mods):
        overlay_kc = self._overlay_keycode(keysym)
        if overlay_kc:
            return overlay_kc, (), None
    return kc, tuple(mods), group

def _modifiers_engage(self, mods: Iterable[int]) -> bool:
    """Whether holding these keycodes actually selects a shifted level."""
    effective = getattr(self, "_effective_mod_keycodes", None)
    if effective is None:
        return True
    return all(kc in effective for kc in mods)

def _down_mod_keycodes(self, held_keysyms: Iterable[int]) -> set:
    """Shift/AltGr keycodes currently down: the client-held keysyms the
    handler tracks, plus this shim's own synthesized holds. Tracked state
    only, so no press ever pays a query_keymap server round trip."""
    down = set()
    for mods in self._synth_mods.values():
        down.update(mods)
    for ks in held_keysyms:
        if ks == 0xFFE1:
            down.add(self._shift_kc)
        elif ks == 0xFFE2:
            down.add(self._shift_r_kc)
        elif ks in (0xFE03, 0xFF7E):
            down.add(self._altgr_kc)
    down.discard(0)
    return down

def _mods_to_lift(self, required: Container[int], down: Container[int]) -> list:
    """Down Shift/AltGr keycodes the target level does not want. A required
    Shift is satisfied by either side, so neither is lifted then."""
    lift = []
    if self._shift_kc not in required:
        lift.extend(kc for kc in (self._shift_kc, self._shift_r_kc)
                    if kc in down)
    if self._altgr_kc and self._altgr_kc not in required and self._altgr_kc in down:
        lift.append(self._altgr_kc)
    return lift

def press(self, keysym: int, neutralize: bool = False,
          held_keysyms: Iterable[int] = ()) -> None:
    """Press a keysym via XTEST, synthesizing the modifiers its level needs.

    Only modifiers not already down are synthesized (a required Shift held
    on either side counts), and only those are undone by release().

    Args:
        neutralize: Lift a held Shift/AltGr the level does not want around
            the press — it would select a different glyph, or push an
            overlay bind onto its empty AltGr levels; chords pass False so
            Ctrl+Shift+X keeps its held modifiers.
        held_keysyms: Level-selecting modifier keysyms the client itself
            holds, consulted instead of a per-press server query.
    """
    kc, mods, group = self._resolve(keysym)
    self._enter_group(keysym, group)
    down = self._down_mod_keycodes(held_keysyms)
    lifted = self._mods_to_lift(set(mods), down) if neutralize else []
    for m in lifted:
        xtest.fake_input(self._d, Xlib.X.KeyRelease, m)
    synth = [m for m in mods
             if m not in down
             and not (m == self._shift_kc and self._shift_r_kc in down)]
    for m in synth:
        xtest.fake_input(self._d, Xlib.X.KeyPress, m)
    if synth:
        self._synth_mods[keysym] = synth
    xtest.fake_input(self._d, Xlib.X.KeyPress, kc)
    self._pressed_kc[keysym] = kc
    for m in reversed(lifted):
        xtest.fake_input(self._d, Xlib.X.KeyPress, m)
    self._d.flush()

def release(self, keysym: int) -> None:
    """Release a keysym, replaying its press-time keycode.

    Only an untracked press is re-resolved: the layout may have changed
    mid-keystroke, so a re-resolve of a tracked one could miss the key.
    """
    kc = self._pressed_kc.pop(keysym, None)
    if kc is None:
        kc, _, _ = self._resolve(keysym)
    self._leave_group(keysym)
    xtest.fake_input(self._d, Xlib.X.KeyRelease, kc)
    for m in reversed(self._synth_mods.pop(keysym, ())):
        xtest.fake_input(self._d, Xlib.X.KeyRelease, m)
    self._settle_group()
    self._d.flush()

def _enter_group(self, keysym: int, group: Optional[int]) -> None:
    """Lock the group a press needs, queued ahead of the press itself.

    Without a hold of ours in force, a group-1 keysym injects under the
    server's own lock (a later group locked by the user's desktop switcher
    then decides, as for any single-group layout) and only a keysym of a
    later group reads the locked group — one round trip — and switches
    when it differs. While a hold is in force every key is tracked under
    it, so a group-1 key typed during a Cyrillic run switches back and the
    original lock returns once the last tracked key is up.
    """
    if group is None or self._xkb is None:
        return
    hold = self._group_hold
    if hold is None:
        if group == 0:
            return
        before = self._xkb.locked_group()
        if before == group:
            return
        self._xkb.lock_group(group)
        self._group_hold = [before, group, {keysym: group}]
        return
    self._cancel_group_restore()
    if hold[1] != group:
        self._xkb.lock_group(group)
        hold[1] = group
    hold[2][keysym] = group

def _leave_group(self, keysym: int) -> None:
    """Take a key out of the group hold ahead of its release, putting its
    press-time group back if a later key moved the lock on, so the release
    carries the keysym the press did."""
    hold = self._group_hold
    if hold is None:
        return
    group = hold[2].pop(keysym, None)
    if group is not None and group != hold[1]:
        self._xkb.lock_group(group)
        hold[1] = group

def _settle_group(self) -> None:
    """With no tracked key left down, schedule the restore of the lock
    found before the switch — immediately when no event loop is running to
    defer it."""
    hold = self._group_hold
    if hold is None or hold[2]:
        return
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        self._restore_group()
        return
    self._cancel_group_restore()
    self._group_restore = loop.call_later(self._GROUP_LINGER_S, self._restore_group)

def _cancel_group_restore(self) -> None:
    if self._group_restore is not None:
        self._group_restore.cancel()
        self._group_restore = None

def _restore_group(self) -> None:
    """The linger elapsed: put the group lock back unless a key of ours
    is still down under it."""
    self._group_restore = None
    hold = self._group_hold
    if hold is None or hold[2]:
        return
    self.release_group_lock()

def release_group_lock(self) -> None:
    """Put back the group lock found before this shim switched it, now:
    after a keyboard reset, or before the connection closes, no keystroke
    follows that a lingering lock would serve, and a lock left behind
    would have the desktop translate every later key under it."""
    self._cancel_group_restore()
    hold, self._group_hold = self._group_hold, None
    if hold is None or hold[0] == hold[1] or self._xkb is None:
        return
    try:
        self._xkb.lock_group(hold[0])
        self._d.flush()
    except Exception as e:
        logger_webrtc_input.debug(f"group lock restore failed: {e}")