Some web work. Firefox still being weird.

This commit is contained in:
Storm Dragon
2026-07-06 18:22:19 -04:00
parent 3926ed0044
commit 323d0a3975
11 changed files with 584 additions and 1 deletions
+26
View File
@@ -1042,6 +1042,32 @@ class Cthulhu(GObject.Object):
def getLogger(self) -> logger.Logger: # New getter for the logger
return self.logger
def get_debug_snapshot(self) -> Dict[str, Dict[str, object]]:
"""Returns read-only runtime counters useful for diagnostics."""
speechServer = speech.getSpeechServer()
snapshot: Dict[str, Dict[str, object]] = {
"cthulhu_state": {
"active_window_present": cthulhu_state.activeWindow is not None,
"locus_of_focus_present": cthulhu_state.locusOfFocus is not None,
"active_script_present": cthulhu_state.activeScript is not None,
"pending_self_hosted_focus_present": cthulhu_state.pendingSelfHostedFocus is not None,
"device_present": cthulhu_state.device is not None,
"pause_atspi_churn": cthulhu_state.pauseAtspiChurn,
"prioritized_desktop_context_token_present":
cthulhu_state.prioritizedDesktopContextToken is not None,
},
"speech": {
"speech_server_present": speechServer is not None,
"speech_server": speechServer.__class__.__name__ if speechServer else "",
},
"event_manager": self.eventManager.get_debug_snapshot(),
"script_manager": self.scriptManager.get_debug_snapshot(),
"input_event_manager": input_event_manager.get_manager().get_debug_snapshot(),
"plugin_system_manager": self.pluginSystemManager.get_debug_snapshot(),
}
return snapshot
def addKeyGrab(self, binding: Any) -> List[int]:
return addKeyGrab(binding)
+11
View File
@@ -698,6 +698,17 @@ class CthulhuDBusServiceInterface(Publishable):
debug.print_message(debug.LEVEL_INFO, msg, True)
return result
def GetRuntimeStateSnapshot(self) -> str: # pylint: disable=invalid-name
"""Returns read-only runtime state counters for diagnostics."""
msg = "DBUS SERVICE: GetRuntimeStateSnapshot called."
debug.print_message(debug.LEVEL_INFO, msg, True)
from . import cthulhu # pylint: disable=import-outside-toplevel
snapshot = cthulhu.cthulhuApp.get_debug_snapshot()
return debug.format_runtime_state_snapshot(snapshot)
def Quit(self) -> bool: # pylint: disable=invalid-name
"""Quits Cthulhu."""
+15 -1
View File
@@ -45,7 +45,7 @@ import subprocess
import sys
import time
import types
from typing import Any, Generator, Optional, TextIO, Pattern, TYPE_CHECKING
from typing import Any, Generator, Mapping, Optional, TextIO, Pattern, TYPE_CHECKING
from datetime import datetime
import gi
@@ -291,6 +291,20 @@ def print_log_tokens(level: int, prefix: str, tokens: list[Any], reason: Optiona
text = format_log_message(prefix, _format_tokens(tokens), reason)
printMessage(level, text, timestamp, stack)
def format_runtime_state_snapshot(snapshot: Mapping[str, Any]) -> str:
"""Formats read-only runtime state counters for debug capture."""
lines = ["CTHULHU RUNTIME STATE SNAPSHOT"]
for section, values in snapshot.items():
lines.append(f"[{section}]")
if not isinstance(values, Mapping):
lines.append(f"value: {values}")
continue
for key in sorted(values):
lines.append(f"{key}: {values[key]}")
return "\n".join(lines)
def reset_startup_timing() -> None:
"""Reset the startup timing baseline."""
+24
View File
@@ -194,6 +194,30 @@ class EventManager:
self._deactivateKeyHandling()
debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Deactivated', True)
def get_debug_snapshot(self) -> Dict[str, object]:
"""Returns read-only counters useful for long-uptime diagnostics."""
return {
"active": self._active,
"async_mode": self._asyncMode,
"event_queue_size": self._eventQueue.qsize(),
"prioritized_event_present": self._prioritizedEvent is not None,
"idle_source_id": self._gidleId,
"prioritized_idle_source_id": self._prioritizedIdleId,
"events_suspended": self._eventsSuspended,
"suspendable_event_types": len(self._suspendableEvents),
"events_triggering_suspension": len(self._eventsTriggeringSuspension),
"ignored_event_types": len(self._ignoredEvents),
"script_listener_types": len(self._scriptListenerCounts),
"parents_of_defunct_descendants": len(self._parentsOfDefunctDescendants),
"cmdline_cache_size": len(self._cmdlineCache),
"relevance_burst_history_size": len(self._relevanceBurstHistory),
"churn_suppressed": self._churnSuppressed,
"prioritized_context_token_present": self._prioritizedContextToken is not None,
"key_handling_active": self._keyHandlingActive,
"input_event_manager_present": self._inputEventManager is not None,
}
def set_compositor_state_adapter(self, adapter: Any) -> None:
"""Stores the compositor state adapter used for startup wiring."""
+15
View File
@@ -109,6 +109,21 @@ class InputEventManager:
debug.print_message(debug.LEVEL_INFO, msg, True)
self._device = None
def get_debug_snapshot(self) -> Dict[str, object]:
"""Returns read-only counters useful for long-uptime diagnostics."""
return {
"device_active": self._device is not None,
"pointer_watcher_active": self._pointer_moved_id != 0,
"mapped_keycodes": len(self._mapped_keycodes),
"mapped_keysyms": len(self._mapped_keysyms),
"grabbed_bindings": len(self._grabbed_bindings),
"paused": self._paused,
"suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None,
"last_input_event_present": self._last_input_event is not None,
"last_non_modifier_key_event_present": self._last_non_modifier_key_event is not None,
}
def enable_pointer_monitoring(self) -> bool:
"""Enables pointer monitoring on the current device, if possible."""
+22
View File
@@ -1102,3 +1102,25 @@ class PluginSystemManager:
for pluginInfo in self.plugins:
if ForceAllPlugins or pluginInfo.loaded:
self.unloadPlugin(pluginInfo)
def get_debug_snapshot(self) -> Dict[str, object]:
"""Returns read-only counters useful for long-uptime diagnostics."""
loadedPlugins = [info for info in self._plugins.values() if info.loaded]
try:
pluggyRegisteredPlugins = len(self.plugin_manager.get_plugins())
except Exception:
pluggyRegisteredPlugins = -1
return {
"scanned_plugins": len(self._plugins),
"active_plugins": len(self._active_plugins),
"loaded_plugins": len(loadedPlugins),
"builtin_plugins": sum(1 for info in self._plugins.values() if info.builtin),
"hidden_plugins": sum(1 for info in self._plugins.values() if info.hidden),
"plugin_keybinding_contexts": len(self._plugin_keybindings),
"plugin_keybindings": sum(len(bindings) for bindings in self._plugin_keybindings.values()),
"global_keybindings": len(self._global_bindings),
"last_active_script_present": self._last_active_script is not None,
"pluggy_registered_plugins": pluggyRegisteredPlugins,
}
+25
View File
@@ -396,6 +396,31 @@ class ScriptManager:
reason
)
def get_debug_snapshot(self) -> Dict[str, object]:
"""Returns read-only counters useful for long-uptime diagnostics."""
activeScript = cthulhu_state.activeScript
activeApp = getattr(activeScript, "app", None)
activeAppName = ""
if activeApp is not None:
try:
activeAppName = AXObject.get_name(activeApp) or ""
except Exception:
activeAppName = "[unavailable]"
return {
"active": self._active,
"active_script": activeScript.__class__.__name__ if activeScript else "",
"active_script_app": activeAppName,
"default_script_present": self._defaultScript is not None,
"app_scripts": len(self.appScripts),
"toolkit_script_apps": len(self.toolkitScripts),
"toolkit_scripts": sum(len(scripts) for scripts in self.toolkitScripts.values()),
"custom_script_apps": len(self.customScripts),
"custom_scripts": sum(len(scripts) for scripts in self.customScripts.values()),
"sleep_mode_scripts": len(self._sleepModeScripts),
}
def _get_script_for_app_replicant(self, app: Atspi.Accessible) -> Optional[Script]:
if not self._active:
return None
+19
View File
@@ -2116,6 +2116,11 @@ class Script(default.Script):
else:
self._clearSyntheticWebSelection()
if self.utilities.lastInputEventWasCharNav() and (event.source, event.detail1) == (obj, offset):
msg = "WEB: Event ignored: Character navigation already presented this context"
debug.printMessage(debug.LEVEL_INFO, msg, True)
return True
if self._lastCommandWasCaretNav:
msg = "WEB: Event ignored: Last command was caret nav"
debug.printMessage(debug.LEVEL_INFO, msg, True)
@@ -2624,6 +2629,20 @@ 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)
@@ -230,6 +230,7 @@ class Utilities(script_utilities.Utilities):
self._shouldInferLabelFor = {}
self._treatAsTextObject = {}
self._treatAsDiv = {}
self.clearContentCache()
self._paths = {}
self._contextPathsRolesAndNames = {}
self._canHaveCaretContextDecision = {}
@@ -1793,6 +1794,21 @@ class Utilities(script_utilities.Utilities):
self._canHaveCaretContextDecision = {}
return rv
def _trimContentsBeforeEmbeddedObjectBoundary(self, contents, obj, offset):
if not contents:
return contents
firstObj, firstStart, firstEnd, firstString = contents[0]
if firstObj != obj or not (firstStart < offset < firstEnd):
return contents
char = AXText.get_character_at_offset(obj, firstStart)[0]
if char != self.EMBEDDED_OBJECT_CHARACTER:
return contents
string = AXText.get_substring(obj, offset, firstEnd)
return [[firstObj, offset, firstEnd, string]] + contents[1:]
def _getLineContentsAtOffset(self, obj, offset, layoutMode=None, useCache=True):
startTime = time.time()
if not obj:
@@ -1881,6 +1897,13 @@ class Utilities(script_utilities.Utilities):
self._debugContentsInfo(obj, offset, objects, "Line (not layout mode)")
return objects
if self.isLink(obj) and objects and objects[0][0] == obj and objects[0][1] > 0:
if useCache:
self._currentLineContents = objects
self._debugContentsInfo(obj, offset, objects, "Line (wrapped link)")
return objects
firstObj, firstStart, firstEnd, firstString = objects[0]
if (extents[2] == 0 and extents[3] == 0) or self.isMath(firstObj):
extents = self.getExtents(firstObj, firstStart, firstEnd)
@@ -2027,6 +2050,13 @@ class Utilities(script_utilities.Utilities):
tokens = ["WEB: Previous context is: ", obj, ", ", offset]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
if obj == firstObj and 0 <= offset < firstOffset and self.isLink(obj):
contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False)
if contents and contents != line:
msg = "WEB: Using same-link non-layout previous line contents."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return contents
contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache)
if not contents:
tokens = ["WEB: Could not get line contents for ", obj, ", ", offset]
@@ -2092,6 +2122,23 @@ class Utilities(script_utilities.Utilities):
tokens = ["WEB: Next context is: ", obj, ", ", offset]
debug.printTokens(debug.LEVEL_INFO, tokens, True)
if obj == lastObj and offset > lastOffset and self.isLink(obj):
contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False)
if contents and contents != line:
msg = "WEB: Using same-link non-layout next line contents."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return contents
if self.isLink(lastObj):
start, end = self.getHyperlinkRange(lastObj)
if obj == AXObject.get_parent(lastObj) and end >= 0 and offset >= end:
contents = self.getLineContentsAtOffset(obj, offset, layoutMode=False, useCache=False)
contents = self._trimContentsBeforeEmbeddedObjectBoundary(contents, obj, offset)
if contents and contents != line:
msg = "WEB: Using non-layout line after split link."
debug.printMessage(debug.LEVEL_INFO, msg, True)
return contents
contents = self.getLineContentsAtOffset(obj, offset, layoutMode, useCache)
if line == contents:
obj, offset = self.nextContext(obj, offset, True)