diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 3fedfa59b..953657611 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -137,6 +137,7 @@ improvising or retrying when something is off. | Judge any executor's write, including an agent's or a script's, from the system of record | Released | Synthetic | `openadapt-flow verify-run`, [VERIFY_RUN.md](packages/flow/docs/VERIFY_RUN.md) | | Prove that nothing outside the declared change moved (frame conditions), and attribute concurrent edits by other users through the app's audit log or a modified-by field; the receipt signs which parts were checked | On main | Synthetic (MockMed, SQLite audit log) | `verify-run`, [VERIFICATION_COVERAGE.md](packages/flow/docs/VERIFICATION_COVERAGE.md) | | Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` | +| Check that the saved record matches its source document, not only the parameters: intake keeps each value's line, a second reading must agree before a guarded save, and a round trip compares the read-back with the document | On main | Synthetic (MockMed, text referral) | `openadapt_flow/provenance.py`, `intake`, `verify-run --provenance`, guarded save | | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` | | Draft a record check from the demonstration's write steps (`scaffold-verifier`) | Released | Synthetic | `openadapt-flow scaffold-verifier` (a person fills in the draft) | | Build a working record check from the reads the app made while the person viewed the saved record, and test it against injected bad writes before anyone reviews it | On main ([#1284](https://github.com/OpenAdaptAI/OpenAdapt/pull/1284)) | Synthetic (MockMed, OpenEMR fixture) | `openadapt-flow scaffold-verifier` | diff --git a/llms-full.txt b/llms-full.txt index 90ba715ef..1676f43b3 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -534,6 +534,7 @@ improvising or retrying when something is off. | Judge any executor's write, including an agent's or a script's, from the system of record | Released | Synthetic | `openadapt-flow verify-run`, [VERIFY_RUN.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFY_RUN.md) | | Prove that nothing outside the declared change moved (frame conditions), and attribute concurrent edits by other users through the app's audit log or a modified-by field; the receipt signs which parts were checked | On main | Synthetic (MockMed, SQLite audit log) | `verify-run`, [VERIFICATION_COVERAGE.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFICATION_COVERAGE.md) | | Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` | +| Check that the saved record matches its source document, not only the parameters: intake keeps each value's line, a second reading must agree before a guarded save, and a round trip compares the read-back with the document | On main | Synthetic (MockMed, text referral) | `openadapt_flow/provenance.py`, `intake`, `verify-run --provenance`, guarded save | | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` | | Draft a record check from the demonstration's write steps (`scaffold-verifier`) | Released | Synthetic | `openadapt-flow scaffold-verifier` (a person fills in the draft) | | Build a working record check from the reads the app made while the person viewed the saved record, and test it against injected bad writes before anyone reviews it | On main ([#1284](https://github.com/OpenAdaptAI/OpenAdapt/pull/1284)) | Synthetic (MockMed, OpenEMR fixture) | `openadapt-flow scaffold-verifier` | diff --git a/packages/flow/docs/GUARDED_SAVE.md b/packages/flow/docs/GUARDED_SAVE.md index b3c536243..f762d9d91 100644 --- a/packages/flow/docs/GUARDED_SAVE.md +++ b/packages/flow/docs/GUARDED_SAVE.md @@ -157,6 +157,32 @@ directly inside it, the paired phone also shows the request. The receipt binds t `evidence.authorization.approver` names a person, and `self_approved` is `false`. +## Values with a source document + +A task can say where its values came from. Add `provenance` (the +`params.provenance.json` intake writes, as an object) and `sources` (paths to +the documents it cites, relative to the task file): + +```json +{ + "schema_version": "openadapt.guard-task/v1", + "task_id": "referral-0001", + "expected_record": {"patient_id": "p1"}, + "intended_fields": {"note": "Referral to orthopedics for knee pain"}, + "provenance": {"schema_version": "openadapt.param-provenance/v1", "params": {"note": {"value": "Referral to orthopedics for knee pain", "source": {"kind": "document", "sha256": "sha256:...", "line": 3, "label": "Note"}, "extractor": "intake:agent"}}}, + "sources": ["referral.pdf"] +} +``` + +`begin` reads each source a second time with the deterministic label parse. +If that reading does not confirm a value (a different value, no value, or a +value that the label parse itself produced, which is not a second reading), +the save goes to a person under any approval mode, and the approval request +names the fields. `confirm` then compares the saved record with the source +document, not only with the task, so a misread value that was saved +correctly is still refuted. A task whose values differ from its own +provenance is refused. + ## The receipt `confirm` writes an `openadapt.effect-receipt/v1` file, the same format diff --git a/packages/flow/docs/INTAKE.md b/packages/flow/docs/INTAKE.md index de5b55692..9230d3304 100644 --- a/packages/flow/docs/INTAKE.md +++ b/packages/flow/docs/INTAKE.md @@ -27,6 +27,13 @@ openadapt-flow run ./new-referral --params-file params.json fields, writes nothing, and exits with code 2. Pass `--accept NAME` after you have checked a value. +5. Writes `params.provenance.json` beside `params.json` + (`openadapt.param-provenance/v1`): for each value, the SHA-256 of the + document's bytes, the line it was read from, the label, and who read it + (`intake:local`, `intake:agent`, or `person` for a value you typed). A + later check uses it to compare the saved record with the document itself + ([`VERIFICATION_COVERAGE.md`](VERIFICATION_COVERAGE.md)). + Confidence is a fixed score from the kind of match, not a probability. An exact label match scores 0.95; a match on only the last word of the name (for example `Name:` for `patient_name`) scores 0.6. diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index 756b5daa4..eaecbb7f0 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -162,6 +162,13 @@ The receipt's `checked` field names each part of the claim behind - `source`: the inputs match their source. `instruction` means the inputs were the task's own parameters (`--params`), with no document behind them. +With `--provenance params.provenance.json --source referral.pdf`, the source +part is a round trip: for each value with a document source, a second, +deterministic reading of the cited document must equal the field the store +returned. A value misread before the save and saved faithfully passes the +delta and fails here. `evidence.source` names the fields that differ and the +extractors used, never the values. + Each part is `held`, `refuted` or `not_checked`. `verified` requires `delta` and `frame` to be `held` and `source` to be `held` or `instruction`. The field is inside `receipt_sha256`, so a Seal signs it. `evidence.frame` holds diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 0834f52f2..73bd89bb3 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3345,6 +3345,27 @@ def _cmd_attach_seal(args: argparse.Namespace) -> int: return 1 +def _verify_run_sources(args: argparse.Namespace) -> dict[str, Any]: + """``--provenance`` and ``--source`` for verify-run's round trip.""" + + from openadapt_flow.verify_run import VerifyRunError + + path = getattr(args, "provenance", None) + sources = list(getattr(args, "source", None) or []) + if not path: + if sources: + raise VerifyRunError("--source needs --provenance") + return {} + from openadapt_flow.provenance import load_provenance, read_sources + + try: + provenance = load_provenance(path) + documents = read_sources([Path(p) for p in sources]) + except (OSError, ValueError) as exc: + raise VerifyRunError(f"cannot read the provenance or a source: {exc}") from exc + return {"provenance": provenance, "documents": documents} + + def _cmd_verify_run(args: argparse.Namespace) -> int: """``verify-run``: judge an effect another executor claims, from the store.""" from openadapt_flow.verify_run import ( @@ -3407,6 +3428,7 @@ def _json(path: str) -> Any: approver=_json(args.approver) if args.approver else None, remediates=args.remediates, correlation_id=args.correlation_id, + **_verify_run_sources(args), ) except VerifyRunError as exc: print(f"verify-run REFUSED: {exc}", file=sys.stderr) @@ -9011,6 +9033,23 @@ def build_parser() -> argparse.ArgumentParser: metavar="JSON", help="Read the store now and write a private baseline, then exit", ) + p.add_argument( + "--provenance", + default=None, + metavar="JSON", + help=( + "Where each parameter came from (openadapt.param-provenance/v1, as " + "intake writes it). The saved record is then compared with the " + "source document, not only with --params" + ), + ) + p.add_argument( + "--source", + action="append", + default=None, + metavar="FILE", + help="A source document the provenance cites (repeatable)", + ) p.add_argument( "--out", default=None, metavar="DIR", help="Directory for effect-receipt.json" ) diff --git a/packages/flow/openadapt_flow/guard.py b/packages/flow/openadapt_flow/guard.py index 59b869c98..92b04395a 100644 --- a/packages/flow/openadapt_flow/guard.py +++ b/packages/flow/openadapt_flow/guard.py @@ -15,6 +15,12 @@ ``VERIFIED``, ``REFUTED`` or ``RECONCILIATION_REQUIRED`` with an ``openadapt.effect-receipt/v1`` receipt. +When the task says where its values came from (``provenance`` and the +``sources`` it cites), :meth:`GuardedSave.begin` reads the sources a second, +independent way. A value the second reading does not confirm sends the save +to a person, whatever the approval mode, and :meth:`Guard.confirm` compares +the saved record with the source document, not only with the task. + Once a task has been committed it is never committed again. ``VERIFIED`` means it is done; any other verdict means a person must look at the record, and :meth:`GuardedSave.begin` refuses the task until they do. @@ -51,7 +57,7 @@ _OPAQUE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$") _TASK_KEYS = {"schema_version", "task_id", "expected_record", "intended_fields"} -_TASK_OPTIONAL = {"authorization_ref"} +_TASK_OPTIONAL = {"authorization_ref", "provenance", "sources"} class GuardVerdict(str, Enum): @@ -90,6 +96,10 @@ class GuardTask: expected_record: Mapping[str, str] intended_fields: Mapping[str, Any] authorization_ref: str = "" + #: Where each value came from (``openadapt.param-provenance/v1``). + provenance: Optional[Any] = None + #: The source documents the provenance cites, by ``sha256:`` of their bytes. + documents: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: if not isinstance(self.task_id, str) or not _OPAQUE.match(self.task_id): @@ -139,12 +149,39 @@ def load_task(path: str | Path) -> GuardTask: raise GuardRefused( "task_invalid", "expected_record and intended_fields are objects" ) - return GuardTask( + provenance = None + documents: dict[str, Any] = {} + if "provenance" in raw: + from openadapt_flow.provenance import ProvenanceFile, read_sources + + try: + provenance = ProvenanceFile.model_validate(raw["provenance"]) + sources = raw.get("sources") or [] + if not isinstance(sources, list): + raise ValueError("sources must be a list of paths") + base = Path(path).resolve().parent + documents = read_sources([base / str(p) for p in sources]) + except (OSError, ValueError) as exc: + raise GuardRefused( + "task_invalid", f"provenance or a source can't be read: {exc}" + ) from exc + elif "sources" in raw: + raise GuardRefused("task_invalid", "sources needs provenance") + task = GuardTask( task_id=raw["task_id"], expected_record=dict(raw["expected_record"]), intended_fields=dict(raw["intended_fields"]), authorization_ref=str(raw.get("authorization_ref") or ""), + provenance=provenance, + documents=documents, ) + if provenance is not None and provenance.matches(task.params): + raise GuardRefused( + "task_invalid", + "the task's values differ from the values its provenance records: " + + ", ".join(provenance.matches(task.params)), + ) + return task def check_identity( @@ -317,6 +354,16 @@ def begin( ) guard_id = f"guard-{secrets.token_hex(16)}" guard = Guard(self, task, guard_id, agent_id=agent_id) + if task.provenance is not None and task.provenance.has_documents(): + # A second, independent reading of the sources. A value it does + # not confirm goes to a person before anything is written. + from openadapt_flow.provenance import cross_check, default_extractor + + check = cross_check( + task.provenance, task.documents, default_extractor(task.provenance) + ) + guard.source_check = check.status + guard.source_held_fields = check.disagreements status, missing = check_identity(task.expected_record, screen_text) guard.identity = status if status != "verified": @@ -358,6 +405,9 @@ class Guard: state: str = "new" identity: str = "unchecked" approval: Optional[dict[str, str]] = None + #: ``agree`` | ``hold`` | ``instruction``, or None without provenance. + source_check: Optional[str] = None + source_held_fields: list[str] = field(default_factory=list) result: Optional[GuardResult] = None _request: Any = field(default=None, repr=False) _approver: Optional[dict[str, str]] = field(default=None, repr=False) @@ -378,6 +428,8 @@ def _record(self, reason: Optional[str] = None) -> dict[str, Any]: "identity": self.identity, "approval_mode": self.owner.approval, "approval": self.approval, + "source_check": self.source_check, + "source_held_fields": list(self.source_held_fields), "params_sha256": _sha256_hex(self.task.params), "verdict": self.result.verdict.value if self.result else None, "outcome": self.result.outcome if self.result else None, @@ -424,7 +476,7 @@ def commit( + (f" (not found: {', '.join(missing)})" if missing else "") + "; nothing may be written", ) - if self.owner.approval == "each_item": + if self.owner.approval == "each_item" or self.source_check == "hold": waiting = self._await_approval( screenshot_png, wait_s, reason=reason, screen_text=screen_text ) @@ -480,7 +532,17 @@ def _await_approval( workflow_name="guarded-save", step_id="save", step_index=0, - step_intent="Save " + ", ".join(sorted(self.task.intended_fields)), + step_intent=( + "Save " + + ", ".join(sorted(self.task.intended_fields)) + + ( + ". A second reading of the source did not confirm: " + + ", ".join(self.source_held_fields) + + ". Compare them with the source document" + if self.source_check == "hold" + else "" + ) + ), mode=ApprovalMode.EACH_ITEM, bundle_content_digest=None, runtime_inputs_digest=_sha256_hex(self.task.params), @@ -574,6 +636,8 @@ def confirm(self) -> GuardResult: approver=self._approver, correlation_id=self.guard_id, write_attempted=True, + provenance=self.task.provenance, + documents=self.task.documents, ) receipt_path = str(write_result(receipt_doc, self.dir)) except Exception as exc: # unreadable store, timeout, bad config diff --git a/packages/flow/openadapt_flow/intake/cli.py b/packages/flow/openadapt_flow/intake/cli.py index 8235da210..a9a3dd3da 100644 --- a/packages/flow/openadapt_flow/intake/cli.py +++ b/packages/flow/openadapt_flow/intake/cli.py @@ -81,8 +81,15 @@ def cmd_intake(args: argparse.Namespace) -> int: ) return 2 Path(args.out).write_text(json.dumps(result.to_params(), indent=2) + "\n") + # Where each value came from: the document's digest and the line, so a + # later check can compare the saved record with the document itself. + from openadapt_flow.provenance import document_sha256, from_intake + + prov_path = Path(args.out).with_suffix(".provenance.json") + prov = from_intake(result, document_sha256(args.document)) + prov_path.write_text(prov.model_dump_json(indent=2) + "\n") print( - f"Wrote {args.out}. Run: " + f"Wrote {args.out} and {prov_path}. Run: " + command(f"run {args.bundle} --params-file {args.out}") ) return 0 diff --git a/packages/flow/openadapt_flow/provenance.py b/packages/flow/openadapt_flow/provenance.py new file mode 100644 index 000000000..8fcf5907e --- /dev/null +++ b/packages/flow/openadapt_flow/provenance.py @@ -0,0 +1,448 @@ +"""Where each written value came from, and whether the record matches it. + +A record check confirms that the store holds the values it was given. If the +referral said ``02139`` and the extraction read ``02193``, the save of +``02193`` passes every check: the error happened before the save. This module +closes that gap in three steps (``docs/VERIFICATION_COVERAGE.md`` section 3): + +1. **Provenance.** Every parameter can name its source: a line of a document + (by the document's digest), a worklist cell, or the authorized instruction + itself. Intake already finds the line for every field; :func:`from_intake` + keeps it. +2. **A second extraction must agree before the save.** :func:`cross_check` + reads the same source with a different method and compares value by + value. A disagreement, a value the second method can't find, or two + extractions by the same method hold the item for a person. +3. **A round trip after the save.** :func:`round_trip` compares the record as + read back with the source document, not with the parameters, so a + correctly persisted wrong entry is refuted. + +Values compare by kind (text, code, date, currency) with the same rules the +screen read-back uses (:mod:`openadapt_flow.runtime.effects.readback_values`), +read exactly (no OCR leniency). Nothing here calls a model: a model extractor +is something a caller passes in. + +What this can't catch: a source document that is itself wrong, and a value +both methods misread the same way. The receipt names both methods so a +reviewer can judge how independent they were. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Literal, Optional, Protocol + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +PROVENANCE_SCHEMA: Literal["openadapt.param-provenance/v1"] = ( + "openadapt.param-provenance/v1" +) + +SourceKind = Literal["document", "worklist_cell", "instruction"] +SourceStatus = Literal["held", "refuted", "not_checked", "instruction"] + +#: Extractor methods. Two extractions are independent only when their methods +#: differ. +METHOD_OF: dict[str, str] = { + "intake:local": "label_parse", + "intake:agent": "model_line", + "ocr": "ocr", + "worklist": "cell_read", +} + + +def method_of(extractor: str) -> str: + """The method family of an extractor name (``model:`` is ``model``).""" + + if extractor in METHOD_OF: + return METHOD_OF[extractor] + return extractor.split(":", 1)[0] + + +class SourceRef(BaseModel): + """Where one value came from.""" + + model_config = ConfigDict(extra="forbid") + + kind: SourceKind + #: ``document`` / ``worklist_cell``: ``sha256:`` of the source file bytes. + sha256: Optional[str] = Field(default=None, pattern=r"^sha256:[0-9a-f]{64}$") + #: ``document``: the 1-based line that carries the value. + line: Optional[int] = Field(default=None, ge=1) + #: ``document``: the label before the value on that line. + label: Optional[str] = None + #: ``worklist_cell``: 1-based row and the column name. + row: Optional[int] = Field(default=None, ge=1) + column: Optional[str] = None + + @model_validator(mode="after") + def _shape(self) -> "SourceRef": + if self.kind == "document" and (self.sha256 is None or self.line is None): + raise ValueError("a document source needs sha256 and line") + if self.kind == "worklist_cell" and ( + self.sha256 is None or self.row is None or not self.column + ): + raise ValueError("a worklist cell needs sha256, row and column") + return self + + +class ParamProvenance(BaseModel): + model_config = ConfigDict(extra="forbid") + + value: str + source: SourceRef + #: Who read the value: ``intake:local``, ``intake:agent``, ``model:``, + #: ``ocr``, ``worklist``, ``person``. + extractor: str = Field(min_length=1) + #: ``text`` | ``code`` | ``date`` | ``currency``. Unset: inferred. + value_kind: Optional[Literal["text", "code", "date", "currency"]] = None + + +class ProvenanceFile(BaseModel): + """``params.provenance.json``: one entry per parameter that has a source.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: Literal["openadapt.param-provenance/v1"] = PROVENANCE_SCHEMA + params: dict[str, ParamProvenance] + + def documents(self) -> set[str]: + return { + p.source.sha256 + for p in self.params.values() + if p.source.kind != "instruction" and p.source.sha256 + } + + def has_documents(self) -> bool: + return any(p.source.kind != "instruction" for p in self.params.values()) + + def matches(self, params: Mapping[str, object]) -> list[str]: + """Names whose provenance value differs from the run's value.""" + + return sorted( + name + for name, p in self.params.items() + if name in params and str(params[name]) != p.value + ) + + +def sha256_bytes(data: bytes) -> str: + return "sha256:" + hashlib.sha256(data).hexdigest() + + +def document_sha256(path: Path | str) -> str: + return sha256_bytes(Path(path).read_bytes()) + + +def from_intake( + result: Any, document_sha: str, *, extractor: Optional[str] = None +) -> ProvenanceFile: + """Provenance for an :class:`~openadapt_flow.intake.extract.IntakeResult`. + + A value a person confirmed or typed keeps its line when it has one; its + extractor is ``person``. A value with no line is not given a source. + """ + + name_of = extractor or f"intake:{result.mode}" + params: dict[str, ParamProvenance] = {} + for name, fld in result.fields.items(): + if fld.value is None or fld.source_line is None: + continue + params[name] = ParamProvenance( + value=str(fld.value), + source=SourceRef( + kind="document", + sha256=document_sha, + line=fld.source_line, + label=fld.label, + ), + extractor="person" if fld.confirmed and fld.confidence >= 1.0 else name_of, + ) + return ProvenanceFile(params=params) + + +# -- second extraction ----------------------------------------------------------- + + +class Extractor(Protocol): + """A second way to read a value from a source document.""" + + #: Recorded in the receipt; :func:`method_of` decides independence. + name: str + + def read(self, document: Any, name: str, ref: SourceRef) -> Optional[str]: + """The value of parameter ``name`` per this method, or None.""" + + +@dataclass +class LabelExtractor: + """Deterministic: find the parameter's label anywhere in the document. + + The intake label parser, run over the whole document (not only the cited + line), so a citation to the wrong line is caught too. ``specs`` gives each + parameter's type for normalization; ``aliases`` extra labels. + """ + + specs: Mapping[str, Any] + aliases: Mapping[str, list[str]] = field(default_factory=dict) + name: str = "intake:local" + + def read(self, document: Any, name: str, ref: SourceRef) -> Optional[str]: + from openadapt_flow.intake.extract import extract_fields + + spec = self.specs.get(name) + if spec is None: + return None + aliases = dict(self.aliases) + if ref.label: + aliases[name] = [*aliases.get(name, []), ref.label] + found = extract_fields(document, {name: spec}, aliases=aliases).fields[name] + if found.error or found.value is None: + return None + return str(found.value) + + +@dataclass +class CallableExtractor: + """Wrap a caller's function (a second model, OCR) as an extractor.""" + + name: str + fn: Callable[[Any, str, SourceRef], Optional[str]] + + def read(self, document: Any, name: str, ref: SourceRef) -> Optional[str]: + return self.fn(document, name, ref) + + +def _kind(prov: ParamProvenance, value: str) -> str: + from openadapt_flow.runtime.effects.readback_values import infer_value_kind + + return prov.value_kind or infer_value_kind(value) + + +def same_value(a: str, b: str, kind: str) -> bool: + from openadapt_flow.runtime.effects.readback_values import compare_value + + return compare_value(a, b, kind=kind, source="structure") == "match" # type: ignore[arg-type] + + +@dataclass +class FieldCheck: + name: str + #: ``agree`` | ``disagree`` | ``missing`` | ``instruction`` | + #: ``not_independent`` | ``no_document`` + result: str + #: In memory only: never put in a receipt (values are data). + first: Optional[str] = None + second: Optional[str] = None + + +@dataclass +class CrossCheck: + """The pre-save comparison of two extractions.""" + + status: Literal["agree", "hold", "instruction"] + second_extractor: str + fields: list[FieldCheck] = field(default_factory=list) + + @property + def disagreements(self) -> list[str]: + return [f.name for f in self.fields if f.result not in ("agree", "instruction")] + + def summary(self) -> dict[str, Any]: + counts: dict[str, int] = {} + for f in self.fields: + counts[f.result] = counts.get(f.result, 0) + 1 + return { + "status": self.status, + "second_extractor": self.second_extractor, + "fields": dict(sorted(counts.items())), + "held_fields": self.disagreements, + } + + +def cross_check( + provenance: ProvenanceFile, + documents: Mapping[str, Any], + second: Extractor, +) -> CrossCheck: + """Compare every sourced value with a second, independent extraction. + + ``documents`` maps a document's ``sha256:`` to its text + (:class:`~openadapt_flow.intake.document.IntakeDocument`). A source whose + document isn't supplied can't be checked, so it holds. + """ + + checks: list[FieldCheck] = [] + for name, prov in sorted(provenance.params.items()): + if prov.source.kind == "instruction": + checks.append(FieldCheck(name, "instruction")) + continue + if method_of(prov.extractor) == method_of(second.name): + checks.append(FieldCheck(name, "not_independent", prov.value)) + continue + document = documents.get(prov.source.sha256 or "") + if document is None: + checks.append(FieldCheck(name, "no_document", prov.value)) + continue + other = second.read(document, name, prov.source) + if other is None: + checks.append(FieldCheck(name, "missing", prov.value)) + elif same_value(prov.value, other, _kind(prov, prov.value)): + checks.append(FieldCheck(name, "agree", prov.value, other)) + else: + checks.append(FieldCheck(name, "disagree", prov.value, other)) + sourced = [c for c in checks if c.result != "instruction"] + if not sourced: + status: Literal["agree", "hold", "instruction"] = "instruction" + elif all(c.result == "agree" for c in sourced): + status = "agree" + else: + status = "hold" + return CrossCheck(status, second.name, checks) + + +# -- round trip ------------------------------------------------------------------ + + +@dataclass +class RoundTrip: + """The read-back record against the source document.""" + + status: SourceStatus + second_extractor: str + reason: str + fields: list[FieldCheck] = field(default_factory=list) + + def summary(self) -> dict[str, Any]: + counts: dict[str, int] = {} + for f in self.fields: + counts[f.result] = counts.get(f.result, 0) + 1 + return { + "status": self.status, + "second_extractor": self.second_extractor, + "fields": dict(sorted(counts.items())), + "mismatched_fields": [ + f.name for f in self.fields if f.result == "disagree" + ], + } + + +def bound_fields(effects: Any) -> dict[str, str]: + """Parameter name -> record field, from the effects' selectors and values.""" + + out: dict[str, str] = {} + for effect in effects: + for record_field, expr in effect.match.items(): + if expr.param is not None: + out.setdefault(expr.param, record_field) + if effect.field and effect.value is not None and effect.value.param: + out.setdefault(effect.value.param, effect.field) + return out + + +def round_trip( + provenance: ProvenanceFile, + record: Optional[Mapping[str, Any]], + fields: Mapping[str, str], + documents: Mapping[str, Any], + second: Extractor, +) -> RoundTrip: + """Compare ``record`` (as read back) with the source of each value. + + ``fields`` maps a parameter to the record field it was written to + (:func:`bound_fields`). For each sourced parameter the second method reads + the source; the record's field must equal that reading. The comparison + never consults the parameters, so a value misread before the save and + saved faithfully is refuted. + """ + + sourced = { + name: prov + for name, prov in sorted(provenance.params.items()) + if prov.source.kind != "instruction" + } + if not sourced: + return RoundTrip("instruction", second.name, "every input is the instruction") + if record is None: + return RoundTrip( + "not_checked", + second.name, + "no single saved record was read back to compare with its source", + ) + checks: list[FieldCheck] = [] + for name, prov in sourced.items(): + record_field = fields.get(name) + if record_field is None or record_field not in record: + checks.append(FieldCheck(name, "not_read_back")) + continue + document = documents.get(prov.source.sha256 or "") + if document is None: + checks.append(FieldCheck(name, "no_document")) + continue + source_value = second.read(document, name, prov.source) + stored = "" if record.get(record_field) is None else str(record[record_field]) + if source_value is None: + checks.append(FieldCheck(name, "missing", stored)) + elif same_value(stored, source_value, _kind(prov, source_value)): + checks.append(FieldCheck(name, "agree", stored, source_value)) + else: + checks.append(FieldCheck(name, "disagree", stored, source_value)) + if any(c.result == "disagree" for c in checks): + names = [c.name for c in checks if c.result == "disagree"] + return RoundTrip( + "refuted", + second.name, + "the saved record differs from its source document in " + ", ".join(names), + checks, + ) + if all(c.result == "agree" for c in checks): + return RoundTrip( + "held", + second.name, + f"{len(checks)} saved field(s) match their source document", + checks, + ) + unchecked = [c.name for c in checks if c.result != "agree"] + return RoundTrip( + "not_checked", + second.name, + "could not compare with the source: " + ", ".join(unchecked), + checks, + ) + + +def default_extractor(provenance: ProvenanceFile) -> LabelExtractor: + """The deterministic label parse, typed from the provenance itself. + + A parameter recorded as a date is read as a date; everything else as + text. Each cited label is an alias, so the parse looks for the same label + the first reader cited, anywhere in the document. + """ + + from openadapt_flow.ir import ParamKind, ParamSpec + + specs: dict[str, Any] = {} + aliases: dict[str, list[str]] = {} + for name, prov in provenance.params.items(): + kind = ParamKind.DATE if prov.value_kind == "date" else ParamKind.STRING + specs[name] = ParamSpec(name=name, type=kind) + if prov.source.label: + aliases[name] = [prov.source.label] + return LabelExtractor(specs, aliases) + + +def read_sources(paths: list[Path]) -> dict[str, Any]: + """``sha256:`` of each file's bytes -> its text, read as intake reads it.""" + + from openadapt_flow.intake.document import read_document + + return {document_sha256(path): read_document(path) for path in paths} + + +def load_provenance(path: Path | str) -> ProvenanceFile: + import json + + return ProvenanceFile.model_validate(json.loads(Path(path).read_text())) diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 00bc06ef7..1d9d4100e 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -492,6 +492,65 @@ def read_of(name: str) -> str: ) +def _target_record( + effects: Sequence[Effect], verdicts: Sequence[EffectVerdict] +) -> Optional[dict[str, Any]]: + """The one record every confirmed write/read-back effect matched, if any.""" + + from openadapt_flow.runtime.effects.effect import stable_id + + found: Optional[dict[str, Any]] = None + for effect, verdict in zip(effects, verdicts): + if effect.kind not in (EffectKind.RECORD_WRITTEN, EffectKind.FIELD_EQUALS): + continue + if effect.kind is EffectKind.RECORD_WRITTEN and effect.expected_count == 0: + continue # a guard: "nothing here" + if not verdict.confirmed or len(verdict.matched_records) != 1: + return None + record = verdict.matched_records[0] + if found is not None and stable_id(found) != stable_id(record): + return None + found = record + return found + + +def _judge_source( + provenance: Optional[Any], + documents: Mapping[str, Any], + second: Optional[Any], + params: Mapping[str, object], + effects: Sequence[Effect], + verdicts: Sequence[EffectVerdict], + delta_held: bool, +) -> tuple[str, dict[str, Any]]: + """``checked.source`` and its evidence (counts and names, no values).""" + + from openadapt_flow.provenance import ( + bound_fields, + default_extractor, + round_trip, + ) + + if provenance is None or not provenance.has_documents(): + return "instruction", {"status": "instruction"} + differ = provenance.matches(params) + if differ: + # The values checked are not the values the source provenance records. + return "refuted", { + "status": "refuted", + "params_differ_from_provenance": differ, + } + extractor = second or default_extractor(provenance) + record = _target_record(effects, verdicts) if delta_held else None + result = round_trip(provenance, record, bound_fields(effects), documents, extractor) + evidence = result.summary() + evidence["first_extractors"] = sorted( + {p.extractor for p in provenance.params.values()} + ) + evidence["documents"] = len(provenance.documents()) + return result.status, evidence + + def _part(verdict: EffectVerdictLabel, proved: bool) -> str: if proved: return "held" @@ -834,6 +893,9 @@ def verify_run( correlation_id: Optional[str] = None, write_attempted: bool = False, coverage: Optional[CoverageRecord] = None, + provenance: Optional[Any] = None, + documents: Optional[Mapping[str, Any]] = None, + second_extractor: Optional[Any] = None, ) -> dict[str, Any]: """Read the system of record and return ``{"receipt", "evidence"}``. @@ -1044,12 +1106,24 @@ def verify_run( and strong_enough and oracle_tier in (2, 3) ) - # Inputs come from whoever authorized the task (``--params``): the - # instruction is their source. + # The source part: with provenance, the record as read back must match + # the source document (a round trip). Without it the inputs are the + # authorized instruction itself (``--params``). + source_status, source_evidence = _judge_source( + provenance, + documents or {}, + second_extractor, + params, + # Unbound: the field each parameter was written to is named by its + # ``{param: ...}`` reference, which binding replaces with the value. + list(cfg.effects), + verdicts, + read_proves_write, + ) checked: dict[str, Any] = { "delta": _part(delta_verdict, read_proves_write), "frame": frame.status, - "source": "instruction", + "source": source_status, } admitted = True if cfg.admission is not None: @@ -1139,6 +1213,7 @@ def verify_run( "effects": rows, }, "frame": frame.evidence(frame_spec), + "source": source_evidence, "checked": checked, "verifier": { "name": "openadapt-flow verify-run", diff --git a/packages/flow/tests/test_provenance.py b/packages/flow/tests/test_provenance.py new file mode 100644 index 000000000..f2711d5b6 --- /dev/null +++ b/packages/flow/tests/test_provenance.py @@ -0,0 +1,330 @@ +"""Provenance: each written value names its source, and the record matches it. + +See ``docs/VERIFICATION_COVERAGE.md`` section 3. The referral is made up. +""" + +from __future__ import annotations + +from openadapt_flow.intake.document import IntakeDocument +from openadapt_flow.intake.extract import extract_fields +from openadapt_flow.ir import ParamKind, ParamSpec +from openadapt_flow.provenance import ( + CallableExtractor, + LabelExtractor, + ParamProvenance, + ProvenanceFile, + SourceRef, + cross_check, + from_intake, + method_of, + round_trip, + sha256_bytes, +) +from openadapt_flow.runtime.effects import Effect, EffectKind + +REFERRAL = IntakeDocument( + "referral.txt", + "text", + [ + "Referral to: Example Family Clinic", + "Last name: Okafor", + "Date of birth: 03/07/1991", + "Postal code: 02139", + "Home phone: 555-0142", + ], +) +DOC = sha256_bytes(REFERRAL.text.encode("utf-8")) +SPECS = { + "lname": ParamSpec(name="lname", type=ParamKind.STRING), + "DOB": ParamSpec(name="DOB", type=ParamKind.DATE), + "postal_code": ParamSpec(name="postal_code", type=ParamKind.STRING), + "phone_home": ParamSpec(name="phone_home", type=ParamKind.STRING), +} +ALIASES = { + "lname": ["last name"], + "DOB": ["date of birth"], + "phone_home": ["home phone"], +} +LABELS = LabelExtractor(SPECS, ALIASES) + + +def _prov(**values: str) -> ProvenanceFile: + lines = {"lname": 2, "DOB": 3, "postal_code": 4, "phone_home": 5} + return ProvenanceFile( + params={ + name: ParamProvenance( + value=value, + source=SourceRef(kind="document", sha256=DOC, line=lines[name]), + extractor="model:first-reader", + ) + for name, value in values.items() + } + ) + + +GOOD = { + "lname": "Okafor", + "DOB": "1991-03-07", + "postal_code": "02139", + "phone_home": "555-0142", +} + + +def test_intake_keeps_the_line_it_read() -> None: + result = extract_fields(REFERRAL, SPECS, aliases=ALIASES) + prov = from_intake(result, DOC) + assert prov.params["postal_code"].source.line == 4 + assert prov.params["postal_code"].extractor == "intake:local" + assert prov.params["DOB"].value == "1991-03-07" + + +def test_two_methods_that_agree_let_the_save_go() -> None: + check = cross_check(_prov(**GOOD), {DOC: REFERRAL}, LABELS) + assert check.status == "agree" + assert check.disagreements == [] + + +def test_a_misread_value_holds_the_save() -> None: + misread = {**GOOD, "postal_code": "02193"} + check = cross_check(_prov(**misread), {DOC: REFERRAL}, LABELS) + assert check.status == "hold" + assert check.disagreements == ["postal_code"] + # The summary carries names and counts, never the values. + assert "02193" not in str(check.summary()) + + +def test_the_same_method_twice_is_not_a_second_extraction() -> None: + result = extract_fields(REFERRAL, SPECS, aliases=ALIASES) + check = cross_check(from_intake(result, DOC), {DOC: REFERRAL}, LABELS) + assert check.status == "hold" + assert {f.result for f in check.fields} == {"not_independent"} + assert method_of("intake:local") != method_of("model:second-reader") + + +def test_a_missing_document_holds() -> None: + assert cross_check(_prov(**GOOD), {}, LABELS).status == "hold" + + +def test_instruction_inputs_need_no_second_extraction() -> None: + prov = ProvenanceFile( + params={ + "lname": ParamProvenance( + value="Okafor", source=SourceRef(kind="instruction"), extractor="person" + ) + } + ) + assert cross_check(prov, {}, LABELS).status == "instruction" + + +def _effects() -> list[Effect]: + match = {"lname": {"param": "lname"}} + return [ + Effect(kind=EffectKind.RECORD_WRITTEN, match=match), + *( + Effect( + kind=EffectKind.FIELD_EQUALS, + match=match, + field=name, + value={"param": name}, + ) + for name in ("DOB", "postal_code", "phone_home") + ), + ] + + +def test_round_trip_holds_for_a_faithful_record() -> None: + from openadapt_flow.provenance import bound_fields + + record = { + "id": 7, + "lname": "Okafor", + "DOB": "1991-03-07", + "postal_code": "02139", + "phone_home": "555-0142", + } + result = round_trip( + _prov(**GOOD), record, bound_fields(_effects()), {DOC: REFERRAL}, LABELS + ) + assert result.status == "held", result.reason + + +def test_round_trip_refutes_a_correctly_saved_wrong_entry() -> None: + """The params said 02193, the store holds 02193, the referral says 02139.""" + + from openadapt_flow.provenance import bound_fields + + misread = {**GOOD, "postal_code": "02193"} + record = {**GOOD, "id": 7, "postal_code": "02193"} + result = round_trip( + _prov(**misread), record, bound_fields(_effects()), {DOC: REFERRAL}, LABELS + ) + assert result.status == "refuted" + assert result.summary()["mismatched_fields"] == ["postal_code"] + + +def test_round_trip_without_a_record_is_not_checked() -> None: + result = round_trip( + _prov(**GOOD), None, {"lname": "lname"}, {DOC: REFERRAL}, LABELS + ) + assert result.status == "not_checked" + + +def test_a_caller_supplies_another_method() -> None: + def ocr_like(document, name, ref): + return document.lines[ref.line - 1].split(":", 1)[1].strip() + + check = cross_check( + _prov(postal_code="02139"), {DOC: REFERRAL}, CallableExtractor("ocr", ocr_like) + ) + assert check.status == "agree" + + +# -- verify-run and the guarded save --------------------------------------------- + +import json # noqa: E402 +from pathlib import Path # noqa: E402 + +import pytest # noqa: E402 + +from openadapt_flow.mockmed.fault_server import serve as fault_serve # noqa: E402 + +SOURCE_TEXT = "Referral\nPatient: p1\nNote: Referral to orthopedics for knee pain\n" +TRUE_NOTE = "Referral to orthopedics for knee pain" +MISREAD_NOTE = "Referral to orthopaedics for knee pain" + + +@pytest.fixture +def sor(): + url, db, stop = fault_serve() + try: + yield url.rstrip("/"), db + finally: + stop() + + +def _source(tmp_path: Path) -> tuple[Path, str]: + path = tmp_path / "referral.txt" + path.write_text(SOURCE_TEXT) + from openadapt_flow.provenance import document_sha256 + + return path, document_sha256(path) + + +def _note_provenance(sha: str, value: str) -> ProvenanceFile: + return ProvenanceFile( + params={ + "note": ParamProvenance( + value=value, + source=SourceRef(kind="document", sha256=sha, line=3, label="Note"), + extractor="model:first-reader", + ) + } + ) + + +def test_verify_run_refutes_a_faithfully_saved_misread(sor, tmp_path: Path) -> None: + import requests + + from openadapt_flow.guard_demo import verify_yaml + from openadapt_flow.provenance import read_sources + from openadapt_flow.verify_run import ( + capture_baseline, + parse_verify_config, + verify_run, + ) + + base, _db = sor + path, sha = _source(tmp_path) + cfg = parse_verify_config(verify_yaml(base)) + + def run(note: str, name: str) -> dict: + params = {"patient_id": "p1", "note": note} + baseline = tmp_path / f"{name}.json" + capture_baseline(cfg, params, baseline) + requests.post( + f"{base}/api/encounter", + json={"patient_id": "p1", "type": "Referral", "note": note}, + timeout=5, + ) + return verify_run( + cfg, + params=params, + run_record={"executor_kind": "agent", "executor_id": "test-agent"}, + authorization_ref="ticket-0000-source", + baseline_path=baseline, + private_dir=tmp_path / name, + provenance=_note_provenance(sha, note), + documents=read_sources([path]), + ) + + good = run(TRUE_NOTE, "good") + assert good["receipt"]["checked"] == { + "delta": "held", + "frame": "held", + "source": "held", + } + assert good["receipt"]["outcome"] == "verified" + bad = run(MISREAD_NOTE, "bad") + assert bad["receipt"]["checked"]["delta"] == "held" + assert bad["receipt"]["checked"]["source"] == "refuted" + assert bad["receipt"]["outcome"] == "reconciliation_required" + assert bad["evidence"]["source"]["mismatched_fields"] == ["note"] + assert MISREAD_NOTE not in json.dumps(bad) + + +def _task_file(tmp_path: Path, note: str) -> Path: + path, sha = _source(tmp_path) + task = { + "schema_version": "openadapt.guard-task/v1", + "task_id": f"source-{abs(hash(note)) % 10**8:08d}", + "expected_record": {"patient_id": "p1"}, + "intended_fields": {"note": note}, + "provenance": json.loads(_note_provenance(sha, note).model_dump_json()), + "sources": [path.name], + } + out = tmp_path / "task.json" + out.write_text(json.dumps(task)) + return out + + +def test_the_guard_sends_a_disagreement_to_a_person(sor, tmp_path: Path) -> None: + from openadapt_flow.guard import GuardedSave, GuardRefused, load_task + from openadapt_flow.guard_demo import FakeAgent, verify_yaml + from openadapt_flow.verify_run import parse_verify_config + + base, _db = sor + guarded = GuardedSave( + parse_verify_config(verify_yaml(base)), tmp_path / "state", approval="not_gated" + ) + agent = FakeAgent(base) + screen = agent.open_chart("p1") + + agreed = guarded.begin( + load_task(_task_file(tmp_path, TRUE_NOTE)), screen_text=screen + ) + assert agreed.source_check == "agree" + assert agreed.commit().status == "go" + + misread = guarded.begin( + load_task(_task_file(tmp_path, MISREAD_NOTE)), screen_text=screen + ) + assert misread.source_check == "hold" + assert misread.source_held_fields == ["note"] + # Not gated, but this save goes to a person: commit needs what they see. + with pytest.raises(GuardRefused) as refused: + misread.commit() + assert refused.value.code == "screenshot_required" + + +def test_a_task_whose_values_differ_from_its_provenance_is_refused( + tmp_path: Path, +) -> None: + from openadapt_flow.guard import GuardRefused, load_task + + path = _task_file(tmp_path, TRUE_NOTE) + task = json.loads(path.read_text()) + task["intended_fields"]["note"] = MISREAD_NOTE + path.write_text(json.dumps(task)) + with pytest.raises(GuardRefused) as refused: + load_task(path) + assert refused.value.code == "task_invalid"