benchmark: verification coverage, fresh runs on OpenEMR and openIMIS (harness)

packages/flow/benchmark/scoreboard/tasks/openimis.py

@@ -111,7 +111,10 @@
 
 
 def fixture() -> OpenIMISFixture:
-    return OpenIMISFixture()
+    # A second copy runs on its own ports (OPENIMIS_HTTP_PORT, OPENIMIS_DB_PORT)
+    # and compose project (OPENIMIS_COMPOSE_PROJECT); the defaults are unchanged.
+    port = os.environ.get("OPENIMIS_HTTP_PORT")
+    return OpenIMISFixture(http_port=int(port)) if port else OpenIMISFixture()
 
 
 def claim_form_url(fx: Optional[OpenIMISFixture] = None) -> str:

packages/flow/benchmark/verification_coverage/README.md

@@ -0,0 +1,87 @@
+# Verification coverage: fresh runs, per fault class
+
+Measures the design in
+[`docs/VERIFICATION_COVERAGE.md`](../../docs/VERIFICATION_COVERAGE.md): frame
+conditions, the fault gate, and provenance. It reruns the OpenEMR fault
+experiment of OpenAdaptAI/OpenAdapt#1365 as fresh agent runs, not rebuilt
+database states, and runs the same suite on openIMIS. The results are in
+[`RESULTS.md`](RESULTS.md).
+
+## Protocol
+
+For each item, a computer-use agent (Claude Sonnet 5.5, the scoreboard's
+agent) enters one record through the app's own UI, on a local fixture with
+synthetic data. A fault is injected only in the fixture's database, by a
+trigger installed before the agent starts or a poller that acts once the new
+record lands. The app itself is unchanged.
+
+Three `verify-run` engines judge the same save. Each runs from its own
+checkout and takes its own baseline before the agent starts:
+
+| Engine | Commit | What it checks |
+|---|---|---|
+| `e1365` | the engine #1365 measured with | the declared record; another record removed |
+| `main` | main before this work | the same, plus #1375: another record changed, per effect |
+| `after` | this work | the frame with the app's own attribution, and the round trip against the source document |
+
+Ground truth is read from the fixture's database after the run. A bad save
+is caught when the engine's outcome is anything but `verified`. A clean save
+that is refused is a false alarm.
+
+## Fault classes
+
+| Class | What happens | The save is |
+|---|---|---|
+| `ok` | nothing | clean |
+| `partial` | a trigger blanks a field of the new record | bad |
+| `wrong_value` | a trigger changes a field of the new record | bad |
+| `duplicate` | the poller copies the new record | bad |
+| `no_persist` | the poller removes the new record after it lands | bad |
+| `collateral_loss` | the poller removes another, pre-existing record | bad |
+| `collateral_edit` | the poller edits another record, with no audit trail | bad |
+| `collateral_edit_bot` | the same edit, attributed to the bot's own user | bad |
+| `concurrent_edit_other` | the same edit, attributed to another user | clean: a legitimate concurrent edit |
+| `misread_source` | the source document holds one value; the agent is told another (a misread before the save) and saves it faithfully | bad |
+
+Attribution comes from the app. OpenEMR writes each patient-record change to
+its audit log (`log`, with the user and `patient_id`); the `after` engine
+reads it with a SELECT-only database user (`audit_read`). openIMIS stamps
+each claim row with its last writer and time (`AuditUserID`,
+`ValidityFrom`); the `after` engine reads those (`record_field`). For the two
+attributed edits, the poller writes what the app would have written: a
+`patient-record-update` log row in OpenEMR, the writer and time on the
+openIMIS row.
+
+For every item the `after` engine also gets the source document (a synthetic
+referral or claim form) and the provenance a first reader would produce. The
+"Held before the save" column is the pre-save cross-check: a deterministic
+second reading of the document disagreeing with what the agent was told.
+
+## The fault gate
+
+`openemr_gate.py` runs the fault gate (`openadapt_flow/fault_gate.py`) on the
+OpenEMR fixture with no model calls, for two reads of the same contract: the
+full patient list (the #1365 check) and a search for the run's email (the
+kind of read a synthesized check picks).
+
+## Running
+
+The fixtures run as their own compose projects under the shared-machine
+rules for this workspace (a lease, at most two fixture sessions, a disk and
+load floor). With a fixture up and the three engines checked out:
+
+```bash
+PYTHONPATH=packages/flow/benchmark python -m verification_coverage.openemr_run \
+  --run-id <id> --per-fault 5 \
+  --engine-e1365 <checkout> --engine-main <checkout> --engine-after <checkout>
+```
+
+```bash
+python -m verification_coverage.summarize --write
+```
+
+## What is public
+
+The harness, the fault classes and these rows are public: synthetic data
+only. Coverage floors a deployment chooses and any grown fault corpus are
+not part of this directory.

