From 522a7686774be80f5d8f3e3b5eb542832dd62e79 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 13:06:01 -0400 Subject: [PATCH 1/3] verify-run: frame conditions -- nothing outside the declared delta changed A record check said a save was right while another chart was edited in place (#1365). The frame judge compares the whole read set before and after the save, once per contract: any record outside the declared records that was added, removed or changed, or a target changed outside its declared fields, is collateral. A change that the app's audit log or a modified-by field with a new version attributes to another user is a concurrent edit and does not count; a change by the bot's own ids or one nothing explains does. - runtime/effects/frame.py: FrameSpec, FrameAttribution (audit_read, record_field), judge_frame. - verify-run: optional frame: section; the audit read is taken with the baseline; the frame judge replaces the per-effect collateral rules when a baseline exists; without one the frame is not_checked and the run is not verified. - Receipt: optional signed field checked {delta, frame, source}; source is 'instruction' (the --params). verified requires every part. evidence.frame carries counts and field names only. - countersign verify shows and checks it; qualify ignores frame:. Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 1 + llms-full.txt | 1 + packages/flow/docs/VERIFY_RUN.md | 92 ++- .../openadapt_flow/countersign/document.py | 19 + .../flow/openadapt_flow/countersign/render.py | 25 + .../flow/openadapt_flow/qualify/config.py | 7 + .../openadapt_flow/runtime/effects/frame.py | 562 ++++++++++++++++++ packages/flow/openadapt_flow/verify_run.py | 245 +++++++- packages/flow/tests/test_frame_conditions.py | 511 ++++++++++++++++ packages/flow/tests/test_verify_run.py | 6 +- 10 files changed, 1441 insertions(+), 28 deletions(-) create mode 100644 packages/flow/openadapt_flow/runtime/effects/frame.py create mode 100644 packages/flow/tests/test_frame_conditions.py 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..fdfe3f5ec 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -80,34 +80,88 @@ 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 + 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. + +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..3113c6c36 --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -0,0 +1,562 @@ +"""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. + +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 + + @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") + 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" + ) + 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 + #: 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" + ), + } + + +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, str]] = None, + audit_before: Optional[list[dict[str, Any]]] = None, + audit_after: Optional[list[dict[str, Any]]] = None, +) -> 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``) -> surface name. 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. + """ + + 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 + + ignore = set(spec.ignore_fields) + may_change = ( + set(spec.target_fields) + if spec.target_fields is not None + else declared_fields(effects) + ) + + 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 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 routes.get(str(index), name) == 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 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 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 == "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..7dabcf500 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,120 @@ 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, +) -> 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, str]] = None + if isinstance(routed, dict) and routed: + routes = { + str(i): surface_of(effect) or "" for i, effect in enumerate(bound_effects) + } + 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, + ) + + +def _part(verdict: EffectVerdictLabel, proved: bool) -> str: + if proved: + return "held" + return "refuted" if verdict == "REFUTED" else "not_checked" + + # -- hashing ------------------------------------------------------------------ @@ -447,6 +587,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 +880,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 +897,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 +933,39 @@ 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), + ) + 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 +995,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 +1032,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 +1071,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 +1085,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 +1121,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 +1160,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 +1196,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 +1243,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..4563aa654 --- /dev/null +++ b/packages/flow/tests/test_frame_conditions.py @@ -0,0 +1,511 @@ +"""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", + **{"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", **{"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 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 From 5fee616dd0ee5763a61a7c46b22ec774db43c72c Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 13:16:08 -0400 Subject: [PATCH 2/3] frame: a content twin is a view of the same read scaffold-verifier reads each table by id and by content (records_drop_fields: [id]). The frame judged the twin as its own slice, so the saved row looked like an undeclared addition there. Surfaces that differ only in dropped fields now share their declared records. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/runtime/effects/frame.py | 9 ++- packages/flow/openadapt_flow/verify_run.py | 22 +++++- packages/flow/tests/test_frame_conditions.py | 78 +++++++++++++++++++ 3 files changed, 101 insertions(+), 8 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py index 3113c6c36..fc6d9e18e 100644 --- a/packages/flow/openadapt_flow/runtime/effects/frame.py +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -287,7 +287,7 @@ def judge_frame( slices: Mapping[str, tuple[EffectState, Optional[list[dict[str, Any]]]]], *, params: Mapping[str, object] = {}, - routes: Optional[Mapping[str, str]] = None, + routes: Optional[Mapping[str, Sequence[str]]] = None, audit_before: Optional[list[dict[str, Any]]] = None, audit_after: Optional[list[dict[str, Any]]] = None, ) -> FrameVerdict: @@ -299,8 +299,9 @@ def judge_frame( 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``) -> surface name. Unset: every effect - belongs to every slice. + 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. """ @@ -402,7 +403,7 @@ def note(change: FrameChange) -> None: mine = [ effect for index, effect in enumerate(effects) - if routes is None or routes.get(str(index), name) == name + if routes is None or name in routes.get(str(index), (name,)) ] selectors = [s for effect in mine for s in declared_selectors(effect)] diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 7dabcf500..73340da0f 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -407,11 +407,25 @@ def judge_run_frame( after = None routed = getattr(verifier, "verifiers", None) slices: dict[str, tuple[EffectState, Optional[list[dict[str, Any]]]]] = {} - routes: Optional[dict[str, str]] = None + routes: Optional[dict[str, list[str]]] = None if isinstance(routed, dict) and routed: - routes = { - str(i): surface_of(effect) or "" for i, effect in enumerate(bound_effects) - } + # 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 = ( diff --git a/packages/flow/tests/test_frame_conditions.py b/packages/flow/tests/test_frame_conditions.py index 4563aa654..4ff54f7a8 100644 --- a/packages/flow/tests/test_frame_conditions.py +++ b/packages/flow/tests/test_frame_conditions.py @@ -509,3 +509,81 @@ def test_countersign_states_every_part() -> None: 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() From 81034b7930c1f5f195090ed39b5acdf5db46e10c Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 14:10:08 -0400 Subject: [PATCH 3/3] frame: attribution only from evidence the bot can't write The safety review of #1443 set three conditions on excusing a change as another user's concurrent edit: - record_field attribution must name fields the app sets on the server, declared per app in server_set; a field the run writes, or a run parameter names, refuses the attribution (every change unattributed). - the executor's own id joins the bot's ids: never another user. - another actor's change to a record the save declares blocks verified (the saved value may not stand); only changes outside the declared delta are excused. Server-set bookkeeping fields are ignored as content: they say who wrote, not what changed. Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/VERIFY_RUN.md | 18 +++- .../openadapt_flow/runtime/effects/frame.py | 102 ++++++++++++++++-- packages/flow/openadapt_flow/verify_run.py | 4 + packages/flow/tests/test_frame_conditions.py | 101 ++++++++++++++++- 4 files changed, 217 insertions(+), 8 deletions(-) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index fdfe3f5ec..097f28ff7 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -100,7 +100,7 @@ frame: 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 + 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 @@ -119,6 +119,22 @@ 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 diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py index fc6d9e18e..15f3fd3db 100644 --- a/packages/flow/openadapt_flow/runtime/effects/frame.py +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -25,6 +25,18 @@ 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``. @@ -87,6 +99,11 @@ class FrameAttribution(BaseModel): 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 @@ -108,12 +125,23 @@ def _shape(self) -> "FrameAttribution": 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 @@ -167,6 +195,8 @@ class FrameChange: #: 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 @@ -199,6 +229,9 @@ def evidence(self, spec: FrameSpec) -> dict[str, Any]: "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 + ), } @@ -290,6 +323,7 @@ def judge_frame( 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``. @@ -304,6 +338,8 @@ def judge_frame( 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 @@ -317,14 +353,23 @@ def judge_frame( "not supply; changes can't be attributed", mode, ) - self_ids = resolved + self_ids = resolved | {str(a) for a in run_actors if str(a)} ignore = set(spec.ignore_fields) - may_change = ( - set(spec.target_fields) - if spec.target_fields is not None - else declared_fields(effects) - ) + 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: @@ -338,6 +383,29 @@ def key_of(record: Mapping[str, Any]) -> Any: 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]], @@ -441,6 +509,20 @@ def declared(record: Mapping[str, Any]) -> bool: ) 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 = [ @@ -526,6 +608,9 @@ def _reason(verdict: FrameVerdict) -> str: 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 @@ -558,6 +643,11 @@ def _reason(verdict: FrameVerdict) -> str: 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 73340da0f..8af3cb2de 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -392,6 +392,7 @@ def judge_run_frame( 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``.""" @@ -462,6 +463,7 @@ def read_of(name: str) -> str: routes=routes, audit_before=audit_before, audit_after=audit_after, + run_actors=run_actors, ) @@ -967,6 +969,8 @@ def verify_run( 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( diff --git a/packages/flow/tests/test_frame_conditions.py b/packages/flow/tests/test_frame_conditions.py index 4ff54f7a8..7961a1318 100644 --- a/packages/flow/tests/test_frame_conditions.py +++ b/packages/flow/tests/test_frame_conditions.py @@ -147,6 +147,7 @@ def _record_field(**kw) -> FrameSpec: kind="record_field", actor_field="by", version_field="v", + server_set=["by", "v"], **{"self": ["bot"]}, **kw, ) @@ -178,7 +179,10 @@ def test_a_change_without_a_new_version_is_unattributed() -> None: def test_self_ids_bind_to_run_params() -> None: spec = FrameSpec( attribution=FrameAttribution( - kind="record_field", actor_field="by", **{"self": [{"param": "bot_user"}]} + kind="record_field", + actor_field="by", + server_set=["by"], + **{"self": [{"param": "bot_user"}]}, ) ) edited = {**OTHER, "phone_home": "555-0199", "by": "bot"} @@ -587,3 +591,98 @@ def test_a_content_twin_is_a_view_of_the_same_read(tmp_path: Path) -> None: ) 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