From e3dd16a518f098bda9ffd41c16218a20c8ef91cd Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 01:34:38 -0400 Subject: [PATCH 1/6] feat(readback): require a state reset and a person's admission for a demonstrated read-back A read-back taught by demonstration now has to throw away client state before it reads (a reload, a new search, a closed and reopened window, or a declared fresh session). The compiler refuses a check without one, so a stale form buffer can't pass. The compiled check is marked demonstrated and confirms nothing until a person admits it; the admission is bound to a SHA-256 of the exact check. Fields compare by kind (text, code, date, currency). An OCR read that differs only in O/0 or l/1, or a numeric date that reads two ways, is INDETERMINATE, never CONFIRMED. The macOS and Linux backends read marked fields from the accessibility tree. Effect evidence gains a derived assurance grade (independent_readback, screen_readback, unchecked) behind an adapter point for the openadapt-types enum; the success receipt records the weakest grade as effect_assurance. New CLI: record --with-check, check mark, check show, check admit. Signed-off-by: Richard Abrich --- packages/flow/openadapt_flow/__main__.py | 35 + packages/flow/openadapt_flow/assurance.py | 90 +++ packages/flow/openadapt_flow/authoring.py | 74 +- .../openadapt_flow/backends/linux_backend.py | 9 + .../openadapt_flow/backends/macos_backend.py | 9 + packages/flow/openadapt_flow/cli_check.py | 328 +++++++++ .../compiler/readback_authoring.py | 313 ++++++++- packages/flow/openadapt_flow/ir.py | 14 + packages/flow/openadapt_flow/receipt.py | 47 +- .../openadapt_flow/runtime/effects/effect.py | 95 ++- .../runtime/effects/onscreen.py | 138 +++- .../runtime/effects/readback_values.py | 247 +++++++ .../runtime/effects/ui_readback.py | 11 +- .../test_demonstrated_readback_verifier.py | 636 ++++++++++++++++++ .../flow/tests/test_readback_authoring.py | 17 +- packages/flow/tests/test_run_receipt.py | 1 + 16 files changed, 2012 insertions(+), 52 deletions(-) create mode 100644 packages/flow/openadapt_flow/assurance.py create mode 100644 packages/flow/openadapt_flow/cli_check.py create mode 100644 packages/flow/openadapt_flow/runtime/effects/readback_values.py create mode 100644 packages/flow/tests/test_demonstrated_readback_verifier.py diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 3baf2e40e..ab8eb735c 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -1047,6 +1047,30 @@ def _record_until_stop(args: argparse.Namespace): def _cmd_record(args: argparse.Namespace) -> int: + """Record, then (with ``--with-check``) ask for the demonstrated check.""" + with_check = bool(getattr(args, "with_check", False)) + if with_check: + from openadapt_flow.cli_check import RECORD_GUIDANCE + + print(RECORD_GUIDANCE) + code = _cmd_record_capture(args) + if code == 0 and with_check: + from openadapt_flow.cli_check import guided_check + + recording = Path(args.out) + if not (recording / "events.jsonl").is_file(): + print("record: no events.jsonl to mark a check in; skipped") + elif not sys.stdin.isatty(): + print( + "Mark the check with 'openadapt-flow check mark " + f"{recording} --start-event N --field PARAM=X,Y,W,H'." + ) + else: + guided_check(recording) + return code + + +def _cmd_record_capture(args: argparse.Namespace) -> int: # The interactive (web) recorder installs in-page DOM listeners against a # headed Playwright page. The DESKTOP recorder (--backend windows) captures # the operator's native OS input over the win_agent contract; both emit the @@ -6772,6 +6796,15 @@ def build_parser() -> argparse.ArgumentParser: "pre-profile default." ), ) + p.add_argument( + "--with-check", + action="store_true", + help=( + "After the save, also show how you'd check it saved (reload or " + "search again, open the record). Flow then asks which fields show " + "the values you typed and writes check.json" + ), + ) _add_backend_flags(p) p.set_defaults(func=_cmd_record) @@ -9561,8 +9594,10 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: pr.set_defaults(func=_cmd_connector) from openadapt_flow.cli_admit import register_admit_parser + from openadapt_flow.cli_check import register_check_parser register_admit_parser(sub) + register_check_parser(sub) return parser diff --git a/packages/flow/openadapt_flow/assurance.py b/packages/flow/openadapt_flow/assurance.py new file mode 100644 index 000000000..886ad84e2 --- /dev/null +++ b/packages/flow/openadapt_flow/assurance.py @@ -0,0 +1,90 @@ +"""How a write was checked, in three grades: the read-back assurance. + +``VerificationTier`` (``openadapt_flow.verification``) ranks evidence in four +steps, and the Seal ladder (``openadapt_types.oracle.OracleTier``) ranks it in +four other steps. A reviewer usually needs one plainer answer: was the save +read back independently of the session that typed it, read back on the same +screen, or not checked at all? :func:`assurance_for` gives that answer: + +- ``independent_readback``: a read that does not go through the acting + session. An API, database, FHIR, file or email read (tier 1), or the + application's record page read in a separate signed-in session + (``ui_readback``, tier 2). +- ``screen_readback``: the acting session read the value back off the screen, + after re-opening the record (tier 3) or on the screen the save left + (tier 4). +- ``unchecked``: no effect verifier confirmed the write. + +Policy rule in the current code. A separate-session UI read counts as +independent here because :meth:`VerificationTier.is_independent_of_actuation` +is True for tier 2, and :data:`openadapt_flow.verification.VERIFIED_EFFECT_TIER` +lets tier 2 reach ``VERIFIED``. It is still not a system-of-record read: the +``regulated`` policy pack requires tier 1, and the Seal ladder puts a second +session at oracle tier 1, below the production Seal minimum of 2. So +``independent_readback`` does not by itself mean a production Seal. + +ADAPTER POINT. The three names are the shared verification-grade vocabulary +being added to ``openadapt-types``. Until that enum is published, this module +defines them as plain strings. When it lands, replace :data:`Assurance` and +the constants below with the ``openadapt_types`` enum and keep +:func:`assurance_for` as the only mapping from Flow tiers. +""" + +from __future__ import annotations + +from typing import Final, Literal, Optional + +from openadapt_flow.verification import VerificationTier + +__all__ = [ + "ASSURANCE_GRADES", + "INDEPENDENT_READBACK", + "SCREEN_READBACK", + "UNCHECKED", + "Assurance", + "assurance_for", + "weakest_assurance", +] + +# -- ADAPTER POINT: replace with the openadapt-types enum when it is published. +Assurance = Literal["independent_readback", "screen_readback", "unchecked"] +INDEPENDENT_READBACK: Final = "independent_readback" +SCREEN_READBACK: Final = "screen_readback" +UNCHECKED: Final = "unchecked" +#: Strongest first. +ASSURANCE_GRADES: tuple[Assurance, ...] = ( + INDEPENDENT_READBACK, + SCREEN_READBACK, + UNCHECKED, +) +# -- end ADAPTER POINT + + +def assurance_for( + tier: Optional[int | VerificationTier], *, confirmed: bool = True +) -> Assurance: + """The assurance grade of one effect check. + + ``tier`` is the verifier's ``VerificationTier`` (1 to 4) and ``confirmed`` + whether its final verdict was CONFIRMED. An unconfirmed check, or one with + no tier, is ``unchecked``: a halted or unverifiable write was not shown to + have persisted, whatever verifier ran. + """ + + if not confirmed or tier is None: + return UNCHECKED + try: + value = VerificationTier(int(tier)) + except ValueError: + return UNCHECKED + if value.is_independent_of_actuation(): + return INDEPENDENT_READBACK + return SCREEN_READBACK + + +def weakest_assurance(grades: list[Assurance]) -> Assurance: + """The weakest grade among ``grades``; ``unchecked`` when empty.""" + + if not grades: + return UNCHECKED + return max(grades, key=ASSURANCE_GRADES.index) diff --git a/packages/flow/openadapt_flow/authoring.py b/packages/flow/openadapt_flow/authoring.py index 6095d39f0..39897659b 100644 --- a/packages/flow/openadapt_flow/authoring.py +++ b/packages/flow/openadapt_flow/authoring.py @@ -268,10 +268,13 @@ class CompileResult(_CompileResultRequired, total=False): """Authoring compile is not success of the job. ``distilled`` (steps in, steps out, dropped steps) is present when the - recording was distilled before compile. + recording was distilled before compile. ``check_needs_admission`` lists + the save steps whose demonstrated check a person must admit + (``openadapt-flow check admit``) before it can confirm a save. """ distilled: dict[str, Any] + check_needs_admission: list[str] class AdmitResult(TypedDict, total=False): @@ -510,6 +513,7 @@ def __init__( self._draft_bundle: Optional[Path] = None self._check_start: Optional[int] = None self._check_fields: list[Any] = [] + self._check_reset: Any = None self._backend_hints = _clean_backend_hints(kind, backend_hints) self._actors: dict[int, str] = {} #: Values typed for each parameter in this session, kept in memory @@ -795,12 +799,15 @@ def continue_input(self, *, operator_confirmed: bool = True) -> dict[str, Any]: return {"recorded": True, "param": pause.param} def begin_check(self) -> dict[str, Any]: - """Ask the person to show how to check the save. + """Ask the person to show how they'd check the save persisted. Events after this call are the check, not the task: the person - navigates to the saved record, then marks fields with - :meth:`mark_field`. Compile turns them into the save's read-back - (:mod:`openadapt_flow.compiler.readback_authoring`). + throws away what the screen remembers (a reload, a new search, a + reopened window), navigates to the saved record, then marks fields + with :meth:`mark_field`. Compile turns them into the save's read-back + (:mod:`openadapt_flow.compiler.readback_authoring`). The compiled + check confirms nothing until a person admits it + (``openadapt-flow check admit``). """ from openadapt_flow.compiler.readback_authoring import CHECK_PROMPT @@ -811,6 +818,33 @@ def begin_check(self) -> dict[str, Any]: self._check_start = recorder.event_count return {"prompt": CHECK_PROMPT, "checking": True} + def mark_reset(self, kind: str) -> dict[str, Any]: + """Name the last recorded step of the check as its state reset. + + For a reset the compiler can't recognize from the recording: a click + on the browser's reload button, a window reopened from the taskbar, + or signing out and back in. ``kind`` is ``reload``, ``new_search``, + ``reopen_window`` or ``fresh_session``. The compiled check records it + as declared, and the person who admits the check sees that. + """ + from openadapt_flow.compiler.readback_authoring import DeclaredReset + + recorder = self._require_recording() + if self._check_start is None: + raise AuthoringError("call begin_check first", code="no_check") + last = recorder.event_count - 1 + if last < self._check_start: + raise AuthoringError( + "record the reset step first, then name it", code="invalid_reset" + ) + try: + self._check_reset = DeclaredReset(kind=kind, event=last) # type: ignore[arg-type] + except ValueError as exc: + raise AuthoringError( + f"unknown reset kind {kind!r}", code="invalid_reset" + ) from exc + return {"reset": kind, "event": last} + def mark_field( self, *, @@ -818,8 +852,14 @@ def mark_field( node_id: Optional[str] = None, backend_pixels: Optional[Mapping[str, Any]] = None, locator: Optional[str] = None, + value_kind: Optional[str] = None, + date_order: Optional[str] = None, ) -> dict[str, Any]: - """Mark the element that shows the value typed for ``param``.""" + """Mark the element that shows the value typed for ``param``. + + ``value_kind`` (``text``, ``code``, ``date``, ``currency``) says how + to compare it; unset, the compiler infers it from the typed value. + """ from openadapt_flow.compiler.readback_authoring import MarkedField self._require_recording() @@ -857,6 +897,8 @@ def mark_field( region=target.region, locator=locator, structure_matched=structure_matched, + value_kind=value_kind, # type: ignore[arg-type] + date_order=date_order, # type: ignore[arg-type] ) ) return { @@ -958,6 +1000,15 @@ def compile( } if distilled is not None: result["distilled"] = distilled + from openadapt_flow.compiler.readback_authoring import demonstrated_checks + + pending = [ + step.id + for step, effect in demonstrated_checks(workflow) + if effect.readback is not None and effect.readback.admission is None + ] + if pending: + result["check_needs_admission"] = pending digest = None manifest = getattr(workflow, "manifest", None) raw_digest = getattr(manifest, "content_digest", None) @@ -1101,10 +1152,17 @@ def _finish_recording(self) -> Path: rec_dir = recorder.finish() self._stamp_provenance(rec_dir) if self._check_start is not None and self._check_fields: - from openadapt_flow.compiler.readback_authoring import write_check + from openadapt_flow.compiler.readback_authoring import ( + CHECK_PROMPT, + write_check, + ) write_check( - rec_dir, start_event=self._check_start, fields=self._check_fields + rec_dir, + start_event=self._check_start, + fields=self._check_fields, + reset=self._check_reset, + prompt=CHECK_PROMPT, ) self._recording_dir = rec_dir self._recorder = None diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index 7287d5c5a..bd9eea3f9 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -583,6 +583,15 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def displayed_text_at(self, x: int, y: int) -> Optional[str]: + """The text of the one AT-SPI element under (x, y), for a read-back. + + Same read as :meth:`structured_text_at`: on this backend that is the + element's own text, not a row. A demonstrated read-back compares it + with the expected value for equality. + """ + return self.structured_text_at(x, y) + def structured_text_at(self, x: int, y: int) -> Optional[str]: try: window, global_x, global_y = self._global_point( diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index dd0822834..7ff4466f9 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -1224,6 +1224,15 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def displayed_text_at(self, x: int, y: int) -> Optional[str]: + """The text of the one AX element under (x, y), for a read-back. + + Same read as :meth:`structured_text_at`: on this backend that is the + element's own text, not a row. A demonstrated read-back compares it + with the expected value for equality. + """ + return self.structured_text_at(x, y) + def structured_text_at(self, x: int, y: int) -> Optional[str]: """Return the exact AX text at/around pixel (x, y), or None.""" try: diff --git a/packages/flow/openadapt_flow/cli_check.py b/packages/flow/openadapt_flow/cli_check.py new file mode 100644 index 000000000..0a94608d7 --- /dev/null +++ b/packages/flow/openadapt_flow/cli_check.py @@ -0,0 +1,328 @@ +"""CLI: ``openadapt-flow check``, a read-back taught by demonstration. + +- ``check mark``: say where the check starts in a recording and which fields + it read. Writes ``check.json`` beside the recording; ``compile`` turns it + into the save's read-back. +- ``check show``: list a bundle's demonstrated checks and whether a person + has admitted each one. +- ``check admit``: a person admits the checks. It needs an interactive + terminal and a typed confirmation, so neither a script nor the agent that + drove the demonstration can admit them. + +``record --with-check`` prints the prompt "Now show how you'd check it +saved." and, after the recording, asks where the check started and which +fields it read. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Callable, Optional + +from openadapt_flow.compiler.readback_authoring import ( + CHECK_PROMPT, + CheckError, + DeclaredReset, + MarkedField, + admit_check, + demonstrated_checks, + describe_check, + write_check, +) + +#: What the person types to admit a check. +ADMIT_WORD = "admit" + +RECORD_GUIDANCE = ( + f"{CHECK_PROMPT}\n" + "Do the task and save. Then, in the same recording, show how an auditor " + "would prove the save persisted:\n" + " 1. throw away what the screen remembers: reload (F5), run a new " + "search, or close and reopen the window;\n" + " 2. open the saved record, or a read-only view of it;\n" + " 3. stop recording with the saved values on screen.\n" + "Flow then asks which fields show the values you typed." +) + + +def register_check_parser(sub: argparse._SubParsersAction) -> None: + parser = sub.add_parser( + "check", + help=( + "Teach a save's read-back by demonstration, and admit it. Nothing " + "confirms a save until a person admits the check." + ), + ) + verbs = parser.add_subparsers(dest="check_cmd", required=True) + + mark = verbs.add_parser( + "mark", help="Write check.json: where the check starts and what it reads" + ) + mark.add_argument("recording", help="Recording directory") + mark.add_argument( + "--start-event", + type=int, + required=True, + help="Index of the first recorded event of the check", + ) + mark.add_argument( + "--field", + action="append", + required=True, + metavar="PARAM=X,Y,W,H", + help="A field that shows the value typed for PARAM, in frame pixels", + ) + mark.add_argument( + "--kind", + action="append", + default=[], + metavar="PARAM=KIND", + help="Compare PARAM as text, code, date or currency (default: inferred)", + ) + mark.add_argument( + "--date-order", + action="append", + default=[], + metavar="PARAM=dmy|mdy", + help="Order of a numeric date field", + ) + mark.add_argument( + "--reset", + default=None, + metavar="KIND@EVENT", + help=( + "Name the state reset when Flow can't recognize it: reload, " + "new_search, reopen_window or fresh_session, at a recorded event" + ), + ) + mark.set_defaults(func=cmd_check_mark) + + show = verbs.add_parser("show", help="List a bundle's demonstrated checks") + show.add_argument("bundle", help="Bundle directory") + show.add_argument("--json", action="store_true", dest="as_json") + show.set_defaults(func=cmd_check_show) + + admit = verbs.add_parser( + "admit", + help="Admit a bundle's demonstrated checks (a person, at a terminal)", + ) + admit.add_argument("bundle", help="Bundle directory") + admit.add_argument("--step", default=None, help="Admit only this save step") + admit.add_argument("--by", default=None, help="Your name (default: the login name)") + admit.set_defaults(func=cmd_check_admit) + + +def _region(text: str) -> tuple[int, int, int, int]: + parts = [int(p) for p in text.split(",")] + if len(parts) != 4: + raise ValueError(f"a region is X,Y,W,H, not {text!r}") + return (parts[0], parts[1], parts[2], parts[3]) + + +def _pairs(items: list[str]) -> dict[str, str]: + out: dict[str, str] = {} + for item in items: + name, sep, value = item.partition("=") + if not sep or not name: + raise ValueError(f"expected NAME=VALUE, got {item!r}") + out[name] = value + return out + + +def build_marks( + fields: list[str], + *, + kinds: Optional[list[str]] = None, + date_orders: Optional[list[str]] = None, +) -> list[MarkedField]: + kind_of = _pairs(kinds or []) + order_of = _pairs(date_orders or []) + marks: list[MarkedField] = [] + for spec in fields: + param, sep, region = spec.partition("=") + if not sep or not param: + raise ValueError(f"expected PARAM=X,Y,W,H, got {spec!r}") + marks.append( + MarkedField( + param=param, + region=_region(region), + value_kind=kind_of.get(param), # type: ignore[arg-type] + date_order=order_of.get(param), # type: ignore[arg-type] + ) + ) + return marks + + +def parse_reset(text: Optional[str]) -> Optional[DeclaredReset]: + if not text: + return None + kind, sep, event = text.partition("@") + if not sep: + raise ValueError(f"expected KIND@EVENT, got {text!r}") + return DeclaredReset(kind=kind, event=int(event)) # type: ignore[arg-type] + + +def cmd_check_mark(args: argparse.Namespace) -> int: + try: + marks = build_marks(args.field, kinds=args.kind, date_orders=args.date_order) + reset = parse_reset(args.reset) + path = write_check( + Path(args.recording), + start_event=args.start_event, + fields=marks, + reset=reset, + prompt=CHECK_PROMPT, + ) + except (ValueError, OSError) as exc: + print(f"check mark: {exc}", file=sys.stderr) + return 2 + print(f"Wrote {path}. Compile the recording; a person then runs 'check admit'.") + return 0 + + +def _load(bundle: str): + from openadapt_flow.ir import Workflow + + return Workflow.load(Path(bundle)) + + +def cmd_check_show(args: argparse.Namespace) -> int: + workflow = _load(args.bundle) + found = demonstrated_checks(workflow) + rows = [] + for step, effect in found: + spec = effect.readback + assert spec is not None + rows.append( + { + "step": step.id, + "reset": None if spec.reset is None else spec.reset.kind, + "reset_evidence": None if spec.reset is None else spec.reset.evidence, + "fields": [f.name for f in spec.fields], + "admitted": spec.admitted(), + "admitted_by": (spec.admission.admitted_by if spec.admission else None), + } + ) + if args.as_json: + print(json.dumps(rows, indent=2, sort_keys=True)) + return 0 + if not rows: + print("No demonstrated check in this bundle.") + return 0 + for (step, effect), row in zip(found, rows): + print(describe_check(step, effect)) + state = ( + f"admitted by {row['admitted_by']}" if row["admitted"] else "NOT admitted" + ) + print(f" -> {state}\n") + return 0 + + +def cmd_check_admit( + args: argparse.Namespace, + *, + ask: Callable[[str], str] = input, + interactive: Optional[Callable[[], bool]] = None, +) -> int: + is_tty = interactive or (lambda: sys.stdin.isatty() and sys.stdout.isatty()) + if not is_tty(): + print( + "check admit: a person admits a check at an interactive terminal; " + "it can't be admitted by a script or by the demonstration", + file=sys.stderr, + ) + return 2 + from openadapt_flow.ir import Workflow + + bundle = Path(args.bundle) + workflow = Workflow.load(bundle) + found = [ + (step, effect) + for step, effect in demonstrated_checks(workflow) + if args.step is None or step.id == args.step + ] + if not found: + print("check admit: no demonstrated check to admit", file=sys.stderr) + return 2 + for step, effect in found: + print(describe_check(step, effect)) + print() + answer = ask( + f"Type '{ADMIT_WORD}' to accept these checks as proof that the save " + "persisted (screen read-back, same session): " + ) + if answer.strip().lower() != ADMIT_WORD: + print("Not admitted. Nothing changed.") + return 1 + try: + updated, steps = admit_check(workflow, admitted_by=args.by, step_id=args.step) + except CheckError as exc: + print(f"check admit: {exc}", file=sys.stderr) + return 2 + updated.save(bundle) + print(f"Admitted the check on {', '.join(steps)}.") + return 0 + + +def _summarize_event(index: int, event: dict) -> str: + kind = event.get("kind", "?") + if kind == "type": + what = f"type {event.get('param') or repr(event.get('text', ''))}" + elif kind in ("key", "hotkey"): + mods = "+".join([*event.get("modifiers", []), str(event.get("key", ""))]) + what = f"{kind} {mods}" + elif "x" in event: + what = f"{kind} at ({event.get('x')}, {event.get('y')})" + else: + what = str(kind) + return f" {index:3d} {what}" + + +def guided_check( + recording: Path, + *, + ask: Callable[[str], str] = input, + echo: Callable[[str], None] = print, +) -> Optional[Path]: + """After ``record --with-check``: ask where the check started and what it read. + + Returns the written ``check.json``, or None when the person skipped it. + """ + lines = (recording / "events.jsonl").read_text(encoding="utf-8").splitlines() + events = [json.loads(line) for line in lines if line.strip()] + echo(CHECK_PROMPT) + echo("Recorded events:") + for index, event in enumerate(events): + echo(_summarize_event(index, event)) + start = ask("Index of the first event of the check (empty to skip): ").strip() + if not start: + echo("No check marked. Mark it later with 'openadapt-flow check mark'.") + return None + fields: list[str] = [] + while True: + field = ask( + "A field that shows a value you typed, PARAM=X,Y,W,H in frame " + "pixels (empty when done): " + ).strip() + if not field: + break + fields.append(field) + if not fields: + echo("No field marked, so there is nothing to check.") + return None + reset = ask( + "If Flow can't see your reset (a click on reload, a sign-in), name it " + "as KIND@EVENT (empty to let Flow find it): " + ).strip() + path = write_check( + recording, + start_event=int(start), + fields=build_marks(fields), + reset=parse_reset(reset or None), + prompt=CHECK_PROMPT, + ) + echo(f"Wrote {path}. After compile, a person runs 'openadapt-flow check admit'.") + return path diff --git a/packages/flow/openadapt_flow/compiler/readback_authoring.py b/packages/flow/openadapt_flow/compiler/readback_authoring.py index 5a6692a6c..3379533cb 100644 --- a/packages/flow/openadapt_flow/compiler/readback_authoring.py +++ b/packages/flow/openadapt_flow/compiler/readback_authoring.py @@ -21,42 +21,65 @@ marked region must show the value typed for its parameter on the screen where it was marked, or compiling refuses. +State reset. Before it reads, the check must throw away what the client +remembers: reload the page, run a new search, close and reopen the window, or +start a fresh session. Otherwise it could read the form buffer or a +client-side page the save left behind, and a save that never reached the +record would pass. :func:`detect_reset` recognizes a reload key, a closed and +reopened window, and a typed and submitted search in the check's steps. The +person can also name the step in ``check.json`` (``reset``); the compiled +check records whether the step was detected or only declared. A check with no +reset is refused. + +Admission. The compiled check is marked ``demonstrated`` and has no +admission, so it confirms nothing until a person admits it +(:func:`admit_check`, ``openadapt-flow check admit``). The demonstration, +whether a person or an agent drove it, can't approve the check it taught. + How strong this is. The check reads the screen of the same application in the same session that did the save (``docs/EFFECT_KIT.md``): -- With a re-navigation of at least - :data:`~openadapt_flow.runtime.effects.effect.MIN_RENAV_ACTIONS` actions, it - is a persisted-state re-acquisition (Flow verification tier 3). Re-opening - the record catches a save that only the form showed. It does not catch a - page served from a cache the write filled, a duplicate submit, or a lost - update by another writer. -- With a shorter navigation (the person marked fields on the screen the save - left), it re-reads the save's own screen (tier 4): a consistency signal only. +- Every compiled check has a state reset, so it is a persisted-state + re-acquisition (Flow verification tier 3). Re-opening the record catches a + save that only the form showed. It does not catch a page served from a + cache the write filled, a duplicate submit, or a lost update by another + writer. +- Fields marked on the screen the save left, with no reset before them, are + refused. Before the reset rule they compiled to a re-read of the save's own + screen (tier 4). That read can still be configured by hand + (``effects.kind: onscreen``), not taught. - It is never a system-of-record read (tier 1) and never an independent - session (tier 2, ``ui_readback``). Under ``problems_only`` approval a save - checked this way still goes to a person (``onscreen_readback``). + session (tier 2, ``ui_readback``). Its assurance grade is + ``screen_readback`` (:mod:`openadapt_flow.assurance`). Under + ``problems_only`` approval a save checked this way still goes to a person + (``onscreen_readback``). Zero model calls at runtime: OCR of fixed regions and replayed input. """ from __future__ import annotations +import getpass import json +from datetime import datetime, timezone from pathlib import Path -from typing import Any, Optional +from typing import Any, Literal, Optional from pydantic import BaseModel, ConfigDict, Field from openadapt_flow.ir import ActionKind, Step, Workflow from openadapt_flow.runtime.effects.effect import ( - MIN_RENAV_ACTIONS, Effect, EffectKind, + ReadbackAdmission, ReadbackField, ReadbackNav, + ReadbackReset, + ReadbackResetKind, ReadbackSpec, ValueExpr, ) +from openadapt_flow.runtime.effects.readback_values import ValueKind, infer_value_kind __all__ = [ "CHECK_FILE", @@ -64,14 +87,51 @@ "CheckError", "CheckMarks", "CheckOutcome", + "DeclaredReset", "MarkedField", + "admit_check", "apply_check", + "demonstrated_checks", + "describe_check", + "detect_reset", "load_check", "write_check", ] CHECK_FILE = "check.json" -CHECK_PROMPT = "Show how to check the save" +CHECK_PROMPT = "Now show how you'd check it saved." + +#: Keys that reload the page or screen. +RELOAD_CHORDS = frozenset( + { + "f5", + "ctrl+f5", + "ctrl+r", + "control+r", + "ctrl+shift+r", + "control+shift+r", + "meta+r", + "cmd+r", + "command+r", + "meta+shift+r", + "cmd+shift+r", + } +) +#: Keys that close a window or tab. +CLOSE_CHORDS = frozenset( + { + "alt+f4", + "ctrl+f4", + "ctrl+w", + "control+w", + "meta+w", + "cmd+w", + "command+w", + "ctrl+shift+w", + "meta+q", + "cmd+q", + } +) class CheckError(ValueError): @@ -93,6 +153,21 @@ class MarkedField(BaseModel): #: structure equaled the value typed for :attr:`param`. The field is then #: read from the structure at runtime instead of by OCR. structure_matched: bool = False + #: How to compare the field: ``text``, ``code``, ``date`` or ``currency``. + #: Unset, the compiler infers it from the typed value. + value_kind: Optional[ValueKind] = None + #: ``date`` only: ``dmy`` or ``mdy`` for a numeric date. + date_order: Optional[Literal["dmy", "mdy"]] = None + + +class DeclaredReset(BaseModel): + """The person's word for which recorded event resets client state.""" + + model_config = ConfigDict(extra="forbid") + + kind: ReadbackResetKind + #: Recorded event index of the boundary (it must be part of the check). + event: int = Field(ge=0) class CheckMarks(BaseModel): @@ -105,6 +180,12 @@ class CheckMarks(BaseModel): #: the task). start_event: int fields: list[MarkedField] = Field(default_factory=list) + #: The step the person names as the state reset, when the compiler can't + #: recognize it from the recording (for example a click on a browser's + #: reload button, or signing out and back in). + reset: Optional[DeclaredReset] = None + #: The prompt the person was shown before the check was recorded. + prompt: Optional[str] = None class CheckOutcome(BaseModel): @@ -115,6 +196,10 @@ class CheckOutcome(BaseModel): fields: list[str] different_path: bool tier: str + reset: Optional[str] = None + reset_evidence: Optional[str] = None + #: Always True: a person must admit the check before it can confirm. + needs_admission: bool = True def write_check( @@ -122,9 +207,13 @@ def write_check( *, start_event: int, fields: list[MarkedField], + reset: Optional[DeclaredReset] = None, + prompt: Optional[str] = None, ) -> Path: """Store the check's marks beside the recording.""" - marks = CheckMarks(start_event=start_event, fields=fields) + marks = CheckMarks( + start_event=start_event, fields=fields, reset=reset, prompt=prompt + ) path = Path(recording_dir) / CHECK_FILE path.write_text(marks.model_dump_json(indent=2) + "\n", encoding="utf-8") return path @@ -162,14 +251,59 @@ def _as_nav(step: Step, workflow: Workflow) -> ReadbackNav: if step.text: return ReadbackNav(action="type", text=step.text) raise CheckError(f"the check's typing step {step.id} has no text") - if step.action is ActionKind.KEY and step.key: - return ReadbackNav(action="key", key=step.key) + if step.action in (ActionKind.KEY, ActionKind.HOTKEY) and step.key: + return ReadbackNav(action="key", key=_chord(step)) raise CheckError( f"the check's step {step.id} ({step.action.value}) can't be replayed as " "a read-back navigation. Use clicks, typing and keys" ) +def _chord(step: Step) -> str: + """The key as replayed, with its modifiers (``ctrl+r``).""" + if step.modifiers and step.key: + return "+".join([*step.modifiers, step.key]) + return step.key or "" + + +def _reset_kind_at(steps: list[Step], index: int) -> Optional[ReadbackResetKind]: + """The reset ``steps[index]`` performs, recognized from the recording.""" + step = steps[index] + if step.action in (ActionKind.KEY, ActionKind.HOTKEY) and step.key: + chord = _chord(step).lower().replace(" ", "") + if chord in RELOAD_CHORDS: + return "reload" + if chord in CLOSE_CHORDS and index + 1 < len(steps): + return "reopen_window" + if step.action is ActionKind.TYPE and (step.param or step.text): + submitted = any( + later.action in (ActionKind.CLICK, ActionKind.DOUBLE_CLICK) + or ( + later.action in (ActionKind.KEY, ActionKind.HOTKEY) + and (later.key or "").lower() in ("enter", "return") + ) + for later in steps[index + 1 :] + ) + if submitted: + return "new_search" + return None + + +def detect_reset(steps: list[Step]) -> Optional[ReadbackReset]: + """The first state-busting step among the check's ``steps``, if any. + + Recognized: a reload key (F5, Ctrl/Cmd+R), a window closed by key and + then reopened by a later step, and a search typed and then submitted (by + Enter or a click). A fresh sign-in can't be recognized: the check can't + replay a password. A person declares it instead. + """ + for index in range(len(steps)): + kind = _reset_kind_at(steps, index) + if kind is not None: + return ReadbackReset(kind=kind, at=index, evidence="detected") + return None + + def _norm(text: str) -> str: return " ".join(text.lower().split()) @@ -235,7 +369,10 @@ def apply_check( save = saves[-1] renavigation = [_as_nav(step, workflow) for step in check] - different_path = len(renavigation) >= MIN_RENAV_ACTIONS + reset = _reset(check, marks) + # The reset re-fetches the record, so the read takes a different path + # from the save even when the reset is the check's only step. + different_path = True params = dict(workflow.params) seen: set[str] = set() @@ -280,10 +417,12 @@ def apply_check( value=ValueExpr(param=field.param), locator=field.locator, source="structure" if field.structure_matched else "ocr", + value_kind=field.value_kind or infer_value_kind(str(params[field.param])), + date_order=field.date_order, ) for field in marks.fields ] - tier = "persisted_state_reacquisition" if different_path else "immediate_screen" + tier = "persisted_state_reacquisition" effect = Effect( kind=EffectKind.FIELD_EQUALS, field=marks.fields[0].param, @@ -292,14 +431,17 @@ def apply_check( readback=ReadbackSpec( region=marks.fields[0].region, different_path=different_path, - renavigation=renavigation if different_path else [], + renavigation=renavigation, fields=read_fields, + demonstrated=True, + reset=reset, ), probe=( - f"UI read-back shown by the person: {len(renavigation)} recorded " - f"navigation step(s), then read {len(read_fields)} marked field(s) " + f"UI read-back shown by demonstration: {len(renavigation)} recorded " + f"navigation step(s) with a {reset.kind.replace('_', ' ')} " + f"({reset.evidence}), then read {len(read_fields)} marked field(s) " "off the screen. Same application and session as the save; not a " - "system-of-record read." + "system-of-record read. Needs a person's admission." ), ) kept = [ @@ -316,9 +458,136 @@ def apply_check( fields=[f.param for f in marks.fields], different_path=different_path, tier=tier, + reset=reset.kind, + reset_evidence=reset.evidence, ) return updated, outcome +def _reset(check: list[Step], marks: CheckMarks) -> ReadbackReset: + """The check's state reset, or :class:`CheckError` when it shows none.""" + declared = marks.reset + if declared is not None: + positions = [_event_index(step) for step in check] + if declared.event not in positions: + raise CheckError( + f"the reset is declared at event {declared.event}, which isn't " + "a step of the check" + ) + at = positions.index(declared.event) + seen = _reset_kind_at(check, at) + return ReadbackReset( + kind=declared.kind, + at=at, + evidence="detected" if seen == declared.kind else "declared", + ) + detected = detect_reset(check) + if detected is None: + raise CheckError( + "the check reads the record without first throwing away what the " + "screen remembers, so it could read the form the save left behind. " + "Before reading, reload the page, run a new search, close and " + "reopen the window, or sign in again, and record that step " + "(or name it as the reset in check.json)" + ) + return detected + + +def demonstrated_checks(workflow: Workflow) -> list[tuple[Step, Effect]]: + """Every save step's read-back taught by demonstration.""" + found: list[tuple[Step, Effect]] = [] + for step in getattr(workflow, "steps", None) or []: + for effect in getattr(step, "effects", None) or []: + if effect.readback is not None and effect.readback.demonstrated: + found.append((step, effect)) + return found + + +def describe_check(step: Step, effect: Effect) -> str: + """What a person is asked to admit, in plain words.""" + spec = effect.readback + assert spec is not None + lines = [f"Save step {step.id} ({step.intent}) is checked by:"] + for i, nav in enumerate(spec.renavigation): + if nav.action == "click": + what = f"click at {nav.point}" + elif nav.action == "type": + what = ( + f"type this run's {nav.param!r}" if nav.param else f"type {nav.text!r}" + ) + else: + what = f"press {nav.key}" + marker = "" + if spec.reset is not None and spec.reset.at == i: + marker = f" <- state reset: {spec.reset.kind} ({spec.reset.evidence})" + lines.append(f" {i + 1}. {what}{marker}") + for field in spec.fields: + lines.append( + f" then read {field.name!r} ({field.value_kind}, by {field.source}) " + f"at {tuple(field.region)}" + ) + lines.append( + "This is a screen read-back in the session that saved: assurance " + "screen_readback, never a system-of-record read." + ) + return "\n".join(lines) + + +def admit_check( + workflow: Workflow, + *, + admitted_by: Optional[str] = None, + step_id: Optional[str] = None, + now: Optional[datetime] = None, +) -> tuple[Workflow, list[str]]: + """Record a person's admission of the demonstrated checks. + + Binds each admission to the exact check (``spec_digest``). The caller + saves the workflow (which re-seals the bundle manifest) and is + responsible for asking a person: the CLI requires a terminal and a typed + confirmation. Returns the updated workflow and the admitted step ids. + """ + who = (admitted_by or getpass.getuser() or "").strip() + if not who: + raise CheckError("an admission needs the name of the person admitting it") + stamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + when = stamp.replace(microsecond=0).isoformat().replace("+00:00", "Z") + admitted: list[str] = [] + steps = [] + for step in workflow.steps: + effects = [] + for effect in step.effects: + spec = effect.readback + if ( + spec is not None + and spec.demonstrated + and (step_id is None or step.id == step_id) + ): + if spec.reset is None: + raise CheckError( + f"the check on {step.id} has no state reset and can't " + "be admitted" + ) + clean = spec.model_copy(update={"admission": None}) + admission = ReadbackAdmission( + admitted_by=who, + admitted_at=when, + spec_digest=clean.admission_digest(), + ) + effect = effect.model_copy( + update={ + "readback": clean.model_copy(update={"admission": admission}) + } + ) + admitted.append(step.id) + effects.append(effect) + steps.append(step.model_copy(update={"effects": effects})) + if not admitted: + raise CheckError( + "no demonstrated check to admit" + (f" on {step_id}" if step_id else "") + ) + return workflow.model_copy(update={"steps": steps}), admitted + + def outcome_json(outcome: CheckOutcome) -> str: return json.dumps(outcome.model_dump(), indent=2, sort_keys=True) diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index 2af0f4c27..01a8dc797 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -2999,6 +2999,20 @@ class EffectVerificationEvidence(BaseModel): #: keeps its exact bytes. endpoint_scope: Literal["loopback", "external", "unknown"] = "unknown" + @property + def assurance(self) -> str: + """How the write was checked: ``independent_readback``, + ``screen_readback`` or ``unchecked`` (:mod:`openadapt_flow.assurance`). + + Derived from :attr:`verification_tier` and :attr:`final_verdict`, so + it adds nothing to the stored record and can't disagree with it. + """ + from openadapt_flow.assurance import assurance_for + + return assurance_for( + self.verification_tier, confirmed=self.final_verdict == "confirmed" + ) + @model_serializer(mode="wrap") def _serialize_compatible(self, handler: Any) -> dict[str, Any]: """Omit the additive ``endpoint_scope`` default from serialized evidence. diff --git a/packages/flow/openadapt_flow/receipt.py b/packages/flow/openadapt_flow/receipt.py index 2d80c7244..0fe7bb074 100644 --- a/packages/flow/openadapt_flow/receipt.py +++ b/packages/flow/openadapt_flow/receipt.py @@ -40,7 +40,7 @@ from pathlib import Path from typing import Annotated, Any, Literal, Optional, cast -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator from openadapt_flow.ir import RunReport from openadapt_flow.verification import VerificationTier @@ -87,6 +87,16 @@ "persisted_state_reacquisition", ] +#: How the weakest effect was checked (:mod:`openadapt_flow.assurance`). A +#: success receipt never carries ``unchecked``. +ReceiptAssurance = Literal["independent_readback", "screen_readback"] + +_ASSURANCE_BY_TIER: dict[str, ReceiptAssurance] = { + "independent_system": "independent_readback", + "independent_session": "independent_readback", + "persisted_state_reacquisition": "screen_readback", +} + ReceiptSubstrate = Literal["web", "windows", "macos", "linux", "rdp", "citrix"] #: ``synthetic-tutorial`` marks a run against the bundled synthetic application, @@ -152,6 +162,10 @@ class RunReceipt(BaseModel): rung_histogram: dict[ReceiptRung, ReceiptCount] = Field(default_factory=dict) evidence_classes: list[ReceiptEvidenceClass] = Field(default_factory=list) effect_tier_reached: ReceiptEffectTier + #: The read-back assurance grade of the weakest effect, derived from + #: :attr:`effect_tier_reached`. Optional so receipts written before it + #: existed still validate (``exclude_none`` keeps their digest). + effect_assurance: Optional[ReceiptAssurance] = None authorization_required: ReceiptCount authorization_confirmed: ReceiptCount identity_required: ReceiptCount @@ -233,6 +247,11 @@ def _complete_verified_contract(self) -> "RunReceipt": raise ValueError("VERIFIED receipt lacks effect evidence") if "effect_tier_4" in self.evidence_classes: raise ValueError("immediate-screen evidence cannot produce VERIFIED") + if ( + self.effect_assurance is not None + and self.effect_assurance != _ASSURANCE_BY_TIER[self.effect_tier_reached] + ): + raise ValueError("effect assurance disagrees with the effect tier") if (self.model_calls > 0) != ("model" in self.evidence_classes): raise ValueError("model-call count and evidence class disagree") if self.model_calls > 0 and self.external_network_calls != "observed": @@ -242,6 +261,18 @@ def _complete_verified_contract(self) -> "RunReceipt": raise ValueError("receipt digest does not match its canonical payload") return self + @model_serializer(mode="wrap") + def _drop_unset_assurance(self, handler: Any) -> dict[str, Any]: + """Keep the exact bytes of receipts that carry no ``effect_assurance``. + + The signed production projection hashes ``model_dump``; an added + ``null`` key would change that hash for every older receipt. + """ + data: dict[str, Any] = handler(self) + if self.effect_assurance is None: + data.pop("effect_assurance", None) + return data + def canonical_json(self, *, include_digest: bool = True) -> bytes: """Deterministic bytes for hashing and for the publish diff.""" @@ -776,6 +807,7 @@ def _build_receipt( "effect evidence" ) + weakest_tier = _weakest_confirmed_tier(report) rungs: dict[str, int] = {} for rung, count in dict(report.rung_counts).items(): if rung not in ReceiptRung.__args__: # type: ignore[attr-defined] @@ -804,7 +836,8 @@ def _build_receipt( duration_ms=int(round(float(report.total_ms))), rung_histogram=rungs, # type: ignore[arg-type] evidence_classes=evidence_classes, # type: ignore[arg-type] - effect_tier_reached=_weakest_confirmed_tier(report), + effect_tier_reached=weakest_tier, + effect_assurance=_ASSURANCE_BY_TIER[weakest_tier], authorization_required=int(required.authorization), authorization_confirmed=int(passed.authorization), identity_required=int(required.identity), @@ -873,6 +906,11 @@ def render_receipt_markdown(receipt: RunReceipt) -> str: f"| heals | {receipt.heals} |", f"| effects confirmed | {effects} |", f"| weakest effect evidence | `{receipt.effect_tier_reached}` |", + *( + [f"| read-back assurance | `{receipt.effect_assurance}` |"] + if receipt.effect_assurance + else [] + ), f"| identity-armed steps | {identity} |", f"| over-halts | {receipt.over_halt_count} |", f"| resolution ladder | {ladder} |", @@ -939,6 +977,11 @@ def _receipt_rows(receipt: RunReceipt) -> list[tuple[str, str]]: f"{receipt.effects_confirmed} of {receipt.effects_required}", ), ("effect evidence", receipt.effect_tier_reached), + *( + [("read-back assurance", receipt.effect_assurance)] + if receipt.effect_assurance + else [] + ), ( "identity-armed", f"{receipt.identity_armed} of {receipt.identity_applicable}", diff --git a/packages/flow/openadapt_flow/runtime/effects/effect.py b/packages/flow/openadapt_flow/runtime/effects/effect.py index aa33a9c8a..93edc7ac7 100644 --- a/packages/flow/openadapt_flow/runtime/effects/effect.py +++ b/packages/flow/openadapt_flow/runtime/effects/effect.py @@ -247,15 +247,68 @@ class ReadbackField(BaseModel): #: region's center from the page structure (DOM, or accessibility tree); #: when the backend can't, the read is INDETERMINATE, never OCR. source: Literal["ocr", "structure"] = "ocr" + #: How the read and the expected value are compared + #: (:mod:`openadapt_flow.runtime.effects.readback_values`): ``text``, + #: ``code`` (case and separators ignored), ``date`` or ``currency``. + value_kind: Literal["text", "code", "date", "currency"] = "text" + #: ``date`` only: the order of a numeric date such as ``03/04/2026``. + #: Unset, a numeric date that reads two ways is ambiguous. + date_order: Optional[Literal["dmy", "mdy"]] = None @model_serializer(mode="wrap") def _drop_default_source(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) if self.source == "ocr": data.pop("source", None) + if self.value_kind == "text": + data.pop("value_kind", None) + if self.date_order is None: + data.pop("date_order", None) return data +#: State-busting boundaries a demonstrated check can show before it reads. +ReadbackResetKind = Literal["reload", "new_search", "reopen_window", "fresh_session"] + + +class ReadbackReset(BaseModel): + """The step of a demonstrated check that throws away client-side state. + + Without one, the check could read a form buffer or a client-side page the + save left behind, and a save that never reached the record would pass. + """ + + model_config = ConfigDict(extra="forbid") + + kind: ReadbackResetKind + #: Index into :attr:`ReadbackSpec.renavigation` of the boundary step. + at: int = Field(ge=0) + #: ``detected``: the compiler recognized the step (a reload key, a closed + #: window, a typed and submitted search). ``declared``: the person said + #: which step it is and the compiler could not check it from the + #: recording; the person who admits the check sees which. + evidence: Literal["detected", "declared"] = "detected" + + +class ReadbackAdmission(BaseModel): + """A person's acceptance of a read-back taught by demonstration. + + The demonstration that taught the check cannot approve it. Until a person + admits it, the check can't confirm a save (the verifier rules + ``INDETERMINATE``). The admission is bound to the exact check by + :attr:`spec_digest`; a changed check needs a new admission. + """ + + model_config = ConfigDict(extra="forbid") + + admitted_by: str = Field(min_length=1) + #: ISO-8601 UTC timestamp. + admitted_at: str = Field(min_length=1) + spec_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + #: Who admitted it. Only a person can. + role: Literal["person"] = "person" + + class ReadbackSpec(BaseModel): """The on-screen read-back oracle mined from the demonstration. @@ -290,16 +343,46 @@ class ReadbackSpec(BaseModel): #: Several fields to read after one re-navigation. When set, every field #: must read back; :attr:`region` is the first field's region. fields: list[ReadbackField] = Field(default_factory=list) + #: True when a person or agent taught this check by demonstration + #: (``check.json``). Such a check confirms nothing until + #: :attr:`admission` records that a person accepted it. + demonstrated: bool = False + #: The state-busting step the demonstrated check shows before it reads. + reset: Optional[ReadbackReset] = None + #: A person's acceptance of this exact check. + admission: Optional[ReadbackAdmission] = None @model_serializer(mode="wrap") def _drop_empty_fields(self, handler: Any) -> dict[str, Any]: - """Keep the bytes of read-back contracts that predate ``fields``.""" + """Keep the bytes of read-back contracts that predate these fields.""" data: dict[str, Any] = handler(self) if not self.fields: data.pop("fields", None) + if not self.demonstrated: + data.pop("demonstrated", None) + if self.reset is None: + data.pop("reset", None) + if self.admission is None: + data.pop("admission", None) return data + def admission_digest(self) -> str: + """SHA-256 of the check a person admits: everything but the admission.""" + + payload = self.model_copy(update={"admission": None}).model_dump(mode="json") + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + def admitted(self) -> bool: + """True when a person admitted this exact (unresolved) check.""" + + return ( + self.admission is not None + and self.admission.role == "person" + and self.admission.spec_digest == self.admission_digest() + ) + def _resolve_readback( spec: Optional[ReadbackSpec], params: Mapping[str, object] @@ -310,6 +393,10 @@ def _resolve_readback( unchanged, so older contracts behave as before. A navigation whose parameter is missing keeps its ``param`` and the verifier refuses it. """ + if spec is not None and spec.admission is not None and not spec.admitted(): + # The admission was for another check. Drop it so the verifier sees an + # unadmitted check and refuses to confirm with it. + spec = spec.model_copy(deep=True, update={"admission": None}) if spec is None or ( not spec.fields and not any(nav.param for nav in spec.renavigation) ): @@ -813,6 +900,12 @@ class EffectVerdict(BaseModel): #: the read cannot certify the CURRENT state -> STALE, an INDETERMINATE #: refinement, never a pass. Additive; default False. stale: bool = False + #: How the check was done, when the verifier states it: + #: ``independent_readback``, ``screen_readback`` or ``unchecked`` + #: (:mod:`openadapt_flow.assurance`). None for verifiers that don't. + assurance: Optional[ + Literal["independent_readback", "screen_readback", "unchecked"] + ] = None @property def should_halt(self) -> bool: diff --git a/packages/flow/openadapt_flow/runtime/effects/onscreen.py b/packages/flow/openadapt_flow/runtime/effects/onscreen.py index 87db5a5fa..3d470a568 100644 --- a/packages/flow/openadapt_flow/runtime/effects/onscreen.py +++ b/packages/flow/openadapt_flow/runtime/effects/onscreen.py @@ -49,6 +49,7 @@ import difflib from typing import Any, Optional +from openadapt_flow.assurance import SCREEN_READBACK, UNCHECKED from openadapt_flow.runtime.effects.adapter import ConnectionProbe, VerifierAdapterBase from openadapt_flow.runtime.effects.effect import ( MIN_RENAV_ACTIONS, @@ -60,6 +61,7 @@ Region, Verdict, ) +from openadapt_flow.runtime.effects.readback_values import compare_value from openadapt_flow.verification import VerificationTier # Fuzzy-match floor for accepting an OCR read-back as the expected value. @@ -101,10 +103,20 @@ def _is_persisted_reacquisition(spec: Any) -> bool: if spec is None or not spec.different_path or spec.region is None: return False _, _, width, height = spec.region + reset = getattr(spec, "reset", None) + if getattr(spec, "demonstrated", False) and reset is None: + # A demonstrated check must show a state-busting step; the compiler + # refuses one without it, and a hand-edited contract earns nothing. + return False + # A reset (a reload, a new search, a reopened window) re-fetches the + # record by itself, so one action is enough when it is the reset. + enough = len(spec.renavigation) >= MIN_RENAV_ACTIONS or ( + reset is not None and 0 <= reset.at < len(spec.renavigation) + ) return ( width > 0 and height > 0 - and len(spec.renavigation) >= MIN_RENAV_ACTIONS + and enough and all(_valid_renavigation_step(nav) for nav in spec.renavigation) ) @@ -292,12 +304,20 @@ def _renavigate( # -- direct API ---------------------------------------------------------- def read_back( - self, expected_text: str, region: Optional[Region] = None + self, + expected_text: str, + region: Optional[Region] = None, + *, + value_kind: str = "text", + date_order: Optional[str] = None, ) -> EffectVerdict: """Rule on whether ``expected_text`` is on the saved-value region NOW. - - CONFIRMED — the expected value reads back (fuzzy) in the region. - - INDETERMINATE — the region has no readable text -> HALT (cannot certify). + - CONFIRMED — the expected value reads back in the region (fuzzy + for free text; normalized-exact for codes, dates and amounts). + - INDETERMINATE — the region has no readable text, or the read differs + from the expected value only in glyphs OCR confuses (``O``/``0``, + ``l``/``1``) or is a date that reads two ways -> HALT. - REFUTED — the region has readable text that is NOT the expected value -> HALT (affirmative on-screen contradiction). """ @@ -323,7 +343,31 @@ def read_back( observed_value=text or None, expected_value=expected_text, ) - if self._match(text, expected_text): + comparison = compare_value( + text, + expected_text, + kind=value_kind, # type: ignore[arg-type] + source="ocr", + date_order=date_order, + exact=False, + ) + if comparison == "ambiguous": + return EffectVerdict( + verdict=Verdict.INDETERMINATE, + kind=EffectKind.FIELD_EQUALS, + substrate=self.substrate, + reason=( + "the read differs from the expected value only in " + "characters OCR can't tell apart (O/0, l/1) or reads as " + "more than one date; cannot certify write" + ), + observed_value=text, + expected_value=expected_text, + ) + matched = comparison == "match" or ( + value_kind == "text" and self._match(text, expected_text) + ) + if matched: return EffectVerdict( verdict=Verdict.CONFIRMED, kind=EffectKind.FIELD_EQUALS, @@ -394,6 +438,21 @@ def verify( different_path = bool(spec is not None and spec.different_path) region = spec.region if spec is not None else None + # A check taught by demonstration confirms nothing until a person + # admits it. Effect.resolve drops an admission made for another check. + if spec is not None and spec.demonstrated and spec.admission is None: + return EffectVerdict( + verdict=Verdict.INDETERMINATE, + kind=expected.kind, + substrate=self.substrate, + reason=( + "the read-back was taught by demonstration and no person " + "has admitted it; cannot certify — HALT" + ), + expected_value=expected_text, + assurance=UNCHECKED, + ) + # DIFFERENT-PATH: re-open the record by an independent path BEFORE # reading. Any failure to re-navigate => INDETERMINATE (HALT), never a # CONFIRM off the write's own surface. @@ -434,11 +493,30 @@ def verify( update={ "kind": expected.kind, "reason": f"{verdict.reason} — {posture}", + "assurance": ( + SCREEN_READBACK + if verdict.verdict is Verdict.CONFIRMED + else UNCHECKED + ), } ) - def _read_structure(self, expected: str, region: Region) -> EffectVerdict: - """Read the element at ``region``'s center from the page structure.""" + def _read_structure( + self, + expected: str, + region: Region, + *, + value_kind: str = "text", + date_order: Optional[str] = None, + ) -> EffectVerdict: + """Read the element at ``region``'s center from the page structure. + + ``displayed_text_at`` reads the DOM in the browser and the + accessibility tree on macOS (AX) and Linux (AT-SPI). A pixel-only + backend (RDP, Citrix) and the Windows backend, whose UIA read returns + a whole row rather than one element, don't offer it, and the read is + INDETERMINATE; it never falls back to OCR. + """ reader = getattr(self._backend, "displayed_text_at", None) if not callable(reader): return EffectVerdict( @@ -464,7 +542,41 @@ def _read_structure(self, expected: str, region: Region) -> EffectVerdict: observed = _normalize(text) # Equality, not containment: a container element would contain the # value next to other fields and confirm by accident. - matched = observed == _normalize(expected) + comparison = compare_value( + observed, + expected, + kind=value_kind, # type: ignore[arg-type] + source="structure", + date_order=date_order, + exact=True, + ) + matched = comparison == "match" + contained = ( + not matched + and compare_value( + observed, + expected, + kind=value_kind, # type: ignore[arg-type] + source="structure", + date_order=date_order, + exact=False, + ) + == "match" + ) + if not matched and (comparison == "ambiguous" or contained): + # The element holds more than the field (an accessibility row, + # a container) or a date that reads two ways: not proof either way. + return EffectVerdict( + verdict=Verdict.INDETERMINATE, + kind=EffectKind.FIELD_EQUALS, + substrate=self.substrate, + reason=( + "the marked element shows the value among other text, or " + "a date that reads two ways" + ), + observed_value=observed, + expected_value=expected, + ) return EffectVerdict( verdict=Verdict.CONFIRMED if matched else Verdict.REFUTED, kind=EffectKind.FIELD_EQUALS, @@ -497,10 +609,16 @@ def _read_fields(self, fields: list, params: dict) -> EffectVerdict: substrate=self.substrate, reason=f"no expected value for field {field.name!r}", ) + kind = getattr(field, "value_kind", "text") + order = getattr(field, "date_order", None) if getattr(field, "source", "ocr") == "structure": - one = self._read_structure(want, tuple(field.region)) + one = self._read_structure( + want, tuple(field.region), value_kind=kind, date_order=order + ) else: - one = self.read_back(want, region=tuple(field.region)) + one = self.read_back( + want, region=tuple(field.region), value_kind=kind, date_order=order + ) expected_all.append(f"{field.name}={want}") observed.append(f"{field.name}={one.observed_value or ''}") if one.verdict is Verdict.INDETERMINATE: diff --git a/packages/flow/openadapt_flow/runtime/effects/readback_values.py b/packages/flow/openadapt_flow/runtime/effects/readback_values.py new file mode 100644 index 000000000..f70d4d7df --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/effects/readback_values.py @@ -0,0 +1,247 @@ +"""Compare a value read back off a screen with the value the run typed. + +A read-back field is read one of three ways: OCR of a screen region, the +page structure (DOM), or the native accessibility tree. This module decides, +for one field, whether the read shows the expected value. It has three +answers: + +- ``match``: the read shows the expected value; +- ``mismatch``: the read is legible and shows another value; +- ``ambiguous``: the read differs from the expected value only in characters + OCR can't tell apart (``O``/``0``, ``l``/``1``/``I``), or a date that reads + two ways. The verifier turns this into ``INDETERMINATE``, never a pass. + +Values are compared after normalizing by kind: + +- ``text``: case and runs of whitespace are ignored; +- ``code`` (an MRN, claim number, order code): case, spaces, hyphens, dots + and slashes are ignored, and nothing else; +- ``date``: both sides are parsed to a calendar date. ``03/04/2026`` is + ambiguous unless the field declares the order (``date_order``); +- ``currency``: both sides are parsed to a decimal amount, ignoring the + currency symbol, thousands separators and a trailing ``.00``. + +Glyph confusions are only considered for OCR. Structure and accessibility +reads return the real characters, so a ``0`` that should be an ``O`` there is +a mismatch. + +No model calls: string normalization, ``datetime.strptime`` and ``Decimal``. +""" + +from __future__ import annotations + +import re +from datetime import date, datetime +from decimal import Decimal, InvalidOperation +from typing import Literal, Optional + +__all__ = [ + "OCR_CONFUSABLE", + "Comparison", + "ValueKind", + "compare_value", + "infer_value_kind", + "normalize_value", +] + +ValueKind = Literal["text", "code", "date", "currency"] +Comparison = Literal["match", "mismatch", "ambiguous"] +ReadSource = Literal["ocr", "structure"] + +#: Characters OCR collapses on a pixel-only display, folded to one +#: representative. Two values equal only after this folding are ambiguous. +OCR_CONFUSABLE = str.maketrans({"o": "0", "l": "1", "i": "1", "|": "1"}) + +_CODE_SEPARATORS = re.compile(r"[\s\-./]+") +_CURRENCY = re.compile(r"^[^\d\-(]*\(?(-?)\s*([\d.,\s]+)\)?[^\d]*$") + +#: Unambiguous layouts, tried on both sides. +_DATE_FORMATS = ( + "%Y-%m-%d", + "%Y/%m/%d", + "%b %d, %Y", + "%B %d, %Y", + "%d %b %Y", + "%d %B %Y", + "%b %d %Y", + "%B %d %Y", + "%d-%b-%Y", + "%Y%m%d", +) +_DAY_MONTH = ("%d/%m/%Y", "%d-%m-%Y", "%d.%m.%Y") +_MONTH_DAY = ("%m/%d/%Y", "%m-%d-%Y", "%m.%d.%Y") + + +def _text(value: str) -> str: + return " ".join(value.lower().split()) + + +def _code(value: str) -> str: + return _CODE_SEPARATORS.sub("", value.lower()) + + +def _dates(value: str, order: Optional[str]) -> set[date]: + """Every calendar date ``value`` can mean.""" + + cleaned = " ".join(value.split()) + found: set[date] = set() + for layout in _DATE_FORMATS: + try: + found.add(datetime.strptime(cleaned, layout).date()) + except ValueError: + continue + numeric: tuple[str, ...] + if order == "dmy": + numeric = _DAY_MONTH + elif order == "mdy": + numeric = _MONTH_DAY + else: + numeric = _DAY_MONTH + _MONTH_DAY + for layout in numeric: + try: + found.add(datetime.strptime(cleaned, layout).date()) + except ValueError: + continue + return found + + +def _amount(value: str) -> Optional[Decimal]: + match = _CURRENCY.match(value.strip()) + if match is None: + return None + sign, digits = match.groups() + digits = digits.replace(" ", "") + if "," in digits and "." in digits: + digits = digits.replace(",", "") + elif digits.count(",") == 1 and len(digits.split(",")[1]) == 2: + digits = digits.replace(",", ".") # 12,50 written with a decimal comma + else: + digits = digits.replace(",", "") + try: + amount = Decimal(digits) + except InvalidOperation: + return None + negative = bool(sign) or value.strip().startswith("(") + return -amount if negative else amount + + +def normalize_value( + value: str, kind: ValueKind = "text", *, date_order: Optional[str] = None +) -> Optional[str]: + """The canonical form of ``value`` for ``kind``, or None if it doesn't parse. + + A date that reads two ways normalizes to None; use :func:`compare_value` + to tell an ambiguous date from a wrong one. + """ + + if kind == "code": + return _code(value) or None + if kind == "date": + found = _dates(value, date_order) + return found.pop().isoformat() if len(found) == 1 else None + if kind == "currency": + amount = _amount(value) + return None if amount is None else format(amount.normalize(), "f") + return _text(value) or None + + +def _folded_equal(observed: str, expected: str) -> bool: + return observed.translate(OCR_CONFUSABLE) == expected.translate(OCR_CONFUSABLE) + + +def compare_value( + observed: str, + expected: str, + *, + kind: ValueKind = "text", + source: ReadSource = "ocr", + date_order: Optional[str] = None, + exact: bool = True, +) -> Comparison: + """Judge one field read against its expected value. + + ``exact`` is for ``text`` and ``code``: True compares the whole field (a + structure read of one element); False accepts the expected value inside a + longer read (an OCR region that also shows a label such as ``MRN:``). Fuzzy OCR + matching for free text stays in + :func:`openadapt_flow.runtime.effects.onscreen.value_matches`; this + function only adds the kinds and the glyph rule. + """ + + ocr = source == "ocr" + if kind == "date": + seen = _dates(observed, date_order) + want = _dates(expected, date_order) + if not seen or not want: + if ocr and _dates(observed.translate(OCR_CONFUSABLE), date_order) & want: + return "ambiguous" + return "mismatch" + if not seen & want: + return "mismatch" + # A read that can mean the expected date or another date proves nothing. + return "match" if len(seen) == 1 and len(want) == 1 else "ambiguous" + if kind == "currency": + seen_amount = _amount(observed) + want_amount = _amount(expected) + if seen_amount is not None and want_amount is not None: + if seen_amount == want_amount: + return "match" + return "mismatch" + if ocr and want_amount is not None: + folded = _amount(observed.lower().translate(OCR_CONFUSABLE)) + if folded == want_amount: + return "ambiguous" + return "mismatch" + if kind == "code": + seen_code, want_code = _code(observed), _code(expected) + if not want_code: + return "mismatch" + if seen_code == want_code or (not exact and want_code in seen_code): + return "match" + if ocr and ( + _folded_equal(seen_code, want_code) + or ( + not exact + and want_code.translate(OCR_CONFUSABLE) + in seen_code.translate(OCR_CONFUSABLE) + ) + ): + return "ambiguous" + return "mismatch" + seen_text, want_text = _text(observed), _text(expected) + if not want_text: + return "mismatch" + if seen_text == want_text or (not exact and want_text in seen_text): + return "match" + if ocr: + folded_seen = seen_text.translate(OCR_CONFUSABLE) + folded_want = want_text.translate(OCR_CONFUSABLE) + if folded_seen == folded_want or (not exact and folded_want in folded_seen): + return "ambiguous" + return "mismatch" + + +_ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_NUMERIC_DATE = re.compile(r"^\d{1,2}[/.\-]\d{1,2}[/.\-]\d{4}$") +_MONEY = re.compile( + r"^[$€£¥]\s?-?[\d,]+(\.\d{2})?$|^-?[\d,]+\.\d{2}\s?(USD|CAD|EUR|GBP)$" +) +_CODE_SHAPE = re.compile(r"^(?=.*\d)(?=.*[A-Za-z])[A-Za-z0-9\-./]{4,}$") + + +def infer_value_kind(value: str) -> ValueKind: + """Guess the kind of a demonstrated value, conservatively. + + Only clear shapes are recognized: an ISO or numeric date, an amount with a + currency symbol or code, and a single token mixing letters and digits (a + code). Anything else is ``text``. + """ + + stripped = value.strip() + if _ISO_DATE.match(stripped) or _NUMERIC_DATE.match(stripped): + return "date" + if _MONEY.match(stripped): + return "currency" + if " " not in stripped and _CODE_SHAPE.match(stripped): + return "code" + return "text" diff --git a/packages/flow/openadapt_flow/runtime/effects/ui_readback.py b/packages/flow/openadapt_flow/runtime/effects/ui_readback.py index 32e509f97..aec1b63dd 100644 --- a/packages/flow/openadapt_flow/runtime/effects/ui_readback.py +++ b/packages/flow/openadapt_flow/runtime/effects/ui_readback.py @@ -51,6 +51,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator +from openadapt_flow.assurance import INDEPENDENT_READBACK, UNCHECKED from openadapt_flow.runtime.effects._common import judge_records from openadapt_flow.runtime.effects.adapter import ConnectionProbe, VerifierAdapterBase from openadapt_flow.runtime.effects.effect import ( @@ -1066,9 +1067,17 @@ def work(session: ReadbackSession) -> EffectVerdict: substrate=self.substrate, reason=f"{failure}; cannot certify the save; HALT", unavailable=True, + assurance=UNCHECKED, ) assert isinstance(verdict, EffectVerdict) - return verdict + # Separate session: independent of the one that typed, under the + # current policy rule (openadapt_flow.assurance). Not a + # system-of-record read. + return verdict.model_copy( + update={ + "assurance": INDEPENDENT_READBACK if verdict.confirmed else UNCHECKED + } + ) def _without_absence_claim(verdict: EffectVerdict) -> EffectVerdict: diff --git a/packages/flow/tests/test_demonstrated_readback_verifier.py b/packages/flow/tests/test_demonstrated_readback_verifier.py new file mode 100644 index 000000000..c98c1ecc5 --- /dev/null +++ b/packages/flow/tests/test_demonstrated_readback_verifier.py @@ -0,0 +1,636 @@ +"""Read-back verifiers taught by demonstration, for apps with no API. + +The person records the save, then shows how an auditor proves it persisted: +throw away client state (reload, new search, reopened window), open the +record, read the fields. These tests cover what that check must refuse and +how its verdict is graded. MockMed's encounter save on fake backends: no +browser, no model calls. +""" + +from __future__ import annotations + +import argparse +import json +from types import SimpleNamespace + +import pytest + +from openadapt_flow.assurance import ( + INDEPENDENT_READBACK, + SCREEN_READBACK, + UNCHECKED, + assurance_for, + weakest_assurance, +) +from openadapt_flow.cli_check import ( + build_marks, + cmd_check_admit, + cmd_check_show, + guided_check, + parse_reset, +) +from openadapt_flow.compiler.readback_authoring import ( + CHECK_PROMPT, + CheckError, + DeclaredReset, + MarkedField, + admit_check, + apply_check, + detect_reset, + load_check, + write_check, +) +from openadapt_flow.ir import ActionKind, EffectVerificationEvidence, Step +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectKind, + EffectState, + ReadbackField, + ReadbackNav, + ReadbackReset, + ReadbackSpec, + ValueExpr, + Verdict, +) +from openadapt_flow.runtime.effects.onscreen import OnScreenReadbackVerifier +from openadapt_flow.runtime.effects.readback_values import ( + compare_value, + infer_value_kind, + normalize_value, +) +from openadapt_flow.verification import VerificationTier +from tests.test_readback_authoring import ( + NOTE_BOX, + RUN, + SHOWS, + VISIT_BOX, + RecordBackend, + _anchor, + _marked_screen, + _recording, + _run, + _workflow, +) + +MRN_BOX = (100, 320, 120, 30) + + +# -- the four outcomes, through a real replay ------------------------------------- + + +def test_a_correct_write_is_confirmed_with_zero_model_calls(tmp_path, monkeypatch): + backend = RecordBackend() + report = _run(tmp_path, monkeypatch, backend, RUN) + assert report.success, report.halt + assert report.model_calls == 0 + (evidence,) = report.results[-1].effect_evidence + assert evidence.final_verdict == "confirmed" + assert evidence.verification_tier == int( + VerificationTier.PERSISTED_STATE_REACQUISITION + ) + assert evidence.assurance == SCREEN_READBACK + + +def test_a_wrong_value_is_refuted(tmp_path, monkeypatch): + report = _run(tmp_path, monkeypatch, RecordBackend(saved_visit="Triage"), RUN) + assert not report.success + assert any("REFUTED" in line for line in report.results[-1].effect_results) + assert report.results[-1].effect_evidence[-1].assurance == UNCHECKED + + +def test_a_missing_record_halts_as_indeterminate(tmp_path, monkeypatch): + # Nothing was saved: the re-opened record shows no value, which proves + # nothing either way, so the run halts for reconciliation. + report = _run(tmp_path, monkeypatch, RecordBackend(phantom=True), RUN) + assert not report.success + assert any("INDETERMINATE" in line for line in report.results[-1].effect_results) + + +# -- state reset ------------------------------------------------------------------ + + +def _clicks_only_workflow(): + """A check that clicks away and back: no reload, no search.""" + wf = _workflow() + steps = wf.steps[:3] + [ + Step( + id="step_003", + intent="click 'History' tab", + action=ActionKind.CLICK, + anchor=_anchor(10, 10), + ), + Step( + id="step_004", + intent="click 'Encounter' tab", + action=ActionKind.CLICK, + anchor=_anchor(50, 60), + ), + ] + return wf.model_copy(update={"steps": steps}) + + +def test_a_stale_buffer_check_without_reset_is_refused_at_compile(tmp_path): + # Two clicks used to count as re-opening the record (tier 3). Switching + # tabs in a client-side page re-renders the same buffer, so it is refused. + rec = _recording(tmp_path, n_events=5) + with pytest.raises(CheckError, match="without first throwing away"): + apply_check(_clicks_only_workflow(), rec, vision=_marked_screen(SHOWS)) + + +def test_a_new_search_is_recognized_as_the_reset(tmp_path): + updated, outcome = apply_check( + _workflow(), _recording(tmp_path), vision=_marked_screen(SHOWS) + ) + assert (outcome.reset, outcome.reset_evidence) == ("new_search", "detected") + spec = updated.steps[2].effects[0].readback + assert spec.demonstrated and spec.reset == ReadbackReset( + kind="new_search", at=1, evidence="detected" + ) + assert spec.admission is None and outcome.needs_admission + + +@pytest.mark.parametrize( + ("key", "modifiers", "kind"), + [("F5", [], "reload"), ("r", ["ctrl"], "reload"), ("w", ["meta"], "reopen_window")], +) +def test_reload_and_reopen_keys_are_resets(key, modifiers, kind): + steps = [ + Step( + id="step_010", + intent="key", + action=ActionKind.HOTKEY if modifiers else ActionKind.KEY, + key=key, + modifiers=modifiers, + ), + Step( + id="step_011", + intent="reopen", + action=ActionKind.CLICK, + anchor=_anchor(5, 5), + ), + ] + reset = detect_reset(steps) + assert reset is not None and reset.kind == kind and reset.at == 0 + + +def test_a_close_key_with_nothing_after_it_is_not_a_reset(): + steps = [Step(id="step_010", intent="close", action=ActionKind.KEY, key="Alt+F4")] + assert detect_reset(steps) is None + + +def test_a_declared_reset_is_recorded_as_declared(tmp_path): + rec = tmp_path / "rec" + _recording(tmp_path, n_events=5) + marks = load_check(rec) + write_check( + rec, + start_event=3, + fields=marks.fields, + reset=DeclaredReset(kind="reload", event=3), + ) + updated, outcome = apply_check( + _clicks_only_workflow(), rec, vision=_marked_screen(SHOWS) + ) + assert (outcome.reset, outcome.reset_evidence) == ("reload", "declared") + + +def test_a_reset_declared_outside_the_check_is_refused(tmp_path): + rec = tmp_path / "rec" + _recording(tmp_path, n_events=5) + marks = load_check(rec) + write_check( + rec, + start_event=3, + fields=marks.fields, + reset=DeclaredReset(kind="reload", event=1), + ) + with pytest.raises(CheckError, match="isn't a step of the check"): + apply_check(_clicks_only_workflow(), rec, vision=_marked_screen(SHOWS)) + + +def test_a_hand_edited_demonstrated_check_without_reset_earns_screen_tier(): + spec = _spec(reset=None) + effect = _effect(spec) + assert OnScreenReadbackVerifier.verification_tier_for(effect) is ( + VerificationTier.IMMEDIATE_SCREEN + ) + + +# -- admission -------------------------------------------------------------------- + + +class ScreenBackend: + """Shows ``text`` in every region after the re-navigation.""" + + def __init__(self, text): + self.text = text + self.calls = [] + + def screenshot(self): + return b"png" + + def click(self, x, y, *, double=False): + self.calls.append(("click", x, y)) + + def type_text(self, text): + self.calls.append(("type", text)) + + def press(self, key): + self.calls.append(("press", key)) + + +def _ocr_showing(text, confidence=0.95): + return SimpleNamespace( + ocr=lambda png, region=None: [SimpleNamespace(text=text, confidence=confidence)] + ) + + +def _spec(*, reset="default", kind="code", admitted=False, region=MRN_BOX): + spec = ReadbackSpec( + region=region, + different_path=True, + renavigation=[ + ReadbackNav(action="key", key="F5"), + ReadbackNav(action="click", point=(50, 60)), + ], + fields=[ + ReadbackField( + name="mrn", + region=region, + value=ValueExpr(param="mrn"), + value_kind=kind, + ) + ], + demonstrated=True, + reset=(ReadbackReset(kind="reload", at=0) if reset == "default" else reset), + ) + if admitted: + effect = _effect(spec) + wf = _workflow().model_copy( + update={ + "steps": [_workflow().steps[2].model_copy(update={"effects": [effect]})] + } + ) + wf, _ = admit_check(wf, admitted_by="reviewer") + spec = wf.steps[0].effects[0].readback + return spec + + +def _effect(spec): + return Effect( + kind=EffectKind.FIELD_EQUALS, + field="mrn", + value=ValueExpr(param="mrn"), + readback=spec, + ) + + +def _verify(spec, shows, params=None, confidence=0.95): + params = params or {"mrn": "MG44O8"} + effect = _effect(spec).resolve(params) + verifier = OnScreenReadbackVerifier( + ScreenBackend(shows), vision=_ocr_showing(shows, confidence) + ) + before = EffectState(substrate="onscreen", reachable=False) + return verifier.verify(effect, before, params) + + +def test_an_unadmitted_check_confirms_nothing(): + verdict = _verify(_spec(), "MG44O8") + assert verdict.verdict is Verdict.INDETERMINATE + assert "no person has admitted it" in verdict.reason + assert verdict.assurance == UNCHECKED + + +def test_an_admitted_check_confirms_as_a_screen_readback(): + spec = _spec(admitted=True) + assert spec.admitted() and spec.admission.admitted_by == "reviewer" + verdict = _verify(spec, "MG44O8") + assert verdict.verdict is Verdict.CONFIRMED, verdict.reason + assert verdict.assurance == SCREEN_READBACK + + +def test_an_admission_does_not_carry_over_to_a_changed_check(): + spec = _spec(admitted=True) + moved = spec.model_copy( + update={ + "fields": [spec.fields[0].model_copy(update={"region": (0, 0, 50, 20)})] + } + ) + assert not moved.admitted() + verdict = _verify(moved, "MG44O8") + assert verdict.verdict is Verdict.INDETERMINATE + + +def test_a_check_without_reset_cannot_be_admitted(): + effect = _effect(_spec(reset=None)) + wf = _workflow().model_copy( + update={ + "steps": [_workflow().steps[2].model_copy(update={"effects": [effect]})] + } + ) + with pytest.raises(CheckError, match="no state reset"): + admit_check(wf, admitted_by="reviewer") + + +# -- OCR ambiguity and normalization ---------------------------------------------- + + +@pytest.mark.parametrize( + ("expected", "shows"), + [("MG44O8", "MG4408"), ("AL1001", "A11001"), ("ID-10O1", "ID-1001")], +) +def test_ocr_glyph_confusion_is_indeterminate_never_a_pass(expected, shows): + verdict = _verify(_spec(admitted=True), shows, params={"mrn": expected}) + assert verdict.verdict is Verdict.INDETERMINATE + assert "can't tell apart" in verdict.reason + + +def test_a_confusable_last_glyph_in_free_text_is_indeterminate(): + # The fuzzy text match would accept 9 of 10 characters; the glyph rule + # stops it. + spec = _spec(admitted=True, kind="text") + verdict = _verify(spec, "acct 10021", params={"mrn": "acct 1002l"}) + assert verdict.verdict is Verdict.INDETERMINATE + + +def test_a_legible_different_code_is_refuted(): + verdict = _verify(_spec(admitted=True), "MG4471") + assert verdict.verdict is Verdict.REFUTED + + +def test_a_structure_read_has_no_glyph_excuse(): + assert compare_value("MG4408", "MG44O8", kind="code", source="structure") == ( + "mismatch" + ) + + +@pytest.mark.parametrize( + ("observed", "expected", "kind", "order", "result"), + [ + ("2026-03-04", "Mar 4, 2026", "date", None, "match"), + ("4 March 2026", "2026-03-04", "date", None, "match"), + ("03/04/2026", "2026-03-04", "date", None, "ambiguous"), + ("03/04/2026", "2026-03-04", "date", "mdy", "match"), + ("03/04/2026", "2026-03-04", "date", "dmy", "mismatch"), + ("2026-03-05", "2026-03-04", "date", None, "mismatch"), + ("$1,250.00", "1250", "currency", None, "match"), + ("USD 1250.50", "$1,250.5", "currency", None, "match"), + ("$1,205.00", "1250", "currency", None, "mismatch"), + ("$1,25O.00", "1250", "currency", None, "ambiguous"), + ("mg-44-08", "MG4408", "code", None, "match"), + ("MG44 08", "MG4408", "code", None, "match"), + ], +) +def test_values_are_normalized_by_kind(observed, expected, kind, order, result): + assert ( + compare_value(observed, expected, kind=kind, source="ocr", date_order=order) + == result + ) + + +def test_normalize_and_infer(): + assert normalize_value("Mar 4, 2026", "date") == "2026-03-04" + assert normalize_value("03/04/2026", "date") is None # two readings + assert normalize_value("$1,250.00", "currency") == "1250" + assert normalize_value("mg-44-08", "code") == "mg4408" + assert infer_value_kind("2026-03-04") == "date" + assert infer_value_kind("$12.50") == "currency" + assert infer_value_kind("MG44O8") == "code" + assert infer_value_kind("chest pain follow up") == "text" + + +def test_the_compiler_infers_a_code_field(tmp_path): + wf = _workflow().model_copy( + update={"params": {**_workflow().params, "visit": "ENC-2041"}} + ) + rec = _recording(tmp_path) + shows = {NOTE_BOX: SHOWS[NOTE_BOX], VISIT_BOX: "ENC-2041"} + updated, _ = apply_check(wf, rec, vision=_marked_screen(shows)) + kinds = [f.value_kind for f in updated.steps[2].effects[0].readback.fields] + assert kinds == ["text", "code"] + + +# -- non-DOM surfaces ------------------------------------------------------------- + + +class A11yBackend(ScreenBackend): + """A native backend whose accessibility tree shows ``element``.""" + + def __init__(self, element): + super().__init__("") + self.element = element + + def displayed_text_at(self, x, y): + return self.element + + +def _verify_structure(element, expected="MG44O8"): + spec = _spec(admitted=True) + spec = spec.model_copy( + update={"fields": [spec.fields[0].model_copy(update={"source": "structure"})]} + ) + # Re-admit: the field changed. + effect = _effect(spec.model_copy(update={"admission": None})) + wf = _workflow().model_copy( + update={ + "steps": [_workflow().steps[2].model_copy(update={"effects": [effect]})] + } + ) + wf, _ = admit_check(wf, admitted_by="reviewer") + params = {"mrn": expected} + resolved = wf.steps[0].effects[0].resolve(params) + verifier = OnScreenReadbackVerifier( + A11yBackend(element), vision=_ocr_showing("unused") + ) + return verifier.verify( + resolved, EffectState(substrate="onscreen", reachable=False), params + ) + + +def test_an_accessibility_read_confirms_the_exact_value(): + assert _verify_structure("MG44O8").verdict is Verdict.CONFIRMED + + +def test_an_accessibility_row_that_contains_the_value_is_indeterminate(): + # A whole row (name, DOB, MRN) is not the field. + verdict = _verify_structure("Ada Park 1980-02-01 MG44O8", expected="MG44O8") + assert verdict.verdict is Verdict.INDETERMINATE + + +def test_an_accessibility_read_of_another_value_is_refuted(): + assert _verify_structure("MG4471").verdict is Verdict.REFUTED + + +def test_native_backends_offer_displayed_text_but_windows_does_not(): + from openadapt_flow.backends.linux_backend import LinuxBackend + from openadapt_flow.backends.macos_backend import MacOSBackend + from openadapt_flow.backends.windows_backend import WindowsBackend + + assert hasattr(MacOSBackend, "displayed_text_at") + assert hasattr(LinuxBackend, "displayed_text_at") + # UIA's point read returns a row without the clicked cell, not a field. + assert not hasattr(WindowsBackend, "displayed_text_at") + + +# -- graded assurance ------------------------------------------------------------- + + +def test_assurance_grades_follow_the_tier(): + assert assurance_for(VerificationTier.INDEPENDENT_SYSTEM) == INDEPENDENT_READBACK + # Current policy: a separate signed-in session is independent of actuation. + assert assurance_for(VerificationTier.INDEPENDENT_SESSION) == INDEPENDENT_READBACK + assert ( + assurance_for(VerificationTier.PERSISTED_STATE_REACQUISITION) == SCREEN_READBACK + ) + assert assurance_for(VerificationTier.IMMEDIATE_SCREEN) == SCREEN_READBACK + assert assurance_for(1, confirmed=False) == UNCHECKED + assert assurance_for(None) == UNCHECKED + assert weakest_assurance([INDEPENDENT_READBACK, SCREEN_READBACK]) == ( + SCREEN_READBACK + ) + assert weakest_assurance([]) == UNCHECKED + + +def test_evidence_assurance_is_derived_and_not_stored(): + evidence = EffectVerificationEvidence( + effect_contract_hash="sha256:" + "0" * 64, + substrate="rest", + verification_tier=1, + initial_verdict="confirmed", + final_verdict="confirmed", + ) + assert evidence.assurance == INDEPENDENT_READBACK + assert "assurance" not in evidence.model_dump(mode="json") + + +def test_old_readback_contracts_serialize_as_before(): + spec = ReadbackSpec( + region=(1, 2, 3, 4), + different_path=True, + renavigation=[ReadbackNav(action="click", point=(1, 1))] * 2, + fields=[ + ReadbackField(name="a", region=(1, 2, 3, 4), value=ValueExpr(param="a")) + ], + ) + dumped = spec.model_dump(mode="json") + assert not {"demonstrated", "reset", "admission"} & set(dumped) + assert not {"value_kind", "date_order"} & set(dumped["fields"][0]) + + +# -- the CLI ---------------------------------------------------------------------- + + +def _bundle(tmp_path): + updated, _ = apply_check( + _workflow(), _recording(tmp_path), vision=_marked_screen(SHOWS) + ) + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + updated.save(bundle) + return bundle + + +def _admit_args(bundle, by="Richard"): + return argparse.Namespace(bundle=str(bundle), step=None, by=by) + + +def test_admit_refuses_without_a_terminal(tmp_path, capsys): + bundle = _bundle(tmp_path) + code = cmd_check_admit(_admit_args(bundle), interactive=lambda: False) + assert code == 2 + assert "interactive terminal" in capsys.readouterr().err + + +def test_admit_needs_the_typed_word(tmp_path): + from openadapt_flow.ir import Workflow + + bundle = _bundle(tmp_path) + code = cmd_check_admit( + _admit_args(bundle), ask=lambda _: "y", interactive=lambda: True + ) + assert code == 1 + assert Workflow.load(bundle).steps[2].effects[0].readback.admission is None + + +def test_a_person_admits_at_the_terminal_and_it_is_recorded(tmp_path, capsys): + from openadapt_flow.ir import Workflow + + bundle = _bundle(tmp_path) + code = cmd_check_admit( + _admit_args(bundle), ask=lambda _: "admit", interactive=lambda: True + ) + assert code == 0 + spec = Workflow.load(bundle).steps[2].effects[0].readback + assert spec.admitted() + assert spec.admission.admitted_by == "Richard" + assert spec.admission.role == "person" + out = capsys.readouterr().out + assert "state reset: new_search (detected)" in out + + cmd_check_show(argparse.Namespace(bundle=str(bundle), as_json=True)) + rows = json.loads(capsys.readouterr().out) + assert rows[0]["admitted"] is True and rows[0]["reset"] == "new_search" + + +def test_guided_check_asks_and_writes_check_json(tmp_path): + rec = tmp_path / "rec" + rec.mkdir() + (rec / "events.jsonl").write_text( + "\n".join( + json.dumps(e) + for e in [ + {"kind": "type", "text": "x"}, + {"kind": "key", "key": "Enter"}, + {"kind": "key", "key": "F5"}, + {"kind": "click", "x": 5, "y": 5}, + ] + ) + + "\n" + ) + answers = iter(["2", "note=100,200,300,40", "", ""]) + said: list[str] = [] + path = guided_check(rec, ask=lambda _: next(answers), echo=said.append) + assert path is not None and said[0] == CHECK_PROMPT + marks = load_check(rec) + assert marks.start_event == 2 and marks.prompt == CHECK_PROMPT + assert marks.fields == [MarkedField(param="note", region=(100, 200, 300, 40))] + + +def test_cli_parsers(): + marks = build_marks(["mrn=1,2,3,4"], kinds=["mrn=code"]) + assert marks[0].value_kind == "code" and marks[0].region == (1, 2, 3, 4) + assert parse_reset("fresh_session@7") == DeclaredReset( + kind="fresh_session", event=7 + ) + with pytest.raises(ValueError): + parse_reset("reload") + + +def test_check_commands_through_main(tmp_path, capsys): + from openadapt_flow.__main__ import main + + bundle = _bundle(tmp_path / "b") + # Under pytest stdin is not a terminal, so admission is refused. + assert main(["check", "admit", str(bundle)]) == 2 + rec = tmp_path / "r" + rec.mkdir() + code = main( + [ + "check", + "mark", + str(rec), + "--start-event", + "3", + "--field", + "mrn=1,2,30,10", + "--kind", + "mrn=code", + "--reset", + "fresh_session@3", + ] + ) + assert code == 0 + marks = load_check(rec) + assert marks.reset == DeclaredReset(kind="fresh_session", event=3) + assert marks.fields[0].value_kind == "code" diff --git a/packages/flow/tests/test_readback_authoring.py b/packages/flow/tests/test_readback_authoring.py index d28948337..4972d64c8 100644 --- a/packages/flow/tests/test_readback_authoring.py +++ b/packages/flow/tests/test_readback_authoring.py @@ -22,6 +22,7 @@ CHECK_PROMPT, CheckError, MarkedField, + admit_check, apply_check, load_check, write_check, @@ -171,15 +172,13 @@ def test_check_becomes_the_saves_readback_and_leaves_the_task(tmp_path): ) -def test_marking_on_the_saves_own_screen_is_only_a_screen_check(tmp_path): +def test_marking_on_the_saves_own_screen_is_refused(tmp_path): + # No step between the save and the marks: the check would read the form + # the save left behind. Before the reset rule this compiled to tier 4. wf = _workflow().model_copy(update={"steps": _workflow().steps[:3]}) rec = _recording(tmp_path, start_event=3, n_events=3) - updated, outcome = apply_check(wf, rec, vision=_marked_screen(SHOWS)) - assert not outcome.different_path and outcome.tier == "immediate_screen" - effect = updated.steps[2].effects[0] - assert OnScreenReadbackVerifier.verification_tier_for(effect) is ( - VerificationTier.IMMEDIATE_SCREEN - ) + with pytest.raises(CheckError, match="without first throwing away"): + apply_check(wf, rec, vision=_marked_screen(SHOWS)) def test_refuses_a_box_that_does_not_show_the_typed_value(tmp_path): @@ -257,6 +256,7 @@ def _run(tmp_path, monkeypatch, backend, params): updated, _ = apply_check( _workflow(), _recording(tmp_path), vision=_marked_screen(SHOWS) ) + updated, _ = admit_check(updated, admitted_by="reviewer") monkeypatch.setattr(vision_mod, "ocr", _screen_of(backend)) vision = FakeVision() vision.text_results = { @@ -405,7 +405,7 @@ def test_compile_recording_applies_the_check(tmp_path, monkeypatch): session.type_text(NOTE, param="note") session.press("Enter") session.begin_check() - session.click(10, 10) + session.press("F5") session.click(50, 60) session.mark_field( param="note", backend_pixels={"x": 100, "y": 200, "w": 300, "h": 40} @@ -453,6 +453,7 @@ def _structure_workflow(tmp_path): ) # No OCR is needed: the fields were checked against the live page. updated, _ = apply_check(_workflow(), rec, vision=_marked_screen({})) + updated, _ = admit_check(updated, admitted_by="reviewer") return updated diff --git a/packages/flow/tests/test_run_receipt.py b/packages/flow/tests/test_run_receipt.py index 2b3e777c0..e539bfc74 100644 --- a/packages/flow/tests/test_run_receipt.py +++ b/packages/flow/tests/test_run_receipt.py @@ -79,6 +79,7 @@ "rung_histogram", "evidence_classes", "effect_tier_reached", + "effect_assurance", "authorization_required", "authorization_confirmed", "identity_required", From a2342ff4b949635d094e978d1475838dbc1e7af5 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 01:34:39 -0400 Subject: [PATCH 2/6] docs: read-back taught by demonstration, state reset, admission and assurance grades Signed-off-by: Richard Abrich --- packages/flow/docs/EFFECT_KIT.md | 210 +++++++++++++++++++++++-------- packages/flow/docs/LIMITS.md | 11 ++ 2 files changed, 171 insertions(+), 50 deletions(-) diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 4470ba860..5e25a8374 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -230,56 +230,166 @@ Frappe Lending reference matrix (`benchmark/frappe_lending/fixture.py`, PR every other audited table by exactly 0. This is a harness-level companion to the verifier (it brackets a whole run, not one step). -## A read-back the person shows ("Show how to check the save") - -For an application with no API, database or export, the person can show how -they would check the save at the end of the recording. In an authoring -session, `begin_check()` returns the prompt "Show how to check the save". -The person then navigates to the saved record (recorded like any other step) -and marks each field that shows a value they typed, with `mark_field(param=..., -node_id=...)`. The marks are stored beside the recording as `check.json`, and -`compile_recording` turns them into an `onscreen` read-back on the save step -(`openadapt_flow/compiler/readback_authoring.py`): - -- the navigation leaves the task and becomes the read-back's re-navigation. A - value typed into a search box (for example the patient's name) is replayed - with this run's value; -- each marked field is a screen region, compared with the run parameter typed - into the form. Every field must read back. One unreadable field makes the - check `INDETERMINATE`, and a readable field with another value makes it - `REFUTED`. Either one halts the run; -- compiling checks the marks against the demonstration. Each marked region must - show the value typed for its parameter, or compiling fails. A check that - types a secret, marks an input that wasn't typed, or comes before any save is - refused. - -Each field is read one of two ways: - -- **From the page structure.** When the backend can read the text an element - shows (`displayed_text_at`; the browser backend reads the DOM) and, at - marking time, the marked element showed exactly the value typed for its - parameter, the field is compiled with `source: structure`. At runtime the - element at the field's center must show exactly the expected value. A - backend that can't read the structure makes the check `INDETERMINATE`; it - does not fall back to OCR. The page body is never read as a field. -- **By OCR** of the marked region, otherwise. - -Reading the structure avoids OCR misreads. It is not more independent: it is -still the same application and session, and the tier is the same. Native -accessibility reads (macOS, Linux) and template matching aren't implemented. - -How strong it is, compared with a read outside the application: - -| The check | Tier | What it can catch | What it can't | -|---|---|---|---| -| re-opens the record (at least two navigation actions) | 3, `reacquired-state` | a save only the form showed; a wrong value in a marked field | a page served from a cache the write filled; a duplicate submit; a lost update by another writer; fields nobody marked | -| fields marked on the screen the save left | 4, `screen-consistency` | a value the save's own screen shows wrong | anything the screen paints but the record never got | -| `ui_readback` (separate signed-in browser session) | 2, `independent-session` | the above, without sharing the acting session | it is still the same application, not its record store | -| REST, SQL, FHIR and other system-of-record reads | 1, `independent-system` | what the record store holds | | - -A check the person shows is never a system-of-record read. Under -`problems_only` approval, a save checked this way still goes to a person -(`onscreen_readback`). At runtime it makes no model calls. +## A read-back taught by demonstration ("Now show how you'd check it saved") + +For an application with no API, database or export (a legacy desktop app, +or one reached over Citrix or RDP), the record check has to go through the +application's own screens. The person records the save, then records how an +auditor would prove it persisted: throw away what the screen remembers, +open the saved record, read the fields. `compile` turns the second part into +an `onscreen` read-back on the save step +(`openadapt_flow/compiler/readback_authoring.py`). + +### Recording the check + +There are three ways to record it: + +- **CLI.** `openadapt-flow record --with-check ...` prints the prompt "Now + show how you'd check it saved." The person does the task, saves, and goes + on to the check in the same recording. Afterwards the CLI lists the + recorded events and asks where the check starts and which fields show the + typed values (`PARAM=X,Y,W,H`, in frame pixels). For a recording made + without the flag, `openadapt-flow check mark REC --start-event N --field + PARAM=X,Y,W,H` writes the same file. +- **Authoring session** (an agent or the Desktop app driving Flow). Calling + `begin_check()` returns the prompt. Every event after it belongs to the + check. `mark_field(param=..., node_id=...)` marks a field, and + `mark_reset(kind)` names the last step as the reset when Flow can't + recognize it. +- **By hand.** `check.json` beside the recording holds `start_event`, the + marked `fields`, an optional declared `reset`, and the `prompt` the person + saw. + +### What compiling checks + +- **A state reset before the read.** The check must contain a step that + throws away client-side state: a reload (F5, Ctrl+R or Cmd+R), a new + search (text typed, then submitted with Enter or a click), or a window + closed by key and then reopened. Flow can't recognize a click on a + browser's reload button or a fresh sign-in from the recording, so the + person names those (`--reset KIND@EVENT`, or `mark_reset`). Their kinds + are `reload`, `new_search`, `reopen_window` and `fresh_session`. The + compiled check records whether the reset was `detected` or only + `declared`. A check with no reset is refused. Without one, the check could + read the form the save left behind, or a page the browser rendered without + fetching, and a save that never reached the record would pass. Before this + rule, fields marked on the screen the save left compiled into a re-read of + that screen (tier 4). They are now refused; that weaker read is only + available as a hand-configured `effects.kind: onscreen`. +- **The marks match the demonstration.** Each marked region must show the + value typed for its parameter on the screen where it was marked. A check + that types a secret, marks a value that wasn't typed, or comes before any + save is refused. +- **The navigation leaves the task.** It becomes the read-back's + re-navigation, replayed after the save. A value typed into a search box + (for example the patient's name) is replayed with this run's value. + +### A person admits the check + +The compiled check is marked `demonstrated` and has no admission. Until a +person admits it, it can't confirm a save: the verifier rules +`INDETERMINATE` and the run halts. The demonstration that taught the check, +whether a person or an agent drove it, can't approve it. + +```bash +openadapt-flow check show BUNDLE # what each check does, and who admitted it +openadapt-flow check admit BUNDLE # a person, at a terminal, types "admit" +``` + +`check admit` prints each check (the replayed steps, the reset and whether it +was detected or declared, the fields and how each is read) and asks the +person to type `admit`. It refuses to run without an interactive terminal. +The admission records who admitted the check, when (UTC), and a SHA-256 of +the exact check (`readback.admission`). A check that changes after admission +(a moved field, another reset) no longer matches that digest. Its admission +is dropped when the run resolves the contract, so it can't confirm a save +until a person admits it again. `admit` re-seals the bundle manifest. + +### How a field is read + +- **From the page structure.** At marking time, the backend must be able to + read the text an element shows (`displayed_text_at`), and the marked element + must show exactly the value typed for its parameter. The field is then + compiled with `source: structure`, and at runtime the element must show + exactly the expected value. The browser backend reads the DOM. The macOS + (AX) and Linux (AT-SPI) backends read the accessibility element under the + point. The Windows backend doesn't offer this read: its UI Automation read + returns a whole row without the clicked cell. A backend that can't read + the structure makes the check `INDETERMINATE`; it doesn't fall back to OCR. + An element that shows the value among other text (a row, a container) + makes the check `INDETERMINATE`, not `CONFIRMED`. +- **By OCR** of the marked region, otherwise. This is the only option on + RDP and Citrix. + +Each field is compared by kind (`value_kind`). When the mark doesn't set +one, compiling infers it from the typed value: + +| Kind | Compared after | Example that matches | +|---|---|---| +| `text` | lowercasing and collapsing whitespace; OCR may also match approximately | `Chest pain` / `chest pain` | +| `code` | lowercasing and removing spaces, hyphens, dots and slashes | `MG-44-08` / `mg4408` | +| `date` | parsing to a calendar date | `Mar 4, 2026` / `2026-03-04` | +| `currency` | parsing to an amount, ignoring the symbol and thousands separators | `$1,250.00` / `1250` | + +A numeric date such as `03/04/2026` reads two ways. Unless the field sets +`date_order` (`dmy` or `mdy`), it is `INDETERMINATE`. + +OCR can't tell some characters apart on a pixel-only display: `O` and `0`, +and `l`, `I` and `1`. When an OCR read differs from the expected value only +in those characters, the field is `INDETERMINATE`, never `CONFIRMED`. A +legible read that differs in any other character is `REFUTED`. Structure and +accessibility reads return the real characters, so the same difference there +is `REFUTED`. + +### Outcomes + +| What the re-opened record shows | Verdict | Run | +|---|---|---| +| every marked field shows the expected value | `CONFIRMED` | continues | +| a legible field shows another value | `REFUTED` | halts | +| a field is blank or unreadable (for example, nothing was saved) | `INDETERMINATE` | halts; reconciliation | +| a field differs only in OCR-confusable characters, or a date reads two ways | `INDETERMINATE` | halts | +| the check isn't admitted, or its re-navigation can't be replayed | `INDETERMINATE` | halts | + +A blank field never counts as proof that nothing was saved. + +### How strong it is + +| The check | Tier | Assurance | What it can catch | What it can't | +|---|---|---|---|---| +| taught by demonstration (reset, then read) | 3, `reacquired-state` | `screen_readback` | a save only the form showed; a wrong value in a marked field | a page served from a cache the write filled; a duplicate submit; a lost update by another writer; fields nobody marked | +| hand-configured `onscreen`, same screen | 4, `screen-consistency` | `screen_readback` | a value the save's own screen shows wrong | anything the screen paints but the record never got | +| `ui_readback` (separate signed-in browser session) | 2, `independent-session` | `independent_readback` | the above, without sharing the acting session | it is still the same application, not its record store | +| REST, SQL, FHIR and other system-of-record reads | 1, `independent-system` | `independent_readback` | what the record store holds | | + +A check taught by demonstration is never a system-of-record read and never +makes a run `VERIFIED`. Under `problems_only` approval, a save checked this +way still goes to a person (`onscreen_readback`). At runtime it makes no +model calls: replayed input, OCR of fixed regions, and structure reads. + +### Read-back assurance + +Each piece of effect evidence carries one of three grades +(`openadapt_flow/assurance.py`), and the on-screen and separate-session +verifiers also state it on their verdicts: + +- `independent_readback`: an API, database, FHIR or file read, or the + record page read in a separate signed-in session; +- `screen_readback`: the acting session read the value off the screen; +- `unchecked`: no verifier confirmed the write. + +The grade comes from the verifier's tier and final verdict, so it can't +disagree with them. The success receipt records the weakest effect's grade +as `effect_assurance`, next to `effect_tier_reached`. + +Current rule: a separate-session UI read (tier 2) counts as +`independent_readback`, because it doesn't go through the session that +typed, and tier 2 can reach `VERIFIED`. It isn't a system-of-record read. +The `regulated` policy pack requires tier 1, and the Seal ladder in +`openadapt-types` puts a second session at oracle tier 1, below the +production Seal minimum of 2. These three names are the shared vocabulary +being added to `openadapt-types`; until it ships, `assurance.py` defines +them locally behind a marked adapter point. ## Separate-session UI read-back (`ui_readback`) diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index 713d64b26..f11ec49b7 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -332,6 +332,17 @@ transactional guarantee. The ultimate safety net is never the read-back: it is the **identity gate** (right record) plus **halt-on-ambiguity** (never guess a target), with read-back as additive assurance on top. +A read-back taught by demonstration +([`EFFECT_KIT.md`](EFFECT_KIT.md#a-read-back-taught-by-demonstration-now-show-how-youd-check-it-saved)) +has three more limits. It must reset client state (a reload, a new search, a +reopened window) before it reads. Flow recognizes a reload key, a closed and +reopened window, and a submitted search from the recording; any other reset +is the person's declaration, and Flow records it as declared rather than +checked. On a pixel-only display, a read that differs from the expected +value only in `O`/`0` or `l`/`1` is `INDETERMINATE`: the check halts instead +of guessing. It confirms nothing until a person admits it. Its assurance +grade is `screen_readback`, never `independent_readback`. + **Without a read that is independent of the acting session, the run cannot be `VERIFIED`.** That read is either a system-of-record read (REST, FHIR, SQL, file; tier 1) or the application's record page read in a separate signed-in From 0755a59f2e36642b86f61acf74b06c97c188f6a8 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 02:33:58 -0400 Subject: [PATCH 3/6] fix(assurance): grade separate-session UI reads as screen_readback independent_readback now means a read of the system of record that does not go through the UI (REST, SQL, FHIR, GraphQL, file). A ui_readback in a separate session goes through the same interface the executor used, so it is screen_readback on evidence and on the run receipt. Signed-off-by: Richard Abrich --- packages/flow/docs/EFFECT_KIT.md | 20 ++++++------- packages/flow/openadapt_flow/assurance.py | 30 +++++++++---------- packages/flow/openadapt_flow/receipt.py | 2 +- .../test_demonstrated_readback_verifier.py | 2 +- 4 files changed, 27 insertions(+), 27 deletions(-) diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 5e25a8374..70eda4cf2 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -359,7 +359,7 @@ A blank field never counts as proof that nothing was saved. |---|---|---|---|---| | taught by demonstration (reset, then read) | 3, `reacquired-state` | `screen_readback` | a save only the form showed; a wrong value in a marked field | a page served from a cache the write filled; a duplicate submit; a lost update by another writer; fields nobody marked | | hand-configured `onscreen`, same screen | 4, `screen-consistency` | `screen_readback` | a value the save's own screen shows wrong | anything the screen paints but the record never got | -| `ui_readback` (separate signed-in browser session) | 2, `independent-session` | `independent_readback` | the above, without sharing the acting session | it is still the same application, not its record store | +| `ui_readback` (separate signed-in browser session) | 2, `independent-session` | `screen_readback` | the above, without sharing the acting session | it is still the same application, not its record store | | REST, SQL, FHIR and other system-of-record reads | 1, `independent-system` | `independent_readback` | what the record store holds | | A check taught by demonstration is never a system-of-record read and never @@ -373,21 +373,21 @@ Each piece of effect evidence carries one of three grades (`openadapt_flow/assurance.py`), and the on-screen and separate-session verifiers also state it on their verdicts: -- `independent_readback`: an API, database, FHIR or file read, or the - record page read in a separate signed-in session; -- `screen_readback`: the acting session read the value off the screen; +- `independent_readback`: a read of the system of record that doesn't go + through the UI: an API, database, FHIR, GraphQL or file read; +- `screen_readback`: a read through the application's screens, whether in + the acting session or in a separate signed-in session; - `unchecked`: no verifier confirmed the write. The grade comes from the verifier's tier and final verdict, so it can't disagree with them. The success receipt records the weakest effect's grade as `effect_assurance`, next to `effect_tier_reached`. -Current rule: a separate-session UI read (tier 2) counts as -`independent_readback`, because it doesn't go through the session that -typed, and tier 2 can reach `VERIFIED`. It isn't a system-of-record read. -The `regulated` policy pack requires tier 1, and the Seal ladder in -`openadapt-types` puts a second session at oracle tier 1, below the -production Seal minimum of 2. These three names are the shared vocabulary +Current rule: a separate-session UI read (tier 2) is `screen_readback`. It +doesn't go through the session that typed, but it goes through the same +interface, so a UI bug or a stale view can fool both reads. A screen-only +target such as Citrix gets an honest screen-checked grade, never +`independent_readback`. These three names are the shared vocabulary being added to `openadapt-types`; until it ships, `assurance.py` defines them locally behind a marked adapter point. diff --git a/packages/flow/openadapt_flow/assurance.py b/packages/flow/openadapt_flow/assurance.py index 886ad84e2..cbb0299e3 100644 --- a/packages/flow/openadapt_flow/assurance.py +++ b/packages/flow/openadapt_flow/assurance.py @@ -6,22 +6,22 @@ read back independently of the session that typed it, read back on the same screen, or not checked at all? :func:`assurance_for` gives that answer: -- ``independent_readback``: a read that does not go through the acting - session. An API, database, FHIR, file or email read (tier 1), or the - application's record page read in a separate signed-in session - (``ui_readback``, tier 2). -- ``screen_readback``: the acting session read the value back off the screen, - after re-opening the record (tier 3) or on the screen the save left - (tier 4). +- ``independent_readback``: a read of the system of record that does not go + through the UI: an API, database, FHIR, GraphQL, file or email read + (tier 1). +- ``screen_readback``: a read through the application's own screens. That is + the record page read in a separate signed-in session (``ui_readback``, + tier 2), the acting session re-opening the record (tier 3), or the screen + the save left (tier 4). - ``unchecked``: no effect verifier confirmed the write. -Policy rule in the current code. A separate-session UI read counts as -independent here because :meth:`VerificationTier.is_independent_of_actuation` -is True for tier 2, and :data:`openadapt_flow.verification.VERIFIED_EFFECT_TIER` -lets tier 2 reach ``VERIFIED``. It is still not a system-of-record read: the -``regulated`` policy pack requires tier 1, and the Seal ladder puts a second -session at oracle tier 1, below the production Seal minimum of 2. So -``independent_readback`` does not by itself mean a production Seal. +Policy rule. A separate-session UI read goes through the same interface the +executor used, so a UI bug or a stale view can fool both. It is therefore +``screen_readback`` here, even though +:meth:`VerificationTier.is_independent_of_actuation` is True for tier 2. A +screen-only target such as Citrix gets an honest "screen-checked" grade, +never ``independent_readback``. ``independent_readback`` still does not by +itself mean a production Seal. ADAPTER POINT. The three names are the shared verification-grade vocabulary being added to ``openadapt-types``. Until that enum is published, this module @@ -77,7 +77,7 @@ def assurance_for( value = VerificationTier(int(tier)) except ValueError: return UNCHECKED - if value.is_independent_of_actuation(): + if value is VerificationTier.INDEPENDENT_SYSTEM: return INDEPENDENT_READBACK return SCREEN_READBACK diff --git a/packages/flow/openadapt_flow/receipt.py b/packages/flow/openadapt_flow/receipt.py index 0fe7bb074..4420a5c8c 100644 --- a/packages/flow/openadapt_flow/receipt.py +++ b/packages/flow/openadapt_flow/receipt.py @@ -93,7 +93,7 @@ _ASSURANCE_BY_TIER: dict[str, ReceiptAssurance] = { "independent_system": "independent_readback", - "independent_session": "independent_readback", + "independent_session": "screen_readback", "persisted_state_reacquisition": "screen_readback", } diff --git a/packages/flow/tests/test_demonstrated_readback_verifier.py b/packages/flow/tests/test_demonstrated_readback_verifier.py index c98c1ecc5..ba7bf7d16 100644 --- a/packages/flow/tests/test_demonstrated_readback_verifier.py +++ b/packages/flow/tests/test_demonstrated_readback_verifier.py @@ -479,7 +479,7 @@ def test_native_backends_offer_displayed_text_but_windows_does_not(): def test_assurance_grades_follow_the_tier(): assert assurance_for(VerificationTier.INDEPENDENT_SYSTEM) == INDEPENDENT_READBACK # Current policy: a separate signed-in session is independent of actuation. - assert assurance_for(VerificationTier.INDEPENDENT_SESSION) == INDEPENDENT_READBACK + assert assurance_for(VerificationTier.INDEPENDENT_SESSION) == SCREEN_READBACK assert ( assurance_for(VerificationTier.PERSISTED_STATE_REACQUISITION) == SCREEN_READBACK ) From b4cfe692fbaaeac7b27674c57c012c4134080c9f Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sat, 17 Oct 2026 16:07:23 -0400 Subject: [PATCH 4/6] fix(readback): grade separate-session reads as screen_readback and refuse agent admissions - ui_readback stated independent_readback on a confirmed tier-2 verdict, which disagreed with assurance_for and the receipt. It now states screen_readback, as the policy rule says. - admit_check refuses an admission in an agent, model, bot or service name, and ReadbackSpec.admitted() ignores such an admission written by hand, so the verifier rules INDETERMINATE. - The compiled check records demonstrated_by from the recording's meta.json, and describe_check prints it. - Tests for each, and a corrected comment in the grade test. - EFFECT_KIT.md no longer claims a demonstration can't approve its own check; it says which identity checks exist and what they don't prove. Signed-off-by: Richard Abrich --- packages/flow/docs/EFFECT_KIT.md | 13 ++++- .../compiler/readback_authoring.py | 33 +++++++++-- .../openadapt_flow/runtime/effects/effect.py | 58 ++++++++++++++++++- .../runtime/effects/ui_readback.py | 12 ++-- .../test_demonstrated_readback_verifier.py | 56 +++++++++++++++++- .../flow/tests/test_ui_readback_verifier.py | 14 +++++ 6 files changed, 170 insertions(+), 16 deletions(-) diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 70eda4cf2..3754def7c 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -288,8 +288,17 @@ There are three ways to record it: The compiled check is marked `demonstrated` and has no admission. Until a person admits it, it can't confirm a save: the verifier rules -`INDETERMINATE` and the run halts. The demonstration that taught the check, -whether a person or an agent drove it, can't approve it. +`INDETERMINATE` and the run halts. The check records who drove the +demonstration (`demonstrated_by`, from the recording's `meta.json`), and +`check show` and `check admit` print it. + +An admission in the name of an agent, a model or a service (`agent`, +`agent:...`, `model:...`, `bot:...`, `service:...` and similar) is refused, +and the verifier ignores such an admission if it gets into the bundle some +other way. These are identity checks, not proof that a person acted. Code +that calls `admit_check` with a person's name, or that writes the admission +and re-seals the bundle, can still admit a check. The interactive terminal +that `check admit` requires is what keeps a script from doing it by accident. ```bash openadapt-flow check show BUNDLE # what each check does, and who admitted it diff --git a/packages/flow/openadapt_flow/compiler/readback_authoring.py b/packages/flow/openadapt_flow/compiler/readback_authoring.py index 3379533cb..fdac4376f 100644 --- a/packages/flow/openadapt_flow/compiler/readback_authoring.py +++ b/packages/flow/openadapt_flow/compiler/readback_authoring.py @@ -31,10 +31,13 @@ check records whether the step was detected or only declared. A check with no reset is refused. -Admission. The compiled check is marked ``demonstrated`` and has no -admission, so it confirms nothing until a person admits it -(:func:`admit_check`, ``openadapt-flow check admit``). The demonstration, -whether a person or an agent drove it, can't approve the check it taught. +Admission. The compiled check is marked ``demonstrated``, records who drove +the demonstration (``demonstrated_by``), and has no admission, so it confirms +nothing until a person admits it (:func:`admit_check`, ``openadapt-flow check +admit``). The command needs an interactive terminal and a typed word. An +admission in an agent, model or service name is refused, and the verifier +ignores one that got into the bundle anyway. These are identity checks: code +that can call :func:`admit_check` with a person's name can admit. How strong this is. The check reads the screen of the same application in the same session that did the save (``docs/EFFECT_KIT.md``): @@ -78,6 +81,7 @@ ReadbackResetKind, ReadbackSpec, ValueExpr, + is_agent_identity, ) from openadapt_flow.runtime.effects.readback_values import ValueKind, infer_value_kind @@ -434,6 +438,7 @@ def apply_check( renavigation=renavigation, fields=read_fields, demonstrated=True, + demonstrated_by=_demonstrated_by(recording), reset=reset, ), probe=( @@ -503,11 +508,25 @@ def demonstrated_checks(workflow: Workflow) -> list[tuple[Step, Effect]]: return found +def _demonstrated_by(recording: Path) -> Optional[str]: + """``demonstrated_by`` from the recording's ``meta.json``, if it says.""" + + try: + meta = json.loads((recording / "meta.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + value = meta.get("demonstrated_by") if isinstance(meta, dict) else None + return str(value) if value else None + + def describe_check(step: Step, effect: Effect) -> str: """What a person is asked to admit, in plain words.""" spec = effect.readback assert spec is not None - lines = [f"Save step {step.id} ({step.intent}) is checked by:"] + lines = [ + f"Save step {step.id} ({step.intent}) is checked by " + f"(demonstrated by {spec.demonstrated_by or 'unknown'}):" + ] for i, nav in enumerate(spec.renavigation): if nav.action == "click": what = f"click at {nav.point}" @@ -550,6 +569,10 @@ def admit_check( who = (admitted_by or getpass.getuser() or "").strip() if not who: raise CheckError("an admission needs the name of the person admitting it") + if is_agent_identity(who): + raise CheckError( + f"{who!r} is not a person; an agent or model can't admit a check" + ) stamp = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) when = stamp.replace(microsecond=0).isoformat().replace("+00:00", "Z") admitted: list[str] = [] diff --git a/packages/flow/openadapt_flow/runtime/effects/effect.py b/packages/flow/openadapt_flow/runtime/effects/effect.py index 93edc7ac7..a388ea561 100644 --- a/packages/flow/openadapt_flow/runtime/effects/effect.py +++ b/packages/flow/openadapt_flow/runtime/effects/effect.py @@ -290,6 +290,52 @@ class ReadbackReset(BaseModel): evidence: Literal["detected", "declared"] = "detected" +#: Names that are not a person: an admission by one of them admits nothing. +_NOT_A_PERSON = frozenset( + { + "", + "agent", + "agent_and_person", + "person", + "model", + "assistant", + "cua", + "llm", + "ai", + "bot", + "system", + "service", + "auto", + "automatic", + "unknown", + } +) +_NOT_A_PERSON_PREFIXES = ( + "agent:", + "model:", + "cua:", + "llm:", + "ai:", + "bot:", + "service:", + "svc:", + "system:", +) + + +def is_agent_identity(name: object) -> bool: + """True when ``name`` is empty or names an agent, a model or a service. + + An identity check only: Flow can't prove a person typed a name, but it + can refuse the names it knows are not people. + """ + + if not isinstance(name, str): + return True + lower = name.strip().lower() + return lower in _NOT_A_PERSON or lower.startswith(_NOT_A_PERSON_PREFIXES) + + class ReadbackAdmission(BaseModel): """A person's acceptance of a read-back taught by demonstration. @@ -347,6 +393,9 @@ class ReadbackSpec(BaseModel): #: (``check.json``). Such a check confirms nothing until #: :attr:`admission` records that a person accepted it. demonstrated: bool = False + #: Who drove the demonstration, from the recording's ``meta.json``: + #: ``person``, ``agent`` or ``agent_and_person``. None when unknown. + demonstrated_by: Optional[str] = None #: The state-busting step the demonstrated check shows before it reads. reset: Optional[ReadbackReset] = None #: A person's acceptance of this exact check. @@ -361,6 +410,8 @@ def _drop_empty_fields(self, handler: Any) -> dict[str, Any]: data.pop("fields", None) if not self.demonstrated: data.pop("demonstrated", None) + if self.demonstrated_by is None: + data.pop("demonstrated_by", None) if self.reset is None: data.pop("reset", None) if self.admission is None: @@ -375,11 +426,16 @@ def admission_digest(self) -> str: return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() def admitted(self) -> bool: - """True when a person admitted this exact (unresolved) check.""" + """True when a person admitted this exact (unresolved) check. + + An admission in the name of an agent, a model or a service admits + nothing, however it got into the bundle. + """ return ( self.admission is not None and self.admission.role == "person" + and not is_agent_identity(self.admission.admitted_by) and self.admission.spec_digest == self.admission_digest() ) diff --git a/packages/flow/openadapt_flow/runtime/effects/ui_readback.py b/packages/flow/openadapt_flow/runtime/effects/ui_readback.py index aec1b63dd..ca8f903d0 100644 --- a/packages/flow/openadapt_flow/runtime/effects/ui_readback.py +++ b/packages/flow/openadapt_flow/runtime/effects/ui_readback.py @@ -51,7 +51,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator -from openadapt_flow.assurance import INDEPENDENT_READBACK, UNCHECKED +from openadapt_flow.assurance import SCREEN_READBACK, UNCHECKED from openadapt_flow.runtime.effects._common import judge_records from openadapt_flow.runtime.effects.adapter import ConnectionProbe, VerifierAdapterBase from openadapt_flow.runtime.effects.effect import ( @@ -1070,13 +1070,11 @@ def work(session: ReadbackSession) -> EffectVerdict: assurance=UNCHECKED, ) assert isinstance(verdict, EffectVerdict) - # Separate session: independent of the one that typed, under the - # current policy rule (openadapt_flow.assurance). Not a - # system-of-record read. + # A separate session reads the same application's screens, so under + # the policy rule in openadapt_flow.assurance this is a screen + # read-back, not a system-of-record read (tier 2). return verdict.model_copy( - update={ - "assurance": INDEPENDENT_READBACK if verdict.confirmed else UNCHECKED - } + update={"assurance": SCREEN_READBACK if verdict.confirmed else UNCHECKED} ) diff --git a/packages/flow/tests/test_demonstrated_readback_verifier.py b/packages/flow/tests/test_demonstrated_readback_verifier.py index ba7bf7d16..e2d4ffcef 100644 --- a/packages/flow/tests/test_demonstrated_readback_verifier.py +++ b/packages/flow/tests/test_demonstrated_readback_verifier.py @@ -322,6 +322,58 @@ def test_an_admission_does_not_carry_over_to_a_changed_check(): assert verdict.verdict is Verdict.INDETERMINATE +def _admitted_by(spec, who): + """A hand-written admission with a correct digest, in ``who``'s name.""" + from openadapt_flow.runtime.effects.effect import ReadbackAdmission + + clean = spec.model_copy(update={"admission": None}) + return clean.model_copy( + update={ + "admission": ReadbackAdmission( + admitted_by=who, + admitted_at="2026-10-17T00:00:00Z", + spec_digest=clean.admission_digest(), + ) + } + ) + + +@pytest.mark.parametrize( + "who", ["agent", "agent:anthropic", "model:x", "bot:ci", "person"] +) +def test_an_agent_cannot_admit_a_check(who): + effect = _effect(_spec()) + wf = _workflow().model_copy( + update={ + "steps": [_workflow().steps[2].model_copy(update={"effects": [effect]})] + } + ) + with pytest.raises(CheckError, match="not a person"): + admit_check(wf, admitted_by=who) + # Written into the bundle by hand, with the right digest, it admits nothing. + forged = _admitted_by(_spec(), who) + assert not forged.admitted() + verdict = _verify(forged, "MG44O8") + assert verdict.verdict is Verdict.INDETERMINATE + assert verdict.assurance == UNCHECKED + # The same hand-written admission in a person's name is the control case. + assert _verify(_admitted_by(_spec(), "reviewer"), "MG44O8").confirmed + + +def test_the_check_records_who_demonstrated_it(tmp_path): + rec = _recording(tmp_path) + meta = rec / "meta.json" + data = json.loads(meta.read_text()) if meta.is_file() else {} + meta.write_text(json.dumps({**data, "demonstrated_by": "agent"})) + updated, _ = apply_check(_workflow(), rec, vision=_marked_screen(SHOWS)) + step = updated.steps[2] + spec = step.effects[0].readback + assert spec.demonstrated_by == "agent" + from openadapt_flow.compiler.readback_authoring import describe_check + + assert "demonstrated by agent" in describe_check(step, step.effects[0]) + + def test_a_check_without_reset_cannot_be_admitted(): effect = _effect(_spec(reset=None)) wf = _workflow().model_copy( @@ -478,7 +530,9 @@ def test_native_backends_offer_displayed_text_but_windows_does_not(): def test_assurance_grades_follow_the_tier(): assert assurance_for(VerificationTier.INDEPENDENT_SYSTEM) == INDEPENDENT_READBACK - # Current policy: a separate signed-in session is independent of actuation. + # A separate signed-in session still reads the application's screens, so + # it is a screen read-back even though it is independent of actuation. + assert VerificationTier.INDEPENDENT_SESSION.is_independent_of_actuation() assert assurance_for(VerificationTier.INDEPENDENT_SESSION) == SCREEN_READBACK assert ( assurance_for(VerificationTier.PERSISTED_STATE_REACQUISITION) == SCREEN_READBACK diff --git a/packages/flow/tests/test_ui_readback_verifier.py b/packages/flow/tests/test_ui_readback_verifier.py index 8eba71b8c..16e215050 100644 --- a/packages/flow/tests/test_ui_readback_verifier.py +++ b/packages/flow/tests/test_ui_readback_verifier.py @@ -194,6 +194,20 @@ def test_confirms_the_saved_row() -> None: assert driver.credentials == [(READER_USERNAME, READER_PASSWORD)] * 2 +def test_a_separate_session_read_is_graded_screen_readback() -> None: + # Tier 2 reads the application's own record page, so it is not an + # independent (system-of-record) read-back. See openadapt_flow.assurance. + confirmed = _verifier(FakeDriver([_page(_row("7"))])).verify( + _note_equals(), EMPTY_BEFORE + ) + assert confirmed.verdict is Verdict.CONFIRMED + assert confirmed.assurance == "screen_readback" + refuted = _verifier(FakeDriver([_page(_row("7", note=""))])).verify( + _note_equals(), EMPTY_BEFORE + ) + assert refuted.assurance == "unchecked" + + def test_partial_save_is_refuted_and_labeled_as_a_record_page() -> None: verdict = _verifier(FakeDriver([_page(_row("7", note=""))])).verify( _note_equals(), EMPTY_BEFORE From 094ded69dffe490f494035c40389831de69e345e Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 17 Oct 2026 17:31:26 -0400 Subject: [PATCH 5/6] refactor(assurance): take the grade vocabulary from the published contract The assurance module kept its three grades as plain strings until the shared enum was published. In the monorepo it is published as openadapt_flow.contracts.approval.ReadBackTier with the RFC-0002 wire type VerificationTierLabel. Use that type for Assurance, point the docstring at openadapt_flow.contracts instead of openadapt_types, and test that the grades and assurance_for agree with ReadBackTier. The mapping is unchanged: only tier 1 is independent_readback. Signed-off-by: Richard Abrich --- packages/flow/openadapt_flow/assurance.py | 26 +++++++++---------- .../test_demonstrated_readback_verifier.py | 20 ++++++++++++++ 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/packages/flow/openadapt_flow/assurance.py b/packages/flow/openadapt_flow/assurance.py index cbb0299e3..b50f357cf 100644 --- a/packages/flow/openadapt_flow/assurance.py +++ b/packages/flow/openadapt_flow/assurance.py @@ -1,7 +1,7 @@ """How a write was checked, in three grades: the read-back assurance. ``VerificationTier`` (``openadapt_flow.verification``) ranks evidence in four -steps, and the Seal ladder (``openadapt_types.oracle.OracleTier``) ranks it in +steps, and the Seal ladder (``openadapt_flow.contracts.oracle.OracleTier``) ranks it in four other steps. A reviewer usually needs one plainer answer: was the save read back independently of the session that typed it, read back on the same screen, or not checked at all? :func:`assurance_for` gives that answer: @@ -23,17 +23,18 @@ never ``independent_readback``. ``independent_readback`` still does not by itself mean a production Seal. -ADAPTER POINT. The three names are the shared verification-grade vocabulary -being added to ``openadapt-types``. Until that enum is published, this module -defines them as plain strings. When it lands, replace :data:`Assurance` and -the constants below with the ``openadapt_types`` enum and keep -:func:`assurance_for` as the only mapping from Flow tiers. +The three names are the shared verification-grade vocabulary published as +:class:`openadapt_flow.contracts.approval.ReadBackTier` (RFC-0002 +``verification_tier``). :data:`Assurance` is that contract's wire type and the +constants below are its labels; :func:`assurance_for` stays the only mapping +from Flow tiers. """ from __future__ import annotations from typing import Final, Literal, Optional +from openadapt_flow.contracts.approval import VerificationTierLabel from openadapt_flow.verification import VerificationTier __all__ = [ @@ -46,18 +47,17 @@ "weakest_assurance", ] -# -- ADAPTER POINT: replace with the openadapt-types enum when it is published. -Assurance = Literal["independent_readback", "screen_readback", "unchecked"] -INDEPENDENT_READBACK: Final = "independent_readback" -SCREEN_READBACK: Final = "screen_readback" -UNCHECKED: Final = "unchecked" -#: Strongest first. +#: The RFC-0002 ``verification_tier`` wire type from the shared contract. +Assurance = VerificationTierLabel +INDEPENDENT_READBACK: Final[Literal["independent_readback"]] = "independent_readback" +SCREEN_READBACK: Final[Literal["screen_readback"]] = "screen_readback" +UNCHECKED: Final[Literal["unchecked"]] = "unchecked" +#: Strongest first, in the contract's order. ASSURANCE_GRADES: tuple[Assurance, ...] = ( INDEPENDENT_READBACK, SCREEN_READBACK, UNCHECKED, ) -# -- end ADAPTER POINT def assurance_for( diff --git a/packages/flow/tests/test_demonstrated_readback_verifier.py b/packages/flow/tests/test_demonstrated_readback_verifier.py index e2d4ffcef..487a82b1a 100644 --- a/packages/flow/tests/test_demonstrated_readback_verifier.py +++ b/packages/flow/tests/test_demonstrated_readback_verifier.py @@ -688,3 +688,23 @@ def test_check_commands_through_main(tmp_path, capsys): marks = load_check(rec) assert marks.reset == DeclaredReset(kind="fresh_session", event=3) assert marks.fields[0].value_kind == "code" + + +def test_assurance_grades_are_the_published_contract_labels() -> None: + """The grades are the shared ReadBackTier labels, strongest first.""" + + from openadapt_flow.assurance import ASSURANCE_GRADES, assurance_for + from openadapt_flow.contracts.approval import ReadBackTier + + assert ASSURANCE_GRADES == tuple(tier.label for tier in sorted(ReadBackTier)) + for flow_tier in (1, 2, 3, 4): + grade = assurance_for(flow_tier) + expected = ( + ReadBackTier.INDEPENDENT_READBACK + if flow_tier == 1 + else ReadBackTier.SCREEN_READBACK + ) + assert ReadBackTier.from_label(grade) is expected + assert ReadBackTier.from_label(assurance_for(1, confirmed=False)) is ( + ReadBackTier.UNCHECKED + ) From 069d5927562ba2d5666c9b5a2e3e5e4a7da536cc Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 17 Oct 2026 18:12:59 -0400 Subject: [PATCH 6/6] fix(check): name the check verbs through command(); keep runtime off contracts Two monorepo guards the play branch predates: - Printed next steps must name a verb through cli_hints.command(), so the launcher spelling (openadapt flow ...) is shown when that is what the person ran. record --with-check and check mark/guided check printed a literal openadapt-flow check ... hint. - openadapt_flow.interop.types must not load openadapt_flow.contracts at import time. The previous commit imported VerificationTierLabel into openadapt_flow.assurance, which the runtime imports. Spell the Literal out again and test that it equals the contract's. Signed-off-by: Richard Abrich --- packages/flow/openadapt_flow/__main__.py | 7 ++++--- packages/flow/openadapt_flow/assurance.py | 10 ++++++---- packages/flow/openadapt_flow/cli_check.py | 8 +++++--- .../flow/tests/test_demonstrated_readback_verifier.py | 6 ++++++ 4 files changed, 21 insertions(+), 10 deletions(-) diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index ab8eb735c..fde626574 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -1056,15 +1056,16 @@ def _cmd_record(args: argparse.Namespace) -> int: code = _cmd_record_capture(args) if code == 0 and with_check: from openadapt_flow.cli_check import guided_check + from openadapt_flow.cli_hints import command recording = Path(args.out) if not (recording / "events.jsonl").is_file(): print("record: no events.jsonl to mark a check in; skipped") elif not sys.stdin.isatty(): - print( - "Mark the check with 'openadapt-flow check mark " - f"{recording} --start-event N --field PARAM=X,Y,W,H'." + mark = command( + f"check mark {recording} --start-event N --field PARAM=X,Y,W,H" ) + print(f"Mark the check with '{mark}'.") else: guided_check(recording) return code diff --git a/packages/flow/openadapt_flow/assurance.py b/packages/flow/openadapt_flow/assurance.py index b50f357cf..e47693b21 100644 --- a/packages/flow/openadapt_flow/assurance.py +++ b/packages/flow/openadapt_flow/assurance.py @@ -26,7 +26,7 @@ The three names are the shared verification-grade vocabulary published as :class:`openadapt_flow.contracts.approval.ReadBackTier` (RFC-0002 ``verification_tier``). :data:`Assurance` is that contract's wire type and the -constants below are its labels; :func:`assurance_for` stays the only mapping +constants below are its labels (a test keeps them equal); :func:`assurance_for` stays the only mapping from Flow tiers. """ @@ -34,7 +34,6 @@ from typing import Final, Literal, Optional -from openadapt_flow.contracts.approval import VerificationTierLabel from openadapt_flow.verification import VerificationTier __all__ = [ @@ -47,8 +46,11 @@ "weakest_assurance", ] -#: The RFC-0002 ``verification_tier`` wire type from the shared contract. -Assurance = VerificationTierLabel +#: The RFC-0002 ``verification_tier`` wire type. It spells +#: ``openadapt_flow.contracts.approval.VerificationTierLabel`` out rather than +#: importing it, so importing Flow's runtime does not load the contracts +#: package; tests/test_demonstrated_readback_verifier.py pins the two equal. +Assurance = Literal["independent_readback", "screen_readback", "unchecked"] INDEPENDENT_READBACK: Final[Literal["independent_readback"]] = "independent_readback" SCREEN_READBACK: Final[Literal["screen_readback"]] = "screen_readback" UNCHECKED: Final[Literal["unchecked"]] = "unchecked" diff --git a/packages/flow/openadapt_flow/cli_check.py b/packages/flow/openadapt_flow/cli_check.py index 0a94608d7..79afbbd5a 100644 --- a/packages/flow/openadapt_flow/cli_check.py +++ b/packages/flow/openadapt_flow/cli_check.py @@ -22,6 +22,7 @@ from pathlib import Path from typing import Callable, Optional +from openadapt_flow.cli_hints import command from openadapt_flow.compiler.readback_authoring import ( CHECK_PROMPT, CheckError, @@ -179,7 +180,8 @@ def cmd_check_mark(args: argparse.Namespace) -> int: except (ValueError, OSError) as exc: print(f"check mark: {exc}", file=sys.stderr) return 2 - print(f"Wrote {path}. Compile the recording; a person then runs 'check admit'.") + admit = command("check admit") + print(f"Wrote {path}. Compile the recording; a person then runs '{admit}'.") return 0 @@ -299,7 +301,7 @@ def guided_check( echo(_summarize_event(index, event)) start = ask("Index of the first event of the check (empty to skip): ").strip() if not start: - echo("No check marked. Mark it later with 'openadapt-flow check mark'.") + echo(f"No check marked. Mark it later with '{command('check mark')}'.") return None fields: list[str] = [] while True: @@ -324,5 +326,5 @@ def guided_check( reset=parse_reset(reset or None), prompt=CHECK_PROMPT, ) - echo(f"Wrote {path}. After compile, a person runs 'openadapt-flow check admit'.") + echo(f"Wrote {path}. After compile, a person runs '{command('check admit')}'.") return path diff --git a/packages/flow/tests/test_demonstrated_readback_verifier.py b/packages/flow/tests/test_demonstrated_readback_verifier.py index 487a82b1a..8a4a2aaea 100644 --- a/packages/flow/tests/test_demonstrated_readback_verifier.py +++ b/packages/flow/tests/test_demonstrated_readback_verifier.py @@ -697,6 +697,12 @@ def test_assurance_grades_are_the_published_contract_labels() -> None: from openadapt_flow.contracts.approval import ReadBackTier assert ASSURANCE_GRADES == tuple(tier.label for tier in sorted(ReadBackTier)) + from typing import get_args + + from openadapt_flow.assurance import Assurance + from openadapt_flow.contracts.approval import VerificationTierLabel + + assert get_args(Assurance) == get_args(VerificationTierLabel) for flow_tier in (1, 2, 3, 4): grade = assurance_for(flow_tier) expected = (