_WaylandKeymapOwner

def __init__(self, wayland_input: Any, base_keymap_text: str) -> None:
    if libxkb is None:
        raise RuntimeError("libxkbcommon unavailable")
    if not base_keymap_text:
        raise RuntimeError("empty compositor keymap")
    self._input = wayland_input
    self._base_text = base_keymap_text
    self._map = {}
    self._overlay = {}
    self._overlay_order = []
    self._pressed = {}
    self._mod_refs = {}
    self._down = set()
    self._build_map()

def _build_map(self) -> None:
    """Compile the base keymap and index every keysym plus the overlay pool.

    One walk over the keymap fills the keysym-to-(keycode, level) map,
    collects overlay candidates, and resolves the synth-modifier keycodes.
    """
    ctx = libxkb.xkb_context_new(0)
    if not ctx:
        raise RuntimeError("xkb_context_new failed")
    try:
        # Trailing arguments are TEXT_V1 and NO_FLAGS.
        km = libxkb.xkb_keymap_new_from_string(
            ctx, self._base_text.encode(), 1, 0)
        if not km:
            raise RuntimeError("keymap compile failed")
        try:
            lo = libxkb.xkb_keymap_min_keycode(km)
            hi = libxkb.xkb_keymap_max_keycode(km)
            syms = ctypes.POINTER(ctypes.c_uint32)()
            unbound, shadowable = [], []
            for kc in range(lo, hi + 1):
                levels = min(4, libxkb.xkb_keymap_num_levels_for_key(km, kc, 0))
                seen = spare = 0
                for level in range(levels):
                    n = libxkb.xkb_keymap_key_get_syms_by_level(
                        km, kc, 0, level, ctypes.byref(syms))
                    for i in range(n):
                        sym = syms[i]
                        if not sym:
                            continue
                        seen += 1
                        if self._VENDOR_FIRST <= sym <= self._VENDOR_LAST:
                            spare += 1
                        if sym not in self._map:
                            self._map[sym] = (kc, level)
                if kc < self._SUB256_CEILING and kc >= max(lo, 9):
                    if not seen:
                        unbound.append(kc)
                    elif seen == spare:
                        shadowable.append(kc)
            self._overlay_codes = unbound + shadowable + list(range(
                self._OVERFLOW_BASE_KEYCODE,
                self._OVERFLOW_BASE_KEYCODE + self._OVERFLOW_SLOTS))
        finally:
            libxkb.xkb_keymap_unref(km)
    finally:
        libxkb.xkb_context_unref(ctx)
    # Fallbacks are the conventional evdev+8 keycodes.
    self._shift_kc = self._map.get(0xFFE1, (50, 0))[0]
    self._shift_r_kc = self._map.get(0xFFE2, (62, 0))[0]
    self._altgr_kc = self._map.get(0xFE03, (108, 0))[0]

def resolves(self, keysym: int) -> bool:
    """Whether the base keymap carries this keysym on some key/level, so it
    can be injected without an overlay bind."""
    return keysym in self._map

def _held_conflicts(self, required: Container[int]) -> list:
    """Shift/AltGr keycodes currently down that the target level rejects.

    Held (client-pressed or synthesized), they would shift the injected key
    onto a different glyph. A required Shift is satisfied by either side,
    so neither Shift keycode is lifted then.
    """
    shift_wanted = self._shift_kc in required
    out = []
    for kc in (self._shift_kc, self._shift_r_kc, self._altgr_kc):
        if not kc or kc not in self._down or kc in required:
            continue
        if shift_wanted and kc in (self._shift_kc, self._shift_r_kc):
            continue
        out.append(kc)
    return out

def _inject(self, kc: int, state: int) -> None:
    (self._down.add if state else self._down.discard)(kc)
    self._input.inject_key(kc, state)

def _mods_for_level(self, level: int) -> tuple:
    """The Shift/AltGr keycodes whose held state selects this keymap level."""
    mods = []
    if level & 1:
        mods.append(self._shift_kc)
    if level & 2:
        mods.append(self._altgr_kc)
    return tuple(mods)

