effects: read a paged FHIR or REST record list whole, or not at all

CAPABILITIES.md

@@ -214,6 +214,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, 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` |
 

llms-full.txt

@@ -611,6 +611,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, 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` |
 

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`, `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` |
@@ -1045,6 +1045,31 @@ 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=<name>`); 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` 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
 the same but whose fields changed reads as one record removed and one added,

packages/flow/openadapt_flow/deployment.py

@@ -387,6 +387,15 @@ 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
+    #: ``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
@@ -612,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"):
+        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
@@ -1964,6 +1978,8 @@ def sanitized(value: Any, *, key: str = "") -> Any:
         "fresh_session",
         "desktop_readback",
         "deferred_report",
+        "records_next_path",
+        "records_total_path",
     ):
         if config.get(added) is None:
             config.pop(added, None)
@@ -2116,6 +2132,8 @@ 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,
+            total_path=cfg.records_total_path,
         )
 
     if kind == "fhir":

packages/flow/openadapt_flow/runtime/effects/fhir.py

@@ -67,6 +67,35 @@
 }
 
 
+#: 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 _origin(url: str) -> str:
+    from urllib.parse import urlsplit
+
+    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):
     """Supplies bearer tokens that can expire."""
 
@@ -155,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 ----------------------------------------------------------
 
@@ -213,61 +244,121 @@ 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
-        if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle":
-            return None
-        entries = bundle.get("entry", [])
-        if not isinstance(entries, list):
+        # ``_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
+            return bundle
+
+        bundle = get(url, self.search_params)
+        if bundle is None:
             return None
         out: list[dict[str, Any]] = []
-        for entry in entries:
-            if not isinstance(entry, dict):
+        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
-            resource = entry.get("resource")
-            if not isinstance(resource, dict):
+            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):
+                # 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)}"
+                )
                 return None
-            out.append(self._flatten(resource))
+            bundle = get(next_url, None)
+            if bundle is None:
+                return None  # a lost page makes the set partial
+        else:
+            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)
 
     # -- EffectVerifier protocol --------------------------------------------
 
     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(
@@ -278,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)