Add i3 focus tracking and harden X11 handoff

src/cthulhu/compositor_state_adapter.py

@@ -29,6 +29,7 @@ from .compositor_state_types import (
     RESUME_ATSPI_CHURN,
     WORKSPACE_STATE_CHANGED,
 )
+from .compositor_state_i3 import I3WorkspaceBackend
 from .compositor_state_wayland import NullWorkspaceBackend, WaylandSharedProtocolsBackend
 from .wnck_support import get_session_type
 
@@ -42,6 +43,7 @@ class CompositorStateAdapter:
     ) -> None:
         if workspace_backends is None:
             workspace_backends = [
+                I3WorkspaceBackend(),
                 WaylandSharedProtocolsBackend(),
                 NullWorkspaceBackend(),
             ]
@@ -70,13 +72,15 @@ class CompositorStateAdapter:
             if not self._backend_is_available(backend):
                 continue
             self._workspaceBackend = backend
+            self._snapshot = DesktopContextSnapshot(
+                session_type=self.get_session_type(),
+                backend_name=self._backend_name(),
+            )
             self._backend_activate(backend)
             break
 
-        self._snapshot = DesktopContextSnapshot(
-            session_type=self.get_session_type(),
-            backend_name=self._backend_name(),
-        )
+        if self._workspaceBackend is None:
+            self._snapshot = DesktopContextSnapshot(session_type=self.get_session_type())
         self.sync_accessible_context("activate")
 
     def deactivate(self) -> None:
@@ -104,6 +108,9 @@ class CompositorStateAdapter:
             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"],
+            focused_window_id=workspaceSnapshot.focused_window_id,
+            focused_window_title=workspaceSnapshot.focused_window_title,
+            focused_workspace_empty=workspaceSnapshot.focused_workspace_empty,
         )
 
         previous_token = self._snapshot.active_window_token
@@ -145,27 +152,49 @@ class CompositorStateAdapter:
         signal_name: str,
         workspace_ids: Optional[Iterable[str]] = None,
         reason: str = "",
+        payload: Optional[dict[str, Any]] = None,
     ) -> None:
         if workspace_ids is not None:
+            payload = payload or {}
+            focusedWorkspaceEmpty = payload.get(
+                "focused_workspace_empty",
+                self._snapshot.focused_workspace_empty,
+            )
+            clearAccessibleContext = focusedWorkspaceEmpty is True
             snapshot = replace(
                 self._snapshot,
                 active_workspace_ids=frozenset(str(workspaceId) for workspaceId in workspace_ids if workspaceId),
                 workspace_transition_pending=signal_name == DESKTOP_TRANSITION_STARTED,
+                active_window_token="" if clearAccessibleContext else self._snapshot.active_window_token,
+                locus_of_focus_token="" if clearAccessibleContext else self._snapshot.locus_of_focus_token,
+                active_window_pid=-1 if clearAccessibleContext else self._snapshot.active_window_pid,
+                locus_of_focus_pid=-1 if clearAccessibleContext else self._snapshot.locus_of_focus_pid,
+                active_window_name="" if clearAccessibleContext else self._snapshot.active_window_name,
+                locus_of_focus_name="" if clearAccessibleContext else self._snapshot.locus_of_focus_name,
+                focused_window_id=payload.get(
+                    "focused_window_id",
+                    self._snapshot.focused_window_id,
+                ),
+                focused_window_title=payload.get(
+                    "focused_window_title",
+                    self._snapshot.focused_window_title,
+                ),
+                focused_workspace_empty=focusedWorkspaceEmpty,
             )
             self._snapshot = snapshot
             cthulhu_state.compositorSnapshot = snapshot
 
             if signal_name == DESKTOP_TRANSITION_STARTED:
                 cthulhu_state.pauseAtspiChurn = True
-                self._emit(signal_name, reason, snapshot)
-                self._emit(PAUSE_ATSPI_CHURN, reason, snapshot)
+                self._emit(signal_name, reason, snapshot, payload)
+                self._emit(PAUSE_ATSPI_CHURN, reason, snapshot, payload)
                 return
 
             if signal_name == DESKTOP_TRANSITION_FINISHED:
                 cthulhu_state.pauseAtspiChurn = False
-                self._emit(signal_name, reason, snapshot)
-                self._emit(RESUME_ATSPI_CHURN, reason, snapshot)
-                self._emit(FLUSH_STALE_ATSPI_EVENTS, reason, snapshot)
+                self._emit(signal_name, reason, snapshot, payload)
+                self._emit(RESUME_ATSPI_CHURN, reason, snapshot, payload)
+                self._emit(FLUSH_STALE_ATSPI_EVENTS, reason, snapshot, payload)
                 return
 
             if signal_name == PAUSE_ATSPI_CHURN:
