diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 94b152ca3..4da794024 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -177,7 +177,10 @@ improvising or retrying when something is off. | Run a step as a direct API call (`ApiActuator`, `ApiBinding`) | Released, hand-authored bindings | Synthetic benchmarks | `runtime/actuators/api.py` | | Capture a recording's network traffic and compile a clean save into an API step, with the same record check | In review ([#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269)) | Synthetic patients on a local OpenEMR test copy | network capture, `ApiBinding` | | Fall back from the API step to page structure when the API is unavailable, before anything is sent | In review ([#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269)) | Synthetic | [#1269](https://github.com/OpenAdaptAI/OpenAdapt/pull/1269) | -| Run a worklist on several workers at once, one signed-in session each, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--workers` | +| Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | +| Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | +| A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | +| Record where each step's time goes in every run report | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | ## When the app changes diff --git a/packages/flow/benchmark/openemr_throughput/README.md b/packages/flow/benchmark/openemr_throughput/README.md new file mode 100644 index 000000000..e308b239f --- /dev/null +++ b/packages/flow/benchmark/openemr_throughput/README.md @@ -0,0 +1,57 @@ +# OpenEMR referral worklist runner + +Runs the OpenEMR referral program over a worklist on a local OpenEMR 8.0.0.3 +test copy, with 1 to 8 workers, either speed profile, and optional pipelined +read-back, and judges every item from the database afterwards. Every patient, +address, phone number (555-01xx) and email (`example.invalid`) is made up. +OpenAdapt isn't affiliated with the OpenEMR project. + +Speed figures are published only through the +scoreboard (`benchmark/scoreboard/`, added in [#1303](https://github.com/OpenAdaptAI/OpenAdapt/pull/1303)) (arm 4, `program_fast`, and arm 6, `parallel`). +This folder holds the runner and the profiler, not results. + +| File | What it is | +| --- | --- | +| `harness.py` | Runs one worklist on N workers against the fixture and judges every item afterwards | +| `profiler.py` | Exclusive-time profiler for one replay, deeper than the per-step `timings` in `report.json` | + +## What one run does + +1. **Setup, not timed.** Reset the database to the fixture's verified + baseline, add 8 clerk accounts cloned from the fixture's actor user, seed + the existing chart Rosa Delgado, and generate M distinct synthetic + referrals. First names have 4 or more letters (see LIMITS.md, "Short + parameter values"). +2. **The worklist.** `run_worklist_pool` over the M referrals on N workers. + Each worker opens its own browser context and signs in once as its own + clerk. For each item it claims, it closes any open patient chart (OpenEMR + keeps the last patient open after a save, which moves the form), opens a + blank New Patient form, and replays the program with the engine's own + read-back of the saved chart through the OpenEMR REST API. `--processes` + runs each worker in its own process. +3. **Judge, not timed.** The test setup reads every item back from the + database with SQL. An item is correct when exactly one new chart exists + for it with 12 of 12 fields matching; an unrun item must have no chart. + +`--halt-at I` replaces item I with a possible duplicate of the seeded chart; +the engine must halt on it and the pool must stop every worker. +`--speed fast` applies the `fast` profile (`openadapt_flow.runtime.speed`). +`--profile` (one worker) writes an exclusive-time profile per item. + +Each run writes `summary.json` (counts, wall time, sign-in time per worker, +host load, engine commit, speed settings) and `items.json` (worker, state, +judge's verdict and seconds per item) into `--out/`, for the +scoreboard's importer. + +## Run + +Needs podman, the fixture's baseline state, and the compiled referral bundle +(not in this repository). Timed runs wait until the host's 1-minute load +average is under 12. + +```bash +THROUGHPUT_UI_PORT=9911 THROUGHPUT_API_PORT=9910 \ +python -m benchmark.openemr_throughput.harness run --bundle BUNDLE \ + --fixture-state STATE --out OUT --results OUT/runs.jsonl \ + --tag n4 --workers 4 --items 32 --speed fast --processes --gate-load 12 +``` diff --git a/packages/flow/benchmark/openemr_throughput/harness.py b/packages/flow/benchmark/openemr_throughput/harness.py new file mode 100644 index 000000000..75903b856 --- /dev/null +++ b/packages/flow/benchmark/openemr_throughput/harness.py @@ -0,0 +1,517 @@ +"""Worklist throughput on a local OpenEMR test copy (synthetic data only). + + python -m benchmark.openemr_throughput.harness run \\ + --bundle BUNDLE --fixture-state STATE --tag TAG --workers N --items M \\ + [--window K] [--halt-at I] [--profile] + +What one run does: + +1. Setup, not timed: reset the fixture database to its verified baseline, + add 8 synthetic clerk accounts (``oa_clerk1`` .. ``oa_clerk8``), seed the + existing chart Rosa Delgado, and generate M distinct made-up referrals. +2. Timed: :func:`openadapt_flow.runtime.worklist_pool.run_worklist_pool` over + the M referrals on N workers. Each worker opens its own browser context + and signs in as its own clerk (counted in the time). For each item it + claims, it closes any open patient chart, opens the blank New Patient + form, and replays the compiled bundle with the engine's own read-back of + the saved chart through the OpenEMR REST API. +3. Judge, not timed: the test setup reads every item back from the database + with SQL (12 of 12 fields, exactly one new chart, none for unrun items). + +``--halt-at I`` replaces item I with a possible duplicate of the seeded +chart; the engine must halt on it and the pool must stop every worker. +``--profile`` (N = 1 only) records an exclusive-time profile per item. + +The fixture is :mod:`benchmark.openemr_local.fixture` (pinned OpenEMR 8.0.0.3 +images) with its own compose project and ports, set by environment: +``THROUGHPUT_UI_PORT`` (HTTP) and ``THROUGHPUT_API_PORT`` (HTTPS). The bundle +is the referral program compiled with DOM identity; it is not in this +repository. +""" + +from __future__ import annotations + +import argparse +import importlib.util +import json +import os +import shutil +import sys +import threading +import time +from pathlib import Path +from typing import Any + +HERE = Path(__file__).resolve().parent +FLOW = HERE.parents[1] + +#: The existing chart a possible-duplicate referral collides with. +EXISTING_ROSA = dict( + title="Mrs.", fname="Rosa", lname="Delgado", dob="1958-03-27", sex="Female", + street="311 BIRCH COURT", city="EXAMPLETON", state="MA", state_label="Massachusetts", + postal_code="02139", phone_home="555-0119", email="r.delgado.prior@example.invalid", +) +#: Same name, different date of birth: OpenEMR offers the existing chart and +#: the engine must stop (the r6 case of the referral timing benchmark). +POSSIBLE_DUPLICATE = dict( + title="Mrs.", fname="Rosa", lname="Delgado", dob="1958-03-22", sex="Female", + street="7 MILL ROAD", city="EXAMPLETON", state="MA", state_label="Massachusetts", + postal_code="02139", phone_home="555-0188", email="rosa.delgado@example.invalid", +) + +# First names are 4+ letters: the identity check refuses to bind a parameter +# value shorter than 4 characters (see docs/LIMITS.md). +FIRST = { + "Female": ["Adela", "Beatrix", "Clarice", "Dorinda", "Edwina", "Florence", "Gisela", + "Harriet", "Isadora", "Juniper", "Katrina", "Lorena", "Marisol", "Nerissa", + "Ophelia", "Perpetua"], + "Male": ["Ambrose", "Barnaby", "Casimir", "Desmond", "Emrys", "Fitzroy", "Gideon", + "Horatio", "Ignatius", "Jasper", "Leopold", "Lucian", "Montague", "Nathaniel", + "Oswin", "Peregrine"], +} +LAST = ["Quillfern", "Rowanby", "Sedgewick", "Thornvale", "Umberly", "Vantree", "Wexmoor", + "Yarrowby"] +STREETS = ["LANTERN COURT", "SYNTHETIC WAY", "EXAMPLE ROAD", "PLACEHOLDER LANE"] +CLERKS = [f"oa_clerk{i}" for i in range(1, 9)] + + +def synthetic_referrals(n: int) -> list[dict[str, str]]: + """n distinct made-up patients (555-01xx phones, example.invalid emails).""" + out = [] + for i in range(n): + sex = "Female" if i % 2 == 0 else "Male" + fname = FIRST[sex][(i // 2) % 16] + last = LAST[(i // 32) % 8] + ("" if i < 256 else str(i // 256)) + out.append(dict( + title="Ms." if sex == "Female" else "Mr.", fname=fname, + lname=f"{last}{chr(65 + (i % 26))}", + dob=f"19{50 + (i % 45):02d}-{1 + (i % 12):02d}-{1 + (i % 27):02d}", sex=sex, + street=f"{10 + i} {STREETS[i % 4]}", city="EXAMPLETON", state="MA", + state_label="Massachusetts", postal_code="02139", + phone_home=f"555-01{i % 100:02d}", + email=f"{fname.lower()}.{last.lower()}.{i}@example.invalid")) + return out + + +# -- fixture --------------------------------------------------------------- + + +def make_fixture(state: Path) -> Any: + from benchmark.openemr_local import fixture as fx + + ui_port = os.environ.get("THROUGHPUT_UI_PORT", "9911") + api_port = os.environ.get("THROUGHPUT_API_PORT", "9910") + fx.LOCK_PATH = state.parent / "environment.lock.json" + + class ThroughputFixture(fx.OpenEMRFixture): + """Same pinned images, compose file and baseline; own project and ports.""" + + @property + def ui_base_url(self) -> str: + return f"http://127.0.0.1:{ui_port}" + + @property + def api_base_url(self) -> str: + return f"https://127.0.0.1:{api_port}" + + def _validate_runtime_env(self) -> None: + v = self._runtime_values() + assert v.get("OPENEMR_HTTP_PORT") == ui_port + assert v.get("OPENEMR_HTTPS_PORT") == api_port + assert v.get("OPENEMR_IMAGE") == self.lock["services"]["openemr"] + assert v.get("MARIADB_IMAGE") == self.lock["services"]["mariadb"] + self.runtime_env.chmod(0o600) + + return ThroughputFixture(state) + + +def add_clerks(fixture: Any, n: int = 8) -> list[str]: + """Clone the fixture's actor user into n clerks (own username and ACL row).""" + stmts = ["USE openemr"] + for i, name in enumerate(CLERKS[:n], 1): + aro = 100 + i + stmts += [ + "DROP TEMPORARY TABLE IF EXISTS t_u", + "CREATE TEMPORARY TABLE t_u SELECT * FROM users WHERE username='openadapt_actor'", + f"UPDATE t_u SET id=0, uuid=NULL, username='{name}', fname='Clerk', lname='{i}'", + "INSERT INTO users SELECT * FROM t_u", + f"SET @uid=(SELECT id FROM users WHERE username='{name}')", + "INSERT INTO users_secure (id,username,password,last_update_password) " + f"SELECT @uid,'{name}',password,last_update_password FROM users_secure " + "WHERE username='openadapt_actor'", + "INSERT INTO gacl_aro (id,section_value,value,order_value,name,hidden) " + f"VALUES ({aro},'users','{name}',10,'Clerk {i}',0)", + f"INSERT INTO gacl_groups_aro_map (group_id,aro_id) VALUES (11,{aro})", + f"INSERT INTO `groups` (name,user) VALUES ('Default','{name}')", + ] + stmts.append(f"UPDATE gacl_aro_seq SET id={100 + n}") + fixture._db(";\n".join(stmts) + ";") + count = fixture._db( + "SELECT COUNT(*) FROM openemr.users WHERE username LIKE 'oa_clerk%'" + ).decode().strip() + if count != str(n): + raise RuntimeError(f"expected {n} clerks, found {count}") + return CLERKS[:n] + + +SQL_FIELDS = ("id", "title", "fname", "lname", "DOB", "sex", "street", "city", "state", + "postal_code", "phone_home", "email", "country_code") + + +def sql_patients(fixture: Any) -> list[dict[str, str]]: + out = fixture._db( + "SELECT " + ",".join(SQL_FIELDS) + " FROM openemr.patient_data ORDER BY id" + ).decode() + return [ + {k: ("" if v == "NULL" else v) for k, v in zip(SQL_FIELDS, line.split("\t"))} + for line in out.splitlines() + ] + + +def container_started(fixture: Any) -> str: + import subprocess + + project = fixture.lock["project"] + names = [f"{project}-openemr-1", f"{project}-db-1"] + r = subprocess.run( + ["podman", "inspect", "-f", "{{.Name}}={{.State.StartedAt}}/{{.State.Running}}", *names], + capture_output=True, text=True, + ) + return r.stdout.strip().replace("\n", "|") + + +# -- load ------------------------------------------------------------------ + + +def wait_quiet(threshold: float, max_wait_s: float, sustain: int = 4) -> dict[str, Any]: + """Start only once the 1-minute load average stays below ``threshold``.""" + t0 = time.time() + ok = 0 + while True: + one = os.getloadavg()[0] + ok = ok + 1 if one < threshold else 0 + if ok >= sustain or time.time() - t0 > max_wait_s: + return {"threshold": threshold, "waited_s": round(time.time() - t0, 1), + "gave_up": ok < sustain, "load_at_start": os.getloadavg()} + time.sleep(5.0) + + +class LoadSampler: + def __init__(self, every_s: float = 2.0) -> None: + self.samples: list[float] = [] + self._stop = threading.Event() + self._t = threading.Thread(target=self._loop, args=(every_s,), daemon=True) + self._t.start() + + def _loop(self, every_s: float) -> None: + while not self._stop.is_set(): + self.samples.append(os.getloadavg()[0]) + self._stop.wait(every_s) + + def stop(self) -> dict[str, Any]: + self._stop.set() + self._t.join(timeout=5) + s = self.samples + return {"load1_mean": round(sum(s) / len(s), 2) if s else None, + "load1_max": round(max(s), 2) if s else None, "ncpu": os.cpu_count()} + + +# -- the run --------------------------------------------------------------- + + +def _demo_module() -> Any: + spec = importlib.util.spec_from_file_location( + "openemr_local_demo", FLOW / "scripts" / "openemr_local_demo.py" + ) + demo = importlib.util.module_from_spec(spec) + sys.modules["openemr_local_demo"] = demo + spec.loader.exec_module(demo) + return demo + + +def build_worker(user: str, opts: dict[str, Any]) -> Any: + """One worker signed in as ``user``. Module-level so a child process can + build it (``ProcessWorker``); ``opts`` holds only plain values.""" + from openadapt_flow.benchmark.openemr_local import ( + OpenEMRPatientOracle, + SyntheticPatientSpec, + ) + from openadapt_flow.cli_worklist_pool import WebPoolWorker + from openadapt_flow.ir import Workflow + from openadapt_flow.runtime import Replayer + + from openadapt_flow.runtime.speed import apply_speed_profile + + os.environ.setdefault("OPENADAPT_FLOW_BROWSER", "bundled") + apply_speed_profile(opts.get("speed", "standard")) + os.environ["THROUGHPUT_UI_PORT"] = opts["ui_port"] + os.environ["THROUGHPUT_API_PORT"] = opts["api_port"] + if opts.get("ocr_threads") is not None: + os.environ["OPENADAPT_FLOW_OCR_THREADS"] = str(opts["ocr_threads"]) + if opts.get("page_quiet_ms") is not None: + from openadapt_flow.vision import settle as settle_mod + + settle_mod.PAGE_QUIET_WINDOW_S = opts["page_quiet_ms"] / 1000.0 + demo = sys.modules.get("openemr_local_demo") or _demo_module() + fixture = make_fixture(Path(opts["fixture_state"])) + bundle = Path(opts["bundle"]) + workflow = Workflow.load(bundle) + values = fixture._runtime_values() + + class _Clerk: + ui_base_url = fixture.ui_base_url + + def _runtime_values(self) -> dict[str, str]: + return {**values, "OPENEMR_ACTOR_USER": user} + + class _PageOnly: + def __init__(self, page: Any) -> None: + self.page = page + + def open_form(page: Any) -> None: + """Start each item where the recording started: no chart open, blank form.""" + if page.evaluate( + "() => !!(window.app_view_model && app_view_model.application_data.patient())" + ): + # The chart's own close button (OpenEMR clearPatient), no finder. + with page.expect_response(lambda r: "unset_session_ajax.php" in r.url, timeout=30_000): + page.evaluate("() => clearPatient(false)") + page.wait_for_function("() => !app_view_model.application_data.patient()", timeout=30_000) + frame = page.frame(name=demo.FORM_FRAME_NAME) + if frame is not None and demo.FORM_PATH in frame.url: + try: + if frame.locator("#form_fname").input_value(timeout=2000) == "": + return # sign-in left the blank form open + except Exception: # noqa: BLE001 - fall through and reload it + pass + page.evaluate( + "({path, name}) => { window.navigateTab(path, name); window.activateTabByName(name, true); }", + {"path": demo.FORM_PATH, "name": demo.FORM_FRAME_NAME}, + ) + frame = demo._form_frame(page) + frame.locator("#form_fname").wait_for(state="visible", timeout=60_000) + if frame.locator("#form_fname").input_value() != "": + raise RuntimeError("the New Patient form was not blank") + + session: list[Any] = [] + + def run_item(page: Any, idx: int, row: dict[str, Any], item_dir: Path, pipeline: Any) -> Any: + load_start = round(os.getloadavg()[0], 2) + spec = SyntheticPatientSpec(**row) + if not session: + session.append(fixture.token_session("oracle")) + oracle = OpenEMRPatientOracle(fixture.api_base_url, session[0], spec) + backend = demo._BasePlaywrightBackend(page) + prof = None + if opts.get("profile"): + from benchmark.openemr_throughput.profiler import RunProfiler + + prof = RunProfiler() + prof.install(backend) + try: + with (prof.span("task") if prof else _null()): + rep = Replayer( + backend, effect_verifier=oracle.verifier, readback_pipeline=pipeline + ).run(workflow, params=spec.params(), bundle_dir=bundle, run_dir=item_dir) + finally: + if prof: + prof.uninstall() + if prof: + (item_dir / "profile.json").write_text(json.dumps( + {"buckets": prof.buckets(), "calls": prof.summary(), + "parents": prof.parents(), "steps": prof.step_buckets()}, indent=1)) + if not opts.get("keep_pngs"): + for png in (item_dir / "steps").glob("*.png"): + png.unlink() + # Host load at both ends of the item: the scoreboard times a row only + # when both are at or under its limit. + (item_dir / "host_load.json").write_text(json.dumps( + {"start": load_start, "end": round(os.getloadavg()[0], 2)})) + return rep + + worker = WebPoolWorker( + name=user, entry_url=fixture.ui_base_url, run_item_on_page=run_item, + sign_in=lambda page: demo._preauthenticate_browser(_PageOnly(page), _Clerk()), + open_entry=open_form, readback_window=opts.get("window", 0), + ) + importlib.import_module("openadapt_flow.vision.ocr")._get_engine() + return worker + + +def run(a: argparse.Namespace) -> dict[str, Any]: + from openadapt_flow._browser_setup import ensure_chromium_installed + from openadapt_flow.benchmark.openemr_local import ( + SyntheticPatientSpec, + ) + from openadapt_flow.ir import Workflow + from openadapt_flow.runtime.worklist_pool import PoolPolicy, run_worklist_pool + + if a.profile and a.workers != 1: + raise SystemExit("--profile needs --workers 1") + from openadapt_flow.runtime.speed import apply_speed_profile, current_speed_settings + + apply_speed_profile(a.speed) + if a.ocr_threads is not None: + os.environ["OPENADAPT_FLOW_OCR_THREADS"] = str(a.ocr_threads) + if a.page_quiet_ms is not None: + from openadapt_flow.vision import settle as settle_mod + + settle_mod.PAGE_QUIET_WINDOW_S = a.page_quiet_ms / 1000.0 + os.environ.setdefault("OPENADAPT_FLOW_BROWSER", "bundled") + _demo_module() # registers the module for build_worker in this process + out = Path(a.out) / a.tag + out.mkdir(parents=True, exist_ok=False) + fixture = make_fixture(Path(a.fixture_state)) + t = time.time() + digest = fixture.reset() + clerks = add_clerks(fixture) + seed = fixture.token_session("actor").post( + fixture.api_base_url + "/apis/default/api/patient", + json=dict(SyntheticPatientSpec(**EXISTING_ROSA).fields), + ).status_code + rows = synthetic_referrals(a.items) + if a.halt_at is not None: + rows[a.halt_at] = dict(POSSIBLE_DUPLICATE) + before_ids = {r["id"] for r in sql_patients(fixture)} + bundle = Path(a.bundle) + workflow = Workflow.load(bundle) + engine_sha = os.popen(f"git -C {FLOW} rev-parse HEAD").read().strip() + meta = { + "tag": a.tag, "workers": a.workers, "items": a.items, "window": a.window, + "halt_at": a.halt_at, "profile": a.profile, "engine_sha": engine_sha, + **current_speed_settings(), + "engine_dirty": bool(os.popen(f"git -C {FLOW} status --porcelain -- openadapt_flow").read().strip()), + "bundle_digest": getattr(workflow.manifest, "content_digest", None), + "baseline": digest[:12], "seed_status": seed, "setup_s": round(time.time() - t, 2), + "started": time.strftime("%Y-%m-%d %H:%M:%S"), + } + ensure_chromium_installed() + opts = { + "fixture_state": str(a.fixture_state), "bundle": str(a.bundle), + "ui_port": os.environ.get("THROUGHPUT_UI_PORT", "9911"), + "api_port": os.environ.get("THROUGHPUT_API_PORT", "9910"), + "window": a.window, "profile": a.profile, "keep_pngs": a.keep_pngs, + "page_quiet_ms": a.page_quiet_ms, "ocr_threads": a.ocr_threads, + "speed": a.speed, + } + if a.processes: + from openadapt_flow.runtime.worklist_pool import ProcessWorker + + workers = [ + ProcessWorker(user, "benchmark.openemr_throughput.harness:build_worker", + {"user": user, "opts": opts}) + for user in clerks[: a.workers] + ] + else: + # Load the OCR model before the clock, as a warm worker would. + importlib.import_module("openadapt_flow.vision.ocr")._get_engine() + workers = [build_worker(user, opts) for user in clerks[: a.workers]] + meta["processes"] = bool(a.processes) + policy = PoolPolicy(workers=a.workers, readback_window=a.window, + items_independent=True, max_workers=8) + free_gb = shutil.disk_usage(str(out)).free / 1e9 + if free_gb < 8: + raise SystemExit(f"disk below 8 GB ({free_gb:.1f} GB free)") + gate = wait_quiet(a.gate_load, a.gate_max_wait_s) + sampler = LoadSampler() + c0 = container_started(fixture) + t0 = time.time() + report = run_worklist_pool( + rows, workers, run_dir=out / "pool", policy=policy, + on_event=lambda m: print(time.strftime("%H:%M:%S"), m, flush=True), + ) + wall = time.time() - t0 + load = sampler.stop() + c1 = container_started(fixture) + + # Judge (not timed): the test setup's own SQL read-back of every item. + new_rows = [r for r in sql_patients(fixture) if r["id"] not in before_ids] + items = [] + for item in report.items: + spec = SyntheticPatientSpec(**rows[item["idx"]]) + mine = [r for r in new_rows if r["lname"] == spec.lname and r["email"] == spec.email] + fields = dict(spec.fields) + match = sum(1 for k, v in fields.items() if len(mine) == 1 and mine[0].get(k) == v) + duplicate_case = a.halt_at is not None and item["idx"] == a.halt_at + res = item.get("result") or {} + items.append({ + "idx": item["idx"], "state": item["state"], "worker": item["worker"], + "new_charts": len(mine), "readback": f"{match}/{len(fields)}", + "correct": (len(mine) == 0) if duplicate_case else (len(mine) == 1 and match == len(fields)), + "duplicate_case": duplicate_case, "wall_s": res.get("wall_s"), + "readback_mode": res.get("readback"), + "claimed_at": item["claimed_at"], "finished_at": item["finished_at"], + }) + ran = [i for i in items if i["state"] != "pending"] + saved = [i for i in ran if i["correct"] and i["new_charts"] == 1] + summary = { + **meta, "wall_s": round(wall, 3), "pool_wall_s": round(report.wall_s, 3), + "sign_in_s": {w.name: round(w.open_s, 2) for w in report.per_worker}, + "counts": report.counts(), "stop": report.stop, "items_run": len(ran), + "saved_correct": len(saved), + # A wrong write: a chart that doesn't match its item, a second chart, + # or a chart for an item that never ran. A halt that saved nothing is + # not one of these; it is in not_saved. + "wrong_or_extra_charts": sum( + 1 for i in items + if (i["new_charts"] >= 1 and not i["correct"]) or i["new_charts"] > 1 + ), + "not_saved": sum(1 for i in ran if i["new_charts"] == 0 and not i["duplicate_case"]), + "new_charts_total": len(new_rows), + "entries_per_hour": round(len(saved) * 3600 / wall, 1), + "written_after_failed_readback": report.written_after_failed_readback, + "per_worker": [{"name": w.name, "items": len(w.items), "states": w.states, + "busy_s": round(w.busy_s, 1)} for w in report.per_worker], + **load, "gate": gate, "valid": c0 == c1, "ended": time.strftime("%Y-%m-%d %H:%M:%S"), + } + (out / "summary.json").write_text(json.dumps(summary, indent=1, default=str)) + (out / "items.json").write_text(json.dumps(items, indent=1)) + return summary + + +class _null: + def __enter__(self) -> None: + return None + + def __exit__(self, *exc: object) -> None: + return None + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + sub = ap.add_subparsers(dest="cmd", required=True) + r = sub.add_parser("run") + r.add_argument("--bundle", required=True) + r.add_argument("--fixture-state", required=True) + r.add_argument("--out", required=True, help="evidence root; each run gets OUT/TAG") + r.add_argument("--tag", required=True) + r.add_argument("--workers", type=int, required=True) + r.add_argument("--items", type=int, required=True) + r.add_argument("--window", type=int, default=0) + r.add_argument("--halt-at", type=int, default=None) + r.add_argument("--profile", action="store_true") + r.add_argument("--speed", choices=["standard", "fast"], default="standard", + help="named speed profile (openadapt_flow.runtime.speed)") + r.add_argument("--ocr-threads", type=int, default=None, + help="OCR engine threads (OPENADAPT_FLOW_OCR_THREADS); default all cores") + r.add_argument("--page-quiet-ms", type=int, default=None, + help="page-quiet settle window (runtime.page_quiet_ms); default 100") + r.add_argument("--keep-pngs", action="store_true") + r.add_argument("--processes", action="store_true", + help="run each worker in its own process (ProcessWorker)") + r.add_argument("--gate-load", type=float, default=12.0) + r.add_argument("--gate-max-wait-s", type=float, default=7200.0) + r.add_argument("--results", required=True, + help="JSONL file to append the run summary to (evidence, not committed)") + a = ap.parse_args() + summary = run(a) + Path(a.results).parent.mkdir(parents=True, exist_ok=True) + with open(a.results, "a") as fh: + fh.write(json.dumps({k: v for k, v in summary.items()}, default=str) + "\n") + print(json.dumps({k: summary[k] for k in ( + "tag", "workers", "window", "wall_s", "items_run", "saved_correct", + "entries_per_hour", "counts", "stop", "load1_mean", "valid")}, default=str)) + + +if __name__ == "__main__": + main() diff --git a/packages/flow/benchmark/openemr_throughput/profiler.py b/packages/flow/benchmark/openemr_throughput/profiler.py new file mode 100644 index 000000000..dd412e81d --- /dev/null +++ b/packages/flow/benchmark/openemr_throughput/profiler.py @@ -0,0 +1,349 @@ +"""Exclusive-time profiler for one replay run. + +Wraps a fixed list of Replayer, backend and vision functions. Each wrapped +call's time is split into its own (exclusive) time and the time spent in +other wrapped calls it made, so the buckets add up to the wall time of the +run with no double counting. Time not inside any wrapped call is engine +bookkeeping ("engine other"). + +The profiler only measures. It changes no behavior: every wrapper calls the +original function with the same arguments and returns its result. + +Usage:: + + prof = RunProfiler() + prof.install(backend) # wraps this backend instance + classes + with prof.span("task"): + report = replayer.run(...) + prof.uninstall() + prof.summary() # {name: {"n":..., "excl_s":..., "incl_s":...}} +""" + +from __future__ import annotations + +import functools +import inspect +import threading +import time +from collections import defaultdict +from contextlib import contextmanager +from typing import Any, Callable, Iterator + +# Replayer methods worth naming. Each is wrapped on the class. +REPLAYER_METHODS = ( + "_run_step", + "_resolve_step", + "_apply_step_gates", + "_wait_starting_state_settled", + "_verify_identity", + "_verify_identity_ocr", + "_verify_dedicated_identity_signal", + "_surface_identity_check", + "_act", + "_wait_for_dialog", + "_handle_interstitials", + "_verify_typed_input", + "_verify_selected_option", + "_check_postconditions", + "_poll_postconditions", + "_verify_effects", + "_settle_seeded", + "_save_step_png", + "_wait_for_scroll_transition_and_settle", + "_revalidate_consequential_actuation", + "_finalize_report", +) + +# Backend methods. Wrapped on the instance so other backends are untouched. +BACKEND_METHODS = ( + "screenshot", + "page_quiescent", + "page_change_marker", + "click", + "right_click", + "type_text", + "type_text_guarded", + "press", + "press_guarded", + "scroll", + "locate_structural", + "act_structural", + "act_guarded_coordinate", + "structured_text_at", + "text_value_at", + "displayed_text_at", + "focused_text_value", + "focused_field_label", + "focused_select_label_at", + "visible_text_present", + "take_dialogs", + "arm_dialog", + "structural_locator_at", + "submit_binding_at", + "workflow_state_identity", + "session_identity", + "application_identity", +) + +# (module, function) pairs in the vision layer. +VISION_FUNCS = ( + ("openadapt_flow.vision.settle", "wait_settled"), + ("openadapt_flow.vision.settle", "wait_settled_result"), + ("openadapt_flow.vision.settle", "wait_settled_seeded"), + ("openadapt_flow.vision.ocr", "ocr"), + ("openadapt_flow.vision.match", "find_template"), + ("openadapt_flow.vision.match", "find_text"), + ("openadapt_flow.vision.match", "find_text_candidates"), + ("openadapt_flow.vision.hashing", "phash_png"), +) + +# Report buckets: which named calls count toward each. +BUCKETS: dict[str, tuple[str, ...]] = { + "waits: fixed sleeps (poll intervals)": ("time.sleep",), + "waits: page signals (DOM/network quiet)": ( + "be.page_quiescent", + "be.page_change_marker", + ), + "waits: screen settle (frame compare)": ( + "vision.wait_settled", + "vision.wait_settled_result", + "vision.wait_settled_seeded", + "rp._settle_seeded", + "rp._wait_starting_state_settled", + "rp._wait_for_scroll_transition_and_settle", + "vision.phash_png", + ), + "screenshots (capture)": ("be.screenshot",), + "step evidence PNGs (write)": ("rp._save_step_png",), + "identity: structure (DOM)": ( + "rp._verify_identity", + "rp._verify_dedicated_identity_signal", + "rp._surface_identity_check", + "be.structured_text_at", + "be.workflow_state_identity", + "be.session_identity", + "be.application_identity", + ), + "identity: OCR": ("rp._verify_identity_ocr", "vision.ocr@identity"), + "OCR (other)": ("vision.ocr",), + "target resolution": ( + "rp._resolve_step", + "be.locate_structural", + "be.structural_locator_at", + "be.submit_binding_at", + "vision.find_template", + "vision.find_text", + "vision.find_text_candidates", + ), + "typing and keys": ( + "be.type_text", + "be.type_text_guarded", + "be.press", + "be.press_guarded", + ), + "clicks and scrolls": ( + "be.click", + "be.right_click", + "be.act_structural", + "be.act_guarded_coordinate", + "be.scroll", + ), + "input read-back (field value)": ( + "rp._verify_typed_input", + "rp._verify_selected_option", + "be.text_value_at", + "be.displayed_text_at", + "be.focused_text_value", + "be.focused_field_label", + "be.focused_select_label_at", + ), + "postconditions and dialogs": ( + "rp._check_postconditions", + "rp._poll_postconditions", + "be.visible_text_present", + "rp._wait_for_dialog", + "be.take_dialogs", + "be.arm_dialog", + "rp._handle_interstitials", + ), + "save read-back (system of record)": ("rp._verify_effects",), + "engine other": ( + "rp._run_step", + "rp._act", + "rp._apply_step_gates", + "rp._revalidate_consequential_actuation", + "rp._finalize_report", + "span.task", + ), +} + + +class RunProfiler: + """Stack-based exclusive-time profiler (main thread only).""" + + def __init__(self) -> None: + self._thread = threading.get_ident() + self._stack: list[list[Any]] = [] # [name, start, child_time] + self.stats: dict[str, dict[str, float]] = defaultdict( + lambda: {"n": 0, "excl_s": 0.0, "incl_s": 0.0} + ) + self.steps: dict[str, dict[str, float]] = defaultdict( + lambda: defaultdict(float) + ) + self.by_parent: dict[str, dict[str, float]] = defaultdict( + lambda: defaultdict(float) + ) + self._current_step: str | None = None + self._undo: list[Callable[[], None]] = [] + + # -- core ------------------------------------------------------------- + + def _enter(self, name: str) -> None: + self._stack.append([name, time.perf_counter(), 0.0]) + + def _exit(self) -> None: + name, start, child = self._stack.pop() + incl = time.perf_counter() - start + excl = incl - child + if name == "vision.ocr" and any( + f[0] in ("rp._verify_identity_ocr", "rp._verify_identity") + for f in self._stack + ): + name = "vision.ocr@identity" + parent = next( + (f[0] for f in reversed(self._stack) if f[0].startswith("rp.")), "-" + ) + self.by_parent[name][parent] += excl + st = self.stats[name] + st["n"] += 1 + st["excl_s"] += excl + st["incl_s"] += incl + if self._current_step: + self.steps[self._current_step][name] += excl + if self._stack: + self._stack[-1][2] += incl + + def _wrap( + self, name: str, fn: Callable[..., Any], *, static: bool = False + ) -> Callable[..., Any]: + prof = self + + @functools.wraps(fn) + def wrapper(*a: Any, **k: Any) -> Any: + if threading.get_ident() != prof._thread: + return fn(*a, **k) + prev_step = prof._current_step + if name == "rp._run_step": + step = ( + a[0 if static else 1] + if len(a) > (0 if static else 1) + else k.get("step") + ) + prof._current_step = getattr(step, "id", None) or prev_step + prof._enter(name) + try: + return fn(*a, **k) + finally: + prof._exit() + if name == "rp._run_step": + prof._current_step = prev_step + + return wrapper + + @contextmanager + def span(self, name: str) -> Iterator[None]: + self._enter(f"span.{name}") + try: + yield + finally: + self._exit() + + # -- install ---------------------------------------------------------- + + def install(self, backend: Any) -> None: + import importlib + + from openadapt_flow.runtime.replayer import Replayer + + for m in REPLAYER_METHODS: + raw = inspect.getattr_static(Replayer, m, None) + if raw is None: + continue + if isinstance(raw, staticmethod): + wrapped: Any = staticmethod( + self._wrap(f"rp.{m}", raw.__func__, static=True) + ) + else: + wrapped = self._wrap(f"rp.{m}", raw) + setattr(Replayer, m, wrapped) + self._undo.append(lambda m=m, raw=raw: setattr(Replayer, m, raw)) + for m in BACKEND_METHODS: + orig = getattr(backend, m, None) + if orig is None or not callable(orig): + continue + setattr(backend, m, self._wrap(f"be.{m}", orig)) + self._undo.append(lambda m=m: backend.__dict__.pop(m, None)) + for mod_name, fn_name in VISION_FUNCS: + mod = importlib.import_module(mod_name) + orig = getattr(mod, fn_name, None) + if orig is None: + continue + wrapped = self._wrap(f"vision.{fn_name}", orig) + setattr(mod, fn_name, wrapped) + self._undo.append( + lambda mod=mod, fn_name=fn_name, orig=orig: setattr(mod, fn_name, orig) + ) + # Modules that imported the function by name keep their own + # reference; patch those too. + for other in ("openadapt_flow.runtime.replayer", "openadapt_flow.vision"): + om = importlib.import_module(other) + if getattr(om, fn_name, None) is orig: + setattr(om, fn_name, wrapped) + self._undo.append( + lambda om=om, fn_name=fn_name, orig=orig: setattr( + om, fn_name, orig + ) + ) + orig_sleep = time.sleep + time.sleep = self._wrap("time.sleep", orig_sleep) + self._undo.append(lambda: setattr(time, "sleep", orig_sleep)) + + def uninstall(self) -> None: + while self._undo: + self._undo.pop()() + + # -- output ----------------------------------------------------------- + + def summary(self) -> dict[str, dict[str, float]]: + return { + k: {kk: round(vv, 4) for kk, vv in v.items()} for k, v in self.stats.items() + } + + def parents(self) -> dict[str, dict[str, float]]: + """Exclusive time of each call, split by the nearest Replayer caller.""" + return { + n: {p: round(v, 4) for p, v in sorted(per.items(), key=lambda x: -x[1])} + for n, per in self.by_parent.items() + } + + def buckets(self) -> dict[str, float]: + out: dict[str, float] = {} + named = set() + for bucket, names in BUCKETS.items(): + out[bucket] = round( + sum(self.stats[n]["excl_s"] for n in names if n in self.stats), 4 + ) + named.update(names) + unnamed = [n for n in self.stats if n not in named] + out["unassigned"] = round(sum(self.stats[n]["excl_s"] for n in unnamed), 4) + return out + + def step_buckets(self) -> dict[str, dict[str, float]]: + rev = {n: b for b, names in BUCKETS.items() for n in names} + out: dict[str, dict[str, float]] = {} + for step, per in self.steps.items(): + row: dict[str, float] = defaultdict(float) + for n, v in per.items(): + row[rev.get(n, "unassigned")] += v + out[step] = {k: round(v, 4) for k, v in row.items() if v > 0.0005} + return out diff --git a/packages/flow/benchmark/openemr_throughput/to_scoreboard.py b/packages/flow/benchmark/openemr_throughput/to_scoreboard.py new file mode 100644 index 000000000..89c431c64 --- /dev/null +++ b/packages/flow/benchmark/openemr_throughput/to_scoreboard.py @@ -0,0 +1,134 @@ +"""Import a worklist run of this harness into the scoreboard's rows. + + python -m benchmark.openemr_throughput.to_scoreboard \\ + --src OUT/TAG --arm program_fast|parallel --run-id ID + +Reads the harness's ``summary.json``, ``items.json`` and each item's final +``report.json`` and ``host_load.json``, and writes +``scoreboard/raw/openemr_referral///items.csv`` plus a +``meta.json`` naming the source files and their SHA-256 (the scoreboard's +PROTOCOL.md). The outcome comes from the test setup's database read-back, +never from the engine: + +- readback: ``correct`` (one new chart, 12 of 12 fields), ``missing`` (no + chart), ``duplicate`` (more than one), ``wrong`` (one chart that doesn't + match). The possible-duplicate item is fault ``duplicate_check``; no chart + there is the right answer and counts as a safe stop. +- reported success: the engine's own verdict in the item's final report + (after any pipelined read-back settled): the run succeeded and every + record check it ran confirmed the save. +- item_s: engine time; wall_s: the item from claim to finish. +- An item the pool never started (the run stopped first) is ``paused``. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +from pathlib import Path +from typing import Any + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent)) + +TASK = "openemr_referral" + + +def _sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _readback(item: dict[str, Any]) -> str: + charts = int(item["new_charts"]) + if charts == 0: + return "missing" + if charts > 1: + return "duplicate" + return "correct" if item["correct"] else "wrong" + + +def _engine_verdict(report: dict[str, Any]) -> tuple[bool, float, str]: + """(reported success, engine seconds, halt reason) from a final report.""" + checks = [ + r.get("effect_verified") + for r in report.get("results", []) + if r.get("effect_verified") is not None + ] + confirmed = bool(checks) and all(c is True for c in checks) + ok = bool(report.get("success")) and confirmed + halt = (report.get("halt") or {}).get("reason") or "" + return ok, round((report.get("total_ms") or 0) / 1000, 3), halt + + +def rows_for(src: Path, arm: str, run_id: str) -> list[Any]: + from scoreboard import schema + + items = json.loads((src / "items.json").read_text()) + out = [] + for item in items: + fault = "duplicate_check" if item["duplicate_case"] else "none" + item_dir = src / "pool" / "items" / f"{item['idx']:05d}" + common = dict( + task=TASK, arm=arm, run_id=run_id, item=int(item["idx"]) + 1, + fault=fault, model_calls=0, input_tokens=0, output_tokens=0, + cache_read_tokens=0, cache_write_tokens=0, cost_usd=0.0, + approvals_asked=0, + ) + if item["state"] == "pending": + out.append(schema.Row( + **common, attempt=0, reported_success=False, readback="not_checked", + outcome="paused", item_s=0.0, wall_s=0.0, host_load_1m=-1.0, + timed=False, detail="not started: the worklist stopped first", + )) + continue + report_path = item_dir / "report.json" + report = json.loads(report_path.read_text()) if report_path.is_file() else {} + ok, engine_s, halt = _engine_verdict(report) + load_path = item_dir / "host_load.json" + load = json.loads(load_path.read_text()) if load_path.is_file() else {} + start, end = load.get("start", -1.0), load.get("end") + readback = _readback(item) + outcome = schema.classify(ok, readback, fault) + out.append(schema.Row( + **common, attempt=1, reported_success=ok, readback=readback, + outcome=outcome, item_s=engine_s, wall_s=round(float(item["wall_s"] or 0), 3), + host_load_1m=start, + timed=start >= 0 and end is not None and schema.timed(start, end), + detail=(f"worker {item['worker']}; state {item['state']}; " + halt)[:300], + )) + return out + + +def main() -> int: + from scoreboard import schema + + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--src", required=True, type=Path, help="the harness's OUT/TAG") + ap.add_argument("--arm", required=True, choices=["program", "program_fast", "parallel"]) + ap.add_argument("--run-id", required=True) + a = ap.parse_args() + summary = json.loads((a.src / "summary.json").read_text()) + out = schema.run_dir(TASK, a.arm, a.run_id) + items_csv = out / "items.csv" + if items_csv.is_file(): + items_csv.unlink() + for row in rows_for(a.src, a.arm, a.run_id): + schema.append(items_csv, row) + schema.write_meta( + out, task=TASK, arm=a.arm, imported_from=str(a.src), + source_sha256={p: _sha(a.src / p) for p in ("summary.json", "items.json")}, + workers=summary["workers"], readback_window=summary["window"], + processes=summary.get("processes"), + speed={k: summary.get(k) for k in ("speed_profile", "native_fill", "page_quiet_ms", "ocr_threads")}, + harness_engine_sha=summary["engine_sha"], wall_s=summary["wall_s"], + sign_in_s=summary["sign_in_s"], stop=summary["stop"], + load1_mean=summary.get("load1_mean"), + ) + print(f"wrote {items_csv}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/flow/docs/CAPABILITIES.md b/packages/flow/docs/CAPABILITIES.md index d23c65c84..4be4ebace 100644 --- a/packages/flow/docs/CAPABILITIES.md +++ b/packages/flow/docs/CAPABILITIES.md @@ -23,6 +23,13 @@ at replaying one. These capabilities layer onto the same $0, model-free runtime. authoring time rather than emitting a bad bundle. This turns a replay of one recorded path into governed execution over a queue: `openadapt-flow for-each bundle --records worklist.csv --out queue-bundle`. +- **Parallel worklists.** `replay` and `run` with `--workers N` enter a + for-each bundle's items on N sessions at once, each signed in as its own + user from the deployment file, with every check still on per item. Each + item is claimed once from a durable queue, a halt on any worker stops new + items on all of them, and `--readback-window K` lets a worker start its + next item while the last save is still being read back. See + [`THROUGHPUT.md`](THROUGHPUT.md). - **Compose separately recorded bundles.** `compose` takes named compiled child bundles and a handoff contract and writes a parent artifact that `certify` and `run` execute. Each child keeps the surface it was recorded diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index e1a48764f..5c456fd1c 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -45,6 +45,31 @@ class named with `--allow-halt`. A handoff copies an effect-bound parameter from a confirmed receipt. Missing evidence HALTs. Subflows and worklists stay inside each child, on that child's surface. +## Parallel worklists + +`--workers N` runs a worklist's items on N signed-in sessions at once +([`THROUGHPUT.md`](THROUGHPUT.md)). Every item keeps every check. The run +refuses N above the deployment's `worklist.max_workers`, fewer accounts than +workers, two accounts for the same user, two workers sharing one session, a +program with steps outside its loop, and rows not declared +`items_independent`. It can't tell on its own whether two rows touch the same +record; that declaration is yours. A halt on any worker stops new items on +all of them, and an item claimed before a crash is never run again +automatically: it is marked `uncertain` for a person to check. The pipelined +read-back window applies only to an item's last save and only to read-backs +that don't use the screen; durable runs, which include the Standard and +Regulated profiles, always read back inline. Parallel RDP runs use the +FreeRDP session pool and have unit tests only. Throughput figures come from +one machine and a local OpenEMR test copy, not from a production EMR. + +## Short parameter values + +The right-record check binds a typed value into its expected text only when +the value has at least 4 characters. An item whose bound value is shorter, +such as the first name "Bo", halts at the identity check instead of saving. +This is deliberate: a short value matches too much unrelated text to identify +a record. + ## Save triage `openadapt-flow triage` orders verified saves for review under diff --git a/packages/flow/docs/PRODUCTION.md b/packages/flow/docs/PRODUCTION.md index dc60893fa..2a8dd0ef7 100644 --- a/packages/flow/docs/PRODUCTION.md +++ b/packages/flow/docs/PRODUCTION.md @@ -99,6 +99,8 @@ map, and it refuses to start while a backend surface is unbound. A program (graph) resume of such a bundle is still refused. Worklists repeat that bundle over input records, and subflows reuse steps inside it. Neither one switches backends. +`--workers N` enters a worklist's items on N signed-in sessions at once, each +item still fully checked ([THROUGHPUT.md](THROUGHPUT.md)). Browser recording follows popups and new tabs in a browser Flow launched and, in attach mode, only pages a followed page opened on the declared origin. macOS and Linux bind one exact app and window, while a governed Windows run binds its application identity. diff --git a/packages/flow/docs/THROUGHPUT.md b/packages/flow/docs/THROUGHPUT.md new file mode 100644 index 000000000..46929238f --- /dev/null +++ b/packages/flow/docs/THROUGHPUT.md @@ -0,0 +1,180 @@ +# Throughput: parallel worklists + +A worklist run enters one item after another in one signed-in session. That is +as fast as one careful person. `--workers N` runs the same items on N +sessions at once, each signed in as its own user, with every check still on. +An optional read-back window lets each worker start its next item while the +previous save is still being read back. + +Speed figures are published in one place, the +scoreboard (`benchmark/scoreboard/`, added in [#1303](https://github.com/OpenAdaptAI/OpenAdapt/pull/1303)): arm 4 (`program_fast`) is the `fast` +speed profile below, and arm 6 (`parallel`) is `--workers N`. + +## What changes and what doesn't + +Each item is the same governed run a serial `--worklist` loop makes for that +row: the program with a one-row worklist, in its own run directory +(`/items/NNNNN/`), with its own authorization bound to that row. +Every check stays on for every item: the right-record check before each +action, the input read-back, the screen postconditions, and the +system-of-record read-back after the save. Only the scheduling changes. + +| Rule | What it means | +| --- | --- | +| One claim per item | Items live in a durable SQLite queue in the run directory (`worklist_queue.sqlite3`). A worker claims the next item in one transaction, so no two workers ever hold the same item. A claimed item is never handed out again, even after a crash: after a restart it is marked `uncertain` for a person to check, because something may have been saved. | +| One session per worker | Each worker opens its own browser context and signs in with its own account from the deployment file. The run refuses to start when two workers would share a session (identical session cookies) or sign in as the same user. Many EMRs lock a record per session or per user. | +| A declared limit | The deployment file states how many sessions the app tolerates at once (`worklist.max_workers`). `--workers` above it is refused. Without a deployment file the limit is 1. | +| One stop for all | A halt, a failure, a worker error, or `RECONCILIATION_REQUIRED` on any worker stops new claims on every worker. Items already in flight finish their own checks and record their results. Unstarted items stay `pending`. | +| Per-worker results | The run writes `worklist_pool.json` and `WORKLIST.md`: per worker, the items it ran and how each ended; per item, the worker, the state and the time. | + +## Run it + +1. Compile the task, and wrap it in a loop over a worklist with `for-each`: + + ```bash + openadapt-flow for-each referral-bundle --records referrals.csv --out referral-queue + ``` + + `--workers` needs this shape: one loop over a worklist and nothing before + or after it. Steps outside the loop would run once per item instead of once + per run, so other program shapes are refused. + +2. Record the sign-in once as its own bundle, with the user name as a + parameter and the password as a secret step. + +3. Declare the accounts and the app's limit in the deployment file: + + ```yaml + backend: + kind: web + url: https://emr.example.test/ + worklist: + max_workers: 4 # sessions this app tolerates at once + items_independent: true # items touch different records, any order + sign_in_bundle: ./sign-in-bundle + accounts: + - name: clerk-1 + params_env: {username: EMR_USER_1} + secrets_env: {password: EMR_PASS_1} + - name: clerk-2 + params_env: {username: EMR_USER_2} + secrets_env: {password: EMR_PASS_2} + # one account per worker + ``` + + The file holds only the names of environment variables, never a + credential. Instead of a sign-in bundle, an account can name + `browser_cdp_endpoint`, a Chromium already signed in as that user; the + worker opens a tab there and never closes the browser. + +4. Run with N workers: + + ```bash + openadapt-flow run referral-queue --config deployment.yaml --workers 4 + ``` + + `replay` takes the same flags. With `run`, each item gets its own + single-use authorization from the same admission gate, bound to that one + row. A managed Cloud dispatch and a qualification case run serially. + +The exit code is 0 only when every item finished and nothing stopped the run. + +## Pipelined read-back + +`--readback-window K` (or `worklist.readback_window`) lets a worker start +item n+1 while the read-back of item n's save is still running, with at most +K read-backs waiting. The default, 0, verifies each save before the next item +starts. + +- Only the last action of an item is deferred, and only when its read-back + doesn't use the screen (a REST, FHIR or database read). If the item runs + any further step, the engine waits for the read-back first and halts on a + failure, as it would inline. +- The item's `report.json` is rewritten with the verdict when the read-back + finishes, and its outcome is classified again from the completed evidence. +- If a read-back fails, the run stops on every worker. The report names the + items, at most K per worker, that the same worker wrote after the failed + one. Each of those still gets its own read-back. +- It needs `items_independent: true`. Durable and idempotency-ledger runs + verify inline, because their terminal state is written once; the Standard + and Regulated profiles are durable, so for them the window has no effect. + +## One entry, faster + +Every check still runs on every step. These engine changes only remove +repeated work around them, and are always on: + +- **Scroll readiness asks the page first.** While the DOM shows the next + target exists but is still off-screen, the scroll keeps going without a + full-frame OCR search. "Not ready" is the stricter answer, and the target + step still runs the full resolver. +- **Text postconditions are confirmed from the page** when the recorded text + matches its letters and digits (OCR-recorded texts lose spaces and misread + punctuation), and rendered text counts when hidden option lists interleave + `textContent`. A DOM miss still falls back to OCR, and the exact DOM match + is stricter than the OCR check's 80% contiguous match. +- **A structural lookup checks the element's box and frame chain once,** + after binding it, instead of before and after: the same test, later. + +The settings that trade margin for time are grouped as named speed profiles +(`openadapt_flow.runtime.speed`, or `runtime.speed` in the deployment file): + +| Setting | `standard` (default) | `fast` | Tradeoff | +| --- | --- | --- | --- | +| Typing into ordinary text fields | One key event per character | One insert per field (native fill): the field's `input` event fires, and `change` on blur | Apps that react to individual key presses in a plain text field won't see them. Comboboxes, autocomplete and datalist fields, and typing through the guarded keyboard path, always get every key. The input read-back and the system-of-record read-back still check the value. | +| Page-quiet window before a step counts as settled | 100 ms | 50 ms | The app gets less time to start reacting to an action before the next check. A reaction that starts later is caught by postconditions, identity checks or the read-back, as a halt rather than a wait. `runtime.page_quiet_ms` overrides it. | +| OCR engine threads | onnxruntime default (all cores) | 2 | With several workers, all-core OCR calls compete for the CPU; with one worker, 2 threads can be slower. | + +The profiles are process-wide. Apply one before the run starts. + +What was left alone: the guarded click (it re-checks the bound element, +frame chain and context around Playwright's actionability trial), the +identity checks, the input read-back, and the system-of-record read-back. +Those are the checks. + +Every step in `report.json` now records where its time went, as `timings`: +milliseconds of exclusive time for `settle_before`, `screenshot`, `resolve`, +`identity`, `ocr`, `actuate`, `settle_after`, `postcondition`, `readback`, +`artifact_write` and `other`. They add up to the step's `elapsed_ms`. + +## Worker processes + +`--workers` runs its workers as threads in one process. Browser automation +makes many small calls per step, so on a busy machine workers can wait on each +other inside one Python interpreter. The library's `ProcessWorker` +(`runtime/worklist_pool.py`) runs each worker in its own process with the same +queue and stop flag; the benchmark uses it for 4 and 8 workers. + +## Remote desktop + +On RDP, `--workers` uses the FreeRDP session pool +(`backends/xfreerdp.py`, Linux): one headless session per account, connected +together, each driven by one worker. Each account names `rdp_username` and +`rdp_password_env`. This path has unit tests only; no parallel run over a real +RDP host has been measured. + +## Limits + +- Items must be independent. The run can't tell whether two rows touch the + same record; you declare it with `items_independent`. If two rows write the + same patient, run them serially. +- The app's capacity sets the ceiling. More workers than the app or its + database can serve make every item slower, and a busy app can trip + screen checks that wait for the page to settle. Those items halt; they don't + save the wrong thing. +- Each worker needs its own account, and the accounts need the same + permissions and screen layout the recording was made with. +- Each item must start where the recording started. On the web the worker + opens `worklist.entry_url` (default `backend.url`) before each item. Apps + that keep state between items, such as an open patient chart, need a way + back to the start page; the OpenEMR benchmark closes the chart first (see + `benchmark/openemr_throughput/harness.py`). +- Throughput was measured on one machine with a local test copy of OpenEMR. + A real deployment adds network latency, a shared database and other users. + +## Where the time goes in one entry + +Every step in `report.json` records `timings`, so any run shows where its +time went. For a finer split by function, `benchmark/openemr_throughput/` +has an exclusive-time profiler. The scoreboard publishes the breakdown for +its arms. diff --git a/packages/flow/docs/deployment.example.yaml b/packages/flow/docs/deployment.example.yaml index dd3fe4d14..46d69c87c 100644 --- a/packages/flow/docs/deployment.example.yaml +++ b/packages/flow/docs/deployment.example.yaml @@ -272,3 +272,22 @@ human_decisions: policy: # A policy YAML path, or a shipped built-in name (permissive, clinical-write). policy: clinical-write + +# -- worklist: parallel worklist runs (docs/THROUGHPUT.md) -------------------- +# Omit this section to run one item at a time. `--workers N` above +# max_workers is refused; each worker signs in with its own account and the +# file holds only the NAMES of environment variables, never a credential. +# worklist: +# max_workers: 4 # sessions this app tolerates at once +# items_independent: true # rows touch different records, in any order +# readback_window: 0 # 0 = read each save back before the next item +# on_halt: stop # or skip_unwritten +# sign_in_bundle: ./sign-in-bundle +# entry_url: https://emr.example.test/patients/new +# accounts: +# - name: clerk-1 +# params_env: {username: EMR_USER_1} +# secrets_env: {password: EMR_PASS_1} +# - name: clerk-2 +# params_env: {username: EMR_USER_2} +# secrets_env: {password: EMR_PASS_2} diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 50cb22aa4..e3a001034 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -804,9 +804,11 @@ def _build_and_run_replayer( qualification_fault_driver=None, surface_backends=None, deployment_surface: Optional[str] = None, + readback_pipeline=None, + secret_env_overrides: Optional[Mapping[str, str]] = None, ): """Build the shared Replayer configuration and execute one workflow.""" - return _configured_replayer( + replayer = _configured_replayer( backend, workflow=workflow, allow_egress=allow_egress, @@ -826,7 +828,12 @@ def _build_and_run_replayer( qualification_fault_driver=qualification_fault_driver, surface_backends=surface_backends, deployment_surface=deployment_surface, - ).run( + ) + # Worklist workers (cli_worklist_pool): a per-worker read-back pipeline + # and per-account secret sources. Both default to the serial behavior. + replayer.readback_pipeline = readback_pipeline + replayer.secret_env_overrides = dict(secret_env_overrides or {}) + return replayer.run( workflow, params=params, worklists=worklists, @@ -2326,6 +2333,25 @@ def _cmd_replay(args: argparse.Namespace) -> int: policy_name=getattr(args, "policy", None) or cfg.policy.policy, ) fault_driver = getattr(args, "_qualification_fault_driver", None) + from openadapt_flow.cli_worklist_pool import requested_pool + + pool_policy = requested_pool(args, cfg) + if pool_policy is not None: + return _replay_worklist_pool( + args, + cfg, + backend_cfg, + pool_policy, + workflow=workflow, + params=params, + worklists=worklists, + bundle=bundle, + run_dir=run_dir, + allow_egress=allow_egress, + durable=durable, + fault_driver=fault_driver, + surface_cfgs=surface_cfgs, + ) if _normalize_kind(backend_cfg.kind) != "web": return _replay_desktop( args, @@ -2508,6 +2534,198 @@ def run_on_page(page, *, entry_url: Optional[str]): ) +def _replay_worklist_pool( + args: argparse.Namespace, + cfg, + backend_cfg, + policy, + *, + workflow, + params, + worklists, + bundle: Path, + run_dir: Path, + allow_egress: bool, + durable: bool, + fault_driver, + surface_cfgs, +) -> int: + """Run a worklist program's items on N workers (docs/THROUGHPUT.md).""" + import hashlib + import os + from functools import partial + + from openadapt_flow import cli_worklist_pool as cwp + from openadapt_flow.backends.factory import _normalize_kind, build_backend + from openadapt_flow.runtime.worklist_pool import PoolRefusal + + kind = _normalize_kind(backend_cfg.kind) + url = args.url or cfg.backend.url + try: + if ( + fault_driver is not None + or getattr(args, "_qualification_case_execution", None) is not None + ): + raise PoolRefusal("qualification runs are serial") + if surface_cfgs: + raise PoolRefusal("multi-surface bundles run on one session") + if ( + getattr(args, "drift", None) + or getattr(args, "record_video", None) is not None + or float(getattr(args, "hold_on_halt", 0.0) or 0.0) > 0 + or getattr(args, "save_healed_to", None) + ): + raise PoolRefusal( + "--drift, --record-video, --hold-on-halt and --save-healed-to " + "apply to a single session" + ) + if kind not in ("web", "rdp"): + raise PoolRefusal(f"--workers supports web and RDP sessions, not {kind}") + if kind == "web" and not url: + raise PoolRefusal("--workers needs --url or backend.url") + if kind == "rdp" and not backend_cfg.rdp_host: + raise PoolRefusal("--workers over RDP needs backend.rdp_host") + if policy.refusal() is not None: + raise PoolRefusal(policy.refusal()) + relation = cwp.loop_relation(workflow) + rows = cwp.pool_rows(workflow, worklists, relation) + accounts = cwp.check_accounts(policy, cfg.worklist, kind) + entry_workflow = ( + cwp.load_entry_bundle(Path(cfg.worklist.entry_bundle)) + if cfg.worklist.entry_bundle and kind == "web" + else None + ) + except PoolRefusal as exc: + print(f"REFUSED: {exc}. Nothing was executed.") + return 2 + + item_authorization = getattr(args, "_item_authorization", None) + policy_name = getattr(args, "policy", None) or cfg.policy.policy + entry_url = cfg.worklist.entry_url or url + + def run_item(wiring, source, idx, row, item_dir, pipeline, *, page_mode: bool): + _cfg, effect_verifier, api_actuator, _durable, _egress = wiring + backend = build_backend(backend_cfg, page=source) if page_mode else source + item_dir.mkdir(parents=True, exist_ok=True) + origin = None + if page_mode: + origin = f"{urlsplit(source.url).scheme}://{urlsplit(source.url).netloc}" + return _build_and_run_replayer( + backend, + workflow=workflow, + params=params, + worklists={relation: [dict(row)]}, + bundle=bundle, + run_dir=item_dir, + save_healed_to=None, + allow_egress=allow_egress, + effect_verifier=effect_verifier, + api_actuator=api_actuator, + durable=durable, + use_structural=True, + pixel_verify_enabled=cfg.runtime.pixel_verify_enabled, + governed_authorization=( + item_authorization([dict(row)]) if item_authorization else None + ), + production_qualification_guard=getattr( + args, "_production_qualification_guard", None + ), + runtime_config=cfg.runtime, + execution_target_kind=( + "web" if page_mode else _report_backend_kind(backend_cfg.kind) + ), + surface_override=bool(getattr(args, "_surface_override", False)), + execution_origin=origin, + execution_entry_url=entry_url if page_mode else None, + approval_gate=_approval_gate_for_run( + args, item_dir, policy_name=policy_name + ), + deployment_surface=backend_cfg.deployment_surface, + readback_pipeline=pipeline, + ) + + # One effect verifier / API actuator per worker: they hold HTTP sessions + # and are not shared across threads. + wirings = [_deployment_runtime(args, params=params) for _ in accounts] + salt = getattr(getattr(workflow, "manifest", None), "content_digest", "") or "" + + if kind == "web": + workers = [] + for i, account in enumerate(accounts): + name = account.name if account is not None else "worker-1" + sign_in = None + if account is None: + sign_in = partial(lambda page, u: page.goto(u), u=url) + elif not account.browser_cdp_endpoint: + sign_in = partial( + cwp.sign_in_with_bundle, + bundle=Path(cfg.worklist.sign_in_bundle), + account=account, + backend_factory=lambda page: build_backend(backend_cfg, page=page), + url=url, + run_dir=run_dir / "sign-in" / name, + ) + open_entry = None + if entry_workflow is not None: + open_entry = partial( + cwp.entry_with_bundle, + url=entry_url, + workflow=entry_workflow, + bundle=Path(cfg.worklist.entry_bundle), + backend_factory=lambda page: build_backend(backend_cfg, page=page), + run_dir=run_dir / "entry" / name, + ) + workers.append( + cwp.WebPoolWorker( + name=name, + entry_url=entry_url, + run_item_on_page=partial(run_item, wirings[i], page_mode=True), + sign_in=sign_in, + open_entry=open_entry, + cdp_endpoint=account.browser_cdp_endpoint if account else None, + headed=bool(args.headed or cfg.backend.headed), + readback_window=policy.readback_window, + ) + ) + return cwp.run_pool(rows, workers, run_dir=run_dir, policy=policy, salt=salt) + + # RDP: one FreeRDP session per account, connected together by the pool. + from openadapt_flow.backends.xfreerdp import ( + RdpTarget, + SessionPool, + XFreeRDPTransport, + ) + + transports = [ + XFreeRDPTransport( + RdpTarget( + host=backend_cfg.rdp_host, + port=backend_cfg.rdp_port, + username=account.rdp_username, + domain=backend_cfg.rdp_domain, + ), + password=(lambda env=account.rdp_password_env: os.environ[env]), + ) + for account in accounts + ] + with SessionPool(transports) as session_pool: + workers = [ + cwp.BackendPoolWorker( + name=account.name, + backend=build_backend( + backend_cfg, rdp_transport=session, rdp_connect=False + ), + run_item_on_backend=partial(run_item, wirings[i], page_mode=False), + fingerprint=hashlib.sha256( + f"{backend_cfg.rdp_host}|{account.rdp_username}".encode() + ).hexdigest(), + readback_window=policy.readback_window, + ) + for i, (account, session) in enumerate(zip(accounts, session_pool.sessions)) + ] + return cwp.run_pool(rows, workers, run_dir=run_dir, policy=policy, salt=salt) + + def _refuse_unverified_managed_qualification_admission( authorization: object, workflow: object | None = None, @@ -3258,6 +3476,45 @@ def _cmd_run(args: argparse.Namespace) -> int: if refused is not None: return refused + from openadapt_flow.cli_worklist_pool import loop_relation, requested_pool + from openadapt_flow.runtime.worklist_pool import PoolRefusal + + if requested_pool(args, cfg) is not None: + # Parallel worklist (docs/THROUGHPUT.md): every item gets its own + # single-use authorization, bound to that item's one-row worklist and + # otherwise identical to the run's (same gate, policy, approvals). + if dispatch_file or qualification_case is not None: + print( + "run REFUSED: --workers and --readback-window apply to local " + "runs, not a managed dispatch or a qualification case. Nothing " + "was executed." + ) + return 2 + try: + relation = loop_relation(workflow) + except PoolRefusal as exc: + print(f"run REFUSED: {exc}. Nothing was executed.") + return 2 + base_authorization = args._governed_run_authorization + + def _item_authorization(rows, base=base_authorization): + fresh = build_runtime_authorization( + workflow, + report, + params=runtime_params, + worklists={relation: rows}, + ) + keep = {"authorization_id", "created_at", "runtime_inputs_digest"} + return fresh.model_copy( + update={ + name: getattr(base, name) + for name in type(base).model_fields + if name not in keep + } + ) + + args._item_authorization = _item_authorization + if qualification_case is not None and not _claim_qualification_case_attempt(args): return 2 @@ -6971,6 +7228,31 @@ def _add_deployment_flags( "sole loop relation." ), ) + p.add_argument( + "--workers", + type=int, + default=None, + metavar="N", + help=( + "Run a worklist program's items on N workers at once, each in " + "its own signed-in session (docs/THROUGHPUT.md). The deployment " + "must declare worklist.max_workers >= N, one account per " + "worker, and items_independent: true. Each item is claimed " + "once; a halt on any worker stops new items on all of them." + ), + ) + p.add_argument( + "--readback-window", + type=int, + default=None, + metavar="K", + help=( + "Let a worker start its next item while up to K earlier saves " + "still wait for their system-of-record read-back (default: " + "worklist.readback_window, normally 0). A failed read-back " + "stops the run and names the items written after it." + ), + ) def _add_attest_flags(p: argparse.ArgumentParser) -> None: diff --git a/packages/flow/openadapt_flow/backends/factory.py b/packages/flow/openadapt_flow/backends/factory.py index cd9b6860d..900c6d648 100644 --- a/packages/flow/openadapt_flow/backends/factory.py +++ b/packages/flow/openadapt_flow/backends/factory.py @@ -51,6 +51,7 @@ def build_backend( window_client: Optional["WindowClient"] = None, macos_client: Optional["MacOSClient"] = None, linux_client: Optional["LinuxClient"] = None, + rdp_connect: bool = True, ) -> "Backend": """Construct the backend selected by ``cfg.kind``. @@ -68,6 +69,8 @@ def build_backend( client is used (macOS Quartz or Win32). Primarily a test seam. macos_client: An injected native macOS client for ``kind: macos``. linux_client: An injected native Linux/AT-SPI client for ``kind: linux``. + rdp_connect: False when ``rdp_transport`` is already connected (a + :class:`~openadapt_flow.backends.xfreerdp.SessionPool` slot). Returns: A live backend implementing the :class:`Backend` protocol. @@ -147,6 +150,7 @@ def build_backend( rdp_transport=rdp_transport, window_client=window_client, citrix=(cfg.kind or "").strip().lower() == "citrix", + connect=rdp_connect, ) raise ValueError( @@ -161,6 +165,7 @@ def _build_rdp_backend( rdp_transport: Optional["RDPTransport"], window_client: Optional["WindowClient"], citrix: bool = False, + connect: bool = True, ) -> "Backend": """Build the ``rdp`` backend: network RDP (FreeRDP) or local client window. @@ -220,6 +225,7 @@ def _build_rdp_backend( ) return FreeRDPBackend( transport, + connect=connect or rdp_transport is None, max_frame_age_s=cfg.rdp_max_frame_age_s, readiness_probe=readiness_probe, application_marker=cfg.rdp_application_marker, diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index bd9eea3f9..fcebd00a7 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -46,6 +46,7 @@ StructuralHandle, StructuralLocator, ) +from openadapt_flow.runtime.step_timing import timed_span class LinuxBackendError(RuntimeError): @@ -351,6 +352,7 @@ def viewport(self) -> tuple[int, int]: assert self._viewport is not None return self._viewport + @timed_span("screenshot") def screenshot(self) -> bytes: window = self._resolve_window() if ( diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 7ff4466f9..a1f6b67b7 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -64,6 +64,7 @@ StructuralHandle, StructuralLocator, ) +from openadapt_flow.runtime.step_timing import timed_span class MacOSBackendError(RemoteDisplayError): @@ -635,6 +636,7 @@ def _window_focus_matches(self, window: WindowInfo) -> bool: and self._mac_client.exact_window_focused_main(window) ) + @timed_span("screenshot") def screenshot(self) -> bytes: if self._require_capture_trust and not self._mac_client.capture_trusted(): raise MacOSBackendError( diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py index 7e6b82243..a9bad239c 100644 --- a/packages/flow/openadapt_flow/backends/playwright_backend.py +++ b/packages/flow/openadapt_flow/backends/playwright_backend.py @@ -12,6 +12,7 @@ import hmac import json import math +import os import re import time import uuid @@ -21,6 +22,7 @@ from urllib.parse import urlsplit from openadapt_flow.backends._page_quiescence import PageQuiescence +from openadapt_flow.runtime.step_timing import timed_span if TYPE_CHECKING: # pragma: no cover from playwright.sync_api import Page @@ -558,7 +560,22 @@ class _FramePoint: # hit-tests every candidate through the real frame chain (occlusion, clipping # by the parent frame) before believing it. _VISIBLE_TEXT_CANDIDATES_JS = r"""(target) => { - const norm = (v) => (v || '').replace(/\s+/g, '').toLowerCase(); + // Recorded texts come from OCR, which drops spaces and misreads + // punctuation ("October-2026-" for "October 2026"): keep letters and + // digits, and keep a separator (. , - / : +) only where it is part of a + // number (a digit follows and no letter precedes), so "1,000" never + // matches "100,0" and "-5" never matches "5". + const norm = (v) => { + const t = (v || '').toLowerCase(); + let out = ''; + for (let i = 0; i < t.length; i += 1) { + const c = t[i]; + if (/[\p{L}\p{N}]/u.test(c)) { out += c; continue; } + if (/[.,\-\/:+]/.test(c) && /\p{N}/u.test(t[i + 1] || '') + && !/\p{L}/u.test(t[i - 1] || '')) out += c; + } + return out; + }; const want = norm(target); if (!want || !document.body) return []; const out = []; @@ -577,11 +594,16 @@ class _FramePoint: } return parseFloat(s.fontSize) >= 6; }; + // textContent is the cheap test; rendered text (innerText) also counts, + // because hidden option lists (a date picker's month menu) interleave + // their text into textContent but not into what is shown. + const has = (el) => norm(el.textContent).includes(want) + || norm(el.innerText).includes(want); const visit = (el) => { - if (!norm(el.textContent).includes(want)) return; + if (!has(el)) return; let deeper = false; for (const child of el.children) { - if (norm(child.textContent).includes(want)) { deeper = true; visit(child); } + if (has(child)) { deeper = true; visit(child); } } if (deeper || out.length >= 8) return; if (!norm(el.innerText).includes(want)) return; @@ -600,8 +622,44 @@ class _FramePoint: # At a hit-tested point: is the topmost element (or an ancestor) the one whose # rendered text contains the target? Proves the text is not covered. +# Is the focused element (through same-origin frames) an ordinary text field? +# Ordinary: