Tighten up relase and reclaim clode for terminals.

src/cthulhu/compositor_state_adapter.py

@@ -63,6 +63,15 @@ class CompositorStateAdapter:
     def get_snapshot(self) -> DesktopContextSnapshot:
         return self._snapshot
 
+    def refresh_workspace_context(self, reason: str) -> DesktopContextSnapshot:
+        """Refreshes and returns the selected backend's current workspace context."""
+
+        backend = self._workspaceBackend
+        refreshContext = getattr(backend, "refresh_context", None)
+        if callable(refreshContext):
+            refreshContext(reason)
+        return self._snapshot
+
     def get_session_type(self) -> str:
         return get_session_type()
 

src/cthulhu/compositor_state_i3.py

@@ -92,6 +92,11 @@ class I3WorkspaceBackend:
         self._emitSignal = None
         self._lastContext = None
 
+    def refresh_context(self, reason: str) -> None:
+        """Refreshes the cached i3 focus context without emitting a transition."""
+
+        self._refresh_context(reason, transition=False)
+
     def _ensure_connection(self) -> bool:
         if self._connection is not None:
             return True

src/cthulhu/event_manager.py

@@ -174,6 +174,7 @@ class EventManager:
 
         debug.printMessage(debug.LEVEL_INFO, 'EVENT MANAGER: Activating keyboard handling', True)
         self._inputEventManager = input_event_manager.get_manager()
+        self._inputEventManager.set_compositor_state_adapter(self._compositorStateAdapter)
         self._inputEventManager.start_key_watcher()
         cthulhu_state.device = self._inputEventManager._device
         self._keyHandlingActive = True
@@ -277,6 +278,8 @@ class EventManager:
             self._compositorStateAdapter.remove_listener(self._handle_compositor_signal)
 
         self._compositorStateAdapter = adapter
+        if self._inputEventManager is not None:
+            self._inputEventManager.set_compositor_state_adapter(adapter)
         if adapter is not None and hasattr(adapter, "add_listener"):
             adapter.add_listener(self._handle_compositor_signal)
 

src/cthulhu/input_event_manager.py

@@ -134,6 +134,12 @@ class InputEventManager:
         self._xtermRecoverySourceId: int = 0
         self._xtermRecoveryPollCount: int = 0
         self._xtermHandoffHistory: List[str] = []
+        self._compositorStateAdapter: Optional[Any] = None
+
+    def set_compositor_state_adapter(self, adapter: Optional[Any]) -> None:
+        """Stores the compositor adapter used to refresh XTerm handoff state."""
+
+        self._compositorStateAdapter = adapter
 
     def activate_device(self) -> Atspi.Device:
         """Creates and returns the AT-SPI device used by this manager."""
@@ -847,9 +853,17 @@ class InputEventManager:
 
         snapshot = cthulhu_state.compositorSnapshot
         usingI3Context = snapshot is not None and snapshot.backend_name == "i3-ipc"
+        if usingI3Context and self._lastEwmhXtermMatch is True:
+            adapter = self._compositorStateAdapter
+            refreshContext = getattr(adapter, "refresh_workspace_context", None)
+            if callable(refreshContext):
+                snapshot = refreshContext("xterm-handoff-check")
+                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
+            self._lastEwmhActiveWindowId = None
+            self._lastEwmhXtermMatch = False
             return False
 
         if usingI3Context and snapshot.focused_window_id is not None:

tests/test_compositor_state_adapter_regressions.py

@@ -42,6 +42,7 @@ class FakeWorkspaceBackend:
         self.name = name
         self.activate_calls = []
         self.deactivate_calls = []
+        self.refresh_calls = []
 
     def is_available(self, session_type: str | None = None) -> bool:
         return self.available
@@ -52,6 +53,9 @@ class FakeWorkspaceBackend:
     def deactivate(self, emit_signal=None) -> None:
         self.deactivate_calls.append(emit_signal)
 
+    def refresh_context(self, reason: str) -> None:
+        self.refresh_calls.append(reason)
+
 
 class CompositorStateAdapterRegressionTests(unittest.TestCase):
     def setUp(self) -> None:
@@ -111,6 +115,16 @@ class CompositorStateAdapterRegressionTests(unittest.TestCase):
         self.assertEqual(snapshot.focused_window_title, "9 devel")
         self.assertFalse(snapshot.focused_workspace_empty)
 
+    def test_refresh_workspace_context_queries_selected_backend(self) -> None:
+        backend = FakeWorkspaceBackend(True, "i3-ipc")
+        adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[backend])
+        adapter.activate()
+
+        snapshot = adapter.refresh_workspace_context("xterm-handoff-check")
+
+        self.assertEqual(backend.refresh_calls, ["xterm-handoff-check"])
+        self.assertIs(snapshot, adapter.get_snapshot())
+
     def test_confirmed_empty_workspace_clears_stale_accessible_context(self) -> None:
         backend = FakeWorkspaceBackend(True, "i3-ipc")
         adapter = compositor_state_adapter.CompositorStateAdapter(workspace_backends=[backend])

