diff --git a/packages/capture/openadapt_capture/capture.py b/packages/capture/openadapt_capture/capture.py index 30189f8f5..4f50deb90 100644 --- a/packages/capture/openadapt_capture/capture.py +++ b/packages/capture/openadapt_capture/capture.py @@ -1322,6 +1322,22 @@ def desktop_capture(self) -> dict | None: return config.get("capture_desktop") return None + @property + def locale_facts(self) -> dict | None: + """Locale, date order and keyboard layout read when recording began. + + ``{"locale", "date_order", "keyboard_layout"}``, each a normalized + string or None when the OS didn't report it (see + ``locale_facts.py``). None for a recording made before capture + persisted them. Converters carry this as it is and never replace it + with the converting computer's settings. + """ + + config = getattr(self._recording, "config", None) + if isinstance(config, dict): + return config.get("capture_locale") + return None + def narration(self) -> list | None: """Scrubbed spoken narration as timed sentences, or None. diff --git a/packages/capture/openadapt_capture/locale_facts.py b/packages/capture/openadapt_capture/locale_facts.py new file mode 100644 index 000000000..118fdd8cb --- /dev/null +++ b/packages/capture/openadapt_capture/locale_facts.py @@ -0,0 +1,445 @@ +"""Locale, date order and keyboard layout at recording time (#1897). + +Read once when a recording starts and persisted in the recording's config +under ``capture_locale``, so a converter carries the recording computer's +facts instead of reading its own. A fact the OS doesn't report is ``None``; +it is never filled in with ``en_US``, ``mdy`` or ``us``. + +The readers match ``openadapt_flow.locale_invariants``, which reads the same +facts at replay; Flow's tests check the two agree. Standard library only, so +capture gains no dependency. + +* macOS: ``AppleLocale``, ``AppleICUDateFormatStrings``, and the current + keyboard layout and input method from Text Input Sources. +* Windows: ``GetUserDefaultLocaleName``, ``GetLocaleInfoEx`` (short date), and + the HKL of the foreground window's thread. +* Linux: the user locale via ``setlocale(LC_TIME, "")``, and the X server's + active XKB group. Wayland is unknown. +""" + +from __future__ import annotations + +import locale +import os +import re +import subprocess +import sys +import threading +from typing import Callable, Literal, Optional + +DateOrder = Literal["mdy", "dmy", "ymd"] + +__all__ = ["DateOrder", "observe_locale_facts", "read_os_locale_facts"] + + +def _norm(value: str) -> str: + return value.strip().replace("-", "_").lower() + + +def _locale_family(value: str) -> str: + """``en_us.utf-8`` / ``de_de@rg=chzzzz`` -> ``en_us`` / ``de_de``.""" + + return re.split(r"[.@]", _norm(value), maxsplit=1)[0] + + +_STRFTIME = re.compile(r"%[-_0^#]?([a-zA-Z])") +_STRFTIME_KIND = { + "d": "d", + "e": "d", + "m": "m", + "b": "m", + "B": "m", + "h": "m", + "y": "y", + "Y": "y", + "C": "y", +} + + +def date_order_from_format(fmt: str) -> Optional[DateOrder]: + """The day/month/year order of a strftime (``%d.%m.%Y``, ``%e/%m/%y``, + ``%D``) or ICU/Windows (``M/d/yyyy``, ``dd.MM.yyyy``) date pattern. + + None when the pattern doesn't name all three parts. + """ + + text = (fmt or "").strip() + if not text: + return None + seen: list[str] = [] + if "%" in text: + text = text.replace("%D", "%m/%d/%y").replace("%F", "%Y-%m-%d") + for m in _STRFTIME.finditer(text): + kind = _STRFTIME_KIND.get(m.group(1)) + if kind and kind not in seen: + seen.append(kind) + else: + # Drop quoted literals, then read runs of pattern letters. A run of + # three or more d/E is a weekday name, not the day of the month. + text = re.sub(r"'[^']*'", "", text) + for m in re.finditer(r"([dMLyuE])\1*", text): + letter, size = m.group(1), len(m.group(0)) + if letter == "E" or (letter == "d" and size >= 3): + continue + kind = {"d": "d", "M": "m", "L": "m", "y": "y", "u": "y"}[letter] + if kind not in seen: + seen.append(kind) + order = "".join(seen) + if order in ("mdy", "dmy", "ymd"): + return order # type: ignore[return-value] + return None + + +# -- OS readers (no GUI) ------------------------------------------------------------ +# +# Kept in step with openadapt_flow.locale_invariants (the replay-time +# reader); Flow's tests/test_locale_invariants.py checks the two agree. + +_Run = Callable[[list[str]], Optional[str]] +_GroupReader = Callable[[str], Optional[int]] +_LOCALE_LOCK = threading.Lock() + + +def _run(cmd: list[str]) -> Optional[str]: + try: + out = subprocess.run( + cmd, capture_output=True, text=True, timeout=5, check=False + ) + except (OSError, subprocess.SubprocessError): + return None + if out.returncode != 0: + return None + return out.stdout + + +def _posix_user_time_locale() -> tuple[Optional[str], Optional[DateOrder]]: + """The user's LC_TIME locale and its date order, read after + ``setlocale(LC_TIME, "")`` and restored afterwards. + + ``C``/``POSIX`` is "not configured", so both come back unknown. + """ + + with _LOCALE_LOCK: + saved = locale.setlocale(locale.LC_TIME) + try: + try: + name = locale.setlocale(locale.LC_TIME, "") + except locale.Error: + return None, None + if not name or name in ("C", "POSIX") or name.startswith("C."): + return None, None + try: + d_fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] + except (AttributeError, ValueError): + d_fmt = "" + finally: + locale.setlocale(locale.LC_TIME, saved) + return name, date_order_from_format(d_fmt) + + +def _posix_date_order_for(name: str) -> Optional[DateOrder]: + with _LOCALE_LOCK: + saved = locale.setlocale(locale.LC_TIME) + try: + for candidate in (f"{name}.UTF-8", name): + try: + locale.setlocale(locale.LC_TIME, candidate) + except locale.Error: + continue + try: + fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] + except (AttributeError, ValueError): + return None + return date_order_from_format(fmt) + return None + finally: + locale.setlocale(locale.LC_TIME, saved) + + +# The current keyboard layout and input source from Text Input Sources, the +# API the focused application's key events go through. It runs in a child +# interpreter: HIToolbox may assert (and kill the process) when called off +# the main thread, and the replay gate can run on a worker thread. +_TIS_SCRIPT = """ +import ctypes +cf = ctypes.CDLL("/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation") +hi = ctypes.CDLL("/System/Library/Frameworks/Carbon.framework/Carbon") +hi.TISCopyCurrentKeyboardLayoutInputSource.restype = ctypes.c_void_p +hi.TISCopyCurrentKeyboardInputSource.restype = ctypes.c_void_p +hi.TISGetInputSourceProperty.restype = ctypes.c_void_p +hi.TISGetInputSourceProperty.argtypes = [ctypes.c_void_p, ctypes.c_void_p] +cf.CFStringGetCString.argtypes = [ + ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long, ctypes.c_uint32 +] +cf.CFStringGetCString.restype = ctypes.c_bool +cf.CFRelease.argtypes = [ctypes.c_void_p] +key = ctypes.c_void_p.in_dll(hi, "kTISPropertyInputSourceID") +def ident(src): + if not src: + return "" + value = hi.TISGetInputSourceProperty(src, key) + buf = ctypes.create_string_buffer(512) + if not value or not cf.CFStringGetCString(value, buf, 512, 0x08000100): + return "" + return buf.value.decode() +layout = hi.TISCopyCurrentKeyboardLayoutInputSource() +source = hi.TISCopyCurrentKeyboardInputSource() +try: + print(ident(layout)) + print(ident(source)) +finally: + for ref in (layout, source): + if ref: + cf.CFRelease(ref) +""" + + +def _macos_layout(run: _Run) -> Optional[str]: + """The layout the focused application receives keys through. + + Text Input Sources first: the current keyboard layout, plus the current + input method when one (an IME) sits on top of it. A frozen app can't run + a child interpreter, so it falls back to HIToolbox's saved current layout. + With per-document input sources turned on, both name the source of the + application that has focus at the time of the check. + """ + + lines: list[str] = [] + if not getattr(sys, "frozen", False): + out = run([sys.executable, "-I", "-c", _TIS_SCRIPT]) or "" + lines = [line.strip() for line in out.splitlines()] + layout_id = lines[0] if lines else "" + if layout_id: + source_id = lines[1] if len(lines) > 1 else "" + layout = layout_id.rsplit(".", 1)[-1] + if source_id and source_id != layout_id: + layout += f"+{source_id}" + return layout + saved = ( + run( + [ + "defaults", + "read", + "com.apple.HIToolbox", + "AppleCurrentKeyboardLayoutInputSourceID", + ] + ) + or "" + ).strip() + return saved.rsplit(".", 1)[-1] if saved else None + + +def _macos(run: _Run) -> tuple[Optional[str], Optional[DateOrder], Optional[str]]: + raw_locale = (run(["defaults", "read", "-g", "AppleLocale"]) or "").strip() + name = _locale_family(raw_locale) if raw_locale else None + order: Optional[DateOrder] = None + custom = run(["defaults", "read", "-g", "AppleICUDateFormatStrings"]) or "" + # A user-set short date format: {1 = "dd.MM.yy"; ...} + m = re.search(r'\b1\s*=\s*"([^"]+)"', custom) + if m: + order = date_order_from_format(m.group(1)) + if order is None and name: + lang, _, region = name.partition("_") + order = _posix_date_order_for(f"{lang}_{region.upper()}" if region else lang) + return name, order, _macos_layout(run) + + +def klid_from_hkl( + hkl: int, layout_id_to_klid: Callable[[int], Optional[str]] +) -> Optional[str]: + """The KLID (``00000407``, ``00010409``, ``e0010411``) behind a Windows + HKL, the per-thread handle ``GetKeyboardLayout`` returns. + + The high word is the layout: ``Fxxx`` is a ``Layout Id`` that the + registry maps to its KLID, ``Exxx`` is an IME, and anything else is the + layout's language id. None when the handle is empty or unmappable. + """ + + hkl &= 0xFFFFFFFF + lang = hkl & 0xFFFF + device = (hkl >> 16) & 0xFFFF + if not lang: + return None + if device & 0xF000 == 0xF000: + return layout_id_to_klid(device & 0x0FFF) + if device & 0xF000 == 0xE000: + return f"{device:04X}{lang:04X}" + return f"{(device or lang):08X}" + + +def _windows_klid_for_layout_id( + layout_id: int, +) -> Optional[str]: # pragma: no cover - Windows only + import winreg + + base = r"SYSTEM\CurrentControlSet\Control\Keyboard Layouts" + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, base) as root: # type: ignore[attr-defined] + index = 0 + while True: + try: + name = winreg.EnumKey(root, index) # type: ignore[attr-defined] + except OSError: + return None + index += 1 + try: + with winreg.OpenKey(root, name) as key: # type: ignore[attr-defined] + value, _ = winreg.QueryValueEx(key, "Layout Id") # type: ignore[attr-defined] + if int(str(value), 16) == layout_id: + return str(name) + except (OSError, ValueError): + continue + + +def _windows() -> tuple[ + Optional[str], Optional[DateOrder], Optional[str] +]: # pragma: no cover - exercised on Windows CI + import ctypes + + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + user32 = ctypes.windll.user32 # type: ignore[attr-defined] + name: Optional[str] = None + order: Optional[DateOrder] = None + layout: Optional[str] = None + buf = ctypes.create_unicode_buffer(85) + if kernel32.GetUserDefaultLocaleName(buf, 85): + name = buf.value or None + if name: + fmt = ctypes.create_unicode_buffer(80) + LOCALE_SSHORTDATE = 0x1F + if kernel32.GetLocaleInfoEx(name, LOCALE_SSHORTDATE, fmt, 80): + order = date_order_from_format(fmt.value) + # The layout is per thread. Read the foreground window's thread, the one + # that receives keystrokes, not this process's own thread. No foreground + # window (a service or locked session) leaves the layout unknown. + user32.GetForegroundWindow.restype = ctypes.c_void_p + user32.GetWindowThreadProcessId.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + user32.GetWindowThreadProcessId.restype = ctypes.c_ulong + user32.GetKeyboardLayout.argtypes = [ctypes.c_ulong] + user32.GetKeyboardLayout.restype = ctypes.c_void_p + hwnd = user32.GetForegroundWindow() + if hwnd: + thread_id = user32.GetWindowThreadProcessId(hwnd, None) + hkl = user32.GetKeyboardLayout(thread_id) if thread_id else None + if hkl: + layout = klid_from_hkl(int(hkl), _windows_klid_for_layout_id) + return name, order, layout + + +def _xkb_active_group(display: str) -> Optional[int]: + """The X server's active (effective) XKB group, via ``XkbGetState``.""" + + import ctypes + import ctypes.util + + path = ctypes.util.find_library("X11") + if not path: + return None + x11 = ctypes.CDLL(path) + x11.XOpenDisplay.argtypes = [ctypes.c_char_p] + x11.XOpenDisplay.restype = ctypes.c_void_p + x11.XkbGetState.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p] + x11.XkbGetState.restype = ctypes.c_int + x11.XCloseDisplay.argtypes = [ctypes.c_void_p] + dpy = x11.XOpenDisplay(display.encode()) + if not dpy: + return None + try: + # XkbStateRec starts with ``unsigned char group``; 64 bytes covers it. + state = (ctypes.c_ubyte * 64)() + XkbUseCoreKbd = 0x0100 + if x11.XkbGetState(dpy, XkbUseCoreKbd, ctypes.byref(state)) != 0: + return None + return int(state[0]) + finally: + x11.XCloseDisplay(dpy) + + +def _xkb_layout( + run: _Run, env: dict[str, str], group: _GroupReader = _xkb_active_group +) -> Optional[str]: + """The XKB layout of the X server's active group. + + The X server's keymap (``setxkbmap -query``) lists the configured + layouts; ``XkbGetState`` says which group is active. When the group + can't be read, only a single-layout keymap is unambiguous. Wayland has no + common API for the compositor's active layout, and ``XKB_DEFAULT_LAYOUT`` + and ``localectl`` describe configuration, not the active group, so those + cases are unknown. + """ + + if env.get("WAYLAND_DISPLAY") or env.get("XDG_SESSION_TYPE", "").lower() == ( + "wayland" + ): + return None + display = env.get("DISPLAY") + if not display: + return None + query = run(["setxkbmap", "-query"]) or "" + found = re.search(r"^layout:\s*(\S+)", query, re.M) + if found is None: + return None + layouts = found.group(1).split(",") + variant = re.search(r"^variant:\s*(\S*)", query, re.M) + variants = variant.group(1).split(",") if variant else [] + try: + active = group(display) + except Exception: # noqa: BLE001 - an unreadable group is unknown + active = None + if active is None: + if len(layouts) != 1: + return None + active = 0 + if active >= len(layouts) or not layouts[active]: + return None + chosen = variants[active] if active < len(variants) else "" + return layouts[active] + (f"+{chosen}" if chosen else "") + + +def _posix( + run: _Run, env: dict[str, str], group: _GroupReader +) -> tuple[Optional[str], Optional[DateOrder], Optional[str]]: + name, order = _posix_user_time_locale() + return ( + _locale_family(name) if name else None, + order, + _xkb_layout(run, env, group), + ) + + +def read_os_locale_facts( + *, + platform: Optional[str] = None, + run: Optional[_Run] = None, + env: Optional[dict[str, str]] = None, + xkb_group: Optional[_GroupReader] = None, +) -> dict[str, Optional[str]]: + """This computer's locale, date order and the keyboard layout of the + input target, as plain normalized strings. Never raises; a fact that + can't be read is ``None``.""" + + plat = platform or sys.platform + runner = run or _run + environ = dict(os.environ if env is None else env) + try: + if plat == "darwin": + name, order, layout = _macos(runner) + elif plat == "win32": + name, order, layout = _windows() + else: + name, order, layout = _posix( + runner, environ, xkb_group or _xkb_active_group + ) + except Exception: # noqa: BLE001 - an unreadable fact is unknown, not a guess + name, order, layout = None, None, None + return { + "locale": _norm(name) if name else None, + "date_order": order, + "keyboard_layout": _norm(layout) if layout else None, + } + + +def observe_locale_facts() -> dict[str, Optional[str]]: + """The facts to persist with a new recording: ``locale``, ``date_order`` + and ``keyboard_layout``, each a normalized string or ``None``.""" + + return read_os_locale_facts() diff --git a/packages/capture/openadapt_capture/recorder.py b/packages/capture/openadapt_capture/recorder.py index 131ef7174..b441bf92b 100644 --- a/packages/capture/openadapt_capture/recorder.py +++ b/packages/capture/openadapt_capture/recorder.py @@ -80,6 +80,7 @@ create_input_observer, ) from openadapt_capture.live import LiveEventTap, publish_live +from openadapt_capture.locale_facts import observe_locale_facts from openadapt_capture.structural import ( StructuralObservationRequest, StructuralObserver, @@ -2605,13 +2606,17 @@ def create_recording( "platform": sys.platform, "task_description": task_description, } - capture_config: dict[str, Any] = {} + # The keyboard layout, locale and date order the user records under + # (#1897). Replay types by keycode, so it refuses on another layout or + # date order. Read now, on the recording computer, and persisted even + # when every fact is unknown: an incomplete observation must refuse at + # replay, not look like a recording from before these facts existed. + capture_config: dict[str, Any] = {"capture_locale": observe_locale_facts()} if window_capture_info is not None: capture_config["capture_window"] = window_capture_info if desktop_capture_info is not None: capture_config["capture_desktop"] = desktop_capture_info - if capture_config: - recording_data["config"] = capture_config + recording_data["config"] = capture_config # Several writer processes share this file: give it a write log. engine, Session = create_db(db_path, journal_mode=SQLITE_CAPTURE_JOURNAL_MODE) session = Session() diff --git a/packages/capture/tests/test_locale_facts.py b/packages/capture/tests/test_locale_facts.py new file mode 100644 index 000000000..f0c61dbcd --- /dev/null +++ b/packages/capture/tests/test_locale_facts.py @@ -0,0 +1,76 @@ +"""Recording-time locale, date order and keyboard layout (#1897). + +Capture reads these when a recording starts and persists them, so a later +conversion carries the recording computer's facts instead of its own. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from openadapt_capture import locale_facts, recorder +from openadapt_capture.capture import CaptureSession + + +def _stub_display(monkeypatch) -> 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 + ) + monkeypatch.setattr(recorder.platform, "get_display_pixel_ratio", lambda: 2.0) + + +def test_recording_persists_the_facts_read_when_it_starts( + monkeypatch, tmp_path +) -> None: + _stub_display(monkeypatch) + facts = {"locale": "de_de", "date_order": "dmy", "keyboard_layout": "de"} + monkeypatch.setattr(recorder, "observe_locale_facts", lambda: dict(facts)) + + recording, _ = recorder.create_recording("t", capture_dir=str(tmp_path / "c")) + + assert recording.config["capture_locale"] == facts + with CaptureSession.load(str(tmp_path / "c")) as session: + assert session.locale_facts == facts + + +def test_an_unreadable_observation_is_persisted_not_dropped( + monkeypatch, tmp_path +) -> None: + # Every fact unknown is still a new, bound recording. Dropping the key + # would make it look like a recording from before #1897 and skip the check. + _stub_display(monkeypatch) + unknown = {"locale": None, "date_order": None, "keyboard_layout": None} + monkeypatch.setattr(recorder, "observe_locale_facts", lambda: dict(unknown)) + + recording, _ = recorder.create_recording("t", capture_dir=str(tmp_path / "c")) + + assert recording.config["capture_locale"] == unknown + with CaptureSession.load(str(tmp_path / "c")) as session: + assert session.locale_facts == unknown + + +def test_a_legacy_recording_has_no_locale_facts() -> None: + session = CaptureSession("/tmp/unused", None, SimpleNamespace(config=None)) + assert session.locale_facts is None + session = CaptureSession("/tmp/unused", None, SimpleNamespace(config={})) + assert session.locale_facts is None + + +def test_the_reader_never_raises_and_never_guesses() -> None: + def broken(cmd: list[str]): + raise RuntimeError("no OS") + + seen = locale_facts.read_os_locale_facts(platform="darwin", run=broken) + assert seen == {"locale": None, "date_order": None, "keyboard_layout": None} + + +def test_the_x_layout_is_the_active_group_not_the_configured_list() -> None: + def run(cmd: list[str]): + return "rules: evdev\nlayout: us,de\nvariant: ,nodeadkeys\n" + + env = {"DISPLAY": ":0"} + assert locale_facts._xkb_layout(run, env, lambda d: 0) == "us" + assert locale_facts._xkb_layout(run, env, lambda d: 1) == "de+nodeadkeys" + assert locale_facts._xkb_layout(run, env, lambda d: None) is None diff --git a/packages/flow/openadapt_flow/adapters/capture.py b/packages/flow/openadapt_flow/adapters/capture.py index 839e6d686..633e8df36 100644 --- a/packages/flow/openadapt_flow/adapters/capture.py +++ b/packages/flow/openadapt_flow/adapters/capture.py @@ -309,6 +309,41 @@ def _desktop_capture_meta(session: "CaptureSession") -> Optional[dict[str, Any]] return None +def _capture_locale_meta(session: "CaptureSession") -> Optional[dict[str, Any]]: + """The locale facts capture recorded with the session, else None (#1897). + + Reads ``CaptureSession.locale_facts`` and falls back to the recording's + ``config`` JSON (key ``capture_locale``), like the scope metadata above. + None means a session recorded before capture persisted these facts. A + present but malformed value refuses, so a damaged binding can't drop the + check. Unknown facts stay ``None``; they are never filled in from this + computer. + """ + + raw = getattr(session, "locale_facts", None) + if raw is None: + recording = getattr(session, "_recording", None) + config = getattr(recording, "config", None) + if isinstance(config, dict): + raw = config.get("capture_locale") + if raw is None: + return None + if not isinstance(raw, dict): + raise ValueError( + "capture session declares malformed locale facts; expected an " + "object or null" + ) + from openadapt_flow.locale_invariants import LocaleInvariants + + try: + facts = LocaleInvariants.model_validate(raw) + except ValueError as exc: + raise ValueError( + "capture session locale facts do not match the locale schema" + ) from exc + return facts.model_dump() + + def _is_viewport(value: Any) -> TypeGuard[Sequence[float]]: """True when ``value`` is an exact positive integer pixel pair.""" return ( @@ -1266,6 +1301,16 @@ def convert_capture( "source": "openadapt-capture", "task_description": session.task_description, } + # Native replay types by keycode, so the recording computer's locale, + # date order and keyboard layout travel with the recording and are + # checked before replay (#1897). They are the facts capture persisted + # when recording started, copied as they are, unknowns included. This + # computer's facts are never read here: conversion can run later or + # elsewhere. A session from before capture persisted them stays + # unbound. + recorded_locale = _capture_locale_meta(session) + if recorded_locale is not None: + meta["locale_invariants"] = recorded_locale # Name the space every recorded x/y is in, so a consumer fails closed # instead of assuming. Window and desktop scope are mutually exclusive # (a session carrying both is refused above). A session that declares diff --git a/packages/flow/openadapt_flow/compiler/compile.py b/packages/flow/openadapt_flow/compiler/compile.py index e4ce15e8f..7a73ad7e8 100644 --- a/packages/flow/openadapt_flow/compiler/compile.py +++ b/packages/flow/openadapt_flow/compiler/compile.py @@ -2086,6 +2086,28 @@ def _apply_param_overrides( return secret_values +def _locale_invariants_from_meta(meta: dict[str, Any]) -> Any: + """The recording's locale, date order and keyboard layout (#1897). + + Absent stays absent (an older recording is not bound). Present but + malformed refuses, so a damaged binding can't silently drop the check. + """ + + raw = meta.get("locale_invariants") + if raw is None: + return None + from openadapt_flow.locale_invariants import LocaleInvariants + + if not isinstance(raw, dict): + raise ValueError("meta.json locale_invariants must be an object") + try: + return LocaleInvariants.model_validate(raw) + except Exception as exc: + raise ValueError( + "meta.json locale_invariants does not match the locale schema" + ) from exc + + def compile_recording( recording_dir: Path | str, out_bundle_dir: Path | str, @@ -2327,6 +2349,7 @@ def _compile_recording( surface, execution_mode = _surface_binding_from_meta( meta, backend_hints, target_surface ) + locale_invariants = _locale_invariants_from_meta(meta) events = _load_events(recording) params: dict[str, str] = dict(meta.get("params") or {}) secret_params: list[str] = list(meta.get("secret_params") or []) @@ -3615,6 +3638,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: recording_id=meta.get("id"), viewport=tuple(viewport) if viewport else None, backend_hints=backend_hints, + locale_invariants=locale_invariants, surface=surface, execution_mode=execution_mode, params=params, diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index dd946abc4..f8b2ac01a 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -55,6 +55,7 @@ # Type-only import for the Step.effects forward reference. The RUNTIME # import is at the BOTTOM of this module (see the note there) to avoid a # circular import through openadapt_flow.runtime's package init. + from openadapt_flow.locale_invariants import LocaleInvariants from openadapt_flow.qualification import QualificationProject from openadapt_flow.runtime.effects.effect import Effect @@ -3295,6 +3296,9 @@ class Workflow(BaseModel): # or a hosted run summary. Empty for browser/native bundles, preserving # their serialized form through the compatibility serializer below. backend_hints: Optional[BackendHints] = None + #: Locale, date order and keyboard layout recorded with the demonstration + #: (#1897). Checked before replay; omitted from serialized bytes when unset. + locale_invariants: Optional["LocaleInvariants"] = None # -- surface binding (roadmap Section 5) -------------------------------- # The exact execution surface this workflow was recorded/qualified on and # the execution mode that surface implies. A bound workflow refuses to run @@ -3414,6 +3418,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) if self.backend_hints is None: data.pop("backend_hints", None) + if self.locale_invariants is None: + data.pop("locale_invariants", None) if self.surface is None: data.pop("surface", None) if self.execution_mode is None: @@ -6400,6 +6406,7 @@ def save( # OCR/cv2/model deps) runtime package loads cleanly and Step's schema can be # completed. Effect enters this module's globals so ``model_rebuild`` resolves # the forward reference; bundles with no effects are unaffected. +from openadapt_flow.locale_invariants import LocaleInvariants # noqa: E402,F401 from openadapt_flow.qualification import QualificationProject # noqa: E402,F401 from openadapt_flow.runtime.effects.effect import Effect # noqa: E402,F401 diff --git a/packages/flow/openadapt_flow/locale_invariants.py b/packages/flow/openadapt_flow/locale_invariants.py new file mode 100644 index 000000000..c3c11f335 --- /dev/null +++ b/packages/flow/openadapt_flow/locale_invariants.py @@ -0,0 +1,723 @@ +"""Locale, date order and keyboard layout as checked replay invariants (#1897). + +A worklist date like ``03/04/2026`` is ambiguous, and keycode replay on a +non-matching keyboard layout types the wrong punctuation. A recording carries +:class:`LocaleInvariants`; before replay the runtime compares them to this +computer and refuses on a mismatch. + +The recording's facts are read once by the capture producer when recording +starts (``openadapt_capture.locale_facts``), persisted in the recording, and +carried unchanged through conversion and compile. A converter never puts its +own computer's facts on a recording. + +Facts come from the OS, not from Python's start-up ``C`` locale: + +* macOS: the global ``AppleLocale`` and ``AppleICUDateFormatStrings``, and the + current keyboard layout and input method from Text Input Sources (in a + child interpreter; HIToolbox's saved current layout when that can't run). +* Windows: ``GetUserDefaultLocaleName``, ``GetLocaleInfoEx`` (short date), and + ``GetKeyboardLayout`` for the foreground window's thread, mapped to a KLID. +* Linux and other POSIX: the user locale via ``setlocale(LC_TIME, "")`` and + ``nl_langinfo(D_FMT)``, and the layout of the X server's active XKB group + (``XkbGetState`` with ``setxkbmap -query``). + +Limits: the layout is that of whatever has keyboard focus when it is read. +At recording time that is the recorded application. At replay the runtime +checks once before any action and again after focus, immediately before +keyboard delivery, so a layout change caused by the focusing click is +caught. Wayland has no common API for the active layout, and +``XKB_DEFAULT_LAYOUT``/``localectl`` name configuration, not the active +group, so on Wayland, or with several X layouts and no readable group, the +layout is unknown. + +A fact that can't be read is ``None`` (unknown). Unknown is never filled in +with ``en_US``, ``mdy`` or ``us``. A workflow with no binding at all is a +recording made before #1897 and is not checked. A bound workflow refuses +replay when any fact it needs is unknown, on either side. No GUI is opened. +""" + +from __future__ import annotations + +import locale +import os +import re +import subprocess +import sys +import threading +from dataclasses import dataclass +from typing import Any, Callable, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +DateOrder = Literal["mdy", "dmy", "ymd"] + +__all__ = [ + "DateOrder", + "LocaleInvariantMismatch", + "LocaleInvariantResult", + "LocaleInvariants", + "check_locale_invariants", + "date_order_from_format", + "enforce_workflow_locale_invariants", + "klid_from_hkl", + "observe_locale_invariants", + "read_os_locale_facts", + "workflow_uses_keystroke_typing", + "worklist_date_needs_explicit_format", + "worklist_number_needs_explicit_format", +] + + +class LocaleInvariants(BaseModel): + """Environment facts recorded with a demonstration and checked at replay. + + ``None`` means the fact could not be read on that computer. + """ + + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + locale: Optional[str] = Field( + default=None, + min_length=2, + max_length=64, + description="BCP-47 / POSIX locale id, e.g. en_US, fr_CA, de_DE.", + ) + date_order: Optional[DateOrder] = Field( + default=None, + description="How the OS orders numeric dates: mdy, dmy, or ymd.", + ) + keyboard_layout: Optional[str] = Field( + default=None, + min_length=1, + max_length=128, + description=( + "Keyboard layout id as the OS reports it (macOS input source, " + "Windows KLID, or XKB layout[+variant])." + ), + ) + + @field_validator("locale", "keyboard_layout") + @classmethod + def _normalize_token(cls, value: Optional[str]) -> Optional[str]: + if value is None: + return None + text = value.strip().replace("-", "_") + if not text: + raise ValueError("must not be empty") + return text.lower() + + def unknown(self) -> list[str]: + return [ + name + for name in ("locale", "date_order", "keyboard_layout") + if getattr(self, name) is None + ] + + +class LocaleInvariantMismatch(RuntimeError): + """This computer's locale, date order or layout disagrees with the recording, + or can't be read.""" + + def __init__(self, reason: str, *, mismatched: list[str]): + super().__init__(reason) + self.reason = reason + self.mismatched = list(mismatched) + + +@dataclass(frozen=True) +class LocaleInvariantResult: + ok: bool + mismatched: tuple[str, ...] = () + unknown: tuple[str, ...] = () + unrecorded: tuple[str, ...] = () + reason: str = "" + + def raise_if_refused(self) -> None: + if not self.ok: + raise LocaleInvariantMismatch( + self.reason, + mismatched=[*self.mismatched, *self.unknown, *self.unrecorded], + ) + + +def _norm(value: str) -> str: + return value.strip().replace("-", "_").lower() + + +def _locale_family(value: str) -> str: + """``en_us.utf-8`` / ``de_de@rg=chzzzz`` -> ``en_us`` / ``de_de``.""" + + return re.split(r"[.@]", _norm(value), maxsplit=1)[0] + + +def check_locale_invariants( + recorded: LocaleInvariants, + current: LocaleInvariants, + *, + keystroke_steps: bool = True, +) -> LocaleInvariantResult: + """Compare recorded invariants to this computer. Fails closed. + + Each fact that matters must have been read when recording, be readable + here, and be equal. The keyboard layout only matters when a step types by + keystroke (``keystroke_steps``); locale and date order always matter, + because worklist parsing and masked date fields depend on them. A fact + the recording could not read refuses: there is nothing to check it + against, and an incomplete observation is not an older, unbound one. + """ + + names = ["locale", "date_order"] + (["keyboard_layout"] if keystroke_steps else []) + mismatched: list[str] = [] + unknown: list[str] = [] + unrecorded: list[str] = [] + for name in names: + want = getattr(recorded, name) + if want is None: + unrecorded.append(name) + continue + have = getattr(current, name) + if have is None: + unknown.append(name) + continue + if name == "locale": + same = _locale_family(want) == _locale_family(have) + else: + same = _norm(str(want)) == _norm(str(have)) + if not same: + mismatched.append(name) + if mismatched or unknown or unrecorded: + parts = [] + if mismatched: + parts.append("does not match the recording: " + ", ".join(mismatched)) + if unknown: + parts.append("could not be read on this computer: " + ", ".join(unknown)) + if unrecorded: + parts.append("could not be read when recording: " + ", ".join(unrecorded)) + return LocaleInvariantResult( + ok=False, + mismatched=tuple(mismatched), + unknown=tuple(unknown), + unrecorded=tuple(unrecorded), + reason="replay refused: " + + "; ".join(parts) + + " (refuse rather than type the wrong punctuation or date)", + ) + return LocaleInvariantResult(ok=True, reason="locale invariants match") + + +# -- date format parsing ---------------------------------------------------------- + +_STRFTIME = re.compile(r"%[-_0^#]?([a-zA-Z])") +_STRFTIME_KIND = { + "d": "d", + "e": "d", + "m": "m", + "b": "m", + "B": "m", + "h": "m", + "y": "y", + "Y": "y", + "C": "y", +} + + +def date_order_from_format(fmt: str) -> Optional[DateOrder]: + """The day/month/year order of a strftime (``%d.%m.%Y``, ``%e/%m/%y``, + ``%D``) or ICU/Windows (``M/d/yyyy``, ``dd.MM.yyyy``) date pattern. + + None when the pattern doesn't name all three parts. + """ + + text = (fmt or "").strip() + if not text: + return None + seen: list[str] = [] + if "%" in text: + text = text.replace("%D", "%m/%d/%y").replace("%F", "%Y-%m-%d") + for m in _STRFTIME.finditer(text): + kind = _STRFTIME_KIND.get(m.group(1)) + if kind and kind not in seen: + seen.append(kind) + else: + # Drop quoted literals, then read runs of pattern letters. A run of + # three or more d/E is a weekday name, not the day of the month. + text = re.sub(r"'[^']*'", "", text) + for m in re.finditer(r"([dMLyuE])\1*", text): + letter, size = m.group(1), len(m.group(0)) + if letter == "E" or (letter == "d" and size >= 3): + continue + kind = {"d": "d", "M": "m", "L": "m", "y": "y", "u": "y"}[letter] + if kind not in seen: + seen.append(kind) + order = "".join(seen) + if order in ("mdy", "dmy", "ymd"): + return order # type: ignore[return-value] + return None + + +# -- OS readers (no GUI) ------------------------------------------------------------ +# +# Kept in step with openadapt_capture.locale_facts (the recording-time +# producer); tests/test_locale_invariants.py checks the two agree. + +_Run = Callable[[list[str]], Optional[str]] +_GroupReader = Callable[[str], Optional[int]] +_LOCALE_LOCK = threading.Lock() + + +def _run(cmd: list[str]) -> Optional[str]: + try: + out = subprocess.run( + cmd, capture_output=True, text=True, timeout=5, check=False + ) + except (OSError, subprocess.SubprocessError): + return None + if out.returncode != 0: + return None + return out.stdout + + +def _posix_user_time_locale() -> tuple[Optional[str], Optional[DateOrder]]: + """The user's LC_TIME locale and its date order, read after + ``setlocale(LC_TIME, "")`` and restored afterwards. + + ``C``/``POSIX`` is "not configured", so both come back unknown. + """ + + with _LOCALE_LOCK: + saved = locale.setlocale(locale.LC_TIME) + try: + try: + name = locale.setlocale(locale.LC_TIME, "") + except locale.Error: + return None, None + if not name or name in ("C", "POSIX") or name.startswith("C."): + return None, None + try: + d_fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] + except (AttributeError, ValueError): + d_fmt = "" + finally: + locale.setlocale(locale.LC_TIME, saved) + return name, date_order_from_format(d_fmt) + + +def _posix_date_order_for(name: str) -> Optional[DateOrder]: + with _LOCALE_LOCK: + saved = locale.setlocale(locale.LC_TIME) + try: + for candidate in (f"{name}.UTF-8", name): + try: + locale.setlocale(locale.LC_TIME, candidate) + except locale.Error: + continue + try: + fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] + except (AttributeError, ValueError): + return None + return date_order_from_format(fmt) + return None + finally: + locale.setlocale(locale.LC_TIME, saved) + + +# The current keyboard layout and input source from Text Input Sources, the +# API the focused application's key events go through. It runs in a child +# interpreter: HIToolbox may assert (and kill the process) when called off +# the main thread, and the replay gate can run on a worker thread. +_TIS_SCRIPT = """ +import ctypes +cf = ctypes.CDLL("/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation") +hi = ctypes.CDLL("/System/Library/Frameworks/Carbon.framework/Carbon") +hi.TISCopyCurrentKeyboardLayoutInputSource.restype = ctypes.c_void_p +hi.TISCopyCurrentKeyboardInputSource.restype = ctypes.c_void_p +hi.TISGetInputSourceProperty.restype = ctypes.c_void_p +hi.TISGetInputSourceProperty.argtypes = [ctypes.c_void_p, ctypes.c_void_p] +cf.CFStringGetCString.argtypes = [ + ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long, ctypes.c_uint32 +] +cf.CFStringGetCString.restype = ctypes.c_bool +cf.CFRelease.argtypes = [ctypes.c_void_p] +key = ctypes.c_void_p.in_dll(hi, "kTISPropertyInputSourceID") +def ident(src): + if not src: + return "" + value = hi.TISGetInputSourceProperty(src, key) + buf = ctypes.create_string_buffer(512) + if not value or not cf.CFStringGetCString(value, buf, 512, 0x08000100): + return "" + return buf.value.decode() +layout = hi.TISCopyCurrentKeyboardLayoutInputSource() +source = hi.TISCopyCurrentKeyboardInputSource() +try: + print(ident(layout)) + print(ident(source)) +finally: + for ref in (layout, source): + if ref: + cf.CFRelease(ref) +""" + + +def _macos_layout(run: _Run) -> Optional[str]: + """The layout the focused application receives keys through. + + Text Input Sources first: the current keyboard layout, plus the current + input method when one (an IME) sits on top of it. A frozen app can't run + a child interpreter, so it falls back to HIToolbox's saved current layout. + With per-document input sources turned on, both name the source of the + application that has focus at the time of the check. + """ + + lines: list[str] = [] + if not getattr(sys, "frozen", False): + out = run([sys.executable, "-I", "-c", _TIS_SCRIPT]) or "" + lines = [line.strip() for line in out.splitlines()] + layout_id = lines[0] if lines else "" + if layout_id: + source_id = lines[1] if len(lines) > 1 else "" + layout = layout_id.rsplit(".", 1)[-1] + if source_id and source_id != layout_id: + layout += f"+{source_id}" + return layout + saved = ( + run( + [ + "defaults", + "read", + "com.apple.HIToolbox", + "AppleCurrentKeyboardLayoutInputSourceID", + ] + ) + or "" + ).strip() + return saved.rsplit(".", 1)[-1] if saved else None + + +def _macos(run: _Run) -> tuple[Optional[str], Optional[DateOrder], Optional[str]]: + raw_locale = (run(["defaults", "read", "-g", "AppleLocale"]) or "").strip() + name = _locale_family(raw_locale) if raw_locale else None + order: Optional[DateOrder] = None + custom = run(["defaults", "read", "-g", "AppleICUDateFormatStrings"]) or "" + # A user-set short date format: {1 = "dd.MM.yy"; ...} + m = re.search(r'\b1\s*=\s*"([^"]+)"', custom) + if m: + order = date_order_from_format(m.group(1)) + if order is None and name: + lang, _, region = name.partition("_") + order = _posix_date_order_for(f"{lang}_{region.upper()}" if region else lang) + return name, order, _macos_layout(run) + + +def klid_from_hkl( + hkl: int, layout_id_to_klid: Callable[[int], Optional[str]] +) -> Optional[str]: + """The KLID (``00000407``, ``00010409``, ``e0010411``) behind a Windows + HKL, the per-thread handle ``GetKeyboardLayout`` returns. + + The high word is the layout: ``Fxxx`` is a ``Layout Id`` that the + registry maps to its KLID, ``Exxx`` is an IME, and anything else is the + layout's language id. None when the handle is empty or unmappable. + """ + + hkl &= 0xFFFFFFFF + lang = hkl & 0xFFFF + device = (hkl >> 16) & 0xFFFF + if not lang: + return None + if device & 0xF000 == 0xF000: + return layout_id_to_klid(device & 0x0FFF) + if device & 0xF000 == 0xE000: + return f"{device:04X}{lang:04X}" + return f"{(device or lang):08X}" + + +def _windows_klid_for_layout_id( + layout_id: int, +) -> Optional[str]: # pragma: no cover - Windows only + import winreg + + base = r"SYSTEM\CurrentControlSet\Control\Keyboard Layouts" + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, base) as root: # type: ignore[attr-defined] + index = 0 + while True: + try: + name = winreg.EnumKey(root, index) # type: ignore[attr-defined] + except OSError: + return None + index += 1 + try: + with winreg.OpenKey(root, name) as key: # type: ignore[attr-defined] + value, _ = winreg.QueryValueEx(key, "Layout Id") # type: ignore[attr-defined] + if int(str(value), 16) == layout_id: + return str(name) + except (OSError, ValueError): + continue + + +def _windows() -> tuple[ + Optional[str], Optional[DateOrder], Optional[str] +]: # pragma: no cover - exercised on Windows CI + import ctypes + + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + user32 = ctypes.windll.user32 # type: ignore[attr-defined] + name: Optional[str] = None + order: Optional[DateOrder] = None + layout: Optional[str] = None + buf = ctypes.create_unicode_buffer(85) + if kernel32.GetUserDefaultLocaleName(buf, 85): + name = buf.value or None + if name: + fmt = ctypes.create_unicode_buffer(80) + LOCALE_SSHORTDATE = 0x1F + if kernel32.GetLocaleInfoEx(name, LOCALE_SSHORTDATE, fmt, 80): + order = date_order_from_format(fmt.value) + # The layout is per thread. Read the foreground window's thread, the one + # that receives keystrokes, not this process's own thread. No foreground + # window (a service or locked session) leaves the layout unknown. + user32.GetForegroundWindow.restype = ctypes.c_void_p + user32.GetWindowThreadProcessId.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + user32.GetWindowThreadProcessId.restype = ctypes.c_ulong + user32.GetKeyboardLayout.argtypes = [ctypes.c_ulong] + user32.GetKeyboardLayout.restype = ctypes.c_void_p + hwnd = user32.GetForegroundWindow() + if hwnd: + thread_id = user32.GetWindowThreadProcessId(hwnd, None) + hkl = user32.GetKeyboardLayout(thread_id) if thread_id else None + if hkl: + layout = klid_from_hkl(int(hkl), _windows_klid_for_layout_id) + return name, order, layout + + +def _xkb_active_group(display: str) -> Optional[int]: + """The X server's active (effective) XKB group, via ``XkbGetState``.""" + + import ctypes + import ctypes.util + + path = ctypes.util.find_library("X11") + if not path: + return None + x11 = ctypes.CDLL(path) + x11.XOpenDisplay.argtypes = [ctypes.c_char_p] + x11.XOpenDisplay.restype = ctypes.c_void_p + x11.XkbGetState.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p] + x11.XkbGetState.restype = ctypes.c_int + x11.XCloseDisplay.argtypes = [ctypes.c_void_p] + dpy = x11.XOpenDisplay(display.encode()) + if not dpy: + return None + try: + # XkbStateRec starts with ``unsigned char group``; 64 bytes covers it. + state = (ctypes.c_ubyte * 64)() + XkbUseCoreKbd = 0x0100 + if x11.XkbGetState(dpy, XkbUseCoreKbd, ctypes.byref(state)) != 0: + return None + return int(state[0]) + finally: + x11.XCloseDisplay(dpy) + + +def _xkb_layout( + run: _Run, env: dict[str, str], group: _GroupReader = _xkb_active_group +) -> Optional[str]: + """The XKB layout of the X server's active group. + + The X server's keymap (``setxkbmap -query``) lists the configured + layouts; ``XkbGetState`` says which group is active. When the group + can't be read, only a single-layout keymap is unambiguous. Wayland has no + common API for the compositor's active layout, and ``XKB_DEFAULT_LAYOUT`` + and ``localectl`` describe configuration, not the active group, so those + cases are unknown. + """ + + if env.get("WAYLAND_DISPLAY") or env.get("XDG_SESSION_TYPE", "").lower() == ( + "wayland" + ): + return None + display = env.get("DISPLAY") + if not display: + return None + query = run(["setxkbmap", "-query"]) or "" + found = re.search(r"^layout:\s*(\S+)", query, re.M) + if found is None: + return None + layouts = found.group(1).split(",") + variant = re.search(r"^variant:\s*(\S*)", query, re.M) + variants = variant.group(1).split(",") if variant else [] + try: + active = group(display) + except Exception: # noqa: BLE001 - an unreadable group is unknown + active = None + if active is None: + if len(layouts) != 1: + return None + active = 0 + if active >= len(layouts) or not layouts[active]: + return None + chosen = variants[active] if active < len(variants) else "" + return layouts[active] + (f"+{chosen}" if chosen else "") + + +def _posix( + run: _Run, env: dict[str, str], group: _GroupReader +) -> tuple[Optional[str], Optional[DateOrder], Optional[str]]: + name, order = _posix_user_time_locale() + return ( + _locale_family(name) if name else None, + order, + _xkb_layout(run, env, group), + ) + + +def read_os_locale_facts( + *, + platform: Optional[str] = None, + run: Optional[_Run] = None, + env: Optional[dict[str, str]] = None, + xkb_group: Optional[_GroupReader] = None, +) -> dict[str, Optional[str]]: + """This computer's locale, date order and the keyboard layout of the + input target, as plain normalized strings. Never raises; a fact that + can't be read is ``None``.""" + + plat = platform or sys.platform + runner = run or _run + environ = dict(os.environ if env is None else env) + try: + if plat == "darwin": + name, order, layout = _macos(runner) + elif plat == "win32": + name, order, layout = _windows() + else: + name, order, layout = _posix( + runner, environ, xkb_group or _xkb_active_group + ) + except Exception: # noqa: BLE001 - an unreadable fact is unknown, not a guess + name, order, layout = None, None, None + return { + "locale": _norm(name) if name else None, + "date_order": order, + "keyboard_layout": _norm(layout) if layout else None, + } + + +def observe_locale_invariants( + *, + date_order: Optional[DateOrder] = None, + keyboard_layout: Optional[str] = None, + locale_name: Optional[str] = None, + platform: Optional[str] = None, + run: Optional[_Run] = None, + env: Optional[dict[str, str]] = None, + xkb_group: Optional[_GroupReader] = None, +) -> LocaleInvariants: + """Read this computer's locale, date order and the input target's + keyboard layout. + + Explicit arguments override what the OS reports. A fact the OS doesn't + report stays ``None``. + """ + + seen = read_os_locale_facts( + platform=platform, run=run, env=env, xkb_group=xkb_group + ) + try: + observed = LocaleInvariants.model_validate(seen) + except ValueError: + observed = LocaleInvariants() + return LocaleInvariants( + locale=locale_name if locale_name is not None else observed.locale, + date_order=date_order if date_order is not None else observed.date_order, + keyboard_layout=( + keyboard_layout if keyboard_layout is not None else observed.keyboard_layout + ), + ) + + +# -- worklist values ------------------------------------------------------------- + + +def worklist_date_needs_explicit_format(raw: str) -> bool: + """True unless a worklist date can be read without a declared order. + + Fails closed: ISO ``YYYY-MM-DD`` and dates with a month name are fine, and + so is a numeric date with a four-digit year whose order is forced (a part + above 12, or day equal to month). Anything else that looks numeric, such + as ``03/04/2026``, ``03/04/26``, ``03.04.2026`` or ``3-4-26``, needs a + declared format. + """ + + value = raw.strip() + if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): + return False + if re.search(r"[A-Za-z]", value): + return False + m = re.fullmatch(r"(\d{1,2})([/.\-])(\d{1,2})\2(\d{4})", value) + if m: + a, b = int(m.group(1)), int(m.group(3)) + return not (a > 12 or b > 12 or a == b) + return bool(re.search(r"\d", value)) + + +def worklist_number_needs_explicit_format(raw: str) -> bool: + """True when a number's decimal separator can't be told from grouping. + + ``1.234,56`` and ``1,234.56`` use both separators and are unambiguous. A + single separator followed by exactly three digits (``1.234``, ``1,234``) + reads as a thousand or as one-point-something, so it needs a declared + format. + """ + + value = raw.strip().lstrip("+-") + if "," in value and "." in value: + return False + return bool(re.fullmatch(r"\d{1,3}[.,]\d{3}", value)) + + +# -- workflow gate ---------------------------------------------------------------- + + +def workflow_uses_keystroke_typing(workflow: Any) -> bool: + """True when any step may type by keycode. + + There is no recorded semantic-input mode yet, so every TYPE, KEY and + HOTKEY step counts as keystroke input. That is the conservative reading. + """ + + from openadapt_flow.ir import ActionKind + + for step in getattr(workflow, "steps", None) or (): + kind = getattr(step, "action", None) or getattr(step, "kind", None) + value = getattr(kind, "value", kind) + if value in ( + ActionKind.TYPE.value, + ActionKind.KEY.value, + ActionKind.HOTKEY.value, + ): + return True + return False + + +def enforce_workflow_locale_invariants( + workflow: Any, + *, + current: Optional[LocaleInvariants] = None, +) -> None: + """Raise :class:`LocaleInvariantMismatch` when a workflow's recorded + invariants disagree with this computer or can't be checked. + + ``None`` means a recording from before #1897, which carries no binding. + A binding whose facts are unknown (all of them, even) is a new recording + whose observation was incomplete, and it refuses. + """ + + recorded = getattr(workflow, "locale_invariants", None) + if recorded is None: + return + current = current or observe_locale_invariants() + keystroke = workflow_uses_keystroke_typing(workflow) + check_locale_invariants( + recorded, current, keystroke_steps=keystroke + ).raise_if_refused() diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 0e9e10f3d..4977090ff 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -1894,6 +1894,17 @@ def _run_unsealed_steps( campaign_refusal = self._qualification_campaign_refusal(workflow) if campaign_refusal is not None: raise ValueError(campaign_refusal) + # Locale / date-order / keyboard layout (#1897): refuse before any + # keystroke when the recording's environment does not match this one. + from openadapt_flow.locale_invariants import ( + LocaleInvariantMismatch, + enforce_workflow_locale_invariants, + ) + + try: + enforce_workflow_locale_invariants(workflow) + except LocaleInvariantMismatch as exc: + raise ValueError(exc.reason) from exc if bundle_dir is not None: # Every execution path runs through here, so a JIT draft with no # valid admission, or an admitted one run on another app or @@ -14270,6 +14281,13 @@ def _act_without_dialog( baseline_field_value = self._text_value_at(field_point) if baseline_field_value is None: baseline_field_value = self._text_value_at(None) + # Focus (this step's click or the previous click-to-focus) may have + # changed the active keyboard layout. Recheck before any keystroke. + locale_refusal = self._require_locale_invariants_before_keyboard(workflow) + if locale_refusal is not None: + result.safety_halt = True + result.failure_category = "safety_halt" + return locale_refusal if step.selection_commit_key is not None: selection_commit_key = step.selection_commit_key assert step.selection_region is not None @@ -14406,6 +14424,11 @@ def _act_without_dialog( if step.action is ActionKind.HOTKEY else step.key ) + locale_refusal = self._require_locale_invariants_before_keyboard(workflow) + if locale_refusal is not None: + result.safety_halt = True + result.failure_category = "safety_halt" + return locale_refusal requires_atomic_keyboard = self._requires_atomic_identity_keyboard( step, workflow ) @@ -14879,6 +14902,31 @@ def environment_call() -> QualificationEnvironmentObservation: set_guard(self._require_qualification_environment_current) return None, observed, ids + def _require_locale_invariants_before_keyboard( + self, workflow: Workflow + ) -> Optional[str]: + """Recheck locale/layout after focus, before keystroke delivery (#1897). + + The run-start gate is not enough: a focusing click can change the + active input layout before TYPE/KEY/HOTKEY. Re-observe the target + facts now and refuse when they disagree with the recording or when + a needed fact cannot be read. A workflow with no binding (pre-#1897) + is unchecked. + """ + + if getattr(workflow, "locale_invariants", None) is None: + return None + from openadapt_flow.locale_invariants import ( + LocaleInvariantMismatch, + enforce_workflow_locale_invariants, + ) + + try: + enforce_workflow_locale_invariants(workflow) + except LocaleInvariantMismatch as exc: + return exc.reason + return None + def _require_qualification_environment_current(self) -> None: """Recheck the bound environment immediately before an input edge.""" @@ -18001,6 +18049,12 @@ def _verify_typed_input( if refusal is not None: result.input_verified = False return refusal + locale_refusal = self._require_locale_invariants_before_keyboard(workflow) + if locale_refusal is not None: + result.input_verified = False + result.safety_halt = True + result.failure_category = "safety_halt" + return locale_refusal self._require_qualification_environment_current() self._deliver_backend_call(result, lambda: self.backend.type_text(text)) landed, _changed = self._typed_input_landed( diff --git a/packages/flow/tests/test_capture_adapter.py b/packages/flow/tests/test_capture_adapter.py index c2d16fd97..14eded9b6 100644 --- a/packages/flow/tests/test_capture_adapter.py +++ b/packages/flow/tests/test_capture_adapter.py @@ -567,8 +567,10 @@ def test_meta_matches_recorder_contract(converted: Path) -> None: assert meta["source"] == "openadapt-capture" assert meta["task_description"] == "add a note" # Regression: a NON-window session's meta carries exactly the recorder - # contract keys — no window-mode fields may leak into it. - assert set(meta.keys()) == { + # contract keys — no window-mode fields may leak into it. This fixture is + # a session from before capture persisted locale facts (#1897), so it + # stays unbound. The converting computer's own facts are never added. + expected = { "id", "created_at", "viewport", @@ -577,6 +579,86 @@ def test_meta_matches_recorder_contract(converted: Path) -> None: "source", "task_description", } + assert set(meta.keys()) == expected + + +# -- locale facts (#1897): recorded by capture, carried, never re-observed ---- + + +def _forbid_observing_this_computer(monkeypatch) -> None: + import openadapt_flow.locale_invariants as li + + def refuse(*args, **kwargs): + raise AssertionError("conversion read the converting computer's locale") + + monkeypatch.setattr(li, "observe_locale_invariants", refuse) + monkeypatch.setattr(li, "read_os_locale_facts", refuse) + + +def test_conversion_carries_the_recorded_locale_facts( + tmp_path: Path, monkeypatch +) -> None: + import json + + _forbid_observing_this_computer(monkeypatch) + recorded = {"locale": "de_de", "date_order": "dmy", "keyboard_layout": "de"} + capture_dir = make_capture( + tmp_path, + demo_rows(), + config={"pixel_ratio": PIXEL_RATIO, "capture_locale": recorded}, + ) + convert_capture(capture_dir, tmp_path / "recording", settle_s=1.0) + meta = json.loads((tmp_path / "recording" / "meta.json").read_text()) + assert meta["locale_invariants"] == recorded + + +def test_an_unreadable_recording_observation_stays_bound_and_refuses( + tmp_path: Path, monkeypatch +) -> None: + import json + + from openadapt_flow.compiler.compile import _locale_invariants_from_meta + from openadapt_flow.ir import ActionKind, Step, Workflow + from openadapt_flow.locale_invariants import ( + LocaleInvariantMismatch, + LocaleInvariants, + enforce_workflow_locale_invariants, + ) + + _forbid_observing_this_computer(monkeypatch) + unknown = {"locale": None, "date_order": None, "keyboard_layout": None} + capture_dir = make_capture( + tmp_path, + demo_rows(), + config={"pixel_ratio": PIXEL_RATIO, "capture_locale": unknown}, + ) + convert_capture(capture_dir, tmp_path / "recording", settle_s=1.0) + meta = json.loads((tmp_path / "recording" / "meta.json").read_text()) + # Kept, not dropped: dropping it made a new recording look unbound. + assert meta["locale_invariants"] == unknown + bound = _locale_invariants_from_meta(meta) + assert bound is not None + workflow = Workflow( + name="demo", + locale_invariants=bound, + steps=[Step(id="s1", action=ActionKind.TYPE, intent="type a note")], + ) + here = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + with pytest.raises(LocaleInvariantMismatch, match="when recording"): + enforce_workflow_locale_invariants(workflow, current=here) + + +def test_malformed_recorded_locale_facts_refuse_conversion(tmp_path: Path) -> None: + capture_dir = make_capture( + tmp_path, + demo_rows(), + config={ + "pixel_ratio": PIXEL_RATIO, + "capture_locale": {"date_order": "dym"}, + }, + ) + with pytest.raises(ValueError, match="locale"): + convert_capture(capture_dir, tmp_path / "recording", settle_s=1.0) def test_frames_selected_from_video(converted: Path) -> None: diff --git a/packages/flow/tests/test_locale_invariants.py b/packages/flow/tests/test_locale_invariants.py new file mode 100644 index 000000000..f1acba259 --- /dev/null +++ b/packages/flow/tests/test_locale_invariants.py @@ -0,0 +1,740 @@ +"""Locale, date order and keyboard layout invariants (#1897).""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys + +import pytest + +from openadapt_flow.ir import ActionKind, Step, Workflow +from openadapt_flow.locale_invariants import ( + LocaleInvariantMismatch, + LocaleInvariants, + check_locale_invariants, + date_order_from_format, + enforce_workflow_locale_invariants, + observe_locale_invariants, + worklist_date_needs_explicit_format, + worklist_number_needs_explicit_format, +) + + +def test_matching_invariants_pass() -> None: + recorded = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + current = LocaleInvariants( + locale="en_US.UTF-8", date_order="mdy", keyboard_layout="us" + ) + assert check_locale_invariants(recorded, current).ok + + +def test_date_order_mismatch_refuses() -> None: + recorded = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + current = LocaleInvariants(locale="en_US", date_order="dmy", keyboard_layout="us") + result = check_locale_invariants(recorded, current) + assert not result.ok + assert result.mismatched == ("date_order",) + with pytest.raises(LocaleInvariantMismatch): + result.raise_if_refused() + + +def test_keyboard_mismatch_refuses_when_keystrokes_used() -> None: + recorded = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + current = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="fr") + assert not check_locale_invariants(recorded, current).ok + # Semantic value-setting may ignore keyboard layout. + assert check_locale_invariants(recorded, current, keystroke_steps=False).ok + + +def test_locale_mismatch_refuses() -> None: + recorded = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + current = LocaleInvariants(locale="fr_CA", date_order="mdy", keyboard_layout="us") + result = check_locale_invariants(recorded, current) + assert result.mismatched == ("locale",) + + +def test_worklist_ambiguous_numeric_date_needs_explicit_format() -> None: + assert worklist_date_needs_explicit_format("03/04/2026") + assert not worklist_date_needs_explicit_format("2026-03-04") + assert not worklist_date_needs_explicit_format("14/03/2026") # day > 12 + assert not worklist_date_needs_explicit_format("03 March 2026") + assert not worklist_date_needs_explicit_format("03/03/2026") # equal parts + + +def test_workflow_field_omitted_when_unset() -> None: + import json + + wf = Workflow(name="demo") + payload = json.loads(wf.model_dump_json()) + assert "locale_invariants" not in payload + loaded = Workflow.model_validate_json(wf.model_dump_json()) + assert loaded.locale_invariants is None + + +def test_workflow_round_trips_locale_invariants() -> None: + inv = LocaleInvariants(locale="fr_CA", date_order="dmy", keyboard_layout="ca") + wf = Workflow(name="demo", locale_invariants=inv) + loaded = Workflow.model_validate_json(wf.model_dump_json()) + assert loaded.locale_invariants is not None + assert loaded.locale_invariants.date_order == "dmy" + assert loaded.locale_invariants.keyboard_layout == "ca" + + +def test_enforce_refuses_mismatched_workflow() -> None: + wf = Workflow( + name="demo", + locale_invariants=LocaleInvariants( + locale="de_DE", date_order="dmy", keyboard_layout="de" + ), + steps=[ + Step(id="s1", action=ActionKind.TYPE, intent="type a date"), + ], + ) + current = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + with pytest.raises(LocaleInvariantMismatch) as excinfo: + enforce_workflow_locale_invariants(wf, current=current) + assert "date_order" in excinfo.value.mismatched + assert "locale" in excinfo.value.mismatched + + +def test_observe_accepts_explicit_overrides() -> None: + observed = observe_locale_invariants( + locale_name="fr_FR", date_order="dmy", keyboard_layout="fr" + ) + assert observed.locale.startswith("fr_fr") + assert observed.date_order == "dmy" + assert observed.keyboard_layout == "fr" + + +# ---------------------------------------------- review fixes: real detection + + +def _observe_under(env_locale: str) -> dict: + """Observe in a fresh interpreter whose environment names ``env_locale``. + + A real process, so Python's start-up ``C`` LC_TIME is in play: this is + the case that used to report ``mdy`` for every locale. + """ + + env = {k: v for k, v in os.environ.items() if not k.startswith("LC_")} + env.update(LANG=env_locale, LC_ALL=env_locale) + code = ( + "import json, sys\n" + "from openadapt_flow.locale_invariants import _posix_user_time_locale\n" + "name, order = _posix_user_time_locale()\n" + "print(json.dumps({'name': name, 'order': order}))\n" + ) + out = subprocess.run( + [sys.executable, "-c", code], + env=env, + capture_output=True, + text=True, + check=True, + timeout=60, + ) + return json.loads(out.stdout) + + +def _locale_available(name: str) -> bool: + out = subprocess.run( + [ + sys.executable, + "-c", + f"import locale; locale.setlocale(locale.LC_TIME, {name!r})", + ], + capture_output=True, + timeout=60, + ) + return out.returncode == 0 + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX locale reader") +@pytest.mark.parametrize( + "name,order", + [ + ("de_DE.UTF-8", "dmy"), + ("fr_FR.UTF-8", "dmy"), + ("en_GB.UTF-8", "dmy"), + ("en_US.UTF-8", "mdy"), + ], +) +def test_real_user_locale_sets_the_date_order(name: str, order: str) -> None: + if not _locale_available(name): + pytest.skip(f"{name} is not installed on this host") + seen = _observe_under(name) + assert seen["order"] == order + assert seen["name"] and seen["name"].lower().startswith(name[:5].lower()) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX locale reader") +def test_c_locale_is_unknown_not_en_us() -> None: + seen = _observe_under("C") + assert seen == {"name": None, "order": None} + + +def test_unknown_current_facts_refuse_instead_of_defaulting() -> None: + recorded = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + result = check_locale_invariants(recorded, LocaleInvariants()) + assert not result.ok + assert set(result.unknown) == {"locale", "date_order", "keyboard_layout"} + with pytest.raises(LocaleInvariantMismatch): + result.raise_if_refused() + + +def test_an_unreadable_layout_is_unknown_not_the_locale_language() -> None: + # The old code reported the locale's language ("en") as the layout. + seen = observe_locale_invariants(platform="linux", run=lambda cmd: None, env={}) + assert seen.keyboard_layout is None + + +def test_layout_change_under_the_same_locale_is_caught() -> None: + # en_US with a French AZERTY layout: the old code reported "en" for both. + def run(cmd: list[str]): + if cmd[:2] == ["setxkbmap", "-query"]: + return "rules: evdev\nmodel: pc105\nlayout: fr\n" + return None + + current = observe_locale_invariants( + platform="linux", run=run, env={"DISPLAY": ":0"}, xkb_group=lambda d: None + ) + assert current.keyboard_layout == "fr" + recorded = LocaleInvariants( + locale=current.locale, date_order=current.date_order, keyboard_layout="us" + ) + result = check_locale_invariants(recorded, current) + assert "keyboard_layout" in result.mismatched + + +def test_configured_layouts_without_an_x_server_are_unknown() -> None: + # localectl and XKB_DEFAULT_LAYOUT name configuration, not the layout + # the input target is using, so they no longer stand in for it. + def run(cmd: list[str]): + if cmd[:2] == ["localectl", "status"]: + return " X11 Layout: ca\n X11 Variant: multix\n" + return None + + env = {"XKB_DEFAULT_LAYOUT": "ca", "XKB_DEFAULT_VARIANT": "multix"} + seen = observe_locale_invariants(platform="linux", run=run, env=env) + assert seen.keyboard_layout is None + + +def test_macos_reader_uses_os_region_and_input_source() -> None: + answers = { + "AppleLocale": "de_DE@rg=chzzzz\n", + "AppleICUDateFormatStrings": '{\n 1 = "dd.MM.yy";\n}\n', + "AppleCurrentKeyboardLayoutInputSourceID": "com.apple.keylayout.German\n", + } + + def run(cmd: list[str]): + return answers.get(cmd[-1]) + + seen = observe_locale_invariants(platform="darwin", run=run) + assert seen.locale == "de_de" + assert seen.date_order == "dmy" + assert seen.keyboard_layout == "german" + + +@pytest.mark.skipif(sys.platform != "darwin", reason="real macOS defaults") +def test_real_macos_reader_reads_every_fact() -> None: + seen = observe_locale_invariants() + assert seen.locale and seen.date_order and seen.keyboard_layout + + +@pytest.mark.skipif(sys.platform != "win32", reason="real Windows locale APIs") +def test_real_windows_reader_reads_every_fact() -> None: + seen = observe_locale_invariants() + assert seen.locale and seen.date_order and seen.keyboard_layout + + +@pytest.mark.parametrize( + "fmt,order", + [ + ("%d.%m.%Y", "dmy"), + ("%e/%m/%y", "dmy"), + ("%-d/%-m/%Y", "dmy"), + ("%D", "mdy"), + ("%F", "ymd"), + ("M/d/yyyy", "mdy"), + ("dd.MM.yyyy", "dmy"), + ("yyyy-MM-dd", "ymd"), + ("dddd, d MMMM yyyy", "dmy"), + ("", None), + ("%H:%M", None), + ], +) +def test_date_order_from_strftime_and_icu_patterns(fmt: str, order) -> None: + assert date_order_from_format(fmt) == order + + +@pytest.mark.parametrize("raw", ["03/04/26", "03.04.2026", "3-4-26", "04/03", "0304"]) +def test_unrecognized_numeric_dates_need_a_format(raw: str) -> None: + assert worklist_date_needs_explicit_format(raw) + + +def test_decimal_separators_that_read_two_ways_need_a_format() -> None: + assert worklist_number_needs_explicit_format("1.234") + assert worklist_number_needs_explicit_format("1,234") + assert not worklist_number_needs_explicit_format("1.234,56") + assert not worklist_number_needs_explicit_format("1,234.56") + assert not worklist_number_needs_explicit_format("12.5") + + +def test_compile_carries_the_recording_invariants(tmp_path) -> None: + from openadapt_flow.compiler.compile import _locale_invariants_from_meta + + assert _locale_invariants_from_meta({}) is None + inv = _locale_invariants_from_meta( + {"locale_invariants": {"locale": "de_DE", "date_order": "dmy"}} + ) + assert inv.locale == "de_de" and inv.keyboard_layout is None + with pytest.raises(ValueError): + _locale_invariants_from_meta({"locale_invariants": {"date_order": "dym"}}) + with pytest.raises(ValueError): + _locale_invariants_from_meta({"locale_invariants": "de_DE"}) + + +# ------------------------------- re-review fixes: unknown recording facts + + +def test_a_layout_the_recording_could_not_read_refuses_keystrokes() -> None: + # Used to be skipped, so keystrokes replayed on any layout. + recorded = LocaleInvariants(locale="de_DE", date_order="dmy") + current = LocaleInvariants(locale="de_DE", date_order="dmy", keyboard_layout="us") + result = check_locale_invariants(recorded, current) + assert not result.ok + assert result.unrecorded == ("keyboard_layout",) + assert "when recording" in result.reason + # No keystroke step: the layout isn't needed. + assert check_locale_invariants(recorded, current, keystroke_steps=False).ok + + +def _typing_workflow(invariants) -> Workflow: + return Workflow( + name="demo", + locale_invariants=invariants, + steps=[Step(id="s1", action=ActionKind.TYPE, intent="type a date")], + ) + + +def test_a_bound_recording_with_every_fact_unknown_refuses() -> None: + here = LocaleInvariants(locale="en_US", date_order="mdy", keyboard_layout="us") + with pytest.raises(LocaleInvariantMismatch) as excinfo: + enforce_workflow_locale_invariants( + _typing_workflow(LocaleInvariants()), current=here + ) + assert set(excinfo.value.mismatched) == {"locale", "date_order", "keyboard_layout"} + # It survives the bundle round trip as a binding, not as "unbound". + loaded = Workflow.model_validate_json( + _typing_workflow(LocaleInvariants()).model_dump_json() + ) + assert loaded.locale_invariants is not None + with pytest.raises(LocaleInvariantMismatch): + enforce_workflow_locale_invariants(loaded, current=here) + + +def test_a_recording_from_before_the_binding_is_not_checked() -> None: + enforce_workflow_locale_invariants( + _typing_workflow(None), current=LocaleInvariants() + ) + + +def test_compile_keeps_an_all_unknown_binding() -> None: + from openadapt_flow.compiler.compile import _locale_invariants_from_meta + + unknown = {"locale": None, "date_order": None, "keyboard_layout": None} + bound = _locale_invariants_from_meta({"locale_invariants": unknown}) + assert bound is not None and bound.unknown() == list(unknown) + + +# ----------------------------- re-review fixes: the active input target + + +def _xkb_query(layouts: str, variants: str = "") -> str: + return f"rules: evdev\nlayout: {layouts}\nvariant: {variants}\n" + + +def test_the_x_server_wins_over_xkb_default_layout() -> None: + def run(cmd: list[str]): + return _xkb_query("fr") if cmd[:2] == ["setxkbmap", "-query"] else None + + seen = observe_locale_invariants( + platform="linux", + run=run, + env={"DISPLAY": ":0", "XKB_DEFAULT_LAYOUT": "us"}, + xkb_group=lambda d: 0, + ) + assert seen.keyboard_layout == "fr" + + +def test_the_active_xkb_group_picks_the_layout() -> None: + def run(cmd: list[str]): + return _xkb_query("us,de", ",nodeadkeys") + + def at(group): + return observe_locale_invariants( + platform="linux", run=run, env={"DISPLAY": ":0"}, xkb_group=lambda d: group + ).keyboard_layout + + # Switching groups changes the observed layout; it used to read "us,de" + # both times and compare equal. + assert at(0) == "us" + assert at(1) == "de+nodeadkeys" + # Several layouts and no readable group: unknown, not the list. + assert at(None) is None + assert at(5) is None + + +def test_wayland_layout_is_unknown() -> None: + def run(cmd: list[str]): + return _xkb_query("us") + + seen = observe_locale_invariants( + platform="linux", + run=run, + env={"DISPLAY": ":0", "WAYLAND_DISPLAY": "wayland-0"}, + xkb_group=lambda d: 0, + ) + assert seen.keyboard_layout is None + + +def test_macos_reads_text_input_sources_first() -> None: + def run(cmd: list[str]): + if "-c" in cmd: + return "com.apple.keylayout.German\ncom.apple.keylayout.German\n" + if cmd[-1] == "AppleCurrentKeyboardLayoutInputSourceID": + return "com.apple.keylayout.US\n" # stale saved value + return None + + seen = observe_locale_invariants(platform="darwin", run=run) + assert seen.keyboard_layout == "german" + + +def test_macos_input_method_is_part_of_the_layout() -> None: + def run(cmd: list[str]): + if "-c" in cmd: + return ( + "com.apple.keylayout.ABC\n" + "com.apple.inputmethod.Kotoeri.RomajiTyping.Japanese\n" + ) + return None + + seen = observe_locale_invariants(platform="darwin", run=run) + assert seen.keyboard_layout == ( + "abc+com.apple.inputmethod.kotoeri.romajityping.japanese" + ) + + +@pytest.mark.parametrize( + "hkl,klid", + [ + (0x04090409, "00000409"), # US + (0x04070407, "00000407"), # German + (0x08090409, "00000809"), # UK layout, English (US) language + (0xF0020409, "00010409"), # US-Dvorak, via its Layout Id + (0xE0010411, "E0010411"), # Japanese IME + (0xFFFFFFFFF0020409, "00010409"), # sign-extended 64-bit handle + (0, None), + (0xF0090409, None), # Layout Id the registry doesn't know + ], +) +def test_windows_klid_from_the_foreground_threads_hkl(hkl: int, klid) -> None: + from openadapt_flow.locale_invariants import klid_from_hkl + + def layout_id(value: int): + return {0x002: "00010409"}.get(value) + + assert klid_from_hkl(hkl, layout_id) == klid + + +def test_windows_reader_reads_the_foreground_thread() -> None: + import inspect + + from openadapt_flow import locale_invariants + + source = inspect.getsource(locale_invariants._windows) + assert "GetForegroundWindow" in source + assert "GetWindowThreadProcessId" in source + assert "GetKeyboardLayoutNameW" not in source + + +@pytest.mark.parametrize( + "platform,run,env", + [ + ("linux", lambda cmd: _xkb_query("us,de", ",nodeadkeys"), {"DISPLAY": ":0"}), + ("linux", lambda cmd: _xkb_query("fr"), {"DISPLAY": ":1"}), + ("linux", lambda cmd: None, {}), + ( + "darwin", + lambda cmd: { + "AppleLocale": "de_DE@rg=chzzzz\n", + "AppleICUDateFormatStrings": '{\n 1 = "dd.MM.yy";\n}\n', + }.get(cmd[-1], "com.apple.keylayout.German\ncom.apple.keylayout.German\n"), + None, + ), + ], +) +def test_capture_producer_reads_the_same_facts(platform, run, env) -> None: + # Capture records the facts and Flow checks them; the two readers must + # agree or every replay would refuse (or worse, match by accident). + facts = pytest.importorskip("openadapt_capture.locale_facts") + kwargs = dict(platform=platform, run=run, env=env, xkb_group=lambda d: 1) + produced = facts.read_os_locale_facts(**kwargs) + observed = observe_locale_invariants(**kwargs) + assert LocaleInvariants.model_validate(produced) == observed + assert produced == observed.model_dump() + + +def test_capture_producer_readers_are_the_same_code() -> None: + import inspect + + from openadapt_flow import locale_invariants + + facts = pytest.importorskip("openadapt_capture.locale_facts") + shared = [ + "_norm", + "_locale_family", + "date_order_from_format", + "_run", + "_posix_user_time_locale", + "_posix_date_order_for", + "_macos_layout", + "_macos", + "klid_from_hkl", + "_windows_klid_for_layout_id", + "_windows", + "_xkb_active_group", + "_xkb_layout", + "_posix", + "read_os_locale_facts", + ] + for name in shared: + assert inspect.getsource(getattr(facts, name)) == inspect.getsource( + getattr(locale_invariants, name) + ), name + assert facts._TIS_SCRIPT == locale_invariants._TIS_SCRIPT + + +# -- replayer: recheck after focus, before keyboard delivery ---------------------- + + +def _locale_replayer_harness(tmp_path, monkeypatch, *, layouts): + """Deterministic Replayer control with a scripted current layout. + + ``layouts`` is consumed by each ``observe_locale_invariants`` call. A + focusing click can advance the script (simulate an app that switches the + input layout on focus) by appending to ``layouts`` from the backend. + """ + + import io + + from PIL import Image + + from openadapt_flow.ir import ActionKind, Anchor, Step, Workflow + from openadapt_flow.runtime.replayer import Replayer + + observations: list[str | None] = [] + script = list(layouts) + + def observe(**_kwargs): + from openadapt_flow.locale_invariants import LocaleInvariants + + layout = script.pop(0) if script else "us" + observations.append(layout) + return LocaleInvariants( + locale="en_US", date_order="mdy", keyboard_layout=layout + ) + + monkeypatch.setattr( + "openadapt_flow.locale_invariants.observe_locale_invariants", observe + ) + + def make_png(size=(300, 200), color=(240, 240, 240)) -> bytes: + buf = io.BytesIO() + Image.new("RGB", size, color).save(buf, format="PNG") + return buf.getvalue() + + class Match: + def __init__(self, point, region, confidence=0.95): + self.point = point + self.region = region + self.confidence = confidence + + class Vision: + def __init__(self): + self.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20)), + Match(point=(110, 105), region=(100, 100, 50, 20)), + ] + self.pixels_changed_results: list = [] + + def find_template(self, *a, **k): + return self.template_results.pop(0) if self.template_results else None + + def find_structural_template(self, *a, **k): + return None + + def find_text(self, *a, **k): + return None + + def text_present(self, *a, **k): + return False + + def ocr(self, *a, **k): + return [] + + def pixels_changed(self, *a, **k): + if self.pixels_changed_results: + return self.pixels_changed_results.pop(0) + return True + + def phash_png(self, *a, **k): + return "aa" + + def phash_distance(self, *a, **k): + return 0 + + def wait_settled(self, backend, **k): + return backend.screenshot() + + class Backend: + def __init__(self, on_click=None): + self._frame = make_png() + self.actions: list = [] + self._on_click = on_click + self._viewport = (300, 200) + self._text_value = "" + + @property + def viewport(self): + return self._viewport + + def screenshot(self): + return self._frame + + def click(self, x, y, *, double=False): + self.actions.append(("click", x, y, double)) + if self._on_click is not None: + self._on_click() + + def type_text(self, text): + self.actions.append(("type", text)) + self._text_value = self._text_value + text + + def press(self, key): + self.actions.append(("press", key)) + + def text_value_at(self, x, y): + return self._text_value + + def focused_text_value(self): + return self._text_value + + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "btn.png").write_bytes(make_png((50, 20))) + run_dir = tmp_path / "run" + + from openadapt_flow.locale_invariants import LocaleInvariants + + workflow = Workflow( + name="locale-focus", + locale_invariants=LocaleInvariants( + locale="en_US", date_order="mdy", keyboard_layout="us" + ), + steps=[ + Step( + id="s1", + intent="focus field", + action=ActionKind.CLICK, + anchor=Anchor( + template="templates/btn.png", + region=(100, 100, 50, 20), + click_point=(110, 105), + ocr_text="Field", + ), + ), + Step( + id="s2", + intent="type note", + action=ActionKind.TYPE, + text="hello", + ), + ], + ) + return observations, script, Backend, Vision, workflow, bundle, run_dir, Replayer + + +def test_layout_change_after_focus_refuses_type_before_delivery( + tmp_path, monkeypatch +) -> None: + """P2 at 760632e99: a focusing click can switch the layout before TYPE. + + The run-start gate only saw US. After the click the controlled layout is + German; TYPE must refuse before keystroke delivery. + """ + + observations, script, Backend, Vision, workflow, bundle, run_dir, Replayer = ( + _locale_replayer_harness(tmp_path, monkeypatch, layouts=["us"]) + ) + + def on_click(): + # App switches the active input layout when the field takes focus. + script.append("de") + + backend = Backend(on_click=on_click) + vision = Vision() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + workflow, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert ("type", "hello") not in backend.actions + assert observations == ["us", "de"] + assert any(r.error and "keyboard_layout" in r.error for r in report.results), [ + r.error for r in report.results + ] + + +def test_stable_layout_still_passes_after_focus_recheck(tmp_path, monkeypatch) -> None: + """Positive: the same US layout before and after focus still succeeds.""" + + observations, script, Backend, Vision, workflow, bundle, run_dir, Replayer = ( + _locale_replayer_harness(tmp_path, monkeypatch, layouts=["us"]) + ) + + def on_click(): + script.append("us") + + backend = Backend(on_click=on_click) + vision = Vision() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + workflow, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is True + assert ("type", "hello") in backend.actions + assert observations == ["us", "us"] + + +def test_unavailable_layout_after_focus_refuses_type(tmp_path, monkeypatch) -> None: + """Refuse when the post-focus target observation cannot read the layout.""" + + observations, script, Backend, Vision, workflow, bundle, run_dir, Replayer = ( + _locale_replayer_harness(tmp_path, monkeypatch, layouts=["us"]) + ) + + def on_click(): + script.append(None) + + backend = Backend(on_click=on_click) + vision = Vision() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + workflow, bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert ("type", "hello") not in backend.actions + assert observations == ["us", None] + assert any( + r.error and "could not be read on this computer" in r.error + for r in report.results + ), [r.error for r in report.results]