diff --git a/openadapt_capture/platform/__init__.py b/openadapt_capture/platform/__init__.py index 9ff1c52..6999728 100644 --- a/openadapt_capture/platform/__init__.py +++ b/openadapt_capture/platform/__init__.py @@ -31,11 +31,21 @@ def get_display_pixel_ratio() -> float: ... @staticmethod - def is_accessibility_enabled() -> bool: + def is_accessibility_enabled() -> bool | None: """Check if accessibility permissions are enabled.""" ... +class DisplayMetricsUnavailable(RuntimeError): + """A display measurement could not be taken. + + Raised instead of returning a plausible default. A fabricated pixel ratio + or screen size is written into the recording and is indistinguishable from + a measured one, so every downstream coordinate is silently reinterpreted + at the wrong scale while the recording reports success. + """ + + def get_platform() -> str: """Get the current platform identifier. @@ -77,18 +87,26 @@ def get_screen_dimensions() -> tuple[int, int]: Returns: Tuple of (width, height) in physical pixels. + + Raises: + DisplayMetricsUnavailable: If the screen could not be measured. A + plausible default is never returned: it would be recorded as a + real measurement. """ try: provider = get_platform_provider() return provider.get_screen_dimensions() - except (NotImplementedError, ImportError): + except (NotImplementedError, ImportError) as exc: # Fallback to generic implementation try: from PIL import ImageGrab screenshot = ImageGrab.grab() return screenshot.size - except Exception: - return (1920, 1080) # Default fallback + except Exception as grab_exc: + raise DisplayMetricsUnavailable( + f"Could not measure screen dimensions: {exc}; " + f"the generic screenshot fallback also failed: {grab_exc}" + ) from grab_exc def get_display_pixel_ratio() -> float: @@ -99,31 +117,43 @@ def get_display_pixel_ratio() -> float: Returns: Pixel ratio (e.g., 1.0 for standard displays, 2.0 for Retina). + + Raises: + DisplayMetricsUnavailable: If the ratio could not be measured. `1.0` is + never substituted: a Retina display whose probe failed would be + recorded as a standard display and every captured coordinate would + be reinterpreted at half scale. """ try: provider = get_platform_provider() return provider.get_display_pixel_ratio() - except (NotImplementedError, ImportError): - return 1.0 + except (NotImplementedError, ImportError) as exc: + raise DisplayMetricsUnavailable( + f"Could not measure the display pixel ratio: {exc}" + ) from exc -def is_accessibility_enabled() -> bool: +def is_accessibility_enabled() -> bool | None: """Check if accessibility permissions are enabled. On macOS, this checks if the application has accessibility permissions required for keyboard and mouse event capture. Returns: - True if accessibility is enabled, False otherwise. + True if accessibility is verified enabled, False if it is verified + disabled, and None if the permission state could not be determined. + None is not a synonym for True: a caller gating "safe to record" on an + undetermined state would start a recording that captures no input. """ try: provider = get_platform_provider() return provider.is_accessibility_enabled() except (NotImplementedError, ImportError): - return True # Assume enabled on unknown platforms + return None __all__ = [ + "DisplayMetricsUnavailable", "get_platform", "get_platform_provider", "get_screen_dimensions", diff --git a/openadapt_capture/platform/darwin.py b/openadapt_capture/platform/darwin.py index 06555f5..71d1e5c 100644 --- a/openadapt_capture/platform/darwin.py +++ b/openadapt_capture/platform/darwin.py @@ -10,6 +10,8 @@ import sys +from openadapt_capture.platform import DisplayMetricsUnavailable + if sys.platform != "darwin": raise ImportError("This module is only available on macOS") @@ -30,12 +32,15 @@ def get_screen_dimensions() -> tuple[int, int]: Returns: Tuple of (width, height) in physical pixels. + + Raises: + DisplayMetricsUnavailable: If the screen could not be measured. """ try: from PIL import ImageGrab screenshot = ImageGrab.grab() return screenshot.size - except Exception: + except Exception as grab_exc: # Fallback using Quartz try: import Quartz @@ -44,8 +49,12 @@ def get_screen_dimensions() -> tuple[int, int]: width = Quartz.CGDisplayPixelsWide(main_display) height = Quartz.CGDisplayPixelsHigh(main_display) return (width, height) - except Exception: - return (1920, 1080) + except Exception as quartz_exc: + raise DisplayMetricsUnavailable( + "Could not measure macOS screen dimensions: " + f"ImageGrab failed ({grab_exc}) and Quartz failed " + f"({quartz_exc})" + ) from quartz_exc @staticmethod def get_display_pixel_ratio() -> float: @@ -73,8 +82,11 @@ def get_display_pixel_ratio() -> float: if logical_width > 0: return physical_width / logical_width - return 1.0 - except ImportError: + raise DisplayMetricsUnavailable( + "macOS reported a non-positive logical display width; " + "the pixel ratio could not be measured" + ) + except ImportError as import_exc: # Try using Quartz directly try: import Quartz @@ -91,21 +103,35 @@ def get_display_pixel_ratio() -> float: if logical_width > 0: return physical_width / logical_width - return 1.0 - except Exception: - return 1.0 - except Exception: - return 1.0 + raise DisplayMetricsUnavailable( + "Quartz did not report a usable display mode; " + "the pixel ratio could not be measured" + ) + except DisplayMetricsUnavailable: + raise + except Exception as quartz_exc: + raise DisplayMetricsUnavailable( + "Could not measure the macOS pixel ratio: mss/PIL missing " + f"({import_exc}) and Quartz failed ({quartz_exc})" + ) from quartz_exc + except DisplayMetricsUnavailable: + raise + except Exception as exc: + raise DisplayMetricsUnavailable( + f"Could not measure the macOS pixel ratio: {exc}" + ) from exc @staticmethod - def is_accessibility_enabled() -> bool: + def is_accessibility_enabled() -> bool | None: """Check if accessibility permissions are enabled. macOS requires accessibility permissions for capturing keyboard and mouse events globally. Returns: - True if accessibility is enabled, False otherwise. + True if verified enabled, False if verified disabled, and None if + the permission state could not be determined. None must not be + read as True: an unchecked permission is not a granted one. """ try: import Quartz # noqa: F401 - needed for ApplicationServices @@ -136,9 +162,9 @@ def is_accessibility_enabled() -> bool: ) return result.returncode == 0 except Exception: - return True # Assume enabled if we can't check + return None # Undetermined - never report this as enabled except Exception: - return True # Assume enabled if we can't check + return None # Undetermined - never report this as enabled @staticmethod def get_active_window_info() -> dict | None: diff --git a/openadapt_capture/platform/linux.py b/openadapt_capture/platform/linux.py index f0fd2fb..0afddac 100644 --- a/openadapt_capture/platform/linux.py +++ b/openadapt_capture/platform/linux.py @@ -10,6 +10,7 @@ import sys +from openadapt_capture.platform import DisplayMetricsUnavailable from openadapt_capture.x11_threads import ensure_xlib_thread_support if not sys.platform.startswith("linux"): @@ -99,7 +100,10 @@ def get_screen_dimensions() -> tuple[int, int]: except Exception: pass - return (1920, 1080) + raise DisplayMetricsUnavailable( + "Could not measure Linux screen dimensions: none of mss/PIL, " + "xdpyinfo, or wlr-randr produced a usable result" + ) @staticmethod def get_display_pixel_ratio() -> float: @@ -164,10 +168,13 @@ def get_display_pixel_ratio() -> float: except Exception: pass - return 1.0 + raise DisplayMetricsUnavailable( + "Could not measure the Linux pixel ratio: gsettings, GDK_SCALE, " + "QT_SCALE_FACTOR, and the mss/PIL comparison all failed" + ) @staticmethod - def is_accessibility_enabled() -> bool: + def is_accessibility_enabled() -> bool | None: """Check if input capture is available. On Linux, this typically requires: @@ -175,7 +182,9 @@ def is_accessibility_enabled() -> bool: - Wayland: Portal permissions or root access Returns: - True if input capture is likely available, False otherwise. + True if input capture is verified available, False if it is + verified unavailable, and None if the state could not be + determined. None must not be read as True. """ import os @@ -213,8 +222,9 @@ def is_accessibility_enabled() -> bool: except Exception: pass - # Assume enabled if we can't determine - return True + # Undetermined: the portal probe could not run. Never report this as + # enabled - a caller would start a recording that captures no input. + return None @staticmethod def get_active_window_info() -> dict | None: diff --git a/openadapt_capture/platform/windows.py b/openadapt_capture/platform/windows.py index eaa9172..3575193 100644 --- a/openadapt_capture/platform/windows.py +++ b/openadapt_capture/platform/windows.py @@ -10,6 +10,8 @@ import sys +from openadapt_capture.platform import DisplayMetricsUnavailable + if sys.platform != "win32": raise ImportError("This module is only available on Windows") @@ -35,7 +37,7 @@ def get_screen_dimensions() -> tuple[int, int]: from PIL import ImageGrab screenshot = ImageGrab.grab() return screenshot.size - except Exception: + except Exception as grab_exc: # Fallback using ctypes try: import ctypes @@ -53,8 +55,12 @@ def get_screen_dimensions() -> tuple[int, int]: width = user32.GetSystemMetrics(0) # SM_CXSCREEN height = user32.GetSystemMetrics(1) # SM_CYSCREEN return (width, height) - except Exception: - return (1920, 1080) + except Exception as ctypes_exc: + raise DisplayMetricsUnavailable( + "Could not measure Windows screen dimensions: ImageGrab " + f"failed ({grab_exc}) and GetSystemMetrics failed " + f"({ctypes_exc})" + ) from ctypes_exc @staticmethod def get_display_pixel_ratio() -> float: @@ -96,32 +102,42 @@ def get_display_pixel_ratio() -> float: except Exception: pass - return 1.0 - except Exception: - return 1.0 + raise DisplayMetricsUnavailable( + "Could not measure the Windows pixel ratio: neither " + "GetDpiForMonitor nor GetDeviceCaps produced a result" + ) + except DisplayMetricsUnavailable: + raise + except Exception as exc: + raise DisplayMetricsUnavailable( + f"Could not measure the Windows pixel ratio: {exc}" + ) from exc @staticmethod - def is_accessibility_enabled() -> bool: + def is_accessibility_enabled() -> bool | None: """Check if the application can capture input events. - On Windows, input capture typically works without special permissions, - but we check if we're running with sufficient privileges. + On Windows, input capture works for both administrator and standard + users, so a successful probe of the Win32 privilege API is what makes + the True answer a *checked* one rather than an assumed one. Returns: - True if input capture is available, False otherwise. + True if the Win32 privilege API answered, and None if it could not + be reached at all. None must not be read as True: the previous + version computed `IsUserAnAdmin()`, discarded the result, and + returned True from every branch including the failure handler, so + the answer carried no information. """ try: import ctypes - # Check if running as administrator - try: - ctypes.windll.shell32.IsUserAnAdmin() - # Even non-admin can typically capture input - return True - except Exception: - return True # Assume enabled - except Exception: + # The result is intentionally not used to gate the answer: a + # standard user can capture input. Reaching the API at all is what + # is being verified. + ctypes.windll.shell32.IsUserAnAdmin() return True + except Exception: + return None @staticmethod def get_active_window_info() -> dict | None: diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index b352d76..3719c58 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -1144,6 +1144,11 @@ def read_window_events( """ utils.set_start_time(recording.timestamp) + # Refuse at the boundary. Without this the loop below polls a backend that + # can never answer, never sets started_event, and the recording hangs in + # startup with no stated cause. + window.require_impl() + logger.info("Starting") prev_window_data = {} started = False @@ -1310,7 +1315,15 @@ def create_recording( timestamp = utils.set_start_time() monitor_width, monitor_height = utils.get_monitor_dims() - pixel_ratio = platform.get_display_pixel_ratio() + try: + pixel_ratio = platform.get_display_pixel_ratio() + except platform.DisplayMetricsUnavailable as exc: + # Persist NULL, not 1.0. The column is nullable precisely so that + # "unknown" is representable; recording 1.0 would report an + # unmeasured display as a verified standard-density one and every + # downstream coordinate would be rescaled from a fabricated number. + logger.error(f"Display pixel ratio could not be measured: {exc}") + pixel_ratio = None double_click_distance_pixels = utils.get_double_click_distance_pixels() double_click_interval_seconds = utils.get_double_click_interval_seconds() recording_data = { @@ -1825,15 +1838,21 @@ def record( # DIFFERENT window (whichever is focused), so it stays off. if config.RECORD_WINDOW_DATA and window_scope is None: window_event_reader = threading.Thread( - target=read_window_events, + target=_run_task_fail_loud, daemon=True, args=( - event_q, - terminate_processing, - recording, - task_started_events.setdefault( - "window_event_reader", threading.Event() + "window_event_reader", + read_window_events, + ( + event_q, + terminate_processing, + recording, + task_started_events.setdefault( + "window_event_reader", threading.Event() + ), ), + terminate_processing, + task_errors, ), ) window_event_reader.start() diff --git a/openadapt_capture/structural_observer/windows.py b/openadapt_capture/structural_observer/windows.py index 938d204..2464258 100644 --- a/openadapt_capture/structural_observer/windows.py +++ b/openadapt_capture/structural_observer/windows.py @@ -398,8 +398,15 @@ def _observe_with_runtime( if target is None: return None - element = _as_element(target) info = _element_info(target) + if info is None: + # The element vanished or the COM read failed. Emitting an observation + # whose element carries no identity fields would be positive evidence + # that windows_uia observed this action, indistinguishable from an + # element that legitimately exposes nothing. Report no observation. + return None + + element = _as_element(target) process_id = _present_int(_safe_value(info, "process_id")) process = None if process_id is not None: diff --git a/openadapt_capture/window/__init__.py b/openadapt_capture/window/__init__.py index eeb0a98..56321ea 100644 --- a/openadapt_capture/window/__init__.py +++ b/openadapt_capture/window/__init__.py @@ -10,7 +10,19 @@ from openadapt_capture.config import config + +class WindowCaptureUnavailable(RuntimeError): + """No window-capture backend could be loaded for this process. + + Raised rather than letting every window query return an empty result + forever. An unavailable backend and a momentarily unreadable window + otherwise look identical, and a recording proceeds with a silently empty + window timeline. + """ + + impl = None +impl_unavailable_reason: str | None = None try: if sys.platform == "darwin": from . import _macos as impl @@ -19,9 +31,31 @@ elif sys.platform.startswith("linux"): from . import _linux as impl else: - logger.warning(f"Unsupported platform for window capture: {sys.platform}") + impl_unavailable_reason = ( + f"Unsupported platform for window capture: {sys.platform}" + ) + logger.warning(impl_unavailable_reason) except ImportError as exc: - logger.warning(f"Window capture not available: {exc}") + impl_unavailable_reason = f"Window capture backend failed to import: {exc}" + logger.error(impl_unavailable_reason) + + +def require_impl() -> Any: + """Return the window-capture backend, or refuse. + + Returns: + The platform backend module. + + Raises: + WindowCaptureUnavailable: If no backend loaded. Call this before + starting a window-reading loop so an absent backend surfaces as a + recording-startup failure instead of an empty window timeline. + """ + if impl is None: + raise WindowCaptureUnavailable( + impl_unavailable_reason or "No window-capture backend is loaded" + ) + return impl def get_active_window_data( @@ -34,11 +68,13 @@ def get_active_window_data( Returns: dict or None: A dictionary containing information about the active window, - or None if the state is not available. + or None if the state is not available. `None` is the documented + unavailable answer; an empty dict is never returned, because a + caller cannot tell it from a window with no fields. """ state = get_active_window_state(include_window_data) if not state: - return {} + return None title = state["title"] left = state["left"] top = state["top"] diff --git a/openadapt_capture/window/_linux.py b/openadapt_capture/window/_linux.py index 14bc905..094f6d4 100644 --- a/openadapt_capture/window/_linux.py +++ b/openadapt_capture/window/_linux.py @@ -166,10 +166,13 @@ def get_active_element_state(x: int, y: int) -> dict | None: y (int): The y-coordinate of the element. Returns: - dict or None: A dictionary containing the state of the active element. + None. Linux element-level state retrieval is not implemented (it would + need AT-SPI). None means "not observed"; the previous placeholder dict + was persisted verbatim onto every ActionEvent, so a consumer could not + tell invented state from captured accessibility data. """ - # Placeholder: Implement element-level state retrieval if necessary. - return {"x": x, "y": y, "state": "placeholder"} + del x, y # No provider to query. + return None def main() -> None: diff --git a/tests/test_no_fabricated_capture.py b/tests/test_no_fabricated_capture.py new file mode 100644 index 0000000..9de0171 --- /dev/null +++ b/tests/test_no_fabricated_capture.py @@ -0,0 +1,198 @@ +"""Regressions against reporting a capture that did not happen. + +Every test here pins the boundary between "measured / observed" and "could not +measure / could not observe". A plausible default returned from the second case +is written into the recording and is indistinguishable from a real measurement. +""" + +from __future__ import annotations + +import importlib +import queue +import sys +import threading +from types import ModuleType, SimpleNamespace + +import pytest + +from openadapt_capture import platform as capture_platform +from openadapt_capture import recorder, window +from openadapt_capture.platform import DisplayMetricsUnavailable +from openadapt_capture.structural import StructuralObservationRequest +from openadapt_capture.structural_observer.windows import WindowsUIAStructuralObserver +from openadapt_capture.window import WindowCaptureUnavailable + + +def _unavailable_provider(*_args, **_kwargs): + raise NotImplementedError("Platform not supported: test") + + +class TestDisplayMetricsAreMeasuredOrRefused: + def test_pixel_ratio_raises_instead_of_returning_one(self, monkeypatch) -> None: + # 1.0 is a real, common answer. Returning it for an unmeasurable + # display records a Retina screen as standard density and rescales + # every captured coordinate by half. + monkeypatch.setattr( + capture_platform, "get_platform_provider", _unavailable_provider + ) + + with pytest.raises(DisplayMetricsUnavailable): + capture_platform.get_display_pixel_ratio() + + def test_screen_dimensions_raise_instead_of_returning_1920x1080( + self, monkeypatch + ) -> None: + monkeypatch.setattr( + capture_platform, "get_platform_provider", _unavailable_provider + ) + + class _BrokenImageGrab: + @staticmethod + def grab(): + raise OSError("no display") + + monkeypatch.setitem(sys.modules, "PIL.ImageGrab", _BrokenImageGrab) + + with pytest.raises(DisplayMetricsUnavailable): + capture_platform.get_screen_dimensions() + + def test_accessibility_state_is_none_when_undetermined(self, monkeypatch) -> None: + # `True` here used to mean "I did not check". + monkeypatch.setattr( + capture_platform, "get_platform_provider", _unavailable_provider + ) + + assert capture_platform.is_accessibility_enabled() is None + + def test_recording_stores_null_pixel_ratio_when_unmeasurable( + self, monkeypatch, tmp_path + ) -> None: + monkeypatch.setattr(recorder.utils, "get_monitor_dims", lambda: (1280, 720)) + monkeypatch.setattr( + recorder.utils, "get_double_click_distance_pixels", lambda: 5.0 + ) + monkeypatch.setattr( + recorder.utils, "get_double_click_interval_seconds", lambda: 0.5 + ) + + def _unmeasurable() -> float: + raise DisplayMetricsUnavailable("probe failed") + + monkeypatch.setattr( + recorder.platform, "get_display_pixel_ratio", _unmeasurable + ) + + recording, _db_path = recorder.create_recording( + "test task", capture_dir=str(tmp_path / "capture") + ) + + # NULL, not 1.0: the column is nullable so that unknown is + # representable in the recording itself. + assert recording.pixel_ratio is None + + +class TestWindowBackendAvailability: + def test_require_impl_raises_when_no_backend_loaded(self, monkeypatch) -> None: + monkeypatch.setattr(window, "impl", None) + monkeypatch.setattr( + window, "impl_unavailable_reason", "backend failed to import" + ) + + with pytest.raises(WindowCaptureUnavailable): + window.require_impl() + + def test_active_window_data_is_none_not_empty_dict(self, monkeypatch) -> None: + # The annotation and docstring both promise None. Returning {} made + # "no backend" look like a window with no fields. + monkeypatch.setattr(window, "get_active_window_state", lambda _flag: None) + + assert window.get_active_window_data() is None + + def test_window_reader_refuses_instead_of_polling_forever( + self, monkeypatch + ) -> None: + monkeypatch.setattr(window, "impl", None) + monkeypatch.setattr( + window, "impl_unavailable_reason", "backend failed to import" + ) + started_event = threading.Event() + terminate = threading.Event() + + with pytest.raises(WindowCaptureUnavailable): + recorder.read_window_events( + queue.Queue(), + terminate, + SimpleNamespace(timestamp=100.0), + started_event, + ) + + # The pre-fix loop spun on an empty result and never announced + # readiness, so recording startup hung with no stated cause. + assert not started_event.is_set() + + +class _WrapperWithUnreadableElement: + """A UIA wrapper whose element_info read fails, as on a vanished element.""" + + @property + def element_info(self): + raise OSError("COM call failed") + + def friendly_class_name(self) -> str | None: + return "Button" + + def parent(self): + return None + + def top_level_parent(self): + return self + + def descendants(self, **_kwargs): + return [] + + def window_text(self) -> str | None: + return None + + +class TestStructuralObservationIsEvidence: + def test_unreadable_element_yields_no_observation(self) -> None: + target = _WrapperWithUnreadableElement() + observer = WindowsUIAStructuralObserver( + runtime=SimpleNamespace( + from_point=lambda _x, _y: target, + focused_element=lambda: target, + ), + process_name_resolver=lambda _pid: None, + clock=lambda: 101.25, + ) + + observed = observer.observe( + StructuralObservationRequest( + event_timestamp=101.0, + action_name="click", + x=25, + y=35, + ) + ) + + # An observation with an identity-free element is positive evidence + # that windows_uia observed this action. It must not be emitted when + # nothing could be read. + assert observed is None + + +class TestLinuxElementState: + def test_element_state_is_none_not_a_placeholder_dict(self, monkeypatch) -> None: + # Import the Linux backend on any host by stubbing its X11 dependency: + # the function under test never touches X11, and gating this test on + # Linux would leave the regression unproven on the PR matrix. + for name in ("xcffib", "xcffib.xproto"): + stub = ModuleType(name) + stub.Connection = object # annotation target at import time + monkeypatch.setitem(sys.modules, name, stub) + monkeypatch.delitem(sys.modules, "openadapt_capture.window._linux", raising=False) + _linux = importlib.import_module("openadapt_capture.window._linux") + + # The placeholder dict was persisted verbatim onto every ActionEvent + # and read downstream as captured accessibility data. + assert _linux.get_active_element_state(10, 20) is None