SelkiesGamepad

def __init__(self, js_interposer_socket_path: str,
             evdev_interposer_socket_path: str,
             loop: Optional[asyncio.AbstractEventLoop] = None,
             uinput_enabled: bool = False) -> None:
    self.js_sock_path = js_interposer_socket_path
    self.evdev_sock_path = evdev_interposer_socket_path
    self.loop = loop or asyncio.get_running_loop()

    self.uinput_enabled = uinput_enabled
    self.uinput = None

    self.mapper = None
    self.config_payload_cache = None

    self.js_server = None
    self.evdev_server = None
    self.js_clients = {}
    self.evdev_clients = {}

    self.events_queue = asyncio.Queue(maxsize=4096)
    self.running = False
    self._event_processor_task = None

    self._held_controls = set()
    self._js_state = {}
    self._evdev_state = {}

    self.on_rumble: Optional[Callable[[float, float, int], None]] = None
    self._ff_playing: dict = {}
    self._ff_gain = 0xffff
    self._ff_timer: Optional[asyncio.TimerHandle] = None
    self._ff_sent: Optional[tuple] = (0, 0)
    self._ff_renew_at: Optional[float] = None

def set_config(self, client_input_name: str, client_num_btns: int,
               client_num_axes: int) -> None:
    """Build the mapper and cache the js_config_t payload served to interposer clients."""
    self.mapper = GamepadMapper(STANDARD_XPAD_CONFIG, client_input_name, client_num_btns, client_num_axes)

    js_idx = 0 
    match = re.search(r"selkies_js(\d+)\.sock$", self.js_sock_path)
    if match:
        js_idx = int(match.group(1))
    else:
        logger_selkies_gamepad.warning(
            f"Failed to parse js_index from {self.js_sock_path}, "
            f"defaulting to 0 for payload name generation if needed."
        )

    payload_controller_config = {
        "name": STANDARD_XPAD_CONFIG.get("name", f"Selkies Virtual JS{js_idx}"),
        "vendor_id": STANDARD_XPAD_CONFIG.get("vendor_id", 0x0000),
        "product_id": STANDARD_XPAD_CONFIG.get("product_id", 0x0000),
        "version": STANDARD_XPAD_CONFIG.get("version", 0x0114),
        "buttons": STANDARD_XPAD_CONFIG.get("btn_map", []), 
        "axes": STANDARD_XPAD_CONFIG.get("axes_map", [])
    }

    self.config_payload_cache = self._make_interposer_config_payload(js_idx, payload_controller_config)

    logger_selkies_gamepad.debug(
        f"Gamepad configured. JS socket: {self.js_sock_path}, EVDEV socket: {self.evdev_sock_path}. "
        f"Using fixed config: {STANDARD_XPAD_CONFIG['name']}"
    )

def ensure_uinput(self) -> None:
    """Bring this slot's kernel device up, once, if kernel gamepads are on.
    A failure downgrades the slot to the interposer sockets rather than
    killing input."""
    if not self.uinput_enabled or self.uinput is not None:
        return
    match = re.search(r"selkies_js(\d+)\.sock$", self.js_sock_path)
    device = UInputGamepad(os.path.basename(self.js_sock_path),
                           pad_phys(int(match.group(1)) if match else 0))
    try:
        nodes = device.create()
    except OSError as e:
        self.uinput_enabled = False
        logger_selkies_gamepad.error(
            f"Gamepad {self.js_sock_path}: could not create a kernel device ({e}); "
            "this slot now reaches applications only through the Input Interposer."
        )
        return
    self.uinput = device
    try:
        self.loop.add_reader(device.fd, self._service_uinput_ff)
    except (OSError, ValueError) as e:
        # Only an emulated /dev/uinput (tests/tools/uinput_shim.c) is a
        # file the loop cannot watch; a kernel device always can be.
        logger_selkies_gamepad.warning(
            f"Gamepad {self.js_sock_path}: force feedback requests cannot be watched ({e})."
        )
    logger_selkies_gamepad.info(
        f"Gamepad {self.js_sock_path}: kernel device ready ({', '.join(nodes) or 'node path unknown'})."
    )
    unreadable = [node for node in nodes if not os.access(node, os.R_OK)]
    if unreadable:
        logger_selkies_gamepad.warning(
            f"Gamepad {self.js_sock_path}: {', '.join(unreadable)} is not readable by this user, "
            "so applications cannot open it. Add the account to the 'input' group."
        )

