diff --git a/README.md b/README.md index 0c19a458..77d6fb04 100644 --- a/README.md +++ b/README.md @@ -453,6 +453,25 @@ checkpoint and resume; Agent Skill and MCP emission. Those are in [docs/CAPABILITIES.md](docs/CAPABILITIES.md), and the whole documentation set is at [docs.openadapt.ai](https://docs.openadapt.ai). +## Reward worker for a training loop + +`openadapt-flow serve-reward` scores a model's training episode by reading the +system of record after the episode ends, through the same effect oracles the +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. + +```bash +pip install 'openadapt-flow[reward]' +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. Adapters for TRL's `GRPOTrainer` and verl's reward manager are in +`openadapt_flow.reward.callables`. See +[docs/REWARD_WORKER.md](docs/REWARD_WORKER.md). + ## Development ```bash diff --git a/docs/REWARD_WORKER.md b/docs/REWARD_WORKER.md new file mode 100644 index 00000000..ac7f17a4 --- /dev/null +++ b/docs/REWARD_WORKER.md @@ -0,0 +1,209 @@ +# The reward worker + +`openadapt-flow serve-reward` scores one training episode by reading the +system of record after the episode ends. It answers one question: did the +terminal effect the reward contract requires actually land, and did nothing +the contract forbids land with it? The answer comes back as a signed +`RewardEvidenceReceiptV1` from `openadapt-types`. + +A reward receipt is not an Execute Seal. Execute takes a qualified program +with zero model use. A model rollout is not one, so it never receives an +Execute receipt, and the reward receipt never claims that Flow governed the +policy's actions. The two receipts carry different schema ids, and the +reward receipt has no `execution_id`, `workflow_digest`, `qualification_id`, +or `contracts` block, so one cannot be passed off as the other. + +## What runs where + +Three processes, on three machines, and only one of them sees the data. + +The **organization worker** is this command. It runs inside the customer +network, next to the system of record, and holds the only credential that +can read it. It reads through one oracle recipe (a REST document, a +read-only SQL query, a FHIR search, a directory listing, or a JSON dump for +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 +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. + +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 +shape `openadapt_evals.reward.receipts.EpisodeDescriptor` sends: +`episode_id`, `policy_checkpoint_id`, `policy_update`, +`reward_contract_digest`, and optional `task_id`, `environment_id`, and +`metadata`. The digest must be the contract this worker serves, or the +episode is refused. The trainer never gets a credential for the system of +record. `openadapt_flow.reward.callables` carries the adapters for TRL's +`GRPOTrainer` and verl's reward manager, and `HttpRewardClient` 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. + +## The outcome table + +The worker judges every required effect and every forbidden effect with +the same three-valued judge the runtime uses +(`openadapt_flow/runtime/effects/_common.py`). The runtime's own signal +about how the episode ended is an input, never the verdict. The rules fire in +this order. + +| Condition | `reward_outcome` | Scalar | +|---|---|---| +| Runtime signal `failed_platform` | `failed_platform` | unscored | +| Store unreachable at read time | `failed_platform`, uncertainty `oracle_unavailable` | unscored | +| Any verdict INDETERMINATE (stale read, no baseline for `count_new_only`) | `reconciliation_required`, uncertainty `effect_uncertain` | unscored | +| Any forbidden effect present | `wrong_effect` | 0 or the declared penalty | +| Signal `completed`, every required effect CONFIRMED | `verified` | the declared positive reward | +| Signal `completed`, a required effect REFUTED (absent, duplicated, wrong value) | `wrong_effect` | 0 or the declared penalty | +| Signal `halted_before_effect`, `refused`, or `rejected_policy`, store shows no required effect | that outcome | 0 or the declared penalty | +| Same signals, but a required effect is present anyway | `reconciliation_required`, uncertainty `effect_uncertain` | unscored | + +Unscored is never 0.0. The receipt carries `scalar_reward: null` and the +envelope says `unscored: true`. Zero would teach the policy that an +unreadable store is the same as a wrong write. It is not. + +`certified` is true only at oracle tier 2 or 3 with a certificate that is +current at the episode's policy update. Tier 0 (visual, OCR) and tier 1 +(second UI session) receipts are `development_only` and can never be +certified, whatever the screen shows. A verified tier-2 receipt whose +certificate expired is still `verified`, still scored, and not certified. + +## The MockMed run + +```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. + +`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. + +Three episodes to post, with the bearer token and contract digest the banner +prints: + +```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", + "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", + "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", + "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. +``` + +Then the tier-0 worker, in a second terminal, with that bundle's own digest: + +```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", + "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 MockMed banner lie fixture yields 0 because the seeded contract declares +`wrong_effect_reward: 0.0`. The contract default is -1.0. A penalty is a +training choice the contract states; the worker never picks one. + +## Routes + +| Route | Body in | Body out | +|---|---|---| +| `GET /health` | none | issuer, key fingerprint, contract digest, oracle tier | +| `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 | + +Every route but `/health` needs `Authorization: Bearer `. The +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. + +The OpenAI route mirrors the only custom-grader contract OpenAI documents, +the `python` grader's `grade(sample, item) -> float` (graders guide and +reinforcement fine-tuning guide at developers.openai.com, read 2026-09-01). +OpenAI documents no grader that calls a user-hosted URL, and its python +grader runs without network access, so a hosted RFT job cannot reach this +worker. The route exists for a self-hosted loop that already speaks that +shape. Its schema has no "do not score" value, and OpenAI's own rule is that +an exception or a bad float "will be marked as invalid and return a 0 +grade". This worker refuses that: an unscored episode answers 422 with +`error: unscored`, and a wrapper must drop the sample before any grader +sees it. + +## Trainer adapters + +`trl_reward_function(scorer, policy_checkpoint_id=..., reward_contract_digest=...)` +returns a function +with the signature TRL's `GRPOTrainer` expects for `reward_funcs`: +`(prompts, completions, completion_ids, trainer_state, **kwargs) -> +list[float | None]`. The dataset carries `episode_id` and `oracle_identity` +columns; the policy update is `trainer_state.global_step`. An unscored +episode returns `None`, which TRL documents as "this reward function does +not apply to this sample" and excludes. + +`verl_compute_score(scorer, policy_checkpoint_id=..., reward_contract_digest=...)` +returns a +`compute_score(data_source, solution_str, ground_truth, extra_info)` for +`custom_reward_function.path`. verl has no None sentinel, so an unscored +episode returns `{"score": nan, "openadapt_unscored": true, ...}`. NaN on +purpose: a group that keeps it produces a NaN loss instead of a quiet 0. +`drop_unscored(rewards, *aligned)` and `scored_groups(groups)` remove those +samples before the advantage is computed. + +## Scope + +The worker reads the store once after the episode and signs what it read. +It never sees the screen or the trajectory, so it cannot grade how the +policy got there. If the store cannot be read, it says so and scores +nothing. diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index eb8a49be..b5f231e5 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -61,6 +61,9 @@ UI over bundles / runs / skill libraries; requires the ``console`` extra). - ``serve-execute`` — host the public Execute HTTP+MCP contract on this machine (self-signed local receipts; not an OpenAdapt production Seal). +- ``serve-reward`` — score training episodes by reading the system of + record through an independent oracle (self-signed reward receipts; not + an Execute Seal). - ``emit-skill`` — emit an Agent Skills folder for a bundle. - ``emit-mcp`` — emit a standalone MCP ``server.py`` for a bundle. - ``connector`` — the BYOC (bring-your-own-cloud) outbound-pull daemon: @@ -7668,6 +7671,55 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: ) p.set_defaults(func=_cmd_serve_execute) + p = sub.add_parser( + "serve-reward", + help=( + "Score training episodes by reading the system of record through " + "an independent oracle. Receipts are self-signed reward receipts, " + "not Execute Seals. Needs `pip install 'openadapt-flow[reward]'`" + ), + ) + p.add_argument( + "--contract", + default=None, + help=( + "Reward contract bundle directory (contract.json, " + "required_effects.json, forbidden_effects.json, oracle.json, " + "optional certificate.json). Defaults to the seeded MockMed " + "tier-2 bundle when --seed-mockmed is given." + ), + ) + p.add_argument( + "--port", + type=int, + default=8788, + help="Port (default: 8788)", + ) + p.add_argument( + "--host", + default="127.0.0.1", + help="Bind address (default: 127.0.0.1)", + ) + p.add_argument( + "--data-dir", + default=None, + help="Local data directory (default: ~/.openadapt/reward-ref)", + ) + p.add_argument( + "--token", + default=None, + help="Bearer token (default: generated on first start in --data-dir)", + ) + p.add_argument( + "--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)" + ), + ) + p.set_defaults(func=_cmd_serve_reward) + p = sub.add_parser( "business-decisions", help="Customer-runner typed-decision relay; it never resumes or acts.", @@ -7967,6 +8019,61 @@ def _cmd_serve_execute(args: argparse.Namespace) -> int: return 0 +def _cmd_serve_reward(args: argparse.Namespace) -> int: + from importlib.util import find_spec + + missing = [ + name + for name in ("fastapi", "uvicorn", "openadapt_types") + if find_spec(name) is None + ] + if missing: + raise SystemExit( + f"serve-reward needs {', '.join(missing)} — install the " + "reward extra: pip install 'openadapt-flow[reward]'" + ) + from openadapt_flow.reward import REWARD_NOTICE + from openadapt_flow.reward.serve import serve + from openadapt_flow.reward.worker import RewardWorker, default_data_dir + + data_dir = Path(args.data_dir) if args.data_dir else default_data_dir() + contract = args.contract + seeded_paths: dict[str, Path] = {} + if args.seed_mockmed: + from openadapt_flow.execute.keys import ( + fingerprint_of, + load_or_create_private_key, + ) + from openadapt_flow.reward.seed import seed_mockmed + + key = load_or_create_private_key(data_dir) + issuer_key_id = "self_signed:" + fingerprint_of(key.public_key()) + seeded_paths = seed_mockmed(data_dir, key, issuer_key_id) + if contract is None: + contract = str(seeded_paths["tier2"]) + if contract is None: + raise SystemExit("serve-reward needs --contract or --seed-mockmed") + worker = RewardWorker(contract, data_dir, token=args.token) + print("openadapt-flow reference reward worker") + print(f" http://{args.host}:{args.port}") + print(f" data dir {data_dir}") + print(f" contract {worker.bundle.directory}") + print(f" digest {worker.contract.digest}") + print( + f" oracle {worker.bundle.oracle.channel.value} " + f"(tier {int(worker.bundle.oracle.tier)})" + ) + print(f" certificate {'present' if worker.certificate else 'absent'}") + print(f" token {worker.token}") + print(" issuer self_signed") + print(f" fingerprint {worker.fingerprint}") + print(f" {REWARD_NOTICE}") + for label, path in seeded_paths.items(): + print(f" seeded {label:<6} {path}") + serve(worker, host=args.host, port=args.port) + return 0 + + def _connector_flags(args: argparse.Namespace) -> dict[str, object]: """Collect the connector CLI flags into the settings-resolution dict.""" keys = ( diff --git a/openadapt_flow/reward/__init__.py b/openadapt_flow/reward/__init__.py new file mode 100644 index 00000000..1532a3e3 --- /dev/null +++ b/openadapt_flow/reward/__init__.py @@ -0,0 +1,32 @@ +"""MIT reference reward worker: verified terminal effects for a training loop. + +A reward receipt states one thing: OpenAdapt read the terminal effect of one +episode through an independent oracle and judged it against one reward +contract. It never states that Flow governed the policy's actions. A model +rollout is not a qualified program, so it never receives an Execute receipt +or an Execute Seal. The two receipts carry different schema ids and cannot +be exchanged for one another. + +The worker signs receipts with a local Ed25519 key, the same way the +reference Execute server signs its local receipts. The key lives in a +sibling data directory. Evidence bytes stay on this machine; a receipt +carries digests only. +""" + +from __future__ import annotations + +from typing import Literal + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8788 +DEFAULT_DATA_DIRNAME = "reward-ref" +REWARD_NOTICE: Literal[ + "Reward receipt. Not an Execute Seal. Flow did not govern the policy." +] = "Reward receipt. Not an Execute Seal. Flow did not govern the policy." + +__all__ = [ + "DEFAULT_HOST", + "DEFAULT_PORT", + "DEFAULT_DATA_DIRNAME", + "REWARD_NOTICE", +] diff --git a/openadapt_flow/reward/calibration.py b/openadapt_flow/reward/calibration.py new file mode 100644 index 00000000..92854c10 --- /dev/null +++ b/openadapt_flow/reward/calibration.py @@ -0,0 +1,162 @@ +"""Exact false-accept bounds for a synthetic-scope reward certificate. + +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. + +The bound is exact: it uses the binomial tail directly, not a normal +approximation. For zero failures it reduces to ``1 - alpha ** (1 / n)``. +""" + +from __future__ import annotations + +import math +import random +from dataclasses import dataclass +from typing import Any, Callable, Sequence + +from openadapt_types.reward import RewardOutcomeV1 + + +def binomial_cdf(k: int, n: int, p: float) -> float: + """P(X <= k) for X ~ Binomial(n, p), computed with exact coefficients.""" + + if k < 0: + return 0.0 + if k >= n: + return 1.0 + if p <= 0.0: + return 1.0 + if p >= 1.0: + return 0.0 + total = 0.0 + for i in range(k + 1): + total += math.comb(n, i) * (p**i) * ((1.0 - p) ** (n - i)) + return min(1.0, total) + + +def clopper_pearson_upper( + failures: int, trials: int, *, confidence: float = 0.95 +) -> float: + """One-sided exact upper bound on a binomial proportion. + + The smallest ``p`` with ``P(X <= failures | trials, p) <= 1 - confidence``. + """ + + if trials <= 0: + raise ValueError("trials must be positive") + if not 0 <= failures <= trials: + raise ValueError("failures must lie in [0, trials]") + if not 0.0 < confidence < 1.0: + raise ValueError("confidence must lie in (0, 1)") + alpha = 1.0 - confidence + if failures == trials: + return 1.0 + if failures == 0: + return 1.0 - alpha ** (1.0 / trials) + low, high = 0.0, 1.0 + for _ in range(200): + mid = (low + high) / 2.0 + if binomial_cdf(failures, trials, mid) > alpha: + low = mid + else: + high = mid + return high + + +@dataclass(frozen=True) +class CalibrationResult: + """What the seed ran and what it found.""" + + trials: int + false_accepts: int + confidence: float + epsilon: float + generator_seed: int + + def as_metadata(self) -> dict[str, Any]: + return { + "calibration_trials": self.trials, + "calibration_false_accepts": self.false_accepts, + "calibration_confidence": self.confidence, + "calibration_generator_seed": self.generator_seed, + "calibration_method": "clopper_pearson_one_sided_exact", + } + + +FAULT_CLASSES: tuple[str, ...] = ( + "extra_record", + "duplicate_record", + "missing_record", + "wrong_type", + "forbidden_present", +) + + +def extradup_trials( + checker: Callable[[Sequence[dict[str, Any]], dict[str, str]], RewardOutcomeV1], + *, + trials: int, + generator_seed: int, + confidence: float = 0.95, +) -> 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. + """ + + 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: + false_accepts += 1 + return CalibrationResult( + trials=trials, + false_accepts=false_accepts, + confidence=confidence, + epsilon=clopper_pearson_upper(false_accepts, trials, confidence=confidence), + generator_seed=generator_seed, + ) + + +def faulted_store(fault: str, patient: str, rng: random.Random) -> list[dict[str, Any]]: + """A MockMed-shaped store carrying one fault for ``patient``.""" + + 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}") diff --git a/openadapt_flow/reward/callables.py b/openadapt_flow/reward/callables.py new file mode 100644 index 00000000..43835ab4 --- /dev/null +++ b/openadapt_flow/reward/callables.py @@ -0,0 +1,276 @@ +"""Reward-function adapters for TRL GRPO and verl. + +Both adapters turn a trainer's per-sample data into an episode descriptor, +ask a scorer for a signed receipt, and hand back the scalar. An UNSCORED +episode never becomes 0.0: + +* TRL: the adapter returns ``None`` for that sample. TRL documents ``None`` + as "this reward function does not apply to this sample" and excludes it + from the reward calculation. +* verl: the reward manager has no such sentinel, so the adapter returns + :data:`UNSCORED_REWARD` (``nan``) together with ``"openadapt_unscored": + True`` in the dict. A NaN poisons a group's advantage instead of + silently training on 0.0. :func:`drop_unscored` removes those samples + from a group before the loss is computed; wire it in, or filter on the + ``openadapt_unscored`` key. + +TRL contract (read 2026-09-01): +https://huggingface.co/docs/trl/main/en/grpo_trainer#using-a-custom-reward-function + reward_func(prompts, completions, completion_ids, trainer_state, **kwargs) + -> list[float | None] +Every dataset column except ``prompt`` arrives in ``**kwargs`` as a list +aligned with ``completions``. ``trainer_state.global_step`` is the policy +update counter the certificate expiry is denominated in. + +verl contract (read 2026-09-01): +https://verl.readthedocs.io/en/latest/preparation/reward_function.html +https://github.com/volcengine/verl/blob/main/verl/workers/reward_manager/naive.py + compute_score(data_source, solution_str, ground_truth, extra_info=None) + -> float | dict (a dict must carry "score"; other keys are logged) +Configured through ``custom_reward_function.path`` and +``custom_reward_function.name``. +""" + +from __future__ import annotations + +import math +from typing import ( + Any, + Callable, + Iterable, + Mapping, + Optional, + Protocol, + Sequence, + TypeVar, +) + +from openadapt_types.reward import RewardEvidenceReceiptV1 + +from openadapt_flow.reward.models import EpisodeDescriptorV1 + +#: The verl-side sentinel for an episode the oracle could not score. +#: ``nan`` on purpose: a trainer that forgets to drop it sees a NaN loss, +#: not a quiet 0.0 that teaches the policy that uncertainty is failure. +UNSCORED_REWARD: float = float("nan") + +T = TypeVar("T") + + +def is_unscored(value: Any) -> bool: + """True for ``None`` (TRL) and for the NaN sentinel (verl).""" + + if value is None: + return True + return isinstance(value, float) and math.isnan(value) + + +def drop_unscored( + rewards: Sequence[Any], *aligned: Sequence[T] +) -> tuple[list[float], list[list[T]]]: + """Filter a group down to the scored samples. + + Returns the kept rewards and, for each aligned sequence passed, the + kept items in the same positions. A group that loses every sample + comes back empty; the trainer must skip that group. + """ + + keep = [index for index, value in enumerate(rewards) if not is_unscored(value)] + kept_rewards = [float(rewards[index]) for index in keep] + kept_aligned = [[seq[index] for index in keep] for seq in aligned] + return kept_rewards, kept_aligned + + +class RewardScorer(Protocol): + """Anything that turns an episode descriptor into a reward envelope. + + :class:`openadapt_flow.reward.worker.RewardWorker` scores in-process. + :class:`HttpRewardClient` calls a worker over HTTP from a trainer node. + """ + + def score_episode(self, payload: Mapping[str, Any]) -> dict[str, Any]: ... + + +class HttpRewardClient: + """Thin client for ``POST /v1/rewards`` on a reward worker.""" + + def __init__(self, base_url: str, token: str, *, timeout_s: float = 30.0) -> None: + self.base_url = base_url.rstrip("/") + self.token = token + self.timeout_s = timeout_s + + def score_episode(self, payload: Mapping[str, Any]) -> dict[str, Any]: + import httpx + + response = httpx.post( + f"{self.base_url}/v1/rewards", + json=dict(payload), + headers={"Authorization": f"Bearer {self.token}"}, + timeout=self.timeout_s, + ) + if response.status_code == 409: + raise RuntimeError( + f"episode already scored: {response.json().get('detail')}" + ) + response.raise_for_status() + return dict(response.json()) + + +def scalar_of(envelope: Mapping[str, Any]) -> Optional[float]: + """The receipt's scalar, or ``None`` when the episode is unscored.""" + + receipt = RewardEvidenceReceiptV1.model_validate(envelope["receipt"]) + return receipt.scalar_reward + + +def episode_from_columns( + *, + episode_id: str, + policy_checkpoint_id: str, + policy_update: int, + reward_contract_digest: str, + oracle_identity: Optional[Mapping[str, Any]], + runtime_signal: str = "completed", +) -> dict[str, Any]: + """Build the wire payload for one episode, in the trainer client's shape. + + ``oracle_identity`` may be ``None`` when the environment registered the + identity with ``RewardWorker.begin_episode`` before the rollout. + """ + + metadata: dict[str, Any] = {"runtime_signal": runtime_signal} + if oracle_identity is not None: + metadata["oracle_identity"] = { + str(k): str(v) for k, v in oracle_identity.items() + } + return EpisodeDescriptorV1( + episode_id=str(episode_id), + policy_checkpoint_id=str(policy_checkpoint_id), + policy_update=int(policy_update), + reward_contract_digest=reward_contract_digest, + metadata=metadata, + ).model_dump(mode="json", exclude_none=True) + + +# -- TRL ---------------------------------------------------------------------- + + +def trl_reward_function( + scorer: RewardScorer, + *, + policy_checkpoint_id: str, + reward_contract_digest: str, + episode_column: str = "episode_id", + identity_column: str = "oracle_identity", + signal_column: str = "runtime_signal", +) -> Callable[..., list[Optional[float]]]: + """Build a ``reward_funcs`` entry for ``trl.GRPOTrainer``. + + The dataset carries one row per episode with ``episode_column`` (the + episode id the environment ran under), optionally ``identity_column`` + (a dict of the oracle identity keys; omit it when the environment + registers the identity with ``begin_episode``), and optionally + ``signal_column``. The policy update comes from + ``trainer_state.global_step``. Each call returns one float per + completion, or ``None`` for an unscored episode. + """ + + def openadapt_verified_effect_reward( + prompts: Sequence[Any], + completions: Sequence[Any], + completion_ids: Optional[Sequence[Any]] = None, + trainer_state: Any = None, + **kwargs: Any, + ) -> list[Optional[float]]: + del prompts, completion_ids + episodes = kwargs.get(episode_column) + if episodes is None: + raise KeyError(f"dataset must carry {episode_column!r}") + identities = kwargs.get(identity_column) or [None] * len(completions) + signals = kwargs.get(signal_column) or ["completed"] * len(completions) + policy_update = int(getattr(trainer_state, "global_step", 0) or 0) + rewards: list[Optional[float]] = [] + for episode_id, identity, signal in zip(episodes, identities, signals): + envelope = scorer.score_episode( + episode_from_columns( + episode_id=episode_id, + policy_checkpoint_id=policy_checkpoint_id, + policy_update=policy_update, + reward_contract_digest=reward_contract_digest, + oracle_identity=identity, + runtime_signal=signal, + ) + ) + rewards.append(scalar_of(envelope)) + return rewards + + return openadapt_verified_effect_reward + + +# -- verl --------------------------------------------------------------------- + + +def verl_compute_score( + scorer: RewardScorer, + *, + policy_checkpoint_id: str, + reward_contract_digest: str, + episode_key: str = "openadapt_episode", +) -> Callable[..., dict[str, Any]]: + """Build a verl ``compute_score`` over a reward worker. + + ``extra_info[episode_key]`` must hold ``episode_id`` and + ``policy_update``, plus ``oracle_identity`` unless the environment + registered it with ``begin_episode``, and optionally ``runtime_signal``. The returned dict + carries ``score`` (the scalar, or :data:`UNSCORED_REWARD`), the receipt + id, and the flags a filter needs. verl stores every extra key in the + batch's non-tensor data, so ``openadapt_unscored`` survives to the point + where :func:`drop_unscored` can act on it. + """ + + def compute_score( + data_source: Any, + solution_str: Any, + ground_truth: Any, + extra_info: Optional[Mapping[str, Any]] = None, + ) -> dict[str, Any]: + del data_source, solution_str, ground_truth + info = dict(extra_info or {}) + episode = info.get(episode_key) + if not isinstance(episode, Mapping): + raise KeyError(f"extra_info must carry {episode_key!r}") + envelope = scorer.score_episode( + episode_from_columns( + episode_id=str(episode["episode_id"]), + policy_checkpoint_id=policy_checkpoint_id, + policy_update=int(episode.get("policy_update", 0)), + reward_contract_digest=reward_contract_digest, + oracle_identity=episode.get("oracle_identity"), + runtime_signal=str(episode.get("runtime_signal", "completed")), + ) + ) + receipt = envelope["receipt"] + scalar = scalar_of(envelope) + return { + "score": UNSCORED_REWARD if scalar is None else float(scalar), + "openadapt_unscored": scalar is None, + "openadapt_reward_outcome": receipt["reward_outcome"], + "openadapt_receipt_id": receipt["receipt_id"], + "openadapt_certified": bool(receipt["certified"]), + "openadapt_development_only": bool(receipt["development_only"]), + } + + return compute_score + + +def scored_groups( + groups: Iterable[Sequence[Mapping[str, Any]]], +) -> list[list[Mapping[str, Any]]]: + """Drop unscored samples from each verl group of ``compute_score`` dicts.""" + + kept: list[list[Mapping[str, Any]]] = [] + for group in groups: + rewards = [item["score"] for item in group] + _kept_rewards, (kept_items,) = drop_unscored(rewards, list(group)) + kept.append(kept_items) + return kept diff --git a/openadapt_flow/reward/models.py b/openadapt_flow/reward/models.py new file mode 100644 index 00000000..8fdc2ea0 --- /dev/null +++ b/openadapt_flow/reward/models.py @@ -0,0 +1,457 @@ +"""Local reward-worker models: the contract bundle, the episode, the recipe. + +The portable contract and receipt are ``RewardContractV1`` and +``RewardEvidenceReceiptV1`` from ``openadapt-types``. The contract names its +required effects, forbidden effects, and oracle by digest only. This module +holds the local, digest-checked bundle that carries the bytes behind those +digests, plus the episode descriptor a trainer submits. + +Extra keys are forbidden everywhere. Screenshot, OCR, and parameter fields +have no place in a receipt; the same allow-list discipline as +:mod:`openadapt_flow.execute.models` applies. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Literal, Optional + +from openadapt_types.oracle import OracleChannel, OracleTier, tier_of +from openadapt_types.process_capability import _digest_payload +from openadapt_types.reward import ( + RewardCertificateV1, + RewardContractV1, +) +from pydantic import ( + BaseModel, + ConfigDict, + Field, + StrictBool, + StrictInt, + StrictStr, + field_validator, + model_validator, +) + +from openadapt_flow.reward import REWARD_NOTICE +from openadapt_flow.runtime.effects.effect import Effect + +CONTRACT_FILE = "contract.json" +REQUIRED_EFFECTS_FILE = "required_effects.json" +FORBIDDEN_EFFECTS_FILE = "forbidden_effects.json" +ORACLE_FILE = "oracle.json" +CERTIFICATE_FILE = "certificate.json" + +_OPAQUE_ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$" + +#: Fields a shareable reward receipt or evidence summary must never carry. +#: Mirrors the Execute allow-list; the reward path adds the trainer-side +#: names a rollout buffer tends to attach to an episode. +FORBIDDEN_RECEIPT_KEYS = frozenset( + { + "screenshot", + "screenshots", + "frames", + "frame", + "video", + "trajectory", + "observations", + "observation", + "actions", + "action", + "ocr", + "ocr_text", + "typed_value", + "typed_values", + "parameters", + "parameter", + "prompt", + "prompts", + "completion", + "completions", + "completion_ids", + "url", + "hostname", + "coordinate", + "coordinates", + "application_name", + "organization_name", + "user_name", + "workflow_name", + "phi", + "image", + "after_png", + "before_png", + "note", + "record_id", + "records", + } +) + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +RuntimeSignal = Literal[ + "completed", + "halted_before_effect", + "refused", + "rejected_policy", + "failed_platform", +] + + +class EpisodeDescriptorV1(_Strict): + """What a trainer submits to have one episode's terminal effect read. + + The wire shape is the one ``openadapt_evals.reward.receipts. + EpisodeDescriptor`` sends: ``episode_id``, ``policy_checkpoint_id``, + ``policy_update``, ``reward_contract_digest``, and optional ``task_id``, + ``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. + + ``runtime_signal`` (or ``metadata["runtime_signal"]``) is what the + episode runtime reported about its own end. The oracle read decides; the + signal only picks which zero-or-penalty outcome applies when the store + agrees that nothing landed. + """ + + schema_version: Literal["openadapt.reward-episode/v1"] = ( + "openadapt.reward-episode/v1" + ) + episode_id: StrictStr = Field(pattern=_OPAQUE_ID_PATTERN) + policy_checkpoint_id: StrictStr = Field(pattern=_OPAQUE_ID_PATTERN) + policy_update: StrictInt = Field(ge=0) + reward_contract_digest: StrictStr = Field(pattern=r"^sha256:[0-9a-f]{64}$") + task_id: Optional[StrictStr] = Field(default=None, pattern=_OPAQUE_ID_PATTERN) + environment_id: Optional[StrictStr] = Field( + default=None, pattern=_OPAQUE_ID_PATTERN + ) + metadata: dict[StrictStr, Any] = Field(default_factory=dict) + oracle_identity: Optional[dict[StrictStr, StrictStr]] = Field( + default=None, min_length=1, max_length=32 + ) + runtime_signal: Optional[RuntimeSignal] = None + + @field_validator("oracle_identity") + @classmethod + def _identity_values( + cls, values: Optional[dict[str, str]] + ) -> Optional[dict[str, str]]: + if values is None: + return None + return _clean_identity(values) + + @field_validator("metadata") + @classmethod + def _metadata_keys(cls, values: dict[str, Any]) -> dict[str, Any]: + bad = FORBIDDEN_RECEIPT_KEYS.intersection(values) + if bad: + raise ValueError( + "metadata carries rollout or PHI keys: " + ", ".join(sorted(bad)) + ) + return values + + def resolved_identity(self) -> Optional[dict[str, str]]: + """The oracle identity the descriptor itself carries, if any.""" + + if self.oracle_identity is not None: + return dict(self.oracle_identity) + raw = self.metadata.get("oracle_identity") + if raw is None: + return None + if not isinstance(raw, dict) or not raw: + raise ValueError("metadata.oracle_identity must be a non-empty object") + return _clean_identity({str(k): str(v) for k, v in raw.items()}) + + def resolved_signal(self) -> str: + if self.runtime_signal is not None: + return self.runtime_signal + raw = self.metadata.get("runtime_signal", "completed") + if raw not in RUNTIME_SIGNALS: + raise ValueError(f"metadata.runtime_signal {raw!r} is not a known signal") + return str(raw) + + +RUNTIME_SIGNALS = frozenset( + { + "completed", + "halted_before_effect", + "refused", + "rejected_policy", + "failed_platform", + } +) + + +def _clean_identity(values: dict[str, str]) -> dict[str, str]: + for key, value in values.items(): + if not key or not value: + raise ValueError("oracle_identity keys and values must be non-empty") + return dict(sorted(values.items())) + + +class OracleRecipeV1(_Strict): + """How the worker reads the system of record. Interface only. + + A recipe carries no credential. ``headers_env`` and ``token_env`` name + environment variables the worker reads at start; the bundle on disk + never holds the secret. A per-system-of-record recipe for a real + deployment stays private; this public shape is the mechanism. + """ + + kind: Literal["json_file", "screen_dump", "rest", "sqlite", "fhir", "file_arrival"] + #: ``json_file`` / ``screen_dump``: a JSON document on disk. Relative + #: paths resolve against the bundle directory. + path: Optional[StrictStr] = None + #: ``json_file`` / ``rest``: key of the records list in the document + #: (``null`` when the document is the list). + records_key: Optional[StrictStr] = "records" + #: ``rest`` / ``fhir``: base URL of the read endpoint. + base_url: Optional[StrictStr] = None + #: ``rest``: path of the records document. + records_path: StrictStr = "/api/db" + #: ``rest``: environment variable holding a JSON object of headers. + headers_env: Optional[StrictStr] = None + #: ``sqlite``: database file. ``query``: one read-only SELECT. + query: Optional[StrictStr] = None + #: ``fhir``: resource type and search parameters; ``token_env`` names the + #: bearer token variable. + resource_type: StrictStr = "Observation" + search_params: dict[StrictStr, StrictStr] = Field(default_factory=dict) + field_paths: Optional[dict[StrictStr, StrictStr]] = None + token_env: Optional[StrictStr] = None + #: ``file_arrival``: watched directory and glob. + pattern: StrictStr = "*" + timeout_s: float = 5.0 + + @model_validator(mode="after") + def _kind_fields(self) -> "OracleRecipeV1": + needs_path = {"json_file", "screen_dump", "sqlite", "file_arrival"} + if self.kind in needs_path and not self.path: + raise ValueError(f"oracle recipe {self.kind} requires path") + if self.kind in {"rest", "fhir"} and not self.base_url: + raise ValueError(f"oracle recipe {self.kind} requires base_url") + if self.kind == "sqlite" and not self.query: + raise ValueError("oracle recipe sqlite requires query") + return self + + @property + def channel(self) -> OracleChannel: + return _RECIPE_CHANNEL[self.kind] + + @property + def tier(self) -> OracleTier: + return tier_of(self.channel) + + @property + def digest(self) -> str: + return _digest_payload(self.model_dump(mode="json")) + + def resolve_path(self, base_dir: Path) -> Path: + path = Path(self.path or "") + return path if path.is_absolute() else base_dir / path + + def headers(self) -> Optional[dict[str, str]]: + if not self.headers_env: + return None + raw = os.environ.get(self.headers_env, "") + if not raw: + return None + parsed = json.loads(raw) + if not isinstance(parsed, dict): + raise ValueError(f"{self.headers_env} must hold a JSON object") + return {str(k): str(v) for k, v in parsed.items()} + + def token(self) -> Optional[str]: + if not self.token_env: + return None + 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. +_RECIPE_CHANNEL: dict[str, OracleChannel] = { + "json_file": OracleChannel.FILE, + "screen_dump": OracleChannel.OCR, + "rest": OracleChannel.API, + "sqlite": OracleChannel.DB, + "fhir": OracleChannel.API, + "file_arrival": OracleChannel.FILE, +} + + +class RewardBundle(_Strict): + """One reward contract with the bytes behind its digests, checked. + + 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, or when an effect references an identity + key the contract does not declare. + """ + + model_config = ConfigDict(extra="forbid", frozen=True, arbitrary_types_allowed=True) + + directory: Path + contract: RewardContractV1 + required_effects: tuple[Effect, ...] + forbidden_effects: tuple[Effect, ...] + oracle: OracleRecipeV1 + certificate: Optional[RewardCertificateV1] = None + + @classmethod + def load(cls, directory: Path | str) -> "RewardBundle": + base = Path(directory).expanduser().resolve() + if not base.is_dir(): + raise BundleError(f"reward contract directory is missing: {base}") + contract = RewardContractV1.model_validate(_read_json(base / CONTRACT_FILE)) + required_raw = _read_json(base / REQUIRED_EFFECTS_FILE) + forbidden_raw = _read_json(base / FORBIDDEN_EFFECTS_FILE) + oracle_raw = _read_json(base / ORACLE_FILE) + for name, raw, want in ( + ( + REQUIRED_EFFECTS_FILE, + required_raw, + contract.required_effect_contract_digest, + ), + ( + FORBIDDEN_EFFECTS_FILE, + forbidden_raw, + contract.forbidden_effect_contract_digest, + ), + (ORACLE_FILE, oracle_raw, contract.oracle.oracle_contract_digest), + ): + got = _digest_payload(raw) + if got != want: + raise BundleError( + f"{name} digest {got} does not match the contract's {want}" + ) + oracle = OracleRecipeV1.model_validate(oracle_raw) + if oracle.channel is not contract.oracle.channel: + raise BundleError( + f"oracle recipe channel {oracle.channel.value} does not match " + f"the contract channel {contract.oracle.channel.value}" + ) + required = tuple(_effects(required_raw, REQUIRED_EFFECTS_FILE)) + forbidden = tuple(_effects(forbidden_raw, FORBIDDEN_EFFECTS_FILE)) + if not required: + raise BundleError("a reward contract requires at least one required effect") + declared = set(contract.oracle.identity_keys) + for effect in (*required, *forbidden): + missing = effect.referenced_params() - declared + if missing: + names = ", ".join(sorted(missing)) + raise BundleError( + f"effect references identity keys the contract does not " + f"declare: {names}" + ) + certificate: Optional[RewardCertificateV1] = None + cert_path = base / CERTIFICATE_FILE + if cert_path.is_file(): + certificate = RewardCertificateV1.model_validate(_read_json(cert_path)) + if certificate.reward_contract_digest != contract.digest: + raise BundleError( + "certificate.json binds a different reward contract digest" + ) + return cls( + directory=base, + contract=contract, + required_effects=required, + forbidden_effects=forbidden, + oracle=oracle, + certificate=certificate, + ) + + @property + def identity_keys(self) -> tuple[str, ...]: + return tuple(self.contract.oracle.identity_keys) + + def check_identity(self, identity: dict[str, str]) -> None: + """Refuse an identity whose key set differs from the contract's.""" + + want = set(self.identity_keys) + got = set(identity) + extra = sorted(got - want) + missing = sorted(want - got) + if extra: + raise IdentityError( + "oracle_identity carries keys the contract does not declare: " + + ", ".join(extra) + ) + if missing: + raise IdentityError( + "oracle_identity is missing declared keys: " + ", ".join(missing) + ) + + +class BundleError(ValueError): + """The bundle on disk does not match its contract.""" + + +class IdentityError(ValueError): + """The episode's oracle identity does not fit the contract.""" + + +class SelfSignedRewardEnvelopeV1(_Strict): + """Local envelope around a portable reward receipt. + + ``execute_seal`` and ``production_seal`` are always false. ``issuer`` is + always ``self_signed``. The receipt inside is signed on its own; this + envelope adds the local key fingerprint and the notice. + """ + + # No ``schema_version`` here on purpose: the trainer-side client treats a + # body with ``receipt`` and no ``schema_version`` as an envelope. + envelope: Literal["openadapt.reward-self-signed-envelope/v1"] = ( + "openadapt.reward-self-signed-envelope/v1" + ) + issuer: Literal["self_signed"] = "self_signed" + issuer_key_fingerprint: StrictStr = Field(pattern=r"^sha256:[0-9a-f]{64}$") + execute_seal: Literal[False] = False + production_seal: Literal[False] = False + verify_host: Literal["local"] = "local" + flow_governed_policy: Literal[False] = False + notice: Literal[ + "Reward receipt. Not an Execute Seal. Flow did not govern the policy." + ] = REWARD_NOTICE + unscored: StrictBool + receipt: dict[str, Any] + + +def assert_no_forbidden_keys(payload: dict[str, Any]) -> None: + """Refuse a receipt dict that carries a PHI, screenshot, or rollout field.""" + + extra = FORBIDDEN_RECEIPT_KEYS.intersection(payload) + if extra: + names = ", ".join(sorted(extra)) + raise ValueError(f"reward receipt forbids extra/PHI keys: {names}") + + +def _read_json(path: Path) -> Any: + if not path.is_file(): + raise BundleError(f"reward bundle file is missing: {path}") + try: + return json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise BundleError(f"reward bundle file is not JSON: {path}") from exc + + +def _effects(raw: Any, name: str) -> list[Effect]: + if not isinstance(raw, list): + raise BundleError(f"{name} must be a JSON list of effects") + effects: list[Effect] = [] + for item in raw: + if not isinstance(item, dict): + raise BundleError(f"{name} entries must be JSON objects") + effects.append(Effect.model_validate(item)) + return effects diff --git a/openadapt_flow/reward/oracles.py b/openadapt_flow/reward/oracles.py new file mode 100644 index 00000000..0e5638e1 --- /dev/null +++ b/openadapt_flow/reward/oracles.py @@ -0,0 +1,193 @@ +"""Oracle adapters for the reward worker. + +Every adapter here satisfies ``openadapt_types.oracle.OracleAdapter``: a +``channel`` and a read-only ``read(identity)`` that returns one +``OracleObservation``. The channel sets the tier. The observation's value +carries the system-of-record records the shared judge consumes, plus a +``reachable`` flag; an unreachable read is a value with ``reachable`` +false, never a guessed empty list. + +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. +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any, Mapping, Optional + +from openadapt_types.oracle import OracleAdapter, OracleChannel, OracleObservation + +from openadapt_flow.reward.models import OracleRecipeV1 +from openadapt_flow.runtime.effects.effect import EffectState + + +def observation( + channel: OracleChannel, + identity: Mapping[str, str], + records: Optional[list[dict[str, Any]]], +) -> OracleObservation: + """Wrap a raw read as one observation; ``None`` records mean unreachable.""" + + return OracleObservation( + channel=channel, + identity=dict(identity), + value={ + "reachable": records is not None, + "records": list(records or []), + }, + ) + + +def records_of(observed: OracleObservation) -> Optional[list[dict[str, Any]]]: + """Return the records an observation carries, or ``None`` if unreachable.""" + + if not observed.value.get("reachable"): + return None + raw = observed.value.get("records") + if not isinstance(raw, list): + return None + return [dict(item) for item in raw if isinstance(item, dict)] + + +def effect_state_of(observed: OracleObservation, substrate: str) -> EffectState: + """Project an observation onto the judge's pre-state shape.""" + + records = records_of(observed) + return EffectState( + substrate=substrate, + reachable=records is not None, + records=records or [], + ) + + +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. + """ + + 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: + try: + body = json.loads(self.path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return observation(self.channel, identity, None) + if self.records_key is None: + records = body + elif isinstance(body, dict): + records = body.get(self.records_key) + else: + records = None + if not isinstance(records, list): + return observation(self.channel, identity, None) + return observation( + self.channel, + identity, + [item for item in records if isinstance(item, dict)], + ) + + +class VerifierOracle: + """Adapt an effect-kit verifier's fresh read into an oracle observation. + + The kit verifier owns the transport (REST GET, read-only SELECT, FHIR + search, directory listing). This adapter only asks it for a fresh + snapshot and forwards the records. + """ + + def __init__(self, verifier: Any, *, channel: OracleChannel) -> None: + self.verifier = verifier + self.channel = channel + + def read(self, identity: Mapping[str, str]) -> OracleObservation: + try: + state = self.verifier.capture_post_state(None) + except Exception: # noqa: BLE001 - an unreadable store is not a guess + return observation(self.channel, identity, None) + if not getattr(state, "reachable", False): + return observation(self.channel, identity, None) + 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.""" + + if recipe.kind == "json_file": + return JsonDocumentOracle( + recipe.resolve_path(base_dir), + channel=OracleChannel.FILE, + records_key=recipe.records_key, + ) + if recipe.kind == "screen_dump": + return JsonDocumentOracle( + recipe.resolve_path(base_dir), + channel=OracleChannel.OCR, + records_key=recipe.records_key, + ) + if recipe.kind == "rest": + from openadapt_flow.runtime.effects.rest import RestRecordVerifier + + return VerifierOracle( + RestRecordVerifier( + str(recipe.base_url), + records_path=recipe.records_path, + records_key=recipe.records_key, + headers=recipe.headers(), + timeout_s=recipe.timeout_s, + ), + channel=OracleChannel.API, + ) + if recipe.kind == "sqlite": + from openadapt_flow.runtime.effects.sql import SqlRecordVerifier + + database = recipe.resolve_path(base_dir) + + def connect() -> sqlite3.Connection: + uri = f"file:{database}?mode=ro" + conn = sqlite3.connect(uri, uri=True) + conn.row_factory = sqlite3.Row + return conn + + return VerifierOracle( + SqlRecordVerifier(connect, str(recipe.query), timeout_s=recipe.timeout_s), + channel=OracleChannel.DB, + ) + if recipe.kind == "fhir": + from openadapt_flow.runtime.effects.fhir import FhirEffectVerifier + + return VerifierOracle( + FhirEffectVerifier( + str(recipe.base_url), + resource_type=recipe.resource_type, + search_params=dict(recipe.search_params), + field_paths=recipe.field_paths, + access_token=recipe.token(), + timeout_s=recipe.timeout_s, + ), + channel=OracleChannel.API, + ) + if recipe.kind == "file_arrival": + from openadapt_flow.runtime.effects.file_arrival import FileArrivalVerifier + + return VerifierOracle( + FileArrivalVerifier(recipe.resolve_path(base_dir), pattern=recipe.pattern), + channel=OracleChannel.FILE, + ) + raise ValueError(f"unknown oracle recipe kind {recipe.kind!r}") diff --git a/openadapt_flow/reward/seed.py b/openadapt_flow/reward/seed.py new file mode 100644 index 00000000..525d0877 --- /dev/null +++ b/openadapt_flow/reward/seed.py @@ -0,0 +1,320 @@ +"""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. +* ``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. + +Nothing here is a production recipe. Both bundles are synthetic. +""" + +from __future__ import annotations + +import base64 +import json +from pathlib import Path +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +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.models import ( + CERTIFICATE_FILE, + CONTRACT_FILE, + FORBIDDEN_EFFECTS_FILE, + ORACLE_FILE, + REQUIRED_EFFECTS_FILE, + RewardBundle, +) + +MOCKMED_TASK_ID = "task_mockmed_encounter_note" +MOCKMED_ENVIRONMENT_ID = "environment_mockmed_synthetic" +MOCKMED_CONTRACT_ID = "reward_contract_mockmed" +MOCKMED_TIER0_CONTRACT_ID = "reward_contract_mockmed_tier0" +MOCKMED_HONEST_PATIENT = "patient-honest-0001" +MOCKMED_LIE_PATIENT = "patient-lie-0002" +MOCKMED_DUPLICATE_PATIENT = "patient-dup-0003" +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%). +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", + }, +] + +_SCREEN: list[dict[str, Any]] = [ + {"patient_id": MOCKMED_HONEST_PATIENT, "type": "Triage", "status": "saved"}, + {"patient_id": MOCKMED_LIE_PATIENT, "type": "Triage", "status": "saved"}, +] + +_REQUIRED_EFFECTS: list[dict[str, Any]] = [ + { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}, "type": "Triage"}, + "expected_count": 1, + } +] + +_FORBIDDEN_EFFECTS: list[dict[str, Any]] = [ + { + "kind": "record_written", + "match": {"patient_id": {"param": "patient_id"}, "type": "Discharge"}, + "expected_count": 1, + } +] + + +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.""" + + data_dir = Path(data_dir) + store = data_dir / "mockmed" + store.mkdir(parents=True, exist_ok=True) + _write(store / "records.json", {"records": _RECORDS}) + _write(store / "screen.json", {"records": _SCREEN}) + + tier2 = data_dir / "contracts" / "mockmed" + write_bundle( + tier2, + contract_id=MOCKMED_CONTRACT_ID, + oracle={ + "kind": "json_file", + "path": str(store / "records.json"), + "records_key": "records", + }, + channel="file", + key=key, + issuer_key_id=issuer_key_id, + certify=True, + ) + tier0 = data_dir / "contracts" / "mockmed-tier0" + write_bundle( + tier0, + contract_id=MOCKMED_TIER0_CONTRACT_ID, + oracle={ + "kind": "screen_dump", + "path": str(store / "screen.json"), + "records_key": "records", + }, + channel="ocr", + key=key, + issuer_key_id=issuer_key_id, + certify=False, + ) + return {"tier2": tier2, "tier0": tier0} + + +def mockmed_episode( + patient_id: str, + *, + episode_id: str, + contract_digest: str, + policy_update: int = 0, + runtime_signal: str = "completed", +) -> dict[str, Any]: + """The wire payload for one seeded episode, in the trainer client's shape.""" + + return { + "episode_id": episode_id, + "policy_checkpoint_id": MOCKMED_CHECKPOINT, + "policy_update": policy_update, + "reward_contract_digest": contract_digest, + "task_id": MOCKMED_TASK_ID, + "environment_id": MOCKMED_ENVIRONMENT_ID, + "metadata": { + "oracle_identity": {"patient_id": patient_id}, + "runtime_signal": runtime_signal, + }, + } + + +def write_bundle( + directory: Path, + *, + contract_id: str, + oracle: dict[str, Any], + channel: str, + key: Ed25519PrivateKey, + issuer_key_id: str, + certify: bool, + required_effects: list[dict[str, Any]] | None = None, + forbidden_effects: list[dict[str, Any]] | None = None, +) -> None: + """Write one synthetic bundle. Tests pass their own effects.""" + + directory.mkdir(parents=True, exist_ok=True) + required = _canonical( + _REQUIRED_EFFECTS if required_effects is None else required_effects + ) + forbidden = _canonical( + _FORBIDDEN_EFFECTS if forbidden_effects is None else forbidden_effects + ) + oracle_doc = _canonical(oracle) + corpus_digest = _digest_payload({"corpus": "synthetic-mockmed", "size": 0}) + contract = RewardContractV1.model_validate( + { + "contract_id": contract_id, + "contract_version": "version_0001", + "task_id": MOCKMED_TASK_ID, + "task_digest": _digest_payload({"task": MOCKMED_TASK_ID}), + "environment_id": MOCKMED_ENVIRONMENT_ID, + "environment_digest": _digest_payload( + {"environment": MOCKMED_ENVIRONMENT_ID} + ), + "required_effect_contract_digest": _digest_payload(required), + "forbidden_effect_contract_digest": _digest_payload(forbidden), + "oracle": { + "channel": channel, + "identity_keys": ["patient_id"], + "oracle_contract_digest": _digest_payload(oracle_doc), + }, + "components": [{"name": "terminal_effect", "weight": 1.0}], + # The synthetic banner lie yields 0, not the default -1 penalty: + # this fixture demonstrates "no reward", not a tuned penalty. + "scoring": {"wrong_effect_reward": 0.0}, + "certificate_policy": { + "epsilon": 0.05, + "delta": 0.05, + "threshold": 0.5, + "calibration_corpus_digest": corpus_digest, + "expiry_policy_updates": CERTIFICATE_EXPIRY_UPDATES, + }, + } + ) + _write(directory / CONTRACT_FILE, contract.model_dump(mode="json")) + _write(directory / REQUIRED_EFFECTS_FILE, required) + _write(directory / FORBIDDEN_EFFECTS_FILE, forbidden) + _write(directory / ORACLE_FILE, oracle_doc) + if certify: + calibration = calibrate_bundle(directory) + certificate = self_signed_certificate( + contract, + key=key, + issuer_key_id=issuer_key_id, + issued_at_policy_update=0, + expiry_policy_updates=CERTIFICATE_EXPIRY_UPDATES, + epsilon=calibration.epsilon, + delta=1.0 - calibration.confidence, + ) + _write(directory / CERTIFICATE_FILE, certificate.model_dump(mode="json")) + _write(directory / CALIBRATION_FILE, calibration.as_metadata()) + + +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. + """ + + from openadapt_types.oracle import OracleChannel + + from openadapt_flow.reward.oracles import observation + from openadapt_flow.reward.worker import judge_episode + from openadapt_flow.runtime.effects.effect import EffectState + + bundle = RewardBundle.load(directory) + channel = OracleChannel(bundle.contract.oracle.channel) + + 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 + + return extradup_trials( + checker, + trials=CALIBRATION_TRIALS, + generator_seed=CALIBRATION_SEED, + confidence=CALIBRATION_CONFIDENCE, + ) + + +def self_signed_certificate( + contract: RewardContractV1, + *, + key: Ed25519PrivateKey, + issuer_key_id: str, + issued_at_policy_update: int, + expiry_policy_updates: int, + epsilon: float, + delta: float, + certificate_id: str = "reward_certificate_mockmed", +) -> RewardCertificateV1: + """A certificate signed by the local key, synthetic scope. + + ``epsilon`` comes from :func:`calibrate_bundle`, never from a constant. + A production-scope certificate is calibrated on a held-out corpus by + the OpenAdapt control service and is not published. This one exists so + the seeded run can show a certified receipt next to an uncertified one. + """ + + from openadapt_flow.reward.worker import _now + + unsigned = { + "schema_version": "openadapt.reward-certificate/v1", + "certificate_id": certificate_id, + "reward_contract_digest": contract.digest, + "checker_configuration_digest": _digest_payload({"checker": "synthetic"}), + "epsilon": epsilon, + "delta": delta, + "threshold": contract.certificate_policy.threshold, + "calibration_corpus_digest": contract.certificate_policy.calibration_corpus_digest, + "calibration_scope": "synthetic", + "issued_at_policy_update": issued_at_policy_update, + "expiry_policy_updates": expiry_policy_updates, + "issued_at": _now(), + "issuer": "self_signed", + "issuer_key_id": issuer_key_id, + } + signature = base64.b64encode(key.sign(canonical_json_bytes(unsigned))).decode( + "ascii" + ) + return RewardCertificateV1.model_validate( + {**unsigned, "signature_algorithm": "ed25519", "signature": signature} + ) + + +def _canonical(value: Any) -> Any: + return json.loads(json.dumps(value, sort_keys=True)) + + +def _write(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) diff --git a/openadapt_flow/reward/serve.py b/openadapt_flow/reward/serve.py new file mode 100644 index 00000000..e36003b8 --- /dev/null +++ b/openadapt_flow/reward/serve.py @@ -0,0 +1,228 @@ +"""Loopback HTTP surface for the reference reward worker. + +Routes: + +* ``GET /health``: issuer, key fingerprint, contract digest, oracle tier. +* ``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 + episode still gets a receipt; the envelope says ``unscored: true`` and + the receipt carries no scalar. +* ``GET /v1/rewards/{receipt_id}``: read a stored envelope. +* ``POST /v1/graders/openai``: the OpenAI grader shape, see below. + +Every route but ``/health`` needs the local bearer token. + +OpenAI grader route. The OpenAI reinforcement fine-tuning graders guide +(https://developers.openai.com/api/docs/guides/graders, read 2026-09-01) +and the RFT guide +(https://developers.openai.com/api/docs/guides/reinforcement-fine-tuning, +same date) document five grader types: ``string_check``, +``text_similarity``, ``score_model``, ``python``, and ``multi``. None of +them calls a user-hosted HTTP endpoint, and the ``python`` grader runs +with no network access. The only documented custom-grader contract is the +``python`` grader's function:: + + def grade(sample: dict[str, Any], item: dict[str, Any]) -> float + +where ``sample`` holds the model output (``output_text``, ``output_json``, +``output_tools``, ``choices``) and ``item`` holds the dataset row, and the +returned float lies in ``[0, 1]``. The guide also states that an +exception or an invalid float "will be marked as invalid and return a 0 +grade". That is the rule this worker must not follow for an unscored +episode. + +So this route takes ``{"sample": ..., "item": ...}``, reads the episode +descriptor from ``item``, and answers ``{"score": float}`` in ``[0, 1]`` +plus the receipt id. An unscored episode answers HTTP 422 with +``{"error": "unscored", ...}``. The schema has no "do not score" value, so +a wrapper that feeds a hosted grader must drop that sample before the +grader sees it; forwarding the 422 as 0 would teach the policy that +uncertainty is failure. +""" + +from __future__ import annotations + +import hmac +from pathlib import Path +from typing import Any, Optional + +from fastapi import FastAPI, Header, HTTPException, Request +from fastapi.responses import JSONResponse + +from openadapt_flow import __version__ +from openadapt_flow.reward import REWARD_NOTICE +from openadapt_flow.reward.worker import RewardWorker, RewardWorkerError + +OPENAI_GRADER_ROUTE = "/v1/graders/openai" + + +def create_app(worker: RewardWorker) -> FastAPI: + """Build the one-process HTTP app over a worker.""" + + app = FastAPI( + title="OpenAdapt reference reward worker", + version=__version__, + docs_url=None, + redoc_url=None, + openapi_url=None, + ) + app.state.reward = worker + + @app.exception_handler(RewardWorkerError) + async def _worker_error(_request: Request, exc: RewardWorkerError) -> JSONResponse: + return JSONResponse(status_code=exc.status_code, content=exc.body()) + + @app.get("/health") + def health() -> dict[str, Any]: + return { + "status": "ok", + "service": "openadapt-reward-ref", + "issuer": "self_signed", + "issuer_key_fingerprint": worker.fingerprint, + "reward_contract_digest": worker.contract.digest, + "oracle_channel": worker.bundle.oracle.channel.value, + "oracle_tier": int(worker.bundle.oracle.tier), + "certificate_present": worker.certificate is not None, + "execute_seal": False, + "production_seal": False, + "notice": REWARD_NOTICE, + } + + @app.post("/v1/rewards", status_code=200, response_model=None) + async def create_reward( + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(worker, authorization) + payload = await _json_object(request) + envelope = worker.score_episode(payload) + return JSONResponse( + status_code=200, content=envelope, headers=_issuer_headers(worker) + ) + + @app.get("/v1/rewards/{receipt_id}", response_model=None) + def get_reward( + receipt_id: str, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(worker, authorization) + return JSONResponse( + content=worker.get_receipt(receipt_id), headers=_issuer_headers(worker) + ) + + @app.post(OPENAI_GRADER_ROUTE, response_model=None) + async def openai_grader( + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(worker, authorization) + payload = await _json_object(request) + item = payload.get("item") + if not isinstance(item, dict): + raise HTTPException( + status_code=400, detail="body must carry sample and item objects" + ) + if not isinstance(payload.get("sample"), dict): + raise HTTPException( + status_code=400, detail="body must carry sample and item objects" + ) + envelope = worker.score_episode(_episode_from_item(item)) + receipt = envelope["receipt"] + scalar = receipt["scalar_reward"] + if scalar is None: + return JSONResponse( + status_code=422, + content={ + "error": "unscored", + "detail": ( + "the oracle could not score this episode; drop the " + "sample, do not grade it 0" + ), + "reward_outcome": receipt["reward_outcome"], + "uncertainty": receipt["uncertainty"], + "receipt_id": receipt["receipt_id"], + }, + headers=_issuer_headers(worker), + ) + positive = worker.contract.scoring.verified_reward + return JSONResponse( + content={ + "score": max(0.0, min(1.0, float(scalar) / positive)), + "scalar_reward": scalar, + "reward_outcome": receipt["reward_outcome"], + "certified": receipt["certified"], + "development_only": receipt["development_only"], + "receipt_id": receipt["receipt_id"], + }, + headers=_issuer_headers(worker), + ) + + return app + + +def serve( + worker: RewardWorker, + *, + host: str = "127.0.0.1", + port: int = 8788, +) -> None: + """Block on uvicorn. Caller prints the banner before this.""" + + import uvicorn + + uvicorn.run(create_app(worker), host=host, port=port, log_level="info") + + +def _episode_from_item(item: dict[str, Any]) -> dict[str, Any]: + """The dataset row carries the episode descriptor fields by name.""" + + fields: dict[str, Any] = { + key: item[key] + for key in ( + "episode_id", + "policy_checkpoint_id", + "policy_update", + "reward_contract_digest", + "task_id", + "environment_id", + "metadata", + "oracle_identity", + "runtime_signal", + ) + if key in item + } + return fields + + +def _require_bearer(worker: RewardWorker, authorization: Optional[str]) -> None: + scheme, separator, token = (authorization or "").partition(" ") + if separator != " " or scheme.lower() != "bearer": + raise HTTPException(status_code=401, detail="bearer token required") + if not hmac.compare_digest(token.strip(), worker.token): + raise HTTPException(status_code=401, detail="invalid bearer token") + + +def _issuer_headers(worker: RewardWorker) -> dict[str, str]: + return { + "X-OpenAdapt-Issuer": "self_signed", + "X-OpenAdapt-Issuer-Fingerprint": worker.fingerprint, + "X-OpenAdapt-Execute-Seal": "false", + "X-OpenAdapt-Production-Seal": "false", + } + + +async def _json_object(request: Request) -> dict[str, Any]: + try: + payload = await request.json() + except Exception as exc: + raise HTTPException(status_code=400, detail="body must be JSON") from exc + if not isinstance(payload, dict): + raise HTTPException(status_code=400, detail="body must be a JSON object") + return payload + + +def default_data_dir() -> Path: + from openadapt_flow.reward.worker import default_data_dir as _default + + return _default() diff --git a/openadapt_flow/reward/worker.py b/openadapt_flow/reward/worker.py new file mode 100644 index 00000000..a9458337 --- /dev/null +++ b/openadapt_flow/reward/worker.py @@ -0,0 +1,553 @@ +"""The reward worker: read the store, judge, score, sign. + +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; +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 + three-valued judge (``runtime/effects/_common.py``); +4. map the verdicts and the runtime's own signal onto ``RewardOutcomeV1``; +5. call the pure ``openadapt_types.reward.score`` helper; +6. write the evidence locally, sign the receipt, store both. + +INDETERMINATE never becomes 0. It becomes an unscored outcome with a +stated uncertainty, and the trainer drops the sample. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Literal, Optional, cast +from uuid import uuid4 + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from openadapt_types.oracle import OracleAdapter, OracleObservation +from openadapt_types.process_capability import canonical_json_bytes +from openadapt_types.reward import ( + REWARD_SCORING_CLASS, + RewardCertificateV1, + RewardContractV1, + RewardEvidenceReceiptV1, + RewardOutcomeV1, + RewardScoringClassV1, + RewardUncertaintyStateV1, + certificate_state, + score, +) +from pydantic import ValidationError + +from openadapt_flow.execute.keys import ( + fingerprint_of, + load_or_create_private_key, + load_or_create_token, +) +from openadapt_flow.reward.models import ( + EpisodeDescriptorV1, + IdentityError, + RewardBundle, + SelfSignedRewardEnvelopeV1, + assert_no_forbidden_keys, +) +from openadapt_flow.reward.oracles import ( + build_oracle, + effect_state_of, + records_of, +) +from openadapt_flow.runtime.effects._common import judge_records +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectKind, + EffectState, + EffectVerdict, + Verdict, +) + + +class RewardWorkerError(Exception): + """Typed failure with an HTTP status for the reference server.""" + + def __init__(self, status_code: int, error: str, detail: str) -> None: + super().__init__(detail) + self.status_code = status_code + self.error = error + self.detail = detail + + def body(self) -> dict[str, str]: + return {"error": self.error, "detail": self.detail} + + +class Judgement: + """The outcome mapping's result before scoring.""" + + __slots__ = ("outcome", "uncertainty", "required", "forbidden", "reason") + + def __init__( + self, + outcome: RewardOutcomeV1, + uncertainty: RewardUncertaintyStateV1, + required: list[EffectVerdict], + forbidden: list[EffectVerdict], + reason: str, + ) -> None: + self.outcome = outcome + self.uncertainty = uncertainty + self.required = required + self.forbidden = forbidden + self.reason = reason + + +_SIGNAL_OUTCOME: dict[str, RewardOutcomeV1] = { + "halted_before_effect": RewardOutcomeV1.HALTED_BEFORE_EFFECT, + "refused": RewardOutcomeV1.REFUSED, + "rejected_policy": RewardOutcomeV1.REJECTED_POLICY, +} + + +def judge_episode( + bundle: RewardBundle, + identity: dict[str, str], + runtime_signal: str, + before: EffectState, + observed: OracleObservation, +) -> Judgement: + """Map oracle verdicts and the runtime's signal onto a reward outcome. + + The table, in the order the rules fire: + + * runtime signal ``failed_platform`` -> ``FAILED_PLATFORM`` (unscored). + * store unreachable -> ``FAILED_PLATFORM``, uncertainty + ``oracle_unavailable`` (unscored). + * any required or forbidden verdict INDETERMINATE for another reason -> + ``RECONCILIATION_REQUIRED``, uncertainty ``effect_uncertain`` + (unscored). + * any forbidden effect present -> ``WRONG_EFFECT``. + * signal ``completed``: every required effect CONFIRMED -> ``VERIFIED``; + any REFUTED -> ``WRONG_EFFECT`` (the banner lie lands here: the screen + said saved, the store holds nothing). + * signal ``halted_before_effect`` / ``refused`` / ``rejected_policy``: + no required effect present -> that outcome; a required effect present + anyway -> ``RECONCILIATION_REQUIRED`` with ``effect_uncertain``, since + the runtime and the store disagree and a person settles it. + """ + + substrate = bundle.oracle.channel.value + if runtime_signal == "failed_platform": + return Judgement( + RewardOutcomeV1.FAILED_PLATFORM, + RewardUncertaintyStateV1.NONE, + [], + [], + "runtime reported a platform failure", + ) + current = records_of(observed) + if current is None: + return Judgement( + RewardOutcomeV1.FAILED_PLATFORM, + RewardUncertaintyStateV1.ORACLE_UNAVAILABLE, + [], + [], + "system of record unreachable at read time", + ) + required = [ + judge_records(_bind(effect, identity), before, current, substrate=substrate) + for effect in bundle.required_effects + ] + forbidden = [ + judge_records(_bind(effect, identity), before, current, substrate=substrate) + for effect in bundle.forbidden_effects + ] + indeterminate = [ + verdict + for verdict in (*required, *forbidden) + if verdict.verdict is Verdict.INDETERMINATE + ] + if indeterminate: + return Judgement( + RewardOutcomeV1.RECONCILIATION_REQUIRED, + RewardUncertaintyStateV1.EFFECT_UNCERTAIN, + required, + forbidden, + indeterminate[0].reason, + ) + present_forbidden = [v for v in forbidden if _forbidden_present(v)] + if present_forbidden: + return Judgement( + RewardOutcomeV1.WRONG_EFFECT, + RewardUncertaintyStateV1.NONE, + required, + forbidden, + "a forbidden effect is present in the system of record", + ) + all_confirmed = all(v.verdict is Verdict.CONFIRMED for v in required) + any_present = any(_required_present(v) for v in required) + if runtime_signal == "completed": + if all_confirmed: + return Judgement( + RewardOutcomeV1.VERIFIED, + RewardUncertaintyStateV1.NONE, + required, + forbidden, + "every required effect is present exactly as declared", + ) + refuted = next(v for v in required if v.verdict is Verdict.REFUTED) + return Judgement( + RewardOutcomeV1.WRONG_EFFECT, + RewardUncertaintyStateV1.NONE, + required, + forbidden, + refuted.reason, + ) + signalled = _SIGNAL_OUTCOME[runtime_signal] + if any_present: + return Judgement( + RewardOutcomeV1.RECONCILIATION_REQUIRED, + RewardUncertaintyStateV1.EFFECT_UNCERTAIN, + required, + forbidden, + f"runtime reported {runtime_signal} but a required effect is present", + ) + return Judgement( + signalled, + RewardUncertaintyStateV1.NONE, + required, + forbidden, + f"runtime reported {runtime_signal}; the store shows no required effect", + ) + + +def _bind(effect: Effect, identity: dict[str, str]) -> Effect: + return effect.resolve(identity) + + +def _required_present(verdict: EffectVerdict) -> bool: + if verdict.verdict is Verdict.CONFIRMED: + return True + return bool(verdict.observed_count) or bool(verdict.matched_records) + + +def _forbidden_present(verdict: EffectVerdict) -> bool: + """A forbidden effect is present when the store holds any matching record.""" + + if verdict.kind is EffectKind.FIELD_EQUALS: + return verdict.verdict is Verdict.CONFIRMED + if verdict.verdict is Verdict.CONFIRMED: + return True + return bool(verdict.observed_count) or bool(verdict.matched_records) + + +class RewardWorker: + """One reward contract, one oracle, one local key, on one machine.""" + + def __init__( + self, + bundle: RewardBundle | Path | str, + data_dir: Path | str, + *, + oracle: OracleAdapter | None = None, + token: str | None = None, + ) -> None: + self.bundle = ( + bundle if isinstance(bundle, RewardBundle) else RewardBundle.load(bundle) + ) + self.data_dir = Path(data_dir).expanduser() + self.data_dir.mkdir(parents=True, exist_ok=True) + self._lock = threading.RLock() + self._key: Ed25519PrivateKey = load_or_create_private_key(self.data_dir) + self.token = load_or_create_token(self.data_dir, token) + self.fingerprint = fingerprint_of(self._key.public_key()) + self.oracle: OracleAdapter = oracle or build_oracle( + self.bundle.oracle, self.bundle.directory + ) + if self.oracle.channel is not self.bundle.oracle.channel: + raise ValueError( + "oracle adapter channel does not match the contract's channel" + ) + (self.data_dir / "rewards").mkdir(parents=True, exist_ok=True) + (self.data_dir / "baselines").mkdir(parents=True, exist_ok=True) + + # -- public surface ----------------------------------------------------- + + @property + def contract(self) -> RewardContractV1: + return self.bundle.contract + + @property + def certificate(self) -> Optional[RewardCertificateV1]: + return self.bundle.certificate + + @property + def issuer_key_id(self) -> str: + return "self_signed:" + self.fingerprint + + 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. + """ + + 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")}, + ) + return state + + def score_episode( + self, payload: dict[str, Any] | EpisodeDescriptorV1 + ) -> dict[str, Any]: + """Read, judge, score, sign. Returns the stored envelope as a dict.""" + + try: + episode = ( + payload + if isinstance(payload, EpisodeDescriptorV1) + else EpisodeDescriptorV1.model_validate(payload) + ) + declared_identity = episode.resolved_identity() + signal = episode.resolved_signal() + except (ValidationError, ValueError) as exc: + raise RewardWorkerError(422, "invalid_episode", str(exc)) from exc + self._check_binding(episode) + with self._lock: + existing = self._episode_index(episode.episode_id) + if existing is not None: + raise RewardWorkerError( + 409, + "duplicate_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", + ) + try: + self.bundle.check_identity(identity) + except IdentityError as exc: + raise RewardWorkerError(422, "identity_mismatch", str(exc)) from exc + observed = self.oracle.read(identity) + judged = judge_episode(self.bundle, identity, signal, before, observed) + envelope = self._issue(episode, observed, before, judged) + return envelope + + def _check_binding(self, episode: EpisodeDescriptorV1) -> None: + if episode.reward_contract_digest != self.contract.digest: + raise RewardWorkerError( + 422, + "contract_mismatch", + f"this worker serves contract {self.contract.digest}, the episode " + f"names {episode.reward_contract_digest}", + ) + if episode.task_id is not None and episode.task_id != self.contract.task_id: + raise RewardWorkerError( + 422, "contract_mismatch", "task_id does not match the contract" + ) + if ( + episode.environment_id is not None + and episode.environment_id != self.contract.environment_id + ): + raise RewardWorkerError( + 422, "contract_mismatch", "environment_id does not match the contract" + ) + + def get_receipt(self, receipt_id: str) -> dict[str, Any]: + path = self.data_dir / "rewards" / receipt_id / "envelope.json" + if not path.is_file(): + raise RewardWorkerError(404, "not_found", "no such reward receipt") + payload = json.loads(path.read_text(encoding="utf-8")) + assert_no_forbidden_keys(payload.get("receipt") or {}) + return cast( + dict[str, Any], + SelfSignedRewardEnvelopeV1.model_validate(payload).model_dump(mode="json"), + ) + + def verify_receipt(self, receipt: RewardEvidenceReceiptV1) -> bool: + """Check a receipt's signature against this worker's key.""" + + from cryptography.exceptions import InvalidSignature + + try: + self._key.public_key().verify( + base64.b64decode(receipt.signature), + canonical_json_bytes(receipt.unsigned_payload()), + ) + except (InvalidSignature, ValueError): + return False + return True + + # -- issue -------------------------------------------------------------- + + def _issue( + self, + episode: EpisodeDescriptorV1, + observed: OracleObservation, + before: EffectState, + judged: Judgement, + ) -> dict[str, Any]: + tier = observed.tier + scored = score( + judged.outcome, + tier, + self.certificate, + episode.policy_update, + scoring=self.contract.scoring, + ) + state = certificate_state(self.certificate, episode.policy_update) + receipt_id = _new_id("reward_receipt") + evidence = { + "episode_id": episode.episode_id, + "oracle_channel": observed.channel.value, + "oracle_identity": dict(observed.identity), + "baseline_reachable": before.reachable, + "observed": observed.value, + "required_verdicts": [v.model_dump(mode="json") for v in judged.required], + "forbidden_verdicts": [v.model_dump(mode="json") for v in judged.forbidden], + "reason": judged.reason, + } + evidence_digest = ( + "sha256:" + hashlib.sha256(canonical_json_bytes(evidence)).hexdigest() + ) + components: dict[str, float] = {} + if scored.scalar is not None: + total = sum(c.weight for c in self.contract.components) + for component in self.contract.components: + components[component.name] = scored.scalar * component.weight / total + unsigned = { + "schema_version": "openadapt.reward-evidence-receipt/v1", + "receipt_id": receipt_id, + "reward_contract_digest": self.contract.digest, + "policy_checkpoint_id": episode.policy_checkpoint_id, + "policy_update": episode.policy_update, + "episode_id": episode.episode_id, + "oracle_tier": int(tier), + "reward_outcome": judged.outcome.value, + "evidence_digest": evidence_digest, + "reward_components": components, + "scalar_reward": scored.scalar, + "certificate_id": ( + self.certificate.certificate_id + if self.certificate is not None + else None + ), + "certificate_digest": ( + self.certificate.digest if self.certificate is not None else None + ), + "certificate_state": state.value, + "calibration_corpus_digest": ( + self.certificate.calibration_corpus_digest + if self.certificate is not None + else None + ), + "calibration_scope": ( + self.certificate.calibration_scope.value + if self.certificate is not None + else None + ), + "uncertainty": judged.uncertainty.value, + "certified": scored.certified, + "development_only": scored.development_only, + "issuer_key_id": self.issuer_key_id, + "nonce": _new_id("nonce"), + "issued_at": _now(), + } + signature = base64.b64encode( + self._key.sign(canonical_json_bytes(unsigned)) + ).decode("ascii") + receipt = RewardEvidenceReceiptV1.model_validate( + {**unsigned, "signature_algorithm": "ed25519", "signature": signature} + ) + payload = receipt.model_dump(mode="json") + assert_no_forbidden_keys(payload) + envelope = SelfSignedRewardEnvelopeV1( + issuer_key_fingerprint=self.fingerprint, + unscored=REWARD_SCORING_CLASS[judged.outcome] + is RewardScoringClassV1.UNSCORED, + receipt=payload, + ) + directory = self.data_dir / "rewards" / receipt_id + self._write_json(directory / "evidence.json", evidence) + self._write_json(directory / "receipt.json", payload) + envelope_payload = envelope.model_dump(mode="json") + self._write_json(directory / "envelope.json", envelope_payload) + self._write_json( + self.data_dir / "episodes" / f"{episode.episode_id}.json", + {"receipt_id": receipt_id}, + ) + return envelope_payload + + # -- storage ------------------------------------------------------------ + + def _baseline( + self, episode_id: str + ) -> tuple[Optional[dict[str, str]], EffectState]: + path = self.data_dir / "baselines" / f"{episode_id}.json" + if path.is_file(): + payload = json.loads(path.read_text("utf-8")) + identity = {str(k): str(v) for k, v in dict(payload["identity"]).items()} + return identity, EffectState.model_validate(payload["state"]) + # No baseline was captured: the delta is unknowable, so any effect + # that needs one (count_new_only, exact_new_set) judges INDETERMINATE. + return None, EffectState( + substrate=self.bundle.oracle.channel.value, reachable=False + ) + + def _episode_index(self, episode_id: str) -> Optional[str]: + path = self.data_dir / "episodes" / f"{episode_id}.json" + if not path.is_file(): + return None + payload = json.loads(path.read_text(encoding="utf-8")) + return str(payload.get("receipt_id") or "") or None + + 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") + tmp.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + tmp.replace(path) + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +def _now() -> str: + return ( + datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def default_data_dir() -> Path: + from openadapt_flow.reward import DEFAULT_DATA_DIRNAME + + return Path.home() / ".openadapt" / DEFAULT_DATA_DIRNAME + + +OutcomeLiteral = Literal[ + "verified", + "halted_before_effect", + "refused", + "rejected_policy", + "wrong_effect", + "reconciliation_required", + "failed_platform", +] diff --git a/pyproject.toml b/pyproject.toml index 1798314c..0b45f9be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,7 +44,7 @@ dependencies = [ "cryptography>=42.0", # Portable ProcessContract v1 capability, artifact, authentication, and # process-receipt contracts. The healthy runtime remains model-neutral. - "openadapt-types>=0.13.0,<0.14.0", + "openadapt-types>=0.17.0,<0.18.0", ] [project.optional-dependencies] @@ -72,11 +72,12 @@ dev = [ # uvicorn for the console's boot smoke test. "fastapi>=0.110", "uvicorn>=0.29", - # Types 0.13 keeps the existing entity and reviewed-copy contracts and adds - # the ProcessContract v1 capability, artifact, authentication, and receipt - # contracts. A peer must still negotiate each schema; the dependency alone - # never upgrades an existing decision surface. - "openadapt-types>=0.13.0,<0.14.0", + # Types 0.17 keeps the entity, reviewed-copy, ProcessContract, and Execute + # contracts and adds the reward contract, certificate (with calibration + # scope and issuer), and reward evidence receipt the reward worker signs. + # A peer must still negotiate each schema; the dependency alone never + # upgrades an existing decision surface. + "openadapt-types>=0.17.0,<0.18.0", # Engineering-hygiene gates (lint+format, type-check, coverage). Pinned to # majors so CI and local dev run the same checkers. "ruff==0.16.3", @@ -99,7 +100,7 @@ grounding = ["openadapt-grounding>=0.1.0"] console = [ "fastapi>=0.110", "uvicorn>=0.29", - "openadapt-types>=0.13.0,<0.14.0", + "openadapt-types>=0.17.0,<0.18.0", ] # Reference Execute server: `openadapt-flow serve-execute`. Same public # request schema as Cloud Execute, hosted in this process with a local @@ -107,10 +108,18 @@ console = [ execute = [ "fastapi>=0.110", "uvicorn>=0.29", - "openadapt-types>=0.13.0,<0.14.0", + "openadapt-types>=0.17.0,<0.18.0", ] # WindowsBackend: HTTP client for the WAA (Windows Agent Arena) server. windows = ["requests>=2.31", "pywin32>=312; platform_system == 'Windows'"] +# Reference reward worker: `openadapt-flow serve-reward`. Reads the system of +# record after a training episode and signs a reward receipt with a local key. +# A reward receipt is not an Execute Seal; the worker never imports Cloud. +reward = [ + "fastapi>=0.110", + "uvicorn>=0.29", + "openadapt-types>=0.17.0,<0.18.0", +] # Native macOS window capture/input. Imported lazily; other platforms never # install or import these framework bindings. macos = [ @@ -172,7 +181,7 @@ capture = ["openadapt-capture>=1.2.0"] # field-exact against the released schemas. The `interop-types` CI job # type-checks and tests the boundaries against the real package; each schema is # consumed only by an explicitly negotiated peer. -interop = ["openadapt-types>=0.13.0,<0.14.0"] +interop = ["openadapt-types>=0.17.0,<0.18.0"] [project.scripts] openadapt-flow = "openadapt_flow.__main__:main" diff --git a/tests/test_reward_worker.py b/tests/test_reward_worker.py new file mode 100644 index 00000000..dceb5e19 --- /dev/null +++ b/tests/test_reward_worker.py @@ -0,0 +1,656 @@ +"""Reference reward worker: outcome mapping, certificate, boundary, adapters.""" + +from __future__ import annotations + +import json +import math +from pathlib import Path +from typing import Any + +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("openadapt_types.reward") + +from fastapi.testclient import TestClient # noqa: E402 +from openadapt_types.oracle import OracleChannel # noqa: E402 +from openadapt_types.reward import ( # noqa: E402 + RewardCalibrationScopeV1, + RewardCertificateV1, + RewardEvidenceReceiptV1, + RewardOutcomeV1, +) + +from openadapt_flow.reward.calibration import ( # noqa: E402 + clopper_pearson_upper, + extradup_trials, +) +from openadapt_flow.reward.callables import ( # noqa: E402 + UNSCORED_REWARD, + drop_unscored, + is_unscored, + trl_reward_function, + verl_compute_score, +) +from openadapt_flow.reward.models import ( # noqa: E402 + CERTIFICATE_FILE, + FORBIDDEN_RECEIPT_KEYS, + RewardBundle, + assert_no_forbidden_keys, +) +from openadapt_flow.reward.oracles import JsonDocumentOracle # noqa: E402 +from openadapt_flow.reward.seed import ( # noqa: E402 + CALIBRATION_FILE, + CALIBRATION_TRIALS, + MOCKMED_DUPLICATE_PATIENT, + MOCKMED_HONEST_PATIENT, + MOCKMED_LIE_PATIENT, + mockmed_episode, + seed_mockmed, +) +from openadapt_flow.reward.serve import OPENAI_GRADER_ROUTE, create_app # noqa: E402 +from openadapt_flow.reward.worker import RewardWorker, RewardWorkerError # noqa: E402 + + +@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) + paths = seed_mockmed( + data_dir, key, "self_signed:" + fingerprint_of(key.public_key()) + ) + return {"data_dir": data_dir, **paths} + + +def _worker( + seeded: dict[str, Any], which: str = "tier2", **kwargs: Any +) -> RewardWorker: + return RewardWorker(seeded[which], seeded["data_dir"], token="test-token", **kwargs) + + +def _episode( + worker: RewardWorker, patient_id: str, episode_id: str, **kwargs: Any +) -> dict[str, Any]: + return mockmed_episode( + patient_id, + episode_id=episode_id, + contract_digest=worker.contract.digest, + **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) + + +# -- outcome mapping ---------------------------------------------------------- + + +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") + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED + assert receipt.oracle_tier == 2 + assert receipt.certified is True + assert receipt.development_only is False + assert receipt.scalar_reward == 1.0 + assert receipt.reward_components == {"terminal_effect": 1.0} + assert receipt.certificate_state.value == "current" + assert receipt.calibration_scope is RewardCalibrationScopeV1.SYNTHETIC + assert receipt.production_certified is False + assert receipt.reward_contract_digest == worker.contract.digest + assert envelope["unscored"] is False + assert envelope["execute_seal"] is False + assert envelope["production_seal"] is False + assert envelope["flow_governed_policy"] is False + assert worker.verify_receipt(receipt) + + +def test_verified_tier2_expired_certificate_is_not_certified( + seeded: dict[str, Any], +) -> None: + 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, + ) + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED + assert receipt.certified is False + assert receipt.certificate_state.value == "expired" + assert receipt.scalar_reward == 1.0 + + +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") + ) + 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 + assert receipt.certified is False + assert receipt.certificate_state.value == "absent" + assert receipt.calibration_scope is None + + +def test_banner_lie_yields_zero(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + envelope = worker.score_episode( + _episode(worker, MOCKMED_LIE_PATIENT, "episode_banner_lie_01") + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.WRONG_EFFECT + assert receipt.scalar_reward == 0.0 + assert receipt.certified is True + assert envelope["unscored"] is False + + +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") + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.WRONG_EFFECT + assert receipt.scalar_reward == 0.0 + evidence = json.loads( + ( + seeded["data_dir"] / "rewards" / receipt.receipt_id / "evidence.json" + ).read_text() + ) + assert evidence["required_verdicts"][0]["observed_count"] == 2 + + +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) + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.status_code == 409 + assert excinfo.value.error == "duplicate_episode" + + +def test_indeterminate_is_unscored_not_zero( + seeded: dict[str, Any], tmp_path: Path +) -> None: + worker = _worker(seeded, oracle=_unreachable(tmp_path)) + envelope = worker.score_episode( + _episode(worker, MOCKMED_HONEST_PATIENT, "episode_unreachable_1") + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.FAILED_PLATFORM + assert receipt.uncertainty.value == "oracle_unavailable" + assert receipt.scalar_reward is None + assert receipt.reward_components == {} + assert envelope["unscored"] is True + + +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") + # No baseline: the delta is unknowable, so the judge is INDETERMINATE and + # the episode is unscored, never 0. + envelope = worker.score_episode( + _episode(worker, MOCKMED_HONEST_PATIENT, "episode_new_only_01") + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.RECONCILIATION_REQUIRED + assert receipt.uncertainty.value == "effect_uncertain" + assert receipt.scalar_reward is 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)) + envelope = worker.score_episode( + { + "episode_id": "episode_new_only_02", + "policy_checkpoint_id": "policy_checkpoint_mockmed_0", + "policy_update": 0, + "reward_contract_digest": worker.contract.digest, + } + ) + assert _receipt(envelope).reward_outcome is RewardOutcomeV1.VERIFIED + + +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", + ) + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.HALTED_BEFORE_EFFECT + assert receipt.scalar_reward == 0.0 + + +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", + ) + ) + receipt = _receipt(envelope) + assert receipt.reward_outcome is RewardOutcomeV1.RECONCILIATION_REQUIRED + assert receipt.scalar_reward is None + assert envelope["unscored"] is True + + +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") + ) + assert _receipt(envelope).reward_outcome is RewardOutcomeV1.WRONG_EFFECT + + +# -- refusals ------------------------------------------------------------------- + + +def test_extra_identity_key_is_refused(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_extra_key_01") + payload["metadata"]["oracle_identity"]["encounter_id"] = "enc-1" + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.status_code == 422 + assert excinfo.value.error == "identity_mismatch" + + +def test_missing_identity_is_refused(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_no_identity_1") + del payload["metadata"]["oracle_identity"] + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.error == "identity_missing" + + +def test_wrong_contract_digest_is_refused(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_wrong_contract") + payload["reward_contract_digest"] = "sha256:" + "0" * 64 + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.error == "contract_mismatch" + + +def test_extra_episode_field_is_refused(seeded: dict[str, Any]) -> None: + worker = _worker(seeded) + payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_extra_field_1") + payload["screenshot"] = "iVBORw0KGgo=" + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.error == "invalid_episode" + payload = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_extra_field_2") + payload["metadata"]["screenshots"] = ["iVBORw0KGgo="] + with pytest.raises(RewardWorkerError) as excinfo: + worker.score_episode(payload) + assert excinfo.value.error == "invalid_episode" + + +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") + ) + receipt = envelope["receipt"] + assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(receipt) + assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(envelope) + flat = json.dumps(envelope) + assert MOCKMED_HONEST_PATIENT not in flat + assert "iVBOR" not in flat + with pytest.raises(ValueError, match="forbids"): + assert_no_forbidden_keys({**receipt, "screenshots": []}) + for name in ("execution_id", "workflow_digest", "qualification_id", "contracts"): + assert name not in receipt + + +def test_bundle_refuses_tampered_effects(seeded: dict[str, Any]) -> None: + bundle_dir = seeded["tier2"] + path = bundle_dir / "required_effects.json" + effects = json.loads(path.read_text()) + effects[0]["expected_count"] = 2 + path.write_text(json.dumps(effects)) + with pytest.raises(ValueError, match="digest"): + RewardBundle.load(bundle_dir) + + +def test_certificate_bound_is_recomputable(seeded: dict[str, Any]) -> None: + bundle_dir = seeded["tier2"] + certificate = RewardCertificateV1.model_validate( + json.loads((bundle_dir / CERTIFICATE_FILE).read_text()) + ) + assert certificate.calibration_scope is RewardCalibrationScopeV1.SYNTHETIC + assert certificate.issuer.value == "self_signed" + calibration = json.loads((bundle_dir / CALIBRATION_FILE).read_text()) + assert calibration["calibration_trials"] == CALIBRATION_TRIALS + assert calibration["calibration_false_accepts"] == 0 + recomputed = clopper_pearson_upper( + calibration["calibration_false_accepts"], + calibration["calibration_trials"], + confidence=calibration["calibration_confidence"], + ) + assert certificate.epsilon == recomputed + assert certificate.epsilon == pytest.approx( + 1.0 - 0.05 ** (1.0 / CALIBRATION_TRIALS) + ) + + +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 + result = extradup_trials( + lambda records, identity: RewardOutcomeV1.VERIFIED, + trials=10, + generator_seed=1, + ) + assert result.false_accepts == 10 + assert result.epsilon == 1.0 + + +# -- HTTP surface --------------------------------------------------------------- + + +def _client( + seeded: dict[str, Any], which: str = "tier2", **kwargs: Any +) -> tuple[TestClient, RewardWorker]: + worker = _worker(seeded, which, **kwargs) + client = TestClient(create_app(worker)) + client.headers["Authorization"] = "Bearer test-token" + return client, worker + + +def test_http_reward_roundtrip(seeded: dict[str, Any]) -> None: + client, worker = _client(seeded) + bare = TestClient(client.app) + health = bare.get("/health").json() + assert health["issuer"] == "self_signed" + assert health["execute_seal"] is False + assert health["oracle_tier"] == 2 + assert bare.post("/v1/rewards", json={}).status_code == 401 + created = client.post( + "/v1/rewards", json=_episode(worker, MOCKMED_HONEST_PATIENT, "episode_http_01") + ) + assert created.status_code == 200 + assert created.headers["X-OpenAdapt-Execute-Seal"] == "false" + receipt_id = created.json()["receipt"]["receipt_id"] + fetched = client.get(f"/v1/rewards/{receipt_id}") + assert fetched.status_code == 200 + assert fetched.json() == created.json() + assert client.get("/v1/rewards/reward_receipt_missing").status_code == 404 + again = client.post( + "/v1/rewards", json=_episode(worker, MOCKMED_HONEST_PATIENT, "episode_http_01") + ) + assert again.status_code == 409 + + +def test_http_matches_the_evals_client_wire_shape(seeded: dict[str, Any]) -> None: + """Round-trip the exact JSON ``openadapt_evals.reward.receipts`` sends. + + ``EpisodeDescriptor.as_payload()`` drops ``None`` fields and always sends + ``metadata``; ``HttpRewardEndpoint`` requires HTTP 200 and parses either + a bare receipt or ``{"receipt": ...}`` when the body has no top-level + ``schema_version``. + """ + + client, worker = _client(seeded) + payload = { + "episode_id": "episode_evals_client_1", + "policy_checkpoint_id": "policy_checkpoint_evals", + "policy_update": 4, + "reward_contract_digest": worker.contract.digest, + "metadata": {"oracle_identity": {"patient_id": MOCKMED_HONEST_PATIENT}}, + } + response = client.post("/v1/rewards", json=payload) + assert response.status_code == 200 + body = response.json() + assert "receipt" in body and "schema_version" not in body + receipt = RewardEvidenceReceiptV1.model_validate(body["receipt"]) + assert receipt.episode_id == "episode_evals_client_1" + assert receipt.policy_checkpoint_id == "policy_checkpoint_evals" + assert receipt.policy_update == 4 + assert receipt.reward_outcome is RewardOutcomeV1.VERIFIED + minimal = { + "episode_id": "episode_evals_client_2", + "policy_checkpoint_id": "policy_checkpoint_evals", + "policy_update": 4, + "reward_contract_digest": worker.contract.digest, + "metadata": {}, + } + refused = client.post("/v1/rewards", json=minimal) + assert refused.status_code == 422 + assert refused.json()["error"] == "identity_missing" + + +def test_openai_grader_route_contract(seeded: dict[str, Any]) -> None: + client, worker = _client(seeded) + item = _episode(worker, MOCKMED_HONEST_PATIENT, "episode_grader_ok_1") + scored = client.post( + OPENAI_GRADER_ROUTE, + json={"sample": {"output_text": "saved"}, "item": item}, + ) + assert scored.status_code == 200 + body = scored.json() + assert body["score"] == 1.0 + assert 0.0 <= body["score"] <= 1.0 + assert body["certified"] is True + lie = client.post( + OPENAI_GRADER_ROUTE, + json={ + "sample": {"output_text": "saved"}, + "item": _episode(worker, MOCKMED_LIE_PATIENT, "episode_grader_lie_1"), + }, + ) + assert lie.json()["score"] == 0.0 + assert client.post(OPENAI_GRADER_ROUTE, json={"item": item}).status_code == 400 + + +def test_openai_grader_route_refuses_to_grade_unscored( + seeded: dict[str, Any], tmp_path: Path +) -> None: + client, worker = _client(seeded, oracle=_unreachable(tmp_path)) + response = client.post( + OPENAI_GRADER_ROUTE, + json={ + "sample": {"output_text": "saved"}, + "item": _episode(worker, MOCKMED_HONEST_PATIENT, "episode_grader_x_1"), + }, + ) + assert response.status_code == 422 + assert response.json()["error"] == "unscored" + assert "score" not in response.json() + + +# -- trainer callables ---------------------------------------------------------- + + +class _State: + global_step = 3 + + +def test_trl_reward_function_contract(seeded: dict[str, Any], tmp_path: Path) -> None: + worker = _worker(seeded) + reward = trl_reward_function( + worker, + policy_checkpoint_id="policy_checkpoint_trl_1", + reward_contract_digest=worker.contract.digest, + ) + assert reward.__name__ == "openadapt_verified_effect_reward" + rewards = reward( + prompts=["p1", "p2"], + completions=["c1", "c2"], + completion_ids=[[1], [2]], + trainer_state=_State(), + episode_id=["episode_trl_0001", "episode_trl_0002"], + oracle_identity=[ + {"patient_id": MOCKMED_HONEST_PATIENT}, + {"patient_id": MOCKMED_LIE_PATIENT}, + ], + ) + assert rewards == [1.0, 0.0] + unreachable = _worker(seeded, oracle=_unreachable(tmp_path)) + reward2 = trl_reward_function( + unreachable, + policy_checkpoint_id="policy_checkpoint_trl_2", + reward_contract_digest=unreachable.contract.digest, + ) + rewards2 = reward2( + prompts=["p"], + completions=["c"], + trainer_state=_State(), + episode_id=["episode_trl_0003"], + oracle_identity=[{"patient_id": MOCKMED_HONEST_PATIENT}], + ) + assert rewards2 == [None] + kept, (kept_completions,) = drop_unscored([1.0, None, 0.0], ["a", "b", "c"]) + assert kept == [1.0, 0.0] + assert kept_completions == ["a", "c"] + + +def test_verl_compute_score_contract(seeded: dict[str, Any], tmp_path: Path) -> None: + worker = _worker(seeded) + compute_score = verl_compute_score( + worker, + policy_checkpoint_id="policy_checkpoint_verl", + reward_contract_digest=worker.contract.digest, + ) + result = compute_score( + data_source="mockmed", + solution_str="saved", + ground_truth=None, + extra_info={ + "openadapt_episode": { + "episode_id": "episode_verl_0001", + "oracle_identity": {"patient_id": MOCKMED_HONEST_PATIENT}, + "policy_update": 0, + } + }, + ) + assert result["score"] == 1.0 + assert result["openadapt_unscored"] is False + unreachable = _worker(seeded, oracle=_unreachable(tmp_path)) + compute2 = verl_compute_score( + unreachable, + policy_checkpoint_id="policy_checkpoint_verl", + reward_contract_digest=unreachable.contract.digest, + ) + result2 = compute2( + "mockmed", + "saved", + None, + { + "openadapt_episode": { + "episode_id": "episode_verl_0002", + "oracle_identity": {"patient_id": MOCKMED_HONEST_PATIENT}, + } + }, + ) + assert math.isnan(result2["score"]) + assert result2["openadapt_unscored"] is True + assert is_unscored(UNSCORED_REWARD) + assert not is_unscored(0.0) + kept, _ = drop_unscored([result["score"], result2["score"]]) + assert kept == [1.0] + + +# -- CLI and boundary ----------------------------------------------------------- + + +def test_cli_wires_serve_reward() -> None: + from openadapt_flow.__main__ import build_parser + + args = build_parser().parse_args( + ["serve-reward", "--contract", "/tmp/c", "--port", "8788", "--seed-mockmed"] + ) + assert args.command == "serve-reward" + assert args.contract == "/tmp/c" + assert args.port == 8788 + assert args.seed_mockmed is True + assert args.func.__name__ == "_cmd_serve_reward" + + +def test_source_boundary_has_no_cloud_modules() -> None: + root = Path(__file__).resolve().parents[1] / "openadapt_flow" / "reward" + names = {path.name for path in root.glob("*.py")} + assert names.isdisjoint( + {"tenant.py", "billing.py", "stripe.py", "control_plane.py"} + ) + joined = "\n".join(path.read_text(encoding="utf-8") for path in root.glob("*.py")) + assert "openadapt_cloud" not in joined + assert "openadapt-cloud" not in joined + assert "app.openadapt.ai" not in joined + assert "ExecuteEvidenceReceiptV1" not in joined diff --git a/uv.lock b/uv.lock index 80b02f81..7f9506ab 100644 --- a/uv.lock +++ b/uv.lock @@ -2213,6 +2213,11 @@ privacy = [ rdp = [ { name = "aardwolf" }, ] +reward = [ + { name = "fastapi" }, + { name = "openadapt-types" }, + { name = "uvicorn" }, +] service = [ { name = "fastapi" }, { name = "uvicorn", extra = ["standard"] }, @@ -2238,6 +2243,7 @@ requires-dist = [ { name = "fastapi", marker = "extra == 'console'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'dev'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'execute'", specifier = ">=0.110" }, + { name = "fastapi", marker = "extra == 'reward'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'service'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'service-mlx'", specifier = ">=0.110" }, { name = "httpx", specifier = ">=0.27" }, @@ -2252,11 +2258,12 @@ requires-dist = [ { name = "openadapt-capture", marker = "extra == 'capture'", specifier = ">=1.2.0" }, { name = "openadapt-grounding", marker = "extra == 'grounding'", specifier = ">=0.1.0" }, { name = "openadapt-privacy", extras = ["presidio"], marker = "extra == 'privacy'", specifier = ">=1.0.0" }, - { name = "openadapt-types", specifier = ">=0.13.0,<0.14.0" }, - { name = "openadapt-types", marker = "extra == 'console'", specifier = ">=0.13.0,<0.14.0" }, - { name = "openadapt-types", marker = "extra == 'dev'", specifier = ">=0.13.0,<0.14.0" }, - { name = "openadapt-types", marker = "extra == 'execute'", specifier = ">=0.13.0,<0.14.0" }, - { name = "openadapt-types", marker = "extra == 'interop'", specifier = ">=0.13.0,<0.14.0" }, + { name = "openadapt-types", specifier = ">=0.17.0,<0.18.0" }, + { name = "openadapt-types", marker = "extra == 'console'", specifier = ">=0.17.0,<0.18.0" }, + { name = "openadapt-types", marker = "extra == 'dev'", specifier = ">=0.17.0,<0.18.0" }, + { name = "openadapt-types", marker = "extra == 'execute'", specifier = ">=0.17.0,<0.18.0" }, + { name = "openadapt-types", marker = "extra == 'interop'", specifier = ">=0.17.0,<0.18.0" }, + { name = "openadapt-types", marker = "extra == 'reward'", specifier = ">=0.17.0,<0.18.0" }, { name = "opencv-python", specifier = ">=4.9" }, { name = "pillow", specifier = ">=10.0" }, { name = "playwright", marker = "extra == 'browser'", specifier = ">=1.44" }, @@ -2281,10 +2288,11 @@ requires-dist = [ { name = "uvicorn", marker = "extra == 'console'", specifier = ">=0.29" }, { name = "uvicorn", marker = "extra == 'dev'", specifier = ">=0.29" }, { name = "uvicorn", marker = "extra == 'execute'", specifier = ">=0.29" }, + { name = "uvicorn", marker = "extra == 'reward'", specifier = ">=0.29" }, { name = "uvicorn", extras = ["standard"], marker = "extra == 'service'", specifier = ">=0.29" }, { name = "uvicorn", extras = ["standard"], marker = "extra == 'service-mlx'", specifier = ">=0.29" }, ] -provides-extras = ["browser", "dev", "grounder", "grounding", "console", "execute", "windows", "macos", "linux", "rdp", "privacy", "hosted", "service", "service-mlx", "capture", "interop"] +provides-extras = ["browser", "dev", "grounder", "grounding", "console", "execute", "windows", "reward", "macos", "linux", "rdp", "privacy", "hosted", "service", "service-mlx", "capture", "interop"] [[package]] name = "openadapt-grounding" @@ -2324,14 +2332,14 @@ presidio = [ [[package]] name = "openadapt-types" -version = "0.13.0" +version = "0.17.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/31/60/5bf6f51cb885390ed0176610f3875bce6ccda2df39417b80b4e3649be453/openadapt_types-0.13.0.tar.gz", hash = "sha256:b044ac809cf8891bd7d4aaf44a8b44b8f1b44f7847717e685375ac313ddd831a", size = 175106, upload-time = "2026-08-31T00:26:26.323Z" } +sdist = { url = "https://files.pythonhosted.org/packages/17/d5/549afc2b6bdde8a0a80d4eec67644275df251ee1654f3189f0cc66235730/openadapt_types-0.17.0.tar.gz", hash = "sha256:43dc00d82d6d8feb2cfaa16960224a85933440d7dee3b5f3c60618873535287b", size = 212574, upload-time = "2026-09-01T23:50:50.837Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/44/76/063d4d6a83bded9a4aabdbc809dca65c528a95cfaaffec67de0919d6f4f3/openadapt_types-0.13.0-py3-none-any.whl", hash = "sha256:6549cf6179c0b851ff1e1f788f4600a741741e4d54a1a0edfdcec49d1ed5d273", size = 123629, upload-time = "2026-08-31T00:26:24.74Z" }, + { url = "https://files.pythonhosted.org/packages/55/9a/bec1d51cba5749732ceeba6e4abf7abe6aeaa22c849a1edb26b97c7ab479/openadapt_types-0.17.0-py3-none-any.whl", hash = "sha256:79738ab529f71241963725c6a40603deae032ff60c61d66baf33d7a248b1afa9", size = 155033, upload-time = "2026-09-01T23:50:49.083Z" }, ] [[package]]