"""Map and style the real Wine game window on the host X11 display.""" from __future__ import annotations import argparse import ctypes import ctypes.util import os from pathlib import Path import select import subprocess import sys import time from collections.abc import Collection, Iterable, Mapping, Sequence from .desktop_identity import ( APPLICATION_NAME, APPLICATION_WM_CLASS, APPLICATION_WM_INSTANCE, ) from .window_diagnostics import WindowDiagnosticRecorder WINDOW_TITLE = APPLICATION_NAME WINDOW_DISCOVERY_TIMEOUT_SECONDS = 60.0 WINDOW_READY_DELAY_SECONDS = 0.5 WINDOW_SURFACE_SETTLE_SECONDS = 0.1 WINDOW_SURFACE_REFRESH_DELAY_SECONDS = 0.15 WINDOW_NUDGE_CONFIRM_TIMEOUT_SECONDS = 0.15 WINDOW_STATE_CONFIRM_TIMEOUT_SECONDS = 0.75 WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS = 0.5 WINDOW_ACTION_CONFIRM_TIMEOUT_SECONDS = 0.75 WINDOW_SCALING_EVENT_TIMEOUT_SECONDS = 0.5 WINDOW_SCALING_REPAIR_INTERVAL_SECONDS = 0.05 WINDOW_CONFIRM_POLL_SECONDS = 0.025 WINDOW_FOCUS_DELAY_SECONDS = 0.2 WINDOW_STARTUP_GUARD_SECONDS = 12.0 WINDOW_STARTUP_POLL_SECONDS = 0.1 WINDOW_ACTIVATION_DELAY_SECONDS = 0.2 WINDOW_STARTUP_RECOVERY_LIMIT = 3 WINDOW_SCALING_REPAIR_LIMIT = 3 WINDOW_SCALING_EVENT_BATCH_LIMIT = 64 _X_REVERT_TO_PARENT = 2 _X_CURRENT_TIME = 0 _X_CLIENT_MESSAGE = 33 _X_PROPERTY_NOTIFY = 28 _X_PROPERTY_CHANGE_MASK = 1 << 22 _X_SUBSTRUCTURE_NOTIFY_MASK = 1 << 19 _X_SUBSTRUCTURE_REDIRECT_MASK = 1 << 20 _X_WM_STATE_WITHDRAWN = 0 _X_WM_STATE_NORMAL = 1 _X_WM_STATE_ICONIC = 3 _NET_WM_STATE_REMOVE = 0 _NET_WM_STATE_SOURCE_APPLICATION = 1 _NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING = 2 WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS = 1.0 GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS = 0.25 GAMESCOPE_HOST_FOCUS_STATE_ENV = "PTLINUX_GAMESCOPE_HOST_FOCUS_STATE" GAMESCOPE_HOST_FOCUS_STATE_MAX_AGE_SECONDS = 0.6 _P_MIN_SIZE = 1 << 4 _P_MAX_SIZE = 1 << 5 _P_RESIZE_INC = 1 << 6 _P_BASE_SIZE = 1 << 8 _MOTIF_HINTS_FUNCTIONS = 1 << 0 _MOTIF_HINTS_DECORATIONS = 1 << 1 _MOTIF_ALL = 1 << 0 _MOTIF_FUNC_RESIZE = 1 << 1 _MOTIF_FUNC_MAXIMIZE = 1 << 4 _MOTIF_DECOR_RESIZE = 1 << 2 _MOTIF_DECOR_MAXIMIZE = 1 << 6 _X_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p) def _process_exit_status(returncode: int) -> int: """Convert Popen's negative signal code before passing it through SystemExit.""" if returncode < 0: return 128 + min(-returncode, 127) return min(returncode, 255) def _ignore_x11_error(_display: object, _event: object) -> int: """Keep a stale X11 window race from terminating the mapper process.""" return 0 def _matches_game_window_title(title: str, expected_title: str) -> bool: """Match the game title without accepting dialogs that merely mention it.""" candidate = title.strip().casefold() expected = expected_title.strip().casefold() if not candidate or not expected: return False return candidate == expected or candidate.startswith(f"{expected} ") def _select_game_window( windows: Iterable[tuple[int, str]], expected_title: str, excluded_window_ids: Collection[int], ) -> tuple[int, str] | None: for window, title in windows: if window in excluded_window_ids: continue if _matches_game_window_title(title, expected_title): return window, title return None def _protocols_without( protocols: Sequence[int], blocked_protocol: int, ) -> tuple[int, ...]: """Remove only the selected atom while preserving every other protocol.""" return tuple(protocol for protocol in protocols if protocol != blocked_protocol) def _motif_value_with_enabled_action(value: int, action: int) -> int: """Enable one Motif action while preserving its inclusive/exclusive mode.""" if value & _MOTIF_ALL: return value & ~action return value | action def _motif_value_allows_action(value: int, action: int) -> bool: """Interpret a Motif inclusive or all-except action mask.""" if value & _MOTIF_ALL: return value & action == 0 return value & action != 0 def _wm_state_name(value: int | None) -> str: return { _X_WM_STATE_WITHDRAWN: "withdrawn", _X_WM_STATE_NORMAL: "normal", _X_WM_STATE_ICONIC: "iconic", }.get(value, "unknown" if value is None else f"other_{value}") def _motif_hints_with_scaling(hints: Sequence[int]) -> tuple[int, ...]: """Add native resize/maximize actions while retaining every other hint.""" updated = list(hints) if len(updated) < 3: return tuple(updated) flags = updated[0] if flags & _MOTIF_HINTS_FUNCTIONS: for action in (_MOTIF_FUNC_RESIZE, _MOTIF_FUNC_MAXIMIZE): updated[1] = _motif_value_with_enabled_action(updated[1], action) if flags & _MOTIF_HINTS_DECORATIONS: for action in (_MOTIF_DECOR_RESIZE, _MOTIF_DECOR_MAXIMIZE): updated[2] = _motif_value_with_enabled_action(updated[2], action) return tuple(updated) def _normal_hints_without_fixed_size( hints: Sequence[int], current_size: tuple[int, int], ) -> tuple[int, ...]: """Relax Wine's exact fixed-size pair while retaining unrelated constraints.""" updated = list(hints) if len(updated) < 9: return tuple(updated) current_width, current_height = current_size has_minimum = updated[0] & _P_MIN_SIZE != 0 has_maximum = updated[0] & _P_MAX_SIZE != 0 has_fixed_pair = ( has_minimum and has_maximum and updated[5] == updated[7] and updated[6] == updated[8] ) maximum_is_current = ( has_maximum and updated[7] == current_width and updated[8] == current_height ) if has_fixed_pair: updated[0] &= ~(_P_MIN_SIZE | _P_MAX_SIZE) elif maximum_is_current: updated[0] &= ~_P_MAX_SIZE return tuple(updated) def _normal_hints_with_resize_increments( hints: Sequence[int], current_size: tuple[int, int], increments: tuple[int, int], ) -> tuple[int, ...]: """Add resize increments while preserving every unrelated size hint.""" updated = list(hints) if len(updated) < 18: updated.extend(0 for _ in range(18 - len(updated))) width, height = current_size width_increment, height_increment = increments updated[0] |= _P_RESIZE_INC | _P_BASE_SIZE updated[9] = width_increment updated[10] = height_increment updated[15] = width % width_increment updated[16] = height % height_increment return tuple(updated) class GameboxRuntimeError(RuntimeError): """Raised when the game command or X11 mapper cannot be started.""" class _MotifWmHints(ctypes.Structure): _fields_ = [ ("flags", ctypes.c_ulong), ("functions", ctypes.c_ulong), ("decorations", ctypes.c_ulong), ("input_mode", ctypes.c_long), ("status", ctypes.c_ulong), ] class _XClassHint(ctypes.Structure): _fields_ = [ ("res_name", ctypes.c_char_p), ("res_class", ctypes.c_char_p), ] class _XClientMessageData(ctypes.Union): _fields_ = [ ("bytes", ctypes.c_char * 20), ("shorts", ctypes.c_short * 10), ("longs", ctypes.c_long * 5), ] class _XClientMessageEvent(ctypes.Structure): _fields_ = [ ("type", ctypes.c_int), ("serial", ctypes.c_ulong), ("send_event", ctypes.c_int), ("display", ctypes.c_void_p), ("window", ctypes.c_ulong), ("message_type", ctypes.c_ulong), ("format", ctypes.c_int), ("data", _XClientMessageData), ] class _XPropertyEvent(ctypes.Structure): _fields_ = [ ("type", ctypes.c_int), ("serial", ctypes.c_ulong), ("send_event", ctypes.c_int), ("display", ctypes.c_void_p), ("window", ctypes.c_ulong), ("atom", ctypes.c_ulong), ("time", ctypes.c_ulong), ("state", ctypes.c_int), ] class _XEvent(ctypes.Union): _fields_ = [ ("type", ctypes.c_int), ("client_message", _XClientMessageEvent), ("property", _XPropertyEvent), ("padding", ctypes.c_long * 24), ] class _X11: def __init__(self) -> None: library_path = ctypes.util.find_library("X11") or "libX11.so.6" try: self.library = ctypes.CDLL(library_path) except OSError as exc: raise GameboxRuntimeError("the X11 client library is unavailable") from exc self._configure_signatures() self._error_handler = _X_ERROR_HANDLER(_ignore_x11_error) # type: ignore[arg-type] self._previous_error_handler = self.library.XSetErrorHandler( ctypes.cast(self._error_handler, ctypes.c_void_p), ) self.display = self.library.XOpenDisplay(None) if not self.display: self._restore_error_handler() raise GameboxRuntimeError("could not open the host X11 display") self._atoms: dict[bytes, int] = {} def close(self) -> None: try: if self.display: self.library.XCloseDisplay(self.display) self.display = None finally: self._restore_error_handler() def root_window_ids(self) -> set[int]: return set(self._root_windows()) def find_first_window( self, title_fragment: str, *, excluded_window_ids: Collection[int], ) -> tuple[int, str] | None: windows: list[tuple[int, str]] = [] for window in self._root_windows(): title = self._window_title(window) if title is not None: windows.append((window, title)) selected = _select_game_window(windows, title_fragment, excluded_window_ids) if selected is None: return None window, title = selected return self._client_window(window), title def find_window_by_pid(self, expected_pid: int) -> int | None: """Find a top-level host window created by one native process.""" queue = self._root_windows() visited: set[int] = set() while queue and len(visited) < 4096: window = queue.pop(0) if window in visited: continue visited.add(window) pid_values = self._long_property(window, b"_NET_WM_PID", 1) if pid_values and pid_values[0] == expected_pid: return window queue.extend(self._child_windows(window)) return None def set_resize_increments( self, window: int, current_size: tuple[int, int], increments: tuple[int, int], ) -> bool: """Quantize host resize requests without replacing other WM hints.""" if any(value <= 0 for value in (*current_size, *increments)): raise ValueError("window sizes and resize increments must be positive") hints = self._long_property(window, b"WM_NORMAL_HINTS", 18) or () updated = _normal_hints_with_resize_increments( hints, current_size, increments, ) self._replace_long_property( window, b"WM_NORMAL_HINTS", b"WM_SIZE_HINTS", updated, ) self.library.XSync(self.display, 0) confirmed = self._long_property(window, b"WM_NORMAL_HINTS", 18) return bool( confirmed and len(confirmed) >= 17 and confirmed[0] & _P_RESIZE_INC and confirmed[9:11] == increments ) def prefer_composited_presentation(self, window: int) -> bool: """Keep the host compositor active for this window at every size.""" self._replace_long_property( window, b"_NET_WM_BYPASS_COMPOSITOR", b"CARDINAL", (_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,), ) self.library.XSync(self.display, 0) return self._long_property( window, b"_NET_WM_BYPASS_COMPOSITOR", 1, ) == (_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,) def _root_windows(self) -> list[int]: root = self.library.XDefaultRootWindow(self.display) return self._child_windows(root) def _child_windows(self, window: int) -> list[int]: root_return = ctypes.c_ulong() parent_return = ctypes.c_ulong() children = ctypes.POINTER(ctypes.c_ulong)() child_count = ctypes.c_uint() status = self.library.XQueryTree( self.display, window, ctypes.byref(root_return), ctypes.byref(parent_return), ctypes.byref(children), ctypes.byref(child_count), ) if status == 0: return [] try: return [int(children[index]) for index in range(child_count.value)] finally: if children: self.library.XFree(ctypes.cast(children, ctypes.c_void_p)) def _client_window(self, window: int) -> int: """Resolve a window-manager frame to the ICCCM client it contains.""" queue = [window] visited: set[int] = set() while queue and len(visited) < 64: candidate = queue.pop(0) if candidate in visited: continue visited.add(candidate) if self._has_wm_state(candidate): return candidate queue.extend(self._child_windows(candidate)) return window def _intern_atom(self, name: bytes, *, only_if_exists: bool) -> int: """Return one atom ID without repeating X-server lookups.""" try: atoms = self._atoms except AttributeError: # Keep helpers usable in isolated tests that construct the mapper without # opening a real X11 display. atoms = {} self._atoms = atoms cached_atom = atoms.get(name) if cached_atom is not None: return cached_atom atom = int( self.library.XInternAtom( self.display, name, int(only_if_exists), ) ) # A missing only-if-exists lookup must remain retryable: a later call may # create the atom or another client may publish it. if atom != 0: atoms[name] = atom return atom def _has_wm_state(self, window: int) -> bool: wm_state = self._intern_atom(b"WM_STATE", only_if_exists=True) if wm_state == 0: return False actual_type = ctypes.c_ulong() actual_format = ctypes.c_int() item_count = ctypes.c_ulong() bytes_after = ctypes.c_ulong() value = ctypes.POINTER(ctypes.c_ubyte)() status = self.library.XGetWindowProperty( self.display, window, wm_state, 0, 0, 0, 0, ctypes.byref(actual_type), ctypes.byref(actual_format), ctypes.byref(item_count), ctypes.byref(bytes_after), ctypes.byref(value), ) try: return status == 0 and actual_type.value != 0 finally: if value: self.library.XFree(ctypes.cast(value, ctypes.c_void_p)) def window_state(self, window: int) -> int | None: """Return the ICCCM WM_STATE value for a managed client.""" values = self._long_property(window, b"WM_STATE", 2) if not values: return None return values[0] def window_is_borderless(self, window: int) -> bool | None: """Return whether the client currently requests no Motif decorations.""" hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5) if hints is None or len(hints) < 3: return None if hints[0] & _MOTIF_HINTS_DECORATIONS == 0: return False return hints[2] == 0 def window_is_fullscreen_or_maximized(self, window: int) -> bool: """Return whether EWMH currently marks the client as screen-filling.""" states = self._long_property(window, b"_NET_WM_STATE", 64) if not states: return False return any( atom != 0 and atom in states for atom in self._screen_filling_state_atoms() ) def window_net_wm_state_flags(self, window: int) -> dict[str, bool | None]: """Read EWMH state flags that distinguish minimize, remap, and raise.""" states = self._long_property(window, b"_NET_WM_STATE", 64) def state(name: bytes) -> bool | None: atom = self._intern_atom(name, only_if_exists=True) if states is None or atom == 0: return None return atom in states return { "net_wm_state_present": states is not None, "fullscreen": state(b"_NET_WM_STATE_FULLSCREEN"), "maximized_horizontal": state(b"_NET_WM_STATE_MAXIMIZED_HORZ"), "maximized_vertical": state(b"_NET_WM_STATE_MAXIMIZED_VERT"), "hidden": state(b"_NET_WM_STATE_HIDDEN"), "above": state(b"_NET_WM_STATE_ABOVE"), "demands_attention": state(b"_NET_WM_STATE_DEMANDS_ATTENTION"), "skip_taskbar": state(b"_NET_WM_STATE_SKIP_TASKBAR"), } def window_diagnostic_snapshot(self, window: int) -> dict[str, object]: """Read the WM properties needed to diagnose focus and maximize behavior.""" state = self.window_state(window) size = self.window_size(window) net_state_flags = self.window_net_wm_state_flags(window) allowed_actions = self._long_property( window, b"_NET_WM_ALLOWED_ACTIONS", 64, ) def contains(values: tuple[int, ...] | None, atom: int) -> bool | None: if values is None or atom == 0: return None return atom in values resize_action = self._intern_atom( b"_NET_WM_ACTION_RESIZE", only_if_exists=True, ) maximize_horizontal_action = self._intern_atom( b"_NET_WM_ACTION_MAXIMIZE_HORZ", only_if_exists=True, ) maximize_vertical_action = self._intern_atom( b"_NET_WM_ACTION_MAXIMIZE_VERT", only_if_exists=True, ) minimize_action = self._intern_atom( b"_NET_WM_ACTION_MINIMIZE", only_if_exists=True, ) normal_hints = self._long_property(window, b"WM_NORMAL_HINTS", 18) normal_flags = normal_hints[0] if normal_hints else None has_minimum = bool(normal_flags & _P_MIN_SIZE) if normal_flags is not None else None has_maximum = bool(normal_flags & _P_MAX_SIZE) if normal_flags is not None else None fixed_size = ( bool( has_minimum and has_maximum and normal_hints is not None and len(normal_hints) >= 9 and normal_hints[5:7] == normal_hints[7:9] ) if normal_flags is not None else None ) motif_hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5) motif_flags = motif_hints[0] if motif_hints and len(motif_hints) >= 3 else None motif_functions = motif_hints[1] if motif_flags is not None else None motif_decorations = motif_hints[2] if motif_flags is not None else None return { "wm_state": _wm_state_name(state), "active": self.window_is_active(window), "focused": self.window_is_focused(window), "width": size[0] if size is not None else None, "height": size[1] if size is not None else None, "borderless": self.window_is_borderless(window), **net_state_flags, "allowed_actions_present": allowed_actions is not None, "allows_resize": contains(allowed_actions, resize_action), "allows_minimize": contains(allowed_actions, minimize_action), "allows_maximize_horizontal": contains( allowed_actions, maximize_horizontal_action, ), "allows_maximize_vertical": contains( allowed_actions, maximize_vertical_action, ), "normal_hints_present": normal_hints is not None, "normal_minimum_size": has_minimum, "normal_maximum_size": has_maximum, "normal_fixed_size": fixed_size, "normal_resize_increment": ( bool(normal_flags & _P_RESIZE_INC) if normal_flags is not None else None ), "motif_hints_present": motif_flags is not None, "motif_allows_resize": ( _motif_value_allows_action(motif_functions, _MOTIF_FUNC_RESIZE) if ( motif_flags is not None and motif_flags & _MOTIF_HINTS_FUNCTIONS and motif_functions is not None ) else None ), "motif_allows_maximize": ( _motif_value_allows_action(motif_functions, _MOTIF_FUNC_MAXIMIZE) if ( motif_flags is not None and motif_flags & _MOTIF_HINTS_FUNCTIONS and motif_functions is not None ) else None ), "motif_resize_decoration": ( _motif_value_allows_action(motif_decorations, _MOTIF_DECOR_RESIZE) if ( motif_flags is not None and motif_flags & _MOTIF_HINTS_DECORATIONS and motif_decorations is not None ) else None ), "motif_maximize_decoration": ( _motif_value_allows_action(motif_decorations, _MOTIF_DECOR_MAXIMIZE) if ( motif_flags is not None and motif_flags & _MOTIF_HINTS_DECORATIONS and motif_decorations is not None ) else None ), } def force_windowed_state(self, window: int) -> bool: """Remove fullscreen/maximized EWMH state and wait for WM acknowledgement. The state property must not be edited directly on a mapped client. EWMH requires a client message to the root window so KWin/XWayland can restore the window's previous geometry before the render surface is refreshed. Return ``True`` only when a screen-filling state was present and its removal was confirmed. Return ``False`` when the window was already normal. """ states = set(self._long_property(window, b"_NET_WM_STATE", 64) or ()) fullscreen, maximized_vertical, maximized_horizontal = ( self._screen_filling_state_atoms() ) maximized = tuple( atom for atom in (maximized_vertical, maximized_horizontal) if atom != 0 and atom in states ) fullscreen_is_set = fullscreen != 0 and fullscreen in states if not maximized and not fullscreen_is_set: return False state_atom = self._intern_atom(b"_NET_WM_STATE", only_if_exists=True) if state_atom == 0: raise GameboxRuntimeError("the window manager has no EWMH state atom") requests: list[tuple[int, int]] = [] if maximized: requests.append( (maximized[0], maximized[1] if len(maximized) > 1 else 0) ) if fullscreen_is_set: requests.append((fullscreen, 0)) for first_state, second_state in requests: if not self._send_root_client_message( window, state_atom, ( _NET_WM_STATE_REMOVE, first_state, second_state, _NET_WM_STATE_SOURCE_APPLICATION, 0, ), ): raise GameboxRuntimeError( "the window manager rejected the windowed-state request" ) if not self._wait_for_windowed_state(window): raise GameboxRuntimeError( "the window manager did not confirm normal windowed state" ) return True def enable_window_scaling(self, window: int, *, confirm: bool = True) -> bool: """Expose native resize/maximize actions without changing the game binary. Wine marks this game with identical current-size ``PMinSize`` and ``PMaxSize`` constraints and excludes resize/maximize from its Motif hints. Remove only that exact fixed-size pair and preserve unrelated aspect, increment, and geometry hints. Initial setup is retained only when the window manager confirms resize plus both maximize actions. Event-driven maintenance can skip the confirmation because support was already proven. """ current_size = self.window_size(window) normal_hints = self._long_property(window, b"WM_NORMAL_HINTS", 18) motif_hints = self._long_property(window, b"_MOTIF_WM_HINTS", 5) updated_normal_hints = ( _normal_hints_without_fixed_size(normal_hints, current_size) if normal_hints is not None and current_size is not None else None ) updated_motif_hints = ( _motif_hints_with_scaling(motif_hints) if motif_hints is not None else None ) normal_changed = ( normal_hints is not None and updated_normal_hints is not None and updated_normal_hints != normal_hints ) motif_changed = motif_hints is not None and updated_motif_hints != motif_hints if not normal_changed and not motif_changed: if not confirm: return True # ``_NET_WM_ALLOWED_ACTIONS`` belongs to the window manager. Wait for # its answer even when the client hints were already resizable; Muffin # can publish the property shortly after the SDL window is discovered. self.library.XSync(self.display, 0) return self._wait_for_scaling_actions(window) confirmed = False try: if normal_changed and updated_normal_hints is not None: self._replace_long_property( window, b"WM_NORMAL_HINTS", b"WM_SIZE_HINTS", updated_normal_hints, ) if motif_changed and updated_motif_hints is not None: self._replace_long_property( window, b"_MOTIF_WM_HINTS", b"_MOTIF_WM_HINTS", updated_motif_hints, ) if not confirm: self.library.XFlush(self.display) confirmed = True return True self.library.XSync(self.display, 0) confirmed = self._wait_for_scaling_actions(window) return confirmed finally: if not confirmed: restore_errors: list[Exception] = [] if motif_changed and motif_hints is not None: try: self._replace_long_property( window, b"_MOTIF_WM_HINTS", b"_MOTIF_WM_HINTS", motif_hints, ) except Exception as exc: restore_errors.append(exc) if normal_changed and normal_hints is not None: try: self._replace_long_property( window, b"WM_NORMAL_HINTS", b"WM_SIZE_HINTS", normal_hints, ) except Exception as exc: restore_errors.append(exc) try: self.library.XSync(self.display, 0) except Exception as exc: restore_errors.append(exc) if restore_errors: raise GameboxRuntimeError( "could not fully restore the original game-window hints" ) from restore_errors[0] def window_allows_scaling(self, window: int) -> bool: actions = self._long_property(window, b"_NET_WM_ALLOWED_ACTIONS", 64) if not actions: return False resize = self._intern_atom( b"_NET_WM_ACTION_RESIZE", only_if_exists=True, ) horizontal = self._intern_atom( b"_NET_WM_ACTION_MAXIMIZE_HORZ", only_if_exists=True, ) vertical = self._intern_atom( b"_NET_WM_ACTION_MAXIMIZE_VERT", only_if_exists=True, ) return resize != 0 and horizontal != 0 and vertical != 0 and { resize, horizontal, vertical, }.issubset(actions) def _wait_for_scaling_actions(self, window: int) -> bool: deadline = time.monotonic() + WINDOW_ACTION_CONFIRM_TIMEOUT_SECONDS while not self.window_allows_scaling(window): remaining = deadline - time.monotonic() if remaining <= 0: return False time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining)) return True def select_scaling_hint_events(self, window: int) -> None: """Subscribe this X11 connection to client-property changes.""" if self.library.XSelectInput( self.display, window, _X_PROPERTY_CHANGE_MASK, ) == 0: raise GameboxRuntimeError( "could not subscribe to game-window property changes" ) self.library.XSync(self.display, 0) def wait_for_scaling_hint_change( self, window: int, *, timeout_seconds: float = WINDOW_SCALING_EVENT_TIMEOUT_SECONDS, ) -> bool: """Wait efficiently for Wine to replace either maintained hint property.""" if self.library.XPending(self.display) == 0: connection = self.library.XConnectionNumber(self.display) if connection < 0: raise GameboxRuntimeError("the X11 display connection is unavailable") try: readable, _, _ = select.select( (connection,), (), (), timeout_seconds, ) except InterruptedError: return False if not readable: return False watched_atoms = { self._intern_atom(b"WM_NORMAL_HINTS", only_if_exists=False), self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False), self._intern_atom(b"_NET_WM_ALLOWED_ACTIONS", only_if_exists=False), } changed = False events_read = 0 while ( events_read < WINDOW_SCALING_EVENT_BATCH_LIMIT and self.library.XPending(self.display) > 0 ): event = _XEvent() self.library.XNextEvent(self.display, ctypes.byref(event)) events_read += 1 if ( event.type == _X_PROPERTY_NOTIFY and int(event.property.window) == window and int(event.property.atom) in watched_atoms ): changed = True return changed def _screen_filling_state_atoms(self) -> tuple[int, int, int]: return ( self._intern_atom(b"_NET_WM_STATE_FULLSCREEN", only_if_exists=True), self._intern_atom(b"_NET_WM_STATE_MAXIMIZED_VERT", only_if_exists=True), self._intern_atom(b"_NET_WM_STATE_MAXIMIZED_HORZ", only_if_exists=True), ) def _wait_for_windowed_state(self, window: int) -> bool: deadline = time.monotonic() + WINDOW_STATE_CONFIRM_TIMEOUT_SECONDS while self.window_is_fullscreen_or_maximized(window): remaining = deadline - time.monotonic() if remaining <= 0: return False time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining)) return True def _long_property( self, window: int, name: bytes, maximum_items: int, ) -> tuple[int, ...] | None: atom = self._intern_atom(name, only_if_exists=True) if atom == 0: return None actual_type = ctypes.c_ulong() actual_format = ctypes.c_int() item_count = ctypes.c_ulong() bytes_after = ctypes.c_ulong() value = ctypes.POINTER(ctypes.c_ubyte)() status = self.library.XGetWindowProperty( self.display, window, atom, 0, maximum_items, 0, 0, ctypes.byref(actual_type), ctypes.byref(actual_format), ctypes.byref(item_count), ctypes.byref(bytes_after), ctypes.byref(value), ) try: if ( status != 0 or actual_type.value == 0 or actual_format.value != 32 or not value ): return None longs = ctypes.cast(value, ctypes.POINTER(ctypes.c_ulong)) return tuple( int(longs[index] & 0xFFFFFFFF) for index in range(min(item_count.value, maximum_items)) ) finally: if value: self.library.XFree(ctypes.cast(value, ctypes.c_void_p)) def _replace_long_property( self, window: int, name: bytes, property_type: bytes, values: Sequence[int], ) -> None: atom = self._intern_atom(name, only_if_exists=False) type_atom = self._intern_atom(property_type, only_if_exists=False) if atom == 0 or type_atom == 0: raise GameboxRuntimeError("could not create an X11 window-property atom") property_values = (ctypes.c_ulong * len(values))( *(value & 0xFFFFFFFF for value in values) ) self.library.XChangeProperty( self.display, window, atom, type_atom, 32, 0, ctypes.cast(property_values, ctypes.POINTER(ctypes.c_ubyte)), len(values), ) def window_size(self, window: int) -> tuple[int, int] | None: """Return the current X11 client dimensions.""" root = ctypes.c_ulong() x = ctypes.c_int() y = ctypes.c_int() width = ctypes.c_uint() height = ctypes.c_uint() border_width = ctypes.c_uint() depth = ctypes.c_uint() if self.library.XGetGeometry( self.display, window, ctypes.byref(root), ctypes.byref(x), ctypes.byref(y), ctypes.byref(width), ctypes.byref(height), ctypes.byref(border_width), ctypes.byref(depth), ) == 0: return None return width.value, height.value def _wait_for_window_size( self, window: int, expected_size: tuple[int, int], *, timeout_seconds: float = WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS, ) -> bool: """Wait briefly for an asynchronous WM ConfigureRequest decision.""" deadline = time.monotonic() + timeout_seconds while self.window_size(window) != expected_size: remaining = deadline - time.monotonic() if remaining <= 0: return False time.sleep(min(WINDOW_CONFIRM_POLL_SECONDS, remaining)) return True def _resize_window_and_wait( self, window: int, target_size: tuple[int, int], *, timeout_seconds: float = WINDOW_GEOMETRY_CONFIRM_TIMEOUT_SECONDS, ) -> bool: width, height = target_size self.library.XResizeWindow(self.display, window, width, height) self.library.XSync(self.display, 0) return self._wait_for_window_size( window, target_size, timeout_seconds=timeout_seconds, ) def map_window( self, window: int, *, borderless: bool, ) -> tuple[int, int] | None: # Capture the real client size for both decorated and borderless windows. # Wine/DXVK can otherwise retain a desktop-sized render surface inside a # smaller decorated client on some window managers. target_size = self.window_size(window) if borderless: self._remove_decorations(window) self.library.XMapRaised(self.display, window) self.library.XFlush(self.display) return target_size def refresh_render_surface( self, window: int, *, target_size: tuple[int, int] | None = None, ) -> bool: """Nudge the mapped client size once after Wine creates the window. Some Wine/DXVK combinations retain the host desktop-sized render surface inside a smaller top-level game window. A one-pixel resize followed by an exact restore makes Wine reconcile the surface with the real client size. Never iconify or unmap a Wine client for this refresh. Wine can process the resulting state change after startup and minimize the game at its login screen. """ if target_size is None: target_size = self.window_size(window) if target_size is None: return False original_width, original_height = target_size if original_width > 1: nudged_width = original_width - 1 nudged_height = original_height elif original_height > 1: nudged_width = original_width nudged_height = original_height - 1 else: return False try: nudge_confirmed = self._resize_window_and_wait( window, (nudged_width, nudged_height), timeout_seconds=WINDOW_NUDGE_CONFIRM_TIMEOUT_SECONDS, ) if nudge_confirmed: time.sleep(WINDOW_SURFACE_REFRESH_DELAY_SECONDS) finally: # Always request and verify the exact starting dimensions, including when # KWin declined the nudge or the short wait was interrupted. restore_confirmed = self._resize_window_and_wait( window, (original_width, original_height), ) return nudge_confirmed and restore_confirmed def focus_window(self, window: int) -> None: """Raise the mapped game client and give it keyboard focus.""" self.library.XMapRaised(self.display, window) self.restore_input_focus(window) def restore_input_focus(self, window: int) -> None: """Give a mapped client keyboard focus without raising its host window.""" self.library.XSetInputFocus( self.display, window, _X_REVERT_TO_PARENT, _X_CURRENT_TIME, ) self.library.XSync(self.display, 0) def window_is_focused(self, window: int) -> bool: focused = ctypes.c_ulong() revert_to = ctypes.c_int() if self.library.XGetInputFocus( self.display, ctypes.byref(focused), ctypes.byref(revert_to), ) != 1: return False return int(focused.value) == window def window_is_active(self, window: int) -> bool: """Return whether the desktop already considers this client active.""" root = self.library.XDefaultRootWindow(self.display) active = self._long_property(root, b"_NET_ACTIVE_WINDOW", 1) return bool(active and active[0] == window) def minimize_window(self, window: int) -> bool: """Ask the host window manager to keep a remapped client minimized.""" screen = self.library.XDefaultScreen(self.display) accepted = self.library.XIconifyWindow(self.display, window, screen) != 0 self.library.XSync(self.display, 0) return accepted def activate_window(self, window: int) -> bool: """Ask the window manager to restore and activate one client.""" active_window = self._intern_atom( b"_NET_ACTIVE_WINDOW", only_if_exists=False, ) if active_window == 0: return False return self._send_root_client_message( window, active_window, (1, _X_CURRENT_TIME, 0, 0, 0), ) def _send_root_client_message( self, window: int, message_type: int, data: Sequence[int], ) -> bool: """Send one 32-bit EWMH client message to the root window.""" if len(data) != 5: raise ValueError("an X11 client message requires exactly five values") event = _XEvent() event.client_message.type = _X_CLIENT_MESSAGE event.client_message.serial = 0 event.client_message.send_event = 1 event.client_message.display = self.display event.client_message.window = window event.client_message.message_type = message_type event.client_message.format = 32 for index, value in enumerate(data): event.client_message.data.longs[index] = value status = self.library.XSendEvent( self.display, self.library.XDefaultRootWindow(self.display), 0, _X_SUBSTRUCTURE_NOTIFY_MASK | _X_SUBSTRUCTURE_REDIRECT_MASK, ctypes.byref(event), ) self.library.XSync(self.display, 0) return status != 0 def set_window_identity( self, window: int, *, resource_name: str = APPLICATION_WM_INSTANCE, resource_class: str = APPLICATION_WM_CLASS, ) -> None: """Replace Proton's generic steamapp identity before taskbar matching.""" for label, value in ( ("resource name", resource_name), ("resource class", resource_class), ): if not value or "\x00" in value: raise GameboxRuntimeError(f"invalid X11 {label}") hints = _XClassHint( res_name=resource_name.encode("utf-8"), res_class=resource_class.encode("utf-8"), ) if self.library.XSetClassHint( self.display, window, ctypes.byref(hints), ) == 0: raise GameboxRuntimeError("could not set the game window identity") self.library.XSync(self.display, 0) def suppress_wm_ping(self, window: int) -> bool: """Stop the WM hang prompt for one window, preserving its other protocols.""" ping_atom = self._intern_atom(b"_NET_WM_PING", only_if_exists=False) if ping_atom == 0: return False protocols = self._wm_protocols(window) if protocols is None: return False filtered_protocols = _protocols_without(protocols, ping_atom) if filtered_protocols == protocols: return True if filtered_protocols: protocol_values = (ctypes.c_ulong * len(filtered_protocols))( *filtered_protocols ) status = self.library.XSetWMProtocols( self.display, window, protocol_values, len(filtered_protocols), ) else: status = self.library.XSetWMProtocols( self.display, window, ctypes.POINTER(ctypes.c_ulong)(), 0, ) if status == 0: return False self.library.XSync(self.display, 0) return self._wm_protocols(window) == filtered_protocols def _wm_protocols(self, window: int) -> tuple[int, ...] | None: protocols = ctypes.POINTER(ctypes.c_ulong)() protocol_count = ctypes.c_int() status = self.library.XGetWMProtocols( self.display, window, ctypes.byref(protocols), ctypes.byref(protocol_count), ) if status == 0: return None try: return tuple( int(protocols[index]) for index in range(protocol_count.value) ) finally: if protocols: self.library.XFree(ctypes.cast(protocols, ctypes.c_void_p)) def _remove_decorations(self, window: int) -> None: atom = self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False) if atom == 0: return hints = _MotifWmHints( flags=1 << 1, functions=0, decorations=0, input_mode=0, status=0, ) self.library.XChangeProperty( self.display, window, atom, atom, 32, 0, ctypes.cast(ctypes.byref(hints), ctypes.POINTER(ctypes.c_ubyte)), 5, ) def _window_title(self, window: int) -> str | None: value = ctypes.c_char_p() if self.library.XFetchName(self.display, window, ctypes.byref(value)) == 0: return None try: if value.value is None: return None return value.value.decode("utf-8", errors="replace") finally: self.library.XFree(ctypes.cast(value, ctypes.c_void_p)) def _restore_error_handler(self) -> None: if self._error_handler is None: return self.library.XSetErrorHandler(self._previous_error_handler) self._error_handler = None def _configure_signatures(self) -> None: self.library.XSetErrorHandler.argtypes = [ctypes.c_void_p] self.library.XSetErrorHandler.restype = ctypes.c_void_p self.library.XOpenDisplay.argtypes = [ctypes.c_char_p] self.library.XOpenDisplay.restype = ctypes.c_void_p self.library.XDefaultRootWindow.argtypes = [ctypes.c_void_p] self.library.XDefaultRootWindow.restype = ctypes.c_ulong self.library.XDefaultScreen.argtypes = [ctypes.c_void_p] self.library.XDefaultScreen.restype = ctypes.c_int self.library.XIconifyWindow.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int, ] self.library.XIconifyWindow.restype = ctypes.c_int self.library.XQueryTree.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)), ctypes.POINTER(ctypes.c_uint), ] self.library.XQueryTree.restype = ctypes.c_int self.library.XFetchName.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_char_p), ] self.library.XFetchName.restype = ctypes.c_int self.library.XInternAtom.argtypes = [ ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int, ] self.library.XInternAtom.restype = ctypes.c_ulong self.library.XChangeProperty.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_int, ctypes.c_int, ctypes.POINTER(ctypes.c_ubyte), ctypes.c_int, ] self.library.XChangeProperty.restype = ctypes.c_int self.library.XGetWindowProperty.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_long, ctypes.c_long, ctypes.c_int, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.POINTER(ctypes.c_ubyte)), ] self.library.XGetWindowProperty.restype = ctypes.c_int self.library.XSetClassHint.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(_XClassHint), ] self.library.XSetClassHint.restype = ctypes.c_int self.library.XGetWMProtocols.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.POINTER(ctypes.c_ulong)), ctypes.POINTER(ctypes.c_int), ] self.library.XGetWMProtocols.restype = ctypes.c_int self.library.XSetWMProtocols.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong), ctypes.c_int, ] self.library.XSetWMProtocols.restype = ctypes.c_int self.library.XMapRaised.argtypes = [ctypes.c_void_p, ctypes.c_ulong] self.library.XMapRaised.restype = ctypes.c_int self.library.XSelectInput.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_long, ] self.library.XSelectInput.restype = ctypes.c_int self.library.XConnectionNumber.argtypes = [ctypes.c_void_p] self.library.XConnectionNumber.restype = ctypes.c_int self.library.XPending.argtypes = [ctypes.c_void_p] self.library.XPending.restype = ctypes.c_int self.library.XNextEvent.argtypes = [ ctypes.c_void_p, ctypes.POINTER(_XEvent), ] self.library.XNextEvent.restype = ctypes.c_int self.library.XGetGeometry.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_uint), ctypes.POINTER(ctypes.c_uint), ctypes.POINTER(ctypes.c_uint), ctypes.POINTER(ctypes.c_uint), ] self.library.XGetGeometry.restype = ctypes.c_int self.library.XResizeWindow.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_uint, ctypes.c_uint, ] self.library.XResizeWindow.restype = ctypes.c_int self.library.XSetInputFocus.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int, ctypes.c_ulong, ] self.library.XSetInputFocus.restype = ctypes.c_int self.library.XGetInputFocus.argtypes = [ ctypes.c_void_p, ctypes.POINTER(ctypes.c_ulong), ctypes.POINTER(ctypes.c_int), ] self.library.XGetInputFocus.restype = ctypes.c_int self.library.XSendEvent.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int, ctypes.c_long, ctypes.POINTER(_XEvent), ] self.library.XSendEvent.restype = ctypes.c_int self.library.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int] self.library.XSync.restype = ctypes.c_int self.library.XFlush.argtypes = [ctypes.c_void_p] self.library.XFlush.restype = ctypes.c_int self.library.XFree.argtypes = [ctypes.c_void_p] self.library.XFree.restype = ctypes.c_int self.library.XCloseDisplay.argtypes = [ctypes.c_void_p] self.library.XCloseDisplay.restype = ctypes.c_int def _window_diagnostic_details(mapper: _X11, window: int) -> dict[str, object]: """Keep diagnostic reads from affecting window supervision.""" try: return mapper.window_diagnostic_snapshot(window) except Exception as exc: return {"snapshot_error": type(exc).__name__} def _sample_window_diagnostic_details( diagnostics: WindowDiagnosticRecorder | None, mapper: _X11, window: int, ) -> dict[str, object]: if diagnostics is None or not diagnostics.accepts_samples: return {} return _window_diagnostic_details(mapper, window) def _prepare_game_window( mapper: _X11, window: int, *, borderless: bool, suppress_wm_ping: bool, ) -> None: if suppress_wm_ping: try: ping_suppressed = mapper.suppress_wm_ping(window) except Exception as exc: print( f"Warning: could not suppress game-window hang warnings ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) else: if ping_suppressed: print( "Game-window hang warnings suppressed for this launch.", file=sys.stderr, flush=True, ) else: print( "Warning: could not verify game-window hang-warning " "suppression; continuing normally", file=sys.stderr, flush=True, ) target_size = mapper.map_window(window, borderless=borderless) forced_windowed = False try: forced_windowed = mapper.force_windowed_state(window) except Exception as exc: print( f"Warning: could not force the game into normal windowed mode ({exc}); " "continuing with surface synchronization", file=sys.stderr, flush=True, ) else: if forced_windowed: print( "Game window forced out of fullscreen/maximized mode.", file=sys.stderr, flush=True, ) if target_size is not None or forced_windowed: time.sleep(WINDOW_SURFACE_SETTLE_SECONDS) if forced_windowed: # Let KWin finish restoring the pre-fullscreen client geometry before it # becomes the authoritative surface size. try: target_size = mapper.window_size(window) except Exception as exc: print( f"Warning: could not read the restored game-window size ({exc}); " "retrying during surface synchronization", file=sys.stderr, flush=True, ) target_size = None try: if target_size is None: refreshed = mapper.refresh_render_surface(window) else: refreshed = mapper.refresh_render_surface( window, target_size=target_size, ) except Exception as exc: print( f"Warning: could not synchronize the game surface ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) else: if refreshed: print( "Game render surface synchronized with its window.", file=sys.stderr, flush=True, ) else: print( "Warning: the window manager declined the game-surface " "refresh; continuing normally", file=sys.stderr, flush=True, ) def _apply_game_window_identity(mapper: _X11, window: int) -> None: try: mapper.set_window_identity(window) except Exception as exc: print( f"Warning: could not assign the PristonTale EU taskbar identity ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) def _focus_game_window(mapper: _X11, window: int) -> None: """Let the dashboard process its ready event before focusing the game.""" time.sleep(WINDOW_FOCUS_DELAY_SECONDS) try: mapper.focus_window(window) except Exception as exc: print( f"Warning: could not focus the game window ({exc}); continuing normally", file=sys.stderr, flush=True, ) def _repair_game_window_focus( mapper: _X11, window: int, *, diagnostics: WindowDiagnosticRecorder | None = None, ) -> bool: """Repair Wine's input focus only while its desktop window remains active.""" try: state = mapper.window_state(window) active = mapper.window_is_active(window) focused = mapper.window_is_focused(window) if active else False if diagnostics is not None: diagnostics.transition( "focus_gate", (state, active, focused), "focus_state_changed", window_id=window, wm_state=_wm_state_name(state), active=active, focused=focused, ) if state == _X_WM_STATE_ICONIC: return False if not active: return False if focused: return True except Exception: return False try: mapper.restore_input_focus(window) repaired = mapper.window_is_focused(window) if diagnostics is not None: diagnostics.record( "focus_repair", window_id=window, succeeded=repaired, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) return repaired except Exception as exc: if diagnostics is not None: diagnostics.record( "focus_repair_failed", window_id=window, error_type=type(exc).__name__, ) print( f"Warning: could not repair game-window focus ({exc}); continuing " "normally", file=sys.stderr, flush=True, ) return False def _repair_gamescope_game_window_focus( mapper: _X11, window: int, *, diagnostics: WindowDiagnosticRecorder | None = None, ) -> bool: """Restore nested focus without asking the desktop to activate Gamescope.""" host_focus = _gamescope_host_focus_status() if diagnostics is not None: diagnostics.transition( "host_focus_bridge", (host_focus.get("allowed"), host_focus.get("bridge")), "host_focus_bridge_changed", **host_focus, ) if host_focus["allowed"] is not True: return False try: state = mapper.window_state(window) focused = mapper.window_is_focused(window) if diagnostics is not None: diagnostics.transition( "nested_focus_gate", (state, focused), "nested_focus_state_changed", window_id=window, wm_state=_wm_state_name(state), focused=focused, ) if state == _X_WM_STATE_ICONIC: return False if focused: return True except Exception: return False try: mapper.restore_input_focus(window) repaired = mapper.window_is_focused(window) if diagnostics is not None: diagnostics.record( "nested_focus_repair", window_id=window, succeeded=repaired, host_focus=host_focus, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) return repaired except Exception as exc: if diagnostics is not None: diagnostics.record( "nested_focus_repair_failed", window_id=window, error_type=type(exc).__name__, host_focus=host_focus, ) print( f"Warning: could not restore nested game input focus ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) return False def _gamescope_host_focus_is_allowed( *, environment: Mapping[str, str] | None = None, now: float | None = None, ) -> bool: """Honor the outer X11 window's current focus without trusting stale state.""" return _gamescope_host_focus_status( environment=environment, now=now, )["allowed"] is True def _gamescope_host_focus_status( *, environment: Mapping[str, str] | None = None, now: float | None = None, ) -> dict[str, object]: """Explain whether the nested supervisor may repair input focus.""" env = os.environ if environment is None else environment state_path = env.get(GAMESCOPE_HOST_FOCUS_STATE_ENV, "").strip() if not state_path: # Direct invocations and native Wayland Gamescope have no X11 host-state # bridge. Retain their existing nested focus behavior. return {"allowed": True, "bridge": "absent"} if state_path == "unavailable": return {"allowed": False, "bridge": "unavailable"} try: fields = Path(state_path).read_text(encoding="ascii").split() if len(fields) != 2 or fields[0] not in {"0", "1"}: return {"allowed": False, "bridge": "invalid"} published_at = float(fields[1]) except (OSError, UnicodeError, ValueError): return {"allowed": False, "bridge": "unreadable"} observed_at = time.monotonic() if now is None else now age = observed_at - published_at age_ms = round(age * 1000) if age < 0.0: return {"allowed": False, "bridge": "future", "age_ms": age_ms} if age > GAMESCOPE_HOST_FOCUS_STATE_MAX_AGE_SECONDS: return {"allowed": False, "bridge": "stale", "age_ms": age_ms} active = fields[0] == "1" return { "allowed": active, "bridge": "active" if active else "inactive", "age_ms": age_ms, } def _resolve_gamescope_game_window( mapper: _X11, window: int, *, window_title: str, excluded_window_ids: Collection[int], diagnostics: WindowDiagnosticRecorder | None = None, ) -> int | None: """Return a valid nested client across Wine's login-window transition.""" try: state = mapper.window_state(window) except Exception: state = None if diagnostics is not None: diagnostics.transition( "nested_window_state", (window, state), "nested_window_state_changed", window_id=window, wm_state=_wm_state_name(state), ) if state == _X_WM_STATE_ICONIC: # IconicState can be a real desktop minimize request. Monitoring must never # turn that user-controlled state back into a visible window. return None if state is not None and state != _X_WM_STATE_WITHDRAWN: return window try: found = mapper.find_first_window( window_title, excluded_window_ids=set(excluded_window_ids) | {window}, ) except Exception: return None if found is None: return None replacement, title = found _apply_game_window_identity(mapper, replacement) if diagnostics is not None: diagnostics.record( "nested_window_replaced", previous_window_id=window, window_id=replacement, **_sample_window_diagnostic_details(diagnostics, mapper, replacement), ) print( f"Game login window changed; following it inside Gamescope: {title}", file=sys.stderr, flush=True, ) return replacement def _enable_game_window_scaling( mapper: _X11, window: int, *, diagnostics: WindowDiagnosticRecorder | None = None, ) -> bool: """Enable confirmed resize/maximize actions without risking game startup.""" try: enabled = mapper.enable_window_scaling(window) except Exception as exc: if diagnostics is not None: diagnostics.record( "scaling_configuration_failed", window_id=window, error_type=type(exc).__name__, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) print( f"Warning: could not enable game-window scaling ({exc}); " "the fixed-size window will continue normally", file=sys.stderr, flush=True, ) return False if diagnostics is not None: diagnostics.record( "scaling_configured", window_id=window, enabled=enabled, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) if enabled: print( "Game window is resizable with persistent maximize support.", file=sys.stderr, flush=True, ) else: print( "Warning: the window manager did not confirm safe scaling actions; " "the game remains fixed-size", file=sys.stderr, flush=True, ) return enabled def _maintain_game_window_scaling( mapper: _X11, process: subprocess.Popen[bytes], window: int, *, diagnostics: WindowDiagnosticRecorder | None = None, ) -> int: """Keep host scaling hints across Wine maximize/restore transitions.""" try: mapper.select_scaling_hint_events(window) except Exception as exc: if diagnostics is not None: diagnostics.record( "scaling_monitor_failed", window_id=window, error_type=type(exc).__name__, ) print( f"Warning: could not monitor game-window scaling hints ({exc}); " "leaving the game window fixed-size for consistency", file=sys.stderr, flush=True, ) return int(process.wait()) scaling_enabled = ( _enable_game_window_scaling(mapper, window) if diagnostics is None else _enable_game_window_scaling( mapper, window, diagnostics=diagnostics, ) ) if not scaling_enabled: return int(process.wait()) repair_failures = 0 next_scaling_repair_at = 0.0 next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS while process.poll() is None: try: hints_changed = mapper.wait_for_scaling_hint_change(window) except Exception as exc: print( f"Warning: game-window scaling monitoring stopped ({exc}); " "the game will continue normally", file=sys.stderr, flush=True, ) break now = time.monotonic() if now >= next_focus_repair_at: if diagnostics is None: _repair_game_window_focus(mapper, window) else: _repair_game_window_focus( mapper, window, diagnostics=diagnostics, ) next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS if not hints_changed and now < next_scaling_repair_at: continue try: mapper.enable_window_scaling( window, confirm=False, ) except Exception as exc: repair_failures += 1 if diagnostics is not None: diagnostics.record( "scaling_maintenance_failed", window_id=window, error_type=type(exc).__name__, repair_failures=repair_failures, ) print( f"Warning: could not restore game-window scaling hints ({exc})", file=sys.stderr, flush=True, ) if repair_failures >= WINDOW_SCALING_REPAIR_LIMIT: print( "Warning: scaling maintenance was disabled after repeated failures; " "the game will continue normally", file=sys.stderr, flush=True, ) break else: repair_failures = 0 if hints_changed and diagnostics is not None: diagnostics.record( "scaling_hints_repaired", window_id=window, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) next_scaling_repair_at = time.monotonic() + WINDOW_SCALING_REPAIR_INTERVAL_SECONDS return int(process.wait()) def _maintain_game_window_focus( mapper: _X11, process: subprocess.Popen[bytes], window: int, *, diagnostics: WindowDiagnosticRecorder | None = None, ) -> int: """Keep the keyboard-focus repair while retaining Wine's fixed-size hints.""" print( "Game window resizing is disabled to prevent repeated Direct3D " "device resets.", file=sys.stderr, flush=True, ) next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS while process.poll() is None: time.sleep(WINDOW_SCALING_EVENT_TIMEOUT_SECONDS) now = time.monotonic() if now < next_focus_repair_at: continue if diagnostics is None: _repair_game_window_focus(mapper, window) else: _repair_game_window_focus( mapper, window, diagnostics=diagnostics, ) next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS return int(process.wait()) def _maintain_gamescope_game_window_focus( mapper: _X11, process: subprocess.Popen[bytes], window: int, *, window_title: str = WINDOW_TITLE, existing_window_ids: Collection[int] = (), diagnostics: WindowDiagnosticRecorder | None = None, ) -> int: """Keep a valid, focused nested client through login and host resizing.""" active_window: int = window retired_window_ids = set(existing_window_ids) next_focus_repair_at = ( time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS ) while process.poll() is None: time.sleep(GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS) now = time.monotonic() if now < next_focus_repair_at: continue if diagnostics is None: resolved_window = _resolve_gamescope_game_window( mapper, active_window, window_title=window_title, excluded_window_ids=retired_window_ids, ) else: resolved_window = _resolve_gamescope_game_window( mapper, active_window, window_title=window_title, excluded_window_ids=retired_window_ids, diagnostics=diagnostics, ) if resolved_window is None: next_focus_repair_at = ( time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS ) continue resolved_window_id = resolved_window if resolved_window_id != active_window: retired_window_ids.add(active_window) active_window = resolved_window_id if diagnostics is None: _repair_gamescope_game_window_focus(mapper, active_window) else: _repair_gamescope_game_window_focus( mapper, active_window, diagnostics=diagnostics, ) next_focus_repair_at = ( time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS ) return int(process.wait()) def _guard_game_window_startup( mapper: _X11, process: subprocess.Popen[bytes], window: int, *, borderless: bool, diagnostics: WindowDiagnosticRecorder | None = None, ) -> bool: """Recover Wine's login-screen window transition without hiding the client.""" deadline = time.monotonic() + WINDOW_STARTUP_GUARD_SECONDS try: initial_borderless = ( mapper.window_is_borderless(window) if borderless else None ) # A readable decorated value can already be Wine's fast login reset. The # initial preparation immediately before this guard requested borderless. borderless_was_applied = borderless and initial_borderless is not None except Exception as exc: if diagnostics is not None: diagnostics.record( "startup_monitor_failed", window_id=window, error_type=type(exc).__name__, ) print( f"Warning: could not monitor the game startup window ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) return False if diagnostics is not None: diagnostics.record( "startup_guard_started", window_id=window, borderless_requested=borderless, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) recovery_count = 0 while process.poll() is None: if time.monotonic() >= deadline: return True try: state = mapper.window_state(window) screen_filling = mapper.window_is_fullscreen_or_maximized(window) borderless_now = ( mapper.window_is_borderless(window) if borderless else None ) except Exception as exc: if diagnostics is not None: diagnostics.record( "startup_monitor_failed", window_id=window, error_type=type(exc).__name__, ) print( f"Warning: could not inspect the game startup window ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) return False became_iconic = state == _X_WM_STATE_ICONIC entered_screen_filling_mode = screen_filling is True login_style_reset = borderless_was_applied and borderless_now is False if diagnostics is not None: diagnostics.transition( "startup_window_state", (state, screen_filling, borderless_now), "startup_state_changed", window_id=window, wm_state=_wm_state_name(state), screen_filling=screen_filling, borderless=borderless_now, ) if became_iconic: print( "Game startup window was minimized; leaving it minimized.", file=sys.stderr, flush=True, ) return False if ( not entered_screen_filling_mode and not login_style_reset ): time.sleep(WINDOW_STARTUP_POLL_SECONDS) continue recovery_count += 1 if entered_screen_filling_mode: reason = "entered fullscreen/maximized mode" else: reason = "changed its window layout" print( f"Game startup window {reason}; restoring and refocusing it.", file=sys.stderr, flush=True, ) if diagnostics is not None: diagnostics.record( "startup_recovery_requested", window_id=window, reason=reason, recovery_count=recovery_count, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) try: mapper.activate_window(window) time.sleep(WINDOW_ACTIVATION_DELAY_SECONDS) _prepare_game_window( mapper, window, borderless=borderless, suppress_wm_ping=False, ) mapper.activate_window(window) mapper.focus_window(window) if borderless: borderless_was_applied = ( mapper.window_is_borderless(window) is True ) except Exception as exc: if diagnostics is not None: diagnostics.record( "startup_recovery_failed", window_id=window, reason=reason, error_type=type(exc).__name__, ) print( f"Warning: could not recover the game startup window ({exc}); " "continuing normally", file=sys.stderr, flush=True, ) return False if diagnostics is not None: diagnostics.record( "startup_recovery_completed", window_id=window, reason=reason, **_sample_window_diagnostic_details(diagnostics, mapper, window), ) if recovery_count >= WINDOW_STARTUP_RECOVERY_LIMIT: print( "Warning: the game repeatedly changed its startup window; " "stopped automatic recovery.", file=sys.stderr, flush=True, ) return False time.sleep(WINDOW_STARTUP_POLL_SECONDS) return False def run_game_window( command: Sequence[str], window_title: str = WINDOW_TITLE, *, borderless: bool = False, allow_window_resizing: bool = False, gamescope_scaling: bool = False, suppress_wm_ping: bool = False, ) -> int: if not command: raise GameboxRuntimeError("game command cannot be empty") diagnostics = WindowDiagnosticRecorder("game_window") diagnostics.record( "runtime_configured", borderless=borderless, allow_window_resizing=allow_window_resizing, gamescope_scaling=gamescope_scaling, suppress_wm_ping=suppress_wm_ping, display_available=bool(os.environ.get("DISPLAY")), ) diagnostic_sink = diagnostics if diagnostics.enabled else None mapper: _X11 | None = None existing_window_ids: set[int] = set() try: created_mapper = _X11() existing_window_ids = created_mapper.root_window_ids() mapper = created_mapper except GameboxRuntimeError as error: diagnostics.record( "x11_unavailable", error_type=type(error).__name__, ) print( f"Warning: {error}; continuing without X11 window styling", file=sys.stderr, flush=True, ) try: process = subprocess.Popen([str(value) for value in command]) except BaseException: if mapper is not None: mapper.close() diagnostics.close(process_started=False, managed=mapper is not None) raise if mapper is None: print( f"Game window ready: unmanaged {window_title}", file=sys.stderr, flush=True, ) try: return int(process.wait()) finally: diagnostics.close(process_started=True, managed=False) game_window: int | None = None game_title: str | None = None game_window_created_at: float | None = None deadline = time.monotonic() + WINDOW_DISCOVERY_TIMEOUT_SECONDS try: while process.poll() is None and time.monotonic() < deadline: now = time.monotonic() if game_window is None: found = mapper.find_first_window( window_title, excluded_window_ids=existing_window_ids, ) if found is not None: window, title = found game_window = window game_title = title game_window_created_at = now _apply_game_window_identity(mapper, window) if diagnostic_sink is not None: diagnostic_sink.record( "game_window_discovered", window_id=window, **_sample_window_diagnostic_details( diagnostic_sink, mapper, window, ), ) print( f"Game window created; preparing it: {title}", file=sys.stderr, flush=True, ) if ( game_window is not None and game_window_created_at is not None and now - game_window_created_at >= WINDOW_READY_DELAY_SECONDS ): if gamescope_scaling: # Gamescope owns the nested child's fullscreen/maximized state and # surface geometry. Trying to restore or nudge that window makes the # two window managers fight and can produce visible frame flicker. print( f"Game window ready: {game_title or window_title}", file=sys.stderr, flush=True, ) if diagnostic_sink is not None: diagnostic_sink.record( "game_window_ready", window_id=game_window, gamescope_scaling=True, host_focus=_gamescope_host_focus_status(), **_sample_window_diagnostic_details( diagnostic_sink, mapper, game_window, ), ) if _gamescope_host_focus_is_allowed(): _focus_game_window(mapper, game_window) print( "Game presentation is isolated; Gamescope handles " "aspect-preserving outer-window resizing.", file=sys.stderr, flush=True, ) return _maintain_gamescope_game_window_focus( mapper, process, game_window, window_title=window_title, existing_window_ids=existing_window_ids, **( {"diagnostics": diagnostic_sink} if diagnostic_sink is not None else {} ), ) _prepare_game_window( mapper, game_window, borderless=borderless, suppress_wm_ping=suppress_wm_ping, ) print( f"Game window ready: {game_title or window_title}", file=sys.stderr, flush=True, ) if diagnostic_sink is not None: diagnostic_sink.record( "game_window_ready", window_id=game_window, gamescope_scaling=False, **_sample_window_diagnostic_details( diagnostic_sink, mapper, game_window, ), ) _focus_game_window(mapper, game_window) startup_stable = _guard_game_window_startup( mapper, process, game_window, borderless=borderless, **( {"diagnostics": diagnostic_sink} if diagnostic_sink is not None else {} ), ) if ( allow_window_resizing and not gamescope_scaling and startup_stable and not borderless ): return _maintain_game_window_scaling( mapper, process, game_window, **( {"diagnostics": diagnostic_sink} if diagnostic_sink is not None else {} ), ) if not borderless: if allow_window_resizing and not startup_stable: print( "Warning: live window resizing was not enabled because the " "startup window did not stabilize.", file=sys.stderr, flush=True, ) return _maintain_game_window_focus( mapper, process, game_window, **( {"diagnostics": diagnostic_sink} if diagnostic_sink is not None else {} ), ) return int(process.wait()) time.sleep(0.1) if process.poll() is None: print( f"Warning: no new game window matching {window_title!r} appeared", file=sys.stderr, flush=True, ) if game_window is not None and not gamescope_scaling: _prepare_game_window( mapper, game_window, borderless=borderless, suppress_wm_ping=suppress_wm_ping, ) return int(process.wait()) finally: diagnostics.close( process_started=True, managed=True, window_discovered=game_window is not None, gamescope_scaling=gamescope_scaling, ) mapper.close() def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--window-title", default=WINDOW_TITLE) parser.add_argument("--borderless", action="store_true") parser.add_argument( "--allow-window-resizing", action="store_true", help="enable unsafe native client resizing for diagnostics", ) parser.add_argument("--gamescope-scaling", action="store_true") parser.add_argument("--suppress-wm-ping", action="store_true") parser.add_argument("command", nargs=argparse.REMAINDER) arguments = parser.parse_args(argv) command = arguments.command if command and command[0] == "--": command = command[1:] try: return _process_exit_status( run_game_window( command, arguments.window_title, borderless=arguments.borderless, allow_window_resizing=arguments.allow_window_resizing, gamescope_scaling=arguments.gamescope_scaling, suppress_wm_ping=arguments.suppress_wm_ping, ) ) except (GameboxRuntimeError, OSError) as exc: print(f"Error: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())