tests/test_compositor_state_i3_regressions.py

@@ -153,6 +153,26 @@ class I3WorkspaceBackendRegressionTests(unittest.TestCase):
             ],
         )
 
+    def test_public_refresh_emits_updated_context_without_transition(self) -> None:
+        backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace())
+        backend._active = True
+        backend._emitSignal = mock.Mock()
+        payload = {
+            "focused_window_id": 0x2600003,
+            "focused_window_title": "Browser",
+            "focused_workspace_empty": False,
+        }
+
+        with mock.patch.object(backend, "_query_context", return_value=({"8"}, payload)):
+            backend.refresh_context("xterm-handoff-check")
+
+        backend._emitSignal.assert_called_once_with(
+            compositor_state_types.WORKSPACE_STATE_CHANGED,
+            {"8"},
+            "xterm-handoff-check",
+            payload,
+        )
+
     def test_query_failure_emits_unknown_instead_of_preserving_stale_i3_context(self):
         backend = compositor_state_i3.I3WorkspaceBackend(i3ipc_module=types.SimpleNamespace())
         backend._active = True

tests/test_event_manager_compositor_context_regressions.py

@@ -64,6 +64,16 @@ class EventManagerCompositorContextRegressionTests(unittest.TestCase):
 
         adapter.add_listener.assert_called_once_with(self.manager._handle_compositor_signal)
 
+    def test_set_compositor_state_adapter_updates_active_input_manager(self) -> None:
+        adapter = mock.Mock()
+        self.manager._inputEventManager = mock.Mock()
+
+        self.manager.set_compositor_state_adapter(adapter)
+
+        self.manager._inputEventManager.set_compositor_state_adapter.assert_called_once_with(
+            adapter
+        )
+
     def test_pause_signal_updates_churn_state_and_resume_clears_it(self) -> None:
         snapshot = compositor_state_types.DesktopContextSnapshot(session_type="wayland")
 

tests/test_input_event_manager_x11_focus_regressions.py

@@ -658,6 +658,43 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
         matcher.assert_called_once_with(0x2600003)
         getEwmhWindowId.assert_not_called()
 
+    def test_xterm_match_refreshes_i3_context_before_reusing_cached_match(self) -> None:
+        manager = input_event_manager.InputEventManager()
+        staleSnapshot = compositor_state_types.DesktopContextSnapshot(
+            session_type="x11",
+            backend_name="i3-ipc",
+            focused_window_id=0x200000C,
+            focused_workspace_empty=False,
+        )
+        freshSnapshot = compositor_state_types.DesktopContextSnapshot(
+            session_type="x11",
+            backend_name="i3-ipc",
+            focused_window_id=0x2600003,
+            focused_workspace_empty=False,
+        )
+        adapter = mock.Mock()
+
+        def refresh_context(_reason: str) -> compositor_state_types.DesktopContextSnapshot:
+            input_event_manager.cthulhu_state.compositorSnapshot = freshSnapshot
+            return freshSnapshot
+
+        adapter.refresh_workspace_context.side_effect = refresh_context
+        input_event_manager.cthulhu_state.compositorSnapshot = staleSnapshot
+        manager._lastEwmhActiveWindowId = 0x200000C
+        manager._lastEwmhXtermMatch = True
+
+        manager.set_compositor_state_adapter(adapter)
+        with mock.patch.object(
+            manager,
+            "_x11_window_id_xterm_match",
+            return_value=False,
+        ) as matcher:
+            result = manager._active_x11_window_xterm_match()
+
+        self.assertFalse(result)
+        adapter.refresh_workspace_context.assert_called_once_with("xterm-handoff-check")
+        matcher.assert_called_once_with(0x2600003)
+
     def test_i3_exact_xid_initializes_x11_before_matching_xterm(self):
         manager = input_event_manager.InputEventManager()
         snapshot = compositor_state_types.DesktopContextSnapshot(
@@ -696,6 +733,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
 
     def test_confirmed_empty_i3_workspace_is_definitely_not_xterm(self):
         manager = input_event_manager.InputEventManager()
+        manager._lastEwmhActiveWindowId = 0x200000C
+        manager._lastEwmhXtermMatch = True
         snapshot = compositor_state_types.DesktopContextSnapshot(
             session_type="x11",
             backend_name="i3-ipc",
@@ -710,6 +749,8 @@ class InputEventManagerX11FocusRegressionTests(unittest.TestCase):
             result = manager._active_x11_window_xterm_match()
 
         self.assertFalse(result)
+        self.assertIsNone(manager._lastEwmhActiveWindowId)
+        self.assertFalse(manager._lastEwmhXtermMatch)
         getEwmhWindowId.assert_not_called()
         matcher.assert_not_called()