@@ -173,7 +202,7 @@ class CompositorStateAdapter:
             elif signal_name == RESUME_ATSPI_CHURN:
                 cthulhu_state.pauseAtspiChurn = False
 
-            self._emit(signal_name, reason, snapshot)
+            self._emit(signal_name, reason, snapshot, payload)
             return
 
         normalized = (signal_name or "").strip().lower()

src/cthulhu/compositor_state_i3.py

@@ -0,0 +1,259 @@
+#!/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.
+
+"""Optional i3 IPC backend for authoritative X11 workspace and focus state."""
+
+from __future__ import annotations
+
+import importlib
+import threading
+from collections.abc import Mapping
+from typing import Any, Optional
+
+from gi.repository import GLib
+
+from . import debug
+from .compositor_state_types import (
+    DESKTOP_TRANSITION_FINISHED,
+    DESKTOP_TRANSITION_STARTED,
+    WORKSPACE_STATE_CHANGED,
+)
+from .wnck_support import get_session_type
+
+
+class I3WorkspaceBackend:
+    """Tracks the focused i3 workspace and its real foreground X11 window."""
+
+    name = "i3-ipc"
+
+    def __init__(self, *, i3ipc_module: Any = None) -> None:
+        self._i3ipc = i3ipc_module
+        self._connection: Any = None
+        self._emitSignal: Any = None
+        self._eventThread: Optional[threading.Thread] = None
+        self._active = False
+        self._lastContext: Optional[tuple[Any, ...]] = None
+
+    def is_available(self, session_type: str | None = None) -> bool:
+        effectiveSessionType = (session_type or get_session_type()).strip().lower()
+        if effectiveSessionType != "x11":
+            return False
+
+        if self._i3ipc is None:
+            try:
+                self._i3ipc = importlib.import_module("i3ipc")
+            except ImportError:
+                return False
+
+        return self._ensure_connection()
+
+    def activate(self, emit_signal: Any = None) -> None:
+        self.deactivate()
+        self._emitSignal = emit_signal
+        if emit_signal is None or not self.is_available():
+            return
+
+        self._active = True
+        self._subscribe()
+        self._refresh_context("activate", transition=False)
+        self._eventThread = threading.Thread(
+            target=self._run_event_loop,
+            name="cthulhu-i3-ipc",
+            daemon=True,
+        )
+        self._eventThread.start()
+
+    def deactivate(self, emit_signal: Any = None) -> None:
+        self._active = False
+        connection = self._connection
+        if connection is not None:
+            quitLoop = getattr(connection, "main_quit", None)
+            if callable(quitLoop):
+                try:
+                    quitLoop()
+                except Exception:
+                    pass
+
+            commandSocket = getattr(connection, "_cmd_socket", None)
+            if commandSocket is not None:
+                try:
+                    commandSocket.close()
+                except Exception:
+                    pass
+
+        self._connection = None
+        self._eventThread = None
+        self._emitSignal = None
+        self._lastContext = None
+
+    def _ensure_connection(self) -> bool:
+        if self._connection is not None:
+            return True
+        if self._i3ipc is None:
+            return False
+
+        connectionClass = getattr(self._i3ipc, "Connection", None)
+        if not callable(connectionClass):
+            return False
+
+        try:
+            try:
+                self._connection = connectionClass(auto_reconnect=True)
+            except TypeError:
+                self._connection = connectionClass()
+            self._connection.get_workspaces()
+        except Exception as error:
+            self._connection = None
+            msg = f"COMPOSITOR STATE: i3 IPC unavailable: {error}"
+            debug.printMessage(debug.LEVEL_INFO, msg, True)
+            return False
+
+        return True
+
+    def _subscribe(self) -> None:
+        if self._connection is None or self._i3ipc is None:
+            return
+
+        onEvent = getattr(self._connection, "on", None)
+        eventType = getattr(self._i3ipc, "Event", None)
+        if not callable(onEvent) or eventType is None:
+            return
+
+        onEvent(eventType.WORKSPACE, self._handle_ipc_event)
+        onEvent(eventType.WINDOW, self._handle_ipc_event)
+
+    def _run_event_loop(self) -> None:
+        connection = self._connection
+        runLoop = getattr(connection, "main", None)
+        if not callable(runLoop):
+            return
+
+        try:
+            runLoop()
+        except Exception as error:
+            if self._active:
+                msg = f"COMPOSITOR STATE: i3 IPC event loop stopped: {error}"
+                debug.printMessage(debug.LEVEL_WARNING, msg, True)
+
+    def _handle_ipc_event(self, _connection: Any, event: Any) -> None:
+        if not self._active:
+            return
+
+        change = str(getattr(event, "change", "") or "changed")
+        GLib.idle_add(self._refresh_context, f"i3-{change}", True)
+
+    def _query_context(self) -> tuple[set[str], dict[str, Any]]:
+        if self._connection is None:
+            raise RuntimeError("i3 IPC is not connected")
+
+        workspaceReplies = self._connection.get_workspaces()
+        focusedReply = next(
+            (workspace for workspace in workspaceReplies if bool(getattr(workspace, "focused", False))),
+            None,
+        )
+        if focusedReply is None:
+            raise RuntimeError("i3 did not report a focused workspace")
+
+        workspaceName = str(getattr(focusedReply, "name", "") or "").strip()
+        if not workspaceName:
+            raise RuntimeError("i3 focused workspace has no name")
+
+        tree = self._connection.get_tree()
+        focusedWorkspace = next(
+            (
+                workspace
+                for workspace in tree.workspaces()
+                if str(getattr(workspace, "name", "") or "") == workspaceName
+            ),
+            None,
+        )
+        if focusedWorkspace is None:
+            raise RuntimeError(f"i3 tree is missing focused workspace {workspaceName}")
+
+        windowNodes = [
+            node
+            for node in focusedWorkspace
+            if self._positive_window_id(getattr(node, "window", None)) is not None
+        ]
+        focusedNode = tree.find_focused()
+        focusedWindowId: Optional[int] = None
+        focusedWindowTitle = ""
+        if focusedNode is not None and focusedNode.workspace() is focusedWorkspace:
+            focusedWindowId = self._positive_window_id(getattr(focusedNode, "window", None))
+            if focusedWindowId is not None:
+                focusedWindowTitle = str(
+                    getattr(focusedNode, "window_title", None)
+                    or getattr(focusedNode, "name", None)
+                    or ""
+                ).strip()
+
+        payload = {
+            "focused_window_id": focusedWindowId,
+            "focused_window_title": focusedWindowTitle,
+            "focused_workspace_empty": not windowNodes,
+        }
+        return {workspaceName}, payload
+
+    @staticmethod
+    def _positive_window_id(value: Any) -> Optional[int]:
+        try:
+            windowId = int(value)
+        except (TypeError, ValueError):
+            return None
+        return windowId if windowId > 0 else None
+
+    def _refresh_context(self, reason: str, transition: bool = True) -> bool:
+        if not self._active:
+            return False
+
+        try:
+            workspaceIds, payload = self._query_context()
+        except Exception as error:
+            msg = f"COMPOSITOR STATE: Could not query i3 context: {error}"
+            debug.printMessage(debug.LEVEL_INFO, msg, True)
+            unknownPayload = {
+                "focused_window_id": None,
+                "focused_window_title": "",
+                "focused_workspace_empty": None,
+            }
+            self._lastContext = None
+            self._emit_signal(
+                WORKSPACE_STATE_CHANGED,
+                set(),
+                f"{reason}: i3-context-unknown",
+                unknownPayload,
+            )
+            return False
+
+        context = (
+            frozenset(workspaceIds),
+            payload["focused_window_id"],
+            payload["focused_window_title"],
+            payload["focused_workspace_empty"],
+        )
+        if context == self._lastContext:
+            return False
+        self._lastContext = context
+
+        if transition:
+            self._emit_signal(DESKTOP_TRANSITION_STARTED, workspaceIds, reason, payload)
+        self._emit_signal(WORKSPACE_STATE_CHANGED, workspaceIds, reason, payload)
+        if transition:
+            self._emit_signal(DESKTOP_TRANSITION_FINISHED, workspaceIds, reason, payload)
+        return False
+
+    def _emit_signal(
+        self,
+        signalType: str,
+        workspaceIds: set[str],
+        reason: str,
+        payload: Mapping[str, Any],
+    ) -> None:
+        if callable(self._emitSignal):
+            self._emitSignal(signalType, workspaceIds, reason, dict(payload))

