_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)
            spares = unbound + shadowable
            text = [kc for kc in self._TEXT_SPARES if kc in spares]
            self._text_codes = len(text)
            self._overlay_codes = text + [kc for kc in spares if kc not in text] + 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. Once the text spares are bound, the oldest bind on one
    not held down is recycled before any other keycode is taken, and 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.
    """
    keysyms = list(dict.fromkeys(keysyms))
    # The batch's own binds are as untouchable as a held key: one swap binds
    # them all, so a recycled one would type the other keysym.
    held = {kc for kc, _ in self._pressed.values()}
    held.update(self._overlay[k] for k in keysyms if k in self._overlay)
    out = {}
    fresh = False
    for keysym in keysyms:
        kc = self._overlay.get(keysym)
        if kc is None:
            victim = None
            if len(self._overlay) >= self._text_codes:
                text = self._overlay_codes[:self._text_codes]
                victim = next((s for s in self._overlay_order
                               if self._overlay[s] in text and self._overlay[s] not in held), None)
                if victim is None and 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
            if victim is not None:
                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 one keymap swap per run.

    Every missing keysym of a run resolves in a single swap (no per-char
    swap storm). A run holds as many missing keysyms as there are text
    spares not held down, so a longer text goes out in runs that fit them,
    each swap riding the channel behind the taps before it. 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
        (every overlay keycode is held down, which only the first run can
        meet); True once the full text is injected.
    """
    keysyms = []
    for ch in text:
        ks = character_to_layout_keysym(ch)
        if ks not in self._map:
            ks = universal_text_keysym(ch)
            if ks is None:
                continue
        keysyms.append(ks)
    held = {kc for kc, _ in self._pressed.values()}
    room = sum(kc not in held for kc in self._overlay_codes[:self._text_codes])
    runs = [([], {})]
    for ks in keysyms:
        run, missing = runs[-1]
        if ks not in self._map and ks not in missing:
            if room and len(missing) >= room:
                run, missing = [], {}
                runs.append((run, missing))
            missing[ks] = None
        run.append(ks)
    lifted = []
    for i, (run, missing) in enumerate(runs):
        overlay = self._overlay_bind_many(missing) if missing else {}
        # Resolve the run before touching state: the False path must have
        # typed nothing and charged no modifier refs.
        resolved_keys = []
        for ks in run:
            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)
        if not i and neutralize:
            # 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(())
            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)
        if i == len(runs) - 1:
            events += [(kc, 1) for kc in reversed(lifted)]
        self._inject_run(events)
    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)