From eb182f2a4ca239ce6794e372f3f8213a67798db6 Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 13:20:47 -0400 Subject: [PATCH 1/2] fault gate: admit a record check only after it catches its faults A check built from the same place as the save proves little until it has been seen to fail. The gate runs the check against a test copy of the system of record, writes seven kinds of bad save to the copy's store (wrong value per field, look-alike record, unrelated edit, partial, duplicate, missing, delayed) and lets the check read them through its own read, end to end through verify-run. A read whose scope can't see a fault fails the gate. - openadapt_flow/fault_gate.py: run_gate, check_digest (effects, read shape and frame; not the store's location), CoverageRecord (openadapt.check-coverage/v1), AdmissionPolicy, and two public test copies (MemoryRestCopy, SqliteCopy). - verify-run: admission: section; checked.coverage in the signed receipt ({killed, applicable, sha256, policy}); verified requires policy met. - scaffold-verifier --fault-gate: the gate on a copy of --db. Without it, the recorded-state test is written in the same format, incomplete on purpose (it can't test a wrong value over the right record or a late write). - countersign shows the coverage; qualify ignores admission:. Co-Authored-By: Claude Opus 5.5 --- CAPABILITIES.md | 1 + llms-full.txt | 1 + packages/flow/docs/VERIFY_RUN.md | 36 + packages/flow/openadapt_flow/__main__.py | 10 + .../openadapt_flow/countersign/document.py | 10 +- .../flow/openadapt_flow/countersign/render.py | 6 + packages/flow/openadapt_flow/fault_gate.py | 789 ++++++++++++++++++ .../flow/openadapt_flow/qualify/config.py | 2 +- .../openadapt_flow/record_check_synthesis.py | 177 +++- .../flow/openadapt_flow/scaffold_verifier.py | 12 +- packages/flow/openadapt_flow/verify_run.py | 73 +- packages/flow/tests/test_fault_gate.py | 298 +++++++ 12 files changed, 1406 insertions(+), 9 deletions(-) create mode 100644 packages/flow/openadapt_flow/fault_gate.py create mode 100644 packages/flow/tests/test_fault_gate.py diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 2e285085e..9f934a704 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -137,6 +137,7 @@ improvising or retrying when something is off. | 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) | +| Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` | | 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 cbfa88a16..cf56e3051 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -534,6 +534,7 @@ improvising or retrying when something is off. | 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) | +| Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` | | 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 fdfe3f5ec..8c7c29a25 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -153,6 +153,42 @@ 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. +## Admission: a check shown to catch its faults + +A record check proves little until it has been seen to fail. The fault gate +(`openadapt_flow/fault_gate.py`, [`VERIFICATION_COVERAGE.md`](VERIFICATION_COVERAGE.md)) +runs the check against a test copy of the system of record. It writes seven +kinds of bad save to the copy's store, and the check reads each one through +its own read: a wrong value in each declared field, the values on a +look-alike record, an edit to another record, a partial write, a duplicate, +no write, and a write that lands after the check's timeout. A read whose +scope can't see a fault fails the gate, not only a judge that misses it. + +The gate writes `check_coverage.json` (`openadapt.check-coverage/v1`): faults +killed over faults applicable, per class, bound to the check by its digest +(the effects, the read and the frame; not where the store is). An +`admission:` section makes verify-run require it: + +```yaml +admission: + coverage: check_coverage.json # relative to verify.yaml + min_coverage: 1.0 # default: every applicable fault caught + require_classes: [unrelated_edit] + refuse_test_copies: [recorded_states] # the default +``` + +Under a policy, `checked.coverage` in the receipt holds the totals, the +record's digest and `policy` (`met`, or why not, such as +`coverage_for_another_check`). `verified` requires `met`. A changed check +needs a new coverage record. Without `admission:` nothing about coverage is +required or stated. + +`openadapt-flow scaffold-verifier --db app.sqlite --fault-gate` runs the gate +on a copy of the database and keeps the check a draft when a fault survives. +Without `--fault-gate` it writes the coverage of its recorded-state test, +which can't test a wrong value written over the right record or a late +write, so that record is incomplete and refused by default. + ## Verdicts | What the store shows | `effect_verdict` | `outcome` | diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index feb6e2974..cc3f461e6 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -4829,6 +4829,7 @@ def optional_path(name: str) -> Optional[Path]: recording=optional_path("recording"), bundle=optional_path("bundle"), complete_read=bool(getattr(args, "complete_read", False)), + fault_gate=bool(getattr(args, "fault_gate", False)), ) if built is not None: print(built.summary) @@ -9410,6 +9411,15 @@ def build_parser() -> argparse.ArgumentParser: "(one store, no other tables). Recorded as declared, not proven" ), ) + p.add_argument( + "--fault-gate", + action="store_true", + help=( + "With --db: write each bad save to a copy of the database and let " + "the check read it through its own read (the fault gate). A fault " + "the read can't see keeps the check a draft" + ), + ) p.add_argument( "--recording", default=None, diff --git a/packages/flow/openadapt_flow/countersign/document.py b/packages/flow/openadapt_flow/countersign/document.py index a1e21b2e2..bd6f78c70 100644 --- a/packages/flow/openadapt_flow/countersign/document.py +++ b/packages/flow/openadapt_flow/countersign/document.py @@ -651,12 +651,20 @@ def verify_countersigned( 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"}: + if not isinstance(parts, dict) or set(parts) - {"coverage"} != { + "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") + or ( + "coverage" in parts + and (parts.get("coverage") or {}).get("policy") != "met" + ) ): report.problems.append("verified_without_every_part") # Seal signature: openadapt_types.seal, unchanged. diff --git a/packages/flow/openadapt_flow/countersign/render.py b/packages/flow/openadapt_flow/countersign/render.py index 92594d9d4..3e5c54a12 100644 --- a/packages/flow/openadapt_flow/countersign/render.py +++ b/packages/flow/openadapt_flow/countersign/render.py @@ -98,6 +98,12 @@ def checked_line(checked: dict[str, Any]) -> str: out.append("source: the inputs were the authorized instruction") else: out.append(f"{part} ({_PART_TEXT[part]}): {state}") + cover = checked.get("coverage") + if isinstance(cover, dict): + out.append( + f"fault coverage: {cover.get('killed')} of {cover.get('applicable')} " + f"faults caught, admission policy {cover.get('policy')}" + ) return "; ".join(out) diff --git a/packages/flow/openadapt_flow/fault_gate.py b/packages/flow/openadapt_flow/fault_gate.py new file mode 100644 index 000000000..3d37e110a --- /dev/null +++ b/packages/flow/openadapt_flow/fault_gate.py @@ -0,0 +1,789 @@ +"""Admit a record check only after it is shown to catch the faults it claims. + +A record check is built from the same place as the save, so a passing check +proves little until it has been seen to fail. This gate runs the check +against a **test copy** of the system of record and applies a fixed suite of +bad saves to the copy's **store**. The check reads them through **its own +read**, so a read whose scope can't see a fault fails the gate, not only a +judge that misses it. (Mutating the records a check already read, as +``record_check_synthesis.prove_against_faults`` does, tests the judge only.) + +Each case resets the copy, takes the check's baseline, performs the clean +save, applies one fault, and runs the check end to end through +:func:`openadapt_flow.verify_run.verify_run`. The clean save must come out +``verified``; each fault must not. + +================= =========================================== =========== +Fault class The copy holds after the save Claim +================= =========================================== =========== +``wrong_value`` the target with one declared field changed delta +``wrong_record`` the target's values on a look-alike instead delta +``unrelated_edit`` the clean save plus another record edited frame +``partial`` the target with its later fields blank delta +``duplicate`` the target twice delta +``missing`` no target delta +``delayed`` no target until after the check's timeout delta +================= =========================================== =========== + +Coverage is faults killed over faults applicable, recorded in +``check_coverage.json`` (``openadapt.check-coverage/v1``) and bound to the +check by :func:`check_digest`. ``verify-run`` reads it under an +``admission:`` policy and signs the totals into the receipt. + +The mechanism, the seven operators and this public suite are public. A grown +fault corpus and deployment coverage floors stay private +(``source-policy``: ``grown_corpus``, ``deployment_thresholds``). +""" + +from __future__ import annotations + +import copy +import hashlib +import json +import tempfile +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Mapping, Optional, Protocol, Sequence + +from pydantic import BaseModel, ConfigDict, Field + +from openadapt_flow.runtime.effects.effect import Effect, EffectKind + +COVERAGE_SCHEMA = "openadapt.check-coverage/v1" +SUITE_ID = "openadapt.fault-suite/public" +SUITE_VERSION = "1.0.0" +COVERAGE_FILE = "check_coverage.json" + +FAULT_CLASSES: tuple[str, ...] = ( + "wrong_value", + "wrong_record", + "unrelated_edit", + "partial", + "duplicate", + "missing", + "delayed", +) +#: Which part of the claim each fault class tests. +CLAIM_OF: dict[str, str] = { + name: ("frame" if name == "unrelated_edit" else "delta") for name in FAULT_CLASSES +} + +#: Verifier keys that say where the store is or how to sign in, not what is +#: read. A test copy changes them; the check is the same check. +_LOCATION_KEYS = frozenset( + { + "base_url", + "auth", + "access_token", + "access_token_env", + "smart_backend", + "verify_tls", + "sqlite_database", + "sql_driver", + "sql_connect_args", + "sql_password_env", + "secret_env", + "root", + "session", + } +) + + +class GateError(ValueError): + """The gate could not run. Nothing it produced may admit a check.""" + + +# -- the test copy ------------------------------------------------------------- + + +class TestCopy(Protocol): + """A disposable copy of the system of record, with privileged writes. + + The check reads the copy through its own verifier, built from the same + ``verify.yaml`` with :meth:`verifier_overrides` applied. The gate writes + through :meth:`update`, :meth:`insert` and :meth:`delete`, which bypass the + app on purpose: a fault is what the app should never do. + """ + + #: Short name recorded in the coverage file (``sqlite``, ``rest_memory``, + #: a fixture name). + name: str + #: The record field the copy's write primitives address records by. + key_field: str + + def reset(self) -> None: + """Restore the state before the save.""" + + def save(self) -> Any: + """Perform the clean save, as the app would. Returns the target's key.""" + + def rows(self) -> list[dict[str, Any]]: + """Every record of the slice, read with privilege.""" + + def update(self, key: Any, changes: Mapping[str, Any]) -> None: ... + + def insert(self, record: Mapping[str, Any]) -> Any: ... + + def delete(self, key: Any) -> None: ... + + def verifier_overrides(self) -> dict[str, Any]: + """Verifier keys that point the check's read at this copy.""" + + +# -- results ------------------------------------------------------------------- + + +@dataclass +class FaultCase: + fault: str + detail: str + applicable: bool = True + #: True when the check did not come out verified. + killed: bool = False + #: True when the fault could not be built on this copy (not credited). + untestable: bool = False + outcome: str = "" + checked: dict[str, Any] = field(default_factory=dict) + + def as_dict(self) -> dict[str, Any]: + return { + "fault": self.fault, + "detail": self.detail, + "applicable": self.applicable, + "result": ( + "not applicable" + if not self.applicable + else "untestable" + if self.untestable + else "killed" + if self.killed + else "SURVIVED" + ), + "outcome": self.outcome, + "checked": self.checked, + } + + +class ClassCoverage(BaseModel): + model_config = ConfigDict(extra="forbid") + applicable: int = 0 + killed: int = 0 + + +class CoverageRecord(BaseModel): + """``check_coverage.json``.""" + + model_config = ConfigDict(extra="forbid") + + schema_version: str = COVERAGE_SCHEMA + check_sha256: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + suite: dict[str, str] + test_copy: str + claims: list[str] + clean_pass: bool + classes: dict[str, ClassCoverage] + killed: int + applicable: int + cases: list[dict[str, Any]] = Field(default_factory=list) + + @property + def complete(self) -> bool: + """The clean save passed and every applicable fault was killed.""" + + return ( + self.clean_pass and self.applicable > 0 and self.killed == self.applicable + ) + + def digest(self) -> str: + payload = self.model_dump(mode="json") + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + def totals(self) -> dict[str, Any]: + """What the receipt carries: the totals and this record's digest.""" + + return { + "killed": self.killed, + "applicable": self.applicable, + "sha256": self.digest(), + } + + +class AdmissionPolicy(BaseModel): + """``admission:`` in ``verify.yaml``. + + The public default is the rule, not a tuned number: the clean save passes + and every applicable fault is killed. A deployment can also require + classes to be applicable (``require_classes``, for example every check + must claim a frame) and refuse coverage measured only on recorded states. + """ + + model_config = ConfigDict(extra="forbid") + + #: Path to ``check_coverage.json``, relative to ``verify.yaml``. + coverage: str + #: Fraction of applicable faults that must be killed. 1.0 by default. + min_coverage: float = Field(default=1.0, ge=0.0, le=1.0) + require_classes: list[str] = Field(default_factory=list) + #: Test-copy kinds whose coverage this policy refuses. + refuse_test_copies: list[str] = Field(default_factory=lambda: ["recorded_states"]) + + def problem(self, record: CoverageRecord, check_sha256: str) -> Optional[str]: + """None when ``record`` admits the check ``check_sha256``.""" + + if record.schema_version != COVERAGE_SCHEMA: + return "coverage_schema" + if record.check_sha256 != check_sha256: + return "coverage_for_another_check" + if record.test_copy in self.refuse_test_copies: + return "coverage_test_copy_refused" + if not record.clean_pass: + return "coverage_clean_save_failed" + if record.applicable == 0: + return "coverage_empty" + if record.killed / record.applicable < self.min_coverage: + return "coverage_below_policy" + for name in self.require_classes: + cls = record.classes.get(name) + if cls is None or cls.applicable == 0: + return "coverage_class_missing" + return None + + +# -- the check's identity ------------------------------------------------------ + + +def _read_shape(section: Mapping[str, Any]) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, value in sorted(section.items()): + if key in _LOCATION_KEYS: + continue + if key == "surfaces" and isinstance(value, Mapping): + out[key] = {name: _read_shape(sub) for name, sub in sorted(value.items())} + else: + out[key] = value + return out + + +def check_digest(cfg: Any) -> str: + """SHA-256 of what the check is: its contract, its read and its frame. + + The contract alone (``verify_config_sha256``) does not name the read, and + a narrower read with the same effects is a weaker check. Where the store + is and how to sign in to it are left out, so a test copy runs the same + check as production. + """ + + verifier = cfg.verifier.model_dump(mode="json", exclude_defaults=True) + frame = cfg.frame.model_dump(mode="json", by_alias=True) if cfg.frame else None + payload = { + "effects": [e.model_dump(mode="json") for e in cfg.effects], + "read": _read_shape(verifier), + "frame": frame, + } + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +# -- fault operators ----------------------------------------------------------- + + +def _perturb(value: Any) -> Any: + """A value that differs from ``value`` and still looks like one.""" + + if isinstance(value, bool): + return not value + if isinstance(value, int): + return value + 1 + if isinstance(value, float): + return value + 1.0 + text = "" if value is None else str(value) + if text and text[-1].isdigit(): + return text[:-1] + str((int(text[-1]) + 1) % 10) + return text + "x" if text else "x" + + +def _declared(effects: Sequence[Effect], params: Mapping[str, object]) -> list[str]: + """Fields the contract pins, in declaration order.""" + + names: list[str] = [] + for effect in effects: + bound = effect.resolve(params) + for key in bound.match: + if key not in names: + names.append(key) + if bound.field and bound.field not in names: + names.append(bound.field) + return names + + +def _target_row(copy_: TestCopy, key: Any) -> Optional[dict[str, Any]]: + for row in copy_.rows(): + if str(row.get(copy_.key_field)) == str(key): + return dict(row) + return None + + +@dataclass +class _Prepared: + """The copy after reset, baseline and clean save.""" + + key: Any + target: dict[str, Any] + + +Apply = Callable[[TestCopy, _Prepared], Optional[Callable[[], None]]] + + +def _plan( + effects: Sequence[Effect], + params: Mapping[str, object], + sample: dict[str, Any], + key_field: str, +) -> list[tuple[str, str, Apply]]: + """The suite as (class, detail, operator). Operators may return a + callable to run after the check starts (``delayed``).""" + + fields = [f for f in _declared(effects, params) if f in sample and f != key_field] + plan: list[tuple[str, str, Apply]] = [] + + for name in fields: + + def wrong(c: TestCopy, p: _Prepared, name: str = name) -> None: + c.update(p.key, {name: _perturb(p.target.get(name))}) + + plan.append(("wrong_value", f"{name} changed", wrong)) + + def wrong_record(c: TestCopy, p: _Prepared) -> None: + twin = {k: v for k, v in p.target.items() if k != key_field} + identity = fields[0] if fields else None + if identity is not None: + twin[identity] = _perturb(twin.get(identity)) + c.delete(p.key) + c.insert(twin) + + plan.append( + ("wrong_record", "the values saved on a look-alike record", wrong_record) + ) + + def unrelated(c: TestCopy, p: _Prepared) -> None: + others = [r for r in c.rows() if str(r.get(key_field)) != str(p.key)] + if not others: + raise _Untestable("the copy holds no other record to edit") + victim = others[0] + name = next( + ( + k + for k, v in victim.items() + if k != key_field + and isinstance(v, (str, int, float)) + and v not in ("",) + ), + None, + ) + if name is None: + raise _Untestable("the other record has no editable field") + c.update(victim[key_field], {name: _perturb(victim[name])}) + + plan.append(("unrelated_edit", "another record in the slice edited", unrelated)) + + later = [f for f in fields[len(fields) // 2 :] if f in sample] or fields[-1:] + + def partial(c: TestCopy, p: _Prepared) -> None: + if not later: + raise _Untestable("the contract pins no field to leave blank") + c.update(p.key, {name: "" for name in later}) + + plan.append(("partial", f"{', '.join(later)} left blank", partial)) + + def duplicate(c: TestCopy, p: _Prepared) -> None: + c.insert({k: v for k, v in p.target.items() if k != key_field}) + + plan.append(("duplicate", "the target saved twice", duplicate)) + + def missing(c: TestCopy, p: _Prepared) -> None: + c.delete(p.key) + + plan.append(("missing", "nothing saved", missing)) + + def delayed(c: TestCopy, p: _Prepared) -> Callable[[], None]: + c.delete(p.key) + record = {k: v for k, v in p.target.items() if k != key_field} + return lambda: c.insert(record) + + plan.append(("delayed", "the target lands after the check's timeout", delayed)) + return plan + + +class _Untestable(Exception): + pass + + +# -- the gate ------------------------------------------------------------------ + + +def _check_timeout(effects: Sequence[Effect]) -> float: + return max((e.timeout_s for e in effects), default=5.0) + + +def _point_at(verifier: Any, overrides: Mapping[str, Any]) -> Any: + """``verifier`` with its store location replaced, surfaces included.""" + + if verifier.surfaces: + return verifier.model_copy( + update={ + "surfaces": { + name: _point_at(sub, overrides) + for name, sub in verifier.surfaces.items() + } + } + ) + return verifier.model_copy(update=dict(overrides)) + + +def run_gate( + cfg: Any, + params: Mapping[str, object], + copy_: TestCopy, + *, + claims: Sequence[str] = ("delta", "frame"), + workdir: Optional[Path] = None, + delay_margin_s: float = 1.0, +) -> CoverageRecord: + """Run the public fault suite on ``copy_`` and return the coverage record. + + ``cfg`` is the check's parsed ``verify.yaml``. Its read is pointed at the + copy with :meth:`TestCopy.verifier_overrides`; nothing else changes. + """ + + from openadapt_flow.verify_run import capture_baseline, verify_run + + effects = list(cfg.effects) + if not any( + e.kind in (EffectKind.RECORD_WRITTEN, EffectKind.FIELD_EQUALS) for e in effects + ): + raise GateError("the gate needs a record_written or field_equals effect") + verifier = _point_at(cfg.verifier, copy_.verifier_overrides()) + # The gate is what produces the coverage an admission policy asks for. + local = cfg.model_copy(update={"verifier": verifier, "admission": None}) + own = Path(tempfile.mkdtemp(prefix="fault-gate-")) if workdir is None else workdir + own.mkdir(parents=True, exist_ok=True) + runs = 0 + + def prepare() -> tuple[Path, _Prepared]: + nonlocal runs + runs += 1 + copy_.reset() + baseline = own / f"baseline-{runs}.json" + capture_baseline(local, params, baseline) + key = copy_.save() + target = _target_row(copy_, key) + if target is None: + raise GateError("the clean save left no record the copy can find") + return baseline, _Prepared(key, target) + + def judge(baseline: Path) -> dict[str, Any]: + return verify_run( + local, + params=params, + run_record={ + "executor_kind": "script", + "executor_id": "openadapt-fault-gate", + }, + authorization_ref="openadapt-fault-gate", + baseline_path=baseline, + private_dir=own / f"run-{runs}", + )["receipt"] + + baseline, prepared = prepare() + clean = judge(baseline) + clean_pass = clean["outcome"] == "verified" + cases = [ + FaultCase( + "clean", + "the save as the app makes it", + applicable=False, + killed=not clean_pass, + outcome=clean["outcome"], + checked=dict(clean.get("checked") or {}), + ) + ] + for fault, detail, apply in _plan( + effects, params, prepared.target, copy_.key_field + ): + case = FaultCase(fault, detail, applicable=CLAIM_OF[fault] in claims) + cases.append(case) + if not case.applicable: + continue + try: + baseline, prepared = prepare() + later = apply(copy_, prepared) + except _Untestable as exc: + case.untestable = True + case.detail += f" ({exc})" + continue + timer: Optional[threading.Timer] = None + if later is not None: + timer = threading.Timer(_check_timeout(effects) + delay_margin_s, later) + timer.start() + try: + receipt = judge(baseline) + finally: + if timer is not None: + timer.join() + case.outcome = receipt["outcome"] + case.checked = dict(receipt.get("checked") or {}) + case.killed = receipt["outcome"] != "verified" + copy_.reset() + + classes: dict[str, ClassCoverage] = {} + for case in cases[1:]: + entry = classes.setdefault(case.fault, ClassCoverage()) + if case.applicable: + entry.applicable += 1 + if case.killed and not case.untestable: + entry.killed += 1 + return CoverageRecord( + check_sha256=check_digest(cfg), + suite={"id": SUITE_ID, "version": SUITE_VERSION}, + test_copy=copy_.name, + claims=sorted(set(claims)), + clean_pass=clean_pass, + classes=classes, + killed=sum(c.killed for c in classes.values()), + applicable=sum(c.applicable for c in classes.values()), + cases=[c.as_dict() for c in cases], + ) + + +def write_coverage(record: CoverageRecord, out_dir: Path) -> Path: + path = Path(out_dir) / COVERAGE_FILE + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(record.model_dump(mode="json"), indent=2, sort_keys=True) + "\n" + ) + return path + + +def load_coverage(path: Path) -> CoverageRecord: + try: + return CoverageRecord.model_validate(json.loads(Path(path).read_text())) + except (OSError, ValueError) as exc: + raise GateError(f"cannot read coverage record {path}: {exc}") from exc + + +# -- public test copies ---------------------------------------------------------- + + +class MemoryRestCopy: + """An in-memory record store served over HTTP: the sample test copy. + + Serves ``GET `` as ``{records_key: [...]}``. ``seed`` is the + state before the save, ``target`` the record the clean save writes. + """ + + name = "rest_memory" + + def __init__( + self, + seed: Sequence[Mapping[str, Any]], + target: Mapping[str, Any], + *, + records_path: str = "/api/db", + records_key: Optional[str] = "records", + key_field: str = "id", + ) -> None: + self.key_field = key_field + self._seed = [dict(r) for r in seed] + self._target = dict(target) + self._records: list[dict[str, Any]] = [] + self._lock = threading.Lock() + self._next = 1000 + self.records_path = records_path + self.records_key = records_key + self._server, self.base_url = self._serve() + + def _serve(self) -> tuple[Any, str]: + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + store = self + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - http.server API + if self.path != store.records_path: + self.send_response(404) + self.end_headers() + return + with store._lock: + rows = copy.deepcopy(store._records) + body = rows if store.records_key is None else {store.records_key: rows} + data = json.dumps(body).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *_args: Any) -> None: + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server, f"http://127.0.0.1:{server.server_address[1]}" + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + + def reset(self) -> None: + with self._lock: + self._records = [dict(r) for r in self._seed] + + def save(self) -> Any: + return self.insert(self._target) + + def rows(self) -> list[dict[str, Any]]: + with self._lock: + return copy.deepcopy(self._records) + + def update(self, key: Any, changes: Mapping[str, Any]) -> None: + with self._lock: + for row in self._records: + if str(row.get(self.key_field)) == str(key): + row.update(changes) + + def insert(self, record: Mapping[str, Any]) -> Any: + with self._lock: + self._next += 1 + row = {**dict(record), self.key_field: self._next} + self._records.append(row) + return self._next + + def delete(self, key: Any) -> None: + with self._lock: + self._records = [ + r for r in self._records if str(r.get(self.key_field)) != str(key) + ] + + def verifier_overrides(self) -> dict[str, Any]: + return { + "base_url": self.base_url, + "records_path": self.records_path, + "records_key": self.records_key, + } + + +class SqliteCopy: + """A copy of a SQLite database file, one table as the slice. + + ``source`` is copied before every case. ``save`` runs the clean save's + SQL (or inserts ``target``). The check's read runs against the copy. + """ + + name = "sqlite" + + def __init__( + self, + source: Path, + table: str, + target: Mapping[str, Any], + *, + key_field: str = "id", + saved_key: Any = None, + workdir: Optional[Path] = None, + ) -> None: + import re + + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,63}", table) or not re.fullmatch( + r"[A-Za-z_][A-Za-z0-9_]{0,63}", key_field + ): + raise GateError("table and key field must be plain identifiers") + self.source = Path(source) + self.table = table + self.key_field = key_field + self._target = {k: v for k, v in dict(target).items() if k != key_field} + #: When ``source`` already holds the saved record (a database copied + #: after the demonstration), its key: reset removes it. + self.saved_key = saved_key + base = ( + Path(tempfile.mkdtemp(prefix="fault-gate-db-")) + if workdir is None + else workdir + ) + self.path = base / "copy.sqlite" + + def _connect(self) -> Any: + import sqlite3 + + con = sqlite3.connect(self.path) + con.row_factory = sqlite3.Row + return con + + def _columns(self, con: Any) -> set[str]: + return {row[1] for row in con.execute(f'PRAGMA table_info("{self.table}")')} + + def reset(self) -> None: + import shutil + + self.path.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(self.source, self.path) + if self.saved_key is not None: + self.delete(self.saved_key) + + def save(self) -> Any: + return self.insert(self._target) + + def rows(self) -> list[dict[str, Any]]: + con = self._connect() + try: + return [dict(r) for r in con.execute(f'SELECT * FROM "{self.table}"')] + finally: + con.close() + + def _names(self, con: Any, names: Sequence[str]) -> list[str]: + known = self._columns(con) + bad = [n for n in names if n not in known] + if bad: + raise GateError(f"unknown column(s) {bad} in {self.table}") + return list(names) + + def update(self, key: Any, changes: Mapping[str, Any]) -> None: + con = self._connect() + try: + names = self._names(con, list(changes)) + sets = ", ".join(f'"{n}" = ?' for n in names) + con.execute( + f'UPDATE "{self.table}" SET {sets} WHERE "{self.key_field}" = ?', + [changes[n] for n in names] + [key], + ) + con.commit() + finally: + con.close() + + def insert(self, record: Mapping[str, Any]) -> Any: + con = self._connect() + try: + names = self._names(con, [k for k in record if k != self.key_field]) + cur = con.execute( + f'INSERT INTO "{self.table}" ({", ".join(chr(34) + n + chr(34) for n in names)}) ' + f"VALUES ({', '.join('?' for _ in names)})", + [record[n] for n in names], + ) + con.commit() + return cur.lastrowid + finally: + con.close() + + def delete(self, key: Any) -> None: + con = self._connect() + try: + con.execute( + f'DELETE FROM "{self.table}" WHERE "{self.key_field}" = ?', [key] + ) + con.commit() + finally: + con.close() + + def verifier_overrides(self) -> dict[str, Any]: + return {"sqlite_database": str(self.path)} diff --git a/packages/flow/openadapt_flow/qualify/config.py b/packages/flow/openadapt_flow/qualify/config.py index 50c408375..8736faf5f 100644 --- a/packages/flow/openadapt_flow/qualify/config.py +++ b/packages/flow/openadapt_flow/qualify/config.py @@ -403,7 +403,7 @@ def sub(match: re.Match[str]) -> str: #: ``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",) +VERIFY_RUN_ONLY_SECTIONS: tuple[str, ...] = ("frame", "admission") def parse_verify_config( diff --git a/packages/flow/openadapt_flow/record_check_synthesis.py b/packages/flow/openadapt_flow/record_check_synthesis.py index c606e2d2d..547d790ce 100644 --- a/packages/flow/openadapt_flow/record_check_synthesis.py +++ b/packages/flow/openadapt_flow/record_check_synthesis.py @@ -1419,6 +1419,10 @@ class Proposal: faults: list[FaultResult] live: Optional[dict[str, Any]] = None identity: Optional[dict[str, Any]] = None + #: The fault gate's coverage record (``openadapt_flow/fault_gate.py``): + #: on a copy of the database with ``fault_gate``, else mapped from the + #: recorded-state faults above. + coverage: Optional[Any] = None @property def misses(self) -> list[str]: @@ -1432,6 +1436,14 @@ def misses(self) -> list[str]: out += [f"read: {item}" for item in self.search.problems] if self.live is not None: out += [f"live {m}" for m in self.live.get("misses", [])] + if self.coverage is not None and self.coverage.test_copy != "recorded_states": + if not self.coverage.clean_pass: + out.append("fault gate: the clean save did not verify on the copy") + out += [ + f"fault gate {case['fault']}: {case['detail']} ({case['result']})" + for case in self.coverage.cases + if case["applicable"] and case["result"] != "killed" + ] return out @property @@ -1481,6 +1493,16 @@ def proof(self) -> dict[str, Any]: "misses": self.misses, "notes": list(self.search.notes), "identity": self.identity, + "coverage": ( + None + if self.coverage is None + else { + "test_copy": self.coverage.test_copy, + "killed": self.coverage.killed, + "applicable": self.coverage.applicable, + "sha256": self.coverage.digest(), + } + ), } @@ -1491,6 +1513,7 @@ def synthesize( db_path: Optional[Path] = None, verifier: Optional[Mapping[str, Any]] = None, complete_read: bool = False, + fault_gate: bool = False, ) -> Proposal: """Find the read, build the check, and run the recorded-state fault test. @@ -1549,7 +1572,144 @@ def synthesize( params, closed_world=search.closed_world, ) - return Proposal(params, search, effects, section, faults) + proposal = Proposal(params, search, effects, section, faults) + cfg = gate_config(proposal) + if fault_gate and db_path is not None: + proposal.coverage = gate_on_database(proposal, Path(db_path), cfg) + if proposal.coverage is None and cfg is not None: + proposal.coverage = coverage_from_recorded(proposal.faults, cfg) + return proposal + + +# --------------------------------------------------------------------------- +# The fault gate (docs/VERIFICATION_COVERAGE.md section 2) +# --------------------------------------------------------------------------- + +#: Recorded-state fault -> fault-gate class. +_RECORDED_CLASS = { + "no_persist": "missing", + "partial": "partial", + "duplicate": "duplicate", + "wrong_record": "wrong_record", + "stale": "unrelated_edit", + "changed": "unrelated_edit", + "deleted": "unrelated_edit", + "collateral_unaudited": "unrelated_edit", +} + + +def gate_config(proposal: Proposal) -> Optional[Any]: + """The proposed check as ``verify-run`` would read it once confirmed.""" + + from openadapt_flow.verify_run import VerifyRunError, parse_verify_config + + if not proposal.effects or not proposal.verifier: + return None + try: + return parse_verify_config( + { + "schema_version": "openadapt.verify/v1", + "verifier": proposal.verifier, + "effects": [ + e.model_copy( + update={"needs_operator_confirmation": False} + ).model_dump(mode="json", exclude_defaults=True) + for e in proposal.effects + ], + } + ) + except VerifyRunError: + return None + + +def coverage_from_recorded(faults: list[FaultResult], cfg: Any) -> Any: + """Coverage of the recorded-state faults, in the fault gate's format. + + A recorded state can't hold a wrong value written on top of the right + record or a write that lands late, so ``wrong_value`` and ``delayed`` + count as applicable and not killed. An admission policy refuses this + kind by default. + """ + + from openadapt_flow.fault_gate import ( + SUITE_ID, + SUITE_VERSION, + ClassCoverage, + CoverageRecord, + check_digest, + ) + + classes = { + name: ClassCoverage() + for name in ( + "wrong_value", + "wrong_record", + "unrelated_edit", + "partial", + "duplicate", + "missing", + "delayed", + ) + } + for name in ("wrong_value", "delayed"): + classes[name].applicable = 1 + cases: list[dict[str, Any]] = [] + clean_pass = True + for result in faults: + if result.correct: + clean_pass = clean_pass and result.ok + continue + mapped = _RECORDED_CLASS.get(result.fault) + if mapped is None: + continue + classes[mapped].applicable += 1 + if result.ok: + classes[mapped].killed += 1 + cases.append( + { + "fault": mapped, + "detail": result.detail, + "applicable": True, + "result": "killed" if result.ok else result.as_dict()["result"], + "outcome": "", + "checked": {}, + } + ) + return CoverageRecord( + check_sha256=check_digest(cfg), + suite={"id": SUITE_ID, "version": SUITE_VERSION}, + test_copy="recorded_states", + claims=["delta", "frame"], + clean_pass=clean_pass, + classes=classes, + killed=sum(c.killed for c in classes.values()), + applicable=sum(c.applicable for c in classes.values()), + cases=cases, + ) + + +def gate_on_database(proposal: Proposal, db_path: Path, cfg: Any) -> Optional[Any]: + """Run the fault gate on a copy of ``db_path``, the saved row removed. + + Needs a target table whose rows carry an ``id``; otherwise the gate + can't address the saved row and the recorded-state coverage stands. + """ + + from openadapt_flow.fault_gate import SqliteCopy, run_gate + + if cfg is None: + return None + target = next((s for s in proposal.search.surfaces if s.is_target), None) + if ( + target is None + or not target.source.startswith("table ") + or not target.target + or "id" not in target.target + ): + return None + table = target.source.split(" ", 1)[1] + copy_ = SqliteCopy(db_path, table, target.target, saved_key=target.target["id"]) + return run_gate(cfg, proposal.params, copy_) # --------------------------------------------------------------------------- @@ -1606,7 +1766,10 @@ def render(proposal: Proposal) -> str: def write(proposal: Proposal, out_dir: Path) -> tuple[Path, Path]: - """Write the proposed check and its proof; return both paths.""" + """Write the proposed check and its proof; return both paths. + + The fault gate's coverage record goes beside them (``check_coverage.json``). + """ out_dir.mkdir(parents=True, exist_ok=True) check = out_dir / PROPOSED_NAME @@ -1617,6 +1780,16 @@ def write(proposal: Proposal, out_dir: Path) -> tuple[Path, Path]: proof, json.dumps(proposal.proof(), indent=2, sort_keys=True, default=str) + "\n", ) + if proposal.coverage is not None: + from openadapt_flow.fault_gate import COVERAGE_FILE + + _sealed.write_text( + out_dir / COVERAGE_FILE, + json.dumps( + proposal.coverage.model_dump(mode="json"), indent=2, sort_keys=True + ) + + "\n", + ) return check, proof diff --git a/packages/flow/openadapt_flow/scaffold_verifier.py b/packages/flow/openadapt_flow/scaffold_verifier.py index 9b62e180c..200b9b805 100644 --- a/packages/flow/openadapt_flow/scaffold_verifier.py +++ b/packages/flow/openadapt_flow/scaffold_verifier.py @@ -518,6 +518,7 @@ def build_record_check( recording: Optional[Path] = None, bundle: Optional[Path] = None, complete_read: bool = False, + fault_gate: bool = False, ) -> Optional[RecordCheckResult]: """Build, fault-test and write a proposed record check. @@ -525,6 +526,11 @@ def build_record_check( ``recording`` / ``bundle`` (a bundle alone holds no reads, and identity coverage needs both). Returns ``None`` when no read returns the saved record, so the caller can fall back to the TODO draft. + + ``fault_gate`` with ``db`` runs the fault gate on a copy of the database + (``openadapt_flow/fault_gate.py``): the faults are written to the copy and + the check reads them through its own read. A fault it misses keeps the + check a draft. Either way ``check_coverage.json`` is written. """ import tempfile @@ -567,7 +573,11 @@ def build_record_check( ) try: proposal = rcs.synthesize( - recording_dir, db_path=db, verifier=section, complete_read=complete_read + recording_dir, + db_path=db, + verifier=section, + complete_read=complete_read, + fault_gate=fault_gate, ) except rcs.SynthesisError as exc: if db is None and section is None: diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 73340da0f..1ddd703f7 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -61,9 +61,16 @@ CHECKED_HOW_VALUES, checked_how_for_flow_tier, ) -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, PrivateAttr from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.fault_gate import ( + AdmissionPolicy, + CoverageRecord, + GateError, + check_digest, + load_coverage, +) from openadapt_flow.qualify.result import verify_config_sha256 from openadapt_flow.runtime.effects.adapter import verifier_identity from openadapt_flow.runtime.effects.effect import ( @@ -142,6 +149,8 @@ #: ``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.coverage``, present only under an admission policy. +COVERAGE_KEYS = frozenset({"killed", "applicable", "sha256", "policy"}) CHECKED_VALUES = ("held", "refuted", "not_checked") SOURCE_VALUES = CHECKED_VALUES + ("instruction",) @@ -203,6 +212,20 @@ class VerifyConfig(BaseModel): #: (``runtime/effects/frame.py``). Unset: the default frame (the whole #: read set, compared by ``id``, no attribution). frame: Optional[FrameSpec] = None + #: The check's admission policy: its fault-gate coverage record and the + #: minimum it must meet (``openadapt_flow/fault_gate.py``). Unset: no + #: coverage is required or stated. + admission: Optional[AdmissionPolicy] = None + #: Where ``admission.coverage`` resolves (the file's directory). + _base_dir: Optional[Path] = PrivateAttr(default=None) + + def coverage_path(self) -> Optional[Path]: + if self.admission is None: + return None + path = Path(self.admission.coverage) + if not path.is_absolute() and self._base_dir is not None: + path = self._base_dir / path + return path def load_verify_config(path: str | Path) -> VerifyConfig: @@ -215,7 +238,9 @@ def load_verify_config(path: str | Path) -> VerifyConfig: data = yaml.safe_load(source.read_text(encoding="utf-8")) or {} except yaml.YAMLError as exc: raise VerifyRunError(f"cannot parse {source}: {exc}") from exc - return parse_verify_config(data, source=f"verify config {source}") + cfg = parse_verify_config(data, source=f"verify config {source}") + cfg._base_dir = source.resolve().parent + return cfg #: ``openadapt.verify/v1`` sections only qualify reads (RFC-0002, Qualify-only @@ -806,6 +831,7 @@ def verify_run( remediates: Optional[str] = None, correlation_id: Optional[str] = None, write_attempted: bool = False, + coverage: Optional[CoverageRecord] = None, ) -> dict[str, Any]: """Read the system of record and return ``{"receipt", "evidence"}``. @@ -1016,15 +1042,38 @@ def verify_run( ) # Inputs come from whoever authorized the task (``--params``): the # instruction is their source. - checked = { + checked: dict[str, Any] = { "delta": _part(delta_verdict, read_proves_write), "frame": frame.status, "source": "instruction", } + admitted = True + if cfg.admission is not None: + # The check must have been shown to catch the faults it claims. + record = coverage + problem: Optional[str] = None + if record is None: + path = cfg.coverage_path() + try: + record = load_coverage(path) if path is not None else None + except GateError: + record = None + if record is None: + problem = "coverage_missing" + else: + problem = cfg.admission.problem(record, check_digest(cfg)) + admitted = problem is None + checked["coverage"] = { + "killed": record.killed if record is not None else 0, + "applicable": record.applicable if record is not None else 0, + "sha256": record.digest() if record is not None else None, + "policy": "met" if admitted else str(problem), + } claim_holds = ( read_proves_write and frame.held and checked["source"] in ("held", "instruction") + and admitted ) if claim_holds and self_approved is True: # The write landed, but the executor approved its own write: no @@ -1213,9 +1262,24 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: # Optional: receipts issued before docs/VERIFICATION_COVERAGE.md lack it. if "checked" in receipt: parts = receipt["checked"] + cover = parts.get("coverage") if isinstance(parts, Mapping) else None if ( not isinstance(parts, Mapping) - or set(parts) != set(CHECKED_PARTS) + or set(parts) - {"coverage"} != set(CHECKED_PARTS) + or ( + "coverage" in parts + and ( + not isinstance(cover, Mapping) + or set(cover) != COVERAGE_KEYS + or not isinstance(cover.get("killed"), int) + or not isinstance(cover.get("applicable"), int) + or not isinstance(cover.get("policy"), str) + or ( + cover.get("sha256") is not None + and not _DIGEST.match(str(cover.get("sha256"))) + ) + ) + ) or parts.get("delta") not in CHECKED_VALUES or parts.get("frame") not in CHECKED_VALUES or parts.get("source") not in SOURCE_VALUES @@ -1225,6 +1289,7 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: parts["delta"] != "held" or parts["frame"] != "held" or parts["source"] not in ("held", "instruction") + or (cover is not None and cover.get("policy") != "met") ): return "verified_without_every_part" if outcome in ("verified", "rolled_back_verified") and ( diff --git a/packages/flow/tests/test_fault_gate.py b/packages/flow/tests/test_fault_gate.py new file mode 100644 index 000000000..0fa3e3831 --- /dev/null +++ b/packages/flow/tests/test_fault_gate.py @@ -0,0 +1,298 @@ +"""The fault gate: a record check is admitted only after it catches its faults. + +See ``docs/VERIFICATION_COVERAGE.md`` section 2. Records are made up. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from openadapt_flow.fault_gate import ( + AdmissionPolicy, + MemoryRestCopy, + SqliteCopy, + check_digest, + load_coverage, + run_gate, + write_coverage, +) +from openadapt_flow.verify_run import parse_verify_config + +PARAMS = {"patient_id": "p1", "note": "Follow-up in two weeks", "type": "Triage"} +SEED = [{"id": 1, "patient_id": "p2", "type": "Consult", "note": "Allergy: penicillin"}] +TARGET = {"patient_id": "p1", "type": "Triage", "note": "Follow-up in two weeks"} + + +def _effects() -> list[dict]: + match = {"patient_id": {"param": "patient_id"}, "note": {"param": "note"}} + return [ + { + "kind": "record_written", + "match": match, + "count_new_only": True, + "timeout_s": 0.3, + }, + { + "kind": "field_equals", + "match": match, + "field": "type", + "value": {"param": "type"}, + "timeout_s": 0.3, + }, + ] + + +def _rest_cfg(base: str = "http://127.0.0.1:1"): + return parse_verify_config( + { + "schema_version": "openadapt.verify/v1", + "verifier": {"kind": "rest", "base_url": base, "records_path": "/api/db"}, + "effects": _effects(), + } + ) + + +@pytest.fixture +def rest_copy(): + store = MemoryRestCopy(SEED, TARGET) + try: + yield store + finally: + store.close() + + +def test_a_full_read_kills_every_fault(rest_copy, tmp_path: Path) -> None: + record = run_gate(_rest_cfg(), PARAMS, rest_copy, workdir=tmp_path) + assert record.clean_pass + survived = [c for c in record.cases if c["result"] == "SURVIVED"] + assert not survived, survived + assert record.complete + assert set(record.classes) == { + "wrong_value", + "wrong_record", + "unrelated_edit", + "partial", + "duplicate", + "missing", + "delayed", + } + # One wrong-value case per declared field. + assert record.classes["wrong_value"].applicable == 3 + assert record.killed == record.applicable == 9 + + +def test_a_check_that_claims_no_frame_is_not_tested_on_one( + rest_copy, tmp_path: Path +) -> None: + record = run_gate( + _rest_cfg(), PARAMS, rest_copy, claims=("delta",), workdir=tmp_path + ) + assert record.classes["unrelated_edit"].applicable == 0 + assert record.applicable == 8 + + +def _sqlite(tmp_path: Path) -> Path: + db = tmp_path / "app.sqlite" + con = sqlite3.connect(db) + con.executescript( + "CREATE TABLE enc (id INTEGER PRIMARY KEY, patient_id TEXT, type TEXT, note TEXT);" + "INSERT INTO enc VALUES (1, 'p2', 'Consult', 'Allergy: penicillin');" + ) + con.commit() + con.close() + return db + + +def _sql_cfg(db: Path, *, scoped: bool): + query = "SELECT id, patient_id, type, note FROM enc" + verifier = {"kind": "sql", "sqlite_database": str(db), "sql_query": query} + if scoped: + verifier["sql_query"] = query + " WHERE patient_id = :p" + verifier["sql_query_params"] = {"p": {"param": "patient_id"}} + return parse_verify_config( + { + "schema_version": "openadapt.verify/v1", + "verifier": verifier, + "effects": _effects(), + } + ) + + +def test_a_read_scoped_to_the_patient_misses_the_unrelated_edit(tmp_path: Path) -> None: + """The blind spot of #1365: the check can't see another patient's chart.""" + + db = _sqlite(tmp_path) + copy = SqliteCopy(db, "enc", TARGET, workdir=tmp_path / "copy") + record = run_gate( + _sql_cfg(db, scoped=True), PARAMS, copy, workdir=tmp_path / "gate" + ) + assert record.clean_pass + assert record.classes["unrelated_edit"].killed == 0 + assert not record.complete + survived = [c["fault"] for c in record.cases if c["result"] == "SURVIVED"] + assert survived == ["unrelated_edit"] + + +def test_the_same_table_read_whole_kills_it(tmp_path: Path) -> None: + db = _sqlite(tmp_path) + copy = SqliteCopy(db, "enc", TARGET, workdir=tmp_path / "copy") + record = run_gate( + _sql_cfg(db, scoped=False), PARAMS, copy, workdir=tmp_path / "gate" + ) + assert record.complete, [c for c in record.cases if c["result"] != "killed"] + + +def test_the_check_digest_ignores_where_the_store_is() -> None: + a = _rest_cfg("http://127.0.0.1:1") + b = _rest_cfg("http://10.0.0.2:8080") + assert check_digest(a) == check_digest(b) + narrower = parse_verify_config( + { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "rest", + "base_url": "http://127.0.0.1:1", + "records_path": "/api/db?patient=p1", + }, + "effects": _effects(), + } + ) + assert check_digest(narrower) != check_digest(a) + + +def test_admission_policy(rest_copy, tmp_path: Path) -> None: + cfg = _rest_cfg() + record = run_gate(cfg, PARAMS, rest_copy, workdir=tmp_path / "gate") + path = write_coverage(record, tmp_path) + loaded = load_coverage(path) + assert loaded.digest() == record.digest() + policy = AdmissionPolicy(coverage=str(path)) + assert policy.problem(loaded, check_digest(cfg)) is None + assert policy.problem(loaded, "sha256:" + "0" * 64) == "coverage_for_another_check" + weak = loaded.model_copy(update={"killed": loaded.killed - 1}) + assert policy.problem(weak, check_digest(cfg)) == "coverage_below_policy" + recorded = loaded.model_copy(update={"test_copy": "recorded_states"}) + assert policy.problem(recorded, check_digest(cfg)) == "coverage_test_copy_refused" + strict = AdmissionPolicy(coverage=str(path), require_classes=["unrelated_edit"]) + no_frame = run_gate( + cfg, PARAMS, rest_copy, claims=("delta",), workdir=tmp_path / "g2" + ) + assert strict.problem(no_frame, check_digest(cfg)) == "coverage_class_missing" + + +# -- verify-run under an admission policy ---------------------------------------- + + +def test_verify_run_signs_the_coverage_it_was_admitted_with( + rest_copy, tmp_path: Path +) -> None: + import yaml + + from openadapt_flow.verify_run import ( + capture_baseline, + load_verify_config, + receipt_problem, + verify_run, + ) + + record = run_gate(_rest_cfg(), PARAMS, rest_copy, workdir=tmp_path / "gate") + write_coverage(record, tmp_path) + doc = { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "rest", + "base_url": rest_copy.base_url, + "records_path": "/api/db", + }, + "effects": _effects(), + "admission": {"coverage": "check_coverage.json"}, + } + path = tmp_path / "verify.yaml" + path.write_text(yaml.safe_dump(doc)) + + def run(name: str) -> dict: + cfg = load_verify_config(path) + rest_copy.reset() + baseline = tmp_path / f"{name}.json" + capture_baseline(cfg, PARAMS, baseline) + rest_copy.save() + return verify_run( + cfg, + params=PARAMS, + run_record={"executor_kind": "agent", "executor_id": "test-agent"}, + authorization_ref="ticket-0000-admission", + baseline_path=baseline, + private_dir=tmp_path / name, + )["receipt"] + + admitted = run("admitted") + assert admitted["outcome"] == "verified" + assert admitted["checked"]["coverage"] == {**record.totals(), "policy": "met"} + assert receipt_problem(admitted) is None + + (tmp_path / "check_coverage.json").unlink() + missing = run("missing") + assert missing["outcome"] == "reconciliation_required" + assert missing["checked"]["coverage"]["policy"] == "coverage_missing" + + other = record.model_copy(update={"check_sha256": "sha256:" + "1" * 64}) + write_coverage(other, tmp_path) + stale = run("stale") + assert stale["checked"]["coverage"]["policy"] == "coverage_for_another_check" + assert stale["outcome"] != "verified" + forged = { + **admitted, + "checked": { + **admitted["checked"], + "coverage": { + **admitted["checked"]["coverage"], + "policy": "coverage_below_policy", + }, + }, + } + assert receipt_problem(forged) == "verified_without_every_part" + + +# -- scaffold-verifier on the gate ------------------------------------------------ + + +def test_scaffold_verifier_gate_finds_the_scoped_read_blind_spot( + tmp_path: Path, +) -> None: + """The synthesized read is scoped to the patient; another patient's row + edited in the same table is invisible to it. The recorded-state test + can't see that; the gate on a database copy can.""" + + from openadapt_flow.record_check_synthesis import synthesize + from tests._record_check_fixtures import NOTE, write_recording + + db = tmp_path / "app.db" + con = sqlite3.connect(db) + con.executescript( + "CREATE TABLE encounters (id INTEGER PRIMARY KEY, patient_id TEXT, note TEXT);" + "CREATE TABLE billing (id INTEGER PRIMARY KEY, patient_id TEXT, amount TEXT);" + ) + con.execute("INSERT INTO encounters VALUES (1, 'p2', 'an older note')") + con.execute("INSERT INTO encounters VALUES (2, 'p1', ?)", (NOTE,)) + con.execute("INSERT INTO billing VALUES (1, 'p2', '10.00')") + con.commit() + con.close() + rec = write_recording(tmp_path / "rec", {"patient_id": "p1", "note": NOTE}) + + recorded = synthesize(rec, db_path=db) + assert recorded.coverage is not None + assert recorded.coverage.test_copy == "recorded_states" + assert not recorded.coverage.complete # wrong_value and delayed untested + + gated = synthesize(rec, db_path=db, fault_gate=True) + cov = gated.coverage + assert cov is not None and cov.test_copy == "sqlite" + assert cov.clean_pass + survived = {c["fault"] for c in cov.cases if c["result"] == "SURVIVED"} + assert "unrelated_edit" in survived + assert not gated.working + assert any(m.startswith("fault gate unrelated_edit") for m in gated.misses) + assert gated.proof()["coverage"]["test_copy"] == "sqlite" From 7d1f13a18f2b567109e1d27a3af4dad02e003b4d Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 14:19:18 -0400 Subject: [PATCH 2/2] scaffold-verifier: read a scoped target table whole as well The fault gate showed that a --db check whose read is scoped to the patient can't see an edit to another patient's row in the same table. Synthesis now adds an unscoped read of the target table, so the frame sees every row the save didn't declare, and the gated check kills every fault in the suite. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/record_check_synthesis.py | 22 +++++++- packages/flow/tests/test_fault_gate.py | 55 +++++++++++++------ 2 files changed, 58 insertions(+), 19 deletions(-) diff --git a/packages/flow/openadapt_flow/record_check_synthesis.py b/packages/flow/openadapt_flow/record_check_synthesis.py index 547d790ce..7b88b89a3 100644 --- a/packages/flow/openadapt_flow/record_check_synthesis.py +++ b/packages/flow/openadapt_flow/record_check_synthesis.py @@ -884,6 +884,26 @@ def reads_from_sqlite( scope=sorted(set(scope.values())), ) ) + if scope: + # The scoped read can't see another subject's rows in the same + # table, so an edit there would pass. Read the table whole too: + # the frame then sees every row the save didn't declare. + search.surfaces.append( + Surface( + name=_unique_name(f"{table}_all", search.taken), + config={ + "kind": "sql", + "sqlite_database": str(db_path), + "sql_query": query, + }, + after=rows, + before=_minus(rows, row), + before_recorded=False, + source=f"table {table}, every row", + target=row, + bound=bound, + ) + ) finally: conn.close() if complete: @@ -1707,7 +1727,7 @@ def gate_on_database(proposal: Proposal, db_path: Path, cfg: Any) -> Optional[An or "id" not in target.target ): return None - table = target.source.split(" ", 1)[1] + table = target.source.split(" ", 1)[1].split(",", 1)[0] copy_ = SqliteCopy(db_path, table, target.target, saved_key=target.target["id"]) return run_gate(cfg, proposal.params, copy_) diff --git a/packages/flow/tests/test_fault_gate.py b/packages/flow/tests/test_fault_gate.py index 0fa3e3831..f8a39f52e 100644 --- a/packages/flow/tests/test_fault_gate.py +++ b/packages/flow/tests/test_fault_gate.py @@ -259,14 +259,7 @@ def run(name: str) -> dict: # -- scaffold-verifier on the gate ------------------------------------------------ -def test_scaffold_verifier_gate_finds_the_scoped_read_blind_spot( - tmp_path: Path, -) -> None: - """The synthesized read is scoped to the patient; another patient's row - edited in the same table is invisible to it. The recorded-state test - can't see that; the gate on a database copy can.""" - - from openadapt_flow.record_check_synthesis import synthesize +def _scoped_db(tmp_path: Path): from tests._record_check_fixtures import NOTE, write_recording db = tmp_path / "app.db" @@ -280,19 +273,45 @@ def test_scaffold_verifier_gate_finds_the_scoped_read_blind_spot( con.execute("INSERT INTO billing VALUES (1, 'p2', '10.00')") con.commit() con.close() - rec = write_recording(tmp_path / "rec", {"patient_id": "p1", "note": NOTE}) + return db, write_recording(tmp_path / "rec", {"patient_id": "p1", "note": NOTE}) + + +def test_the_gate_finds_a_read_scoped_to_the_patient(tmp_path: Path) -> None: + """A read scoped to the patient can't see another patient's row in the + same table. The recorded-state test can't show that; the gate on a + database copy can.""" + + from openadapt_flow import record_check_synthesis as rcs + + db, rec = _scoped_db(tmp_path) + proposal = rcs.synthesize(rec, db_path=db) + # Only the patient-scoped read, as synthesis built it before it learned + # to read the target table whole. + proposal.search.surfaces = [ + s for s in proposal.search.surfaces if "every row" not in s.source + ] + proposal.effects = rcs.effects_for(proposal.search.surfaces) + proposal.verifier = rcs.verifier_section(proposal.search.surfaces) + cov = rcs.gate_on_database(proposal, db, rcs.gate_config(proposal)) + assert cov is not None and cov.clean_pass + survived = {c["fault"] for c in cov.cases if c["result"] == "SURVIVED"} + assert survived == {"unrelated_edit"} + + recorded = rcs.coverage_from_recorded(proposal.faults, rcs.gate_config(proposal)) + assert recorded.test_copy == "recorded_states" + assert not recorded.complete # wrong_value and delayed untested + - recorded = synthesize(rec, db_path=db) - assert recorded.coverage is not None - assert recorded.coverage.test_copy == "recorded_states" - assert not recorded.coverage.complete # wrong_value and delayed untested +def test_scaffold_verifier_on_the_gate_reads_the_target_table_whole( + tmp_path: Path, +) -> None: + from openadapt_flow.record_check_synthesis import synthesize + db, rec = _scoped_db(tmp_path) gated = synthesize(rec, db_path=db, fault_gate=True) + assert any("every row" in s.source for s in gated.search.surfaces) cov = gated.coverage assert cov is not None and cov.test_copy == "sqlite" - assert cov.clean_pass - survived = {c["fault"] for c in cov.cases if c["result"] == "SURVIVED"} - assert "unrelated_edit" in survived - assert not gated.working - assert any(m.startswith("fault gate unrelated_edit") for m in gated.misses) + assert cov.complete, [c for c in cov.cases if c["result"] != "killed"] + assert gated.working, gated.misses assert gated.proof()["coverage"]["test_copy"] == "sqlite"