From 9db6b5545353bdf89a3b92538ce50952e4a7a914 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 12:53:11 -0500 Subject: [PATCH 1/5] Locale, date order and keyboard layout as replay invariants (#1897) Record LocaleInvariants on the workflow (locale, date_order, keyboard_layout). Before replay, refuse when the current environment disagrees, unless the run has no keystroke typing. Ambiguous worklist numeric dates require an explicit format. Unit tests cover mismatches and bundle round-trip. --- packages/flow/openadapt_flow/ir.py | 7 + .../flow/openadapt_flow/locale_invariants.py | 241 ++++++++++++++++++ .../flow/openadapt_flow/runtime/replayer.py | 11 + packages/flow/tests/test_locale_invariants.py | 101 ++++++++ 4 files changed, 360 insertions(+) create mode 100644 packages/flow/openadapt_flow/locale_invariants.py create mode 100644 packages/flow/tests/test_locale_invariants.py diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index dd946abc4b..f8b2ac01a0 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 0000000000..56627b0b01 --- /dev/null +++ b/packages/flow/openadapt_flow/locale_invariants.py @@ -0,0 +1,241 @@ +"""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. Recordings may +carry :class:`LocaleInvariants`; before replay the runtime compares them to +the current environment and refuses on a mismatch, unless every typed step +sets values semantically (DOM/accessibility value) rather than by keystrokes. + +Observing the current environment uses only the Python ``locale`` module and +optional explicit overrides — no GUI is opened. +""" + +from __future__ import annotations + +import locale +from dataclasses import dataclass +from typing import Any, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +DateOrder = Literal["mdy", "dmy", "ymd"] + +__all__ = [ + "DateOrder", + "LocaleInvariantMismatch", + "LocaleInvariantResult", + "LocaleInvariants", + "check_locale_invariants", + "observe_locale_invariants", + "worklist_date_needs_explicit_format", +] + + +class LocaleInvariants(BaseModel): + """Environment facts recorded with a demonstration and checked at replay.""" + + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + locale: str = Field( + min_length=2, + max_length=32, + description="BCP-47 / POSIX locale id, e.g. en_US, fr_CA, de_DE.", + ) + date_order: DateOrder = Field( + description="How numeric dates are ordered: mdy, dmy, or ymd.", + ) + keyboard_layout: str = Field( + min_length=1, + max_length=64, + description=( + "Keyboard layout id as the OS reports it (e.g. us, fr, " + "ca-multilingual, de)." + ), + ) + + @field_validator("locale", "keyboard_layout") + @classmethod + def _normalize_token(cls, value: str) -> str: + text = value.strip().replace("-", "_") + if not text: + raise ValueError("must not be empty") + return text.lower() + + +class LocaleInvariantMismatch(RuntimeError): + """Current locale/date-order/keyboard layout disagrees with the recording.""" + + 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, ...] = () + reason: str = "" + + def raise_if_refused(self) -> None: + if not self.ok: + raise LocaleInvariantMismatch(self.reason, mismatched=list(self.mismatched)) + + +def _norm_locale(value: str) -> str: + return value.strip().replace("-", "_").lower() + + +def _locale_family(value: str) -> str: + """``en_us.utf-8`` -> ``en_us``; compare without encoding suffix.""" + + base = _norm_locale(value).split(".", 1)[0] + return base + + +def check_locale_invariants( + recorded: LocaleInvariants, + current: LocaleInvariants, + *, + keystroke_steps: bool = True, +) -> LocaleInvariantResult: + """Compare recorded invariants to the current environment. + + When ``keystroke_steps`` is False (every input step sets values + semantically), a keyboard-layout mismatch is allowed; locale and date + order still must match because worklist parsing and masked date fields + depend on them. + """ + + mismatched: list[str] = [] + if _locale_family(recorded.locale) != _locale_family(current.locale): + mismatched.append("locale") + if recorded.date_order != current.date_order: + mismatched.append("date_order") + if keystroke_steps and ( + _norm_locale(recorded.keyboard_layout) != _norm_locale(current.keyboard_layout) + ): + mismatched.append("keyboard_layout") + if mismatched: + return LocaleInvariantResult( + ok=False, + mismatched=tuple(mismatched), + reason=( + "recording environment does not match this computer: " + + ", ".join(mismatched) + + " (refuse rather than type the wrong punctuation or date)" + ), + ) + return LocaleInvariantResult(ok=True, reason="locale invariants match") + + +def observe_locale_invariants( + *, + date_order: Optional[DateOrder] = None, + keyboard_layout: Optional[str] = None, + locale_name: Optional[str] = None, +) -> LocaleInvariants: + """Read the current environment without opening a GUI. + + ``date_order`` and ``keyboard_layout`` must be supplied by the caller or by + the OS-specific capture layer when available; this helper only invents a + date order from the locale's ``d_fmt`` when ``date_order`` is omitted, and + defaults ``keyboard_layout`` to the locale language tag when unset (a + conservative placeholder until capture fills the real layout id). + """ + + if locale_name is None: + lang, _enc = locale.getlocale(locale.LC_TIME) + if not lang: + lang, _enc = locale.getlocale() + locale_name = lang or "en_US" + order = date_order + if order is None: + try: + d_fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] + except (AttributeError, ValueError): + d_fmt = "" + order = _date_order_from_d_fmt(d_fmt) or "mdy" + layout = keyboard_layout or _locale_family(locale_name).split("_", 1)[0] or "us" + return LocaleInvariants( + locale=locale_name, date_order=order, keyboard_layout=layout + ) + + +def _date_order_from_d_fmt(d_fmt: str) -> Optional[DateOrder]: + text = d_fmt.lower() + if not text: + return None + # Prefer the first day/month/year marker that appears. + positions = { + "d": min((i for i, c in enumerate(text) if c == "d"), default=10**9), + "m": min((i for i, c in enumerate(text) if c == "m"), default=10**9), + "y": min((i for i, c in enumerate(text) if c == "y"), default=10**9), + } + order = "".join(sorted(positions, key=positions.get)) # type: ignore[arg-type] + mapping = {"mdy": "mdy", "dmy": "dmy", "ymd": "ymd"} + return mapping.get(order) # type: ignore[return-value] + + +def worklist_date_needs_explicit_format(raw: str) -> bool: + """True when a numeric date is ambiguous without a declared order. + + Worklist parsing must never guess ``03/04/2026``. Named-month and ISO + dates are fine; equal day/month parts and day>12 clues are unambiguous. + """ + + import re + + 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 not m: + return False + a, b = int(m.group(1)), int(m.group(3)) + if a > 12 or b > 12 or a == b: + return False + return True + + +def workflow_uses_keystroke_typing(workflow: Any) -> bool: + """True when any step may type by keycode rather than a semantic value set.""" + + from openadapt_flow.ir import ActionKind + + semantic = False # reserved for a future Step.input_mode flag + _ = semantic + 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, + "type", + "key", + "hotkey", + ): + # A step that sets a DOM/accessibility value is still typed as TYPE + # today; until input_mode is recorded, treat TYPE as keystroke. + return True + return False + + +def enforce_workflow_locale_invariants( + workflow: Any, + *, + current: Optional[LocaleInvariants] = None, +) -> None: + """Raise :class:`LocaleInvariantMismatch` when a bound workflow disagrees.""" + + 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 0e9e10f3d3..3ce4cc1b11 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 diff --git a/packages/flow/tests/test_locale_invariants.py b/packages/flow/tests/test_locale_invariants.py new file mode 100644 index 0000000000..433715840c --- /dev/null +++ b/packages/flow/tests/test_locale_invariants.py @@ -0,0 +1,101 @@ +"""Locale, date order and keyboard layout invariants (#1897).""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.ir import ActionKind, Step, Workflow +from openadapt_flow.locale_invariants import ( + LocaleInvariantMismatch, + LocaleInvariants, + check_locale_invariants, + enforce_workflow_locale_invariants, + observe_locale_invariants, + worklist_date_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" From db3cf2b34e41ec60c2899264f23587fdbc051616 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 21:22:36 -0500 Subject: [PATCH 2/5] fix(flow): read locale, date order and layout from the OS, fail closed (#1897) Date order now comes from the user locale (setlocale(LC_TIME, "") on POSIX, AppleLocale/AppleICUDateFormatStrings on macOS, GetLocaleInfoEx on Windows), and the keyboard layout from the OS (HIToolbox input source, Windows KLID, XKB). A fact that can't be read is unknown, never en_US/mdy/us, and an unknown current fact for a recorded one refuses replay. Native capture conversion records the invariants and compile carries them into the workflow, so the gate is live for keycode replay. Worklist date shapes it can't read now need a declared format; ambiguous decimal separators are flagged; %e, %D, %F and ICU/Windows patterns parse. Adds real-process tests under C, de_DE, fr_FR, en_GB and en_US, and real macOS/Windows reader tests. --- .../flow/openadapt_flow/adapters/capture.py | 9 + .../flow/openadapt_flow/compiler/compile.py | 24 + .../flow/openadapt_flow/locale_invariants.py | 445 ++++++++++++++---- packages/flow/tests/test_locale_invariants.py | 197 ++++++++ 4 files changed, 579 insertions(+), 96 deletions(-) diff --git a/packages/flow/openadapt_flow/adapters/capture.py b/packages/flow/openadapt_flow/adapters/capture.py index 839e6d6862..56018b41c4 100644 --- a/packages/flow/openadapt_flow/adapters/capture.py +++ b/packages/flow/openadapt_flow/adapters/capture.py @@ -1266,6 +1266,15 @@ def convert_capture( "source": "openadapt-capture", "task_description": session.task_description, } + # Native replay types by keycode, so the computer's locale, date + # order and keyboard layout travel with the recording and are checked + # before replay (#1897). Conversion runs on the recording computer; + # facts it can't read are left unknown, never guessed. + from openadapt_flow.locale_invariants import observe_locale_invariants + + invariants = observe_locale_invariants() + if invariants.unknown() != ["locale", "date_order", "keyboard_layout"]: + meta["locale_invariants"] = invariants.model_dump() # 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 e4ce15e8f0..7a73ad7e8e 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/locale_invariants.py b/packages/flow/openadapt_flow/locale_invariants.py index 56627b0b01..b2d0ac5514 100644 --- a/packages/flow/openadapt_flow/locale_invariants.py +++ b/packages/flow/openadapt_flow/locale_invariants.py @@ -1,20 +1,35 @@ """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. Recordings may -carry :class:`LocaleInvariants`; before replay the runtime compares them to -the current environment and refuses on a mismatch, unless every typed step -sets values semantically (DOM/accessibility value) rather than by keystrokes. - -Observing the current environment uses only the Python ``locale`` module and -optional explicit overrides — no GUI is opened. +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. + +Facts come from the OS, not from Python's start-up ``C`` locale: + +* macOS: the global ``AppleLocale``, ``AppleICUDateFormatStrings`` and the + HIToolbox current keyboard layout, read with ``defaults``. +* Windows: ``GetUserDefaultLocaleName``, ``GetLocaleInfoEx`` (short date) and + ``GetKeyboardLayoutNameW``. +* Linux and other POSIX: the user locale via ``setlocale(LC_TIME, "")`` and + ``nl_langinfo(D_FMT)``, and the XKB layout from ``XKB_DEFAULT_LAYOUT``, + ``setxkbmap -query`` or ``localectl status``. + +A fact that can't be read is ``None`` (unknown). Unknown is never filled in +with ``en_US``, ``mdy`` or ``us``. At replay, an unknown current value for a +recorded fact refuses. 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, Literal, Optional +from typing import Any, Callable, Literal, Optional from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -26,44 +41,64 @@ "LocaleInvariantResult", "LocaleInvariants", "check_locale_invariants", + "date_order_from_format", + "enforce_workflow_locale_invariants", "observe_locale_invariants", + "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.""" + """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: str = Field( + locale: Optional[str] = Field( + default=None, min_length=2, - max_length=32, + max_length=64, description="BCP-47 / POSIX locale id, e.g. en_US, fr_CA, de_DE.", ) - date_order: DateOrder = Field( - description="How numeric dates are ordered: mdy, dmy, or ymd.", + date_order: Optional[DateOrder] = Field( + default=None, + description="How the OS orders numeric dates: mdy, dmy, or ymd.", ) - keyboard_layout: str = Field( + keyboard_layout: Optional[str] = Field( + default=None, min_length=1, - max_length=64, + max_length=128, description=( - "Keyboard layout id as the OS reports it (e.g. us, fr, " - "ca-multilingual, de)." + "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: str) -> str: + 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): - """Current locale/date-order/keyboard layout disagrees with the recording.""" + """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) @@ -75,22 +110,24 @@ def __init__(self, reason: str, *, mismatched: list[str]): class LocaleInvariantResult: ok: bool mismatched: tuple[str, ...] = () + unknown: tuple[str, ...] = () reason: str = "" def raise_if_refused(self) -> None: if not self.ok: - raise LocaleInvariantMismatch(self.reason, mismatched=list(self.mismatched)) + raise LocaleInvariantMismatch( + self.reason, mismatched=[*self.mismatched, *self.unknown] + ) -def _norm_locale(value: str) -> str: +def _norm(value: str) -> str: return value.strip().replace("-", "_").lower() def _locale_family(value: str) -> str: - """``en_us.utf-8`` -> ``en_us``; compare without encoding suffix.""" + """``en_us.utf-8`` / ``de_de@rg=chzzzz`` -> ``en_us`` / ``de_de``.""" - base = _norm_locale(value).split(".", 1)[0] - return base + return re.split(r"[.@]", _norm(value), maxsplit=1)[0] def check_locale_invariants( @@ -99,114 +136,334 @@ def check_locale_invariants( *, keystroke_steps: bool = True, ) -> LocaleInvariantResult: - """Compare recorded invariants to the current environment. + """Compare recorded invariants to this computer. Fails closed. - When ``keystroke_steps`` is False (every input step sets values - semantically), a keyboard-layout mismatch is allowed; locale and date - order still must match because worklist parsing and masked date fields - depend on them. + Each fact the recording holds must be readable here and 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 itself is not compared. """ + names = ["locale", "date_order"] + (["keyboard_layout"] if keystroke_steps else []) mismatched: list[str] = [] - if _locale_family(recorded.locale) != _locale_family(current.locale): - mismatched.append("locale") - if recorded.date_order != current.date_order: - mismatched.append("date_order") - if keystroke_steps and ( - _norm_locale(recorded.keyboard_layout) != _norm_locale(current.keyboard_layout) - ): - mismatched.append("keyboard_layout") - if mismatched: + unknown: list[str] = [] + for name in names: + want = getattr(recorded, name) + if want is None: + 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: + 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)) return LocaleInvariantResult( ok=False, mismatched=tuple(mismatched), - reason=( - "recording environment does not match this computer: " - + ", ".join(mismatched) - + " (refuse rather than type the wrong punctuation or date)" - ), + unknown=tuple(unknown), + 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) ------------------------------------------------------------ + +_Run = Callable[[list[str]], Optional[str]] +_LOCALE_LOCK = threading.Lock() + + +def _run(cmd: list[str]) -> Optional[str]: + try: + out = subprocess.run( + cmd, capture_output=True, text=True, timeout=3, 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) + + +def _macos(run: _Run) -> LocaleInvariants: + 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) + layout_raw = ( + run( + [ + "defaults", + "read", + "com.apple.HIToolbox", + "AppleCurrentKeyboardLayoutInputSourceID", + ] + ) + or "" + ).strip() + layout = layout_raw.rsplit(".", 1)[-1] if layout_raw else None + return LocaleInvariants(locale=name, date_order=order, keyboard_layout=layout) + + +def _windows() -> LocaleInvariants: # 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) + klid = ctypes.create_unicode_buffer(9) + if user32.GetKeyboardLayoutNameW(klid): + layout = klid.value or None + return LocaleInvariants(locale=name, date_order=order, keyboard_layout=layout) + + +def _layout_from(text: str, layout_re: str, variant_re: str) -> Optional[str]: + layout = re.search(layout_re, text, re.M) + if layout is None: + return None + variant = re.search(variant_re, text, re.M) + return layout.group(1) + (f"+{variant.group(1)}" if variant else "") + + +def _xkb_layout(run: _Run, env: dict[str, str]) -> Optional[str]: + if env.get("XKB_DEFAULT_LAYOUT"): + variant = env.get("XKB_DEFAULT_VARIANT") + return env["XKB_DEFAULT_LAYOUT"] + (f"+{variant}" if variant else "") + if env.get("DISPLAY"): + found = _layout_from( + run(["setxkbmap", "-query"]) or "", + r"^layout:\s*(\S+)", + r"^variant:\s*(\S+)", + ) + if found: + return found + return _layout_from( + run(["localectl", "status"]) or "", + r"X11 Layout:\s*(\S+)", + r"X11 Variant:\s*(\S+)", + ) + + +def _posix(run: _Run, env: dict[str, str]) -> LocaleInvariants: + name, order = _posix_user_time_locale() + return LocaleInvariants( + locale=_locale_family(name) if name else None, + date_order=order, + keyboard_layout=_xkb_layout(run, env), + ) + + 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, ) -> LocaleInvariants: - """Read the current environment without opening a GUI. + """Read this computer's locale, date order and keyboard layout. - ``date_order`` and ``keyboard_layout`` must be supplied by the caller or by - the OS-specific capture layer when available; this helper only invents a - date order from the locale's ``d_fmt`` when ``date_order`` is omitted, and - defaults ``keyboard_layout`` to the locale language tag when unset (a - conservative placeholder until capture fills the real layout id). + Explicit arguments override what the OS reports. A fact the OS doesn't + report stays ``None``. """ - if locale_name is None: - lang, _enc = locale.getlocale(locale.LC_TIME) - if not lang: - lang, _enc = locale.getlocale() - locale_name = lang or "en_US" - order = date_order - if order is None: - try: - d_fmt = locale.nl_langinfo(locale.D_FMT) # type: ignore[attr-defined] - except (AttributeError, ValueError): - d_fmt = "" - order = _date_order_from_d_fmt(d_fmt) or "mdy" - layout = keyboard_layout or _locale_family(locale_name).split("_", 1)[0] or "us" + plat = platform or sys.platform + runner = run or _run + environ = dict(os.environ if env is None else env) + try: + if plat == "darwin": + seen = _macos(runner) + elif plat == "win32": + seen = _windows() + else: + seen = _posix(runner, environ) + except Exception: # noqa: BLE001 - an unreadable fact is unknown, not a guess + seen = LocaleInvariants() return LocaleInvariants( - locale=locale_name, date_order=order, keyboard_layout=layout + locale=locale_name if locale_name is not None else seen.locale, + date_order=date_order if date_order is not None else seen.date_order, + keyboard_layout=( + keyboard_layout if keyboard_layout is not None else seen.keyboard_layout + ), ) -def _date_order_from_d_fmt(d_fmt: str) -> Optional[DateOrder]: - text = d_fmt.lower() - if not text: - return None - # Prefer the first day/month/year marker that appears. - positions = { - "d": min((i for i, c in enumerate(text) if c == "d"), default=10**9), - "m": min((i for i, c in enumerate(text) if c == "m"), default=10**9), - "y": min((i for i, c in enumerate(text) if c == "y"), default=10**9), - } - order = "".join(sorted(positions, key=positions.get)) # type: ignore[arg-type] - mapping = {"mdy": "mdy", "dmy": "dmy", "ymd": "ymd"} - return mapping.get(order) # type: ignore[return-value] +# -- worklist values ------------------------------------------------------------- def worklist_date_needs_explicit_format(raw: str) -> bool: - """True when a numeric date is ambiguous without a declared order. + """True unless a worklist date can be read without a declared order. - Worklist parsing must never guess ``03/04/2026``. Named-month and ISO - dates are fine; equal day/month parts and day>12 clues are unambiguous. + 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. """ - import re - 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 not m: - return False - a, b = int(m.group(1)), int(m.group(3)) - if a > 12 or b > 12 or a == b: + 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 True + 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 rather than a semantic value set.""" + """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 - semantic = False # reserved for a future Step.input_mode flag - _ = semantic for step in getattr(workflow, "steps", None) or (): kind = getattr(step, "action", None) or getattr(step, "kind", None) value = getattr(kind, "value", kind) @@ -214,12 +471,7 @@ def workflow_uses_keystroke_typing(workflow: Any) -> bool: ActionKind.TYPE.value, ActionKind.KEY.value, ActionKind.HOTKEY.value, - "type", - "key", - "hotkey", ): - # A step that sets a DOM/accessibility value is still typed as TYPE - # today; until input_mode is recorded, treat TYPE as keystroke. return True return False @@ -229,7 +481,8 @@ def enforce_workflow_locale_invariants( *, current: Optional[LocaleInvariants] = None, ) -> None: - """Raise :class:`LocaleInvariantMismatch` when a bound workflow disagrees.""" + """Raise :class:`LocaleInvariantMismatch` when a workflow's recorded + invariants disagree with this computer or can't be checked here.""" recorded = getattr(workflow, "locale_invariants", None) if recorded is None: diff --git a/packages/flow/tests/test_locale_invariants.py b/packages/flow/tests/test_locale_invariants.py index 433715840c..6d3de0f0e7 100644 --- a/packages/flow/tests/test_locale_invariants.py +++ b/packages/flow/tests/test_locale_invariants.py @@ -2,6 +2,11 @@ from __future__ import annotations +import json +import os +import subprocess +import sys + import pytest from openadapt_flow.ir import ActionKind, Step, Workflow @@ -9,9 +14,11 @@ 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, ) @@ -99,3 +106,193 @@ def test_observe_accepts_explicit_overrides() -> None: 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"} + ) + 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_xkb_variant_and_localectl_fallback() -> None: + def run(cmd: list[str]): + if cmd[:2] == ["localectl", "status"]: + return " X11 Layout: ca\n X11 Variant: multix\n" + return None + + seen = observe_locale_invariants(platform="linux", run=run, env={}) + assert seen.keyboard_layout == "ca+multix" + + +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"}) + + +def test_a_fact_the_recording_could_not_read_is_not_compared() -> None: + recorded = LocaleInvariants(locale="de_DE", date_order="dmy") + current = LocaleInvariants(locale="de_DE", date_order="dmy", keyboard_layout="us") + assert check_locale_invariants(recorded, current).ok From 19fc329029cb1ccd70475a56d76f81463c1e4204 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 21:41:22 -0500 Subject: [PATCH 3/5] test(flow): capture meta contract includes the host's locale invariants (#1897) --- packages/flow/tests/test_capture_adapter.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/flow/tests/test_capture_adapter.py b/packages/flow/tests/test_capture_adapter.py index c2d16fd97c..36fe9d06d5 100644 --- a/packages/flow/tests/test_capture_adapter.py +++ b/packages/flow/tests/test_capture_adapter.py @@ -567,8 +567,14 @@ 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. The one + # addition is this computer's locale, date order and keyboard layout + # (#1897), present whenever the OS reports any of them, and exactly what + # it reports. + from openadapt_flow.locale_invariants import observe_locale_invariants + + host = observe_locale_invariants() + expected = { "id", "created_at", "viewport", @@ -577,6 +583,10 @@ def test_meta_matches_recorder_contract(converted: Path) -> None: "source", "task_description", } + if host.unknown() != ["locale", "date_order", "keyboard_layout"]: + expected.add("locale_invariants") + assert meta["locale_invariants"] == host.model_dump() + assert set(meta.keys()) == expected def test_frames_selected_from_video(converted: Path) -> None: From 760632e99b773f5e6eab86e5399c118e71df73eb Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 22:24:35 -0500 Subject: [PATCH 4/5] fix(flow,capture): record locale facts at capture, refuse unknown ones, read the active layout (#1897) Re-review BLOCK at 19fc32902: - P1: an unknown recorded fact no longer skips the check. A bound workflow refuses when a fact it needs (locale and date order always, the layout for keystroke steps) could not be read when recording. A binding with every fact unknown stays a binding through capture, conversion, compile and the bundle, so it refuses instead of looking like a pre-#1897 one. - P1: capture's create_recording reads the facts when recording starts and persists them under config "capture_locale" (CaptureSession.locale_facts). convert_capture copies them as they are and never reads the converting computer. A legacy session without them stays unbound; a malformed value refuses conversion. - P2: the layout is the input target's. Linux reads the X server's active XKB group (XkbGetState) against setxkbmap's layout list; several layouts with no readable group, Wayland, XKB_DEFAULT_LAYOUT and localectl are unknown. macOS reads Text Input Sources (layout plus any active input method) in a child interpreter. Windows maps the foreground window thread's HKL to its KLID instead of GetKeyboardLayoutNameW. The capture reader is a stdlib-only copy of Flow's; a test checks the two are the same source. --- packages/capture/openadapt_capture/capture.py | 16 + .../capture/openadapt_capture/locale_facts.py | 445 ++++++++++++++++++ .../capture/openadapt_capture/recorder.py | 11 +- packages/capture/tests/test_locale_facts.py | 76 +++ .../flow/openadapt_flow/adapters/capture.py | 54 ++- .../flow/openadapt_flow/locale_invariants.py | 387 +++++++++++---- packages/flow/tests/test_capture_adapter.py | 92 +++- packages/flow/tests/test_locale_invariants.py | 230 ++++++++- 8 files changed, 1203 insertions(+), 108 deletions(-) create mode 100644 packages/capture/openadapt_capture/locale_facts.py create mode 100644 packages/capture/tests/test_locale_facts.py diff --git a/packages/capture/openadapt_capture/capture.py b/packages/capture/openadapt_capture/capture.py index 30189f8f5c..4f50deb906 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 0000000000..118fdd8cbc --- /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 131ef71748..b441bf92b4 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 0000000000..f0c61dbcd0 --- /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 56018b41c4..633e8df369 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,15 +1301,16 @@ def convert_capture( "source": "openadapt-capture", "task_description": session.task_description, } - # Native replay types by keycode, so the computer's locale, date - # order and keyboard layout travel with the recording and are checked - # before replay (#1897). Conversion runs on the recording computer; - # facts it can't read are left unknown, never guessed. - from openadapt_flow.locale_invariants import observe_locale_invariants - - invariants = observe_locale_invariants() - if invariants.unknown() != ["locale", "date_order", "keyboard_layout"]: - meta["locale_invariants"] = invariants.model_dump() + # 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/locale_invariants.py b/packages/flow/openadapt_flow/locale_invariants.py index b2d0ac5514..e4dd2bb7a9 100644 --- a/packages/flow/openadapt_flow/locale_invariants.py +++ b/packages/flow/openadapt_flow/locale_invariants.py @@ -5,19 +5,33 @@ :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``, ``AppleICUDateFormatStrings`` and the - HIToolbox current keyboard layout, read with ``defaults``. -* Windows: ``GetUserDefaultLocaleName``, ``GetLocaleInfoEx`` (short date) and - ``GetKeyboardLayoutNameW``. +* 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 XKB layout from ``XKB_DEFAULT_LAYOUT``, - ``setxkbmap -query`` or ``localectl status``. + ``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, +which is the recorded application at recording time but may be another +window when replay is checked. 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``. At replay, an unknown current value for a -recorded fact refuses. No GUI is opened. +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 @@ -43,7 +57,9 @@ "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", @@ -111,12 +127,14 @@ 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.reason, + mismatched=[*self.mismatched, *self.unknown, *self.unrecorded], ) @@ -138,19 +156,22 @@ def check_locale_invariants( ) -> LocaleInvariantResult: """Compare recorded invariants to this computer. Fails closed. - Each fact the recording holds must be readable here and 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 itself is not compared. + 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: @@ -162,16 +183,19 @@ def check_locale_invariants( same = _norm(str(want)) == _norm(str(have)) if not same: mismatched.append(name) - if mismatched or unknown: + 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)", @@ -230,15 +254,19 @@ def date_order_from_format(fmt: str) -> Optional[DateOrder]: # -- 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=3, check=False + cmd, capture_output=True, text=True, timeout=5, check=False ) except (OSError, subprocess.SubprocessError): return None @@ -291,7 +319,80 @@ def _posix_date_order_for(name: str) -> Optional[DateOrder]: locale.setlocale(locale.LC_TIME, saved) -def _macos(run: _Run) -> LocaleInvariants: +# 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 @@ -303,22 +404,58 @@ def _macos(run: _Run) -> LocaleInvariants: if order is None and name: lang, _, region = name.partition("_") order = _posix_date_order_for(f"{lang}_{region.upper()}" if region else lang) - layout_raw = ( - run( - [ - "defaults", - "read", - "com.apple.HIToolbox", - "AppleCurrentKeyboardLayoutInputSourceID", - ] - ) - or "" - ).strip() - layout = layout_raw.rsplit(".", 1)[-1] if layout_raw else None - return LocaleInvariants(locale=name, date_order=order, keyboard_layout=layout) + return name, order, _macos_layout(run) -def _windows() -> LocaleInvariants: # pragma: no cover - exercised on Windows CI +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] @@ -334,48 +471,136 @@ def _windows() -> LocaleInvariants: # pragma: no cover - exercised on Windows C LOCALE_SSHORTDATE = 0x1F if kernel32.GetLocaleInfoEx(name, LOCALE_SSHORTDATE, fmt, 80): order = date_order_from_format(fmt.value) - klid = ctypes.create_unicode_buffer(9) - if user32.GetKeyboardLayoutNameW(klid): - layout = klid.value or None - return LocaleInvariants(locale=name, date_order=order, keyboard_layout=layout) + # 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 -def _layout_from(text: str, layout_re: str, variant_re: str) -> Optional[str]: - layout = re.search(layout_re, text, re.M) - if layout is None: + path = ctypes.util.find_library("X11") + if not path: return None - variant = re.search(variant_re, text, re.M) - return layout.group(1) + (f"+{variant.group(1)}" if variant else "") - - -def _xkb_layout(run: _Run, env: dict[str, str]) -> Optional[str]: - if env.get("XKB_DEFAULT_LAYOUT"): - variant = env.get("XKB_DEFAULT_VARIANT") - return env["XKB_DEFAULT_LAYOUT"] + (f"+{variant}" if variant else "") - if env.get("DISPLAY"): - found = _layout_from( - run(["setxkbmap", "-query"]) or "", - r"^layout:\s*(\S+)", - r"^variant:\s*(\S+)", - ) - if found: - return found - return _layout_from( - run(["localectl", "status"]) or "", - r"X11 Layout:\s*(\S+)", - r"X11 Variant:\s*(\S+)", - ) + 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]) -> LocaleInvariants: +def _posix( + run: _Run, env: dict[str, str], group: _GroupReader +) -> tuple[Optional[str], Optional[DateOrder], Optional[str]]: name, order = _posix_user_time_locale() - return LocaleInvariants( - locale=_locale_family(name) if name else None, - date_order=order, - keyboard_layout=_xkb_layout(run, env), + 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, @@ -384,30 +609,27 @@ def observe_locale_invariants( 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 keyboard layout. + """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``. """ - plat = platform or sys.platform - runner = run or _run - environ = dict(os.environ if env is None else env) + seen = read_os_locale_facts( + platform=platform, run=run, env=env, xkb_group=xkb_group + ) try: - if plat == "darwin": - seen = _macos(runner) - elif plat == "win32": - seen = _windows() - else: - seen = _posix(runner, environ) - except Exception: # noqa: BLE001 - an unreadable fact is unknown, not a guess - seen = LocaleInvariants() + observed = LocaleInvariants.model_validate(seen) + except ValueError: + observed = LocaleInvariants() return LocaleInvariants( - locale=locale_name if locale_name is not None else seen.locale, - date_order=date_order if date_order is not None else seen.date_order, + 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 seen.keyboard_layout + keyboard_layout if keyboard_layout is not None else observed.keyboard_layout ), ) @@ -482,7 +704,12 @@ def enforce_workflow_locale_invariants( current: Optional[LocaleInvariants] = None, ) -> None: """Raise :class:`LocaleInvariantMismatch` when a workflow's recorded - invariants disagree with this computer or can't be checked here.""" + 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: diff --git a/packages/flow/tests/test_capture_adapter.py b/packages/flow/tests/test_capture_adapter.py index 36fe9d06d5..14eded9b61 100644 --- a/packages/flow/tests/test_capture_adapter.py +++ b/packages/flow/tests/test_capture_adapter.py @@ -567,13 +567,9 @@ 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. The one - # addition is this computer's locale, date order and keyboard layout - # (#1897), present whenever the OS reports any of them, and exactly what - # it reports. - from openadapt_flow.locale_invariants import observe_locale_invariants - - host = observe_locale_invariants() + # 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", @@ -583,12 +579,88 @@ def test_meta_matches_recorder_contract(converted: Path) -> None: "source", "task_description", } - if host.unknown() != ["locale", "date_order", "keyboard_layout"]: - expected.add("locale_invariants") - assert meta["locale_invariants"] == host.model_dump() 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: """Before frames precede each action; after frames show its effect.""" import cv2 diff --git a/packages/flow/tests/test_locale_invariants.py b/packages/flow/tests/test_locale_invariants.py index 6d3de0f0e7..b27bb2b583 100644 --- a/packages/flow/tests/test_locale_invariants.py +++ b/packages/flow/tests/test_locale_invariants.py @@ -197,7 +197,7 @@ def run(cmd: list[str]): return None current = observe_locale_invariants( - platform="linux", run=run, env={"DISPLAY": ":0"} + platform="linux", run=run, env={"DISPLAY": ":0"}, xkb_group=lambda d: None ) assert current.keyboard_layout == "fr" recorded = LocaleInvariants( @@ -207,14 +207,17 @@ def run(cmd: list[str]): assert "keyboard_layout" in result.mismatched -def test_xkb_variant_and_localectl_fallback() -> None: +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 - seen = observe_locale_invariants(platform="linux", run=run, env={}) - assert seen.keyboard_layout == "ca+multix" + 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: @@ -292,7 +295,222 @@ def test_compile_carries_the_recording_invariants(tmp_path) -> None: _locale_invariants_from_meta({"locale_invariants": "de_DE"}) -def test_a_fact_the_recording_could_not_read_is_not_compared() -> None: +# ------------------------------- 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") - assert check_locale_invariants(recorded, current).ok + 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 From 6b87bd651f8eabaa8da2aeb4d0049fb415faa29b Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 22:52:36 -0500 Subject: [PATCH 5/5] fix(flow): recheck keyboard layout after focus before typing (#1897) Re-review BLOCK at 760632e99: The run-start locale gate is not enough. A focusing click can change the active input layout before TYPE/KEY/HOTKEY. Recheck the target facts after focus and immediately before keyboard delivery, and refuse when they disagree with the recording or when a needed fact cannot be read. A stable-layout positive still passes. --- .../flow/openadapt_flow/locale_invariants.py | 14 +- .../flow/openadapt_flow/runtime/replayer.py | 43 ++++ packages/flow/tests/test_locale_invariants.py | 224 ++++++++++++++++++ 3 files changed, 275 insertions(+), 6 deletions(-) diff --git a/packages/flow/openadapt_flow/locale_invariants.py b/packages/flow/openadapt_flow/locale_invariants.py index e4dd2bb7a9..c3c11f3355 100644 --- a/packages/flow/openadapt_flow/locale_invariants.py +++ b/packages/flow/openadapt_flow/locale_invariants.py @@ -21,12 +21,14 @@ ``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, -which is the recorded application at recording time but may be another -window when replay is checked. 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. +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 diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 3ce4cc1b11..4977090ffe 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -14281,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 @@ -14417,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 ) @@ -14890,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.""" @@ -18012,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_locale_invariants.py b/packages/flow/tests/test_locale_invariants.py index b27bb2b583..f1acba2590 100644 --- a/packages/flow/tests/test_locale_invariants.py +++ b/packages/flow/tests/test_locale_invariants.py @@ -514,3 +514,227 @@ def test_capture_producer_readers_are_the_same_code() -> None: 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]