provenance: compare the saved record with its source document

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` |

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` |

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

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.

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

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"
     )

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

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