src/cthulhu/compositor_state_types.py

@@ -38,6 +38,9 @@ class DesktopContextSnapshot:
     locus_of_focus_pid: int = -1
     active_window_name: str = ""
     locus_of_focus_name: str = ""
+    focused_window_id: Optional[int] = None
+    focused_window_title: str = ""
+    focused_workspace_empty: Optional[bool] = None
 
 
 @dataclass(frozen=True, slots=True)

src/cthulhu/event_manager.py

@@ -60,6 +60,7 @@ from .compositor_state_types import (
     PAUSE_ATSPI_CHURN,
     PRIORITIZE_FOCUS,
     RESUME_ATSPI_CHURN,
+    WORKSPACE_STATE_CHANGED,
 )
 
 if TYPE_CHECKING:
@@ -113,6 +114,7 @@ class EventManager:
         self._compositorStateAdapter: Optional[Any] = None
         self._churnSuppressed: bool = cthulhu_state.pauseAtspiChurn
         self._prioritizedContextToken: Optional[str] = cthulhu_state.prioritizedDesktopContextToken
+        self._desktopContextConfirmedEmpty: bool = False
         self._relevanceBurstWindow: float = 0.15
         self._relevanceBurstHistory: Dict[Tuple[str, str, str], float] = {}
 
@@ -292,6 +294,9 @@ class EventManager:
             if not contextToken:
                 contextToken = snapshotToken or None
 
