refactor: add relevance gate helpers

src/cthulhu/event_manager.py

@@ -58,6 +58,9 @@ if TYPE_CHECKING:
 class EventManager:
 
     EMBEDDED_OBJECT_CHARACTER: str = '\ufffc'
+    RELEVANCE_KEEP: str = "keep"
+    RELEVANCE_COLLAPSE: str = "collapse"
+    RELEVANCE_DROP: str = "drop"
 
     def __init__(self, app: Cthulhu, asyncMode: bool = True) -> None:
         debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Initializing', True)
@@ -95,6 +98,11 @@ class EventManager:
                                            'object:state-changed:defunct',
                                            'object:property-change:accessible-parent']
         self._parentsOfDefunctDescendants: List[Atspi.Accessible] = []
+        self._compositorStateAdapter: Optional[Any] = None
+        self._churnSuppressed: bool = cthulhu_state.pauseAtspiChurn
+        self._prioritizedContextToken: Optional[str] = cthulhu_state.prioritizedDesktopContextToken
+        self._relevanceBurstWindow: float = 0.15
+        self._relevanceBurstHistory: Dict[Tuple[str, str], float] = {}
 
         cthulhu_state.device = None
         self._keyHandlingActive: bool = False
@@ -227,18 +235,163 @@ class EventManager:
         return "steamwebhelper" in cmdline
 
     def _isSteamNotificationEvent(self, event: Atspi.Event) -> bool:
-        for obj in (event.any_data, event.source):
+        return self._isLiveOrNotificationEvent(event)
-            if not isinstance(obj, Atspi.Accessible):
+
-                continue
+    def _isLiveOrNotificationObject(self, obj: Any) -> bool:
+        if obj is None:
+            return False
+
+        try:
             if AXUtilities.is_notification(obj) or AXUtilities.is_alert(obj):
                 return True
+        except Exception:
+            return False
+
+        try:
             liveAttr = AXObject.get_attribute(obj, 'live')
             containerLive = AXObject.get_attribute(obj, 'container-live')
-            if liveAttr in ('assertive', 'polite') or containerLive in ('assertive', 'polite'):
+        except Exception:
+            return False
+
+        if liveAttr in ('assertive', 'polite') or containerLive in ('assertive', 'polite'):
+            return True
+
+        return False
+
+    def _isLiveOrNotificationEvent(self, event: Atspi.Event) -> bool:
+        for obj in (event.any_data, event.source):
+            if self._isLiveOrNotificationObject(obj):
                 return True
 
         return False
 
+    def _isPhaseOneWebRole(self, role: Atspi.Role) -> bool:
+        return role in (
+            Atspi.Role.LINK,
+            Atspi.Role.PUSH_BUTTON,
+            Atspi.Role.CHECK_BOX,
+            Atspi.Role.RADIO_BUTTON,
+            Atspi.Role.LIST_ITEM,
+            Atspi.Role.LIST,
+            Atspi.Role.SECTION,
+        )
+
+    def _eventTouchesCurrentFocusContext(self, event: Atspi.Event) -> bool:
+        locusOfFocus = cthulhu_state.locusOfFocus
+        if locusOfFocus is None:
+            return False
+
+        if event.source in (locusOfFocus, getattr(event, "any_data", None)):
+            return True
+
+        focused = getattr(cthulhu_state, "pendingSelfHostedFocus", None)
+        if focused is not None and event.source in (focused, getattr(event, "any_data", None)):
+            return True
+
+        return False
+
+    def _shouldPreserveForRelevanceGate(self, event: Atspi.Event) -> bool:
+        if event.type.startswith("window"):
+            return True
+
+        if event.type.startswith("mouse:button"):
+            return True
+
+        if event.type.startswith("object:state-changed:focused") and event.detail1:
+            return True
+
+        if self._eventTouchesCurrentFocusContext(event):
+            return True
+
+        return self._isLiveOrNotificationEvent(event)
+
+    def _eventRoleForRelevance(self, event: Atspi.Event) -> Optional[Atspi.Role]:
+        try:
+            return AXObject.get_role(event.source)
+        except Exception:
+            return None
+
+    def _appNameForRelevance(self, app: Atspi.Accessible) -> str:
+        try:
+            appName = AXObject.get_name(app)
+        except Exception:
+            return ""
+
+        return (appName or "").strip().lower()
+
+    def _relevanceEventGroup(self, event: Atspi.Event, role: Atspi.Role) -> Optional[str]:
+        if event.type.startswith("object:children-changed:") and self._isPhaseOneWebRole(role):
+            return "children-changed"
+
+        if event.type.startswith("object:property-change:accessible-name") and self._isPhaseOneWebRole(role):
+            return "accessible-name"
+
+        return None
+
+    def _classifyRelevance(self, event: Atspi.Event, app: Atspi.Accessible) -> str:
+        if self._is_obsolete_by_context(event):
+            return self.RELEVANCE_DROP
+
+        if self._shouldPreserveForRelevanceGate(event):
+            return self.RELEVANCE_KEEP
+
+        if event.type.startswith("object:state-changed:focused"):
+            return self.RELEVANCE_KEEP if event.detail1 else self.RELEVANCE_DROP
+
+        role = self._eventRoleForRelevance(event)
+        eventGroup = self._relevanceEventGroup(event, role)
+        if eventGroup is None:
+            return self.RELEVANCE_KEEP
+
+        if event.type.startswith("object:property-change:accessible-name") \
+           and self._isPhaseOneWebRole(role):
+            try:
+                focused = AXUtilities.is_focused(event.source)
+            except Exception:
+                focused = False
+            if not focused:
+                return self.RELEVANCE_DROP
+
+        if self._isBurstRelevanceEvent(event, app, eventGroup):
+            return self.RELEVANCE_COLLAPSE
+
+        return self.RELEVANCE_KEEP
+
+    def _isBurstRelevanceEvent(
+        self,
+        event: Atspi.Event,
+        app: Atspi.Accessible,
+        eventGroup: Optional[str] = None,
+    ) -> bool:
+        if self._shouldPreserveForRelevanceGate(event):
+            return False
+
+        if eventGroup is None:
+            role = self._eventRoleForRelevance(event)
+            eventGroup = self._relevanceEventGroup(event, role)
+
+        if eventGroup is None:
+            return False
+
+        appName = self._appNameForRelevance(app) or "unknown-app"
+        burstKey = (appName, eventGroup)
+        now = time.monotonic()
+        lastSeen = self._relevanceBurstHistory.get(burstKey)
+        self._relevanceBurstHistory[burstKey] = now
+        return lastSeen is not None and (now - lastSeen) < self._relevanceBurstWindow
+
+    def _isSteamBurstChurnEvent(self, event: Atspi.Event, app: Atspi.Accessible) -> bool:
+        if not self._isSteamApp(app) or self._isLiveOrNotificationEvent(event):
+            return False
+
+        if event.type.startswith("object:state-changed:focused"):
+            return not event.detail1
+
+        if event.type.startswith("object:children-changed:"):
+            return self._isBurstRelevanceEvent(event, app, "children-changed")
+
+        return False
+
     def _ignore(self, event: Atspi.Event) -> bool:
         """Returns True if this event should be ignored."""
 
