From c39f3bda47257736f25ec6e8aca333e086d3c7e8 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Thu, 30 Jul 2026 18:53:56 -0400 Subject: [PATCH] Add i3 focus tracking and harden X11 handoff --- src/cthulhu/compositor_state_adapter.py | 49 +++- src/cthulhu/compositor_state_i3.py | 259 ++++++++++++++++++ src/cthulhu/compositor_state_types.py | 3 + src/cthulhu/event_manager.py | 48 ++++ src/cthulhu/input_event_manager.py | 197 +++++++++++-- src/cthulhu/meson.build | 1 + .../plugins/WindowTitleReader/plugin.py | 156 +++++++++-- src/cthulhu/scripts/web/script.py | 14 - ...st_compositor_state_adapter_regressions.py | 60 +++- tests/test_compositor_state_i3_regressions.py | 181 ++++++++++++ ..._manager_compositor_context_regressions.py | 69 +++++ ...put_event_manager_x11_focus_regressions.py | 201 +++++++++++++- tests/test_runtime_state_snapshot.py | 20 +- tests/test_web_input_regressions.py | 6 +- .../test_window_title_fallback_regressions.py | 168 +++++++++++- tests/test_wine_access_controller_contract.py | 22 ++ wine-access/CMakeLists.txt | 1 + 17 files changed, 1379 insertions(+), 76 deletions(-) create mode 100644 src/cthulhu/compositor_state_i3.py create mode 100644 tests/test_compositor_state_i3_regressions.py diff --git a/src/cthulhu/compositor_state_adapter.py b/src/cthulhu/compositor_state_adapter.py index a00081c..8a77cfd 100644 --- a/src/cthulhu/compositor_state_adapter.py +++ b/src/cthulhu/compositor_state_adapter.py @@ -29,6 +29,7 @@ from .compositor_state_types import ( RESUME_ATSPI_CHURN, WORKSPACE_STATE_CHANGED, ) +from .compositor_state_i3 import I3WorkspaceBackend from .compositor_state_wayland import NullWorkspaceBackend, WaylandSharedProtocolsBackend from .wnck_support import get_session_type @@ -42,6 +43,7 @@ class CompositorStateAdapter: ) -> None: if workspace_backends is None: workspace_backends = [ + I3WorkspaceBackend(), WaylandSharedProtocolsBackend(), NullWorkspaceBackend(), ] @@ -70,13 +72,15 @@ class CompositorStateAdapter: if not self._backend_is_available(backend): continue self._workspaceBackend = backend + self._snapshot = DesktopContextSnapshot( + session_type=self.get_session_type(), + backend_name=self._backend_name(), + ) self._backend_activate(backend) break - self._snapshot = DesktopContextSnapshot( - session_type=self.get_session_type(), - backend_name=self._backend_name(), - ) + if self._workspaceBackend is None: + self._snapshot = DesktopContextSnapshot(session_type=self.get_session_type()) self.sync_accessible_context("activate") def deactivate(self) -> None: @@ -104,6 +108,9 @@ class CompositorStateAdapter: locus_of_focus_pid=locus_of_focus_snapshot["pid"], active_window_name=active_window_snapshot["name"], locus_of_focus_name=locus_of_focus_snapshot["name"], + focused_window_id=workspaceSnapshot.focused_window_id, + focused_window_title=workspaceSnapshot.focused_window_title, + focused_workspace_empty=workspaceSnapshot.focused_workspace_empty, ) previous_token = self._snapshot.active_window_token @@ -145,27 +152,49 @@ class CompositorStateAdapter: signal_name: str, workspace_ids: Optional[Iterable[str]] = None, reason: str = "", + payload: Optional[dict[str, Any]] = None, ) -> None: if workspace_ids is not None: + payload = payload or {} + focusedWorkspaceEmpty = payload.get( + "focused_workspace_empty", + self._snapshot.focused_workspace_empty, + ) + clearAccessibleContext = focusedWorkspaceEmpty is True snapshot = replace( self._snapshot, active_workspace_ids=frozenset(str(workspaceId) for workspaceId in workspace_ids if workspaceId), workspace_transition_pending=signal_name == DESKTOP_TRANSITION_STARTED, + active_window_token="" if clearAccessibleContext else self._snapshot.active_window_token, + locus_of_focus_token="" if clearAccessibleContext else self._snapshot.locus_of_focus_token, + active_window_pid=-1 if clearAccessibleContext else self._snapshot.active_window_pid, + locus_of_focus_pid=-1 if clearAccessibleContext else self._snapshot.locus_of_focus_pid, + active_window_name="" if clearAccessibleContext else self._snapshot.active_window_name, + locus_of_focus_name="" if clearAccessibleContext else self._snapshot.locus_of_focus_name, + focused_window_id=payload.get( + "focused_window_id", + self._snapshot.focused_window_id, + ), + focused_window_title=payload.get( + "focused_window_title", + self._snapshot.focused_window_title, + ), + focused_workspace_empty=focusedWorkspaceEmpty, ) self._snapshot = snapshot cthulhu_state.compositorSnapshot = snapshot if signal_name == DESKTOP_TRANSITION_STARTED: cthulhu_state.pauseAtspiChurn = True - self._emit(signal_name, reason, snapshot) - self._emit(PAUSE_ATSPI_CHURN, reason, snapshot) + self._emit(signal_name, reason, snapshot, payload) + self._emit(PAUSE_ATSPI_CHURN, reason, snapshot, payload) return if signal_name == DESKTOP_TRANSITION_FINISHED: cthulhu_state.pauseAtspiChurn = False - self._emit(signal_name, reason, snapshot) - self._emit(RESUME_ATSPI_CHURN, reason, snapshot) - self._emit(FLUSH_STALE_ATSPI_EVENTS, reason, snapshot) + self._emit(signal_name, reason, snapshot, payload) + self._emit(RESUME_ATSPI_CHURN, reason, snapshot, payload) + self._emit(FLUSH_STALE_ATSPI_EVENTS, reason, snapshot, payload) return if signal_name == PAUSE_ATSPI_CHURN: @@ -173,7 +202,7 @@ class CompositorStateAdapter: elif signal_name == RESUME_ATSPI_CHURN: cthulhu_state.pauseAtspiChurn = False - self._emit(signal_name, reason, snapshot) + self._emit(signal_name, reason, snapshot, payload) return normalized = (signal_name or "").strip().lower() diff --git a/src/cthulhu/compositor_state_i3.py b/src/cthulhu/compositor_state_i3.py new file mode 100644 index 0000000..8187bd8 --- /dev/null +++ b/src/cthulhu/compositor_state_i3.py @@ -0,0 +1,259 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2026 Stormux +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. + +"""Optional i3 IPC backend for authoritative X11 workspace and focus state.""" + +from __future__ import annotations + +import importlib +import threading +from collections.abc import Mapping +from typing import Any, Optional + +from gi.repository import GLib + +from . import debug +from .compositor_state_types import ( + DESKTOP_TRANSITION_FINISHED, + DESKTOP_TRANSITION_STARTED, + WORKSPACE_STATE_CHANGED, +) +from .wnck_support import get_session_type + + +class I3WorkspaceBackend: + """Tracks the focused i3 workspace and its real foreground X11 window.""" + + name = "i3-ipc" + + def __init__(self, *, i3ipc_module: Any = None) -> None: + self._i3ipc = i3ipc_module + self._connection: Any = None + self._emitSignal: Any = None + self._eventThread: Optional[threading.Thread] = None + self._active = False + self._lastContext: Optional[tuple[Any, ...]] = None + + def is_available(self, session_type: str | None = None) -> bool: + effectiveSessionType = (session_type or get_session_type()).strip().lower() + if effectiveSessionType != "x11": + return False + + if self._i3ipc is None: + try: + self._i3ipc = importlib.import_module("i3ipc") + except ImportError: + return False + + return self._ensure_connection() + + def activate(self, emit_signal: Any = None) -> None: + self.deactivate() + self._emitSignal = emit_signal + if emit_signal is None or not self.is_available(): + return + + self._active = True + self._subscribe() + self._refresh_context("activate", transition=False) + self._eventThread = threading.Thread( + target=self._run_event_loop, + name="cthulhu-i3-ipc", + daemon=True, + ) + self._eventThread.start() + + def deactivate(self, emit_signal: Any = None) -> None: + self._active = False + connection = self._connection + if connection is not None: + quitLoop = getattr(connection, "main_quit", None) + if callable(quitLoop): + try: + quitLoop() + except Exception: + pass + + commandSocket = getattr(connection, "_cmd_socket", None) + if commandSocket is not None: + try: + commandSocket.close() + except Exception: + pass + + self._connection = None + self._eventThread = None + self._emitSignal = None + self._lastContext = None + + def _ensure_connection(self) -> bool: + if self._connection is not None: + return True + if self._i3ipc is None: + return False + + connectionClass = getattr(self._i3ipc, "Connection", None) + if not callable(connectionClass): + return False + + try: + try: + self._connection = connectionClass(auto_reconnect=True) + except TypeError: + self._connection = connectionClass() + self._connection.get_workspaces() + except Exception as error: + self._connection = None + msg = f"COMPOSITOR STATE: i3 IPC unavailable: {error}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + return False + + return True + + def _subscribe(self) -> None: + if self._connection is None or self._i3ipc is None: + return + + onEvent = getattr(self._connection, "on", None) + eventType = getattr(self._i3ipc, "Event", None) + if not callable(onEvent) or eventType is None: + return + + onEvent(eventType.WORKSPACE, self._handle_ipc_event) + onEvent(eventType.WINDOW, self._handle_ipc_event) + + def _run_event_loop(self) -> None: + connection = self._connection + runLoop = getattr(connection, "main", None) + if not callable(runLoop): + return + + try: + runLoop() + except Exception as error: + if self._active: + msg = f"COMPOSITOR STATE: i3 IPC event loop stopped: {error}" + debug.printMessage(debug.LEVEL_WARNING, msg, True) + + def _handle_ipc_event(self, _connection: Any, event: Any) -> None: + if not self._active: + return + + change = str(getattr(event, "change", "") or "changed") + GLib.idle_add(self._refresh_context, f"i3-{change}", True) + + def _query_context(self) -> tuple[set[str], dict[str, Any]]: + if self._connection is None: + raise RuntimeError("i3 IPC is not connected") + + workspaceReplies = self._connection.get_workspaces() + focusedReply = next( + (workspace for workspace in workspaceReplies if bool(getattr(workspace, "focused", False))), + None, + ) + if focusedReply is None: + raise RuntimeError("i3 did not report a focused workspace") + + workspaceName = str(getattr(focusedReply, "name", "") or "").strip() + if not workspaceName: + raise RuntimeError("i3 focused workspace has no name") + + tree = self._connection.get_tree() + focusedWorkspace = next( + ( + workspace + for workspace in tree.workspaces() + if str(getattr(workspace, "name", "") or "") == workspaceName + ), + None, + ) + if focusedWorkspace is None: + raise RuntimeError(f"i3 tree is missing focused workspace {workspaceName}") + + windowNodes = [ + node + for node in focusedWorkspace + if self._positive_window_id(getattr(node, "window", None)) is not None + ] + focusedNode = tree.find_focused() + focusedWindowId: Optional[int] = None + focusedWindowTitle = "" + if focusedNode is not None and focusedNode.workspace() is focusedWorkspace: + focusedWindowId = self._positive_window_id(getattr(focusedNode, "window", None)) + if focusedWindowId is not None: + focusedWindowTitle = str( + getattr(focusedNode, "window_title", None) + or getattr(focusedNode, "name", None) + or "" + ).strip() + + payload = { + "focused_window_id": focusedWindowId, + "focused_window_title": focusedWindowTitle, + "focused_workspace_empty": not windowNodes, + } + return {workspaceName}, payload + + @staticmethod + def _positive_window_id(value: Any) -> Optional[int]: + try: + windowId = int(value) + except (TypeError, ValueError): + return None + return windowId if windowId > 0 else None + + def _refresh_context(self, reason: str, transition: bool = True) -> bool: + if not self._active: + return False + + try: + workspaceIds, payload = self._query_context() + except Exception as error: + msg = f"COMPOSITOR STATE: Could not query i3 context: {error}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + unknownPayload = { + "focused_window_id": None, + "focused_window_title": "", + "focused_workspace_empty": None, + } + self._lastContext = None + self._emit_signal( + WORKSPACE_STATE_CHANGED, + set(), + f"{reason}: i3-context-unknown", + unknownPayload, + ) + return False + + context = ( + frozenset(workspaceIds), + payload["focused_window_id"], + payload["focused_window_title"], + payload["focused_workspace_empty"], + ) + if context == self._lastContext: + return False + self._lastContext = context + + if transition: + self._emit_signal(DESKTOP_TRANSITION_STARTED, workspaceIds, reason, payload) + self._emit_signal(WORKSPACE_STATE_CHANGED, workspaceIds, reason, payload) + if transition: + self._emit_signal(DESKTOP_TRANSITION_FINISHED, workspaceIds, reason, payload) + return False + + def _emit_signal( + self, + signalType: str, + workspaceIds: set[str], + reason: str, + payload: Mapping[str, Any], + ) -> None: + if callable(self._emitSignal): + self._emitSignal(signalType, workspaceIds, reason, dict(payload)) diff --git a/src/cthulhu/compositor_state_types.py b/src/cthulhu/compositor_state_types.py index ca54ce6..5880b6d 100644 --- a/src/cthulhu/compositor_state_types.py +++ b/src/cthulhu/compositor_state_types.py @@ -38,6 +38,9 @@ class DesktopContextSnapshot: locus_of_focus_pid: int = -1 active_window_name: str = "" locus_of_focus_name: str = "" + focused_window_id: Optional[int] = None + focused_window_title: str = "" + focused_workspace_empty: Optional[bool] = None @dataclass(frozen=True, slots=True) diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 5903865..332696f 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -60,6 +60,7 @@ from .compositor_state_types import ( PAUSE_ATSPI_CHURN, PRIORITIZE_FOCUS, RESUME_ATSPI_CHURN, + WORKSPACE_STATE_CHANGED, ) if TYPE_CHECKING: @@ -113,6 +114,7 @@ class EventManager: self._compositorStateAdapter: Optional[Any] = None self._churnSuppressed: bool = cthulhu_state.pauseAtspiChurn self._prioritizedContextToken: Optional[str] = cthulhu_state.prioritizedDesktopContextToken + self._desktopContextConfirmedEmpty: bool = False self._relevanceBurstWindow: float = 0.15 self._relevanceBurstHistory: Dict[Tuple[str, str, str], float] = {} @@ -292,6 +294,9 @@ class EventManager: if not contextToken: contextToken = snapshotToken or None + if signalType in (WORKSPACE_STATE_CHANGED, DESKTOP_TRANSITION_FINISHED): + self._update_i3_workspace_context(snapshot) + if signalType in (DESKTOP_FOCUS_CONTEXT_CHANGED, PRIORITIZE_FOCUS): self._prioritizedContextToken = contextToken cthulhu_state.prioritizedDesktopContextToken = contextToken @@ -318,6 +323,43 @@ class EventManager: if signalType == FLUSH_STALE_ATSPI_EVENTS: self._flush_stale_atspi_events() + def _update_i3_workspace_context(self, snapshot: Any) -> None: + """Clears stale accessible state when i3 confirms a blank workspace.""" + + if snapshot is None or snapshot.backend_name != "i3-ipc": + return + + workspaceEmpty = snapshot.focused_workspace_empty + if workspaceEmpty is None: + return + + if not workspaceEmpty: + if self._desktopContextConfirmedEmpty: + self._desktopContextConfirmedEmpty = False + GLib.idle_add(self._sync_focus_on_startup) + return + + if self._desktopContextConfirmedEmpty: + return + + self._desktopContextConfirmedEmpty = True + activeScript = self.app.scriptManager.get_active_script() + if activeScript is not None: + activeScript.presentationInterrupt() + + manager = focus_manager.get_manager() + if manager is not None: + manager.clear_state("confirmed empty i3 workspace") + cthulhu_state.pendingSelfHostedFocus = None + self._prioritizedContextToken = None + cthulhu_state.prioritizedDesktopContextToken = None + + defaultScript = self.app.scriptManager.get_default_script() + self.app.scriptManager.set_active_script( + defaultScript, + "focus: confirmed-empty-i3-workspace", + ) + def ignoreEventTypes(self, eventTypeList: List[str]) -> None: for eventType in eventTypeList: if eventType not in self._ignoredEvents: @@ -1771,6 +1813,12 @@ class EventManager: debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True) eType = event.type + if self._desktopContextConfirmedEmpty \ + and not self._is_notification_event_during_xterm_handoff(event): + msg = "EVENT MANAGER: Ignoring AT-SPI event on confirmed empty i3 workspace." + debug.printMessage(debug.LEVEL_INFO, msg, True) + return + refreshXtermHandoff = self._event_refreshes_xterm_handoff(event) isNotification = self._is_notification_event_during_xterm_handoff(event) suppressForXterm = not isNotification and self._inputEventManager is not None and ( diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index 3c7bebf..4059ad3 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -125,12 +125,15 @@ class InputEventManager: self._netActiveWindowAtom = None self._x11AnyPropertyType = None self._didAttemptX11Display: bool = False + self._lastEwmhReadWindowId: Optional[int] = None self._lastEwmhActiveWindowId: Optional[int] = None self._lastEwmhXtermMatch: Optional[bool] = None self._scriptWithSuspendedGrabsForXterm = None self._xtermFocusScreen = None self._xtermFocusHandlerId: int = 0 self._xtermRecoverySourceId: int = 0 + self._xtermRecoveryPollCount: int = 0 + self._xtermHandoffHistory: List[str] = [] def activate_device(self) -> Atspi.Device: """Creates and returns the AT-SPI device used by this manager.""" @@ -229,6 +232,14 @@ class InputEventManager: "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None, "xterm_focus_monitor_active": self._xtermFocusHandlerId != 0, "xterm_recovery_timer_active": self._xtermRecoverySourceId != 0, + "xterm_recovery_source_present": self._xterm_recovery_source_is_present(), + "xterm_recovery_poll_count": self._xtermRecoveryPollCount, + "xterm_handoff_history": " | ".join(self._xtermHandoffHistory), + "last_ewmh_read_window_id": ( + hex(self._lastEwmhReadWindowId) + if self._lastEwmhReadWindowId is not None + else "" + ), "last_ewmh_active_window_id": ( hex(self._lastEwmhActiveWindowId) if self._lastEwmhActiveWindowId is not None @@ -578,22 +589,7 @@ class InputEventManager: def _get_ewmh_active_x11_window_id(self) -> Optional[int]: """Returns _NET_ACTIVE_WINDOW from the X11 root window when available.""" - if not self._didAttemptX11Display: - self._didAttemptX11Display = True - try: - from Xlib import X # pylint: disable=import-outside-toplevel - from Xlib import display as xDisplay # pylint: disable=import-outside-toplevel - - self._x11Display = xDisplay.Display() - self._x11Root = self._x11Display.screen().root - self._netActiveWindowAtom = self._x11Display.intern_atom("_NET_ACTIVE_WINDOW") - self._x11AnyPropertyType = X.AnyPropertyType - except Exception as error: - msg = f"INPUT EVENT MANAGER: EWMH active-window lookup unavailable: {error}" - debug.print_message(debug.LEVEL_INFO, msg, True) - self._close_x11_display() - - if self._x11Root is None or self._netActiveWindowAtom is None: + if not self._ensure_x11_display(): return None try: @@ -612,6 +608,33 @@ class InputEventManager: self._didAttemptX11Display = False return None + def _ensure_x11_display(self) -> bool: + """Initializes the optional exact-X11 metadata connection.""" + + if self._x11Display is not None: + return True + + if not self._didAttemptX11Display: + self._didAttemptX11Display = True + try: + from Xlib import X # pylint: disable=import-outside-toplevel + from Xlib import display as xDisplay # pylint: disable=import-outside-toplevel + + self._x11Display = xDisplay.Display() + self._x11Root = self._x11Display.screen().root + self._netActiveWindowAtom = self._x11Display.intern_atom("_NET_ACTIVE_WINDOW") + self._x11AnyPropertyType = X.AnyPropertyType + except Exception as error: + msg = f"INPUT EVENT MANAGER: EWMH active-window lookup unavailable: {error}" + debug.print_message(debug.LEVEL_INFO, msg, True) + self._close_x11_display() + + return ( + self._x11Display is not None + and self._x11Root is not None + and self._netActiveWindowAtom is not None + ) + def _close_x11_display(self) -> None: """Closes the optional X11 display used for EWMH focus lookup.""" @@ -620,6 +643,7 @@ class InputEventManager: self._x11Root = None self._netActiveWindowAtom = None self._x11AnyPropertyType = None + self._lastEwmhReadWindowId = None self._lastEwmhActiveWindowId = None self._lastEwmhXtermMatch = None if x11Display is None: @@ -764,9 +788,22 @@ class InputEventManager: self._poll_xterm_grab_recovery, ) + def _xterm_recovery_source_is_present(self) -> bool: + """Returns whether GLib still owns the recorded recovery source.""" + + if not self._xtermRecoverySourceId: + return False + + try: + context = GLib.MainContext.default() + return context.find_source_by_id(self._xtermRecoverySourceId) is not None + except Exception: + return False + def _poll_xterm_grab_recovery(self) -> bool: """Updates XTerm handoff state from the authoritative EWMH active window.""" + self._xtermRecoveryPollCount += 1 if self._device is None: self._xtermRecoverySourceId = 0 return False @@ -782,6 +819,15 @@ class InputEventManager: debug.print_message(debug.LEVEL_INFO, msg, True) return True + def _record_xterm_handoff_action(self, action: str) -> None: + """Retains a short history of XTerm focus and grab-state transitions.""" + + if self._xtermHandoffHistory and self._xtermHandoffHistory[-1] == action: + return + + self._xtermHandoffHistory.append(action) + self._xtermHandoffHistory = self._xtermHandoffHistory[-12:] + def _on_active_x11_window_changed(self, screen: Any, _previousWindow: Any) -> None: """Updates grab suspension as soon as X11 focus changes.""" @@ -799,7 +845,24 @@ class InputEventManager: def _active_x11_window_xterm_match(self) -> Optional[bool]: """Returns whether the active X11 window is XTerm, or None when unknown.""" - windowId = self._get_ewmh_active_x11_window_id() + snapshot = cthulhu_state.compositorSnapshot + usingI3Context = snapshot is not None and snapshot.backend_name == "i3-ipc" + if usingI3Context and snapshot.focused_workspace_empty is True: + self._record_xterm_handoff_action("i3:confirmed-empty-workspace") + self._lastEwmhReadWindowId = None + return False + + if usingI3Context and snapshot.focused_window_id is not None: + windowId = snapshot.focused_window_id + else: + windowId = self._get_ewmh_active_x11_window_id() + + previousWindowId = self._lastEwmhReadWindowId + if windowId != previousWindowId: + previousText = hex(previousWindowId) if previousWindowId is not None else "unknown" + currentText = hex(windowId) if windowId is not None else "unknown" + self._record_xterm_handoff_action(f"ewmh:{previousText}->{currentText}") + self._lastEwmhReadWindowId = windowId if ( windowId is not None and windowId == self._lastEwmhActiveWindowId @@ -807,8 +870,10 @@ class InputEventManager: ): return self._lastEwmhXtermMatch - window = self._get_active_x11_window(ewmhWindowId=windowId) - match = self._x11_window_xterm_match(window) + if windowId is None: + return None + + match = self._x11_window_id_xterm_match(windowId) if windowId is not None and match is not None: self._lastEwmhActiveWindowId = windowId self._lastEwmhXtermMatch = match @@ -870,6 +935,80 @@ class InputEventManager: return self._identifier_is_xterm(executable) + def _get_x11_window_metadata( + self, + windowId: int, + ) -> Optional[Tuple[List[str], int]]: + """Returns identifiers and pid directly from the exact X11 window id.""" + + if not self._ensure_x11_display(): + return None + + try: + window = self._x11Display.create_resource_object("window", windowId) + identifiers = [ + value + for value in (window.get_wm_class() or ()) + if isinstance(value, str) and value.strip() + ] + + name = window.get_wm_name() + if isinstance(name, str) and name.strip(): + identifiers.append(name) + + netWmNameAtom = self._x11Display.intern_atom("_NET_WM_NAME") + utf8StringAtom = self._x11Display.intern_atom("UTF8_STRING") + nameProperty = window.get_full_property(netWmNameAtom, utf8StringAtom) + if nameProperty is not None and nameProperty.value: + value = nameProperty.value + if isinstance(value, bytes): + value = value.decode(errors="ignore") + if isinstance(value, str) and value.strip(): + identifiers.append(value) + + netWmPidAtom = self._x11Display.intern_atom("_NET_WM_PID") + pidProperty = window.get_full_property( + netWmPidAtom, + self._x11AnyPropertyType, + ) + pid = -1 + if pidProperty is not None and pidProperty.value: + pid = int(pidProperty.value[0]) + except Exception as error: + msg = ( + "INPUT EVENT MANAGER: Could not read metadata for X11 window " + f"{hex(windowId)}: {error}" + ) + debug.print_message(debug.LEVEL_INFO, msg, True) + return None + + return identifiers, pid + + def _x11_window_id_xterm_match(self, windowId: int) -> Optional[bool]: + """Returns whether the exact EWMH X11 window id identifies XTerm.""" + + metadata = self._get_x11_window_metadata(windowId) + if metadata is None: + return None + + identifiers, pid = metadata + if any(self._identifier_is_xterm(value) for value in identifiers): + return True + + if pid > 0: + try: + with open(f"/proc/{pid}/cmdline", "rb") as cmdlineFile: + executable = cmdlineFile.read().split(b"\0", 1)[0].decode(errors="ignore") + if self._identifier_is_xterm(executable): + return True + except OSError: + pass + + if identifiers or pid > 0: + return False + + return None + def _active_x11_window_is_xterm(self) -> bool: """Returns True when the active X11 window appears to be XTerm.""" @@ -1043,7 +1182,17 @@ class InputEventManager: """Suspends active-script key grabs so XTerm/Fenrir can receive them.""" script = script_manager.get_manager().get_active_script() - if script is None or script is self._scriptWithSuspendedGrabsForXterm: + if script is None: + self._record_xterm_handoff_action("suspend:skipped-no-active-script") + return + + if script is self._scriptWithSuspendedGrabsForXterm: + grabIds = getattr(script, "grab_ids", []) + scriptGrabCount = len(grabIds) if isinstance(grabIds, list) else -1 + self._record_xterm_handoff_action( + f"suspend:already-marked:script-grabs={scriptGrabCount}:" + f"all-grabs={len(self._grabbed_bindings)}" + ) return if self._scriptWithSuspendedGrabsForXterm is not None: @@ -1060,6 +1209,9 @@ class InputEventManager: debug.print_tokens(debug.LEVEL_INFO, tokens, True) removeGrabs() self._scriptWithSuspendedGrabsForXterm = script + self._record_xterm_handoff_action( + f"suspend:success:all-grabs={len(self._grabbed_bindings)}" + ) self._ensure_xterm_recovery_timer() def _restore_key_grabs_after_xterm(self) -> None: @@ -1071,6 +1223,7 @@ class InputEventManager: activeScript = script_manager.get_manager().get_active_script() if script is not activeScript: + self._record_xterm_handoff_action("restore:dropped-active-script-changed") tokens = [ "INPUT EVENT MANAGER: Not restoring XTerm-suspended key grabs; active script changed:", script, @@ -1092,6 +1245,7 @@ class InputEventManager: try: addGrabs() except Exception as error: + self._record_xterm_handoff_action(f"restore:failed:{type(error).__name__}") msg = f"INPUT EVENT MANAGER: Could not restore XTerm-suspended key grabs: {error}" debug.print_message(debug.LEVEL_INFO, msg, True) removeGrabs = getattr(script, "removeKeyGrabs", None) @@ -1110,6 +1264,9 @@ class InputEventManager: return self._scriptWithSuspendedGrabsForXterm = None + self._record_xterm_handoff_action( + f"restore:success:all-grabs={len(self._grabbed_bindings)}" + ) # pylint: disable=too-many-arguments # pylint: disable=too-many-positional-arguments diff --git a/src/cthulhu/meson.build b/src/cthulhu/meson.build index f822e7f..3c9a005 100644 --- a/src/cthulhu/meson.build +++ b/src/cthulhu/meson.build @@ -34,6 +34,7 @@ cthulhu_python_sources = files([ 'cmdnames.py', 'colornames.py', 'compositor_state_adapter.py', + 'compositor_state_i3.py', 'compositor_state_types.py', 'compositor_state_wayland.py', 'common_keyboardmap.py', diff --git a/src/cthulhu/plugins/WindowTitleReader/plugin.py b/src/cthulhu/plugins/WindowTitleReader/plugin.py index dee1439..b1c55ff 100644 --- a/src/cthulhu/plugins/WindowTitleReader/plugin.py +++ b/src/cthulhu/plugins/WindowTitleReader/plugin.py @@ -80,7 +80,7 @@ class WindowTitleReader(Plugin): self, overwrite=True, ) - if xlibAvailable: + if xlibAvailable or self._get_i3_title_context() is not None: self._start_tracking() self._activated = True debug.printMessage(debug.LEVEL_INFO, "WindowTitleReader: Activated", True) @@ -136,7 +136,7 @@ class WindowTitleReader(Plugin): self._present_message("Window title reader off") return True - if not xlibAvailable: + if not xlibAvailable and self._get_i3_title_context() is None: self._present_message("Window title reader unavailable") debug.printMessage( debug.LEVEL_INFO, @@ -167,7 +167,7 @@ class WindowTitleReader(Plugin): self._present_message("Window title reader on") return True - if not xlibAvailable: + if not xlibAvailable and self._get_i3_title_context() is None: if notify_user: self._present_message("Window title reader unavailable") debug.printMessage( @@ -218,9 +218,9 @@ class WindowTitleReader(Plugin): return True try: - self._display = xDisplay.Display() - self._root = self._display.screen().root - self._init_atoms() + if self._get_i3_title_context() is None: + if not self._ensure_x11_display(): + return False self._pollSourceId = GLib.timeout_add(self._pollIntervalMs, self._poll_window_title) self._poll_window_title() return True @@ -273,17 +273,24 @@ class WindowTitleReader(Plugin): return False try: - activeWindow = self._get_active_window() - if not activeWindow: + i3Context = self._get_i3_title_context() + if i3Context is not None: + activeWindowId, windowTitle = i3Context + else: + if not self._ensure_x11_display(): + self._lastTitle = None + return True + activeWindow = self._get_active_window() + if not activeWindow: + self._lastTitle = None + return True + activeWindowId = activeWindow.id + windowTitle = self._get_current_title(activeWindow) + + if activeWindowId is None or not windowTitle: self._lastTitle = None return True - windowTitle = self._get_current_title(activeWindow) - if not windowTitle: - self._lastTitle = None - return True - - activeWindowId = activeWindow.id if self._lastActiveWindowId is None: self._lastActiveWindowId = activeWindowId elif activeWindowId != self._lastActiveWindowId: @@ -321,12 +328,25 @@ class WindowTitleReader(Plugin): def get_fallback_title(self, atspiTitle=None): """Returns the X11 title when AT-SPI has not exposed an equivalent title.""" + i3Context = self._get_i3_title_context() + if i3Context is not None: + _windowId, titleText = i3Context + if not titleText: + return "" + if atspiTitle is None: + atspiTitle = self._get_atspi_title() + if self._titles_match(atspiTitle, titleText): + return "" + return titleText + if not xlibAvailable and self._display is None: return "" startedTracking = self._pollSourceId is not None if not startedTracking and not self._start_tracking(): return "" + if not self._ensure_x11_display(): + return "" activeWindow = self._get_active_window() titleText = self._get_current_title(activeWindow) if activeWindow else "" @@ -341,6 +361,65 @@ class WindowTitleReader(Plugin): return titleText + def _get_i3_title_context(self): + """Returns authoritative i3 window id/title, or None when i3 is not active.""" + + if not self.app: + return None + + try: + adapter = self.app.getCompositorStateAdapter() + snapshot = adapter.get_snapshot() if adapter else None + except Exception: + return None + + if snapshot is None or getattr(snapshot, "backend_name", "") != "i3-ipc": + return None + + if getattr(snapshot, "focused_workspace_empty", None) is True: + return (None, "") + + windowId = getattr(snapshot, "focused_window_id", None) + titleText = str(getattr(snapshot, "focused_window_title", "") or "").strip() + if windowId is None: + return None + + if not titleText or self._is_wine_desktop_title(titleText): + if not self._ensure_x11_display(): + return None + try: + activeWindow = self._display.create_resource_object("window", windowId) + except Exception: + return None + titleText = self._get_current_title(activeWindow) + if not titleText: + return None + + return (windowId, titleText) + + def _ensure_x11_display(self): + """Lazily opens the generic X11 title source when i3 IPC is unavailable.""" + + if self._display is not None: + return True + if not xlibAvailable: + return False + + try: + self._display = xDisplay.Display() + self._root = self._display.screen().root + self._init_atoms() + except Exception as error: + debug.printMessage( + debug.LEVEL_INFO, + f"WindowTitleReader: Failed to open X11 fallback: {error}", + True, + ) + self._cleanup_display() + return False + + return True + def _get_atspi_title(self): if not self.app: return "" @@ -386,9 +465,11 @@ class WindowTitleReader(Plugin): def _get_wine_desktop_title(self, desktopWindow): focusWindow = self._get_focus_window() - if focusWindow and focusWindow.id != desktopWindow.id: + if focusWindow \ + and focusWindow.id != desktopWindow.id \ + and self._window_is_descendant_of(focusWindow, desktopWindow): focusTitle = self._get_window_title(focusWindow) - if focusTitle and not self._is_wine_desktop_title(focusTitle): + if self._is_usable_wine_child_title(focusWindow, focusTitle): return focusTitle return self._find_child_title(desktopWindow) @@ -404,7 +485,7 @@ class WindowTitleReader(Plugin): window = childQueue.pop(0) checkedCount += 1 titleText = self._get_window_title(window) - if titleText and not self._is_wine_desktop_title(titleText): + if self._is_usable_wine_child_title(window, titleText): return titleText try: @@ -414,6 +495,47 @@ class WindowTitleReader(Plugin): return "" + def _window_is_descendant_of(self, window, ancestor): + if not window or not ancestor: + return False + + current = window + for _hop in range(64): + if current.id == ancestor.id: + return True + try: + parent = current.query_tree().parent + except xError.XError: + return False + if not parent or parent.id == current.id: + return False + current = parent + + return False + + def _is_usable_wine_child_title(self, window, titleText): + if not titleText or self._is_wine_desktop_title(titleText): + return False + + normalizedTitle = " ".join(titleText.casefold().split()) + if normalizedTitle in ("default ime", "program manager"): + return False + + try: + windowClass = window.get_wm_class() or () + except xError.XError: + windowClass = () + normalizedClass = { + str(value or "").strip().casefold() + for value in windowClass + } + if "cthulhu-wine-access.exe" in normalizedClass: + return False + if normalizedTitle == "nvda" and "nvda.exe" in normalizedClass: + return False + + return True + def _is_wine_desktop_title(self, titleText): return self._wineDesktopLabel.lower() in titleText.lower() diff --git a/src/cthulhu/scripts/web/script.py b/src/cthulhu/scripts/web/script.py index db73dec..1c03126 100644 --- a/src/cthulhu/scripts/web/script.py +++ b/src/cthulhu/scripts/web/script.py @@ -2660,20 +2660,6 @@ class Script(default.Script): self.utilities.setCaretContext(event.source, 0) obj, offset = event.source, 0 - if self._lastCommandWasCaretNav \ - and obj is not event.source \ - and AXUtilities.is_focusable(event.source) \ - and AXUtilities.is_focused(event.source) \ - and self.utilities.inDocumentContent(event.source) \ - and self.utilities.isLink(event.source) \ - and obj \ - and AXUtilities.is_section(obj): - msg = "WEB: Event handled: Replacing stale section context with focused link" - debug.printMessage(debug.LEVEL_INFO, msg, True) - cthulhu.setLocusOfFocus(event, event.source, False) - self.utilities.setCaretContext(event.source, 0) - return True - if self._lastCommandWasCaretNav: msg = "WEB: Event ignored: Last command was caret nav" debug.printMessage(debug.LEVEL_INFO, msg, True) diff --git a/tests/test_compositor_state_adapter_regressions.py b/tests/test_compositor_state_adapter_regressions.py index fda089c..45f8917 100644 --- a/tests/test_compositor_state_adapter_regressions.py +++ b/tests/test_compositor_state_adapter_regressions.py @@ -11,6 +11,7 @@ import cthulhu as cthulhu_package from cthulhu import cthulhu_state from cthulhu import compositor_state_adapter +from cthulhu import compositor_state_i3 from cthulhu import compositor_state_types from cthulhu import compositor_state_wayland @@ -74,7 +75,7 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase): self.assertTrue(callable(selectedBackend.activate_calls[0])) self.assertEqual(adapter.get_snapshot().backend_name, "selected") - def test_default_workspace_backends_include_wayland_then_null_backend(self) -> None: + def test_default_workspace_backends_include_i3_wayland_then_null_backend(self) -> None: adapter = compositor_state_adapter.CompositorStateAdapter() backend_types = [type(backend) for backend in adapter._workspaceBackends] @@ -82,11 +83,68 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase): self.assertEqual( backend_types, [ + compositor_state_i3.I3WorkspaceBackend, compositor_state_wayland.WaylandSharedProtocolsBackend, compositor_state_wayland.NullWorkspaceBackend, ], ) + def test_workspace_payload_records_i3_foreground_context(self) -> None: + backend = FakeWorkspaceBackend(True, "i3-ipc") + adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[backend]) + + adapter.activate() + backend.activate_calls[-1]( + compositor_state_types.WORKSPACE_STATE_CHANGED, + {"8"}, + "i3-window-focus", + { + "focused_window_id": 0x200000C, + "focused_window_title": "9 devel", + "focused_workspace_empty": False, + }, + ) + + snapshot = adapter.get_snapshot() + self.assertEqual(snapshot.active_workspace_ids, frozenset({"8"})) + self.assertEqual(snapshot.focused_window_id, 0x200000C) + self.assertEqual(snapshot.focused_window_title, "9 devel") + self.assertFalse(snapshot.focused_workspace_empty) + + def test_confirmed_empty_workspace_clears_stale_accessible_context(self) -> None: + backend = FakeWorkspaceBackend(True, "i3-ipc") + adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[backend]) + adapter.activate() + adapter._snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + active_window_token="42:Browser", + locus_of_focus_token="42:Web content", + active_window_pid=42, + locus_of_focus_pid=42, + active_window_name="Browser", + locus_of_focus_name="Web content", + ) + + backend.activate_calls[-1]( + compositor_state_types.WORKSPACE_STATE_CHANGED, + {"9"}, + "i3-workspace-focus", + { + "focused_window_id": None, + "focused_window_title": "", + "focused_workspace_empty": True, + }, + ) + + snapshot = adapter.get_snapshot() + self.assertEqual(snapshot.active_window_token, "") + self.assertEqual(snapshot.locus_of_focus_token, "") + self.assertEqual(snapshot.active_window_pid, -1) + self.assertEqual(snapshot.locus_of_focus_pid, -1) + self.assertEqual(snapshot.active_window_name, "") + self.assertEqual(snapshot.locus_of_focus_name, "") + def test_wayland_backend_is_unavailable_without_wayland_session(self) -> None: backend = compositor_state_wayland.WaylandSharedProtocolsBackend() diff --git a/tests/test_compositor_state_i3_regressions.py b/tests/test_compositor_state_i3_regressions.py new file mode 100644 index 0000000..69ffbdd --- /dev/null +++ b/tests/test_compositor_state_i3_regressions.py @@ -0,0 +1,181 @@ +import sys +import types +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cthulhu import compositor_state_i3 +from cthulhu import compositor_state_types + + +class FakeNode: + def __init__( + self, + *, + node_type="con", + name="", + window=None, + window_title=None, + focused=False, + children=None, + ): + self.type = node_type + self.name = name + self.window = window + self.window_title = window_title + self.focused = focused + self.nodes = list(children or []) + self.floating_nodes = [] + self.parent = None + for child in self.nodes: + child.parent = self + + def __iter__(self): + pending = list(self.nodes) + list(self.floating_nodes) + while pending: + node = pending.pop(0) + yield node + pending.extend(node.nodes) + pending.extend(node.floating_nodes) + + def workspaces(self): + return [node for node in self if node.type == "workspace"] + + def find_focused(self): + if self.focused: + return self + return next((node for node in self if node.focused), None) + + def workspace(self): + node = self + while node is not None and node.type != "workspace": + node = node.parent + return node + + +class FakeWorkspaceReply: + def __init__(self, name, focused): + self.name = name + self.focused = focused + + +class FakeConnection: + def __init__(self, tree, workspaces): + self.tree = tree + self.workspaceReplies = workspaces + + def get_tree(self): + return self.tree + + def get_workspaces(self): + return self.workspaceReplies + + +class I3WorkspaceBackendRegressionTests(unittest.TestCase): + def test_query_reports_focused_window_id_title_and_nonempty_workspace(self): + focusedWindow = FakeNode( + name="AT-SPI misses this title", + window=0x200000C, + window_title="Accurate i3 title", + focused=True, + ) + workspace = FakeNode( + node_type="workspace", + name="8", + children=[focusedWindow], + ) + root = FakeNode(node_type="root", children=[workspace]) + connection = FakeConnection(root, [FakeWorkspaceReply("8", True)]) + backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace()) + backend._connection = connection + + workspaceIds, payload = backend._query_context() + + self.assertEqual(workspaceIds, {"8"}) + self.assertEqual(payload["focused_window_id"], 0x200000C) + self.assertEqual(payload["focused_window_title"], "Accurate i3 title") + self.assertFalse(payload["focused_workspace_empty"]) + + def test_query_reports_confirmed_empty_focused_workspace(self): + workspace = FakeNode( + node_type="workspace", + name="9", + focused=True, + ) + root = FakeNode(node_type="root", children=[workspace]) + connection = FakeConnection(root, [FakeWorkspaceReply("9", True)]) + backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace()) + backend._connection = connection + + workspaceIds, payload = backend._query_context() + + self.assertEqual(workspaceIds, {"9"}) + self.assertIsNone(payload["focused_window_id"]) + self.assertEqual(payload["focused_window_title"], "") + self.assertTrue(payload["focused_workspace_empty"]) + + def test_refresh_emits_started_state_and_finished_with_foreground_payload(self): + backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace()) + backend._active = True + backend._emitSignal = mock.Mock() + payload = { + "focused_window_id": None, + "focused_window_title": "", + "focused_workspace_empty": True, + } + + with mock.patch.object(backend, "_query_context", return_value=({"9"}, payload)): + self.assertFalse(backend._refresh_context("workspace-focus", transition=True)) + + self.assertEqual( + backend._emitSignal.call_args_list, + [ + mock.call( + compositor_state_types.DESKTOP_TRANSITION_STARTED, + {"9"}, + "workspace-focus", + payload, + ), + mock.call( + compositor_state_types.WORKSPACE_STATE_CHANGED, + {"9"}, + "workspace-focus", + payload, + ), + mock.call( + compositor_state_types.DESKTOP_TRANSITION_FINISHED, + {"9"}, + "workspace-focus", + payload, + ), + ], + ) + + def test_query_failure_emits_unknown_instead_of_preserving_stale_i3_context(self): + backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace()) + backend._active = True + backend._emitSignal = mock.Mock() + backend._lastContext = (frozenset({"8"}), 0x200000C, "Browser", False) + + with ( + mock.patch.object(backend, "_query_context", side_effect=OSError("IPC disconnected")), + mock.patch.object(compositor_state_i3.debug, "printMessage"), + ): + self.assertFalse(backend._refresh_context("window-focus", transition=True)) + + backend._emitSignal.assert_called_once_with( + compositor_state_types.WORKSPACE_STATE_CHANGED, + set(), + "window-focus: i3-context-unknown", + { + "focused_window_id": None, + "focused_window_title": "", + "focused_workspace_empty": None, + }, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_event_manager_compositor_context_regressions.py b/tests/test_event_manager_compositor_context_regressions.py index 235319b..587f69b 100644 --- a/tests/test_event_manager_compositor_context_regressions.py +++ b/tests/test_event_manager_compositor_context_regressions.py @@ -91,6 +91,75 @@ class EventManagerCompositorContextRegressionTests(unittest.TestCase): self.assertFalse(cthulhu_state.pauseAtspiChurn) self.assertFalse(self.manager._churnSuppressed) + def test_confirmed_empty_i3_workspace_clears_stale_web_context(self) -> None: + staleScript = mock.Mock() + defaultScript = mock.Mock() + focusManager = mock.Mock() + self.manager.app.scriptManager.get_active_script.return_value = staleScript + self.manager.app.scriptManager.get_default_script.return_value = defaultScript + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + active_workspace_ids=frozenset({"9"}), + focused_workspace_empty=True, + ) + + with mock.patch.object( + event_manager.focus_manager, + "get_manager", + return_value=focusManager, + ): + self.manager._handle_compositor_signal( + compositor_state_types.CompositorStateEvent( + compositor_state_types.DESKTOP_TRANSITION_FINISHED, + reason="workspace-focus", + snapshot=snapshot, + ) + ) + + staleScript.presentationInterrupt.assert_called_once_with() + focusManager.clear_state.assert_called_once_with("confirmed empty i3 workspace") + self.manager.app.scriptManager.set_active_script.assert_called_once_with( + defaultScript, + "focus: confirmed-empty-i3-workspace", + ) + self.assertTrue(self.manager._desktopContextConfirmedEmpty) + + def test_atspi_event_is_ignored_while_i3_workspace_is_confirmed_empty(self) -> None: + self.manager._desktopContextConfirmedEmpty = True + event = FakeEvent("object:state-changed:focused", source=object(), detail1=1) + self.manager._get_scriptForEvent = mock.Mock() + + with ( + mock.patch.object(event_manager.debug, "printObjectEvent"), + mock.patch.object(event_manager.debug, "printMessage"), + ): + self.manager._processObjectEvent(event) + + self.manager._get_scriptForEvent.assert_not_called() + + def test_nonempty_i3_workspace_schedules_focus_resync_after_empty_workspace(self) -> None: + self.manager._desktopContextConfirmedEmpty = True + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + active_workspace_ids=frozenset({"8"}), + focused_window_id=0x200000C, + focused_workspace_empty=False, + ) + + with mock.patch.object(event_manager.GLib, "idle_add") as idleAdd: + self.manager._handle_compositor_signal( + compositor_state_types.CompositorStateEvent( + compositor_state_types.WORKSPACE_STATE_CHANGED, + reason="window-focus", + snapshot=snapshot, + ) + ) + + self.assertFalse(self.manager._desktopContextConfirmedEmpty) + idleAdd.assert_called_once_with(self.manager._sync_focus_on_startup) + def test_stale_context_event_is_obsolete_while_churn_is_paused(self) -> None: self.manager._churnSuppressed = True self.manager._prioritizedContextToken = "current" diff --git a/tests/test_input_event_manager_x11_focus_regressions.py b/tests/test_input_event_manager_x11_focus_regressions.py index cdbb955..cd4677f 100644 --- a/tests/test_input_event_manager_x11_focus_regressions.py +++ b/tests/test_input_event_manager_x11_focus_regressions.py @@ -6,9 +6,20 @@ from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cthulhu import input_event_manager +from cthulhu import compositor_state_types class InputEventManagerX11FocusRegressionTests(unittest.TestCase): + def setUp(self): + self.compositorSnapshot = input_event_manager.cthulhu_state.compositorSnapshot + input_event_manager.cthulhu_state.compositorSnapshot = None + self.addCleanup( + setattr, + input_event_manager.cthulhu_state, + "compositorSnapshot", + self.compositorSnapshot, + ) + def test_active_x11_window_differs_from_cached_atspi_window_by_pid(self): manager = input_event_manager.InputEventManager() cachedWindow = object() @@ -599,7 +610,6 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): def test_xterm_match_reuses_cached_metadata_while_ewmh_window_id_is_unchanged(self): manager = input_event_manager.InputEventManager() - activeWindow = object() with ( mock.patch.object( @@ -609,12 +619,7 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): ), mock.patch.object( manager, - "_get_active_x11_window", - return_value=activeWindow, - ) as getActiveWindow, - mock.patch.object( - manager, - "_x11_window_xterm_match", + "_x11_window_id_xterm_match", return_value=False, ) as matcher, ): @@ -623,8 +628,165 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): self.assertFalse(firstResult) self.assertFalse(secondResult) - getActiveWindow.assert_called_once_with(ewmhWindowId=0x1400010) - matcher.assert_called_once_with(activeWindow) + matcher.assert_called_once_with(0x1400010) + + def test_xterm_match_prefers_i3_focused_window_id_over_stale_ewmh_id(self): + manager = input_event_manager.InputEventManager() + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + focused_window_id=0x2600003, + focused_workspace_empty=False, + ) + + with ( + mock.patch.object(input_event_manager.cthulhu_state, "compositorSnapshot", snapshot), + mock.patch.object( + manager, + "_get_ewmh_active_x11_window_id", + return_value=0x200000C, + ) as getEwmhWindowId, + mock.patch.object( + manager, + "_x11_window_id_xterm_match", + return_value=False, + ) as matcher, + ): + result = manager._active_x11_window_xterm_match() + + self.assertFalse(result) + matcher.assert_called_once_with(0x2600003) + getEwmhWindowId.assert_not_called() + + def test_i3_exact_xid_initializes_x11_before_matching_xterm(self): + manager = input_event_manager.InputEventManager() + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + focused_window_id=0x200000C, + focused_workspace_empty=False, + ) + window = mock.Mock() + window.get_wm_class.return_value = ("xterm", "XTerm") + window.get_wm_name.return_value = "devel" + window.get_full_property.return_value = None + display = mock.Mock() + display.create_resource_object.return_value = window + + def initializeX11(): + manager._x11Display = display + manager._x11AnyPropertyType = object() + return True + + with ( + mock.patch.object(input_event_manager.cthulhu_state, "compositorSnapshot", snapshot), + mock.patch.object( + manager, + "_ensure_x11_display", + side_effect=initializeX11, + ) as ensureX11, + mock.patch.object(manager, "_get_ewmh_active_x11_window_id") as getEwmhWindowId, + ): + result = manager._active_x11_window_xterm_match() + + self.assertTrue(result) + ensureX11.assert_called_once_with() + display.create_resource_object.assert_called_once_with("window", 0x200000C) + getEwmhWindowId.assert_not_called() + + def test_confirmed_empty_i3_workspace_is_definitely_not_xterm(self): + manager = input_event_manager.InputEventManager() + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + focused_workspace_empty=True, + ) + + with ( + mock.patch.object(input_event_manager.cthulhu_state, "compositorSnapshot", snapshot), + mock.patch.object(manager, "_get_ewmh_active_x11_window_id") as getEwmhWindowId, + mock.patch.object(manager, "_x11_window_id_xterm_match") as matcher, + ): + result = manager._active_x11_window_xterm_match() + + self.assertFalse(result) + getEwmhWindowId.assert_not_called() + matcher.assert_not_called() + + def test_unknown_i3_context_falls_back_to_generic_ewmh_window_id(self): + manager = input_event_manager.InputEventManager() + snapshot = compositor_state_types.DesktopContextSnapshot( + session_type="x11", + backend_name="i3-ipc", + focused_window_id=None, + focused_workspace_empty=None, + ) + + with ( + mock.patch.object(input_event_manager.cthulhu_state, "compositorSnapshot", snapshot), + mock.patch.object( + manager, + "_get_ewmh_active_x11_window_id", + return_value=0x2600003, + ), + mock.patch.object( + manager, + "_x11_window_id_xterm_match", + return_value=False, + ) as matcher, + ): + result = manager._active_x11_window_xterm_match() + + self.assertFalse(result) + matcher.assert_called_once_with(0x2600003) + + def test_ewmh_brave_window_does_not_use_stale_wnck_xterm_metadata(self): + manager = input_event_manager.InputEventManager() + + with ( + mock.patch.object( + manager, + "_get_ewmh_active_x11_window_id", + return_value=0x2600003, + ), + mock.patch.object( + manager, + "_x11_window_id_xterm_match", + return_value=False, + ) as directMatcher, + mock.patch.object(manager, "_get_active_x11_window") as getWnckWindow, + mock.patch.object(manager, "_x11_window_xterm_match") as wnckMatcher, + ): + result = manager._active_x11_window_xterm_match() + + self.assertFalse(result) + directMatcher.assert_called_once_with(0x2600003) + getWnckWindow.assert_not_called() + wnckMatcher.assert_not_called() + + def test_direct_x11_metadata_identifies_brave_as_non_xterm(self): + manager = input_event_manager.InputEventManager() + + with mock.patch.object( + manager, + "_get_x11_window_metadata", + return_value=(["brave-origin", "Brave-origin", "New Tab - Brave Origin"], 1234), + ): + result = manager._x11_window_id_xterm_match(0x2600003) + + self.assertFalse(result) + + def test_direct_x11_metadata_identifies_xterm(self): + manager = input_event_manager.InputEventManager() + + with mock.patch.object( + manager, + "_get_x11_window_metadata", + return_value=(["xterm", "XTerm", "devel"], 1234), + ): + result = manager._x11_window_id_xterm_match(0x200000C) + + self.assertTrue(result) def test_xterm_focus_change_suspends_grabs_without_keyboard_event(self): manager = input_event_manager.InputEventManager() @@ -687,9 +849,30 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase): result = manager._poll_xterm_grab_recovery() self.assertTrue(result) + self.assertEqual(manager._xtermRecoveryPollCount, 1) restore.assert_called_once_with() suspend.assert_not_called() + def test_xterm_match_records_last_raw_ewmh_window_id(self): + manager = input_event_manager.InputEventManager() + + with ( + mock.patch.object( + manager, + "_get_ewmh_active_x11_window_id", + return_value=0x1400010, + ), + mock.patch.object( + manager, + "_x11_window_id_xterm_match", + return_value=False, + ), + ): + result = manager._active_x11_window_xterm_match() + + self.assertFalse(result) + self.assertEqual(manager._lastEwmhReadWindowId, 0x1400010) + def test_xterm_focus_poll_survives_transient_matcher_failure(self): manager = input_event_manager.InputEventManager() manager._device = object() diff --git a/tests/test_runtime_state_snapshot.py b/tests/test_runtime_state_snapshot.py index 6596020..e388825 100644 --- a/tests/test_runtime_state_snapshot.py +++ b/tests/test_runtime_state_snapshot.py @@ -82,12 +82,23 @@ class RuntimeStateSnapshotTests(unittest.TestCase): manager._scriptWithSuspendedGrabsForXterm = object() manager._xtermFocusHandlerId = 17 manager._xtermRecoverySourceId = 23 + manager._xtermRecoveryPollCount = 42 + manager._xtermHandoffHistory = [ + "ewmh:0x200000c->0x1400010", + "restore:success", + ] + manager._lastEwmhReadWindowId = 0x1400010 manager._lastEwmhActiveWindowId = 0x1400010 manager._lastEwmhXtermMatch = False manager._last_input_event = object() manager._last_non_modifier_key_event = None - snapshot = manager.get_debug_snapshot() + with mock.patch.object( + manager, + "_xterm_recovery_source_is_present", + return_value=True, + ): + snapshot = manager.get_debug_snapshot() self.assertTrue(snapshot["device_active"]) self.assertTrue(snapshot["pointer_watcher_active"]) @@ -97,6 +108,13 @@ class RuntimeStateSnapshotTests(unittest.TestCase): self.assertTrue(snapshot["suspended_xterm_script_present"]) self.assertTrue(snapshot["xterm_focus_monitor_active"]) self.assertTrue(snapshot["xterm_recovery_timer_active"]) + self.assertTrue(snapshot["xterm_recovery_source_present"]) + self.assertEqual(42, snapshot["xterm_recovery_poll_count"]) + self.assertEqual( + "ewmh:0x200000c->0x1400010 | restore:success", + snapshot["xterm_handoff_history"], + ) + self.assertEqual("0x1400010", snapshot["last_ewmh_read_window_id"]) self.assertEqual("0x1400010", snapshot["last_ewmh_active_window_id"]) self.assertFalse(snapshot["last_ewmh_xterm_match"]) self.assertFalse(snapshot["last_non_modifier_key_event_present"]) diff --git a/tests/test_web_input_regressions.py b/tests/test_web_input_regressions.py index 1582de5..f196012 100644 --- a/tests/test_web_input_regressions.py +++ b/tests/test_web_input_regressions.py @@ -1858,7 +1858,7 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase): setLocusOfFocus.assert_called_once_with(event, source, False) testScript.utilities.setCaretContext.assert_called_once_with(source, 0) - def test_focused_link_replaces_stale_section_context_after_caret_nav(self): + def test_delayed_focused_link_does_not_rewind_section_context_after_caret_nav(self): testScript = self._make_dynamic_script() document = object() section = object() @@ -1883,8 +1883,8 @@ class WebDynamicContentRecoveryRegressionTests(unittest.TestCase): result = web_script.Script.onFocusedChanged(testScript, event) self.assertTrue(result) - setLocusOfFocus.assert_called_once_with(event, link, False) - testScript.utilities.setCaretContext.assert_called_once_with(link, 0) + setLocusOfFocus.assert_not_called() + testScript.utilities.setCaretContext.assert_not_called() testScript.utilities.searchForCaretContext.assert_not_called() def test_browse_mode_sticky_blocks_web_app_descendant_focus_claim(self): testScript = self._make_dynamic_script() diff --git a/tests/test_window_title_fallback_regressions.py b/tests/test_window_title_fallback_regressions.py index aef52f3..266f994 100644 --- a/tests/test_window_title_fallback_regressions.py +++ b/tests/test_window_title_fallback_regressions.py @@ -34,6 +34,7 @@ class WindowTitleFallbackRegressionTests(unittest.TestCase): activeWindow = mock.Mock(id=200) with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), mock.patch.object(plugin, "_get_active_window", return_value=activeWindow), mock.patch.object(plugin, "_get_current_title", return_value="XTerm"), mock.patch.object(plugin, "_schedule_fallback_title") as scheduleFallback, @@ -47,7 +48,10 @@ class WindowTitleFallbackRegressionTests(unittest.TestCase): plugin._pollSourceId = 1 plugin._lastActiveWindowId = 100 - with mock.patch.object(plugin, "_get_active_window", return_value=None): + with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), + mock.patch.object(plugin, "_get_active_window", return_value=None), + ): self.assertTrue(plugin._poll_window_title()) self.assertEqual(plugin._lastActiveWindowId, 100) @@ -58,6 +62,7 @@ class WindowTitleFallbackRegressionTests(unittest.TestCase): activeWindow = mock.Mock() with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), mock.patch.object(plugin, "_get_active_window", return_value=activeWindow), mock.patch.object(plugin, "_get_current_title", return_value="Terminal"), ): @@ -69,11 +74,172 @@ class WindowTitleFallbackRegressionTests(unittest.TestCase): activeWindow = mock.Mock() with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), mock.patch.object(plugin, "_get_active_window", return_value=activeWindow), mock.patch.object(plugin, "_get_current_title", return_value="Game Window"), ): self.assertEqual(plugin.get_fallback_title("Wine Desktop"), "Game Window") + def test_fallback_prefers_i3_title_and_window_id(self): + plugin = WindowTitleReader() + plugin.app = mock.Mock() + snapshot = mock.Mock( + backend_name="i3-ipc", + focused_workspace_empty=False, + focused_window_id=0x200000C, + focused_window_title="Accurate i3 title", + ) + plugin.app.getCompositorStateAdapter.return_value.get_snapshot.return_value = snapshot + + with mock.patch.object(plugin, "_get_active_window") as getX11Window: + self.assertEqual(plugin.get_fallback_title("Incomplete title"), "Accurate i3 title") + + getX11Window.assert_not_called() + + def test_i3_wine_desktop_title_uses_existing_child_window_resolution(self): + plugin = WindowTitleReader() + plugin.app = mock.Mock() + snapshot = mock.Mock( + backend_name="i3-ipc", + focused_workspace_empty=False, + focused_window_id=0x2400007, + focused_window_title="Wine Desktop", + ) + desktopWindow = mock.Mock() + plugin.app.getCompositorStateAdapter.return_value.get_snapshot.return_value = snapshot + plugin._display = mock.Mock() + plugin._display.create_resource_object.return_value = desktopWindow + + with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), + mock.patch.object( + plugin, + "_get_window_title", + return_value="Wine Desktop", + ), + mock.patch.object( + plugin, + "_get_wine_desktop_title", + return_value="Actual Game Window", + ) as getWineTitle, + ): + self.assertEqual( + plugin.get_fallback_title("Wine Desktop"), + "Actual Game Window", + ) + + plugin._display.create_resource_object.assert_called_once_with("window", 0x2400007) + getWineTitle.assert_called_once_with(desktopWindow) + + def test_wine_title_resolution_ignores_unrelated_x11_focus(self): + plugin = WindowTitleReader() + desktopWindow = mock.Mock(id=10) + unrelatedFocus = mock.Mock(id=20) + + with ( + mock.patch.object(plugin, "_get_focus_window", return_value=unrelatedFocus), + mock.patch.object( + plugin, + "_window_is_descendant_of", + return_value=False, + ) as isDescendant, + mock.patch.object(plugin, "_get_window_title", return_value="Unrelated Terminal"), + mock.patch.object( + plugin, + "_find_child_title", + return_value="Actual Game Window", + ) as findChildTitle, + ): + self.assertEqual( + plugin._get_wine_desktop_title(desktopWindow), + "Actual Game Window", + ) + + isDescendant.assert_called_once_with(unrelatedFocus, desktopWindow) + findChildTitle.assert_called_once_with(desktopWindow) + + def test_wine_child_title_filter_rejects_infrastructure_windows(self): + plugin = WindowTitleReader() + helperWindow = mock.Mock() + helperWindow.get_wm_class.return_value = ( + "cthulhu-wine-access.exe", + "cthulhu-wine-access.exe", + ) + gameWindow = mock.Mock() + gameWindow.get_wm_class.return_value = ("game.exe", "game.exe") + + self.assertFalse(plugin._is_usable_wine_child_title(gameWindow, "Default IME")) + self.assertFalse(plugin._is_usable_wine_child_title(helperWindow, "NVDA")) + self.assertTrue(plugin._is_usable_wine_child_title(gameWindow, "Actual Game Window")) + + def test_confirmed_empty_i3_workspace_does_not_fall_back_to_stale_x11_title(self): + plugin = WindowTitleReader() + plugin.app = mock.Mock() + snapshot = mock.Mock( + backend_name="i3-ipc", + focused_workspace_empty=True, + focused_window_id=None, + focused_window_title="", + ) + plugin.app.getCompositorStateAdapter.return_value.get_snapshot.return_value = snapshot + + with mock.patch.object(plugin, "_get_active_window") as getX11Window: + self.assertEqual(plugin.get_fallback_title("Browser"), "") + + getX11Window.assert_not_called() + + def test_unknown_i3_context_falls_back_to_generic_x11_title(self): + plugin = WindowTitleReader() + plugin.app = mock.Mock() + snapshot = mock.Mock( + backend_name="i3-ipc", + focused_workspace_empty=None, + focused_window_id=None, + focused_window_title="", + ) + plugin.app.getCompositorStateAdapter.return_value.get_snapshot.return_value = snapshot + plugin._pollSourceId = 1 + activeWindow = mock.Mock() + + with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True) as ensureX11, + mock.patch.object(plugin, "_get_active_window", return_value=activeWindow), + mock.patch.object(plugin, "_get_current_title", return_value="Generic X title"), + ): + self.assertEqual(plugin.get_fallback_title("Incomplete"), "Generic X title") + + ensureX11.assert_called_once_with() + + def test_known_i3_xid_with_missing_title_uses_exact_x11_title(self): + plugin = WindowTitleReader() + plugin.app = mock.Mock() + snapshot = mock.Mock( + backend_name="i3-ipc", + focused_workspace_empty=False, + focused_window_id=0x2600003, + focused_window_title="", + ) + activeWindow = mock.Mock() + plugin.app.getCompositorStateAdapter.return_value.get_snapshot.return_value = snapshot + plugin._display = mock.Mock() + plugin._display.create_resource_object.return_value = activeWindow + + with ( + mock.patch.object(plugin, "_ensure_x11_display", return_value=True), + mock.patch.object( + plugin, + "_get_current_title", + return_value="Exact X11 title", + ) as getCurrentTitle, + ): + self.assertEqual( + plugin.get_fallback_title("Incomplete"), + "Exact X11 title", + ) + + plugin._display.create_resource_object.assert_called_once_with("window", 0x2600003) + getCurrentTitle.assert_called_once_with(activeWindow) + def test_present_title_uses_fallback_instead_of_atspi_title(self): presenter = where_am_i_presenter.WhereAmIPresenter() script = mock.Mock() diff --git a/tests/test_wine_access_controller_contract.py b/tests/test_wine_access_controller_contract.py index ae17ca8..15e1e60 100644 --- a/tests/test_wine_access_controller_contract.py +++ b/tests/test_wine_access_controller_contract.py @@ -19,6 +19,28 @@ class WineAccessControllerContractTests(unittest.TestCase): match.group(0), ) + def test_helper_uses_windows_subsystem_without_a_console_window(self): + cmakePath = Path(__file__).resolve().parents[1] / "wine-access" / "CMakeLists.txt" + source = cmakePath.read_text(encoding="utf-8") + + self.assertIn( + "target_link_options(cthulhu-wine-access PRIVATE -mwindows)", + source, + ) + + def test_compatibility_window_stays_hidden_zero_sized_and_nonactivating(self): + sourcePath = Path(__file__).resolve().parents[1] / "wine-access" / "main.cpp" + source = sourcePath.read_text(encoding="utf-8") + + self.assertIn("WS_EX_NOACTIVATE | WS_EX_TOOLWINDOW", source) + self.assertIsNotNone(re.search( + r"CreateWindowExW\(WS_EX_NOACTIVATE \| WS_EX_TOOLWINDOW,.*?" + r"L\"NVDA\", 0, 0, 0, 0, 0,", + source, + re.DOTALL, + )) + self.assertIn("ShowWindow(window, SW_HIDE)", source) + if __name__ == "__main__": unittest.main() diff --git a/wine-access/CMakeLists.txt b/wine-access/CMakeLists.txt index 90b3dbb..cdc2b4c 100644 --- a/wine-access/CMakeLists.txt +++ b/wine-access/CMakeLists.txt @@ -23,6 +23,7 @@ target_compile_features(cthulhu-wine-access PRIVATE cxx_std_20) target_include_directories(cthulhu-wine-access PRIVATE "${generated_dir}") target_link_libraries(cthulhu-wine-access PRIVATE PkgConfig::GIO oleacc ole32 oleaut32 rpcrt4 user32) +target_link_options(cthulhu-wine-access PRIVATE -mwindows) target_compile_options(cthulhu-wine-access PRIVATE $<$:-Wall -Wextra>)