From 4ec5247758e4b929325b5c9d9eb734e4232663c1 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 12:42:30 -0400 Subject: [PATCH 1/2] Add the lifecycle harness: workload, operator, oracle, cost model, metric A seeded referral stream with an exception mix and its ground truth, a scripted operator charged from a versioned cost model, an independent SQL oracle, the arm interface (lifecycle-arm/1), and the metric (verified completed items per operator-hour with x0.5/x2 bands). The OpenEMR adapter injects faults and server-side UI changes; an in-memory app and toy arms check the bookkeeping in tests. Co-Authored-By: Claude Opus 5.5 --- packages/flow/benchmark/lifecycle/__init__.py | 0 packages/flow/benchmark/lifecycle/apps.py | 577 ++++++++++++++++++ packages/flow/benchmark/lifecycle/arm.py | 146 +++++ .../flow/benchmark/lifecycle/cost_model.py | 106 ++++ packages/flow/benchmark/lifecycle/fake.py | 217 +++++++ packages/flow/benchmark/lifecycle/metrics.py | 271 ++++++++ packages/flow/benchmark/lifecycle/operator.py | 145 +++++ packages/flow/benchmark/lifecycle/oracle.py | 185 ++++++ packages/flow/benchmark/lifecycle/report.py | 179 ++++++ packages/flow/benchmark/lifecycle/run.py | 385 ++++++++++++ packages/flow/benchmark/lifecycle/workload.py | 341 +++++++++++ .../flow/tests/test_lifecycle_benchmark.py | 178 ++++++ 12 files changed, 2730 insertions(+) create mode 100644 packages/flow/benchmark/lifecycle/__init__.py create mode 100644 packages/flow/benchmark/lifecycle/apps.py create mode 100644 packages/flow/benchmark/lifecycle/arm.py create mode 100644 packages/flow/benchmark/lifecycle/cost_model.py create mode 100644 packages/flow/benchmark/lifecycle/fake.py create mode 100644 packages/flow/benchmark/lifecycle/metrics.py create mode 100644 packages/flow/benchmark/lifecycle/operator.py create mode 100644 packages/flow/benchmark/lifecycle/oracle.py create mode 100644 packages/flow/benchmark/lifecycle/report.py create mode 100644 packages/flow/benchmark/lifecycle/run.py create mode 100644 packages/flow/benchmark/lifecycle/workload.py create mode 100644 packages/flow/tests/test_lifecycle_benchmark.py diff --git a/packages/flow/benchmark/lifecycle/__init__.py b/packages/flow/benchmark/lifecycle/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/packages/flow/benchmark/lifecycle/apps.py b/packages/flow/benchmark/lifecycle/apps.py new file mode 100644 index 000000000..4a97259da --- /dev/null +++ b/packages/flow/benchmark/lifecycle/apps.py @@ -0,0 +1,577 @@ +"""The apps a lifecycle run drives, and every event the harness injects into them. + +OpenEMR 8.0.0.3 runs as its own Compose project (``openadapt-openemr-lifecycle``, +loopback ports 9421 for the browser and 9420 for the API) from the pinned +fixture in ``benchmark/openemr_local``, with synthetic data only. Nothing here +touches another session's copy. + +Injected events, each logged with exactly what was changed: + +- other clerks: a chart created through the REST API as the actor account, + or an unrelated chart's home phone changed through SQL; +- forced logout: every PHP session file in the app container is deleted; +- slow save: a ``BEFORE INSERT`` trigger on ``patient_data`` that sleeps for + the item's patient only; +- failed save: the same trigger raising an error, a set number of times (the + counter lives in a non-transactional table, so it survives the rollback); +- UI changes, served to every client as a vendor release would be: + - ``theme``: a stylesheet that changes colours, fonts and spacing on the + New Patient form and its duplicate-check popup; + - ``relabel``: the save button "Confirm Create New Patient" becomes + "Save new patient", and "Create New Patient" becomes "Check and create"; + - ``reorder``: the demographics layout (``layout_options``) puts Birth Sex + before DOB, and Home Phone and Contact Email before Address; + - ``interstitial``: a notice over the New Patient form that must be + dismissed ("Continue") before the form takes input. + A version bump is not run: the lock pins one OpenEMR image and a second + pinned image would need its own baseline (see the report's limits). +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import threading +import time +from pathlib import Path +from typing import Any, Optional + +HERE = Path(__file__).resolve().parent +BENCH = HERE.parent +PROJECT = "openadapt-openemr-lifecycle" +UI_PORT = "9421" +API_PORT = "9420" +WEBROOT = "/var/www/localhost/htdocs/openemr" +FORM = f"{WEBROOT}/interface/new/new_comprehensive.php" +POPUP = f"{WEBROOT}/interface/new/new_search_popup.php" +DRIFT_JS = f"{WEBROOT}/public/lifecycle-drift.js" +SQL_FIELDS = ( + "id", + "pid", + "title", + "fname", + "lname", + "DOB", + "sex", + "street", + "city", + "state", + "postal_code", + "phone_home", + "email", + "country_code", +) + +THEME_CSS = ( + "body{background:#fbf6ee !important;font-family:Georgia,serif !important;}" + ".card,.card-header,.nav-tabs .nav-link{background:#efe4d2 !important;}" + ".btn-primary{background:#6b3d8f !important;border-color:#6b3d8f !important;}" + "label,.label_custom{color:#5a3218 !important;}" + ".form-control{border-radius:0 !important;border-color:#a08060 !important;}" +) +INTERSTITIAL_JS = r""" +(function(){ + if (!/new_comprehensive\.php/.test(location.pathname)) return; + function show(){ + if (document.getElementById('oa-lifecycle-notice')) return; + var d=document.createElement('div'); + d.id='oa-lifecycle-notice'; + d.setAttribute('role','dialog'); + d.style.cssText='position:fixed;inset:0;background:rgba(0,0,0,.45);z-index:99999;display:flex;align-items:center;justify-content:center'; + d.innerHTML='
'+ + '

Registration form updated

'+ + '

Contact details are now checked when you save. Review the new layout before you continue.

'+ + '
'; + document.body.appendChild(d); + document.getElementById('oa-lifecycle-continue').onclick=function(){d.remove();}; + } + if (document.readyState==='loading') document.addEventListener('DOMContentLoaded',show); else show(); +})(); +""" + + +def podman_env() -> dict[str, str]: + """Docker Compose pointed at the benchmark Podman machine.""" + + env = dict(os.environ) + if not env.get("DOCKER_HOST"): + sock = subprocess.run( + [ + "podman", + "machine", + "inspect", + "openadapt-benchmark", + "--format", + "{{.ConnectionInfo.PodmanSocket.Path}}", + ], + capture_output=True, + text=True, + check=False, + ).stdout.strip() + if sock: + env["DOCKER_HOST"] = f"unix://{sock}" + return env + + +def make_fixture(state_dir: Path) -> Any: + import benchmark.openemr_local.fixture as fx + + class LifecycleFixture(fx.OpenEMRFixture): + def __init__(self, state: Path) -> None: + super().__init__(state) + self.lock = {**self.lock, "project": PROJECT} + + @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 _expected(self) -> dict[str, str]: + return { + "OPENEMR_IMAGE": self.lock["services"]["openemr"], + "MARIADB_IMAGE": self.lock["services"]["mariadb"], + "OPENEMR_ACTOR_USER": "openadapt_actor", + "OPENEMR_HTTP_PORT": UI_PORT, + "OPENEMR_HTTPS_PORT": API_PORT, + } + + def _validate_runtime_env(self) -> None: + values = self._runtime_values() + for k, v in self._expected().items(): + if values.get(k) != v: + raise fx.FixtureError(f"runtime.env {k} differs") + self.runtime_env.chmod(0o600) + + def runtime_preflight(self) -> None: + # The shared machine's rules (~/oa/scratch/fixture-lease/README.md): + # at least 15 GiB free and a 1-minute load under 40 before any + # container starts. The pinned images are already in the Podman + # machine. + import shutil + + free = shutil.disk_usage("/System/Volumes/Data").free / 2**30 + if free < 15: + raise fx.FixtureError(f"only {free:.1f} GiB free on the host") + load = os.getloadavg()[0] + if load >= 40: + raise fx.FixtureError(f"1-minute load is {load:.0f}; wait for < 40") + + def prepare(self) -> None: + import secrets + + self._protect_state_dir() + if not self.runtime_env.exists(): + values = { + **self._expected(), + "MARIADB_ROOT_PASSWORD": secrets.token_urlsafe(32), + "OPENEMR_DB_PASSWORD": secrets.token_urlsafe(32), + "OPENEMR_ACTOR_PASSWORD": secrets.token_urlsafe(32), + } + self._write_private_exclusive( + self.runtime_env, + ( + "# generated local synthetic fixture; never commit\n" + + "".join(f"{k}={v}\n" for k, v in values.items()) + ).encode(), + ) + super().prepare() + + def _run(self, argv, *, input_bytes=None, timeout_s=300.0): # type: ignore[override] + # Same as the base, with Compose pointed at the Podman machine. + if argv[:2] == ["docker", "compose"] or argv[:1] == ["docker"]: + proc = subprocess.run( + argv, + input=input_bytes, + capture_output=True, + timeout=timeout_s, + env=podman_env(), + check=False, + ) + if proc.returncode != 0: + raise fx.FixtureError( + f"{' '.join(argv[:4])} failed: {proc.stderr.decode()[-800:]}" + ) + return proc + return super()._run(argv, input_bytes=input_bytes, timeout_s=timeout_s) + + return LifecycleFixture(state_dir) + + +class OpenEMRApp: + """The app adapter the harness drives: reset, read, inject, change.""" + + def __init__(self, state_dir: Optional[Path] = None) -> None: + self.state_dir = Path( + state_dir + or os.environ.get("LIFECYCLE_OPENEMR_STATE") + or Path.home() / "oa" / "scratch" / "lifecycle" / "openemr-state" + ) + self.fx = make_fixture(self.state_dir) + self.db_container = f"{PROJECT}-db-1" + self.app_container = f"{PROJECT}-openemr-1" + self.ui_url = self.fx.ui_base_url + self.api_url = self.fx.api_base_url + self.drift_on: list[str] = [] + self._fault_lock = threading.Lock() + from benchmark.openemr_throughput.harness import CLERKS + + #: Added before the baseline snapshot, so every reset keeps them. + self._clerks: list[str] = list(CLERKS[:3]) + + # -- bring-up ---------------------------------------------------------- + + def bootstrapped(self) -> bool: + return all( + p.is_file() + for p in ( + self.fx.runtime_env, + self.fx.clients_path, + self.fx.snapshot_path, + self.fx.snapshot_hash_path, + ) + ) + + def bring_up(self, say: Any = print) -> str: + if self.bootstrapped(): + say("starting the existing copy") + self.fx._compose("up", "-d", timeout_s=900.0) + self.fx._wait_ready(timeout_s=900.0) + return "restarted" + self.fx.runtime_preflight() + say("checking the pinned OpenEMR release") + self.fx.prepare() + self.fx.image_identity() + say("starting; OpenEMR installs itself on first start") + self.fx._compose("up", "-d", timeout_s=900.0) + self.fx._wait_ready(timeout_s=1800.0) + self.fx._verify_running_source() + say("adding synthetic API clients") + self.fx.bootstrap() + self.add_clerks(3) + self.fx.snapshot() + return "created" + + def describe(self) -> dict[str, Any]: + return { + "app": "openemr", + "version": "8.0.0.3", + "project": PROJECT, + "ui_url": self.ui_url, + "api_url": self.api_url, + "images": self.fx.lock["services"], + "baseline_sha256": self.fx.baseline_hash() if self.bootstrapped() else None, + } + + def extra(self) -> dict[str, Any]: + """What arms may use. ``rest_read_session()`` returns an httpx client + with a read-only bearer token (``user/patient.rs``), the same REST + read-back OpenAdapt's record check uses; ``rest_write_session()`` one + with the actor's create/update scope. Arms never get the oracle's SQL.""" + + return { + "fixture_state": str(self.state_dir), + "fixture_lock": str(self.state_dir / "environment.lock.json"), + "project": PROJECT, + "ui_port": int(UI_PORT), + "api_port": int(API_PORT), + "rest_read_session": lambda: self.fx.token_session("oracle"), + "rest_write_session": lambda: self.fx.token_session("actor"), + } + + def point_hooks_here(self) -> None: + """verify.yaml's hooks (sign-in, REST read-back) read these variables.""" + + lock = self.state_dir / "environment.lock.json" + lock.write_text(json.dumps(self.fx.lock, indent=2) + "\n") + os.environ["OPENEMR_FIXTURE_STATE"] = str(self.state_dir) + os.environ["OPENEMR_FIXTURE_LOCK"] = str(lock) + os.environ["OPENEMR_UI_URL"] = self.ui_url + os.environ["OPENEMR_API_URL"] = self.api_url + os.environ.update({k: v for k, v in podman_env().items() if k == "DOCKER_HOST"}) + + # -- accounts ------------------------------------------------------------ + + def add_clerks(self, n: int) -> list[str]: + from benchmark.openemr_throughput.harness import add_clerks + + self._clerks = add_clerks(self.fx, n) + return self._clerks + + def accounts(self) -> list[tuple[str, str]]: + pw = self.fx._runtime_values()["OPENEMR_ACTOR_PASSWORD"] + users = ["openadapt_actor"] + [ + c for c in self._clerks if c != "openadapt_actor" + ] + return [(u, pw) for u in users] + + # -- direct database channel (the oracle's) ----------------------------- + + def sql(self, statement: str, *, timeout_s: float = 120.0) -> str: + proc = subprocess.run( + [ + "podman", + "exec", + "-i", + self.db_container, + "sh", + "-c", + 'MYSQL_PWD="$MARIADB_ROOT_PASSWORD" exec mariadb -uroot --batch --raw ' + "--skip-column-names", + ], + input=statement.encode(), + capture_output=True, + timeout=timeout_s, + check=False, + ) + if proc.returncode != 0: + raise RuntimeError(f"sql failed: {proc.stderr.decode()[-500:]}") + return proc.stdout.decode() + + def rows(self) -> list[dict[str, str]]: + out = self.sql( + "SELECT " + ",".join(SQL_FIELDS) + " FROM openemr.patient_data ORDER BY id;" + ) + return [ + { + k: ("" if v == "NULL" else v) + for k, v in zip(SQL_FIELDS, line.split("\t")) + } + for line in out.splitlines() + if line + ] + + def app_exec(self, script: str, *, timeout_s: float = 60.0) -> str: + proc = subprocess.run( + ["podman", "exec", self.app_container, "sh", "-c", script], + capture_output=True, + timeout=timeout_s, + check=False, + ) + if proc.returncode != 0: + raise RuntimeError(f"app exec failed: {proc.stderr.decode()[-500:]}") + return proc.stdout.decode() + + # -- state --------------------------------------------------------------- + + def reset(self) -> str: + self.clear_all_events() + return self.fx.reset() + + def seed_existing(self, people: list[dict[str, str]]) -> list[str]: + return [self.create_chart(p, who="seed") for p in people] + + def create_chart(self, fields: dict[str, str], *, who: str) -> str: + """A chart created through the REST API (another clerk, the operator + doing an item by hand, or the seed). Returns its patient_data id.""" + + body = { + "title": fields["title"], + "fname": fields["fname"].strip(), + "lname": fields["lname"].strip(), + "DOB": fields["DOB"], + "sex": fields["sex"], + "street": fields["street"], + "city": fields["city"], + "state": "MA", + "postal_code": fields["postal_code"], + "phone_home": fields.get("phone_home", ""), + "email": fields["email"], + "country_code": fields.get("country_code", "USA"), + } + before = {r["id"] for r in self.rows()} + session = self.fx.token_session("actor") + try: + resp = session.post(f"{self.api_url}/apis/default/api/patient", json=body) + finally: + session.close() + if resp.status_code not in (200, 201): + raise RuntimeError(f"{who} create returned HTTP {resp.status_code}") + new = [r["id"] for r in self.rows() if r["id"] not in before] + if len(new) != 1: + raise RuntimeError(f"{who} create made {len(new)} rows") + return new[0] + + # -- item-level events ------------------------------------------------------ + + def inject( + self, + inj: dict[str, Any], + truth: Any, + *, + harness: set[str], + stop: Optional[threading.Event] = None, + ) -> dict[str, Any]: + event = inj["event"] + if event == "other_clerk_creates": + cid = self.create_chart(inj["person"], who="other clerk") + return {"event": event, "chart_id": cid} + if event == "force_logout": + n = self.app_exec( + "n=$(ls /tmp/sess_* 2>/dev/null | wc -l); rm -f /tmp/sess_*; echo $n" + ) + return {"event": event, "sessions_ended": int(n.strip() or 0)} + if event == "other_clerk_edits_unrelated": + # wait a moment so the edit lands while the item is under way + if stop is not None: + stop.wait(5.0) + rows = [r for r in self.rows() if r["id"] not in harness] + target = rows[truth.seq % len(rows)] if rows else None + if target is None: + return {"event": event, "touched_ids": []} + phone = f"555-01{truth.seq % 100:02d}" + self.sql( + f"UPDATE openemr.patient_data SET phone_home='{phone}' " + f"WHERE id={int(target['id'])};" + ) + return {"event": event, "touched_ids": [target["id"]], "phone_home": phone} + if event in ("slow_save", "failed_save"): + email = truth.person["email"] + assert re.fullmatch(r"[a-z0-9.@-]+", email) + if event == "slow_save": + body = f"IF NEW.email = '{email}' THEN SET @oa_slept = SLEEP({int(inj['seconds'])}); END IF;" + else: + body = ( + f"IF NEW.email = '{email}' AND (SELECT n FROM lifecycle.fault_budget " + f"WHERE k='{email}') > 0 THEN " + f"UPDATE lifecycle.fault_budget SET n = n - 1 WHERE k='{email}'; " + "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'lifecycle: injected save failure'; " + "END IF;" + ) + self.sql( + "CREATE DATABASE IF NOT EXISTS lifecycle; " + "CREATE TABLE IF NOT EXISTS lifecycle.fault_budget " + "(k VARCHAR(190) PRIMARY KEY, n INT) ENGINE=Aria TRANSACTIONAL=0; " + f"REPLACE INTO lifecycle.fault_budget VALUES ('{email}', {int(inj.get('times', 0))}); " + "DROP TRIGGER IF EXISTS openemr.oa_lifecycle_fault; " + "CREATE TRIGGER openemr.oa_lifecycle_fault BEFORE INSERT ON openemr.patient_data " + f"FOR EACH ROW BEGIN {body} END;" + ) + return {"event": event, "trigger": body} + raise ValueError(f"unknown injection {event}") + + def clear_item_events(self) -> list[str]: + out = self.sql( + "SELECT TRIGGER_NAME FROM information_schema.TRIGGERS " + "WHERE TRIGGER_NAME='oa_lifecycle_fault';" + ) + if out.strip(): + self.sql("DROP TRIGGER IF EXISTS openemr.oa_lifecycle_fault;") + return ["oa_lifecycle_fault"] + return [] + + # -- UI changes ---------------------------------------------------------------- + + def _ensure_originals(self) -> None: + self.app_exec( + f"for f in {FORM} {POPUP}; do [ -f $f.oa-orig ] || cp -p $f $f.oa-orig; done" + ) + + def _write_drift_js(self) -> None: + parts = [] + if "theme" in self.drift_on: + parts.append( + "(function(){var s=document.createElement('style');s.textContent=" + + json.dumps(THEME_CSS) + + ";(document.head||document.documentElement).appendChild(s);})();" + ) + if "interstitial" in self.drift_on: + parts.append(INTERSTITIAL_JS) + js = "\n".join(parts) + self.app_exec(f"cat > {DRIFT_JS} <<'OAEOF'\n{js}\nOAEOF\nchmod 644 {DRIFT_JS}") + + def apply_drift(self, kind: str) -> dict[str, Any]: + self._ensure_originals() + if kind not in self.drift_on: + self.drift_on.append(kind) + changed: dict[str, Any] = {"kind": kind} + if kind in ("theme", "interstitial"): + self._write_drift_js() + tag = '' + for f in (FORM, POPUP): + self.app_exec( + f"grep -q lifecycle-drift.js {f} || " + f"sed -i 's##{tag}#' {f}" + ) + changed["files"] = [FORM, POPUP, DRIFT_JS] + elif kind == "relabel": + self.app_exec( + f"sed -i \"s/xla('Confirm Create New Patient')/xla('Save new patient')/\" {POPUP}; " + f"sed -i \"s/xlt('Create New Patient')/xlt('Check and create')/; " + f"s/xla('Create New Patient')/xla('Check and create')/; " + f"s/xlj('Create New Patient')/xlj('Check and create')/\" {FORM}" + ) + changed["labels"] = { + "Confirm Create New Patient": "Save new patient", + "Create New Patient": "Check and create", + } + elif kind == "reorder": + sql = ( + "UPDATE openemr.layout_options SET seq=85 WHERE form_id='DEM' AND field_id='sex';" + "UPDATE openemr.layout_options SET seq=0 WHERE form_id='DEM' AND field_id='phone_home';" + "UPDATE openemr.layout_options SET seq=0 WHERE form_id='DEM' AND field_id='email';" + ) + self.sql(sql) + changed["sql"] = sql + else: + raise ValueError(f"unknown UI change {kind}") + return changed + + def clear_all_events(self) -> None: + self.drift_on = [] + try: + self.app_exec( + f"for f in {FORM} {POPUP}; do [ -f $f.oa-orig ] && cp -p $f.oa-orig $f; done; " + f"rm -f {DRIFT_JS}; true" + ) + except (RuntimeError, subprocess.SubprocessError): + pass + try: + self.clear_item_events() + self.sql( + "UPDATE openemr.layout_options SET seq=100 WHERE form_id='DEM' AND field_id='sex';" + "UPDATE openemr.layout_options SET seq=11 WHERE form_id='DEM' AND field_id='phone_home';" + "UPDATE openemr.layout_options SET seq=14 WHERE form_id='DEM' AND field_id='email';" + ) + except (RuntimeError, subprocess.SubprocessError): + pass + + +def make_app(name: str) -> Any: + if name == "fake": + from .fake import FakeApp + + return FakeApp() + if name == "openemr": + app = OpenEMRApp() + app.point_hooks_here() + return app + raise ValueError(f"unknown app {name}") + + +def main() -> int: + """Bring the lifecycle OpenEMR copy up (and set it up on first use).""" + + import argparse + + p = argparse.ArgumentParser(description=main.__doc__) + p.add_argument("command", choices=("up", "status", "down")) + args = p.parse_args() + app = OpenEMRApp() + if args.command == "up": + t = time.monotonic() + how = app.bring_up() + print(json.dumps({"how": how, "seconds": round(time.monotonic() - t, 1)})) + elif args.command == "status": + print(json.dumps({"bootstrapped": app.bootstrapped(), "ui": app.ui_url})) + else: + app.fx._compose("stop", timeout_s=300.0) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/benchmark/lifecycle/arm.py b/packages/flow/benchmark/lifecycle/arm.py new file mode 100644 index 000000000..3a1e1fecb --- /dev/null +++ b/packages/flow/benchmark/lifecycle/arm.py @@ -0,0 +1,146 @@ +"""The interface every comparison arm implements. + +The harness owns the worklist and its ground truth, the app and every injected +event, the independent oracle, the simulated operator and the cost model. An +arm owns only how it sets itself up and how it does one item. + +Two rules keep arms comparable: + +- An arm never charges itself operator time. Every human touch (an approval, + a stop, a question, a reconciliation, a repair review, an item done by hand) + goes through ``operator.ask``, which returns the decision and the seconds the + versioned cost model charges for it. +- An arm never judges itself. ``ItemResult.claimed`` is what the arm says; the + oracle reads the system of record after every item and decides. + +Arms are loaded from ``module:factory`` strings; ``factory(options) -> Arm``. +""" + +from __future__ import annotations + +import importlib +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal, Optional, Protocol + +INTERFACE_VERSION = "lifecycle-arm/1" + +RequestKind = Literal[ + "approval", "stop", "reconcile", "repair_review", "question", "manual_entry" +] + + +@dataclass(frozen=True) +class WorkItem: + """One unit of work as the source document states it. + + ``params`` may be missing fields or hold malformed values: they are what + the referral says, not what is true. The harness keeps the truth. + """ + + seq: int + item_id: str + day: int + app: str + params: dict[str, str] + source_text: str + + +@dataclass(frozen=True) +class Env: + app: str + ui_url: str + api_url: Optional[str] + #: Synthetic (user, password) pairs the arm may sign in with. + accounts: tuple[tuple[str, str], ...] + #: The arm's private scratch directory for this run. + workdir: Path + seed: int + #: App-specific extras (for example the fixture's container names). + extra: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class OperatorRequest: + kind: RequestKind + item_id: Optional[str] + #: The product's own reason code, or the agent's question, verbatim. + reason: str + #: Where the request surfaced, e.g. "cli:openadapt-flow durable resume". + surface: str + detail: dict[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class OperatorDecision: + #: approve | deny | resume | skip | do_by_hand | accept_repair | + #: reject_repair | answer | confirm_saved | confirm_not_saved + action: str + charged_s: float + #: Which scripted-policy rule decided it. + rule: str + answer: Optional[str] = None + + +class Operator(Protocol): + def ask(self, request: OperatorRequest) -> OperatorDecision: ... + + +@dataclass +class ItemResult: + item_id: str + claimed: Literal["done", "stopped", "error"] + #: The product's final status or the agent's last line, verbatim. + product_status: str + #: The arm's own wall time for the item (first action to return). + wall_s: float + model_calls: int = 0 + cost_usd: float = 0.0 + #: Evidence path relative to the run directory. + evidence: str = "" + #: Maintenance the arm did inside this item (for example re-authoring a + #: script after it failed on a changed screen). Same shape as a setup + #: phase; charged as maintenance, not as per-item work. + phases: list[dict[str, Any]] = field(default_factory=list) + #: Free-form, kept in the ledger. ``detail["usage"]`` may hold a token + #: breakdown: [{"model", "input", "output", "cache_read", "cache_write", + #: "calls", "usd"}]. + detail: dict[str, Any] = field(default_factory=dict) + + +class Arm(Protocol): + name: str + + def describe(self) -> dict[str, Any]: + """The exact configuration, recorded in the ledger as is.""" + + def setup(self, env: Env, operator: Operator) -> list[dict[str, Any]]: + """Teach, compile, qualify, admit. One dict per phase: + ``{"phase", "wall_s", "attended_s", "model_calls", "cost_usd"}``. + ``attended_s`` is the part of the wall time a person must watch.""" + + def run_item(self, item: WorkItem, env: Env, operator: Operator) -> ItemResult: ... + + def after_drift( + self, event: dict[str, Any], env: Env, operator: Operator + ) -> list[dict[str, Any]]: + """Maintenance an arm does on its own initiative after a UI change it + was told about (most arms are not told, and return []).""" + + def close(self) -> None: ... + + +class SpendCapReached(RuntimeError): + """Raised by an arm before a model call that could cross a spend cap. The + runner stops that arm, records the rest of its items as paused, and goes + on with the next arm.""" + + +def load_arm(spec: str, options: Optional[dict[str, Any]] = None) -> Arm: + """Build an arm from ``module:factory``.""" + + module, _, name = spec.partition(":") + if not module or not name: + raise ValueError(f"arm must look like module:factory, got {spec!r}") + factory = getattr(importlib.import_module(module), name) + return factory(dict(options or {})) diff --git a/packages/flow/benchmark/lifecycle/cost_model.py b/packages/flow/benchmark/lifecycle/cost_model.py new file mode 100644 index 000000000..fca658045 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/cost_model.py @@ -0,0 +1,106 @@ +"""The operator cost model: how many seconds each kind of human touch is charged. + +Versioned. A change to any number is a new version with a new key, never an +edit of an old one, so every results ledger names the exact numbers it was +charged with. The report shows every result at x0.5 and x2 of these numbers, +because none of them is measured from real operators yet (see CALIBRATION). + +What each charge covers: + +- ``approval``: read an approval card (the item's values and the save it will + make) and press Approve or Deny. +- ``stop``: read a stop's reason, look at the item in the app, and decide + (resume, skip, or do it by hand). Doing it by hand is charged separately. +- ``reconcile``: the product could not tell whether a save landed. Open the + app, find the record, compare it with the item, answer. +- ``repair_review``: read a proposed repair (the changed step, before and + after screens, probe results) and approve or reject it. +- ``question``: answer a free-text question from an agent. +- ``manual_entry``: do one item entirely by hand in the app (the all-manual + baseline, and the fallback for any item automation leaves undone). +- ``interrupt``: the resumption cost of any interruption, added once per + operator request (switching from other work to the queue and back). +- Setup and maintenance phases are charged at the wall time a person is + attending them, plus fixed preparation where a phase needs it. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, replace +from typing import Any + +CALIBRATION = ( + "No timings from real operators exist yet. Desktop and the console record " + "when a request was raised and when it was answered, but every answer on " + "record was given by a script, so those gaps measure the harness, not a " + "person. Until Desktop has real operator telemetry, the per-touch numbers " + "below are assumptions: keystroke-level estimates for reading a card and " + "clicking (approval), plus looking a record up in the app (stop, " + "reconcile), and a typed-entry estimate for a 12-field registration form " + "(manual entry: about 120 characters at 40 words a minute, plus reading " + "the referral and the duplicate search). The interrupt charge is the " + "resumption lag reported in interruption studies (tens of seconds)." +) + + +@dataclass(frozen=True) +class CostModel: + version: str + approval_s: float + stop_s: float + reconcile_s: float + repair_review_s: float + question_s: float + manual_entry_s: float + interrupt_s: float + #: Preparing a demonstration (pick the example, open the app) before the + #: recorded wall time is charged. + teach_prep_s: float + #: Reviewing a compiled program's parameters and steps before admission. + compile_review_s: float + #: Signing an admission once qualification passed. + admit_s: float + + def per_request(self, kind: str) -> float: + """Charged seconds for one operator request of ``kind``.""" + + base = { + "approval": self.approval_s, + "stop": self.stop_s, + "reconcile": self.reconcile_s, + "repair_review": self.repair_review_s, + "question": self.question_s, + "manual_entry": self.manual_entry_s, + }[kind] + return base + self.interrupt_s + + def scaled(self, factor: float) -> "CostModel": + """Every charge times ``factor`` (the sensitivity bands).""" + + fields = { + k: v * factor + for k, v in asdict(self).items() + if isinstance(v, (int, float)) and not isinstance(v, bool) + } + return replace(self, version=f"{self.version}x{factor:g}", **fields) + + def to_dict(self) -> dict[str, Any]: + return {**asdict(self), "calibration": CALIBRATION} + + +V1 = CostModel( + version="lifecycle-cost/2026-10-20.1", + approval_s=20.0, + stop_s=90.0, + reconcile_s=120.0, + repair_review_s=120.0, + question_s=60.0, + manual_entry_s=180.0, + interrupt_s=30.0, + teach_prep_s=300.0, + compile_review_s=600.0, + admit_s=300.0, +) + +MODELS = {V1.version: V1} +BANDS = (0.5, 1.0, 2.0) diff --git a/packages/flow/benchmark/lifecycle/fake.py b/packages/flow/benchmark/lifecycle/fake.py new file mode 100644 index 000000000..913ac5d99 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/fake.py @@ -0,0 +1,217 @@ +"""An in-memory app and two toy arms, for tests and for checking a new arm. + +``FakeApp`` keeps a patient table in memory and honours the same injected +events as the OpenEMR adapter (a failed save raises, a forced logout signs the +arm out, a UI change is just a flag). ``naive`` types whatever the referral +says and always claims done; ``careful`` asks the operator about anything odd +and stops on a failed save. Neither is a model of a product; they exist so the +harness's bookkeeping can be checked end to end in a second. +""" + +from __future__ import annotations + +import threading +import time +from typing import Any, Optional + +from .arm import Env, ItemResult, OperatorRequest, WorkItem + + +class FakeApp: + ui_url = "fake://app" + api_url = None + + def __init__(self) -> None: + self.rows_: list[dict[str, str]] = [] + self.next_id = 1 + self.fail_saves: dict[str, int] = {} + self.signed_in = True + self.drift_on: list[str] = [] + + def describe(self) -> dict[str, Any]: + return {"app": "fake", "version": "0"} + + def extra(self) -> dict[str, Any]: + return {"app": self} + + def accounts(self) -> list[tuple[str, str]]: + return [("clerk", "synthetic")] + + def reset(self) -> str: + self.rows_, self.next_id, self.fail_saves = [], 1, {} + return "fake-baseline" + + def rows(self) -> list[dict[str, str]]: + return [dict(r) for r in self.rows_] + + def _insert(self, fields: dict[str, str]) -> str: + email = fields.get("email", "") + if self.fail_saves.get(email, 0) > 0: + self.fail_saves[email] -= 1 + raise RuntimeError("save failed") + rid = str(self.next_id) + self.next_id += 1 + row = {"id": rid, "pid": rid, "state": "MA"} + row.update({k: v for k, v in fields.items() if k != "state_label"}) + self.rows_.append(row) + return rid + + def seed_existing(self, people: list[dict[str, str]]) -> list[str]: + return [self._insert(p) for p in people] + + def create_chart(self, fields: dict[str, str], *, who: str) -> str: + return self._insert( + { + **fields, + "fname": fields["fname"].strip(), + "lname": fields["lname"].strip(), + } + ) + + def inject( + self, + inj: dict[str, Any], + truth: Any, + *, + harness: set[str], + stop: Optional[threading.Event] = None, + ) -> dict[str, Any]: + ev = inj["event"] + if ev == "other_clerk_creates": + return { + "event": ev, + "chart_id": self.create_chart(inj["person"], who="clerk"), + } + if ev == "force_logout": + self.signed_in = False + return {"event": ev, "sessions_ended": 1} + if ev == "other_clerk_edits_unrelated": + target = next((r for r in self.rows_ if r["id"] not in harness), None) + if target: + target["phone_home"] = "555-0199" + return {"event": ev, "touched_ids": [target["id"]] if target else []} + if ev == "failed_save": + self.fail_saves[truth.person["email"]] = int(inj.get("times", 1)) + return {"event": ev} + if ev == "slow_save": + return {"event": ev} + raise ValueError(ev) + + def clear_item_events(self) -> list[str]: + self.fail_saves.clear() + return [] + + def apply_drift(self, kind: str) -> dict[str, Any]: + self.drift_on.append(kind) + return {"kind": kind} + + def clear_all_events(self) -> None: + self.drift_on = [] + self.fail_saves.clear() + self.signed_in = True + + # what an arm sees + def find(self, fname: str, lname: str) -> list[dict[str, str]]: + return [ + r + for r in self.rows_ + if r["fname"].lower() == fname.strip().lower() + and r["lname"].lower() == lname.strip().lower() + ] + + +class _ToyArm: + name = "toy" + careful = False + + def __init__(self, options: dict[str, Any]) -> None: + self.options = options + + def describe(self) -> dict[str, Any]: + return {"kind": self.name, "toy": True, "refused_features": []} + + def setup(self, env: Env, operator: Any) -> list[dict[str, Any]]: + return [ + { + "phase": "teach", + "wall_s": 60.0, + "attended_s": 60.0, + "model_calls": 0, + "cost_usd": 0.0, + } + ] + + def after_drift( + self, event: dict[str, Any], env: Env, operator: Any + ) -> list[dict[str, Any]]: + return [] + + def close(self) -> None: + pass + + def run_item(self, item: WorkItem, env: Env, operator: Any) -> ItemResult: + app: FakeApp = env.extra["app"] + t0 = time.time() + p = item.params + if not app.signed_in: + app.signed_in = True # signs in again + if self.careful: + odd = ( + "DOB" not in p + or "/" in p.get("DOB", "") + or p["fname"] != p["fname"].strip() + or app.find(p["fname"], p["lname"]) + ) + if odd: + d = operator.ask( + OperatorRequest( + kind="stop", + item_id=item.item_id, + reason="looks odd", + surface="toy", + detail={"resumable": False}, + ) + ) + return ItemResult( + item.item_id, "stopped", f"stopped: {d.action}", time.time() - t0 + ) + try: + app._insert(dict(p)) + except RuntimeError as exc: + if self.careful: + d = operator.ask( + OperatorRequest( + kind="stop", + item_id=item.item_id, + reason=str(exc), + surface="toy", + detail={"resumable": True}, + ) + ) + if d.action == "resume": + app._insert(dict(p)) + return ItemResult( + item.item_id, "done", "saved after resume", time.time() - t0 + ) + return ItemResult(item.item_id, "stopped", str(exc), time.time() - t0) + return ItemResult( + item.item_id, "done", "saved (error ignored)", time.time() - t0 + ) + return ItemResult(item.item_id, "done", "saved", time.time() - t0) + + +class NaiveArm(_ToyArm): + name = "naive" + + +class CarefulArm(_ToyArm): + name = "careful" + careful = True + + +def naive(options: dict[str, Any]) -> NaiveArm: + return NaiveArm(options) + + +def careful(options: dict[str, Any]) -> CarefulArm: + return CarefulArm(options) diff --git a/packages/flow/benchmark/lifecycle/metrics.py b/packages/flow/benchmark/lifecycle/metrics.py new file mode 100644 index 000000000..65fcb98c4 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/metrics.py @@ -0,0 +1,271 @@ +"""The metric, computed only from the results ledger. + +Headline: verified completed items per operator-hour. + +- A *verified completed item* is one whose final state the independent + oracle judges right (the intended chart exactly once, or no new chart when + none is wanted) and that the operator did not have to do by hand. +- *Operator hours* are every charged second: setup, per-item requests + (approvals, stops, reconciliations, repair reviews, questions, items done by + hand) and maintenance after each UI change. + +Charges are recomputed here from the ledger's request kinds and the cost +model, so the sensitivity bands (x0.5, x2) use the same rows. + +Outcome of an item, from the oracle's read after the arm returned (before any +by-hand fix), crossed with what the arm claimed: + +===================== ================================================== +``correct`` right end state, arm claimed done +``right_stop`` no chart wanted, arm stopped, nothing written +``false_stop`` chart wanted, item data right, no item-level fault, + arm stopped with nothing written (split by whether a + UI change was in force) +``fault_stop`` chart wanted, an item-level fault was injected + (expiry, slow or failed save, concurrent edit), arm + stopped with nothing written +``stopped_saved`` arm stopped but the right chart is there +``silent_wrong`` arm claimed done; the oracle disagrees +``flagged_wrong`` a wrong write landed and the arm stopped +``error`` the arm crashed; judged like a stop +===================== ================================================== +""" + +from __future__ import annotations + +import math +from collections import Counter, defaultdict +from typing import Any, Iterable, Optional + +from .cost_model import BANDS, CostModel + +ITEM_FAULTS = { + "session_expiry", + "slow_save", + "failed_save", + "concurrent_same", + "concurrent_other", +} +DATA_EXCEPTIONS = { + "dup_existing", + "dup_resend", + "ambiguous_same_person", + "ambiguous_new_person", + "missing_required", + "missing_optional", + "malformed_dob", + "malformed_case", +} + + +def classify(*, claimed: str, oracle: str, chart_wanted: bool, exception: str) -> str: + """One item's outcome. ``oracle`` is "correct", "absent", "wrong" or + "duplicate" (for no-chart items "correct" means no new chart).""" + + done = claimed == "done" + if oracle == "correct": + if done: + return "correct" + return "stopped_saved" if chart_wanted else "right_stop" + if not chart_wanted: + # a new chart landed where none was wanted + return "silent_wrong" if done else "flagged_wrong" + if oracle == "absent": + if done: + return "silent_wrong" + if exception in ITEM_FAULTS: + return "fault_stop" + if exception in DATA_EXCEPTIONS: + return "data_stop" + return "false_stop" + return "silent_wrong" if done else "flagged_wrong" + + +def wilson(k: int, n: int, z: float = 1.96) -> Optional[tuple[float, float]]: + if n == 0: + return None + p = k / n + d = 1 + z * z / n + c = p + z * z / (2 * n) + r = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) + return (round(max(0.0, (c - r) / d), 4), round(min(1.0, (c + r) / d), 4)) + + +def _charge(cost: CostModel, row: dict[str, Any]) -> float: + s = cost.per_request(row["kind"]) + if ( + row["action"] == "do_by_hand" + and row["kind"] != "manual_entry" + and row.get("by_hand") + ): + s += cost.manual_entry_s + return s + + +def _phase_charge(cost: CostModel, row: dict[str, Any]) -> float: + fixed = { + "teach": cost.teach_prep_s, + "compile": cost.compile_review_s, + "admit": cost.admit_s, + }.get(row["phase"], 0.0) + return float(row.get("attended_s", 0.0)) + fixed + + +def compute(rows: Iterable[dict[str, Any]], cost: CostModel) -> dict[str, Any]: + rows = list(rows) + items = [r for r in rows if r["type"] == "item_result"] + ops = [r for r in rows if r["type"] == "operator"] + setup = [r for r in rows if r["type"] == "phase" and r.get("stage") == "setup"] + maint = [ + r for r in rows if r["type"] == "phase" and r.get("stage") == "maintenance" + ] + drifts = [r for r in rows if r["type"] == "drift_applied"] + + by_hand = { + r["item_id"] for r in ops if r["action"] == "do_by_hand" and r.get("item_id") + } + outcomes = Counter(r["outcome"] for r in items) + final_ok = [r for r in items if r["final_oracle"] == "correct"] + completed = [ + r + for r in final_ok + if r["item_id"] not in by_hand and r["outcome"] != "silent_wrong" + ] + + # operator requests attributed to maintenance: repair reviews and any + # request raised in the first `maintenance_window` items after a drift + item_ops = [r for r in ops if r["kind"] != "repair_review"] + repair_ops = [r for r in ops if r["kind"] == "repair_review"] + + silent = [r for r in items if r["outcome"] == "silent_wrong"] + + def bands() -> dict[str, Any]: + out = {} + for factor in BANDS: + c = cost.scaled(factor) if factor != 1.0 else cost + setup_s = sum(_phase_charge(c, r) for r in setup) + maint_s = sum(_phase_charge(c, r) for r in maint) + sum( + _charge(c, r) for r in repair_ops + ) + item_s = sum(_charge(c, r) for r in item_ops) + hours = (setup_s + maint_s + item_s) / 3600.0 + # Silent wrongs cost nobody anything during the run. If each is + # found later, someone must look it up and put it right. + cleanup_s = len(silent) * (c.per_request("reconcile") + c.manual_entry_s) + hours_cleanup = hours + cleanup_s / 3600.0 + n_all = len(items) + manual_hours = n_all * c.manual_entry_s / 3600.0 + out[f"x{factor:g}"] = { + "setup_min": round(setup_s / 60, 2), + "maintenance_min": round(maint_s / 60, 2), + "per_item_operator_min_total": round(item_s / 60, 2), + "per_item_operator_min": round(item_s / 60 / n_all, 3) + if n_all + else None, + "operator_hours": round(hours, 3), + "verified_completed_per_operator_hour": ( + round(len(completed) / hours, 2) if hours else None + ), + "operator_hours_with_cleanup": round(hours_cleanup, 3), + "verified_completed_per_operator_hour_with_cleanup": ( + round(len(completed) / hours_cleanup, 2) if hours_cleanup else None + ), + "manual_only_items_per_hour": ( + round(3600.0 / c.manual_entry_s, 2) if c.manual_entry_s else None + ), + "manual_only_hours": round(manual_hours, 3), + "work_removed_fraction": ( + round(1 - hours / manual_hours, 3) if manual_hours else None + ), + } + return out + + stops = [r for r in items if r["claimed"] != "done"] + wanted = [r for r in items if r["chart_wanted"]] + clean_wanted = [ + r + for r in wanted + if r["exception"] not in ITEM_FAULTS and r["exception"] not in DATA_EXCEPTIONS + ] + false_stops = [r for r in items if r["outcome"] == "false_stop"] + done_claims = [r for r in items if r["claimed"] == "done"] + silent = [r for r in items if r["outcome"] == "silent_wrong"] + + per_drift: dict[str, Any] = {} + for d in drifts: + kind = d["kind"] + window = [r for r in items if r["seq"] >= d["first_seq"]] + nxt = [x["first_seq"] for x in drifts if x["first_seq"] > d["first_seq"]] + if nxt: + window = [r for r in window if r["seq"] < min(nxt)] + per_drift[kind] = { + "first_seq": d["first_seq"], + "items": len(window), + "outcomes": dict(Counter(r["outcome"] for r in window)), + "repair_reviews": sum( + 1 + for r in repair_ops + if window and window[0]["seq"] <= r.get("seq", -1) <= window[-1]["seq"] + ), + } + + wall = [r["wall_s"] for r in items if r.get("wall_s") is not None] + loads = [r["load_1m"] for r in items if r.get("load_1m") is not None] + return { + "cost_model": cost.version, + "items": len(items), + "outcomes": dict(outcomes), + "verified_completed": len(completed), + "final_correct": len(final_ok), + "done_by_hand": len(by_hand), + "operator_requests": dict(Counter(r["kind"] for r in ops)), + "operator_actions": dict(Counter(r["action"] for r in ops)), + "false_stop_rate": { + "k": len(false_stops), + "n": len(clean_wanted), + "rate": round(len(false_stops) / len(clean_wanted), 4) + if clean_wanted + else None, + "ci95": wilson(len(false_stops), len(clean_wanted)), + "with_ui_change_in_force": sum( + 1 for r in false_stops if r.get("drift_active") + ), + "definition": "stops with nothing written on items whose data was right and " + "that carried no item-level fault, over all such items", + }, + "silent_wrong_rate": { + "k": len(silent), + "n": len(done_claims), + "rate": round(len(silent) / len(done_claims), 4) if done_claims else None, + "ci95": wilson(len(silent), len(done_claims)), + "of_all_items": round(len(silent) / len(items), 4) if items else None, + "definition": "items the arm claimed done that the independent oracle judged wrong, " + "over all items the arm claimed done", + }, + "stops": len(stops), + "by_exception": { + k: dict(Counter(r["outcome"] for r in items if r["exception"] == k)) + for k in sorted({r["exception"] for r in items}) + }, + "per_drift": per_drift, + "wall": { + "total_s": round(sum(wall), 1), + "median_item_s": round(sorted(wall)[len(wall) // 2], 2) if wall else None, + "load_1m_median": round(sorted(loads)[len(loads) // 2], 1) + if loads + else None, + "load_1m_max": round(max(loads), 1) if loads else None, + }, + "model": { + "calls": sum(r.get("model_calls", 0) for r in items), + "cost_usd": round(sum(r.get("cost_usd", 0.0) for r in items), 4), + }, + "bands": bands(), + } + + +def group_by_arm(rows: Iterable[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: + out: dict[str, list[dict[str, Any]]] = defaultdict(list) + for r in rows: + out[r.get("arm", "")].append(r) + return dict(out) diff --git a/packages/flow/benchmark/lifecycle/operator.py b/packages/flow/benchmark/lifecycle/operator.py new file mode 100644 index 000000000..5c0ce5339 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/operator.py @@ -0,0 +1,145 @@ +"""The simulated operator: a scripted, deterministic policy over ground truth. + +The operator answers every request an arm raises. It decides from the event +ledger's truth for the item and from what it can see in the app (the same +read-only look a person would take), never from the arm's own verdict. Each +answer is charged from the cost model and written to the results ledger with +the rule that decided it. + +The policy, in order: + +- ``approval``: approve when the item should produce a chart and the values + on the card (when the card shows any) match the truth; otherwise deny. +- ``stop`` / ``reconcile``: + - the item should produce no new chart: skip it (return to sender); + - the right chart is already there: confirm it is saved; + - the arm offers a resume and this item has not been resumed yet: resume; + - otherwise do the item by hand. +- ``repair_review``: accept when a UI change is in force and the repair's + probes passed; otherwise reject. +- ``question``: answer from the truth's rationale. +- ``manual_entry``: do the item by hand. + +An optional model operator can be added beside this one; it must be labelled +as such in the ledger (``operator.kind``). +""" + +from __future__ import annotations + +from typing import Any, Callable, Optional + +from .arm import OperatorDecision, OperatorRequest +from .cost_model import CostModel +from .workload import Truth + +NAME_FIELDS = ("fname", "lname") + + +def values_match(shown: dict[str, str], intended: dict[str, str]) -> bool: + """Do the values on a card match the truth (names compared trimmed and + case-insensitively, as a person reads them)?""" + + for key, want in intended.items(): + if key not in shown: + continue + got = shown[key] + if key in NAME_FIELDS: + if got.strip().lower() != want.strip().lower(): + return False + elif got != want: + return False + return True + + +class ScriptedOperator: + kind = "scripted" + + def __init__( + self, + cost: CostModel, + truth: dict[str, Truth], + *, + look: Callable[[Truth], str], + do_by_hand: Callable[[Truth], dict[str, Any]], + drift_active: Callable[[], list[str]], + record: Callable[[dict[str, Any]], None], + ) -> None: + """``look(truth)`` returns "correct", "absent" or "wrong": what a person + sees when they search the app for this item's patient. ``do_by_hand`` + enters the item as the operator would (and records that it did). + ``record`` appends a row to the results ledger.""" + + self.cost = cost + self.truth = truth + self.look = look + self.do_by_hand = do_by_hand + self.drift_active = drift_active + self.record = record + self.resumed: set[str] = set() + self.by_hand: set[str] = set() + self.skipped: set[str] = set() + + def _decide(self, req: OperatorRequest) -> tuple[str, str, Optional[str]]: + t = self.truth.get(req.item_id or "") + if req.kind == "repair_review": + ok = bool(req.detail.get("probes_passed")) + if self.drift_active() and ok: + return "accept_repair", "repair_drift_and_probes_passed", None + return "reject_repair", "repair_no_drift_or_probe_failed", None + if t is None: + return "deny", "no_truth_for_item", None + if req.kind == "approval": + shown = req.detail.get("values") or {} + if t.intended is not None and values_match(shown, t.intended): + return "approve", "approve_intended_write", None + return "deny", "deny_unintended_write", None + if req.kind == "question": + return "answer", "answer_from_truth", t.rationale + if req.kind == "manual_entry": + return "do_by_hand", "manual_entry_requested", None + # stop or reconcile + if t.intended is None: + return "skip", "skip_no_chart_wanted", None + seen = self.look(t) + if seen == "correct": + return "confirm_saved", "chart_already_correct", None + if seen == "wrong": + return "do_by_hand", "fix_wrong_chart_by_hand", None + if req.detail.get("resumable") and req.item_id not in self.resumed: + return "resume", "resume_once", None + return "do_by_hand", "enter_by_hand", None + + def ask(self, req: OperatorRequest) -> OperatorDecision: + action, rule, answer = self._decide(req) + charged = self.cost.per_request(req.kind) + extra: dict[str, Any] = {} + if action == "resume" and req.item_id: + self.resumed.add(req.item_id) + if action == "skip" and req.item_id: + self.skipped.add(req.item_id) + if action == "do_by_hand" and req.item_id and req.item_id not in self.by_hand: + self.by_hand.add(req.item_id) + t = self.truth[req.item_id] + if req.kind != "manual_entry": + charged += self.cost.manual_entry_s + extra = self.do_by_hand(t) + decision = OperatorDecision( + action=action, charged_s=charged, rule=rule, answer=answer + ) + self.record( + { + "type": "operator", + "operator": self.kind, + "item_id": req.item_id, + "kind": req.kind, + "reason": req.reason, + "surface": req.surface, + "action": action, + "rule": rule, + "charged_s": charged, + "cost_model": self.cost.version, + "detail": {k: v for k, v in req.detail.items() if k != "values"}, + **({"by_hand": extra} if extra else {}), + } + ) + return decision diff --git a/packages/flow/benchmark/lifecycle/oracle.py b/packages/flow/benchmark/lifecycle/oracle.py new file mode 100644 index 000000000..ad06b11f7 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/oracle.py @@ -0,0 +1,185 @@ +"""The independent oracle: reads OpenEMR's database directly, never the arm. + +It runs as the database root inside the fixture's own database container, a +channel no arm uses. It judges an item twice: + +- right after the arm returns (the arm's own outcome), counting only charts + the arm could have written: rows the harness wrote itself (another clerk, + the operator by hand) are attributed by row id and left out; +- after the operator has dealt with the item (the item's final state), over + every chart for that person. + +A chart matches the truth when every field matches: names trimmed and +compared without case (as a person reads a name), every other field exactly. +""" + +from __future__ import annotations + +import hashlib +from typing import Any, Callable, Iterable, Optional + +from .workload import Truth + +FIELDS = ( + "id", + "pid", + "title", + "fname", + "lname", + "DOB", + "sex", + "street", + "city", + "state", + "postal_code", + "phone_home", + "email", + "country_code", +) +#: Truth key -> patient_data column. +COLUMN = { + "title": "title", + "fname": "fname", + "lname": "lname", + "DOB": "DOB", + "sex": "sex", + "street": "street", + "city": "city", + "postal_code": "postal_code", + "phone_home": "phone_home", + "email": "email", + "country_code": "country_code", +} +STATE_CODE = {"Massachusetts": "MA"} +NAME_KEYS = {"fname", "lname"} + +Rows = list[dict[str, str]] + + +def norm_name(v: str) -> str: + return " ".join(v.split()).lower() + + +def field_diffs(row: dict[str, str], intended: dict[str, str]) -> list[dict[str, str]]: + diffs = [] + for key, col in COLUMN.items(): + want = intended.get(key, "") + got = row.get(col, "") + same = norm_name(got) == norm_name(want) if key in NAME_KEYS else got == want + if not same: + diffs.append({"field": key, "want": want, "got": got}) + want_state = STATE_CODE.get(intended.get("state_label", ""), "") + if row.get("state", "") != want_state: + diffs.append( + {"field": "state", "want": want_state, "got": row.get("state", "")} + ) + return diffs + + +def same_person(row: dict[str, str], person: dict[str, str]) -> bool: + return ( + norm_name(row.get("fname", "")) == norm_name(person["fname"]) + and norm_name(row.get("lname", "")) == norm_name(person["lname"]) + and row.get("DOB", "") == person.get("DOB", "") + ) + + +def _names(row: dict[str, str]) -> tuple[str, str]: + return norm_name(row.get("fname", "")), norm_name(row.get("lname", "")) + + +def rows_digest(rows: Iterable[dict[str, str]]) -> str: + h = hashlib.sha256() + for r in sorted(rows, key=lambda r: int(r["id"])): + h.update(("\t".join(r.get(k, "") for k in FIELDS) + "\n").encode()) + return h.hexdigest() + + +class Oracle: + def __init__(self, sql_rows: Callable[[], Rows]) -> None: + self.sql_rows = sql_rows + + def snapshot(self) -> Rows: + return self.sql_rows() + + def judge_arm( + self, + truth: Truth, + before: Rows, + after: Rows, + *, + harness_ids: set[str], + harness_touched: set[str], + ) -> dict[str, Any]: + """The arm's outcome: "correct", "absent", "wrong" or "duplicate".""" + + before_ids = {r["id"] for r in before} + new = [ + r for r in after if r["id"] not in before_ids and r["id"] not in harness_ids + ] + old_before = {r["id"]: r for r in before if r["id"] not in harness_touched} + old_after = {r["id"]: r for r in after if r["id"] in old_before} + collateral = sorted(i for i in old_before if old_before[i] != old_after.get(i)) + detail: dict[str, Any] = { + "new_charts": len(new), + "collateral_ids": collateral, + "new_ids": [r["id"] for r in new], + } + if collateral: + return {"verdict": "wrong", "why": "collateral", **detail} + if truth.intended is None: + if new: + return { + "verdict": "wrong", + "why": "chart written where none wanted", + **detail, + } + return {"verdict": "correct", "why": "no new chart", **detail} + if not new: + return {"verdict": "absent", "why": "no new chart", **detail} + if len(new) > 1: + return {"verdict": "duplicate", "why": f"{len(new)} new charts", **detail} + diffs = field_diffs(new[0], truth.intended) + if diffs: + return { + "verdict": "wrong", + "why": "field mismatch", + "diffs": diffs, + **detail, + } + return {"verdict": "correct", "why": "one chart, every field right", **detail} + + def judge_final( + self, truth: Truth, before_item: Rows, now: Rows, *, other_clerk_ids: set[str] + ) -> dict[str, Any]: + """The item's final state, whoever wrote it (other clerks aside). + + Every chart written since the item started that carries this + person's name or email counts, so a chart with a mistyped date of + birth is still found.""" + + prior = {r["id"] for r in before_item} + new = [ + r + for r in now + if r["id"] not in prior + and r["id"] not in other_clerk_ids + and ( + _names(r) == _names(truth.person) + or r.get("email") == truth.person["email"] + ) + ] + if truth.intended is None: + return {"verdict": "wrong" if new else "correct", "new_charts": len(new)} + if len(new) == 1 and not field_diffs(new[0], truth.intended): + return {"verdict": "correct", "new_charts": 1} + return {"verdict": "absent" if not new else "wrong", "new_charts": len(new)} + + def look(self, truth: Truth, before_item: Optional[Rows] = None) -> str: + """What a person sees searching the app for this item's patient: + "correct", "absent" or "wrong".""" + + v = self.judge_final( + truth, before_item or [], self.sql_rows(), other_clerk_ids=set() + ) + return v["verdict"] diff --git a/packages/flow/benchmark/lifecycle/report.py b/packages/flow/benchmark/lifecycle/report.py new file mode 100644 index 000000000..90e97705d --- /dev/null +++ b/packages/flow/benchmark/lifecycle/report.py @@ -0,0 +1,179 @@ +"""Compute ``results.json`` and the one-page ``REPORT.md`` from a run's ledger. + + python -m lifecycle.report lifecycle/results/ + +Everything is recomputed from ``ledger.jsonl``; nothing else is read. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any + +from .cost_model import BANDS, CALIBRATION, MODELS +from .metrics import compute, group_by_arm + + +def load(run_dir: Path) -> list[dict[str, Any]]: + return [ + json.loads(line) + for line in (run_dir / "ledger.jsonl").read_text().splitlines() + if line.strip() + ] + + +def results(run_dir: Path) -> dict[str, Any]: + rows = load(run_dir) + start = next(r for r in rows if r["type"] == "run_start") + cost = MODELS[start["cost_model"]["version"]] + arms = {} + for arm, arm_rows in group_by_arm(rows).items(): + if not arm: + continue + begin = next((r for r in arm_rows if r["type"] == "arm_start"), {}) + m = compute(arm_rows, cost) + m["describe"] = begin.get("describe", {}) + m["complete"] = any(r["type"] == "arm_end" for r in arm_rows) + arms[arm] = m + return { + "run": {k: v for k, v in start.items() if k not in ("t", "arm")}, + "arms": arms, + } + + +def _pct(x: Any) -> str: + return "n/a" if x is None else f"{100 * x:.1f}%" + + +def _ci(ci: Any) -> str: + return "" if not ci else f" (95% CI {100 * ci[0]:.1f}-{100 * ci[1]:.1f}%)" + + +def render(res: dict[str, Any]) -> str: + run = res["run"] + lines = [ + "# Lifecycle soak report", + "", + f"Protocol `{run['protocol']}`, workload `{run['workload']}`, seed {run['seed']}, " + f"{run['items']} referrals over {run['days']} simulated days" + + (f" (first {run['stop_after']} run)" if run.get("stop_after") else "") + + f". Engine commit `{run['engine_commit'][:12]}`. App: {run['app'].get('app', '')} " + f"{run['app'].get('version', '')}, synthetic data only. " + f"Cost model `{run['cost_model']['version']}`.", + "", + ] + for name, m in res["arms"].items(): + b = m["bands"] + lines += [ + f"## Arm `{name}`" + ("" if m["complete"] else " (incomplete)"), + "", + "| | x0.5 | x1 | x2 |", + "|---|---:|---:|---:|", + ] + rows = [ + ( + "Verified completed items per operator-hour", + "verified_completed_per_operator_hour", + ), + ( + "... if every silent wrong is later found and fixed", + "verified_completed_per_operator_hour_with_cleanup", + ), + ("All-manual items per operator-hour", "manual_only_items_per_hour"), + ("Operator work removed vs all-manual", "work_removed_fraction"), + ("Setup minutes", "setup_min"), + ("Per-item operator minutes (mean)", "per_item_operator_min"), + ("Maintenance minutes (all UI changes)", "maintenance_min"), + ("Operator hours in total", "operator_hours"), + ] + for label, key in rows: + vals = [] + for f in BANDS: + v = b[f"x{f:g}"][key] + vals.append( + _pct(v) + if key == "work_removed_fraction" + else ("n/a" if v is None else f"{v:g}") + ) + lines.append(f"| {label} | " + " | ".join(vals) + " |") + fs, sw = m["false_stop_rate"], m["silent_wrong_rate"] + lines += [ + "", + f"- Items: {m['items']}; verified completed without hand entry: " + f"{m['verified_completed']}; right in the end: {m['final_correct']}; " + f"done by hand: {m['done_by_hand']}.", + f"- False stops: {fs['k']} of {fs['n']} clean items, {_pct(fs['rate'])}{_ci(fs['ci95'])}; " + f"{fs['with_ui_change_in_force']} of them with a UI change in force.", + f"- Silent wrong: {sw['k']} of {sw['n']} items claimed done, {_pct(sw['rate'])}{_ci(sw['ci95'])} " + "(judged by direct SQL, never the product's verdict).", + f"- Operator requests: {json.dumps(m['operator_requests'])}; actions: " + f"{json.dumps(m['operator_actions'])}.", + f"- Outcomes: {json.dumps(m['outcomes'])}.", + f"- Wall: {m['wall']['total_s']} s of item time, median {m['wall']['median_item_s']} s " + f"an item; host 1-minute load median {m['wall']['load_1m_median']}, max " + f"{m['wall']['load_1m_max']}.", + f"- Model calls {m['model']['calls']}, list-price cost ${m['model']['cost_usd']}.", + "", + "Outcome by exception:", + "", + "| Exception | Outcomes |", + "|---|---|", + ] + for k, v in m["by_exception"].items(): + lines.append( + f"| {k} | {', '.join(f'{a} {n}' for a, n in sorted(v.items()))} |" + ) + if m["per_drift"]: + lines += [ + "", + "After each UI change (items until the next change):", + "", + "| Change | From item | Items | Outcomes |", + "|---|---:|---:|---|", + ] + for k, v in m["per_drift"].items(): + lines.append( + f"| {k} | {v['first_seq']} | {v['items']} | " + f"{', '.join(f'{a} {n}' for a, n in sorted(v['outcomes'].items()))} |" + ) + refused = m["describe"].get("refused_features") + if refused is not None: + lines += [ + "", + "Features the compatibility matrix refused for this configuration:", + "", + ] + lines += [f"- `{r['feature']}`: {r['reason']}" for r in refused] or [ + "- none" + ] + lines.append("") + lines += [ + "## Assumptions", + "", + f"Operator seconds per touch (`{run['cost_model']['version']}`): " + + ", ".join( + f"{k.removesuffix('_s')} {v:g}" + for k, v in run["cost_model"].items() + if k.endswith("_s") + ) + + ". Every request also pays the interrupt charge. Bands scale every number.", + "", + CALIBRATION, + "", + ] + return "\n".join(lines) + + +def write_report(run_dir: Path) -> dict[str, Any]: + res = results(run_dir) + (run_dir / "results.json").write_text( + json.dumps(res, indent=2, sort_keys=True) + "\n" + ) + (run_dir / "REPORT.md").write_text(render(res)) + return res + + +if __name__ == "__main__": + write_report(Path(sys.argv[1])) diff --git a/packages/flow/benchmark/lifecycle/run.py b/packages/flow/benchmark/lifecycle/run.py new file mode 100644 index 000000000..e64088388 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/run.py @@ -0,0 +1,385 @@ +"""Run the lifecycle soak: one seeded worklist, every arm, one results ledger. + + cd packages/flow/benchmark + python -m lifecycle.run --app openemr --seed 7 --items 300 --days 10 \\ + --arm lifecycle.arms.openadapt:make --out lifecycle/results/ + +Each arm gets the app reset to the same baseline with the same existing +charts, then the same stream of items, UI changes and faults. The harness +applies every event, asks the oracle after every item, and writes every row +to ``ledger.jsonl``; ``results.json`` and ``REPORT.md`` are computed from that +file alone (``python -m lifecycle.report `` recomputes them). +""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import subprocess +import threading +import time +from pathlib import Path +from typing import Any, Callable, Optional + +from .arm import INTERFACE_VERSION, Env, ItemResult, OperatorRequest, load_arm +from .cost_model import MODELS, V1 +from .metrics import classify +from .operator import ScriptedOperator +from .oracle import Oracle +from .workload import WORKLOAD_VERSION, Truth, Workload, generate + +HERE = Path(__file__).resolve().parent +PROTOCOL_VERSION = "lifecycle/2026-10-20.1" + + +def load_1m() -> Optional[float]: + try: + return round(os.getloadavg()[0], 1) + except OSError: + return None + + +class Ledger: + """Append-only JSON lines; every row is flushed as it is written.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._lock = threading.Lock() + self._fh = path.open("a", encoding="utf-8") + self.arm = "" + + def write(self, row: dict[str, Any]) -> None: + row = {"t": round(time.time(), 3), "arm": self.arm, **row} + with self._lock: + self._fh.write(json.dumps(row, sort_keys=True, default=str) + "\n") + self._fh.flush() + + def close(self) -> None: + self._fh.close() + + +class _CountingOperator: + """Wraps the operator to know which items already reached a person.""" + + def __init__(self, inner: ScriptedOperator) -> None: + self.inner = inner + self.asked: dict[str, int] = {} + + def ask(self, req: OperatorRequest) -> Any: + if req.item_id: + self.asked[req.item_id] = self.asked.get(req.item_id, 0) + 1 + return self.inner.ask(req) + + +def engine_commit() -> str: + try: + return subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=HERE, + capture_output=True, + text=True, + check=True, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError): + return "unknown" + + +def run_arm( + *, + arm_spec: str, + arm_options: dict[str, Any], + app: Any, + workload: Workload, + ledger: Ledger, + out: Path, + cost_name: str, + stop_after: Optional[int] = None, + say: Callable[[str], None] = print, +) -> None: + arm = load_arm(arm_spec, arm_options) + ledger.arm = arm.name + truth = workload.truth + cost = MODELS[cost_name] + + say(f"[{arm.name}] resetting the app to its baseline") + app.reset() + app.clear_all_events() + existing_ids = app.seed_existing(workload.existing) + ledger.write( + { + "type": "arm_start", + "arm_spec": arm_spec, + "options": arm_options, + "describe": arm.describe(), + "existing_ids": existing_ids, + "interface": INTERFACE_VERSION, + } + ) + + oracle = Oracle(app.rows) + harness_ids: set[str] = set() + harness_touched: set[str] = set() + other_clerk_ids: set[str] = set() + active_drift: list[str] = [] + current: dict[str, Any] = {"before": None, "seq": 0} + + def do_by_hand(t: Truth) -> dict[str, Any]: + """The operator enters the item in the app themself.""" + if t.intended is None: + return {} + new_id = app.create_chart(t.intended, who="operator") + harness_ids.add(new_id) + return {"chart_id": new_id} + + def record(row: dict[str, Any]) -> None: + ledger.write({**row, "seq": current["seq"]}) + + scripted = ScriptedOperator( + cost, + truth, + look=lambda t: oracle.look(t, current["before"]), + do_by_hand=do_by_hand, + drift_active=lambda: list(active_drift), + record=record, + ) + operator = _CountingOperator(scripted) + env = Env( + app=workload.items[0].app, + ui_url=app.ui_url, + api_url=app.api_url, + accounts=tuple(app.accounts()), + workdir=out / "arms" / arm.name, + seed=workload.seed, + extra=app.extra(), + ) + env.workdir.mkdir(parents=True, exist_ok=True) + + say(f"[{arm.name}] setup") + for phase in arm.setup(env, operator): + ledger.write({"type": "phase", "stage": "setup", **phase}) + + drift_by_seq = {d["first_seq"]: d for d in workload.drift} + items = workload.items[: stop_after or len(workload.items)] + for item in items: + current["seq"] = item.seq + t = truth[item.item_id] + if item.seq in drift_by_seq: + d = drift_by_seq[item.seq] + applied = app.apply_drift(d["kind"]) + active_drift.append(d["kind"]) + ledger.write({"type": "drift_applied", **d, "applied": applied}) + say( + f"[{arm.name}] day {d['day']}: UI change '{d['kind']}' from item {item.seq}" + ) + for phase in arm.after_drift(d, env, operator): + ledger.write( + { + "type": "phase", + "stage": "maintenance", + "drift": d["kind"], + **phase, + } + ) + + before = oracle.snapshot() + current["before"] = before + injected: list[dict[str, Any]] = [] + for inj in t.inject: + if inj["when"] == "before_item": + injected.append(app.inject(inj, t, harness=harness_ids)) + for inj in injected: + if inj.get("chart_id"): + other_clerk_ids.add(inj["chart_id"]) + harness_ids.add(inj["chart_id"]) + stop_during = threading.Event() + during: list[threading.Thread] = [] + during_out: list[dict[str, Any]] = [] + for inj in t.inject: + if inj["when"] == "during_item": + th = threading.Thread( + target=lambda i=inj: during_out.append( + app.inject(i, t, harness=harness_ids, stop=stop_during) + ), + daemon=True, + ) + th.start() + during.append(th) + + load_start = load_1m() + started = time.time() + try: + result = arm.run_item(item, env, operator) + except Exception as exc: # noqa: BLE001 - an arm crash is a result + if type(exc).__name__ == "SpendCapReached": + stop_during.set() + app.clear_item_events() + ledger.write( + { + "type": "arm_paused", + "seq": item.seq, + "reason": str(exc)[:300], + "paused_items": len(items) - item.seq + 1, + } + ) + say(f"[{arm.name}] spend cap reached at item {item.seq}; arm paused") + break + result = ItemResult( + item_id=item.item_id, + claimed="error", + product_status=f"{type(exc).__name__}: {exc}"[:500], + wall_s=time.time() - started, + ) + stop_during.set() + for th in during: + th.join(timeout=60) + cleared = app.clear_item_events() + for phase in result.phases: + ledger.write( + { + "type": "phase", + "stage": "maintenance", + "seq": item.seq, + "within_item": item.item_id, + **phase, + } + ) + for d in during_out: + harness_touched.update(d.get("touched_ids", [])) + + after = oracle.snapshot() + verdict = oracle.judge_arm( + t, before, after, harness_ids=harness_ids, harness_touched=harness_touched + ) + outcome = classify( + claimed=result.claimed, + oracle=verdict["verdict"], + chart_wanted=t.intended is not None, + exception=t.exception, + ) + # A stop the arm never put in front of a person still needs one. + if result.claimed != "done" and not operator.asked.get(item.item_id): + operator.ask( + OperatorRequest( + kind="stop", + item_id=item.item_id, + reason=result.product_status[:300], + surface="harness:stopped-item queue", + detail={"resumable": False}, + ) + ) + final = oracle.judge_final( + t, before, oracle.snapshot(), other_clerk_ids=other_clerk_ids + ) + row = { + "type": "item_result", + "seq": item.seq, + "item_id": item.item_id, + "day": item.day, + "exception": t.exception, + "chart_wanted": t.intended is not None, + "drift_active": list(active_drift), + "claimed": result.claimed, + "product_status": result.product_status, + "oracle": verdict, + "outcome": outcome, + "final_oracle": final["verdict"], + "final": final, + "wall_s": round(result.wall_s, 2), + "harness_wall_s": round(time.time() - started, 2), + "model_calls": result.model_calls, + "cost_usd": result.cost_usd, + "evidence": result.evidence, + "detail": result.detail, + "load_1m": load_start, + "load_1m_end": load_1m(), + "injected": injected + during_out, + "cleared": cleared, + } + ledger.write(row) + say( + f"[{arm.name}] {item.seq:4d} {item.item_id} {t.exception:22s} " + f"claimed={result.claimed:7s} oracle={verdict['verdict']:9s} -> {outcome:13s} " + f"final={final['verdict']:8s} {result.wall_s:6.1f}s load={load_start}" + ) + arm.close() + app.clear_all_events() + ledger.write({"type": "arm_end"}) + + +def main(argv: Optional[list[str]] = None) -> int: + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("--app", default="openemr", choices=("openemr", "fake")) + p.add_argument( + "--arm", + action="append", + required=True, + help="module:factory; repeat for several arms", + ) + p.add_argument("--arm-options", default="{}", help="JSON options for every arm") + p.add_argument("--seed", type=int, default=7) + p.add_argument("--items", type=int, default=300) + p.add_argument("--days", type=int, default=10) + p.add_argument("--existing", type=int, default=12) + p.add_argument( + "--stop-after", + type=int, + default=None, + help="run only the first N items (a smoke run)", + ) + p.add_argument("--cost-model", default=V1.version, choices=sorted(MODELS)) + p.add_argument("--out", type=Path, required=True) + args = p.parse_args(argv) + + out: Path = args.out + out.mkdir(parents=True, exist_ok=False) + workload = generate(args.seed, args.items, args.days, args.existing) + (out / "workload.jsonl").write_text( + "".join(json.dumps(r, sort_keys=True) + "\n" for r in workload.ledger_rows()) + ) + from .apps import make_app + + app = make_app(args.app) + ledger = Ledger(out / "ledger.jsonl") + ledger.write( + { + "type": "run_start", + "protocol": PROTOCOL_VERSION, + "workload": WORKLOAD_VERSION, + "seed": args.seed, + "items": args.items, + "days": args.days, + "stop_after": args.stop_after, + "cost_model": MODELS[args.cost_model].to_dict(), + "engine_commit": engine_commit(), + "host": platform.node().split(".")[0], + "platform": platform.platform(), + "cpus": os.cpu_count(), + "app": app.describe(), + "argv": argv if argv is not None else None, + } + ) + try: + for spec in args.arm: + run_arm( + arm_spec=spec, + arm_options=json.loads(args.arm_options), + app=app, + workload=workload, + ledger=ledger, + out=out, + cost_name=args.cost_model, + stop_after=args.stop_after, + ) + finally: + ledger.write({"type": "run_end"}) + ledger.close() + from .report import write_report + + write_report(out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/benchmark/lifecycle/workload.py b/packages/flow/benchmark/lifecycle/workload.py new file mode 100644 index 000000000..c10b8d16c --- /dev/null +++ b/packages/flow/benchmark/lifecycle/workload.py @@ -0,0 +1,341 @@ +"""A seeded worklist stream with an exception mix, and its ground truth. + +Every item is a referral a clerk would register as a new patient. Every +exception is drawn from a seeded generator and written to the event ledger +before the run starts, so the operator and the oracle judge from the same +truth and two runs with one seed see the same stream. + +The mix is stress-weighted, not measured: about a quarter of items carry an +exception so that a few hundred items exercise every kind several times. Real +intake mixes are not known to us; the report says so. + +Item-level exceptions (``EXCEPTION_MIX``): + +- ``dup_existing``: the patient already has a chart (same name and date of + birth). Right outcome: no new chart. +- ``dup_resend``: a referral already in this stream is sent again (a fax + resend). Right outcome: no second chart. +- ``ambiguous_same_person``: same name as an existing chart, date of birth off + by a typo. It is that person. Right outcome: no new chart. +- ``ambiguous_new_person``: same name as an existing chart, a different + person (different date of birth and address). Right outcome: a new chart. +- ``missing_required``: no date of birth. Right outcome: no chart; the + referral goes back to the sender. +- ``missing_optional``: no home phone. Right outcome: a chart with a blank + home phone. +- ``malformed_dob``: date of birth written MM/DD/YYYY. Right outcome: a chart + with the true date, or a stop; never a wrong date. +- ``malformed_case``: names in capitals with stray spaces. Right outcome: a + chart with the names as the person's name (case-insensitive match, trimmed). +- ``concurrent_same``: another clerk registers the same patient seconds + before this item runs. Right outcome: exactly one chart for the patient. +- ``concurrent_other``: another clerk edits an unrelated chart while this item + runs. Right outcome: this item's chart; the other edit is not collateral. +- ``session_expiry``: the arm's sessions are ended (forced logout) just + before the item. Right outcome: a chart, or a safe stop. +- ``slow_save``: the save takes several seconds longer than usual. Right + outcome: a chart. +- ``failed_save``: the save fails once (the app returns an error). Right + outcome: no chart from the failed attempt; a chart after a retry, or a safe + stop. + +Day-level events (``DRIFT_SCHEDULE``) stay on from their day to the end, as a +vendor release would: a theme change, a relabel of the save button, a field +reorder, and an added interstitial notice. +""" + +from __future__ import annotations + +import json +import random +from dataclasses import asdict, dataclass, field +from typing import Any, Optional + +from .arm import WorkItem + +WORKLOAD_VERSION = "lifecycle-workload/2026-10-20.1" + +EXCEPTION_MIX: dict[str, float] = { + "dup_existing": 0.03, + "dup_resend": 0.03, + "ambiguous_same_person": 0.02, + "ambiguous_new_person": 0.02, + "missing_required": 0.02, + "missing_optional": 0.02, + "malformed_dob": 0.02, + "malformed_case": 0.02, + "concurrent_same": 0.015, + "concurrent_other": 0.02, + "session_expiry": 0.02, + "slow_save": 0.02, + "failed_save": 0.015, +} + +#: (day, kind) — each stays on from that day to the end of the run. +DRIFT_SCHEDULE: tuple[tuple[int, str], ...] = ( + (2, "theme"), + (4, "relabel"), + (6, "reorder"), + (8, "interstitial"), +) + +# Invented names. Short names are kept on purpose: they occur in real +# worklists and some identity checks treat them differently. +FIRST = ( + "Nadia Marcus Ines Theo Grace Arlo Priya Tomas Wen Esme Kofi Liesel Rafael " + "Odile Bram Yuki Soren Amara Dmitri Luz Ana Bo Jae Ola Cyrus Mireille " + "Henrik Talia Oren Signe Mateo Leilani Fergus Ilse Kwame Rosalind Zeno " + "Petra Anouk Calum Dalia Emeric Farida Gideon Hollis Ida Jasper Kalinda" +).split() +LAST = ( + "Brennan Feld Carvalho Lindqvist Adeyemi Whitcombe Raman Okafor Vance " + "Halvorsen Quist Marchetti Szabo Tanaka Delacroix Mbeki Novak Ferreira " + "Ashby Kowalczyk Lund Pereira Ostrowski Nakamura Esposito Fairweather " + "Galloway Hartigan Ibarra Jablonski Kestrel Larkspur Merriweather Nyberg " + "Oyelaran Pemberton Quennell Rasmussen Stavros Thorne Ulloa Vasquez " + "Wexford Yarrow Zielinski Abernathy Birchall Caddick Dunmore Ng Li" +).split() +STREETS = ( + "WILLOW BEND ROAD QUARRY HILL DRIVE LANTERN COURT FERRY STREET CEDAR ROW " + "BEACON TERRACE STATION AVENUE MILL ROAD ORCHARD LANE HARBOR WAY" +).split() +STREET_NAMES = [ + " ".join(STREETS[i : i + 3]) if i + 3 <= len(STREETS) else STREETS[i] + for i in range(0, len(STREETS), 3) +] +TITLES = {"Female": ("Ms.", "Mrs."), "Male": ("Mr.",)} + + +@dataclass +class Truth: + """What is true about one item, kept by the harness, never shown to arms.""" + + item_id: str + seq: int + day: int + exception: str + #: The chart the item should produce, or None when no new chart is right. + intended: Optional[dict[str, str]] + #: The person this item is about (the true values), for the oracle's + #: "exactly one chart for this person" check. + person: dict[str, str] + rationale: str + #: Item-level injections the harness must apply around this item. + inject: list[dict[str, Any]] = field(default_factory=list) + + +@dataclass +class Workload: + seed: int + days: int + items: list[WorkItem] + truth: dict[str, Truth] + #: Charts that exist before the run (seeded so duplicates have targets). + existing: list[dict[str, str]] + drift: list[dict[str, Any]] + + def ledger_rows(self) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [ + { + "type": "workload", + "version": WORKLOAD_VERSION, + "seed": self.seed, + "days": self.days, + "items": len(self.items), + "exception_mix": EXCEPTION_MIX, + } + ] + rows += [{"type": "existing_chart", **c} for c in self.existing] + rows += [{"type": "drift_scheduled", **d} for d in self.drift] + for item in self.items: + t = self.truth[item.item_id] + rows.append({"type": "item", **asdict(item), "truth": asdict(t)}) + return rows + + +def _person(rng: random.Random, used: set[tuple[str, str]], n: int) -> dict[str, str]: + while True: + fname, lname = rng.choice(FIRST), rng.choice(LAST) + if (fname, lname) not in used: + used.add((fname, lname)) + break + sex = rng.choice(("Female", "Male")) + year = rng.randint(1935, 2004) + dob = f"{year:04d}-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}" + number = rng.randint(2, 498) + street = f"{number} {rng.choice(STREET_NAMES)}" + return { + "title": rng.choice(TITLES[sex]), + "fname": fname, + "lname": lname, + "DOB": dob, + "sex": sex, + "street": street, + "city": "EXAMPLETON", + "state_label": "Massachusetts", + "postal_code": "02139", + "phone_home": f"555-01{rng.randint(0, 99):02d}", + "email": f"{fname.lower()}.{lname.lower()}.{n}@example.invalid", + "country_code": "USA", + } + + +def _source_text(params: dict[str, str]) -> str: + lines = ["REFERRAL - new patient registration requested", ""] + labels = ( + ("title", "Title"), + ("fname", "First name"), + ("lname", "Last name"), + ("DOB", "Date of birth"), + ("sex", "Sex"), + ("street", "Street"), + ("city", "City"), + ("state_label", "State"), + ("postal_code", "ZIP"), + ("phone_home", "Home phone"), + ("email", "Email"), + ("country_code", "Country"), + ) + for key, label in labels: + lines.append(f"{label}: {params.get(key, '')}") + return "\n".join(lines) + "\n" + + +def _pick(rng: random.Random) -> str: + r = rng.random() + acc = 0.0 + for kind, p in EXCEPTION_MIX.items(): + acc += p + if r < acc: + return kind + return "clean" + + +def generate(seed: int, n_items: int, days: int, n_existing: int = 12) -> Workload: + """The stream for ``seed``: ``n_items`` referrals over ``days`` days.""" + + rng = random.Random(seed) + used: set[tuple[str, str]] = set() + existing = [_person(rng, used, 9000 + i) for i in range(n_existing)] + items: list[WorkItem] = [] + truth: dict[str, Truth] = {} + sent: list[tuple[WorkItem, dict[str, str]]] = [] + per_day = max(1, -(-n_items // days)) + + for seq in range(1, n_items + 1): + item_id = f"ref-{seq:05d}" + day = (seq - 1) // per_day + kind = _pick(rng) + person = _person(rng, used, seq) + params = dict(person) + intended: Optional[dict[str, str]] = dict(person) + inject: list[dict[str, Any]] = [] + why = "a new referral; register it" + + if kind == "dup_existing": + person = dict(rng.choice(existing)) + params, intended = dict(person), None + why = "this patient already has a chart" + elif kind == "dup_resend": + originals = [s for s in sent if truth[s[0].item_id].intended is not None] + if originals: + orig, orig_person = rng.choice(originals) + person, params, intended = dict(orig_person), dict(orig.params), None + why = f"a resend of {orig.item_id}" + else: + kind = "clean" + elif kind == "ambiguous_same_person": + person = dict(rng.choice(existing)) + params, intended = dict(person), None + y, m, d = person["DOB"].split("-") + params["DOB"] = f"{y}-{m}-{int(d) % 28 + 1:02d}" + why = "same person as an existing chart; the date of birth has a typo" + elif kind == "ambiguous_new_person": + other = rng.choice(existing) + person["fname"], person["lname"] = other["fname"], other["lname"] + person["email"] = ( + f"{other['fname'].lower()}.{other['lname'].lower()}.{seq}@example.invalid" + ) + params, intended = dict(person), dict(person) + why = "same name as an existing chart, a different person" + elif kind == "missing_required": + params.pop("DOB") + intended = None + why = "no date of birth; return the referral to the sender" + elif kind == "missing_optional": + params["phone_home"] = "" + person["phone_home"] = "" + intended = dict(person) + why = "no home phone; register without one" + elif kind == "malformed_dob": + y, m, d = person["DOB"].split("-") + params["DOB"] = f"{m}/{d}/{y}" + why = "date of birth written MM/DD/YYYY; the true date is ISO" + elif kind == "malformed_case": + params["fname"] = f" {person['fname'].upper()} " + params["lname"] = person["lname"].upper() + " " + why = "names in capitals with stray spaces" + elif kind == "concurrent_same": + intended = None + inject.append( + { + "event": "other_clerk_creates", + "person": dict(person), + "when": "before_item", + } + ) + why = "another clerk registered this patient moments earlier" + elif kind == "concurrent_other": + inject.append( + {"event": "other_clerk_edits_unrelated", "when": "during_item"} + ) + why = "another clerk edits an unrelated chart meanwhile" + elif kind == "session_expiry": + inject.append({"event": "force_logout", "when": "before_item"}) + why = "sessions ended just before the item" + elif kind == "slow_save": + inject.append( + { + "event": "slow_save", + "seconds": rng.choice((4, 6, 8)), + "when": "during_item", + } + ) + why = "the save is slow" + elif kind == "failed_save": + inject.append({"event": "failed_save", "times": 1, "when": "during_item"}) + why = "the save fails once" + + item = WorkItem( + seq=seq, + item_id=item_id, + day=day, + app="openemr", + params=params, + source_text=_source_text(params), + ) + items.append(item) + truth[item_id] = Truth( + item_id=item_id, + seq=seq, + day=day, + exception=kind, + intended=intended, + person=person, + rationale=why, + inject=inject, + ) + sent.append((item, person)) + + drift = [ + {"day": d, "kind": k, "first_seq": d * per_day + 1} + for d, k in DRIFT_SCHEDULE + if d < days + ] + return Workload( + seed=seed, days=days, items=items, truth=truth, existing=existing, drift=drift + ) + + +def dumps(workload: Workload) -> str: + return "".join(json.dumps(r, sort_keys=True) + "\n" for r in workload.ledger_rows()) diff --git a/packages/flow/tests/test_lifecycle_benchmark.py b/packages/flow/tests/test_lifecycle_benchmark.py new file mode 100644 index 000000000..34dba0575 --- /dev/null +++ b/packages/flow/tests/test_lifecycle_benchmark.py @@ -0,0 +1,178 @@ +"""The lifecycle harness's bookkeeping, end to end on the in-memory app. + +No browser, no fixture, no model: the toy arms in ``benchmark/lifecycle/fake.py`` +drive ``FakeApp`` so the workload, the operator, the oracle, the metric and +the report can be checked in a second. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +BENCH = Path(__file__).resolve().parents[1] / "benchmark" +sys.path.insert(0, str(BENCH)) + +from lifecycle import cost_model, metrics, run, workload # noqa: E402 +from lifecycle.arm import OperatorRequest # noqa: E402 +from lifecycle.operator import ScriptedOperator # noqa: E402 + + +def test_workload_is_seeded_and_covers_every_exception() -> None: + a = workload.generate(7, 400, 10) + b = workload.generate(7, 400, 10) + assert workload.dumps(a) == workload.dumps(b) + assert workload.dumps(workload.generate(8, 400, 10)) != workload.dumps(a) + kinds = {t.exception for t in a.truth.values()} + assert set(workload.EXCEPTION_MIX) <= kinds + assert [d["kind"] for d in a.drift] == [ + "theme", + "relabel", + "reorder", + "interstitial", + ] + # the arm never sees the truth + for item in a.items: + assert "truth" not in json.dumps(item.params) + missing = [i for i in a.items if a.truth[i.item_id].exception == "missing_required"] + assert missing and all("DOB" not in i.params for i in missing) + + +@pytest.mark.parametrize( + "claimed,oracle,wanted,exception,expected", + [ + ("done", "correct", True, "clean", "correct"), + ("stopped", "correct", False, "dup_existing", "right_stop"), + ("stopped", "absent", True, "clean", "false_stop"), + ("stopped", "absent", True, "failed_save", "fault_stop"), + ("stopped", "absent", True, "malformed_dob", "data_stop"), + ("done", "absent", True, "clean", "silent_wrong"), + ("done", "wrong", False, "dup_existing", "silent_wrong"), + ("stopped", "wrong", True, "clean", "flagged_wrong"), + ("stopped", "correct", True, "slow_save", "stopped_saved"), + ], +) +def test_classify(claimed, oracle, wanted, exception, expected) -> None: + assert ( + metrics.classify( + claimed=claimed, oracle=oracle, chart_wanted=wanted, exception=exception + ) + == expected + ) + + +def test_cost_model_bands_scale_every_charge() -> None: + half = cost_model.V1.scaled(0.5) + assert half.per_request("stop") == pytest.approx( + 0.5 * cost_model.V1.per_request("stop") + ) + assert half.version.endswith("x0.5") + assert "calibration" in cost_model.V1.to_dict() + + +def _operator(truth, look="absent"): + rows = [] + op = ScriptedOperator( + cost_model.V1, + truth, + look=lambda t: look, + do_by_hand=lambda t: {"chart_id": "9"}, + drift_active=lambda: ["relabel"], + record=rows.append, + ) + return op, rows + + +def test_operator_policy() -> None: + w = workload.generate(7, 200, 10) + dup = next(t for t in w.truth.values() if t.exception == "dup_existing") + clean = next(t for t in w.truth.values() if t.exception == "clean") + op, rows = _operator(w.truth) + assert op.ask(OperatorRequest("stop", dup.item_id, "dup", "test")).action == "skip" + first = op.ask( + OperatorRequest("stop", clean.item_id, "x", "test", {"resumable": True}) + ) + again = op.ask( + OperatorRequest("stop", clean.item_id, "x", "test", {"resumable": True}) + ) + assert (first.action, again.action) == ("resume", "do_by_hand") + assert ( + again.charged_s + == cost_model.V1.per_request("stop") + cost_model.V1.manual_entry_s + ) + ok = op.ask( + OperatorRequest( + "approval", clean.item_id, "save?", "test", {"values": dict(clean.intended)} + ) + ) + bad = op.ask( + OperatorRequest( + "approval", dup.item_id, "save?", "test", {"values": dict(dup.person)} + ) + ) + assert (ok.action, bad.action) == ("approve", "deny") + repair = op.ask( + OperatorRequest( + "repair_review", None, "relabel", "test", {"probes_passed": True} + ) + ) + assert repair.action == "accept_repair" + assert all(r["cost_model"] == cost_model.V1.version for r in rows) + op2, _ = _operator(w.truth, look="correct") + assert ( + op2.ask(OperatorRequest("reconcile", clean.item_id, "?", "t")).action + == "confirm_saved" + ) + + +def test_end_to_end_on_the_fake_app(tmp_path: Path) -> None: + out = tmp_path / "run" + assert ( + run.main( + [ + "--app", + "fake", + "--arm", + "lifecycle.fake:naive", + "--arm", + "lifecycle.fake:careful", + "--items", + "200", + "--days", + "10", + "--out", + str(out), + ] + ) + == 0 + ) + res = json.loads((out / "results.json").read_text()) + naive, careful = res["arms"]["naive"], res["arms"]["careful"] + # the naive arm claims everything done and the oracle catches its wrongs + assert naive["silent_wrong_rate"]["k"] > 0 + assert naive["operator_requests"] == {} + assert careful["silent_wrong_rate"]["k"] == 0 + assert careful["final_correct"] == careful["items"] == 200 + # a silent wrong is never a verified completion + assert naive["verified_completed"] == naive["outcomes"]["correct"] + for arm in (naive, careful): + b = arm["bands"] + assert ( + b["x0.5"]["verified_completed_per_operator_hour"] + > b["x1"]["verified_completed_per_operator_hour"] + > b["x2"]["verified_completed_per_operator_hour"] + ) + assert ( + b["x1"]["verified_completed_per_operator_hour_with_cleanup"] + <= b["x1"]["verified_completed_per_operator_hour"] + ) + report = (out / "REPORT.md").read_text() + assert "Verified completed items per operator-hour" in report + assert "No timings from real operators" in report + # the ledger alone reproduces the results + from lifecycle.report import results + + assert json.loads(json.dumps(results(out), sort_keys=True)) == res From 1560d4fa8fbfd7a023e91d0b35cf4af5a2a5f520 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 13:44:30 -0400 Subject: [PATCH 2/2] Lifecycle: OpenAdapt arm, OpenEMR adapter fixes, README, stops export The OpenAdapt arm runs the certified referral program on the governed Standard path (or one CLI run per item with mode=cli) and records what the product refused. Trigger SQL uses a delimiter; ports and project are overridable; the fixture is in Env.extra; arms may report maintenance phases inside an item. Workload 2026-10-20.2 widens the surname pool. Co-Authored-By: Claude Opus 5.5 --- packages/flow/benchmark/lifecycle/README.md | 104 ++ packages/flow/benchmark/lifecycle/apps.py | 15 +- .../flow/benchmark/lifecycle/openadapt_arm.py | 953 ++++++++++++++++++ packages/flow/benchmark/lifecycle/report.py | 37 + packages/flow/benchmark/lifecycle/run.py | 2 +- packages/flow/benchmark/lifecycle/workload.py | 20 +- 6 files changed, 1117 insertions(+), 14 deletions(-) create mode 100644 packages/flow/benchmark/lifecycle/README.md create mode 100644 packages/flow/benchmark/lifecycle/openadapt_arm.py diff --git a/packages/flow/benchmark/lifecycle/README.md b/packages/flow/benchmark/lifecycle/README.md new file mode 100644 index 000000000..7b585c503 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/README.md @@ -0,0 +1,104 @@ +# Lifecycle soak: does OpenAdapt remove work under ordinary business conditions? + +A "synthetic customer": a seeded stream of referrals is worked through on a +real local OpenEMR 8.0.0.3, with duplicates, bad source documents, other +clerks, forced logouts, slow and failed saves and four UI changes, while a +scripted operator answers every stop. The result is one number with its +parts, computed only from the run's ledger. + +**Verified completed items per operator-hour**: items whose final state the +independent oracle judges right and that the operator did not do by hand, +divided by every operator hour the run charged (setup, per-item requests, +maintenance after UI changes). + +Reported beside it, never folded in: + +- setup minutes (teach, compile, qualify, admit); +- per-item operator minutes (approvals, stops, reconciliations, repair reviews); +- maintenance minutes after each UI change; +- false-stop rate: stops with nothing written on items whose data was right + and that carried no injected fault, over all such items (Wilson 95% CI), and + how many of them happened with a UI change in force; +- silent-wrong rate: items the arm claimed done that the oracle judged wrong, + over all items it claimed done; the oracle is a direct SQL read of + `patient_data` from inside the database container, never the product's + verdict; +- wall time, host load, model calls and list-price cost; +- the same headline if every silent wrong is later found and fixed + (reconcile + re-entry per silent wrong), and the all-manual baseline + (every item entered by hand). + +## Files + +| File | What it is | +|---|---| +| `arm.py` | The interface every arm implements (`lifecycle-arm/1`) | +| `workload.py` | The seeded referral stream, its exception mix and ground truth | +| `apps.py` | The OpenEMR adapter: reset, SQL oracle channel, every injected event | +| `oracle.py` | The independent judge (per item, and the item's final state) | +| `operator.py` | The scripted operator and its decision rules | +| `cost_model.py` | Seconds charged per operator touch, versioned, with its calibration note | +| `metrics.py` | The metric and its parts, from the ledger | +| `report.py` | `results.json`, `REPORT.md` and `stops.jsonl` from the ledger | +| `run.py` | The runner | +| `openadapt_arm.py` | The OpenAdapt arm | +| `fake.py` | An in-memory app and two toy arms, for tests and for checking a new arm | +| `results//` | Committed runs: `ledger.jsonl`, `workload.jsonl`, `results.json`, `REPORT.md`, `stops.jsonl` | + +## Run it + +```bash +cd packages/flow/benchmark +python -m lifecycle.apps up # the lifecycle OpenEMR copy, ports 9420/9421 +python -m lifecycle.run --app openemr --seed 7 --items 300 --days 10 \ + --arm lifecycle.openadapt_arm:make --out lifecycle/results/ +python -m lifecycle.report lifecycle/results/ # recompute from the ledger +``` + +`--app fake --arm lifecycle.fake:careful` runs the whole pipeline in a second +with no fixture. Two runs with the same seed see the same items, events and +UI changes, in the same order. + +## The workload + +Each item is one new-patient referral (the 12 fields of OpenEMR's New +Patient form). About a quarter carry an exception (`workload.py` has the +exact mix and the right outcome of each): duplicates of existing charts, +resent referrals, same-name patients who are and are not the same person, a +missing date of birth, a missing home phone, a US-format date of birth, names +in capitals with stray spaces, another clerk registering the same patient +first, another clerk editing an unrelated chart mid-item, a forced logout, a +slow save and a save that fails once. The mix is stress-weighted to exercise +every kind several times, not a measured intake distribution. + +UI changes are applied in the app server, so every client sees them, and +stay on to the end of the run: a theme (day 2), a relabel of both create +buttons (day 4), a reordered demographics layout (day 6) and a notice that +must be dismissed before the form takes input (day 8). A version bump is not +run: the fixture pins one OpenEMR image. + +Every injected event, with exactly what was changed, is a row in the ledger. + +## The operator + +`operator.py` answers every request from the ledger's truth and from a +look at the app, never from the arm's verdict, and records the rule that +decided. Requests and their charges come from `cost_model.py`. No real +operator timings exist yet, so every charge is an assumption, stated in the +model and shown at x0.5 and x2. + +## Adding an arm + +Implement `arm.Arm` and pass `--arm module:factory`. Route every human touch +through `operator.ask`; never charge yourself and never judge yourself. +`Env.extra` carries the fixture's ports and read-only and actor REST +sessions; arms never get the oracle's SQL. Raise `arm.SpendCapReached` to +pause an arm at a spend cap. + +## Limits + +- One machine, shared with other work; every item row records the 1-minute + load. Wall times are not comparable across runs at different loads. +- Synthetic data, a synthetic workload mix, and assumed operator costs. +- The OpenAdapt arm's configuration, and what the product refused, is in + each run's `results.json` (`arms..describe`) and its report. diff --git a/packages/flow/benchmark/lifecycle/apps.py b/packages/flow/benchmark/lifecycle/apps.py index 4a97259da..a79a8fd14 100644 --- a/packages/flow/benchmark/lifecycle/apps.py +++ b/packages/flow/benchmark/lifecycle/apps.py @@ -40,9 +40,11 @@ HERE = Path(__file__).resolve().parent BENCH = HERE.parent -PROJECT = "openadapt-openemr-lifecycle" -UI_PORT = "9421" -API_PORT = "9420" +#: Another session can run its own copy: set these three (and +#: LIFECYCLE_OPENEMR_STATE) to its own project, ports and state folder. +PROJECT = os.environ.get("LIFECYCLE_OPENEMR_PROJECT", "openadapt-openemr-lifecycle") +UI_PORT = os.environ.get("LIFECYCLE_UI_PORT", "9421") +API_PORT = os.environ.get("LIFECYCLE_API_PORT", "9420") WEBROOT = "/var/www/localhost/htdocs/openemr" FORM = f"{WEBROOT}/interface/new/new_comprehensive.php" POPUP = f"{WEBROOT}/interface/new/new_search_popup.php" @@ -281,6 +283,7 @@ def extra(self) -> dict[str, Any]: "api_port": int(API_PORT), "rest_read_session": lambda: self.fx.token_session("oracle"), "rest_write_session": lambda: self.fx.token_session("actor"), + "fixture": self.fx, } def point_hooks_here(self) -> None: @@ -447,9 +450,11 @@ def inject( "CREATE TABLE IF NOT EXISTS lifecycle.fault_budget " "(k VARCHAR(190) PRIMARY KEY, n INT) ENGINE=Aria TRANSACTIONAL=0; " f"REPLACE INTO lifecycle.fault_budget VALUES ('{email}', {int(inj.get('times', 0))}); " - "DROP TRIGGER IF EXISTS openemr.oa_lifecycle_fault; " + "DROP TRIGGER IF EXISTS openemr.oa_lifecycle_fault;\n" + "DELIMITER //\n" "CREATE TRIGGER openemr.oa_lifecycle_fault BEFORE INSERT ON openemr.patient_data " - f"FOR EACH ROW BEGIN {body} END;" + f"FOR EACH ROW BEGIN {body} END//\n" + "DELIMITER ;\n" ) return {"event": event, "trigger": body} raise ValueError(f"unknown injection {event}") diff --git a/packages/flow/benchmark/lifecycle/openadapt_arm.py b/packages/flow/benchmark/lifecycle/openadapt_arm.py new file mode 100644 index 000000000..0f418c492 --- /dev/null +++ b/packages/flow/benchmark/lifecycle/openadapt_arm.py @@ -0,0 +1,953 @@ +"""The OpenAdapt arm: the certified referral program, run the way a customer runs it. + +Every item is one ``openadapt-flow run`` (a subprocess, the CLI a customer +uses) in an already signed-in Chromium that the arm keeps open, the way a +clerk's browser stays signed in all day (``--browser-cdp-endpoint``). The +configuration is the production one as far as the CLI allows it here: + +- durable runs (``--durable``): a stop pauses at the last verified step; +- per-item approval in ``problems_only`` mode; +- the REST record check from ``benchmark/openemr_local/verify.yaml`` (the + read-only OAuth client) confirming every save; +- self-repair on, with the Claude bridge repair agent from + ``benchmark/auto_repair`` (only when a model key file is given); +- the ``clinical-write`` certification policy. + +What the CLI refused is recorded in ``describe()["refused_features"]``, +verbatim where the product printed it. Every person-facing step goes through +the product's own commands: ``item-approval show|approve|decline``, +``approve`` and ``resume``. The arm never reads the oracle. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Optional + +from .arm import Env, ItemResult, OperatorRequest, WorkItem + +HERE = Path(__file__).resolve().parent +FLOW = HERE.parents[1] +DEFAULT_BUNDLE = Path.home() / "oa/scratch/quickstart-openemr/e2e-2/bundle" +CERTIFIED_DIGEST = "ec096afa2b614326b00ec3155f256041c19d86f3998761ce9d772cfd911460b6" +SIGN_IN_POLL_WAIT_S = 45.0 +BACKGROUND_POLL = "execute_background_services.php" + + +def _cli() -> list[str]: + return [sys.executable, "-m", "openadapt_flow"] + + +def _env(extra: Optional[dict[str, str]] = None) -> dict[str, str]: + env = dict(os.environ) + env["OPENADAPT_NO_LAUNCHER_CHECK"] = "1" + env["PYTHONPATH"] = os.pathsep.join( + [str(FLOW / "benchmark"), str(FLOW), env.get("PYTHONPATH", "")] + ).rstrip(os.pathsep) + env.update(extra or {}) + return env + + +class _As: + """The fixture, signed in as one of its users.""" + + def __init__(self, fx: Any, user: str) -> None: + self.ui_base_url = fx.ui_base_url + self._values = {**fx._runtime_values(), "OPENEMR_ACTOR_USER": user} + + def _runtime_values(self) -> dict[str, str]: + return dict(self._values) + + +class OpenAdaptArm: + name = "openadapt" + + def __init__(self, options: dict[str, Any]) -> None: + self.options = options + self.bundle_src = Path(options.get("bundle") or DEFAULT_BUNDLE) + self.cdp_port = int(options.get("cdp_port", 9425)) + self.key_file = options.get("repair_key_file") + self.approval_wait = int(options.get("approval_wait_s", 600)) + self.item_timeout = float(options.get("item_timeout_s", 1500)) + self.user = options.get("user", "oa_clerk1") + #: "governed": the Standard governed path the certified runs use + #: (in-process Replayer with the OpenEMR backend). "cli": one + #: ``openadapt-flow run`` per item, attached over CDP. + self.mode = options.get("mode", "governed") + self.refused: list[dict[str, str]] = [] + self._pw: Any = None + self._browser: Any = None + self.page: Any = None + self.fx: Any = None + + # -- description ----------------------------------------------------------- + + def describe(self) -> dict[str, Any]: + if self.mode == "governed": + return { + "kind": "openadapt", + "mode": "governed", + "surface": "the Standard governed run the certified referral runs use: " + "run gate on the certified bundle under clinical-write, governed " + "authorization built from it, durable, settled-state detection, the " + "REST record check (verify.yaml), the OpenEMR browser backend " + "(qualify.campaign._run)", + "bundle_digest": CERTIFIED_DIGEST, + "profile": "standard", + "durable": True, + "operator_surface": "the run's report.json; a stop is re-run from a " + "fresh form after the operator checks the record", + "refused_features": self.refused, + } + return { + "kind": "openadapt", + "mode": "cli", + "surface": "openadapt-flow run (CLI subprocess per item), attached to a " + "signed-in Chromium over CDP", + "bundle_digest": CERTIFIED_DIGEST, + "durable": True, + "approval_mode": "problems_only", + "policy": "clinical-write", + "record_check": "REST, read-only OAuth client (verify.yaml)", + "self_repair": bool(self.key_file), + "repair_agent": "benchmark.auto_repair.measure_mockmed:BridgeRepairAgent " + "(claude-sonnet-5-5)" + if self.key_file + else None, + "refused_features": self.refused, + } + + # -- browser ------------------------------------------------------------------- + + def _launch(self) -> None: + if self.mode == "governed": + self.backend, self._close_backend = ( + self._demo().OpenEMRPlaywrightBackend.launch( + self.fx.ui_base_url, headless=True + ) + ) + self.page = self.backend.page + return + from playwright.sync_api import sync_playwright + + self._pw = sync_playwright().start() + self._browser = self._pw.chromium.launch( + headless=True, args=[f"--remote-debugging-port={self.cdp_port}"] + ) + context = self._browser.new_context(viewport={"width": 1280, "height": 800}) + self.page = context.new_page() + + def _demo(self) -> Any: + from benchmark.openemr_api_tier.common import load_demo + + return load_demo() + + def _signed_in(self) -> bool: + try: + return bool( + self.page.evaluate( + "() => typeof window.navigateTab === 'function' && " + "!document.querySelector('#authUser')" + ) + ) + except Exception: # noqa: BLE001 - a crashed page is signed out + return False + + def sign_in(self) -> None: + demo = self._demo() + polled = {"seen": False} + + def saw(req: Any) -> None: + if BACKGROUND_POLL in req.url: + polled["seen"] = True + + self.page.context.on("requestfinished", saw) + backend = getattr(self, "backend", None) or _DemoBackend(self.page) + demo._preauthenticate_browser(backend, _As(self.fx, self.user)) + deadline = time.monotonic() + SIGN_IN_POLL_WAIT_S + while not polled["seen"] and time.monotonic() < deadline: + self.page.wait_for_timeout(250) + self.page.wait_for_timeout(1000) + + def open_form(self) -> None: + """A blank New Patient form in the tab shell, no chart open.""" + + demo, page = self._demo(), self.page + if page.evaluate( + "() => !!(window.app_view_model && app_view_model.application_data.patient())" + ): + 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 + ) + 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.wait_for_load_state("load", timeout=60_000) + frame.locator("#form_fname").wait_for(state="visible", timeout=60_000) + page.wait_for_timeout(500) + + def _ready(self) -> dict[str, Any]: + """Signed in at a blank form. After a stop, or when the session has + ended, the clerk starts again from the sign-in page (which also + closes a duplicate-check dialog a stopped run left open); that is + part of starting the next item, not a person's task.""" + + note: dict[str, Any] = {} + if self._signed_in() and not getattr(self, "_dirty", False): + try: + self.open_form() + return note + except Exception as exc: # noqa: BLE001 - e.g. the session ended + note["open_form_error"] = f"{type(exc).__name__}: {exc}"[:200] + note["signed_in_again"] = True + self._dirty = False + self.page.goto(self.fx.ui_base_url) + if not self._signed_in(): + self.sign_in() + else: + self.page.wait_for_timeout(1500) + self.open_form() + return note + + # -- setup --------------------------------------------------------------------- + + def setup(self, env: Env, operator: Any) -> list[dict[str, Any]]: + from benchmark.lifecycle.apps import make_fixture + + self.env = env + self.fx = make_fixture(Path(env.extra["fixture_state"])) + self.work = env.workdir + self.bundle = self.work / "bundle" + if self.bundle.exists(): + shutil.rmtree(self.bundle) + shutil.copytree(self.bundle_src, self.bundle) + manifest = json.loads((self.bundle / "manifest.json").read_text()) + digest = manifest.get("content_digest") + if digest != CERTIFIED_DIGEST: + raise RuntimeError(f"bundle digest {digest} is not the certified program") + if self.mode == "governed": + return self._setup_governed() + self._write_config() + if not self.key_file: + self.refused.append( + { + "feature": "self_repair", + "reason": "no repair model key file was given to this run, so no repair " + "agent is configured (repair is on only with an agent)", + } + ) + self._try_seal() + self._launch() + self.page.goto(self.fx.ui_base_url) + self.sign_in() + self._probe_standard() + # Pipelined read-back is refused under a durable run; recorded from the + # compatibility matrix and confirmed by each report's read-back mode. + from openadapt_flow.compatibility import cell + + c = cell("pipelined_readback", "durable") + self.refused.append({"feature": "pipelined_readback", "reason": c.reason or ""}) + phases = self._setup_phases() + return phases + + def _write_config(self) -> None: + # The record check reads the REST API over the fixture's loopback HTTP + # port: the HTTPS port's certificate is self-signed and the REST + # verifier takes no CA option. + api = self.fx.ui_base_url + cfg = { + "backend": { + "kind": "web", + "url": self.fx.ui_base_url, + "browser_cdp_endpoint": f"http://127.0.0.1:{self.cdp_port}", + }, + "effects": { + "kind": "rest", + "base_url": api, + "records_path": "/apis/default/api/patient?_limit=1000", + "records_key": "data", + "auth": {"bearer_env": "SOR_BEARER_TOKEN"}, + }, + "runtime": {"approval_mode": "problems_only"}, + } + if self.key_file: + cfg["repair"] = { + "agent": "benchmark.auto_repair.measure_mockmed:BridgeRepairAgent", + "screen_access_by": "lifecycle-soak (synthetic data)", + "model_provider": "anthropic", + "model_key_file": str(self.key_file), + "store": str(self.work / "repair-store"), + "approval_wait_s": self.approval_wait, + } + import yaml + + self.config = self.work / "deploy.yaml" + self.config.write_text(yaml.safe_dump(cfg, sort_keys=False)) + + def _setup_governed(self) -> list[dict[str, Any]]: + from openadapt_flow.compatibility import cell + from openadapt_flow.ir import Workflow + from openadapt_flow.qualify.campaign import _default_hooks, lookup_plan + from openadapt_flow.qualify.config import load_verify_config + + config = load_verify_config(FLOW / "benchmark/openemr_local/verify.yaml") + surface, structure = lookup_plan(config) + self.gov = { + "workflow": Workflow.load(self.bundle), + "config": config, + "hooks": _default_hooks(config, None), + "lookup": {"use_structural": structure, "deployment_surface": surface}, + } + for feature, reason in ( + ( + "per_item_approval", + "the governed run path takes no approval gate (only " + "`openadapt-flow run` wires one), so no save is put to a person", + ), + ( + "self_repair", + "the governed run path wires no repair agent (only " + "`openadapt-flow run` does)", + ), + ("pipelined_readback", cell("pipelined_readback", "durable").reason or ""), + ( + "cli_run", + "`openadapt-flow run` cannot run this program: with an attached " + "signed-in browser every item halts at step_000 (structured identity of the " + "title field not found by the generic web backend); without one the browser " + "starts signed out. Measured in the CLI smoke run.", + ), + ( + "signed_admission", + "not issued for this program yet; the run gate on the " + "certified bundle authorizes each run, as in the certified referral runs", + ), + ): + self.refused.append({"feature": feature, "reason": reason}) + self._launch() + self.page.goto(self.fx.ui_base_url) + self.sign_in() + return self._setup_phases() + + def _run_governed(self, item: WorkItem, run_dir: Path, n: int) -> Any: + import dataclasses + + from openadapt_flow.qualify.campaign import TrialContext, _run + from openadapt_flow.run_gate import ( + build_runtime_authorization, + evaluate_run_gate, + ) + + g = self.gov + hooks = dataclasses.replace( + g["hooks"], open_session=lambda: (self.backend, lambda: None) + ) + params = {k: str(v) for k, v in item.params.items()} + ctx = TrialContext( + case_id="lifecycle", + expected="VERIFIED", + oracle_kind="exact_record", + trial=n, + trial_dir=run_dir, + run_dir=run_dir, + run_id=f"lifecycle-{item.item_id}-{n}-{int(time.time())}", + params=params, + certified_run=True, + state={"policy": "clinical-write"}, + ) + + def authorize(_verifier: Any, gate_kwargs: dict[str, Any]) -> Any: + gate = evaluate_run_gate(g["workflow"], **gate_kwargs) + if not gate.passed: + raise RuntimeError("run gate refused: " + gate.render()[-300:]) + return build_runtime_authorization(g["workflow"], gate, params=params) + + report, _ = _run( + workflow=g["workflow"], + bundle_dir=self.bundle, + config=g["config"], + hooks=hooks, + ctx=ctx, + setup=type("Setup", (), {"observer": None})(), # type: ignore[arg-type] + authorize=authorize, + fault_driver=None, + marks={"start": time.time()}, + lookup=g["lookup"], + ) + return json.loads(report.model_dump_json()) + + def _item_governed(self, item: WorkItem, operator: Any) -> ItemResult: + t0 = time.time() + idir = self.work / "items" / item.item_id + idir.mkdir(parents=True, exist_ok=True) + detail: dict[str, Any] = {"log": []} + log = detail["log"] + calls, cost, report, run_dir = 0, 0.0, {}, idir / "run-1" + for n in (1, 2): + run_dir = idir / f"run-{n}" + try: + detail.update(self._ready()) + report = self._run_governed(item, run_dir, n) + except Exception as exc: # noqa: BLE001 - a crash is a stop + report = { + "execution_outcome": "ERROR", + "halt": {"reason": f"{type(exc).__name__}: {exc}"[:300]}, + } + calls += int(report.get("model_calls") or 0) + cost += float(report.get("est_model_cost_usd") or 0.0) + tx = str(report.get("transaction_outcome") or "") + if tx == "VERIFIED": + return self._result( + item, "done", report, t0, calls, cost, detail, run_dir + ) + self._dirty = True + halt = report.get("halt") or {} + unsure = tx == "RECONCILIATION_REQUIRED" + reason = ( + f"{report.get('execution_outcome')}/{tx}: {halt.get('state_id', '')} " + f"{str(halt.get('reason', ''))[:200]}" + ) + decision = operator.ask( + OperatorRequest( + kind="reconcile" if unsure else "stop", + item_id=item.item_id, + reason=reason.strip(), + surface="report.json; re-run from a fresh form", + detail={ + "resumable": n == 1, + "transaction_outcome": tx, + "halt_step": halt.get("state_id"), + }, + ) + ) + log.append({"run": n, "stop": reason.strip(), "decision": decision.action}) + if decision.action != "resume": + break + return self._result(item, "stopped", report, t0, calls, cost, detail, run_dir) + + def _try_seal(self) -> None: + """Encrypt the bundle at rest, as the default gated path requires. + + ``openadapt-flow seal`` with a fresh local key; when it refuses, the + refusal is recorded verbatim and the run uses ``--allow-unencrypted``. + """ + + import base64 + import secrets + + src = self.work / "seal-source" + dest = self.work / "sealed-bundle" + for d in (src, dest): + if d.exists(): + shutil.rmtree(d) + shutil.copytree(self.bundle, src) + # A benchmark sidecar Flow does not know; seal refuses unknown files. + (src / "benchmark-contract.json").unlink(missing_ok=True) + key = base64.b64encode(secrets.token_bytes(32)).decode() + proc = subprocess.run( + [*_cli(), "seal", str(src), "-o", str(dest)], + capture_output=True, + text=True, + env=_env({"OPENADAPT_BUNDLE_KEY": key}), + timeout=600, + ) + out = (proc.stdout + proc.stderr).strip() + (self.work / "seal-output.txt").write_text(out + "\n") + if proc.returncode == 0 and (dest / "manifest.json").exists(): + self.bundle, self.bundle_key = dest, key + self.unencrypted = False + else: + self.bundle_key, self.unencrypted = None, True + line = next((x for x in out.splitlines() if "REFUSED" in x), out[-300:]) + self.refused.append( + { + "feature": "encrypted_bundle", + "reason": line.strip() + " (the run uses --allow-unencrypted)", + } + ) + + def _token_env(self) -> dict[str, str]: + session = self.fx.token_session("oracle") + token = session.headers["Authorization"].split(" ", 1)[1] + session.close() + env = {"SOR_BEARER_TOKEN": token} + if getattr(self, "bundle_key", None): + env["OPENADAPT_BUNDLE_KEY"] = self.bundle_key + return env + + def _run_cmd( + self, + run_dir: Path, + params: Path, + *, + profile: Optional[str] = None, + dry_run: bool = False, + ) -> list[str]: + cmd = [ + *_cli(), + "run", + str(self.bundle), + "--config", + str(self.config), + "--url", + self.fx.ui_base_url, + "--run-dir", + str(run_dir), + "--params-file", + str(params), + "--durable", + "--approval-mode", + "problems_only", + "--approval-wait", + str(self.approval_wait), + "--policy", + "clinical-write", + ] + if profile: + cmd += ["--profile", profile] + elif getattr(self, "unencrypted", False): + cmd.append("--allow-unencrypted") + if dry_run: + cmd.append("--dry-run") + return cmd + + def _probe_standard(self) -> None: + """Ask the CLI for the Standard profile once and record its answer. + + A real run (a dry run stops before the admission check), pointed at a + port where nothing listens, so it can write nothing even if admitted. + """ + + import yaml + + probe = self.work / "standard-probe" + probe.mkdir(parents=True, exist_ok=True) + params = probe / "params.json" + params.write_text(json.dumps({"fname": "Probe"})) + cfg = yaml.safe_load(self.config.read_text()) + cfg["backend"] = {"kind": "web", "url": "http://127.0.0.1:9439"} + probe_cfg = probe / "deploy.yaml" + probe_cfg.write_text(yaml.safe_dump(cfg, sort_keys=False)) + cmd = self._run_cmd(probe / "run", params, profile="standard") + cmd[cmd.index("--config") + 1] = str(probe_cfg) + cmd[cmd.index("--url") + 1] = "http://127.0.0.1:9439" + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + env=_env(self._token_env()), + timeout=600, + cwd=self.work, + ) + out = proc.stdout + proc.stderr + (probe / "output.txt").write_text(out) + m = re.search(r"[^\n]*signed qualification admission[^\n]*", out) + self.refused.append( + { + "feature": "standard_profile", + "reason": ( + m.group(0).strip() if m else f"exit {proc.returncode}: {out[-300:]}" + ), + } + ) + + def _setup_phases(self) -> list[dict[str, Any]]: + """Teach, compile, qualify and admit, from the certified program's own + records (they were not re-run in this soak).""" + + phases: list[dict[str, Any]] = [] + demo = self.bundle_src.parent / "demonstration.jsonl" + teach_s = None + if demo.exists(): + ts = [ + json.loads(line).get("timestamp") + for line in demo.read_text().splitlines() + if line.strip() + ] + ts = [t for t in ts if isinstance(t, (int, float))] + if len(ts) > 1: + teach_s = max(ts) - min(ts) + phases.append( + { + "phase": "teach", + "wall_s": teach_s or 36.8, + "attended_s": teach_s or 36.8, + "model_calls": 0, + "cost_usd": 0.0, + "source": "the demonstration's recorded duration", + } + ) + phases.append( + { + "phase": "compile", + "wall_s": 12.2, + "attended_s": 0.0, + "model_calls": 0, + "cost_usd": 0.0, + "source": "quickstart e2e-2 log (compile 12.2 s)", + } + ) + phases.append( + { + "phase": "qualify", + "wall_s": 549.6, + "attended_s": 0.0, + "model_calls": 0, + "cost_usd": 0.0, + "source": "qual10 QUALIFICATION.md: 18 trials, 549.6 s, unattended", + } + ) + phases.append( + { + "phase": "admit", + "wall_s": 0.0, + "attended_s": 0.0, + "model_calls": 0, + "cost_usd": 0.0, + "source": "fixed charge (cost model admit_s); the signed admission " + "itself does not exist yet", + } + ) + return phases + + # -- one item ---------------------------------------------------------------------- + + def _approvals(self, run_dir: Path) -> list[dict[str, Any]]: + if not (run_dir / "item-approval").exists(): + return [] + proc = subprocess.run( + [*_cli(), "item-approval", "show", str(run_dir)], + capture_output=True, + text=True, + env=_env(), + timeout=300, + ) + if proc.returncode != 0: + return [] + try: + data = json.loads(proc.stdout) + except ValueError: + return [] + if isinstance(data, dict): + data = data.get("requests", []) + return [r for r in data if isinstance(r, dict)] + + def _answer( + self, + item: WorkItem, + run_dir: Path, + req: dict[str, Any], + operator: Any, + log: list[dict[str, Any]], + ) -> None: + rid = req.get("request_id") or req.get("id") + sha = req.get("request_sha256") or req.get("digest") or req.get("sha256") + text = json.dumps(req) + is_repair = "repair" in text.lower() + values = {} + for f in req.get("fields") or []: + if isinstance(f, dict): + key = f.get("param") or f.get("name") or f.get("label") + if key: + values[str(key)] = str(f.get("value", "")) + kind = "repair_review" if is_repair else "approval" + decision = operator.ask( + OperatorRequest( + kind=kind, + item_id=item.item_id, + reason=str( + req.get("reason") or req.get("workflow_name") or "save approval" + ), + surface="cli:openadapt-flow item-approval", + detail={ + "values": values or dict(item.params), + "probes_passed": is_repair, + "request_id": rid, + }, + ) + ) + verb = ( + "approve" if decision.action in ("approve", "accept_repair") else "decline" + ) + cmd = [ + *_cli(), + "item-approval", + verb, + str(run_dir), + "--request-id", + str(rid), + "--request-sha256", + str(sha), + "--decided-by", + "human", + "--route", + "desktop_run_it", + "--unauthenticated", + ] + if verb == "decline": + cmd += ["--reason", "wrong_values"] + proc = subprocess.run( + cmd, capture_output=True, text=True, env=_env(), timeout=300 + ) + log.append( + { + "answered": rid, + "verb": verb, + "kind": kind, + "exit": proc.returncode, + "stderr": proc.stderr[-300:], + } + ) + + def _drive( + self, + cmd: list[str], + run_dir: Path, + item: WorkItem, + operator: Any, + log: list[dict[str, Any]], + ) -> int: + """Run one CLI command; answer every approval it asks for meanwhile.""" + + out = open(run_dir.parent / f"{run_dir.name}-{len(log)}.log", "w") + proc = subprocess.Popen( + cmd, + stdout=out, + stderr=subprocess.STDOUT, + env=_env(self._token_env()), + cwd=self.work, + ) + answered: set[str] = set() + deadline = time.monotonic() + self.item_timeout + while proc.poll() is None: + if time.monotonic() > deadline: + proc.kill() + log.append({"timeout": True}) + break + for req in self._approvals(run_dir): + rid = str(req.get("request_id") or req.get("id")) + status = str(req.get("status", "")).lower() + if rid in answered or status in ("answered", "consumed", "expired"): + continue + answered.add(rid) + self._answer(item, run_dir, req, operator, log) + time.sleep(2.0) + out.close() + code = proc.wait() + text = Path(out.name).read_text(errors="replace") + for line in text.splitlines(): + if line.startswith("Auto-repair is off for this run:") and not any( + r["feature"] == "self_repair" for r in self.refused + ): + self.refused.append({"feature": "self_repair", "reason": line.strip()}) + return code + + def _report(self, run_dir: Path) -> dict[str, Any]: + path = run_dir / "report.json" + if not path.exists(): + return {} + try: + return json.loads(path.read_text()) + except ValueError: + return {} + + def run_item(self, item: WorkItem, env: Env, operator: Any) -> ItemResult: + if self.mode == "governed": + return self._item_governed(item, operator) + t0 = time.time() + idir = self.work / "items" / item.item_id + idir.mkdir(parents=True, exist_ok=True) + run_dir = idir / "run" + params = idir / "params.json" + params.write_text(json.dumps(item.params)) + params.chmod(0o600) + detail: dict[str, Any] = {"log": []} + log = detail["log"] + try: + detail.update(self._ready()) + except Exception as exc: # noqa: BLE001 - the session could not be opened + detail["session_error"] = f"{type(exc).__name__}: {exc}"[:300] + self._drive(self._run_cmd(run_dir, params), run_dir, item, operator, log) + report = self._report(run_dir) + model_calls = int(report.get("model_calls") or 0) + cost = float(report.get("est_model_cost_usd") or 0.0) + + for attempt in range(2): + tx = str(report.get("transaction_outcome") or "") + ex = str(report.get("execution_outcome") or "") + if tx == "VERIFIED": + return self._result( + item, "done", report, t0, model_calls, cost, detail, run_dir + ) + esc = {} + if (run_dir / "pending_escalation.json").exists(): + try: + esc = json.loads((run_dir / "pending_escalation.json").read_text()) + except ValueError: + esc = {} + category = str(esc.get("category", "")) + unsure = tx == "RECONCILIATION_REQUIRED" or category.startswith( + ("delivery_uncertain", "effect_") + ) + halt = report.get("halt") or {} + reason = ( + f"{tx or ex or 'no report'}: {category} " + f"{halt.get('state_id', '')} {str(halt.get('reason', ''))[:200]}" + ) + decision = operator.ask( + OperatorRequest( + kind="reconcile" if unsure else "stop", + item_id=item.item_id, + reason=reason.strip(), + surface="cli:openadapt-flow explain/approve/resume", + detail={ + "resumable": bool(esc), + "category": category, + "transaction_outcome": tx, + "halt_step": halt.get("state_id"), + }, + ) + ) + log.append({"stop": reason.strip(), "decision": decision.action}) + if decision.action != "resume": + return self._result( + item, "stopped", report, t0, model_calls, cost, detail, run_dir + ) + # The person resolves the stop, approves, and resumes the run. + try: + self._ready_for_resume() + except Exception as exc: # noqa: BLE001 + detail["resume_session_error"] = f"{type(exc).__name__}: {exc}"[:300] + subprocess.run( + [ + *_cli(), + "approve", + str(run_dir), + "--approver", + "operator", + "--resolution", + "checked the item and resumed", + ], + capture_output=True, + text=True, + env=_env(), + timeout=300, + ) + self._drive( + [ + *_cli(), + "resume", + str(run_dir), + "--config", + str(self.config), + "--url", + self.fx.ui_base_url, + "--params-file", + str(params), + ], + run_dir, + item, + operator, + log, + ) + report = self._report(run_dir) + model_calls += int(report.get("model_calls") or 0) + cost += float(report.get("est_model_cost_usd") or 0.0) + log.append({"resumed": report.get("transaction_outcome")}) + return self._result( + item, "stopped", report, t0, model_calls, cost, detail, run_dir + ) + + def _ready_for_resume(self) -> None: + """Before a resume: if the session ended, sign in again and open the + form; otherwise leave the page where the run paused.""" + + if not self._signed_in(): + self.page.goto(self.fx.ui_base_url) + self.sign_in() + self.open_form() + + def _result( + self, + item: WorkItem, + claimed: str, + report: dict[str, Any], + t0: float, + calls: int, + cost: float, + detail: dict[str, Any], + run_dir: Path, + ) -> ItemResult: + halt = report.get("halt") or {} + approval = report.get("approval") or {} + detail.update( + { + "execution_outcome": report.get("execution_outcome"), + "transaction_outcome": report.get("transaction_outcome"), + "halt_step": halt.get("state_id"), + "halt_reason": str(halt.get("reason", ""))[:300], + "approval_mode": approval.get("mode"), + "approval_decision": approval.get("decision"), + "self_repair": report.get("self_repair"), + "readback": report.get("readback"), + } + ) + status = " / ".join( + str(x) + for x in ( + report.get("execution_outcome"), + report.get("transaction_outcome"), + ) + if x + ) + return ItemResult( + item_id=item.item_id, + claimed=claimed, # type: ignore[arg-type] + product_status=status or "no report", + wall_s=time.time() - t0, + model_calls=calls, + cost_usd=cost, + evidence=str(run_dir.relative_to(self.env.workdir.parent.parent)), + detail=detail, + ) + + def after_drift( + self, event: dict[str, Any], env: Env, operator: Any + ) -> list[dict[str, Any]]: + return [] # the product is not told about UI changes; it meets them + + def close(self) -> None: + if getattr(self, "_close_backend", None): + self._close_backend() + return + try: + if self._browser is not None: + self._browser.close() + finally: + if self._pw is not None: + self._pw.stop() + + +class _DemoBackend: + """The one attribute the demo's sign-in uses.""" + + def __init__(self, page: Any) -> None: + self.page = page + + +def make(options: dict[str, Any]) -> OpenAdaptArm: + return OpenAdaptArm(options) diff --git a/packages/flow/benchmark/lifecycle/report.py b/packages/flow/benchmark/lifecycle/report.py index 90e97705d..9ae3efc3d 100644 --- a/packages/flow/benchmark/lifecycle/report.py +++ b/packages/flow/benchmark/lifecycle/report.py @@ -166,8 +166,45 @@ def render(res: dict[str, Any]) -> str: return "\n".join(lines) +def stops(run_dir: Path) -> list[dict[str, Any]]: + """One row per item an arm did not claim done, with the oracle's view of + whether anything was actually wrong (for false-stop root-causing).""" + + out = [] + for r in load(run_dir): + if r["type"] != "item_result" or r["claimed"] == "done": + continue + d = r.get("detail") or {} + repair = d.get("self_repair") or {} + out.append( + { + "arm": r["arm"], + "seq": r["seq"], + "item_id": r["item_id"], + "evidence": r.get("evidence"), + "app": "openemr", + "exception": r["exception"], + "drift_active": r.get("drift_active"), + "halt_step": d.get("halt_step"), + "halt_reason": d.get("halt_reason"), + "execution_outcome": d.get("execution_outcome"), + "transaction_outcome": d.get("transaction_outcome"), + "repair": repair if isinstance(repair, dict) else str(repair), + "outcome": r["outcome"], + "anything_wrong": r["outcome"] not in ("false_stop",), + "oracle": r["oracle"].get("verdict"), + "final": r["final_oracle"], + "product_status": r["product_status"], + } + ) + return out + + def write_report(run_dir: Path) -> dict[str, Any]: res = results(run_dir) + (run_dir / "stops.jsonl").write_text( + "".join(json.dumps(s, sort_keys=True) + "\n" for s in stops(run_dir)) + ) (run_dir / "results.json").write_text( json.dumps(res, indent=2, sort_keys=True) + "\n" ) diff --git a/packages/flow/benchmark/lifecycle/run.py b/packages/flow/benchmark/lifecycle/run.py index e64088388..658096c68 100644 --- a/packages/flow/benchmark/lifecycle/run.py +++ b/packages/flow/benchmark/lifecycle/run.py @@ -235,7 +235,7 @@ def record(row: dict[str, Any]) -> None: for th in during: th.join(timeout=60) cleared = app.clear_item_events() - for phase in result.phases: + for phase in [*result.phases, *result.detail.get("maintenance_phases", [])]: ledger.write( { "type": "phase", diff --git a/packages/flow/benchmark/lifecycle/workload.py b/packages/flow/benchmark/lifecycle/workload.py index c10b8d16c..f126e2b5c 100644 --- a/packages/flow/benchmark/lifecycle/workload.py +++ b/packages/flow/benchmark/lifecycle/workload.py @@ -53,7 +53,7 @@ from .arm import WorkItem -WORKLOAD_VERSION = "lifecycle-workload/2026-10-20.1" +WORKLOAD_VERSION = "lifecycle-workload/2026-10-20.2" EXCEPTION_MIX: dict[str, float] = { "dup_existing": 0.03, @@ -87,14 +87,18 @@ "Henrik Talia Oren Signe Mateo Leilani Fergus Ilse Kwame Rosalind Zeno " "Petra Anouk Calum Dalia Emeric Farida Gideon Hollis Ida Jasper Kalinda" ).split() -LAST = ( - "Brennan Feld Carvalho Lindqvist Adeyemi Whitcombe Raman Okafor Vance " - "Halvorsen Quist Marchetti Szabo Tanaka Delacroix Mbeki Novak Ferreira " - "Ashby Kowalczyk Lund Pereira Ostrowski Nakamura Esposito Fairweather " - "Galloway Hartigan Ibarra Jablonski Kestrel Larkspur Merriweather Nyberg " - "Oyelaran Pemberton Quennell Rasmussen Stavros Thorne Ulloa Vasquez " - "Wexford Yarrow Zielinski Abernathy Birchall Caddick Dunmore Ng Li" +#: 300 invented surnames (20 stems x 15 endings). OpenEMR's duplicate search +#: lists any chart that shares a last name, so the size of this pool sets how +#: often a new patient meets a namesake; a real practice has common surnames +#: too. Short surnames are kept for the same reason as short first names. +_STEMS = ( + "Ash Birch Cald Dun Ell Fair Gal Hart Ivers Jess Kell Lang Mar Nor Oak " + "Pem Quen Rad Stan Thorn" ).split() +_ENDINGS = ( + "by ford wick ton ley more well wood field brook combe stead mere gate ridge" +).split() +LAST = [a + b for a in _STEMS for b in _ENDINGS][:298] + ["Ng", "Li"] STREETS = ( "WILLOW BEND ROAD QUARRY HILL DRIVE LANTERN COURT FERRY STREET CEDAR ROW " "BEACON TERRACE STATION AVENUE MILL ROAD ORCHARD LANE HARBOR WAY"