def _overlay_text(self) -> str:
    """Base keymap text with the occupied overlay slots bound at level 0.

    Only slots in use are declared, at the keycodes the pool handed out
    (mostly existing sub-256 codes being shadowed); hex keysym literals
    need no names. The base's own maximum is kept: a pc105 keymap binds a
    couple of hundred keycodes above 255, and lowering the ceiling would
    drop every one of them.
    """
    base = self._base_text
    max_at = base.index("maximum = ")
    max_end = base.index(";", max_at)
    used = sorted(self._overlay.values())
    base_max = int(base[max_at + len("maximum = "):max_end].strip())
    parts = [base[:max_at],
             f"maximum = {max([base_max] + used)}"]
    rest = base[max_end:]
    kc_end = rest.index("};")
    parts.append(rest[:kc_end])
    for kc in used:
        parts.append(f"\t<UC{kc:03}> = {kc};\n")
    rest = rest[kc_end:]
    sym_at = rest.index("xkb_symbols")
    open_at = rest.index("{", sym_at)
    depth = 0
    close_at = None
    for idx in range(open_at, len(rest)):
        ch = rest[idx]
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                close_at = idx
                break
    if close_at is None:
        raise RuntimeError("unbalanced xkb_symbols section")
    parts.append(rest[:close_at])
    for keysym, kc in self._overlay.items():
        parts.append(
            f"\tkey <UC{kc:03}> {{ [ {overlay_bind_keysym(keysym):#x} ] }};\n")
    parts.append(rest[close_at:])
    return "".join(parts)

def _overlay_bind_many(self, keysyms: Iterable[int]) -> dict:
    """Assign overlay keycodes to a batch of keysyms in ONE keymap swap.

    A swap costs milliseconds on the compositor thread (which also drives
    input and rendering), so binding a burst one at a time would stall it
    proportionally. A full pool recycles the oldest slot not held down:
    rebinding a pressed keycode would make its release report a different
    symbol than its press did. The swap rides the same command channel as
    the key events and never awaits a reply, so it drains before the keys
    that need it while this loop is never blocked on the compositor;
    `set_keymap_overlay` hands over just the binds, the `set_keymap_string`
    fallback re-sends the whole keymap text (a redundant compile far side).

    Returns:
        `{keysym: keycode}` for every requested keysym; a keysym that could
        not be bound (every slot held down) maps to 0.
    """
    held = {kc for kc, _ in self._pressed.values()}
    out = {}
    fresh = False
    for keysym in dict.fromkeys(keysyms):
        kc = self._overlay.get(keysym)
        if kc is None:
            if len(self._overlay) >= len(self._overlay_codes):
                victim = next(
                    (s for s in self._overlay_order if self._overlay[s] not in held),
                    None,
                )
                if victim is None:
                    out[keysym] = 0
                    continue
                self._overlay_order.remove(victim)
                kc = self._overlay.pop(victim)
            else:
                kc = self._overlay_codes[len(self._overlay)]
            self._overlay[keysym] = kc
            self._overlay_order.append(keysym)
            held.add(kc)
            fresh = True
        out[keysym] = kc
    if fresh:
        binds = [(kc, overlay_bind_keysym(sym))
                 for sym, kc in self._overlay.items()]
        splice = getattr(self._input, "set_keymap_overlay", None)
        if splice is not None:
            splice(binds)
        else:
            self._input.set_keymap_string(self._overlay_text())
    return out

def _overlay_bind(self, keysym: int) -> int:
    return self._overlay_bind_many([keysym])[keysym]

def _tap(self, kc: int, mods: Iterable[int],
         into: Optional[list] = None) -> None:
    """Momentary press+release with refcounted modifier synthesis; a
    modifier the client itself holds down is left alone.

    With `into`, the events are appended to that list instead of injected, so a
    caller typing a run of characters can hand the compositor one ordered batch —
    an event at a time costs a channel send and a calloop wake each, on the thread
    that also renders."""
    out = [] if into is None else into
    synthed = []
    for m in mods:
        if m in self._down and m not in self._mod_refs:
            continue
        self._mod_refs[m] = self._mod_refs.get(m, 0) + 1
        if self._mod_refs[m] == 1:
            out.append((m, 1))
        synthed.append(m)
    out.append((kc, 1))
    out.append((kc, 0))
    for m in reversed(synthed):
        refs = self._mod_refs.get(m, 0) - 1
        if refs <= 0:
            self._mod_refs.pop(m, None)
            out.append((m, 0))
        else:
            self._mod_refs[m] = refs
    if into is None:
        self._inject_run(out)

def _inject_run(self, events: list) -> None:
    """Deliver an ordered run of (keycode, state) events, batched when the
    compositor accepts a batch."""
    if not events:
        return
    for kc, state in events:
        (self._down.add if state else self._down.discard)(kc)
    batch = getattr(self._input, "inject_keys", None)
    if batch is not None:
        batch(events)
        return
    for kc, state in events:
        self._input.inject_key(kc, state)

