feat: add compositor state adapter scaffold

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()

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)

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
 

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:

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',