+        if signalType in (WORKSPACE_STATE_CHANGED, DESKTOP_TRANSITION_FINISHED):
+            self._update_i3_workspace_context(snapshot)
+
         if signalType in (DESKTOP_FOCUS_CONTEXT_CHANGED, PRIORITIZE_FOCUS):
             self._prioritizedContextToken = contextToken
             cthulhu_state.prioritizedDesktopContextToken = contextToken
@@ -318,6 +323,43 @@ class EventManager:
         if signalType == FLUSH_STALE_ATSPI_EVENTS:
             self._flush_stale_atspi_events()
 
+    def _update_i3_workspace_context(self, snapshot: Any) -> None:
+        """Clears stale accessible state when i3 confirms a blank workspace."""
+
+        if snapshot is None or snapshot.backend_name != "i3-ipc":
+            return
+
+        workspaceEmpty = snapshot.focused_workspace_empty
+        if workspaceEmpty is None:
+            return
+
+        if not workspaceEmpty:
+            if self._desktopContextConfirmedEmpty:
+                self._desktopContextConfirmedEmpty = False
+                GLib.idle_add(self._sync_focus_on_startup)
+            return
+
+        if self._desktopContextConfirmedEmpty:
+            return
+
+        self._desktopContextConfirmedEmpty = True
+        activeScript = self.app.scriptManager.get_active_script()
+        if activeScript is not None:
+            activeScript.presentationInterrupt()
+
+        manager = focus_manager.get_manager()
+        if manager is not None:
+            manager.clear_state("confirmed empty i3 workspace")
+        cthulhu_state.pendingSelfHostedFocus = None
+        self._prioritizedContextToken = None
+        cthulhu_state.prioritizedDesktopContextToken = None
+
+        defaultScript = self.app.scriptManager.get_default_script()
+        self.app.scriptManager.set_active_script(
+            defaultScript,
+            "focus: confirmed-empty-i3-workspace",
+        )
+
     def ignoreEventTypes(self, eventTypeList: List[str]) -> None:
         for eventType in eventTypeList:
             if eventType not in self._ignoredEvents:
@@ -1771,6 +1813,12 @@ class EventManager:
         debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True)
         eType = event.type
 