packages/flow/benchmark/verification_coverage/exec_mariadb.py

@@ -0,0 +1,80 @@
+"""A minimal DB-API connection to a fixture's MariaDB through ``docker exec``.
+
+The OpenEMR fixture publishes no database port. This lets a ``kind: sql``
+verifier (``sql_driver: verification_coverage.exec_mariadb``) read it as a
+SELECT-only user, the way a deployment would read through its own driver.
+Benchmark plumbing only: one statement, no parameters, rows as text.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from typing import Any, Optional
+
+
+class _Cursor:
+    def __init__(self, conn: "Connection") -> None:
+        self._conn = conn
+        self.description: Optional[list[tuple[str, ...]]] = None
+        self._rows: list[tuple[Any, ...]] = []
+
+    def execute(self, sql: str, params: Any = None) -> None:
+        if params:
+            raise ValueError("exec_mariadb takes no query parameters")
+        out = subprocess.run(
+            [
+                "docker",
+                "exec",
+                "-i",
+                "-e",
+                "MYSQL_PWD",
+                self._conn.container,
+                "mariadb",
+                "-u",
+                self._conn.user,
+                "--batch",
+                "--raw",
+                self._conn.database,
+                "-e",
+                sql,
+            ],
+            capture_output=True,
+            text=True,
+            timeout=60,
+            env={**os.environ, "MYSQL_PWD": self._conn.password},
+            check=True,
+        ).stdout
+        lines = out.splitlines()
+        if not lines:
+            self.description, self._rows = [], []
+            return
+        header = lines[0].split("\t")
+        self.description = [(name,) for name in header]
+        self._rows = [
+            tuple(None if cell == "NULL" else cell for cell in line.split("\t"))
+            for line in lines[1:]
+        ]
+
+    def fetchall(self) -> list[tuple[Any, ...]]:
+        return list(self._rows)
+
+
+class Connection:
+    def __init__(self, container: str, user: str, password: str, database: str):
+        self.container = container
+        self.user = user
+        self.password = password
+        self.database = database
+
+    def cursor(self) -> _Cursor:
+        return _Cursor(self)
+
+    def close(self) -> None:
+        return None
+
+
+def connect(
+    container: str, user: str, password: str = "", database: str = "openemr"
+) -> Connection:
+    return Connection(container, user, password, database)

packages/flow/benchmark/verification_coverage/openemr_gate.py

@@ -0,0 +1,159 @@
+"""The fault gate on an OpenEMR test copy (no model calls).
+
+    python -m verification_coverage.openemr_gate --run-id <id>
+
+Runs ``openadapt_flow.fault_gate.run_gate`` against the fixture: the clean
+save is the actor's own ``POST /api/patient``; the faults are privileged SQL
+on ``patient_data``; the check reads through OpenEMR's REST API with the
+read-only oracle client, exactly as ``verify-run`` does in production. Two
+checks are measured:
+
+- ``full``: the patient list, as in #1365 (``/api/patient``);
+- ``scoped``: the same contract read through a search for the run's email,
+  the kind of read a synthesized check picks. It can't see another chart.
+
+Writes ``results/openemr/<run-id>/gate-<check>.json``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Any, Mapping
+
+HERE = Path(__file__).resolve().parent
+BENCH = HERE.parent
+sys.path.insert(0, str(BENCH))
+
+from scoreboard.claims_openemr import (  # noqa: E402
+    EXISTING,
+    TOKEN_ENV,
+    VERIFY_YAML,
+    _token_env,
+    reset_and_seed,
+)
+from scoreboard.tasks import openemr  # noqa: E402
+
+FIELDS = openemr.SQL_FIELDS
+
+
+def _sql_value(value: Any) -> str:
+    text = "" if value is None else str(value)
+    if not re.fullmatch(r"[A-Za-z0-9 ._@:/+-]*", text):
+        raise ValueError("unexpected characters in a synthetic value")
+    return f"'{text}'"
+
+
+class OpenEMRCopy:
+    """The fixture as a :class:`openadapt_flow.fault_gate.TestCopy`."""
+
+    name = "openemr_fixture"
+    key_field = "id"
+
+    def __init__(self, values: Mapping[str, str]) -> None:
+        self.values = dict(values)
+
+    def reset(self) -> None:
+        reset_and_seed()
+
+    def save(self) -> Any:
+        fx = openemr.fixture()
+        body = {k: v for k, v in self.values.items() if k != "state_label"}
+        body["state"] = "MA"
+        session = fx.token_session("actor")
+        try:
+            status = session.post(
+                f"{fx.api_base_url}/apis/default/api/patient", json=body
+            ).status_code
+        finally:
+            session.close()
+        if status not in (200, 201):
+            raise RuntimeError(f"the clean save returned HTTP {status}")
+        rows = [r for r in openemr.patients() if r["email"] == self.values["email"]]
+        return rows[-1]["id"]
+
+    def rows(self) -> list[dict[str, Any]]:
+        return [dict(r) for r in openemr.patients()]
+
+    def update(self, key: Any, changes: Mapping[str, Any]) -> None:
+        sets = ", ".join(
+            f"{name} = {_sql_value(value)}"
+            for name, value in changes.items()
+            if name in FIELDS and name != "id"
+        )
+        openemr.fixture()._db(
+            f"UPDATE openemr.patient_data SET {sets} WHERE id = {int(key)}"
+        )
+
+    def insert(self, record: Mapping[str, Any]) -> Any:
+        db = openemr.fixture()._db
+        donor = db("SELECT MIN(id) FROM openemr.patient_data").decode().split()[0]
+        sets = ", ".join(
+            f"{name} = {_sql_value(value)}"
+            for name, value in record.items()
+            if name in FIELDS and name != "id"
+        )
+        db(
+            "CREATE TEMPORARY TABLE openemr.vc_row AS SELECT * FROM "
+            f"openemr.patient_data WHERE id = {int(donor)}; "
+            "UPDATE openemr.vc_row SET id = (SELECT MAX(id) + 1 FROM "
+            "openemr.patient_data), pid = (SELECT MAX(pid) + 1 FROM "
+            "openemr.patient_data), uuid = UNHEX(REPLACE(UUID(), '-', ''))"
+            + (f", {sets}" if sets else "")
+            + "; INSERT INTO openemr.patient_data SELECT * FROM openemr.vc_row;"
+        )
+        return db("SELECT MAX(id) FROM openemr.patient_data").decode().split()[0]
+
+    def delete(self, key: Any) -> None:
+        openemr.fixture()._db(f"DELETE FROM openemr.patient_data WHERE id = {int(key)}")
+
+    def verifier_overrides(self) -> dict[str, Any]:
+        return {}
+
+
+def main() -> int:
+    import yaml
+
+    from openadapt_flow.fault_gate import run_gate
+    from openadapt_flow.verify_run import parse_verify_config
+
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--item", type=int, default=990)
+    args = parser.parse_args()
+    out = HERE / "results" / "openemr" / args.run_id
+    out.mkdir(parents=True, exist_ok=True)
+    fx = openemr.fixture()
+    values = openemr.item_values(args.item)
+    params = {k: v for k, v in values.items() if k != "state_label"}
+    os.environ.update(_token_env())
+    full = yaml.safe_load(VERIFY_YAML.format(api=fx.ui_base_url, token_env=TOKEN_ENV))
+    scoped = json.loads(json.dumps(full))
+    scoped["verifier"]["records_path"] = "/apis/default/api/patient?email={email}"
+    scoped["verifier"]["path_params"] = {"email": {"param": "email"}}
+    assert EXISTING["email"] != values["email"]
+    summary = {}
+    for name, doc in (("full", full), ("scoped", scoped)):
+        cfg = parse_verify_config(doc)
+        record = run_gate(cfg, params, OpenEMRCopy(values))
+        (out / f"gate-{name}.json").write_text(
+            json.dumps(record.model_dump(mode="json"), indent=2, sort_keys=True) + "\n"
+        )
+        summary[name] = {
+            "killed": record.killed,
+            "applicable": record.applicable,
+            "clean_pass": record.clean_pass,
+            "survived": [c["fault"] for c in record.cases if c["result"] == "SURVIVED"],
+        }
+        print(name, summary[name], flush=True)
+    fx.reset()
+    (out / "gate-summary.json").write_text(json.dumps(summary, indent=2) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())