def type_text(self, text: str, neutralize: bool = False) -> bool:
    """Type text as momentary taps with at most ONE keymap swap.

    Every missing keysym resolves in a single swap (no per-char swap storm).
    Each char prefers its canonical layout keysym (a ru layout types ф on
    its own key) before falling to the overlay.

    Args:
        text: Characters to tap out in order.
        neutralize: Lift conflicting held Shift/AltGr around the whole run
            so the taps land on their resolved levels.

    Returns:
        False, having typed nothing, when a char cannot be bound at all;
        True once the full run is injected.
    """
    keysyms = []
    for ch in text:
        ks = character_to_layout_keysym(ch)
        if ks not in self._map:
            cp = ord(ch)
            ks = cp if 0x20 <= cp <= 0xFF else (0x01000000 | cp)
        keysyms.append(ks)
    missing = [ks for ks in dict.fromkeys(keysyms) if ks not in self._map]
    overlay = self._overlay_bind_many(missing) if missing else {}
    # Resolve everything before touching state: the False path must have
    # typed nothing and charged no modifier refs.
    resolved_keys = []
    for ks in keysyms:
        resolved = self._map.get(ks)
        if resolved is None and overlay.get(ks):
            resolved = (overlay[ks], 0)
        if resolved is None:
            return False
        resolved_keys.append(resolved)
    # Lift conflicts before building the taps so _down reflects the lift and
    # a shifted char inside the run synthesizes its Shift normally.
    lifted = self._held_conflicts(()) if neutralize else []
    for kc in lifted:
        self._inject(kc, 0)
    events = []
    for kc, level in resolved_keys:
        self._tap(kc, self._mods_for_level(level), into=events)
    self._inject_run(events + [(kc, 1) for kc in reversed(lifted)])
    return True

def press(self, keysym: int, neutralize: bool = False) -> None:
    """Press a keysym, overlay-binding it first when the base layout lacks it.

    Only modifiers not already down are synthesized; a modifier the client
    holds as its own key is neither charged nor released here.

    Args:
        neutralize: Lift a conflicting held Shift/AltGr around the press and
            restore it after: a client layout's Shift pairing rarely matches
            the seat layout's, so the held modifier would move the key onto
            a different glyph. Chords pass False so Ctrl+Shift+X passes
            through untouched.
    """
    held = self._pressed.get(keysym)
    if held is not None:
        # Auto-repeat re-press: the first press already charged the refcounts.
        self._inject(held[0], 1)
        return
    resolved = self._map.get(keysym)
    if resolved is not None:
        kc, level = resolved
        mods = self._mods_for_level(level)
    else:
        kc = self._overlay_bind(keysym)
        if not kc:
            # Pool exhausted (every slot held down); keycode 0 is not a key.
            return
        mods = ()
    lifted = self._held_conflicts(set(mods)) if neutralize else []
    for m in lifted:
        self._inject(m, 0)
    synthed = []
    for m in mods:
        already = (m in self._down
                   or (m == self._shift_kc and self._shift_r_kc in self._down))
        if already and m not in self._mod_refs:
            continue
        self._mod_refs[m] = self._mod_refs.get(m, 0) + 1
        if self._mod_refs[m] == 1:
            self._inject(m, 1)
        synthed.append(m)
    self._pressed[keysym] = (kc, tuple(synthed))
    self._inject(kc, 1)
    for m in reversed(lifted):
        self._inject(m, 1)

def release(self, keysym: int) -> None:
    """Release a pressed keysym and un-refcount the modifiers its press synthesized."""
    held = self._pressed.pop(keysym, None)
    if held is None:
        return
    kc, mods = held
    self._inject(kc, 0)
    for m in reversed(mods):
        refs = self._mod_refs.get(m, 0) - 1
        if refs <= 0:
            self._mod_refs.pop(m, None)
            self._inject(m, 0)
        else:
            self._mod_refs[m] = refs

def reset(self) -> None:
    """Release every held key and synthetic modifier."""
    for keysym in list(self._pressed):
        try:
            self.release(keysym)
        except Exception:
            self._pressed.pop(keysym, None)
    for m in list(self._mod_refs):
        try:
            self._inject(m, 0)
        except Exception:
            pass
    self._mod_refs.clear()
    self._down.clear()

def adopt_held(self, previous: "_WaylandKeymapOwner") -> None:
    """Carry what a previous owner left down across a base-layout change.

    Held keys and synthesized modifiers are physical keycodes the compositor
    still has pressed, valid under any base keymap, so the releases that
    follow the change must reach them: a fresh owner would drop them as
    no-ops and leave the keys held for good. Overlay binds are not carried —
    the compositor rebuilds its keymap from the new base without them.
    """
    self._pressed.update(previous._pressed)
    self._mod_refs.update(previous._mod_refs)
    self._down.update(previous._down)