From adf0a3e3a44549d0f587ec40c4aff7c7781e9009 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 1 Sep 2026 20:09:37 -0400 Subject: [PATCH 1/3] feat(reward): MIT reference reward worker with self-signed reward receipts Add openadapt_flow/reward: a loopback HTTP worker that reads the system of record after a training episode through an independent oracle, judges required and forbidden effects with the shared three-valued judge, scores with openadapt_types.reward.score, and signs a RewardEvidenceReceiptV1 with the same local Ed25519 key mechanism as the reference Execute server. INDETERMINATE and an unreachable store map to unscored outcomes, never 0. Tier 0 and 1 reads are development_only and never certified. The seeded MockMed certificate carries synthetic scope and an epsilon computed as the exact one-sided 95% Clopper-Pearson bound from 300 ExtraDup trials the seed runs through the bundle's own judge. The POST /v1/rewards wire shape matches openadapt_evals.reward.receipts. The OpenAI grader route mirrors the python grader's grade(sample, item) contract and answers 422 for an unscored episode. TRL and verl adapters return None and NaN respectively for unscored samples, with a helper that drops them from a group. Co-Authored-By: Claude Fable 5.1 --- README.md | 19 + docs/REWARD_WORKER.md | 209 +++++++++ openadapt_flow/reward/__init__.py | 32 ++ openadapt_flow/reward/calibration.py | 162 +++++++ openadapt_flow/reward/callables.py | 276 +++++++++++ openadapt_flow/reward/models.py | 457 +++++++++++++++++++ openadapt_flow/reward/oracles.py | 193 ++++++++ openadapt_flow/reward/seed.py | 320 +++++++++++++ openadapt_flow/reward/serve.py | 228 ++++++++++ openadapt_flow/reward/worker.py | 553 ++++++++++++++++++++++ tests/test_reward_worker.py | 656 +++++++++++++++++++++++++++ 11 files changed, 3105 insertions(+) create mode 100644 docs/REWARD_WORKER.md create mode 100644 openadapt_flow/reward/__init__.py create mode 100644 openadapt_flow/reward/calibration.py create mode 100644 openadapt_flow/reward/callables.py create mode 100644 openadapt_flow/reward/models.py create mode 100644 openadapt_flow/reward/oracles.py create mode 100644 openadapt_flow/reward/seed.py create mode 100644 openadapt_flow/reward/serve.py create mode 100644 openadapt_flow/reward/worker.py create mode 100644 tests/test_reward_worker.py 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/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/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 From 7b76a3bb3f84ad32776afdb0c4db64f0b6cf1e18 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 1 Sep 2026 20:10:57 -0400 Subject: [PATCH 2/3] build: pin openadapt-types 0.17 and add the reward extra Co-Authored-By: Claude Fable 5.1 --- pyproject.toml | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1798314c..be5ae51b 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 @@ -111,6 +112,14 @@ execute = [ ] # 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" From 76abf869212e7b0a06ca41dfff15061c11b80393 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 1 Sep 2026 20:44:48 -0400 Subject: [PATCH 3/3] feat(cli): wire serve-reward and relock openadapt-types 0.17 Co-Authored-By: Claude Fable 5.1 --- openadapt_flow/__main__.py | 107 +++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- uv.lock | 26 +++++---- 3 files changed, 125 insertions(+), 10 deletions(-) diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index dbc32c2d..2064586d 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -60,6 +60,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: @@ -7654,6 +7657,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.", @@ -7953,6 +8005,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/pyproject.toml b/pyproject.toml index be5ae51b..0b45f9be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -108,7 +108,7 @@ 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'"] 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]]