From 19b3cdc52cc0d0193951df42a689938961e9d29b Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 23:22:41 -0400 Subject: [PATCH 1/2] fix(capture): keep only input the recorded window received Window-scoped recording attributed every key and click to the target window whenever the window's identity and bounds were unchanged. Keys typed after switching to another app, including a password prompt, and clicks on a window that overlaps the target's rectangle were stored as target actions. Each native input is now admitted only if the target received it: - A key press needs keyboard focus on the target window or its process, both when the native hook receives it and when it is delivered. - A click, scroll or move needs the target to be the topmost window at the point. A target that is off screen (a ScreenCaptureKit capture of a covered window) never receives clicks. - A release, or a move while a button is held, continues a press the target accepted and is kept, so programs keep matched pairs. A release whose press was dropped is dropped too. - An unknown answer or a probe failure is a refusal. Dropped input is removed before logging, structural observation or persistence; its journal position is discarded. A marker with a reason code, the input kind, a timestamp and a coarse point is appended to window-input-gaps.jsonl. Markers never hold key identity or text. The probes sit behind an injectable seam on WindowCaptureScope (focus_probe). Platform probes: NSWorkspace frontmost pid cross-checked with the front-most on-screen Quartz window, and the first layer-0 window containing the point (macOS); GetForegroundWindow and WindowFromPoint with GetAncestor(GA_ROOT) (Windows); _NET_ACTIVE_WINDOW and the client inside the topmost viewable root child (X11). Desktop recording is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../capture/openadapt_capture/recorder.py | 264 ++++++- .../openadapt_capture/window_capture.py | 111 ++- .../openadapt_capture/window_capture_linux.py | 59 ++ .../openadapt_capture/window_input_focus.py | 242 +++++++ packages/capture/tests/test_window_capture.py | 36 +- .../capture/tests/test_window_input_focus.py | 677 ++++++++++++++++++ 6 files changed, 1380 insertions(+), 9 deletions(-) create mode 100644 packages/capture/openadapt_capture/window_input_focus.py create mode 100644 packages/capture/tests/test_window_input_focus.py diff --git a/packages/capture/openadapt_capture/recorder.py b/packages/capture/openadapt_capture/recorder.py index c257c369c..38af558e5 100644 --- a/packages/capture/openadapt_capture/recorder.py +++ b/packages/capture/openadapt_capture/recorder.py @@ -21,6 +21,7 @@ import hashlib import io +import json import math import multiprocessing import os @@ -90,6 +91,7 @@ WindowCaptureError, WindowCapturePermissionError, WindowCaptureScope, + WindowInputNotOnTarget, build_window_scope, prepare_macos_window_capture_runtime, ) @@ -286,6 +288,7 @@ class _JournalEntry: event: Event | None = None error: BaseException | None = None ready: bool = False + discarded: bool = False class EventReservation: @@ -331,6 +334,16 @@ def fail(self, error: BaseException) -> None: self._finished = True self._journal._condition.notify_all() + def discard(self) -> None: + """Release this position without an event; the consumer skips it.""" + if self._finished: + raise RuntimeError("the event journal reservation is already complete") + with self._journal._condition: + self._entry.discarded = True + self._entry.ready = True + self._finished = True + self._journal._condition.notify_all() + class WindowActionReservation: """A source position and the exact published geometry at native receipt.""" @@ -340,10 +353,36 @@ def __init__( reservation: EventReservation, window_scope: WindowCaptureScope, geometry: tuple[Any, float, float, tuple[int, int, int, int], int], + keyboard_focus_at_receipt: Any = None, ) -> None: self._reservation = reservation self._window_scope = window_scope self._geometry = geometry + self._keyboard_focus_at_receipt = keyboard_focus_at_receipt + self._admitted = False + + def admit(self, point: tuple[float, float] | None) -> None: + """Accept input only if the target window received it. + + Raises: + WindowInputNotOnTarget: The input went to another window. + """ + if point is None: + self._window_scope.assert_input_on_target( + self._geometry[0], + keyboard_focus_at_receipt=self._keyboard_focus_at_receipt, + ) + else: + self._window_scope.assert_input_on_target(self._geometry[0], point=point) + self._admitted = True + + def admit_continuation(self) -> None: + """Accept the release or drag of a press the target already received.""" + self._admitted = True + + def discard(self) -> None: + """Drop this input: its journal position completes without an event.""" + self._reservation.discard() @property def source_ordinal(self) -> int: @@ -359,6 +398,10 @@ def bind( y: float | None, ) -> tuple[float, float, int] | int: """Bind normalized input to geometry reserved at native receipt.""" + if not self._admitted: + raise EventJournalOrderingError( + "window-scoped input must be admitted as on-target before binding" + ) if x is not None and y is not None: return self._window_scope.translate_reserved_geometry( self._geometry, @@ -412,6 +455,94 @@ def fail(self, error: BaseException) -> None: del error +WINDOW_INPUT_GAPS_FILENAME = "window-input-gaps.jsonl" +WINDOW_INPUT_GAP_SCHEMA_VERSION = "openadapt.capture.window-input-gap/v1" +_WINDOW_INPUT_GAP_POINT_GRID = 10.0 + + +class WindowInputGapLog: + """Markers for input dropped because it did not go to the target window. + + One ``focus_left_target`` marker opens each run of dropped input (and a + new one is written when the reason changes); ``focus_returned_to_target`` + closes it at the next accepted input. A marker holds a reason code, the + input kind, a timestamp and, for pointer input, a point rounded to a + coarse grid. It never holds key identity or text. + + Markers are appended to ``window-input-gaps.jsonl`` in the capture + directory, created only when input is dropped, and kept in ``records``. + """ + + def __init__(self, capture_dir: str | os.PathLike[str] | None = None) -> None: + self._path = ( + None if capture_dir is None else os.path.join(capture_dir, WINDOW_INPUT_GAPS_FILENAME) + ) + self._lock = threading.Lock() + self._active_reason: str | None = None + self.records: list[dict[str, Any]] = [] + + def _append_locked(self, record: dict[str, Any]) -> None: + self.records.append(record) + if self._path is None: + return + line = (json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n").encode() + fd = os.open( + self._path, + os.O_WRONLY | os.O_CREAT | os.O_APPEND | getattr(os, "O_BINARY", 0), + 0o600, + ) + try: + os.write(fd, line) + os.fsync(fd) + finally: + os.close(fd) + + def left_target( + self, + reason: str, + input_kind: str, + timestamp: float | None, + point: tuple[float, float] | None = None, + ) -> None: + """Open (or re-reason) a run of dropped input.""" + with self._lock: + if self._active_reason == reason: + return + self._active_reason = reason + record: dict[str, Any] = { + "schema_version": WINDOW_INPUT_GAP_SCHEMA_VERSION, + "event": "focus_left_target", + "reason": reason, + "input": input_kind, + "timestamp": timestamp, + } + if point is not None: + grid = _WINDOW_INPUT_GAP_POINT_GRID + record["coarse_point"] = [ + round(point[0] / grid) * grid, + round(point[1] / grid) * grid, + ] + self._append_locked(record) + logger.info( + "window-scoped input dropped: it did not go to the target window (reason={})", + reason, + ) + + def returned(self, timestamp: float | None) -> None: + """Close the current run of dropped input, if any.""" + with self._lock: + if self._active_reason is None: + return + self._active_reason = None + self._append_locked( + { + "schema_version": WINDOW_INPUT_GAP_SCHEMA_VERSION, + "event": "focus_returned_to_target", + "timestamp": timestamp, + } + ) + + class OrderedEventJournal: """A causal FIFO journal with pre-observation reservations.""" @@ -463,6 +594,12 @@ def reserve_window_action( ) else: binding = window_scope.generation_for_action() + except WindowInputNotOnTarget: + entry.discarded = True + entry.ready = True + reservation._finished = True + self._condition.notify_all() + raise except BaseException as exc: entry.error = exc entry.ready = True @@ -480,6 +617,9 @@ def reserve_window_action_receipt( timestamp = float(timestamp) if not math.isfinite(timestamp): raise EventJournalOrderingError("event timestamps must be finite") + # Read keyboard focus first, as close to native receipt as possible: + # delivery can lag long enough for focus to return to the target. + keyboard_focus = window_scope.keyboard_focus_at_receipt() with window_scope.observation_boundary(): with self._condition: entry = self._reserve_locked(timestamp) @@ -492,7 +632,7 @@ def reserve_window_action_receipt( reservation._finished = True self._condition.notify_all() raise - return WindowActionReservation(reservation, window_scope, geometry) + return WindowActionReservation(reservation, window_scope, geometry, keyboard_focus) def put(self, event: Event, block: bool = True, timeout: float | None = None) -> int: del block, timeout @@ -535,6 +675,8 @@ def get(self, block: bool = True, timeout: float | None = None) -> Event: while True: if self._entries and self._entries[0].ready: entry = self._entries.popleft() + if entry.discarded: + continue if entry.error is not None: raise EventJournalReservationError( "an ordered event producer failed after reservation" @@ -556,6 +698,8 @@ def get_nowait(self) -> Event: def empty(self) -> bool: with self._condition: + while self._entries and self._entries[0].ready and self._entries[0].discarded: + self._entries.popleft() return not self._entries def qsize(self) -> int: @@ -2441,6 +2585,90 @@ def create_recording( return recording, db_path +def _key_identity(event: ObservedKey) -> tuple[str | None, str | None, str | None]: + """Pair a key release with its press, in memory only.""" + return (event.key_vk, event.key_name, event.canonical_key_name) + + +class _WindowInputAdmission: + """Decide, per native input, whether the target window received it. + + Keys and pointer presses, scrolls and moves are probed. A release (or a + move while a button is held) continues a press the target accepted and is + admitted without a probe, so programs keep matched press/release pairs. + A release whose press was dropped is dropped too. + + Runs on the single input delivery thread. + """ + + def __init__(self, scope: WindowCaptureScope, gap_log: WindowInputGapLog) -> None: + self._scope = scope + self._gap_log = gap_log + self._held_keys: set[tuple[str | None, str | None, str | None]] = set() + self._held_buttons: set[str] = set() + + @staticmethod + def _discard(reservation: object | None) -> None: + if isinstance(reservation, WindowActionReservation): + reservation.discard() + + def dropped(self, event: ObservedInput, reason: str) -> None: + """Record a marker for dropped input, never its content.""" + if isinstance(event, ObservedMouseMove): + return # moves off the target are routine; they open no gap + if isinstance(event, ObservedKey): + self._gap_log.left_target(reason, "keyboard", event.timestamp) + else: + self._gap_log.left_target(reason, "pointer", event.timestamp, (event.x, event.y)) + + def admit(self, event: ObservedInput, reservation: object | None) -> bool: + """Return whether to record ``event``; a dropped event is discarded.""" + continuation = False + if isinstance(event, ObservedKey) and not event.pressed: + identity = _key_identity(event) + if identity not in self._held_keys: + self._discard(reservation) + return False + self._held_keys.discard(identity) + continuation = True + elif isinstance(event, ObservedMouseButton) and not event.pressed: + if event.button not in self._held_buttons: + self._discard(reservation) + return False + self._held_buttons.discard(event.button) + continuation = True + elif isinstance(event, ObservedMouseMove) and self._held_buttons: + continuation = True + + if continuation: + if isinstance(reservation, WindowActionReservation): + reservation.admit_continuation() + return True + + point = None if isinstance(event, ObservedKey) else (event.x, event.y) + try: + if isinstance(reservation, WindowActionReservation): + reservation.admit(point) + else: + window = self._scope.reserve_action_geometry()[0] + if point is None: + self._scope.assert_input_on_target(window) + else: + self._scope.assert_input_on_target(window, point=point) + except WindowInputNotOnTarget as exc: + self._discard(reservation) + self.dropped(event, exc.reason) + return False + + if isinstance(event, ObservedKey): + self._held_keys.add(_key_identity(event)) + elif isinstance(event, ObservedMouseButton): + self._held_buttons.add(event.button) + if not isinstance(event, ObservedMouseMove): + self._gap_log.returned(event.timestamp) + return True + + def read_input_events( event_q: queue.Queue, terminate_processing: multiprocessing.Event, @@ -2453,14 +2681,45 @@ def read_input_events( terminal_frame_finished: threading.Event | None = None, terminal_frame_cancelled: threading.Event | None = None, authentication: AuthenticationHandoffController | None = None, + input_gap_log: WindowInputGapLog | None = None, ) -> None: - """Read globally ordered keyboard and mouse events from one native observer.""" + """Read globally ordered keyboard and mouse events from one native observer. + + In window-scoped mode each input is admitted only if it went to the + target window (see :func:`_admit_window_input`). Other input is dropped + before logging, structural observation, or persistence, and + ``input_gap_log`` records a marker without the input's content. + """ stop_sequences = [sequence for sequence in config.STOP_SEQUENCES if sequence] stop_sequence_indices = [0 for _ in stop_sequences] + window_admission = ( + _WindowInputAdmission(coordinate_scope, input_gap_log or WindowInputGapLog()) + if isinstance(coordinate_scope, WindowCaptureScope) + else None + ) def on_observed( event: ObservedInput, reservation: EventReservation | WindowActionReservation | None = None, + ) -> None: + if window_admission is None: + dispatch_observed(event, reservation) + return + if getattr(event, "injected", False): + dispatch_observed(event, reservation) + return + if not window_admission.admit(event, reservation): + return + try: + dispatch_observed(event, reservation) + except WindowInputNotOnTarget as exc: + # Only the legacy unreserved path re-probes after admission; its + # journal position was already discarded. + window_admission.dropped(event, exc.reason) + + def dispatch_observed( + event: ObservedInput, + reservation: EventReservation | WindowActionReservation | None = None, ) -> None: if isinstance(event, ObservedMouseMove): on_move( @@ -3209,6 +3468,7 @@ def record( terminal_frame_finished, terminal_frame_cancelled, authentication, + WindowInputGapLog(capture_dir) if window_scope is not None else None, ) input_event_reader = threading.Thread( target=_run_task_fail_loud, diff --git a/packages/capture/openadapt_capture/window_capture.py b/packages/capture/openadapt_capture/window_capture.py index 41bdb2cae..ee82866cb 100644 --- a/packages/capture/openadapt_capture/window_capture.py +++ b/packages/capture/openadapt_capture/window_capture.py @@ -48,6 +48,12 @@ from loguru import logger +from openadapt_capture.window_input_focus import ( + FocusOwner, + InputFocusProbe, + platform_input_focus_probe, +) + if TYPE_CHECKING: from PIL import Image @@ -140,6 +146,28 @@ class WindowCaptureUnavailableError(WindowCaptureError): """The exact-window capture provider is not available.""" +INPUT_KEYBOARD_FOCUS_ELSEWHERE = "keyboard_focus_elsewhere" +INPUT_KEYBOARD_FOCUS_UNKNOWN = "keyboard_focus_unknown" +INPUT_POINTER_OVER_OTHER_WINDOW = "pointer_over_other_window" +INPUT_POINTER_TARGET_UNKNOWN = "pointer_target_unknown" +INPUT_TARGET_NOT_ON_SCREEN = "target_not_on_screen" +_NOT_PROBED = object() + + +class WindowInputNotOnTarget(WindowCaptureError): + """Native input went to a window other than the recorded target. + + This is not a capture failure. The recorder drops the input and records a + marker with ``reason``. The exception never carries key identity, text, or + an exact point, so another application's input cannot reach a recording. + """ + + def __init__(self, reason: str) -> None: + """Record a stable reason code.""" + super().__init__(f"native input was not on the target window ({reason})") + self.reason = reason + + @dataclass(frozen=True) class WindowTarget: """How to find the window to record: case-insensitive substrings. @@ -255,9 +283,15 @@ def __init__( resolver: ResolverFn | None = None, capturer: CapturerFn | None = None, frame_rate: float | None = None, + focus_probe: InputFocusProbe | None = None, ) -> None: - """Initialize the scope for ``target``.""" + """Initialize the scope for ``target``. + + ``focus_probe`` reports keyboard focus and the topmost window at a + point. It defaults to the platform probe; tests inject a fake. + """ self.target = target + self._focus_probe = focus_probe or platform_input_focus_probe() self._resolver = resolver or resolve_window self._capture_provider = None if capturer is None and sys.platform == "darwin": @@ -596,10 +630,73 @@ def translate(self, x: float, y: float) -> tuple[float, float]: px, py, _generation = self.translate_with_generation(x, y) return px, py + def keyboard_focus_at_receipt(self) -> FocusOwner | None: + """Read keyboard focus inside a native input hook; ``None`` if unknown.""" + try: + return self._focus_probe.keyboard_focus(at_receipt=True) + except Exception: + return None + + def assert_input_on_target( + self, + window: TargetWindow, + *, + point: tuple[float, float] | None = None, + keyboard_focus_at_receipt: object = _NOT_PROBED, + ) -> None: + """Refuse input that ``window`` did not receive. + + A pointer event (``point`` set) needs ``window`` topmost at the point. + A keyboard event needs keyboard focus on ``window`` or its process, + both at native receipt (when a receipt reading is given) and now. An + unknown answer or a probe failure is a refusal. + + Raises: + WindowInputNotOnTarget: With a stable reason code. + """ + if point is not None: + if not window.on_screen: + # A visibility-independent provider can capture a covered + # window, but a click there reached another window. + raise WindowInputNotOnTarget(INPUT_TARGET_NOT_ON_SCREEN) + try: + owner = self._focus_probe.window_at(float(point[0]), float(point[1])) + except Exception: + owner = None + if owner is None: + raise WindowInputNotOnTarget(INPUT_POINTER_TARGET_UNKNOWN) + if owner.window_id != window.window_id: + raise WindowInputNotOnTarget(INPUT_POINTER_OVER_OTHER_WINDOW) + return + try: + live = self._focus_probe.keyboard_focus() + except Exception: + live = None + readings = [live] + if keyboard_focus_at_receipt is not _NOT_PROBED: + readings.append(keyboard_focus_at_receipt) + for owner in readings: + if not isinstance(owner, FocusOwner): + raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_UNKNOWN) + if owner.window_id is None and owner.pid is None: + raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_UNKNOWN) + on_target = (owner.window_id is not None and owner.window_id == window.window_id) or ( + owner.pid is not None and owner.pid == window.pid + ) + if not on_target: + raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_ELSEWHERE) + def _geometry_for_action( self, + *, + point: tuple[float, float] | None = None, + keyboard: bool = False, ) -> tuple[TargetWindow, float, float, tuple[int, int, int, int], int]: - """Return the published geometry after exact live revalidation.""" + """Return the published geometry after exact live revalidation. + + With ``point`` or ``keyboard`` set, the input must also have gone to + the target (:meth:`assert_input_on_target`). + """ self._assert_display_topology() with self._lock: window = self._published_window @@ -624,6 +721,8 @@ def _geometry_for_action( "the target moved or resized after the last published frame; " "wait for a matching frame before recording input" ) + if point is not None or keyboard: + self.assert_input_on_target(live, point=point) self._assert_display_topology() return window, scale_x, scale_y, content_rect, generation @@ -684,8 +783,8 @@ def translate_reserved_geometry( ) def generation_for_action(self) -> int: - """Bind a non-pointer action to the exact published frame epoch.""" - return self._geometry_for_action()[4] + """Bind a keyboard action to the exact published frame epoch.""" + return self._geometry_for_action(keyboard=True)[4] def assert_current(self) -> None: """Revalidate the bound process, bounds, and display topology.""" @@ -693,7 +792,9 @@ def assert_current(self) -> None: def translate_with_generation(self, x: float, y: float) -> tuple[float, float, int]: """Translate against the exact published frame after revalidation.""" - window, scale_x, scale_y, content_rect, generation = self._geometry_for_action() + window, scale_x, scale_y, content_rect, generation = self._geometry_for_action( + point=(x, y) + ) return ( (x - window.bounds[0]) * scale_x + content_rect[0], (y - window.bounds[1]) * scale_y + content_rect[1], diff --git a/packages/capture/openadapt_capture/window_capture_linux.py b/packages/capture/openadapt_capture/window_capture_linux.py index dde9b4425..034ed8324 100644 --- a/packages/capture/openadapt_capture/window_capture_linux.py +++ b/packages/capture/openadapt_capture/window_capture_linux.py @@ -209,6 +209,65 @@ def _pid(self, window_id: int) -> int: values = _u32_values(payload) if value_format == 32 else [] return values[0] if len(values) == 1 else 0 + def window_pid(self, window_id: int) -> int | None: + """Return the ``_NET_WM_PID`` of one window, or ``None`` when absent.""" + try: + pid = self._pid(window_id) + except Exception: + return None + return pid or None + + def active_window(self) -> int | None: + """Return the EWMH ``_NET_ACTIVE_WINDOW`` client, or ``None``.""" + value_format, payload = self._property(self._root, "_NET_ACTIVE_WINDOW", length=1) + values = _u32_values(payload) if value_format == 32 else [] + return values[0] if len(values) == 1 and values[0] else None + + def client_window_at(self, x: int, y: int) -> int | None: + """Return the client inside the topmost viewable root child at a point. + + Root children are window-manager frames (or unmanaged windows) in + bottom-to-top stacking order. The EWMH client inside the topmost frame + that contains the point is returned in the same identifier space as + the resolver. An unmanaged window, such as an override-redirect popup, + is returned as itself, so it never compares equal to the target. + """ + children = list(self._conn.core.QueryTree(self._root).reply().children) + frame = None + for child in reversed(children): + try: + attributes = self._conn.core.GetWindowAttributes(child).reply() + if int(attributes.map_state) != self._xproto.MapState.Viewable: + continue + if int(attributes._class) != self._xproto.WindowClass.InputOutput: + continue + geometry = self._conn.core.GetGeometry(child).reply() + except Exception: + continue + border = int(geometry.border_width) + left, top = int(geometry.x), int(geometry.y) + width = int(geometry.width) + 2 * border + height = int(geometry.height) + 2 * border + if left <= x < left + width and top <= y < top + height: + frame = int(child) + break + if frame is None: + return None + clients = set(self.window_ids()) + pending = [(frame, 0)] + while pending: + window_id, depth = pending.pop(0) + if window_id in clients: + return window_id + if depth >= 3: + continue + try: + nested = self._conn.core.QueryTree(window_id).reply().children + except Exception: + continue + pending.extend((int(child), depth + 1) for child in nested) + return frame + def window_record(self, window_id: int) -> X11WindowRecord | None: """Read one viewability, identity, and root-pixel geometry snapshot.""" try: diff --git a/packages/capture/openadapt_capture/window_input_focus.py b/packages/capture/openadapt_capture/window_input_focus.py new file mode 100644 index 000000000..a07b37940 --- /dev/null +++ b/packages/capture/openadapt_capture/window_input_focus.py @@ -0,0 +1,242 @@ +"""Which window receives native input: keyboard focus and topmost-at-point. + +Window-scoped recording stores input only when it went to the recorded +target window. Bounds alone cannot show that: a user can switch to another +application and type, or click a window that overlaps the target's +rectangle. These probes report, for one instant, which window or process +has keyboard focus and which top-level window is topmost at a screen point. + +Each probe returns a :class:`FocusOwner` or ``None`` when the answer cannot +be determined. Callers treat ``None`` and any exception as "not on the +target" (fail closed). + +Platform bindings are imported lazily inside each probe so that importing +this module never touches a display (tests/test_headless_import.py). +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from typing import Iterator, Protocol + + +@dataclass(frozen=True) +class FocusOwner: + """A window and/or process observed as an input receiver. + + ``window_id`` uses the same identifier space as + :attr:`openadapt_capture.window_capture.TargetWindow.window_id` on the + same platform (CGWindowID, top-level HWND, or X11 client window). + """ + + window_id: int | None = None + pid: int | None = None + + +class InputFocusProbe(Protocol): + """Injectable seam that reports which window receives native input.""" + + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Return the keyboard-focus owner, or ``None`` when unknown. + + ``at_receipt`` is set when the call runs inside a native input hook. + Implementations must then answer from a cheap local read. + """ + ... + + def window_at(self, x: float, y: float) -> FocusOwner | None: + """Return the topmost top-level window at a global screen point.""" + ... + + +def _point_in_bounds(x: float, y: float, bounds: tuple[float, float, float, float]) -> bool: + left, top, width, height = bounds + return left <= x < left + width and top <= y < top + height + + +class MacOSInputFocusProbe: + """macOS: frontmost application and front-to-back Quartz window list. + + Bounds come from ``kCGWindowBounds``: global screen points with a + top-left origin, the same space the macOS resolver binds. + """ + + @staticmethod + def _frontmost_pid() -> int | None: + from AppKit import NSWorkspace + + app = NSWorkspace.sharedWorkspace().frontmostApplication() + if app is None: + return None + pid = int(app.processIdentifier()) + return pid if pid > 0 else None + + @staticmethod + def _onscreen_app_windows() -> Iterator[dict]: + """Yield on-screen layer-0 windows front to back, lazily.""" + import Quartz + + options = ( + Quartz.kCGWindowListOptionOnScreenOnly | Quartz.kCGWindowListExcludeDesktopElements + ) + windows = Quartz.CGWindowListCopyWindowInfo(options, Quartz.kCGNullWindowID) + for info in windows or []: + if int(info.get("kCGWindowLayer", 0) or 0) != 0: + continue # menu bar, status items, overlays + if float(info.get("kCGWindowAlpha", 1.0) or 0.0) <= 0.0: + continue # invisible windows do not receive clicks + b = info.get("kCGWindowBounds", {}) or {} + bounds = ( + float(b.get("X", 0.0)), + float(b.get("Y", 0.0)), + float(b.get("Width", 0.0)), + float(b.get("Height", 0.0)), + ) + if bounds[2] <= 0 or bounds[3] <= 0: + continue + yield { + "window_id": int(info.get("kCGWindowNumber", 0) or 0), + "pid": int(info.get("kCGWindowOwnerPID", 0) or 0), + "bounds": bounds, + } + + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Return the frontmost application's process. + + Inside the event tap only the cheap ``NSWorkspace`` read runs. At + delivery the answer must also agree with the owner of the front-most + on-screen application window. ``NSWorkspace`` is updated through the + main run loop and can be stale in a process that does not run one, so + a disagreement is reported as unknown. + """ + pid = self._frontmost_pid() + if pid is None or at_receipt: + return None if pid is None else FocusOwner(pid=pid) + front = next(self._onscreen_app_windows(), None) + if front is None or front["pid"] != pid: + return None + return FocusOwner(pid=pid) + + def window_at(self, x: float, y: float) -> FocusOwner | None: + """Return the first front-to-back layer-0 window containing the point.""" + for info in self._onscreen_app_windows(): + if _point_in_bounds(x, y, info["bounds"]): + return FocusOwner(window_id=info["window_id"], pid=info["pid"]) + return None + + +class WindowsInputFocusProbe: + """Windows: foreground window and ``WindowFromPoint`` root ancestor.""" + + _GA_ROOT = 2 + + @staticmethod + def _user32(): + import ctypes + import ctypes.wintypes as wintypes + + # A private handle: setting prototypes here must not change the + # shared ``ctypes.windll.user32`` the resolver uses. + user32 = ctypes.WinDLL("user32") + user32.GetForegroundWindow.restype = wintypes.HWND + user32.GetForegroundWindow.argtypes = [] + user32.WindowFromPoint.restype = wintypes.HWND + user32.WindowFromPoint.argtypes = [wintypes.POINT] + user32.GetAncestor.restype = wintypes.HWND + user32.GetAncestor.argtypes = [wintypes.HWND, wintypes.UINT] + user32.GetWindowThreadProcessId.restype = wintypes.DWORD + user32.GetWindowThreadProcessId.argtypes = [ + wintypes.HWND, + ctypes.POINTER(wintypes.DWORD), + ] + return user32 + + def _owner(self, user32, hwnd) -> FocusOwner | None: + import ctypes + import ctypes.wintypes as wintypes + + if not hwnd: + return None + root = user32.GetAncestor(hwnd, self._GA_ROOT) or hwnd + pid = wintypes.DWORD() + user32.GetWindowThreadProcessId(root, ctypes.byref(pid)) + return FocusOwner(window_id=int(root), pid=int(pid.value) or None) + + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Return the foreground top-level window and its process.""" + del at_receipt # GetForegroundWindow is already a cheap local read + user32 = self._user32() + return self._owner(user32, user32.GetForegroundWindow()) + + def window_at(self, x: float, y: float) -> FocusOwner | None: + """Return the root ancestor of the window under the physical point.""" + import ctypes.wintypes as wintypes + + user32 = self._user32() + return self._owner(user32, user32.WindowFromPoint(wintypes.POINT(int(x), int(y)))) + + +class X11InputFocusProbe: + """X11: ``_NET_ACTIVE_WINDOW`` and the topmost viewable root child.""" + + def __init__(self, client_factory=None) -> None: + self._client_factory = client_factory + + def _client(self): + from openadapt_capture.window_capture_linux import X11CompositeClient + + return (self._client_factory or X11CompositeClient)() + + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Return the EWMH active client window and its ``_NET_WM_PID``.""" + del at_receipt + with self._client() as client: + window_id = client.active_window() + if not window_id: + return None + return FocusOwner(window_id=int(window_id), pid=client.window_pid(window_id)) + + def window_at(self, x: float, y: float) -> FocusOwner | None: + """Return the client window inside the topmost root child at the point.""" + with self._client() as client: + window_id = client.client_window_at(int(x), int(y)) + if not window_id: + return None + return FocusOwner(window_id=int(window_id), pid=client.window_pid(window_id)) + + +class UnavailableInputFocusProbe: + """A platform without a focus probe: every answer is unknown.""" + + def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None: + """Report unknown so window-scoped input fails closed.""" + del at_receipt + return None + + def window_at(self, x: float, y: float) -> FocusOwner | None: + """Report unknown so window-scoped input fails closed.""" + del x, y + return None + + +def platform_input_focus_probe() -> InputFocusProbe: + """Return this platform's probe; constructing it touches no display.""" + if sys.platform == "darwin": + return MacOSInputFocusProbe() + if sys.platform == "win32": + return WindowsInputFocusProbe() + if sys.platform.startswith("linux"): + return X11InputFocusProbe() + return UnavailableInputFocusProbe() + + +__all__ = [ + "FocusOwner", + "InputFocusProbe", + "MacOSInputFocusProbe", + "UnavailableInputFocusProbe", + "WindowsInputFocusProbe", + "X11InputFocusProbe", + "platform_input_focus_probe", +] diff --git a/packages/capture/tests/test_window_capture.py b/packages/capture/tests/test_window_capture.py index 9f4e548be..c1d3e232f 100644 --- a/packages/capture/tests/test_window_capture.py +++ b/packages/capture/tests/test_window_capture.py @@ -51,6 +51,7 @@ build_window_scope, translate_point, ) +from openadapt_capture.window_input_focus import FocusOwner def test_record_closes_window_scope_when_startup_fails(monkeypatch, tmp_path): @@ -776,6 +777,7 @@ def capturer(window): WindowTarget(owner="FakeApp"), resolver=fake.resolver, capturer=capturer, + focus_probe=fake, ) scope.bind_display_topology( { @@ -2068,18 +2070,45 @@ def test_build_window_scope_with_owner(self): # --------------------------------------------------------------------------- +TARGET = "target" + + class FakePlatform: - """Injectable resolver/capturer simulating a movable Retina window.""" + """Injectable resolver/capturer/focus probe simulating a movable Retina window. + + ``keyboard_owner`` and ``pointer_owner`` model which window receives + input: ``TARGET`` (the fake window), a ``FocusOwner`` for another + window, ``None`` (unknown), or an exception instance to raise. + """ def __init__(self, bounds=(300.0, 150.0, 800.0, 600.0), scale=2.0): self.bounds = bounds self.scale = scale self.window_id = 42 + self.pid = 1234 self.title = "Fake Window" self.missing = False self.on_screen = True self.visibility_independent = False self.process_start_time = 123.5 + self.keyboard_owner = TARGET + self.pointer_owner = TARGET + self.receipt_keyboard_owner = None # None: same as keyboard_owner + + def _owner(self, value): + if value == TARGET: + return FocusOwner(window_id=self.window_id, pid=self.pid) + if isinstance(value, BaseException): + raise value + return value + + def keyboard_focus(self, *, at_receipt=False): + if at_receipt and self.receipt_keyboard_owner is not None: + return self._owner(self.receipt_keyboard_owner) + return self._owner(self.keyboard_owner) + + def window_at(self, x, y): + return self._owner(self.pointer_owner) def resolver(self, target: WindowTarget): if self.missing: @@ -2088,7 +2117,7 @@ def resolver(self, target: WindowTarget): window_id=self.window_id, owner="FakeApp", title=self.title, - pid=1234, + pid=self.pid, bounds=self.bounds, on_screen=self.on_screen, visibility_independent=self.visibility_independent, @@ -2113,6 +2142,7 @@ def scope(fake): WindowTarget(owner="FakeApp"), resolver=fake.resolver, capturer=fake.capturer, + focus_probe=fake, ) result.bind_display_topology( { @@ -2144,6 +2174,7 @@ def capturer(window): WindowTarget(owner="FakeApp"), resolver=fake.resolver, capturer=capturer, + focus_probe=fake, ) capture_scope.bind_display_topology( { @@ -2208,6 +2239,7 @@ def test_resize_uses_exact_axis_scales_after_integer_rounding(self, fake): WindowTarget(owner="FakeApp"), resolver=fake.resolver, capturer=lambda _window: next(images), + focus_probe=fake, ) scope.bind_display_topology( { diff --git a/packages/capture/tests/test_window_input_focus.py b/packages/capture/tests/test_window_input_focus.py new file mode 100644 index 000000000..03f6f645d --- /dev/null +++ b/packages/capture/tests/test_window_input_focus.py @@ -0,0 +1,677 @@ +"""Window-scoped recording keeps only input the target window received. + +These tests drive the real ``record()`` pipeline (journal, input delivery, +screen reader, and the SQLite writer processes) with display-free fakes: the +window resolver, capturer and focus probe come from ``FakePlatform`` and the +native input observer is a ``ThreadedInputObserver`` the test feeds directly. +They run headless on Linux, macOS and Windows. +""" + +from __future__ import annotations + +import json +import multiprocessing +import os +import sqlite3 +import threading +import time +import uuid +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import openadapt_capture.recorder as recorder_module +from openadapt_capture import utils +from openadapt_capture.input_observer import ( + ObservedKey, + ObservedMouseButton, + ObservedMouseMove, + ObservedMouseScroll, + ThreadedInputObserver, +) +from openadapt_capture.recorder import ( + WINDOW_INPUT_GAPS_FILENAME, + WindowInputGapLog, +) +from openadapt_capture.window_capture import ( + INPUT_KEYBOARD_FOCUS_ELSEWHERE, + INPUT_KEYBOARD_FOCUS_UNKNOWN, + INPUT_POINTER_OVER_OTHER_WINDOW, + INPUT_POINTER_TARGET_UNKNOWN, + INPUT_TARGET_NOT_ON_SCREEN, + WindowCaptureScope, + WindowInputNotOnTarget, + WindowTarget, +) +from openadapt_capture.window_input_focus import FocusOwner +from tests.test_window_capture import TARGET, FakePlatform + +SLACK = FocusOwner(window_id=7, pid=555) +PASSWORD_DIALOG = FocusOwner(window_id=8, pid=666) +# Inside the fake target's rectangle (300, 150, 800, 600). +INSIDE = (400.0, 300.0) + + +class FedObserver(ThreadedInputObserver): + """A native observer whose events the test emits through the receipt path.""" + + def __init__(self, callback): + super().__init__( + callback, + observe_keyboard=True, + observe_mouse=True, + capture_mouse_moves=True, + shutdown_timeout=2.0, + ) + self._release_loop = threading.Event() + + def _setup(self): + return + + def _run_loop(self): + self._release_loop.wait() + + def _teardown(self): + return + + def _wake(self): + self._release_loop.set() + + def feed(self, event) -> None: + """Emit one event as a native hook would and wait for its delivery.""" + receipt = self._reserve_receipt(event.timestamp) + self._emit(event, receipt=receipt) + deadline = time.monotonic() + 10 + while not bool(getattr(receipt, "finished", False)): + if time.monotonic() >= deadline: + raise AssertionError("native receipt was not completed") + time.sleep(0.001) + + +def _now() -> float: + return utils.get_timestamp() + + +def key(char: str, pressed: bool, *, name: str | None = None) -> ObservedKey: + return ObservedKey( + pressed=pressed, + key_name=name, + key_char=char, + key_vk=name or char, + canonical_key_name=name, + canonical_key_char=char, + canonical_key_vk=name or char, + timestamp=_now(), + ) + + +def type_text(observer: FedObserver, text: str, *, canary: str | None = None) -> None: + for index, char in enumerate(text): + name = None if canary is None else f"{canary}-{index}" + observer.feed(key(char, True, name=name)) + observer.feed(key(char, False, name=name)) + + +def click(observer: FedObserver, point=INSIDE) -> None: + for pressed in (True, False): + observer.feed( + ObservedMouseButton( + x=point[0], + y=point[1], + button="left", + pressed=pressed, + timestamp=_now(), + ) + ) + + +class WindowRecording: + """Run ``record()`` in window mode against ``FakePlatform``.""" + + def __init__(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fake: FakePlatform): + self.fake = fake + self.capture_dir = tmp_path / "capture" + self.observers: list[FedObserver] = [] + self.error: BaseException | None = None + self.terminate_processing = multiprocessing.Event() + self.terminate_recording = multiprocessing.Event() + self._status_recv, self._status_send = multiprocessing.Pipe(duplex=False) + + for name, value in { + "RECORD_VIDEO": False, + "RECORD_IMAGES": True, + "RECORD_AUDIO": False, + "RECORD_BROWSER_EVENTS": False, + "RECORD_WINDOW_DATA": False, + "RECORD_STRUCTURAL_OBSERVATIONS": False, + "RECORD_READ_ACTIVE_ELEMENT_STATE": False, + "PLOT_PERFORMANCE": False, + "SCREEN_CAPTURE_FPS": 20.0, + }.items(): + monkeypatch.setattr(recorder_module.config, name, value) + + def build_scope(owner, title, **_kwargs): + return WindowCaptureScope( + WindowTarget(owner=owner, title=title), + resolver=fake.resolver, + capturer=fake.capturer, + focus_probe=fake, + ) + + def create_observer(callback, **_kwargs): + observer = FedObserver(callback) + self.observers.append(observer) + return observer + + # Headless: no display metric or screen grab may be read. + monkeypatch.setattr(recorder_module.utils, "get_monitor_dims", lambda: (1920, 1080)) + monkeypatch.setattr(recorder_module.utils, "get_double_click_distance_pixels", lambda: 5) + monkeypatch.setattr(recorder_module.utils, "get_double_click_interval_seconds", lambda: 0.5) + monkeypatch.setattr(recorder_module.platform, "get_display_pixel_ratio", lambda: 2.0) + + def no_display(*_args, **_kwargs): + raise AssertionError("window-scoped recording read the physical display") + + monkeypatch.setattr(recorder_module.utils, "take_screenshot", no_display) + monkeypatch.setattr(recorder_module.utils, "get_process_local_sct", no_display) + monkeypatch.setattr(recorder_module, "build_window_scope", build_scope) + monkeypatch.setattr(recorder_module, "create_input_observer", create_observer) + monkeypatch.setattr( + recorder_module.DesktopCaptureScope, + "current", + lambda: SimpleNamespace( + snapshot=lambda: { + "schema_version": "openadapt.capture.display-topology/v1", + "topology_sha256": "a" * 64, + }, + assert_current=lambda **_kwargs: None, + ), + ) + + def __enter__(self) -> FedObserver: + def run(): + try: + recorder_module.record( + "window input focus", + capture_dir=str(self.capture_dir), + terminate_processing=self.terminate_processing, + terminate_recording=self.terminate_recording, + status_pipe=self._status_send, + window_owner="FakeApp", + ) + except BaseException as exc: # surfaced by __exit__ + self.error = exc + + self._thread = threading.Thread(target=run, daemon=True) + self._thread.start() + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + if self.error is not None: + raise self.error + if self._status_recv.poll(0.05): + if self._status_recv.recv().get("type") == "record.started": + break + else: + raise AssertionError("record() did not start") + assert len(self.observers) == 1 + return self.observers[0] + + def __exit__(self, exc_type, exc, traceback) -> None: + # Let the screen reader publish an after-frame for the last action. + time.sleep(0.3) + self.terminate_processing.set() + self._thread.join(timeout=90) + assert not self._thread.is_alive(), "record() did not stop" + if exc_type is None and self.error is not None: + raise self.error + + def actions(self) -> list[dict]: + db_path = self.capture_dir / "recording.db" + connection = sqlite3.connect(db_path) + connection.row_factory = sqlite3.Row + try: + rows = connection.execute( + "SELECT * FROM action_event ORDER BY source_ordinal" + ).fetchall() + finally: + connection.close() + return [dict(row) for row in rows] + + def gaps(self) -> list[dict]: + path = self.capture_dir / WINDOW_INPUT_GAPS_FILENAME + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text().splitlines()] + + def assert_nowhere(self, *needles: str) -> None: + """No file the recording wrote may contain any ``needles``.""" + files = [path for path in self.capture_dir.rglob("*") if path.is_file()] + assert files, "the recording wrote no files" + for path in files: + data = path.read_bytes() + for needle in needles: + for encoded in (needle.encode("utf-8"), needle.encode("utf-16-le")): + assert encoded not in data, f"{needle!r} was stored in {path.name}" + + +def _key_rows(actions: list[dict]) -> list[dict]: + return [row for row in actions if row["name"] in ("press", "release")] + + +def _click_rows(actions: list[dict]) -> list[dict]: + return [row for row in actions if row["name"] == "click"] + + +@pytest.fixture +def fake(): + return FakePlatform() + + +def test_alt_tab_and_type_drops_keys_and_never_stores_them(tmp_path, monkeypatch, fake, capfd): + canary = f"pw-canary-{uuid.uuid4().hex}" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + type_text(observer, "ok") + fake.keyboard_owner = SLACK # the user switched to another app + type_text(observer, "hunter2", canary=canary) + fake.keyboard_owner = TARGET + type_text(observer, "go") + + keys = _key_rows(recording.actions()) + assert "".join(row["key_char"] for row in keys if row["name"] == "press") == "okgo" + assert all(row["key_name"] is None or canary not in row["key_name"] for row in keys) + gaps = recording.gaps() + left = [gap for gap in gaps if gap["event"] == "focus_left_target"] + assert len(left) == 1 + assert left[0]["reason"] == INPUT_KEYBOARD_FOCUS_ELSEWHERE + assert left[0]["input"] == "keyboard" + assert isinstance(left[0]["timestamp"], float) + assert set(left[0]) == {"schema_version", "event", "reason", "input", "timestamp"} + assert [gap["event"] for gap in gaps] == ["focus_left_target", "focus_returned_to_target"] + recording.assert_nowhere(canary, "hunter2") + output = capfd.readouterr() + assert canary not in output.out + output.err + + +def test_click_on_overlapping_window_inside_target_rectangle_is_dropped( + tmp_path, monkeypatch, fake +): + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + click(observer) + fake.pointer_owner = SLACK # another window covers this point + click(observer) + observer.feed( + ObservedMouseScroll(x=INSIDE[0], y=INSIDE[1], dx=0.0, dy=-3.0, timestamp=_now()) + ) + fake.pointer_owner = TARGET + + actions = recording.actions() + clicks = _click_rows(actions) + assert [row["mouse_pressed"] for row in clicks] == [1, 0] + assert not [row for row in actions if row["name"] == "scroll"] + (left,) = [gap for gap in recording.gaps() if gap["event"] == "focus_left_target"] + assert left["reason"] == INPUT_POINTER_OVER_OTHER_WINDOW + assert left["input"] == "pointer" + assert left["coarse_point"] == [400.0, 300.0] + + +def test_password_dialog_over_target_drops_keys_and_clicks(tmp_path, monkeypatch, fake): + canary = f"pw-canary-{uuid.uuid4().hex}" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + click(observer) + # A different process's password prompt is topmost and focused. + fake.keyboard_owner = PASSWORD_DIALOG + fake.pointer_owner = PASSWORD_DIALOG + click(observer) + type_text(observer, "s3cr3t!", canary=canary) + observer.feed(key("\r", True, name="enter")) + # The dialog closes before the Enter release is delivered. + fake.keyboard_owner = TARGET + fake.pointer_owner = TARGET + observer.feed(key("\r", False, name="enter")) + click(observer) + + actions = recording.actions() + assert _key_rows(actions) == [] # the release of a dropped press is dropped too + assert len(_click_rows(actions)) == 4 + reasons = [gap["reason"] for gap in recording.gaps() if gap["event"] == "focus_left_target"] + assert reasons == [INPUT_POINTER_OVER_OTHER_WINDOW, INPUT_KEYBOARD_FOCUS_ELSEWHERE] + recording.assert_nowhere(canary, "s3cr3t!") + + +@pytest.mark.parametrize("failure", [None, RuntimeError("probe failed")], ids=["none", "raises"]) +def test_unknown_focus_fails_closed(tmp_path, monkeypatch, fake, failure): + canary = f"pw-canary-{uuid.uuid4().hex}" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + fake.keyboard_owner = failure + fake.pointer_owner = failure + type_text(observer, "abc", canary=canary) + click(observer) + fake.keyboard_owner = TARGET + fake.pointer_owner = TARGET + + actions = recording.actions() + assert _key_rows(actions) == [] + assert _click_rows(actions) == [] + reasons = [gap["reason"] for gap in recording.gaps() if gap["event"] == "focus_left_target"] + assert reasons == [INPUT_KEYBOARD_FOCUS_UNKNOWN, INPUT_POINTER_TARGET_UNKNOWN] + recording.assert_nowhere(canary) + + +def test_input_on_target_is_recorded_as_before(tmp_path, monkeypatch, fake): + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + observer.feed(ObservedMouseMove(x=310.0, y=170.0, timestamp=_now())) + click(observer, (310.0, 170.0)) + type_text(observer, "hi") + + actions = recording.actions() + assert [row["name"] for row in actions] == [ + "move", + "click", + "click", + "press", + "release", + "press", + "release", + ] + first_click = _click_rows(actions)[0] + assert (first_click["mouse_x"], first_click["mouse_y"]) == (20.0, 40.0) + assert all(row["window_geometry_generation"] is not None for row in actions) + assert recording.gaps() == [] + assert not (recording.capture_dir / WINDOW_INPUT_GAPS_FILENAME).exists() + + +def test_focus_change_between_receipt_and_delivery_is_dropped(tmp_path, monkeypatch, fake): + """Focus read inside the native hook is binding even if it returns later.""" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + fake.receipt_keyboard_owner = PASSWORD_DIALOG + observer.feed(key("x", True)) + observer.feed(key("x", False)) + fake.receipt_keyboard_owner = None + + assert _key_rows(recording.actions()) == [] + + +def test_release_of_an_accepted_press_survives_a_focus_change(tmp_path, monkeypatch, fake): + """Cmd+Tab: the press reached the target, so the release is kept.""" + recording = WindowRecording(tmp_path, monkeypatch, fake) + with recording as observer: + observer.feed(key("", True, name="cmd")) + observer.feed(key("\t", True, name="tab")) + fake.keyboard_owner = SLACK + observer.feed(key("\t", False, name="tab")) + observer.feed(key("", False, name="cmd")) + + keys = _key_rows(recording.actions()) + assert [(row["name"], row["key_name"]) for row in keys] == [ + ("press", "cmd"), + ("press", "tab"), + ("release", "tab"), + ("release", "cmd"), + ] + assert recording.gaps() == [] + + +# --------------------------------------------------------------------------- +# Scope-level decisions (no pipeline) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def scope(fake): + result = WindowCaptureScope( + WindowTarget(owner="FakeApp"), + resolver=fake.resolver, + capturer=fake.capturer, + focus_probe=fake, + ) + result.bind_display_topology( + { + "schema_version": "openadapt.capture.display-topology/v1", + "topology_sha256": "a" * 64, + }, + lambda **_kwargs: None, + ) + result.capture_frame() + return result + + +def test_offscreen_visibility_independent_target_refuses_clicks(fake): + fake.on_screen = False + fake.visibility_independent = True + window = fake.resolver(WindowTarget(owner="FakeApp")) + scope = WindowCaptureScope( + WindowTarget(owner="FakeApp"), + resolver=fake.resolver, + capturer=fake.capturer, + focus_probe=fake, + ) + with pytest.raises(WindowInputNotOnTarget) as excinfo: + scope.assert_input_on_target(window, point=INSIDE) + assert excinfo.value.reason == INPUT_TARGET_NOT_ON_SCREEN + + +def test_keyboard_focus_on_target_process_is_accepted(scope, fake): + fake.keyboard_owner = FocusOwner(window_id=None, pid=fake.pid) + assert isinstance(scope.generation_for_action(), int) + + +def test_legacy_translation_refuses_other_window(scope, fake): + fake.pointer_owner = SLACK + with pytest.raises(WindowInputNotOnTarget): + scope.translate(*INSIDE) + fake.pointer_owner = TARGET + assert scope.translate(310.0, 170.0) == (20.0, 40.0) + + +def test_assert_current_does_not_need_focus(scope, fake): + """Stopping from another app's UI must not fail the recording.""" + fake.keyboard_owner = SLACK + fake.pointer_owner = SLACK + scope.assert_current() + + +def test_gap_log_never_records_more_than_reason_kind_time_and_coarse_point(tmp_path): + log = WindowInputGapLog(tmp_path) + log.left_target(INPUT_POINTER_OVER_OTHER_WINDOW, "pointer", 1.5, (123.4, 456.7)) + log.left_target(INPUT_POINTER_OVER_OTHER_WINDOW, "pointer", 1.6, (999.0, 999.0)) + log.returned(2.0) + lines = (tmp_path / WINDOW_INPUT_GAPS_FILENAME).read_text().splitlines() + assert [json.loads(line) for line in lines] == log.records + assert log.records[0]["coarse_point"] == [120.0, 460.0] + assert len(log.records) == 2 + if os.name == "posix": + assert (tmp_path / WINDOW_INPUT_GAPS_FILENAME).stat().st_mode & 0o077 == 0 + + +# --------------------------------------------------------------------------- +# Platform probes with faked native bindings +# --------------------------------------------------------------------------- + + +def _quartz_window(window_id, pid, bounds, *, layer=0, alpha=1.0): + x, y, w, h = bounds + return { + "kCGWindowNumber": window_id, + "kCGWindowOwnerPID": pid, + "kCGWindowLayer": layer, + "kCGWindowAlpha": alpha, + "kCGWindowBounds": {"X": x, "Y": y, "Width": w, "Height": h}, + } + + +@pytest.fixture +def fake_macos(monkeypatch): + import sys + + state = SimpleNamespace(frontmost_pid=1234, windows=[]) + quartz = SimpleNamespace( + kCGWindowListOptionOnScreenOnly=1, + kCGWindowListExcludeDesktopElements=16, + kCGNullWindowID=0, + CGWindowListCopyWindowInfo=lambda _options, _window: list(state.windows), + ) + app = SimpleNamespace(processIdentifier=lambda: state.frontmost_pid) + workspace = SimpleNamespace(frontmostApplication=lambda: app) + appkit = SimpleNamespace( + NSWorkspace=SimpleNamespace(sharedWorkspace=lambda: workspace), + ) + monkeypatch.setitem(sys.modules, "Quartz", quartz) + monkeypatch.setitem(sys.modules, "AppKit", appkit) + return state + + +def test_macos_probe_requires_window_list_to_agree_with_frontmost_app(fake_macos): + from openadapt_capture.window_input_focus import MacOSInputFocusProbe + + probe = MacOSInputFocusProbe() + fake_macos.windows = [ + _quartz_window(1, 99, (0, 0, 2000, 25), layer=25), # menu bar extra + _quartz_window(42, 1234, (300, 150, 800, 600)), + ] + assert probe.keyboard_focus() == FocusOwner(pid=1234) + # NSWorkspace can be stale; the front-most app window belongs to Slack. + fake_macos.windows.insert(1, _quartz_window(7, 555, (0, 0, 500, 500))) + assert probe.keyboard_focus() is None + # Inside the native hook only the cheap read runs. + assert probe.keyboard_focus(at_receipt=True) == FocusOwner(pid=1234) + + +def test_macos_probe_window_at_is_front_to_back_layer_zero(fake_macos): + from openadapt_capture.window_input_focus import MacOSInputFocusProbe + + probe = MacOSInputFocusProbe() + fake_macos.windows = [ + _quartz_window(3, 77, (0, 0, 2000, 2000), layer=101), # overlay + _quartz_window(4, 78, (0, 0, 2000, 2000), alpha=0.0), # invisible + _quartz_window(7, 555, (350, 250, 200, 200)), # Slack over the target + _quartz_window(42, 1234, (300, 150, 800, 600)), + ] + assert probe.window_at(400.0, 300.0) == FocusOwner(window_id=7, pid=555) + assert probe.window_at(310.0, 170.0) == FocusOwner(window_id=42, pid=1234) + assert probe.window_at(5000.0, 5000.0) is None + + +def test_windows_probe_uses_root_ancestor_of_foreground_and_point(monkeypatch): + import ctypes + + try: + import ctypes.wintypes # noqa: F401 + except (ImportError, ValueError): + pytest.skip("ctypes.wintypes is unavailable on this interpreter") + from openadapt_capture.window_input_focus import WindowsInputFocusProbe + + class Function: + def __init__(self, impl): + self.impl = impl + + def __call__(self, *args): + return self.impl(*args) + + roots = {101: 42, 202: 7} + pids = {42: 1234, 7: 555} + points = [] + + def thread_process_id(hwnd, pid_ref): + pid_ref._obj.value = pids[int(hwnd)] + return 1 + + def window_from_point(point): + points.append((point.x, point.y)) + return 202 + + user32 = SimpleNamespace( + GetForegroundWindow=Function(lambda: 101), + WindowFromPoint=Function(window_from_point), + GetAncestor=Function(lambda hwnd, flag: roots[int(hwnd)] if flag == 2 else None), + GetWindowThreadProcessId=Function(thread_process_id), + ) + monkeypatch.setattr(ctypes, "WinDLL", lambda _name: user32, raising=False) + + probe = WindowsInputFocusProbe() + assert probe.keyboard_focus() == FocusOwner(window_id=42, pid=1234) + assert probe.window_at(400.7, 300.2) == FocusOwner(window_id=7, pid=555) + assert points == [(400, 300)] + user32.GetForegroundWindow = Function(lambda: None) + assert probe.keyboard_focus() is None + + +class _X11Reply(SimpleNamespace): + def reply(self): + return self + + +class _FakeX11Core: + """QueryTree/GetWindowAttributes/GetGeometry over a fake window tree.""" + + VIEWABLE = 2 + INPUT_OUTPUT = 1 + + def __init__(self, tree, geometry, unmapped=()): + self.tree = tree + self.geometry = geometry + self.unmapped = set(unmapped) + + def QueryTree(self, window): # noqa: N802 - X11 request name + return _X11Reply(children=list(self.tree.get(window, []))) + + def GetWindowAttributes(self, window): # noqa: N802 + return _X11Reply( + map_state=0 if window in self.unmapped else self.VIEWABLE, + _class=self.INPUT_OUTPUT, + ) + + def GetGeometry(self, window): # noqa: N802 + x, y, w, h = self.geometry[window] + return _X11Reply(x=x, y=y, width=w, height=h, border_width=0) + + +def test_x11_client_window_at_maps_topmost_frame_to_its_client(): + from openadapt_capture.window_capture_linux import X11CompositeClient + + root = 1 + # Root children bottom-to-top: target frame 10 (client 42), Slack frame + # 20 (client 7), then an unmapped frame 30 that covers everything. + core = _FakeX11Core( + tree={root: [10, 20, 30], 10: [11], 11: [42], 20: [7]}, + geometry={ + 10: (300, 150, 800, 600), + 20: (350, 250, 200, 200), + 30: (0, 0, 1920, 1080), + }, + unmapped={30}, + ) + client = object.__new__(X11CompositeClient) + client._conn = SimpleNamespace(core=core) + client._root = root + client._xproto = SimpleNamespace( + MapState=SimpleNamespace(Viewable=_FakeX11Core.VIEWABLE), + WindowClass=SimpleNamespace(InputOutput=_FakeX11Core.INPUT_OUTPUT), + ) + client.window_ids = lambda: [42, 7] + + assert client.client_window_at(400, 300) == 7 + assert client.client_window_at(310, 170) == 42 + assert client.client_window_at(1500, 900) is None + + +def test_x11_probe_reports_active_window_and_point_owner(): + from contextlib import nullcontext + + from openadapt_capture.window_input_focus import X11InputFocusProbe + + client = SimpleNamespace( + active_window=lambda: 42, + client_window_at=lambda x, y: 7 if (x, y) == (400, 300) else None, + window_pid=lambda window_id: {42: 1234, 7: 555}.get(window_id), + ) + probe = X11InputFocusProbe(client_factory=lambda: nullcontext(client)) + assert probe.keyboard_focus() == FocusOwner(window_id=42, pid=1234) + assert probe.window_at(400.4, 300.9) == FocusOwner(window_id=7, pid=555) + assert probe.window_at(0, 0) is None From 94cec2a870136d5ef02fc4c4652dfb59c6556250 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 23:25:16 -0400 Subject: [PATCH 2/2] fix(capture): skip minimized and cloaked windows when resolving on Windows IsWindowVisible is true for a minimized window and for a window DWM cloaks (another virtual desktop, a suspended UWP frame). Neither shows pixels at its bounds, so the region grab would capture whatever else is there, and the largest-match rule could prefer such a window over the visible one. The resolver now skips windows for which IsIconic is true or DwmGetWindowAttribute(DWMWA_CLOAKED) is non-zero. An unreadable cloak state counts as cloaked. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_capture/window_capture.py | 27 +++++ .../tests/test_window_resolver_windows.py | 110 ++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 packages/capture/tests/test_window_resolver_windows.py diff --git a/packages/capture/openadapt_capture/window_capture.py b/packages/capture/openadapt_capture/window_capture.py index ee82866cb..6931d3d02 100644 --- a/packages/capture/openadapt_capture/window_capture.py +++ b/packages/capture/openadapt_capture/window_capture.py @@ -2104,6 +2104,8 @@ def _resolve_window_windows(target: WindowTarget) -> TargetWindow | None: def _enum_cb(hwnd: int, _lparam: int) -> bool: if not user32.IsWindowVisible(hwnd): return True + if user32.IsIconic(hwnd) or _window_cloaked(hwnd): + return True # minimized, or hidden by DWM (another virtual desktop) length = user32.GetWindowTextLengthW(hwnd) buf = ctypes.create_unicode_buffer(length + 1) user32.GetWindowTextW(hwnd, buf, length + 1) @@ -2151,6 +2153,31 @@ def _enum_cb(hwnd: int, _lparam: int) -> bool: return max(matches, key=lambda m: m.bounds[2] * m.bounds[3]) +def _window_cloaked(hwnd: int) -> bool: + """Return whether DWM cloaks ``hwnd``; an unreadable state counts as cloaked. + + A cloaked window (on another virtual desktop, or a suspended UWP frame) + passes ``IsWindowVisible`` but shows no pixels, so a region grab of its + bounds would capture whatever else is on screen there. + """ + import ctypes + import ctypes.wintypes as wintypes + + try: + dwmapi = ctypes.windll.dwmapi + except (AttributeError, OSError): # pragma: no cover - supported Windows has DWM + return True + DWMWA_CLOAKED = 14 + cloaked = wintypes.DWORD() + res = dwmapi.DwmGetWindowAttribute( + wintypes.HWND(hwnd), + wintypes.DWORD(DWMWA_CLOAKED), + ctypes.byref(cloaked), + ctypes.sizeof(cloaked), + ) + return res != 0 or cloaked.value != 0 + + def _window_rect(hwnd: int) -> tuple[int, int, int, int] | None: """Return DWM physical bounds; never mix DPI-virtualized coordinates.""" import ctypes diff --git a/packages/capture/tests/test_window_resolver_windows.py b/packages/capture/tests/test_window_resolver_windows.py new file mode 100644 index 000000000..5f099cfc9 --- /dev/null +++ b/packages/capture/tests/test_window_resolver_windows.py @@ -0,0 +1,110 @@ +"""The Win32 window resolver skips windows that show no pixels. + +Runs on every platform: ``ctypes.windll``, ``WINFUNCTYPE`` and ``psutil`` +are replaced with fakes. +""" + +from __future__ import annotations + +import ctypes +from types import SimpleNamespace + +import pytest + +import openadapt_capture.window_capture as window_capture_module +from openadapt_capture.window_capture import WindowTarget + +try: + import ctypes.wintypes # noqa: F401 +except (ImportError, ValueError): # pragma: no cover - very old interpreters + pytest.skip("ctypes.wintypes is unavailable on this interpreter", allow_module_level=True) + +DWMWA_EXTENDED_FRAME_BOUNDS = 9 +DWMWA_CLOAKED = 14 + + +@pytest.fixture +def win32(monkeypatch): + """Fake Win32: three visible 'Target' windows, one minimized, one cloaked.""" + windows = { + 101: {"title": "Target", "iconic": False, "cloaked": 0, "rect": (0, 0, 400, 300)}, + 102: {"title": "Target", "iconic": True, "cloaked": 0, "rect": (0, 0, 1600, 1200)}, + 103: {"title": "Target", "iconic": False, "cloaked": 2, "rect": (0, 0, 1800, 1000)}, + } + + def get_text_length(hwnd): + return len(windows[int(hwnd)]["title"]) + + def get_text(hwnd, buffer, _size): + buffer.value = windows[int(hwnd)]["title"] + return len(buffer.value) + + def get_pid(hwnd, pid_ref): + pid_ref._obj.value = 4000 + int(hwnd) + return 1 + + def enum_windows(callback, _lparam): + for hwnd in windows: + if not callback(hwnd, 0): + break + return True + + def dwm_attribute(hwnd, attribute, value_ref, _size): + window = windows[int(hwnd.value)] + if attribute.value == DWMWA_CLOAKED: + value_ref._obj.value = window["cloaked"] + elif attribute.value == DWMWA_EXTENDED_FRAME_BOUNDS: + rect = value_ref._obj + rect.left, rect.top, rect.right, rect.bottom = window["rect"] + else: + return 1 + return 0 + + user32 = SimpleNamespace( + IsWindowVisible=lambda _hwnd: True, + IsIconic=lambda hwnd: windows[int(hwnd)]["iconic"], + GetWindowTextLengthW=get_text_length, + GetWindowTextW=get_text, + GetWindowThreadProcessId=get_pid, + EnumWindows=enum_windows, + ) + dwmapi = SimpleNamespace(DwmGetWindowAttribute=dwm_attribute) + monkeypatch.setattr( + ctypes, "windll", SimpleNamespace(user32=user32, dwmapi=dwmapi), raising=False + ) + monkeypatch.setattr(ctypes, "WINFUNCTYPE", lambda *_types: lambda fn: fn, raising=False) + + import psutil + + class Process: + def __init__(self, pid): + self.pid = pid + + def name(self): + return "target.exe" + + def create_time(self): + return 100.0 + + monkeypatch.setattr(psutil, "Process", Process) + return windows + + +def test_minimized_and_cloaked_windows_are_skipped(win32): + window = window_capture_module._resolve_window_windows(WindowTarget(title="Target")) + assert window is not None + assert window.window_id == 101 # the larger minimized/cloaked matches lose + + +def test_unreadable_cloak_state_is_treated_as_cloaked(win32, monkeypatch): + win32[101]["cloaked"] = 0 + dwmapi = ctypes.windll.dwmapi + original = dwmapi.DwmGetWindowAttribute + + def failing(hwnd, attribute, value_ref, size): + if attribute.value == DWMWA_CLOAKED: + return 0x80070057 # E_INVALIDARG + return original(hwnd, attribute, value_ref, size) + + monkeypatch.setattr(dwmapi, "DwmGetWindowAttribute", failing) + assert window_capture_module._resolve_window_windows(WindowTarget(title="Target")) is None