def _emit_uinput(self, ev_type: int, ev_code: int, ev_value: float) -> None:
    """Mirror one event onto the kernel device, tearing it down on write failure."""
    self.ensure_uinput()
    if self.uinput is None:
        return
    try:
        self.uinput.emit(ev_type, ev_code, ev_value)
    except OSError as e:
        logger_selkies_gamepad.error(
            f"Gamepad {self.js_sock_path}: kernel device write failed ({e}); tearing it down."
        )
        self._drop_uinput()
        self.uinput_enabled = False

def _drop_uinput(self) -> None:
    """Destroy the kernel device, and forget the effects applications played on it."""
    if self.uinput is None:
        return
    if self.uinput.fd is not None:
        self.loop.remove_reader(self.uinput.fd)
    self.uinput.destroy()
    self.uinput = None
    self.ff_forget("uinput")

def _service_uinput_ff(self) -> None:
    """Answer the kernel device's queued effect requests and mix what they played."""
    if self.uinput is None:
        return
    try:
        records = self.uinput.service_ff()
    except OSError as e:
        logger_selkies_gamepad.warning(f"Gamepad {self.js_sock_path}: force feedback request failed: {e}")
        return
    for record in records:
        self.ff_record("uinput", record)

def ff_record(self, source: Any, record: tuple) -> None:
    """Apply one FF_RECORD_FMT record an application's handle produced.

    A play starts after its delay and lasts its length times its repeat
    count, or until stopped when its length is 0; a play of an effect
    already playing replaces it, as a replay does on a kernel pad.
    """
    kind, _, effect_id, strong, weak, length, delay, count, _ = record
    key = (source, effect_id)
    if kind == FF_RECORD_PLAY and count:
        start = self.loop.time() + delay / 1000.0
        end = start + count * length / 1000.0 if length else None
        self._ff_playing[key] = [strong, weak, start, end]
    elif kind in (FF_RECORD_PLAY, FF_RECORD_STOP):
        self._ff_playing.pop(key, None)
    elif kind == FF_RECORD_GAIN:
        self._ff_gain = count
    self._ff_update()

def ff_replay(self) -> None:
    """Hand the mix playing now to `on_rumble` again, for a client that has
    just taken the slot: an unchanged mix is otherwise not sent until its
    renewal."""
    if any(self._ff_sent):
        self._ff_sent = None
        self._ff_update()

def ff_forget(self, source: Any) -> None:
    """Stop every effect one source played: its handle closed, or its device went."""
    for key in [k for k in self._ff_playing if k[0] == source]:
        del self._ff_playing[key]
    self._ff_update()

