Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 38 additions & 7 deletions packages/capture/openadapt_capture/recorder.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@
observe_structural_action,
)
from openadapt_capture.window_capture import (
FOCUS_EVIDENCE_WINDOW,
WindowCaptureAmbiguousError,
WindowCaptureError,
WindowCapturePermissionError,
Expand Down Expand Up @@ -361,20 +362,26 @@ def __init__(
self._keyboard_focus_at_receipt = keyboard_focus_at_receipt
self._admitted = False

def admit(self, point: tuple[float, float] | None) -> None:
def admit(self, point: tuple[float, float] | None) -> str:
"""Accept input only if the target window received it.

Returns:
The focus evidence level (see ``assert_input_on_target``).

Raises:
WindowInputNotOnTarget: The input went to another window.
"""
if point is None:
self._window_scope.assert_input_on_target(
evidence = 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)
evidence = self._window_scope.assert_input_on_target(
self._geometry[0], point=point
)
self._admitted = True
return evidence

def admit_continuation(self) -> None:
"""Accept the release or drag of a press the target already received."""
Expand Down Expand Up @@ -469,8 +476,14 @@ class WindowInputGapLog:
input kind, a timestamp and, for pointer input, a point rounded to a
coarse grid. It never holds key identity or text.

A ``keyboard_focus_evidence`` marker is written when the evidence for
accepted keyboard input changes level: ``process`` means the OS named
only the target's process, not the target window, so the input could
have gone to another window of the same app.

Markers are appended to ``window-input-gaps.jsonl`` in the capture
directory, created only when input is dropped, and kept in ``records``.
directory, created only when input is dropped or its evidence weakens,
and kept in ``records``.
"""

def __init__(self, capture_dir: str | os.PathLike[str] | None = None) -> None:
Expand All @@ -479,6 +492,7 @@ def __init__(self, capture_dir: str | os.PathLike[str] | None = None) -> None:
)
self._lock = threading.Lock()
self._active_reason: str | None = None
self._keyboard_evidence = FOCUS_EVIDENCE_WINDOW
self.records: list[dict[str, Any]] = []

def _append_locked(self, record: dict[str, Any]) -> None:
Expand Down Expand Up @@ -528,6 +542,21 @@ def left_target(
reason,
)

def keyboard_evidence(self, evidence: str, timestamp: float | None) -> None:
"""Mark a change in how accepted keyboard input reached the target."""
with self._lock:
if evidence == self._keyboard_evidence:
return
self._keyboard_evidence = evidence
self._append_locked(
{
"schema_version": WINDOW_INPUT_GAP_SCHEMA_VERSION,
"event": "keyboard_focus_evidence",
"evidence": evidence,
"timestamp": timestamp,
}
)

def returned(self, timestamp: float | None) -> None:
"""Close the current run of dropped input, if any."""
with self._lock:
Expand Down Expand Up @@ -2648,18 +2677,20 @@ def admit(self, event: ObservedInput, reservation: object | None) -> bool:
point = None if isinstance(event, ObservedKey) else (event.x, event.y)
try:
if isinstance(reservation, WindowActionReservation):
reservation.admit(point)
evidence = reservation.admit(point)
else:
window = self._scope.reserve_action_geometry()[0]
if point is None:
self._scope.assert_input_on_target(window)
evidence = self._scope.assert_input_on_target(window)
else:
self._scope.assert_input_on_target(window, point=point)
evidence = 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._gap_log.keyboard_evidence(evidence, event.timestamp)
if isinstance(event, ObservedKey):
self._held_keys.add(_key_identity(event))
elif isinstance(event, ObservedMouseButton):
Expand Down
37 changes: 28 additions & 9 deletions packages/capture/openadapt_capture/window_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,11 +147,17 @@ class WindowCaptureUnavailableError(WindowCaptureError):


INPUT_KEYBOARD_FOCUS_ELSEWHERE = "keyboard_focus_elsewhere"
INPUT_KEYBOARD_FOCUS_IN_DIALOG = "keyboard_focus_in_dialog"
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()
# How an accepted input was shown to reach the target. ``window``: the OS
# named the target window as the receiver. ``process``: it named only the
# target's process, which cannot rule out another window of the same app.
FOCUS_EVIDENCE_WINDOW = "window"
FOCUS_EVIDENCE_PROCESS = "process"


class WindowInputNotOnTarget(WindowCaptureError):
Expand Down Expand Up @@ -643,13 +649,21 @@ def assert_input_on_target(
*,
point: tuple[float, float] | None = None,
keyboard_focus_at_receipt: object = _NOT_PROBED,
) -> None:
) -> str:
"""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.
A keyboard event needs keyboard focus on ``window``, both at native
receipt (when a receipt reading is given) and now. A reading that
names a window must name ``window`` itself: another window of the
same process (a second browser window, an app's password dialog) is
a refusal. Only a reading that names no window can fall back to the
process, and a dialog or sheet never counts. An unknown answer or a
probe failure is a refusal.

Returns:
``FOCUS_EVIDENCE_WINDOW`` when some reading named the target
window, else ``FOCUS_EVIDENCE_PROCESS`` (weaker evidence).

Raises:
WindowInputNotOnTarget: With a stable reason code.
Expand All @@ -667,24 +681,29 @@ def assert_input_on_target(
raise WindowInputNotOnTarget(INPUT_POINTER_TARGET_UNKNOWN)
if owner.window_id != window.window_id:
raise WindowInputNotOnTarget(INPUT_POINTER_OVER_OTHER_WINDOW)
return
return FOCUS_EVIDENCE_WINDOW
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)
evidence = FOCUS_EVIDENCE_PROCESS
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:
if owner.dialog:
raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_IN_DIALOG)
if owner.window_id is not None:
if owner.window_id != window.window_id:
raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_ELSEWHERE)
evidence = FOCUS_EVIDENCE_WINDOW
elif owner.pid != window.pid:
raise WindowInputNotOnTarget(INPUT_KEYBOARD_FOCUS_ELSEWHERE)
return evidence

def _geometry_for_action(
self,
Expand Down
151 changes: 139 additions & 12 deletions packages/capture/openadapt_capture/window_input_focus.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,15 @@ class FocusOwner:
``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).
A probe that can name the focused window must set it: the scope then
requires that exact window, and a matching ``pid`` alone is not enough.
``dialog`` is set when focus is in a dialog or sheet, which never counts
as the target even when the target's own process owns it.
"""

window_id: int | None = None
pid: int | None = None
dialog: bool = False


class InputFocusProbe(Protocol):
Expand All @@ -55,11 +60,57 @@ def _point_in_bounds(x: float, y: float, bounds: tuple[float, float, float, floa
return left <= x < left + width and top <= y < top + height


# Quartz window levels: kCGNormalWindowLevel and kCGStatusWindowLevel.
# A window between them is raised above document windows: a floating panel
# (3), a modal panel such as an NSAlert or password prompt (8), a torn-off
# menu (3). The status level and above are the menu bar, status items and
# system overlays, which never take an app's keyboard focus.
_MACOS_NORMAL_LAYER = 0
_MACOS_STATUS_LAYER = 25
# Accessibility subroles of a focused window that is a dialog, not a
# document window. Typing there never counts as typing into the target.
_MACOS_DIALOG_SUBROLES = frozenset({"AXDialog", "AXSystemDialog"})
_MACOS_AX_FOCUS_TIMEOUT_SECONDS = 0.25
_MACOS_AX_MAX_ANCESTORS = 32


def _macos_ax_window_number(element: object) -> int | None:
"""Return the CGWindowID behind an AX window element, or ``None``.

Uses the long-standing HIServices ``_AXUIElementGetWindow`` call. There
is no public API for this mapping; any failure is reported as unknown.
"""
try:
import ctypes

import objc

library = ctypes.CDLL(
"/System/Library/Frameworks/ApplicationServices.framework/ApplicationServices"
)
get_window = library._AXUIElementGetWindow
get_window.restype = ctypes.c_int32
get_window.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)]
number = ctypes.c_uint32(0)
if get_window(objc.pyobjc_id(element), ctypes.byref(number)) != 0:
return None
return int(number.value) or None
except Exception:
return None


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.

The focused window is the front-most on-screen layer-0 window of the
frontmost application. A raised window of that application (a modal or
floating panel) in front of it takes keyboard focus, so it is reported
as a dialog. When Accessibility can name the application's
``AXFocusedWindow``, it must agree with that window, and a focused dialog
or sheet is reported as such.
"""

@staticmethod
Expand All @@ -74,15 +125,21 @@ def _frontmost_pid() -> int | None:

@staticmethod
def _onscreen_app_windows() -> Iterator[dict]:
"""Yield on-screen layer-0 windows front to back, lazily."""
"""Yield on-screen app windows front to back, lazily.

These are document windows (layer 0) and the raised windows above
them up to, not including, the status level: floating panels and
modal panels. Each item carries its ``layer``.
"""
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:
layer = int(info.get("kCGWindowLayer", 0) or 0)
if not _MACOS_NORMAL_LAYER <= layer < _MACOS_STATUS_LAYER:
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
Expand All @@ -99,27 +156,97 @@ def _onscreen_app_windows() -> Iterator[dict]:
"window_id": int(info.get("kCGWindowNumber", 0) or 0),
"pid": int(info.get("kCGWindowOwnerPID", 0) or 0),
"bounds": bounds,
"layer": layer,
}