@@ -315,6 +468,9 @@ class EventManager:
                 else:
                     return _ignore_with_reason("inactive-app", "event not from active app")
 
+        if self._isSteamBurstChurnEvent(event, app):
+            return _ignore_with_reason("steam-burst-churn", "event is low-value Steam churn")
+
         if event.type.startswith('object:text-changed') \
            and self.EMBEDDED_OBJECT_CHARACTER in event.any_data \
            and not event.any_data.replace(self.EMBEDDED_OBJECT_CHARACTER, ""):
@@ -418,32 +574,27 @@ class EventManager:
                 tokens = ["EVENT MANAGER: Locus of focus:", cthulhu_state.locusOfFocus]
                 debug.printTokens(debug.LEVEL_INFO, tokens, True)
 
-            defunct = AXObject.is_dead(event.any_data) or AXUtilities.is_defunct(event.any_data)
+            elif event.any_data == cthulhu_state.locusOfFocus:
-            if defunct:
+                msg = 'EVENT MANAGER: Locus of focus is being added.'
-                _log_ignore("defunct-child", "potentially defunct child/descendant")
+                debug.printMessage(debug.LEVEL_INFO, msg, True)
-                if AXUtilities.manages_descendants(event.source) \
+                return False
-                   and event.source not in self._parentsOfDefunctDescendants:
-                    self._parentsOfDefunctDescendants.append(event.source)
-                return True
 
-            if event.source in self._parentsOfDefunctDescendants:
+            elif event.any_data == cthulhu_state.pendingSelfHostedFocus:
-                self._parentsOfDefunctDescendants.remove(event.source)
+                msg = 'EVENT MANAGER: Pending self-hosted focus is being added.'
+                debug.printMessage(debug.LEVEL_INFO, msg, True)
+                return False
 
-            # This should be safe. We do not have a reason to present a newly-added,
+            elif self._eventsSuspended:
-            # but not focused image. We do not need to update live regions for images.
+                if event.type.endswith('add') and not self._eventTypeIsSuspended(event.type):
-            # This is very likely a completely and utterly useless event for us. The
+                    # Cthulhu is known to miss text insertion in Gtk3 when text insertions
-            # reason for ignoring it here rather than quickly processing it is the
+                    # are suspended. So while we do need to suspend some of the resulting
-            # potential for event floods like we're seeing from matrix.org.
+                    # events, we also need to make sure the text insertion isn't suspended.
-            if AXUtilities.is_image(event.any_data):
+                    if AXUtilities.is_text(event.any_data):
-                return _ignore_with_reason("child-image", "role filtered")
+                        return _ignore_with_reason("suspended-text", "child text is not suspended")
+                elif not self._eventTypeIsSuspended(event.type):
+                    return _ignore_with_reason("suspended-event-type", "event type is not suspended")
 
-            # In normal apps we would have caught this from the parent role.
+        return False
-            # But gnome-shell has panel parents adding/removing menu items.
-            if event.type.startswith('object:children-changed'):
-                if AXUtilities.is_menu_item(event.any_data):
-                    return _ignore_with_reason("child-menu-item", "child role filtered")
-
-        return _allow_with_reason("no-cause", "no ignore condition met")
 
     def _addToQueue(self, event: Any, asyncMode: bool) -> None:
         debugging = debug.debugEventQueue