def _ff_update(self) -> None:
    """Mix the effects playing now and hand the result to `on_rumble`.

    The motors take the sum of every playing effect's magnitudes, scaled
    by the gain and capped, as a memless pad mixes them. The client is told
    how long to hold the mix: until the next effect starts or ends, or,
    where that is further off than FF_LEASE_MS or never (an effect without
    an end), for FF_LEASE_MS at a time, renewed at half that; the Gamepad
    API plays at most 5 s at once, and a client that stops hearing stops
    shaking. A mix that did not change is not sent again until a renewal
    is due, and the timer wakes at the next start, end, or renewal.
    """
    if self._ff_timer is not None:
        self._ff_timer.cancel()
        self._ff_timer = None
    now = self.loop.time()
    for key in [k for k, v in self._ff_playing.items() if v[3] is not None and v[3] <= now]:
        del self._ff_playing[key]
    strong = weak = 0
    boundaries = []
    endless = False
    for s_mag, w_mag, start, end in self._ff_playing.values():
        if start > now:
            boundaries.append(start)
            continue
        strong += s_mag
        weak += w_mag
        if end is None:
            endless = True
        else:
            boundaries.append(end)
    strong = min(0xffff, strong * self._ff_gain // 0xffff)
    weak = min(0xffff, weak * self._ff_gain // 0xffff)
    mix = (strong, weak)
    lease = FF_LEASE_MS / 1000.0
    change = min((t for t in boundaries if t > now), default=None)
    leased = any(mix) and (endless or change is None or change - now > lease)
    hold = lease if leased else (change - now if any(mix) and change is not None else 0.0)
    renew = self._ff_renew_at is not None and now >= self._ff_renew_at
    if mix != self._ff_sent or renew:
        self._ff_sent = mix
        self._ff_renew_at = now + lease / 2 if leased else None
        if self.on_rumble is not None:
            try:
                self.on_rumble(strong / 0xffff, weak / 0xffff, int(round(hold * 1000)))
            except Exception:
                logger_selkies_gamepad.debug("rumble relay failed", exc_info=True)
    wake = [t for t in boundaries if t > now]
    if self._ff_renew_at is not None:
        wake.append(self._ff_renew_at)
    if wake:
        self._ff_timer = self.loop.call_at(min(wake), self._ff_update)

def _make_interposer_config_payload(self, js_index: int, controller_config: dict) -> bytes:
    """Create the js_config_t payload sent to the C interposer.

    The payload is always exactly C_INTERPOSER_STRUCT_SIZE bytes; every
    failure path returns a zeroed buffer of that size rather than raising,
    so a client handshake never dies on a malformed config.
    """
    struct_fmt = base_struct_fmt = "undefined"
    try:
        name_str = controller_config.get("name", f"Selkies Virtual JS{js_index}")
        name_bytes_utf8 = name_str.encode('utf-8')
        if len(name_bytes_utf8) >= CONTROLLER_NAME_MAX_LEN:
            name_bytes_for_pack = name_bytes_utf8[:CONTROLLER_NAME_MAX_LEN - 1] + b'\0'
        else:
            name_bytes_for_pack = name_bytes_utf8.ljust(CONTROLLER_NAME_MAX_LEN, b'\0')

        if len(name_bytes_for_pack) != CONTROLLER_NAME_MAX_LEN:
             logger_selkies_gamepad.error(f"CRITICAL: name_bytes_for_pack is not {CONTROLLER_NAME_MAX_LEN} bytes long! Got {len(name_bytes_for_pack)}")
             return b'\0' * C_INTERPOSER_STRUCT_SIZE

        raw_vendor = controller_config.get("vendor_id")
        if isinstance(raw_vendor, str):
            vendor_id = int(raw_vendor, 16)
        elif isinstance(raw_vendor, int):
            vendor_id = raw_vendor
        else:
            vendor_id = 0x045e
        raw_product = controller_config.get("product_id")
        if isinstance(raw_product, str):
            product_id = int(raw_product, 16)
        elif isinstance(raw_product, int):
            product_id = raw_product
        else:
            product_id = 0x028e
        raw_version = controller_config.get("version")
        if isinstance(raw_version, str):
            version_id = int(raw_version, 16)
        elif isinstance(raw_version, int):
            version_id = raw_version
        else:
            version_id = 0x0114

        buttons_evdev_codes = controller_config.get("buttons", [])
        axes_evdev_codes = controller_config.get("axes", [])

        # Counts clamped to the array capacity: a count above the truncated
        # map length would drive an out-of-bounds read in the C interposer.
        num_actual_btns = min(len(buttons_evdev_codes), INTERPOSER_MAX_BTNS)
        num_actual_axes = min(len(axes_evdev_codes), INTERPOSER_MAX_AXES)

        padded_btn_map_for_pack = list(buttons_evdev_codes)
        if len(padded_btn_map_for_pack) > INTERPOSER_MAX_BTNS:
            logger_selkies_gamepad.warning(f"Controller '{name_str}' has {len(padded_btn_map_for_pack)} buttons, truncating to {INTERPOSER_MAX_BTNS} for config.")
            padded_btn_map_for_pack = padded_btn_map_for_pack[:INTERPOSER_MAX_BTNS]
        else:
            padded_btn_map_for_pack.extend([0] * (INTERPOSER_MAX_BTNS - len(padded_btn_map_for_pack)))

        padded_axes_map_for_pack = list(axes_evdev_codes)
        if len(padded_axes_map_for_pack) > INTERPOSER_MAX_AXES:
            logger_selkies_gamepad.warning(f"Controller '{name_str}' has {len(padded_axes_map_for_pack)} axes, truncating to {INTERPOSER_MAX_AXES} for config.")
            padded_axes_map_for_pack = padded_axes_map_for_pack[:INTERPOSER_MAX_AXES]
        else:
            padded_axes_map_for_pack.extend([0] * (INTERPOSER_MAX_AXES - len(padded_axes_map_for_pack)))

        base_struct_fmt = f"={CONTROLLER_NAME_MAX_LEN}sxHHHHH{INTERPOSER_MAX_BTNS}H{INTERPOSER_MAX_AXES}B"

        size_without_explicit_end_padding = struct.calcsize(base_struct_fmt)

        padding_needed = C_INTERPOSER_STRUCT_SIZE - size_without_explicit_end_padding

        if padding_needed < 0:
            logger_selkies_gamepad.error(
                f"CRITICAL STRUCT SIZE ERROR: Python base packed size ({size_without_explicit_end_padding}) "
                f"is larger than C interposer expected size ({C_INTERPOSER_STRUCT_SIZE}). "
                f"This means constants (MAX_BTNS, MAX_AXES, NAME_LEN) or field types/order "
                f"differ between Python 'base_struct_fmt' and C 'js_config_t'."
            )
            return b'\0' * C_INTERPOSER_STRUCT_SIZE

        struct_fmt = f"{base_struct_fmt}{padding_needed}x"

        python_final_packed_size = struct.calcsize(struct_fmt)
        if python_final_packed_size != C_INTERPOSER_STRUCT_SIZE:
            logger_selkies_gamepad.error(
                f"CRITICAL FINAL PYTHON PACKED SIZE MISMATCH for js_config_t! "
                f"C interposer expects: {C_INTERPOSER_STRUCT_SIZE}, "
                f"Python struct.pack calculated final size: {python_final_packed_size} using format '{struct_fmt}'. "
                f"This indicates an issue with padding calculation logic or the base_struct_fmt."
            )
            return b'\0' * C_INTERPOSER_STRUCT_SIZE

        logger_selkies_gamepad.debug(f"Using final struct_fmt: '{struct_fmt}' for js_config, packing to size {python_final_packed_size}")

        payload_args = [
            name_bytes_for_pack,
            vendor_id,
            product_id,
            version_id,
            num_actual_btns,
            num_actual_axes,
        ]
        payload_args.extend(padded_btn_map_for_pack)
        payload_args.extend(padded_axes_map_for_pack)

        payload = struct.pack(struct_fmt, *payload_args)

        log_display_name = name_bytes_for_pack.split(b'\0',1)[0].decode('utf-8', errors='replace')
        logger_selkies_gamepad.debug(f"Packed js_config payload for '{name_str}' (js{js_index}): "
                     f"len={len(payload)} bytes. "
                     f"Name='{log_display_name}', "
                     f"Vendor=0x{vendor_id:04x}, Product=0x{product_id:04x}, Version=0x{version_id:04x}, "
                     f"Reported Buttons={num_actual_btns} (Array capacity: {INTERPOSER_MAX_BTNS}), "
                     f"Reported Axes={num_actual_axes} (Array capacity: {INTERPOSER_MAX_AXES})")

        if len(payload) != C_INTERPOSER_STRUCT_SIZE:
            logger_selkies_gamepad.error(f"FINAL PAYLOAD SIZE MISMATCH AFTER PACKING! Expected {C_INTERPOSER_STRUCT_SIZE}, got {len(payload)}. This is very bad.")
            return b'\0' * C_INTERPOSER_STRUCT_SIZE
        return payload

    except struct.error as e:
        current_struct_fmt = struct_fmt if struct_fmt != "undefined" else base_struct_fmt
        logger_selkies_gamepad.error(f"Error packing joystick config for js{js_index} with format '{current_struct_fmt}': {e}")
        config_to_log = controller_config if 'controller_config' in locals() else {}
        logger_selkies_gamepad.error(f"Controller config was: {config_to_log}")
        return b'\0' * C_INTERPOSER_STRUCT_SIZE
    except Exception as e:
        config_to_log = controller_config if 'controller_config' in locals() else {}
        logging.exception(f"Unexpected error creating interposer config payload for js{js_index} with config {config_to_log}: {e}")
        return b'\0' * C_INTERPOSER_STRUCT_SIZE

async def _handle_interposer_client(self, reader: asyncio.StreamReader,
                                    writer: asyncio.StreamWriter,
                                    is_evdev_socket: bool) -> None:
    """Per-client handshake and lifetime: send config, read the client's
    architecture byte, register the writer for event fan-out, then hold the
    connection open until shutdown or disconnect.

    A JS client first gets its current state replayed as INIT events
    (joydev semantics); the snapshot, its write, and the registration share
    one loop step, so no broadcast can interleave and the client's first
    live event strictly follows its snapshot. evdev has no in-band INIT:
    those clients poll state through the interposer's ioctl emulation.
    """
    peername = writer.get_extra_info('peername')
    socket_type_str = "EVDEV" if is_evdev_socket else "JS"
    clients_dict = self.evdev_clients if is_evdev_socket else self.js_clients
    sock_path = self.evdev_sock_path if is_evdev_socket else self.js_sock_path
    log_prefix = f"Gamepad {sock_path} Client {peername} ({socket_type_str}):"
    logger_selkies_gamepad.debug(f"{log_prefix} Handler started.")
    sock = writer.get_extra_info('socket')
    if sock is not None:
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, INTERPOSER_SNDBUF)
        except OSError as e:
            logger_selkies_gamepad.debug(f"{log_prefix} Send buffer left at the host's default: {e}")

    try:
        if not self.config_payload_cache:
            logger_selkies_gamepad.error(f"{log_prefix} Config payload not ready. Aborting handler.")
            return
        logger_selkies_gamepad.debug(f"{log_prefix} Preparing to send config payload. Length: {len(self.config_payload_cache)}, Expected C size: {EXPECTED_C_STRUCT_SIZE}, First 16 bytes: {self.config_payload_cache[:16].hex()}")
        writer.write(self.config_payload_cache)
        await writer.drain()
        logger_selkies_gamepad.debug(f"{log_prefix} Sent config payload.")

        arch_byte = await reader.readexactly(1)
        client_sizeof_long = struct.unpack("=B", arch_byte)[0]
        client_arch_bits = client_sizeof_long * 8
        logger_selkies_gamepad.debug(f"{log_prefix} Received arch specifier: {client_sizeof_long} bytes ({client_arch_bits}-bit).")

        if not is_evdev_socket:
            writer.write(self.init_state_burst())
        clients_dict[writer] = {'arch_bits': client_arch_bits}
        await writer.drain()
        logger_selkies_gamepad.debug(f"{log_prefix} Added to active list. Total {socket_type_str} clients: {len(clients_dict)}.")

        # An evdev handle writes its force feedback back as FF_RECORD_FMT
        # records; a joydev one has none, and what it writes is dropped.
        # Either way the read is what notices the application closing it.
        while self.running and not writer.is_closing():
            if is_evdev_socket:
                record = await reader.readexactly(FF_RECORD_SIZE)
                self.ff_record(writer, struct.unpack(FF_RECORD_FMT, record))
            elif not await reader.read(4096):
                break

        if not self.running:
            logger_selkies_gamepad.debug(f"{log_prefix} Exiting handler normally because self.running is False.")
        if writer.is_closing():
            logger_selkies_gamepad.debug(f"{log_prefix} Exiting handler normally because writer.is_closing() is True (client likely closed connection).")

    except (asyncio.IncompleteReadError, ConnectionResetError, BrokenPipeError) as e:
        logger_selkies_gamepad.debug(f"{log_prefix} Disconnected (expected error): {type(e).__name__} - {e}")
    except Exception as e:
        logger_selkies_gamepad.error(f"{log_prefix} Unhandled error in handler: {e}", exc_info=True)
    finally:
        logger_selkies_gamepad.debug(f"{log_prefix} Entering finally block.")
        self.ff_forget(writer)
        if writer in clients_dict:
            del clients_dict[writer]
            logger_selkies_gamepad.debug(f"{log_prefix} Removed from active list. Total {socket_type_str} clients now: {len(clients_dict)}.")
        elif self.running:
            logger_selkies_gamepad.warning(f"{log_prefix} Writer not found in active list during finally block.")

        if not writer.is_closing():
            logger_selkies_gamepad.debug(f"{log_prefix} Explicitly closing writer in finally block.")
            writer.close()
            await writer.wait_closed()
        logger_selkies_gamepad.debug(f"{log_prefix} Handler finished.")

async def _run_single_server(self, interposer_socket_path: str,
                             is_evdev_socket: bool) -> Optional[asyncio.AbstractServer]:
    """Bind one interposer Unix server (unlinking a stale socket file first);
    None on failure.

    A file at the path that another account owns is refused: in a shared
    directory it is that account's listener, and every application of this
    session would open it as its gamepad.
    """
    sock_dir = os.path.dirname(interposer_socket_path)
    if sock_dir and not os.path.exists(sock_dir):
        try: os.makedirs(sock_dir, exist_ok=True)
        except OSError as e:
            logger_selkies_gamepad.error(f"Failed to create directory {sock_dir} for socket: {e}")
            return None

    try:
        owner = os.lstat(interposer_socket_path).st_uid
    except FileNotFoundError:
        owner = None
    if owner is not None and owner != os.geteuid():
        logger_selkies_gamepad.error(
            f"{interposer_socket_path} belongs to uid {owner}, not this session: gamepads are not "
            f"served there; point js_socket_path at a directory of this user's own.")
        return None
    if owner is not None:
        try:
            os.unlink(interposer_socket_path)
            logger_selkies_gamepad.debug(f"Removed existing socket file: {interposer_socket_path}")
        except OSError as e:
            logger_selkies_gamepad.error(f"Could not remove the stale socket at {interposer_socket_path}: {e}")
            return None

    try:
        server = await asyncio.start_unix_server(
            lambda r, w: self._handle_interposer_client(r, w, is_evdev_socket),
            path=interposer_socket_path
        )
        addr = server.sockets[0].getsockname() if server.sockets else interposer_socket_path
        logger_selkies_gamepad.debug(f"{'EVDEV' if is_evdev_socket else 'JS'} interposer server listening on {addr}")
        return server
    except Exception as e:
        logger_selkies_gamepad.error(f"Failed to start {'EVDEV' if is_evdev_socket else 'JS'} server on {interposer_socket_path}: {e}", exc_info=True)
        return None

async def run_servers(self) -> None:
    """Start both interposer servers and the event processor; runs until close()."""
    if not self.mapper:
        logger_selkies_gamepad.error("Mapper not set. Call set_config() before run_servers().")
        return

    self.running = True
    if self._event_processor_task is None or self._event_processor_task.done():
        self._event_processor_task = asyncio.create_task(self._process_event_queue())

    self.js_server = await self._run_single_server(self.js_sock_path, is_evdev_socket=False)
    self.evdev_server = await self._run_single_server(self.evdev_sock_path, is_evdev_socket=True)

    if not self.js_server and not self.evdev_server:
        logger_selkies_gamepad.error("Neither JS nor EVDEV interposer server could be started. Stopping.")
        self.running = False
        if self._event_processor_task and not self._event_processor_task.done():
            self._event_processor_task.cancel()
        return

    while self.running:
        await asyncio.sleep(1)
    logger_selkies_gamepad.debug("run_servers loop exited.")

def send_event(self, client_event_idx: int, client_value: float,
               is_button_event: bool) -> None:
    """Map one client control change and queue it for fan-out to every client.

    On overflow the oldest queued event is dropped to make room — for a
    gamepad the freshest state matters and a stale sample is worthless —
    so a slowly draining client cannot back-pressure into unbounded
    growth; the shutdown sentinel (None) is re-enqueued if evicted.
    """
    if not self.mapper or not self.running:
        return
    event_package = self.mapper.get_mapped_events(client_event_idx, client_value, is_button_event)
    if event_package:
        control = (is_button_event, client_event_idx)
        if client_value:
            self._held_controls.add(control)
        else:
            self._held_controls.discard(control)
        js_data = event_package.get('js_event_data')
        if js_data:
            _, value, ev_type, number = struct.unpack("=IhBB", js_data)
            self._js_state[(ev_type, number)] = value
        evdev_template = event_package.get('evdev_event_template')
        if evdev_template:
            self._evdev_state[evdev_template[:2]] = evdev_template[2]
        logger_selkies_gamepad.debug(f"Gamepad {self.js_sock_path}: Queuing event: {event_package}")
        try:
            self.events_queue.put_nowait(event_package)
        except asyncio.QueueFull:
            try:
                dropped = self.events_queue.get_nowait()
                self.events_queue.task_done()
                if dropped is None:
                    self.events_queue.put_nowait(None)
                    return
            except asyncio.QueueEmpty:
                pass
            try:
                self.events_queue.put_nowait(event_package)
            except asyncio.QueueFull:
                logger_selkies_gamepad.warning(
                    f"Gamepad {self.js_sock_path}: event queue full; dropping event."
                )

def reset_state(self) -> None:
    """Emit a neutral value for every control still held non-neutral, so a
    dropped association leaves no stuck button or off-center axis behind on
    the app driving this pad."""
    for is_button_event, client_event_idx in list(self._held_controls):
        self.send_event(client_event_idx, 0, is_button_event)

def init_state_burst(self) -> bytes:
    """joydev-parity state replay for a newly connected JS client: every
    control's current value as JS_EVENT_INIT-flagged events, so an app
    opening the pad mid-hold starts from the true state (and input during
    the connect handshake is covered as state, not lost edges). Buttons
    rest at 0, stick/hat axes at center, triggers at the mapper's released
    value."""
    mapping = STANDARD_XPAD_CONFIG["mapping"]
    parts = []
    for idx in range(len(STANDARD_XPAD_CONFIG["btn_map"])):
        value = self._js_state.get((JS_EVENT_BUTTON, idx), 0)
        parts.append(get_js_event_packed(JS_EVENT_BUTTON | JS_EVENT_INIT, idx, value))
    for idx in range(len(STANDARD_XPAD_CONFIG["axes_map"])):
        rest = normalize_axis_value(
            0,
            idx in mapping["trigger_internal_abstract_axis_indices"],
            idx in mapping["hat_internal_abstract_axis_indices"],
            for_js_event=True,
        )
        value = self._js_state.get((JS_EVENT_AXIS, idx), rest)
        parts.append(get_js_event_packed(JS_EVENT_AXIS | JS_EVENT_INIT, idx, value))
    return b"".join(parts)

def evdev_state_burst(self, arch_bits: int) -> bytes:
    """Every control this pad has moved, at its latest value, packed for
    an evdev handle of `arch_bits`; a control never moved is still at rest."""
    return b"".join(get_evdev_events_packed(ev_type, ev_code, value, arch_bits)
                    for (ev_type, ev_code), value in self._evdev_state.items())

async def _process_event_queue(self) -> None:
    """Drain the event queue until the None sentinel, fanning each event out
    to JS, EVDEV, and uinput consumers.

    Nothing here waits on a client: each write is queued on its
    connection (`_fan_out`), so a handle the application has stopped
    reading never holds back delivery to the handles it reads.
    """
    logger_selkies_gamepad.debug(f"Gamepad {self.js_sock_path}: Event processor started.")
    while self.running:
        try:
            event_package = await self.events_queue.get()
            if event_package is None:
                self.events_queue.task_done()
                break

            logger_selkies_gamepad.debug(f"Gamepad {self.js_sock_path}: Dequeued event: {event_package}")

            js_data = event_package.get('js_event_data')
            evdev_template = event_package.get('evdev_event_template') 

            if js_data:
                for writer, client_info in list(self.js_clients.items()):
                    self._fan_out(writer, client_info, js_data, False)

            if evdev_template:
                ev_type, ev_code, ev_value = evdev_template
                self._emit_uinput(ev_type, ev_code, ev_value)
                for writer, client_info in list(self.evdev_clients.items()):
                    evdev_data = get_evdev_events_packed(ev_type, ev_code, ev_value,
                                                         client_info.get('arch_bits', 64))
                    self._fan_out(writer, client_info, evdev_data, True)

            self.events_queue.task_done()
        except asyncio.CancelledError:
            logger_selkies_gamepad.debug(f"Gamepad {self.js_sock_path}: Event processor task canceled.")
            break
        except Exception as e:
            logger_selkies_gamepad.error(f"Gamepad {self.js_sock_path}: Unhandled error in event processor: {e}", exc_info=True)
    logger_selkies_gamepad.debug(f"Gamepad {self.js_sock_path}: Event processor stopped.")

def _fan_out(self, writer: asyncio.StreamWriter, client_info: dict, data: bytes,
             is_evdev: bool) -> None:
    """Queue one event on an interposer connection, or shed it if the
    application has stopped reading that handle.

    Applications commonly hold a pad handle they never read (SDL, Wine,
    or a launcher probing the node beside the one the game reads), and
    the kernel treats such a reader as behind, never as gone: evdev drops
    its queue for SYN_DROPPED and joydev replays the pad's state, and the
    fd stays open. Here, once a connection's buffered writes reach the
    transport's high-water mark, the handle gets no more events until it
    reads the backlog down, and is then handed the pad's current state
    (`_resync_when_drained`), so it resumes where every other handle is.
    Closing it instead would leave the application a dead fd it may still
    mean to read; waiting on it would freeze every other handle of the pad.
    """
    if writer.is_closing() or 'resync' in client_info:
        return
    transport = writer.transport
    if transport.get_write_buffer_size() >= transport.get_write_buffer_limits()[1]:
        client_info['resync'] = self.loop.create_task(
            self._resync_when_drained(writer, client_info, is_evdev))
        return
    writer.write(data)

async def _resync_when_drained(self, writer: asyncio.StreamWriter, client_info: dict,
                               is_evdev: bool) -> None:
    """Wait for a shed handle to read its backlog down, then write the
    pad's current state to it and let events through again.

    A joydev handle gets the JS_EVENT_INIT replay it gets on open, as
    joydev itself gives a reader whose buffer overran; an evdev handle
    gets every control this pad has moved at its latest value, which a
    reader that skips SYN_DROPPED and one that resyncs alike take as
    ordinary events.
    """
    kind = "EVDEV" if is_evdev else "JS"
    sock_path = self.evdev_sock_path if is_evdev else self.js_sock_path
    logger_selkies_gamepad.debug(
        f"Gamepad {sock_path}: an {kind} handle is not being read; holding its events until it is.")
    try:
        await writer.drain()
        if writer.is_closing():
            return
        if is_evdev:
            writer.write(self.evdev_state_burst(client_info.get('arch_bits', 64)))
        else:
            writer.write(self.init_state_burst())
        logger_selkies_gamepad.debug(f"Gamepad {sock_path}: an {kind} handle caught up; resynced its state.")
    except (ConnectionResetError, BrokenPipeError):
        pass
    finally:
        client_info.pop('resync', None)

async def close(self) -> None:
    """Stop servers, drop clients, unlink socket files, destroy the kernel device."""
    logger_selkies_gamepad.debug(f"Closing gamepad services for JS:{self.js_sock_path}, EVDEV:{self.evdev_sock_path}")
    self.running = False

    # A client handler waits in a read of its connection, and a server's
    # wait_closed() waits for every connection to close, so the clients go
    # first.
    for writer in list(self.js_clients.keys()) + list(self.evdev_clients.keys()):
        if not writer.is_closing(): writer.close()

    if self.js_server:
        self.js_server.close()
        await self.js_server.wait_closed()
        self.js_server = None
        logger_selkies_gamepad.debug(f"JS interposer server {self.js_sock_path} closed.")
    if self.evdev_server:
        self.evdev_server.close()
        await self.evdev_server.wait_closed()
        self.evdev_server = None
        logger_selkies_gamepad.debug(f"EVDEV interposer server {self.evdev_sock_path} closed.")

    for writer in list(self.js_clients.keys()):
        if not writer.is_closing(): writer.close()
    self.js_clients.clear()
    for writer in list(self.evdev_clients.keys()):
        if not writer.is_closing(): writer.close()
    self.evdev_clients.clear()

    if self._event_processor_task and not self._event_processor_task.done():
        try:
            self.events_queue.put_nowait(None) 
            await asyncio.wait_for(self._event_processor_task, timeout=2.0)
        except asyncio.TimeoutError:
            logger_selkies_gamepad.warning("Event processor task timed out on close, canceling.")
            self._event_processor_task.cancel()
        except asyncio.CancelledError:
            pass 
        except Exception as e:
            logger_selkies_gamepad.error(f"Exception stopping event processor: {e}")
    self._event_processor_task = None

    for sock_path in [self.js_sock_path, self.evdev_sock_path]:
        if sock_path and os.path.exists(sock_path):
            try:
                os.unlink(sock_path)
                logger_selkies_gamepad.debug(f"Removed socket file: {sock_path}")
            except OSError as e:
                logger_selkies_gamepad.warning(f"Could not remove socket file {sock_path} on close: {e}")

    self._drop_uinput()

    logger_selkies_gamepad.debug("Gamepad services fully closed.")