focus_on: check who has the focus after a key that changes nothing visible

packages/flow/openadapt_flow/backends/playwright_backend.py

@@ -396,6 +396,44 @@ class _FramePoint:
     ' [role="radio"], [role="switch"]'
 )
 
+#: The focused element's own facts, in its document (``focus_identity``).
+_FOCUS_IDENTITY_JS = r"""(el) => {
+    const doc = el.ownerDocument;
+    if (el === doc.body || el === doc.documentElement) return {none: true};
+    const count = (sel) => {
+        try { return doc.querySelectorAll(sel).length; } catch (e) { return 0; }
+    };
+    const testid = el.getAttribute('data-testid') || null;
+    const id = el.id || null;
+    let active = false;
+    const activeId = el.getAttribute('aria-activedescendant');
+    if (activeId && doc.getElementById(activeId)) active = true;
+    if (!active) {
+        for (const attr of ['aria-controls', 'aria-owns']) {
+            const ref = el.getAttribute(attr);
+            const box = ref ? doc.getElementById(ref) : null;
+            if (box && box.querySelector('[role="option"][aria-selected="true"]')) {
+                active = true;
+                break;
+            }
+        }
+    }
+    let top = false;
+    try { top = doc.defaultView === doc.defaultView.top; } catch (e) { top = false; }
+    return {
+        none: false,
+        tag: el.localName,
+        testid: testid,
+        testid_count: testid ? count('[data-testid="' + CSS.escape(testid) + '"]') : 0,
+        id: id,
+        id_count: id ? count('#' + CSS.escape(id)) : 0,
+        active_option: active,
+        top: top,
+    };
+}"""
+#: The first line of an ARIA snapshot: ``- role "name"``.
+_ARIA_LINE = re.compile(r'^\s*-\s+([a-z]+)(?:\s+"((?:[^"\\]|\\.)*)")?')
+
 _BIND_COORDINATE_TARGET_JS = (
     r"""(args) => {
     const hit = document.elementFromPoint(args.x, args.y);
@@ -1870,6 +1908,61 @@ def focused_element_matches(self, selector: str) -> Optional[bool]:
         )
         return result if isinstance(result, bool) else None
 
+    def focus_identity(self) -> Optional[dict[str, Any]]:
+        """Who has the focus: its testid, id, role and name, and how many share them.
+
+        For the ``focus_on`` postcondition. ``testid_count`` / ``id_count``
+        count the elements with the same ``data-testid`` / ``id`` in the
+        focused element's document; ``role_name_count`` counts the elements
+        with the same role and accessible name on the page (Playwright's own
+        accessibility tree, as replay resolves a role and a name). Role and
+        name are read only in the top document. ``active_option`` is whether
+        an option of the focused control is active: its
+        ``aria-activedescendant`` names an element, or the listbox it
+        controls has an ``aria-selected`` option. ``{"none": True}`` when
+        nothing has the focus; None when it can't be read.
+        """
+
+        info = self._focused_element_call(_FOCUS_IDENTITY_JS)
+        if not isinstance(info, dict):
+            return None
+        if info.get("none"):
+            return {"none": True}
+        role: Optional[str] = None
+        name: Optional[str] = None
+        role_name_count = 0
+        if info.get("top"):
+            try:
+                snapshot = self.page.locator(":focus").aria_snapshot(timeout=1000)
+            except Exception:  # noqa: BLE001 - no snapshot: no role and name
+                snapshot = ""
+            first = snapshot.splitlines()[0] if snapshot else ""
+            match = _ARIA_LINE.match(first)
+            if match:
+                role = match.group(1)
+                name = (match.group(2) or "").replace('\\"', '"').strip() or None
+            if role and name:
+                try:
+                    role_name_count = self.page.get_by_role(
+                        role,  # type: ignore[arg-type]
+                        name=name,
+                        exact=True,
+                    ).count()
+                except Exception:  # noqa: BLE001 - uncountable is not unique
+                    role_name_count = 0
+        return {
+            "none": False,
+            "tag": info.get("tag"),
+            "testid": info.get("testid"),
+            "testid_count": int(info.get("testid_count") or 0),
+            "id": info.get("id"),
+            "id_count": int(info.get("id_count") or 0),
+            "role": role,
+            "name": name,
+            "role_name_count": role_name_count,
+            "active_option": info.get("active_option") is True,
+        }
+
     def element_texts(
         self, selector: str, frame_path: tuple[str, ...] = ()
     ) -> Optional[list[str]]:

packages/flow/openadapt_flow/compiler/compile.py

@@ -1125,9 +1125,79 @@ def _structural_postconditions(event: dict) -> list[Postcondition]:
         pcs.append(Postcondition(kind=PostconditionKind.URL_CHANGED))
     elif title_before and title_after and title_before != title_after:
         pcs.append(Postcondition(kind=PostconditionKind.TITLE_CHANGED))
+    if not pcs and event.get("kind") in ("key", "hotkey"):
+        focus = _focus_postcondition(
+            event.get("focus_before"), event.get("focus_after")
+        )
+        if focus is not None:
+            pcs.append(focus)
     return pcs
 
 
+def _focus_postcondition(before: Any, after: Any) -> Optional[Postcondition]:
+    """A ``focus_on`` check for a key that moved the focus, or None.
+
+    Only when the element focused after the key has a specific identity that
+    no other element on the page shared at teach time: a ``data-testid``,
+    then an ``id``, then a role with a non-empty accessible name. A tag, or
+    a role with no name, is not specific, and the step keeps no check. The
+    key must have moved the focus or made an option active; a key that
+    changed neither proves nothing about it.
+    """
+
+    if not isinstance(after, dict) or after.get("none") is not False:
+        return None
+    before = before if isinstance(before, dict) else {}
+
+    def ident(o: dict) -> tuple:
+        return (
+            o.get("tag"),
+            o.get("testid"),
+            o.get("id"),
+            o.get("role"),
+            o.get("name"),
+        )
+
+    moved = ident(before) != ident(after)
+    activated = after.get("active_option") is True and not (
+        not moved and before.get("active_option") is True
+    )
+    if not (moved or activated):
+        return None
+    common: dict[str, Any] = {"focus_active_option": activated}
+    testid, ident_id = after.get("testid"), after.get("id")
+    role, name = after.get("role"), after.get("name")
+    if isinstance(testid, str) and testid.strip() and after.get("testid_count") == 1:
+        return Postcondition(
+            kind=PostconditionKind.FOCUS_ON,
+            focus_by="testid",
+            focus_value=testid,
+            **common,
+        )
+    if isinstance(ident_id, str) and ident_id.strip() and after.get("id_count") == 1:
+        return Postcondition(
+            kind=PostconditionKind.FOCUS_ON,
+            focus_by="id",
+            focus_value=ident_id,
+            **common,
+        )
+    if (
+        isinstance(role, str)
+        and role.strip()
+        and isinstance(name, str)
+        and name.strip()
+        and after.get("role_name_count") == 1
+    ):
+        return Postcondition(
+            kind=PostconditionKind.FOCUS_ON,
+            focus_by="role_name",
+            focus_role=role,
+            focus_name=name,
+            **common,
+        )
+    return None
+
+
 def lint_param_leakage(workflow: Workflow, param_values: tuple[str, ...]) -> list[str]:
     """Scan a compiled workflow for demo parameter values baked in as
     literals outside the designated parameter slots.
@@ -1164,6 +1234,12 @@ def lint_param_leakage(workflow: Workflow, param_values: tuple[str, ...]) -> lis
     violations: list[str] = []
     for step in workflow.steps:
         for pc in step.expect:
+            for focus in (pc.focus_name, pc.focus_value):
+                if focus and _contains_excluded(focus, values):
+                    violations.append(
+                        f"{step.id}: focus_on postcondition {focus!r} embeds a "
+                        "demo parameter value"
+                    )
             if pc.text and _contains_excluded(pc.text, values):
                 violations.append(
                     f"{step.id}: {pc.kind.value} postcondition "

packages/flow/openadapt_flow/ir.py

@@ -528,6 +528,16 @@ class PostconditionKind(str, Enum):
     # shown, so a list that has not loaded yet never passes. Unreadable rows
     # fail the check.
     NO_DUPLICATE_CANDIDATE = "no_duplicate_candidate"
+    # A key press that moves the focus and changes nothing visible (a Tab
+    # onto an icon button, an ArrowDown that highlights an option): after the
+    # step the focused element has a specific identity, ``focus_by`` (a
+    # ``testid``, an ``id``, or a ``role_name``: a role and a non-empty
+    # accessible name), and no other element on the page has it. With
+    # ``focus_active_option`` an option of the focused control is active
+    # (``aria-activedescendant``, or ``aria-selected`` in the listbox it
+    # controls). Unobservable focus, or no specific identity, fails the check
+    # (:func:`focus_identity_is_specific`).
+    FOCUS_ON = "focus_on"
 
 
 class Postcondition(BaseModel):
@@ -555,6 +565,15 @@ class Postcondition(BaseModel):
     candidate_rows: Optional[str] = Field(default=None, min_length=1, max_length=256)
     #: NO_DUPLICATE_CANDIDATE: the frame chain the list is in (outermost first).
     candidate_frame_path: Optional[tuple[str, ...]] = Field(default=None, max_length=8)
+    #: FOCUS_ON: how the focused element is identified.
+    focus_by: Optional[Literal["testid", "id", "role_name"]] = None
+    #: FOCUS_ON: the ``data-testid`` or ``id`` (``focus_by`` testid / id).
+    focus_value: Optional[str] = Field(default=None, min_length=1, max_length=256)
+    #: FOCUS_ON: the ARIA role and accessible name (``focus_by`` role_name).
+    focus_role: Optional[str] = Field(default=None, min_length=1, max_length=64)
+    focus_name: Optional[str] = Field(default=None, min_length=1, max_length=256)
+    #: FOCUS_ON: an option of the focused control must be active.
+    focus_active_option: bool = False
     #: NO_DUPLICATE_CANDIDATE: parameters whose value in any row makes it a
     #: real candidate.
     candidate_params: Optional[tuple[str, ...]] = Field(
@@ -571,12 +590,44 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
             data.pop("field_selector", None)
         if self.param is None:
             data.pop("param", None)
-        for name in ("candidate_rows", "candidate_frame_path", "candidate_params"):
+        for name in (
+            "candidate_rows",
+            "candidate_frame_path",
+            "candidate_params",
+            "focus_by",
+            "focus_value",
+            "focus_role",
+            "focus_name",
+        ):
             if getattr(self, name) is None:
                 data.pop(name, None)
+        if not self.focus_active_option:
+            data.pop("focus_active_option", None)
         return data
 
 
+def focus_identity_is_specific(pc: "Postcondition") -> bool:
+    """Whether a FOCUS_ON postcondition names one element, not a kind of element.
+
+    A ``testid`` or an ``id`` with a value, or a role with a non-empty
+    accessible name. A tag alone, or a role with no name, names no element:
+    such a postcondition asserts nothing and never passes.
+    """
+
+    if pc.kind is not PostconditionKind.FOCUS_ON:
+        return False
+    if pc.focus_by in ("testid", "id"):
+        return bool(pc.focus_value and pc.focus_value.strip())
+    if pc.focus_by == "role_name":
+        return bool(
+            pc.focus_role
+            and pc.focus_role.strip()
+            and pc.focus_name
+            and pc.focus_name.strip()
+        )
+    return False
+
+
 IdentitySignalKeyValue = Literal[
     "subject_name",
     "record_id",

packages/flow/openadapt_flow/policy.py

@@ -28,7 +28,14 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from openadapt_flow.ir import ActionKind, Step, Workflow
+from openadapt_flow.ir import (
+    ActionKind,
+    Postcondition,
+    PostconditionKind,
+    Step,
+    Workflow,
+    focus_identity_is_specific,
+)
 from openadapt_flow.risk import classify_step_risk, is_local_edit, step_text
 from openadapt_flow.traversal import iter_workflow_steps
 from openadapt_flow.verification import declared_effect_is_independent_read
@@ -139,7 +146,15 @@ def is_vacuous(step: Step) -> bool:
         return False
     if step.action in (ActionKind.TYPE, ActionKind.SELECT_OPTION):
         return False
-    return expects_effect(step) and not step.expect
+    return expects_effect(step) and not any(_asserts(pc) for pc in step.expect)
+
+
+def _asserts(pc: Postcondition) -> bool:
+    """Whether a postcondition can fail: a FOCUS_ON must name one element."""
+
+    if pc.kind is PostconditionKind.FOCUS_ON:
+        return focus_identity_is_specific(pc)
+    return True
 
 
 def has_screen_postcondition(step: Step) -> bool:

packages/flow/openadapt_flow/recorder.py

@@ -661,6 +661,10 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None:
         """Capture before frame, act, wait settle, capture after, log event."""
         before = self._backend.screenshot()
         structural_before = self._structural_state()
+        # Who has the focus before and after a key press, for a key that moves
+        # the focus and changes nothing visible (the ``focus_on``
+        # postcondition). Key events only.
+        focus_before = self._focus_identity(event)
         # Structured identity of the clicked target (DOM / a11y text), when
         # the backend exposes it: captured on the BEFORE frame, before the
         # action, so the compiler can store it on the anchor and the replayer
@@ -757,6 +761,14 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None:
                 },
             }
         after = self._wait_settled()
+        if focus_before is not None:
+            focus_after = self._focus_identity(event)
+            if focus_after is not None:
+                event = {
+                    **event,
+                    "focus_before": focus_before,
+                    "focus_after": focus_after,
+                }
         opened = self._claim_new_pages(self._i)
         if opened:
             event = {**event, "opens": opened}
@@ -1046,6 +1058,20 @@ def _identity_census_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
             return None
         return census if isinstance(census, dict) else None
 
+    def _focus_identity(self, event: dict[str, Any]) -> Optional[dict[str, Any]]:
+        """The focused element's identity around a key event, if readable."""
+
+        if event.get("kind") not in ("key", "hotkey"):
+            return None
+        reader = getattr(self._backend, "focus_identity", None)
+        if not callable(reader):
+            return None
+        try:
+            value = reader()
+        except Exception:
+            return None
+        return value if isinstance(value, dict) else None
+
     def _structural_state(self) -> dict[str, Any]:
         """Structural observations the backend can provide right now.