feat(reward): MIT reference reward worker with self-signed reward receipts

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

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 <token>`. 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.

openadapt_flow/__main__.py

@@ -61,6 +61,9 @@
   UI over bundles / runs / skill libraries; requires the ``console`` extra).
 - ``serve-execute`` — host the public Execute HTTP+MCP contract on this
   machine (self-signed local receipts; not an OpenAdapt production Seal).
+- ``serve-reward`` — score training episodes by reading the system of
+  record through an independent oracle (self-signed reward receipts; not
+  an Execute Seal).
 - ``emit-skill`` — emit an Agent Skills folder for a bundle.
 - ``emit-mcp`` — emit a standalone MCP ``server.py`` for a bundle.
 - ``connector`` — the BYOC (bring-your-own-cloud) outbound-pull daemon:
@@ -7668,6 +7671,55 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None:
     )
     p.set_defaults(func=_cmd_serve_execute)
 
+    p = sub.add_parser(
+        "serve-reward",
+        help=(
+            "Score training episodes by reading the system of record through "
+            "an independent oracle. Receipts are self-signed reward receipts, "
+            "not Execute Seals. Needs `pip install 'openadapt-flow[reward]'`"
+        ),
+    )
+    p.add_argument(
+        "--contract",
+        default=None,
+        help=(
+            "Reward contract bundle directory (contract.json, "
+            "required_effects.json, forbidden_effects.json, oracle.json, "
+            "optional certificate.json). Defaults to the seeded MockMed "
+            "tier-2 bundle when --seed-mockmed is given."
+        ),
+    )
+    p.add_argument(
+        "--port",
+        type=int,
+        default=8788,
+        help="Port (default: 8788)",
+    )
+    p.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Bind address (default: 127.0.0.1)",
+    )
+    p.add_argument(
+        "--data-dir",
+        default=None,
+        help="Local data directory (default: ~/.openadapt/reward-ref)",
+    )
+    p.add_argument(
+        "--token",
+        default=None,
+        help="Bearer token (default: generated on first start in --data-dir)",
+    )
+    p.add_argument(
+        "--seed-mockmed",
+        action="store_true",
+        help=(
+            "Write the synthetic MockMed reward bundles (tier-2 file oracle "
+            "with a calibrated synthetic certificate; tier-0 screen dump)"
+        ),
+    )
+    p.set_defaults(func=_cmd_serve_reward)
+
     p = sub.add_parser(
         "business-decisions",
         help="Customer-runner typed-decision relay; it never resumes or acts.",
@@ -7967,6 +8019,61 @@ def _cmd_serve_execute(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_serve_reward(args: argparse.Namespace) -> int:
+    from importlib.util import find_spec
+
+    missing = [
+        name
+        for name in ("fastapi", "uvicorn", "openadapt_types")
+        if find_spec(name) is None
+    ]
+    if missing:
+        raise SystemExit(
+            f"serve-reward needs {', '.join(missing)} — install the "
+            "reward extra:  pip install 'openadapt-flow[reward]'"
+        )
+    from openadapt_flow.reward import REWARD_NOTICE
+    from openadapt_flow.reward.serve import serve
+    from openadapt_flow.reward.worker import RewardWorker, default_data_dir
+
+    data_dir = Path(args.data_dir) if args.data_dir else default_data_dir()
+    contract = args.contract
+    seeded_paths: dict[str, Path] = {}
+    if args.seed_mockmed:
+        from openadapt_flow.execute.keys import (
+            fingerprint_of,
+            load_or_create_private_key,
+        )
+        from openadapt_flow.reward.seed import seed_mockmed
+
+        key = load_or_create_private_key(data_dir)
+        issuer_key_id = "self_signed:" + fingerprint_of(key.public_key())
+        seeded_paths = seed_mockmed(data_dir, key, issuer_key_id)
+        if contract is None:
+            contract = str(seeded_paths["tier2"])
+    if contract is None:
+        raise SystemExit("serve-reward needs --contract <bundle dir> 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 = (

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",
+]