diff --git a/README.md b/README.md index 65cb1fdf..02b34e36 100644 --- a/README.md +++ b/README.md @@ -466,11 +466,22 @@ runtime uses. It returns a signed `RewardEvidenceReceiptV1`: the terminal effect landed, or it didn't, or the store couldn't be read and the episode is unscored. Unscored is never 0. +The worker is not in a published release yet, and the release carrying it has +no date. These two commands are what will work once it lands: + ```bash pip install 'openadapt-flow[reward]' openadapt-flow serve-reward --seed-mockmed --port 8788 ``` +To run it today, install from the repository head: + +```bash +git clone https://github.com/OpenAdaptAI/openadapt-flow +cd openadapt-flow && pip install -e '.[reward]' +python -m openadapt_flow serve-reward --seed-mockmed --port 8788 +``` + A reward receipt isn't an Execute Seal. A model rollout isn't a qualified program, so it never gets one, and the receipt never says Flow governed the policy. The adapters for TRL's `GRPOTrainer` and verl's reward manager live in diff --git a/docs/EFFECT_KIT.md b/docs/EFFECT_KIT.md index 27e3f00e..d85dbdad 100644 --- a/docs/EFFECT_KIT.md +++ b/docs/EFFECT_KIT.md @@ -69,6 +69,7 @@ effects: patient_id: {param: patient_id} # binds to THIS episode's patient type: Triage expected_count: 1 + count_new_only: true # and to THIS episode's write ``` Wrong, and `RewardBundle.load` refuses it: @@ -88,6 +89,40 @@ patient whose record was never read. The load-time guard closes that. A bundle whose required effects select no record by a declared identity key fails to load, and the error names the missing keys and the selector to add. +## The claim names the change + +A required effect that names its subject can still be satisfied by a row that +was already in the store. `record_written` and `field_equals` are statements +about the store's current contents, so a rollout that did nothing collects +the full reward whenever the subject already has a matching row. + +`count_new_only` and `exact_new_set` are the two kinds the judge settles +against the pre-episode baseline, so a reward contract's required effects +must include at least one of them. `RewardBundle.load` refuses a contract +where none does. The required effects are judged as a conjunction, so one +change claim is enough to make `verified` mean the episode added something: +pair a `count_new_only` write with as many `field_equals` read-backs as the +contract needs. + +The baseline comes from `begin_episode`, which the environment calls before +the rollout runs. Without it the change claim is INDETERMINATE and the +episode is unscored, never zero. + +## The tier comes from the mechanism, not the recipe name + +`json_file` and `screen_dump` build the same reader over a JSON document on +the worker's own disk, so they read through the same channel, `ocr`, at tier +0. A contract that declares `file` for a `json_file` recipe fails to load. +Nothing in a JSON document separates a system-of-record dump from a screen +scrape, and letting the recipe kind decide would let the same bytes earn tier +0 under one name and tier 2 under the other. + +`sqlite` earns tier 2 because the worker opens a real database read-only and +runs one SELECT through the engine, and it checks the file header, so a +screen dump renamed `store.db` is refused. `rest` and `fhir` earn it on the +network read the worker performs; whether the endpoint is the customer's +system of record is the bundle author's claim and nobody checks it here. + `idempotency_key` counts as a selector too, because it filters the matched set by `key_field`. `value` does not. On a `field_equals` contract `match` chooses the record and `value` asserts its content, so a subject id in `value` diff --git a/docs/REWARD_WORKER.md b/docs/REWARD_WORKER.md index 2a22fcbf..7392bc25 100644 --- a/docs/REWARD_WORKER.md +++ b/docs/REWARD_WORKER.md @@ -25,12 +25,13 @@ the synthetic fixture), judges the read, and signs the receipt with a local Ed25519 key under `~/.openadapt/reward-ref/`. Evidence bytes (the records it read, the verdicts) stay on that disk. The receipt carries only digests. -The **OpenAdapt control service** is off the high-volume path. It issues and -revokes reward certificates and publishes the calibration corpus digest a +The **OpenAdapt control service** is off the high-volume path. It issues +reward certificates and publishes the calibration corpus digest a certificate names. It never sees an episode. Today the only certificate that exists is synthetic scope, signed by the worker's own key for the MockMed fixture. A production-scope certificate needs the Phase-1 calibration on a -held-out corpus, which is not published. +held-out corpus, which is not published. Nothing revokes a certificate yet; +expiry counts policy updates and is the only thing that ends one. The **trainer node** runs the policy and the optimizer. It submits an episode descriptor to the worker and gets the receipt back. The descriptor is the @@ -44,17 +45,48 @@ record. The trainer-side adapters live in `openadapt_evals.reward` `openadapt_flow.reward.callables` keeps only `HttpRewardClient`, the payload builder, and the receipt's scalar, for the trip between the two machines. -The oracle still has to know which record to read. That identity comes from -one of three places: `metadata.oracle_identity` on the descriptor, an -`oracle_identity` field beside it, or a registration the environment made -with `RewardWorker.begin_episode(episode_id, identity)` before the rollout -ran. The last one also captures the pre-episode baseline, which is what a -`count_new_only` effect needs to tell this episode's write from a record -that was already there. Its keys must match the contract's `identity_keys` -exactly. Matching keys does not mean the oracle read the right record: -it returns the whole collection, so a required effect must select the -subject with a `{param: ...}` reference, and `RewardBundle.load` refuses -a bundle where none does (`docs/EFFECT_KIT.md`). +## Register the episode before the rollout runs + +The environment calls `POST /v1/episodes`, or `RewardWorker.begin_episode` +in process, with the episode id and the subject. Two facts get fixed there, +and both decide what the receipt may say later. + +The **subject** is the record the oracle reads. Registration happens before +anyone knows how the episode will turn out, and the descriptor arrives after, +so the registration wins. A descriptor may repeat the registered subject and +gets a 422 `identity_conflict` if it names a different one. Re-registering +the same episode under a different subject is refused too. Without a +registration the worker still accepts `oracle_identity` on the descriptor, or +`metadata.oracle_identity`, and the keys must match the contract's +`identity_keys` exactly. + +The **baseline** is the store as it stood before the rollout. Required +effects have to include at least one claim about change (`count_new_only` on +a `record_written` effect, or an `exact_new_set` effect), so without a +baseline the judge rules INDETERMINATE and the episode is unscored. That is +what stops a rollout that did nothing from collecting the full reward on a +row that was already there. `RewardBundle.load` refuses a contract whose +required effects only describe the store's current contents. + +Matching keys does not mean the oracle read the right record: it returns the +whole collection, so a required effect must select the subject with a +`{param: ...}` reference, and `RewardBundle.load` refuses a bundle where none +does (`docs/EFFECT_KIT.md`). + +## The policy update only moves forward + +A certificate expires after a stated number of policy updates, and +`policy_update` on the descriptor is the only thing that moves an episode +towards that expiry. It arrives on the wire from the trainer, so the worker +keeps its own ledger under `/policy_updates/` holding the highest +update this contract has scored, and refuses anything below it with 422 +`policy_update_regressed`. Count 0, then 999, then 10^9, and the certificate +reads expired from then on; counting back to 0 is refused rather than +honoured. + +The mark is per contract, not per policy checkpoint. Per checkpoint, a +trainer could rename its checkpoint and start counting from zero again, and +expiry would never arrive. ## The outcome table @@ -85,70 +117,157 @@ current at the episode's policy update. Tier 0 (visual, OCR) and tier 1 certified, whatever the screen shows. A verified tier-2 receipt whose certificate expired is still `verified`, still scored, and not certified. +## What the tier rests on + +The recipe kind picks the channel and the channel sets the tier, so the tier +has to rest on a mechanism the worker can check rather than on a word the +bundle author chose. + +`json_file` and `screen_dump` both hand `JsonDocumentOracle` a JSON document +on the worker's own disk. Nothing in those bytes separates a +system-of-record dump from a screen scrape, so both read through `ocr` at +tier 0, and a contract that declares `file` for a `json_file` recipe fails to +load. One adapter, one channel: the same document cannot be worth tier 0 +under one kind and tier 2 under the other. + +`sqlite` is tier 2 because the worker opens a real database read-only and +runs one SELECT through the engine. It checks the file header, so a screen +dump renamed `store.db` is refused before the worker starts. `build_oracle` +also refuses to hand back any adapter whose channel differs from the recipe +table's, so the two places that have to agree about a tier cannot drift +apart quietly. + +`rest` and `fhir` are tier 2 on the strength of the read being a network +call to an endpoint the bundle names. The worker verifies that it made that +call. It cannot verify that the endpoint is the customer's system of record +rather than a server the trainer stood up, and a self-signed synthetic-scope +certificate does not attest to it. Whoever admits the bundle owns that +check. + ## The MockMed run +The reward worker is not in a published release yet, and the release +carrying it has no date. These two commands are what will work once it +lands: + ```bash pip install 'openadapt-flow[reward]' openadapt-flow serve-reward --seed-mockmed --port 8788 ``` -`--seed-mockmed` writes two contract bundles and their fixtures under the -data directory and serves the tier-2 one when `--contract` is omitted. - -`contracts/mockmed` reads `mockmed/records.json` through the `json_file` -recipe, channel `file`, tier 2. Before it signs the synthetic certificate, -the seed runs 300 ExtraDup trials through the bundle's own judge: each -trial plants one fault (an extra record, a duplicate, a missing record, a -wrong type, or a forbidden discharge) and asks whether the judge accepts it. -The certificate's `epsilon` is the exact one-sided 95% Clopper-Pearson bound -from those counts, the same method the evals proof run uses (its 0 of 15 -gives 0.181036), and `calibration.json` beside it records the trial count -and the false-accept count so you can recompute the bound. With 300 trials -and zero false accepts the bound is 0.0099. The certificate carries -`calibration_scope: synthetic` and `issuer: self_signed`; the types contract -refuses a self-signed certificate with any other scope. +Until then, run it from a checkout of the repository head: + +```bash +git clone https://github.com/OpenAdaptAI/openadapt-flow +cd openadapt-flow && pip install -e '.[reward]' +python -m openadapt_flow serve-reward --seed-mockmed --port 8788 +``` + +`--seed-mockmed` writes two contract bundles and their stores under the data +directory and serves the tier-2 one when `--contract` is omitted. + +`contracts/mockmed` reads `mockmed/records.db` through the `sqlite` recipe, +channel `db`, tier 2. The database starts with two rows for the duplicate +patient and nothing for anybody else, because the required effect asserts a +change: an episode has to write something before it can be verified. + +Before it signs the synthetic certificate the seed runs 300 trials through +the bundle's own judge. The records those trials plant come from the +bundle's own required and forbidden effects, so the fault lands on the rows +this contract talks about. Each trial plants one of six faults: an extra +record, a duplicate, a missing record, a wrong field value, the right record +under another patient, or a forbidden discharge. The certificate's `epsilon` +is the exact one-sided 95% Clopper-Pearson bound from those counts, the same +method the evals proof run uses (its 0 of 15 gives 0.181036). With 300 +trials and zero false accepts the bound is 0.0099. `calibration.json` beside +the certificate records the trial count, the false-accept count, which fault +classes applied, and the corpus digest, so you can recompute the bound and +see what it covers. The certificate carries `calibration_scope: synthetic` +and `issuer: self_signed`; the types contract refuses a self-signed +certificate with any other scope. + +The corpus digest is derived from the contract, and `RewardBundle.load` +refuses a certificate that names a different one. A bound measured on +somebody else's records does not apply to these effects. `contracts/mockmed-tier0` reads `mockmed/screen.json` through the -`screen_dump` recipe, channel `ocr`, tier 0. The dump shows the banner-lie -episode as saved. +`screen_dump` recipe, channel `ocr`, tier 0. -Three episodes to post, with the bearer token and contract digest the banner -prints: +Three episodes to run, with the bearer token and contract digest the banner +prints. Each one registers first, then writes what the episode would have +written, then asks for the reward: ```bash TOKEN=... # printed on start, also in ~/.openadapt/reward-ref/token DIGEST=... # printed on start as "digest", also GET /health -post() { curl -s -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \ - -d "$1" http://127.0.0.1:8788/v1/rewards; } - -post '{"episode_id":"episode_honest_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", +DB=~/.openadapt/reward-ref/mockmed/records.db +call() { curl -s -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \ + -d "$2" "http://127.0.0.1:8788$1"; } +save() { sqlite3 "$DB" \ + "INSERT INTO encounters (patient_id, type, status) VALUES ('$1', 'Triage', 'saved')"; } + +# 1. The honest episode: registered, wrote its row, verified. +call /v1/episodes '{"episode_id":"episode_honest_01", + "oracle_identity":{"patient_id":"patient-honest-0001"}}' +save patient-honest-0001 +call /v1/rewards '{"episode_id":"episode_honest_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", "policy_update":0,"reward_contract_digest":"'$DIGEST'", "metadata":{"oracle_identity":{"patient_id":"patient-honest-0001"}}}' # -> reward_outcome verified, scalar_reward 1.0, certified true, # calibration_scope synthetic -post '{"episode_id":"episode_lie_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", +# 2. The banner lie: registered, wrote nothing to the database. +call /v1/episodes '{"episode_id":"episode_lie_01", + "oracle_identity":{"patient_id":"patient-lie-0002"}}' +call /v1/rewards '{"episode_id":"episode_lie_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", "policy_update":0,"reward_contract_digest":"'$DIGEST'", "metadata":{"oracle_identity":{"patient_id":"patient-lie-0002"}}}' # -> reward_outcome wrong_effect, scalar_reward 0.0. The screen said saved. # The store holds no record. -post '{"episode_id":"episode_dup_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", +# 3. The double submit: two rows where the contract allows one. +call /v1/episodes '{"episode_id":"episode_dup_01", + "oracle_identity":{"patient_id":"patient-dup-0003"}}' +save patient-dup-0003 && save patient-dup-0003 +call /v1/rewards '{"episode_id":"episode_dup_01","policy_checkpoint_id":"policy_checkpoint_mockmed_0", "policy_update":0,"reward_contract_digest":"'$DIGEST'", "metadata":{"oracle_identity":{"patient_id":"patient-dup-0003"}}}' -# -> reward_outcome wrong_effect. Two Triage records where the contract -# allows one. +# -> reward_outcome wrong_effect. Two new Triage records. ``` -Then the tier-0 worker, in a second terminal, with that bundle's own digest: +Register an episode, write nothing, and score it, and the answer is +`wrong_effect` with `scalar_reward` 0.0 even when the patient already has a +row. The required effect asks what this episode added. + +Then the tier-0 worker, in a second terminal, on the same banner-lie +episode. Write the banner into the screen dump and nothing into the +database, and the two channels disagree: ```bash openadapt-flow serve-reward --contract ~/.openadapt/reward-ref/contracts/mockmed-tier0 --port 8789 -post '{"episode_id":"episode_lie_02","policy_checkpoint_id":"policy_checkpoint_mockmed_0", +DIGEST0=... # that bundle's own digest, from its banner or GET /health +call0() { curl -s -H "Authorization: Bearer $TOKEN" -H 'content-type: application/json' \ + -d "$2" "http://127.0.0.1:8789$1"; } + +call0 /v1/episodes '{"episode_id":"episode_lie_02", + "oracle_identity":{"patient_id":"patient-lie-0002"}}' + +# The banner appears. Nothing reaches the database. +python - <<'PY' +import json, pathlib +p = pathlib.Path.home() / ".openadapt/reward-ref/mockmed/screen.json" +body = json.loads(p.read_text()) +body["records"].append({"id": 900, "patient_id": "patient-lie-0002", + "type": "Triage", "status": "saved"}) +p.write_text(json.dumps(body, indent=2)) +PY + +call0 /v1/rewards '{"episode_id":"episode_lie_02","policy_checkpoint_id":"policy_checkpoint_mockmed_0", "policy_update":0,"reward_contract_digest":"'$DIGEST0'", "metadata":{"oracle_identity":{"patient_id":"patient-lie-0002"}}}' # -> reward_outcome verified, development_only true, certified false. -# The OCR dump agrees with the banner. That is why tier 0 cannot certify. +# The OCR dump agrees with the banner while the database holds nothing. +# That is why tier 0 cannot certify. ``` The MockMed banner lie fixture yields 0 because the seeded contract declares @@ -160,6 +279,7 @@ training choice the contract states; the worker never picks one. | Route | Body in | Body out | |---|---|---| | `GET /health` | none | issuer, key fingerprint, contract digest, oracle tier | +| `POST /v1/episodes` | `episode_id` and `oracle_identity` | the registered subject and whether the baseline read | | `POST /v1/rewards` | the episode descriptor | the self-signed envelope, 200, receipt under `receipt` | | `GET /v1/rewards/{receipt_id}` | none | the stored envelope | | `POST /v1/graders/openai` | `{"sample": ..., "item": ...}` | `{"score": 0..1, ...}` or 422 | @@ -169,9 +289,19 @@ envelope carries `issuer: self_signed`, `execute_seal: false`, `production_seal: false`, `flow_governed_policy: false`, `unscored`, and the receipt. It has no top-level `schema_version`, which is how the evals client tells an envelope from a bare receipt. Submitting the same `episode_id` -twice returns 409; a reward is issued once. A descriptor that names a -different contract digest, or none of the three identity sources, returns -422. +twice returns 409; a reward is issued once. + +The 422 cases, all of them refusals to sign something the read does not +support: + +| `error` | When | +|---|---| +| `contract_mismatch` | the descriptor names a contract this worker does not serve | +| `identity_missing` | no registration, no `oracle_identity`, no `metadata.oracle_identity` | +| `identity_mismatch` | the subject's keys are not the contract's `identity_keys` | +| `identity_conflict` | the descriptor names a subject other than the registered one | +| `policy_update_regressed` | the policy update is below the highest this contract has scored | +| `invalid_episode` | the descriptor carries a rollout, screenshot, or PHI field | The OpenAI route mirrors the only custom-grader contract OpenAI documents, the `python` grader's `grade(sample, item) -> float` (graders guide and diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 072d1b74..0311ed9e 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -7794,8 +7794,9 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: "--seed-mockmed", action="store_true", help=( - "Write the synthetic MockMed reward bundles (tier-2 file oracle " - "with a calibrated synthetic certificate; tier-0 screen dump)" + "Write the synthetic MockMed reward bundles (tier-2 read-only " + "SQLite oracle with a calibrated synthetic certificate; tier-0 " + "screen dump)" ), ) p.set_defaults(func=_cmd_serve_reward) @@ -8150,6 +8151,14 @@ def _cmd_serve_reward(args: argparse.Namespace) -> int: print(f" {REWARD_NOTICE}") for label, path in seeded_paths.items(): print(f" seeded {label:<6} {path}") + if seeded_paths: + print(f" store {data_dir / 'mockmed' / 'records.db'}") + print(f" screen {data_dir / 'mockmed' / 'screen.json'}") + print( + " Register each episode with POST /v1/episodes BEFORE the " + "rollout runs. A required effect asserts a change, so the worker " + "needs the pre-episode baseline and fixes the subject in advance." + ) serve(worker, host=args.host, port=args.port) return 0 diff --git a/openadapt_flow/reward/calibration.py b/openadapt_flow/reward/calibration.py index 92854c10..64a85557 100644 --- a/openadapt_flow/reward/calibration.py +++ b/openadapt_flow/reward/calibration.py @@ -2,25 +2,41 @@ A certificate's ``epsilon`` is a bound on the probability that the checker accepts an episode it should have refused. The seed does not invent that -number. It runs the checker over ``n`` synthetic ExtraDup trials (one -extra record, one duplicate record, one missing record, one wrong-type -record per trial, chosen by a seeded generator), counts the false -accepts, and reports the exact one-sided Clopper-Pearson upper bound at -the stated confidence. +number. It runs the checker over ``n`` synthetic trials, one fault per trial +chosen by a seeded generator, counts the false accepts, and reports the exact +one-sided Clopper-Pearson upper bound at the stated confidence. The bound is exact: it uses the binomial tail directly, not a normal approximation. For zero failures it reduces to ``1 - alpha ** (1 / n)``. + +**The corpus comes from the contract under calibration.** A fixed corpus +measures nothing. Feed a store of triage rows to a contract about radiology +rows and every trial refutes for a reason that has nothing to do with the +fault, the false-accept count is zero because the checker rejects everything, +and the bound is the best number the method can produce while bounding no +real behaviour. So :func:`corpus_from_effects` reads the contract's own +required and forbidden effects and builds the records they describe, and +:func:`extradup_trials` refuses to report a bound unless a clean store built +that way actually earns ``VERIFIED``. """ from __future__ import annotations +import json import math import random from dataclasses import dataclass -from typing import Any, Callable, Sequence +from typing import Any, Callable, Iterable, Mapping, Optional, Sequence +from openadapt_types.process_capability import _digest_payload from openadapt_types.reward import RewardOutcomeV1 +from openadapt_flow.runtime.effects.effect import Effect, EffectKind, ValueExpr + + +class CalibrationRefused(ValueError): + """The corpus cannot exercise the contract, so no bound may be issued.""" + def binomial_cdf(k: int, n: int, p: float) -> float: """P(X <= k) for X ~ Binomial(n, p), computed with exact coefficients.""" @@ -68,6 +84,196 @@ def clopper_pearson_upper( return high +#: Every fault a trial may plant. ``wrong_subject`` is the write that landed +#: on somebody else: the record the contract requires exists, correct in every +#: field, under another subject's identity. ``RewardOutcomeV1.WRONG_EFFECT`` +#: names that mode ("a terminal effect that differs from the required one"), +#: so a bound that never planted it did not bound what the receipt claims. +FAULT_CLASSES: tuple[str, ...] = ( + "extra_record", + "duplicate_record", + "missing_record", + "wrong_field", + "wrong_subject", + "forbidden_present", +) + +#: The suffix a ``wrong_field`` trial appends to one declared literal. +_WRONG_FIELD_SUFFIX = "__calibration_wrong" + + +@dataclass(frozen=True) +class CorpusRecipe: + """The records a calibration corpus plants, read off one contract. + + ``intended`` are the records a clean episode leaves behind: one merged + template per record the required effects describe, repeated as many times + as the contract requires it. ``forbidden`` are the records the contract + forbids. Both are templates: a field value is either ``{"literal": ...}`` + or ``{"param": ...}``, and a ``param`` resolves against the trial's + subject identity, exactly as the judge resolves it. + + ``identity_fields`` are the row-id fields an ``exact_new_set`` effect + reads. Every planted record carries them, and ``id``, so newness can be + enumerated. + """ + + identity_keys: tuple[str, ...] + intended: tuple[dict[str, dict[str, str]], ...] + forbidden: tuple[dict[str, dict[str, str]], ...] + identity_fields: tuple[str, ...] + + def as_payload(self) -> dict[str, Any]: + """The canonical form the corpus digest is taken over.""" + + return { + "corpus": "openadapt.reward-derived-corpus/v1", + "identity_keys": list(self.identity_keys), + "intended": [dict(record) for record in self.intended], + "forbidden": [dict(record) for record in self.forbidden], + "identity_fields": list(self.identity_fields), + } + + @property + def applicable_faults(self) -> tuple[str, ...]: + """The fault classes this contract can actually be perturbed by. + + A contract that forbids nothing has no ``forbidden_present`` mode to + false-accept, and a contract whose required records carry no literal + field beyond the subject has no ``wrong_field`` mode. Those classes + are vacuous here rather than untested, and the result names the set + that was sampled so a reader is not left to assume. + """ + + applicable = ["missing_record", "wrong_subject"] + if self.intended: + applicable[:0] = ["extra_record", "duplicate_record"] + if self._mutable_field() is not None: + applicable.append("wrong_field") + if self.forbidden: + applicable.append("forbidden_present") + return tuple(sorted(set(applicable))) + + def _mutable_field(self) -> Optional[tuple[int, str]]: + """The first declared literal a ``wrong_field`` trial may spoil.""" + + for position, record in enumerate(self.intended): + for field in sorted(record): + if field in self.identity_keys or field in self.identity_fields: + continue + if field == "id": + continue + if "literal" in record[field]: + return position, field + return None + + def records( + self, identity: Mapping[str, str], *, first_id: int + ) -> list[dict[str, Any]]: + """Resolve ``intended`` against one subject, with distinct row ids.""" + + return _resolve_all(self.intended, identity, self.identity_fields, first_id) + + def forbidden_records( + self, identity: Mapping[str, str], *, first_id: int + ) -> list[dict[str, Any]]: + return _resolve_all(self.forbidden, identity, self.identity_fields, first_id) + + +def corpus_from_effects( + required: Sequence[Effect], + forbidden: Sequence[Effect], + identity_keys: Sequence[str], +) -> CorpusRecipe: + """Read the records a contract describes off its own effects.""" + + identity_fields = tuple( + sorted( + { + effect.identity_field + for effect in (*required, *forbidden) + if effect.kind is EffectKind.EXACT_NEW_SET + } + ) + ) + return CorpusRecipe( + identity_keys=tuple(sorted(identity_keys)), + intended=tuple(_templates(required)), + forbidden=tuple(_templates(forbidden)), + identity_fields=identity_fields, + ) + + +def corpus_digest(corpus: CorpusRecipe) -> str: + """The digest a certificate must name for this contract's corpus. + + Derived, so a certificate cannot name a corpus it was not calibrated on + and a bundle cannot keep a stale digest after its effects change. + """ + + return _digest_payload(corpus.as_payload()) + + +def corpus_digest_for( + required: Sequence[Effect], + forbidden: Sequence[Effect], + identity_keys: Sequence[str], +) -> str: + return corpus_digest(corpus_from_effects(required, forbidden, identity_keys)) + + +def faulted_store( + fault: str, + identity: Mapping[str, str], + corpus: CorpusRecipe, + rng: random.Random, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """One trial's pre-state and post-state, carrying exactly one fault. + + The pre-state holds rows belonging to other subjects and nothing of this + subject's, so a ``count_new_only`` effect can attribute what it finds. The + post-state adds the corpus records, spoiled by ``fault``. + """ + + other = _other_subject(identity, rng) + before = _resolve_all( + corpus.intended, other, corpus.identity_fields, 100 + rng.randrange(0, 400) + ) + clean = corpus.records(identity, first_id=1) + + if fault == "missing_record": + return before, list(before) + if fault == "wrong_subject": + # The record landed, correct in every field, on somebody else. + return before, [*before, *corpus.records(other, first_id=901)] + if fault == "extra_record": + extra = corpus.records(identity, first_id=501)[:1] + return before, [*before, *clean, *extra] + if fault == "duplicate_record": + # The same row twice, id and all: a re-delivered write, not a + # second distinct record. + return before, [*before, *clean, *clean[:1]] + if fault == "wrong_field": + spoiled = corpus._mutable_field() + if spoiled is None: + raise CalibrationRefused( + "this contract declares no literal field a wrong_field trial " + "could spoil" + ) + position, field = spoiled + current = [dict(record) for record in clean] + current[position][field] = f"{current[position][field]}{_WRONG_FIELD_SUFFIX}" + return before, [*before, *current] + if fault == "forbidden_present": + if not corpus.forbidden: + raise CalibrationRefused("this contract forbids nothing") + return ( + before, + [*before, *clean, *corpus.forbidden_records(identity, first_id=701)], + ) + raise ValueError(f"unknown fault class {fault!r}") + + @dataclass(frozen=True) class CalibrationResult: """What the seed ran and what it found.""" @@ -77,6 +283,8 @@ class CalibrationResult: confidence: float epsilon: float generator_seed: int + corpus_digest: str + fault_classes: tuple[str, ...] def as_metadata(self) -> dict[str, Any]: return { @@ -85,40 +293,65 @@ def as_metadata(self) -> dict[str, Any]: "calibration_confidence": self.confidence, "calibration_generator_seed": self.generator_seed, "calibration_method": "clopper_pearson_one_sided_exact", + "calibration_corpus_digest": self.corpus_digest, + "calibration_fault_classes": list(self.fault_classes), } -FAULT_CLASSES: tuple[str, ...] = ( - "extra_record", - "duplicate_record", - "missing_record", - "wrong_type", - "forbidden_present", -) +#: ``checker(before, current, identity)`` -> the outcome the worker assigns +#: when the store held ``before`` at the start of the episode and holds +#: ``current`` at the end. +Checker = Callable[ + [Sequence[dict[str, Any]], Sequence[dict[str, Any]], dict[str, str]], + RewardOutcomeV1, +] def extradup_trials( - checker: Callable[[Sequence[dict[str, Any]], dict[str, str]], RewardOutcomeV1], + checker: Checker, + corpus: CorpusRecipe, *, trials: int, generator_seed: int, confidence: float = 0.95, + corpus_digest: str, ) -> CalibrationResult: """Run the checker over faulted stores and bound its false-accept rate. - ``checker(records, identity)`` returns the reward outcome the checker - assigns when the store holds ``records`` and the episode claims the - required effect for ``identity``. A false accept is ``VERIFIED`` on a - faulted store. + A false accept is ``VERIFIED`` on a faulted store. + + Refuses before it counts anything when the corpus does not exercise the + contract: a clean store built from the contract's own effects must earn + ``VERIFIED``, and at least one fault class must be applicable. Without + the first, every trial refutes for a reason the fault did not cause and + zero false accepts means only that the checker rejects everything. """ + applicable = corpus.applicable_faults + if not applicable: + raise CalibrationRefused( + "no fault class applies to this contract, so its false-accept " + "rate cannot be sampled and no certificate may be issued" + ) + control_identity = _trial_identity(corpus.identity_keys, "control") + control_before, control_current = _clean_store(corpus, control_identity) + control = checker(control_before, control_current, control_identity) + if control is not RewardOutcomeV1.VERIFIED: + raise CalibrationRefused( + "a clean store built from this contract's own required effects " + f"judges {control.value}, not verified, so the corpus does not " + "exercise the contract and a zero false-accept count would bound " + "nothing. Check that the required effects describe records the " + "same read can hold." + ) + rng = random.Random(generator_seed) false_accepts = 0 for index in range(trials): - fault = FAULT_CLASSES[rng.randrange(len(FAULT_CLASSES))] - patient = f"patient-cal-{index:04d}" - records = faulted_store(fault, patient, rng) - if checker(records, {"patient_id": patient}) is RewardOutcomeV1.VERIFIED: + fault = applicable[rng.randrange(len(applicable))] + identity = _trial_identity(corpus.identity_keys, f"{index:04d}") + before, current = faulted_store(fault, identity, corpus, rng) + if checker(before, current, identity) is RewardOutcomeV1.VERIFIED: false_accepts += 1 return CalibrationResult( trials=trials, @@ -126,37 +359,111 @@ def extradup_trials( confidence=confidence, epsilon=clopper_pearson_upper(false_accepts, trials, confidence=confidence), generator_seed=generator_seed, + corpus_digest=corpus_digest, + fault_classes=applicable, ) -def faulted_store(fault: str, patient: str, rng: random.Random) -> list[dict[str, Any]]: - """A MockMed-shaped store carrying one fault for ``patient``.""" +# -- deriving the records a contract describes ------------------------------- - noise = [ - { - "id": 100 + i, - "patient_id": f"patient-other-{rng.randrange(10_000):04d}", - "type": "Triage", - "status": "saved", - } - for i in range(rng.randrange(0, 4)) - ] - intended = {"id": 1, "patient_id": patient, "type": "Triage", "status": "saved"} - if fault == "extra_record": - extra = {"id": 2, "patient_id": patient, "type": "Triage", "status": "saved"} - return [*noise, intended, extra] - if fault == "duplicate_record": - return [*noise, intended, dict(intended, id=3)] - if fault == "missing_record": - return noise - if fault == "wrong_type": - return [*noise, dict(intended, type="Consult")] - if fault == "forbidden_present": - discharge = { - "id": 4, - "patient_id": patient, - "type": "Discharge", - "status": "saved", - } - return [*noise, intended, discharge] - raise ValueError(f"unknown fault class {fault!r}") + +def _expr_template(expr: ValueExpr) -> dict[str, str]: + if expr.param is not None: + return {"param": expr.param} + return {"literal": str(expr.literal)} + + +def _selector_template( + selector: Mapping[str, ValueExpr], +) -> dict[str, dict[str, str]]: + return {field: _expr_template(expr) for field, expr in selector.items()} + + +def _key_of(template: Mapping[str, dict[str, str]]) -> str: + return json.dumps(template, sort_keys=True) + + +def _templates(effects: Iterable[Effect]) -> list[dict[str, dict[str, str]]]: + """One merged record template per record the effects describe. + + Effects that select the same record contribute to one template: a + ``field_equals`` read-back of a row a ``record_written`` effect also + requires describes ONE row, not two, and planting two would make the + clean store refute on cardinality. + """ + + merged: dict[str, dict[str, dict[str, str]]] = {} + counts: dict[str, int] = {} + order: list[str] = [] + + def add(template: dict[str, dict[str, str]], key_template: Any, count: int) -> None: + key = _key_of(key_template) + if key not in merged: + merged[key] = {} + counts[key] = 0 + order.append(key) + merged[key].update(template) + counts[key] = max(counts[key], count) + + for effect in effects: + match = _selector_template(effect.match) + if effect.kind is EffectKind.EXACT_NEW_SET: + for selector in effect.new_records: + member = {**match, **_selector_template(selector)} + add(member, member, 1) + continue + if effect.expected_count <= 0: + # An absence claim plants nothing; a clean store satisfies it. + continue + template = dict(match) + if effect.idempotency_key is not None: + template[effect.key_field] = _expr_template(effect.idempotency_key) + if effect.kind is EffectKind.FIELD_EQUALS: + if effect.field and effect.value is not None: + template[effect.field] = _expr_template(effect.value) + add(template, match, 1) + continue + add(template, match, effect.expected_count) + + records: list[dict[str, dict[str, str]]] = [] + for key in order: + records.extend(dict(merged[key]) for _ in range(counts[key])) + return records + + +def _resolve_all( + templates: Sequence[Mapping[str, dict[str, str]]], + identity: Mapping[str, str], + identity_fields: Sequence[str], + first_id: int, +) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for offset, template in enumerate(templates): + row_id = first_id + offset + record: dict[str, Any] = {"id": row_id} + for field in identity_fields: + record[field] = row_id + for field, expr in template.items(): + if "param" in expr: + record[field] = identity.get(expr["param"], "") + else: + record[field] = expr["literal"] + records.append(record) + return records + + +def _clean_store( + corpus: CorpusRecipe, identity: Mapping[str, str] +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """The control trial: nothing wrong, so the checker must say verified.""" + + return [], corpus.records(identity, first_id=1) + + +def _trial_identity(identity_keys: Sequence[str], tag: str) -> dict[str, str]: + return {key: f"{key}-cal-{tag}" for key in identity_keys} + + +def _other_subject(identity: Mapping[str, str], rng: random.Random) -> dict[str, str]: + suffix = rng.randrange(10_000) + return {key: f"{key}-other-{suffix:04d}" for key in identity} diff --git a/openadapt_flow/reward/models.py b/openadapt_flow/reward/models.py index 979e1e5e..6223f843 100644 --- a/openadapt_flow/reward/models.py +++ b/openadapt_flow/reward/models.py @@ -113,12 +113,15 @@ class EpisodeDescriptorV1(_Strict): ``environment_id``, ``metadata``. The digest binds the receipt to the contract this worker serves; a different digest is refused. - The oracle needs to know which record to read. That identity comes from - one of three places, checked in this order: the ``oracle_identity`` - field, ``metadata["oracle_identity"]``, or a registration made by - ``RewardWorker.begin_episode`` before the rollout ran. Its keys must be - exactly the contract's ``oracle.identity_keys``; an extra key is refused, - a missing key is refused. + The oracle needs to know which record to read. A registration made by + ``RewardWorker.begin_episode`` before the rollout ran decides that, and + the descriptor cannot replace it: a descriptor that names a different + subject is refused, because the registration was made before anyone knew + how the episode would end and the descriptor arrives after. The + ``oracle_identity`` field and ``metadata["oracle_identity"]`` still name + the subject for an episode with no registration, in that order. Its keys + must be exactly the contract's ``oracle.identity_keys``; an extra key is + refused, a missing key is refused. ``runtime_signal`` (or ``metadata["runtime_signal"]``) is what the episode runtime reported about its own end. The oracle read decides; the @@ -279,10 +282,20 @@ def token(self) -> Optional[str]: return os.environ.get(self.token_env) or None -#: The recipe kind sets the channel, and the channel sets the tier. A -#: payload cannot upgrade a screen dump into a system-of-record read. +#: The channel each recipe kind reads through, and the channel sets the tier. +#: A payload cannot upgrade a screen dump into a system-of-record read. +#: +#: This table must agree with the adapter +#: :func:`openadapt_flow.reward.oracles.build_oracle` returns, and that +#: function refuses to hand back an adapter whose channel differs. Kinds that +#: share an adapter therefore share a channel: ``json_file`` and +#: ``screen_dump`` both build :class:`~openadapt_flow.reward.oracles. +#: JsonDocumentOracle` over a JSON document on the worker's own disk, so both +#: read through ``ocr`` at tier 0. Nothing in those bytes separates a +#: system-of-record dump from a screen scrape, and letting the recipe kind +#: pick would let one document earn two tiers. _RECIPE_CHANNEL: dict[str, OracleChannel] = { - "json_file": OracleChannel.FILE, + "json_file": OracleChannel.OCR, "screen_dump": OracleChannel.OCR, "rest": OracleChannel.API, "sqlite": OracleChannel.DB, @@ -328,16 +341,27 @@ class RewardBundle(_Strict): Loading refuses when any digest in the contract disagrees with the file it names, when the oracle recipe's channel disagrees with the contract's oracle channel, when an effect references an identity - key the contract does not declare, or when the required effects - between them select no record by a declared identity key. + key the contract does not declare, when the required effects + between them select no record by a declared identity key, or when no + required effect makes a claim about change. - That last rule is what ties a judgement to a subject. An oracle reads a + The identity rule is what ties a judgement to a subject. An oracle reads a whole collection and does not filter it by ``oracle_identity`` (see :meth:`openadapt_flow.reward.oracles.VerifierOracle.read`), so the required effect's own selector is the only place the subject is applied. A contract that declares ``identity_keys`` and then matches on content alone would accept a write that landed on somebody else, and the receipt would still name the declared subject. + + The change rule is what ties a judgement to an episode. A plain + ``record_written`` effect is a statement about the store's current + contents, so it is satisfied by a row that was already there before the + rollout started, and an episode that did nothing at all earns the full + reward. Only ``count_new_only`` and ``exact_new_set`` compare the read + against the pre-episode baseline, so at least one required effect must be + one of those. The required effects are judged as a conjunction, so one + change claim is enough: ``VERIFIED`` then means every required effect + held AND the episode added a record. """ model_config = ConfigDict(extra="forbid", frozen=True, arbitrary_types_allowed=True) @@ -410,6 +434,16 @@ def load(cls, directory: Path | str) -> "RewardBundle": f"to the named subject. Add {{{example}}} to the match selector " f"of a required effect." ) + if not any(effect.requires_baseline for effect in required): + raise BundleError( + "required_effects assert only what the store holds now, so " + "they are satisfied by a record that was already there and a " + "rollout that did nothing scores the full reward. At least " + "one required effect must be a claim about change, which the " + "judge can only settle against the pre-episode baseline: set " + '"count_new_only": true on a record_written effect, or ' + "declare an exact_new_set effect." + ) certificate: Optional[RewardCertificateV1] = None cert_path = base / CERTIFICATE_FILE if cert_path.is_file(): @@ -418,6 +452,7 @@ def load(cls, directory: Path | str) -> "RewardBundle": raise BundleError( "certificate.json binds a different reward contract digest" ) + _check_corpus_digest(contract, certificate, required, forbidden) return cls( directory=base, contract=contract, @@ -449,6 +484,47 @@ def check_identity(self, identity: dict[str, str]) -> None: ) +def _check_corpus_digest( + contract: RewardContractV1, + certificate: RewardCertificateV1, + required: tuple[Effect, ...], + forbidden: tuple[Effect, ...], +) -> None: + """Refuse a certificate calibrated on a corpus that is not this contract's. + + ``epsilon`` bounds how often the checker accepts an episode it should have + refused, and it only bounds that for the records it was measured on. A + corpus of triage rows says nothing about a contract that asks for + radiology rows: every trial refutes for a reason the planted fault did not + cause, the false-accept count is zero, and the bound is the best number + the method can produce while bounding nothing. + + So the corpus is derived from the contract's own effects and named by + digest, and both the policy and the certificate must name that digest. + """ + + from openadapt_flow.reward.calibration import corpus_digest_for + + want = corpus_digest_for(required, forbidden, contract.oracle.identity_keys) + for label, got in ( + ( + "certificate_policy.calibration_corpus_digest", + contract.certificate_policy.calibration_corpus_digest, + ), + ( + "certificate.calibration_corpus_digest", + certificate.calibration_corpus_digest, + ), + ): + if got != want: + raise BundleError( + f"{label} is {got}, and the corpus derived from this " + f"contract's own required and forbidden effects digests to " + f"{want}. A bound measured on other records does not apply " + f"to these effects." + ) + + class BundleError(ValueError): """The bundle on disk does not match its contract.""" diff --git a/openadapt_flow/reward/oracles.py b/openadapt_flow/reward/oracles.py index 119b2b04..d5903376 100644 --- a/openadapt_flow/reward/oracles.py +++ b/openadapt_flow/reward/oracles.py @@ -10,7 +10,18 @@ The REST, SQL, FHIR, and file-arrival adapters wrap the effect-verifier kit (``docs/EFFECT_KIT.md``) so the read logic, the read-only SQL whitelist, and the "unreadable means INDETERMINATE" rule are the same code the runtime -uses. ``json_file`` and ``screen_dump`` are the synthetic MockMed fixtures. +uses. ``json_file`` and ``screen_dump`` read a JSON document on disk, which +is what the synthetic tier-0 fixture needs. + +The channel belongs to the adapter, not to the recipe kind. Two recipe +kinds that build the same adapter over the same bytes read through the same +channel and earn the same tier. ``json_file`` and ``screen_dump`` are that +pair: both hand :class:`JsonDocumentOracle` a JSON document on the worker's +own disk, and nothing in the bytes tells a system-of-record dump from a +screen scrape. So both read through ``ocr`` and both stay at tier 0. +Claiming tier 2 needs a channel whose mechanism the worker can check: a +SQLite database file it opens read-only, a REST or FHIR endpoint it calls +over the network, or a directory it lists. """ from __future__ import annotations @@ -66,22 +77,33 @@ def effect_state_of(observed: OracleObservation, substrate: str) -> EffectState: class JsonDocumentOracle: - """Read a JSON document of records from disk. - - ``channel`` is ``file`` for a system-of-record dump and ``ocr`` for the - synthetic screen dump. The same reader, two tiers, on purpose: the tier - comes from what the document is, not from how it is parsed. + """Read a JSON document of records from disk. Always channel ``ocr``. + + The channel is a class attribute and no caller may set it. Both the + ``json_file`` and the ``screen_dump`` recipe kinds build this adapter, so + a bundle that could pick the channel could hand the same bytes two + different tiers: point ``screen_dump`` at a document and the receipt says + tier 0 and ``development_only``; point ``json_file`` at the same document + and it says tier 2 and ``certified``. + + Tier 0 is the honest floor for this reader. A JSON document on the + worker's own disk carries nothing that separates a system-of-record dump + from a screen scrape, and whoever writes the file writes the answer. Use + ``sqlite``, ``rest``, ``fhir``, or ``file_arrival`` when the read really + is a record channel. """ + #: Fixed. See the class docstring for why this is not a constructor + #: argument. + channel: OracleChannel = OracleChannel.OCR + def __init__( self, path: Path | str, *, - channel: OracleChannel = OracleChannel.FILE, records_key: Optional[str] = "records", ) -> None: self.path = Path(path) - self.channel = channel self.records_key = records_key def read(self, identity: Mapping[str, str]) -> OracleObservation: @@ -137,19 +159,50 @@ def read(self, identity: Mapping[str, str]) -> OracleObservation: return observation(self.channel, identity, list(state.records)) -def build_oracle(recipe: OracleRecipeV1, base_dir: Path) -> OracleAdapter: - """Construct the adapter a recipe names. Secrets come from the environment.""" +class OracleMechanismError(ValueError): + """The recipe cannot read through the channel it claims.""" - if recipe.kind == "json_file": - return JsonDocumentOracle( - recipe.resolve_path(base_dir), - channel=OracleChannel.FILE, - records_key=recipe.records_key, + +#: The first sixteen bytes of every SQLite database file, from the file +#: format specification. A JSON screen dump renamed ``store.db`` does not +#: carry them, so the ``db`` channel cannot be claimed by renaming a file. +_SQLITE_MAGIC = b"SQLite format 3\x00" + + +def assert_sqlite_database(path: Path) -> Path: + """Refuse a ``sqlite`` recipe path that is not a SQLite database file. + + The ``db`` channel is tier 2. What earns that tier is the mechanism: the + worker opens a real database read-only and runs one SELECT through the + engine. Checking the file header is what stops the recipe kind from being + the whole claim. + """ + + try: + with path.open("rb") as handle: + header = handle.read(len(_SQLITE_MAGIC)) + except OSError as exc: + raise OracleMechanismError( + f"oracle recipe sqlite cannot open {path}: {exc}" + ) from exc + if header != _SQLITE_MAGIC: + raise OracleMechanismError( + f"{path} is not a SQLite database file, so this recipe cannot " + "read through the db channel. A JSON document reads through the " + "ocr channel at tier 0; use the screen_dump or json_file kind " + "for one." ) - if recipe.kind == "screen_dump": + return path + + +def _adapter_for(recipe: OracleRecipeV1, base_dir: Path) -> OracleAdapter: + """Construct the adapter a recipe names. Secrets come from the environment.""" + + if recipe.kind in {"json_file", "screen_dump"}: + # One adapter, one channel. Neither kind may claim the other's tier; + # see the JsonDocumentOracle docstring. return JsonDocumentOracle( recipe.resolve_path(base_dir), - channel=OracleChannel.OCR, records_key=recipe.records_key, ) if recipe.kind == "rest": @@ -168,7 +221,7 @@ def build_oracle(recipe: OracleRecipeV1, base_dir: Path) -> OracleAdapter: if recipe.kind == "sqlite": from openadapt_flow.runtime.effects.sql import SqlRecordVerifier - database = recipe.resolve_path(base_dir) + database = assert_sqlite_database(recipe.resolve_path(base_dir)) def connect() -> sqlite3.Connection: uri = f"file:{database}?mode=ro" @@ -202,3 +255,24 @@ def connect() -> sqlite3.Connection: channel=OracleChannel.FILE, ) raise ValueError(f"unknown oracle recipe kind {recipe.kind!r}") + + +def build_oracle(recipe: OracleRecipeV1, base_dir: Path) -> OracleAdapter: + """Build the adapter and refuse it when its channel is not the recipe's. + + ``OracleRecipeV1.channel`` is a table keyed on the recipe kind, and the + contract's declared channel is checked against that table when the bundle + loads. Neither of those reads the adapter. This function closes the loop: + the adapter the worker actually holds decides, and a table that drifts + away from it stops the worker instead of shipping a tier the read cannot + support. + """ + + adapter = _adapter_for(recipe, base_dir) + if adapter.channel is not recipe.channel: + raise OracleMechanismError( + f"oracle recipe {recipe.kind} declares channel " + f"{recipe.channel.value} but its adapter reads through " + f"{adapter.channel.value}" + ) + return adapter diff --git a/openadapt_flow/reward/seed.py b/openadapt_flow/reward/seed.py index 33c117f9..4eb97719 100644 --- a/openadapt_flow/reward/seed.py +++ b/openadapt_flow/reward/seed.py @@ -1,17 +1,26 @@ """Synthetic MockMed reward fixtures for ``serve-reward --seed-mockmed``. -Two bundles and one records file: - -* ``contracts/mockmed`` reads ``mockmed/records.json`` through the - ``json_file`` recipe (channel ``file``, tier 2) and carries a self-signed - certificate issued at policy update 0. Episode ``honest`` finds its - encounter and scores ``verified``. Episode ``banner-lie`` finds nothing: - the screen said saved, the store holds no record, ``wrong_effect`` at - the contract's declared penalty of 0. +Two bundles and two stores: + +* ``contracts/mockmed`` reads ``mockmed/records.db`` through the ``sqlite`` + recipe (channel ``db``, tier 2) and carries a self-signed certificate + issued at policy update 0. The database starts with two rows for the + duplicate patient and nothing for anybody else, so an episode has to write + something before it can be verified. * ``contracts/mockmed-tier0`` reads ``mockmed/screen.json`` through the - ``screen_dump`` recipe (channel ``ocr``, tier 0). The screen dump shows - the banner-lie encounter as saved. The receipt is ``development_only`` - and never certified, whatever the dump says. + ``screen_dump`` recipe (channel ``ocr``, tier 0). Its receipts are + ``development_only`` and never certified, whatever the dump says. + +Read the two together and the fixture makes its point: write the banner into +the screen dump and nothing into the database, and the tier-0 worker says +``verified`` while the tier-2 worker says ``wrong_effect`` about the same +episode. + +The tier-2 store is a SQLite database rather than a JSON document because the +tier has to rest on something the worker can check. Both JSON recipe kinds +build the same reader over the same kind of bytes, so neither can claim a +record channel; ``openadapt_flow.reward.oracles.assert_sqlite_database`` +refuses a ``sqlite`` recipe whose file is not a real database. Nothing here is a production recipe. Both bundles are synthetic. """ @@ -20,6 +29,7 @@ import base64 import json +import sqlite3 from pathlib import Path from typing import Any @@ -27,7 +37,12 @@ from openadapt_types.process_capability import _digest_payload, canonical_json_bytes from openadapt_types.reward import RewardCertificateV1, RewardContractV1 -from openadapt_flow.reward.calibration import CalibrationResult, extradup_trials +from openadapt_flow.reward.calibration import ( + CalibrationResult, + corpus_digest_for, + corpus_from_effects, + extradup_trials, +) from openadapt_flow.reward.models import ( CERTIFICATE_FILE, CONTRACT_FILE, @@ -36,6 +51,7 @@ REQUIRED_EFFECTS_FILE, RewardBundle, ) +from openadapt_flow.runtime.effects.effect import Effect MOCKMED_TASK_ID = "task_mockmed_encounter_note" MOCKMED_ENVIRONMENT_ID = "environment_mockmed_synthetic" @@ -47,36 +63,27 @@ MOCKMED_CHECKPOINT = "policy_checkpoint_mockmed_0" CERTIFICATE_EXPIRY_UPDATES = 1000 CALIBRATION_FILE = "calibration.json" -#: ExtraDup trials the seed runs before it signs the synthetic certificate. -#: 300 trials with zero false accepts bound the rate at 0.0099 (95%). +#: Trials the seed runs before it signs the synthetic certificate. 300 trials +#: with zero false accepts bound the rate at 0.0099 (95%). CALIBRATION_TRIALS = 300 CALIBRATION_SEED = 20260901 CALIBRATION_CONFIDENCE = 0.95 -_RECORDS: list[dict[str, Any]] = [ - { - "id": 1, - "patient_id": MOCKMED_HONEST_PATIENT, - "type": "Triage", - "status": "saved", - }, - { - "id": 2, - "patient_id": MOCKMED_DUPLICATE_PATIENT, - "type": "Triage", - "status": "saved", - }, - { - "id": 3, - "patient_id": MOCKMED_DUPLICATE_PATIENT, - "type": "Triage", - "status": "saved", - }, +#: The tier-2 store's one table and the read-only SELECT the oracle runs. +MOCKMED_TABLE = "encounters" +MOCKMED_QUERY = f"SELECT id, patient_id, type, status FROM {MOCKMED_TABLE}" + +#: What the database holds before any episode runs. The honest patient and +#: the banner-lie patient have no row: a required effect now asserts a change, +#: so a row that was already there earns nothing. +_ROWS: list[dict[str, Any]] = [ + {"id": 2, "patient_id": MOCKMED_DUPLICATE_PATIENT, "type": "Triage"}, + {"id": 3, "patient_id": MOCKMED_DUPLICATE_PATIENT, "type": "Triage"}, ] +#: What the screen shows before any episode runs. _SCREEN: list[dict[str, Any]] = [ - {"patient_id": MOCKMED_HONEST_PATIENT, "type": "Triage", "status": "saved"}, - {"patient_id": MOCKMED_LIE_PATIENT, "type": "Triage", "status": "saved"}, + {"id": 2, "patient_id": MOCKMED_DUPLICATE_PATIENT, "type": "Triage"}, ] _REQUIRED_EFFECTS: list[dict[str, Any]] = [ @@ -84,6 +91,10 @@ "kind": "record_written", "match": {"patient_id": {"param": "patient_id"}, "type": "Triage"}, "expected_count": 1, + # The claim is about what this episode added, not about what the + # store happens to hold. Without it a rollout that did nothing + # scores the full reward whenever the subject already had a row. + "count_new_only": True, } ] @@ -99,12 +110,13 @@ def seed_mockmed( data_dir: Path, key: Ed25519PrivateKey, issuer_key_id: str ) -> dict[str, Path]: - """Write both bundles and the records files. Returns the bundle paths.""" + """Write both bundles and both stores. Returns the bundle paths.""" data_dir = Path(data_dir) store = data_dir / "mockmed" store.mkdir(parents=True, exist_ok=True) - _write(store / "records.json", {"records": _RECORDS}) + database = store / "records.db" + write_mockmed_database(database, _ROWS) _write(store / "screen.json", {"records": _SCREEN}) tier2 = data_dir / "contracts" / "mockmed" @@ -112,11 +124,11 @@ def seed_mockmed( tier2, contract_id=MOCKMED_CONTRACT_ID, oracle={ - "kind": "json_file", - "path": str(store / "records.json"), - "records_key": "records", + "kind": "sqlite", + "path": str(database), + "query": MOCKMED_QUERY, }, - channel="file", + channel="db", key=key, issuer_key_id=issuer_key_id, certify=True, @@ -138,6 +150,63 @@ def seed_mockmed( return {"tier2": tier2, "tier0": tier0} +def write_mockmed_database(path: Path, rows: list[dict[str, Any]]) -> Path: + """Create the synthetic tier-2 store as a real SQLite database.""" + + path.parent.mkdir(parents=True, exist_ok=True) + if path.exists(): + path.unlink() + connection = sqlite3.connect(path) + try: + connection.execute( + f"CREATE TABLE {MOCKMED_TABLE} (" + "id INTEGER PRIMARY KEY, patient_id TEXT NOT NULL, " + "type TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'saved')" + ) + connection.executemany( + f"INSERT INTO {MOCKMED_TABLE} (id, patient_id, type, status) " + "VALUES (:id, :patient_id, :type, :status)", + [{"status": "saved", **row} for row in rows], + ) + connection.commit() + finally: + connection.close() + return path + + +def write_mockmed_encounter( + database: Path, patient_id: str, *, type_: str = "Triage" +) -> None: + """Add one encounter row, standing in for what an episode would write.""" + + connection = sqlite3.connect(database) + try: + connection.execute( + f"INSERT INTO {MOCKMED_TABLE} (patient_id, type, status) " + "VALUES (?, ?, 'saved')", + (patient_id, type_), + ) + connection.commit() + finally: + connection.close() + + +def write_mockmed_banner(screen: Path, patient_id: str) -> None: + """Add one row to the screen dump, standing in for a save banner.""" + + body = json.loads(screen.read_text(encoding="utf-8")) + records = list(body.get("records") or []) + records.append( + { + "id": 100 + len(records), + "patient_id": patient_id, + "type": "Triage", + "status": "saved", + } + ) + _write(screen, {"records": records}) + + def mockmed_episode( patient_id: str, *, @@ -185,7 +254,14 @@ def write_bundle( _FORBIDDEN_EFFECTS if forbidden_effects is None else forbidden_effects ) oracle_doc = _canonical(oracle) - corpus_digest = _digest_payload({"corpus": "synthetic-mockmed", "size": 0}) + keys = ["patient_id"] if identity_keys is None else list(identity_keys) + # The corpus a certificate is calibrated on comes from the contract's own + # effects, so its digest is computed here and not chosen. + corpus_digest = corpus_digest_for( + [Effect.model_validate(item) for item in required], + [Effect.model_validate(item) for item in forbidden], + keys, + ) contract = RewardContractV1.model_validate( { "contract_id": contract_id, @@ -200,9 +276,7 @@ def write_bundle( "forbidden_effect_contract_digest": _digest_payload(forbidden), "oracle": { "channel": channel, - "identity_keys": ( - ["patient_id"] if identity_keys is None else list(identity_keys) - ), + "identity_keys": keys, "oracle_contract_digest": _digest_payload(oracle_doc), }, "components": [{"name": "terminal_effect", "weight": 1.0}], @@ -238,11 +312,15 @@ def write_bundle( def calibrate_bundle(directory: Path) -> CalibrationResult: - """Run the seeded ExtraDup trials through this bundle's judge. - - The certificate's ``epsilon`` is the exact one-sided Clopper-Pearson - bound from these counts. The counts are written beside the certificate - so a reader can recompute the bound. + """Run the seeded trials through this bundle's judge. + + The corpus is read off the bundle's own required and forbidden effects, + so the faults it plants are faults in the records this contract talks + about. The certificate's ``epsilon`` is the exact one-sided + Clopper-Pearson bound from these counts, and ``calibration.json`` beside + the certificate records the counts, the corpus digest, and which fault + classes applied, so a reader can recompute the bound and see what it + covers. """ from openadapt_types.oracle import OracleChannel @@ -253,17 +331,24 @@ def calibrate_bundle(directory: Path) -> CalibrationResult: bundle = RewardBundle.load(directory) channel = OracleChannel(bundle.contract.oracle.channel) + corpus = corpus_from_effects( + bundle.required_effects, + bundle.forbidden_effects, + bundle.identity_keys, + ) - def checker(records: Any, identity: dict[str, str]) -> Any: - before = EffectState(substrate=channel.value, reachable=False) - observed = observation(channel, identity, list(records)) - return judge_episode(bundle, identity, "completed", before, observed).outcome + def checker(before: Any, current: Any, identity: dict[str, str]) -> Any: + pre = EffectState(substrate=channel.value, reachable=True, records=list(before)) + observed = observation(channel, identity, list(current)) + return judge_episode(bundle, identity, "completed", pre, observed).outcome return extradup_trials( checker, + corpus, trials=CALIBRATION_TRIALS, generator_seed=CALIBRATION_SEED, confidence=CALIBRATION_CONFIDENCE, + corpus_digest=bundle.contract.certificate_policy.calibration_corpus_digest, ) diff --git a/openadapt_flow/reward/serve.py b/openadapt_flow/reward/serve.py index 128bacf6..317b53e3 100644 --- a/openadapt_flow/reward/serve.py +++ b/openadapt_flow/reward/serve.py @@ -3,6 +3,11 @@ Routes: * ``GET /health``: issuer, key fingerprint, contract digest, oracle tier. +* ``POST /v1/episodes``: register one episode's subject and capture the + pre-episode baseline. The environment calls this BEFORE the rollout runs. + It is what makes the graded subject a fact settled in advance rather than + a field the trainer fills in once it knows how the episode went, and it is + what gives a ``count_new_only`` effect something to compare against. * ``POST /v1/rewards``: episode descriptor in, self-signed envelope out (200, the receipt under ``receipt``). The descriptor is the shape ``openadapt_evals.reward.receipts.EpisodeDescriptor`` sends. An unscored @@ -91,6 +96,37 @@ def health() -> dict[str, Any]: "notice": REWARD_NOTICE, } + @app.post("/v1/episodes", status_code=200, response_model=None) + async def begin_episode( + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(worker, authorization) + payload = await _json_object(request) + episode_id = payload.get("episode_id") + identity = payload.get("oracle_identity") + if not isinstance(episode_id, str) or not episode_id: + raise HTTPException(status_code=400, detail="episode_id is required") + if not isinstance(identity, dict) or not identity: + raise HTTPException( + status_code=400, detail="oracle_identity must be a non-empty object" + ) + subject = {str(key): str(value) for key, value in identity.items()} + try: + state = worker.begin_episode(episode_id, subject) + except ValueError as exc: # IdentityError and friends + raise RewardWorkerError(422, "identity_mismatch", str(exc)) from exc + return JSONResponse( + status_code=200, + content={ + "episode_id": episode_id, + "oracle_identity": subject, + "baseline_reachable": state.reachable, + "baseline_record_count": len(state.records), + }, + headers=_issuer_headers(worker), + ) + @app.post("/v1/rewards", status_code=200, response_model=None) async def create_reward( request: Request, diff --git a/openadapt_flow/reward/worker.py b/openadapt_flow/reward/worker.py index a9458337..e6da8430 100644 --- a/openadapt_flow/reward/worker.py +++ b/openadapt_flow/reward/worker.py @@ -3,7 +3,8 @@ One worker holds one reward contract bundle, one oracle adapter, and one local signing key. ``score_episode`` is the whole path: -1. check the episode's identity keys against the contract; +1. settle which subject is graded and check its keys against the contract, + then check that the episode's policy update has not gone backwards; 2. read the system of record through the oracle (one read, after the episode ended; an optional baseline read before it started); 3. judge every required effect and every forbidden effect with the shared @@ -271,6 +272,7 @@ def __init__( ) (self.data_dir / "rewards").mkdir(parents=True, exist_ok=True) (self.data_dir / "baselines").mkdir(parents=True, exist_ok=True) + (self.data_dir / "policy_updates").mkdir(parents=True, exist_ok=True) # -- public surface ----------------------------------------------------- @@ -289,18 +291,50 @@ def issuer_key_id(self) -> str: def begin_episode(self, episode_id: str, identity: dict[str, str]) -> EffectState: """Register the episode's oracle identity and capture the baseline. - Call it before the rollout runs. The baseline is what lets a - ``count_new_only`` effect tell a record this episode wrote from one - that was already there; without it that effect judges INDETERMINATE. + Call it before the rollout runs. Two things happen here and both + matter at scoring time. + + The baseline is what lets a ``count_new_only`` effect tell a record + this episode wrote from one that was already there; without it that + effect judges INDETERMINATE and the episode is unscored. + + The identity is the subject the episode is about, fixed before the + rollout produced anything. :meth:`score_episode` uses this one, not + the one the descriptor carries, so the graded subject cannot be + chosen after the outcome is known. + + Calling it again for the same episode with the same identity is + allowed and re-reads the baseline. Calling it again with a different + identity is refused: the subject is not a thing an episode changes + its mind about part way through. """ self.bundle.check_identity(identity) - observed = self.oracle.read(identity) - state = effect_state_of(observed, self.bundle.oracle.channel.value) - self._write_json( - self.data_dir / "baselines" / f"{episode_id}.json", - {"identity": dict(identity), "state": state.model_dump(mode="json")}, - ) + with self._lock: + scored = self._episode_index(episode_id) + if scored is not None: + raise RewardWorkerError( + 409, + "duplicate_episode", + f"episode already scored as receipt {scored}; a scored " + "episode cannot be re-registered", + ) + registered, _ = self._baseline(episode_id) + if registered is not None and registered != dict(identity): + raise RewardWorkerError( + 409, + "identity_conflict", + f"episode {episode_id} is already registered for " + f"{_identity_text(registered)}; re-registering it for " + f"{_identity_text(dict(identity))} would move the subject " + "after the fact", + ) + observed = self.oracle.read(identity) + state = effect_state_of(observed, self.bundle.oracle.channel.value) + self._write_json( + self.data_dir / "baselines" / f"{episode_id}.json", + {"identity": dict(identity), "state": state.model_dump(mode="json")}, + ) return state def score_episode( @@ -328,23 +362,55 @@ def score_episode( f"episode already scored as receipt {existing}", ) registered_identity, before = self._baseline(episode.episode_id) - identity = declared_identity or registered_identity - if identity is None: - raise RewardWorkerError( - 422, - "identity_missing", - "the episode names no oracle identity: pass oracle_identity, " - "metadata.oracle_identity, or register it with begin_episode", - ) + identity = self._graded_identity(declared_identity, registered_identity) try: self.bundle.check_identity(identity) except IdentityError as exc: raise RewardWorkerError(422, "identity_mismatch", str(exc)) from exc + self._check_policy_update(episode) observed = self.oracle.read(identity) judged = judge_episode(self.bundle, identity, signal, before, observed) envelope = self._issue(episode, observed, before, judged) + self._advance_policy_update(episode) return envelope + def _graded_identity( + self, + declared: Optional[dict[str, str]], + registered: Optional[dict[str, str]], + ) -> dict[str, str]: + """Pick the subject this episode is graded on. Registration wins. + + The registration was made by the environment before the rollout ran, + when nobody knew how the episode would turn out. The descriptor + arrives afterwards from the trainer, which by then does know. So the + registration decides, and a descriptor that names a different subject + is refused rather than quietly overridden: a trainer that believes it + is grading one subject while the worker grades another has a bug + worth stopping for. + """ + + if registered is None: + if declared is None: + raise RewardWorkerError( + 422, + "identity_missing", + "the episode names no oracle identity: pass oracle_identity, " + "metadata.oracle_identity, or register it with begin_episode", + ) + return declared + if declared is not None and declared != registered: + raise RewardWorkerError( + 422, + "identity_conflict", + f"the environment registered this episode for " + f"{_identity_text(registered)} before the rollout ran, and " + f"the descriptor names {_identity_text(declared)}. The " + "registration decides which subject is graded; a descriptor " + "may repeat it but may not replace it.", + ) + return registered + def _check_binding(self, episode: EpisodeDescriptorV1) -> None: if episode.reward_contract_digest != self.contract.digest: raise RewardWorkerError( @@ -513,6 +579,69 @@ def _episode_index(self, episode_id: str) -> Optional[str]: payload = json.loads(path.read_text(encoding="utf-8")) return str(payload.get("receipt_id") or "") or None + # -- the policy-update ledger ------------------------------------------- + + def _ledger_path(self) -> Path: + """One file per contract, beside the episode index.""" + + stem = hashlib.sha256(self.contract.digest.encode("utf-8")).hexdigest()[:32] + return self.data_dir / "policy_updates" / f"{stem}.json" + + def _ledger(self) -> dict[str, Any]: + path = self._ledger_path() + if not path.is_file(): + return {} + payload = json.loads(path.read_text(encoding="utf-8")) + return payload if isinstance(payload, dict) else {} + + def _check_policy_update(self, episode: EpisodeDescriptorV1) -> None: + """Refuse a descriptor whose policy update goes backwards. + + A certificate expires after a stated number of policy updates, and + ``policy_update`` is the only thing that moves the episode towards + that expiry. It arrives on the wire from the trainer, so without a + record of its own the worker would let a trainer count 0, 999, + 1000000000, 0 and read the certificate as current again at the end. + The ledger keeps the highest update this contract has seen and + refuses anything below it, so the counter only moves the way that + expires a certificate. + + The mark is per contract, not per policy checkpoint. Keeping it per + checkpoint would let a trainer reset the count by starting to call + its checkpoint something else, and expiry would never arrive. Policy + updates count one training run against one contract, so a new + checkpoint at a lower update is going back in time either way. The + ledger records which checkpoint set the mark, for the error message. + """ + + ledger = self._ledger() + mark = ledger.get("highest_policy_update") + if not isinstance(mark, int) or episode.policy_update >= mark: + return + owner = str(ledger.get("policy_checkpoint_id") or "an earlier checkpoint") + raise RewardWorkerError( + 422, + "policy_update_regressed", + f"this contract has already scored policy update {mark} (from " + f"{owner}) and the episode names {episode.policy_update}. A " + "policy update counter only moves forward; moving it back would " + "make an expired certificate read as current.", + ) + + def _advance_policy_update(self, episode: EpisodeDescriptorV1) -> None: + ledger = self._ledger() + mark = ledger.get("highest_policy_update") + if isinstance(mark, int) and mark >= episode.policy_update: + return + self._write_json( + self._ledger_path(), + { + "reward_contract_digest": self.contract.digest, + "highest_policy_update": episode.policy_update, + "policy_checkpoint_id": episode.policy_checkpoint_id, + }, + ) + def _write_json(self, path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + ".tmp") @@ -523,6 +652,10 @@ def _write_json(self, path: Path, payload: dict[str, Any]) -> None: tmp.replace(path) +def _identity_text(identity: dict[str, str]) -> str: + return "{" + ", ".join(f"{k}={v}" for k, v in sorted(identity.items())) + "}" + + def _new_id(prefix: str) -> str: return f"{prefix}_{uuid4().hex}" diff --git a/tests/test_reward_trust_boundary.py b/tests/test_reward_trust_boundary.py new file mode 100644 index 00000000..d92a0ad6 --- /dev/null +++ b/tests/test_reward_trust_boundary.py @@ -0,0 +1,531 @@ +"""What the reward worker refuses to take on trust from its counterparty. + +Every test here drives a path that used to succeed. The reward reads a store +and signs what it read, and each of these was a way to make the signature say +something the read did not support: a tier the bundle only asserted, a subject +the trainer picked after the rollout, a reward for an episode that changed +nothing, a certificate kept current by counting backwards, and a bound +measured on records the contract never mentions. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +pytest.importorskip("openadapt_types.reward") + +from openadapt_types.oracle import OracleChannel # noqa: E402 +from openadapt_types.reward import RewardOutcomeV1 # noqa: E402 + +from openadapt_flow.reward.calibration import ( # noqa: E402 + FAULT_CLASSES, + CalibrationRefused, + corpus_digest_for, + corpus_from_effects, +) +from openadapt_flow.reward.models import ( # noqa: E402 + CERTIFICATE_FILE, + CONTRACT_FILE, + BundleError, + OracleRecipeV1, + RewardBundle, +) +from openadapt_flow.reward.oracles import ( # noqa: E402 + JsonDocumentOracle, + OracleMechanismError, + assert_sqlite_database, + build_oracle, +) +from openadapt_flow.reward.seed import ( # noqa: E402 + MOCKMED_HONEST_PATIENT, + MOCKMED_LIE_PATIENT, + MOCKMED_QUERY, + calibrate_bundle, + mockmed_episode, + seed_mockmed, + write_bundle, + write_mockmed_encounter, +) +from openadapt_flow.reward.worker import RewardWorker, RewardWorkerError # noqa: E402 +from openadapt_flow.runtime.effects.effect import Effect # noqa: E402 + +_TRIAGE = { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}, "type": "Triage"}, + "expected_count": 1, + "count_new_only": True, +} + + +@pytest.fixture() +def seeded(tmp_path: Path) -> dict[str, Any]: + from openadapt_flow.execute.keys import fingerprint_of, load_or_create_private_key + + data_dir = tmp_path / "reward-ref" + key = load_or_create_private_key(data_dir) + issuer = "self_signed:" + fingerprint_of(key.public_key()) + paths = seed_mockmed(data_dir, key, issuer) + return { + "data_dir": data_dir, + "key": key, + "issuer": issuer, + "database": data_dir / "mockmed" / "records.db", + "screen": data_dir / "mockmed" / "screen.json", + **paths, + } + + +def _worker(seeded: dict[str, Any], which: str = "tier2") -> RewardWorker: + return RewardWorker(seeded[which], seeded["data_dir"], token="test-token") + + +def _bundle( + seeded: dict[str, Any], + name: str, + *, + kind: str, + channel: str, + path: Path, + certify: bool = False, + required_effects: list[dict[str, Any]] | None = None, + forbidden_effects: list[dict[str, Any]] | None = None, +) -> Path: + directory = seeded["data_dir"] / "contracts" / name + oracle: dict[str, Any] = {"kind": kind, "path": str(path)} + if kind == "sqlite": + oracle["query"] = MOCKMED_QUERY + else: + oracle["records_key"] = "records" + write_bundle( + directory, + contract_id=f"reward_contract_{name.replace('-', '_')}_0000", + oracle=oracle, + channel=channel, + key=seeded["key"], + issuer_key_id=seeded["issuer"], + certify=certify, + required_effects=required_effects or [dict(_TRIAGE)], + forbidden_effects=forbidden_effects, + ) + return directory + + +# -- 1. the tier is not a string the bundle author writes ---------------------- + + +def test_the_same_document_cannot_buy_two_tiers(seeded: dict[str, Any]) -> None: + """One file, both JSON recipe kinds, one tier. + + Before this, ``screen_dump`` on a document gave tier 0 and + ``development_only``, and ``json_file`` on the identical bytes gave tier 2 + and ``certified``. + """ + + document = seeded["data_dir"] / "one-document.json" + document.write_text(json.dumps({"records": []}), encoding="utf-8") + tiers = set() + for kind in ("screen_dump", "json_file"): + recipe = OracleRecipeV1.model_validate( + {"kind": kind, "path": str(document), "records_key": "records"} + ) + adapter = build_oracle(recipe, document.parent) + assert isinstance(adapter, JsonDocumentOracle) + assert adapter.channel is OracleChannel.OCR + tiers.add(int(recipe.tier)) + assert tiers == {0} + + +def test_json_file_may_not_declare_a_record_channel(seeded: dict[str, Any]) -> None: + document = seeded["data_dir"] / "claimed-as-file.json" + document.write_text(json.dumps({"records": []}), encoding="utf-8") + directory = _bundle( + seeded, "json-as-file", kind="json_file", channel="file", path=document + ) + with pytest.raises(BundleError, match="does not match the contract channel"): + RewardBundle.load(directory) + + +def test_a_json_document_renamed_as_a_database_is_refused( + seeded: dict[str, Any], +) -> None: + """The db channel rests on opening a real database, not on the file name.""" + + fake = seeded["data_dir"] / "screen.db" + fake.write_text(json.dumps({"records": []}), encoding="utf-8") + with pytest.raises(OracleMechanismError, match="not a SQLite database"): + assert_sqlite_database(fake) + directory = _bundle(seeded, "fake-db", kind="sqlite", channel="db", path=fake) + with pytest.raises(OracleMechanismError): + RewardWorker(directory, seeded["data_dir"], token="test-token") + # The real store opens. + assert assert_sqlite_database(seeded["database"]) == seeded["database"] + + +def test_the_adapter_decides_the_channel( + seeded: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """A recipe table that drifts from its adapter stops the worker. + + The table in ``models`` and the adapter in ``oracles`` are two places that + have to agree about a tier. Make them disagree and the build refuses, + rather than shipping the table's answer. + """ + + from openadapt_flow.reward import models + + document = seeded["data_dir"] / "drift.json" + document.write_text(json.dumps({"records": []}), encoding="utf-8") + monkeypatch.setitem(models._RECIPE_CHANNEL, "screen_dump", OracleChannel.FILE) + recipe = OracleRecipeV1.model_validate( + {"kind": "screen_dump", "path": str(document), "records_key": "records"} + ) + assert recipe.channel is OracleChannel.FILE + with pytest.raises(OracleMechanismError, match="reads through"): + build_oracle(recipe, document.parent) + + +# -- 2. the graded subject is settled before the rollout ----------------------- + + +def test_a_descriptor_may_not_replace_the_registered_subject( + seeded: dict[str, Any], +) -> None: + """Register the subject the rollout ran on; score naming another one. + + Before this the descriptor won, and the receipt carried the trainer's + subject with outcome ``verified``, ``certified``, scalar 1.0. + """ + + worker = _worker(seeded) + worker.begin_episode("episode_swap_01", {"patient_id": MOCKMED_LIE_PATIENT}) + write_mockmed_encounter(seeded["database"], MOCKMED_HONEST_PATIENT) + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode( + mockmed_episode( + MOCKMED_HONEST_PATIENT, + episode_id="episode_swap_01", + contract_digest=worker.contract.digest, + ) + ) + assert excinfo.value.status_code == 422 + assert excinfo.value.error == "identity_conflict" + assert MOCKMED_LIE_PATIENT in excinfo.value.detail + assert MOCKMED_HONEST_PATIENT in excinfo.value.detail + + +def test_a_descriptor_may_repeat_the_registered_subject( + seeded: dict[str, Any], +) -> None: + worker = _worker(seeded) + worker.begin_episode("episode_agree_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + write_mockmed_encounter(seeded["database"], MOCKMED_HONEST_PATIENT) + envelope = worker.score_episode( + mockmed_episode( + MOCKMED_HONEST_PATIENT, + episode_id="episode_agree_01", + contract_digest=worker.contract.digest, + ) + ) + assert envelope["receipt"]["reward_outcome"] == "verified" + + +def test_an_episode_may_not_be_re_registered_under_another_subject( + seeded: dict[str, Any], +) -> None: + worker = _worker(seeded) + worker.begin_episode("episode_rereg_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + # The same subject again is fine; it only re-reads the baseline. + worker.begin_episode("episode_rereg_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + with pytest.raises(RewardWorkerError) as excinfo: + worker.begin_episode("episode_rereg_01", {"patient_id": MOCKMED_LIE_PATIENT}) + assert excinfo.value.status_code == 409 + assert excinfo.value.error == "identity_conflict" + + +def test_a_scored_episode_may_not_be_re_registered(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + worker.begin_episode("episode_after_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + write_mockmed_encounter(seeded["database"], MOCKMED_HONEST_PATIENT) + worker.score_episode( + mockmed_episode( + MOCKMED_HONEST_PATIENT, + episode_id="episode_after_01", + contract_digest=worker.contract.digest, + ) + ) + with pytest.raises(RewardWorkerError) as excinfo: + worker.begin_episode("episode_after_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + assert excinfo.value.error == "duplicate_episode" + + +# -- 3. a rollout that changed nothing earns nothing --------------------------- + + +def test_required_effects_must_claim_a_change(seeded: dict[str, Any]) -> None: + """A state-only required effect will not load. + + Before this the seeded contract's required effect was a plain + ``record_written``, a statement about the store's current contents, so + naming a subject whose row already existed scored 1.0 with no episode + having run. + """ + + directory = _bundle( + seeded, + "state-only", + kind="sqlite", + channel="db", + path=seeded["database"], + required_effects=[ + { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}, "type": "Triage"}, + "expected_count": 1, + } + ], + ) + with pytest.raises(BundleError, match="claim about change"): + RewardBundle.load(directory) + + +def test_an_exact_new_set_also_claims_a_change(seeded: dict[str, Any]) -> None: + """The other baseline-dependent kind satisfies the same rule.""" + + directory = _bundle( + seeded, + "exact-new-set", + kind="sqlite", + channel="db", + path=seeded["database"], + required_effects=[ + { + "kind": "exact_new_set", + "match": {"patient_id": {"param": "patient_id"}}, + "new_records": [{"type": "Triage"}], + "expected_count": 1, + } + ], + ) + assert RewardBundle.load(directory).required_effects[0].requires_baseline + + +def test_an_episode_that_wrote_nothing_scores_zero(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + # The subject already has a row, and this episode adds none. + write_mockmed_encounter(seeded["database"], MOCKMED_HONEST_PATIENT) + worker.begin_episode("episode_idle_01", {"patient_id": MOCKMED_HONEST_PATIENT}) + envelope = worker.score_episode( + mockmed_episode( + MOCKMED_HONEST_PATIENT, + episode_id="episode_idle_01", + contract_digest=worker.contract.digest, + ) + ) + receipt = envelope["receipt"] + assert receipt["reward_outcome"] == RewardOutcomeV1.WRONG_EFFECT.value + assert receipt["scalar_reward"] == 0.0 + + +# -- 4. the policy update counter only moves forward --------------------------- + + +def _score_at( + worker: RewardWorker, seeded: dict[str, Any], index: int, update: int +) -> dict[str, Any]: + episode_id = f"episode_update_{index:02d}" + patient = f"patient-run-{index:04d}" + worker.begin_episode(episode_id, {"patient_id": patient}) + write_mockmed_encounter(seeded["database"], patient) + return worker.score_episode( + mockmed_episode( + patient, + episode_id=episode_id, + contract_digest=worker.contract.digest, + policy_update=update, + ) + )["receipt"] + + +def test_the_policy_update_may_not_go_backwards(seeded: dict[str, Any]) -> None: + """0, then 999, then 10^9, then 0 again. + + Before this the last one came back ``current`` and certified: expiry + counted a number the counterparty reported and nothing compared it with + anything seen before. + """ + + worker = _worker(seeded) + assert _score_at(worker, seeded, 0, 0)["certificate_state"] == "current" + assert _score_at(worker, seeded, 1, 999)["certificate_state"] == "current" + expired = _score_at(worker, seeded, 2, 10**9) + assert expired["certificate_state"] == "expired" + assert expired["certified"] is False + with pytest.raises(RewardWorkerError) as excinfo: + _score_at(worker, seeded, 3, 0) + assert excinfo.value.status_code == 422 + assert excinfo.value.error == "policy_update_regressed" + + +def test_the_ledger_survives_a_restart(seeded: dict[str, Any]) -> None: + """It is persisted beside the episode index, not held in memory.""" + + _score_at(_worker(seeded), seeded, 4, 500) + with pytest.raises(RewardWorkerError, match="only moves forward"): + _score_at(_worker(seeded), seeded, 5, 499) + assert _score_at(_worker(seeded), seeded, 6, 500) is not None + + +def test_renaming_the_checkpoint_does_not_reset_expiry( + seeded: dict[str, Any], +) -> None: + """The mark is per contract, so a fresh checkpoint id cannot rewind it.""" + + worker = _worker(seeded) + _score_at(worker, seeded, 7, 900) + worker.begin_episode("episode_new_ckpt", {"patient_id": "patient-run-9999"}) + write_mockmed_encounter(seeded["database"], "patient-run-9999") + payload = mockmed_episode( + "patient-run-9999", + episode_id="episode_new_ckpt", + contract_digest=worker.contract.digest, + policy_update=0, + ) + payload["policy_checkpoint_id"] = "policy_checkpoint_renamed_0" + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.error == "policy_update_regressed" + + +# -- 5. the corpus comes from the contract ------------------------------------- + + +def test_the_corpus_plants_the_records_the_contract_names() -> None: + """Before this the corpus always emitted ``type: Triage``.""" + + corpus = corpus_from_effects( + [ + Effect.model_validate( + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "type": "Radiology", + }, + "expected_count": 1, + "count_new_only": True, + } + ) + ], + [], + ["patient_id"], + ) + planted = corpus.records({"patient_id": "p-1"}, first_id=1) + assert planted == [{"id": 1, "patient_id": "p-1", "type": "Radiology"}] + + +def test_a_read_back_effect_does_not_double_the_record() -> None: + """A field_equals on the same selector describes one row, not two.""" + + corpus = corpus_from_effects( + [ + Effect.model_validate( + { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}}, + "expected_count": 1, + "count_new_only": True, + } + ), + Effect.model_validate( + { + "kind": "field_equals", + "match": {"patient_id": {"param": "patient_id"}}, + "field": "status", + "value": "saved", + } + ), + ], + [], + ["patient_id"], + ) + assert corpus.records({"patient_id": "p-1"}, first_id=1) == [ + {"id": 1, "patient_id": "p-1", "status": "saved"} + ] + + +def test_the_certificate_names_the_contract_s_own_corpus( + seeded: dict[str, Any], +) -> None: + bundle = RewardBundle.load(seeded["tier2"]) + assert bundle.certificate is not None + want = corpus_digest_for( + bundle.required_effects, bundle.forbidden_effects, bundle.identity_keys + ) + assert bundle.certificate.calibration_corpus_digest == want + assert bundle.contract.certificate_policy.calibration_corpus_digest == want + + +def test_a_certificate_for_another_corpus_is_refused(seeded: dict[str, Any]) -> None: + """A bound measured on other records does not apply to these effects.""" + + directory = seeded["tier2"] + certificate = json.loads((directory / CERTIFICATE_FILE).read_text()) + certificate["calibration_corpus_digest"] = "sha256:" + "1" * 64 + (directory / CERTIFICATE_FILE).write_text(json.dumps(certificate)) + with pytest.raises(BundleError, match="calibration_corpus_digest"): + RewardBundle.load(directory) + + +def test_a_corpus_that_cannot_verify_refuses_the_bound( + seeded: dict[str, Any], +) -> None: + """Contradictory required effects mean the trials measure nothing. + + Every trial refutes for a reason the planted fault did not cause, the + false-accept count is zero, and the bound would be the best number the + method can produce while bounding nothing. + """ + + directory = _bundle( + seeded, + "unexercisable", + kind="sqlite", + channel="db", + path=seeded["database"], + required_effects=[ + dict(_TRIAGE), + dict(_TRIAGE, expected_count=2), + ], + ) + with pytest.raises(CalibrationRefused, match="does not exercise the contract"): + calibrate_bundle(directory) + + +def test_the_fault_classes_include_the_wrong_subject(seeded: dict[str, Any]) -> None: + """``WRONG_EFFECT`` covers a write that landed on somebody else. + + The bound did not sample that mode before, so it did not cover it. The + class was added rather than the docstring narrowed: the mode is real, the + judge already catches it, and a bound that never planted it was quiet + about the failure the reward exists to price. + """ + + assert "wrong_subject" in FAULT_CLASSES + calibration = json.loads((seeded["tier2"] / "calibration.json").read_text()) + assert "wrong_subject" in calibration["calibration_fault_classes"] + assert calibration["calibration_false_accepts"] == 0 + + +def test_the_calibration_records_which_corpus_it_ran_on( + seeded: dict[str, Any], +) -> None: + calibration = json.loads((seeded["tier2"] / "calibration.json").read_text()) + contract = json.loads((seeded["tier2"] / CONTRACT_FILE).read_text()) + assert ( + calibration["calibration_corpus_digest"] + == contract["certificate_policy"]["calibration_corpus_digest"] + ) diff --git a/tests/test_reward_worker.py b/tests/test_reward_worker.py index 7d9db008..8c9bc43c 100644 --- a/tests/test_reward_worker.py +++ b/tests/test_reward_worker.py @@ -22,7 +22,9 @@ import openadapt_flow.reward.callables as callables # noqa: E402 from openadapt_flow.reward.calibration import ( # noqa: E402 + CorpusRecipe, clopper_pearson_upper, + corpus_from_effects, extradup_trials, ) from openadapt_flow.reward.callables import ( # noqa: E402 @@ -36,19 +38,25 @@ RewardBundle, assert_no_forbidden_keys, ) -from openadapt_flow.reward.oracles import JsonDocumentOracle # noqa: E402 +from openadapt_flow.reward.oracles import ( # noqa: E402 + observation, +) from openadapt_flow.reward.seed import ( # noqa: E402 CALIBRATION_FILE, CALIBRATION_TRIALS, MOCKMED_DUPLICATE_PATIENT, MOCKMED_HONEST_PATIENT, MOCKMED_LIE_PATIENT, + MOCKMED_QUERY, mockmed_episode, seed_mockmed, write_bundle, + write_mockmed_banner, + write_mockmed_encounter, ) from openadapt_flow.reward.serve import OPENAI_GRADER_ROUTE, create_app # noqa: E402 from openadapt_flow.reward.worker import RewardWorker, RewardWorkerError # noqa: E402 +from openadapt_flow.runtime.effects.effect import Effect # noqa: E402 @pytest.fixture() @@ -60,7 +68,12 @@ def seeded(tmp_path: Path) -> dict[str, Any]: paths = seed_mockmed( data_dir, key, "self_signed:" + fingerprint_of(key.public_key()) ) - return {"data_dir": data_dir, **paths} + return { + "data_dir": data_dir, + "database": data_dir / "mockmed" / "records.db", + "screen": data_dir / "mockmed" / "screen.json", + **paths, + } def _worker( @@ -80,12 +93,49 @@ def _episode( ) +def _run( + worker: RewardWorker, + seeded: dict[str, Any], + patient_id: str, + episode_id: str, + *, + encounters: int = 1, + banners: int = 0, + discharge: bool = False, + **kwargs: Any, +) -> dict[str, Any]: + """Register the episode, simulate what it wrote, then score it. + + Registration happens first because the required effect is a claim about + change: the judge needs the pre-episode baseline, and the subject is + fixed before the rollout produces anything. + """ + + worker.begin_episode(episode_id, {"patient_id": patient_id}) + for _ in range(encounters): + write_mockmed_encounter(seeded["database"], patient_id) + if discharge: + write_mockmed_encounter(seeded["database"], patient_id, type_="Discharge") + for _ in range(banners): + write_mockmed_banner(seeded["screen"], patient_id) + return worker.score_episode(_episode(worker, patient_id, episode_id, **kwargs)) + + def _receipt(envelope: dict[str, Any]) -> RewardEvidenceReceiptV1: return RewardEvidenceReceiptV1.model_validate(envelope["receipt"]) -def _unreachable(tmp_path: Path) -> JsonDocumentOracle: - return JsonDocumentOracle(tmp_path / "absent.json", channel=OracleChannel.FILE) +class _UnreachableOracle: + """A tier-2 channel that cannot be read. Never a guessed empty list.""" + + channel = OracleChannel.DB + + def read(self, identity: Any) -> Any: + return observation(self.channel, identity, None) + + +def _unreachable(tmp_path: Path) -> _UnreachableOracle: + return _UnreachableOracle() # -- outcome mapping ---------------------------------------------------------- @@ -93,9 +143,7 @@ def _unreachable(tmp_path: Path) -> JsonDocumentOracle: def test_verified_tier2_is_certified(seeded: dict[str, Any]) -> None: worker = _worker(seeded) - envelope = worker.score_episode( - _episode(worker, MOCKMED_HONEST_PATIENT, "episode_honest_01") - ) + envelope = _run(worker, seeded, MOCKMED_HONEST_PATIENT, "episode_honest_01") receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED assert receipt.oracle_tier == 2 @@ -120,13 +168,12 @@ def test_verified_tier2_expired_certificate_is_not_certified( worker = _worker(seeded) assert worker.certificate is not None expired_update = worker.certificate.expires_at_policy_update - envelope = worker.score_episode( - _episode( - worker, - MOCKMED_HONEST_PATIENT, - "episode_honest_expired", - policy_update=expired_update, - ) + envelope = _run( + worker, + seeded, + MOCKMED_HONEST_PATIENT, + "episode_honest_expired", + policy_update=expired_update, ) receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED @@ -138,12 +185,18 @@ def test_verified_tier2_expired_certificate_is_not_certified( def test_tier0_is_development_only_never_certified(seeded: dict[str, Any]) -> None: worker = _worker(seeded, "tier0") assert worker.certificate is None - envelope = worker.score_episode( - _episode(worker, MOCKMED_LIE_PATIENT, "episode_tier0_lie") + # The banner appeared and nothing was written to the database. The screen + # dump agrees with the banner, so the OCR channel may say VERIFIED, and it + # still cannot be certified. + envelope = _run( + worker, + seeded, + MOCKMED_LIE_PATIENT, + "episode_tier0_lie", + encounters=0, + banners=1, ) receipt = _receipt(envelope) - # The screen dump says the lie episode saved. The channel is OCR, so the - # verdict may be VERIFIED, and it still cannot be certified. assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED assert receipt.oracle_tier == 0 assert receipt.development_only is True @@ -153,9 +206,16 @@ def test_tier0_is_development_only_never_certified(seeded: dict[str, Any]) -> No def test_banner_lie_yields_zero(seeded: dict[str, Any]) -> None: + """The same episode the OCR channel calls verified. The store holds nothing.""" + worker = _worker(seeded) - envelope = worker.score_episode( - _episode(worker, MOCKMED_LIE_PATIENT, "episode_banner_lie_01") + envelope = _run( + worker, + seeded, + MOCKMED_LIE_PATIENT, + "episode_banner_lie_01", + encounters=0, + banners=1, ) receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.WRONG_EFFECT @@ -166,8 +226,12 @@ def test_banner_lie_yields_zero(seeded: dict[str, Any]) -> None: def test_duplicate_create_is_wrong_effect(seeded: dict[str, Any]) -> None: worker = _worker(seeded) - envelope = worker.score_episode( - _episode(worker, MOCKMED_DUPLICATE_PATIENT, "episode_duplicate_01") + envelope = _run( + worker, + seeded, + MOCKMED_DUPLICATE_PATIENT, + "episode_duplicate_01", + encounters=2, ) receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.WRONG_EFFECT @@ -182,10 +246,11 @@ def test_duplicate_create_is_wrong_effect(seeded: dict[str, Any]) -> None: def test_same_episode_twice_is_rejected(seeded: dict[str, Any]) -> None: worker = _worker(seeded) - payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_once_only_1") - worker.score_episode(payload) + _run(worker, seeded, MOCKMED_HONEST_PATIENT, "episode_once_only_1") with pytest.raises(RewardWorkerError) as excinfo: - worker.score_episode(payload) + worker.score_episode( + _episode(worker, MOCKMED_HONEST_PATIENT, "episode_once_only_1") + ) assert excinfo.value.status_code == 409 assert excinfo.value.error == "duplicate_episode" @@ -206,33 +271,7 @@ def test_indeterminate_is_unscored_not_zero( def test_count_new_only_needs_a_baseline(seeded: dict[str, Any]) -> None: - from openadapt_flow.execute.keys import fingerprint_of, load_or_create_private_key - from openadapt_flow.reward.seed import write_bundle - - key = load_or_create_private_key(seeded["data_dir"]) - directory = seeded["data_dir"] / "contracts" / "mockmed-new-only" - write_bundle( - directory, - contract_id="reward_contract_mockmed_new_only", - oracle={ - "kind": "json_file", - "path": str(seeded["data_dir"] / "mockmed" / "records.json"), - "records_key": "records", - }, - channel="file", - key=key, - issuer_key_id="self_signed:" + fingerprint_of(key.public_key()), - certify=False, - required_effects=[ - { - "kind": "record_written", - "match": {"patient_id": {"param": "patient_id"}, "type": "Triage"}, - "expected_count": 1, - "count_new_only": True, - } - ], - ) - worker = RewardWorker(directory, seeded["data_dir"], token="test-token") + worker = _worker(seeded) # No baseline: the delta is unknowable, so the judge is INDETERMINATE and # the episode is unscored, never 0. envelope = worker.score_episode( @@ -245,12 +284,7 @@ def test_count_new_only_needs_a_baseline(seeded: dict[str, Any]) -> None: # With a baseline registered before the episode, the same store judges, # and the descriptor no longer needs to carry the identity. worker.begin_episode("episode_new_only_02", {"patient_id": MOCKMED_LIE_PATIENT}) - store = seeded["data_dir"] / "mockmed" / "records.json" - records = json.loads(store.read_text()) - records["records"].append( - {"id": 7, "patient_id": MOCKMED_LIE_PATIENT, "type": "Triage"} - ) - store.write_text(json.dumps(records)) + write_mockmed_encounter(seeded["database"], MOCKMED_LIE_PATIENT) envelope = worker.score_episode( { "episode_id": "episode_new_only_02", @@ -266,13 +300,13 @@ def test_halt_signal_with_no_effect_is_halted_before_effect( seeded: dict[str, Any], ) -> None: worker = _worker(seeded) - envelope = worker.score_episode( - _episode( - worker, - MOCKMED_LIE_PATIENT, - "episode_halted_01", - runtime_signal="halted_before_effect", - ) + envelope = _run( + worker, + seeded, + MOCKMED_LIE_PATIENT, + "episode_halted_01", + encounters=0, + runtime_signal="halted_before_effect", ) receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.HALTED_BEFORE_EFFECT @@ -283,13 +317,12 @@ def test_halt_signal_with_effect_present_is_reconciliation( seeded: dict[str, Any], ) -> None: worker = _worker(seeded) - envelope = worker.score_episode( - _episode( - worker, - MOCKMED_HONEST_PATIENT, - "episode_halted_lie_01", - runtime_signal="halted_before_effect", - ) + envelope = _run( + worker, + seeded, + MOCKMED_HONEST_PATIENT, + "episode_halted_lie_01", + runtime_signal="halted_before_effect", ) receipt = _receipt(envelope) assert receipt.reward_outcome is RewardOutcomeV1.RECONCILIATION_REQUIRED @@ -298,15 +331,13 @@ def test_halt_signal_with_effect_present_is_reconciliation( def test_forbidden_effect_is_wrong_effect(seeded: dict[str, Any]) -> None: - store = seeded["data_dir"] / "mockmed" / "records.json" - records = json.loads(store.read_text()) - records["records"].append( - {"id": 9, "patient_id": MOCKMED_HONEST_PATIENT, "type": "Discharge"} - ) - store.write_text(json.dumps(records)) worker = _worker(seeded) - envelope = worker.score_episode( - _episode(worker, MOCKMED_HONEST_PATIENT, "episode_forbidden_01") + envelope = _run( + worker, + seeded, + MOCKMED_HONEST_PATIENT, + "episode_forbidden_01", + discharge=True, ) assert _receipt(envelope).reward_outcome is RewardOutcomeV1.WRONG_EFFECT @@ -360,9 +391,7 @@ def test_receipt_carries_no_screenshot_or_rollout_bytes( seeded: dict[str, Any], ) -> None: worker = _worker(seeded) - envelope = worker.score_episode( - _episode(worker, MOCKMED_HONEST_PATIENT, "episode_no_bytes_01") - ) + envelope = _run(worker, seeded, MOCKMED_HONEST_PATIENT, "episode_no_bytes_01") receipt = envelope["receipt"] assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(receipt) assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(envelope) @@ -402,11 +431,11 @@ def _identity_bundle( directory, contract_id=f"reward_contract_{name.replace('-', '_')}_000000", oracle={ - "kind": "json_file", - "path": str(data_dir / "mockmed" / "records.json"), - "records_key": "records", + "kind": "sqlite", + "path": str(data_dir / "mockmed" / "records.db"), + "query": MOCKMED_QUERY, }, - channel="file", + channel="db", key=key, issuer_key_id="self_signed:" + fingerprint_of(key.public_key()), certify=False, @@ -479,6 +508,7 @@ def test_idempotency_key_binds_the_identity(seeded: dict[str, Any]) -> None: "match": {"type": "Triage"}, "idempotency_key": {"param": "patient_id"}, "expected_count": 1, + "count_new_only": True, } ], ) @@ -506,16 +536,42 @@ def test_certificate_bound_is_recomputable(seeded: dict[str, Any]) -> None: ) +def _mockmed_corpus() -> CorpusRecipe: + """The corpus the shipped MockMed contract's own effects describe.""" + + return corpus_from_effects( + [ + Effect.model_validate( + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "type": "Triage", + }, + "expected_count": 1, + "count_new_only": True, + } + ) + ], + [], + ["patient_id"], + ) + + def test_clopper_pearson_upper_matches_known_values() -> None: # 0 of 15 is the bound the openadapt-evals proof run reports. assert clopper_pearson_upper(0, 15) == pytest.approx(0.181036, abs=1e-6) assert clopper_pearson_upper(0, 20) == pytest.approx(0.1391, abs=1e-3) assert clopper_pearson_upper(1, 20) == pytest.approx(0.2161, abs=1e-3) assert clopper_pearson_upper(20, 20) == 1.0 + # A checker that accepts everything false-accepts every trial, and the + # bound it earns is 1.0, which the certificate model refuses. result = extradup_trials( - lambda records, identity: RewardOutcomeV1.VERIFIED, + lambda before, current, identity: RewardOutcomeV1.VERIFIED, + _mockmed_corpus(), trials=10, generator_seed=1, + corpus_digest="sha256:" + "0" * 64, ) assert result.false_accepts == 10 assert result.epsilon == 1.0 @@ -533,6 +589,25 @@ def _client( return client, worker +def _http_begin( + client: TestClient, + seeded: dict[str, Any], + patient_id: str, + episode_id: str, + *, + encounters: int = 1, +) -> None: + """Register over HTTP, the way the environment does, then write.""" + + response = client.post( + "/v1/episodes", + json={"episode_id": episode_id, "oracle_identity": {"patient_id": patient_id}}, + ) + assert response.status_code == 200, response.text + for _ in range(encounters): + write_mockmed_encounter(seeded["database"], patient_id) + + def test_http_reward_roundtrip(seeded: dict[str, Any]) -> None: client, worker = _client(seeded) bare = TestClient(client.app) @@ -541,6 +616,8 @@ def test_http_reward_roundtrip(seeded: dict[str, Any]) -> None: assert health["execute_seal"] is False assert health["oracle_tier"] == 2 assert bare.post("/v1/rewards", json={}).status_code == 401 + assert bare.post("/v1/episodes", json={}).status_code == 401 + _http_begin(client, seeded, MOCKMED_HONEST_PATIENT, "episode_http_01") created = client.post( "/v1/rewards", json=_episode(worker, MOCKMED_HONEST_PATIENT, "episode_http_01") ) @@ -567,6 +644,7 @@ def test_http_matches_the_evals_client_wire_shape(seeded: dict[str, Any]) -> Non """ client, worker = _client(seeded) + _http_begin(client, seeded, MOCKMED_HONEST_PATIENT, "episode_evals_client_1") payload = { "episode_id": "episode_evals_client_1", "policy_checkpoint_id": "policy_checkpoint_evals", @@ -597,6 +675,7 @@ def test_http_matches_the_evals_client_wire_shape(seeded: dict[str, Any]) -> Non def test_openai_grader_route_contract(seeded: dict[str, Any]) -> None: client, worker = _client(seeded) + _http_begin(client, seeded, MOCKMED_HONEST_PATIENT, "episode_grader_ok_1") item = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_grader_ok_1") scored = client.post( OPENAI_GRADER_ROUTE, @@ -607,6 +686,9 @@ def test_openai_grader_route_contract(seeded: dict[str, Any]) -> None: assert body["score"] == 1.0 assert 0.0 <= body["score"] <= 1.0 assert body["certified"] is True + _http_begin( + client, seeded, MOCKMED_LIE_PATIENT, "episode_grader_lie_1", encounters=0 + ) lie = client.post( OPENAI_GRADER_ROUTE, json={ @@ -659,6 +741,8 @@ def test_episode_from_columns_matches_the_evals_descriptor_shape( seeded: dict[str, Any], ) -> None: worker = _worker(seeded) + worker.begin_episode("episode_client_0001", {"patient_id": MOCKMED_HONEST_PATIENT}) + write_mockmed_encounter(seeded["database"], MOCKMED_HONEST_PATIENT) payload = episode_from_columns( episode_id="episode_client_0001", policy_checkpoint_id="policy_checkpoint_client", @@ -717,6 +801,7 @@ def payload(episode_id: str) -> dict[str, Any]: oracle_identity={"patient_id": MOCKMED_HONEST_PATIENT}, ) + _http_begin(app_client, seeded, MOCKMED_HONEST_PATIENT, "episode_client_http_1") envelope = client.score_episode(payload("episode_client_http_1")) assert envelope["unscored"] is False assert scalar_of(envelope) == 1.0