def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None:
"""Return the frontmost application's process.
@staticmethod
def _ax_focused_window(pid: int) -> tuple[int | None, bool] | None:
"""Return ``(window number, is dialog)`` for the app's focused window.

``None`` when Accessibility cannot name a focused window (not
trusted, timed out, or the app exposes none).
"""
try:
import ApplicationServices as ax

success = ax.kAXErrorSuccess

def attribute(element: object, name: str) -> object | None:
error, value = ax.AXUIElementCopyAttributeValue(element, name, None)
return value if error == success else None

application = ax.AXUIElementCreateApplication(pid)
if (
ax.AXUIElementSetMessagingTimeout(
application, _MACOS_AX_FOCUS_TIMEOUT_SECONDS
)
!= success
):
return None
window = attribute(application, "AXFocusedWindow")
if window is None or attribute(window, "AXRole") != "AXWindow":
return None
dialog = attribute(window, "AXSubrole") in _MACOS_DIALOG_SUBROLES
# A sheet hangs off its parent window. Focus inside it still
# names the parent as the focused window.
element = attribute(application, "AXFocusedUIElement")
for _ in range(_MACOS_AX_MAX_ANCESTORS):
if element is None or dialog:
break
if attribute(element, "AXRole") == "AXSheet":
dialog = True
break
element = attribute(element, "AXParent")
return _macos_ax_window_number(window), dialog
except Exception:
return None

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.
def keyboard_focus(self, *, at_receipt: bool = False) -> FocusOwner | None:
"""Return the focused window and its process.

