diff --git a/CAPABILITIES.md b/CAPABILITIES.md index f66900f16..d911945c7 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -229,6 +229,7 @@ improvising or retrying when something is off. | Repair by default: once a repair agent is configured, every stop that wrote nothing goes to it without anyone starting it; a stop after a possible write goes to reconciliation; every stop is a row in a ledger with the funnel from stop to resumed item | On main ([#1360](https://github.com/OpenAdaptAI/OpenAdapt/pull/1360)) | Synthetic (MockMed, 22 drift modes, real Claude agent: 6 of 21 repairable stops resumed and verified, 0 wrong saves in 66 runs); routing over recorded OpenEMR stops; live openIMIS relabel: 3 of 3 reached repair, 0 resumed; silent-saves 1.1.0 UI-change blocks: 18 of 28 items that always stopped now correct, 0 wrong | `run` (on by default, `--no-auto-repair`), `repair:` in `deploy.yaml`, [benchmark/auto_repair](packages/flow/benchmark/auto_repair/README.md) | | Repair a relabeled save button: prove from the recorded identity hash that only the label changed (same control id, kind and form), re-qualify the fix on the new release, and have a person admit the certified copy afresh, since a save button's identity is part of the record check | On main | Real app (local OpenEMR, synthetic data): after the save button was relabeled, the stopped item re-ran and saved once after 8.3 operator minutes and 763 s of re-qualification; 896 s from stop to done ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | `openadapt_flow/repair/relabel.py`, [SELF_REPAIR.md](packages/flow/docs/SELF_REPAIR.md#a-relabeled-control-proof-that-only-the-label-changed) | | A duplicate check that reads the app's match list (`no_duplicate_candidate`): stop only on a listed record with the item's last name or date of birth, fail closed on an unreadable or unloaded list | On main | Real app (local OpenEMR, synthetic data): false stops per 100 items 70.4 before, 16.4 after; 0 silent wrongs in 30 saves ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | [DUPLICATE_CHECK.md](packages/flow/docs/DUPLICATE_CHECK.md) | +| A duplicate check that asks the system of record (generic FHIR R4 `Patient` search; OpenEMR REST and SQL adapters): every page read, stop on an exact or close match, proceed otherwise, fail closed on any read problem. One source for qualification, runs and resume; a program can pin it | On main | Synthetic: unit and HTTP tests; live OpenEMR 8.0.0.3 search decided 6 of 6 known referrals right through FHIR | [DUPLICATE_CHECK.md](packages/flow/docs/DUPLICATE_CHECK.md) | | A switched-off `problems_only` gate asks a person before the save instead of stopping the item | On main | Synthetic; real app in [#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513) | [APPROVAL_MODES.md](packages/flow/docs/APPROVAL_MODES.md#when-the-mode-turns-itself-off) | | Fleet drift patches: a verified fix with no patient data, shared across sites under policy | Released, local reference hub | Simulator | `openadapt-flow fleet`, [FLEET_DRIFT_PATCHES.md](packages/flow/docs/FLEET_DRIFT_PATCHES.md) | diff --git a/llms-full.txt b/llms-full.txt index acb6358d3..548258200 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -626,6 +626,7 @@ improvising or retrying when something is off. | Repair by default: once a repair agent is configured, every stop that wrote nothing goes to it without anyone starting it; a stop after a possible write goes to reconciliation; every stop is a row in a ledger with the funnel from stop to resumed item | On main ([#1360](https://github.com/OpenAdaptAI/OpenAdapt/pull/1360)) | Synthetic (MockMed, 22 drift modes, real Claude agent: 6 of 21 repairable stops resumed and verified, 0 wrong saves in 66 runs); routing over recorded OpenEMR stops; live openIMIS relabel: 3 of 3 reached repair, 0 resumed; silent-saves 1.1.0 UI-change blocks: 18 of 28 items that always stopped now correct, 0 wrong | `run` (on by default, `--no-auto-repair`), `repair:` in `deploy.yaml`, [benchmark/auto_repair](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/auto_repair/README.md) | | Repair a relabeled save button: prove from the recorded identity hash that only the label changed (same control id, kind and form), re-qualify the fix on the new release, and have a person admit the certified copy afresh, since a save button's identity is part of the record check | On main | Real app (local OpenEMR, synthetic data): after the save button was relabeled, the stopped item re-ran and saved once after 8.3 operator minutes and 763 s of re-qualification; 896 s from stop to done ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | `openadapt_flow/repair/relabel.py`, [SELF_REPAIR.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/SELF_REPAIR.md#a-relabeled-control-proof-that-only-the-label-changed) | | A duplicate check that reads the app's match list (`no_duplicate_candidate`): stop only on a listed record with the item's last name or date of birth, fail closed on an unreadable or unloaded list | On main | Real app (local OpenEMR, synthetic data): false stops per 100 items 70.4 before, 16.4 after; 0 silent wrongs in 30 saves ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | [DUPLICATE_CHECK.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/DUPLICATE_CHECK.md) | +| A duplicate check that asks the system of record (generic FHIR R4 `Patient` search; OpenEMR REST and SQL adapters): every page read, stop on an exact or close match, proceed otherwise, fail closed on any read problem. One source for qualification, runs and resume; a program can pin it | On main | Synthetic: unit and HTTP tests; live OpenEMR 8.0.0.3 search decided 6 of 6 known referrals right through FHIR | [DUPLICATE_CHECK.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/DUPLICATE_CHECK.md) | | A switched-off `problems_only` gate asks a person before the save instead of stopping the item | On main | Synthetic; real app in [#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513) | [APPROVAL_MODES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/APPROVAL_MODES.md#when-the-mode-turns-itself-off) | | Fleet drift patches: a verified fix with no patient data, shared across sites under policy | Released, local reference hub | Simulator | `openadapt-flow fleet`, [FLEET_DRIFT_PATCHES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/FLEET_DRIFT_PATCHES.md) | diff --git a/packages/flow/benchmark/dup_check/README.md b/packages/flow/benchmark/dup_check/README.md new file mode 100644 index 000000000..a2265980c --- /dev/null +++ b/packages/flow/benchmark/dup_check/README.md @@ -0,0 +1,163 @@ +# Duplicate check: screen against record, offline + +In the production soak (#1513), the duplicate check was the largest operator +cost after approvals. It stops a new patient who shares a last name or date +of birth with an existing chart. This measures the check that asks the system +of record instead (`runtime/checks/duplicate_record.py`) against the screen +rule, on the same stream and the same patient table. + +```bash +python -m benchmark.dup_check.measure --out DIR # per-item rows, kept out of the repo +``` + +The per-item rows of the run below are private, in openadapt-corpus +(`reliability_recipes/verification_coverage/dup_check_offline-2026-10-24/`); +this README keeps the aggregate tables. No app, no model calls. `measure.py` says exactly what each check sees. The +record check reads an in-memory table with FHIR search semantics +(`TableSource`): a case-insensitive name prefix and an exact birth date. + +## The exact numbers + +A miss is a second chart for a patient who already has one. + +**Test set:** 500 items from the soak's seed-7 generator +(`lifecycle.workload.generate(7, 500, 10)`), with planted duplicates and extra +charts: 694 items, 440 new patients and 241 true duplicates. +- 47 duplicates come from the generator itself: 12 existing patients referred + again, 19 resends, 10 same people with a mistyped date of birth, and 6 + registered by another clerk moments before. +- 194 are planted variants of existing patients. + +The set grew in four rounds. The first report's "12 of 110" was round one; +every number here is on the full 241. + +| | screen check | record check, first version | record check, now | +|---|---:|---:|---:| +| missed, of 241 | 42 | 90 | 24 | +| missed, of the 47 from the generator | 0 | 0 | 0 | +| false stops, of 440 new patients | 252 (36.3 per 100 items) | 2 (0.3) | 2 (0.3) | + +The soak's own 55 items, with the same planting (249 items, 202 true +duplicates): screen 42 missed and 10 false stops; record 24 missed and 0 +false stops. + +**Every kind** (planted count; missed by the screen, by the first version of +the record check, and by the record check now): + +| kind | planted | screen | v1 | now | what catches it now | +|---|---:|---:|---:|---:|---| +| from the generator (`dup_existing`, `dup_resend`, `ambiguous_same_person`, `concurrent_same`) | 47 | 0 | 0 | 0 | exact match, or same name with the date one field off | +| married name (`planted_married`) | 12 | 0 | 0 | 0 | same first name and date of birth | +| married name, new address (`pl_married_moved`) | 12 | 0 | 0 | 0 | same first name and date of birth | +| married name + mistyped day (`heldout_married_dob_typo`) | 12 | 12 | 12 | 0 | first name, a near date, a shared phone, street or email | +| hyphenated name (`planted_hyphenated`) | 12 | 0 | 0 | 0 | a shared surname part | +| second surname, "Garcia Lopez" for "Garcia" (`pl_second_surname`) | 6 | 0 | 0 | 0 | a shared surname part | +| day and month swapped (`planted_swapped_dob`) | 3 | 0 | 0 | 0 | same name, swapped date | +| same name, mistyped day (`pl_ambiguous_same_person`) | 12 | 0 | 0 | 0 | same name, date one field off | +| year digits transposed (`r2_year_transposed`) | 11 | 0 | 0 | 0 | same name, date one field off | +| typo in the first name (`heldout_first_typo`) | 12 | 0 | 0 | 0 | a similar name, same date | +| typo in the last name (`heldout_last_typo`) | 12 | 0 | 0 | 0 | a similar name, same date | +| typo in the last name + mistyped day (`rv_last_typo_dob_typo`) | 12 | 12 | 12 | 0 | same first name, similar last name, a near date | +| initial for the first name (`pl_initial`) | 12 | 0 | 12 | 0 | the initial rule | +| nickname, Bob for Robert (`r2_nickname`) | 6 | 0 | 6 | 0 | the nickname table | +| nickname + mistyped day (`rv_nickname_dob_typo`) | 6 | 0 | 6 | 0 | the nickname table, same last name, a near date | +| first and last name swapped (`r2_name_swap`) | 12 | 0 | 12 | 0 | the crossed names | +| both names misspelled, same contact (`pl_both_typos_same_contact`) | 12 | 0 | 0 | 0 | similar names, same date | +| prior name on the chart, moved, mistyped day (`r2_prior_name_moved_dob_typo`) | 6 | 6 | 6 | 0 | the chart's birth name | +| married name, moved, mistyped day (`limit_married_moved_dob_typo`) | 12 | 12 | 12 | **12** | known limit | +| new first and last name, moved, same day (`limit_new_name_moved_same_dob`) | 12 | 0 | 12 | **12** | known limit | + +The rows add to 241. Every match the record check makes is `close` or +`exact`, and both stop the item for a person. + +**Known limit: 24 misses, uncatchable without extra data.** +- 12 are a new last name with a new address, phone and email, and a mistyped + day of birth. They share only the first name with the chart. +- 12 are new first and last names with new contact details. They share only + the date of birth, as hundreds of patients do. + +Catching them needs an MRN, an insurance member id, or a person. The screen +stops the second kind only because it stops every new patient who shares a +date of birth with anyone. Stopping on a shared date of birth alone would +bring back the screen's `shared_dob` false stops (33 here). + +**The trade-off.** +- Misses fell from 90 (first version) to 24, and all 24 are the known limit. +- False stops stayed at 2 of 440 new patients. Both are `shared_dob` items + whose first name equals that of the chart born that day, which the + married-name rule sends to a person on purpose. +- One rule was narrowed along the way. "Another name, one shared contact + detail and a near date of birth" stopped two new patients, because the + generator's phone pool has only 100 numbers (`555-01xx`). It now needs two + shared contact details for a near date, or one for the same date. That + narrowing is tuned to this stream. + +## Live OpenEMR (2026-10-24) + +`live.py` on the pinned OpenEMR 8.0.0.3 fixture. It seeded 23 synthetic +charts, 13 of them born on one day, and probed both channels. Raw output +(`probe.json`, `decisions.json`) is private, in openadapt-corpus +`reliability_recipes/verification_coverage/live-2026-10-24/dup_check/`. + +What OpenEMR does: + +| | REST `/api/patient` | FHIR `Patient` | +|---|---|---| +| page size | `_limit` honored | `_count` honored | +| next page | `links.next`, but built on OpenEMR's configured site address, not the address the request went to; no total | no next link at all, and `Bundle.total` equals the rows returned, so `_count=5` silently returns 5 of 13 | +| with no page size | all 120 of 120 rows sharing a birth date | all 120 of 120 | +| `lname` / `family` | prefix match (`Hartwell` finds `Hartwell1`...) | prefix match | +| birth name | not exposed | a birth name isn't returned as an old name; searching it finds nothing | + +The check's decisions on six referrals with known answers: +- exact; +- a married name; +- a nickname; +- a swapped day and month; +- a new patient sharing only the birth date; +- a namesake. + +| channel | right | notes | +|---|---:|---| +| FHIR, as the check sends it (no `_count`) | 6 / 6 | | +| REST, 500 rows a page | 6 / 6 | | +| REST, 5 rows a page | 4 / 6 | the two that needed a second page were `unreadable`: the next link points to the site address, another origin, so the check refused it (fail closed) | + +So: +- REST stays opt-in. Its paging only works where OpenEMR's site address is + the address the check reads. +- `FhirPatientSource` never sends `_count`. OpenEMR returned all rows without + it (up to 120 tested), and the total can't reveal a cap, so a larger result + set than that is untested. +- Prior names are only searchable through the SQL adapter. + +## How the set was built + +For every existing chart, `plant` inserts duplicates at seeded positions: +- round one: a married name, a hyphenated name and a swapped date, then the + held-out first-name typo, last-name typo, and married name plus a mistyped + day; +- round two (`r2_`): a crossed name, a transposed year, and nicknames and + prior names on extra charts; +- round three (`pl_`): the six cases production-lifecycle named; +- round four (`rv_`): the two the review reproduced. + +Each item meets the existing and extra charts, every earlier item's right +outcome from the soak's truth, and another clerk's chart for +`concurrent_same`. + +The rules for each round were written knowing that round's kinds, so only +round one's held-out kinds were a blind test. They are why the second round +exists. + +## Limits + +- Offline. The channel adapters' paging and fail-closed behavior is tested + against a real local HTTP server (`tests/test_duplicate_record_check.py`), + not against OpenEMR. +- The search for prior names assumes a FHIR `family` search also matches a + patient's old names. The SQL adapter does (`birth_lname`). + `patient_history`'s previous names are not read. +- One seed. The screen rule is `no_duplicate_candidate` applied to OpenEMR's + listing as the soak describes it: any chart sharing the first name, the + last name or the date of birth. It is not a run of the browser. diff --git a/packages/flow/benchmark/dup_check/__init__.py b/packages/flow/benchmark/dup_check/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/packages/flow/benchmark/dup_check/live.py b/packages/flow/benchmark/dup_check/live.py new file mode 100644 index 000000000..e81b7a185 --- /dev/null +++ b/packages/flow/benchmark/dup_check/live.py @@ -0,0 +1,306 @@ +"""The record duplicate check against live OpenEMR: search semantics and paging. + + python -m benchmark.dup_check.live --out DIR + +Needs the fixture lease and ``OPENEMR_FIXTURE_STATE`` / ``OPENEMR_UI_PORT`` / +``OPENEMR_API_PORT``. No model calls. Resets the fixture first, then seeds 23 +synthetic charts through OpenEMR's REST API, 13 of them born on one day, +so a small page size forces several pages. + +It records what OpenEMR's channels actually do (``probe.json``): + +- REST ``/api/patient``: whether ``_limit`` caps a page, whether a reply + carries ``links.next`` or a total, the body's keys, and whether ``lname`` + matches exactly or by prefix; +- FHIR ``Patient``: whether ``_count`` pages with ``Bundle.link[next]`` and + ``Bundle.total``, whether ``family`` matches by prefix, and whether a stored + birth name comes back as an old name; + +and then runs ``check_duplicate`` through each channel on six referrals with +known answers (``decisions.json``). +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent)) + +from openadapt_flow.runtime.checks.duplicate_record import ( # noqa: E402 + FhirPatientSource, + RestPatientSource, + check_duplicate, +) + +DOB = "1977-05-14" +SCOPE = "openid api:oemr api:fhir user/patient.crus user/Patient.read user/Patient.rs" + + +def _people() -> list[dict[str, str]]: + base = { + "title": "Ms.", + "sex": "Female", + "street": "1 PROBE ROAD", + "city": "EXAMPLETON", + "state": "MA", + "postal_code": "02139", + "country_code": "USA", + } + out = [] + for i in range(13): # one shared birth day: forces pages at a small size + out.append( + { + **base, + "fname": f"Shared{i}", + "lname": f"Dayborn{i}", + "DOB": DOB, + "phone_home": f"555-06{i:02d}", + "email": f"shared.{i}@example.invalid", + } + ) + for i in range(6): # one last-name stem: prefix or exact matching + out.append( + { + **base, + "fname": f"Stem{i}", + "lname": f"Hartwell{'' if i == 0 else i}", + "DOB": f"1960-01-{i + 10:02d}", + "phone_home": f"555-07{i:02d}", + "email": f"stem.{i}@example.invalid", + } + ) + out.append( + { + **base, + "fname": "Rosalind", + "lname": "Okafor", + "DOB": "1990-04-09", + "phone_home": "617-555-0142", + "email": "r.okafor@example.invalid", + } + ) + out.append( + { + **base, + "fname": "Robert", + "lname": "Ashby", + "DOB": "1950-02-03", + "phone_home": "555-0800", + "email": "r.ashby@example.invalid", + } + ) + out.append( + { + **base, + "fname": "Ada", + "lname": "Quill", + "DOB": "1984-03-07", + "phone_home": "555-0801", + "email": "a.quill@example.invalid", + } + ) + out.append( + { + **base, + "fname": "Mabel", + "lname": "Ortiz", + "DOB": "1990-01-21", + "phone_home": "555-0802", + "email": "m.ortiz@example.invalid", + } + ) + return out + + +#: (referral, expected decision) +REFERRALS = [ + ({"fname": "Ada", "lname": "Quill", "DOB": "1984-03-07"}, "exact"), + ( + {"fname": "Rosalind", "lname": "Jesswood", "DOB": "1990-04-09"}, + "close", + ), # married + ({"fname": "Bob", "lname": "Ashby", "DOB": "1950-02-03"}, "close"), # nickname + ( + {"fname": "Ada", "lname": "Quill", "DOB": "1984-07-03"}, + "close", + ), # swapped day/month + ({"fname": "Nadia", "lname": "Fresh", "DOB": DOB}, "none"), # shares only the day + ({"fname": "Tomas", "lname": "Hartwell", "DOB": "1999-12-01"}, "none"), # namesake +] + + +def main() -> int: + from verification_coverage.independent import live # the fixture helpers + + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--out", type=Path, required=True) + args = parser.parse_args() + args.out.mkdir(parents=True, exist_ok=True) + fx = live.fx() + live.reset() + clients = json.loads(fx.clients_path.read_text()) + live.sql( + f"UPDATE openemr.oauth_clients SET scope = {live.q(SCOPE)} " + f"WHERE client_id = {live.q(clients['actor_client_id'])};" + ) + import httpx + + values = fx._runtime_values() + session = httpx.Client(verify=False, timeout=30) + resp = session.post( + f"{fx.api_base_url}/oauth2/default/token", + data={ + "grant_type": "password", + "client_id": clients["actor_client_id"], + "scope": SCOPE, + "user_role": "users", + "username": values["OPENEMR_ACTOR_USER"], + "password": values["OPENEMR_ACTOR_PASSWORD"], + }, + ) + if resp.status_code != 200: + raise SystemExit(f"token: HTTP {resp.status_code}") + token = resp.json()["access_token"] + auth = {"Authorization": f"Bearer {token}"} + api = f"{fx.api_base_url}/apis/default/api" + fhir = f"{fx.api_base_url}/apis/default/fhir" + for person in _people(): + r = session.post(f"{api}/patient", json=person, headers=auth) + if r.status_code not in (200, 201): + raise SystemExit(f"seed: HTTP {r.status_code} {r.text[:200]}") + # A married chart keeps her birth name, as OpenEMR's demographics allow. + live.sql( + "UPDATE openemr.patient_data SET birth_fname = 'Rosalind', " + "birth_lname = 'Jesswood' WHERE email = 'r.okafor@example.invalid'" + ) + + probe: dict[str, Any] = {} + + def rest(params: dict[str, Any]) -> dict[str, Any]: + r = session.get(f"{api}/patient", params=params, headers=auth) + body = ( + r.json() + if r.headers.get("content-type", "").startswith("application/json") + else {} + ) + data = body.get("data") if isinstance(body, dict) else None + return { + "status": r.status_code, + "keys": sorted(body) if isinstance(body, dict) else type(body).__name__, + "rows": len(data) if isinstance(data, list) else None, + "links": body.get("links") if isinstance(body, dict) else None, + "lnames": sorted({d.get("lname") for d in data or []}) + if isinstance(data, list) + else None, + } + + probe["rest_dob_all"] = rest({"DOB": DOB}) + probe["rest_dob_limit5"] = rest({"DOB": DOB, "_limit": 5}) + probe["rest_dob_limit5_offset5"] = rest({"DOB": DOB, "_limit": 5, "_offset": 5}) + probe["rest_lname_exact"] = rest({"lname": "Hartwell"}) + probe["rest_no_match"] = rest({"DOB": "1801-01-01"}) + + def fhir_search(params: dict[str, Any]) -> dict[str, Any]: + pages, total, ids, url, q = 0, None, [], f"{fhir}/Patient", dict(params) + links_seen = [] + while url and pages < 20: + r = session.get( + url, params=q, headers={**auth, "Accept": "application/fhir+json"} + ) + pages += 1 + if r.status_code != 200: + return {"status": r.status_code, "text": r.text[:200]} + b = r.json() + total = b.get("total", total) + ids += [e["resource"]["id"] for e in b.get("entry", []) or []] + nxt = [ + link["url"] + for link in b.get("link", []) + if link.get("relation") == "next" + ] + links_seen.append([link.get("relation") for link in b.get("link", [])]) + url, q = (nxt[0] if nxt else None), None + return { + "pages": pages, + "total": total, + "read": len(ids), + "unique": len(set(ids)), + "link_relations": links_seen[:3], + } + + probe["fhir_birthdate"] = fhir_search({"birthdate": DOB}) + probe["fhir_birthdate_count5"] = fhir_search({"birthdate": DOB, "_count": 5}) + probe["fhir_family_prefix"] = fhir_search({"family": "Hartwell"}) + r = session.get( + f"{fhir}/Patient", + params={"family": "Jesswood"}, + headers={**auth, "Accept": "application/fhir+json"}, + ) + b = r.json() if r.status_code == 200 else {} + probe["fhir_family_birth_name"] = { + "status": r.status_code, + "total": b.get("total"), + "names": [e["resource"].get("name") for e in b.get("entry", []) or []][:2], + } + (args.out / "probe.json").write_text( + json.dumps(probe, indent=2, sort_keys=True) + "\n" + ) + print(json.dumps(probe, indent=1, sort_keys=True)[:4000]) + + decisions = [] + channels = { + "fhir_page5": lambda: FhirPatientSource( + fhir, + token=lambda: token, + verify_tls=False, + session=httpx.Client(verify=False, timeout=30), + ), + "rest_page5": lambda: RestPatientSource( + fx.api_base_url, + headers=auth, + page_size=5, + session=httpx.Client(verify=False, timeout=30), + ), + "rest_page500": lambda: RestPatientSource( + fx.api_base_url, + headers=auth, + session=httpx.Client(verify=False, timeout=30), + ), + } + for name, make in channels.items(): + for want, expected in REFERRALS: + result = check_duplicate(make(), want) + decisions.append( + { + "channel": name, + "referral": want, + "expected": expected, + "decision": result.decision, + "reason": result.reason, + "pages": result.pages, + "candidates": [c.why for c in result.candidates], + } + ) + print( + name, + want["fname"], + want["lname"], + want["DOB"], + "expected", + expected, + "->", + result.decision, + "|", + result.reason[:90], + ) + (args.out / "decisions.json").write_text(json.dumps(decisions, indent=2) + "\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/benchmark/dup_check/live_rates.py b/packages/flow/benchmark/dup_check/live_rates.py new file mode 100644 index 000000000..817210839 --- /dev/null +++ b/packages/flow/benchmark/dup_check/live_rates.py @@ -0,0 +1,163 @@ +"""False stops and missed duplicates through live OpenEMR FHIR, on the soak's stream. + + python -m benchmark.dup_check.live_rates --out DIR [--long-items 500] + +The offline measurement (``measure.py``) reads an in-memory table with FHIR +search semantics. This one asks a real OpenEMR 8.0.0.3: it resets the pinned +fixture, creates the existing charts through OpenEMR's REST API (and stores +each planted chart's birth name the way OpenEMR's demographics do), then for +every item in the planted seed-7 stream runs ``check_duplicate`` through +``FhirPatientSource`` and, when the right outcome is a new chart, creates it. +So the record check meets the same charts as offline, served by OpenEMR. + +Needs the fixture lease and ``OPENEMR_FIXTURE_STATE`` / ``OPENEMR_UI_PORT`` / +``OPENEMR_API_PORT``. No model calls. Writes ``rows.json`` (per item, private +evidence) and ``summary.json`` to ``--out``. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path +from typing import Any + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent)) + +from benchmark.dup_check import measure # noqa: E402 +from openadapt_flow.runtime.checks.duplicate_record import ( # noqa: E402 + FhirPatientSource, + check_duplicate, +) + +SCOPE = "openid api:oemr api:fhir user/patient.crus user/Patient.read user/Patient.rs" +API_FIELDS = ( + "title", + "fname", + "lname", + "DOB", + "sex", + "street", + "city", + "postal_code", + "phone_home", + "email", + "country_code", +) + + +def main() -> int: + from verification_coverage.independent import live # the fixture helpers + + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--long-items", type=int, default=500) + parser.add_argument("--budget-s", type=float, default=3000.0) + args = parser.parse_args() + args.out.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + + import httpx + + fx = live.fx() + live.reset() + clients = json.loads(fx.clients_path.read_text()) + live.sql( + f"UPDATE openemr.oauth_clients SET scope = {live.q(SCOPE)} " + f"WHERE client_id = {live.q(clients['actor_client_id'])};" + ) + values = fx._runtime_values() + session = httpx.Client(verify=False, timeout=30) + tokens: dict[str, Any] = {} + + def token() -> str: + # Tokens live an hour; take a fresh one every 30 minutes. + if not tokens or time.monotonic() - tokens["at"] > 1800: + r = session.post( + f"{fx.api_base_url}/oauth2/default/token", + data={ + "grant_type": "password", + "client_id": clients["actor_client_id"], + "scope": SCOPE, + "user_role": "users", + "username": values["OPENEMR_ACTOR_USER"], + "password": values["OPENEMR_ACTOR_PASSWORD"], + }, + ) + r.raise_for_status() + tokens.update(value=r.json()["access_token"], at=time.monotonic()) + return str(tokens["value"]) + + def create(person: dict[str, str]) -> None: + body = {k: str(person.get(k, "") or "") for k in API_FIELDS} + body["DOB"] = measure.iso(body["DOB"]) + body["state"] = "MA" + r = session.post( + f"{fx.api_base_url}/apis/default/api/patient", + json=body, + headers={"Authorization": f"Bearer {token()}"}, + ) + if r.status_code not in (200, 201): + raise RuntimeError(f"create: HTTP {r.status_code} {r.text[:200]}") + if person.get("prior_lname"): + live.sql( + "UPDATE openemr.patient_data SET birth_fname = " + f"{live.q(person['fname'])}, birth_lname = {live.q(person['prior_lname'])} " + f"WHERE email = {live.q(person['email'])}" + ) + + existing, items = measure.from_generator(7, args.long_items, 10) + extra, extra_items = measure.extra_existing(existing) + stream = measure.plant(existing, items + extra_items, 7) + for person in existing + extra: + create(person) + source = FhirPatientSource( + f"{fx.api_base_url}/apis/default/fhir", + token=token, + session=httpx.Client(verify=False, timeout=30), + ) + rows = [] + for item in stream: + if time.monotonic() - started > args.budget_s: + raise SystemExit( + f"stopped at the {args.budget_s:.0f} s budget after {len(rows)} items" + ) + for person in item.before: + create(person) + params = { + k: (v.strip() if isinstance(v, str) else v) for k, v in item.params.items() + } + params["DOB"] = measure.iso(params.get("DOB", "")) + result = check_duplicate(source, params) + is_dup = item.exception in measure.DUPLICATE_KINDS + rows.append( + { + "item": item.item_id, + "exception": item.exception, + "right": "new chart" + if item.intended is not None + else ("no chart: duplicate" if is_dup else "no chart"), + "record": result.decision, + "record_why": [c.why for c in result.candidates], + "reason": result.reason if result.decision == "unreadable" else "", + } + ) + if item.intended is not None: + create(item.intended) + summary = measure.summarize([{**r, "screen": "proceed"} for r in rows]) + summary.pop("screen", None) + summary["unreadable_reasons"] = sorted({r["reason"] for r in rows if r["reason"]}) + summary["seconds"] = round(time.monotonic() - started) + (args.out / "rows.json").write_text(json.dumps(rows, indent=1) + "\n") + (args.out / "summary.json").write_text( + json.dumps(summary, indent=2, sort_keys=True) + "\n" + ) + print(json.dumps(summary, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/benchmark/dup_check/measure.py b/packages/flow/benchmark/dup_check/measure.py new file mode 100644 index 000000000..512a85876 --- /dev/null +++ b/packages/flow/benchmark/dup_check/measure.py @@ -0,0 +1,516 @@ +"""Screen duplicate check against record duplicate check, offline, on the soak's stream. + + python -m benchmark.dup_check.measure --out DIR + +The stream is the production soak's seed-7 workload (#1513): the committed +55-item stream it ran, and a longer seed-7 stream from the same generator. +Planted into each are true duplicates the soak lacks: an existing patient +referred again under a married name, under a hyphenated name, and with day +and month swapped. + +The patient table each item meets is the existing charts plus every earlier +item's right outcome (the soak's truth), plus another clerk's chart for +``concurrent_same``. So both checks see the same records, and only the check +differs. + +- **screen**: the rule ``no_duplicate_candidate`` applies to OpenEMR's + possible-duplicate list (``production_arm.with_duplicate_list_check``): + OpenEMR lists every chart that shares the first name, the last name or the + date of birth, and the check stops when a listed row carries every word of + the run's last name or shows its date of birth. +- **record**: ``runtime.checks.duplicate_record.check_duplicate`` over the + same table with FHIR search semantics; ``exact`` and ``close`` stop. + +A *false stop* is a stop on an item whose right outcome is a new chart. A +*missed duplicate* is a proceed on an item whose right outcome is no new +chart because the patient already has one. Dates are converted to ISO first, +as intake does since #1514. +""" + +from __future__ import annotations + +import argparse +import json +import random +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +from benchmark.lifecycle import workload as wl +from openadapt_flow.runtime.checks.duplicate_record import ( + PatientRecord, + TableSource, + check_duplicate, +) + +HERE = Path(__file__).resolve().parent +SOAK = HERE.parent / "lifecycle" / "results" / "production-2026-10-24" / "after" +#: Right outcome "no new chart" because the person already has one. +DUPLICATE_KINDS = { + "dup_existing", + "dup_resend", + "ambiguous_same_person", + "concurrent_same", + "planted_married", + "planted_hyphenated", + "planted_swapped_dob", + "heldout_first_typo", + "heldout_last_typo", + "heldout_married_dob_typo", + "r2_nickname", + "r2_name_swap", + "r2_year_transposed", + "r2_prior_name_moved_dob_typo", + "pl_married_moved", + "pl_second_surname", + "pl_initial", + "pl_both_typos_same_contact", + "pl_ambiguous_same_person", + "rv_nickname_dob_typo", + "rv_last_typo_dob_typo", + "limit_married_moved_dob_typo", + "limit_new_name_moved_same_dob", +} +#: Formal first names given to extra existing charts, and the nickname each +#: is referred under (``r2_nickname``). +FORMAL = ( + ("Robert", "Bob"), + ("Elizabeth", "Liz"), + ("William", "Bill"), + ("Katherine", "Kate"), + ("Theodore", "Theo"), + ("Margaret", "Peggy"), +) +MARRIED = ("Okafor", "Lindqvist", "Brennan", "Moreau", "Castellanos", "Whitcombe") + + +@dataclass +class Item: + item_id: str + exception: str + params: dict[str, str] + #: The chart the right outcome creates, or None. + intended: Optional[dict[str, str]] + #: Charts that appear just before the item (another clerk's). + before: list[dict[str, str]] + + +def iso(value: str) -> str: + m = re.fullmatch(r"(\d{2})/(\d{2})/(\d{4})", value or "") + return f"{m.group(3)}-{m.group(1)}-{m.group(2)}" if m else (value or "") + + +def from_ledger(path: Path) -> tuple[list[dict[str, str]], list[Item]]: + existing, items = [], [] + for line in path.read_text().splitlines(): + row = json.loads(line) + if row["type"] == "existing_chart": + existing.append({k: v for k, v in row.items() if k != "type"}) + elif row["type"] == "item": + t = row["truth"] + before = [ + i["person"] + for i in t["inject"] + if i.get("event") == "other_clerk_creates" + ] + items.append( + Item( + row["item_id"], t["exception"], row["params"], t["intended"], before + ) + ) + return existing, items + + +def from_generator( + seed: int, n: int, days: int +) -> tuple[list[dict[str, str]], list[Item]]: + w = wl.generate(seed, n, days) + items = [] + for it in w.items: + t = w.truth[it.item_id] + before = [ + i["person"] for i in t.inject if i.get("event") == "other_clerk_creates" + ] + items.append(Item(it.item_id, t.exception, dict(it.params), t.intended, before)) + return list(w.existing), items + + +def plant(existing: list[dict[str, str]], items: list[Item], seed: int) -> list[Item]: + """Insert true duplicates of existing patients at seeded positions.""" + + rng = random.Random(seed) + planted: list[Item] = [] + for n, person in enumerate(existing): + married = dict(person, lname=MARRIED[n % len(MARRIED)]) + planted.append(Item(f"plant-m-{n}", "planted_married", married, None, [])) + hyph = dict( + person, lname=f"{person['lname']}-{MARRIED[(n + 1) % len(MARRIED)]}" + ) + planted.append(Item(f"plant-h-{n}", "planted_hyphenated", hyph, None, [])) + y, m, d = person["DOB"].split("-") + if int(d) <= 12 and d != m: + planted.append( + Item( + f"plant-s-{n}", + "planted_swapped_dob", + dict(person, DOB=f"{y}-{d}-{m}"), + None, + [], + ) + ) + # Held out: variants the record check's rules were not written for. + typo = lambda s: s[:-2] + s[-1] + s[-2] if len(s) > 3 else s + "e" # noqa: E731 + planted.append( + Item( + f"held-f-{n}", + "heldout_first_typo", + dict(person, fname=typo(person["fname"])), + None, + [], + ) + ) + planted.append( + Item( + f"held-l-{n}", + "heldout_last_typo", + dict(person, lname=typo(person["lname"])), + None, + [], + ) + ) + planted.append( + Item( + f"held-md-{n}", + "heldout_married_dob_typo", + dict( + person, + lname=MARRIED[(n + 2) % len(MARRIED)], + DOB=f"{y}-{m}-{int(d) % 28 + 1:02d}", + ), + None, + [], + ) + ) + planted += round_two(n, person) + out = list(items) + for p in planted: + out.insert(rng.randint(0, len(out)), p) + return out + + +def moved(person: dict[str, str], n: int) -> dict[str, str]: + """The same person at a new address, phone and email.""" + + return dict( + person, + street=f"{900 + n} NEW PLACE", + phone_home=f"555-02{n:02d}", + email=f"moved.{n}@example.invalid", + ) + + +def day_typo(dob: str) -> str: + y, m, d = dob.split("-") + return f"{y}-{m}-{int(d) % 28 + 1:02d}" + + +def round_two(n: int, person: dict[str, str]) -> list[Item]: + """Kinds added after the first measurement (rules written for them, + except the two ``limit_`` kinds, which no rule here can catch).""" + + out = [ + Item( + f"r2-sw-{n}", + "r2_name_swap", + dict(person, fname=person["lname"], lname=person["fname"]), + None, + [], + ) + ] + y, m, d = person["DOB"].split("-") + ty = y[:2] + y[3] + y[2] + if ty != y and 1900 <= int(ty) <= 2010: + out.append( + Item( + f"r2-ty-{n}", + "r2_year_transposed", + dict(person, DOB=f"{ty}-{m}-{d}"), + None, + [], + ) + ) + + # The six cases production-lifecycle named (``pl_``). + def typo(v: str) -> str: + return v[:-2] + v[-1] + v[-2] if len(v) > 3 else v + "e" + + # The review's reproduced gap (``rv_``): a last-name typo plus a day typo. + out.append( + Item( + f"rv-lt-{n}", + "rv_last_typo_dob_typo", + dict(person, lname=typo(person["lname"]), DOB=day_typo(person["DOB"])), + None, + [], + ) + ) + + out += [ + Item( + f"pl-mm-{n}", + "pl_married_moved", + dict(moved(person, n), lname=MARRIED[(n + 4) % len(MARRIED)]), + None, + [], + ), + Item( + f"pl-in-{n}", + "pl_initial", + dict(person, fname=person["fname"][0] + "."), + None, + [], + ), + Item( + f"pl-bt-{n}", + "pl_both_typos_same_contact", + dict( + person, + fname=typo(person["fname"]), + lname=typo(person["lname"]), + street=f"{n} SOMEWHERE ELSE", + ), + None, + [], + ), + Item( + f"pl-as-{n}", + "pl_ambiguous_same_person", + dict(person, DOB=day_typo(person["DOB"])), + None, + [], + ), + ] + out.append( + Item( + f"lim-mm-{n}", + "limit_married_moved_dob_typo", + dict( + moved(person, n), + lname=MARRIED[(n + 3) % len(MARRIED)], + DOB=day_typo(person["DOB"]), + ), + None, + [], + ) + ) + out.append( + Item( + f"lim-nn-{n}", + "limit_new_name_moved_same_dob", + dict(moved(person, n), fname="Unrelated", lname="Renamed"), + None, + [], + ) + ) + return out + + +def extra_existing( + existing: list[dict[str, str]], +) -> tuple[list[dict[str, str]], list[Item]]: + """Charts with formal first names or a stored prior name, and their + true-duplicate referrals.""" + + extra, items = [], [] + for n, (formal, nick) in enumerate(FORMAL): + base = existing[n % len(existing)] + chart = dict( + base, + fname=formal, + lname=f"Formal{n}", + email=f"formal.{n}@example.invalid", + phone_home=f"555-03{n:02d}", + street=f"{700 + n} FORMAL ROAD", + ) + extra.append(chart) + items.append( + Item(f"r2-nk-{n}", "r2_nickname", dict(chart, fname=nick), None, []) + ) + items.append( + Item( + f"rv-nd-{n}", + "rv_nickname_dob_typo", + dict(chart, fname=nick, DOB=day_typo(chart["DOB"])), + None, + [], + ) + ) + for n in range(6): + base = existing[(n + 5) % len(existing)] + chart = dict( + base, + fname=f"Second{n}", + lname=f"Garcia{n} Lopez{n}", + email=f"second.{n}@example.invalid", + phone_home=f"555-05{n:02d}", + street=f"{600 + n} SECOND STREET", + ) + extra.append(chart) + items.append( + Item( + f"pl-ss-{n}", + "pl_second_surname", + dict(moved(chart, 70 + n), lname=f"Garcia{n}"), + None, + [], + ) + ) + for n in range(6): + base = existing[(n + 3) % len(existing)] + chart = dict( + base, + lname=f"Married{n}", + prior_lname=f"Maiden{n}", + email=f"prior.{n}@example.invalid", + phone_home=f"555-04{n:02d}", + street=f"{800 + n} PRIOR LANE", + DOB=base["DOB"], + ) + extra.append(chart) + referral = dict( + moved(chart, 50 + n), lname=f"Maiden{n}", DOB=day_typo(chart["DOB"]) + ) + referral.pop("prior_lname") + items.append( + Item(f"r2-pn-{n}", "r2_prior_name_moved_dob_typo", referral, None, []) + ) + return extra, items + + +def record_of(i: int, person: dict[str, str]) -> PatientRecord: + prior = ( + ((person["fname"], person["prior_lname"]),) if person.get("prior_lname") else () + ) + return PatientRecord( + str(i), + person["fname"], + person["lname"], + person.get("DOB", ""), + phone=person.get("phone_home", ""), + street=person.get("street", ""), + email=person.get("email", ""), + prior_names=prior, + ) + + +def words(value: str) -> list[str]: + return re.findall(r"[0-9a-z]+", (value or "").casefold()) + + +def screen_stops(params: dict[str, str], table: list[PatientRecord]) -> bool: + """The no_duplicate_candidate rule on OpenEMR's possible-duplicate list.""" + + lname, dob = params.get("lname", ""), params.get("DOB", "") + if not words(lname) or not dob: + return True # a missing parameter fails the check closed + fname = params.get("fname", "") + listed = [ + r + for r in table + if words(r.fname) == words(fname) + or words(r.lname) == words(lname) + or r.dob == dob + ] + for r in listed: + row = set(words(f"{r.fname} {r.lname} {r.dob}")) + if all(w in row for w in words(lname)) or r.dob == dob: + return True + return False + + +def measure(existing: list[dict[str, str]], items: list[Item]) -> dict[str, Any]: + table = [record_of(i, p) for i, p in enumerate(existing)] + rows = [] + for item in items: + for person in item.before: + table.append(record_of(len(table), person)) + params = { + k: (v.strip() if isinstance(v, str) else v) for k, v in item.params.items() + } + params["DOB"] = iso(params.get("DOB", "")) + is_dup = item.exception in DUPLICATE_KINDS + new_chart = item.intended is not None + screen = screen_stops(params, table) + result = check_duplicate(TableSource(table), params) + rows.append( + { + "item": item.item_id, + "exception": item.exception, + "right": "new chart" + if new_chart + else ("no chart: duplicate" if is_dup else "no chart"), + "screen": "stop" if screen else "proceed", + "record": result.decision, + "record_why": [c.why for c in result.candidates], + } + ) + if item.intended is not None: + table.append(record_of(len(table), item.intended)) + return {"rows": rows, "summary": summarize(rows)} + + +def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]: + out: dict[str, Any] = {"items": len(rows)} + new = [r for r in rows if r["right"] == "new chart"] + dups = [r for r in rows if r["right"] == "no chart: duplicate"] + out["new_chart_items"], out["true_duplicates"] = len(new), len(dups) + for check, stops in ( + ("screen", lambda r: r["screen"] == "stop"), + ("record", lambda r: r["record"] != "none"), + ): + false_stops = [r for r in new if stops(r)] + missed = [r for r in dups if not stops(r)] + out[check] = { + "false_stops": len(false_stops), + "false_stops_per_100_items": round(100 * len(false_stops) / len(rows), 1), + "false_stops_by_exception": _count(false_stops), + "missed_duplicates": len(missed), + "missed_by_exception": _count(missed), + "caught_duplicates": len(dups) - len(missed), + } + out["record"]["decisions"] = _count(rows, key="record") + return out + + +def _count(rows: list[dict[str, Any]], key: str = "exception") -> dict[str, int]: + counts: dict[str, int] = {} + for r in rows: + counts[r[key]] = counts.get(r[key], 0) + 1 + return dict(sorted(counts.items())) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + # Per-item rows are raw evidence: write them outside the repository. + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--long-items", type=int, default=500) + args = parser.parse_args() + args.out.mkdir(parents=True, exist_ok=True) + runs = {} + for name, (existing, items) in ( + ("soak-seed7-55", from_ledger(SOAK / "workload.jsonl")), + (f"seed7-{args.long_items}", from_generator(7, args.long_items, 10)), + ): + extra, extra_items = extra_existing(existing) + runs[name] = measure(existing + extra, plant(existing, items + extra_items, 7)) + for name, run in runs.items(): + (args.out / f"{name}.json").write_text( + json.dumps(run, indent=2, sort_keys=True) + "\n" + ) + print(name, json.dumps(run["summary"], sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/benchmark/lifecycle/production_arm.py b/packages/flow/benchmark/lifecycle/production_arm.py index 3e1dfcbe0..f13c03145 100644 --- a/packages/flow/benchmark/lifecycle/production_arm.py +++ b/packages/flow/benchmark/lifecycle/production_arm.py @@ -772,6 +772,9 @@ def with_duplicate_list_check(source: Path, out: Path) -> Path: candidate_rows="tr.oneresult", candidate_frame_path=("#modalframe",), candidate_params=("lname", "DOB"), + # Qualified with the record check (OpenEMR's patient API): a run + # without that channel fails the step instead of reading the list. + candidate_source="record", ) for pc in step.expect ] diff --git a/packages/flow/docs/DUPLICATE_CHECK.md b/packages/flow/docs/DUPLICATE_CHECK.md index 02fffc8ae..82c2d223b 100644 --- a/packages/flow/docs/DUPLICATE_CHECK.md +++ b/packages/flow/docs/DUPLICATE_CHECK.md @@ -61,3 +61,114 @@ on the OpenEMR referral program, and the soak admits that version afresh `04/20/1944`, ISO), and a row date that reads either way counts when either reading matches. Other values are matched word for word, so name only parameters the list shows as they were typed. + +## Asking the system of record instead + +`no_duplicate_candidate` can search the system of record for the patient +instead of reading the on-screen list (`runtime/checks/duplicate_record.py`, a +check whose source is the record). + +The channels: + +| Source | Scope | Used | +|---|---|---| +| `FhirPatientSource` | generic FHIR R4: `Patient` search with standard parameters, for any R4 server | whenever the run's effect verifier reads FHIR | +| `RestPatientSource` | OpenEMR adapter: OpenEMR's standard REST `/api/patient` | only for a program pinned to the record check (below) | +| `SqlPatientSource` | OpenEMR adapter, fallback: a read-only query on `patient_data` (MySQL or MariaDB `%s` parameters; `paramstyle="qmark"` for SQLite) | only when wired by hand | + +**One source for every path.** `Replayer` resolves the source itself, from +its own effect verifier and the program's pin (`resolve_duplicate_source`). +So qualification, a run, a resume and the lifecycle proof that read the same +system of record use the same check: +- no construction path can choose another; +- a qualification hook can't set one; +- a verifier the resolution doesn't recognize gives no source rather than an + error. + +`tests/test_duplicate_source_paths.py` scans every `Replayer(...)` +construction to keep it so. + +**The program decides, and qualification sees it.** The postcondition's +`candidate_source` is part of the qualified program: +- `record`: the record check, with the FHIR or OpenEMR REST channel. A run + with no record channel fails the step and never reads the screen instead. +- `screen`: the on-screen list, even when a channel exists. +- unset: the FHIR channel when the run has one, else the screen. + +**Choosing the channel.** `candidate_channel` picks one: +- `fhir` with an OpenEMR REST record check (`.../apis/default/api/patient`) + searches OpenEMR's FHIR API on the same server (`.../apis/default/fhir`), + with the record check's bearer token. That token's client then needs the + FHIR read scope, `api:fhir user/Patient.read`. Without it every search is + refused, and the check is `unreadable`. +- `rest` uses only the REST adapter. +- Unset, the record check's own channel decides. + +**Parameter names.** The check reads `fname`, `lname`, `DOB`, and when +present `mrn`, `phone_home`, `email` and `street`: OpenEMR's referral names. +A program for another app maps its own with `candidate_record_fields`, for +example `{"fname": "given", "lname": "family", "DOB": "birth_date"}`. + +The check searches: +- by the date of birth, and by the date with day and month swapped; +- by each part of the last and first name, and by the first name; +- by MRN, phone and email when the run has them. + +It reads every page, and it decides from whole records: + +| Decision | When | What happens | +|---|---|---| +| `exact` | same first name, last name and date of birth, or the same MRN | stop: the chart exists | +| `close` | see the list below | stop for a person; the halt names each record and why | +| `none` | no record is this patient | proceed | +| `unreadable` | see the list below | stop (fail closed) | + +A record is `close` when it has: +- the same or a similar name, a nickname, an initial or the chart's prior + name, with day and month swapped, or with the date of birth one field off + and at most one name part differing (or a shared phone, email or street); +- the same date of birth with a similar name or a shared surname part; +- first and last name swapped; +- the same date of birth and first name under another last name; +- another name with the same date of birth and a shared contact detail, or a + near date of birth and two. + +A record date that is only a year or a year and month counts as no date of +birth, so a name match is `close`, never a non-match. + +The check is `unreadable` when: +- the channel is unreachable, or refuses the credentials; +- the reply is not a result set, or a REST body carries any error field or + no `data` list; +- a reported total doesn't match what was read; +- the search runs past 50 pages, or a REST page is full with no next link; +- a next-page link points to another host; +- a record has no id; +- the run's date of birth is missing or not ISO. + +Sharing only a last name, or only a date of birth, doesn't stop the item. + +**The stop reason.** A record-check stop reads `failed: no_duplicate_candidate +(record check via : , ...)`. The channel is +`none` when a pinned program had no channel. Decision `unreadable` is a +problem with the program or its channel. `exact` and `close` are about the +item. + +**On OpenEMR 8.0.0.3** (measured live, `benchmark/dup_check/README.md`): +- REST pages with `links.next`, but the link carries OpenEMR's configured site + address. The check refuses a link to another origin, so on a deployment + whose site address differs from the address the check reads, a + multi-page REST answer is `unreadable`. +- FHIR returns no next link, and its `Bundle.total` equals the rows returned. + The check sends no `_count`, and OpenEMR returned every row (up to 120 + tested). +- Both match last names by prefix. A birth name is not searchable through + FHIR. + +Not catchable from the referral alone: a person whose name changed, who moved, +and whose date of birth is mistyped or merely shared. Nothing the referral +carries ties them to the chart; that needs an MRN, an insurance id, or a +person. + +Measured offline on the production soak's seed-7 stream, with true +duplicates planted: `benchmark/dup_check/README.md`. diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index 9a3fd2cbd..619063b3f 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -514,6 +514,21 @@ class Postcondition(BaseModel): candidate_params: Optional[tuple[str, ...]] = Field( default=None, min_length=1, max_length=8 ) + #: NO_DUPLICATE_CANDIDATE: where the check's evidence must come from. + #: ``record``: the system of record, and a run with no record channel + #: fails the step (never the screen instead). ``screen``: the listed rows, + #: even when a record channel exists. Unset: the record when the run has + #: a record channel, else the screen. + candidate_source: Optional[Literal["record", "screen"]] = None + #: NO_DUPLICATE_CANDIDATE, record source: the run parameter that fills + #: each field the record check reads (``fname``, ``lname``, ``DOB``, + #: ``mrn``, ``phone_home``, ``email``, ``street``). Unset: those names. + candidate_record_fields: Optional[dict[str, str]] = None + #: NO_DUPLICATE_CANDIDATE, record source: which channel to search. + #: ``fhir`` with an OpenEMR REST record check searches OpenEMR's FHIR API + #: on the same server; ``rest``, OpenEMR's REST API only. Unset: the + #: channel of the run's record check. + candidate_channel: Optional[Literal["fhir", "rest"]] = None @model_serializer(mode="wrap") def _serialize_compatible(self, handler: Any) -> dict[str, Any]: @@ -525,7 +540,14 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: data.pop("field_selector", None) if self.param is None: data.pop("param", None) - for name in ("candidate_rows", "candidate_frame_path", "candidate_params"): + for name in ( + "candidate_rows", + "candidate_frame_path", + "candidate_params", + "candidate_source", + "candidate_record_fields", + "candidate_channel", + ): if getattr(self, name) is None: data.pop(name, None) return data @@ -4381,6 +4403,14 @@ class StepResult(BaseModel): #: on that path alone; None otherwise, and then omitted from the #: serialized result. scroll_target_ready_before: Optional[bool] = None + #: Set when ``no_duplicate_candidate`` failed this step: ``exact`` or + #: ``close`` (a record matched), ``candidate`` (a listed row matched on the + #: screen): about the item; ``unreadable`` (the record channel or the + #: screen list could not answer, or a record-pinned check had no + #: channel): about the program. None otherwise, and then omitted. + duplicate_check: Optional[Literal["exact", "close", "candidate", "unreadable"]] = ( + None + ) @model_serializer(mode="wrap") def _serialize_compatible(self, handler: Any) -> dict[str, Any]: @@ -4399,6 +4429,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: data.pop("check_to_click", None) if isinstance(data, dict) and self.scroll_target_ready_before is None: data.pop("scroll_target_ready_before", None) + if isinstance(data, dict) and self.duplicate_check is None: + data.pop("duplicate_check", None) return data diff --git a/packages/flow/openadapt_flow/qualify/campaign.py b/packages/flow/openadapt_flow/qualify/campaign.py index d07913cad..0fa9401e0 100644 --- a/packages/flow/openadapt_flow/qualify/campaign.py +++ b/packages/flow/openadapt_flow/qualify/campaign.py @@ -756,7 +756,12 @@ def _run( kwargs["checkpoint_key"] = crypto.resolve_key(None) if hooks.replayer_kwargs is not None: - kwargs.update(hooks.replayer_kwargs(ctx)) + extra = dict(hooks.replayer_kwargs(ctx)) + if "duplicate_record_source" in extra: + # The duplicate check is resolved from the run's own verifier + # on every path; a hook can't qualify under another one. + raise ValueError("replayer_kwargs may not set duplicate_record_source") + kwargs.update(extra) # The signed lookup settings win over any hook, so a hook can't turn # structure back on during a pass that must not use it. kwargs.update(lookup or {}) diff --git a/packages/flow/openadapt_flow/runtime/checks/__init__.py b/packages/flow/openadapt_flow/runtime/checks/__init__.py new file mode 100644 index 000000000..c8610527a --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/checks/__init__.py @@ -0,0 +1,14 @@ +"""Checks tagged with where their evidence comes from. + +``docs/design/DECOMPOSITION.md`` S5: a check reads either the screen or the +system of record. This package holds checks whose source is the record. +""" + +from __future__ import annotations + +from typing import Literal + +#: Where a check's evidence comes from. +CheckSource = Literal["screen", "record"] + +__all__ = ["CheckSource"] diff --git a/packages/flow/openadapt_flow/runtime/checks/duplicate_record.py b/packages/flow/openadapt_flow/runtime/checks/duplicate_record.py new file mode 100644 index 000000000..1993e7acf --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/checks/duplicate_record.py @@ -0,0 +1,1088 @@ +"""Is this new patient already in the system of record? Asked of the record. + +The screen check (``no_duplicate_candidate``) reads the app's possible- +duplicate list and stops on any row that shares the last name or the date of +birth. Every new patient who shares either with an existing chart stops for a +person, and that is most of the operator time in the production soak. + +This check asks the system of record instead (FHIR ``Patient`` search, the +app's REST API, or, as a documented fallback, a read-only SQL query) and +decides from whole records: + +- ``exact``: same first name, last name and date of birth, or the same MRN. + Stop: the chart exists. +- ``close``: a record that may be this person. Stop for a person, with both + records shown. Close is any of: + - the same name with a date of birth one field off (a typo) or with day and + month swapped; + - the same date of birth with a similar first name and a similar or + hyphen-sharing last name; + - the same date of birth and the same first name, whatever the last name + (a married or changed name). +- ``none``: no record is this person. Proceed. +- ``unreadable``: the channel was unreachable, refused the credentials, + returned something that is not a result set, or could not be read to the + last page. Stop (fail closed). + +Sharing only a last name, or only a date of birth, is not a candidate: those +are the ordinary coincidences a practice sees. +""" + +from __future__ import annotations + +import re +import unicodedata +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Literal, Mapping, Optional, Protocol +from urllib.parse import urljoin, urlsplit + +from openadapt_flow.runtime.checks import CheckSource + +Decision = Literal["none", "exact", "close", "unreadable"] + +#: The fields the check reads, and the run parameter that fills each one by +#: default (OpenEMR's referral parameter names). A program for another app +#: maps its own names with ``candidate_record_fields``. +RECORD_FIELDS: dict[str, str] = { + "fname": "fname", + "lname": "lname", + "DOB": "DOB", + "mrn": "mrn", + "phone_home": "phone_home", + "email": "email", + "street": "street", +} + +#: A paging chain longer than this is read as unreadable, not truncated. +MAX_PAGES = 50 + + +@dataclass(frozen=True) +class PatientRecord: + """One chart as the record channel returns it.""" + + record_id: str + fname: str + lname: str + dob: str # ISO YYYY-MM-DD, or "" when the record has none + mrn: str = "" + phone: str = "" + street: str = "" + email: str = "" + #: Earlier names, (first, last): a birth or maiden name, a previous name. + prior_names: tuple[tuple[str, str], ...] = () + + +@dataclass +class Candidate: + record: PatientRecord + match: Literal["exact", "close"] + why: str + + +@dataclass +class DuplicateCheckResult: + decision: Decision + source: CheckSource = "record" + channel: str = "" + reason: str = "" + candidates: list[Candidate] = field(default_factory=list) + #: Search requests made and pages read, for the receipt. + searches: int = 0 + pages: int = 0 + + @property + def proceed(self) -> bool: + return self.decision == "none" + + +class Unreadable(Exception): + """The record channel could not give a complete answer.""" + + +class PatientSource(Protocol): + """A read-only patient search on the system of record.""" + + channel: str + + def by_birthdate(self, dob: str) -> list[PatientRecord]: ... + + def by_family(self, family: str) -> list[PatientRecord]: ... + + def by_identifier(self, mrn: str) -> list[PatientRecord]: ... + + def by_given(self, given: str) -> list[PatientRecord]: ... + + def by_phone(self, phone: str) -> list[PatientRecord]: ... + + def by_email(self, email: str) -> list[PatientRecord]: ... + + +# -- matching ------------------------------------------------------------------- + + +def norm_name(value: str) -> str: + """Case, accents, spaces and punctuation removed (hyphens kept).""" + + text = unicodedata.normalize("NFKD", value or "") + text = "".join(c for c in text if not unicodedata.combining(c)).casefold() + text = re.sub(r"[^a-z\- ]", "", text) + return re.sub(r"\s+", " ", text).strip() + + +def name_parts(value: str) -> set[str]: + return {p for p in re.split(r"[\s\-]+", norm_name(value)) if p} + + +def edit_distance(a: str, b: str) -> int: + """Damerau-Levenshtein (optimal string alignment).""" + + d = [[0] * (len(b) + 1) for _ in range(len(a) + 1)] + for i in range(len(a) + 1): + d[i][0] = i + for j in range(len(b) + 1): + d[0][j] = j + for i in range(1, len(a) + 1): + for j in range(1, len(b) + 1): + cost = 0 if a[i - 1] == b[j - 1] else 1 + d[i][j] = min(d[i - 1][j] + 1, d[i][j - 1] + 1, d[i - 1][j - 1] + cost) + if i > 1 and j > 1 and a[i - 1] == b[j - 2] and a[i - 2] == b[j - 1]: + d[i][j] = min(d[i][j], d[i - 2][j - 2] + 1) + return d[len(a)][len(b)] + + +def soundex(value: str) -> str: + letters = re.sub(r"[^a-z]", "", norm_name(value)) + if not letters: + return "" + codes = { + c: str(n) + for n, group in enumerate(("bfpv", "cgjkqsxz", "dt", "l", "mn", "r"), start=1) + for c in group + } + out, last = letters[0].upper(), codes.get(letters[0], "") + for c in letters[1:]: + code = codes.get(c, "") + if code and code != last: + out += code + if c not in "hw": + last = code + return (out + "000")[:4] + + +def similar_name(a: str, b: str) -> bool: + """Equal, one or two edits apart (by length), or the same Soundex code.""" + + x, y = norm_name(a).replace("-", " "), norm_name(b).replace("-", " ") + if not x or not y: + return False + if x == y: + return True + limit = 1 if min(len(x), len(y)) <= 5 else 2 + if edit_distance(x, y) <= limit: + return True + return len(x) > 3 and len(y) > 3 and soundex(x) == soundex(y) + + +def swapped(dob: str) -> Optional[str]: + """The date with day and month swapped, when that is another valid date.""" + + y, m, d = dob.split("-") + if m == d or int(d) > 12: + return None + return f"{y}-{d}-{m}" + + +def one_field_off(a: str, b: str) -> bool: + """Two ISO dates that differ in exactly one of year, month and day.""" + + pa, pb = a.split("-"), b.split("-") + return sum(x != y for x, y in zip(pa, pb)) == 1 + + +#: Common English given-name equivalents. Generic, not drawn from any test +#: workload; a deployment can extend it (``NICKNAMES.update``). +_NICKNAME_GROUPS = ( + "robert bob bobby rob robbie", + "william bill billy will willie liam", + "elizabeth liz beth betty eliza lizzie libby", + "katherine kate kathy katie kat", + "catherine cathy cate kate", + "margaret maggie meg peggy marge greta", + "james jim jimmy jamie", + "john jack johnny jon", + "richard rick ricky dick rich", + "michael mike mikey mick", + "thomas tom tommy", + "joseph joe joey", + "charles charlie chuck chas", + "edward ed eddie ted ned", + "anthony tony", + "daniel dan danny", + "christopher chris kit", + "alexander alex sasha xander", + "alexandra alex sandra sasha lexi", + "patricia pat patty trish tricia", + "jennifer jen jenny", + "rebecca becky becca", + "susan sue susie suzy", + "deborah deb debbie", + "nicholas nick nicky", + "benjamin ben benny", + "samuel sam sammy", + "samantha sam sammie", + "matthew matt matty", + "andrew andy drew", + "stephen steve stevie", + "steven steve stevie", + "jonathan jon jonny", + "timothy tim timmy", + "theodore theo ted teddy", + "dorothy dot dottie dolly", + "victoria vicky tori", + "abigail abby gail", + "gregory greg", + "kenneth ken kenny", + "ronald ron ronnie", + "donald don donnie", + "lawrence larry laurie", + "frederick fred freddie", + "isabella bella izzy isabel", + "jacqueline jackie", + "josephine jo josie", + "eleanor ellie nora nell", + "francis frank frankie", + "frances fran frankie", + "henry hank harry hal", + "harold harry hal", + "raymond ray", + "rafael rafa", + "dmitri dima", + "anastasia ana stacy", + "ana anna annie", + "ines ine", + "nadia nadya", + "mateo mat teo", +) +NICKNAMES: dict[str, set[str]] = {} +for _group in _NICKNAME_GROUPS: + _names = set(_group.split()) + for _name in _names: + NICKNAMES.setdefault(_name, set()).update(_names) + + +def same_given(a: str, b: str) -> bool: + """The same first name: similar spelling, an initial, or a known nickname.""" + + if similar_name(a, b): + return True + x, y = norm_name(a), norm_name(b) + # An initial ("J." for James) stands for any name with that first letter. + if x and y and (len(x) == 1 or len(y) == 1) and x[0] == y[0]: + return True + return bool(x and y and y in NICKNAMES.get(x, ())) + + +def dob_near(a: str, b: str) -> bool: + """A mistyped date of birth: one field off, day and month swapped, or two + digits of one field transposed (1984 and 1948).""" + + if not a or not b or a == b: + return False + return one_field_off(a, b) or swapped(a) == b + + +def _digits(value: str) -> str: + return re.sub(r"\D", "", value or "") + + +def _street(value: str) -> str: + return re.sub( + r"\s+", " ", re.sub(r"[^0-9a-z ]", "", (value or "").casefold()) + ).strip() + + +def contacts_shared(want: Mapping[str, str], record: PatientRecord) -> list[str]: + """Every contact detail the referral and the chart share.""" + + out = [] + phone = _digits(want.get("phone_home") or "") + if len(phone) >= 7 and phone == _digits(record.phone): + out.append("same phone") + email = (want.get("email") or "").strip().casefold() + if email and email == record.email.strip().casefold(): + out.append("same email") + street = _street(want.get("street") or "") + if street and street == _street(record.street): + out.append("same street address") + return out + + +def tie_breaker(want: Mapping[str, str], record: PatientRecord) -> Optional[str]: + """A contact detail the referral and the chart share, if any.""" + + phone = _digits(want.get("phone_home") or "") + if len(phone) >= 7 and phone == _digits(record.phone): + return "same phone" + email = (want.get("email") or "").strip().casefold() + if email and email == record.email.strip().casefold(): + return "same email" + street = _street(want.get("street") or "") + if street and street == _street(record.street): + return "same street address" + return None + + +def classify(want: Mapping[str, str], record: PatientRecord) -> Optional[Candidate]: + """Is ``record`` this new patient? + + ``want``: fname, lname, DOB, and when known mrn, phone_home, email, + street. The record's prior names are tried as well as its current one. + """ + + mrn = (want.get("mrn") or "").strip() + if mrn and record.mrn and mrn == record.mrn.strip(): + return Candidate(record, "exact", "same MRN") + names = [(record.fname, record.lname), *record.prior_names] + for i, (first, last) in enumerate(names): + found = _classify_name(want, record, first, last) + if found is not None: + if i: + found.why += " (a prior name on the chart)" + if found.match == "exact": + found.match = "close" + return found + # A full name change: only the contact details tie the two together. + shared = contacts_shared(want, record) + if ( + shared + and record.dob + and ( + record.dob == want["DOB"] + or (len(shared) >= 2 and dob_near(want["DOB"], record.dob)) + ) + ): + return Candidate( + record, + "close", + f"another name; {', '.join(shared)}; " + + ("the same" if record.dob == want["DOB"] else "a near") + + " date of birth", + ) + return None + + +def _classify_name( + want: Mapping[str, str], record: PatientRecord, first: str, last: str +) -> Optional[Candidate]: + fname, lname, dob = want["fname"], want["lname"], want["DOB"] + same_first = norm_name(fname) == norm_name(first) + same_last = norm_name(lname) == norm_name(last) + if same_first and same_last and dob == record.dob: + return Candidate( + record, "exact", "same first name, last name and date of birth" + ) + first_like = same_given(fname, first) + last_like = similar_name(lname, last) or bool(name_parts(lname) & name_parts(last)) + near = dob_near(dob, record.dob) + if first_like and last_like: + if not record.dob: + return Candidate( + record, "close", "same name; the record has no date of birth" + ) + if record.dob == dob: + return Candidate(record, "close", "same date of birth, a similar name") + if record.dob == swapped(dob): + return Candidate(record, "close", "same name, day and month swapped") + # A mistyped date of birth with a name that differs in at most one + # part (a nickname, an initial or a typo in the other), or with a + # shared contact detail. + if near and (same_first or same_last or tie_breaker(want, record)): + return Candidate( + record, "close", "same or similar name, a mistyped date of birth" + ) + swapped_names = norm_name(fname) == norm_name(last) and norm_name( + lname + ) == norm_name(first) + if swapped_names and (record.dob == dob or near): + return Candidate(record, "close", "first and last name swapped") + if record.dob == dob and first_like: + return Candidate( + record, "close", "same date of birth and first name, another last name" + ) + if near and first_like and tie_breaker(want, record): + return Candidate( + record, + "close", + f"same first name, another last name, {tie_breaker(want, record)}, " + "a mistyped date of birth", + ) + return None + + +ISO_DATE = re.compile(r"\d{4}-\d{2}-\d{2}") + + +def full_date(value: Any) -> str: + """A record's date of birth as full ISO, or "" when it has none or only a + year or a year and month. A partial date is never a non-match: the + record is judged as one with no date of birth, so a name match is close.""" + + text = str(value or "").strip()[:10] + return text if ISO_DATE.fullmatch(text) else "" + + +def check_duplicate( + source: PatientSource, want: Mapping[str, str] +) -> DuplicateCheckResult: + """Ask ``source`` whether the patient in ``want`` already has a chart.""" + + result = DuplicateCheckResult("unreadable", channel=source.channel) + fname, lname = (want.get("fname") or "").strip(), (want.get("lname") or "").strip() + dob = (want.get("DOB") or "").strip() + if not fname or not lname or not ISO_DATE.fullmatch(dob): + result.reason = ( + "the run's first name, last name or ISO date of birth is missing" + ) + return result + queries: list[tuple[Callable[[str], list[PatientRecord]], str]] = [ + (source.by_birthdate, dob) + ] + if swapped(dob): + queries.append((source.by_birthdate, swapped(dob))) # type: ignore[arg-type] + # Each part of the last name, and of the first name (a swapped name). + queries += [ + (source.by_family, part) + for part in sorted(name_parts(lname) | name_parts(fname)) + ] + queries.append((source.by_given, fname)) + if (want.get("mrn") or "").strip(): + queries.append((source.by_identifier, want["mrn"].strip())) + if len(_digits(want.get("phone_home") or "")) >= 7: + queries.append((source.by_phone, want["phone_home"].strip())) + if (want.get("email") or "").strip(): + queries.append((source.by_email, want["email"].strip())) + seen: dict[str, PatientRecord] = {} + try: + for query, value in queries: + for record in query(value): + if not record.record_id.strip(): + raise Unreadable("a record came back with no id") + seen.setdefault(record.record_id, record) + result.searches += 1 + except Unreadable as exc: + result.reason = str(exc) + return result + result.pages = getattr(source, "pages_read", 0) + found = [c for c in (classify(want, r) for r in seen.values()) if c is not None] + result.candidates = sorted( + found, key=lambda c: (c.match != "exact", c.record.record_id) + ) + if any(c.match == "exact" for c in found): + result.decision = "exact" + result.reason = "a chart for this patient exists" + elif found: + result.decision = "close" + result.reason = "a chart may be this patient; a person decides" + else: + result.decision = "none" + result.reason = f"no chart matches among {len(seen)} record(s) read" + return result + + +# -- channels ------------------------------------------------------------------- + + +def _get_json( + session: Any, url: str, params: Optional[dict], headers: dict, timeout_s: float +) -> tuple[Any, str]: + try: + resp = session.get(url, params=params, headers=headers, timeout=timeout_s) + except Exception as exc: # noqa: BLE001 - transport failure is unreadable + raise Unreadable(f"the record channel is unreachable ({type(exc).__name__})") + if resp.status_code in (401, 403): + raise Unreadable( + f"the record channel refused the credentials (HTTP {resp.status_code})" + ) + if resp.status_code // 100 != 2: + raise Unreadable(f"the record channel returned HTTP {resp.status_code}") + try: + return resp.json(), str(getattr(resp, "url", url)) + except Exception: # noqa: BLE001 - unparseable body is unreadable + raise Unreadable("the record channel returned a body that is not JSON") + + +def _same_origin(base: str, url: str) -> str: + """``url`` if it is on ``base``'s scheme, host and port, else unreadable: + a next-page link to another host must not receive the credentials.""" + + a, b = urlsplit(base), urlsplit(url) + if (a.scheme, a.netloc) != (b.scheme, b.netloc): + raise Unreadable("a next-page link points to another host") + return url + + +class FhirPatientSource: + """Generic FHIR R4: ``Patient`` search, every page (``Bundle.link[next]``). + + Uses only standard R4 search parameters (``birthdate``, ``family``, + ``given``, ``identifier``, ``phone``, ``email``), so it serves any R4 + server, OpenEMR's included. + + A page chain longer than :data:`MAX_PAGES`, a ``Bundle.total`` that does + not equal the entries read, or an entry that is not a ``Patient`` makes + the answer unreadable. + """ + + channel = "fhir" + + def __init__( + self, + base_url: str, + *, + token: Optional[Callable[[], Optional[str]]] = None, + session: Any = None, + timeout_s: float = 10.0, + verify_tls: bool = True, + mrn_system: Optional[str] = None, + ) -> None: + self.base_url = base_url.rstrip("/") + self._token = token + self._session = session + self.timeout_s = timeout_s + self.verify_tls = verify_tls + self.mrn_system = mrn_system + self.pages_read = 0 + + def _client(self) -> Any: + if self._session is None: + import httpx + + self._session = httpx.Client(verify=self.verify_tls) + return self._session + + def _headers(self) -> dict[str, str]: + headers = {"Accept": "application/fhir+json"} + if self._token is not None: + token = self._token() + if not token: + raise Unreadable("no access token for the record channel") + headers["Authorization"] = f"Bearer {token}" + return headers + + def search(self, params: dict[str, str]) -> list[PatientRecord]: + url: Optional[str] = f"{self.base_url}/Patient" + query: Optional[dict[str, str]] = params + total: Optional[int] = None + out: list[PatientRecord] = [] + pages = 0 + while url is not None: + if pages >= MAX_PAGES: + raise Unreadable( + f"more than {MAX_PAGES} pages; the search was not read to the end" + ) + body, at = _get_json( + self._client(), url, query, self._headers(), self.timeout_s + ) + pages += 1 + self.pages_read += 1 + if not isinstance(body, dict) or body.get("resourceType") != "Bundle": + raise Unreadable("the record channel did not return a FHIR Bundle") + if total is None and isinstance(body.get("total"), int): + total = body["total"] + entries = body.get("entry", []) + if not isinstance(entries, list): + raise Unreadable("the Bundle's entries are not a list") + for entry in entries: + resource = entry.get("resource") if isinstance(entry, dict) else None + if ( + not isinstance(resource, dict) + or resource.get("resourceType") != "Patient" + ): + raise Unreadable("a Bundle entry is not a Patient") + out.append(self._record(resource)) + url, query = None, None + for link in body.get("link", []) or []: + if isinstance(link, dict) and link.get("relation") == "next": + raw = link.get("url") + if not isinstance(raw, str) or not raw.strip(): + raise Unreadable("a next-page link has no URL") + url = _same_origin(self.base_url, urljoin(at, raw.strip())) + if total is not None and total != len(out): + raise Unreadable( + f"the search reported {total} record(s) and {len(out)} were read" + ) + return out + + def _record(self, resource: dict[str, Any]) -> PatientRecord: + names = [n for n in resource.get("name") or [] if isinstance(n, dict)] + current = [n for n in names if n.get("use") not in ("old", "maiden")] + official = next( + (n for n in current if n.get("use") == "official"), + current[0] if current else {}, + ) + + def pair(n: dict[str, Any]) -> tuple[str, str]: + given = n.get("given") or [] + return (str(given[0]) if given else "", str(n.get("family") or "")) + + mrn = "" + for ident in resource.get("identifier") or []: + if not isinstance(ident, dict): + continue + codes = [ + c.get("code") for c in (ident.get("type") or {}).get("coding", []) or [] + ] + if (self.mrn_system and ident.get("system") == self.mrn_system) or ( + not self.mrn_system and ("MR" in codes or "PT" in codes) + ): + mrn = str(ident.get("value") or "") + break + telecom = [t for t in resource.get("telecom") or [] if isinstance(t, dict)] + phone = next( + ( + str(t.get("value") or "") + for t in telecom + if t.get("system") == "phone" and t.get("use") in (None, "home") + ), + "", + ) + email = next( + (str(t.get("value") or "") for t in telecom if t.get("system") == "email"), + "", + ) + address = next( + (a for a in resource.get("address") or [] if isinstance(a, dict)), {} + ) + lines = address.get("line") or [] + first, last = pair(official) + prior = tuple( + pair(n) for n in names if n is not official and pair(n) != (first, last) + ) + return PatientRecord( + record_id=str(resource.get("id") or ""), + fname=first, + lname=last, + dob=full_date(resource.get("birthDate")), + mrn=mrn, + phone=phone, + street=str(lines[0]) if lines else "", + email=email, + prior_names=prior, + ) + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + return self.search({"birthdate": dob}) + + def by_family(self, family: str) -> list[PatientRecord]: + return self.search({"family": family}) + + def by_identifier(self, mrn: str) -> list[PatientRecord]: + return self.search({"identifier": mrn}) + + def by_given(self, given: str) -> list[PatientRecord]: + return self.search({"given": given}) + + def by_phone(self, phone: str) -> list[PatientRecord]: + return self.search({"phone": phone}) + + def by_email(self, email: str) -> list[PatientRecord]: + return self.search({"email": email}) + + +class SqlPatientSource: + """OpenEMR adapter, fallback: a read-only query on ``patient_data``. + + The columns (``pid``, ``pubpid``, ``DOB``, ``birth_lname``, ...) are + OpenEMR's. + + For a deployment with no FHIR or REST read. ``connect`` returns a DB-API + connection with a SELECT-only account. OpenEMR runs on MySQL or MariaDB, + so parameters default to ``%s`` (``paramstyle="format"``); pass + ``paramstyle="qmark"`` for SQLite. The SQL itself is portable (the + wildcard is bound, not concatenated). + """ + + channel = "sql" + + def __init__( + self, + connect: Callable[[], Any], + *, + table: str = "patient_data", + paramstyle: Literal["format", "qmark"] = "format", + ) -> None: + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.]{0,63}", table): + raise ValueError("table must be a plain identifier") + self._connect = connect + self.table = table + self.mark = "%s" if paramstyle == "format" else "?" + self.pages_read = 0 + + def _query(self, where: str, *values: str) -> list[PatientRecord]: + try: + conn = self._connect() + except Exception as exc: # noqa: BLE001 + raise Unreadable(f"the database is unreachable ({type(exc).__name__})") + try: + cur = conn.cursor() + cur.execute( + "SELECT pid, fname, lname, DOB, pubpid, phone_home, street, email, " + f"birth_fname, birth_lname FROM {self.table} " + f"WHERE {where.replace('?', self.mark)}", + values, + ) + rows = cur.fetchall() + except Exception as exc: # noqa: BLE001 + raise Unreadable(f"the database query failed ({type(exc).__name__})") + finally: + try: + conn.close() + except Exception: # noqa: BLE001 + pass + self.pages_read += 1 + out = [] + for pid, f, ln, d, mrn, phone, street, email, bf, bl in rows: + prior = ((str(bf or f or ""), str(bl)),) if bl else () + out.append( + PatientRecord( + str(pid), + str(f or ""), + str(ln or ""), + full_date(d), + str(mrn or ""), + str(phone or ""), + str(street or ""), + str(email or ""), + prior, + ) + ) + return out + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + return self._query("DOB = ?", dob) + + def by_family(self, family: str) -> list[PatientRecord]: + # FHIR's string search is a case-insensitive prefix; match it. + prefix = family.replace("%", "").replace("_", "") + "%" + return self._query( + "LOWER(lname) LIKE LOWER(?) OR LOWER(birth_lname) LIKE LOWER(?)", + prefix, + prefix, + ) + + def by_identifier(self, mrn: str) -> list[PatientRecord]: + return self._query("pubpid = ?", mrn) + + def by_given(self, given: str) -> list[PatientRecord]: + return self._query("LOWER(fname) = LOWER(?)", given) + + def by_phone(self, phone: str) -> list[PatientRecord]: + return self._query("phone_home = ?", phone) + + def by_email(self, email: str) -> list[PatientRecord]: + return self._query("LOWER(email) = LOWER(?)", email) + + +class TableSource: + """An in-memory patient table with FHIR search semantics (offline use).""" + + channel = "table" + + def __init__(self, records: Iterable[PatientRecord]) -> None: + self.records = list(records) + self.pages_read = 0 + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + self.pages_read += 1 + return [r for r in self.records if r.dob == dob] + + def by_family(self, family: str) -> list[PatientRecord]: + self.pages_read += 1 + prefix = norm_name(family) + return [ + r + for r in self.records + if any( + norm_name(last).startswith(prefix) + for last in (r.lname, *(n[1] for n in r.prior_names)) + ) + ] + + def by_identifier(self, mrn: str) -> list[PatientRecord]: + self.pages_read += 1 + return [r for r in self.records if r.mrn and r.mrn == mrn] + + def by_given(self, given: str) -> list[PatientRecord]: + self.pages_read += 1 + prefix = norm_name(given) + return [r for r in self.records if norm_name(r.fname).startswith(prefix)] + + def by_phone(self, phone: str) -> list[PatientRecord]: + self.pages_read += 1 + return [ + r for r in self.records if r.phone and _digits(r.phone) == _digits(phone) + ] + + def by_email(self, email: str) -> list[PatientRecord]: + self.pages_read += 1 + return [r for r in self.records if r.email.casefold() == email.casefold()] + + +class RestPatientSource: + """OpenEMR adapter: its standard REST API, ``GET .../api/patient?DOB=...``. + + The query parameters and fields (``fname``, ``lname``, ``DOB``, + ``pubpid``, ``phone_home``) are OpenEMR's. + + OpenEMR answers with ``{"data": [...]}``. A next-page link + (``links.next``) is followed. A full page with no next link may be + truncated, so it makes the answer unreadable rather than incomplete. + """ + + channel = "rest" + + def __init__( + self, + base_url: str, + *, + path: str = "/apis/default/api/patient", + headers: Optional[dict[str, str]] = None, + session: Any = None, + timeout_s: float = 10.0, + page_size: int = 500, + ) -> None: + self.base_url = base_url.rstrip("/") + self.path = path.split("?", 1)[0] + self.headers = dict(headers or {}) + self._session = session + self.timeout_s = timeout_s + self.page_size = page_size + self.pages_read = 0 + + def _client(self) -> Any: + if self._session is None: + import httpx + + self._session = httpx.Client() + return self._session + + def search(self, params: dict[str, str]) -> list[PatientRecord]: + url: Optional[str] = f"{self.base_url}{self.path}" + query: Optional[dict[str, Any]] = {**params, "_limit": self.page_size} + out: list[PatientRecord] = [] + pages = 0 + while url is not None: + if pages >= MAX_PAGES: + raise Unreadable( + f"more than {MAX_PAGES} pages; the search was not read to the end" + ) + body, at = _get_json( + self._client(), url, query, self.headers, self.timeout_s + ) + pages += 1 + self.pages_read += 1 + if not isinstance(body, dict): + raise Unreadable("the patient search did not return an object") + errors = [ + k + for k in ("validationErrors", "internalErrors", "errors", "error") + if body.get(k) + ] + if errors: + raise Unreadable("the patient search reported " + ", ".join(errors)) + rows = body.get("data") + if not isinstance(rows, list): + raise Unreadable("the patient search did not return a data list") + for row in rows: + if not isinstance(row, dict): + raise Unreadable("a patient row is not an object") + out.append( + PatientRecord( + record_id=str( + row.get("uuid") or row.get("pid") or row.get("id") or "" + ), + fname=str(row.get("fname") or ""), + lname=str(row.get("lname") or ""), + dob=full_date(row.get("DOB")), + mrn=str(row.get("pubpid") or ""), + phone=str(row.get("phone_home") or ""), + street=str(row.get("street") or ""), + email=str(row.get("email") or ""), + prior_names=( + ( + ( + str( + row.get("birth_fname") or row.get("fname") or "" + ), + str(row["birth_lname"]), + ), + ) + if row.get("birth_lname") + else () + ), + ) + ) + links = body.get("links") if isinstance(body, dict) else None + nxt = links.get("next") if isinstance(links, dict) else None + if isinstance(nxt, str) and nxt.strip(): + url = _same_origin(self.base_url, urljoin(at, nxt.strip())) + query = None + elif len(rows) >= self.page_size: + raise Unreadable( + "a full page with no next link; the search may be truncated" + ) + else: + url = None + return out + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + return self.search({"DOB": dob}) + + def by_family(self, family: str) -> list[PatientRecord]: + # OpenEMR matches lname exactly here, so compare prefixes locally + # against an exact read: a hyphenated or married name is still found + # through the date of birth. + return self.search({"lname": family}) + + def by_identifier(self, mrn: str) -> list[PatientRecord]: + return self.search({"pubpid": mrn}) + + def by_given(self, given: str) -> list[PatientRecord]: + return self.search({"fname": given}) + + def by_phone(self, phone: str) -> list[PatientRecord]: + return self.search({"phone_home": phone}) + + def by_email(self, email: str) -> list[PatientRecord]: + return self.search({"email": email}) + + +def _unwrap(verifier: Any) -> Any: + """The record verifier inside a known wrapper, else ``verifier``.""" + + seen = 0 + while verifier is not None and seen < 4: + inner = None + for name in ("inner", "_inner", "wrapped", "_wrapped", "verifier", "_verifier"): + candidate = getattr(verifier, name, None) + if candidate is not None and candidate is not verifier: + inner = candidate + break + if inner is None: + return verifier + verifier, seen = inner, seen + 1 + return verifier + + +def source_for_verifier( + verifier: Any, + *, + allow_rest: bool = False, + channel: Optional[Literal["fhir", "rest"]] = None, +) -> Optional[PatientSource]: + """The patient search on the same channel as a run's record check. + + A FHIR verifier gives a generic FHIR R4 ``Patient`` search at its base URL + with its credentials. An OpenEMR REST verifier on ``.../api/patient`` + gives the OpenEMR REST adapter, but only with ``allow_rest`` (a program + pinned to the record check). Anything else, or a verifier missing the + attributes these need, gives None. + + ``channel`` (the program's ``candidate_channel``) chooses: ``fhir`` with + an OpenEMR REST verifier searches OpenEMR's FHIR API on the same server + (``.../apis//fhir``) with the verifier's bearer token, which then + needs the FHIR read scope (``api:fhir user/Patient.read``); ``rest`` + uses only the REST adapter. + """ + + from openadapt_flow.runtime.effects import FhirEffectVerifier + from openadapt_flow.runtime.effects.rest import RestRecordVerifier + + verifier = _unwrap(verifier) + try: + if channel == "rest" and not isinstance(verifier, RestRecordVerifier): + return None + if isinstance(verifier, FhirEffectVerifier): + access_token = getattr(verifier, "access_token", None) + token_source = getattr(verifier, "token_source", None) + + def token() -> Optional[str]: + if token_source is not None: + return token_source.access_token() + return access_token + + return FhirPatientSource( + verifier.base_url, + token=token if (access_token or token_source) else None, + verify_tls=getattr(verifier, "verify_tls", True), + timeout_s=max(float(getattr(verifier, "timeout_s", 10.0)), 10.0), + ) + records_path = str(getattr(verifier, "records_path", "") or "") + if ( + channel == "fhir" + and isinstance(verifier, RestRecordVerifier) + and "/api/patient" in records_path + ): + # OpenEMR serves FHIR beside its REST API: .../apis/default/fhir. + site = records_path.split("?", 1)[0].split("/api/patient", 1)[0] + bearer = str( + (getattr(verifier, "headers", None) or {}).get("Authorization", "") + ) + token = bearer.removeprefix("Bearer ").strip() + return FhirPatientSource( + f"{verifier.base_url}{site}/fhir", + token=(lambda: token) if token else None, + session=verifier._get_session(), + timeout_s=max(float(getattr(verifier, "timeout_s", 10.0)), 10.0), + ) + if channel == "fhir": + return None + if ( + allow_rest + and isinstance(verifier, RestRecordVerifier) + and "/api/patient" in records_path + ): + return RestPatientSource( + verifier.base_url, + path=records_path, + headers=getattr(verifier, "headers", None), + session=verifier._get_session(), + timeout_s=max(float(getattr(verifier, "timeout_s", 10.0)), 10.0), + ) + except Exception: # noqa: BLE001 - an unusable verifier gives no source + return None + return None + + +#: ``Replayer(duplicate_record_source=...)``: derive the source from the run's +#: own effect verifier (the default). Only tests and embedders pass a source. +AUTO = "auto" + + +def resolve_duplicate_source( + effect_verifier: Any, + requested: Any = AUTO, + *, + pinned_record: bool = False, + channel: Optional[Literal["fhir", "rest"]] = None, +) -> Any: + """The one place a run's duplicate-check source is decided. + + Every ``Replayer`` resolves its source here from the effect verifier it is + given, so qualification, a run, a resume and a proof that read the same + system of record use the same check. With no pin, only a FHIR channel is + used; a program pinned to the record check (``candidate_source: + record``) also accepts OpenEMR's REST channel. ``requested`` is ``AUTO`` + or a source object (tests, embedding). + """ + + if requested is None: + return None + if requested != AUTO: + return requested + if effect_verifier is None: + return None + return source_for_verifier( + effect_verifier, allow_rest=pinned_record, channel=channel + ) diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index a841389f8..c3f54c966 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -868,6 +868,7 @@ def __init__( state_verifier: Optional[Any] = None, effect_verifier: Optional[EffectVerifier] = None, effect_compensator: Optional[Any] = None, + duplicate_record_source: Any = "auto", api_actuator: Optional[Any] = None, web_form_actuator: Optional[Any] = None, api_steps: bool = True, @@ -1020,6 +1021,14 @@ def __init__( # makes a model call — the $0 runtime guarantee is preserved. self.effect_verifier = effect_verifier self.effect_compensator = effect_compensator + # The duplicate check's record channel (``runtime/checks/ + # duplicate_record.py``). Resolved from this run's own effect verifier + # when the check runs (``_duplicate_source``), so every construction + # path (qualification, run, resume, proof) that reads the same system + # of record uses the same check. Only tests and embedders pass one. + self._duplicate_requested = duplicate_record_source + self._last_duplicate_check: Optional[Any] = None + self._last_duplicate_outcome: Optional[str] = None # ``run``-only, bundle-bound admission capability. The ordinary # ``replay`` path leaves this unset and preserves its permissive # behavior. When present it carries exact identity/effect decisions @@ -7770,13 +7779,15 @@ def _run_switch_surface_step( ok, frame = self._poll_postconditions(step, frame, bundle_dir, start_state) if not ok: frame = self.vision.wait_settled(self.backend) - failed = [ - self._describe_postcondition(pc) + failed_pcs = [ + pc for pc in step.expect if not self._postcondition_passes( pc, frame, bundle_dir, start_state ) ] + self._stamp_duplicate_check(result, failed_pcs) + failed = [self._describe_postcondition(pc) for pc in failed_pcs] result.after_png = self._save_step_png( run_dir, evidence_step_id, "after", frame ) @@ -17573,6 +17584,7 @@ def _check_postconditions( and not self._postcondition_confirms_a_write(step, workflow) ): failed_pcs = self._drift_oracle_rescue(failed_pcs, frame_png, result) + self._stamp_duplicate_check(result, failed_pcs) failed = [self._describe_postcondition(pc) for pc in failed_pcs] return not failed_pcs, frame_png, failed @@ -17715,16 +17727,32 @@ def _malformed_date_params(workflow: Workflow, params: dict[str, Any]) -> list[s return problems def _no_duplicate_candidate(self, pc: Any, frame_png: bytes) -> bool: - """The app's possible-duplicate list has loaded and names no candidate. + """The duplicate check passes: no chart is, or may be, this patient. - A row is a candidate when its words include every word of this run's + Records why it failed in ``self._last_duplicate_outcome`` (``exact``, + ``close``, ``candidate`` or ``unreadable``; None when it passed), which + the halted step's ``StepResult.duplicate_check`` carries. + """ + + outcome = self._duplicate_outcome(pc, frame_png) + self._last_duplicate_outcome = outcome + return outcome is None + + def _duplicate_outcome(self, pc: Any, frame_png: bytes) -> Optional[str]: + """None when the check passes, else why it failed. + + With a record source (``runtime/checks/duplicate_record.py``): the + decision, ``exact``, ``close`` or ``unreadable``. On the screen: the + app's possible-duplicate list has loaded and names no candidate. A + row is a candidate when its words include every word of this run's value of any ``pc.candidate_params`` parameter (case-insensitive). A value that is an ISO date is also compared as a date: every date the row shows, in any format intake reads (``Apr 20, 1944``, ``20.04.1944``, ``04/20/1944``), is normalized and compared, and a row date that reads either way counts when either reading matches. - Fails closed: an unreadable list, a missing or empty parameter, or an - empty list without its "no matches" marker all fail the check. + A listed candidate is ``candidate``. Fails closed as ``unreadable``: + an unreadable list, a missing or empty parameter, or an empty list + without its "no matches" marker. """ params = ( @@ -17732,15 +17760,44 @@ def _no_duplicate_candidate(self, pc: Any, frame_png: bytes) -> bool: or getattr(self, "_run_params", None) or {} ) + self._last_duplicate_check = None + pinned = getattr(pc, "candidate_source", None) + source = ( + None + if pinned == "screen" + else self._duplicate_source( + pinned == "record", getattr(pc, "candidate_channel", None) + ) + ) + if pinned == "record" and source is None: + # The program was qualified with the record check; without a + # record channel it must not fall back to the screen. + from openadapt_flow.runtime.checks.duplicate_record import ( + DuplicateCheckResult, + ) + + self._last_duplicate_check = DuplicateCheckResult( + "unreadable", + channel="none", + reason="the program requires the record check and this run has " + "no record channel", + ) + return "unreadable" + if source is not None: + if self._no_duplicate_record(source, params, pc): + return None + checked = self._last_duplicate_check + decision = getattr(checked, "decision", "unreadable") + return decision if decision in ("exact", "close") else "unreadable" names = tuple(pc.candidate_params or ()) if not pc.candidate_rows or not names or not pc.text: - return False + return "unreadable" reader = getattr(self.backend, "element_texts", None) if not callable(reader): - return False + return "unreadable" rows = reader(pc.candidate_rows, tuple(pc.candidate_frame_path or ())) if rows is None: - return False + return "unreadable" def words(value: str) -> list[str]: return re.findall(r"[0-9a-z]+", value.casefold()) @@ -17751,7 +17808,7 @@ def words(value: str) -> list[str]: text = runtime_param_text(params[name]) if name in params else "" value = words(text) if not value: - return False + return "unreadable" wanted.append(value) if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text.strip()): wanted_dates.add(text.strip()) @@ -17759,15 +17816,84 @@ def words(value: str) -> list[str]: # Nothing listed: only the list's own "no matches" marker shows # the search finished. if self._dom_text_visible(pc.text): - return True - return self.vision.text_present(frame_png, pc.text) + return None + if self.vision.text_present(frame_png, pc.text): + return None + return "unreadable" # the list never loaded for row in rows: row_words = set(words(row)) if any(all(w in row_words for w in value) for value in wanted): - return False + return "candidate" if wanted_dates and wanted_dates & _row_dates(row): - return False - return True + return "candidate" + return None + + def _stamp_duplicate_check(self, result: Any, failed_pcs: Any) -> None: + """Put the duplicate check's failure on the halted step's result.""" + + if result is None or not any( + getattr(pc.kind, "value", pc.kind) == "no_duplicate_candidate" + for pc in failed_pcs + ): + return + result.duplicate_check = ( + getattr(self, "_last_duplicate_outcome", None) or "unreadable" + ) + + def _duplicate_source( + self, pinned_record: bool = False, channel: Any = None + ) -> Any: + """This run's duplicate-check record source, or None (the screen).""" + + from openadapt_flow.runtime.checks.duplicate_record import ( + resolve_duplicate_source, + ) + + return resolve_duplicate_source( + self.effect_verifier, + getattr(self, "_duplicate_requested", "auto"), + pinned_record=pinned_record, + channel=channel, + ) + + def _no_duplicate_record( + self, source: Any, params: dict[str, Any], pc: Any = None + ) -> bool: + """The duplicate check with the system of record as its source. + + Proceeds only when the record channel answers completely and no chart + is this patient (``decision == "none"``). An exact or close match, or + an unreadable channel, fails the check, and the halt names the + records found and why (``_describe_postcondition``). + """ + + from openadapt_flow.runtime.checks.duplicate_record import ( + RECORD_FIELDS, + check_duplicate, + ) + + # Which run parameter fills each field: OpenEMR's names by default, + # ``pc.candidate_record_fields`` for any other program. + mapped = getattr(pc, "candidate_record_fields", None) or {} + names = {**RECORD_FIELDS, **dict(mapped)} + want = { + field: runtime_param_text(params[name]) if name in params else "" + for field, name in names.items() + } + try: + result = check_duplicate(source, want) + except Exception as exc: # noqa: BLE001 - any failure fails closed + from openadapt_flow.runtime.checks.duplicate_record import ( + DuplicateCheckResult, + ) + + result = DuplicateCheckResult( + "unreadable", + channel=getattr(source, "channel", ""), + reason=f"the record check failed ({type(exc).__name__})", + ) + self._last_duplicate_check = result + return result.proceed def _field_value_shown(self, pc: Any) -> bool: """The focused field shows this run's value of ``pc.param``. @@ -17800,10 +17926,21 @@ def squash(value: str) -> str: value = read("focused_text_value") return value is not None and value == expected - @staticmethod - def _describe_postcondition(pc: Any) -> str: + def _describe_postcondition(self, pc: Any) -> str: """Human-readable one-liner for a postcondition (for error messages).""" kind = pc.kind.value if hasattr(pc.kind, "value") else pc.kind + checked = getattr(self, "_last_duplicate_check", None) + if kind == "no_duplicate_candidate" and checked is not None: + found = "; ".join( + f"{c.match} record {c.record.record_id} ({c.why})" + for c in checked.candidates + ) + return ( + f"no_duplicate_candidate (record check via {checked.channel}: " + f"{checked.decision}, {checked.reason}" + + (f": {found}" if found else "") + + ")" + ) if kind == "local_edit": return "local_edit (URL, title, page count and focus unchanged)" if kind == "field_value": diff --git a/packages/flow/public-artifacts.json b/packages/flow/public-artifacts.json index c8f2dafa6..fb115f7be 100644 --- a/packages/flow/public-artifacts.json +++ b/packages/flow/public-artifacts.json @@ -166,6 +166,29 @@ "benchmark/lending_fault_model/resolution_ladder_results.json", "benchmark/lending_fault_model/results.json", "benchmark/lending_fault_model/swer_results.json", + "benchmark/lifecycle/results/production-2026-10-24/after/arm-ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/after/ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/after/results.json", + "benchmark/lifecycle/results/production-2026-10-24/after/stops.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/after/switch/log-1792873972.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/after/switch/log-1792878330.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/after/switch/switch.json", + "benchmark/lifecycle/results/production-2026-10-24/after/workload.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/before/arm-ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/before/ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/before/results.json", + "benchmark/lifecycle/results/production-2026-10-24/before/stops.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/before/switch/log-1792868032.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/before/switch/switch.json", + "benchmark/lifecycle/results/production-2026-10-24/before/workload.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/arm-ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/ledger.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/results.json", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/stops.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/switch/log-1792865787.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/switch/log-1792866771.jsonl", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/switch/switch.json", + "benchmark/lifecycle/results/production-2026-10-24/recovery-smoke/workload.jsonl", "benchmark/lifecycle_proof/results/earlier-attempts/proof-12-ledger.jsonl", "benchmark/lifecycle_proof/results/earlier-attempts/proof-13-ledger.jsonl", "benchmark/lifecycle_proof/results/earlier-attempts/proof-14-ledger.jsonl", diff --git a/packages/flow/tests/test_duplicate_record_check.py b/packages/flow/tests/test_duplicate_record_check.py new file mode 100644 index 000000000..17bd07818 --- /dev/null +++ b/packages/flow/tests/test_duplicate_record_check.py @@ -0,0 +1,512 @@ +"""The duplicate check with the system of record as its source. + +The screen check stops on any listed chart that shares the last name or the +date of birth. This check asks the record channel for the patient and stops +only on a chart that is, or may be, this patient, and on any channel failure. +The FHIR tests run against a real local HTTP server serving OpenEMR-shaped +``Patient`` bundles, page by page. +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any, Callable, Optional +from urllib.parse import parse_qs, urlsplit + +import pytest + +from openadapt_flow.runtime.checks.duplicate_record import ( + FhirPatientSource, + PatientRecord, + RestPatientSource, + SqlPatientSource, + TableSource, + check_duplicate, + classify, +) + +WANT = {"fname": "Rosalind", "lname": "Jesswood", "DOB": "1990-04-09"} + + +def rec( + fname: str, lname: str, dob: str, rid: str = "1", mrn: str = "" +) -> PatientRecord: + return PatientRecord(rid, fname, lname, dob, mrn) + + +@pytest.mark.parametrize( + ("record", "match"), + [ + (rec("Rosalind", "Jesswood", "1990-04-09"), "exact"), + (rec(" ROSALIND ", "jesswood", "1990-04-09"), "exact"), + (rec("Ana", "Other", "1950-01-01", mrn="P77"), "exact"), # same MRN + (rec("Rosalind", "Okafor", "1990-04-09"), "close"), # married name + (rec("Rosalind", "Jesswood-Okafor", "1990-04-09"), "close"), # hyphenated + (rec("Rosalind", "Jesswood", "1990-09-04"), "close"), # day and month swapped + (rec("Rosalind", "Jesswood", "1990-04-10"), "close"), # a typo in the day + (rec("Rosalnid", "Jesswood", "1990-04-09"), "close"), # a typo in the name + (rec("Rosalind", "Jesswood", ""), "close"), # the record has no birth date + (rec("Ana", "Jesswood", "1961-01-01"), None), # a namesake + (rec("Ana", "Oakwood", "1990-04-09"), None), # a shared birth date + (rec("Rosalind", "Jesswood", "1971-11-23"), None), # same name, another person + ], +) +def test_classify(record: PatientRecord, match: Optional[str]) -> None: + found = classify({**WANT, "mrn": "P77"}, record) + assert (found.match if found else None) == match + + +def test_decisions_and_fail_closed_inputs() -> None: + table = TableSource( + [ + rec("Ana", "Jesswood", "1961-01-01", "1"), + rec("Ana", "Oakwood", "1990-04-09", "2"), + ] + ) + assert check_duplicate(table, WANT).decision == "none" + table.records.append(rec("Rosalind", "Jesswood", "1990-04-09", "3")) + result = check_duplicate(table, WANT) + assert result.decision == "exact" and result.source == "record" + assert [c.record.record_id for c in result.candidates] == ["3"] + # Not an ISO date (intake converts first), or a missing name: stop. + assert ( + check_duplicate(table, {**WANT, "DOB": "04/09/1990"}).decision == "unreadable" + ) + assert check_duplicate(table, {**WANT, "lname": " "}).decision == "unreadable" + + +# -- FHIR over HTTP --------------------------------------------------------------- + + +def patient(pid: str, given: str, family: str, dob: str) -> dict[str, Any]: + """Shaped like OpenEMR 8's FHIR Patient (official name, PT identifier).""" + + return { + "resourceType": "Patient", + "id": pid, + "name": [{"use": "official", "family": family, "given": [given]}], + "birthDate": dob, + "identifier": [{"type": {"coding": [{"code": "PT"}]}, "value": f"P{pid}"}], + } + + +class Server: + def __init__( + self, handle: Callable[[str, dict[str, list[str]]], tuple[int, Any]] + ) -> None: + outer = self + self.requests: list[str] = [] + + class H(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 + parts = urlsplit(self.path) + outer.requests.append(self.path) + status, body = handle(parts.path, parse_qs(parts.query)) + data = json.dumps(body).encode() + self.send_response(status) + self.send_header("Content-Type", "application/fhir+json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *args: Any) -> None: + pass + + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), H) + self.url = f"http://127.0.0.1:{self.httpd.server_address[1]}/fhir" + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + + def close(self) -> None: + self.httpd.shutdown() + + +@pytest.fixture +def serve(): + servers: list[Server] = [] + + def make(handle: Callable[[str, dict[str, list[str]]], tuple[int, Any]]) -> Server: + servers.append(Server(handle)) + return servers[-1] + + yield make + for s in servers: + s.close() + + +PEOPLE = [patient(str(i), "Ana", f"Name{i}", "1990-04-09") for i in range(7)] + + +def paged(page_size: int, people: list[dict[str, Any]], total: Optional[int] = None): + def handle(path: str, query: dict[str, list[str]]) -> tuple[int, Any]: + page = int(query.get("_page", ["0"])[0]) + if "birthdate" in query: + match = [p for p in people if p["birthDate"] == query["birthdate"][0]] + elif "family" in query: + f = query["family"][0].lower() + match = [p for p in people if p["name"][0]["family"].lower().startswith(f)] + else: + match = [] + chunk = match[page * page_size : (page + 1) * page_size] + links = [{"relation": "self", "url": "Patient"}] + if (page + 1) * page_size < len(match): + links.append( + { + "relation": "next", + "url": f"Patient?_page={page + 1}&" + + "&".join(f"{k}={v[0]}" for k, v in query.items() if k != "_page"), + } + ) + return 200, { + "resourceType": "Bundle", + "type": "searchset", + "total": len(match) if total is None else total, + "link": links, + "entry": [{"resource": r} for r in chunk], + } + + return handle + + +def test_fhir_reads_every_page_and_finds_the_chart_on_the_last(serve) -> None: + people = PEOPLE + [patient("99", "Rosalind", "Jesswood", "1990-04-09")] + server = serve(paged(3, people)) + result = check_duplicate(FhirPatientSource(server.url), WANT) + assert result.decision == "exact" + assert result.candidates[0].record.mrn == "P99" + assert result.pages >= 3 + + +def test_fhir_no_match_proceeds(serve) -> None: + server = serve(paged(3, PEOPLE)) + result = check_duplicate(FhirPatientSource(server.url), WANT) + assert result.decision == "none", result.reason + + +@pytest.mark.parametrize( + "handle", + [ + paged(3, PEOPLE, total=9), # fewer read than the server reported + lambda p, q: (401, {"error": "invalid_token"}), + lambda p, q: (500, {}), + lambda p, q: (200, {"resourceType": "OperationOutcome"}), + lambda p, q: ( + 200, + { + "resourceType": "Bundle", + "entry": [{"resource": {"resourceType": "Observation"}}], + }, + ), + # A next link that never ends. + lambda p, q: ( + 200, + { + "resourceType": "Bundle", + "entry": [], + "link": [{"relation": "next", "url": "Patient?again=1"}], + }, + ), + ], +) +def test_fhir_fails_closed(serve, handle) -> None: + server = serve(handle) + assert check_duplicate(FhirPatientSource(server.url), WANT).decision == "unreadable" + + +def test_fhir_unreachable_fails_closed() -> None: + source = FhirPatientSource("http://127.0.0.1:9/fhir", timeout_s=0.5) + assert check_duplicate(source, WANT).decision == "unreadable" + + +def test_rest_full_page_without_next_link_fails_closed(serve) -> None: + rows = [ + {"uuid": str(i), "fname": "Ana", "lname": f"N{i}", "DOB": "1990-04-09"} + for i in range(5) + ] + server = serve(lambda p, q: (200, {"data": rows})) + source = RestPatientSource(server.url, path="/api/patient", page_size=5) + assert check_duplicate(source, WANT).decision == "unreadable" + source = RestPatientSource(server.url, path="/api/patient", page_size=50) + assert check_duplicate(source, WANT).decision == "none" + + +def test_sql_fallback(tmp_path: Path) -> None: + db = tmp_path / "openemr.db" + con = sqlite3.connect(db) + con.execute( + # OpenEMR's patient_data columns the fallback reads. + "CREATE TABLE patient_data (pid INTEGER, fname TEXT, lname TEXT, DOB TEXT, " + "pubpid TEXT, phone_home TEXT, street TEXT, email TEXT, birth_fname TEXT, " + "birth_lname TEXT)" + ) + con.execute( + "INSERT INTO patient_data VALUES (4, 'Rosalind', 'Jesswood-Okafor', " + "'1990-04-09', '4', '', '', '', NULL, NULL)" + ) + # A chart stored under a married name, with the birth name kept. + con.execute( + "INSERT INTO patient_data VALUES (5, 'Rosalind', 'Okafor', '1990-04-10', " + "'5', '', '', '', 'Rosalind', 'Jesswood')" + ) + con.commit() + con.close() + source = SqlPatientSource(lambda: sqlite3.connect(db), paramstyle="qmark") + result = check_duplicate(source, WANT) + assert result.decision == "close" and result.channel == "sql" + assert {c.record.record_id for c in result.candidates} == {"4", "5"} + + +# -- in the replayer -------------------------------------------------------------- + + +def _replayer_check(source: Any, params: dict[str, str]) -> tuple[bool, str]: + from openadapt_flow.ir import Postcondition, PostconditionKind + from openadapt_flow.runtime.replayer import Replayer + from tests.test_no_duplicate_candidate import MARKER, _Backend + from tests.test_replayer import FakeVision + + # The screen would stop: its list shows a namesake. + backend = _Backend(["5\tJESSWOOD, Ana\t1961-01-01\tFemale"]) + replayer = Replayer(backend, vision=FakeVision(), duplicate_record_source=source) + replayer._current_params = dict(params) + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text=MARKER, + candidate_rows="tr.oneresult", + candidate_frame_path=("#modalframe",), + candidate_params=("lname", "DOB"), + ) + passed = replayer._postcondition_passes(pc, b"", None, {}) + assert backend.read == [] # the screen list is not read + return passed, replayer._describe_postcondition(pc) + + +def test_the_replayer_asks_the_record_when_a_channel_is_configured() -> None: + table = TableSource([rec("Ana", "Jesswood", "1961-01-01", "5")]) + assert _replayer_check(table, WANT)[0] is True + table.records.append(rec("Rosalind", "Okafor", "1990-04-09", "8")) + passed, why = _replayer_check(table, WANT) + assert passed is False + # #1517 classes "failed: no_duplicate_candidate" as an item-input stop. + assert why.startswith("no_duplicate_candidate ") + assert "record check via table: close" in why + assert "close record 8 (same date of birth and first name" in why + + +def test_the_replayer_fails_closed_when_the_channel_raises() -> None: + class Broken: + channel = "fhir" + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + raise RuntimeError("boom") + + passed, why = _replayer_check(Broken(), WANT) + assert passed is False and "unreadable" in why + + +# -- round two: nicknames, swapped names, prior names, contact tie-breakers ------- + +FULL = { + **WANT, + "phone_home": "617-555-0142", + "email": "r.j@example.test", + "street": "41 Larch Way", +} + + +def full( + fname: str, + lname: str, + dob: str, + *, + phone: str = "", + email: str = "", + street: str = "", + prior: tuple = (), +) -> PatientRecord: + return PatientRecord("9", fname, lname, dob, "", phone, street, email, prior) + + +@pytest.mark.parametrize( + ("record", "match"), + [ + # A married name and a mistyped day, tied by the phone. + (full("Rosalind", "Okafor", "1990-04-10", phone="(617) 555-0142"), "close"), + # The same, with nothing to tie them: not a candidate. + (full("Rosalind", "Okafor", "1990-04-10"), None), + # A nickname. + (full("Roz", "Jesswood", "1990-04-09"), None), # not in the table + (full("Rosalind", "Jesswood", "1990-04-09"), "exact"), + # First and last name swapped. + (full("Jesswood", "Rosalind", "1990-04-09"), "close"), + # A prior (maiden) name on the chart, with a mistyped day. + ( + full("Rosalind", "Okafor", "1990-04-10", prior=(("Rosalind", "Jesswood"),)), + "close", + ), + # A full name change: the same birth date and the same email. + (full("Mary", "Okafor", "1990-04-09", email="R.J@example.test"), "close"), + # A full name change with a near birth date needs two contact details. + (full("Mary", "Okafor", "1990-04-10", email="r.j@example.test"), None), + ( + full( + "Mary", + "Okafor", + "1990-04-10", + email="r.j@example.test", + street="41 LARCH WAY", + ), + "close", + ), + ], +) +def test_round_two(record: PatientRecord, match: Optional[str]) -> None: + found = classify(FULL, record) + assert (found.match if found else None) == match, found + + +def test_nickname_table() -> None: + want = {"fname": "Bob", "lname": "Ashby", "DOB": "1950-02-03"} + assert classify(want, rec("Robert", "Ashby", "1950-02-03")).match == "close" + assert classify(want, rec("Roberta", "Ashby", "1971-02-03")) is None + + +def test_fhir_reads_old_names_and_contacts(serve) -> None: + resource = patient("12", "Rosalind", "Okafor", "1990-04-10") + resource["name"].append( + {"use": "maiden", "family": "Jesswood", "given": ["Rosalind"]} + ) + resource["telecom"] = [{"system": "phone", "use": "home", "value": "617-555-0142"}] + server = serve( + lambda p, q: ( + 200, + {"resourceType": "Bundle", "total": 1, "entry": [{"resource": resource}]}, + ) + ) + result = check_duplicate(FhirPatientSource(server.url), FULL) + assert result.decision == "close" + record = result.candidates[0].record + assert record.prior_names == (("Rosalind", "Jesswood"),) and record.phone + + +@pytest.mark.parametrize( + ("want", "record"), + [ + # An initial for the first name. + ( + {"fname": "J.", "lname": "Ashby", "DOB": "1950-02-03"}, + rec("James", "Ashby", "1950-02-03"), + ), + # A second surname on the chart, the first one on the referral. + ( + {"fname": "Lucia", "lname": "Garcia", "DOB": "1981-06-15"}, + rec("Lucia", "Garcia Lopez", "1981-06-15"), + ), + ], +) +def test_production_lifecycle_cases_stop( + want: dict[str, str], record: PatientRecord +) -> None: + found = classify(want, record) + assert found is not None and found.match == "close" + + +def test_an_initial_does_not_match_another_letter() -> None: + want = {"fname": "J.", "lname": "Ashby", "DOB": "1950-02-03"} + assert classify(want, rec("Mary", "Ashby", "1950-02-03")) is None + + +# -- review fixes: every read problem is unreadable, never "none" ---------------- + + +@pytest.mark.parametrize( + "body", + [ + {"internalErrors": ["db down"]}, # no data, an error + {"data": [], "validationErrors": {"DOB": "bad"}}, # an error next to data + {}, # no data at all + {"data": {"0": {}}}, # data that isn't a list + ], +) +def test_rest_error_bodies_are_unreadable(serve, body) -> None: + server = serve(lambda p, q: (200, body)) + source = RestPatientSource(server.url, path="/api/patient") + assert check_duplicate(source, WANT).decision == "unreadable" + + +@pytest.mark.parametrize("partial", ["1990", "1990-04", ""]) +def test_a_partial_record_birth_date_is_no_date_not_a_mismatch(serve, partial) -> None: + resource = patient("31", "Rosalind", "Jesswood", partial) + if not partial: + resource.pop("birthDate") + bundle = {"resourceType": "Bundle", "total": 1, "entry": [{"resource": resource}]} + server = serve(lambda p, q: (200, bundle)) + assert check_duplicate(FhirPatientSource(server.url), WANT).decision == "close" + + +def test_a_record_with_no_id_is_unreadable() -> None: + table = TableSource([rec("Ana", "Other", "1990-04-09", rid=" ")]) + assert check_duplicate(table, WANT).decision == "unreadable" + + +def test_a_next_link_to_another_host_is_unreadable(serve) -> None: + elsewhere = "https://elsewhere.example.test/Patient?p=2" + bundle = { + "resourceType": "Bundle", + "entry": [], + "link": [{"relation": "next", "url": elsewhere}], + } + server = serve(lambda p, q: (200, bundle)) + result = check_duplicate(FhirPatientSource(server.url), WANT) + assert result.decision == "unreadable" and "another host" in result.reason + + +@pytest.mark.parametrize( + ("want", "record"), + [ + # A nickname and a one-field date typo. + ( + {"fname": "Bob", "lname": "Ashby", "DOB": "1950-05-07"}, + rec("Robert", "Ashby", "1950-05-06"), + ), + # A last-name typo and a one-field date typo. + ( + {"fname": "Rosalind", "lname": "Jeswood", "DOB": "1990-04-10"}, + rec("Rosalind", "Jesswood", "1990-04-09"), + ), + ], +) +def test_review_recall_gaps_stop(want, record) -> None: + found = classify(want, record) + assert found is not None and found.match == "close" + + +def test_a_program_maps_its_own_parameter_names() -> None: + from openadapt_flow.ir import Postcondition, PostconditionKind + from openadapt_flow.runtime.replayer import Replayer + from tests.test_no_duplicate_candidate import MARKER + from tests.test_no_duplicate_candidate import _Backend as ListBackend + from tests.test_replayer import FakeVision + + table = TableSource([rec("Rosalind", "Jesswood", "1990-04-09", "7")]) + replayer = Replayer( + ListBackend([]), vision=FakeVision(), duplicate_record_source=table + ) + replayer._current_params = { + "given": "Rosalind", + "family": "Jesswood", + "birth": "1990-04-09", + } + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text=MARKER, + candidate_rows="tr", + candidate_params=("family",), + candidate_source="record", + candidate_record_fields={"fname": "given", "lname": "family", "DOB": "birth"}, + ) + assert replayer._postcondition_passes(pc, b"", None, {}) is False + assert replayer._last_duplicate_check.decision == "exact" diff --git a/packages/flow/tests/test_duplicate_source_paths.py b/packages/flow/tests/test_duplicate_source_paths.py new file mode 100644 index 000000000..e18dd8b71 --- /dev/null +++ b/packages/flow/tests/test_duplicate_source_paths.py @@ -0,0 +1,349 @@ +"""Every governed way of building a Replayer resolves one duplicate-check source. + +Qualification, a run, a resume and the lifecycle proof must use the same +duplicate check, or a program is admitted under one check and run under +another. The source is resolved by ``Replayer`` from the run's own effect +verifier and the program's pin (``resolve_duplicate_source``). These tests +pin that: + +- no construction path picks a source of its own (an AST scan of every + ``Replayer(...)`` call); +- the governed paths all hand the Replayer their effect verifier; +- the same verifier gives the same source on every path; +- OpenEMR's REST adapter is used only by a program pinned to the record; +- resolution never crashes on a verifier it doesn't know; +- a program pinned to the record check fails closed without a channel. +""" + +from __future__ import annotations + +import ast +from pathlib import Path +from typing import Any + +import pytest + +from openadapt_flow.ir import Postcondition, PostconditionKind +from openadapt_flow.runtime.checks.duplicate_record import ( + FhirPatientSource, + PatientRecord, + RestPatientSource, + TableSource, + resolve_duplicate_source, +) +from openadapt_flow.runtime.effects import FhirEffectVerifier +from openadapt_flow.runtime.effects.rest import RestRecordVerifier + +FLOW = Path(__file__).resolve().parents[1] +#: Governed construction paths: (file, enclosing function) of a Replayer(...) +#: call that runs a program against the system of record. +GOVERNED = { + ("openadapt_flow/qualify/campaign.py", "_run"), + ("benchmark/lifecycle_proof/run.py", "run"), + ("benchmark/lifecycle_proof/run.py", "execute"), + ("benchmark/lifecycle_proof/run.py", "_resume"), + ("openadapt_flow/deployment.py", "build_replayer"), +} + + +def _replayer_calls() -> list[tuple[str, str, ast.Call]]: + out = [] + for root in ("openadapt_flow", "benchmark"): + for path in (FLOW / root).rglob("*.py"): + try: + tree = ast.parse(path.read_text(encoding="utf-8")) + except (SyntaxError, UnicodeDecodeError): + continue + rel = str(path.relative_to(FLOW)) + for fn in ast.walk(tree): + if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for node in ast.walk(fn): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "Replayer" + ): + out.append((rel, fn.name, node)) + return out + + +def test_no_construction_path_picks_its_own_duplicate_source() -> None: + calls = _replayer_calls() + assert calls, "the scan found no Replayer constructions" + choosing = { + (rel, fn) + for rel, fn, call in calls + if any(k.arg == "duplicate_record_source" for k in call.keywords) + } + assert choosing == set() + + +def test_every_governed_path_hands_the_replayer_its_effect_verifier() -> None: + calls = _replayer_calls() + found = {(rel, fn) for rel, fn, _ in calls} + missing = GOVERNED - found + assert not missing, f"governed paths moved or renamed: {sorted(missing)}" + for rel, fn, call in calls: + if (rel, fn) not in GOVERNED: + continue + names = {k.arg for k in call.keywords} + # ``None`` is ``**kwargs`` (qualify), checked by name below. + assert "effect_verifier" in names or None in names, (rel, fn) + campaign = (FLOW / "openadapt_flow/qualify/campaign.py").read_text() + assert "effect_verifier=verifier" in campaign + # A qualification hook can't swap the duplicate check. + assert "replayer_kwargs may not set duplicate_record_source" in campaign + + +class _Backend: + url, page_title, page_count = "http://x/", "x", 1 + + +def _fhir() -> FhirEffectVerifier: + return FhirEffectVerifier( + "https://emr.example.test/apis/default/fhir", access_token="t" + ) + + +def _rest() -> RestRecordVerifier: + return RestRecordVerifier( + "https://emr.example.test", + records_path="/apis/default/api/patient?_limit=1000", + records_key="data", + headers={"Authorization": "Bearer t"}, + ) + + +@pytest.mark.parametrize( + ("verifier", "pinned", "kind"), + [ + (_fhir, False, FhirPatientSource), + (_fhir, True, FhirPatientSource), + # OpenEMR's REST adapter only for a program pinned to the record. + (_rest, False, type(None)), + (_rest, True, RestPatientSource), + ], +) +def test_the_same_verifier_gives_the_same_source_on_every_path( + verifier: Any, pinned: bool, kind: type +) -> None: + from openadapt_flow.deployment import build_replayer + from openadapt_flow.runtime.replayer import Replayer + from tests.test_replayer import FakeVision + + v = verifier() + direct = Replayer(_Backend(), vision=FakeVision(), effect_verifier=v) + deployed = build_replayer( + _Backend(), + allow_egress=False, + effect_verifier=v, + api_actuator=None, + durable=False, + use_structural=False, + ) + for replayer in (direct, deployed): + source = replayer._duplicate_source(pinned) + assert isinstance(source, kind) + if source is not None: + assert source.base_url == v.base_url + + +def test_resolution_is_tolerant_and_rest_is_opt_in() -> None: + assert resolve_duplicate_source(None) is None + table = TableSource([]) + assert resolve_duplicate_source(None, table) is table + assert resolve_duplicate_source(_rest()) is None + assert isinstance( + resolve_duplicate_source(_rest(), pinned_record=True), RestPatientSource + ) + + class Double(RestRecordVerifier): + """A test double that skips RestRecordVerifier.__init__.""" + + def __init__(self) -> None: + pass + + # No crash at construction; no source, so a pinned program fails closed. + assert resolve_duplicate_source(Double(), pinned_record=True) is None + + class Wrapper: + def __init__(self, inner: Any) -> None: + self.inner = inner + + assert isinstance(resolve_duplicate_source(Wrapper(_fhir())), FhirPatientSource) + + +def test_a_replayer_builds_with_an_unknown_verifier_double() -> None: + from openadapt_flow.runtime.replayer import Replayer + from tests.test_replayer import FakeVision + + class Double(RestRecordVerifier): + def __init__(self) -> None: + pass + + replayer = Replayer(_Backend(), vision=FakeVision(), effect_verifier=Double()) + assert replayer._duplicate_source(True) is None + + +def _check(source: Any, pinned: Any) -> tuple[bool, list[Any]]: + from openadapt_flow.runtime.replayer import Replayer + from tests.test_no_duplicate_candidate import MARKER + from tests.test_no_duplicate_candidate import _Backend as ListBackend + from tests.test_replayer import FakeVision + + backend = ListBackend([], marker=True) # the screen would pass + replayer = Replayer(backend, vision=FakeVision(), duplicate_record_source=source) + replayer._current_params = {"fname": "Ada", "lname": "Quill", "DOB": "1984-03-07"} + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text=MARKER, + candidate_rows="tr.oneresult", + candidate_frame_path=("#modalframe",), + candidate_params=("lname", "DOB"), + candidate_source=pinned, + ) + return replayer._postcondition_passes(pc, b"", None, {}), backend.read + + +def test_a_program_pinned_to_the_record_fails_closed_without_a_channel() -> None: + from openadapt_flow.runtime.replayer import Replayer + from tests.test_replayer import FakeVision + + passed, read = _check(None, "record") + assert passed is False and read == [] # never the screen instead + replayer = Replayer(_Backend(), vision=FakeVision()) + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text="x", + candidate_rows="tr", + candidate_params=("lname",), + candidate_source="record", + ) + replayer._current_params = {} + replayer._postcondition_passes(pc, b"", None, {}) + why = replayer._describe_postcondition(pc) + # The reason the switch policy reads (#1517): a structured, stable phrase. + assert why.startswith("no_duplicate_candidate (record check via none: unreadable") + + +def test_a_program_pinned_to_the_screen_reads_the_screen() -> None: + table = TableSource([PatientRecord("1", "Ada", "Quill", "1984-03-07")]) + passed, read = _check(table, "screen") + assert passed is True and read # the list was read; the record wasn't asked + + +def test_unset_candidate_fields_serialize_as_before() -> None: + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text="x", + candidate_rows="tr", + candidate_params=("lname",), + ) + dumped = pc.model_dump() + assert "candidate_source" not in dumped + assert "candidate_record_fields" not in dumped + + +# -- the structured outcome on the halted step (#1517 reads it) ----------------- + + +def _halted(source: Any, rows: Any, pinned: Any = None) -> Any: + from openadapt_flow.ir import StepResult + from openadapt_flow.runtime.replayer import Replayer + from tests.test_no_duplicate_candidate import MARKER + from tests.test_no_duplicate_candidate import _Backend as ListBackend + from tests.test_replayer import FakeVision + + replayer = Replayer( + ListBackend(rows, marker=False), + vision=FakeVision(), + duplicate_record_source=source, + ) + replayer._current_params = {"fname": "Ada", "lname": "Quill", "DOB": "1984-03-07"} + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text=MARKER, + candidate_rows="tr.oneresult", + candidate_frame_path=("#modalframe",), + candidate_params=("lname", "DOB"), + candidate_source=pinned, + ) + result = StepResult(step_id="s", intent="x", ok=False) + failed = [] if replayer._postcondition_passes(pc, b"", None, {}) else [pc] + replayer._stamp_duplicate_check(result, failed) + return result + + +@pytest.mark.parametrize( + ("source", "rows", "pinned", "outcome"), + [ + ( + TableSource([PatientRecord("1", "Ada", "Quill", "1984-03-07")]), + [], + None, + "exact", + ), + ( + TableSource([PatientRecord("2", "Ada", "Okafor", "1984-03-07")]), + [], + None, + "close", + ), + (None, [], "record", "unreadable"), # pinned, no channel + (None, ["5\tQuill, Ana\t1961-01-01"], "screen", "candidate"), + (None, None, "screen", "unreadable"), # the list can't be read + (None, [], "screen", "unreadable"), # never loaded (no marker) + ], +) +def test_the_halted_step_carries_the_duplicate_outcome( + source, rows, pinned, outcome +) -> None: + result = _halted(source, rows, pinned) + assert result.duplicate_check == outcome + assert result.model_dump()["duplicate_check"] == outcome + + +def test_an_unreadable_record_channel_is_unreadable() -> None: + class Broken: + channel = "fhir" + + def by_birthdate(self, dob: str) -> list[PatientRecord]: + raise RuntimeError("down") + + assert _halted(Broken(), []).duplicate_check == "unreadable" + + +def test_a_passing_or_unrelated_step_serializes_as_before() -> None: + from openadapt_flow.ir import StepResult + + result = _halted(TableSource([]), []) + assert result.duplicate_check is None + assert ( + "duplicate_check" + not in StepResult(step_id="s", intent="x", ok=True).model_dump() + ) + + +def test_a_program_can_choose_fhir_beside_an_openemr_rest_check() -> None: + source = resolve_duplicate_source(_rest(), pinned_record=True, channel="fhir") + assert isinstance(source, FhirPatientSource) + assert source.base_url == "https://emr.example.test/apis/default/fhir" + assert source._headers()["Authorization"] == "Bearer t" + # Asked for REST with a FHIR check, or FHIR with an unrelated REST read: + # no source, so a record-pinned program fails closed. + assert resolve_duplicate_source(_fhir(), pinned_record=True, channel="rest") is None + other = RestRecordVerifier("https://x.test", records_path="/api/db") + assert resolve_duplicate_source(other, pinned_record=True, channel="fhir") is None + + +def test_the_channel_choice_serializes_only_when_set() -> None: + pc = Postcondition( + kind=PostconditionKind.NO_DUPLICATE_CANDIDATE, + text="x", + candidate_rows="tr", + candidate_params=("lname",), + candidate_source="record", + candidate_channel="fhir", + ) + assert pc.model_dump()["candidate_channel"] == "fhir"