+        if self._desktopContextConfirmedEmpty \
+           and not self._is_notification_event_during_xterm_handoff(event):
+            msg = "EVENT MANAGER: Ignoring AT-SPI event on confirmed empty i3 workspace."
+            debug.printMessage(debug.LEVEL_INFO, msg, True)
+            return
+
         refreshXtermHandoff = self._event_refreshes_xterm_handoff(event)
         isNotification = self._is_notification_event_during_xterm_handoff(event)
         suppressForXterm = not isNotification and self._inputEventManager is not None and (

src/cthulhu/input_event_manager.py

@@ -125,12 +125,15 @@ class InputEventManager:
         self._netActiveWindowAtom = None
         self._x11AnyPropertyType = None
         self._didAttemptX11Display: bool = False
+        self._lastEwmhReadWindowId: Optional[int] = None
         self._lastEwmhActiveWindowId: Optional[int] = None
         self._lastEwmhXtermMatch: Optional[bool] = None
         self._scriptWithSuspendedGrabsForXterm = None
         self._xtermFocusScreen = None
         self._xtermFocusHandlerId: int = 0
         self._xtermRecoverySourceId: int = 0
+        self._xtermRecoveryPollCount: int = 0
+        self._xtermHandoffHistory: List[str] = []
 
     def activate_device(self) -> Atspi.Device:
         """Creates and returns the AT-SPI device used by this manager."""
@@ -229,6 +232,14 @@ class InputEventManager:
             "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None,
             "xterm_focus_monitor_active": self._xtermFocusHandlerId != 0,
             "xterm_recovery_timer_active": self._xtermRecoverySourceId != 0,
+            "xterm_recovery_source_present": self._xterm_recovery_source_is_present(),
+            "xterm_recovery_poll_count": self._xtermRecoveryPollCount,
+            "xterm_handoff_history": " | ".join(self._xtermHandoffHistory),
+            "last_ewmh_read_window_id": (
+                hex(self._lastEwmhReadWindowId)
+                if self._lastEwmhReadWindowId is not None
+                else ""
+            ),
             "last_ewmh_active_window_id": (
                 hex(self._lastEwmhActiveWindowId)
                 if self._lastEwmhActiveWindowId is not None
@@ -578,22 +589,7 @@ class InputEventManager:
     def _get_ewmh_active_x11_window_id(self) -> Optional[int]:
         """Returns _NET_ACTIVE_WINDOW from the X11 root window when available."""
 
-        if not self._didAttemptX11Display:
-            self._didAttemptX11Display = True
-            try:
-                from Xlib import X  # pylint: disable=import-outside-toplevel
-                from Xlib import display as xDisplay  # pylint: disable=import-outside-toplevel
-
-                self._x11Display = xDisplay.Display()
-                self._x11Root = self._x11Display.screen().root
-                self._netActiveWindowAtom = self._x11Display.intern_atom("_NET_ACTIVE_WINDOW")
-                self._x11AnyPropertyType = X.AnyPropertyType
-            except Exception as error:
-                msg = f"INPUT EVENT MANAGER: EWMH active-window lookup unavailable: {error}"
-                debug.print_message(debug.LEVEL_INFO, msg, True)
-                self._close_x11_display()
-
-        if self._x11Root is None or self._netActiveWindowAtom is None:
+        if not self._ensure_x11_display():
             return None
 
         try:
@@ -612,6 +608,33 @@ class InputEventManager:
             self._didAttemptX11Display = False
             return None
 
+    def _ensure_x11_display(self) -> bool:
+        """Initializes the optional exact-X11 metadata connection."""
+
+        if self._x11Display is not None:
+            return True
+
+        if not self._didAttemptX11Display:
+            self._didAttemptX11Display = True
+            try:
+                from Xlib import X  # pylint: disable=import-outside-toplevel
+                from Xlib import display as xDisplay  # pylint: disable=import-outside-toplevel
+
+                self._x11Display = xDisplay.Display()
+                self._x11Root = self._x11Display.screen().root
+                self._netActiveWindowAtom = self._x11Display.intern_atom("_NET_ACTIVE_WINDOW")
+                self._x11AnyPropertyType = X.AnyPropertyType
+            except Exception as error:
+                msg = f"INPUT EVENT MANAGER: EWMH active-window lookup unavailable: {error}"
+                debug.print_message(debug.LEVEL_INFO, msg, True)
+                self._close_x11_display()
+
+        return (
+            self._x11Display is not None
+            and self._x11Root is not None
+            and self._netActiveWindowAtom is not None
+        )
+
     def _close_x11_display(self) -> None:
         """Closes the optional X11 display used for EWMH focus lookup."""
 
@@ -620,6 +643,7 @@ class InputEventManager:
         self._x11Root = None
         self._netActiveWindowAtom = None
         self._x11AnyPropertyType = None
+        self._lastEwmhReadWindowId = None
         self._lastEwmhActiveWindowId = None
         self._lastEwmhXtermMatch = None
         if x11Display is None:
@@ -764,9 +788,22 @@ class InputEventManager:
             self._poll_xterm_grab_recovery,
         )
 
+    def _xterm_recovery_source_is_present(self) -> bool:
+        """Returns whether GLib still owns the recorded recovery source."""
+
+        if not self._xtermRecoverySourceId:
+            return False
+
+        try:
+            context = GLib.MainContext.default()
+            return context.find_source_by_id(self._xtermRecoverySourceId) is not None
+        except Exception:
+            return False
+
     def _poll_xterm_grab_recovery(self) -> bool:
         """Updates XTerm handoff state from the authoritative EWMH active window."""
 
+        self._xtermRecoveryPollCount += 1
         if self._device is None:
             self._xtermRecoverySourceId = 0
             return False
@@ -782,6 +819,15 @@ class InputEventManager:
             debug.print_message(debug.LEVEL_INFO, msg, True)
         return True
 
+    def _record_xterm_handoff_action(self, action: str) -> None:
+        """Retains a short history of XTerm focus and grab-state transitions."""
+
+        if self._xtermHandoffHistory and self._xtermHandoffHistory[-1] == action:
+            return
+
+        self._xtermHandoffHistory.append(action)
+        self._xtermHandoffHistory = self._xtermHandoffHistory[-12:]
+
     def _on_active_x11_window_changed(self, screen: Any, _previousWindow: Any) -> None:
         """Updates grab suspension as soon as X11 focus changes."""
 
@@ -799,7 +845,24 @@ class InputEventManager:
     def _active_x11_window_xterm_match(self) -> Optional[bool]:
         """Returns whether the active X11 window is XTerm, or None when unknown."""
 
-        windowId = self._get_ewmh_active_x11_window_id()
+        snapshot = cthulhu_state.compositorSnapshot
+        usingI3Context = snapshot is not None and snapshot.backend_name == "i3-ipc"
+        if usingI3Context and snapshot.focused_workspace_empty is True:
+            self._record_xterm_handoff_action("i3:confirmed-empty-workspace")
+            self._lastEwmhReadWindowId = None
+            return False
+
+        if usingI3Context and snapshot.focused_window_id is not None:
+            windowId = snapshot.focused_window_id
+        else:
+            windowId = self._get_ewmh_active_x11_window_id()
+
+        previousWindowId = self._lastEwmhReadWindowId
+        if windowId != previousWindowId:
+            previousText = hex(previousWindowId) if previousWindowId is not None else "unknown"
+            currentText = hex(windowId) if windowId is not None else "unknown"
+            self._record_xterm_handoff_action(f"ewmh:{previousText}->{currentText}")
+        self._lastEwmhReadWindowId = windowId
         if (
             windowId is not None
             and windowId == self._lastEwmhActiveWindowId
@@ -807,8 +870,10 @@ class InputEventManager:
         ):
             return self._lastEwmhXtermMatch
 
-        window = self._get_active_x11_window(ewmhWindowId=windowId)
-        match = self._x11_window_xterm_match(window)
+        if windowId is None:
+            return None
+
+        match = self._x11_window_id_xterm_match(windowId)
         if windowId is not None and match is not None:
             self._lastEwmhActiveWindowId = windowId
             self._lastEwmhXtermMatch = match
@@ -870,6 +935,80 @@ class InputEventManager:
 
         return self._identifier_is_xterm(executable)
 
+    def _get_x11_window_metadata(
+        self,
+        windowId: int,
+    ) -> Optional[Tuple[List[str], int]]:
+        """Returns identifiers and pid directly from the exact X11 window id."""
+
+        if not self._ensure_x11_display():
+            return None
+
+        try:
+            window = self._x11Display.create_resource_object("window", windowId)
+            identifiers = [
+                value
+                for value in (window.get_wm_class() or ())
+                if isinstance(value, str) and value.strip()
+            ]
+
+            name = window.get_wm_name()
+            if isinstance(name, str) and name.strip():
+                identifiers.append(name)
+
+            netWmNameAtom = self._x11Display.intern_atom("_NET_WM_NAME")
+            utf8StringAtom = self._x11Display.intern_atom("UTF8_STRING")
+            nameProperty = window.get_full_property(netWmNameAtom, utf8StringAtom)
+            if nameProperty is not None and nameProperty.value:
+                value = nameProperty.value
+                if isinstance(value, bytes):
+                    value = value.decode(errors="ignore")
+                if isinstance(value, str) and value.strip():
+                    identifiers.append(value)
+
+            netWmPidAtom = self._x11Display.intern_atom("_NET_WM_PID")
+            pidProperty = window.get_full_property(
+                netWmPidAtom,
+                self._x11AnyPropertyType,
+            )
+            pid = -1
+            if pidProperty is not None and pidProperty.value:
+                pid = int(pidProperty.value[0])
+        except Exception as error:
+            msg = (
+                "INPUT EVENT MANAGER: Could not read metadata for X11 window "
+                f"{hex(windowId)}: {error}"
+            )
+            debug.print_message(debug.LEVEL_INFO, msg, True)
+            return None
+
+        return identifiers, pid
+
+    def _x11_window_id_xterm_match(self, windowId: int) -> Optional[bool]:
+        """Returns whether the exact EWMH X11 window id identifies XTerm."""
+
+        metadata = self._get_x11_window_metadata(windowId)
+        if metadata is None:
+            return None
+
+        identifiers, pid = metadata
+        if any(self._identifier_is_xterm(value) for value in identifiers):
+            return True
+
+        if pid > 0:
+            try:
+                with open(f"/proc/{pid}/cmdline", "rb") as cmdlineFile:
+                    executable = cmdlineFile.read().split(b"\0", 1)[0].decode(errors="ignore")
+                if self._identifier_is_xterm(executable):
+                    return True
+            except OSError:
+                pass
+
+        if identifiers or pid > 0:
+            return False
+
+        return None
+
     def _active_x11_window_is_xterm(self) -> bool:
         """Returns True when the active X11 window appears to be XTerm."""
 
@@ -1043,7 +1182,17 @@ class InputEventManager:
         """Suspends active-script key grabs so XTerm/Fenrir can receive them."""
 
         script = script_manager.get_manager().get_active_script()
-        if script is None or script is self._scriptWithSuspendedGrabsForXterm:
+        if script is None:
+            self._record_xterm_handoff_action("suspend:skipped-no-active-script")
+            return
+
+        if script is self._scriptWithSuspendedGrabsForXterm:
+            grabIds = getattr(script, "grab_ids", [])
+            scriptGrabCount = len(grabIds) if isinstance(grabIds, list) else -1
+            self._record_xterm_handoff_action(
+                f"suspend:already-marked:script-grabs={scriptGrabCount}:"
+                f"all-grabs={len(self._grabbed_bindings)}"
+            )
             return
 
         if self._scriptWithSuspendedGrabsForXterm is not None:
@@ -1060,6 +1209,9 @@ class InputEventManager:
         debug.print_tokens(debug.LEVEL_INFO, tokens, True)
         removeGrabs()
         self._scriptWithSuspendedGrabsForXterm = script
+        self._record_xterm_handoff_action(
+            f"suspend:success:all-grabs={len(self._grabbed_bindings)}"
+        )
         self._ensure_xterm_recovery_timer()
 
     def _restore_key_grabs_after_xterm(self) -> None:
@@ -1071,6 +1223,7 @@ class InputEventManager:
 
         activeScript = script_manager.get_manager().get_active_script()
         if script is not activeScript:
+            self._record_xterm_handoff_action("restore:dropped-active-script-changed")
             tokens = [
                 "INPUT EVENT MANAGER: Not restoring XTerm-suspended key grabs; active script changed:",
                 script,
@@ -1092,6 +1245,7 @@ class InputEventManager:
         try:
             addGrabs()
         except Exception as error:
+            self._record_xterm_handoff_action(f"restore:failed:{type(error).__name__}")
             msg = f"INPUT EVENT MANAGER: Could not restore XTerm-suspended key grabs: {error}"
             debug.print_message(debug.LEVEL_INFO, msg, True)
             removeGrabs = getattr(script, "removeKeyGrabs", None)
@@ -1110,6 +1264,9 @@ class InputEventManager:
             return
 
         self._scriptWithSuspendedGrabsForXterm = None
+        self._record_xterm_handoff_action(
+            f"restore:success:all-grabs={len(self._grabbed_bindings)}"
+        )
 
     # pylint: disable=too-many-arguments
     # pylint: disable=too-many-positional-arguments

src/cthulhu/meson.build

@@ -34,6 +34,7 @@ cthulhu_python_sources = files([
   'cmdnames.py',
   'colornames.py',
   'compositor_state_adapter.py',
+  'compositor_state_i3.py',
   'compositor_state_types.py',
   'compositor_state_wayland.py',
   'common_keyboardmap.py',

src/cthulhu/plugins/WindowTitleReader/plugin.py

@@ -80,7 +80,7 @@ class WindowTitleReader(Plugin):
             self,
             overwrite=True,
         )
-        if xlibAvailable:
+        if xlibAvailable or self._get_i3_title_context() is not None:
             self._start_tracking()
         self._activated = True
         debug.printMessage(debug.LEVEL_INFO, "WindowTitleReader: Activated", True)
@@ -136,7 +136,7 @@ class WindowTitleReader(Plugin):
             self._present_message("Window title reader off")
             return True
 
-        if not xlibAvailable:
+        if not xlibAvailable and self._get_i3_title_context() is None:
             self._present_message("Window title reader unavailable")
             debug.printMessage(
                 debug.LEVEL_INFO,
@@ -167,7 +167,7 @@ class WindowTitleReader(Plugin):
                     self._present_message("Window title reader on")
                 return True
 
-            if not xlibAvailable:
+            if not xlibAvailable and self._get_i3_title_context() is None:
                 if notify_user:
                     self._present_message("Window title reader unavailable")
                 debug.printMessage(
@@ -218,9 +218,9 @@ class WindowTitleReader(Plugin):
             return True
 
         try:
-            self._display = xDisplay.Display()
-            self._root = self._display.screen().root
-            self._init_atoms()
+            if self._get_i3_title_context() is None:
+                if not self._ensure_x11_display():
+                    return False
             self._pollSourceId = GLib.timeout_add(self._pollIntervalMs, self._poll_window_title)
             self._poll_window_title()
             return True
@@ -273,17 +273,24 @@ class WindowTitleReader(Plugin):
             return False
 
         try:
-            activeWindow = self._get_active_window()
-            if not activeWindow:
+            i3Context = self._get_i3_title_context()
+            if i3Context is not None:
+                activeWindowId, windowTitle = i3Context
+            else:
+                if not self._ensure_x11_display():
+                    self._lastTitle = None
+                    return True
+                activeWindow = self._get_active_window()
+                if not activeWindow:
+                    self._lastTitle = None
+                    return True
+                activeWindowId = activeWindow.id
+                windowTitle = self._get_current_title(activeWindow)
+
+            if activeWindowId is None or not windowTitle:
                 self._lastTitle = None
                 return True
 
-            windowTitle = self._get_current_title(activeWindow)
-            if not windowTitle:
-                self._lastTitle = None
-                return True
-
-            activeWindowId = activeWindow.id
             if self._lastActiveWindowId is None:
                 self._lastActiveWindowId = activeWindowId
             elif activeWindowId != self._lastActiveWindowId:
@@ -321,12 +328,25 @@ class WindowTitleReader(Plugin):
     def get_fallback_title(self, atspiTitle=None):
         """Returns the X11 title when AT-SPI has not exposed an equivalent title."""
 
+        i3Context = self._get_i3_title_context()
+        if i3Context is not None:
+            _windowId, titleText = i3Context
+            if not titleText:
+                return ""
+            if atspiTitle is None:
+                atspiTitle = self._get_atspi_title()
+            if self._titles_match(atspiTitle, titleText):
+                return ""
+            return titleText
+
         if not xlibAvailable and self._display is None:
             return ""
 
         startedTracking = self._pollSourceId is not None
         if not startedTracking and not self._start_tracking():
             return ""
+        if not self._ensure_x11_display():
+            return ""
 
         activeWindow = self._get_active_window()
         titleText = self._get_current_title(activeWindow) if activeWindow else ""
@@ -341,6 +361,65 @@ class WindowTitleReader(Plugin):
 
         return titleText
 
+    def _get_i3_title_context(self):
+        """Returns authoritative i3 window id/title, or None when i3 is not active."""
+
+        if not self.app:
+            return None
+
+        try:
+            adapter = self.app.getCompositorStateAdapter()
+            snapshot = adapter.get_snapshot() if adapter else None
+        except Exception:
+            return None
+
+        if snapshot is None or getattr(snapshot, "backend_name", "") != "i3-ipc":
+            return None
+
+        if getattr(snapshot, "focused_workspace_empty", None) is True:
+            return (None, "")
+
+        windowId = getattr(snapshot, "focused_window_id", None)
+        titleText = str(getattr(snapshot, "focused_window_title", "") or "").strip()
+        if windowId is None:
+            return None
+
+        if not titleText or self._is_wine_desktop_title(titleText):
+            if not self._ensure_x11_display():
+                return None
+            try:
+                activeWindow = self._display.create_resource_object("window", windowId)
+            except Exception:
+                return None
+            titleText = self._get_current_title(activeWindow)
+            if not titleText:
+                return None
+
+        return (windowId, titleText)
+
+    def _ensure_x11_display(self):
+        """Lazily opens the generic X11 title source when i3 IPC is unavailable."""
+
+        if self._display is not None:
+            return True
+        if not xlibAvailable:
+            return False
+
+        try:
+            self._display = xDisplay.Display()
+            self._root = self._display.screen().root
+            self._init_atoms()
+        except Exception as error:
+            debug.printMessage(
+                debug.LEVEL_INFO,
+                f"WindowTitleReader: Failed to open X11 fallback: {error}",
+                True,
+            )
+            self._cleanup_display()
+            return False
+
+        return True
+
     def _get_atspi_title(self):
         if not self.app:
             return ""
@@ -386,9 +465,11 @@ class WindowTitleReader(Plugin):
 
     def _get_wine_desktop_title(self, desktopWindow):
         focusWindow = self._get_focus_window()
-        if focusWindow and focusWindow.id != desktopWindow.id:
+        if focusWindow \
+           and focusWindow.id != desktopWindow.id \
+           and self._window_is_descendant_of(focusWindow, desktopWindow):
             focusTitle = self._get_window_title(focusWindow)
-            if focusTitle and not self._is_wine_desktop_title(focusTitle):
+            if self._is_usable_wine_child_title(focusWindow, focusTitle):
                 return focusTitle
 
         return self._find_child_title(desktopWindow)
@@ -404,7 +485,7 @@ class WindowTitleReader(Plugin):
             window = childQueue.pop(0)
             checkedCount += 1
             titleText = self._get_window_title(window)
-            if titleText and not self._is_wine_desktop_title(titleText):
+            if self._is_usable_wine_child_title(window, titleText):
                 return titleText
 
             try:
@@ -414,6 +495,47 @@ class WindowTitleReader(Plugin):
 
         return ""
 
+    def _window_is_descendant_of(self, window, ancestor):
+        if not window or not ancestor:
+            return False
+
+        current = window
+        for _hop in range(64):
+            if current.id == ancestor.id:
+                return True
+            try:
+                parent = current.query_tree().parent
+            except xError.XError:
+                return False
+            if not parent or parent.id == current.id:
+                return False
+            current = parent
+
+        return False
+
+    def _is_usable_wine_child_title(self, window, titleText):
+        if not titleText or self._is_wine_desktop_title(titleText):
+            return False
+
+        normalizedTitle = " ".join(titleText.casefold().split())
+        if normalizedTitle in ("default ime", "program manager"):
+            return False
+
+        try:
+            windowClass = window.get_wm_class() or ()
+        except xError.XError:
+            windowClass = ()
+        normalizedClass = {
+            str(value or "").strip().casefold()
+            for value in windowClass
+        }
+        if "cthulhu-wine-access.exe" in normalizedClass:
+            return False
+        if normalizedTitle == "nvda" and "nvda.exe" in normalizedClass:
+            return False
+
+        return True
+
     def _is_wine_desktop_title(self, titleText):
         return self._wineDesktopLabel.lower() in titleText.lower()
 

src/cthulhu/scripts/web/script.py

@@ -2660,20 +2660,6 @@ 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)