diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 874ac4595..2e285085e 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -136,6 +136,7 @@ improvising or retrying when something is off. | Never retry an uncertain save (`RECONCILIATION_REQUIRED`) | Released | Synthetic (fault server) | [VERIFICATION.md](packages/flow/docs/VERIFICATION.md) | | One check-level scale, 0 to 5, from "unchecked" to "counterparty confirmed" | Released | Unit tests | `openadapt_flow/check_level.py` | | 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) | | 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 0ea008a4a..cbfa88a16 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -533,6 +533,7 @@ improvising or retrying when something is off. | Never retry an uncertain save (`RECONCILIATION_REQUIRED`) | Released | Synthetic (fault server) | [VERIFICATION.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFICATION.md) | | One check-level scale, 0 to 5, from "unchecked" to "counterparty confirmed" | Released | Unit tests | `openadapt_flow/check_level.py` | | 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) | | 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/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index 698daae1a..097f28ff7 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -80,34 +80,104 @@ unless given, and when present they are signed with the rest of the receipt. ## Records outside the target -With a baseline, a `record_written` check also refutes when a record that -was there before the agent started and is not the target has changed -(`forbid_collateral_change`), not only when it is gone -(`forbid_collateral_loss`). The agent's write can be right while it also -edited another patient's record in place, and only the baseline shows that. -verify-run turns this on for every `record_written` effect when it judges. -The contract in the file is unchanged, so its digest is still the one qualify -reports, and `evidence.readback.effects[].collateral_change_checked` says the -check ran. To skip it for one effect, for example when the read covers a -table other people edit during the run, set `forbid_collateral_change: false` -on that effect in `effects:`. Fields that change on their own, such as a -modification time, belong in the verifier's `records_drop_fields`. - -A changed record is not proof that the executor changed it: another user may -have edited it during the run. The verdict says so, and the outcome is -`reconciliation_required`, so a person checks the record before anything is +With a baseline, verify-run also proves that nothing outside the declared +delta changed: no record that the effects don't name was added, removed or +edited, and a pre-existing target changed only in the fields the effects +name. This is the frame check ([`VERIFICATION_COVERAGE.md`](VERIFICATION_COVERAGE.md)). +It reads the same read set the effects use, once after the save, and judges +the whole contract at once, so two effects on one read set never flag each +other's targets. The agent's write can be right while it also edited another +patient's record in place, and only the baseline shows that. + +Without a baseline the frame can't be checked, and the run is not +`verified`. + +A `frame:` section says how records are compared and who may change them: + +```yaml +frame: + identity_field: pid # default: id, else the whole record + ignore_fields: [date] # fields that change on their own + target_fields: [phone_home] # default: the fields the effects name + attribution: # optional + kind: audit_read # or record_field, with server_set: [...] + actor_field: user + record_field: patient_id + self: [{param: bot_user}] # the bot's own user or session ids + read: # a second read-only read: the app's audit log + kind: sql + sql_driver: pymysql + sql_query: SELECT id, user, patient_id FROM log WHERE event LIKE 'patient-record-%' +``` + +Each change outside the declared delta is attributed. A change that the +audit log (`audit_read`) or the record's own modified-by field with a new +version (`record_field`) attributes to another user is a concurrent edit: it +is listed in `evidence.frame.concurrent_count` and does not count against the +save. A change by one of the bot's own ids, or one nothing explains, is +collateral, and the outcome is `reconciliation_required`. With `audit_read`, +a log entry by the bot for a record no effect names is collateral too, even +when the read shows no change. + +Only evidence the bot can't write counts: + +- `audit_read` reads the app's own audit log or change feed. +- `record_field` reads fields the app sets on the server. List them in + `server_set` (they must include `actor_field` and any `version_field`); + that list is the per-app statement that no user or API client can write + them. A field the run itself writes, or that a run parameter names, is + never evidence: attribution that uses one is refused, and every change + stays unattributed (`evidence.frame.attribution: attribution_refused`). +- The run's own ids (`self`, and the executor's id from the run record) are + never "another user". +- Another user's change to a record the save declares is never excused. The + saved value may no longer stand, so the outcome is + `reconciliation_required` (`evidence.frame.target_changed_by_other`). Only + changes outside the declared delta can be concurrent edits. + +Without `attribution`, every change outside the delta is collateral. A +change is not proof that the executor made it: another user may have edited +the record during the run. A person checks the record before anything is retried or undone. Values compare by type. A `Decimal` or `datetime` read from a database equals the JSON form the baseline stored it in, while two strings compare exactly, so `"02139"` changed to `"2139"` is a change. +When the frame is judged, the per-effect collateral rules +(`forbid_collateral_loss`, `forbid_collateral_change`) apply only where the +file sets them explicitly; the frame covers the rest. Without a baseline, +`forbid_collateral_change` is turned on as before, though it can't refute +without a baseline either. + +## What the receipt says was checked + +The receipt's `checked` field names each part of the claim behind +`verified` and how it came out: + +```json +"checked": {"delta": "held", "frame": "held", "source": "instruction"} +``` + +- `delta`: the declared records, read from the system of record. +- `frame`: nothing else in the read set changed. +- `source`: the inputs match their source. `instruction` means the inputs + were the task's own parameters (`--params`), with no document behind them. + +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 +the counts behind the frame part (records compared, collateral, concurrent) +and the digest of the `frame:` section. A receipt issued before the field +existed has neither and still checks. + ## Verdicts | What the store shows | `effect_verdict` | `outcome` | |---|---|---| -| The declared record, exactly once, through a system-of-record read | `CONFIRMED` | `verified` | +| The declared record, exactly once, through a system-of-record read, and nothing else in the read set changed | `CONFIRMED` | `verified` | +| The declared record, but no baseline, so the frame can't be checked | `CONFIRMED` | `reconciliation_required` | | The same, but the executor approved its own write | `CONFIRMED` | `rejected_policy` | -| Nothing that matches | `REFUTED` | `halted_before_effect` | -| A duplicate, a wrong value, only some of the effects, or another record that vanished or changed | `REFUTED` | `reconciliation_required` | +| Nothing that matches, and nothing else in the read set moved | `REFUTED` | `halted_before_effect` | +| A duplicate, a wrong value, only some of the effects, or a record outside the delta that was added, vanished or changed | `REFUTED` | `reconciliation_required` | | The store could not be read, no baseline, or a separate-session read | `INDETERMINATE` or `CONFIRMED` | `reconciliation_required` | Only a read through the system of record produces `verified`. A screen diff --git a/packages/flow/openadapt_flow/countersign/document.py b/packages/flow/openadapt_flow/countersign/document.py index 316462a75..a1e21b2e2 100644 --- a/packages/flow/openadapt_flow/countersign/document.py +++ b/packages/flow/openadapt_flow/countersign/document.py @@ -497,6 +497,9 @@ class CountersignReport: #: The check level the receipt states, or None on a receipt #: issued before ``checked_how`` existed. checked_how: Optional[str] = None + #: Which parts of the claim behind ``verified`` were checked (delta, + #: frame, source), or None on a receipt issued before ``checked``. + checked: Optional[dict[str, Any]] = None seal: dict[str, Any] = field(default_factory=dict) evidence: dict[str, Any] = field(default_factory=dict) statement: dict[str, Any] = field(default_factory=dict) @@ -620,6 +623,11 @@ def verify_countersigned( outcome=receipt.get("outcome"), oracle_tier=receipt.get("oracle_tier"), checked_how=receipt.get("checked_how"), + checked=( + dict(receipt["checked"]) + if isinstance(receipt.get("checked"), dict) + else None + ), accept_non_flow=accept_non_flow, ) if schema not in (EXECUTE_RECEIPT_SCHEMA, EFFECT_RECEIPT_SCHEMA): @@ -641,6 +649,16 @@ def verify_countersigned( ) if stated_tier is None or stated_tier != receipt.get("verification_tier"): report.problems.append("checked_how_mismatch") + if schema == EFFECT_RECEIPT_SCHEMA and "checked" in receipt: + parts = receipt.get("checked") + if not isinstance(parts, dict) or set(parts) != {"delta", "frame", "source"}: + report.problems.append("checked_malformed") + elif receipt.get("outcome") in _SEALABLE and ( + parts.get("delta") != "held" + or parts.get("frame") != "held" + or parts.get("source") not in ("held", "instruction") + ): + report.problems.append("verified_without_every_part") # Seal signature: openadapt_types.seal, unchanged. try: keys = parse_key_registry(list(registry)) @@ -677,6 +695,7 @@ def verify_countersigned( != receipt.get("verification_tier") or (ev.get("readback") or {}).get("checked_how") != receipt.get("checked_how") + or ev.get("checked") != receipt.get("checked") ) report.evidence = { "status": "evidence_mismatch" if mismatch else "bound", diff --git a/packages/flow/openadapt_flow/countersign/render.py b/packages/flow/openadapt_flow/countersign/render.py index e05988246..92594d9d4 100644 --- a/packages/flow/openadapt_flow/countersign/render.py +++ b/packages/flow/openadapt_flow/countersign/render.py @@ -81,6 +81,26 @@ def executor_line(report: CountersignReport) -> str: } +_PART_TEXT = { + "delta": "the declared change", + "frame": "nothing else in the slice changed", + "source": "the inputs match their source", +} + + +def checked_line(checked: dict[str, Any]) -> str: + """One clause per part of the claim, in the receipt's own words.""" + + out = [] + for part in ("delta", "frame", "source"): + state = str(checked.get(part)) + if part == "source" and state == "instruction": + out.append("source: the inputs were the authorized instruction") + else: + out.append(f"{part} ({_PART_TEXT[part]}): {state}") + return "; ".join(out) + + def tier_line(report: CountersignReport) -> str: if report.receipt_schema == EXECUTE_RECEIPT_SCHEMA: return f"not stated by Execute v1 (oracle_tier {report.oracle_tier})" @@ -163,6 +183,11 @@ def render_report(report: CountersignReport, *, show_names: bool = True) -> str: if report.checked_how is not None else [] ), + *( + [("Checked", checked_line(report.checked))] + if report.checked is not None + else [] + ), ("Effect verdict", str(report.effect_verdict or "not stated (Execute v1)")), ("Outcome", str(report.outcome)), ("Receipt schema", str(report.receipt_schema)), diff --git a/packages/flow/openadapt_flow/qualify/config.py b/packages/flow/openadapt_flow/qualify/config.py index 2ec3c4671..50c408375 100644 --- a/packages/flow/openadapt_flow/qualify/config.py +++ b/packages/flow/openadapt_flow/qualify/config.py @@ -400,6 +400,12 @@ def sub(match: re.Match[str]) -> str: return value +#: ``openadapt.verify/v1`` sections only ``verify-run`` reads. Qualify accepts +#: and ignores them, so one file serves both (the mirror of verify-run's +#: ``QUALIFY_ONLY_SECTIONS``). +VERIFY_RUN_ONLY_SECTIONS: tuple[str, ...] = ("frame",) + + def parse_verify_config( data: Any, *, @@ -410,6 +416,7 @@ def parse_verify_config( if not isinstance(data, dict): raise QualifyConfigError("verify.yaml must be a YAML mapping") + data = {k: v for k, v in data.items() if k not in VERIFY_RUN_ONLY_SECTIONS} missing: list[str] = [] data = _expand(data, missing, env) if missing: diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py new file mode 100644 index 000000000..15f3fd3db --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -0,0 +1,653 @@ +"""Frame conditions: prove that nothing outside the declared delta changed. + +An effect contract says what a save must do. It says nothing about what the +save must not do, so a check built from it can't see an edit to a record it +never names (OpenAdaptAI/OpenAdapt#1365: another chart edited in place, 5 of 5 +saves reported verified). A frame condition closes that gap the way a +procedure's ``modifies`` clause does in program verification: the contract +declares the records and fields the save may change, and everything else in +the slice must be unchanged. + +The slice is the verifier's read set: the records it returns before and after +the save, through the same read-only channel the effect check uses. The judge +runs once for the whole contract, not once per effect, so two effects on one +read set never flag each other's targets. + +Every change outside the declared records is attributed: + +- ``other``: an audit entry, or the record's own modified-by field with a + moved version, names an actor that is not the bot. A concurrent edit by + another user. Listed, and not counted against the save. +- ``self``: the bot's own user or session id. Collateral. +- ``unattributed``: nothing names an actor. Collateral too: a change nobody + can explain is never assumed to be someone else's. + +An audit entry by the bot for a record no effect declares is collateral even +when the read shows nothing: the log shows a write the contract doesn't cover. + +Only evidence the bot can't write counts (the safety review of #1443): + +- attribution comes from the app's audit log, or from record fields the + frame declares as server-set (``server_set``); a field the run itself + writes is never evidence of who changed a record, so attribution that + names one is refused and every change stays unattributed; +- the run's own identities (``self`` and the executor's id) are never + "another actor"; +- another actor's change to a record the save declares is never excused: + the saved value may no longer stand, so it blocks like collateral. Only + changes outside the declared delta can be concurrent edits. + +The judge reads nothing itself; callers pass the snapshots. Fail-safe: a +missing baseline, an unreadable read, a record without its identity field or +an unresolved self id makes the frame ``not_checked``, never ``held``. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectKind, + EffectState, + ValueExpr, + record_matches, + stable_id, +) + +FrameStatus = Literal["held", "refuted", "not_checked"] +Attribution = Literal["other", "self", "unattributed"] +ChangeKind = Literal["added", "removed", "changed"] + +#: How many changes one reason line names before it stops listing kinds. +_SAMPLE_LIMIT = 5 + + +class FrameAttribution(BaseModel): + """Where the judge learns who changed a record. + + ``audit_read``: a second read-only read (``read``, the same shape as an + ``effects:`` verifier section) returns the app's audit log or change + feed. Entries present after the save and not before are the window's + entries. ``record_field`` names the entry field that refers to a record + (compared with :attr:`FrameSpec.identity_field`), ``actor_field`` the + user or session. + + ``record_field`` attribution: each record carries its last modifier in + ``actor_field`` and, optionally, a version or ETag in ``version_field``. + A changed record whose version did not move is unattributed. + """ + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + kind: Literal["audit_read", "record_field"] + #: Entry (or record) field naming who made the change. Dotted paths + #: reach nested fields (``meta.lastUpdatedBy``). + actor_field: str = Field(min_length=1) + #: The bot's own user or session ids. Literals or run parameters. + self_actors: list[ValueExpr] = Field(alias="self", min_length=1) + #: ``audit_read`` only: the entry field naming the changed record. + record_field: Optional[str] = None + #: ``audit_read`` only: the read-only read of the log, as an ``effects:`` + #: verifier section. Built by the caller (``verify-run``). + read: Optional[dict[str, Any]] = None + #: ``record_field`` only: a version, revision or ETag on each record. + version_field: Optional[str] = None + #: ``record_field`` only: the record fields this app sets on the server + #: and no user or API client can write. ``actor_field`` and + #: ``version_field`` must be among them: declaring them is the per-app + #: statement that they are evidence. + server_set: list[str] = Field(default_factory=list) + + @model_validator(mode="before") + @classmethod + def _coerce_self(cls, data: Any) -> Any: + if isinstance(data, dict): + key = "self" if "self" in data else "self_actors" + values = data.get(key) + if isinstance(values, list): + data = dict(data) + data[key] = [ + ValueExpr(literal=v) if isinstance(v, str) else v for v in values + ] + return data + + @model_validator(mode="after") + def _shape(self) -> "FrameAttribution": + if self.kind == "audit_read": + if not self.record_field or self.read is None: + raise ValueError("audit_read attribution needs record_field and read") + if self.version_field is not None: + raise ValueError("version_field applies to record_field attribution") + if self.server_set: + raise ValueError("server_set applies to record_field attribution") + else: + if self.read is not None or self.record_field is not None: + raise ValueError( + "record_field attribution reads the records themselves; " + "read and record_field apply to audit_read" + ) + needed = {self.actor_field} | ( + {self.version_field} if self.version_field else set() + ) + missing = sorted(needed - set(self.server_set)) + if missing: + raise ValueError( + "record_field attribution needs its fields declared server-set " + "(fields no user or API client can write): " + ", ".join(missing) + ) + return self + + +class FrameSpec(BaseModel): + """``frame:`` in ``verify.yaml``: the slice that must otherwise not move.""" + + model_config = ConfigDict(extra="forbid") + + #: Records are compared by this field. Unset: :func:`stable_id` (``id`` + #: when present, else the whole record). + identity_field: Optional[str] = None + #: Fields that change on their own on every record (a last-seen time). + ignore_fields: list[str] = Field(default_factory=list) + #: Fields of a pre-existing target record the save may change. Unset: + #: the fields the effects name. + target_fields: Optional[list[str]] = None + attribution: Optional[FrameAttribution] = None + + @model_validator(mode="after") + def _identity_for_audit(self) -> "FrameSpec": + if ( + self.attribution is not None + and self.attribution.kind == "audit_read" + and not self.identity_field + ): + raise ValueError( + "audit_read attribution compares log entries with records by " + "identity_field; set it" + ) + return self + + def digest(self) -> str: + """SHA-256 of the frame as declared, for binding into evidence.""" + + payload = self.model_dump(mode="json", by_alias=True) + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +@dataclass +class FrameChange: + """One record outside the declared delta that moved.""" + + surface: str + change: ChangeKind + attribution: Attribution + #: Field names that changed (names only, never values). + fields: list[str] = field(default_factory=list) + #: True for a declared record that changed outside its target fields. + on_target: bool = False + #: True for a bot audit entry on an undeclared record the read shows as + #: unchanged. + audit_only: bool = False + #: True when another actor changed a record this save declares. + other_on_target: bool = False + #: In memory only, for the operator's console; never put in a receipt. + key: Any = None + + +@dataclass +class FrameVerdict: + status: FrameStatus + reason: str + attribution: str = "none" + records_compared: int = 0 + collateral: list[FrameChange] = field(default_factory=list) + concurrent: list[FrameChange] = field(default_factory=list) + + @property + def held(self) -> bool: + return self.status == "held" + + def evidence(self, spec: FrameSpec) -> dict[str, Any]: + """Closed fields only: counts and field names, no records or actors.""" + + return { + "status": self.status, + "frame_sha256": spec.digest(), + "attribution": self.attribution, + "records_compared": self.records_compared, + "collateral_count": len(self.collateral), + "concurrent_count": len(self.concurrent), + "collateral_fields": sorted({f for c in self.collateral for f in c.fields}), + "collateral_kinds": sorted({c.change for c in self.collateral}), + "unattributed_count": sum( + 1 for c in self.collateral if c.attribution == "unattributed" + ), + "target_changed_by_other": sum( + 1 for c in self.collateral if c.other_on_target + ), + } + + +def _not_checked(reason: str, attribution: str = "none") -> FrameVerdict: + return FrameVerdict(status="not_checked", reason=reason, attribution=attribution) + + +def get_path(record: Mapping[str, Any], path: str) -> Any: + """``record[a][b]`` for ``path`` ``a.b``; None when any step is missing.""" + + if path in record: + return record[path] + node: Any = record + for part in path.split("."): + if not isinstance(node, Mapping) or part not in node: + return None + node = node[part] + return node + + +def declared_selectors(effect: Effect) -> list[dict[str, str]]: + """The selectors naming the records ``effect`` may write. + + An ``exact_new_set``'s ``match`` is a scope, not a target, so its members + are the declared records. An idempotency key narrows the selector. + """ + + def plain(selector: Mapping[str, Any]) -> dict[str, str]: + out = {k: str(v) for k, v in selector.items()} + if effect.idempotency_key is not None: + out[effect.key_field] = str(effect.idempotency_key) + return out + + if effect.kind is EffectKind.EXACT_NEW_SET: + return [plain(s) for s in effect.new_records] + return [plain(effect.match)] + + +def declared_fields(effects: Sequence[Effect]) -> set[str]: + """Every field an effect names: selector keys and read-back fields.""" + + names: set[str] = set() + for effect in effects: + for selector in declared_selectors(effect): + names.update(selector) + if effect.field: + names.add(effect.field) + return names + + +def _same(before: Any, after: Any) -> bool: + from openadapt_flow.runtime.effects._common import _same_value + + return _same_value(before, after) + + +def _changed_fields( + before: Mapping[str, Any], after: Mapping[str, Any], ignore: set[str] +) -> list[str]: + return sorted( + k + for k in set(before) | set(after) + if k not in ignore and not _same(before.get(k), after.get(k)) + ) + + +def _without(record: Mapping[str, Any], ignore: set[str]) -> dict[str, Any]: + return {k: v for k, v in record.items() if k not in ignore} + + +def _resolve_self( + spec: FrameAttribution, params: Mapping[str, object] +) -> Optional[set[str]]: + out: set[str] = set() + for expr in spec.self_actors: + value = expr.resolve(params) + if value is None or value == "": + return None + out.add(value) + return out + + +def judge_frame( + spec: FrameSpec, + effects: Sequence[Effect], + slices: Mapping[str, tuple[EffectState, Optional[list[dict[str, Any]]]]], + *, + params: Mapping[str, object] = {}, + routes: Optional[Mapping[str, Sequence[str]]] = None, + audit_before: Optional[list[dict[str, Any]]] = None, + audit_after: Optional[list[dict[str, Any]]] = None, + run_actors: Sequence[str] = (), +) -> FrameVerdict: + """Judge the frame of ``effects`` over ``slices``. + + Args: + spec: The declared frame. + effects: The contract's effects, already bound to the run's params. + slices: Surface name -> (baseline, records read after the save, or + ``None`` when unreadable). One verifier is one slice named ``""``. + params: Run params, to resolve the bot's own ids. + routes: Effect index (as ``str``) -> the surface names it declares + records in (its own surface and any view of the same read). Unset: + every effect belongs to every slice. + audit_before: ``audit_read`` entries read with the baseline. + audit_after: ``audit_read`` entries read after the save. + run_actors: More ids of this run (the executor's), never "another + actor". + """ + + attribution = spec.attribution + mode: str = attribution.kind if attribution is not None else "none" + self_ids: set[str] = set() + if attribution is not None: + resolved = _resolve_self(attribution, params) + if resolved is None: + return _not_checked( + "the frame's own-actor ids reference a parameter the run did " + "not supply; changes can't be attributed", + mode, + ) + self_ids = resolved | {str(a) for a in run_actors if str(a)} + + ignore = set(spec.ignore_fields) + if attribution is not None and attribution.kind == "record_field": + # Server-set bookkeeping moves on every write; it says who, not what. + ignore |= set(attribution.server_set) + written = declared_fields(effects) + may_change = set(spec.target_fields) if spec.target_fields is not None else written + if attribution is not None and attribution.kind == "record_field": + # A field this run writes (or any run parameter names) is not evidence + # of who changed a record: refuse the attribution, fail closed. + evidence = {attribution.actor_field} | ( + {attribution.version_field} if attribution.version_field else set() + ) + if evidence & (written | may_change | set(params)): + attribution = None + mode = "attribution_refused" + + def key_of(record: Mapping[str, Any]) -> Any: + if spec.identity_field: + value = record.get(spec.identity_field) + return None if value is None else str(value) + return stable_id(_without(record, ignore)) + + window_entries: Optional[list[dict[str, Any]]] = None + if mode == "audit_read": + if audit_before is not None and audit_after is not None: + seen = {stable_id(e) for e in audit_before} + window_entries = [e for e in audit_after if stable_id(e) not in seen] + + def other_on_target( + key: Any, + before: Optional[Mapping[str, Any]], + after: Optional[Mapping[str, Any]], + ) -> bool: + """Did another actor change this declared record in the window?""" + + if attribution is None: + return False + if attribution.kind == "audit_read": + return window_entries is not None and any( + str(get_path(e, attribution.record_field or "")) == str(key) + and get_path(e, attribution.actor_field) not in (None, "") + and str(get_path(e, attribution.actor_field)) not in self_ids + for e in window_entries + ) + if after is None: + return False + if before is not None and not _changed_fields(before, after, ignore): + return False # untouched in the window + actor = get_path(after, attribution.actor_field) + return actor not in (None, "") and str(actor) not in self_ids + + def attribute( + key: Any, + before: Optional[Mapping[str, Any]], + after: Optional[Mapping[str, Any]], + ) -> Attribution: + if attribution is None: + return "unattributed" + if attribution.kind == "audit_read": + if window_entries is None: + return "unattributed" + actors = { + str(get_path(e, attribution.actor_field)) + for e in window_entries + if str(get_path(e, attribution.record_field or "")) == str(key) + and get_path(e, attribution.actor_field) not in (None, "") + } + if actors & self_ids: + return "self" + return "other" if actors else "unattributed" + # record_field: the record says who last changed it. + if after is None: + return "unattributed" # a removed record can't say who removed it + actor = get_path(after, attribution.actor_field) + if actor in (None, ""): + return "unattributed" + if str(actor) in self_ids: + return "self" + if attribution.version_field: + new_version = get_path(after, attribution.version_field) + old_version = ( + None if before is None else get_path(before, attribution.version_field) + ) + if new_version in (None, "") or ( + before is not None and _same(old_version, new_version) + ): + return "unattributed" + elif before is not None and _same( + get_path(before, attribution.actor_field), actor + ): + # Same last modifier as before the save: nothing says the other + # actor made this change rather than an earlier one. + return "unattributed" + return "other" + + collateral: list[FrameChange] = [] + concurrent: list[FrameChange] = [] + declared_keys: set[str] = set() + compared = 0 + + def note(change: FrameChange) -> None: + if change.attribution == "other": + concurrent.append(change) + else: + collateral.append(change) + + for name, (before_state, after_records) in slices.items(): + if not before_state.reachable: + return _not_checked( + "no readable baseline, so what else changed can't be known", mode + ) + if after_records is None: + return _not_checked("the slice could not be read after the save", mode) + mine = [ + effect + for index, effect in enumerate(effects) + if routes is None or name in routes.get(str(index), (name,)) + ] + selectors = [s for effect in mine for s in declared_selectors(effect)] + + def declared(record: Mapping[str, Any]) -> bool: + return any(record_matches(dict(record), s) for s in selectors) + + before_map: dict[Any, dict[str, Any]] = {} + after_map: dict[Any, dict[str, Any]] = {} + for label, records, into in ( + ("baseline", before_state.records, before_map), + ("post-save", after_records, after_map), + ): + for record in records: + key = key_of(record) + if key is None: + return _not_checked( + f"a {label} record carries no {spec.identity_field!r}; " + "the slice can't be compared record by record", + mode, + ) + if key in into and spec.identity_field: + return _not_checked( + f"two {label} records share one {spec.identity_field!r}; " + "the slice can't be compared record by record", + mode, + ) + into[key] = dict(record) + compared += len(set(before_map) | set(after_map)) + + for key in set(before_map) | set(after_map): + before = before_map.get(key) + after = after_map.get(key) + is_declared = (before is not None and declared(before)) or ( + after is not None and declared(after) + ) + if is_declared: + declared_keys.add(str(key)) + if other_on_target(key, before, after): + # The saved value may not stand: never excused. + collateral.append( + FrameChange( + name, + "changed", + "other", + _changed_fields(before or {}, after or {}, ignore), + on_target=True, + other_on_target=True, + key=key, + ) + ) + continue + if before is None or after is None: + continue # a new or removed target: the delta check's case + stray = [ + f + for f in _changed_fields(before, after, ignore) + if f not in may_change + ] + if stray: + note( + FrameChange( + name, + "changed", + attribute(key, before, after), + stray, + on_target=True, + key=key, + ) + ) + continue + if before is None and after is not None: + note( + FrameChange( + name, + "added", + attribute(key, None, after), + sorted(_without(after, ignore)), + key=key, + ) + ) + elif after is None and before is not None: + note( + FrameChange(name, "removed", attribute(key, before, None), key=key) + ) + elif before is not None and after is not None: + fields = _changed_fields(before, after, ignore) + if fields: + note( + FrameChange( + name, + "changed", + attribute(key, before, after), + fields, + key=key, + ) + ) + + if mode == "audit_read": + if window_entries is None: + # The log could not be read. Every change stays unattributed, and + # the bot's own log entries can't be checked: say so. + mode = "audit_read_unavailable" + else: + moved = {str(c.key) for c in collateral + concurrent} + for entry in window_entries: + actor = get_path(entry, attribution.actor_field) # type: ignore[union-attr] + ref = get_path(entry, attribution.record_field or "") # type: ignore[union-attr] + if actor is None or str(actor) not in self_ids: + continue + if str(ref) in declared_keys or str(ref) in moved: + continue + moved.add(str(ref)) + collateral.append( + FrameChange("", "changed", "self", audit_only=True, key=ref) + ) + + verdict = FrameVerdict( + status="refuted" if collateral else "held", + reason="", + attribution=mode, + records_compared=compared, + collateral=collateral, + concurrent=concurrent, + ) + verdict.reason = _reason(verdict) + return verdict + + +def _reason(verdict: FrameVerdict) -> str: + parts: list[str] = [] + if verdict.collateral: + by_kind: dict[str, int] = {} + for change in verdict.collateral: + label = ( + "written by this run per the audit log" + if change.audit_only + else "saved record changed by another actor (the saved value may " + "not stand)" + if change.other_on_target + else f"{change.change} outside the declared fields of a target" + if change.on_target + else change.change + ) + by_kind[label] = by_kind.get(label, 0) + 1 + kinds = ", ".join(f"{n} {k}" for k, n in sorted(by_kind.items())) + fields = sorted({f for c in verdict.collateral for f in c.fields}) + unattributed = sum( + 1 for c in verdict.collateral if c.attribution == "unattributed" + ) + parts.append( + f"{len(verdict.collateral)} record(s) outside the declared delta " + f"moved ({kinds}" + + ( + f"; field(s): {', '.join(fields[:_SAMPLE_LIMIT])}" + f"{'' if len(fields) <= _SAMPLE_LIMIT else ', ...'}" + if fields + else "" + ) + + f"); {unattributed} of them no actor explains -- collateral, a " + "person should reconcile before anything is retried or undone" + ) + else: + parts.append( + f"nothing outside the declared delta changed across " + f"{verdict.records_compared} record(s)" + ) + if verdict.concurrent: + parts.append( + f"{len(verdict.concurrent)} change(s) attributed to another actor " + "(concurrent edits, not counted against this save)" + ) + if verdict.attribution == "attribution_refused": + parts.append( + "attribution refused: it names a field this run writes, which is no " + "evidence of who changed a record" + ) + if verdict.attribution == "audit_read_unavailable": + parts.append("the audit log could not be read, so no change was attributed") + return "; ".join(parts) diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index e40332f98..8af3cb2de 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -73,6 +73,7 @@ EffectVerdict, Verdict, ) +from openadapt_flow.runtime.effects.frame import FrameSpec, FrameVerdict, judge_frame from openadapt_flow.verification import ( EffectContractMutationError, VerificationTier, @@ -134,6 +135,16 @@ "independent_system_of_record": 4, } +#: Where the baseline keeps the audit log read for frame attribution. Not a +#: surface name: those are plain identifiers. +FRAME_AUDIT_DETAIL = "openadapt:frame_audit" + +#: ``checked`` (``docs/VERIFICATION_COVERAGE.md``): which parts of the claim +#: behind ``verified`` were checked, and how each came out. +CHECKED_PARTS: tuple[str, ...] = ("delta", "frame", "source") +CHECKED_VALUES = ("held", "refuted", "not_checked") +SOURCE_VALUES = CHECKED_VALUES + ("instruction",) + #: Receipt fields beyond Execute v1. The last two are optional. _NEW_RECEIPT_FIELDS = ( "executor_kind", @@ -188,6 +199,10 @@ class VerifyConfig(BaseModel): verifier: EffectsConfig #: Same shape as ``effect_contract.yaml`` ``effects:``. effects: list[Effect] = Field(min_length=1) + #: What else in the read set must stay unchanged, and who may change it + #: (``runtime/effects/frame.py``). Unset: the default frame (the whole + #: read set, compared by ``id``, no attribution). + frame: Optional[FrameSpec] = None def load_verify_config(path: str | Path) -> VerifyConfig: @@ -286,6 +301,17 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo ) if cfg.verifier.candidates: raise VerifyRunError("verify-run does not support verifier candidates yet") + attribution = cfg.frame.attribution if cfg.frame is not None else None + if attribution is not None and attribution.read is not None: + try: + audit = EffectsConfig.model_validate(attribution.read) + except Exception as exc: + raise VerifyRunError(f"invalid frame.attribution.read: {exc}") from exc + if audit.kind in _NOT_INDEPENDENT or audit.surfaces or audit.candidates: + raise VerifyRunError( + "frame.attribution.read must be one read-only system-of-record " + "read (rest, sql, fhir, graphql, ...)" + ) for effect in cfg.effects: if effect.needs_operator_confirmation: raise VerifyRunError( @@ -317,6 +343,136 @@ def _audit_effect(effect: Effect, bound: Effect) -> Effect: return bound.model_copy(update={"forbid_collateral_change": True}) +def _frame_effect(effect: Effect, bound: Effect) -> Effect: + """The effect as verify-run judges it when the frame is judged. + + The frame judge covers every record outside the declared delta, once for + the whole contract, so the per-effect collateral rules would only repeat + it -- and, judged one effect at a time, flag another effect's target. They + stay on only where the file sets them explicitly. + """ + + if "forbid_collateral_loss" in effect.model_fields_set: + return bound + return bound.model_copy(update={"forbid_collateral_loss": False}) + + +def _post_records(state: Any) -> Optional[list[dict[str, Any]]]: + if not isinstance(state, EffectState) or not state.reachable: + return None + return list(state.records) + + +def _audit_verifier(cfg: VerifyConfig, params: Mapping[str, object]) -> Any: + attribution = cfg.frame.attribution if cfg.frame is not None else None + if attribution is None or attribution.read is None: + return None + try: + return build_effect_verifier( + EffectsConfig.model_validate(attribution.read), params + ) + except ValueError as exc: + raise VerifyRunError(f"cannot build the audit read: {exc}") from exc + + +def _read_audit(verifier: Any) -> Optional[list[dict[str, Any]]]: + if verifier is None: + return None + try: + return _post_records(verifier.capture_pre_state()) + except Exception: # noqa: BLE001 - an unreadable log attributes nothing + return None + + +def judge_run_frame( + cfg: VerifyConfig, + verifier: Any, + before: EffectState, + bound_effects: Sequence[Effect], + params: Mapping[str, object], + *, + audit_verifier: Any = None, + run_actors: Sequence[str] = (), +) -> FrameVerdict: + """Read the slice after the save and judge the frame against ``before``.""" + + from openadapt_flow.runtime.effects.surfaces import surface_of + + spec = cfg.frame or FrameSpec() + try: + capture = getattr(verifier, "capture_post_state", None) or ( + verifier.capture_pre_state + ) + after = capture() + except Exception: # noqa: BLE001 - unreadable -> not checked + after = None + routed = getattr(verifier, "verifiers", None) + slices: dict[str, tuple[EffectState, Optional[list[dict[str, Any]]]]] = {} + routes: Optional[dict[str, list[str]]] = None + if isinstance(routed, dict) and routed: + # Surfaces that differ only in dropped fields (a content twin) are + # views of one read: a record an effect declares in one is declared + # in each. + def read_of(name: str) -> str: + sub = cfg.verifier.surfaces.get(name) + if sub is None: + return name + shape = sub.model_dump(mode="json", exclude={"records_drop_fields"}) + return json.dumps(shape, sort_keys=True, default=str) + + views: dict[str, list[str]] = {} + for name in routed: + views.setdefault(read_of(name), []).append(name) + routes = {} + for i, effect in enumerate(bound_effects): + own = surface_of(effect) or "" + routes[str(i)] = views.get(read_of(own), [own]) if own in routed else [own] + for name in routed: + sub = before.detail.get(name, {}) if isinstance(before.detail, dict) else {} + post = ( + after.detail.get(name, {}) + if isinstance(after, EffectState) and isinstance(after.detail, dict) + else {} + ) + slices[name] = ( + EffectState( + substrate=str(sub.get("substrate", "")), + reachable=bool(before.reachable and sub.get("reachable", False)), + records=list(sub.get("records", [])), + ), + list(post.get("records", [])) if post.get("reachable") else None, + ) + else: + slices[""] = (before, _post_records(after)) + audit_before: Optional[list[dict[str, Any]]] = None + audit_after: Optional[list[dict[str, Any]]] = None + if audit_verifier is not None: + stored = ( + before.detail.get(FRAME_AUDIT_DETAIL) + if isinstance(before.detail, dict) + else None + ) + if isinstance(stored, dict) and stored.get("reachable"): + audit_before = list(stored.get("records", [])) + audit_after = _read_audit(audit_verifier) + return judge_frame( + spec, + bound_effects, + slices, + params=params, + routes=routes, + audit_before=audit_before, + audit_after=audit_after, + run_actors=run_actors, + ) + + +def _part(verdict: EffectVerdictLabel, proved: bool) -> str: + if proved: + return "held" + return "refuted" if verdict == "REFUTED" else "not_checked" + + # -- hashing ------------------------------------------------------------------ @@ -447,6 +603,20 @@ def capture_baseline( state = verifier.capture_pre_state() if not isinstance(state, EffectState): raise VerifyRunError("verifier returned no readable baseline") + audit_verifier = _audit_verifier(cfg, params) + if audit_verifier is not None: + entries = _read_audit(audit_verifier) + state = state.model_copy( + update={ + "detail": { + **state.detail, + FRAME_AUDIT_DETAIL: { + "reachable": entries is not None, + "records": entries or [], + }, + } + } + ) payload = { "schema_version": BASELINE_SCHEMA, "execution_id": execution_id, @@ -726,6 +896,7 @@ def verify_run( contract_digest = verify_config_sha256(identity, list(cfg.effects)) verdicts: list[EffectVerdict] = [] rows: list[dict[str, Any]] = [] + bound_effects: list[Effect] = [] weakest: Optional[VerificationTier] = None tiers: list[Optional[VerificationTier]] = [] for effect in cfg.effects: @@ -742,7 +913,12 @@ def verify_run( ) except (KeyError, ValueError) as exc: raise VerifyRunError(f"effect contract cannot bind params: {exc}") from exc - judged = _audit_effect(effect, bound) + judged = ( + _frame_effect(effect, bound) + if baseline is not None + else _audit_effect(effect, bound) + ) + bound_effects.append(bound) tier = verifier_effect_tier(verifier, judged) try: verdict = verify_effect_without_mutation(verifier, judged, before) @@ -773,12 +949,41 @@ def verify_run( "expected_count": verdict.expected_count, "unavailable": verdict.unavailable, "stale": verdict.stale, - "collateral_change_checked": judged.forbid_collateral_change, + # With a baseline the frame judge covers every record outside + # the delta (evidence.frame); without one nothing can. + "collateral_change_checked": judged.forbid_collateral_change + or baseline is not None, } ) - effect_verdict = _aggregate(verdicts) + delta_verdict = _aggregate(verdicts) observed_effect = _observed(verdicts) + # The frame: nothing outside the declared delta moved. Without a baseline + # it can't be known, so it is not checked and the run is not verified. + frame_spec = cfg.frame or FrameSpec() + if baseline is not None: + frame = judge_run_frame( + cfg, + verifier, + before, + bound_effects, + params, + audit_verifier=_audit_verifier(cfg, params), + # The executor's own id is this run, never "another user". + run_actors=[str(run_record.get("executor_id") or "")], + ) + else: + frame = FrameVerdict( + status="not_checked", + reason="no baseline, so what else changed can't be known", + ) + # A record outside the declared delta moved: the save's effect on the + # store is not the declared one, whatever the per-effect reads say. + effect_verdict: EffectVerdictLabel = ( + "REFUTED" if frame.status == "refuted" else delta_verdict + ) + if frame.status == "refuted" and observed_effect in ("absent", "present"): + observed_effect = "conflicting" verification_tier = readback_class(verdicts, tiers) unread = verification_tier == "unchecked" # The check level the weakest read reached (RFC-0002 open question 5): @@ -808,21 +1013,34 @@ def verify_run( delivery_uncertain = False read_proves_write = ( - effect_verdict == "CONFIRMED" + delta_verdict == "CONFIRMED" and verification_tier == "independent_readback" and strong_enough and oracle_tier in (2, 3) ) - if read_proves_write and self_approved is True: + # Inputs come from whoever authorized the task (``--params``): the + # instruction is their source. + checked = { + "delta": _part(delta_verdict, read_proves_write), + "frame": frame.status, + "source": "instruction", + } + claim_holds = ( + read_proves_write + and frame.held + and checked["source"] in ("held", "instruction") + ) + if claim_holds and self_approved is True: # The write landed, but the executor approved its own write: no # oversight separate from itself. Never verified. outcome = ExecuteTerminalOutcomeV1.REJECTED_POLICY - elif read_proves_write: + elif claim_holds: outcome = ExecuteTerminalOutcomeV1.VERIFIED elif ( effect_verdict == "REFUTED" and observed_effect == "absent" and not write_attempted + and frame.status != "refuted" ): # The store shows nothing landed, whatever the executor said. outcome = ExecuteTerminalOutcomeV1.HALTED_BEFORE_EFFECT @@ -832,6 +1050,7 @@ def verify_run( delivery_uncertain = True verified = outcome is ExecuteTerminalOutcomeV1.VERIFIED landed = read_proves_write + frame_held = frame.held evidence = { "schema_version": EFFECT_EVIDENCE_SCHEMA, @@ -870,6 +1089,8 @@ def verify_run( "max_execution_window_s": cfg.max_execution_window_s, "effects": rows, }, + "frame": frame.evidence(frame_spec), + "checked": checked, "verifier": { "name": "openadapt-flow verify-run", "model_calls": 0, @@ -882,7 +1103,8 @@ def verify_run( authorization_passed=verified, # Identity is the contract's record selector, judged in the store. identity_passed=landed, - postcondition_passed=landed, + # The postcondition includes the frame: nothing else moved. + postcondition_passed=landed and frame_held, effect_passed=landed, minimum_effect_strength=EffectStrengthV1(cfg.minimum_effect_strength), observed_effect_strength=( @@ -917,6 +1139,7 @@ def verify_run( receipt["effect_verdict"] = effect_verdict receipt["verification_tier"] = verification_tier receipt["checked_how"] = checked_how + receipt["checked"] = checked if remediates is not None: receipt["remediates"] = remediates if correlation_id is not None: @@ -955,7 +1178,7 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: v1_names | set(_NEW_RECEIPT_FIELDS) | set(_OPTIONAL_RECEIPT_FIELDS) - | {"checked_how"} + | {"checked_how", "checked"} ) if set(receipt) - allowed: return "unknown_field" @@ -991,6 +1214,23 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: ): return "checked_how" outcome = receipt.get("outcome") + # Optional: receipts issued before docs/VERIFICATION_COVERAGE.md lack it. + if "checked" in receipt: + parts = receipt["checked"] + if ( + not isinstance(parts, Mapping) + or set(parts) != set(CHECKED_PARTS) + or parts.get("delta") not in CHECKED_VALUES + or parts.get("frame") not in CHECKED_VALUES + or parts.get("source") not in SOURCE_VALUES + ): + return "checked" + if outcome in ("verified", "rolled_back_verified") and ( + parts["delta"] != "held" + or parts["frame"] != "held" + or parts["source"] not in ("held", "instruction") + ): + return "verified_without_every_part" if outcome in ("verified", "rolled_back_verified") and ( receipt["effect_verdict"] != "CONFIRMED" or receipt["verification_tier"] != "independent_readback" @@ -1021,6 +1261,13 @@ def evidence_problem( or executor.get("kind") != receipt.get("executor_kind") or readback.get("verification_tier") != receipt.get("verification_tier") or readback.get("checked_how") != receipt.get("checked_how") + or evidence.get("checked") != receipt.get("checked") + ): + return "evidence_mismatch" + frame = evidence.get("frame") + if "checked" in receipt and ( + not isinstance(frame, Mapping) + or frame.get("status") != receipt["checked"].get("frame") ): return "evidence_mismatch" if salting.get("scheme") != SALT_SCHEME or not _DIGEST.match( diff --git a/packages/flow/tests/test_frame_conditions.py b/packages/flow/tests/test_frame_conditions.py new file mode 100644 index 000000000..7961a1318 --- /dev/null +++ b/packages/flow/tests/test_frame_conditions.py @@ -0,0 +1,688 @@ +"""Frame conditions: nothing outside the declared delta changed. + +See ``docs/VERIFICATION_COVERAGE.md`` section 1. Records are made up. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest +import requests + +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.runtime.effects import Effect, EffectKind +from openadapt_flow.runtime.effects.effect import EffectState +from openadapt_flow.runtime.effects.frame import ( + FrameAttribution, + FrameSpec, + judge_frame, +) +from openadapt_flow.verify_run import ( + capture_baseline, + parse_verify_config, + receipt_problem, + verify_run, +) + +OTHER = { + "id": 1, + "lname": "Priorchart", + "phone_home": "555-0188", + "by": "clinic2", + "v": 3, +} +TARGET = {"id": 2, "lname": "Newchart", "phone_home": "555-0131", "by": "bot", "v": 1} + + +def _effect(**match: str) -> Effect: + return Effect( + kind=EffectKind.RECORD_WRITTEN, + match=match or {"lname": TARGET["lname"]}, + count_new_only=True, + ) + + +def _state(*records: dict) -> EffectState: + return EffectState( + substrate="t", reachable=True, records=[dict(r) for r in records] + ) + + +def _judge(before, after, effects=None, spec=None, **kw): + return judge_frame( + spec or FrameSpec(), + effects or [_effect()], + {"": (before, after)}, + **kw, + ) + + +# -- the judge ----------------------------------------------------------------- + + +def test_clean_save_holds() -> None: + verdict = _judge(_state(OTHER), [dict(OTHER), dict(TARGET)]) + assert verdict.status == "held" + assert verdict.records_compared == 2 + + +def test_edit_to_another_record_is_collateral() -> None: + edited = {**OTHER, "phone_home": "555-0199"} + verdict = _judge(_state(OTHER), [edited, dict(TARGET)]) + assert verdict.status == "refuted" + assert verdict.collateral[0].fields == ["phone_home"] + assert "phone_home" in verdict.reason + + +def test_an_undeclared_addition_is_collateral() -> None: + stray = {"id": 9, "lname": "Junk", "phone_home": "", "by": "bot", "v": 1} + verdict = _judge(_state(OTHER), [dict(OTHER), dict(TARGET), stray]) + assert verdict.status == "refuted" + assert verdict.collateral[0].change == "added" + + +def test_a_removal_is_collateral() -> None: + verdict = _judge(_state(OTHER), [dict(TARGET)]) + assert verdict.status == "refuted" + assert verdict.collateral[0].change == "removed" + + +def test_no_baseline_is_not_checked() -> None: + before = EffectState(substrate="t", reachable=False) + assert _judge(before, [dict(TARGET)]).status == "not_checked" + + +def test_unreadable_after_is_not_checked() -> None: + assert _judge(_state(OTHER), None).status == "not_checked" + + +def test_two_effects_on_one_read_do_not_flag_each_other() -> None: + # An update to an existing record plus a new one. Per-effect collateral + # rules flag the other effect's target; the frame declares both. + updated = {**OTHER, "phone_home": "555-0100"} + effects = [ + _effect(lname=TARGET["lname"]), + Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"lname": OTHER["lname"], "phone_home": "555-0100"}, + ), + ] + spec = FrameSpec(ignore_fields=["by", "v"]) + verdict = _judge(_state(OTHER), [updated, dict(TARGET)], effects, spec) + assert verdict.status == "held", verdict.reason + + +def test_a_target_may_change_only_in_its_declared_fields() -> None: + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"lname": OTHER["lname"], "phone_home": "555-0100"}, + ) + changed = {**OTHER, "phone_home": "555-0100", "street": "1 Elsewhere"} + spec = FrameSpec(ignore_fields=["by", "v"]) + verdict = _judge(_state({**OTHER, "street": "9 Road"}), [changed], [effect], spec) + assert verdict.status == "refuted" + assert verdict.collateral[0].on_target + assert verdict.collateral[0].fields == ["street"] + + +def test_ignored_fields_do_not_count() -> None: + touched = {**OTHER, "v": 4} + spec = FrameSpec(ignore_fields=["v"]) + assert _judge(_state(OTHER), [touched, dict(TARGET)], spec=spec).status == "held" + + +def test_missing_identity_field_is_not_checked() -> None: + spec = FrameSpec(identity_field="pid") + assert _judge(_state(OTHER), [dict(OTHER)], spec=spec).status == "not_checked" + + +# -- attribution: record_field ------------------------------------------------- + + +def _record_field(**kw) -> FrameSpec: + return FrameSpec( + attribution=FrameAttribution( + kind="record_field", + actor_field="by", + version_field="v", + server_set=["by", "v"], + **{"self": ["bot"]}, + **kw, + ) + ) + + +def test_another_users_edit_with_a_new_version_is_concurrent() -> None: + edited = {**OTHER, "phone_home": "555-0199", "by": "clinic3", "v": 4} + verdict = _judge(_state(OTHER), [edited, dict(TARGET)], spec=_record_field()) + assert verdict.status == "held" + assert len(verdict.concurrent) == 1 + + +def test_the_bots_own_edit_is_collateral() -> None: + edited = {**OTHER, "phone_home": "555-0199", "by": "bot", "v": 4} + verdict = _judge(_state(OTHER), [edited, dict(TARGET)], spec=_record_field()) + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "self" + + +def test_a_change_without_a_new_version_is_unattributed() -> None: + # A direct database write: content moved, the version and modifier did not. + edited = {**OTHER, "phone_home": "555-0199"} + verdict = _judge(_state(OTHER), [edited, dict(TARGET)], spec=_record_field()) + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "unattributed" + + +def test_self_ids_bind_to_run_params() -> None: + spec = FrameSpec( + attribution=FrameAttribution( + kind="record_field", + actor_field="by", + server_set=["by"], + **{"self": [{"param": "bot_user"}]}, + ) + ) + edited = {**OTHER, "phone_home": "555-0199", "by": "bot"} + assert ( + _judge(_state(OTHER), [edited, dict(TARGET)], spec=spec).status == "not_checked" + ) + verdict = _judge( + _state(OTHER), [edited, dict(TARGET)], spec=spec, params={"bot_user": "bot"} + ) + assert verdict.status == "refuted" + + +# -- attribution: audit_read --------------------------------------------------- + + +def _audit_spec() -> FrameSpec: + return FrameSpec( + identity_field="id", + attribution=FrameAttribution( + kind="audit_read", + actor_field="user", + record_field="record", + read={"kind": "sql"}, + **{"self": ["bot"]}, + ), + ) + + +OLD_ENTRY = {"id": 100, "user": "clinic2", "record": 1} + + +def test_an_audited_edit_by_another_user_is_concurrent() -> None: + edited = {**OTHER, "phone_home": "555-0199"} + entries = [OLD_ENTRY, {"id": 101, "user": "clinic3", "record": 1}] + verdict = _judge( + _state(OTHER), + [edited, dict(TARGET)], + spec=_audit_spec(), + audit_before=[OLD_ENTRY], + audit_after=entries, + ) + assert verdict.status == "held" + assert len(verdict.concurrent) == 1 + + +def test_an_old_log_entry_explains_nothing() -> None: + edited = {**OTHER, "phone_home": "555-0199"} + verdict = _judge( + _state(OTHER), + [edited, dict(TARGET)], + spec=_audit_spec(), + audit_before=[OLD_ENTRY], + audit_after=[OLD_ENTRY], + ) + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "unattributed" + + +def test_the_bot_and_another_user_on_one_record_is_collateral() -> None: + edited = {**OTHER, "phone_home": "555-0199"} + entries = [ + {"id": 101, "user": "clinic3", "record": 1}, + {"id": 102, "user": "bot", "record": 1}, + ] + verdict = _judge( + _state(OTHER), + [edited, dict(TARGET)], + spec=_audit_spec(), + audit_before=[], + audit_after=entries, + ) + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "self" + + +def test_a_bot_log_entry_on_an_undeclared_record_is_collateral() -> None: + # The read shows nothing changed (an edit and its revert, or a field the + # read doesn't return), but the log says the bot wrote record 1. + entries = [ + {"id": 101, "user": "bot", "record": 1}, + {"id": 102, "user": "bot", "record": 2}, + ] + verdict = _judge( + _state(OTHER), + [dict(OTHER), dict(TARGET)], + spec=_audit_spec(), + audit_before=[], + audit_after=entries, + ) + assert verdict.status == "refuted" + assert [c.audit_only for c in verdict.collateral] == [True] + + +def test_an_unreadable_log_attributes_nothing() -> None: + edited = {**OTHER, "phone_home": "555-0199"} + verdict = _judge( + _state(OTHER), + [edited, dict(TARGET)], + spec=_audit_spec(), + audit_before=[], + audit_after=None, + ) + assert verdict.status == "refuted" + assert verdict.attribution == "audit_read_unavailable" + + +def test_attribution_shape_is_validated() -> None: + with pytest.raises(ValueError): + FrameAttribution(kind="audit_read", actor_field="user", **{"self": ["bot"]}) + with pytest.raises(ValueError): + FrameAttribution(kind="record_field", actor_field="by", **{"self": []}) + with pytest.raises(ValueError): + FrameSpec( + attribution=FrameAttribution( + kind="audit_read", + actor_field="user", + record_field="record", + read={"kind": "sql"}, + **{"self": ["bot"]}, + ) + ) + + +# -- verify-run ---------------------------------------------------------------- + + +@pytest.fixture +def sor(): + url, db, stop = fault_serve() + try: + yield url.rstrip("/"), db + finally: + stop() + + +def _yaml(base: str, **extra: object) -> dict: + return { + "schema_version": "openadapt.verify/v1", + "verifier": {"kind": "rest", "base_url": base, "records_path": "/api/db"}, + "effects": [ + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "note": {"param": "note"}, + }, + "expected_count": 1, + "count_new_only": True, + "timeout_s": 0.5, + } + ], + **extra, + } + + +PARAMS = {"patient_id": "p1", "note": "Follow-up in two weeks"} + + +def _save(base: str) -> None: + requests.post( + f"{base}/api/encounter", + json={"patient_id": "p1", "type": "Triage", "note": PARAMS["note"]}, + timeout=5, + ) + + +def _verify(cfg, tmp_path: Path, baseline: Path | None) -> dict: + return verify_run( + cfg, + params=PARAMS, + run_record={"executor_kind": "agent", "executor_id": "test-agent"}, + authorization_ref="ticket-0000-frame", + baseline_path=baseline, + private_dir=tmp_path / "out", + ) + + +def test_verify_run_clean_save_states_every_part(sor, tmp_path: Path) -> None: + base, db = sor + db.add("p2", "Consult", "Allergy: penicillin") + cfg = parse_verify_config(_yaml(base)) + baseline = tmp_path / "baseline.json" + capture_baseline(cfg, PARAMS, baseline) + _save(base) + result = _verify(cfg, tmp_path, baseline) + receipt = result["receipt"] + assert receipt["outcome"] == "verified" + assert receipt["checked"] == { + "delta": "held", + "frame": "held", + "source": "instruction", + } + assert result["evidence"]["frame"]["status"] == "held" + assert result["evidence"]["frame"]["records_compared"] == 2 + assert receipt_problem(receipt) is None + + +def test_verify_run_without_a_baseline_is_not_verified(sor, tmp_path: Path) -> None: + base, _db = sor + cfg = parse_verify_config(_yaml(base)) + # Absolute count: the delta alone would confirm. + cfg.effects[0] = cfg.effects[0].model_copy(update={"count_new_only": False}) + _save(base) + receipt = _verify(cfg, tmp_path, None)["receipt"] + assert receipt["effect_verdict"] == "CONFIRMED" + assert receipt["checked"]["frame"] == "not_checked" + assert receipt["outcome"] == "reconciliation_required" + + +def test_verify_run_stray_addition_is_refuted(sor, tmp_path: Path) -> None: + base, db = sor + cfg = parse_verify_config(_yaml(base)) + baseline = tmp_path / "baseline.json" + capture_baseline(cfg, PARAMS, baseline) + _save(base) + db.add("p9", "Triage", "a stray note") + result = _verify(cfg, tmp_path, baseline) + assert result["receipt"]["checked"]["delta"] == "held" + assert result["receipt"]["checked"]["frame"] == "refuted" + assert result["receipt"]["effect_verdict"] == "REFUTED" + assert result["receipt"]["outcome"] == "reconciliation_required" + assert result["evidence"]["frame"]["collateral_kinds"] == ["added"] + + +def test_verify_run_attributes_through_an_audit_read(tmp_path: Path) -> None: + """A SQL store with its own audit log: another user's edit is concurrent.""" + + db = tmp_path / "app.sqlite" + con = sqlite3.connect(db) + con.executescript( + "CREATE TABLE chart (id INTEGER PRIMARY KEY, lname TEXT, phone TEXT);" + "CREATE TABLE log (id INTEGER PRIMARY KEY, user TEXT, chart_id INTEGER);" + "INSERT INTO chart VALUES (1, 'Priorchart', '555-0188');" + "INSERT INTO log VALUES (1, 'clinic2', 1);" + ) + con.commit() + + def doc(self_user: str) -> dict: + return { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT id, lname, phone FROM chart", + }, + "effects": [ + { + "kind": "record_written", + "match": {"lname": {"param": "lname"}}, + "count_new_only": True, + "timeout_s": 0.2, + } + ], + "frame": { + "identity_field": "id", + "attribution": { + "kind": "audit_read", + "actor_field": "user", + "record_field": "chart_id", + "self": [self_user], + "read": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT id, user, chart_id FROM log", + }, + }, + }, + } + + params = {"lname": "Newchart"} + + def run(edit_user: str, name: str) -> dict: + cfg = parse_verify_config(doc("bot")) + baseline = tmp_path / f"{name}.json" + capture_baseline(cfg, params, baseline) + con.execute("DELETE FROM chart WHERE id > 1") + con.execute("UPDATE chart SET phone = '555-0188' WHERE id = 1") + con.execute("INSERT INTO chart (lname, phone) VALUES ('Newchart', '555-0131')") + con.execute( + "INSERT INTO log (user, chart_id) VALUES ('bot', last_insert_rowid())" + ) + con.execute("UPDATE chart SET phone = '555-0199' WHERE id = 1") + con.execute("INSERT INTO log (user, chart_id) VALUES (?, 1)", (edit_user,)) + con.commit() + return verify_run( + cfg, + params=params, + run_record={"executor_kind": "agent", "executor_id": "test-agent"}, + authorization_ref="ticket-0000-audit", + baseline_path=baseline, + private_dir=tmp_path / name, + ) + + try: + concurrent = run("clinic3", "concurrent") + assert concurrent["receipt"]["outcome"] == "verified", concurrent["evidence"] + assert concurrent["evidence"]["frame"]["concurrent_count"] == 1 + own = run("bot", "own") + assert own["receipt"]["checked"]["frame"] == "refuted" + assert own["receipt"]["outcome"] == "reconciliation_required" + finally: + con.close() + + +def test_receipt_check_refuses_verified_with_an_unchecked_part( + sor, tmp_path: Path +) -> None: + base, _db = sor + cfg = parse_verify_config(_yaml(base)) + baseline = tmp_path / "baseline.json" + capture_baseline(cfg, PARAMS, baseline) + _save(base) + receipt = _verify(cfg, tmp_path, baseline)["receipt"] + forged = {**receipt, "checked": {**receipt["checked"], "frame": "not_checked"}} + assert receipt_problem(forged) == "verified_without_every_part" + assert receipt_problem({**receipt, "checked": {"delta": "held"}}) == "checked" + + +def test_qualify_accepts_a_frame_section() -> None: + from openadapt_flow.qualify.config import VERIFY_RUN_ONLY_SECTIONS + + assert "frame" in VERIFY_RUN_ONLY_SECTIONS + + +def test_countersign_states_every_part() -> None: + from openadapt_flow.countersign.render import checked_line + + line = checked_line({"delta": "held", "frame": "refuted", "source": "instruction"}) + assert "frame (nothing else in the slice changed): refuted" in line + assert "the inputs were the authorized instruction" in line + + +def test_a_content_twin_is_a_view_of_the_same_read(tmp_path: Path) -> None: + """scaffold-verifier reads each table twice: by id, and by content + (``records_drop_fields: [id]``). The saved row is declared in both.""" + + db = tmp_path / "app.sqlite" + con = sqlite3.connect(db) + con.executescript( + "CREATE TABLE enc (id INTEGER PRIMARY KEY, patient_id TEXT, note TEXT);" + "INSERT INTO enc VALUES (1, 'p2', 'an older note');" + ) + con.commit() + read = {"kind": "sql", "sqlite_database": str(db), "sql_query": "SELECT * FROM enc"} + match = {"patient_id": {"param": "patient_id"}, "note": {"param": "note"}} + cfg = parse_verify_config( + { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "surfaces", + "surfaces": { + "enc": read, + "enc.content": {**read, "records_drop_fields": ["id"]}, + }, + }, + "effects": [ + { + "kind": "record_written", + "match": match, + "count_new_only": True, + "probe": "surface=enc|the saved row", + "timeout_s": 0.2, + }, + { + "kind": "record_written", + "match": {"_none": "-"}, + "expected_count": 0, + "count_new_only": True, + "probe": "surface=enc.content|no row changes", + "timeout_s": 0.2, + }, + ], + } + ) + params = {"patient_id": "p1", "note": "Follow-up"} + baseline = tmp_path / "b.json" + capture_baseline(cfg, params, baseline) + con.execute("INSERT INTO enc (patient_id, note) VALUES ('p1', 'Follow-up')") + con.commit() + result = verify_run( + cfg, + params=params, + run_record={"executor_kind": "script"}, + authorization_ref="ticket-0000-twin", + baseline_path=baseline, + private_dir=tmp_path / "out", + ) + assert result["receipt"]["outcome"] == "verified", result["evidence"]["frame"] + con.execute("UPDATE enc SET note = 'edited' WHERE id = 1") + con.commit() + baseline2 = tmp_path / "b2.json" + con.execute("DELETE FROM enc WHERE id > 1") + con.execute("UPDATE enc SET note = 'an older note' WHERE id = 1") + con.commit() + capture_baseline(cfg, params, baseline2) + con.execute("INSERT INTO enc (patient_id, note) VALUES ('p1', 'Follow-up')") + con.execute("UPDATE enc SET note = 'edited' WHERE id = 1") + con.commit() + edited = verify_run( + cfg, + params=params, + run_record={"executor_kind": "script"}, + authorization_ref="ticket-0000-twin", + baseline_path=baseline2, + private_dir=tmp_path / "out2", + ) + assert edited["receipt"]["checked"]["frame"] == "refuted" + con.close() + + +# -- the safety review's conditions (#1443) ------------------------------------ + + +def test_record_field_attribution_must_declare_its_fields_server_set() -> None: + """(a) Only fields the app sets on the server, declared per app, count.""" + + with pytest.raises(ValueError, match="server-set"): + FrameAttribution( + kind="record_field", + actor_field="by", + version_field="v", + **{"self": ["bot"]}, + ) + with pytest.raises(ValueError, match="server-set"): + FrameAttribution( + kind="record_field", + actor_field="by", + version_field="v", + server_set=["by"], + **{"self": ["bot"]}, + ) + + +def test_attribution_from_a_field_the_run_writes_is_refused() -> None: + """(a) A field a step typed into is no evidence of who changed a record.""" + + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"lname": TARGET["lname"], "by": "bot"}, + count_new_only=True, + ) + edited = {**OTHER, "phone_home": "555-0199", "by": "clinic3", "v": 4} + verdict = _judge(_state(OTHER), [edited, dict(TARGET)], [effect], _record_field()) + assert verdict.attribution == "attribution_refused" + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "unattributed" + # The same field named by a run parameter is refused too. + verdict = _judge( + _state(OTHER), [edited, dict(TARGET)], spec=_record_field(), params={"by": "x"} + ) + assert verdict.attribution == "attribution_refused" + + +def test_the_runs_own_executor_is_never_another_actor() -> None: + """(b) The named user must differ from the run's own identity.""" + + edited = {**OTHER, "phone_home": "555-0199", "by": "agent-7", "v": 4} + verdict = _judge( + _state(OTHER), + [edited, dict(TARGET)], + spec=_record_field(), + run_actors=["agent-7"], + ) + assert verdict.status == "refuted" + assert verdict.collateral[0].attribution == "self" + + +def test_another_users_change_to_the_saved_record_blocks() -> None: + """(c) The saved value may not stand: never excused, record_field.""" + + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"lname": OTHER["lname"], "phone_home": "555-0100"}, + ) + # The bot set phone_home; another user then wrote the same record. + saved = {**OTHER, "phone_home": "555-0100", "by": "clinic3", "v": 4} + verdict = _judge(_state(OTHER), [saved], [effect], _record_field()) + assert verdict.status == "refuted" + assert verdict.collateral[0].other_on_target + assert "may not stand" in verdict.reason + # A record the bot itself wrote last is the delta's business. + own = {**OTHER, "phone_home": "555-0100", "by": "bot", "v": 4} + assert _judge(_state(OTHER), [own], [effect], _record_field()).status == "held" + + +def test_an_audited_edit_of_the_saved_record_by_another_user_blocks() -> None: + """(c) Same rule through the audit log: an entry on the target by another + user blocks, while one on an undeclared record is a concurrent edit.""" + + entries = [ + {"id": 101, "user": "bot", "record": 2}, + {"id": 102, "user": "clinic3", "record": 2}, + ] + verdict = _judge( + _state(OTHER), + [dict(OTHER), dict(TARGET)], + spec=_audit_spec(), + audit_before=[], + audit_after=entries, + ) + assert verdict.status == "refuted" + assert verdict.collateral[0].other_on_target + assert verdict.evidence(_audit_spec())["target_changed_by_other"] == 1 diff --git a/packages/flow/tests/test_verify_run.py b/packages/flow/tests/test_verify_run.py index 1d51dc457..34d3a0d13 100644 --- a/packages/flow/tests/test_verify_run.py +++ b/packages/flow/tests/test_verify_run.py @@ -216,7 +216,10 @@ def test_wrong_patient_is_refuted(cfg, sor, tmp_path): base, _db = sor result = _run(cfg, tmp_path, pre=lambda: _actor_writes(base, fault="wrong_record")) assert result["receipt"]["effect_verdict"] == "REFUTED" - assert result["receipt"]["outcome"] == "halted_before_effect" + # The intended record is absent, but a record was written to another + # patient: the frame sees it, so this is not "nothing happened". + assert result["receipt"]["checked"]["frame"] == "refuted" + assert result["receipt"]["outcome"] == "reconciliation_required" def test_duplicate_write_needs_reconciliation(cfg, sor, tmp_path): @@ -459,6 +462,7 @@ def test_receipt_keeps_every_execute_v1_field(cfg, sor, tmp_path): "executor_ref_sha256", "effect_verdict", "verification_tier", + "checked", "checked_how", } assert receipt["schema_version"] == EFFECT_RECEIPT_SCHEMA