Inside the event tap only the cheap ``NSWorkspace`` read runs and the
answer names the process alone. At delivery the front-most on-screen
application window must belong to that process. ``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. That
window is the focused window, unless one of the application's raised
windows is in front of it (a dialog), or Accessibility names a
different focused window (unknown) or a dialog or sheet.
"""
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)
raised: dict | None = None
front: dict | None = None
for info in self._onscreen_app_windows():
if info["layer"] != _MACOS_NORMAL_LAYER:
if info["pid"] == pid and raised is None:
raised = info
continue
front = info
break
if front is None or front["pid"] != pid:
return None
return FocusOwner(pid=pid)
if raised is not None:
# A modal or floating panel of the app sits over its document
# windows. Keys typed while it is up go to the panel.
return FocusOwner(window_id=raised["window_id"] or None, pid=pid, dialog=True)
window_id = front["window_id"] or None
focused = self._ax_focused_window(pid)
if focused is not None:
ax_window_id, dialog = focused
if dialog:
return FocusOwner(window_id=ax_window_id or window_id, pid=pid, dialog=True)
if ax_window_id is not None and ax_window_id != window_id:
return None
return FocusOwner(window_id=window_id, pid=pid)

def window_at(self, x: float, y: float) -> FocusOwner | None:
"""Return the first front-to-back layer-0 window containing the point."""
"""Return the first front-to-back app window containing the point.

A raised panel over the target (a modal password prompt) is returned
as itself, so a click on it never counts as a click on the target.
"""
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"])
Expand Down
Loading
Loading