From 19b2a35a716f8e35fa69aaaea497cc7b9b789eed Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 24 Oct 2026 23:48:05 -0400 Subject: [PATCH 1/2] effects: read a paged FHIR or REST record list whole, or not at all The FHIR verifier read only the first page of a search, so a record on page two read as absent. It now follows the Bundle's next links on the same origin and checks total. The REST verifier follows the next page's URL at records_next_path when it is set (relative links resolve against the page). A next page on another origin, a failed page, more than 50 pages, or a total above what was read makes the read unreadable, never a confirmation or a refutation from the first pages. Unset, records_next_path drops out of the config so existing digests hold. FHIR/REST pagination also proposed by @yunusarda-sudo in #1511 (superseded by this change). Co-authored-by: yunusarda-sudo Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 1 + llms-full.txt | 2 + packages/flow/docs/EFFECT_KIT.md | 11 +- packages/flow/openadapt_flow/deployment.py | 9 +- .../openadapt_flow/runtime/effects/fhir.py | 71 ++++- .../openadapt_flow/runtime/effects/rest.py | 79 ++++- packages/flow/tests/test_effect_paging.py | 273 ++++++++++++++++++ 7 files changed, 422 insertions(+), 24 deletions(-) create mode 100644 packages/flow/tests/test_effect_paging.py diff --git a/CAPABILITIES.md b/CAPABILITIES.md index e541d413f..dbbf51508 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -210,6 +210,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page or a short read is unreadable | On main | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/llms-full.txt b/llms-full.txt index 02ba1a00d..64bd29116 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -563,6 +563,7 @@ improvising or retrying when something is off. | Learn one program with branches from two or more recordings, or say which recordings are missing | Released | Synthetic | `openadapt-flow induce`, `disambiguate`; [CAPABILITIES.md (flow)](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/CAPABILITIES.md) | | Learn a guarded branch from several halts fixed the same way | Released | Synthetic | `openadapt-flow teach-branch`, `learning/branch_teach.py` | | Run a program over a worklist | Released | Real app | `openadapt-flow for-each` | +| Learn a worklist loop from recordings that search a record, open it, work it and go back to the list, even when the last row skips the return or rows differ (a field some rows lack, a write some rows get behind a screen check a person accepts). Each pass selects the row's own record by its values, verifies it, and checks that row's write. Certified and governed runs refuse it until qualification covers it | On main ([#1435](https://github.com/OpenAdaptAI/OpenAdapt/pull/1435)) | Synthetic (held-out rows run through the interpreter against an in-memory app) | `openadapt-flow induce`; `Step.target_param` | | Typed business decisions inside a run | Released | Synthetic | `business_decision` states, `openadapt-flow business-decisions` | | Watch many real runs in the background and mine candidate branches (shadow mode) | On main ([#1212](https://github.com/OpenAdaptAI/OpenAdapt/pull/1212)) | Synthetic | [SHADOW_MINING.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/capture/docs/SHADOW_MINING.md), [LEARN_FROM_MANY_RUNS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/LEARN_FROM_MANY_RUNS.md) | | Fill a workflow's inputs from a referral PDF, fax image or email | Released | Synthetic | `openadapt-flow intake`, [INTAKE.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/INTAKE.md) | @@ -606,6 +607,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page or a short read is unreadable | On main | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 393d94754..066685dad 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -1035,7 +1035,7 @@ Per-kind required fields: | `ui_readback` | `ui_readback` section: `record_url`, `sign_in` (env-var names for the read-back account), `ready_selector`, `row_selector`, `fields`, `identity` | `url_params`, `empty_selector`, `not_found_selector`, `read_post_paths`, per-field `attribute` / `date_format` / `casefold`, `truncation_markers`, `navigation_timeout_s` | | `desktop_readback` | `desktop_readback` section: `session` (a `backend:`-style config for the read-back session), `search` (`box`, `query`), `ready`, `identity`, `fields` | `credentials` (env-var names), `sign_in`, `search.submit`, `open.results`, `not_found`, per-control `shows` / `region` / `at`, per-field `date_format` / `casefold`, `min_ocr_confidence`, `navigation_timeout_s`, `settle_s` | | `deferred_report` | `deferred_report` section: `report` (`format`, and for csv/text its columns or pattern and `accept`), `key`, `deadline_hours`, `ledger` | `report.reject`, `describe` | -| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `path_params`, `auth`, `records_drop_fields` | +| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `records_next_path`, `path_params`, `auth`, `records_drop_fields` | | `graphql` | `base_url` + `graphql_query` | `graphql_variables` (`{param: ...}` entity binding), `graphql_records_path`, `graphql_freshness_field` + `graphql_freshness_window_s`, `auth` | | `fhir` | `base_url` | `resource_type`, `search_params`, `search_param_exprs`, `field_paths`, `access_token_env` or `smart_backend`, `verify_tls`, `records_drop_fields` | | `sql` | `sql_query` + (`sqlite_database` or `sql_driver`) | `sql_query_params`, `sql_connect_args`, `sql_password_env`, `records_drop_fields` | @@ -1045,6 +1045,15 @@ Per-kind required fields: | `document-hash` | `root` | `glob` | | `surfaces` | `surfaces`: name -> one plain verifier config (any kind above) | each effect's `probe` names its surface (`surface=`); an effect naming none, or an unknown one, halts | +A paged list is read whole or not at all. `fhir` follows the Bundle's `next` +links. `rest` follows the next page's URL at `records_next_path` (a dotted +path in the body, such as `links.next`; a relative URL resolves against the +page it came from). Without `records_next_path` one response is the whole +read, as before. A next page on another origin, a page that fails, more than +50 pages, or a FHIR `total` above the number of resources read makes the read +unreadable (INDETERMINATE), never a confirmation or a refutation from the +first pages. + `records_drop_fields` (`rest`, `sql`, `fhir`) removes fields from every record before judging. `[id]` compares records by content: a record whose id stayed the same but whose fields changed reads as one record removed and one added, diff --git a/packages/flow/openadapt_flow/deployment.py b/packages/flow/openadapt_flow/deployment.py index 3fa342ef1..ed165dcbe 100644 --- a/packages/flow/openadapt_flow/deployment.py +++ b/packages/flow/openadapt_flow/deployment.py @@ -387,6 +387,11 @@ class EffectsConfig(BaseModel): base_url: Optional[str] = None records_path: str = "/api/db" records_key: Optional[str] = "records" + #: ``rest`` only: dotted path in the response body to the next page's URL + #: (``links.next``). Set, the read follows every page on the same origin + #: and reads as unreadable when a page fails or the chain is too long. + #: Unset: one response is the whole read, as before. + records_next_path: Optional[str] = None #: Secret-isolated auth headers for the records read (bearer_env / #: header+value_env / basic_env). Never a credential literal. auth: Optional[AuthRef] = None @@ -612,7 +617,7 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: """Keep configs without a desktop read-back byte-identical.""" data: dict[str, Any] = handler(self) - for added in ("desktop_readback", "deferred_report"): + for added in ("desktop_readback", "deferred_report", "records_next_path"): if data.get(added, "") is None: data.pop(added, None) return data @@ -1887,6 +1892,7 @@ def sanitized(value: Any, *, key: str = "") -> Any: "fresh_session", "desktop_readback", "deferred_report", + "records_next_path", ): if config.get(added) is None: config.pop(added, None) @@ -2039,6 +2045,7 @@ def _build_effect_verifier_unredacted( timeout_s=cfg.timeout_s, poll_interval_s=cfg.poll_interval_s, drop_fields=tuple(cfg.records_drop_fields), + next_path=cfg.records_next_path, ) if kind == "fhir": diff --git a/packages/flow/openadapt_flow/runtime/effects/fhir.py b/packages/flow/openadapt_flow/runtime/effects/fhir.py index 912e340b6..c3db82d91 100644 --- a/packages/flow/openadapt_flow/runtime/effects/fhir.py +++ b/packages/flow/openadapt_flow/runtime/effects/fhir.py @@ -67,6 +67,31 @@ } +#: A search result longer than this many pages reads as unreadable, never as +#: the first pages. +MAX_PAGES = 50 + + +def _next_link(bundle: dict[str, Any]) -> Optional[str]: + """The Bundle's ``next`` page URL, if any.""" + + links = bundle.get("link") + if not isinstance(links, list): + return None + for link in links: + if isinstance(link, dict) and link.get("relation") == "next": + url = link.get("url") + return url if isinstance(url, str) and url else None + return None + + +def _same_origin(url: str, base_url: str) -> bool: + from urllib.parse import urlsplit + + a, b = urlsplit(url), urlsplit(base_url) + return (a.scheme, a.netloc) == (b.scheme, b.netloc) + + class AccessTokenSource(Protocol): """Supplies bearer tokens that can expire.""" @@ -240,19 +265,45 @@ def _search(self) -> Optional[list[dict[str, Any]]]: bundle = resp.json() except Exception: # noqa: BLE001 - unparseable body is unreadable return None - if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle": - return None - entries = bundle.get("entry", []) - if not isinstance(entries, list): - return None out: list[dict[str, Any]] = [] - for entry in entries: - if not isinstance(entry, dict): + total: Any = None + for page in range(MAX_PAGES): + if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle": + return None + if page == 0: + total = bundle.get("total") + entries = bundle.get("entry", []) + if not isinstance(entries, list): + return None + for entry in entries: + if not isinstance(entry, dict): + return None + resource = entry.get("resource") + if not isinstance(resource, dict): + return None + out.append(self._flatten(resource)) + next_url = _next_link(bundle) + if next_url is None: + break + if not _same_origin(next_url, self.base_url): + return None # a next page elsewhere is not this system of record + try: + resp = session.get( + next_url, headers=headers, timeout=self.timeout_s, **tls_kwargs + ) + except Exception: # noqa: BLE001 - a lost page makes the set partial + return None + if resp.status_code // 100 != 2: return None - resource = entry.get("resource") - if not isinstance(resource, dict): + try: + bundle = resp.json() + except Exception: # noqa: BLE001 return None - out.append(self._flatten(resource)) + else: + return None # more pages than MAX_PAGES: never a truncated read + if isinstance(total, int) and not isinstance(total, bool) and total > len(out): + # The server says there are more matches than it returned. + return None return drop_record_fields(out, self.drop_fields) # -- EffectVerifier protocol -------------------------------------------- diff --git a/packages/flow/openadapt_flow/runtime/effects/rest.py b/packages/flow/openadapt_flow/runtime/effects/rest.py index d9e9f608d..1ae755145 100644 --- a/packages/flow/openadapt_flow/runtime/effects/rest.py +++ b/packages/flow/openadapt_flow/runtime/effects/rest.py @@ -22,6 +22,7 @@ import time from typing import Any, Optional +from urllib.parse import urljoin, urlsplit from openadapt_flow.runtime.effects._common import drop_record_fields, judge_records from openadapt_flow.runtime.effects.adapter import VerifierAdapterBase @@ -32,6 +33,25 @@ ) from openadapt_flow.verification import VerificationTier +#: A paged read longer than this many pages reads as unreadable. +MAX_PAGES = 50 + +_UNREADABLE = object() + + +def _dotted(body: Any, path: str) -> Any: + node = body + for part in path.split("."): + if not isinstance(node, dict): + return None + node = node.get(part) + return node + + +def _same_origin(url: str, base_url: str) -> bool: + a, b = urlsplit(url), urlsplit(base_url) + return (a.scheme, a.netloc) == (b.scheme, b.netloc) + class RestRecordVerifier(VerifierAdapterBase): """Verify effects against a JSON REST system of record. @@ -55,6 +75,10 @@ class RestRecordVerifier(VerifierAdapterBase): records by their whole content, so a change to another record reads as that record removed and a new one added, and the collateral-loss check catches it). Empty -> records as read. + next_path: Dotted path in the body to the next page's URL. Set, every + page on the same origin is read (at most :data:`MAX_PAGES`); a + failed page or a longer chain reads as unreadable. Unset, one + response is the whole read. """ substrate = "rest" @@ -71,8 +95,10 @@ def __init__( timeout_s: float = 5.0, poll_interval_s: float = 0.2, drop_fields: tuple[str, ...] = (), + next_path: Optional[str] = None, ) -> None: self.base_url = base_url.rstrip("/") + self.next_path = next_path self.drop_fields = tuple(drop_fields) self.records_path = records_path self.records_key = records_key @@ -107,13 +133,9 @@ def _get_session(self) -> Any: self._session = requests.Session() return self._session - def _fetch_records(self) -> Optional[list[dict[str, Any]]]: - """GET the system-of-record document and extract the records list. + def _get_json(self, url: str) -> Any: + """One GET; the parsed body, or ``_UNREADABLE``. Never raises.""" - Returns ``None`` -- read as unreadable, forcing INDETERMINATE -- on - any transport error, non-2xx status, or shape mismatch. Never raises. - """ - url = f"{self.base_url}{self.records_path}" try: # ``headers`` is passed ONLY when configured, so injected # ``requests``-style sessions with a narrower ``get`` signature @@ -125,23 +147,56 @@ def _fetch_records(self) -> Optional[list[dict[str, Any]]]: else: resp = self._get_session().get(url, timeout=self.timeout_s) except Exception: # noqa: BLE001 - any transport failure is unreadable - return None + return _UNREADABLE if resp.status_code // 100 != 2: - return None + return _UNREADABLE try: - body = resp.json() + return resp.json() except Exception: # noqa: BLE001 - unparseable body is unreadable - return None + return _UNREADABLE + + def _records_of(self, body: Any) -> Optional[list[Any]]: if self.records_key is None: records = body elif isinstance(body, dict): records = body.get(self.records_key) else: return None - if not isinstance(records, list): + return records if isinstance(records, list) else None + + def _fetch_records(self) -> Optional[list[dict[str, Any]]]: + """GET the system-of-record document and extract the records list. + + Returns ``None`` -- read as unreadable, forcing INDETERMINATE -- on + any transport error, non-2xx status, or shape mismatch, and, with + ``next_path``, on a failed page, a page on another origin or a chain + longer than :data:`MAX_PAGES`. Never raises. + """ + url = f"{self.base_url}{self.records_path}" + collected: list[Any] = [] + for _page in range(MAX_PAGES): + body = self._get_json(url) + if body is _UNREADABLE: + return None + records = self._records_of(body) + if records is None: + return None + collected.extend(records) + if self.next_path is None: + break + following = _dotted(body, self.next_path) + if following in (None, ""): + break + if not isinstance(following, str): + return None + following = urljoin(url, following) + if not _same_origin(following, self.base_url): + return None + url = following + else: return None return drop_record_fields( - [r for r in records if isinstance(r, dict)], self.drop_fields + [r for r in collected if isinstance(r, dict)], self.drop_fields ) # -- EffectVerifier protocol -------------------------------------------- diff --git a/packages/flow/tests/test_effect_paging.py b/packages/flow/tests/test_effect_paging.py new file mode 100644 index 000000000..cccbf9815 --- /dev/null +++ b/packages/flow/tests/test_effect_paging.py @@ -0,0 +1,273 @@ +"""A paged read is the whole read, or unreadable: never its first page. + +The FHIR verifier follows a Bundle's ``next`` links and checks ``total``; the +REST verifier follows ``records_next_path`` when it is set. A page on another +origin, a failed page, a chain longer than ``MAX_PAGES`` or a ``total`` above +what was read makes the read unreadable, so no check confirms or refutes from +part of the store. Records are made up. +""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Callable +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + +import pytest + +from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.runtime.effects import fhir, rest +from openadapt_flow.runtime.effects.fhir import FhirEffectVerifier +from openadapt_flow.runtime.effects.rest import RestRecordVerifier + +PATIENTS = [{"id": str(i), "fname": f"F{i}", "lname": f"L{i}"} for i in range(1, 6)] + +Page = Callable[[str, int], tuple[int, dict]] + + +@pytest.fixture +def serve(): + servers: list[ThreadingHTTPServer] = [] + + def start(page: Page) -> str: + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - http.server API + parts = urlsplit(self.path) + number = int(parse_qs(parts.query).get("page", ["0"])[0]) + status, body = page(parts.path, number) + data = json.dumps(body).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *_args) -> None: + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + servers.append(server) + return f"http://127.0.0.1:{server.server_address[1]}" + + yield start + for server in servers: + server.shutdown() + server.server_close() + + +# -- FHIR ------------------------------------------------------------------------ + + +def _bundle(url: str, number: int, *, per: int = 2, total: object = None) -> dict: + rows = PATIENTS[number * per : number * per + per] + bundle: dict = { + "resourceType": "Bundle", + "type": "searchset", + "entry": [ + { + "resource": { + "resourceType": "Patient", + "id": p["id"], + "name": [{"family": p["lname"], "given": [p["fname"]]}], + } + } + for p in rows + ], + "link": [{"relation": "self", "url": f"{url}/fhir/Patient?page={number}"}], + } + if (number + 1) * per < len(PATIENTS): + bundle["link"].append( + {"relation": "next", "url": f"{url}/fhir/Patient?page={number + 1}"} + ) + if total is not None: + bundle["total"] = total + return bundle + + +def _fhir(url: str) -> FhirEffectVerifier: + return FhirEffectVerifier( + f"{url}/fhir", + resource_type="Patient", + field_paths={"id": "id", "lname": "name.0.family"}, + access_token="test-token-not-a-secret", + timeout_s=2.0, + ) + + +def _read(verifier) -> list | None: + state = verifier.capture_pre_state() + return list(state.records) if state.reachable else None + + +def test_fhir_reads_every_page(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, total=5))) + holder["url"] = url + records = _read(_fhir(url)) + assert records is not None + assert [r["id"] for r in records] == ["1", "2", "3", "4", "5"] + + +def test_fhir_one_page_without_a_next_link_is_the_whole_read(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10))) + holder["url"] = url + assert len(_read(_fhir(url)) or []) == 5 + + +def test_fhir_next_page_on_another_origin_is_unreadable(serve) -> None: + def page(path: str, n: int) -> tuple[int, dict]: + bundle = _bundle("http://203.0.113.9", n) + return 200, bundle + + url = serve(page) + assert _read(_fhir(url)) is None + + +def test_fhir_failed_later_page_is_unreadable(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + if n == 2: + return 503, {"resourceType": "OperationOutcome"} + return 200, _bundle(holder["url"], n) + + holder["url"] = serve(page) + assert _read(_fhir(holder["url"])) is None + + +def test_fhir_total_above_the_read_is_unreadable(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10, total=7))) + holder["url"] = url + assert _read(_fhir(url)) is None + + +def test_fhir_endless_next_links_are_unreadable(serve, monkeypatch) -> None: + monkeypatch.setattr(fhir, "MAX_PAGES", 4) + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + bundle = _bundle(holder["url"], 0) + bundle["link"][-1]["url"] = f"{holder['url']}/fhir/Patient?page={n + 1}" + return 200, bundle + + holder["url"] = serve(page) + assert _read(_fhir(holder["url"])) is None + + +# -- REST ------------------------------------------------------------------------ + + +def _rest_page( + n: int, *, per: int = 2, next_url: Callable[[int], str] | None = None +) -> dict: + rows = PATIENTS[n * per : n * per + per] + more = (n + 1) * per < len(PATIENTS) + nxt = next_url(n + 1) if (more and next_url) else None + return {"data": rows, "links": {"next": nxt}} + + +def _rest(url: str, next_path: str | None = "links.next") -> RestRecordVerifier: + return RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + timeout_s=2.0, + next_path=next_path, + ) + + +def test_rest_follows_relative_and_absolute_next_links(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + # Page 1 links absolutely, page 2 relatively. + return 200, _rest_page( + n, + next_url=lambda m: ( + f"{holder['url']}/api/patient?page={m}" + if m == 1 + else f"/api/patient?page={m}" + ), + ) + + holder["url"] = serve(page) + records = _read(_rest(holder["url"])) + assert records is not None and [r["id"] for r in records] == list("12345") + + +def test_rest_without_next_path_reads_one_response(serve) -> None: + holder: dict = {} + holder["url"] = serve( + lambda path, n: ( + 200, + _rest_page(n, next_url=lambda m: f"/api/patient?page={m}"), + ) + ) + records = _read(_rest(holder["url"], next_path=None)) + assert records is not None and len(records) == 2 + + +def test_rest_next_page_on_another_origin_is_unreadable(serve) -> None: + url = serve( + lambda path, n: ( + 200, + _rest_page( + n, next_url=lambda m: f"http://203.0.113.9/api/patient?page={m}" + ), + ) + ) + assert _read(_rest(url)) is None + + +def test_rest_failed_later_page_is_unreadable(serve) -> None: + def page(path: str, n: int) -> tuple[int, dict]: + if n == 1: + return 500, {"error": "unavailable"} + return 200, _rest_page(n, next_url=lambda m: f"/api/patient?page={m}") + + assert _read(_rest(serve(page))) is None + + +def test_rest_a_next_link_that_is_not_text_is_unreadable(serve) -> None: + url = serve(lambda path, n: (200, {"data": PATIENTS[:2], "links": {"next": 7}})) + assert _read(_rest(url)) is None + + +def test_rest_endless_next_links_are_unreadable(serve, monkeypatch) -> None: + monkeypatch.setattr(rest, "MAX_PAGES", 4) + url = serve( + lambda path, n: ( + 200, + {"data": PATIENTS[:1], "links": {"next": f"/api/patient?page={n + 1}"}}, + ) + ) + assert _read(_rest(url)) is None + + +def test_records_next_path_in_a_config(serve) -> None: + holder: dict = {} + holder["url"] = serve( + lambda path, n: ( + 200, + _rest_page(n, next_url=lambda m: f"/api/patient?page={m}"), + ) + ) + section = { + "kind": "rest", + "base_url": holder["url"], + "records_path": "/api/patient", + "records_key": "data", + "records_next_path": "links.next", + } + verifier = build_effect_verifier(EffectsConfig.model_validate(section)) + assert len(_read(verifier) or []) == 5 + # Unset, the config serializes as before, so existing digests hold. + plain = EffectsConfig.model_validate( + {k: v for k, v in section.items() if k != "records_next_path"} + ) + assert "records_next_path" not in plain.model_dump(mode="json") From a6efabd1956225a72a8b477dacef881793d2a4be Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 25 Oct 2026 01:16:00 -0400 Subject: [PATCH 2/2] effects: exact FHIR total, REST records_total_path, no redirects Review of #1538: - FHIR: a total other than the number of matches read (included resources don't count) is unreadable, not only a total above it. - REST: records_total_path checks the record count against a total in the first page; without it a REST read trusts the page chain (documented). - A requests session doesn't follow redirects, so the configured auth headers never reach another host, and a response that landed on another origin is unreadable. - Each unreadable read says why (verdict reason, detail.read_problem), such as "next link origin X != base_url Y". OpenEMR builds REST next links from its site address; EFFECT_KIT.md says to set base_url to it, with a test. - The CAPABILITIES row names this PR. Co-authored-by: yunusarda-sudo Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 2 +- llms-full.txt | 2 +- packages/flow/docs/EFFECT_KIT.md | 24 ++- packages/flow/openadapt_flow/deployment.py | 13 +- .../openadapt_flow/runtime/effects/fhir.py | 138 +++++++++++------ .../openadapt_flow/runtime/effects/rest.py | 83 +++++++++- packages/flow/tests/test_effect_paging.py | 143 +++++++++++++++++- 7 files changed, 341 insertions(+), 64 deletions(-) diff --git a/CAPABILITIES.md b/CAPABILITIES.md index dbbf51508..2ab0cece1 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -210,7 +210,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | -| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page or a short read is unreadable | On main | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](packages/flow/docs/EFFECT_KIT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page, a redirect or a count that doesn't match the total is unreadable | On main ([#1538](https://github.com/OpenAdaptAI/OpenAdapt/pull/1538)) | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/llms-full.txt b/llms-full.txt index 64bd29116..080b3397e 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -607,7 +607,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | -| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page or a short read is unreadable | On main | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/EFFECT_KIT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page, a redirect or a count that doesn't match the total is unreadable | On main ([#1538](https://github.com/OpenAdaptAI/OpenAdapt/pull/1538)) | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 066685dad..81d3ec6f8 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -1035,7 +1035,7 @@ Per-kind required fields: | `ui_readback` | `ui_readback` section: `record_url`, `sign_in` (env-var names for the read-back account), `ready_selector`, `row_selector`, `fields`, `identity` | `url_params`, `empty_selector`, `not_found_selector`, `read_post_paths`, per-field `attribute` / `date_format` / `casefold`, `truncation_markers`, `navigation_timeout_s` | | `desktop_readback` | `desktop_readback` section: `session` (a `backend:`-style config for the read-back session), `search` (`box`, `query`), `ready`, `identity`, `fields` | `credentials` (env-var names), `sign_in`, `search.submit`, `open.results`, `not_found`, per-control `shows` / `region` / `at`, per-field `date_format` / `casefold`, `min_ocr_confidence`, `navigation_timeout_s`, `settle_s` | | `deferred_report` | `deferred_report` section: `report` (`format`, and for csv/text its columns or pattern and `accept`), `key`, `deadline_hours`, `ledger` | `report.reject`, `describe` | -| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `records_next_path`, `path_params`, `auth`, `records_drop_fields` | +| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `records_next_path`, `records_total_path`, `path_params`, `auth`, `records_drop_fields` | | `graphql` | `base_url` + `graphql_query` | `graphql_variables` (`{param: ...}` entity binding), `graphql_records_path`, `graphql_freshness_field` + `graphql_freshness_window_s`, `auth` | | `fhir` | `base_url` | `resource_type`, `search_params`, `search_param_exprs`, `field_paths`, `access_token_env` or `smart_backend`, `verify_tls`, `records_drop_fields` | | `sql` | `sql_query` + (`sqlite_database` or `sql_driver`) | `sql_query_params`, `sql_connect_args`, `sql_password_env`, `records_drop_fields` | @@ -1050,9 +1050,25 @@ links. `rest` follows the next page's URL at `records_next_path` (a dotted path in the body, such as `links.next`; a relative URL resolves against the page it came from). Without `records_next_path` one response is the whole read, as before. A next page on another origin, a page that fails, more than -50 pages, or a FHIR `total` above the number of resources read makes the read -unreadable (INDETERMINATE), never a confirmation or a refutation from the -first pages. +50 pages, or a FHIR `total` other than the number of matches read (included +resources don't count) makes the read unreadable (INDETERMINATE), never a +confirmation or a refutation from the first pages. + +REST has no standard total. With `records_total_path` (a dotted path in the +first page, such as `meta.total`), a read whose record count differs, or a +first page without the total, is unreadable. Without it, a REST read trusts +the page chain. + +A redirect is a failed read: the configured auth headers never follow it to +another host. Why a read was unreadable is in the verdict's reason and the +state's `detail.read_problem`, for example +`next link origin https://localhost:9300 != base_url http://localhost:9520`. + +OpenEMR builds its REST next links from its configured site address +(`site_addr_oath`), not from the address the request used. Read through a +mapped port or another host name, every next link names another origin and +the read is unreadable. Set `base_url` to the site address OpenEMR is +configured with. `records_drop_fields` (`rest`, `sql`, `fhir`) removes fields from every record before judging. `[id]` compares records by content: a record whose id stayed diff --git a/packages/flow/openadapt_flow/deployment.py b/packages/flow/openadapt_flow/deployment.py index ed165dcbe..a634da344 100644 --- a/packages/flow/openadapt_flow/deployment.py +++ b/packages/flow/openadapt_flow/deployment.py @@ -392,6 +392,10 @@ class EffectsConfig(BaseModel): #: and reads as unreadable when a page fails or the chain is too long. #: Unset: one response is the whole read, as before. records_next_path: Optional[str] = None + #: ``rest`` only: dotted path in the first page's body to the number of + #: records the read should return (``meta.total``). Set, a read whose + #: count differs reads as unreadable. Unset: the read trusts the chain. + records_total_path: Optional[str] = None #: Secret-isolated auth headers for the records read (bearer_env / #: header+value_env / basic_env). Never a credential literal. auth: Optional[AuthRef] = None @@ -617,7 +621,12 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: """Keep configs without a desktop read-back byte-identical.""" data: dict[str, Any] = handler(self) - for added in ("desktop_readback", "deferred_report", "records_next_path"): + for added in ( + "desktop_readback", + "deferred_report", + "records_next_path", + "records_total_path", + ): if data.get(added, "") is None: data.pop(added, None) return data @@ -1893,6 +1902,7 @@ def sanitized(value: Any, *, key: str = "") -> Any: "desktop_readback", "deferred_report", "records_next_path", + "records_total_path", ): if config.get(added) is None: config.pop(added, None) @@ -2046,6 +2056,7 @@ def _build_effect_verifier_unredacted( poll_interval_s=cfg.poll_interval_s, drop_fields=tuple(cfg.records_drop_fields), next_path=cfg.records_next_path, + total_path=cfg.records_total_path, ) if kind == "fhir": diff --git a/packages/flow/openadapt_flow/runtime/effects/fhir.py b/packages/flow/openadapt_flow/runtime/effects/fhir.py index c3db82d91..be28e37bc 100644 --- a/packages/flow/openadapt_flow/runtime/effects/fhir.py +++ b/packages/flow/openadapt_flow/runtime/effects/fhir.py @@ -85,11 +85,15 @@ def _next_link(bundle: dict[str, Any]) -> Optional[str]: return None -def _same_origin(url: str, base_url: str) -> bool: +def _origin(url: str) -> str: from urllib.parse import urlsplit - a, b = urlsplit(url), urlsplit(base_url) - return (a.scheme, a.netloc) == (b.scheme, b.netloc) + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}" + + +def _same_origin(url: str, base_url: str) -> bool: + return _origin(url) == _origin(base_url) class AccessTokenSource(Protocol): @@ -180,6 +184,8 @@ def __init__( # (``requests``-style) session; flipped only for the httpx fallback # client built in :meth:`_get_session`. self._verify_per_request = True + #: Why the last read was unreadable, for the state and the verdict. + self.read_problem: Optional[str] = None # -- transport ---------------------------------------------------------- @@ -238,71 +244,102 @@ def _search(self) -> Optional[list[dict[str, Any]]]: transport error, non-2xx (incl. auth 401/403), non-Bundle body, or malformed entries. Never raises. """ + self.read_problem = None url = f"{self.base_url}/{self.resource_type}" headers = self._headers() if headers is None: + self.read_problem = "no access token" return None session = self._get_session() # ``verify`` is passed per request only for a ``requests``-style client; # the httpx fallback already bound it at construction (see - # ``_get_session``), and passing it again would raise. - tls_kwargs = {"verify": self.verify_tls} if self._verify_per_request else {} - try: - resp = session.get( - url, - params=self.search_params, - headers=headers, - timeout=self.timeout_s, - **tls_kwargs, - ) - except Exception: # noqa: BLE001 - transport failure is unreadable - return None - if resp.status_code == 401 and self.token_source is not None: - self.token_source.invalidate() - if resp.status_code // 100 != 2: - return None - try: - bundle = resp.json() - except Exception: # noqa: BLE001 - unparseable body is unreadable - return None - out: list[dict[str, Any]] = [] - total: Any = None - for page in range(MAX_PAGES): + # ``_get_session``), and passing it again would raise. A ``requests`` + # session must not follow a redirect with the bearer token. + tls_kwargs: dict[str, Any] = ( + {"verify": self.verify_tls} if self._verify_per_request else {} + ) + if (type(session).__module__ or "").startswith("requests"): + tls_kwargs["allow_redirects"] = False + + def get(target: str, params: Optional[dict[str, str]]) -> Any: + """One page: the parsed Bundle, or None with ``read_problem``.""" + + try: + kwargs = dict(tls_kwargs) + if params is not None: + kwargs["params"] = params + resp = session.get( + target, headers=headers, timeout=self.timeout_s, **kwargs + ) + except Exception as exc: # noqa: BLE001 - transport failure is unreadable + self.read_problem = f"GET {target} failed: {type(exc).__name__}" + return None + if resp.status_code == 401 and self.token_source is not None: + self.token_source.invalidate() + if resp.status_code // 100 != 2: + self.read_problem = f"GET {target} returned HTTP {resp.status_code}" + return None + landed = getattr(resp, "url", None) + landed = str(landed) if landed is not None else "" + if landed and _origin(landed) != _origin(target): + self.read_problem = ( + f"GET {target} was redirected to origin {_origin(landed)}" + ) + return None + try: + bundle = resp.json() + except Exception: # noqa: BLE001 - unparseable body is unreadable + self.read_problem = f"GET {target} returned a body that is not JSON" + return None if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle": + self.read_problem = f"GET {target} returned no Bundle" return None - if page == 0: - total = bundle.get("total") + return bundle + + bundle = get(url, self.search_params) + if bundle is None: + return None + out: list[dict[str, Any]] = [] + matches = 0 + total: Any = bundle.get("total") + for _page in range(MAX_PAGES): entries = bundle.get("entry", []) if not isinstance(entries, list): + self.read_problem = "a Bundle's entry is not a list" return None for entry in entries: if not isinstance(entry, dict): + self.read_problem = "a Bundle entry is not an object" return None resource = entry.get("resource") if not isinstance(resource, dict): + self.read_problem = "a Bundle entry has no resource" return None + mode = (entry.get("search") or {}).get("mode") + if mode not in ("include", "outcome"): + matches += 1 out.append(self._flatten(resource)) next_url = _next_link(bundle) if next_url is None: break if not _same_origin(next_url, self.base_url): - return None # a next page elsewhere is not this system of record - try: - resp = session.get( - next_url, headers=headers, timeout=self.timeout_s, **tls_kwargs + # A next page elsewhere is not this system of record. + self.read_problem = ( + f"next link origin {_origin(next_url)} != base_url " + f"{_origin(self.base_url)}" ) - except Exception: # noqa: BLE001 - a lost page makes the set partial - return None - if resp.status_code // 100 != 2: - return None - try: - bundle = resp.json() - except Exception: # noqa: BLE001 return None + bundle = get(next_url, None) + if bundle is None: + return None # a lost page makes the set partial else: - return None # more pages than MAX_PAGES: never a truncated read - if isinstance(total, int) and not isinstance(total, bool) and total > len(out): - # The server says there are more matches than it returned. + self.read_problem = f"more than {MAX_PAGES} pages" + return None # never a truncated read + if isinstance(total, int) and not isinstance(total, bool) and total != matches: + # The server counts a different number of matches than it returned. + self.read_problem = ( + f"the Bundle's total is {total}, the read returned {matches} matches" + ) return None return drop_record_fields(out, self.drop_fields) @@ -310,15 +347,18 @@ def _search(self) -> Optional[list[dict[str, Any]]]: def capture_pre_state(self, context: Any = None) -> EffectState: records = self._search() + detail: dict[str, Any] = { + "base_url": self.base_url, + "resource_type": self.resource_type, + "search_params": self.search_params, + } + if records is None and self.read_problem: + detail["read_problem"] = self.read_problem return EffectState( substrate=self.substrate, reachable=records is not None, records=records or [], - detail={ - "base_url": self.base_url, - "resource_type": self.resource_type, - "search_params": self.search_params, - }, + detail=detail, ) def verify( @@ -329,6 +369,10 @@ def verify( while True: current = self._search() last = judge_records(expected, before, current, substrate=self.substrate) + if current is None and self.read_problem: + last = last.model_copy( + update={"reason": f"{last.reason} ({self.read_problem})"} + ) if last.confirmed or time.monotonic() >= deadline: return last time.sleep(self.poll_interval_s) diff --git a/packages/flow/openadapt_flow/runtime/effects/rest.py b/packages/flow/openadapt_flow/runtime/effects/rest.py index 1ae755145..390d7ce6d 100644 --- a/packages/flow/openadapt_flow/runtime/effects/rest.py +++ b/packages/flow/openadapt_flow/runtime/effects/rest.py @@ -48,9 +48,25 @@ def _dotted(body: Any, path: str) -> Any: return node +def _origin(url: str) -> str: + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}" + + def _same_origin(url: str, base_url: str) -> bool: - a, b = urlsplit(url), urlsplit(base_url) - return (a.scheme, a.netloc) == (b.scheme, b.netloc) + return _origin(url) == _origin(base_url) + + +def _no_redirects(session: Any) -> dict[str, Any]: + """Keyword that stops ``requests`` following a redirect. + + A redirect could carry the configured auth headers to another host, so a + 3xx is a failed read. ``httpx`` doesn't follow redirects by default; an + injected session keeps its own ``get`` signature. + """ + + module = type(session).__module__ or "" + return {"allow_redirects": False} if module.startswith("requests") else {} class RestRecordVerifier(VerifierAdapterBase): @@ -79,6 +95,13 @@ class RestRecordVerifier(VerifierAdapterBase): page on the same origin is read (at most :data:`MAX_PAGES`); a failed page or a longer chain reads as unreadable. Unset, one response is the whole read. + total_path: Dotted path in the first page's body to the number of + records the server holds for the read. Set, a read whose record + count differs, or a body without it, reads as unreadable. Unset, + the read trusts the page chain. + + Why the last read was unreadable is in :attr:`read_problem` (and in the + state's ``detail`` and the verdict's reason). """ substrate = "rest" @@ -96,9 +119,12 @@ def __init__( poll_interval_s: float = 0.2, drop_fields: tuple[str, ...] = (), next_path: Optional[str] = None, + total_path: Optional[str] = None, ) -> None: self.base_url = base_url.rstrip("/") self.next_path = next_path + self.total_path = total_path + self.read_problem: Optional[str] = None self.drop_fields = tuple(drop_fields) self.records_path = records_path self.records_key = records_key @@ -136,23 +162,34 @@ def _get_session(self) -> Any: def _get_json(self, url: str) -> Any: """One GET; the parsed body, or ``_UNREADABLE``. Never raises.""" + session = self._get_session() try: # ``headers`` is passed ONLY when configured, so injected # ``requests``-style sessions with a narrower ``get`` signature # (tests, custom oracles) keep working exactly as before. if self.headers: - resp = self._get_session().get( - url, timeout=self.timeout_s, headers=self.headers + resp = session.get( + url, + timeout=self.timeout_s, + headers=self.headers, + **_no_redirects(session), ) else: - resp = self._get_session().get(url, timeout=self.timeout_s) - except Exception: # noqa: BLE001 - any transport failure is unreadable + resp = session.get(url, timeout=self.timeout_s) + except Exception as exc: # noqa: BLE001 - any transport failure is unreadable + self.read_problem = f"GET {url} failed: {type(exc).__name__}" return _UNREADABLE if resp.status_code // 100 != 2: + self.read_problem = f"GET {url} returned HTTP {resp.status_code}" + return _UNREADABLE + landed = getattr(resp, "url", None) + if isinstance(landed, str) and landed and not _same_origin(landed, url): + self.read_problem = f"GET {url} was redirected to origin {_origin(landed)}" return _UNREADABLE try: return resp.json() except Exception: # noqa: BLE001 - unparseable body is unreadable + self.read_problem = f"GET {url} returned a body that is not JSON" return _UNREADABLE def _records_of(self, body: Any) -> Optional[list[Any]]: @@ -172,15 +209,20 @@ def _fetch_records(self) -> Optional[list[dict[str, Any]]]: ``next_path``, on a failed page, a page on another origin or a chain longer than :data:`MAX_PAGES`. Never raises. """ + self.read_problem = None url = f"{self.base_url}{self.records_path}" collected: list[Any] = [] - for _page in range(MAX_PAGES): + total: Any = None + for page in range(MAX_PAGES): body = self._get_json(url) if body is _UNREADABLE: return None records = self._records_of(body) if records is None: + self.read_problem = f"GET {url}: no records list in the body" return None + if page == 0 and self.total_path is not None: + total = _dotted(body, self.total_path) collected.extend(records) if self.next_path is None: break @@ -188,13 +230,31 @@ def _fetch_records(self) -> Optional[list[dict[str, Any]]]: if following in (None, ""): break if not isinstance(following, str): + self.read_problem = f"the next link at {self.next_path} is not a URL" return None following = urljoin(url, following) if not _same_origin(following, self.base_url): + # OpenEMR, for one, builds next links from its configured site + # address, which can differ from the address the read uses. + self.read_problem = ( + f"next link origin {_origin(following)} != base_url " + f"{_origin(self.base_url)}" + ) return None url = following else: + self.read_problem = f"more than {MAX_PAGES} pages" return None + if self.total_path is not None: + if isinstance(total, bool) or not isinstance(total, int): + self.read_problem = f"no record total at {self.total_path}" + return None + if total != len(collected): + self.read_problem = ( + f"the record total at {self.total_path} is {total}, the read " + f"returned {len(collected)}" + ) + return None return drop_record_fields( [r for r in collected if isinstance(r, dict)], self.drop_fields ) @@ -203,11 +263,14 @@ def _fetch_records(self) -> Optional[list[dict[str, Any]]]: def capture_pre_state(self, context: Any = None) -> EffectState: records = self._fetch_records() + detail: dict[str, Any] = {"url": f"{self.base_url}{self.records_path}"} + if records is None and self.read_problem: + detail["read_problem"] = self.read_problem return EffectState( substrate=self.substrate, reachable=records is not None, records=records or [], - detail={"url": f"{self.base_url}{self.records_path}"}, + detail=detail, ) def verify( @@ -222,6 +285,10 @@ def verify( while True: current = self._fetch_records() last = judge_records(expected, before, current, substrate=self.substrate) + if current is None and self.read_problem: + last = last.model_copy( + update={"reason": f"{last.reason} ({self.read_problem})"} + ) if last.confirmed or time.monotonic() >= deadline: return last time.sleep(self.poll_interval_s) diff --git a/packages/flow/tests/test_effect_paging.py b/packages/flow/tests/test_effect_paging.py index cccbf9815..8f7740a0c 100644 --- a/packages/flow/tests/test_effect_paging.py +++ b/packages/flow/tests/test_effect_paging.py @@ -24,7 +24,9 @@ PATIENTS = [{"id": str(i), "fname": f"F{i}", "lname": f"L{i}"} for i in range(1, 6)] -Page = Callable[[str, int], tuple[int, dict]] +Page = Callable[[str, int], tuple] +#: Request headers every test server saw, in order. +SEEN: list[dict] = [] @pytest.fixture @@ -36,9 +38,12 @@ class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 - http.server API parts = urlsplit(self.path) number = int(parse_qs(parts.query).get("page", ["0"])[0]) - status, body = page(parts.path, number) + SEEN.append(dict(self.headers)) + status, body, *extra = page(parts.path, number) data = json.dumps(body).encode() self.send_response(status) + for name, value in (extra[0] if extra else {}).items(): + self.send_header(name, value) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() @@ -271,3 +276,137 @@ def test_records_next_path_in_a_config(serve) -> None: {k: v for k, v in section.items() if k != "records_next_path"} ) assert "records_next_path" not in plain.model_dump(mode="json") + + +# -- review of #1538 --------------------------------------------------------------- + + +def test_fhir_total_below_the_read_is_unreadable(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10, total=3))) + holder["url"] = url + verifier = _fhir(url) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"] == ( + "the Bundle's total is 3, the read returned 5 matches" + ) + + +def test_fhir_included_resources_are_not_matches(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple: + bundle = _bundle(holder["url"], n, per=10, total=5) + bundle["entry"].append( + { + "resource": {"resourceType": "Practitioner", "id": "dr1"}, + "search": {"mode": "include"}, + } + ) + return 200, bundle + + holder["url"] = serve(page) + assert len(_read(_fhir(holder["url"])) or []) == 6 + + +def test_fhir_next_origin_is_named(serve) -> None: + url = serve(lambda path, n: (200, _bundle("http://203.0.113.9", n))) + state = _fhir(url).capture_pre_state() + assert state.detail["read_problem"] == ( + f"next link origin http://203.0.113.9 != base_url {url}" + ) + + +def test_openemr_next_links_name_the_site_address(serve) -> None: + """OpenEMR builds REST next links from its configured site address. Read + through a mapped port, the link names another origin: unreadable, with + both origins in the reason, never page one alone.""" + + site = "https://localhost:9300" + + def page(path: str, n: int) -> tuple: + return 200, { + "validationErrors": [], + "internalErrors": [], + "data": PATIENTS[n * 2 : n * 2 + 2], + "links": {"next": f"{site}/apis/default/api/patient?page={n + 1}"}, + } + + url = serve(page) + verifier = _rest(url) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"] == ( + f"next link origin {site} != base_url {url}" + ) + + +def _rest_total(url: str, path: str = "meta.total") -> RestRecordVerifier: + return RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + timeout_s=2.0, + next_path="links.next", + total_path=path, + ) + + +@pytest.mark.parametrize("total,ok", [(5, True), (6, False), (4, False), (None, False)]) +def test_rest_total_path(serve, total, ok) -> None: + def page(path: str, n: int) -> tuple: + body = _rest_page(n, next_url=lambda m: f"/api/patient?page={m}") + if total is not None: + body["meta"] = {"total": total} + return 200, body + + verifier = _rest_total(serve(page)) + state = verifier.capture_pre_state() + assert state.reachable is ok + if not ok: + assert "total" in state.detail["read_problem"] + + +def test_a_redirect_never_carries_the_auth_headers(serve) -> None: + """A 3xx is a failed read; the configured headers never reach the other + host.""" + + SEEN.clear() + other = serve(lambda path, n: (200, {"data": PATIENTS, "links": {}})) + url = serve(lambda path, n: (302, {}, {"Location": f"{other}/api/patient"})) + verifier = RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + headers={"X-Api-Key": "test-key-not-a-secret"}, + timeout_s=2.0, + ) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"].endswith("returned HTTP 302") + assert len(SEEN) == 1 # the other host was never asked + + SEEN.clear() + fhir_url = serve(lambda path, n: (302, {}, {"Location": f"{other}/fhir/Patient"})) + state = _fhir(fhir_url).capture_pre_state() + assert not state.reachable + assert len(SEEN) == 1 + + +def test_the_verdict_names_why_the_read_failed(serve) -> None: + from openadapt_flow.runtime.effects import Effect, EffectKind + + url = serve( + lambda path, n: ( + 200, + _rest_page( + n, next_url=lambda m: f"http://203.0.113.9/api/patient?page={m}" + ), + ) + ) + verifier = _rest(url) + before = verifier.capture_pre_state() + effect = Effect(kind=EffectKind.RECORD_WRITTEN, match={"lname": "L5"}, timeout_s=0) + verdict = verifier.verify(effect, before) + assert "next link origin http://203.0.113.9" in verdict.reason