diff --git a/src/cthulhu/compositor_state_adapter.py b/src/cthulhu/compositor_state_adapter.py new file mode 100644 index 0000000..6e28f7b --- /dev/null +++ b/src/cthulhu/compositor_state_adapter.py @@ -0,0 +1,177 @@ +#!/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. + +"""Adapter that normalizes compositor and AT-SPI desktop context signals.""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable +from typing import Any, Optional + +from . import cthulhu_state +from .ax_object import AXObject +from .compositor_state_types import ( + CompositorStateEvent, + DesktopContextSnapshot, + DESKTOP_FOCUS_CONTEXT_CHANGED, + FLUSH_STALE_ATSPI_EVENTS, + PAUSE_ATSPI_CHURN, + PRIORITIZE_FOCUS, + RESUME_ATSPI_CHURN, +) +from .wnck_support import get_session_type + + +class CompositorStateAdapter: + """Normalizes compositor state and desktop-focus context changes.""" + + def __init__( + self, + app: Optional[Any] = None, + event_manager: Optional[Any] = None, + workspace_backends: Optional[Iterable[Any]] = None, + ) -> None: + self.app = app + self.event_manager = event_manager + self._workspaceBackends = list(workspace_backends or []) + self._workspaceBackend: Optional[Any] = None + self._listeners: list[Callable[[CompositorStateEvent], None]] = [] + self._snapshot = DesktopContextSnapshot(session_type=self.get_session_type()) + + def add_listener(self, listener: Callable[[CompositorStateEvent], None]) -> None: + if listener not in self._listeners: + self._listeners.append(listener) + + def remove_listener(self, listener: Callable[[CompositorStateEvent], None]) -> None: + if listener in self._listeners: + self._listeners.remove(listener) + + def get_snapshot(self) -> DesktopContextSnapshot: + return self._snapshot + + def get_session_type(self) -> str: + return get_session_type() + + def activate(self) -> None: + self._workspaceBackend = None + for backend in self._workspaceBackends: + if not self._backend_is_available(backend): + continue + self._workspaceBackend = backend + self._backend_activate(backend) + break + + self.sync_accessible_context("activate") + + def deactivate(self) -> None: + if self._workspaceBackend is not None: + self._backend_deactivate(self._workspaceBackend) + self._workspaceBackend = None + cthulhu_state.pauseAtspiChurn = False + + def sync_accessible_context(self, reason: str) -> DesktopContextSnapshot: + active_window = cthulhu_state.activeWindow + locus_of_focus = cthulhu_state.locusOfFocus + + active_window_snapshot = self._build_object_snapshot(active_window) + locus_of_focus_snapshot = self._build_object_snapshot(locus_of_focus) + snapshot = DesktopContextSnapshot( + session_type=self.get_session_type(), + active_window_token=active_window_snapshot["token"], + locus_of_focus_token=locus_of_focus_snapshot["token"], + active_window_pid=active_window_snapshot["pid"], + 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"], + ) + + previous_token = self._snapshot.active_window_token + self._snapshot = snapshot + cthulhu_state.compositorSnapshot = snapshot + + if snapshot.active_window_token != previous_token: + cthulhu_state.prioritizedDesktopContextToken = snapshot.active_window_token or None + self._emit( + DESKTOP_FOCUS_CONTEXT_CHANGED, + reason, + snapshot, + ) + self._emit( + PRIORITIZE_FOCUS, + reason, + snapshot, + ) + + return snapshot + + def _build_object_snapshot(self, obj: Any) -> dict[str, Any]: + if obj is None: + return {"token": "", "pid": -1, "name": ""} + + pid = AXObject.get_process_id(obj) + name = (AXObject.get_name(obj) or "").strip() + if not name: + name = AXObject.get_role_name(obj) or AXObject.get_accessible_id(obj) or "unknown" + + token = "" + if pid >= 0 or name: + token = f"{max(pid, 0)}:{name}" + + return {"token": token, "pid": pid, "name": name} + + def _emit(self, event_type: str, reason: str, snapshot: Optional[DesktopContextSnapshot]) -> None: + event = CompositorStateEvent(event_type, reason=reason, snapshot=snapshot) + for listener in list(self._listeners): + listener(event) + + def _handle_workspace_signal(self, signal_name: str, *_args: Any) -> None: + normalized = (signal_name or "").strip().lower() + if not normalized: + return + + if "transition" not in normalized and "workspace" not in normalized: + return + + if any(marker in normalized for marker in ("start", "begin", "enter", "prepare", "opening")): + cthulhu_state.pauseAtspiChurn = True + self._emit(PAUSE_ATSPI_CHURN, signal_name, self._snapshot) + return + + if any(marker in normalized for marker in ("end", "complete", "finished", "finish", "leave", "exit", "stop", "done")): + cthulhu_state.pauseAtspiChurn = False + self._emit(RESUME_ATSPI_CHURN, signal_name, self._snapshot) + self._emit(FLUSH_STALE_ATSPI_EVENTS, signal_name, self._snapshot) + return + + if "flush" in normalized or "stale" in normalized: + self._emit(FLUSH_STALE_ATSPI_EVENTS, signal_name, self._snapshot) + + def _backend_is_available(self, backend: Any) -> bool: + if hasattr(backend, "is_available"): + try: + return bool(backend.is_available(self.get_session_type())) + except TypeError: + return bool(backend.is_available()) + if hasattr(backend, "available"): + return bool(backend.available) + return True + + def _backend_activate(self, backend: Any) -> None: + if hasattr(backend, "activate"): + try: + backend.activate(self) + except TypeError: + backend.activate() + + def _backend_deactivate(self, backend: Any) -> None: + if hasattr(backend, "deactivate"): + try: + backend.deactivate(self) + except TypeError: + backend.deactivate() diff --git a/src/cthulhu/compositor_state_types.py b/src/cthulhu/compositor_state_types.py new file mode 100644 index 0000000..9be6d5a --- /dev/null +++ b/src/cthulhu/compositor_state_types.py @@ -0,0 +1,44 @@ +#!/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. + +"""Normalized compositor-state event vocabulary and snapshots.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Mapping, Optional + +DESKTOP_FOCUS_CONTEXT_CHANGED = "DESKTOP_FOCUS_CONTEXT_CHANGED" +PRIORITIZE_FOCUS = "PRIORITIZE_FOCUS" +PAUSE_ATSPI_CHURN = "PAUSE_ATSPI_CHURN" +RESUME_ATSPI_CHURN = "RESUME_ATSPI_CHURN" +FLUSH_STALE_ATSPI_EVENTS = "FLUSH_STALE_ATSPI_EVENTS" + + +@dataclass(frozen=True, slots=True) +class DesktopContextSnapshot: + """Immutable snapshot of the desktop context tracked by the adapter.""" + + session_type: str + active_window_token: str = "" + locus_of_focus_token: str = "" + active_window_pid: int = -1 + locus_of_focus_pid: int = -1 + active_window_name: str = "" + locus_of_focus_name: str = "" + + +@dataclass(frozen=True, slots=True) +class CompositorStateEvent: + """Normalized event emitted by the compositor state adapter.""" + + type: str + reason: str = "" + snapshot: Optional[DesktopContextSnapshot] = None + payload: Mapping[str, Any] = field(default_factory=dict) diff --git a/src/cthulhu/cthulhu.py b/src/cthulhu/cthulhu.py index 711803a..8f0e71b 100644 --- a/src/cthulhu/cthulhu.py +++ b/src/cthulhu/cthulhu.py @@ -51,6 +51,7 @@ if TYPE_CHECKING: from .input_event import InputEvent from .input_event_manager import InputEventManager from .event_manager import EventManager + from .compositor_state_adapter import CompositorStateAdapter from .signal_manager import SignalManager from .dynamic_api_manager import DynamicApiManager from .speech import Speech @@ -258,6 +259,7 @@ except Exception: from . import braille from . import debug from . import event_manager +from . import compositor_state_adapter from . import keybindings from . import learn_mode_presenter from . import logger @@ -492,6 +494,7 @@ def loadUserSettings(script: Optional[Any] = None, inputEvent: Optional[Any] = N cthulhu_modifier_manager.getManager().refreshCthulhuModifiers("Loading user settings.") # Activate core systems FIRST before loading plugins + cthulhuApp.compositorStateAdapter.activate() cthulhuApp.scriptManager.activate() cthulhuApp.eventManager.activate() @@ -777,6 +780,7 @@ def shutdown(script: Optional[Any] = None, inputEvent: Optional[Any] = None) -> # Deactivate the event manager first so that it clears its queue and will not # accept new events. Then let the script manager unregister script event listeners. + cthulhuApp.compositorStateAdapter.deactivate() cthulhuApp.eventManager.deactivate() cthulhuApp.scriptManager.deactivate() @@ -953,6 +957,11 @@ class Cthulhu(GObject.Object): self.resourceManager: resource_manager.ResourceManager = resource_manager.ResourceManager(self) self.settingsManager: SettingsManager = settings_manager.SettingsManager(self) # Directly instantiate self.eventManager: EventManager = event_manager.EventManager(self) # Directly instantiate + self.compositorStateAdapter: CompositorStateAdapter = compositor_state_adapter.CompositorStateAdapter( + self, + self.eventManager, + ) + self.eventManager.set_compositor_state_adapter(self.compositorStateAdapter) self.scriptManager: ScriptManager = script_manager.ScriptManager(self) # Directly instantiate script_manager._manager = self.scriptManager self.logger: logger.Logger = logger.Logger() # Directly instantiate @@ -986,6 +995,9 @@ class Cthulhu(GObject.Object): def getEventManager(self) -> EventManager: return self.eventManager + def getCompositorStateAdapter(self) -> CompositorStateAdapter: + return self.compositorStateAdapter + def getSettingsManager(self) -> SettingsManager: return self.settingsManager diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 7d177c4..5a29b02 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -181,6 +181,11 @@ class EventManager: self._deactivateKeyHandling() debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Deactivated', True) + def set_compositor_state_adapter(self, adapter: Any) -> None: + """Stores the compositor state adapter used for startup wiring.""" + + self._compositorStateAdapter = adapter + def ignoreEventTypes(self, eventTypeList: List[str]) -> None: for eventType in eventTypeList: if eventType not in self._ignoredEvents: diff --git a/src/cthulhu/meson.build b/src/cthulhu/meson.build index 30a299f..8e9d9eb 100644 --- a/src/cthulhu/meson.build +++ b/src/cthulhu/meson.build @@ -33,6 +33,8 @@ cthulhu_python_sources = files([ 'chnames.py', 'cmdnames.py', 'colornames.py', + 'compositor_state_adapter.py', + 'compositor_state_types.py', 'common_keyboardmap.py', 'cthulhuVersion.py', 'cthulhu_modifier_manager.py', diff --git a/tests/test_compositor_state_adapter_regressions.py b/tests/test_compositor_state_adapter_regressions.py new file mode 100644 index 0000000..dea50bf --- /dev/null +++ b/tests/test_compositor_state_adapter_regressions.py @@ -0,0 +1,101 @@ +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cthulhu import cthulhu_state +from cthulhu import compositor_state_adapter +from cthulhu import compositor_state_types + + +class FakeWorkspaceBackend: + def __init__(self, available: bool, name: str) -> None: + self.available = available + self.name = name + self.activate_calls = [] + self.deactivate_calls = [] + + def is_available(self, session_type: str | None = None) -> bool: + return self.available + + def activate(self, adapter=None) -> None: + self.activate_calls.append(adapter) + + def deactivate(self, adapter=None) -> None: + self.deactivate_calls.append(adapter) + + +class CompositorStateAdapterRegressionTests(unittest.TestCase): + def setUp(self) -> None: + cthulhu_state.compositorSnapshot = None + cthulhu_state.pauseAtspiChurn = False + cthulhu_state.prioritizedDesktopContextToken = None + + def test_activate_selects_first_available_backend(self) -> None: + unavailableBackend = FakeWorkspaceBackend(False, "unavailable") + selectedBackend = FakeWorkspaceBackend(True, "selected") + adapter = compositor_state_adapter.CompositorStateAdapter( + workspace_backends=[unavailableBackend, selectedBackend] + ) + + adapter.activate() + + self.assertEqual(unavailableBackend.activate_calls, []) + self.assertEqual(selectedBackend.activate_calls, [adapter]) + self.assertIs(adapter._workspaceBackend, selectedBackend) + + def test_sync_accessible_context_emits_focus_context_changed_when_active_window_token_changes(self) -> None: + adapter = compositor_state_adapter.CompositorStateAdapter() + events = [] + adapter.add_listener(events.append) + firstWindow = object() + secondWindow = object() + + def get_process_id(obj): + return 111 if obj is firstWindow else 222 + + def get_name(obj): + return "Terminal" + + with ( + mock.patch.object(compositor_state_adapter.AXObject, "get_process_id", side_effect=get_process_id), + mock.patch.object(compositor_state_adapter.AXObject, "get_name", side_effect=get_name), + ): + cthulhu_state.activeWindow = firstWindow + cthulhu_state.locusOfFocus = firstWindow + adapter.sync_accessible_context("startup") + + events.clear() + cthulhu_state.activeWindow = secondWindow + cthulhu_state.locusOfFocus = secondWindow + adapter.sync_accessible_context("workspace transition") + + self.assertIn( + compositor_state_types.DESKTOP_FOCUS_CONTEXT_CHANGED, + [event.type for event in events], + ) + self.assertIn( + compositor_state_types.PRIORITIZE_FOCUS, + [event.type for event in events], + ) + + def test_sync_accessible_context_builds_stable_active_window_tokens(self) -> None: + adapter = compositor_state_adapter.CompositorStateAdapter() + window = object() + + with ( + mock.patch.object(compositor_state_adapter.AXObject, "get_process_id", return_value=4242), + mock.patch.object(compositor_state_adapter.AXObject, "get_name", return_value="Terminal"), + ): + cthulhu_state.activeWindow = window + cthulhu_state.locusOfFocus = window + snapshot = adapter.sync_accessible_context("startup") + + self.assertEqual(snapshot.active_window_token, "4242:Terminal") + self.assertEqual(cthulhu_state.compositorSnapshot.active_window_token, "4242:Terminal") + + +if __name__ == "__main__": + unittest.main()