From 629785d548ef631a7a451371e5c13dc249a7ca0b Mon Sep 17 00:00:00 2001 From: Hunter Jozwiak Date: Thu, 9 Apr 2026 10:53:47 -0400 Subject: [PATCH] refactor: add relevance gate helpers --- src/cthulhu/event_manager.py | 205 ++++++++++++++++++++++++++++++----- 1 file changed, 178 insertions(+), 27 deletions(-) diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 8dc37b7..d1be833 100644 --- a/src/cthulhu/event_manager.py +++ b/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): - if not isinstance(obj, Atspi.Accessible): - continue + return self._isLiveOrNotificationEvent(event) + + 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) - if defunct: - _log_ignore("defunct-child", "potentially defunct child/descendant") - if AXUtilities.manages_descendants(event.source) \ - and event.source not in self._parentsOfDefunctDescendants: - self._parentsOfDefunctDescendants.append(event.source) - return True + elif event.any_data == cthulhu_state.locusOfFocus: + msg = 'EVENT MANAGER: Locus of focus is being added.' + debug.printMessage(debug.LEVEL_INFO, msg, True) + return False - if event.source in self._parentsOfDefunctDescendants: - self._parentsOfDefunctDescendants.remove(event.source) + elif event.any_data == cthulhu_state.pendingSelfHostedFocus: + 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, - # but not focused image. We do not need to update live regions for images. - # This is very likely a completely and utterly useless event for us. The - # reason for ignoring it here rather than quickly processing it is the - # potential for event floods like we're seeing from matrix.org. - if AXUtilities.is_image(event.any_data): - return _ignore_with_reason("child-image", "role filtered") + elif self._eventsSuspended: + if event.type.endswith('add') and not self._eventTypeIsSuspended(event.type): + # Cthulhu is known to miss text insertion in Gtk3 when text insertions + # are suspended. So while we do need to suspend some of the resulting + # events, we also need to make sure the text insertion isn't suspended. + if AXUtilities.is_text(event.any_data): + 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. - # 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") + return False def _addToQueue(self, event: Any, asyncMode: bool) -> None: debugging = debug.debugEventQueue