diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 9f5e5a866..3fedfa59b 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -136,6 +136,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 9a989292b..90ba715ef 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -533,6 +533,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 097f28ff7..756b5daa4 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -169,6 +169,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 b5cc7fd91..0834f52f2 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -4843,6 +4843,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) @@ -9424,6 +9425,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..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: @@ -1419,6 +1439,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 +1456,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 +1513,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 +1533,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 +1592,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].split(",", 1)[0] + copy_ = SqliteCopy(db_path, table, target.target, saved_key=target.target["id"]) + return run_gate(cfg, proposal.params, copy_) # --------------------------------------------------------------------------- @@ -1606,7 +1786,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 +1800,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 8af3cb2de..00bc06ef7 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 @@ -808,6 +833,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"}``. @@ -1020,15 +1046,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 @@ -1217,9 +1266,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 @@ -1229,6 +1293,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..f8a39f52e --- /dev/null +++ b/packages/flow/tests/test_fault_gate.py @@ -0,0 +1,317 @@ +"""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 _scoped_db(tmp_path: Path): + 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() + 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 + + +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.complete, [c for c in cov.cases if c["result"] != "killed"] + assert gated.working, gated.misses + assert gated.proof()["coverage"]["test_copy"] == "sqlite"