diff --git a/packages/flow/docs/ENSURE_RECORD.md b/packages/flow/docs/ENSURE_RECORD.md index f24905ee01..2399b6a548 100644 --- a/packages/flow/docs/ENSURE_RECORD.md +++ b/packages/flow/docs/ENSURE_RECORD.md @@ -244,3 +244,45 @@ Measured on Medplum 5.2.8 (`benchmark/fhir_medplum/readpath/probe.json`): server serves (`MedplumFixture.restrict_agent`). - The verifier's own client reads Patient and AuditEvent, and writes nothing. + +### Reference identity + +A contract can supply `reference_schema: {definition, sha256}`. `definition` +is a FHIR resource StructureDefinition snapshot; `sha256` pins its canonical +JSON (`sort_keys=True`, separators `,` and `:`). The schema's resource type +must match the contract. Only elements explicitly typed `Reference` receive +reference comparisons. Sliced paths, choice types and custom target profiles +need additional resolution support and are refused. + +Bind the task's target to the Reference element or its `.reference` member. +The reader resolves each supplied and stored target before judging. Relative +and absolute references to the same configured FHIR base identify the same +logical resource, including references to its historical versions. Resource +types and bases remain separate. A requested version is read and its version +ID checked separately; duplicate identity does not depend on that version. +An explicitly requested reference version still constrains the stored field, +and the written record keeps its separate last-version binding. Identifier +references need a system and value and a complete search that resolves exactly +one target. A reference carrying both a literal and an +identifier must resolve consistently through both. Failed reads, ambiguous +identifiers and unsupported external or contained references stay uncertain. +The judge reads only the collected evidence. Identity searches send local +`Type/id` values. A search that omits a freshly read, known matching record +is unreadable; callers of `find_matches` can supply those resource IDs through +`expected_ids`. An initial search can be empty when no match is yet known. + +Within a schema-proven Reference, `display` is optional presentation text. +It may differ only when the contract independently binds that Reference's +target identity. The verifier does not certify that this text describes the +target, even on a closed contract. A field named `display` elsewhere keeps its +existing scalar comparison. Other undeclared elements remain subject to the +closed-record check. Contracts without `reference_schema` retain their previous behavior +and serialization. + +The FHIR generator accepts optional `structure_definition.json` and +`reference_targets.json` files in its teach directory. The latter contains +the target resources captured for its demonstration checks; runtime +verification reads the targets again. Metadata and captured targets must be +supplied before the generator can use reference identity. SQL generation +continues to use declared foreign keys and unique target matches. OpenAPI +metadata and denormalized SQL label relationships are not inferred. diff --git a/packages/flow/openadapt_flow/ensure_record.py b/packages/flow/openadapt_flow/ensure_record.py index e084b01dfe..55d5f33501 100644 --- a/packages/flow/openadapt_flow/ensure_record.py +++ b/packages/flow/openadapt_flow/ensure_record.py @@ -40,7 +40,7 @@ import re import time from collections import Counter -from collections.abc import Callable, Collection, Mapping +from collections.abc import Callable, Collection, Mapping, Sequence from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from email.utils import parsedate_to_datetime @@ -48,7 +48,14 @@ from typing import Any, Literal, Optional from urllib.parse import urlsplit -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator + +from openadapt_flow.fhir_reference import ( + ReferenceEvidence, + ReferenceReader, + ReferenceSchema, + ReferenceUnreadable, +) Outcome = Literal["found_existing", "created_verified", "needs_review"] Compare = Literal["exact", "casefold", "digits", "date", "instant"] @@ -215,6 +222,8 @@ class EnsureContract(BaseModel): resource_type: str = Field(min_length=1) base_url: str = Field(min_length=1) access_token_env: Optional[str] = None + #: Exact schema metadata proving which elements are FHIR References. + reference_schema: Optional[ReferenceSchema] = None #: The agent's own actor references (``Practitioner/``). self_actors: list[str] = Field(alias="self", min_length=1) #: Task parameter -> where and how the record stores it. @@ -259,9 +268,37 @@ class EnsureContract(BaseModel): @model_validator(mode="after") def _bound(self) -> "EnsureContract": + if self.reference_schema is not None: + if self.reference_schema.definition["type"] != self.resource_type: + raise ValueError("the Reference schema describes another resource type") + paths = self.reference_schema.paths() + for name, field in self.fields.items(): + plain = _bare(field.path) + if plain in paths or plain.removesuffix(".reference") in paths: + if field.compare != "exact" or field.map: + raise ValueError( + "Reference identity cannot use scalar normalization" + ) + if ( + plain.endswith(".display") + and plain.removesuffix(".display") in paths + ): + parent = field.path.removesuffix(".display") + if not any( + f.path in (parent, parent + ".reference") + for f in self.fields.values() + ): + raise ValueError( + f"{name}: a Reference display needs a bound target identity" + ) for rule in self.identity: if not rule: raise ValueError("an identity rule names no parameter") + if any( + p in self.fields and reference_display(self, self.fields[p].path) + for p in rule + ): + raise ValueError("Reference display cannot identify a record") missing = [p for p in rule if p not in self.fields] if missing: raise ValueError(f"identity names unknown fields: {missing}") @@ -288,6 +325,13 @@ def _bound(self) -> "EnsureContract": ) return self + @model_serializer(mode="wrap") + def _serialize_reference_schema(self, handler: Any) -> Any: + data = handler(self) + if self.reference_schema is None: + data.pop("reference_schema", None) + return data + @property def corrections(self) -> bool: """Whether a correction may count: both switches must say so.""" @@ -457,8 +501,69 @@ def normalize(value: Any, rule: FieldRule, *, task: bool = False) -> Optional[st return text -def same(contract: EnsureContract, name: str, stored: Any, wanted: Any) -> bool: +def reference_field( + contract: EnsureContract, path: str +) -> Optional[tuple[str, tuple[str, ...]]]: + """A field's typed Reference parent and target types, never a name guess.""" + if contract.reference_schema is None: + return None + paths = contract.reference_schema.paths() + for parent in (path, path.removesuffix(".reference")): + if _bare(parent) in paths: + return parent, paths[_bare(parent)] + return None + + +def reference_display(contract: EnsureContract, path: str) -> bool: + return ( + path.endswith(".display") + and reference_field(contract, path.removesuffix(".display")) is not None + ) + + +def _display_has_identity( + contract: EnsureContract, path: str, params: Mapping[str, Any] +) -> bool: + parent = path.removesuffix(".display") + return any( + f.path in (parent, parent + ".reference") and params.get(n) not in (None, "") + for n, f in contract.fields.items() + ) + + +def _field_values( + contract: EnsureContract, resource: Mapping[str, Any], path: str +) -> list[Any]: + ref = reference_field(contract, path) + return extract_all(resource, ref[0] if ref else path) + + +def same( + contract: EnsureContract, + name: str, + stored: Any, + wanted: Any, + references: Optional[ReferenceEvidence] = None, + *, + logical: bool = False, +) -> bool: rule = contract.fields[name] + ref = reference_field(contract, rule.path) + if ref is not None: + if stored is None or wanted is None: + return False + if references is None: + raise ReferenceUnreadable( + "reference comparison needs resolved target evidence" + ) + if references.identity(stored, ref[1]) != references.identity(wanted, ref[1]): + return False + wanted_version = references.version(wanted, ref[1]) + return ( + logical + or wanted_version is None + or references.version(stored, ref[1]) == wanted_version + ) a, b = normalize(stored, rule), normalize(wanted, rule, task=True) return a is not None and b is not None and a == b @@ -553,7 +658,10 @@ def is_void(contract: EnsureContract, resource: Mapping[str, Any]) -> bool: def is_person( - contract: EnsureContract, resource: Mapping[str, Any], params: Mapping[str, Any] + contract: EnsureContract, + resource: Mapping[str, Any], + params: Mapping[str, Any], + references: Optional[ReferenceEvidence] = None, ) -> bool: """True when the resource meets any identity rule for these values. @@ -567,8 +675,8 @@ def is_person( return any( all( any( - same(contract, p, value, params.get(p)) - for value in extract_all(resource, contract.fields[p].path) + same(contract, p, value, params.get(p), references, logical=True) + for value in _field_values(contract, resource, contract.fields[p].path) ) for p in rule ) @@ -578,13 +686,32 @@ def is_person( def wrong_fields( - contract: EnsureContract, resource: Mapping[str, Any], params: Mapping[str, Any] + contract: EnsureContract, + resource: Mapping[str, Any], + params: Mapping[str, Any], + references: Optional[ReferenceEvidence] = None, ) -> list[str]: + for name, rule in contract.fields.items(): + if ( + name in params + and reference_display(contract, rule.path) + and not _display_has_identity(contract, rule.path, params) + ): + raise ReferenceUnreadable( + "Reference display has no supplied target identity" + ) return sorted( name for name, rule in contract.fields.items() if name in params - and not same(contract, name, extract(resource, rule.path), params[name]) + and not reference_display(contract, rule.path) + and not same( + contract, + name, + next(iter(_field_values(contract, resource, rule.path)), None), + params[name], + references, + ) ) @@ -613,7 +740,13 @@ def _expected_leaves( ) -> Counter[str]: want: Counter[str] = Counter() for name, rule in contract.fields.items(): - if params is not None and params.get(name) in (None, ""): + if params is not None and ( + params.get(name) in (None, "") + or ( + reference_display(contract, rule.path) + and not _display_has_identity(contract, rule.path, params) + ) + ): continue # the task didn't give it, so the record mustn't hold it from openadapt_flow.runtime.effects.fhir import parse_filter, split_path @@ -634,6 +767,23 @@ def _expected_leaves( for path in filters: if path != leaf: want[path] += 1 + # These allowances apply only to a schema-proven Reference with an + # independently checked identity supplied by this task. + for name, rule in contract.fields.items(): + ref = reference_field(contract, rule.path) + if ref is None or (params is not None and params.get(name) in (None, "")): + continue + parent = _bare(ref[0]) + if rule.path == ref[0]: + want.pop(parent, None) + for member in ( + "reference", + "type", + "identifier.system", + "identifier.value", + "display", + ): + want[parent + "." + member] = max(1, want[parent + "." + member]) return want @@ -663,7 +813,10 @@ def exempt(path: str) -> bool: def holds_only_task_values( - contract: EnsureContract, resource: Mapping[str, Any], params: Mapping[str, Any] + contract: EnsureContract, + resource: Mapping[str, Any], + params: Mapping[str, Any], + references: Optional[ReferenceEvidence] = None, ) -> bool: """True when every value the version holds is the task's (``consistent``). @@ -676,12 +829,22 @@ def holds_only_task_values( if is_void(contract, resource): return False for name, rule in contract.fields.items(): - held = [v for v in extract_all(resource, rule.path) if v not in (None, "")] + if reference_display(contract, rule.path): + if extract_all(resource, rule.path) and not _display_has_identity( + contract, rule.path, params + ): + return False + continue + held = [ + v + for v in _field_values(contract, resource, rule.path) + if v not in (None, "") + ] if not held: continue if params.get(name) in (None, ""): return False - if not all(same(contract, name, v, params[name]) for v in held): + if not all(same(contract, name, v, params[name], references) for v in held): return False # On any contract, open or closed: a version in between may hold nothing # the task didn't give (another person's identifier, an address). @@ -689,16 +852,19 @@ def holds_only_task_values( def _between_ok( - contract: EnsureContract, version: Mapping[str, Any], params: Mapping[str, Any] + contract: EnsureContract, + version: Mapping[str, Any], + params: Mapping[str, Any], + references: Optional[ReferenceEvidence] = None, ) -> bool: mode = contract.operations.correction.between if mode == "consistent": - return holds_only_task_values(contract, version, params) + return holds_only_task_values(contract, version, params, references) if mode == "either": - return is_person(contract, version, params) or holds_only_task_values( - contract, version, params - ) - return is_person(contract, version, params) + return is_person( + contract, version, params, references + ) or holds_only_task_values(contract, version, params, references) + return is_person(contract, version, params, references) def _strip(node: Any, path: str) -> Any: @@ -996,6 +1162,28 @@ def resources( out.append(res) return out + def reference_evidence( + self, + contract: EnsureContract, + params: Mapping[str, Any], + resources: Collection[Mapping[str, Any]], + ) -> ReferenceEvidence: + def fetch(path: str) -> Mapping[str, Any]: + _, body = self.fetch(f"{self.base_url}/{path}") + return body + + reader = ReferenceReader(self.base_url, fetch, self.resources) + for name, rule in contract.fields.items(): + ref = reference_field(contract, rule.path) + if ref is None: + continue + if params.get(name) not in (None, ""): + reader.collect(params[name], ref[1]) + for resource in resources: + for value in extract_all(resource, ref[0]): + reader.collect(value, ref[1]) + return reader.snapshot() + def server_time(self) -> datetime: """The server's clock, from the ``Date`` header of a metadata read.""" @@ -1065,13 +1253,27 @@ class Change: reads: list[tuple[Optional[datetime], frozenset[str]]] = field(default_factory=list) -def find_matches( +def _find_matches( contract: EnsureContract, reader: FhirReader, params: Mapping[str, Any] ) -> tuple[dict[str, dict[str, Any]], int]: """Every live record that is the person, by each identity rule's search.""" seen: dict[str, dict[str, Any]] = {} queries: list[dict[str, str]] = [] + expected = ( + reader.reference_evidence(contract, params, []) + if contract.reference_schema + else ReferenceEvidence() + ) + + def query_value(name: str) -> str: + ref = reference_field(contract, contract.fields[name].path) + return ( + expected.search_value(params[name], ref[1], reader.base_url) + if ref + else str(params[name]).strip() + ) + for rule in contract.identity: given = [p for p in rule if params.get(p) not in (None, "")] if len(given) < len(rule): @@ -1082,20 +1284,46 @@ def find_matches( # (an allergy's substance code) only narrows nothing: the search uses # the rule's other values, and the contract compares it after. searchable = [p for p in rule if p in contract.search] - queries.append({contract.search[p]: str(params[p]).strip() for p in searchable}) + queries.append({contract.search[p]: query_value(p) for p in searchable}) if len(searchable) > 1: - queries += [ - {contract.search[p]: str(params[p]).strip()} for p in searchable - ] + queries += [{contract.search[p]: query_value(p)} for p in searchable] if not queries: raise Unreadable("the task gives the values of no identity rule") for query in queries: for res in reader.resources(contract.resource_type, {**query, "_count": "200"}): seen[str(res["id"])] = res - matches = {i: r for i, r in seen.items() if is_person(contract, r, params)} + references = ( + reader.reference_evidence(contract, params, list(seen.values())) + if contract.reference_schema + else None + ) + matches = { + i: r for i, r in seen.items() if is_person(contract, r, params, references) + } return matches, len(seen) +def _require_matches(matches: Mapping[str, Any], expected_ids: Sequence[str]) -> None: + if set(expected_ids) - set(matches): + raise Unreadable("identifying search did not return a known matching record") + + +def find_matches( + contract: EnsureContract, + reader: FhirReader, + params: Mapping[str, Any], + *, + expected_ids: Sequence[str] = (), +) -> tuple[dict[str, dict[str, Any]], int]: + """Search logical identities; an omitted known match is unreadable, never absent.""" + try: + matches, candidates = _find_matches(contract, reader, params) + _require_matches(matches, expected_ids) + return matches, candidates + except ReferenceUnreadable as exc: + raise Unreadable(str(exc)) from exc + + def capture_baseline( contract: EnsureContract, params: Mapping[str, Any], @@ -1241,7 +1469,7 @@ def _review(reasons: list[str], evidence: dict[str, Any]) -> EnsureResult: return EnsureResult("needs_review", None, reasons, evidence) -def judge( +def _judge( contract: EnsureContract, params: Mapping[str, Any], baseline: Baseline, @@ -1249,6 +1477,7 @@ def judge( after_matches: Mapping[str, Mapping[str, Any]], created: Optional[Mapping[str, Any]], existing_now: Optional[Mapping[str, Any]] = None, + references: Optional[ReferenceEvidence] = None, ) -> EnsureResult: """Decide the outcome from the snapshots. Reads nothing itself. @@ -1336,7 +1565,7 @@ def judge( r for r in creates if r.split("/", 1)[1] not in after_matches - and transient_ok(contract, mine[r], selfs, params, changes) + and transient_ok(contract, mine[r], selfs, params, changes, references) ] creates = [r for r in creates if r not in transient] if transient: @@ -1357,7 +1586,9 @@ def judge( tid = target.split("/", 1)[1] corrected = bool(set(mine[target].actions) - {"C"}) if corrected: - refused = correction_refused(contract, mine[target], selfs, params) + refused = correction_refused( + contract, mine[target], selfs, params, references + ) if refused: reasons.append(refused) if target in others or (changes[target].authors - selfs): @@ -1372,12 +1603,12 @@ def judge( unbound = read_back_unbound(changes[target], created) if unbound: reasons.append(unbound) - wrong = wrong_fields(contract, created, params) + wrong = wrong_fields(contract, created, params, references) if wrong: reasons.append("wrong or missing: " + ", ".join(wrong)) if is_void(contract, created): reasons.append("the created record asserts nothing (void)") - elif not is_person(contract, created, params): + elif not is_person(contract, created, params, references): reasons.append("the created record is not the person") if contract.closed: extra_paths = extra_elements(contract, created, params) @@ -1418,6 +1649,32 @@ def judge( return _review(reasons, evidence) +def judge( + contract: EnsureContract, + params: Mapping[str, Any], + baseline: Baseline, + changes: Mapping[str, Change], + after_matches: Mapping[str, Mapping[str, Any]], + created: Optional[Mapping[str, Any]], + existing_now: Optional[Mapping[str, Any]] = None, + references: Optional[ReferenceEvidence] = None, +) -> EnsureResult: + """Judge snapshots without I/O; missing target evidence stays uncertain.""" + try: + return _judge( + contract, + params, + baseline, + changes, + after_matches, + created, + existing_now, + references, + ) + except ReferenceUnreadable as exc: + return _review([f"unreadable: {exc}"], {"unreadable": True}) + + def _version_meta(version: Mapping[str, Any]) -> tuple[Optional[datetime], str]: meta = version.get("meta") or {} if not isinstance(meta, Mapping): @@ -1515,6 +1772,7 @@ def correction_refused( change: Change, selfs: set[str], params: Mapping[str, Any], + references: Optional[ReferenceEvidence] = None, ) -> Optional[str]: """Why a create followed by more writes can't count as one write, or None. @@ -1544,7 +1802,9 @@ def correction_refused( return f"{changed}, and with settle_s under 1 s its reads can't be told apart" if change.versions is None: return f"{changed}, and without _history its versions can't be counted" - if any(not _between_ok(contract, body, params) for body in change.bodies): + if any( + not _between_ok(contract, body, params, references) for body in change.bodies + ): # A version that was someone else (or no one) may have been found by # a search no audit event names, and acted on. if contract.operations.correction.between != "person": @@ -1597,6 +1857,7 @@ def transient_ok( selfs: set[str], params: Mapping[str, Any], changes: Mapping[str, Change] = {}, + references: Optional[ReferenceEvidence] = None, ) -> bool: """True when ``change`` is a record the agent created and deleted again that ``cardinality.own_transient`` lets count as not created. @@ -1637,7 +1898,9 @@ def transient_ok( return False if change.versions != sum(1 for _, a in timed if a in ("C", "U")): return False - if not all(holds_only_task_values(contract, b, params) for b in change.bodies): + if not all( + holds_only_task_values(contract, b, params, references) for b in change.bodies + ): return False return not any(when is None or not who or who - selfs for when, who in change.reads) @@ -1794,13 +2057,37 @@ def changes_since(until: Optional[datetime]) -> dict[str, Change]: late = _agents_part(changes_since(None), selfs) if late != _agents_part(changes, selfs): raise Unreadable("the agent wrote after the window closed") + resources = [r for r in (created, existing_now) if isinstance(r, Mapping)] + resources += [body for change in changes.values() for body in change.bodies] + resources += list(after_matches.values()) + references = ( + reader.reference_evidence(contract, params, resources) + if contract.reference_schema + else None + ) + if contract.reference_schema: + known = [ + str(r["id"]) + for r in (created, existing_now) + if isinstance(r, Mapping) + and r.get("id") is not None + and is_person(contract, r, params, references) + ] + _require_matches(after_matches, known) result = judge( - contract, params, baseline, changes, after_matches, created, existing_now + contract, + params, + baseline, + changes, + after_matches, + created, + existing_now, + references, ) # Where the read window closed, for a write receipt's scope. result.evidence["window_until"] = until.isoformat() return result - except Unreadable as exc: + except (Unreadable, ReferenceUnreadable) as exc: return _review([f"unreadable: {exc}"], {"unreadable": True}) diff --git a/packages/flow/openadapt_flow/fhir_reference.py b/packages/flow/openadapt_flow/fhir_reference.py new file mode 100644 index 0000000000..ab9847dde7 --- /dev/null +++ b/packages/flow/openadapt_flow/fhir_reference.py @@ -0,0 +1,365 @@ +"""FHIR Reference typing and read-only target-resolution evidence. + +StructureDefinition metadata identifies Reference elements. Display text never +identifies a target. Judging consumes a snapshot; only ReferenceReader performs +reads, and it never follows a reference outside the configured FHIR base. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urlsplit + +from pydantic import BaseModel, ConfigDict, model_validator + +_RESOURCE = re.compile(r"[A-Z][A-Za-z0-9]*") +_ID = re.compile(r"[A-Za-z0-9\-.]{1,64}") +_PROFILE = "http://hl7.org/fhir/StructureDefinition/" + + +class ReferenceUnreadable(RuntimeError): + """The supplied evidence cannot establish one reference target.""" + + +def definition_digest(definition: Mapping[str, Any]) -> str: + return hashlib.sha256( + json.dumps(definition, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + + +class ReferenceSchema(BaseModel): + """A contract pins the exact StructureDefinition snapshot it trusts.""" + + model_config = ConfigDict(extra="forbid") + definition: dict[str, Any] + sha256: str + + @model_validator(mode="after") + def _snapshot(self) -> "ReferenceSchema": + d = self.definition + if self.sha256 != definition_digest(d): + raise ValueError("the Reference schema digest does not match") + if ( + d.get("resourceType") != "StructureDefinition" + or d.get("kind") != "resource" + or not isinstance(d.get("type"), str) + or not _RESOURCE.fullmatch(d["type"]) + or not d.get("url") + or not isinstance(d.get("snapshot"), dict) + or not isinstance(d["snapshot"].get("element"), list) + ): + raise ValueError( + "Reference typing needs a resource StructureDefinition snapshot" + ) + seen: set[str] = set() + for e in d["snapshot"]["element"]: + if not isinstance(e, dict) or not isinstance(e.get("path"), str): + raise ValueError("a schema element needs its path") + path = e["path"] + if path in seen: + # Slices and choice types need more context than a plain path. + raise ValueError("duplicate/sliced schema paths are unsupported") + seen.add(path) + if path != d["type"] and not path.startswith(d["type"] + "."): + raise ValueError("a schema element belongs to another resource type") + return self + + def paths(self) -> dict[str, tuple[str, ...]]: + out = {} + prefix = self.definition["type"] + "." + for e in self.definition["snapshot"]["element"]: + types = e.get("type") or [] + if not isinstance(types, list): + raise ValueError("schema element types must be a list") + if len(types) != 1 or not isinstance(types[0], dict): + continue + t = types[0] + if t.get("code") != "Reference": + continue + path = e["path"][len(prefix) :] + if not re.fullmatch(r"[A-Za-z][A-Za-z0-9]*(\.[A-Za-z][A-Za-z0-9]*)*", path): + raise ValueError("choice or non-plain Reference paths are unsupported") + profiles = t.get("targetProfile") or [] + if not isinstance(profiles, list) or any( + not isinstance(x, str) for x in profiles + ): + raise ValueError("Reference target profiles must be URLs") + targets = tuple( + p[len(_PROFILE) :] for p in profiles if p.startswith(_PROFILE) + ) + if len(targets) != len(profiles) or any( + not _RESOURCE.fullmatch(t) for t in targets + ): + raise ValueError("custom target profiles need a profile resolver") + out[path] = targets + return out + + +def reference_value(value: Any) -> dict[str, Any]: + if isinstance(value, str): + value = {"reference": value} + if not isinstance(value, Mapping): + raise ReferenceUnreadable( + "a Reference is neither an object nor a literal reference" + ) + # Evidence contains identity material only. Unsupported Identifier + # metadata must not be silently dropped or copied into a proof snapshot. + out = {k: value[k] for k in ("reference", "type") if k in value} + if any(not isinstance(v, str) for v in out.values()): + raise ReferenceUnreadable("literal reference and type must be strings") + if "reference" in out: + try: + parts = urlsplit(out["reference"]) + except ValueError as exc: + raise ReferenceUnreadable("invalid literal reference URI") from exc + if parts.username or parts.password or parts.query or parts.fragment: + raise ReferenceUnreadable( + "reference URI credentials, queries and fragments are unsupported" + ) + if "type" in out and not _RESOURCE.fullmatch(out["type"]): + raise ReferenceUnreadable("unsupported Reference.type") + if "identifier" in value: + identifier = value["identifier"] + if not isinstance(identifier, Mapping) or set(identifier) - {"system", "value"}: + raise ReferenceUnreadable("unsupported Reference.identifier metadata") + if any(not isinstance(v, str) for v in identifier.values()): + raise ReferenceUnreadable("identifier system and value must be strings") + out["identifier"] = dict(identifier) + return out + + +def _key(value: Any, targets: Sequence[str]) -> str: + try: + return json.dumps([reference_value(value), sorted(targets)], sort_keys=True) + except (TypeError, ValueError) as exc: + raise ReferenceUnreadable( + "a Reference contains non-JSON identity data" + ) from exc + + +@dataclass(frozen=True) +class ReferenceEvidence: + """Resolved identities and explicit failures, with no reader attached.""" + + identities: Mapping[str, str] = field(default_factory=dict) + problems: Mapping[str, str] = field(default_factory=dict) + versions: Mapping[str, str] = field(default_factory=dict) + + def identity(self, value: Any, targets: Sequence[str]) -> str: + key = _key(value, targets) + if key not in self.identities: + raise ReferenceUnreadable( + self.problems.get(key, "reference target was not resolved") + ) + return self.identities[key] + + def version(self, value: Any, targets: Sequence[str]) -> str | None: + self.identity(value, targets) + return self.versions.get(_key(value, targets)) + + def search_value(self, value: Any, targets: Sequence[str], base_url: str) -> str: + identity = self.identity(value, targets) + prefix = base_url.rstrip("/") + "/" + if not identity.startswith(prefix): + raise ReferenceUnreadable( + "reference search identity belongs to another base" + ) + relative = identity[len(prefix) :] + parts = relative.split("/") + if ( + len(parts) != 2 + or not _RESOURCE.fullmatch(parts[0]) + or not _ID.fullmatch(parts[1]) + ): + raise ReferenceUnreadable("reference search needs one logical Type/id") + return relative + + +class ReferenceReader: + """Collect target evidence through a FHIR reader's GET and full search.""" + + def __init__( + self, + base_url: str, + fetch: Callable[[str], Mapping[str, Any]], + search: Callable[[str, Mapping[str, str]], Sequence[Mapping[str, Any]]], + ) -> None: + self.base = base_url.rstrip("/") + self.fetch = fetch + self.search = search + self._identities: dict[str, str] = {} + self._problems: dict[str, str] = {} + self._versions: dict[str, str] = {} + + def snapshot(self) -> ReferenceEvidence: + return ReferenceEvidence( + dict(self._identities), dict(self._problems), dict(self._versions) + ) + + def collect(self, value: Any, targets: Sequence[str]) -> None: + try: + self.resolve(value, targets) + except ReferenceUnreadable: + pass # retain the refusal; a pure comparison decides whether it needs it + + def _literal(self, value: Any) -> tuple[str, str, str]: + if not isinstance(value, str) or value != value.strip(): + raise ReferenceUnreadable("a literal reference needs one exact URI") + url = value if urlsplit(value).scheme else self.base + "/" + value + p, b = urlsplit(url), urlsplit(self.base) + if ( + p.scheme not in ("http", "https") + or p.scheme != b.scheme + or p.netloc != b.netloc + or p.username + or p.password + or p.query + or p.fragment + or not p.path.startswith(b.path.rstrip("/") + "/") + ): + raise ReferenceUnreadable("a reference outside the FHIR base is unresolved") + tail = p.path[len(b.path.rstrip("/")) + 1 :].split("/") + if ( + len(tail) not in (2, 4) + or not _RESOURCE.fullmatch(tail[0]) + or not _ID.fullmatch(tail[1]) + ): + raise ReferenceUnreadable("unsupported literal reference identity") + if len(tail) == 4 and (tail[2] != "_history" or not _ID.fullmatch(tail[3])): + raise ReferenceUnreadable("unsupported versioned reference") + # Logical identity names the resource. The requested version stays + # in the read path and separate evidence; it never hides a duplicate. + return self.base + "/" + "/".join(tail[:2]), tail[0], "/".join(tail) + + def resolve(self, value: Any, targets: Sequence[str]) -> str: + key = _key(value, targets) + if key in self._identities: + return self._identities[key] + if key in self._problems: + raise ReferenceUnreadable(self._problems[key]) + try: + identity, version = self._resolve(reference_value(value), targets) + except Exception as exc: # a transport failure cannot become a match + problem = ( + str(exc) + if isinstance(exc, ReferenceUnreadable) + else "reference target read failed" + ) + self._problems[key] = problem + raise ReferenceUnreadable(problem) from exc + self._identities[key] = identity + if version is not None: + self._versions[key] = version + return identity + + def _resolve( + self, ref: Mapping[str, Any], targets: Sequence[str] + ) -> tuple[str, str | None]: + literal = ref.get("reference") + identifier = ref.get("identifier") + declared = ref.get("type") + if declared is not None and ( + not isinstance(declared, str) or not _RESOURCE.fullmatch(declared) + ): + raise ReferenceUnreadable("unsupported Reference.type") + target = None + identity = None + version = None + rtype = declared + if literal is not None: + identity, rtype, path = self._literal(literal) + if declared is not None and declared != rtype: + raise ReferenceUnreadable( + "Reference.type contradicts its literal target" + ) + target = self.fetch(path) + if ( + not isinstance(target, Mapping) + or target.get("resourceType") != rtype + or target.get("id") != path.split("/")[1] + ): + raise ReferenceUnreadable("the target read returned another identity") + if ( + "/_history/" in path + and (target.get("meta") or {}).get("versionId") != path.split("/")[-1] + ): + raise ReferenceUnreadable("the target read returned another version") + if "/_history/" in path: + version = path.split("/")[-1] + permitted = tuple(t for t in targets if t != "Resource") + if rtype is not None and permitted and rtype not in permitted: + raise ReferenceUnreadable("the target type contradicts the schema") + if identifier is not None: + if not isinstance(identifier, Mapping): + raise ReferenceUnreadable("Reference.identifier is not an Identifier") + system, value = identifier.get("system"), identifier.get("value") + if not all( + isinstance(v, str) and v and v == v.strip() for v in (system, value) + ): + raise ReferenceUnreadable( + "an identifier needs its exact system and value" + ) + if any(c in str(system) + str(value) for c in "|,$\\"): + raise ReferenceUnreadable("identifier search escaping is unsupported") + candidates = [rtype] if rtype is not None else list(permitted) + if not candidates: + raise ReferenceUnreadable( + "identifier resolution needs a declared target type" + ) + matches: dict[str, Mapping[str, Any]] = {} + for kind in candidates: + for res in self.search( + kind, {"identifier": f"{system}|{value}", "_count": "200"} + ): + if ( + res.get("resourceType") != kind + or not isinstance(res.get("id"), str) + or not _ID.fullmatch(res["id"]) + ): + raise ReferenceUnreadable( + "identifier search returned an invalid target identity" + ) + ids = res.get("identifier") + if not isinstance(ids, list) or not any( + isinstance(i, Mapping) + and i.get("system") == system + and i.get("value") == value + for i in ids + ): + raise ReferenceUnreadable( + "identifier search returned an unverified identifier" + ) + matches[f"{self.base}/{kind}/{res['id']}"] = res + if len(matches) != 1: + raise ReferenceUnreadable( + "identifier search did not resolve exactly one target" + ) + found = next(iter(matches)) + if identity is not None: + current = identity.split("/_history/", 1)[0] + if current != found: + raise ReferenceUnreadable( + "Reference.reference and identifier contradict" + ) + ids = target.get("identifier") if target is not None else None + if not isinstance(ids, list) or not any( + isinstance(i, Mapping) + and i.get("system") == system + and i.get("value") == value + for i in ids + ): + raise ReferenceUnreadable( + "the literal target does not hold its claimed identifier" + ) + else: + identity = found + if identity is None: + raise ReferenceUnreadable( + "a display alone does not identify a reference target" + ) + return identity, version diff --git a/packages/flow/openadapt_flow/generate/ensure_contract.py b/packages/flow/openadapt_flow/generate/ensure_contract.py index 4ecabfbbaf..89fba55bfd 100644 --- a/packages/flow/openadapt_flow/generate/ensure_contract.py +++ b/packages/flow/openadapt_flow/generate/ensure_contract.py @@ -50,10 +50,19 @@ EnsureContract, FieldRule, extra_elements, + extract_all, is_person, normalize, + reference_display, + reference_field, wrong_fields, ) +from openadapt_flow.fhir_reference import ( + ReferenceReader, + ReferenceSchema, + ReferenceUnreadable, + definition_digest, +) from openadapt_flow.generate import fhir_tables as tables SCHEMA = "openadapt.ensure-record/v1" @@ -96,6 +105,8 @@ class Teach: facts: Mapping[str, Any] deployment: Mapping[str, Any] confirmed: set[str] = field(default_factory=set) + structure_definition: Optional[Mapping[str, Any]] = None + reference_targets: Sequence[Mapping[str, Any]] = () def _resources(doc: Any) -> list[dict[str, Any]]: @@ -163,6 +174,12 @@ def load_teach(root: str | Path, confirmed: Sequence[str] = ()) -> Teach: # A confirmation is a person's, given to the command, never read # from the teach data. confirmed=set(confirmed), + structure_definition=_load(base / "structure_definition.json") + if (base / "structure_definition.json").exists() + else None, + reference_targets=_resources(_load(base / "reference_targets.json")) + if (base / "reference_targets.json").exists() + else [], ) @@ -486,11 +503,60 @@ def _change_feed( } +def _reference_reader(teach: Teach) -> ReferenceReader: + """Captured targets admit the demonstrations only; runtime reads anew.""" + + def fetch(path: str) -> Mapping[str, Any]: + parts = path.split("/") + found = [ + r + for r in teach.reference_targets + if r.get("resourceType") == parts[0] + and r.get("id") == parts[1] + and (len(parts) == 2 or (r.get("meta") or {}).get("versionId") == parts[3]) + ] + if len(found) != 1: + raise ReferenceUnreadable( + "demonstration target was not captured exactly once" + ) + return found[0] + + def search(kind: str, query: Mapping[str, str]) -> Sequence[Mapping[str, Any]]: + system, value = query["identifier"].split("|", 1) + return [ + r + for r in teach.reference_targets + if r.get("resourceType") == kind + and any( + isinstance(i, Mapping) + and i.get("system") == system + and i.get("value") == value + for i in r.get("identifier", []) + ) + ] + + return ReferenceReader(str(teach.deployment.get("base_url") or ""), fetch, search) + + def generate(teach: Teach) -> Generated: rtype = teach.resource_type refusals: list[str] = [] to_confirm: list[str] = [] prov: dict[str, str] = {} + schema = None + if teach.structure_definition is not None: + try: + schema = ReferenceSchema( + definition=dict(teach.structure_definition), + sha256=definition_digest(teach.structure_definition), + ) + if schema.definition["type"] != rtype: + raise ValueError("the schema describes another resource type") + schema.paths() + except ValueError as exc: + return Generated(None, [f"Reference schema: {exc}"], [], {}, []) + typed = schema.paths() if schema else {} + references = _reference_reader(teach) made: list[tuple[Demo, dict[str, Any]]] = [] for demo in teach.demos: @@ -541,7 +607,17 @@ def generate(teach: Teach) -> Generated: compare=kind, # type: ignore[arg-type] map=map_for(s.plain, values_at[s.path]) or {}, ) - if _matches(s.value, wanted, rule): + parent = s.plain.removesuffix(".reference") + if parent in typed: + try: + matched = references.resolve( + s.value, typed[parent] + ) == references.resolve(wanted, typed[parent]) + except ReferenceUnreadable: + matched = False + else: + matched = _matches(s.value, wanted, rule) + if matched: if s.ambiguous: ambiguous_hit = True elif s.group is None: @@ -587,6 +663,16 @@ def generate(teach: Teach) -> Generated: sample.group and any(b.startswith(sample.group + ".") for b in bound) ): continue + parent = path.removesuffix(".display") + if ( + path.endswith(".display") + and sample.plain.removesuffix(".display") in typed + and parent + ".reference" in bound + ): + prov[f"presentation.{path}"] = ( + "FHIR Reference.display; the same Reference has a bound target identity" + ) + continue vals = values_at[path] present_in = sum(1 for ss in per_demo if any(s.path == path for s in ss)) if sample.ambiguous: @@ -680,6 +766,9 @@ def generate(teach: Teach) -> Generated: "resource_type": rtype, "base_url": teach.deployment.get("base_url"), } + if schema is not None: + contract["reference_schema"] = schema.model_dump() + prov["reference_schema"] = "the hash-pinned StructureDefinition snapshot" if teach.deployment.get("access_token_env"): contract["access_token_env"] = teach.deployment["access_token_env"] contract["self"] = list(teach.deployment.get("self") or []) @@ -698,7 +787,7 @@ def generate(teach: Teach) -> Generated: except Exception as exc: # pragma: no cover - a generator bug refusals.append(f"the generated contract does not load: {exc}") else: - refusals += self_check(model, made, checks) + refusals += self_check(model, made, checks, references) return Generated(contract, refusals, to_confirm, prov, checks) @@ -756,28 +845,47 @@ def self_check( contract: EnsureContract, made: Sequence[tuple[Demo, dict[str, Any]]], checks: list[str], + references: Optional[ReferenceReader] = None, ) -> list[str]: """The contract passes every demonstration and catches every perturbation.""" problems: list[str] = [] for demo, resource in made: params = {k: v for k, v in demo.params.items() if k in contract.fields} - wrong = wrong_fields(contract, resource, params) + if references is not None: + for name, field in contract.fields.items(): + ref = reference_field(contract, field.path) + if ref is not None: + if name in params: + references.collect(params[name], ref[1]) + for value in extract_all(resource, ref[0]): + references.collect(value, ref[1]) + evidence = references.snapshot() if references else None + try: + wrong = wrong_fields(contract, resource, params, evidence) + except ReferenceUnreadable as exc: + problems.append(f"{demo.name}: {exc}") + continue if wrong: problems.append(f"{demo.name} reads as wrong: {wrong}") - if not is_person(contract, resource, params): + if not is_person(contract, resource, params, evidence): problems.append(f"{demo.name} is not the one its identity rules name") extra = extra_elements(contract, resource, params) if extra: problems.append(f"{demo.name} holds elements the contract doesn't: {extra}") for name, rule in contract.fields.items(): - if params.get(name) in (None, ""): + if params.get(name) in (None, "") or reference_display(contract, rule.path): continue bent = copy.deepcopy(resource) if not _set_first(bent, rule.path, lambda v, r=rule: _perturb(v, r)): problems.append(f"{demo.name}: can't perturb {name}") - elif name not in wrong_fields(contract, bent, params): - problems.append(f"{demo.name}: a changed {name} reads as right") + else: + try: + missed = name not in wrong_fields(contract, bent, params, evidence) + except ReferenceUnreadable: + missed = False # an unresolved changed target cannot confirm + if missed: + problems.append(f"{demo.name}: a changed {name} reads as right") bent = copy.deepcopy(resource) bent["extension"] = [{"url": "urn:openadapt:probe", "valueString": "x"}] if not extra_elements(contract, bent, params): @@ -785,7 +893,7 @@ def self_check( if not problems: checks.append( f"{len(made)} demonstrations read back right and closed; " - f"{len(contract.fields)} field perturbations and one extra element " + f"{sum(not reference_display(contract, f.path) for f in contract.fields.values())} field perturbations and one extra element " "caught in each" ) return problems diff --git a/packages/flow/openadapt_flow/shadow.py b/packages/flow/openadapt_flow/shadow.py index 90a12db1d0..866b85f611 100644 --- a/packages/flow/openadapt_flow/shadow.py +++ b/packages/flow/openadapt_flow/shadow.py @@ -425,7 +425,8 @@ def _unchecked(self, name: str, why: str) -> Episode: return ep def _fhir_episode(self, method: str, path: str, doc: Any) -> Episode: - from openadapt_flow.ensure_record import extract, normalize + from openadapt_flow.ensure_record import extract, normalize, reference_field + from openadapt_flow.fhir_reference import reference_value segs = [s for s in urlsplit(path).path.split("/") if s] if any(s == "_search" or s.startswith("$") for s in segs): @@ -446,7 +447,13 @@ def _fhir_episode(self, method: str, path: str, doc: Any) -> Episode: if isinstance(doc, dict) and doc.get("resourceType") == rtype: for name, rule in contract.fields.items(): value = extract(doc, rule.path) - if value is not None and not isinstance(value, (dict, list)): + if value is not None and ( + not isinstance(value, (dict, list)) + or ( + reference_field(contract, rule.path) is not None + and isinstance(value, dict) + ) + ): params[name] = value done = self._by_ref.get(ref) if ref else None if done is not None and done.checked: @@ -462,7 +469,15 @@ def _fhir_episode(self, method: str, path: str, doc: Any) -> Episode: key = ( rtype, tuple( - (p, normalize(params[p], contract.fields[p], task=True)) + ( + p, + json.dumps(reference_value(params[p]), sort_keys=True) + if reference_field(contract, contract.fields[p].path) + is not None + else normalize( + params[p], contract.fields[p], task=True + ), + ) for p in rule ), ) @@ -773,6 +788,7 @@ def _recheck_fhir(self, ep: Episode) -> None: verify_ensure, wrong_fields, ) + from openadapt_flow.fhir_reference import ReferenceUnreadable contract = ep.contract rid = str(ep.result.record_id or "") @@ -786,7 +802,13 @@ def _recheck_fhir(self, ep: Episode) -> None: reader = FhirReader(contract) try: _, now = reader.fetch(f"{reader.base_url}/{ref}") - except Unreadable as exc: + references = reader.reference_evidence( + contract, ep.params, [now] if isinstance(now, dict) else [] + ) + # Resolve all supplied and stored identities before a late verdict. + if isinstance(now, dict): + wrong_fields(contract, now, ep.params, references) + except (Unreadable, ReferenceUnreadable) as exc: gone = re.search(r"returned HTTP (404|410)\b", str(exc)) ep.late_moved = bool(gone) ep.late_verdict = REFUTED if gone else UNCERTAIN @@ -797,10 +819,12 @@ def _recheck_fhir(self, ep: Episode) -> None: ] return moved: list[str] = [] - if not isinstance(now, dict) or not is_person(contract, now, ep.params): + if not isinstance(now, dict) or not is_person( + contract, now, ep.params, references + ): moved.append("the confirmed record is no longer the person") else: - wrong = wrong_fields(contract, now, ep.params) + wrong = wrong_fields(contract, now, ep.params, references) if wrong: moved.append( "the confirmed record now holds other values: " + ", ".join(wrong) @@ -871,9 +895,10 @@ def _check_fhir(self, ep: Episode, base: Any, statuses: list[int]) -> None: Unreadable, extra_elements, extract, - same, verify_ensure, + wrong_fields, ) + from openadapt_flow.fhir_reference import ReferenceUnreadable contract = ep.contract with self._lock: @@ -937,6 +962,11 @@ def _check_fhir(self, ep: Episode, base: Any, statuses: list[int]) -> None: for ref, res in stored.items(): if not isinstance(res, dict) or "unreadable" in res: continue + references = reader.reference_evidence(contract, ep.params, [res]) + try: + mismatches = wrong_fields(contract, res, ep.params, references) + except ReferenceUnreadable: + mismatches = list(contract.fields) for name, rule in contract.fields.items(): got = extract(res, rule.path) sent = ep.params.get(name) @@ -948,8 +978,7 @@ def _check_fhir(self, ep: Episode, base: Any, statuses: list[int]) -> None: "field": name, "sent": sent, "stored": got, - "same": name in ep.params - and same(contract, name, got, ep.params[name]), + "same": name in ep.params and name not in mismatches, } ) if contract.closed: diff --git a/packages/flow/tests/test_fhir_reference.py b/packages/flow/tests/test_fhir_reference.py new file mode 100644 index 0000000000..7c6af7370c --- /dev/null +++ b/packages/flow/tests/test_fhir_reference.py @@ -0,0 +1,748 @@ +"""Schema-proven Reference identities; every record and endpoint is synthetic.""" + +from __future__ import annotations + +import copy +from datetime import datetime, timezone + +import pytest + +from openadapt_flow.ensure_record import ( + Baseline, + Change, + EnsureContract, + FhirReader, + judge, +) +from openadapt_flow.fhir_reference import ReferenceSchema, definition_digest + +BASE = "https://records.example.test/fhir/R4" +BOT = "Practitioner/verifier-agent" + + +def schema(resource="Observation", path="subject"): + definition = { + "resourceType": "StructureDefinition", + "kind": "resource", + "url": f"http://hl7.org/fhir/StructureDefinition/{resource}", + "type": resource, + "snapshot": { + "element": [ + {"path": resource}, + { + "path": f"{resource}.{path}", + "type": [ + { + "code": "Reference", + "targetProfile": [ + "http://hl7.org/fhir/StructureDefinition/Patient" + ], + } + ], + }, + ] + }, + } + return {"definition": definition, "sha256": definition_digest(definition)} + + +def contract(**over): + data = { + "resource_type": "Observation", + "base_url": BASE, + "self": [BOT], + "reference_schema": schema(), + "closed": True, + "fields": { + "patient": {"path": "subject.reference"}, + "display": {"path": "subject.display"}, + "note": {"path": "valueString"}, + }, + "identity": [["patient", "note"]], + "search": {"patient": "patient"}, + } + data.update(over) + return EnsureContract.model_validate(data) + + +def record(**over): + data = { + "resourceType": "Observation", + "id": "o", + "subject": {"reference": "Patient/a", "display": "Store's presentation"}, + "valueString": "approved", + } + data.update(over) + return data + + +def parameters(**over): + data = { + "patient": "Patient/a", + "display": "Task's presentation", + "note": "approved", + } + data.update(over) + return data + + +class Store: + def __init__(self): + self.records = { + "Patient/a": { + "resourceType": "Patient", + "id": "a", + "identifier": [{"system": "urn:synthetic:registry", "value": "A"}], + }, + "Patient/b": { + "resourceType": "Patient", + "id": "b", + "identifier": [{"system": "urn:synthetic:registry", "value": "B"}], + }, + } + self.calls = [] + self.status = 200 + self.total = None + + def get(self, url, params=None): + self.calls.append((url, params)) + assert url.startswith(BASE + "/"), ( + "never send a read to an external origin/base" + ) + path = url[len(BASE) + 1 :] + if self.status != 200: + return self.status, {}, {} + if params and "identifier" in params: + system, value = params["identifier"].split("|", 1) + rows = [ + r + for r in self.records.values() + if r["resourceType"] == path + and any( + i["system"] == system and i["value"] == value + for i in r.get("identifier", []) + ) + ] + return ( + 200, + {}, + { + "resourceType": "Bundle", + "type": "searchset", + "total": self.total if self.total is not None else len(rows), + "entry": [{"resource": r} for r in rows], + }, + ) + return (200, {}, self.records[path]) if path in self.records else (404, {}, {}) + + +def verdict(c, params, resource, store): + reader = FhirReader(c, get=store.get) + evidence = reader.reference_evidence(c, params, [resource]) + calls = list(store.calls) + result = judge( + c, + params, + Baseline(started=datetime.now(timezone.utc), matches={}), + {"Observation/o": Change(ref="Observation/o", actions={"C"}, authors={BOT})}, + {"o": resource}, + resource, + references=evidence, + ) + assert store.calls == calls, "pure judging must not perform reads" + return result + + +def test_only_reference_display_differs_confirmed(): + assert ( + verdict(contract(), parameters(), record(), Store()).outcome + == "created_verified" + ) + + +def test_wrong_target_with_matching_display_refuted(): + result = verdict( + contract(), + parameters(), + record(subject={"reference": "Patient/b", "display": "Task's presentation"}), + Store(), + ) + assert result.outcome == "needs_review" and not result.evidence.get("unreadable") + assert any("wrong or missing: patient" in r for r in result.reasons) + + +@pytest.mark.parametrize( + "subject", + [ + {"reference": "Patient/a"}, + {"reference": BASE + "/Patient/a", "display": "anything"}, + ], +) +def test_display_is_optional_and_relative_absolute_target_identity_agrees(subject): + assert ( + verdict(contract(), parameters(), record(subject=subject), Store()).outcome + == "created_verified" + ) + + +def test_identifier_only_resolves_to_same_literal_target(): + wanted = { + "type": "Patient", + "identifier": {"system": "urn:synthetic:registry", "value": "A"}, + } + assert ( + verdict(contract(), parameters(patient=wanted), record(), Store()).outcome + == "created_verified" + ) + + assert ( + verdict( + contract(), + parameters(), + record(subject={"identifier": wanted["identifier"]}), + Store(), + ).outcome + == "created_verified" + ) + + +def test_literal_and_identifier_must_agree(): + subject = { + "reference": "Patient/a", + "identifier": {"system": "urn:synthetic:registry", "value": "B"}, + "display": "Task's presentation", + } + result = verdict(contract(), parameters(), record(subject=subject), Store()) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + assert "contradict" in result.reasons[0] + + +def test_consistent_literal_identifier_is_verified(): + subject = { + "reference": "Patient/a", + "identifier": {"system": "urn:synthetic:registry", "value": "A"}, + } + assert ( + verdict(contract(), parameters(), record(subject=subject), Store()).outcome + == "created_verified" + ) + + +@pytest.mark.parametrize( + "subject", + [ + {"reference": "https://other.example.test/fhir/R4/Patient/a"}, + {"reference": "https://records.example.test/another/Patient/a"}, + {"reference": "Patient/a", "type": "Practitioner"}, + {"reference": "Patient/missing"}, + {"reference": "#contained"}, + {"reference": "urn:uuid:synthetic"}, + {"reference": "Patient/a?token=x"}, + {"reference": "Patient/a/_history/old"}, + {"display": "Task's presentation"}, + {"identifier": {"value": "A"}}, + ], +) +def test_unresolved_or_contradictory_identity_never_confirms(subject): + result = verdict(contract(), parameters(), record(subject=subject), Store()) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_same_id_in_another_resource_type_is_not_same_target(): + store = Store() + store.records["Practitioner/a"] = {"resourceType": "Practitioner", "id": "a"} + result = verdict( + contract(), parameters(), record(subject={"reference": "Practitioner/a"}), store + ) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_identifier_duplicate_is_uncertain_even_when_one_literal_matches(): + store = Store() + store.records["Patient/b"]["identifier"] = copy.deepcopy( + store.records["Patient/a"]["identifier"] + ) + result = verdict( + contract(), + parameters( + patient={"identifier": {"system": "urn:synthetic:registry", "value": "A"}} + ), + record(), + store, + ) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_identifier_partial_search_is_uncertain(): + store = Store() + store.total = 2 + result = verdict( + contract(), + parameters( + patient={"identifier": {"system": "urn:synthetic:registry", "value": "A"}} + ), + record(), + store, + ) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +@pytest.mark.parametrize("status", [401, 403, 500]) +def test_failed_target_read_is_uncertain(status): + store = Store() + store.status = status + result = verdict(contract(), parameters(), record(), store) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_target_read_must_echo_the_requested_identity(): + store = Store() + store.records["Patient/a"]["id"] = "b" + result = verdict(contract(), parameters(), record(), store) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_unknown_metadata_does_not_relax_display(): + c = contract(reference_schema=None) + result = verdict(c, parameters(), record(), Store()) + assert result.outcome == "needs_review" + assert any("display" in r for r in result.reasons) + assert "reference_schema" not in c.model_dump() + + +def test_unrelated_field_named_display_remains_exact(): + c = contract( + fields={ + "patient": {"path": "subject.reference"}, + "display": {"path": "display"}, + "note": {"path": "valueString"}, + } + ) + result = verdict(c, parameters(), record(display="Different scalar"), Store()) + assert result.outcome == "needs_review" and not result.evidence.get("unreadable") + + +def test_display_parameter_without_target_parameter_is_uncertain(): + params = parameters() + del params["patient"] + assert verdict(contract(), params, record(), Store()).evidence["unreadable"] + + +@pytest.mark.parametrize( + "subject", + [ + {"reference": "Patient/a", "display": ["one", "two"]}, + { + "reference": "Patient/a", + "extension": [{"url": "urn:synthetic:unapproved", "valueString": "x"}], + }, + ], +) +def test_reference_does_not_open_closed_record_members(subject): + result = verdict(contract(), parameters(), record(subject=subject), Store()) + assert result.outcome == "needs_review" and any( + "not given by the task" in r for r in result.reasons + ) + + +def test_missing_resolution_evidence_is_uncertain(): + r = record() + result = judge( + contract(), + parameters(), + Baseline(started=datetime.now(timezone.utc), matches={}), + {"Observation/o": Change(ref="Observation/o", actions={"C"}, authors={BOT})}, + {"o": r}, + r, + ) + assert result.evidence["unreadable"] + + +def test_digest_and_type_metadata_cannot_be_edited_silently(): + pin = schema() + pin["definition"]["snapshot"]["element"][1]["type"][0]["code"] = "string" + with pytest.raises(ValueError, match="digest"): + ReferenceSchema.model_validate(pin) + with pytest.raises(ValueError, match="another resource"): + contract(reference_schema=schema("Encounter")) + + +def test_display_cannot_be_the_record_identity(): + with pytest.raises(ValueError, match="display cannot"): + contract(identity=[["display"]], search={"display": "patient"}) + + +def test_metadata_without_an_independent_identity_binding_refuses(): + with pytest.raises(ValueError, match="bound target identity"): + contract( + fields={"display": {"path": "subject.display"}}, + identity=[["display"]], + search={"display": "patient"}, + ) + + +def test_legacy_contract_canonical_digest_and_round_trip_are_unchanged(): + import hashlib + import json + + doc = { + "resource_type": "Patient", + "base_url": BASE, + "self": [BOT], + "fields": {"name": {"path": "name.family"}}, + "identity": [["name"]], + "search": {"name": "family"}, + } + c = EnsureContract.model_validate(doc) + raw = c.model_dump(mode="json", by_alias=True) + # Recorded from the unmodified model at 800f450047c3195bb28a5e6b6b8f862a5e21a16f. + digest = hashlib.sha256( + json.dumps(raw, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert digest == "5847c5c359a0d8836f3d5b7c9e63c0cee3c088ee46646b833a915daaa3ff7c30" + assert "reference_schema" not in raw + assert ( + EnsureContract.model_validate(raw).model_dump(mode="json", by_alias=True) == raw + ) + + +def test_reference_evidence_does_not_carry_unrelated_target_fields(): + store = Store() + store.records["Patient/a"]["unrelated"] = "SYNTHETIC-UNRELATED-TARGET-DATA" + c = contract() + evidence = FhirReader(c, get=store.get).reference_evidence( + c, parameters(), [record()] + ) + assert "SYNTHETIC-UNRELATED-TARGET-DATA" not in repr(evidence) + assert verdict(c, parameters(), record(), store).outcome == "created_verified" + + +def test_unsupported_identifier_metadata_stays_uncertain_and_out_of_evidence(): + wanted = { + "identifier": { + "system": "urn:synthetic:registry", + "value": "A", + "assigner": {"display": "SYNTHETIC-UNRELATED-ASSIGNER"}, + } + } + c = contract() + evidence = FhirReader(c, get=Store().get).reference_evidence( + c, parameters(patient=wanted), [record()] + ) + assert "SYNTHETIC-UNRELATED-ASSIGNER" not in repr(evidence) + assert verdict(c, parameters(patient=wanted), record(), Store()).evidence[ + "unreadable" + ] + + +def test_reference_uri_credentials_and_queries_are_not_persisted(): + c = contract() + for uri in ( + "https://synthetic-user:PRIVATE-MARKER@records.example.test/fhir/R4/Patient/a", + "Patient/a?token=PRIVATE-MARKER", + ): + resource = record(subject={"reference": uri}) + evidence = FhirReader(c, get=Store().get).reference_evidence( + c, parameters(), [resource] + ) + assert "PRIVATE-MARKER" not in repr(evidence) + assert verdict(c, parameters(), resource, Store()).evidence["unreadable"] + + +@pytest.mark.parametrize( + "earlier_target,expected", + [("Patient/a", "created_verified"), ("Patient/b", "needs_review")], +) +def test_reference_identity_remains_checked_in_every_corrected_version( + earlier_target, expected +): + from datetime import timedelta + + c = contract(settle_s=1, operations={"correction": {"between": "consistent"}}) + params = parameters() + first = record( + subject={"reference": earlier_target, "display": "Earlier presentation"} + ) + final = record() + store = Store() + evidence = FhirReader(c, get=store.get).reference_evidence( + c, params, [first, final] + ) + now = datetime.now(timezone.utc) + first["meta"] = {"versionId": "1", "lastUpdated": now.isoformat()} + final["meta"] = { + "versionId": "2", + "lastUpdated": (now + timedelta(seconds=1)).isoformat(), + } + change = Change( + ref="Observation/o", + actions={"C", "U"}, + authors={BOT}, + versions=2, + bodies=[first, final], + events=[ + (now, "C", frozenset({BOT})), + (now + timedelta(seconds=1), "U", frozenset({BOT})), + ], + ) + result = judge( + c, + params, + Baseline(started=now, matches={}), + {"Observation/o": change}, + {"o": final}, + final, + references=evidence, + ) + assert result.outcome == expected + assert not result.evidence.get("unreadable") + + +@pytest.mark.parametrize( + "target,read_version,expected,reason", + [ + ("Patient/a", "2", "created_verified", None), + ("Patient/a", "1", "needs_review", "not the last version"), + ("Patient/b", "2", "needs_review", "wrong or missing: patient"), + ("Patient/a", None, "needs_review", "no versionId"), + ], +) +def test_reference_identity_and_last_written_version_are_both_required( + target, read_version, expected, reason +): + now = datetime.now(timezone.utc) + written = record( + subject={"reference": target, "display": "Task's presentation"}, + meta={"versionId": "2", "lastUpdated": now.isoformat()}, + ) + read_back = copy.deepcopy(written) + if read_version is None: + del read_back["meta"]["versionId"] + else: + read_back["meta"]["versionId"] = read_version + c = contract() + params = parameters() + store = Store() + evidence = FhirReader(c, get=store.get).reference_evidence( + c, params, [written, read_back] + ) + calls = list(store.calls) + change = Change( + ref="Observation/o", + actions={"C"}, + authors={BOT}, + versions=1, + bodies=[written], + events=[(now, "C", frozenset({BOT}))], + ) + result = judge( + c, + params, + Baseline(started=now, matches={}), + {"Observation/o": change}, + {"o": read_back}, + read_back, + references=evidence, + ) + assert result.outcome == expected + assert not result.evidence.get("unreadable") + assert store.calls == calls, "pure judging must not perform reads" + if reason: + assert any(reason in item for item in result.reasons) + else: + assert result.evidence["versions"] == [{"version": "2", "author": BOT}] + + +class SearchStore(Store): + def __init__(self, resources=(), *, strict=False): + super().__init__() + self.resources = list(resources) + self.strict = strict + self.queries = [] + for version in ("1", "2"): + target = copy.deepcopy(self.records["Patient/a"]) + target["meta"] = {"versionId": version} + self.records[f"Patient/a/_history/{version}"] = target + + def get(self, url, params=None): + if url == BASE + "/Observation": + self.queries.append(dict(params or {})) + rows = self.resources + if self.strict and (params or {}).get("patient") != "Patient/a": + rows = [] + return ( + 200, + {}, + { + "resourceType": "Bundle", + "type": "searchset", + "total": len(rows), + "entry": [{"resource": r} for r in rows], + }, + ) + return super().get(url, params) + + +@pytest.mark.parametrize( + "wanted,stored", + [("Patient/a", "Patient/a/_history/1"), ("Patient/a/_history/1", "Patient/a")], +) +def test_versioned_reference_cannot_hide_existing_record_or_allow_duplicate( + wanted, stored +): + from openadapt_flow.ensure_record import find_matches + + c = contract() + params = parameters(patient=wanted) + existing = record(subject={"reference": stored, "display": "Other presentation"}) + store = SearchStore([existing]) + reader = FhirReader(c, get=store.get) + matches, candidates = find_matches(c, reader, params) + assert candidates == 1 and set(matches) == {"o"} + baseline = Baseline(started=datetime.now(timezone.utc), matches=matches) + created = record(id="new", subject={"reference": wanted}) + store.resources.append(created) + after, _ = find_matches(c, reader, params) + assert set(after) == {"o", "new"} + evidence = reader.reference_evidence(c, params, list(after.values())) + result = judge( + c, + params, + baseline, + { + "Observation/new": Change( + ref="Observation/new", actions={"C"}, authors={BOT} + ) + }, + after, + created, + references=evidence, + ) + assert result.outcome == "needs_review" + assert result.evidence["before_matches"] == 1 + assert result.evidence["after_matches"] == 2 + assert not result.evidence.get("unreadable") + + +@pytest.mark.parametrize( + "wanted", + [ + "Patient/a", + BASE + "/Patient/a", + "Patient/a/_history/1", + BASE + "/Patient/a/_history/1", + ], +) +def test_reference_search_uses_local_logical_type_id(wanted): + from openadapt_flow.ensure_record import find_matches + + c = contract() + store = SearchStore([record()], strict=True) + matches, count = find_matches( + c, FhirReader(c, get=store.get), parameters(patient=wanted) + ) + assert set(matches) == {"o"} and count == 1 + assert store.queries == [{"patient": "Patient/a", "_count": "200"}] + + +def test_missing_expected_reference_search_match_is_unreadable(): + from openadapt_flow.ensure_record import Unreadable, find_matches + + c = contract() + reader = FhirReader(c, get=SearchStore().get) + with pytest.raises(Unreadable, match="known matching record"): + find_matches(c, reader, parameters(), expected_ids=["o"]) + + +def test_initial_reference_search_may_be_empty_without_expected_record(): + from openadapt_flow.ensure_record import find_matches + + c = contract() + assert find_matches(c, FhirReader(c, get=SearchStore().get), parameters()) == ( + {}, + 0, + ) + + +def test_logical_reference_identity_retains_verified_version_separately(): + c = contract() + params = parameters(patient="Patient/a/_history/1") + evidence = FhirReader(c, get=SearchStore().get).reference_evidence( + c, params, [record()] + ) + assert evidence.identity(params["patient"], ("Patient",)) == BASE + "/Patient/a" + assert evidence.identity("Patient/a", ("Patient",)) == BASE + "/Patient/a" + assert evidence.version(params["patient"], ("Patient",)) == "1" + assert evidence.version("Patient/a", ("Patient",)) is None + + +def test_requested_reference_version_is_still_checked_outside_person_identity(): + from openadapt_flow.ensure_record import is_person, wrong_fields + + c = contract() + params = parameters(patient="Patient/a/_history/1") + resource = record(subject={"reference": "Patient/a/_history/2"}) + evidence = FhirReader(c, get=SearchStore().get).reference_evidence( + c, params, [resource] + ) + assert is_person(c, resource, params, evidence) + assert wrong_fields(c, resource, params, evidence) == ["patient"] + + +def test_version_target_read_must_still_echo_requested_version(): + store = SearchStore() + store.records["Patient/a/_history/1"]["meta"]["versionId"] = "2" + result = verdict( + contract(), parameters(patient="Patient/a/_history/1"), record(), store + ) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + + +def test_closed_typed_reference_does_not_certify_its_display_text(): + resource = record( + subject={"reference": "Patient/a", "display": "An unrelated person's label"} + ) + assert ( + verdict(contract(), parameters(), resource, Store()).outcome + == "created_verified" + ) + + +def test_matching_display_does_not_make_a_different_target_an_existing_person(): + from openadapt_flow.ensure_record import find_matches + + c = contract() + resource = record( + subject={"reference": "Patient/b", "display": "Task's presentation"} + ) + store = SearchStore([resource]) + assert find_matches(c, FhirReader(c, get=store.get), parameters()) == ({}, 1) + + +@pytest.mark.parametrize("created", [False, True]) +def test_verification_refuses_search_omission_of_fresh_known_match( + monkeypatch, created +): + from openadapt_flow import ensure_record + + c = contract(change_feed={"audit": None, "sweep_types": ["Observation"]}) + resource = record() + store = SearchStore() + store.records["Observation/o"] = resource + now = datetime.now(timezone.utc) + changes = ( + {"Observation/o": Change(ref="Observation/o", actions={"C"}, authors={BOT})} + if created + else {} + ) + monkeypatch.setattr(ensure_record, "read_changes", lambda *args: changes) + monkeypatch.setattr(FhirReader, "server_time", lambda self: now) + result = ensure_record.verify_ensure( + c, + parameters(), + Baseline(started=now, matches={} if created else {"o": resource}), + reader=FhirReader(c, get=store.get), + sleep=lambda seconds: None, + ) + assert result.outcome == "needs_review" and result.evidence["unreadable"] + assert any("known matching record" in reason for reason in result.reasons) diff --git a/packages/flow/tests/test_generate_ensure_contract.py b/packages/flow/tests/test_generate_ensure_contract.py index ed32381f62..513caef5af 100644 --- a/packages/flow/tests/test_generate_ensure_contract.py +++ b/packages/flow/tests/test_generate_ensure_contract.py @@ -338,3 +338,88 @@ def test_a_date_looking_string_outside_a_date_element_compares_exactly(teach): out = generate(teach) assert out.admitted, out.refusals assert "compare" not in out.contract["fields"]["directive_date"] + + +def _reference_teach(*, with_schema=True): + from tests.test_fhir_reference import BASE, schema + + t = _allergy_teach() + t.deployment = dict(t.deployment, base_url=BASE) + t.confirmed = {"identity"} + t.reference_targets = [ + {"resourceType": "Patient", "id": f"p{i + 1}"} for i in range(len(t.demos)) + ] + if with_schema: + t.structure_definition = schema("AllergyIntolerance", "patient")["definition"] + for i, d in enumerate(t.demos): + d.params["presentation"] = f"Task text {i}" + d.after[0]["patient"]["display"] = d.params["presentation"] + return t + + +def test_typed_reference_display_is_non_identity_and_other_fields_stay_strict(): + from openadapt_flow.ensure_record import FhirReader, wrong_fields + from tests.test_fhir_reference import Store + + t = _reference_teach() + result = generate(t) + assert result.admitted, result.refusals + c = EnsureContract.model_validate(result.contract) + store = Store() + store.records = {f"Patient/p{i + 1}": r for i, r in enumerate(t.reference_targets)} + + # Use the fixture's synthetic base, not the Store helper's default base. + def get(url, params=None): + key = url[len(c.base_url.rstrip("/")) + 1 :] + return 200, {}, store.records[key] + + d = t.demos[0] + res = copy.deepcopy(d.after[0]) + res["patient"]["display"] = "Different rendering" + evidence = FhirReader(c, get=get).reference_evidence(c, d.params, [res]) + assert wrong_fields(c, res, d.params, evidence) == [] + res["patient"]["reference"] = "Patient/p2" + res["patient"]["display"] = d.params["presentation"] + evidence = FhirReader(c, get=get).reference_evidence(c, d.params, [res]) + assert "patient" in wrong_fields(c, res, d.params, evidence) + res["code"]["text"] = "Wrong scalar" + assert "substance" in wrong_fields(c, res, d.params, evidence) + + +def test_reference_like_names_without_schema_keep_display_strict(): + t = _reference_teach(with_schema=False) + result = generate(t) + assert result.admitted + c = EnsureContract.model_validate(result.contract) + from openadapt_flow.ensure_record import wrong_fields + + res = copy.deepcopy(t.demos[0].after[0]) + res["patient"]["display"] = "Different rendering" + assert "presentation" in wrong_fields(c, res, t.demos[0].params) + + +def test_typed_reference_without_captured_target_refuses_generation(): + t = _reference_teach() + t.reference_targets = [] + result = generate(t) + assert not result.admitted + assert any("patient" in reason for reason in result.refusals) + + +def test_unbound_typed_reference_display_does_not_become_a_constant_requirement(): + t = _reference_teach() + for d in t.demos: + del d.params["presentation"] + result = generate(t) + assert result.admitted, result.refusals + assert not any( + entry.startswith("patient.display") + for entry in result.contract.get("allow", []) + ) + + +def test_typed_display_alone_cannot_replace_a_target_binding(): + t = _reference_teach() + for d in t.demos: + del d.params["patient"] + assert not generate(t).admitted diff --git a/packages/flow/tests/test_shadow_fhir.py b/packages/flow/tests/test_shadow_fhir.py index 00afd19636..e1f77cb28c 100644 --- a/packages/flow/tests/test_shadow_fhir.py +++ b/packages/flow/tests/test_shadow_fhir.py @@ -410,3 +410,113 @@ def test_a_lost_retry_in_a_known_episode_makes_a_late_refutation_uncertain( env["calls"]["ev"] = {"agent_writes": 1} sh.run_rechecks(now=True, only=[sh.episodes[0]]) assert sh.episodes[0].late_verdict == "UNCERTAIN" + + +@pytest.mark.parametrize( + "target,expected", + [ + ("Patient/p1", "CONFIRMED"), + ("Patient/p2", "REFUTED"), + ("Patient/missing", "UNCERTAIN"), + ], +) +def test_reference_identity_and_display_use_the_same_late_path( + env, monkeypatch, target, expected +): + from openadapt_flow.fhir_reference import ReferenceSchema + from tests.test_fhir_reference import schema + + stored = _confirmed(env, monkeypatch) + ep = env["sh"].episodes[0] + c = ep.contract + c.reference_schema = ReferenceSchema.model_validate( + schema("AllergyIntolerance", "patient") + ) + # The existing synthetic contract binds the patient reference and scalar + # clinical values. A stored display is presentation even when unbound. + stored["now"] = _stored(target) + stored["now"]["patient"]["display"] = "A different presentation" + + def fetch(self, url, params=None): + if "/Patient/" in url: + pid = url.rsplit("/", 1)[-1] + if pid == "missing": + raise ensure_record.Unreadable("synthetic missing target") + return {}, {"resourceType": "Patient", "id": pid} + return {}, stored["now"] + + monkeypatch.setattr(ensure_record.FhirReader, "fetch", fetch) + env["sh"].run_rechecks(now=True) + assert ep.late_verdict == expected, ep.late_reasons + + +def test_multivalued_reference_late_exception_stays_uncertain(env, monkeypatch): + from tests.test_fhir_reference import schema + + stored = _confirmed(env, monkeypatch) + ep = env["sh"].episodes[0] + # A repeated ancestor preserves scalar Reference leaves. A list directly + # in AllergyIntolerance.patient is malformed and refuses before this path. + data = ep.contract.model_dump(by_alias=True) + data.update( + resource_type="CareTeam", + reference_schema=schema("CareTeam", "participant.member"), + fields={ + "patient": {"path": "participant.member.reference"}, + "code": {"path": "category.coding.code"}, + }, + allow=[], + ) + ep.contract = ensure_record.EnsureContract.model_validate(data) + ep.params = {"patient": ep.params["patient"], "code": ep.params["code"]} + stored["now"] = { + "resourceType": "CareTeam", + "id": "a1", + "participant": [ + {"member": {"reference": "Patient/p2"}}, + {"member": {"reference": "Patient/missing"}}, + ], + "category": [{"coding": [{"code": ep.params["code"]}]}], + } + + def fetch(self, url, params=None): + if "/Patient/" in url: + pid = url.rsplit("/", 1)[-1] + if pid == "missing": + raise ensure_record.Unreadable("synthetic missing target") + return {}, {"resourceType": "Patient", "id": pid} + return {}, stored["now"] + + monkeypatch.setattr(ensure_record.FhirReader, "fetch", fetch) + env["sh"].run_rechecks(now=True) + assert ep.late_verdict == "UNCERTAIN" + assert any("ReferenceUnreadable" in reason for reason in ep.late_reasons) + + +def test_invalid_singular_reference_array_is_refused(env, monkeypatch): + from openadapt_flow.fhir_reference import ReferenceSchema + from tests.test_fhir_reference import schema + + stored = _confirmed(env, monkeypatch) + ep = env["sh"].episodes[0] + ep.contract.reference_schema = ReferenceSchema.model_validate( + schema("AllergyIntolerance", "patient") + ) + stored["now"] = _stored("Patient/p2") + stored["now"]["patient"] = [ + {"reference": "Patient/p2"}, + {"reference": "Patient/missing"}, + ] + + def fetch(self, url, params=None): + if "/Patient/" in url: + pid = url.rsplit("/", 1)[-1] + if pid == "missing": + raise ensure_record.Unreadable("synthetic missing target") + return {}, {"resourceType": "Patient", "id": pid} + return {}, stored["now"] + + monkeypatch.setattr(ensure_record.FhirReader, "fetch", fetch) + env["sh"].run_rechecks(now=True) + assert ep.late_verdict == "REFUTED" + assert any("no longer the person" in reason for reason in ep.late_reasons)